diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/conftest.py | 125 | ||||
| -rw-r--r-- | tests/tasks/test_receiver.py | 886 | ||||
| -rw-r--r-- | tests/test_forms.py | 130 | ||||
| -rw-r--r-- | tests/test_views.py | 183 |
4 files changed, 1324 insertions, 0 deletions
diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..94ba650 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,125 @@ +import uuid +from collections.abc import Callable, Iterator +from types import ModuleType + +import httpx +import pytest +from flask import Flask +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, +] + + +@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(): + db.create_all() + + yield application + + with application.app_context(): + db.session.remove() + db.drop_all() + db.engine.dispose() + + +@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. + """ + + 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 diff --git a/tests/tasks/test_receiver.py b/tests/tasks/test_receiver.py new file mode 100644 index 0000000..4545121 --- /dev/null +++ b/tests/tasks/test_receiver.py @@ -0,0 +1,886 @@ +import logging +import uuid +from collections.abc import Callable +from types import ModuleType + +import httpx +import pytest +from flask import Flask + +from webmentions_ssg import DATABASE as db +from webmentions_ssg.models import ReceivedWebmention + +SOURCE_URL = "https://source.example/article" +TARGET_URL = "https://dennisfink.me/blog/example/" + +ReceivedWebmentionFactory = Callable[..., uuid.UUID] +HTTPXMockInstaller = Callable[ + [Callable[[httpx.Request], httpx.Response]], + None, +] + + +def get_webmention_state( + app: Flask, + identifier: uuid.UUID, +) -> tuple[str, str | None]: + with app.app_context(): + webmention = db.session.get( + ReceivedWebmention, + identifier, + ) + + assert webmention is not None + + return ( + webmention.status, + webmention.failure_reason, + ) + + +@pytest.mark.parametrize( + ( + "body", + "source_url", + "target_url", + "expected", + ), + [ + pytest.param( + f'<a href="{TARGET_URL}">Reply</a>', + SOURCE_URL, + TARGET_URL, + True, + id="a-href", + ), + pytest.param( + f'<area href="{TARGET_URL}" alt="Target">', + SOURCE_URL, + TARGET_URL, + True, + id="area-href", + ), + pytest.param( + f'<link href="{TARGET_URL}" rel="alternate">', + SOURCE_URL, + TARGET_URL, + True, + id="link-href", + ), + pytest.param( + f'<img src="{TARGET_URL}" alt="">', + SOURCE_URL, + TARGET_URL, + True, + id="img-src", + ), + pytest.param( + f'<audio src="{TARGET_URL}"></audio>', + SOURCE_URL, + TARGET_URL, + True, + id="audio-src", + ), + pytest.param( + f'<video src="{TARGET_URL}"></video>', + SOURCE_URL, + TARGET_URL, + True, + id="video-src", + ), + pytest.param( + (f'<audio><source src="{TARGET_URL}"></audio>'), + SOURCE_URL, + TARGET_URL, + True, + id="audio-source-src", + ), + pytest.param( + (f'<video><source src="{TARGET_URL}"></video>'), + SOURCE_URL, + TARGET_URL, + True, + id="video-source-src", + ), + pytest.param( + f'<iframe src="{TARGET_URL}"></iframe>', + SOURCE_URL, + TARGET_URL, + True, + id="iframe-src", + ), + pytest.param( + f'<embed src="{TARGET_URL}">', + SOURCE_URL, + TARGET_URL, + True, + id="embed-src", + ), + pytest.param( + f'<script src="{TARGET_URL}"></script>', + SOURCE_URL, + TARGET_URL, + True, + id="script-src", + ), + pytest.param( + (f'<video><track src="{TARGET_URL}"></video>'), + SOURCE_URL, + TARGET_URL, + True, + id="track-src", + ), + pytest.param( + f'<input type="image" src="{TARGET_URL}" alt="">', + SOURCE_URL, + TARGET_URL, + True, + id="image-input-src", + ), + pytest.param( + f'<input type="IMAGE" src="{TARGET_URL}" alt="">', + SOURCE_URL, + TARGET_URL, + True, + id="image-input-case-insensitive", + ), + pytest.param( + f'<blockquote cite="{TARGET_URL}">Quotation</blockquote>', + SOURCE_URL, + TARGET_URL, + True, + id="blockquote-cite", + ), + pytest.param( + f'<q cite="{TARGET_URL}">Quotation</q>', + SOURCE_URL, + TARGET_URL, + True, + id="q-cite", + ), + pytest.param( + f'<ins cite="{TARGET_URL}">Addition</ins>', + SOURCE_URL, + TARGET_URL, + True, + id="ins-cite", + ), + pytest.param( + f'<del cite="{TARGET_URL}">Removal</del>', + SOURCE_URL, + TARGET_URL, + True, + id="del-cite", + ), + pytest.param( + '<a href="../target/">Reply</a>', + "https://source.example/posts/article/", + "https://source.example/posts/target/", + True, + id="relative-href", + ), + pytest.param( + '<base href="https://dennisfink.me/blog/"><a href="example/">Reply</a>', + SOURCE_URL, + TARGET_URL, + True, + id="base-url", + ), + pytest.param( + f'<img cite="{TARGET_URL}" alt="">', + SOURCE_URL, + TARGET_URL, + False, + id="img-cite-invalid", + ), + pytest.param( + f'<blockquote src="{TARGET_URL}">Quote</blockquote>', + SOURCE_URL, + TARGET_URL, + False, + id="blockquote-src-invalid", + ), + pytest.param( + f'<a src="{TARGET_URL}">Reply</a>', + SOURCE_URL, + TARGET_URL, + False, + id="a-src-invalid", + ), + pytest.param( + f'<div href="{TARGET_URL}"></div>', + SOURCE_URL, + TARGET_URL, + False, + id="div-href-invalid", + ), + pytest.param( + f'<link src="{TARGET_URL}">', + SOURCE_URL, + TARGET_URL, + False, + id="link-src-invalid", + ), + pytest.param( + f'<input type="text" src="{TARGET_URL}">', + SOURCE_URL, + TARGET_URL, + False, + id="text-input-src-invalid", + ), + pytest.param( + (f'<picture><source src="{TARGET_URL}"></picture>'), + SOURCE_URL, + TARGET_URL, + False, + id="picture-source-src-invalid", + ), + pytest.param( + f'<base href="{TARGET_URL}">', + SOURCE_URL, + TARGET_URL, + False, + id="base-is-not-mention", + ), + pytest.param( + (f'<a href="{TARGET_URL}more">Different page</a>'), + SOURCE_URL, + TARGET_URL, + False, + id="longer-url", + ), + pytest.param( + (f'<img cite="{TARGET_URL}" src="https://example.com/image.jpg" alt="">'), + SOURCE_URL, + TARGET_URL, + False, + id="invalid-cite-does-not-override-valid-src", + ), + pytest.param( + f"<p>{TARGET_URL}</p>", + SOURCE_URL, + TARGET_URL, + False, + id="text-content", + ), + pytest.param( + '<a href="https://example.com/">Other site</a>', + SOURCE_URL, + TARGET_URL, + False, + id="missing-target", + ), + ], +) +def test_html_mentions_target( + receiver_module: ModuleType, + body: str, + source_url: str, + target_url: str, + expected: bool, +) -> None: + assert ( + receiver_module.html_mentions_target( + body.encode(), + source_url, + target_url, + ) + is expected + ) + + +@pytest.mark.parametrize( + ("body", "target_url", "expected"), + [ + (TARGET_URL, TARGET_URL, True), + (f"This post replies to {TARGET_URL}", TARGET_URL, True), + ( + f"https://example.com/first {TARGET_URL} https://example.com/last", + TARGET_URL, + True, + ), + (f"{TARGET_URL}more", TARGET_URL, False), + ("https://dennisfink.me/blog/other/", TARGET_URL, False), + ("/blog/example/", TARGET_URL, False), + ("There are no links here.", TARGET_URL, False), + ], +) +def test_text_mentions_target( + receiver_module: ModuleType, + body: str, + target_url: str, + expected: bool, +) -> None: + assert ( + receiver_module.text_mentions_target( + body, + target_url, + ) + is expected + ) + + +def test_fetch_source_returns_response_and_body( + app: Flask, + receiver_module: ModuleType, + install_httpx_mock: HTTPXMockInstaller, +) -> None: + captured_request: httpx.Request | None = None + + def handler( + request: httpx.Request, + ) -> httpx.Response: + nonlocal captured_request + captured_request = request + + return httpx.Response( + 200, + headers={ + "Content-Type": "text/html", + }, + content=b"<p>Document</p>", + ) + + install_httpx_mock(handler) + + with app.app_context(): + response, body = receiver_module.fetch_source(SOURCE_URL) + + assert response.status_code == 200 + assert body == b"<p>Document</p>" + + assert captured_request is not None + assert captured_request.url == SOURCE_URL + + accept = captured_request.headers["Accept"] + + assert "text/html" in accept + assert "application/xhtml+xml" in accept + assert "text/plain" in accept + + assert captured_request.headers["User-Agent"] == ( + f"{receiver_module.APP_NAME}/{receiver_module.VERSION} ReceivedWebmention" + ) + + +@pytest.mark.parametrize( + ("status_code", "exception_name"), + [ + (400, "VerificationError"), + (404, "VerificationError"), + (410, "SourceGoneError"), + (408, "TemporaryFetchError"), + (425, "TemporaryFetchError"), + (429, "TemporaryFetchError"), + (500, "TemporaryFetchError"), + (503, "TemporaryFetchError"), + ], +) +def test_fetch_source_maps_http_status_to_exception( + app: Flask, + receiver_module: ModuleType, + install_httpx_mock: HTTPXMockInstaller, + status_code: int, + exception_name: str, +) -> None: + install_httpx_mock(lambda request: httpx.Response(status_code)) + + exception_type = getattr( + receiver_module, + exception_name, + ) + + with ( + app.app_context(), + pytest.raises( + exception_type, + match=f"HTTP {status_code}", + ), + ): + receiver_module.fetch_source(SOURCE_URL) + + +def test_fetch_source_propagates_network_error( + app: Flask, + receiver_module: ModuleType, + install_httpx_mock: HTTPXMockInstaller, +) -> None: + def handler( + request: httpx.Request, + ) -> httpx.Response: + raise httpx.ConnectError( + "Connection refused", + request=request, + ) + + install_httpx_mock(handler) + + with ( + app.app_context(), + pytest.raises( + httpx.ConnectError, + match="Connection refused", + ), + ): + receiver_module.fetch_source(SOURCE_URL) + + +def test_fetch_source_follows_redirect( + app: Flask, + receiver_module: ModuleType, + install_httpx_mock: HTTPXMockInstaller, +) -> None: + requested_paths: list[str] = [] + + def handler( + request: httpx.Request, + ) -> httpx.Response: + requested_paths.append(request.url.path) + + match request.url.path: + case "/start": + return httpx.Response( + 302, + headers={ + "Location": "/final", + }, + ) + + case "/final": + return httpx.Response( + 200, + headers={ + "Content-Type": "text/html", + }, + content=b"Final document", + ) + + case _: + raise AssertionError(f"Unexpected URL: {request.url}") + + install_httpx_mock(handler) + + with app.app_context(): + response, body = receiver_module.fetch_source("https://source.example/start") + + assert requested_paths == [ + "/start", + "/final", + ] + assert response.url.path == "/final" + assert body == b"Final document" + + +def test_fetch_source_enforces_redirect_limit( + app: Flask, + receiver_module: ModuleType, + install_httpx_mock: HTTPXMockInstaller, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setitem( + app.config, + "WEBMENTIONS_SSG_MAX_REDIRECTS", + 1, + ) + + install_httpx_mock( + lambda request: httpx.Response( + 302, + headers={ + "Location": "/another", + }, + ) + ) + + with ( + app.app_context(), + pytest.raises(httpx.TooManyRedirects), + ): + receiver_module.fetch_source("https://source.example/start") + + +def test_fetch_source_rejects_declared_oversized_body( + app: Flask, + receiver_module: ModuleType, + install_httpx_mock: HTTPXMockInstaller, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setitem( + app.config, + "WEBMENTIONS_SSG_MAX_SOURCE_BYTES", + 10, + ) + + install_httpx_mock( + lambda request: httpx.Response( + 200, + headers={ + "Content-Type": "text/html", + "Content-Length": "11", + }, + content=b"x" * 11, + ) + ) + + with ( + app.app_context(), + pytest.raises( + receiver_module.VerificationError, + match="Source document is too large", + ), + ): + receiver_module.fetch_source(SOURCE_URL) + + +def test_fetch_source_rejects_streamed_oversized_body( + app: Flask, + receiver_module: ModuleType, + install_httpx_mock: HTTPXMockInstaller, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setitem( + app.config, + "WEBMENTIONS_SSG_MAX_SOURCE_BYTES", + 10, + ) + + class BodyStream(httpx.SyncByteStream): + def __iter__(self): + yield b"x" * 6 + yield b"x" * 6 + + install_httpx_mock( + lambda request: httpx.Response( + 200, + headers={ + "Content-Type": "text/html", + }, + stream=BodyStream(), + ) + ) + + with ( + app.app_context(), + pytest.raises( + receiver_module.VerificationError, + match="Source document is too large", + ), + ): + receiver_module.fetch_source(SOURCE_URL) + + +@pytest.mark.parametrize( + ("content_type", "body", "expected"), + [ + pytest.param( + "text/html; charset=utf-8", + f'<a href="{TARGET_URL}">Reply</a>'.encode(), + True, + id="html", + ), + pytest.param( + "TEXT/HTML; CHARSET=UTF-8", + f'<a href="{TARGET_URL}">Reply</a>'.encode(), + True, + id="case-insensitive-html", + ), + pytest.param( + "application/xhtml+xml", + b'<a href="https://example.com/">Other</a>', + False, + id="xhtml-without-target", + ), + pytest.param( + "text/plain; charset=utf-8", + f"Reply to {TARGET_URL}".encode(), + True, + id="plain-text-utf-8", + ), + pytest.param( + "text/plain; charset=iso-8859-1", + (f"Grüße. Reply to {TARGET_URL}").encode("iso-8859-1"), + True, + id="plain-text-declared-encoding", + ), + pytest.param( + "text/plain; charset=utf-8", + b"\xff Reply to " + TARGET_URL.encode(), + True, + id="plain-text-invalid-byte", + ), + pytest.param( + "text/plain", + b"No target here.", + False, + id="plain-text-without-target", + ), + ], +) +def test_source_mentions_target_by_media_type( + receiver_module: ModuleType, + monkeypatch: pytest.MonkeyPatch, + content_type: str, + body: bytes, + expected: bool, +) -> None: + response = httpx.Response( + 200, + headers={ + "Content-Type": content_type, + }, + content=body, + request=httpx.Request( + "GET", + SOURCE_URL, + ), + ) + + monkeypatch.setattr( + receiver_module, + "fetch_source", + lambda source_url: ( + response, + body, + ), + ) + + assert ( + receiver_module.source_mentions_target( + SOURCE_URL, + TARGET_URL, + ) + is expected + ) + + +@pytest.mark.parametrize( + ("content_type", "expected_media_type"), + [ + ("application/json", "application/json"), + ("application/pdf", "application/pdf"), + ("", "missing"), + ], +) +def test_source_mentions_target_rejects_unsupported_media_type( + receiver_module: ModuleType, + monkeypatch: pytest.MonkeyPatch, + content_type: str, + expected_media_type: str, +) -> None: + headers = {} + + if content_type: + headers["Content-Type"] = content_type + + response = httpx.Response( + 200, + headers=headers, + content=b"Document", + request=httpx.Request( + "GET", + SOURCE_URL, + ), + ) + + monkeypatch.setattr( + receiver_module, + "fetch_source", + lambda source_url: ( + response, + b"Document", + ), + ) + + with pytest.raises( + receiver_module.VerificationError, + match=(f"Unsupported source content type: {expected_media_type}"), + ): + receiver_module.source_mentions_target( + SOURCE_URL, + TARGET_URL, + ) + + +def test_verify_webmention_marks_row_verifying_before_check( + app: Flask, + receiver_module: ModuleType, + make_webmention: ReceivedWebmentionFactory, + monkeypatch: pytest.MonkeyPatch, +) -> None: + identifier = make_webmention( + status="failed", + failure_reason="Earlier failure", + ) + + def verify_source( + source_url: str, + target_url: str, + ) -> bool: + webmention = db.session.get( + ReceivedWebmention, + identifier, + ) + + assert webmention is not None + assert webmention.status == "verifying" + assert webmention.failure_reason is None + assert source_url == webmention.source + assert target_url == webmention.target + + return True + + monkeypatch.setattr( + receiver_module, + "source_mentions_target", + verify_source, + ) + + receiver_module.verify_webmention.call_local(identifier) + + assert get_webmention_state( + app, + identifier, + ) == ( + "verified", + None, + ) + + +@pytest.mark.parametrize( + ("outcome", "expected_status", "expected_reason"), + [ + ("verified", "verified", None), + ("missing", "deleted", "Source does not mention target"), + ("gone", "deleted", "Source returned HTTP 410"), + ("permanent-failure", "failed", "Source returned HTTP 404"), + ], +) +def test_verify_webmention_persists_final_state( + app: Flask, + receiver_module: ModuleType, + make_webmention: ReceivedWebmentionFactory, + monkeypatch: pytest.MonkeyPatch, + outcome: str, + expected_status: str, + expected_reason: str | None, +) -> None: + identifier = make_webmention( + status="received", + ) + + def verify_source( + source_url: str, + target_url: str, + ) -> bool: + match outcome: + case "verified": + return True + + case "missing": + return False + + case "gone": + raise receiver_module.SourceGoneError("Source returned HTTP 410") + + case "permanent-failure": + raise receiver_module.VerificationError("Source returned HTTP 404") + + case _: + raise AssertionError(f"Unexpected outcome: {outcome}") + + monkeypatch.setattr( + receiver_module, + "source_mentions_target", + verify_source, + ) + + receiver_module.verify_webmention.call_local(identifier) + + assert get_webmention_state( + app, + identifier, + ) == ( + expected_status, + expected_reason, + ) + + +@pytest.mark.parametrize( + "failure", + [ + "temporary-http", + "network", + ], +) +def test_verify_webmention_persists_retryable_failure_and_reraises( + app: Flask, + receiver_module: ModuleType, + make_webmention: ReceivedWebmentionFactory, + monkeypatch: pytest.MonkeyPatch, + failure: str, +) -> None: + identifier = make_webmention() + + def verify_source( + source_url: str, + target_url: str, + ) -> bool: + match failure: + case "temporary-http": + raise receiver_module.TemporaryFetchError("Source returned HTTP 503") + + case "network": + raise httpx.ConnectError( + "Connection refused", + request=httpx.Request( + "GET", + source_url, + ), + ) + + case _: + raise AssertionError(f"Unexpected failure: {failure}") + + monkeypatch.setattr( + receiver_module, + "source_mentions_target", + verify_source, + ) + + match failure: + case "temporary-http": + expected_exception = receiver_module.TemporaryFetchError + expected_reason = "Source returned HTTP 503" + + case "network": + expected_exception = httpx.ConnectError + expected_reason = "Connection refused" + + case _: + raise AssertionError(f"Unexpected failure: {failure}") + + with pytest.raises( + expected_exception, + match=expected_reason, + ): + receiver_module.verify_webmention.call_local(identifier) + + assert get_webmention_state( + app, + identifier, + ) == ( + "failed", + expected_reason, + ) + + +def test_verify_webmention_ignores_unknown_identifier( + receiver_module: ModuleType, + caplog: pytest.LogCaptureFixture, +) -> None: + identifier = uuid.uuid7() + + with caplog.at_level(logging.WARNING): + receiver_module.verify_webmention.call_local(identifier) + + assert f"Cannot verify unknown ReceivedWebmention {identifier}" in caplog.text diff --git a/tests/test_forms.py b/tests/test_forms.py new file mode 100644 index 0000000..9be6ca8 --- /dev/null +++ b/tests/test_forms.py @@ -0,0 +1,130 @@ +import pytest +from flask import Flask +from werkzeug.datastructures import MultiDict + +from webmentions_ssg.forms import EndpointForm + +VALID_SOURCE = "https://source.example/post" +VALID_TARGET = "https://dennisfink.me/blog/example/" + + +@pytest.mark.parametrize( + ( + "form_data", + "invalid_field", + "expected_error", + ), + [ + pytest.param( + { + "target": VALID_TARGET, + }, + "source", + "This field is required.", + id="source-required", + ), + pytest.param( + { + "source": "not a URL", + "target": VALID_TARGET, + }, + "source", + "Invalid URL.", + id="source-url", + ), + pytest.param( + { + "source": "ftp://source.example/post", + "target": VALID_TARGET, + }, + "source", + "source must begin with http or https", + id="source-scheme", + ), + pytest.param( + { + "source": VALID_TARGET, + "target": VALID_TARGET, + }, + "source", + None, + id="source-not-equal-to-target", + ), + pytest.param( + { + "source": VALID_SOURCE, + }, + "target", + "This field is required.", + id="target-required", + ), + pytest.param( + { + "source": VALID_SOURCE, + "target": "not a URL", + }, + "target", + "Invalid URL.", + id="target-url", + ), + pytest.param( + { + "source": VALID_SOURCE, + "target": "ftp://dennisfink.me/blog/example/", + }, + "target", + "target must begin with http or https", + id="target-scheme", + ), + pytest.param( + { + "source": VALID_SOURCE, + "target": "https://example.com/post", + }, + "target", + None, + id="target-allowed-hostname", + ), + ], +) +def test_endpoint_form_rejects_invalid_data( + app: Flask, + form_data: dict[str, str], + invalid_field: str, + expected_error: str | None, +) -> None: + with app.test_request_context( + "/endpoint", + method="POST", + ): + form = EndpointForm( + formdata=MultiDict(form_data), + meta={"csrf": False}, + ) + + assert not form.validate() + assert invalid_field in form.errors + + if expected_error is not None: + assert expected_error in form.errors[invalid_field] + + +def test_endpoint_form_accepts_valid_data( + app: Flask, +) -> None: + with app.test_request_context( + "/endpoint", + method="POST", + ): + form = EndpointForm( + formdata=MultiDict( + { + "source": VALID_SOURCE, + "target": VALID_TARGET, + } + ), + meta={"csrf": False}, + ) + + assert form.validate() + assert form.errors == {} diff --git a/tests/test_views.py b/tests/test_views.py new file mode 100644 index 0000000..d7fd798 --- /dev/null +++ b/tests/test_views.py @@ -0,0 +1,183 @@ +import uuid +from types import ModuleType + +import pytest +import sqlalchemy as sa +from flask import Flask +from flask.testing import FlaskClient + +from webmentions_ssg import DATABASE as db +from webmentions_ssg.models import ReceivedWebmention + + +def test_endpoint_only_accepts_post( + client: FlaskClient, +) -> None: + response = client.get("/endpoint") + + assert response.status_code == 405 + + +def test_endpoint_returns_form_errors( + client: FlaskClient, +) -> None: + response = client.post( + "/endpoint", + data={ + "source": "https://source.example/post", + "target": "https://example.com/post", + }, + ) + + assert response.status_code == 400 + + errors = response.get_json() + + assert errors is not None + assert "target" in errors + + +def test_endpoint_creates_webmention( + app: Flask, + client: FlaskClient, + views_module: ModuleType, + monkeypatch: pytest.MonkeyPatch, +) -> None: + queued: list[uuid.UUID] = [] + + monkeypatch.setattr( + views_module, + "verify_webmention", + queued.append, + ) + + source = "https://source.example/post" + target = "https://dennisfink.me/blog/example/" + + response = client.post( + "/endpoint", + data={ + "source": source, + "target": target, + }, + ) + + assert response.status_code == 201 + + with app.app_context(): + webmention = db.session.scalar(sa.select(ReceivedWebmention)) + + assert webmention is not None + assert webmention.source == source + assert webmention.target == target + assert webmention.status == "received" + assert webmention.failure_reason is None + + identifier = webmention.uuid + + assert response.headers["Location"].endswith(str(identifier)) + + assert queued == [identifier] + + +def test_endpoint_is_idempotent( + app: Flask, + client: FlaskClient, + views_module: ModuleType, + monkeypatch: pytest.MonkeyPatch, +) -> 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/"), + } + + first_response = client.post( + "/endpoint", + data=data, + ) + second_response = client.post( + "/endpoint", + data=data, + ) + + 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(): + webmentions = db.session.scalars(sa.select(ReceivedWebmention)).all() + + assert len(webmentions) == 1 + + webmention = webmentions[0] + + assert webmention.source == data["source"] + assert webmention.target == data["target"] + assert webmention.status == "received" + assert webmention.failure_reason is None + + identifier = webmention.uuid + + assert queued == [ + identifier, + identifier, + ] + + +def test_resending_resets_failure_state( + app: Flask, + client: FlaskClient, + views_module: ModuleType, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr( + views_module, + "verify_webmention", + lambda identifier: None, + ) + + source = "https://source.example/post" + target = "https://dennisfink.me/blog/example/" + + with app.app_context(): + existing = ReceivedWebmention( + uuid=uuid.uuid7(), + source=source, + target=target, + status="failed", + failure_reason="Previous failure", + ) + + db.session.add(existing) + db.session.commit() + + identifier = existing.uuid + + response = client.post( + "/endpoint", + data={ + "source": source, + "target": target, + }, + ) + + assert response.status_code == 201 + + with app.app_context(): + webmention = db.session.get( + ReceivedWebmention, + identifier, + ) + + assert webmention is not None + assert webmention.status == "received" + assert webmention.failure_reason is None |