"""Cost-measure confidentiality on the audit-log read paths (2026-07-29).

``can_access_audit`` admits ``auditor`` + ``management``, both OUTSIDE
the invoice circle (``can_view_invoice_measure`` = finance /
cloud_admin / global_admin). ``cloud.scope_alert.fired`` audit rows
carry the evaluated cloud-bill amounts in ``detail``
(``current_usd`` / ``threshold_usd`` / ``baseline_usd`` /
``fx_rate_cny_per_usd``); on a net-evaluated alert those are
invoice-basis figures, and dividing ``current_usd`` by the list-basis
``/cmdb/billing/daily`` number for the same (scope, day) — which the
same reader may legitimately fetch — yields the account's discount rate.

The fix is read-time redaction, not a narrower gate: the DB row stays
complete and a circle reader still sees everything. These tests pin
every read path (JSON list, single-row, trace, per-resource history,
XLSX export), both basis directions, and the fail-safe branch for a
money-bearing action nobody classified.
"""
from __future__ import annotations

import io
import uuid

import pytest
import pytest_asyncio
from httpx import AsyncClient
from openpyxl import load_workbook
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession

from app.audit.models import AuditLog
from app.core.masking import (
    AUDIT_ACTION_COST_BASIS,
    AUDIT_COST_BASIS_KEY,
    REDACTED_AMOUNT,
    REDACTED_REASON,
    is_money_detail_key,
    redact_invoice_basis_amounts,
)
from app.identity.person_models import IdentityPerson

SCOPE_ALERT_ACTION = "cloud.scope_alert.fired"
SCOPE_RESOURCE_TYPE = "cmdb_cloud_scope"

# The HISTORICAL (pre-#2097) writer shape — full amounts in the row.
# Every row written before 2026-07-29 looks like this, and the read-side
# redactor exists precisely because those rows are immutable. The
# CURRENT writer withholds the amounts on net (see
# ``test_current_writer_net_shape_keeps_the_withhold_markers``).
# Numbers are the 2026-07-27 gcp-autonomous-prod spike.
_ALERT_DETAIL = {
    "rule_type": "daily_spike",
    "period_key": "2026-07-28",
    "threshold_usd": "1059.00",
    "current_usd": "1163.42",
    "baseline_usd": "706.00",
    "multiplier": 1.5,
    "fx_rate_cny_per_usd": 7.21,
    "dispatched": True,
    "recipients_count": 3,
    "integration_id": "b9f0c8c2-0000-0000-0000-000000000001",
}
_ABSOLUTE_KEYS = (
    "threshold_usd", "current_usd", "baseline_usd", "fx_rate_cny_per_usd",
)


def _net_detail() -> dict:
    """A net-evaluated alert row as the writer will stamp it."""
    return {**_ALERT_DETAIL, AUDIT_COST_BASIS_KEY: "invoice"}


def _list_detail() -> dict:
    return {**_ALERT_DETAIL, AUDIT_COST_BASIS_KEY: "list"}


async def _seed_alert(
    db: AsyncSession, *, detail: dict, action: str = SCOPE_ALERT_ACTION,
) -> tuple[str, str]:
    """Insert one system-actor alert audit row. Returns (resource_id,
    trace_id) — both unique so every assertion filters to this row only
    (the test DB is shared and append-only)."""
    resource_id = f"scope-{uuid.uuid4()}"
    trace_id = uuid.uuid4()
    db.add(AuditLog(
        id=uuid.uuid4(),
        trace_id=trace_id,
        actor_type="system",
        actor_user_id=None,
        action=action,
        resource_type=SCOPE_RESOURCE_TYPE,
        resource_id=resource_id,
        status="success",
        detail=detail,
    ))
    await db.commit()
    return resource_id, str(trace_id)


@pytest_asyncio.fixture
async def management_headers(db_session: AsyncSession):
    """Management role — inside ``can_access_audit``, outside the
    invoice circle."""
    from app.auth.models import AuthUserRole
    from app.core.security import create_access_token

    user_id = uuid.uuid4()
    db_session.add(IdentityPerson(
        id=user_id,
        work_email=f"mgmt-{user_id.hex[:8]}@test.com",
        display_name="Management Reader",
        employment_status="active",
        source_system="successfactors",
        hris_person_id=f"sf-mgmt-{user_id.hex[:8]}",
    ))
    await db_session.flush()
    role_id = (await db_session.execute(
        text("SELECT id FROM auth_roles WHERE name = 'management'"),
    )).scalar_one()
    db_session.add(AuthUserRole(
        user_id=user_id,
        role_id=uuid.UUID(role_id) if isinstance(role_id, str) else role_id,
    ))
    await db_session.commit()
    token = create_access_token(str(user_id), ["management"])
    return {"Authorization": f"Bearer {token}"}


# ── Unit: the redactor itself ────────────────────────────────────────


def test_net_row_drops_absolutes_but_keeps_the_ratio():
    out = redact_invoice_basis_amounts(
        _net_detail(), action=SCOPE_ALERT_ACTION, can_view_invoice=False,
    )
    for key in _ABSOLUTE_KEYS:
        assert out[key] == REDACTED_AMOUNT, key
    # A dimensionless same-basis ratio cancels the discount — it stays,
    # and so does every non-money forensic field.
    assert out["multiplier"] == 1.5
    assert out["rule_type"] == "daily_spike"
    assert out["period_key"] == "2026-07-28"
    assert out["recipients_count"] == 3
    assert out["dispatched"] is True
    assert out["integration_id"] == _ALERT_DETAIL["integration_id"]
    assert out["_redacted_reason"] == REDACTED_REASON
    assert sorted(out["_redacted_fields"]) == sorted(_ABSOLUTE_KEYS)
    # Never mutate the caller's dict — the ORM row must stay complete.
    assert _net_detail()["current_usd"] == "1163.42"


def test_current_writer_net_shape_keeps_the_withhold_markers():
    """The CURRENT writer's net row (post-#2097): no absolute amounts,
    ``amounts_withheld`` policy markers instead. The read-side redactor
    must NOT eat those markers — their entire purpose is to tell the
    OUT-OF-CIRCLE reader "withheld by policy, not a cron failure" — and
    the only money-shaped key left (the FX rate) still redacts."""
    detail = {
        AUDIT_COST_BASIS_KEY: "invoice",
        "rule_type": "daily_spike",
        "period_key": "2026-07-28",
        "measure": "net",
        "multiplier": 1.5,
        "actual_multiple": 1.6479,
        "fx_rate_cny_per_usd": 7.21,
        "dispatched": True,
        "recipients_count": 3,
        "integration_id": "b9f0c8c2-0000-0000-0000-000000000001",
        "amounts_withheld": True,
        "amounts_withheld_reason": "invoice_basis_confidential",
    }
    out = redact_invoice_basis_amounts(
        dict(detail), action=SCOPE_ALERT_ACTION, can_view_invoice=False,
    )
    assert out["amounts_withheld"] is True
    assert out["amounts_withheld_reason"] == "invoice_basis_confidential"
    assert out["actual_multiple"] == 1.6479
    assert out["measure"] == "net"
    assert out["fx_rate_cny_per_usd"] == REDACTED_AMOUNT
    assert out["_redacted_fields"] == ["fx_rate_cny_per_usd"]


def test_circle_reader_gets_the_untouched_net_row():
    detail = _net_detail()
    assert redact_invoice_basis_amounts(
        detail, action=SCOPE_ALERT_ACTION, can_view_invoice=True,
    ) == detail


def test_list_basis_row_is_never_redacted():
    detail = _list_detail()
    assert redact_invoice_basis_amounts(
        detail, action=SCOPE_ALERT_ACTION, can_view_invoice=False,
    ) == detail


def test_row_without_a_basis_marker_redacts():
    """Every ``cloud.scope_alert.fired`` row written before this fix
    lacks the marker. "Cannot prove list" ⇒ treat as invoice."""
    out = redact_invoice_basis_amounts(
        dict(_ALERT_DETAIL), action=SCOPE_ALERT_ACTION, can_view_invoice=False,
    )
    assert out["current_usd"] == REDACTED_AMOUNT
    assert out["multiplier"] == 1.5


def test_unknown_money_bearing_action_fails_safe():
    action = "cloud.some_future_cost_alert.fired"
    assert action not in AUDIT_ACTION_COST_BASIS
    out = redact_invoice_basis_amounts(
        {"net_spend_usd": "900.00", "ratio": 1.2, "scope": "abc"},
        action=action, can_view_invoice=False,
    )
    assert out["net_spend_usd"] == REDACTED_AMOUNT
    assert out["ratio"] == 1.2
    assert out["scope"] == "abc"


def test_unknown_action_without_money_is_untouched():
    detail = {"reason": "manual", "count": 4}
    assert redact_invoice_basis_amounts(
        detail, action="something.brand_new", can_view_invoice=False,
    ) == detail


def test_non_cloud_money_actions_pass_through():
    """Asset / finance / AI-key money has no list-vs-invoice duality, so
    redacting it would blind the auditor for nothing."""
    detail = {"amount": "1200.00", "currency": "CNY"}
    for action in (
        "software_addon.request", "finance.budget.create",
        "apikey.budget_alert_sent",
    ):
        assert redact_invoice_basis_amounts(
            detail, action=action, can_view_invoice=False,
        ) == detail, action


def test_every_money_bearing_audit_writer_is_classified():
    """The registration rule (DECISIONS 2026-07-30 ③) enforced by a
    source scan instead of a promise.

    Fail-closed is the right DEFAULT, but for a NON-cloud writer it is
    also a silent regression: an auditor investigating a procurement
    dispute gets ``"[redacted]"`` where the amount used to be, and
    nothing tells them the redaction was a classification oversight.
    So every ``audit.log(action=..., detail={...})`` call site whose
    literal detail keys include a money SHAPE must be in the table.

    Scans the literal ``detail=`` dicts with ``ast`` (no imports, no
    DB): a new writer that spells its keys inline gets caught here. A
    writer that builds ``detail`` in a variable is invisible to this
    scan — hence the docs rule as well; this test raises the floor, it
    does not replace reading the rule.
    """
    import ast
    from pathlib import Path

    app_root = Path(__file__).resolve().parents[2] / "app"
    unclassified: list[str] = []
    for py in app_root.rglob("*.py"):
        try:
            tree = ast.parse(py.read_text(encoding="utf-8"))
        except SyntaxError:  # pragma: no cover — app must always parse
            continue
        for node in ast.walk(tree):
            if not isinstance(node, ast.Call):
                continue
            kw = {k.arg: k.value for k in node.keywords if k.arg}
            action_node, detail_node = kw.get("action"), kw.get("detail")
            if not isinstance(action_node, ast.Constant):
                continue
            if not isinstance(action_node.value, str):
                continue
            if not isinstance(detail_node, ast.Dict):
                continue
            money = [
                k.value for k in detail_node.keys
                if isinstance(k, ast.Constant)
                and isinstance(k.value, str)
                and is_money_detail_key(k.value)
            ]
            if money and action_node.value not in AUDIT_ACTION_COST_BASIS:
                unclassified.append(
                    f"{py.relative_to(app_root.parent)}:{node.lineno} "
                    f"action={action_node.value!r} money_keys={money}",
                )
    assert not unclassified, (
        "audit writers with money-shaped detail keys that are NOT in "
        "app.core.masking.AUDIT_ACTION_COST_BASIS — an out-of-circle "
        "reader (auditor / management) will see '[redacted]' there. "
        "Classify each one (COST_BASIS_NON_CLOUD for asset / "
        "procurement / finance money, which has no list-vs-invoice "
        "duality):\n  " + "\n  ".join(sorted(unclassified))
    )


def test_nested_money_map_is_redacted_wholesale():
    out = redact_invoice_basis_amounts(
        {"cost_by_service": {"Compute Engine": 812.4, "BigQuery": 351.0},
         "nested": {"current_usd": "1163.42", "note": "ok"}},
        action="cloud.unclassified.alert", can_view_invoice=False,
    )
    assert out["cost_by_service"] == REDACTED_AMOUNT
    assert out["nested"]["current_usd"] == REDACTED_AMOUNT
    assert out["nested"]["note"] == "ok"
    assert "nested.current_usd" in out["_redacted_fields"]


# ── API: /audit/logs ─────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_logs_redacts_net_row_for_auditor(
    client: AsyncClient, db_session, auditor_headers,
):
    resource_id, _ = await _seed_alert(db_session, detail=_net_detail())
    resp = await client.get(
        "/api/v1/audit/logs",
        params={"action": SCOPE_ALERT_ACTION, "resource_id": resource_id},
        headers=auditor_headers,
    )
    assert resp.status_code == 200, resp.text
    [row] = resp.json()["data"]
    for key in _ABSOLUTE_KEYS:
        assert row["detail"][key] == REDACTED_AMOUNT, key
    assert row["detail"]["multiplier"] == 1.5
    assert "1163.42" not in resp.text


@pytest.mark.asyncio
async def test_logs_redacts_net_row_for_management(
    client: AsyncClient, db_session, management_headers,
):
    resource_id, _ = await _seed_alert(db_session, detail=_net_detail())
    resp = await client.get(
        "/api/v1/audit/logs",
        params={"action": SCOPE_ALERT_ACTION, "resource_id": resource_id},
        headers=management_headers,
    )
    assert resp.status_code == 200, resp.text
    [row] = resp.json()["data"]
    assert row["detail"]["current_usd"] == REDACTED_AMOUNT


@pytest.mark.asyncio
async def test_logs_keeps_net_row_intact_for_global_admin(
    client: AsyncClient, db_session, admin_headers,
):
    resource_id, _ = await _seed_alert(db_session, detail=_net_detail())
    resp = await client.get(
        "/api/v1/audit/logs",
        params={"action": SCOPE_ALERT_ACTION, "resource_id": resource_id},
        headers=admin_headers,
    )
    assert resp.status_code == 200, resp.text
    [row] = resp.json()["data"]
    assert row["detail"]["current_usd"] == "1163.42"
    assert row["detail"]["fx_rate_cny_per_usd"] == 7.21
    assert "_redacted_fields" not in row["detail"]


@pytest.mark.asyncio
async def test_logs_list_basis_row_visible_to_everyone(
    client: AsyncClient, db_session, auditor_headers, admin_headers,
    management_headers,
):
    resource_id, _ = await _seed_alert(db_session, detail=_list_detail())
    for headers in (auditor_headers, management_headers, admin_headers):
        resp = await client.get(
            "/api/v1/audit/logs",
            params={"action": SCOPE_ALERT_ACTION, "resource_id": resource_id},
            headers=headers,
        )
        assert resp.status_code == 200, resp.text
        [row] = resp.json()["data"]
        assert row["detail"]["current_usd"] == "1163.42"
        assert row["detail"]["baseline_usd"] == "706.00"
        assert "_redacted_fields" not in row["detail"]


@pytest.mark.asyncio
async def test_log_detail_and_trace_paths_redact(
    client: AsyncClient, db_session, auditor_headers,
):
    resource_id, trace_id = await _seed_alert(
        db_session, detail=_net_detail(),
    )
    listed = await client.get(
        "/api/v1/audit/logs",
        params={"action": SCOPE_ALERT_ACTION, "resource_id": resource_id},
        headers=auditor_headers,
    )
    log_id = listed.json()["data"][0]["id"]

    single = await client.get(
        f"/api/v1/audit/logs/{log_id}", headers=auditor_headers,
    )
    assert single.status_code == 200, single.text
    assert single.json()["data"]["detail"]["current_usd"] == REDACTED_AMOUNT

    trace = await client.get(
        f"/api/v1/audit/trace/{trace_id}", headers=auditor_headers,
    )
    assert trace.status_code == 200, trace.text
    [row] = [
        r for r in trace.json()["data"] if r["resource_id"] == resource_id
    ]
    assert row["detail"]["baseline_usd"] == REDACTED_AMOUNT


# ── API: /audit/resources/history ────────────────────────────────────


@pytest.mark.asyncio
async def test_resource_history_redacts_for_auditor_not_cloud_admin(
    client: AsyncClient, db_session, auditor_headers, cloud_admin_headers,
):
    """``cmdb*`` history admits cloud_admin (in the circle) AND auditor
    (outside it) — the same endpoint must answer differently."""
    resource_id, _ = await _seed_alert(db_session, detail=_net_detail())
    params = {
        "resource_type": SCOPE_RESOURCE_TYPE, "resource_id": resource_id,
    }

    aud = await client.get(
        "/api/v1/audit/resources/history",
        params=params, headers=auditor_headers,
    )
    assert aud.status_code == 200, aud.text
    [row] = aud.json()["data"]
    assert row["detail"]["current_usd"] == REDACTED_AMOUNT
    assert "1163.42" not in aud.text

    ca = await client.get(
        "/api/v1/audit/resources/history",
        params=params, headers=cloud_admin_headers,
    )
    assert ca.status_code == 200, ca.text
    [row] = ca.json()["data"]
    assert row["detail"]["current_usd"] == "1163.42"


# ── API: /audit/export (XLSX) ────────────────────────────────────────


def _detail_cell(content: bytes) -> str:
    ws = load_workbook(io.BytesIO(content)).active
    col = next(
        c for c in range(1, ws.max_column + 1)
        if ws.cell(row=2, column=c).value == "detail"
    )
    return ws.cell(row=3, column=col).value or ""


@pytest.mark.asyncio
async def test_export_redacts_net_row_for_auditor(
    client: AsyncClient, db_session, auditor_headers,
):
    resource_id, _ = await _seed_alert(db_session, detail=_net_detail())
    resp = await client.get(
        "/api/v1/audit/export",
        params={"resource_id": resource_id},
        headers=auditor_headers,
    )
    assert resp.status_code == 200, resp.text
    cell = _detail_cell(resp.content)
    assert "1163.42" not in cell
    assert "706.00" not in cell
    assert "7.21" not in cell
    assert REDACTED_REASON in cell
    # The ratio and the forensic context survive the export too.
    assert "1.5" in cell
    assert "daily_spike" in cell


@pytest.mark.asyncio
async def test_export_keeps_net_row_for_global_admin(
    client: AsyncClient, db_session, admin_headers,
):
    resource_id, _ = await _seed_alert(db_session, detail=_net_detail())
    resp = await client.get(
        "/api/v1/audit/export",
        params={"resource_id": resource_id},
        headers=admin_headers,
    )
    assert resp.status_code == 200, resp.text
    cell = _detail_cell(resp.content)
    assert "1163.42" in cell
    assert REDACTED_REASON not in cell


@pytest.mark.asyncio
async def test_export_unknown_money_action_fails_safe_for_auditor(
    client: AsyncClient, db_session, auditor_headers,
):
    resource_id, _ = await _seed_alert(
        db_session,
        detail={"net_spend_usd": "912.55", "scope_slug": "prod-a"},
        action="cloud.future_cost_alert.fired",
    )
    resp = await client.get(
        "/api/v1/audit/export",
        params={"resource_id": resource_id},
        headers=auditor_headers,
    )
    assert resp.status_code == 200, resp.text
    cell = _detail_cell(resp.content)
    assert "912.55" not in cell
    assert "prod-a" in cell