File size: 1,841 Bytes
f592f80 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 | import sys,unittest,threading,urllib.error
from pathlib import Path
sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'engine'))
import providers,atlas_analysis
class Attempts(unittest.TestCase):
def test_partial_api_text_available_for_transform_only(self):
import io,json
original=providers.secure_urlopen
try:
for provider,data in [('OpenAI',{'status':'incomplete','output':[{'content':[{'type':'output_text','text':'Partial draft'}]}]}),('VT ARC',{'choices':[{'finish_reason':'length','message':{'content':'Partial draft'}}]})]:
providers.secure_urlopen=lambda *a,**k:io.StringIO(json.dumps(data))
with self.assertRaises(ValueError):providers.request(provider,'fake','key','prompt','input')
result=providers.request(provider,'fake','key','prompt','input',allow_partial=True)
self.assertEqual(result['text'],'Partial draft');self.assertTrue(result['truncated'])
finally:providers.secure_urlopen=original
def test_local_invalid_assessment_is_not_retried(self):
class Engine:
calls=0
def generate(self,*args,**kwargs):
self.calls+=1;return {'text':'{}','tokens':1,'truncated':False}
e=Engine()
with self.assertRaises(atlas_analysis.AnalysisError):providers.evaluate('A passage.','Local Qwen','Qwen','',threading.Event(),e)
self.assertEqual(e.calls,1)
def test_api_transport_failure_is_not_retried(self):
calls=[];original=atlas_analysis.secure_urlopen
def fail(*a,**k):
calls.append(1);raise urllib.error.HTTPError('https://example.invalid',503,'Unavailable',{},None)
atlas_analysis.secure_urlopen=fail
try:
with self.assertRaises(atlas_analysis.RequestError):providers.evaluate('A passage.','OpenAI',atlas_analysis.MODEL,'test',threading.Event())
self.assertEqual(len(calls),1)
finally:atlas_analysis.secure_urlopen=original
if __name__=='__main__':unittest.main()
|