path: root/tests/test_views.py

# SPDX-FileCopyrightText: 2026 Dennis Fink <me+coding@dennisfink.me>
#
# SPDX-License-Identifier: BSD-3-Clause

import uuid
from datetime import UTC, datetime
from types import ModuleType
from unittest.mock import ANY, Mock, call

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,
    SentWebmention,
    SentWebmentionStatus,
    Source,
    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():
        user = User(username="admin", password="test")

        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(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


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)

    response = client.post("/endpoint")

    assert response.status_code == 400
    assert response.get_json() == errors


def test_endpoint_creates_webmention(
    app: Flask,
    client: FlaskClient,
    views_module: ModuleType,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    form = mock_endpoint_form(views_module, monkeypatch, valid=True)
    verify_webmention = Mock()
    monkeypatch.setattr(views_module, "verify_webmention", verify_webmention)

    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_URL
        assert webmention.target == TARGET_URL
        assert webmention.status == "received"
        assert webmention.failure_reason is None

        identifier = webmention.uuid

    assert response.headers["Location"].endswith(str(identifier))
    verify_webmention.assert_called_once_with(identifier)


def test_endpoint_is_idempotent(
    app: Flask,
    client: FlaskClient,
    views_module: ModuleType,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    mock_endpoint_form(views_module, monkeypatch, valid=True)
    verify_webmention = Mock()
    monkeypatch.setattr(views_module, "verify_webmention", verify_webmention)

    first_response = client.post("/endpoint")
    second_response = client.post("/endpoint")

    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 == SOURCE_URL
        assert webmention.target == TARGET_URL
        assert webmention.status == "received"
        assert webmention.failure_reason is None

        identifier = webmention.uuid

    assert verify_webmention.call_args_list == [call(identifier), call(identifier)]


def test_resending_resets_failure_state(
    app: Flask,
    client: FlaskClient,
    views_module: ModuleType,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    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_URL,
            target=TARGET_URL,
            status="failed",
            failure_reason="Previous failure",
        )

        db.session.add(existing)
        db.session.commit()

        identifier = existing.uuid

    response = client.post("/endpoint")

    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

    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


def mock_login_form(
    views_module: ModuleType,
    monkeypatch: pytest.MonkeyPatch,
    *,
    valid: bool,
    username: str = "admin",
    password: str = "test",
) -> Mock:
    form = Mock()
    form.validate_on_submit.return_value = valid
    form.username.data = username
    form.password.data = password

    monkeypatch.setattr(views_module.forms, "LoginForm", Mock(return_value=form))
    return form


def mock_admin_action_form(
    views_module: ModuleType, monkeypatch: pytest.MonkeyPatch, *, valid: bool
) -> Mock:
    form = Mock()
    form.validate_on_submit.return_value = valid

    monkeypatch.setattr(views_module.forms, "AdminActionForm", Mock(return_value=form))
    return form


def create_received_webmention(
    app: Flask,
    *,
    source: str = SOURCE_URL,
    target: str = TARGET_URL,
    status: str = "verified",
    failure_reason: str | None = None,
) -> uuid.UUID:
    with app.app_context():
        webmention = ReceivedWebmention(
            uuid=uuid.uuid7(),
            source=source,
            target=target,
            status=status,
            failure_reason=failure_reason,
        )

        db.session.add(webmention)
        db.session.commit()

        return webmention.uuid


def create_resendable_sent_webmention(app: Flask) -> tuple[uuid.UUID, uuid.UUID]:
    now = datetime.now(UTC)

    with app.app_context():
        source = Source(
            path="blog/resend/index.html",
            url="https://dennisfink.me/blog/resend/",
            content_hash="1" * 32,
            revision=3,
            last_seen_at=now,
            revised_at=now,
        )
        webmention = SentWebmention(
            target="https://example.com/resend",
            active=True,
            desired_revision=3,
            processed_revision=3,
            sent_revision=2,
            status=SentWebmentionStatus.FAILED,
            failure_reason="Previous failure",
            endpoint="https://example.com/webmention",
            response_status=500,
            status_url="https://example.com/webmention/status/123",
        )
        source.sent_webmentions.append(webmention)

        db.session.add(source)
        db.session.commit()

        return source.uuid, webmention.uuid


def test_login_renders_form(client: FlaskClient) -> None:
    response = client.get("/login")

    assert response.status_code == 200
    assert "login" in response.get_data(as_text=True).lower()


def test_login_rejects_invalid_credentials(
    app: Flask,
    client: FlaskClient,
    views_module: ModuleType,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    with app.app_context():
        db.session.add(User(username="admin", password="test"))
        db.session.commit()

    mock_login_form(
        views_module,
        monkeypatch,
        valid=True,
        username="admin",
        password="wrong-password",
    )

    response = client.post("/login")

    assert response.status_code == 302
    assert response.headers["Location"].endswith("/login")

    with client.session_transaction() as session:
        assert "_user_id" not in session
        assert ("danger", "Invalid username or password") in session["_flashes"]


def test_login_redirects_to_local_next_page(
    app: Flask,
    client: FlaskClient,
    views_module: ModuleType,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    with app.app_context():
        user = User(username="admin", password="test")
        db.session.add(user)
        db.session.commit()
        user_id = user.id

    mock_login_form(views_module, monkeypatch, valid=True)

    response = client.post("/login?next=/received?page=2")

    assert response.status_code == 302
    assert response.headers["Location"].endswith("/received?page=2")

    with client.session_transaction() as session:
        assert session["_user_id"] == str(user_id)


def test_login_rejects_external_next_page(
    app: Flask,
    client: FlaskClient,
    views_module: ModuleType,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    with app.app_context():
        db.session.add(User(username="admin", password="test"))
        db.session.commit()

    mock_login_form(views_module, monkeypatch, valid=True)

    response = client.post("/login?next=https://evil.example/phishing")

    assert response.status_code == 302
    assert response.headers["Location"].endswith("/")
    assert "evil.example" not in response.headers["Location"]


def test_login_redirects_authenticated_user(app: Flask, client: FlaskClient) -> None:
    log_in(app, client)

    response = client.get("/login")

    assert response.status_code == 302
    assert response.headers["Location"].endswith("/")


def test_logout_ends_session(app: Flask, client: FlaskClient) -> None:
    log_in(app, client)

    response = client.get("/logout")

    assert response.status_code == 302
    assert response.headers["Location"].endswith("/")

    protected_response = client.get("/received")
    assert protected_response.status_code == 302
    assert "/login" in protected_response.headers["Location"]


def test_received_requires_login(client: FlaskClient) -> None:
    response = client.get("/received")

    assert response.status_code == 302
    assert "/login" in response.headers["Location"]


def test_received_lists_webmentions(app: Flask, client: FlaskClient) -> None:
    log_in(app, client)
    create_received_webmention(app)

    response = client.get("/received")

    assert response.status_code == 200
    html = response.get_data(as_text=True)
    assert SOURCE_URL in html
    assert TARGET_URL in html


@pytest.mark.parametrize("action", ["delete", "reverify"])
def test_received_admin_actions_require_login(client: FlaskClient, action: str) -> None:
    response = client.post(f"/received/{uuid.uuid7()}/{action}")

    assert response.status_code == 302
    assert "/login" in response.headers["Location"]


@pytest.mark.parametrize("action", ["delete", "reverify"])
def test_received_admin_actions_reject_invalid_form(
    app: Flask,
    client: FlaskClient,
    views_module: ModuleType,
    monkeypatch: pytest.MonkeyPatch,
    action: str,
) -> None:
    log_in(app, client)
    identifier = create_received_webmention(app)
    form = mock_admin_action_form(views_module, monkeypatch, valid=False)

    response = client.post(f"/received/{identifier}/{action}")

    assert response.status_code == 400
    form.validate_on_submit.assert_called_once_with()


@pytest.mark.parametrize("action", ["delete", "reverify"])
def test_received_admin_actions_return_404_for_unknown_webmention(
    app: Flask,
    client: FlaskClient,
    views_module: ModuleType,
    monkeypatch: pytest.MonkeyPatch,
    action: str,
) -> None:
    log_in(app, client)
    mock_admin_action_form(views_module, monkeypatch, valid=True)

    response = client.post(f"/received/{uuid.uuid7()}/{action}")

    assert response.status_code == 404


def test_delete_received_webmention(
    app: Flask,
    client: FlaskClient,
    views_module: ModuleType,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    log_in(app, client)
    identifier = create_received_webmention(app)
    mock_admin_action_form(views_module, monkeypatch, valid=True)

    response = client.post(f"/received/{identifier}/delete?page=3")

    assert response.status_code == 302
    assert response.headers["Location"].endswith("/received?page=3")

    with app.app_context():
        assert db.session.get(ReceivedWebmention, identifier) is None


def test_reverify_received_webmention(
    app: Flask,
    client: FlaskClient,
    views_module: ModuleType,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    log_in(app, client)
    identifier = create_received_webmention(
        app, status="failed", failure_reason="Previous failure"
    )
    mock_admin_action_form(views_module, monkeypatch, valid=True)
    verify_webmention = Mock()
    monkeypatch.setattr(views_module, "verify_webmention", verify_webmention)

    response = client.post(f"/received/{identifier}/reverify?page=2")

    assert response.status_code == 302
    assert response.headers["Location"].endswith("/received?page=2")

    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

    verify_webmention.assert_called_once_with(identifier)


def test_rescan_sent_sources_requires_login(client: FlaskClient) -> None:
    response = client.post("/sent/rescan")

    assert response.status_code == 302
    assert "/login" in response.headers["Location"]


@pytest.mark.parametrize("valid", [False, True])
def test_rescan_sent_sources(
    app: Flask,
    client: FlaskClient,
    views_module: ModuleType,
    monkeypatch: pytest.MonkeyPatch,
    valid: bool,
) -> None:
    log_in(app, client)
    form = mock_admin_action_form(views_module, monkeypatch, valid=valid)
    manual_scan_sources = Mock()
    monkeypatch.setattr(views_module, "manual_scan_sources", manual_scan_sources)

    response = client.post("/sent/rescan")

    assert response.status_code == 302
    assert response.headers["Location"].endswith("/sent")
    form.validate_on_submit.assert_called_once_with()

    if valid:
        manual_scan_sources.assert_called_once_with()
    else:
        manual_scan_sources.assert_not_called()


def test_resend_sent_webmention_requires_login(client: FlaskClient) -> None:
    response = client.post(f"/sent/{uuid.uuid7()}/{uuid.uuid7()}/resend")

    assert response.status_code == 302
    assert "/login" in response.headers["Location"]


def test_resend_sent_webmention_rejects_invalid_form(
    app: Flask,
    client: FlaskClient,
    views_module: ModuleType,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    log_in(app, client)
    source_identifier, webmention_identifier = create_resendable_sent_webmention(app)
    mock_admin_action_form(views_module, monkeypatch, valid=False)

    response = client.post(f"/sent/{source_identifier}/{webmention_identifier}/resend")

    assert response.status_code == 400


def test_resend_sent_webmention_returns_404_for_unknown_webmention(
    app: Flask,
    client: FlaskClient,
    views_module: ModuleType,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    log_in(app, client)
    source_identifier = create_sent_source(app)
    mock_admin_action_form(views_module, monkeypatch, valid=True)

    response = client.post(f"/sent/{source_identifier}/{uuid.uuid7()}/resend")

    assert response.status_code == 404


def test_resend_sent_webmention_rejects_wrong_source(
    app: Flask,
    client: FlaskClient,
    views_module: ModuleType,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    log_in(app, client)
    _, webmention_identifier = create_resendable_sent_webmention(app)
    wrong_source_identifier = create_sent_source(app)
    mock_admin_action_form(views_module, monkeypatch, valid=True)

    response = client.post(
        f"/sent/{wrong_source_identifier}/{webmention_identifier}/resend"
    )

    assert response.status_code == 404


def test_resend_sent_webmention_reopens_current_revision(
    app: Flask,
    client: FlaskClient,
    views_module: ModuleType,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    log_in(app, client)
    source_identifier, webmention_identifier = create_resendable_sent_webmention(app)
    mock_admin_action_form(views_module, monkeypatch, valid=True)
    send_webmention = Mock()
    monkeypatch.setattr(views_module, "send_webmention", send_webmention)

    response = client.post(f"/sent/{source_identifier}/{webmention_identifier}/resend")

    assert response.status_code == 302
    assert response.headers["Location"].endswith(f"/sent/{source_identifier}")

    with app.app_context():
        webmention = db.session.get(SentWebmention, webmention_identifier)
        assert webmention is not None
        assert webmention.desired_revision == 3
        assert webmention.processed_revision is None
        assert webmention.sent_revision == 2
        assert webmention.status is None
        assert webmention.failure_reason is None
        assert webmention.endpoint is None
        assert webmention.response_status is None
        assert webmention.status_url is None

    send_webmention.assert_called_once_with(webmention_identifier)


def test_status_returns_404_for_unknown_webmention(client: FlaskClient) -> None:
    response = client.get(f"/status/{uuid.uuid7()}")

    assert response.status_code == 404


def test_status_displays_webmention(app: Flask, client: FlaskClient) -> None:
    identifier = create_received_webmention(
        app, status="failed", failure_reason="Verification failed"
    )

    response = client.get(f"/status/{identifier}")

    assert response.status_code == 200
    html = response.get_data(as_text=True)
    assert SOURCE_URL in html
    assert TARGET_URL in html
    assert "failed" in html
    assert "Verification failed" in html


def test_endpoint_recovers_from_concurrent_insert(
    client: FlaskClient, views_module: ModuleType, monkeypatch: pytest.MonkeyPatch
) -> None:
    mock_endpoint_form(views_module, monkeypatch, valid=True)

    competing_identifier = uuid.uuid7()
    competing_webmention = ReceivedWebmention(
        uuid=competing_identifier,
        source=SOURCE_URL,
        target=TARGET_URL,
        status="failed",
        failure_reason="Competing request failed",
    )

    initial_result = Mock()
    initial_result.scalar_one_or_none.return_value = None
    competing_result = Mock()
    competing_result.scalar_one.return_value = competing_webmention

    execute = Mock(side_effect=[initial_result, competing_result])
    add = Mock()
    rollback = Mock()
    commit = Mock(
        side_effect=[
            sa.exc.IntegrityError(
                "INSERT INTO received_webmention ...",
                {},
                Exception("duplicate source/target"),
            ),
            None,
        ]
    )
    verify_webmention = Mock()

    monkeypatch.setattr(views_module.db.session, "execute", execute)
    monkeypatch.setattr(views_module.db.session, "add", add)
    monkeypatch.setattr(views_module.db.session, "rollback", rollback)
    monkeypatch.setattr(views_module.db.session, "commit", commit)
    monkeypatch.setattr(views_module, "verify_webmention", verify_webmention)

    response = client.post("/endpoint")

    assert response.status_code == 201
    assert response.headers["Location"].endswith(f"/status/{competing_identifier}")

    assert execute.call_count == 2
    add.assert_called_once_with(ANY)
    rollback.assert_called_once_with()
    assert commit.call_count == 2

    assert competing_webmention.status == "received"
    assert competing_webmention.failure_reason is None
    verify_webmention.assert_called_once_with(competing_identifier)