"""API tests: Gmail accounts manager, Send Test Email, public unsubscribe flow,
and the full campaign lifecycle (wizard options, create/patch/delete/clear-all,
validate-step, preview, start/pause/resume/cancel, progress, recipients, export).

Real behavior is asserted everywhere; nothing is faked. Where real Google OAuth
cannot be exercised (no credentials in CI), the app's honest error responses are
asserted instead of mocked successes.
"""
import csv
import io
import json
import secrets
import uuid
from datetime import datetime, timedelta, timezone

import pytest


# --------------------------------------------------------------------------- helpers
def csrf(client) -> str:
    for c in client.cookies.jar:
        if c.name == "mailer_csrf":
            return c.value
    raise AssertionError("CSRF cookie missing")


def H(client):
    return {"X-CSRF-Token": csrf(client)}


def user_id_of(client) -> int:
    r = client.get("/auth/me")
    assert r.status_code == 200
    return r.json()["id"]


def celery_available() -> bool:
    """True when a real Redis broker is reachable (campaign start enqueues jobs)."""
    try:
        import redis as _r
        from app.core.config import settings

        conn = _r.from_url(settings.REDIS_URL, socket_timeout=1)
        conn.ping()
        conn.close()
        return True
    except Exception:
        return False


requires_broker = pytest.mark.skipif(not celery_available(),
                                     reason="Redis broker not available")


def _seed_account(db, user_id, email=None, status="authorized", method="api",
                  with_token=True, refresh="refresh-abc"):
    """Insert a GmailAccount (+ OAuthToken) directly. Tokens are fake strings —
    the app never uses them in these endpoints except honest-failure paths."""
    from app.models import GmailAccount, OAuthToken

    acc = GmailAccount(user_id=user_id,
                       email=email or f"g{uuid.uuid4().hex[:8]}@gmail.com",
                       display_name="Test Account", status=status,
                       sending_method=method, is_active=(status != "revoked"))
    db.add(acc)
    db.flush()
    if with_token:
        db.add(OAuthToken(account_id=acc.id,
                          access_token="fake-access-token",
                          refresh_token=refresh,
                          expires_at=datetime.now(timezone.utc) + timedelta(hours=1)))
    db.commit()
    return acc.id


def _seed_campaign_fixture(db, user_id, n_recipients=3):
    """RecipientList + recipients + subject + template + authorized account.
    Returns ids dict for building CampaignIn payloads."""
    from app.models import (EmailTemplate, GmailAccount, OAuthToken, Recipient,
                            RecipientList, SubjectLine)

    rl = RecipientList(user_id=user_id, name=f"camp-list-{uuid.uuid4().hex[:6]}")
    db.add(rl)
    db.flush()
    for i in range(1, n_recipients + 1):
        db.add(Recipient(list_id=rl.id, email=f"camp{i}@x.com",
                         first_name=f"Camp{i}", last_name="User",
                         unsub_token=secrets.token_urlsafe(24), is_valid=True))
    subj = SubjectLine(user_id=user_id, subject="Your invoice %RandomInvoiceNumber% is ready")
    db.add(subj)
    tpl = EmailTemplate(user_id=user_id, name=f"tpl-{uuid.uuid4().hex[:6]}",
                        subject_hint="Invoice",
                        body_html="<p>Hi %FirstName%, pay %RandomTotal% now.</p>")
    db.add(tpl)
    acc = GmailAccount(user_id=user_id, email=f"camp-acc-{uuid.uuid4().hex[:6]}@gmail.com",
                       status="authorized", sending_method="api")
    db.add(acc)
    db.flush()
    db.add(OAuthToken(account_id=acc.id, access_token="fake-access",
                      refresh_token="fake-refresh"))
    db.commit()
    return {"list_id": rl.id, "subject_id": subj.id, "template_id": tpl.id,
            "account_id": acc.id}


def _cIn(fixture, **over):
    base = {
        "name": "Test Campaign",
        "recipient_list_id": fixture["list_id"],
        "subject_set_ids": [fixture["subject_id"]],
        "template_ids": [fixture["template_id"]],
        "message_type": "html",
        "attach_mode": "none",
        "account_ids": [fixture["account_id"]],
        "rotation_strategy": "sequential",
        "subject_rotation": "round_robin",
        "template_sequence_mode": "repeat",
        "phone_selection": "round_robin",
    }
    base.update(over)
    return base


# --------------------------------------------------------------------------- accounts
class TestAccountsApi:

    def test_list_empty(self, api):
        d = api.get("/gmail/accounts").json()
        assert d["accounts"] == [] and d["count"] == 0

    def test_seeded_account_shape(self, api, db):
        aid = _seed_account(db, user_id_of(api))
        d = api.get("/gmail/accounts").json()
        assert d["count"] == 1
        a = d["accounts"][0]
        assert {"id", "email", "status", "sending_method", "daily_limit",
                "campaign_limit", "sends_today", "has_refresh_token",
                "authorized"} <= set(a)
        assert a["id"] == aid and a["authorized"] is True
        assert a["has_refresh_token"] is True

    def test_patch_updates_limits_and_method(self, api, db):
        aid = _seed_account(db, user_id_of(api), method="api")
        r = api.patch(f"/gmail/accounts/{aid}", headers=H(api),
                      json={"sending_method": "smtp", "daily_limit": 50,
                            "campaign_limit": 25})
        assert r.status_code == 200, r.text
        d = r.json()
        assert d["sending_method"] == "smtp"
        assert d["daily_limit"] == 50 and d["campaign_limit"] == 25

    def test_patch_404_unknown(self, api):
        r = api.patch("/gmail/accounts/999", headers=H(api),
                      json={"sending_method": "api", "daily_limit": 5,
                            "campaign_limit": 5})
        assert r.status_code == 404

    def test_connect_without_config_is_honest_400(self, api):
        r = api.post("/gmail/accounts/connect", headers=H(api))
        assert r.status_code == 400
        assert "client configuration" in r.text.lower()

    def test_reauthorize_without_config_400(self, api, db):
        aid = _seed_account(db, user_id_of(api))
        r = api.post(f"/gmail/accounts/{aid}/reauthorize", headers=H(api))
        assert r.status_code == 400

    def test_revoke_and_delete_account(self, api, db):
        aid = _seed_account(db, user_id_of(api))
        r = api.post(f"/gmail/accounts/{aid}/revoke", headers=H(api))
        assert r.status_code == 200, r.text
        d = r.json()
        assert d["ok"] is True and "revoked_at_google" in d
        # revoked → is_active False, status revoked, token deleted
        d2 = api.get("/gmail/accounts").json()
        assert d2["accounts"][0]["status"] == "revoked"
        assert d2["accounts"][0]["authorized"] is False
        from app.models import OAuthToken
        assert db.query(OAuthToken).count() == 0
        # DELETE removes the row entirely
        r3 = api.delete(f"/gmail/accounts/{aid}", headers=H(api))
        assert r3.status_code == 200
        assert api.get("/gmail/accounts").json()["count"] == 0

    def test_revoke_deletes_token_row_and_sets_flags(self, api, db):
        aid = _seed_account(db, user_id_of(api))
        api.post(f"/gmail/accounts/{aid}/revoke", headers=H(api))
        from app.models import GmailAccount
        acc = db.get(GmailAccount, aid)
        assert acc.status == "revoked" and acc.is_active is False
        assert acc.revoked_by_user is True and acc.revoked_at is not None

    def test_clear_all(self, api, db):
        _seed_account(db, user_id_of(api), email="a1@x.com")
        _seed_account(db, user_id_of(api), email="a2@x.com")
        r = api.post("/gmail/accounts/clear-all", headers=H(api))
        assert r.status_code == 200
        d = r.json()
        assert d["removed"] == 2
        assert api.get("/gmail/accounts").json()["count"] == 0
        # audit preserved
        acts = api.get("/logs/audit").json()
        assert any(a["action"] == "gmail.clear_all" for a in acts["audit"])

    def test_account_isolation(self, api, db, second_user):
        # second user's account must be invisible and untouchable
        aid = _seed_account(db, user_id_of(api))
        assert second_user.get("/gmail/accounts").json()["accounts"] == []
        r = second_user.patch(f"/gmail/accounts/{aid}", headers=H(second_user),
                              json={"sending_method": "api", "daily_limit": 1,
                                    "campaign_limit": 1})
        assert r.status_code == 404
        r2 = second_user.post(f"/gmail/accounts/{aid}/revoke", headers=H(second_user))
        assert r2.status_code == 404

    def test_import_config_valid_json(self, api):
        cfg = {"installed": {"client_id": "abc-123.apps.googleusercontent.com",
                             "client_secret": "SECRET-XYZ",
                             "project_id": "my-project",
                             "auth_uri": "https://accounts.google.com/o/oauth2/auth",
                             "token_uri": "https://oauth2.googleapis.com/token"}}
        r = api.post("/gmail/accounts/import-config", headers=H(api),
                     files={"file": ("client_secret.json", json.dumps(cfg).encode(),
                                     "application/json")})
        assert r.status_code in (200, 201), r.text
        d = r.json()
        assert d["ok"] is True and d["project_id"] == "my-project"
        # secret is NEVER echoed back, only a preview
        assert "SECRET-XYZ" not in r.text
        assert d["client_id_preview"].startswith("abc-123")
        # after import, connect can build a real authorization URL
        r2 = api.post("/gmail/accounts/connect", headers=H(api))
        assert r2.status_code == 200, r2.text
        assert "accounts.google.com" in r2.json()["authorization_url"]

    def test_import_config_rejects_invalid(self, api):
        r = api.post("/gmail/accounts/import-config", headers=H(api),
                     files={"file": ("bad.json", b'{"nope": true}', "application/json")})
        assert r.status_code == 400
        r2 = api.post("/gmail/accounts/import-config", headers=H(api),
                      files={"file": ("notjson.json", b"not json", "application/json")})
        assert r2.status_code == 400
        r3 = api.post("/gmail/accounts/import-config", headers=H(api),
                      files={"file": ("cfg.txt", b"{}", "text/plain")})
        assert r3.status_code == 400

    def test_requires_auth(self, api):
        from starlette.testclient import TestClient
        anon = TestClient(api.app)
        assert anon.get("/gmail/accounts").status_code == 401
        assert anon.post("/gmail/accounts/clear-all").status_code == 401

    def test_csrf_required_on_mutations(self, api):
        r = api.post("/gmail/accounts/clear-all")
        assert r.status_code == 403
        r2 = api.patch("/gmail/accounts/1", json={})
        assert r2.status_code == 403


class TestBulkSmtpAccountsApi:
    """'Bulk Add via SMTP' — paste many 'email:app_password' lines at once.
    Real network SMTP logins are mocked here (unit-of-work level); the actual
    smtplib wiring is covered end-to-end in tests/test_gmail_service.py."""

    def _mock_verify_ok(self, monkeypatch):
        from app.services import gmail_service as gs
        monkeypatch.setattr(gs, "verify_smtp_login",
                            lambda email, pw: {"ok": True, "email_address": email,
                                                "detail": "SMTP login verified."})

    def _mock_verify_selective(self, monkeypatch, bad_emails):
        from app.services import gmail_service as gs
        def _fn(email, pw):
            if email in bad_emails:
                return {"ok": False, "detail": "SMTP authentication failed: bad creds"}
            return {"ok": True, "email_address": email, "detail": "ok"}
        monkeypatch.setattr(gs, "verify_smtp_login", _fn)

    def test_bulk_add_creates_accounts_with_real_verification(self, api, monkeypatch):
        self._mock_verify_ok(monkeypatch)
        lines = "one@gmail.com:abcd efgh ijkl mnop\ntwo@gmail.com:wxyz wxyz wxyz wxyz"
        r = api.post("/gmail/accounts/bulk-smtp", headers=H(api),
                     json={"lines": lines, "daily_limit": 100, "campaign_limit": 50})
        assert r.status_code == 200, r.text
        d = r.json()
        assert d["total"] == 2 and d["succeeded"] == 2 and d["failed"] == 0
        accs = api.get("/gmail/accounts").json()["accounts"]
        assert accs and {a["email"] for a in accs} == {"one@gmail.com", "two@gmail.com"}
        for a in accs:
            assert a["sending_method"] == "smtp"
            assert a["auth_provider"] == "smtp_password"
            assert a["daily_limit"] == 100 and a["campaign_limit"] == 50
            assert a["authorized"] is True

    def test_bulk_add_never_echoes_password(self, api, monkeypatch):
        self._mock_verify_ok(monkeypatch)
        r = api.post("/gmail/accounts/bulk-smtp", headers=H(api),
                     json={"lines": "secret@gmail.com:abcd efgh ijkl mnop"})
        assert "abcdefghijklmnop" not in r.text.lower().replace(" ", "")
        assert api.get("/gmail/accounts").json()["accounts"][0]["email"] == "secret@gmail.com"

    def test_bulk_add_rejects_bad_syntax_lines(self, api, monkeypatch):
        self._mock_verify_ok(monkeypatch)
        lines = "good@gmail.com:abcd efgh ijkl mnop\nnotanemail:short\njustoneword"
        r = api.post("/gmail/accounts/bulk-smtp", headers=H(api), json={"lines": lines})
        d = r.json()
        assert d["total"] == 3 and d["succeeded"] == 1 and d["failed"] == 2
        failed = [x for x in d["results"] if not x["ok"]]
        assert len(failed) == 2

    def test_bulk_add_reports_failed_smtp_login_without_saving(self, api, monkeypatch):
        self._mock_verify_selective(monkeypatch, {"bad@gmail.com"})
        lines = "good@gmail.com:abcd efgh ijkl mnop\nbad@gmail.com:wxyz wxyz wxyz wxyz"
        r = api.post("/gmail/accounts/bulk-smtp", headers=H(api), json={"lines": lines})
        d = r.json()
        assert d["succeeded"] == 1 and d["failed"] == 1
        emails = {a["email"] for a in api.get("/gmail/accounts").json()["accounts"]}
        assert emails == {"good@gmail.com"}  # the failed login was never persisted

    def test_bulk_add_updates_existing_account_in_place(self, api, monkeypatch, db):
        self._mock_verify_ok(monkeypatch)
        r1 = api.post("/gmail/accounts/bulk-smtp", headers=H(api),
                      json={"lines": "dup@gmail.com:abcd efgh ijkl mnop",
                            "daily_limit": 10, "campaign_limit": 5})
        assert r1.json()["succeeded"] == 1
        r2 = api.post("/gmail/accounts/bulk-smtp", headers=H(api),
                      json={"lines": "dup@gmail.com:newp assw ordh eree",
                            "daily_limit": 77, "campaign_limit": 33})
        assert r2.json()["succeeded"] == 1
        accs = api.get("/gmail/accounts").json()["accounts"]
        assert len(accs) == 1  # updated, not duplicated
        assert accs[0]["daily_limit"] == 77 and accs[0]["campaign_limit"] == 33

    def test_bulk_add_deduplicates_within_same_batch(self, api, monkeypatch):
        self._mock_verify_ok(monkeypatch)
        lines = "same@gmail.com:abcd efgh ijkl mnop\nsame@gmail.com:abcd efgh ijkl mnop"
        r = api.post("/gmail/accounts/bulk-smtp", headers=H(api), json={"lines": lines})
        d = r.json()
        assert d["succeeded"] == 1 and d["failed"] == 1  # 2nd flagged as duplicate
        assert api.get("/gmail/accounts").json()["count"] == 1

    def test_bulk_add_requires_auth_and_csrf(self, api):
        from starlette.testclient import TestClient
        anon = TestClient(api.app)
        assert anon.post("/gmail/accounts/bulk-smtp",
                         json={"lines": "a@gmail.com:x"}).status_code == 401
        r = api.post("/gmail/accounts/bulk-smtp", json={"lines": "a@gmail.com:x"})
        assert r.status_code == 403

    def test_bulk_add_rejects_empty_input(self, api):
        r = api.post("/gmail/accounts/bulk-smtp", headers=H(api), json={"lines": "   \n  \n"})
        assert r.status_code == 400

    def test_bulk_add_smtp_account_can_actually_send(self, api, monkeypatch, db):
        """The whole point of this feature: an account added via bulk-SMTP must
        be usable by the real send path (send_tasks.send_one) without any
        OAuth token — verifying the root-cause bug fix end-to-end."""
        self._mock_verify_ok(monkeypatch)
        api.post("/gmail/accounts/bulk-smtp", headers=H(api),
                json={"lines": "sender@gmail.com:abcd efgh ijkl mnop"})
        from app.models import GmailAccount
        acc = db.query(GmailAccount).filter_by(email="sender@gmail.com").one()
        assert acc.auth_provider == "smtp_password"
        assert acc.smtp_password == "abcdefghijklmnop"  # decrypts correctly
        from app.services.gmail_service import build_gmail_client_from_account
        client = build_gmail_client_from_account(acc, oauth_token=None)
        assert client.method == "smtp"
        assert client.smtp_password == "abcdefghijklmnop"


# --------------------------------------------------------------------------- test-send
class TestTestSendApi:

    def test_account_listing_authorized_only(self, api, db):
        _seed_account(db, user_id_of(api), email="ok@x.com", status="authorized")
        _seed_account(db, user_id_of(api), email="expired@x.com", status="expired")
        d = api.get("/test-send/accounts").json()
        assert [a["email"] for a in d["accounts"]] == ["ok@x.com"]

    def test_templates_listing(self, api, db):
        fixture = _seed_campaign_fixture(db, user_id_of(api))
        d = api.get("/test-send/templates").json()
        assert fixture["template_id"] in [t["id"] for t in d["templates"]]

    def test_send_without_oauth_token_honest_failure(self, api, db):
        aid = _seed_account(db, user_id_of(api), with_token=False)
        r = api.post("/test-send/send", headers=H(api),
                     json={"account_id": aid, "to_email": "dest@example.com",
                           "subject": "Hello", "body_html": "<p>Test</p>"})
        assert r.status_code == 200, r.text
        d = r.json()
        # REAL behavior: no stored token → honest failure, never fake success
        assert d["ok"] is False
        assert "No OAuth token" in d["detail"]
        assert d["stage"] in ("perm_failed", "temp_failed")

    def test_send_unknown_account_404(self, api):
        r = api.post("/test-send/send", headers=H(api),
                     json={"account_id": 999, "to_email": "dest@example.com",
                           "subject": "Hello", "body_html": "<p>Hi</p>"})
        assert r.status_code == 404

    def test_send_expired_account_409(self, api, db):
        aid = _seed_account(db, user_id_of(api), status="expired")
        r = api.post("/test-send/send", headers=H(api),
                     json={"account_id": aid, "to_email": "dest@example.com",
                           "subject": "Hello", "body_html": "<p>Hi</p>"})
        assert r.status_code == 409

    def test_send_requires_body_or_template(self, api, db):
        aid = _seed_account(db, user_id_of(api))
        r = api.post("/test-send/send", headers=H(api),
                     json={"account_id": aid, "to_email": "dest@example.com",
                           "subject": "Hello"})
        assert r.status_code == 400

    def test_send_substitutes_variables_before_send(self, api, db):
        # covers the pre-send pipeline end-to-end up to the real network call;
        # the call itself fails honestly without Google credentials
        aid = _seed_account(db, user_id_of(api))
        r = api.post("/test-send/send", headers=H(api),
                     json={"account_id": aid, "to_email": "dest@example.com",
                           "subject": "Invoice %RandomInvoiceNumber%",
                           "body_html": "<p>Dear %FirstName%, total %RandomTotal%.</p>",
                           "use_random_variables": True})
        assert r.status_code == 200, r.text
        d = r.json()
        assert d["ok"] is False  # no real Google creds in CI — honest failure
        assert "elapsed_ms" in d

    def test_send_csrf_and_auth(self, api, db):
        aid = _seed_account(db, user_id_of(api))
        no_csrf = api.post("/test-send/send",
                           json={"account_id": aid, "to_email": "d@x.com",
                                 "subject": "s", "body_html": "<p>x</p>"})
        assert no_csrf.status_code == 403
        from starlette.testclient import TestClient
        anon = TestClient(api.app)
        r = anon.post("/test-send/send",
                      json={"account_id": aid, "to_email": "d@x.com",
                            "subject": "s", "body_html": "<p>x</p>"})
        assert r.status_code == 401


# --------------------------------------------------------------------------- unsubscribe (public)
class TestUnsubscribeApi:

    def _seed(self, db, user_id):
        from app.models import Recipient, RecipientList
        rl = RecipientList(user_id=user_id, name="unsub-list")
        db.add(rl)
        db.flush()
        rec = Recipient(list_id=rl.id, email="victim@x.com",
                        first_name="Vic", last_name="Tim",
                        unsub_token="T" * 32, is_valid=True)
        db.add(rec)
        db.commit()
        return rec

    def test_landing_page_shows_email_and_form(self, api, db):
        rec = self._seed(db, user_id_of(api))
        r = api.get(f"/unsubscribe/{rec.unsub_token}")
        assert r.status_code == 200
        assert "victim@x.com" in r.text
        assert "Confirm unsubscribe" in r.text
        assert f'<form method="post" action="/unsubscribe/{rec.unsub_token}"' in r.text

    def test_short_token_rejected(self, api):
        r = api.get("/unsubscribe/shorttoken")
        assert r.status_code == 200
        assert "Invalid link" in r.text

    def test_unknown_token_invalid_link(self, api):
        r = api.get("/unsubscribe/{target}".format(target="Z" * 32))
        assert r.status_code == 200
        assert "Invalid link" in r.text

    def test_confirm_adds_suppression_and_event(self, api, db):
        # with campaign context: suppression + EmailEvent recorded
        from app.models import Campaign, CampaignRecipient
        rec = self._seed(db, user_id_of(api))
        camp = Campaign(user_id=user_id_of(api), name="unsub-camp",
                        recipient_list_id=rec.list_id, status="completed")
        db.add(camp)
        db.flush()
        db.add(CampaignRecipient(campaign_id=camp.id, recipient_id=rec.id,
                                 email=rec.email, seq=1, status="sent"))
        db.commit()
        r = api.post(f"/unsubscribe/{rec.unsub_token}")
        assert r.status_code == 200
        assert "has been added" in r.text
        from app.models import EmailEvent, SuppressionEntry
        row = db.query(SuppressionEntry).filter_by(email="victim@x.com").one()
        assert row.reason == "unsubscribed"
        assert row.source == "unsubscribe-link"
        assert row.user_id == user_id_of(api)
        ev = db.query(EmailEvent).filter_by(event_type="unsubscribed").all()
        assert ev and ev[0].campaign_recipient_id is not None
        db.refresh(rec)
        assert rec.status == "unsubscribed"

    def test_confirm_without_campaign_context_still_suppresses(self, api, db):
        # bare recipient, no campaign: suppression added, no event crash
        from app.models import SuppressionEntry
        rec = self._seed(db, user_id_of(api))
        r = api.post(f"/unsubscribe/{rec.unsub_token}")
        assert r.status_code == 200 and "has been added" in r.text
        assert db.query(SuppressionEntry).filter_by(email="victim@x.com").count() == 1

    def test_confirm_idempotent_no_duplicate_suppression(self, api, db):
        rec = self._seed(db, user_id_of(api))
        api.post(f"/unsubscribe/{rec.unsub_token}")
        api.post(f"/unsubscribe/{rec.unsub_token}")
        from app.models import SuppressionEntry
        assert db.query(SuppressionEntry).filter_by(email="victim@x.com").count() == 1

    def test_landing_already_unsubscribed_message(self, api, db):
        rec = self._seed(db, user_id_of(api))
        api.post(f"/unsubscribe/{rec.unsub_token}")
        r = api.get(f"/unsubscribe/{rec.unsub_token}")
        assert "Already unsubscribed" in r.text

    def test_status_json(self, api, db):
        from app.models import Recipient
        rec = self._seed(db, user_id_of(api))
        r = api.get(f"/unsubscribe-status/{rec.unsub_token}")
        assert r.status_code == 200
        d = r.json()
        assert d["subscribed"] is True and d["email_domain"] == "x.com"
        api.post(f"/unsubscribe/{rec.unsub_token}")
        r2 = api.get(f"/unsubscribe-status/{rec.unsub_token}")
        assert r2.json()["subscribed"] is False
        # invalid token → graceful JSON
        r3 = api.get("/unsubscribe-status/badtoken")
        assert r3.status_code == 200
        assert r3.json()["subscribed"] is False

    def test_public_flow_needs_no_auth(self, api, db):
        # the token IS the credential: anonymous client can unsubscribe
        from starlette.testclient import TestClient
        rec = self._seed(db, user_id_of(api))
        anon = TestClient(api.app)
        r = anon.post(f"/unsubscribe/{rec.unsub_token}")
        assert r.status_code == 200
        assert "has been added" in r.text


# --------------------------------------------------------------------------- campaigns
class TestCampaignsApi:

    def test_wizard_options_lists_everything(self, api, db):
        f = _seed_campaign_fixture(db, user_id_of(api))
        d = api.get("/campaigns/wizard/options").json()
        assert f["list_id"] in [l["id"] for l in d["lists"]]
        assert f["subject_id"] in [s["id"] for s in d["subjects"]]
        assert f["template_id"] in [t["id"] for t in d["templates"]]
        assert f["account_id"] in [a["id"] for a in d["accounts"]]
        assert "limits" in d and d["limits"]["max_recipients"] == 2000

    def test_create_and_get(self, api, db):
        f = _seed_campaign_fixture(db, user_id_of(api))
        r = api.post("/campaigns/create", headers=H(api), json=_cIn(f, name="My Camp"))
        assert r.status_code in (200, 201), r.text
        d = r.json()
        assert d["name"] == "My Camp" and d["status"] == "draft"
        assert d["subject_set_ids"] == [f["subject_id"]]
        r2 = api.get(f"/campaigns/{d['id']}")
        assert r2.status_code == 200 and r2.json()["name"] == "My Camp"

    def test_create_validation_errors_are_specific(self, api, db):
        f = _seed_campaign_fixture(db, user_id_of(api))
        # missing subjects
        r = api.post("/campaigns/create", headers=H(api),
                     json=_cIn(f, subject_set_ids=[]))
        assert r.status_code == 400
        errs = r.json()["detail"]["errors"]
        assert any("subject" in e.lower() for e in errs)
        # missing accounts
        r2 = api.post("/campaigns/create", headers=H(api),
                      json=_cIn(f, account_ids=[]))
        assert r2.status_code == 400
        assert any("Gmail account" in e for e in r2.json()["detail"]["errors"])
        # bad enum
        r3 = api.post("/campaigns/create", headers=H(api),
                      json=_cIn(f, rotation_strategy="bogus"))
        assert r3.status_code == 400
        # cross-user reference
        r4 = api.post("/campaigns/create", headers=H(api),
                      json=_cIn(f, subject_set_ids=[999]))
        assert r4.status_code == 400
        assert any("not found" in e for e in r4.json()["detail"]["errors"])

    def test_create_rejects_weights_without_weighted_strategy(self, api, db):
        f = _seed_campaign_fixture(db, user_id_of(api))
        r = api.post("/campaigns/create", headers=H(api),
                     json=_cIn(f, rotation_weights={str(f["account_id"]): 3}))
        assert r.status_code == 400

    def test_patch_campaign(self, api, db):
        f = _seed_campaign_fixture(db, user_id_of(api))
        cid = api.post("/campaigns/create", headers=H(api),
                       json=_cIn(f, name="Before")).json()["id"]
        r = api.post(f"/campaigns/{cid}/patch", headers=H(api),
                     json=_cIn(f, name="After", max_per_account=50))
        assert r.status_code == 200, r.text
        d = r.json()
        assert d["name"] == "After" and d["max_per_account"] == 50

    def test_patch_rejected_while_running(self, api, db):
        f = _seed_campaign_fixture(db, user_id_of(api))
        cid = api.post("/campaigns/create", headers=H(api),
                       json=_cIn(f, name="R")).json()["id"]
        from app.models import Campaign
        c = db.get(Campaign, cid)
        c.status = "running"
        db.commit()
        r = api.post(f"/campaigns/{cid}/patch", headers=H(api),
                     json=_cIn(f, name="Nope"))
        assert r.status_code == 409

    def test_delete_and_clear_all(self, api, db):
        f = _seed_campaign_fixture(db, user_id_of(api))
        c1 = api.post("/campaigns/create", headers=H(api),
                      json=_cIn(f, name="D1")).json()["id"]
        c2 = api.post("/campaigns/create", headers=H(api),
                      json=_cIn(f, name="D2")).json()["id"]
        c3 = api.post("/campaigns/create", headers=H(api),
                      json=_cIn(f, name="D3")).json()["id"]
        r = api.post(f"/campaigns/{c1}/delete", headers=H(api))
        assert r.status_code == 200 and r.json()["deleted"] == c1
        # running campaign cannot be deleted
        from app.models import Campaign
        c = db.get(Campaign, c2)
        c.status = "running"
        db.commit()
        r2 = api.post(f"/campaigns/{c2}/delete", headers=H(api))
        assert r2.status_code == 409
        # clear-all skips running, deletes the rest
        r3 = api.post("/campaigns/clear-all", headers=H(api))
        assert r3.status_code == 200
        d = r3.json()
        assert d["deleted"] >= 1 and d["skipped"] >= 1
        remaining = api.get("/campaigns/list").json()
        assert remaining["total"] == 1  # only the running one survives
        assert remaining["campaigns"][0]["id"] == c2

    def test_list_filters(self, api, db):
        f = _seed_campaign_fixture(db, user_id_of(api))
        for nm in ("alpha one", "beta two"):
            api.post("/campaigns/create", headers=H(api), json=_cIn(f, name=nm))
        allr = api.get("/campaigns/list").json()
        assert allr["total"] == 2
        s = api.get("/campaigns/list", params={"search": "alpha"}).json()
        assert s["total"] == 1 and s["campaigns"][0]["name"] == "alpha one"
        st = api.get("/campaigns/list", params={"status": "draft"}).json()
        assert st["total"] == 2

    def test_validate_step_errors(self, api):
        r = api.post("/campaigns/validate-step", headers=H(api),
                     json={"step": 1, "data": {"name": ""}})
        assert r.status_code == 200
        assert any("name" in e.lower() for e in r.json()["errors"])
        r2 = api.post("/campaigns/validate-step", headers=H(api),
                      json={"step": 3, "data": {"subject_set_ids": []}})
        assert any("subject" in e.lower() for e in r2.json()["errors"])
        r3 = api.post("/campaigns/validate-step", headers=H(api),
                      json={"step": 9, "data": {"max_per_account": 99999}})
        assert any("max_per_account" in e for e in r3.json()["errors"])
        r4 = api.post("/campaigns/validate-step", headers=H(api),
                      json={"step": 11, "data": {"unsubscribe_enabled": False}})
        assert any("unsubscribe" in e.lower() for e in r4.json()["errors"])
        ok = api.post("/campaigns/validate-step", headers=H(api),
                      json={"step": 1, "data": {"name": "Good"}})
        assert ok.json()["errors"] == []

    def test_preview_renders_real_messages(self, api, db):
        f = _seed_campaign_fixture(db, user_id_of(api), n_recipients=3)
        cid = api.post("/campaigns/create", headers=H(api),
                       json=_cIn(f, name="Prev")).json()["id"]
        r = api.post(f"/campaigns/{cid}/preview?sample_size=3", headers=H(api))
        assert r.status_code == 200, r.text
        d = r.json()
        assert d["sample_size"] == 3
        msgs = d["messages"]
        assert len(msgs) == 3
        m0 = msgs[0]
        assert "Camp1" in m0["body_html"]          # substituted first name
        assert "%FirstName%" not in m0["body_html"]  # no raw tokens remain
        assert "camp-acc" in m0["from_account"]     # real account email
        assert m0["subject"]                        # subject rendered
        # invoice numbers unique per recipient
        invs = [m["raw_variables"]["invoice_number"] for m in msgs]
        assert len(set(invs)) == len(invs)
        assert all(m["raw_variables"]["amount"] is not None for m in msgs)
        # preview must NOT create campaign_recipients (nothing persisted)
        from app.models import CampaignRecipient
        assert db.query(CampaignRecipient).count() == 0

    def test_preview_requires_list_with_valid_recipients(self, api, db):
        f = _seed_campaign_fixture(db, user_id_of(api), n_recipients=1)
        cid = api.post("/campaigns/create", headers=H(api),
                       json=_cIn(f, name="P2")).json()["id"]
        from app.models import Recipient
        for rec in db.query(Recipient).all():
            rec.is_valid = False
        db.commit()
        r = api.post(f"/campaigns/{cid}/preview", headers=H(api))
        assert r.status_code == 400

    @requires_broker
    def test_start_materializes_and_queues(self, api, db):
        f = _seed_campaign_fixture(db, user_id_of(api), n_recipients=3)
        cid = api.post("/campaigns/create", headers=H(api),
                       json=_cIn(f, name="Start Me")).json()["id"]
        r = api.post(f"/campaigns/{cid}/start", headers=H(api))
        assert r.status_code == 200, r.text
        d = r.json()
        assert d["status"] == "running" and d["queued"] == 3
        from app.models import CampaignRecipient, SendJob
        crs = db.query(CampaignRecipient).filter_by(campaign_id=cid).all()
        assert len(crs) == 3
        assert all(cr.status == "queued" for cr in crs)
        assert db.query(SendJob).filter_by(campaign_id=cid).count() == 3
        # each campaign_recipient has generated values stored
        assert all(cr.generated_invoice_number for cr in crs)

    @requires_broker
    def test_pause_resume_cancel_flow(self, api, db):
        f = _seed_campaign_fixture(db, user_id_of(api), n_recipients=2)
        cid = api.post("/campaigns/create", headers=H(api),
                       json=_cIn(f, name="PRC")).json()["id"]
        api.post(f"/campaigns/{cid}/start", headers=H(api))
        r = api.post(f"/campaigns/{cid}/pause", headers=H(api))
        assert r.status_code == 200 and r.json()["status"] == "paused"
        # pause is idempotent-guarded: cannot pause a paused campaign
        r2 = api.post(f"/campaigns/{cid}/pause", headers=H(api))
        assert r2.status_code == 409
        r3 = api.post(f"/campaigns/{cid}/resume", headers=H(api))
        assert r3.status_code == 200 and r3.json()["status"] == "running"
        r4 = api.post(f"/campaigns/{cid}/cancel", headers=H(api))
        assert r4.status_code == 200 and r4.json()["status"] == "cancelled"

    def test_pause_non_running_409(self, api, db):
        f = _seed_campaign_fixture(db, user_id_of(api))
        cid = api.post("/campaigns/create", headers=H(api),
                       json=_cIn(f, name="Draft")).json()["id"]
        r = api.post(f"/campaigns/{cid}/pause", headers=H(api))
        assert r.status_code == 409
        r2 = api.post(f"/campaigns/{cid}/resume", headers=H(api))
        assert r2.status_code == 409

    @requires_broker
    def test_resume_requires_pending(self, api, db):
        f = _seed_campaign_fixture(db, user_id_of(api), n_recipients=2)
        cid = api.post("/campaigns/create", headers=H(api),
                       json=_cIn(f, name="RQ")).json()["id"]
        api.post(f"/campaigns/{cid}/start", headers=H(api))
        api.post(f"/campaigns/{cid}/pause", headers=H(api))
        # mark all recipients sent → resume has nothing to queue
        from app.models import CampaignRecipient
        for cr in db.query(CampaignRecipient).filter_by(campaign_id=cid).all():
            cr.status = "sent"
        db.commit()
        r = api.post(f"/campaigns/{cid}/resume", headers=H(api))
        assert r.status_code == 409

    @requires_broker
    def test_progress_after_start(self, api, db):
        f = _seed_campaign_fixture(db, user_id_of(api), n_recipients=3)
        cid = api.post("/campaigns/create", headers=H(api),
                       json=_cIn(f, name="Prog")).json()["id"]
        api.post(f"/campaigns/{cid}/start", headers=H(api))
        d = api.get(f"/campaigns/{cid}/progress").json()
        assert d["campaign_id"] == cid
        assert d["status"] == "running"
        assert d["total"] == 3 and d["queued"] == 3
        assert d["jobs_total"] == 3
        assert d["progress_pct"] == 0.0

    @requires_broker
    def test_recipients_monitor_table(self, api, db):
        f = _seed_campaign_fixture(db, user_id_of(api), n_recipients=3)
        cid = api.post("/campaigns/create", headers=H(api),
                       json=_cIn(f, name="Mon")).json()["id"]
        api.post(f"/campaigns/{cid}/start", headers=H(api))
        d = api.get(f"/campaigns/{cid}/recipients").json()
        assert d["total"] == 3
        rows = d["recipients"]
        assert [r["seq"] for r in rows] == [1, 2, 3]
        r0 = rows[0]
        assert {"seq", "email", "status", "account", "subject", "template",
                "invoice_number", "amount", "attempts", "message_id"} <= set(r0)
        assert r0["account"] and "camp-acc" in r0["account"]
        assert r0["subject"]  # subject text resolved
        # filters work
        s = api.get(f"/campaigns/{cid}/recipients", params={"status": "queued"}).json()
        assert s["total"] == 3
        s2 = api.get(f"/campaigns/{cid}/recipients", params={"search": "camp2"}).json()
        assert s2["total"] == 1

    @requires_broker
    def test_export_csv(self, api, db):
        f = _seed_campaign_fixture(db, user_id_of(api), n_recipients=2)
        cid = api.post("/campaigns/create", headers=H(api),
                       json=_cIn(f, name="Export Me")).json()["id"]
        api.post(f"/campaigns/{cid}/start", headers=H(api))
        r = api.get(f"/campaigns/{cid}/export")
        assert r.status_code == 200
        assert "text/csv" in r.headers["content-type"]
        rows = list(csv.reader(io.StringIO(r.text)))
        assert rows[0][0] == "seq"
        assert len(rows) == 3  # header + 2 recipients
        assert rows[1][1] == "camp1@x.com"
        assert rows[1][2] == "queued"

    @requires_broker
    def test_sse_stream_first_events(self, api, db, db_engine):
        """SSE contract end-to-end. starlette's TestClient runs the whole ASGI app
        to completion before returning the response, so an endless stream would
        deadlock the client. A background thread marks the campaign completed
        mid-stream: the generator then emits connected -> progress -> keepalive(s)
        -> progress -> done and terminates on its own."""
        from app.models import Campaign
        from sqlalchemy.orm import Session as SyncSess
        import threading, time as _time

        f = _seed_campaign_fixture(db, user_id_of(api), n_recipients=2)
        cid = api.post("/campaigns/create", headers=H(api),
                       json=_cIn(f, name="SSE")).json()["id"]
        r = api.post(f"/campaigns/{cid}/start", headers=H(api))
        assert r.status_code == 200, r.text

        def flip_to_completed_after(seconds: float):
            _time.sleep(seconds)
            with SyncSess(db_engine, expire_on_commit=False) as s:
                c = s.get(Campaign, cid)
                if c is not None:
                    c.status = "completed"
                    s.commit()

        th = threading.Thread(target=flip_to_completed_after, args=(3.0,), daemon=True)
        th.start()

        frames = []
        t0 = _time.monotonic()
        with api.stream("GET", f"/campaigns/{cid}/stream") as resp:
            assert resp.status_code == 200
            assert resp.headers["content-type"].startswith("text/event-stream")
            for line in resp.iter_lines():
                frames.append(line)
                assert _time.monotonic() - t0 < 30, frames  # safety net
        th.join(timeout=5)
        assert any("event: connected" in fr for fr in frames), frames
        assert any("event: progress" in fr for fr in frames), frames
        assert any("event: done" in fr for fr in frames), frames
        # the done frame carries the live counter payload
        done_idx = next(i for i, fr in enumerate(frames) if "event: done" in fr)
        done_data = frames[done_idx + 1]
        assert done_data.startswith("data: "), done_data
        import json as _json
        payload = _json.loads(done_data[len("data: "):])
        assert payload["status"] == "completed"
        assert payload["total"] == 2

    def test_campaign_isolation(self, api, db, second_user):
        f = _seed_campaign_fixture(db, user_id_of(api))
        cid = api.post("/campaigns/create", headers=H(api),
                       json=_cIn(f, name="Mine")).json()["id"]
        assert second_user.get(f"/campaigns/{cid}").status_code == 404
        r = second_user.post(f"/campaigns/{cid}/start", headers=H(second_user))
        assert r.status_code == 404

    def test_requires_auth_and_csrf(self, api):
        from starlette.testclient import TestClient
        anon = TestClient(api.app)
        assert anon.get("/campaigns/list").status_code == 401
        assert anon.post("/campaigns/create", json={"name": "x"}).status_code == 401
        r = api.post("/campaigns/create", json={"name": "NoCsrf"})
        assert r.status_code == 403
        r2 = api.post("/campaigns/validate-step", json={"step": 1, "data": {}})
        assert r2.status_code == 403