Merge branch 'main' into issues/659/alias

Author: codingjoe

health_check/checks.py

@@ -11,13 +11,15 @@
 from django import db
 from django.conf import settings
 from django.core.cache import CacheKeyWarning, caches
+from django.core.cache.backends.base import InvalidCacheBackendError
 from django.core.files.base import ContentFile
+from django.core.files.storage import InvalidStorageError, storages
 from django.core.files.storage import Storage as DjangoStorage
-from django.core.files.storage import storages
 from django.core.mail import get_connection
 from django.core.mail.backends.base import BaseEmailBackend
 from django.db import connections
 from django.db.models import Expression
+from django.utils.connection import ConnectionDoesNotExist
 
 from health_check.base import HealthCheck
 from health_check.exceptions import (
@@ -62,7 +64,10 @@ class Cache(HealthCheck):
     )
 
     async def run(self):
-        cache = caches[self.alias]
+        try:
+            cache = caches[self.alias]
+        except InvalidCacheBackendError as e:
+            raise ServiceUnavailable("Cache alias does not exist") from e
         # Use an isolated key per probe run to avoid cross-process write races.
         cache_key = f"{self.key_prefix}:{uuid.uuid4().hex}"
         cache_value = f"itworks-{datetime.datetime.now().timestamp()}"
@@ -109,7 +114,10 @@ class Database(HealthCheck):
     alias: str = "default"
 
     def run(self):
-        connection = connections[self.alias]
+        try:
+            connection = connections[self.alias]
+        except ConnectionDoesNotExist as e:
+            raise ServiceUnavailable("Database alias does not exist") from e
         result = None
         try:
             compiler = connection.ops.compiler("SQLCompiler")(
@@ -237,7 +245,10 @@ class Storage(HealthCheck):
 
     @property
     def storage(self) -> DjangoStorage:
-        return storages[self.alias]
+        try:
+            return storages[self.alias]
+        except InvalidStorageError as e:
+            raise ServiceUnavailable("Storage alias does not exist") from e
 
     def get_file_name(self):
         return f"health_check_storage_test/test-{uuid.uuid4()}.txt"

health_check/views.py

@@ -249,7 +249,7 @@ def _render_feed(self, feed_class):
         """Generate RSS or Atom feed with health check results."""
         feed = feed_class(
             title="Health Check Status",
-            link=self.request.build_absolute_uri(),
+            link=self.request.build_absolute_uri(self.request.path),
             description="Current status of system health checks",
             feed_url=self.request.build_absolute_uri(),
         )
@@ -262,8 +262,8 @@ def _render_feed(self, feed_class):
             )
             feed.add_item(
                 title=repr(result.check),
-                link=self.request.build_absolute_uri(),
-                description=f"{result.check!r}\nResponse time: {result.time_taken:.3f}s",
+                link=self.request.build_absolute_uri(self.request.path),
+                description=str(result.error) if result.error else "OK",
                 pubdate=published_at,
                 updateddate=published_at,
                 author_name=self.feed_author,

tests/test_checks.py

@@ -19,6 +19,16 @@
 class TestCache:
     """Test the Cache health check."""
 
+    @pytest.mark.asyncio
+    async def test_run_check__invalid_alias(self):
+        """Raise ServiceUnavailable when cache alias does not exist."""
+        check = Cache(alias="nonexistent-alias")
+        result = await check.get_result()
+        assert isinstance(result.error, ServiceUnavailable), (
+            "Expected ServiceUnavailable for a non-existent cache alias"
+        )
+        assert "Cache alias does not exist" in str(result.error)
+
     @pytest.mark.asyncio
     async def test_run_check__cache_working(self):
         """Cache backend successfully sets and retrieves values."""
@@ -112,6 +122,17 @@ async def _aget(_key):
 class TestDatabase:
     """Test the Database health check."""
 
+    @pytest.mark.django_db
+    @pytest.mark.asyncio
+    async def test_run_check__invalid_alias(self):
+        """Raise ServiceUnavailable when database alias does not exist."""
+        check = Database(alias="nonexistent-alias")
+        result = await check.get_result()
+        assert isinstance(result.error, ServiceUnavailable), (
+            "Expected ServiceUnavailable for a non-existent database alias"
+        )
+        assert "Database alias does not exist" in str(result.error)
+
     @pytest.mark.django_db
     @pytest.mark.asyncio
     async def test_run_check__database_available(self):
@@ -146,6 +167,16 @@ async def test_run_check__locmem_backend(self):
 class TestStorage:
     """Test the Storage health check."""
 
+    @pytest.mark.asyncio
+    async def test_run_check__invalid_alias(self):
+        """Raise ServiceUnavailable when storage alias does not exist."""
+        check = Storage(alias="nonexistent-alias")
+        result = await check.get_result()
+        assert isinstance(result.error, ServiceUnavailable), (
+            "Expected ServiceUnavailable for a non-existent storage alias"
+        )
+        assert "Storage alias does not exist" in str(result.error)
+
     @pytest.mark.asyncio
     async def test_run_check__default_storage(self):
         """Storage check completes without exceptions."""

tests/test_views.py

@@ -566,6 +566,91 @@ async def run(self):
             "Failed check should use current timestamp in mixed feed"
         )
 
+    @pytest.mark.asyncio
+    async def test_get__rss_feed_description_includes_status(self, health_check_view):
+        """RSS feed item description includes status message."""
+        feedparser = pytest.importorskip("feedparser")
+
+        class FailingBackend(HealthCheck):
+            async def run(self):
+                raise HealthCheckException("Something went wrong")
+
+        response = await health_check_view([FailingBackend], format_param="rss")
+        feed = feedparser.parse(response.content.decode("utf-8"))
+        assert len(feed.entries) == 1
+        entry = feed.entries[0]
+        assert "Something went wrong" in entry.summary, (
+            "Feed description should include the actual error message"
+        )
+
+    @pytest.mark.asyncio
+    async def test_get__rss_feed_description_healthy_shows_ok(self, health_check_view):
+        """RSS feed item description shows OK for healthy checks."""
+        feedparser = pytest.importorskip("feedparser")
+
+        class SuccessBackend(HealthCheck):
+            async def run(self):
+                pass
+
+        response = await health_check_view([SuccessBackend], format_param="rss")
+        feed = feedparser.parse(response.content.decode("utf-8"))
+        assert len(feed.entries) == 1
+        entry = feed.entries[0]
+        assert "OK" in entry.summary, (
+            "Feed description should show OK for healthy checks"
+        )
+
+    @pytest.mark.asyncio
+    async def test_get__rss_feed_link_excludes_format_param(self, health_check_view):
+        """RSS feed item link does not include the format query parameter."""
+        feedparser = pytest.importorskip("feedparser")
+
+        class SuccessBackend(HealthCheck):
+            async def run(self):
+                pass
+
+        response = await health_check_view([SuccessBackend], format_param="rss")
+        feed = feedparser.parse(response.content.decode("utf-8"))
+        assert len(feed.entries) == 1
+        entry = feed.entries[0]
+        assert "format=rss" not in entry.link, (
+            "Feed item link should not include the format query parameter"
+        )
+
+    @pytest.mark.asyncio
+    async def test_get__atom_feed_description_includes_status(self, health_check_view):
+        """Atom feed item description includes status message."""
+        feedparser = pytest.importorskip("feedparser")
+
+        class FailingBackend(HealthCheck):
+            async def run(self):
+                raise HealthCheckException("Database unreachable")
+
+        response = await health_check_view([FailingBackend], format_param="atom")
+        feed = feedparser.parse(response.content.decode("utf-8"))
+        assert len(feed.entries) == 1
+        entry = feed.entries[0]
+        assert "Database unreachable" in entry.summary, (
+            "Feed description should include the actual error message"
+        )
+
+    @pytest.mark.asyncio
+    async def test_get__atom_feed_link_excludes_format_param(self, health_check_view):
+        """Atom feed item link does not include the format query parameter."""
+        feedparser = pytest.importorskip("feedparser")
+
+        class SuccessBackend(HealthCheck):
+            async def run(self):
+                pass
+
+        response = await health_check_view([SuccessBackend], format_param="atom")
+        feed = feedparser.parse(response.content.decode("utf-8"))
+        assert len(feed.entries) == 1
+        entry = feed.entries[0]
+        assert "format=atom" not in entry.link, (
+            "Feed item link should not include the format query parameter"
+        )
+
     @pytest.mark.asyncio
     async def test_get_plugins__with_string_import(self):
         """Import check from string path."""