tests: add mocks

Author: giancarloromeo

services/notifications/src/simcore_service_notifications/core/application.py

@@ -33,8 +33,7 @@ def _initialise_logger(settings: ApplicationSettings):
     )
 
 
-def create_app() -> FastAPI:
-    settings = ApplicationSettings.create_from_envs()
+def create_app(settings: ApplicationSettings) -> FastAPI:
     _logger.debug(settings.model_dump_json(indent=2))
 
     _initialise_logger(settings)

services/notifications/src/simcore_service_notifications/main.py

@@ -1,3 +1,6 @@
 from simcore_service_notifications.core.application import create_app
+from simcore_service_notifications.core.settings import ApplicationSettings
 
-the_app = create_app()
+_settings = ApplicationSettings.create_from_envs()
+
+the_app = create_app(_settings)

services/notifications/src/simcore_service_notifications/modules/celery/worker_main.py

@@ -28,7 +28,7 @@
 assert _settings.NOTIFICATIONS_CELERY  # nosec
 the_celery_app = create_celery_app(_settings.NOTIFICATIONS_CELERY)
 
-app_server = FastAPIAppServer(app=create_app())
+app_server = FastAPIAppServer(app=create_app(_settings))
 
 
 def worker_init_wrapper(sender, **_kwargs):

services/notifications/tests/conftest.py

@@ -2,18 +2,40 @@
 # pylint: disable=unused-argument
 
 
+import datetime
+from collections.abc import AsyncIterator
+from functools import partial
 from pathlib import Path
+from typing import Any
 
 import pytest
+from celery import Celery  # type: ignore[import-untyped]
+from celery.contrib.testing.worker import (  # type: ignore[import-untyped]
+    TestWorkController,
+    start_worker,
+)
+from celery.signals import worker_init, worker_shutdown  # type: ignore[import-untyped]
+from celery.worker.worker import WorkController  # type: ignore[import-untyped]
+from celery_library.signals import on_worker_init, on_worker_shutdown
 from models_library.basic_types import BootModeEnum
 from pytest_simcore.helpers.monkeypatch_envs import EnvVarsDict, setenvs_from_dict
+from servicelib.fastapi.celery.app_server import FastAPIAppServer
+from simcore_service_notifications.core.application import create_app
+from simcore_service_notifications.core.settings import ApplicationSettings
+from simcore_service_notifications.modules.celery.tasks import (
+    TaskQueue,  # type: ignore[import-untyped]
+)
+from simcore_service_notifications.modules.celery.tasks import (
+    setup_worker_tasks,
+)
 
 pytest_plugins = [
     "pytest_simcore.docker_compose",
     "pytest_simcore.docker_swarm",
     "pytest_simcore.environment_configs",
     "pytest_simcore.postgres_service",
     "pytest_simcore.rabbit_service",
+    "pytest_simcore.redis_service",
     "pytest_simcore.repository_paths",
 ]
 
@@ -40,3 +62,53 @@ def mock_environment(
             "SC_BOOT_MODE": BootModeEnum.DEBUG,
         },
     )
+
+
+@pytest.fixture(scope="session")
+def celery_config() -> dict[str, Any]:
+    return {
+        "broker_connection_retry_on_startup": True,
+        "broker_url": "memory://localhost//",
+        "result_backend": "cache+memory://localhost//",
+        "result_expires": datetime.timedelta(days=7),
+        "result_extended": True,
+        "pool": "threads",
+        "task_default_queue": "default",
+        "task_send_sent_event": True,
+        "task_track_started": True,
+        "worker_send_task_events": True,
+    }
+
+
+@pytest.fixture
+async def with_celery_worker(
+    mock_environment: EnvVarsDict,
+    celery_app: Celery,
+    monkeypatch: pytest.MonkeyPatch,
+) -> AsyncIterator[TestWorkController]:
+    # Signals must be explicitily connected
+    monkeypatch.setenv("NOTIFICATIONS_WORKER_MODE", "true")
+    app_settings = ApplicationSettings.create_from_envs()
+
+    def _on_worker_init_wrapper(sender: WorkController, **_kwargs):
+        assert app_settings.NOTIFICATIONS_CELERY  # nosec
+        return partial(
+            on_worker_init,
+            FastAPIAppServer(app=create_app(app_settings)),
+            app_settings.NOTIFICATIONS_CELERY,
+        )(sender, **_kwargs)
+
+    worker_init.connect(_on_worker_init_wrapper)
+    worker_shutdown.connect(on_worker_shutdown)
+
+    setup_worker_tasks(celery_app)
+
+    with start_worker(
+        celery_app,
+        pool="threads",
+        concurrency=1,
+        loglevel="info",
+        perform_ping_check=False,
+        queues=",".join(queue.value for queue in TaskQueue),
+    ) as worker:
+        yield worker

services/notifications/tests/unit/conftest.py

@@ -11,6 +11,7 @@
 from pytest_simcore.helpers.monkeypatch_envs import EnvVarsDict, setenvs_from_dict
 from settings_library.rabbit import RabbitSettings
 from simcore_service_notifications.core.application import create_app
+from simcore_service_notifications.core.settings import ApplicationSettings
 
 
 @pytest.fixture
@@ -37,7 +38,7 @@ def app_environment(
 
 @pytest.fixture
 async def initialized_app(app_environment: EnvVarsDict) -> AsyncIterator[FastAPI]:
-    app: FastAPI = create_app()
+    app: FastAPI = create_app(ApplicationSettings.create_from_envs())
 
     async with LifespanManager(app, startup_timeout=30, shutdown_timeout=30):
         yield app