274 lines
13 KiB
Python
274 lines
13 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
|
|
|
|
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 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_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_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)
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|