moved stuff around

Author: sanderegg

services/storage/tests/conftest.py

@@ -21,6 +21,9 @@
 import simcore_service_storage
 from asgi_lifespan import LifespanManager
 from aws_library.s3 import SimcoreS3API
+from celery import Celery
+from celery.contrib.testing.worker import TestWorkController, start_worker
+from celery.signals import worker_init, worker_shutdown
 from faker import Faker
 from fakeredis.aioredis import FakeRedis
 from fastapi import FastAPI
@@ -41,7 +44,7 @@
 from models_library.users import UserID
 from models_library.utils.fastapi_encoders import jsonable_encoder
 from pydantic import ByteSize, TypeAdapter
-from pytest_mock import MockerFixture
+from pytest_mock import MockerFixture, MockFixture
 from pytest_simcore.helpers.fastapi import url_from_operation_id
 from pytest_simcore.helpers.httpx_assert_checks import assert_status
 from pytest_simcore.helpers.logging_tools import log_context
@@ -57,15 +60,22 @@
 )
 from pytest_simcore.helpers.typing_env import EnvVarsDict
 from servicelib.aiohttp import status
+from servicelib.rabbitmq._client_rpc import RabbitMQRPCClient
 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.api._worker_tasks.tasks import setup_worker_tasks
 from simcore_service_storage.core.application import create_app
 from simcore_service_storage.core.settings import ApplicationSettings
 from simcore_service_storage.datcore_dsm import DatCoreDataManager
 from simcore_service_storage.dsm import get_dsm_provider
 from simcore_service_storage.models import FileMetaData, FileMetaDataAtDB, S3BucketName
+from simcore_service_storage.modules.celery.signals import (
+    on_worker_init,
+    on_worker_shutdown,
+)
+from simcore_service_storage.modules.celery.worker import CeleryTaskQueueWorker
 from simcore_service_storage.modules.s3 import get_s3_client
 from simcore_service_storage.simcore_s3_dsm import SimcoreS3DataManager
 from sqlalchemy import literal_column
@@ -959,3 +969,66 @@ def celery_config() -> dict[str, Any]:
         "task_track_started": True,
         "task_send_sent_event": True,
     }
+
+
+@pytest.fixture
+def mock_celery_app(mocker: MockFixture, celery_config: dict[str, Any]) -> None:
+    celery_app = Celery(**celery_config)
+
+    mocker.patch(
+        "simcore_service_storage.modules.celery.create_app",
+        return_value=celery_app,
+    )
+
+
+@pytest.fixture
+async def celery_worker_controller(
+    app_environment: EnvVarsDict,
+    initialized_app: FastAPI,
+    celery_app: Celery,
+    monkeypatch: pytest.MonkeyPatch,
+) -> AsyncIterator[TestWorkController]:
+    # Signals must be explicitily connected
+    worker_init.connect(on_worker_init)
+    worker_shutdown.connect(on_worker_shutdown)
+
+    setup_worker_tasks(celery_app)
+
+    monkeypatch.setenv("STORAGE_WORKER_MODE", "true")
+    with start_worker(
+        celery_app,
+        pool="threads",
+        concurrency=1,
+        loglevel="info",
+        perform_ping_check=False,
+        worker_kwargs={"hostname": "celery@worker1"},
+    ) as worker:
+        worker_init.send(sender=worker)
+
+        # NOTE: wait for worker to be ready (sic)
+        await asyncio.sleep(5)
+        yield worker
+
+        worker_shutdown.send(sender=worker)
+
+
+@pytest.fixture
+def celery_worker(
+    celery_worker_controller: TestWorkController,
+) -> CeleryTaskQueueWorker:
+    assert isinstance(celery_worker_controller.app, Celery)
+    return CeleryTaskQueueWorker(celery_worker_controller.app)
+
+
+@pytest.fixture
+async def storage_rabbitmq_rpc_client(
+    rabbitmq_rpc_client: Callable[[str], Awaitable[RabbitMQRPCClient]],
+) -> RabbitMQRPCClient:
+    rpc_client = await rabbitmq_rpc_client("pytest_storage_rpc_client")
+    assert rpc_client
+    return rpc_client
+
+
+@pytest.fixture
+def product_name(faker: Faker) -> str:
+    return faker.name()

services/storage/tests/unit/test_rpc_handlers_paths.py

@@ -10,14 +10,10 @@
 import asyncio
 import datetime
 import random
-from collections.abc import AsyncIterator, Awaitable, Callable
 from pathlib import Path
 from typing import Any, TypeAlias
 
 import pytest
-from celery import Celery
-from celery.contrib.testing.worker import TestWorkController, start_worker
-from celery.signals import worker_init, worker_shutdown
 from faker import Faker
 from fastapi import FastAPI
 from models_library.api_schemas_rpc_async_jobs.async_jobs import (
@@ -29,19 +25,12 @@
 from models_library.rabbitmq_basic_types import RPCMethodName
 from models_library.users import UserID
 from pydantic import ByteSize, TypeAdapter
-from pytest_mock import MockFixture
 from pytest_simcore.helpers.storage_utils import FileIDDict, ProjectWithFilesParams
-from pytest_simcore.helpers.typing_env import EnvVarsDict
 from servicelib.rabbitmq._client_rpc import RabbitMQRPCClient
 from servicelib.rabbitmq.rpc_interfaces.async_jobs.async_jobs import (
     wait_and_get_result,
 )
 from servicelib.rabbitmq.rpc_interfaces.storage.paths import compute_path_size
-from simcore_service_storage.api._worker_tasks.tasks import setup_worker_tasks
-from simcore_service_storage.modules.celery.signals import (
-    on_worker_init,
-    on_worker_shutdown,
-)
 from simcore_service_storage.modules.celery.worker import CeleryTaskQueueWorker
 from simcore_service_storage.simcore_s3_dsm import SimcoreS3DataManager
 
@@ -103,69 +92,6 @@ async def _assert_compute_path_size(
     pytest.fail("Job did not finish")
 
 
-@pytest.fixture
-def mock_celery_app(mocker: MockFixture, celery_config: dict[str, Any]) -> None:
-    celery_app = Celery(**celery_config)
-
-    mocker.patch(
-        "simcore_service_storage.modules.celery.create_app",
-        return_value=celery_app,
-    )
-
-
-@pytest.fixture
-async def celery_worker_controller(
-    app_environment: EnvVarsDict,
-    initialized_app: FastAPI,
-    celery_app: Celery,
-    monkeypatch: pytest.MonkeyPatch,
-) -> AsyncIterator[TestWorkController]:
-    # Signals must be explicitily connected
-    worker_init.connect(on_worker_init)
-    worker_shutdown.connect(on_worker_shutdown)
-
-    setup_worker_tasks(celery_app)
-
-    monkeypatch.setenv("STORAGE_WORKER_MODE", "true")
-    with start_worker(
-        celery_app,
-        pool="threads",
-        concurrency=1,
-        loglevel="info",
-        perform_ping_check=False,
-        worker_kwargs={"hostname": "celery@worker1"},
-    ) as worker:
-        worker_init.send(sender=worker)
-
-        # NOTE: wait for worker to be ready (sic)
-        await asyncio.sleep(5)
-        yield worker
-
-        worker_shutdown.send(sender=worker)
-
-
-@pytest.fixture
-def celery_worker(
-    celery_worker_controller: TestWorkController,
-) -> CeleryTaskQueueWorker:
-    assert isinstance(celery_worker_controller.app, Celery)
-    return CeleryTaskQueueWorker(celery_worker_controller.app)
-
-
-@pytest.fixture
-async def storage_rabbitmq_rpc_client(
-    rabbitmq_rpc_client: Callable[[str], Awaitable[RabbitMQRPCClient]],
-) -> RabbitMQRPCClient:
-    rpc_client = await rabbitmq_rpc_client("pytest_storage_rpc_client")
-    assert rpc_client
-    return rpc_client
-
-
-@pytest.fixture
-def product_name(faker: Faker) -> str:
-    return faker.name()
-
-
 @pytest.mark.parametrize(
     "location_id",
     [SimcoreS3DataManager.get_location_id()],