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()