cache queries and revalidate with incremental static regeneration

dnscms/tests/test_revalidation.py

@@ -0,0 +1,127 @@
+import logging
+import urllib.error
+
+import pytest
+
+from dnscms import revalidation
+from events.models import EventCategory
+from tests.conftest import GenericPageFactory
+
+WEBHOOK_URL = "https://frontend.example.com/api/revalidate"
+
+
+class InlineThread:
+    def __init__(self, target=None, args=(), kwargs=None, daemon=None):
+        self._target = target
+        self._args = args
+        self._kwargs = kwargs or {}
+
+    def start(self):
+        self._target(*self._args, **self._kwargs)
+
+
+@pytest.fixture
+def webhook_calls(monkeypatch, settings):
+    settings.REVALIDATE_WEBHOOK_SECRET = "s3cret"
+    settings.FRONTEND_BASE_URL = "https://frontend.example.com"
+    calls = []
+    monkeypatch.setattr(revalidation.threading, "Thread", InlineThread)
+    monkeypatch.setattr(
+        revalidation, "_send_webhook", lambda url, secret, reason: calls.append((url, secret))
+    )
+    return calls
+
+
+def test_page_publish_triggers_webhook(
+    home_page, webhook_calls, django_capture_on_commit_callbacks
+):
+    page = GenericPageFactory(parent=home_page)
+    with django_capture_on_commit_callbacks(execute=True):
+        page.save_revision().publish()
+
+    assert webhook_calls == [(WEBHOOK_URL, "s3cret")]
+
+
+def test_page_unpublish_triggers_webhook(
+    home_page, webhook_calls, django_capture_on_commit_callbacks
+):
+    page = GenericPageFactory(parent=home_page)
+    page.save_revision().publish()
+    webhook_calls.clear()
+
+    with django_capture_on_commit_callbacks(execute=True):
+        page.unpublish()
+
+    assert webhook_calls == [(WEBHOOK_URL, "s3cret")]
+
+
+def test_page_delete_triggers_webhook(
+    home_page, webhook_calls, django_capture_on_commit_callbacks
+):
+    # live=False isolates the post_delete path: deleting a live page would
+    # also fire page_unpublished (a harmless duplicate purge in production)
+    page = GenericPageFactory(parent=home_page, live=False)
+    webhook_calls.clear()
+
+    with django_capture_on_commit_callbacks(execute=True):
+        page.delete()
+
+    assert webhook_calls == [(WEBHOOK_URL, "s3cret")]
+
+
+@pytest.mark.django_db
+def test_snippet_save_triggers_webhook(webhook_calls, django_capture_on_commit_callbacks):
+    with django_capture_on_commit_callbacks(execute=True):
+        EventCategory.objects.create(name="Konsert", slug="konsert")
+
+    assert webhook_calls == [(WEBHOOK_URL, "s3cret")]
+
+
+def test_unset_secret_skips_and_warns_once(settings, monkeypatch, caplog):
+    settings.REVALIDATE_WEBHOOK_SECRET = ""
+    monkeypatch.setattr(revalidation, "_warned_unconfigured", False)
+
+    with caplog.at_level(logging.WARNING, logger="dnscms.revalidation"):
+        revalidation.trigger_frontend_revalidation("test")
+        revalidation.trigger_frontend_revalidation("test")
+
+    warnings = [r for r in caplog.records if r.levelno == logging.WARNING]
+    assert len(warnings) == 1
+
+
+def test_send_webhook_failure_is_swallowed(monkeypatch, caplog):
+    def raise_urlerror(request, timeout):
+        raise urllib.error.URLError("connection refused")
+
+    monkeypatch.setattr(revalidation.urllib.request, "urlopen", raise_urlerror)
+
+    with caplog.at_level(logging.ERROR, logger="dnscms.revalidation"):
+        revalidation._send_webhook(WEBHOOK_URL, "s3cret", "test")
+
+    assert any("frontend revalidation failed" in r.message for r in caplog.records)
+
+
+def test_send_webhook_request_shape(monkeypatch):
+    seen = {}
+
+    class FakeResponse:
+        status = 200
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *exc):
+            return False
+
+    def fake_urlopen(request, timeout):
+        seen["request"] = request
+        seen["timeout"] = timeout
+        return FakeResponse()
+
+    monkeypatch.setattr(revalidation.urllib.request, "urlopen", fake_urlopen)
+    revalidation._send_webhook(WEBHOOK_URL, "s3cret", "test")
+
+    request = seen["request"]
+    assert request.get_method() == "POST"
+    assert request.get_header("X-revalidate-secret") == "s3cret"
+    assert seen["timeout"] == revalidation.WEBHOOK_TIMEOUT_SECONDS