"""对象目录与样例适配守卫（任务书 P1-A / P1-F / P1-H）。""" from __future__ import annotations import re import pytest from catalog import sample_adapters as sa # 方案 §7.2 列出的 26 个 API_INBOUND 代码契约 EXPECTED_INBOUND_CONTRACTS = { "MDM_ITEM", "MDM_CUSTOMER", "MDM_SUPPLIER", "MDM_LOCATION", "MDM_SOURCE_LIST", "MDM_EMPLOYEE_HEADCOUNT", "S1_SALES_ORDER_ENTRY", "S1_REQUIREMENT_EXAMINE_RESULT", "S1_REQUIREMENT_EXAMINE_DETAIL", "S2_WORK_ORDER_SCHEDULE", "S3_PURCHASE_ORDER", "S3_PURCHASE_RECEIPT", "S4_SHIPMENT", "S4_IQC", "S4_RETURN", "S4_SHORTAGE", "S5_WORK_ORDER_BOM", "S5_INVENTORY_TXN", "S5_INVENTORY_OPENING_BALANCE", "S5_INVENTORY_BALANCE_MONTHLY", "S6_WORK_ORDER_LINE", "S6_REPORT_TXN", "S7_FQC_TASK_TXN", "S7_SALES_ORDER_LINE", "S7_FINISHED_ONHAND", "S7_FINISHED_OPENING_BALANCE", } RUN_ID = "20260930120000-abc123" def test_catalog_loads_and_has_three_channels(): catalog = sa.load_catalog() assert catalog["objects"], "catalog must not be empty" for obj in catalog["objects"]: for channel in ("dbSync", "apiPull", "apiInbound"): assert channel in obj, f"{obj['objectCode']} missing {channel}" assert "supported" in obj[channel] # 不支持必须写明原因，不得把“配置未开”写成“代码不支持” for channel in ("dbSync", "apiPull", "apiInbound"): if not obj[channel]["supported"]: assert obj[channel].get("reason"), f"{obj['objectCode']}.{channel} unsupported without reason" def test_catalog_covers_all_26_inbound_contracts(): registered = {(o.get("apiInbound") or {}).get("entityCode") for o in sa.load_catalog()["objects"] if (o.get("apiInbound") or {}).get("supported")} assert registered == EXPECTED_INBOUND_CONTRACTS, ( f"missing={sorted(EXPECTED_INBOUND_CONTRACTS - registered)}, " f"unexpected={sorted(registered - EXPECTED_INBOUND_CONTRACTS)}") def test_every_inbound_contract_has_field_spec(): for code in EXPECTED_INBOUND_CONTRACTS: assert code in sa.INBOUND_SPECS, f"no field spec for {code}" assert any(required for _, required, _ in sa.INBOUND_SPECS[code]), f"{code} has no required field" def test_contract_version_matches_field_spec_semantics(): """INBOUND_SPECS 按 IdentityFields（v2 中立集合）造数，目录必须统一登记 v2。 C# 端 NormalizeVersion 只将显式 v2 路由到 IdentityFields；若目录写 v1 而适配表按 v2 生成，就会出现“按 v1 造数、按 v2 校验”的口径错配。 """ for obj in sa.load_catalog()["objects"]: inbound = obj.get("apiInbound") or {} if inbound.get("supported"): assert inbound.get("contractVersion") == "v2", \ f"{obj['objectCode']} contractVersion 应为 v2，实为 {inbound.get('contractVersion')}" def test_builtin_sample_rows_satisfy_required_fields(): """内置最小样例必须能填满契约必填项（否则预检会把造数缺陷当成配置问题）。""" run_id = "20260930120000-abc123" for obj in sa.load_catalog()["objects"]: if obj.get("sample") or not (obj.get("apiInbound") or {}).get("supported"): continue adapted = sa.adapt_inbound_rows(obj, run_id) assert not adapted["missing_required"], \ f"{obj['objectCode']} builtin sample misses {adapted['missing_required']}" def test_sample_paths_exist(): for obj in sa.load_catalog()["objects"]: if obj.get("sample"): assert (sa.REPO_ROOT / obj["sample"]).exists(), f"sample missing: {obj['sample']}" else: assert obj["objectCode"] in sa.BUILTIN_SAMPLES, \ f"{obj['objectCode']} has neither sample file nor builtin rows" def test_run_id_format(): assert re.match(r"^\d{14}-[0-9a-f]{6}$", sa.new_run_id()) def test_db_rows_carry_sim_prefix_and_increment_column(): obj = sa.find_object("ITEM") rows = sa.adapt_db_rows(obj, RUN_ID) assert rows for row in rows: assert str(row["bizKey"]).startswith(f"SIM-{RUN_ID}-") assert row.get("sourceUpdatedAt"), "increment column must be filled" assert row.get("Domain") == "SIM" def test_prefix_is_idempotent(): obj = sa.find_object("ITEM") once = sa.adapt_db_rows(obj, RUN_ID) twice = sa.adapt_db_rows(obj, RUN_ID, once) assert [r["bizKey"] for r in once] == [r["bizKey"] for r in twice] def test_api_rows_sorted_by_bizkey_for_cursor(): obj = sa.find_object("SALES_ORDER_ENTRY") rows = sa.adapt_api_rows(obj, RUN_ID) keys = [str(r["bizKey"]) for r in rows] assert keys == sorted(keys) @pytest.mark.parametrize("object_code,entity_code", [ ("ITEM", "MDM_ITEM"), ("SALES_ORDER_ENTRY", "S1_SALES_ORDER_ENTRY"), ("WORK_ORDER", "S2_WORK_ORDER_SCHEDULE"), ("PURCHASE_ORDER", "S3_PURCHASE_ORDER"), ("IQC", "S4_IQC"), ("INVENTORY_TXN", "S5_INVENTORY_TXN"), ("REPORT_TXN", "S6_REPORT_TXN"), ("FQC_TASK", "S7_FQC_TASK_TXN"), ]) def test_inbound_rows_use_contract_field_names(object_code, entity_code): obj = sa.find_object(object_code) assert (obj["apiInbound"] or {}).get("entityCode") == entity_code adapted = sa.adapt_inbound_rows(obj, RUN_ID) spec_fields = {f for f, _, _ in sa.INBOUND_SPECS[entity_code]} for row in adapted["rows"]: unknown = set(row) - spec_fields - {"sourceUpdatedAt", "sourceVersion", "op"} assert not unknown, f"{entity_code} produced non-contract fields: {sorted(unknown)}" assert str(row.get("sourceUpdatedAt")), "sourceUpdatedAt is required by common contract" def test_inbound_reports_missing_required_instead_of_faking(): obj = sa.find_object("SALES_ORDER_ENTRY") adapted = sa.adapt_inbound_rows(obj, RUN_ID, rows=[{"bizKey": "X-1", "qty": 5}]) # bill_no / entry_seq / seorder_id 无别名来源，必须报告缺失而不是编造 assert {"bill_no", "entry_seq", "seorder_id"} <= set(adapted["missing_required"]) def test_inbound_rejects_object_without_contract(): obj = sa.find_object("SALES_ORDER_HEAD") assert not (obj["apiInbound"] or {}).get("supported") with pytest.raises(ValueError): sa.adapt_inbound_rows(obj, RUN_ID) def test_envelope_shape(): env = sa.inbound_envelope([{"a": 1}], snapshot_id="SNAP-1", seq=2) assert env == {"data": {"list": [{"a": 1}], "snapshotId": "SNAP-1", "seq": 2}} def test_scenarios_reference_known_objects_and_mark_unsupported(): scenarios = sa.load_scenarios()["scenarios"] assert len(scenarios) == 3 for sc in scenarios: assert sc["nodes"] for node in sc["nodes"]: obj = sa.find_object(node["objectCode"]) assert obj, f"scenario {sc['scenarioId']} references unknown object {node['objectCode']}" spec = {"DB_SYNC": obj["dbSync"], "API_PULL": obj["apiPull"], "API_INBOUND": obj["apiInbound"]}[node["channel"]] if not spec.get("supported"): assert node.get("inboundNote") or spec.get("reason"), \ f"{node['objectCode']} unsupported channel without note" def test_samples_contain_no_real_looking_secrets(): """样例不得含真实凭据形态字段。""" banned = re.compile(r"(password|secret|access_?key|token|jwt)", re.IGNORECASE) for obj in sa.load_catalog()["objects"]: for row in sa.load_canonical_rows(obj): for key in row: assert not banned.search(str(key)), f"{obj['objectCode']} sample has credential-like field {key}"