fix some tests

Author: sanderegg

services/director-v2/src/simcore_service_director_v2/modules/dask_client.py

@@ -359,7 +359,7 @@ async def send_computation_tasks(
             try:
                 # This instance is created only once so it can be reused in calls below
                 node_ports = await dask_utils.create_node_ports(
-                    db_engine=self.app.state.engine,
+                    db_engine=self.app.state.asyncpg_engine,
                     user_id=user_id,
                     project_id=project_id,
                     node_id=node_id,
@@ -439,14 +439,14 @@ async def get_tasks_status(self, job_ids: list[str]) -> list[DaskClientTaskState
         def _get_pipeline_statuses(
             dask_scheduler: distributed.Scheduler,
         ) -> dict[dask.typing.Key, DaskSchedulerTaskState | None]:
-            statuses: dict[
-                dask.typing.Key, DaskSchedulerTaskState | None
-            ] = dask_scheduler.get_task_status(keys=job_ids)
+            statuses: dict[dask.typing.Key, DaskSchedulerTaskState | None] = (
+                dask_scheduler.get_task_status(keys=job_ids)
+            )
             return statuses
 
-        task_statuses: dict[
-            dask.typing.Key, DaskSchedulerTaskState | None
-        ] = await self.backend.client.run_on_scheduler(_get_pipeline_statuses)
+        task_statuses: dict[dask.typing.Key, DaskSchedulerTaskState | None] = (
+            await self.backend.client.run_on_scheduler(_get_pipeline_statuses)
+        )
         assert isinstance(task_statuses, dict)  # nosec
 
         _logger.debug("found dask task statuses: %s", f"{task_statuses=}")
@@ -578,10 +578,10 @@ def _get_worker_used_resources(
 
         with log_catch(_logger, reraise=False):
             # NOTE: this runs directly on the dask-scheduler and may rise exceptions
-            used_resources_per_worker: dict[
-                str, dict[str, Any]
-            ] = await dask_utils.wrap_client_async_routine(
-                self.backend.client.run_on_scheduler(_get_worker_used_resources)
+            used_resources_per_worker: dict[str, dict[str, Any]] = (
+                await dask_utils.wrap_client_async_routine(
+                    self.backend.client.run_on_scheduler(_get_worker_used_resources)
+                )
             )
 
             # let's update the scheduler info, with default to 0s since sometimes

services/director-v2/tests/unit/conftest.py

@@ -48,6 +48,7 @@
 def disable_postgres(mocker) -> None:
     def mock_setup(app: FastAPI, *args, **kwargs) -> None:
         app.state.engine = mock.AsyncMock()
+        app.state.asyncpg_engine = mock.AsyncMock()
 
     mocker.patch("simcore_service_director_v2.modules.db.setup", side_effect=mock_setup)
 
@@ -152,7 +153,7 @@ def mock_service_inspect(
 
 @pytest.fixture
 def scheduler_data_from_service_inspect(
-    mock_service_inspect: Mapping[str, Any]
+    mock_service_inspect: Mapping[str, Any],
 ) -> SchedulerData:
     return SchedulerData.from_service_inspect(mock_service_inspect)
 

services/director-v2/tests/unit/test_modules_dask_client.py

@@ -189,7 +189,7 @@ async def dask_client(
     }[request.param]()
 
     try:
-        assert client.app.state.engine is not None
+        assert client.app.state.asyncpg_engine is not None
 
         # check we can run some simple python script
         def _square(x):
@@ -207,8 +207,8 @@ def neg(x):
         result = await future
         assert result == -285
     except AttributeError:
-        # enforces existance of 'app.state.engine' and sets to None
-        client.app.state.engine = None
+        # enforces existance of 'app.state.asyncpg_engine' and sets to None
+        client.app.state.asyncpg_engine = None
 
     return client
 
@@ -390,7 +390,9 @@ def fct_that_raise_exception():
     )  # type: ignore
     assert task_exception
     assert isinstance(task_exception, exc)
-    task_traceback = await future.traceback(timeout=_ALLOW_TIME_FOR_GATEWAY_TO_CREATE_WORKERS)  # type: ignore
+    task_traceback = await future.traceback(
+        timeout=_ALLOW_TIME_FOR_GATEWAY_TO_CREATE_WORKERS
+    )  # type: ignore
     assert task_traceback
     trace = traceback.format_exception(task_exception)
     assert trace
@@ -625,7 +627,9 @@ def fake_sidecar_fct(
     assert published_computation_task[0].job_id in list_of_persisted_datasets
     assert list_of_persisted_datasets[0] == published_computation_task[0].job_id
     # get the persisted future from the scheduler back
-    task_future = await dask_client.backend.client.get_dataset(name=published_computation_task[0].job_id)  # type: ignore
+    task_future = await dask_client.backend.client.get_dataset(
+        name=published_computation_task[0].job_id
+    )  # type: ignore
     assert task_future
     assert isinstance(task_future, distributed.Future)
     assert task_future.key == published_computation_task[0].job_id

services/director-v2/tests/unit/test_modules_osparc_variables.py

@@ -280,7 +280,6 @@ def compose_spec():
 def test_auto_inject_environments_added_to_all_services_in_compose(
     compose_spec: ComposeSpecLabelDict,
 ):
-
     before = deepcopy(compose_spec)
 
     after = auto_inject_environments(compose_spec)
@@ -290,7 +289,6 @@ def test_auto_inject_environments_added_to_all_services_in_compose(
 
     auto_injected_envs = set(_NEW_ENVIRONMENTS.keys())
     for name, service in compose_spec.get("services", {}).items():
-
         # all services have environment specs
         assert service["environment"], f"expected in {name} service"