Added examples/range (celery#169)

Author: Nusnus

.github/workflows/examples.yml

@@ -55,6 +55,39 @@ jobs:
         run: |
           pytest -vv tests -n auto
 
+  range:
+    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/range
+        run: |
+          python -m pip install --upgrade pip
+          pip install -r requirements.txt
+
+      - name: Run tests
+        working-directory: examples/range
+        timeout-minutes: 5
+        run: |
+          pytest -vv tests -n auto
+
   django:
     runs-on: ${{ matrix.os }}
 

examples/range/pytest.ini

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

examples/range/requirements.txt

@@ -0,0 +1,22 @@
+from __future__ import annotations
+
+import requests
+from packaging.version import parse as parse_version
+
+
+def get_celery_versions(start_version: str, end_version: str) -> list[str]:
+    url = "https://pypi.org/pypi/celery/json"
+    response = requests.get(url)
+    data = response.json()
+    all_versions = data["releases"].keys()
+
+    filtered_versions = [
+        v
+        for v in all_versions
+        if (
+            parse_version(start_version) <= parse_version(v) <= parse_version(end_version)
+            and not parse_version(v).is_prerelease
+        )
+    ]
+
+    return sorted(filtered_versions, key=parse_version)

examples/range/tests/__init__.py

@@ -0,0 +1,19 @@
+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 ping
+from tests.conftest import get_celery_versions
+
+
+class TestRange:
+    @pytest.fixture(scope="session", params=get_celery_versions("v4.4.7", "v5.0.0"))
+    def celery_version(self, request: pytest.FixtureRequest) -> str:
+        return request.param
+
+    def test_ping(self, celery_setup: CeleryTestSetup, celery_version: str):
+        sig: Signature = ping.s()
+        res: AsyncResult = sig.apply_async()
+        assert res.get(timeout=RESULT_TIMEOUT) == "pong"

examples/range/tests/conftest.py

@@ -0,0 +1,73 @@
+import pytest
+from pytest_docker_tools import build
+from pytest_docker_tools import container
+from pytest_docker_tools import fxtr
+
+from pytest_celery import DEFAULT_WORKER_CONTAINER_TIMEOUT
+from pytest_celery import DEFAULT_WORKER_VOLUME
+from pytest_celery import WORKER_DOCKERFILE_ROOTDIR
+from pytest_celery import CeleryTestSetup
+from pytest_celery import CeleryTestWorker
+from pytest_celery import CeleryWorkerCluster
+from pytest_celery import CeleryWorkerContainer
+from tests.conftest import get_celery_versions
+
+versions_range = get_celery_versions("v5.0.0", "v5.0.5")
+versions_list = ["v4.4.7", "v5.2.7", "v5.3.0"]
+
+
+def generate_workers(versions: list[str]) -> list[str]:
+    worker_containers = list()
+    for v in versions:
+        img = f"worker_v{v.replace('.', '_')}_image"
+        globals()[img] = build(
+            path=WORKER_DOCKERFILE_ROOTDIR,
+            tag=f"pytest-celery/examples/worker:v{v}",
+            buildargs={
+                "CELERY_VERSION": v,
+                "CELERY_LOG_LEVEL": fxtr("default_worker_celery_log_level"),
+                "CELERY_WORKER_NAME": fxtr("default_worker_celery_worker_name"),
+                "CELERY_WORKER_QUEUE": fxtr("default_worker_celery_worker_queue"),
+            },
+        )
+        cnt = f"worker_v{v.replace('.', '_')}_container"
+        globals()[cnt] = container(
+            image="{" + f"{img}.id" + "}",
+            environment=fxtr("default_worker_env"),
+            network="{default_pytest_celery_network.name}",
+            volumes={"{default_worker_volume.name}": DEFAULT_WORKER_VOLUME},
+            wrapper_class=CeleryWorkerContainer,
+            timeout=DEFAULT_WORKER_CONTAINER_TIMEOUT,
+        )
+        worker_containers.append(cnt)
+    return worker_containers
+
+
+class TestClusterList:
+    @pytest.fixture(params=[generate_workers(versions_list)])
+    def celery_worker_cluster(self, request: pytest.FixtureRequest) -> CeleryWorkerCluster:
+        nodes: list[CeleryWorkerContainer] = [request.getfixturevalue(worker) for worker in request.param]
+        cluster = CeleryWorkerCluster(*nodes)
+        yield cluster
+        cluster.teardown()
+
+    def test_worker_cluster_with_fixed_list(self, celery_setup: CeleryTestSetup, subtests):
+        worker: CeleryTestWorker
+        for version, worker in zip(versions_list, celery_setup.worker_cluster):
+            with subtests.test(msg=f"Found worker {version} in cluster"):
+                assert f"{worker.hostname()} {version}" in worker.logs()
+
+
+class TestClusterRange:
+    @pytest.fixture(params=[generate_workers(versions_range)])
+    def celery_worker_cluster(self, request: pytest.FixtureRequest) -> CeleryWorkerCluster:
+        nodes: list[CeleryWorkerContainer] = [request.getfixturevalue(worker) for worker in request.param]
+        cluster = CeleryWorkerCluster(*nodes)
+        yield cluster
+        cluster.teardown()
+
+    def test_worker_cluster_with_versions_range(self, celery_setup: CeleryTestSetup, subtests):
+        worker: CeleryTestWorker
+        for version, worker in zip(versions_range, celery_setup.worker_cluster):
+            with subtests.test(msg=f"Found worker v{version} in cluster"):
+                assert f"{worker.hostname()} v{version}" in worker.logs()