"""汇联易 external-actuals coordinator — mapping / classification / FX / dedup / graceful degradation. The live 汇联易 expense API can't be exercised from CI, so these pin the LOGIC the coordinator owns (document→statement mapping, the no-double-count classification switch, USD conversion at the month-frozen rate, idempotent keys, and graceful skip when no connector). The exact vendor field names are validated on the first dry-run against the real gateway (documented in docs/ops). """ from __future__ import annotations from decimal import Decimal import pytest import pytest_asyncio from app.core.coordinators import finance_huilianyi_coordinator as hly from app.domains.finance.fx import upsert_monthly_rate PERIOD = "2026-05" CNY_RATE = Decimal("0.10") @pytest_asyncio.fixture async def fx(db_session): # Pin the month rate so resolve_usd_per_unit hits the frozen row (no network). await upsert_monthly_rate(db_session, "CNY", PERIOD, CNY_RATE, source="manual", frozen=True) await db_session.flush() return db_session def _doc(): return { "documentCode": "BX-001", "accountingDate": "2026-05-20", "documentStatus": "PAID", "paymentStatus": "PAID", "employeeOID": "emp-9", "currency": "CNY", "apportionments": [ # Non-IT (差旅费) → promoted into the ledger as cost_type='expense'. {"expenseType": "差旅费", "amount": "1000", "costCenterCode": "CC-RD", "departmentName": "研发", "merchantName": "携程"}, # IT-overlap (云服务费, NOT on the promote whitelist) → # reconciliation-only, NOT promoted (avoids double-count vs Helix cloud). {"expenseType": "云服务费", "amount": "500", "costCenterCode": "CC-RD", "departmentName": "研发", "merchantName": "AWS", "invoiceNo": "INV-9"}, ], } @pytest.mark.asyncio async def test_map_apportions_classifies_and_converts(fx): rows = await hly.map_documents_to_statements(fx, [_doc()]) assert len(rows) == 2 by_cat = {r.expense_category: r for r in rows} travel = by_cat["差旅费"] assert travel.period == PERIOD assert travel.cost_center == "CC-RD" assert travel.department == "研发" assert travel.amount_native == Decimal("1000.000000") assert travel.amount_usd == Decimal("100.000000") # 1000 × 0.10 assert travel.fx_basis == "month_frozen" assert travel.promote_to_ledger is True assert travel.mapped_cost_type == "expense" assert travel.line_no == 0 cloud = by_cat["云服务费"] assert cloud.amount_usd == Decimal("50.000000") assert cloud.invoice_no == "INV-9" # IT-overlap stays external-only — the no-double-count guard. assert cloud.promote_to_ledger is False assert cloud.mapped_cost_type is None assert cloud.line_no == 1 @pytest.mark.asyncio async def test_owner_oid_not_stored_as_person_id(fx): # 汇联易 employeeOID is a vendor id, not a Helix person id → must NOT # be stored as owner_person_id (would mislabel attribution). rows = await hly.map_documents_to_statements(fx, [_doc()]) assert all(r.owner_person_id is None for r in rows) @pytest.mark.asyncio async def test_zero_line_does_not_grab_doc_total(fx): # A ¥0 apportionment line must be skipped, NOT fall through to the # document total (the old `_pick(line) or _pick(doc)` ghost-row bug). doc = { "documentCode": "BX-Z", "accountingDate": "2026-05-10", "currency": "CNY", "totalAmount": "9999", "apportionments": [ {"expenseType": "差旅费", "amount": "300", "costCenterCode": "CC-A"}, {"expenseType": "餐饮费", "amount": "0", "costCenterCode": "CC-A"}, ], } rows = await hly.map_documents_to_statements(fx, [doc]) # Only the ¥300 line survives; the ¥0 line is dropped (NOT ¥9999). assert len(rows) == 1 assert rows[0].amount_native == Decimal("300.000000") assert all(r.amount_native != Decimal("9999.000000") for r in rows) @pytest.mark.asyncio async def test_statement_keys_unique_and_stable(fx): r1 = await hly.map_documents_to_statements(fx, [_doc()]) r2 = await hly.map_documents_to_statements(fx, [_doc()]) keys1 = {r.statement_key for r in r1} assert len(keys1) == 2 # two lines → two distinct keys # Stable across runs (idempotent upsert key). assert keys1 == {r.statement_key for r in r2} @pytest.mark.asyncio async def test_promote_whitelist_is_overridable(fx): from app.integrations.models import PlatformSetting # Override so ONLY 餐饮费 promotes — 差旅费 should now stay external-only. fx.add(PlatformSetting(key=hly.PROMOTE_CATEGORIES_SETTING, value=["餐饮费"])) await fx.flush() rows = await hly.map_documents_to_statements(fx, [_doc()]) travel = next(r for r in rows if r.expense_category == "差旅费") assert travel.promote_to_ledger is False # no longer whitelisted @pytest.mark.asyncio async def test_pull_actuals_no_connector_skips_gracefully(fx): # No identity_sync_huilianyi integration row exists → soft skip, no raise. result = await hly.pull_actuals(fx, PERIOD) assert result["skipped"] == "no_connector" assert result["rows"] == 0 @pytest.mark.asyncio async def test_sync_cost_centers_collapses_in_batch_duplicates(fx, monkeypatch): """汇联易 returning the same code twice in one pull must NOT trip the (source, code) unique constraint — regression for the actuals-sync cron failing with IntegrityError on uq_finance_cost_center_src_code.""" from sqlalchemy import select from app.domains.finance.models import FinanceCostCenter async def _fake_build_client(_db): return object(), {"cost_center": "/api/open/costCenter"} async def _fake_fetch(_client, _path): # Same code twice (+ a distinct one), as the live gateway can return. return [ {"code": "CC-RD", "name": "研发"}, {"code": "CC-RD", "name": "研发(改名)"}, {"code": "CC-OPS", "name": "运维"}, ] monkeypatch.setattr(hly, "_build_client", _fake_build_client) monkeypatch.setattr(hly, "_fetch_cost_centers", _fake_fetch) monkeypatch.setattr(hly, "_safe_close", lambda _c: _noop()) result = await hly.sync_cost_centers(fx) await fx.flush() assert result["rows"] == 3 # parsed count, duplicates included rows = (await fx.execute( select(FinanceCostCenter).where(FinanceCostCenter.source == hly.SOURCE) )).scalars().all() by_code = {r.code: r for r in rows} assert set(by_code) == {"CC-RD", "CC-OPS"} # one row per code, no collision assert by_code["CC-RD"].name == "研发(改名)" # last occurrence wins async def _noop(): return None def test_as_list_tolerates_array_and_wrapped_shapes(): assert hly._as_list([{"a": 1}]) == [{"a": 1}] assert hly._as_list({"data": [{"a": 1}]}) == [{"a": 1}] assert hly._as_list({"data": {"records": [{"a": 1}]}}) == [{"a": 1}] assert hly._as_list({"unexpected": 1}) == [] assert hly._as_list(None) == [] def test_doc_period_parses_date_shapes(): assert hly._doc_period({"accountingDate": "2026-05-20"}) == "2026-05" assert hly._doc_period({"submitDate": "2026/05/20"}) == "2026-05" assert hly._doc_period({"bookingDate": "20260520"}) == "2026-05" assert hly._doc_period({}) is None # ── pagination: the expense/cost-center pulls MUST page through every result ── # # Regression: ``_fetch_expense_documents`` / ``_fetch_cost_centers`` issued a # single ``page=1, size=200/500`` request and returned that one page. For a # ~500-person month the expense docs exceed one page, so every doc past the # cap was silently dropped → the reconciliation ledger understated actuals # with no error surfaced. class _FakeHlyClient: """Serves a fixed doc list in page/size windows, like the 汇联易 gateway.""" def __init__(self, docs, *, ignore_page: bool = False): self.docs = docs self.ignore_page = ignore_page self.calls: list[tuple[int, int]] = [] async def request(self, method, path, params=None): params = params or {} page = int(params.get("page", 1)) size = int(params.get("size", 200)) self.calls.append((page, size)) effective_page = 1 if self.ignore_page else page start = (effective_page - 1) * size return 200, self.docs[start:start + size], "" @pytest.mark.asyncio async def test_fetch_expense_documents_pages_through_all(): docs = [{"documentCode": f"BX-{i:04d}", "currency": "CNY"} for i in range(450)] client = _FakeHlyClient(docs) got = await hly._fetch_expense_documents(client, "/expense", PERIOD) # Every doc returned, none dropped, none duplicated. assert len(got) == 450 assert {d["documentCode"] for d in got} == {d["documentCode"] for d in docs} # Paged 200 + 200 + 50(short → stop). assert client.calls == [(1, 200), (2, 200), (3, 200)] @pytest.mark.asyncio async def test_fetch_cost_centers_pages_through_all(): items = [{"code": f"CC-{i:04d}", "name": f"n{i}"} for i in range(1200)] client = _FakeHlyClient(items) got = await hly._fetch_cost_centers(client, "/cc") assert len(got) == 1200 assert client.calls == [(1, 500), (2, 500), (3, 500)] # 500+500+200(short) @pytest.mark.asyncio async def test_fetch_all_pages_runaway_raises(monkeypatch): """A server that keeps returning full, fresh pages past the runaway guard must RAISE (partial actuals understate the ledger) — never truncate.""" from app.domains.identity_sync.runners.huilianyi import HuilianyiError monkeypatch.setattr(hly, "_MAX_PAGES", 3) docs = [{"documentCode": f"BX-{i}", "currency": "CNY"} for i in range(800)] client = _FakeHlyClient(docs) with pytest.raises(HuilianyiError): await hly._fetch_expense_documents(client, "/expense", PERIOD) @pytest.mark.asyncio async def test_fetch_all_pages_stops_when_server_ignores_page(): """A gateway that ignores the ``page`` param (returns the same full page every time) must stop after a full page yields nothing new — not spin to the runaway guard and raise.""" docs = [{"documentCode": f"BX-{i}", "currency": "CNY"} for i in range(200)] client = _FakeHlyClient(docs, ignore_page=True) got = await hly._fetch_expense_documents(client, "/expense", PERIOD) assert len(got) == 200 # page1 full → page2 same 200 (all seen, fresh==0) → stop. assert client.calls == [(1, 200), (2, 200)]