"""The exposure ledger, and who is allowed to touch the console.

The ledger is the compliance artefact. Everything here is about it being
*honest*: deduped so "shown 400 times" means something, published-only so
an author's own preview clicks do not inflate coverage, and stamped with
the version and language actually rendered.
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime, timedelta

import pytest
from sqlalchemy import select

from app.domains.security_awareness.models import SecawImpression
from tests.test_security_awareness.conftest import (
    ADMIN_BASE,
    SELF_BASE,
    article_payload,
    comic_payload,
    headers_for,
    make_person,
)

pytestmark = pytest.mark.asyncio


async def _publish(client, headers, payload=None) -> str:
    r = await client.post(
        f"{ADMIN_BASE}/materials", json=payload or article_payload(),
        headers=headers,
    )
    assert r.status_code == 201, r.text
    return r.json()["data"]["id"]


async def _impressions(db, material_id: str) -> list[SecawImpression]:
    return list((await db.execute(
        select(SecawImpression).where(
            SecawImpression.material_id == uuid.UUID(material_id),
        )
    )).scalars().all())


async def test_opening_a_material_records_exactly_one_exposure(
    client, db_session, security_officer, reader,
):
    """The write lives on the read route rather than in a client-issued
    follow-up: a compliance claim that depends on the browser remembering
    to report is a claim that quietly under-counts."""
    officer, oh = security_officer
    person, rh = reader
    mid = await _publish(client, oh)

    await client.get(f"{SELF_BASE}/materials/{mid}?lang=en", headers=rh)

    rows = await _impressions(db_session, mid)
    assert len(rows) == 1
    assert rows[0].person_id == person.id
    assert rows[0].surface == "detail"
    assert rows[0].material_version == 1
    # zh-only material read by an English reader → the ledger records the
    # block that ACTUALLY rendered, not what was asked for.
    assert rows[0].lang == "zh"


async def test_refreshing_does_not_manufacture_exposures(
    client, db_session, security_officer, reader,
):
    """This idempotence is what makes a state-changing GET acceptable."""
    officer, oh = security_officer
    _p, rh = reader
    mid = await _publish(client, oh)

    for _ in range(5):
        await client.get(f"{SELF_BASE}/materials/{mid}", headers=rh)

    assert len(await _impressions(db_session, mid)) == 1


async def test_a_new_exposure_is_recorded_once_the_window_has_passed(
    client, db_session, security_officer, reader,
):
    """Coming back tomorrow IS a second exposure — the window collapses
    a reading session, not a habit."""
    officer, oh = security_officer
    _p, rh = reader
    mid = await _publish(client, oh)

    await client.get(f"{SELF_BASE}/materials/{mid}", headers=rh)
    rows = await _impressions(db_session, mid)
    assert len(rows) == 1

    # Age the only row past the dedupe window.
    rows[0].occurred_at = datetime.now(UTC) - timedelta(hours=2)
    await db_session.commit()

    await client.get(f"{SELF_BASE}/materials/{mid}", headers=rh)
    assert len(await _impressions(db_session, mid)) == 2


async def test_surfaces_dedupe_independently(
    client, db_session, security_officer, reader,
):
    """Being shown the card in the ambient bubble and opening the article
    are different claims; an auditor is entitled to tell them apart."""
    officer, oh = security_officer
    _p, rh = reader
    mid = await _publish(client, oh)

    await client.get(f"{SELF_BASE}/materials/{mid}", headers=rh)
    r = await client.post(
        f"{SELF_BASE}/impressions",
        json={"material_id": mid, "surface": "bubble"},
        headers=rh,
    )
    assert r.status_code == 202
    assert r.json()["data"]["recorded"] is True

    rows = await _impressions(db_session, mid)
    assert sorted(r.surface for r in rows) == ["bubble", "detail"]

    # A second bubble report inside the window collapses.
    r = await client.post(
        f"{SELF_BASE}/impressions",
        json={"material_id": mid, "surface": "bubble"},
        headers=rh,
    )
    assert r.json()["data"]["recorded"] is False
    assert len(await _impressions(db_session, mid)) == 2


async def test_only_the_detail_surface_marks_a_material_as_read(
    client, security_officer, reader,
):
    """Being shown a card is not having read the thing."""
    officer, oh = security_officer
    _p, rh = reader
    mid = await _publish(client, oh)

    await client.post(
        f"{SELF_BASE}/impressions",
        json={"material_id": mid, "surface": "bubble"}, headers=rh,
    )
    card = (await client.get(
        f"{SELF_BASE}/materials", headers=rh,
    )).json()["data"][0]
    assert card["viewed_at"] is None


async def test_an_unknown_surface_is_refused(client, security_officer, reader):
    officer, oh = security_officer
    _p, rh = reader
    mid = await _publish(client, oh)
    r = await client.post(
        f"{SELF_BASE}/impressions",
        json={"material_id": mid, "surface": "billboard"}, headers=rh,
    )
    assert r.status_code == 422


async def test_a_draft_preview_records_nothing(
    client, db_session, security_officer,
):
    """An author looking at their own draft is not workforce reach, and
    counting it would inflate coverage with the author's own clicks."""
    officer, oh = security_officer
    mid = await _publish(client, oh, article_payload(publish=False))

    r = await client.get(f"{ADMIN_BASE}/materials/{mid}/preview", headers=oh)
    assert r.status_code == 200
    assert r.json()["data"]["title"] == "口令怎么设"
    assert await _impressions(db_session, mid) == []

    # And the explicit report endpoint refuses too, rather than trusting
    # the caller's claim about an unpublished row.
    r = await client.post(
        f"{SELF_BASE}/impressions",
        json={"material_id": mid, "surface": "bubble"}, headers=oh,
    )
    assert r.json()["data"]["recorded"] is False
    assert await _impressions(db_session, mid) == []


async def test_admin_console_separates_reach_from_impression_count(
    client, db_session, security_officer, reader,
):
    """"shown 900 times" and "reached 300 people" answer different
    questions; a console printing only the first invites exactly the
    wrong conclusion about coverage."""
    officer, oh = security_officer
    _p, rh = reader
    mid = await _publish(client, oh)

    await client.get(f"{SELF_BASE}/materials/{mid}", headers=rh)
    await client.post(
        f"{SELF_BASE}/impressions",
        json={"material_id": mid, "surface": "bubble"}, headers=rh,
    )

    row = (await client.get(
        f"{ADMIN_BASE}/materials", headers=oh,
    )).json()["data"][0]
    assert row["reached_people"] == 1
    assert row["impressions"] == 2


async def test_author_name_and_email_travel_separately(
    client, security_officer,
):
    """The platform spent months un-splitting a "姓名 · 邮箱" string that a
    dozen frontends each truncated their own way (#3417). A new surface
    does not get to re-create that."""
    officer, oh = security_officer
    await _publish(client, oh)

    row = (await client.get(
        f"{ADMIN_BASE}/materials", headers=oh,
    )).json()["data"][0]
    assert row["created_by_name"] == "officer"
    assert row["created_by_email"] == "officer@test.com"
    assert "·" not in (row["created_by_name"] or "")


# ── RBAC ────────────────────────────────────────────────────────────


async def test_an_employee_cannot_reach_the_console(client, security_officer, reader):
    officer, oh = security_officer
    _p, rh = reader
    mid = await _publish(client, oh)

    for method, path, body in (
        ("get", f"{ADMIN_BASE}/materials", None),
        ("get", f"{ADMIN_BASE}/meta", None),
        ("post", f"{ADMIN_BASE}/materials", article_payload()),
        ("put", f"{ADMIN_BASE}/materials/{mid}", {"is_pinned": True}),
        ("post", f"{ADMIN_BASE}/materials/{mid}/status", {"status": "archived"}),
        ("post", f"{ADMIN_BASE}/materials/{mid}/duplicate", None),
        ("delete", f"{ADMIN_BASE}/materials/{mid}", None),
        ("get", f"{ADMIN_BASE}/materials/{mid}/preview", None),
    ):
        call = getattr(client, method)
        r = await call(path, headers=rh) if body is None else await call(
            path, json=body, headers=rh,
        )
        assert r.status_code == 403, f"{method.upper()} {path} → {r.status_code}"


async def test_security_officer_can_run_the_whole_library(
    client, security_officer,
):
    """安全员 IS the awareness desk — it must not need a global admin to
    publish a comic."""
    officer, oh = security_officer
    mid = await _publish(client, oh, comic_payload(officer.id))
    assert (await client.put(
        f"{ADMIN_BASE}/materials/{mid}", json={"is_pinned": True}, headers=oh,
    )).status_code == 200
    assert (await client.post(
        f"{ADMIN_BASE}/materials/{mid}/status", json={"status": "archived"},
        headers=oh,
    )).status_code == 200
    assert (await client.get(
        f"{ADMIN_BASE}/meta", headers=oh,
    )).json()["data"]["can_manage"] is True


async def test_management_reads_but_cannot_write(
    client, db_session, security_officer,
):
    """The leadership view is read-only platform-wide, and the console
    must not render write affordances that would 403."""
    officer, oh = security_officer
    mid = await _publish(client, oh)

    boss = await make_person(db_session, "boss@test.com", role="management")
    await db_session.commit()
    bh = headers_for(boss, ["management"])

    meta = await client.get(f"{ADMIN_BASE}/meta", headers=bh)
    assert meta.status_code == 200
    assert meta.json()["data"]["can_manage"] is False

    assert (await client.get(
        f"{ADMIN_BASE}/materials", headers=bh,
    )).status_code == 200
    assert (await client.post(
        f"{ADMIN_BASE}/materials/{mid}/status", json={"status": "archived"},
        headers=bh,
    )).status_code == 403


async def test_upload_purpose_is_gated_to_the_security_desk(
    client, security_officer, reader,
):
    """Every other purpose is open on the write side because an employee
    legitimately produces the evidence. Awareness material is authored by
    one team; leaving it open would let anyone spend the bucket quota on
    content only the security desk is supposed to write."""
    _officer, oh = security_officer
    _p, rh = reader
    body = {
        "purpose": "awareness_material",
        "filename": "p1.png",
        "content_type": "image/png",
        "size_bytes": 1024,
        "checksum_sha256": "A" * 43 + "=",
    }
    assert (await client.post(
        "/api/v1/uploads/presign-put", json=body, headers=rh,
    )).status_code == 403

    # The officer gets past the gate. Whether S3 is configured in this
    # environment is a different question — the assertion is only that
    # the RBAC gate is not what stops them.
    r = await client.post("/api/v1/uploads/presign-put", json=body, headers=oh)
    assert r.status_code != 403