added rpc interface for create_output_dirs

Author: Andrei Neagu

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

@@ -2,6 +2,7 @@
 
 from models_library.projects_nodes_io import NodeID
 from models_library.rabbitmq_basic_types import RPCMethodName
+from models_library.services import ServiceOutput
 from pydantic import TypeAdapter
 
 from ....logging_utils import log_decorator
@@ -27,3 +28,19 @@ async def toggle_ports_io(
         enable_inputs=enable_inputs,
     )
     assert result is None  # nosec
+
+
+@log_decorator(_logger, level=logging.DEBUG)
+async def create_output_dirs(
+    rabbitmq_rpc_client: RabbitMQRPCClient,
+    *,
+    node_id: NodeID,
+    outputs_labels: dict[str, ServiceOutput]
+) -> None:
+    rpc_namespace = get_rpc_namespace(node_id)
+    result = await rabbitmq_rpc_client.request(
+        rpc_namespace,
+        TypeAdapter(RPCMethodName).validate_python("create_output_dirs"),
+        outputs_labels=outputs_labels,
+    )
+    assert result is None  # nosec

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

@@ -1,4 +1,5 @@
 from fastapi import FastAPI
+from models_library.services import ServiceOutput
 from servicelib.rabbitmq import RPCRouter
 
 from ...services import container_extensions
@@ -13,3 +14,10 @@ async def toggle_ports_io(
     await container_extensions.toggle_ports_io(
         app, enable_outputs=enable_outputs, enable_inputs=enable_inputs
     )
+
+
+@router.expose()
+async def create_output_dirs(
+    app: FastAPI, *, outputs_labels: dict[str, ServiceOutput]
+) -> None:
+    await container_extensions.create_output_dirs(app, outputs_labels=outputs_labels)

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

@@ -1,13 +1,19 @@
 # pylint:disable=unused-argument
 # pylint:disable=redefined-outer-name
+# pylint:disable=protected-access
 
+import asyncio
+from typing import Final
 from unittest.mock import AsyncMock
 
 import pytest
 from fastapi import FastAPI
+from models_library.services import ServiceOutput
+from pydantic import TypeAdapter
 from pytest_mock import MockerFixture
 from servicelib.rabbitmq import RabbitMQRPCClient
 from servicelib.rabbitmq.rpc_interfaces.dynamic_sidecar import container_extensions
+from simcore_service_dynamic_sidecar.core.application import AppState
 from simcore_service_dynamic_sidecar.core.settings import ApplicationSettings
 from simcore_service_dynamic_sidecar.modules.inputs import InputsState
 from simcore_service_dynamic_sidecar.modules.outputs._watcher import OutputsWatcher
@@ -16,6 +22,8 @@
     "rabbit",
 ]
 
+_WAIT_FOR_OUTPUTS_WATCHER: Final[float] = 0.1
+
 
 def _assert_inputs_pulling(app: FastAPI, is_enabled: bool) -> None:
     inputs_state: InputsState = app.state.inputs_state
@@ -64,3 +72,62 @@ async def test_toggle_ports_io(
 
     _assert_inputs_pulling(app, enabled)
     _assert_outputs_event_propagation(spy_output_watcher, enabled)
+
+
+@pytest.fixture
+def mock_outputs_labels() -> dict[str, ServiceOutput]:
+    return {
+        "output_port_1": TypeAdapter(ServiceOutput).validate_python(
+            ServiceOutput.model_json_schema()["examples"][3]
+        ),
+        "output_port_2": TypeAdapter(ServiceOutput).validate_python(
+            ServiceOutput.model_json_schema()["examples"][3]
+        ),
+    }
+
+
+@pytest.fixture
+def mock_event_filter_enqueue(
+    app: FastAPI, monkeypatch: pytest.MonkeyPatch
+) -> AsyncMock:
+    mock = AsyncMock(return_value=None)
+    outputs_watcher: OutputsWatcher = app.state.outputs_watcher
+    monkeypatch.setattr(outputs_watcher._event_filter, "enqueue", mock)  # noqa: SLF001
+    return mock
+
+
+async def test_container_create_outputs_dirs(
+    app: FastAPI,
+    rpc_client: RabbitMQRPCClient,
+    mock_outputs_labels: dict[str, ServiceOutput],
+    mock_event_filter_enqueue: AsyncMock,
+):
+    app_state = AppState(app)
+
+    # by default outputs-watcher it is disabled
+    result = await container_extensions.toggle_ports_io(
+        rpc_client,
+        node_id=app_state.settings.DY_SIDECAR_NODE_ID,
+        enable_outputs=True,
+        enable_inputs=True,
+    )
+    assert result is None
+    await asyncio.sleep(_WAIT_FOR_OUTPUTS_WATCHER)
+
+    assert mock_event_filter_enqueue.call_count == 0
+
+    result = await container_extensions.create_output_dirs(
+        rpc_client,
+        node_id=app_state.settings.DY_SIDECAR_NODE_ID,
+        outputs_labels=mock_outputs_labels,
+    )
+
+    for dir_name in mock_outputs_labels:
+        assert (app_state.mounted_volumes.disk_outputs_path / dir_name).is_dir()
+
+    await asyncio.sleep(_WAIT_FOR_OUTPUTS_WATCHER)
+    EXPECT_EVENTS_WHEN_CREATING_OUTPUT_PORT_KEY_DIRS = 0
+    assert (
+        mock_event_filter_enqueue.call_count
+        == EXPECT_EVENTS_WHEN_CREATING_OUTPUT_PORT_KEY_DIRS
+    )