From 5b39f70aa14cac635b6084170e9d0bd3f548b911 Mon Sep 17 00:00:00 2001 From: Dennis Fink Date: Mon, 10 Aug 2026 20:00:28 +0200 Subject: test: simplify mocks and localize test fixtures Replace shared HTTPX and DNS fixtures with targeted unittest.mock patches close to the tests that need them. Keep conftest focused on application and client setup, while moving receiver-specific helpers into the receiver test module. --- tests/conftest.py | 122 +-------- tests/tasks/test_receiver.py | 627 +++++++++++++++++-------------------------- tests/test_forms.py | 12 +- tests/test_url_security.py | 18 +- tests/test_views.py | 72 +++-- 5 files changed, 289 insertions(+), 562 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index f1ef4b5..45658b4 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,8 +1,5 @@ -import uuid -from collections.abc import Callable, Iterator -from types import ModuleType +from collections.abc import Iterator -import httpx import pytest from flask import Flask from flask.testing import FlaskClient @@ -10,29 +7,10 @@ from flask.testing import FlaskClient from webmentions_ssg import DATABASE as db from webmentions_ssg import create_app from webmentions_ssg.config import TestingConfig -from webmentions_ssg.models import ReceivedWebmention - -HTTPHandler = Callable[ - [httpx.Request], - httpx.Response, -] - - -class FakeHostAnswers: - def __init__(self, *addresses: str): - self._addresses = addresses - - def addresses(self): - return iter(self._addresses) @pytest.fixture def app() -> Iterator[Flask]: - """ - Create a new Flask application and in-memory database - for every test. - """ - application = create_app(TestingConfig) with application.app_context(): @@ -49,101 +27,3 @@ def app() -> Iterator[Flask]: @pytest.fixture def client(app: Flask) -> FlaskClient: return app.test_client() - - -@pytest.fixture -def receiver_module(app: Flask) -> ModuleType: - """ - Ensure the application and Huey extension are initialized - before retrieving the tasks module. - """ - - from webmentions_ssg.tasks import receiver - - return receiver - - -@pytest.fixture -def views_module(app: Flask) -> ModuleType: - from webmentions_ssg import views - - return views - - -@pytest.fixture -def make_webmention( - app: Flask, -) -> Callable[..., uuid.UUID]: - def create( - *, - source: str = "https://source.example/post", - target: str = ("https://dennisfink.me/blog/example/"), - status: str = "received", - failure_reason: str | None = None, - ) -> uuid.UUID: - identifier = uuid.uuid7() - - with app.app_context(): - webmention = ReceivedWebmention( - uuid=identifier, - source=source, - target=target, - status=status, - failure_reason=failure_reason, - ) - - db.session.add(webmention) - db.session.commit() - - return identifier - - return create - - -@pytest.fixture -def install_httpx_mock( - monkeypatch: pytest.MonkeyPatch, - receiver_module: ModuleType, -) -> Callable[[HTTPHandler], None]: - """ - Replace the HTTPX transport without replacing HTTPX itself. - """ - - monkeypatch.setattr( - receiver_module, - "ensure_public_url", - lambda url: None, - ) - - real_client = httpx.Client - - def install(handler: HTTPHandler) -> None: - transport = httpx.MockTransport(handler) - - def create_client( - *args, - **kwargs, - ) -> httpx.Client: - return real_client( - *args, - transport=transport, - **kwargs, - ) - - monkeypatch.setattr( - receiver_module.httpx, - "Client", - create_client, - ) - - return install - - -@pytest.fixture -def public_dns( - monkeypatch: pytest.MonkeyPatch, -) -> None: - monkeypatch.setattr( - "webmentions_ssg.url_security.dns.resolver.resolve_name", - lambda hostname: FakeHostAnswers("93.184.216.34"), - ) diff --git a/tests/tasks/test_receiver.py b/tests/tasks/test_receiver.py index 4545121..19630c3 100644 --- a/tests/tasks/test_receiver.py +++ b/tests/tasks/test_receiver.py @@ -1,7 +1,7 @@ import logging import uuid -from collections.abc import Callable from types import ModuleType +from unittest.mock import Mock, patch import httpx import pytest @@ -9,15 +9,45 @@ 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, +) SOURCE_URL = "https://source.example/article" TARGET_URL = "https://dennisfink.me/blog/example/" -ReceivedWebmentionFactory = Callable[..., uuid.UUID] -HTTPXMockInstaller = Callable[ - [Callable[[httpx.Request], httpx.Response]], - None, -] + +@pytest.fixture +def receiver(app: Flask) -> ModuleType: + from webmentions_ssg.tasks import receiver + + return receiver + + +def create_webmention( + app: Flask, + *, + source: str = SOURCE_URL, + target: str = TARGET_URL, + status: str = "received", + failure_reason: str | None = None, +) -> uuid.UUID: + identifier = uuid.uuid7() + + with app.app_context(): + webmention = ReceivedWebmention( + uuid=identifier, + source=source, + target=target, + status=status, + failure_reason=failure_reason, + ) + + db.session.add(webmention) + db.session.commit() + + return identifier def get_webmention_state( @@ -38,257 +68,200 @@ def get_webmention_state( ) +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 + + @pytest.mark.parametrize( - ( - "body", - "source_url", - "target_url", - "expected", - ), + ("body", "expected"), [ pytest.param( f'Reply', - SOURCE_URL, - TARGET_URL, True, id="a-href", ), pytest.param( f'Target', - SOURCE_URL, - TARGET_URL, True, id="area-href", ), pytest.param( f'', - SOURCE_URL, - TARGET_URL, True, id="link-href", ), pytest.param( f'', - SOURCE_URL, - TARGET_URL, True, id="img-src", ), pytest.param( f'', - SOURCE_URL, - TARGET_URL, True, id="audio-src", ), pytest.param( f'', - SOURCE_URL, - TARGET_URL, True, id="video-src", ), pytest.param( - (f''), - SOURCE_URL, - TARGET_URL, + f'', True, id="audio-source-src", ), pytest.param( - (f''), - SOURCE_URL, - TARGET_URL, + f'', True, id="video-source-src", ), pytest.param( f'', - SOURCE_URL, - TARGET_URL, True, id="iframe-src", ), pytest.param( f'', - SOURCE_URL, - TARGET_URL, True, id="embed-src", ), pytest.param( f'', - SOURCE_URL, - TARGET_URL, True, id="script-src", ), pytest.param( - (f''), - SOURCE_URL, - TARGET_URL, + f'', True, id="track-src", ), pytest.param( f'', - SOURCE_URL, - TARGET_URL, True, id="image-input-src", ), pytest.param( f'', - SOURCE_URL, - TARGET_URL, True, id="image-input-case-insensitive", ), pytest.param( f'
Quotation
', - SOURCE_URL, - TARGET_URL, True, id="blockquote-cite", ), pytest.param( f'Quotation', - SOURCE_URL, - TARGET_URL, True, id="q-cite", ), pytest.param( f'Addition', - SOURCE_URL, - TARGET_URL, True, id="ins-cite", ), pytest.param( f'Removal', - SOURCE_URL, - TARGET_URL, True, id="del-cite", ), - pytest.param( - 'Reply', - "https://source.example/posts/article/", - "https://source.example/posts/target/", - True, - id="relative-href", - ), pytest.param( 'Reply', - SOURCE_URL, - TARGET_URL, True, id="base-url", ), pytest.param( f'', - SOURCE_URL, - TARGET_URL, False, id="img-cite-invalid", ), pytest.param( f'
Quote
', - SOURCE_URL, - TARGET_URL, False, id="blockquote-src-invalid", ), pytest.param( f'Reply', - SOURCE_URL, - TARGET_URL, False, id="a-src-invalid", ), pytest.param( f'
', - SOURCE_URL, - TARGET_URL, False, id="div-href-invalid", ), pytest.param( f'', - SOURCE_URL, - TARGET_URL, False, id="link-src-invalid", ), pytest.param( f'', - SOURCE_URL, - TARGET_URL, False, id="text-input-src-invalid", ), pytest.param( - (f''), - SOURCE_URL, - TARGET_URL, + f'', False, id="picture-source-src-invalid", ), pytest.param( f'', - SOURCE_URL, - TARGET_URL, False, id="base-is-not-mention", ), pytest.param( - (f'Different page'), - SOURCE_URL, - TARGET_URL, + f'Different page', False, id="longer-url", ), pytest.param( - (f''), - SOURCE_URL, - TARGET_URL, + f'', False, id="invalid-cite-does-not-override-valid-src", ), pytest.param( f"

{TARGET_URL}

", - SOURCE_URL, - TARGET_URL, False, id="text-content", ), pytest.param( 'Other site', - SOURCE_URL, - TARGET_URL, False, id="missing-target", ), ], ) def test_html_mentions_target( - receiver_module: ModuleType, + receiver: ModuleType, body: str, - source_url: str, - target_url: str, expected: bool, ) -> None: assert ( - receiver_module.html_mentions_target( + receiver.html_mentions_target( body.encode(), - source_url, - target_url, + SOURCE_URL, + TARGET_URL, ) is expected ) +def test_html_mentions_relative_target( + receiver: ModuleType, +) -> None: + assert receiver.html_mentions_target( + b'Reply', + "https://source.example/posts/article/", + "https://source.example/posts/target/", + ) + + @pytest.mark.parametrize( ("body", "target_url", "expected"), [ @@ -306,63 +279,97 @@ def test_html_mentions_target( ], ) def test_text_mentions_target( - receiver_module: ModuleType, + receiver: ModuleType, body: str, target_url: str, expected: bool, ) -> None: - assert ( - receiver_module.text_mentions_target( - body, - target_url, - ) - is expected - ) + assert receiver.text_mentions_target(body, target_url) is expected -def test_fetch_source_returns_response_and_body( - app: Flask, - receiver_module: ModuleType, - install_httpx_mock: HTTPXMockInstaller, +@patch("webmentions_ssg.tasks.receiver.ensure_public_url") +def test_ensure_public_request_accepts_public_url( + ensure_public_url: Mock, + receiver: ModuleType, ) -> None: - captured_request: httpx.Request | None = None + receiver.ensure_public_request(httpx.Request("GET", SOURCE_URL)) - def handler( - request: httpx.Request, - ) -> httpx.Response: - nonlocal captured_request - captured_request = request + ensure_public_url.assert_called_once_with(SOURCE_URL) - return httpx.Response( - 200, - headers={ - "Content-Type": "text/html", - }, - content=b"

Document

", - ) - install_httpx_mock(handler) +@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)) - with app.app_context(): - response, body = receiver_module.fetch_source(SOURCE_URL) + ensure_public_url.assert_called_once_with(SOURCE_URL) - assert response.status_code == 200 - assert body == b"

Document

" - assert captured_request is not None - assert captured_request.url == 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)) - accept = captured_request.headers["Accept"] + ensure_public_url.assert_called_once_with(SOURCE_URL) - assert "text/html" in accept - assert "application/xhtml+xml" in accept - assert "text/plain" in accept - assert captured_request.headers["User-Agent"] == ( - f"{receiver_module.APP_NAME}/{receiver_module.VERSION} ReceivedWebmention" +@patch("webmentions_ssg.tasks.receiver.httpx.Client") +def test_fetch_source_returns_response_and_body( + httpx_client: Mock, + app: Flask, + receiver: ModuleType, +) -> None: + response = httpx.Response( + 200, + headers={"Content-Type": "text/html"}, + content=b"

Document

", + request=httpx.Request("GET", SOURCE_URL), ) + client = 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"

Document

" + client.stream.assert_called_once_with("GET", SOURCE_URL) + + options = httpx_client.call_args.kwargs + + assert options["headers"] == { + "Accept": "text/html, application/xhtml+xml;q=0.9, text/plain;q=0.8", + "User-Agent": f"{receiver.APP_NAME}/{receiver.VERSION} ReceivedWebmention", + } + assert options["follow_redirects"] is True + assert options["max_redirects"] == app.config.get( + "WEBMENTIONS_SSG_MAX_REDIRECTS", + 20, + ) + assert options["trust_env"] is False + assert options["event_hooks"] == { + "request": [receiver.ensure_public_request], + } + + +@patch("webmentions_ssg.tasks.receiver.httpx.Client") @pytest.mark.parametrize( ("status_code", "exception_name"), [ @@ -377,19 +384,22 @@ def test_fetch_source_returns_response_and_body( ], ) def test_fetch_source_maps_http_status_to_exception( + httpx_client: Mock, app: Flask, - receiver_module: ModuleType, - install_httpx_mock: HTTPXMockInstaller, + receiver: ModuleType, status_code: int, exception_name: str, ) -> None: - install_httpx_mock(lambda request: httpx.Response(status_code)) - - exception_type = getattr( - receiver_module, - exception_name, + set_stream_response( + httpx_client, + httpx.Response( + status_code, + request=httpx.Request("GET", SOURCE_URL), + ), ) + exception_type = getattr(receiver, exception_name) + with ( app.app_context(), pytest.raises( @@ -397,23 +407,20 @@ def test_fetch_source_maps_http_status_to_exception( match=f"HTTP {status_code}", ), ): - receiver_module.fetch_source(SOURCE_URL) + 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_module: ModuleType, - install_httpx_mock: HTTPXMockInstaller, + receiver: ModuleType, ) -> None: - def handler( - request: httpx.Request, - ) -> httpx.Response: - raise httpx.ConnectError( - "Connection refused", - request=request, - ) - - install_httpx_mock(handler) + client = httpx_client.return_value.__enter__.return_value + client.stream.side_effect = httpx.ConnectError( + "Connection refused", + request=httpx.Request("GET", SOURCE_URL), + ) with ( app.app_context(), @@ -422,153 +429,74 @@ def test_fetch_source_propagates_network_error( match="Connection refused", ), ): - receiver_module.fetch_source(SOURCE_URL) - - -def test_fetch_source_follows_redirect( - app: Flask, - receiver_module: ModuleType, - install_httpx_mock: HTTPXMockInstaller, -) -> None: - requested_paths: list[str] = [] - - def handler( - request: httpx.Request, - ) -> httpx.Response: - requested_paths.append(request.url.path) - - match request.url.path: - case "/start": - return httpx.Response( - 302, - headers={ - "Location": "/final", - }, - ) - - case "/final": - return httpx.Response( - 200, - headers={ - "Content-Type": "text/html", - }, - content=b"Final document", - ) - - case _: - raise AssertionError(f"Unexpected URL: {request.url}") - - install_httpx_mock(handler) - - with app.app_context(): - response, body = receiver_module.fetch_source("https://source.example/start") - - assert requested_paths == [ - "/start", - "/final", - ] - assert response.url.path == "/final" - assert body == b"Final document" - - -def test_fetch_source_enforces_redirect_limit( - app: Flask, - receiver_module: ModuleType, - install_httpx_mock: HTTPXMockInstaller, - monkeypatch: pytest.MonkeyPatch, -) -> None: - monkeypatch.setitem( - app.config, - "WEBMENTIONS_SSG_MAX_REDIRECTS", - 1, - ) - - install_httpx_mock( - lambda request: httpx.Response( - 302, - headers={ - "Location": "/another", - }, - ) - ) - - with ( - app.app_context(), - pytest.raises(httpx.TooManyRedirects), - ): - receiver_module.fetch_source("https://source.example/start") + 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_module: ModuleType, - install_httpx_mock: HTTPXMockInstaller, - monkeypatch: pytest.MonkeyPatch, + receiver: ModuleType, ) -> None: - monkeypatch.setitem( - app.config, - "WEBMENTIONS_SSG_MAX_SOURCE_BYTES", - 10, - ) + app.config["WEBMENTIONS_SSG_MAX_SOURCE_BYTES"] = 10 - install_httpx_mock( - lambda request: httpx.Response( + set_stream_response( + httpx_client, + httpx.Response( 200, headers={ "Content-Type": "text/html", "Content-Length": "11", }, content=b"x" * 11, - ) + request=httpx.Request("GET", SOURCE_URL), + ), ) with ( app.app_context(), pytest.raises( - receiver_module.VerificationError, + receiver.VerificationError, match="Source document is too large", ), ): - receiver_module.fetch_source(SOURCE_URL) + 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_module: ModuleType, - install_httpx_mock: HTTPXMockInstaller, - monkeypatch: pytest.MonkeyPatch, + receiver: ModuleType, ) -> None: - monkeypatch.setitem( - app.config, - "WEBMENTIONS_SSG_MAX_SOURCE_BYTES", - 10, - ) + app.config["WEBMENTIONS_SSG_MAX_SOURCE_BYTES"] = 10 class BodyStream(httpx.SyncByteStream): def __iter__(self): yield b"x" * 6 yield b"x" * 6 - install_httpx_mock( - lambda request: httpx.Response( + set_stream_response( + httpx_client, + httpx.Response( 200, - headers={ - "Content-Type": "text/html", - }, + headers={"Content-Type": "text/html"}, stream=BodyStream(), - ) + request=httpx.Request("GET", SOURCE_URL), + ), ) with ( app.app_context(), pytest.raises( - receiver_module.VerificationError, + receiver.VerificationError, match="Source document is too large", ), ): - receiver_module.fetch_source(SOURCE_URL) + receiver.fetch_source(SOURCE_URL) +@patch("webmentions_ssg.tasks.receiver.fetch_source") @pytest.mark.parametrize( ("content_type", "body", "expected"), [ @@ -617,42 +545,25 @@ def test_fetch_source_rejects_streamed_oversized_body( ], ) def test_source_mentions_target_by_media_type( - receiver_module: ModuleType, - monkeypatch: pytest.MonkeyPatch, + fetch_source: Mock, + receiver: ModuleType, content_type: str, body: bytes, expected: bool, ) -> None: response = httpx.Response( 200, - headers={ - "Content-Type": content_type, - }, + headers={"Content-Type": content_type}, content=body, - request=httpx.Request( - "GET", - SOURCE_URL, - ), + request=httpx.Request("GET", SOURCE_URL), ) + fetch_source.return_value = response, body - monkeypatch.setattr( - receiver_module, - "fetch_source", - lambda source_url: ( - response, - body, - ), - ) - - assert ( - receiver_module.source_mentions_target( - SOURCE_URL, - TARGET_URL, - ) - is expected - ) + assert receiver.source_mentions_target(SOURCE_URL, TARGET_URL) is expected + fetch_source.assert_called_once_with(SOURCE_URL) +@patch("webmentions_ssg.tasks.receiver.fetch_source") @pytest.mark.parametrize( ("content_type", "expected_media_type"), [ @@ -662,8 +573,8 @@ def test_source_mentions_target_by_media_type( ], ) def test_source_mentions_target_rejects_unsupported_media_type( - receiver_module: ModuleType, - monkeypatch: pytest.MonkeyPatch, + fetch_source: Mock, + receiver: ModuleType, content_type: str, expected_media_type: str, ) -> None: @@ -676,38 +587,25 @@ def test_source_mentions_target_rejects_unsupported_media_type( 200, headers=headers, content=b"Document", - request=httpx.Request( - "GET", - SOURCE_URL, - ), - ) - - monkeypatch.setattr( - receiver_module, - "fetch_source", - lambda source_url: ( - response, - b"Document", - ), + request=httpx.Request("GET", SOURCE_URL), ) + fetch_source.return_value = response, b"Document" with pytest.raises( - receiver_module.VerificationError, - match=(f"Unsupported source content type: {expected_media_type}"), + receiver.VerificationError, + match=f"Unsupported source content type: {expected_media_type}", ): - receiver_module.source_mentions_target( - SOURCE_URL, - TARGET_URL, - ) + receiver.source_mentions_target(SOURCE_URL, TARGET_URL) +@patch("webmentions_ssg.tasks.receiver.source_mentions_target") def test_verify_webmention_marks_row_verifying_before_check( + source_mentions_target: Mock, app: Flask, - receiver_module: ModuleType, - make_webmention: ReceivedWebmentionFactory, - monkeypatch: pytest.MonkeyPatch, + receiver: ModuleType, ) -> None: - identifier = make_webmention( + identifier = create_webmention( + app, status="failed", failure_reason="Earlier failure", ) @@ -729,13 +627,9 @@ def test_verify_webmention_marks_row_verifying_before_check( return True - monkeypatch.setattr( - receiver_module, - "source_mentions_target", - verify_source, - ) + source_mentions_target.side_effect = verify_source - receiver_module.verify_webmention.call_local(identifier) + receiver.verify_webmention.call_local(identifier) assert get_webmention_state( app, @@ -746,6 +640,7 @@ def test_verify_webmention_marks_row_verifying_before_check( ) +@patch("webmentions_ssg.tasks.receiver.source_mentions_target") @pytest.mark.parametrize( ("outcome", "expected_status", "expected_reason"), [ @@ -756,45 +651,32 @@ def test_verify_webmention_marks_row_verifying_before_check( ], ) def test_verify_webmention_persists_final_state( + source_mentions_target: Mock, app: Flask, - receiver_module: ModuleType, - make_webmention: ReceivedWebmentionFactory, - monkeypatch: pytest.MonkeyPatch, + receiver: ModuleType, outcome: str, expected_status: str, expected_reason: str | None, ) -> None: - identifier = make_webmention( - status="received", - ) - - def verify_source( - source_url: str, - target_url: str, - ) -> bool: - match outcome: - case "verified": - return True - - case "missing": - return False - - case "gone": - raise receiver_module.SourceGoneError("Source returned HTTP 410") - - case "permanent-failure": - raise receiver_module.VerificationError("Source returned HTTP 404") - - case _: - raise AssertionError(f"Unexpected outcome: {outcome}") - - monkeypatch.setattr( - receiver_module, - "source_mentions_target", - verify_source, - ) + identifier = create_webmention(app) + + match outcome: + case "verified": + source_mentions_target.return_value = True + case "missing": + source_mentions_target.return_value = False + case "gone": + source_mentions_target.side_effect = receiver.SourceGoneError( + "Source returned HTTP 410" + ) + case "permanent-failure": + source_mentions_target.side_effect = receiver.VerificationError( + "Source returned HTTP 404" + ) + case _: + raise AssertionError(f"Unexpected outcome: {outcome}") - receiver_module.verify_webmention.call_local(identifier) + receiver.verify_webmention.call_local(identifier) assert get_webmention_state( app, @@ -805,6 +687,7 @@ def test_verify_webmention_persists_final_state( ) +@patch("webmentions_ssg.tasks.receiver.source_mentions_target") @pytest.mark.parametrize( "failure", [ @@ -813,74 +696,48 @@ def test_verify_webmention_persists_final_state( ], ) def test_verify_webmention_persists_retryable_failure_and_reraises( + source_mentions_target: Mock, app: Flask, - receiver_module: ModuleType, - make_webmention: ReceivedWebmentionFactory, - monkeypatch: pytest.MonkeyPatch, + receiver: ModuleType, failure: str, ) -> None: - identifier = make_webmention() - - def verify_source( - source_url: str, - target_url: str, - ) -> bool: - match failure: - case "temporary-http": - raise receiver_module.TemporaryFetchError("Source returned HTTP 503") - - case "network": - raise httpx.ConnectError( - "Connection refused", - request=httpx.Request( - "GET", - source_url, - ), - ) - - case _: - raise AssertionError(f"Unexpected failure: {failure}") - - monkeypatch.setattr( - receiver_module, - "source_mentions_target", - verify_source, - ) + identifier = create_webmention(app) match failure: case "temporary-http": - expected_exception = receiver_module.TemporaryFetchError - expected_reason = "Source returned HTTP 503" - + exception = receiver.TemporaryFetchError("Source returned HTTP 503") case "network": - expected_exception = httpx.ConnectError - expected_reason = "Connection refused" - + exception = httpx.ConnectError( + "Connection refused", + request=httpx.Request("GET", SOURCE_URL), + ) case _: raise AssertionError(f"Unexpected failure: {failure}") + source_mentions_target.side_effect = exception + with pytest.raises( - expected_exception, - match=expected_reason, + type(exception), + match=str(exception), ): - receiver_module.verify_webmention.call_local(identifier) + receiver.verify_webmention.call_local(identifier) assert get_webmention_state( app, identifier, ) == ( "failed", - expected_reason, + str(exception), ) def test_verify_webmention_ignores_unknown_identifier( - receiver_module: ModuleType, + receiver: ModuleType, caplog: pytest.LogCaptureFixture, ) -> None: identifier = uuid.uuid7() with caplog.at_level(logging.WARNING): - receiver_module.verify_webmention.call_local(identifier) + receiver.verify_webmention.call_local(identifier) assert f"Cannot verify unknown ReceivedWebmention {identifier}" in caplog.text diff --git a/tests/test_forms.py b/tests/test_forms.py index 2d5fc01..417c655 100644 --- a/tests/test_forms.py +++ b/tests/test_forms.py @@ -1,3 +1,5 @@ +from unittest.mock import patch + import pytest from flask import Flask from werkzeug.datastructures import MultiDict @@ -8,6 +10,10 @@ VALID_SOURCE = "https://source.example/post" VALID_TARGET = "https://dennisfink.me/blog/example/" +@patch( + "webmentions_ssg.forms.validators.ensure_public_url", + new=lambda url: None, +) @pytest.mark.parametrize( ( "form_data", @@ -89,7 +95,6 @@ VALID_TARGET = "https://dennisfink.me/blog/example/" ) def test_endpoint_form_rejects_invalid_data( app: Flask, - public_dns: None, form_data: dict[str, str], invalid_field: str, expected_error: str | None, @@ -110,9 +115,12 @@ def test_endpoint_form_rejects_invalid_data( assert expected_error in form.errors[invalid_field] +@patch( + "webmentions_ssg.forms.validators.ensure_public_url", + new=lambda url: None, +) def test_endpoint_form_accepts_valid_data( app: Flask, - public_dns: None, ) -> None: with app.test_request_context( "/endpoint", diff --git a/tests/test_url_security.py b/tests/test_url_security.py index fba6d7d..b10e6ff 100644 --- a/tests/test_url_security.py +++ b/tests/test_url_security.py @@ -1,3 +1,5 @@ +from unittest.mock import Mock, patch + import pytest from webmentions_ssg.url_security import ( @@ -6,14 +8,7 @@ from webmentions_ssg.url_security import ( ) -class FakeAnswers: - def __init__(self, *addresses: str): - self._addresses = addresses - - def addresses(self): - return iter(self._addresses) - - +@patch("webmentions_ssg.url_security.dns.resolver.resolve_name") @pytest.mark.parametrize( ("url", "resolved_addresses"), [ @@ -29,14 +24,11 @@ class FakeAnswers: ], ) def test_ensure_public_url_rejects_non_public_addresses( - monkeypatch: pytest.MonkeyPatch, + resolve_name: Mock, url: str, resolved_addresses: list[str], ) -> None: - monkeypatch.setattr( - "webmentions_ssg.url_security.dns.resolver.resolve_name", - lambda hostname: FakeAnswers(*resolved_addresses), - ) + resolve_name.return_value.addresses.return_value = resolved_addresses with pytest.raises(NonPublicAddressError): ensure_public_url(url) diff --git a/tests/test_views.py b/tests/test_views.py index 70ed732..06e546c 100644 --- a/tests/test_views.py +++ b/tests/test_views.py @@ -1,7 +1,6 @@ import uuid -from types import ModuleType +from unittest.mock import Mock, call, patch -import pytest import sqlalchemy as sa from flask import Flask from flask.testing import FlaskClient @@ -18,6 +17,10 @@ def test_endpoint_only_accepts_post( assert response.status_code == 405 +@patch( + "webmentions_ssg.forms.validators.ensure_public_url", + new=lambda url: None, +) def test_endpoint_returns_form_errors( client: FlaskClient, ) -> None: @@ -37,21 +40,16 @@ def test_endpoint_returns_form_errors( assert "target" in errors +@patch( + "webmentions_ssg.forms.validators.ensure_public_url", + new=lambda url: None, +) +@patch("webmentions_ssg.views.verify_webmention") def test_endpoint_creates_webmention( + verify_webmention: Mock, app: Flask, client: FlaskClient, - views_module: ModuleType, - monkeypatch: pytest.MonkeyPatch, - public_dns: None, ) -> None: - queued: list[uuid.UUID] = [] - - monkeypatch.setattr( - views_module, - "verify_webmention", - queued.append, - ) - source = "https://source.example/post" target = "https://dennisfink.me/blog/example/" @@ -77,28 +75,22 @@ def test_endpoint_creates_webmention( identifier = webmention.uuid assert response.headers["Location"].endswith(str(identifier)) - - assert queued == [identifier] + verify_webmention.assert_called_once_with(identifier) +@patch( + "webmentions_ssg.forms.validators.ensure_public_url", + new=lambda url: None, +) +@patch("webmentions_ssg.views.verify_webmention") def test_endpoint_is_idempotent( + verify_webmention: Mock, app: Flask, client: FlaskClient, - views_module: ModuleType, - monkeypatch: pytest.MonkeyPatch, - public_dns: None, ) -> None: - queued: list[uuid.UUID] = [] - - monkeypatch.setattr( - views_module, - "verify_webmention", - queued.append, - ) - data = { - "source": ("https://source.example/post"), - "target": ("https://dennisfink.me/blog/example/"), + "source": "https://source.example/post", + "target": "https://dennisfink.me/blog/example/", } first_response = client.post( @@ -112,7 +104,6 @@ def test_endpoint_is_idempotent( assert first_response.status_code == 201 assert second_response.status_code == 201 - assert first_response.headers["Location"] == second_response.headers["Location"] with app.app_context(): @@ -129,25 +120,22 @@ def test_endpoint_is_idempotent( identifier = webmention.uuid - assert queued == [ - identifier, - identifier, + assert verify_webmention.call_args_list == [ + call(identifier), + call(identifier), ] +@patch( + "webmentions_ssg.forms.validators.ensure_public_url", + new=lambda url: None, +) +@patch("webmentions_ssg.views.verify_webmention") def test_resending_resets_failure_state( + verify_webmention: Mock, app: Flask, client: FlaskClient, - views_module: ModuleType, - monkeypatch: pytest.MonkeyPatch, - public_dns: None, ) -> None: - monkeypatch.setattr( - views_module, - "verify_webmention", - lambda identifier: None, - ) - source = "https://source.example/post" target = "https://dennisfink.me/blog/example/" @@ -184,3 +172,5 @@ def test_resending_resets_failure_state( assert webmention is not None assert webmention.status == "received" assert webmention.failure_reason is None + + verify_webmention.assert_called_once_with(identifier) -- cgit v1.3.1