Pine
26fa546f67
- 七端口 RBAC、select-identity、JWT、审计 - FastAPI + SQLAlchemy + SQLite，/auth /opc /admin /agents 等路由
139 lines
5.3 KiB
Python
139 lines
5.3 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""运营端（operator 端口）子角色权限与账号管理测试。"""
|
||
from __future__ import annotations
|
||
|
||
from fastapi.testclient import TestClient
|
||
|
||
|
||
def login(client: TestClient, username: str, password: str = "123456") -> str:
|
||
res = client.post("/auth/login", json={"username": username, "password": password})
|
||
assert res.status_code == 200, res.text
|
||
return res.json()["token"]
|
||
|
||
|
||
def auth(token: str) -> dict:
|
||
return {"Authorization": f"Bearer {token}"}
|
||
|
||
|
||
def user_id_by_username(client: TestClient, token: str, username: str) -> str:
|
||
users = client.get("/admin/users", headers=auth(token)).json()
|
||
return next(u["id"] for u in users if u["username"] == username)
|
||
|
||
|
||
def select_identity(client: TestClient, token: str, port: str) -> str:
|
||
"""把中性令牌切换到指定端口身份，返回绑定身份的令牌。"""
|
||
body = client.post("/auth/login", json={"username": "pine", "password": "123456"}).json()
|
||
ident = next(i for i in body["identities"] if i["port"] == port)["id"]
|
||
sel = client.post(
|
||
"/auth/select-identity", headers=auth(token), json={"identity_id": ident}
|
||
).json()
|
||
return sel["token"]
|
||
|
||
|
||
def test_operator_admin_cannot_escalate_to_super_admin(client):
|
||
# 管理员不可创建/提升 超管/管理员（特权角色）
|
||
admin_t = login(client, "op_admin")
|
||
res = client.post(
|
||
"/admin/users",
|
||
headers=auth(admin_t),
|
||
json={"username": "13800001111", "password": "123456",
|
||
"role": "operator", "sub_role": "op_super_admin"},
|
||
)
|
||
assert res.status_code == 403
|
||
# 管理员可创建普通运营账号（客服）
|
||
ok = client.post(
|
||
"/admin/users",
|
||
headers=auth(admin_t),
|
||
json={"username": "13800002222", "password": "123456",
|
||
"role": "operator", "sub_role": "op_customer_service"},
|
||
)
|
||
assert ok.status_code == 200
|
||
# 管理员把某账号提升为超管 → 403
|
||
target_id = user_id_by_username(client, admin_t, "13800002222")
|
||
up = client.post(
|
||
f"/admin/users/{target_id}/role",
|
||
headers=auth(admin_t),
|
||
json={"role": "operator", "sub_role": "op_super_admin"},
|
||
)
|
||
assert up.status_code == 403
|
||
|
||
|
||
def test_operator_cs_cannot_change_config(client):
|
||
# 客服可建任务/服务商，但不可改系统配置
|
||
cs_t = login(client, "op_cs")
|
||
assert (
|
||
client.put("/admin/config/platform.fee.task",
|
||
headers=auth(cs_t), json={"value": "0.01"}).status_code == 403
|
||
)
|
||
assert client.post("/admin/tasks", headers=auth(cs_t),
|
||
json={"title": "客服建任务"}).status_code == 200
|
||
|
||
|
||
def test_operator_finance_cannot_manage_tasks(client):
|
||
# 财务可看结算，但不可建任务
|
||
fin_t = login(client, "op_fin")
|
||
assert (
|
||
client.post("/admin/tasks", headers=auth(fin_t),
|
||
json={"title": "财务建任务"}).status_code == 403
|
||
)
|
||
assert client.get("/operator/settlements", headers=auth(fin_t)).status_code == 200
|
||
|
||
|
||
def test_operator_invalid_role_combo_rejected(client):
|
||
# 非法/跨界角色组合被白名单拒绝
|
||
admin_t = login(client, "op_admin")
|
||
res = client.post(
|
||
"/admin/users",
|
||
headers=auth(admin_t),
|
||
json={"username": "13800003333", "password": "123456",
|
||
"role": "operator", "sub_role": "gov_province"},
|
||
)
|
||
assert res.status_code == 400
|
||
|
||
|
||
def test_operator_grant_perm_only_super_admin(client):
|
||
# 配置角色权限仅超管可用
|
||
admin_t = login(client, "op_admin")
|
||
assert (
|
||
client.post("/admin/roles/operator%7Cop_admin/permissions",
|
||
headers=auth(admin_t), json={"permissions": []}).status_code == 403
|
||
)
|
||
# pine 以超管身份可配置
|
||
neutral = login(client, "pine")
|
||
sa_t = select_identity(client, neutral, "operator")
|
||
ok = client.post(
|
||
"/admin/roles/operator%7Cop_analyst/permissions",
|
||
headers=auth(sa_t), json={"permissions": ["menu:audit", "action:data.export"]},
|
||
)
|
||
assert ok.status_code == 200
|
||
assert "menu:audit" in ok.json()["permissions"]
|
||
|
||
|
||
def test_operator_role_change_takes_effect(client):
|
||
# 改角色后：令牌失效 + 身份同步，重新登录体现新角色权限
|
||
admin_t = login(client, "op_admin")
|
||
created = client.post(
|
||
"/admin/users",
|
||
headers=auth(admin_t),
|
||
json={"username": "13800004444", "password": "123456",
|
||
"role": "operator", "sub_role": "op_customer_service"},
|
||
).json()
|
||
# 该账号现可建任务（客服有 task.manage）
|
||
cs_t = login(client, "13800004444")
|
||
assert client.post("/admin/tasks", headers=auth(cs_t),
|
||
json={"title": "t"}).status_code == 200
|
||
# 管理员将其改为分析师（无 task.manage）
|
||
target_id = user_id_by_username(client, admin_t, "13800004444")
|
||
up = client.post(
|
||
f"/admin/users/{target_id}/role",
|
||
headers=auth(admin_t),
|
||
json={"role": "operator", "sub_role": "op_analyst"},
|
||
)
|
||
assert up.status_code == 200
|
||
# 旧令牌已失效
|
||
assert client.get("/admin/users", headers=auth(cs_t)).status_code == 401
|
||
# 新登录为分析师：不可建任务
|
||
an_t = login(client, "13800004444")
|
||
assert client.post("/admin/tasks", headers=auth(an_t),
|
||
json={"title": "t2"}).status_code == 403
|