Feature: New worker fixture "default_worker_command"

docs/getting-started/next-steps.rst

@@ -263,6 +263,7 @@ And build our container using the standard `pytest-docker-tools <https://pypi.or
         volumes={"{default_worker_volume.name}": defaults.DEFAULT_WORKER_VOLUME},
         wrapper_class=MyWorkerContainer,
         timeout=defaults.DEFAULT_WORKER_CONTAINER_TIMEOUT,
+        command=fxtr("default_worker_command"),
     )
 
 Pytest Integration

docs/userguide/examples/django.rst

@@ -144,6 +144,7 @@ and `container <https://github.com/Jc2k/pytest-docker-tools?tab=readme-ov-file#c
         },
         wrapper_class=DjangoWorkerContainer,
         timeout=defaults.DEFAULT_WORKER_CONTAINER_TIMEOUT,
+        command=fxtr("default_worker_command"),
     )
 
 In this case, we also mount the project directory to ``/src`` in the container, so that we can install the project

docs/userguide/examples/myworker.rst

@@ -161,6 +161,7 @@ These fixtures may be overridden if required.
         volumes={"{default_worker_volume.name}": defaults.DEFAULT_WORKER_VOLUME},
         wrapper_class=MyWorkerContainer,
         timeout=defaults.DEFAULT_WORKER_CONTAINER_TIMEOUT,
+        command=fxtr("default_worker_command"),
     )
 
 Lastly, we wrap the container in a fixture to allow it to be injected into the test environment

examples/django/tests/conftest.py

@@ -58,6 +58,7 @@ def worker_queue(cls) -> str:
     },
     wrapper_class=DjangoWorkerContainer,
     timeout=defaults.DEFAULT_WORKER_CONTAINER_TIMEOUT,
+    command=fxtr("default_worker_command"),
 )
 
 

examples/myworker/tests/myworker/myworker.py

@@ -50,6 +50,7 @@ def worker_queue(cls) -> str:
     volumes={"{default_worker_volume.name}": defaults.DEFAULT_WORKER_VOLUME},
     wrapper_class=MyWorkerContainer,
     timeout=defaults.DEFAULT_WORKER_CONTAINER_TIMEOUT,
+    command=fxtr("default_worker_command"),
 )
 
 

src/pytest_celery/api/container.py

@@ -71,9 +71,6 @@ def command(cls, *args: str) -> list[str]:
                 the first element, followed by the command-line arguments.
         """
 
-        # To be used with pytest_docker_tools.container using the command
-        # kwarg with the class method as value
-        # e.g. command=MyContainer.command()
         raise NotImplementedError("CeleryTestContainer.command")
 
     def teardown(self) -> None:

src/pytest_celery/vendors/worker/container.py

@@ -32,6 +32,22 @@ class CeleryWorkerContainer(CeleryTestContainer):
         dependencies of your project.
     """
 
+    @classmethod
+    def command(cls, *args: str) -> list[str]:
+        args = args or tuple()
+        return [
+            "celery",
+            "-A",
+            "app",
+            "worker",
+            f"--loglevel={cls.log_level()}",
+            "-n",
+            f"{cls.worker_name()}@%h",
+            "-Q",
+            f"{cls.worker_queue()}",
+            *args,
+        ]
+
     def _wait_port(self, port: str) -> int:
         # Not needed for worker container
         raise NotImplementedError

src/pytest_celery/vendors/worker/fixtures.py

@@ -91,6 +91,7 @@ def default_worker_container_session_cls() -> type[CeleryWorkerContainer]:
     volumes={"{default_worker_volume.name}": DEFAULT_WORKER_VOLUME},
     wrapper_class=CeleryWorkerContainer,
     timeout=DEFAULT_WORKER_CONTAINER_TIMEOUT,
+    command=fxtr("default_worker_command"),
 )
 
 celery_base_worker_image = build(
@@ -161,6 +162,19 @@ def default_worker_celery_worker_queue(default_worker_container_session_cls: typ
     return default_worker_container_session_cls.worker_queue()
 
 
+@pytest.fixture
+def default_worker_command(default_worker_container_cls: type[CeleryWorkerContainer]) -> list[str]:
+    """Command to run the container.
+
+    Args:
+        default_worker_container_cls (type[CeleryWorkerContainer]): See also: :ref:`vendor-class`.
+
+    Returns:
+        list[str]: Docker CMD instruction.
+    """
+    return default_worker_container_cls.command()
+
+
 @pytest.fixture
 def default_worker_env(
     default_worker_container_cls: type[CeleryWorkerContainer],

tests/integration/api/custom_setup/conftest.py

@@ -52,6 +52,7 @@ def worker_queue(cls) -> str:
     volumes={"{default_worker_volume.name}": DEFAULT_WORKER_VOLUME},
     wrapper_class=Celery4WorkerContainer,
     timeout=DEFAULT_WORKER_CONTAINER_TIMEOUT,
+    command=fxtr("default_worker_command"),
 )
 
 
@@ -110,6 +111,7 @@ def default_worker_container_session_cls() -> type[CeleryWorkerContainer]:
     volumes={"{default_worker_volume.name}": DEFAULT_WORKER_VOLUME},
     wrapper_class=Celery5WorkerContainer,
     timeout=DEFAULT_WORKER_CONTAINER_TIMEOUT,
+    command=fxtr("default_worker_command"),
 )
 
 

tests/integration/conftest.py

@@ -56,4 +56,5 @@ def default_worker_container_session_cls() -> type[CeleryWorkerContainer]:
     volumes={"{default_worker_volume.name}": DEFAULT_WORKER_VOLUME},
     wrapper_class=IntegrationWorkerContainer,
     timeout=DEFAULT_WORKER_CONTAINER_TIMEOUT,
+    command=fxtr("default_worker_command"),
 )

tests/integration/vendors/test_worker.py

@@ -50,6 +50,14 @@ def default_worker_app_module(self, request: pytest.FixtureRequest) -> ModuleTyp
         def test_replacing_app_module(self, container: CeleryWorkerContainer, default_worker_app_module: ModuleType):
             assert container.app_module() == default_worker_app_module
 
+    class test_replacing_command:
+        @pytest.fixture
+        def default_worker_command(self, default_worker_command: list[str]) -> list[str]:
+            return [*default_worker_command, "--pool", "solo"]
+
+        def test_replacing_command(self, container: CeleryWorkerContainer):
+            assert container.logs().count("solo") == 1
+
 
 class test_base_test_worker:
     def test_config(self, celery_setup_worker: CeleryTestWorker):

tests/smoke/conftest.py

@@ -48,6 +48,7 @@ def worker_queue(cls) -> str:
     volumes={"{default_worker_volume.name}": DEFAULT_WORKER_VOLUME},
     wrapper_class=CeleryLatestWorkerContainer,
     timeout=DEFAULT_WORKER_CONTAINER_TIMEOUT,
+    command=fxtr("default_worker_command"),
 )
 
 
@@ -102,6 +103,7 @@ def default_worker_container_session_cls() -> type[CeleryWorkerContainer]:
     volumes={"{default_worker_volume.name}": DEFAULT_WORKER_VOLUME},
     wrapper_class=SmokeWorkerContainer,
     timeout=DEFAULT_WORKER_CONTAINER_TIMEOUT,
+    command=fxtr("default_worker_command"),
 )
 
 

tests/smoke/test_worker.py

@@ -0,0 +1,24 @@
+from __future__ import annotations
+
+import pytest
+
+from pytest_celery import CeleryTestSetup
+
+
+@pytest.mark.parametrize(
+    "pool",
+    [
+        "solo",
+        "prefork",
+        "threads",
+    ],
+)
+class test_replacing_pool:
+    @pytest.fixture
+    def default_worker_command(self, default_worker_command: list[str], pool: str) -> list[str]:
+        return [*default_worker_command, "--pool", pool]
+
+    def test_pool_from_celery_banner(self, celery_setup: CeleryTestSetup, pool: str):
+        if pool == "threads":
+            pool = "thread"
+        celery_setup.worker.assert_log_exists(pool)