email-connect/tests/test_integration_send_scan.py

"""End-to-end send-and-scan continuity against the local mail harness.

Every case here needs the harness (`docker compose -f tests/harness/docker-compose.yml
up -d`) and is skipped without it, so the default suite stays offline.

What these tests may assert is bounded: a message reaching a harness mailbox is
`provider_accepted` transport evidence and nothing more. Asserting delivery,
awareness, identity, or authorization from it would be a defect in the test.
"""

from __future__ import annotations

import imaplib
import io
import json
import tempfile
from email import policy
from email.parser import BytesParser
from pathlib import Path

import pytest

import harness
from email_connect.scanner import scan_mailbox
from email_connect.transactional import SQLiteDeliveryStore, TransactionalApplication

requires_harness = pytest.mark.skipif(
    not harness.available(),
    reason="mail harness not running: docker compose -f tests/harness/docker-compose.yml up -d",
)

TOKEN = "harness-ingest-token"
PORTAL = "https://users.harness.email-connect.test"


def application(tmp_path: Path) -> TransactionalApplication:
    return TransactionalApplication(
        SQLiteDeliveryStore(str(tmp_path / "mail.db")),
        harness.smtp_provider(),
        TOKEN,
        PORTAL,
    )


def invoke(app, payload, *, path="/v1/send", key=None, token=TOKEN):
    raw = json.dumps(payload).encode()
    result = {}
    env = {
        "PATH_INFO": path,
        "REQUEST_METHOD": "POST",
        "CONTENT_LENGTH": str(len(raw)),
        "wsgi.input": io.BytesIO(raw),
        "HTTP_AUTHORIZATION": f"Bearer {token}",
    }
    if path == "/v1/send":
        env["HTTP_IDEMPOTENCY_KEY"] = key if key is not None else payload.get("id")
    body = b"".join(app(env, lambda status, _headers: result.update(status=status)))
    return result["status"], json.loads(body)


def invitation(event_id: str, recipient: str, typ: str = "family_member.invited") -> dict:
    return {
        "id": event_id,
        "type": typ,
        "source": "user-engine",
        "data": {"primary_email": recipient, "invitation_id": f"inv-{event_id}"},
    }


def delivered_messages(user_address: str) -> list:
    """Every message in a harness mailbox, parsed."""

    connection = imaplib.IMAP4(harness.HOST, harness.IMAP_PORT)
    try:
        connection.login(user_address, harness.PASSWORD)
        connection.select("INBOX", readonly=True)
        _status, data = connection.uid("search", None, "ALL")
        uids = (data[0] or b"").split()
        messages = []
        for uid in uids:
            _status, fetched = connection.uid("fetch", uid, "(BODY.PEEK[])")
            for item in fetched:
                if isinstance(item, tuple) and len(item) >= 2 and isinstance(item[1], bytes):
                    messages.append(BytesParser(policy=policy.default).parsebytes(item[1]))
        return messages
    finally:
        connection.logout()


@requires_harness
def test_accepted_invitation_reaches_the_mailbox_and_is_correlatable(tmp_path):
    harness.reset()
    recipient = harness.address("invitation delivered")

    status, body = invoke(application(tmp_path), invitation("evt-1", recipient))

    assert status.startswith("202")
    assert body["status"] == "accepted"
    assert body["evidence_ceiling"] == "provider_accepted"

    messages = delivered_messages(recipient)
    assert len(messages) == 1
    assert messages[0]["To"] == recipient
    assert f"{PORTAL}/invitations/inv-evt-1" in messages[0].get_content()
    # The returned reference is the sent Message-ID, so an accepted send can be
    # tied to a message later observed in a mailbox.
    assert body["reference"] == str(messages[0]["Message-ID"]).strip()


@requires_harness
def test_scanner_ingests_a_delivered_invitation_without_claiming_delivery(tmp_path):
    harness.reset()
    recipient = harness.address("invitation scanned")
    invoke(application(tmp_path), invitation("evt-2", recipient))

    with tempfile.TemporaryDirectory() as tmp:
        root = Path(tmp)
        config = harness.mailbox_config(
            recipient,
            storage_path=str(root / "state.sqlite"),
            reports_dir=str(root / "reports"),
        )
        result = scan_mailbox(config)
        report = result.report_path.read_text(encoding="utf-8")

    assert result.scan.messages_seen == 1
    assert result.scan.messages_parsed == 1

    # An ordinary outbound invitation sitting in a mailbox is not return-path
    # evidence. The scanner must not turn it into a delivery, awareness,
    # identity, or interaction claim.
    for forbidden in (
        "notification.endpoint.delivered",
        "notification.message.delivered",
        "interaction.reply_received",
        "interaction.opened",
        "identity.verified",
    ):
        assert forbidden not in report


@requires_harness
def test_duplicate_event_sends_once_and_returns_the_same_reference(tmp_path):
    harness.reset()
    recipient = harness.address("invitation idempotent")
    app = application(tmp_path)

    first_status, first = invoke(app, invitation("evt-3", recipient))
    second_status, second = invoke(app, invitation("evt-3", recipient))

    assert first_status.startswith("202") and first["status"] == "accepted"
    assert second_status.startswith("200") and second["status"] == "duplicate"
    assert second["reference"] == first["reference"]
    assert len(delivered_messages(recipient)) == 1


@requires_harness
def test_resend_is_a_separate_event_and_delivers_again(tmp_path):
    harness.reset()
    recipient = harness.address("invitation resent")
    app = application(tmp_path)

    invoke(app, invitation("evt-4", recipient))
    status, body = invoke(app, invitation("evt-5", recipient, "family_invitation.resent"))

    assert status.startswith("202") and body["status"] == "accepted"
    messages = delivered_messages(recipient)
    assert len(messages) == 2
    assert len({message["Message-ID"] for message in messages}) == 2


@requires_harness
def test_suppressed_recipient_receives_nothing(tmp_path):
    harness.reset()
    recipient = harness.address("invitation suppressed")
    store = SQLiteDeliveryStore(str(tmp_path / "mail.db"))
    store.suppress(recipient, "complaint")
    app = TransactionalApplication(store, harness.smtp_provider(), TOKEN, PORTAL)

    status, body = invoke(app, invitation("evt-6", recipient))

    assert status.startswith("400")
    assert body["error"] == "recipient_suppressed"
    assert delivered_messages(recipient) == []


@requires_harness
def test_rejected_requests_never_reach_the_provider(tmp_path):
    harness.reset()
    recipient = harness.address("invitation rejected")
    app = application(tmp_path)

    unauthorized = invoke(app, invitation("evt-7", recipient), token="wrong")
    not_allowed = invoke(app, invitation("evt-8", recipient, "arbitrary.send"))
    key_mismatch = invoke(app, invitation("evt-9", recipient), key="other")

    assert unauthorized[0].startswith("401")
    assert not_allowed[1]["error"] == "template_not_allowed"
    assert key_mismatch[1]["error"] == "idempotency_key_mismatch"
    assert delivered_messages(recipient) == []


@requires_harness
def test_verification_mail_is_delivered_and_carries_no_authorization(tmp_path):
    harness.reset()
    recipient = harness.address("verification delivered")
    app = application(tmp_path)

    status, body = invoke(
        app,
        {
            "registration_id": "reg-1",
            "normalized_email": recipient,
            "preferred_username": "harness-user",
            "client_id": "portal",
            "tenant": "harness",
            "correlation_id": "corr-1",
        },
        path="/v1/registration-verifications",
    )

    assert status.startswith("2"), (status, body)
    messages = delivered_messages(recipient)
    assert len(messages) == 1
    assert f"{PORTAL}/registration/verify?handle=" in messages[0].get_content()
    # Delivering a verification link is not proof that anyone verified anything.
    assert body.get("authorization", False) is False