lovife
9f6d26c650
账号唯一性是 (tenant_id, username)，跨租户重名本身是合法行为（登录靠租户码区分）；但默认值与 e2e 残留会让它变成默认状态：不带租户码登录会撞上 40012（该账号存在于多个单位），平台侧看账号也分不清哪个 admin 属于哪所学校。三处一起收口 —— 1) scripts/e2e-m0.py 跑完不再留租户：以前每跑一次就留下一个 e2e###### 租户 + 里面叫 admin 的管理员 + demo 下 3 个测试组织，库里那两个孤儿租户就是这么来的。现在按本次跑出的 id 自清（崩溃路径也兜住），不越界成 LIKE 'e2e%' 全清。 2) scripts/e2e-m1-tenant-domain.py 自己造、自己清「另一个租户里也有 admin」的夹具：它原本靠第 1 条的残留凑出 40012 这个前提，残留一被清掉就会静默反转成 8 条 FAIL。顺带修掉同族的第二处隐式依赖 —— 它假定 t14.localhost 是个真实租户，而那个租户早就不存在了。 3) PC 端「开通租户」的登录账号默认值不再写死 admin，改为跟随租户编码生成 admin_<编码>；运营一旦手改就不再被编码覆盖。 验证：全量 python scripts/verify-all.py --with-e2e 通过（37 脚本 / 1597 断言 / 0 失败），跑完库里仍只有 1 个租户、0 处跨租户重名；PC 端 pnpm typecheck 0 错、pnpm lint 干净、check-wiring 未接线 0 条，浏览器实演默认值跟随与手改不回退。
421 lines
18 KiB
Python
421 lines
18 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
M0 验收测试。
|
||
|
||
把《05-开发任务拆解》里 M0 的验收标准变成可重复执行的程序：
|
||
|
||
[x] 能开通租户 → 建组织树 → 各角色登录看到不同菜单
|
||
[x] 跨租户访问被拦截
|
||
[x] schema-postgresql.sql 在测试环境零错误执行 ← 由 Flyway 迁移保证
|
||
|
||
用法
|
||
----
|
||
python scripts/e2e-m0.py
|
||
python scripts/e2e-m0.py --base http://localhost:8080
|
||
|
||
退出码：0 = 全部通过；1 = 有失败项（可直接用于 CI）。
|
||
|
||
依赖：cryptography；且需要能 `docker exec xt-pg`（**清理本脚本造出的测试租户与测试组织**）。
|
||
|
||
⚠️ 关于"跑完要清干净"（2026-10 补）
|
||
本脚本以前**完全不清理**：每跑一次就留下一个 `e2e######` 租户（里面有一个叫 `admin`
|
||
的管理员）+ demo 租户下的 3 个测试组织。跑两轮之后，库里同一时刻就存在两个
|
||
`admin` —— 而"账号跨单位重名"这条保护（40012）一旦被别的东西当作前提，
|
||
清理残留这一个动作就会把那个脚本**静默反转**（见 `e2e-m1-tenant-domain.py` 文件头）。
|
||
所以：跑完（含中途崩溃）必须把自己造的东西删干净，且**只删自己建的**。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import base64
|
||
import json
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
import urllib.error
|
||
import urllib.request
|
||
|
||
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_TENANT_IDS: list[int] = []
|
||
CREATED_ORG_IDS: list[int] = []
|
||
|
||
|
||
# ── HTTP ──────────────────────────────────────────────
|
||
|
||
def http(method: str, url: str, body=None, token: str | None = None,
|
||
timeout: int = 30) -> tuple[int, dict]:
|
||
data = json.dumps(body, ensure_ascii=False).encode("utf-8") if body is not None else None
|
||
headers = {"Content-Type": "application/json; charset=utf-8"}
|
||
if token:
|
||
headers["Authorization"] = f"Bearer {token}"
|
||
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, {"code": -1, "msg": raw[:200]}
|
||
|
||
|
||
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("utf-8"), padding.PKCS1v15())).decode()
|
||
|
||
|
||
def login(base: str, username: str, password: str, tenant_code: str | None = None) -> dict:
|
||
payload = {"username": username, "password": rsa_encrypt(password)}
|
||
if tenant_code:
|
||
payload["tenantCode"] = tenant_code
|
||
try:
|
||
_, body = http("POST", f"{base}/api/v1/auth/login", payload)
|
||
except Exception as e: # 网络层失败也要报成一条用例，而不是崩掉
|
||
return {"code": -1, "msg": f"请求失败: {e}"}
|
||
return body or {"code": -1, "msg": "空响应"}
|
||
|
||
|
||
# ── 断言 ──────────────────────────────────────────────
|
||
|
||
def data_of(body: dict) -> dict:
|
||
"""安全取 data：响应里 data 可能是 null。"""
|
||
if not isinstance(body, dict):
|
||
return {}
|
||
d = body.get("data")
|
||
return d if isinstance(d, dict) else {}
|
||
|
||
|
||
def list_of(body: dict) -> list:
|
||
d = body.get("data")
|
||
if isinstance(d, list):
|
||
return d
|
||
if isinstance(d, dict) and isinstance(d.get("list"), list):
|
||
return d["list"]
|
||
return []
|
||
|
||
|
||
def check(name: str, cond: bool, detail: str = "") -> bool:
|
||
if cond:
|
||
PASSED.append(name)
|
||
print(f" [PASS] {name}")
|
||
else:
|
||
FAILED.append((name, detail))
|
||
print(f" [FAIL] {name} {detail}")
|
||
return cond
|
||
|
||
|
||
def expect_code(name: str, body: dict, code: int) -> bool:
|
||
actual = body.get("code") if isinstance(body, dict) else None
|
||
return check(name, actual == code,
|
||
f"期望 code={code}，实际 code={actual} msg={body.get('msg') if isinstance(body, dict) else body}")
|
||
|
||
|
||
def finish() -> int:
|
||
print("\n" + "=" * 66)
|
||
total = len(PASSED) + len(FAILED)
|
||
if FAILED:
|
||
print(f" 结果: {len(PASSED)}/{total} 通过 —— 有失败项")
|
||
print("\n 失败项：")
|
||
for name, detail in FAILED:
|
||
print(f" - {name}")
|
||
if detail:
|
||
print(f" {detail}")
|
||
else:
|
||
print(f" 结果: {len(PASSED)}/{total} 全部通过")
|
||
print("=" * 66)
|
||
return 0 if not FAILED else 1
|
||
|
||
|
||
# ── 清理 ──────────────────────────────────────────────
|
||
|
||
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: # noqa: BLE001
|
||
return False, str(e)
|
||
|
||
|
||
_TENANT_TABLES: list[str] | None = None
|
||
|
||
|
||
def tables_with_tenant_id() -> list[str]:
|
||
"""
|
||
带 `tenant_id` 的表 —— **动态枚举**，不手抄表名。
|
||
|
||
手抄的清单在"新增了一张业务表"那天就开始漏，而漏掉的那张表恰好很难被注意到
|
||
（租户没了、行还在）。`dev-clean-e2e.py` 是同一口径。
|
||
"""
|
||
global _TENANT_TABLES
|
||
if _TENANT_TABLES is None:
|
||
_, out = sql("SELECT table_name FROM information_schema.columns "
|
||
"WHERE table_schema='public' AND column_name='tenant_id'")
|
||
_TENANT_TABLES = [t.strip() for t in out.splitlines() if t.strip()]
|
||
return _TENANT_TABLES
|
||
|
||
|
||
def cleanup_all() -> None:
|
||
"""
|
||
删掉本脚本造出来的数据。**幂等**，可以重复调用。
|
||
|
||
⚠️ 只删**自己建的**（按本次跑出来的 id）：这里刻意不做
|
||
`tenant_code LIKE 'e2e%'` 那种"顺手全清" —— 那是 `dev-clean-e2e.py` 的活。
|
||
测试脚本越界删别人建的东西，就会让"跑一个脚本把另一个脚本的在途数据删了"
|
||
这种事变得可能。
|
||
"""
|
||
for oid in list(CREATED_ORG_IDS):
|
||
sql(f"DELETE FROM sys_org WHERE id = {oid}")
|
||
CREATED_ORG_IDS.remove(oid)
|
||
|
||
for tid in list(CREATED_TENANT_IDS):
|
||
for table in tables_with_tenant_id():
|
||
sql(f"DELETE FROM {table} WHERE tenant_id = {tid}")
|
||
sql(f"DELETE FROM sys_tenant WHERE id = {tid}")
|
||
CREATED_TENANT_IDS.remove(tid)
|
||
|
||
|
||
# ── 用例 ──────────────────────────────────────────────
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--base", default=os.environ.get("XT_BASE", "http://localhost:8080"))
|
||
args = ap.parse_args()
|
||
base = args.base.rstrip("/")
|
||
stamp = str(int(time.time()))[-6:]
|
||
new_code = f"e2e{stamp}"
|
||
|
||
if not os.path.exists(PUB_KEY):
|
||
sys.exit(f"找不到 RSA 公钥：{PUB_KEY}")
|
||
|
||
print("=" * 66)
|
||
print(" M0 验收测试")
|
||
print(f" 目标: {base}")
|
||
print("=" * 66)
|
||
|
||
# ── A. 认证 ───────────────────────────────────────
|
||
print("\n[A] 认证与鉴权")
|
||
|
||
body = login(base, "platform", "Platform@123")
|
||
if not expect_code("A1 平台运营登录成功", body, 0):
|
||
return finish()
|
||
platform_token = data_of(body).get("token")
|
||
|
||
# ⚠️ admin 这个账号名在多个租户下都存在（E2E 每次会给新租户也建一个 admin），
|
||
# 所以必须带 tenantCode —— 这正是服务端「拒绝猜测、要求指明单位」的设计。
|
||
body = login(base, "admin", "Admin@123", tenant_code="demo")
|
||
expect_code("A2 学校管理员登录成功（带租户码）", body, 0)
|
||
demo_token = data_of(body).get("token")
|
||
demo_menus = set(data_of(body).get("menus") or [])
|
||
|
||
# 覆盖「账号跨租户重名」这条保护。
|
||
# 首次运行时 admin 只存在于 demo 一个租户（此时正常登录是对的）；
|
||
# 反复运行后会出现多个同名 admin（此时必须要求指明单位）。
|
||
# 两种都算通过，唯一不允许的是「随便挑一个」。
|
||
body = login(base, "admin", "Admin@123")
|
||
code_n = body.get("code")
|
||
msg_n = str(body.get("msg", ""))
|
||
check("A2b 同名账号跨租户时：要么唯一可直接登录，要么要求指明单位——绝不猜",
|
||
code_n == 0 or (code_n == 40012 and "多个单位" in msg_n),
|
||
f"code={code_n} msg={msg_n}")
|
||
|
||
expect_code("A3 密码错误返回 40012", login(base, "admin", "WrongPassword!1", "demo"), 40012)
|
||
|
||
# 用每次唯一的账号名，避免上一轮的失败计数触发验证码要求
|
||
ghost = f"ghost_{stamp}"
|
||
expect_code("A4 账号不存在也返回 40012（不泄露账号是否存在）",
|
||
login(base, ghost, "Whatever@123"), 40012)
|
||
|
||
_, b = http("GET", f"{base}/api/v1/auth/me")
|
||
expect_code("A5 未带 token 访问受保护接口返回 40101", b, 40101)
|
||
|
||
_, b = http("GET", f"{base}/api/v1/auth/me", token=demo_token)
|
||
expect_code("A6 /auth/me 正常返回", b, 0)
|
||
check("A7 返回角色码 SCHOOL_ADMIN", data_of(b).get("roleCodes") == ["SCHOOL_ADMIN"],
|
||
str(data_of(b).get("roleCodes")))
|
||
check("A8 学校管理员菜单不含 tenant（无平台权限）", "tenant" not in demo_menus,
|
||
f"menus={sorted(demo_menus)}")
|
||
|
||
body = login(base, "platform", "Platform@123")
|
||
plat_menus = set(data_of(body).get("menus") or [])
|
||
check("A9 平台运营菜单含 tenant", "tenant" in plat_menus, f"menus={sorted(plat_menus)}")
|
||
check("A10 两种角色菜单不同（权限渲染生效）", demo_menus != plat_menus,
|
||
f"demo={sorted(demo_menus)} platform={sorted(plat_menus)}")
|
||
|
||
# ── B. 开通租户 ───────────────────────────────────
|
||
print("\n[B] 开通租户")
|
||
|
||
_, b = http("GET", f"{base}/api/v1/tenants", token=demo_token)
|
||
expect_code("B1 学校管理员无租户管理权限（40301）", b, 40301)
|
||
|
||
_, b = http("POST", f"{base}/api/v1/tenants", {
|
||
"tenantName": f"E2E 测试学校 {stamp}",
|
||
"tenantCode": new_code,
|
||
"tenantType": 1,
|
||
"contactName": "测试联系人",
|
||
"contactPhone": "13900139000",
|
||
"expireTime": "2027-12-31",
|
||
"adminUsername": "admin",
|
||
"adminRealName": "E2E 管理员",
|
||
"adminPhone": "13900139001",
|
||
}, token=platform_token)
|
||
expect_code("B2 平台运营开通租户成功", b, 0)
|
||
created = data_of(b)
|
||
new_tenant_id = created.get("id")
|
||
if isinstance(new_tenant_id, int):
|
||
CREATED_TENANT_IDS.append(new_tenant_id) # 登记，收尾与崩溃路径都要清它
|
||
new_pwd = created.get("adminInitialPassword")
|
||
check("B3 返回初始密码（仅此一次）", bool(new_pwd), json.dumps(created, ensure_ascii=False)[:200])
|
||
check("B4 返回小程序码参数", created.get("miniProgramScene") == f"t{new_tenant_id}",
|
||
str(created.get("miniProgramScene")))
|
||
if new_tenant_id is None:
|
||
return finish()
|
||
|
||
body = login(base, "admin", new_pwd or "x", tenant_code=new_code)
|
||
expect_code("B5 新租户管理员用初始密码可登录（证明租户作用域切换生效）", body, 0)
|
||
new_token = data_of(body).get("token")
|
||
check("B6 新管理员归属租户正确",
|
||
data_of(body).get("tenant", {}).get("tenantCode") == new_code,
|
||
str(data_of(body).get("tenant")))
|
||
|
||
_, b = http("GET", f"{base}/api/v1/tenants?pageNum=1&pageSize=50", token=platform_token)
|
||
expect_code("B7 租户列表可查", b, 0)
|
||
ids = [t.get("id") for t in list_of(b)]
|
||
check("B8 列表含新建租户", new_tenant_id in ids, f"ids={ids}")
|
||
|
||
_, b = http("POST", f"{base}/api/v1/tenants", {
|
||
"tenantName": "重复编码", "tenantCode": new_code, "tenantType": 1,
|
||
"contactName": "x", "contactPhone": "13900139000", "expireTime": "2027-12-31",
|
||
"adminUsername": "admin", "adminRealName": "x", "adminPhone": "13900139002",
|
||
}, token=platform_token)
|
||
expect_code("B9 租户编码重复被拒（40001）", b, 40001)
|
||
|
||
# ── C. 组织树 ─────────────────────────────────────
|
||
print("\n[C] 组织架构")
|
||
|
||
def mk_org(token, parent_id, name, otype, code):
|
||
_, r = http("POST", f"{base}/api/v1/orgs", {
|
||
"parentId": parent_id, "orgName": name, "orgType": otype, "orgCode": code,
|
||
}, token=token)
|
||
return r
|
||
|
||
b = mk_org(demo_token, None, f"E2E 示范学校 {stamp}", 1, f"ROOT{stamp}")
|
||
expect_code("C1 建学校节点", b, 0)
|
||
school_id = b.get("data")
|
||
|
||
b = mk_org(demo_token, school_id, f"E2E 机电系 {stamp}", 2, f"DEPT{stamp}")
|
||
expect_code("C2 建院系节点", b, 0)
|
||
dept_id = b.get("data")
|
||
|
||
b = mk_org(demo_token, dept_id, f"E2E 24数控1班 {stamp}", 3, f"CLS{stamp}")
|
||
expect_code("C3 建班级节点", b, 0)
|
||
class_id = b.get("data")
|
||
# 登记（C1/C2/C3 若失败，对应的 data 是 None，不能登记成 0/None）
|
||
for oid in (school_id, dept_id, class_id):
|
||
if isinstance(oid, int):
|
||
CREATED_ORG_IDS.append(oid)
|
||
|
||
expect_code("C4 同级重复组织编码被拒（40001）",
|
||
mk_org(demo_token, school_id, "重复编码", 2, f"DEPT{stamp}"), 40001)
|
||
expect_code("C5 组织类型层级颠倒被拒（40001）",
|
||
mk_org(demo_token, class_id, "层级颠倒", 1, f"BAD{stamp}"), 40001)
|
||
|
||
_, b = http("GET", f"{base}/api/v1/orgs/tree?withCount=true", token=demo_token)
|
||
expect_code("C6 组织树查询", b, 0)
|
||
tree = list_of(b)
|
||
|
||
def find(nodes, target):
|
||
for n in nodes:
|
||
if n.get("id") == target:
|
||
return n
|
||
hit = find(n.get("children") or [], target)
|
||
if hit:
|
||
return hit
|
||
return None
|
||
|
||
check("C7 三层结构正确（班级挂在院系下）", find(tree, class_id) is not None,
|
||
f"树中未见 id={class_id}")
|
||
root = find(tree, school_id)
|
||
check("C8 学校 → 院系 层级正确",
|
||
root is not None and dept_id in [c.get("id") for c in (root.get("children") or [])],
|
||
f"children={[c.get('id') for c in (root.get('children') or [])] if root else None}")
|
||
|
||
_, b = http("DELETE", f"{base}/api/v1/orgs/{dept_id}", token=demo_token)
|
||
expect_code("C9 有下级的组织不可删除（40015）", b, 40015)
|
||
|
||
# ── D. 跨租户隔离 ─────────────────────────────────
|
||
print("\n[D] 跨租户隔离")
|
||
|
||
_, b = http("GET", f"{base}/api/v1/orgs/tree", token=new_token)
|
||
expect_code("D1 新租户可查自己的组织树", b, 0)
|
||
dumped = json.dumps(list_of(b), ensure_ascii=False)
|
||
check("D2 新租户看不到 demo 租户的组织（租户条件自动注入）",
|
||
str(school_id) not in dumped, f"新租户树里出现了 demo 的组织 id={school_id}")
|
||
|
||
_, b = http("GET", f"{base}/api/v1/tenants", token=new_token)
|
||
expect_code("D3 新租户管理员调租户列表被拒（40301）", b, 40301)
|
||
|
||
# ── E. 异常与安全 ─────────────────────────────────
|
||
print("\n[E] 异常与安全")
|
||
|
||
_, b = http("POST", f"{base}/api/v1/auth/login", {"username": "admin", "password": "not-encrypted"})
|
||
check("E1 密码未加密（非法密文）也被拒，不泄露内部信息",
|
||
b.get("code") in (40012, 40001), f"code={b.get('code')} msg={b.get('msg')}")
|
||
|
||
_, b = http("GET", f"{base}/api/v1/orgs/tree", token="garbage.token.value")
|
||
expect_code("E2 伪造 token 返回 40101", b, 40101)
|
||
|
||
_, b = http("GET", f"{base}/api/v1/orgs/abc", token=demo_token)
|
||
check("E3 路径参数类型错误返回 40001", b.get("code") == 40001, f"code={b.get('code')}")
|
||
|
||
_, b = http("POST", f"{base}/api/v1/orgs", {"orgName": "", "orgType": 99, "orgCode": ""},
|
||
token=demo_token)
|
||
check("E4 参数校验失败返回 40001 且带字段级信息", b.get("code") == 40001,
|
||
f"code={b.get('code')}")
|
||
|
||
# ── F. 清理 ───────────────────────────────────────
|
||
print("\n[F] 清理本次造出的测试数据")
|
||
n_tenant, n_org = len(CREATED_TENANT_IDS), len(CREATED_ORG_IDS)
|
||
cleanup_all()
|
||
print(f" 已清理 {n_tenant} 个测试租户（连它的角色与账号）"
|
||
f"、{n_org} 个挂在 demo 下的测试组织")
|
||
check("F1 测试租户与测试组织都删掉了", not CREATED_TENANT_IDS and not CREATED_ORG_IDS,
|
||
f"tenant={CREATED_TENANT_IDS} org={CREATED_ORG_IDS}")
|
||
|
||
return finish()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# 崩溃也要清：否则每次失败的运行都往库里丢一个 e2e 租户和一个叫 admin 的账号
|
||
# （跨租户同名账号就是这么攒出来的，见文件头 ⚠️）
|
||
try:
|
||
_code = main()
|
||
except BaseException:
|
||
print("\n ⚠ 脚本异常退出，正在清理已造出的测试数据…")
|
||
cleanup_all()
|
||
raise
|
||
raise SystemExit(_code)
|