115 lines
12 KiB
Python
115 lines
12 KiB
Python
"""AC84–90: permissions, hidden dependencies, immutable quality and mock provenance."""
|
||
import json,os,tempfile,unittest
|
||
from pathlib import Path
|
||
from unittest.mock import patch,MagicMock
|
||
from fastapi import FastAPI
|
||
from fastapi.testclient import TestClient
|
||
from api import workbench_configuration as api,lease_library as library_api
|
||
from services import assistant_store as store,component_registry as registry,catalog_configuration as catalog,trajectory_quality as quality,review_plan,lease_case_library as library
|
||
from services.project_assistant import ProjectAssistantService
|
||
from services.component_evaluation import evaluate
|
||
from services.prescreen_explanation import explain
|
||
BASE='/api/xingbang/risk-agent/project-assistant/library'
|
||
ADMIN={'X-Xingbang-Demo-Role':'admin'};MANAGER={'X-Xingbang-Demo-Role':'manager'}
|
||
class ConfigurationTests(unittest.TestCase):
|
||
def setUp(self):
|
||
self.temp=tempfile.TemporaryDirectory(ignore_cleanup_errors=True);self.root=Path(self.temp.name);self.svc=ProjectAssistantService(self.root/'db.sqlite3');library.initialize(self.svc.db_path)
|
||
with self.svc.connect() as db:catalog.seed_fact_source(db)
|
||
self.patches=[patch.object(api,'service',lambda:self.svc),patch.object(library_api,'service',lambda:self.svc),patch.object(quality,'PATH',self.root/'quality.private.json'),patch.dict(os.environ,{'XINGBANG_AUTH_USERS':'','XINGBANG_REQUIRE_AUTH':'0','XINGBANG_DEMO_ROLE_LOGIN':'1','TYPESAFE_API_KEY':''})]
|
||
for p in self.patches:p.start()
|
||
app=FastAPI();app.include_router(library_api.router);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_hidden_new_assembly_old_version_preserved(self):
|
||
with self.svc.connect() as db:
|
||
old=registry.resolve(db,'corporate.cashflow');original=store.digest(old);catalog.set_visibility(db,old,'admin',True)
|
||
with self.assertRaisesRegex(ValueError,'COMPONENT_HIDDEN'):review_plan.assemble(db,'corporate.prescreen')
|
||
self.assertEqual(store.digest(registry.resolve(db,'corporate.cashflow')),original)
|
||
with self.assertRaisesRegex(ValueError,'COMPONENT_HIDDEN'):review_plan.propose_plan(db,dict({k:v for k,v in registry.resolve(db,'corporate.prescreen').items() if k!='allowed_tools'},id='new.plan'),'admin','DEMO-A')
|
||
catalog.set_visibility(db,old,'admin',False);self.assertEqual(review_plan.assemble(db,'corporate.prescreen')['plan']['id'],'corporate.prescreen')
|
||
def test_inventory_scope_and_role(self):
|
||
with self.svc.connect() as db:
|
||
for project_id,key in [('DEMO-A','private.skill'),('DEMO-B','other.skill')]:registry.draft(db,{'id':key,'name':'现金流草稿','kind':'skill','engine':'required','inputs':['facility.amount'],'version':'1.0.0','project_id':project_id},'admin')
|
||
catalog.set_visibility(db,registry.resolve(db,'corporate.cashflow'),'admin',True)
|
||
def query(headers):return self.client.post(BASE+'/inventory',headers=headers,json={'project_id':'DEMO-A','message':'我有几个skills？'}).json()['items']
|
||
manager=query(MANAGER);admin=query(ADMIN)
|
||
self.assertNotIn('private.skill@1.0.0',[r['id'] for r in manager]);self.assertNotIn('corporate.cashflow',[r['id'] for r in manager])
|
||
self.assertIn('private.skill@1.0.0',[r['id'] for r in admin]);self.assertNotIn('other.skill@1.0.0',[r['id'] for r in admin])
|
||
self.assertEqual(self.client.post(BASE+'/inventory',headers=MANAGER,json={'project_id':'NOT-ALLOWED','message':'skills'}).status_code,403)
|
||
def test_template_create_evaluate_independent_publish(self):
|
||
body={'project_id':'DEMO-A','kind':'template','name':'精简预审','content':{'sections':[{'id':'overview','title':'项目概况','dependencies':['*']}]}}
|
||
self.assertEqual(self.client.post(BASE+'/configuration/cards',headers=MANAGER,json=body).status_code,403)
|
||
result=self.client.post(BASE+'/configuration/cards',headers=ADMIN,json=body);self.assertEqual(result.status_code,201,result.text);draft=result.json()
|
||
with self.svc.connect() as db:
|
||
self.assertTrue(evaluate(db,draft['id'],'test')['passed'])
|
||
with self.assertRaises(PermissionError):registry.release(db,draft['id'],'demo-admin')
|
||
self.assertEqual(registry.release(db,draft['id'],'other-admin')['sections'][0]['title'],'项目概况')
|
||
path=BASE+'/cards/'+draft['component_id']+'/visibility'
|
||
self.assertEqual(self.client.patch(path,headers=MANAGER,json={'hidden':True}).status_code,403)
|
||
self.assertEqual(self.client.patch(path,headers=ADMIN,json={'hidden':True}).status_code,200)
|
||
def test_data_card_reject_write(self):
|
||
body={'project_id':'DEMO-A','kind':'data_card','name':'财务事实','content':{'source_id':'source.project-facts','sql':'SELECT * FROM finance WHERE project_id=:project_id','tables':['finance'],'mapping':{'facility.amount':'amount'}}}
|
||
result=self.client.post(BASE+'/configuration/cards',headers=ADMIN,json=body);self.assertEqual(result.status_code,201,result.text)
|
||
with self.svc.connect() as db:self.assertTrue(evaluate(db,result.json()['id'],'test')['passed'])
|
||
body['content']['sql']='DELETE FROM finance';self.assertEqual(self.client.post(BASE+'/configuration/cards',headers=ADMIN,json=body).status_code,422)
|
||
def test_natural_case_migration_idempotent(self):
|
||
with self.svc.connect() as db:before=library.read_query(db,'SELECT * FROM lease_cases ORDER BY id')
|
||
library.initialize(self.svc.db_path)
|
||
with self.svc.connect() as db:after=library.read_query(db,'SELECT * FROM lease_cases ORDER BY id')
|
||
self.assertEqual(before,after);self.assertEqual(len({r['lessee'] for r in after}),240);self.assertTrue(all('模拟' not in r['lessee'] and r['data_domain']=='demo' for r in after))
|
||
self.assertTrue(all('补充' in r['supplements'] and r['total_assets']>r['total_liabilities'] for r in after))
|
||
def test_private_settings_and_address(self):
|
||
payload={'provider':'local','endpoint':'http://127.0.0.1:11434/v1','model':'qwen3:8b','api_key':'secret-test'}
|
||
self.assertEqual(self.client.put(BASE+'/quality/settings',headers=MANAGER,json=payload).status_code,403)
|
||
result=self.client.put(BASE+'/quality/settings',headers=ADMIN,json=payload);self.assertEqual(result.status_code,200);self.assertNotIn('secret-test',result.text);self.assertTrue(result.json()['key_configured'])
|
||
payload['endpoint']='https://evil.example/v1';result=self.client.put(BASE+'/quality/settings',headers=ADMIN,json=payload);self.assertEqual(result.status_code,422);self.assertEqual(result.json()['error']['code'],'CONFIGURATION_INVALID')
|
||
self.assertNotIn('api_key',self.client.get(BASE+'/quality/settings',headers=ADMIN).json())
|
||
def test_quality_disabled_immutable_and_cross_project(self):
|
||
run={'id':'run-test','project_id':'DEMO-A','status':'partial','steps':[{'step_id':'s1','status':'partial','skill':{'id':'corporate.cashflow'},'findings':[{'code':'c1','result':'pass','evidence_ids':[]}]}],'component_versions':[{'id':'corporate.cashflow','version':'1.0.0'}],'probability':None,'components':{'plan':{'scope':'demo'}}}
|
||
with self.svc.connect() as db:store.put(db,'run',run,immutable=True);baseline=store.digest(store.get(db,'run',run['id']))
|
||
result=self.client.post(BASE+'/projects/DEMO-A/quality',headers=MANAGER,json={'run_id':run['id']});self.assertEqual(result.status_code,201,result.text);body=result.json()
|
||
self.assertEqual(body['judgment']['status'],'skipped');self.assertEqual({c['code'] for c in body['checks']},{'CHECK_INCOMPLETE','PASS_WITHOUT_REFERENCE'})
|
||
with self.svc.connect() as db:self.assertEqual(store.digest(store.get(db,'run',run['id'])),baseline)
|
||
self.assertEqual(self.client.post(BASE+'/projects/DEMO-B/quality',headers=MANAGER,json={'run_id':run['id']}).status_code,404)
|
||
def test_jev_no_key_no_fallback(self):
|
||
config={'provider':'jev','revision':2,'model':'jev-latest','allow_external':True};run={'id':'r','project_id':'DEMO-A','status':'completed','components':{'plan':{'scope':'demo'}}}
|
||
with patch('services.trajectory_quality.httpx.Client') as transport:result=quality.inspect(run,config);transport.assert_not_called()
|
||
self.assertEqual(result['judgment']['status'],'failed');self.assertEqual(result['judgment']['answers'],{});self.assertIsNone(result['approval_probability'])
|
||
def test_local_and_jev_contract_validation(self):
|
||
state={'steps':[],'components':[],'status':'completed'};config={'provider':'local','model':'qwen3:8b','endpoint':'http://127.0.0.1:11434/v1'};response=MagicMock();response.json.return_value={'choices':[{'message':{'content':json.dumps({'missing':False,'unsupported':False,'deviation':True})}}]}
|
||
with patch('services.trajectory_quality.httpx.Client') as factory:
|
||
client=factory.return_value.__enter__.return_value;client.post.return_value=response
|
||
self.assertTrue(quality.invoke(config,state)['answers']['deviation']['value']);self.assertNotIn('tools',client.post.call_args.kwargs['json'])
|
||
response.json.return_value={'model':'jev-1.13.0','answers':{k:{'type':'noul','noul':0.1} for k in quality.QUESTIONS}};config={'provider':'jev','model':'jev-latest','api_key':'secret','allow_external':True}
|
||
self.assertEqual(quality.invoke(config,state)['model'],'jev-1.13.0');self.assertEqual(client.post.call_args.args[0],'https://api.typesafe.ai/v1/systemone')
|
||
response.json.return_value['answers']['missing']['noul']=float('nan')
|
||
with self.assertRaises(ValueError):quality.invoke(config,state)
|
||
def test_explanation_frozen_without_fake_contributions(self):
|
||
card={'id':'skill.one','name':'现金流','version':'2.0.0','code':'cash.missing'};run={'components':{'skills':[card]},'steps':[{'skill':card,'findings':[{'code':'cash.missing','result':'unknown','title':'缺少流水','evidence_ids':[]}]}],'input_snapshot_id':'snapshot-original'}
|
||
result=explain(run,{'dimension_results':[{'id':'skill.one','stage':0,'state':'unknown'}]})
|
||
self.assertIsNone(result['probability_contributions']);self.assertEqual(result['items'][0]['version'],'2.0.0');self.assertEqual(result['items'][0]['checks'][0]['result'],'unknown')
|
||
def test_malformed_card_content_is_422(self):
|
||
for content in ({'sections':['wrong']},{'sections':[]},{'sections':[{'id':'overview','title':None}]}):
|
||
result=self.client.post(BASE+'/configuration/cards',headers=ADMIN,json={'project_id':'DEMO-A','kind':'template','name':'bad','content':content})
|
||
self.assertEqual(result.status_code,422,result.text)
|
||
def test_frozen_report_template_survives_later_release(self):
|
||
from copy import deepcopy
|
||
with self.svc.connect() as db:
|
||
template=registry.resolve(db,'template.ai-pre-review');project=store.get(db,'project','DEMO-A')
|
||
snapshot=dict(deepcopy(project),id='snapshot-template');store.put(db,'snapshot',snapshot,immutable=True)
|
||
run={'id':'run-template','project_id':'DEMO-A','status':'completed','snapshot_id':snapshot['id'],'components':{'template':deepcopy(template)},'steps':[],'findings':[],'summary':{},'evidence':[]}
|
||
store.put(db,'run',run,immutable=True)
|
||
new=deepcopy({k:v for k,v in template.items() if k not in ('revision','content_hash')});new['version']='1.0.1';new['sections'][0]['title']='新版章节'
|
||
draft=registry.draft(db,new,'editor');self.assertTrue(evaluate(db,draft['id'],'independent')['passed']);registry.release(db,draft['id'],'publisher')
|
||
report=self.svc.create_report('DEMO-A',run['id'],'manager','template.ai-pre-review')
|
||
self.assertEqual(report['template']['version'],template['version']);self.assertEqual(report['sections'][0]['title'],template['sections'][0]['title'])
|
||
def test_builtin_data_card_reads_only_current_confirmed_facts(self):
|
||
from services.assistant_data import read_cards
|
||
with self.svc.connect() as db:
|
||
project=store.get(db,'project','DEMO-A');source=registry.resolve(db,'source.project-facts');card=registry.resolve(db,'data.initial-finance')
|
||
values,sources,datasets,errors=read_cards(project,{'legacy_cards':[dict(card,connection=source['id'])],'legacy_connections':[dict(source,data_domain='demo')]})
|
||
self.assertFalse(errors,errors);self.assertEqual(values['facility.amount'],project['values']['facility.amount']);self.assertEqual(len(datasets),1);self.assertEqual(values['financial.total_assets'],project['values']['financial.total_assets'])
|
||
if __name__=='__main__':unittest.main()
|