Added examples/rabbitmq_management

.github/workflows/examples.yml

@@ -88,6 +88,39 @@ jobs:
         run: |
           pytest -vv tests -n auto
 
+  rabbitmq_management:
+    runs-on: ${{ matrix.os }}
+
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ["3.12"]
+        os: ["ubuntu-latest"]
+
+    steps:
+      - name: Install apt packages
+        if: startsWith(matrix.os, 'ubuntu-')
+        run: |
+          sudo apt update
+      - uses: actions/checkout@v4
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python-version }}
+          cache: 'pip'
+          cache-dependency-path: '**/setup.py'
+      - name: Install dependencies
+        working-directory: examples/rabbitmq_management
+        run: |
+          python -m pip install --upgrade pip
+          pip install -r requirements.txt
+
+      - name: Run tests
+        working-directory: examples/rabbitmq_management
+        timeout-minutes: 5
+        run: |
+          pytest -vv tests -n auto
+
   django:
     runs-on: ${{ matrix.os }}
 

examples/rabbitmq_management/requirements.txt

@@ -0,0 +1,4 @@
+pytest>=7.4.4
+# pytest-celery>=1.0.0
+git+https://github.com/celery/pytest-celery.git
+pytest-xdist>=3.5.0

examples/rabbitmq_management/tests/conftest.py

@@ -0,0 +1,42 @@
+import pytest
+
+from pytest_celery import RABBITMQ_PORTS
+from pytest_celery import CeleryBrokerCluster
+from pytest_celery import RabbitMQContainer
+from pytest_celery import RabbitMQTestBroker
+
+
+@pytest.fixture
+def default_rabbitmq_broker_image() -> str:
+    return "rabbitmq:management"
+
+
+@pytest.fixture
+def default_rabbitmq_broker_ports() -> dict:
+    # Expose the management UI port
+    ports = RABBITMQ_PORTS.copy()
+    ports.update({"15672/tcp": None})
+    return ports
+
+
+class RabbitMQManagementTestBroker(RabbitMQTestBroker):
+    def get_management_url(self) -> str:
+        ip = self.container.attrs["NetworkSettings"]["Ports"]["15672/tcp"][0]["HostIp"]
+        port = self.container.attrs["NetworkSettings"]["Ports"]["15672/tcp"][0]["HostPort"]
+        # Opening this link during debugging allows you to see the RabbitMQ management UI
+        # in your browser
+        return f"http://{ip}:{port}"
+
+
+@pytest.fixture
+def celery_rabbitmq_broker(default_rabbitmq_broker: RabbitMQContainer) -> RabbitMQTestBroker:
+    broker = RabbitMQManagementTestBroker(default_rabbitmq_broker)
+    yield broker
+    broker.teardown()
+
+
+@pytest.fixture
+def celery_broker_cluster(celery_rabbitmq_broker: RabbitMQTestBroker) -> CeleryBrokerCluster:  # type: ignore
+    cluster = CeleryBrokerCluster(celery_rabbitmq_broker)  # type: ignore
+    yield cluster
+    cluster.teardown()

examples/rabbitmq_management/tests/test_management_broker.py

@@ -0,0 +1,23 @@
+import requests
+from requests.auth import HTTPBasicAuth
+
+from pytest_celery import CeleryTestSetup
+from tests.conftest import RabbitMQManagementTestBroker
+
+
+def test_login_to_broker_alone(celery_rabbitmq_broker: RabbitMQManagementTestBroker):
+    api = celery_rabbitmq_broker.get_management_url() + "/api/whoami"
+    response = requests.get(api, auth=HTTPBasicAuth("guest", "guest"))
+    assert response.status_code == 200
+    assert response.json()["name"] == "guest"
+    assert response.json()["tags"] == ["administrator"]
+
+
+def test_broker_in_setup(celery_setup: CeleryTestSetup):
+    assert isinstance(celery_setup.broker, RabbitMQManagementTestBroker)
+    api = celery_setup.broker.get_management_url() + "/api/queues"
+    response = requests.get(api, auth=HTTPBasicAuth("guest", "guest"))
+    assert response.status_code == 200
+    res = response.json()
+    assert isinstance(res, list)
+    assert len(list(filter(lambda queues: celery_setup.worker.hostname() in queues["name"], res))) == 1