"""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": "", }, ) 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)