fixed most tests

Author: Andrei Neagu

packages/service-library/src/servicelib/aiohttp/long_running_tasks/_manager.py

@@ -1,11 +1,14 @@
 import datetime
 
 from aiohttp import web
+from settings_library.rabbit import RabbitSettings
 from settings_library.redis import RedisSettings
 
+from ...long_running_tasks import lrt_api
 from ...long_running_tasks.base_long_running_manager import BaseLongRunningManager
-from ...long_running_tasks.models import TaskContext
+from ...long_running_tasks.models import RabbitNamespace, TaskContext
 from ...long_running_tasks.task import RedisNamespace, TasksManager
+from ...rabbitmq._client_rpc import RabbitMQRPCClient
 from ._constants import APP_LONG_RUNNING_MANAGER_KEY
 from ._request import get_task_context
 
@@ -17,7 +20,9 @@ def __init__(
         stale_task_check_interval: datetime.timedelta,
         stale_task_detect_timeout: datetime.timedelta,
         redis_settings: RedisSettings,
+        rabbit_settings: RabbitSettings,
         redis_namespace: RedisNamespace,
+        rabbit_namespace: RabbitNamespace,
     ):
         self._app = app
         self._tasks_manager = TasksManager(
@@ -26,17 +31,52 @@ def __init__(
             redis_settings=redis_settings,
             redis_namespace=redis_namespace,
         )
+        self._rabbit_namespace = rabbit_namespace
+        self.rabbit_settings = rabbit_settings
+        self._rpc_server: RabbitMQRPCClient | None = None
+        self._rpc_client: RabbitMQRPCClient | None = None
 
     @property
     def tasks_manager(self) -> TasksManager:
         return self._tasks_manager
 
+    @property
+    def rpc_server(self) -> RabbitMQRPCClient:
+        assert self._rpc_server is not None  # nosec
+        return self._rpc_server
+
+    @property
+    def rpc_client(self) -> RabbitMQRPCClient:
+        assert self._rpc_client is not None  # nosec
+        return self._rpc_client
+
+    @property
+    def rabbit_namespace(self) -> RabbitNamespace:
+        return self._rabbit_namespace
+
     async def setup(self) -> None:
         await self._tasks_manager.setup()
+        self._rpc_server = await RabbitMQRPCClient.create(
+            client_name=f"lrt-server-{self.rabbit_namespace}",
+            settings=self.rabbit_settings,
+        )
+        self._rpc_client = await RabbitMQRPCClient.create(
+            client_name=f"lrt-client-{self.rabbit_namespace}",
+            settings=self.rabbit_settings,
+        )
+        await lrt_api.register_rabbit_routes(self)
 
     async def teardown(self) -> None:
         await self._tasks_manager.teardown()
 
+        if self._rpc_server is not None:
+            await self._rpc_server.close()
+            self._rpc_server = None
+
+        if self._rpc_client is not None:
+            await self._rpc_client.close()
+            self._rpc_client = None
+
     @staticmethod
     def get_task_context(request: web.Request) -> TaskContext:
         return get_task_context(request)

packages/service-library/src/servicelib/aiohttp/long_running_tasks/_routes.py

@@ -29,6 +29,7 @@ async def list_tasks(request: web.Request) -> web.Response:
                 abort_href=f"{request.app.router['remove_task'].url_for(task_id=t.task_id)}",
             )
             for t in await lrt_api.list_tasks(
+                long_running_manager.rpc_client,
                 long_running_manager,
                 long_running_manager.get_task_context(request),
             )
@@ -42,6 +43,7 @@ async def get_task_status(request: web.Request) -> web.Response:
     long_running_manager = get_long_running_manager(request.app)
 
     task_status = await lrt_api.get_task_status(
+        long_running_manager.rpc_client,
         long_running_manager,
         long_running_manager.get_task_context(request),
         path_params.task_id,
@@ -56,6 +58,7 @@ async def get_task_result(request: web.Request) -> web.Response | Any:
 
     # NOTE: this might raise an exception that will be catched by the _error_handlers
     return await lrt_api.get_task_result(
+        long_running_manager.rpc_client,
         long_running_manager,
         long_running_manager.get_task_context(request),
         path_params.task_id,
@@ -68,6 +71,7 @@ async def remove_task(request: web.Request) -> web.Response:
     long_running_manager = get_long_running_manager(request.app)
 
     await lrt_api.remove_task(
+        long_running_manager.rpc_client,
         long_running_manager,
         long_running_manager.get_task_context(request),
         path_params.task_id,

packages/service-library/src/servicelib/aiohttp/long_running_tasks/_server.py

@@ -8,6 +8,7 @@
 from aiohttp.web import HTTPException
 from common_library.json_serialization import json_dumps
 from pydantic import AnyHttpUrl, TypeAdapter
+from settings_library.rabbit import RabbitSettings
 from settings_library.redis import RedisSettings
 
 from ...aiohttp import status
@@ -20,7 +21,7 @@
     DEFAULT_STALE_TASK_CHECK_INTERVAL,
     DEFAULT_STALE_TASK_DETECT_TIMEOUT,
 )
-from ...long_running_tasks.models import TaskContext, TaskGet
+from ...long_running_tasks.models import RabbitNamespace, TaskContext, TaskGet
 from ...long_running_tasks.task import RedisNamespace, RegisteredTaskName
 from ..typing_extension import Handler
 from . import _routes
@@ -63,6 +64,7 @@ async def start_long_running_task(
     task_id = None
     try:
         task_id = await lrt_api.start_task(
+            long_running_manager.rpc_client,
             long_running_manager,
             registerd_task_name,
             fire_and_forget=fire_and_forget,
@@ -97,7 +99,12 @@ async def start_long_running_task(
     except asyncio.CancelledError:
         # remove the task, the client was disconnected
         if task_id:
-            await lrt_api.remove_task(long_running_manager, task_context, task_id)
+            await lrt_api.remove_task(
+                long_running_manager.rpc_client,
+                long_running_manager,
+                task_context,
+                task_id,
+            )
         raise
 
 
@@ -142,6 +149,8 @@ def setup(
     router_prefix: str,
     redis_settings: RedisSettings,
     redis_namespace: RedisNamespace,
+    rabbit_settings: RabbitSettings,
+    rabbit_namespace: RabbitNamespace,
     handler_check_decorator: Callable = _no_ops_decorator,
     task_request_context_decorator: Callable = _no_task_context_decorator,
     stale_task_check_interval: datetime.timedelta = DEFAULT_STALE_TASK_CHECK_INTERVAL,
@@ -170,7 +179,9 @@ async def on_cleanup_ctx(app: web.Application) -> AsyncGenerator[None, None]:
                 stale_task_check_interval=stale_task_check_interval,
                 stale_task_detect_timeout=stale_task_detect_timeout,
                 redis_settings=redis_settings,
+                rabbit_settings=rabbit_settings,
                 redis_namespace=redis_namespace,
+                rabbit_namespace=rabbit_namespace,
             )
         )
 

packages/service-library/tests/aiohttp/long_running_tasks/test_long_running_tasks.py

@@ -23,33 +23,34 @@
 from servicelib.aiohttp.rest_middlewares import append_rest_middlewares
 from servicelib.long_running_tasks.models import TaskGet, TaskId, TaskStatus
 from servicelib.long_running_tasks.task import TaskContext
+from settings_library.rabbit import RabbitSettings
 from settings_library.redis import RedisSettings
 from tenacity.asyncio import AsyncRetrying
 from tenacity.retry import retry_if_exception_type
 from tenacity.stop import stop_after_delay
 from tenacity.wait import wait_fixed
 
 pytest_simcore_core_services_selection = [
-    "redis",
-]
-
-pytest_simcore_ops_services_selection = [
-    "redis-commander",
+    "rabbit",
 ]
 
 
 @pytest.fixture
 def app(
-    server_routes: web.RouteTableDef, redis_service: RedisSettings
+    server_routes: web.RouteTableDef,
+    use_in_memory_redis: RedisSettings,
+    rabbit_service: RabbitSettings,
 ) -> web.Application:
     app = web.Application()
     app.add_routes(server_routes)
     # this adds enveloping and error middlewares
     append_rest_middlewares(app, api_version="")
     long_running_tasks.server.setup(
         app,
-        redis_settings=redis_service,
+        redis_settings=use_in_memory_redis,
         redis_namespace="test",
+        rabbit_settings=rabbit_service,
+        rabbit_namespace="test",
         router_prefix="/futures",
     )
 
@@ -109,7 +110,14 @@ async def test_workflow(
         ("generated item", 0.8),
         ("finished", 1.0),
     ]
-    assert all(x in progress_updates for x in EXPECTED_MESSAGES)
+    async for attempt in AsyncRetrying(
+        wait=wait_fixed(0.1),
+        stop=stop_after_delay(10),
+        reraise=True,
+        retry=retry_if_exception_type(AssertionError),
+    ):
+        with attempt:
+            assert all(x in progress_updates for x in EXPECTED_MESSAGES)
     # now get the result
     result_url = client.app.router["get_task_result"].url_for(task_id=task_id)
     result = await client.get(f"{result_url}")

packages/service-library/tests/aiohttp/long_running_tasks/test_long_running_tasks_client.py

@@ -15,13 +15,20 @@
     long_running_task_request,
 )
 from servicelib.aiohttp.rest_middlewares import append_rest_middlewares
+from settings_library.rabbit import RabbitSettings
 from settings_library.redis import RedisSettings
 from yarl import URL
 
+pytest_simcore_core_services_selection = [
+    "rabbit",
+]
+
 
 @pytest.fixture
 def app(
-    server_routes: web.RouteTableDef, use_in_memory_redis: RedisSettings
+    server_routes: web.RouteTableDef,
+    use_in_memory_redis: RedisSettings,
+    rabbit_service: RabbitSettings,
 ) -> web.Application:
     app = web.Application()
     app.add_routes(server_routes)
@@ -31,6 +38,8 @@ def app(
         app,
         redis_settings=use_in_memory_redis,
         redis_namespace="test",
+        rabbit_settings=rabbit_service,
+        rabbit_namespace="test",
         router_prefix="/futures",
     )
 
@@ -58,7 +67,7 @@ async def test_long_running_task_request_raises_400(
     client: TestClient, long_running_task_url: URL
 ):
     # missing parameters raises
-    with pytest.raises(ClientResponseError):
+    with pytest.raises(ClientResponseError):  # noqa: PT012
         async for _ in long_running_task_request(
             client.session, long_running_task_url, None
         ):
@@ -95,7 +104,7 @@ async def test_long_running_task_request_timeout(
 ):
     assert client.app
     task: LRTask | None = None
-    with pytest.raises(asyncio.TimeoutError):
+    with pytest.raises(asyncio.TimeoutError):  # noqa: PT012
         async for task in long_running_task_request(
             client.session,
             long_running_task_url.with_query(num_strings=10, sleep_time=1),

packages/service-library/tests/aiohttp/long_running_tasks/test_long_running_tasks_with_task_context.py

@@ -27,10 +27,11 @@
 from servicelib.aiohttp.typing_extension import Handler
 from servicelib.long_running_tasks.models import TaskGet, TaskId
 from servicelib.long_running_tasks.task import TaskContext
+from settings_library.rabbit import RabbitSettings
 from settings_library.redis import RedisSettings
 
 pytest_simcore_core_services_selection = [
-    "redis",
+    "rabbit",
 ]
 # WITH TASK CONTEXT
 # NOTE: as the long running task framework may be used in any number of services
@@ -67,16 +68,19 @@ async def _test_task_context_decorator(
 def app_with_task_context(
     server_routes: web.RouteTableDef,
     task_context_decorator,
-    redis_service: RedisSettings,
+    use_in_memory_redis: RedisSettings,
+    rabbit_service: RabbitSettings,
 ) -> web.Application:
     app = web.Application()
     app.add_routes(server_routes)
     # this adds enveloping and error middlewares
     append_rest_middlewares(app, api_version="")
     long_running_tasks.server.setup(
         app,
-        redis_settings=redis_service,
+        redis_settings=use_in_memory_redis,
         redis_namespace="test",
+        rabbit_settings=rabbit_service,
+        rabbit_namespace="test",
         router_prefix="/futures_with_task_context",
         task_request_context_decorator=task_context_decorator,
     )

services/web/server/src/simcore_service_webserver/long_running_tasks.py

@@ -10,16 +10,18 @@
 )
 from servicelib.aiohttp.long_running_tasks.server import setup
 from servicelib.aiohttp.typing_extension import Handler
+from servicelib.long_running_tasks.models import RabbitNamespace
 from servicelib.long_running_tasks.task import RedisNamespace
 
-from . import redis
+from . import rabbitmq, redis
 from ._meta import API_VTAG
 from .login.decorators import login_required
 from .models import AuthenticatedRequestContext
 
 _logger = logging.getLogger(__name__)
 
-_LONG_RUNNING_TASKS_NAMESPACE: Final[RedisNamespace] = "webserver-legacy"
+_LRT_REDIS_NAMESPACE: Final[RedisNamespace] = "webserver-legacy"
+_LRT_RABBIT_NAMESPACE: Final[RabbitNamespace] = "webserver-legacy"
 
 
 def webserver_request_context_decorator(handler: Handler):
@@ -37,10 +39,13 @@ async def _test_task_context_decorator(
 
 @ensure_single_setup(__name__, logger=_logger)
 def setup_long_running_tasks(app: web.Application) -> None:
+
     setup(
         app,
         redis_settings=redis.get_plugin_settings(app),
-        redis_namespace=_LONG_RUNNING_TASKS_NAMESPACE,
+        rabbit_settings=rabbitmq.get_plugin_settings(app),
+        redis_namespace=_LRT_REDIS_NAMESPACE,
+        rabbit_namespace=_LRT_RABBIT_NAMESPACE,
         router_prefix=f"/{API_VTAG}/tasks-legacy",
         handler_check_decorator=login_required,
         task_request_context_decorator=webserver_request_context_decorator,