diff options
Diffstat (limited to '')
| -rw-r--r-- | tests/tasks/test_receiver.py | 396 | ||||
| -rw-r--r-- | tests/tasks/test_scanner.py | 1558 | ||||
| -rw-r--r-- | tests/tasks/test_sender.py | 788 |
3 files changed, 2453 insertions, 289 deletions
diff --git a/tests/tasks/test_receiver.py b/tests/tasks/test_receiver.py index 19630c3..811ab63 100644 --- a/tests/tasks/test_receiver.py +++ b/tests/tasks/test_receiver.py @@ -9,10 +9,7 @@ from flask import Flask from webmentions_ssg import DATABASE as db from webmentions_ssg.models import ReceivedWebmention -from webmentions_ssg.url_security import ( - AddressResolutionError, - NonPublicAddressError, -) +from webmentions_ssg.url_security import AddressResolutionError SOURCE_URL = "https://source.example/article" TARGET_URL = "https://dennisfink.me/blog/example/" @@ -50,28 +47,16 @@ def create_webmention( return identifier -def get_webmention_state( - app: Flask, - identifier: uuid.UUID, -) -> tuple[str, str | None]: +def get_webmention_state(app: Flask, identifier: uuid.UUID) -> tuple[str, str | None]: with app.app_context(): - webmention = db.session.get( - ReceivedWebmention, - identifier, - ) + webmention = db.session.get(ReceivedWebmention, identifier) assert webmention is not None - return ( - webmention.status, - webmention.failure_reason, - ) + return (webmention.status, webmention.failure_reason) -def set_stream_response( - httpx_client: Mock, - response: httpx.Response, -) -> Mock: +def set_stream_response(httpx_client: Mock, response: httpx.Response) -> Mock: client = httpx_client.return_value.__enter__.return_value client.stream.return_value.__enter__.return_value = response return client @@ -80,65 +65,25 @@ def set_stream_response( @pytest.mark.parametrize( ("body", "expected"), [ + pytest.param(f'<a href="{TARGET_URL}">Reply</a>', True, id="a-href"), + pytest.param(f'<area href="{TARGET_URL}" alt="Target">', True, id="area-href"), pytest.param( - f'<a href="{TARGET_URL}">Reply</a>', - True, - id="a-href", - ), - pytest.param( - f'<area href="{TARGET_URL}" alt="Target">', - True, - id="area-href", - ), - pytest.param( - f'<link href="{TARGET_URL}" rel="alternate">', - True, - id="link-href", - ), - pytest.param( - f'<img src="{TARGET_URL}" alt="">', - True, - id="img-src", - ), - pytest.param( - f'<audio src="{TARGET_URL}"></audio>', - True, - id="audio-src", - ), - pytest.param( - f'<video src="{TARGET_URL}"></video>', - True, - id="video-src", - ), - pytest.param( - f'<audio><source src="{TARGET_URL}"></audio>', - True, - id="audio-source-src", + f'<link href="{TARGET_URL}" rel="alternate">', True, id="link-href" ), + pytest.param(f'<img src="{TARGET_URL}" alt="">', True, id="img-src"), + pytest.param(f'<audio src="{TARGET_URL}"></audio>', True, id="audio-src"), + pytest.param(f'<video src="{TARGET_URL}"></video>', True, id="video-src"), pytest.param( - f'<video><source src="{TARGET_URL}"></video>', - True, - id="video-source-src", + f'<audio><source src="{TARGET_URL}"></audio>', True, id="audio-source-src" ), pytest.param( - f'<iframe src="{TARGET_URL}"></iframe>', - True, - id="iframe-src", + f'<video><source src="{TARGET_URL}"></video>', True, id="video-source-src" ), + pytest.param(f'<iframe src="{TARGET_URL}"></iframe>', True, id="iframe-src"), + pytest.param(f'<embed src="{TARGET_URL}">', True, id="embed-src"), + pytest.param(f'<script src="{TARGET_URL}"></script>', True, id="script-src"), pytest.param( - f'<embed src="{TARGET_URL}">', - True, - id="embed-src", - ), - pytest.param( - f'<script src="{TARGET_URL}"></script>', - True, - id="script-src", - ), - pytest.param( - f'<video><track src="{TARGET_URL}"></video>', - True, - id="track-src", + f'<video><track src="{TARGET_URL}"></video>', True, id="track-src" ), pytest.param( f'<input type="image" src="{TARGET_URL}" alt="">', @@ -155,51 +100,23 @@ def set_stream_response( True, id="blockquote-cite", ), - pytest.param( - f'<q cite="{TARGET_URL}">Quotation</q>', - True, - id="q-cite", - ), - pytest.param( - f'<ins cite="{TARGET_URL}">Addition</ins>', - True, - id="ins-cite", - ), - pytest.param( - f'<del cite="{TARGET_URL}">Removal</del>', - True, - id="del-cite", - ), + pytest.param(f'<q cite="{TARGET_URL}">Quotation</q>', True, id="q-cite"), + pytest.param(f'<ins cite="{TARGET_URL}">Addition</ins>', True, id="ins-cite"), + pytest.param(f'<del cite="{TARGET_URL}">Removal</del>', True, id="del-cite"), pytest.param( '<base href="https://dennisfink.me/blog/"><a href="example/">Reply</a>', True, id="base-url", ), - pytest.param( - f'<img cite="{TARGET_URL}" alt="">', - False, - id="img-cite-invalid", - ), + pytest.param(f'<img cite="{TARGET_URL}" alt="">', False, id="img-cite-invalid"), pytest.param( f'<blockquote src="{TARGET_URL}">Quote</blockquote>', False, id="blockquote-src-invalid", ), - pytest.param( - f'<a src="{TARGET_URL}">Reply</a>', - False, - id="a-src-invalid", - ), - pytest.param( - f'<div href="{TARGET_URL}"></div>', - False, - id="div-href-invalid", - ), - pytest.param( - f'<link src="{TARGET_URL}">', - False, - id="link-src-invalid", - ), + pytest.param(f'<a src="{TARGET_URL}">Reply</a>', False, id="a-src-invalid"), + pytest.param(f'<div href="{TARGET_URL}"></div>', False, id="div-href-invalid"), + pytest.param(f'<link src="{TARGET_URL}">', False, id="link-src-invalid"), pytest.param( f'<input type="text" src="{TARGET_URL}">', False, @@ -210,51 +127,28 @@ def set_stream_response( False, id="picture-source-src-invalid", ), + pytest.param(f'<base href="{TARGET_URL}">', False, id="base-is-not-mention"), pytest.param( - f'<base href="{TARGET_URL}">', - False, - id="base-is-not-mention", - ), - pytest.param( - f'<a href="{TARGET_URL}more">Different page</a>', - False, - id="longer-url", + f'<a href="{TARGET_URL}more">Different page</a>', False, id="longer-url" ), pytest.param( f'<img cite="{TARGET_URL}" src="https://example.com/image.jpg" alt="">', False, id="invalid-cite-does-not-override-valid-src", ), + pytest.param(f"<p>{TARGET_URL}</p>", False, id="text-content"), pytest.param( - f"<p>{TARGET_URL}</p>", - False, - id="text-content", - ), - pytest.param( - '<a href="https://example.com/">Other site</a>', - False, - id="missing-target", + '<a href="https://example.com/">Other site</a>', False, id="missing-target" ), ], ) -def test_html_mentions_target( - receiver: ModuleType, - body: str, - expected: bool, -) -> None: +def test_html_mentions_target(receiver: ModuleType, body: str, expected: bool) -> None: assert ( - receiver.html_mentions_target( - body.encode(), - SOURCE_URL, - TARGET_URL, - ) - is expected + receiver.html_mentions_target(body.encode(), SOURCE_URL, TARGET_URL) is expected ) -def test_html_mentions_relative_target( - receiver: ModuleType, -) -> None: +def test_html_mentions_relative_target(receiver: ModuleType) -> None: assert receiver.html_mentions_target( b'<a href="../target/">Reply</a>', "https://source.example/posts/article/", @@ -279,63 +173,14 @@ def test_html_mentions_relative_target( ], ) def test_text_mentions_target( - receiver: ModuleType, - body: str, - target_url: str, - expected: bool, + receiver: ModuleType, body: str, target_url: str, expected: bool ) -> None: assert receiver.text_mentions_target(body, target_url) is expected -@patch("webmentions_ssg.tasks.receiver.ensure_public_url") -def test_ensure_public_request_accepts_public_url( - ensure_public_url: Mock, - receiver: ModuleType, -) -> None: - receiver.ensure_public_request(httpx.Request("GET", SOURCE_URL)) - - ensure_public_url.assert_called_once_with(SOURCE_URL) - - -@patch( - "webmentions_ssg.tasks.receiver.ensure_public_url", - side_effect=NonPublicAddressError("Non-public address"), -) -def test_ensure_public_request_rejects_non_public_address( - ensure_public_url: Mock, - receiver: ModuleType, -) -> None: - with pytest.raises( - receiver.VerificationError, - match="Source resolves to a non-public address", - ): - receiver.ensure_public_request(httpx.Request("GET", SOURCE_URL)) - - ensure_public_url.assert_called_once_with(SOURCE_URL) - - -@patch( - "webmentions_ssg.tasks.receiver.ensure_public_url", - side_effect=AddressResolutionError("Could not resolve hostname"), -) -def test_ensure_public_request_maps_dns_failure_to_temporary_error( - ensure_public_url: Mock, - receiver: ModuleType, -) -> None: - with pytest.raises( - receiver.TemporaryFetchError, - match="Source hostname could not be resolved", - ): - receiver.ensure_public_request(httpx.Request("GET", SOURCE_URL)) - - ensure_public_url.assert_called_once_with(SOURCE_URL) - - @patch("webmentions_ssg.tasks.receiver.httpx.Client") def test_fetch_source_returns_response_and_body( - httpx_client: Mock, - app: Flask, - receiver: ModuleType, + httpx_client: Mock, app: Flask, receiver: ModuleType ) -> None: response = httpx.Response( 200, @@ -360,13 +205,10 @@ def test_fetch_source_returns_response_and_body( } assert options["follow_redirects"] is True assert options["max_redirects"] == app.config.get( - "WEBMENTIONS_SSG_MAX_REDIRECTS", - 20, + "WEBMENTIONS_SSG_MAX_REDIRECTS", 20 ) assert options["trust_env"] is False - assert options["event_hooks"] == { - "request": [receiver.ensure_public_request], - } + assert options["event_hooks"] == {"request": [receiver.ensure_public_request]} @patch("webmentions_ssg.tasks.receiver.httpx.Client") @@ -392,51 +234,34 @@ def test_fetch_source_maps_http_status_to_exception( ) -> None: set_stream_response( httpx_client, - httpx.Response( - status_code, - request=httpx.Request("GET", SOURCE_URL), - ), + httpx.Response(status_code, request=httpx.Request("GET", SOURCE_URL)), ) exception_type = getattr(receiver, exception_name) - with ( - app.app_context(), - pytest.raises( - exception_type, - match=f"HTTP {status_code}", - ), - ): + with app.app_context(), pytest.raises(exception_type, match=f"HTTP {status_code}"): receiver.fetch_source(SOURCE_URL) @patch("webmentions_ssg.tasks.receiver.httpx.Client") def test_fetch_source_propagates_network_error( - httpx_client: Mock, - app: Flask, - receiver: ModuleType, + httpx_client: Mock, app: Flask, receiver: ModuleType ) -> None: client = httpx_client.return_value.__enter__.return_value client.stream.side_effect = httpx.ConnectError( - "Connection refused", - request=httpx.Request("GET", SOURCE_URL), + "Connection refused", request=httpx.Request("GET", SOURCE_URL) ) with ( app.app_context(), - pytest.raises( - httpx.ConnectError, - match="Connection refused", - ), + pytest.raises(httpx.ConnectError, match="Connection refused"), ): receiver.fetch_source(SOURCE_URL) @patch("webmentions_ssg.tasks.receiver.httpx.Client") def test_fetch_source_rejects_declared_oversized_body( - httpx_client: Mock, - app: Flask, - receiver: ModuleType, + httpx_client: Mock, app: Flask, receiver: ModuleType ) -> None: app.config["WEBMENTIONS_SSG_MAX_SOURCE_BYTES"] = 10 @@ -444,10 +269,7 @@ def test_fetch_source_rejects_declared_oversized_body( httpx_client, httpx.Response( 200, - headers={ - "Content-Type": "text/html", - "Content-Length": "11", - }, + headers={"Content-Type": "text/html", "Content-Length": "11"}, content=b"x" * 11, request=httpx.Request("GET", SOURCE_URL), ), @@ -455,19 +277,14 @@ def test_fetch_source_rejects_declared_oversized_body( with ( app.app_context(), - pytest.raises( - receiver.VerificationError, - match="Source document is too large", - ), + pytest.raises(receiver.VerificationError, match="Source document is too large"), ): receiver.fetch_source(SOURCE_URL) @patch("webmentions_ssg.tasks.receiver.httpx.Client") def test_fetch_source_rejects_streamed_oversized_body( - httpx_client: Mock, - app: Flask, - receiver: ModuleType, + httpx_client: Mock, app: Flask, receiver: ModuleType ) -> None: app.config["WEBMENTIONS_SSG_MAX_SOURCE_BYTES"] = 10 @@ -488,10 +305,7 @@ def test_fetch_source_rejects_streamed_oversized_body( with ( app.app_context(), - pytest.raises( - receiver.VerificationError, - match="Source document is too large", - ), + pytest.raises(receiver.VerificationError, match="Source document is too large"), ): receiver.fetch_source(SOURCE_URL) @@ -537,10 +351,7 @@ def test_fetch_source_rejects_streamed_oversized_body( id="plain-text-invalid-byte", ), pytest.param( - "text/plain", - b"No target here.", - False, - id="plain-text-without-target", + "text/plain", b"No target here.", False, id="plain-text-without-target" ), ], ) @@ -600,24 +411,14 @@ def test_source_mentions_target_rejects_unsupported_media_type( @patch("webmentions_ssg.tasks.receiver.source_mentions_target") def test_verify_webmention_marks_row_verifying_before_check( - source_mentions_target: Mock, - app: Flask, - receiver: ModuleType, + source_mentions_target: Mock, app: Flask, receiver: ModuleType ) -> None: identifier = create_webmention( - app, - status="failed", - failure_reason="Earlier failure", + app, status="failed", failure_reason="Earlier failure" ) - def verify_source( - source_url: str, - target_url: str, - ) -> bool: - webmention = db.session.get( - ReceivedWebmention, - identifier, - ) + 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" @@ -631,13 +432,7 @@ def test_verify_webmention_marks_row_verifying_before_check( receiver.verify_webmention.call_local(identifier) - assert get_webmention_state( - app, - identifier, - ) == ( - "verified", - None, - ) + assert get_webmention_state(app, identifier) == ("verified", None) @patch("webmentions_ssg.tasks.receiver.source_mentions_target") @@ -678,28 +473,13 @@ def test_verify_webmention_persists_final_state( receiver.verify_webmention.call_local(identifier) - assert get_webmention_state( - app, - identifier, - ) == ( - expected_status, - expected_reason, - ) + assert get_webmention_state(app, identifier) == (expected_status, expected_reason) @patch("webmentions_ssg.tasks.receiver.source_mentions_target") -@pytest.mark.parametrize( - "failure", - [ - "temporary-http", - "network", - ], -) +@pytest.mark.parametrize("failure", ["temporary-http", "network"]) def test_verify_webmention_persists_retryable_failure_and_reraises( - source_mentions_target: Mock, - app: Flask, - receiver: ModuleType, - failure: str, + source_mentions_target: Mock, app: Flask, receiver: ModuleType, failure: str ) -> None: identifier = create_webmention(app) @@ -708,32 +488,21 @@ def test_verify_webmention_persists_retryable_failure_and_reraises( exception = receiver.TemporaryFetchError("Source returned HTTP 503") case "network": exception = httpx.ConnectError( - "Connection refused", - request=httpx.Request("GET", SOURCE_URL), + "Connection refused", request=httpx.Request("GET", SOURCE_URL) ) case _: raise AssertionError(f"Unexpected failure: {failure}") source_mentions_target.side_effect = exception - with pytest.raises( - type(exception), - match=str(exception), - ): + with pytest.raises(type(exception), match=str(exception)): receiver.verify_webmention.call_local(identifier) - assert get_webmention_state( - app, - identifier, - ) == ( - "failed", - str(exception), - ) + assert get_webmention_state(app, identifier) == ("failed", str(exception)) def test_verify_webmention_ignores_unknown_identifier( - receiver: ModuleType, - caplog: pytest.LogCaptureFixture, + receiver: ModuleType, caplog: pytest.LogCaptureFixture ) -> None: identifier = uuid.uuid7() @@ -741,3 +510,52 @@ def test_verify_webmention_ignores_unknown_identifier( receiver.verify_webmention.call_local(identifier) assert f"Cannot verify unknown ReceivedWebmention {identifier}" in caplog.text + + +@patch("webmentions_ssg.tasks.receiver.is_public_url", return_value=True) +def test_ensure_public_request_accepts_public_url( + is_public_url: Mock, receiver: ModuleType +) -> None: + receiver.ensure_public_request(httpx.Request("GET", SOURCE_URL)) + + is_public_url.assert_called_once_with(SOURCE_URL) + + +@patch("webmentions_ssg.tasks.receiver.is_public_url", return_value=False) +def test_ensure_public_request_rejects_non_public_address( + is_public_url: Mock, receiver: ModuleType +) -> None: + with pytest.raises( + receiver.VerificationError, match="Source resolves to a non-public address" + ): + receiver.ensure_public_request(httpx.Request("GET", SOURCE_URL)) + + is_public_url.assert_called_once_with(SOURCE_URL) + + +@patch( + "webmentions_ssg.tasks.receiver.is_public_url", + side_effect=AddressResolutionError("Could not resolve hostname"), +) +def test_ensure_public_request_maps_dns_failure_to_temporary_error( + is_public_url: Mock, receiver: ModuleType +) -> None: + with pytest.raises( + receiver.TemporaryFetchError, match="Source hostname could not be resolved" + ): + receiver.ensure_public_request(httpx.Request("GET", SOURCE_URL)) + + is_public_url.assert_called_once_with(SOURCE_URL) + + +@patch( + "webmentions_ssg.tasks.receiver.is_public_url", + side_effect=ValueError("No hostname was specified"), +) +def test_ensure_public_request_rejects_url_without_hostname( + is_public_url: Mock, receiver: ModuleType +) -> None: + with pytest.raises(receiver.VerificationError, match="Source URL has no hostname"): + receiver.ensure_public_request(httpx.Request("GET", SOURCE_URL)) + + is_public_url.assert_called_once_with(SOURCE_URL) diff --git a/tests/tasks/test_scanner.py b/tests/tasks/test_scanner.py new file mode 100644 index 0000000..65031b4 --- /dev/null +++ b/tests/tasks/test_scanner.py @@ -0,0 +1,1558 @@ +from pathlib import Path +from types import ModuleType +from uuid import UUID + +import pytest +import sqlalchemy as sa +from bs4 import BeautifulSoup +from flask import Flask +from pytest import MonkeyPatch + +from webmentions_ssg import DATABASE as db +from webmentions_ssg.models import SentWebmention, SentWebmentionStatus, Source + +BASE_URL = "https://dennisfink.me/blog/" +SOURCE_URL = f"{BASE_URL}example/" + + +@pytest.fixture +def scanner_module(app: Flask) -> ModuleType: + # Importing scanner registers Huey tasks. The dependency on the + # app fixture guarantees that Huey has been initialized first. + _ = app + + from webmentions_ssg.tasks import scanner + + return scanner + + +def run_scan(app: Flask, scanner_module: ModuleType) -> None: + with app.app_context(): + scanner_module.scan_sources() + + +def parse_html(html: str) -> BeautifulSoup: + return BeautifulSoup(html, "html5lib") + + +def write_post( + root: Path, + slug: str, + content: str, + *, + canonical: str | None = None, + entry_url: str | None = None, +) -> Path: + directory = root / slug + directory.mkdir(parents=True, exist_ok=True) + + source_url = f"{BASE_URL}{slug}/" + + if canonical is None: + canonical = source_url + + if entry_url is None: + entry_url = source_url + + path = directory / "index.html" + path.write_text( + f""" + <!doctype html> + <html> + <head> + <link + rel="canonical" + href="{canonical}" + > + </head> + <body> + <article class="h-entry"> + <a + class="u-url" + href="{entry_url}" + > + Permalink + </a> + + <div class="e-content"> + {content} + </div> + </article> + </body> + </html> + """, + encoding="utf-8", + ) + + return path + + +def configure_scanner(app: Flask, root: Path, *, base_url: str | None = None) -> None: + app.config["WEBMENTIONS_SSG_SOURCE_DIRECTORY"] = str(root) + app.config["WEBMENTIONS_SSG_SOURCE_BASE_URL"] = base_url + + +# --------------------------------------------------------------------------- +# Microformats parsing +# --------------------------------------------------------------------------- + + +def test_parse_entry_returns_microformats_entry(scanner_module: ModuleType) -> None: + document = parse_html( + f""" + <article class="h-entry"> + <a class="u-url" href="{SOURCE_URL}"> + Permalink + </a> + + <div class="e-content"> + Hello world + </div> + </article> + """ + ) + + element = document.find(class_="h-entry") + + assert element is not None + + entry = scanner_module.parse_entry(element, SOURCE_URL) + + assert "h-entry" in entry["type"] + assert entry["properties"]["url"] == [SOURCE_URL] + + +def test_parse_entry_rejects_non_entry(scanner_module: ModuleType) -> None: + document = parse_html( + """ + <article> + <p>Hello world</p> + </article> + """ + ) + + element = document.find("article") + + assert element is not None + + with pytest.raises( + scanner_module.SourceScanError, match="Expected exactly one parsed h-entry" + ): + scanner_module.parse_entry(element, SOURCE_URL) + + +def test_primary_entry_returns_source_entry(scanner_module: ModuleType) -> None: + document = parse_html( + f""" + <article class="h-entry" id="source"> + <a class="u-url" href="{SOURCE_URL}"> + Permalink + </a> + + <p class="p-name"> + Example post + </p> + + <div class="e-content"> + Hello world + </div> + </article> + """ + ) + + element, entry = scanner_module.primary_entry(document, SOURCE_URL) + + assert element["id"] == "source" + assert entry["properties"]["name"] == ["Example post"] + assert entry["properties"]["url"] == [SOURCE_URL] + + +def test_primary_entry_rejects_missing_h_entry(scanner_module: ModuleType) -> None: + document = parse_html( + """ + <main> + <p>No microformats here.</p> + </main> + """ + ) + + with pytest.raises( + scanner_module.SourceScanError, match="Expected exactly one h-entry, found 0" + ): + scanner_module.primary_entry(document, SOURCE_URL) + + +def test_primary_entry_rejects_multiple_h_entries(scanner_module: ModuleType) -> None: + document = parse_html( + f""" + <article class="h-entry"> + <a class="u-url" href="{SOURCE_URL}"> + First + </a> + </article> + + <article class="h-entry"> + <a + class="u-url" + href="https://example.com/other/" + > + Second + </a> + </article> + """ + ) + + with pytest.raises( + scanner_module.SourceScanError, match="Expected exactly one h-entry, found 2" + ): + scanner_module.primary_entry(document, SOURCE_URL) + + +def test_primary_entry_rejects_nested_h_entry(scanner_module: ModuleType) -> None: + document = parse_html( + f""" + <article class="h-entry"> + <a class="u-url" href="{SOURCE_URL}"> + Permalink + </a> + + <div class="e-content"> + <article class="h-entry"> + <a + class="u-url" + href="https://example.com/nested/" + > + Nested + </a> + </article> + </div> + </article> + """ + ) + + with pytest.raises( + scanner_module.SourceScanError, match="Expected exactly one h-entry, found 2" + ): + scanner_module.primary_entry(document, SOURCE_URL) + + +def test_primary_entry_rejects_missing_u_url(scanner_module: ModuleType) -> None: + document = parse_html( + """ + <article class="h-entry"> + <div class="e-content"> + Hello world + </div> + </article> + """ + ) + + with pytest.raises(scanner_module.SourceScanError, match="does not match"): + scanner_module.primary_entry(document, SOURCE_URL) + + +def test_primary_entry_rejects_nonmatching_u_url(scanner_module: ModuleType) -> None: + document = parse_html( + """ + <article class="h-entry"> + <a + class="u-url" + href="https://example.com/other/" + > + Other + </a> + + <div class="e-content"> + Hello world + </div> + </article> + """ + ) + + with pytest.raises(scanner_module.SourceScanError, match="does not match"): + scanner_module.primary_entry(document, SOURCE_URL) + + +# --------------------------------------------------------------------------- +# e-content +# --------------------------------------------------------------------------- + + +def test_content_element_returns_e_content(scanner_module: ModuleType) -> None: + document = parse_html( + """ + <article class="h-entry"> + <div class="e-content" id="content"> + Hello world + </div> + </article> + """ + ) + + entry = document.find(class_="h-entry") + + assert entry is not None + + content = scanner_module.content_element(entry) + + assert content["id"] == "content" + + +def test_content_element_rejects_missing_e_content(scanner_module: ModuleType) -> None: + document = parse_html( + """ + <article class="h-entry"> + <p>Hello world</p> + </article> + """ + ) + + entry = document.find(class_="h-entry") + + assert entry is not None + + with pytest.raises( + scanner_module.SourceScanError, match="Expected exactly one e-content, found 0" + ): + scanner_module.content_element(entry) + + +def test_content_element_rejects_multiple_e_content(scanner_module: ModuleType) -> None: + document = parse_html( + """ + <article class="h-entry"> + <div class="e-content"> + First + </div> + + <div class="e-content"> + Second + </div> + </article> + """ + ) + + entry = document.find(class_="h-entry") + + assert entry is not None + + with pytest.raises( + scanner_module.SourceScanError, match="Expected exactly one e-content, found 2" + ): + scanner_module.content_element(entry) + + +# --------------------------------------------------------------------------- +# Canonical URLs +# --------------------------------------------------------------------------- + + +def test_canonical_url_returns_absolute_url(scanner_module: ModuleType) -> None: + document = parse_html( + f""" + <html> + <head> + <link + rel="canonical" + href="{SOURCE_URL}" + > + </head> + </html> + """ + ) + + result = scanner_module.canonical_url( + document, relative_path=Path("example/index.html"), base_url=None + ) + + assert result == SOURCE_URL + + +def test_canonical_url_returns_none_when_missing(scanner_module: ModuleType) -> None: + document = parse_html( + """ + <html> + <head> + <title>Example</title> + </head> + </html> + """ + ) + + result = scanner_module.canonical_url( + document, relative_path=Path("example/index.html"), base_url=None + ) + + assert result is None + + +def test_canonical_url_resolves_relative_url_with_base_url( + scanner_module: ModuleType, +) -> None: + document = parse_html( + """ + <html> + <head> + <link + rel="canonical" + href="./canonical/" + > + </head> + </html> + """ + ) + + result = scanner_module.canonical_url( + document, relative_path=Path("example/index.html"), base_url=BASE_URL + ) + + assert result == ("https://dennisfink.me/blog/example/canonical/") + + +def test_relative_canonical_requires_base_url(scanner_module: ModuleType) -> None: + document = parse_html( + """ + <html> + <head> + <link + rel="canonical" + href="./canonical/" + > + </head> + </html> + """ + ) + + with pytest.raises(scanner_module.SourceScanError, match="relative canonical"): + scanner_module.canonical_url( + document, relative_path=Path("example/index.html"), base_url=None + ) + + +def test_empty_canonical_is_rejected(scanner_module: ModuleType) -> None: + document = parse_html( + """ + <html> + <head> + <link + rel="canonical" + href=" " + > + </head> + </html> + """ + ) + + with pytest.raises(scanner_module.SourceScanError, match="empty canonical URL"): + scanner_module.canonical_url( + document, relative_path=Path("example/index.html"), base_url=BASE_URL + ) + + +# --------------------------------------------------------------------------- +# Target extraction +# --------------------------------------------------------------------------- + + +def test_scan_source_uses_canonical_without_base_url( + scanner_module: ModuleType, tmp_path: Path +) -> None: + path = write_post( + tmp_path, + "example", + """ + <a href="https://example.com/target"> + Target + </a> + """, + ) + + scanned = scanner_module.scan_source_file(path, root=tmp_path, base_url=None) + + assert scanned.path == "example/index.html" + assert scanned.url == SOURCE_URL + assert scanned.targets == frozenset({"https://example.com/target"}) + assert len(scanned.content_hash) == 32 + + +def test_scan_source_falls_back_to_base_url( + scanner_module: ModuleType, tmp_path: Path +) -> None: + directory = tmp_path / "example" + directory.mkdir() + + path = directory / "index.html" + path.write_text( + """ + <!doctype html> + <html> + <body> + <article class="h-entry"> + <a class="u-url" href="./"> + Permalink + </a> + + <div class="e-content"> + <a href="https://example.com/target"> + Target + </a> + </div> + </article> + </body> + </html> + """, + encoding="utf-8", + ) + + scanned = scanner_module.scan_source_file(path, root=tmp_path, base_url=BASE_URL) + + assert scanned.url == SOURCE_URL + assert scanned.targets == frozenset({"https://example.com/target"}) + + +def test_scan_source_requires_canonical_or_base_url( + scanner_module: ModuleType, tmp_path: Path +) -> None: + directory = tmp_path / "example" + directory.mkdir() + + path = directory / "index.html" + path.write_text( + """ + <!doctype html> + <html> + <body> + <article class="h-entry"> + <a class="u-url" href="./"> + Permalink + </a> + + <div class="e-content"> + <p>Hello world.</p> + </div> + </article> + </body> + </html> + """, + encoding="utf-8", + ) + + with pytest.raises(scanner_module.SourceScanError, match="no canonical URL"): + scanner_module.scan_source_file(path, root=tmp_path, base_url=None) + + +def test_scan_source_extracts_only_content_links( + scanner_module: ModuleType, tmp_path: Path +) -> None: + directory = tmp_path / "example" + directory.mkdir() + + path = directory / "index.html" + path.write_text( + f""" + <!doctype html> + <html> + <head> + <link + rel="canonical" + href="{SOURCE_URL}" + > + </head> + <body> + <article class="h-entry"> + <a class="u-url" href="{SOURCE_URL}"> + Permalink + </a> + + <a href="https://example.com/metadata"> + Metadata link + </a> + + <div class="e-content"> + <a href="https://example.com/content"> + Content link + </a> + </div> + </article> + </body> + </html> + """, + encoding="utf-8", + ) + + scanned = scanner_module.scan_source_file(path, root=tmp_path, base_url=None) + + assert scanned.targets == frozenset({"https://example.com/content"}) + + +def test_scan_source_extracts_relative_content_link( + scanner_module: ModuleType, tmp_path: Path +) -> None: + path = write_post( + tmp_path, + "example", + """ + <a href="../other/"> + Other post + </a> + """, + ) + + scanned = scanner_module.scan_source_file(path, root=tmp_path, base_url=None) + + assert scanned.targets == frozenset({"https://dennisfink.me/blog/other/"}) + + +@pytest.mark.parametrize( + "property_name", ("in-reply-to", "like-of", "repost-of", "bookmark-of") +) +def test_scan_source_extracts_reaction_properties( + scanner_module: ModuleType, tmp_path: Path, property_name: str +) -> None: + directory = tmp_path / "example" + directory.mkdir() + + target = f"https://example.com/{property_name}" + + path = directory / "index.html" + path.write_text( + f""" + <!doctype html> + <html> + <head> + <link + rel="canonical" + href="{SOURCE_URL}" + > + </head> + <body> + <article class="h-entry"> + <a class="u-url" href="{SOURCE_URL}"> + Permalink + </a> + + <a + class="u-{property_name}" + href="{target}" + > + Reaction target + </a> + + <div class="e-content"> + <p>Post content.</p> + </div> + </article> + </body> + </html> + """, + encoding="utf-8", + ) + + scanned = scanner_module.scan_source_file(path, root=tmp_path, base_url=None) + + assert scanned.targets == frozenset({target}) + + +def test_scan_source_extracts_anchor_from_e_content( + scanner_module: ModuleType, tmp_path: Path +) -> None: + path = write_post( + tmp_path, + "example", + """ + <a href="https://example.com/target"> + Target + </a> + """, + ) + + scanned = scanner_module.scan_source_file(path, root=tmp_path, base_url=None) + + assert scanned.targets == frozenset({"https://example.com/target"}) + + +@pytest.mark.parametrize("tag", ("link", "area")) +def test_scan_source_ignores_non_anchor_href_elements( + scanner_module: ModuleType, tmp_path: Path, tag: str +) -> None: + path = write_post( + tmp_path, + "example", + f""" + <{tag} href="https://example.com/target"> + """, + ) + + scanned = scanner_module.scan_source_file(path, root=tmp_path, base_url=None) + + assert scanned.targets == frozenset() + + +def test_scan_source_ignores_self_fragment( + scanner_module: ModuleType, tmp_path: Path +) -> None: + path = write_post( + tmp_path, + "example", + """ + <a href="#section"> + Section + </a> + + <a href="https://example.com/target"> + Target + </a> + """, + ) + + scanned = scanner_module.scan_source_file(path, root=tmp_path, base_url=None) + + assert scanned.targets == frozenset({"https://example.com/target"}) + + +# --------------------------------------------------------------------------- +# Database reconciliation +# --------------------------------------------------------------------------- + + +def test_scan_creates_source_and_webmentions( + app: Flask, scanner_module: ModuleType, tmp_path: Path, monkeypatch: MonkeyPatch +) -> None: + configure_scanner(app, tmp_path) + + write_post( + tmp_path, + "example", + """ + <a href="https://example.com/a"> + A + </a> + + <a href="https://example.com/b"> + B + </a> + """, + ) + + queued: list[UUID] = [] + + monkeypatch.setattr(scanner_module, "send_webmention", queued.append) + + run_scan(app, scanner_module) + + with app.app_context(): + source = db.session.scalar(sa.select(Source)) + + assert source is not None + assert source.path == "example/index.html" + assert source.url == SOURCE_URL + assert source.revision == 1 + assert source.deleted_at is None + assert len(source.content_hash) == 32 + + webmentions = list( + db.session.scalars( + sa.select(SentWebmention).order_by(SentWebmention.target) + ) + ) + + assert [webmention.target for webmention in webmentions] == [ + "https://example.com/a", + "https://example.com/b", + ] + + assert all(webmention.active for webmention in webmentions) + assert all(webmention.desired_revision == 1 for webmention in webmentions) + assert all(webmention.processed_revision is None for webmention in webmentions) + + identifiers = {webmention.uuid for webmention in webmentions} + + assert set(queued) == identifiers + + +def test_unchanged_source_does_not_increment_revision( + app: Flask, scanner_module: ModuleType, tmp_path: Path, monkeypatch: MonkeyPatch +) -> None: + configure_scanner(app, tmp_path) + + write_post( + tmp_path, + "example", + """ + <a href="https://example.com/a"> + A + </a> + """, + ) + + queued: list[UUID] = [] + + monkeypatch.setattr(scanner_module, "send_webmention", queued.append) + + run_scan(app, scanner_module) + + with app.app_context(): + webmention = db.session.scalar(sa.select(SentWebmention)) + + assert webmention is not None + + webmention.processed_revision = 1 + webmention.sent_revision = 1 + webmention.status = SentWebmentionStatus.SENT + + db.session.commit() + + queued.clear() + + run_scan(app, scanner_module) + + with app.app_context(): + source = db.session.scalar(sa.select(Source)) + + assert source is not None + assert source.revision == 1 + + assert queued == [] + + +def test_updated_source_increments_revision( + app: Flask, scanner_module: ModuleType, tmp_path: Path, monkeypatch: MonkeyPatch +) -> None: + configure_scanner(app, tmp_path) + + path = write_post( + tmp_path, + "example", + """ + <p>Original content</p> + + <a href="https://example.com/a"> + A + </a> + """, + ) + + queued: list[UUID] = [] + + monkeypatch.setattr(scanner_module, "send_webmention", queued.append) + + run_scan(app, scanner_module) + + with app.app_context(): + webmention = db.session.scalar(sa.select(SentWebmention)) + + assert webmention is not None + + webmention.processed_revision = 1 + webmention.sent_revision = 1 + webmention.status = SentWebmentionStatus.SENT + + db.session.commit() + + queued.clear() + + path.write_text( + f""" + <!doctype html> + <html> + <head> + <link + rel="canonical" + href="{SOURCE_URL}" + > + </head> + <body> + <article class="h-entry"> + <a class="u-url" href="{SOURCE_URL}"> + Permalink + </a> + + <div class="e-content"> + <p>Updated content</p> + + <a href="https://example.com/a"> + A + </a> + </div> + </article> + </body> + </html> + """, + encoding="utf-8", + ) + + run_scan(app, scanner_module) + + with app.app_context(): + source = db.session.scalar(sa.select(Source)) + webmention = db.session.scalar(sa.select(SentWebmention)) + + assert source is not None + assert webmention is not None + + assert source.revision == 2 + + assert webmention.active + assert webmention.desired_revision == 2 + assert webmention.processed_revision == 1 + assert webmention.sent_revision == 1 + assert webmention.pending + + identifier = webmention.uuid + + assert queued == [identifier] + + +def test_new_target_is_added_on_update( + app: Flask, scanner_module: ModuleType, tmp_path: Path, monkeypatch: MonkeyPatch +) -> None: + configure_scanner(app, tmp_path) + + path = write_post( + tmp_path, + "example", + """ + <p>No links yet.</p> + """, + ) + + queued: list[UUID] = [] + + monkeypatch.setattr(scanner_module, "send_webmention", queued.append) + + run_scan(app, scanner_module) + + assert queued == [] + + path.write_text( + f""" + <!doctype html> + <html> + <head> + <link + rel="canonical" + href="{SOURCE_URL}" + > + </head> + <body> + <article class="h-entry"> + <a class="u-url" href="{SOURCE_URL}"> + Permalink + </a> + + <div class="e-content"> + <a href="https://example.com/new"> + New + </a> + </div> + </article> + </body> + </html> + """, + encoding="utf-8", + ) + + run_scan(app, scanner_module) + + with app.app_context(): + source = db.session.scalar(sa.select(Source)) + webmention = db.session.scalar(sa.select(SentWebmention)) + + assert source is not None + assert webmention is not None + + assert source.revision == 2 + + assert webmention.target == ("https://example.com/new") + assert webmention.active + assert webmention.desired_revision == 2 + assert webmention.processed_revision is None + assert webmention.sent_revision is None + assert webmention.pending + + identifier = webmention.uuid + + assert queued == [identifier] + + +def test_removed_sent_target_is_queued_again( + app: Flask, scanner_module: ModuleType, tmp_path: Path, monkeypatch: MonkeyPatch +) -> None: + configure_scanner(app, tmp_path) + + path = write_post( + tmp_path, + "example", + """ + <a href="https://example.com/sent"> + Sent + </a> + """, + ) + + queued: list[UUID] = [] + + monkeypatch.setattr(scanner_module, "send_webmention", queued.append) + + run_scan(app, scanner_module) + + with app.app_context(): + webmention = db.session.scalar(sa.select(SentWebmention)) + + assert webmention is not None + + webmention.processed_revision = 1 + webmention.sent_revision = 1 + webmention.status = SentWebmentionStatus.SENT + + identifier = webmention.uuid + db.session.commit() + + queued.clear() + + path.write_text( + f""" + <!doctype html> + <html> + <head> + <link + rel="canonical" + href="{SOURCE_URL}" + > + </head> + <body> + <article class="h-entry"> + <a class="u-url" href="{SOURCE_URL}"> + Permalink + </a> + + <div class="e-content"> + <p>The link is gone.</p> + </div> + </article> + </body> + </html> + """, + encoding="utf-8", + ) + + run_scan(app, scanner_module) + + with app.app_context(): + source = db.session.scalar(sa.select(Source)) + webmention = db.session.scalar(sa.select(SentWebmention)) + + assert source is not None + assert webmention is not None + + assert source.revision == 2 + + assert not webmention.active + assert webmention.desired_revision == 2 + assert webmention.processed_revision == 1 + assert webmention.sent_revision == 1 + assert webmention.pending + + assert queued == [identifier] + + +def test_removed_unsent_target_is_not_queued( + app: Flask, scanner_module: ModuleType, tmp_path: Path, monkeypatch: MonkeyPatch +) -> None: + configure_scanner(app, tmp_path) + + path = write_post( + tmp_path, + "example", + """ + <a href="https://example.com/unsent"> + Unsent + </a> + """, + ) + + queued: list[UUID] = [] + + monkeypatch.setattr(scanner_module, "send_webmention", queued.append) + + run_scan(app, scanner_module) + + queued.clear() + + path.write_text( + f""" + <!doctype html> + <html> + <head> + <link + rel="canonical" + href="{SOURCE_URL}" + > + </head> + <body> + <article class="h-entry"> + <a class="u-url" href="{SOURCE_URL}"> + Permalink + </a> + + <div class="e-content"> + <p>The link is gone.</p> + </div> + </article> + </body> + </html> + """, + encoding="utf-8", + ) + + run_scan(app, scanner_module) + + with app.app_context(): + source = db.session.scalar(sa.select(Source)) + webmention = db.session.scalar(sa.select(SentWebmention)) + + assert source is not None + assert webmention is not None + + assert source.revision == 2 + + assert not webmention.active + assert webmention.desired_revision == 2 + assert webmention.processed_revision == 2 + assert webmention.sent_revision is None + assert not webmention.pending + + assert queued == [] + + +# --------------------------------------------------------------------------- +# Source deletion/restoration +# --------------------------------------------------------------------------- + + +def test_deleted_source_queues_previously_sent_webmention( + app: Flask, scanner_module: ModuleType, tmp_path: Path, monkeypatch: MonkeyPatch +) -> None: + configure_scanner(app, tmp_path) + + path = write_post( + tmp_path, + "example", + """ + <a href="https://example.com/a"> + A + </a> + """, + ) + + queued: list[UUID] = [] + + monkeypatch.setattr(scanner_module, "send_webmention", queued.append) + + run_scan(app, scanner_module) + + with app.app_context(): + webmention = db.session.scalar(sa.select(SentWebmention)) + + assert webmention is not None + + webmention.processed_revision = 1 + webmention.sent_revision = 1 + webmention.status = SentWebmentionStatus.SENT + + identifier = webmention.uuid + db.session.commit() + + queued.clear() + path.unlink() + + run_scan(app, scanner_module) + + with app.app_context(): + source = db.session.scalar(sa.select(Source)) + webmention = db.session.scalar(sa.select(SentWebmention)) + + assert source is not None + assert webmention is not None + + assert source.revision == 2 + assert source.deleted_at is not None + + assert not webmention.active + assert webmention.desired_revision == 2 + assert webmention.processed_revision == 1 + assert webmention.sent_revision == 1 + assert webmention.pending + + assert queued == [identifier] + + +def test_deleted_source_does_not_queue_unsent_webmention( + app: Flask, scanner_module: ModuleType, tmp_path: Path, monkeypatch: MonkeyPatch +) -> None: + configure_scanner(app, tmp_path) + + path = write_post( + tmp_path, + "example", + """ + <a href="https://example.com/a"> + A + </a> + """, + ) + + queued: list[UUID] = [] + + monkeypatch.setattr(scanner_module, "send_webmention", queued.append) + + run_scan(app, scanner_module) + + queued.clear() + path.unlink() + + run_scan(app, scanner_module) + + with app.app_context(): + source = db.session.scalar(sa.select(Source)) + webmention = db.session.scalar(sa.select(SentWebmention)) + + assert source is not None + assert webmention is not None + + assert source.revision == 2 + assert source.deleted_at is not None + + assert not webmention.active + assert webmention.desired_revision == 2 + assert webmention.processed_revision == 2 + assert webmention.sent_revision is None + assert not webmention.pending + + assert queued == [] + + +def test_restored_source_creates_new_revision( + app: Flask, scanner_module: ModuleType, tmp_path: Path, monkeypatch: MonkeyPatch +) -> None: + configure_scanner(app, tmp_path) + + path = write_post( + tmp_path, + "example", + """ + <a href="https://example.com/a"> + A + </a> + """, + ) + + queued: list[UUID] = [] + + monkeypatch.setattr(scanner_module, "send_webmention", queued.append) + + run_scan(app, scanner_module) + + with app.app_context(): + webmention = db.session.scalar(sa.select(SentWebmention)) + + assert webmention is not None + + webmention.processed_revision = 1 + webmention.sent_revision = 1 + webmention.status = SentWebmentionStatus.SENT + + db.session.commit() + + path.unlink() + run_scan(app, scanner_module) + + queued.clear() + + write_post( + tmp_path, + "example", + """ + <a href="https://example.com/a"> + A + </a> + """, + ) + + run_scan(app, scanner_module) + + with app.app_context(): + source = db.session.scalar(sa.select(Source)) + webmention = db.session.scalar(sa.select(SentWebmention)) + + assert source is not None + assert webmention is not None + + assert source.revision == 3 + assert source.deleted_at is None + + assert webmention.active + assert webmention.desired_revision == 3 + assert webmention.sent_revision == 1 + assert webmention.pending + + identifier = webmention.uuid + + assert queued == [identifier] + + +# --------------------------------------------------------------------------- +# Queue recovery and invalid sources +# --------------------------------------------------------------------------- + + +def test_pending_webmention_is_requeued_on_next_scan( + app: Flask, scanner_module: ModuleType, tmp_path: Path, monkeypatch: MonkeyPatch +) -> None: + configure_scanner(app, tmp_path) + + write_post( + tmp_path, + "example", + """ + <a href="https://example.com/a"> + A + </a> + """, + ) + + queued: list[UUID] = [] + + monkeypatch.setattr(scanner_module, "send_webmention", queued.append) + + run_scan(app, scanner_module) + + assert len(queued) == 1 + + identifier = queued[0] + queued.clear() + + run_scan(app, scanner_module) + + assert queued == [identifier] + + +def test_invalid_known_source_is_not_deleted( + app: Flask, + scanner_module: ModuleType, + tmp_path: Path, + monkeypatch: MonkeyPatch, + caplog: pytest.LogCaptureFixture, +) -> None: + configure_scanner(app, tmp_path) + + path = write_post( + tmp_path, + "example", + """ + <p>Hello world.</p> + """, + ) + + monkeypatch.setattr(scanner_module, "send_webmention", lambda identifier: None) + + run_scan(app, scanner_module) + + path.write_text( + f""" + <!doctype html> + <html> + <head> + <link + rel="canonical" + href="{SOURCE_URL}" + > + </head> + <body> + <p>The h-entry disappeared.</p> + </body> + </html> + """, + encoding="utf-8", + ) + + run_scan(app, scanner_module) + + with app.app_context(): + source = db.session.scalar(sa.select(Source)) + + assert source is not None + assert source.revision == 1 + assert source.deleted_at is None + + assert "Could not scan source" in caplog.text + assert "Expected exactly one h-entry, found 0" in caplog.text + + +def test_scan_sources_accepts_valid_base_url( + app: Flask, scanner_module: ModuleType, tmp_path: Path, monkeypatch: MonkeyPatch +) -> None: + configure_scanner(app, tmp_path, base_url=BASE_URL) + + directory = tmp_path / "example" + directory.mkdir() + + path = directory / "index.html" + path.write_text( + """ + <!doctype html> + <html> + <body> + <article class="h-entry"> + <a class="u-url" href="./"> + Permalink + </a> + + <div class="e-content"> + <p>Hello world.</p> + </div> + </article> + </body> + </html> + """, + encoding="utf-8", + ) + + monkeypatch.setattr(scanner_module, "send_webmention", lambda identifier: None) + + 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 + + +def test_scan_sources_rejects_invalid_base_url( + app: Flask, scanner_module: ModuleType, tmp_path: Path +) -> None: + configure_scanner(app, tmp_path, base_url="not-a-url") + + with pytest.raises(RuntimeError, match="absolute HTTP or HTTPS URL"): + run_scan(app, scanner_module) + + +@pytest.mark.parametrize( + "href", + ( + "https://dennisfink.me/blog/other/", + "https://www.dennisfink.me/blog/other/", + "https://webmentions.dennisfink.me/endpoint", + "https://foo.bar.dennisfink.me/example", + ), +) +def test_scan_source_ignores_matching_hostname( + scanner_module: ModuleType, tmp_path: Path, href: str +) -> None: + path = write_post( + tmp_path, + "example", + f""" + <a href="{href}"> + Ignored + </a> + + <a href="https://example.com/target"> + External + </a> + """, + ) + + scanned = scanner_module.scan_source_file( + path, + root=tmp_path, + base_url=None, + ignored_hostnames=("dennisfink.me", "*.dennisfink.me"), + ) + + assert scanned.targets == frozenset({"https://example.com/target"}) + + +@pytest.mark.parametrize( + "href", + ( + "https://notdennisfink.me/example", + "https://dennisfink.me.example.com/example", + "https://example.com/example", + ), +) +def test_scan_source_keeps_nonmatching_hostname( + scanner_module: ModuleType, tmp_path: Path, href: str +) -> None: + path = write_post( + tmp_path, + "example", + f""" + <a href="{href}"> + Target + </a> + """, + ) + + scanned = scanner_module.scan_source_file( + path, + root=tmp_path, + base_url=None, + ignored_hostnames=("dennisfink.me", "*.dennisfink.me"), + ) + + assert scanned.targets == frozenset({href}) + + +def test_scan_source_ignores_reaction_to_matching_hostname( + scanner_module: ModuleType, tmp_path: Path +) -> None: + directory = tmp_path / "example" + directory.mkdir() + + path = directory / "index.html" + path.write_text( + f""" + <!doctype html> + <html> + <head> + <link + rel="canonical" + href="{SOURCE_URL}" + > + </head> + <body> + <article class="h-entry"> + <a + class="u-url" + href="{SOURCE_URL}" + > + Permalink + </a> + + <a + class="u-in-reply-to" + href="https://www.dennisfink.me/blog/other/" + > + Reply target + </a> + + <div class="e-content"> + <p>Reply content.</p> + </div> + </article> + </body> + </html> + """, + encoding="utf-8", + ) + + scanned = scanner_module.scan_source_file( + path, + root=tmp_path, + base_url=None, + ignored_hostnames=("dennisfink.me", "*.dennisfink.me"), + ) + + assert scanned.targets == frozenset() diff --git a/tests/tasks/test_sender.py b/tests/tasks/test_sender.py new file mode 100644 index 0000000..86799f7 --- /dev/null +++ b/tests/tasks/test_sender.py @@ -0,0 +1,788 @@ +import logging +import uuid +from datetime import datetime, timezone +from types import ModuleType +from unittest.mock import MagicMock, Mock + +import httpx +import pytest +from flask import Flask + +from webmentions_ssg import DATABASE as db +from webmentions_ssg.models import SentWebmention, SentWebmentionStatus, Source + +SOURCE_URL = "https://dennisfink.me/blog/example/" +TARGET_URL = "https://example.com/post" +ENDPOINT_URL = "https://example.com/webmention" +STATUS_URL = "https://example.com/webmention/status/123" + + +def create_sent_webmention( + app: Flask, + *, + active: bool = True, + desired_revision: int = 1, + processed_revision: int | None = None, + sent_revision: int | None = None, + status: SentWebmentionStatus | None = None, + failure_reason: str | None = None, + endpoint: str | None = None, + response_status: int | None = None, + status_url: str | None = None, +) -> uuid.UUID: + now = datetime.now(timezone.utc) + + with app.app_context(): + source = Source( + path="example/index.html", + url=SOURCE_URL, + content_hash="0" * 32, + revision=desired_revision, + last_seen_at=now, + revised_at=now, + ) + + webmention = SentWebmention( + target=TARGET_URL, + active=active, + desired_revision=desired_revision, + processed_revision=processed_revision, + sent_revision=sent_revision, + status=status, + failure_reason=failure_reason, + endpoint=endpoint, + response_status=response_status, + status_url=status_url, + ) + + source.sent_webmentions.append(webmention) + + db.session.add(source) + db.session.commit() + + return webmention.uuid + + +def mock_sender_requests( + sender_module: ModuleType, + monkeypatch: pytest.MonkeyPatch, + *, + endpoint: str | None = ENDPOINT_URL, + result: tuple[int, str | None] = (202, None), +) -> tuple[Mock, Mock, Mock, Mock]: + httpx_client = MagicMock() + client = httpx_client.return_value.__enter__.return_value + + discover = Mock(return_value=endpoint) + post = Mock(return_value=result) + + monkeypatch.setattr(sender_module.httpx, "Client", httpx_client) + monkeypatch.setattr(sender_module, "discover_webmention_endpoint", discover) + monkeypatch.setattr(sender_module, "post_webmention", post) + + return httpx_client, client, discover, post + + +@pytest.fixture +def sender_module(app: Flask) -> ModuleType: + # Importing sender registers Huey tasks. The dependency on the + # app fixture guarantees that Huey has been initialized first. + _ = app + + from webmentions_ssg.tasks import sender + + return sender + + +@pytest.mark.parametrize( + ("value", "expected"), + ( + ( + '<https://example.com/webmention>; rel="webmention"', + ("https://example.com/webmention", {"webmention"}), + ), + ( + "<https://example.com/webmention>; rel=webmention", + ("https://example.com/webmention", {"webmention"}), + ), + ( + '<https://example.com/webmention>; rel="webmention alternate"', + ("https://example.com/webmention", {"webmention", "alternate"}), + ), + ( + '<https://example.com/webmention>; REL="WebMention Alternate"', + ("https://example.com/webmention", {"webmention", "alternate"}), + ), + ( + "<https://example.com/webmention>; " + 'type="text/html"; rel="webmention"; title="Endpoint"', + ("https://example.com/webmention", {"webmention"}), + ), + ( + '<https://example.com/stylesheet>; type="text/css"', + ("https://example.com/stylesheet", set()), + ), + ), +) +def test_parse_link_value( + sender_module: ModuleType, value: str, expected: tuple[str, set[str]] +) -> None: + assert sender_module.parse_link_value(value) == expected + + +@pytest.mark.parametrize( + "value", ("", "https://example.com/webmention", "<https://example.com/webmention") +) +def test_parse_link_value_rejects_malformed_value( + sender_module: ModuleType, value: str +) -> None: + assert sender_module.parse_link_value(value) is None + + +@pytest.mark.parametrize( + ("target", "headers", "html", "expected_endpoint"), + ( + pytest.param( + "https://example.com/test/1", + (("Link", "</test/1/webmention?head=true>; rel=webmention"),), + "", + "https://example.com/test/1/webmention?head=true", + id="1-http-link-unquoted-rel-relative-url", + ), + pytest.param( + "https://example.com/test/2", + ( + ( + "Link", + "<https://example.com/test/2/webmention?head=true>; rel=webmention", + ), + ), + "", + "https://example.com/test/2/webmention?head=true", + id="2-http-link-unquoted-rel-absolute-url", + ), + pytest.param( + "https://example.com/test/3", + (), + """ + <link + rel="webmention" + href="/test/3/webmention" + > + """, + "https://example.com/test/3/webmention", + id="3-html-link-relative-url", + ), + pytest.param( + "https://example.com/test/4", + (), + """ + <link + rel="webmention" + href="https://example.com/test/4/webmention" + > + """, + "https://example.com/test/4/webmention", + id="4-html-link-absolute-url", + ), + pytest.param( + "https://example.com/test/5", + (), + """ + <a + rel="webmention" + href="/test/5/webmention" + > + Endpoint + </a> + """, + "https://example.com/test/5/webmention", + id="5-html-a-relative-url", + ), + pytest.param( + "https://example.com/test/6", + (), + """ + <a + rel="webmention" + href="https://example.com/test/6/webmention" + > + Endpoint + </a> + """, + "https://example.com/test/6/webmention", + id="6-html-a-absolute-url", + ), + pytest.param( + "https://example.com/test/7", + ( + ( + "LinK", + "<https://example.com/test/7/webmention?head=true>; rel=webmention", + ), + ), + "", + "https://example.com/test/7/webmention?head=true", + id="7-http-link-strange-casing", + ), + pytest.param( + "https://example.com/test/8", + ( + ( + "Link", + '<https://example.com/test/8/webmention?head=true>; rel="webmention"', + ), + ), + "", + "https://example.com/test/8/webmention?head=true", + id="8-http-link-quoted-rel", + ), + pytest.param( + "https://example.com/test/9", + (), + """ + <link + rel="webmention somethingelse" + href="https://example.com/test/9/webmention" + > + """, + "https://example.com/test/9/webmention", + id="9-multiple-html-rel-values", + ), + pytest.param( + "https://example.com/test/10", + ( + ( + "Link", + "<https://example.com/test/10/webmention?head=true>; " + 'rel="webmention somethingelse"', + ), + ), + "", + "https://example.com/test/10/webmention?head=true", + id="10-multiple-link-header-rel-values", + ), + pytest.param( + "https://example.com/test/11", + (("Link", '</test/11/webmention>; rel="webmention"'),), + """ + <link + rel="webmention" + href="/test/11/webmention/error" + > + + <a + rel="webmention" + href="/test/11/webmention/error" + > + Wrong endpoint + </a> + """, + "https://example.com/test/11/webmention", + id="11-http-link-precedence", + ), + pytest.param( + "https://example.com/test/12", + (), + """ + <link + rel="not-webmention" + href="/test/12/webmention/error" + > + + <a + rel="webmention" + href="/test/12/webmention" + > + Correct endpoint + </a> + """, + "https://example.com/test/12/webmention", + id="12-exact-rel-match", + ), + pytest.param( + "https://example.com/test/13", + (), + """ + <!-- + <a + rel="webmention" + href="/test/13/webmention/error" + > + False endpoint + </a> + --> + + <a + rel="webmention" + href="/test/13/webmention" + > + Correct endpoint + </a> + """, + "https://example.com/test/13/webmention", + id="13-endpoint-inside-html-comment", + ), + pytest.param( + "https://example.com/test/14", + (), + """ + <code> + <a + rel="webmention" + href="/test/14/webmention/error" + > + False endpoint + </a> + </code> + + <a + rel="webmention" + href="/test/14/webmention" + > + Correct endpoint + </a> + """, + "https://example.com/test/14/webmention", + id="14-endpoint-in-escaped-html", + ), + pytest.param( + "https://example.com/test/15", + (), + """ + <link + rel="webmention" + href="" + > + """, + "https://example.com/test/15", + id="15-empty-href", + ), + pytest.param( + "https://example.com/test/16", + (), + """ + <a + rel="webmention" + href="/test/16/webmention" + > + First endpoint + </a> + + <link + rel="webmention" + href="/test/16/webmention/error" + > + """, + "https://example.com/test/16/webmention", + id="16-a-before-link", + ), + pytest.param( + "https://example.com/test/17", + (), + """ + <link + rel="webmention" + href="/test/17/webmention" + > + + <a + rel="webmention" + href="/test/17/webmention/error" + > + Second endpoint + </a> + """, + "https://example.com/test/17/webmention", + id="17-link-before-a", + ), + pytest.param( + "https://example.com/test/18", + ( + ("Link", '<https://example.com/test/18/webmention/error>; rel="other"'), + ( + "Link", + "<https://example.com/test/18/webmention?head=true>; " + 'rel="webmention"', + ), + ), + "", + "https://example.com/test/18/webmention?head=true", + id="18-multiple-http-link-headers", + ), + pytest.param( + "https://example.com/test/19", + ( + ( + "Link", + "<https://example.com/test/19/webmention/error>; " + 'rel="other", ' + "<https://example.com/test/19/webmention?head=true>; " + 'rel="webmention"', + ), + ), + "", + "https://example.com/test/19/webmention?head=true", + id="19-single-header-multiple-values", + ), + pytest.param( + "https://example.com/test/20", + (), + """ + <link rel="webmention"> + + <a + rel="webmention" + href="/test/20/webmention" + > + Correct endpoint + </a> + """, + "https://example.com/test/20/webmention", + id="20-link-without-href", + ), + pytest.param( + "https://example.com/test/21", + (), + """ + <link + rel="webmention" + href="/test/21/webmention?query=yes" + > + """, + "https://example.com/test/21/webmention?query=yes", + id="21-endpoint-query-string", + ), + pytest.param( + "https://example.com/test/22", + (), + """ + <link + rel="webmention" + href="22/webmention" + > + """, + "https://example.com/test/22/webmention", + id="22-endpoint-relative-to-path", + ), + ), +) +def test_discovery( + app: Flask, + sender_module: ModuleType, + target: str, + headers: tuple[tuple[str, str], ...], + html: str, + expected_endpoint: str, +) -> None: + def handler(request: httpx.Request) -> httpx.Response: + if request.method == "HEAD": + return httpx.Response(200, headers=headers) + + return httpx.Response(200, headers={"Content-Type": "text/html"}, text=html) + + with ( + app.app_context(), + httpx.Client( + transport=httpx.MockTransport(handler), follow_redirects=True + ) as client, + ): + endpoint = sender_module.discover_webmention_endpoint(client, target) + + assert endpoint == expected_endpoint + + +@pytest.mark.parametrize( + ("target", "responses", "expected_endpoint"), + ( + pytest.param( + "https://example.com/test/23/page", + { + "/test/23/page": (302, {"Location": "page/redirect-key"}), + "/test/23/page/redirect-key": ( + 200, + {"Link": ("<webmention-endpoint/endpoint-key>; rel=webmention")}, + ), + }, + ("https://example.com/test/23/page/webmention-endpoint/endpoint-key"), + id="relative-endpoint-after-redirect", + ), + ), +) +def test_discovery_redirect( + app: Flask, + sender_module: ModuleType, + target: str, + responses: dict[str, tuple[int, dict[str, str]]], + expected_endpoint: str, +) -> None: + def handler(request: httpx.Request) -> httpx.Response: + try: + status, headers = responses[request.url.path] + except KeyError: + raise AssertionError(f"Unexpected request: {request.url}") from None + + return httpx.Response(status, headers=headers) + + with ( + app.app_context(), + httpx.Client( + transport=httpx.MockTransport(handler), follow_redirects=True + ) as client, + ): + endpoint = sender_module.discover_webmention_endpoint(client, target) + + assert endpoint == expected_endpoint + + +def test_discovery_stops_after_head(app: Flask, sender_module: ModuleType) -> None: + requests: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + requests.append(request) + + 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, "https://example.com/post" + ) + + assert endpoint == "https://example.com/webmention" + assert [request.method for request in requests] == ["HEAD"] + + +def test_discovery_falls_back_to_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={"Content-Type": "text/html"}, + text=""" + <link + rel="webmention" + href="/webmention" + > + """, + ) + + with ( + app.app_context(), + httpx.Client( + transport=httpx.MockTransport(handler), follow_redirects=True + ) as client, + ): + endpoint = sender_module.discover_webmention_endpoint( + client, "https://example.com/post" + ) + + assert endpoint == "https://example.com/webmention" + assert [request.method for request in requests] == ["HEAD", "GET"] + + +def test_send_webmention_persists_success( + app: Flask, sender_module: ModuleType, monkeypatch: pytest.MonkeyPatch +) -> None: + identifier = create_sent_webmention(app) + + _, client, discover, post = mock_sender_requests( + sender_module, monkeypatch, result=(201, STATUS_URL) + ) + + sender_module.send_webmention.call_local(identifier) + + discover.assert_called_once_with(client, TARGET_URL) + post.assert_called_once_with( + client, endpoint=ENDPOINT_URL, source=SOURCE_URL, target=TARGET_URL + ) + + with app.app_context(): + webmention = db.session.get(SentWebmention, identifier) + + assert webmention is not None + assert webmention.processed_revision == 1 + assert webmention.sent_revision == 1 + assert webmention.status == SentWebmentionStatus.SENT + assert webmention.failure_reason is None + assert webmention.endpoint == ENDPOINT_URL + assert webmention.response_status == 201 + assert webmention.status_url == STATUS_URL + assert webmention.last_attempted_at is not None + assert webmention.last_sent_at is not None + assert not webmention.pending + + +def test_send_webmention_marks_unsupported_target_processed( + app: Flask, sender_module: ModuleType, monkeypatch: pytest.MonkeyPatch +) -> None: + identifier = create_sent_webmention( + app, + desired_revision=2, + processed_revision=1, + sent_revision=1, + status=SentWebmentionStatus.SENT, + endpoint="https://old.example/webmention", + response_status=201, + status_url="https://old.example/status/123", + ) + + _, client, discover, post = mock_sender_requests( + sender_module, monkeypatch, endpoint=None + ) + + sender_module.send_webmention.call_local(identifier) + + discover.assert_called_once_with(client, TARGET_URL) + post.assert_not_called() + + with app.app_context(): + webmention = db.session.get(SentWebmention, identifier) + + assert webmention is not None + assert webmention.processed_revision == 2 + assert webmention.sent_revision == 1 + assert webmention.status == SentWebmentionStatus.UNSUPPORTED + assert webmention.failure_reason == "No Webmention endpoint discovered" + assert webmention.endpoint is None + assert webmention.response_status is None + assert webmention.status_url is None + assert webmention.last_attempted_at is not None + assert not webmention.pending + + +def test_send_webmention_persists_permanent_failure( + app: Flask, sender_module: ModuleType, monkeypatch: pytest.MonkeyPatch +) -> None: + identifier = create_sent_webmention( + app, + desired_revision=2, + processed_revision=1, + sent_revision=1, + status=SentWebmentionStatus.SENT, + status_url="https://old.example/status/123", + ) + + mock_sender_requests(sender_module, monkeypatch, result=(400, None)) + + sender_module.send_webmention.call_local(identifier) + + with app.app_context(): + webmention = db.session.get(SentWebmention, identifier) + + assert webmention is not None + assert webmention.processed_revision == 2 + assert webmention.sent_revision == 1 + assert webmention.status == SentWebmentionStatus.FAILED + assert webmention.failure_reason == "Webmention endpoint returned HTTP 400" + assert webmention.endpoint == ENDPOINT_URL + assert webmention.response_status == 400 + assert webmention.status_url is None + assert not webmention.pending + + +def test_send_webmention_persists_temporary_failure_and_reraises( + app: Flask, sender_module: ModuleType, monkeypatch: pytest.MonkeyPatch +) -> None: + identifier = create_sent_webmention( + app, + desired_revision=2, + processed_revision=1, + sent_revision=1, + status=SentWebmentionStatus.SENT, + status_url="https://old.example/status/123", + ) + + mock_sender_requests(sender_module, monkeypatch, result=(503, None)) + + with pytest.raises( + sender_module.TemporarySenderError, + match="Webmention endpoint returned HTTP 503", + ): + sender_module.send_webmention.call_local(identifier) + + with app.app_context(): + webmention = db.session.get(SentWebmention, identifier) + + assert webmention is not None + assert webmention.processed_revision == 1 + assert webmention.sent_revision == 1 + assert webmention.status == SentWebmentionStatus.FAILED + assert webmention.failure_reason == "Webmention endpoint returned HTTP 503" + assert webmention.endpoint == ENDPOINT_URL + assert webmention.response_status == 503 + assert webmention.status_url is None + assert webmention.pending + + +def test_send_webmention_ignores_processed_revision( + app: Flask, sender_module: ModuleType, monkeypatch: pytest.MonkeyPatch +) -> None: + identifier = create_sent_webmention( + app, + desired_revision=2, + processed_revision=2, + sent_revision=2, + status=SentWebmentionStatus.SENT, + ) + + httpx_client, _, discover, post = mock_sender_requests(sender_module, monkeypatch) + + sender_module.send_webmention.call_local(identifier) + + httpx_client.assert_not_called() + discover.assert_not_called() + post.assert_not_called() + + +def test_send_webmention_ignores_result_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 (202, 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 + + +def test_send_webmention_ignores_unknown_identifier( + sender_module: ModuleType, caplog: pytest.LogCaptureFixture +) -> None: + identifier = uuid.uuid7() + + with caplog.at_level(logging.WARNING): + sender_module.send_webmention.call_local(identifier) + + assert f"Cannot send unknown SentWebmention {identifier}" in caplog.text |