Pine
26fa546f67
- 七端口 RBAC、select-identity、JWT、审计 - FastAPI + SQLAlchemy + SQLite，/auth /opc /admin /agents 等路由
118 lines
4.2 KiB
Python
118 lines
4.2 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""RBAC 端到端测试：角色 / 权限 / 数据范围 / 审计。"""
|
|
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 test_operator_can_list_users_enterprise_cannot(client):
|
|
op = login(client, "pine") # operator / op_super_admin
|
|
ent = login(client, "ent01") # enterprise
|
|
assert client.get("/admin/users", headers=auth(op)).status_code == 200
|
|
assert client.get("/admin/users", headers=auth(ent)).status_code == 403
|
|
|
|
|
|
def test_assign_role_requires_permission(client):
|
|
# op_analyst 有 operator 角色但缺 action:user.assign_role
|
|
analyst = login(client, "op_analyst")
|
|
res = client.post(
|
|
"/admin/users/u_ent_01/role",
|
|
headers=auth(analyst),
|
|
json={"role": "carrier"},
|
|
)
|
|
assert res.status_code == 403
|
|
|
|
|
|
def test_super_admin_can_assign_role(client):
|
|
op = login(client, "pine")
|
|
res = client.post(
|
|
"/admin/users/u_ent_01/role",
|
|
headers=auth(op),
|
|
json={"role": "carrier", "org_id": "o_car_001", "region_id": "r_dist_wh"},
|
|
)
|
|
assert res.status_code == 200, res.text
|
|
body = res.json()
|
|
assert body["user"]["role"] == "carrier"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 数据范围层级（上级可看下级，下级不可看上级）
|
|
# ---------------------------------------------------------------------------
|
|
def test_government_province_sees_all_regions(client):
|
|
prov = login(client, "gov_prov")
|
|
regions = client.get("/government/regions", headers=auth(prov)).json()
|
|
ids = {r["id"] for r in regions}
|
|
assert "r_prov_yn" in ids and "r_dist_wh" in ids and "r_dist_ql" in ids
|
|
|
|
|
|
def test_government_district_sees_only_own_region(client):
|
|
dist = login(client, "gov_dist") # r_dist_wh
|
|
regions = client.get("/government/regions", headers=auth(dist)).json()
|
|
ids = {r["id"] for r in regions}
|
|
assert ids == {"r_dist_wh"}
|
|
|
|
|
|
def test_district_gov_cannot_access_out_of_scope_enterprise(client):
|
|
dist = login(client, "gov_dist") # 只覆盖 r_dist_wh
|
|
# o_ent_prov 在 r_prov_yn，超出区县范围
|
|
res = client.get("/government/enterprises/o_ent_prov", headers=auth(dist))
|
|
assert res.status_code == 403
|
|
|
|
|
|
def test_province_gov_can_access_any_enterprise(client):
|
|
prov = login(client, "gov_prov")
|
|
res = client.get("/government/enterprises/o_ent_prov", headers=auth(prov))
|
|
assert res.status_code == 200
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 禁用用户 / 审计
|
|
# ---------------------------------------------------------------------------
|
|
def test_disabled_user_token_rejected(client):
|
|
op = login(client, "pine")
|
|
ent = login(client, "ent01") # 禁用前取得的 token
|
|
# 禁用 ent01
|
|
res = client.post(
|
|
"/admin/users/u_ent_01/status",
|
|
headers=auth(op),
|
|
json={"status": "disabled"},
|
|
)
|
|
assert res.status_code == 200
|
|
# 已登录的 ent01 token 立即失效
|
|
assert client.get("/auth/verify", headers=auth(ent)).status_code == 401
|
|
# 重新登录被拒（禁用）
|
|
assert (
|
|
client.post(
|
|
"/auth/login", json={"username": "ent01", "password": "123456"},
|
|
).status_code
|
|
== 403
|
|
)
|
|
|
|
|
|
def test_role_change_and_data_view_write_audit(client):
|
|
op = login(client, "pine")
|
|
prov = login(client, "gov_prov")
|
|
client.post(
|
|
"/admin/users/u_ent_01/role",
|
|
headers=auth(op),
|
|
json={"role": "provider"},
|
|
)
|
|
client.get("/government/enterprises", headers=auth(prov))
|
|
logs = client.get("/admin/audit-logs", headers=auth(op)).json()
|
|
actions = {l["action"] for l in logs}
|
|
assert "role.assign" in actions
|
|
assert "data.view" in actions
|