Add test ensuring ingestion works without db access (#3322)

Handling alert payloads should work without db access (but still
requires cached integrations information)

engine/apps/integrations/tests/test_views.py

@@ -2,11 +2,24 @@ from unittest.mock import call, patch
 
 import pytest
 from django.core.files.uploadedfile import SimpleUploadedFile
+from django.db import OperationalError
 from django.urls import reverse
+from pytest_django.plugin import _DatabaseBlocker
 from rest_framework import status
 from rest_framework.test import APIClient
 
 from apps.alerts.models import AlertReceiveChannel
+from apps.integrations.mixins import AlertChannelDefiningMixin
+
+
+class DatabaseBlocker(_DatabaseBlocker):
+    """Customize pytest_django db blocker to raise OperationalError exception."""
+
+    def _blocking_wrapper(*args, **kwargs):
+        __tracebackhide__ = True
+        __tracebackhide__  # Silence pyflakes
+        # mimic DB unavailable error
+        raise OperationalError("Database access disabled")
 
 
 @pytest.mark.django_db
@@ -184,3 +197,99 @@ def test_integration_universal_endpoint_not_allow_files(
     assert response.status_code == status.HTTP_400_BAD_REQUEST
 
     assert not mock_create_alert.apply_async.called
+
+
+@patch("apps.integrations.views.create_alert")
+@pytest.mark.parametrize(
+    "integration_type",
+    [
+        arc_type
+        for arc_type in AlertReceiveChannel.INTEGRATION_TYPES
+        if arc_type not in ["amazon_sns", "grafana", "alertmanager", "grafana_alerting", "maintenance"]
+    ],
+)
+@pytest.mark.django_db
+def test_integration_universal_endpoint_works_without_db(
+    mock_create_alert, make_organization_and_user, make_alert_receive_channel, integration_type
+):
+    organization, user = make_organization_and_user()
+    alert_receive_channel = make_alert_receive_channel(
+        organization=organization,
+        author=user,
+        integration=integration_type,
+    )
+
+    client = APIClient()
+    url = reverse(
+        "integrations:universal",
+        kwargs={"integration_type": integration_type, "alert_channel_key": alert_receive_channel.token},
+    )
+
+    # populate cache
+    AlertChannelDefiningMixin().update_alert_receive_channel_cache()
+
+    # disable DB access
+    with DatabaseBlocker().block():
+        data = {"foo": "bar"}
+        response = client.post(url, data, format="json")
+
+    assert response.status_code == status.HTTP_200_OK
+
+    mock_create_alert.apply_async.assert_called_once_with(
+        [],
+        {
+            "title": None,
+            "message": None,
+            "image_url": None,
+            "link_to_upstream_details": None,
+            "alert_receive_channel_pk": alert_receive_channel.pk,
+            "integration_unique_data": None,
+            "raw_request_data": data,
+        },
+    )
+
+
+@patch("apps.integrations.views.create_alertmanager_alerts")
+@pytest.mark.django_db
+def test_integration_grafana_endpoint_without_db_has_alerts(
+    mock_create_alertmanager_alerts, settings, make_organization_and_user, make_alert_receive_channel
+):
+    settings.DEBUG = False
+
+    integration_type = "grafana"
+    organization, user = make_organization_and_user()
+    alert_receive_channel = make_alert_receive_channel(
+        organization=organization,
+        author=user,
+        integration=integration_type,
+    )
+
+    client = APIClient()
+    url = reverse("integrations:grafana", kwargs={"alert_channel_key": alert_receive_channel.token})
+
+    data = {
+        "alerts": [
+            {
+                "foo": 123,
+            },
+            {
+                "foo": 456,
+            },
+        ]
+    }
+
+    # populate cache
+    AlertChannelDefiningMixin().update_alert_receive_channel_cache()
+
+    # disable DB access
+    with DatabaseBlocker().block():
+        response = client.post(url, data, format="json")
+
+    assert response.status_code == status.HTTP_200_OK
+
+    mock_create_alertmanager_alerts.apply_async.assert_has_calls(
+        [
+            call((alert_receive_channel.pk, data["alerts"][0])),
+            call((alert_receive_channel.pk, data["alerts"][1])),
+        ]
+    )

engine/apps/labels/tests/factories.py

@@ -10,7 +10,7 @@ from common.utils import UniqueFaker
 
 
 class LabelKeyFactory(factory.DjangoModelFactory):
-    id = UniqueFaker("sentence", nb_words=3)
+    id = UniqueFaker("pystr", max_chars=36)
     name = UniqueFaker("sentence", nb_words=3)
 
     class Meta:
@@ -18,7 +18,7 @@ class LabelKeyFactory(factory.DjangoModelFactory):
 
 
 class LabelValueFactory(factory.DjangoModelFactory):
-    id = UniqueFaker("sentence", nb_words=3)
+    id = UniqueFaker("pystr", max_chars=36)
     name = UniqueFaker("sentence", nb_words=3)
 
     class Meta: