extraced ServiceManager

Author: Andrei Neagu

packages/pytest-simcore/src/pytest_simcore/helpers/docker.py

@@ -1,17 +1,24 @@
+import contextlib
 import json
 import logging
 import os
 import re
 import subprocess
+from collections.abc import AsyncIterator, Callable
+from contextlib import AbstractAsyncContextManager
 from enum import Enum
 from pathlib import Path
-from typing import Any
+from typing import Any, Protocol
 
 import docker
 import yaml
+from servicelib.rabbitmq import RabbitMQClient
+from servicelib.redis import RedisClientSDK
 from tenacity import retry
 from tenacity.after import after_log
-from tenacity.stop import stop_after_attempt
+from tenacity.asyncio import AsyncRetrying
+from tenacity.retry import retry_if_exception_type
+from tenacity.stop import stop_after_attempt, stop_after_delay
 from tenacity.wait import wait_fixed
 
 
@@ -272,3 +279,56 @@ def save_docker_infos(destination_dir: Path):
             f"wrote docker log and json files for {len(all_containers)} containers in ",
             destination_dir,
         )
+
+
+class _ClientWithPingProtocol(Protocol):
+    async def ping(self) -> bool: ...
+
+
+class ServiceManager:
+    def __init__(
+        self,
+        redis_client: RedisClientSDK,
+        rabbit_client: RabbitMQClient,
+        paused_container: Callable[[str], AbstractAsyncContextManager[None]],
+    ) -> None:
+        self.redis_client = redis_client
+        self.rabbit_client = rabbit_client
+        self.paused_container = paused_container
+
+    @contextlib.asynccontextmanager
+    async def _paused_container(
+        self, container_name: str, client: _ClientWithPingProtocol
+    ) -> AsyncIterator[None]:
+        async with self.paused_container(container_name):
+            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 await client.ping() is False
+            yield
+
+        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 await client.ping() is True
+
+    @contextlib.asynccontextmanager
+    async def pause_rabbit(self) -> AsyncIterator[None]:
+        async with self._paused_container("rabbit", self.rabbit_client):
+            yield
+
+    @contextlib.asynccontextmanager
+    async def pause_redis(self) -> AsyncIterator[None]:
+        # save db for clean restore point
+        await self.redis_client.redis.save()
+
+        async with self._paused_container("redis", self.redis_client):
+            yield

packages/service-library/tests/deferred_tasks/test_deferred_tasks.py

@@ -2,16 +2,15 @@
 # pylint:disable=unused-argument
 
 import asyncio
-import contextlib
 import datetime
 import itertools
 import json
 import random
 import sys
-from collections.abc import AsyncIterable, AsyncIterator, Awaitable, Callable
+from collections.abc import AsyncIterable, Awaitable, Callable
 from contextlib import AbstractAsyncContextManager, AsyncExitStack, suppress
 from pathlib import Path
-from typing import Any, Protocol
+from typing import Any
 
 import psutil
 import pytest
@@ -20,6 +19,7 @@
 from common_library.serialization import model_dump_with_secrets
 from pydantic import NonNegativeFloat, NonNegativeInt
 from pytest_mock import MockerFixture
+from pytest_simcore.helpers.docker import ServiceManager
 from servicelib.rabbitmq import RabbitMQClient
 from servicelib.redis import RedisClientSDK
 from servicelib.sequences_utils import partition_gen
@@ -332,59 +332,6 @@ async def rabbit_client(
     return create_rabbitmq_client("pinger")
 
 
-class ClientWithPingProtocol(Protocol):
-    async def ping(self) -> bool: ...
-
-
-class ServiceManager:
-    def __init__(
-        self,
-        redis_client: RedisClientSDK,
-        rabbit_client: RabbitMQClient,
-        paused_container: Callable[[str], AbstractAsyncContextManager[None]],
-    ) -> None:
-        self.redis_client = redis_client
-        self.rabbit_client = rabbit_client
-        self.paused_container = paused_container
-
-    @contextlib.asynccontextmanager
-    async def _paused_container(
-        self, container_name: str, client: ClientWithPingProtocol
-    ) -> AsyncIterator[None]:
-        async with self.paused_container(container_name):
-            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 await client.ping() is False
-            yield
-
-        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 await client.ping() is True
-
-    @contextlib.asynccontextmanager
-    async def pause_rabbit(self) -> AsyncIterator[None]:
-        async with self._paused_container("rabbit", self.rabbit_client):
-            yield
-
-    @contextlib.asynccontextmanager
-    async def pause_redis(self) -> AsyncIterator[None]:
-        # save db for clean restore point
-        await self.redis_client.redis.save()
-
-        async with self._paused_container("redis", self.redis_client):
-            yield
-
-
 @pytest.fixture
 def mock_default_socket_timeout(mocker: MockerFixture) -> None:
     mocker.patch(