"""第二阶段（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"] == "乙"