diff options
Diffstat (limited to 'tests/tasks')
| -rw-r--r-- | tests/tasks/test_receiver.py | 886 |
1 files changed, 886 insertions, 0 deletions
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 |