add waiting for calls

Author: sanderegg

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

@@ -12,6 +12,7 @@
 from unittest import mock
 
 import pytest
+from tenacity import retry, stop_after_delay, wait_fixed
 from _helpers import PublishedProject
 from fastapi import FastAPI
 from pytest_mock import MockerFixture
@@ -25,6 +26,7 @@
 from simcore_service_director_v2.modules.comp_scheduler._worker import (
     _get_scheduler_worker,
 )
+from settings_library.rabbit import RabbitSettings
 
 pytest_simcore_core_services_selection = ["postgres", "rabbit", "redis"]
 pytest_simcore_ops_services_selection = ["adminer"]
@@ -93,19 +95,21 @@ def with_scheduling_concurrency(
     )
 
 
+@pytest.mark.testit
 @pytest.mark.parametrize("scheduling_concurrency", [1, 50, 100])
 @pytest.mark.parametrize(
     "queue_name", [SchedulePipelineRabbitMessage.get_channel_name()]
 )
 async def test_worker_scheduling_parallelism(
+    rabbit_service: RabbitSettings,
+    ensure_parametrized_queue_is_empty: None,
     scheduling_concurrency: int,
     with_scheduling_concurrency: EnvVarsDict,
     with_disabled_auto_scheduling: mock.Mock,
     mocked_scheduler_api: mock.Mock,
     initialized_app: FastAPI,
     publish_project: Callable[[], Awaitable[PublishedProject]],
     run_metadata: RunMetadataDict,
-    ensure_parametrized_queue_is_empty: None,
 ):
     with_disabled_auto_scheduling.assert_called_once()
 
@@ -125,8 +129,15 @@ async def _project_pipeline_creation_workflow() -> None:
             use_on_demand_clusters=False,
         )
 
+    # whatever scheduling concurrency we call in here, we shall always see the same number of calls to the scheduler
     await asyncio.gather(
         *(_project_pipeline_creation_workflow() for _ in range(scheduling_concurrency))
     )
+    # the call to run the pipeline is async so we need to wait here
     mocked_scheduler_api.assert_called()
-    assert mocked_scheduler_api.call_count == scheduling_concurrency
+
+    @retry(stop=stop_after_delay(5), reraise=True, wait=wait_fixed(0.5))
+    def _assert_expected_called():
+        assert mocked_scheduler_api.call_count == scheduling_concurrency
+
+    _assert_expected_called()