🎨Improve testing on DB listener (#8019)

Author: sanderegg

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

@@ -10,11 +10,15 @@
 import pytest
 import sqlalchemy as sa
 from faker import Faker
+from models_library.products import ProductName
 from models_library.projects import ProjectAtDB, ProjectID
 from models_library.projects_nodes_io import NodeID
+from pytest_simcore.helpers.postgres_tools import insert_and_get_row_lifespan
 from simcore_postgres_database.models.comp_pipeline import StateType, comp_pipeline
 from simcore_postgres_database.models.comp_tasks import comp_tasks
+from simcore_postgres_database.models.products import products
 from simcore_postgres_database.models.projects import ProjectType, projects
+from simcore_postgres_database.models.projects_to_products import projects_to_products
 from simcore_postgres_database.models.services import services_access_rights
 from simcore_postgres_database.models.users import UserRole, UserStatus, users
 from simcore_postgres_database.utils_projects_nodes import (
@@ -63,9 +67,22 @@ def creator(**user_kwargs) -> dict[str, Any]:
     print(f"<-- deleted users {created_user_ids=}")
 
 
+@pytest.fixture
+async def product_db(
+    sqlalchemy_async_engine: AsyncEngine, product: dict[str, Any]
+) -> AsyncIterator[dict[str, Any]]:
+    async with insert_and_get_row_lifespan(  # pylint:disable=contextmanager-generator-missing-cleanup
+        sqlalchemy_async_engine,
+        table=products,
+        values=product,
+        pk_col=products.c.name,
+    ) as created_product:
+        yield created_product
+
+
 @pytest.fixture
 async def project(
-    sqlalchemy_async_engine: AsyncEngine, faker: Faker
+    sqlalchemy_async_engine: AsyncEngine, faker: Faker, product_name: ProductName
 ) -> AsyncIterator[Callable[..., Awaitable[ProjectAtDB]]]:
     created_project_ids: list[str] = []
 
@@ -113,6 +130,12 @@ async def creator(
                     for node_id in inserted_project.workbench
                 ],
             )
+            await con.execute(
+                projects_to_products.insert().values(
+                    project_uuid=f"{inserted_project.uuid}",
+                    product_name=product_name,
+                )
+            )
         print(f"--> created {inserted_project=}")
         created_project_ids.append(f"{inserted_project.uuid}")
         return inserted_project

services/director-v2/tests/conftest.py

@@ -43,6 +43,7 @@
     "pytest_simcore.docker_registry",
     "pytest_simcore.docker_swarm",
     "pytest_simcore.environment_configs",
+    "pytest_simcore.faker_products_data",
     "pytest_simcore.faker_projects_data",
     "pytest_simcore.faker_users_data",
     "pytest_simcore.minio_service",
@@ -355,7 +356,6 @@ async def wrapper(*args, **kwargs):
 
 @pytest.fixture
 def mock_osparc_variables_api_auth_rpc(mocker: MockerFixture) -> None:
-
     fake_data = ApiKeyGet.model_validate(ApiKeyGet.model_json_schema()["examples"][0])
 
     async def _create(

services/director-v2/tests/integration/01/test_computation_api.py

@@ -192,6 +192,7 @@ def test_invalid_computation(
 async def test_start_empty_computation_is_refused(
     async_client: httpx.AsyncClient,
     create_registered_user: Callable,
+    product_db: dict[str, Any],
     project: Callable[..., Awaitable[ProjectAtDB]],
     osparc_product_name: str,
     osparc_product_api_base_url: str,
@@ -397,6 +398,7 @@ async def test_run_partial_computation(
     wait_for_catalog_service: Callable[[UserID, str], Awaitable[None]],
     async_client: httpx.AsyncClient,
     create_registered_user: Callable,
+    product_db: dict[str, Any],
     project: Callable[..., Awaitable[ProjectAtDB]],
     update_project_workbench_with_comp_tasks: Callable,
     fake_workbench_without_outputs: dict[str, Any],
@@ -549,6 +551,7 @@ async def test_run_computation(
     wait_for_catalog_service: Callable[[UserID, str], Awaitable[None]],
     async_client: httpx.AsyncClient,
     create_registered_user: Callable,
+    product_db: dict[str, Any],
     project: Callable[..., Awaitable[ProjectAtDB]],
     fake_workbench_without_outputs: dict[str, Any],
     update_project_workbench_with_comp_tasks: Callable,
@@ -667,6 +670,7 @@ async def test_run_computation(
 async def test_abort_computation(
     async_client: httpx.AsyncClient,
     create_registered_user: Callable,
+    product_db: dict[str, Any],
     project: Callable[..., Awaitable[ProjectAtDB]],
     fake_workbench_without_outputs: dict[str, Any],
     fake_workbench_computational_pipeline_details: PipelineDetails,
@@ -746,6 +750,7 @@ async def test_abort_computation(
 async def test_update_and_delete_computation(
     async_client: httpx.AsyncClient,
     create_registered_user: Callable,
+    product_db: dict[str, Any],
     project: Callable[..., Awaitable[ProjectAtDB]],
     fake_workbench_without_outputs: dict[str, Any],
     fake_workbench_computational_pipeline_details_not_started: PipelineDetails,
@@ -874,6 +879,7 @@ async def test_update_and_delete_computation(
 async def test_pipeline_with_no_computational_services_still_create_correct_comp_tasks_in_db(
     async_client: httpx.AsyncClient,
     create_registered_user: Callable,
+    product_db: dict[str, Any],
     project: Callable[..., Awaitable[ProjectAtDB]],
     jupyter_service: dict[str, Any],
     osparc_product_name: str,
@@ -920,6 +926,7 @@ async def test_pipeline_with_no_computational_services_still_create_correct_comp
 async def test_pipeline_with_control_loop_made_of_dynamic_services_is_allowed(
     client: TestClient,
     create_registered_user: Callable,
+    product_db: dict[str, Any],
     project: Callable[..., Awaitable[ProjectAtDB]],
     jupyter_service: dict[str, Any],
     osparc_product_name: str,
@@ -991,6 +998,7 @@ async def test_pipeline_with_control_loop_made_of_dynamic_services_is_allowed(
 async def test_pipeline_with_cycle_containing_a_computational_service_is_forbidden(
     client: TestClient,
     create_registered_user: Callable,
+    product_db: dict[str, Any],
     project: Callable[..., Awaitable[ProjectAtDB]],
     sleeper_service: dict[str, Any],
     jupyter_service: dict[str, Any],
@@ -1075,6 +1083,7 @@ async def test_pipeline_with_cycle_containing_a_computational_service_is_forbidd
 async def test_burst_create_computations(
     async_client: httpx.AsyncClient,
     create_registered_user: Callable,
+    product_db: dict[str, Any],
     project: Callable[..., Awaitable[ProjectAtDB]],
     fake_workbench_without_outputs: dict[str, Any],
     update_project_workbench_with_comp_tasks: Callable,

services/director-v2/tests/integration/02/test_dynamic_services_routes.py

@@ -97,7 +97,9 @@ def user_id(user_db: dict[str, Any]) -> UserID:
 
 @pytest.fixture
 async def project_id(
-    user_db: dict[str, Any], project: Callable[..., Awaitable[ProjectAtDB]]
+    user_db: dict[str, Any],
+    project: Callable[..., Awaitable[ProjectAtDB]],
+    product_db: dict[str, Any],
 ) -> str:
     prj = await project(user=user_db)
     return f"{prj.uuid}"

services/director-v2/tests/integration/02/test_dynamic_sidecar_nodeports_integration.py

@@ -28,6 +28,7 @@
 )
 from models_library.api_schemas_directorv2.computations import ComputationGet
 from models_library.clusters import ClusterAuthentication
+from models_library.products import ProductName
 from models_library.projects import (
     Node,
     NodesDict,
@@ -164,6 +165,7 @@ async def minimal_configuration(
     ensure_swarm_and_networks: None,
     minio_s3_settings_envs: EnvVarsDict,
     current_user: dict[str, Any],
+    product_db: dict[str, Any],
     osparc_product_name: str,
 ) -> AsyncIterator[None]:
     await wait_for_catalog_service(current_user["id"], osparc_product_name)
@@ -917,6 +919,13 @@ async def _assert_retrieve_completed(
                 ), "TIP: Message missing suggests that the data was never uploaded: look in services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/modules/nodeports.py"
 
 
+def product_name(osparc_product_name: ProductName) -> ProductName:
+    """
+    override the product name to be used in these tests
+    """
+    return osparc_product_name
+
+
 @pytest.mark.flaky(max_runs=3)
 async def test_nodeports_integration(
     cleanup_services_and_networks: None,

services/director-v2/tests/integration/02/test_mixed_dynamic_sidecar_and_legacy_project.py

@@ -139,6 +139,7 @@ def user_dict(create_registered_user: Callable) -> dict[str, Any]:
 async def dy_static_file_server_project(
     minimal_configuration: None,
     user_dict: dict[str, Any],
+    product_db: dict[str, Any],
     project: Callable[..., Awaitable[ProjectAtDB]],
     dy_static_file_server_service: dict,
     dy_static_file_server_dynamic_sidecar_service: dict,

services/director-v2/tests/unit/with_dbs/comp_scheduler/test_api_route_computations.py

@@ -88,6 +88,7 @@ def minimal_configuration(
     faker: Faker,
     with_disabled_auto_scheduling: mock.Mock,
     with_disabled_scheduler_publisher: mock.Mock,
+    product_db: dict[str, Any],
 ):
     monkeypatch.setenv("DIRECTOR_V2_DYNAMIC_SIDECAR_ENABLED", "false")
     monkeypatch.setenv("COMPUTATIONAL_BACKEND_DASK_CLIENT_ENABLED", "1")
@@ -363,11 +364,6 @@ def _mocked_get_pricing_unit(request, pricing_plan_id: int) -> httpx.Response:
         yield respx_mock
 
 
-@pytest.fixture
-def product_name(faker: Faker) -> str:
-    return faker.name()
-
-
 @pytest.fixture
 def product_api_base_url(faker: Faker) -> AnyHttpUrl:
     return TypeAdapter(AnyHttpUrl).validate_python(faker.url())
@@ -379,7 +375,7 @@ async def test_computation_create_validators(
     fake_workbench_without_outputs: dict[str, Any],
     product_name: str,
     product_api_base_url: AnyHttpUrl,
-    faker: Faker,
+    product_db: dict[str, Any],
 ):
     user = create_registered_user()
     proj = await project(user, workbench=fake_workbench_without_outputs)
@@ -911,7 +907,7 @@ async def test_get_computation_from_not_started_computation_task(
             node_states={
                 t.node_id: NodeState(
                     modified=True,
-                    currentStatus=RunningState.NOT_STARTED,
+                    current_status=RunningState.NOT_STARTED,
                     progress=None,
                     dependencies={
                         NodeID(node)
@@ -983,7 +979,7 @@ async def test_get_computation_from_published_computation_task(
             node_states={
                 t.node_id: NodeState(
                     modified=True,
-                    currentStatus=RunningState.PUBLISHED,
+                    current_status=RunningState.PUBLISHED,
                     dependencies={
                         NodeID(node)
                         for node, next_nodes in fake_workbench_adjacency.items()

services/director-v2/tests/unit/with_dbs/comp_scheduler/test_api_route_computations_tasks.py

@@ -120,6 +120,7 @@ async def project_id(
     project: Callable[..., Awaitable[ProjectAtDB]],
     create_pipeline: Callable[..., Awaitable[CompPipelineAtDB]],
     create_tasks: Callable[..., Awaitable[list[CompTaskAtDB]]],
+    product_db: dict[str, Any],
 ) -> ProjectID:
     """project uuid of a saved project (w/ tasks up-to-date)"""
 
@@ -133,6 +134,7 @@ async def project_id(
     )
     # insert tasks -> comp_tasks
     comp_tasks = await create_tasks(user=user, project=proj)
+    assert comp_tasks
 
     return proj.uuid
 
@@ -166,7 +168,7 @@ async def test_get_all_tasks_log_files(
     assert resp.status_code == status.HTTP_200_OK
     log_files = TypeAdapter(list[TaskLogFileGet]).validate_json(resp.text)
     assert log_files
-    assert all(l.download_link for l in log_files)
+    assert all(file.download_link for file in log_files)
 
 
 async def test_get_task_logs_file(

services/director-v2/tests/unit/with_dbs/comp_scheduler/test_api_rpc_computations.py

@@ -40,6 +40,7 @@ async def test_rpc_list_computation_runs_and_tasks(
     create_tasks: Callable[..., Awaitable[list[CompTaskAtDB]]],
     create_comp_run: Callable[..., Awaitable[CompRunsAtDB]],
     rpc_client: RabbitMQRPCClient,
+    product_db: dict[str, Any],
 ):
     user = create_registered_user()
     proj = await project(user, workbench=fake_workbench_without_outputs)
@@ -119,6 +120,7 @@ async def test_rpc_list_computation_runs_with_filtering(
     create_tasks: Callable[..., Awaitable[list[CompTaskAtDB]]],
     create_comp_run: Callable[..., Awaitable[CompRunsAtDB]],
     rpc_client: RabbitMQRPCClient,
+    product_db: dict[str, Any],
 ):
     user = create_registered_user()
 
@@ -175,6 +177,7 @@ async def test_rpc_list_computation_runs_history(
     create_tasks: Callable[..., Awaitable[list[CompTaskAtDB]]],
     create_comp_run: Callable[..., Awaitable[CompRunsAtDB]],
     rpc_client: RabbitMQRPCClient,
+    product_db: dict[str, Any],
 ):
     user = create_registered_user()
 
@@ -186,6 +189,7 @@ async def test_rpc_list_computation_runs_history(
     comp_tasks = await create_tasks(
         user=user, project=proj, state=StateType.PUBLISHED, progress=None
     )
+    assert comp_tasks
     comp_runs_1 = await create_comp_run(
         user=user,
         project=proj,
@@ -195,6 +199,7 @@ async def test_rpc_list_computation_runs_history(
         iteration=1,
         dag_adjacency_list=fake_workbench_adjacency,
     )
+    assert comp_runs_1
     comp_runs_2 = await create_comp_run(
         user=user,
         project=proj,
@@ -204,6 +209,7 @@ async def test_rpc_list_computation_runs_history(
         iteration=2,
         dag_adjacency_list=fake_workbench_adjacency,
     )
+    assert comp_runs_2
     comp_runs_3 = await create_comp_run(
         user=user,
         project=proj,
@@ -213,6 +219,7 @@ async def test_rpc_list_computation_runs_history(
         iteration=3,
         dag_adjacency_list=fake_workbench_adjacency,
     )
+    assert comp_runs_3
 
     output = await rpc_computations.list_computations_iterations_page(
         rpc_client, product_name="osparc", user_id=user["id"], project_ids=[proj.uuid]

services/director-v2/tests/unit/with_dbs/comp_scheduler/test_db_repositories_comp_runs.py

@@ -9,7 +9,7 @@
 import datetime
 import random
 from collections.abc import Awaitable, Callable
-from typing import cast
+from typing import Any, cast
 
 import arrow
 import pytest
@@ -56,6 +56,7 @@ async def test_get(
     fake_project_id: ProjectID,
     publish_project: Callable[[], Awaitable[PublishedProject]],
     create_comp_run: Callable[..., Awaitable[CompRunsAtDB]],
+    product_db: dict[str, Any],
 ):
     with pytest.raises(ComputationalRunNotFoundError):
         await CompRunsRepository(sqlalchemy_async_engine).get(
@@ -88,6 +89,7 @@ async def test_list(
     publish_project: Callable[[], Awaitable[PublishedProject]],
     run_metadata: RunMetadataDict,
     faker: Faker,
+    product_db: dict[str, Any],
 ):
     assert await CompRunsRepository(sqlalchemy_async_engine).list_() == []
 
@@ -269,6 +271,7 @@ async def test_create(
     run_metadata: RunMetadataDict,
     faker: Faker,
     publish_project: Callable[[], Awaitable[PublishedProject]],
+    product_db: dict[str, Any],
 ):
     with pytest.raises(ProjectNotFoundError):
         await CompRunsRepository(sqlalchemy_async_engine).create(
@@ -331,6 +334,7 @@ async def test_update(
     run_metadata: RunMetadataDict,
     faker: Faker,
     publish_project: Callable[[], Awaitable[PublishedProject]],
+    product_db: dict[str, Any],
 ):
     # this updates nothing but also does not complain
     updated = await CompRunsRepository(sqlalchemy_async_engine).update(
@@ -371,6 +375,7 @@ async def test_set_run_result(
     run_metadata: RunMetadataDict,
     faker: Faker,
     publish_project: Callable[[], Awaitable[PublishedProject]],
+    product_db: dict[str, Any],
 ):
     published_project = await publish_project()
     created = await CompRunsRepository(sqlalchemy_async_engine).create(
@@ -419,6 +424,7 @@ async def test_mark_for_cancellation(
     run_metadata: RunMetadataDict,
     faker: Faker,
     publish_project: Callable[[], Awaitable[PublishedProject]],
+    product_db: dict[str, Any],
 ):
     published_project = await publish_project()
     created = await CompRunsRepository(sqlalchemy_async_engine).create(
@@ -451,6 +457,7 @@ async def test_mark_for_scheduling(
     run_metadata: RunMetadataDict,
     faker: Faker,
     publish_project: Callable[[], Awaitable[PublishedProject]],
+    product_db: dict[str, Any],
 ):
     published_project = await publish_project()
     created = await CompRunsRepository(sqlalchemy_async_engine).create(
@@ -485,6 +492,7 @@ async def test_mark_scheduling_done(
     run_metadata: RunMetadataDict,
     faker: Faker,
     publish_project: Callable[[], Awaitable[PublishedProject]],
+    product_db: dict[str, Any],
 ):
     published_project = await publish_project()
     created = await CompRunsRepository(sqlalchemy_async_engine).create(