Exercise ordinary input in private display and retain inventory UI failure diagnosis
This commit is contained in:
@@ -0,0 +1,112 @@
|
||||
"""Runner policy/provenance tests only: no processes, display, or input calls."""
|
||||
import copy
|
||||
import hashlib
|
||||
import json
|
||||
from pathlib import Path
|
||||
import sys
|
||||
import tempfile
|
||||
from types import SimpleNamespace
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'tools'))
|
||||
import headless_game as r
|
||||
from test_ordinary_input_runner import records, NONCE
|
||||
|
||||
class ExerciseTests(unittest.TestCase):
|
||||
def test_optin_preflight_before_side_effects_and_default_unchanged(self):
|
||||
r.validate_input_exercise(SimpleNamespace())
|
||||
for project in ('core','chat'):
|
||||
args=SimpleNamespace(exercise_player_input=True,observe_player_input=True,inspect_headless_display=True,project=project,settle=50)
|
||||
r.validate_input_exercise(args)
|
||||
for change in ({'observe_player_input':False},{'inspect_headless_display':False},{'settle':49},{'project':r.CONTROL},{'project':'ai'},{'packed_core':Path('/x')},{'packed_addons':Path('/x')},{'packed_session':Path('/x')}):
|
||||
bad=copy.copy(args)
|
||||
for key,value in change.items():setattr(bad,key,value)
|
||||
with self.subTest(change=change),patch.object(r,'packed_request',side_effect=AssertionError('must refuse before filesystem')):
|
||||
with self.assertRaises(ValueError):r.run(bad)
|
||||
|
||||
def test_nested_command_keeps_original_boundary_and_proton_argv(self):
|
||||
with tempfile.TemporaryDirectory() as folder:
|
||||
root=Path(folder);(root/'tools').mkdir();profile=root/'profile';profile.mkdir()
|
||||
for file in ('headless_input_proof.py','headless_display_probe.py'):(root/'tools'/file).write_text('# fake helper')
|
||||
base=['gamescope','--backend','headless','--','/proton','run','/game','-profile','Z:/profile','-errOrdinaryInputObserve',NONCE]
|
||||
wrapped,display=r.display_probe_command(base,root,Path('/compat'),profile,Path('/game'),NONCE,':0')
|
||||
command,context=r.input_exercise_command(wrapped,root,Path('/compat'),profile,Path('/game'),NONCE,':0',display)
|
||||
self.assertEqual(command[:4],base[:4]);self.assertEqual(command[-len(base[4:]):],base[4:])
|
||||
child_start=command.index('--',4)+1
|
||||
self.assertEqual(command[child_start:],wrapped[4:]);self.assertEqual(context['command'],wrapped[4:])
|
||||
self.assertEqual(context['nonce'],NONCE)
|
||||
with self.assertRaises(ValueError):r.input_exercise_command(base,root,Path('/compat'),profile,Path('/game'),NONCE,':0',display)
|
||||
with self.assertRaises(ValueError):r.input_exercise_command(wrapped,root,Path('/compat'),profile,Path('/game'),NONCE,':0',None)
|
||||
Path(context['report']).write_text('{}')
|
||||
with self.assertRaises(ValueError):r.input_exercise_command(wrapped,root,Path('/compat'),profile,Path('/game'),NONCE,':0',display)
|
||||
|
||||
def test_native_movement_requires_action_and_real_motion_before_menu(self):
|
||||
static=records(menus=True)
|
||||
evidence=r.input_observer_evidence(static,NONCE)
|
||||
self.assertTrue(evidence['sampler_complete']);self.assertFalse(evidence['movement']['action_observed']);self.assertFalse(evidence['movement']['physical_motion_observed'])
|
||||
# First post-READY sample is also the native opening sample: final movement
|
||||
# can precede opening during the same sample interval.
|
||||
line='sample=22 real=3200 world=7200 SAMPLE origin=<1, 2, 3> forward=0 input=<0, 0, 0> velocity=<0, 0, 0>'
|
||||
moving=static.replace(line,'sample=22 real=3200 world=7200 SAMPLE origin=<1.2, 2, 3> forward=1 input=<0, 0, 1> velocity=<1, 0, 0>')
|
||||
good=r.input_observer_evidence(moving,NONCE)
|
||||
self.assertTrue(good['sampler_complete']);self.assertTrue(good['movement']['action_observed']);self.assertTrue(good['movement']['physical_motion_observed']);self.assertTrue(good['movement']['released_input_observed'])
|
||||
# Native release requires a later sample, with both controls neutral.
|
||||
for held in ('forward=1 input=<0, 0, 0>', 'forward=0 input=<0, 0.2, 0>'):
|
||||
after=moving.split('INVENTORY_CLOSED',1)
|
||||
bad=after[0]+'INVENTORY_CLOSED'+after[1].replace('forward=0 input=<0, 0, 0>',held)
|
||||
self.assertFalse(r.input_observer_evidence(bad,NONCE)['movement']['released_input_observed'])
|
||||
for text,key in ((moving.replace('forward=1','forward=0'),'action_observed'),(moving.replace('input=<0, 0, 1>','input=<0, 0, 0>'),'action_observed'),(moving.replace('origin=<1.2, 2, 3>','origin=<1, 2, 3>'),'physical_motion_observed'),(moving.replace('velocity=<1, 0, 0>','velocity=<0, 0, 0>'),'physical_motion_observed')):
|
||||
self.assertFalse(r.input_observer_evidence(text,NONCE)['movement'][key])
|
||||
|
||||
def make_proof(self,root):
|
||||
session=root/'logs_session';session.mkdir()
|
||||
context={'report':str(root/'report.json'),'helper_sha256':'h','ownership_helper_sha256':'d','prefix':'/compat/pfx','profile':'/profile','game_executable':'/game','command':['display-wrapper'],'nonce':NONCE}
|
||||
owner={'window':10,'pid':20,'server_pid':20,'process':{'pid':20,'start':5}}
|
||||
binding={'display':':99'}
|
||||
(session/'display-ownership.json').write_text(json.dumps({'binding':binding,'owned_windows':[owner]}))
|
||||
report={**context,'schema':1,'kind':'headless-private-keyboard-experiment','sequence_completed':True,'hard_supervision':True,'all_attempted_presses_released':True,'input_sent':True,'paired_release_errors':[], 'console':str(session/'console.log'),'binding':binding,'owner':owner,'observer':{'events':['READY','INVENTORY_OPEN','INVENTORY_CLOSED']},'events':[{'key':key,'down':down,'monotonic':1+i*.2} for i,(key,down) in enumerate([('W',True),('W',False),('Tab',True),('Tab',False),('Tab',True),('Tab',False)])]}
|
||||
observation={'nonce':NONCE,'sampler_complete':True,'inventory_open_observed':True,'inventory_close_observed':True,'movement':{'action_observed':True,'physical_motion_observed':True,'released_input_observed':True}}
|
||||
return session,context,report,observation
|
||||
|
||||
def test_live_poll_ignores_incomplete_foreign_or_success_and_latches_exact_failure(self):
|
||||
with tempfile.TemporaryDirectory() as folder:
|
||||
session,context,report,observation=self.make_proof(Path(folder));path=Path(context['report'])
|
||||
self.assertFalse(r.input_exercise_poll_failure(context))
|
||||
for raw in (b'',b'{',b'null',b'[]',json.dumps(report).encode()):
|
||||
path.write_bytes(raw);self.assertFalse(r.input_exercise_poll_failure(context))
|
||||
failed={**report,'sequence_completed':False,'error':'key release not confirmed'}
|
||||
path.write_text(json.dumps({**failed,'nonce':'foreign'}));self.assertFalse(r.input_exercise_poll_failure(context))
|
||||
for detail in ({'error':'native helper refused'}, {'supervision_failed':True}, {'sequence_completed':False,'whole_process_group_cleanup_verified':False}):
|
||||
path.write_text(json.dumps({**report,**detail}))
|
||||
self.assertTrue(r.input_exercise_poll_failure(context))
|
||||
self.assertEqual(r.input_exercise_poll_result(context,None,False),'input-exercise-failed')
|
||||
self.assertEqual(r.input_exercise_poll_result(context,'game-started',False),'input-exercise-failed')
|
||||
self.assertEqual(r.input_exercise_poll_result(context,None,True),'game-failed')
|
||||
self.assertEqual(r.input_exercise_poll_result(context,'process-exited',False),'process-exited')
|
||||
self.assertIsNone(r.input_exercise_poll_result(None,None,False))
|
||||
# Initial false status without the finally completion field is not terminal.
|
||||
path.write_text(json.dumps({**report,'sequence_completed':False}))
|
||||
self.assertFalse(r.input_exercise_poll_failure(context))
|
||||
|
||||
def test_exact_bytes_archived_and_verified(self):
|
||||
with tempfile.TemporaryDirectory() as folder:
|
||||
session,context,report,observation=self.make_proof(Path(folder))
|
||||
raw=(json.dumps(report,indent=3)+'\n').encode();Path(context['report']).write_bytes(raw)
|
||||
status,result=r.input_exercise_result('game-started',context,session,observation,{'verified':True})
|
||||
self.assertEqual(status,'game-started');self.assertTrue(result['verified'],result);self.assertFalse(result['gameplay_qualified'])
|
||||
self.assertEqual((session/'input-proof.json').read_bytes(),raw);self.assertEqual(result['sha256'],hashlib.sha256(raw).hexdigest())
|
||||
|
||||
def test_report_or_native_failure_cannot_qualify_or_replace_prior_failure(self):
|
||||
mutations=[lambda p:p.update(sequence_completed=False),lambda p:p.update(hard_supervision=False),lambda p:p.update(supervision_failed=True),lambda p:p.update(all_attempted_presses_released=False),lambda p:p.update(error='failed'),lambda p:p.update(paired_release_errors=['failed']),lambda p:p.update(unconfirmed_held_key='W'),lambda p:p.update(nonce='wrong'),lambda p:p.update(helper_sha256='wrong'),lambda p:p.update(command=['wrong']),lambda p:p.update(console='/other/console.log'),lambda p:p['events'].pop(),lambda p:p['events'][0].update(down=1),lambda p:p['events'][0].update(monotonic=float('nan')),lambda p:p['events'][-1].update(monotonic=100),lambda p:p['owner'].update(pid=99),lambda p:p['observer'].update(events=['READY'])]
|
||||
for change in mutations:
|
||||
with tempfile.TemporaryDirectory() as folder:
|
||||
session,context,report,observation=self.make_proof(Path(folder));change(report);Path(context['report']).write_text(json.dumps(report))
|
||||
status,result=r.input_exercise_result('game-started',context,session,observation,{'verified':True})
|
||||
self.assertEqual(status,'input-exercise-incomplete');self.assertFalse(result['verified']);self.assertTrue((session/'input-proof.json').is_file())
|
||||
for prior in ('game-failed','cleanup-failed','source-changed','timeout','display-probe-incomplete','game-started'):
|
||||
with tempfile.TemporaryDirectory() as folder:
|
||||
session,context,report,observation=self.make_proof(Path(folder));Path(context['report']).write_text(json.dumps(report));observation['movement']['released_input_observed']=False
|
||||
status,result=r.input_exercise_result(prior,context,session,observation,{'verified':True})
|
||||
self.assertEqual(status,'input-exercise-incomplete' if prior=='game-started' else prior);self.assertFalse(result['verified'])
|
||||
|
||||
if __name__=='__main__':unittest.main()
|
||||
@@ -0,0 +1,307 @@
|
||||
#!/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 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(), .9)
|
||||
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_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)
|
||||
|
||||
|
||||
if __name__ == '__main__': unittest.main()
|
||||
Reference in New Issue
Block a user