Pine
26fa546f67
- 七端口 RBAC、select-identity、JWT、审计 - FastAPI + SQLAlchemy + SQLite，/auth /opc /admin /agents 等路由
74 lines
2.7 KiB
Python
74 lines
2.7 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""账号↔端口多身份：登录返回身份列表、选择身份签发新令牌、按身份访问。"""
|
|
from __future__ import annotations
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
|
|
def login(client: TestClient, username: str, password: str = "123456") -> dict:
|
|
res = client.post("/auth/login", json={"username": username, "password": password})
|
|
assert res.status_code == 200, res.text
|
|
return res.json()
|
|
|
|
|
|
def auth(token: str) -> dict:
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
def test_single_identity_auto_login(client):
|
|
"""单身份账号（opc01）登录即绑定该身份，可直接访问对应端口。"""
|
|
body = login(client, "opc01")
|
|
assert len(body["identities"]) == 1
|
|
assert body["identities"][0]["port"] == "opc"
|
|
assert body["role"] == "opc_member"
|
|
assert body.get("identity_id") == body["identities"][0]["id"]
|
|
|
|
res = client.get("/opc/dashboard", headers=auth(body["token"]))
|
|
assert res.status_code == 200, res.text
|
|
|
|
|
|
def test_multi_identity_login_returns_all_and_neutral(client):
|
|
"""多身份账号（ent01：企业 + OPC）登录返回全部身份，令牌为中性（回退单角色）。"""
|
|
body = login(client, "ent01")
|
|
ports = [i["port"] for i in body["identities"]]
|
|
assert "enterprise" in ports and "opc" in ports
|
|
|
|
# 中性令牌按 users.role 回退为 enterprise：访问 OPC 端口应 403
|
|
res = client.get("/opc/dashboard", headers=auth(body["token"]))
|
|
assert res.status_code == 403
|
|
|
|
|
|
def test_select_identity_grants_opc_access(client):
|
|
"""选择 OPC 身份后，新令牌可访问 OPC 端口，且 /me 反映该身份。"""
|
|
body = login(client, "ent01")
|
|
opc_identity = next(i for i in body["identities"] if i["port"] == "opc")
|
|
|
|
res = client.post(
|
|
"/auth/select-identity",
|
|
headers=auth(body["token"]),
|
|
json={"identity_id": opc_identity["id"]},
|
|
)
|
|
assert res.status_code == 200, res.text
|
|
selected = res.json()
|
|
assert selected["role"] == "opc_member"
|
|
assert selected["identity_id"] == opc_identity["id"]
|
|
|
|
dash = client.get("/opc/dashboard", headers=auth(selected["token"]))
|
|
assert dash.status_code == 200, dash.text
|
|
|
|
me = client.get("/auth/me", headers=auth(selected["token"])).json()
|
|
assert me["role"] == "opc_member"
|
|
assert me.get("port") == "opc"
|
|
|
|
|
|
def test_select_identity_rejects_other_users_identity(client):
|
|
"""不能选择不属于当前账号的身份。"""
|
|
body = login(client, "opc01")
|
|
other = "ident_ent01_opc" # 属于 ent01 的身份 id
|
|
res = client.post(
|
|
"/auth/select-identity",
|
|
headers=auth(body["token"]),
|
|
json={"identity_id": other},
|
|
)
|
|
assert res.status_code == 404
|