Added myworker example (#164)

Author: Nusnus

.github/workflows/examples.yml

@@ -22,6 +22,39 @@ permissions:
   contents: read # to fetch code (actions/checkout)
 
 jobs:
+  myworker:
+    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/myworker
+        run: |
+          python -m pip install --upgrade pip
+          pip install -r requirements.txt
+
+      - name: Run tests
+        working-directory: examples/myworker
+        timeout-minutes: 5
+        run: |
+          pytest -vv tests -n auto
+
   django:
     runs-on: ${{ matrix.os }}
 

examples/myworker/pytest.ini

@@ -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/myworker/requirements.txt

@@ -0,0 +1,20 @@
+import pytest
+
+from pytest_celery.vendors.rabbitmq.defaults import RABBITMQ_PORTS
+from tests.myworker.myworker import myworker_container  # noqa
+from tests.myworker.myworker import myworker_image  # noqa
+from tests.myworker.myworker import myworker_worker  # noqa
+
+
+@pytest.fixture
+def default_rabbitmq_broker_image() -> str:
+    # Useful for debugging
+    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

examples/myworker/tests/__init__.py

@@ -0,0 +1,33 @@
+FROM python:3.11-bookworm
+
+# Create a user to run the worker
+RUN adduser --disabled-password --gecos "" test_user
+
+# Install system dependencies
+RUN apt-get update && apt-get install -y build-essential git
+
+# Set arguments
+ARG CELERY_LOG_LEVEL=INFO
+ARG CELERY_WORKER_NAME=my_worker
+ARG CELERY_WORKER_QUEUE=celery
+ENV LOG_LEVEL=$CELERY_LOG_LEVEL
+ENV WORKER_NAME=$CELERY_WORKER_NAME
+ENV WORKER_QUEUE=$CELERY_WORKER_QUEUE
+
+# Install packages
+WORKDIR /src
+
+COPY --chown=test_user:test_user requirements.txt .
+RUN pip install --no-cache-dir --upgrade pip
+RUN pip install -r ./requirements.txt
+RUN git clone https://github.com/celery/celery.git
+
+WORKDIR /src/celery
+
+RUN pip install -e .
+
+# Switch to the test_user
+USER test_user
+
+# Start the celery worker
+CMD celery -A app worker --loglevel=$LOG_LEVEL -n $WORKER_NAME@%h -Q $WORKER_QUEUE

examples/myworker/tests/conftest.py

@@ -0,0 +1,60 @@
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+from celery import Celery
+from pytest_docker_tools import build
+from pytest_docker_tools import container
+from pytest_docker_tools import fxtr
+
+from pytest_celery import CeleryTestWorker
+from pytest_celery import CeleryWorkerContainer
+from pytest_celery import defaults
+
+
+class MyWorkerContainer(CeleryWorkerContainer):
+    @property
+    def client(self) -> Any:
+        return self
+
+    @classmethod
+    def version(cls) -> str:
+        return "Celery main branch"
+
+    @classmethod
+    def log_level(cls) -> str:
+        return "INFO"
+
+    @classmethod
+    def worker_name(cls) -> str:
+        return "my_worker"
+
+    @classmethod
+    def worker_queue(cls) -> str:
+        return "myworker"
+
+
+myworker_image = build(
+    path=".",
+    dockerfile="tests/myworker/Dockerfile",
+    tag="pytest-celery/myworker:example",
+    buildargs=MyWorkerContainer.buildargs(),
+)
+
+
+myworker_container = container(
+    image="{myworker_image.id}",
+    environment=fxtr("default_worker_env"),
+    network="{default_pytest_celery_network.name}",
+    volumes={"{default_worker_volume.name}": defaults.DEFAULT_WORKER_VOLUME},
+    wrapper_class=MyWorkerContainer,
+    timeout=defaults.DEFAULT_WORKER_CONTAINER_TIMEOUT,
+)
+
+
+@pytest.fixture
+def myworker_worker(myworker_container: MyWorkerContainer, celery_setup_app: Celery) -> CeleryTestWorker:
+    worker = CeleryTestWorker(myworker_container, app=celery_setup_app)
+    yield worker
+    worker.teardown()

examples/myworker/tests/myworker/Dockerfile

@@ -0,0 +1,27 @@
+import pytest
+from celery.canvas import Signature
+from celery.result import AsyncResult
+
+from pytest_celery import RESULT_TIMEOUT
+from pytest_celery import CeleryTestSetup
+from pytest_celery import CeleryTestWorker
+from pytest_celery import CeleryWorkerCluster
+from pytest_celery import ping
+
+
+@pytest.fixture
+def celery_worker_cluster(
+    celery_worker: CeleryTestWorker,
+    myworker_worker: CeleryTestWorker,
+) -> CeleryWorkerCluster:
+    cluster = CeleryWorkerCluster(celery_worker, myworker_worker)  # type: ignore
+    yield cluster
+    cluster.teardown()
+
+
+def test_ping(celery_setup: CeleryTestSetup):
+    worker: CeleryTestWorker
+    for worker in celery_setup.worker_cluster:
+        sig: Signature = ping.s()
+        res: AsyncResult = sig.apply_async(queue=worker.worker_queue)
+        assert res.get(timeout=RESULT_TIMEOUT) == "pong"