improve unit tests

Author: matusdrobuliak66

packages/postgres-database/src/simcore_postgres_database/utils_comp_run_snapshot_tasks.py

@@ -55,6 +55,6 @@ async def update_for_run_id_and_node_id(
         )
         row = await result.one_or_none()
         if row is None:
-            raise ValueError("improve message")
-            # return None
+            msg = f"update for run_id={run_id} and node_id={node_id} did not return any row"
+            raise ValueError(msg)
         return row

packages/postgres-database/src/simcore_postgres_database/utils_comp_runs.py

@@ -68,5 +68,6 @@ async def get_latest_run_id_for_project(
         result = await _conn.execute(base_select_query)
         row = result.one_or_none()
         if not row:
-            raise ValueError("improve")
+            msg = f"get_latest_run_id_for_project did not return any row for project_id={project_id}"
+            raise ValueError(msg)
         return cast(PositiveInt, row.run_id)

services/director-v2/src/simcore_service_director_v2/api/routes/computations.py

@@ -199,7 +199,7 @@ async def _try_start_pipeline(
     project: ProjectAtDB,
     users_repo: UsersRepository,
     projects_metadata_repo: ProjectsMetadataRepository,
-    filtered_comp_tasks_in_db: list[CompTaskAtDB],
+    tasks_to_run: list[CompTaskAtDB],
 ) -> None:
     if not minimal_dag.nodes():
         # 2 options here: either we have cycles in the graph or it's really done
@@ -242,7 +242,7 @@ async def _try_start_pipeline(
         )
         or {},
         use_on_demand_clusters=computation.use_on_demand_clusters,
-        filtered_comp_tasks_in_db=filtered_comp_tasks_in_db,
+        tasks_to_run=tasks_to_run,
     )
 
 
@@ -359,7 +359,7 @@ async def create_computation(  # noqa: PLR0913 # pylint: disable=too-many-positi
                 project=project,
                 users_repo=users_repo,
                 projects_metadata_repo=projects_metadata_repo,
-                filtered_comp_tasks_in_db=filtered_tasks,
+                tasks_to_run=filtered_tasks,
             )
 
         pipeline_state = utils.get_pipeline_state_from_task_states(filtered_tasks)

services/director-v2/src/simcore_service_director_v2/modules/comp_scheduler/_manager.py

@@ -41,7 +41,7 @@ async def run_new_pipeline(
     project_id: ProjectID,
     run_metadata: RunMetadataDict,
     use_on_demand_clusters: bool,
-    filtered_comp_tasks_in_db: list[CompTaskAtDB],
+    tasks_to_run: list[CompTaskAtDB],
 ) -> None:
     """Sets a new pipeline to be scheduled on the computational resources."""
     # ensure the pipeline exists and is populated with something
@@ -69,7 +69,7 @@ async def run_new_pipeline(
             "run_id": new_run.run_id,
             # "submit": datetime.fromisoformat(task.submit)
         }
-        for task in filtered_comp_tasks_in_db
+        for task in tasks_to_run
     ]
     await CompRunsSnapshotTasksRepository.instance(db_engine).batch_create(
         data=db_create_snaphot_tasks

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

@@ -21,6 +21,7 @@ class PublishedProject:
     project: ProjectAtDB
     pipeline: CompPipelineAtDB
     tasks: list[CompTaskAtDB]
+    tasks_to_run: list[CompTaskAtDB]
 
 
 @dataclass(kw_only=True)

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

@@ -156,7 +156,7 @@ async def test_schedule_all_pipelines(
         project_id=published_project.project.uuid,
         run_metadata=run_metadata,
         use_on_demand_clusters=False,
-        filtered_comp_tasks_in_db=[],
+        tasks_to_run=published_project.tasks_to_run,
     )
     # this directly schedule a new pipeline
     scheduler_rabbit_client_parser.assert_called_once_with(
@@ -258,7 +258,7 @@ async def test_schedule_all_pipelines_logs_error_if_it_find_old_pipelines(
         project_id=published_project.project.uuid,
         run_metadata=run_metadata,
         use_on_demand_clusters=False,
-        filtered_comp_tasks_in_db=[],
+        tasks_to_run=published_project.tasks_to_run,
     )
     # this directly schedule a new pipeline
     scheduler_rabbit_client_parser.assert_called_once_with(
@@ -342,7 +342,7 @@ async def test_empty_pipeline_is_not_scheduled(
             project_id=empty_project.uuid,
             run_metadata=run_metadata,
             use_on_demand_clusters=False,
-            filtered_comp_tasks_in_db=[],
+            tasks_to_run=[],
         )
     await assert_comp_runs_empty(sqlalchemy_async_engine)
     scheduler_rabbit_client_parser.assert_not_called()
@@ -358,7 +358,7 @@ async def test_empty_pipeline_is_not_scheduled(
             project_id=empty_project.uuid,
             run_metadata=run_metadata,
             use_on_demand_clusters=False,
-            filtered_comp_tasks_in_db=[],
+            tasks_to_run=[],
         )
     assert len(caplog.records) == 1
     assert "no computational dag defined" in caplog.records[0].message

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

@@ -169,7 +169,7 @@ async def _assert_start_pipeline(
         project_id=published_project.project.uuid,
         run_metadata=run_metadata,
         use_on_demand_clusters=False,
-        filtered_comp_tasks_in_db=[],
+        tasks_to_run=published_project.tasks_to_run,
     )
 
     # check the database is correctly updated, the run is published
@@ -1125,7 +1125,7 @@ async def test_broken_pipeline_configuration_is_not_scheduled_and_aborted(
         project_id=sleepers_project.uuid,
         run_metadata=run_metadata,
         use_on_demand_clusters=False,
-        filtered_comp_tasks_in_db=[],
+        tasks_to_run=[],
     )
     with_disabled_scheduler_publisher.assert_called_once()
     # we shall have a a new comp_runs row with the new pipeline job
@@ -1253,7 +1253,7 @@ async def test_handling_of_disconnected_scheduler_dask(
         project_id=published_project.project.uuid,
         run_metadata=run_metadata,
         use_on_demand_clusters=False,
-        filtered_comp_tasks_in_db=[],
+        tasks_to_run=published_project.tasks_to_run,
     )
 
     # since there is no cluster, there is no dask-scheduler,
@@ -1769,7 +1769,7 @@ async def test_pipeline_with_on_demand_cluster_with_not_ready_backend_waits(
         project_id=published_project.project.uuid,
         run_metadata=run_metadata,
         use_on_demand_clusters=True,
-        filtered_comp_tasks_in_db=[],
+        tasks_to_run=published_project.tasks_to_run,
     )
 
     # we ask to use an on-demand cluster, therefore the tasks are published first
@@ -1874,7 +1874,7 @@ async def test_pipeline_with_on_demand_cluster_with_no_clusters_keeper_fails(
         project_id=published_project.project.uuid,
         run_metadata=run_metadata,
         use_on_demand_clusters=True,
-        filtered_comp_tasks_in_db=[],
+        tasks_to_run=published_project.tasks_to_run,
     )
 
     # we ask to use an on-demand cluster, therefore the tasks are published first

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

@@ -69,7 +69,7 @@ async def test_worker_properly_autocalls_scheduler_api(
         project_id=published_project.project.uuid,
         run_metadata=run_metadata,
         use_on_demand_clusters=False,
-        filtered_comp_tasks_in_db=[],
+        tasks_to_run=published_project.tasks_to_run,
     )
     mocked_get_scheduler_worker.assert_called_once_with(initialized_app)
     mocked_get_scheduler_worker.return_value.apply.assert_called_once_with(
@@ -127,7 +127,7 @@ async def _project_pipeline_creation_workflow() -> None:
             project_id=published_project.project.uuid,
             run_metadata=run_metadata,
             use_on_demand_clusters=False,
-            filtered_comp_tasks_in_db=[],
+            tasks_to_run=published_project.tasks_to_run,
         )
 
     # whatever scheduling concurrency we call in here, we shall always see the same number of calls to the scheduler

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

@@ -22,7 +22,7 @@
 from pydantic.main import BaseModel
 from simcore_postgres_database.models.comp_pipeline import StateType, comp_pipeline
 from simcore_postgres_database.models.comp_runs import comp_runs
-from simcore_postgres_database.models.comp_tasks import comp_tasks
+from simcore_postgres_database.models.comp_tasks import NodeClass, comp_tasks
 from simcore_service_director_v2.models.comp_pipelines import CompPipelineAtDB
 from simcore_service_director_v2.models.comp_runs import (
     CompRunsAtDB,
@@ -231,16 +231,23 @@ async def publish_project(
 
     async def _() -> PublishedProject:
         created_project = await project(user, workbench=fake_workbench_without_outputs)
+        created_pipeline = await create_pipeline(
+            project_id=f"{created_project.uuid}",
+            dag_adjacency_list=fake_workbench_adjacency,
+        )
+        created_tasks = await create_tasks(
+            user=user, project=created_project, state=StateType.PUBLISHED
+        )
+        tasks_to_run = [
+            t for t in created_tasks if t.node_class == NodeClass.COMPUTATIONAL
+        ]
+
         return PublishedProject(
             user=user,
             project=created_project,
-            pipeline=await create_pipeline(
-                project_id=f"{created_project.uuid}",
-                dag_adjacency_list=fake_workbench_adjacency,
-            ),
-            tasks=await create_tasks(
-                user=user, project=created_project, state=StateType.PUBLISHED
-            ),
+            pipeline=created_pipeline,
+            tasks=created_tasks,
+            tasks_to_run=tasks_to_run,
         )
 
     return _
@@ -266,27 +273,31 @@ async def running_project(
     user = create_registered_user()
     created_project = await project(user, workbench=fake_workbench_without_outputs)
     now_time = arrow.utcnow().datetime
+    created_tasks = await create_tasks(
+        user=user,
+        project=created_project,
+        state=StateType.RUNNING,
+        progress=0.0,
+        start=now_time,
+    )
+    tasks_to_run = [t for t in created_tasks if t.node_class == NodeClass.COMPUTATIONAL]
+
     return RunningProject(
         user=user,
         project=created_project,
         pipeline=await create_pipeline(
             project_id=f"{created_project.uuid}",
             dag_adjacency_list=fake_workbench_adjacency,
         ),
-        tasks=await create_tasks(
-            user=user,
-            project=created_project,
-            state=StateType.RUNNING,
-            progress=0.0,
-            start=now_time,
-        ),
+        tasks=created_tasks,
         runs=await create_comp_run(
             user=user,
             project=created_project,
             started=now_time,
             result=StateType.RUNNING,
         ),
         task_to_callback_mapping={},
+        tasks_to_run=tasks_to_run,
     )
 
 
@@ -302,6 +313,14 @@ async def running_project_mark_for_cancellation(
 ) -> RunningProject:
     user = create_registered_user()
     created_project = await project(user, workbench=fake_workbench_without_outputs)
+    created_tasks = await create_tasks(
+        user=user,
+        project=created_project,
+        state=StateType.RUNNING,
+        progress=0.0,
+        start=now_time,
+    )
+    tasks_to_run = [t for t in created_tasks if t.node_class == NodeClass.COMPUTATIONAL]
     now_time = arrow.utcnow().datetime
     return RunningProject(
         user=user,
@@ -310,13 +329,7 @@ async def running_project_mark_for_cancellation(
             project_id=f"{created_project.uuid}",
             dag_adjacency_list=fake_workbench_adjacency,
         ),
-        tasks=await create_tasks(
-            user=user,
-            project=created_project,
-            state=StateType.RUNNING,
-            progress=0.0,
-            start=now_time,
-        ),
+        tasks=created_tasks,
         runs=await create_comp_run(
             user=user,
             project=created_project,
@@ -325,6 +338,7 @@ async def running_project_mark_for_cancellation(
             cancelled=now_time + datetime.timedelta(seconds=5),
         ),
         task_to_callback_mapping={},
+        tasks_to_run=tasks_to_run,
     )