"""Temporary-directory and small authority tests only; never touch real ROOT.""" import copy import hashlib import json from pathlib import Path import tempfile import types import unittest from unittest.mock import patch import engineering_prepare as p LOCAL = Path(__file__).resolve().parent AI = LOCAL.parent / 'pn-recruitment-audit-v1/publication.py' def valid_status(): final = dict(status=p.FINAL_STATUS, primary_trials_per_stage=120, repeats_per_stage=24, zero_overlay_exact=True, static25_repeat_exact=True, old_integrity_pass=True, frozen_five_reference_pass=True, frozen_eligibility_and_mask_unchanged=True, base_weights_unchanged=True, training=False, learned_candidate=False) scientific = dict(status=p.FINAL_STATUS, primary_trials_per_stage=120, technical_repeats_per_stage=24, zero_overlay_exact=True, static25_repeat_exact=True, training=False, learned_candidate=False) old = {'root-' + str(i): {'pass': True, 'checked': 1, 'mismatches': [], 'manifest_sha256': 'a'*64, 'checksums_sha256': 'b'*64} for i in range(28)} refs = {'pass_all': True, 'trials': [{'trial': name, 'exact': {key: True for key in ('step', 'neuron_idx', 'counts', 'v', 'g')}} for name in p.REFERENCES]} return final, scientific, old, copy.deepcopy(old), refs def planned(): return [dict(name='fixed-' + str(i), cohort='seed_validation' if i < 60 else 'novel_to_aj', repeat=r) for i in range(120) for r in ((0, 1) if i < 24 else (0,))] class AuthorityTests(unittest.TestCase): def test_required_final_status_and_all_28_roots_five_references(self): p.validate_final_status(*valid_status()) for mutation in ('status', 'hidden-learning', 'root', 'reference', 'root-hash'): args = copy.deepcopy(valid_status()) if mutation == 'status': args[0]['status'] = 'INCOMPLETE' elif mutation == 'hidden-learning': args[1]['learned_candidate'] = True elif mutation == 'root': args[3].pop('root-27') elif mutation == 'reference': args[4]['trials'][2]['exact']['g'] = False else: args[3]['root-1']['manifest_sha256'] = 'c' * 64 with self.subTest(mutation=mutation), self.assertRaises(ValueError): p.validate_final_status(*args) def test_first_actual_planned_primary_is_selected_without_dynamic_ranking(self): rows = planned() self.assertEqual(p.validate_planned(rows), rows[0]) for altered in (rows[:-1], rows + [rows[0]], [dict(rows[0], name='../escape'), *rows[1:]], [rows[1], rows[1], *rows[2:]]): with self.assertRaises(ValueError): p.validate_planned(altered) def test_three_external_sha_anchors_have_no_defaults_or_placeholders(self): for anchors in ((None, 'a'*64, 'b'*64), ('', '', ''), ('x'*64, 'a'*64, 'b'*64)): with self.assertRaisesRegex(ValueError, 'actual external'): p._prepare(Path('not-read'), Path('not-read'), Path('not-read'), *anchors) def test_path_inventory_rejects_duplicates_and_traversal(self): for name in ('../bad', '/absolute', 'a\\b', 'a/../b', 'C:/bad'): with self.assertRaises(ValueError): p._relative(name) row = {'path': 'good.json', 'bytes': 3, 'sha256': 'a'*64} with self.assertRaisesRegex(ValueError, 'Duplicate'): p._index([row, row]) def test_pinned_AI_verifier_rechecks_complete_synthetic_seal(self): helper = p.load_module('phase4an_test_AI_publication', AI, p.AI_HELPER_SHA) with tempfile.TemporaryDirectory() as directory: root = Path(directory).resolve() payload = root / 'small.json'; payload.write_text('{}', encoding='utf-8') row = {'path': payload.name, 'bytes': payload.stat().st_size, 'sha256': p.digest(payload)} p.save(root / 'manifest.json', {'root': str(root), 'files': [row]}) (root / 'checksums.sha256').write_text(row['sha256'] + ' small.json\n' + p.digest(root / 'manifest.json') + ' manifest.json\n', encoding='utf-8') self.assertTrue(helper.verify_seal(root)['pass_all']) payload.write_text('{"changed":true}', encoding='utf-8') with self.assertRaises(ValueError): helper.verify_seal(root) def test_finalizer_authority_helper_checks_external_anchors_and_both_full_seals(self): helper = p.load_module('phase4an_test_AI_publication', AI, p.AI_HELPER_SHA) with tempfile.TemporaryDirectory() as directory: root = Path(directory).resolve() model, experiment = root / p.AM_MODEL_REL, root / p.AM_EXP_REL model.mkdir(parents=True); experiment.mkdir(parents=True) p.save(model / 'payload.json', {'synthetic': True}) p.save(experiment / 'protocol/preregistered.json', {'planned': []}) p.save(experiment / 'audit/final-status.json', {'status': p.FINAL_STATUS}) for tree in (model, experiment): rows = [dict(path=path.relative_to(tree).as_posix(), bytes=path.stat().st_size, sha256=p.digest(path)) for path in sorted(tree.rglob('*')) if path.is_file()] p.save(tree / 'manifest.json', {'root': str(tree), 'files': rows}) (tree / 'checksums.sha256').write_text(''.join(row['sha256'] + ' ' + row['path'] + '\n' for row in rows) + p.digest(tree / 'manifest.json') + ' manifest.json\n', encoding='utf-8') prereg = p.digest(experiment / 'protocol/preregistered.json') anchors = dict(am_model_manifest_sha256=p.digest(model / 'manifest.json'), am_experiment_manifest_sha256=p.digest(experiment / 'manifest.json'), am_final_sha256=p.digest(experiment / 'audit/final-status.json'), am_preregistration_sha256=prereg, AI_publication_helper_sha256=p.AI_HELPER_SHA) with patch.object(p, 'ROOT', root), patch.object(p, 'AM_PREREG_SHA', prereg), patch.object(p, 'load_module', return_value=helper): self.assertEqual(set(p.verify_authority(anchors)), {'model', 'experiment'}) (model / 'payload.json').write_text('changed', encoding='utf-8') with self.assertRaises(ValueError): p.verify_authority(anchors) def test_old_integrity_bridge_restores_import_path_and_rejects_failed_root(self): import sys with tempfile.TemporaryDirectory() as directory: root = Path(directory).resolve(); expected = valid_status()[3] before_path = list(sys.path) def good(): sys.path.insert(0, 'synthetic-old-context-search-path') return copy.deepcopy(expected) module = types.SimpleNamespace(ROOT=root, integrity=good) with patch.object(p, 'load_module', return_value=module): self.assertEqual(p._old_integrity(root, 'a'*64, expected), expected) self.assertEqual(sys.path, before_path) bad = copy.deepcopy(expected); bad['root-1']['pass'] = False module.integrity = lambda: bad with patch.object(p, 'load_module', return_value=module), self.assertRaisesRegex(ValueError, '28-root'): p._old_integrity(root, 'a'*64, expected) self.assertEqual(sys.path, before_path) def test_all_144_stage_completions_must_match_sealed_inventory(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory).resolve(); experiment = root / p.AM_EXP_REL; experiment.mkdir(parents=True) rows, index, completed = planned(), {}, [] def add(relative, value): path = experiment / relative; p.save(path, value) index[relative] = {'path': relative, 'bytes': path.stat().st_size, 'sha256': p.digest(path)} for row in rows: prefix = 'gpu/zero/' + row['name'] + '-r' + str(row['repeat']) + '/' folder = experiment / prefix.rstrip('/') case = root / p.AK_REL / 'cases' / row['cohort'] / row['name'] add(prefix + 'summary.json', {'case': row['name'], 'complete': True, 'pass_all': True, 'all_finite': True, 'steps': 1000, 'completed_ticks': 1000}) add(prefix + 'comparison.json', {'pass_all': True, 'overlay_name': 'zero', 'case_dir': str(case), 'right': str(folder), 'mode': 'zero', 'left': str(root / p.AK_REL / 'gpu' / row['cohort'] / (row['name'] + '-r' + str(row['repeat']))), 'checks': {'synthetic_gate': True}, 'failed_checks': [], 'full_queue_reconstructed': True}) inventory = [dict(path=k[len(prefix):], bytes=v['bytes'], sha256=v['sha256']) for k, v in sorted(index.items()) if k.startswith(prefix)] add(prefix + 'output-hashes.json', inventory) completed.append(dict(name=row['name'], repeat=row['repeat'], output_hashes_sha256=index[prefix+'output-hashes.json']['sha256'], comparison_sha256=index[prefix+'comparison.json']['sha256'])) receipt = {'stage': 'zero', 'pass_all': True, 'trials': 144, 'primary_trials': 120, 'technical_repeats': 24, 'training': False, 'completed': completed} add('audit/zero-complete.json', receipt) self.assertTrue(p.validate_stage(experiment, index, 'zero', rows, root)['pass_all']) wrong = copy.deepcopy(rows); wrong[0], wrong[1] = wrong[1], wrong[0] with self.assertRaisesRegex(ValueError, 'order/identity'): p.validate_stage(experiment, index, 'zero', wrong, root) # A missing sealed payload entry cannot be hidden by a PASS flag. altered = dict(index); altered.pop('gpu/zero/fixed-0-r0/summary.json') with self.assertRaises(ValueError): p.validate_stage(experiment, altered, 'zero', rows, root) class PreparationSafetyTests(unittest.TestCase): def test_exclusive_layout_allows_only_nonexistent_or_empty_experiment(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory).resolve() model = root / 'models/male-cns-lif/olfactory-conditioning-lab-v1'; model.mkdir(parents=True) experiment = root / 'experiments/phase4an-engineering-validation'; experiment.parent.mkdir() self.assertEqual(p._layout(root, model, experiment)[2], experiment) experiment.mkdir(); self.assertEqual(p._layout(root, model, experiment)[2], experiment) (experiment / 'preserve.txt').write_text('partial', encoding='utf-8') with self.assertRaisesRegex(ValueError, 'populated'): p._layout(root, model, experiment) self.assertEqual((experiment / 'preserve.txt').read_text(), 'partial') def test_required_runtime_workflow_scope_sources_must_all_exist(self): with tempfile.TemporaryDirectory() as directory: model = Path(directory).resolve() for name in p.SOURCE_FILES: (model / name).write_text('# pure test source\n', encoding='utf-8') result = p.source_inventory(model) self.assertEqual(len(result), 21) self.assertIn(str(model / 'engineering_workflow.py'), result) self.assertIn(str(model / 'engineering_scopes.py'), result) self.assertIn(str(model / 'engineering_analysis.py'), result) (model / 'engineering_workflow.py').unlink() with self.assertRaisesRegex(ValueError, 'Required engineering source'): p.source_inventory(model) def test_byte_budget_reserve_is_not_a_license_to_delete_partials(self): with tempfile.TemporaryDirectory() as directory: experiment = Path(directory).resolve() / 'new' experiment.mkdir(); (experiment / 'partial').write_bytes(b'x') with patch.object(p.shutil, 'disk_usage', return_value=types.SimpleNamespace(free=30 * 1024**3)): self.assertEqual(p._budget(experiment)['used_bytes'], 1) with patch.object(p.shutil, 'disk_usage', return_value=types.SimpleNamespace(free=5 * 1024**3)): with self.assertRaisesRegex(ValueError, 'Insufficient'): p._budget(experiment) self.assertTrue((experiment / 'partial').is_file()) def test_row_mapping_rejects_unplanned_or_escaping_paths(self): with tempfile.TemporaryDirectory() as directory: exp = Path(directory).resolve(); case = exp / 'cases/safe-row'; case.mkdir(parents=True) reg = {'plan': {'rows': [{'row_id': 'safe-row'}]}, 'cases': {'safe-row': {'case_dir': str(case)}}} with patch.object(p, 'EXP', exp): self.assertEqual(p.row_path(reg, 'safe-row'), case) with self.assertRaises(ValueError): p.row_path(reg, '../outside') reg['cases']['safe-row']['case_dir'] = str(exp) with self.assertRaises(ValueError): p.row_path(reg, 'safe-row') if __name__ == '__main__': unittest.main()