path: root/tests/tasks/test_receiver.py

import logging
import uuid
from types import ModuleType
from unittest.mock import Mock, patch

import httpx
import pytest
from flask import Flask

from webmentions_ssg import DATABASE as db
from webmentions_ssg.models import ReceivedWebmention
from webmentions_ssg.url_security import AddressResolutionError

SOURCE_URL = "https://source.example/article"
TARGET_URL = "https://dennisfink.me/blog/example/"


@pytest.fixture
def receiver(app: Flask) -> ModuleType:
    from webmentions_ssg.tasks import receiver

    return receiver


def create_webmention(
    app: Flask,
    *,
    source: str = SOURCE_URL,
    target: str = TARGET_URL,
    status: str = "received",
    failure_reason: str | None = None,
) -> uuid.UUID:
    identifier = uuid.uuid7()

    with app.app_context():
        webmention = ReceivedWebmention(
            uuid=identifier,
            source=source,
            target=target,
            status=status,
            failure_reason=failure_reason,
        )

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

    return identifier


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)


def set_stream_response(httpx_client: Mock, response: httpx.Response) -> Mock:
    client = httpx_client.return_value.__enter__.return_value
    client.stream.return_value.__enter__.return_value = response
    return client


@pytest.mark.parametrize(
    ("body", "expected"),
    [
        pytest.param(f'<a href="{TARGET_URL}">Reply</a>', True, id="a-href"),
        pytest.param(f'<area href="{TARGET_URL}" alt="Target">', True, id="area-href"),
        pytest.param(
            f'<link href="{TARGET_URL}" rel="alternate">', True, id="link-href"
        ),
        pytest.param(f'<img src="{TARGET_URL}" alt="">', True, id="img-src"),
        pytest.param(f'<audio src="{TARGET_URL}"></audio>', True, id="audio-src"),
        pytest.param(f'<video src="{TARGET_URL}"></video>', True, id="video-src"),
        pytest.param(
            f'<audio><source src="{TARGET_URL}"></audio>', True, id="audio-source-src"
        ),
        pytest.param(
            f'<video><source src="{TARGET_URL}"></video>', True, id="video-source-src"
        ),
        pytest.param(f'<iframe src="{TARGET_URL}"></iframe>', True, id="iframe-src"),
        pytest.param(f'<embed src="{TARGET_URL}">', True, id="embed-src"),
        pytest.param(f'<script src="{TARGET_URL}"></script>', True, id="script-src"),
        pytest.param(
            f'<video><track src="{TARGET_URL}"></video>', True, id="track-src"
        ),
        pytest.param(
            f'<input type="image" src="{TARGET_URL}" alt="">',
            True,
            id="image-input-src",
        ),
        pytest.param(
            f'<input type="IMAGE" src="{TARGET_URL}" alt="">',
            True,
            id="image-input-case-insensitive",
        ),
        pytest.param(
            f'<blockquote cite="{TARGET_URL}">Quotation</blockquote>',
            True,
            id="blockquote-cite",
        ),
        pytest.param(f'<q cite="{TARGET_URL}">Quotation</q>', True, id="q-cite"),
        pytest.param(f'<ins cite="{TARGET_URL}">Addition</ins>', True, id="ins-cite"),
        pytest.param(f'<del cite="{TARGET_URL}">Removal</del>', True, id="del-cite"),
        pytest.param(
            '<base href="https://dennisfink.me/blog/"><a href="example/">Reply</a>',
            True,
            id="base-url",
        ),
        pytest.param(f'<img cite="{TARGET_URL}" alt="">', False, id="img-cite-invalid"),
        pytest.param(
            f'<blockquote src="{TARGET_URL}">Quote</blockquote>',
            False,
            id="blockquote-src-invalid",
        ),
        pytest.param(f'<a src="{TARGET_URL}">Reply</a>', False, id="a-src-invalid"),
        pytest.param(f'<div href="{TARGET_URL}"></div>', False, id="div-href-invalid"),
        pytest.param(f'<link src="{TARGET_URL}">', False, id="link-src-invalid"),
        pytest.param(
            f'<input type="text" src="{TARGET_URL}">',
            False,
            id="text-input-src-invalid",
        ),
        pytest.param(
            f'<picture><source src="{TARGET_URL}"></picture>',
            False,
            id="picture-source-src-invalid",
        ),
        pytest.param(f'<base href="{TARGET_URL}">', False, id="base-is-not-mention"),
        pytest.param(
            f'<a href="{TARGET_URL}more">Different page</a>', False, id="longer-url"
        ),
        pytest.param(
            f'<img cite="{TARGET_URL}" src="https://example.com/image.jpg" alt="">',
            False,
            id="invalid-cite-does-not-override-valid-src",
        ),
        pytest.param(f"<p>{TARGET_URL}</p>", False, id="text-content"),
        pytest.param(
            '<a href="https://example.com/">Other site</a>', False, id="missing-target"
        ),
    ],
)
def test_html_mentions_target(receiver: ModuleType, body: str, expected: bool) -> None:
    assert (
        receiver.html_mentions_target(body.encode(), SOURCE_URL, TARGET_URL) is expected
    )


def test_html_mentions_relative_target(receiver: ModuleType) -> None:
    assert receiver.html_mentions_target(
        b'<a href="../target/">Reply</a>',
        "https://source.example/posts/article/",
        "https://source.example/posts/target/",
    )


@pytest.mark.parametrize(
    ("body", "target_url", "expected"),
    [
        (TARGET_URL, TARGET_URL, True),
        (f"This post replies to {TARGET_URL}", TARGET_URL, True),
        (
            f"https://example.com/first {TARGET_URL} https://example.com/last",
            TARGET_URL,
            True,
        ),
        (f"{TARGET_URL}more", TARGET_URL, False),
        ("https://dennisfink.me/blog/other/", TARGET_URL, False),
        ("/blog/example/", TARGET_URL, False),
        ("There are no links here.", TARGET_URL, False),
    ],
)
def test_text_mentions_target(
    receiver: ModuleType, body: str, target_url: str, expected: bool
) -> None:
    assert receiver.text_mentions_target(body, target_url) is expected


@patch("webmentions_ssg.tasks.receiver.httpx.Client")
def test_fetch_source_returns_response_and_body(
    httpx_client: Mock, app: Flask, receiver: ModuleType
) -> None:
    response = httpx.Response(
        200,
        headers={"Content-Type": "text/html"},
        content=b"<p>Document</p>",
        request=httpx.Request("GET", SOURCE_URL),
    )
    client = set_stream_response(httpx_client, response)

    with app.app_context():
        fetched_response, body = receiver.fetch_source(SOURCE_URL)

    assert fetched_response is response
    assert body == b"<p>Document</p>"
    client.stream.assert_called_once_with("GET", SOURCE_URL)

    options = httpx_client.call_args.kwargs

    assert options["headers"] == {
        "Accept": "text/html, application/xhtml+xml;q=0.9, text/plain;q=0.8",
        "User-Agent": f"{receiver.APP_NAME}/{receiver.VERSION} ReceivedWebmention",
    }
    assert options["follow_redirects"] is True
    assert options["max_redirects"] == app.config.get(
        "WEBMENTIONS_SSG_MAX_REDIRECTS", 20
    )
    assert options["trust_env"] is False
    assert options["event_hooks"] == {"request": [receiver.ensure_public_request]}


@patch("webmentions_ssg.tasks.receiver.httpx.Client")
@pytest.mark.parametrize(
    ("status_code", "exception_name"),
    [
        (400, "VerificationError"),
        (404, "VerificationError"),
        (410, "SourceGoneError"),
        (408, "TemporaryFetchError"),
        (425, "TemporaryFetchError"),
        (429, "TemporaryFetchError"),
        (500, "TemporaryFetchError"),
        (503, "TemporaryFetchError"),
    ],
)
def test_fetch_source_maps_http_status_to_exception(
    httpx_client: Mock,
    app: Flask,
    receiver: ModuleType,
    status_code: int,
    exception_name: str,
) -> None:
    set_stream_response(
        httpx_client,
        httpx.Response(status_code, request=httpx.Request("GET", SOURCE_URL)),
    )

    exception_type = getattr(receiver, exception_name)

    with app.app_context(), pytest.raises(exception_type, match=f"HTTP {status_code}"):
        receiver.fetch_source(SOURCE_URL)


@patch("webmentions_ssg.tasks.receiver.httpx.Client")
def test_fetch_source_propagates_network_error(
    httpx_client: Mock, app: Flask, receiver: ModuleType
) -> None:
    client = httpx_client.return_value.__enter__.return_value
    client.stream.side_effect = httpx.ConnectError(
        "Connection refused", request=httpx.Request("GET", SOURCE_URL)
    )

    with (
        app.app_context(),
        pytest.raises(httpx.ConnectError, match="Connection refused"),
    ):
        receiver.fetch_source(SOURCE_URL)


@patch("webmentions_ssg.tasks.receiver.httpx.Client")
def test_fetch_source_rejects_declared_oversized_body(
    httpx_client: Mock, app: Flask, receiver: ModuleType
) -> None:
    app.config["WEBMENTIONS_SSG_MAX_SOURCE_BYTES"] = 10

    set_stream_response(
        httpx_client,
        httpx.Response(
            200,
            headers={"Content-Type": "text/html", "Content-Length": "11"},
            content=b"x" * 11,
            request=httpx.Request("GET", SOURCE_URL),
        ),
    )

    with (
        app.app_context(),
        pytest.raises(receiver.VerificationError, match="Source document is too large"),
    ):
        receiver.fetch_source(SOURCE_URL)


@patch("webmentions_ssg.tasks.receiver.httpx.Client")
def test_fetch_source_rejects_streamed_oversized_body(
    httpx_client: Mock, app: Flask, receiver: ModuleType
) -> None:
    app.config["WEBMENTIONS_SSG_MAX_SOURCE_BYTES"] = 10

    class BodyStream(httpx.SyncByteStream):
        def __iter__(self):
            yield b"x" * 6
            yield b"x" * 6

    set_stream_response(
        httpx_client,
        httpx.Response(
            200,
            headers={"Content-Type": "text/html"},
            stream=BodyStream(),
            request=httpx.Request("GET", SOURCE_URL),
        ),
    )

    with (
        app.app_context(),
        pytest.raises(receiver.VerificationError, match="Source document is too large"),
    ):
        receiver.fetch_source(SOURCE_URL)


@patch("webmentions_ssg.tasks.receiver.fetch_source")
@pytest.mark.parametrize(
    ("content_type", "body", "expected"),
    [
        pytest.param(
            "text/html; charset=utf-8",
            f'<a href="{TARGET_URL}">Reply</a>'.encode(),
            True,
            id="html",
        ),
        pytest.param(
            "TEXT/HTML; CHARSET=UTF-8",
            f'<a href="{TARGET_URL}">Reply</a>'.encode(),
            True,
            id="case-insensitive-html",
        ),
        pytest.param(
            "application/xhtml+xml",
            b'<a href="https://example.com/">Other</a>',
            False,
            id="xhtml-without-target",
        ),
        pytest.param(
            "text/plain; charset=utf-8",
            f"Reply to {TARGET_URL}".encode(),
            True,
            id="plain-text-utf-8",
        ),
        pytest.param(
            "text/plain; charset=iso-8859-1",
            (f"Grüße. Reply to {TARGET_URL}").encode("iso-8859-1"),
            True,
            id="plain-text-declared-encoding",
        ),
        pytest.param(
            "text/plain; charset=utf-8",
            b"\xff Reply to " + TARGET_URL.encode(),
            True,
            id="plain-text-invalid-byte",
        ),
        pytest.param(
            "text/plain", b"No target here.", False, id="plain-text-without-target"
        ),
    ],
)
def test_source_mentions_target_by_media_type(
    fetch_source: Mock,
    receiver: ModuleType,
    content_type: str,
    body: bytes,
    expected: bool,
) -> None:
    response = httpx.Response(
        200,
        headers={"Content-Type": content_type},
        content=body,
        request=httpx.Request("GET", SOURCE_URL),
    )
    fetch_source.return_value = response, body

    assert receiver.source_mentions_target(SOURCE_URL, TARGET_URL) is expected
    fetch_source.assert_called_once_with(SOURCE_URL)


@patch("webmentions_ssg.tasks.receiver.fetch_source")
@pytest.mark.parametrize(
    ("content_type", "expected_media_type"),
    [
        ("application/json", "application/json"),
        ("application/pdf", "application/pdf"),
        ("", "missing"),
    ],
)
def test_source_mentions_target_rejects_unsupported_media_type(
    fetch_source: Mock,
    receiver: ModuleType,
    content_type: str,
    expected_media_type: str,
) -> None:
    headers = {}

    if content_type:
        headers["Content-Type"] = content_type

    response = httpx.Response(
        200,
        headers=headers,
        content=b"Document",
        request=httpx.Request("GET", SOURCE_URL),
    )
    fetch_source.return_value = response, b"Document"

    with pytest.raises(
        receiver.VerificationError,
        match=f"Unsupported source content type: {expected_media_type}",
    ):
        receiver.source_mentions_target(SOURCE_URL, TARGET_URL)


@patch("webmentions_ssg.tasks.receiver.source_mentions_target")
def test_verify_webmention_marks_row_verifying_before_check(
    source_mentions_target: Mock, app: Flask, receiver: ModuleType
) -> None:
    identifier = create_webmention(
        app, status="failed", failure_reason="Earlier failure"
    )

    def verify_source(source_url: str, target_url: str) -> bool:
        webmention = db.session.get(ReceivedWebmention, identifier)

        assert webmention is not None
        assert webmention.status == "verifying"
        assert webmention.failure_reason is None
        assert source_url == webmention.source
        assert target_url == webmention.target

        return True

    source_mentions_target.side_effect = verify_source

    receiver.verify_webmention.call_local(identifier)

    assert get_webmention_state(app, identifier) == ("verified", None)


@patch("webmentions_ssg.tasks.receiver.source_mentions_target")
@pytest.mark.parametrize(
    ("outcome", "expected_status", "expected_reason"),
    [
        ("verified", "verified", None),
        ("missing", "deleted", "Source does not mention target"),
        ("gone", "deleted", "Source returned HTTP 410"),
        ("permanent-failure", "failed", "Source returned HTTP 404"),
    ],
)
def test_verify_webmention_persists_final_state(
    source_mentions_target: Mock,
    app: Flask,
    receiver: ModuleType,
    outcome: str,
    expected_status: str,
    expected_reason: str | None,
) -> None:
    identifier = create_webmention(app)

    match outcome:
        case "verified":
            source_mentions_target.return_value = True
        case "missing":
            source_mentions_target.return_value = False
        case "gone":
            source_mentions_target.side_effect = receiver.SourceGoneError(
                "Source returned HTTP 410"
            )
        case "permanent-failure":
            source_mentions_target.side_effect = receiver.VerificationError(
                "Source returned HTTP 404"
            )
        case _:
            raise AssertionError(f"Unexpected outcome: {outcome}")

    receiver.verify_webmention.call_local(identifier)

    assert get_webmention_state(app, identifier) == (expected_status, expected_reason)


@patch("webmentions_ssg.tasks.receiver.source_mentions_target")
@pytest.mark.parametrize("failure", ["temporary-http", "network"])
def test_verify_webmention_persists_retryable_failure_and_reraises(
    source_mentions_target: Mock, app: Flask, receiver: ModuleType, failure: str
) -> None:
    identifier = create_webmention(app)

    match failure:
        case "temporary-http":
            exception = receiver.TemporaryFetchError("Source returned HTTP 503")
        case "network":
            exception = httpx.ConnectError(
                "Connection refused", request=httpx.Request("GET", SOURCE_URL)
            )
        case _:
            raise AssertionError(f"Unexpected failure: {failure}")

    source_mentions_target.side_effect = exception

    with pytest.raises(type(exception), match=str(exception)):
        receiver.verify_webmention.call_local(identifier)

    assert get_webmention_state(app, identifier) == ("failed", str(exception))


def test_verify_webmention_ignores_unknown_identifier(
    receiver: ModuleType, caplog: pytest.LogCaptureFixture
) -> None:
    identifier = uuid.uuid7()

    with caplog.at_level(logging.WARNING):
        receiver.verify_webmention.call_local(identifier)

    assert f"Cannot verify unknown ReceivedWebmention {identifier}" in caplog.text


@patch("webmentions_ssg.tasks.receiver.is_public_url", return_value=True)
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)


@patch("webmentions_ssg.tasks.receiver.is_public_url", return_value=False)
def test_ensure_public_request_rejects_non_public_address(
    is_public_url: Mock, receiver: ModuleType
) -> None:
    with pytest.raises(
        receiver.VerificationError, match="Source resolves to a non-public address"
    ):
        receiver.ensure_public_request(httpx.Request("GET", SOURCE_URL))

    is_public_url.assert_called_once_with(SOURCE_URL)


@patch(
    "webmentions_ssg.tasks.receiver.is_public_url",
    side_effect=AddressResolutionError("Could not resolve hostname"),
)
def test_ensure_public_request_maps_dns_failure_to_temporary_error(
    is_public_url: Mock, receiver: ModuleType
) -> None:
    with pytest.raises(
        receiver.TemporaryFetchError, match="Source hostname could not be resolved"
    ):
        receiver.ensure_public_request(httpx.Request("GET", SOURCE_URL))

    is_public_url.assert_called_once_with(SOURCE_URL)


@patch(
    "webmentions_ssg.tasks.receiver.is_public_url",
    side_effect=ValueError("No hostname was specified"),
)
def test_ensure_public_request_rejects_url_without_hostname(
    is_public_url: Mock, receiver: ModuleType
) -> None:
    with pytest.raises(receiver.VerificationError, match="Source URL has no hostname"):
        receiver.ensure_public_request(httpx.Request("GET", SOURCE_URL))

    is_public_url.assert_called_once_with(SOURCE_URL)


@patch("webmentions_ssg.tasks.receiver.BeautifulSoup")
def test_html_mentions_target_ignores_non_string_attribute(
    beautiful_soup: Mock, receiver: ModuleType
) -> None:
    document = beautiful_soup.return_value
    document.select_one.return_value = None

    element = Mock()
    element.get.return_value = ["not", "a", "string"]
    document.select.return_value = [element]

    assert not receiver.html_mentions_target(b"<html></html>", SOURCE_URL, TARGET_URL)


@patch("webmentions_ssg.tasks.receiver.httpx.Client")
def test_fetch_source_ignores_invalid_content_length(
    httpx_client: Mock, app: Flask, receiver: ModuleType
) -> None:
    response = httpx.Response(
        200,
        headers={"Content-Type": "text/html", "Content-Length": "invalid"},
        content=b"<p>Document</p>",
        request=httpx.Request("GET", SOURCE_URL),
    )
    set_stream_response(httpx_client, response)

    with app.app_context():
        fetched_response, body = receiver.fetch_source(SOURCE_URL)

    assert fetched_response is response
    assert body == b"<p>Document</p>"


@patch("webmentions_ssg.tasks.receiver.fetch_source")
def test_source_mentions_target_falls_back_for_unknown_encoding(
    fetch_source: Mock, receiver: ModuleType
) -> None:
    body = f"Reply to {TARGET_URL}".encode()

    response = httpx.Response(
        200,
        headers={"Content-Type": "text/plain"},
        content=body,
        request=httpx.Request("GET", SOURCE_URL),
    )
    response.encoding = "unknown-encoding"
    fetch_source.return_value = response, body

    assert receiver.source_mentions_target(SOURCE_URL, TARGET_URL)
    fetch_source.assert_called_once_with(SOURCE_URL)