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