tests: fix initialization

Author: giancarloromeo

packages/service-library/src/servicelib/rabbitmq/rpc_interfaces/notifications/__init__.py

@@ -2,7 +2,7 @@
 
 from models_library.api_schemas_notifications import NOTIFICATIONS_RPC_NAMESPACE
 from models_library.rabbitmq_basic_types import RPCMethodName
-from models_library.rpc.notifications.messages import BaseRecipient, NotificationMessage
+from models_library.rpc.notifications.messages import NotificationMessage, Recipient
 from pydantic import NonNegativeInt, TypeAdapter
 
 from ... import RabbitMQRPCClient
@@ -14,7 +14,7 @@ async def send_notification_message(
     rabbitmq_rpc_client: RabbitMQRPCClient,
     *,
     message: NotificationMessage,
-    recipients: list[BaseRecipient],
+    recipients: list[Recipient],
 ) -> None:
     await rabbitmq_rpc_client.request(
         NOTIFICATIONS_RPC_NAMESPACE,

packages/service-library/src/servicelib/rabbitmq/rpc_interfaces/notifications/messages.py

@@ -7,6 +7,7 @@
     EmailRecipient,
     NotificationMessage,
 )
+from servicelib.celery.models import TaskID
 
 _logger = logging.getLogger(__name__)
 
@@ -15,6 +16,7 @@
 
 async def send_email(
     task: Task,
+    task_id: TaskID,
     message: NotificationMessage,
     recipient: EmailRecipient,
 ) -> None:

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

@@ -16,16 +16,18 @@
 _logger = logging.getLogger(__name__)
 
 
-_TASK_QUEUE_PREFIX: str = "notifications"
+_NOTIFICATIONS_PREFIX: str = "notifications"
 
 
 class TaskQueue(StrEnum):
-    DEFAULT = f"{_TASK_QUEUE_PREFIX}.default"
+    DEFAULT = f"{_NOTIFICATIONS_PREFIX}.default"
 
 
 def setup_worker_tasks(app: Celery) -> None:
     register_celery_types()
     register_pydantic_types(NotificationMessage, EmailRecipient, SMSRecipient)
 
     with log_context(_logger, logging.INFO, msg="worker tasks registration"):
-        register_task(app, send_email, EMAIL_CHANNEL_NAME)
+        register_task(
+            app, send_email, ".".join((_NOTIFICATIONS_PREFIX, EMAIL_CHANNEL_NAME))
+        )

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

@@ -1,12 +1,12 @@
-from enum import StrEnum
+import logging
 
 from models_library.rpc.notifications.messages import NotificationMessage, Recipient
 from servicelib.celery.models import TaskContext
 from servicelib.celery.task_manager import TaskManager
 
+from ..modules.celery.tasks import TaskQueue
 
-class TaskQueues(StrEnum):
-    DEFAULT = "notifications.default"
+_logger = logging.getLogger(__name__)
 
 
 async def send_notification(
@@ -19,7 +19,7 @@ async def send_notification(
         await task_manager.send_task(
             name=f"notifications.{recipient.type}",
             context=TaskContext(),
-            queue=TaskQueues.DEFAULT,
+            queue=TaskQueue.DEFAULT,
             message=message,
             recipient=recipient,
         )

services/notifications/src/simcore_service_notifications/services/notifications_service.py

@@ -1,32 +1,11 @@
 # pylint: disable=redefined-outer-name
 # pylint: disable=unused-argument
 
-
-import datetime
-from collections.abc import AsyncIterator, Awaitable, Callable
-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
-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 servicelib.rabbitmq import RabbitMQRPCClient
-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,
-    setup_worker_tasks,
-)
 
 pytest_plugins = [
     "pytest_simcore.docker_compose",
@@ -60,62 +39,3 @@ 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(
-    app_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
-
-
-@pytest.fixture
-async def notifications_rabbitmq_rpc_client(
-    rabbitmq_rpc_client: Callable[[str], Awaitable[RabbitMQRPCClient]],
-) -> RabbitMQRPCClient:
-    rpc_client = await rabbitmq_rpc_client("pytest_notifications_rpc_client")
-    assert rpc_client
-    return rpc_client

services/notifications/tests/conftest.py

@@ -1,18 +1,31 @@
 # pylint: disable=redefined-outer-name
 # pylint: disable=unused-argument
 
-from collections.abc import AsyncIterator
-from typing import Final
+import datetime
+from collections.abc import AsyncIterator, Awaitable, Callable
+from functools import partial
+from typing import Any, Final
 
 import pytest
 from asgi_lifespan import LifespanManager
+from celery import Celery
+from celery.contrib.testing.worker import start_worker
+from celery.signals import worker_init, worker_shutdown
+from celery.worker.worker import WorkController
+from celery_library.signals import on_worker_init, on_worker_shutdown
 from fastapi import FastAPI
-from fastapi.testclient import TestClient
+from pytest_mock import MockerFixture
 from pytest_simcore.helpers.monkeypatch_envs import EnvVarsDict, setenvs_from_dict
+from servicelib.fastapi.celery.app_server import FastAPIAppServer
+from servicelib.rabbitmq import RabbitMQRPCClient
 from settings_library.rabbit import RabbitSettings
 from settings_library.redis import RedisSettings
 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,
+    setup_worker_tasks,
+)
 
 _LIFESPAN_TIMEOUT: Final[int] = 30
 
@@ -56,13 +69,77 @@ def app_settings(
 
 
 @pytest.fixture
-async def initialized_app(app_settings: ApplicationSettings) -> AsyncIterator[FastAPI]:
+async def fastapi_app(app_settings: ApplicationSettings) -> AsyncIterator[FastAPI]:
     app: FastAPI = create_app(app_settings)
 
     async with LifespanManager(app, startup_timeout=30, shutdown_timeout=30):
         yield app
 
 
+@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
+def mock_celery_app(mocker: MockerFixture, celery_config: dict[str, Any]) -> Celery:
+    celery_app = Celery(**celery_config)
+
+    for module in ("simcore_service_notifications.clients.celery.create_app",):
+        mocker.patch(module, return_value=celery_app)
+
+    return celery_app
+
+
+@pytest.fixture
+async def mock_celery_worker(
+    app_environment: EnvVarsDict,
+    celery_app: Celery,
+    fastapi_app: FastAPI,
+    monkeypatch: pytest.MonkeyPatch,
+) -> AsyncIterator[Any]:
+    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=fastapi_app),
+            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="debug",
+        perform_ping_check=False,
+        queues=",".join(queue.value for queue in TaskQueue),
+    ) as worker:
+        yield worker
+
+
 @pytest.fixture
-def test_client(initialized_app: FastAPI) -> TestClient:
-    return TestClient(initialized_app)
+async def notifications_rabbitmq_rpc_client(
+    rabbitmq_rpc_client: Callable[[str], Awaitable[RabbitMQRPCClient]],
+) -> RabbitMQRPCClient:
+    rpc_client = await rabbitmq_rpc_client("pytest_notifications_rpc_client")
+    assert rpc_client
+    return rpc_client

services/notifications/tests/unit/conftest.py

@@ -1,5 +1,4 @@
-from celery.contrib.testing.worker import TestWorkController
-from fastapi import FastAPI
+import pytest
 from models_library.rpc.notifications.messages import NotificationMessage
 from servicelib.rabbitmq import RabbitMQRPCClient
 from servicelib.rabbitmq.rpc_interfaces.notifications.messages import (
@@ -13,10 +12,13 @@
 ]
 
 
+@pytest.mark.usefixtures(
+    "mock_celery_app",
+    "mock_celery_worker",
+    "fastapi_app",
+)
 async def test_send_email(
-    initialized_app: FastAPI,
     notifications_rabbitmq_rpc_client: RabbitMQRPCClient,
-    with_celery_worker: TestWorkController,
 ):
     await send_notification_message(
         notifications_rabbitmq_rpc_client,