"""Unit tests for FeishuContactRunner + FeishuDepartmentRunner. Verified against real PingCAP Feishu tenant 2026-05-28 — the design under test is **lookup-and-adopt by name** for depts + **slice-replace merge on department_ids** for users + **custom_attrs catalog translation** for the wire format. Covers: * Department runner — adopt-by-name (PATCH existing od-XXX), POST create when name not in catalog, soft_delete idempotent on absent, leader translation failure not blocking, parent_dept_code → od-XXX via helix_dept_to_external index with root fallback. * User runner — employee_no primary lookup, email fallback, POST on genuine miss, department_ids merge (Helix-managed slice replaced, geographic/Authing depts preserved), custom_attrs catalog translation by i18n_name + unknown-schema drop, freeze idempotent on already-frozen / user-absent. * Pure helpers — _normalise_attr_name, _coerce_department_ids, _parse_feishu_code. * Preset snapshot — mapping_rules target_field lock. * DEPT_KINDS registry + dispatcher priority. """ from __future__ import annotations from typing import Any import pytest from app.approval.feishu_client import FeishuApprovalClient from app.core.exceptions import AppException from app.domains.identity_sync.mapping import ( MappingConfig, render_payload, resolve_country_iso2, ) from app.domains.identity_sync.presets import ( FEISHU_CONTACT_USER_MAPPING_RULES, get_preset, ) from app.domains.identity_sync.runners.feishu_contact import ( FeishuContactRunner, _coerce_department_ids, _normalise_attr_name, _parse_feishu_code, ) from app.domains.identity_sync.runners.feishu_department import ( FeishuDepartmentRunner, _normalise_name, ) # ── Helpers ───────────────────────────────────────────────────────── def _client() -> FeishuApprovalClient: return FeishuApprovalClient(app_id="cli_x", app_secret="s") class _Calls(list): """Captures every (method, path, json_body) tuple sent to _request. Test inspects the full wire trace per call — important for paths like dept_ids-merge that issue GET-then-PATCH. """ def __init__(self, responses: list[Any]) -> None: super().__init__() self._responses = list(responses) async def __call__( self, method: str, path: str, *, json_body: dict[str, Any] | None = None, **_kw, ) -> dict[str, Any]: self.append({"method": method, "path": path, "json_body": json_body}) if not self._responses: raise RuntimeError(f"unexpected call {method} {path}") nxt = self._responses.pop(0) if isinstance(nxt, BaseException): raise nxt return nxt def _err(code: int, msg: str = "boom") -> AppException: """Build the AppException shape FeishuApprovalClient raises.""" return AppException( code="FEISHU_API_ERROR", message=f"Feishu API error {code}: {msg}", status_code=502, ) # ── Department runner ─────────────────────────────────────────────── class TestFeishuDepartmentRunner: """The lookup-and-adopt design under test: * Catalog primed with Feishu's existing name → od-XXX index. * Runner adopts by name on first sweep (PATCH existing od-XXX, not POST a duplicate); subsequent sweeps use prior_external_id direct. * The 7 geographic-extra depts that PingCAP's Feishu has (Tokyo Office, SG Office, ...) are NEVER in scope for the runner — they have no HRIS counterpart so the orchestrator never enqueues them. """ @pytest.mark.asyncio async def test_adopt_existing_dept_by_name(self, monkeypatch): """HRIS pushes 'Engineering Group' for the first time. Catalog already has the same-name Feishu dept under od-581b2e24... Runner PATCHes that od-XXX directly — no POST, no duplicate. """ runner = FeishuDepartmentRunner( client=_client(), name_to_external={ "Engineering Group": "od-581b2e24", "HR": "od-d7bc410e", }, ) calls = _Calls([{}]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert_dept( dept_code="1809808", dept_name="Engineering Group", parent_dept_code=None, leader_email=None, dept_order=1, prior_external_id=None, ) assert out.status == "success" assert out.external_id == "od-581b2e24" assert len(calls) == 1 assert calls[0]["method"] == "PATCH" assert "od-581b2e24" in calls[0]["path"] body = calls[0]["json_body"] assert "department_id" not in body assert "open_department_id" not in body assert body["code"] == "1809808" assert body["parent_department_id"] == "0" @pytest.mark.asyncio async def test_steady_state_patch_by_prior_external_id(self, monkeypatch): runner = FeishuDepartmentRunner( client=_client(), name_to_external={}, ) calls = _Calls([{}]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert_dept( dept_code="1809808", dept_name="Engineering Group", parent_dept_code=None, leader_email=None, dept_order=None, prior_external_id="od-581b2e24", ) assert out.status == "success" assert out.external_id == "od-581b2e24" assert len(calls) == 1 assert "od-581b2e24" in calls[0]["path"] @pytest.mark.asyncio async def test_create_new_when_name_not_in_catalog(self, monkeypatch): runner = FeishuDepartmentRunner( client=_client(), name_to_external={"Engineering Group": "od-581b2e24"}, ) calls = _Calls([ {"department": {"open_department_id": "od-NEW999"}}, ]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert_dept( dept_code="9999999", dept_name="New Innovation Lab", parent_dept_code=None, leader_email=None, dept_order=None, prior_external_id=None, ) assert out.status == "success" assert out.external_id == "od-NEW999" assert len(calls) == 1 assert calls[0]["method"] == "POST" body = calls[0]["json_body"] assert body["name"] == "New Innovation Lab" assert body["code"] == "9999999" @pytest.mark.asyncio async def test_parent_dept_code_translates_via_index(self, monkeypatch): runner = FeishuDepartmentRunner( client=_client(), name_to_external={"Infra Platform": "od-deadbeef"}, helix_dept_to_external={"1809808": "od-581b2e24"}, ) calls = _Calls([{}]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert_dept( dept_code="1809900", dept_name="Infra Platform", parent_dept_code="1809808", leader_email=None, dept_order=None, prior_external_id=None, ) assert out.status == "success" body = calls[0]["json_body"] assert body["parent_department_id"] == "od-581b2e24" @pytest.mark.asyncio async def test_unresolved_parent_falls_back_to_root(self, monkeypatch): runner = FeishuDepartmentRunner( client=_client(), name_to_external={"Infra Platform": "od-deadbeef"}, ) calls = _Calls([{}]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert_dept( dept_code="1809900", dept_name="Infra Platform", parent_dept_code="UNKNOWN-XXX", leader_email=None, dept_order=None, prior_external_id=None, ) assert out.status == "success" assert calls[0]["json_body"]["parent_department_id"] == "0" @pytest.mark.asyncio async def test_leader_lookup_miss_does_not_block(self, monkeypatch): runner = FeishuDepartmentRunner( client=_client(), name_to_external={"HR": "od-d7bc410e"}, ) async def fake_resolve(email: str) -> str: raise AppException( code="FEISHU_USER_NOT_FOUND", message=f"no user for {email}", status_code=400, ) monkeypatch.setattr(runner._client, "resolve_user_id_by_email", fake_resolve) calls = _Calls([{}]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert_dept( dept_code="1809817", dept_name="HR", parent_dept_code=None, leader_email="boss@pingcap.com", dept_order=None, prior_external_id=None, ) assert out.status == "success" body = calls[0]["json_body"] assert "leader_user_id" not in body @pytest.mark.asyncio async def test_soft_delete_already_deleted_is_success(self, monkeypatch): runner = FeishuDepartmentRunner(client=_client()) calls = _Calls([_err(230002, "already deleted")]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.soft_delete( dept_code="1809808", prior_external_id="od-581b2e24", ) assert out.status == "soft_deleted" assert out.external_id is None # Cleared on already-gone. @pytest.mark.asyncio async def test_soft_delete_sends_status_is_deleted(self, monkeypatch): runner = FeishuDepartmentRunner(client=_client()) calls = _Calls([{}]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.soft_delete( dept_code="1809808", prior_external_id="od-581b2e24", ) assert out.status == "soft_deleted" assert out.external_id == "od-581b2e24" assert calls[0]["method"] == "PATCH" assert calls[0]["json_body"] == {"status": {"is_deleted": True}} def test_lookup_external_id_by_name_returns_od_from_catalog(self): runner = FeishuDepartmentRunner( client=_client(), name_to_external={ "Engineering Group": "od-eng", "HR": "od-hr", }, ) # Normalisation handles case + whitespace. assert runner.lookup_external_id_by_name("engineering group") == "od-eng" assert runner.lookup_external_id_by_name("HR") == "od-hr" assert runner.lookup_external_id_by_name("Unknown Dept") is None assert runner.lookup_external_id_by_name("") is None assert runner.lookup_external_id_by_name(None) is None @pytest.mark.asyncio async def test_soft_delete_no_external_id_is_noop_success(self, monkeypatch): """If we never adopted (no external_id), there's nothing to delete on Feishu — treat as already done. """ runner = FeishuDepartmentRunner(client=_client()) calls = _Calls([]) # No calls expected. monkeypatch.setattr(runner._client, "_request", calls) out = await runner.soft_delete( dept_code="ghost", prior_external_id=None, ) assert out.status == "soft_deleted" assert out.external_id is None assert len(calls) == 0 class TestSiblingScopedAdoption: """Sibling-scoped (parent_od, name) adoption + unique-name fallback. Pins the fix for same-named depts under different parents (two "Sales") which the old flat name → od index aliased to one od and thrashed. The matching is shared with the dry-run via ``match_existing`` so the two can't drift. """ @staticmethod def _tree_responses() -> list[Any]: # BFS from "0": APAC + NA top-level, each with a child "Sales". # _Calls pops in order; BFS queries "0", then od-apac, od-na, # then the two leaf ods (empty). return [ {"items": [ {"name": "APAC", "open_department_id": "od-apac"}, {"name": "NA", "open_department_id": "od-na"}, ], "has_more": False}, {"items": [ {"name": "Sales", "open_department_id": "od-sales-apac"}, ], "has_more": False}, {"items": [ {"name": "Sales", "open_department_id": "od-sales-na"}, ], "has_more": False}, {"items": [], "has_more": False}, {"items": [], "has_more": False}, ] @pytest.mark.asyncio async def test_prime_catalog_builds_sibling_and_count_indices( self, monkeypatch, ): runner = FeishuDepartmentRunner(client=_client()) calls = _Calls(self._tree_responses()) monkeypatch.setattr(runner._client, "_request", calls) await runner.prime_catalog() # Flat index keeps the public office lookup working (Sales is # last-writer-wins, but its count flags the collision). assert runner._name_to_external["apac"] == "od-apac" assert runner._sibling_to_external[("od-apac", "sales")] == "od-sales-apac" assert runner._sibling_to_external[("od-na", "sales")] == "od-sales-na" assert runner._sibling_to_external[("0", "apac")] == "od-apac" assert runner._name_counts["sales"] == 2 assert runner._name_counts["apac"] == 1 @pytest.mark.asyncio async def test_match_existing_disambiguates_same_name(self, monkeypatch): runner = FeishuDepartmentRunner(client=_client()) calls = _Calls(self._tree_responses()) monkeypatch.setattr(runner._client, "_request", calls) await runner.prime_catalog() # Each "Sales" resolves to its OWN od by parent. assert runner.match_existing( dept_name="Sales", parent_od="od-apac", ) == ("od-sales-apac", "sibling") assert runner.match_existing( dept_name="Sales", parent_od="od-na", ) == ("od-sales-na", "sibling") # Colliding name with no sibling hit → create (never guess). assert runner.match_existing( dept_name="Sales", parent_od="od-unknown", ) == (None, "create") # Unique name → flat fallback adopts even on a parent mismatch. assert runner.match_existing( dept_name="APAC", parent_od="od-some-other-parent", ) == ("od-apac", "unique-name") @pytest.mark.asyncio async def test_two_same_named_depts_adopt_distinct_ods(self, monkeypatch): """End-to-end: a sweep adopting both "Sales" depts PATCHes the correct distinct od for each, never a POST-duplicate.""" runner = FeishuDepartmentRunner( client=_client(), # parents already adopted earlier in the (topo-ordered) sweep. helix_dept_to_external={"APAC": "od-apac", "NA": "od-na"}, ) # prime the catalog, then two PATCH responses for the adoptions. calls = _Calls([*self._tree_responses(), {}, {}]) monkeypatch.setattr(runner._client, "_request", calls) await runner.prime_catalog() out_apac = await runner.upsert_dept( dept_code="SALES-APAC", dept_name="Sales", parent_dept_code="APAC", leader_email=None, dept_order=2, prior_external_id=None, ) out_na = await runner.upsert_dept( dept_code="SALES-NA", dept_name="Sales", parent_dept_code="NA", leader_email=None, dept_order=2, prior_external_id=None, ) assert out_apac.external_id == "od-sales-apac" assert out_na.external_id == "od-sales-na" # Both were PATCH (adopt), not POST (create-duplicate). patches = [c for c in calls if c["method"] == "PATCH"] assert len(patches) == 2 assert any("od-sales-apac" in c["path"] for c in patches) assert any("od-sales-na" in c["path"] for c in patches) assert not [c for c in calls if c["method"] == "POST"] @pytest.mark.asyncio async def test_two_new_same_named_depts_both_create_not_adopt( self, monkeypatch, ): """Regression: when neither "Sales" exists on Feishu yet, creating the first must NOT make the second (different parent) adopt it via the unique-name fallback. Both POST-create distinct ods. Guards the create-tail: ``_name_counts`` must stay the pre-sweep snapshot (no "sales") so the fallback never fires mid-sweep. """ runner = FeishuDepartmentRunner( client=_client(), helix_dept_to_external={"APAC": "od-apac", "NA": "od-na"}, ) # Catalog has APAC + NA but NO "Sales" anywhere. calls = _Calls([ {"items": [ {"name": "APAC", "open_department_id": "od-apac"}, {"name": "NA", "open_department_id": "od-na"}, ], "has_more": False}, {"items": [], "has_more": False}, {"items": [], "has_more": False}, {"department": {"open_department_id": "od-sales-1"}}, {"department": {"open_department_id": "od-sales-2"}}, ]) monkeypatch.setattr(runner._client, "_request", calls) await runner.prime_catalog() out1 = await runner.upsert_dept( dept_code="SALES-APAC", dept_name="Sales", parent_dept_code="APAC", leader_email=None, dept_order=2, prior_external_id=None, ) out2 = await runner.upsert_dept( dept_code="SALES-NA", dept_name="Sales", parent_dept_code="NA", leader_email=None, dept_order=2, prior_external_id=None, ) assert out1.external_id == "od-sales-1" assert out2.external_id == "od-sales-2" # Both were POST (create), never a PATCH adopting the other. posts = [c for c in calls if c["method"] == "POST"] assert len(posts) == 2 assert not [c for c in calls if c["method"] == "PATCH"] # ── User runner ───────────────────────────────────────────────────── class TestFeishuContactRunner: """The slice-replace merge + email→mobile lookup + catalog-aware custom_attrs design under test. """ @pytest.mark.asyncio async def test_lookup_by_email_then_patch(self, monkeypatch): """Email-based lookup (the only supported path on Feishu) → PATCH existing user. Authing-created accounts get adopted, not recreated. """ runner = FeishuContactRunner(client=_client()) calls = _Calls([ # POST /contact/v3/users/batch_get_id (email→user_id) { "user_list": [ {"email": "alice@pingcap.com", "user_id": "u_alice"}, ], }, # GET current dept_ids {"user": {"department_ids": []}}, # PATCH success {}, ]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert( person_email="alice@pingcap.com", payload={ "email": "alice@pingcap.com", "name": "Alice", "employee_no": "P0001", }, prior_external_id=None, ) assert out.status == "success" assert out.external_id == "u_alice" # First call is the email batch_get_id, NOT the employee_no GET # (that endpoint doesn't accept user_id_type=employee_no — code # 99992402 — so we removed that primary lookup). assert calls[0]["path"].startswith("/contact/v3/users/batch_get_id") assert calls[0]["method"] == "POST" @pytest.mark.asyncio async def test_create_on_genuine_miss(self, monkeypatch): runner = FeishuContactRunner(client=_client()) calls = _Calls([ # email lookup miss (empty user_list) {"user_list": []}, # employee_no last-resort index: directory walk (no mobile in # payload → mobile fallback skipped). Root has no child depts, # dept "0" has no matching user → genuine miss → POST. {"items": [], "has_more": False}, # GET /departments (root children) {"items": [], "has_more": False}, # GET find_by_department(0) # Directory v1 create (3a0adf7e2 — geo fixed at creation). {"employee_id": "u_new42"}, # Contact projection PATCH that applies Helix's richer fields. {}, ]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert( person_email="newhire@pingcap.com", payload={ "email": "newhire@pingcap.com", "name": "Newbie", "employee_no": "P0700", # Required since 3a0adf7e2 — a create without an explicit # residence country is BLOCKED, not silently defaulted. "_helix_residence_country": "CN", }, prior_external_id=None, ) assert out.status == "success" assert out.external_id == "u_new42" assert out.detail["geo_name"] == "cn" # Create is Directory v1; the Contact write is the follow-up # projection onto the id Directory returned. create = calls[-2] assert create["method"] == "POST" assert create["path"].startswith("/directory/v1/employees") assert create["json_body"]["employee"]["email"] == "newhire@pingcap.com" assert calls[-1]["method"] == "PATCH" assert "u_new42" in calls[-1]["path"] # No HRIS dept resolved + no helix index → fallback to root "0", # now carried by the projection rather than the create body. assert calls[-1]["json_body"]["department_ids"] == ["0"] @pytest.mark.asyncio async def test_employee_no_index_matches_when_email_and_mobile_miss( self, monkeypatch, ): """Residual duplicate gap: a person with NO mobile whose Feishu account is registered under a different email misses BOTH the email and mobile lookups → would POST a DUPLICATE (no mobile for Feishu's server-side dedup). The last-resort employee_no directory index finds the existing account by the HRIS natural key and routes to PATCH. """ runner = FeishuContactRunner(client=_client()) calls = _Calls([ {"user_list": []}, # 1. email miss {"items": [], "has_more": False}, # 2. dept BFS root {"items": [ # 3. users in dept 0 {"user_id": "u_dir", "employee_no": "P0555"}, ], "has_more": False}, {"user": {"department_ids": []}}, # 4. GET current depts {}, # 5. PATCH success ]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert( person_email="ghost@pingcap.com", # not the Feishu-registered email payload={"email": "ghost@pingcap.com", "name": "Ghost", "employee_no": "P0555"}, prior_external_id=None, ) assert out.status == "success" assert out.external_id == "u_dir" # Matched by employee_no → PATCH, never a POST-create-duplicate. assert calls[-1]["method"] == "PATCH" assert "u_dir" in calls[-1]["path"] assert any("find_by_department" in c["path"] for c in calls) @pytest.mark.asyncio async def test_employee_no_index_miss_still_creates(self, monkeypatch): """Directory built but the employee_no isn't on Feishu → the person genuinely doesn't exist → POST-create is correct (no false adopt).""" runner = FeishuContactRunner(client=_client()) calls = _Calls([ {"user_list": []}, # email miss {"items": [], "has_more": False}, # dept BFS root {"items": [ # dir has OTHER people {"user_id": "u_other", "employee_no": "P0001"}, ], "has_more": False}, {"employee_id": "u_created"}, # Directory v1 create {}, # Contact projection ]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert( person_email="truly.new@pingcap.com", payload={ "email": "truly.new@pingcap.com", "name": "New", "employee_no": "P0999", "_helix_residence_country": "CN", }, prior_external_id=None, ) assert out.status == "success" assert out.external_id == "u_created" assert calls[-2]["method"] == "POST" assert calls[-2]["path"].startswith("/directory/v1/employees") @pytest.mark.asyncio async def test_employee_no_index_built_once_per_sweep(self, monkeypatch): """The directory walk is cached on the runner: a second person in the same sweep reuses the index — the expensive walk runs exactly once.""" runner = FeishuContactRunner(client=_client()) calls = _Calls([ # person A: email miss → build index (dept + users, has A & B) → PATCH {"user_list": []}, {"items": [], "has_more": False}, {"items": [ {"user_id": "u_a", "employee_no": "P0011"}, {"user_id": "u_b", "employee_no": "P0022"}, ], "has_more": False}, {"user": {"department_ids": []}}, {}, # person B: email miss → index CACHED (no dir walk) → PATCH {"user_list": []}, {"user": {"department_ids": []}}, {}, ]) monkeypatch.setattr(runner._client, "_request", calls) out_a = await runner.upsert( person_email="a@pingcap.com", payload={"email": "a@pingcap.com", "name": "A", "employee_no": "P0011"}, prior_external_id=None, ) out_b = await runner.upsert( person_email="b@pingcap.com", payload={"email": "b@pingcap.com", "name": "B", "employee_no": "P0022"}, prior_external_id=None, ) assert out_a.external_id == "u_a" assert out_b.external_id == "u_b" # The full directory walk (find_by_department) happened exactly ONCE. assert sum("find_by_department" in c["path"] for c in calls) == 1 @pytest.mark.asyncio async def test_employee_no_index_build_failure_fails_loud_not_posts( self, monkeypatch, ): """FAIL-LOUD: if the directory walk fails (non-systemic), the person is FAILED (retry next sweep) — NEVER POST-created. This fallback is the last line of defense against a duplicate (no Feishu mobile-dedup backstop), so a swallowed failure that fell through to POST would re-create the exact duplicate the feature prevents.""" runner = FeishuContactRunner(client=_client()) calls = _Calls([ {"user_list": []}, # email miss (no mobile in payload) _err(131006, "dept page failed"), # dept-BFS GET raises → walk fails ]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert( person_email="ghost@pingcap.com", payload={"email": "ghost@pingcap.com", "name": "Ghost", "employee_no": "P0555"}, prior_external_id=None, ) assert out.status == "failed" assert "index build failed" in (out.error or "") # Crucially: NO create was attempted. (The email lookup is itself a # POST to batch_get_id; the create is POST /contact/v3/users — assert # that specific create endpoint was never hit.) assert not any( c["method"] == "POST" and "batch_get_id" not in c["path"] for c in calls ) @pytest.mark.asyncio async def test_mobile_fallback_finds_account_and_patches_wrong_email( self, monkeypatch, ): """Regression (P10012 王静文, 2026-07-13): an existing Feishu account whose registered email differs from work_email (``jingwen.wang1`` vs ``jingwen.wang``) is MISSED by the email lookup. Before the fix it fell through to POST, Feishu matched it by mobile server-side and returned the existing id as a "create" WITHOUT applying the correct email — freezing the wrong email behind hash-skip. Now a mobile fallback lookup resolves the id, routing the person through PATCH, which pushes email=work_email and self-corrects it. """ runner = FeishuContactRunner(client=_client()) calls = _Calls([ # 1. email batch_get_id → MISS (account is under .wang1) {"user_list": []}, # 2. mobile batch_get_id → HIT (mobile fallback resolves the id) {"user_list": [ {"mobile": "+8618801061613", "user_id": "u_jingwen"}, ]}, # 3. hijack-guard verify GET: matched account's employee_no # matches this person's → adopt the hit. {"user": {"employee_no": "P10012"}}, # 4. GET current dept_ids {"user": {"department_ids": []}}, # 5. PATCH success {}, ]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert( person_email="jingwen.wang@pingcap.com", payload={ "email": "jingwen.wang@pingcap.com", "name": "王静文", "employee_no": "P10012", "mobile": "+8618801061613", }, prior_external_id=None, ) assert out.status == "success" assert out.external_id == "u_jingwen" # Email lookup first (by emails), then mobile fallback (by mobiles). assert calls[0]["path"].startswith("/contact/v3/users/batch_get_id") assert "emails" in calls[0]["json_body"] assert calls[1]["path"].startswith("/contact/v3/users/batch_get_id") assert calls[1]["json_body"]["mobiles"] == ["+8618801061613"] # Resolved → PATCH (update existing), NEVER a POST create (which # would leave the wrong email in place). assert calls[-1]["method"] == "PATCH" assert "u_jingwen" in calls[-1]["path"] # The PATCH carries the CORRECTED work_email, self-healing .wang1. assert calls[-1]["json_body"]["email"] == "jingwen.wang@pingcap.com" @pytest.mark.asyncio async def test_no_mobile_in_payload_still_creates_on_miss( self, monkeypatch, ): """Backward-compat: when the payload carries no mobile, the mobile fallback is skipped entirely (no extra batch_get_id call) and a genuine email miss still goes straight to POST create. """ runner = FeishuContactRunner(client=_client()) calls = _Calls([ {"user_list": []}, # email miss {"employee_id": "u_created"}, # Directory v1 create {}, # Contact projection ]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert( person_email="nomobile@pingcap.com", payload={ "email": "nomobile@pingcap.com", "name": "No Mobile", "_helix_residence_country": "CN", }, prior_external_id=None, ) assert out.status == "success" assert out.external_id == "u_created" # Exactly three calls: email lookup + Directory create + Contact # projection. Still NO mobile lookup — that's what this pins. assert len(calls) == 3 assert not any("mobiles" in (c["json_body"] or {}) for c in calls) assert calls[1]["path"].startswith("/directory/v1/employees") # A payload with no mobile produces no ``mobile`` on the create body. assert "mobile" not in calls[1]["json_body"]["employee"] assert calls[2]["method"] == "PATCH" @pytest.mark.asyncio async def test_body_uses_phone_country_over_home_country(self, monkeypatch): """International assignee: home country CN, host-country phone (+65). The synthetic ``mobile_country_iso2`` (SG) takes precedence so the wire mobile is +65…, not +86… — and it's popped from the body.""" runner = FeishuContactRunner(client=_client()) calls = _Calls([ {"user": {"department_ids": []}}, # GET current dept_ids {}, # PATCH success ]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert( person_email="zhaohui@pingcap.com", payload={ "email": "zhaohui@pingcap.com", "name": "Zita Zhao", "mobile": "81738066", "country": "CN", "mobile_country_iso2": "SG", # phone's own country }, prior_external_id="u_z", ) assert out.status == "success" body = calls[-1]["json_body"] assert body["mobile"] == "+6581738066" # SG dialing code, not +86 assert "mobile_country_iso2" not in body # synthetic popped @pytest.mark.parametrize( ("cc", "country", "national", "expected"), [ # The prod cohort this fixes. HK staff all carry mainland SIMs, # so phone-country and work-country genuinely differ; the ISO-3 # value-map used to miss "CN" and weld the work country's code on. ("CN", "HKG", "17688850795", "+8617688850795"), # P10027 Nikita ("CN", "USA", "15652349467", "+8615652349467"), # P0001 Max Liu ("CN", "MYS", "15682051776", "+8615682051776"), # was 41004-dropped ("TW", "USA", "979895750", "+886979895750"), # was 41004-dropped ("MY", "CHN", "189411338", "+60189411338"), # was 41004-dropped ("BR", "BRA", "21972264199", "+5521972264199"), # was dropped: BRA unmapped # SF's ISO-3 picklist values must keep working — they are why the # old value-map existed. ("SWE", "SWE", "703360083", "+46703360083"), # Free text resolves to nothing → work-country fallback, unchanged. ("GB / UK", "GBR", "7835364431", "+447835364431"), # No phone country at all → work country, the pre-existing path. ("", "JPN", "9012345678", "+819012345678"), ], ) @pytest.mark.asyncio async def test_wire_mobile_follows_the_phone_not_the_workplace( self, monkeypatch, cc, country, national, expected, ): """End-to-end: preset rules → rendered payload → wire body. Renders through the REAL preset rules (not a hand-built payload) so a regression in the rule's source_template fails here, and asserts the number Feishu actually receives. """ ctx = { "work_email": "p@pingcap.com", "display_name": "Test Person", "mobile_phone": national, "country": country, "mobile_country_code": cc, "mobile_country_iso2": ( resolve_country_iso2(cc) or resolve_country_iso2(country) ), } cfg = MappingConfig.from_dict( {"rules": [ r for r in FEISHU_CONTACT_USER_MAPPING_RULES if r["target_field"] in ( "mobile", "country", "mobile_country_iso2", "mobile_country_iso3", ) ]}, ) payload = render_payload(cfg, ctx) payload["email"] = "p@pingcap.com" payload["name"] = "Test Person" runner = FeishuContactRunner(client=_client()) calls = _Calls([ {"user": {"department_ids": []}}, # GET current dept_ids {}, # PATCH success ]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert( person_email="p@pingcap.com", payload=payload, prior_external_id="u_x", ) assert out.status == "success" body = calls[-1]["json_body"] assert body.get("mobile") == expected # Synthetics never reach the wire. assert "mobile_country_iso2" not in body assert "mobile_country_iso3" not in body @pytest.mark.asyncio async def test_patch_mobile_invalid_retries_without_mobile(self, monkeypatch): """Defense-in-depth: if Feishu still rejects the mobile (41004), the PATCH retries WITHOUT the mobile slot so the rest of the fields land and the existing number is preserved — the person isn't a total failure over one bad optional field.""" runner = FeishuContactRunner(client=_client()) calls = _Calls([ {"user": {"department_ids": []}}, # GET current dept_ids _err(41004, "mobile is invalid error"), # PATCH rejects mobile {}, # retry (no mobile) OK ]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert( person_email="bad.phone@pingcap.com", payload={ "email": "bad.phone@pingcap.com", "name": "Bad Phone", "mobile": "+8681738066", }, prior_external_id="u_bp", ) assert out.status == "success" assert "mobile not pushed" in (out.error or "") # Retry PATCH dropped mobile but kept the rest. retry_body = calls[-1]["json_body"] assert "mobile" not in retry_body assert retry_body["email"] == "bad.phone@pingcap.com" @pytest.mark.asyncio async def test_create_mobile_invalid_retries_without_mobile(self, monkeypatch): """The Directory v1 create tolerates a mobile-invalid (41004): retry the POST without the mobile slot so the new hire STILL gets an account, and record the dropped slot so the ledger shows the number didn't land. History: this drop-and-retry lived in the Contact v3 create body, which 3a0adf7e2 made unreachable when it routed every create through Directory v1 — so a hire whose HRIS phone Feishu wouldn't validate got NO Feishu account at all and went red every sweep until someone fixed the number. #3274 pinned that hard failure honestly rather than papering over it; this restores the tolerance on the new surface. Same deliberate degraded success as the PATCH path (#1956) — never fail a whole person over one optional field. """ runner = FeishuContactRunner(client=_client()) calls = _Calls([ {"user_list": []}, # email lookup miss {"user_list": []}, # mobile fallback miss → create _err(41004, "mobile is invalid error"), # Directory create rejects mobile {"employee_id": "u_new"}, # retry (no mobile) creates {}, # Contact projection PATCH OK ]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert( person_email="newbad@pingcap.com", payload={ "email": "newbad@pingcap.com", "name": "New Bad", "mobile": "+8681738066", "_helix_residence_country": "CN", }, prior_external_id=None, ) assert out.status == "success" assert out.external_id == "u_new" # The account was created by a SECOND Directory POST that dropped the # mobile — the first one carried it and was rejected. creates = [ c for c in calls if c["method"] == "POST" and c["path"].startswith("/directory/v1/employees") ] assert len(creates) == 2 assert creates[0]["json_body"]["employee"]["mobile"] == "+8681738066" assert "mobile" not in creates[1]["json_body"]["employee"] # …and the rest of the employee payload survived the retry. assert creates[1]["json_body"]["employee"]["email"] == "newbad@pingcap.com" assert creates[1]["json_body"]["options"]["geo_name"] == "cn" # The Contact projection doesn't re-send the number Feishu just # refused (that would only buy a guaranteed second 41004). patches = [c for c in calls if c["method"] == "PATCH"] assert len(patches) == 1 assert "mobile" not in (patches[0]["json_body"] or {}) # Degraded success, not a clean green: the ledger records the drop. assert "without mobile" in (out.error or "") mobile_slots = [ s for s in (out.slot_outcomes or []) if s.get("slot") == "mobile" ] assert len(mobile_slots) == 1 assert mobile_slots[0]["status"] == "dropped" assert mobile_slots[0]["feishu_code"] == 41004 assert mobile_slots[0]["value"] == "+8681738066" @pytest.mark.asyncio async def test_create_mobile_fallback_failure_is_still_a_failure(self, monkeypatch): """The drop-and-retry is one retry, not a swallow: if the mobile-less create ALSO fails, the person is reported failed (with the fallback named) instead of a silent green.""" runner = FeishuContactRunner(client=_client()) calls = _Calls([ {"user_list": []}, # email lookup miss {"user_list": []}, # mobile fallback miss → create _err(41004, "mobile is invalid error"), # Directory create rejects mobile _err(99991663, "tenant access token invalid"), # retry also fails ]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert( person_email="worse@pingcap.com", payload={ "email": "worse@pingcap.com", "name": "Worse", "mobile": "+8681738066", "_helix_residence_country": "CN", }, prior_external_id=None, ) assert out.status == "failed" assert "mobile-fallback" in (out.error or "") @pytest.mark.asyncio async def test_create_non_mobile_error_is_not_retried(self, monkeypatch): """Only 41004 earns the drop-and-retry. Any other Directory refusal still fails on the first call — no blind second POST that could double-create.""" runner = FeishuContactRunner(client=_client()) calls = _Calls([ {"user_list": []}, # email lookup miss {"user_list": []}, # mobile fallback miss → create _err(99991663, "tenant access token invalid"), ]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert( person_email="boom@pingcap.com", payload={ "email": "boom@pingcap.com", "name": "Boom", "mobile": "+8681738066", "_helix_residence_country": "CN", }, prior_external_id=None, ) assert out.status == "failed" assert len(calls) == 3 # no retry POST assert not out.slot_outcomes # nothing was "dropped" @pytest.mark.asyncio async def test_create_without_employee_type_needs_no_default(self, monkeypatch): """A future hire with no effective-dated employee_type_code still gets an account, and no synthetic default is invented. This test used to assert a create-only ``employee_type=1`` fallback (2026-07-13, P10011 Eddie Jiang), which existed because Contact v3 POST rejected a create without it (99992402). 3a0adf7e2 moved creation to Directory v1, whose body carries only name/email/mobile — the field isn't required there, and the follow-up Contact call is a PATCH, where drop-on-empty is the correct semantics (never clobber a good downstream value). So the default is gone by construction, and asserting its absence is what keeps the intent honest. """ runner = FeishuContactRunner(client=_client()) calls = _Calls([ {"user_list": []}, # email lookup miss → create {"employee_id": "u_new"}, # Directory v1 create OK {}, # Contact projection ]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert( person_email="future.hire@pingcap.com", payload={ "email": "future.hire@pingcap.com", "name": "Future Hire", "_helix_residence_country": "CN", }, prior_external_id=None, ) assert out.status == "success" assert out.external_id == "u_new" # Directory create takes no employee_type… assert "employee_type" not in calls[1]["json_body"]["employee"] # …and the projection PATCH does not invent one either. assert "employee_type" not in calls[2]["json_body"] @pytest.mark.asyncio async def test_create_keeps_explicit_employee_type(self, monkeypatch): """A populated employee_type still reaches Feishu on a create — the mapped value (e.g. 5 顾问) is sent verbatim, now on the Contact projection PATCH rather than the (Directory v1) create body.""" runner = FeishuContactRunner(client=_client()) calls = _Calls([ {"user_list": []}, {"employee_id": "u_new2"}, # Directory v1 create {}, # Contact projection ]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert( person_email="consultant@pingcap.com", payload={ "email": "consultant@pingcap.com", "name": "Consultant", "employee_type": 5, "_helix_residence_country": "CN", }, prior_external_id=None, ) assert out.status == "success" assert calls[-1]["method"] == "PATCH" assert calls[-1]["json_body"].get("employee_type") == 5 @pytest.mark.asyncio async def test_skip_list_suppresses_create(self, monkeypatch): """Admin skip-list: a missing person on the skip-list is NOT created — the runner returns 'skipped' and issues NO POST, so no junk/duplicate account is made (e.g. placeholder email, leaver leftover). """ runner = FeishuContactRunner( client=_client(), skip_provision_emails=frozenset({"skip.me@pingcap.com"}), ) calls = _Calls([ # email lookup miss (empty user_list) — would normally create {"user_list": []}, ]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert( person_email="skip.me@pingcap.com", payload={"email": "skip.me@pingcap.com", "name": "Skip Me"}, prior_external_id=None, ) assert out.status == "skipped" assert out.external_id is None # Only the lookup ran — NO POST create. assert len(calls) == 1 assert calls[0]["path"].startswith("/contact/v3/users/batch_get_id") @pytest.mark.asyncio async def test_reopen_frozen_asserts_active_on_patch(self, monkeypatch): """reopen_frozen=True → the PATCH asserts status.is_frozen=false so an in-scope, HRIS-active person frozen on Feishu is reopened. """ runner = FeishuContactRunner(client=_client(), reopen_frozen=True) calls = _Calls([ {"user_list": [{"email": "bob@pingcap.com", "user_id": "u_bob"}]}, {"user": {"department_ids": []}}, # GET current dept_ids {}, # PATCH success ]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert( person_email="bob@pingcap.com", payload={"email": "bob@pingcap.com", "name": "Bob"}, prior_external_id=None, ) assert out.status == "success" patch = calls[-1] assert patch["method"] == "PATCH" assert patch["json_body"]["status"]["is_frozen"] is False @pytest.mark.asyncio async def test_skip_list_suppresses_reopen_but_still_updates(self, monkeypatch): """Skip-list wins over reopen_frozen: a skip-listed existing user is still UPDATED (PATCH), but status is NOT asserted — an intentionally-frozen account stays frozen. """ runner = FeishuContactRunner( client=_client(), reopen_frozen=True, skip_provision_emails=frozenset({"frozen.keep@pingcap.com"}), ) calls = _Calls([ {"user_list": [ {"email": "frozen.keep@pingcap.com", "user_id": "u_fk"}, ]}, {"user": {"department_ids": []}}, # GET current dept_ids {}, # PATCH success ]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert( person_email="frozen.keep@pingcap.com", payload={"email": "frozen.keep@pingcap.com", "name": "FK"}, prior_external_id=None, ) assert out.status == "success" patch = calls[-1] assert patch["method"] == "PATCH" # Still updates, but must NOT reopen (no status assertion). assert "status" not in (patch["json_body"] or {}) @pytest.mark.asyncio async def test_dept_ids_merge_preserves_geographic_depts(self, monkeypatch): """User exists with 2 dept_ids on Feishu: - od-eng (Engineering Group, Helix-managed) - od-tokyo (Tokyo Office, Authing-managed, NOT in Helix's index) Helix's mapping pushes department_code = HRIS Engineering code. Expected merge: replace od-eng with od-eng (or any reshuffle of the Helix slice) + KEEP od-tokyo untouched. """ runner = FeishuContactRunner( client=_client(), helix_dept_to_external={ "1809808": "od-eng", # Engineering Group (Helix-managed) "1809817": "od-hr", # HR (Helix-managed) }, ) calls = _Calls([ # GET current dept_ids: Eng + Tokyo {"user": {"department_ids": ["od-eng", "od-tokyo"]}}, # PATCH success {}, ]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert( person_email="alice@pingcap.com", payload={ "email": "alice@pingcap.com", "name": "Alice", "department_ids": "1809808", # Eng Group HRIS code }, prior_external_id="u_alice", ) assert out.status == "success" body = calls[-1]["json_body"] # od-tokyo preserved (Helix doesn't manage it). assert "od-tokyo" in body["department_ids"] # od-eng present (Helix-managed slice). assert "od-eng" in body["department_ids"] # Sanity: order is helix-first then non-helix. assert body["department_ids"][0] == "od-eng" assert body["department_ids"][-1] == "od-tokyo" @pytest.mark.asyncio async def test_dept_ids_merge_drops_helix_dept_removed_from_hris( self, monkeypatch, ): """User used to be in 2 Helix-managed depts (od-eng + od-hr). Now HRIS only has them in Eng. Merge result: od-eng kept, od-hr DROPPED (Helix's slice owns the HRIS-managed depts). Geographic Tokyo Office preserved. """ runner = FeishuContactRunner( client=_client(), helix_dept_to_external={ "1809808": "od-eng", "1809817": "od-hr", }, ) calls = _Calls([ {"user": {"department_ids": ["od-eng", "od-hr", "od-tokyo"]}}, {}, ]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert( person_email="alice@pingcap.com", payload={ "email": "alice@pingcap.com", "name": "Alice", "department_ids": "1809808", # only Eng in HRIS now }, prior_external_id="u_alice", ) assert out.status == "success" body = calls[-1]["json_body"] # od-hr (Helix-managed but removed from HRIS) → dropped. assert "od-hr" not in body["department_ids"] # od-eng kept (still in HRIS). assert "od-eng" in body["department_ids"] # od-tokyo kept (non-Helix). assert "od-tokyo" in body["department_ids"] @pytest.mark.asyncio async def test_dept_ids_merge_get_failure_skips_slot(self, monkeypatch): """If GET current dept_ids fails, runner PATCHes WITHOUT department_ids slot — Feishu leaves them untouched (better than risking a clobber) — but the outcome is ``failed``, NOT success: the pending dept change was NOT applied and a plain success would store the payload hash and freeze the person forever (#1737). The ``failed`` keeps the hash unwritten so the next sweep retries. """ runner = FeishuContactRunner( client=_client(), helix_dept_to_external={"1809808": "od-eng"}, ) calls = _Calls([ _err(403, "no permission to read user"), # GET fails {}, # PATCH succeeds without dept_ids ]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert( person_email="alice@pingcap.com", payload={ "email": "alice@pingcap.com", "department_ids": "1809808", }, prior_external_id="u_alice", ) assert out.status == "failed" assert "current-dept read failed" in (out.error or "") # PATCH body must omit department_ids entirely. body = calls[-1]["json_body"] assert "department_ids" not in body # ...and therefore MUST omit orders too (Feishu 44002 rejects # orders without department_ids). assert "orders" not in body # ── Primary-department (orders) — the SV-Office-shadows-PreSales fix ─ def test_build_dept_orders_floats_primary_first(self): """Primary dept floats to index 0 with the highest department_order + is_primary_dept=True; the rest non-primary.""" orders = FeishuContactRunner._build_dept_orders( ["od-a", "od-b", "od-c"], "od-b", ) assert orders[0]["department_id"] == "od-b" assert orders[0]["is_primary_dept"] is True assert orders[0]["department_order"] == 3 assert [o["department_id"] for o in orders] == ["od-b", "od-a", "od-c"] assert all(o["is_primary_dept"] is False for o in orders[1:]) def test_build_dept_orders_no_primary_when_absent(self): """Falsy / absent primary → no dept flagged primary (Feishu keeps its own choice); still emits well-formed order entries.""" assert FeishuContactRunner._build_dept_orders([], None) == [] orders = FeishuContactRunner._build_dept_orders(["od-a", "od-b"], None) assert all(o["is_primary_dept"] is False for o in orders) # Primary not among dept_ids → also not forced. orders2 = FeishuContactRunner._build_dept_orders(["od-a"], "od-x") assert all(o["is_primary_dept"] is False for o in orders2) @pytest.mark.asyncio async def test_create_sets_business_dept_as_primary(self, monkeypatch): """New hire with an adopted business dept + an office node. The create MUST carry ``orders`` flagging the business dept primary so the contact card shows the real department (PreSales), not the appended office node (SV Office) — the exact 2026-07 bug. Since 3a0adf7e2 the departments ride the Contact projection PATCH (Directory v1 creates the bare employee), so the ``orders`` slot is asserted there — the guarantee is unchanged, the carrier moved. """ runner = FeishuContactRunner( client=_client(), helix_dept_to_external={"2181695": "od-presales"}, office_name_to_external={"SV Office": "od-sv"}, ) calls = _Calls([ {"user_list": []}, # email lookup miss → create {"employee_id": "u_new"}, # Directory v1 create {}, # Contact projection PATCH ]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert( person_email="ritika@pingcap.com", payload={ "email": "ritika@pingcap.com", "name": "Ritika", "department_ids": "2181695", # PreSales HRIS code "office_dept_name": "SV Office", "_helix_residence_country": "CN", }, prior_external_id=None, ) assert out.status == "success" assert calls[-1]["method"] == "PATCH" body = calls[-1]["json_body"] assert body["department_ids"] == ["od-presales", "od-sv"] primary = [o for o in body["orders"] if o["is_primary_dept"]] assert len(primary) == 1 assert primary[0]["department_id"] == "od-presales" @pytest.mark.asyncio async def test_patch_flips_primary_from_office_to_business( self, monkeypatch, ): """Existing user whose primary is WRONGLY the office node (the state Ritika was in). PATCH must send ``orders`` flipping the business dept to primary, keeping the office node non-primary. """ runner = FeishuContactRunner( client=_client(), helix_dept_to_external={"2181695": "od-presales"}, office_name_to_external={"SV Office": "od-sv"}, ) calls = _Calls([ # GET current: only the office node, and it's primary (bug). {"user": { "department_ids": ["od-sv"], "orders": [ {"department_id": "od-sv", "is_primary_dept": True}, ], }}, {}, # PATCH ]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert( person_email="ritika@pingcap.com", payload={ "email": "ritika@pingcap.com", "department_ids": "2181695", "office_dept_name": "SV Office", }, prior_external_id="u_ritika", ) assert out.status == "success" body = calls[-1]["json_body"] assert "od-presales" in body["department_ids"] assert "od-sv" in body["department_ids"] primary = [o for o in body["orders"] if o["is_primary_dept"]] assert [o["department_id"] for o in primary] == ["od-presales"] @pytest.mark.asyncio async def test_patch_no_orders_when_business_already_primary( self, monkeypatch, ): """A correctly-synced user (business dept already primary) must NOT get an ``orders`` slot — otherwise every sweep would churn every user's primary. Zero-blast steady state. """ runner = FeishuContactRunner( client=_client(), helix_dept_to_external={"2181695": "od-presales"}, office_name_to_external={"SV Office": "od-sv"}, ) calls = _Calls([ {"user": { "department_ids": ["od-presales", "od-sv"], "orders": [ {"department_id": "od-presales", "is_primary_dept": True}, {"department_id": "od-sv", "is_primary_dept": False}, ], }}, {}, # PATCH ]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert( person_email="peer@pingcap.com", payload={ "email": "peer@pingcap.com", "department_ids": "2181695", "office_dept_name": "SV Office", }, prior_external_id="u_peer", ) assert out.status == "success" body = calls[-1]["json_body"] assert "orders" not in body @pytest.mark.asyncio async def test_custom_attrs_catalog_translation(self, monkeypatch): """Mapping rule renders custom_attrs.pinyin + .location + .company_entity + .unknown. Runner translates the known three to Feishu's [{type, id, value: {text}}] list using the catalog; drops .unknown with a log. """ runner = FeishuContactRunner( client=_client(), custom_attr_catalog={ "pinyin": {"id": "C-6982120454421233667", "type": "TEXT"}, "company_entity": {"id": "C-7418413146041008130", "type": "TEXT"}, "location": {"id": "C-7457353071188230163", "type": "TEXT"}, # Also map "公司" → same Feishu schema as company_entity # (zh_cn i18n name should hit the same row). }, ) calls = _Calls([ {"user": {"department_ids": []}}, # GET dept_ids {}, # PATCH ]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert( person_email="alice@pingcap.com", payload={ "email": "alice@pingcap.com", "name": "Alice", "custom_attrs": { "pinyin": "Liyang Zhang", "location": "Beijing Office", "company_entity": "PingCAP (US), Inc.", "unknown": "should be dropped", }, }, prior_external_id="u_alice", ) assert out.status == "success" body = calls[-1]["json_body"] attrs = body.get("custom_attrs") or [] # Three known schemas translated. ids_sent = {a["id"] for a in attrs} assert "C-6982120454421233667" in ids_sent assert "C-7418413146041008130" in ids_sent assert "C-7457353071188230163" in ids_sent # No leak of the unknown schema name as an id. assert "unknown" not in ids_sent # All translated entries carry the Feishu shape. for entry in attrs: assert entry["type"] == "TEXT" assert "id" in entry assert "value" in entry assert "text" in entry["value"] @pytest.mark.asyncio async def test_custom_attrs_omitted_when_catalog_probe_failed( self, monkeypatch, ): """When ``prime_custom_attrs`` couldn't reach Feishu (the catalog is None on the runner), the user push omits the ``custom_attrs`` slot ENTIRELY rather than passing raw names through (which would 400 the whole PATCH). Feishu PATCH with the slot absent leaves it unchanged. Degrades gracefully — the rest of the user push still goes through. """ runner = FeishuContactRunner( client=_client(), custom_attr_catalog=None, # probe-failed state ) calls = _Calls([ {"user": {"department_ids": []}}, {}, # PATCH success ]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert( person_email="alice@pingcap.com", payload={ "email": "alice@pingcap.com", "name": "Alice", "custom_attrs": { "pinyin": "Alice", "company_entity": "PingCAP", }, }, prior_external_id="u_alice", ) assert out.status == "success" # No custom_attrs in the PATCH body — Feishu leaves the slot # alone instead of 400ing on a raw-name id. assert "custom_attrs" not in calls[-1]["json_body"] @pytest.mark.asyncio async def test_custom_attrs_skipped_when_unrecognised_type( self, monkeypatch, ): """Catalog-known schemas of non-TEXT types (HREF / ENUM / GENERIC_USER) are dropped in v1 — runner only wires TEXT. """ runner = FeishuContactRunner( client=_client(), custom_attr_catalog={ "hrbp": {"id": "C-7143156434032312324", "type": "GENERIC_USER"}, "pinyin": {"id": "C-6982120454421233667", "type": "TEXT"}, }, ) calls = _Calls([ {"user": {"department_ids": []}}, {}, ]) monkeypatch.setattr(runner._client, "_request", calls) await runner.upsert( person_email="alice@pingcap.com", payload={ "email": "alice@pingcap.com", "name": "Alice", "custom_attrs": { "pinyin": "Liyang Zhang", "hrbp": "ou_someoneelse", }, }, prior_external_id="u_alice", ) body = calls[-1]["json_body"] attrs = body.get("custom_attrs") or [] ids = [a["id"] for a in attrs] # Only the TEXT one made it through. assert ids == ["C-6982120454421233667"] # ── HRBP (GENERIC_USER custom attr) push ───────────────────────── def test_build_hrbp_custom_attr_shape(self): """Exact Feishu wire shape, verified live 2026-06-16: ``{type: GENERIC_USER, id, value: {generic_user: {id, type: 1}}}``. Schema id comes from the live catalog (HRBP i18n_name), not a hardcoded constant.""" runner = FeishuContactRunner( client=_client(), custom_attr_catalog={ "hrbp": {"id": "C-7143156434032312324", "type": "GENERIC_USER"}, }, ) attr = runner._build_hrbp_custom_attr("9531fc89") assert attr == { "type": "GENERIC_USER", "id": "C-7143156434032312324", "value": {"generic_user": {"id": "9531fc89", "type": 1}}, } def test_build_hrbp_custom_attr_none_cases(self): cat = {"hrbp": {"id": "C-714", "type": "GENERIC_USER"}} # No resolved user_id → None (slot omitted, vendor keeps its value). assert FeishuContactRunner( client=_client(), custom_attr_catalog=cat, )._build_hrbp_custom_attr(None) is None # Catalog probe failed → None (can't know the schema id). assert FeishuContactRunner( client=_client(), custom_attr_catalog=None, )._build_hrbp_custom_attr("u1") is None # Tenant has no HRBP attr → None. assert FeishuContactRunner( client=_client(), custom_attr_catalog={"pinyin": {"id": "C-1", "type": "TEXT"}}, )._build_hrbp_custom_attr("u1") is None # HRBP attr exists but is the wrong type → None (don't send a # malformed value Feishu would 400 on). assert FeishuContactRunner( client=_client(), custom_attr_catalog={"hrbp": {"id": "C-714", "type": "TEXT"}}, )._build_hrbp_custom_attr("u1") is None @pytest.mark.asyncio async def test_hrbp_pushed_via_generic_user_attr(self, monkeypatch): """End-to-end: ``hrbp_user_id_email`` resolves to a Feishu user_id and lands as the HRBP GENERIC_USER custom attr in the PATCH body, ALONGSIDE the TEXT custom_attrs. The synthetic ``hrbp_user_id_email`` field must NOT leak into the wire body.""" runner = FeishuContactRunner( client=_client(), custom_attr_catalog={ "hrbp": {"id": "C-7143156434032312324", "type": "GENERIC_USER"}, "pinyin": {"id": "C-pinyin", "type": "TEXT"}, }, ) async def fake_resolve(email: str) -> str: if email == "hrbp.boss@pingcap.com": return "u_hrbp" raise AppException( code="FEISHU_USER_NOT_FOUND", message="no account", status_code=404, ) monkeypatch.setattr(runner._client, "resolve_user_id_by_email", fake_resolve) calls = _Calls([{"user": {"department_ids": []}}, {}]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert( person_email="alice@pingcap.com", payload={ "email": "alice@pingcap.com", "name": "Alice", "hrbp_user_id_email": "hrbp.boss@pingcap.com", "custom_attrs": {"pinyin": "Liyang Zhang"}, }, prior_external_id="u_alice", ) assert out.status == "success" body = calls[-1]["json_body"] # Synthetic resolution field never reaches Feishu. assert "hrbp_user_id_email" not in body attrs = body.get("custom_attrs") or [] hrbp = [a for a in attrs if a["id"] == "C-7143156434032312324"] assert hrbp == [{ "type": "GENERIC_USER", "id": "C-7143156434032312324", "value": {"generic_user": {"id": "u_hrbp", "type": 1}}, }] # TEXT attr still present too. assert any(a["id"] == "C-pinyin" for a in attrs) @pytest.mark.asyncio async def test_hrbp_omitted_when_unresolvable(self, monkeypatch): """HRBP with no Feishu account (FEISHU_USER_NOT_FOUND) → the HRBP attr is omitted (Feishu keeps its existing value); the rest of the push still goes through.""" runner = FeishuContactRunner( client=_client(), custom_attr_catalog={ "hrbp": {"id": "C-714", "type": "GENERIC_USER"}, }, ) async def fake_resolve(email: str) -> str: raise AppException( code="FEISHU_USER_NOT_FOUND", message="no account", status_code=404, ) monkeypatch.setattr(runner._client, "resolve_user_id_by_email", fake_resolve) calls = _Calls([{"user": {"department_ids": []}}, {}]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert( person_email="alice@pingcap.com", payload={ "email": "alice@pingcap.com", "name": "Alice", "hrbp_user_id_email": "ghost@pingcap.com", }, prior_external_id="u_alice", ) assert out.status == "success" body = calls[-1]["json_body"] assert "hrbp_user_id_email" not in body # No HRBP attr (and no custom_attrs slot at all here). assert not any( a.get("id") == "C-714" for a in (body.get("custom_attrs") or []) ) @pytest.mark.asyncio async def test_auth_failure_propagates_from_freeze_path( self, monkeypatch, ): """The freeze (deactivate) path mirrors upsert: systemic failures (bad creds / rate limit) propagate so an offboarding sweep doesn't write N freeze-failed state rows under the same bad credentials. """ runner = FeishuContactRunner(client=_client()) async def fake_resolve(_email: str) -> str: raise AppException( code="FEISHU_TOKEN_ERROR", message="Failed to get tenant_access_token", status_code=502, ) monkeypatch.setattr( runner._client, "resolve_user_id_by_email", fake_resolve, ) with pytest.raises(AppException) as ei: await runner.deactivate( person_email="leaver@pingcap.com", prior_external_id=None, ) assert ei.value.code == "FEISHU_TOKEN_ERROR" @pytest.mark.asyncio async def test_freeze_patch_rate_limit_propagates(self, monkeypatch): """A tenant rate-limit error on the freeze PATCH itself also halts the sweep — not recorded as per-leaver failure. """ runner = FeishuContactRunner(client=_client()) calls = _Calls([_err(99991671, "rate limited")]) monkeypatch.setattr(runner._client, "_request", calls) with pytest.raises(AppException) as ei: await runner.deactivate( person_email="leaver@pingcap.com", prior_external_id="u_leaver", ) assert "99991671" in (ei.value.message or "") @pytest.mark.asyncio async def test_freeze_success_verified_is_deactivated(self, monkeypatch): """Happy path: PATCH is_frozen=true succeeds AND the directory re-read confirms is_frozen=true → deactivated. This is the path that had NO test before the false-success fix.""" runner = FeishuContactRunner(client=_client()) calls = _Calls([ {}, # PATCH ok {"user": {"status": {"is_frozen": True, "is_resigned": False}}}, # verify GET ]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.deactivate( person_email="leaver@pingcap.com", prior_external_id="u_leaver", ) assert out.status == "deactivated" assert out.external_id == "u_leaver" assert calls[0]["method"] == "PATCH" assert calls[0]["json_body"] == {"status": {"is_frozen": True}} assert calls[1]["method"] == "GET" # verify re-read @pytest.mark.asyncio async def test_freeze_patch_ok_but_still_active_is_failed(self, monkeypatch): """REGRESSION (P01108, 2026-07-24): Feishu returns SUCCESS on the freeze PATCH but the account stays active (is_frozen=false) — e.g. a leaver with an in-flight self-approval. Feishu has no session-revoke API, so this is a leaver still logged in. Must be a hard failure (retry + alert), NOT a false green.""" runner = FeishuContactRunner(client=_client()) calls = _Calls([ {}, # PATCH "ok" {"user": {"status": { "is_frozen": False, "is_resigned": False, "is_exited": False, "is_activated": True, }}}, # verify GET — STILL ACTIVE ]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.deactivate( person_email="leaver@pingcap.com", prior_external_id="u_leaver", ) assert out.status == "failed" assert out.external_id == "u_leaver" assert "freeze did not take" in (out.error or "") @pytest.mark.asyncio async def test_freeze_already_frozen_verifies_to_deactivated(self, monkeypatch): """230002 'already frozen' is no longer trusted blindly — the re-read confirms is_frozen=true, then it's deactivated.""" runner = FeishuContactRunner(client=_client()) calls = _Calls([ _err(230002, "already frozen"), {"user": {"status": {"is_frozen": True}}}, # verify GET ]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.deactivate( person_email="leaver@pingcap.com", prior_external_id="u_leaver", ) assert out.status == "deactivated" assert out.external_id == "u_leaver" assert calls[0]["json_body"] == {"status": {"is_frozen": True}} @pytest.mark.asyncio async def test_freeze_benign_code_but_still_active_is_failed(self, monkeypatch): """A benign idempotent code (already-frozen) that does NOT match the real directory state surfaces as a failure, not a false green.""" runner = FeishuContactRunner(client=_client()) calls = _Calls([ _err(230002, "already frozen"), {"user": {"status": {"is_frozen": False, "is_resigned": False}}}, ]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.deactivate( person_email="leaver@pingcap.com", prior_external_id="u_leaver", ) assert out.status == "failed" @pytest.mark.asyncio async def test_freeze_verify_read_failure_is_failed(self, monkeypatch): """If the freeze cannot be VERIFIED (status re-read errors), fail safe (retry next sweep) rather than claim a false success.""" runner = FeishuContactRunner(client=_client()) calls = _Calls([ {}, # PATCH ok _err(90002, "transient read blip"), # verify GET fails (non-systemic) ]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.deactivate( person_email="leaver@pingcap.com", prior_external_id="u_leaver", ) assert out.status == "failed" @pytest.mark.asyncio async def test_freeze_user_absent_is_success(self, monkeypatch): runner = FeishuContactRunner(client=_client()) calls = _Calls([_err(99991663, "user gone")]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.deactivate( person_email="gone@pingcap.com", prior_external_id="u_gone", ) assert out.status == "deactivated" @pytest.mark.asyncio async def test_freeze_user_resigned_verifies_to_deactivated(self, monkeypatch): """42006 'user has resigned' → the re-read confirms is_resigned=true (a terminal off-state, access cut) so it stays idempotent success and does NOT re-alert daily. But it's now VERIFIED, not blindly trusted. """ runner = FeishuContactRunner(client=_client()) calls = _Calls([ _err(42006, "user has resigned"), {"user": {"status": {"is_resigned": True, "is_frozen": False}}}, ]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.deactivate( person_email="resigned@pingcap.com", prior_external_id="u_resigned", ) assert out.status == "deactivated" assert out.external_id == "u_resigned" assert out.error is None @pytest.mark.asyncio async def test_freeze_lookup_miss_is_success_no_call(self, monkeypatch): runner = FeishuContactRunner(client=_client()) async def fake_resolve(email: str) -> str: raise AppException( code="FEISHU_USER_NOT_FOUND", message="not found", status_code=400, ) monkeypatch.setattr(runner._client, "resolve_user_id_by_email", fake_resolve) calls = _Calls([]) # No PATCH expected. monkeypatch.setattr(runner._client, "_request", calls) out = await runner.deactivate( person_email="ghost@pingcap.com", prior_external_id=None, ) assert out.status == "deactivated" assert out.external_id is None assert len(calls) == 0 def test_supports_delete_is_false(self): runner = FeishuContactRunner(client=_client()) assert runner.supports_delete is False assert runner.supports_deactivate is True @pytest.mark.asyncio async def test_auth_failure_propagates_from_email_lookup( self, monkeypatch, ): """A FEISHU_AUTH_FAILURE-coded error from the email lookup propagates out of upsert (not returned as PersonOutcome) so the orchestrator can short-circuit the whole sweep with one clear error instead of burning rate limit on N more lookups under the same bad creds. """ runner = FeishuContactRunner(client=_client()) async def fake_resolve(_email: str) -> str: raise AppException( code="FEISHU_API_ERROR", message="Feishu API error 99991661: invalid tenant_access_token", status_code=502, ) monkeypatch.setattr( runner._client, "resolve_user_id_by_email", fake_resolve, ) with pytest.raises(AppException) as ei: await runner.upsert( person_email="anyone@pingcap.com", payload={"email": "anyone@pingcap.com", "name": "X"}, prior_external_id=None, ) assert "99991661" in (ei.value.message or "") @pytest.mark.asyncio async def test_token_error_propagates_from_email_lookup( self, monkeypatch, ): """``FEISHU_TOKEN_ERROR`` exceptions are raised BEFORE any request goes out (the token-acquisition itself failed). They carry no numeric Feishu code, so the per-numeric-code check wouldn't match. ``is_systemic_feishu_failure`` detects them via exception code string so they still halt the sweep. """ runner = FeishuContactRunner(client=_client()) async def fake_resolve(_email: str) -> str: raise AppException( code="FEISHU_TOKEN_ERROR", message="Failed to get Feishu tenant access token", status_code=502, ) monkeypatch.setattr( runner._client, "resolve_user_id_by_email", fake_resolve, ) with pytest.raises(AppException) as ei: await runner.upsert( person_email="anyone@pingcap.com", payload={"email": "anyone@pingcap.com", "name": "X"}, prior_external_id=None, ) assert ei.value.code == "FEISHU_TOKEN_ERROR" @pytest.mark.asyncio async def test_auth_failure_propagates_from_leader_lookup( self, monkeypatch, ): """Same auth-failure propagation rule for the leader lookup path — rate-limit / token error during leader resolution halts the sweep instead of silently dropping the leader slot for every user. """ runner = FeishuContactRunner(client=_client()) # Email lookup succeeds; leader lookup blows up under auth. resolve_calls: list[str] = [] async def fake_resolve(email: str) -> str: resolve_calls.append(email) if email == "alice@pingcap.com": return "u_alice" raise AppException( code="FEISHU_API_ERROR", message="Feishu API error 99991671: rate limit", status_code=502, ) monkeypatch.setattr( runner._client, "resolve_user_id_by_email", fake_resolve, ) with pytest.raises(AppException) as ei: await runner.upsert( person_email="alice@pingcap.com", payload={ "email": "alice@pingcap.com", "name": "Alice", "leader_user_id_email": "boss@pingcap.com", }, prior_external_id="u_alice", ) assert "99991671" in (ei.value.message or "") @pytest.mark.asyncio async def test_230003_with_dept_message_retries_without_dept( self, monkeypatch, ): """Feishu's overloaded 230003 code (user-absent vs dept-not-found) is disambiguated by checking whether the error message mentions "department" or "部门". When the message suggests a dept-scope error AND the request carried department_ids, the runner retries without department_ids instead of treating the user as absent and recreating. The retry succeeds on the non-dept fields but the dept slot was dropped — a partial push — so the outcome is ``failed`` (not a hash-freezing ``success``); the next sweep retries once dept sync re-adopts the stale od (#1737). """ runner = FeishuContactRunner( client=_client(), helix_dept_to_external={"1809808": "od-stale"}, ) calls = _Calls([ {"user": {"department_ids": ["od-stale", "od-keeper"]}}, # First PATCH fails with 230003 + dept-shaped message. AppException( code="FEISHU_API_ERROR", message=( "Feishu API error 230003: department id not found" ), status_code=502, ), # Fallback PATCH without department_ids succeeds. {}, ]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert( person_email="alice@pingcap.com", payload={ "email": "alice@pingcap.com", "department_ids": "1809808", }, prior_external_id="u_alice", ) assert out.status == "failed" assert "partial dept push" in (out.error or "") # The second PATCH must omit department_ids (the retry shape). assert "department_ids" not in calls[-1]["json_body"] # NO POST issued — the runner did NOT treat 230003 as user- # absent here. assert all(c["method"] != "POST" for c in calls) @pytest.mark.asyncio async def test_230003_with_user_message_recreates(self, monkeypatch): """Mirror of the above — 230003 with a user-scope message and NO department_ids in the body falls through to recreate, not to the dept-scope retry. Regression anchor for the residency-hint strip: ``_patch_user`` removes ``_helix_residence_country`` before its Contact PATCH, but the user-absent branch hands the SAME body to ``_create_user``, which blocks without it. Stripping it unconditionally turned every PATCH→recreate into ``blocked/missing_residence_country`` with no create attempted at all, so a deleted Feishu account was never rebuilt. The hint is now restored on the recreate branch. """ runner = FeishuContactRunner(client=_client()) calls = _Calls([ {"user": {"department_ids": []}}, # PATCH 230003 user-absent (no dept terminology). AppException( code="FEISHU_API_ERROR", message="Feishu API error 230003: user not found", status_code=502, ), # Recreate goes through Directory v1 (geo fixed at creation, # 3a0adf7e2) — NOT the old Contact v3 POST. {"employee_id": "u_re_alice"}, {}, # Contact projection PATCH ]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert( person_email="alice@pingcap.com", payload={ "email": "alice@pingcap.com", "name": "Alice", "_helix_residence_country": "CN", }, prior_external_id="u_alice", ) assert out.status == "success" assert out.external_id == "u_re_alice" methods = [c["method"] for c in calls] assert methods == ["GET", "PATCH", "POST", "PATCH"] assert calls[2]["path"].startswith("/directory/v1/employees") assert calls[2]["json_body"]["options"]["geo_name"] == "cn" # The projection lands on the NEWLY created id, not the dead one. assert "u_re_alice" in calls[3]["path"] @pytest.mark.asyncio async def test_projection_absent_does_not_create_a_second_account( self, monkeypatch, ): """Read-after-write window: the Contact projection that follows a create can see the brand-new user as absent. That must NOT be taken as "the person doesn't exist" and recreate them — that builds a SECOND account for the same hire, and recurses. This is the flip side of the recreate above: once the residency hint survives the PATCH→create handoff, the projection's own user-absent branch becomes a live path to a duplicate account, so the projection runs with ``allow_recreate=False``. The account exists and its id is known, so the outcome is a retryable failure that KEEPS the mapping — the next sweep PATCHes the existing user and converges. """ runner = FeishuContactRunner(client=_client()) calls = _Calls([ {"user_list": []}, # email miss → create {"employee_id": "u_fresh"}, # Directory create OK _err(230003, "user not found"), # projection PATCH: not there yet ]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert( person_email="fresh@pingcap.com", payload={ "email": "fresh@pingcap.com", "name": "Fresh", "_helix_residence_country": "CN", }, prior_external_id=None, ) assert out.status == "failed" assert out.external_id == "u_fresh" # mapping survives → next sweep PATCHes # Exactly ONE create. No second Directory POST. assert len([ c for c in calls if c["method"] == "POST" and c["path"].startswith("/directory/v1/employees") ]) == 1 # ── Office dimension (second dept axis) ───────────────────────────── class TestOfficeDimension: """Office-location → Feishu office dept is a SECOND dimension on top of HRIS BU. Both slices are Helix-managed; the merge step treats them symmetrically. """ @pytest.mark.asyncio async def test_office_dept_name_is_appended_to_department_ids( self, monkeypatch, ): runner = FeishuContactRunner( client=_client(), helix_dept_to_external={"1809808": "od-eng"}, # HRIS Eng Group office_name_to_external={"Tokyo Office": "od-tokyo"}, ) calls = _Calls([ # GET current dept_ids: empty (new user) {"user": {"department_ids": []}}, {}, # PATCH success ]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert( person_email="alice@pingcap.com", payload={ "email": "alice@pingcap.com", "name": "Alice", "department_ids": "1809808", "office_dept_name": "Tokyo Office", }, prior_external_id="u_alice", ) assert out.status == "success" body = calls[-1]["json_body"] # BOTH HRIS BU and office dept end up on the user. assert "od-eng" in body["department_ids"] assert "od-tokyo" in body["department_ids"] # ``office_dept_name`` is consumed — never leaks to the wire body. assert "office_dept_name" not in body @pytest.mark.asyncio async def test_relocation_replaces_old_office(self, monkeypatch): """User was in od-tokyo; HRIS says they're now in Singapore. Merge MUST drop od-tokyo and add od-sg (office is Helix-managed). """ runner = FeishuContactRunner( client=_client(), helix_dept_to_external={"1809808": "od-eng"}, office_name_to_external={ "Tokyo Office": "od-tokyo", "SG Office": "od-sg", }, ) calls = _Calls([ {"user": {"department_ids": ["od-eng", "od-tokyo"]}}, {}, ]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert( person_email="alice@pingcap.com", payload={ "email": "alice@pingcap.com", "department_ids": "1809808", "office_dept_name": "SG Office", }, prior_external_id="u_alice", ) assert out.status == "success" body = calls[-1]["json_body"] # Old Tokyo gone, new SG present, Eng BU retained. assert "od-tokyo" not in body["department_ids"] assert "od-sg" in body["department_ids"] assert "od-eng" in body["department_ids"] @pytest.mark.asyncio async def test_office_miss_does_not_break_user_push(self, monkeypatch): """Admin hasn't curated a mapping for "Shanghai" → no office_dept_name rendered, OR rendered name not in the live Feishu catalog. User still gets pushed with just the HRIS BU dept; office slot skipped silently. """ runner = FeishuContactRunner( client=_client(), helix_dept_to_external={"1809808": "od-eng"}, office_name_to_external={"Tokyo Office": "od-tokyo"}, ) calls = _Calls([ {"user": {"department_ids": []}}, {}, ]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert( person_email="alice@pingcap.com", payload={ "email": "alice@pingcap.com", "department_ids": "1809808", # No office_dept_name (admin didn't map Shanghai → # any office dept). }, prior_external_id="u_alice", ) assert out.status == "success" body = calls[-1]["json_body"] assert body["department_ids"] == ["od-eng"] @pytest.mark.asyncio async def test_office_in_helix_managed_universe(self, monkeypatch): """Even when a user's payload doesn't carry office_dept_name today (admin didn't curate this person's work_location), their existing od-tokyo on Feishu is STILL considered Helix-managed because Tokyo Office is in office_name_to_external. So if the user moves OUT of Tokyo (work_location cleared), the next sync DROPS od-tokyo from their depts. """ runner = FeishuContactRunner( client=_client(), helix_dept_to_external={"1809808": "od-eng"}, office_name_to_external={"Tokyo Office": "od-tokyo"}, ) calls = _Calls([ {"user": {"department_ids": ["od-eng", "od-tokyo", "od-authing-only"]}}, {}, ]) monkeypatch.setattr(runner._client, "_request", calls) out = await runner.upsert( person_email="alice@pingcap.com", payload={ "email": "alice@pingcap.com", "department_ids": "1809808", # No office_dept_name on this push. }, prior_external_id="u_alice", ) assert out.status == "success" body = calls[-1]["json_body"] # od-tokyo dropped (Helix-managed but not in this push). assert "od-tokyo" not in body["department_ids"] # od-authing-only preserved (NOT in either Helix index). assert "od-authing-only" in body["department_ids"] # Current HRIS BU kept. assert "od-eng" in body["department_ids"] def test_default_office_mapping_covers_pingcap_offices(self): from app.domains.identity_sync.presets import ( OFFICE_LOCATION_TO_FEISHU_DEPT, ) # Sanity check the shipped defaults — the offices PingCAP # actually has people in today. assert OFFICE_LOCATION_TO_FEISHU_DEPT["Tokyo"] == "Tokyo Office" assert OFFICE_LOCATION_TO_FEISHU_DEPT["Singapore"] == "SG Office" assert OFFICE_LOCATION_TO_FEISHU_DEPT["San Mateo"] == "SV Office" assert OFFICE_LOCATION_TO_FEISHU_DEPT["Selangor"] == "MY Office" def test_office_dept_rule_keyed_on_office_code_not_city(self): """The Feishu office-dept dimension must be driven by the HRIS office (``office_code``), NOT work_location (residence city) — an office spans many cities, so the old ``{location}`` rule under- assigned. Mirrors the Google Office-group axis.""" from app.domains.identity_sync.presets import ( FEISHU_CONTACT_USER_MAPPING_RULES, OFFICE_CODE_TO_FEISHU_LOCATION, ) office_rules = [ r for r in FEISHU_CONTACT_USER_MAPPING_RULES if r.get("target_field") == "office_dept_name" ] assert len(office_rules) == 1 rule = office_rules[0] assert rule["source_template"] == "{office_code}" # SV/SG/Tokyo/MY office codes map to the Feishu office-dept names. assert rule["value_map"]["2653"] == "SV Office" assert rule["value_map"]["2661"] == "SG Office" assert rule["value_map"]["2667"] == "Tokyo Office" assert rule["value_map"]["2865"] == "MY Office" assert rule["value_map"] == OFFICE_CODE_TO_FEISHU_LOCATION # ── Office mapping extraction (service helper) ────────────────────── class TestExtractOfficeNames: def test_returns_value_map_values(self): from app.domains.identity_sync.service import IdentitySyncService cfg = { "rules": [ { "source_template": "{location}", "value_map": { "Tokyo": "Tokyo Office", "Singapore": "SG Office", }, "target_field": "office_dept_name", }, # Decoy: another rule with a different target_field. { "source_template": "{country}", "value_map": {"CN": "China"}, "target_field": "country", }, ], } names = IdentitySyncService._extract_office_names_from_mapping(cfg) assert names == frozenset({"Tokyo Office", "SG Office"}) def test_empty_when_no_office_rule(self): from app.domains.identity_sync.service import IdentitySyncService names = IdentitySyncService._extract_office_names_from_mapping( {"rules": []}, ) assert names == frozenset() def test_empty_when_none_input(self): from app.domains.identity_sync.service import IdentitySyncService assert ( IdentitySyncService._extract_office_names_from_mapping(None) == frozenset() ) # ── Pure-function helpers ─────────────────────────────────────────── class TestNormaliseAttrName: def test_simple(self): assert _normalise_attr_name("PinYin") == "pinyin" def test_with_space(self): # Critical case: Feishu has "Company Entity" with space; the # runner indexes it to "company_entity" so the mapping rule # `custom_attrs.company_entity` resolves. assert _normalise_attr_name("Company Entity") == "company_entity" assert _normalise_attr_name("Job Function") == "job_function" def test_strips_punctuation(self): assert _normalise_attr_name("[D] Email") == "d_email" def test_empty(self): assert _normalise_attr_name("") == "" assert _normalise_attr_name(None) == "" def test_zh_unchanged(self): # zh_cn locale strings — non-ASCII gets normalised to "_" runs; # acceptable because the runner ALSO indexes the default + en_us # locales (PingCAP's "Company Entity" has both Chinese and # English i18n entries, English wins). assert _normalise_attr_name("公司") == "" class TestCoerceDepartmentIds: def test_string_single(self): assert _coerce_department_ids("1809808") == ["1809808"] def test_string_csv(self): assert _coerce_department_ids("1809808,1809817") == ["1809808", "1809817"] def test_list(self): assert _coerce_department_ids(["a", "b"]) == ["a", "b"] def test_empty(self): assert _coerce_department_ids("") == [] assert _coerce_department_ids(None) == [] class TestResponseMentionsDepartment: """The dept-vs-user 230003 disambiguator. Verify both EN and zh-CN markers match, and unrelated messages don't. """ def test_en_department_substring_matches(self): from app.domains.identity_sync.runners.feishu_contact import ( _response_mentions_department, ) exc = AppException( code="FEISHU_API_ERROR", message="Feishu API error 230003: department id not found", status_code=502, ) assert _response_mentions_department(exc) is True def test_zh_cn_部门_substring_matches(self): from app.domains.identity_sync.runners.feishu_contact import ( _response_mentions_department, ) exc = AppException( code="FEISHU_API_ERROR", message="Feishu API error 230003: 部门不存在", status_code=502, ) assert _response_mentions_department(exc) is True def test_user_message_does_not_match(self): from app.domains.identity_sync.runners.feishu_contact import ( _response_mentions_department, ) exc = AppException( code="FEISHU_API_ERROR", message="Feishu API error 230003: user not found", status_code=502, ) assert _response_mentions_department(exc) is False class TestParseFeishuCode: def test_extracts_from_canonical_format(self): assert _parse_feishu_code(_err(230002, "already done")) == 230002 def test_returns_none_on_unknown_format(self): exc = AppException(code="OTHER", message="boom", status_code=500) assert _parse_feishu_code(exc) is None class TestIsSystemicFeishuFailure: """The systemic-failure heuristic drives sweep-halt-on-bad-creds. Pin every code we care about so a future Feishu code change can't silently disable the circuit-breaker. """ def test_token_error_is_systemic(self): from app.approval.feishu_client import is_systemic_feishu_failure exc = AppException( code="FEISHU_TOKEN_ERROR", message="Failed to get tenant_access_token", status_code=502, ) assert is_systemic_feishu_failure(exc) is True def test_invalid_tenant_access_token_is_systemic(self): from app.approval.feishu_client import is_systemic_feishu_failure assert is_systemic_feishu_failure(_err(99991661, "token bad")) is True def test_rate_limit_is_systemic(self): from app.approval.feishu_client import is_systemic_feishu_failure assert is_systemic_feishu_failure(_err(99991671, "rate limited")) is True assert is_systemic_feishu_failure(_err(99991672, "freq limit")) is True def test_invalid_app_creds_are_systemic(self): from app.approval.feishu_client import is_systemic_feishu_failure assert is_systemic_feishu_failure(_err(91402, "invalid app_id")) is True assert is_systemic_feishu_failure(_err(91403, "invalid secret")) is True def test_per_record_errors_are_NOT_systemic(self): from app.approval.feishu_client import is_systemic_feishu_failure # Real per-record errors — must NOT halt the sweep. assert is_systemic_feishu_failure(_err(230002, "already frozen")) is False assert is_systemic_feishu_failure(_err(230003, "user not found")) is False assert is_systemic_feishu_failure(_err(99991663, "resource gone")) is False assert is_systemic_feishu_failure(_err(99992402, "field validation")) is False def test_unknown_exception_codes_are_NOT_systemic(self): from app.approval.feishu_client import is_systemic_feishu_failure exc = AppException(code="OTHER", message="boom", status_code=500) assert is_systemic_feishu_failure(exc) is False class TestNormaliseDeptName: def test_passes_simple_through_lowered(self): assert _normalise_name("Engineering Group") == "engineering group" def test_collapses_whitespace(self): assert _normalise_name(" Engineering Group ") == "engineering group" # ── Preset snapshot ───────────────────────────────────────────────── class TestFeishuPresets: def test_combined_preset_resolves(self): p = get_preset("feishu_contact") assert p is not None assert p.kind == "feishu_contact" assert p.requires_integration_kind == "identity_sync_feishu" # No runner_options today — dept phase is admin-invariant, # user phase falls back to Feishu root "0" on hire-day misses. assert p.runner_options == {} def test_user_mapping_rules_snapshot(self): """Mapped custom_attrs align with Feishu live state: pinyin, company_entity, location, contact_email, github_id (+ HRBP GENERIC_USER). gender / employee_type are standard Feishu fields sourced from SF (migration 344); github_id from SF cust_Other (migration 369); Job Function / Job Family are intentionally NOT mapped (no clean SF source). """ targets = [r["target_field"] for r in FEISHU_CONTACT_USER_MAPPING_RULES] assert targets == [ "email", "name", "en_name", "employee_no", "department_ids", "office_dept_name", # second dept dimension (work_location → office) "leader_user_id_email", # HRBP — SF EmpJobRelationships (2448) → Feishu GENERIC_USER # custom attr; runner resolves email → user_id. "hrbp_user_id_email", "job_title", # city — pushed from HRIS (2026-07-01 decision). ``country`` # drives Feishu Region; value-mapped ISO-3 → ISO-2. "city", "country", # SF-sourced enums (migration 344), int-coerced by the runner. "gender", "employee_type", # Authing-parity fields (2026-06-11): the legacy 飞书#0 # task pushed 手机号 + 入职日期; join_time is converted to # unix seconds by the runner's body builder. "mobile", # Synthetic: raw HRIS country (ISO-3) so the runner can tell a # set-but-unmapped country (drop the mobile) from an empty one # (+86 default). Popped before building the Feishu body. "mobile_country_iso3", # Synthetic: the mobile's OWN country ISO-2 (SF PerPhone.countryCode # → externalCode). Takes precedence in the runner's mobile # formatter so a host-country phone on an international assignee # (home CN, phone +65 SG) formats correctly. Popped before the # Feishu body. "mobile_country_iso2", "join_time", "custom_attrs.pinyin", "custom_attrs.company_entity", "custom_attrs.location", "custom_attrs.contact_email", # GitHub ID — SF cust_Other.cust_id (migration 369), backfilled # since the 2026-06-26 "no source" ADR. Resolves to the live # "GitHub ID" schema by normalised i18n_name. "custom_attrs.github_id", # Job Function / Job Family — SF FOJobFunction / cust_JobFamily # names (2026-07-01 decision: HRIS source of truth, may overwrite # Feishu's 北森-era value). PinYin above is derived CJK-only. "custom_attrs.job_function", "custom_attrs.job_family", # NOTE: ``Beisen ID`` NOT mapped (北森 retired, no SF source → # preserve). ``Last Updated`` NOT mapped — a fresh timestamp # every render would defeat the payload-hash skip. ] def test_phone_country_rule_sources_the_normalised_token(self): """The phone-country slot reads the DERIVED ``{mobile_country_iso2}`` token, never raw ``{mobile_country_code}`` behind an ISO-3 value-map. SF's CountryCode picklist externalCodes are mixed ISO-2 / ISO-3 / free text (live 2026-09-01: sf-cn-prod 17/19 ISO-2, sf-eu-prod 17/27). An ISO-3-keyed map misses every already-correct ISO-2 value, so the formatter fell back to the person's WORK country and welded the wrong dialing code onto 18 prod numbers (and got 9 more rejected 41004). A value_map here at all is the bug — assert its absence, not its contents. """ rule = next( rule for rule in FEISHU_CONTACT_USER_MAPPING_RULES if rule["target_field"] == "mobile_country_iso2" ) assert rule["source_template"] == "{mobile_country_iso2}" assert not rule.get("value_map") def test_preset_carries_only_user_phase_rules(self): """Preset's mapping_rules are USER-phase only — dept phase runs from a fixed projection inside the runner, not from admin-customizable mapping rules. """ p = get_preset("feishu_contact") assert p is not None # The mapping_rules attribute exactly equals USER rules. assert [r["target_field"] for r in p.mapping_rules] == [ r["target_field"] for r in FEISHU_CONTACT_USER_MAPPING_RULES ] # ── Topological sort ─────────────────────────────────────────────── class _DeptNode: """Minimal IdentityOrgNode stand-in for topo-sort tests.""" def __init__(self, code: str, parent: str | None, level: int | None = None): self.org_node_source_id = code self.parent_org_node_source_id = parent self.org_level = level self.org_node_name = code class TestTopologicalSort: def test_parent_before_child(self): from app.domains.identity_sync.service import IdentitySyncService nodes = [ _DeptNode("C", "B"), _DeptNode("A", None), _DeptNode("B", "A"), ] ordered = IdentitySyncService._topological_sort_depts(nodes) codes = [n.org_node_source_id for n in ordered] assert codes.index("A") < codes.index("B") assert codes.index("B") < codes.index("C") def test_detached_subtree_still_flushed(self): from app.domains.identity_sync.service import IdentitySyncService nodes = [_DeptNode("ORPHAN", "DOES_NOT_EXIST")] ordered = IdentitySyncService._topological_sort_depts(nodes) assert [n.org_node_source_id for n in ordered] == ["ORPHAN"] # ── Service-level structural tests ───────────────────────────────── class TestSystemicFailurePropagatesThroughSweep: """The R3+logic-scan fix: ``_sweep``'s ``except`` clause MUST disambiguate ``is_systemic_feishu_failure(e)`` AppException re-raises from the runner and propagate them, instead of catching them under bare ``except Exception`` and converting to per-person ``PersonOutcome.failed``. Without this, the entire circuit-break introduced in R1/R2/R3 is dead code (the runner's ``raise`` gets eaten by the orchestrator). Verified by inspecting the source rather than running a full sweep — full-sweep tests would need a real DB session fixture. """ def test_sweep_disambiguates_appexception_systemic(self): import inspect from app.domains.identity_sync.service import IdentitySyncService src = inspect.getsource(IdentitySyncService._sweep) # The sweep MUST catch AppException SEPARATELY and consult # is_systemic_feishu_failure before swallowing. assert "except AppException" in src assert "is_systemic_feishu_failure" in src def test_dept_phase_disambiguates_appexception_systemic(self): import inspect from app.domains.identity_sync.service import IdentitySyncService src = inspect.getsource( IdentitySyncService._run_feishu_dept_phase, ) # Same gate on Phase A — both upsert_dept and soft_delete # call sites must propagate systemic AppExceptions. # # Two equivalent shapes count, because the upsert_dept site moved to # a SAVEPOINT (2026-08-06): a DB-level failure inside the runner used # to deactivate the transaction, taking every later dept in the sweep # AND the systemic branch's own ``self.db.commit()`` down with it. # ``best_effort`` yields the exception on an outcome instead of an # ``except`` clause, so the disambiguation reads # ``isinstance(e, AppException)`` there. What must stay true is that # BOTH sites consult ``is_systemic_feishu_failure`` and re-raise. import re as _re disambiguations = ( src.count("except AppException") + len(_re.findall(r"isinstance\(\w+, AppException\)", src)) ) assert disambiguations >= 2, ( "both upsert_dept and soft_delete must disambiguate AppException" ) assert src.count("is_systemic_feishu_failure") >= 2 # …and each must actually propagate rather than swallow. raises = _re.findall(r"\n\s+raise\b[^\n]*", src) assert len(raises) >= 2, ( f"both sites must propagate, found {raises}" ) class TestLoadHelixDeptIndexScopedByTarget: """The logic-scan fix: ``_load_helix_dept_index`` must filter by ``target_id`` so a future second feishu_contact target (e.g. staging vs prod) doesn't leak its dept adoptions into the first target's user-runner Phase B index. """ def test_signature_requires_target_id(self): import inspect from app.domains.identity_sync.service import IdentitySyncService sig = inspect.signature(IdentitySyncService._load_helix_dept_index) # Must be a kwarg, MUST be present. assert "target_id" in sig.parameters def test_sql_filters_by_target_id(self): import inspect from app.domains.identity_sync.service import IdentitySyncService src = inspect.getsource( IdentitySyncService._load_helix_dept_index, ) # SQL must reference target_id == target_id binding. assert "IdentitySyncDepartmentState.target_id == target_id" in src class TestBrakeMarkerConstant: """The brake-marker substring written by Phase A is the contract ``_load_helix_dept_index`` keys off to exclude brake-blocked rows. Drift between the two ends silently disables the brake guard. """ def test_brake_message_format_matches_filter_marker(self): from app.domains.identity_sync.service import IdentitySyncService # Anything the brake writes must START WITH this marker so the # SQL LIKE filter catches it. marker = IdentitySyncService._DEPT_BRAKE_ERROR_MARKER assert marker == "mass-action brake (dept):" # The marker must appear in BOTH brake messages the service # writes — the per-row last_error AND the run-summary error. # Defended by the LIKE filter, but the human-readable text # must also start with the marker for grep-ability. import inspect from app.domains.identity_sync import service as svc_mod src = inspect.getsource(svc_mod) # The brake constant appears in (a) the constant declaration, # (b) the SQL LIKE filter doc, (c) two writes inside # _run_feishu_dept_phase. So at least 4 occurrences. assert src.count(marker) >= 4 class TestFinaliseRunSyntheticErrorMessage: """When Phase B succeeded but Phase A had failures and the caller didn't pass an explicit error_message, ``_finalise_run`` must synthesise one referencing the dept counters so the runs list view surfaces Phase A trouble. Without this, admin sees ``status=partial`` + ``dept_failed=N`` + ``error_message=None`` in the list view and can't tell what went wrong from the list. """ def test_synthesis_logic_present(self): import inspect from app.domains.identity_sync.service import IdentitySyncService src = inspect.getsource(IdentitySyncService._finalise_run) # Defensive: the synthesis MUST mention dept_failed + # result_summary.errors so future refactors don't quietly # drop this guarantee. assert "summary.dept_failed > 0" in src assert "result_summary.errors" in src class TestFeishuProvisioningRunnerOptions: """``_validate_runner_options`` for the Authing-parity provisioning knobs: ``reopen_frozen`` (bool, default off) and the admin skip-list ``skip_provision_person_ids`` (list of person UUID strings). The feishu_contact branch used to return ``{}`` and would silently drop these — this guards the round-trip + rejection of bad ids. """ def test_defaults_off_and_empty(self): from app.domains.identity_sync.service import _validate_runner_options opts = _validate_runner_options("feishu_contact", {}) assert opts == {"reopen_frozen": False, "skip_provision_person_ids": []} def test_valid_reopen_and_skiplist_preserved(self): import uuid from app.domains.identity_sync.service import _validate_runner_options a, b = str(uuid.uuid4()), str(uuid.uuid4()) opts = _validate_runner_options( "feishu_contact", {"reopen_frozen": True, "skip_provision_person_ids": [a, b, a]}, ) assert opts["reopen_frozen"] is True # de-duplicated, order preserved assert opts["skip_provision_person_ids"] == [a, b] def test_bad_uuid_in_skiplist_rejected(self): import pytest from app.core.exceptions import AppException from app.domains.identity_sync.service import _validate_runner_options with pytest.raises(AppException): _validate_runner_options( "feishu_contact", {"skip_provision_person_ids": ["not-a-uuid"]}, )