100% tested

Author: sanderegg

services/director-v2/src/simcore_service_director_v2/models/comp_runs.py

@@ -49,6 +49,7 @@ class CompRunsAtDB(BaseModel):
     cancelled: datetime.datetime | None
     metadata: RunMetadataDict = RunMetadataDict()
     use_on_demand_clusters: bool
+    last_scheduled: datetime.datetime | None
 
     @validator("result", pre=True)
     @classmethod
@@ -99,6 +100,7 @@ class Config:
                     "modified": "2021-03-01 13:07:34.19161",
                     "cancelled": None,
                     "use_on_demand_clusters": False,
+                    "last_scheduled": None,
                 },
                 {
                     "run_id": 432,
@@ -111,6 +113,7 @@ class Config:
                     "modified": "2021-03-01 13:07:34.19161",
                     "cancelled": None,
                     "use_on_demand_clusters": False,
+                    "last_scheduled": None,
                 },
                 {
                     "run_id": 43243,
@@ -132,6 +135,7 @@ class Config:
                         "some-other-metadata-which-is-an-array": [1, 3, 4],
                     },
                     "use_on_demand_clusters": False,
+                    "last_scheduled": None,
                 },
                 {
                     "run_id": 43243,
@@ -147,6 +151,7 @@ class Config:
                     "cancelled": None,
                     "metadata": None,
                     "use_on_demand_clusters": False,
+                    "last_scheduled": None,
                 },
             ]
         }

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

@@ -74,7 +74,7 @@ async def list(
             conditions.append(
                 or_(
                     comp_runs.c.last_scheduled.is_(None),
-                    comp_runs.c.last_scheduled >= scheduled_cutoff,
+                    comp_runs.c.last_scheduled <= scheduled_cutoff,
                 )
             )
 

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

@@ -9,14 +9,18 @@
 
 
 import asyncio
+import logging
 from collections.abc import AsyncIterator, Callable
+from typing import Any, Awaitable
 from unittest import mock
 
 import pytest
 import sqlalchemy as sa
 from _helpers import PublishedProject
 from fastapi import FastAPI
 from models_library.clusters import DEFAULT_CLUSTER_ID
+from models_library.projects import ProjectAtDB
+from models_library.projects_state import RunningState
 from pytest_mock.plugin import MockerFixture
 from pytest_simcore.helpers.monkeypatch_envs import setenvs_from_dict
 from pytest_simcore.helpers.typing_env import EnvVarsDict
@@ -26,10 +30,14 @@
 from settings_library.rabbit import RabbitSettings
 from settings_library.redis import RedisSettings
 from simcore_postgres_database.models.comp_runs import comp_runs
+from simcore_service_director_v2.core.errors import PipelineNotFoundError
+from simcore_service_director_v2.models.comp_pipelines import CompPipelineAtDB
 from simcore_service_director_v2.models.comp_runs import CompRunsAtDB, RunMetadataDict
 from simcore_service_director_v2.modules.comp_scheduler._distributed_scheduler import (
+    _SCHEDULER_INTERVAL,
     run_new_pipeline,
     schedule_pipelines,
+    stop_pipeline,
 )
 from simcore_service_director_v2.modules.comp_scheduler._models import (
     SchedulePipelineRabbitMessage,
@@ -135,7 +143,7 @@ async def slow_limited_gather(*args, **kwargs):
     mock_function.assert_called_once()
 
 
-async def test_schedule_pipelines_with_non_scheduled_runs(
+async def test_schedule_pipelines(
     initialized_app: FastAPI,
     published_project: PublishedProject,
     sqlalchemy_async_engine: AsyncEngine,
@@ -153,11 +161,121 @@ async def test_schedule_pipelines_with_non_scheduled_runs(
         run_metadata=run_metadata,
         use_on_demand_clusters=False,
     )
+    # this directly schedule a new pipeline
     scheduler_rabbit_client_parser.assert_called_once_with(
         SchedulePipelineRabbitMessage(
             user_id=published_project.project.prj_owner,
             project_id=published_project.project.uuid,
             iteration=1,
         ).body()
     )
+    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.project_uuid == published_project.project.uuid
+    assert comp_run.user_id == published_project.project.prj_owner
+    assert comp_run.iteration == 1
+    assert comp_run.cancelled is None
+    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
+    start_modified_time = comp_run.modified
+
+    await asyncio.sleep(_SCHEDULER_INTERVAL.total_seconds() - 1)
+
+    # this will now not schedule the pipeline since it was last scheduled
+    await schedule_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.cancelled is None
+    assert comp_run.modified == start_modified_time
+
+    await asyncio.sleep(_SCHEDULER_INTERVAL.total_seconds() + 1)
+    # this will now schedule the pipeline since the time passed
+    await schedule_pipelines(initialized_app)
+    scheduler_rabbit_client_parser.assert_called_once_with(
+        SchedulePipelineRabbitMessage(
+            user_id=published_project.project.prj_owner,
+            project_id=published_project.project.uuid,
+            iteration=1,
+        ).body()
+    )
+    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.cancelled is None
+    assert comp_run.modified > start_modified_time
+
+    # now we stop the pipeline, which should instantly trigger a schedule
+    await stop_pipeline(
+        initialized_app,
+        user_id=published_project.project.prj_owner,
+        project_id=published_project.project.uuid,
+    )
+    await schedule_pipelines(initialized_app)
+    scheduler_rabbit_client_parser.assert_called_once_with(
+        SchedulePipelineRabbitMessage(
+            user_id=published_project.project.prj_owner,
+            project_id=published_project.project.uuid,
+            iteration=1,
+        ).body()
+    )
+    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.cancelled is not None
+
+
+async def test_empty_pipeline_is_not_scheduled(
+    initialized_app: FastAPI,
+    registered_user: Callable[..., dict[str, Any]],
+    project: Callable[..., Awaitable[ProjectAtDB]],
+    pipeline: Callable[..., CompPipelineAtDB],
+    run_metadata: RunMetadataDict,
+    sqlalchemy_async_engine: AsyncEngine,
+    scheduler_rabbit_client_parser: mock.AsyncMock,
+    caplog: pytest.LogCaptureFixture,
+):
+    await _assert_comp_runs_empty(sqlalchemy_async_engine)
+    user = registered_user()
+    empty_project = await project(user)
+
+    # the project is not in the comp_pipeline, therefore scheduling it should fail
+    with pytest.raises(PipelineNotFoundError):
+        await run_new_pipeline(
+            initialized_app,
+            user_id=user["id"],
+            project_id=empty_project.uuid,
+            cluster_id=DEFAULT_CLUSTER_ID,
+            run_metadata=run_metadata,
+            use_on_demand_clusters=False,
+        )
+    await _assert_comp_runs_empty(sqlalchemy_async_engine)
+    scheduler_rabbit_client_parser.assert_not_called()
+
+    # create the empty pipeline now
+    pipeline(project_id=f"{empty_project.uuid}")
+
+    # creating a run with an empty pipeline is useless, check the scheduler is not kicking in
+    with caplog.at_level(logging.WARNING):
+        await run_new_pipeline(
+            initialized_app,
+            user_id=user["id"],
+            project_id=empty_project.uuid,
+            cluster_id=DEFAULT_CLUSTER_ID,
+            run_metadata=run_metadata,
+            use_on_demand_clusters=False,
+        )
+    assert len(caplog.records) == 1
+    assert "no computational dag defined" in caplog.records[0].message
+    await _assert_comp_runs_empty(sqlalchemy_async_engine)
+    scheduler_rabbit_client_parser.assert_not_called()