lovife
0bc7ee02dc
小程序端从零建起来了，并在微信开发者工具里真机模拟跑通： 扫 scene=t1-k17 → 显示「示范职业学校 / 2026年9月机电专业认定材料收集 / 尚未发布」。 交付： · frontend-mp/（uni-app，官方 uni-preset-vue#vite-ts 模板），appid 已填 · src/pages/index/index.vue 扫码落地页：读 scene → 调后端 → 显示 「哪个学校 / 填什么 / 现在能不能填」 · GET /api/v1/mp/scene-context（匿名）：登录前的落地页查询 · 主题色与 PC 管理端同一套（腾讯云 TDesign 色板），两端视觉一致 为什么落地页必然是匿名接口： 学生扫开的第一屏要回答"这是哪个学校、要填什么、能不能填"， 而这发生在绑定身份之前。它与契约里的 POST /mp/login（code+scene 换 session） 不是一回事 —— 那个要换 openid，属 M3。 ⚠️ 一个必踩的坑（撞上并绕开了）： FillTenantLineHandler.getTenantId() 在**匿名请求下直接抛 40101**， 所以不能用 taskMapper.selectById()。正确写法是专用 Mapper + 每条查询 @InterceptorIgnore(tenantLine="true") 并显式带 tenant_id（项目注释称此为 一次代码审查点）。见 module/mp/mapper/MpSceneMapper.java —— 把这几条集中在 一个文件里，正是为了让"关掉框架保护后由人保证隔离"这件事有地方审。 ⚠️ scene 可穷举（t1-k1、t1-k2…），所以匿名接口： ① 只返回落地页必须的最少信息，不往里加"顺带能查出来"的东西； ② 按 IP 限流（20 次/分钟，Redis INCR，复用项目的 GATE_RATE_LIMITED 40071）。 E2E 专门验了"前 20 次通过、第 21 次起被限、清掉计数器立刻恢复"。 状态文案分五种而不是合并成"当前不可填报"：学生的下一步动作完全不同 —— 「已截止」要找班主任、「已暂停」要等通知、「尚未发布」是根本不该现在扫。 顺带记两个环境坑（写进交接摘要）： · 微信开发者工具的「服务端口」默认关闭，HBuilderX 运行时报 `IDE service port disabled` 然后退出，**调不起开发者工具**。 解法：开发者工具 → 设置 → 安全设置 → 服务端口 → 开启； 或直接「导入项目」dist/dev/mp-weixin 绕过。 · HBuilderX 把这个项目当 CLI 型项目，产物在 dist/dev/mp-weixin， **不是** unpackage/。模拟扫码要在「编译模式」里设 进入场景=扫描二维码、 启动参数 scene=t1-k17。 · <style lang="scss"> 需要装 sass（模板没带），已 pnpm add -D sass。 ⚠️ 对上线计划最重要的一条结论（已写进交接摘要）： 开发版（develop）只有开发者能扫、且会过期；体验版（trial）只有体验成员能扫； **只有正式版（release）学生才能扫**，而它需要提交审核，涉及小程序类目与资质。 所以 C1「小程序资质法务确认」不是"以后再说"，它直接决定学生能不能开始填报。 已知缺口： · 填报主流程（/mp/login 门禁核验 表单填写）属 M3； 现在点「开始填报」给的是明确的"开发中"提示，不是静默无反应。 · frontend-mp 还没套项目既有的 ESLint/Stylelint 规范。 验收：e2e-m1-scene.py 31/31；八条 E2E 合计 389 项断言全通过； 微信开发者工具里人工确认页面渲染与 scene 传递正确。
386 lines
18 KiB
Python
386 lines
18 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
扫码落地页（GET /api/v1/mp/scene-context）验收测试。
|
||
|
||
覆盖 FR-105（统一小程序 + 扫码带参 scene 识别租户）：
|
||
[x] 正常 scene：返回租户名、任务名、状态、能否填报、给学生的话
|
||
[x] 非法 scene / 不存在的租户 / 不存在的任务：各自给出**不同的**提示
|
||
[x] 租户级 scene（不带任务）：提示去扫任务码
|
||
[x] 带班级：班级名只在"该班确实在任务覆盖范围里"时才返回
|
||
[x] 跨租户：A 租户的 scene 拼上 B 租户的任务 id → 必须当作"不存在"
|
||
[x] 状态文案随任务状态变化（收集中/已暂停/已截止）
|
||
[x] **限流**：scene 可穷举，必须能挡住批量扫描
|
||
[x] 这个接口**匿名**就能访问（学生登录前就要用）
|
||
|
||
为什么这些值得逐条断言：
|
||
这是学生扫码后的第一屏，而且它是**匿名**接口、scene 又可穷举（t1-k1、t1-k2…）。
|
||
一旦哪个分支写错，后果不是"报错"，而是"把不该给的信息给出去了"，
|
||
或者"扫对了码却告诉他这个码不是本系统的" —— 后者会让学生直接放弃。
|
||
|
||
用法
|
||
python scripts/e2e-m1-scene.py
|
||
|
||
依赖：cryptography；且需要能 `docker exec xt-pg` / `xt-redis`
|
||
退出码：0 = 全部通过
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import base64
|
||
import json
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
import urllib.error
|
||
import urllib.parse
|
||
import urllib.request
|
||
import uuid
|
||
|
||
if hasattr(sys.stdout, "reconfigure"):
|
||
sys.stdout.reconfigure(encoding="utf-8")
|
||
|
||
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
PUB_KEY = os.path.join(ROOT, "deploy", "rsa-public-key-dev.pem")
|
||
|
||
PG_CONTAINER = os.environ.get("XT_PG_CONTAINER", "xt-pg")
|
||
PG_USER = os.environ.get("PG_USER", "fill")
|
||
PG_DB = os.environ.get("PG_DB", "filldb")
|
||
PG_PWD = os.environ.get("PG_PASSWORD", "fill_dev_pwd")
|
||
REDIS_CONTAINER = os.environ.get("XT_REDIS_CONTAINER", "xt-redis")
|
||
REDIS_PWD = os.environ.get("REDIS_PASSWORD", "redis_dev_pwd")
|
||
|
||
PASSED: list[str] = []
|
||
FAILED: list[tuple[str, str]] = []
|
||
|
||
CREATED_TASK_IDS: list[int] = []
|
||
CREATED_TENANT_IDS: list[int] = []
|
||
|
||
|
||
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) -> tuple[bool, str]:
|
||
try:
|
||
r = subprocess.run(
|
||
["docker", "exec", "-e", f"PGPASSWORD={PG_PWD}", PG_CONTAINER,
|
||
"psql", "-U", PG_USER, "-d", PG_DB, "-t", "-A", "-c", stmt],
|
||
capture_output=True, text=True, timeout=40)
|
||
return r.returncode == 0, (r.stdout or r.stderr).strip()
|
||
except Exception as e:
|
||
return False, str(e)
|
||
|
||
|
||
def redis(args: list[str]) -> tuple[bool, str]:
|
||
try:
|
||
r = subprocess.run(
|
||
["docker", "exec", "-e", f"REDISCLI_AUTH={REDIS_PWD}", REDIS_CONTAINER,
|
||
"redis-cli"] + args, capture_output=True, text=True, timeout=30)
|
||
return r.returncode == 0, (r.stdout or r.stderr).strip()
|
||
except Exception as e:
|
||
return False, str(e)
|
||
|
||
|
||
def cleanup_all() -> None:
|
||
if CREATED_TASK_IDS:
|
||
ids = ",".join(str(i) for i in CREATED_TASK_IDS)
|
||
for stmt in [
|
||
f"DELETE FROM collect_task_org WHERE task_id IN ({ids})",
|
||
f"DELETE FROM collect_material_def WHERE task_id IN ({ids})",
|
||
f"DELETE FROM collect_roster WHERE task_id IN ({ids})",
|
||
f"DELETE FROM collect_task WHERE id IN ({ids})",
|
||
]:
|
||
sql(stmt)
|
||
for tid in CREATED_TENANT_IDS:
|
||
for stmt in [
|
||
f"DELETE FROM sys_user_role WHERE user_id IN (SELECT id FROM sys_user WHERE tenant_id = {tid})",
|
||
f"DELETE FROM sys_user WHERE tenant_id = {tid}",
|
||
f"DELETE FROM sys_org WHERE tenant_id = {tid}",
|
||
f"DELETE FROM sys_role WHERE tenant_id = {tid}",
|
||
f"DELETE FROM collect_task WHERE tenant_id = {tid}",
|
||
f"DELETE FROM sys_tenant WHERE id = {tid}",
|
||
]:
|
||
sql(stmt)
|
||
redis(["DEL", "xt:mp:scene:rate:127.0.0.1"])
|
||
redis(["DEL", "xt:mp:scene:rate:localhost"])
|
||
|
||
|
||
def rsa_encrypt(plain: str) -> str:
|
||
from cryptography.hazmat.primitives import serialization
|
||
from cryptography.hazmat.primitives.asymmetric import padding
|
||
with open(PUB_KEY, "rb") as f:
|
||
pub = serialization.load_pem_public_key(f.read())
|
||
return base64.b64encode(pub.encrypt(plain.encode(), padding.PKCS1v15())).decode()
|
||
|
||
|
||
def call(method: str, url: str, token: str | None = None,
|
||
body: dict | None = None, timeout: int = 30) -> tuple[int, dict]:
|
||
headers = {"Content-Type": "application/json; charset=utf-8"}
|
||
if token:
|
||
headers["Authorization"] = f"Bearer {token}"
|
||
data = json.dumps(body, ensure_ascii=False).encode("utf-8") if body is not None else None
|
||
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
||
return resp.status, json.loads(resp.read().decode("utf-8"))
|
||
except urllib.error.HTTPError as e:
|
||
raw = e.read().decode("utf-8")
|
||
try:
|
||
return e.code, json.loads(raw)
|
||
except json.JSONDecodeError:
|
||
return e.code, {"raw": raw}
|
||
|
||
|
||
def login(base: str, username: str, password: str, tenant: str | None = None):
|
||
payload = {"username": username, "password": rsa_encrypt(password)}
|
||
if tenant:
|
||
payload["tenantCode"] = tenant
|
||
return call("POST", f"{base}/api/v1/auth/login", body=payload)
|
||
|
||
|
||
def scene(base: str, value: str, token: str | None = None) -> dict:
|
||
"""调 scene-context，返回 data。"""
|
||
url = f"{base}/api/v1/mp/scene-context?scene={urllib.parse.quote(value)}"
|
||
_, body = call("GET", url, token)
|
||
return body.get("data") or {}
|
||
|
||
|
||
def scene_raw(base: str, value: str) -> tuple[int, dict]:
|
||
url = f"{base}/api/v1/mp/scene-context?scene={urllib.parse.quote(value)}"
|
||
return call("GET", url)
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser(description="扫码落地页 E2E")
|
||
ap.add_argument("--base", default=os.environ.get("XT_BASE", "http://localhost:8080"))
|
||
ap.add_argument("--tenant", default=os.environ.get("XT_TENANT", "demo"))
|
||
ap.add_argument("--password", default="Admin@123")
|
||
args = ap.parse_args()
|
||
base = args.base.rstrip("/")
|
||
|
||
print("=" * 72)
|
||
print(f"扫码落地页 E2E base={base}")
|
||
print(" （本接口是匿名的 —— 学生登录前就要用；测试也刻意不带 token）")
|
||
print("=" * 72)
|
||
|
||
suffix = uuid.uuid4().hex[:6]
|
||
redis(["DEL", "xt:mp:scene:rate:127.0.0.1"])
|
||
|
||
# ── 0. 准备（用管理员造数据，但 scene 查询都不带 token）────
|
||
print("\n[0] 准备")
|
||
status, body = login(base, "admin", args.password, args.tenant)
|
||
if body.get("code") != 0:
|
||
fail("管理员登录", f"code={body.get('code')} msg={body.get('msg')}")
|
||
return 1
|
||
token = body["data"]["token"]
|
||
tenant_id = (body["data"].get("tenant") or {}).get("id")
|
||
ok(f"管理员登录（租户 id={tenant_id}）")
|
||
|
||
# 本次要用的任务
|
||
status, body = call("POST", f"{base}/api/v1/tasks", token, {
|
||
"taskName": f"E2E落地页 {suffix}", "collectMode": 1, "batchNo": "2005",
|
||
"majorName": "数控加工", "certType": 1,
|
||
"startTime": "2026-09-28", "endTime": "2026-12-31",
|
||
})
|
||
if body.get("code") != 0:
|
||
fail("建任务", f"code={body.get('code')} msg={body.get('msg')}")
|
||
return 1
|
||
task_id = body["data"]["id"]
|
||
CREATED_TASK_IDS.append(task_id)
|
||
ok(f"建任务（id={task_id}，草稿状态）")
|
||
|
||
status, body = call("GET", f"{base}/api/v1/orgs/tree", token)
|
||
classes: list[dict] = []
|
||
|
||
def walk(nodes):
|
||
for n in nodes or []:
|
||
if n.get("orgType") == 3:
|
||
classes.append(n)
|
||
walk(n.get("children"))
|
||
|
||
walk(body.get("data"))
|
||
if not classes:
|
||
fail("取班级", "demo 租户里没有班级")
|
||
return 1
|
||
|
||
# ── 1. 正常 scene（匿名）─────────────────────────────
|
||
print("\n[1] 正常 scene（匿名，不带 token）")
|
||
d = scene(base, f"t{tenant_id}-k{task_id}")
|
||
check("返回租户名（学生要能确认扫对了学校）", bool(d.get("tenantName")),
|
||
f"tenantName={d.get('tenantName')}")
|
||
check("返回任务名", d.get("taskName") == f"E2E落地页 {suffix}", f"taskName={d.get('taskName')}")
|
||
check("草稿状态 → 尚未发布", d.get("taskStatusText") == "尚未发布", f"{d.get('taskStatusText')}")
|
||
check("草稿不可填报", d.get("canSubmit") is False, f"canSubmit={d.get('canSubmit')}")
|
||
check("提示语是「尚未开始」（不是笼统的不可填报）",
|
||
"尚未开始" in (d.get("message") or ""), f"message={d.get('message')}")
|
||
check("valid 为 true", d.get("valid") is True, f"valid={d.get('valid')}")
|
||
|
||
# ── 2. 各种非法输入要给不同的提示 ─────────────────────
|
||
print("\n[2] 非法输入：提示要各不相同（否则学生不知道该找谁）")
|
||
d_garbage = scene(base, "garbage")
|
||
d_no_tenant = scene(base, "t99999999-k1")
|
||
d_no_task = scene(base, f"t{tenant_id}-k99999999")
|
||
check("乱码 → 不是本系统的二维码", "不是本系统生成" in (d_garbage.get("message") or ""),
|
||
f"{d_garbage.get('message')}")
|
||
check("不存在的租户 → 单位不存在", "单位不存在" in (d_no_tenant.get("message") or ""),
|
||
f"{d_no_tenant.get('message')}")
|
||
check("不存在的任务 → 任务不存在", "任务不存在" in (d_no_task.get("message") or ""),
|
||
f"{d_no_task.get('message')}")
|
||
msgs = {d_garbage.get("message"), d_no_tenant.get("message"), d_no_task.get("message")}
|
||
check("三种失败的提示互不相同（不是同一句话敷衍）", len(msgs) == 3, f"{msgs}")
|
||
|
||
# ── 3. 租户级 scene ──────────────────────────────────
|
||
print("\n[3] 租户级 scene（M0 开通租户时约定的 t{tenantId}）")
|
||
d_tenant = scene(base, f"t{tenant_id}")
|
||
check("租户级码有效但提示去扫任务码",
|
||
d_tenant.get("valid") is True and "填报二维码" in (d_tenant.get("message") or ""),
|
||
f"valid={d_tenant.get('valid')} message={d_tenant.get('message')}")
|
||
check("租户级码没有任务信息", d_tenant.get("taskId") is None, f"{d_tenant.get('taskId')}")
|
||
|
||
# ── 4. 带班级 ────────────────────────────────────────
|
||
print("\n[4] 带班级的 scene")
|
||
d_not_covered = scene(base, f"t{tenant_id}-k{task_id}-o{classes[0]['id']}")
|
||
check("班级不在覆盖范围时不返回 orgName（不能凭空显示一个班）",
|
||
d_not_covered.get("orgName") is None, f"orgName={d_not_covered.get('orgName')}")
|
||
|
||
call("PUT", f"{base}/api/v1/tasks/{task_id}/orgs", token,
|
||
{"orgs": [{"orgId": classes[0]["id"]}]})
|
||
d_covered = scene(base, f"t{tenant_id}-k{task_id}-o{classes[0]['id']}")
|
||
check("班级在覆盖范围时返回班级名",
|
||
d_covered.get("orgName") == classes[0]["orgName"],
|
||
f"orgName={d_covered.get('orgName')} 期望={classes[0]['orgName']}")
|
||
check("带班不影响任务信息", d_covered.get("taskId") == task_id, f"{d_covered.get('taskId')}")
|
||
|
||
# ── 5. 跨租户 ────────────────────────────────────────
|
||
print("\n[5] 跨租户：A 租户的 scene 拼 B 租户的任务 id")
|
||
status, plat = login(base, "platform", "Platform@123")
|
||
other_task_id = None
|
||
if plat.get("code") == 0:
|
||
plat_token = plat["data"]["token"]
|
||
other_code = f"scn{suffix}"
|
||
status, body = call("POST", f"{base}/api/v1/tenants", plat_token, {
|
||
"tenantName": f"E2E扫码隔离 {suffix}", "tenantCode": other_code, "tenantType": 1,
|
||
"contactName": "测试", "contactPhone": "13900139601", "expireTime": "2027-12-31",
|
||
"adminUsername": "admin", "adminRealName": "隔离管理员", "adminPhone": "13900139602",
|
||
})
|
||
if body.get("code") == 0:
|
||
other_tenant_id = body["data"]["id"]
|
||
other_pwd = body["data"]["adminInitialPassword"]
|
||
CREATED_TENANT_IDS.append(other_tenant_id)
|
||
login(base, "admin", other_pwd, other_code)
|
||
_, ob = login(base, "admin", other_pwd, other_code)
|
||
if ob.get("code") == 0:
|
||
ot = ob["data"]["token"]
|
||
_, tb = call("POST", f"{base}/api/v1/tasks", ot, {
|
||
"taskName": f"别校的任务 {suffix}", "collectMode": 1, "batchNo": "2011",
|
||
"majorName": "x", "certType": 1,
|
||
"startTime": "2026-09-28", "endTime": "2026-12-31",
|
||
})
|
||
if tb.get("code") == 0:
|
||
other_task_id = tb["data"]["id"]
|
||
sql(f"DELETE FROM collect_task WHERE id = {other_task_id}") # 用 SQL 收尾更省事
|
||
if other_task_id:
|
||
d_cross = scene(base, f"t{tenant_id}-k{other_task_id}")
|
||
check("用 A 租户的租户号 + B 租户的任务 id → 任务不存在（不跨租户泄露）",
|
||
"任务不存在" in (d_cross.get("message") or ""), f"message={d_cross.get('message')}")
|
||
d_cross2 = scene(base, f"t{CREATED_TENANT_IDS[0]}-k{task_id}")
|
||
check("反过来（B 的租户号 + A 的任务 id）也不泄露",
|
||
"任务不存在" in (d_cross2.get("message") or ""), f"message={d_cross2.get('message')}")
|
||
else:
|
||
fail("跨租户用例", "第二个租户没造出来，这条隔离断言没执行")
|
||
|
||
# ── 6. 状态文案随状态变化 ────────────────────────────
|
||
print("\n[6] 状态文案：五种状态给五句不同的话")
|
||
sql(f"INSERT INTO collect_roster (tenant_id, task_id, real_name, id_card_no, id_card_hash) "
|
||
f"VALUES ({tenant_id}, {task_id}, 'E2E学生A', 'seed', 'seed')")
|
||
sql(f"INSERT INTO collect_material_def (tenant_id, task_id, material_code, material_name, required) "
|
||
f"VALUES ({tenant_id}, {task_id}, 'IDCARD', '身份证', 1)")
|
||
|
||
call("POST", f"{base}/api/v1/tasks/{task_id}/publish", token, {"force": True})
|
||
d_collecting = scene(base, f"t{tenant_id}-k{task_id}")
|
||
check("收集中 → 可填报", d_collecting.get("canSubmit") is True,
|
||
f"canSubmit={d_collecting.get('canSubmit')} status={d_collecting.get('taskStatusText')}")
|
||
check("收集中 → 提示「开始填报」",
|
||
"开始填报" in (d_collecting.get("message") or ""), f"{d_collecting.get('message')}")
|
||
|
||
call("POST", f"{base}/api/v1/tasks/{task_id}/pause", token, {})
|
||
d_paused = scene(base, f"t{tenant_id}-k{task_id}")
|
||
check("已暂停 → 不可填报", d_paused.get("canSubmit") is False, f"{d_paused.get('canSubmit')}")
|
||
check("已暂停 → 提示等恢复通知（不是「已截止」，学生的下一步动作完全不同）",
|
||
"暂停" in (d_paused.get("message") or ""), f"{d_paused.get('message')}")
|
||
|
||
call("POST", f"{base}/api/v1/tasks/{task_id}/resume", token)
|
||
call("POST", f"{base}/api/v1/tasks/{task_id}/close", token, {})
|
||
d_closed = scene(base, f"t{tenant_id}-k{task_id}")
|
||
check("已截止 → 不可填报", d_closed.get("canSubmit") is False, f"{d_closed.get('canSubmit')}")
|
||
check("已截止 → 提示联系班主任补交（§7.12：码不做失效，靠这句话说明状态）",
|
||
"截止" in (d_closed.get("message") or "") and "班主任" in (d_closed.get("message") or ""),
|
||
f"{d_closed.get('message')}")
|
||
check("截止后仍能查到任务（码永久有效）", d_closed.get("taskName") == f"E2E落地页 {suffix}",
|
||
f"{d_closed.get('taskName')}")
|
||
|
||
# ── 7. 限流 ──────────────────────────────────────────
|
||
print("\n[7] 限流：scene 可穷举，必须挡得住批量扫描")
|
||
rate_key = None
|
||
okk, out = redis(["--scan", "--pattern", "xt:mp:scene:rate:*"])
|
||
if okk and out.strip():
|
||
rate_key = out.strip().splitlines()[0]
|
||
if rate_key:
|
||
redis(["DEL", rate_key])
|
||
codes = []
|
||
for i in range(25):
|
||
st, b = scene_raw(base, f"t{tenant_id}-k{task_id}")
|
||
codes.append((st, b.get("code")))
|
||
# 前 20 次应成功（code=0），第 21 次起应被限流
|
||
passed_n = sum(1 for st, c in codes if c == 0)
|
||
limited_n = sum(1 for st, c in codes if c == 40071)
|
||
check("前 20 次正常通过", passed_n == 20, f"通过 {passed_n} 次")
|
||
check("第 21 次起被限流（40071）", limited_n == 5, f"被限 {limited_n} 次")
|
||
check("限流用的是项目的 GATE_RATE_LIMITED(40071)，不是自制错误码",
|
||
any(c == 40071 for _, c in codes), f"{set(c for _, c in codes)}")
|
||
redis(["DEL", rate_key])
|
||
_, after = scene_raw(base, f"t{tenant_id}-k{task_id}")
|
||
check("清掉计数器后立刻恢复（说明是限流不是封禁）", after.get("code") == 0,
|
||
f"code={after.get('code')}")
|
||
else:
|
||
fail("限流用例", "没找到限流键，说明限流可能没生效")
|
||
|
||
# ── 8. 清理 ──────────────────────────────────────────
|
||
print("\n[8] 清理测试数据")
|
||
cleanup_all()
|
||
ok(f"已清理 {len(CREATED_TASK_IDS)} 个任务、{len(CREATED_TENANT_IDS)} 个测试租户")
|
||
|
||
print("\n" + "=" * 72)
|
||
print(f"通过 {len(PASSED)} 失败 {len(FAILED)} 共 {len(PASSED) + len(FAILED)}")
|
||
if FAILED:
|
||
print("\n失败项：")
|
||
for name, detail in FAILED:
|
||
print(f" · {name}\n {detail}")
|
||
print("=" * 72)
|
||
return 1 if FAILED else 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
code = main()
|
||
except BaseException:
|
||
print("\n ⚠ 脚本异常退出，正在清理…")
|
||
cleanup_all()
|
||
raise
|
||
raise SystemExit(code)
|