properly test sending messages

packages/pytest-simcore/src/pytest_simcore/rabbit_service.py

@@ -11,6 +11,7 @@
 import aio_pika
 import pytest
 import tenacity
+from servicelib.rabbitmq import RabbitMQClient
 from settings_library.rabbit import RabbitSettings
 from tenacity.before_sleep import before_sleep_log
 from tenacity.stop import stop_after_attempt
@@ -32,7 +33,7 @@ async def wait_till_rabbit_responsive(url: str) -> None:
     await connection.close()
 
 
-@pytest.fixture(scope="function")
+@pytest.fixture
 async def rabbit_settings(
     docker_stack: dict, testing_environ_vars: dict  # stack is up
 ) -> RabbitSettings:
@@ -55,7 +56,7 @@ async def rabbit_settings(
     return settings
 
 
-@pytest.fixture(scope="function")
+@pytest.fixture
 async def rabbit_service(
     rabbit_settings: RabbitSettings, monkeypatch: pytest.MonkeyPatch
 ) -> RabbitSettings:
@@ -73,7 +74,7 @@ async def rabbit_service(
     return rabbit_settings
 
 
-@pytest.fixture(scope="function")
+@pytest.fixture
 async def rabbit_connection(
     rabbit_settings: RabbitSettings,
 ) -> AsyncIterator[aio_pika.abc.AbstractConnection]:
@@ -103,7 +104,7 @@ def _connection_close_callback(sender: Any, exc: Optional[BaseException] = None)
     assert connection.is_closed
 
 
-@pytest.fixture(scope="function")
+@pytest.fixture
 async def rabbit_channel(
     rabbit_connection: aio_pika.abc.AbstractConnection,
 ) -> AsyncIterator[aio_pika.abc.AbstractChannel]:
@@ -118,3 +119,13 @@ def _channel_close_callback(sender: Any, exc: Optional[BaseException] = None):
         channel.close_callbacks.add(_channel_close_callback)
         yield channel
     assert channel.is_closed
+
+
+@pytest.fixture
+async def rabbit_client(
+    rabbit_settings: RabbitSettings,
+) -> AsyncIterator[RabbitMQClient]:
+    client = RabbitMQClient("pytest", settings=rabbit_settings)
+    assert client
+    yield client
+    await client.close()

services/autoscaling/tests/unit/conftest.py

@@ -32,6 +32,7 @@
 from moto.server import ThreadedMotoServer
 from pydantic import ByteSize, PositiveInt
 from pytest import MonkeyPatch
+from pytest_mock.plugin import MockerFixture
 from pytest_simcore.helpers.utils_docker import get_localhost_ip
 from pytest_simcore.helpers.utils_envs import EnvVarsDict, setenvs_from_dict
 from settings_library.rabbit import RabbitSettings
@@ -104,6 +105,21 @@ def app_environment(
     return mock_env_devel_environment | envs
 
 
+@pytest.fixture
+def disable_dynamic_service_background_task(mocker: MockerFixture) -> Iterator[None]:
+    mocker.patch(
+        "simcore_service_autoscaling.dynamic_scaling.start_periodic_task",
+        autospec=True,
+    )
+
+    mocker.patch(
+        "simcore_service_autoscaling.dynamic_scaling.stop_periodic_task",
+        autospec=True,
+    )
+
+    yield
+
+
 @pytest.fixture
 def disabled_rabbitmq(app_environment: EnvVarsDict, monkeypatch: pytest.MonkeyPatch):
     monkeypatch.delenv("RABBIT_HOST")

services/autoscaling/tests/unit/test_dynamic_scaling_core.py

@@ -18,21 +18,6 @@
 from simcore_service_autoscaling.utils_aws import EC2Client
 
 
-@pytest.fixture
-def disable_dynamic_service_background_task(mocker: MockerFixture) -> Iterator[None]:
-    mocker.patch(
-        "simcore_service_autoscaling.dynamic_scaling.start_periodic_task",
-        autospec=True,
-    )
-
-    mocker.patch(
-        "simcore_service_autoscaling.dynamic_scaling.stop_periodic_task",
-        autospec=True,
-    )
-
-    yield
-
-
 @pytest.fixture
 def aws_instance_private_dns() -> str:
     return "ip-10-23-40-12.ec2.internal"

services/autoscaling/tests/unit/test_rabbitmq.py

@@ -15,14 +15,24 @@
     RabbitAutoscalingMessage,
     RabbitMessageBase,
 )
+from pytest_mock.plugin import MockerFixture
+from servicelib.rabbitmq import RabbitMQClient
 from settings_library.rabbit import RabbitSettings
 from simcore_service_autoscaling.core.errors import ConfigurationError
 from simcore_service_autoscaling.rabbitmq import get_rabbitmq_client, send_message
 from tenacity import retry
+from tenacity._asyncio import AsyncRetrying
 from tenacity.retry import retry_if_exception_type
 from tenacity.stop import stop_after_delay
 from tenacity.wait import wait_fixed
 
+_TENACITY_RETRY_PARAMS = dict(
+    reraise=True,
+    retry=retry_if_exception_type(AssertionError),
+    stop=stop_after_delay(30),
+    wait=wait_fixed(0.5),
+)
+
 # Selection of core and tool services started in this swarm fixture (integration)
 pytest_simcore_core_services_selection = [
     "rabbit",
@@ -83,12 +93,27 @@ def test_rabbitmq_initializes(
 
 
 async def test_send_message(
+    disable_dynamic_service_background_task,
     enabled_rabbitmq: RabbitSettings,
     initialized_app: FastAPI,
     rabbit_message: RabbitMessageBase,
+    rabbit_client: RabbitMQClient,
+    mocker: MockerFixture,
 ):
+    mocked_message_handler = mocker.AsyncMock(return_value=True)
+    await rabbit_client.subscribe(rabbit_message.channel_name, mocked_message_handler)
     await send_message(initialized_app, message=rabbit_message)
 
+    async for attempt in AsyncRetrying(**_TENACITY_RETRY_PARAMS):
+        with attempt:
+            print(
+                f"--> checking for message in rabbit exchange {rabbit_message.channel_name}, {attempt.retry_state.retry_object.statistics}"
+            )
+            mocked_message_handler.assert_called_once_with(
+                rabbit_message.json().encode()
+            )
+            print("... message received")
+
 
 async def test_send_message_with_disabled_rabbit_does_not_raise(
     disabled_rabbitmq: None,
@@ -109,12 +134,7 @@ async def _switch_off_rabbit_mq_instance(async_docker_client: aiodocker.Docker)
         *(async_docker_client.services.delete(s["ID"]) for s in rabbit_services)
     )
 
-    @retry(
-        retry=retry_if_exception_type(AssertionError),
-        reraise=True,
-        wait=wait_fixed(0.5),
-        stop=stop_after_delay(30),
-    )
+    @retry(**_TENACITY_RETRY_PARAMS)
     async def _check_service_task_gone(service: Mapping[str, Any]) -> None:
         print(
             f"--> checking if service {service['ID']}:{service['Spec']['Name']} is really gone..."