added RPC endpoint

Author: Andrei Neagu

packages/service-library/src/servicelib/rabbitmq/rpc_interfaces/dynamic_sidecar/long_running_tasks.py

@@ -0,0 +1,25 @@
+import logging
+
+from models_library.projects_nodes_io import NodeID
+from models_library.rabbitmq_basic_types import RPCMethodName
+from pydantic import TypeAdapter
+
+from ....logging_utils import log_decorator
+from ... import RabbitMQRPCClient
+from ._utils import get_rpc_namespace
+
+_logger = logging.getLogger(__name__)
+
+
+@log_decorator(_logger, level=logging.DEBUG)
+async def cleanup_local_long_running_tasks(
+    rabbitmq_rpc_client: RabbitMQRPCClient,
+    *,
+    node_id: NodeID,
+) -> None:
+    rpc_namespace = get_rpc_namespace(node_id)
+    result = await rabbitmq_rpc_client.request(
+        rpc_namespace,
+        TypeAdapter(RPCMethodName).validate_python("cleanup_local_long_running_tasks"),
+    )
+    assert result is None  # nosec

services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/api/rpc/_long_running_tasks.py

@@ -0,0 +1,11 @@
+from fastapi import FastAPI
+from servicelib.rabbitmq import RPCRouter
+
+from ...services import long_running_tasks
+
+router = RPCRouter()
+
+
+@router.expose()
+async def cleanup_local_long_running_tasks(app: FastAPI) -> None:
+    await long_running_tasks.cleanup_long_running_tasks(app)

services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/api/rpc/routes.py

@@ -4,11 +4,12 @@
 
 from ...core.rabbitmq import get_rabbitmq_rpc_server
 from ...core.settings import ApplicationSettings
-from . import _disk, _disk_usage, _volumes
+from . import _disk, _disk_usage, _long_running_tasks, _volumes
 
 ROUTERS: list[RPCRouter] = [
     _disk_usage.router,
     _disk.router,
+    _long_running_tasks.router,
     _volumes.router,
 ]
 

services/dynamic-sidecar/tests/unit/test_api_rest_long_running_tasks.py renamed to services/dynamic-sidecar/tests/unit/api/rest/test_long_running_tasks.py

@@ -25,13 +25,12 @@
 ]
 
 
-async def sleeping_forever(progress: TaskProgress) -> None:
+async def sleeping_very_long(progress: TaskProgress) -> None:
     _ = progress
-    while True:  # noqa: ASYNC110
-        await asyncio.sleep(1)
+    await asyncio.sleep(10_000)
 
 
-TaskRegistry.register(sleeping_forever)
+TaskRegistry.register(sleeping_very_long)
 
 
 @pytest.fixture
@@ -68,7 +67,7 @@ async def test_cleanup_long_running_tasks(test_client: TestClient) -> None:
     await lrt_api.start_task(
         long_running_manager.rpc_client,
         long_running_manager.lrt_namespace,
-        sleeping_forever.__name__,
+        sleeping_very_long.__name__,
     )
 
     _assert_long_running_tasks_count(long_running_manager, count=1)

services/dynamic-sidecar/tests/unit/api/rpc/test__long_running_tasks.py

@@ -0,0 +1,84 @@
+# pylint: disable=protected-access
+# pylint: disable=redefined-outer-name
+# pylint: disable=unused-argument
+
+import asyncio
+import json
+
+import pytest
+from common_library.serialization import model_dump_with_secrets
+from fastapi import FastAPI
+from models_library.api_schemas_long_running_tasks.base import TaskProgress
+from pytest_simcore.helpers.monkeypatch_envs import EnvVarsDict, setenvs_from_dict
+from servicelib.fastapi.long_running_tasks._manager import FastAPILongRunningManager
+from servicelib.fastapi.long_running_tasks.server import (
+    get_long_running_manager_from_app,
+)
+from servicelib.long_running_tasks import lrt_api
+from servicelib.long_running_tasks.task import TaskRegistry
+from servicelib.rabbitmq import RabbitMQRPCClient
+from servicelib.rabbitmq.rpc_interfaces.dynamic_sidecar import long_running_tasks
+from settings_library.rabbit import RabbitSettings
+from simcore_service_dynamic_sidecar.core.settings import ApplicationSettings
+
+pytest_simcore_core_services_selection = [
+    "rabbit",
+]
+
+
+async def sleeping_very_long(progress: TaskProgress) -> None:
+    _ = progress
+    await asyncio.sleep(10_000)
+
+
+TaskRegistry.register(sleeping_very_long)
+
+
+@pytest.fixture
+def mock_environment(
+    monkeypatch: pytest.MonkeyPatch,
+    rabbit_service: RabbitSettings,
+    mock_environment: EnvVarsDict,
+) -> EnvVarsDict:
+    return setenvs_from_dict(
+        monkeypatch,
+        {
+            **mock_environment,
+            "RABBIT_SETTINGS": json.dumps(
+                model_dump_with_secrets(rabbit_service, show_secrets=True)
+            ),
+        },
+    )
+
+
+def _assert_long_running_tasks_count(
+    long_running_manager: FastAPILongRunningManager, *, count: int
+) -> None:
+    assert (
+        len(long_running_manager.tasks_manager._created_tasks) == count  # noqa: SLF001
+    )
+
+
+async def test_cleanup_long_running_tasks(
+    app: FastAPI, rpc_client: RabbitMQRPCClient
+) -> None:
+    settings: ApplicationSettings = app.state.settings
+    long_running_manager = get_long_running_manager_from_app(app)
+
+    _assert_long_running_tasks_count(long_running_manager, count=0)
+
+    await lrt_api.start_task(
+        long_running_manager.rpc_client,
+        long_running_manager.lrt_namespace,
+        sleeping_very_long.__name__,
+    )
+
+    _assert_long_running_tasks_count(long_running_manager, count=1)
+
+    result = await long_running_tasks.cleanup_local_long_running_tasks(
+        rpc_client,
+        node_id=settings.DY_SIDECAR_NODE_ID,
+    )
+    assert result is None
+
+    _assert_long_running_tasks_count(long_running_manager, count=0)