From e0f65d7ff582b32f1c7b5508bba8cda807aaae5f Mon Sep 17 00:00:00 2001 From: Dennis Fink Date: Sat, 15 Aug 2026 20:54:53 +0200 Subject: feat(sender): add outgoing Webmention pipeline Scan generated h-entry documents for Webmention targets and track source revisions and delivery state in the database. Discover target endpoints, send Webmentions asynchronously, and reconcile updated, removed, restored, and deleted sources across scans. Add authenticated views for inspecting sent Webmentions and make the scanner schedule configurable. --- tests/test_views.py | 202 ++++++++++++++++++++++++++++++++++++---------------- 1 file changed, 140 insertions(+), 62 deletions(-) (limited to 'tests/test_views.py') diff --git a/tests/test_views.py b/tests/test_views.py index 06e546c..aac281e 100644 --- a/tests/test_views.py +++ b/tests/test_views.py @@ -1,29 +1,89 @@ import uuid +from datetime import datetime, timezone from unittest.mock import Mock, call, patch +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 +from webmentions_ssg.models import ( + ReceivedWebmention, + SentWebmention, + SentWebmentionStatus, + Source, + User, +) -def test_endpoint_only_accepts_post( - client: FlaskClient, -) -> None: +def log_in(app: Flask, client: FlaskClient) -> None: + with app.app_context(): + user = User(username="admin") + + db.session.add(user) + db.session.commit() + + user_id = user.id + + with client.session_transaction() as session: + session["_user_id"] = str(user_id) + session["_fresh"] = True + + +def create_sent_source(app: Flask) -> uuid.UUID: + now = datetime.now(timezone.utc) + + with app.app_context(): + source = Source( + path="blog/example/index.html", + url="https://dennisfink.me/blog/example/", + content_hash="0" * 32, + revision=2, + last_seen_at=now, + revised_at=now, + ) + + source.sent_webmentions.extend( + [ + SentWebmention( + target="https://example.com/b", + active=True, + desired_revision=2, + processed_revision=2, + sent_revision=2, + status=SentWebmentionStatus.SENT, + endpoint="https://example.com/webmention", + response_status=202, + ), + SentWebmention( + target="https://example.com/a", + active=True, + desired_revision=2, + processed_revision=2, + sent_revision=1, + status=SentWebmentionStatus.FAILED, + failure_reason="Webmention endpoint returned HTTP 400", + endpoint="https://example.com/webmention", + response_status=400, + ), + ] + ) + + db.session.add(source) + db.session.commit() + + return source.uuid + + +def test_endpoint_only_accepts_post(client: FlaskClient) -> None: response = client.get("/endpoint") 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: +@patch("webmentions_ssg.forms.validators.is_public_url", new=lambda url: True) +def test_endpoint_returns_form_errors(client: FlaskClient) -> None: response = client.post( "/endpoint", data={ @@ -40,26 +100,15 @@ 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.forms.validators.is_public_url", new=lambda url: True) @patch("webmentions_ssg.views.verify_webmention") def test_endpoint_creates_webmention( - verify_webmention: Mock, - app: Flask, - client: FlaskClient, + verify_webmention: Mock, app: Flask, client: FlaskClient ) -> None: source = "https://source.example/post" target = "https://dennisfink.me/blog/example/" - response = client.post( - "/endpoint", - data={ - "source": source, - "target": target, - }, - ) + response = client.post("/endpoint", data={"source": source, "target": target}) assert response.status_code == 201 @@ -78,29 +127,18 @@ def test_endpoint_creates_webmention( verify_webmention.assert_called_once_with(identifier) -@patch( - "webmentions_ssg.forms.validators.ensure_public_url", - new=lambda url: None, -) +@patch("webmentions_ssg.forms.validators.is_public_url", new=lambda url: True) @patch("webmentions_ssg.views.verify_webmention") def test_endpoint_is_idempotent( - verify_webmention: Mock, - app: Flask, - client: FlaskClient, + verify_webmention: Mock, app: Flask, client: FlaskClient ) -> None: 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, - ) + 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 @@ -120,21 +158,13 @@ def test_endpoint_is_idempotent( identifier = webmention.uuid - assert verify_webmention.call_args_list == [ - call(identifier), - call(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.forms.validators.is_public_url", new=lambda url: True) @patch("webmentions_ssg.views.verify_webmention") def test_resending_resets_failure_state( - verify_webmention: Mock, - app: Flask, - client: FlaskClient, + verify_webmention: Mock, app: Flask, client: FlaskClient ) -> None: source = "https://source.example/post" target = "https://dennisfink.me/blog/example/" @@ -153,24 +183,72 @@ def test_resending_resets_failure_state( identifier = existing.uuid - response = client.post( - "/endpoint", - data={ - "source": source, - "target": target, - }, - ) + response = client.post("/endpoint", data={"source": source, "target": target}) assert response.status_code == 201 with app.app_context(): - webmention = db.session.get( - ReceivedWebmention, - identifier, - ) + webmention = db.session.get(ReceivedWebmention, identifier) assert webmention is not None assert webmention.status == "received" assert webmention.failure_reason is None verify_webmention.assert_called_once_with(identifier) + + +@pytest.mark.parametrize("path", ["/sent", f"/sent/{uuid.uuid7()}"]) +def test_sent_views_require_login(client: FlaskClient, path: str) -> None: + response = client.get(path) + + assert response.status_code == 302 + assert "/login" in response.headers["Location"] + + +def test_sent_lists_sources(app: Flask, client: FlaskClient) -> None: + log_in(app, client) + identifier = create_sent_source(app) + + response = client.get("/sent") + + assert response.status_code == 200 + + html = response.get_data(as_text=True) + + assert "blog/example/index.html" in html + assert "https://dennisfink.me/blog/example/" in html + assert f"/sent/{identifier}" in html + + +def test_sent_source_lists_webmentions(app: Flask, client: FlaskClient) -> None: + log_in(app, client) + identifier = create_sent_source(app) + + response = client.get(f"/sent/{identifier}") + + assert response.status_code == 200 + + html = response.get_data(as_text=True) + + assert "blog/example/index.html" in html + assert "https://dennisfink.me/blog/example/" in html + + first_target = "https://example.com/a" + second_target = "https://example.com/b" + + assert first_target in html + assert second_target in html + assert html.index(first_target) < html.index(second_target) + + assert "Webmention endpoint returned HTTP 400" in html + assert "https://example.com/webmention" in html + + +def test_sent_source_returns_404_for_unknown_source( + app: Flask, client: FlaskClient +) -> None: + log_in(app, client) + + response = client.get(f"/sent/{uuid.uuid7()}") + + assert response.status_code == 404 -- cgit v1.3.1