"""End-to-end proof that the plan history window actually hides data.

The unit tests in test_trial_gating.py pin the plan arithmetic. These pin the
part that can actually leak: that an out-of-window ticket is absent from the
inbox listing AND not reachable by URL, for the same logged-in user, against a
real tenant schema — and that a paid plan sees both.

Seeds two threads into the Free Sandbox tenant (one recent, one 200 days old),
exercises the real routes through the Flask test client, then cleans up.

Skipped automatically if the sandbox org is absent (see
scripts/create_test_orgs.py) so the suite still runs on a bare checkout.
"""
import datetime as dt

import pytest
from sqlalchemy import text

from supporthub.app.db import engine, session_scope

SANDBOX_SLUG = "free-sandbox"
SANDBOX_EMAIL = "free@local.test"
SANDBOX_PASSWORD = "Test123!Free"

RECENT_SUBJECT = "WINDOWTEST recent thread"
OLD_SUBJECT = "WINDOWTEST old thread"


def _sandbox():
    with engine.connect() as c:
        return c.execute(text(
            "SELECT id, schema_name FROM public.organizations WHERE slug=:s"),
            {"s": SANDBOX_SLUG}).one_or_none()


pytestmark = pytest.mark.skipif(
    _sandbox() is None,
    reason="free-sandbox org not provisioned (run scripts/create_test_orgs.py)",
)


@pytest.fixture()
def seeded():
    """Insert one in-window and one out-of-window ticket; yield their ids."""
    org = _sandbox()
    schema = org.schema_name
    now = dt.datetime.utcnow()
    recent = now - dt.timedelta(days=2)
    old = now - dt.timedelta(days=200)
    ids = {}

    with engine.begin() as c:
        c.execute(text(f"SET search_path TO {schema}"))
        for key, subject, when in (
            ("recent", RECENT_SUBJECT, recent),
            ("old", OLD_SUBJECT, old),
        ):
            tid = c.execute(text(
                "INSERT INTO addons_threads (provider_thread_id, subject, created_at,"
                " last_activity_at) VALUES (:p, :s, :c, :l) RETURNING id"),
                {"p": f"windowtest-{key}", "s": subject, "c": when, "l": when}).scalar()
            ticket_id = c.execute(text(
                "INSERT INTO tickets (thread_id, first_seen_at, last_activity_at, status)"
                " VALUES (:t, :f, :l, 'open') RETURNING id"),
                {"t": tid, "f": when, "l": when}).scalar()
            ids[key] = {"thread": tid, "ticket": ticket_id}

    yield ids

    with engine.begin() as c:
        c.execute(text(f"SET search_path TO {schema}"))
        c.execute(text("DELETE FROM tickets WHERE id = ANY(:ids)"),
                  {"ids": [v["ticket"] for v in ids.values()]})
        c.execute(text("DELETE FROM addons_threads WHERE id = ANY(:ids)"),
                  {"ids": [v["thread"] for v in ids.values()]})


@pytest.fixture()
def client():
    import re
    from supporthub.app.main import app
    c = app.test_client()
    r = c.get("/login")
    token = re.search(r'name="csrf_token"[^>]*value="([^"]+)"',
                      r.get_data(as_text=True)).group(1)
    r = c.post("/login", data={"email": SANDBOX_EMAIL, "password": SANDBOX_PASSWORD,
                              "csrf_token": token})
    assert r.status_code in (200, 302), "sandbox login failed"
    return c


@pytest.fixture()
def as_plan():
    """Temporarily set the sandbox org's stored plan, restoring it afterwards."""
    org = _sandbox()
    originals = {}

    def _set(plan, trial_days=None):
        with session_scope(schema="public") as s:
            row = s.execute(text(
                "SELECT plan, trial_ends_at FROM public.organizations WHERE id=:i"),
                {"i": org.id}).one()
            originals.setdefault("plan", row.plan)
            originals.setdefault("trial_ends_at", row.trial_ends_at)
            s.execute(text(
                "UPDATE public.organizations SET plan=:p, trial_ends_at=:t WHERE id=:i"),
                {"p": plan, "i": org.id,
                 "t": (dt.datetime.utcnow() + dt.timedelta(days=trial_days))
                      if trial_days else None})

    yield _set

    if originals:
        with session_scope(schema="public") as s:
            s.execute(text(
                "UPDATE public.organizations SET plan=:p, trial_ends_at=:t WHERE id=:i"),
                {"p": originals["plan"], "t": originals["trial_ends_at"], "i": org.id})


def test_free_plan_inbox_hides_the_out_of_window_ticket(client, seeded, as_plan):
    as_plan("free")
    body = client.get("/inbox").get_data(as_text=True)
    assert RECENT_SUBJECT in body, "in-window ticket should be listed"
    assert OLD_SUBJECT not in body, "out-of-window ticket LEAKED into the inbox"


def test_free_plan_cannot_open_the_out_of_window_ticket_by_url(client, seeded, as_plan):
    """The window must not be cosmetic — the direct URL has to be refused too."""
    as_plan("free")
    assert client.get(f"/ticket/{seeded['recent']['ticket']}").status_code == 200
    assert client.get(f"/ticket/{seeded['old']['ticket']}").status_code == 403


def test_free_plan_cannot_export_the_out_of_window_ticket(client, seeded, as_plan):
    as_plan("free")
    r = client.get(f"/ticket/{seeded['old']['ticket']}/export.md")
    assert r.status_code == 403, "export must honour the window, not bypass it"


def test_trial_gets_the_same_30_day_window_as_free(client, seeded, as_plan):
    as_plan("free", trial_days=15)
    body = client.get("/inbox").get_data(as_text=True)
    assert RECENT_SUBJECT in body
    assert OLD_SUBJECT not in body
    assert client.get(f"/ticket/{seeded['old']['ticket']}").status_code == 403


def test_paid_plan_sees_everything(client, seeded, as_plan):
    """The same user, same data, on MAX — proves the filter is plan-driven."""
    as_plan("max")
    body = client.get("/inbox").get_data(as_text=True)
    assert RECENT_SUBJECT in body
    assert OLD_SUBJECT in body, "MAX must see full history"
    assert client.get(f"/ticket/{seeded['old']['ticket']}").status_code == 200


def test_upgrading_restores_access_to_previously_hidden_data(client, seeded, as_plan):
    """Downgrade hides; upgrade brings it back. Nothing is deleted in between."""
    as_plan("free")
    assert client.get(f"/ticket/{seeded['old']['ticket']}").status_code == 403
    as_plan("max")
    assert client.get(f"/ticket/{seeded['old']['ticket']}").status_code == 200


def test_badge_counts_agree_with_the_windowed_list(client, seeded, as_plan):
    """The symptom this prevents: "Open 40" sitting above a list of 5.

    The counts came from separate COUNT() queries that ignored the window, so the
    badge contradicted the page next to it. Asserted via the AJAX endpoint (the
    same helper the context processor uses) so both copies are covered.
    """
    as_plan("max")
    full = client.get("/api/sidebar-counts").get_json()
    as_plan("free")
    windowed = client.get("/api/sidebar-counts").get_json()

    assert full["all"]["total"] > windowed["all"]["total"], (
        "free must count fewer tickets than max over the same data "
        f"(max={full['all']['total']} free={windowed['all']['total']})"
    )
    # And the count must match what the list actually shows.
    body = client.get("/inbox").get_data(as_text=True)
    assert OLD_SUBJECT not in body
    assert RECENT_SUBJECT in body


def test_badge_counts_are_not_served_stale_across_a_plan_change(client, seeded, as_plan):
    """The counts are cached for 30s; the cutoff must be part of the cache key or
    an upgrade shows the old restricted numbers for half a minute."""
    as_plan("free")
    before = client.get("/api/sidebar-counts").get_json()["all"]["total"]
    as_plan("max")
    after = client.get("/api/sidebar-counts").get_json()["all"]["total"]
    assert after > before, "plan change did not bust the sidebar count cache"


def test_full_account_export_ignores_the_window(client, seeded, as_plan):
    """Portability: a downgraded user must still be able to RETRIEVE the history
    the window hides, or "hidden, not deleted" is an unverifiable claim.

    Deliberately the one read path that is NOT window-restricted.
    """
    as_plan("free")
    # The window hides it from the app...
    assert OLD_SUBJECT not in client.get("/inbox").get_data(as_text=True)
    assert client.get(f"/ticket/{seeded['old']['ticket']}").status_code == 403
    # ...but the account export still contains it.
    r = client.get("/settings/export-all.md")
    assert r.status_code == 200, r.status_code
    body = r.get_data(as_text=True)
    assert OLD_SUBJECT in body, "portability export must include hidden history"
    assert RECENT_SUBJECT in body


def test_full_account_export_is_audit_logged(client, seeded, as_plan):
    """A bulk export of every customer conversation must never be invisible."""
    org = _sandbox()
    with engine.connect() as c:
        c.execute(text(f"SET search_path TO {org.schema_name}"))
        before = c.execute(text(
            "SELECT count(*) FROM audit_log WHERE action='export_account'")).scalar()
    as_plan("free")
    assert client.get("/settings/export-all.md").status_code == 200
    with engine.connect() as c:
        c.execute(text(f"SET search_path TO {org.schema_name}"))
        after = c.execute(text(
            "SELECT count(*) FROM audit_log WHERE action='export_account'")).scalar()
    assert after == before + 1, "account export was not recorded in the audit log"