"""跨域打标:哪些行会被标上,以及哪些行**绝不能**被标上。

这个文件里一半的用例是负例,理由是这套东西的失败模式不对称:漏标一个
CBG 员工的后果是一条待办没消掉;误标一个在职员工的后果是他离职时不会
被回收。所以每一条正向匹配旁边都放一条它不该越过的边界。
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime

import pytest

from app.core.coordinators.partner_tag_coordinator import sync_partner_tags
from app.domains.asset.models import ExternalAccountRegistry
from app.domains.github_org import constants as gh_const
from app.domains.github_org.models import GithubOrg, GithubOrgMembership
from app.domains.google_workspace.models import GoogleOrphanAccount
from app.domains.partner_directory.models import (
    LINK_APPLIED,
    LINK_DISMISSED,
    LINK_PENDING_REVIEW,
    MATCH_EMAIL_LOCALPART,
    MATCH_GITHUB_ID,
    MATCH_NAME,
    SURFACE_GITHUB_ORG,
    SURFACE_GOOGLE_ORPHAN,
    PartnerDirectoryLink,
    PartnerDirectoryMember,
)
from app.identity.person_models import IdentityPerson
from app.integrations.models import Integration

pytestmark = pytest.mark.asyncio

LOCAL_DOMAIN = "pingcap.com"
PARTNER_DOMAIN = "pingcap.cn"


async def _integration(db) -> Integration:
    row = Integration(
        id=uuid.uuid4(),
        kind="partner_directory",
        provider="feishu",
        name="CBG飞书",
        slug=f"cbg-{uuid.uuid4().hex[:8]}",
        enabled=True,
        config_public={
            "app_id": "cli_test",
            "partner_email_domain": PARTNER_DOMAIN,
            "local_email_domain": LOCAL_DOMAIN,
        },
    )
    db.add(row)
    await db.flush()
    return row


async def _member(
    db, integration, *, email_name=None, name=None, github=None,
    external_id=None,
) -> PartnerDirectoryMember:
    email = f"{email_name}@{PARTNER_DOMAIN}" if email_name else None
    row = PartnerDirectoryMember(
        id=str(uuid.uuid4()),
        integration_id=str(integration.id),
        external_id=external_id or f"ou_{uuid.uuid4().hex[:12]}",
        name=name,
        email=email,
        email_name=email_name,
        github_login=github,
        github_login_lower=(github or "").lower() or None,
        is_active=True,
    )
    db.add(row)
    await db.flush()
    return row


async def _orphan(db, email, *, display_name=None) -> GoogleOrphanAccount:
    row = GoogleOrphanAccount(
        id=uuid.uuid4(),
        primary_email=email,
        google_user_id=f"gid-{uuid.uuid4().hex[:8]}",
        display_name=display_name or email,
        suspended=False,
        is_admin=False,
    )
    db.add(row)
    await db.flush()
    return row


async def _person(db, email, *, name="我们的人", status="active") -> IdentityPerson:
    row = IdentityPerson(
        id=uuid.uuid4(),
        work_email=email,
        display_name=name,
        employment_status=status,
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
        employee_no=f"E{uuid.uuid4().hex[:6]}",
    )
    db.add(row)
    await db.flush()
    return row


async def _github_membership(
    db, login, *, person_id=None, profile=None,
) -> GithubOrgMembership:
    org = GithubOrg(
        id=str(uuid.uuid4()),
        slug=f"org{uuid.uuid4().hex[:6]}",
        slug_lower=None,
        display_name="Test Org",
        is_active=True,
    )
    org.slug_lower = org.slug.lower()
    db.add(org)
    await db.flush()
    row = GithubOrgMembership(
        id=str(uuid.uuid4()),
        org_id=org.id,
        person_id=person_id,
        github_login=login,
        github_login_lower=login.lower(),
        state=gh_const.MEMBERSHIP_STATE_ACTIVE,
        github_profile=profile,
    )
    db.add(row)
    await db.flush()
    return row


async def _external_seat(db, integration, email) -> None:
    """一条 ``category='external'`` 的厂商坐席行。

    要建 software + plan 两个 FK 父 —— 这个用例要覆盖的正是 vendor_seat
    这条路径(生产上漏掉的那一个就是从这里进来的),换成没有 FK 负担的
    Google 孤儿面就测不到真实形状了。
    """
    from decimal import Decimal

    from app.domains.asset.models import (
        AssetSoftwareCatalog,
        SoftwareExternalHolder,
        SoftwareLicensePlan,
    )

    sw = AssetSoftwareCatalog(
        id=uuid.uuid4(), name=f"sw-{uuid.uuid4().hex[:6]}",
        asset_type="SaaS", status="active",
    )
    db.add(sw)
    await db.flush()
    plan = SoftwareLicensePlan(
        id=uuid.uuid4(), software_id=sw.id, name="seats",
        price=Decimal("10.00"), billing_cycle="monthly", currency="USD",
        status="active", max_seats=None,
    )
    db.add(plan)
    await db.flush()
    now = datetime.now(UTC)
    db.add(SoftwareExternalHolder(
        id=uuid.uuid4(), integration_id=uuid.UUID(str(integration.id)),
        software_id=sw.id, plan_id=plan.id, provider="feishu",
        holder_key=email, email=email, display_name=email.split("@")[0],
        category="external", resolved=False,
        first_observed_at=now, last_observed_at=now,
    ))
    await db.flush()


async def _run(db, integration, **kw):
    return await sync_partner_tags(
        db,
        integration_id=str(integration.id),
        local_email_domain=LOCAL_DOMAIN,
        trace_id="test",
        **kw,
    )


async def _links(db, integration) -> list[PartnerDirectoryLink]:
    from sqlalchemy import select

    return list((await db.execute(
        select(PartnerDirectoryLink).where(
            PartnerDirectoryLink.integration_id == str(integration.id),
        ),
    )).scalars().all())


async def _registry(db, email) -> ExternalAccountRegistry | None:
    from sqlalchemy import func, select

    return (await db.execute(
        select(ExternalAccountRegistry).where(
            func.lower(ExternalAccountRegistry.email) == email.lower(),
        ),
    )).scalars().first()


# ── 主路径:跨域邮箱名 ────────────────────────────────────────────


async def test_localpart_across_domains_tags_a_google_orphan(db_session):
    """🔴 这条就是整个功能存在的理由。

    生产里 ``google_orphan_accounts`` 的 ``@pingcap.cn`` 是 **0 行** ——
    按完整邮箱串匹配,Gmail 那 32 个 CBG 账号一个都对不上。
    """
    integration = await _integration(db_session)
    await _member(db_session, integration, email_name="chenshuang", name="陈霜")
    await _orphan(db_session, f"chenshuang@{LOCAL_DOMAIN}")

    outcome = await _run(db_session, integration)

    assert outcome.matched == 1
    assert outcome.applied == 1
    assert outcome.by_method == {MATCH_EMAIL_LOCALPART: 1}
    row = await _registry(db_session, f"chenshuang@{LOCAL_DOMAIN}")
    assert row is not None
    assert row.classification == "external_guest"
    link = (await _links(db_session, integration))[0]
    assert link.surface == SURFACE_GOOGLE_ORPHAN
    assert link.state == LINK_APPLIED
    assert link.applied_at is not None


async def test_localpart_does_not_reach_a_foreign_domain(db_session):
    """同名邮箱名 + 不是我们的域 ⇒ 不匹配。

    外部坐席里有 gmail / qq 之类的地址,一个 ``chenshuang@gmail.com``
    和对方的 ``chenshuang@pingcap.cn`` 邮箱名相同纯属巧合。
    """
    integration = await _integration(db_session)
    await _member(db_session, integration, email_name="chenshuang")
    await _orphan(db_session, "chenshuang@gmail.com")

    outcome = await _run(db_session, integration)

    assert outcome.matched == 0
    assert await _registry(db_session, "chenshuang@gmail.com") is None


async def test_ambiguous_mailbox_name_matches_nobody(db_session):
    """对方名册里两个人共用一个邮箱名 ⇒ 这个键作废,不是挑第一个。

    「随机挑一个」在一个会往生产写标签的匹配器里不是可接受的语义。
    """
    integration = await _integration(db_session)
    await _member(db_session, integration, email_name="li.wei", name="李伟")
    # 同一个 email_name,不同的人(对方两个域 / 历史遗留都出现过)。
    dup = await _member(db_session, integration, name="李薇")
    dup.email = f"li.wei@{PARTNER_DOMAIN}.legacy"
    dup.email_name = "li.wei"
    await db_session.flush()
    await _orphan(db_session, f"li.wei@{LOCAL_DOMAIN}")

    outcome = await _run(db_session, integration)

    assert outcome.matched == 0
    assert any("email_name" in e for e in outcome.errors)


# ── GitHub ────────────────────────────────────────────────────────


async def test_github_id_confirms_an_unlinked_handle(db_session):
    integration = await _integration(db_session)
    await _member(
        db_session, integration, email_name="octo", name="章鱼", github="OctoCat",
    )
    membership = await _github_membership(db_session, "octocat")

    outcome = await _run(db_session, integration)

    assert outcome.by_method == {MATCH_GITHUB_ID: 1}
    await db_session.refresh(membership)
    assert membership.external_confirmed_at is not None
    assert membership.external_note is not None
    assert "章鱼" in membership.external_note
    # 名册命中的一定是人:名册里每一行都有姓名、工号和企业邮箱。机器账号
    # 分类器只能靠 handle 命名约定猜(GitHub 自己的 type 字段对 bot 也报
    # User),这里有比它强的证据。
    assert membership.external_kind == gh_const.EXTERNAL_KIND_OUTSIDE_HUMAN


async def test_github_handle_linked_to_a_person_is_never_touched(db_session):
    """🔴 已经认出是我们的人的 handle 不进候选集。

    这是跨域邮箱名匹配之所以安全的那个保证 —— 它靠的是查询的 WHERE
    子句,不是「应该不会撞名」。
    """
    integration = await _integration(db_session)
    person = IdentityPerson(
        id=uuid.uuid4(),
        work_email=f"octo@{LOCAL_DOMAIN}",
        display_name="我们的人",
        employment_status="active",
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
        employee_no=f"E{uuid.uuid4().hex[:6]}",
    )
    db_session.add(person)
    await db_session.flush()
    await _member(db_session, integration, email_name="octo", github="octocat")
    membership = await _github_membership(
        db_session, "octocat", person_id=str(person.id),
    )

    outcome = await _run(db_session, integration)

    await db_session.refresh(membership)
    assert membership.external_confirmed_at is None
    assert not any(
        link.surface == SURFACE_GITHUB_ORG
        for link in await _links(db_session, integration)
    )
    assert outcome.by_surface.get(SURFACE_GITHUB_ORG) is None


async def test_second_run_reports_the_github_tag_as_still_applied(db_session):
    """🔴 第二天再跑,已经确认过的 handle 不能报成「被拒」。

    同步每天都会重新走一遍所有命中。第一版代码在 handle 已确认时返回
    「该 handle 已经被标记为外部人员」当作没写成 —— 于是从第二天起,
    生产上 41 条 GitHub 链接会全部长期显示被拒,而它们一条问题都没有。
    """
    integration = await _integration(db_session)
    await _member(
        db_session, integration, email_name="octo", name="章鱼", github="octocat",
    )
    membership = await _github_membership(db_session, "octocat")

    first = await _run(db_session, integration)
    assert first.applied == 1

    second = await _run(db_session, integration)

    assert second.matched == 1
    assert second.applied == 1
    assert second.skipped == 0
    await db_session.refresh(membership)
    assert membership.external_confirmed_at is not None
    link = (await _links(db_session, integration))[0]
    assert link.apply_skipped_reason is None


async def test_a_stale_external_seat_for_an_active_employee_is_not_tagged(db_session):
    """🔴 生产第一跑就漏的那一个。

    `software_external_holder.category='external'` 是**上一次坐席同步的
    快照**,不是此刻的事实:生产里 18 个邮箱在那张表上挂着 external,而
    `identity_persons` 现在说他们在职。其中 `chunlin.zhuang@pingcap.com`
    恰好在 CBG 名册里有一个同邮箱名的人(其实就是同一个人,两边通讯录
    都有),于是被自动标成了 external_guest —— 她离职时那个标会让回收
    流程豁免她的账号。

    安全性必须来自权威表的当前状态,不能来自另一个域的快照。
    """
    integration = await _integration(db_session)
    await _member(db_session, integration, email_name="chunlin.zhuang", name="庄清苓")
    email = f"chunlin.zhuang@{LOCAL_DOMAIN}"
    await _person(db_session, email, name="庄清苓 Thalie Zhuang")
    # 陈旧快照:坐席表说这个邮箱是外部人,而 HRIS 说她在职。
    await _external_seat(db_session, integration, email)

    outcome = await _run(db_session, integration)

    assert outcome.matched == 0
    assert await _registry(db_session, email) is None
    assert await _links(db_session, integration) == []


async def test_an_active_employees_alias_is_also_protected(db_session):
    """别名也算「这个人现在持有的地址」。

    邮箱改过名的人,旧地址留在 email_aliases 里,而 Google 那边旧地址
    仍然是一个活账号 —— 那正是最容易被当成孤儿账号的形状。
    """
    integration = await _integration(db_session)
    await _member(db_session, integration, email_name="old.name", name="改过名的人")
    old_email = f"old.name@{LOCAL_DOMAIN}"
    person = await _person(db_session, f"new.name@{LOCAL_DOMAIN}", name="改过名的人")
    person.email_aliases = [old_email]
    await db_session.flush()
    await _orphan(db_session, old_email)

    outcome = await _run(db_session, integration)

    assert outcome.matched == 0
    assert await _registry(db_session, old_email) is None


async def test_a_departed_employees_address_is_still_taggable(db_session):
    """闸只挡在职的人 —— 离职后去了合作方,正是这套东西要标的主案例。"""
    integration = await _integration(db_session)
    await _member(db_session, integration, email_name="moved.on", name="去了对面")
    email = f"moved.on@{LOCAL_DOMAIN}"
    await _person(db_session, email, name="去了对面", status="left")
    await _orphan(db_session, email)

    outcome = await _run(db_session, integration)

    assert outcome.applied == 1
    row = await _registry(db_session, email)
    assert row is not None
    assert row.classification == "external_guest"


# ── 不覆盖人的判断 ────────────────────────────────────────────────


async def test_admin_classification_is_not_clobbered(db_session):
    """管理员写过 ``shared_service`` 的行不被降级成 ``external_guest``。

    而且「没写成」不是静默的:原因落在链接行上,列表页看得见。
    """
    integration = await _integration(db_session)
    await _member(db_session, integration, email_name="release-bot")
    email = f"release-bot@{LOCAL_DOMAIN}"
    await _orphan(db_session, email)
    db_session.add(ExternalAccountRegistry(
        id=uuid.uuid4(), email=email, classification="shared_service",
        note="CI 发布机器人", created_by=uuid.uuid4(),
    ))
    await db_session.flush()

    outcome = await _run(db_session, integration)

    assert outcome.matched == 1
    assert outcome.applied == 0
    assert outcome.skipped == 1
    row = await _registry(db_session, email)
    assert row is not None
    assert row.classification == "shared_service"
    link = (await _links(db_session, integration))[0]
    assert link.state == LINK_APPLIED
    assert link.applied_at is None
    assert "shared_service" in (link.apply_skipped_reason or "")


async def test_already_external_guest_counts_as_applied_not_refused(db_session):
    """🔴 标签已经在位 ≠ 写入被拒。

    生产首次试算:41 个命中的 Google 孤儿账号里 **29 个此前已经被人
    工点成 external_guest**。如果把它们报成 skipped,每天的同步汇报会
    长期写着「41 命中 / 12 生效 / 29 被拒」—— 而那 29 条一个问题都没有。
    一个常年非零的「被拒」数会训练人不看这个数。
    """
    integration = await _integration(db_session)
    await _member(db_session, integration, email_name="chenshuang", name="陈霜")
    email = f"chenshuang@{LOCAL_DOMAIN}"
    await _orphan(db_session, email)
    db_session.add(ExternalAccountRegistry(
        id=uuid.uuid4(), email=email, classification="external_guest",
        note="CBG保留 陈霜使用",
    ))
    await db_session.flush()

    outcome = await _run(db_session, integration)

    assert outcome.matched == 1
    assert outcome.applied == 1
    assert outcome.skipped == 0
    link = (await _links(db_session, integration))[0]
    assert link.apply_skipped_reason is None


# ── 弱信号 ────────────────────────────────────────────────────────


async def test_name_only_match_queues_and_writes_nothing(db_session):
    integration = await _integration(db_session)
    await _member(db_session, integration, email_name="zhang.san", name="张三")
    await _orphan(
        db_session, f"totally.different@{LOCAL_DOMAIN}", display_name="张三",
    )

    outcome = await _run(db_session, integration)

    assert outcome.by_method == {MATCH_NAME: 1}
    assert outcome.applied == 0
    assert outcome.pending_review == 1
    assert await _registry(db_session, f"totally.different@{LOCAL_DOMAIN}") is None
    link = (await _links(db_session, integration))[0]
    assert link.state == LINK_PENDING_REVIEW


async def test_dismissed_link_is_never_revived_by_a_later_sync(db_session):
    """否掉是终态。每天把人已经拒绝过的东西重新提出来,一周后就没人看了。"""
    integration = await _integration(db_session)
    await _member(db_session, integration, email_name="chenshuang")
    email = f"chenshuang@{LOCAL_DOMAIN}"
    await _orphan(db_session, email)

    await _run(db_session, integration)
    link = (await _links(db_session, integration))[0]
    link.state = LINK_DISMISSED
    await db_session.flush()

    outcome = await _run(db_session, integration)

    await db_session.refresh(link)
    assert link.state == LINK_DISMISSED
    assert outcome.applied == 0
    assert outcome.skipped == 1


# ── 试算 ──────────────────────────────────────────────────────────


async def test_dry_run_matches_without_writing_anything(db_session):
    integration = await _integration(db_session)
    await _member(db_session, integration, email_name="chenshuang")
    email = f"chenshuang@{LOCAL_DOMAIN}"
    await _orphan(db_session, email)

    outcome = await _run(db_session, integration, apply=False)

    assert outcome.matched == 1
    assert outcome.applied == 0
    assert await _registry(db_session, email) is None
    assert (await _links(db_session, integration))[0].state == LINK_PENDING_REVIEW


# ── 名册退役 ──────────────────────────────────────────────────────


async def test_incomplete_snapshot_does_not_retire_members(db_session):
    """🔴 读不全的一次同步不能把名册清空。

    否则权限一过期,几十个账号的 CBG 标签当天集体变成僵尸链接,第二天
    权限恢复又全部复活 —— 而中间那一天离职流程会把 CBG 保留的人当成
    该回收的人。
    """
    from app.domains.partner_directory.service import sync_roster
    from app.integrations.driver_base import (
        PartnerDirectoryDriver,
        PartnerDirectoryMemberRecord,
        PartnerDirectorySnapshot,
    )

    integration = await _integration(db_session)
    existing = await _member(
        db_session, integration, email_name="still.here", external_id="ou_keep",
    )

    class _Partial(PartnerDirectoryDriver):
        provider = "test"

        async def list_members(self) -> PartnerDirectorySnapshot:
            return PartnerDirectorySnapshot(
                members=[PartnerDirectoryMemberRecord(external_id="ou_other")],
                departments_scanned=10,
                departments_unreadable=9,
                complete=False,
            )

    driver = _Partial(integration_id=str(integration.id), config={})
    result = await sync_roster(
        db_session, integration_id=str(integration.id), driver=driver,
    )

    assert result["retired"] == 0
    await db_session.refresh(existing)
    assert existing.deleted_at is None

    class _Complete(_Partial):
        async def list_members(self) -> PartnerDirectorySnapshot:
            return PartnerDirectorySnapshot(
                members=[PartnerDirectoryMemberRecord(external_id="ou_other")],
                departments_scanned=10,
                departments_unreadable=0,
                complete=True,
            )

    result = await sync_roster(
        db_session, integration_id=str(integration.id),
        driver=_Complete(integration_id=str(integration.id), config={}),
    )
    assert result["retired"] == 1
    await db_session.refresh(existing)
    assert existing.deleted_at is not None
    # 刻意**不**断言 ``deleted_at <= now()``:``deleted_at`` 是不带 fsp 的
    # MySQL DATETIME,写 36.733 秒读回来是 37 秒(DATETIME 是四舍五入,
    # 不是截断),于是一个刚写下的时间戳可以比当前时刻晚半秒。那条断言
    # 会随机红,而且它测的不是这个用例关心的事。
    assert isinstance(existing.deleted_at, datetime)