From d0e245d63e014d268310976fc45429f08d2cbbe9 Mon Sep 17 00:00:00 2001
From: Dennis Fink
Date: Wed, 19 Aug 2026 19:49:55 +0200
Subject: refactor(core): improve typing and test coverage

Restructure database and user CLI commands, make passwords write-only
model properties, and use the UTC datetime constant throughout the
application.

Add PEP 287-style documentation and expand receiver, scanner, sender,
and URL security tests to cover error paths and edge cases.
---
 tests/tasks/test_receiver.py |  52 ++++++
 tests/tasks/test_scanner.py  | 203 +++++++++++++++++++++++
 tests/tasks/test_sender.py   | 377 ++++++++++++++++++++++++++++++++++++++++++-
 3 files changed, 630 insertions(+), 2 deletions(-)

(limited to 'tests/tasks')

diff --git a/tests/tasks/test_receiver.py b/tests/tasks/test_receiver.py
index 811ab63..7e8f531 100644
--- a/tests/tasks/test_receiver.py
+++ b/tests/tasks/test_receiver.py
@@ -559,3 +559,55 @@ def test_ensure_public_request_rejects_url_without_hostname(
         receiver.ensure_public_request(httpx.Request("GET", SOURCE_URL))
 
     is_public_url.assert_called_once_with(SOURCE_URL)
+
+
+@patch("webmentions_ssg.tasks.receiver.BeautifulSoup")
+def test_html_mentions_target_ignores_non_string_attribute(
+    beautiful_soup: Mock, receiver: ModuleType
+) -> None:
+    document = beautiful_soup.return_value
+    document.select_one.return_value = None
+
+    element = Mock()
+    element.get.return_value = ["not", "a", "string"]
+    document.select.return_value = [element]
+
+    assert not receiver.html_mentions_target(b"<html></html>", SOURCE_URL, TARGET_URL)
+
+
+@patch("webmentions_ssg.tasks.receiver.httpx.Client")
+def test_fetch_source_ignores_invalid_content_length(
+    httpx_client: Mock, app: Flask, receiver: ModuleType
+) -> None:
+    response = httpx.Response(
+        200,
+        headers={"Content-Type": "text/html", "Content-Length": "invalid"},
+        content=b"<p>Document</p>",
+        request=httpx.Request("GET", SOURCE_URL),
+    )
+    set_stream_response(httpx_client, response)
+
+    with app.app_context():
+        fetched_response, body = receiver.fetch_source(SOURCE_URL)
+
+    assert fetched_response is response
+    assert body == b"<p>Document</p>"
+
+
+@patch("webmentions_ssg.tasks.receiver.fetch_source")
+def test_source_mentions_target_falls_back_for_unknown_encoding(
+    fetch_source: Mock, receiver: ModuleType
+) -> None:
+    body = f"Reply to {TARGET_URL}".encode()
+
+    response = httpx.Response(
+        200,
+        headers={"Content-Type": "text/plain"},
+        content=body,
+        request=httpx.Request("GET", SOURCE_URL),
+    )
+    response.encoding = "unknown-encoding"
+    fetch_source.return_value = response, body
+
+    assert receiver.source_mentions_target(SOURCE_URL, TARGET_URL)
+    fetch_source.assert_called_once_with(SOURCE_URL)
diff --git a/tests/tasks/test_scanner.py b/tests/tasks/test_scanner.py
index 65031b4..89e2995 100644
--- a/tests/tasks/test_scanner.py
+++ b/tests/tasks/test_scanner.py
@@ -1556,3 +1556,206 @@ def test_scan_source_ignores_reaction_to_matching_hostname(
     )
 
     assert scanned.targets == frozenset()
+
+
+def test_canonical_url_rejects_non_http_resolved_url(
+    scanner_module: ModuleType,
+) -> None:
+    document = parse_html(
+        """
+        <html>
+          <head>
+            <link
+              rel="canonical"
+              href="mailto:example@example.com"
+            >
+          </head>
+        </html>
+        """
+    )
+
+    with pytest.raises(
+        scanner_module.SourceScanError,
+        match="Canonical URL is not a valid HTTP or HTTPS URL",
+    ):
+        scanner_module.canonical_url(
+            document, relative_path=Path("example/index.html"), base_url=BASE_URL
+        )
+
+
+def test_property_urls_ignores_non_mapping_properties(
+    scanner_module: ModuleType,
+) -> None:
+    entry = {"properties": "invalid"}
+
+    assert tuple(scanner_module.property_urls(entry, "url")) == ()
+
+
+@pytest.mark.parametrize(
+    ("parsed", "message"),
+    [
+        pytest.param(
+            [],
+            "Microformats parser did not return a document object",
+            id="non-object-document",
+        ),
+        pytest.param(
+            {"items": [None]}, "Parsed h-entry is not an object", id="non-object-entry"
+        ),
+        pytest.param(
+            {"items": [{"type": ["h-card"]}]},
+            "Parsed microformats item is not an h-entry",
+            id="wrong-item-type",
+        ),
+    ],
+)
+def test_parse_entry_rejects_invalid_parser_output(
+    scanner_module: ModuleType, monkeypatch: MonkeyPatch, parsed: object, message: str
+) -> None:
+    document = parse_html(
+        """
+        <article class="h-entry">
+          <p>Hello world</p>
+        </article>
+        """
+    )
+
+    element = document.find(class_="h-entry")
+
+    assert element is not None
+
+    monkeypatch.setattr(scanner_module.mf2py, "parse", lambda **kwargs: parsed)
+
+    with pytest.raises(scanner_module.SourceScanError, match=message):
+        scanner_module.parse_entry(element, SOURCE_URL)
+
+
+@pytest.mark.parametrize(
+    "value",
+    [
+        pytest.param(" ", id="empty"),
+        pytest.param("mailto:example@example.com", id="non-http"),
+    ],
+)
+def test_normalize_target_ignores_invalid_target(
+    scanner_module: ModuleType, value: str
+) -> None:
+    assert (
+        scanner_module.normalize_target(
+            value, base_url=SOURCE_URL, source_url=SOURCE_URL
+        )
+        is None
+    )
+
+
+def test_changed_source_url_is_rejected(
+    app: Flask,
+    scanner_module: ModuleType,
+    tmp_path: Path,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    configure_scanner(app, tmp_path)
+
+    write_post(
+        tmp_path,
+        "example",
+        """
+        <p>Hello world.</p>
+        """,
+    )
+
+    run_scan(app, scanner_module)
+
+    changed_url = f"{BASE_URL}changed/"
+
+    write_post(
+        tmp_path,
+        "example",
+        """
+        <p>Hello world.</p>
+        """,
+        canonical=changed_url,
+        entry_url=changed_url,
+    )
+
+    run_scan(app, scanner_module)
+
+    with app.app_context():
+        source = db.session.scalar(sa.select(Source))
+
+        assert source is not None
+        assert source.url == SOURCE_URL
+        assert source.revision == 1
+
+    assert "changed public URL" in caplog.text
+
+
+def test_scan_sources_returns_when_source_directory_is_not_configured(
+    app: Flask, scanner_module: ModuleType
+) -> None:
+    app.config["WEBMENTIONS_SSG_SOURCE_DIRECTORY"] = None
+
+    run_scan(app, scanner_module)
+
+    with app.app_context():
+        assert db.session.scalar(sa.select(Source)) is None
+
+
+def test_scan_sources_rejects_source_directory_that_is_not_directory(
+    app: Flask, scanner_module: ModuleType, tmp_path: Path
+) -> None:
+    path = tmp_path / "source.html"
+    path.write_text("<p>Hello world.</p>", encoding="utf-8")
+
+    configure_scanner(app, path)
+
+    with pytest.raises(RuntimeError, match="Source directory is not a directory"):
+        run_scan(app, scanner_module)
+
+
+def test_scan_sources_ignores_root_index(
+    app: Flask, scanner_module: ModuleType, tmp_path: Path
+) -> None:
+    configure_scanner(app, tmp_path)
+
+    (tmp_path / "index.html").write_text(
+        f"""
+        <!doctype html>
+        <html>
+          <head>
+            <link
+              rel="canonical"
+              href="{BASE_URL}"
+            >
+          </head>
+          <body>
+            <article class="h-entry">
+              <a class="u-url" href="{BASE_URL}">
+                Permalink
+              </a>
+
+              <div class="e-content">
+                <p>Blog index.</p>
+              </div>
+            </article>
+          </body>
+        </html>
+        """,
+        encoding="utf-8",
+    )
+
+    write_post(
+        tmp_path,
+        "example",
+        """
+        <p>Hello world.</p>
+        """,
+    )
+
+    run_scan(app, scanner_module)
+
+    with app.app_context():
+        sources = list(db.session.scalars(sa.select(Source)))
+
+        assert len(sources) == 1
+        assert sources[0].path == "example/index.html"
diff --git a/tests/tasks/test_sender.py b/tests/tasks/test_sender.py
index 86799f7..1c8ef99 100644
--- a/tests/tasks/test_sender.py
+++ b/tests/tasks/test_sender.py
@@ -1,6 +1,6 @@
 import logging
 import uuid
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from types import ModuleType
 from unittest.mock import MagicMock, Mock
 
@@ -10,6 +10,7 @@ from flask import Flask
 
 from webmentions_ssg import DATABASE as db
 from webmentions_ssg.models import SentWebmention, SentWebmentionStatus, Source
+from webmentions_ssg.url_security import AddressResolutionError
 
 SOURCE_URL = "https://dennisfink.me/blog/example/"
 TARGET_URL = "https://example.com/post"
@@ -30,7 +31,7 @@ def create_sent_webmention(
     response_status: int | None = None,
     status_url: str | None = None,
 ) -> uuid.UUID:
-    now = datetime.now(timezone.utc)
+    now = datetime.now(UTC)
 
     with app.app_context():
         source = Source(
@@ -786,3 +787,375 @@ def test_send_webmention_ignores_unknown_identifier(
         sender_module.send_webmention.call_local(identifier)
 
     assert f"Cannot send unknown SentWebmention {identifier}" in caplog.text
+
+
+def test_ensure_public_request_accepts_public_url(
+    sender_module: ModuleType, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    is_public_url = Mock(return_value=True)
+    monkeypatch.setattr(sender_module, "is_public_url", is_public_url)
+
+    sender_module.ensure_public_request(httpx.Request("GET", TARGET_URL))
+
+    is_public_url.assert_called_once_with(TARGET_URL)
+
+
+def test_ensure_public_request_rejects_non_public_address(
+    sender_module: ModuleType, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    is_public_url = Mock(return_value=False)
+    monkeypatch.setattr(sender_module, "is_public_url", is_public_url)
+
+    with pytest.raises(
+        sender_module.PermanentSenderError,
+        match="Request resolves to a non-public address",
+    ):
+        sender_module.ensure_public_request(httpx.Request("GET", TARGET_URL))
+
+    is_public_url.assert_called_once_with(TARGET_URL)
+
+
+def test_ensure_public_request_maps_dns_failure_to_temporary_error(
+    sender_module: ModuleType, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    is_public_url = Mock(
+        side_effect=AddressResolutionError("Could not resolve hostname")
+    )
+    monkeypatch.setattr(sender_module, "is_public_url", is_public_url)
+
+    with pytest.raises(
+        sender_module.TemporarySenderError,
+        match="Request hostname could not be resolved",
+    ):
+        sender_module.ensure_public_request(httpx.Request("GET", TARGET_URL))
+
+    is_public_url.assert_called_once_with(TARGET_URL)
+
+
+def test_resolve_endpoint_rejects_non_http_url(sender_module: ModuleType) -> None:
+    response = httpx.Response(200, request=httpx.Request("GET", TARGET_URL))
+
+    with pytest.raises(
+        sender_module.PermanentSenderError, match="Invalid Webmention endpoint"
+    ):
+        sender_module.resolve_endpoint(response, "mailto:example@example.com")
+
+
+def test_endpoint_from_headers_ignores_malformed_link(
+    sender_module: ModuleType,
+) -> None:
+    response = httpx.Response(
+        200,
+        headers=[("Link", "not-a-link"), ("Link", "</webmention>; rel=webmention")],
+        request=httpx.Request("GET", TARGET_URL),
+    )
+
+    assert sender_module.endpoint_from_headers(response) == ENDPOINT_URL
+
+
+def test_endpoint_from_html_accepts_string_rel(
+    sender_module: ModuleType, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    element = Mock()
+    element.get.side_effect = lambda name: {
+        "rel": "webmention alternate",
+        "href": "/webmention",
+    }[name]
+
+    document = Mock()
+    document.find_all.return_value = [element]
+
+    beautiful_soup = Mock(return_value=document)
+    monkeypatch.setattr(sender_module, "BeautifulSoup", beautiful_soup)
+
+    response = httpx.Response(200, request=httpx.Request("GET", TARGET_URL))
+
+    assert sender_module.endpoint_from_html(response, b"") == ENDPOINT_URL
+
+
+def test_endpoint_from_html_ignores_elements_without_rel_and_non_string_href(
+    sender_module: ModuleType, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    missing_rel = Mock()
+    missing_rel.get.return_value = None
+
+    invalid_href = Mock()
+    invalid_href.get.side_effect = lambda name: {
+        "rel": ["webmention"],
+        "href": ["invalid"],
+    }[name]
+
+    document = Mock()
+    document.find_all.return_value = [missing_rel, invalid_href]
+
+    beautiful_soup = Mock(return_value=document)
+    monkeypatch.setattr(sender_module, "BeautifulSoup", beautiful_soup)
+
+    response = httpx.Response(200, request=httpx.Request("GET", TARGET_URL))
+
+    assert sender_module.endpoint_from_html(response, b"") is None
+
+
+def test_read_target_body_rejects_declared_oversized_body(
+    app: Flask, sender_module: ModuleType
+) -> None:
+    app.config["WEBMENTIONS_SSG_MAX_TARGET_BYTES"] = 10
+
+    response = httpx.Response(200, headers={"Content-Length": "11"}, content=b"x" * 11)
+
+    with (
+        app.app_context(),
+        pytest.raises(
+            sender_module.PermanentSenderError, match="Target document is too large"
+        ),
+    ):
+        sender_module.read_target_body(response)
+
+
+def test_read_target_body_ignores_invalid_content_length(
+    app: Flask, sender_module: ModuleType
+) -> None:
+    response = httpx.Response(
+        200, headers={"Content-Length": "invalid"}, content=b"Document"
+    )
+
+    with app.app_context():
+        body = sender_module.read_target_body(response)
+
+    assert body == b"Document"
+
+
+def test_read_target_body_rejects_streamed_oversized_body(
+    app: Flask, sender_module: ModuleType
+) -> None:
+    app.config["WEBMENTIONS_SSG_MAX_TARGET_BYTES"] = 10
+
+    class BodyStream(httpx.SyncByteStream):
+        def __iter__(self):
+            yield b"x" * 6
+            yield b"x" * 6
+
+    response = httpx.Response(200, stream=BodyStream())
+
+    with (
+        app.app_context(),
+        pytest.raises(
+            sender_module.PermanentSenderError, match="Target document is too large"
+        ),
+    ):
+        sender_module.read_target_body(response)
+
+
+def test_discovery_uses_link_header_from_get(
+    app: Flask, sender_module: ModuleType
+) -> None:
+    requests: list[httpx.Request] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        requests.append(request)
+
+        if request.method == "HEAD":
+            return httpx.Response(200)
+
+        return httpx.Response(200, headers={"Link": "</webmention>; rel=webmention"})
+
+    with (
+        app.app_context(),
+        httpx.Client(
+            transport=httpx.MockTransport(handler), follow_redirects=True
+        ) as client,
+    ):
+        endpoint = sender_module.discover_webmention_endpoint(client, TARGET_URL)
+
+    assert endpoint == ENDPOINT_URL
+    assert [request.method for request in requests] == ["HEAD", "GET"]
+
+
+@pytest.mark.parametrize(
+    ("status_code", "exception_name"),
+    [
+        pytest.param(503, "TemporarySenderError", id="temporary"),
+        pytest.param(404, "PermanentSenderError", id="permanent"),
+    ],
+)
+def test_discovery_maps_get_status_to_exception(
+    app: Flask, sender_module: ModuleType, status_code: int, exception_name: str
+) -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        if request.method == "HEAD":
+            return httpx.Response(200)
+
+        return httpx.Response(status_code)
+
+    exception_type = getattr(sender_module, exception_name)
+
+    with (
+        app.app_context(),
+        httpx.Client(transport=httpx.MockTransport(handler)) as client,
+        pytest.raises(exception_type, match=f"Target returned HTTP {status_code}"),
+    ):
+        sender_module.discover_webmention_endpoint(client, TARGET_URL)
+
+
+def test_discovery_returns_none_for_unsupported_media_type(
+    app: Flask, sender_module: ModuleType
+) -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        if request.method == "HEAD":
+            return httpx.Response(200)
+
+        return httpx.Response(
+            200, headers={"Content-Type": "application/json"}, content=b"{}"
+        )
+
+    with (
+        app.app_context(),
+        httpx.Client(transport=httpx.MockTransport(handler)) as client,
+    ):
+        endpoint = sender_module.discover_webmention_endpoint(client, TARGET_URL)
+
+    assert endpoint is None
+
+
+@pytest.mark.parametrize(
+    ("status_code", "location", "expected_status_url"),
+    [
+        pytest.param(
+            201, "/webmention/status/123", STATUS_URL, id="created-with-location"
+        ),
+        pytest.param(202, None, None, id="accepted"),
+    ],
+)
+def test_post_webmention(
+    sender_module: ModuleType,
+    status_code: int,
+    location: str | None,
+    expected_status_url: str | None,
+) -> None:
+    requests: list[httpx.Request] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        requests.append(request)
+
+        headers = {}
+
+        if location is not None:
+            headers["Location"] = location
+
+        return httpx.Response(status_code, headers=headers)
+
+    with httpx.Client(transport=httpx.MockTransport(handler)) as client:
+        result = sender_module.post_webmention(
+            client, endpoint=ENDPOINT_URL, source=SOURCE_URL, target=TARGET_URL
+        )
+
+    assert result == (status_code, expected_status_url)
+    assert len(requests) == 1
+
+    request = requests[0]
+
+    assert request.method == "POST"
+    assert str(request.url) == ENDPOINT_URL
+
+    form = httpx.QueryParams(request.content.decode())
+
+    assert form["source"] == SOURCE_URL
+    assert form["target"] == TARGET_URL
+
+
+def test_send_webmention_ignores_unsupported_result_for_superseded_revision(
+    app: Flask, sender_module: ModuleType, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    identifier = create_sent_webmention(app)
+
+    _, _, discover, post = mock_sender_requests(
+        sender_module, monkeypatch, endpoint=None
+    )
+
+    def supersede_revision(*args, **kwargs) -> None:
+        webmention = db.session.get(SentWebmention, identifier)
+
+        assert webmention is not None
+
+        webmention.desired_revision = 2
+        db.session.commit()
+
+    discover.side_effect = supersede_revision
+
+    sender_module.send_webmention.call_local(identifier)
+
+    post.assert_not_called()
+
+    with app.app_context():
+        webmention = db.session.get(SentWebmention, identifier)
+
+        assert webmention is not None
+        assert webmention.desired_revision == 2
+        assert webmention.processed_revision is None
+        assert webmention.sent_revision is None
+        assert webmention.status is None
+        assert webmention.pending
+
+
+def test_send_webmention_ignores_temporary_failure_for_superseded_revision(
+    app: Flask, sender_module: ModuleType, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    identifier = create_sent_webmention(app)
+
+    _, _, _, post = mock_sender_requests(sender_module, monkeypatch)
+
+    def supersede_revision(*args, **kwargs) -> tuple[int, None]:
+        webmention = db.session.get(SentWebmention, identifier)
+
+        assert webmention is not None
+
+        webmention.desired_revision = 2
+        db.session.commit()
+
+        raise sender_module.TemporarySenderError("Temporary failure")
+
+    post.side_effect = supersede_revision
+
+    sender_module.send_webmention.call_local(identifier)
+
+    with app.app_context():
+        webmention = db.session.get(SentWebmention, identifier)
+
+        assert webmention is not None
+        assert webmention.desired_revision == 2
+        assert webmention.processed_revision is None
+        assert webmention.sent_revision is None
+        assert webmention.status is None
+        assert webmention.pending
+
+
+def test_send_webmention_ignores_permanent_failure_for_superseded_revision(
+    app: Flask, sender_module: ModuleType, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    identifier = create_sent_webmention(app)
+
+    _, _, _, post = mock_sender_requests(sender_module, monkeypatch)
+
+    def supersede_revision(*args, **kwargs) -> tuple[int, None]:
+        webmention = db.session.get(SentWebmention, identifier)
+
+        assert webmention is not None
+
+        webmention.desired_revision = 2
+        db.session.commit()
+
+        return (400, None)
+
+    post.side_effect = supersede_revision
+
+    sender_module.send_webmention.call_local(identifier)
+
+    with app.app_context():
+        webmention = db.session.get(SentWebmention, identifier)
+
+        assert webmention is not None
+        assert webmention.desired_revision == 2
+        assert webmention.processed_revision is None
+        assert webmention.sent_revision is None
+        assert webmention.status is None
+        assert webmention.pending
-- 
cgit v1.3.1