mail-address-semantics/tests/test_end_to_end.py

"""Full path: raw mail -> gateway -> MAS event -> HTTP app -> ticket lifecycle."""

import json
import subprocess
import sys
import threading
import urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

import pytest

from mas_core.discovery import Response, discover
from mas_gateway.core import Resolver, build_event, gateway, message_meta_from_rfc822
from mas_gateway.main import make_webhook_server
from ticket_service import TicketService, make_server

ROOT = Path(__file__).resolve().parents[1]
STATUTE = ROOT / "examples/statute/great-service.example/.well-known/mas"
D = "great-service.example"
CUSTOMER = "customer@example.net"


def raw(to, subject, body="hello", sender=CUSTOMER, extra=""):
    return (f"From: Customer <{sender}>\r\nTo: {to}\r\nSubject: {subject}\r\n"
            f"Message-ID: <{abs(hash((to, subject, body)))}@example.net>\r\n{extra}\r\n{body}\r\n").encode()


def serve(server):
    threading.Thread(target=server.serve_forever, daemon=True).start()
    return server


@pytest.fixture
def env():
    ids = iter(f"GS-{n}" for n in (41872, 41873, 41874))
    svc = TicketService(id_factory=lambda: next(ids))
    srv = serve(make_server(svc, port=0))
    resolver = Resolver.with_statute_files([STATUTE])
    app = f"http://127.0.0.1:{srv.server_address[1]}/mas-events"

    def send(to, subject, **kw):
        [d] = gateway([to], message_meta_from_rfc822(raw(to, subject, **kw)), resolver, app)
        return d
    yield svc, send, app, resolver
    srv.shutdown()


def test_canonical_lifecycle(env):
    svc, send, *_ = env
    # Flow A: create via P2 interaction intake
    d = send(f"ticket!callback@{D}", "My router is broken")
    assert d.status == 201 and d.response["resource_address"] == f"ticket=GS-41872@{D}"
    assert d.response["requested_interaction"] == "callback"
    [reply] = svc.outbox()
    assert reply["mail_from"] == f"ticket=GS-41872@{D}" and reply["mail_to"] == CUSTOMER
    assert "GS-41872" not in reply["subject"]           # no identifier in the subject

    # Flow B: correlate purely by address; the subject is unrelated
    d = send(f"ticket=GS-41872@{D}", "The problem happened again")
    assert d.response == {"status": "correlated", "ticket_id": "GS-41872", "facet": None}
    d = send(f"ticket=GS-41872+comments@{D}", "a comment")
    assert d.response["facet"] == "comments"
    d = send(f"ticket=GS-41872@{D}", "comment (facet stripped in transit)")
    assert d.response["ticket_id"] == "GS-41872"         # facet loss degrades safely
    assert len(svc.ticket("GS-41872")["messages"]) == 4

    # Flow C: closed resource stays addressable but immutable
    assert svc.close("GS-41872")
    d = send(f"ticket=GS-41872@{D}", "Hello again?")
    assert d.response["status"] == "closed" and d.response["mutated"] is False
    assert d.response["notified"] is True
    t = svc.ticket("GS-41872")
    assert t["status"] == "closed"
    assert [m["disposition"] for m in t["messages"]][-1] == "archived-after-close"
    d = send(f"ticket=GS-41872@{D}", "From a stranger", sender="other@example.org")
    assert d.response["notified"] is False                # no notice to non-participants

    # Flow D: unknown resource -> quarantined, no reply
    before = len(svc.outbox())
    d = send(f"ticket=DOES-NOT-EXIST@{D}", "guess")
    assert d.status == 202 and d.response["status"] == "unknown-resource"
    assert len(svc.outbox()) == before and len(svc.quarantine()) == 1


def test_default_ids_are_lower_case():
    from ticket_service import default_ticket_id
    tid = default_ticket_id()
    assert tid == tid.lower() and tid.startswith("gs-") and len(tid) == 13


def test_lower_cased_id_correlates(env):
    # IONOS forwards and Google Groups members arrive lower-cased (MAS-WP-0003 T04)
    svc, send, *_ = env
    send(f"ticket!callback@{D}", "request")
    d = send(f"ticket=gs-41872+comments@{D}", "via a mailing list")
    assert d.response == {"status": "correlated", "ticket_id": "GS-41872", "facet": "comments"}
    assert len(svc.ticket("gs-41872")["messages"]) == 2
    assert svc.close("gs-41872")
    d = send(f"ticket=gs-41872@{D}", "after close")
    assert d.response["ticket_id"] == "GS-41872" and d.response["notified"] is True
    assert svc.outbox()[-1]["mail_from"] == f"ticket=GS-41872@{D}"   # replies use the issued id


def test_send_cap_holds_replies(env):
    svc, send, *_ = env
    sent = []
    svc.mailer.smtp, svc.mailer.max_sent = ("127.0.0.1", 25), 2
    svc.mailer._send = lambda *a: sent.append(a)
    for n in range(3):
        send(f"ticket!callback@{D}", f"request {n}")
    assert len(sent) == 2 and svc.send_cap_reached()
    out = svc.outbox()
    assert [(r["sent"], r["held"]) for r in out] == [(1, None), (1, None), (0, "send-cap-reached")]
    assert out[2]["ticket_id"] == out[2]["case_ref"]


def test_only_declared_intake_creates(env):
    # SCP-3: "support" is a function without intake_for, so it creates nothing
    svc, send, *_ = env
    d = send(f"support@{D}", "hello")
    assert d.response["status"] == "not-handled" and svc.outbox() == []


def test_no_auto_reply_to_automated_mail(env):
    svc, send, *_ = env
    send(f"ticket@{D}", "out of office", extra="Auto-Submitted: auto-replied\r\n")
    assert svc.outbox() == []


def test_interaction_not_enabled_is_not_honoured(env):
    svc, send, *_ = env
    d = send(f"ticket!chat@{D}", "please chat")
    assert d.status == 201 and d.response["requested_interaction"] is None
    assert "not available" in svc.outbox()[0]["body"]


def test_unconfirmed_namespace_is_not_acted_on(env):
    svc, _, app, _ = env
    resolver = Resolver(discover_fn=lambda host, **kw: discover(
        host, transport=lambda u, t: Response(404, {}, b"")))
    [d] = gateway([f"ticket=GS-41872@{D}"], {"from": CUSTOMER, "subject": "x"}, resolver, app)
    assert d.status == 422 and d.response["status"] == "namespace-unconfirmed"


def test_non_mas_recipient_still_forwarded(env):
    _, _, _, resolver = env
    ev = build_event(f'"not mas"@{D}', {}, resolver)
    assert ev["mas"]["valid"] is False and ev["recipient"] == f'"not mas"@{D}'
    # SCP-1: unnecessary quotes are equivalent to the unquoted form
    ev = build_event(f'"ticket=GS-1"@{D}', {}, resolver)
    assert ev["mas"]["valid"] is True and ev["mas"]["equivalence_key"] == "ticket=GS-1"
    assert ev["recipient"] == f'"ticket=GS-1"@{D}'   # the event keeps what arrived


def test_pipe_transport_subprocess(env):
    svc, _, app, _ = env
    to = f"ticket!callback@{D}"
    p = subprocess.run([sys.executable, "-m", "mas_gateway.main", "--statute", str(STATUTE), "pipe",
                        "--recipient", to, "--app-url", app], input=raw(to, "via pipe"),
                       capture_output=True, cwd=ROOT)
    assert p.returncode == 0, p.stderr
    assert svc.ticket("GS-41872")["requested_interaction"] == "callback"
    p = subprocess.run([sys.executable, "-m", "mas_gateway.main", "--statute", str(STATUTE), "pipe",
                        "--recipient", to, "--app-url", "http://127.0.0.1:9/none"],
                       input=raw(to, "app down"), capture_output=True, cwd=ROOT)
    assert p.returncode == 75                              # tempfail, never a bounce


def test_webhook_input(env):
    svc, _, app, resolver = env
    gw = serve(make_webhook_server("127.0.0.1:0", app, resolver))
    try:
        url = f"http://127.0.0.1:{gw.server_address[1]}/inbound"
        body = json.dumps({"recipient": f"ticket!callback@{D}", "from": CUSTOMER,
                           "subject": "via webhook", "text": "hi"}).encode()
        req = urllib.request.Request(url, body, {"Content-Type": "application/json"})
        with urllib.request.urlopen(req) as r:
            [d] = json.loads(r.read())["deliveries"]
        assert d["status"] == 201 and d["response"]["ticket_id"] == "GS-41872"
    finally:
        gw.shutdown()


def test_gateway_is_resource_type_agnostic():
    """Another resource type routes through the unchanged gateway."""
    got = []

    class App(BaseHTTPRequestHandler):
        def do_POST(self):
            got.append(json.loads(self.rfile.read(int(self.headers["Content-Length"]))))
            self.send_response(204)
            self.end_headers()

        def log_message(self, *a):
            pass

    srv = serve(ThreadingHTTPServer(("127.0.0.1", 0), App))
    try:
        resolver = Resolver.with_statute_files([ROOT / "mas_conformance/statutes/valid-example.json"])
        gateway(["invoice=RE=2026-4711+dispute@coulomb.social"], {"from": CUSTOMER},
                resolver, f"http://127.0.0.1:{srv.server_address[1]}/")
        mas = got[0]["mas"]
        assert (mas["kind"], mas["resource_type"], mas["resource_id"], mas["facet"]) == \
            ("resource", "invoice", "RE=2026-4711", "dispute")
    finally:
        srv.shutdown()


# -- sender authentication (spec 14 item 10) ----------------------------------------

AR_OURS = "Authentication-Results: great-service.example; dkim=pass header.d=example.net; " \
          "spf=pass smtp.mailfrom=example.net; dmarc=pass header.from=example.net\r\n"
RECEIVED = "Received: from mx.example.net by great-service.example; Sun, 18 Oct 2026 20:00:00 +0000\r\n"


def _meta(headers):
    return message_meta_from_rfc822(raw(f"ticket!callback@{D}", "hi", extra=headers),
                                    trusted_authserv_id=D)


def test_authentication_read_from_our_own_header():
    a = _meta(AR_OURS + RECEIVED)["authentication"]
    assert a == {"dkim": "pass", "spf": "pass", "dmarc": "pass", "header_from": "example.net"}


def test_forged_or_foreign_authentication_is_ignored():
    forged = RECEIVED + AR_OURS                      # below our Received: brought by the sender
    foreign = AR_OURS.replace("great-service.example;", "other.example;") + RECEIVED
    assert _meta(forged)["authentication"] is None
    assert _meta(foreign)["authentication"] is None
    assert message_meta_from_rfc822(raw("x@y", "s", extra=AR_OURS))["authentication"] is None  # no trusted id


def test_reply_only_to_authenticated_sender(env):
    svc, send, *_ = env
    svc.mailer.require_dmarc_pass = True
    sent = []
    svc.mailer.smtp, svc.mailer._send = ("127.0.0.1", 25), lambda *a: sent.append(a)
    send(f"ticket!callback@{D}", "spoofed")                         # no verdict
    assert sent == [] and svc.outbox() == [] and svc.ticket("GS-41872") is not None


def test_authenticated_sender_gets_reply(env):
    svc, _, _, resolver = env
    svc.mailer.require_dmarc_pass = True
    sent = []
    svc.mailer.smtp, svc.mailer._send = ("127.0.0.1", 25), lambda *a: sent.append(a)
    event = build_event(f"ticket!callback@{D}", _meta(AR_OURS + RECEIVED), resolver)
    status, body = svc.handle(event)
    assert status == 201 and len(sent) == 1 and sent[0][1] == CUSTOMER