275 lines
20 KiB
Python
275 lines
20 KiB
Python
"""Captured F-only adapter on fake XTest; native log reader and pure policy live."""
|
|
import copy
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest.mock import patch
|
|
sys.path[:0]=[str(Path(__file__).resolve().parents[1]/'tools'),str(Path(__file__).resolve().parent)]
|
|
import headless_input_proof as h
|
|
import prison_held_action_proof as policy
|
|
from test_prison_held_action_proof import Scene
|
|
from test_prison_held_action_evidence import build as raw_fixture,raw
|
|
|
|
class Fn:
|
|
def __init__(self,fn):self.fn=fn
|
|
def __call__(self,*a):return self.fn(*a)
|
|
|
|
class Rig:
|
|
def __init__(self,scene=None):
|
|
self.scene=scene or Scene();self.keys=set();self.calls=[];self.fail_down=False;self.fail_up=False;self.pointer_busy=False
|
|
self.scene.guard_cost=.01;self.codes={0x77:25,0xff09:23,0x66:41}
|
|
def keymap(d,state):
|
|
for i in range(32):state[i]=bytes([sum(1<<(code%8) for code in self.keys if code//8==i)])
|
|
return 1
|
|
self.x=SimpleNamespace(XKeysymToKeycode=Fn(lambda d,k:self.codes.get(k,0)),XKeycodeToKeysym=Fn(lambda d,c,g:next((k for k,v in self.codes.items() if v==c),0)),
|
|
XQueryKeymap=Fn(keymap),XFlush=Fn(lambda d:0))
|
|
def keyevent(d,code,down,delay):
|
|
self.calls.append((d,code,down,self.scene.now))
|
|
if down:
|
|
self.keys.add(code);self.scene.native_held=True;self.scene.down=self.scene.now
|
|
return 0 if self.fail_down else 1
|
|
if self.fail_up:return 0
|
|
self.keys.discard(code);self.scene.native_held=False;self.scene.up=self.scene.now
|
|
return 1
|
|
self.xt=SimpleNamespace(XTestQueryExtension=Fn(lambda *a:1),XTestFakeKeyEvent=Fn(keyevent))
|
|
self.query=SimpleNamespace(d=1,x=self.x)
|
|
self.owner={'window':100,'pid':9,'process':{'start':1}}
|
|
def check():self.scene.check();return {'verified':True}
|
|
self.guard=SimpleNamespace(query=self.query,owner=copy.deepcopy(self.owner),check=check)
|
|
with patch.object(h.C,'CDLL',return_value=self.xt):self.keyboard=h.Keyboard(self.query)
|
|
def pointer(*a):
|
|
if self.pointer_busy:raise h.Refused('pointer outside original window')
|
|
self.mouse=SimpleNamespace(query=self.query,owner=copy.deepcopy(self.owner),keyboard=self.keyboard,validate_pointer=pointer,enable_coarse_facing=True)
|
|
self.report=self.scene.report;self.report.update(paired_release_errors=[],held_action_requested=True,coarse_facing_requested=True)
|
|
self.adapter=h.HeldFKey(self.guard,self.keyboard,self.mouse,self.report,self.scene.clock,self.scene.sleep,44.5,enabled=True)
|
|
def down(self):return self.adapter.down_f(self.guard,deadline=35,not_after=30.5)
|
|
def up(self,cleanup=False):return self.adapter.up_f(self.guard,deadline=35,cleanup=cleanup)
|
|
def run(self):
|
|
return policy.perform_held_action(self.guard,self.keyboard.require_idle,self.adapter,self.scene.observe,self.scene.geometry,self.scene.calibration,self.scene.evidence,self.report,
|
|
clock=self.scene.clock,sleep=self.scene.sleep,notify=self.scene.notifications.append)
|
|
|
|
class AdapterTests(unittest.TestCase):
|
|
def test_actual_keyboard_stays_wtab_and_only_explicit_adapter_can_f(self):
|
|
r=Rig();self.assertEqual(set(r.keyboard.codes),{'W','Tab'})
|
|
with self.assertRaises(ValueError):r.keyboard.send('F',True)
|
|
with self.assertRaises(ValueError):h.HeldFKey(r.guard,r.keyboard,r.mouse,r.report,r.scene.clock,r.scene.sleep,44.5)
|
|
down=r.down();self.assertEqual(down,{'key':'F','down_monotonic':r.scene.now})
|
|
self.assertEqual(r.calls[0][1:3],(41,1));self.assertEqual(r.report['unconfirmed_held_key'],'F')
|
|
with self.assertRaises(ValueError):r.down()
|
|
r.up();self.assertFalse(r.keys);self.assertIsNone(r.report['unconfirmed_held_key'])
|
|
self.assertEqual([set(e) for e in r.report['held_events']],[{'key','down','monotonic','sent'}]*2)
|
|
self.assertEqual([e['down'] for e in r.report['held_events']],[True,False])
|
|
def test_real_policy_uses_actual_f_adapter_and_preserves_release_evidence(self):
|
|
r=Rig();result=r.run()
|
|
self.assertTrue(result['completed']);self.assertTrue(result['release_confirmed']);self.assertFalse(r.keys)
|
|
self.assertEqual([c[2] for c in r.calls],[1,0])
|
|
self.assertEqual(r.report['held_events'][0]['monotonic'],result['down_monotonic'])
|
|
self.assertEqual(r.report['held_events'][1]['monotonic'],result['up_monotonic'])
|
|
self.assertLess(result['up_monotonic'],result['held_deadline_monotonic'])
|
|
self.assertEqual(r.report['held_release_errors'],[])
|
|
def test_uncertain_down_cleanup_retains_original_code_and_one_attempt(self):
|
|
r=Rig();r.fail_down=True
|
|
with self.assertRaisesRegex(ValueError,'uncertain'):r.run()
|
|
self.assertEqual([c[2] for c in r.calls],[1,0]);self.assertFalse(r.keys)
|
|
self.assertFalse(r.report['held_events'][0]['sent']);self.assertTrue(r.report['held_events'][1]['finally'])
|
|
self.assertFalse(r.report['held_action_proof']['completed'])
|
|
with self.assertRaises(ValueError):r.down()
|
|
def test_uncertain_up_and_lost_owner_never_redirect(self):
|
|
for fault in ('up','owner','connection','guard','pointer'):
|
|
r=Rig();r.down()
|
|
if fault=='up':r.fail_up=True
|
|
if fault=='owner':r.guard.owner['pid']=10
|
|
if fault=='connection':r.mouse.query=object()
|
|
if fault=='guard':r.guard.check=lambda:(_ for _ in ()).throw(h.Refused('original owner lost'))
|
|
if fault=='pointer':r.pointer_busy=True
|
|
with self.subTest(fault=fault),self.assertRaises(ValueError):r.up(cleanup=True)
|
|
self.assertEqual(r.report['unconfirmed_held_key'],'F');self.assertTrue(r.keys)
|
|
self.assertTrue(r.report['held_release_errors'])
|
|
if fault!='up':self.assertEqual(len(r.calls),1)
|
|
def test_mapping_capture_and_late_cleanup_use_original_keycode(self):
|
|
r=Rig();r.down();r.codes[0x66]=42;r.scene.sleep(6)
|
|
with self.assertRaises(ValueError):r.up()
|
|
r.up(cleanup=True);self.assertEqual(r.calls[-1][1],41);self.assertFalse(r.keys)
|
|
self.assertTrue(r.report['held_events'][-1]['finally'])
|
|
def test_native_final_guard_rechecks_six_seconds_pose_age_and_idle(self):
|
|
for fault in ('late','idle','pointer','foreign','mapping','lease','not_after'):
|
|
r=Rig()
|
|
if fault=='late':r.scene.guard_cost=.3
|
|
if fault=='idle':r.keys.add(25)
|
|
if fault=='pointer':r.pointer_busy=True
|
|
if fault=='foreign':r.mouse.owner['pid']=11
|
|
if fault=='mapping':r.codes[0x66]=42
|
|
with self.subTest(fault=fault),self.assertRaises(ValueError):
|
|
r.adapter.down_f(r.guard,deadline=40 if fault=='lease' else 35,not_after=40 if fault=='not_after' else 30.5)
|
|
self.assertEqual(r.calls,[])
|
|
def test_release_has_two_attempt_cap_and_no_other_adapter_methods(self):
|
|
r=Rig();r.down();r.fail_up=True
|
|
for _ in range(2):
|
|
with self.assertRaises(ValueError):r.up(True)
|
|
n=len(r.calls)
|
|
with self.assertRaisesRegex(ValueError,'repeated'):r.up(True)
|
|
self.assertEqual(len(r.calls),n)
|
|
for method in ('send','move','pulse'):self.assertFalse(hasattr(r.adapter,method))
|
|
|
|
class ObserverTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):cls.frames,*cls.proofs=raw_fixture()
|
|
def test_real_native_reader_keeps_world_turn_receipts_and_moving_original_geometry(self):
|
|
frames=copy.deepcopy(self.frames);selection=self.proofs[3];anchor=selection['final_sample']
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
p=Path(folder)/'console.log';p.write_text('');now=[0.]
|
|
o=h.Observer(p,'test',clock=lambda:now[0],enable_look=True,enable_approach=True,enable_selection=True,enable_held=True)
|
|
try:
|
|
text=raw(frames);lines=sorted(text.splitlines(True),key=lambda line:int(__import__('re').search(r'sample=(\d+)',line)[1]))
|
|
initial=[];future=[]
|
|
for line in lines:
|
|
import re
|
|
m=re.search(r'sample=(\d+)',line)
|
|
(initial if int(m[1])<=anchor else future).append(line)
|
|
now[0]=frames[anchor]['real_ms']/1000
|
|
with p.open('a') as f:f.write(''.join(initial))
|
|
while o.offset<p.stat().st_size:o.poll()
|
|
first=o.selection_snapshots();self.assertIn('turn',first[-1]);self.assertEqual(first[-1]['world_ms'],frames[anchor]['world_ms'] if 'world_ms' in frames[anchor] else frames[anchor]['real_ms'])
|
|
receipt=first[-1]['received_monotonic'];o.begin_held(anchor)
|
|
cursor=0;found=False
|
|
while cursor<len(future):
|
|
# One complete native sample at a time, preserving receipt time.
|
|
block=[];i=int(__import__('re').search(r'sample=(\d+)',future[cursor])[1])
|
|
while cursor<len(future) and int(__import__('re').search(r'sample=(\d+)',future[cursor])[1])==i:block.append(future[cursor]);cursor+=1
|
|
now[0]=frames[i]['real_ms']/1000
|
|
with p.open('a') as f:f.write(''.join(block))
|
|
batch=o.held_snapshots()
|
|
if batch[0]['sample']==anchor:self.assertEqual(batch[0]['received_monotonic'],receipt)
|
|
if batch[-1]['phase']==2:
|
|
found=True;self.assertNotEqual(batch[-1]['geometry']['context'],first[-1]['geometry']['context']);self.assertIn('world_ms',batch[-1])
|
|
self.assertTrue(found)
|
|
with self.assertRaises(ValueError):o.begin_held(anchor)
|
|
with self.assertRaises(ValueError):o.approach_data()
|
|
finally:o.close()
|
|
def test_cancelled_or_replaced_console_cannot_re_adopt(self):
|
|
for fault in ('replace','cancel'):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
p=Path(folder)/'console.log';p.write_text('');o=h.Observer(p,'test',enable_look=True,enable_approach=True,enable_selection=True,enable_held=True)
|
|
try:
|
|
if fault=='replace':p.rename(p.with_suffix('.old'));p.write_text('')
|
|
else:p.write_text('ERR INPUT OBSERVE: run=test sample=0 real=0 world=0 CANCELLED original-binding-changed\n')
|
|
with self.subTest(fault=fault),self.assertRaises(ValueError):o.held_data()
|
|
finally:o.close()
|
|
def test_compaction_preserves_raw_world_reference_without_duplicate_pose(self):
|
|
full={'sample':1,'real_ms':2,'world_ms':3,'received_monotonic':4.,'origin':[1,2,3],'turn':[0,0,0]}
|
|
report={'approach_requested':True,'held_action_requested':True,'held_action_proof':{'samples':[full]}}
|
|
compact=h.report_for_write(report)
|
|
self.assertEqual(compact['held_action_proof']['samples'],[{k:full[k] for k in ('sample','real_ms','world_ms','received_monotonic')}]);self.assertIn('origin',full)
|
|
|
|
class FullIntegrationTests(unittest.TestCase):
|
|
def test_actual_f_adapter_policy_and_unmocked_final_raw_verifier(self):
|
|
from test_prison_coarse_facing import RawTests
|
|
from ordinary_gate_choices import join_choice_snapshots
|
|
RawTests.setUpClass()
|
|
frames,look,route,facing,selection=copy.deepcopy(RawTests.fixture)
|
|
scene=Scene();end=selection['samples'][-1]
|
|
scene.now=selection['finished_monotonic'];scene.next_sample=(end['real_ms']-2000)/1000+.1
|
|
scene.geometry=copy.deepcopy(selection['geometry']);scene.origin=list(end['origin'])
|
|
scene.camera=list(end['camera']);scene.direction=list(end['direction']);scene.actions=copy.deepcopy(selection['actions'])
|
|
make=scene.make
|
|
def snapshot():
|
|
item=make();item['sample']=int(round((2000+scene.now*1000)/100));item['real_ms']=2000+int(round(scene.now*1000));item['world_ms']=item['real_ms']
|
|
item['choices'].update(nonce='test',sample=item['sample'],real_ms=item['real_ms'])
|
|
if item['phase']==1:item['choices']['selected_id']=scene.actions[1]['id']
|
|
return item
|
|
scene.make=snapshot;scene.snapshot=copy.deepcopy(end);scene.snapshot.update(turn=[0,0,0],world_ms=end['real_ms'])
|
|
scene.history=[copy.deepcopy(scene.snapshot)];scene.last_read=end['sample']-1
|
|
for item in selection['samples']:item.update(turn=[0,0,0],world_ms=item['real_ms'])
|
|
scene.report={'look_proof':look,'approach_proof':route,'facing_proof':facing,'action_selection_proof':selection}
|
|
rig=Rig(scene);report=rig.report
|
|
def content():
|
|
for item in scene.history:frames[item['sample']]=copy.deepcopy(item)
|
|
text=raw(frames);return text,join_choice_snapshots(text,'test')
|
|
def begin(anchor):
|
|
self.assertEqual(anchor,selection['final_sample']);scene.last_read=anchor-1
|
|
observer=SimpleNamespace(enable_held=True,nonce='test',ready_at=0.,ready_real=2000,queued_real=0,
|
|
approach_data=content,held_data=content,begin_held=begin,held_snapshots=scene.observe)
|
|
with patch.object(h,'notify',side_effect=scene.notifications.append):
|
|
h.run_held_action(rig.guard,rig.keyboard,observer,rig.mouse,report,scene.clock,scene.sleep)
|
|
self.assertTrue(report['held_action_proof']['completed']);self.assertEqual([c[2] for c in rig.calls],[1,0])
|
|
self.assertEqual(report['held_release_errors'],[]);self.assertFalse(rig.keys)
|
|
self.assertTrue(any(e['kind']=='held-key-pending' for e in scene.notifications))
|
|
|
|
class SupervisorTests(unittest.TestCase):
|
|
def pending(self):
|
|
s=h.HeldSupervisor(True);s.consume({'kind':'held-observer-deadline','deadline_monotonic':44.5},2)
|
|
s.consume({'kind':'held-action-pending','deadline_monotonic':44.0},30)
|
|
s.consume({'kind':'held-key-pending','key':'F','deadline_monotonic':35},30)
|
|
return s
|
|
def test_five_second_lease_does_not_replace_other_key_supervision(self):
|
|
s=self.pending();self.assertFalse(s.expired(31.6));self.assertTrue(s.expired(35));self.assertEqual(s.key,'F')
|
|
self.assertFalse(s.consume({'kind':'release-confirmed','key':'W'},31.7));self.assertEqual(s.key,'F')
|
|
s.cleanup_ack({'kind':'held-key-release-confirmed','key':'F'});self.assertIsNone(s.key)
|
|
def test_f_and_observer_deadlines_cannot_be_refunded_or_renewed(self):
|
|
for event in ({'kind':'held-key-pending','key':'F','deadline_monotonic':36},{'kind':'held-action-pending','deadline_monotonic':44.5},{'kind':'held-observer-deadline','deadline_monotonic':45},{'kind':'held-key-release-confirmed','key':'W'}):
|
|
s=self.pending();s.consume(event,31);self.assertTrue(s.expired(31))
|
|
s=self.pending();s.consume({'kind':'held-key-release-confirmed','key':'F'},34);self.assertFalse(s.expired(43.9));self.assertTrue(s.expired(44))
|
|
self.assertFalse(s.expired(44,proof_finished=True))
|
|
def test_unrequested_missing_outer_lease_busy_or_late_ack_refuses(self):
|
|
s=h.HeldSupervisor(False);s.consume({'kind':'held-observer-deadline','deadline_monotonic':44},2);self.assertTrue(s.failed)
|
|
for before in ('missing','busy'):
|
|
s=h.HeldSupervisor(True)
|
|
if before=='busy':s.consume({'kind':'held-observer-deadline','deadline_monotonic':44},2)
|
|
s.consume({'kind':'held-key-pending','key':'F','deadline_monotonic':35},30,other_pending=before=='busy');self.assertTrue(s.failed)
|
|
s=self.pending();s.consume({'kind':'held-key-release-confirmed','key':'F'},35);self.assertTrue(s.failed);self.assertIsNone(s.key)
|
|
def test_existing_supervise_stops_stuck_f_and_retains_uncertainty(self):
|
|
events=[{'kind':'held-observer-deadline','deadline_monotonic':44},{'kind':'held-key-pending','key':'F','deadline_monotonic':5.2}]
|
|
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
|
|
def start(self):pass
|
|
def is_alive(self):return self.alive
|
|
def terminate(self):pass
|
|
def join(self,*a):self.alive=False
|
|
def kill(self):self.alive=False
|
|
process=Process();pipe=Pipe();clock=[0.]
|
|
def ticking():clock[0]+=.1;return clock[0]
|
|
context=SimpleNamespace(Pipe=lambda **kw:(pipe,pipe),Process=lambda **kw:process)
|
|
with patch.object(h.multiprocessing,'get_context',return_value=context),patch.object(h.time,'monotonic',side_effect=ticking),patch.object(h.signal,'signal'),patch.object(h,'stop_exact_child'),patch.object(h,'supervision_failure') as failed:
|
|
self.assertEqual(h.supervise(['--enable-private-held-action-proof']),1)
|
|
self.assertEqual(failed.call_args.args[1],'F');self.assertGreaterEqual(clock[0],5.2)
|
|
|
|
class FlagTests(unittest.TestCase):
|
|
def test_exact_native_nonce_and_explicit_flag_chain(self):
|
|
base=['--enable-private-keyboard-proof','--prefix','/x','--game-executable','/g','--profile','/p','--report','/r','--host-display',':0','--nonce','N']
|
|
flags=['--enable-private-look-proof','--enable-private-approach-proof','--enable-private-alignment-proof','--enable-private-selection-proof','--enable-private-coarse-approach-proof','--enable-private-coarse-facing-proof','--enable-private-held-action-proof']
|
|
self.assertFalse(h.arguments(base).enable_private_held_action_proof)
|
|
args=h.arguments(base+flags)
|
|
for missing in flags[:-1]:
|
|
with self.assertRaises(SystemExit),patch('sys.stderr'):h.arguments(base+[f for f in flags if f!=missing])
|
|
native=['-errOrdinaryGateObserve','N','-errOrdinaryGateChoices','N','-errOrdinaryHeldAction','N'];h.gate_options(native,args)
|
|
for extra in ([],['-errOrdinaryHeldAction','wrong'],['-errOrdinaryHeldActionAlias','N'],['-errOrdinaryHeldAction','N']*2):
|
|
with self.assertRaises(ValueError):h.gate_options(native[:-2]+extra,args)
|
|
with self.assertRaises(ValueError):h.gate_options(native,h.arguments(base+flags[:-1]))
|
|
def test_sequence_finally_preserves_later_uncertain_f_over_clean_approach(self):
|
|
observer=SimpleNamespace(events=['READY'],poll=lambda:None,fresh=lambda:None,ready_at=0.,ready_real=2000,queued_real=0,
|
|
enable_approach=True,enable_selection=True,enable_held=True)
|
|
report={'held_action_requested':True,'coarse_facing_requested':True,'coarse_approach_requested':True}
|
|
guard=SimpleNamespace(check=lambda:None);keyboard=SimpleNamespace(require_idle=lambda:None)
|
|
def approach(*args):report['approach_proof']={'completed':True,'unconfirmed_held_key':None,'paired_release_errors':[]}
|
|
def held(*args):
|
|
report['unconfirmed_held_key']='F';report['paired_release_errors'].append('captured owner lost during F release')
|
|
raise h.Refused('F remains uncertain')
|
|
with patch.object(h,'perform_look'),patch.object(h,'run_approach',side_effect=approach),patch.object(h,'run_selection'),patch.object(h,'run_held_action',side_effect=held),patch.object(h,'notify'):
|
|
with self.assertRaisesRegex(ValueError,'F remains uncertain'):
|
|
h.sequence(guard,keyboard,observer,report,clock=lambda:1.,sleep=lambda n:None,mouse=object(),approach=True,alignment=True,selection=True,coarse_approach=True,coarse_facing=True,held_action=True)
|
|
self.assertEqual(report['unconfirmed_held_key'],'F');self.assertFalse(report['all_attempted_presses_released']);self.assertFalse(report['sequence_completed'])
|
|
|
|
def test_no_dynamic_held_capability_from_report_alone(self):
|
|
with self.assertRaisesRegex(ValueError,'Held action'):
|
|
h.sequence(None,None,None,{'held_action_requested':True},held_action=False)
|
|
with self.assertRaises(ValueError):h.Observer(Path('/missing'),'N',enable_held=True)
|
|
|
|
if __name__=='__main__':unittest.main()
|