createing tests for repository

Author: sanderegg

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

@@ -1,6 +1,7 @@
 import logging
 from typing import Final
 
+import arrow
 import networkx as nx
 from aiopg.sa import Engine
 from fastapi import FastAPI
@@ -123,13 +124,15 @@ async def schedule_all_pipelines(app: FastAPI) -> None:
     with log_context(_logger, logging.DEBUG, msg="scheduling pipelines"):
         db_engine = get_db_engine(app)
         runs_to_schedule = await CompRunsRepository.instance(db_engine).list(
-            filter_by_state=SCHEDULED_STATES, need_scheduling=True
+            filter_by_state=SCHEDULED_STATES,
+            never_scheduled=True,
+            processed_before=arrow.utcnow().datetime - SCHEDULER_INTERVAL,
         )
         possibly_lost_scheduled_pipelines = await CompRunsRepository.instance(
             db_engine
         ).list(
             filter_by_state=SCHEDULED_STATES,
-            scheduled_since=SCHEDULER_INTERVAL * _LOST_TASKS_FACTOR,
+            scheduled_after=SCHEDULER_INTERVAL * _LOST_TASKS_FACTOR,
         )
         if possibly_lost_scheduled_pipelines:
             _logger.error(

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

@@ -233,7 +233,7 @@ async def _set_schedule_done(
         project_id: ProjectID,
         iteration: Iteration,
     ) -> None:
-        await CompRunsRepository.instance(self.db_engine).mark_as_scheduled_done(
+        await CompRunsRepository.instance(self.db_engine).mark_scheduling_done(
             user_id=user_id,
             project_id=project_id,
             iteration=iteration,

services/director-v2/src/simcore_service_director_v2/modules/db/repositories/comp_runs.py

@@ -55,11 +55,12 @@ async def get(
 
     async def list(
         self,
+        *,
         filter_by_state: set[RunningState] | None = None,
-        need_scheduling: bool | None = None,
-        scheduled_since: datetime.timedelta | None = None,
+        never_scheduled: bool = False,
+        processed_before: datetime.datetime | None = None,
+        scheduled_after: datetime.timedelta | None = None,
     ) -> list[CompRunsAtDB]:
-
         conditions = []
         if filter_by_state:
             conditions.append(
@@ -72,13 +73,14 @@ async def list(
             )
 
         scheduling_or_conditions = []
-        if need_scheduling is not None:
-            scheduling_or_conditions.append(comp_runs.c.last_scheduled.is_(None))
-        if scheduled_since is not None:
-            scheduled_cutoff = arrow.utcnow().datetime - scheduled_since
-            scheduling_or_conditions.append(
-                comp_runs.c.last_scheduled <= scheduled_cutoff
-            )
+        if never_scheduled:
+            scheduling_or_conditions.append(comp_runs.c.scheduled.is_(None))
+        if scheduled_after is not None:
+            scheduled_cutoff = arrow.utcnow().datetime - scheduled_after
+            scheduling_or_conditions.append(comp_runs.c.scheduled <= scheduled_cutoff)
+
+        if processed_before is not None:
+            scheduling_or_conditions.append(comp_runs.c.processed <= processed_before)
 
         if scheduling_or_conditions:
             conditions.append(sa.or_(*scheduling_or_conditions))
@@ -189,15 +191,16 @@ async def mark_for_scheduling(
             user_id,
             project_id,
             iteration,
-            last_scheduled=arrow.utcnow().datetime,
+            scheduled=arrow.utcnow().datetime,
+            processed=None,
         )
 
-    async def mark_as_scheduled_done(
+    async def mark_scheduling_done(
         self, *, user_id: UserID, project_id: ProjectID, iteration: PositiveInt
     ) -> CompRunsAtDB | None:
         return await self.update(
             user_id,
             project_id,
             iteration,
-            last_scheduled=None,
+            processed=arrow.utcnow().datetime,
         )

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

@@ -0,0 +1,51 @@
+import pytest
+from models_library.projects import ProjectID
+from models_library.users import UserID
+from simcore_service_director_v2.core.errors import ComputationalRunNotFoundError
+from simcore_service_director_v2.modules.db.repositories.comp_runs import (
+    CompRunsRepository,
+)
+
+pytest_simcore_core_services_selection = [
+    "postgres",
+]
+pytest_simcore_ops_services_selection = [
+    "adminer",
+]
+
+
+async def test_get(aiopg_engine, user_id: UserID, project_id: ProjectID):
+    with pytest.raises(ComputationalRunNotFoundError):
+        await CompRunsRepository(aiopg_engine).get(user_id, project_id)
+
+
+async def test_list():
+    ...
+
+
+async def test_create():
+    ...
+
+
+async def test_update():
+    ...
+
+
+async def test_delete():
+    ...
+
+
+async def test_set_run_result():
+    ...
+
+
+async def test_mark_for_cancellation():
+    ...
+
+
+async def test_mark_for_scheduling():
+    ...
+
+
+async def test_mark_scheduling_done():
+    ...

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

@@ -177,21 +177,21 @@ async def test_schedule_all_pipelines(
     assert comp_run.cluster_id == DEFAULT_CLUSTER_ID
     assert comp_run.metadata == run_metadata
     assert comp_run.result is RunningState.PUBLISHED
-    assert comp_run.last_scheduled is not None
-    start_schedule_time = comp_run.last_scheduled
+    assert comp_run.scheduled is not None
+    start_schedule_time = comp_run.scheduled
     start_modified_time = comp_run.modified
 
     # this will now not schedule the pipeline since it was already scheduled
     await schedule_all_pipelines(initialized_app)
     scheduler_rabbit_client_parser.assert_not_called()
     comp_runs = await assert_comp_runs(sqlalchemy_async_engine, expected_total=1)
     comp_run = comp_runs[0]
-    assert comp_run.last_scheduled == start_schedule_time, "scheduled time changed!"
+    assert comp_run.scheduled == start_schedule_time, "scheduled time changed!"
     assert comp_run.cancelled is None
     assert comp_run.modified == start_modified_time
 
     # once the worker is done, the schedule time is set back to None
-    await CompRunsRepository(aiopg_engine).mark_as_scheduled_done(
+    await CompRunsRepository(aiopg_engine).mark_scheduling_done(
         user_id=comp_run.user_id,
         project_id=comp_run.project_uuid,
         iteration=comp_run.iteration,
@@ -210,9 +210,9 @@ async def test_schedule_all_pipelines(
     scheduler_rabbit_client_parser.reset_mock()
     comp_runs = await assert_comp_runs(sqlalchemy_async_engine, expected_total=1)
     comp_run = comp_runs[0]
-    assert comp_run.last_scheduled is not None
-    assert comp_run.last_scheduled > start_schedule_time
-    last_schedule_time = comp_run.last_scheduled
+    assert comp_run.scheduled is not None
+    assert comp_run.scheduled > start_schedule_time
+    last_schedule_time = comp_run.scheduled
     assert comp_run.cancelled is None
     assert comp_run.modified > start_modified_time
 
@@ -233,8 +233,8 @@ async def test_schedule_all_pipelines(
     scheduler_rabbit_client_parser.reset_mock()
     comp_runs = await assert_comp_runs(sqlalchemy_async_engine, expected_total=1)
     comp_run = comp_runs[0]
-    assert comp_run.last_scheduled is not None
-    assert comp_run.last_scheduled > last_schedule_time
+    assert comp_run.scheduled is not None
+    assert comp_run.scheduled > last_schedule_time
     assert comp_run.cancelled is not None
 
 
@@ -277,16 +277,16 @@ async def test_schedule_all_pipelines_logs_error_if_it_find_old_pipelines(
     assert comp_run.cluster_id == DEFAULT_CLUSTER_ID
     assert comp_run.metadata == run_metadata
     assert comp_run.result is RunningState.PUBLISHED
-    assert comp_run.last_scheduled is not None
-    start_schedule_time = comp_run.last_scheduled
+    assert comp_run.scheduled is not None
+    start_schedule_time = comp_run.scheduled
     start_modified_time = comp_run.modified
 
     # this will now not schedule the pipeline since it was already scheduled
     await schedule_all_pipelines(initialized_app)
     scheduler_rabbit_client_parser.assert_not_called()
     comp_runs = await assert_comp_runs(sqlalchemy_async_engine, expected_total=1)
     comp_run = comp_runs[0]
-    assert comp_run.last_scheduled == start_schedule_time, "scheduled time changed!"
+    assert comp_run.scheduled == start_schedule_time, "scheduled time changed!"
     assert comp_run.cancelled is None
     assert comp_run.modified == start_modified_time
 
@@ -295,7 +295,7 @@ async def test_schedule_all_pipelines_logs_error_if_it_find_old_pipelines(
         comp_run.user_id,
         comp_run.project_uuid,
         comp_run.iteration,
-        last_scheduled=datetime.datetime.now(tz=datetime.UTC)
+        scheduled=datetime.datetime.now(tz=datetime.UTC)
         - SCHEDULER_INTERVAL * (_LOST_TASKS_FACTOR + 1),
     )
     with caplog.at_level(logging.ERROR):
@@ -313,8 +313,8 @@ async def test_schedule_all_pipelines_logs_error_if_it_find_old_pipelines(
     scheduler_rabbit_client_parser.reset_mock()
     comp_runs = await assert_comp_runs(sqlalchemy_async_engine, expected_total=1)
     comp_run = comp_runs[0]
-    assert comp_run.last_scheduled is not None
-    assert comp_run.last_scheduled > start_schedule_time
+    assert comp_run.scheduled is not None
+    assert comp_run.scheduled > start_schedule_time
     assert comp_run.cancelled is None
     assert comp_run.modified > start_modified_time