#!/usr/bin/env python # -*- coding: utf-8 -*- """ M1-02 任务覆盖班级与班主任配置 验收测试。 覆盖《03-接口设计》§7.6： [x] GET /tasks/{id}/orgs 只返回生效的覆盖关系，班级名与班主任名装配正确 [x] PUT /tasks/{id}/orgs 整体替换语义、软移除、历史行复用 [x] 校验：班级必须是 orgType=3、必须同租户、同一个班不能重复 [x] 校验：初审人必须是本租户内已启用且持班主任角色的账号、主副不能同一人 [x] 状态限制：草稿/收集中/已暂停可改，已截止/已归档不可改 [x] 权限：班主任无 task:manage，不能改 为什么这些校验值得逐条断言： 「指定初审人」本质上是一次**授权** —— 那个人将能看到该班学生的身份证号与学籍材料。 所以"不能把 A 校的班挂到 B 校的任务上""不能把非班主任设为初审人" 不是体验问题，是数据隔离与授权边界问题。每一条都要有断言守着。 用法 python scripts/e2e-m1-orgs.py python scripts/e2e-m1-orgs.py --base http://localhost:8080 依赖：cryptography；且需要能 `docker exec xt-pg`（造发布状态与清理测试租户） 退出码：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") PASSED: list[str] = [] FAILED: list[tuple[str, str]] = [] CREATED_TASK_IDS: list[int] = [] CREATED_USER_IDS: list[int] = [] CREATED_TENANT_IDS: list[int] = [] #: 清理时要用的上下文。放在模块级，是为了让**脚本中途崩溃也能清理**。 #: #: ⚠️ 教训：本脚本第一次运行时在准备阶段抛了 TypeError 直接退出， #: 清理段根本没被执行，于是把两个测试账号（hte688ed2 / adm688ed2）留在了库里。 #: 脚本崩了就留下垃圾数据，是自动化测试最容易犯也最容易被忽略的问题 —— #: 因为"跑通了"的时候它不会暴露。 CTX: dict[str, str | None] = {"base": None, "token": None} def cleanup_all() -> None: """ 清理所有本脚本造出来的数据。**幂等**，可以重复调用；单项失败不影响其余项。 被 `main()` 末尾与 `__main__` 的 finally 各调用一次 —— 前者的输出好看， 后者保证崩溃时也能兜住。 """ base = CTX.get("base") token = CTX.get("token") 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) if base and token: for uid in CREATED_USER_IDS: try: call("DELETE", f"{base}/api/v1/users/{uid}", token) except Exception: pass 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) # ── 断言 ────────────────────────────────────────────── 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 show(body: dict, indent: int = 9) -> None: pad = " " * indent text = json.dumps(body, ensure_ascii=False) print(f"{pad}完整响应：{text[:400]}{'…' if len(text) > 400 else ''}") def blocking_hint(body: dict) -> str: """ 把 40001 的结构化校验明细摊平成一句人话。 后端对参数校验失败返回的是 `data.blocking[]`（每项含 field 与 title）， 而不是只给一句"存在阻断性校验问题" —— 这对前端很重要（能直接高亮出错字段）， 但脚本只打印 msg 的话就完全看不出是哪个字段错了，排查要多花好几轮。 """ items = ((body.get("data") or {}).get("blocking")) or [] if not items: return "" parts = [f"{i.get('field')}: {i.get('title')}" for i in items if i] return "　明细[" + "；".join(parts) + "]" if parts else "" def sql(stmt: str) -> tuple[bool, str]: """直接执行 SQL —— 只用于造/清测试夹具。""" 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) # ── HTTP ──────────────────────────────────────────── 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 main() -> int: ap = argparse.ArgumentParser(description="M1-02 覆盖班级与班主任 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"M1-02 覆盖班级与班主任 E2E　base={base}　租户={args.tenant}") print("=" * 72) suffix = uuid.uuid4().hex[:6] # ── 0. 准备 ──────────────────────────────────────── print("\n[0] 准备") status, body = login(base, "admin", args.password, args.tenant) if body.get("code") != 0: fail("管理员登录", f"HTTP {status} code={body.get('code')} msg={body.get('msg')}") return 1 token = body["data"]["token"] tenant_id = (body["data"].get("tenant") or {}).get("id") CTX["base"] = base CTX["token"] = token ok(f"管理员登录（租户 id={tenant_id}）") status, body = call("GET", f"{base}/api/v1/orgs/tree", token) classes: list[dict] = [] school: list[dict] = [] def walk(nodes): for n in nodes or []: if n.get("orgType") == 3: classes.append(n) if n.get("orgType") == 1: school.append(n) walk(n.get("children")) walk(body.get("data")) if len(classes) < 2 or not school: fail("准备组织数据", f"demo 租户需要至少 2 个班级和 1 个学校节点，实际 班级={len(classes)} 学校={len(school)}") return 1 ok(f"取到班级与学校节点：{classes[0]['orgName']} / {classes[1]['orgName']} / {school[0]['orgName']}") # 造一个班主任账号（指定初审人时必须持有班主任角色） # ⚠️ userType 是接口的 @NotNull 字段。界面上"用户类型由角色推导"是前端的职责， # 接口这层仍要求显式传，且服务端会校验它与角色一致（3=班主任，2=学校管理员） ht_user = f"hte{suffix}" status, body = call("POST", f"{base}/api/v1/users", token, { "username": ht_user, "realName": f"E2E班主任{suffix}", "phone": "13900139101", "roleCodes": ["HEAD_TEACHER"], "userType": 3, }) if body.get("code") != 0: fail("造班主任账号", f"code={body.get('code')} msg={body.get('msg')}{blocking_hint(body)}") return 1 head_id = body["data"]["id"] head_pwd = body["data"]["initialPassword"] CREATED_USER_IDS.append(head_id) ok(f"造班主任账号（id={head_id}）") # 再造一个非班主任（学校管理员），用于验证"非班主任不能当初审人" admin2_user = f"adm{suffix}" status, body = call("POST", f"{base}/api/v1/users", token, { "username": admin2_user, "realName": f"E2E管理员{suffix}", "phone": "13900139102", "roleCodes": ["SCHOOL_ADMIN"], "userType": 2, }) non_teacher_id = None if body.get("code") == 0: non_teacher_id = body["data"]["id"] CREATED_USER_IDS.append(non_teacher_id) ok(f"造非班主任账号（id={non_teacher_id}）") else: fail("造非班主任账号", f"code={body.get('code')} msg={body.get('msg')}{blocking_hint(body)}") # 造第二个租户（跨租户隔离用） other_org_id = None status, plat = login(base, "platform", "Platform@123") if plat.get("code") != 0: fail("平台管理员登录", f"code={plat.get('code')} msg={plat.get('msg')}") else: plat_token = plat["data"]["token"] other_code = f"e2e{suffix}" status, body = call("POST", f"{base}/api/v1/tenants", plat_token, { "tenantName": f"E2E 隔离学校 {suffix}", "tenantCode": other_code, "tenantType": 1, "contactName": "测试联系人", "contactPhone": "13900139103", "expireTime": "2027-12-31", "adminUsername": "admin", "adminRealName": "E2E 隔离管理员", "adminPhone": "13900139104", }) if body.get("code") != 0: fail("开通第二个租户", f"code={body.get('code')} msg={body.get('msg')}") else: other_tenant_id = body["data"]["id"] other_pwd = body["data"]["adminInitialPassword"] CREATED_TENANT_IDS.append(other_tenant_id) status, ob = login(base, "admin", other_pwd, other_code) if ob.get("code") != 0: fail("第二个租户管理员登录", f"code={ob.get('code')} msg={ob.get('msg')}") else: other_token = ob["data"]["token"] # 在新租户里建一个班（先建学校→再建班） status, body = call("POST", f"{base}/api/v1/orgs", other_token, { "parentId": 0, "orgName": f"隔离学校{suffix}", "orgType": 1, "orgCode": f"ISOL{suffix}", "sortNo": 1, }) parent = body.get("data") status, body = call("POST", f"{base}/api/v1/orgs", other_token, { "parentId": parent, "orgName": f"隔离班{suffix}", "orgType": 3, "orgCode": f"ISOLC{suffix}", "sortNo": 1, }) if body.get("code") == 0: other_org_id = body["data"] ok(f"第二个租户与它的班级已就绪（orgId={other_org_id}）") else: fail("在第二个租户里建班级", f"code={body.get('code')} msg={body.get('msg')}") # 造一个任务 batch = "2001" status, body = call("POST", f"{base}/api/v1/tasks", token, { "taskName": f"E2E覆盖范围 {suffix}", "collectMode": 1, "batchNo": batch, "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} code={body['data']['taskCode']}）") # ── 1. 空任务 ─────────────────────────────────────── print("\n[1] 空任务的覆盖范围") status, body = call("GET", f"{base}/api/v1/tasks/{task_id}/orgs", token) check("新任务的覆盖范围是空数组（不是 null）", body.get("code") == 0 and body.get("data") == [], f"code={body.get('code')} data={body.get('data')}") # ── 2. 设置覆盖班级 ───────────────────────────────── print("\n[2] 设置覆盖班级与班主任") c1, c2 = classes[0]["id"], classes[1]["id"] status, body = call("PUT", f"{base}/api/v1/tasks/{task_id}/orgs", token, { "orgs": [ {"orgId": c1, "headUserId": head_id, "expectedCount": 28}, {"orgId": c2, "headUserId": None, "expectedCount": 31}, ], }) check("保存成功", body.get("code") == 0, f"code={body.get('code')} msg={body.get('msg')}") status, body = call("GET", f"{base}/api/v1/tasks/{task_id}/orgs", token) rows = body.get("data") or [] check("返回 2 个班", len(rows) == 2, f"实际 {len(rows)} 行：{json.dumps(rows, ensure_ascii=False)[:200]}") by_org = {r["orgId"]: r for r in rows} r1 = by_org.get(c1) or {} r2 = by_org.get(c2) or {} check("第一个班：班级名正确", r1.get("orgName") == classes[0]["orgName"], f"{r1.get('orgName')} != {classes[0]['orgName']}") check("第一个班：初审人姓名带账号（同名同姓时能区分）", (r1.get("headUserName") or "").startswith(f"E2E班主任{suffix}") and f"（{ht_user}）" in (r1.get("headUserName") or ""), f"headUserName={r1.get('headUserName')}") check("第一个班：应填人数保存正确", r1.get("expectedCount") == 28, f"{r1.get('expectedCount')}") check("第一个班：响应里不再有「协办人」字段（V19 已删列）", "deputyUserIds" not in r1 and "deputyUserNames" not in r1, f"keys={sorted(r1.keys())}") check("第二个班：初审人为空（表示取班级负责人，不是「没人审」）", r2.get("headUserId") is None and r2.get("headUserName") is None, f"headUserId={r2.get('headUserId')} headUserName={r2.get('headUserName')}") check("第二个班：应填人数保存正确", r2.get("expectedCount") == 31, f"{r2.get('expectedCount')}") check("status 恒为 1（只返回生效行）", all(r.get("status") == 1 for r in rows), f"{[r.get('status') for r in rows]}") # ── 3. 校验（全部拒绝）────────────────────────────── print("\n[3] 校验：不该放行的都要拦住") status, body = call("PUT", f"{base}/api/v1/tasks/{task_id}/orgs", token, { "orgs": [{"orgId": c1}, {"orgId": c1}], }) check("同一个班出现两次被拒", body.get("code") == 40001, f"code={body.get('code')} msg={body.get('msg')}") status, body = call("PUT", f"{base}/api/v1/tasks/{task_id}/orgs", token, { "orgs": [{"orgId": 999999999}], }) check("不存在的班级被拒 40001", body.get("code") == 40001, f"code={body.get('code')} msg={body.get('msg')}") status, body = call("PUT", f"{base}/api/v1/tasks/{task_id}/orgs", token, { "orgs": [{"orgId": school[0]["id"]}], }) check("把「学校」当成一个班被拒 40001", body.get("code") == 40001, f"code={body.get('code')} msg={body.get('msg')}") check("拒绝理由讲清楚了「不是班级」（而不是笼统的校验失败）", "不是班级" in (body.get("msg") or ""), f"msg={body.get('msg')}") if non_teacher_id: status, body = call("PUT", f"{base}/api/v1/tasks/{task_id}/orgs", token, { "orgs": [{"orgId": c1, "headUserId": non_teacher_id}], }) check("把非班主任设为初审人被拒 40001", body.get("code") == 40001, f"code={body.get('code')} msg={body.get('msg')}") check("拒绝理由讲清楚了「不是班主任」并给出怎么办", "不是班主任" in (body.get("msg") or "") and "账号管理" in (body.get("msg") or ""), f"msg={body.get('msg')}") okk, out = sql("SELECT count(*) FROM information_schema.columns " "WHERE table_name = 'collect_task_org' AND column_name = 'deputy_user_ids'") check("⚠️ collect_task_org.deputy_user_ids 列已被 V19 删除（协办人机制下线）", okk and out.strip() == "0", f"查询结果={out}") if other_org_id: status, body = call("PUT", f"{base}/api/v1/tasks/{task_id}/orgs", token, { "orgs": [{"orgId": other_org_id}], }) check("跨租户的班级被拒 40001（数据隔离边界）", body.get("code") == 40001, f"code={body.get('code')} msg={body.get('msg')}") check("跨租户的班被当作「不存在」，不透露它存在", "不存在" in (body.get("msg") or ""), f"msg={body.get('msg')}") else: fail("跨租户隔离用例", "第二个租户没造出来，这条最重要的隔离断言没能执行") status, body = call("GET", f"{base}/api/v1/tasks/{task_id}/orgs", token) check("以上失败都没有污染数据（仍是 2 个班）", len(body.get("data") or []) == 2, f"实际 {len(body.get('data') or [])} 行") # ── 4. 软移除与历史行复用 ─────────────────────────── print("\n[4] 软移除与历史行复用") status, body = call("PUT", f"{base}/api/v1/tasks/{task_id}/orgs", token, { "orgs": [{"orgId": c1, "headUserId": head_id, "expectedCount": 28}], }) check("只提交 1 个班（整体替换）成功", body.get("code") == 0, f"code={body.get('code')} msg={body.get('msg')}") status, body = call("GET", f"{base}/api/v1/tasks/{task_id}/orgs", token) check("另一个班已从覆盖范围消失", len(body.get("data") or []) == 1, f"实际 {len(body.get('data') or [])} 行") okk, out = sql(f"SELECT count(*) FROM collect_task_org WHERE task_id = {task_id} " f"AND org_id = {c2} AND status = 0") check("被移除的班是软移除（status=0 的行还在，不是物理删除）", okk and out.strip() == "1", f"查询结果={out}") okk, before = sql(f"SELECT count(*) FROM collect_task_org WHERE task_id = {task_id}") status, body = call("PUT", f"{base}/api/v1/tasks/{task_id}/orgs", token, { "orgs": [{"orgId": c1, "headUserId": head_id, "expectedCount": 28}, {"orgId": c2, "expectedCount": 31}], }) check("把移除的班加回来成功", body.get("code") == 0, f"code={body.get('code')} msg={body.get('msg')}") okk, after = sql(f"SELECT count(*) FROM collect_task_org WHERE task_id = {task_id}") check("加回来时复用历史行（行数没有增加）", okk and before.strip() == after.strip(), f"移除后 {before.strip()} 行 → 加回后 {after.strip()} 行") status, body = call("GET", f"{base}/api/v1/tasks/{task_id}/orgs", token) check("加回后仍是 2 个班生效", len(body.get("data") or []) == 2, f"实际 {len(body.get('data') or [])} 行") # 清空 status, body = call("PUT", f"{base}/api/v1/tasks/{task_id}/orgs", token, {"orgs": []}) check("允许清空（合法中间态，发布时由 40019 拦）", body.get("code") == 0, f"code={body.get('code')} msg={body.get('msg')}") status, body = call("GET", f"{base}/api/v1/tasks/{task_id}/orgs", token) check("清空后为空数组", body.get("data") == [], f"{body.get('data')}") # ── 5. 状态限制 ───────────────────────────────────── print("\n[5] 状态限制") # 恢复 2 个班，然后造发布前置数据走到"收集中" call("PUT", f"{base}/api/v1/tasks/{task_id}/orgs", token, { "orgs": [{"orgId": c1, "headUserId": head_id, "expectedCount": 28}, {"orgId": c2, "expectedCount": 31}], }) sql(f"INSERT INTO collect_roster (tenant_id, task_id, org_id, real_name, id_card_no, id_card_hash) " f"VALUES ({tenant_id}, {task_id}, {c1}, '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)") status, body = call("POST", f"{base}/api/v1/tasks/{task_id}/publish", token, {"force": True}) if check("任务已发布（收集中）", body.get("code") == 0, f"code={body.get('code')} msg={body.get('msg')}"): # 故意有一条不传 expectedCount —— 它必须存成"未设置"（null），不能变成 0。 # 若变成 0，"期望 0 人、实际 31 人"这种假告警会出现在每个没填的班上。 mid_orgs = [{"orgId": c1, "headUserId": head_id, "expectedCount": 28}, {"orgId": c2}] if len(classes) > 2: mid_orgs.append({"orgId": classes[2]["id"]}) status, body = call("PUT", f"{base}/api/v1/tasks/{task_id}/orgs", token, {"orgs": mid_orgs}) check("收集中允许改覆盖范围（中途加班是真实需求）", body.get("code") == 0, f"code={body.get('code')} msg={body.get('msg')}") status, body = call("GET", f"{base}/api/v1/tasks/{task_id}/orgs", token) rows2 = {r["orgId"]: r for r in (body.get("data") or [])} check("不传 expectedCount 时存为 null（不是 0）", c2 in rows2 and rows2[c2].get("expectedCount") is None, f"expectedCount={rows2.get(c2, {}).get('expectedCount')}") # 清空已填的应填人数：这一条正是之前触发 500 的路径 status, body = call("PUT", f"{base}/api/v1/tasks/{task_id}/orgs", token, { "orgs": [{"orgId": c1, "headUserId": head_id}], }) check("可以把已填的应填人数清空（不再 500）", body.get("code") == 0, f"code={body.get('code')} msg={body.get('msg')}") status, body = call("GET", f"{base}/api/v1/tasks/{task_id}/orgs", token) check("清空后读回是 null（清空真的落库了，不是静默忽略）", (body.get("data") or [{}])[0].get("expectedCount") is None, f"{body.get('data')}") status, body = call("PUT", f"{base}/api/v1/tasks/{task_id}/orgs", token, { "orgs": [{"orgId": c1, "headUserId": head_id, "expectedCount": 28}], }) check("重新填入应填人数成功", body.get("code") == 0, f"code={body.get('code')} msg={body.get('msg')}") call("POST", f"{base}/api/v1/tasks/{task_id}/close", token, {}) status, body = call("GET", f"{base}/api/v1/tasks/{task_id}", token) closed = (body.get("data") or {}).get("status") == 2 check("任务已截止", closed, f"status={(body.get('data') or {}).get('status')}") status, body = call("PUT", f"{base}/api/v1/tasks/{task_id}/orgs", token, { "orgs": [{"orgId": c1, "headUserId": head_id}], }) check("已截止不可改覆盖范围", body.get("code") == 40001, f"code={body.get('code')} msg={body.get('msg')}") call("POST", f"{base}/api/v1/tasks/{task_id}/archive", token, {}) status, body = call("PUT", f"{base}/api/v1/tasks/{task_id}/orgs", token, { "orgs": [{"orgId": c1}], }) check("已归档不可改覆盖范围", body.get("code") == 40001, f"code={body.get('code')} msg={body.get('msg')}") status, body = call("GET", f"{base}/api/v1/tasks/{task_id}/orgs", token) check("已归档仍可读覆盖范围（只读不等于看不见）", body.get("code") == 0 and len(body.get("data") or []) == 1, f"code={body.get('code')} rows={len(body.get('data') or [])}") # ── 6. 权限 ──────────────────────────────────────── print("\n[6] 权限：班主任只能看不能改") status, body = login(base, ht_user, head_pwd, args.tenant) if check("班主任能登录", body.get("code") == 0, f"code={body.get('code')} msg={body.get('msg')}"): ht_token = body["data"]["token"] status, body = call("PUT", f"{base}/api/v1/tasks/{task_id}/orgs", ht_token, { "orgs": [{"orgId": c1}], }) check("班主任改覆盖范围被拒（无 task:manage）", body.get("code") == 40301, f"code={body.get('code')} msg={body.get('msg')}") status, body = call("GET", f"{base}/api/v1/tasks/{task_id}/orgs", ht_token) check("班主任能读覆盖范围（他需要知道自己带哪些班）", body.get("code") == 0, f"code={body.get('code')} msg={body.get('msg')}") # ── 7. 清理 ──────────────────────────────────────── print("\n[7] 清理测试数据") cleanup_all() ok(f"已清理 {len(CREATED_TASK_IDS)} 个任务、{len(CREATED_USER_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)