79 lines
2.4 KiB
Python
79 lines
2.4 KiB
Python
|
"""Wave 0：JWT / RBAC 测试。"""
|
|||
|
|
|
|||
|
|
from app.gateway.jwt_service import decode_token, issue_token
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_issue_and_decode_staff_token():
|
|||
|
|
token, expires = issue_token("STAFF-20001", "staff")
|
|||
|
|
assert expires > 0
|
|||
|
|
payload = decode_token(token)
|
|||
|
|
assert payload["sub"] == "STAFF-20001"
|
|||
|
|
assert "analyst" in payload["roles"]
|
|||
|
|
assert "agent:analyst:chat" in payload["permissions"]
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_login_endpoint(client):
|
|||
|
|
resp = client.post("/api/auth/login", json={"actor_id": "STAFF-20001", "token_type": "staff"})
|
|||
|
|
assert resp.status_code == 200
|
|||
|
|
body = resp.json()
|
|||
|
|
assert body["code"] == 0
|
|||
|
|
assert "access_token" in body["data"]
|
|||
|
|
assert "trace_id" in body
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_chat_requires_auth(client):
|
|||
|
|
resp = client.post(
|
|||
|
|
"/api/chat",
|
|||
|
|
json={"message": "hello"},
|
|||
|
|
headers={"X-Agent-Type": "analyst"},
|
|||
|
|
)
|
|||
|
|
assert resp.status_code == 401
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_chat_analyst_ok(client):
|
|||
|
|
login = client.post("/api/auth/login", json={"actor_id": "STAFF-20001", "token_type": "staff"})
|
|||
|
|
token = login.json()["data"]["access_token"]
|
|||
|
|
resp = client.post(
|
|||
|
|
"/api/chat",
|
|||
|
|
json={"message": "上季度收益率"},
|
|||
|
|
headers={
|
|||
|
|
"Authorization": f"Bearer {token}",
|
|||
|
|
"X-Agent-Type": "analyst",
|
|||
|
|
},
|
|||
|
|
)
|
|||
|
|
assert resp.status_code == 200
|
|||
|
|
body = resp.json()
|
|||
|
|
assert body["code"] == 0
|
|||
|
|
assert "reply" in body["data"]
|
|||
|
|
assert body["data"]["agent_type"] == "analyst"
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_agent_type_mismatch_forbidden(client):
|
|||
|
|
login = client.post("/api/auth/login", json={"actor_id": "STAFF-20001", "token_type": "staff"})
|
|||
|
|
token = login.json()["data"]["access_token"]
|
|||
|
|
resp = client.post(
|
|||
|
|
"/api/chat",
|
|||
|
|
json={"message": "hello"},
|
|||
|
|
headers={
|
|||
|
|
"Authorization": f"Bearer {token}",
|
|||
|
|
"X-Agent-Type": "customer",
|
|||
|
|
},
|
|||
|
|
)
|
|||
|
|
assert resp.status_code == 403
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_advisor_not_assigned_forbidden(client, mock_db):
|
|||
|
|
mock_db["advisor"].is_assigned.return_value = False
|
|||
|
|
login = client.post("/api/auth/login", json={"actor_id": "STAFF-10086", "token_type": "staff"})
|
|||
|
|
token = login.json()["data"]["access_token"]
|
|||
|
|
resp = client.post(
|
|||
|
|
"/api/chat",
|
|||
|
|
json={"message": "查持仓", "customer_id": "CUST-1010"},
|
|||
|
|
headers={
|
|||
|
|
"Authorization": f"Bearer {token}",
|
|||
|
|
"X-Agent-Type": "advisor",
|
|||
|
|
},
|
|||
|
|
)
|
|||
|
|
assert resp.status_code == 403
|
|||
|
|
assert mock_db["audit"].insert.called
|