mini-openclaw/backend/test_risk_agent_config.py

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