"""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()