53 lines
6.0 KiB
Python
53 lines
6.0 KiB
Python
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
from io import BytesIO
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
from api import risk_agent_config as risk
|
|
from api.collateral_agent import router
|
|
|
|
|
|
class CollateralTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp=tempfile.TemporaryDirectory();self.dbpatch=patch.object(risk,"DB_PATH",Path(self.temp.name)/"test.sqlite");self.dbpatch.start()
|
|
self.env=patch.dict("os.environ",{"XINGBANG_AUTH_USERS":"","XINGBANG_REQUIRE_AUTH":"0","XINGBANG_LOCAL_MODEL_URL":""});self.env.start()
|
|
app=FastAPI();app.include_router(risk.router);app.include_router(router);self.client=TestClient(app);self.base="/api/xingbang/risk-agent";self.root=self.base+"/collateral"
|
|
self.run=self.client.post(self.base+"/plans/plan-default/run",json={"project_id":"XB-2026-001"}).json()
|
|
self.rule=self.client.post(self.root+"/rules",json={"name":"测试增信规则","required_kinds":["asset","insurance"],"allowed_asset_types":["设备"],"required_coverages":["财产险"],"beneficiary":"兴邦金租","basis":"测试制度 v1"}).json();self.rule=self.client.post(self.root+f"/rules/{self.rule['id']}/status",json={"version":self.rule["version"],"kind":"publish","note":"独立审核发布"}).json()
|
|
self.case=self.client.post(self.root+"/cases",json={"run_id":self.run["id"],"lease_end":"2029-01-01"}).json()
|
|
def tearDown(self):self.client.close();self.dbpatch.stop();self.env.stop();self.temp.cleanup()
|
|
def act(self,kind,**extra):
|
|
r=self.client.post(f"{self.root}/cases/{self.case['id']}/actions",json={"version":self.case["version"],"kind":kind,"note":"人工核对",**extra});
|
|
if r.status_code==200:self.case=r.json()
|
|
return r
|
|
def measure(self,**extra):
|
|
base={"version":self.case["version"],"measure":{"kind":"asset","name":"设备 A","resource_key":"asset-a","amount":"5000","prior_claims":"0","asset_type":"设备","ownership_clear":True,"enforceable":True,"evidence":{"reference":"合同-1","as_of":"2026-09-01"}}};base["measure"].update(extra);r=self.client.post(f"{self.root}/cases/{self.case['id']}/measures",json=base);self.assertEqual(r.status_code,200,r.text);self.case=r.json();return self.case["measures"][-1]
|
|
def test_file_reference_and_confirm_gate(self):
|
|
self.assertEqual(self.client.post(f"{self.root}/cases/{self.case['id']}/actions",json={"version":self.case["version"],"kind":"confirm","item_id":"none","note":"x"}).status_code,404)
|
|
self.measure();item=self.case["measures"][-1];self.assertEqual(self.act("confirm",item_id=item["id"]).status_code,200)
|
|
def test_pdf_page_quote_is_checked(self):
|
|
r=self.client.post(f"{self.root}/cases/{self.case['id']}/documents?version={self.case['version']}",files={"file":("evidence.txt",b"Policy number P-001\nBeneficiary Xingbang", "text/plain")});self.assertEqual(r.status_code,200,r.text);self.case=r.json();doc=self.case["documents"][0]
|
|
self.assertEqual(self.measure(evidence={"document_id":doc["id"],"page":1,"quote":"Policy number P-001","reference":"保单 P-001","as_of":"2026-09-01"})["evidence"]["document_id"],doc["id"])
|
|
bad={"version":self.case["version"],"measure":{"kind":"asset","name":"错引","resource_key":"x","amount":"1","evidence":{"document_id":doc["id"],"page":1,"quote":"不存在","reference":"x","as_of":"2026-09-01"}}};self.assertEqual(self.client.post(f"{self.root}/cases/{self.case['id']}/measures",json=bad).status_code,422)
|
|
def test_duplicate_and_non_double_count_recovery(self):
|
|
first=self.measure();self.assertEqual(self.act("confirm",item_id=first["id"]).status_code,200)
|
|
second=self.measure(name="同一资产副本",resource_key="asset-a");self.assertEqual(self.act("confirm",item_id=second["id"]).status_code,200)
|
|
r=self.client.post(f"{self.root}/cases/{self.case['id']}/evaluate",json={"version":self.case["version"],"rules_id":self.rule["id"],"haircut":20,"note":"测算"});self.assertEqual(r.status_code,200,r.text);self.case=r.json();self.assertEqual(self.case["result"]["scenario"]["asset_recovery"],"0.0000")
|
|
def test_measure_change_invalidates_result_and_model_is_scoped(self):
|
|
m=self.measure();self.act("confirm",item_id=m["id"]);r=self.client.post(f"{self.root}/cases/{self.case['id']}/evaluate",json={"version":self.case["version"],"rules_id":self.rule["id"],"haircut":20,"note":"测算"});self.assertEqual(r.status_code,200);self.case=r.json();self.measure(amount="6000");self.assertTrue(self.case["stale"])
|
|
self.assertEqual(self.client.post(f"{self.root}/cases/{self.case['id']}/advice",json={"version":self.case["version"],"note":"生成"}).status_code,409)
|
|
def test_failed_model_preserves_result(self):
|
|
m=self.measure();self.act("confirm",item_id=m["id"]);r=self.client.post(f"{self.root}/cases/{self.case['id']}/evaluate",json={"version":self.case["version"],"rules_id":self.rule["id"],"haircut":20,"note":"测算"});self.case=r.json();version=self.case["version"]
|
|
r=self.client.post(f"{self.root}/cases/{self.case['id']}/advice",json={"version":version,"note":"生成"});self.assertEqual(r.status_code,503);current=self.client.get(self.root+"/workspace").json()["cases"][0];self.assertEqual(current["version"],version);self.assertIsNone(current["advice"])
|
|
def test_manager_project_scope(self):
|
|
users=[{"id":"m","role":"manager","token":"m-token","projects":[]}];self.client.headers["authorization"]="Bearer m-token"
|
|
with patch.dict("os.environ",{"XINGBANG_AUTH_USERS":json.dumps(users)}):self.assertEqual(self.client.get(self.root+"/workspace").json()["cases"],[])
|
|
def test_input_validation(self):
|
|
self.assertEqual(self.client.post(self.root+"/cases",json={"run_id":self.run["id"],"lease_end":"2020-01-01"}).status_code,200)
|
|
self.assertEqual(self.client.post(self.root+f"/cases/{self.case['id']}/documents?version={self.case['version']}",files={"file":("bad.exe",b"x","application/octet-stream")}).status_code,422)
|
|
|
|
if __name__=="__main__":unittest.main()
|