path: root/tests/tasks/test_sender.py

import logging
import uuid
from datetime import datetime, timezone
from types import ModuleType
from unittest.mock import MagicMock, Mock

import httpx
import pytest
from flask import Flask

from webmentions_ssg import DATABASE as db
from webmentions_ssg.models import SentWebmention, SentWebmentionStatus, Source

SOURCE_URL = "https://dennisfink.me/blog/example/"
TARGET_URL = "https://example.com/post"
ENDPOINT_URL = "https://example.com/webmention"
STATUS_URL = "https://example.com/webmention/status/123"


def create_sent_webmention(
    app: Flask,
    *,
    active: bool = True,
    desired_revision: int = 1,
    processed_revision: int | None = None,
    sent_revision: int | None = None,
    status: SentWebmentionStatus | None = None,
    failure_reason: str | None = None,
    endpoint: str | None = None,
    response_status: int | None = None,
    status_url: str | None = None,
) -> uuid.UUID:
    now = datetime.now(timezone.utc)

    with app.app_context():
        source = Source(
            path="example/index.html",
            url=SOURCE_URL,
            content_hash="0" * 32,
            revision=desired_revision,
            last_seen_at=now,
            revised_at=now,
        )

        webmention = SentWebmention(
            target=TARGET_URL,
            active=active,
            desired_revision=desired_revision,
            processed_revision=processed_revision,
            sent_revision=sent_revision,
            status=status,
            failure_reason=failure_reason,
            endpoint=endpoint,
            response_status=response_status,
            status_url=status_url,
        )

        source.sent_webmentions.append(webmention)

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

        return webmention.uuid


def mock_sender_requests(
    sender_module: ModuleType,
    monkeypatch: pytest.MonkeyPatch,
    *,
    endpoint: str | None = ENDPOINT_URL,
    result: tuple[int, str | None] = (202, None),
) -> tuple[Mock, Mock, Mock, Mock]:
    httpx_client = MagicMock()
    client = httpx_client.return_value.__enter__.return_value

    discover = Mock(return_value=endpoint)
    post = Mock(return_value=result)

    monkeypatch.setattr(sender_module.httpx, "Client", httpx_client)
    monkeypatch.setattr(sender_module, "discover_webmention_endpoint", discover)
    monkeypatch.setattr(sender_module, "post_webmention", post)

    return httpx_client, client, discover, post


@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(
    ("value", "expected"),
    (
        (
            '<https://example.com/webmention>; rel="webmention"',
            ("https://example.com/webmention", {"webmention"}),
        ),
        (
            "<https://example.com/webmention>; rel=webmention",
            ("https://example.com/webmention", {"webmention"}),
        ),
        (
            '<https://example.com/webmention>; rel="webmention alternate"',
            ("https://example.com/webmention", {"webmention", "alternate"}),
        ),
        (
            '<https://example.com/webmention>; REL="WebMention Alternate"',
            ("https://example.com/webmention", {"webmention", "alternate"}),
        ),
        (
            "<https://example.com/webmention>; "
            'type="text/html"; rel="webmention"; title="Endpoint"',
            ("https://example.com/webmention", {"webmention"}),
        ),
        (
            '<https://example.com/stylesheet>; type="text/css"',
            ("https://example.com/stylesheet", set()),
        ),
    ),
)
def test_parse_link_value(
    sender_module: ModuleType, value: str, expected: tuple[str, set[str]]
) -> None:
    assert sender_module.parse_link_value(value) == expected


@pytest.mark.parametrize(
    "value", ("", "https://example.com/webmention", "<https://example.com/webmention")
)
def test_parse_link_value_rejects_malformed_value(
    sender_module: ModuleType, value: str
) -> None:
    assert sender_module.parse_link_value(value) is None


@pytest.mark.parametrize(
    ("target", "headers", "html", "expected_endpoint"),
    (
        pytest.param(
            "https://example.com/test/1",
            (("Link", "</test/1/webmention?head=true>; rel=webmention"),),
            "",
            "https://example.com/test/1/webmention?head=true",
            id="1-http-link-unquoted-rel-relative-url",
        ),
        pytest.param(
            "https://example.com/test/2",
            (
                (
                    "Link",
                    "<https://example.com/test/2/webmention?head=true>; rel=webmention",
                ),
            ),
            "",
            "https://example.com/test/2/webmention?head=true",
            id="2-http-link-unquoted-rel-absolute-url",
        ),
        pytest.param(
            "https://example.com/test/3",
            (),
            """
            <link
              rel="webmention"
              href="/test/3/webmention"
            >
            """,
            "https://example.com/test/3/webmention",
            id="3-html-link-relative-url",
        ),
        pytest.param(
            "https://example.com/test/4",
            (),
            """
            <link
              rel="webmention"
              href="https://example.com/test/4/webmention"
            >
            """,
            "https://example.com/test/4/webmention",
            id="4-html-link-absolute-url",
        ),
        pytest.param(
            "https://example.com/test/5",
            (),
            """
            <a
              rel="webmention"
              href="/test/5/webmention"
            >
              Endpoint
            </a>
            """,
            "https://example.com/test/5/webmention",
            id="5-html-a-relative-url",
        ),
        pytest.param(
            "https://example.com/test/6",
            (),
            """
            <a
              rel="webmention"
              href="https://example.com/test/6/webmention"
            >
              Endpoint
            </a>
            """,
            "https://example.com/test/6/webmention",
            id="6-html-a-absolute-url",
        ),
        pytest.param(
            "https://example.com/test/7",
            (
                (
                    "LinK",
                    "<https://example.com/test/7/webmention?head=true>; rel=webmention",
                ),
            ),
            "",
            "https://example.com/test/7/webmention?head=true",
            id="7-http-link-strange-casing",
        ),
        pytest.param(
            "https://example.com/test/8",
            (
                (
                    "Link",
                    '<https://example.com/test/8/webmention?head=true>; rel="webmention"',
                ),
            ),
            "",
            "https://example.com/test/8/webmention?head=true",
            id="8-http-link-quoted-rel",
        ),
        pytest.param(
            "https://example.com/test/9",
            (),
            """
            <link
              rel="webmention somethingelse"
              href="https://example.com/test/9/webmention"
            >
            """,
            "https://example.com/test/9/webmention",
            id="9-multiple-html-rel-values",
        ),
        pytest.param(
            "https://example.com/test/10",
            (
                (
                    "Link",
                    "<https://example.com/test/10/webmention?head=true>; "
                    'rel="webmention somethingelse"',
                ),
            ),
            "",
            "https://example.com/test/10/webmention?head=true",
            id="10-multiple-link-header-rel-values",
        ),
        pytest.param(
            "https://example.com/test/11",
            (("Link", '</test/11/webmention>; rel="webmention"'),),
            """
            <link
              rel="webmention"
              href="/test/11/webmention/error"
            >

            <a
              rel="webmention"
              href="/test/11/webmention/error"
            >
              Wrong endpoint
            </a>
            """,
            "https://example.com/test/11/webmention",
            id="11-http-link-precedence",
        ),
        pytest.param(
            "https://example.com/test/12",
            (),
            """
            <link
              rel="not-webmention"
              href="/test/12/webmention/error"
            >

            <a
              rel="webmention"
              href="/test/12/webmention"
            >
              Correct endpoint
            </a>
            """,
            "https://example.com/test/12/webmention",
            id="12-exact-rel-match",
        ),
        pytest.param(
            "https://example.com/test/13",
            (),
            """
            <!--
              <a
                rel="webmention"
                href="/test/13/webmention/error"
              >
                False endpoint
              </a>
            -->

            <a
              rel="webmention"
              href="/test/13/webmention"
            >
              Correct endpoint
            </a>
            """,
            "https://example.com/test/13/webmention",
            id="13-endpoint-inside-html-comment",
        ),
        pytest.param(
            "https://example.com/test/14",
            (),
            """
            <code>
              &lt;a
                rel="webmention"
                href="/test/14/webmention/error"
              &gt;
                False endpoint
              &lt;/a&gt;
            </code>

            <a
              rel="webmention"
              href="/test/14/webmention"
            >
              Correct endpoint
            </a>
            """,
            "https://example.com/test/14/webmention",
            id="14-endpoint-in-escaped-html",
        ),
        pytest.param(
            "https://example.com/test/15",
            (),
            """
            <link
              rel="webmention"
              href=""
            >
            """,
            "https://example.com/test/15",
            id="15-empty-href",
        ),
        pytest.param(
            "https://example.com/test/16",
            (),
            """
            <a
              rel="webmention"
              href="/test/16/webmention"
            >
              First endpoint
            </a>

            <link
              rel="webmention"
              href="/test/16/webmention/error"
            >
            """,
            "https://example.com/test/16/webmention",
            id="16-a-before-link",
        ),
        pytest.param(
            "https://example.com/test/17",
            (),
            """
            <link
              rel="webmention"
              href="/test/17/webmention"
            >

            <a
              rel="webmention"
              href="/test/17/webmention/error"
            >
              Second endpoint
            </a>
            """,
            "https://example.com/test/17/webmention",
            id="17-link-before-a",
        ),
        pytest.param(
            "https://example.com/test/18",
            (
                ("Link", '<https://example.com/test/18/webmention/error>; rel="other"'),
                (
                    "Link",
                    "<https://example.com/test/18/webmention?head=true>; "
                    'rel="webmention"',
                ),
            ),
            "",
            "https://example.com/test/18/webmention?head=true",
            id="18-multiple-http-link-headers",
        ),
        pytest.param(
            "https://example.com/test/19",
            (
                (
                    "Link",
                    "<https://example.com/test/19/webmention/error>; "
                    'rel="other", '
                    "<https://example.com/test/19/webmention?head=true>; "
                    'rel="webmention"',
                ),
            ),
            "",
            "https://example.com/test/19/webmention?head=true",
            id="19-single-header-multiple-values",
        ),
        pytest.param(
            "https://example.com/test/20",
            (),
            """
            <link rel="webmention">

            <a
              rel="webmention"
              href="/test/20/webmention"
            >
              Correct endpoint
            </a>
            """,
            "https://example.com/test/20/webmention",
            id="20-link-without-href",
        ),
        pytest.param(
            "https://example.com/test/21",
            (),
            """
            <link
              rel="webmention"
              href="/test/21/webmention?query=yes"
            >
            """,
            "https://example.com/test/21/webmention?query=yes",
            id="21-endpoint-query-string",
        ),
        pytest.param(
            "https://example.com/test/22",
            (),
            """
            <link
              rel="webmention"
              href="22/webmention"
            >
            """,
            "https://example.com/test/22/webmention",
            id="22-endpoint-relative-to-path",
        ),
    ),
)
def test_discovery(
    app: Flask,
    sender_module: ModuleType,
    target: str,
    headers: tuple[tuple[str, str], ...],
    html: str,
    expected_endpoint: str,
) -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        if request.method == "HEAD":
            return httpx.Response(200, headers=headers)

        return httpx.Response(200, headers={"Content-Type": "text/html"}, text=html)

    with (
        app.app_context(),
        httpx.Client(
            transport=httpx.MockTransport(handler), follow_redirects=True
        ) as client,
    ):
        endpoint = sender_module.discover_webmention_endpoint(client, target)

    assert endpoint == expected_endpoint


@pytest.mark.parametrize(
    ("target", "responses", "expected_endpoint"),
    (
        pytest.param(
            "https://example.com/test/23/page",
            {
                "/test/23/page": (302, {"Location": "page/redirect-key"}),
                "/test/23/page/redirect-key": (
                    200,
                    {"Link": ("<webmention-endpoint/endpoint-key>; rel=webmention")},
                ),
            },
            ("https://example.com/test/23/page/webmention-endpoint/endpoint-key"),
            id="relative-endpoint-after-redirect",
        ),
    ),
)
def test_discovery_redirect(
    app: Flask,
    sender_module: ModuleType,
    target: str,
    responses: dict[str, tuple[int, dict[str, str]]],
    expected_endpoint: str,
) -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        try:
            status, headers = responses[request.url.path]
        except KeyError:
            raise AssertionError(f"Unexpected request: {request.url}") from None

        return httpx.Response(status, headers=headers)

    with (
        app.app_context(),
        httpx.Client(
            transport=httpx.MockTransport(handler), follow_redirects=True
        ) as client,
    ):
        endpoint = sender_module.discover_webmention_endpoint(client, target)

    assert endpoint == expected_endpoint


def test_discovery_stops_after_head(app: Flask, sender_module: ModuleType) -> None:
    requests: list[httpx.Request] = []

    def handler(request: httpx.Request) -> httpx.Response:
        requests.append(request)

        return httpx.Response(200, headers={"Link": "</webmention>; rel=webmention"})

    with (
        app.app_context(),
        httpx.Client(
            transport=httpx.MockTransport(handler), follow_redirects=True
        ) as client,
    ):
        endpoint = sender_module.discover_webmention_endpoint(
            client, "https://example.com/post"
        )

    assert endpoint == "https://example.com/webmention"
    assert [request.method for request in requests] == ["HEAD"]


def test_discovery_falls_back_to_get(app: Flask, sender_module: ModuleType) -> None:
    requests: list[httpx.Request] = []

    def handler(request: httpx.Request) -> httpx.Response:
        requests.append(request)

        if request.method == "HEAD":
            return httpx.Response(200)

        return httpx.Response(
            200,
            headers={"Content-Type": "text/html"},
            text="""
            <link
              rel="webmention"
              href="/webmention"
            >
            """,
        )

    with (
        app.app_context(),
        httpx.Client(
            transport=httpx.MockTransport(handler), follow_redirects=True
        ) as client,
    ):
        endpoint = sender_module.discover_webmention_endpoint(
            client, "https://example.com/post"
        )

    assert endpoint == "https://example.com/webmention"
    assert [request.method for request in requests] == ["HEAD", "GET"]


def test_send_webmention_persists_success(
    app: Flask, sender_module: ModuleType, monkeypatch: pytest.MonkeyPatch
) -> None:
    identifier = create_sent_webmention(app)

    _, client, discover, post = mock_sender_requests(
        sender_module, monkeypatch, result=(201, STATUS_URL)
    )

    sender_module.send_webmention.call_local(identifier)

    discover.assert_called_once_with(client, TARGET_URL)
    post.assert_called_once_with(
        client, endpoint=ENDPOINT_URL, source=SOURCE_URL, target=TARGET_URL
    )

    with app.app_context():
        webmention = db.session.get(SentWebmention, identifier)

        assert webmention is not None
        assert webmention.processed_revision == 1
        assert webmention.sent_revision == 1
        assert webmention.status == SentWebmentionStatus.SENT
        assert webmention.failure_reason is None
        assert webmention.endpoint == ENDPOINT_URL
        assert webmention.response_status == 201
        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(
    app: Flask, sender_module: ModuleType, monkeypatch: pytest.MonkeyPatch
) -> None:
    identifier = create_sent_webmention(
        app,
        desired_revision=2,
        processed_revision=1,
        sent_revision=1,
        status=SentWebmentionStatus.SENT,
        endpoint="https://old.example/webmention",
        response_status=201,
        status_url="https://old.example/status/123",
    )

    _, client, discover, post = mock_sender_requests(
        sender_module, monkeypatch, endpoint=None
    )

    sender_module.send_webmention.call_local(identifier)

    discover.assert_called_once_with(client, TARGET_URL)
    post.assert_not_called()

    with app.app_context():
        webmention = db.session.get(SentWebmention, identifier)

        assert webmention is not None
        assert webmention.processed_revision == 2
        assert webmention.sent_revision == 1
        assert webmention.status == SentWebmentionStatus.UNSUPPORTED
        assert webmention.failure_reason == "No Webmention endpoint discovered"
        assert webmention.endpoint is None
        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(
    app: Flask, sender_module: ModuleType, monkeypatch: pytest.MonkeyPatch
) -> None:
    identifier = create_sent_webmention(
        app,
        desired_revision=2,
        processed_revision=1,
        sent_revision=1,
        status=SentWebmentionStatus.SENT,
        status_url="https://old.example/status/123",
    )

    mock_sender_requests(sender_module, monkeypatch, result=(400, None))

    sender_module.send_webmention.call_local(identifier)

    with app.app_context():
        webmention = db.session.get(SentWebmention, identifier)

        assert webmention is not None
        assert webmention.processed_revision == 2
        assert webmention.sent_revision == 1
        assert webmention.status == SentWebmentionStatus.FAILED
        assert webmention.failure_reason == "Webmention endpoint returned HTTP 400"
        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(
    app: Flask, sender_module: ModuleType, monkeypatch: pytest.MonkeyPatch
) -> None:
    identifier = create_sent_webmention(
        app,
        desired_revision=2,
        processed_revision=1,
        sent_revision=1,
        status=SentWebmentionStatus.SENT,
        status_url="https://old.example/status/123",
    )

    mock_sender_requests(sender_module, monkeypatch, result=(503, None))

    with pytest.raises(
        sender_module.TemporarySenderError,
        match="Webmention endpoint returned HTTP 503",
    ):
        sender_module.send_webmention.call_local(identifier)

    with app.app_context():
        webmention = db.session.get(SentWebmention, identifier)

        assert webmention is not None
        assert webmention.processed_revision == 1
        assert webmention.sent_revision == 1
        assert webmention.status == SentWebmentionStatus.FAILED
        assert webmention.failure_reason == "Webmention endpoint returned HTTP 503"
        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(
    app: Flask, sender_module: ModuleType, monkeypatch: pytest.MonkeyPatch
) -> None:
    identifier = create_sent_webmention(
        app,
        desired_revision=2,
        processed_revision=2,
        sent_revision=2,
        status=SentWebmentionStatus.SENT,
    )

    httpx_client, _, discover, post = mock_sender_requests(sender_module, monkeypatch)

    sender_module.send_webmention.call_local(identifier)

    httpx_client.assert_not_called()
    discover.assert_not_called()
    post.assert_not_called()


def test_send_webmention_ignores_result_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 (202, 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


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