Pine
ac2df72b21
- 删 UserIdentity 模型/IdentityRepository/db.identities，账号统一单一角色(users.role) - auth.py 去 select-identity/_identity_summaries，登录恒返回 identities=[]；_login_response_for_user 按单角色签发 - 删多端口路由 rbac_government/investor/developer 及其 main 挂载 - seed 去多身份回填(_migrate_identities/_ensure_port_agents/多端身份/端口标签映射) - dependencies: require_port 去端口隔离(单角色放宽)、optional_current_user 修 identity_id 多余参数 - user_admin_service create_user 不再建端口身份；schemas 去 SelectIdentityRequest - 新增迁移 0014_drop_user_identities(drop user_identities 表) - 测试改写为单角色契约(登录空 identities、无 select-identity、账号级智能体) Co-Authored-By: Claude <noreply@anthropic.com>
207 lines
9.0 KiB
Python
207 lines
9.0 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""各业务端口工作台端点测试（企业/载体/服务商/政务）。"""
|
|
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_port_dashboards_accessible_by_own_role(client):
|
|
cases = [
|
|
("ent01", "/enterprise/dashboard"),
|
|
("car01", "/carrier/dashboard"),
|
|
("pro01", "/provider/dashboard"),
|
|
("gov_prov", "/government/dashboard"),
|
|
]
|
|
for username, path in cases:
|
|
token = login(client, username)
|
|
res = client.get(path, headers=auth(token))
|
|
assert res.status_code == 200, f"{path}: {res.text}"
|
|
body = res.json()
|
|
assert len(body.get("stats", [])) >= 1, f"{path}: no stats"
|
|
assert body.get("port") == path.split("/")[1]
|
|
|
|
|
|
def test_port_dashboard_rejects_other_role(client):
|
|
token = login(client, "opc01") # OPC 非企业
|
|
res = client.get("/enterprise/dashboard", headers=auth(token))
|
|
assert res.status_code == 403
|
|
|
|
|
|
def test_port_subpages_by_role(client):
|
|
cases = [
|
|
("ent01", ["/enterprise/talents", "/enterprise/tasks"]),
|
|
("car01", ["/carrier/opc", "/carrier/services"]),
|
|
("pro01", ["/provider/orders", "/provider/services"]),
|
|
("gov_prov", ["/government/opc", "/government/policies", "/government/carriers", "/government/data"]),
|
|
("opc01", ["/opc/agent", "/opc/tasks", "/opc/finance", "/opc/messages"]),
|
|
]
|
|
for username, paths in cases:
|
|
token = login(client, username)
|
|
for p in paths:
|
|
res = client.get(p, headers=auth(token))
|
|
assert res.status_code == 200, f"{p}: {res.text}"
|
|
|
|
|
|
def test_port_subpage_rejects_other_role(client):
|
|
token = login(client, "opc01")
|
|
res = client.get("/enterprise/talents", headers=auth(token))
|
|
assert res.status_code == 403
|
|
|
|
|
|
def test_enterprise_publish_task(client):
|
|
token = login(client, "ent01")
|
|
res = client.post(
|
|
"/enterprise/tasks", headers=auth(token),
|
|
json={"title": "测试发布", "category": "设计创意", "budget_min": 1000, "budget_max": 3000},
|
|
)
|
|
assert res.status_code == 200, res.text
|
|
assert res.json()["status"] == "pending"
|
|
|
|
|
|
def test_enterprise_tender_flow(client):
|
|
# 企业发布并提交 → 已发布；OPC 投标 → 企业中标 → 验收通过
|
|
ent = login(client, "ent01")
|
|
opc = login(client, "opc01")
|
|
task = client.post("/enterprise/tasks", headers=auth(ent),
|
|
json={"title": "投标测试任务", "mode": "bid", "category": "设计创意",
|
|
"budget_min": 1000, "budget_max": 3000}).json()
|
|
tid = task["id"]
|
|
sub = client.post(f"/enterprise/tasks/{tid}/submit", headers=auth(ent))
|
|
assert sub.status_code == 200 and sub.json()["status"] == "published"
|
|
|
|
bid = client.post(f"/opc/tasks/{tid}/bid", headers=auth(opc),
|
|
json={"quote": 1500, "plan": "两周交付"}).json()
|
|
bids = client.get(f"/enterprise/tasks/{tid}/bids", headers=auth(ent)).json()
|
|
assert any(b["id"] == bid["id"] for b in bids["items"])
|
|
|
|
won = client.post(f"/enterprise/tasks/{tid}/bids/{bid['id']}/win", headers=auth(ent))
|
|
assert won.status_code == 200 and won.json()["status"] == "win"
|
|
|
|
delivered = client.post(f"/opc/tasks/{tid}/deliver", headers=auth(opc))
|
|
assert delivered.status_code == 200 and delivered.json()["status"] == "delivered"
|
|
|
|
done = client.post(f"/enterprise/tasks/{tid}/review", headers=auth(ent), json={"action": "accept"})
|
|
assert done.status_code == 200 and done.json()["status"] == "completed"
|
|
|
|
|
|
def test_government_scope_filtering(client):
|
|
# 省级可见全省 OPC；区县级(五华区)仅可见本区县 OPC
|
|
prov = login(client, "gov_prov")
|
|
dist = login(client, "gov_dist")
|
|
prov_opc = client.get("/government/opc", headers=auth(prov)).json()["items"]
|
|
dist_opc = client.get("/government/opc", headers=auth(dist)).json()["items"]
|
|
assert len(prov_opc) >= len(dist_opc)
|
|
# gov_dist 数据范围不含省级
|
|
data = client.get("/government/data", headers=auth(prov)).json()
|
|
assert data["stats"][0]["key"] == "totalOPC"
|
|
|
|
|
|
def test_subsidy_three_level_approval(client):
|
|
# 补贴三级审批：区县(初审)→市(复审)→省(终审)→发放
|
|
gov_dist = login(client, "gov_dist")
|
|
sub = client.get("/government/subsidies", headers=auth(gov_dist)).json()["items"][0]
|
|
aid = sub["id"]
|
|
res = client.post(f"/government/subsidies/{aid}/approve", headers=auth(gov_dist))
|
|
assert res.status_code == 200 and res.json()["status"] == "district_review"
|
|
|
|
|
|
def test_carrier_referral(client):
|
|
car = login(client, "car01")
|
|
res = client.post("/carrier/referrals", headers=auth(car),
|
|
json={"provider_name": "云超财税服务所", "opc_name": "OPC个人创业者"})
|
|
assert res.status_code == 200 and res.json()["status"] == "referred"
|
|
|
|
|
|
def test_training_enroll(client):
|
|
inv = login(client, "inv01")
|
|
res = client.post("/investor/trainings/商业计划书（BP）撰写实战/enroll", headers=auth(inv))
|
|
assert res.status_code == 200 and res.json()["status"] == "enrolled"
|
|
mine = client.get("/investor/my-trainings", headers=auth(inv)).json()["items"]
|
|
assert len(mine) >= 1
|
|
|
|
|
|
def test_ecosystem_crosscut(client):
|
|
# 通知、信用、合同、结算、争议、撮合、路演加入
|
|
opc = login(client, "opc01")
|
|
ent = login(client, "ent01")
|
|
inv = login(client, "inv01")
|
|
op = login(client, "pine") # operator 超级管理员
|
|
|
|
ntf = client.get("/notifications", headers=auth(opc)).json()
|
|
assert "items" in ntf and ntf["unread"] >= 1
|
|
|
|
credit = client.get("/me/credit", headers=auth(opc)).json()
|
|
assert "credit_score" in credit
|
|
|
|
# 撮合
|
|
matches = client.get("/investor/matches", headers=auth(inv)).json()
|
|
assert "items" in matches
|
|
|
|
# 路演加入
|
|
join = client.post("/roadshows/rs_001/join", headers=auth(inv))
|
|
assert join.status_code == 200 and join.json()["joined"] is True
|
|
|
|
|
|
def test_agents_single_role(client):
|
|
# 单角色：智能体归账号所有（不再按端口隔离），可增改删；他人不可见/不可改
|
|
pine = login(client, "pine")
|
|
created = client.post("/agents", headers=auth(pine), json={"name": "我的助手"}).json()
|
|
assert created["name"] == "我的助手"
|
|
mine = client.get("/agents", headers=auth(pine)).json()
|
|
assert any(a["id"] == created["id"] for a in mine)
|
|
# 他人(OPC)不可见
|
|
opc = login(client, "opc01")
|
|
other = client.get("/agents", headers=auth(opc)).json()
|
|
assert all(a["id"] != created["id"] for a in other)
|
|
# 本人可更新/删除
|
|
upd = client.put(f"/agents/{created['id']}", headers=auth(pine), json={"name": "助手v2"})
|
|
assert upd.status_code == 200 and upd.json()["name"] == "助手v2"
|
|
assert client.delete(f"/agents/{created['id']}", headers=auth(pine)).status_code == 200
|
|
|
|
|
|
def test_org_members(client):
|
|
# 企业管理员查看/添加机构成员；一账号挂多机构
|
|
ent = login(client, "ent01")
|
|
members = client.get("/org/o_ent_001/members", headers=auth(ent)).json()["items"]
|
|
assert any(m["user_id"] == "u_ent_01" and m["is_admin"] for m in members)
|
|
assert any(m["user_id"] == "u_ent_fin" and m["role"] == "finance" for m in members)
|
|
# 非机构管理员（普通成员）不可查看
|
|
opc = login(client, "opc01")
|
|
res = client.get("/org/o_ent_001/members", headers=auth(opc))
|
|
assert res.status_code == 403
|
|
|
|
|
|
def test_phone_account_register_validation(client):
|
|
# 注册需 11 位手机号
|
|
res = client.post("/auth/register", json={"username": "abc", "password": "123456"})
|
|
assert res.status_code == 400
|
|
res2 = client.post("/auth/register", json={"username": "13800001234", "password": "123456"})
|
|
assert res2.status_code in (200, 403) # 403=演示端已存在用户禁止注册
|
|
|
|
|
|
def test_admin_create_user_and_org_member(client):
|
|
# 运营端新增账号（不同类型）
|
|
body = client.post("/auth/login", json={"username": "pine", "password": "123456"}).json()
|
|
token = body["token"]
|
|
op_id = next(i for i in body["identities"] if i["port"] == "operator")["id"]
|
|
ot = client.post("/auth/select-identity", headers=auth(token), json={"identity_id": op_id}).json()["token"]
|
|
created = client.post("/admin/users", headers=auth(ot),
|
|
json={"username": "13900009990", "password": "123456",
|
|
"nickname": "新账号", "role": "provider", "sub_role": "admin"}).json()
|
|
assert created["username"] == "13900009990" and created["role"] == "provider"
|
|
# 企业管理员按用户名添加机构成员
|
|
ent = login(client, "ent01")
|
|
added = client.post("/org/o_ent_001/members", headers=auth(ent),
|
|
json={"username": "13900009990", "role": "publisher"}).json()
|
|
assert added["role"] == "publisher"
|