New Example: myutils

.github/workflows/examples.yml

@@ -159,3 +159,36 @@ jobs:
         run: |
           export DJANGO_SETTINGS_MODULE=proj.settings
           pytest -vv tests -n auto
+
+  myutils:
+    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/myutils
+        run: |
+          python -m pip install --upgrade pip
+          pip install -r requirements.txt
+
+      - name: Run tests
+        working-directory: examples/myutils
+        timeout-minutes: 10
+        run: |
+          pytest -vv tests -n auto

examples/myutils/pytest.ini

@@ -0,0 +1,5 @@
+[pytest]
+log_cli = true
+log_cli_level = INFO
+log_cli_format = %(asctime)s [%(levelname)8s] %(message)s (%(filename)s:%(lineno)s)
+log_cli_date_format = %Y-%m-%d %H:%M:%S

examples/myutils/requirements.txt

@@ -0,0 +1,3 @@
+pytest>=7.4.4
+pytest-celery[celery]@git+https://github.com/celery/pytest-celery.git
+pytest-xdist>=3.5.0

examples/myutils/tests/conftest.py

@@ -0,0 +1,29 @@
+from types import ModuleType
+
+import pytest
+
+from pytest_celery import CeleryTestWorker
+
+
+class MyWorker(CeleryTestWorker):
+
+    def myfunc(self) -> bool:
+        exit_code, output = self.container.exec_run(
+            'python -c "from utils import myfunc; print(myfunc())"',
+        )
+        if exit_code != 0:
+            raise RuntimeError(f"Error: {output}")
+        output = output.decode("utf-8")
+        return output.strip()
+
+
+@pytest.fixture
+def default_worker_cls() -> type[CeleryTestWorker]:
+    return MyWorker
+
+
+@pytest.fixture
+def default_worker_utils_module() -> ModuleType:
+    from tests import myutils
+
+    return myutils

examples/myutils/tests/myutils.py

@@ -0,0 +1,5 @@
+from pytest_celery.vendors.worker.content.utils import get_running_processes_info  # noqa
+
+
+def myfunc():
+    return "foo"

examples/myutils/tests/test_myutils.py

@@ -0,0 +1,13 @@
+from pytest_celery import CeleryTestSetup
+from tests.conftest import MyWorker
+
+
+def test_myfunc_in_worker(celery_worker: MyWorker):
+    assert celery_worker.myfunc() == "foo"
+    assert celery_worker.get_running_processes_info()
+
+
+def test_myfunc_in_setup_worker(celery_setup: CeleryTestSetup):
+    celery_worker: MyWorker = celery_setup.worker
+    assert celery_worker.myfunc() == "foo"
+    assert celery_worker.get_running_processes_info()