fixed tests

Author: sanderegg

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

@@ -107,7 +107,7 @@ async def _distribute_pipeline(
     # TODO: we should use the transaction and the asyncpg engine here to ensure 100% consistency
     # async with transaction_context(get_asyncpg_engine(app)) as connection:
     await rabbitmq_client.publish(
-        SchedulePipelineRabbitMessage.channel_name,
+        SchedulePipelineRabbitMessage.get_channel_name(),
         SchedulePipelineRabbitMessage(
             user_id=run.user_id,
             project_id=run.project_uuid,

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

@@ -22,6 +22,7 @@
 from pytest_simcore.helpers.typing_env import EnvVarsDict
 from servicelib.rabbitmq._client import RabbitMQClient
 from servicelib.redis import CouldNotAcquireLockError
+from servicelib.utils import limited_gather
 from settings_library.rabbit import RabbitSettings
 from settings_library.redis import RedisSettings
 from simcore_postgres_database.models.comp_runs import comp_runs
@@ -36,9 +37,7 @@
 from sqlalchemy.ext.asyncio import AsyncEngine
 
 pytest_simcore_core_services_selection = ["postgres", "rabbit", "redis"]
-pytest_simcore_ops_services_selection = [
-    "adminer",
-]
+pytest_simcore_ops_services_selection = ["adminer", "redis-commander"]
 
 
 @pytest.fixture
@@ -80,7 +79,7 @@ async def _assert_comp_runs(
     async with sqlalchemy_async_engine.connect() as conn:
         list_of_comp_runs = [
             CompRunsAtDB.from_orm(row)
-            async for row in await conn.stream(sa.select().select_from(comp_runs))
+            for row in await conn.execute(sa.select(comp_runs))
         ]
     assert len(list_of_comp_runs) == expected_total
     return list_of_comp_runs
@@ -106,25 +105,34 @@ async def test_schedule_pipelines_empty_db(
     await _assert_comp_runs_empty(sqlalchemy_async_engine)
 
 
-async def test_schedule_pipelines_concurently_raises_and_only_one_runs(
-    initialized_app: FastAPI,
+async def test_schedule_pipelines_concurently_runs_exclusively_and_raises(
+    initialized_app: FastAPI, mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch
 ):
     CONCURRENCY = 5
-    # TODO: this can be flaky as an empty scheduling is very short
-    with pytest.raises(
-        CouldNotAcquireLockError,
-        match=".+ computational-distributed-scheduler",
-    ):
-        await asyncio.gather(
-            *(schedule_pipelines(initialized_app) for _ in range(CONCURRENCY))
-        )
+    # NOTE: this ensure no flakyness as empty scheduling is very fast
+    original_function = limited_gather
+
+    async def slow_limited_gather(*args, **kwargs):
+        result = await original_function(*args, **kwargs)
+        await asyncio.sleep(3)  # to ensure flakyness does not occur
+        return result
+
+    mock_function = mocker.patch(
+        "simcore_service_director_v2.modules.comp_scheduler._distributed_scheduler.limited_gather",
+        autospec=True,
+        side_effect=slow_limited_gather,
+    )
 
     results = await asyncio.gather(
         *(schedule_pipelines(initialized_app) for _ in range(CONCURRENCY)),
         return_exceptions=True,
     )
 
-    assert results.count(None) == 1, "Only one task should have run"
+    assert results.count(None) == 1, f"Only one task should have run: {results}"
+    for r in results:
+        if r:
+            assert isinstance(r, CouldNotAcquireLockError)
+    mock_function.assert_called_once()
 
 
 async def test_schedule_pipelines_with_non_scheduled_runs(
@@ -135,8 +143,8 @@ async def test_schedule_pipelines_with_non_scheduled_runs(
     scheduler_rabbit_client_parser: mock.AsyncMock,
 ):
     await _assert_comp_runs_empty(sqlalchemy_async_engine)
-    # now we schedule a pipeline
     assert published_project.project.prj_owner
+    # now we schedule a pipeline
     await run_new_pipeline(
         initialized_app,
         user_id=published_project.project.prj_owner,
@@ -145,5 +153,11 @@ async def test_schedule_pipelines_with_non_scheduled_runs(
         run_metadata=run_metadata,
         use_on_demand_clusters=False,
     )
-    scheduler_rabbit_client_parser.assert_called_once_with()
+    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()
+    )
     comp_runs = await _assert_comp_runs(sqlalchemy_async_engine, expected_total=1)