feat: 核心服务端基础框架（身份/平台 API）

- 七端口 RBAC、select-identity、JWT、审计 - FastAPI + SQLAlchemy + SQLite，/auth /opc /admin /agents 等路由
2026-08-23 22:35:59 +08:00
commit 26fa546f67
39 changed files with 8354 additions and 0 deletions
@@ -0,0 +1,236 @@
+# -*- 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_agent_per_port_isolation(client):
+    # 同一账号在不同端口拥有各自独立的智能体（多端口彻底隔离）
+    body = client.post("/auth/login", json={"username": "pine", "password": "123456"}).json()
+    token = body["token"]
+    opc_ident = next(i for i in body["identities"] if i["port"] == "opc")
+    op_id = next(i for i in body["identities"] if i["port"] == "operator")
+    # 运营端：创建专属智能体
+    sel_op = client.post("/auth/select-identity", headers=auth(token), json={"identity_id": op_id["id"]}).json()
+    created = client.post("/agents", headers=auth(sel_op["token"]), json={"name": "运营专属助手"}).json()
+    assert created["port"] == "operator"
+    op_agents = client.get("/agents", headers=auth(sel_op["token"])).json()
+    assert any(a["id"] == created["id"] for a in op_agents)
+    # OPC 端：看不到运营端的专属智能体（彻底隔离）
+    sel_opc = client.post("/auth/select-identity", headers=auth(token), json={"identity_id": opc_ident["id"]}).json()
+    opc_agents = client.get("/agents", headers=auth(sel_opc["token"])).json()
+    assert all(a["id"] != created["id"] for a in opc_agents)
+    assert all(a["port"] == "opc" for a in opc_agents)
+
+
+def test_agent_neutral_token_fully_isolated(client):
+    # 多身份账号登录后未 select-identity 的中性令牌：不得跨端口访问任何智能体
+    body = client.post("/auth/login", json={"username": "pine", "password": "123456"}).json()
+    neutral = body["token"]
+    # 中性令牌（未解析端口身份）访问智能体一律 403
+    assert client.get("/agents", headers=auth(neutral)).status_code == 403
+    # 先选身份创建 opc 智能体
+    opc_ident = next(i for i in body["identities"] if i["port"] == "opc")["id"]
+    ot = client.post(
+        "/auth/select-identity", headers=auth(neutral), json={"identity_id": opc_ident}
+    ).json()["token"]
+    created = client.post("/agents", headers=auth(ot), json={"name": "OPC专属"}).json()
+    assert created["port"] == "opc"
+    # 中性令牌改/删该智能体：403（不能跨端口或绕过身份）
+    assert (
+        client.put(
+            f"/agents/{created['id']}", headers=auth(neutral), json={"name": "x"}
+        ).status_code
+        == 403
+    )
+    assert client.delete(f"/agents/{created['id']}", headers=auth(neutral)).status_code == 403
+    # 带身份令牌仍可正常管理（回归）
+    upd = client.put(
+        f"/agents/{created['id']}", headers=auth(ot), json={"name": "OPC专属v2"}
+    )
+    assert upd.status_code == 200 and upd.json()["name"] == "OPC专属v2"
+
+
+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"