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