Digital Drosophila Archive
217 lines
13 KiB
Python
217 lines
13 KiB
Python
"""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()
|