"""Live dashboard tests: /progress polling endpoint, /recipients monitor table,
and the /stream SSE endpoint across ALL terminal states (completed/cancelled/
failed) plus payload-shape and isolation checks.

test_api_accounts_campaigns_testsend_unsubscribe.py already covers the SSE
happy path (connected -> progress -> done on completed) and basic progress/
recipients smoke tests; this module focuses on the remaining dashboard
surface: multi-status progress counters, monitor filtering/search edge cases,
and the cancelled/failed SSE termination paths (same generator, different
"done" trigger).

NOTE on SSE testing: starlette's TestClient (0.41.3) drives the ENTIRE ASGI
app to completion inside `portal.call` before returning the response, so an
endless stream would deadlock at `with api.stream(...)`. Every SSE test here
uses a background thread that flips the campaign to a terminal status shortly
after the stream opens, so the server-side generator sees the DB change (via
its 5s poll fallback or the Redis pub/sub tick) and emits `done` + returns on
its own — never via a client-side early break.
"""
import csv
import io
import json
import secrets
import threading
import time as _time
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
    and the SSE generator's pub/sub subscription has a live server)."""
    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_campaign_fixture(db, user_id, n_recipients=3):
    """RecipientList + recipients + subject + template + authorized account."""
    from app.models import (EmailTemplate, GmailAccount, OAuthToken, Recipient,
                            RecipientList, SubjectLine)

    rl = RecipientList(user_id=user_id, name=f"dash-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"dash{i}@x.com",
                         first_name=f"Dash{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"dash-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": "Dash 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


def _flip_status_after(db_engine, cid, status, seconds, extra=None):
    """Background-thread helper: after `seconds`, set Campaign.status (+ any
    extra column values) and commit — used to drive the SSE generator to a
    terminal state without a real worker."""
    from app.models import Campaign
    from sqlalchemy.orm import Session as SyncSess

    def _run():
        _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 = status
                for k, v in (extra or {}).items():
                    setattr(c, k, v)
                s.commit()

    th = threading.Thread(target=_run, daemon=True)
    th.start()
    return th


def _read_sse(api, cid, timeout=30.0):
    """Consume the SSE stream to completion (generator self-terminates on
    `done`) and return the raw frame list."""
    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 < timeout, frames
    return frames


def _done_payload(frames):
    done_idx = next(i for i, fr in enumerate(frames) if "event: done" in fr)
    data_line = frames[done_idx + 1]
    assert data_line.startswith("data: "), data_line
    return json.loads(data_line[len("data: "):])


# --------------------------------------------------------------------------- /progress
class TestProgressEndpoint:
    def test_progress_shape_draft_campaign(self, api, db):
        """Draft (never started) campaign: progress is a well-formed zero state,
        not an error — the endpoint must work regardless of lifecycle stage."""
        f = _seed_campaign_fixture(db, user_id_of(api), n_recipients=4)
        cid = api.post("/campaigns/create", headers=H(api),
                       json=_cIn(f, name="Draft")).json()["id"]
        d = api.get(f"/campaigns/{cid}/progress").json()
        assert d["campaign_id"] == cid
        assert d["status"] == "draft"
        assert d["jobs_total"] == 0
        assert d["progress_pct"] == 0.0
        assert d["completed"] is False
        assert d["started_at"] is None and d["finished_at"] is None

    def test_progress_requires_ownership(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="Owned")).json()["id"]
        r = second_user.get(f"/campaigns/{cid}/progress")
        assert r.status_code == 404

    def test_progress_nonexistent_campaign_404(self, api):
        r = api.get("/campaigns/999999/progress")
        assert r.status_code == 404

    @requires_broker
    def test_progress_reflects_manual_status_transitions(self, api, db, db_engine):
        """Progress counters reflect whatever CampaignRecipient rows say, live —
        simulate a worker completing 2/3 sends and verify counts/percent."""
        from app.models import Campaign, CampaignRecipient
        from sqlalchemy.orm import Session as SyncSess

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

        with SyncSess(db_engine, expire_on_commit=False) as s:
            rows = (s.query(CampaignRecipient)
                    .filter_by(campaign_id=cid).order_by(CampaignRecipient.seq).all())
            rows[0].status = "sent"
            rows[1].status = "perm_failed"
            # rows[2] stays queued
            camp = s.get(Campaign, cid)
            camp.sent_count = 1
            camp.failed_count = 1
            s.commit()

        d = api.get(f"/campaigns/{cid}/progress").json()
        assert d["total"] == 3
        assert d["sent"] == 1
        assert d["perm_failed"] == 1
        assert d["queued"] == 1
        # done = sent + perm_failed + suppressed + skipped = 2 of 3
        assert d["progress_pct"] == round(100.0 * 2 / 3, 1)
        assert d["completed"] is False  # status still "running", not "completed"


# --------------------------------------------------------------------------- /recipients
class TestRecipientsMonitor:
    @requires_broker
    def test_status_filter_matches_only_that_status(self, api, db, db_engine):
        from app.models import Campaign, CampaignRecipient
        from sqlalchemy.orm import Session as SyncSess

        f = _seed_campaign_fixture(db, user_id_of(api), n_recipients=3)
        cid = api.post("/campaigns/create", headers=H(api),
                       json=_cIn(f, name="Filt")).json()["id"]
        api.post(f"/campaigns/{cid}/start", headers=H(api))
        with SyncSess(db_engine, expire_on_commit=False) as s:
            rows = (s.query(CampaignRecipient)
                    .filter_by(campaign_id=cid).order_by(CampaignRecipient.seq).all())
            rows[0].status = "sent"
            s.commit()

        sent_only = api.get(f"/campaigns/{cid}/recipients",
                            params={"status": "sent"}).json()
        assert sent_only["total"] == 1
        assert sent_only["recipients"][0]["status"] == "sent"

        queued_only = api.get(f"/campaigns/{cid}/recipients",
                              params={"status": "queued"}).json()
        assert queued_only["total"] == 2

    @requires_broker
    def test_search_matches_email_substring_case_insensitive(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="Search")).json()["id"]
        api.post(f"/campaigns/{cid}/start", headers=H(api))
        d = api.get(f"/campaigns/{cid}/recipients", params={"search": "DASH2"}).json()
        assert d["total"] == 1
        assert "dash2@x.com" in d["recipients"][0]["email"]

    def test_recipients_requires_ownership(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="Priv")).json()["id"]
        r = second_user.get(f"/campaigns/{cid}/recipients")
        assert r.status_code == 404

    def test_recipients_empty_before_start(self, api, db):
        """Draft campaign has no materialized CampaignRecipient rows yet."""
        f = _seed_campaign_fixture(db, user_id_of(api), n_recipients=3)
        cid = api.post("/campaigns/create", headers=H(api),
                       json=_cIn(f, name="Empty")).json()["id"]
        d = api.get(f"/campaigns/{cid}/recipients").json()
        assert d["total"] == 0
        assert d["recipients"] == []


# --------------------------------------------------------------------------- SSE stream
@requires_broker
class TestSSEStreamTerminalStates:
    def test_stream_terminates_on_cancelled(self, api, db, db_engine):
        f = _seed_campaign_fixture(db, user_id_of(api), n_recipients=2)
        cid = api.post("/campaigns/create", headers=H(api),
                       json=_cIn(f, name="Cancelled")).json()["id"]
        r = api.post(f"/campaigns/{cid}/start", headers=H(api))
        assert r.status_code == 200, r.text

        th = _flip_status_after(db_engine, cid, "cancelled", 3.0)
        frames = _read_sse(api, cid)
        th.join(timeout=5)

        assert any("event: connected" in fr for fr in frames), frames
        assert any("event: done" in fr for fr in frames), frames
        payload = _done_payload(frames)
        assert payload["status"] == "cancelled"

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

        th = _flip_status_after(db_engine, cid, "failed", 3.0)
        frames = _read_sse(api, cid)
        th.join(timeout=5)

        assert any("event: done" in fr for fr in frames), frames
        payload = _done_payload(frames)
        assert payload["status"] == "failed"

    def test_stream_does_not_emit_done_while_running(self, api, db, db_engine):
        """Before the terminal flip, only connected/progress/keepalive frames
        appear — never a premature `done` for a still-running campaign."""
        f = _seed_campaign_fixture(db, user_id_of(api), n_recipients=2)
        cid = api.post("/campaigns/create", headers=H(api),
                       json=_cIn(f, name="StillRunning")).json()["id"]
        r = api.post(f"/campaigns/{cid}/start", headers=H(api))
        assert r.status_code == 200, r.text

        # flip late enough that we can sample a few seconds of "still running"
        th = _flip_status_after(db_engine, cid, "completed", 4.0)
        frames = _read_sse(api, cid)
        th.join(timeout=5)

        done_positions = [i for i, fr in enumerate(frames) if "event: done" in fr]
        assert len(done_positions) == 1  # exactly one terminal event, at the end
        assert done_positions[0] == max(
            i for i, fr in enumerate(frames) if fr.startswith("event:")
        )

    def test_stream_progress_payload_matches_progress_endpoint_shape(self, api, db,
                                                                     db_engine):
        """The `progress` SSE event payload must carry the same fields/semantics
        as GET /progress (both built from the same _progress_payload helper)."""
        f = _seed_campaign_fixture(db, user_id_of(api), n_recipients=2)
        cid = api.post("/campaigns/create", headers=H(api),
                       json=_cIn(f, name="Shape")).json()["id"]
        r = api.post(f"/campaigns/{cid}/start", headers=H(api))
        assert r.status_code == 200, r.text

        th = _flip_status_after(db_engine, cid, "completed", 3.0)
        frames = _read_sse(api, cid)
        th.join(timeout=5)

        progress_idx = next(i for i, fr in enumerate(frames) if "event: progress" in fr)
        progress_payload = json.loads(frames[progress_idx + 1][len("data: "):])
        expected_keys = {"campaign_id", "status", "total", "sent", "temp_failed",
                         "perm_failed", "suppressed", "skipped", "pending", "queued",
                         "processing", "jobs_total", "completed", "progress_pct",
                         "started_at", "finished_at"}
        assert expected_keys <= set(progress_payload)
        assert progress_payload["campaign_id"] == cid

    def test_stream_requires_ownership(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="NotYours")).json()["id"]
        r = second_user.get(f"/campaigns/{cid}/stream")
        assert r.status_code == 404