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>
131 lines
8.5 KiB
Python
131 lines
8.5 KiB
Python
import tempfile
|
|
import unittest
|
|
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
|
|
|
|
|
|
class RiskAgentTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp=tempfile.TemporaryDirectory()
|
|
self.patch=patch.object(engine,"DB_PATH",Path(self.temp.name)/"catalog.sqlite")
|
|
self.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.patch.stop();self.temp.cleanup()
|
|
def run_plan(self,project="XB-2026-001"):
|
|
return self.client.post(self.base+"/plans/plan-default/run",json={"project_id":project})
|
|
def test_real_values_and_history(self):
|
|
response=self.run_plan();self.assertEqual(response.status_code,200,response.text)
|
|
result=response.json()
|
|
self.assertAlmostEqual(result["values"]["metrics.debt_ratio"],13200/18000)
|
|
self.assertTrue(any(f["category"]=="信息缺失" for f in result["findings"]))
|
|
self.assertEqual(self.client.get(self.base+"/runs/history").json()[0]["id"],result["id"])
|
|
def test_financial_leasing_skills_and_plan_are_loaded(self):
|
|
catalog=self.client.get(self.base+"/catalog").json()
|
|
skill_ids={item["id"] for item in catalog["skills"]}
|
|
self.assertTrue({"skill-fl-lessee","skill-fl-asset","skill-fl-sale-leaseback","skill-fl-guarantee","skill-fl-review-report"}.issubset(skill_ids))
|
|
plan=next(item for item in catalog["plans"] if item["id"]=="plan-financial-leasing-corporate")
|
|
self.assertEqual(len(plan["cards"]),5)
|
|
native={item["name"] for item in self.client.get(self.base+"/native-skills/list").json()}
|
|
self.assertIn("financial-leasing-asset-review",native)
|
|
self.assertNotIn("get_weather",native)
|
|
self.assertNotIn("get_weather_open",native)
|
|
def test_sql_write_denied(self):
|
|
card=dict(engine.SEED["cards"][0],sql="DELETE FROM project")
|
|
with self.assertRaises(Exception):engine.query_card(card,engine.SEED["connections"][0],"XB-2026-001")
|
|
def test_parameter_binding(self):
|
|
result=self.run_plan("' OR 1=1 --").json()
|
|
self.assertEqual(result["values"],{})
|
|
self.assertTrue(result["findings"])
|
|
def test_edit_changes_result_preserves_snapshot(self):
|
|
first=self.run_plan().json()
|
|
card=self.client.get(self.base+"/catalog").json()["cards"][1]
|
|
card["sql"]="SELECT project_id, period, scope, assets, 1000 AS liabilities, cashflow FROM finance WHERE project_id=:project_id"
|
|
response=self.client.put(self.base+"/cards/"+card["id"],json={"name":card["name"],"data":card})
|
|
self.assertEqual(response.status_code,200,response.text)
|
|
second=self.run_plan().json()
|
|
self.assertAlmostEqual(second["values"]["metrics.debt_ratio"],1000/18000)
|
|
self.assertNotEqual(first["values"],second["values"])
|
|
old=self.client.get(self.base+"/runs/history").json()[1]
|
|
self.assertEqual(first,old)
|
|
def test_referenced_archive_and_stale_save_rejected(self):
|
|
self.assertEqual(self.client.delete(self.base+"/cards/card-finance").status_code,409)
|
|
item=self.client.get(self.base+"/catalog").json()["skills"][0]
|
|
url=self.base+"/skills/"+item["id"]
|
|
payload={"name":item["name"],"data":item}
|
|
self.assertEqual(self.client.put(url,json=payload).status_code,200)
|
|
self.assertEqual(self.client.put(url,json=payload).status_code,409)
|
|
def test_disabled_dependency_blocks_run(self):
|
|
item=self.client.get(self.base+"/catalog").json()["cards"][1];item["status"]="draft"
|
|
self.client.put(self.base+"/cards/"+item["id"],json={"name":item["name"],"data":item})
|
|
self.assertEqual(self.run_plan().status_code,409)
|
|
def test_model_never_calls_external_url(self):
|
|
with patch.dict("os.environ",{"XINGBANG_LOCAL_MODEL_URL":"https://example.com/v1"}):
|
|
with self.assertRaises(ValueError):engine.local_model({}, {}, {})
|
|
def test_official_deepseek_model_url_is_allowed(self):
|
|
with patch.dict("os.environ",{"XINGBANG_LOCAL_MODEL_URL":"https://api.deepseek.com"}), patch("graph.risk_task.invoke_risk_task",return_value={"text":"ok"}) as invoke:
|
|
result=engine.local_model({}, {}, {})
|
|
self.assertEqual(result,{"text":"ok"})
|
|
self.assertEqual(invoke.call_args.args[3],"https://api.deepseek.com")
|
|
def test_crud_custom_skill(self):
|
|
response=self.client.post(self.base+"/skills",json={"name":"新检查","data":{"engine":"quality","inputs":["*"],"output":"quality"}})
|
|
self.assertEqual(response.status_code,200,response.text)
|
|
item=response.json();self.assertEqual(item["status"],"draft")
|
|
self.assertEqual(self.client.delete(self.base+"/skills/"+item["id"]).status_code,200)
|
|
def test_admin_imports_skill_folder_as_native_skill_card(self):
|
|
content="---\nname: local-leasing-check\ndisplay_name: 本地金租核查\ndescription: 核查租赁物证据\ncategory: 租赁物核验\ninputs: [lease.total_value]\nversion: 1.0.0\n---\n先读取授权证据，不作审批结论。"
|
|
base=Path(self.temp.name)/"native"
|
|
with patch.object(engine,"SKILLS_BASE_DIR",base):
|
|
response=self.client.post(self.base+"/native-skills/import",json={"skills":[{"skill_md":content}]})
|
|
self.assertEqual(response.status_code,201,response.text)
|
|
imported=response.json()["skills"][0]
|
|
self.assertEqual(imported["name"],"本地金租核查")
|
|
self.assertEqual(imported["native_skills"],["imported/local-leasing-check/SKILL.md"])
|
|
self.assertEqual(imported["status"],"published")
|
|
self.assertTrue((base/"skills"/"imported"/"local-leasing-check"/"SKILL.md").exists())
|
|
self.assertEqual(self.client.post(self.base+"/native-skills/import",json={"skills":[{"skill_md":content}]}).status_code,409)
|
|
def test_currency_units_are_normalized_with_raw_evidence(self):
|
|
item=self.client.get(self.base+"/catalog").json()["cards"][3]
|
|
item["unit"]="元"
|
|
self.client.put(self.base+"/cards/"+item["id"],json={"name":item["name"],"data":item})
|
|
result=self.client.post(self.base+"/plans/plan-energy/run",json={"project_id":"XB-2026-001"}).json()
|
|
self.assertAlmostEqual(result["values"]["metrics.coverage"],.54/5000)
|
|
self.assertEqual(result["evidence"]["lease.total_value"][0]["raw_value"],5400)
|
|
self.assertEqual(result["evidence"]["lease.total_value"][0]["unit"],"万元")
|
|
def test_local_model_response_is_separate_advisory(self):
|
|
plan=self.client.get(self.base+"/catalog").json()["plans"][0]
|
|
plan["skills"].append("skill-trend")
|
|
self.client.put(self.base+"/plans/"+plan["id"],json={"name":plan["name"],"data":plan})
|
|
with patch.object(engine,"local_model",return_value="请核实负债变化原因"):
|
|
result=self.run_plan().json()
|
|
self.assertEqual(result["advisories"][0]["text"],"请核实负债变化原因")
|
|
def test_cross_period_ratio_is_blocked(self):
|
|
catalog=self.client.get(self.base+"/catalog").json()
|
|
card=catalog["cards"][1]
|
|
card["mapping"].pop("financial.total_liabilities")
|
|
self.client.put(self.base+"/cards/"+card["id"],json={"name":card["name"],"data":card})
|
|
other=dict(card,name="上年负债",mapping={"financial.total_liabilities":"liabilities"},period="2024")
|
|
created=self.client.post(self.base+"/cards",json={"name":other["name"],"data":other}).json()
|
|
created["status"]="published"
|
|
self.client.put(self.base+"/cards/"+created["id"],json={"name":created["name"],"data":created})
|
|
plan=catalog["plans"][0];plan["cards"].append(created["id"])
|
|
self.client.put(self.base+"/plans/"+plan["id"],json={"name":plan["name"],"data":plan})
|
|
result=self.run_plan().json()
|
|
self.assertNotIn("metrics.debt_ratio",result["values"])
|
|
self.assertTrue(any("期间或口径不一致" in f["title"] for f in result["findings"]))
|
|
def test_missing_model_endpoint_is_not_success(self):
|
|
rules=self.client.get(self.base+"/catalog").json()["rules"][0]
|
|
rules["natural_language"]="分析偿债风险"
|
|
self.client.put(self.base+"/rules/"+rules["id"],json={"name":rules["name"],"data":rules})
|
|
with patch.dict("os.environ",{"XINGBANG_LOCAL_MODEL_URL":""}):
|
|
result=self.run_plan().json()
|
|
self.assertTrue(result["summary"]["incomplete"])
|
|
self.assertEqual(result["steps"][-1]["status"],"未完成")
|
|
|
|
if __name__=="__main__":unittest.main()
|