"""SCIM group push — dept tree → Groups for AWS IAM Identity Center.

Two layers:

* ``ScimRunner`` group methods — wire decision tree pinned against a
  scripted ``_request`` (externalId-first adoption with displayName
  fallback, 409-create race, replace-members 404 sentinel, idempotent
  single add).
* ``_run_scim_group_phase`` — DB-integration: member map from the
  target's own user-state rows, hash skip, duplicate-displayName
  guard, out-of-scope groups marked excluded (never deleted).
"""
from __future__ import annotations

import uuid
from typing import Any

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

from app.domains.identity_sync.models import (
    IdentitySyncDepartmentState,
    IdentitySyncState,
    IdentitySyncTarget,
)
from app.domains.identity_sync.presets import AWS_IDC_MAPPING_RULES
from app.domains.identity_sync.runners.scim import ScimRunner
from app.domains.identity_sync.service import IdentitySyncService
from app.identity.person_models import (
    IdentityOrgMembership,
    IdentityOrgNode,
    IdentityPerson,
)

pytestmark = pytest.mark.asyncio


def _runner() -> ScimRunner:
    return ScimRunner(base_url="https://idc.example/scim/v2", api_token="t")


class _Scripted:
    def __init__(self, script: list[tuple[str, str, tuple[int, Any, str]]]):
        self.script = list(script)
        self.calls: list[tuple[str, str, dict | None, dict | None]] = []

    async def __call__(self, method, path, *, json_body=None, params=None,
                       retry_on_429=2):
        self.calls.append((method, path, json_body, params))
        for i, (m, prefix, resp) in enumerate(self.script):
            if m == method and path.startswith(prefix):
                self.script.pop(i)
                return resp
        raise AssertionError(f"unexpected call {method} {path} {params}")


def _wire(runner: ScimRunner, script) -> _Scripted:
    calls = _Scripted(script)
    runner._request = calls  # type: ignore[method-assign]
    return calls


LIST_EMPTY = {"schemas": ["urn:ietf:params:scim:api:messages:2.0:ListResponse"],
              "Resources": [], "totalResults": 0}


def _list_with(gid: str):
    return {"schemas": ["urn:ietf:params:scim:api:messages:2.0:ListResponse"],
            "Resources": [{"id": gid, "displayName": "x"}], "totalResults": 1}


class TestGroupWireMethods:
    async def test_find_external_id_first(self):
        r = _runner()
        calls = _wire(r, [("GET", "Groups", (200, _list_with("g-1"), ""))])
        gid = await r.find_group_id(display_name="Finance", external_code="1809819")
        assert gid == "g-1"
        # Only ONE lookup — externalId hit short-circuits.
        assert len(calls.calls) == 1
        assert 'externalId eq "1809819"' in calls.calls[0][3]["filter"]

    async def test_find_falls_back_to_display_name(self):
        r = _runner()
        calls = _wire(r, [
            ("GET", "Groups", (200, LIST_EMPTY, "")),          # externalId miss
            ("GET", "Groups", (200, _list_with("g-2"), "")),   # displayName hit
        ])
        gid = await r.find_group_id(display_name="Finance", external_code="1809819")
        assert gid == "g-2"
        assert 'displayName eq "Finance"' in calls.calls[1][3]["filter"]

    async def test_filter_value_quoted(self):
        r = _runner()
        calls = _wire(r, [("GET", "Groups", (200, LIST_EMPTY, ""))])
        await r.find_group_id(display_name='R"D Team')
        assert '"R\\"D Team"' in calls.calls[0][3]["filter"]

    async def test_create_409_race_adopts(self):
        r = _runner()
        _wire(r, [
            ("POST", "Groups", (409, None, "duplicate")),
            ("GET", "Groups", (200, _list_with("g-3"), "")),
        ])
        gid, err = await r.create_group(display_name="Finance",
                                        external_code="1809819")
        assert gid == "g-3" and err is None

    async def test_members_404_sentinel(self):
        # The current-membership GET 404 → group vanished server-side.
        r = _runner()
        _wire(r, [("GET", "Groups/g-9", (404, None, "gone"))])
        err = await r.replace_group_members(group_id="g-9", member_ids=["u1"])
        assert err == "group_absent"

    async def test_members_diff_add_and_remove(self):
        # IdC rejects op:replace on members — reconcile via add/remove
        # deltas against the current set read from GET /Groups/{id}.
        r = _runner()
        calls = _wire(r, [
            ("GET", "Groups/g-1", (200, {
                "id": "g-1", "displayName": "Eng",
                "members": [{"value": "keep"}, {"value": "gone"}],
            }, "")),
            ("PATCH", "Groups/g-1", (204, None, "")),   # add newcomer
            ("PATCH", "Groups/g-1", (204, None, "")),   # remove leaver
        ])
        err = await r.replace_group_members(
            group_id="g-1", member_ids=["keep", "new"], display_name="Eng",
        )
        assert err is None
        # displayName already "Eng" → no displayName PATCH, just diff ops.
        add = calls.calls[1][2]["Operations"][0]
        assert add["op"] == "add" and add["value"] == [{"value": "new"}]
        rem = calls.calls[2][2]["Operations"][0]
        assert rem["op"] == "remove"
        # IdC's documented per-member removal syntax (filter path).
        assert rem["path"] == 'members[value eq "gone"]'

    async def test_members_remove_falls_back_to_value_list(self):
        # If IdC rejects the filter-path remove, retry the value-list
        # form so the leaver is still dropped.
        r = _runner()
        calls = _wire(r, [
            ("GET", "Groups/g-1", (200, {
                "id": "g-1", "displayName": "X",
                "members": [{"value": "gone"}],
            }, "")),
            ("PATCH", "Groups/g-1", (400, None, "filter path not supported")),
            ("PATCH", "Groups/g-1", (204, None, "")),
        ])
        err = await r.replace_group_members(
            group_id="g-1", member_ids=[], display_name="X",
        )
        assert err is None
        op1 = calls.calls[1][2]["Operations"][0]
        assert op1["op"] == "remove" and op1["path"] == 'members[value eq "gone"]'
        op2 = calls.calls[2][2]["Operations"][0]
        assert op2["op"] == "remove" and op2["path"] == "members"
        assert op2["value"] == [{"value": "gone"}]

    async def test_members_multi_remove_is_batched_in_one_patch(self):
        # 2026-07-07 perf: a bulk membership drop must be ONE batched PATCH
        # (multiple remove ops), not one serial PATCH-per-member with
        # per-call retry — otherwise a reorg dropping hundreds of members
        # stalls the sync target for minutes.
        r = _runner()
        calls = _wire(r, [
            ("GET", "Groups/g-1", (200, {
                "id": "g-1", "displayName": "X",
                "members": [{"value": "a"}, {"value": "b"}, {"value": "c"}],
            }, "")),
            ("PATCH", "Groups/g-1", (204, None, "")),  # single batched remove
        ])
        err = await r.replace_group_members(
            group_id="g-1", member_ids=[], display_name="X",
        )
        assert err is None
        patches = [c for c in calls.calls if c[0] == "PATCH"]
        assert len(patches) == 1, "multi-member removes must be one batched PATCH"
        ops = patches[0][2]["Operations"]
        assert len(ops) == 3 and all(o["op"] == "remove" for o in ops)
        assert {o["path"] for o in ops} == {
            'members[value eq "a"]',
            'members[value eq "b"]',
            'members[value eq "c"]',
        }

    async def test_members_add_batch_409_falls_back_per_member(self):
        # IdC eventual consistency: the GET missed a just-added member,
        # so the batch add 409s (atomic — none land). Fall back to
        # per-member adds, treating "already a member" as success so the
        # genuinely-new member still lands and the dept doesn't fail.
        r = _runner()
        calls = _wire(r, [
            ("GET", "Groups/g-1", (200, {
                "id": "g-1", "displayName": "X", "members": [],
            }, "")),
            ("PATCH", "Groups/g-1", (409, None, "uniqueness")),  # batch rejected
            ("PATCH", "Groups/g-1", (409, None, "uniqueness")),  # existing: ok
            ("PATCH", "Groups/g-1", (204, None, "")),            # new: lands
        ])
        err = await r.replace_group_members(
            group_id="g-1", member_ids=["existing", "new"], display_name="X",
        )
        assert err is None
        assert len(calls.calls) == 4
        assert calls.calls[1][2]["Operations"][0]["value"] == [
            {"value": "existing"}, {"value": "new"},
        ]
        assert calls.calls[2][2]["Operations"][0]["value"] == [{"value": "existing"}]
        assert calls.calls[3][2]["Operations"][0]["value"] == [{"value": "new"}]

    async def test_members_add_per_member_real_400_propagates(self):
        # A genuine (non-uniqueness) 400 on a per-member add — e.g. a
        # member id IdC rejects — must surface as a non-fatal member
        # error, NOT be swallowed and NOT return "group_absent".
        r = _runner()
        _wire(r, [
            ("GET", "Groups/g-1", (200, {
                "id": "g-1", "displayName": "X", "members": [],
            }, "")),
            ("PATCH", "Groups/g-1", (400, None, "uniqueness")),      # batch → fallback
            ("PATCH", "Groups/g-1", (400, None, "invalid member id")),  # real bad id
        ])
        err = await r.replace_group_members(
            group_id="g-1", member_ids=["bad"], display_name="X",
        )
        assert err is not None and err != "group_absent"
        assert "member ops failed" in err and "add bad → 400" in err

    async def test_member_add_404_is_not_group_absent(self):
        # THE regression: a 404 on a MEMBER add (bad/absent user id) when
        # the group provably exists (GET 200) must NOT be reported as
        # group_absent — that misclassification cleared the valid group
        # id and looped forever, never landing the other 6 of 7 members
        # (prod: Dataplane Fabric / Clinic Service). Good members land;
        # the bad one is isolated as a non-fatal, actionable error.
        # to_add is sorted, so "aaa-good" is attempted before "zzz-bad";
        # the script's per-member responses follow that order.
        r = _runner()
        calls = _wire(r, [
            ("GET", "Groups/g-1", (200, {
                "id": "g-1", "displayName": "X", "members": [],
            }, "")),
            ("PATCH", "Groups/g-1", (404, None, "member not found")),  # batch 404
            ("PATCH", "Groups/g-1", (204, None, "")),                  # aaa-good lands
            ("PATCH", "Groups/g-1", (404, None, "user id absent")),    # zzz-bad fails
        ])
        err = await r.replace_group_members(
            group_id="g-1", member_ids=["aaa-good", "zzz-bad"], display_name="X",
        )
        assert err != "group_absent"
        assert err is not None and "add zzz-bad → 404" in err
        # the good member WAS added per-member (not aborted by the bad one)
        added = [c for c in calls.calls if c[0] == "PATCH"
                 and c[2]["Operations"][0].get("op") == "add"
                 and c[2]["Operations"][0]["value"] == [{"value": "aaa-good"}]]
        assert len(added) == 1

    async def test_members_add_chunks_over_idc_cap(self):
        # IdC caps membership changes per PATCH at 100; 130 newcomers
        # (current empty) = two add chunks (100 + 30), never a replace.
        r = _runner()
        calls = _wire(r, [
            ("GET", "Groups/g-big", (200, {
                "id": "g-big", "displayName": "Big", "members": [],
            }, "")),
            ("PATCH", "Groups/g-big", (204, None, "")),
            ("PATCH", "Groups/g-big", (204, None, "")),
        ])
        members = [f"u{i}" for i in range(130)]
        err = await r.replace_group_members(
            group_id="g-big", member_ids=members, display_name="Big",
        )
        assert err is None
        assert calls.calls[0][0] == "GET"
        add1 = calls.calls[1][2]["Operations"][0]
        assert add1["op"] == "add" and len(add1["value"]) == 100
        add2 = calls.calls[2][2]["Operations"][0]
        assert add2["op"] == "add" and len(add2["value"]) == 30

    async def test_display_name_reject_still_lands_membership(self):
        # displayName drift is its own scalar replace; if IdC rejects it
        # the membership diff still applies (access > label).
        r = _runner()
        calls = _wire(r, [
            ("GET", "Groups/g-1", (200, {
                "id": "g-1", "displayName": "Old", "members": [],
            }, "")),
            ("PATCH", "Groups/g-1", (400, None, "displayName not patchable")),
            ("PATCH", "Groups/g-1", (204, None, "")),
        ])
        err = await r.replace_group_members(
            group_id="g-1", member_ids=["u1"], display_name="New Name",
        )
        assert err is not None and err.startswith("name_patch_failed")
        dn = calls.calls[1][2]["Operations"][0]
        assert dn["path"] == "displayName"
        add = calls.calls[2][2]["Operations"][0]
        assert add["op"] == "add" and add["path"] == "members"

    async def test_add_member_idempotent_on_409(self):
        r = _runner()
        _wire(r, [("PATCH", "Groups/g-1", (409, None, "uniqueness"))])
        assert await r.add_group_member(group_id="g-1", member_id="u1") is None


# ── Phase orchestration (DB) ─────────────────────────────────────────
async def _node(db, *, src_id, name, parent=None, level=1):
    db.add(IdentityOrgNode(
        id=uuid.uuid4(), source_system="successfactors",
        org_node_source_id=src_id, org_node_name=name,
        parent_org_node_source_id=parent, org_level=level,
        node_status="active",
    ))
    await db.flush()


async def _person_in(db, *, email, dept_src, employee_no):
    pid = uuid.uuid4()
    db.add(IdentityPerson(
        id=pid, work_email=email, display_name=email.split("@")[0],
        employment_status="active", source_system="successfactors",
        hris_person_id=f"sf-{email}", employee_no=employee_no,
        is_china_employee=True, country="CHN",
    ))
    await db.flush()
    db.add(IdentityOrgMembership(
        id=uuid.uuid4(), person_id=pid, source_system="successfactors",
        org_node_source_id=dept_src, org_node_name=dept_src,
        is_primary=True, membership_status="active",
    ))
    await db.flush()
    return pid


async def _scim_target(db, *, push_groups=True) -> IdentitySyncTarget:
    tgt = IdentitySyncTarget(
        id=uuid.uuid4(),
        name="Test AWS IdC", slug="test-aws-idc", kind="scim",
        enabled=False, on_offboard="deactivate",
        mapping_config={"rules": [dict(r) for r in AWS_IDC_MAPPING_RULES]},
        filter_config={
            "include_offboarded": False, "departments": [],
            "include_subtree": True, "countries": [],
            "employment_statuses": ["active", "on_leave"],
            "job_title_includes": [], "person_ids": [], "mode": "and",
        },
        schedule_config={"trigger": "manual", "cron_expr": None,
                         "timezone": "UTC"},
        runner_options={
            "base_url": "https://idc.example/scim/v2",
            "auth_scheme": "bearer", "extra_headers": {},
            "verify_ssl": True, "timeout_seconds": 30,
            "rate_limit_per_minute": 60, "push_groups": push_groups,
        },
    )
    db.add(tgt)
    await db.flush()
    return tgt


class _PhaseRunner(ScimRunner):
    """ScimRunner with the wire methods recorded + scripted in-memory."""

    def __init__(self):
        super().__init__(base_url="https://idc.example/scim/v2", api_token="t")
        self.existing: dict[str, str] = {}      # external_code → group id
        self.members: dict[str, list[str]] = {}  # group id → members
        self.created: list[str] = []

    async def find_group_id(self, *, display_name, external_code=None):
        return self.existing.get(external_code or display_name)

    async def create_group(self, *, display_name, external_code=None):
        gid = f"g-{external_code or display_name}"
        self.existing[external_code or display_name] = gid
        self.created.append(gid)
        return gid, None

    async def replace_group_members(self, *, group_id, member_ids,
                                     display_name=None):
        self.members[group_id] = list(member_ids)
        self.names = getattr(self, "names", {})
        self.names[group_id] = display_name
        return None


class TestGroupPhase:
    async def test_groups_created_with_direct_members_and_hash_skip(
        self, db_session: AsyncSession,
    ):
        await _node(db_session, src_id="ROOT", name="PingCAP", level=1)
        await _node(db_session, src_id="FIN", name="Finance", parent="ROOT", level=2)
        p1 = await _person_in(db_session, email="a@pingcap.com",
                              dept_src="FIN", employee_no="P00001")
        p2 = await _person_in(db_session, email="b@pingcap.com",
                              dept_src="ROOT", employee_no="P00002")
        target = await _scim_target(db_session)
        # User-state rows: the sweep already pushed both users.
        for pid, sid in ((p1, "scim-u1"), (p2, "scim-u2")):
            db_session.add(IdentitySyncState(
                id=uuid.uuid4(), target_id=target.id, person_id=pid,
                external_id=sid, last_status="success",
            ))
        await db_session.commit()

        svc = IdentitySyncService(db_session)
        runner = _PhaseRunner()
        summary = await svc._run_scim_group_phase(target=target, runner=runner)
        assert summary.dept_failed == 0
        assert summary.dept_created == 2
        # DIRECT members only — b is in ROOT, not rolled into FIN.
        assert runner.members["g-FIN"] == ["scim-u1"]
        assert runner.members["g-ROOT"] == ["scim-u2"]
        await db_session.commit()

        # Second run with identical membership → hash skip, no writes.
        runner2 = _PhaseRunner()
        runner2.existing = dict(runner.existing)
        summary2 = await svc._run_scim_group_phase(target=target, runner=runner2)
        assert summary2.dept_failed == 0
        assert summary2.dept_created == 0 and summary2.dept_updated == 0
        assert runner2.members == {}  # nothing re-pushed

    async def test_failed_user_state_with_external_id_keeps_membership(
        self, db_session: AsyncSession,
    ):
        # Membership mirrors ACCOUNT EXISTENCE: a transient upsert
        # failure must NOT strip the person from their AWS group
        # (replace-all would revoke their permission sets).
        await _node(db_session, src_id="ROOT", name="PingCAP", level=1)
        p1 = await _person_in(db_session, email="ok@pingcap.com",
                              dept_src="ROOT", employee_no="P00011")
        p2 = await _person_in(db_session, email="flap@pingcap.com",
                              dept_src="ROOT", employee_no="P00012")
        p3 = await _person_in(db_session, email="gone@pingcap.com",
                              dept_src="ROOT", employee_no="P00013")
        target = await _scim_target(db_session)
        rows = [
            (p1, "scim-ok", "success"),
            (p2, "scim-flap", "failed"),       # transient 500 last sweep
            (p3, "scim-gone", "deactivated"),  # we shut this account
        ]
        for pid, sid, st in rows:
            db_session.add(IdentitySyncState(
                id=uuid.uuid4(), target_id=target.id, person_id=pid,
                external_id=sid, last_status=st,
            ))
        await db_session.commit()

        svc = IdentitySyncService(db_session)
        runner = _PhaseRunner()
        await svc._run_scim_group_phase(target=target, runner=runner)
        # failed-but-existing stays; deactivated leaves.
        assert sorted(runner.members["g-ROOT"]) == ["scim-flap", "scim-ok"]

    async def test_duplicate_display_name_disambiguated_by_parent(
        self, db_session: AsyncSession,
    ):
        # Two real "Sales" depts under different business groups. AWS IdC
        # group displayName is a FLAT unique namespace, so instead of
        # failing both (the old behaviour), qualify each by its immediate
        # parent and sync them as distinct groups.
        await _node(db_session, src_id="ROOT", name="PingCAP", level=1)
        await _node(db_session, src_id="JP",
                    name="Japan Business Group", parent="ROOT", level=2)
        await _node(db_session, src_id="EMEA", name="EMEA",
                    parent="ROOT", level=2)
        await _node(db_session, src_id="S1", name="Sales",
                    parent="JP", level=3)
        await _node(db_session, src_id="S2", name="Sales",
                    parent="EMEA", level=3)
        pa = await _person_in(db_session, email="x@pingcap.com",
                              dept_src="S1", employee_no="P00003")
        pb = await _person_in(db_session, email="y@pingcap.com",
                              dept_src="S2", employee_no="P00004")
        target = await _scim_target(db_session)
        for pid, sid in ((pa, "scim-x"), (pb, "scim-y")):
            db_session.add(IdentitySyncState(
                id=uuid.uuid4(), target_id=target.id, person_id=pid,
                external_id=sid, last_status="success",
            ))
        await db_session.commit()

        svc = IdentitySyncService(db_session)
        runner = _PhaseRunner()
        summary = await svc._run_scim_group_phase(target=target, runner=runner)
        # No failures — both same-named depts sync as their own group.
        assert summary.dept_failed == 0
        assert runner.members["g-S1"] == ["scim-x"]
        assert runner.members["g-S2"] == ["scim-y"]
        # displayName qualified by immediate parent → globally unique.
        assert runner.names["g-S1"] == "Sales (Japan Business Group)"
        assert runner.names["g-S2"] == "Sales (EMEA)"
        # Non-colliding ancestor keeps its plain name.
        assert runner.names["g-JP"] == "Japan Business Group"
        assert not any("circuit breaker" in e for e in summary.errors)

    async def test_duplicate_display_name_same_parent_falls_back_to_code(
        self, db_session: AsyncSession,
    ):
        # Pathological: same name AND same parent → parent-qualification
        # still collides, so append the HRIS dept code to guarantee every
        # pushed displayName is unique.
        await _node(db_session, src_id="ROOT", name="PingCAP", level=1)
        await _node(db_session, src_id="ENG1", name="Engineering",
                    parent="ROOT", level=2)
        await _node(db_session, src_id="ENG2", name="Engineering",
                    parent="ROOT", level=2)
        pa = await _person_in(db_session, email="x@pingcap.com",
                              dept_src="ENG1", employee_no="P00003")
        pb = await _person_in(db_session, email="y@pingcap.com",
                              dept_src="ENG2", employee_no="P00004")
        target = await _scim_target(db_session)
        for pid, sid in ((pa, "scim-x"), (pb, "scim-y")):
            db_session.add(IdentitySyncState(
                id=uuid.uuid4(), target_id=target.id, person_id=pid,
                external_id=sid, last_status="success",
            ))
        await db_session.commit()

        svc = IdentitySyncService(db_session)
        runner = _PhaseRunner()
        summary = await svc._run_scim_group_phase(target=target, runner=runner)
        assert summary.dept_failed == 0
        assert runner.names["g-ENG1"] == "Engineering (PingCAP) [ENG1]"
        assert runner.names["g-ENG2"] == "Engineering (PingCAP) [ENG2]"
        assert not any("circuit breaker" in e for e in summary.errors)

    async def test_out_of_scope_group_excluded_never_deleted(
        self, db_session: AsyncSession,
    ):
        await _node(db_session, src_id="ROOT", name="PingCAP", level=1)
        p = await _person_in(db_session, email="z@pingcap.com",
                             dept_src="ROOT", employee_no="P00005")
        target = await _scim_target(db_session)
        db_session.add(IdentitySyncState(
            id=uuid.uuid4(), target_id=target.id, person_id=p,
            external_id="scim-z", last_status="success",
        ))
        # A group from a dept that no longer exists in HRIS.
        db_session.add(IdentitySyncDepartmentState(
            id=uuid.uuid4(), target_id=target.id, dept_code="GONE",
            external_id="g-GONE", last_status="success",
        ))
        await db_session.commit()

        svc = IdentitySyncService(db_session)
        runner = _PhaseRunner()
        await svc._run_scim_group_phase(target=target, runner=runner)
        await db_session.commit()

        st = (await db_session.execute(
            select(IdentitySyncDepartmentState).where(
                IdentitySyncDepartmentState.target_id == target.id,
                IdentitySyncDepartmentState.dept_code == "GONE",
            )
        )).scalar_one()
        assert st.last_status == "excluded"
        # The vendor group id is retained for the admin to retire.
        assert st.external_id == "g-GONE"


class TestPushGroupsValidation:
    async def test_push_groups_round_trips_through_options(self):
        from app.domains.identity_sync.service import _validate_runner_options

        opts = _validate_runner_options("scim", {
            "base_url": "https://idc.example/scim/v2",
            "push_groups": True,
        })
        assert opts["push_groups"] is True
        opts = _validate_runner_options("scim", {
            "base_url": "https://idc.example/scim/v2",
        })
        assert opts["push_groups"] is False


class _EnsureRunner(_PhaseRunner):
    """PhaseRunner + a recording ``add_group_member`` so the event-path
    single-member add can be asserted without a wire call."""

    def __init__(self):
        super().__init__()
        self.added: list[tuple[str, str]] = []

    async def add_group_member(self, *, group_id, member_id):
        self.added.append((group_id, member_id))
        return None


class TestSinglePersonGroupEnsure:
    """Event-path (single-person) group handling — a new hire in a dept
    whose SCIM Group hasn't been swept yet must still land in their group
    immediately (the group is created on the spot), not silently skipped
    until the nightly sweep.
    """

    async def test_add_member_creates_missing_group_then_adds(
        self, db_session: AsyncSession,
    ):
        await _node(db_session, src_id="ROOT", name="PingCAP", level=1)
        await _node(db_session, src_id="ENG", name="Engineering",
                    parent="ROOT", level=2)
        # An in-scope member so the dept is in scope for the ensure.
        await _person_in(db_session, email="new@pingcap.com",
                         dept_src="ENG", employee_no="P00021")
        target = await _scim_target(db_session)
        await db_session.commit()

        svc = IdentitySyncService(db_session)
        runner = _EnsureRunner()
        # No dept_state row for ENG yet (never swept).
        await svc._add_member_to_dept_group(
            target=target, runner=runner,
            dept_code="ENG", scim_user_id="scim-new",
        )
        await db_session.commit()

        # Group created (externalId=ENG) AND the member added to it.
        assert runner.created == ["g-ENG"]
        assert ("g-ENG", "scim-new") in runner.added
        # Group id persisted so the nightly sweep + later adds reuse it.
        st = (await db_session.execute(
            select(IdentitySyncDepartmentState).where(
                IdentitySyncDepartmentState.target_id == target.id,
                IdentitySyncDepartmentState.dept_code == "ENG",
            )
        )).scalar_one()
        assert st.external_id == "g-ENG"

    async def test_add_member_reuses_existing_group_no_create(
        self, db_session: AsyncSession,
    ):
        await _node(db_session, src_id="ROOT", name="PingCAP", level=1)
        await _node(db_session, src_id="ENG", name="Engineering",
                    parent="ROOT", level=2)
        await _person_in(db_session, email="e@pingcap.com",
                         dept_src="ENG", employee_no="P00022")
        target = await _scim_target(db_session)
        db_session.add(IdentitySyncDepartmentState(
            id=uuid.uuid4(), target_id=target.id, dept_code="ENG",
            external_id="g-existing", last_status="success",
        ))
        await db_session.commit()

        svc = IdentitySyncService(db_session)
        runner = _EnsureRunner()
        await svc._add_member_to_dept_group(
            target=target, runner=runner,
            dept_code="ENG", scim_user_id="scim-e",
        )
        # Group already adopted → NO create, member added to existing id.
        assert runner.created == []
        assert ("g-existing", "scim-e") in runner.added

    async def test_ensure_skips_out_of_scope_dept(
        self, db_session: AsyncSession,
    ):
        # ENG has a member but the filter targets OTHER only → ENG is out
        # of scope, so no group is created for it.
        await _node(db_session, src_id="ROOT", name="PingCAP", level=1)
        await _node(db_session, src_id="ENG", name="Engineering",
                    parent="ROOT", level=2)
        await _node(db_session, src_id="OTHER", name="Other",
                    parent="ROOT", level=2)
        await _person_in(db_session, email="eng@pingcap.com",
                         dept_src="ENG", employee_no="P00023")
        target = await _scim_target(db_session)
        target.filter_config = {**target.filter_config, "departments": ["OTHER"]}
        await db_session.commit()

        svc = IdentitySyncService(db_session)
        runner = _EnsureRunner()
        gid = await svc._ensure_scim_dept_group(
            target=target, runner=runner, dept_code="ENG",
        )
        assert gid is None
        assert runner.created == []


class TestPersonDeptChain:
    async def test_chain_is_leaf_plus_ancestors(
        self, db_session: AsyncSession,
    ):
        await _node(db_session, src_id="ROOT", name="PingCAP", level=1)
        await _node(db_session, src_id="BG", name="NA & EMEA",
                    parent="ROOT", level=2)
        await _node(db_session, src_id="PS", name="PreSales",
                    parent="BG", level=3)
        pid = await _person_in(db_session, email="r@pingcap.com",
                               dept_src="PS", employee_no="P00024")
        await db_session.commit()

        svc = IdentitySyncService(db_session)
        chain = await svc._person_dept_chain(pid)
        # Leaf + every ancestor up the active tree.
        assert chain == {"PS", "BG", "ROOT"}

    async def test_chain_empty_without_primary_membership(
        self, db_session: AsyncSession,
    ):
        pid = uuid.uuid4()
        db_session.add(IdentityPerson(
            id=pid, work_email="nomem@pingcap.com", display_name="nomem",
            employment_status="active", source_system="successfactors",
            hris_person_id="sf-nomem", employee_no="P00025",
        ))
        await db_session.commit()
        svc = IdentitySyncService(db_session)
        assert await svc._person_dept_chain(pid) == set()