"""HTTP + MCP surface for the evidence-query catalogue.

The surface is where an access-forensics feature becomes dangerous: it
turns "who had access to what" into something fetchable. So the tests
here are mostly about who is refused, and about the platform gates a new
read path is easy to forget.
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime, timedelta

import pytest

from app.audit.models import AuditLog
from app.core.access_cycle_listener import register_listener
from app.domains.asset.models import (
    AssetSoftwareCatalog,
    SoftwareLicensePlan,
    SoftwareSeatAllocation,
)

register_listener()

BASE = "/api/v1/auditor"


async def _seed_seat(db, person_id):
    software = AssetSoftwareCatalog(name=f"ApiSW {uuid.uuid4().hex[:6]}")
    db.add(software)
    await db.flush()
    plan = SoftwareLicensePlan(
        software_id=software.id, name="Pro", price=1, billing_cycle="monthly",
    )
    db.add(plan)
    await db.flush()
    db.add(SoftwareSeatAllocation(
        plan_id=plan.id, user_id=person_id, status="active",
        allocated_at=datetime.now(UTC) - timedelta(days=3),
    ))
    await db.flush()
    return software


# ── catalogue ─────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_catalog_lists_every_registered_query(client, auditor_headers):
    resp = await client.get(f"{BASE}/evidence-queries", headers=auditor_headers)
    assert resp.status_code == 200
    items = resp.json()["data"]["items"]
    keys = {i["key"] for i in items}
    assert "privileged_role_roster" in keys
    assert "revocation_proof" in keys
    for item in items:
        assert set(item["title"]) == {"zh-CN", "en-US", "ja-JP"}


@pytest.mark.asyncio
async def test_employee_cannot_read_the_catalog(client, employee_headers):
    """The catalogue names every access question the platform can answer;
    it is itself security-sensitive reconnaissance."""
    resp = await client.get(
        f"{BASE}/evidence-queries", headers=employee_headers,
    )
    assert resp.status_code == 403


@pytest.mark.asyncio
async def test_anonymous_is_refused(client):
    resp = await client.get(f"{BASE}/evidence-queries")
    assert resp.status_code in (401, 403)


# ── running ───────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_run_freezes_a_bundle(client, db_session, admin_headers, employee_user):
    employee, _ = employee_user
    software = await _seed_seat(db_session, employee.id)

    resp = await client.post(
        f"{BASE}/evidence-queries/software_seat_roster/run",
        headers=admin_headers,
        json={"params": {"software_id": str(software.id)}},
    )
    assert resp.status_code == 200
    data = resp.json()["data"]
    assert data["row_count"] == 1
    assert data["content_hash"]
    assert data["generated_via"] == "manual"


@pytest.mark.asyncio
async def test_employee_cannot_run_a_query(client, employee_headers):
    resp = await client.post(
        f"{BASE}/evidence-queries/software_seat_roster/run",
        headers=employee_headers,
        json={"params": {}},
    )
    assert resp.status_code == 403


@pytest.mark.asyncio
async def test_unknown_query_key_is_404(client, admin_headers):
    resp = await client.post(
        f"{BASE}/evidence-queries/definitely_not_a_query/run",
        headers=admin_headers,
        json={"params": {}},
    )
    assert resp.status_code == 404


@pytest.mark.asyncio
async def test_unknown_param_is_400_not_ignored(client, admin_headers):
    resp = await client.post(
        f"{BASE}/evidence-queries/software_seat_roster/run",
        headers=admin_headers,
        json={"params": {"softwar_id": str(uuid.uuid4())}},
    )
    assert resp.status_code == 400
    assert resp.json()["errors"][0]["code"] == "EVIDENCE_QUERY_UNKNOWN_PARAM"


# ── reading bundles ───────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_bundle_rows_are_paged_and_cited(
    client, db_session, admin_headers, employee_user,
):
    employee, _ = employee_user
    software = await _seed_seat(db_session, employee.id)
    run = await client.post(
        f"{BASE}/evidence-queries/software_seat_roster/run",
        headers=admin_headers,
        json={"params": {"software_id": str(software.id)}},
    )
    bundle_id = run.json()["data"]["id"]

    resp = await client.get(
        f"{BASE}/evidence-bundles/{bundle_id}/rows", headers=admin_headers,
    )
    assert resp.status_code == 200
    body = resp.json()
    # Nested inside ``data`` — the shape the frontend's PaginatedList
    # reads. Asserting it at envelope level passed while the page
    # crashed on ``meta.total`` in a live run.
    assert body["data"]["meta"]["total"] == 1
    row = body["data"]["items"][0]
    assert row["source_table"] == "software_seat_allocations"
    assert row["source_pk"]


@pytest.mark.asyncio
async def test_missing_bundle_is_404(client, admin_headers):
    resp = await client.get(
        f"{BASE}/evidence-bundles/{uuid.uuid4()}", headers=admin_headers,
    )
    assert resp.status_code == 404


@pytest.mark.asyncio
async def test_employee_cannot_read_bundles(client, employee_headers):
    resp = await client.get(f"{BASE}/evidence-bundles", headers=employee_headers)
    assert resp.status_code == 403


@pytest.mark.asyncio
async def test_page_size_cap_is_enforced(client, admin_headers):
    """Above the cap must 422, not silently clamp — the frontend pins its
    page size to this bound."""
    resp = await client.get(
        f"{BASE}/evidence-bundles?page_size=500", headers=admin_headers,
    )
    assert resp.status_code == 422


# ── the cost-visibility gate a new read path is easy to forget ────────


@pytest.mark.asyncio
async def test_invoice_amounts_are_redacted_outside_the_circle(
    client, db_session, auditor_headers, admin_user,
):
    """Evidence rows sourced from the audit trail carry the original
    ``audit_logs.detail`` verbatim, which can contain cloud money. This
    is a NEW read path over those same rows; skipping the platform's
    invoice-measure gate here would be a leak wearing a new endpoint.
    """
    admin, _ = admin_user
    resource_id = str(uuid.uuid4())
    db_session.add(AuditLog(
        trace_id=uuid.uuid4(),
        actor_user_id=admin.id,
        actor_type="user",
        action="cloud.scope.update",
        status="success",
        resource_type="cmdb_cloud_scope",
        resource_id=resource_id,
        detail={"invoice_amount_usd": 12345.67},
    ))
    await db_session.flush()

    run = await client.post(
        f"{BASE}/evidence-queries/object_lifecycle/run",
        headers=auditor_headers,
        json={"params": {
            "resource_type": "cmdb_cloud_scope",
            "resource_id": resource_id,
        }},
    )
    assert run.status_code == 200
    bundle_id = run.json()["data"]["id"]

    rows = await client.get(
        f"{BASE}/evidence-bundles/{bundle_id}/rows", headers=auditor_headers,
    )
    detail = rows.json()["data"]["items"][0]["payload"]["detail"]
    assert detail.get("invoice_amount_usd") != 12345.67, (
        "an unregistered money-bearing audit action must not pass through "
        "un-redacted to a reader outside the invoice-measure circle"
    )


@pytest.mark.asyncio
async def test_non_cloud_payloads_are_not_over_redacted(
    client, db_session, auditor_headers, employee_user,
):
    """The mirror-image failure.

    The redaction table is keyed by audit action and fails closed on
    unknown ones. Pointing it at a whole evidence payload with a
    synthetic action would blank out money-shaped fields on rows that
    have nothing to do with cloud cost — and that regression is silent.
    """
    employee, _ = employee_user
    software = await _seed_seat(db_session, employee.id)
    run = await client.post(
        f"{BASE}/evidence-queries/software_seat_roster/run",
        headers=auditor_headers,
        json={"params": {"software_id": str(software.id)}},
    )
    bundle_id = run.json()["data"]["id"]
    rows = await client.get(
        f"{BASE}/evidence-bundles/{bundle_id}/rows", headers=auditor_headers,
    )
    payload = rows.json()["data"]["items"][0]["payload"]
    assert payload["plan"] == "Pro"
    assert payload["still_held"] is True


# ── MCP twins ─────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_mcp_tools_are_registered_with_the_right_mutation_flag():
    """``run`` writes a bundle and an audit row.

    Annotating it read-only would let a client replay it as a cache fill
    and quietly accumulate dated evidence artefacts.
    """
    from app.mcp.server import mcp_server

    tools = await mcp_server.list_tools()
    by_name = {t.name: t for t in tools}
    assert "auditor_list_evidence_queries" in by_name
    assert "auditor_run_evidence_query" in by_name

    listing = by_name["auditor_list_evidence_queries"].annotations
    running = by_name["auditor_run_evidence_query"].annotations
    assert listing.readOnlyHint is True
    assert running.readOnlyHint is False


@pytest.mark.asyncio
async def test_tool_count_hint_matches_reality():
    """Guards the NON-NEGOTIABLE hint rule for this PR's two tools.

    (The hint was also six behind on main before this change — six tools
    had landed without a bump, which is exactly the drift the rule
    exists to prevent. Corrected to the true count here.)
    """
    import re

    from app.mcp.server import mcp_server
    from app.mcp.tools import _REGISTERED_TOOLS_COUNT_HINT

    tools = await mcp_server.list_tools()
    assert _REGISTERED_TOOLS_COUNT_HINT == len(tools)
    assert re.match(r"^\d+$", str(_REGISTERED_TOOLS_COUNT_HINT))


# ── evidence pack module ──────────────────────────────────────────────


@pytest.mark.asyncio
async def test_access_evidence_is_a_registered_pack_module():
    """The eight pre-existing modules are all detection outputs; none of
    them answers the CC6.x access questions an auditor opens with."""
    from app.domains.auditor.constants import (
        ALL_EVIDENCE_MODULES,
        EVIDENCE_MODULE_ACCESS_EVIDENCE,
    )

    assert EVIDENCE_MODULE_ACCESS_EVIDENCE in ALL_EVIDENCE_MODULES


@pytest.mark.asyncio
async def test_pack_module_renders_frozen_bundles_not_fresh_queries(
    db_session, admin_user, employee_user,
):
    """The pack must render STORED rows.

    A pack that re-derived its numbers at build time would disagree with
    the bundle an auditor already reviewed, and the hash printed next to
    it would no longer verify anything.
    """
    from app.domains.auditor.constants import EVIDENCE_MODULE_ACCESS_EVIDENCE
    from app.domains.auditor.evidence_query.service import run_evidence_query
    from app.domains.auditor.models import AuditorEvidencePack
    from app.domains.auditor.services import EvidencePackService

    admin, _ = admin_user
    employee, _ = employee_user
    software = await _seed_seat(db_session, employee.id)
    bundle = await run_evidence_query(
        db_session, query_key="software_seat_roster",
        params={"software_id": str(software.id)},
        actor_user_id=admin.id,
    )

    pack = AuditorEvidencePack(
        title="Q3 pack",
        period_start=datetime.now(UTC) - timedelta(days=1),
        period_end=datetime.now(UTC) + timedelta(days=1),
        modules=[EVIDENCE_MODULE_ACCESS_EVIDENCE],
        requested_by_person_id=admin.id,
        status="pending",
    )
    db_session.add(pack)
    await db_session.flush()

    section = await EvidencePackService._render_module(
        db_session,
        module_name=EVIDENCE_MODULE_ACCESS_EVIDENCE,
        pack=pack,
    )
    assert "1 frozen evidence bundle" in section["summary"]
    # The stored hash appears verbatim, so a reader can verify it against
    # the bundle they were shown.
    assert bundle.content_hash[:16] in str(section["table"])
    assert "software_seat_allocations" in section["csv"]


@pytest.mark.asyncio
async def test_pack_csv_quotes_labels_containing_commas(
    db_session, admin_user, employee_user,
):
    """A comma in a display name silently shifts every later column.

    In a compliance artefact that goes to an external auditor, that is a
    corrupted exhibit rather than a cosmetic bug.
    """
    from app.domains.auditor.constants import EVIDENCE_MODULE_ACCESS_EVIDENCE
    from app.domains.auditor.evidence_query.service import run_evidence_query
    from app.domains.auditor.models import AuditorEvidencePack
    from app.domains.auditor.services import EvidencePackService

    admin, _ = admin_user
    employee, _ = employee_user
    employee.display_name = "Doe, Jane"
    await db_session.flush()

    software = await _seed_seat(db_session, employee.id)
    await run_evidence_query(
        db_session, query_key="software_seat_roster",
        params={"software_id": str(software.id)},
        actor_user_id=admin.id,
    )

    pack = AuditorEvidencePack(
        title="Comma pack",
        period_start=datetime.now(UTC) - timedelta(days=1),
        period_end=datetime.now(UTC) + timedelta(days=1),
        modules=[EVIDENCE_MODULE_ACCESS_EVIDENCE],
        requested_by_person_id=admin.id,
        status="pending",
    )
    db_session.add(pack)
    await db_session.flush()

    section = await EvidencePackService._render_module(
        db_session,
        module_name=EVIDENCE_MODULE_ACCESS_EVIDENCE,
        pack=pack,
    )
    data_line = section["csv"].splitlines()[1]
    assert '"Doe, Jane"' in data_line


@pytest.mark.asyncio
async def test_the_mcp_twin_redacts_exactly_like_the_http_one(
    db_session, admin_user,
):
    """A twin that skips the gate IS the way around the gate.

    ``auditor_run_evidence_query`` returns rows inline, and
    ``object_lifecycle`` embeds raw ``audit_logs.detail`` — which can
    carry invoice-basis amounts that an ``auditor``, sitting outside the
    invoice circle, must never receive. The HTTP path redacts them. If
    the MCP path does not, the platform red line simply has a second
    door, and the registration test above would never notice: it checks
    names and mutation flags, not payloads.

    Asserted against the shared helper on an auditor-shaped user, which
    is the same call the tool makes.
    """
    from app.domains.auditor.evidence_query.router import _redacted_payload

    admin, _ = admin_user
    resource_id = str(uuid.uuid4())
    db_session.add(AuditLog(
        trace_id=uuid.uuid4(), actor_user_id=admin.id, actor_type="user",
        action="cloud.scope.update", status="success",
        resource_type="cmdb_cloud_scope", resource_id=resource_id,
        detail={"invoice_amount_usd": 12345.67},
    ))
    await db_session.flush()

    from app.domains.auditor.evidence_query.service import (
        list_bundle_rows,
        run_evidence_query,
    )

    bundle = await run_evidence_query(
        db_session, query_key="object_lifecycle",
        params={
            "resource_type": "cmdb_cloud_scope", "resource_id": resource_id,
        },
        actor_user_id=admin.id,
    )
    rows, _ = await list_bundle_rows(db_session, bundle.id, page_size=200)
    assert rows, "the audit row must reach the bundle"

    class _Auditor:
        role_names = ["employee", "auditor"]

    redacted = _redacted_payload(rows[0].payload, _Auditor())
    assert redacted["detail"].get("invoice_amount_usd") != 12345.67, (
        "the MCP twin must not hand an auditor money the HTTP twin hides"
    )
    # And the raw row really did carry it — otherwise this proves nothing.
    assert rows[0].payload["detail"]["invoice_amount_usd"] == 12345.67

    # The tool must actually call it: a redaction nobody invokes is
    # decoration, which is exactly how this shipped the first time.
    import inspect

    from app.mcp import tools

    src = inspect.getsource(tools.auditor_run_evidence_query)
    assert "_redacted_payload(r.payload" in src

async def _freeze_one(client, admin_headers, person_id) -> str:
    """Freeze one bundle and hand back its id."""
    resp = await client.post(
        f"{BASE}/evidence-queries/person_footprint/run",
        json={"params": {"person_id": str(person_id)}},
        headers=admin_headers,
    )
    assert resp.status_code == 200, resp.text
    return resp.json()["data"]["id"]


@pytest.mark.asyncio
async def test_leadership_read_tier_cannot_open_a_persons_footprint(
    client, db_session, admin_headers, employee_user,
):
    """``can_view_security`` admits ``management`` for aggregate posture.

    A bundle is not aggregate posture: it is one named person's complete
    access footprint with display names in the clear, and any holder
    could enumerate any colleague's — executives included. Reading one
    back must not be broader than producing one.
    """
    from app.auth.models import AuthUserRole
    from app.core.security import create_access_token
    from app.identity.person_models import IdentityPerson
    from tests.conftest import _get_role_id

    employee, _ = employee_user
    await _seed_seat(db_session, employee.id)
    bundle_id = await _freeze_one(client, admin_headers, employee.id)

    leader_id = uuid.uuid4()
    db_session.add(IdentityPerson(
        id=leader_id, work_email=f"leader-{leader_id.hex[:6]}@test.com",
        display_name="Leadership Reader", employment_status="active",
        hris_person_id=f"sf-mgmt-{leader_id.hex[:6]}",
    ))
    await db_session.flush()
    db_session.add(AuthUserRole(
        user_id=leader_id, role_id=await _get_role_id(db_session, "management"),
    ))
    await db_session.flush()
    await db_session.commit()
    headers = {
        "Authorization": "Bearer " + create_access_token(
            str(leader_id), ["employee", "management"],
        )
    }

    for path in (
        f"{BASE}/evidence-bundles",
        f"{BASE}/evidence-bundles/{bundle_id}",
        f"{BASE}/evidence-bundles/{bundle_id}/rows",
    ):
        assert (await client.get(path, headers=headers)).status_code == 403, path
@pytest.mark.asyncio
async def test_opening_a_bundle_writes_an_audit_row(
    client, db_session, admin_headers, employee_user,
):
    """Running the query was audited; reading the result back was not.

    So the act an auditor actually performs when looking into a named
    colleague left no trace, while producing the evidence did. This
    module's docstring calls an unaudited forensics tool a governance
    hole — that claim only ever covered the write path.
    """
    from sqlalchemy import select

    employee, _ = employee_user
    await _seed_seat(db_session, employee.id)
    bundle_id = await _freeze_one(client, admin_headers, employee.id)

    resp = await client.get(
        f"{BASE}/evidence-bundles/{bundle_id}/rows", headers=admin_headers,
    )
    assert resp.status_code == 200

    rows = (await db_session.execute(
        select(AuditLog).where(
            AuditLog.action == "auditor.evidence_bundle.read",
            AuditLog.resource_id == bundle_id,
        )
    )).scalars().all()
    assert rows, "reading a frozen footprint must be traceable"
    assert rows[-1].detail["subject_person_id"] == str(employee.id)
    assert rows[-1].detail["included_rows"] is True