195 lines
14 KiB
Python
195 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""Synthetic evidence ONLY: disposable end-to-end check() positives and corruptions.
|
|
|
|
No Godot/OS execution, no actual acceptance proof. Mutations update hashes so the
|
|
semantic gate, not merely a stale digest, must reject invalid state.
|
|
"""
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
import check_p08_migration_evidence as c
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
BANNER = 'Godot Engine v4.7.2.stable.official.ed1daf0bf - https://godotengine.org\n'
|
|
|
|
def typed(value):
|
|
if isinstance(value,dict): return {'type':'Dictionary','entries':[[typed(k),typed(v)] for k,v in value.items()]}
|
|
if isinstance(value,list): return {'type':'Array','element_type':0,'element_class':'','items':[typed(x) for x in value]}
|
|
kind = 'Nil' if value is None else 'bool' if type(value) is bool else 'int' if type(value) is int else 'String'
|
|
return {'type':kind,'text':json.dumps(value)}
|
|
|
|
class Proof:
|
|
def __init__(self, path):
|
|
self.path = path
|
|
runner = c.module('p08_synthetic_source',ROOT/'tools/run_all_tests.py')
|
|
self.runner = runner
|
|
source = {n:p['sha256'] for n,p in runner.source_refs(ROOT).items()}
|
|
self.index = {'schema':'p08-evidence-v1','commit':'e196578c1a4ad2fdf84642565bd102e524be386f',
|
|
'source_map':source,'spec_sha256':c.SPEC_SHA,'marker_contract_sha256':c.sha(ROOT/c.CONTRACT),
|
|
'inputs':{str(p.relative_to(ROOT)):c.sha(p) for p in (ROOT/'docs/spec/inputs/p08-02').iterdir() if p.is_file()},
|
|
'cases':[],'conditions':{k:[x for x in c.CASES if k!='unsolvable_preservation' or x.endswith('refusal')] for k in c.CONDITIONS},
|
|
'artifacts':{},'acceptance':'pending','collection_error':None,'phases':[],
|
|
'os_proof':'os.json','regression':'regression.json'}
|
|
for number,name in enumerate(c.CASES): self.case(name,number)
|
|
self.os(); self.regression(); self.save()
|
|
def put(self,name,value):
|
|
path = self.path/name; path.parent.mkdir(parents=True,exist_ok=True)
|
|
path.write_bytes(value if isinstance(value,bytes) else json.dumps(value).encode())
|
|
self.index['artifacts'][name] = {'bytes':path.stat().st_size,'sha256':c.sha(path)}
|
|
def variant(self,prefix,name,value):
|
|
node = typed(value)
|
|
self.put(prefix+name+'.json',node)
|
|
# Deliberately not a native Variant: this is an isolated Python proof fixture.
|
|
self.put(prefix+name+'.var',json.dumps(node).encode())
|
|
def save(self): (self.path/'index.json').write_text(json.dumps(self.index))
|
|
def state(self,operations):
|
|
jobs = [{'job_id':j,'state':'PLANNED' if j=='j2' else 'BUILDING','progress':0 if j=='j2' else 466,'reserved_cells':[[x,2,0] for x in range(offset,offset+6)],'construction_cost':{'gold':1}}
|
|
for j,offset in [('j1',0),('j2',10)]]
|
|
facilities = [{'instance_id':'dorm','occupied_cells':[[30,0,1]]},{'instance_id':'clinic','occupied_cells':[[31,0,1]]}]
|
|
guests = [{'guest_id':g} for g in ['p08.walker']+['g'+str(n) for n in range(5)]]
|
|
staff = [{'staff_id':'patient','health':0},{'staff_id':'carrier'},{'staff_id':'staff_gina','carried_inventory_id':'cargo'}]
|
|
r = {'operations':{'enabled':operations},'guests':{'guests':guests},'staff':{'staff':staff,'tasks':[
|
|
{'task_id':'rescue','target_entity_id':'patient','rescue_facility_id':'clinic','state':'MOVING','assigned_staff_id':'carrier'}]},
|
|
'construction':{'jobs':jobs,'facilities':facilities},'facilities':{'instances':copy.deepcopy(facilities)},
|
|
'world':{'occupancies':[{'entity_id':j['job_id'],'cells':copy.deepcopy(j['reserved_cells'])} for j in jobs]+
|
|
[{'entity_id':f['instance_id'],'cells':copy.deepcopy(f['occupied_cells'])} for f in facilities]},
|
|
'inventory':{'inventories':[{'owner_inventory_id':x,'stacks':[{'item_id':'ingredient/raw_food','quantity':4}]} for x in ('source','dest','cargo')],
|
|
'reservations':[{'reservation_id':'reserve'}]},
|
|
'logistics_requests':[{'request_id':'food','source_inventory_id':'source','destination_inventory_id':'dest','assigned_carrier_id':'staff_gina','reservation_ids':['reserve'],'state':'IN_TRANSIT'}],
|
|
'movement_credits':[{'credit':1000}],
|
|
'elevators':{'trips':[{'request_id':'ev'+str(n),'actor_id':'g'+str(n)} for n in range(5)],'shafts':[{'boarded':['g0','g1','g2','g3']}]}}
|
|
return {'systems':{'runtime':r}}
|
|
def phase(self,name,role,pid):
|
|
folder = 'runs/'+name+'/'; test = 'tests/fixtures/p08-02/'+role+'.gd'
|
|
command = ['synthetic-native','--script',test]
|
|
start = {'actual_pid':pid,'command':command,'live_identity_exact':True,'event':'ENGINE_STARTED','owner_pid':1,'monotonic_ns':1}
|
|
end = dict(start,event='ENGINE_ENDED',monotonic_ns=2,exit_code=0,engine_ended=True)
|
|
self.put(folder+name+'/started.json',start); self.put(folder+name+'/ended.json',end)
|
|
self.put(folder+name+'/outer-owned-finally.json',{'actual_captured_pids':[pid],'all_captured_ended':True,'outer_cleanup_errors':[],'foreign_process_or_group_signals':0})
|
|
self.put(folder+'raw.log',(BANNER+'P08 mixed '+role+': 1 checks, 0 failures\n').encode())
|
|
receipt = {'phase':name,'actual_pid':pid,'command':command,'exit_code':0,'engine_ended':True,'test':test,
|
|
'source_sha256':self.index['source_map'][test],'log':'raw.log',
|
|
'native_process':{'exit_code':0,'helper_exit_code':0,'timed_out':False,'lifecycle_error':None},
|
|
'actual_private_profile':{'report':{'fresh_before_parent_initialize':True,'profile_writes_before_guard':0,'actual_user_data_dir':'diagnostic-private-'+name,'expected_user_data_dir':'diagnostic-private-'+name}}}
|
|
self.put(folder+name+'-receipt.json',receipt)
|
|
return folder+name+'-receipt.json'
|
|
def case(self,name,number):
|
|
operations = name.startswith('on-'); refusal = name.endswith('refusal'); prefix = 'pairs/'+name+'/'
|
|
phases = [self.phase(name+'-'+role,role,100+2*number+i) for i,role in enumerate(('writer','reader'))]
|
|
self.index['cases'].append({'id':name,'operations':'on' if operations else 'off','kind':'refusal' if refusal else 'positive','directory':prefix[:-1],'phases':phases})
|
|
self.index['phases'] += phases
|
|
identity = {'starting_job':'j2','diagnostic_budget':{'seed':42588,'food_quantity':8,'food_cost':24,'signing':1,'costs':[{'cost':{'gold':2,'materials':1}}],'initial':{'gold':0,'materials':0},'needed':{'gold':27,'materials':1},'deltas':{'gold':27,'materials':1}},'jobs':['j1','j2'],'dorm':'dorm','rescue_task':'rescue','logistics_requests':['food'],'ev_requests':['ev'+str(n) for n in range(5)]}
|
|
original = self.state(operations)
|
|
self.variant(prefix,'identity',identity); self.variant(prefix,'original',original); self.variant(prefix,'after-live',original)
|
|
before = {}; after = {}
|
|
for member in c.MEMBERS:
|
|
raw = ('synthetic '+member).encode()
|
|
for package in ('original','result','backup'): self.put(prefix+package+'/formal/'+member,raw)
|
|
before['/'+member] = after['/'+member] = hashlib.sha256(raw).hexdigest()
|
|
before['.pre_migration.bak/'+member] = after['.pre_migration.bak/'+member] = hashlib.sha256(raw).hexdigest()
|
|
self.variant(prefix,'before-members',before); self.variant(prefix,'after-members',after)
|
|
checkpoint = copy.deepcopy(original)
|
|
if not refusal:
|
|
r = checkpoint['systems']['runtime']
|
|
for job in r['construction']['jobs']: job['reserved_cells'] = job['reserved_cells'][:4]
|
|
for row in r['world']['occupancies'][:2]: row['cells'] = row['cells'][:4]
|
|
events = [{'event_type':'ELEVATOR_ARRIVED','payload':{'request_id':rid}} for rid in identity['ev_requests']]
|
|
for i,role in enumerate(('writer','reader')):
|
|
self.variant(prefix,role+'-process',{'pid':100+2*number+i,'failures':0,'checks':1,'operations':operations,'refusal':refusal})
|
|
for label in c.LABELS:
|
|
state = copy.deepcopy(checkpoint)
|
|
if label == 'minute-160':
|
|
state['systems']['runtime']['staff']['staff'][0]['health'] = 100
|
|
state['systems']['runtime']['logistics_requests'][0]['state'] = 'COMPLETED'
|
|
self.variant(prefix,role+'-'+label,events if label=='events' else state)
|
|
def os(self):
|
|
proof = {'commit':self.index['commit'],'source_map':self.index['source_map'],'input_method':'native_os','setup':'controlled_old_save','actions':[]}
|
|
for name in ('preview','cancel','confirm','changed_source_rejection'):
|
|
row = {'action':name,'input_receipt':'os/'+name+'/input.json'}
|
|
self.put(row['input_receipt'],{'method':'native_os','event_type':'mouse','target_pid':500,'timestamp_ns':1})
|
|
for field in ('screen_before','screen_after','model_before','model_after','formal_before','formal_after'):
|
|
row[field] = 'os/'+name+'/'+field+'.json'; self.put(row[field],{'synthetic':'diagnostic'})
|
|
proof['actions'].append(row)
|
|
self.put('os.json',proof)
|
|
def regression(self):
|
|
contracts = c.module('p08_synthetic_markers',ROOT/'tools/strict_test_markers.py').load_contracts(ROOT/c.CONTRACT)
|
|
roster = self.runner.roster(ROOT); rows = []
|
|
for i,test in enumerate(roster):
|
|
contract = contracts[test]
|
|
def marker(template):
|
|
for key in ('checks','passed'): template = template.replace('<'+key+'>','1')
|
|
for key in ('failures','failed','skipped'): template = template.replace('<'+key+'>','0')
|
|
return template+'\n'
|
|
raw = BANNER+marker(contract['template'])
|
|
for auxiliary in contract.get('auxiliary',[]): raw += marker(auxiliary['template'])
|
|
if contract.get('json_report'):
|
|
raw += contract['json_report']['prefix']+json.dumps({'check_summary':{'total':1,'passed':1,'failed':0,'skipped':0},'status':'PASS','runtime_errors':[],'unmet_fixture_fields':[]})+'\n'
|
|
log = str(i)+'.log'; self.put('regression/'+log,raw.encode())
|
|
rows.append({'test':test,'log':log,'exit_code':0,'engine_ended':True})
|
|
self.put('regression/input.json',{'selected_source_map':self.index['source_map'],'roster':roster})
|
|
self.put('regression/ended.json',{'error':None,'cleanup_errors':[],'guard_errors':[],'phases':rows})
|
|
self.put('regression.json',{'commit':self.index['commit'],'directory':'regression','input':'regression/input.json','ended':'regression/ended.json'})
|
|
|
|
class EvidenceTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.temp = tempfile.TemporaryDirectory(prefix='p08-checker-')
|
|
cls.proof = Proof(Path(cls.temp.name))
|
|
cls.baseline = copy.deepcopy(cls.proof.index)
|
|
@classmethod
|
|
def tearDownClass(cls): cls.temp.cleanup()
|
|
def test_positive_reaches_whole_check(self):
|
|
self.assertEqual(c.check(ROOT,self.proof.path),[])
|
|
def test_corruptions_cannot_be_accepted(self):
|
|
p = self.proof
|
|
mutations = {
|
|
'duplicate_actor':lambda r:r['guests']['guests'].append(copy.deepcopy(r['guests']['guests'][0])),
|
|
'orphan_occupancy':lambda r:r['world']['occupancies'][0].update(entity_id='orphan'),
|
|
'lost_patient':lambda r:r['staff']['staff'].pop(0),
|
|
'lost_cargo':lambda r:r['inventory']['inventories'].pop(),
|
|
'lost_ev':lambda r:r['elevators']['trips'].pop(),
|
|
'missing_start_stage':lambda r:r['construction']['jobs'][1].update(state='BUILDING',progress=1),
|
|
}
|
|
path = 'pairs/off-positive/original.json'
|
|
original = (p.path/path).read_bytes()
|
|
for label,mutate in mutations.items():
|
|
with self.subTest(label=label):
|
|
node = c.decode(json.loads(original)); mutate(node['systems']['runtime'])
|
|
p.put(path,typed(node)); p.save()
|
|
self.assertTrue(c.check(ROOT,p.path),label)
|
|
p.put(path,original); p.index = copy.deepcopy(self.baseline); p.save()
|
|
path = 'pairs/off-positive/identity.json'
|
|
original = (p.path/path).read_bytes()
|
|
node = c.decode(json.loads(original)); node['diagnostic_budget']['deltas']['gold'] += 1
|
|
p.put(path,typed(node)); p.save()
|
|
self.assertTrue(c.check(ROOT,p.path),'resource_delta_mismatch')
|
|
p.put(path,original); p.index = copy.deepcopy(self.baseline); p.save()
|
|
paths = [('partial_batch','pairs/off-refusal/after-live.json'),('unequal_cross_pid','pairs/on-positive/reader-checkpoint.json'),
|
|
('overwritten_original','pairs/off-refusal/original/formal/world.bin'),('overwritten_backup','pairs/off-refusal/backup/formal/world.bin')]
|
|
for label,path in paths:
|
|
with self.subTest(label=label):
|
|
original = (p.path/path).read_bytes()
|
|
p.put(path,b'{}'); p.save(); self.assertTrue(c.check(ROOT,p.path),label)
|
|
p.put(path,original); p.index = copy.deepcopy(self.baseline); p.save()
|
|
for label in ('missing_on','wrong_source','stale_source','fake_os_signal','missing_phase','duplicate_pid'):
|
|
with self.subTest(label=label):
|
|
changed = None
|
|
if label=='missing_on': p.index['cases'] = p.index['cases'][:2]
|
|
elif label=='wrong_source': p.index['source_map']['autoload/simulation_runtime.gd'] = '0'*64
|
|
elif label=='stale_source': del p.index['source_map']['tools/check_p08_migration_evidence.py']
|
|
elif label=='missing_phase': p.index['cases'][0]['phases'].pop()
|
|
elif label=='fake_os_signal':
|
|
changed = 'os/preview/input.json'; original = (p.path/changed).read_bytes()
|
|
row = json.loads(original); row['method'] = 'signal_emitted'; p.put(changed,row)
|
|
else:
|
|
changed = p.index['cases'][0]['phases'][1]; original = (p.path/changed).read_bytes()
|
|
row = json.loads(original); row['actual_pid'] = 100; p.put(changed,row)
|
|
p.save(); self.assertTrue(c.check(ROOT,p.path),label)
|
|
if changed: p.put(changed,original)
|
|
p.index = copy.deepcopy(self.baseline); p.save()
|
|
|
|
if __name__ == '__main__': unittest.main()
|