Bugfix: Disabled clusters raised exception on access from setup instead of returning None (#196)

Author: Nusnus

src/pytest_celery/api/setup.py

@@ -64,34 +64,34 @@ def app(self) -> Celery:
         return self._app
 
     @property
-    def backend_cluster(self) -> CeleryBackendCluster:
+    def backend_cluster(self) -> CeleryBackendCluster | None:
         """The backend cluster of this setup."""
         return self._backend_cluster
 
     @property
-    def backend(self) -> CeleryTestBackend:
+    def backend(self) -> CeleryTestBackend | None:
         """The first backend node of the backend cluster."""
-        return self._backend_cluster[0]  # type: ignore
+        return self._backend_cluster[0] if self._backend_cluster else None  # type: ignore
 
     @property
-    def broker_cluster(self) -> CeleryBrokerCluster:
+    def broker_cluster(self) -> CeleryBrokerCluster | None:
         """The broker cluster of this setup."""
         return self._broker_cluster
 
     @property
-    def broker(self) -> CeleryTestBroker:
+    def broker(self) -> CeleryTestBroker | None:
         """The first broker node of the broker cluster."""
-        return self._broker_cluster[0]  # type: ignore
+        return self._broker_cluster[0] if self._broker_cluster else None  # type: ignore
 
     @property
-    def worker_cluster(self) -> CeleryWorkerCluster:
+    def worker_cluster(self) -> CeleryWorkerCluster | None:
         """The worker cluster of this setup."""
         return self._worker_cluster
 
     @property
-    def worker(self) -> CeleryTestWorker:
+    def worker(self) -> CeleryTestWorker | None:
         """The first worker node of the worker cluster."""
-        return self._worker_cluster[0]  # type: ignore
+        return self._worker_cluster[0] if self._worker_cluster else None  # type: ignore
 
     @classmethod
     def name(cls) -> str:

tests/integration/api/test_setup.py

@@ -4,8 +4,11 @@
 from celery import Celery
 
 from pytest_celery import RESULT_TIMEOUT
+from pytest_celery import CeleryBackendCluster
+from pytest_celery import CeleryBrokerCluster
 from pytest_celery import CeleryTestSetup
 from pytest_celery import CeleryTestWorker
+from pytest_celery import CeleryWorkerCluster
 from tests.tasks import identity
 
 
@@ -59,3 +62,34 @@ def test_apply_async(self, celery_setup: CeleryTestSetup):
             queue = worker.worker_queue
             r = identity.s("test_ready").apply_async(queue=queue)
             assert r.get(timeout=RESULT_TIMEOUT) == "test_ready"
+
+    class test_disabling_backend_cluster:
+        @pytest.fixture
+        def celery_backend_cluster(self) -> CeleryBackendCluster:
+            return None
+
+        def test_disabling_backend_cluster(self, celery_setup: CeleryTestSetup):
+            assert celery_setup.backend_cluster is None
+            assert celery_setup.backend is None
+
+    class test_disabling_broker_cluster:
+        @pytest.fixture
+        def celery_broker_cluster(self) -> CeleryBrokerCluster:
+            return None
+
+        @pytest.fixture
+        def celery_worker_cluster(self) -> CeleryWorkerCluster:
+            return None
+
+        def test_disabling_broker_cluster(self, celery_setup: CeleryTestSetup):
+            assert celery_setup.broker_cluster is None
+            assert celery_setup.broker is None
+
+    class test_disabling_worker_cluster:
+        @pytest.fixture
+        def celery_worker_cluster(self) -> CeleryWorkerCluster:
+            return None
+
+        def test_disabling_worker_cluster(self, celery_setup: CeleryTestSetup):
+            assert celery_setup.worker_cluster is None
+            assert celery_setup.worker is None

tests/unit/api/test_setup.py

@@ -3,10 +3,13 @@
 import pytest
 from celery import Celery
 
+from pytest_celery import CeleryBackendCluster
+from pytest_celery import CeleryBrokerCluster
 from pytest_celery import CeleryTestBackend
 from pytest_celery import CeleryTestBroker
 from pytest_celery import CeleryTestSetup
 from pytest_celery import CeleryTestWorker
+from pytest_celery import CeleryWorkerCluster
 
 
 class test_celery_test_setup_unit:
@@ -67,3 +70,30 @@ def test_ready(self, celery_setup: CeleryTestSetup, confirmation: dict):
             **confirmation,
             control=False,  # Not supported in unit tests
         )
+
+    class test_disabling_backend_cluster:
+        @pytest.fixture
+        def celery_backend_cluster(self) -> CeleryBackendCluster:
+            return None
+
+        def test_disabling_backend_cluster(self, celery_setup: CeleryTestSetup):
+            assert celery_setup.backend_cluster is None
+            assert celery_setup.backend is None
+
+    class test_disabling_broker_cluster:
+        @pytest.fixture
+        def celery_broker_cluster(self) -> CeleryBrokerCluster:
+            return None
+
+        def test_disabling_broker_cluster(self, celery_setup: CeleryTestSetup):
+            assert celery_setup.broker_cluster is None
+            assert celery_setup.broker is None
+
+    class test_disabling_worker_cluster:
+        @pytest.fixture
+        def celery_worker_cluster(self) -> CeleryWorkerCluster:
+            return None
+
+        def test_disabling_worker_cluster(self, celery_setup: CeleryTestSetup):
+            assert celery_setup.worker_cluster is None
+            assert celery_setup.worker is None