Pine
26fa546f67
- 七端口 RBAC、select-identity、JWT、审计 - FastAPI + SQLAlchemy + SQLite，/auth /opc /admin /agents 等路由
58 lines
2.1 KiB
Python
58 lines
2.1 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""OPC 超级个体端点测试：工作台聚合数据。"""
|
|
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_opc_dashboard_shape(client):
|
|
"""opc_member 可访问工作台，且返回标准化聚合结构（不含前端演示数据）。"""
|
|
token = login(client, "opc01")
|
|
res = client.get("/opc/dashboard", headers=auth(token))
|
|
assert res.status_code == 200, res.text
|
|
body = res.json()
|
|
|
|
stats = body["stats"]
|
|
assert set(stats) == {"inProgressTasks", "completedTasks", "totalEarnings", "creditScore"}
|
|
assert stats["inProgressTasks"] == 2 # ot_001 in_progress + ot_002 urgent
|
|
assert stats["completedTasks"] == 1 # ot_003 completed
|
|
assert stats["creditScore"] == 88
|
|
assert stats["totalEarnings"] == sum( # fin_001..fin_006 income
|
|
[3200, 4500, 2800, 5100, 3900, 4600]
|
|
)
|
|
|
|
# 进行中任务：仅 in_progress / urgent，且最多 3 条
|
|
assert len(body["activeTasks"]) == 2
|
|
assert all(t["status"] in ("in_progress", "urgent") for t in body["activeTasks"])
|
|
|
|
# 智能体建议为 3 条
|
|
assert len(body["agentTips"]) == 3
|
|
|
|
# 月度收入：按 YYYY-MM 聚合升序
|
|
months = [m["month"] for m in body["monthlyIncome"]]
|
|
assert months == sorted(months)
|
|
assert len(body["monthlyIncome"]) <= 6
|
|
|
|
# 最新消息：最多 4 条
|
|
assert len(body["recentMessages"]) <= 4
|
|
assert body["recentMessages"][0]["msg_type"] in (
|
|
"task", "policy", "service", "agent", "system",
|
|
)
|
|
|
|
|
|
def test_opc_dashboard_rejects_other_roles(client):
|
|
"""非 opc_member（如企业端）访问应返回 403。"""
|
|
token = login(client, "ent01")
|
|
res = client.get("/opc/dashboard", headers=auth(token))
|
|
assert res.status_code == 403
|