#!/usr/bin/env python3 """Fake-only input policy tests; never load/connect native X11 or send input.""" import importlib.util import os from pathlib import Path import sys import tempfile from types import SimpleNamespace import copy import json import unittest from unittest.mock import patch sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'tools')) import headless_input_proof as proof class Clock: def __init__(self): self.now = 0 def __call__(self): return self.now def sleep(self, seconds): self.now += seconds class FakeObserver: def __init__(self): self.events = []; self.polls = 0; self.fail = False def poll(self): self.polls += 1 if self.fail: raise proof.Refused('observer cancelled') if not self.events: self.events.append('READY') def fresh(self): self.poll() class FakeGuard: def __init__(self): self.calls = 0; self.fail = False def check(self): self.calls += 1 if self.fail: raise proof.Refused('focus lost') class FakeKeyboard: def __init__(self, observer): self.observer = observer; self.events = []; self.held = None; self.on_send = None def require_idle(self): if self.held: raise proof.Refused('key already held') def send(self, key, down): self.events.append((key, down)); self.held = key if down else None if key == 'Tab' and not down: self.observer.events.append('INVENTORY_OPEN' if len(self.observer.events) == 1 else 'INVENTORY_CLOSED') if self.on_send: self.on_send(key, down) class SequenceTests(unittest.TestCase): def setup_sequence(self): self.clock = Clock(); self.observer = FakeObserver(); self.guard = FakeGuard() self.keyboard = FakeKeyboard(self.observer); self.report = {'input_sent': False} def run_sequence(self): proof.sequence(self.guard, self.keyboard, self.observer, self.report, self.clock, self.clock.sleep) def test_fixed_pairs_and_native_handshake(self): self.setup_sequence(); self.run_sequence() self.assertEqual(self.keyboard.events, [('W', True), ('W', False), ('Tab', True), ('Tab', False), ('Tab', True), ('Tab', False)]) self.assertTrue(self.report['sequence_completed']) self.assertAlmostEqual(self.clock(), 3.4) self.assertGreater(self.guard.calls, len(self.keyboard.events)) def test_focus_loss_aborts_and_attempts_release_without_foreign_event(self): self.setup_sequence() self.keyboard.on_send = lambda key, down: setattr(self.guard, 'fail', True) if down else None with self.assertRaisesRegex(proof.Refused, 'focus lost'): self.run_sequence() self.assertEqual(self.keyboard.events, [('W', True)]) self.assertEqual(self.report['unconfirmed_held_key'], 'W') self.assertEqual(len(self.report['paired_release_errors']), 1) self.assertFalse(self.report['sequence_completed']) def test_observer_cancel_still_releases_on_proven_connection(self): self.setup_sequence() self.keyboard.on_send = lambda key, down: setattr(self.observer, 'fail', True) if down else None with self.assertRaisesRegex(proof.Refused, 'observer cancelled'): self.run_sequence() self.assertEqual(self.keyboard.events, [('W', True), ('W', False)]) self.assertIsNone(self.report['unconfirmed_held_key']) def test_uncertain_native_down_exception_releases(self): self.setup_sequence() def fail(key, down): if down: raise OSError('XTest returned unknown') self.keyboard.on_send = fail with self.assertRaises(OSError): self.run_sequence() self.assertEqual(self.keyboard.events, [('W', True), ('W', False)]) def test_missing_inventory_ack_never_sends_second_tab(self): self.setup_sequence() def erase(key, down): if key == 'Tab' and not down: self.observer.events = ['READY'] self.keyboard.on_send = erase with self.assertRaisesRegex(proof.Refused, 'INVENTORY_OPEN'): self.run_sequence() self.assertEqual(self.keyboard.events.count(('Tab', True)), 1) self.assertLess(self.clock(), 25) def test_final_native_key_state_is_required_before_completion(self): self.setup_sequence() calls = [] original = self.keyboard.require_idle def idle(): calls.append(True) original() if len(calls) == 7: raise proof.Refused('native key state not idle') self.keyboard.require_idle = idle with self.assertRaisesRegex(proof.Refused, 'native key state'): self.run_sequence() self.assertEqual(len(self.keyboard.events), 6) self.assertFalse(self.report['sequence_completed']) self.assertEqual(len(calls), 7) def test_no_ready_no_input(self): self.setup_sequence(); self.observer.poll = lambda: None with self.assertRaisesRegex(proof.Refused, 'READY'): self.run_sequence() self.assertEqual(self.keyboard.events, []) def test_existing_held_key_refuses_without_releasing_it(self): self.setup_sequence(); self.keyboard.held = 'Other' with self.assertRaisesRegex(proof.Refused, 'already held'): self.run_sequence() self.assertEqual(self.keyboard.events, []) class ReleaseStateTests(unittest.TestCase): def test_busy_then_idle_polls_without_resending_input(self): clock = Clock(); guard = FakeGuard(); report = {}; states = [True, True, False] def idle(): if states.pop(0): raise proof.KeyStateBusy('still held') keyboard = SimpleNamespace(require_idle=idle) proof.confirm_release(guard, keyboard, report, 'W', clock, clock.sleep) self.assertEqual(guard.calls, 3) self.assertEqual(report['release_observations'][0]['attempts'], 3) self.assertTrue(report['release_observations'][0]['confirmed']) self.assertAlmostEqual(clock(), .05) def test_persistent_state_remains_bounded_and_unconfirmed(self): clock = Clock(); report = {} def idle(): raise proof.KeyStateBusy('still held') with self.assertRaisesRegex(proof.Refused, '350ms'): proof.confirm_release(FakeGuard(), SimpleNamespace(require_idle=idle), report, 'W', clock, clock.sleep) self.assertLessEqual(clock(), .35) self.assertFalse(report['release_observations'][0]['confirmed']) def test_focus_loss_and_native_query_error_are_never_retried(self): for focus_lost in (False, True): clock = Clock(); guard = FakeGuard(); guard.fail = focus_lost; calls = [] def idle(): calls.append(True) raise proof.Refused('native query failed') with self.assertRaises(proof.Refused): proof.confirm_release(guard, SimpleNamespace(require_idle=idle), {}, 'W', clock, clock.sleep) self.assertEqual(len(calls), 0 if focus_lost else 1) self.assertEqual(clock(), 0) def test_tab_release_has_one_fresh_guard_after_its_hold(self): clock = Clock(); observer = FakeObserver(); guard = FakeGuard(); keyboard = FakeKeyboard(observer) original = guard.check def costly_guard(): original(); clock.sleep(.10) guard.check = costly_guard; report = {'input_sent': False} proof.sequence(guard, keyboard, observer, report, clock, clock.sleep) for down, up in ((2, 3), (4, 5)): self.assertAlmostEqual(report['events'][up]['monotonic'] - report['events'][down]['monotonic'], .15) self.assertTrue(report['sequence_completed']) def test_delayed_w_release_keeps_exact_six_events(self): clock = Clock(); observer = FakeObserver(); guard = FakeGuard(); keyboard = FakeKeyboard(observer) original = keyboard.require_idle; calls = [] def delayed(): calls.append(True) if len(calls) == 2: raise proof.KeyStateBusy('delayed W state') original() keyboard.require_idle = delayed; report = {'input_sent': False} proof.sequence(guard, keyboard, observer, report, clock, clock.sleep) self.assertEqual(len(keyboard.events), 6) self.assertEqual(report['release_observations'][0]['attempts'], 2) self.assertTrue(report['all_attempted_presses_released']) self.assertTrue(report['sequence_completed']) class ObserverTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory(); self.path = Path(self.tmp.name) / 'console.log' self.path.write_text('ERR INPUT OBSERVE: run=test sample=1 real=1 world=1 READY\n') self.clock = Clock(); self.observer = proof.Observer(self.path, 'test', self.clock) def tearDown(self): self.observer.close(); self.tmp.cleanup() def append(self, text): with self.path.open('a') as f: f.write(text) def record(self, sample, real, detail): self.append(f'SCRIPT : ERR INPUT OBSERVE: run=test sample={sample} real={real} world={real}.000 {detail}\n') def test_old_ready_does_not_authorize_and_new_exact_sequence_does(self): self.observer.poll(); self.assertEqual(self.observer.events, []) self.record(2, 20, 'SAMPLE origin=0 0 0'); self.record(2, 20, 'READY') self.observer.fresh(); self.assertEqual(self.observer.events, ['READY']) self.record(3, 30, 'SAMPLE origin=0 0 0'); self.record(3, 30, 'INVENTORY_OPEN') self.observer.fresh(); self.record(4, 40, 'INVENTORY_CLOSED'); self.observer.poll() self.assertEqual(self.observer.events, ['READY', 'INVENTORY_OPEN', 'INVENTORY_CLOSED']) def test_stale_and_cancellation_fail_closed(self): self.record(2, 20, 'SAMPLE x'); self.record(2, 20, 'READY'); self.observer.fresh() self.clock.now = 1.01 with self.assertRaisesRegex(proof.Refused, 'stale'): self.observer.fresh() self.record(3, 30, 'CANCELLED original-admission-lost') with self.assertRaisesRegex(proof.Refused, 'ended'): self.observer.poll() def test_reverse_clock_or_repeated_sample_refused(self): self.record(2, 20, 'SAMPLE x'); self.observer.poll(); self.record(2, 20, 'SAMPLE x') with self.assertRaisesRegex(proof.Refused, 'Repeated'): self.observer.poll() def test_replace_truncate_and_matching_malformed_refused(self): self.path.write_text('') with self.assertRaisesRegex(proof.Refused, 'truncated'): self.observer.poll() def test_different_nonce_ignored(self): self.append('ERR INPUT OBSERVE: run=foreign sample=2 real=20 world=20 READY\n') self.observer.poll(); self.assertEqual(self.observer.raw, []) class GuardTests(unittest.TestCase): def setUp(self): self.args = SimpleNamespace(host_display=':0', prefix=Path('/tmp/p/pfx'), game_executable=Path('/tmp/game'), profile=Path('/tmp/profile'), nonce='test') self.bound = {'display': ':2'} self.owner = {'window': 100, 'pid': 20, 'server_pid': 20, 'process': {'pid': 20, 'start': 99}} self.query = SimpleNamespace(prop=lambda *a: 20, server_pid=lambda *a: 20) def test_full_live_gates_rechecked_before_acceptance(self): with patch.object(proof.display, 'binding', return_value=self.bound) as binding, patch.object(proof.display, 'owned_window', return_value=self.owner), patch.object(proof, 'game_arguments') as argv, patch.object(proof.display, 'input_target', return_value={'verified': True}): proof.Guard(self.args, self.bound, self.query, self.owner).check() self.assertEqual(binding.call_count, 2); argv.assert_called_once() def test_binding_window_focus_failures(self): for bad in ('binding', 'window', 'focus'): with self.subTest(bad=bad), patch.object(proof.display, 'binding', return_value={} if bad == 'binding' else self.bound), patch.object(proof.display, 'owned_window', return_value=None if bad == 'window' else self.owner), patch.object(proof, 'game_arguments'), patch.object(proof.display, 'input_target', return_value={'verified': bad != 'focus'}): with self.assertRaises(proof.Refused): proof.Guard(self.args, self.bound, self.query, self.owner).check() def test_native_argv_profile_nonce_are_exact_not_prefix_matches(self): native = {'pid': 20, 'start': 99, 'args': ['exe', '-errOrdinaryInputObserve', 'test', '-errOrdinaryScenario', 'core', '-profile', 'Z:/tmp/profile']} with patch.object(proof.display, 'process', return_value=native), patch.object(proof.display, 'identity', return_value=self.owner['process']): proof.game_arguments(self.owner, self.args) native['args'][-1] += '-neighbour' with self.assertRaisesRegex(proof.Refused, 'profile'): proof.game_arguments(self.owner, self.args) def test_opt_in_required(self): with self.assertRaises(SystemExit), patch('sys.stderr'): proof.arguments([]) class SupervisionTests(unittest.TestCase): def test_only_exact_child_identity_and_argv_can_receive_stop(self): record = {'identity': {'pid': 222, 'start': 999}, 'command': ['python', 'owned-wrapper']} native = {'pid': 222, 'start': 999, 'uid': os.getuid(), 'args': record['command']} with patch.object(proof.display, 'identity', side_effect=lambda p: {'pid': p['pid'], 'start': p['start']}): for change in ({}, {'start': 1000}, {'args': ['other']}): sent = [] accepted = proof.stop_exact_child(record, lambda p: {**native, **change}, lambda *v: sent.append(v)) self.assertEqual(accepted, not bool(change)) self.assertEqual(len(sent), 0 if change else 1) def test_timeout_records_uncertain_press_in_exact_report(self): import json with tempfile.TemporaryDirectory() as folder: path = Path(folder) / 'proof.json'; path.write_text('') info = path.stat() record = {'path': str(path), 'device': info.st_dev, 'inode': info.st_ino, 'metadata': {'schema': 1, 'input_sent': False}} proof.supervision_failure(record, 'W', 'timeout') result = json.loads(path.read_text()) self.assertTrue(result['supervision_failed']); self.assertTrue(result['input_attempted']) self.assertFalse(result['sequence_completed']); self.assertFalse(result['all_attempted_presses_released']) self.assertEqual(result['unconfirmed_held_key'], 'W') record['inode'] += 1 with self.assertRaises(proof.Refused): proof.supervision_failure(record, None, 'wrong inode') def test_supervisor_preserves_worker_error_and_uncertain_release(self): import json with tempfile.TemporaryDirectory() as folder: path = Path(folder) / 'proof.json' path.write_text(json.dumps({'error': 'actual keymap error', 'all_attempted_presses_released': False, 'unconfirmed_held_key': 'Tab'})) info = path.stat() record = {'path': str(path), 'device': info.st_dev, 'inode': info.st_ino, 'metadata': {}} proof.supervision_failure(record, None, 'worker exited') result = json.loads(path.read_text()) self.assertEqual(result['error'], 'actual keymap error') self.assertEqual(result['supervision_error'], 'worker exited') self.assertFalse(result['all_attempted_presses_released']) self.assertEqual(result['unconfirmed_held_key'], 'Tab') def test_press_watchdog_terminates_worker_and_owned_wrapper_without_retry(self): events = [{'kind': 'report', 'metadata': {}}, {'kind': 'child', 'identity': {}}, {'kind': 'press-pending', 'key': 'W'}] class Pipe: def poll(self, *a): return bool(events) def recv(self): return events.pop(0) def close(self): pass class Process: pid = 123; exitcode = 0; alive = True; terminated = 0 def start(self): pass def is_alive(self): return self.alive def terminate(self): self.terminated += 1 def join(self, *a): self.alive = False def kill(self): self.alive = False child = Process(); pipe = Pipe() context = SimpleNamespace(Pipe=lambda **k: (pipe, pipe), Process=lambda **k: child) times = iter([0, 0, 0, 0, 2]) with patch.object(proof.multiprocessing, 'get_context', return_value=context), patch.object(proof.time, 'monotonic', side_effect=lambda: next(times)), patch.object(proof.signal, 'signal'), patch.object(proof, 'stop_exact_child') as stop, patch.object(proof, 'supervision_failure') as failed: self.assertEqual(proof.supervise([]), 1) self.assertEqual(child.terminated, 1); stop.assert_called_once() self.assertEqual(failed.call_args.args[1], 'W') def test_look_watchdog_stops_exact_child_without_key_or_mouse_retry(self): events = [{'kind': 'report', 'metadata': {}}, {'kind': 'child', 'identity': {}}, {'kind': 'look-pending'}] class Pipe: def poll(self, *a): return bool(events) def recv(self): return events.pop(0) def close(self): pass class Process: pid = 123; exitcode = 0; alive = True; terminated = 0 def start(self): pass def is_alive(self): return self.alive def terminate(self): self.terminated += 1 def join(self, *a): self.alive = False def kill(self): self.alive = False child = Process(); pipe = Pipe() context = SimpleNamespace(Pipe=lambda **k: (pipe, pipe), Process=lambda **k: child) times = iter([0, 0, 0, 0, 4.01]) with patch.object(proof.multiprocessing, 'get_context', return_value=context), patch.object(proof.time, 'monotonic', side_effect=lambda: next(times)), patch.object(proof.signal, 'signal'), patch.object(proof, 'stop_exact_child') as stop, patch.object(proof, 'supervision_failure') as failed: self.assertEqual(proof.supervise([]), 1) self.assertEqual(child.terminated, 1); stop.assert_called_once() self.assertIsNone(failed.call_args.args[1]) def test_wrapper_requires_matching_existing_readonly_wrapper(self): args = SimpleNamespace(command=['/bin/echo', 'bad'], console=None) with self.assertRaises(proof.Refused): proof.wrapper_command(args) args.command = [] with self.assertRaises(proof.Refused): proof.wrapper_command(args) args.console = Path('/tmp/exact/console.log') self.assertEqual(proof.wrapper_command(args), []) def test_ctypes_keyboard_uses_exact_xtest_signatures_and_no_native_library(self): class Function: def __init__(self, call): self.call = call def __call__(self, *args): return self.call(*args) xt = SimpleNamespace(XTestQueryExtension=Function(lambda *a: 1), XTestFakeKeyEvent=Function(lambda *a: 1)) x = SimpleNamespace(XKeysymToKeycode=Function(lambda d,k: 25 if k == 0x77 else 23), XKeycodeToKeysym=Function(lambda d,c,l: 0x77 if c == 25 else 0xff09), XQueryKeymap=Function(lambda *a: 1), XFlush=Function(lambda *a: 0)) with patch.object(proof.C, 'CDLL', return_value=xt): keyboard = proof.Keyboard(SimpleNamespace(x=x, d=123)) keyboard.require_idle(); keyboard.send('W', True); keyboard.send('W', False) self.assertEqual(xt.XTestFakeKeyEvent.argtypes, [proof.C.c_void_p, proof.C.c_uint, proof.C.c_int, proof.C.c_ulong]) self.assertEqual(keyboard.codes, {'W': 25, 'Tab': 23}) with self.assertRaises(proof.Refused): keyboard.send('Escape', True) class LookObserver(FakeObserver): def __init__(self): super().__init__(); self.events=['READY']; self.ready_sample=10 self.look_records=[{'kind':'basis','sample':1,'real_ms':100}]; self.input_samples=[] self.add(10,0); self.add(11,0) def add(self,index,angle,position=None): import math pos=position or [0.,0.,0.] self.look_records.append({'kind':'sample','sample':index,'real_ms':index*100, 'direction':[math.sin(math.radians(angle)),0.,math.cos(math.radians(angle))], 'camera':[0.,1.7,0.],'body':pos,'bodyForward':[0.,0.,1.],'turn':[0.,0.,0.]}) self.input_samples.append({'sample':index,'real_ms':index*100,'origin':pos,'forward':0., 'input':[0.,0.,0.],'velocity':[0.,0.,0.],'inventory':False}) class LookMouse: def __init__(self,observer,clock): self.observer=observer;self.clock=clock;self.events=[];self.ack=True;self.returned=True def move(self,dx,dy,guard): guard.check();self.events.append((dx,dy));self.clock.sleep(.01) if self.ack: self.observer.add(12 if dx>0 else 13,2 if dx>0 or not self.returned else 0) class LookTests(unittest.TestCase): def setup_look(self): self.clock=Clock();self.observer=LookObserver();self.guard=FakeGuard() self.keyboard=FakeKeyboard(self.observer);self.mouse=LookMouse(self.observer,self.clock);self.report={} def perform(self): proof.perform_look(self.guard,self.keyboard,self.observer,self.mouse,self.report,self.clock,self.clock.sleep) def test_two_native_acknowledgements_before_any_keyboard(self): self.setup_look();self.perform() p=self.report['look_proof'] self.assertTrue(p['completed']);self.assertEqual(p['baseline_samples'],[10,11]) self.assertEqual((p['response_sample'],p['return_sample']),(12,13)) self.assertEqual([(m['dx'],m['dy']) for m in p['motions']],[(24,0),(-24,0)]) self.assertEqual([m['after_sample'] for m in p['motions']],[11,12]) self.assertEqual(self.keyboard.events,[]);self.assertLessEqual(p['finished_monotonic']-p['started_monotonic'],4) def test_missing_native_rotation_no_blind_return_or_keyboard(self): self.setup_look();self.mouse.ack=False with self.assertRaisesRegex(proof.Refused,'acknowledgement|four seconds'):self.perform() self.assertEqual(self.mouse.events,[(24,0)]);self.assertEqual(self.keyboard.events,[]) self.assertFalse(self.report['look_proof']['completed']) def test_failed_return_is_not_assumed_from_opposite_pixels(self): self.setup_look();self.mouse.returned=False with self.assertRaises(proof.Refused):self.perform() self.assertEqual(self.mouse.events,[(24,0),(-24,0)]) self.assertFalse(self.report['look_proof']['completed']) def test_early_motion_inventory_or_cancel_refuses_before_mouse(self): for fault in ('origin','input','velocity','forward','inventory','cancel','clock'): self.setup_look() if fault=='origin': self.observer.look_records[-1]['body']=[1.,0.,0.];self.observer.input_samples[-1]['origin']=[1.,0.,0.] elif fault in ('input','velocity'):self.observer.input_samples[-1][fault]=[0.,0.,1.] elif fault=='forward':self.observer.input_samples[-1]['forward']=1. elif fault=='inventory':self.observer.input_samples[-1]['inventory']=True elif fault=='clock':self.observer.input_samples[-1]['real_ms']+=1 else:self.observer.look_records.append({'kind':'cancelled','sample':11,'real_ms':1100}) with self.subTest(fault=fault),self.assertRaises(proof.Refused):self.perform() self.assertEqual(self.mouse.events,[]) def test_default_sequence_has_no_look_or_mouse_and_optin_failure_blocks_w(self): self.setup_look();self.mouse.ack=False with self.assertRaises(proof.Refused):proof.sequence(self.guard,self.keyboard,self.observer,self.report,self.clock,self.clock.sleep,mouse=self.mouse) self.assertEqual(self.keyboard.events,[]) self.setup_look();proof.sequence(self.guard,self.keyboard,self.observer,self.report,self.clock,self.clock.sleep) self.assertNotIn('look_proof',self.report);self.assertEqual(len(self.keyboard.events),6) def test_strict_native_look_parser_and_vectors(self): prefix='SCRIPT : ERR LOOK OBSERVE: run=N sample=1 real=100 ' sample='SAMPLE direction=<0,0,1> camera=<0,1.7,0> body=<0,0,0> bodyForward=<0,0,1> turn=<0,0,0>' result=proof.parse_look_record(prefix+sample,'N');self.assertEqual(result['direction'],[0,0,1]) self.assertEqual(proof.parse_look_record(prefix+'BASIS native-player-camera-world-transform','N')['kind'],'basis') for invalid in (sample.replace('direction=<0,0,1>','direction=<0,0,0>'),sample.replace('camera=<0,1.7,0>','camera='),sample+' extra'): with self.assertRaises(proof.Refused):proof.parse_look_record(prefix+invalid,'N') with self.assertRaises(proof.Refused):proof.parse_look_record(prefix+sample,'foreign') self.assertAlmostEqual(proof.yaw_delta([.01745,0,-.99985],[-.01745,0,-.99985]),2,places=2) def test_stale_pose_and_sampling_gap_refuse_before_mouse(self): self.setup_look() self.observer.input_samples.append({**self.observer.input_samples[-1], 'sample': 17, 'real_ms': 1700}) with self.assertRaisesRegex(proof.Refused, 'stale'): self.perform() self.assertEqual(self.mouse.events, []) self.setup_look(); self.observer.add(13, 0) with self.assertRaisesRegex(proof.Refused, 'sampling gap'): self.perform() self.assertEqual(self.mouse.events, []) def test_real_observer_joins_pinned_console_and_keeps_later_menu_cancel(self): with tempfile.TemporaryDirectory() as folder: path=Path(folder)/'console.log'; path.write_text('old irrelevant lines\n') observer=proof.Observer(path,'N',clock=lambda: 1.,enable_look=True) try: with path.open('a') as out: out.write('ERR INPUT OBSERVE: run=N sample=1 real=100 world=100 SAMPLE origin=<0,0,0> forward=0 input=<0,0,0> velocity=<0,0,0> inventory=0\n') out.write('ERR INPUT OBSERVE: run=N sample=1 real=100 world=100 READY\n') out.write('ERR LOOK OBSERVE: run=N sample=1 real=100 BASIS native-player-camera-world-transform\n') out.write('ERR LOOK OBSERVE: run=N sample=1 real=100 SAMPLE direction=<0,0,1> camera=<0,1.7,0> body=<0,0,0> bodyForward=<0,0,1> turn=<0,0,0>\n') observer.fresh() self.assertEqual(observer.ready_sample,1) self.assertEqual(observer.input_samples[0]['origin'],observer.look_records[-1]['body']) self.assertEqual(observer.input_samples[0]['real_ms'],observer.look_records[-1]['real_ms']) with path.open('a') as out: out.write('ERR LOOK OBSERVE: run=N sample=2 real=200 CANCELLED menu-opened\n') out.write('ERR INPUT OBSERVE: run=N sample=2 real=200 world=200 SAMPLE origin=<0,0,0> forward=0 input=<0,0,0> velocity=<0,0,0> inventory=1\n') out.write('ERR INPUT OBSERVE: run=N sample=2 real=200 world=200 INVENTORY_OPEN\n') observer.fresh() self.assertEqual(observer.events,['READY','INVENTORY_OPEN']) self.assertEqual(observer.look_records[-1]['reason'],'menu-opened') with path.open('a') as out: out.write('ERR LOOK OBSERVE: run=N sample=3 real=300 SAMPLE direction=<0,0,1> camera=<0,1.7,0> body=<0,0,0> bodyForward=<0,0,1> turn=<0,0,0>\n') with self.assertRaisesRegex(proof.Refused,'after cancellation'):observer.poll() finally:observer.close() def test_real_main_report_binds_look_optin_even_before_console_failure(self): import json for enabled in (False,True): with self.subTest(enabled=enabled),tempfile.TemporaryDirectory() as folder: args=SimpleNamespace(enable_private_look_proof=enabled,nonce='N', report=Path(folder)/'proof.json',console=Path(folder)/'console.log', prefix=Path(folder)/'pfx',profile=Path(folder),game_executable=Path(folder)/'game',host_display=':0') with patch.object(proof,'arguments',return_value=args),patch.object(proof,'wrapper_command',return_value=[]),patch.object(proof,'console_path'),patch.object(proof.display,'validate_paths'),patch.object(proof.display,'binding',return_value={'display':':2'}),patch.object(proof,'Observer',side_effect=OSError('bounded test stop')),patch.object(proof.signal,'signal'): self.assertEqual(proof.main([]),1) report=json.loads(args.report.read_text()) self.assertIs(report['look_requested'],enabled) self.assertFalse(report['input_sent']) self.assertEqual(report['error'],'bounded test stop') def test_cli_mouse_optin_requires_keyboard_optin(self): with self.assertRaises(SystemExit),patch('sys.stderr'): proof.arguments(['--enable-private-look-proof']) class PointerTests(unittest.TestCase): def test_pointer_bounds_root_masks_and_child_pid(self): mouse=object.__new__(proof.RelativeMouse) mouse.owner={'window':100,'pid':9};mouse.query=SimpleNamespace(server_pid=lambda child:9 if child==77 else 88) target={'geometry':{'root':5},'focused_clipped_bounds':[0,0,100,100]} base={'root':5,'child':77,'root_x':50,'root_y':50,'mask':0} mouse.pointer=lambda:base.copy();mouse.validate_pointer(target,24,0);mouse.validate_pointer(target,-24,0) for change in ({'root':6},{'mask':1},{'mask':256},{'root_x':90},{'root_y':101},{'child':78}): mouse.pointer=lambda:{**base,**change} with self.subTest(change=change),self.assertRaises(proof.Refused):mouse.validate_pointer(target,24,0) def test_only_fixed_relative_events_use_same_connection(self): class Fn: def __init__(self,fn):self.fn=fn def __call__(self,*args):return self.fn(*args) events=[];xt=SimpleNamespace(XTestFakeRelativeMotionEvent=Fn(lambda *args:events.append(args) or 1)) x=SimpleNamespace(XQueryPointer=Fn(lambda *a:1),XFlush=lambda d:None) keyboard=SimpleNamespace(xt=xt,require_idle=lambda:None);query=SimpleNamespace(x=x,d=123) mouse=proof.RelativeMouse(query,keyboard,{'window':100,'pid':9});mouse.validate_pointer=lambda *a:None guard=SimpleNamespace(check=lambda:{'verified':True}) mouse.move(24,0,guard);mouse.move(-24,0,guard) self.assertEqual(events,[(123,24,0,0),(123,-24,0,0)]) self.assertEqual(xt.XTestFakeRelativeMotionEvent.argtypes,[proof.C.c_void_p,proof.C.c_int,proof.C.c_int,proof.C.c_ulong]) for dx,dy in ((25,0),(24,1),(True,0)): with self.assertRaises(proof.Refused):mouse.move(dx,dy,guard) class ApproachTests(unittest.TestCase): def test_explicit_mode_requires_look_and_matching_native_gate_nonce(self): argv=['--enable-private-keyboard-proof','--prefix','/tmp/pfx','--game-executable','/tmp/game', '--profile','/tmp/profile','--report','/tmp/report','--host-display',':0','--nonce','N'] with self.assertRaises(SystemExit),patch('sys.stderr'): proof.arguments(argv+['--enable-private-approach-proof']) self.assertFalse(proof.arguments(argv).enable_private_approach_proof) args=proof.arguments(argv+['--enable-private-look-proof','--enable-private-approach-proof']) native={'args':['-errOrdinaryInputObserve','N','-errOrdinaryLookObserve','N','-errOrdinaryGateObserve','N', '-errOrdinaryScenario','core','-profile','Z:/tmp/profile']} owner={'pid':123,'process':{'pid':123}} with patch.object(proof.display,'process',return_value=native),patch.object(proof.display,'identity',return_value=owner['process']): proof.game_arguments(owner,args) native['args'][5]='foreign' with self.assertRaisesRegex(proof.Refused,'gate observer nonce'):proof.game_arguments(owner,args) def test_variant_runs_look_then_route_without_default_w_tab_cycle(self): clock=Clock();observer=FakeObserver();observer.enable_approach=True;observer.ready_at=0 guard=FakeGuard();keyboard=FakeKeyboard(observer);report={};order=[];notifications=[] def look(*args):order.append('look');report['look_proof']={'completed':True};clock.sleep(.2) def route(g,k,o,m,r,c,s): self.assertEqual(order,['look']);order.append('approach') k.send('W',True);k.send('W',False) r['approach_proof']={'completed':True,'paired_release_errors':[],'unconfirmed_held_key':None} with patch.object(proof,'perform_look',side_effect=look),patch.object(proof,'run_approach',side_effect=route),patch.object(proof,'notify',side_effect=notifications.append): proof.sequence(guard,keyboard,observer,report,clock,clock.sleep,mouse=object(),approach=True) self.assertEqual(order,['look','approach']);self.assertEqual(keyboard.events,[('W',True),('W',False)]) self.assertTrue(report['sequence_completed']);self.assertTrue(report['all_attempted_presses_released']) self.assertEqual(notifications[0],{'kind':'approach-pending','deadline_monotonic':35}) self.assertEqual(notifications[-1],{'kind':'approach-finished','successful':True}) def test_failed_look_never_calls_route_and_keeps_uncertain_route_release(self): clock=Clock();o=FakeObserver();o.enable_approach=True;o.ready_at=0;k=FakeKeyboard(o);r={} with patch.object(proof,'perform_look',side_effect=proof.Refused('look refusal')),patch.object(proof,'run_approach') as route: with self.assertRaisesRegex(proof.Refused,'look refusal'): proof.sequence(FakeGuard(),k,o,r,clock,clock.sleep,mouse=object(),approach=True) route.assert_not_called();self.assertEqual(k.events,[]) def failed(*args): r['approach_proof']={'completed':False,'unconfirmed_held_key':'W','paired_release_errors':['ownership lost']} raise proof.Refused('route refusal') with patch.object(proof,'perform_look'),patch.object(proof,'run_approach',side_effect=failed): with self.assertRaisesRegex(proof.Refused,'route refusal'): proof.sequence(FakeGuard(),k,o,r,clock,clock.sleep,mouse=object(),approach=True) self.assertEqual(r['unconfirmed_held_key'],'W');self.assertFalse(r['all_attempted_presses_released']) self.assertFalse(r['sequence_completed']);self.assertEqual(r['paired_release_errors'],['ownership lost']) def test_original_ready_deadline_not_renewed_after_calibration(self): clock=Clock();clock.now=34;o=FakeObserver();o.enable_approach=True;o.ready_at=0;k=FakeKeyboard(o);r={} def route(*args):clock.sleep(2) with patch.object(proof,'perform_look'),patch.object(proof,'run_approach',side_effect=route): with self.assertRaisesRegex(proof.Refused,'Original approach deadline'): proof.sequence(FakeGuard(),k,o,r,clock,clock.sleep,mouse=object(),approach=True) self.assertFalse(r['sequence_completed']);self.assertEqual(k.events,[]) def test_w_only_adapter_rejects_other_keys_and_expiry_but_allows_release(self): clock=Clock();o=FakeObserver();k=FakeKeyboard(o);r={'events':[]} adapter=proof.ApproachKeyboard(k,r,clock,35) for key,down in [('F',True),('Tab',True),('W',1)]: with self.assertRaises(proof.Refused):adapter.send(key,down) adapter.send('W',True);clock.now=35 with self.assertRaisesRegex(proof.Refused,'deadline'):adapter.send('W',True) adapter.send('W',False) self.assertEqual(k.events,[('W',True),('W',False)]);self.assertTrue(r['input_sent']) def test_relative_approach_requires_explicit_capability_and_checks_pointer_and_deadline(self): class Fn: def __init__(self,fn):self.fn=fn def __call__(self,*args):return self.fn(*args) events=[];xt=SimpleNamespace(XTestFakeRelativeMotionEvent=Fn(lambda *a:events.append(a) or 1)) x=SimpleNamespace(XQueryPointer=Fn(lambda *a:1),XFlush=lambda d:None) keyboard=SimpleNamespace(xt=xt,require_idle=lambda:None);query=SimpleNamespace(x=x,d=123) mouse=proof.RelativeMouse(query,keyboard,{'window':100,'pid':9});clock=Clock();checks=[] mouse.validate_pointer=lambda *a:checks.append(a);guard=SimpleNamespace(check=lambda:{'verified':True}) with self.assertRaises(proof.Refused):mouse.approach_move(1,0,guard,deadline=35,clock=clock) mouse.enable_approach=True for dx,dy in [(1,0),(-128,0),(128,0)]:mouse.approach_move(dx,dy,guard,deadline=35,clock=clock) self.assertEqual(len(events),3);self.assertEqual(len(checks),6) for dx,dy in [(0,0),(129,0),(-129,0),(True,0),(1,1),(1,False)]: with self.assertRaises(proof.Refused):mouse.approach_move(dx,dy,guard,deadline=35,clock=clock) def delayed(*args):clock.now=35 mouse.validate_pointer=delayed with self.assertRaisesRegex(proof.Refused,'deadline'):mouse.approach_move(1,0,guard,deadline=35,clock=clock) self.assertEqual(len(events),3) def test_raw_join_retains_partial_pair_and_every_intermediate_snapshot(self): from test_ordinary_gate_evidence import fixture,NONCE lines=[s for s in fixture().splitlines() if 'CANCELLED' not in s] with tempfile.TemporaryDirectory() as folder: path=Path(folder)/'console.log';path.write_text('') clock=Clock();clock.now=10 observer=proof.Observer(path,NONCE,clock=clock,enable_look=True,enable_approach=True) try: # Split after SAMPLE but before GEOMETRY: pair29 is newest complete. with path.open('a') as out:out.write('\n'.join(lines[:-1])+'\n') first=observer.approach_snapshots();self.assertEqual([s['sample'] for s in first],[29]) self.assertEqual(first[0]['geometry']['gate'],(6090.,211.344,7154.)) clock.sleep(.1) with path.open('a') as out:out.write(lines[-1]+'\n') following=observer.approach_snapshots();self.assertEqual([s['sample'] for s in following],[30]) received=following[0]['received_monotonic'];clock.sleep(.1) repeated=observer.approach_snapshots();self.assertEqual(repeated,following) self.assertEqual(repeated[0]['received_monotonic'],received) clock.sleep(.5) with self.assertRaisesRegex(proof.Refused,'stale'):observer.approach_snapshots() finally:observer.close() with tempfile.TemporaryDirectory() as folder: path=Path(folder)/'console.log';path.write_text('');clock=Clock() observer=proof.Observer(path,NONCE,clock=clock,enable_look=True,enable_approach=True) try: pivot=next(i for i,s in enumerate(lines) if 'sample=27 ' in s) with path.open('a') as out:out.write('\n'.join(lines[:pivot])+'\n') self.assertEqual(observer.approach_snapshots()[-1]['sample'],26) clock.sleep(.1) with path.open('a') as out:out.write('\n'.join(lines[pivot:])+'\n') self.assertEqual([s['sample'] for s in observer.approach_snapshots()],[27,28,29,30]) finally:observer.close() def test_raw_join_refuses_gate_cancel_foreign_nonce_and_changed_geometry(self): from test_ordinary_gate_evidence import fixture,NONCE clean='\n'.join(s for s in fixture().splitlines() if 'CANCELLED' not in s)+'\n' for change in ('cancel','foreign','geometry','clock','action'): with self.subTest(change=change),tempfile.TemporaryDirectory() as folder: path=Path(folder)/'console.log';path.write_text('') observer=proof.Observer(path,NONCE,clock=lambda:10,enable_look=True,enable_approach=True) try: raw=clean if change=='cancel':raw+='ERR GATE OBSERVE: run='+NONCE+' sample=31 real=4410 CANCELLED native-player-or-prison-context-lost\n' if change=='foreign':raw=raw.replace('ERR GATE OBSERVE: run='+NONCE,'ERR GATE OBSERVE: run=foreign') if change=='geometry':raw=raw.replace('sample=30 real=4300 GEOMETRY body=', 'sample=30 real=4300 GEOMETRY body=').replace('sample=30 real=4300 GEOMETRY body=<6088.5,211.389,7158.5> context=<6092.77', 'sample=30 real=4300 GEOMETRY body=<6088.5,211.389,7158.5> context=<6093.77') if change=='action':raw=raw.replace('duration=4 pressed=0','duration=4 pressed=1') if change=='clock':raw=raw.replace('sample=30 real=4300 GEOMETRY','sample=30 real=4301 GEOMETRY') with path.open('a') as out:out.write(raw) with self.assertRaises(ValueError):observer.approach_snapshots() finally:observer.close() def test_adapter_requires_independent_look_acceptance_before_module(self): observer=SimpleNamespace(approach_data=lambda:('raw',{}),nonce='N');report={'look_proof':{}} with patch('ordinary_look_evidence.validate_look_evidence',return_value={'verified':False,'errors':['native mismatch']}),patch('prison_approach_proof.perform_approach') as run: with self.assertRaisesRegex(proof.Refused,'calibration refused'): proof.run_approach(None,None,observer,None,report,lambda:0,lambda s:None) run.assert_not_called() def test_real_calibration_adapter_preserves_clocks_caps_and_completion_owner(self): import math from test_ordinary_gate_evidence import fixture,NONCE raw='\n'.join(s for s in fixture().splitlines() if 'CANCELLED' not in s)+'\n' angle=math.radians(2) raw='\n'.join(line.replace('direction=<0,0,1>',f'direction=<{math.sin(angle)},0,{math.cos(angle)}>') if 'ERR LOOK OBSERVE:' in line and 'sample=22 ' in line else line for line in raw.splitlines())+'\n' with tempfile.TemporaryDirectory() as folder: path=Path(folder)/'console.log';path.write_text('');clock=Clock();clock.now=10 observer=proof.Observer(path,NONCE,clock=clock,enable_look=True,enable_approach=True) try: with path.open('a') as out:out.write(raw) observer.fresh();clock.sleep(.4) look={'requested':True,'completed':True,'started_monotonic':10.,'finished_monotonic':10.4, 'baseline_samples':[20,21],'response_sample':22,'return_sample':23, 'motions':[{'dx':24,'dy':0,'monotonic':10.1,'after_sample':21,'after_real_ms':3310}, {'dx':-24,'dy':0,'monotonic':10.3,'after_sample':22,'after_real_ms':3420}]} report={'look_proof':look,'events':[],'coarse_approach_requested':True};keyboard=FakeKeyboard(observer);notifications=[];motions=[] mouse=SimpleNamespace(approach_move=lambda *a,**k:motions.append((a,k))) def module(guard,k,m,observe,geometry,calibration,r,**kwargs): self.assertEqual(calibration['ready_monotonic'],10) self.assertEqual(calibration['ready_real_ms'],3091) self.assertEqual(calibration['observer_start_real_ms'],1000) self.assertEqual(kwargs['key_state_busy'],proof.KeyStateBusy) self.assertEqual(kwargs['profile'],'coarse') self.assertEqual(observe()[-1]['geometry'],geometry) self.assertEqual(geometry['nonce'],NONCE) with self.assertRaises(proof.Refused):k.send('F',True) k.send('W',True);k.send('W',False);m.move(12,0,guard) r['approach_proof']={'completed':True} kwargs['notify']({'kind':'approach-pending','deadline_monotonic':45}) kwargs['notify']({'kind':'approach-finished','successful':True}) with patch('prison_approach_proof.perform_approach',side_effect=module),patch.object(proof,'notify',side_effect=notifications.append): proof.run_approach(FakeGuard(),keyboard,observer,mouse,report,clock,clock.sleep) self.assertEqual([e['key'] for e in report['events']],['W','W']) self.assertEqual(motions[0][1]['deadline'],45) self.assertEqual(notifications,[{'kind':'approach-pending','deadline_monotonic':45}]) self.assertTrue(report['input_sent']) finally:observer.close() def test_actual_main_metadata_allows_exactly_one_module_attempt(self): import copy from test_prison_approach_proof import Scene notices=[] with tempfile.TemporaryDirectory() as folder: args=SimpleNamespace(enable_private_look_proof=True,enable_private_approach_proof=True,nonce='test', report=Path(folder)/'proof.json',console=Path(folder)/'console.log',prefix=Path(folder)/'pfx', profile=Path(folder),game_executable=Path(folder)/'game',host_display=':0') with patch.object(proof,'arguments',return_value=args),patch.object(proof,'wrapper_command',return_value=[]),patch.object(proof,'console_path'),patch.object(proof.display,'validate_paths'),patch.object(proof.display,'binding',return_value={'display':':2'}),patch.object(proof,'Observer',side_effect=OSError('metadata-only fixture stop')),patch.object(proof.signal,'signal'),patch.object(proof,'notify',side_effect=lambda event:notices.append(copy.deepcopy(event))): self.assertEqual(proof.main([]),1) metadata=next(event['metadata'] for event in notices if event['kind']=='report') self.assertTrue(metadata['approach_requested']);self.assertNotIn('approach_proof',metadata) self.assertEqual(set(metadata['approach_module_sha256']),set(proof.APPROACH_MODULES)) scene=Scene();self.assertTrue(scene.run(report=metadata)['completed']) with self.assertRaisesRegex(ValueError,'already exists'):scene.run(report=metadata) def test_final_serialization_compacts_success_and_failure_without_mutating_live_samples(self): import copy sample={'sample':42,'real_ms':5200,'received_monotonic':10.2,'origin':[1,2,3],'geometry':{'large':'pose'}} for completed in (True,False): report={'approach_requested':True,'approach_proof':{'completed':completed,'samples':[copy.deepcopy(sample)],'error':'kept'}} encoded=proof.report_for_write(report) self.assertEqual(encoded['approach_proof']['samples'],[{'sample':42,'real_ms':5200,'received_monotonic':10.2}]) self.assertEqual(report['approach_proof']['samples'],[sample]) self.assertEqual(encoded['approach_proof']['error'],'kept') ordinary={'approach_requested':False,'events':[]} self.assertIs(proof.report_for_write(ordinary),ordinary) def test_exact_module_hashes_are_filename_bound_and_missing_module_refuses(self): import hashlib with tempfile.TemporaryDirectory() as folder: root=Path(folder) for name in proof.APPROACH_MODULES:(root/name).write_text(name) with patch.object(proof,'__file__',str(root/'headless_input_proof.py')): values=proof.approach_module_hashes() self.assertEqual(set(values),set(proof.APPROACH_MODULES)) self.assertEqual(values['prison_approach_proof.py'],hashlib.sha256(b'prison_approach_proof.py').hexdigest()) (root/'prison_approach_proof.py').unlink() with self.assertRaises(OSError):proof.approach_module_hashes() def test_max_native_record_budget_compact_report_remains_below_one_mebibyte(self): import copy,json from test_ordinary_gate_evidence import fixture,NONCE template=fixture().splitlines() initial=[line for line in template if 'sample=0 ' in line] basis=[line for line in template if 'sample=1 ' in line and 'BASIS' in line] ordinary=[line for line in template if 'sample=2 ' in line] lines=list(initial) for index in range(1,451): t=1000+index*90;world=10000+index*90 for line in ordinary: line=line.replace('sample=2 real=1220','sample='+str(index)+' real='+str(t)).replace('world=10220','world='+str(world)) if index==1: for ack in basis: if ('ERR LOOK OBSERVE:' in line and 'ERR LOOK OBSERVE:' in ack) or ('SAMPLE phase=' in line and 'ERR GATE OBSERVE:' in ack): lines.append(ack.replace('real=1110','real='+str(t))) lines.append(line) if index==23:lines.append(f'ERR INPUT OBSERVE: run={NONCE} sample={index} real={t+1} world={world} READY') with tempfile.TemporaryDirectory() as folder: path=Path(folder)/'console.log';path.write_text('') observer=proof.Observer(path,NONCE,clock=lambda:10.,enable_look=True,enable_approach=True) try: with path.open('a') as out:out.write('\n'.join(lines)+'\n') while observer.offset=2:observer.events=['READY'] observer.poll=startup_poll def ready_windows(): self.assertGreaterEqual(len(polls),2,'Never bind a transient pre-READY window') self.assertEqual(observer.events,['READY']) return [{}] query.windows=ready_windows def sequence(guard,keyboard,o,report,**kw): self.assertIs(o,observer);self.assertTrue(kw['selection']);self.assertTrue(kw['alignment']);self.assertTrue(kw['approach']) self.assertNotIn('facing_proof',report);self.assertNotIn('action_selection_proof',report) self.assertEqual(set(report['selection_module_sha256']),set(proof.SELECTION_MODULES)) report.update(sequence_completed=True,events=[],paired_release_errors=[]) with patch.object(proof,'arguments',return_value=args),patch.object(proof,'wrapper_command',return_value=[]),patch.object(proof,'console_path'),patch.object(proof.display,'validate_paths'),patch.object(proof.display,'binding',return_value={'display':':2'}),patch.object(proof.display,'X11',return_value=query),patch.object(proof.display,'owned_window',return_value=owner),patch.object(proof.display,'input_target',return_value={'verified':True}),patch.object(proof,'Observer',autospec=True,return_value=observer) as ob,patch.object(proof,'Guard',autospec=True),patch.object(proof,'Keyboard',autospec=True),patch.object(proof,'RelativeMouse',autospec=True) as mouse,patch.object(proof,'sequence',side_effect=sequence),patch.object(proof.signal,'signal'): self.assertEqual(proof.main([]),0) self.assertTrue(ob.call_args.kwargs['enable_selection']);self.assertNotIn('selection',mouse.call_args.kwargs) self.assertTrue(mouse.call_args.kwargs['enable_coarse_approach']);self.assertNotIn('coarse_approach',ob.call_args.kwargs) archived=json.loads(args.report.read_text());self.assertTrue(archived['coarse_approach_requested']);self.assertTrue(archived['selection_requested']);self.assertEqual(archived['wheel_events'],[]) def test_one_native_wheel_pair_and_mask_roundtrip_no_retries(self): r=WheelRig();w=r.wheel();notices=[] with patch.object(proof,'notify',side_effect=notices.append):result=w.pulse(5,r,deadline=3) self.assertEqual(r.calls,[(5,1),(5,0)]);self.assertTrue(result['released']) self.assertLess(result['down_monotonic'],result['up_monotonic']);self.assertEqual(r.report['unconfirmed_held_button'],None) self.assertEqual([x['sent'] for x in r.report['wheel_events']],[True,True]);self.assertEqual(r.report['wheel_release_errors'],[]) self.assertEqual([e['kind'] for e in notices],['button-pending','button-release-confirmed']) self.assertEqual(w.xt.XTestFakeButtonEvent.argtypes,[proof.C.c_void_p,proof.C.c_uint,proof.C.c_int,proof.C.c_ulong]) for button in (5,4,True): with self.assertRaises(proof.Refused):w.pulse(button,r,deadline=3) self.assertEqual(len(r.calls),2) def test_wheel_capability_deadline_foreign_state_and_cleanup(self): for fault in ('disabled','wrong','expired_during_guard','foreign','down_error','up_error','sticky'): r=WheelRig();deadline=3 if fault=='foreign':r.mask=1<<8 if fault=='down_error':r.fail_down=True if fault=='up_error':r.fail_up=True if fault=='sticky':r.sticky=True if fault=='expired_during_guard':deadline=1.05 with self.subTest(fault=fault),self.assertRaises(proof.Refused): w=proof.SelectionWheel(r.mouse,r.report,r.clock,r.clock.sleep,deadline,enabled=fault!='disabled') w.pulse(4 if fault=='wrong' else 5,r,deadline=deadline) if fault in ('disabled','wrong','expired_during_guard','foreign'):self.assertEqual(r.calls,[]) else: self.assertEqual(r.calls,[(5,1),(5,0)]) self.assertTrue(r.report['wheel_release_errors']) if fault=='down_error': self.assertTrue(r.report['wheel_events'][-1]['finally']);self.assertIsNone(r.report['unconfirmed_held_button']) else:self.assertEqual(r.report['unconfirmed_held_button'],5) def test_captured_pointer_prevents_cleanup_writes(self): r=WheelRig();original=r.xt.XTestFakeButtonEvent.fn def send(*a):result=original(*a);r.lost=True;return result r.xt.XTestFakeButtonEvent.fn=send with self.assertRaises(proof.Refused):r.wheel().pulse(5,r,deadline=3) self.assertEqual(r.calls,[(5,1)]);self.assertEqual(r.report['unconfirmed_held_button'],5);self.assertTrue(r.report['wheel_release_errors']) def test_actual_pointer_validator_allows_only_own_button_mask_on_release(self): m=proof.RelativeMouse.__new__(proof.RelativeMouse);m.owner={'pid':7};m.query=SimpleNamespace(server_pid=lambda c:7) state={'root':1,'mask':1<<12,'root_x':5,'root_y':5,'child':0};m.pointer=lambda:state target={'geometry':{'root':1},'focused_clipped_bounds':[0,0,10,10]} with self.assertRaises(proof.Refused):m.validate_pointer(target,0,0) self.assertEqual(m.validate_pointer(target,0,0,allowed_mask=1<<12),state) state['mask']|=1<<8 with self.assertRaises(proof.Refused):m.validate_pointer(target,0,0,allowed_mask=1<<12) state['mask']=0;state['root_x']=10 with self.assertRaises(proof.Refused):m.validate_pointer(target,0,0,allowed_mask=1<<12) def test_choice_join_waits_for_exact_complete_line_and_retains_receipts(self): from test_ordinary_gate_evidence import fixture,NONCE from test_ordinary_gate_choices import line as choice_line from ordinary_gate_evidence import parse_gate_record lines=[] for line in fixture().splitlines(): if 'CANCELLED' in line:continue lines.append(line) record=parse_gate_record(line,NONCE) if record and record['kind']=='geometry': lines.append(choice_line(record['sample'],record['real_ms'],selected='none:-1',context=0,ordered='').replace('actions=2 permissions=2 shown=2','actions=0 permissions=0 shown=0')) with tempfile.TemporaryDirectory() as folder: path=Path(folder)/'console.log';path.write_text('');clock=Clock();clock.now=10 o=proof.Observer(path,NONCE,clock=clock,enable_look=True,enable_approach=True,enable_selection=True) try: with path.open('a') as f:f.write('\n'.join(lines[:-1])+'\n') first=o.approach_snapshots();self.assertEqual(first[-1]['sample'],29) self.assertNotIn(30,o.gate_received);clock.sleep(.1) with path.open('a') as f:f.write(lines[-1]+'\n') after=o.approach_snapshots();self.assertEqual([x['sample'] for x in after],[30]);receipt=after[0]['received_monotonic'] self.assertEqual(set(after[0]['choices']),{'nonce','sample','real_ms','selected_class','selected_id','actions'}) o.selection_previous=28;clock.sleep(.1) batch=o.selection_snapshots();self.assertEqual([x['sample'] for x in batch],[29,30]);self.assertEqual(batch[-1]['received_monotonic'],receipt) self.assertEqual(o.choice_previous,30);self.assertFalse(hasattr(o,'choice_records')) with path.open('a') as f:f.write(lines[-1]+'\n') with self.assertRaisesRegex(proof.Refused,'repeated'):o.poll() finally:o.close() def test_facing_selection_real_modules_through_one_wheel_adapter(self): from test_prison_action_selection_proof import ChoiceScene scene=ChoiceScene();report=scene.report;prior=copy.deepcopy(report['approach_proof']) report['look_proof']={'baseline_samples':[20,21],'response_sample':22};report['events']=[] o=SimpleNamespace(ready_at=2.,ready_real=32000,queued_real=30000,nonce='N',approach_data=lambda:('raw',{}), look_records=[{'kind':'sample','sample':21,'direction':scene.calibration['baseline_direction']},{'kind':'sample','sample':22,'direction':scene.calibration['response_direction']}],alignment_snapshots=scene.observe,alignment_previous=None) def observe_selection(): if o.selection_previous is not None:scene.last_read=o.selection_previous;o.selection_previous=None return scene.observe() o.selection_snapshots=observe_selection r=WheelRig();r.clock=scene.clock;mask=[0];calls=[] def event(d,button,down,delay): calls.append((button,down));scene.sleep(.001);mask[0]=(1<<12) if down else 0 if down:scene.selected_at=scene.now+.15 return 1 def position(target,dx,dy,allowed_mask=0): if mask[0]&~allowed_mask:raise proof.Refused('foreign mask') return {'mask':mask[0]} r.xt.XTestFakeButtonEvent.fn=event mouse=SimpleNamespace(xt=r.xt,query=r.query,keyboard=SimpleNamespace(require_idle=scene.idle),validate_pointer=position, alignment_move=lambda dx,dy,guard,deadline,clock:scene.move(dx,dy,guard,deadline=deadline)) notices=[] def facing_evidence(*args, profile='standard'): self.assertEqual(profile, 'standard') return {'verified':True,'last_sample':report['facing_proof']['final_sample']} with patch('prison_approach_evidence.validate_approach_evidence',return_value=scene.accepted),patch('prison_alignment_evidence.validate_facing_evidence',side_effect=facing_evidence),patch.object(proof,'notify',side_effect=notices.append): proof.run_selection(scene,SimpleNamespace(require_idle=scene.idle),o,mouse,report,scene.clock,scene.sleep) self.assertEqual(calls,[(5,1),(5,0)]);self.assertTrue(report['facing_proof']['completed']);self.assertTrue(report['action_selection_proof']['completed']) self.assertNotIn('alignment_proof',report);self.assertEqual(report['approach_proof'],prior);self.assertEqual(report['events'],[]) self.assertEqual(report['action_selection_proof']['deadline_monotonic'],report['facing_proof']['deadline_monotonic']) finished=[n for n in notices if n['kind']=='alignment-finished'];self.assertEqual(finished,[{'kind':'alignment-finished','successful':True}]) self.assertLess(scene.now,39.) def test_refused_facing_validator_never_allocates_wheel_or_finishes_lease(self): from test_prison_action_selection_proof import ChoiceScene scene=ChoiceScene();report=scene.report report['look_proof']={'baseline_samples':[20,21],'response_sample':22} o=SimpleNamespace(ready_at=2.,ready_real=32000,queued_real=30000,nonce='N',approach_data=lambda:('raw',{}),look_records=[{'kind':'sample','sample':21,'direction':scene.calibration['baseline_direction']},{'kind':'sample','sample':22,'direction':scene.calibration['response_direction']}],alignment_snapshots=scene.observe) mouse=SimpleNamespace(alignment_move=lambda dx,dy,guard,deadline,clock:scene.move(dx,dy,guard,deadline=deadline));notices=[] with patch('prison_approach_evidence.validate_approach_evidence',return_value=scene.accepted),patch('prison_alignment_evidence.validate_facing_evidence',return_value={'verified':False}),patch.object(proof,'SelectionWheel',side_effect=AssertionError('no wheel before validation')),patch.object(proof,'notify',side_effect=notices.append): with self.assertRaisesRegex(proof.Refused,'facing prerequisite'):proof.run_selection(scene,SimpleNamespace(require_idle=scene.idle),o,mouse,report,scene.clock,scene.sleep) self.assertTrue(report['facing_proof']['completed']);self.assertNotIn('action_selection_proof',report) self.assertEqual([n for n in notices if n['kind']=='alignment-finished'],[{'kind':'alignment-finished','successful':False}]) def test_selection_compaction_and_hashes_leave_live_proof_intact(self): full={'sample':1,'real_ms':1,'received_monotonic':1.,'origin':[1,2,3],'choices':{'actions':[]}} report={'approach_requested':True,'alignment_requested':True,'selection_requested':True,'approach_proof':{'samples':[full]},'facing_proof':{'samples':[full]},'action_selection_proof':{'samples':[full]}} encoded=proof.report_for_write(report) for name in ('approach_proof','facing_proof','action_selection_proof'): self.assertEqual(set(encoded[name]['samples'][0]),{'sample','real_ms','received_monotonic'});self.assertIn('origin',report[name]['samples'][0]) with tempfile.TemporaryDirectory() as folder: root=Path(folder) for name in proof.SELECTION_MODULES:(root/name).write_text(name) with patch.object(proof,'__file__',str(root/'headless_input_proof.py')):self.assertEqual(set(proof.module_hashes(proof.SELECTION_MODULES)),{'ordinary_gate_choices.py','prison_action_selection_proof.py','prison_action_selection_evidence.py'}) def test_button_supervisor_lease_is_independent_of_key_release(self): events=[{'kind':'report','metadata':{}},{'kind':'child','identity':{}},{'kind':'button-pending','button':5},{'kind':'release-confirmed','key':'W'}] class Pipe: def poll(self,*a):return bool(events) def recv(self):return events.pop(0) def close(self):pass class Process: pid=123;exitcode=0;alive=True;terminated=0 def start(self):pass def is_alive(self):return self.alive def terminate(self):self.terminated+=1 def join(self,*a):self.alive=False def kill(self):self.alive=False child=Process();pipe=Pipe();clock=Clock() def ticking():clock.now+=.25;return clock.now context=SimpleNamespace(Pipe=lambda **k:(pipe,pipe),Process=lambda **k:child) with patch.object(proof.multiprocessing,'get_context',return_value=context),patch.object(proof.time,'monotonic',side_effect=ticking),patch.object(proof.signal,'signal'),patch.object(proof,'stop_exact_child') as stop,patch.object(proof,'supervision_failure') as failed: self.assertEqual(proof.supervise([]),1);self.assertEqual(child.terminated,1) self.assertEqual(failed.call_args.kwargs['button'],5);self.assertIsNone(failed.call_args.args[1]);stop.assert_called_once() class CoarseApproachTests(unittest.TestCase): def test_opt_in_requires_approach_and_default_is_unchanged(self): argv=['--enable-private-keyboard-proof','--prefix','/tmp/pfx','--game-executable','/tmp/game','--profile','/tmp/profile','--report','/tmp/report','--host-display',':0','--nonce','N'] self.assertFalse(proof.arguments(argv).enable_private_coarse_approach_proof) with self.assertRaises(SystemExit),patch('sys.stderr'):proof.arguments(argv+['--enable-private-coarse-approach-proof']) args=proof.arguments(argv+['--enable-private-coarse-approach-proof','--enable-private-approach-proof','--enable-private-look-proof']) self.assertTrue(args.enable_private_coarse_approach_proof) with self.assertRaisesRegex(proof.Refused,'Coarse approach'): proof.sequence(FakeGuard(),FakeKeyboard(FakeObserver()),FakeObserver(),{},coarse_approach=True) with self.assertRaisesRegex(proof.Refused,'Coarse approach'): proof.sequence(FakeGuard(),FakeKeyboard(FakeObserver()),FakeObserver(),{'coarse_approach_requested':True},approach=True) def test_coarse_mouse_preserves_actual_private_pointer_focus_and_deadline_checks(self): class Fn: def __init__(self,fn):self.fn=fn def __call__(self,*a):return self.fn(*a) events=[];xt=SimpleNamespace(XTestFakeRelativeMotionEvent=Fn(lambda *a:events.append(a) or 1)) query=SimpleNamespace(d=1,x=SimpleNamespace(XQueryPointer=Fn(lambda *a:1),XFlush=lambda d:None),server_pid=lambda child:9) keyboard=SimpleNamespace(xt=xt,require_idle=lambda:None);clock=Clock() target={'geometry':{'root':1},'focused_clipped_bounds':(0,0,1000,720)} checks=[];guard=SimpleNamespace(check=lambda:checks.append(True) or target) mouse=proof.RelativeMouse(query,keyboard,{'window':100,'pid':9},enable_approach=True,enable_coarse_approach=True) mouse.pointer=lambda:{'root':1,'child':0,'root_x':500,'root_y':300,'mask':0} for dx in (384,-384):mouse.approach_move(dx,0,guard,deadline=35,clock=clock) self.assertEqual(len(events),2);self.assertEqual(len(checks),4) for dx,dy in ((385,0),(300,1),(True,0)): with self.assertRaises(proof.Refused):mouse.approach_move(dx,dy,guard,deadline=35,clock=clock) mouse.enable_coarse_approach=False with self.assertRaises(proof.Refused):mouse.approach_move(300,0,guard,deadline=35,clock=clock) mouse.enable_coarse_approach=True mouse.pointer=lambda:{'root':1,'child':0,'root_x':900,'root_y':300,'mask':0} with self.assertRaisesRegex(proof.Refused,'clipped'):mouse.approach_move(300,0,guard,deadline=35,clock=clock) mouse.pointer=lambda:{'root':1,'child':77,'root_x':500,'root_y':300,'mask':0};query.server_pid=lambda child:10 with self.assertRaisesRegex(proof.Refused,'foreign'):mouse.approach_move(300,0,guard,deadline=35,clock=clock) mouse.pointer=lambda:{'root':1,'child':0,'root_x':500,'root_y':300,'mask':0} def late():clock.now=35;return target with self.assertRaisesRegex(proof.Refused,'deadline'):mouse.approach_move(300,0,SimpleNamespace(check=late),deadline=35,clock=clock) self.assertEqual(len(events),2) with self.assertRaisesRegex(proof.Refused,'capability'):proof.RelativeMouse(query,keyboard,{'window':100,'pid':9},enable_coarse_approach=True) if __name__ == '__main__': unittest.main()