Digital Drosophila Archive
122 lines
7.5 KiB
Python
122 lines
7.5 KiB
Python
"""Synthetic-only terminal guard tests, no science, training or final metrics."""
|
|
import copy
|
|
import json
|
|
import shlex
|
|
import sys
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
import normal_finalize as F
|
|
|
|
class FinalizeTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp=tempfile.TemporaryDirectory();self.addCleanup(self.temp.cleanup)
|
|
self.root=Path(self.temp.name).resolve();self.exp=self.root/'exp';self.control=self.root/'control';self.model=self.root/'model'
|
|
for p in (self.exp/'protocol',self.exp/'audit',self.control,self.model):p.mkdir(parents=True)
|
|
self.reg=dict(schema='normal-conditioning-registration-v1',scientific_scope='normal-acquisition-only',
|
|
subsequent_assays_authorized=False,normal_inventory={'fixture':True})
|
|
F.save(self.exp/'protocol/preregistered.json',self.reg)
|
|
self.sha=F.identity(self.exp/'protocol/preregistered.json')['sha256']
|
|
for name in ('launch-attempt.json','launch.json','supervisor-start.json'):
|
|
F.save(self.control/name,dict(synthetic=True,registration_sha256=self.sha))
|
|
self.pins={p.name:F.identity(p) for p in self.control.iterdir()}
|
|
self.kwargs=dict(exp=self.exp,control=self.control,model=self.model,regsha=self.sha,start_pins=self.pins,
|
|
census=lambda:dict(actual_OS_query=True,matching_live_processes=[]))
|
|
|
|
def terminal(self):
|
|
F.save(self.control/'exit.json',dict(registration_sha256=self.sha,
|
|
argv=[str(F.PYTHON),'-B',str(self.model/'normal_runner.py'),'--registration-sha256',self.sha],
|
|
exit_code=0,final_exit_observed=True,complete=True,error=None,
|
|
learning_success_claimed=False,scientific_analysis_required=True))
|
|
F.save(self.exp/'audit/execution-complete.json',dict(complete=True,registration_sha256=self.sha,
|
|
inventory=self.reg['normal_inventory'],learning_success_claimed=False))
|
|
|
|
def rewrite_fixture(self,path,update):
|
|
value=F.read(path);value.update(update)
|
|
# Test-owned file corruption, not application editing/retry behavior.
|
|
path.write_text(json.dumps(value),encoding='utf-8')
|
|
|
|
def test_actual_live_process_refuses_even_with_complete_files_and_no_writes(self):
|
|
self.terminal();before=sorted(str(p) for p in self.root.rglob('*'))
|
|
kwargs={**self.kwargs,'census':lambda:dict(actual_OS_query=True,matching_live_processes=[{'pid':123}])}
|
|
with self.assertRaisesRegex(ValueError,'still live'):F.readiness(**kwargs)
|
|
self.assertEqual(before,sorted(str(p) for p in self.root.rglob('*')))
|
|
|
|
def test_missing_process_without_real_exit_does_not_authorize_terminal_audit(self):
|
|
with self.assertRaisesRegex(ValueError,'missing process alone'):F.readiness(**self.kwargs)
|
|
|
|
def test_success_only_means_ready_not_learning_and_does_not_hash_brain_outputs(self):
|
|
self.terminal();F.save(self.exp/'unrelated-heavy-placeholder.json',{'not_read':True})
|
|
checked=[];original=F.identity
|
|
def identity(path):checked.append(str(path));return original(path)
|
|
with patch.object(F,'identity',side_effect=identity):result=F.readiness(**self.kwargs)
|
|
self.assertTrue(result['ready']);self.assertFalse(result['scientific_success_inferred'])
|
|
self.assertNotIn(str(self.exp/'unrelated-heavy-placeholder.json'),checked)
|
|
|
|
def test_changed_startup_binding_is_rejected(self):
|
|
self.terminal();self.rewrite_fixture(self.control/'launch.json',{'tampered':True})
|
|
with self.assertRaisesRegex(ValueError,'Startup'):F.readiness(**self.kwargs)
|
|
|
|
def test_nonzero_parent_exit_is_not_scientific_failure_or_success(self):
|
|
self.terminal();self.rewrite_fixture(self.control/'exit.json',{'exit_code':1,'complete':False})
|
|
with self.assertRaisesRegex(ValueError,'successful run exit'):F.readiness(**self.kwargs)
|
|
|
|
def test_completion_inventory_or_registration_mismatch_rejected(self):
|
|
self.terminal();self.rewrite_fixture(self.exp/'audit/execution-complete.json',{'inventory':{'shortened':True}})
|
|
with self.assertRaisesRegex(ValueError,'completion binding'):F.readiness(**self.kwargs)
|
|
|
|
def test_stopped_record_always_refuses_completion(self):
|
|
self.terminal();F.save(self.exp/'audit/stopped.json',dict(complete=False))
|
|
with self.assertRaisesRegex(ValueError,'Stopped run'):F.readiness(**self.kwargs)
|
|
|
|
def test_actual_process_query_required_not_caller_assumption(self):
|
|
self.terminal();kw={**self.kwargs,'census':lambda:dict(matching_live_processes=[])}
|
|
with self.assertRaisesRegex(ValueError,'Actual process'):F.readiness(**kw)
|
|
|
|
def test_argv_classifier_ignores_grep_substrings_and_includes_real_children(self):
|
|
worker=self.model.parent/'conditioning-interface-v2/isolated_workers.py'
|
|
validator=self.model.parent/'conditioning-execution-v1/validate_update_cpu.py'
|
|
def row(pid,*argv):return str(pid)+' '+shlex.join(argv)
|
|
rows='\n'.join([
|
|
row(10,'python','-B',str(self.model/'normal_runner.py'),'--registration-sha256','x'),
|
|
row(11,'python','-B',str(worker),'brain','--job',str(self.exp/'jobs/x.json')),
|
|
row(12,'python','-B',str(validator),'--authority',str(self.exp/'jobs/x.json')),
|
|
row(13,'bash','-c','grep '+str(self.model/'normal_runner.py')),
|
|
row(14,'python','-B',str(worker),'brain','--job','/different/exp/jobs/x.json'),
|
|
row(15,'python','-B',str(self.model/'normal_inputs.py'),'--registration-sha256','x')])
|
|
self.assertEqual([r['pid'] for r in F.live_runtime_rows(rows,self.model,self.exp)],[10,11,12,15])
|
|
|
|
def test_exclusive_save_never_overwrites(self):
|
|
path=self.root/'owned.json';F.save(path,{'first':True});before=path.read_bytes()
|
|
with self.assertRaises(FileExistsError):F.save(path,{'second':True})
|
|
self.assertEqual(before,path.read_bytes())
|
|
|
|
def test_real_harmless_CPU_child_exit_and_log_hashes_are_saved(self):
|
|
out=self.root/'new-audit';out.mkdir()
|
|
with patch.object(F,'OUT',out):
|
|
receipt=F.child('fixture',[sys.executable,'-B','-c','print("synthetic CPU fixture only")'],10)
|
|
self.assertEqual(receipt['exit_code'],0);self.assertTrue(receipt['actual_parent_observation'])
|
|
self.assertEqual(receipt['stdout'],F.identity(out/'fixture.stdout.log'))
|
|
self.assertIn('synthetic CPU fixture only',(out/'fixture.stdout.log').read_text())
|
|
|
|
def test_failed_CPU_child_preserves_exit_and_refuses_same_name_retry(self):
|
|
out=self.root/'failed-audit';out.mkdir()
|
|
with patch.object(F,'OUT',out):
|
|
with self.assertRaisesRegex(ValueError,'Audit child failed'):
|
|
F.child('fixture',[sys.executable,'-B','-c','raise SystemExit(7)'],10)
|
|
receipt=F.read(out/'fixture-exit.json')
|
|
self.assertEqual(receipt['exit_code'],7);self.assertFalse(receipt['complete'])
|
|
with self.assertRaises(FileExistsError):F.child('fixture',[sys.executable,'-B','-c','pass'],10)
|
|
|
|
def test_finalization_checks_live_guard_before_source_or_output_changes(self):
|
|
out=self.root/'should-stay-empty';out.mkdir()
|
|
with patch.object(F,'PYTHON',Path(sys.executable)),patch.object(F,'OUT',out),\
|
|
patch.object(F,'readiness',side_effect=ValueError('still live')),\
|
|
patch.object(F,'source_identity') as source:
|
|
with self.assertRaisesRegex(ValueError,'still live'):F.finalize()
|
|
source.assert_not_called()
|
|
self.assertEqual(list(out.iterdir()),[])
|
|
|
|
if __name__=='__main__':unittest.main()
|