208 lines
14 KiB
Python
208 lines
14 KiB
Python
"""Actual facing policy, capabilities, raw validators and runner; fake input only."""
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
import math
|
|
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 prison_alignment_proof as policy
|
|
import prison_alignment_evidence as evidence
|
|
import prison_action_selection_evidence as selected_evidence
|
|
import headless_input_proof as helper
|
|
import headless_game as runner
|
|
from test_prison_action_selection_proof import ChoiceScene
|
|
from test_prison_action_selection_evidence import build,raw
|
|
from test_prison_approach_proof import Scene as ApproachScene
|
|
import test_headless_input_exercise_runner as runner_fixtures
|
|
import test_relative_look_runner as selection_runner_fixtures
|
|
|
|
|
|
def coarse_move(s,dx,dy,guard,*,deadline):
|
|
if type(dx) is not int or not 1<=abs(dx)<=384 or dy!=0:raise AssertionError('coarse fake capability')
|
|
guard.check();guard.check()
|
|
if s.now>=deadline:raise policy.Refused('adapter deadline')
|
|
s.motions.append((dx,dy,s.now,deadline))
|
|
if not s.ignore:s.pending.append((s.now+s.delay,dx*s.gain*(-1 if s.reverse else 1)))
|
|
|
|
|
|
def coarse(s):
|
|
s.report.update(coarse_facing_requested=True,coarse_approach_requested=True,selection_requested=True)
|
|
return policy.perform_facing(s,s.idle,s,s.observe,s.geometry,s.calibration,s.accepted,s.report,
|
|
clock=s.clock,sleep=s.sleep,notify=s.notifications.append,profile='coarse')
|
|
|
|
|
|
class FacingTests(unittest.TestCase):
|
|
def scene(self):
|
|
s=ChoiceScene();s.report['approach_proof']['profile']='coarse'
|
|
s.move=lambda *a,**kw:coarse_move(s,*a,**kw)
|
|
return s
|
|
def test_coarse_reduces_requests_without_new_clock_or_total_budget(self):
|
|
standard=ChoiceScene();ordinary=standard.facing()
|
|
s=self.scene();result=coarse(s)
|
|
self.assertTrue(result['completed']);self.assertEqual(result['profile'],'coarse')
|
|
self.assertNotIn('profile',ordinary)
|
|
self.assertLess(len(s.motions),len(standard.motions));self.assertTrue(any(abs(m[0])>128 for m in s.motions))
|
|
self.assertLessEqual(len(s.motions),6);self.assertLessEqual(sum(abs(m[0]) for m in s.motions),768)
|
|
self.assertEqual(result['deadline_monotonic'],ordinary['deadline_monotonic'])
|
|
self.assertLess(s.now,standard.now);self.assertEqual(s.wheels,[])
|
|
self.assertEqual(s.origin,standard.origin)
|
|
def test_profile_and_explicit_prerequisites_refuse_before_input(self):
|
|
for profile in (True,None,'unknown','coarse'):
|
|
s=self.scene()
|
|
with self.subTest(profile=profile),self.assertRaises(ValueError):
|
|
policy.perform_facing(s,s.idle,s,s.observe,s.geometry,s.calibration,s.accepted,s.report,clock=s.clock,sleep=s.sleep,notify=s.notifications.append,profile=profile)
|
|
self.assertEqual(s.motions,[])
|
|
s=self.scene();s.report['approach_proof']['profile']='standard'
|
|
with self.assertRaisesRegex(ValueError,'accepted coarse'):coarse(s)
|
|
def test_coarse_native_overshoot_body_and_clock_guards(self):
|
|
for fault in ('overshoot','reverse','moving','deadline','focus'):
|
|
s=self.scene()
|
|
if fault=='overshoot':s.gain=.5
|
|
if fault=='reverse':s.reverse=True
|
|
if fault=='moving':s.input=[0,0,1];s.sleep(.1)
|
|
if fault=='deadline':s.calibration['alignment_deadline_monotonic']=31.15
|
|
if fault=='focus':s.focus=False
|
|
with self.subTest(fault=fault),self.assertRaises(ValueError):coarse(s)
|
|
self.assertFalse(s.report.get('facing_proof',{}).get('completed'));self.assertEqual(s.wheels,[])
|
|
def test_policy_refuses_third_384_request_before_write(self):
|
|
s=self.scene();s.angle-=140;s.gain=.011
|
|
response=24*.011*math.pi/180;s.calibration['response_direction']=[math.sin(response),0,math.cos(response)]
|
|
s.snapshot=s.make();s.history=[copy.deepcopy(s.snapshot)]
|
|
with self.assertRaisesRegex(ValueError,'cumulative pixel'):coarse(s)
|
|
self.assertEqual([abs(m[0]) for m in s.motions],[384,384])
|
|
|
|
|
|
class Fn:
|
|
def __init__(self,fn):self.fn=fn
|
|
def __call__(self,*a):return self.fn(*a)
|
|
|
|
|
|
class CapabilityTests(unittest.TestCase):
|
|
def rig(self,coarse_enabled=True):
|
|
events=[];now=[1.];target={'geometry':{'root':1},'focused_clipped_bounds':(0,0,2000,720)}
|
|
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 c:9)
|
|
keyboard=SimpleNamespace(xt=xt,require_idle=lambda:None)
|
|
mouse=helper.RelativeMouse(query,keyboard,{'window':100,'pid':9},enable_approach=True,enable_alignment=True,enable_coarse_approach=True,enable_coarse_facing=coarse_enabled,enable_selection=coarse_enabled)
|
|
mouse.pointer=lambda:{'root':1,'child':0,'root_x':1000,'root_y':300,'mask':0}
|
|
return mouse,SimpleNamespace(check=lambda:target),events,now
|
|
def move(self,m,g,n,dx,dy=0):m.alignment_move(dx,dy,g,deadline=9,clock=lambda:n[0])
|
|
def test_total_retired_before_actual_or_uncertain_write(self):
|
|
m,g,events,n=self.rig()
|
|
for dx in (384,383):self.move(m,g,n,dx)
|
|
self.assertEqual(m.alignment_pixels,767)
|
|
with self.assertRaisesRegex(ValueError,'budget'):self.move(m,g,n,2)
|
|
self.move(m,g,n,1);self.assertEqual(len(events),3)
|
|
with self.assertRaisesRegex(ValueError,'budget'):self.move(m,g,n,1)
|
|
m,g,events,n=self.rig();m.xt.XTestFakeRelativeMotionEvent.fn=lambda *a:0
|
|
with self.assertRaises(ValueError):self.move(m,g,n,384)
|
|
self.assertEqual(m.alignment_pixels,384);self.assertEqual(m.alignment_requests,1)
|
|
def test_six_request_cap_and_default_128(self):
|
|
m,g,e,n=self.rig(False)
|
|
with self.assertRaises(ValueError):self.move(m,g,n,129)
|
|
for _ in range(6):self.move(m,g,n,1)
|
|
with self.assertRaisesRegex(ValueError,'budget'):self.move(m,g,n,1)
|
|
self.assertEqual(len(e),6)
|
|
def test_coarse_retains_pointer_foreign_idle_and_final_deadline_guards(self):
|
|
for fault in ('pointer','foreign','idle','deadline','pitch','bool','385'):
|
|
m,g,e,n=self.rig()
|
|
if fault=='pointer':m.pointer=lambda:{'root':1,'child':0,'root_x':1900,'root_y':300,'mask':0}
|
|
if fault=='foreign':m.pointer=lambda:{'root':1,'child':1,'root_x':1000,'root_y':300,'mask':0};m.query.server_pid=lambda c:10
|
|
if fault=='idle':m.keyboard.require_idle=lambda:(_ for _ in ()).throw(ValueError('busy'))
|
|
if fault=='deadline':
|
|
old=g.check
|
|
def late():n[0]=9;return old()
|
|
g.check=late
|
|
with self.subTest(fault=fault),self.assertRaises(ValueError):self.move(m,g,n,True if fault=='bool' else 385 if fault=='385' else 384,1 if fault=='pitch' else 0)
|
|
self.assertEqual(e,[]);self.assertEqual(m.alignment_pixels,0)
|
|
|
|
|
|
class RawTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
original_run=ApproachScene.run
|
|
with patch.object(ApproachScene,'run',lambda s,*a,**kw:original_run(s,*a,**dict(kw,profile='coarse'))),patch.object(ChoiceScene,'facing',coarse),patch.object(ChoiceScene,'move',coarse_move):
|
|
cls.fixture=build()
|
|
def result(self,change=None,expected='coarse'):
|
|
f,l,a,p,s=copy.deepcopy(self.fixture)
|
|
if change:change(f,l,a,p,s)
|
|
return selected_evidence.validate_selection_evidence(raw(f),'test',l,a,p,s,facing_profile=expected)
|
|
def test_actual_policy_chain_and_raw_parsers_require_independent_profile(self):
|
|
r=self.result();self.assertTrue(r['verified'],r);self.assertFalse(r['held_action_verified'])
|
|
for expected in ('standard',True,None,'unknown'):
|
|
r=self.result(expected=expected);self.assertFalse(r['verified'],r)
|
|
def test_relabelled_missing_or_foreign_profile_refuses(self):
|
|
for change in (lambda f,l,a,p,s:p.pop('profile'),lambda f,l,a,p,s:p.update(profile='standard'),lambda f,l,a,p,s:a.update(profile='standard')):
|
|
r=self.result(change);self.assertFalse(r['verified'],r)
|
|
def test_raw_coarse_intermediate_overshoot_and_budget_refuse(self):
|
|
def overshoot(f,l,a,p,s):
|
|
m=p['motions'][0];i=m['after_sample']+1;base=f[m['after_sample']]['direction'];angle=math.atan2(base[0],base[2])+math.radians(35.01)
|
|
f[i]['direction']=[math.sin(angle),0,math.cos(angle)]
|
|
for change in (overshoot,lambda f,l,a,p,s:p['motions'][0].update(dx=385),lambda f,l,a,p,s:p.update(deadline_monotonic=p['deadline_monotonic']+.01)):
|
|
r=self.result(change);self.assertFalse(r['verified'],r)
|
|
|
|
|
|
class RunnerTests(unittest.TestCase):
|
|
def test_explicit_flags_refuse_before_runner_mutations(self):
|
|
base=dict(exercise_prison_coarse_facing=True,exercise_prison_coarse_approach=True,exercise_prison_selection=True,exercise_prison_alignment=True,observe_prison_choices=True,exercise_prison_approach=True,exercise_relative_look=True,observe_prison_gate=True,exercise_player_input=True,observe_player_input=True,inspect_headless_display=True,project='chat',settle=50)
|
|
runner.validate_input_exercise(SimpleNamespace(**base))
|
|
for change in ({'exercise_prison_selection':False},{'exercise_prison_coarse_approach':False},{'exercise_prison_coarse_facing':1},{'packed_core':Path('/x')},{'project':'vanilla-control'}):
|
|
with patch.object(runner,'packed_request',side_effect=AssertionError('mutation reached')):
|
|
with self.assertRaises(ValueError):runner.run(SimpleNamespace(**(base|change)))
|
|
def test_helper_argv_requires_both_explicit_predecessors(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']
|
|
self.assertFalse(helper.arguments(base).enable_private_coarse_facing_proof)
|
|
self.assertTrue(helper.arguments(base+flags).enable_private_coarse_facing_proof)
|
|
for missing in flags[:-1]:
|
|
with self.assertRaises(SystemExit),patch('sys.stderr'):helper.arguments(base+[f for f in flags if f!=missing])
|
|
def test_command_carries_explicit_capability_and_sealed_context(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
root=Path(folder);(root/'tools').mkdir();profile=root/'profile';profile.mkdir()
|
|
for name in ('headless_input_proof.py','headless_display_probe.py',*helper.APPROACH_MODULES,*helper.ALIGNMENT_MODULES,*helper.SELECTION_MODULES):
|
|
(root/'tools'/name).write_text('# fixture')
|
|
base=['gamescope','--backend','headless','--','/proton','run','/game','-errOrdinaryInputObserve','N','-errOrdinaryLookObserve','N','-errOrdinaryGateObserve','N','-errOrdinaryGateChoices','N']
|
|
wrapped,display=runner.display_probe_command(base,root,Path('/compat'),profile,Path('/game'),'N',':0')
|
|
args=(wrapped,root,Path('/compat'),profile,Path('/game'),'N',':0',display)
|
|
for enabled in (False,True):
|
|
command,context=runner.input_exercise_command(*args,look=True,approach=True,alignment=True,selection=True,coarse_approach=True,coarse_facing=enabled)
|
|
self.assertEqual('--enable-private-coarse-facing-proof' in command,enabled)
|
|
self.assertEqual(context.get('coarse_facing_requested',False),enabled)
|
|
self.assertEqual(context['command'],wrapped[4:])
|
|
with self.assertRaises(ValueError):runner.input_exercise_command(*args,look=True,approach=True,alignment=True,selection=False,coarse_approach=True,coarse_facing=True)
|
|
def test_real_runner_forwards_launch_profile_to_full_raw_validator(self):
|
|
f,l,a,p,s=copy.deepcopy(RawTests.fixture)
|
|
for requested,recorded in ((True,True),(False,True),(True,False)):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
fixture=selection_runner_fixtures.SelectionRunner()
|
|
session,context,report,observation=fixture.make_selection(Path(folder))
|
|
context.update(nonce='test',coarse_approach_requested=True,coarse_facing_requested=requested)
|
|
report.update(nonce='test',coarse_approach_requested=True,coarse_facing_requested=recorded,look_proof=l,approach_proof=a,facing_proof=p,action_selection_proof=s)
|
|
report['events']=copy.deepcopy(a['events'])
|
|
report['wheel_events']=[{'button':5,'down':down,'sent':True,'monotonic':s['wheel'][key]} for down,key in ((True,'down_monotonic'),(False,'up_monotonic'))]
|
|
observation['nonce']='test';(session/'console.log').write_text(raw(f))
|
|
Path(context['report']).write_text(json.dumps(report))
|
|
status,result=runner.input_exercise_result('game-started',context,session,observation,{'verified':True})
|
|
if requested and recorded:
|
|
self.assertEqual(status,'game-started',result);self.assertTrue(result['verified'],result)
|
|
self.assertTrue(result['action_selection_proof']['verified'])
|
|
else:self.assertEqual(status,'input-exercise-incomplete');self.assertFalse(result['verified'])
|
|
|
|
def test_result_rejects_self_authorized_profile_before_raw_credit(self):
|
|
fixture=runner_fixtures.ExerciseTests()
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
session,context,report,observation=fixture.make_proof(Path(folder))
|
|
for flags in ({'coarse_facing_requested':True},{'coarse_facing_requested':1},{'facing_proof':{'profile':'coarse'}}):
|
|
Path(context['report']).write_text(json.dumps(report|flags))
|
|
status,result=runner.input_exercise_result('game-started',context,session,observation,{'verified':True})
|
|
self.assertEqual(status,'input-exercise-incomplete');self.assertFalse(result['verified'])
|
|
self.assertTrue('facing' in result['error'].lower() or 'action-selection' in result['error'].lower())
|
|
(session/'input-proof.json').unlink()
|
|
|
|
if __name__=='__main__':unittest.main()
|