fix: exclude infrastructure probes from API throttling

tests/apps/core/test_views.py

@@ -7,16 +7,28 @@ from unittest.mock import patch
 
 from apps.core import views as core_views
 from apps.core.views import HealthCheckView
-from django.urls import reverse
+from django.core.cache.backends.locmem import LocMemCache
+from django.urls import path, reverse
 from django.utils import timezone
+from drf_yasg import openapi
+from drf_yasg.generators import OpenAPISchemaGenerator
 from rest_framework import status
 from rest_framework.test import APIRequestFactory, APITestCase
+from rest_framework.throttling import AnonRateThrottle
 from rest_framework.views import APIView
 
 from tests.apps.user.factories import UserFactory
 from tests.utils.fixtures import fake
 
 
+class _RejectAllThrottle:
+    def allow_request(self, request, view):
+        return False
+
+    def wait(self):
+        return None
+
+
 class HealthCheckViewTest(APITestCase):
     """Tests for HealthCheckView"""
 
@@ -25,6 +37,12 @@ class HealthCheckViewTest(APITestCase):
         url = reverse("core:health")
         self.assertEqual(url, "/health/")
 
+    def test_detailed_health_check_keeps_global_throttling(self):
+        with patch.object(APIView, "throttle_classes", [_RejectAllThrottle]):
+            response = self.client.get(reverse("core:health"))
+
+        self.assertEqual(response.status_code, status.HTTP_429_TOO_MANY_REQUESTS)
+
     def test_health_check_success(self):
         """Test health check returns healthy status"""
         url = reverse("core:health")
@@ -230,6 +248,13 @@ class LivenessViewTest(APITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         self.assertEqual(response.data["status"], "alive")
 
+    def test_liveness_bypasses_global_throttling(self):
+        with patch.object(APIView, "throttle_classes", [_RejectAllThrottle]):
+            response = self.client.get(reverse("core:liveness"))
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(response.data["status"], "alive")
+
 
 class ReadinessViewTest(APITestCase):
     """Tests for ReadinessView"""
@@ -248,13 +273,6 @@ class ReadinessViewTest(APITestCase):
         self.assertEqual(response.data["status"], "ready")
 
     def test_readiness_bypasses_global_throttling(self):
-        class _RejectAllThrottle:
-            def allow_request(self, request, view):
-                return False
-
-            def wait(self):
-                return None
-
         with patch.object(APIView, "throttle_classes", [_RejectAllThrottle]):
             response = self.client.get(reverse("core:readiness"))
 
@@ -278,7 +296,8 @@ class ReadinessViewTest(APITestCase):
         try:
             core_views.connection = _BrokenConnection()
             url = reverse("core:readiness")
-            response = self.client.get(url)
+            with patch.object(APIView, "throttle_classes", [_RejectAllThrottle]):
+                response = self.client.get(url)
         finally:
             core_views.connection = original_connection
 
@@ -286,6 +305,65 @@ class ReadinessViewTest(APITestCase):
         self.assertEqual(response.data["status"], "not_ready")
 
 
+class HealthProbeThrottlingTest(APITestCase):
+    def test_openapi_keeps_rate_limit_only_on_detailed_health(self):
+        generator = OpenAPISchemaGenerator(
+            info=openapi.Info(title="Health probes", default_version="v1"),
+            patterns=[
+                path("health/", HealthCheckView.as_view()),
+                path("health/live/", core_views.LivenessView.as_view()),
+                path("health/ready/", core_views.ReadinessView.as_view()),
+            ],
+        )
+        schema = generator.get_schema(request=None, public=True)
+        responses = {
+            f"{schema.base_path.rstrip('/')}{url}": item["get"].responses
+            for url, item in schema.paths.items()
+        }
+
+        self.assertNotIn("429", responses["/health/live/"])
+        self.assertNotIn("429", responses["/health/ready/"])
+        self.assertIn("503", responses["/health/ready/"])
+        self.assertIn("429", responses["/health/"])
+
+    def test_probes_do_not_consume_or_obey_exhausted_anonymous_quota(self):
+        probe_cache = LocMemCache(self.id(), {})
+        probe_cache.clear()
+        self.addCleanup(probe_cache.clear)
+
+        class OnePerMinuteAnonThrottle(AnonRateThrottle):
+            rate = "1/min"
+            cache = probe_cache
+
+        with (
+            patch.object(APIView, "throttle_classes", [OnePerMinuteAnonThrottle]),
+            patch.object(
+                HealthCheckView, "_check_database", return_value={"status": "up"}
+            ),
+            patch.object(
+                HealthCheckView, "_check_redis", return_value={"status": "up"}
+            ),
+        ):
+            for quota_exhausted in (False, True):
+                for _ in range(3):
+                    for name, expected_status in (
+                        ("liveness", "alive"),
+                        ("readiness", "ready"),
+                    ):
+                        with self.subTest(probe=name, quota_exhausted=quota_exhausted):
+                            response = self.client.get(reverse(f"core:{name}"))
+                            self.assertEqual(response.status_code, status.HTTP_200_OK)
+                            self.assertEqual(response.data["status"], expected_status)
+
+                if not quota_exhausted:
+                    response = self.client.get(reverse("core:health"))
+                    self.assertEqual(response.status_code, status.HTTP_200_OK)
+                response = self.client.get(reverse("core:health"))
+                self.assertEqual(
+                    response.status_code, status.HTTP_429_TOO_MANY_REQUESTS
+                )
+
+
 class APIVersioningURLTest(APITestCase):
     """Tests for API versioning URL structure"""