From e4a8194e6de25e8a48e2c61ce7f89a8159fd99d9 Mon Sep 17 00:00:00 2001
From: Dennis Fink
Date: Sun, 9 Aug 2026 14:15:29 +0200
Subject: Implement received Webmention handling

Add the Flask application setup, database models and migrations,
authentication, and configuration for development and testing.

Implement asynchronous Webmention verification with Huey, including HTML
and plain-text source validation, retries, status tracking, and size
limits.

Add status, login, and paginated received-Webmention views together with
comprehensive tests for forms, views, and receiver tasks.
---
 tests/tasks/test_receiver.py | 886 +++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 886 insertions(+)
 create mode 100644 tests/tasks/test_receiver.py

(limited to 'tests/tasks/test_receiver.py')

diff --git a/tests/tasks/test_receiver.py b/tests/tasks/test_receiver.py
new file mode 100644
index 0000000..4545121
--- /dev/null
+++ b/tests/tasks/test_receiver.py
@@ -0,0 +1,886 @@
+import logging
+import uuid
+from collections.abc import Callable
+from types import ModuleType
+
+import httpx
+import pytest
+from flask import Flask
+
+from webmentions_ssg import DATABASE as db
+from webmentions_ssg.models import ReceivedWebmention
+
+SOURCE_URL = "https://source.example/article"
+TARGET_URL = "https://dennisfink.me/blog/example/"
+
+ReceivedWebmentionFactory = Callable[..., uuid.UUID]
+HTTPXMockInstaller = Callable[
+    [Callable[[httpx.Request], httpx.Response]],
+    None,
+]
+
+
+def get_webmention_state(
+    app: Flask,
+    identifier: uuid.UUID,
+) -> tuple[str, str | None]:
+    with app.app_context():
+        webmention = db.session.get(
+            ReceivedWebmention,
+            identifier,
+        )
+
+        assert webmention is not None
+
+        return (
+            webmention.status,
+            webmention.failure_reason,
+        )
+
+
+@pytest.mark.parametrize(
+    (
+        "body",
+        "source_url",
+        "target_url",
+        "expected",
+    ),
+    [
+        pytest.param(
+            f'<a href="{TARGET_URL}">Reply</a>',
+            SOURCE_URL,
+            TARGET_URL,
+            True,
+            id="a-href",
+        ),
+        pytest.param(
+            f'<area href="{TARGET_URL}" alt="Target">',
+            SOURCE_URL,
+            TARGET_URL,
+            True,
+            id="area-href",
+        ),
+        pytest.param(
+            f'<link href="{TARGET_URL}" rel="alternate">',
+            SOURCE_URL,
+            TARGET_URL,
+            True,
+            id="link-href",
+        ),
+        pytest.param(
+            f'<img src="{TARGET_URL}" alt="">',
+            SOURCE_URL,
+            TARGET_URL,
+            True,
+            id="img-src",
+        ),
+        pytest.param(
+            f'<audio src="{TARGET_URL}"></audio>',
+            SOURCE_URL,
+            TARGET_URL,
+            True,
+            id="audio-src",
+        ),
+        pytest.param(
+            f'<video src="{TARGET_URL}"></video>',
+            SOURCE_URL,
+            TARGET_URL,
+            True,
+            id="video-src",
+        ),
+        pytest.param(
+            (f'<audio><source src="{TARGET_URL}"></audio>'),
+            SOURCE_URL,
+            TARGET_URL,
+            True,
+            id="audio-source-src",
+        ),
+        pytest.param(
+            (f'<video><source src="{TARGET_URL}"></video>'),
+            SOURCE_URL,
+            TARGET_URL,
+            True,
+            id="video-source-src",
+        ),
+        pytest.param(
+            f'<iframe src="{TARGET_URL}"></iframe>',
+            SOURCE_URL,
+            TARGET_URL,
+            True,
+            id="iframe-src",
+        ),
+        pytest.param(
+            f'<embed src="{TARGET_URL}">',
+            SOURCE_URL,
+            TARGET_URL,
+            True,
+            id="embed-src",
+        ),
+        pytest.param(
+            f'<script src="{TARGET_URL}"></script>',
+            SOURCE_URL,
+            TARGET_URL,
+            True,
+            id="script-src",
+        ),
+        pytest.param(
+            (f'<video><track src="{TARGET_URL}"></video>'),
+            SOURCE_URL,
+            TARGET_URL,
+            True,
+            id="track-src",
+        ),
+        pytest.param(
+            f'<input type="image" src="{TARGET_URL}" alt="">',
+            SOURCE_URL,
+            TARGET_URL,
+            True,
+            id="image-input-src",
+        ),
+        pytest.param(
+            f'<input type="IMAGE" src="{TARGET_URL}" alt="">',
+            SOURCE_URL,
+            TARGET_URL,
+            True,
+            id="image-input-case-insensitive",
+        ),
+        pytest.param(
+            f'<blockquote cite="{TARGET_URL}">Quotation</blockquote>',
+            SOURCE_URL,
+            TARGET_URL,
+            True,
+            id="blockquote-cite",
+        ),
+        pytest.param(
+            f'<q cite="{TARGET_URL}">Quotation</q>',
+            SOURCE_URL,
+            TARGET_URL,
+            True,
+            id="q-cite",
+        ),
+        pytest.param(
+            f'<ins cite="{TARGET_URL}">Addition</ins>',
+            SOURCE_URL,
+            TARGET_URL,
+            True,
+            id="ins-cite",
+        ),
+        pytest.param(
+            f'<del cite="{TARGET_URL}">Removal</del>',
+            SOURCE_URL,
+            TARGET_URL,
+            True,
+            id="del-cite",
+        ),
+        pytest.param(
+            '<a href="../target/">Reply</a>',
+            "https://source.example/posts/article/",
+            "https://source.example/posts/target/",
+            True,
+            id="relative-href",
+        ),
+        pytest.param(
+            '<base href="https://dennisfink.me/blog/"><a href="example/">Reply</a>',
+            SOURCE_URL,
+            TARGET_URL,
+            True,
+            id="base-url",
+        ),
+        pytest.param(
+            f'<img cite="{TARGET_URL}" alt="">',
+            SOURCE_URL,
+            TARGET_URL,
+            False,
+            id="img-cite-invalid",
+        ),
+        pytest.param(
+            f'<blockquote src="{TARGET_URL}">Quote</blockquote>',
+            SOURCE_URL,
+            TARGET_URL,
+            False,
+            id="blockquote-src-invalid",
+        ),
+        pytest.param(
+            f'<a src="{TARGET_URL}">Reply</a>',
+            SOURCE_URL,
+            TARGET_URL,
+            False,
+            id="a-src-invalid",
+        ),
+        pytest.param(
+            f'<div href="{TARGET_URL}"></div>',
+            SOURCE_URL,
+            TARGET_URL,
+            False,
+            id="div-href-invalid",
+        ),
+        pytest.param(
+            f'<link src="{TARGET_URL}">',
+            SOURCE_URL,
+            TARGET_URL,
+            False,
+            id="link-src-invalid",
+        ),
+        pytest.param(
+            f'<input type="text" src="{TARGET_URL}">',
+            SOURCE_URL,
+            TARGET_URL,
+            False,
+            id="text-input-src-invalid",
+        ),
+        pytest.param(
+            (f'<picture><source src="{TARGET_URL}"></picture>'),
+            SOURCE_URL,
+            TARGET_URL,
+            False,
+            id="picture-source-src-invalid",
+        ),
+        pytest.param(
+            f'<base href="{TARGET_URL}">',
+            SOURCE_URL,
+            TARGET_URL,
+            False,
+            id="base-is-not-mention",
+        ),
+        pytest.param(
+            (f'<a href="{TARGET_URL}more">Different page</a>'),
+            SOURCE_URL,
+            TARGET_URL,
+            False,
+            id="longer-url",
+        ),
+        pytest.param(
+            (f'<img cite="{TARGET_URL}" src="https://example.com/image.jpg" alt="">'),
+            SOURCE_URL,
+            TARGET_URL,
+            False,
+            id="invalid-cite-does-not-override-valid-src",
+        ),
+        pytest.param(
+            f"<p>{TARGET_URL}</p>",
+            SOURCE_URL,
+            TARGET_URL,
+            False,
+            id="text-content",
+        ),
+        pytest.param(
+            '<a href="https://example.com/">Other site</a>',
+            SOURCE_URL,
+            TARGET_URL,
+            False,
+            id="missing-target",
+        ),
+    ],
+)
+def test_html_mentions_target(
+    receiver_module: ModuleType,
+    body: str,
+    source_url: str,
+    target_url: str,
+    expected: bool,
+) -> None:
+    assert (
+        receiver_module.html_mentions_target(
+            body.encode(),
+            source_url,
+            target_url,
+        )
+        is expected
+    )
+
+
+@pytest.mark.parametrize(
+    ("body", "target_url", "expected"),
+    [
+        (TARGET_URL, TARGET_URL, True),
+        (f"This post replies to {TARGET_URL}", TARGET_URL, True),
+        (
+            f"https://example.com/first {TARGET_URL} https://example.com/last",
+            TARGET_URL,
+            True,
+        ),
+        (f"{TARGET_URL}more", TARGET_URL, False),
+        ("https://dennisfink.me/blog/other/", TARGET_URL, False),
+        ("/blog/example/", TARGET_URL, False),
+        ("There are no links here.", TARGET_URL, False),
+    ],
+)
+def test_text_mentions_target(
+    receiver_module: ModuleType,
+    body: str,
+    target_url: str,
+    expected: bool,
+) -> None:
+    assert (
+        receiver_module.text_mentions_target(
+            body,
+            target_url,
+        )
+        is expected
+    )
+
+
+def test_fetch_source_returns_response_and_body(
+    app: Flask,
+    receiver_module: ModuleType,
+    install_httpx_mock: HTTPXMockInstaller,
+) -> None:
+    captured_request: httpx.Request | None = None
+
+    def handler(
+        request: httpx.Request,
+    ) -> httpx.Response:
+        nonlocal captured_request
+        captured_request = request
+
+        return httpx.Response(
+            200,
+            headers={
+                "Content-Type": "text/html",
+            },
+            content=b"<p>Document</p>",
+        )
+
+    install_httpx_mock(handler)
+
+    with app.app_context():
+        response, body = receiver_module.fetch_source(SOURCE_URL)
+
+    assert response.status_code == 200
+    assert body == b"<p>Document</p>"
+
+    assert captured_request is not None
+    assert captured_request.url == SOURCE_URL
+
+    accept = captured_request.headers["Accept"]
+
+    assert "text/html" in accept
+    assert "application/xhtml+xml" in accept
+    assert "text/plain" in accept
+
+    assert captured_request.headers["User-Agent"] == (
+        f"{receiver_module.APP_NAME}/{receiver_module.VERSION} ReceivedWebmention"
+    )
+
+
+@pytest.mark.parametrize(
+    ("status_code", "exception_name"),
+    [
+        (400, "VerificationError"),
+        (404, "VerificationError"),
+        (410, "SourceGoneError"),
+        (408, "TemporaryFetchError"),
+        (425, "TemporaryFetchError"),
+        (429, "TemporaryFetchError"),
+        (500, "TemporaryFetchError"),
+        (503, "TemporaryFetchError"),
+    ],
+)
+def test_fetch_source_maps_http_status_to_exception(
+    app: Flask,
+    receiver_module: ModuleType,
+    install_httpx_mock: HTTPXMockInstaller,
+    status_code: int,
+    exception_name: str,
+) -> None:
+    install_httpx_mock(lambda request: httpx.Response(status_code))
+
+    exception_type = getattr(
+        receiver_module,
+        exception_name,
+    )
+
+    with (
+        app.app_context(),
+        pytest.raises(
+            exception_type,
+            match=f"HTTP {status_code}",
+        ),
+    ):
+        receiver_module.fetch_source(SOURCE_URL)
+
+
+def test_fetch_source_propagates_network_error(
+    app: Flask,
+    receiver_module: ModuleType,
+    install_httpx_mock: HTTPXMockInstaller,
+) -> None:
+    def handler(
+        request: httpx.Request,
+    ) -> httpx.Response:
+        raise httpx.ConnectError(
+            "Connection refused",
+            request=request,
+        )
+
+    install_httpx_mock(handler)
+
+    with (
+        app.app_context(),
+        pytest.raises(
+            httpx.ConnectError,
+            match="Connection refused",
+        ),
+    ):
+        receiver_module.fetch_source(SOURCE_URL)
+
+
+def test_fetch_source_follows_redirect(
+    app: Flask,
+    receiver_module: ModuleType,
+    install_httpx_mock: HTTPXMockInstaller,
+) -> None:
+    requested_paths: list[str] = []
+
+    def handler(
+        request: httpx.Request,
+    ) -> httpx.Response:
+        requested_paths.append(request.url.path)
+
+        match request.url.path:
+            case "/start":
+                return httpx.Response(
+                    302,
+                    headers={
+                        "Location": "/final",
+                    },
+                )
+
+            case "/final":
+                return httpx.Response(
+                    200,
+                    headers={
+                        "Content-Type": "text/html",
+                    },
+                    content=b"Final document",
+                )
+
+            case _:
+                raise AssertionError(f"Unexpected URL: {request.url}")
+
+    install_httpx_mock(handler)
+
+    with app.app_context():
+        response, body = receiver_module.fetch_source("https://source.example/start")
+
+    assert requested_paths == [
+        "/start",
+        "/final",
+    ]
+    assert response.url.path == "/final"
+    assert body == b"Final document"
+
+
+def test_fetch_source_enforces_redirect_limit(
+    app: Flask,
+    receiver_module: ModuleType,
+    install_httpx_mock: HTTPXMockInstaller,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setitem(
+        app.config,
+        "WEBMENTIONS_SSG_MAX_REDIRECTS",
+        1,
+    )
+
+    install_httpx_mock(
+        lambda request: httpx.Response(
+            302,
+            headers={
+                "Location": "/another",
+            },
+        )
+    )
+
+    with (
+        app.app_context(),
+        pytest.raises(httpx.TooManyRedirects),
+    ):
+        receiver_module.fetch_source("https://source.example/start")
+
+
+def test_fetch_source_rejects_declared_oversized_body(
+    app: Flask,
+    receiver_module: ModuleType,
+    install_httpx_mock: HTTPXMockInstaller,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setitem(
+        app.config,
+        "WEBMENTIONS_SSG_MAX_SOURCE_BYTES",
+        10,
+    )
+
+    install_httpx_mock(
+        lambda request: httpx.Response(
+            200,
+            headers={
+                "Content-Type": "text/html",
+                "Content-Length": "11",
+            },
+            content=b"x" * 11,
+        )
+    )
+
+    with (
+        app.app_context(),
+        pytest.raises(
+            receiver_module.VerificationError,
+            match="Source document is too large",
+        ),
+    ):
+        receiver_module.fetch_source(SOURCE_URL)
+
+
+def test_fetch_source_rejects_streamed_oversized_body(
+    app: Flask,
+    receiver_module: ModuleType,
+    install_httpx_mock: HTTPXMockInstaller,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setitem(
+        app.config,
+        "WEBMENTIONS_SSG_MAX_SOURCE_BYTES",
+        10,
+    )
+
+    class BodyStream(httpx.SyncByteStream):
+        def __iter__(self):
+            yield b"x" * 6
+            yield b"x" * 6
+
+    install_httpx_mock(
+        lambda request: httpx.Response(
+            200,
+            headers={
+                "Content-Type": "text/html",
+            },
+            stream=BodyStream(),
+        )
+    )
+
+    with (
+        app.app_context(),
+        pytest.raises(
+            receiver_module.VerificationError,
+            match="Source document is too large",
+        ),
+    ):
+        receiver_module.fetch_source(SOURCE_URL)
+
+
+@pytest.mark.parametrize(
+    ("content_type", "body", "expected"),
+    [
+        pytest.param(
+            "text/html; charset=utf-8",
+            f'<a href="{TARGET_URL}">Reply</a>'.encode(),
+            True,
+            id="html",
+        ),
+        pytest.param(
+            "TEXT/HTML; CHARSET=UTF-8",
+            f'<a href="{TARGET_URL}">Reply</a>'.encode(),
+            True,
+            id="case-insensitive-html",
+        ),
+        pytest.param(
+            "application/xhtml+xml",
+            b'<a href="https://example.com/">Other</a>',
+            False,
+            id="xhtml-without-target",
+        ),
+        pytest.param(
+            "text/plain; charset=utf-8",
+            f"Reply to {TARGET_URL}".encode(),
+            True,
+            id="plain-text-utf-8",
+        ),
+        pytest.param(
+            "text/plain; charset=iso-8859-1",
+            (f"Grüße. Reply to {TARGET_URL}").encode("iso-8859-1"),
+            True,
+            id="plain-text-declared-encoding",
+        ),
+        pytest.param(
+            "text/plain; charset=utf-8",
+            b"\xff Reply to " + TARGET_URL.encode(),
+            True,
+            id="plain-text-invalid-byte",
+        ),
+        pytest.param(
+            "text/plain",
+            b"No target here.",
+            False,
+            id="plain-text-without-target",
+        ),
+    ],
+)
+def test_source_mentions_target_by_media_type(
+    receiver_module: ModuleType,
+    monkeypatch: pytest.MonkeyPatch,
+    content_type: str,
+    body: bytes,
+    expected: bool,
+) -> None:
+    response = httpx.Response(
+        200,
+        headers={
+            "Content-Type": content_type,
+        },
+        content=body,
+        request=httpx.Request(
+            "GET",
+            SOURCE_URL,
+        ),
+    )
+
+    monkeypatch.setattr(
+        receiver_module,
+        "fetch_source",
+        lambda source_url: (
+            response,
+            body,
+        ),
+    )
+
+    assert (
+        receiver_module.source_mentions_target(
+            SOURCE_URL,
+            TARGET_URL,
+        )
+        is expected
+    )
+
+
+@pytest.mark.parametrize(
+    ("content_type", "expected_media_type"),
+    [
+        ("application/json", "application/json"),
+        ("application/pdf", "application/pdf"),
+        ("", "missing"),
+    ],
+)
+def test_source_mentions_target_rejects_unsupported_media_type(
+    receiver_module: ModuleType,
+    monkeypatch: pytest.MonkeyPatch,
+    content_type: str,
+    expected_media_type: str,
+) -> None:
+    headers = {}
+
+    if content_type:
+        headers["Content-Type"] = content_type
+
+    response = httpx.Response(
+        200,
+        headers=headers,
+        content=b"Document",
+        request=httpx.Request(
+            "GET",
+            SOURCE_URL,
+        ),
+    )
+
+    monkeypatch.setattr(
+        receiver_module,
+        "fetch_source",
+        lambda source_url: (
+            response,
+            b"Document",
+        ),
+    )
+
+    with pytest.raises(
+        receiver_module.VerificationError,
+        match=(f"Unsupported source content type: {expected_media_type}"),
+    ):
+        receiver_module.source_mentions_target(
+            SOURCE_URL,
+            TARGET_URL,
+        )
+
+
+def test_verify_webmention_marks_row_verifying_before_check(
+    app: Flask,
+    receiver_module: ModuleType,
+    make_webmention: ReceivedWebmentionFactory,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    identifier = make_webmention(
+        status="failed",
+        failure_reason="Earlier failure",
+    )
+
+    def verify_source(
+        source_url: str,
+        target_url: str,
+    ) -> bool:
+        webmention = db.session.get(
+            ReceivedWebmention,
+            identifier,
+        )
+
+        assert webmention is not None
+        assert webmention.status == "verifying"
+        assert webmention.failure_reason is None
+        assert source_url == webmention.source
+        assert target_url == webmention.target
+
+        return True
+
+    monkeypatch.setattr(
+        receiver_module,
+        "source_mentions_target",
+        verify_source,
+    )
+
+    receiver_module.verify_webmention.call_local(identifier)
+
+    assert get_webmention_state(
+        app,
+        identifier,
+    ) == (
+        "verified",
+        None,
+    )
+
+
+@pytest.mark.parametrize(
+    ("outcome", "expected_status", "expected_reason"),
+    [
+        ("verified", "verified", None),
+        ("missing", "deleted", "Source does not mention target"),
+        ("gone", "deleted", "Source returned HTTP 410"),
+        ("permanent-failure", "failed", "Source returned HTTP 404"),
+    ],
+)
+def test_verify_webmention_persists_final_state(
+    app: Flask,
+    receiver_module: ModuleType,
+    make_webmention: ReceivedWebmentionFactory,
+    monkeypatch: pytest.MonkeyPatch,
+    outcome: str,
+    expected_status: str,
+    expected_reason: str | None,
+) -> None:
+    identifier = make_webmention(
+        status="received",
+    )
+
+    def verify_source(
+        source_url: str,
+        target_url: str,
+    ) -> bool:
+        match outcome:
+            case "verified":
+                return True
+
+            case "missing":
+                return False
+
+            case "gone":
+                raise receiver_module.SourceGoneError("Source returned HTTP 410")
+
+            case "permanent-failure":
+                raise receiver_module.VerificationError("Source returned HTTP 404")
+
+            case _:
+                raise AssertionError(f"Unexpected outcome: {outcome}")
+
+    monkeypatch.setattr(
+        receiver_module,
+        "source_mentions_target",
+        verify_source,
+    )
+
+    receiver_module.verify_webmention.call_local(identifier)
+
+    assert get_webmention_state(
+        app,
+        identifier,
+    ) == (
+        expected_status,
+        expected_reason,
+    )
+
+
+@pytest.mark.parametrize(
+    "failure",
+    [
+        "temporary-http",
+        "network",
+    ],
+)
+def test_verify_webmention_persists_retryable_failure_and_reraises(
+    app: Flask,
+    receiver_module: ModuleType,
+    make_webmention: ReceivedWebmentionFactory,
+    monkeypatch: pytest.MonkeyPatch,
+    failure: str,
+) -> None:
+    identifier = make_webmention()
+
+    def verify_source(
+        source_url: str,
+        target_url: str,
+    ) -> bool:
+        match failure:
+            case "temporary-http":
+                raise receiver_module.TemporaryFetchError("Source returned HTTP 503")
+
+            case "network":
+                raise httpx.ConnectError(
+                    "Connection refused",
+                    request=httpx.Request(
+                        "GET",
+                        source_url,
+                    ),
+                )
+
+            case _:
+                raise AssertionError(f"Unexpected failure: {failure}")
+
+    monkeypatch.setattr(
+        receiver_module,
+        "source_mentions_target",
+        verify_source,
+    )
+
+    match failure:
+        case "temporary-http":
+            expected_exception = receiver_module.TemporaryFetchError
+            expected_reason = "Source returned HTTP 503"
+
+        case "network":
+            expected_exception = httpx.ConnectError
+            expected_reason = "Connection refused"
+
+        case _:
+            raise AssertionError(f"Unexpected failure: {failure}")
+
+    with pytest.raises(
+        expected_exception,
+        match=expected_reason,
+    ):
+        receiver_module.verify_webmention.call_local(identifier)
+
+    assert get_webmention_state(
+        app,
+        identifier,
+    ) == (
+        "failed",
+        expected_reason,
+    )
+
+
+def test_verify_webmention_ignores_unknown_identifier(
+    receiver_module: ModuleType,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    identifier = uuid.uuid7()
+
+    with caplog.at_level(logging.WARNING):
+        receiver_module.verify_webmention.call_local(identifier)
+
+    assert f"Cannot verify unknown ReceivedWebmention {identifier}" in caplog.text
-- 
cgit v1.3.1