xiaotutongxue/scripts/e2e-m0.py

#!/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
"""
from __future__ import annotations

import argparse
import base64
import json
import os
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")

PASSED: list[str] = []
FAILED: list[tuple[str, str]] = []


# ── 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 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")
    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")

    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')}")

    return finish()


if __name__ == "__main__":
    raise SystemExit(main())