rfc-app/backend/tests/test_outbound_emails_vertical.py

"""End-to-end integration tests for the v0.18.0 Slice 4
outbound_emails audit table + admin endpoint.

The release adds:
  * `backend/migrations/020_outbound_emails.sql` — the audit table.
  * `record_outbound()` in `email.py` — the write helper every send
    path calls before returning, capturing status='sent' / 'failed'
    / 'deferred' (the dev-fallback path when SMTP_HOST is unset).
  * `GET /api/admin/outbound-emails` — admin-only listing, filterable
    by kind / status / to_address.

These tests prove:
  * Sending OTC / invite / notification mail writes one row per send
    (status='deferred' under tests since SMTP_HOST is unset).
  * The Message-ID on the row matches the envelope's Message-ID
    header (the seam Slice 5 uses for bounce correlation).
  * `kind` is populated per send path.
  * `GET /api/admin/outbound-emails` lists rows newest-first,
    accepts filters, refuses non-admins.
"""
from __future__ import annotations

from test_propose_vertical import (  # noqa: F401
    FakeGitea,
    app_with_fake_gitea,
    provision_user_row,
    sign_in_as,
    tmp_env,
)


# ---------------------------------------------------------------------------
# Write-on-send wiring
# ---------------------------------------------------------------------------


def test_otc_send_writes_outbound_row_with_message_id(app_with_fake_gitea):
    from fastapi.testclient import TestClient
    from app import db, email as email_mod

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        email_mod.reset_sent_envelopes()
        r = client.post("/auth/otc/request", json={"email": "newcomer@ex.co"})
        assert r.status_code == 200

        # Audit row landed.
        rows = db.conn().execute(
            "SELECT id, to_address, kind, status, message_id, error "
            "FROM outbound_emails WHERE to_address = 'newcomer@ex.co'"
        ).fetchall()
        assert len(rows) == 1
        row = rows[0]
        assert row["kind"] == "otc"
        # No SMTP_HOST in tests -> 'deferred', not 'sent'.
        assert row["status"] == "deferred"
        assert row["error"] is None
        # Message-ID matches the envelope's header (the seam Slice 5 uses).
        envelopes = [e for e in email_mod.sent_envelopes() if e.get("kind") == "otc"]
        assert envelopes
        envelope_mid = envelopes[-1]["message"]["Message-ID"]
        assert row["message_id"] == envelope_mid


def test_invite_send_writes_outbound_row(app_with_fake_gitea):
    from fastapi.testclient import TestClient
    from app import db

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        provision_user_row(user_id=500, login="adminQ", role="admin")
        sign_in_as(
            client, user_id=500, gitea_login="adminQ",
            display_name="Admin Q", role="admin",
            email="adminq@test",
        )
        r = client.post(
            "/api/admin/users",
            json={
                "email": "invitee@ex.co",
                "first_name": "Inv", "last_name": "Itee",
                "role": "contributor", "custom_message": "",
            },
        )
        assert r.status_code == 200, r.text

        rows = db.conn().execute(
            "SELECT kind, status, message_id FROM outbound_emails "
            "WHERE to_address = 'invitee@ex.co'"
        ).fetchall()
        assert len(rows) == 1
        assert rows[0]["kind"] == "invite"
        assert rows[0]["status"] == "deferred"
        assert rows[0]["message_id"] is not None


def test_notification_send_writes_outbound_row_with_notification_id(app_with_fake_gitea):
    """Watcher notifications carry a `notification_id` FK so the
    admin can join through to the notifications table to see what
    triggered the send."""
    from fastapi.testclient import TestClient
    from app import db, email as email_mod
    from test_notifications_vertical import PITCH

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        provision_user_row(user_id=2, login="alice", role="contributor")
        provision_user_row(user_id=1, login="ben", role="owner")
        sign_in_as(
            client, user_id=2, gitea_login="alice",
            display_name="Alice", role="contributor", email="alice@test",
        )
        r = client.post("/api/rfcs/propose", json={
            "title": "OHM", "slug": "ohm", "pitch": PITCH, "tags": [],
        })
        email_mod.reset_sent_envelopes()
        # Wipe pre-merge audit rows so the assertion below is unambiguous.
        db.conn().execute("DELETE FROM outbound_emails")
        sign_in_as(
            client, user_id=1, gitea_login="ben",
            display_name="Ben", role="owner", email="ben@test",
        )
        merge_r = client.post(f"/api/proposals/{r.json()['pr_number']}/merge")
        assert merge_r.status_code == 200, merge_r.text

        rows = db.conn().execute(
            "SELECT kind, status, notification_id, message_id "
            "FROM outbound_emails WHERE to_address = 'alice@test'"
        ).fetchall()
        assert rows, "no outbound_emails row for alice@test"
        # At least one notification kind, with a populated FK.
        notif_rows = [r for r in rows if r["kind"] == "notification"]
        assert notif_rows
        for nr in notif_rows:
            assert nr["status"] == "deferred"
            assert nr["notification_id"] is not None
            assert nr["message_id"] is not None


# ---------------------------------------------------------------------------
# Admin endpoint
# ---------------------------------------------------------------------------


def test_admin_outbound_emails_lists_rows(app_with_fake_gitea):
    from fastapi.testclient import TestClient

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        # Generate a few rows.
        client.post("/auth/otc/request", json={"email": "one@ex.co"})

        provision_user_row(user_id=600, login="adminR", role="admin")
        sign_in_as(
            client, user_id=600, gitea_login="adminR",
            display_name="Admin R", role="admin", email="adminr@test",
        )
        client.post("/api/admin/users", json={
            "email": "two@ex.co", "first_name": "T", "last_name": "Wo",
            "role": "contributor", "custom_message": "",
        })

        r = client.get("/api/admin/outbound-emails")
        assert r.status_code == 200, r.text
        items = r.json()["items"]
        kinds = {it["kind"] for it in items}
        assert "otc" in kinds
        assert "invite" in kinds
        # Newest-first.
        ids = [it["id"] for it in items]
        assert ids == sorted(ids, reverse=True)


def test_admin_outbound_emails_filters_by_kind(app_with_fake_gitea):
    from fastapi.testclient import TestClient

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        client.post("/auth/otc/request", json={"email": "filter1@ex.co"})

        provision_user_row(user_id=601, login="adminS", role="admin")
        sign_in_as(
            client, user_id=601, gitea_login="adminS",
            display_name="Admin S", role="admin", email="admins@test",
        )
        client.post("/api/admin/users", json={
            "email": "filter2@ex.co", "first_name": "F", "last_name": "Two",
            "role": "contributor", "custom_message": "",
        })

        r = client.get("/api/admin/outbound-emails?kind=otc")
        assert r.status_code == 200
        items = r.json()["items"]
        assert items
        assert all(it["kind"] == "otc" for it in items)


def test_admin_outbound_emails_filters_by_to_address(app_with_fake_gitea):
    from fastapi.testclient import TestClient

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        client.post("/auth/otc/request", json={"email": "TARGET@ex.co"})
        client.post("/auth/otc/request", json={"email": "other@ex.co"})

        provision_user_row(user_id=602, login="adminT", role="admin")
        sign_in_as(
            client, user_id=602, gitea_login="adminT",
            display_name="Admin T", role="admin", email="admint@test",
        )

        # to_address filter is case-insensitive.
        r = client.get("/api/admin/outbound-emails?to_address=target@ex.co")
        assert r.status_code == 200
        items = r.json()["items"]
        assert items
        assert all(it["to_address"].lower() == "target@ex.co" for it in items)


def test_admin_outbound_emails_refuses_non_admin(app_with_fake_gitea):
    from fastapi.testclient import TestClient

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        provision_user_row(user_id=700, login="contribU", role="contributor")
        sign_in_as(
            client, user_id=700, gitea_login="contribU",
            display_name="Contrib U", role="contributor",
        )
        r = client.get("/api/admin/outbound-emails")
        assert r.status_code == 403


# ---------------------------------------------------------------------------
# v0.18.0 Slice 5: bounce correlation
# ---------------------------------------------------------------------------


def test_bounce_with_message_id_marks_outbound_row_bounced(app_with_fake_gitea):
    """When the bounce body includes the original `message_id`, the
    framework looks it up in outbound_emails and stamps
    status='bounced' on the matching row. The hard-bounce ->
    global-opt-out logic still fires."""
    from fastapi.testclient import TestClient
    from app import db, email as email_mod

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        provision_user_row(user_id=800, login="bouncey", role="contributor")
        db.conn().execute("UPDATE users SET email = 'bouncey@ex.co' WHERE id = 800")

        # Send something to bouncey to land an outbound_emails row.
        email_mod.reset_sent_envelopes()
        client.post("/auth/otc/request", json={"email": "bouncey@ex.co"})
        row = db.conn().execute(
            "SELECT id, message_id, status FROM outbound_emails "
            "WHERE to_address = 'bouncey@ex.co'"
        ).fetchone()
        assert row is not None
        original_id = row["id"]
        message_id = row["message_id"]
        assert row["status"] == "deferred"  # pre-bounce baseline

        # Bounce comes in carrying that message_id.
        r = client.post(
            "/api/webhooks/email-bounce",
            json={
                "email": "bouncey@ex.co",
                "kind": "hard",
                "message_id": message_id,
            },
        )
        assert r.status_code == 200
        body = r.json()
        assert body["matched"] is True
        assert body["correlated_id"] == original_id

        # Audit row stamped.
        post = db.conn().execute(
            "SELECT status, error FROM outbound_emails WHERE id = ?",
            (original_id,),
        ).fetchone()
        assert post["status"] == "bounced"
        assert "bounce (hard)" in (post["error"] or "")

        # Hard-bounce global opt-out still fires.
        urow = db.conn().execute(
            "SELECT email_opt_out_all FROM users WHERE id = 800"
        ).fetchone()
        assert urow["email_opt_out_all"] == 1


def test_bounce_with_unknown_message_id_does_not_crash(app_with_fake_gitea):
    """A message_id the framework doesn't recognize logs but does
    NOT 5xx — bounce providers replay old bounces, and the
    framework can't refuse just because the row was pruned."""
    from fastapi.testclient import TestClient

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        r = client.post(
            "/api/webhooks/email-bounce",
            json={
                "email": "nobody@ex.co",
                "kind": "hard",
                "message_id": "<not-in-our-db@ex.co>",
            },
        )
        assert r.status_code == 200
        assert r.json()["correlated_id"] is None


def test_bounce_without_message_id_still_flips_opt_out(app_with_fake_gitea):
    """Backward compat: providers that don't surface Message-ID
    still get the legacy v1 behavior — match by email + flip the
    global opt-out."""
    from fastapi.testclient import TestClient
    from app import db

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        provision_user_row(user_id=801, login="legacybounce", role="contributor")
        db.conn().execute("UPDATE users SET email = 'legacy@ex.co' WHERE id = 801")

        r = client.post(
            "/api/webhooks/email-bounce",
            json={"email": "legacy@ex.co", "kind": "hard"},
        )
        assert r.status_code == 200
        body = r.json()
        assert body["matched"] is True
        assert body["correlated_id"] is None

        urow = db.conn().execute(
            "SELECT email_opt_out_all FROM users WHERE id = 801"
        ).fetchone()
        assert urow["email_opt_out_all"] == 1


def test_bounced_rows_show_in_admin_endpoint(app_with_fake_gitea):
    """The admin endpoint surfaces bounced rows alongside the rest;
    filtering by `status=bounced` isolates them."""
    from fastapi.testclient import TestClient
    from app import db, email as email_mod

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        provision_user_row(user_id=802, login="adminB", role="admin")
        sign_in_as(
            client, user_id=802, gitea_login="adminB",
            display_name="Admin B", role="admin", email="adminb@test",
        )
        email_mod.reset_sent_envelopes()
        client.post("/auth/otc/request", json={"email": "willbounce@ex.co"})
        row = db.conn().execute(
            "SELECT message_id FROM outbound_emails WHERE to_address = 'willbounce@ex.co'"
        ).fetchone()
        client.post(
            "/api/webhooks/email-bounce",
            json={"email": "willbounce@ex.co", "kind": "hard", "message_id": row["message_id"]},
        )

        r = client.get("/api/admin/outbound-emails?status=bounced")
        assert r.status_code == 200
        items = r.json()["items"]
        assert items
        assert all(it["status"] == "bounced" for it in items)
        assert any(it["to_address"] == "willbounce@ex.co" for it in items)