#!/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)