104 lines
6.3 KiB
Python
104 lines
6.3 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_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_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_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()
|