502 lines
26 KiB
Python
502 lines
26 KiB
Python
#!/usr/bin/env python3
|
|
"""Whole-checker positive and semantic corruption tests; synthetic proof only."""
|
|
import copy
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
from types import SimpleNamespace
|
|
import shutil
|
|
import contextlib
|
|
import io
|
|
import os
|
|
import gzip
|
|
import hashlib
|
|
|
|
sys.dont_write_bytecode = True
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
def module(name, path):
|
|
spec = importlib.util.spec_from_file_location(name, path)
|
|
value = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(value)
|
|
return value
|
|
C = module('placement_checker_test', ROOT/'tools/check_p08_normal_placement_evidence.py')
|
|
F = module('placement_synthetic_fixture', ROOT/'tests/fixtures/p08-03/checker_fixture.py')
|
|
D = module('placement_collector_test', ROOT/'tools/run_p08_normal_placement.py')
|
|
|
|
class EvidenceTests(unittest.TestCase):
|
|
def setUp(self):
|
|
scratch = ROOT/'tests/fixtures/p08-03/.scratch'
|
|
scratch.mkdir(exist_ok=True)
|
|
self.temp = tempfile.TemporaryDirectory(dir=scratch)
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.root, self.out = F.make(Path(self.temp.name), ROOT, C)
|
|
|
|
def edit(self, path, callback):
|
|
p = self.out/path
|
|
data = json.loads(p.read_text())
|
|
callback(data)
|
|
F.put(p, data)
|
|
F.repin(self.out)
|
|
|
|
def reject(self, text):
|
|
errors = C.check(self.root, self.out)
|
|
self.assertTrue(errors, 'corrupted proof accepted')
|
|
self.assertIn(text, ' '.join(errors))
|
|
|
|
def test_positive_whole_checker(self):
|
|
self.assertEqual([], C.check(self.root, self.out))
|
|
|
|
def test_verify_only_cli_uses_build_output_directory(self):
|
|
output = io.StringIO()
|
|
with patch.dict(os.environ, {'BUILD_OUTPUT_DIR':str(self.out.parent)}), contextlib.redirect_stdout(output):
|
|
self.assertEqual(0,C.main(['--verify-only','--project-root',str(self.root)]))
|
|
self.assertEqual('pending',json.loads(output.getvalue())['acceptance'])
|
|
|
|
def test_verify_only_cli_rejects_semantic_corruption(self):
|
|
self.edit('pair/writer-actions.json', lambda d: d['actions'][0].update(status='配置可能'))
|
|
output = io.StringIO()
|
|
with patch.dict(os.environ, {'BUILD_OUTPUT_DIR':str(self.out.parent)}), contextlib.redirect_stdout(output):
|
|
self.assertEqual(1,C.main(['--verify-only','--project-root',str(self.root)]))
|
|
self.assertIn('reason mismatch',json.loads(output.getvalue())['deficits'])
|
|
|
|
def bootstrap_edit(self, mutate):
|
|
self.edit("os-pair/writer-bootstrap-checkpoint.json", mutate)
|
|
self.reject("bootstrap source-defined operation drift")
|
|
|
|
def test_bootstrap_missing_notification(self):
|
|
self.bootstrap_edit(lambda d: C.typed_member(d,"systems","runtime","domain_events","recent_events")["items"].pop())
|
|
|
|
def test_bootstrap_extra_notification(self):
|
|
self.bootstrap_edit(lambda d: C.typed_member(d,"systems","runtime","domain_events","recent_events")["items"].append(
|
|
copy.deepcopy(C.typed_member(d,"systems","runtime","domain_events","recent_events")["items"][-1])))
|
|
|
|
def test_bootstrap_counter_drift(self):
|
|
self.bootstrap_edit(lambda d: C.typed_member(d,"systems","runtime","id_generator","counters","domain_event").update(text="999"))
|
|
|
|
def test_bootstrap_payload_drift(self):
|
|
def mutate(d):
|
|
event = C.typed_member(d,"systems","runtime","domain_events","recent_events")["items"][-1]
|
|
C.typed_member(event,"payload","save_path").update(text='"/wrong/path"')
|
|
self.bootstrap_edit(mutate)
|
|
|
|
def test_bootstrap_unknown_field_drift(self):
|
|
self.bootstrap_edit(lambda d: d["entries"].append([F.typed("unknown"),F.typed(1)]))
|
|
|
|
def test_bootstrap_missing_raw(self):
|
|
(self.out/"os-pair/writer-bootstrap-checkpoint.var").unlink()
|
|
F.repin(self.out)
|
|
self.reject("writer-bootstrap-checkpoint.var")
|
|
|
|
def test_bootstrap_overwritten_raw(self):
|
|
shutil.copyfile(self.out/"os-pair/writer-checkpoint.var", self.out/"os-pair/writer-bootstrap-checkpoint.var")
|
|
F.repin(self.out)
|
|
self.reject("wrong bootstrap raw model")
|
|
|
|
def test_reason_mismatch(self):
|
|
self.edit('pair/writer-actions.json', lambda d: d['actions'][0].update(status='配置可能'))
|
|
self.reject('reason mismatch')
|
|
|
|
def test_world_cell_drift(self):
|
|
self.edit('pair/writer-actions.json', lambda d: d['actions'][0]['position'].update(selected_cell=[9, 9, 0]))
|
|
self.reject('world-cell drift')
|
|
|
|
def test_transformed_viewport_corruption(self):
|
|
self.edit('pair/writer-actions.json', lambda d: d['actions'][0]['position'].update(viewport=[1,2]))
|
|
self.reject('transformed viewport mismatch')
|
|
|
|
def test_stale_source(self):
|
|
(self.root/C.TEST).write_text('changed')
|
|
self.reject('stale source')
|
|
|
|
def test_stale_commit(self):
|
|
self.edit('index.json', lambda d: d.update(commit='0'*40))
|
|
self.reject('stale commit')
|
|
|
|
def test_stale_input(self):
|
|
(self.root/'docs/traceability/original/BASELINE.txt').write_text('changed')
|
|
self.reject('stale input')
|
|
|
|
def test_missing_os(self):
|
|
self.edit('index.json', lambda d: d.update(os_proof=None))
|
|
self.reject('missing artifact')
|
|
|
|
def test_fake_os_method(self):
|
|
self.edit('os-pair/unsupported-os.json', lambda d: d.update(method='handler'))
|
|
self.reject('fake OS provenance')
|
|
|
|
def test_fake_os_pid(self):
|
|
self.edit('os-pair/unsupported-os.json', lambda d: d.update(target_pid=555))
|
|
self.reject('fake OS provenance')
|
|
|
|
def test_same_pid_resume(self):
|
|
self.edit('runs/reader/reader-receipt.json', lambda d: d.update(actual_pid=101))
|
|
self.reject('duplicate actual PID')
|
|
|
|
def test_orphan_occupancy(self):
|
|
path = 'pair/writer-demolition-after.json'
|
|
model = C.decode(json.loads((self.out/path).read_text()))
|
|
model['systems']['runtime']['world']['occupancies'].append({'entity_id': 'ghost', 'cells': [[-8, 0, 0]]})
|
|
F.put(self.out/path, F.typed(model)); F.repin(self.out)
|
|
self.reject('orphan/lost occupancy')
|
|
|
|
def test_partial_demolition(self):
|
|
path = 'pair/writer-demolition-after.json'
|
|
model = C.decode(json.loads((self.out/path).read_text()))
|
|
model['systems']['runtime']['facilities']['instances'].append({'instance_id': 'f1', 'occupied_cells': [[-8, 0, 0]]})
|
|
F.put(self.out/path, F.typed(model)); F.repin(self.out)
|
|
self.reject('partial demolition')
|
|
|
|
def test_refusal_partial_resources(self):
|
|
path = 'pair/writer-refused_demolition-after.json'
|
|
model = C.decode(json.loads((self.out/path).read_text()))
|
|
model['game_state']['resources']['gold'] -= 1
|
|
F.put(self.out/path, F.typed(model)); F.repin(self.out)
|
|
self.reject('refusal partially applied')
|
|
|
|
def test_wrong_resume(self):
|
|
path = 'pair/reader-checkpoint.json'
|
|
original = (self.out/path).read_bytes()
|
|
model = C.decode(json.loads((self.out/path).read_text()))
|
|
model['clock']['absolute_minute'] += 1
|
|
F.put(self.out/path, F.typed(model)); F.repin(self.out)
|
|
self.reject('wrong resume model')
|
|
(self.out/path).write_bytes(original); F.repin(self.out)
|
|
self.edit('pair/reader-save-result-0001.json', lambda d: d.update(result=F.typed({'ok':False,'operation':'load','code':'UNSAFE_PATH'})))
|
|
self.reject('actual save/load failed')
|
|
self.edit('pair/reader-save-result-0001.json', lambda d: d.update(result=F.typed({'ok':True,'operation':'load'}), requested_path='/outside/formal'))
|
|
self.reject('save outside private user profile')
|
|
|
|
def test_notification_payload_is_recomputed_from_returned_dict(self):
|
|
path = 'pair/reader-checkpoint.json'
|
|
original = (self.out/path).read_bytes()
|
|
fields = ('ok','code','message','operation','save_path','reason','request_id',
|
|
'checkpoint_absolute_minute','duration_us','runtime_duration_us',
|
|
'selected_candidate_id','selected_candidate_kind','selected_generation',
|
|
'save_id','saved_absolute_minute')
|
|
for field in fields:
|
|
with self.subTest(field=field):
|
|
data = json.loads(original)
|
|
event = C.typed_member(data,'systems','runtime','domain_events','recent_events')['items'][-1]
|
|
value = C.typed_member(event,'payload',field)
|
|
decoded = C.decode(value)
|
|
changed = not decoded if isinstance(decoded,bool) else decoded+1 if isinstance(decoded,int) else decoded+'-drift'
|
|
value.update(C.scalar(changed,value['type']))
|
|
F.put(self.out/path,data); F.repin(self.out)
|
|
self.reject('source-defined operation drift')
|
|
(self.out/path).write_bytes(original); F.repin(self.out)
|
|
|
|
def test_notification_envelope_cannot_drift(self):
|
|
path = 'pair/writer-checkpoint.json'
|
|
original = (self.out/path).read_bytes()
|
|
for field in ('event_id','event_type','source_service','absolute_minute','causation_command_id','entity_ids'):
|
|
with self.subTest(field=field):
|
|
data = json.loads(original)
|
|
event = C.typed_member(data,'systems','runtime','domain_events','recent_events')['items'][-1]
|
|
value = C.typed_member(event,field)
|
|
if field == 'entity_ids': value['items'].append(C.scalar('unexpected-entity'))
|
|
else:
|
|
old = C.decode(value)
|
|
value.update(C.scalar(old+1 if isinstance(old,int) else old+'-drift',value['type']))
|
|
F.put(self.out/path,data); F.repin(self.out)
|
|
self.reject('source-defined operation drift')
|
|
(self.out/path).write_bytes(original); F.repin(self.out)
|
|
|
|
def test_notification_counter_must_increment_exactly_once(self):
|
|
def corrupt(data):
|
|
C.typed_member(data,'systems','runtime','id_generator','counters','domain_event')['text'] = '9'
|
|
self.edit('pair/reader-checkpoint.json',corrupt)
|
|
self.reject('source-defined operation drift')
|
|
|
|
def test_notification_published_ids_cannot_be_omitted(self):
|
|
self.edit('pair/writer-checkpoint.json',lambda d: C.typed_member(d,'systems','runtime','domain_events','published_ids')['items'].clear())
|
|
self.reject('source-defined operation drift')
|
|
|
|
def test_extra_notification_is_rejected(self):
|
|
def corrupt(data):
|
|
events = C.typed_member(data,'systems','runtime','domain_events','recent_events')['items']
|
|
events.append(copy.deepcopy(events[-1]))
|
|
self.edit('pair/reader-checkpoint.json',corrupt)
|
|
self.reject('source-defined operation drift')
|
|
|
|
def test_existing_event_history_cannot_be_rewritten(self):
|
|
def corrupt(data):
|
|
event = C.typed_member(data,'systems','runtime','domain_events','recent_events')['items'][0]
|
|
C.typed_member(event,'event_type').update(text='"REWRITTEN_HISTORY"')
|
|
self.edit('pair/reader-checkpoint.json',corrupt)
|
|
self.reject('source-defined operation drift')
|
|
|
|
def test_postsave_unrelated_resource_drift_is_rejected(self):
|
|
self.edit('pair/writer-checkpoint.json',lambda d: C.typed_member(d,'game_state','resources','gold').update(text='9999'))
|
|
self.reject('source-defined operation drift')
|
|
|
|
def test_postload_unknown_field_drift_is_rejected(self):
|
|
self.edit('pair/reader-checkpoint.json',lambda d: C.typed_member(d,'systems','runtime','inventory')['entries'].append([C.scalar('unknown'),C.scalar(1)]))
|
|
self.reject('source-defined operation drift')
|
|
|
|
def test_presave_must_match_read_only_formal_model(self):
|
|
self.edit('pair/writer-presave.json',lambda d: C.typed_member(d,'clock','absolute_minute').update(text='361'))
|
|
self.reject('wrong presave formal model')
|
|
|
|
def test_read_only_inspection_must_match_complete_formal_model(self):
|
|
self.edit('pair/reader-save-result-0002.json',lambda d: C.typed_member(d['result'],'snapshot','clock','absolute_minute').update(text='361'))
|
|
self.reject('wrong read-only formal model')
|
|
|
|
def test_inspection_cannot_lose_read_only_attribution(self):
|
|
self.edit('pair/writer-save-result-0002.json',lambda d: C.typed_member(d['result'],'read_only').update(text='false'))
|
|
self.reject('missing read-only formal model')
|
|
|
|
def test_returned_result_payload_tampering_is_rejected(self):
|
|
self.edit('pair/reader-save-result-0001.json',lambda d: C.typed_member(d['result'],'duration_us').update(text='999'))
|
|
self.reject('source-defined operation drift')
|
|
|
|
def test_notification_payload_type_drift_is_rejected(self):
|
|
def corrupt(data):
|
|
event = C.typed_member(data,'systems','runtime','domain_events','recent_events')['items'][-1]
|
|
C.typed_member(event,'payload','operation')['type'] = 'String'
|
|
C.typed_member(event,'payload','operation')['text'] = '"recovery"'
|
|
self.edit('pair/reader-checkpoint.json',corrupt)
|
|
self.reject('source-defined operation drift')
|
|
|
|
def test_operation_count_is_source_derived_with_recent_limit(self):
|
|
model = C.decode(json.loads((self.out/'pair/writer-formal-model.json').read_text()))
|
|
runtime = model['systems']['runtime']
|
|
runtime['id_generator']['counters']['domain_event'] = 900
|
|
stream = runtime['domain_events']
|
|
stream['published_ids'] = ['domain_event.%012d' % n for n in range(1,901)]
|
|
stream['recent_events'] = [{'ordinal':n} for n in range(645,901)]
|
|
expected = C.decode(C.expected_operation(F.typed(model),{'ok':True},'SAVE_CHECKPOINT_SUCCEEDED'))
|
|
new = expected['systems']['runtime']
|
|
self.assertEqual(901,new['id_generator']['counters']['domain_event'])
|
|
self.assertEqual(901,len(new['domain_events']['published_ids']))
|
|
self.assertEqual(256,len(new['domain_events']['recent_events']))
|
|
self.assertEqual(stream['recent_events'][1:],new['domain_events']['recent_events'][:-1])
|
|
self.assertEqual('domain_event.000000000901',new['domain_events']['recent_events'][-1]['event_id'])
|
|
del runtime['id_generator']['counters']
|
|
with self.assertRaisesRegex(ValueError,'missing/duplicate typed model member'):
|
|
C.expected_operation(F.typed(model),{'ok':True},'SAVE_CHECKPOINT_SUCCEEDED')
|
|
|
|
def test_missing_formal_member(self):
|
|
for name in ('formal','formal.bak'):
|
|
with self.subTest(package=name):
|
|
file = self.out/('pair/'+name+'/journal.log')
|
|
original = file.read_bytes()
|
|
file.unlink(); F.repin(self.out)
|
|
self.reject('journal.log')
|
|
file.write_bytes(original); F.repin(self.out)
|
|
self.edit('pair/formal-export.json', lambda d: d['slot_metadata'].update(sha256='0'*64))
|
|
self.reject('slot metadata SHA')
|
|
|
|
def test_missing_model_member(self):
|
|
path = 'pair/writer-unsupported-before.json'
|
|
model = C.decode(json.loads((self.out/path).read_text()))
|
|
del model['systems']['runtime']['inventory']
|
|
F.put(self.out/path, F.typed(model)); F.repin(self.out)
|
|
self.reject('missing whole Runtime members')
|
|
|
|
def test_missing_clamp_boundary(self):
|
|
self.edit('pair/writer-actions.json', lambda d: d['actions'][-1].update(steps=[1]*18))
|
|
self.reject('wheel clamp sequence')
|
|
|
|
def test_no_actual_pan(self):
|
|
self.edit('pair/writer-actions.json', lambda d: d['actions'][-1].update(pan_after=[0,0]))
|
|
self.reject('move-mode wheel did not pan')
|
|
|
|
def test_handler_command_must_remain_headless(self):
|
|
self.edit('runs/writer/writer-receipt.json', lambda d: d['command'].remove('--headless'))
|
|
self.reject('handler command contract')
|
|
|
|
def test_os_headless_masquerade(self):
|
|
self.edit('os-runs/writer/ended.json', lambda d: d['command'].insert(1, '--headless'))
|
|
self.reject('OS native argv')
|
|
|
|
def test_stale_os_recorder(self):
|
|
self.edit('os-runs/writer/ended.json', lambda d: d.update(recorder_sha256='0'*64))
|
|
self.reject('OS controller/recorder SHA')
|
|
|
|
def test_os_process_unended(self):
|
|
self.edit('os-runs/writer/ended.json', lambda d: d.update(engine_ended=False))
|
|
self.reject('OS native failure')
|
|
|
|
def test_os_default_player_changed(self):
|
|
self.edit('os-runs/writer/ended.json', lambda d: d.update(default_after={'changed':True}))
|
|
self.reject('OS protected state changed')
|
|
|
|
def test_os_actual_observation_missing(self):
|
|
self.edit('os-pair/unsupported-1/observed.json', lambda d: d.update(events=[]))
|
|
self.reject('missing OS events')
|
|
|
|
def test_os_ack_for_another_request(self):
|
|
self.edit('os-pair/unsupported-1/ack.json', lambda d: d.update(request_sha256='0'*64))
|
|
self.reject('OS ack request mismatch')
|
|
|
|
def test_os_canvas_mapping_drift(self):
|
|
self.edit('os-pair/unsupported-1/request.json', lambda d: d.update(viewport_to_screen=[1,0,0,1,32,0]))
|
|
# Repin acknowledgement too: test semantics, rather than checksum rejection.
|
|
self.edit('os-pair/unsupported-1/ack.json', lambda d: d.update(request_sha256=C.sha(self.out/'os-pair/unsupported-1/request.json')))
|
|
self.reject('OS canvas mapping drift')
|
|
|
|
def test_os_guide_interception(self):
|
|
self.edit('os-pair/unsupported-1/request.json', lambda d: d.update(guide_visible=True,guide_rect=[0,0,10000,10000]))
|
|
self.edit('os-pair/unsupported-1/ack.json', lambda d: d.update(request_sha256=C.sha(self.out/'os-pair/unsupported-1/request.json')))
|
|
self.reject('OS guide interception')
|
|
|
|
def test_whole_checker_missing_stair_projection(self):
|
|
self.edit('pair/writer-lower_stair-completed.json', lambda d: C.typed_member(d,'systems','runtime','facility_infrastructure','projections')['items'].pop())
|
|
self.reject('missing/extra infrastructure projection')
|
|
|
|
def test_whole_checker_wrong_stair_connector(self):
|
|
self.edit('pair/writer-lower_stair-completed.json', lambda d: C.typed_member(C.typed_member(d,'systems','runtime','path','connectors')['items'][0],'cost').update(text='999'))
|
|
self.reject('wrong infrastructure connector topology')
|
|
|
|
def test_refusal_cannot_be_relabelled(self):
|
|
self.edit('pair/writer-actions.json', lambda d: d['actions'][4].update(expected='OTHER_REASON'))
|
|
self.reject('required refusal reason changed')
|
|
|
|
def test_demolition_cannot_be_relabelled(self):
|
|
self.edit('pair/writer-actions.json', lambda d: d['actions'][9].update(kind='camera'))
|
|
self.reject('operation kind mismatch')
|
|
|
|
def test_png_signature_is_not_a_screenshot(self):
|
|
(self.out/'os-pair/screen.png').write_bytes(b'\x89PNG\r\n\x1a\nFAKE')
|
|
F.repin(self.out)
|
|
self.reject('invalid OS screenshot')
|
|
|
|
def test_omitted_os_lifecycle_request(self):
|
|
self.edit('os-pair/writer-requests.json', lambda d: d['requests'].pop())
|
|
self.reject('omitted OS lifecycle requests')
|
|
|
|
def test_os_audio_boolean_cannot_replace_owner_evidence(self):
|
|
self.edit('os-runs/writer/teardown/owned-audio/owner_1.json', lambda d: d['final_targets'][0].update(alive=True))
|
|
self.reject('real WeakRef release')
|
|
|
|
def collector_mock(self, fail_wait=False):
|
|
"""No native engine: Popen, profiles, live engine probes are all mocked."""
|
|
output = Path(self.temp.name)/'mock-collection'
|
|
stage = output/'stage'; stage.mkdir(parents=True)
|
|
shutil.copyfile(self.root/'project.godot',stage/'project.godot')
|
|
public = (stage/'project.godot').read_bytes()
|
|
(output/'pair').mkdir()
|
|
shutil.copytree(self.out/'pair/formal',output/'pair/initial-formal')
|
|
F.put(output/'pair/diagnostic-budget.json',{'setup':'synthetic mock'})
|
|
refs = C.runner.source_refs(self.root)
|
|
protected = {'mock_default_player':True}
|
|
captured = []; cleaned = []
|
|
def spawn(argv, **kw):
|
|
self.assertNotIn('--headless', argv)
|
|
self.assertEqual(['--path',str(stage),'--script'],argv[1:4])
|
|
role = kw['env']['P08_PLACEMENT_ROLE']
|
|
profile_text = (stage/'project.godot').read_text()
|
|
name = __import__('re').search(r'config/custom_user_dir_name="([^"]+)"',profile_text).group(1)
|
|
pid = 801+len(captured)
|
|
process = SimpleNamespace(pid=pid)
|
|
def wait(timeout):
|
|
self.assertEqual(900,timeout)
|
|
if fail_wait:
|
|
raise TimeoutError('mock Root acknowledgement timeout')
|
|
kw['stdout'].write('P08 passive OS placement: 2 checks, 0 failures\n')
|
|
F.put(output/('os-pair/'+role+'-profile.json'),{'pid':pid,'user_data_dir':str(Path(self.temp.name)/name),'display':'mock'})
|
|
if role == 'writer': shutil.copytree(output/'pair/initial-formal',output/'os-pair/formal')
|
|
return 0
|
|
process.wait = wait; captured.append(process)
|
|
return process
|
|
def cleanup(children):
|
|
cleaned.extend(p.pid for p in children)
|
|
return ([{'pid':p.pid,'engine_ended':True} for p in children], [])
|
|
r = SimpleNamespace(require=C.runner.require, inventory=C.runner.inventory,
|
|
sha=C.sha, read=C.load, write=C.runner.write, copy_source=shutil.copyfile,
|
|
source_refs=lambda root:refs, default_player=lambda root:protected,
|
|
profile_path=lambda name:Path(self.temp.name)/name,
|
|
private_project=C.runner.private_project, fresh_environment=C.runner.fresh_environment,
|
|
SUPPORT=ROOT/'tools/regression', module=lambda *a:SimpleNamespace(ordinary_gate=lambda *a:{'mock':True}))
|
|
outer = SimpleNamespace(native=SimpleNamespace(helper={'live_engines':lambda:[]}),finish_captured=cleanup)
|
|
# Disk admission is an environmental input to this fully mocked collector.
|
|
# Preserve the real collector's gate; make this fixture independent of host free space.
|
|
with patch.object(D.shutil,'disk_usage',return_value=SimpleNamespace(free=1024**3)), patch.object(D.subprocess,'Popen',side_effect=spawn), contextlib.redirect_stdout(io.StringIO()):
|
|
if fail_wait:
|
|
with self.assertRaisesRegex(AssertionError,'native OS collection failed'):
|
|
D.collect_os(r,self.root,output,stage,str(Path(self.temp.name)/'synthetic-engine-not-executable'),'1'*40,refs,protected,outer)
|
|
else:
|
|
phases = D.collect_os(r,self.root,output,stage,str(Path(self.temp.name)/'synthetic-engine-not-executable'),'1'*40,refs,protected,outer)
|
|
self.assertEqual(2,len(phases))
|
|
self.assertEqual([p.pid for p in captured],cleaned)
|
|
self.assertEqual(public,(stage/'project.godot').read_bytes())
|
|
self.assertTrue((output/'os-runs/writer/ended.json').exists())
|
|
|
|
def test_mock_native_controller_two_private_pids(self):
|
|
self.collector_mock()
|
|
|
|
def test_mock_native_timeout_restores_project_and_cleans_owned_child(self):
|
|
self.collector_mock(fail_wait=True)
|
|
|
|
class RealR05InfrastructureTests(unittest.TestCase):
|
|
"""Actual native snapshot, never a fabricated acceptance specimen."""
|
|
def setUp(self):
|
|
folder = ROOT/'tests/fixtures/p08-03'
|
|
provenance = json.loads((folder/'r05-reader-checkpoint-provenance.json').read_bytes())
|
|
raw = gzip.decompress((folder/'r05-reader-checkpoint.json.gz').read_bytes())
|
|
self.assertEqual(provenance['raw_sha256'],hashlib.sha256(raw).hexdigest())
|
|
self.assertEqual(provenance['raw_bytes'],len(raw))
|
|
self.node = json.loads(raw)
|
|
self.model = C.decode(self.node)
|
|
|
|
def test_real_reader_checkpoint(self):
|
|
self.assertEqual(self.model,C.state(self.node))
|
|
|
|
def reject(self, mutate, message):
|
|
runtime = self.model['systems']['runtime']; mutate(runtime)
|
|
with self.assertRaisesRegex(ValueError,message): C.state(F.typed(self.model))
|
|
|
|
def test_missing_projection(self):
|
|
self.reject(lambda r:r['facility_infrastructure']['projections'].pop(),'missing/extra infrastructure projection')
|
|
|
|
def test_extra_projection(self):
|
|
def mutate(r):
|
|
p=copy.deepcopy(r['facility_infrastructure']['projections'][0]); p['instance_id']='ghost'
|
|
r['facility_infrastructure']['projections'].append(p)
|
|
self.reject(mutate,'missing/extra infrastructure projection')
|
|
|
|
def test_wrong_projected_cells(self):
|
|
self.reject(lambda r:r['facility_infrastructure']['projections'][0].update(cells=[[9,9,0]]),'wrong infrastructure projection')
|
|
|
|
def test_missing_route(self):
|
|
self.reject(lambda r:r['spatial_subslots']['entities'].pop(),'wrong infrastructure ROUTE topology')
|
|
|
|
def test_extra_route(self):
|
|
self.reject(lambda r:r['spatial_subslots']['entities'].append({'entity_id':'infra.route:ghost','category':'ROUTE','cells':[[8,4,0]]}),'extra infrastructure ROUTE topology')
|
|
|
|
def test_wrong_route(self):
|
|
self.reject(lambda r:r['spatial_subslots']['entities'][0].update(category='UTILITY'),'wrong infrastructure ROUTE topology')
|
|
|
|
def test_missing_connector(self):
|
|
self.reject(lambda r:r['path']['connectors'].pop(0),'wrong infrastructure connector topology')
|
|
|
|
def test_extra_connector(self):
|
|
def mutate(r):
|
|
c=copy.deepcopy(r['path']['connectors'][0]); c['id']='infra.connector:ghost'; r['path']['connectors'].append(c)
|
|
self.reject(mutate,'extra infrastructure connector topology')
|
|
|
|
def test_wrong_connector(self):
|
|
self.reject(lambda r:r['path']['connectors'][0].update(to=[8,0,0]),'wrong infrastructure connector topology')
|
|
|
|
def test_wrong_projected_walkable(self):
|
|
self.reject(lambda r:r['route_projection']['projected_walkable_cells'].pop(),'wrong projected walkable topology')
|
|
|
|
def test_wrong_path_cells(self):
|
|
self.reject(lambda r:r['path']['cell_traffic'].pop(),'wrong projected walkable topology')
|
|
|
|
def test_ordinary_facility_loss(self):
|
|
self.reject(lambda r:r['facilities']['instances'].clear(),'partial demolition/completion')
|
|
|
|
def test_proxy_occupancy_loss(self):
|
|
self.reject(lambda r:r['world']['occupancies'].pop(),'orphan/lost occupancy')
|
|
|
|
def test_proxy_operational_duplication(self):
|
|
self.reject(lambda r:r['facilities']['instances'].append(copy.deepcopy(r['construction']['facilities'][0])),'partial demolition/completion')
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|