214 lines
8.9 KiB
Python
214 lines
8.9 KiB
Python
# -*- 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 unit_test_guard # noqa: F401，单独运行也禁止联网
|
||
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()
|