email-connect/tests/test_callers.py

"""Admitted caller contract: only synthetic tokens, providers, recipients and ledgers."""

import io
import json
from concurrent.futures import ThreadPoolExecutor

import pytest
from email_connect.callers import CallerPolicy, ScopedTransactionalApplication, load_callers
from email_connect.transactional import ProviderError, SQLiteDeliveryStore, TransactionalApplication


class Provider:
    def __init__(self):
        self.calls = []

    def send(self, *args):
        self.calls.append(args)
        return "synthetic-reference-recipient=member@example.test-token=must-not-leak"

    def check(self):
        return None


def policy(name="caller-a", **changes):
    values = dict(
        caller_id=name,
        token="synthetic-token-" + name,
        source=name,
        bindings=frozenset({("tenant:a", "client:a")}),
        operations=frozenset({"send", "delivery-status", "verify", "consume", "cancel"}),
        templates=frozenset({"family_member.invited"}),
        rate_per_minute=60,
        burst=100,
        max_payload_bytes=32768,
    )
    values.update(changes)
    return CallerPolicy(**values)


def event(caller="caller-a", **changes):
    payload = {
        "id": "evt-1",
        "type": "family_member.invited",
        "source": caller,
        "tenant": "tenant:a",
        "client_id": "client:a",
        "data": {"primary_email": "member@example.test", "invitation_id": "invite-1"},
    }
    payload.update(changes)
    return payload


def verification(**changes):
    result = {
        "registration_id": "reg-1",
        "normalized_email": "member@example.test",
        "preferred_username": "member",
        "tenant": "tenant:a",
        "client_id": "client:a",
        "correlation_id": "trace-1",
    }
    result.update(changes)
    return result


def invoke(app, payload, caller="caller-a", path="/v1/send", token=None, key=None):
    raw = json.dumps(payload).encode()
    response = {}
    env = {
        "PATH_INFO": path,
        "REQUEST_METHOD": "POST",
        "CONTENT_LENGTH": str(len(raw)),
        "wsgi.input": io.BytesIO(raw),
        "HTTP_AUTHORIZATION": "Bearer " + (token or "synthetic-token-" + caller),
        "HTTP_IDEMPOTENCY_KEY": key if key is not None else payload.get("id", ""),
    }
    body = b"".join(
        app(
            env,
            lambda status, headers: response.update(
                status=int(status.split()[0]),
                headers=dict(headers),
            ),
        )
    )
    return response["status"], json.loads(body), response["headers"]


def application(tmp_path, policies=None, clock=None):
    store = SQLiteDeliveryStore(str(tmp_path / "ledger.db"))
    provider = Provider()
    app = ScopedTransactionalApplication(
        store,
        provider,
        "https://issuer.example.test",
        policies or [policy(), policy("caller-b")],
        **({"clock": clock} if clock is not None else {}),
    )
    return app, store, provider


def test_credentials_determine_caller_and_status_is_scoped_and_redacted(tmp_path):
    app, _, provider = application(tmp_path)
    assert invoke(app, event(), token="wrong-credential")[0] == 401
    assert invoke(app, event("caller-b"))[0] == 400
    accepted, body, headers = invoke(app, event())
    assert accepted == 202 and body["event_id"] == "evt-1" and body["reference"].startswith("dvr_")
    assert headers["Cache-Control"] == "private, no-store"
    status, record, _ = invoke(app, {"event_id": "evt-1"}, path="/v1/delivery-status")
    assert status == 200 and record["state"] == "provider_accepted"
    assert record["reference"] == body["reference"]
    _, other, _ = invoke(app, {"event_id": "evt-1"}, caller="caller-b", path="/v1/delivery-status")
    assert other["state"] == "not_found"
    assert len(provider.calls) == 1
    serialized = json.dumps([body, record, other])
    assert "member@example.test" not in serialized and "must-not-leak" not in serialized


@pytest.mark.parametrize(
    "change",
    [
        {"tenant": "tenant:b"},
        {"client_id": "client:b"},
        {"tenant": ["tenant:a"]},
        {"type": "unapproved-template"},
        {"type": []},
        {"source": "user-engine"},
    ],
)
def test_scope_and_template_denials_never_submit(tmp_path, change):
    app, _, provider = application(tmp_path)
    assert invoke(app, event(**change))[0] == 400
    assert provider.calls == []


def test_retry_and_payload_identity_are_independent_across_callers_and_restart(tmp_path):
    app, store, provider = application(tmp_path)
    first = invoke(app, event())[1]
    assert invoke(app, event())[1]["status"] == "duplicate"
    assert (
        invoke(
            app, event(data={"primary_email": "other@example.test", "invitation_id": "invite-1"})
        )[0]
        == 400
    )
    assert invoke(app, event("caller-b"), caller="caller-b")[0] == 202
    assert len(provider.calls) == 2
    store.db.close()
    restarted, _, empty_provider = application(tmp_path)
    assert invoke(restarted, event())[1]["reference"] == first["reference"]
    assert empty_provider.calls == []


def test_unknown_submission_survives_restart_without_resend(tmp_path):
    app, store, provider = application(tmp_path)

    def unknown(*args):
        provider.calls.append(args)
        raise OSError("synthetic SMTP recipient/token must not leak")

    provider.send = unknown
    status, body, _ = invoke(app, event())
    assert status == 422 and body == {"error": "delivery_outcome_unknown", "retryable": False}
    store.db.close()
    restarted, _, next_provider = application(tmp_path)
    assert invoke(restarted, event())[1] == body
    assert len(provider.calls) == 1 and next_provider.calls == []


def test_rate_budget_survives_restart_and_backward_clock_and_is_per_caller(tmp_path):
    now = [100.0]
    policies = [policy(burst=1, rate_per_minute=2), policy("caller-b", burst=1, rate_per_minute=2)]
    app, store, provider = application(tmp_path, policies, clock=lambda: now[0])
    assert invoke(app, event())[0] == 202
    status, _, headers = invoke(app, event(id="evt-2"))
    assert status == 429 and headers["Retry-After"] == "30" and len(provider.calls) == 1
    assert invoke(app, event("caller-b"), caller="caller-b")[0] == 202
    store.db.close()
    restarted, _, next_provider = application(tmp_path, policies, clock=lambda: now[0])
    assert invoke(restarted, event(id="evt-2"))[0] == 429
    now[0] = 90
    assert invoke(restarted, event(id="evt-2"))[0] == 429
    now[0] = 130
    assert invoke(restarted, event(id="evt-2"))[0] == 202
    assert len(next_provider.calls) == 1


def test_concurrent_connections_cannot_exceed_request_burst(tmp_path):
    first, store, provider = application(tmp_path, [policy(burst=1)])
    second = ScopedTransactionalApplication(
        SQLiteDeliveryStore(str(tmp_path / "ledger.db")),
        provider,
        "https://issuer.example.test",
        [policy(burst=1)],
    )
    with ThreadPoolExecutor(max_workers=2) as workers:
        responses = list(workers.map(lambda app: invoke(app, event()), [first, second]))
    assert sorted(r[0] for r in responses) == [202, 429]
    assert len(provider.calls) == 1


def test_verification_owner_and_binding_checked_before_consumption_or_cancellation(tmp_path):
    app, store, provider = application(tmp_path)
    request = verification()
    first = invoke(app, request, path="/v1/registration-verifications")[1]
    assert invoke(app, request, path="/v1/registration-verifications")[1] == first
    handle = provider.calls[0][2].split("handle=", 1)[1].splitlines()[0]
    for path in ("/consume", "/cancel"):
        assert (
            invoke(
                app,
                {"handle": handle},
                caller="caller-b",
                path="/v1/registration-verifications" + path,
            )[0]
            == 400
        )
    assert invoke(app, request, caller="caller-b", path="/v1/registration-verifications")[0] == 202
    assert len(provider.calls) == 2
    legacy = TransactionalApplication(store, provider, "legacy", "https://issuer.example.test")
    raw = json.dumps({"handle": handle}).encode()
    response = {}
    legacy(
        {
            "PATH_INFO": "/v1/registration-verifications/consume",
            "REQUEST_METHOD": "POST",
            "CONTENT_LENGTH": str(len(raw)),
            "wsgi.input": io.BytesIO(raw),
            "HTTP_AUTHORIZATION": "Bearer legacy",
        },
        lambda status, headers: response.update(status=int(status.split()[0])),
    )
    assert response["status"] == 400
    limited = ScopedTransactionalApplication(
        store,
        provider,
        "https://issuer.example.test",
        [policy(bindings=frozenset({("tenant:b", "client:b")}))],
    )
    assert (
        invoke(limited, {"handle": handle}, path="/v1/registration-verifications/consume")[0] == 400
    )
    assert invoke(app, {"handle": handle}, path="/v1/registration-verifications/consume")[0] == 200
    assert invoke(app, {"handle": handle}, path="/v1/registration-verifications/cancel")[0] == 400


def test_suppression_and_unsuppression_require_explicit_owner_operations(tmp_path):
    ops = policy("mail-operator", operations=frozenset({"suppress", "unsuppress"}))
    app, _, provider = application(tmp_path, [policy(), ops])
    payload = {"recipient": "member@example.test", "reason": "complaint"}
    assert invoke(app, payload, path="/v1/suppressions")[0] == 403
    assert invoke(app, payload, caller="mail-operator", path="/v1/suppressions")[0] == 200
    assert invoke(app, event())[1]["error"] == "recipient_suppressed"
    assert not provider.calls
    assert invoke(app, payload, caller="mail-operator", path="/v1/suppressions/remove")[0] == 200
    assert invoke(app, event())[0] == 202


def test_original_caller_legacy_reservations_and_links_need_explicit_adoption(tmp_path):
    app, store, provider = application(tmp_path)
    payload = verification()
    legacy = TransactionalApplication(store, provider, "legacy", "https://issuer.example.test")
    raw = json.dumps(payload).encode()
    result = {}
    legacy(
        {
            "PATH_INFO": "/v1/registration-verifications",
            "REQUEST_METHOD": "POST",
            "CONTENT_LENGTH": str(len(raw)),
            "wsgi.input": io.BytesIO(raw),
            "HTTP_AUTHORIZATION": "Bearer legacy",
        },
        lambda status, headers: result.update(status=status),
    )
    handle = provider.calls[0][2].split("handle=", 1)[1].splitlines()[0]
    adopter = policy("user-engine", adopt_legacy_verifications=True)
    adopted = ScopedTransactionalApplication(
        store, provider, "https://issuer.example.test", [adopter]
    )
    assert (
        invoke(adopted, payload, caller="user-engine", path="/v1/registration-verifications")[0]
        == 202
    )
    assert len(provider.calls) == 1
    assert (
        invoke(
            adopted,
            {"handle": handle},
            caller="user-engine",
            path="/v1/registration-verifications/consume",
        )[0]
        == 200
    )
    with pytest.raises(ValueError, match="original_caller"):
        policy(adopt_legacy_verifications=True)


def test_payload_limit_unknown_operation_and_provider_status_have_no_implicit_grant(tmp_path):
    app, _, provider = application(tmp_path, [policy(max_payload_bytes=100)])
    assert invoke(app, event(data={"primary_email": "x" * 1000}))[0] == 413
    assert invoke(app, {}, path="/v1/status")[0] == 403
    assert invoke(app, {}, path="/unknown")[0] == 404
    assert not provider.calls


@pytest.mark.parametrize(
    "change",
    [
        {"burst": 0},
        {"rate_per_minute": True},
        {"max_payload_bytes": 200000},
        {"token": ""},
        {"bindings": frozenset()},
        {"templates": frozenset({"arbitrary-body"})},
    ],
)
def test_policy_refuses_unbounded_or_unauthorized_configuration(change):
    with pytest.raises(ValueError):
        policy(**change)


def test_policy_loader_uses_independent_files_and_never_echoes_credentials(tmp_path):
    token_file = tmp_path / "token"
    token_file.write_text("synthetic-token-caller-a\n")
    entry = dict(
        caller_id="caller-a",
        source="caller-a",
        bindings=[["tenant:a", "client:a"]],
        operations=["send"],
        templates=["family_member.invited"],
        rate_per_minute=10,
        burst=3,
        max_payload_bytes=8192,
        token_file=str(token_file),
    )
    config = tmp_path / "policy.json"
    config.write_text(json.dumps({"schema_version": 1, "callers": [entry]}))
    loaded = load_callers(config)
    assert loaded[0].token == "synthetic-token-caller-a"
    assert "synthetic-token" not in repr(loaded)
    config.write_text(
        json.dumps({"schema_version": 1, "callers": [entry, dict(entry, caller_id="caller-b")]})
    )
    with pytest.raises(ValueError, match="caller_configuration_invalid") as error:
        load_callers(config)
    assert "synthetic-token" not in str(error.value)
    entry["token"] = "synthetic-embedded-secret"
    config.write_text(json.dumps({"schema_version": 1, "callers": [entry]}))
    with pytest.raises(ValueError, match="caller_configuration_invalid"):
        load_callers(config)


@pytest.mark.parametrize("retryable", [False, True])
def test_definitive_failure_retains_retry_policy_after_restart(tmp_path, retryable):
    app, store, provider = application(tmp_path)

    def reject(*args):
        provider.calls.append(args)
        raise ProviderError(
            "temporary_deferral" if retryable else "permanent_rejection", retryable=retryable
        )

    provider.send = reject
    first = invoke(app, event())
    assert first[0] == (503 if retryable else 422)
    store.db.close()
    restarted, _, next_provider = application(tmp_path)
    result = invoke(restarted, event())
    assert result[0] == (202 if retryable else 422)
    assert len(next_provider.calls) == int(retryable)


def test_legacy_invitation_adoption_preserves_accepted_receipt(tmp_path):
    store = SQLiteDeliveryStore(str(tmp_path / "ledger.db"))
    provider = Provider()
    legacy = TransactionalApplication(
        store, provider, "legacy-token", "https://issuer.example.test"
    )
    old = event("user-engine")
    old.pop("tenant")
    old.pop("client_id")
    assert invoke(legacy, old, token="legacy-token")[0] == 202
    admitted = policy("user-engine", legacy_invitation_binding=("tenant:a", "client:a"))
    scoped = ScopedTransactionalApplication(
        store, provider, "https://issuer.example.test", [admitted]
    )
    assert invoke(scoped, old, caller="user-engine")[1]["status"] == "duplicate"
    assert len(provider.calls) == 1
    assert invoke(scoped, dict(old, id="ec2_reserved"), caller="user-engine")[0] == 400


def test_historical_failed_reservation_is_conservatively_terminal(tmp_path):
    import sqlite3

    path = tmp_path / "old.db"
    with sqlite3.connect(path) as db:
        db.execute(
            "CREATE TABLE send_attempts (event_id TEXT PRIMARY KEY, "
            "fingerprint TEXT NOT NULL, state TEXT NOT NULL)"
        )
        db.execute("INSERT INTO send_attempts VALUES ('old-event','old-fingerprint','failed')")
    store = SQLiteDeliveryStore(str(path))
    with pytest.raises(ProviderError, match="delivery_outcome_unknown"):
        store.reserve_send("old-event", "old-fingerprint")


def test_invalid_policy_refuses_startup_before_database_mutation(tmp_path, monkeypatch):
    from email_connect.transactional import main

    monkeypatch.setenv("EMAIL_CONNECT_CALLERS_FILE", str(tmp_path / "missing.json"))
    monkeypatch.setenv("EMAIL_CONNECT_DATABASE_PATH", str(tmp_path / "not-created.db"))
    with pytest.raises(ValueError, match="caller_configuration_invalid"):
        main()
    assert not (tmp_path / "not-created.db").exists()


@pytest.mark.parametrize("outcome", ["accepted", "unknown", "permanent"])
def test_original_caller_adds_binding_without_replaying_legacy_mail(tmp_path, outcome):
    import hashlib

    store = SQLiteDeliveryStore(str(tmp_path / "ledger.db"))
    provider = Provider()
    old = event("user-engine")
    old.pop("client_id")
    fingerprint = hashlib.sha256(
        json.dumps(old, sort_keys=True, separators=(",", ":")).encode()
    ).hexdigest()
    store.reserve_send(old["id"], fingerprint)
    if outcome == "accepted":
        store.record(old["id"], "synthetic-legacy-reference")
    elif outcome == "permanent":
        store.send_failed(
            old["id"], True, ProviderError("permanent_rejection", retryable=False)
        )
    admitted = policy("user-engine", adopt_legacy_invitations=True)
    scoped = ScopedTransactionalApplication(
        store, provider, "https://issuer.example.test", [admitted]
    )
    bound = dict(old, client_id="client:a")
    status, body, _ = invoke(scoped, bound, caller="user-engine")
    if outcome == "accepted":
        assert status == 200 and body["status"] == "duplicate"
    else:
        assert status == 422
        assert body["error"] == (
            "delivery_outcome_unknown"
            if outcome == "unknown"
            else "permanent_rejection"
        )
    assert provider.calls == []
    assert store.db.execute("SELECT fingerprint FROM send_attempts").fetchall() == [
        (fingerprint,)
    ]
    status, body, _ = invoke(
        scoped,
        dict(bound, data=dict(bound["data"], primary_email="changed@example.test")),
        caller="user-engine",
    )
    assert status == 400 and body["error"] == "idempotency_payload_mismatch"
    status, body, _ = invoke(
        scoped, dict(bound, tenant="tenant:denied"), caller="user-engine"
    )
    assert status == 400 and body["error"] == "caller_scope_denied"
    status, body, _ = invoke(
        scoped,
        {"event_id": old["id"]},
        caller="user-engine",
        path="/v1/delivery-status",
    )
    assert status == 200
    assert (
        body["state"]
        == {
            "accepted": "provider_accepted",
            "unknown": "unknown",
            "permanent": "failed",
        }[outcome]
    )
    assert provider.calls == []


def test_invitation_adoption_requires_original_caller_and_boolean():
    with pytest.raises(ValueError, match="legacy_admission_requires_original_caller"):
        policy(adopt_legacy_invitations=True)
    with pytest.raises(ValueError, match="invalid_legacy_admission"):
        policy("user-engine", adopt_legacy_invitations="true")


@pytest.mark.parametrize(
    "tenant,client,allowed",
    [
        ("tenant:family-one", "client:app", True),
        ("tenant:family-two", "client:app", True),
        ("tenant:platform", "client:app", False),
        ("tenant:family-one", "client:other", False),
        ("tenant:Family-one", "client:app", False),
        ("tenant:family-one/foreign", "client:app", False),
    ],
)
def test_binding_globs_match_whole_case_sensitive_valid_identifiers(
    tmp_path, tenant, client, allowed
):
    admitted = policy(binding_patterns=frozenset({("tenant:family-*", "client:app")}))
    app, _, provider = application(tmp_path, [admitted])
    status, body, _ = invoke(app, event(tenant=tenant, client_id=client))
    assert status == (202 if allowed else 400)
    assert len(provider.calls) == int(allowed)
    if not allowed:
        assert body["error"] == "caller_scope_denied"


def test_glob_binding_applies_to_verification_and_consumption(tmp_path):
    admitted = policy(
        bindings=frozenset(),
        binding_patterns=frozenset({("tenant:family-*", "client:?")}),
    )
    app, _, provider = application(tmp_path, [admitted])
    status, body, _ = invoke(
        app,
        verification(tenant="tenant:family-one"),
        path="/v1/registration-verifications",
    )
    assert status == 202
    handle = provider.calls[0][2].split("handle=", 1)[1].splitlines()[0]
    status, _, _ = invoke(
        app, {"handle": handle}, path="/v1/registration-verifications/consume"
    )
    assert status == 200


@pytest.mark.parametrize(
    "pattern", ["(tenant:.*)", "tenant:[ab]", "tenant:/other", "*" * 129]
)
def test_invalid_or_unbounded_binding_patterns_refuse_startup(pattern):
    with pytest.raises(ValueError, match="invalid_caller_binding_pattern"):
        policy(binding_patterns=frozenset({(pattern, "client:a")}))


def test_new_adopted_invitation_keeps_bound_fingerprint_across_restart(tmp_path):
    admitted = policy("user-engine", adopt_legacy_invitations=True)
    app, store, _ = application(tmp_path, [admitted])
    assert invoke(app, event("user-engine"), caller="user-engine")[0] == 202
    store.db.close()
    recovered, _, provider = application(tmp_path, [admitted])
    assert (
        invoke(recovered, event("user-engine"), caller="user-engine")[1]["status"]
        == "duplicate"
    )
    assert (
        invoke(
            recovered,
            event(
                "user-engine",
                data={
                    "primary_email": "changed@example.test",
                    "invitation_id": "invite-1",
                },
            ),
            caller="user-engine",
        )[0]
        == 400
    )
    assert provider.calls == []