"""Coverage for ``TransferableSummaryService`` — the per-category
counter that powers the "必须转让" tally on
``/admin/offboarding-config``.

Owner reported 2026-05-07: the existing overview only counted
``asset_software_catalog`` rows whose ``offboarding_policy`` is
``must_transfer``, so the tally read **0** even though hundreds of
cloud subscriptions / email groups / IAM users / physical assets
were owned by employees and would need transfer at offboarding time.
This file pins the six-category contract so the next refactor can't
silently drop a stream.
"""
from __future__ import annotations

import uuid
from datetime import datetime
from decimal import Decimal

import pytest

from app.domains.asset.models import AssetItem, AssetSoftwareCatalog
from app.domains.cmdb.models import (
    CmdbCloudScope,
    CmdbDomain,
    CmdbIamUser,
)
from app.domains.domain_registration.models import DomainRegistration
from app.domains.email_group.models import EmailGroup, EmailGroupOwner
from app.domains.offboarding.transferable_summary_service import (
    TransferableSummaryService,
    serialize,
)
from app.identity.person_models import IdentityPerson
from app.integrations.models import Integration


async def _make_integration(db_session, *, kind: str = "cloud") -> Integration:
    iid = uuid.uuid4()
    row = Integration(
        id=iid,
        kind=kind,
        provider="aws",
        name=f"tenant-{str(iid)[:6]}",
        slug=f"t-{str(iid)[:6]}",
        enabled=True,
        is_primary=False,
    )
    db_session.add(row)
    await db_session.flush()
    return row


async def _make_person(db_session, *, status: str = "active") -> IdentityPerson:
    p = IdentityPerson(
        source_system="test",
        hris_person_id=f"P{uuid.uuid4().hex[:8]}",
        employee_no=f"E{uuid.uuid4().hex[:8]}",
        work_email=f"u{uuid.uuid4().hex[:8]}@x.test",
        display_name="Test Person",
        employment_status=status,
    )
    db_session.add(p)
    await db_session.flush()
    return p


@pytest.mark.asyncio
async def test_summary_zero_when_no_seeded_rows(db_session):
    """Empty fixture state — the six counts are all zero. Guards
    against a future refactor that accidentally returns the table
    row count instead of the filtered count."""
    summary = await TransferableSummaryService(db_session).get_summary()
    keys = [c.key for c in summary.categories]
    assert keys == [
        "software", "cloud_scope", "domain",
        "email_group", "asset_item", "iam_user",
    ]
    for c in summary.categories:
        assert c.count == 0
    assert summary.total == 0

    # Wire-shape sanity for the frontend client.
    payload = serialize(summary)
    assert payload["total"] == 0
    assert {c["key"] for c in payload["categories"]} == set(keys)
    for c in payload["categories"]:
        assert "admin_path" in c and c["admin_path"].startswith("/")


@pytest.mark.asyncio
async def test_software_count_only_must_transfer_active(db_session):
    """``offboarding_policy='must_transfer'`` AND ``status='active'``.
    Other policies and archived rows MUST NOT contribute."""
    db_session.add_all([
        AssetSoftwareCatalog(
            name="Real owned tool",
            asset_type="software",
            status="active",
            offboarding_policy="must_transfer",
        ),
        # archived must_transfer — excluded
        AssetSoftwareCatalog(
            name="Archived tool",
            asset_type="software",
            status="archived",
            offboarding_policy="must_transfer",
        ),
        # active but different policy — excluded
        AssetSoftwareCatalog(
            name="SSO tool",
            asset_type="software",
            status="active",
            offboarding_policy="sso_cascade",
        ),
        AssetSoftwareCatalog(
            name="Auto-revoke tool",
            asset_type="software",
            status="active",
            offboarding_policy="auto_revoke",
        ),
    ])
    await db_session.flush()

    summary = await TransferableSummaryService(db_session).get_summary()
    by_key = {c.key: c.count for c in summary.categories}
    assert by_key["software"] == 1
    # And the others stay at zero — counts are disjoint.
    assert by_key["cloud_scope"] == 0
    assert by_key["domain"] == 0


@pytest.mark.asyncio
async def test_cloud_scope_count_only_owned_active_alive(db_session):
    """Active + non-deleted + owner_person_id IS NOT NULL."""
    integ = await _make_integration(db_session)
    owner = await _make_person(db_session)

    db_session.add_all([
        CmdbCloudScope(
            integration_id=str(integ.id),
            scope_type="aws_account",
            external_id=f"acct-{uuid.uuid4().hex[:6]}",
            display_name="Owned + active",
            owner_person_id=str(owner.id),
            budget_usd_monthly=Decimal("100"),
            status="active",
        ),
        CmdbCloudScope(
            integration_id=str(integ.id),
            scope_type="aws_account",
            external_id=f"acct-{uuid.uuid4().hex[:6]}",
            display_name="Unowned — excluded",
            owner_person_id=None,
            status="active",
        ),
        CmdbCloudScope(
            integration_id=str(integ.id),
            scope_type="aws_account",
            external_id=f"acct-{uuid.uuid4().hex[:6]}",
            display_name="Decommissioned — excluded",
            owner_person_id=str(owner.id),
            status="decommissioned",
        ),
        CmdbCloudScope(
            integration_id=str(integ.id),
            scope_type="aws_account",
            external_id=f"acct-{uuid.uuid4().hex[:6]}",
            display_name="Soft-deleted — excluded",
            owner_person_id=str(owner.id),
            status="active",
            deleted_at=datetime(2026, 4, 1, 0, 0, 0),
        ),
    ])
    await db_session.flush()

    summary = await TransferableSummaryService(db_session).get_summary()
    by_key = {c.key: c.count for c in summary.categories}
    assert by_key["cloud_scope"] == 1


@pytest.mark.asyncio
async def test_domain_count_combines_two_streams(db_session):
    """Domain card sums ``domain_registrations`` (self-service) +
    ``cmdb_domains`` (registrar discovery) so admins see one number
    instead of having to add columns from two pages."""
    integ = await _make_integration(db_session, kind="dns")
    owner = await _make_person(db_session)

    from app.domains.domain_registration.models import (
        DOMAIN_REG_STATUS_APPROVED,
        DOMAIN_REG_STATUS_PENDING,
        DOMAIN_REG_STATUS_REVOKED,
    )
    db_session.add_all([
        DomainRegistration(
            person_id=owner.id,
            domain_name="alice.staging.x.test",
            purpose="staging env",
            status=DOMAIN_REG_STATUS_APPROVED,
        ),
        DomainRegistration(
            person_id=owner.id,
            domain_name="alice.pending.x.test",
            purpose="not yet granted",
            status=DOMAIN_REG_STATUS_PENDING,  # excluded — not yet held
        ),
        DomainRegistration(
            person_id=owner.id,
            domain_name="alice.revoked.x.test",
            purpose="old env",
            status=DOMAIN_REG_STATUS_REVOKED,  # excluded
        ),
        CmdbDomain(
            integration_id=str(integ.id),
            domain_name="x.test",
            registrar="aliyun",
            status="active",
            owner_person_id=str(owner.id),
        ),
        CmdbDomain(
            integration_id=str(integ.id),
            domain_name="orphan.x.test",
            registrar="aliyun",
            status="active",
            owner_person_id=None,  # excluded
        ),
    ])
    await db_session.flush()

    summary = await TransferableSummaryService(db_session).get_summary()
    by_key = {c.key: c.count for c in summary.categories}
    assert by_key["domain"] == 2  # 1 from each stream


@pytest.mark.asyncio
async def test_email_group_count_distinct_groups(db_session):
    """Multi-owner groups count once — admins reason about the
    group as the unit of hand-off, not each owner row."""
    g1 = EmailGroup(
        email=f"g1-{uuid.uuid4().hex[:6]}@x.test",
        name="Group 1",
        status="active",
    )
    g2 = EmailGroup(
        email=f"g2-{uuid.uuid4().hex[:6]}@x.test",
        name="Group 2",
        status="active",
    )
    g_inactive = EmailGroup(
        email=f"g3-{uuid.uuid4().hex[:6]}@x.test",
        name="Inactive group",
        status="archived",  # excluded — can't be reassigned
    )
    db_session.add_all([g1, g2, g_inactive])
    await db_session.flush()

    p1 = await _make_person(db_session)
    p2 = await _make_person(db_session)

    db_session.add_all([
        EmailGroupOwner(email_group_id=g1.id, person_id=p1.id),
        EmailGroupOwner(email_group_id=g1.id, person_id=p2.id),  # co-owner
        EmailGroupOwner(email_group_id=g2.id, person_id=p1.id),
        # Owner of an archived group — must NOT contribute.
        EmailGroupOwner(email_group_id=g_inactive.id, person_id=p1.id),
    ])
    await db_session.flush()

    summary = await TransferableSummaryService(db_session).get_summary()
    by_key = {c.key: c.count for c in summary.categories}
    assert by_key["email_group"] == 2  # two distinct ACTIVE groups


@pytest.mark.asyncio
async def test_asset_item_count_in_use_held_only(db_session):
    """Held + in-use status. Disposed / pending-procurement assets
    must not count — they aren't going to require a hand-off."""
    holder = await _make_person(db_session)

    db_session.add_all([
        AssetItem(
            asset_tag=f"AT-{uuid.uuid4().hex[:6]}",
            category="laptop",
            status="assigned",
            assigned_to=holder.id,
        ),
        AssetItem(
            asset_tag=f"AT-{uuid.uuid4().hex[:6]}",
            category="laptop",
            status="claimed",
            assigned_to=holder.id,
        ),
        AssetItem(
            asset_tag=f"AT-{uuid.uuid4().hex[:6]}",
            category="laptop",
            status="borrowed",
            borrowed_by=holder.id,
        ),
        # disposed — excluded
        AssetItem(
            asset_tag=f"AT-{uuid.uuid4().hex[:6]}",
            category="laptop",
            status="disposed",
            assigned_to=holder.id,
        ),
        # idle / unassigned — excluded
        AssetItem(
            asset_tag=f"AT-{uuid.uuid4().hex[:6]}",
            category="laptop",
            status="idle",
            assigned_to=None,
            borrowed_by=None,
        ),
    ])
    await db_session.flush()

    summary = await TransferableSummaryService(db_session).get_summary()
    by_key = {c.key: c.count for c in summary.categories}
    assert by_key["asset_item"] == 3


@pytest.mark.asyncio
async def test_iam_user_count_only_with_owner_alive(db_session):
    """IAM users with ``owner_person_id IS NOT NULL`` and not
    soft-deleted. Unowned IAM users (cleanup-flow territory) are
    NOT in the must-transfer tally."""
    integ = await _make_integration(db_session)
    owner = await _make_person(db_session)

    db_session.add_all([
        CmdbIamUser(
            integration_id=str(integ.id),
            user_name=f"u-{uuid.uuid4().hex[:6]}",
            owner_person_id=str(owner.id),
        ),
        CmdbIamUser(
            integration_id=str(integ.id),
            user_name=f"u-{uuid.uuid4().hex[:6]}",
            owner_person_id=None,  # excluded
        ),
        CmdbIamUser(
            integration_id=str(integ.id),
            user_name=f"u-{uuid.uuid4().hex[:6]}",
            owner_person_id=str(owner.id),
            deleted_at=datetime(2026, 1, 1, 0, 0, 0),  # excluded
        ),
    ])
    await db_session.flush()

    summary = await TransferableSummaryService(db_session).get_summary()
    by_key = {c.key: c.count for c in summary.categories}
    assert by_key["iam_user"] == 1


@pytest.mark.asyncio
async def test_total_sums_categories(db_session):
    """The page header surfaces ``total`` directly. Pin the
    arithmetic so a future refactor can't silently drop a category
    from the sum (the original bug fingerprint was the page showing
    ``0 项`` while the categories were nonzero)."""
    integ = await _make_integration(db_session)
    p = await _make_person(db_session)

    db_session.add_all([
        AssetSoftwareCatalog(
            name="Tool A", asset_type="software",
            status="active", offboarding_policy="must_transfer",
        ),
        CmdbCloudScope(
            integration_id=str(integ.id),
            scope_type="aws_account",
            external_id=f"acct-{uuid.uuid4().hex[:6]}",
            display_name="Owned scope",
            owner_person_id=str(p.id),
            status="active",
        ),
        CmdbIamUser(
            integration_id=str(integ.id),
            user_name=f"u-{uuid.uuid4().hex[:6]}",
            owner_person_id=str(p.id),
        ),
        AssetItem(
            asset_tag=f"AT-{uuid.uuid4().hex[:6]}",
            category="laptop",
            status="assigned",
            assigned_to=p.id,
        ),
    ])
    await db_session.flush()

    summary = await TransferableSummaryService(db_session).get_summary()
    by_key = {c.key: c.count for c in summary.categories}
    assert by_key["software"] == 1
    assert by_key["cloud_scope"] == 1
    assert by_key["iam_user"] == 1
    assert by_key["asset_item"] == 1
    assert by_key["domain"] == 0
    assert by_key["email_group"] == 0
    assert summary.total == 4