1219 lines
90 KiB
Python
1219 lines
90 KiB
Python
#!/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=<NaN,1.7,0>'),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<path.stat().st_size:observer.poll()
|
|
sample=observer.approach_snapshots()[-1]
|
|
full=[{**copy.deepcopy(sample),'sample':i} for i in range(1,451)]
|
|
report={'approach_requested':True,'approach_proof':{'samples':full,'motions':[{'dx':128,'dy':0,'monotonic':10,'after_sample':i,'response_sample':i+1} for i in range(16)],'events':[{'key':'W','down':bool(i%2),'monotonic':10} for i in range(24)]},
|
|
'look_observation':observer.look_records,'gate_observation':observer.gate_records,
|
|
'observer':{'records':observer.raw,'events':observer.events},'provenance_budget':'x'*65536}
|
|
report['alignment_requested']=True
|
|
report['alignment_proof']={'samples':full[-80:],'motions':[{'dx':128,'dy':0,'monotonic':10,'after_sample':i,'response_sample':i+1} for i in range(6)]}
|
|
report['selection_requested']=True
|
|
report['facing_proof']={'samples':full[-80:],'motions':report['alignment_proof']['motions']}
|
|
report['action_selection_proof']={'samples':full[-20:],'wheel':{'button':5,'released':True}}
|
|
report['wheel_events']=[{'button':5,'down':True,'monotonic':10.,'sent':True},{'button':5,'down':False,'monotonic':10.1,'sent':True}]
|
|
payload=json.dumps(proof.report_for_write(report),indent=2).encode()
|
|
self.assertEqual(len(observer.look_records),451);self.assertEqual(len(observer.gate_records),901)
|
|
self.assertLess(len(payload),1024*1024)
|
|
self.assertEqual(len(report['approach_proof']['samples']),450)
|
|
self.assertIn('geometry',report['approach_proof']['samples'][0])
|
|
finally:observer.close()
|
|
|
|
def test_approach_watchdog_cannot_be_renewed_by_module_notification(self):
|
|
events=[{'kind':'report','metadata':{}},{'kind':'child','identity':{}},
|
|
{'kind':'approach-pending','deadline_monotonic':10},
|
|
{'kind':'approach-pending','deadline_monotonic':30}]
|
|
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+=1;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.assertLessEqual(clock.now,11)
|
|
stop.assert_called_once();self.assertIsNone(failed.call_args.args[1])
|
|
|
|
|
|
|
|
class AlignmentIntegrationTests(unittest.TestCase):
|
|
def test_cli_requires_explicit_approach_and_look(self):
|
|
base=['--enable-private-keyboard-proof','--prefix','/tmp/pfx','--game-executable','/tmp/game',
|
|
'--profile','/tmp/profile','--report','/tmp/report','--host-display',':0','--nonce','N']
|
|
for flags in ([],['--enable-private-look-proof'],['--enable-private-approach-proof']):
|
|
with self.assertRaises(SystemExit),patch('sys.stderr'):
|
|
proof.arguments(base+flags+['--enable-private-alignment-proof'])
|
|
args=proof.arguments(base+['--enable-private-look-proof','--enable-private-approach-proof','--enable-private-alignment-proof'])
|
|
self.assertTrue(args.enable_private_alignment_proof)
|
|
self.assertFalse(proof.arguments(base).enable_private_alignment_proof)
|
|
def test_alignment_follows_completed_approach_without_more_keyboard_events(self):
|
|
clock=Clock();o=FakeObserver();o.enable_approach=True;o.ready_at=0;k=FakeKeyboard(o);r={};order=[];notices=[]
|
|
def look(*a):order.append('look')
|
|
def approach(g,k,o,m,r,c,s):
|
|
order.append('approach');k.send('W',True);k.send('W',False)
|
|
r['approach_proof']={'completed':True,'final_sample':99,'samples':[{'sample':99}], 'paired_release_errors':[],'unconfirmed_held_key':None}
|
|
def alignment(g,k,o,m,r,c,s):
|
|
self.assertEqual(order,['look','approach']);self.assertTrue(r['approach_proof']['completed'])
|
|
self.assertEqual(notices[-1],{'kind':'approach-finished','successful':True})
|
|
order.append('alignment');r['alignment_proof']={'completed':True}
|
|
with patch.object(proof,'perform_look',side_effect=look),patch.object(proof,'run_approach',side_effect=approach),patch.object(proof,'run_alignment',side_effect=alignment),patch.object(proof,'notify',side_effect=notices.append):
|
|
proof.sequence(FakeGuard(),k,o,r,clock,clock.sleep,mouse=object(),approach=True,alignment=True)
|
|
self.assertEqual(order,['look','approach','alignment'])
|
|
self.assertEqual(k.events,[('W',True),('W',False)]);self.assertTrue(r['sequence_completed'])
|
|
self.assertEqual(r['approach_proof']['samples'],[{'sample':99}])
|
|
self.assertEqual(r['approach_proof']['final_sample'],99)
|
|
def test_failed_alignment_does_not_erase_approach_or_reactivate_its_lease(self):
|
|
clock=Clock();o=FakeObserver();o.enable_approach=True;o.ready_at=0;k=FakeKeyboard(o);r={};notices=[]
|
|
def approach(*args):r['approach_proof']={'completed':True,'final_sample':100,'paired_release_errors':[],'unconfirmed_held_key':None}
|
|
with patch.object(proof,'perform_look'),patch.object(proof,'run_approach',side_effect=approach),patch.object(proof,'run_alignment',side_effect=proof.Refused('selection missing')),patch.object(proof,'notify',side_effect=notices.append):
|
|
with self.assertRaisesRegex(proof.Refused,'selection missing'):
|
|
proof.sequence(FakeGuard(),k,o,r,clock,clock.sleep,mouse=object(),approach=True,alignment=True)
|
|
self.assertFalse(r['sequence_completed']);self.assertTrue(r['approach_proof']['completed'])
|
|
self.assertTrue(all(e['successful'] for e in notices if e['kind']=='approach-finished'))
|
|
self.assertEqual(k.events,[])
|
|
def test_yaw_only_capability_checks_optin_bounds_original_deadline_and_native_guard(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))
|
|
query=SimpleNamespace(x=SimpleNamespace(XQueryPointer=Fn(lambda *a:1),XFlush=lambda d:None),d=123)
|
|
mouse=proof.RelativeMouse(query,SimpleNamespace(xt=xt,require_idle=lambda:None),{'window':100,'pid':9},enable_approach=True)
|
|
guard=FakeGuard();clock=Clock();mouse.validate_pointer=lambda *a:None
|
|
with self.assertRaises(proof.Refused):mouse.alignment_move(24,0,guard,deadline=8,clock=clock)
|
|
mouse.enable_alignment=True
|
|
for dx in (1,-128,128):mouse.alignment_move(dx,0,guard,deadline=8,clock=clock)
|
|
for dx,dy in ((0,0),(129,0),(-129,0),(1,1),(True,0),(1,False)):
|
|
with self.assertRaises(proof.Refused):mouse.alignment_move(dx,dy,guard,deadline=8,clock=clock)
|
|
wrapped=proof.AlignmentMouse(mouse,{},clock,8)
|
|
with self.assertRaisesRegex(proof.Refused,'extend'):wrapped.move(1,0,guard,deadline=8.001)
|
|
def late(*a):clock.now=8
|
|
mouse.validate_pointer=late
|
|
with self.assertRaisesRegex(proof.Refused,'deadline'):wrapped.move(1,0,guard,deadline=8)
|
|
self.assertEqual(len(events),3)
|
|
def test_snapshot_has_native_camera_action_fields_and_independent_alignment_cursor(self):
|
|
from test_ordinary_gate_evidence import fixture,NONCE
|
|
clean='\n'.join(s for s in fixture().splitlines() if 'CANCELLED' not in s)+'\n'
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
path=Path(folder)/'console.log';path.write_text('')
|
|
o=proof.Observer(path,NONCE,clock=lambda:10.,enable_look=True,enable_approach=True)
|
|
try:
|
|
with path.open('a') as out:out.write(clean)
|
|
first=o.approach_snapshots()[-1];self.assertEqual(first['sample'],30)
|
|
self.assertEqual(first['camera'],(6088.58,213.019,7158.56))
|
|
for key in ('selected','context_selected','cached_performable','progress_bound','performing'):self.assertIs(first[key],False)
|
|
self.assertEqual((first['pressed'],first['progress'],first['duration']),(0,0,4))
|
|
o.alignment_previous=27
|
|
self.assertEqual([s['sample'] for s in o.alignment_snapshots()],[28,29,30])
|
|
self.assertEqual(o.approach_previous,30);self.assertEqual(o.alignment_previous,30)
|
|
self.assertEqual(o.approach_snapshots(),[first])
|
|
finally:o.close()
|
|
def test_failed_independent_approach_blocks_alignment_module(self):
|
|
o=SimpleNamespace(ready_at=0,ready_real=3000,queued_real=1000,nonce='N',approach_data=lambda:('raw',{}))
|
|
with patch('prison_approach_evidence.validate_approach_evidence',return_value={'verified':False,'errors':['raw route mismatch']}),patch('prison_alignment_proof.perform_alignment') as module:
|
|
with self.assertRaisesRegex(proof.Refused,'prerequisite refused'):
|
|
proof.run_alignment(None,None,o,None,{'look_proof':{}},lambda:30,lambda s:None)
|
|
module.assert_not_called()
|
|
def test_alignment_adapter_passes_readonly_keyboard_and_original_lease(self):
|
|
import copy
|
|
clock=Clock();clock.now=34
|
|
route={'completed':True,'final_sample':100,'geometry':{'nonce':'N'},'samples':[{'sample':100,'origin':[1,2,3]}]}
|
|
original=copy.deepcopy(route);report={'approach_proof':route,'look_proof':{'baseline_samples':[20,21],'response_sample':22}}
|
|
o=SimpleNamespace(ready_at=0,ready_real=3000,queued_real=1000,nonce='N',approach_data=lambda:('raw',{}),
|
|
look_records=[{'kind':'sample','sample':21,'direction':[0,0,1]},{'kind':'sample','sample':22,'direction':[.03,0,.999]}],
|
|
alignment_snapshots=lambda:[{'sample':100}],alignment_previous=None)
|
|
keyboard=SimpleNamespace(require_idle=lambda:None);motions=[];notices=[]
|
|
mouse=SimpleNamespace(alignment_move=lambda *a,**k:motions.append((a,k)))
|
|
def module(guard,require_idle,m,observe,geometry,calibration,evidence,r,**kwargs):
|
|
self.assertTrue(callable(require_idle));self.assertFalse(hasattr(require_idle,'send'))
|
|
self.assertEqual(o.alignment_previous,99)
|
|
self.assertEqual(calibration['observer_deadline_monotonic'],42.5)
|
|
self.assertEqual(calibration['alignment_deadline_monotonic'],42)
|
|
m.move(12,0,guard,deadline=42)
|
|
r['alignment_proof']={'completed':True,'samples':[{'sample':100}]}
|
|
kwargs['notify']({'kind':'alignment-pending','deadline_monotonic':42})
|
|
kwargs['notify']({'kind':'alignment-finished','successful':True})
|
|
with patch('prison_approach_evidence.validate_approach_evidence',return_value={'verified':True,'last_sample':100}),patch('prison_alignment_proof.perform_alignment',side_effect=module),patch.object(proof,'notify',side_effect=notices.append):
|
|
proof.run_alignment(FakeGuard(),keyboard,o,mouse,report,clock,clock.sleep)
|
|
self.assertEqual(route,original);self.assertEqual(motions[0][1]['deadline'],42)
|
|
self.assertEqual([n['kind'] for n in notices],['alignment-pending','alignment-pending','alignment-finished'])
|
|
self.assertTrue(notices[-1]['successful'])
|
|
# A later start has only the original remaining lifetime, never a new 8s.
|
|
clock.now=40;notices.clear();report.pop('alignment_proof')
|
|
def late_module(g,k,m,observe,geometry,calibration,evidence,r,**kwargs):
|
|
self.assertEqual(calibration['alignment_deadline_monotonic'],42.5)
|
|
clock.now=42.5;r['alignment_proof']={'completed':True}
|
|
with patch('prison_approach_evidence.validate_approach_evidence',return_value={'verified':True,'last_sample':100}),patch('prison_alignment_proof.perform_alignment',side_effect=late_module),patch.object(proof,'notify',side_effect=notices.append):
|
|
with self.assertRaisesRegex(proof.Refused,'deadline'):proof.run_alignment(FakeGuard(),keyboard,o,mouse,report,clock,clock.sleep)
|
|
self.assertFalse(notices[-1]['successful'])
|
|
def test_actual_alignment_module_runs_through_helper_capabilities(self):
|
|
import copy
|
|
from test_prison_alignment_proof import Scene
|
|
scene=Scene();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)
|
|
mouse=SimpleNamespace(alignment_move=lambda dx,dy,guard,deadline,clock:scene.move(dx,dy,guard,deadline=deadline))
|
|
with patch('prison_approach_evidence.validate_approach_evidence',return_value=scene.accepted):
|
|
proof.run_alignment(scene,SimpleNamespace(require_idle=scene.idle),o,mouse,report,scene.clock,scene.sleep)
|
|
self.assertTrue(report['alignment_proof']['completed']);self.assertGreater(len(scene.motions),0)
|
|
self.assertTrue(report['input_sent']);self.assertEqual(report['events'],[])
|
|
self.assertEqual(report['approach_proof'],prior)
|
|
self.assertLessEqual(report['alignment_proof']['deadline_monotonic'],39.)
|
|
with patch('prison_approach_evidence.validate_approach_evidence',return_value=scene.accepted):
|
|
with self.assertRaisesRegex(ValueError,'already exists'):
|
|
proof.run_alignment(scene,SimpleNamespace(require_idle=scene.idle),o,mouse,report,scene.clock,scene.sleep)
|
|
|
|
def test_conditional_hashes_and_compaction_preserve_indexed_approach(self):
|
|
import copy,hashlib
|
|
report={'approach_requested':True,'alignment_requested':True,'approach_proof':{'final_sample':100,'samples':[{'sample':100,'real_ms':12000,'received_monotonic':10,'origin':[1,2,3]}]},'alignment_proof':{'completed':False,'samples':[{'sample':100,'real_ms':12000,'received_monotonic':10,'camera':[1,3,3]}]}}
|
|
prior=copy.deepcopy(report);encoded=proof.report_for_write(report)
|
|
self.assertEqual(report,prior);self.assertEqual(encoded['approach_proof']['final_sample'],100)
|
|
self.assertEqual(encoded['approach_proof']['samples'],encoded['alignment_proof']['samples'])
|
|
self.assertEqual(set(encoded['alignment_proof']['samples'][0]),{'sample','real_ms','received_monotonic'})
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
root=Path(folder)
|
|
for name in proof.ALIGNMENT_MODULES:(root/name).write_text(name)
|
|
with patch.object(proof,'__file__',str(root/'headless_input_proof.py')):
|
|
values=proof.module_hashes(proof.ALIGNMENT_MODULES)
|
|
self.assertEqual(set(values),{'prison_alignment_proof.py','prison_alignment_evidence.py'})
|
|
self.assertEqual(values['prison_alignment_proof.py'],hashlib.sha256(b'prison_alignment_proof.py').hexdigest())
|
|
def test_alignment_watchdog_is_separate_and_cannot_renew(self):
|
|
events=[{'kind':'report','metadata':{}},{'kind':'child','identity':{}},
|
|
{'kind':'approach-pending','deadline_monotonic':5},{'kind':'approach-finished','successful':True},
|
|
{'kind':'alignment-pending','deadline_monotonic':9},{'kind':'alignment-pending','deadline_monotonic':12}]
|
|
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+=.5;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.assertGreaterEqual(clock.now,9);self.assertLessEqual(clock.now,9.5)
|
|
stop.assert_called_once();self.assertIsNone(failed.call_args.args[1])
|
|
|
|
|
|
|
|
class GateChoicesArguments(unittest.TestCase):
|
|
def test_exact_gate_choices_are_readonly_and_bound_in_wrapper_and_live_process(self):
|
|
args=SimpleNamespace(nonce='N',prefix=Path('/tmp/pfx'),profile=Path('/tmp/profile'),game_executable=Path('/tmp/game'),host_display=':0',report=Path('/tmp/input.json'),console=None,enable_private_look_proof=True,enable_private_approach_proof=True,
|
|
command=[sys.executable,str(Path(proof.display.__file__).resolve()),'placeholder'])
|
|
prefix=[str(Path.home()/'.local/share/Steam/steamapps/common/Proton - Experimental/proton'),'run',str(args.game_executable),'-errOrdinaryInputObserve','N','-errOrdinaryLookObserve','N','-errOrdinaryScenario','core','-profile','Z:/tmp/profile']
|
|
gate=['-errOrdinaryGateObserve','N'];choice=['-errOrdinaryGateChoices','N']
|
|
cases=[(gate,True,True),(gate+choice,True,True),(gate+choice,False,False),(gate,False,True),([],False,True),([],True,False),(choice,True,False),(gate+choice+choice,True,False),(gate+gate,True,False),(gate+['-errOrdinaryGateChoices','foreign'],True,False),(gate+['-errOrdinaryGateChoices'],True,False),(gate+['-errOrdinaryGateChoicesAlias','N'],True,False)]
|
|
for suffix,approach,allowed in cases:
|
|
with self.subTest(suffix=suffix,approach=approach):
|
|
args.enable_private_approach_proof=approach
|
|
native=prefix+suffix;owner={'pid':123,'process':{'pid':123}}
|
|
wrapped=SimpleNamespace(query_worker=False,prefix=args.prefix,profile=args.profile,game_executable=args.game_executable,host_display=args.host_display,report=Path('/tmp/display.json'),command=['--']+native)
|
|
with patch.object(proof.display,'parse_args',return_value=wrapped),patch.object(proof.display,'validate_paths'),patch.object(proof.display,'process',return_value={'args':native}),patch.object(proof.display,'identity',return_value=owner['process']):
|
|
if allowed:
|
|
self.assertEqual(proof.wrapper_command(args),args.command)
|
|
proof.game_arguments(owner,args)
|
|
else:
|
|
with self.assertRaises(proof.Refused):proof.wrapper_command(args)
|
|
with self.assertRaises(proof.Refused):proof.game_arguments(owner,args)
|
|
|
|
|
|
class WheelRig:
|
|
class Fn:
|
|
def __init__(self,fn):self.fn=fn
|
|
def __call__(self,*args):return self.fn(*args)
|
|
def __init__(self,clock=None):
|
|
self.clock=clock or Clock();self.clock.now=max(self.clock.now,1.)
|
|
self.mask=0;self.calls=[];self.fail_down=False;self.fail_up=False;self.sticky=False;self.lost=False
|
|
self.report={};self.guards=0
|
|
def event(d,button,down,delay):
|
|
self.calls.append((button,down));self.clock.sleep(.001)
|
|
if down:self.mask=1<<12
|
|
elif not self.sticky:self.mask=0
|
|
return 0 if (self.fail_down and down) or (self.fail_up and not down) else 1
|
|
self.xt=SimpleNamespace(XTestFakeButtonEvent=self.Fn(event))
|
|
self.query=SimpleNamespace(d=object(),x=SimpleNamespace(XFlush=lambda d:None))
|
|
self.keyboard=SimpleNamespace(require_idle=lambda:None)
|
|
self.mouse=SimpleNamespace(xt=self.xt,query=self.query,keyboard=self.keyboard,validate_pointer=self.position)
|
|
def position(self,target,dx,dy,allowed_mask=0):
|
|
assert dx==dy==0
|
|
if self.mask & ~allowed_mask:raise proof.Refused('foreign button')
|
|
return {'mask':self.mask}
|
|
def check(self):
|
|
self.guards+=1;self.clock.sleep(.03)
|
|
if self.lost:raise proof.Refused('ownership lost')
|
|
return {}
|
|
def wheel(self,deadline=3):return proof.SelectionWheel(self.mouse,self.report,self.clock,self.clock.sleep,deadline,enabled=True)
|
|
|
|
class SelectionIntegration(unittest.TestCase):
|
|
def test_selection_requires_all_optins_and_exact_native_choices(self):
|
|
argv=['--enable-private-keyboard-proof','--prefix','/tmp/pfx','--game-executable','/tmp/game','--profile','/tmp/profile','--report','/tmp/report','--host-display',':0','--nonce','N']
|
|
flags=['--enable-private-selection-proof','--enable-private-alignment-proof','--enable-private-approach-proof','--enable-private-look-proof']
|
|
self.assertFalse(proof.arguments(argv).enable_private_selection_proof)
|
|
for missing in flags[1:]:
|
|
with self.assertRaises(SystemExit),patch('sys.stderr'):proof.arguments(argv+[f for f in flags if f!=missing])
|
|
args=proof.arguments(argv+flags)
|
|
for native in (['-errOrdinaryGateObserve','N'],['-errOrdinaryGateObserve','N','-errOrdinaryGateChoices','other'],['-errOrdinaryGateObserve','N','-errOrdinaryGateChoices','N','-errOrdinaryGateChoices','N']):
|
|
with self.assertRaises(proof.Refused):proof.gate_options(native,args)
|
|
proof.gate_options(['-errOrdinaryGateObserve','N','-errOrdinaryGateChoices','N'],args)
|
|
|
|
def test_selection_sequence_uses_facing_selection_branch_without_default_alignment(self):
|
|
clock=Clock();observer=FakeObserver();observer.enable_approach=True;observer.enable_selection=True;observer.ready_at=0
|
|
keyboard=FakeKeyboard(observer);report={};calls=[]
|
|
def approach(*a):report['approach_proof']={'completed':True};calls.append('approach')
|
|
def select(*a):report.update(facing_proof={'completed':True},action_selection_proof={'completed':True});calls.append('selection')
|
|
with patch.object(proof,'perform_look',side_effect=lambda *a:calls.append('look')),patch.object(proof,'run_approach',side_effect=approach),patch.object(proof,'run_selection',side_effect=select),patch.object(proof,'run_alignment',side_effect=AssertionError('strict alignment must stay separate')):
|
|
proof.sequence(FakeGuard(),keyboard,observer,report,clock,clock.sleep,mouse=object(),approach=True,alignment=True,selection=True)
|
|
self.assertEqual(calls,['look','approach','selection']);self.assertTrue(report['sequence_completed']);self.assertEqual(report['events'],[])
|
|
with self.assertRaises(proof.Refused):proof.sequence(FakeGuard(),keyboard,observer,{},clock,clock.sleep,mouse=object(),approach=True,selection=True)
|
|
|
|
def test_actual_main_wires_selection_only_to_observer_and_sequence(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
root=Path(folder);args=SimpleNamespace(nonce='N',prefix=root/'pfx',profile=root/'profile',game_executable=root/'game',host_display=':0',report=root/'report.json',console=root/'console.log',command=[],enable_private_look_proof=True,enable_private_approach_proof=True,enable_private_alignment_proof=True,enable_private_selection_proof=True,enable_private_coarse_approach_proof=True)
|
|
owner={'window':1,'pid':2,'server_pid':2,'process':{'pid':2}}
|
|
query=SimpleNamespace(windows=lambda:[{}],focus=lambda:(1,0),close=lambda:None)
|
|
observer=SimpleNamespace(identity=(1,2),start_offset=0,offset=0,raw=[],events=[],look_records=[],gate_records=[],close=lambda:None)
|
|
polls=[]
|
|
def startup_poll():
|
|
polls.append(True)
|
|
if len(polls)>=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):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()
|