♻️ Maintenance: enhances pytest_simcore tooling (#7274)

Author: pcrespov

packages/models-library/tests/test_service_settings_labels.py

@@ -5,7 +5,6 @@
 
 import json
 from copy import deepcopy
-from pprint import pformat
 from typing import Any, Final, NamedTuple
 
 import pytest
@@ -33,6 +32,10 @@
 from models_library.services_resources import DEFAULT_SINGLE_SERVICE_NAME
 from models_library.utils.string_substitution import TextTemplate
 from pydantic import BaseModel, TypeAdapter, ValidationError
+from pytest_simcore.pydantic_models import (
+    assert_validation_model,
+    iter_model_examples_in_class,
+)
 
 
 class _Parametrization(NamedTuple):
@@ -89,17 +92,23 @@ def test_service_settings():
         service_setting.set_destination_containers(["random_value1", "random_value2"])
 
 
-@pytest.mark.parametrize("model_cls", [SimcoreServiceLabels])
+@pytest.mark.parametrize(
+    "model_cls, example_name, example_data",
+    iter_model_examples_in_class(SimcoreServiceLabels),
+)
 def test_correctly_detect_dynamic_sidecar_boot(
-    model_cls: type[BaseModel], model_cls_examples: dict[str, dict[str, Any]]
+    model_cls: type[BaseModel], example_name: str, example_data: Any
 ):
-    for name, example in model_cls_examples.items():
-        print(name, ":", pformat(example))
-        model_instance = TypeAdapter(model_cls).validate_python(example)
-        assert model_instance.callbacks_mapping is not None
-        assert model_instance.needs_dynamic_sidecar == (
-            "simcore.service.paths-mapping" in example
-        )
+
+    model_instance = assert_validation_model(
+        model_cls, example_name=example_name, example_data=example_data
+    )
+
+    assert isinstance(model_instance, SimcoreServiceLabels)
+    assert model_instance.callbacks_mapping is not None
+    assert model_instance.needs_dynamic_sidecar == (
+        "simcore.service.paths-mapping" in example_data
+    )
 
 
 def test_raises_error_if_http_entrypoint_is_missing():

packages/pytest-simcore/src/pytest_simcore/helpers/faker_factories.py

@@ -220,6 +220,7 @@ def fake_task(**overrides) -> dict[str, Any]:
 
 
 def random_product(
+    *,
     group_id: int | None = None,
     registration_email_template: str | None = None,
     fake: Faker = DEFAULT_FAKER,
@@ -288,6 +289,29 @@ def random_product(
     return data
 
 
+def random_product_price(
+    *, product_name: str, fake: Faker = DEFAULT_FAKER, **overrides
+) -> dict[str, Any]:
+    from simcore_postgres_database.models.products_prices import products_prices
+
+    data = {
+        "product_name": product_name,
+        "usd_per_credit": fake.pydecimal(left_digits=2, right_digits=2, positive=True),
+        "min_payment_amount_usd": fake.pydecimal(
+            left_digits=2, right_digits=2, positive=True
+        ),
+        "comment": fake.sentence(),
+        "valid_from": fake.date_time_this_decade(),
+        "stripe_price_id": fake.uuid4(),
+        "stripe_tax_rate_id": fake.uuid4(),
+    }
+
+    assert set(data.keys()).issubset({c.name for c in products_prices.columns})
+
+    data.update(overrides)
+    return data
+
+
 def utcnow() -> datetime:
     return datetime.now(tz=UTC)
 

packages/pytest-simcore/src/pytest_simcore/helpers/postgres_tools.py

@@ -6,7 +6,7 @@
 import sqlalchemy as sa
 from psycopg2 import OperationalError
 from simcore_postgres_database.models.base import metadata
-from sqlalchemy.ext.asyncio import AsyncEngine
+from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine
 
 
 class PostgresTestConfig(TypedDict):
@@ -76,25 +76,46 @@ def migrated_pg_tables_context(
 def is_postgres_responsive(url) -> bool:
     """Check if something responds to ``url``"""
     try:
-        engine = sa.create_engine(url)
-        conn = engine.connect()
+        sync_engine = sa.create_engine(url)
+        conn = sync_engine.connect()
         conn.close()
     except OperationalError:
         return False
     return True
 
 
-async def _insert_and_get_row(
-    conn, table: sa.Table, values: dict[str, Any], pk_col: sa.Column, pk_value: Any
+async def _async_insert_and_get_row(
+    conn: AsyncConnection,
+    table: sa.Table,
+    values: dict[str, Any],
+    pk_col: sa.Column,
+    pk_value: Any,
 ):
     result = await conn.execute(table.insert().values(**values).returning(pk_col))
-    row = result.first()
+    row = result.one()
 
     # NOTE: DO NO USE row[pk_col] since you will get a deprecation error (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9)
     assert getattr(row, pk_col.name) == pk_value
 
     result = await conn.execute(sa.select(table).where(pk_col == pk_value))
-    return result.first()
+    return result.one()
+
+
+def _sync_insert_and_get_row(
+    conn: sa.engine.Connection,
+    table: sa.Table,
+    values: dict[str, Any],
+    pk_col: sa.Column,
+    pk_value: Any,
+):
+    result = conn.execute(table.insert().values(**values).returning(pk_col))
+    row = result.one()
+
+    # NOTE: DO NO USE row[pk_col] since you will get a deprecation error (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9)
+    assert getattr(row, pk_col.name) == pk_value
+
+    result = conn.execute(sa.select(table).where(pk_col == pk_value))
+    return result.one()
 
 
 @asynccontextmanager
@@ -108,14 +129,48 @@ async def insert_and_get_row_lifespan(
 ) -> AsyncIterator[dict[str, Any]]:
     # insert & get
     async with sqlalchemy_async_engine.begin() as conn:
-        row = await _insert_and_get_row(
+        row = await _async_insert_and_get_row(
             conn, table=table, values=values, pk_col=pk_col, pk_value=pk_value
         )
 
+    assert row
+
     # NOTE: DO NO USE dict(row) since you will get a deprecation error (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9)
     # pylint: disable=protected-access
     yield row._asdict()
 
     # delete row
     async with sqlalchemy_async_engine.begin() as conn:
         await conn.execute(table.delete().where(pk_col == pk_value))
+
+
+@contextmanager
+def sync_insert_and_get_row_lifespan(
+    sqlalchemy_sync_engine: sa.engine.Engine,
+    *,
+    table: sa.Table,
+    values: dict[str, Any],
+    pk_col: sa.Column,
+    pk_value: Any,
+) -> Iterator[dict[str, Any]]:
+    """sync version of insert_and_get_row_lifespan.
+
+    TIP: more convenient for **module-scope fixtures** that setup the
+    database tables before the app starts since it does not require an `event_loop`
+    fixture (which is funcition-scoped )
+    """
+    # insert & get
+    with sqlalchemy_sync_engine.begin() as conn:
+        row = _sync_insert_and_get_row(
+            conn, table=table, values=values, pk_col=pk_col, pk_value=pk_value
+        )
+
+    assert row
+
+    # NOTE: DO NO USE dict(row) since you will get a deprecation error (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9)
+    # pylint: disable=protected-access
+    yield row._asdict()
+
+    # delete row
+    with sqlalchemy_sync_engine.begin() as conn:
+        conn.execute(table.delete().where(pk_col == pk_value))

packages/pytest-simcore/src/pytest_simcore/pydantic_models.py

@@ -2,14 +2,15 @@
 import importlib
 import inspect
 import itertools
-import json
 import pkgutil
+import warnings
 from collections.abc import Iterator
 from contextlib import suppress
 from types import ModuleType
-from typing import Any, NamedTuple
+from typing import Any, NamedTuple, TypeVar
 
 import pytest
+from common_library.json_serialization import json_dumps
 from pydantic import BaseModel, ValidationError
 
 
@@ -63,18 +64,20 @@ def walk_model_examples_in_package(package: ModuleType) -> Iterator[ModelExample
 def iter_model_examples_in_module(module: object) -> Iterator[ModelExample]:
     """Iterates on all examples defined as BaseModelClass.model_config["json_schema_extra"]["example"]
 
-
     Usage:
+        import some_package.some_module
 
         @pytest.mark.parametrize(
             "model_cls, example_name, example_data",
-            iter_model_examples_in_module(simcore_service_webserver.storage_schemas),
+            iter_model_examples_in_module(some_package.some_module),
         )
         def test_model_examples(
-            model_cls: type[BaseModel], example_name: int, example_data: Any
+            model_cls: type[BaseModel], example_name: str, example_data: Any
         ):
-            print(example_name, ":", json.dumps(example_data))
-            assert model_cls.model_validate(example_data)
+            assert_validation_model(
+                model_cls, example_name=example_name, example_data=example_data
+            )
+
     """
 
     def _is_model_cls(obj) -> bool:
@@ -95,36 +98,69 @@ def _is_model_cls(obj) -> bool:
 
     for model_name, model_cls in inspect.getmembers(module, _is_model_cls):
 
-        schema = model_cls.model_json_schema()
+        yield from iter_model_examples_in_class(model_cls, model_name)
+
+
+def iter_model_examples_in_class(
+    model_cls: type[BaseModel], model_name: str | None = None
+) -> Iterator[ModelExample]:
+    """Iterates on all examples within a base model class
+
+    Usage:
+
+        @pytest.mark.parametrize(
+            "model_cls, example_name, example_data",
+            iter_model_examples_in_class(SomeModelClass),
+        )
+        def test_model_examples(
+            model_cls: type[BaseModel], example_name: str, example_data: Any
+        ):
+            assert_validation_model(
+                model_cls, example_name=example_name, example_data=example_data
+            )
+
+    """
+    assert issubclass(model_cls, BaseModel)  # nosec
 
-        if example := schema.get("example"):
+    if model_name is None:
+        model_name = f"{model_cls.__module__}.{model_cls.__name__}"
+
+    schema = model_cls.model_json_schema()
+
+    if example := schema.get("example"):
+        yield ModelExample(
+            model_cls=model_cls,
+            example_name=f"{model_name}_example",
+            example_data=example,
+        )
+
+    if many_examples := schema.get("examples"):
+        for index, example in enumerate(many_examples):
             yield ModelExample(
                 model_cls=model_cls,
-                example_name=f"{model_name}_example",
+                example_name=f"{model_name}_examples_{index}",
                 example_data=example,
             )
 
-        if many_examples := schema.get("examples"):
-            for index, example in enumerate(many_examples):
-                yield ModelExample(
-                    model_cls=model_cls,
-                    example_name=f"{model_name}_examples_{index}",
-                    example_data=example,
-                )
+
+TBaseModel = TypeVar("TBaseModel", bound=BaseModel)
 
 
 def assert_validation_model(
-    model_cls: type[BaseModel], example_name: int, example_data: Any
-):
+    model_cls: type[TBaseModel], example_name: str, example_data: Any
+) -> TBaseModel:
     try:
-        assert model_cls.model_validate(example_data) is not None
+        model_instance = model_cls.model_validate(example_data)
     except ValidationError as err:
         pytest.fail(
             f"{example_name} is invalid {model_cls.__module__}.{model_cls.__name__}:"
-            f"\n{json.dumps(example_data, indent=1)}"
+            f"\n{json_dumps(example_data, indent=1)}"
             f"\nError: {err}"
         )
 
+    assert isinstance(model_instance, model_cls)
+    return model_instance
+
 
 ## PYDANTIC MODELS & SCHEMAS -----------------------------------------------------
 
@@ -134,7 +170,12 @@ def model_cls_examples(model_cls: type[BaseModel]) -> dict[str, dict[str, Any]]:
     """
     Extracts examples from pydantic model class Config
     """
-
+    warnings.warn(
+        "The 'model_cls_examples' fixture is deprecated and will be removed in a future version. "
+        "Please use 'iter_model_example_in_class' or 'iter_model_examples_in_module' as an alternative.",
+        DeprecationWarning,
+        stacklevel=2,
+    )
     # Use by defining model_cls as test parametrization
     assert model_cls, (
         f"Testing against a {model_cls} model that has NO examples. Add them in Config class. "

packages/service-integration/tests/test_versioning.py

@@ -2,10 +2,17 @@
 # pylint: disable=unused-argument
 # pylint: disable=unused-variable
 
-import json
+
+import itertools
+from typing import Any
 
 import pytest
 from packaging.version import Version
+from pydantic import BaseModel
+from pytest_simcore.pydantic_models import (
+    assert_validation_model,
+    iter_model_examples_in_class,
+)
 from service_integration.versioning import (
     ExecutableVersionInfo,
     ServiceVersionInfo,
@@ -45,11 +52,15 @@ def test_bump_version_string(
 
 
 @pytest.mark.parametrize(
-    "model_cls",
-    [ExecutableVersionInfo, ServiceVersionInfo],
+    "model_cls, example_name, example_data",
+    itertools.chain(
+        iter_model_examples_in_class(ExecutableVersionInfo),
+        iter_model_examples_in_class(ServiceVersionInfo),
+    ),
 )
-def test_version_info_model_examples(model_cls, model_cls_examples):
-    for name, example in model_cls_examples.items():
-        print(name, ":", json.dumps(example, indent=1))
-        model_instance = model_cls(**example)
-        assert model_instance, f"Failed with {name}"
+def test_version_info_model_examples(
+    model_cls: type[BaseModel], example_name: str, example_data: Any
+):
+    assert_validation_model(
+        model_cls, example_name=example_name, example_data=example_data
+    )