Added: default_worker_ports

src/pytest_celery/__init__.py

@@ -103,6 +103,7 @@
     from pytest_celery.vendors.worker.fixtures import default_worker_container_session_cls
     from pytest_celery.vendors.worker.fixtures import default_worker_env
     from pytest_celery.vendors.worker.fixtures import default_worker_initial_content
+    from pytest_celery.vendors.worker.fixtures import default_worker_ports
     from pytest_celery.vendors.worker.fixtures import default_worker_signals
     from pytest_celery.vendors.worker.fixtures import default_worker_tasks
     from pytest_celery.vendors.worker.fixtures import default_worker_utils_module

src/pytest_celery/vendors/worker/Dockerfile

@@ -32,6 +32,8 @@ ENV WORKER_QUEUE=$CELERY_WORKER_QUEUE
 ENV PYTHONUNBUFFERED=1
 ENV PYTHONDONTWRITEBYTECODE=1
 
+EXPOSE 5678
+
 # Install Python dependencies
 RUN pip install --no-cache-dir --upgrade \
     pip \

src/pytest_celery/vendors/worker/container.py

@@ -14,6 +14,7 @@
 from pytest_celery.vendors.worker.defaults import DEFAULT_WORKER_ENV
 from pytest_celery.vendors.worker.defaults import DEFAULT_WORKER_LOG_LEVEL
 from pytest_celery.vendors.worker.defaults import DEFAULT_WORKER_NAME
+from pytest_celery.vendors.worker.defaults import DEFAULT_WORKER_PORTS
 from pytest_celery.vendors.worker.defaults import DEFAULT_WORKER_QUEUE
 from pytest_celery.vendors.worker.defaults import DEFAULT_WORKER_VERSION
 from pytest_celery.vendors.worker.volume import WorkerInitialContent
@@ -197,3 +198,8 @@ def initial_content(
             content.set_config_from_object(worker_app)
 
         return content.generate()
+
+    @classmethod
+    def ports(cls) -> dict | None:
+        """Ports to expose from the worker container."""
+        return DEFAULT_WORKER_PORTS

src/pytest_celery/vendors/worker/defaults.py

@@ -21,6 +21,9 @@
     "PYTHONDONTWRITEBYTECODE": "1",
     "PYTHONPATH": "/app",
 }
+WORKER_DEBUGPY_PORTS = {
+    "5678/tcp": "5678",
+}
 WORKER_VOLUME = {
     "bind": "/app",
     "mode": "rw",
@@ -30,6 +33,7 @@
 DEFAULT_WORKER_LOG_LEVEL = WORKER_LOG_LEVEL
 DEFAULT_WORKER_NAME = WORKER_NAME
 DEFAULT_WORKER_ENV = WORKER_ENV
+DEFAULT_WORKER_PORTS = None
 DEFAULT_WORKER_QUEUE = WORKER_QUEUE
 DEFAULT_WORKER_CONTAINER_TIMEOUT = 60
 DEFAULT_WORKER_VOLUME = WORKER_VOLUME

src/pytest_celery/vendors/worker/fixtures.py

@@ -86,6 +86,7 @@ def default_worker_container_session_cls() -> type[CeleryWorkerContainer]:
 
 default_worker_container = container(
     image="{celery_base_worker_image.id}",
+    ports=fxtr("default_worker_ports"),
     environment=fxtr("default_worker_env"),
     network="{default_pytest_celery_network.name}",
     volumes={"{default_worker_volume.name}": DEFAULT_WORKER_VOLUME},
@@ -237,6 +238,19 @@ def default_worker_initial_content(
     )
 
 
+@pytest.fixture
+def default_worker_ports(default_worker_container_cls: type[CeleryWorkerContainer]) -> dict | None:
+    """Port bindings for this vendor.
+
+    Args:
+        default_worker_container_cls (type[CeleryWorkerContainer]): See also: :ref:`vendor-class`.
+
+    Returns:
+        dict: Port bindings.
+    """
+    return default_worker_container_cls.ports()
+
+
 @pytest.fixture
 def default_worker_app_module(default_worker_container_cls: type[CeleryWorkerContainer]) -> ModuleType:
     """App module for this worker.

tests/integration/conftest.py

@@ -51,6 +51,7 @@ def default_worker_container_session_cls() -> type[CeleryWorkerContainer]:
 
 default_worker_container = container(
     image="{integration_tests_worker_image.id}",
+    ports=fxtr("default_worker_ports"),
     environment=fxtr("default_worker_env"),
     network="{default_pytest_celery_network.name}",
     volumes={"{default_worker_volume.name}": DEFAULT_WORKER_VOLUME},

tests/smoke/conftest.py

@@ -43,6 +43,7 @@ def worker_queue(cls) -> str:
 
 celery_latest_worker_container = container(
     image="{celery_latest_worker_image.id}",
+    ports=fxtr("default_worker_ports"),
     environment=fxtr("default_worker_env"),
     network="{default_pytest_celery_network.name}",
     volumes={"{default_worker_volume.name}": DEFAULT_WORKER_VOLUME},
@@ -96,7 +97,8 @@ def default_worker_container_session_cls() -> type[CeleryWorkerContainer]:
 
 
 default_worker_container = container(
-    image="{smoke_tests_worker_image.id}",  # TODO: Use fixture to avoid defining default_worker_container again
+    image="{smoke_tests_worker_image.id}",
+    ports=fxtr("default_worker_ports"),
     environment=fxtr("default_worker_env"),
     network="{default_pytest_celery_network.name}",
     volumes={"{default_worker_volume.name}": DEFAULT_WORKER_VOLUME},