From 24f519f088ec100ed5ece4d7de0085889875f14f Mon Sep 17 00:00:00 2001 From: Nicholas Devenish Date: Fri, 6 Nov 2026 16:20:54 +0000 Subject: [PATCH 1/2] Run the RabbitMQ integration tests against a real broker The pika transport tests that need a broker were always skipped, both locally and in CI, as nothing provided one. Add a session rabbitmq_broker fixture that uses, in order: RABBITMQ_HOST/RABBITMQ_PORT if set, an existing broker on localhost:5672, or a temporary rabbitmq:3 container started with docker or podman. If no broker can be found or started the tests are skipped locally, but fail when CI is set, so that they can't silently stop running again. Add a rabbitmq service to the GitHub Actions test job. --- .github/workflows/python.yml | 15 +++ tests/transport/conftest.py | 193 +++++++++++++++++++++++++++++++++++ tests/transport/test_pika.py | 24 ++--- 3 files changed, 214 insertions(+), 18 deletions(-) create mode 100644 tests/transport/conftest.py 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..67802ce --- /dev/null +++ b/tests/transport/conftest.py @@ -0,0 +1,193 @@ +""" +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)) + 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 From d9cacb05ccc02ab412cbd80af62b215172ed8550 Mon Sep 17 00:00:00 2001 From: Nicholas Devenish Date: Fri, 6 Nov 2026 16:38:23 +0000 Subject: [PATCH 2/2] Restructure broker fixture so container_id is always assigned before use Resolves a CodeQL 'potentially uninitialized local variable' alert. The previous code was correct, as _no_broker() never returns, but CodeQL doesn't follow NoReturn. --- tests/transport/conftest.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/tests/transport/conftest.py b/tests/transport/conftest.py index 67802ce..346296f 100644 --- a/tests/transport/conftest.py +++ b/tests/transport/conftest.py @@ -187,7 +187,8 @@ def rabbitmq_broker() -> Iterator[RabbitMQBroker]: broker, container_id = _start_broker_container(runtime) except ContainerStartupError as e: _no_broker(str(e)) - try: - yield broker - finally: - _remove_container(runtime, container_id) + else: + try: + yield broker + finally: + _remove_container(runtime, container_id)