fixed fastapi tests

Author: Andrei Neagu

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

@@ -1,10 +1,14 @@
 import datetime
 
 from fastapi import FastAPI
+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 RabbitNamespace
 from ...long_running_tasks.task import RedisNamespace, TasksManager
+from ...rabbitmq._client_rpc import RabbitMQRPCClient
 
 
 class FastAPILongRunningManager(BaseLongRunningManager):
@@ -14,7 +18,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(
@@ -23,13 +29,48 @@ 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) -> str:
+        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

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

@@ -27,7 +27,9 @@ async def list_tasks(
             result_href=str(request.url_for("get_task_result", task_id=t.task_id)),
             abort_href=str(request.url_for("remove_task", task_id=t.task_id)),
         )
-        for t in await lrt_api.list_tasks(long_running_manager, task_context={})
+        for t in await lrt_api.list_tasks(
+            long_running_manager.rpc_client, long_running_manager, task_context={}
+        )
     ]
 
 
@@ -48,7 +50,10 @@ async def get_task_status(
 ) -> TaskStatus:
     assert request  # nosec
     return await lrt_api.get_task_status(
-        long_running_manager, task_context={}, task_id=task_id
+        long_running_manager.rpc_client,
+        long_running_manager,
+        task_context={},
+        task_id=task_id,
     )
 
 
@@ -71,7 +76,10 @@ async def get_task_result(
 ) -> TaskResult | Any:
     assert request  # nosec
     return await lrt_api.get_task_result(
-        long_running_manager, task_context={}, task_id=task_id
+        long_running_manager.rpc_client,
+        long_running_manager,
+        task_context={},
+        task_id=task_id,
     )
 
 
@@ -93,4 +101,9 @@ async def remove_task(
     ],
 ) -> None:
     assert request  # nosec
-    await lrt_api.remove_task(long_running_manager, task_context={}, task_id=task_id)
+    await lrt_api.remove_task(
+        long_running_manager.rpc_client,
+        long_running_manager,
+        task_context={},
+        task_id=task_id,
+    )

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

@@ -1,13 +1,15 @@
 import datetime
 
 from fastapi import APIRouter, FastAPI
+from settings_library.rabbit import RabbitSettings
 from settings_library.redis import RedisSettings
 
 from ...long_running_tasks.constants import (
     DEFAULT_STALE_TASK_CHECK_INTERVAL,
     DEFAULT_STALE_TASK_DETECT_TIMEOUT,
 )
 from ...long_running_tasks.errors import BaseLongRunningError
+from ...long_running_tasks.models import RabbitNamespace
 from ...long_running_tasks.task import RedisNamespace
 from ._error_handlers import base_long_running_error_handler
 from ._manager import FastAPILongRunningManager
@@ -20,6 +22,8 @@ def setup(
     router_prefix: str = "",
     redis_settings: RedisSettings,
     redis_namespace: RedisNamespace,
+    rabbit_settings: RabbitSettings,
+    rabbit_namespace: RabbitNamespace,
     stale_task_check_interval: datetime.timedelta = DEFAULT_STALE_TASK_CHECK_INTERVAL,
     stale_task_detect_timeout: datetime.timedelta = DEFAULT_STALE_TASK_DETECT_TIMEOUT,
 ) -> None:
@@ -47,6 +51,8 @@ async def on_startup() -> None:
                 stale_task_detect_timeout=stale_task_detect_timeout,
                 redis_settings=redis_settings,
                 redis_namespace=redis_namespace,
+                rabbit_settings=rabbit_settings,
+                rabbit_namespace=rabbit_namespace,
             )
         )
         await long_running_manager.setup()

packages/service-library/src/servicelib/long_running_tasks/_rabbit/lrt_client.py

@@ -1,8 +1,9 @@
 import logging
-from typing import Any
+from datetime import timedelta
+from typing import Any, Final
 
 from models_library.rabbitmq_basic_types import RPCMethodName
-from pydantic import TypeAdapter
+from pydantic import PositiveInt, TypeAdapter
 
 from ...logging_utils import log_decorator
 from ...long_running_tasks.task import RegisteredTaskName
@@ -12,6 +13,13 @@
 
 _logger = logging.getLogger(__name__)
 
+_RPC_TIMEOUT_VERY_LONG_REQUEST: Final[PositiveInt] = int(
+    timedelta(minutes=60).total_seconds()
+)
+_RPC_TIMEOUT_NORMAL_REQUEST: Final[PositiveInt] = int(
+    timedelta(seconds=30).total_seconds()
+)
+
 
 @log_decorator(_logger, level=logging.DEBUG)
 async def start_task(
@@ -34,6 +42,7 @@ async def start_task(
         task_name=task_name,
         fire_and_forget=fire_and_forget,
         **task_kwargs,
+        timeout_s=_RPC_TIMEOUT_NORMAL_REQUEST,
     )
     assert isinstance(result, TaskId)  # nosec
     return result
@@ -50,9 +59,9 @@ async def list_tasks(
         get_namespace(namespace),
         TypeAdapter(RPCMethodName).validate_python("list_tasks"),
         task_context=task_context,
+        timeout_s=_RPC_TIMEOUT_NORMAL_REQUEST,
     )
-    assert TypeAdapter(list[TaskBase]).validate_python(result)  # nosec
-    return result
+    return TypeAdapter(list[TaskBase]).validate_python(result)
 
 
 @log_decorator(_logger, level=logging.DEBUG)
@@ -68,6 +77,7 @@ async def get_task_status(
         TypeAdapter(RPCMethodName).validate_python("get_task_status"),
         task_context=task_context,
         task_id=task_id,
+        timeout_s=_RPC_TIMEOUT_NORMAL_REQUEST,
     )
     assert isinstance(result, TaskStatus)  # nosec
     return result
@@ -86,6 +96,7 @@ async def get_task_result(
         TypeAdapter(RPCMethodName).validate_python("get_task_result"),
         task_context=task_context,
         task_id=task_id,
+        timeout_s=_RPC_TIMEOUT_NORMAL_REQUEST,
     )
 
 
@@ -104,5 +115,6 @@ async def remove_task(
         task_context=task_context,
         task_id=task_id,
         reraise_errors=reraise_errors,
+        timeout_s=_RPC_TIMEOUT_VERY_LONG_REQUEST,
     )
     assert result is None  # nosec

packages/service-library/tests/fastapi/long_running_tasks/conftest.py

@@ -9,17 +9,23 @@
 from fastapi import FastAPI
 from httpx import ASGITransport, AsyncClient
 from servicelib.fastapi import long_running_tasks
+from servicelib.rabbitmq._client_rpc import RabbitMQRPCClient
+from settings_library.rabbit import RabbitSettings
 from settings_library.redis import RedisSettings
 
 
 @pytest.fixture
-async def bg_task_app(router_prefix: str, redis_service: RedisSettings) -> FastAPI:
+async def bg_task_app(
+    router_prefix: str, redis_service: RedisSettings, rabbit_service: RabbitSettings
+) -> FastAPI:
     app = FastAPI()
 
     long_running_tasks.server.setup(
         app,
         redis_settings=redis_service,
         redis_namespace="test",
+        rabbit_settings=rabbit_service,
+        rabbit_namespace="test",
         router_prefix=router_prefix,
     )
     return app
@@ -33,3 +39,14 @@ async def async_client(bg_task_app: FastAPI) -> AsyncIterable[AsyncClient]:
         headers={"Content-Type": "application/json"},
     ) as client:
         yield client
+
+
+@pytest.fixture
+async def rabbitmq_rpc_client(
+    rabbit_service: RabbitSettings,
+) -> AsyncIterable[RabbitMQRPCClient]:
+    client = await RabbitMQRPCClient.create(
+        client_name="test-lrt-rpc-client", settings=rabbit_service
+    )
+    yield client
+    await client.close()

packages/service-library/tests/fastapi/long_running_tasks/test_long_running_client.py

@@ -8,7 +8,7 @@
 @pytest.mark.parametrize(
     "error_class, error_args",
     [
-        (HTTPError, dict(message="")),
+        (HTTPError, {"message": ""}),
     ],
 )
 async def test_retry_on_errors(

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

@@ -33,13 +33,18 @@
     TaskStatus,
 )
 from servicelib.long_running_tasks.task import TaskContext, TaskRegistry
+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
 from yarl import URL
 
+pytest_simcore_core_services_selection = [
+    "rabbit",
+]
+
 ITEM_PUBLISH_SLEEP: Final[float] = 0.1
 
 
@@ -81,6 +86,7 @@ async def create_string_list_task(
         fail: bool = False,
     ) -> TaskId:
         return await lrt_api.start_task(
+            long_running_manager.rpc_client,
             long_running_manager,
             _string_list_task.__name__,
             num_strings=num_strings,
@@ -93,12 +99,20 @@ async def create_string_list_task(
 
 @pytest.fixture
 async def app(
-    server_routes: APIRouter, use_in_memory_redis: RedisSettings
+    server_routes: APIRouter,
+    use_in_memory_redis: RedisSettings,
+    rabbit_service: RabbitSettings,
 ) -> AsyncIterator[FastAPI]:
     # overrides fastapi/conftest.py:app
     app = FastAPI(title="test app")
     app.include_router(server_routes)
-    setup_server(app, redis_settings=use_in_memory_redis, redis_namespace="test")
+    setup_server(
+        app,
+        redis_settings=use_in_memory_redis,
+        redis_namespace="test",
+        rabbit_settings=rabbit_service,
+        rabbit_namespace="test",
+    )
     setup_client(app)
     async with LifespanManager(app, startup_timeout=30, shutdown_timeout=30):
         yield app
@@ -188,7 +202,15 @@ 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 check the result
     result_url = app.url_path_for("get_task_result", task_id=task_id)
     result = await client.get(f"{result_url}")

packages/service-library/tests/fastapi/long_running_tasks/test_long_running_tasks_context_manager.py

@@ -28,8 +28,13 @@
     TaskProgress,
 )
 from servicelib.long_running_tasks.task import TaskRegistry
+from settings_library.rabbit import RabbitSettings
 from settings_library.redis import RedisSettings
 
+pytest_simcore_core_services_selection = [
+    "rabbit",
+]
+
 TASK_SLEEP_INTERVAL: Final[PositiveFloat] = 0.1
 
 # UTILS
@@ -71,7 +76,9 @@ async def create_task_user_defined_route(
             FastAPILongRunningManager, Depends(get_long_running_manager)
         ],
     ) -> TaskId:
-        return await lrt_api.start_task(long_running_manager, a_test_task.__name__)
+        return await lrt_api.start_task(
+            long_running_manager.rpc_client, long_running_manager, a_test_task.__name__
+        )
 
     @router.get("/api/failing", status_code=status.HTTP_200_OK)
     async def create_task_which_fails(
@@ -80,15 +87,20 @@ async def create_task_which_fails(
         ],
     ) -> TaskId:
         return await lrt_api.start_task(
-            long_running_manager, a_failing_test_task.__name__
+            long_running_manager.rpc_client,
+            long_running_manager,
+            a_failing_test_task.__name__,
         )
 
     return router
 
 
 @pytest.fixture
 async def bg_task_app(
-    user_routes: APIRouter, router_prefix: str, use_in_memory_redis: RedisSettings
+    user_routes: APIRouter,
+    router_prefix: str,
+    use_in_memory_redis: RedisSettings,
+    rabbit_service: RabbitSettings,
 ) -> AsyncIterable[FastAPI]:
     app = FastAPI()
 
@@ -99,6 +111,8 @@ async def bg_task_app(
         router_prefix=router_prefix,
         redis_settings=use_in_memory_redis,
         redis_namespace="test",
+        rabbit_settings=rabbit_service,
+        rabbit_namespace="test",
     )
     setup_client(app, router_prefix=router_prefix)