From c5492398f100ccb154fa557d657bae2322f95cf7 Mon Sep 17 00:00:00 2001 From: Dennis Fink Date: Fri, 21 Aug 2026 10:37:56 +0200 Subject: test(core): align tests with module ownership Move direct validator, model, and Huey extension coverage into dedicated test modules and mock dependencies where behavior belongs to another module. Expand Huey extension typing and coverage while removing duplicate assertions from scanner, sender, form, and view tests. --- tests/forms/test_forms.py | 115 ++++++++++++++++++ tests/forms/test_validators.py | 114 ++++++++++++++++++ tests/tasks/test_extension.py | 175 +++++++++++++++++++++++++++ tests/tasks/test_receiver.py | 5 - tests/tasks/test_scanner.py | 97 ++++----------- tests/tasks/test_sender.py | 261 ++++++++++++++++++++--------------------- tests/test_forms.py | 96 --------------- tests/test_models.py | 148 +++++++++++++++++++++++ tests/test_url_security.py | 29 +++-- tests/test_views.py | 117 +++++++++++------- 10 files changed, 792 insertions(+), 365 deletions(-) create mode 100644 tests/forms/test_forms.py create mode 100644 tests/forms/test_validators.py create mode 100644 tests/tasks/test_extension.py delete mode 100644 tests/test_forms.py create mode 100644 tests/test_models.py (limited to 'tests') diff --git a/tests/forms/test_forms.py b/tests/forms/test_forms.py new file mode 100644 index 0000000..6fe1849 --- /dev/null +++ b/tests/forms/test_forms.py @@ -0,0 +1,115 @@ +# SPDX-FileCopyrightText: 2026 Dennis Fink +# +# SPDX-License-Identifier: BSD-3-Clause + +from unittest.mock import Mock + +import pytest +from flask import Flask +from werkzeug.datastructures import MultiDict + +from webmentions_ssg.forms import AdminActionForm, EndpointForm, LoginForm + +VALID_SOURCE = "https://source.example/post" +VALID_TARGET = "https://dennisfink.me/blog/example/" + + +@pytest.fixture(autouse=True) +def public_urls(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr( + "webmentions_ssg.forms.validators.is_public_url", Mock(return_value=True) + ) + + +@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_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", + ), + ], +) +def test_endpoint_form_rejects_invalid_field_syntax( + app: Flask, form_data: dict[str, str], invalid_field: str, expected_error: str +) -> None: + with app.test_request_context("/endpoint", method="POST"): + form = EndpointForm(formdata=MultiDict(form_data), meta={"csrf": False}) + + assert not form.validate() + 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 == {} + + +@pytest.mark.parametrize( + ("form_data", "invalid_field"), + [ + pytest.param({"password": "secret"}, "username", id="username-required"), + pytest.param({"username": "admin"}, "password", id="password-required"), + ], +) +def test_login_form_requires_credentials( + app: Flask, form_data: dict[str, str], invalid_field: str +) -> None: + with app.test_request_context("/login", method="POST"): + form = LoginForm(formdata=MultiDict(form_data), meta={"csrf": False}) + + assert not form.validate() + assert "This field is required." in form.errors[invalid_field] + + +def test_login_form_accepts_credentials(app: Flask) -> None: + with app.test_request_context("/login", method="POST"): + form = LoginForm( + formdata=MultiDict({"username": "admin", "password": "secret"}), + meta={"csrf": False}, + ) + assert form.validate() + + +def test_admin_action_form_accepts_submission(app: Flask) -> None: + with app.test_request_context(method="POST"): + form = AdminActionForm(meta={"csrf": False}) + assert form.validate() diff --git a/tests/forms/test_validators.py b/tests/forms/test_validators.py new file mode 100644 index 0000000..c63ee54 --- /dev/null +++ b/tests/forms/test_validators.py @@ -0,0 +1,114 @@ +# SPDX-FileCopyrightText: 2026 Dennis Fink +# +# SPDX-License-Identifier: BSD-3-Clause + +from unittest.mock import Mock + +import pytest +from flask import Flask +from werkzeug.datastructures import MultiDict +from wtforms import Form, StringField, ValidationError + +from webmentions_ssg.forms.validators import AllowedHostname, NotEqualTo, PublicURL +from webmentions_ssg.url_security import AddressResolutionError + + +class ComparisonForm(Form): + source = StringField("Source") + target = StringField("Target") + + +class URLForm(Form): + url = StringField("URL") + + +def test_not_equal_to_accepts_different_values() -> None: + form = ComparisonForm(MultiDict({"source": "source", "target": "target"})) + NotEqualTo("target")(form, form.source) + + +def test_not_equal_to_rejects_equal_values() -> None: + form = ComparisonForm(MultiDict({"source": "same", "target": "same"})) + with pytest.raises(ValidationError, match="Field must not be equal to target"): + NotEqualTo("target")(form, form.source) + + +def test_not_equal_to_uses_custom_message() -> None: + form = ComparisonForm(MultiDict({"source": "same", "target": "same"})) + with pytest.raises(ValidationError, match="Must differ from Target"): + NotEqualTo("target", "Must differ from %(other_label)s")(form, form.source) + + +def test_not_equal_to_rejects_unknown_field() -> None: + form = ComparisonForm(MultiDict({"source": "source", "target": "target"})) + with pytest.raises(ValidationError, match="Invalid field name 'missing'"): + NotEqualTo("missing")(form, form.source) + + +def test_allowed_hostname_accepts_configured_hostname(app: Flask) -> None: + app.config["WEBMENTIONS_SSG_ALLOWED_HOSTNAMES"] = ["dennisfink.me"] + form = URLForm(MultiDict({"url": "https://dennisfink.me/blog/example/"})) + with app.app_context(): + AllowedHostname()(form, form.url) + + +def test_allowed_hostname_rejects_unconfigured_hostname(app: Flask) -> None: + app.config["WEBMENTIONS_SSG_ALLOWED_HOSTNAMES"] = ["dennisfink.me"] + form = URLForm(MultiDict({"url": "https://example.com/post"})) + with app.app_context(), pytest.raises(ValidationError, match="Invalid input"): + AllowedHostname()(form, form.url) + + +def test_allowed_hostname_uses_custom_message(app: Flask) -> None: + app.config["WEBMENTIONS_SSG_ALLOWED_HOSTNAMES"] = ["dennisfink.me"] + form = URLForm(MultiDict({"url": "https://example.com/post"})) + with ( + app.app_context(), + pytest.raises(ValidationError, match="Hostname is not allowed"), + ): + AllowedHostname("Hostname is not allowed")(form, form.url) + + +def test_public_url_accepts_public_url(monkeypatch: pytest.MonkeyPatch) -> None: + is_public_url = Mock(return_value=True) + monkeypatch.setattr("webmentions_ssg.forms.validators.is_public_url", is_public_url) + form = URLForm(MultiDict({"url": "https://example.com/post"})) + PublicURL()(form, form.url) + is_public_url.assert_called_once_with("https://example.com/post") + + +@pytest.mark.parametrize( + "outcome", + [ + pytest.param(False, id="non-public-address"), + pytest.param( + AddressResolutionError("Could not resolve hostname"), id="resolution-error" + ), + pytest.param(ValueError("No hostname was specified"), id="missing-hostname"), + ], +) +def test_public_url_rejects_invalid_url( + monkeypatch: pytest.MonkeyPatch, outcome: bool | Exception +) -> None: + is_public_url = Mock() + + if isinstance(outcome, Exception): + is_public_url.side_effect = outcome + else: + is_public_url.return_value = outcome + + monkeypatch.setattr("webmentions_ssg.forms.validators.is_public_url", is_public_url) + form = URLForm(MultiDict({"url": "https://example.com/post"})) + + with pytest.raises(ValidationError, match="URL must resolve to a public address"): + PublicURL()(form, form.url) + + +def test_public_url_uses_custom_message(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr( + "webmentions_ssg.forms.validators.is_public_url", Mock(return_value=False) + ) + form = URLForm(MultiDict({"url": "https://example.com/post"})) + + with pytest.raises(ValidationError, match="Public URL required"): + PublicURL("Public URL required")(form, form.url) diff --git a/tests/tasks/test_extension.py b/tests/tasks/test_extension.py new file mode 100644 index 0000000..5e9cdfa --- /dev/null +++ b/tests/tasks/test_extension.py @@ -0,0 +1,175 @@ +# SPDX-FileCopyrightText: 2026 Dennis Fink +# +# SPDX-License-Identifier: BSD-3-Clause + +from collections.abc import Callable +from typing import Any + +import huey as huey_package +import pytest +from flask import Flask, has_app_context +from huey import crontab + +from webmentions_ssg.tasks.extension import Huey + + +def test_init_without_app() -> None: + huey = Huey() + assert huey.app is None + with pytest.raises(RuntimeError, match="Huey has not been initialized"): + _ = huey.huey + + +def test_init_with_app(app: Flask) -> None: + huey = Huey(app) + assert huey.app is app + assert huey.huey.name == app.import_name + assert huey.huey.results is True + assert huey.huey.store_none is False + assert huey.huey.utc is True + assert huey.huey.immediate is True + assert app.extensions["huey"] is huey + + +def test_init_app_uses_huey_configuration(app: Flask) -> None: + app.config["HUEY"] = { + "url": "memory://", + "name": "custom-name", + "results": False, + "store_none": True, + "utc": False, + "immediate": False, + } + + huey = Huey() + huey.init_app(app) + + assert huey.huey.name == "custom-name" + assert huey.huey.results is False + assert huey.huey.store_none is True + assert huey.huey.utc is False + assert huey.huey.immediate is False + + +def test_huey_url_takes_precedence_over_huey_configuration(app: Flask) -> None: + app.config["HUEY"] = {"url": "blackhole://"} + app.config["HUEY_URL"] = "memory://" + + huey = Huey(app) + + assert type(huey.huey).__name__ == "MemoryHuey" + + +def task_decorator(huey: Huey, periodic: bool) -> Callable[[Callable[..., Any]], Any]: + if periodic: + return huey.periodic_task(crontab(minute="*")) + + return huey.task() + + +@pytest.mark.parametrize("periodic", [False, True], ids=["task", "periodic-task"]) +def test_task_runs_in_app_context(app: Flask, periodic: bool) -> None: + huey = Huey(app) + + @task_decorator(huey, periodic) + def add(a: int, b: int) -> int: + assert has_app_context() + return a + b + + assert add.call_local(2, 3) == 5 + + +@pytest.mark.parametrize("periodic", [False, True], ids=["task", "periodic-task"]) +def test_task_raises_without_app(app: Flask, periodic: bool) -> None: + huey = Huey(app) + + @task_decorator(huey, periodic) + def task() -> None: + pass + + huey.app = None + + with pytest.raises(RuntimeError, match="Flask app is not available"): + task.call_local() + + +def test_getattr_delegates_to_huey(app: Flask) -> None: + huey = Huey(app) + assert huey.immediate is True + + +@pytest.mark.parametrize( + ("url", "backend_name", "storage_kwargs"), + [ + ("redis://localhost:6379/0", "RedisHuey", {"url": "redis://localhost:6379/0"}), + ( + "rediss://localhost:6379/0", + "RedisHuey", + {"url": "rediss://localhost:6379/0"}, + ), + ( + "redis+priority://localhost:6379/0", + "PriorityRedisHuey", + {"url": "redis://localhost:6379/0"}, + ), + ( + "redis+expire://localhost:6379/0", + "RedisExpireHuey", + {"url": "redis://localhost:6379/0"}, + ), + ( + "redis+priority+expire://localhost:6379/0", + "PriorityRedisExpireHuey", + {"url": "redis://localhost:6379/0"}, + ), + ("sqlite:///var/huey.db", "SqliteHuey", {"filename": "var/huey.db"}), + ("file:///var/huey-queue", "FileHuey", {"path": "var/huey-queue"}), + ( + "postgres://user:password@localhost/database", + "PostgresHuey", + {"dsn": "postgres://user:password@localhost/database"}, + ), + ( + "postgresql://user:password@localhost/database", + "PostgresHuey", + {"dsn": "postgresql://user:password@localhost/database"}, + ), + ("memory://", "MemoryHuey", {}), + ("blackhole://", "BlackHoleHuey", {}), + ], +) +def test_backend_from_url( + monkeypatch: pytest.MonkeyPatch, + url: str, + backend_name: str, + storage_kwargs: dict[str, str], +) -> None: + backend_class = type(f"Test{backend_name}", (), {}) + + monkeypatch.setattr(huey_package, backend_name, backend_class) + + actual_class, actual_storage_kwargs = Huey.backend_from_url(url) + + assert actual_class is backend_class + assert actual_storage_kwargs == storage_kwargs + + +@pytest.mark.parametrize( + ("url", "message"), + [ + ( + "sqlite://var/huey.db", + "SQLite Huey URLs must look like sqlite:///var/huey.db", + ), + ("sqlite:///", "SQLite Huey URL must include a database path"), + ( + "file://var/huey-queue", + "File Huey URLs must look like file:///var/huey-queue", + ), + ("file:///", "File Huey URL must include a directory path"), + ("amqp://localhost", "Unsupported HUEY_URL scheme: 'amqp'"), + ], +) +def test_backend_from_url_rejects_invalid_url(url: str, message: str) -> None: + with pytest.raises(RuntimeError, match=message): + Huey.backend_from_url(url) diff --git a/tests/tasks/test_receiver.py b/tests/tasks/test_receiver.py index 17dd45a..daa271d 100644 --- a/tests/tasks/test_receiver.py +++ b/tests/tasks/test_receiver.py @@ -54,9 +54,7 @@ def create_webmention( 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) @@ -433,9 +431,7 @@ def test_verify_webmention_marks_row_verifying_before_check( return True source_mentions_target.side_effect = verify_source - receiver.verify_webmention.call_local(identifier) - assert get_webmention_state(app, identifier) == ("verified", None) @@ -521,7 +517,6 @@ 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) diff --git a/tests/tasks/test_scanner.py b/tests/tasks/test_scanner.py index 332b8bf..3de5e16 100644 --- a/tests/tasks/test_scanner.py +++ b/tests/tasks/test_scanner.py @@ -4,6 +4,7 @@ from pathlib import Path from types import ModuleType +from unittest.mock import Mock from uuid import UUID import pytest @@ -20,13 +21,11 @@ SOURCE_URL = f"{BASE_URL}example/" @pytest.fixture -def scanner_module(app: Flask) -> ModuleType: - # Importing scanner registers Huey tasks. The dependency on the - # app fixture guarantees that Huey has been initialized first. +def scanner_module(app: Flask, monkeypatch: pytest.MonkeyPatch) -> ModuleType: _ = app - from webmentions_ssg.tasks import scanner + monkeypatch.setattr(scanner, "is_http_url", Mock(return_value=True)) return scanner @@ -96,11 +95,6 @@ def configure_scanner(app: Flask, root: Path, *, base_url: str | None = None) -> app.config["WEBMENTIONS_SSG_SOURCE_BASE_URL"] = base_url -# --------------------------------------------------------------------------- -# Microformats parsing -# --------------------------------------------------------------------------- - - def test_parse_entry_returns_microformats_entry(scanner_module: ModuleType) -> None: document = parse_html( f""" @@ -277,11 +271,6 @@ def test_primary_entry_rejects_nonmatching_u_url(scanner_module: ModuleType) -> scanner_module.primary_entry(document, SOURCE_URL) -# --------------------------------------------------------------------------- -# e-content -# --------------------------------------------------------------------------- - - def test_content_element_returns_e_content(scanner_module: ModuleType) -> None: document = parse_html( """ @@ -346,11 +335,6 @@ def test_content_element_rejects_multiple_e_content(scanner_module: ModuleType) scanner_module.content_element(entry) -# --------------------------------------------------------------------------- -# Canonical URLs -# --------------------------------------------------------------------------- - - def test_canonical_url_returns_absolute_url(scanner_module: ModuleType) -> None: document = parse_html( f""" @@ -393,6 +377,8 @@ def test_canonical_url_returns_none_when_missing(scanner_module: ModuleType) -> def test_canonical_url_resolves_relative_url_with_base_url( scanner_module: ModuleType, ) -> None: + scanner_module.is_http_url.side_effect = [False, True] + document = parse_html( """ @@ -414,6 +400,8 @@ def test_canonical_url_resolves_relative_url_with_base_url( def test_relative_canonical_requires_base_url(scanner_module: ModuleType) -> None: + scanner_module.is_http_url.return_value = False + document = parse_html( """ @@ -453,11 +441,6 @@ def test_empty_canonical_is_rejected(scanner_module: ModuleType) -> None: ) -# --------------------------------------------------------------------------- -# Target extraction -# --------------------------------------------------------------------------- - - def test_scan_source_uses_canonical_without_base_url( scanner_module: ModuleType, tmp_path: Path ) -> None: @@ -657,24 +640,6 @@ def test_scan_source_extracts_reaction_properties( assert scanned.targets == frozenset({target}) -def test_scan_source_extracts_anchor_from_e_content( - scanner_module: ModuleType, tmp_path: Path -) -> None: - path = write_post( - tmp_path, - "example", - """ - - Target - - """, - ) - - scanned = scanner_module.scan_source_file(path, root=tmp_path, base_url=None) - - assert scanned.targets == frozenset({"https://example.com/target"}) - - @pytest.mark.parametrize("tag", ("link", "area")) def test_scan_source_ignores_non_anchor_href_elements( scanner_module: ModuleType, tmp_path: Path, tag: str @@ -714,11 +679,6 @@ def test_scan_source_ignores_self_fragment( assert scanned.targets == frozenset({"https://example.com/target"}) -# --------------------------------------------------------------------------- -# Database reconciliation -# --------------------------------------------------------------------------- - - def test_scan_creates_source_and_webmentions( app: Flask, scanner_module: ModuleType, tmp_path: Path, monkeypatch: MonkeyPatch ) -> None: @@ -900,7 +860,6 @@ def test_updated_source_increments_revision( assert webmention.desired_revision == 2 assert webmention.processed_revision == 1 assert webmention.sent_revision == 1 - assert webmention.pending identifier = webmention.uuid @@ -972,7 +931,6 @@ def test_new_target_is_added_on_update( assert webmention.desired_revision == 2 assert webmention.processed_revision is None assert webmention.sent_revision is None - assert webmention.pending identifier = webmention.uuid @@ -1055,7 +1013,6 @@ def test_removed_sent_target_is_queued_again( assert webmention.desired_revision == 2 assert webmention.processed_revision == 1 assert webmention.sent_revision == 1 - assert webmention.pending assert queued == [identifier] @@ -1124,16 +1081,10 @@ def test_removed_unsent_target_is_not_queued( assert webmention.desired_revision == 2 assert webmention.processed_revision == 2 assert webmention.sent_revision is None - assert not webmention.pending assert queued == [] -# --------------------------------------------------------------------------- -# Source deletion/restoration -# --------------------------------------------------------------------------- - - def test_deleted_source_queues_previously_sent_webmention( app: Flask, scanner_module: ModuleType, tmp_path: Path, monkeypatch: MonkeyPatch ) -> None: @@ -1186,7 +1137,6 @@ def test_deleted_source_queues_previously_sent_webmention( assert webmention.desired_revision == 2 assert webmention.processed_revision == 1 assert webmention.sent_revision == 1 - assert webmention.pending assert queued == [identifier] @@ -1231,7 +1181,6 @@ def test_deleted_source_does_not_queue_unsent_webmention( assert webmention.desired_revision == 2 assert webmention.processed_revision == 2 assert webmention.sent_revision is None - assert not webmention.pending assert queued == [] @@ -1298,18 +1247,12 @@ def test_restored_source_creates_new_revision( assert webmention.active assert webmention.desired_revision == 3 assert webmention.sent_revision == 1 - assert webmention.pending identifier = webmention.uuid assert queued == [identifier] -# --------------------------------------------------------------------------- -# Queue recovery and invalid sources -# --------------------------------------------------------------------------- - - def test_pending_webmention_is_requeued_on_next_scan( app: Flask, scanner_module: ModuleType, tmp_path: Path, monkeypatch: MonkeyPatch ) -> None: @@ -1436,6 +1379,8 @@ def test_scan_sources_accepts_valid_base_url( def test_scan_sources_rejects_invalid_base_url( app: Flask, scanner_module: ModuleType, tmp_path: Path ) -> None: + scanner_module.is_http_url.return_value = False + configure_scanner(app, tmp_path, base_url="not-a-url") with pytest.raises(RuntimeError, match="absolute HTTP or HTTPS URL"): @@ -1565,6 +1510,8 @@ def test_scan_source_ignores_reaction_to_matching_hostname( def test_canonical_url_rejects_non_http_resolved_url( scanner_module: ModuleType, ) -> None: + scanner_module.is_http_url.return_value = False + document = parse_html( """ @@ -1634,19 +1581,21 @@ def test_parse_entry_rejects_invalid_parser_output( 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 +def test_normalize_target_ignores_empty_target(scanner_module: ModuleType) -> None: + assert ( + scanner_module.normalize_target(" ", base_url=SOURCE_URL, source_url=SOURCE_URL) + is None + ) + + +def test_normalize_target_ignores_url_rejected_by_validator( + scanner_module: ModuleType, ) -> None: + scanner_module.is_http_url.return_value = False + assert ( scanner_module.normalize_target( - value, base_url=SOURCE_URL, source_url=SOURCE_URL + "https://example.com/target", base_url=SOURCE_URL, source_url=SOURCE_URL ) is None ) diff --git a/tests/tasks/test_sender.py b/tests/tasks/test_sender.py index dac56a5..6e33887 100644 --- a/tests/tasks/test_sender.py +++ b/tests/tasks/test_sender.py @@ -90,15 +90,59 @@ def mock_sender_requests( @pytest.fixture def sender_module(app: Flask) -> ModuleType: - # Importing sender registers Huey tasks. The dependency on the - # app fixture guarantees that Huey has been initialized first. _ = app - from webmentions_ssg.tasks import sender return sender +@pytest.mark.parametrize( + ("status", "expected"), + [ + pytest.param(200, False, id="success"), + pytest.param(400, False, id="client-error"), + pytest.param(408, True, id="request-timeout"), + pytest.param(425, True, id="too-early"), + pytest.param(429, True, id="too-many-requests"), + pytest.param(500, True, id="server-error-lower-bound"), + pytest.param(599, True, id="server-error-upper-bound"), + pytest.param(600, False, id="outside-server-error-range"), + ], +) +def test_temporary_http_status( + sender_module: ModuleType, status: int, expected: bool +) -> None: + assert sender_module.temporary_http_status(status) is expected + + +@pytest.mark.parametrize( + ("desired_revision", "processed_revision", "revision", "expected"), + [ + pytest.param(2, None, 2, True, id="unprocessed-current"), + pytest.param(2, 1, 2, True, id="older-processed-current"), + pytest.param(2, 2, 2, False, id="already-processed"), + pytest.param(2, 3, 2, False, id="newer-processed"), + pytest.param(3, 1, 2, False, id="superseded"), + ], +) +def test_attempt_is_current( + app: Flask, + sender_module: ModuleType, + desired_revision: int, + processed_revision: int | None, + revision: int, + expected: bool, +) -> None: + identifier = create_sent_webmention( + app, desired_revision=desired_revision, processed_revision=processed_revision + ) + + with app.app_context(): + webmention = db.session.get(SentWebmention, identifier) + assert webmention is not None + assert sender_module.attempt_is_current(webmention, revision) is expected + + @pytest.mark.parametrize( ("value", "expected"), ( @@ -119,8 +163,10 @@ def sender_module(app: Flask) -> ModuleType: ("https://example.com/webmention", {"webmention", "alternate"}), ), ( - "; " - 'type="text/html"; rel="webmention"; title="Endpoint"', + ( + '; type="text/html"; ' + 'rel="webmention"; title="Endpoint"' + ), ("https://example.com/webmention", {"webmention"}), ), ( @@ -159,7 +205,10 @@ def test_parse_link_value_rejects_malformed_value( ( ( "Link", - "; rel=webmention", + ( + ";" + "rel=webmention" + ), ), ), "", @@ -223,7 +272,10 @@ def test_parse_link_value_rejects_malformed_value( ( ( "LinK", - "; rel=webmention", + ( + "; " + "rel=webmention" + ), ), ), "", @@ -235,7 +287,10 @@ def test_parse_link_value_rejects_malformed_value( ( ( "Link", - '; rel="webmention"', + ( + "; " + 'rel="webmention"' + ), ), ), "", @@ -259,8 +314,10 @@ def test_parse_link_value_rejects_malformed_value( ( ( "Link", - "; " - 'rel="webmention somethingelse"', + ( + "; " + 'rel="webmention somethingelse"' + ), ), ), "", @@ -407,8 +464,10 @@ def test_parse_link_value_rejects_malformed_value( ("Link", '; rel="other"'), ( "Link", - "; " - 'rel="webmention"', + ( + "; " + 'rel="webmention"' + ), ), ), "", @@ -420,10 +479,12 @@ def test_parse_link_value_rejects_malformed_value( ( ( "Link", - "; " - 'rel="other", ' - "; " - 'rel="webmention"', + ( + "; " + 'rel="other", ' + "; " + 'rel="webmention"' + ), ), ), "", @@ -625,7 +686,6 @@ def test_send_webmention_persists_success( assert webmention.status_url == STATUS_URL assert webmention.last_attempted_at is not None assert webmention.last_sent_at is not None - assert not webmention.pending def test_send_webmention_marks_unsupported_target_processed( @@ -663,7 +723,6 @@ def test_send_webmention_marks_unsupported_target_processed( assert webmention.response_status is None assert webmention.status_url is None assert webmention.last_attempted_at is not None - assert not webmention.pending def test_send_webmention_persists_permanent_failure( @@ -693,7 +752,6 @@ def test_send_webmention_persists_permanent_failure( assert webmention.endpoint == ENDPOINT_URL assert webmention.response_status == 400 assert webmention.status_url is None - assert not webmention.pending def test_send_webmention_persists_temporary_failure_and_reraises( @@ -727,7 +785,6 @@ def test_send_webmention_persists_temporary_failure_and_reraises( assert webmention.endpoint == ENDPOINT_URL assert webmention.response_status == 503 assert webmention.status_url is None - assert webmention.pending def test_send_webmention_ignores_processed_revision( @@ -750,14 +807,17 @@ def test_send_webmention_ignores_processed_revision( post.assert_not_called() -def test_send_webmention_ignores_result_for_superseded_revision( - app: Flask, sender_module: ModuleType, monkeypatch: pytest.MonkeyPatch +@pytest.mark.parametrize( + "outcome", ["success", "unsupported", "temporary-failure", "permanent-failure"] +) +def test_send_webmention_ignores_superseded_attempt( + app: Flask, sender_module: ModuleType, monkeypatch: pytest.MonkeyPatch, outcome: str ) -> None: identifier = create_sent_webmention(app) - _, _, _, post = mock_sender_requests(sender_module, monkeypatch) + _, _, discover, post = mock_sender_requests(sender_module, monkeypatch) - def supersede_revision(*args, **kwargs) -> tuple[int, None]: + def supersede_revision() -> None: webmention = db.session.get(SentWebmention, identifier) assert webmention is not None @@ -765,9 +825,40 @@ def test_send_webmention_ignores_result_for_superseded_revision( webmention.desired_revision = 2 db.session.commit() - return (202, None) + match outcome: + case "success": + + def post_success(*args, **kwargs) -> tuple[int, None]: + supersede_revision() + return (202, None) + + post.side_effect = post_success + + case "unsupported": + + def discover_unsupported(*args, **kwargs) -> None: + supersede_revision() + + discover.side_effect = discover_unsupported - post.side_effect = supersede_revision + case "temporary-failure": + + def post_temporary_failure(*args, **kwargs) -> None: + supersede_revision() + raise sender_module.TemporarySenderError("Temporary failure") + + post.side_effect = post_temporary_failure + + case "permanent-failure": + + def post_permanent_failure(*args, **kwargs) -> tuple[int, None]: + supersede_revision() + return (400, None) + + post.side_effect = post_permanent_failure + + case _: + raise AssertionError(f"Unexpected outcome: {outcome}") sender_module.send_webmention.call_local(identifier) @@ -779,17 +870,14 @@ def test_send_webmention_ignores_result_for_superseded_revision( 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_unknown_identifier( sender_module: ModuleType, caplog: pytest.LogCaptureFixture ) -> None: identifier = uuid.uuid7() - with caplog.at_level(logging.WARNING): sender_module.send_webmention.call_local(identifier) - assert f"Cannot send unknown SentWebmention {identifier}" in caplog.text @@ -798,9 +886,7 @@ def test_ensure_public_request_accepts_public_url( ) -> 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) @@ -809,13 +895,11 @@ def test_ensure_public_request_rejects_non_public_address( ) -> 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) @@ -826,23 +910,25 @@ def test_ensure_public_request_maps_dns_failure_to_temporary_error( 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: +def test_resolve_endpoint_rejects_invalid_url( + sender_module: ModuleType, monkeypatch: pytest.MonkeyPatch +) -> None: + is_http_url = Mock(return_value=False) + monkeypatch.setattr(sender_module, "is_http_url", is_http_url) 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") + sender_module.resolve_endpoint(response, "/webmention") + is_http_url.assert_called_once_with(ENDPOINT_URL) def test_endpoint_from_headers_ignores_malformed_link( @@ -853,7 +939,6 @@ def test_endpoint_from_headers_ignores_malformed_link( headers=[("Link", "not-a-link"), ("Link", "; rel=webmention")], request=httpx.Request("GET", TARGET_URL), ) - assert sender_module.endpoint_from_headers(response) == ENDPOINT_URL @@ -1065,101 +1150,3 @@ def test_post_webmention( 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_forms.py b/tests/test_forms.py deleted file mode 100644 index fc16858..0000000 --- a/tests/test_forms.py +++ /dev/null @@ -1,96 +0,0 @@ -# SPDX-FileCopyrightText: 2026 Dennis Fink -# -# SPDX-License-Identifier: BSD-3-Clause - -from unittest.mock import patch - -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/" - - -@patch("webmentions_ssg.forms.validators.is_public_url", new=lambda url: True) -@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] - - -@patch("webmentions_ssg.forms.validators.is_public_url", new=lambda url: True) -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_models.py b/tests/test_models.py new file mode 100644 index 0000000..2fa2cb8 --- /dev/null +++ b/tests/test_models.py @@ -0,0 +1,148 @@ +# SPDX-FileCopyrightText: 2026 Dennis Fink +# +# SPDX-License-Identifier: BSD-3-Clause + +import uuid +from collections.abc import Callable +from datetime import UTC, datetime + +import pytest + +from webmentions_ssg.models import ( + ReceivedWebmention, + SentWebmention, + Source, + User, + uuid7_to_datetime, +) + + +def test_user_repr() -> None: + user = User(username="admin", password="secret") + assert repr(user) == "" + + +def test_user_password_is_write_only() -> None: + user = User(username="admin", password="secret") + with pytest.raises(AttributeError, match="Password is write-only"): + _ = user.password + + +def test_user_password_is_hashed_and_can_be_checked() -> None: + user = User(username="admin", password="secret") + assert user.password_hash != "secret" + assert user.check_password("secret") + assert not user.check_password("wrong") + + +@pytest.mark.parametrize( + ("status", "expected"), + [ + ("received", False), + ("verifying", False), + ("verified", True), + ("failed", False), + ("deleted", False), + ], +) +def test_received_webmention_verified(status: str, expected: bool) -> None: + webmention = ReceivedWebmention( + uuid=uuid.uuid7(), + source="https://source.example/post", + target="https://dennisfink.me/blog/example/", + status=status, + ) + + assert webmention.verified is expected + + +@pytest.mark.parametrize( + ("desired_revision", "processed_revision", "expected"), + [(1, None, True), (1, 0, True), (2, 1, True), (1, 1, False), (1, 2, False)], +) +def test_sent_webmention_pending( + desired_revision: int, processed_revision: int | None, expected: bool +) -> None: + webmention = SentWebmention( + target="https://example.com/post", + desired_revision=desired_revision, + processed_revision=processed_revision, + ) + assert webmention.pending is expected + + +def test_uuid7_to_datetime() -> None: + identifier = uuid.uuid7() + timestamp = uuid7_to_datetime(identifier) + + assert timestamp.tzinfo is UTC + assert timestamp.timestamp() == pytest.approx(identifier.time / 1000) + + +@pytest.mark.parametrize( + "factory", + [ + pytest.param( + lambda identifier: ReceivedWebmention( + uuid=identifier, + source="https://source.example/post", + target="https://dennisfink.me/blog/example/", + ), + id="received-webmention", + ), + pytest.param( + lambda identifier: Source( + uuid=identifier, + path="example/index.html", + url="https://dennisfink.me/blog/example/", + content_hash="0" * 32, + revision=1, + last_seen_at=datetime.now(UTC), + revised_at=datetime.now(UTC), + ), + id="source", + ), + pytest.param( + lambda identifier: SentWebmention( + uuid=identifier, target="https://example.com/post", desired_revision=1 + ), + id="sent-webmention", + ), + ], +) +def test_created_at_comes_from_uuid7_timestamp( + factory: Callable[[uuid.UUID], ReceivedWebmention | Source | SentWebmention], +) -> None: + identifier = uuid.uuid7() + model = factory(identifier) + assert model.created_at == uuid7_to_datetime(identifier) + + +def test_received_webmention_repr() -> None: + webmention = ReceivedWebmention( + uuid=uuid.uuid7(), + source="https://source.example/post", + target="https://dennisfink.me/blog/example/", + ) + + assert repr(webmention) == ( + " " + "'https://dennisfink.me/blog/example/'>" + ) + + +def test_source_repr() -> None: + source = Source( + path="example/index.html", + url="https://dennisfink.me/blog/example/", + content_hash="0" * 32, + revision=1, + last_seen_at=datetime.now(UTC), + revised_at=datetime.now(UTC), + ) + assert repr(source) == "" + + +def test_sent_webmention_repr() -> None: + webmention = SentWebmention(target="https://example.com/post", desired_revision=1) + assert repr(webmention) == "" diff --git a/tests/test_url_security.py b/tests/test_url_security.py index db513f9..dee38dd 100644 --- a/tests/test_url_security.py +++ b/tests/test_url_security.py @@ -14,6 +14,27 @@ from webmentions_ssg.url_security import ( ) +@pytest.mark.parametrize( + ("url", "expected"), + [ + pytest.param("http://example.com/", True, id="http"), + pytest.param("https://example.com/path", True, id="https"), + pytest.param("HTTP://EXAMPLE.COM/", True, id="scheme-case-insensitive"), + pytest.param("https://example.com:8443/path", True, id="port"), + pytest.param("http://[2001:db8::1]/", True, id="ipv6"), + pytest.param("ftp://example.com/", False, id="ftp"), + pytest.param("mailto:example@example.com", False, id="mailto"), + pytest.param("/relative/url", False, id="relative"), + pytest.param("//example.com/path", False, id="scheme-relative"), + pytest.param("https:///missing-host", False, id="missing-host"), + pytest.param("", False, id="empty"), + pytest.param("http://[::1", False, id="malformed"), + ], +) +def test_is_http_url(url: str, expected: bool) -> None: + assert is_http_url(url) is expected + + @patch("webmentions_ssg.url_security.dns.resolver.resolve_name") @pytest.mark.parametrize( ("url", "resolved_addresses"), @@ -34,7 +55,6 @@ def test_is_public_url_returns_false_for_non_public_addresses( resolve_name: Mock, url: str, resolved_addresses: list[str] ) -> None: resolve_name.return_value.addresses.return_value = resolved_addresses - assert not is_public_url(url) @@ -53,7 +73,6 @@ def test_is_public_url_returns_true_for_public_addresses( resolve_name: Mock, url: str, resolved_addresses: list[str] ) -> None: resolve_name.return_value.addresses.return_value = resolved_addresses - assert is_public_url(url) @@ -64,7 +83,6 @@ def test_is_public_url_returns_true_for_public_addresses( def test_is_public_url_raises_for_resolution_failure(resolve_name: Mock) -> None: with pytest.raises(AddressResolutionError, match="Could not resolve hostname"): is_public_url("https://nonexistent.example/") - resolve_name.assert_called_once_with("nonexistent.example") @@ -72,9 +90,4 @@ def test_is_public_url_raises_for_resolution_failure(resolve_name: Mock) -> None def test_is_public_url_raises_when_url_has_no_hostname(resolve_name: Mock) -> None: with pytest.raises(ValueError, match="No hostname was specified"): 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 0de8b9b..844da2e 100644 --- a/tests/test_views.py +++ b/tests/test_views.py @@ -4,7 +4,8 @@ import uuid from datetime import UTC, datetime -from unittest.mock import Mock, call, patch +from types import ModuleType +from unittest.mock import Mock, call import pytest import sqlalchemy as sa @@ -20,6 +21,36 @@ from webmentions_ssg.models import ( User, ) +SOURCE_URL = "https://source.example/post" +TARGET_URL = "https://dennisfink.me/blog/example/" + + +@pytest.fixture +def views_module(app: Flask) -> ModuleType: + _ = app + from webmentions_ssg import views + + return views + + +def mock_endpoint_form( + views_module: ModuleType, + monkeypatch: pytest.MonkeyPatch, + *, + valid: bool, + source: str = SOURCE_URL, + target: str = TARGET_URL, + errors: dict[str, list[str]] | None = None, +) -> Mock: + form = Mock() + form.validate_on_submit.return_value = valid + form.source.data = source + form.target.data = target + form.errors = errors or {} + + monkeypatch.setattr(views_module.forms, "EndpointForm", Mock(return_value=form)) + return form + def log_in(app: Flask, client: FlaskClient) -> None: with app.app_context(): @@ -82,46 +113,42 @@ def create_sent_source(app: Flask) -> uuid.UUID: def test_endpoint_only_accepts_post(client: FlaskClient) -> None: response = client.get("/endpoint") - assert response.status_code == 405 -@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={ - "source": "https://source.example/post", - "target": "https://example.com/post", - }, - ) - - assert response.status_code == 400 +def test_endpoint_returns_form_errors( + client: FlaskClient, views_module: ModuleType, monkeypatch: pytest.MonkeyPatch +) -> None: + errors = {"source": ["Invalid source"]} + mock_endpoint_form(views_module, monkeypatch, valid=False, errors=errors) - errors = response.get_json() + response = client.post("/endpoint") - assert errors is not None - assert "target" in errors + assert response.status_code == 400 + assert response.get_json() == errors -@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 + app: Flask, + client: FlaskClient, + views_module: ModuleType, + monkeypatch: pytest.MonkeyPatch, ) -> None: - source = "https://source.example/post" - target = "https://dennisfink.me/blog/example/" + form = mock_endpoint_form(views_module, monkeypatch, valid=True) + verify_webmention = Mock() + monkeypatch.setattr(views_module, "verify_webmention", verify_webmention) - response = client.post("/endpoint", data={"source": source, "target": target}) + response = client.post("/endpoint") assert response.status_code == 201 + form.validate_on_submit.assert_called_once_with() 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.source == SOURCE_URL + assert webmention.target == TARGET_URL assert webmention.status == "received" assert webmention.failure_reason is None @@ -131,18 +158,18 @@ def test_endpoint_creates_webmention( verify_webmention.assert_called_once_with(identifier) -@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 + app: Flask, + client: FlaskClient, + views_module: ModuleType, + monkeypatch: pytest.MonkeyPatch, ) -> None: - data = { - "source": "https://source.example/post", - "target": "https://dennisfink.me/blog/example/", - } + mock_endpoint_form(views_module, monkeypatch, valid=True) + verify_webmention = Mock() + monkeypatch.setattr(views_module, "verify_webmention", verify_webmention) - first_response = client.post("/endpoint", data=data) - second_response = client.post("/endpoint", data=data) + first_response = client.post("/endpoint") + second_response = client.post("/endpoint") assert first_response.status_code == 201 assert second_response.status_code == 201 @@ -155,8 +182,8 @@ def test_endpoint_is_idempotent( webmention = webmentions[0] - assert webmention.source == data["source"] - assert webmention.target == data["target"] + assert webmention.source == SOURCE_URL + assert webmention.target == TARGET_URL assert webmention.status == "received" assert webmention.failure_reason is None @@ -165,19 +192,21 @@ def test_endpoint_is_idempotent( assert verify_webmention.call_args_list == [call(identifier), call(identifier)] -@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 + app: Flask, + client: FlaskClient, + views_module: ModuleType, + monkeypatch: pytest.MonkeyPatch, ) -> None: - source = "https://source.example/post" - target = "https://dennisfink.me/blog/example/" + mock_endpoint_form(views_module, monkeypatch, valid=True) + verify_webmention = Mock() + monkeypatch.setattr(views_module, "verify_webmention", verify_webmention) with app.app_context(): existing = ReceivedWebmention( uuid=uuid.uuid7(), - source=source, - target=target, + source=SOURCE_URL, + target=TARGET_URL, status="failed", failure_reason="Previous failure", ) @@ -187,7 +216,7 @@ def test_resending_resets_failure_state( identifier = existing.uuid - response = client.post("/endpoint", data={"source": source, "target": target}) + response = client.post("/endpoint") assert response.status_code == 201 @@ -252,7 +281,5 @@ 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