tradingSystem/scripts/test_batch18_units.py

# -*- coding: utf-8 -*-
"""
第十八批: 登录与权限 (系统管理员 / 交易员, 可同时持有) 的纯逻辑 —— 不联网、不连库
=====================================================================
只测 app/web/auth.py 里的纯函数, 不导入 app.web.main。覆盖四件事:
  1. 角色判定 roles_from_permissions: 返回**角色集合**, 一个人可同时是管理员和交易员;
  2. 会话票 make_session / verify_session: roles 列表原样签发-校验, 篡改/过期/换密钥/空票一律拒;
  3. bshop 返回解析 parse_bshop_login: 喂真实样例抽字段, code!=200 判失败;
  4. 接口鉴权 authz_decision + is_public_path: 只有运维维护类接口归管理员, 其余登录人都放行。
运行: python scripts/test_batch18_units.py
"""
import os
import sys
import traceback

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from app.web import auth as A                                                   # noqa: E402

RESULTS = []


def case(name):
    def deco(fn):
        RESULTS.append((name, fn))
        return fn
    return deco


SAMPLE = {
    "code": 200, "msg": "登录成功",
    "data": {
        "access_token": "SMxzzyfoIJogItsmiR9Qv0NdnFAoKE-F5WdJPDhzZOA",
        "token_type": "bearer", "expires_in": 604800,
        "user": {"phone": "18795373439", "username": "18795373439", "wechat_bound": True},
        "permissions": [
            {"system_key": "band", "menu_key": "-", "system_name": "波段系统后端", "menu_name": "波段系统后端"},
            {"system_key": "band", "menu_key": "bkld", "system_name": "波段系统后端", "menu_name": "板块轮动"},
            {"system_key": "band", "menu_key": "yhyqx", "system_name": "波段系统后端", "menu_name": "用户与权限"},
            {"system_key": "band_web", "menu_key": "traded", "system_name": "波段系统网页端", "menu_name": "波段监控"},
        ],
        "orgs": [{"id": 277097889622, "org_name": "高姐"},
                 {"id": 489281497140, "org_name": "天盟资本"}],
    },
}


# ---------------------------------------------------------------- 一、角色判定（可多角色）
@case("角色: 管理员标记 + 交易员开放 => 同时是管理员和交易员")
def _roles_both_open():
    r = A.roles_from_permissions(["band:bkld", "pms:admin"], "pms:admin", "")
    assert r == [A.ROLE_ADMIN, A.ROLE_TRADER], r


@case("角色: 只有普通权限 (交易员开放) => 只交易员")
def _roles_trader_only():
    r = A.roles_from_permissions(["band:bkld"], "pms:admin", "")
    assert r == [A.ROLE_TRADER], r


@case("角色: 配了交易员准入, 有管理员无交易员权限 => 只管理员")
def _roles_admin_only():
    r = A.roles_from_permissions(["pms:admin"], "pms:admin", "band:ccgl")
    assert r == [A.ROLE_ADMIN], r


@case("角色: 配了交易员准入, 两者都占 => 同时两个角色 (用户场景)")
def _roles_both_gated():
    r = A.roles_from_permissions(["pms:admin", "band:ccgl"], "pms:admin", "band:ccgl")
    assert r == [A.ROLE_ADMIN, A.ROLE_TRADER], r


@case("角色: 配了交易员准入却一条不占 => 空 (无权进入)")
def _roles_denied():
    r = A.roles_from_permissions(["band:x"], "pms:admin", "band:ccgl")
    assert r == [], r


@case("角色: is_admin 辅助判定")
def _is_admin_helper():
    assert A.is_admin([A.ROLE_ADMIN, A.ROLE_TRADER]) is True
    assert A.is_admin([A.ROLE_TRADER]) is False
    assert A.is_admin([]) is False


# ---------------------------------------------------------------- 二、会话票
@case("会话: roles 列表签发-校验往返, 字段齐全")
def _sess_roundtrip():
    tok = A.make_session("18795373439", "老王", [A.ROLE_ADMIN, A.ROLE_TRADER], "s3cr3t", 12, now=1000.0)
    p = A.verify_session(tok, "s3cr3t", now=1001.0)
    assert p and p["phone"] == "18795373439" and p["username"] == "老王", p
    assert p["roles"] == [A.ROLE_ADMIN, A.ROLE_TRADER] and p["exp"] == 1000 + 12 * 3600, p


@case("会话: 篡改正文或签名 => None")
def _sess_tampered():
    tok = A.make_session("186", "u", [A.ROLE_TRADER], "s3cr3t", 12, now=1000.0)
    b, sig = tok.split(".", 1)
    assert A.verify_session(b + "x." + sig, "s3cr3t", now=1001.0) is None
    assert A.verify_session(b + "." + sig + "x", "s3cr3t", now=1001.0) is None


@case("会话: 过期 / 换密钥 / 空密钥 / 空票 一律拒")
def _sess_reject():
    tok = A.make_session("186", "u", [A.ROLE_TRADER], "s3cr3t", 1, now=1000.0)
    assert A.verify_session(tok, "s3cr3t", now=1000.0 + 3600 + 1) is None, "过期没拒"
    assert A.verify_session(tok, "other", now=1001.0) is None, "换密钥没拒"
    assert A.verify_session(tok, "", now=1001.0) is None, "空密钥没拒"
    assert A.verify_session("", "s3cr3t", now=1001.0) is None, "空票没拒"


# ---------------------------------------------------------------- 三、bshop 解析
@case("解析: 真实样例 => 手机号/用户名/权限/组织")
def _parse_ok():
    r = A.parse_bshop_login(SAMPLE)
    assert r["ok"] is True, r
    assert r["phone"] == "18795373439" and r["username"] == "18795373439", r
    assert "band:bkld" in r["permissions"] and "band:-" in r["permissions"], r["permissions"]
    assert "band:yhyqx" in r["permissions"], r["permissions"]
    assert len(r["orgs"]) == 2 and r["orgs"][0]["org_name"] == "高姐", r["orgs"]


@case("解析: 样例接角色 —— 默认标记只交易员, band:yhyqx 标记则管理员+交易员")
def _parse_then_roles():
    r = A.parse_bshop_login(SAMPLE)
    assert A.roles_from_permissions(r["permissions"], "pms:admin", "") == [A.ROLE_TRADER]
    assert A.roles_from_permissions(r["permissions"], "band:yhyqx", "") == [A.ROLE_ADMIN, A.ROLE_TRADER]


@case("解析: code!=200 / 缺手机号 / 非字典 => 失败")
def _parse_fail():
    assert A.parse_bshop_login({"code": 401, "msg": "密码错误"})["ok"] is False
    assert A.parse_bshop_login({"code": 200, "data": {"user": {}}})["ok"] is False
    assert A.parse_bshop_login("nope")["ok"] is False


# ---------------------------------------------------------------- 四、接口鉴权
@case("鉴权: 交易员工作台上的写操作都放行 (any)")
def _authz_trader_ok():
    trader_paths = [
        ("POST", "/api/commands"),
        ("POST", "/api/commands/CMD-1/cancel"),
        ("POST", "/api/commands/CMD-1/replan"),
        ("POST", "/api/commands/CMD-1/archive"),
        ("POST", "/api/proposals/PRP-1/decide"),
        ("POST", "/api/proposals/PRP-1/archive"),
        ("POST", "/api/instructions/INS-1/cancel"),   # 用户点名: 撤单归交易员
        ("POST", "/api/macro/adopt"),                 # 用户点名: 大盘冷热采纳归交易员
        ("POST", "/api/params"),                      # 工作台上有参数微调面板
        ("POST", "/api/strategies"),
        ("POST", "/api/strategies/validate"),
        ("POST", "/api/strategies/S1/status"),
        ("POST", "/api/strategies/S1/resume-buy"),
        ("POST", "/api/strategies/S1/archive"),
    ]
    for m, path in trader_paths:
        assert A.authz_decision(m, path) == "any", (m, path, A.authz_decision(m, path))


@case("鉴权: 只有系统维护类接口归管理员")
def _authz_admin_only():
    admin_paths = [
        ("POST", "/api/ops/reconcile"),
        ("POST", "/api/ops/daily-settle"),
        ("POST", "/api/ops/macro-scan"),
        ("POST", "/api/ops/plan-refresh"),
        ("POST", "/api/ws-channel/clear-resync"),
        ("POST", "/api/industry/import"),
        ("POST", "/api/proposals"),            # 人工补录 (无 id 段) = 造数调试口
    ]
    for m, path in admin_paths:
        assert A.authz_decision(m, path) == "admin", (m, path, A.authz_decision(m, path))


@case("鉴权: 运维读接口要管理员, 普通读接口谁都能看")
def _authz_gets():
    for path in ["/api/ops/rebuild-preflight", "/api/ops/rebuild-accept",
                 "/api/ops/downstream-schema"]:
        assert A.authz_decision("GET", path) == "admin", path
    for path in ["/api/overview", "/api/positions", "/api/ledger", "/api/proposals"]:
        assert A.authz_decision("GET", path) == "any", path


@case("公开白名单: 首页/静态/登录接口放行, 业务接口不放行")
def _public():
    for p in ["/", "/health", "/favicon.ico", "/static", "/static/vendor/vue.global.js",
              "/api/auth/login", "/api/auth/logout", "/api/me"]:
        assert A.is_public_path(p) is True, p
    for p in ["/api/params", "/api/overview", "/api/commands"]:
        assert A.is_public_path(p) is False, p


def main():
    passed, failed = 0, []
    for name, fn in RESULTS:
        try:
            fn()
            passed += 1
            print(f"  ✓ {name}")
        except Exception as e:
            failed.append((name, e))
            print(f"  ✗ {name}: {type(e).__name__}: {e}")
            traceback.print_exc()
    print()
    if failed:
        print(f"FAILED {len(failed)}/{len(RESULTS)}")
        sys.exit(1)
    print(f"ALL PASS ({passed} cases)")


if __name__ == "__main__":
    main()