qiaoqichaoandClaude Code
d56c699ce6
按 docs/v2/05-implementation-plan.md §1 候选迁移表与 claude_use/T00 白名单， 从 55e2cdb 抽取 224 个文件（2.48 MB）到独立目录 F:/mini-OpenClaw-v2。 - 排除：凭据(.env/server-deploy.conf/旧DELIVERY.md)、运行数据(xingbang_data 等 4 个 SQLite)、依赖(venv/node_modules 符号链接)、归档(_backup*/test-evidence/ ori_file)、Office 临时文件 - 目标目录独立 git 仓库，初始恢复点；Gitea 远程与分支待 OQ-06 Q2 定案 - 未部署、未启动服务、未调用模型 Co-Authored-By: Claude Code <noreply@anthropic.com>
151 lines
9.8 KiB
Python
151 lines
9.8 KiB
Python
import json
|
|
import os
|
|
import sqlite3
|
|
import tempfile
|
|
import unittest
|
|
from copy import deepcopy
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
from api import risk_agent_config as engine
|
|
from api.risk_agent_alignment import align
|
|
from api.risk_agent_connectors import execute, validate_query
|
|
from services.business_demo_database import build_business_demo_database, database_summary
|
|
|
|
|
|
class AlignmentTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.card={"mapping":{"financial.total_assets":"assets"},"unit":"元","alignment":{"project_field":"pid","period_field":"year","scope_field":"scope","selection":"latest","target_scope":"合并"}}
|
|
self.rows=[{"pid":"P1","year":"2024","scope":"合并","assets":10000},{"pid":"P1","year":"2025","scope":"合并","assets":20000}]
|
|
def test_latest_keeps_history_and_converts(self):
|
|
records,selected=align(self.card,self.rows,"P1")
|
|
self.assertEqual(len(records),2)
|
|
self.assertEqual(selected["values"]["financial.total_assets"],2)
|
|
self.assertEqual(selected["period"],"2025")
|
|
def test_wrong_project_rejected(self):
|
|
with self.assertRaises(ValueError): align(self.card,self.rows,"P2")
|
|
def test_duplicates_rejected(self):
|
|
with self.assertRaises(ValueError): align(self.card,self.rows+self.rows,"P1")
|
|
def test_period_selection_and_all(self):
|
|
_,selected=align(self.card,self.rows,"P1","2024")
|
|
self.assertEqual(selected["values"]["financial.total_assets"],1)
|
|
self.card["alignment"]["selection"]="all"
|
|
records,selected=align(self.card,self.rows,"P1")
|
|
self.assertIsNone(selected)
|
|
self.assertEqual(len(records),2)
|
|
def test_mixed_scopes_rejected(self):
|
|
self.card["alignment"].pop("target_scope")
|
|
self.rows[1]["scope"]="母公司"
|
|
with self.assertRaises(ValueError): align(self.card,self.rows,"P1")
|
|
|
|
|
|
class DatabaseTests(unittest.TestCase):
|
|
def test_multitable_business_database_and_configured_cards(self):
|
|
with tempfile.TemporaryDirectory() as root:
|
|
database = build_business_demo_database(Path(root) / "business.sqlite3")
|
|
summary = database_summary(database)
|
|
self.assertEqual(len(summary["tables"]), 11)
|
|
self.assertEqual(summary["tables"]["lease_project"], 3)
|
|
connection = deepcopy(next(item for item in engine.SEED["connections"] if item["id"] == "conn-business-demo"))
|
|
connection["database"] = str(database)
|
|
cards = {item["id"]: item for item in engine.SEED["cards"] if item["id"].startswith("card-business-")}
|
|
with patch.dict(os.environ, {"XINGBANG_SQLITE_ROOT": root}):
|
|
project = engine.query_card(cards["card-business-project"], connection, "DEMO-B")
|
|
finance = engine.query_card(cards["card-business-finance"], connection, "DEMO-B")
|
|
credit = engine.query_card(cards["card-business-credit"], connection, "DEMO-B")
|
|
lease = engine.query_card(cards["card-business-lease"], connection, "DEMO-B")
|
|
repayment = engine.query_card(cards["card-business-repayment"], connection, "DEMO-B")
|
|
self.assertEqual(project[0]["facility_amount"], 5000)
|
|
self.assertEqual(len(finance), 2)
|
|
self.assertIsNone(finance[-1]["operating_cashflow"])
|
|
self.assertEqual(credit[0]["legal_case_count"], 1)
|
|
self.assertEqual(lease[0]["lease_total_value"], 5400)
|
|
self.assertEqual(lease[0]["duplicate_asset_count"], 1)
|
|
self.assertEqual(repayment[0]["evidence_status"], "缺少最新流水")
|
|
|
|
def test_real_sqlite_read_only_and_allowlist(self):
|
|
with tempfile.TemporaryDirectory() as root:
|
|
file=Path(root)/"erp.sqlite"
|
|
db=sqlite3.connect(file)
|
|
db.execute("CREATE TABLE finance(project_id TEXT, assets REAL)")
|
|
db.execute("INSERT INTO finance VALUES ('P1',100)");db.commit();db.close()
|
|
conn={"type":"SQLite","database":str(file),"allowed_tables":["finance"]}
|
|
with patch.dict(os.environ,{"XINGBANG_SQLITE_ROOT":root}):
|
|
self.assertEqual(execute(conn,"SELECT * FROM finance WHERE project_id=:project_id","P1")[0]["assets"],100)
|
|
self.assertEqual(execute(conn,"SELECT * FROM finance WHERE project_id=:project_id","' OR 1=1 --"),[])
|
|
with self.assertRaises(ValueError):execute(conn,"DELETE FROM finance","P1")
|
|
with self.assertRaises(ValueError):execute(conn,"SELECT * FROM sqlite_master","P1")
|
|
def test_sql_parser_rejects_multi_statement_and_writes(self):
|
|
for sql in ("SELECT * FROM finance; DELETE FROM finance", "SELECT * INTO backup FROM finance", "SELECT pg_sleep(5) FROM finance"):
|
|
with self.assertRaises(ValueError):validate_query(sql,["finance"],"postgres")
|
|
|
|
|
|
ACCOUNTS=[{"id":"editor","token":"editor-token","role":"admin"},{"id":"approver","token":"review-token","role":"reviewer"},{"id":"manager","token":"manager-token","role":"manager","projects":["XB-2026-001"]}]
|
|
|
|
class WorkflowTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp=tempfile.TemporaryDirectory()
|
|
self.file_patch=patch.object(engine,"DB_PATH",Path(self.temp.name)/"catalog.sqlite");self.file_patch.start()
|
|
self.auth_patch=patch.dict(os.environ,{"XINGBANG_AUTH_USERS":json.dumps(ACCOUNTS)});self.auth_patch.start()
|
|
app=FastAPI();app.include_router(engine.router)
|
|
self.client=TestClient(app);self.base="/api/xingbang/risk-agent"
|
|
def tearDown(self):
|
|
self.client.close();self.auth_patch.stop();self.file_patch.stop();self.temp.cleanup()
|
|
def call(self,path,method="GET",data=None,token="editor-token"):
|
|
return self.client.request(method,self.base+path,json=data,headers={"Authorization":"Bearer "+token})
|
|
def test_auth_and_project_scope(self):
|
|
self.assertEqual(self.call("/catalog",token="bad").status_code,401)
|
|
self.assertEqual(self.call("/cards","POST",{"name":"unauthorized"},"manager-token").status_code,403)
|
|
catalog=self.call("/catalog",token="manager-token").json()
|
|
self.assertEqual(catalog["cards"],[])
|
|
self.assertEqual(self.call("/plans/plan-default/run","POST",{"project_id":"P2"},"manager-token").status_code,403)
|
|
result=self.call("/plans/plan-default/run","POST",{"project_id":"XB-2026-001"},"manager-token").json()
|
|
self.assertNotIn("configuration",result)
|
|
def test_demo_role_dropdown_is_gated_and_preserves_role_permissions(self):
|
|
header=lambda role:{"X-Xingbang-Demo-Role":role}
|
|
self.assertEqual(self.client.get(self.base+"/access/context",headers=header("manager")).status_code,401)
|
|
with patch.dict(os.environ,{"XINGBANG_DEMO_ROLE_LOGIN":"1"}):
|
|
for role in ("admin","reviewer","manager"):
|
|
response=self.client.get(self.base+"/access/context",headers=header(role))
|
|
self.assertEqual(response.status_code,200,response.text)
|
|
identity=response.json()
|
|
self.assertEqual(identity["role"],role)
|
|
self.assertTrue(identity["demo"])
|
|
self.assertNotIn("*",identity["projects"])
|
|
self.assertEqual(self.client.get(self.base+"/access/context",headers=header("owner")).status_code,401)
|
|
self.assertEqual(self.client.post(self.base+"/cards",headers=header("manager"),json={"name":"x"}).status_code,403)
|
|
self.assertEqual(self.client.post(self.base+"/cards",headers=header("reviewer"),json={"name":"x"}).status_code,403)
|
|
def test_review_publishes_exact_version(self):
|
|
item=self.call("/skills","POST",{"name":"新增检查","data":{"engine":"quality","inputs":["*"]}}).json()
|
|
item["status"]="published"
|
|
self.assertEqual(self.call("/skills/"+item["id"],"PUT",{"name":item["name"],"data":item}).status_code,403)
|
|
pending=self.call("/skills/"+item["id"]+"/submit","POST").json()
|
|
path="/reviews/"+pending["id"]+"/review"
|
|
self.assertEqual(self.call(path,"POST",{"approve":True,"note":"同意"}).status_code,403)
|
|
self.assertEqual(self.call(path,"POST",{"approve":True,"note":"确认输入契约"},"review-token").status_code,200)
|
|
saved=next(x for x in self.call("/catalog").json()["skills"] if x["id"]==item["id"])
|
|
self.assertEqual(saved["status"],"published")
|
|
self.assertEqual(saved["approved_by"],"approver")
|
|
def test_changed_submission_cannot_be_approved(self):
|
|
item=self.call("/skills","POST",{"name":"检查","data":{"engine":"quality","inputs":["*"]}}).json()
|
|
pending=self.call("/skills/"+item["id"]+"/submit","POST").json()
|
|
self.call("/skills/"+item["id"],"PUT",{"name":"修改后","data":item})
|
|
response=self.call("/reviews/"+pending["id"]+"/review","POST",{"approve":True,"note":"同意"},"review-token")
|
|
self.assertEqual(response.status_code,409)
|
|
def test_natural_rules_reach_model_without_explicit_model_skill(self):
|
|
item=self.call("/catalog").json()["rules"][0]
|
|
item.update(status="draft",natural_language="重点关注现金流，不能下审批结论")
|
|
item=self.call("/rules/"+item["id"],"PUT",{"name":item["name"],"data":item}).json()
|
|
pending=self.call("/rules/"+item["id"]+"/submit","POST").json()
|
|
self.call("/reviews/"+pending["id"]+"/review","POST",{"approve":True,"note":"内容已核对"},"review-token")
|
|
with patch.object(engine,"local_model",return_value="请补充现金流材料") as model:
|
|
response=self.call("/plans/plan-default/run","POST",{"project_id":"XB-2026-001"},"manager-token")
|
|
self.assertEqual(response.status_code,200,response.text)
|
|
context=model.call_args.args[1]
|
|
self.assertEqual(context["human_rules"][0]["instructions"],item["natural_language"])
|
|
self.assertTrue(context["peer_datasets"])
|
|
self.assertIn("advisories",response.json())
|
|
|
|
if __name__=="__main__":unittest.main()
|