"""Monthly compliance-posture summary.

This is the only place in the whole feature where a model narrates, so
the tests are mostly about what it is NOT allowed to see or change: the
fact package must pass the platform privacy gate, confidence must be
computed rather than claimed, and a month whose facts have not moved
must not produce a second letter.
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime

import pytest

from app.core.access_cycle_listener import register_listener
from app.domains.auditor.models import (
    AuditorEvidenceBundle,
    AuditorFinding,
    ComplianceControl,
    ComplianceControlStatus,
)
from app.domains.auditor.posture_summary import (
    POSTURE_SCORE_VERSION,
    POSTURE_SUBJECT_TYPE,
    build_posture_facts,
    build_posture_snapshot,
)
from app.intelligence.privacy import assert_llm_safe

register_listener()


async def _seed_controls(db, *, passing: int, failing: int):
    for index in range(passing + failing):
        key = f"CTRL{index}_{uuid.uuid4().hex[:4]}"
        control = ComplianceControl(
            key=key, name=f"Control {index}", category="access",
            signal_type="detective", enabled=True,
        )
        db.add(control)
        await db.flush()
        db.add(ComplianceControlStatus(
            control_id=control.id,
            status="pass" if index < passing else "fail",
            evaluated_at=datetime.now(UTC),
        ))
    await db.flush()


# ── the privacy boundary ──────────────────────────────────────────────


@pytest.mark.asyncio
async def test_fact_package_passes_the_privacy_gate(db_session):
    """The gate is the reason a model may see this at all.

    If a future field ever carries a name, an email or a control key,
    this fails here rather than in a prompt.
    """
    await _seed_controls(db_session, passing=3, failing=2)
    _health, facts, reference_map = await build_posture_facts(
        db_session, period_start="2026-05-01", period_end="2026-05-31",
    )
    assert_llm_safe(facts)
    # Real control keys live only in the local map, never in the package.
    # The refs themselves moved to the reference_map channel; the fact
    # package carries only their count. Both still have to be free of
    # real control keys.
    assert isinstance(facts["failing_control_count"], int)
    assert all(ref.startswith("control_") for ref in reference_map)
    assert len(reference_map) == facts["failing_control_count"], (
        "the count must match the refs actually handed to the model"
    )
    assert not any(
        "CTRL" in str(value) for value in facts.values() if isinstance(value, str)
    )


@pytest.mark.asyncio
async def test_reference_numbering_is_stable_for_the_same_failing_set(
    db_session,
):
    """Unstable numbering would change the facts hash on every run and
    regenerate a settled month for no reason."""
    await _seed_controls(db_session, passing=1, failing=3)
    _h1, facts1, _m1 = await build_posture_facts(
        db_session, period_start="2026-05-01", period_end="2026-05-31",
    )
    _h2, facts2, _m2 = await build_posture_facts(
        db_session, period_start="2026-05-01", period_end="2026-05-31",
    )
    assert facts1 == facts2


# ── deterministic numbers ─────────────────────────────────────────────


@pytest.mark.asyncio
async def test_pass_ratio_and_grade_are_computed_not_claimed(db_session):
    await _seed_controls(db_session, passing=9, failing=1)
    health, facts, _ = await build_posture_facts(
        db_session, period_start="2026-05-01", period_end="2026-05-31",
    )
    assert facts["controls_total"] == 10
    assert facts["controls_passing"] == 9
    assert facts["control_pass_ratio"] == pytest.approx(0.9)
    assert health["overall_score"] == 90
    assert health["grade"] == "B"


@pytest.mark.asyncio
async def test_open_drift_findings_are_counted_by_severity(
    db_session, employee_user,
):
    from app.domains.auditor.access_drift import CONTROL_KEY as DRIFT

    employee, _ = employee_user
    await _seed_controls(db_session, passing=1, failing=0)
    for severity in ("high", "high", "medium"):
        db_session.add(AuditorFinding(
            control_key=DRIFT,
            finding_type="leaver_access_not_revoked",
            dedup_key=uuid.uuid4().hex,
            subject_type="person",
            subject_ref=str(employee.id),
            severity=severity,
            title="drift",
            status="open",
            first_seen_at=datetime.now(UTC).replace(tzinfo=None),
            last_seen_at=datetime.now(UTC).replace(tzinfo=None),
        ))
    await db_session.flush()

    _health, facts, _ = await build_posture_facts(
        db_session, period_start="2026-05-01", period_end="2026-05-31",
    )
    assert facts["open_drift_findings"] == 3
    assert {k.removeprefix("open_drift_"): v for k, v in facts.items() if k.startswith("open_drift_") and k != "open_drift_findings"}["high"] == 2
    assert {k.removeprefix("open_drift_"): v for k, v in facts.items() if k.startswith("open_drift_") and k != "open_drift_findings"}["medium"] == 1


@pytest.mark.asyncio
async def test_evidence_coverage_counts_precompiled_questions(db_session):
    await _seed_controls(db_session, passing=1, failing=0)
    for query_key in ("software_seat_roster", "sso_grant_roster"):
        db_session.add(AuditorEvidenceBundle(
            query_key=query_key,
            query_version="v1",
            params={},
            period_start=datetime(2026, 5, 1, tzinfo=UTC),
            period_end=datetime(2026, 5, 31, tzinfo=UTC),
            row_count=0,
            content_hash=uuid.uuid4().hex,
            generated_via="scheduled",
        ))
    await db_session.flush()

    # A bundle from a DIFFERENT month must not inflate this month's
    # coverage — counting every scheduled bundle ever would report full
    # coverage forever after the first good month, including a month
    # whose pre-compile failed outright.
    db_session.add(AuditorEvidenceBundle(
        query_key="api_key_roster", query_version="v1", params={},
        period_start=datetime(2026, 3, 1, tzinfo=UTC),
        period_end=datetime(2026, 3, 31, tzinfo=UTC),
        row_count=0, content_hash=uuid.uuid4().hex, generated_via="scheduled",
    ))
    await db_session.flush()

    _health, facts, _ = await build_posture_facts(
        db_session, period_start="2026-05-01", period_end="2026-05-31",
    )
    assert facts["evidence_queries_precompiled"] == 2


# ── confidence is deterministic, not a model opinion ──────────────────


@pytest.mark.asyncio
async def test_no_controls_yields_low_confidence(db_session):
    """The platform's low-confidence rule then forces the narrative to
    name the gap instead of writing around it."""
    health, _facts, _ = await build_posture_facts(
        db_session, period_start="2026-05-01", period_end="2026-05-31",
    )
    assert health["confidence_score"] < 70


@pytest.mark.asyncio
async def test_controls_without_evidence_are_mid_confidence(db_session):
    await _seed_controls(db_session, passing=2, failing=0)
    health, _facts, _ = await build_posture_facts(
        db_session, period_start="2026-05-01", period_end="2026-05-31",
    )
    assert health["confidence_score"] == 60


@pytest.mark.asyncio
async def test_controls_plus_evidence_are_high_confidence(db_session):
    await _seed_controls(db_session, passing=2, failing=0)
    db_session.add(AuditorEvidenceBundle(
        query_key="software_seat_roster", query_version="v1", params={},
        period_start=datetime(2026, 5, 1, tzinfo=UTC),
        period_end=datetime(2026, 5, 31, tzinfo=UTC),
        row_count=0, content_hash=uuid.uuid4().hex, generated_via="scheduled",
    ))
    await db_session.flush()
    health, _facts, _ = await build_posture_facts(
        db_session, period_start="2026-05-01", period_end="2026-05-31",
    )
    assert health["confidence_score"] >= 70


# ── snapshot + policy wiring ──────────────────────────────────────────


@pytest.mark.asyncio
async def test_snapshot_is_persisted_under_its_own_subject(db_session):
    await _seed_controls(db_session, passing=2, failing=1)
    result = await build_posture_snapshot(
        db_session, period_start="2026-05-01", period_end="2026-05-31",
    )
    assert result.snapshot.subject_type == POSTURE_SUBJECT_TYPE
    assert result.snapshot.score_version == POSTURE_SCORE_VERSION
    assert result.snapshot.subject_id == "2026-05"


@pytest.mark.asyncio
async def test_same_month_reuses_the_snapshot(db_session):
    """A settled month must not accumulate a new snapshot per run — the
    facts hash is what keeps 'regenerate' free."""
    await _seed_controls(db_session, passing=2, failing=1)
    first = await build_posture_snapshot(
        db_session, period_start="2026-05-01", period_end="2026-05-31",
    )
    second = await build_posture_snapshot(
        db_session, period_start="2026-05-01", period_end="2026-05-31",
    )
    assert first.snapshot.facts_hash == second.snapshot.facts_hash
    # Same facts must not accumulate a second immutable row.
    assert second.created is False


def test_posture_policy_has_a_long_cache_and_short_cooldown():
    from app.intelligence.generation import policy_for

    policy = policy_for(POSTURE_SUBJECT_TYPE)
    # A closed month never changes; the cooldown only guards a double
    # click, and the cache is what makes re-reading it free.
    assert policy.default_cache_days == 30
    assert policy.default_cooldown_minutes == 5


def test_subject_type_is_declared():
    """``SnapshotSubject`` rejects unlisted subjects at construction, so
    a typo fails immediately instead of persisting an unreadable row."""
    from app.intelligence.snapshot_service import SUBJECT_TYPES

    assert POSTURE_SUBJECT_TYPE in SUBJECT_TYPES


# ── notification discipline ───────────────────────────────────────────


@pytest.mark.asyncio
async def test_summary_reaches_only_auditors_and_admins(
    db_session, admin_user, auditor_user, employee_user,
):
    """A drift/compliance report is IT's work. The platform's standing
    rule is that new pushes never reach employees."""
    from app.domains.auditor.posture_summary import notify_posture_summary

    admin, _ = admin_user
    auditor, _ = auditor_user
    employee, _ = employee_user

    sent: list[uuid.UUID] = []

    class _Recorder:
        def __init__(self, _db):
            pass

        async def dispatch(self, env):
            sent.append(env.recipient.id)

    import app.notifications.service as notif

    original = notif.NotificationService
    notif.NotificationService = _Recorder
    try:
        await notify_posture_summary(
            db_session, analysis_id=uuid.uuid4(), period_label="2026-05",
        )
    finally:
        notif.NotificationService = original

    assert admin.id in sent
    assert auditor.id in sent
    assert employee.id not in sent, "employees must never receive this"


@pytest.mark.asyncio
async def test_cache_hit_sends_nothing(db_session, monkeypatch):
    """Sending the same month's letter twice is the cheapest way to make
    a monthly report useless."""
    from app.scheduled_jobs import posture_summary_handler as handler

    class _Result:
        cache_reused = True

        class analysis:  # noqa: N801 - stub
            id = uuid.uuid4()

    async def _fake_generate(*_args, **_kwargs):
        return _Result()

    class _Built:
        snapshot = object()
        created = False

    async def _fake_snapshot(*_args, **_kwargs):
        return _Built()

    notified: list[int] = []

    async def _fake_notify(*_args, **_kwargs):
        notified.append(1)
        return 1

    monkeypatch.setattr(
        "app.intelligence.generation.generate_analysis", _fake_generate,
    )
    monkeypatch.setattr(
        "app.domains.auditor.posture_summary.build_posture_snapshot",
        _fake_snapshot,
    )
    monkeypatch.setattr(
        "app.domains.auditor.posture_summary.notify_posture_summary",
        _fake_notify,
    )

    result = await handler.run_compliance_posture_summary(
        db_session, {"month": "2026-05"},
    )
    assert result["cache_reused"] is True
    assert result["notified"] == 0
    assert notified == []


@pytest.mark.asyncio
async def test_job_registered_in_both_processes():
    from pathlib import Path

    from app.scheduled_jobs.registry import get_handler

    assert get_handler("auditor.compliance_posture_summary") is not None
    root = Path(__file__).resolve().parents[2] / "app"
    for process in ("main.py", "worker.py"):
        assert "posture_summary_handler" in (
            root / process
        ).read_text(encoding="utf-8"), process


@pytest.mark.asyncio
async def test_never_evaluated_controls_are_not_filed_as_failing(db_session):
    """"Nobody has looked" is not "the claim is wrong".

    The outer join that fixed the ratio also made ``status`` None for
    never-evaluated controls, and ``None != "pass"`` quietly filed them
    as failing refs — so the fact package would tell the model eight
    controls were failing while ``controls_by_status`` in the same dict
    called them ``not_evaluated``.
    """
    control = ComplianceControl(
        key=f"CTLX_{uuid.uuid4().hex[:4]}", name="Never run",
        category="access", signal_type="detective", enabled=True,
    )
    db_session.add(control)
    await db_session.flush()

    _health, facts, _ = await build_posture_facts(
        db_session, period_start="2026-05-01", period_end="2026-05-31",
    )
    assert facts["controls_total"] == 1
    assert {k.removeprefix("controls_status_"): v for k, v in facts.items() if k.startswith("controls_status_")}.get("not_evaluated") == 1
    assert facts["failing_control_count"] == 0, (
        "a never-evaluated control must contribute no failing ref"
    )


@pytest.mark.asyncio
async def test_never_evaluated_controls_still_lower_the_ratio(db_session):
    """The other half: they must not be silently excluded either.

    Ten enabled controls of which only two have ever run reported a 100%
    pass ratio and grade A before the outer join.
    """
    await _seed_controls(db_session, passing=2, failing=0)
    for index in range(8):
        db_session.add(ComplianceControl(
            key=f"CTLN{index}_{uuid.uuid4().hex[:4]}", name="Never run",
            category="access", signal_type="detective", enabled=True,
        ))
    await db_session.flush()

    health, facts, _ = await build_posture_facts(
        db_session, period_start="2026-05-01", period_end="2026-05-31",
    )
    assert facts["controls_total"] == 10
    assert facts["control_pass_ratio"] == pytest.approx(0.2)
    assert health["grade"] == "D"


@pytest.mark.asyncio
async def test_the_scored_view_is_citable_as_a_fact(db_session):
    """The model is shown the score, so it cites it.

    ``overall_score`` used to live only in ``health``, so the natural
    citation ``/facts/overall_score`` resolved nowhere and the grounding
    check refused the entire narrative with "Unknown fact reference" —
    the monthly summary could never be produced in production. These
    values derive from ``control_pass_ratio``, which is already a fact;
    publishing the derivation makes it citable rather than inventable.
    """
    from app.domains.auditor.posture_summary import build_posture_facts

    health, facts, _refs = await build_posture_facts(
        db_session, period_start="2026-07-01", period_end="2026-07-31",
    )
    for key in ("overall_score", "confidence_score", "grade"):
        assert key in facts, f"{key} must be citable as /facts/{key}"
        assert facts[key] == health[key], (
            f"{key} must not be able to disagree between facts and health"
        )


@pytest.mark.asyncio
async def test_a_narrative_citing_the_score_passes_grounding(db_session):
    """End-to-end on the exact shape that was failing in production."""
    from app.domains.auditor.posture_summary import build_posture_facts
    from app.intelligence.contracts import validate_intelligence_output

    _health, facts, _refs = await build_posture_facts(
        db_session, period_start="2026-07-01", period_end="2026-07-31",
    )
    payload = {
        "summary_zh": f"2026 年 7 月综合得分 {facts['overall_score']}。",
        "summary_en": f"July 2026 overall score {facts['overall_score']}.",
        "summary_fact_refs": ["/facts/overall_score"],
        "conclusions": [], "risks": [], "opportunities": [],
        "forecasts": [], "recommendations": [], "data_gaps": [],
    }
    validate_intelligence_output(
        payload, facts={"facts": facts},
        allowed_evidence_refs=set(), confidence_score=90,
    )


@pytest.mark.asyncio
async def test_everything_the_model_can_see_is_citable(db_session):
    """The class fix, not another single pointer.

    Production failed twice in a row on this shape: first
    ``/facts/overall_score``, then ``/facts/decision_backlog``. The model
    cites what it is shown, using the label it is shown, so any value
    that reaches it must resolve as a fact. Fixing them one at a time is
    whack-a-mole; this asserts the whole scored breakdown is citable and
    agrees with what the read layer renders.
    """
    from app.domains.auditor.posture_summary import build_posture_facts
    from app.intelligence.contracts import validate_intelligence_output

    health, facts, _refs = await build_posture_facts(
        db_session, period_start="2026-07-01", period_end="2026-07-31",
    )
    for dim in health["dimensions"]:
        key = dim["key"]
        assert key in facts, f"/facts/{key} must resolve — the model cites it"
        assert facts[key] == dim["score"], (
            f"{key} must not differ between the fact package and health"
        )

    # Every scalar fact must be usable as a citation, since the model
    # sees the whole package.
    scalars = [k for k, v in facts.items() if isinstance(v, (int, float, str))]
    payload = {
        "summary_zh": "月度态势。", "summary_en": "Monthly posture.",
        "summary_fact_refs": [f"/facts/{k}" for k in scalars],
        "conclusions": [], "risks": [], "opportunities": [],
        "forecasts": [], "recommendations": [], "data_gaps": [],
    }
    validate_intelligence_output(
        payload, facts={"facts": facts},
        allowed_evidence_refs=set(), confidence_score=90,
    )


@pytest.mark.asyncio
async def test_the_ratio_is_also_published_as_a_percentage(db_session):
    """A narrator writes 45.45%, not 0.4545.

    The grounding guard compares literal number tokens and understands no
    arithmetic, so a model handed the ratio and writing the obvious
    percentage was refused for a number it derived correctly.
    """
    from app.domains.auditor.posture_summary import build_posture_facts
    from app.intelligence.contracts import validate_intelligence_output

    _health, facts, _refs = await build_posture_facts(
        db_session, period_start="2026-07-01", period_end="2026-07-31",
    )
    assert facts["control_pass_percent"] == round(
        facts["control_pass_ratio"] * 100, 2,
    )
    pct = facts["control_pass_percent"]
    validate_intelligence_output(
        {
            "summary_zh": f"控制项通过率 {pct}%。",
            "summary_en": f"Control pass rate {pct}%.",
            "summary_fact_refs": ["/facts/control_pass_percent"],
            "conclusions": [], "risks": [], "opportunities": [],
            "forecasts": [], "recommendations": [], "data_gaps": [],
        },
        facts={"facts": facts}, allowed_evidence_refs=set(),
        confidence_score=90,
    )


@pytest.mark.asyncio
async def test_both_sides_of_the_ratio_are_facts(db_session):
    """A posture letter reports what FAILS, not only what passes.

    Production showed the model writing the complements — 54.55% and a
    failing count — which no fact carried, so figures that were
    arithmetically right and editorially obvious were refused. Rather
    than forbid the sentence, publish the number: how many controls are
    failing is a fact an auditor wants in its own right.
    """
    from app.domains.auditor.posture_summary import build_posture_facts

    _health, facts, _refs = await build_posture_facts(
        db_session, period_start="2026-07-01", period_end="2026-07-31",
    )
    for key in ("controls_failing", "control_fail_percent",
                "controls_not_evaluated"):
        assert key in facts, key
    # The two sides must add up, or the letter contradicts itself.
    if facts["controls_total"]:
        assert (
            facts["control_pass_percent"] + facts["control_fail_percent"]
            <= 100.01
        ), "pass% + fail% cannot exceed 100 (the rest is not-evaluated)"
        assert facts["controls_passing"] + facts["controls_failing"] <= (
            facts["controls_total"]
        )


@pytest.mark.asyncio
async def test_every_fact_is_a_citable_scalar(db_session):
    """Prevent the class, not the instance.

    The model cites facts by JSON Pointer, and the contract refuses a
    pointer landing on a dict or a list. Production refused an entire
    narrative because this package offered ``failing_control_refs`` — a
    LIST — which the model then cited. Any nested value here is the same
    trap waiting for the next run, so the package must be flat.

    Anonymous refs still reach the model, on the ``reference_map``
    channel that ``allowed_evidence_refs`` is built from — which is where
    they belong.
    """
    from app.domains.auditor.posture_summary import build_posture_facts

    _health, facts, reference_map = await build_posture_facts(
        db_session, period_start="2026-07-01", period_end="2026-07-31",
    )
    nested = {
        k: type(v).__name__
        for k, v in facts.items()
        if isinstance(v, (dict, list, tuple, set))
    }
    assert not nested, f"facts must be flat scalars; found {nested}"

    # Every fact must therefore be usable as a citation.
    from app.intelligence.contracts import validate_intelligence_output

    validate_intelligence_output(
        {
            "summary_zh": "月度态势。", "summary_en": "Monthly posture.",
            "summary_fact_refs": [f"/facts/{k}" for k in facts],
            "conclusions": [], "risks": [], "opportunities": [],
            "forecasts": [], "recommendations": [], "data_gaps": [],
        },
        facts={"facts": facts}, allowed_evidence_refs=set(reference_map),
        confidence_score=90,
    )