142 lines
8.6 KiB
Python
142 lines
8.6 KiB
Python
|
"""AC-70..73/75: real SQLite selection and API/model boundaries."""
|
||
|
|
import os
|
||
|
|
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 lease_library as api
|
||
|
|
from services import lease_case_library as lib, assistant_store as store
|
||
|
|
from services.project_assistant import ProjectAssistantService
|
||
|
|
|
||
|
|
BASE = '/api/xingbang/risk-agent/project-assistant/library'
|
||
|
|
ADMIN = {'X-Xingbang-Demo-Role': 'admin'}
|
||
|
|
|
||
|
|
class LeaseLibraryTests(unittest.TestCase):
|
||
|
|
def setUp(self):
|
||
|
|
self.temp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True)
|
||
|
|
self.root = Path(self.temp.name)
|
||
|
|
self.svc = ProjectAssistantService(self.root / 'assistant.sqlite3')
|
||
|
|
lib.initialize(self.svc.db_path)
|
||
|
|
self.patches = [patch.object(api, 'service', lambda: self.svc),
|
||
|
|
patch('services.skill_packages.ROOT', self.root / 'packages'),
|
||
|
|
patch.dict(os.environ, {'XINGBANG_AUTH_USERS':'', 'XINGBANG_REQUIRE_AUTH':'0',
|
||
|
|
'XINGBANG_LOCAL_MODEL_URL':'', 'XINGBANG_DEMO_ROLE_LOGIN':'1'})]
|
||
|
|
for p in self.patches: p.start()
|
||
|
|
app = FastAPI(); app.include_router(api.router)
|
||
|
|
self.client = TestClient(app)
|
||
|
|
|
||
|
|
def tearDown(self):
|
||
|
|
self.client.close()
|
||
|
|
for p in reversed(self.patches): p.stop()
|
||
|
|
self.temp.cleanup()
|
||
|
|
|
||
|
|
def test_seed_is_durable_idempotent_and_corporate(self):
|
||
|
|
lib.initialize(self.svc.db_path)
|
||
|
|
with store.transaction(self.svc.db_path) as db:
|
||
|
|
rows = lib.read_query(db, 'SELECT * FROM lease_cases ORDER BY id')
|
||
|
|
self.assertEqual(len(rows), 240)
|
||
|
|
self.assertEqual({r['data_domain'] for r in rows}, {'demo'})
|
||
|
|
self.assertTrue(all(r['lessee'].endswith('有限公司') and r['asset'] and r['meeting_reason'] for r in rows))
|
||
|
|
self.assertEqual({r['outcome'] for r in rows}, set(lib.OUTCOMES))
|
||
|
|
import json
|
||
|
|
for row in rows:
|
||
|
|
self.assertGreater(row['total_assets'], row['total_liabilities'])
|
||
|
|
snapshot=json.loads(row['initial_snapshot'])
|
||
|
|
self.assertNotIn('outcome',snapshot)
|
||
|
|
self.assertEqual(snapshot['values']['facility.amount'],row['amount'])
|
||
|
|
|
||
|
|
def test_filters_sql_and_natural_agree(self):
|
||
|
|
with self.svc.connect() as db:
|
||
|
|
a = lib.preview(db, 'a', 'filters', filters={'industry':'制造业','outcome':'通过'})
|
||
|
|
b = lib.preview(db, 'a', 'natural', '筛选制造业已经通过的历史案例')
|
||
|
|
c = lib.preview(db, 'a', 'sql', "SELECT * FROM lease_cases WHERE industry='制造业' AND outcome='通过' ORDER BY id")
|
||
|
|
self.assertEqual(a['case_ids'], b['case_ids'])
|
||
|
|
self.assertEqual(a['case_ids'], c['case_ids'])
|
||
|
|
self.assertEqual(a['count'], 12)
|
||
|
|
|
||
|
|
def test_unknown_natural_is_rejected(self):
|
||
|
|
for text in ('制造业或者医疗健康', '没有通过的制造业', '风险小的项目', '金额大于2000万元'):
|
||
|
|
with self.subTest(text=text), self.assertRaises(lib.LibraryError): lib.natural_filters(text)
|
||
|
|
self.assertEqual(lib.natural_filters('制造业，金额不少于2000万元'), {'industry':'制造业','min_amount':2000})
|
||
|
|
|
||
|
|
def test_sql_rejects_writes_system_tables_and_fabrication(self):
|
||
|
|
statements = ['DELETE FROM lease_cases', 'PRAGMA table_info(lease_cases)',
|
||
|
|
'SELECT * FROM assistant_objects', 'SELECT * FROM sqlite_master',
|
||
|
|
"SELECT * FROM lease_cases; DELETE FROM lease_cases", "SELECT * FROM lease_cases UNION SELECT * FROM lease_cases",
|
||
|
|
"SELECT * FROM lease_cases WHERE load_extension('bad')=1", 'SELECT lessee FROM lease_cases']
|
||
|
|
with self.svc.connect() as db:
|
||
|
|
for sql in statements:
|
||
|
|
with self.subTest(sql=sql), self.assertRaises(lib.LibraryError): lib.read_query(db, sql)
|
||
|
|
self.assertEqual(len(lib.read_query(db,'SELECT * FROM lease_cases')),240)
|
||
|
|
self.assertTrue(store.listing(db,'project'), 'Authorizer must be reset after failure')
|
||
|
|
|
||
|
|
def test_confirmation_freezes_exact_scope_and_actor(self):
|
||
|
|
with self.svc.connect() as db:
|
||
|
|
preview = lib.preview(db,'a','filters',filters={'mode':'直接租赁'})
|
||
|
|
confirmed = lib.confirm(db,'a',preview['id'],preview['case_ids'][:2])
|
||
|
|
selection, rows = lib.selected_cases(db,'a',confirmed['id'])
|
||
|
|
self.assertEqual([r['id'] for r in rows],preview['case_ids'][:2])
|
||
|
|
for actor, ids in [('b',None), ('a',[]), ('a',['bad'])]:
|
||
|
|
with self.assertRaises(lib.LibraryError): lib.confirm(db,actor,preview['id'],ids)
|
||
|
|
with self.assertRaises(lib.LibraryError): lib.selected_cases(db,'b',confirmed['id'])
|
||
|
|
self.assertEqual(selection['source_hash'],store.digest(rows))
|
||
|
|
|
||
|
|
def test_preview_expiry_and_changed_cases_fail(self):
|
||
|
|
with self.svc.connect() as db:
|
||
|
|
p = lib.preview(db,'a','filters')
|
||
|
|
selection = lib.confirm(db,'a',p['id'],p['case_ids'][:1])
|
||
|
|
db.execute('UPDATE lease_cases SET amount=1 WHERE id=?',(selection['case_ids'][0],))
|
||
|
|
with self.assertRaises(lib.LibraryError): lib.selected_cases(db,'a',selection['id'])
|
||
|
|
with self.assertRaises(lib.LibraryError): lib.validate_selection({'created_by':'a','expires_at':'2000-01-01T00:00:00+00:00'},'a')
|
||
|
|
|
||
|
|
def test_api_manager_reads_but_cannot_process_and_reviewer_retired(self):
|
||
|
|
manager={'X-Xingbang-Demo-Role':'manager'}
|
||
|
|
self.assertEqual(self.client.get(BASE+'/cases',headers=manager).status_code,200)
|
||
|
|
self.assertEqual(self.client.post(BASE+'/queries',headers=manager,json={'mode':'filters'}).status_code,403)
|
||
|
|
self.assertEqual(self.client.get(BASE+'/cases',headers={'X-Xingbang-Demo-Role':'reviewer'}).status_code,403)
|
||
|
|
|
||
|
|
def confirmed(self):
|
||
|
|
preview=self.client.post(BASE+'/queries',headers=ADMIN,json={'mode':'filters','filters':{'industry':'制造业'}}).json()
|
||
|
|
return self.client.post(BASE+'/selections',headers=ADMIN,json={'preview_id':preview['preview_id'],'selected_ids':preview['case_ids'][:3]}).json()
|
||
|
|
|
||
|
|
def test_unconfigured_model_never_creates_candidate(self):
|
||
|
|
confirmed=self.confirmed()
|
||
|
|
result=self.client.post(BASE+'/extractions',headers=ADMIN,json={'project_id':'DEMO-A','selection_id':confirmed['id'],'kind':'skill'}).json()
|
||
|
|
self.assertEqual(result['status'],'failed')
|
||
|
|
self.assertIsNone(result['candidate_id'])
|
||
|
|
self.assertEqual(result['degradations'][0]['code'],'LIBRARY_MODEL_UNAVAILABLE')
|
||
|
|
with self.svc.connect() as db:
|
||
|
|
self.assertEqual(store.listing(db,'component_draft','DEMO-A'),[])
|
||
|
|
self.assertEqual(len(store.listing(db,'lease_extraction','DEMO-A')),1)
|
||
|
|
|
||
|
|
def test_model_skill_and_rule_are_distinct_governed_drafts(self):
|
||
|
|
confirmed=self.confirmed()
|
||
|
|
proposal={'title':'经营现金流核查','description':'制造业早期材料核查','action':'核对经营现金流材料及来源',
|
||
|
|
'exceptions':'无来源不作通过结论','field':'financial.operating_cashflow','operator':'lt','threshold':'100'}
|
||
|
|
with patch('services.lease_extraction.invoke',return_value=(proposal,{'model':'test-stub','engine':'scoped-agent'})) as call:
|
||
|
|
for kind in ('skill','rule_set'):
|
||
|
|
result=self.client.post(BASE+'/extractions',headers=ADMIN,json={'project_id':'DEMO-A','selection_id':confirmed['id'],'kind':kind}).json()
|
||
|
|
self.assertEqual(result['status'],'completed')
|
||
|
|
with self.svc.connect() as db:
|
||
|
|
draft=store.get(db,'component_draft',result['candidate_id'])
|
||
|
|
self.assertEqual(draft['origin'],'history_derived')
|
||
|
|
self.assertEqual(draft['source_evidence_ids'],confirmed['case_ids'])
|
||
|
|
self.assertEqual(draft['kind'],kind)
|
||
|
|
self.assertEqual(draft['status'],'draft')
|
||
|
|
if kind=='rule_set': self.assertEqual(draft['effect'],'advisory')
|
||
|
|
self.assertEqual(call.call_count,2)
|
||
|
|
self.assertEqual(len(call.call_args.args[2]),3)
|
||
|
|
|
||
|
|
def test_project_and_domain_are_server_guarded(self):
|
||
|
|
confirmed=self.confirmed()
|
||
|
|
result=self.client.post(BASE+'/extractions',headers=ADMIN,json={'project_id':'not-allowed','selection_id':confirmed['id'],'kind':'skill'})
|
||
|
|
self.assertEqual(result.status_code,403)
|
||
|
|
with self.svc.connect() as db:
|
||
|
|
project=store.get(db,'project','DEMO-A'); project['data_domain']='production'; store.put(db,'project',project)
|
||
|
|
result=self.client.post(BASE+'/extractions',headers=ADMIN,json={'project_id':'DEMO-A','selection_id':confirmed['id'],'kind':'skill'})
|
||
|
|
self.assertEqual(result.status_code,403)
|
||
|
|
|
||
|
|
if __name__=='__main__': unittest.main()
|