"""send_one task integration tests: real DB, real Redis, mocked Gmail transport.

Runs the Celery task body INLINE (send_one(cid, crid) — celery Task.__call__)
against the real test database so every status transition, counter update,
attempt row, email event, retry/backoff reschedule and dead-letter push is
verified exactly as a worker would perform it. Only the Gmail transport
(build_gmail_client_from_account) is mocked — everything else is the real code.

NOTE: app imports happen INSIDE tests/fixtures (never at module top level) so
conftest's test_app fixture has already reloaded app.db.session/app.main with
the test DATABASE_URL before app.tasks.send_tasks binds its SyncSession.
"""
import secrets
import uuid
from datetime import datetime, timedelta, timezone

import pytest


# ---------------------------------------------------------------------------
# Seeding helpers (sync Session on the shared test DB, like the engine tests)
# ---------------------------------------------------------------------------
def _make_user(db) -> "User":
    from app.models import User
    u = User(email=f"send{secrets.token_hex(4)}@x.com",
             password_hash="x" * 60, full_name="Send")
    db.add(u)
    db.commit()
    return u


def _make_list(db, user_id, n=1) -> "RecipientList":
    from app.models import Recipient, RecipientList
    rl = RecipientList(name=f"list-{uuid.uuid4().hex[:6]}", user_id=user_id)
    db.add(rl)
    db.commit()
    recs = []
    for i in range(n):
        r = Recipient(list_id=rl.id, email=f"send{i}-{uuid.uuid4().hex[:6]}@x.com",
                      first_name="Sendy", last_name=f"Mc{i}",
                      unsub_token=secrets.token_urlsafe(24))
        db.add(r)
        recs.append(r)
    db.commit()
    return rl, recs


def _make_subject(db, user_id, text="Payment reminder %RandomInvoiceNumber%"):
    from app.models import SubjectLine
    s = SubjectLine(user_id=user_id, subject=text)
    db.add(s)
    db.commit()
    return s


def _make_template(db, user_id, body_html=None):
    from app.models import EmailTemplate
    t = EmailTemplate(user_id=user_id, name=f"tpl-{uuid.uuid4().hex[:6]}",
                      subject_hint="hint",
                      body_html=body_html or
                      "<p>Hi %FirstName%, total %RandomTotal%. "
                      "Invoice %RandomInvoiceNumber%.</p>")
    db.add(t)
    db.commit()
    return t


def _make_account(db, user_id, status="authorized", **over):
    from app.models import GmailAccount, OAuthToken
    a = GmailAccount(user_id=user_id, email=f"send-acc-{uuid.uuid4().hex[:8]}@gmail.com",
                     status=status, sending_method="api",
                     is_active=(status == "authorized"))
    for k, v in over.items():
        setattr(a, k, v)
    db.add(a)
    db.commit()
    if status == "authorized":
        db.add(OAuthToken(account_id=a.id, access_token=f"tok-{uuid.uuid4().hex[:6]}",
                          refresh_token=f"ref-{uuid.uuid4().hex[:6]}",
                          expires_at=datetime.now(timezone.utc) + timedelta(hours=1)))
        db.commit()
    return a


def _seed_sendable(db, **camp_over):
    """User + list(1) + subject + template + authorized account + campaign +
    materialized CampaignRecipient (pending) + SendJob. Returns ids dict."""
    from app.models import Campaign, CampaignRecipient, SendJob
    u = _make_user(db)
    rl, recs = _make_list(db, u.id, n=1)
    subj = _make_subject(db, u.id)
    tpl = _make_template(db, u.id)
    acc = _make_account(db, u.id)
    kw = dict(
        user_id=u.id, name=f"send-{uuid.uuid4().hex[:6]}", status="running",
        recipient_list_id=rl.id, subject_set_ids=[subj.id], template_ids=[tpl.id],
        account_ids=[acc.id],
        send_interval_seconds=0,  # disable per-account Redis rate window by default
    )
    kw.update(camp_over)
    c = Campaign(**kw)
    db.add(c)
    db.commit()
    cr = CampaignRecipient(campaign_id=c.id, recipient_id=recs[0].id, email=recs[0].email,
                           status="pending", seq=1,
                           generated_subject_id=subj.id, generated_template_id=tpl.id,
                           generated_invoice_number="INV-TEST-0001",
                           generated_amount=123.45, generated_shipping=9.99,
                           generated_phone="+15550001",
                           assigned_account_id=acc.id)
    db.add(cr)
    db.commit()
    job = SendJob(campaign_id=c.id, campaign_recipient_id=cr.id, status="queued",
                  max_attempts=(camp_over.get("max_retries") or 3) + 1)
    db.add(job)
    db.commit()
    return {"user": u, "campaign": c, "cr": cr, "job": job, "account": acc,
            "recipient": recs[0], "subject": subj, "template": tpl}


# ---------------------------------------------------------------------------
# Fake Gmail clients (only the transport is faked; everything else is real)
# ---------------------------------------------------------------------------
class _OkClient:
    """Accepts the message like Gmail would; records the MIME for assertions."""
    method = "api"

    def __init__(self):
        self.sent = []

    def send_mime_api(self, mime):
        self.sent.append(mime)
        return {"ok": True, "message_id": "MSG-FAKE-123",
                "thread_id": "TH-FAKE-456",
                "raw_response": {"id": "MSG-FAKE-123", "threadId": "TH-FAKE-456",
                                 "labelIds": ["SENT"]}}


def _http_error(status: int, body: bytes):
    import googleapiclient.errors
    import httplib2
    resp = httplib2.Response({"status": status, "reason": "Error"})
    return googleapiclient.errors.HttpError(resp, body)


class _TempErrorClient:
    """Generic 5xx backend error — temporary, does NOT trigger the account
    cooldown branch (that's for quota/429; see _QuotaClient tests)."""
    method = "api"

    def send_mime_api(self, mime):
        raise _http_error(500, b'{"error": {"message": "Backend Error, try again"}}')


class _PermErrorClient:
    method = "api"

    def send_mime_api(self, mime):
        raise _http_error(403, b'{"error": {"message": "Forbidden - policy violation"}}')


class _BoomClient:
    """Generic exception (connection-level) -> classified temporary."""
    method = "smtp"

    def send_mime_smtp(self, mime):
        raise ConnectionError("Connection reset by peer")


class _RefusedClient:
    """Provider returns ok=False without raising."""
    method = "smtp"

    def send_mime_smtp(self, mime):
        return {"ok": False, "error": "SMTP refused recipients: somebody@example.com"}


class _Recorder:
    """Captures apply_async reschedules instead of touching the broker."""

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

    def __call__(self, *args, **kwargs):
        self.calls.append({"args": args, **kwargs})


@pytest.fixture(autouse=True, scope="module")
def _dispose_task_engine():
    """send_tasks binds the app's pooled SYNC engine at import time; its open
    pool would keep server-side connections alive past the last test and block
    the session-scoped DROP DATABASE teardown (psycopg ObjectInUse). Dispose it
    once, after this module's last test finishes."""
    yield
    try:
        from app.db.session import sync_engine
        sync_engine.dispose()
    except Exception:
        pass


@pytest.fixture()
def send_mod(test_app, monkeypatch):
    """(Re)import send_tasks bound to the test DB.

    Other test modules (e.g. test_security_rotation.py) do a top-level
    `from app.tasks.send_tasks import classify_error`, which runs during
    pytest COLLECTION — before the session-scoped test_app fixture creates the
    test database and reloads app.db.session. That first import caches
    app.tasks.send_tasks in sys.modules with its SyncSession bound to the
    pre-test engine. A later plain `import app.tasks.send_tasks` just returns
    that stale cached module, so send_one() would look up campaigns in the
    wrong database ("campaign deleted" for rows that were just seeded).
    Force a reload here (after depending on test_app) so SyncSession is
    rebound to the current (test) app.db.session.sync_engine.
    """
    import importlib
    import app.tasks.send_tasks as st
    importlib.reload(st)
    recorder = _Recorder()
    monkeypatch.setattr(st.send_one, "apply_async", recorder)
    yield st, recorder
    # monkeypatch auto-undoes


def _call(st, campaign_id, cr_id):
    return st.send_one(campaign_id, cr_id)


# ---------------------------------------------------------------------------
# Early exits & guard paths
# ---------------------------------------------------------------------------
class TestGuardPaths:
    def test_campaign_deleted_is_cancelled(self, db_engine, send_mod):
        from sqlalchemy.orm import Session
        st, _ = send_mod
        with Session(db_engine, expire_on_commit=False) as db:
            f = _seed_sendable(db)
            cid, crid = f["campaign"].id, f["cr"].id
            db.delete(f["campaign"])
            db.commit()
        out = _call(st, cid, crid)
        assert out["ok"] is False and out["status"] == "cancelled"
        assert out["reason"] == "campaign deleted"

    def test_paused_campaign_short_circuits(self, db_engine, send_mod):
        from sqlalchemy.orm import Session
        from app.models import Campaign
        st, _ = send_mod
        with Session(db_engine, expire_on_commit=False) as db:
            f = _seed_sendable(db)
            f["campaign"].status = "paused"
            db.commit()
            cid, crid = f["campaign"].id, f["cr"].id
        out = _call(st, cid, crid)
        assert out["ok"] is False and out["status"] == "cancelled"
        assert "paused" in out["reason"]
        with Session(db_engine, expire_on_commit=False) as db:
            cr = db.get(__import__("app.models", fromlist=["CampaignRecipient"]).CampaignRecipient, crid)
            assert cr.status == "pending"  # untouched

    def test_already_finalized_recipient_no_resend(self, db_engine, send_mod):
        from sqlalchemy.orm import Session
        from app.models import CampaignRecipient
        st, _ = send_mod
        with Session(db_engine, expire_on_commit=False) as db:
            f = _seed_sendable(db)
            f["cr"].status = "sent"
            db.commit()
            cid, crid = f["campaign"].id, f["cr"].id
        out = _call(st, cid, crid)
        assert out == {"ok": True, "status": "sent", "reason": "already finalized"}

    def test_suppression_hit_at_send_time(self, db_engine, send_mod):
        from sqlalchemy.orm import Session
        from app.models import Campaign, CampaignRecipient, EmailEvent, SuppressionEntry
        st, _ = send_mod
        with Session(db_engine, expire_on_commit=False) as db:
            f = _seed_sendable(db)
            db.add(SuppressionEntry(email=f["cr"].email, reason="unsubscribed",
                                    source="test", user_id=f["user"].id))
            db.commit()
            cid, crid = f["campaign"].id, f["cr"].id
        out = _call(st, cid, crid)
        assert out["ok"] is True and out["status"] == "suppressed"
        with Session(db_engine, expire_on_commit=False) as db:
            cr = db.get(CampaignRecipient, crid)
            camp = db.get(Campaign, cid)
            assert cr.status == "suppressed"
            assert "Suppressed" in cr.last_error
            assert camp.suppressed_count == 1
            ev = db.query(EmailEvent).filter_by(campaign_recipient_id=crid,
                                                event_type="suppressed").one()
            assert ev is not None

    def test_unauthorized_assigned_account_skips(self, db_engine, send_mod):
        from sqlalchemy.orm import Session
        from app.models import Campaign, CampaignRecipient
        st, _ = send_mod
        with Session(db_engine, expire_on_commit=False) as db:
            u = _make_user(db)
            rl, recs = _make_list(db, u.id, n=1)
            subj = _make_subject(db, u.id)
            tpl = _make_template(db, u.id)
            bad = _make_account(db, u.id, status="expired")
            from app.models import Campaign, CampaignRecipient, SendJob
            c = Campaign(user_id=u.id, name=f"x-{uuid.uuid4().hex[:6]}", status="running",
                         recipient_list_id=rl.id, subject_set_ids=[subj.id],
                         template_ids=[tpl.id], account_ids=[bad.id],
                         send_interval_seconds=0)
            db.add(c)
            db.commit()
            cr = CampaignRecipient(campaign_id=c.id, recipient_id=recs[0].id,
                                   email=recs[0].email, status="pending", seq=1,
                                   generated_subject_id=subj.id,
                                   generated_template_id=tpl.id,
                                   assigned_account_id=bad.id)
            db.add(cr)
            db.commit()
            cid, crid = c.id, cr.id
        out = _call(st, cid, crid)
        assert out["ok"] is False and out["status"] == "skipped"
        with Session(db_engine, expire_on_commit=False) as db:
            cr = db.get(CampaignRecipient, crid)
            camp = db.get(Campaign, cid)
            assert cr.status == "skipped"
            assert "unavailable" in cr.last_error
            assert camp.skipped_count == 1

    def test_missing_template_perm_fails(self, db_engine, send_mod):
        from sqlalchemy.orm import Session
        from app.models import Campaign, CampaignRecipient
        st, _ = send_mod
        with Session(db_engine, expire_on_commit=False) as db:
            f = _seed_sendable(db)
            f["cr"].generated_template_id = None
            db.commit()
            cid, crid = f["campaign"].id, f["cr"].id
        out = _call(st, cid, crid)
        assert out["ok"] is False and out["status"] == "perm_failed"
        with Session(db_engine, expire_on_commit=False) as db:
            cr = db.get(CampaignRecipient, crid)
            camp = db.get(Campaign, cid)
            assert cr.status == "perm_failed"
            assert "template" in cr.last_error.lower()
            assert camp.failed_count == 1

    def test_missing_oauth_token_perm_fails(self, db_engine, send_mod):
        from sqlalchemy.orm import Session
        from app.models import Campaign, CampaignRecipient, OAuthToken
        st, _ = send_mod
        with Session(db_engine, expire_on_commit=False) as db:
            f = _seed_sendable(db)
            db.query(OAuthToken).filter_by(account_id=f["account"].id).delete()
            db.commit()
            cid, crid = f["campaign"].id, f["cr"].id
        out = _call(st, cid, crid)
        assert out["ok"] is False and out["status"] == "perm_failed"
        assert "token" in out["reason"].lower()
        with Session(db_engine, expire_on_commit=False) as db:
            cr = db.get(CampaignRecipient, crid)
            assert cr.status == "perm_failed"


# ---------------------------------------------------------------------------
# Success path (full real pipeline, only transport faked)
# ---------------------------------------------------------------------------
class TestSuccessPath:
    def test_sent_end_to_end_with_mime_and_headers(self, db_engine, send_mod, monkeypatch):
        from sqlalchemy.orm import Session
        from app.models import Campaign, CampaignRecipient, EmailEvent, SendAttempt
        st, rec = send_mod
        ok_client = _OkClient()
        monkeypatch.setattr(st, "build_gmail_client_from_account",
                            lambda account, token, client_config=None: ok_client)
        with Session(db_engine, expire_on_commit=False) as db:
            f = _seed_sendable(db)
            unsub_token = f["recipient"].unsub_token
            cid, crid, acc_email = f["campaign"].id, f["cr"].id, f["account"].email
            cr_email = f["cr"].email
            subj_text = f["subject"].subject
        out = _call(st, cid, crid)
        assert out == {"ok": True, "status": "sent", "message_id": "MSG-FAKE-123",
                       "thread_id": "TH-FAKE-456"}
        # MIME captured by the fake transport — assert real headers/content
        assert len(ok_client.sent) == 1
        mime = ok_client.sent[0]
        assert mime["From"] == acc_email            # never spoofed
        assert mime["To"] == cr_email
        # Subject is variable-rendered with the same engine as the body —
        # %RandomInvoiceNumber% resolves to the per-recipient generated number
        assert mime["Subject"] == subj_text.replace(
            "%RandomInvoiceNumber%", "INV-TEST-0001")
        lu = mime["List-Unsubscribe"]
        assert unsub_token in lu and lu.startswith("<http")
        assert mime["List-Unsubscribe-Post"] == "List-Unsubscribe=One-Click"
        body = mime.as_string()
        # MIMEText uses base64 CTE — decode payload parts before searching
        import email as _email
        from email import policy as _policy
        _msg = _email.message_from_string(body, policy=_policy.default)
        decoded = " ".join(p.get_payload(decode=True).decode("utf-8", errors="replace")
                           for p in _msg.walk() if p.get_content_type() in ("text/plain", "text/html"))
        assert "Sendy" in decoded                # %FirstName% substituted
        assert "INV-TEST-0001" in decoded        # invoice number substituted
        assert "%FirstName%" not in decoded and "%RandomTotal%" not in decoded
        with Session(db_engine, expire_on_commit=False) as db:
            cr = db.get(CampaignRecipient, crid)
            camp = db.get(Campaign, cid)
            assert cr.status == "sent"
            assert cr.msg_id == "MSG-FAKE-123"
            assert cr.thread_id == "TH-FAKE-456"
            assert cr.sent_at is not None
            assert cr.last_error == ""
            assert cr.attempt_count == 1
            assert camp.sent_count == 1
            att = db.query(SendAttempt).filter_by(campaign_id=cid).one()
            assert att.outcome == "accepted"
            assert att.message_id == "MSG-FAKE-123"
            assert att.method == "api"
            assert att.retry_no == 0
            assert "refresh_token" not in (att.provider_response or "")
            ev = db.query(EmailEvent).filter_by(campaign_id=cid,
                                                event_type="sent").one()
            assert ev.message_id == "MSG-FAKE-123"
            # account usage counters
            from app.models import GmailAccount
            acc = db.query(GmailAccount).filter_by(email=acc_email).one()
            assert acc.sends_today == 1
            assert acc.last_send_date is not None
            assert acc.last_success_at is not None
        assert rec.calls == []  # no reschedules on success

    def test_pdf_attachment_generated_and_sent(self, db_engine, send_mod, monkeypatch):
        """attach_mode=pdf renders the PDF template with variables via real
        WeasyPrint and attaches a real PDF to the MIME message."""
        from sqlalchemy.orm import Session
        from app.models import EmailTemplate
        st, _ = send_mod
        ok_client = _OkClient()
        monkeypatch.setattr(st, "build_gmail_client_from_account",
                            lambda account, token, client_config=None: ok_client)
        with Session(db_engine, expire_on_commit=False) as db:
            f = _seed_sendable(db, attach_mode="pdf", attach_pdf=True)
            pdf_tpl = EmailTemplate(
                user_id=f["user"].id, name=f"pdftpl-{uuid.uuid4().hex[:6]}",
                subject_hint="att",
                body_html="<h1>Invoice %RandomInvoiceNumber%</h1>"
                          "<p>Customer: %FirstName%<br>Amount: %RandomTotal%</p>")
            db.add(pdf_tpl)
            db.commit()
            f["campaign"].attach_pdf_template_id = pdf_tpl.id
            db.commit()
            cid, crid = f["campaign"].id, f["cr"].id
        out = _call(st, cid, crid)
        assert out["ok"] is True and out["status"] == "sent"
        mime = ok_client.sent[0]
        raw = mime.as_string()
        assert "Invoice_INV-TEST-0001.pdf" in raw
        # extract the attachment payload and verify it is a real PDF with the
        # substituted content inside (streams are Flate-compressed, so extract
        # text with a real PDF text extractor — poppler's pdftotext)
        import base64
        import subprocess
        import tempfile
        from email import policy
        msg = __import__("email").message_from_string(raw, policy=policy.default)
        pdf_part = next(p for p in msg.iter_attachments()
                        if p.get_filename() == "Invoice_INV-TEST-0001.pdf")
        data = pdf_part.get_payload(decode=True)
        assert data[:5] == b"%PDF-"
        with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as tf:
            tf.write(data)
            pdf_path = tf.name
        try:
            txt = subprocess.run(["pdftotext", pdf_path, "-"],
                                 capture_output=True, timeout=30).stdout.decode()
        finally:
            import os
            os.unlink(pdf_path)
        assert "Invoice INV-TEST-0001" in txt
        assert "Sendy" in txt               # substituted inside the PDF
        assert "123.45" in txt              # %RandomTotal% substituted in PDF

    def test_plain_message_type_omits_html_part(self, db_engine, send_mod, monkeypatch):
        from sqlalchemy.orm import Session
        st, _ = send_mod
        ok_client = _OkClient()
        monkeypatch.setattr(st, "build_gmail_client_from_account",
                            lambda account, token, client_config=None: ok_client)
        with Session(db_engine, expire_on_commit=False) as db:
            f = _seed_sendable(db, message_type="plain")
            cid, crid = f["campaign"].id, f["cr"].id
        out = _call(st, cid, crid)
        assert out["ok"] is True and out["status"] == "sent"
        msg = __import__("email").message_from_string(ok_client.sent[0].as_string())
        parts = [p.get_content_type() for p in msg.walk() if p.get_content_maintype() == "text"]
        assert "text/plain" in parts
        # html omitted when message_type == plain (empty html_body passed)


# ---------------------------------------------------------------------------
# Error handling: temporary -> retry/backoff; exhausted -> dead letter
# ---------------------------------------------------------------------------
class TestRetryAndBackoff:
    def test_temporary_error_retries_with_exponential_backoff(self, db_engine,
                                                              send_mod, monkeypatch):
        from sqlalchemy.orm import Session
        from app.models import Campaign, CampaignRecipient, SendAttempt, SendJob
        st, rec = send_mod
        monkeypatch.setattr(st, "build_gmail_client_from_account",
                            lambda account, token, client_config=None: _TempErrorClient())
        with Session(db_engine, expire_on_commit=False) as db:
            f = _seed_sendable(db, max_retries=3, retry_delay_seconds=60)
            cid, crid, jid = f["campaign"].id, f["cr"].id, f["job"].id
        out1 = _call(st, cid, crid)
        assert out1["ok"] is False and out1["status"] == "retrying"
        assert len(rec.calls) == 1
        cd1 = rec.calls[0]["countdown"]
        assert 61 <= cd1 <= 66          # 60*2^0 + jitter(0..5) + 1
        assert rec.calls[0]["args"] == [cid, crid]
        with Session(db_engine, expire_on_commit=False) as db:
            cr = db.get(CampaignRecipient, crid)
            job = db.get(SendJob, jid)
            assert cr.status == "queued"
            assert "temporary error" in cr.last_error.lower()
            assert cr.attempt_count == 1
            assert job.status == "retrying"
            assert job.retry_after is not None
            att = db.query(SendAttempt).filter_by(campaign_id=cid).one()
            assert att.outcome == "temp_failed"
            assert att.error_class == "temporary"
            assert att.retry_no == 0

    def test_retry_loop_then_dead_letter_after_max_attempts(self, db_engine,
                                                            send_mod, monkeypatch):
        """4 consecutive temporary failures with max_retries=3: attempts 1-3
        reschedule with doubling backoff, attempt 4 dead-letters (no infinite
        retry), pushes to the REAL Redis DLQ and marks the job dead."""
        from sqlalchemy.orm import Session
        from app.models import Campaign, CampaignRecipient, EmailEvent, SendAttempt, SendJob
        st, rec = send_mod
        monkeypatch.setattr(st, "build_gmail_client_from_account",
                            lambda account, token, client_config=None: _TempErrorClient())
        with Session(db_engine, expire_on_commit=False) as db:
            f = _seed_sendable(db, max_retries=3, retry_delay_seconds=60)
            cid, crid, jid = f["campaign"].id, f["cr"].id, f["job"].id
        outs = [_call(st, cid, crid) for _ in range(4)]
        assert [o["status"] for o in outs] == ["retrying", "retrying", "retrying",
                                               "dead_letter"]
        assert len(rec.calls) == 3     # 3 reschedules, none after dead-letter
        cds = [c["countdown"] for c in rec.calls]
        assert 61 <= cds[0] <= 66      # 60 * 2^0
        assert 121 <= cds[1] <= 126    # 60 * 2^1
        assert 241 <= cds[2] <= 246    # 60 * 2^2
        with Session(db_engine, expire_on_commit=False) as db:
            cr = db.get(CampaignRecipient, crid)
            camp = db.get(Campaign, cid)
            job = db.get(SendJob, jid)
            assert cr.status == "temp_failed"
            assert cr.attempt_count == 4
            assert camp.failed_count == 1
            assert job.status == "dead"
            atts = db.query(SendAttempt).filter_by(campaign_id=cid) \
                .order_by(SendAttempt.id).all()
            assert len(atts) == 4
            assert [a.retry_no for a in atts] == [0, 1, 2, 3]
            ev = db.query(EmailEvent).filter_by(campaign_recipient_id=crid,
                                                event_type="failed").one()
            assert "dead-letter after 4 attempts" in ev.detail
        # DLQ in real Redis
        import redis as redis_lib
        from app.core.config import settings
        rdb = redis_lib.from_url(settings.REDIS_URL, socket_timeout=2)
        try:
            dlq = [e.decode() for e in rdb.lrange("dlq:send", 0, -1)]
            assert any(e.startswith(f"{cid}:{crid}:") for e in dlq)
        finally:
            rdb.delete("dlq:send")

    def test_generic_connection_error_is_temporary(self, db_engine, send_mod,
                                                   monkeypatch):
        from sqlalchemy.orm import Session
        st, rec = send_mod
        monkeypatch.setattr(st, "build_gmail_client_from_account",
                            lambda account, token, client_config=None: _BoomClient())
        with Session(db_engine, expire_on_commit=False) as db:
            f = _seed_sendable(db)
            cid, crid = f["campaign"].id, f["cr"].id
        out = _call(st, cid, crid)
        assert out["ok"] is False and out["status"] == "retrying"
        assert len(rec.calls) == 1
        assert "connection" in out["reason"].lower()

    def test_provider_refusal_without_exception_retries(self, db_engine, send_mod,
                                                        monkeypatch):
        """ok=False from the provider (no exception) flows through the same
        error classifier — never a fake success."""
        from sqlalchemy.orm import Session
        st, rec = send_mod
        monkeypatch.setattr(st, "build_gmail_client_from_account",
                            lambda account, token, client_config=None: _RefusedClient())
        with Session(db_engine, expire_on_commit=False) as db:
            f = _seed_sendable(db)
            cid, crid = f["campaign"].id, f["cr"].id
        out = _call(st, cid, crid)
        assert out["ok"] is False and out["status"] == "retrying"
        assert "SMTP refused" in out["reason"]


# ---------------------------------------------------------------------------
# Permanent failures & provider-driven account/campaign state
# ---------------------------------------------------------------------------
class TestPermanentFailures:
    def test_permanent_403_perm_fails_no_retry(self, db_engine, send_mod, monkeypatch):
        from sqlalchemy.orm import Session
        from app.models import Campaign, CampaignRecipient, GmailAccount, SendAttempt
        st, rec = send_mod
        monkeypatch.setattr(st, "build_gmail_client_from_account",
                            lambda account, token, client_config=None: _PermErrorClient())
        with Session(db_engine, expire_on_commit=False) as db:
            f = _seed_sendable(db)
            cid, crid, aid = f["campaign"].id, f["cr"].id, f["account"].id
        out = _call(st, cid, crid)
        assert out["ok"] is False and out["status"] == "perm_failed"
        assert rec.calls == []          # permanent: no retry, no reschedule
        with Session(db_engine, expire_on_commit=False) as db:
            cr = db.get(CampaignRecipient, crid)
            camp = db.get(Campaign, cid)
            att = db.query(SendAttempt).filter_by(campaign_id=cid).one()
            assert cr.status == "perm_failed"
            assert camp.failed_count == 1
            assert att.error_class == "permanent"
            assert "403" in cr.last_error

    def test_auth_failure_expires_account_and_pauses_campaign(self, db_engine,
                                                              send_mod, monkeypatch):
        """401/invalid-grant style failures mark the account expired and pause
        the whole campaign (protects remaining recipients from burning
        attempts against a dead credential)."""
        import googleapiclient.errors
        import httplib2
        from sqlalchemy.orm import Session
        from app.models import Campaign, CampaignRecipient, GmailAccount

        class _AuthExpiredClient:
            method = "api"

            def send_mime_api(self, mime):
                resp = httplib2.Response({"status": 401, "reason": "Unauthorized"})
                raise googleapiclient.errors.HttpError(
                    resp, b'{"error": "invalid_grant", "message": "Token has been expired or revoked"}')

        st, _ = send_mod
        monkeypatch.setattr(st, "build_gmail_client_from_account",
                            lambda account, token, client_config=None: _AuthExpiredClient())
        with Session(db_engine, expire_on_commit=False) as db:
            f = _seed_sendable(db)
            cid, crid, aid = f["campaign"].id, f["cr"].id, f["account"].id
        out = _call(st, cid, crid)
        assert out["ok"] is False and out["status"] == "perm_failed"
        with Session(db_engine, expire_on_commit=False) as db:
            cr = db.get(CampaignRecipient, crid)
            camp = db.get(Campaign, cid)
            acc = db.get(GmailAccount, aid)
            assert cr.status == "perm_failed"
            assert acc.status == "expired"
            assert camp.status == "paused"

    def test_quota_error_retries_and_marks_account_cooldown(self, db_engine,
                                                            send_mod, monkeypatch):
        """429/quota errors are TEMPORARY: the account gets a 30-minute cooldown
        and the send is retried later (classify_error checks temporary patterns
        first — the old code kept the cooldown branch under 'permanent', which
        was unreachable dead code)."""
        import googleapiclient.errors
        import httplib2
        from sqlalchemy.orm import Session
        from app.models import Campaign, CampaignRecipient, GmailAccount, SendAttempt

        class _QuotaClient:
            method = "api"

            def send_mime_api(self, mime):
                resp = httplib2.Response({"status": 429, "reason": "Too Many Requests"})
                raise googleapiclient.errors.HttpError(
                    resp, b'{"error": {"code": 429, "message": "userRateLimitExceeded quota exceeded"}}')

        st, rec = send_mod
        monkeypatch.setattr(st, "build_gmail_client_from_account",
                            lambda account, token, client_config=None: _QuotaClient())
        with Session(db_engine, expire_on_commit=False) as db:
            f = _seed_sendable(db)
            cid, crid, aid = f["campaign"].id, f["cr"].id, f["account"].id
        out = _call(st, cid, crid)
        assert out["ok"] is False and out["status"] == "retrying"
        assert len(rec.calls) == 1
        with Session(db_engine, expire_on_commit=False) as db:
            acc = db.get(GmailAccount, aid)
            cr = db.get(CampaignRecipient, crid)
            att = db.query(SendAttempt).filter_by(campaign_id=cid).one()
            assert acc.cooldown_until is not None
            assert acc.cooldown_until > datetime.now(timezone.utc)
            assert cr.status == "queued"           # retried, not dead
            assert att.error_class == "temporary"


# ---------------------------------------------------------------------------
# Provider-driven throttling: daily cap, cooldown, per-account rate window
# ---------------------------------------------------------------------------
class TestThrottling:
    def test_daily_limit_reached_pauses_campaign(self, db_engine, send_mod, monkeypatch):
        from sqlalchemy.orm import Session
        from app.models import Campaign, CampaignRecipient, GmailAccount
        st, _ = send_mod
        with Session(db_engine, expire_on_commit=False) as db:
            f = _seed_sendable(db)
            f["account"].daily_limit = 5
            f["account"].sends_today = 5
            f["account"].last_send_date = datetime.now(timezone.utc)
            db.commit()
            cid, crid = f["campaign"].id, f["cr"].id
        out = _call(st, cid, crid)
        assert out["ok"] is False and out["status"] == "paused"
        assert "daily limit" in out["reason"]
        with Session(db_engine, expire_on_commit=False) as db:
            camp = db.get(Campaign, cid)
            cr = db.get(CampaignRecipient, crid)
            assert camp.status == "paused"
            assert cr.status == "pending"  # untouched, resumable later

    def test_account_cooldown_reschedules(self, db_engine, send_mod, monkeypatch):
        from sqlalchemy.orm import Session
        from app.models import CampaignRecipient, SendJob
        st, rec = send_mod
        ok_client = _OkClient()
        monkeypatch.setattr(st, "build_gmail_client_from_account",
                            lambda account, token, client_config=None: ok_client)
        with Session(db_engine, expire_on_commit=False) as db:
            f = _seed_sendable(db)
            f["account"].cooldown_until = datetime.now(timezone.utc) + timedelta(minutes=2)
            db.commit()
            cid, crid, jid = f["campaign"].id, f["cr"].id, f["job"].id
        out = _call(st, cid, crid)
        assert out["ok"] is False and out["status"] == "rescheduled"
        assert out["reason"] == "cooldown"
        assert ok_client.sent == []       # never even attempted
        assert len(rec.calls) == 1
        assert 60 <= rec.calls[0]["countdown"] <= 125  # ~2min cooldown window
        with Session(db_engine, expire_on_commit=False) as db:
            cr = db.get(CampaignRecipient, crid)
            job = db.get(SendJob, jid)
            assert cr.status == "queued"
            assert "cooling down" in cr.last_error
            assert job.status == "retrying"

    def test_per_account_rate_window_reschedules(self, db_engine, send_mod,
                                                 monkeypatch):
        """Real Redis fixed-interval limiter: first send passes, an immediate
        second send for the same account is deferred until the window clears."""
        from sqlalchemy.orm import Session
        from app.models import CampaignRecipient, SendJob
        st, rec = send_mod
        ok_client = _OkClient()
        monkeypatch.setattr(st, "build_gmail_client_from_account",
                            lambda account, token, client_config=None: ok_client)
        with Session(db_engine, expire_on_commit=False) as db:
            f = _seed_sendable(db, send_interval_seconds=30)
            cid, crid, jid, aid = (f["campaign"].id, f["cr"].id, f["job"].id,
                                   f["account"].id)
        out1 = _call(st, cid, crid)
        assert out1["ok"] is True and out1["status"] == "sent"
        # immediate second call: same account -> rate window not elapsed
        with Session(db_engine, expire_on_commit=False) as db:
            cr = db.get(CampaignRecipient, crid)
            cr.status = "pending"          # simulate a re-drive of same account
            db.commit()
        out2 = _call(st, cid, crid)
        assert out2["ok"] is False and out2["status"] == "rescheduled"
        assert out2["reason"] == "rate window"
        assert len(ok_client.sent) == 1    # second attempt never sent
        assert len(rec.calls) == 1
        assert 1 <= rec.calls[0]["countdown"] <= 31
        with Session(db_engine, expire_on_commit=False) as db:
            job = db.get(SendJob, jid)
            assert job.status == "retrying"
        # cleanup the rate window key (account id is unique per test anyway)
        import redis as redis_lib
        from app.core.config import settings
        rdb = redis_lib.from_url(settings.REDIS_URL, socket_timeout=2)
        rdb.delete(f"rlw:acct:{aid}")