lovife
1f3c485ceb
角色与授权（交接摘要 §一·三十五）： - 权限判定合并两个来源：内置角色按代码（BuiltinRole）、自定义角色按库 （sys_role_permission）。实现放 module 层的 DbRoleGrantSource，security 只依赖 接口 RoleGrantSource —— 否则角色服务要调 permissionChecker，两个包迟早成环 - 权限判定必须加缓存：此前是零 IO，而压测基线是"SQL 条数恒定"。缓存放 Redis， 空结果也缓存（否则已删角色的 token 会把库打穿）、Redis 挂了退化成直读库、TTL 只兜底 - V18 建 sys_role.user_type（账号的 user_type 一直从角色反推，自定义角色不在代码映射里） 与 sys_role_menu；8 个接口 + 错误码 40045；契约 119 -> 129 - 内置角色只读（权限码是前后端契约，做成可改 = 一次误操作就能把管理员锁在系统外）； 平台级能力两道防线（界面禁用 + 服务端 40301）；防自锁；改完立即生效（写授权清缓存） - RoleProvisioner 从"已存在就 continue"改成幂等回填 + 每日 03:20 定时任务 —— 修的是"代码改了权限、已开通租户永远停在发放那一刻"，这种漂移不报错也没测试会红 - scripts/e2e-m10-role.py 77/77 需求收敛（交接摘要 §一·三十六，本节的数字为准）： - 删副班主任 / 审核员两个内置角色（6 -> 4）、历史整理器（module/cleanup）、 加入申请（module/join + 学生端 pages/join）、collect_task_org.deputy_user_ids 整列； 审核范围三级回退 -> 两级，连带 6 个 mapper 的 LIKE '%,id,%' 分支 - V19 迁移处理存量：先解 sys_user_role 绑定再软删角色；账号只在"除这两个角色外没有 别的角色"时才软删。必须迁而不能只删代码 —— 角色码删掉后 mergePermissions 返回空集， 存量账号会"能登录、处处 40301"，不抛异常不打日志 - 契约 129 -> 119、错误码 59 -> 54、权限码 30 -> 28、PC 菜单 12 -> 10、 小程序页 10 -> 9、原型 32 -> 29 屏、E2E 39 -> 37 个（断言 1695 条） - UserService.list 对不认识的角色码由"静默丢掉过滤条件"改成失败关闭 （之前传个不存在的 roleCode 会返回全校账号，且看起来完全正常） - 品牌元信息同步改名「深学在线」：pom 名称、/doc.html 标题、配置与启动类注释、 存储自检探针内容；学生端 app 名与全局导航栏标题
561 lines
27 KiB
Python
561 lines
27 KiB
Python
#!/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)
|