"""GRC spine — control catalog, crosswalk, CCM, posture projection.

Covers PR1 of the 安全合规 redesign (migration 250):

* catalog reconcile is correct + idempotent
* control evaluation: freshness gate (no job → not_configured; stale job
  → stale; fresh + clean → pass)
* framework readiness projection (covered_pass / partial / fail / none)
* HTTP RBAC: read open to admin/auditor, 403 for employee; recompute
  gated to manage roles; unknown framework → 404
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime, timedelta

import pytest
from httpx import AsyncClient
from sqlalchemy import func, select

from app.domains.auditor import compliance_catalog as cat
from app.domains.auditor import compliance_service as svc
from app.domains.auditor.models import (
    AccessCertification,
    ComplianceControl,
    ComplianceControlStatus,
)
from app.scheduled_jobs.models import ScheduledJob

# ── catalog reconcile ────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_sync_catalog_seeds_and_is_idempotent(db_session):
    counts = await svc.sync_compliance_catalog(db_session)
    await db_session.commit()
    assert counts["frameworks"] == len(cat.FRAMEWORKS)
    assert counts["controls"] == len(cat.CONTROLS)
    # Derived from the catalog, like the two assertions above. These were
    # hardcoded (16 / 35) and had to be edited by hand every time a
    # control or a requirement was legitimately added — which makes the
    # guard a maintenance tax rather than a check, and tempts the next
    # person to bump the number without looking at what changed. What
    # this test is actually for is that the sync WRITES everything the
    # catalog declares, exactly once.
    assert counts["requirements"] == sum(
        len(v) for v in cat.REQUIREMENTS.values()
    )
    assert counts["mappings"] == sum(
        len(refs)
        for per_control in cat.MAPPINGS.values()
        for refs in per_control.values()
    )

    # Re-run: nothing new.
    again = await svc.sync_compliance_catalog(db_session)
    await db_session.commit()
    assert again == {
        "frameworks": 0, "requirements": 0, "controls": 0, "mappings": 0,
    }


# ── control evaluation / freshness gate ──────────────────────────────


@pytest.mark.asyncio
async def test_recompute_no_jobs_is_not_configured(db_session):
    await svc.sync_compliance_catalog(db_session)
    await db_session.commit()
    res = await svc.recompute_control_status(db_session)
    # Bare test DB has no scheduled_jobs rows → every control's signal is
    # absent → not_configured (never silently "pass").
    assert res["control_count"] == len(cat.CONTROLS)
    assert res["by_status"].get("not_configured") == len(cat.CONTROLS)


@pytest.mark.asyncio
async def test_recompute_is_idempotent_upsert(db_session):
    """Re-running recompute must UPDATE in place, not insert a second
    status row per control (regression: the read-then-insert path raced
    into a UNIQUE(control_id) violation; now an ON DUPLICATE KEY UPDATE
    upsert). Two sequential recomputes leave exactly one row per control."""
    await svc.sync_compliance_catalog(db_session)
    await db_session.commit()

    first = await svc.recompute_control_status(db_session)
    second = await svc.recompute_control_status(db_session)

    assert second["control_count"] == first["control_count"]
    n_status = int((await db_session.execute(
        select(func.count()).select_from(ComplianceControlStatus)
    )).scalar_one())
    n_controls = int((await db_session.execute(
        select(func.count()).select_from(ComplianceControl).where(
            ComplianceControl.enabled.is_(True))
    )).scalar_one())
    assert n_status == n_controls  # one status row per control, no dupes


@pytest.mark.asyncio
async def test_access_certification_control_ignores_privileged_campaign(db_session):
    """The ``access_certification`` control attests general user-access
    review (``software_seat``). A ``privileged_role`` campaign (which the
    cron auto-opens quarterly) must NOT satisfy it — otherwise posture
    shows a false PASS even when no user-access review ever ran. Regression
    for the latest_campaign_status(any-type) masking bug.
    """
    await svc.sync_compliance_catalog(db_session)
    # Only a completed privileged_role campaign exists — no software_seat.
    db_session.add(AccessCertification(
        id=uuid.uuid4(), label="Q-priv", campaign_type="privileged_role",
        scope_config={}, status="completed",
        started_at=datetime.now(UTC).replace(tzinfo=None),
        due_at=datetime.now(UTC).replace(tzinfo=None),
        completed_at=datetime.now(UTC).replace(tzinfo=None),
        snapshot_item_count=3, decided_count=3, revoked_count=0,
    ))
    await db_session.commit()

    await svc.recompute_control_status(db_session)
    ctrl = (await db_session.execute(
        select(ComplianceControl).where(
            ComplianceControl.key == "access_certification")
    )).scalar_one()
    st = (await db_session.execute(
        select(ComplianceControlStatus).where(
            ComplianceControlStatus.control_id == ctrl.id)
    )).scalar_one()
    # No software_seat campaign → not_configured, NOT pass.
    assert st.status == cat.STATUS_NOT_CONFIGURED


def _job(handler_key: str, last_run_at) -> ScheduledJob:
    return ScheduledJob(
        id=uuid.uuid4(),
        name=handler_key,
        display_name=handler_key,
        handler_key=handler_key,
        trigger_type="cron",
        cron_expr="0 * * * *",
        timezone="UTC",
        enabled=True,
        last_run_at=last_run_at,
    )


@pytest.mark.asyncio
async def test_fresh_job_clean_signal_passes(db_session):
    await svc.sync_compliance_catalog(db_session)
    # public_exposure is a "binary clean" control: 0 open findings → pass.
    db_session.add(_job(
        "auditor.public_exposure_scan_daily", datetime.now(UTC),
    ))
    await db_session.commit()

    await svc.recompute_control_status(db_session)
    ctrl = (await db_session.execute(
        select(ComplianceControl).where(
            ComplianceControl.key == "public_exposure_scan")
    )).scalar_one()
    st = (await db_session.execute(
        select(ComplianceControlStatus).where(
            ComplianceControlStatus.control_id == ctrl.id)
    )).scalar_one()
    assert st.status == cat.STATUS_PASS


@pytest.mark.asyncio
async def test_stale_job_is_stale(db_session):
    await svc.sync_compliance_catalog(db_session)
    # Daily control whose job last ran 5 days ago → past the 48h window.
    db_session.add(_job(
        "auditor.public_exposure_scan_daily",
        datetime.now(UTC) - timedelta(days=5),
    ))
    await db_session.commit()

    await svc.recompute_control_status(db_session)
    ctrl = (await db_session.execute(
        select(ComplianceControl).where(
            ComplianceControl.key == "public_exposure_scan")
    )).scalar_one()
    st = (await db_session.execute(
        select(ComplianceControlStatus).where(
            ComplianceControlStatus.control_id == ctrl.id)
    )).scalar_one()
    assert st.status == cat.STATUS_STALE


# ── readiness projection (deterministic, status set by hand) ─────────


async def _set_status(db_session, control_key: str, status: str) -> None:
    ctrl = (await db_session.execute(
        select(ComplianceControl).where(ComplianceControl.key == control_key)
    )).scalar_one()
    row = (await db_session.execute(
        select(ComplianceControlStatus).where(
            ComplianceControlStatus.control_id == ctrl.id)
    )).scalar_one_or_none()
    now = datetime.now(UTC).replace(tzinfo=None)
    if row is None:
        db_session.add(ComplianceControlStatus(
            control_id=ctrl.id, status=status, evaluated_at=now))
    else:
        row.status = status
        row.evaluated_at = now


@pytest.mark.asyncio
async def test_framework_readiness_projection(db_session):
    await svc.sync_compliance_catalog(db_session)
    # Make every control pass → SOC2 fully covered except clauses with no
    # mapped control (none for SOC2 — all 7 are mapped).
    for spec in cat.CONTROLS:
        await _set_status(db_session, spec["key"], cat.STATUS_PASS)
    await db_session.commit()

    fw = await svc.get_framework(db_session, "soc2")
    assert fw["readiness_pct"] == 100
    assert fw["coverage"][cat.COVERAGE_PASS] == fw["requirement_count"]

    # After PR2 every seeded ISO clause is mapped (A.5.15 → the new
    # access_certification control), so with all controls passing the
    # framework is fully covered.
    iso = await svc.get_framework(db_session, "iso27001")
    assert iso["coverage"][cat.COVERAGE_NONE] == 0
    assert iso["readiness_pct"] == 100

    # Flip one control to fail → its requirements drop out of covered_pass.
    await _set_status(db_session, "mfa_coverage", cat.STATUS_FAIL)
    await db_session.commit()
    fw2 = await svc.get_framework(db_session, "soc2")
    assert fw2["readiness_pct"] < 100
    # CC6.1 is mapped to mfa + secret + privileged → partial, not fail.
    cc61 = next(r for r in fw2["requirements"] if r["ref"] == "CC6.1")
    assert cc61["coverage"] == cat.COVERAGE_PARTIAL


@pytest.mark.asyncio
async def test_disabled_control_excluded_from_rollup(db_session):
    """Regression (2026-06-16): a control with enabled=False must drop out
    of the posture control rollup — not surface as a phantom entry.
    _load_graph now filters enabled=True to match recompute_control_status.
    Coverage math (mappings/statuses) is intentionally left untouched.
    """
    await svc.sync_compliance_catalog(db_session)
    for spec in cat.CONTROLS:
        await _set_status(db_session, spec["key"], cat.STATUS_PASS)
    await db_session.commit()

    baseline = await svc.get_posture(db_session)
    assert baseline["controls"]["total"] == len(cat.CONTROLS)

    # Disable one control (keep its PASS status row, so coverage math is
    # unaffected — only the rollup membership changes).
    ctrl = (await db_session.execute(
        select(ComplianceControl).where(ComplianceControl.key == "mfa_coverage")
    )).scalar_one()
    ctrl.enabled = False
    await db_session.commit()

    posture = await svc.get_posture(db_session)
    # The disabled control drops out of the rollup total entirely.
    assert posture["controls"]["total"] == len(cat.CONTROLS) - 1
    # Coverage math is untouched (status rows still intact) → readiness
    # unchanged. Disabling a control does NOT silently rewrite coverage.
    soc2 = next(f for f in posture["frameworks"] if f["key"] == "soc2")
    assert soc2["readiness_pct"] == 100


# ── HTTP surface + RBAC ──────────────────────────────────────────────


@pytest.mark.asyncio
async def test_posture_http_rbac(client: AsyncClient, admin_headers,
                                 auditor_headers, employee_headers):
    # admin + auditor read; employee 403.
    for headers in (admin_headers, auditor_headers):
        resp = await client.get(
            "/api/v1/compliance/posture", headers=headers)
        assert resp.status_code == 200, resp.text
        body = resp.json()
        assert body["success"] is True
        keys = {f["key"] for f in body["data"]["frameworks"]}
        assert {"soc2", "iso27001"} <= keys

    resp = await client.get(
        "/api/v1/compliance/posture", headers=employee_headers)
    assert resp.status_code == 403


@pytest.mark.asyncio
async def test_framework_detail_and_404(client: AsyncClient, admin_headers):
    resp = await client.get(
        "/api/v1/compliance/frameworks/soc2", headers=admin_headers)
    assert resp.status_code == 200, resp.text
    data = resp.json()["data"]
    # From the catalog, not a literal — see the note in
    # ``test_sync_catalog_seeds_and_is_idempotent``.
    from app.domains.auditor import compliance_catalog as cat

    assert data["requirement_count"] == len(cat.REQUIREMENTS["soc2"])
    assert any(r["ref"] == "CC6.1" for r in data["requirements"])

    missing = await client.get(
        "/api/v1/compliance/frameworks/nope", headers=admin_headers)
    assert missing.status_code == 404


@pytest.mark.asyncio
async def test_recompute_requires_manage(client: AsyncClient, admin_headers,
                                         employee_headers):
    ok = await client.post(
        "/api/v1/compliance/controls/recompute", headers=admin_headers)
    assert ok.status_code == 200, ok.text
    assert ok.json()["data"]["control_count"] == len(cat.CONTROLS)

    denied = await client.post(
        "/api/v1/compliance/controls/recompute", headers=employee_headers)
    assert denied.status_code == 403


@pytest.mark.asyncio
async def test_list_controls_carries_status_and_refs(
    client: AsyncClient, admin_headers,
):
    resp = await client.get(
        "/api/v1/compliance/controls", headers=admin_headers)
    assert resp.status_code == 200, resp.text
    controls = resp.json()["data"]["controls"]
    assert len(controls) == len(cat.CONTROLS)
    mfa = next(c for c in controls if c["key"] == "mfa_coverage")
    assert "status" in mfa
    assert any(ref.startswith("soc2:") for ref in mfa["mapped_refs"])
    assert any(ref.startswith("iso27001:") for ref in mfa["mapped_refs"])