1 files changed, 375 insertions, 2 deletions
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