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

"""第二阶段（N1-N4）异常测试能力守卫。

- N2：Mock API 异常注入模式（超时/空响应/非法 JSON/Content-Type/cursor 行为/恢复默认）
- N1：模拟库异常能力（drop 表/无唯一键重复装载/增量列不推进/慢查询视图 DDL）
- N3：入站客户端异常用例（缺/错 AccessKey、缺字段、类型错误、stale、snapshot 失败）
- N4：异常测试报告（字段覆盖任务书 §15 八列、脱敏、导出）

全部进程内验证（TestClient/纯函数），不依赖 Ai-DOP 在线与真实 MySQL。
"""
from __future__ import annotations

import json

import pytest

fastapi_testclient = pytest.importorskip("fastapi.testclient")
TestClient = fastapi_testclient.TestClient

from providers import mock_api  # noqa: E402
from providers import mock_db  # noqa: E402
from catalog import sample_adapters as sa  # noqa: E402


# ---------------------------------------------------------------------------
# N2：Mock API 异常模式
# ---------------------------------------------------------------------------

@pytest.fixture()
def client():
    state = mock_api.MockApiState()
    state.settings.auth_mode = "NONE"
    return TestClient(mock_api.create_app(state)), state


def _set(client, **kw):
    c, state = client
    for key, val in kw.items():
        setattr(state.settings, key, val)


def test_force_status_covers_n2_codes(client):
    """N2 要求 401/403/404/429/500/503：force_status 须对全部六类生效。"""
    c, _ = client
    for code in (401, 403, 404, 429, 500, 503):
        _set(client, force_status=code)
        assert c.get("/api/item").status_code == code, code
    _set(client, force_status=None)
    assert c.get("/api/item").status_code == 200


def test_empty_response_mode(client):
    c, _ = client
    _set(client, empty_response=True)
    resp = c.get("/api/item")
    assert resp.status_code == 200
    assert resp.content == b""


def test_invalid_json_mode(client):
    c, _ = client
    _set(client, invalid_json=True)
    resp = c.get("/api/item")
    assert resp.status_code == 200
    with pytest.raises(Exception):
        resp.json()  # 无法解析为 JSON


def test_wrong_content_type_mode(client):
    c, _ = client
    _set(client, content_type="text/html")
    resp = c.get("/api/item")
    assert resp.status_code == 200
    assert resp.headers["content-type"].startswith("text/html")


def test_timeout_mode_sleeps_configured_seconds(client):
    c, _ = client
    _set(client, timeout_seconds=0.3)
    import time
    started = time.time()
    resp = c.get("/api/item")
    assert resp.status_code == 200
    assert time.time() - started >= 0.3


def test_ignore_cursor_returns_all_rows(client):
    c, _ = client
    rows = [
        {"bizKey": "K-1", "v": 1}, {"bizKey": "K-2", "v": 2}, {"bizKey": "K-3", "v": 3},
    ]
    c.post("/__control/overrides", json={"path": "/api/order-test", "rows": rows})
    # 默认 cursor 过滤：只返回更大行
    after = c.get("/api/order-test", params={"cursor": "K-1"}).json()["data"]["list"]
    assert [r["bizKey"] for r in after] == ["K-2", "K-3"]
    # ignore_cursor：无视 cursor 返回全部行（cursor 不推进观察用）
    _set(client, ignore_cursor=True)
    after = c.get("/api/order-test", params={"cursor": "K-1"}).json()["data"]["list"]
    assert [r["bizKey"] for r in after] == ["K-1", "K-2", "K-3"]


def test_no_sort_keeps_original_order(client):
    c, state = client
    rows = [
        {"bizKey": "Z-2"}, {"bizKey": "A-1"}, {"bizKey": "M-3"},
    ]
    c.post("/__control/overrides", json={"path": "/api/order-test", "rows": rows})
    # 默认排序
    got = [r["bizKey"] for r in c.get("/api/order-test").json()["data"]["list"]]
    assert got == ["A-1", "M-3", "Z-2"]
    # no_sort 保持挂载原顺序（cursor 乱序观察用）
    _set(client, no_sort=True)
    got = [r["bizKey"] for r in c.get("/api/order-test").json()["data"]["list"]]
    assert got == ["Z-2", "A-1", "M-3"]


def test_reset_faults_clears_all_injections(client):
    c, _ = client
    _set(client, force_status=500, empty_response=True, invalid_json=True,
         content_type="text/html", ignore_cursor=True, no_sort=True,
         timeout_seconds=1.0, delay_seconds=0.5)
    resp = c.post("/__control/settings", json={"reset": True}).json()
    assert resp["forceStatus"] is None
    assert resp["emptyResponse"] is False
    assert resp["invalidJson"] is False
    assert resp["contentType"] is None
    assert resp["ignoreCursor"] is False
    assert resp["noSort"] is False
    assert resp["timeoutSeconds"] == 0.0
    assert resp["delaySeconds"] == 0.0
    assert resp["authMode"] == "TOKEN"  # 鉴权模式一并恢复默认
    # 恢复后正常拉取可用
    _set(client, auth_mode="NONE")
    assert c.get("/api/item").status_code == 200


def test_audit_records_active_faults(client):
    c, _ = client
    _set(client, empty_response=True, ignore_cursor=True)
    c.get("/api/item")
    entries = c.get("/__control/audit").json()["entries"]
    assert entries, "audit should record the request"
    faults = entries[-1]["faults"]
    assert "emptyResponse" in faults
    assert "ignoreCursor" in faults


# ---------------------------------------------------------------------------
# N1：模拟库异常能力（纯函数层；真实 MySQL 用例沿用 test_mock_db.py 的跳过约定）
# ---------------------------------------------------------------------------

RUN_ID = "20260930120000-abc123"


def test_ddl_without_unique_key_for_duplicate_bizkey():
    rows = sa.adapt_db_rows(sa.find_object("ITEM"), RUN_ID)
    ddl = mock_db.generate_ddl("sim_item", rows, ["bizKey"], "sourceUpdatedAt",
                               unique_key=False)
    assert "UNIQUE KEY" not in ddl  # 重复业务键场景：无唯一索引，允许重复行共存
    assert "CREATE TABLE IF NOT EXISTS `sim_item`" in ddl


def test_raw_insert_sql_has_no_duplicate_merge():
    """N1 重复业务键：原始 INSERT 不合并（与 upsert 的 ON DUPLICATE KEY 不同）。"""

    class FakeCur:
        def __init__(self):
            self.executed = []

        def executemany(self, sql, rows):
            self.executed.append(sql)

        def __enter__(self):
            return self

        def __exit__(self, *args):
            return False

    class FakeConn:
        def __init__(self, cur):
            self._cur = cur

        def cursor(self):
            return self._cur

        def close(self):
            pass

    cur = FakeCur()
    monkeypatch = pytest.MonkeyPatch()
    monkeypatch.setattr(mock_db, "_use_db", lambda: FakeConn(cur))
    try:
        mock_db.insert_rows_raw("sim_item", [{"bizKey": "K-1", "v": 1},
                                             {"bizKey": "K-1", "v": 2}])
    finally:
        monkeypatch.undo()
    assert cur.executed and "ON DUPLICATE KEY" not in cur.executed[0]


def test_update_rows_can_skip_increment_advance():
    """N1 增量列不推进：advance_increment=False 时 SET 子句不含增量列。"""

    class FakeCur:
        def __init__(self):
            self.executed = []

        def execute(self, sql, args=None):
            self.executed.append(sql)

        def __enter__(self):
            return self

        def __exit__(self, *args):
            return False

        @property
        def rowcount(self):
            return 1

    class FakeConn:
        def __init__(self, cur):
            self._cur = cur

        def cursor(self):
            return self._cur

        def close(self):
            pass

    cur = FakeCur()
    monkeypatch = pytest.MonkeyPatch()
    monkeypatch.setattr(mock_db, "_use_db", lambda: FakeConn(cur))
    try:
        mock_db.update_rows("sim_item", {"itemName": "改后"}, {"bizKey": "K-1"},
                            "sourceUpdatedAt", advance_increment=False)
    finally:
        monkeypatch.undo()
    assert cur.executed
    sql = cur.executed[0]
    assert "`itemName`=%s" in sql
    assert "sourceUpdatedAt" not in sql  # 增量列未被写入


def test_slow_view_ddl_is_controllable():
    ddl = mock_db.build_slow_view_ddl("v_slow_sim_item", "sim_item", 0.5)
    assert ddl.startswith("CREATE OR REPLACE VIEW `v_slow_sim_item`")
    assert "SLEEP(0.5)" in ddl
    assert "FROM `sim_item` t" in ddl


def test_drop_sql_targets_simulator_table():
    """N1 源表不存在：drop 只在模拟库执行（表名反引号包裹，无库名拼接）。"""

    class FakeCur:
        def __init__(self):
            self.executed = []

        def execute(self, sql):
            self.executed.append(sql)

        def __enter__(self):
            return self

        def __exit__(self, *args):
            return False

    class FakeConn:
        def __init__(self, cur):
            self._cur = cur

        def cursor(self):
            return self._cur

        def close(self):
            pass

    cur = FakeCur()
    monkeypatch = pytest.MonkeyPatch()
    monkeypatch.setattr(mock_db, "_use_db", lambda: FakeConn(cur))
    try:
        mock_db.drop_table("sim_item")
        mock_db.drop_view("v_slow_sim_item")
    finally:
        monkeypatch.undo()
    assert cur.executed == ["DROP TABLE IF EXISTS `sim_item`",
                            "DROP VIEW IF EXISTS `v_slow_sim_item`"]


# ---------------------------------------------------------------------------
# N3：入站客户端异常用例（本地捕获请求验证，不触达 Ai-DOP）
# ---------------------------------------------------------------------------

from clients import inbound_client as ic  # noqa: E402


def _fake_urlopen(monkeypatch, captured: list):
    class FakeResp:
        status = 200

        def read(self):
            return b'{"code": 0}'

        def __enter__(self):
            return self

        def __exit__(self, *args):
            return False

    def fake(req, timeout=None):
        captured.append({
            "url": req.full_url,
            "headers": dict(req.headers),
            "data": req.data,
        })
        return FakeResp()

    monkeypatch.setattr(ic.urllib.request, "urlopen", fake)
    return captured


@pytest.fixture()
def envelope():
    return {"data": {"list": [
        {"ItemNum": "SIM-1", "Descr": "甲", "sourceUpdatedAt": "2026-10-01T00:00:00"},
        {"ItemNum": "SIM-2", "Descr": "乙", "sourceUpdatedAt": "2026-10-02T00:00:00"},
    ]}}


def _client():
    return ic.InboundClient("http://127.0.0.1:5005", "SIM-ACCESS-KEY", "sim-secret")


def _h(headers: dict, name: str):
    """urllib 会把 header 名 title-case 化（X-access-key），大小写不敏感取值。"""
    for k, v in headers.items():
        if k.lower() == name.lower():
            return v
    raise KeyError(name)


def test_case_missing_access_key_omits_header(monkeypatch, envelope):
    cap = []
    _fake_urlopen(monkeypatch, cap)
    _client().case_missing_access_key("MDM_ITEM", envelope)
    assert not any(k.lower() == "x-access-key" for k in cap[0]["headers"])


def test_case_wrong_access_key_is_self_consistent(monkeypatch, envelope):
    cap = []
    _fake_urlopen(monkeypatch, cap)
    _client().case_wrong_access_key("MDM_ITEM", envelope)
    sent = cap[0]
    assert _h(sent["headers"], "X-Access-Key") == "SIM-UNKNOWN-ACCESS-KEY"
    # 签名自洽：用发送的 key 重算应等于请求头中的签名
    ts = _h(sent["headers"], "X-Timestamp")
    nonce = _h(sent["headers"], "X-Nonce")
    digest = _h(sent["headers"], "X-Content-SHA256")
    idem = _h(sent["headers"], "Idempotency-Key")
    msg = ic.build_message("POST", "/api/mdp/inbound/MDM_ITEM",
                           "SIM-UNKNOWN-ACCESS-KEY", ts, nonce, digest, idem)
    assert _h(sent["headers"], "X-Signature") == ic.sign("sim-secret", msg)


def test_case_missing_required_field_drops_field(monkeypatch, envelope):
    cap = []
    _fake_urlopen(monkeypatch, cap)
    _client().case_missing_required_field("MDM_ITEM", envelope, "Descr")
    body = json.loads(cap[0]["data"].decode("utf-8"))
    for row in body["data"]["list"]:
        assert "Descr" not in row
    assert envelope["data"]["list"][0]["Descr"] == "甲"  # 原始信封未被修改


def test_case_wrong_field_type_replaces_value(monkeypatch, envelope):
    cap = []
    _fake_urlopen(monkeypatch, cap)
    _client().case_wrong_field_type("MDM_ITEM", envelope, "Descr", 12345)
    body = json.loads(cap[0]["data"].decode("utf-8"))
    assert body["data"]["list"][0]["Descr"] == 12345


def test_case_stale_rows_backdates_timestamp(monkeypatch, envelope):
    cap = []
    _fake_urlopen(monkeypatch, cap)
    _client().case_stale_rows("MDM_ITEM", envelope)
    body = json.loads(cap[0]["data"].decode("utf-8"))
    for row in body["data"]["list"]:
        assert row["sourceUpdatedAt"] == "2000-01-01T00:00:00"


def test_case_unknown_entity_targets_unknown_code(monkeypatch, envelope):
    cap = []
    _fake_urlopen(monkeypatch, cap)
    _client().case_unknown_entity(envelope)
    assert cap[0]["url"].endswith("/api/mdp/inbound/SIM_UNKNOWN_ENTITY_XX")


def test_case_bulk_partial_failure_drops_one_line_field(monkeypatch, envelope):
    cap = []
    _fake_urlopen(monkeypatch, cap)
    lines = envelope["data"]["list"]
    _client().case_bulk_partial_failure("MDM_ITEM", lines, "Descr", bad_idx=1)
    assert _h(cap[0]["headers"], "Content-Type") == "application/x-ndjson"
    ndjson = cap[0]["data"].decode("utf-8").splitlines()
    row0, row1 = json.loads(ndjson[0]), json.loads(ndjson[1])
    assert "Descr" in row0 and "Descr" not in row1  # 仅坏行缺字段


def test_case_snapshot_commit_failure_targets_missing_snapshot(monkeypatch):
    cap = []
    _fake_urlopen(monkeypatch, cap)
    _client().case_snapshot_commit_failure("MDM_ITEM")
    assert "SNAP-NOT-EXIST-0001" in cap[0]["url"]


# ---------------------------------------------------------------------------
# N4：场景目录、预置/恢复、异常报告（app.py 路由，TestClient 进程内）
# ---------------------------------------------------------------------------

import app  # noqa: E402  conftest 已把模拟器根目录加入 sys.path


@pytest.fixture()
def app_client():
    return TestClient(app.app)


def test_fault_catalog_matches_taskbook_counts(app_client):
    """任务书 §12-§14：甲 10 / 乙 16 / 丙 17（原文清单数），N4 八列。"""
    body = app_client.get("/api/fault/catalog").json()
    counts = body["counts"]
    assert counts == {"DB_SYNC": 10, "API_PULL": 16, "API_INBOUND": 17, "total": 43}
    labels = [label for _, label in body["reportFields"]]
    assert labels == ["场景", "模拟端行为", "预期结果", "实际响应",
                      "Ai-DOP 流水状态", "目标数据是否变化", "是否符合预期", "备注"]
    # 每个场景都必须带恢复说明（验收：异常可一键或按说明恢复）
    for s in body["scenarios"]:
        assert s["recover"].strip(), s["id"]


def test_fault_apply_mock_api_and_recover(app_client):
    """N2 预置：apply 改内存态，recover 恢复默认。"""
    r = app_client.post("/api/fault/apply", json={"scenarioId": "N2-05"})
    assert r.status_code == 200 and r.json()["applied"] is True
    assert app.MOCK_STATE.settings.force_status == 500
    # 恢复默认
    r = app_client.post("/api/fault/recover", json={"channel": "API_PULL"})
    assert r.status_code == 200
    assert app.MOCK_STATE.settings.force_status is None
    assert app.MOCK_STATE.settings.empty_response is False


def test_fault_apply_guide_scenario_is_readonly(app_client):
    """N1 指引类场景：模拟器不修改 Ai-DOP 配置，返回 guide 说明。"""
    r = app_client.post("/api/fault/apply", json={"scenarioId": "N1-01"})
    body = r.json()
    assert body["applied"] is False and body["guide"] is True
    assert "人工" in body["note"]


def test_fault_apply_override_missing_bizkey(app_client):
    r = app_client.post("/api/fault/apply", json={"scenarioId": "N2-13"})
    assert r.json()["applied"] is True
    path = r.json()["overridePath"]
    rows = app.MOCK_STATE.overrides[path]
    assert rows and all("bizKey" not in row for row in rows)
    app_client.post("/api/fault/recover", json={"channel": "API_PULL"})
    assert not app.MOCK_STATE.overrides


def test_fault_apply_inbound_without_creds_is_tool_blocked_not_reject(app_client, monkeypatch):
    """N3 apply 缺环境变量时：如实透传 400，不得包装成 expectedReject。

    任务书验收要求「正确拒绝与工具失败区分」：
    Ai-DOP 预检拦截（409）才算 expectedReject（= 测试通过）；
    模拟器自身缺凭据（400）是工具阻塞，请求根本没发出。
    """
    for name in ("AIDOP_SIM_INBOUND_ACCESS_KEY", "AIDOP_SIM_INBOUND_SECRET"):
        monkeypatch.delenv(name, raising=False)
    r = app_client.post("/api/fault/apply", json={"scenarioId": "N3-01", "objectCode": "ITEM"})
    assert r.status_code == 400
    body = r.json()
    assert "expectedReject" not in body
    assert "缺少环境变量" in body["detail"]


class _FakeInboundClient:
    """拦截发往 Ai-DOP 的请求，捕获构造后的信封/块/坏值。"""

    def __init__(self, *a, **k):
        self.calls = []

    def op_push(self, entity, envelope, contract_version=None):
        self.calls.append(("push", entity, json.loads(json.dumps(envelope))))
        return 202, {"accepted": 2, "rejected": 1}, {"X-Test": "1"}

    def op_bulk(self, entity, bulk_rows, contract_version=None):
        self.calls.append(("bulk", entity, json.loads(json.dumps(bulk_rows))))
        return 202, {"accepted": 1, "rejected": 1}, {"X-Test": "1"}

    def case_wrong_field_type(self, entity, envelope, field, bad_value):
        self.calls.append(("wrong-field-type", entity, field,
                           json.loads(json.dumps(bad_value))))
        return 400, {"detail": "type fault"}, {}


def _patch_inbound(monkeypatch, fake):
    monkeypatch.setenv("AIDOP_SIM_INBOUND_ACCESS_KEY", "TEST-AK")
    monkeypatch.setenv("AIDOP_SIM_INBOUND_SECRET", "TEST-SK")
    monkeypatch.setattr(app, "InboundClient", lambda *a, **k: fake)
    monkeypatch.setattr(app.aidop_probe, "precheck_inbound_entity",
                        lambda client, entity, allowed_hosts, contract_version=None:
                        {"configState": "READY", "httpStatus": 200})


def test_fault_apply_partial_reject_duplicates_row_when_single_sample(app_client, monkeypatch):
    """N3-12 回归：样例仅 1 行时复制一行再破坏，保证「1 好 1 坏」部分行拒绝可演示。"""
    fake = _FakeInboundClient()
    _patch_inbound(monkeypatch, fake)
    r = app_client.post("/api/fault/apply", json={"scenarioId": "N3-12", "objectCode": "ITEM"})
    assert r.status_code == 200, r.text
    op, entity, env = fake.calls[0]
    assert op == "push" and entity == "MDM_ITEM"
    rows = env["data"]["list"]
    assert len(rows) == 2
    assert "ItemNum" in rows[0]
    assert "ItemNum" not in rows[1]       # 复制出的坏行缺必填字段


def test_fault_apply_bulk_partial_failure_duplicates_chunk_when_single_row(app_client, monkeypatch):
    """N3-16 回归：样例仅 1 行时复制一个块（新幂等键）再破坏，避免单 chunk 全失败退化。"""
    fake = _FakeInboundClient()
    _patch_inbound(monkeypatch, fake)
    r = app_client.post("/api/fault/apply", json={"scenarioId": "N3-16", "objectCode": "ITEM"})
    assert r.status_code == 200, r.text
    op, entity, chunks = fake.calls[0]
    assert op == "bulk" and entity == "MDM_ITEM"
    assert len(chunks) == 2
    keys = [c["idempotencyKey"] for c in chunks]
    assert len(set(keys)) == 2            # 复制的块带新幂等键
    assert "ItemNum" in chunks[0]["data"]["list"][0]
    assert "ItemNum" not in chunks[1]["data"]["list"][0]


def test_fault_apply_wrong_field_type_uses_structural_hostile_value(app_client, monkeypatch):
    """N3-11 回归：坏值必须是结构敌对值（嵌套对象），同类标量可能被宽容转换不触发校验。"""
    fake = _FakeInboundClient()
    _patch_inbound(monkeypatch, fake)
    r = app_client.post("/api/fault/apply", json={"scenarioId": "N3-11", "objectCode": "ITEM"})
    assert r.status_code == 200, r.text
    op, entity, field, bad = fake.calls[0]
    assert op == "wrong-field-type" and entity == "MDM_ITEM" and field == "ItemNum"
    assert isinstance(bad, dict) and not isinstance(bad, int)
    assert "ItemNum" not in bad


def test_fault_apply_n1_duplicate_bizkey_drops_table_first(app_client, monkeypatch):
    """N1-07 回归：表已存在时 CREATE IF NOT EXISTS 不重建，必须先 DROP 才能去掉旧唯一键。

    真实 MySQL 实证过：不先 DROP 会残留 uk_ 唯一索引，重复行插入报 1062。
    """
    monkeypatch.setenv("AIDOP_SIM_MYSQL_PASSWORD", "x")
    calls = []
    monkeypatch.setattr(app.mock_db, "drop_table", lambda t: calls.append(("drop", t)))
    monkeypatch.setattr(app.mock_db, "init_schema",
                        lambda tables: calls.append(("init", tables["sim_item"].get("unique_key"))))
    monkeypatch.setattr(app.mock_db, "insert_rows_raw",
                        lambda t, rows: calls.append(("insert", len(rows))))
    r = app_client.post("/api/fault/apply", json={"scenarioId": "N1-07", "objectCode": "ITEM"})
    assert r.status_code == 200, r.text
    assert calls[0] == ("drop", "sim_item")
    assert calls[1] == ("init", False)          # 无唯一键表
    assert calls[2] == ("insert", 2)            # 两条相同 bizKey


def test_fault_recover_db_drops_before_rebuild(app_client, monkeypatch):
    """恢复回归：N1-07 把表改成无唯一键后，恢复必须先 DROP 再建
    （init_schema 的 CREATE IF NOT EXISTS 不会改已有表，不 DROP 结构会残留）。"""
    monkeypatch.setenv("AIDOP_SIM_MYSQL_PASSWORD", "x")
    calls = []
    monkeypatch.setattr(app.mock_db, "drop_table", lambda t: calls.append(("drop", t)))
    monkeypatch.setattr(app.mock_db, "init_schema", lambda tables: calls.append(("init", list(tables))))
    monkeypatch.setattr(app.mock_db, "upsert_rows", lambda t, rows: calls.append(("upsert", t)))
    monkeypatch.setattr(app.mock_db, "drop_view", lambda v: calls.append(("dropview", v)))
    r = app_client.post("/api/fault/recover", json={"channel": "DB_SYNC"})
    assert r.status_code == 200, r.text
    idx = calls.index(("drop", "sim_item"))
    assert calls[idx + 1][0] == "init"       # drop 之后才建表
    assert "upsert" in [c[0] for c in calls]  # 样例行装载
    assert ("dropview", "v_slow_sim_item") in calls


def test_fault_report_save_list_export_redacted(app_client, monkeypatch):
    """N4 报告八列 + 脱敏导出。"""
    monkeypatch.setenv("AIDOP_SIM_INBOUND_ACCESS_KEY", "AK-REAL-SECRET-KEY")
    r = app_client.post("/api/fault-report/save", json={
        "scenarioId": "N3-02", "simBehavior": "AccessKey=AK-REAL-SECRET-KEY",
        "expected": "拒绝", "actualResponse": "401", "aidopFlowStatus": "无流水",
        "targetChanged": "否", "pass": "是", "note": "正确拒绝"})
    assert r.status_code == 200, r.text
    report = r.json()["report"]
    assert report["channel"] == "API_INBOUND"
    assert report["scenarioTitle"] == "错误 AccessKey"
    # 脱敏：AccessKey 只保留后四位
    assert "AK-REAL-SECRET-KEY" not in report["simBehavior"]
    assert report["simBehavior"].endswith("-KEY")
    # 列表与导出
    lst = app_client.get("/api/fault-report/list").json()["reports"]
    assert any(item["id"] == report["id"] for item in lst)
    md = app_client.get(f"/api/fault-report/export/{report['id']}")
    assert md.status_code == 200
    assert "异常测试报告" in md.text
    assert "| 场景 |" in md.text and "| 备注 |" in md.text


def test_fault_report_rejects_unknown_scenario(app_client):
    r = app_client.post("/api/fault-report/save", json={"scenarioId": "N9-99"})
    assert r.status_code == 400


def test_fault_report_save_same_scenario_replaces_in_place(app_client, monkeypatch):
    """N4 回归：同场景重复保存 = 原位更新（保留原 id），列表不产生重复行。"""
    monkeypatch.setenv("AIDOP_SIM_INBOUND_ACCESS_KEY", "AK-TEST")
    base = {"scenarioId": "N3-13", "simBehavior": "v1", "expected": "拒绝",
            "actualResponse": "400", "aidopFlowStatus": "无流水",
            "targetChanged": "否", "pass": "是", "note": "第一版"}
    r1 = app_client.post("/api/fault-report/save", json=base).json()
    base["note"] = "第二版"
    r2 = app_client.post("/api/fault-report/save", json=base).json()
    rep = r2["report"]
    assert rep["replaced"] is True
    assert rep["id"] == r1["report"]["id"]      # 原位更新：id 不变
    rows = [x for x in app_client.get("/api/fault-report/list").json()["reports"]
            if x["scenarioId"] == "N3-13"]
    assert len(rows) == 1
    assert rows[0]["note"] == "第二版"


def test_fault_report_save_distinct_scenarios_stay_independent(app_client):
    """N4 回归：不同场景各自一条，互不覆盖。"""
    for sid, note in (("N3-14", "甲"), ("N3-15", "乙")):
        r = app_client.post("/api/fault-report/save", json={
            "scenarioId": sid, "simBehavior": "x", "expected": "拒绝",
            "actualResponse": "403", "aidopFlowStatus": "无流水",
            "targetChanged": "否", "pass": "是", "note": note})
        assert r.status_code == 200
        assert r.json()["report"]["replaced"] is False
    rows = app_client.get("/api/fault-report/list").json()["reports"]
    assert [x for x in rows if x["scenarioId"] == "N3-14"][0]["note"] == "甲"
    assert [x for x in rows if x["scenarioId"] == "N3-15"][0]["note"] == "乙"