"""Shared HRBP / Budget-Owner resolver — ``apply_hris_role_relationships``. This resolver feeds ``identity_persons.hris_hrbp_person_id`` / ``hris_budget_owner_person_id`` from SF ``EmpJobRelationships`` (工作关系, relType 2448=HRBP, 2447=调薪负责人→Budget Owner). Those columns are what the 汇联易 department sync reads to fill the per-BU HRBP / Budget-Owner roles. Regression context (2026-06-16): the resolution originally lived ONLY inside the dormant ``SuccessFactorsSyncService.sync_users`` path, which nothing in production calls — the real admin/cron sync runs through ``hr.SuccessFactorsLifecycleService.confirm_batch``. So the columns stayed NULL for everyone after a real sync and the 汇联易 roles never populated. The fix extracted this shared helper and wired confirm_batch to call it over the WHOLE active roster (not just the per-batch diff). These tests pin the helper's contract; ``test_hr/test_confirm_batch_refreshes_hris_roles`` pins the wiring. """ from __future__ import annotations import uuid import pytest from app.identity.person_models import IdentityPerson from app.identity.successfactors_sync import apply_hris_role_relationships class _FakeClient: def __init__(self, rels: list[dict]): self._rels = rels self.calls = 0 async def list_job_relationships(self, relationship_types): self.calls += 1 return self._rels class _BoomClient: async def list_job_relationships(self, relationship_types): raise RuntimeError("EmpJobRelationships not provisioned for this tenant") def _person(hris_no: str, name: str) -> IdentityPerson: return IdentityPerson( id=uuid.uuid4(), source_system="successfactors", hris_person_id=hris_no, employee_no=f"E{hris_no}", work_email=f"{hris_no.lower()}@x.test", display_name=name, employment_status="active", ) @pytest.mark.asyncio async def test_resolves_hrbp_and_budget_owner_over_roster(db_session): hrbp = _person("P01115", "朱赟") owner = _person("P0843", "倪捷") target = _person("P0989", "冯帅") db_session.add_all([hrbp, owner, target]) await db_session.flush() client = _FakeClient([ {"userId": "P0989", "relationshipType": "2448", "relUserId": "P01115"}, {"userId": "P0989", "relationshipType": "2447", "relUserId": "P0843"}, ]) n = await apply_hris_role_relationships( db_session, client, [hrbp, owner, target], ) assert n == 1 assert target.hris_hrbp_person_id == hrbp.id assert target.hris_budget_owner_person_id == owner.id # People with no declared relationships stay NULL. assert hrbp.hris_hrbp_person_id is None assert owner.hris_budget_owner_person_id is None @pytest.mark.asyncio async def test_resolves_target_not_in_passed_roster_via_db(db_session): """The relUserId may point at a person NOT in the caller's ``people`` list (e.g. confirm_batch passes only this instance's roster but the HRBP lives in another instance). The helper DB-loads missing targets.""" hrbp = _person("P01115", "朱赟") target = _person("P0989", "冯帅") db_session.add_all([hrbp, target]) await db_session.flush() client = _FakeClient([ {"userId": "P0989", "relationshipType": "2448", "relUserId": "P01115"}, ]) # Pass ONLY the target — the HRBP must still resolve from the DB. n = await apply_hris_role_relationships(db_session, client, [target]) assert n == 1 assert target.hris_hrbp_person_id == hrbp.id @pytest.mark.asyncio async def test_self_reference_hrbp_is_dropped_owner_is_kept(db_session): """You can't be your own HR partner, but a senior lead CAN be their own comp/budget owner — only HRBP guards against self-reference.""" p = _person("P0001", "Senior Lead") db_session.add(p) await db_session.flush() client = _FakeClient([ {"userId": "P0001", "relationshipType": "2448", "relUserId": "P0001"}, {"userId": "P0001", "relationshipType": "2447", "relUserId": "P0001"}, ]) await apply_hris_role_relationships(db_session, client, [p]) assert p.hris_hrbp_person_id is None assert p.hris_budget_owner_person_id == p.id @pytest.mark.asyncio async def test_empty_pull_is_noop_does_not_clear(db_session): """An empty (non-erroring) pull must NOT wipe existing pointers — on a whole-roster pass an empty result is "no data this run", not "every relationship removed". This is the 2026-06-16 drain-bug fix (382 → 2 populated when empty pulls cleared the whole roster).""" hrbp = _person("P01115", "朱赟") target = _person("P0989", "冯帅") target.hris_hrbp_person_id = hrbp.id # pre-existing pointer db_session.add_all([hrbp, target]) await db_session.flush() client = _FakeClient([]) # empty pull n = await apply_hris_role_relationships(db_session, client, [hrbp, target]) assert n == 0 # nothing touched assert target.hris_hrbp_person_id == hrbp.id # preserved, NOT cleared @pytest.mark.asyncio async def test_partial_pull_does_not_clear_people_absent_from_it(db_session): """The core drain-bug guard: a pull that omits some people (partial / paginated-short result) must leave THOSE people's columns untouched, only updating the ones it actually returned.""" hrbp = _person("P01115", "朱赟") in_pull = _person("P0989", "冯帅") absent = _person("P0416", "孟凡生") # Both already have a pointer from a prior good sync. in_pull.hris_hrbp_person_id = hrbp.id absent.hris_hrbp_person_id = hrbp.id db_session.add_all([hrbp, in_pull, absent]) await db_session.flush() # Pull only mentions P0989 (e.g. P0416's row was on a dropped page). client = _FakeClient([ {"userId": "P0989", "relationshipType": "2448", "relUserId": "P01115"}, ]) await apply_hris_role_relationships( db_session, client, [hrbp, in_pull, absent], ) assert in_pull.hris_hrbp_person_id == hrbp.id # still set assert absent.hris_hrbp_person_id == hrbp.id # NOT cleared (absent from pull) @pytest.mark.asyncio async def test_present_in_pull_with_one_type_leaves_other_column_untouched(db_session): """A person whose pull row carries only HRBP (no Budget-Owner row this pull) keeps their existing Budget-Owner pointer — we write only the relationship types actually returned.""" hrbp = _person("P01115", "朱赟") owner = _person("P0843", "倪捷") target = _person("P0989", "冯帅") target.hris_budget_owner_person_id = owner.id # pre-existing BO db_session.add_all([hrbp, owner, target]) await db_session.flush() client = _FakeClient([ {"userId": "P0989", "relationshipType": "2448", "relUserId": "P01115"}, ]) await apply_hris_role_relationships( db_session, client, [hrbp, owner, target], ) assert target.hris_hrbp_person_id == hrbp.id # newly set assert target.hris_budget_owner_person_id == owner.id # untouched @pytest.mark.asyncio async def test_unresolvable_target_in_pull_clears_that_pointer(db_session): """A relationship that IS in the pull but whose relUserId no longer resolves to a live person (HRBP left the company) clears that one pointer — that's a present-in-pull resolution miss, not an absence.""" target = _person("P0989", "冯帅") target.hris_hrbp_person_id = uuid.uuid4() # stale pointer db_session.add(target) await db_session.flush() # Pull declares an HRBP whose hris id matches no live person. client = _FakeClient([ {"userId": "P0989", "relationshipType": "2448", "relUserId": "P-GONE"}, ]) await apply_hris_role_relationships(db_session, client, [target]) assert target.hris_hrbp_person_id is None @pytest.mark.asyncio async def test_pull_failure_is_best_effort_and_leaves_columns_untouched(db_session): hrbp = _person("P01115", "朱赟") target = _person("P0989", "冯帅") target.hris_hrbp_person_id = hrbp.id db_session.add_all([hrbp, target]) await db_session.flush() n = await apply_hris_role_relationships(db_session, _BoomClient(), [hrbp, target]) assert n == 0 # nothing touched # The pre-existing pointer must survive a failed pull (don't wipe on error). assert target.hris_hrbp_person_id == hrbp.id @pytest.mark.asyncio async def test_empty_roster_skips_pull(db_session): client = _FakeClient([{"userId": "X", "relationshipType": "2448", "relUserId": "Y"}]) n = await apply_hris_role_relationships(db_session, client, []) assert n == 0 assert client.calls == 0 # no SF call when there's nobody to assign