initial test

Author: sanderegg

packages/service-library/src/servicelib/rabbitmq/rpc_interfaces/async_jobs/async_jobs.py

@@ -23,7 +23,7 @@ async def abort(
     *,
     rpc_namespace: RPCNamespace,
     job_id: AsyncJobId,
-    job_id_data: AsyncJobNameData
+    job_id_data: AsyncJobNameData,
 ) -> AsyncJobAbort:
     result = await rabbitmq_rpc_client.request(
         rpc_namespace,
@@ -41,7 +41,7 @@ async def get_status(
     *,
     rpc_namespace: RPCNamespace,
     job_id: AsyncJobId,
-    job_id_data: AsyncJobNameData
+    job_id_data: AsyncJobNameData,
 ) -> AsyncJobStatus:
     result = await rabbitmq_rpc_client.request(
         rpc_namespace,
@@ -59,7 +59,7 @@ async def get_result(
     *,
     rpc_namespace: RPCNamespace,
     job_id: AsyncJobId,
-    job_id_data: AsyncJobNameData
+    job_id_data: AsyncJobNameData,
 ) -> AsyncJobResult:
     result = await rabbitmq_rpc_client.request(
         rpc_namespace,
@@ -77,7 +77,7 @@ async def list_jobs(
     *,
     rpc_namespace: RPCNamespace,
     filter_: str,
-    job_id_data: AsyncJobNameData
+    job_id_data: AsyncJobNameData,
 ) -> list[AsyncJobGet]:
     result: list[AsyncJobGet] = await rabbitmq_rpc_client.request(
         rpc_namespace,
@@ -95,7 +95,7 @@ async def submit_job(
     rpc_namespace: RPCNamespace,
     method_name: str,
     job_id_data: AsyncJobNameData,
-    **kwargs
+    **kwargs,
 ) -> AsyncJobGet:
     result = await rabbitmq_rpc_client.request(
         rpc_namespace,
@@ -104,5 +104,5 @@ async def submit_job(
         **kwargs,
         timeout_s=_DEFAULT_TIMEOUT_S,
     )
-    assert isinstance(result, AsyncJobGet)
+    assert isinstance(result, AsyncJobGet)  # nosec
     return result

packages/service-library/src/servicelib/rabbitmq/rpc_interfaces/storage/paths.py

@@ -20,13 +20,14 @@ async def compute_path_size(
     product_name: str,
     location_id: LocationID,
     path: Path,
-) -> StorageAsyncJobGet:
+) -> tuple[StorageAsyncJobGet, AsyncJobNameData]:
+    job_id_data = AsyncJobNameData(user_id=user_id, product_name=product_name)
     async_job_rpc_get = await submit_job(
         rabbitmq_rpc_client=client,
         rpc_namespace=STORAGE_RPC_NAMESPACE,
         method_name=RPCMethodName("compute_path_size"),
-        job_id_data=AsyncJobNameData(user_id=user_id, product_name=product_name),
+        job_id_data=job_id_data,
         location_id=location_id,
         path=path,
     )
-    return StorageAsyncJobGet.from_rpc_schema(async_job_rpc_get)
+    return StorageAsyncJobGet.from_rpc_schema(async_job_rpc_get), job_id_data

services/storage/src/simcore_service_storage/api/rpc/_paths.py

@@ -6,20 +6,21 @@
 from models_library.api_schemas_rpc_async_jobs.async_jobs import (
     AsyncJobGet,
     AsyncJobId,
+    AsyncJobNameData,
 )
 from models_library.projects_nodes_io import LocationID
-from models_library.users import UserID
 from servicelib.rabbitmq import RPCRouter
 
 from ...dsm import get_dsm_provider
 
 router = RPCRouter()
 
 
-@router.expose(reraise_if_error_type=())
+@router.expose(reraise_if_error_type=None)
 async def compute_path_size(
     app: FastAPI,
-    user_id: UserID,
+    job_id_data: AsyncJobNameData,
+    # user_id: UserID,
     location_id: LocationID,
     path: Path,
 ) -> AsyncJobGet:
@@ -28,7 +29,8 @@ async def compute_path_size(
     dsm = get_dsm_provider(app).get(location_id)
     # TODO: this must be send to Celery!
     task = asyncio.create_task(
-        dsm.compute_path_size(user_id, path=path), name="THISSHALLGOTOCELERY"
+        dsm.compute_path_size(job_id_data.user_id, path=path),
+        name="THISSHALLGOTOCELERY",
     )
     await asyncio.sleep(5)
 

services/storage/src/simcore_service_storage/api/rpc/routes.py

@@ -6,12 +6,16 @@
 from servicelib.rabbitmq import RPCRouter
 
 from ...modules.rabbitmq import get_rabbitmq_rpc_server
-from . import _async_jobs, _data_export
+from . import _async_jobs, _data_export, _paths
 
 _logger = logging.getLogger(__name__)
 
 
-ROUTERS: list[RPCRouter] = [_data_export.router, _async_jobs.router]
+ROUTERS: list[RPCRouter] = [
+    _async_jobs.router,
+    _data_export.router,
+    _paths.router,
+]
 
 
 def setup_rpc_api_routes(app: FastAPI) -> None:

services/storage/tests/conftest.py

@@ -57,6 +57,7 @@
 from pytest_simcore.helpers.typing_env import EnvVarsDict
 from servicelib.aiohttp import status
 from servicelib.utils import limited_gather
+from settings_library.rabbit import RabbitSettings
 from simcore_postgres_database.models.tokens import tokens
 from simcore_postgres_database.storage_models import file_meta_data, projects, users
 from simcore_service_storage.core.application import create_app
@@ -166,7 +167,6 @@ def app_environment(
     mock_env_devel_environment: EnvVarsDict,
     monkeypatch: pytest.MonkeyPatch,
     external_envfile_dict: EnvVarsDict,
-    mock_rabbit_setup: MockerFixture,
 ) -> EnvVarsDict:
     if external_envfile_dict:
         delenvs_from_dict(monkeypatch, mock_env_devel_environment, raising=False)
@@ -176,6 +176,18 @@ def app_environment(
     return mock_env_devel_environment | envs
 
 
+@pytest.fixture
+def disabled_rabbitmq(app_environment: EnvVarsDict, monkeypatch: pytest.MonkeyPatch):
+    monkeypatch.setenv("STORAGE_RABBITMQ", "null")
+
+
+@pytest.fixture
+def enabled_rabbitmq(
+    app_environment: EnvVarsDict, rabbit_service: RabbitSettings
+) -> RabbitSettings:
+    return rabbit_service
+
+
 @pytest.fixture
 def app_settings(
     app_environment: EnvVarsDict,

services/storage/tests/unit/test_rpc_handlers_paths.py

@@ -7,6 +7,7 @@
 # pylint:disable=unused-variable
 
 
+import logging
 import random
 from collections.abc import Awaitable, Callable
 from pathlib import Path
@@ -16,16 +17,27 @@
 import pytest
 from faker import Faker
 from fastapi import FastAPI
-from models_library.api_schemas_storage.storage_schemas import (
-    PathTotalSizeCreate,
-)
+from models_library.api_schemas_rpc_async_jobs.async_jobs import AsyncJobId
+from models_library.api_schemas_storage import STORAGE_RPC_NAMESPACE
+from models_library.api_schemas_webserver.storage import StorageAsyncJobGet
 from models_library.projects_nodes_io import LocationID, NodeID, SimcoreS3FileID
 from models_library.users import UserID
 from pydantic import ByteSize, TypeAdapter
 from pytest_simcore.helpers.storage_utils import FileIDDict, ProjectWithFilesParams
 from servicelib.rabbitmq._client_rpc import RabbitMQRPCClient
+from servicelib.rabbitmq.rpc_interfaces.async_jobs.async_jobs import (
+    get_result,
+    get_status,
+)
 from servicelib.rabbitmq.rpc_interfaces.storage.paths import compute_path_size
 from simcore_service_storage.simcore_s3_dsm import SimcoreS3DataManager
+from tenacity import (
+    before_sleep_log,
+    retry,
+    retry_if_exception_type,
+    stop_after_delay,
+    wait_fixed,
+)
 
 pytest_simcore_core_services_selection = ["postgres", "rabbit"]
 pytest_simcore_ops_services_selection = ["adminer"]
@@ -59,6 +71,9 @@ def _filter_and_group_paths_one_level_deeper(
     )
 
 
+_logger = logging.getLogger(__name__)
+
+
 async def _assert_compute_path_size(
     rpc_client: RabbitMQRPCClient,
     location_id: LocationID,
@@ -67,16 +82,40 @@ async def _assert_compute_path_size(
     path: Path,
     expected_total_size: int,
 ) -> ByteSize:
-    received = await compute_path_size(
+    received, job_id_data = await compute_path_size(
         rpc_client, user_id=user_id, product_name="", location_id=location_id, path=path
     )
 
-    assert isinstance(received, PathTotalSizeCreate)
+    assert isinstance(received, StorageAsyncJobGet)
 
-    assert received
-    assert received.path == path
-    assert received.size == expected_total_size
-    return received.size
+    @retry(
+        wait=wait_fixed(1),
+        stop=stop_after_delay(10),
+        retry=retry_if_exception_type(AssertionError),
+        before_sleep=before_sleep_log(_logger, logging.WARNING),
+    )
+    async def _wait_for_job_completion(job_id: AsyncJobId) -> None:
+        job_status = await get_status(
+            rpc_client,
+            rpc_namespace=STORAGE_RPC_NAMESPACE,
+            job_id=job_id,
+            job_id_data=job_id_data,
+        )
+        assert job_status.done
+
+    await _wait_for_job_completion(received.job_id)
+    job_result = await get_result(
+        rpc_client,
+        rpc_namespace=STORAGE_RPC_NAMESPACE,
+        job_id=received.job_id,
+        job_id_data=job_id_data,
+    )
+    assert job_result.result is not None
+    assert job_result.error is None
+    response = job_result.result
+    assert isinstance(response, ByteSize)
+    assert response == expected_total_size
+    return response
 
 
 @pytest.mark.parametrize(
@@ -97,8 +136,8 @@ async def _assert_compute_path_size(
     ids=str,
 )
 async def test_path_compute_size(
+    enabled_rabbitmq,
     initialized_app: FastAPI,
-    client: httpx.AsyncClient,
     storage_rabbitmq_rpc_client: RabbitMQRPCClient,
     location_id: LocationID,
     user_id: UserID,