Something went wrong. Try again.
Native control, inspection, UI automation and evidence capture for a running Witcher 3 process.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390"""Safety contracts tested with a fake native world, never a second live owner."""import copyimport jsonfrom pathlib import Pathimport socketimport tempfileimport threadingimport timeimport unittestfrom unittest.mock import patch
from witcher_devkit import debugger, nativefrom witcher_devkit.debug_journal import Journal, durable_json, verify, recover_torn_tailfrom witcher_devkit.debug_runtime import OperationUncertain, Runtime, validate
class World: def __init__(self): self.ticks=0; self.calls=[]; self.hp=100; self.gen=1; self.paused=True; self.events=[] def snapshot(self): return dict(ok=True,generation=self.gen,data=dict(ready=True,ticks=self.ticks,input_ticks=0,remaining_ticks=0, world=dict(devkit_paused=self.paused,player=dict(hp=self.hp,alive=self.hp>0)))) def call(self,command,args='',timeout=5): self.calls.append((command,args)) if command=='snapshot': return self.snapshot() if command in ('advance','input'): self.ticks+=int(args.split('|')[-1]); self.hp-=1 if command=='event_page': return dict(ok=True,generation=self.gen,data=dict(latest=len(self.events),oldest=1,events=self.events[int(args.split('|')[1]):])) return dict(ok=True,generation=self.gen,data={})
class DebuggerTests(unittest.TestCase): def setUp(self): self.tmp=tempfile.TemporaryDirectory(); self.addCleanup(self.tmp.cleanup) self.root=Path(self.tmp.name); self.journal=Journal(self.root/'journal') self.world=World(); self.cancel=threading.Event(); self.runtime=Runtime(self.journal,self.cancel) self.patches=[patch.object(native,'call',side_effect=self.world.call), patch.object(native,'state',return_value={'uncertain':None}),patch.object(native,'ROOT',self.root), patch.object(native,'STATE_ROOT',self.root)] for p in self.patches: p.start(); self.addCleanup(p.stop) self.receipt=dict(token='secret',session_id='s',broker=debugger.identity()) self.broker=debugger.Broker(self.receipt,self.journal)
def request(self,method,**args): return self.broker.request(dict(method=method,token='secret',**args)) def submit(self,opid='a',**extra): return self.request('submit',kind='step',args={'ticks':3},operation_id=opid,wall_seconds=5,expected_generation=None,**extra) def done(self): end=time.monotonic()+3 while self.broker.active and time.monotonic()<end: time.sleep(.005) self.assertIsNone(self.broker.active) return self.broker.operations['a']
def test_idempotent_operation_and_conflicting_arguments(self): self.submit(); self.assertEqual(self.done()['status'],'completed') self.submit(); self.assertEqual(self.world.calls.count(('advance','3')),1) with self.assertRaisesRegex(ValueError,'operation_id_conflict'): self.request('submit',kind='step',args={'ticks':4},operation_id='a',wall_seconds=5,expected_generation=None)
def test_busy_cancel_single_writer(self): entered=threading.Event(); release=threading.Event() def work(*args): entered.set(); release.wait(2); return {'cancel_seen':self.broker.cancelled.is_set()} with patch.object(self.broker.runtime,'execute',side_effect=work): self.submit(); self.assertTrue(entered.wait(1)) with self.assertRaisesRegex(ValueError,'busy'): self.submit('b') self.assertTrue(self.request('cancel',operation_id='a')['cancel_requested']) self.assertEqual(self.request('status')['active_operation'],'a') release.set(); self.assertTrue(self.done()['result']['cancel_seen'])
def test_fsync_intent_exists_before_native_dispatch(self): def work(*args): rows=self.journal.rows(); self.assertEqual(rows[-1]['kind'],'intent'); return {} with patch.object(self.broker.runtime,'execute',side_effect=work): self.submit(); self.done() self.assertEqual(verify(self.root/'journal'/'manifest.json')['pending_operations'],[])
def test_timeout_retains_uncertainty_and_never_replays(self): def failure(*args): self.broker.runtime.mutation_dispatched=True raise TimeoutError('lost ack') with patch.object(self.broker.runtime,'execute',side_effect=failure): self.submit(); self.assertEqual(self.done()['status'],'uncertain') with self.assertRaisesRegex(ValueError,'uncertain'): self.submit('b') self.assertEqual(self.submit()['operation']['status'],'uncertain')
def test_restart_preserves_pending_operation(self): self.journal.append('intent',dict(operation_id='old',status='accepted',fingerprint='f')) recovered=debugger.Broker(self.receipt,Journal(self.journal.folder)) self.assertEqual(recovered.operations['old']['outcome'],'unknown') self.assertTrue(recovered.uncertain)
def test_close_allowed_with_host_uncertainty_and_certain_native_receipt(self): def failure(*args): self.broker.runtime.mutation_dispatched=True raise TimeoutError('lost completion observation') with patch.object(self.broker.runtime,'execute',side_effect=failure): self.submit() self.assertEqual(self.done()['status'],'uncertain') with patch.object(native,'stop',return_value=dict(ok=True,state='stopped')): self.request('submit',kind='close',args={},operation_id='b',wall_seconds=5,expected_generation=None) end=time.monotonic()+3 while self.broker.active and time.monotonic()<end: time.sleep(.005) self.assertEqual(self.broker.operations['b']['status'],'completed') self.assertEqual(self.broker.operations['a']['status'],'uncertain') self.assertTrue(self.broker.uncertain) self.assertTrue(self.broker.closed)
def test_close_stays_blocked_for_unresolved_native_delivery(self): self.broker.uncertain=True with patch.object(native,'state',return_value={'uncertain':{'command':'input','request':7}}), \ patch.object(native,'stop',side_effect=AssertionError('must not exit')): with self.assertRaisesRegex(ValueError,'unresolved native delivery blocks close'): self.request('submit',kind='close',args={},operation_id='b',wall_seconds=5,expected_generation=None) self.assertNotIn('b',self.broker.operations) self.assertTrue(self.broker.uncertain)
def test_close_stays_blocked_when_native_receipt_cannot_be_verified(self): self.broker.uncertain=True with patch.object(native,'state',side_effect=RuntimeError('bad receipt')): with self.assertRaisesRegex(ValueError,'native receipt unavailable'): self.request('submit',kind='close',args={},operation_id='b',wall_seconds=5,expected_generation=None)
def test_close_does_not_assume_missing_native_uncertainty_field_is_certain(self): self.broker.uncertain=True with patch.object(native,'state',return_value={}): with self.assertRaisesRegex(ValueError,'native receipt unavailable'): self.request('submit',kind='close',args={},operation_id='b',wall_seconds=5,expected_generation=None)
def test_explicit_uncertainty_does_not_depend_on_error_text_or_mutation_flag(self): with patch.object(self.broker.runtime,'execute',side_effect=OperationUncertain('incomplete compound evidence')): self.submit() self.assertEqual(self.done()['status'],'uncertain') self.assertTrue(self.broker.uncertain)
def test_cancel_remains_a_request_not_a_native_uncertainty_bypass(self): self.broker.uncertain=True self.broker.active='active-op' result=self.request('cancel',operation_id='active-op') self.assertTrue(result['cancel_requested']) self.assertTrue(self.broker.cancelled.is_set()) self.assertTrue(self.broker.uncertain) self.assertEqual(self.world.calls,[]) self.broker.active=None
def test_stale_generation_rejects_without_action(self): self.request('submit',kind='step',args={'ticks':3},operation_id='a',wall_seconds=5,expected_generation=2) self.assertEqual(self.done()['status'],'rejected') self.assertFalse(any(c=='advance' for c,a in self.world.calls))
def test_token_and_legacy_ownership(self): with self.assertRaisesRegex(ValueError,'token'): self.broker.request({'method':'status','token':'wrong'}) with patch.object(debugger,'descriptor',return_value={'broker':{'pid':-1},'session_id':'owned'}), patch.object(debugger,'identity',return_value={'pid':-1}): with self.assertRaisesRegex(RuntimeError,'session_owned'): debugger.reject_legacy()
def test_input_is_single_dispatch_and_delivery_not_effect(self): result=self.runtime.execute('step',dict(ticks=3,input='0|0|0|0|AttackLight|3'),5) self.assertEqual(self.world.calls.count(('input','0|0|0|0|AttackLight|3')),1) self.assertTrue(result['input_delivered']); self.assertFalse(result['effect_verified']) self.assertEqual(result['ticks_observed'],3)
def test_cancellation_before_dispatch(self): self.cancel.set(); r=self.runtime.execute('step',{'ticks':3},5) self.assertEqual(r['stop_reason'],'cancelled') self.assertFalse(any(c=='advance' for c,a in self.world.calls))
def test_terminal_precedence_prevents_any_new_action(self): self.world.hp=0 for kind,args in [('step',{'ticks':3}),('step',{'ticks':3,'input':'0|0|0|0|Dodge|3'}), ('until',{'ticks':3,'condition':{'kind':'player_dead'}})]: r=self.runtime.execute(kind,args,5) self.assertEqual(r['stop_reason'],'player_death'); self.assertEqual(r['ticks_observed'],0) self.assertFalse(any(c in ('advance','input') for c,a in self.world.calls))
def test_completed_step_retains_collection_failure_separately(self): calls=0; original=self.world.call def call(command,args='',timeout=5): nonlocal calls if command=='event_page': calls+=1 if calls>1: raise TimeoutError('event page unavailable after completed step') return original(command,args,timeout) with patch.object(native,'call',side_effect=call): result=self.runtime.execute('step',{'ticks':3},5) self.assertEqual(result['ticks_observed'],3) self.assertEqual(result['stop_reason'],'observation_gap') self.assertFalse(result['events']['complete']); self.assertTrue(result['events']['errors']) self.assertTrue(result['after']['data']['world']['devkit_paused'])
def test_actor_death_precedes_condition_and_budget(self): original=self.world.call def call(command,args='',timeout=5): if command=='actor': return {'ok':True,'generation':1,'data':{'id':2,'alive':False}} return original(command,args,timeout) with patch.object(native,'call',side_effect=call): r=self.runtime.execute('until',{'ticks':3,'condition':{'kind':'actor_dead','generation':1,'actor_id':2}},5) self.assertEqual(r['stop_reason'],'actor_death'); self.assertEqual(r['ticks_observed'],0) self.assertFalse(any(c=='advance' for c,a in self.world.calls))
def test_until_initial_true_does_not_advance(self): r=self.runtime.execute('until',dict(ticks=10,condition={'kind':'player_health_below','value':101}),5) self.assertEqual(r['ticks_observed'],0); self.assertEqual(r['stop_reason'],'condition')
def test_until_uses_step_bound_and_observed_state(self): r=self.runtime.execute('until',dict(ticks=10,step=2,condition={'kind':'player_health_below','value':98}),5) self.assertEqual(r['ticks_observed'],6); self.assertEqual(r['stop_reason'],'condition') self.assertEqual(r['pause_tick'],6)
def test_event_baseline_excludes_historical_contact(self): self.world.events=[dict(sequence=1,kind='contact',time=9)] r=self.runtime.execute('until',dict(ticks=2,condition={'kind':'event','event_kind':'contact'}),5) self.assertEqual(r['stop_reason'],'tick_budget')
def test_generation_change_stops_without_further_actions(self): old=self.world.call def changing(command,args='',timeout=5): r=old(command,args,timeout) if command=='advance': self.world.gen+=1 return r with patch.object(native,'call',side_effect=changing): with self.assertRaisesRegex(RuntimeError,'generation_changed'): self.runtime.execute('step',{'ticks':3},5) self.assertEqual(self.world.calls.count(('advance','3')),1)
def test_torn_journal_and_tamper_are_not_valid_evidence(self): self.journal.append('note',{}); self.journal.manifest() with self.journal.path.open('ab') as f: f.write(b'{') with self.assertRaisesRegex(ValueError,'torn_tail'): Journal(self.journal.folder) with self.assertRaisesRegex(ValueError,'mismatch'): verify(self.journal.folder/'manifest.json')
def test_concurrent_journal_writers_allocate_sequences_atomically(self): from witcher_devkit import debug_journal original=debug_journal.encode errors=[] def slow_encode(value): time.sleep(.001) # Release the GIL after sequence allocation. return original(value) def writer(index): try: for n in range(20): self.journal.append('note',dict(writer=index,index=n)) except Exception as error: errors.append(error) with patch.object(debug_journal,'encode',side_effect=slow_encode): threads=[threading.Thread(target=writer,args=(n,)) for n in range(6)] for thread in threads: thread.start() for thread in threads: thread.join(5) self.assertFalse(errors) self.assertTrue(all(not thread.is_alive() for thread in threads)) rows=Journal(self.journal.folder).rows() self.assertEqual([row['sequence'] for row in rows],list(range(1,121))) self.journal.manifest() self.assertTrue(verify(self.journal.folder/'manifest.json')['ok'])
def test_corrupt_journal_cannot_block_post_exit_settings_cleanup(self): self.journal.path.write_bytes(b'{"sequence":1,"kind":"note","payload":{}}\n' b'{"sequence":1,"kind":"note","payload":{}}\n') raw=self.journal.path.read_bytes() receipt=dict(self.receipt,folder=str(self.journal.folder)) with patch('witcher_devkit.debug_recovery.closed_cleanup',return_value=dict(ok=True,process_confirmed_closed=True)) as cleanup: result=debugger.recover_closed(receipt) cleanup.assert_called_once_with(receipt) self.assertTrue(result['cleanup_ok']) self.assertFalse(result['journal_integrity']) self.assertIn('journal_sequence_gap',result['journal_error']) self.assertEqual(self.journal.path.read_bytes(),raw) self.assertFalse(self.world.calls) with patch.object(debugger,'identity',return_value=None): status=debugger.Client(receipt).request('status') self.assertEqual(status['state'],'closed') self.assertFalse(status['recovery']['journal_integrity'])
def test_recovery_preserves_torn_tail_and_marks_pending_unknown(self): self.journal.append('intent',{'operation_id':'lost','fingerprint':'f','kind':'step','status':'accepted'}) with self.journal.path.open('ab') as f: f.write(b'{"partial":') original=self.journal.path.read_bytes() journal=recover_torn_tail(self.journal.folder) backup=next(self.journal.folder.glob('journal-torn-*.jsonl')) self.assertEqual(backup.read_bytes(),original) broker=debugger.Broker(self.receipt,journal) self.assertTrue(broker.uncertain) self.assertEqual(Journal(journal.folder).operations['lost']['status'],'interrupted') journal.manifest(); self.assertEqual(verify(journal.folder/'manifest.json')['pending_operations'],[])
def test_quota_refuses_intent_before_native_work(self): self.journal.quota=self.journal.reserve with self.assertRaisesRegex(OSError,'journal_quota'): self.submit() self.assertEqual(self.world.calls,[])
def test_validation_rejects_unbounded_and_unknown_conditions(self): for args in ({'ticks':0},{'ticks':601},{'ticks':True},{'ticks':1,'input':'0|0|0|0|Guard|1'}): with self.assertRaises(ValueError): validate('step',args) with self.assertRaises(ValueError): validate('until',{'ticks':1,'condition':{'kind':'eval','code':'arbitrary'}})
def test_load_waits_for_save_list_and_dispatches_once(self): import hashlib docs=self.root/'docs'; (docs/'gamesaves').mkdir(parents=True) (docs/'gamesaves'/'Fixture.SAV').write_bytes(b'save') calls=[]; lists=iter([[],['fixture']]); gen=iter([1,None,2,2]) def call(command,args='',timeout=5): calls.append((command,args)) if command=='saves': return {'ok':True,'generation':1,'data':next(lists)} if command=='snapshot': generation=next(gen) if generation is None: raise TimeoutError('read during loading') self.world.gen=generation; return self.world.snapshot() return {'ok':True,'generation':2,'data':{}} with patch.object(native.legacy.manage,'DOCS',docs), patch.object(native,'call',side_effect=call): r=self.runtime.execute('load',{'name':'fixture','sha256':hashlib.sha256(b'save').hexdigest()},3) self.assertEqual(calls.count(('load','fixture')),1) self.assertEqual(r['snapshot']['generation'],2)
def test_read_timeout_does_not_poison_certain_session(self): with patch.object(self.broker.runtime,'execute',side_effect=TimeoutError('observation timed out')): self.request('submit',kind='inspect',args={'view':'world'},operation_id='a',wall_seconds=2,expected_generation=None) self.assertEqual(self.done()['status'],'rejected') self.assertFalse(self.broker.uncertain)
def test_handoff_requires_pause_and_invalidates_old_token(self): self.request('submit',kind='handoff',args={},operation_id='a',wall_seconds=5,expected_generation=None) self.assertTrue(self.done()['result']['token_rotated']) with self.assertRaisesRegex(ValueError,'token'): self.request('status') result=self.broker.request(dict(method='status',token=self.broker.receipt['token'])) self.assertEqual(result['state'],'paused')
def test_wall_budget_stops_certain_running_work(self): running=False def call(command,args='',timeout=5): nonlocal running self.world.calls.append((command,args)) if command=='event_page': return {'ok':True,'generation':1,'data':{'latest':0,'oldest':1,'events':[]}} if command=='advance': running=True if command=='release': running=False if command=='snapshot': s=self.world.snapshot() if running: self.world.ticks+=1 s['data']['remaining_ticks']=600 s['data']['world']['devkit_paused']=False return s return {'ok':True,'generation':1,'data':{}} with patch.object(native,'call',side_effect=call): r=self.runtime.execute('run',{'ticks':600},.25) self.assertEqual(r['stop_reason'],'wall_budget') self.assertTrue(r['input_release_verified']) self.assertEqual(self.world.calls.count(('advance','600')),1) self.assertIn(('release',''),self.world.calls)
def test_record_window_survives_owner_restart(self): self.request('submit',kind='record-start',args={},operation_id='a',wall_seconds=5,expected_generation=None) window=self.done()['result']['window_id'] recovered=debugger.Broker(self.receipt,Journal(self.journal.folder)) self.assertEqual(recovered.runtime.window_id,window) result=recovered.runtime.execute('record-stop',{},5) self.assertEqual(result['window_id'],window) self.assertFalse(result['recording'])
def test_dead_owner_still_blocks_legacy_game_access(self): receipt={'broker':{'pid':999,'start':'x'},'session_id':'old','process':{'pid':123}} with patch.object(debugger,'descriptor',return_value=receipt), patch.object(debugger,'identity',return_value=None), patch.object(native.legacy,'process_identity',return_value={'pid':123}): with self.assertRaisesRegex(RuntimeError,'session_owned'): debugger.reject_legacy()
def test_protocol_and_request_fields_fail_before_native_work(self): with self.assertRaisesRegex(ValueError,'unsupported_protocol_version'): self.broker.request({'protocol_version':2,'method':'status','token':'secret'}) with self.assertRaisesRegex(ValueError,'unknown operation request fields'): self.request('submit',kind='step',args={'ticks':1},operation_id='a',wall_seconds=1,extra=True) self.assertEqual(self.world.calls,[]) with self.assertRaises(ValueError): validate('inspect',{'view':'profile'}) with self.assertRaises(ValueError): validate('step',{'ticks':2,'input':''})
def test_condition_precedes_same_boundary_cancellation(self): original=self.world.call def call(command,args='',timeout=5): r=original(command,args,timeout) if command=='advance': self.cancel.set() return r with patch.object(native,'call',side_effect=call): r=self.runtime.execute('until',{'ticks':10,'condition':{'kind':'player_health_below','value':100}},5) self.assertEqual(r['stop_reason'],'condition'); self.assertEqual(r['ticks_observed'],1)
def test_real_socket_client_result_lookup(self): path=str(self.root/'debug.sock') with debugger.Server(path,debugger.Handler) as server: server.broker=self.broker thread=threading.Thread(target=server.serve_forever,daemon=True); thread.start() try: c=debugger.Client(dict(socket=path,token='secret',broker=debugger.identity())) result=c.submit('step',{'ticks':3},'a',5) self.assertEqual(result['operation']['operation_id'],'a') self.assertEqual(c.wait('a',3)['operation']['status'],'completed') finally: server.shutdown(); thread.join()
if __name__=='__main__': unittest.main()