"""Half-yearly access self-attestation (员工访问自查) — engine + self-service.

Covers: the full-footprint snapshot across all five access types
(reviewer = holder, left employees excluded), the holder batch keep/remove
path (keep = attest, remove → the type's existing self-service removal),
the ownership guard, the decide_item rejection of self_attestation items,
and campaign open idempotency.
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime, timedelta
from decimal import Decimal

import pytest

from app.domains.apikey.models import ApiKeyAssignment, ApiKeyProvider
from app.domains.asset.models import (
    AssetSoftwareCatalog,
    SoftwareLicensePlan,
    SoftwareSeatAllocation,
)
from app.domains.auditor import access_certification_service as acs
from app.domains.auditor.models import AccessCertificationItem
from app.domains.cmdb.models import CmdbCloudAccessGrant, CmdbCloudScope
from app.domains.permission.models import SoftwarePermission, UserPermission
from app.domains.sso.models import SsoGrant
from app.identity.person_models import IdentityPerson


def _person(status: str = "active", **kw) -> IdentityPerson:
    return IdentityPerson(
        id=uuid.uuid4(),
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
        work_email=kw.get("email", f"u-{uuid.uuid4().hex[:6]}@x.com"),
        display_name=kw.get("name", "Person"),
        employment_status=status,
    )


async def _software(db) -> AssetSoftwareCatalog:
    cat = AssetSoftwareCatalog(id=uuid.uuid4(), name=f"App-{uuid.uuid4().hex[:4]}")
    db.add(cat)
    await db.flush()
    return cat


async def _seat(db, person_id) -> SoftwareSeatAllocation:
    cat = await _software(db)
    plan = SoftwareLicensePlan(
        id=uuid.uuid4(), software_id=cat.id, name="Standard",
        price=Decimal("10.00"), billing_cycle="monthly",
    )
    db.add(plan)
    await db.flush()
    seat = SoftwareSeatAllocation(
        id=uuid.uuid4(), plan_id=plan.id, user_id=person_id, status="active",
    )
    db.add(seat)
    await db.flush()
    return seat


async def _permission(db, person_id) -> UserPermission:
    cat = await _software(db)
    perm = SoftwarePermission(
        id=uuid.uuid4(), software_id=cat.id, permission_key="admin",
        display_name="Admin", is_active=True,
    )
    db.add(perm)
    await db.flush()
    up = UserPermission(
        id=uuid.uuid4(), user_id=person_id, permission_id=perm.id,
        status="active",
    )
    db.add(up)
    await db.flush()
    return up


async def _cloud_grant(db, person_id) -> CmdbCloudAccessGrant:
    from app.integrations.models import Integration

    integ = Integration(
        id=str(uuid.uuid4()), kind="cloud", provider="aws",
        name="AWS", slug=f"aws-{uuid.uuid4().hex[:6]}",
    )
    db.add(integ)
    await db.flush()
    scope = CmdbCloudScope(
        id=uuid.uuid4(), integration_id=integ.id,
        scope_type="aws_account", external_id=f"acc-{uuid.uuid4().hex[:6]}",
        display_name="Prod AWS",
    )
    db.add(scope)
    await db.flush()
    g = CmdbCloudAccessGrant(
        id=uuid.uuid4(), scope_id=str(scope.id), person_id=str(person_id),
        role_key="viewer", role_display_name="Viewer", status="active",
    )
    db.add(g)
    await db.flush()
    return g


async def _sso(db, person_id) -> SsoGrant:
    g = SsoGrant(
        id=str(uuid.uuid4()), person_id=str(person_id), sso_provider="google",
        system_name="Authing Admin", system_owner_label="IT", status="approved",
    )
    db.add(g)
    await db.flush()
    return g


async def _api_key(db, person_id) -> ApiKeyAssignment:
    prov = ApiKeyProvider(id=uuid.uuid4(), name="Bailian")
    db.add(prov)
    await db.flush()
    ak = ApiKeyAssignment(
        id=uuid.uuid4(), provider_id=prov.id, assigned_to=person_id,
        request_number=f"AK-{uuid.uuid4().hex[:6]}", project_name="proj",
        reason="r", monthly_budget_usd=Decimal("0"), status="active",
    )
    db.add(ak)
    await db.flush()
    return ak


async def _open(db, *, due_days=21):
    return await acs.open_self_attestation_campaign(
        db, period_year=2026, half="H1", due_after_days=due_days,
    )


# ── snapshot ──────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_snapshot_full_footprint_holder_is_reviewer(db_session):
    active = _person(name="Active")
    left = _person(status="left", name="Left")
    db_session.add_all([active, left])
    await db_session.flush()
    await _seat(db_session, active.id)
    await _cloud_grant(db_session, active.id)
    await _permission(db_session, active.id)
    await _sso(db_session, active.id)
    await _api_key(db_session, active.id)
    # the left employee holds a seat too — must be excluded.
    await _seat(db_session, left.id)
    await db_session.commit()

    res = await _open(db_session)
    await db_session.commit()

    items = (await db_session.execute(
        __import__("sqlalchemy").select(AccessCertificationItem).where(
            AccessCertificationItem.certification_id
            == uuid.UUID(res["certification_id"]),
        )
    )).scalars().all()
    # 5 entitlements for the active holder, none for the leaver.
    assert {it.access_type for it in items} == {
        "software_seat", "cloud_grant", "user_permission",
        "sso_grant", "api_key",
    }
    assert len(items) == 5
    for it in items:
        assert it.person_id == active.id
        # holder attests their own access
        assert it.assigned_reviewer_person_id == active.id
        raw = it.raw_payload or {}
        # api keys are attest-only; the rest are removable
        assert raw["removable"] is (it.access_type != "api_key")


@pytest.mark.asyncio
async def test_open_is_idempotent(db_session):
    p = _person()
    db_session.add(p)
    await db_session.flush()
    await _seat(db_session, p.id)
    await db_session.commit()

    first = await _open(db_session)
    await db_session.commit()
    again = await _open(db_session)
    # Re-opening an OPEN campaign is now a top-up rather than a flat skip
    # (#2128): the fan-out moved outside the Redis lock so it can no
    # longer brown-out logins or outrun the lock's TTL, and a crash
    # part-way through it would otherwise strand the un-carded holders —
    # re-opening is refused by the existence check, and the outbox retry
    # worker only backstops rows that already exist.
    #
    # The invariants that actually matter are unchanged and asserted
    # here: ONE campaign, and NOBODY gets a second card.
    assert again["certification_id"] == first["certification_id"]
    assert again["topped_up"] is True
    assert again["notified"] == 0, "a holder was carded twice"


# ── holder batch decision ───────────────────────────────────────────────


@pytest.mark.asyncio
async def test_submit_keep_and_remove_seat(db_session):
    p = _person()
    db_session.add(p)
    await db_session.flush()
    seat = await _seat(db_session, p.id)
    await _permission(db_session, p.id)
    await db_session.commit()
    await _open(db_session)
    await db_session.commit()

    view = await acs.my_open_self_attestation(db_session, person_id=p.id)
    assert view["open"] is True
    assert view["pending_count"] == 2
    by_type = {i["access_type"]: i for i in view["items"]}
    seat_item_id = uuid.UUID(by_type["software_seat"]["id"])
    perm_item_id = uuid.UUID(by_type["user_permission"]["id"])

    res = await acs.submit_self_attestation(
        db_session, actor_person_id=p.id,
        keep_item_ids=[perm_item_id], remove_item_ids=[seat_item_id],
        remove_reason="不再需要",
    )
    await db_session.commit()
    assert res == {"kept": 1, "removed": 1, "skipped": 0, "failed": 0}

    # seat scheduled for cancel-at-period-end (not hard-revoked)
    await db_session.refresh(seat)
    assert seat.status == "active"
    assert seat.cancel_at_period_end is not None
    assert seat.cancel_requested_by == p.id

    seat_item = await db_session.get(AccessCertificationItem, seat_item_id)
    assert seat_item.decision == "revoke"
    assert seat_item.revocation_status == "scheduled"
    perm_item = await db_session.get(AccessCertificationItem, perm_item_id)
    assert perm_item.decision == "keep"


@pytest.mark.asyncio
async def test_submit_failed_removal_isolated_per_item(db_session, monkeypatch):
    """A removal that blows up rolls back ONLY that item (per-item savepoint
    + expire): it stays pending and uncounted, while the rest of the batch
    commits cleanly. Guards the split-brain / count-drift the unisolated
    loop risked."""
    p = _person()
    db_session.add(p)
    await db_session.flush()
    await _seat(db_session, p.id)
    await _seat(db_session, p.id)
    await db_session.commit()
    await _open(db_session)
    await db_session.commit()

    view = await acs.my_open_self_attestation(db_session, person_id=p.id)
    item_ids = [uuid.UUID(i["id"]) for i in view["items"]]
    boom_id = item_ids[0]

    import app.core.coordinators.access_certification_coordinator as coord
    real = coord.enqueue_self_service_removal_for_cert_item

    async def flaky(db, *, item, **kw):
        if item.id == boom_id:
            raise RuntimeError("simulated downstream removal failure")
        return await real(db, item=item, **kw)

    monkeypatch.setattr(
        coord, "enqueue_self_service_removal_for_cert_item", flaky,
    )

    res = await acs.submit_self_attestation(
        db_session, actor_person_id=p.id,
        keep_item_ids=[], remove_item_ids=item_ids,
    )
    await db_session.commit()
    assert res["failed"] == 1
    assert res["removed"] == 1
    # the failed item stayed pending; the other was decided + counted.
    boom = await db_session.get(AccessCertificationItem, boom_id)
    assert boom.decision == "pending"
    assert boom.revocation_status == "none"
    other = await db_session.get(AccessCertificationItem, item_ids[1])
    assert other.decision == "revoke"
    summary = await acs.build_self_attestation_summary(
        db_session, certification_id=boom.certification_id)
    # exactly one decided/revoked — no count drift from the rolled-back item.
    assert summary["decided"] == 1
    assert summary["by_decision"]["revoke"] == 1
    assert summary["by_decision"]["pending"] == 1


@pytest.mark.asyncio
async def test_submit_idempotent_skips_decided(db_session):
    p = _person()
    db_session.add(p)
    await db_session.flush()
    await _seat(db_session, p.id)
    await db_session.commit()
    await _open(db_session)
    await db_session.commit()
    view = await acs.my_open_self_attestation(db_session, person_id=p.id)
    item_id = uuid.UUID(view["items"][0]["id"])

    await acs.submit_self_attestation(
        db_session, actor_person_id=p.id, keep_item_ids=[item_id],
        remove_item_ids=[],
    )
    await db_session.commit()
    # re-submitting the same (already-decided) item is a no-op skip
    res = await acs.submit_self_attestation(
        db_session, actor_person_id=p.id, keep_item_ids=[item_id],
        remove_item_ids=[],
    )
    assert res == {"kept": 0, "removed": 0, "skipped": 1, "failed": 0}


@pytest.mark.asyncio
async def test_cannot_decide_another_persons_item(db_session):
    a = _person(name="A")
    b = _person(name="B")
    db_session.add_all([a, b])
    await db_session.flush()
    await _seat(db_session, a.id)
    await db_session.commit()
    await _open(db_session)
    await db_session.commit()
    view = await acs.my_open_self_attestation(db_session, person_id=a.id)
    a_item = uuid.UUID(view["items"][0]["id"])

    with pytest.raises(Exception) as ei:
        await acs.submit_self_attestation(
            db_session, actor_person_id=b.id, keep_item_ids=[a_item],
            remove_item_ids=[],
        )
    assert getattr(ei.value, "status_code", None) == 403


@pytest.mark.asyncio
async def test_decide_item_rejects_self_attestation(db_session):
    p = _person()
    db_session.add(p)
    await db_session.flush()
    await _seat(db_session, p.id)
    await db_session.commit()
    await _open(db_session)
    await db_session.commit()
    view = await acs.my_open_self_attestation(db_session, person_id=p.id)
    item_id = uuid.UUID(view["items"][0]["id"])

    with pytest.raises(Exception) as ei:
        await acs.decide_item(
            db_session, item_id=item_id, decision="keep", note=None,
            actor_person_id=p.id,
        )
    assert ei.value.code == "SELF_ATTESTATION_ITEM"


@pytest.mark.asyncio
async def test_no_open_campaign_returns_closed(db_session):
    p = _person()
    db_session.add(p)
    await db_session.flush()
    await db_session.commit()
    view = await acs.my_open_self_attestation(db_session, person_id=p.id)
    assert view == {"open": False, "campaign": None, "items": [],
                    "pending_count": 0, "total_count": 0}


@pytest.mark.asyncio
async def test_auto_open_outside_window_is_noop(db_session):
    # Forcing window_days small; the test runs on an arbitrary date — the
    # function must never raise and must return a structured result.
    res = await acs.auto_open_semiannual_self_attestation(
        db_session, window_days=1,
    )
    assert "skipped_reason" in res or "certification_id" in res


# ── PR3: governance summary / team / export / GRC ──────────────────────


@pytest.mark.asyncio
async def test_summary_coverage_and_flagged(db_session):
    p = _person()
    db_session.add(p)
    await db_session.flush()
    await _seat(db_session, p.id)
    await _permission(db_session, p.id)
    await db_session.commit()
    res = await _open(db_session)
    await db_session.commit()
    cid = uuid.UUID(res["certification_id"])

    view = await acs.my_open_self_attestation(db_session, person_id=p.id)
    by_type = {i["access_type"]: i for i in view["items"]}
    seat_item = uuid.UUID(by_type["software_seat"]["id"])
    perm_item = uuid.UUID(by_type["user_permission"]["id"])
    # keep the permission, flag the seat for removal.
    await acs.submit_self_attestation(
        db_session, actor_person_id=p.id,
        keep_item_ids=[perm_item], remove_item_ids=[seat_item],
    )
    await db_session.commit()

    summary = await acs.build_self_attestation_summary(
        db_session, certification_id=cid)
    assert summary is not None
    assert summary["total_items"] == 2
    assert summary["decided"] == 2
    assert summary["coverage_pct"] == 100.0
    assert summary["by_decision"]["keep"] == 1
    assert summary["by_decision"]["revoke"] == 1
    assert len(summary["flagged_removals"]) == 1
    assert summary["flagged_removals"][0]["access_type"] == "software_seat"
    # seat removal is scheduled (cancel-at-period-end)
    assert summary["flagged_disposition"].get("scheduled") == 1
    assert summary["non_responders"] == []


@pytest.mark.asyncio
async def test_summary_non_responders(db_session):
    p = _person(name="Slacker")
    db_session.add(p)
    await db_session.flush()
    await _seat(db_session, p.id)
    await db_session.commit()
    res = await _open(db_session)
    await db_session.commit()
    summary = await acs.build_self_attestation_summary(
        db_session, certification_id=uuid.UUID(res["certification_id"]))
    assert summary["decided"] == 0
    assert summary["coverage_pct"] == 0.0
    assert len(summary["non_responders"]) == 1
    assert summary["non_responders"][0]["pending"] == 1


@pytest.mark.asyncio
async def test_summary_none_for_wrong_type(db_session):
    # A non-self_attestation campaign id returns None.
    cert = await acs.create_campaign(
        db_session, label="Q test 特权", campaign_type="privileged_role",
        scope_config={}, due_at=datetime.now(UTC) + timedelta(days=30),
        actor_person_id=None,
    )
    await db_session.commit()
    assert await acs.build_self_attestation_summary(
        db_session, certification_id=cert.id) is None


@pytest.mark.asyncio
async def test_team_attestation_status(db_session):
    mgr = _person(name="Mgr")
    db_session.add(mgr)
    await db_session.flush()
    r1 = _person(name="R1")
    r2 = _person(name="R2")
    r1.manager_person_id = mgr.id
    r2.manager_person_id = mgr.id
    db_session.add_all([r1, r2])
    await db_session.flush()
    await _seat(db_session, r1.id)
    await _seat(db_session, r2.id)
    await db_session.commit()
    await _open(db_session)
    await db_session.commit()

    # r1 confirms; r2 doesn't.
    v1 = await acs.my_open_self_attestation(db_session, person_id=r1.id)
    await acs.submit_self_attestation(
        db_session, actor_person_id=r1.id,
        keep_item_ids=[uuid.UUID(v1["items"][0]["id"])], remove_item_ids=[],
    )
    await db_session.commit()

    team = await acs.build_team_attestation_status(
        db_session, person_ids=[r1.id, r2.id])
    assert team["open"] is True
    rows = {r["person_id"]: r for r in team["rows"]}
    assert rows[str(r1.id)]["pending"] == 0
    assert rows[str(r2.id)]["pending"] == 1


@pytest.mark.asyncio
async def test_export_docx_and_pdf(db_session):
    from app.domains.auditor import access_review_export as exporter

    p = _person()
    db_session.add(p)
    await db_session.flush()
    await _seat(db_session, p.id)
    await db_session.commit()
    res = await _open(db_session)
    await db_session.commit()
    cid = uuid.UUID(res["certification_id"])

    docx = await exporter.build_docx(db_session, cid)
    assert docx is not None
    buf, stem = docx
    assert buf.getbuffer().nbytes > 0
    assert stem.startswith("access_self_attestation")

    pdf = await exporter.build_pdf(db_session, cid)
    assert pdf is not None
    assert pdf[0].getbuffer().nbytes > 0


@pytest.mark.asyncio
async def test_complete_self_attestation_closes_over_nonresponders(db_session):
    """Round-3 fix (#5): a self-attestation campaign snapshots every active
    employee; non-responders always leave items ``pending`` and an auditor
    cannot disposition them (decide_item refuses self-attestation items).
    ``complete_campaign`` must close over non-responders by stamping the
    remaining pending items to the terminal ``no_response`` disposition, so
    the half-yearly all-employee campaign can actually reach ``completed``
    (previously it blocked forever on a single non-responder)."""
    from sqlalchemy import select

    p = _person(name="NonResponder")
    db_session.add(p)
    await db_session.flush()
    await _seat(db_session, p.id)
    await db_session.commit()

    res = await _open(db_session)
    await db_session.commit()
    cert_id = uuid.UUID(res["certification_id"])

    # Holder never responds → item stays pending. Closing must still succeed.
    cert = await acs.complete_campaign(
        db_session, certification_id=cert_id, actor_person_id=p.id,
    )
    await db_session.commit()
    assert cert.status == acs.STATUS_COMPLETED

    items = (await db_session.execute(
        select(AccessCertificationItem).where(
            AccessCertificationItem.certification_id == cert_id,
        )
    )).scalars().all()
    assert items
    assert all(it.decision == acs.DECISION_NO_RESPONSE for it in items)