"""Leaver cloud-IAM residue detection + governance queue.

The invariant: a person who has left must hold no cloud-side IAM policy
or group membership, **whatever Helix's own grant rows claim**. Prod
2026-08-12 (Tencent 100002610019) is the case these tests encode —
褚杰 ``chujie`` left 2026-06-01, Helix revoked all four CAM grants that
night, the driver's per-policy detach was a no-op against
group-inherited policies, and 2.5 months later the sub-user still had
every one of them. No existing sweep could see it: they all filter on
``active`` / ``pending`` grants, and his were ``revoked``.
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime, timedelta
from unittest.mock import AsyncMock, patch

import pytest
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.cmdb.leaver_iam_residue import (
    detect_leaver_iam_residue,
    reconcile_residue_rows,
    summarize_residue,
)
from app.domains.cmdb.models import (
    RESIDUE_STATUS_DISMISSED,
    RESIDUE_STATUS_OPEN,
    RESIDUE_STATUS_RESOLVED,
    CmdbIamUser,
    CmdbLeaverIamResidue,
)
from app.identity.person_models import IdentityPerson
from app.integrations.models import Integration


async def _person(
    db: AsyncSession,
    email: str,
    *,
    status: str = "left",
    termination_days_ago: int | None = 30,
) -> IdentityPerson:
    term = None
    if termination_days_ago is not None:
        term = (datetime.now(UTC) - timedelta(days=termination_days_ago)).date()
    p = IdentityPerson(
        id=uuid.uuid4(),
        work_email=email,
        display_name=email.split("@")[0],
        employment_status=status,
        termination_date=term,
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
    )
    db.add(p)
    await db.commit()
    return p


async def _integration(db: AsyncSession) -> Integration:
    row = Integration(
        id=uuid.uuid4(),
        kind="cloud",
        provider="tencent",
        name="腾讯云",
        slug=f"tencent-{uuid.uuid4().hex[:6]}",
        enabled=True,
        is_primary=False,
        capabilities=["iam"],
    )
    db.add(row)
    await db.commit()
    return row


async def _iam_user(
    db: AsyncSession,
    integration: Integration,
    *,
    user_name: str,
    policies: list[dict] | None = None,
    groups: list[str] | None = None,
    owner_person_id: str | None = None,
    synced_hours_ago: float = 1.0,
) -> CmdbIamUser:
    now = datetime.now(UTC).replace(tzinfo=None)
    row = CmdbIamUser(
        id=uuid.uuid4(),
        integration_id=str(integration.id),
        user_name=user_name,
        sub_account_id="",
        attached_policies=policies or [],
        groups=groups or [],
        mfa_enabled=False,
        tags={},
        owner_person_id=owner_person_id,
        raw_data={},
        last_synced_at=now - timedelta(hours=synced_hours_ago),
    )
    db.add(row)
    await db.commit()
    return row


@pytest.mark.asyncio
async def test_detects_group_inherited_residue_on_a_leaver(
    db_session: AsyncSession,
) -> None:
    """The chujie case: grants long since revoked in Helix, policies and
    group membership still live on the cloud."""
    person = await _person(db_session, "chujie@pingcap.com")
    integration = await _integration(db_session)
    await _iam_user(
        db_session, integration,
        user_name="chujie",
        policies=[
            {"name": "AdministratorAccess", "arn": ""},
            {"name": "QcloudCollMFAManageAccess", "arn": ""},
        ],
        groups=["管理员"],
        owner_person_id=str(person.id),
    )

    findings = await detect_leaver_iam_residue(db_session)

    assert len(findings) == 1
    f = findings[0]
    assert f.person_id == str(person.id)
    assert f.user_name == "chujie"
    assert f.policy_count == 2
    assert f.group_count == 1
    assert f.match_basis == "owner"
    assert f.mirror_stale is False


@pytest.mark.asyncio
async def test_group_membership_alone_is_residue(
    db_session: AsyncSession,
) -> None:
    """Group membership is *why* the original revoke silently failed —
    the permissions were inherited. An empty policy list with a live
    group is still access."""
    person = await _person(db_session, "lifu.wu@pingcap.com")
    integration = await _integration(db_session)
    await _iam_user(
        db_session, integration,
        user_name="lifu.wu@pingcap.com",
        policies=[],
        groups=["账单计费"],
        owner_person_id=str(person.id),
    )

    findings = await detect_leaver_iam_residue(db_session)

    assert len(findings) == 1
    assert findings[0].policy_count == 0
    assert findings[0].group_count == 1


@pytest.mark.asyncio
async def test_active_employee_and_clean_leaver_are_not_residue(
    db_session: AsyncSession,
) -> None:
    active = await _person(
        db_session, "wuhui.zuo@pingcap.com",
        status="active", termination_days_ago=None,
    )
    clean_leaver = await _person(db_session, "gone.clean@pingcap.com")
    integration = await _integration(db_session)
    await _iam_user(
        db_session, integration,
        user_name="wuhui.zuo@pingcap.com",
        policies=[{"name": "AdministratorAccess", "arn": ""}],
        owner_person_id=str(active.id),
    )
    await _iam_user(
        db_session, integration,
        user_name="gone.clean@pingcap.com",
        policies=[],
        groups=[],
        owner_person_id=str(clean_leaver.id),
    )

    assert await detect_leaver_iam_residue(db_session) == []


@pytest.mark.asyncio
async def test_aws_self_service_policies_are_not_residue(
    db_session: AsyncSession,
) -> None:
    """AWS attaches these to every console user. Counting them produced
    60+ false alerts per leaver in the grants-derivation path before it
    filtered them; this queue must not repeat that."""
    person = await _person(db_session, "someone@pingcap.com")
    integration = await _integration(db_session)
    await _iam_user(
        db_session, integration,
        user_name="someone@pingcap.com",
        policies=[
            {"name": "IAMUserChangePassword",
             "arn": "arn:aws:iam::aws:policy/IAMUserChangePassword"},
        ],
        owner_person_id=str(person.id),
    )

    assert await detect_leaver_iam_residue(db_session) == []


@pytest.mark.asyncio
async def test_unbound_iam_row_matches_the_leaver_by_email(
    db_session: AsyncSession,
) -> None:
    """The owner-resolver only binds ACTIVE people, so a row whose owner
    left before it was ever bound keeps ``owner_person_id = NULL``
    forever. That population is exactly the one this sweep cannot
    afford to miss."""
    person = await _person(db_session, "tangpeng@pingcap.com")
    integration = await _integration(db_session)
    await _iam_user(
        db_session, integration,
        user_name="tangpeng@pingcap.com",
        policies=[{"name": "QcloudCOSFullAccess", "arn": ""}],
        owner_person_id=None,
    )

    findings = await detect_leaver_iam_residue(db_session)

    assert len(findings) == 1
    assert findings[0].person_id == str(person.id)
    assert findings[0].match_basis == "email"


@pytest.mark.asyncio
async def test_iam_row_owned_by_an_active_person_is_never_email_matched(
    db_session: AsyncSession,
) -> None:
    """A row already bound to somebody still employed must not be dragged
    into the queue by a same-address leaver — the owner FK wins."""
    active = await _person(
        db_session, "shared@pingcap.com",
        status="active", termination_days_ago=None,
    )
    integration = await _integration(db_session)
    await _iam_user(
        db_session, integration,
        user_name="left.person@pingcap.com",
        policies=[{"name": "AdministratorAccess", "arn": ""}],
        owner_person_id=str(active.id),
    )
    await _person(db_session, "left.person@pingcap.com")

    assert await detect_leaver_iam_residue(db_session) == []


@pytest.mark.asyncio
async def test_leaver_pending_before_last_working_day_is_not_residue(
    db_session: AsyncSession,
) -> None:
    """``leaver_pending`` means "scheduled to leave", not "gone". Acting
    before the last working day would revoke a colleague who is still at
    their desk — the #1735 / 2026-08-11 incident class."""
    person = await _person(
        db_session, "future.leaver@pingcap.com",
        status="leaver_pending", termination_days_ago=-5,
    )
    integration = await _integration(db_session)
    await _iam_user(
        db_session, integration,
        user_name="future.leaver@pingcap.com",
        policies=[{"name": "AdministratorAccess", "arn": ""}],
        owner_person_id=str(person.id),
    )

    assert await detect_leaver_iam_residue(db_session) == []


@pytest.mark.asyncio
async def test_leaver_pending_past_last_working_day_is_residue(
    db_session: AsyncSession,
) -> None:
    """The window ``cmdb.leaver_grant_sweep`` misses: HRIS can lag days
    behind the actual departure, and ``employment_status='left'`` alone
    would let that gap hide live access."""
    person = await _person(
        db_session, "past.leaver@pingcap.com",
        status="leaver_pending", termination_days_ago=3,
    )
    integration = await _integration(db_session)
    await _iam_user(
        db_session, integration,
        user_name="past.leaver@pingcap.com",
        policies=[{"name": "AdministratorAccess", "arn": ""}],
        owner_person_id=str(person.id),
    )

    findings = await detect_leaver_iam_residue(db_session)
    assert len(findings) == 1
    assert findings[0].person_id == str(person.id)


@pytest.mark.asyncio
async def test_stale_mirror_is_flagged_not_hidden(
    db_session: AsyncSession,
) -> None:
    """An IAM sync that stopped running is itself worth seeing — surface
    the row, but mark it so nobody deletes an account based on a
    snapshot we know is old."""
    person = await _person(db_session, "stale@pingcap.com")
    integration = await _integration(db_session)
    await _iam_user(
        db_session, integration,
        user_name="stale@pingcap.com",
        policies=[{"name": "AdministratorAccess", "arn": ""}],
        owner_person_id=str(person.id),
        synced_hours_ago=48,
    )

    findings = await detect_leaver_iam_residue(db_session)
    assert len(findings) == 1
    assert findings[0].mirror_stale is True


@pytest.mark.asyncio
async def test_reconcile_opens_refreshes_and_auto_resolves(
    db_session: AsyncSession,
) -> None:
    person = await _person(db_session, "cycle@pingcap.com")
    integration = await _integration(db_session)
    iam_row = await _iam_user(
        db_session, integration,
        user_name="cycle@pingcap.com",
        policies=[{"name": "AdministratorAccess", "arn": ""}],
        owner_person_id=str(person.id),
    )

    first = await reconcile_residue_rows(
        db_session, await detect_leaver_iam_residue(db_session),
    )
    assert first.opened == 1
    assert len(first.newly_opened_ids) == 1

    # Second run, residue unchanged: refresh, do NOT re-open (that is
    # what keeps the alert from firing every night).
    second = await reconcile_residue_rows(
        db_session, await detect_leaver_iam_residue(db_session),
    )
    assert second.opened == 0
    assert second.still_open == 1
    assert second.newly_opened_ids == []

    # Access actually removed → the row closes itself.
    iam_row.attached_policies = []
    iam_row.groups = []
    await db_session.commit()
    third = await reconcile_residue_rows(
        db_session, await detect_leaver_iam_residue(db_session),
    )
    assert third.auto_resolved == 1
    row = (await db_session.execute(
        select(CmdbLeaverIamResidue)
    )).scalars().one()
    assert row.status == RESIDUE_STATUS_RESOLVED
    assert row.resolution == "cleared"


@pytest.mark.asyncio
async def test_dismissed_row_is_not_reopened_but_still_auto_closes(
    db_session: AsyncSession,
) -> None:
    """A dismissal is a decision. Re-opening it every night is how a
    governance queue trains people to ignore it — but a dismissed row
    whose residue is genuinely gone must not linger forever either."""
    person = await _person(db_session, "dismissed@pingcap.com")
    integration = await _integration(db_session)
    iam_row = await _iam_user(
        db_session, integration,
        user_name="dismissed@pingcap.com",
        policies=[{"name": "AdministratorAccess", "arn": ""}],
        owner_person_id=str(person.id),
    )
    await reconcile_residue_rows(
        db_session, await detect_leaver_iam_residue(db_session),
    )
    row = (await db_session.execute(
        select(CmdbLeaverIamResidue)
    )).scalars().one()
    row.status = RESIDUE_STATUS_DISMISSED
    row.resolution_note = "转为团队共享账号"
    await db_session.commit()

    await reconcile_residue_rows(
        db_session, await detect_leaver_iam_residue(db_session),
    )
    await db_session.refresh(row)
    assert row.status == RESIDUE_STATUS_DISMISSED

    iam_row.attached_policies = []
    await db_session.commit()
    await reconcile_residue_rows(
        db_session, await detect_leaver_iam_residue(db_session),
    )
    await db_session.refresh(row)
    assert row.status == RESIDUE_STATUS_RESOLVED


@pytest.mark.asyncio
async def test_revoked_iam_row_is_out_of_scope(
    db_session: AsyncSession,
) -> None:
    """``revoked_at`` means an admin already ran the cleanup chain; the
    provider sync drops the row once the delete lands."""
    person = await _person(db_session, "cleaned@pingcap.com")
    integration = await _integration(db_session)
    row = await _iam_user(
        db_session, integration,
        user_name="cleaned@pingcap.com",
        policies=[{"name": "AdministratorAccess", "arn": ""}],
        owner_person_id=str(person.id),
    )
    row.revoked_at = datetime.now(UTC).replace(tzinfo=None)
    await db_session.commit()

    assert await detect_leaver_iam_residue(db_session) == []


@pytest.mark.asyncio
async def test_summary_counts_feed_the_hero_stats(
    db_session: AsyncSession,
) -> None:
    person = await _person(db_session, "stats@pingcap.com")
    integration = await _integration(db_session)
    await _iam_user(
        db_session, integration,
        user_name="stats@pingcap.com",
        policies=[{"name": "AdministratorAccess", "arn": ""}],
        owner_person_id=str(person.id),
    )
    await reconcile_residue_rows(
        db_session, await detect_leaver_iam_residue(db_session),
    )

    summary = await summarize_residue(db_session)
    assert summary["open_count"] == 1
    assert summary["people_count"] == 1
    assert summary["dismissed_count"] == 0


@pytest.mark.asyncio
async def test_http_queue_and_dismiss_flow(
    db_session: AsyncSession,
    client,
    admin_headers: dict[str, str],
) -> None:
    person = await _person(db_session, "http.case@pingcap.com")
    integration = await _integration(db_session)
    await _iam_user(
        db_session, integration,
        user_name="http.case@pingcap.com",
        policies=[{"name": "AdministratorAccess", "arn": ""}],
        groups=["管理员"],
        owner_person_id=str(person.id),
    )

    scan = await client.post(
        "/api/v1/cmdb/leaver-iam-residue/rescan", headers=admin_headers,
    )
    assert scan.status_code == 200
    assert scan.json()["data"]["opened"] == 1

    listing = await client.get(
        "/api/v1/cmdb/leaver-iam-residue?status=open", headers=admin_headers,
    )
    assert listing.status_code == 200
    payload = listing.json()
    rows = payload["data"]
    assert len(rows) == 1
    assert rows[0]["policy_names"] == ["AdministratorAccess"]
    assert rows[0]["group_names"] == ["管理员"]
    assert rows[0]["days_since_leaving"] == 30
    assert payload["meta"]["summary"]["open_count"] == 1

    residue_id = rows[0]["id"]
    dismissed = await client.post(
        f"/api/v1/cmdb/leaver-iam-residue/{residue_id}/dismiss",
        headers=admin_headers,
        json={"reason": "已转为团队共享账号，由 IT 接管"},
    )
    assert dismissed.status_code == 200
    assert dismissed.json()["data"]["status"] == RESIDUE_STATUS_DISMISSED

    # Dismissing twice is a no-op, not a 409 — two admins clicking the
    # same row is a normal race in a shared queue.
    again = await client.post(
        f"/api/v1/cmdb/leaver-iam-residue/{residue_id}/dismiss",
        headers=admin_headers,
        json={"reason": "已转为团队共享账号，由 IT 接管"},
    )
    assert again.status_code == 200

    open_only = await client.get(
        "/api/v1/cmdb/leaver-iam-residue?status=open", headers=admin_headers,
    )
    assert open_only.json()["data"] == []


@pytest.mark.asyncio
async def test_dismiss_requires_a_reason(
    db_session: AsyncSession,
    client,
    admin_headers: dict[str, str],
) -> None:
    """A dismissal with no stated reason is indistinguishable from
    someone clearing their inbox, and this queue is audit evidence."""
    person = await _person(db_session, "noreason@pingcap.com")
    integration = await _integration(db_session)
    await _iam_user(
        db_session, integration,
        user_name="noreason@pingcap.com",
        policies=[{"name": "AdministratorAccess", "arn": ""}],
        owner_person_id=str(person.id),
    )
    await reconcile_residue_rows(
        db_session, await detect_leaver_iam_residue(db_session),
    )
    row = (await db_session.execute(
        select(CmdbLeaverIamResidue)
    )).scalars().one()

    resp = await client.post(
        f"/api/v1/cmdb/leaver-iam-residue/{row.id}/dismiss",
        headers=admin_headers,
        json={"reason": "x"},
    )
    assert resp.status_code == 422
    await db_session.refresh(row)
    assert row.status == RESIDUE_STATUS_OPEN


@pytest.mark.asyncio
async def test_a_dismissed_row_reopens_when_the_access_grows(
    db_session: AsyncSession,
) -> None:
    """A dismissal is a judgement about the residue the admin was SHOWN.

    Dismissing a leaver's read-only account must not silently cover that
    same account later being granted AdministratorAccess — otherwise the
    dismissal is a permanent exemption nobody agreed to.
    """
    person = await _person(db_session, "dismissed.grows@pingcap.com")
    integration = await _integration(db_session)
    iam_row = await _iam_user(
        db_session, integration,
        user_name="dismissed.grows@pingcap.com",
        policies=[{"name": "QcloudCOSGetServiceAccess", "arn": ""}],
        owner_person_id=str(person.id),
    )
    await reconcile_residue_rows(
        db_session, await detect_leaver_iam_residue(db_session),
    )
    row = (await db_session.execute(
        select(CmdbLeaverIamResidue)
    )).scalars().one()
    row.status = RESIDUE_STATUS_DISMISSED
    row.resolution_note = "只读，交接期内保留"
    await db_session.commit()

    # Same residue on the next run: stays dismissed, no re-alert.
    quiet = await reconcile_residue_rows(
        db_session, await detect_leaver_iam_residue(db_session),
    )
    await db_session.refresh(row)
    assert row.status == RESIDUE_STATUS_DISMISSED
    assert quiet.reopened == 0
    assert quiet.newly_opened_ids == []

    # Now the account gains admin — that is a new fact.
    iam_row.attached_policies = [
        {"name": "QcloudCOSGetServiceAccess", "arn": ""},
        {"name": "AdministratorAccess", "arn": ""},
    ]
    await db_session.commit()
    grew = await reconcile_residue_rows(
        db_session, await detect_leaver_iam_residue(db_session),
    )
    await db_session.refresh(row)
    assert row.status == RESIDUE_STATUS_OPEN
    assert grew.reopened == 1
    # Re-alerting is the whole point of reopening.
    assert str(row.id) in grew.newly_opened_ids
    # The original reason survives — it is why somebody thought this was
    # acceptable, and that context matters on the second look.
    assert row.resolution_note == "只读，交接期内保留"


@pytest.mark.asyncio
async def test_reclassified_and_rehired_rows_do_not_claim_cleared(
    db_session: AsyncSession,
) -> None:
    """``cleared`` asserts the cloud-side access is gone. A row that left
    the queue because the account was declared a robot, or because the
    person came back, must not borrow that claim."""
    robot_owner = await _person(db_session, "robot.owner@pingcap.com")
    rehired = await _person(db_session, "boomerang@pingcap.com")
    integration = await _integration(db_session)
    robot = await _iam_user(
        db_session, integration, user_name="robot.owner@pingcap.com",
        policies=[{"name": "AdministratorAccess", "arn": ""}],
        owner_person_id=str(robot_owner.id),
    )
    await _iam_user(
        db_session, integration, user_name="boomerang@pingcap.com",
        policies=[{"name": "AdministratorAccess", "arn": ""}],
        owner_person_id=str(rehired.id),
    )
    await reconcile_residue_rows(
        db_session, await detect_leaver_iam_residue(db_session),
    )

    robot.account_kind = "service"
    rehired.employment_status = "active"
    await db_session.commit()

    await reconcile_residue_rows(
        db_session, await detect_leaver_iam_residue(db_session),
    )
    rows = {
        str(r.person_id): r for r in (await db_session.execute(
            select(CmdbLeaverIamResidue)
        )).scalars().all()
    }
    assert rows[str(robot_owner.id)].resolution == "reclassified"
    assert rows[str(rehired.id)].resolution == "not_a_leaver"
    # The policies are still attached in both cases.
    assert all(r.status == RESIDUE_STATUS_RESOLVED for r in rows.values())


@pytest.mark.asyncio
async def test_genuinely_removed_access_still_reports_cleared(
    db_session: AsyncSession,
) -> None:
    person = await _person(db_session, "reallygone@pingcap.com")
    integration = await _integration(db_session)
    iam_row = await _iam_user(
        db_session, integration, user_name="reallygone@pingcap.com",
        policies=[{"name": "AdministratorAccess", "arn": ""}],
        owner_person_id=str(person.id),
    )
    await reconcile_residue_rows(
        db_session, await detect_leaver_iam_residue(db_session),
    )
    iam_row.attached_policies = []
    await db_session.commit()
    await reconcile_residue_rows(
        db_session, await detect_leaver_iam_residue(db_session),
    )
    row = (await db_session.execute(
        select(CmdbLeaverIamResidue)
    )).scalars().one()
    assert row.resolution == "cleared"


@pytest.mark.asyncio
async def test_admin_alert_card_chrome_is_localized(
    db_session: AsyncSession,
) -> None:
    """Stat labels / subtitle / button do NOT go through the template
    layer, so a bare Chinese string reaches an English or Japanese admin
    untranslated next to a body that localized fine."""
    from app.domains.cmdb.leaver_iam_residue import _alert_admins
    from app.notifications.service import LocalizedText

    person = await _person(db_session, "carded@pingcap.com")
    integration = await _integration(db_session)
    await _iam_user(
        db_session, integration, user_name="carded@pingcap.com",
        policies=[{"name": "AdministratorAccess", "arn": ""}],
        owner_person_id=str(person.id),
    )
    findings = await detect_leaver_iam_residue(db_session)
    result = await reconcile_residue_rows(db_session, findings)

    sent: list = []

    class _Svc:
        def __init__(self, *_a, **_kw) -> None:
            pass

        async def dispatch(self, envelope, **_kw):
            sent.append(envelope)

    with (
        patch("app.notifications.service.NotificationService", _Svc),
        patch(
            "app.scheduled_jobs.failure_notifier.list_admin_recipients",
            new=AsyncMock(return_value=[
                await _person(db_session, "admin@pingcap.com", status="active"),
            ]),
        ),
    ):
        notified = await _alert_admins(
            db_session, findings, result.newly_opened_ids,
        )

    assert notified == 1
    env = sent[0]
    assert isinstance(env.card_subtitle, LocalizedText)
    assert all(isinstance(s.label, LocalizedText) for s in env.card_stats)
    assert all(isinstance(a.label, LocalizedText) for a in env.card_actions)
    # The list itself still rides in context so the template owns the frame.
    assert "residue_list" in env.context