diff --git a/.github/workflows/python.yml b/.github/workflows/python.yml index bf16f76..41d96a9 100644 --- a/.github/workflows/python.yml +++ b/.github/workflows/python.yml @@ -35,6 +35,21 @@ jobs: matrix: python-version: ["3.12", "3.13", "3.14"] + services: + rabbitmq: + image: rabbitmq:3 + ports: + - 5672:5672 + options: >- + --health-cmd "rabbitmq-diagnostics -q check_port_connectivity" + --health-interval 5s + --health-timeout 5s + --health-retries 12 + + env: + RABBITMQ_HOST: localhost + RABBITMQ_PORT: 5672 + steps: - uses: actions/checkout@v7 - uses: actions/download-artifact@v8 diff --git a/tests/transport/conftest.py b/tests/transport/conftest.py new file mode 100644 index 0000000..346296f --- /dev/null +++ b/tests/transport/conftest.py @@ -0,0 +1,194 @@ +""" +Provides a RabbitMQ broker for the transport integration tests. + +The broker is located, in order of preference: + +1. RABBITMQ_HOST / RABBITMQ_PORT environment variables, if either is set +2. An existing broker answering on localhost:5672 +3. A temporary rabbitmq container, started with docker or podman + +If none of these are available, or the container fails to start (e.g. +the image can't be pulled while offline), then the tests needing a broker +are skipped - unless the CI environment variable is set, in which case they +fail, so that a broken CI broker cannot silently skip the tests. +""" + +from __future__ import annotations + +import dataclasses +import logging +import os +import shutil +import subprocess +import time +from collections.abc import Iterator +from typing import NoReturn + +import pika +import pika.exceptions +import pytest + +logger = logging.getLogger(__name__) + +RABBITMQ_IMAGE = "rabbitmq:3" +BROKER_STARTUP_TIMEOUT = 60 + + +@dataclasses.dataclass(frozen=True) +class RabbitMQBroker: + host: str + port: int + username: str = "guest" + password: str = "guest" + + def connection_parameters(self, **kwargs) -> pika.ConnectionParameters: + return pika.ConnectionParameters( + self.host, + self.port, + "/", + pika.PlainCredentials(self.username, self.password), + **kwargs, + ) + + def is_available(self) -> bool: + """Check whether an AMQP connection can be made to this broker.""" + try: + pika.BlockingConnection( + self.connection_parameters( + connection_attempts=1, socket_timeout=1, stack_timeout=2 + ) + ).close() + except pika.exceptions.AMQPError: + return False + return True + + def wait_until_available(self, timeout: float) -> bool: + """Retry connecting until the broker accepts connections, or timeout.""" + start = time.monotonic() + while not self.is_available(): + if time.monotonic() - start > timeout: + return False + time.sleep(0.5) + logger.info( + "RabbitMQ at %s:%d available after %.1fs", + self.host, + self.port, + time.monotonic() - start, + ) + return True + + +def _no_broker(reason: str) -> NoReturn: + if os.environ.get("CI"): + pytest.fail(f"No RabbitMQ broker available in CI: {reason}") + pytest.skip(f"No RabbitMQ broker available: {reason}") + + +def _find_container_runtime() -> str | None: + """Find a container runtime command with a running daemon.""" + for runtime in ["docker", "podman"]: + if command := shutil.which(runtime): + if ( + subprocess.run([command, "version"], capture_output=True).returncode + == 0 + ): + return command + return None + + +class ContainerStartupError(RuntimeError): + """The broker container could not be started, or never became available.""" + + +def _container_host_port(runtime: str, container_id: str) -> int: + """Read the host port that the container's AMQP port is published on.""" + result = subprocess.run( + [runtime, "port", container_id, "5672/tcp"], + capture_output=True, + text=True, + ) + if result.returncode: + raise ContainerStartupError( + f"Could not read container port mapping: {result.stderr.strip()}" + ) + # e.g. "127.0.0.1:49153", possibly followed by further (IPv6) bindings + return int(result.stdout.splitlines()[0].rsplit(":", 1)[1]) + + +def _start_broker_container(runtime: str) -> tuple[RabbitMQBroker, str]: + """ + Start a broker container and wait for it to accept connections. + + Returns: + The broker, and the container ID to remove when finished with it. + + Raises: + ContainerStartupError: If the broker could not be made available. Any + container that was started is removed before raising. + """ + logger.info("Starting %s container with %s", RABBITMQ_IMAGE, runtime) + result = subprocess.run( + [runtime, "run", "-d", "--rm", "-p", "127.0.0.1::5672", RABBITMQ_IMAGE], + capture_output=True, + text=True, + ) + if result.returncode: + raise ContainerStartupError( + f"'{runtime} run {RABBITMQ_IMAGE}' failed (exit {result.returncode}): " + f"{result.stderr.strip()}" + ) + container_id = result.stdout.strip() + try: + broker = RabbitMQBroker( + "127.0.0.1", _container_host_port(runtime, container_id) + ) + if not broker.wait_until_available(BROKER_STARTUP_TIMEOUT): + raise ContainerStartupError( + f"{RABBITMQ_IMAGE} container did not accept connections " + f"within {BROKER_STARTUP_TIMEOUT}s" + ) + except BaseException: + _remove_container(runtime, container_id) + raise + return broker, container_id + + +def _remove_container(runtime: str, container_id: str) -> None: + subprocess.run([runtime, "rm", "-f", container_id], capture_output=True) + + +@pytest.fixture(scope="session") +def rabbitmq_broker() -> Iterator[RabbitMQBroker]: + """A RabbitMQ broker, for tests that need a real server.""" + if "RABBITMQ_HOST" in os.environ or "RABBITMQ_PORT" in os.environ: + broker = RabbitMQBroker( + os.environ.get("RABBITMQ_HOST", "localhost"), + int(os.environ.get("RABBITMQ_PORT", 5672)), + ) + # An explicitly configured broker may still be starting (e.g. a CI + # service container whose health check passed early), so retry + if not broker.wait_until_available(BROKER_STARTUP_TIMEOUT): + _no_broker(f"Could not connect to {broker.host}:{broker.port}") + yield broker + return + + broker = RabbitMQBroker("localhost", 5672) + if broker.is_available(): + yield broker + return + + runtime = _find_container_runtime() + if runtime is None: + _no_broker( + "Nothing on localhost:5672, RABBITMQ_HOST/RABBITMQ_PORT unset, " + "and no running docker or podman to start one" + ) + try: + broker, container_id = _start_broker_container(runtime) + except ContainerStartupError as e: + _no_broker(str(e)) + else: + try: + yield broker + finally: + _remove_container(runtime, container_id) diff --git a/tests/transport/test_pika.py b/tests/transport/test_pika.py index ea998cf..f640468 100644 --- a/tests/transport/test_pika.py +++ b/tests/transport/test_pika.py @@ -49,10 +49,9 @@ def revert_classvariables(): @pytest.fixture -def pikatransport(revert_classvariables, connection_params): - # connection_params is unused here, but implements the fixture skipping - # logic following a single test, instead of attempting a connection for - # every individual test. +def pikatransport(revert_classvariables, rabbitmq_broker): + PikaTransport.defaults["--rabbit-host"] = rabbitmq_broker.host + PikaTransport.defaults["--rabbit-port"] = rabbitmq_broker.port pt = PikaTransport() pt.connect() yield pt @@ -831,27 +830,16 @@ def test_nack_message(mock_pikathread): @pytest.fixture(scope="session") -def connection_params(): +def connection_params(rabbitmq_broker): """Connection Parameters for connecting to a physical RabbitMQ server""" - params = [ - pika.ConnectionParameters( - "localhost", - 5672, - "/", - pika.PlainCredentials("guest", "guest"), + return [ + rabbitmq_broker.connection_parameters( connection_attempts=1, retry_delay=1, socket_timeout=1, stack_timeout=2, ) ] - # Try a connection here to make sure this is valid - try: - bc = pika.BlockingConnection(params) - except BaseException: - pytest.skip("Failed to create test RabbitMQ connection") - bc.close() - return params @pytest.fixture