82 lines
6.5 KiB
Python
82 lines
6.5 KiB
Python
"""AC78..83: SQLite feedback, model settings, private errors and folder conflicts."""
|
|
import os, tempfile, unittest, base64
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
from services import model_settings, assistant_model, assistant_store as store, meeting_feedback
|
|
from services.skill_packages import validate_files
|
|
from services.project_assistant import ProjectAssistantService
|
|
from api import lease_library
|
|
class CompletionTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp=tempfile.TemporaryDirectory(ignore_cleanup_errors=True);self.path=Path(self.temp.name)
|
|
self.svc=ProjectAssistantService(self.path/'assistant.sqlite3')
|
|
self.patches=[patch.object(model_settings,'PATH',self.path/'private.json'),patch.dict(os.environ,{'XINGBANG_LOCAL_MODEL_URL':'','XINGBANG_LOCAL_MODEL_KEY':'','XINGBANG_AUTH_USERS':'','XINGBANG_REQUIRE_AUTH':'0','XINGBANG_DEMO_ROLE_LOGIN':'1'}),patch.object(lease_library,'service',lambda:self.svc)]
|
|
for p in self.patches:p.start()
|
|
app=FastAPI();app.include_router(lease_library.router);self.client=TestClient(app)
|
|
self.base='/api/xingbang/risk-agent/project-assistant/library';self.admin={'X-Xingbang-Demo-Role':'admin'}
|
|
def tearDown(self):
|
|
self.client.close()
|
|
for p in reversed(self.patches):p.stop()
|
|
self.temp.cleanup()
|
|
def test_settings_persist_and_hide_key(self):
|
|
result=self.client.put(self.base+'/model-settings',headers=self.admin,json={'endpoint':'http://127.0.0.1:11434/v1','name':'qwen-test','api_key':'secret-testing-only'})
|
|
self.assertEqual(result.status_code,200);self.assertNotIn('secret-testing-only',result.text)
|
|
self.assertEqual(assistant_model.configuration()['name'],'qwen-test');self.assertEqual(model_settings.api_key(),'secret-testing-only')
|
|
self.assertEqual(self.client.get(self.base+'/model-settings',headers={'X-Xingbang-Demo-Role':'manager'}).status_code,403)
|
|
with self.svc.connect() as db:
|
|
events=db.execute("SELECT body FROM assistant_events WHERE body LIKE '%model.settings_updated%'").fetchall()
|
|
self.assertTrue(events);self.assertNotIn('secret-testing-only',str(events))
|
|
def test_bad_addresses_and_environment_lock(self):
|
|
for url in ['https://example.com/v1','http://user:secret@127.0.0.1/v1','http://127.0.0.1/v1?key=secret','file:///tmp/local']:
|
|
result=self.client.put(self.base+'/model-settings',headers=self.admin,json={'endpoint':url,'name':'qwen'})
|
|
self.assertEqual(result.status_code,422)
|
|
self.assertFalse(model_settings.PATH.exists())
|
|
with patch.dict(os.environ,{'XINGBANG_LOCAL_MODEL_URL':'http://localhost:11434/v1'}):
|
|
result=self.client.put(self.base+'/model-settings',headers=self.admin,json={'endpoint':'http://localhost:1234/v1','name':'qwen'})
|
|
self.assertEqual(result.status_code,409)
|
|
def test_probe_no_business_data_and_redaction(self):
|
|
model_settings.save('http://127.0.0.1:11434/v1','qwen','testing-key')
|
|
with patch('services.model_settings.httpx.Client') as client:
|
|
transport=client.return_value.__enter__.return_value
|
|
transport.post.return_value.json.return_value={'choices':[{'message':{'content':'OK'}}]}
|
|
self.assertTrue(model_settings.probe()['ok'])
|
|
self.assertEqual(transport.post.call_args.kwargs['json']['messages'],[{'role':'user','content':'Reply with OK.'}])
|
|
transport.post.side_effect=RuntimeError('testing-key private provider body')
|
|
result=model_settings.probe();self.assertFalse(result['ok']);self.assertNotIn('testing-key',str(result))
|
|
def pair(self,db,name,label=True,end='2026-01-01T08:00:00+00:00',supersedes=None,snapshot=None):
|
|
snapshot=snapshot or 'snapshot-'+name
|
|
outcome={'id':'outcome-'+name,'project_id':'DEMO-A','meeting_id':'meeting-'+name,'reviewed_snapshot_id':snapshot,'status':'matched','label':label,'occurred_at':'2026-01-02T08:00:00+00:00','supersedes_id':supersedes}
|
|
store.put(db,'review_outcome',outcome)
|
|
store.put(db,'run',{'id':'run-'+name,'project_id':'DEMO-A','input_snapshot_id':snapshot,'ended_at':end,'status':'completed'})
|
|
store.put(db,'prescreen_run',{'id':'run-'+name,'project_id':'DEMO-A','recommendation':'advance','data_domain':'demo','plan_ref':{'id':'plan','version':'1'}})
|
|
return outcome
|
|
def test_feedback_replaced_unobserved_and_late(self):
|
|
with self.svc.connect() as db:
|
|
self.pair(db,'approved');self.pair(db,'rejected',False)
|
|
self.pair(db,'late',end='2026-01-03T08:00:00+00:00');self.pair(db,'deferred',None)
|
|
old=self.pair(db,'old');self.pair(db,'corrected',False,supersedes=old['id'],snapshot=old['reviewed_snapshot_id'])
|
|
result=meeting_feedback.summarize(db,'DEMO-A')
|
|
self.assertEqual(result['observed'],3);self.assertAlmostEqual(result['advance_group']['rate'],1/3)
|
|
self.assertEqual(len(result['excluded']),3);self.assertEqual(result['probability_status'],'not_calibrated')
|
|
def test_ambiguous_empty_and_scope(self):
|
|
with self.svc.connect() as db:
|
|
self.pair(db,'a',snapshot='one');self.pair(db,'b',snapshot='one')
|
|
result=meeting_feedback.summarize(db,'DEMO-A')
|
|
self.assertEqual(result['observed'],0);self.assertIsNone(result['advance_group']['rate'])
|
|
self.assertEqual(self.client.get(self.base+'/projects/PRIVATE-PROJECT/feedback',headers=self.admin).status_code,403)
|
|
def test_case_conflicts(self):
|
|
skill=b'---\nname: sample\ndescription: demo\n---\nbody\n'
|
|
files=[{'path':p,'content_base64':base64.b64encode(c).decode()} for p,c in [('sample/SKILL.md',skill),('sample/references/A.md',b'a'),('sample/references/a.md',b'b')]]
|
|
with self.assertRaises(ValueError):validate_files(files)
|
|
|
|
def test_model_contract_semantic_normalization(self):
|
|
from services.lease_extraction import normalize_proposal
|
|
self.assertEqual(normalize_proposal({'operator':None},'skill')['operator'],'lt')
|
|
self.assertEqual(normalize_proposal({'operator':'<=','threshold':1.2},'rule_set'),{'operator':'lte','threshold':'1.2'})
|
|
self.assertIsNone(normalize_proposal({'operator':None},'rule_set')['operator'])
|
|
with self.assertRaises(ValueError):normalize_proposal({'kind':'skill'},'rule_set')
|
|
self.assertEqual(normalize_proposal({'kind':'rule_set','operator':'<'},'rule_set'),{'operator':'lt'})
|
|
with self.assertRaises(ValueError):normalize_proposal({'threshold':float('inf')},'rule_set')
|