File size: 9,901 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 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 | import sys,os,tempfile,unittest,json,copy
from pathlib import Path
sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'engine'))
_TMP=tempfile.TemporaryDirectory();os.environ['AFFECTOR_SUPPORT']=_TMP.name
import bridge,geometry,documents
bridge.emit=lambda *args,**kwargs:None
class Bridge(unittest.TestCase):
def setUp(self):bridge.run=None;bridge.cancel.clear()
def test_checkpoint_and_verification(self):
original=bridge.providers.request;evaluate=bridge.providers.evaluate
calls=[]
def request(*a,**k):calls.append(a[4]);return {'text':'I felt happy.'}
def score(text,*a,**k):return {'row':{'valence':.8 if 'happy' in text else -.4,'direction':.5,'intensity':.6},'usage':{}}
bridge.providers.request=request;bridge.providers.evaluate=score
try:
g=geometry.new();g['points']=[{'t':0,'v':.8,'d':.5,'i':.6}]
q={'op':'transform','geometry':g,'text':'I felt sad.\n\nI felt lonely.','backend':'OpenAI','model':'fake','key':'SECRET-DO-NOT-SAVE','sampling':{'temperature':0,'top_p':.95,'top_k':40,'max_tokens':128}}
bridge.execute(q);self.assertEqual(len(calls),2)
p=Path(_TMP.name)/'saved.json';bridge.execute({'op':'save_run','path':str(p)})
self.assertNotIn('SECRET',p.read_text());bridge.run=None;bridge.execute({'op':'load_run','path':str(p)});bridge.execute(q);self.assertEqual(len(calls),2)
bridge.execute({'op':'verify','provider':'OpenAI','model':'fake','key':'SECRET'})
v=bridge.run['verification'];self.assertEqual(v['paired_count'],2);self.assertEqual(v['total_count'],2);self.assertEqual(v['after_similarity'],1);self.assertGreater(v['improvement'],0)
finally:bridge.providers.request=original;bridge.providers.evaluate=evaluate
def test_changed_rewrite_policy_starts_fresh_run(self):
original=bridge.providers.request;policy=bridge.REWRITE_POLICY;calls=[]
def rewrite(*a,**k):calls.append(a[3]);return {'text':'Revised.'}
bridge.providers.request=rewrite
try:
q={'op':'transform','geometry':geometry.new(),'text':'First.','backend':'OpenAI','model':'fake','sampling':{'temperature':0,'top_p':.95,'top_k':40,'max_tokens':128}}
bridge.execute(q);bridge.execute(q);self.assertEqual(len(calls),1)
bridge.REWRITE_POLICY='test-new-policy';bridge.execute(q);self.assertEqual(len(calls),2)
self.assertEqual(bridge.run['rewrite_policy'],'test-new-policy')
self.assertIn('Meaning, imagery, and events may change',calls[0])
self.assertIn('with no minimum length',calls[0]);self.assertIn('Aim for at most 6 characters',calls[0])
self.assertNotIn('60 and 100 words',calls[0])
finally:bridge.providers.request=original;bridge.REWRITE_POLICY=policy
def test_partial_rewrite_retained_and_capped(self):
original=bridge.providers.request
bridge.providers.request=lambda *a,**k: {'text':'New. This unfinished draft goes on', 'truncated':True}
try:
bridge.execute({'op':'transform','geometry':geometry.new(),'text':'Original.','backend':'OpenAI','model':'fake','sampling':{'temperature':0,'top_p':.95,'top_k':40,'max_tokens':128}})
r=bridge.run['records']['0']
self.assertEqual(r['text'],'New.')
self.assertEqual(len(r['attempts']),1)
self.assertIn('unfinished draft',r['attempts'][0]['text'])
self.assertEqual(r['character_limit'],11)
self.assertTrue(r['recovery'])
self.assertIn('warning',r);self.assertNotIn('original_retained',r)
finally:bridge.providers.request=original
def test_reset_discards_run_and_automatic_checkpoint_only(self):
checkpoint=bridge.SUPPORT/'last-run.json';checkpoint.write_text('{}')
saved=bridge.SUPPORT/'manual-transfer.json';saved.write_text('keep')
bridge.run={'old':'run'}
result=bridge.execute({'op':'reset_text'})
self.assertTrue(result['reset_text']);self.assertIsNone(bridge.run)
self.assertFalse(checkpoint.exists());self.assertEqual(saved.read_text(),'keep')
bridge.execute({'op':'reset_text'})
def test_rewrite_styles_are_applied_saved_and_restored(self):
original=bridge.providers.request;calls=[]
def request(*a,**k):calls.append(a[3]);return {'text':'Revised.'}
bridge.providers.request=request
try:
q={'op':'transform','geometry':geometry.new(),'text':'Original passage.','backend':'OpenAI','model':'fake','sampling':{'temperature':0,'top_p':.95,'top_k':40,'max_tokens':128},'rewrite_style':'Restrained'}
bridge.execute(q);bridge.execute(q);self.assertEqual(len(calls),1)
self.assertIn('Do not invent characters',calls[0])
p=bridge.SUPPORT/'style.json';bridge.execute({'op':'save_run','path':str(p)})
restored=bridge.execute({'op':'load_run','path':str(p)});self.assertEqual(restored['rewrite_style'],'Restrained')
q['rewrite_style']='Creative';bridge.execute(q);self.assertEqual(len(calls),2)
self.assertIn('Meaning, imagery, and events may change',calls[1]);self.assertEqual(bridge.run['rewrite_style'],'Creative')
finally:bridge.providers.request=original
def test_new_styles_round_trip_and_start_separate_runs(self):
original=bridge.providers.request;calls=[]
def request(*a,**k):calls.append(a[3]);return {'text':'Edited.'}
bridge.providers.request=request
try:
q={'op':'transform','geometry':geometry.new(),'text':'Original passage.','backend':'VT ARC','model':'fake','sampling':{'temperature':0,'top_p':.95,'top_k':40,'max_tokens':128}}
for style,phrase in [('Minimal','nearly all original wording'),('Balanced','modest figurative language')]:
q['rewrite_style']=style;bridge.execute(q)
self.assertIn(phrase,calls[-1]);self.assertEqual(bridge.run['rewrite_style'],style)
p=bridge.SUPPORT/'new-style.json';bridge.execute({'op':'save_run','path':str(p)})
self.assertEqual(bridge.execute({'op':'load_run','path':str(p)})['rewrite_style'],style)
bridge.execute(q)
self.assertEqual(len(calls),2)
q['rewrite_style']='Invalid'
with self.assertRaises(ValueError):bridge.execute(q)
finally:bridge.providers.request=original
def test_paragraph_allowance_carries_forward_and_resets(self):
text='A'*600+' '+'B'*150+'\n\u00a0\n'+'C'*20
spans=[(0,600),(601,751),(754,774)]
records={'0':{'text':'X'*657}}
allowance=documents.rewrite_allowance(text,spans,records,1)
self.assertEqual(allowance,844)
records['1']={'text':'Y'*allowance}
self.assertEqual(len(documents.assemble(text,spans,records).split('\n')[0]),2*751)
self.assertEqual(documents.rewrite_allowance(text,spans,records,2),40)
restored=json.loads(json.dumps(records))
self.assertEqual(documents.rewrite_allowance(text,spans,restored,1),844)
records['0']={'text':'A'*600,'original_retained':True}
self.assertEqual(documents.rewrite_allowance(text,spans,records,1),901)
def test_failed_unit_continues(self):
original=bridge.providers.request;calls=[]
def request(*a,**k):
calls.append(1)
if len(calls)==1:raise ValueError('Invalid rewrite')
return {'text':'New.'}
bridge.providers.request=request
try:
bridge.execute({'op':'transform','geometry':geometry.new(),'text':'First.\n\nSecond.','backend':'VT ARC','model':'fake','sampling':{'temperature':0,'top_p':.95,'top_k':40,'max_tokens':128}})
self.assertTrue(bridge.run['records']['0']['original_retained']);self.assertEqual(bridge.run['records']['1']['text'],'New.');self.assertTrue(bridge.run['complete'])
finally:bridge.providers.request=original
def test_transport_interruptions_continue(self):
original=bridge.providers.request;evaluate=bridge.providers.evaluate
errors=[ConnectionError('offline'),TimeoutError('timeout'),InterruptedError('connection interrupted'),bridge.providers.RequestError('HTTP 503')]
try:
for error in errors:
with self.subTest(error=type(error).__name__):
bridge.run=None;calls=[]
def request(*a,**k):
calls.append(1)
if len(calls)==1:raise error
return {'text':'New.'}
bridge.providers.request=request
bridge.execute({'op':'transform','geometry':geometry.new(),'text':'First.\n\nSecond.','backend':'OpenAI','model':'fake','sampling':{'temperature':0,'top_p':.95,'top_k':40,'max_tokens':128}})
self.assertEqual(len(calls),2);self.assertTrue(bridge.run['complete']);self.assertTrue(bridge.run['records']['0']['original_retained'])
assessed=[]
def assess(*a,**k):
assessed.append(1)
if len(assessed)==1:raise error
return {'row':{'valence':.2,'direction':.5,'intensity':.6},'usage':{}}
bridge.providers.evaluate=assess
bridge.execute({'op':'verify','provider':'OpenAI','model':'fake'})
v=bridge.run['verification'];self.assertEqual(len(assessed),4);self.assertTrue(v['complete']);self.assertEqual(v['failed_assessments'],1);self.assertEqual(v['paired_count'],1)
self.assertIn('before_error',v['scores']['0']);self.assertIn('after_fit',v['scores']['1'])
finally:bridge.providers.request=original;bridge.providers.evaluate=evaluate
def test_summary_distinguishes_rewrites_and_fallbacks(self):
text='One.\n\nTwo.\n\nThree.\n\nFour.'
data={'input':text,'spans':documents.spans(text),'records':{'0':{'text':'A rewrite.'},'1':{'text':'Two.'},'2':{'text':'Three.','error':'timeout','original_retained':True}}}
result=bridge.summarize(data)
self.assertEqual([result[k] for k in ['rewritten','unchanged','failed','pending']],[1,1,1,1])
self.assertIn('1 rewritten',result['message'])
def test_explicit_stop_still_stops(self):
original=bridge.providers.request;calls=[]
def request(*a,**k):
calls.append(1);bridge.cancel.set();raise InterruptedError('User stopped')
bridge.providers.request=request
try:
with self.assertRaises(InterruptedError):
bridge.execute({'op':'transform','geometry':geometry.new(),'text':'First.\n\nSecond.','backend':'OpenAI','model':'fake','sampling':{'temperature':0,'top_p':.95,'top_k':40,'max_tokens':128}})
self.assertEqual(len(calls),1);self.assertFalse(bridge.run['complete']);self.assertEqual(bridge.run['records'],{})
finally:bridge.providers.request=original;bridge.cancel.clear()
if __name__=='__main__':unittest.main()
|