128 lines
3.9 KiB
Python
128 lines
3.9 KiB
Python
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
|