"""汇联易 runner behavior — lookup-first upsert, 离职 idempotency, dept POST/PUT fallbacks, token client error wrapping. Wire calls are stubbed at ``HuilianyiClient.request`` (and ``get_access_token``) so the tests pin the RUNNER's decision tree, not httpx plumbing. Vendor response shapes come from opendocs.huilianyi.com (2026-06-10): * mutations answer ``{"errorCode": "0000", "message": ...}`` * 员工不存在 = 120207, 部门编码已存在 = 120107, 部门编码不存在 = 120110, 子部门/员工未清 = 120102 / 120124 (子部门存在员工) / 120125 (启用的子部门未清) * employee detail GET returns the row (``status``: 1001 active / 1003 resigned) without an errorCode wrapper """ from __future__ import annotations from typing import Any import pytest from app.domains.identity_sync.runners.huilianyi import ( HuilianyiClient, HuilianyiDepartmentRunner, HuilianyiError, HuilianyiRunner, ) pytestmark = pytest.mark.asyncio OK = {"errorCode": "0000", "message": "operation completed"} USER_MISSING = {"errorCode": "120207", "message": "员工不存在"} DEPT_CODE_EXISTS = {"errorCode": "120107", "message": "部门编码已存在"} DEPT_MISSING = {"errorCode": "120110", "message": "部门编码不存在"} DEPT_BUSY = {"errorCode": "120102", "message": "存在子部门或员工"} DEPT_BUSY_ENABLED_CHILD = { "errorCode": "120125", "message": "部门存在启用的子部门，无法禁用", } DEPT_BUSY_HAS_EMPLOYEES = { "errorCode": "120124", "message": "部门或者子部门存在员工,无法禁用", } def _client() -> HuilianyiClient: return HuilianyiClient( base_url="https://api.huilianyi.example", client_id="corp-id", client_secret="corp-secret", ) class _ScriptedCalls: """Replays (status, body, raw) per (method, path-prefix) and logs every request the runner makes.""" def __init__(self, script: list[tuple[str, str, tuple[int, Any, str]]]): self.script = list(script) self.calls: list[tuple[str, str, dict[str, Any] | None]] = [] async def __call__( self, method: str, path: str, *, json_body: dict[str, Any] | None = None, params: dict[str, Any] | None = None, retry_on_429: int = 2, ): self.calls.append((method, path, json_body)) for i, (m, prefix, resp) in enumerate(self.script): if m == method and path.startswith(prefix): self.script.pop(i) status, body, raw = resp return status, body, raw raise AssertionError(f"unexpected call {method} {path}") def _user_runner(script) -> tuple[HuilianyiRunner, _ScriptedCalls]: runner = HuilianyiRunner(client=_client()) calls = _ScriptedCalls(script) runner._client.request = calls # type: ignore[method-assign] return runner, calls def _dept_runner(script) -> tuple[HuilianyiDepartmentRunner, _ScriptedCalls]: runner = HuilianyiDepartmentRunner(client=_client()) calls = _ScriptedCalls(script) runner._client.request = calls # type: ignore[method-assign] return runner, calls PAYLOAD = { "employeeID": "P01431", "fullName": "吴明欣", "email": "mingxin.wu@pingcap.com", "custDeptNumber": "DEP-FIN", "companyCode": "PCBJ", } class TestUserUpsert: async def test_create_when_lookup_misses(self): runner, calls = _user_runner([ ("GET", "/api/open/user/detail/P01431", (404, USER_MISSING, "")), ("POST", "/api/open/user/create/v2", (200, OK, "")), ]) out = await runner.upsert( person_email="mingxin.wu@pingcap.com", payload=dict(PAYLOAD), prior_external_id=None, ) assert out.status == "success" assert out.external_id == "P01431" assert calls.calls[1][0] == "POST" async def test_update_when_lookup_hits_active(self): runner, calls = _user_runner([ ("GET", "/api/open/user/detail/P01431", (200, {"employeeID": "P01431", "status": 1001}, "")), ("PUT", "/api/open/user/v2", (200, OK, "")), ]) out = await runner.upsert( person_email="mingxin.wu@pingcap.com", payload=dict(PAYLOAD), prior_external_id="P01431", ) assert out.status == "success" assert calls.calls[1][0] == "PUT" async def test_resigned_vendor_row_surfaces_not_rehired(self): runner, calls = _user_runner([ ("GET", "/api/open/user/detail/P01431", (200, {"employeeID": "P01431", "status": 1003}, "")), ]) out = await runner.upsert( person_email="mingxin.wu@pingcap.com", payload=dict(PAYLOAD), prior_external_id="P01431", ) assert out.status == "blocked" assert "离职" in (out.error or "") # No mutation attempted against the resigned row. assert len(calls.calls) == 1 async def test_missing_company_code_fails_with_value_map_hint(self): # The mapping engine DROPS a value_map-miss field entirely — # without this guard an unmapped legal entity would silently # keep/misfile the employee under the old company. payload = {k: v for k, v in PAYLOAD.items() if k != "companyCode"} runner, calls = _user_runner([]) out = await runner.upsert( person_email="mingxin.wu@pingcap.com", payload=payload, prior_external_id=None, ) assert out.status == "blocked" assert "value_map" in (out.error or "") assert calls.calls == [] async def test_missing_dept_code_fails_loudly(self): payload = {k: v for k, v in PAYLOAD.items() if k != "custDeptNumber"} runner, calls = _user_runner([]) out = await runner.upsert( person_email="mingxin.wu@pingcap.com", payload=payload, prior_external_id=None, ) assert out.status == "blocked" assert "custDeptNumber" in (out.error or "") assert calls.calls == [] async def test_missing_employee_id_in_payload_fails_loudly(self): runner, calls = _user_runner([]) out = await runner.upsert( person_email="x@pingcap.com", payload={}, prior_external_id=None, ) assert out.status == "failed" assert "employeeID" in (out.error or "") assert calls.calls == [] async def test_put_120207_falls_back_to_create(self): runner, calls = _user_runner([ ("GET", "/api/open/user/detail/P01431", (200, {"employeeID": "P01431", "status": 1001}, "")), ("PUT", "/api/open/user/v2", (200, USER_MISSING, "")), ("POST", "/api/open/user/create/v2", (200, OK, "")), ]) out = await runner.upsert( person_email="mingxin.wu@pingcap.com", payload=dict(PAYLOAD), prior_external_id="P01431", ) assert out.status == "success" assert [c[0] for c in calls.calls] == ["GET", "PUT", "POST"] async def test_create_duplicate_message_falls_back_to_update(self): runner, calls = _user_runner([ ("GET", "/api/open/user/detail/P01431", (404, USER_MISSING, "")), ("POST", "/api/open/user/create/v2", (200, {"errorCode": "120299", "message": "员工工号已存在"}, "")), ("PUT", "/api/open/user/v2", (200, OK, "")), ]) out = await runner.upsert( person_email="mingxin.wu@pingcap.com", payload=dict(PAYLOAD), prior_external_id=None, ) assert out.status == "success" assert [c[0] for c in calls.calls] == ["GET", "POST", "PUT"] class TestUserDeactivate: async def test_deactivate_success(self): runner, calls = _user_runner([ ("DELETE", "/api/open/user/delete/v2/P01431/", (200, OK, "")), ]) out = await runner.deactivate( person_email="mingxin.wu@pingcap.com", prior_external_id="P01431", ) assert out.status == "deactivated" assert out.external_id == "P01431" async def test_deactivate_idempotent_on_user_missing(self): runner, _ = _user_runner([ ("DELETE", "/api/open/user/delete/v2/P01431/", (200, USER_MISSING, "")), ]) out = await runner.deactivate( person_email="mingxin.wu@pingcap.com", prior_external_id="P01431", ) assert out.status == "deactivated" async def test_deactivate_idempotent_on_already_resigned_message(self): # HR keeps resigning leavers by hand through the cutover # window — 离职V2 against an already-resigned row must be # success (goal achieved), not a forever-retried failure. runner, _ = _user_runner([ ("DELETE", "/api/open/user/delete/v2/P01431/", (200, {"errorCode": "120298", "message": "员工已离职"}, "")), ]) out = await runner.deactivate( person_email="mingxin.wu@pingcap.com", prior_external_id="P01431", ) assert out.status == "deactivated" assert out.external_id == "P01431" async def test_deactivate_without_external_id_fails_loudly(self): runner, calls = _user_runner([]) out = await runner.deactivate( person_email="mingxin.wu@pingcap.com", prior_external_id=None, ) assert out.status == "failed" assert calls.calls == [] async def test_supports_delete_is_false(self): # 汇联易 has no hard-delete. The orchestrator short-circuits an # unsupported policy to status='excluded' (NO vendor call), so # _validate_on_offboard refuses on_offboard=delete for this # kind at save time — see TestOnOffboardValidation below. assert HuilianyiRunner.supports_delete is False class TestDeptUpsert: async def test_post_create_new(self): runner, calls = _dept_runner([ ("POST", "/api/open/department/v2", (200, {**OK, "oid": "oid-123", "key": "DEP-FIN"}, "")), ]) out = await runner.upsert_dept( dept_code="DEP-FIN", dept_name="Finance", parent_dept_code=None, leader_email=None, dept_order=None, prior_external_id=None, ) assert out.status == "success" assert out.external_id == "oid-123" assert calls.calls[0][2]["custDeptNumber"] == "DEP-FIN" assert "parentCustDeptNumber" not in calls.calls[0][2] async def test_post_code_exists_falls_back_to_put(self): runner, calls = _dept_runner([ ("POST", "/api/open/department/v2", (200, DEPT_CODE_EXISTS, "")), ("PUT", "/api/open/department/v2", (200, OK, "")), ]) out = await runner.upsert_dept( dept_code="DEP-FIN", dept_name="Finance", parent_dept_code="DEP-ROOT", leader_email=None, dept_order=None, prior_external_id=None, ) assert out.status == "success" assert [c[0] for c in calls.calls] == ["POST", "PUT"] assert calls.calls[1][2]["parentCustDeptNumber"] == "DEP-ROOT" async def test_prior_external_id_puts_first_then_recreates_on_120110(self): runner, calls = _dept_runner([ ("PUT", "/api/open/department/v2", (200, DEPT_MISSING, "")), ("POST", "/api/open/department/v2", (200, OK, "")), ]) out = await runner.upsert_dept( dept_code="DEP-FIN", dept_name="Finance", parent_dept_code=None, leader_email=None, dept_order=None, prior_external_id="oid-123", ) assert out.status == "success" assert [c[0] for c in calls.calls] == ["PUT", "POST"] async def test_roles_and_cost_center_land_in_body(self): """部门负责人/HRBP/Budget Owner → departmentPositionList (positionCode 6103/6105/6102) and cost center → customFormValues[fieldCode=CC].""" runner, calls = _dept_runner([ ("POST", "/api/open/department/v2", (200, {**OK, "oid": "o1"}, "")), ]) out = await runner.upsert_dept( dept_code="2730378", dept_name="Business Operations", parent_dept_code="RootOrg", leader_email=None, dept_order=None, prior_external_id=None, leader_employee_no="P0843", cost_center_label="C0906 BO-Business Operation", hrbp_employee_no="P01115", budget_owner_employee_no="P0843", ) assert out.status == "success" body = calls.calls[0][2] assert "managerEmployeeId" not in body # leader goes via the 6103 role assert body["departmentPositionList"] == [ {"positionCode": "6103", "employeeId": "P0843"}, {"positionCode": "6105", "employeeId": "P01115"}, {"positionCode": "6102", "employeeId": "P0843"}, ] assert body["customFormValues"] == [ {"fieldCode": "CC", "value": "C0906 BO-Business Operation"}, ] async def test_roles_and_cost_center_omitted_when_absent(self): """No roles / no mapped cost center → fields omitted entirely so the vendor leaves any existing values unchanged (never blanks).""" runner, calls = _dept_runner([ ("POST", "/api/open/department/v2", (200, {**OK, "oid": "o1"}, "")), ]) out = await runner.upsert_dept( dept_code="2730378", dept_name="Business Operations", parent_dept_code="RootOrg", leader_email=None, dept_order=None, prior_external_id=None, leader_employee_no=None, cost_center_label=None, hrbp_employee_no=None, budget_owner_employee_no=None, ) assert out.status == "success" body = calls.calls[0][2] assert "departmentPositionList" not in body assert "managerEmployeeId" not in body assert "customFormValues" not in body async def test_partial_roles_only_includes_resolved(self): """Only roles with a resolved 工号 land; the rest are omitted.""" runner, calls = _dept_runner([ ("POST", "/api/open/department/v2", (200, {**OK, "oid": "o1"}, "")), ]) out = await runner.upsert_dept( dept_code="3963958", dept_name="IT Services & Security", parent_dept_code="2730378", leader_email=None, dept_order=None, prior_external_id=None, leader_employee_no="P0989", hrbp_employee_no="P01115", budget_owner_employee_no=None, cost_center_label=None, ) assert out.status == "success" body = calls.calls[0][2] assert body["departmentPositionList"] == [ {"positionCode": "6103", "employeeId": "P0989"}, {"positionCode": "6105", "employeeId": "P01115"}, ] assert "customFormValues" not in body class TestDeptSoftDelete: async def test_disable_success(self): runner, _ = _dept_runner([ ("DELETE", "/api/open/department/delete/v2/DEP-OLD", (200, OK, "")), ]) out = await runner.soft_delete( dept_code="DEP-OLD", prior_external_id="oid-9", ) assert out.status == "soft_deleted" async def test_disable_idempotent_on_missing(self): runner, _ = _dept_runner([ ("DELETE", "/api/open/department/delete/v2/DEP-OLD", (200, DEPT_MISSING, "")), ]) out = await runner.soft_delete( dept_code="DEP-OLD", prior_external_id="oid-9", ) assert out.status == "soft_deleted" async def test_disable_blocked_by_children_fails_with_hint(self): runner, _ = _dept_runner([ ("DELETE", "/api/open/department/delete/v2/DEP-OLD", (200, DEPT_BUSY, "")), ]) out = await runner.soft_delete( dept_code="DEP-OLD", prior_external_id="oid-9", ) assert out.status == "skipped" assert "子部门" in (out.error or "") async def test_disable_blocked_by_enabled_child_120125_gets_hint(self): # Regression: 120125 (部门存在启用的子部门) was unmapped and fell # through to the generic "errorCode=120125" passthrough. It must be # treated like 120102 — a surfaced, actionable human task, never a # bare vendor code. (Prod: custDeptNumber 20250304, 2026-07-05.) runner, _ = _dept_runner([ ("DELETE", "/api/open/department/delete/v2/DEP-OLD", (200, DEPT_BUSY_ENABLED_CHILD, "")), ]) out = await runner.soft_delete( dept_code="DEP-OLD", prior_external_id="oid-9", ) assert out.status == "skipped" assert "子部门" in (out.error or "") # Message quality: the hint must point the human at the vendor # console, and must never leak the bare vendor code as the whole # message (the pre-fix passthrough was exactly # "errorCode=120125"). assert "汇联易后台" in (out.error or "") assert "errorCode=120125" not in (out.error or "") async def test_disable_blocked_by_employees_120124_gets_hint(self): # Regression: 120124 (部门或者子部门存在员工,无法禁用) was the third # dependents-not-cleared code and, like 120125 before it, was # UNMAPPED — 失效 dumped a bare "errorCode=120124" to the generic # failed branch and re-failed every nightly sweep. It must join # 120102 / 120125 as a surfaced, actionable human task (never a # cascade, never a bare code). Codes captured from the #1954 ledger, # NOT guessed: 120124 x9 across 2026-06/07 (custDeptNumbers 2-1-19, # 20240501-1, 2-1-2-1, …); the same read confirmed no already-disabled # idempotency code exists (disabled depts aren't re-swept), so none # was invented here. runner, _ = _dept_runner([ ("DELETE", "/api/open/department/delete/v2/DEP-OLD", (200, DEPT_BUSY_HAS_EMPLOYEES, "")), ]) out = await runner.soft_delete( dept_code="DEP-OLD", prior_external_id="oid-9", ) assert out.status == "skipped" # The friendly hint already covers both 子部门 and 在职员工. assert "员工" in (out.error or "") assert "errorCode" not in (out.error or "") assert "汇联易后台" in (out.error or "") # never leak the bare vendor code as the whole message assert "errorCode=120125" not in (out.error or "") class TestClientToken: async def test_token_missing_in_response_raises(self, monkeypatch): client = _client() class _Resp: status_code = 200 text = '{"weird": true}' def json(self): return {"weird": True} async def fake_post(url, data=None, headers=None): return _Resp() monkeypatch.setattr(client._client, "post", fake_post) with pytest.raises(HuilianyiError): await client.get_access_token() await client.aclose() async def test_token_parsed_from_wrapped_data(self, monkeypatch): client = _client() class _Resp: status_code = 200 text = "..." def json(self): return {"data": {"access_token": "tok-1", "expires_in": 7200}} async def fake_post(url, data=None, headers=None): return _Resp() monkeypatch.setattr(client._client, "post", fake_post) assert await client.get_access_token() == "tok-1" # Cached: second call doesn't hit the wire (post would raise if # called again with a poisoned stub). async def boom(url, data=None, headers=None): # noqa: ANN001 raise AssertionError("token refetched despite valid cache") monkeypatch.setattr(client._client, "post", boom) assert await client.get_access_token() == "tok-1" await client.aclose() class TestClientBreakerAndConfig: async def test_token_auth_style_normalised(self): # Free-text config field: "Basic " must not hard-fail a sweep. client = HuilianyiClient( base_url="https://api.huilianyi.example", client_id="i", client_secret="s", token_auth_style=" Basic ", token_path=" /oauth/token ", ) assert client._token_auth_style == "basic" assert client._token_path == "/oauth/token" await client.aclose() async def test_circuit_breaker_opens_after_consecutive_failures(self, monkeypatch): client = _client() class _Resp: status_code = 500 text = "gateway dead" def json(self): raise ValueError async def dead_post(url, data=None, headers=None): return _Resp() monkeypatch.setattr(client._client, "post", dead_post) for _ in range(client._breaker_threshold): with pytest.raises(HuilianyiError): await client.get_access_token() # Breaker open: fails instantly without touching the wire. async def boom(url, data=None, headers=None): # noqa: ANN001 raise AssertionError("wire touched after breaker opened") monkeypatch.setattr(client._client, "post", boom) with pytest.raises(HuilianyiError, match="circuit open"): await client.get_access_token() with pytest.raises(HuilianyiError, match="circuit open"): await client.request("GET", "/api/open/user/detail/P0001") await client.aclose() class TestOnOffboardValidation: async def test_delete_rejected_for_kinds_without_hard_delete(self): from app.core.exceptions import AppException from app.domains.identity_sync.service import _validate_on_offboard for kind in ("huilianyi", "feishu_contact"): with pytest.raises(AppException) as ei: _validate_on_offboard(kind, "delete") assert ei.value.code == "IDENTITY_SYNC_ON_OFFBOARD_UNSUPPORTED" # SCIM has a real DELETE — allowed. _validate_on_offboard("scim", "delete") # deactivate/keep always fine. _validate_on_offboard("huilianyi", "deactivate") _validate_on_offboard("huilianyi", "keep") with pytest.raises(AppException): _validate_on_offboard("huilianyi", "nuke") class TestPresetAndRegistry: async def test_kind_registered_everywhere(self): from app.domains.identity_sync.presets import get_preset from app.domains.identity_sync.runners import RUNNER_REGISTRY from app.domains.identity_sync.service import SUPPORTED_KINDS assert "huilianyi" in SUPPORTED_KINDS assert RUNNER_REGISTRY["huilianyi"] is HuilianyiRunner preset = get_preset("huilianyi") assert preset is not None assert preset.kind == "huilianyi" assert preset.requires_integration_kind == "identity_sync_huilianyi" targets = { r["target_field"] for r in preset.to_seed()["mapping_config"]["rules"] } # Vendor-required create fields are all mapped. assert {"employeeID", "fullName", "email", "custDeptNumber", "companyCode"} <= targets # Same service-account gate as feishu_contact. assert preset.filter_config["employee_no_pattern"] == r"^P\d+$" async def test_driver_registered(self): import app.integrations.drivers # noqa: F401 — fire registrations from app.integrations.registry import list_drivers kinds = {m.kind for m in list_drivers()} assert "identity_sync_huilianyi" in kinds