"""汇联易 dept HRBP (positionCode 6105) rolls over the SUBTREE roster.

Regression for the 2026-07-14 团建 mis-route: a parent/BU node's DIRECT
members are its leadership cluster (e.g. a CFO whose personal HRBP is a
senior HRBP), whose HRBP differs from the operating teams below. Deriving the
BU's single HRBP from direct-member plurality stamped the BU with the
leadership's HRBP, so a BU-level approval (团建) routed to the wrong HRBP even
though the leaf dept (and the whole subtree) share the team's HRBP.

Fix: HRBP is the most-common ``hris_hrbp_person_id`` across the dept's ENTIRE
subtree (self + descendants). Leader / Budget Owner / cost center stay on the
node's DIRECT members (they are leadership-anchored). These tests drive
``_huilianyi_dept_phase_body`` with a capturing fake runner and assert the
role 工号 each dept is pushed with.
"""
from __future__ import annotations

import uuid

import pytest

from app.domains.identity_sync.models import IdentitySyncTarget
from app.domains.identity_sync.runners.base import DeptOutcome
from app.domains.identity_sync.service import IdentitySyncService, RunSummary
from app.identity.person_models import (
    IdentityOrgMembership,
    IdentityOrgNode,
    IdentityPerson,
)

# BU parent (Helix-root) + one leaf child under it.
_BU = "8000"          # e.g. "Finance"
_LEAF = "8000-1"      # e.g. "Accounting"

# HRBP targets (referenced by members' hris_hrbp_person_id).
_SENIOR_HRBP_NO = "P8001"   # the leadership's (CFO's) senior HRBP
_TEAM_HRBP_NO = "P8002"     # the operating teams' HRBP
# Budget-owner targets (to prove budget owner stays DIRECT-based).
_BU_BUDGET_NO = "P8003"     # CFO's own budget owner
_TEAM_BUDGET_NO = "P8004"   # leaf team's budget owner


class _CapturingDeptRunner:
    """Records every upsert_dept's role 工号 by dept_code."""

    def __init__(self) -> None:
        self.hrbp: dict[str, str | None] = {}
        self.budget: dict[str, str | None] = {}

    async def upsert_dept(
        self, *, dept_code, hrbp_employee_no=None,
        budget_owner_employee_no=None, **_kwargs,
    ):
        self.hrbp[dept_code] = hrbp_employee_no
        self.budget[dept_code] = budget_owner_employee_no
        return DeptOutcome(status="success", external_id=dept_code)

    async def disable_dept(self, *, dept_code, **_kwargs):
        return DeptOutcome(status="soft_deleted", external_id=dept_code)

    async def aclose(self) -> None:
        return None


async def _person(
    db, emp_no: str, *, hrbp_id=None, budget_id=None,
) -> IdentityPerson:
    p = IdentityPerson(
        id=uuid.uuid4(), source_system="successfactors",
        hris_person_id=emp_no, employee_no=emp_no,
        work_email=f"{emp_no.lower()}@pingcap.com",
        display_name=emp_no, employment_status="active",
        hris_hrbp_person_id=hrbp_id,
        hris_budget_owner_person_id=budget_id,
    )
    db.add(p)
    await db.flush()
    return p


async def _seed(db) -> None:
    db.add_all([
        IdentityOrgNode(
            id=uuid.uuid4(), org_node_source_id=_BU, org_node_name="Finance",
            parent_org_node_source_id=None, org_level=1,
            org_path="PingCAP / Finance", node_status="active",
        ),
        IdentityOrgNode(
            id=uuid.uuid4(), org_node_source_id=_LEAF,
            org_node_name="Accounting", parent_org_node_source_id=_BU,
            org_level=2, org_path="PingCAP / Finance / Accounting",
            node_status="active",
        ),
    ])
    # HRBP + budget-owner target persons (not org members → don't skew counts,
    # but present in the roster so the role ids resolve to a 工号).
    senior = await _person(db, _SENIOR_HRBP_NO)
    team = await _person(db, _TEAM_HRBP_NO)
    bu_budget = await _person(db, _BU_BUDGET_NO)
    team_budget = await _person(db, _TEAM_BUDGET_NO)

    # Parent DIRECT member: the "CFO" — HRBP = senior, budget owner = bu_budget.
    cfo = await _person(
        db, "P8010", hrbp_id=senior.id, budget_id=bu_budget.id,
    )
    db.add(IdentityOrgMembership(
        id=uuid.uuid4(), person_id=cfo.id, org_node_source_id=_BU,
        org_node_name="Finance", parent_org_node_source_id=None,
        is_primary=True, org_path="PingCAP / Finance",
    ))
    # Leaf members (the operating team): HRBP = team, budget owner = team_budget.
    for emp_no in ("P8021", "P8022", "P8023"):
        m = await _person(
            db, emp_no, hrbp_id=team.id, budget_id=team_budget.id,
        )
        db.add(IdentityOrgMembership(
            id=uuid.uuid4(), person_id=m.id, org_node_source_id=_LEAF,
            org_node_name="Accounting", parent_org_node_source_id=_BU,
            is_primary=True, org_path="PingCAP / Finance / Accounting",
        ))
    await db.flush()


def _target() -> IdentitySyncTarget:
    return IdentitySyncTarget(
        id=uuid.uuid4(), name="hly", slug="hly-hrbp-subtree", kind="huilianyi",
        enabled=False, on_offboard="deactivate", mapping_config={"rules": []},
        filter_config={
            "employee_no_pattern": r"^P\d+$",
            "employment_statuses": ["active"],
            "include_subtree": True,
        },
        schedule_config={"trigger": "manual", "cron_expr": None, "timezone": "UTC"},
        runner_options={},
    )


async def _run(db) -> _CapturingDeptRunner:
    await _seed(db)
    tgt = _target()
    db.add(tgt)
    await db.flush()
    runner = _CapturingDeptRunner()
    await IdentitySyncService(db)._huilianyi_dept_phase_body(
        target=tgt, runner=runner, summary=RunSummary(),
    )
    return runner


@pytest.mark.asyncio
async def test_bu_hrbp_reflects_subtree_not_leadership(db_session):
    """The BU (parent) HRBP is the operating teams' HRBP (subtree plurality
    3:1), NOT the CFO's senior HRBP — even though the CFO is the only DIRECT
    member of the BU node."""
    runner = await _run(db_session)
    assert runner.hrbp[_BU] == _TEAM_HRBP_NO
    assert runner.hrbp[_BU] != _SENIOR_HRBP_NO


@pytest.mark.asyncio
async def test_leaf_hrbp_unchanged(db_session):
    """A leaf dept has no children, so subtree == direct members — its HRBP
    is the team HRBP, exactly as before."""
    runner = await _run(db_session)
    assert runner.hrbp[_LEAF] == _TEAM_HRBP_NO


@pytest.mark.asyncio
async def test_budget_owner_never_comes_from_sf_comp_approver(db_session):
    """Budget Owner (6102) must NOT be derived from
    ``hris_budget_owner_person_id`` (SF 2447 调薪负责人) any more — that is a
    BU-level comp approver, not a department budget holder (2026-09-04 fix;
    see ``test_huilianyi_budget_owner_is_dept_head_2026_09_04`` for the
    positive cases). Nobody in this fixture manages a dept member, so no
    dept has a derivable head and no ancestor supplies one → the role is
    omitted rather than falling back to the 2447 holders."""
    runner = await _run(db_session)
    assert runner.budget[_BU] not in (_BU_BUDGET_NO, _TEAM_BUDGET_NO)
    assert runner.budget[_LEAF] not in (_BU_BUDGET_NO, _TEAM_BUDGET_NO)
    assert runner.budget[_BU] is None
    assert runner.budget[_LEAF] is None


# ── 2026-07-16 audit: restrict-path parity + deterministic tie-break ─────

async def _node(db, code, name, parent, level):
    db.add(IdentityOrgNode(
        id=uuid.uuid4(), org_node_source_id=code, org_node_name=name,
        parent_org_node_source_id=parent, org_level=level,
        org_path=f"PingCAP / {name}", node_status="active",
    ))


async def _member(db, emp_no, dept_code, dept_name, parent, *, hrbp_id=None):
    m = await _person(db, emp_no, hrbp_id=hrbp_id)
    db.add(IdentityOrgMembership(
        id=uuid.uuid4(), person_id=m.id, org_node_source_id=dept_code,
        org_node_name=dept_name, parent_org_node_source_id=parent,
        is_primary=True, org_path=f"PingCAP / {dept_name}",
    ))
    return m


@pytest.mark.asyncio
async def test_restrict_path_bu_hrbp_matches_full_sweep(db_session):
    """The single-person dept-chain ensure (``restrict_dept_codes`` = one
    leaf + its ancestors) must compute the BU's subtree HRBP over the
    FULL roster, not just the chain's members. Pre-fix, the member pool
    was narrowed to ``in_scope_codes``, so a leaf-1 hire's ensure rolled
    the BU plurality over {leadership (2× senior HRBP) + leaf-1 (1× team
    HRBP)} → senior wins → the exact leadership-skew #1941 fixed gets
    RE-pushed, ping-ponging against every full sweep. Post-fix both
    paths agree: team HRBP (4:2 over the whole subtree)."""
    db = db_session
    # Tree: BU 9000 ── leaf 9000-1 (1 member), leaf 9000-2 (3 members).
    await _node(db, "9000", "Sales", None, 1)
    await _node(db, "9000-1", "Sales-JP", "9000", 2)
    await _node(db, "9000-2", "Sales-CN", "9000", 2)
    senior = await _person(db, "P9101")
    team = await _person(db, "P9102")
    # BU direct members = the leadership cluster (2× senior HRBP).
    await _member(db, "P9110", "9000", "Sales", None, hrbp_id=senior.id)
    await _member(db, "P9111", "9000", "Sales", None, hrbp_id=senior.id)
    # Leaf 1: one member (the hire whose ensure restricts to its chain).
    await _member(db, "P9120", "9000-1", "Sales-JP", "9000", hrbp_id=team.id)
    # Leaf 2: three members — OUTSIDE the restricted chain.
    for emp in ("P9130", "P9131", "P9132"):
        await _member(db, emp, "9000-2", "Sales-CN", "9000", hrbp_id=team.id)
    tgt = _target()
    tgt.slug = "hly-hrbp-restrict"
    db.add(tgt)
    await db.flush()

    # Full sweep — the reference rollup.
    full = _CapturingDeptRunner()
    await IdentitySyncService(db)._huilianyi_dept_phase_body(
        target=tgt, runner=full, summary=RunSummary(),
    )
    # Restricted run — a leaf-1 hire's dept-chain ensure. force_reverify
    # bypasses the dept hash-skip so the runner records what THIS run
    # would push (post-fix, the attrs match the full sweep's, so a plain
    # run would legitimately hash-skip and record nothing).
    restricted = _CapturingDeptRunner()
    await IdentitySyncService(db)._huilianyi_dept_phase_body(
        target=tgt, runner=restricted, summary=RunSummary(),
        restrict_dept_codes={"9000", "9000-1"}, force_reverify=True,
    )

    assert full.hrbp["9000"] == "P9102"  # team HRBP, 4:2
    assert restricted.hrbp["9000"] == full.hrbp["9000"], (
        "restricted ensure must not recompute a different (leadership-"
        "skewed) BU HRBP than the full sweep"
    )
    # The restricted run only pushes the chain.
    assert "9000-2" not in restricted.hrbp


@pytest.mark.asyncio
async def test_hrbp_plurality_tie_is_deterministic(db_session):
    """A 1:1 HRBP tie must resolve identically regardless of the member
    row order (``Counter.most_common`` keys on unordered insertion order
    → the HRBP flapped between sweeps). Tie-break: the holder with the
    smallest 工号 wins."""
    db = db_session
    await _node(db, "9100", "TieA", None, 1)
    await _node(db, "9200", "TieB", None, 1)
    hrbp_hi = await _person(db, "P9202")   # larger 工号
    hrbp_lo = await _person(db, "P9201")   # smaller 工号 → must win ties
    # Dept A: hi-holder's member inserted FIRST; dept B: lo-holder's first.
    await _member(db, "P9210", "9100", "TieA", None, hrbp_id=hrbp_hi.id)
    await _member(db, "P9211", "9100", "TieA", None, hrbp_id=hrbp_lo.id)
    await _member(db, "P9220", "9200", "TieB", None, hrbp_id=hrbp_lo.id)
    await _member(db, "P9221", "9200", "TieB", None, hrbp_id=hrbp_hi.id)
    tgt = _target()
    tgt.slug = "hly-hrbp-tie"
    db.add(tgt)
    await db.flush()

    runner = _CapturingDeptRunner()
    await IdentitySyncService(db)._huilianyi_dept_phase_body(
        target=tgt, runner=runner, summary=RunSummary(),
    )

    assert runner.hrbp["9100"] == "P9201"
    assert runner.hrbp["9200"] == "P9201"