added fixture and refactor tests

Author: Andrei Neagu

services/director-v2/tests/conftest.py

@@ -24,6 +24,7 @@
 from models_library.projects import Node, NodesDict
 from models_library.rpc.webserver.auth.api_keys import ApiKeyGet
 from models_library.users import UserID
+from pydantic import PositiveInt
 from pytest_mock import MockerFixture
 from pytest_simcore.helpers.monkeypatch_envs import (
     setenvs_from_dict,
@@ -364,3 +365,8 @@ async def _create(
         side_effect=_create,
         autospec=True,
     )
+
+
+@pytest.fixture
+def comp_task_run_id() -> PositiveInt:
+    return 42

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

@@ -46,7 +46,7 @@
 from models_library.projects_nodes_io import NodeID
 from models_library.resource_tracker import HardwareInfo
 from models_library.users import UserID
-from pydantic import AnyUrl, ByteSize, TypeAdapter
+from pydantic import AnyUrl, ByteSize, PositiveInt, TypeAdapter
 from pytest_mock.plugin import MockerFixture
 from pytest_simcore.helpers.typing_env import EnvVarsDict
 from settings_library.s3 import S3Settings
@@ -442,6 +442,7 @@ async def test_send_computation_task(
     task_labels: ContainerLabelsDict,
     empty_hardware_info: HardwareInfo,
     faker: Faker,
+    comp_task_run_id: PositiveInt,
 ):
     _DASK_EVENT_NAME = faker.pystr()
 
@@ -503,7 +504,7 @@ def fake_sidecar_fct(
         ),
         metadata=comp_run_metadata,
         hardware_info=empty_hardware_info,
-        run_id=1,
+        run_id=comp_task_run_id,
     )
     assert node_id_to_job_ids
     assert len(node_id_to_job_ids) == 1
@@ -560,6 +561,7 @@ async def test_computation_task_is_persisted_on_dask_scheduler(
     mocked_storage_service_api: respx.MockRouter,
     comp_run_metadata: RunMetadataDict,
     empty_hardware_info: HardwareInfo,
+    comp_task_run_id: PositiveInt,
 ):
     """rationale:
     When a task is submitted to the dask backend, a dask future is returned.
@@ -595,7 +597,7 @@ def fake_sidecar_fct(
         remote_fct=fake_sidecar_fct,
         metadata=comp_run_metadata,
         hardware_info=empty_hardware_info,
-        run_id=1,
+        run_id=comp_task_run_id,
     )
     assert published_computation_task
     assert len(published_computation_task) == 1
@@ -651,6 +653,7 @@ async def test_abort_computation_tasks(
     faker: Faker,
     comp_run_metadata: RunMetadataDict,
     empty_hardware_info: HardwareInfo,
+    comp_task_run_id: PositiveInt,
 ):
     _DASK_EVENT_NAME = faker.pystr()
 
@@ -689,6 +692,7 @@ def fake_remote_fct(
         remote_fct=fake_remote_fct,
         metadata=comp_run_metadata,
         hardware_info=empty_hardware_info,
+        run_id=comp_task_run_id,
     )
     assert published_computation_task
     assert len(published_computation_task) == 1
@@ -740,6 +744,7 @@ async def test_failed_task_returns_exceptions(
     mocked_storage_service_api: respx.MockRouter,
     comp_run_metadata: RunMetadataDict,
     empty_hardware_info: HardwareInfo,
+    comp_task_run_id: PositiveInt,
 ):
     # NOTE: this must be inlined so that the test works,
     # the dask-worker must be able to import the function
@@ -760,6 +765,7 @@ def fake_failing_sidecar_fct(
         remote_fct=fake_failing_sidecar_fct,
         metadata=comp_run_metadata,
         hardware_info=empty_hardware_info,
+        run_id=comp_task_run_id,
     )
     assert published_computation_task
     assert len(published_computation_task) == 1
@@ -802,6 +808,7 @@ async def test_send_computation_task_with_missing_resources_raises(
     mocked_storage_service_api: respx.MockRouter,
     comp_run_metadata: RunMetadataDict,
     empty_hardware_info: HardwareInfo,
+    comp_task_run_id: PositiveInt,
 ):
     # remove the workers that can handle gpu
     scheduler_info = dask_client.backend.client.scheduler_info()
@@ -828,6 +835,7 @@ async def test_send_computation_task_with_missing_resources_raises(
             remote_fct=None,
             metadata=comp_run_metadata,
             hardware_info=empty_hardware_info,
+            run_id=comp_task_run_id,
         )
     mocked_user_completed_cb.assert_not_called()
 
@@ -846,6 +854,7 @@ async def test_send_computation_task_with_hardware_info_raises(
     mocked_storage_service_api: respx.MockRouter,
     comp_run_metadata: RunMetadataDict,
     hardware_info: HardwareInfo,
+    comp_task_run_id: PositiveInt,
 ):
     # NOTE: running on the default cluster will raise missing resources
     with pytest.raises(MissingComputationalResourcesError):
@@ -857,6 +866,7 @@ async def test_send_computation_task_with_hardware_info_raises(
             remote_fct=None,
             metadata=comp_run_metadata,
             hardware_info=hardware_info,
+            run_id=comp_task_run_id,
         )
     mocked_user_completed_cb.assert_not_called()
 
@@ -874,6 +884,7 @@ async def test_too_many_resources_send_computation_task(
     mocked_storage_service_api: respx.MockRouter,
     comp_run_metadata: RunMetadataDict,
     empty_hardware_info: HardwareInfo,
+    comp_task_run_id: PositiveInt,
 ):
     # create an image that needs a huge amount of CPU
     image = Image(
@@ -897,6 +908,7 @@ async def test_too_many_resources_send_computation_task(
             remote_fct=None,
             metadata=comp_run_metadata,
             hardware_info=empty_hardware_info,
+            run_id=comp_task_run_id,
         )
 
     mocked_user_completed_cb.assert_not_called()
@@ -913,6 +925,7 @@ async def test_disconnected_backend_raises_exception(
     mocked_storage_service_api: respx.MockRouter,
     comp_run_metadata: RunMetadataDict,
     empty_hardware_info: HardwareInfo,
+    comp_task_run_id: PositiveInt,
 ):
     # DISCONNECT THE CLUSTER
     await dask_spec_local_cluster.close()  # type: ignore
@@ -925,6 +938,7 @@ async def test_disconnected_backend_raises_exception(
             remote_fct=None,
             metadata=comp_run_metadata,
             hardware_info=empty_hardware_info,
+            run_id=comp_task_run_id,
         )
     mocked_user_completed_cb.assert_not_called()
 
@@ -944,6 +958,7 @@ async def test_changed_scheduler_raises_exception(
     unused_tcp_port_factory: Callable,
     comp_run_metadata: RunMetadataDict,
     empty_hardware_info: HardwareInfo,
+    comp_task_run_id: PositiveInt,
 ):
     # change the scheduler (stop the current one and start another at the same address)
     scheduler_address = URL(dask_spec_local_cluster.scheduler_address)
@@ -973,6 +988,7 @@ async def test_changed_scheduler_raises_exception(
                 remote_fct=None,
                 metadata=comp_run_metadata,
                 hardware_info=empty_hardware_info,
+                run_id=comp_task_run_id,
             )
     mocked_user_completed_cb.assert_not_called()
 
@@ -990,6 +1006,7 @@ async def test_get_tasks_status(
     fail_remote_fct: bool,
     comp_run_metadata: RunMetadataDict,
     empty_hardware_info: HardwareInfo,
+    comp_task_run_id: PositiveInt,
 ):
     # NOTE: this must be inlined so that the test works,
     # the dask-worker must be able to import the function
@@ -1017,6 +1034,7 @@ def fake_remote_fct(
         remote_fct=fake_remote_fct,
         metadata=comp_run_metadata,
         hardware_info=empty_hardware_info,
+        run_id=comp_task_run_id,
     )
     assert published_computation_task
     assert len(published_computation_task) == 1
@@ -1071,6 +1089,7 @@ async def test_dask_sub_handlers(
     fake_task_handlers: TaskHandlers,
     comp_run_metadata: RunMetadataDict,
     empty_hardware_info: HardwareInfo,
+    comp_task_run_id: PositiveInt,
 ):
     dask_client.register_handlers(fake_task_handlers)
     _DASK_START_EVENT = "start"
@@ -1100,6 +1119,7 @@ def fake_remote_fct(
         remote_fct=fake_remote_fct,
         metadata=comp_run_metadata,
         hardware_info=empty_hardware_info,
+        run_id=comp_task_run_id,
     )
     assert published_computation_task
     assert len(published_computation_task) == 1
@@ -1144,6 +1164,7 @@ async def test_get_cluster_details(
     comp_run_metadata: RunMetadataDict,
     empty_hardware_info: HardwareInfo,
     faker: Faker,
+    comp_task_run_id: PositiveInt,
 ):
     cluster_details = await dask_client.get_cluster_details()
     assert cluster_details
@@ -1180,6 +1201,7 @@ def fake_sidecar_fct(
         ),
         metadata=comp_run_metadata,
         hardware_info=empty_hardware_info,
+        run_id=comp_task_run_id,
     )
     assert published_computation_task
     assert len(published_computation_task) == 1

services/director-v2/tests/unit/with_dbs/test_utils_dask.py

@@ -35,7 +35,7 @@
 from models_library.projects import ProjectID
 from models_library.projects_nodes_io import NodeID, SimCoreFileLink, SimcoreS3FileID
 from models_library.users import UserID
-from pydantic import ByteSize, TypeAdapter
+from pydantic import ByteSize, PositiveInt, TypeAdapter
 from pydantic.networks import AnyUrl
 from pytest_mock.plugin import MockerFixture
 from pytest_simcore.helpers.typing_env import EnvVarsDict
@@ -647,6 +647,7 @@ async def test_compute_task_envs(
     run_metadata: RunMetadataDict,
     input_task_envs: ContainerEnvsDict,
     expected_computed_task_envs: ContainerEnvsDict,
+    comp_task_run_id: PositiveInt,
 ):
     sleeper_task: CompTaskAtDB = published_project.tasks[1]
     sleeper_task.image.envs = input_task_envs
@@ -658,6 +659,6 @@ async def test_compute_task_envs(
         node_id=sleeper_task.node_id,
         node_image=sleeper_task.image,
         metadata=run_metadata,
-        run_id=1,
+        run_id=comp_task_run_id,
     )
     assert task_envs == expected_computed_task_envs