"""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