AiDOPWarehouse/tools/integration-simulator/tests/test_sample_adapters.py

"""对象目录与样例适配守卫（任务书 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}"