"""自动加入规则:部门子树匹配、三道闸、以及「缺 handle 不静默跳过」。

最重要的两条:

* ``Engineering`` 命中 ``Engineering / TiDB``,**不**命中
  ``Engineering Support``。裸字符串前缀会吃掉后者,而对外 API 的部门筛选
  为此修过一次 —— 这里从第一天就按路径段比。
* 命中了但**没有 GitHub handle** 不是跳过,是一件待办。生产上在职员工
  只有约 63% 登记了 handle,静默跳过意味着近四成的人永远不会被自动加入,
  而且没有任何地方说过这件事。
"""
from __future__ import annotations

import uuid

import pytest

from app.core.github_org_match import (
    department_subtree_matches,
    rule_matches_person,
)
from app.domains.github_org import auto_join, execution
from app.domains.github_org import constants as const
from app.domains.github_org import service as svc
from app.identity.person_models import IdentityOrgMembership, IdentityPerson
from app.integrations.driver_base import GithubOrgActionResult, GithubOrgDriver

# ``pytest.mark.asyncio`` 只挂在需要 DB 的那一半上 —— 纯匹配语义的
# 用例是同步函数,全局 mark 会让 pytest 对每一条都发一次警告。
_async = pytest.mark.asyncio


# ── 纯匹配语义(不碰 DB) ────────────────────────────────────────


def test_subtree_matches_children_not_siblings():
    """🔴 ``Engineering`` 不该吃掉 ``Engineering Support``。"""
    assert department_subtree_matches(
        rule_path="Engineering", person_segments=["Engineering"],
    )
    assert department_subtree_matches(
        rule_path="Engineering", person_segments=["Engineering", "TiDB"],
    )
    assert not department_subtree_matches(
        rule_path="Engineering", person_segments=["Engineering Support"],
    )
    assert not department_subtree_matches(
        rule_path="Engineering",
        person_segments=["Engineering Support", "Docs"],
    )


def test_deeper_rule_needs_deeper_person():
    assert department_subtree_matches(
        rule_path="Engineering / TiDB",
        person_segments=["Engineering", "TiDB", "Cloud"],
    )
    assert not department_subtree_matches(
        rule_path="Engineering / TiDB", person_segments=["Engineering"],
    )


def test_empty_rule_path_is_wildcard_but_missing_person_data_is_not():
    """通配规则命中所有人;而**指定了部门**的规则遇到缺数据的人不命中。

    fail-closed:自动把人加进 GitHub 组织是一次真实的权限授予,数据缺失时
    应该少给,不该多给。
    """
    assert department_subtree_matches(rule_path=None, person_segments=[])
    assert not department_subtree_matches(
        rule_path="Engineering", person_segments=[],
    )


def test_tenant_root_is_stripped_from_the_rule_too():
    """管理员从组织树上复制的路径可能带租户根 —— 不剥根的后果不是报错,
    是一条永远匹配不上的规则。"""
    assert department_subtree_matches(
        rule_path="PingCAP / Engineering",
        person_segments=["Engineering", "TiDB"],
    )


class _Rule:
    def __init__(self, **kw):
        self.is_active = kw.pop("is_active", True)
        self.deleted_at = kw.pop("deleted_at", None)
        self.match_department_path = kw.pop("match_department_path", None)
        self.match_job_title = kw.pop("match_job_title", None)
        self.match_location = kw.pop("match_location", None)


def test_all_dimensions_must_hold():
    rule = _Rule(match_department_path="Engineering", match_location="CHN0075")
    assert rule_matches_person(
        rule, department_segments_of_person=["Engineering"],
        job_title="SRE", location="CHN0075",
    )
    assert not rule_matches_person(
        rule, department_segments_of_person=["Engineering"],
        job_title="SRE", location="USA0635",
    )


def test_missing_person_dimension_fails_closed():
    rule = _Rule(match_job_title="SRE")
    assert not rule_matches_person(
        rule, department_segments_of_person=["Engineering"],
        job_title=None, location=None,
    )


def test_inactive_and_deleted_rules_never_match():
    for kw in ({"is_active": False}, {"deleted_at": object()}):
        rule = _Rule(match_department_path="Engineering", **kw)
        assert not rule_matches_person(
            rule, department_segments_of_person=["Engineering"],
            job_title=None, location=None,
        )


# ── 规则 CRUD ───────────────────────────────────────────────────


async def _org(db, **kw):
    payload = {
        "slug": "pingcap-inc", "display_name": "PingCAP Inc",
        "auto_join_enabled": True,
    }
    payload.update(kw)
    org = await svc.create_org(db, **payload)
    await db.commit()
    return org


@_async
async def test_rule_with_no_constraint_is_refused(db_session):
    """三个维度全空 = 命中所有人 = 把全公司加进一个 GitHub 组织。"""
    from app.core.exceptions import ValidationError

    org = await _org(db_session)
    with pytest.raises(ValidationError):
        await auto_join.create_rule(db_session, org_id=org.id)


@_async
async def test_department_path_is_normalised_on_write(db_session):
    org = await _org(db_session)
    rule = await auto_join.create_rule(
        db_session, org_id=org.id,
        match_department_path="  PingCAP /Engineering/ TiDB  ",
    )
    assert rule.match_department_path == "Engineering / TiDB"


@_async
async def test_duplicate_rule_is_refused(db_session):
    from app.core.exceptions import ConflictError

    org = await _org(db_session)
    await auto_join.create_rule(
        db_session, org_id=org.id, match_department_path="Engineering",
    )
    await db_session.flush()
    with pytest.raises(ConflictError):
        await auto_join.create_rule(
            db_session, org_id=org.id, match_department_path="Engineering",
        )


# ── 执行 ─────────────────────────────────────────────────────────


class _FakeDriver(GithubOrgDriver):
    provider = "fake"

    def __init__(self) -> None:
        super().__init__(integration_id="fake", config={})
        self.invited: list[str] = []

    async def list_members(self, org, *, role=None):
        return []

    async def list_pending_invitations(self, org):
        return []

    async def invite_member(self, org, login, *, role="member"):
        self.invited.append(login)
        return GithubOrgActionResult(ok=True, invited=True)

    async def remove_member(self, org, login):
        return GithubOrgActionResult(ok=True)


def _patch_driver(monkeypatch, driver):
    from app.domains.github_org import driver_access as da

    async def _fake(db, org):
        return da.OrgDriverResolution(org=org, driver=driver)

    monkeypatch.setattr(execution, "resolve_org_driver", _fake)


async def _person(db, *, dept_path: str, github_id: str | None = "octocat"):
    person = IdentityPerson(
        id=uuid.uuid4(),
        work_email=f"p{uuid.uuid4().hex[:8]}@test.com",
        display_name="Test Person",
        employment_status="active",
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
        github_id=github_id,
    )
    db.add(person)
    await db.flush()
    db.add(
        IdentityOrgMembership(
            person_id=person.id,
            org_node_source_id=f"node-{uuid.uuid4().hex[:8]}",
            org_node_type="department",
            org_node_name=dept_path.split("/")[-1].strip(),
            org_path=f"PingCAP / {dept_path}",
            is_primary=True,
        )
    )
    await db.flush()
    await db.refresh(person, ["org_memberships"])
    return person


@_async
async def test_matching_person_is_invited(db_session, monkeypatch):
    org = await _org(db_session)
    await auto_join.create_rule(
        db_session, org_id=org.id, match_department_path="Engineering",
    )
    await db_session.flush()
    person = await _person(db_session, dept_path="Engineering / TiDB")
    driver = _FakeDriver()
    _patch_driver(monkeypatch, driver)

    out = await auto_join.apply_rules_for_person(
        db_session, person, trace_id="t",
    )
    assert out[auto_join.APPLY_JOINED] == 1
    assert driver.invited == ["octocat"]


@_async
async def test_org_master_switch_blocks_execution(db_session, monkeypatch):
    """规则配了但组织总闸没开 —— 让「先配好、确认命中范围、再开闸」可能。"""
    org = await _org(db_session, auto_join_enabled=False)
    await auto_join.create_rule(
        db_session, org_id=org.id, match_department_path="Engineering",
    )
    await db_session.flush()
    person = await _person(db_session, dept_path="Engineering")
    driver = _FakeDriver()
    _patch_driver(monkeypatch, driver)

    out = await auto_join.apply_rules_for_person(
        db_session, person, trace_id="t",
    )
    assert out[auto_join.APPLY_ORG_DISABLED] == 1
    assert driver.invited == []


@_async
async def test_missing_handle_is_an_ACTION_not_a_silent_skip(
    db_session, monkeypatch,
):
    """🔴 生产上在职员工只有约 63% 登记了 GitHub handle。

    静默跳过意味着近四成的人永远不会被自动加入,而且没有任何地方说过。
    这里要求它:① 单独计数,② 落一条审计,③ 通知本人去补。
    """
    from sqlalchemy import select

    from app.audit.models import AuditLog

    org = await _org(db_session)
    await auto_join.create_rule(
        db_session, org_id=org.id, match_department_path="Engineering",
    )
    await db_session.flush()
    person = await _person(
        db_session, dept_path="Engineering", github_id=None,
    )
    _patch_driver(monkeypatch, _FakeDriver())

    notified: list[str] = []

    async def _fake_notify(db, *, person, summary):
        notified.append(person.work_email)

    monkeypatch.setattr(auto_join, "_notify_missing_handle", _fake_notify)

    out = await auto_join.apply_rules_for_person(
        db_session, person, trace_id="t",
    )
    assert out[auto_join.APPLY_NO_HANDLE] == 1
    assert out[auto_join.APPLY_JOINED] == 0
    assert notified == [person.work_email], "没有通知本人去补登记"

    row = (
        await db_session.execute(
            select(AuditLog).where(
                AuditLog.action == "github_org.auto_join.blocked_no_handle",
            )
        )
    ).scalars().first()
    assert row is not None, "缺 handle 没有留下审计痕迹"


@_async
async def test_leaver_is_never_auto_joined(db_session, monkeypatch):
    org = await _org(db_session)
    await auto_join.create_rule(
        db_session, org_id=org.id, match_department_path="Engineering",
    )
    await db_session.flush()
    person = await _person(db_session, dept_path="Engineering")
    person.employment_status = "leaver_pending"
    await db_session.flush()
    driver = _FakeDriver()
    _patch_driver(monkeypatch, driver)

    out = await auto_join.apply_rules_for_person(
        db_session, person, trace_id="t",
    )
    assert out[auto_join.APPLY_INELIGIBLE] == 1
    assert driver.invited == []


@_async
async def test_already_member_is_not_reinvited(db_session, monkeypatch):
    org = await _org(db_session)
    await auto_join.create_rule(
        db_session, org_id=org.id, match_department_path="Engineering",
    )
    await db_session.flush()
    person = await _person(db_session, dept_path="Engineering")
    await svc.upsert_membership(
        db_session, org=org, github_login="octocat",
        person_id=str(person.id),
        state=const.MEMBERSHIP_STATE_ACTIVE, source=const.SOURCE_IMPORT,
    )
    await db_session.flush()
    driver = _FakeDriver()
    _patch_driver(monkeypatch, driver)

    out = await auto_join.apply_rules_for_person(
        db_session, person, trace_id="t",
    )
    assert out[auto_join.APPLY_ALREADY] == 1
    assert driver.invited == []


@_async
async def test_two_rules_hitting_one_org_execute_once(db_session, monkeypatch):
    """部门规则 + 职位规则同时命中同一个组织 —— 取并集,不重复执行。"""
    org = await _org(db_session)
    await auto_join.create_rule(
        db_session, org_id=org.id, match_department_path="Engineering",
    )
    await auto_join.create_rule(
        db_session, org_id=org.id, match_job_title="SRE",
    )
    await db_session.flush()
    person = await _person(db_session, dept_path="Engineering")
    person.job_title = "SRE"
    await db_session.flush()
    driver = _FakeDriver()
    _patch_driver(monkeypatch, driver)

    out = await auto_join.apply_rules_for_person(
        db_session, person, trace_id="t",
    )
    assert out[auto_join.APPLY_JOINED] == 1
    assert driver.invited == ["octocat"]


@_async
async def test_sibling_department_is_not_matched_end_to_end(
    db_session, monkeypatch,
):
    """端到端再钉一次那条最要命的边界。"""
    org = await _org(db_session)
    await auto_join.create_rule(
        db_session, org_id=org.id, match_department_path="Engineering",
    )
    await db_session.flush()
    person = await _person(db_session, dept_path="Engineering Support")
    driver = _FakeDriver()
    _patch_driver(monkeypatch, driver)

    out = await auto_join.apply_rules_for_person(
        db_session, person, trace_id="t",
    )
    assert out[auto_join.APPLY_JOINED] == 0
    assert driver.invited == []


# ── 预览 ─────────────────────────────────────────────────────────


@_async
async def test_preview_reports_the_handle_gap(db_session):
    """命中数不等于落地数 —— 缺 handle 的那部分必须一起显示。"""
    org = await _org(db_session)
    rule = await auto_join.create_rule(
        db_session, org_id=org.id, match_department_path="Engineering",
    )
    await db_session.flush()
    await _person(db_session, dept_path="Engineering", github_id="a")
    await _person(db_session, dept_path="Engineering", github_id=None)
    await _person(db_session, dept_path="Engineering Support", github_id="c")
    await db_session.flush()

    out = await auto_join.preview_rule(db_session, rule)
    assert out["match_count"] == 2, "兄弟部门被算进去了"
    assert out["without_handle_count"] == 1


# ── 管理端点 ─────────────────────────────────────────────────────


@_async
async def test_rule_endpoints_are_gated(client, employee_headers):
    assert (
        await client.get(
            "/api/v1/github/auto-join-rules", headers=employee_headers,
        )
    ).status_code == 403
    assert (
        await client.post(
            "/api/v1/github/auto-join-rules",
            headers=employee_headers,
            json={"org_id": "x", "match_department_path": "Engineering"},
        )
    ).status_code == 403


@_async
async def test_auditor_reads_rules_but_cannot_write(
    client, auditor_headers, github_admin_headers, db_session,
):
    org = await _org(db_session)
    created = await client.post(
        "/api/v1/github/auto-join-rules",
        headers=github_admin_headers,
        json={"org_id": org.id, "match_department_path": "Engineering"},
    )
    assert created.status_code == 201, created.text
    assert (
        await client.get(
            "/api/v1/github/auto-join-rules", headers=auditor_headers,
        )
    ).status_code == 200
    assert (
        await client.delete(
            f"/api/v1/github/auto-join-rules/{created.json()['data']['id']}",
            headers=auditor_headers,
        )
    ).status_code == 403


@_async
async def test_rule_update_is_audited_by_value(
    client, github_admin_headers, db_session,
):
    """把一条规则从暂停改成生效是一次扩权 —— 审计必须看得出来。"""
    from sqlalchemy import select

    from app.audit.models import AuditLog

    org = await _org(db_session)
    created = await client.post(
        "/api/v1/github/auto-join-rules",
        headers=github_admin_headers,
        json={
            "org_id": org.id, "match_department_path": "Engineering",
            "is_active": False,
        },
    )
    rule_id = created.json()["data"]["id"]
    resp = await client.patch(
        f"/api/v1/github/auto-join-rules/{rule_id}",
        headers=github_admin_headers,
        json={"is_active": True},
    )
    assert resp.status_code == 200, resp.text

    row = (
        await db_session.execute(
            select(AuditLog)
            .where(AuditLog.action == "github_org.auto_join_rule.update")
            .order_by(AuditLog.created_at.desc())
        )
    ).scalars().first()
    assert row is not None
    changes = (row.detail or {}).get("changes") or {}
    assert changes["is_active"] == {"from": False, "to": True}


@_async
async def test_deleting_a_rule_keeps_the_memberships_it_created(
    client, github_admin_headers, db_session, monkeypatch,
):
    """规则消失不该把已经发生的事实一起带走。"""
    org = await _org(db_session)
    created = await client.post(
        "/api/v1/github/auto-join-rules",
        headers=github_admin_headers,
        json={"org_id": org.id, "match_department_path": "Engineering"},
    )
    rule_id = created.json()["data"]["id"]

    person = await _person(db_session, dept_path="Engineering")
    _patch_driver(monkeypatch, _FakeDriver())
    await auto_join.apply_rules_for_person(db_session, person, trace_id="t")
    await db_session.commit()

    resp = await client.delete(
        f"/api/v1/github/auto-join-rules/{rule_id}",
        headers=github_admin_headers,
    )
    assert resp.status_code == 200

    membership = await svc.find_membership(
        db_session, org_id=org.id, github_login="octocat",
    )
    assert membership is not None
    assert membership.state in const.MEMBERSHIP_LIVE_STATES
    assert membership.source == const.SOURCE_AUTO_JOIN


# ── 回归:换了 GitHub 账号不该被重复邀请 ─────────────────────────


@_async
async def test_handle_change_does_not_cause_a_duplicate_invite(
    db_session, monkeypatch,
):
    """🔴 复审第 1 轮抓到的真缺陷(对抗性探针证实过)。

    员工用 handle A 通过审批进了组织,后来在 HRIS 里把 GitHub ID 改成 B。
    「已经在里面了吗」如果按 **handle** 查,按 B 查不到 A 那条记录,于是
    自动加入会**再邀请一次** —— 同一个组织里留下这个人的两条成员资格。

    「他已经有访问权了吗」这个问题的答案永远按**人**算。
    """
    org = await _org(db_session)
    await auto_join.create_rule(
        db_session, org_id=org.id, match_department_path="Engineering",
    )
    await db_session.flush()
    person = await _person(
        db_session, dept_path="Engineering", github_id="old-handle",
    )
    await svc.upsert_membership(
        db_session, org=org, github_login="old-handle",
        person_id=str(person.id), state=const.MEMBERSHIP_STATE_ACTIVE,
        source=const.SOURCE_APPROVAL,
    )
    await db_session.flush()

    person.github_id = "new-handle"
    await db_session.flush()

    driver = _FakeDriver()
    _patch_driver(monkeypatch, driver)
    out = await auto_join.apply_rules_for_person(
        db_session, person, trace_id="t",
    )
    assert driver.invited == [], "换了 handle 就被重复邀请了"
    assert out[auto_join.APPLY_ALREADY] == 1


@_async
async def test_handle_mismatch_is_reported_not_silently_accepted(
    db_session, monkeypatch,
):
    """不自动去撤旧的(那是需要人判断的动作),但必须说出来 ——
    否则台账上会有一条谁也不会去看的陈旧记录。"""
    from sqlalchemy import select

    from app.audit.models import AuditLog

    org = await _org(db_session)
    await auto_join.create_rule(
        db_session, org_id=org.id, match_department_path="Engineering",
    )
    await db_session.flush()
    person = await _person(
        db_session, dept_path="Engineering", github_id="old-handle",
    )
    await svc.upsert_membership(
        db_session, org=org, github_login="old-handle",
        person_id=str(person.id), state=const.MEMBERSHIP_STATE_ACTIVE,
        source=const.SOURCE_APPROVAL,
    )
    await db_session.flush()
    person.github_id = "new-handle"
    await db_session.flush()

    _patch_driver(monkeypatch, _FakeDriver())
    out = await auto_join.apply_rules_for_person(
        db_session, person, trace_id="t",
    )
    entry = out["orgs"][0]
    assert entry["handle_mismatch"]["in_org_as"] == "old-handle"
    assert entry["handle_mismatch"]["now_registered_as"] == "new-handle"

    row = (
        await db_session.execute(
            select(AuditLog).where(
                AuditLog.action == "github_org.auto_join.handle_mismatch",
            )
        )
    ).scalars().first()
    assert row is not None, "handle 不一致没有留下痕迹"