test(scripts): 36 个 E2E 与 6 个机械校验工具（1548 条断言）+ 一键交付自检
新增：check-api-parity（契约 121 接口 ↔ 控制器，自带「必须恰好 121」自检）、 check-entity-columns（基类要求的列 vs 迁移建表列，F-35 那个坑踩了四次之后上的工具）、 check-manual（手册引用的错误码与提示文案对得上代码）、check-test-profile（test profile 真起一次）、 e2e-prodguard、load-test（四场景、按端点分别统计）、verify-all（一条命令跑完交付自检）、 dev-seed-audit（界面验收用的演示数据），以及 M1~M9 各轮 E2E。 压测抓到一个真缺陷并修掉（F-92）：门禁 IP 限流原来计「所有请求」，于是一个班连同一个 WiFi 时 60 个不同学生同时核验只过了 20 个 → 改为只计失败 + 单独的成功上限，20/60 → 60/60。 度量工具的偏差会伪装成产品缺陷（F-60/F-77/F-97），所以每个脚本都断言「动作自己的结论字段」 （submitted / created / canExport），而不是 HTTP 200 + code 0。 验证：verify-all.py --quick 5/5；--with-e2e 见 docs/09 §8。
This commit is contained in:
@@ -0,0 +1,377 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""M6 导出与交付 E2E（《03-接口设计》§12.4 ~ §12.6）。
|
||||
|
||||
它验的不是"能不能打个包"，而是**这条链路上四条硬规则**：
|
||||
|
||||
1. **铁律 10**：有不合格文件 → `40035` 且 `canExport=false`，
|
||||
并且清单要**指名道姓**（谁的哪一项）—— 只说"存在不合格文件"等于没说。
|
||||
2. **只装已通过终审的**：待审/被驳回的单子不许进包（交付物不能有半成品）。
|
||||
3. **目录结构与内容**：班级/姓名/规范件名 + 汇总表；解压后能直接交给认定机构。
|
||||
4. **中文文件名在 Windows 解压不乱码** —— 靠 ZIP 的 UTF-8 + EFS 标志位。
|
||||
这一条在服务器上"看着都对"，只在客户机器上出事，所以**必须在产物里断言**：
|
||||
用 commons-compress 读回 entry 名，并检查 general purpose bit 11 被置位。
|
||||
|
||||
跑法（前提：后端 8080 在跑、`python scripts/dev-seed-audit.py` 已摆好演示数据）：
|
||||
|
||||
python scripts/e2e-m6-export.py
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import importlib.util
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import zipfile
|
||||
|
||||
sys.stdout.reconfigure(encoding="utf-8")
|
||||
|
||||
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
PASSED: list[str] = []
|
||||
FAILED: list[tuple[str, str]] = []
|
||||
|
||||
|
||||
def _load(module_name: str, filename: str):
|
||||
"""复用既有 E2E 的工具（登录 / SQL / 打印口径），不复制第二份。"""
|
||||
path = os.path.join(ROOT, "scripts", filename)
|
||||
spec = importlib.util.spec_from_file_location(module_name, path)
|
||||
mod = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(mod)
|
||||
return mod
|
||||
|
||||
|
||||
R = _load("e2e_m2_roster", "e2e-m2-roster.py")
|
||||
|
||||
BASE = "http://localhost:8080"
|
||||
# 演示任务（`scripts/dev-seed-audit.py` 建的）——按名字找，不写死 id
|
||||
DEMO_TASK_NAME = "【演示】审核台"
|
||||
TEACHER_USER = "demot1"
|
||||
TEACHER_PWD = "Demo1234"
|
||||
|
||||
|
||||
def reset_demo_state() -> int:
|
||||
"""
|
||||
把演示数据摆回"刚提交完、还没审"的初始态。
|
||||
|
||||
为什么必须重摆而不是"接着上一轮跑"：
|
||||
本脚本要验的第一件事是**还有没审完的人时必须拦住导出**（铁律 10）。
|
||||
而上一轮跑完，那三份已经全审过了 → 这条断言必然失败，
|
||||
于是"第二次跑就红"——那种红看着像功能坏了，其实只是夹具状态不对。
|
||||
（本脚本第一次跑就踩了这个：第二轮 canExport 直接是 true。）
|
||||
|
||||
直接复用开发用的种子脚本，避免在这里再写一份造数逻辑（两份必然分叉）。
|
||||
"""
|
||||
seed = _load("dev_seed_audit", "dev-seed-audit.py")
|
||||
argv = sys.argv
|
||||
try:
|
||||
sys.argv = ["dev-seed-audit.py", "--reset"]
|
||||
return seed.main()
|
||||
finally:
|
||||
sys.argv = argv
|
||||
|
||||
|
||||
def ok(name: str) -> None:
|
||||
PASSED.append(name)
|
||||
print(f" [OK] {name}")
|
||||
|
||||
|
||||
def fail(name: str, detail: str) -> None:
|
||||
FAILED.append((name, detail))
|
||||
print(f" [FAIL] {name}\n {detail}")
|
||||
|
||||
|
||||
def check(name: str, cond: bool, detail: str = "") -> bool:
|
||||
if cond:
|
||||
ok(name)
|
||||
else:
|
||||
fail(name, detail or "断言不成立")
|
||||
return cond
|
||||
|
||||
|
||||
def sql(stmt: str) -> str:
|
||||
okk, out = R.sql(stmt)
|
||||
return out
|
||||
|
||||
|
||||
def body_of(method: str, path: str, token: str | None = None, payload: dict | None = None):
|
||||
_, b = R.call(method, f"{BASE}{path}", token, payload)
|
||||
return b
|
||||
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser(description="M6 导出与交付 E2E")
|
||||
args = ap.parse_args()
|
||||
|
||||
print("=" * 72)
|
||||
print(" M6 导出与交付 E2E")
|
||||
print("=" * 72)
|
||||
|
||||
print("\n[准备] 把演示数据重摆到「刚提交、还没审」的初始态")
|
||||
if reset_demo_state() != 0:
|
||||
print(" ✗ 摆演示数据失败，先看 dev-seed-audit.py 的输出")
|
||||
return 2
|
||||
|
||||
_, adm = R.login(BASE, "admin", "Admin@123", "demo")
|
||||
if adm.get("code") != 0:
|
||||
print(f" ✗ 管理员登录失败：{adm}")
|
||||
return 2
|
||||
at = adm["data"]["token"]
|
||||
_, tea = R.login(BASE, TEACHER_USER, TEACHER_PWD, "demo")
|
||||
if tea.get("code") != 0:
|
||||
print(f" ✗ 班主任登录失败：{tea}（先跑 scripts/dev-seed-audit.py）")
|
||||
return 2
|
||||
tt = tea["data"]["token"]
|
||||
|
||||
# 找演示任务
|
||||
tasks = body_of("GET", "/api/v1/tasks?pageNum=1&pageSize=200", at)
|
||||
task = next((t for t in (tasks.get("data") or {}).get("list") or []
|
||||
if t.get("taskName") == DEMO_TASK_NAME), None)
|
||||
if not task:
|
||||
print(f" ✗ 找不到演示任务「{DEMO_TASK_NAME}」，先跑 python scripts/dev-seed-audit.py")
|
||||
return 2
|
||||
task_id = task["id"]
|
||||
print(f"\n[0] 演示任务 id={task_id}")
|
||||
|
||||
# ── 1. 前置检查：有人没审完 → 必须拦住，并说清是谁 ──
|
||||
print("\n[1] 前置检查（还有没审完的人）")
|
||||
pre = body_of("GET", f"/api/v1/exports/precheck?taskId={task_id}", at)
|
||||
check("precheck 接口可用", pre.get("code") == 0, json.dumps(pre, ensure_ascii=False)[:200])
|
||||
d = pre.get("data") or {}
|
||||
check("canExport=false（还有没审完的）", d.get("canExport") is False, f"{d.get('canExport')}")
|
||||
check("阻断里指出了 UNAPPROVED 这一类",
|
||||
any(b.get("type") == "UNAPPROVED" for b in (d.get("blocking") or [])),
|
||||
json.dumps((d.get("blocking") or [])[:3], ensure_ascii=False))
|
||||
check("阻断项带姓名（能指名道姓）",
|
||||
all(b.get("realName") for b in (d.get("blocking") or [])))
|
||||
check("summary.unapproved 与阻断条数一致",
|
||||
(d.get("summary") or {}).get("unapproved") == len([b for b in (d.get("blocking") or [])
|
||||
if b.get("type") == "UNAPPROVED"]),
|
||||
json.dumps(d.get("summary"), ensure_ascii=False))
|
||||
check("quickActions 给出下一步（APPROVE_PENDING）",
|
||||
"APPROVE_PENDING" in (d.get("quickActions") or []),
|
||||
json.dumps(d.get("quickActions"), ensure_ascii=False))
|
||||
|
||||
create = body_of("POST", "/api/v1/exports", at, {
|
||||
"taskId": task_id, "exportName": "不该成功", "scope": {"onlyApproved": False}
|
||||
})
|
||||
check("这是前置检查不过时的 40035（不是别的错）", create.get("code") == 40035,
|
||||
f"code={create.get('code')} msg={create.get('msg')}")
|
||||
check("40035 的 data 里带着完整清单（前端能直接列出来）",
|
||||
isinstance(create.get("data"), dict)
|
||||
and "blocking" in (create.get("data") or {}),
|
||||
json.dumps(create.get("data"), ensure_ascii=False)[:200])
|
||||
check("被拦时**没有**创建导出记录",
|
||||
sql(f"select count(*) from collect_export_task where task_id={task_id} "
|
||||
f"and status <> 3;").strip() == "0")
|
||||
|
||||
# ── 2. 把三个学生审完（班主任初审 + 管理员终审）──
|
||||
print("\n[2] 把演示数据审到全部通过")
|
||||
todos = body_of("GET", "/api/v1/audit/todos?node=1&pageNum=1&pageSize=50", tt).get("data") or {}
|
||||
first_ids = [x["submissionId"] for x in (todos.get("list") or [])]
|
||||
check("班主任看到初审待办", len(first_ids) >= 2, f"todos={first_ids}")
|
||||
for sid in first_ids:
|
||||
b = body_of("POST", f"/api/v1/audit/submissions/{sid}/approve", tt,
|
||||
{"actionSource": "BUTTON"})
|
||||
if b.get("code") != 0:
|
||||
fail(f"初审通过 submission={sid}", json.dumps(b, ensure_ascii=False)[:200])
|
||||
check("初审全部通过", True)
|
||||
|
||||
final = body_of("GET", "/api/v1/audit/todos?node=2&pageNum=1&pageSize=50", at).get("data") or {}
|
||||
passes = [x["submissionId"] for x in (final.get("list") or [])]
|
||||
for sid in passes:
|
||||
b = body_of("POST", f"/api/v1/audit/submissions/{sid}/approve", at,
|
||||
{"actionSource": "BUTTON", "finalOpinion": "E2E 终审通过"})
|
||||
if b.get("code") != 0:
|
||||
fail(f"终审通过 submission={sid}", json.dumps(b, ensure_ascii=False)[:200])
|
||||
passed_cnt = sql(f"select count(*) from collect_submission where task_id={task_id} "
|
||||
f"and status=3 and deleted=0;").strip()
|
||||
check("三份填报单都是「已通过」", passed_cnt == "3", f"status=3 的份数={passed_cnt}")
|
||||
|
||||
# ⚠️ 已通过的单子**不会再出现在待办里**（待办只列待处理的），
|
||||
# 所以后面要用到的 id 必须从库里取 —— 用 todos 的返回值会让脚本
|
||||
# "第一次跑得过、第二次跑就空指针"（本脚本已经踩过一次）。
|
||||
final_ids = [int(x) for x in sql(
|
||||
f"select id from collect_submission where task_id={task_id} and status=3 "
|
||||
f"and deleted=0 order by id limit 3;").split()]
|
||||
check("能从库里取到已通过的填报单 id", len(final_ids) == 3, f"{final_ids}")
|
||||
|
||||
# ── 3. 前置检查应当通过 ──
|
||||
print("\n[3] 审完之后：前置检查应当放行")
|
||||
pre2 = body_of("GET", f"/api/v1/exports/precheck?taskId={task_id}", at).get("data") or {}
|
||||
check("canExport=true", pre2.get("canExport") is True,
|
||||
json.dumps(pre2.get("blocking"), ensure_ascii=False)[:300])
|
||||
check("候选人数 = 3", pre2.get("candidateCount") == 3, f"{pre2.get('candidateCount')}")
|
||||
check("文件数 = 9（3 人 × 3 类材料）", pre2.get("fileCount") == 9, f"{pre2.get('fileCount')}")
|
||||
check("原始体积 > 0", (pre2.get("rawSizeBytes") or 0) > 0, f"{pre2.get('rawSizeBytes')}")
|
||||
|
||||
# ── 4. 创建导出并等完成 ──
|
||||
print("\n[4] 创建导出并轮询到完成")
|
||||
created = body_of("POST", "/api/v1/exports", at, {
|
||||
"taskId": task_id,
|
||||
"exportName": "E2E导出-验收",
|
||||
"scope": {"onlyApproved": True},
|
||||
"format": "ZIP",
|
||||
"generateSummaryExcel": True
|
||||
})
|
||||
check("创建成功", created.get("code") == 0, json.dumps(created, ensure_ascii=False)[:250])
|
||||
export_id = (created.get("data") or {}).get("exportId")
|
||||
check("返回 exportId", isinstance(export_id, int), f"{export_id}")
|
||||
|
||||
detail = {}
|
||||
for _ in range(40):
|
||||
detail = body_of("GET", f"/api/v1/exports/{export_id}", at).get("data") or {}
|
||||
if detail.get("status") in (2, 3):
|
||||
break
|
||||
import time
|
||||
time.sleep(1)
|
||||
check("导出完成（status=2）", detail.get("status") == 2,
|
||||
f"status={detail.get('status')} step={detail.get('stepText')} err={detail.get('errorMsg')}")
|
||||
check("进度 100", detail.get("progress") == 100, f"{detail.get('progress')}")
|
||||
check("产物清单非空", bool(detail.get("resultFiles")), json.dumps(detail.get("resultFiles"))[:200])
|
||||
rf = (detail.get("resultFiles") or [{}])[0]
|
||||
check("产物名带导出名", str(rf.get("name", "")).startswith("E2E导出-验收"), f"{rf.get('name')}")
|
||||
check("产物有体积", (rf.get("size") or 0) > 0, f"{rf.get('size')}")
|
||||
check("产物有 sha256 校验值", str(rf.get("checksum", "")).startswith("sha256:"),
|
||||
f"{rf.get('checksum')}")
|
||||
check("stats 里有压缩率", bool((detail.get("stats") or {}).get("compressRatio")),
|
||||
json.dumps(detail.get("stats"), ensure_ascii=False))
|
||||
|
||||
# ── 5. 下载并验内容 ──
|
||||
print("\n[5] 下载并检查包内容")
|
||||
dl = body_of("GET", f"/api/v1/exports/{export_id}/download", at).get("data") or {}
|
||||
check("下载信息给出地址与文件名", bool(dl.get("url")) and bool(dl.get("fileName")),
|
||||
json.dumps(dl, ensure_ascii=False)[:200])
|
||||
check("下载有效期 > 0 秒", (dl.get("expireSeconds") or 0) > 0, f"{dl.get('expireSeconds')}")
|
||||
|
||||
zip_path = os.path.join(ROOT, "scripts", "_e2e-export.zip")
|
||||
code = os.system(f'curl -s -o "{zip_path}" -H "Authorization: Bearer {at}" '
|
||||
f'"{BASE}{dl.get("url")}"')
|
||||
check("下载接口返回 200 且落地成文件",
|
||||
os.path.exists(zip_path) and os.path.getsize(zip_path) > 1000,
|
||||
f"exit={code} size={os.path.getsize(zip_path) if os.path.exists(zip_path) else 'N/A'}")
|
||||
|
||||
names: list[str] = []
|
||||
with zipfile.ZipFile(zip_path) as z:
|
||||
for info in z.infolist():
|
||||
names.append(info.filename)
|
||||
# ★ UTF-8 标志位（general purpose bit 11）—— Windows 解压看的就是它
|
||||
if info.filename.endswith("身份证.jpg") or info.filename.endswith("汇总表.xlsx"):
|
||||
check(f"「{info.filename}」置了 UTF-8 标志位（bit 11）",
|
||||
bool(info.flag_bits & 0x800), f"flag_bits={info.flag_bits:#x}")
|
||||
# 汇总表内容。⚠️ 文件名按 PRD §5.9 的官方叫法，不是"汇总表.xlsx"
|
||||
xlsx = [n for n in names if n.endswith("（汇总）.xlsx")]
|
||||
check("包里有汇总表（文件名用机构叫法）", bool(xlsx), f"{names[:5]}")
|
||||
if xlsx:
|
||||
from openpyxl import load_workbook
|
||||
with z.open(xlsx[0]) as fh:
|
||||
wb = load_workbook(io.BytesIO(fh.read()))
|
||||
ws = wb.active
|
||||
headers = [c.value for c in ws[1]]
|
||||
check("汇总表 21 列（对齐官方模板 A~U）", len(headers) == 21, f"{len(headers)}：{headers}")
|
||||
check("汇总表表头关键列正确",
|
||||
headers[:4] == ["序号", "姓名", "身份证号", "性别"] and headers[-1] == "备注",
|
||||
f"{headers[:4]} … {headers[-1]}")
|
||||
check("汇总表数据行 = 3", ws.max_row == 4, f"max_row={ws.max_row}")
|
||||
names_col = [ws.cell(row=r, column=2).value for r in range(2, ws.max_row + 1)]
|
||||
check("汇总表里三个学生都在", sorted(n for n in names_col if n) ==
|
||||
sorted(["演示甲", "演示乙", "演示丙"]), f"{names_col}")
|
||||
idcard = str(ws.cell(row=2, column=3).value or "")
|
||||
check("汇总表里的证件号是脱敏的（不含 18 位明文）",
|
||||
"*" in idcard and len(idcard) <= 18, f"{idcard}")
|
||||
|
||||
# ── 目录结构必须与 PRD §5.9 的规范一致（不是"看着合理"就行）──
|
||||
check("根目录是「{批次号}-{任务名称}」",
|
||||
any(n.startswith("202609-【演示】审核台/") for n in names),
|
||||
f"{sorted({n.split('/')[0] for n in names})}")
|
||||
check("第二层是班级", any(n.count("/") >= 2 and n.split("/")[1].startswith("E2E 24数控1班")
|
||||
for n in names),
|
||||
f"{sorted({n.split('/')[1] for n in names if n.count('/') >= 2})}")
|
||||
mats = sorted({n.split("/")[2] for n in names if n.count("/") >= 3})
|
||||
# ⚠️ 用 set 比较，不用"排序后相等"：中文的排序结果取决于码点，
|
||||
# 写死一个顺序会让断言在换一台机器/换一个 Python 版本时莫名其妙地红。
|
||||
check("第三层是**材料类型**（一寸照片/身份证/学籍表）",
|
||||
set(mats) == {"一寸照片", "身份证", "学籍表"}, f"{mats}")
|
||||
check("第四层是文件（班级/材料/文件 三层目录 + 文件名）",
|
||||
all(n.count("/") == 3 for n in names if n.endswith(".jpg")),
|
||||
f"{[n for n in names if n.endswith('.jpg')][:3]}")
|
||||
check("汇总表在根目录（不进子目录）",
|
||||
all(n.count("/") == 1 for n in names if n.endswith("（汇总）.xlsx")),
|
||||
f"{xlsx}")
|
||||
check("文件名是规范件名（带姓名与证件号）",
|
||||
all("演示" in n for n in names if n.endswith(".jpg")),
|
||||
f"{[n for n in names if n.endswith('.jpg')][:3]}")
|
||||
check("没有把草稿/原始件混进去（无 raw/ 前缀）",
|
||||
not any(n.startswith("raw/") or "_raw" in n for n in names), f"{names[:5]}")
|
||||
|
||||
# 目录树预览接口：必须与产物一致（否则"预览"就是谎言）
|
||||
tree = body_of("GET", f"/api/v1/exports/preview?taskId={task_id}&onlyApproved=true",
|
||||
at).get("data") or {}
|
||||
check("目录树预览接口可用", tree.get("root") == "202609-【演示】审核台",
|
||||
f"root={tree.get('root')}")
|
||||
check("预览的班级节点与产物一致",
|
||||
{x.get("name") for x in (tree.get("children") or [])}
|
||||
== {n.split("/")[1] for n in names if n.count("/") >= 2},
|
||||
f"{[x.get('name') for x in (tree.get('children') or [])]}")
|
||||
preview_mats = {m.get("name") for c in (tree.get("children") or [])
|
||||
for m in (c.get("children") or [])}
|
||||
check("预览的材料目录与产物一致", preview_mats == set(mats), f"{preview_mats}")
|
||||
check("预览标注了人数与文件数",
|
||||
tree.get("personCount") == 3 and tree.get("fileCount") == 9,
|
||||
f"person={tree.get('personCount')} file={tree.get('fileCount')}")
|
||||
|
||||
os.remove(zip_path)
|
||||
|
||||
# ── 6. 导出的连带效果 ──
|
||||
print("\n[6] 导出之后的连带约束")
|
||||
#
|
||||
# ⚠️ 这里**不能**断言"撤回返回 40033"：`RevokePolicy` 在导出之后直接把
|
||||
# `canRevoke` 置 false（理由写明"已导出打包，不可撤回"），
|
||||
# 前端因此拿不到 `auditRecordId` → 真去调接口会先撞"缺少要撤回的记录 id"（40001）。
|
||||
# 也就是说 **40033 在界面上不会出现**，它只在"绕过前端直接调接口"时才是可达的。
|
||||
# 所以这里断言的是**用户与前端真正看得见的东西**：policy 说不许撤，且理由点名"已导出"。
|
||||
fdetail = body_of("GET", f"/api/v1/audit/submissions/{final_ids[0]}", at).get("data") or {}
|
||||
rp = fdetail.get("revokePolicy") or {}
|
||||
check("已导出之后 canRevoke=false", rp.get("canRevoke") is False, json.dumps(rp, ensure_ascii=False))
|
||||
check("不能撤的理由点名了「已导出」", "导出" in str(rp.get("reason")),
|
||||
f"{rp.get('reason')}")
|
||||
check("前端拿不到可撤回的记录 id（所以不会误发请求）", rp.get("auditRecordId") is None,
|
||||
f"{rp.get('auditRecordId')}")
|
||||
revoke = body_of("POST", f"/api/v1/audit/submissions/{final_ids[0]}/revoke", at,
|
||||
{"reason": "E2E：导出后不该能撤"})
|
||||
check("绕过前端直接调撤回：被拒（40033 或缺记录 id 都算拒）",
|
||||
revoke.get("code") in (40033, 40001),
|
||||
f"code={revoke.get('code')} msg={revoke.get('msg')}")
|
||||
still = sql(f"select status from collect_submission where id={final_ids[0]};").strip()
|
||||
check("那一单仍然是「已通过」（没被撤掉）", still == "3", f"status={still}")
|
||||
lst = body_of("GET", f"/api/v1/exports?taskId={task_id}&pageNum=1&pageSize=10", at).get("data") or {}
|
||||
check("导出列表里有这次记录", any(x.get("id") == export_id for x in (lst.get("list") or [])),
|
||||
json.dumps(lst, ensure_ascii=False)[:200])
|
||||
check("列表里带任务名", all(x.get("taskName") for x in (lst.get("list") or []) if x.get("taskId")),
|
||||
json.dumps((lst.get("list") or [{}])[0], ensure_ascii=False)[:200])
|
||||
|
||||
# ── 7. 边界 ──
|
||||
print("\n[7] 边界与权限")
|
||||
b = body_of("POST", "/api/v1/exports", at, {"taskId": task_id, "exportName": "空数据",
|
||||
"scope": {"orgIds": [999999]}})
|
||||
check("指定不存在的班 → 没有候选数据被拦住", b.get("code") != 0,
|
||||
f"code={b.get('code')} msg={b.get('msg')}")
|
||||
b = body_of("POST", "/api/v1/exports", at, {"exportName": "没任务"})
|
||||
check("缺 taskId → 40001", b.get("code") == 40001, f"code={b.get('code')}")
|
||||
b = body_of("GET", f"/api/v1/exports/{export_id}", tt)
|
||||
check("班主任（无 export:download）看导出记录 → 40301 或 403",
|
||||
b.get("code") in (40301, 403) or b.get("_http") == 403,
|
||||
f"code={b.get('code')} http={b.get('_http')}")
|
||||
b = body_of("GET", "/api/v1/exports/99999999", at)
|
||||
check("不存在的导出 id → 明确报错而不是 500", b.get("code") == 40001, f"code={b.get('code')}")
|
||||
|
||||
print("\n" + "=" * 72)
|
||||
print(f"通过 {len(PASSED)} 失败 {len(FAILED)} 共 {len(PASSED) + len(FAILED)}")
|
||||
print("=" * 72)
|
||||
for n, det in FAILED:
|
||||
print(f" ✗ {n}\n {det}")
|
||||
return 1 if FAILED else 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user