| 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() |
|
|
