Fix #631 -- Add task_default_queue fallback if task_queues is None

By default Celeries' `task_queues` is None. Even though, there is a
default queue set unless the configuration is altered.

* Drop Celery 3.0 support and drop the old settings fallback.
* Harden test suite by relying less on mocking.

Author: codingjoe

health_check/contrib/celery.py

@@ -3,6 +3,7 @@
 import dataclasses
 import datetime
 import typing
+from types import MappingProxyType
 
 import celery
 from celery.app import app_or_default
@@ -25,7 +26,9 @@ class Ping(HealthCheck):
 
     """
 
-    CORRECT_PING_RESPONSE: typing.ClassVar[dict[str, str]] = {"ok": "pong"}
+    CORRECT_PING_RESPONSE: typing.Final[dict[str, str]] = MappingProxyType(
+        {"ok": "pong"}
+    )
     app: celery.Celery = dataclasses.field(default_factory=app_or_default)
     timeout: datetime.timedelta = dataclasses.field(
         default=datetime.timedelta(seconds=1), repr=False
@@ -59,11 +62,11 @@ def active_workers(self, ping_result):
             yield worker
 
     def check_active_queues(self, *active_workers):
-        defined_queues = {
-            queue.name
-            for queue in getattr(self.app.conf, "task_queues", None)
-            or getattr(self.app.conf, "CELERY_QUEUES", None)
-        }
+        try:
+            defined_queues = {queue.name for queue in self.app.conf.task_queues}
+        except TypeError:
+            # conf.task_queues may be None
+            defined_queues = {self.app.conf.task_default_queue}
         active_queues = {
             queue.get("name")
             for queues in self.app.control.inspect(active_workers)

tests/contrib/test_celery.py

@@ -1,48 +1,30 @@
 """Tests for Celery health check."""
 
+import datetime
+import logging
 from unittest import mock
 
 import pytest
 
 pytest.importorskip("celery")
 
+from kombu import Queue
+
 from health_check.contrib.celery import Ping as CeleryPingHealthCheck
 from health_check.exceptions import ServiceUnavailable
+from tests.testapp.celery import app as celery_app
+
+logger = logging.getLogger(__name__)
 
 
 class TestCelery:
     """Test Celery ping health check."""
 
-    @pytest.mark.asyncio
-    async def test_check_status__success(self):
-        """Report healthy when workers respond correctly."""
-        app = mock.MagicMock()
-        app.ping.return_value = [{"celery@worker1": {"ok": "pong"}}]
-        check = CeleryPingHealthCheck()
-        check.app = app
-
-        result = await check.get_result()
-        assert result.error is None
-
-    @pytest.mark.asyncio
-    async def test_check_status__no_workers(self):
-        """Raise ServiceUnavailable when no workers respond."""
-        mock_app = mock.MagicMock()
-        mock_app.control.ping.return_value = {}
-        check = CeleryPingHealthCheck()
-        check.app = mock_app
-
-        result = await check.get_result()
-        assert result.error is not None
-        assert isinstance(result.error, ServiceUnavailable)
-        assert "unavailable" in str(result.error).lower()
-
     @pytest.mark.asyncio
     async def test_check_status__unexpected_response(self):
         """Raise ServiceUnavailable when worker response is incorrect."""
-        mock_result = {"celery@worker1": {"bad": "response"}}
         mock_app = mock.MagicMock()
-        mock_app.control.ping.return_value = [mock_result]
+        mock_app.control.ping.return_value = [{"celery@worker1": {"bad": "response"}}]
         check = CeleryPingHealthCheck()
         check.app = mock_app
 
@@ -87,24 +69,92 @@ async def test_check_status__unknown_error(self):
             assert result.error is not None
             assert isinstance(result.error, ServiceUnavailable)
 
+    @pytest.mark.asyncio
+    async def test_check_status__success(self):
+        """Report healthy when using real Celery app with configured queues."""
+        default_queue = Queue("default", routing_key="default")
+        celery_app.conf.task_queues = [default_queue]
+        celery_app.conf.task_default_queue = "default"
+
+        check = CeleryPingHealthCheck(app=celery_app)
+
+        # Mock the control methods directly on the app before get_result runs
+        mock_ping = mock.MagicMock(return_value=[{"celery@worker1": {"ok": "pong"}}])
+        mock_inspect_obj = mock.MagicMock()
+        mock_inspect_obj.active_queues.return_value = {
+            "celery@worker1": [{"name": "default"}]
+        }
+        mock_inspect = mock.MagicMock(return_value=mock_inspect_obj)
+
+        original_ping = check.app.control.ping
+        original_inspect = check.app.control.inspect
+
+        try:
+            check.app.control.ping = mock_ping
+            check.app.control.inspect = mock_inspect
+
+            result = await check.get_result()
+            assert result.error is None
+        finally:
+            check.app.control.ping = original_ping
+            check.app.control.inspect = original_inspect
+
+    @pytest.mark.asyncio
+    async def test_check_status__no_workers(self):
+        """Raise ServiceUnavailable when real app receives no worker response."""
+        default_queue = Queue("default", routing_key="default")
+        celery_app.conf.task_queues = [default_queue]
+        celery_app.conf.task_default_queue = "default"
+
+        check = CeleryPingHealthCheck(
+            app=celery_app,
+            timeout=datetime.timedelta(milliseconds=100),
+        )
+        result = await check.get_result()
+
+        assert result.error is not None
+        assert isinstance(result.error, ServiceUnavailable)
+        assert "unavailable" in str(result.error).lower()
+
     @pytest.mark.asyncio
     async def test_check_status__missing_queue_worker(self):
-        """Raise ServiceUnavailable when a defined queue has no active workers."""
-        mock_result = {"celery@worker1": {"ok": "pong"}}
+        """Verify queue validation with real Celery app configuration."""
+        multiple_queues = [
+            Queue("default", routing_key="default"),
+            Queue("integration_queue", routing_key="integration_queue"),
+        ]
+        celery_app.conf.task_queues = multiple_queues
+        celery_app.conf.task_default_queue = "default"
+
+        ping_response = [{"celery@worker1": {"ok": "pong"}}]
+        inspect_response = {"celery@worker1": [{"name": "default"}]}
+
+        check = CeleryPingHealthCheck(app=celery_app)
+        check.app.control.ping = lambda **kwargs: ping_response
+        check.app.control.inspect = lambda *args: mock.MagicMock(
+            active_queues=lambda: inspect_response
+        )
+
+        result = await check.get_result()
+        assert result.error is not None
+        assert isinstance(result.error, ServiceUnavailable)
+        assert "integration_queue" in str(result.error)
+
+    @pytest.mark.asyncio
+    async def test_check_status__raise_type_error__default_queue(self):
+        """Verify that default queue is checked when task_queues is None."""
         mock_app = mock.MagicMock()
-        mock_app.control.ping.return_value = [mock_result]
-        mock_queue = mock.MagicMock()
-        mock_queue.name = "missing_queue"
-        mock_app.conf.task_queues = [mock_queue]
+        mock_app.conf.task_queues = None
+        mock_app.conf.task_default_queue = "default"
+        mock_app.control.ping.return_value = [{"celery@worker1": {"ok": "pong"}}]
         mock_inspect = mock.MagicMock()
         mock_inspect.active_queues.return_value = {
-            "celery@worker1": [{"name": "celery"}]
+            "celery@worker1": [{"name": "default"}]
         }
         mock_app.control.inspect.return_value = mock_inspect
+
         check = CeleryPingHealthCheck()
         check.app = mock_app
 
         result = await check.get_result()
-        assert result.error is not None
-        assert isinstance(result.error, ServiceUnavailable)
-        assert "missing_queue" in str(result.error)
+        assert result.error is None

tests/testapp/settings.py

@@ -52,14 +52,12 @@
 
 USE_TZ = True
 
-CELERY_QUEUES = []
-
 try:
     from kombu import Queue
 except ImportError:
     pass
 else:
-    CELERY_QUEUES += [
+    CELERY_QUEUES = [
         Queue("default"),
         Queue("queue2"),
     ]