Digital Drosophila Archive
229 lines
15 KiB
Python
229 lines
15 KiB
Python
"""Temporary synthetic payloads only; no real publishing, GPU or old writes."""
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
import publish_phase4an as p
|
|
|
|
|
|
AI = Path(__file__).resolve().parent.parent / 'pn-recruitment-audit-v1/publication.py'
|
|
if not AI.is_file():
|
|
AI = p.ROOT / p.HELPER_REL
|
|
|
|
|
|
def write(path, value):
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(value, sort_keys=True), encoding='utf-8')
|
|
|
|
|
|
def read(path):
|
|
return json.loads(path.read_text(encoding='utf-8'))
|
|
|
|
|
|
def seal(exp):
|
|
rows = [dict(path=path.relative_to(exp).as_posix(), bytes=path.stat().st_size, sha256=p.digest(path))
|
|
for path in sorted(exp.rglob('*')) if path.is_file() and path.relative_to(exp).as_posix() not in ('manifest.json', 'checksums.sha256')]
|
|
write(exp / 'manifest.json', {'root': str(exp), 'files': rows})
|
|
(exp / 'checksums.sha256').write_text(''.join(r['sha256'] + ' ' + r['path'] + '\n' for r in rows) +
|
|
p.digest(exp / 'manifest.json') + ' manifest.json\n', encoding='utf-8')
|
|
return p.digest(exp / 'manifest.json'), p.digest(exp / 'audit/final-status.json')
|
|
|
|
|
|
def fixture(root, status=p.PASS):
|
|
exp, model = root / p.EXP_REL, root / p.MODEL_REL
|
|
exp.mkdir(parents=True); model.mkdir(parents=True)
|
|
helper = root / p.HELPER_REL; helper.parent.mkdir(parents=True); helper.write_bytes(AI.read_bytes())
|
|
sources, inputs = {}, {}
|
|
for name in p.SOURCE_FILES:
|
|
raw = ('# synthetic used source: ' + name + '\n').encode()
|
|
original, snapshot = model / name, exp / 'protocol/source-snapshot' / name
|
|
original.write_bytes(raw); snapshot.parent.mkdir(parents=True, exist_ok=True); snapshot.write_bytes(raw)
|
|
sources[str(original)] = inputs[str(snapshot)] = hashlib.sha256(raw).hexdigest()
|
|
(model / 'unregistered-science.py').write_bytes(b'not copied or sealed')
|
|
planned = [dict(row_id='synthetic-' + str(i), steps=348 if i == 31 else 346) for i in range(32)]
|
|
anchors = dict(am_model_manifest_sha256='a'*64, am_experiment_manifest_sha256='b'*64)
|
|
reg = dict(schema='phase4an-engineering-registration-v1', training=False, source_sha256=sources, input_sha256=inputs,
|
|
anchors=anchors, plan=dict(total_calls=32, total_ticks=11074, rows=planned))
|
|
write(exp / 'protocol/preregistered.json', reg); prereg = p.digest(exp / 'protocol/preregistered.json')
|
|
completed = []
|
|
for row in planned:
|
|
identity = row['row_id']
|
|
checkpoint = exp / 'runs' / identity / 'checkpoint-manifest.json'; write(checkpoint, {'synthetic': True})
|
|
runtime = dict(complete=True, training=False, registration_sha256=prereg, row=row, checkpoint_manifest_sha256=p.digest(checkpoint))
|
|
runtime_path = exp / 'receipts' / (identity + '.runtime.json'); write(runtime_path, runtime)
|
|
for suffix in ('.stdout.log', '.stderr.log'):
|
|
target = exp / 'logs' / (identity + suffix); target.parent.mkdir(exist_ok=True); target.write_bytes(b'synthetic log')
|
|
receipt = dict(complete=True, process_exit_code=0, row_id=identity, registration_sha256=prereg,
|
|
runtime_record_sha256=p.digest(runtime_path), checkpoint_manifest_sha256=p.digest(checkpoint),
|
|
stdout_sha256=p.digest(exp / 'logs' / (identity + '.stdout.log')),
|
|
stderr_sha256=p.digest(exp / 'logs' / (identity + '.stderr.log')))
|
|
receipt_path = exp / 'receipts' / (identity + '.json'); write(receipt_path, receipt)
|
|
completed.append(dict(row_id=identity, execution_sha256=p.digest(receipt_path)))
|
|
write(exp / 'audit/execution-complete.json', dict(complete=True, registration_sha256=prereg, completed=completed,
|
|
expected_calls=32, executed_ticks=11074, training=False, learning_pass_claimed=False))
|
|
numerical, covered = status != 'FAIL-NUMERICAL', status == p.PASS
|
|
analysis = dict(schema='phase4an-engineering-analysis-v1', status=status, registration_sha256=prereg,
|
|
source_sha256=sources, training=False, learning_pass_claimed=False, candidate_generated=False, GPU_used=False,
|
|
no_new_brain_runs_during_analysis=True, expected_calls=32, analyzed_calls=32, executed_ticks=11074,
|
|
pass_all=numerical and covered, numerical_exact=numerical, nonvacuous_temporal_coverage=covered)
|
|
write(exp / 'audit/engineering-analysis.json', analysis)
|
|
analysis_sha = p.digest(exp / 'audit/engineering-analysis.json')
|
|
write(exp / 'audit/analysis-execution.json', dict(registration_sha256=prereg, analysis_sha256=analysis_sha))
|
|
final = dict(schema='phase4an-engineering-final-v1', status=status, registration_sha256=prereg,
|
|
analysis_sha256=analysis_sha, analysis_recomputed_exactly=True, completed_calls=32, executed_ticks=11074,
|
|
training=False, learned_candidate=False, learning_pass_claimed=False, model_directory_sealed=False,
|
|
used_engineering_source_snapshots_sealed=True, old_integrity_pass=True, old_integrity_roots=28, AM_seals_preserved=True)
|
|
write(exp / 'audit/final-status.json', final)
|
|
old = {str(i): {'pass': True} for i in range(28)}
|
|
write(exp / 'regression/old-integrity-before.json', old); write(exp / 'regression/old-integrity-after.json', old)
|
|
write(exp / 'regression/AM-seals-after.json', {key: {'pass_all': True, 'manifest_sha256': anchors['am_' + key + '_manifest_sha256']}
|
|
for key in ('model', 'experiment')})
|
|
(exp / 'runs/synthetic-0/queue.jsonl').write_bytes(b'bulk queue\n')
|
|
(exp / 'runs/synthetic-0/state.npz').write_bytes(b'synthetic binary')
|
|
(exp / 'extension-build').mkdir(); (exp / 'extension-build/build.log').write_bytes(b'not selected even log')
|
|
return prereg, seal(exp)
|
|
|
|
|
|
class DeliveryTests(unittest.TestCase):
|
|
def test_success_uses_real_verifier_copies_only_registered_sources_and_small_evidence(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory).resolve(); prereg, anchors = fixture(root)
|
|
with patch.object(p, 'PREREG_SHA', prereg):
|
|
result = p.publish(*anchors, root=root)
|
|
registry = root / p.REGISTRY_REL; receipt = read(registry / 'publication-receipt.json')
|
|
self.assertTrue(result['publication_complete']); self.assertEqual(result['declared_status'], p.PASS)
|
|
self.assertEqual(result['receipt_sha256'], p.digest(registry / 'publication-receipt.json'))
|
|
self.assertEqual(len(list((registry / 'source-snapshot').iterdir())), 21)
|
|
self.assertFalse((registry / 'model').exists())
|
|
self.assertFalse(any(x.name == 'unregistered-science.py' for x in registry.rglob('*')))
|
|
self.assertFalse((registry / 'evidence/runs/synthetic-0/queue.jsonl').exists())
|
|
self.assertFalse((registry / 'evidence/extension-build/build.log').exists())
|
|
self.assertEqual(len(receipt['experiment_payload_not_copied']), 3)
|
|
self.assertNotIn('publication-receipt.json', {x['path'] for x in receipt['files']})
|
|
self.assertTrue(receipt['complete_checksums_reference_bulk_not_copied'])
|
|
self.assertFalse(receipt['numerical_gate_rerun_by_publisher'])
|
|
for row in receipt['files']:
|
|
self.assertEqual(p.digest(registry / row['path']), row['sha256'])
|
|
self.assertEqual(p.digest(root / p.EXP_REL / 'manifest.json'), anchors[0])
|
|
|
|
def test_failed_and_incomplete_status_are_delivered_without_promotion(self):
|
|
for status in ('FAIL-NUMERICAL', 'INCOMPLETE-COVERAGE'):
|
|
with self.subTest(status=status), tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory).resolve(); prereg, anchors = fixture(root, status)
|
|
with patch.object(p, 'PREREG_SHA', prereg):
|
|
result = p.publish(*anchors, root=root)
|
|
receipt = read(root / p.REGISTRY_REL / 'publication-receipt.json')
|
|
self.assertEqual(result['declared_status'], status)
|
|
self.assertFalse(receipt['declared_engineering_pass'])
|
|
self.assertFalse(receipt['learning_pass_claimed'])
|
|
|
|
def test_two_external_hashes_are_mandatory_and_fixed_prereg_is_not_inferred(self):
|
|
with self.assertRaises(TypeError): p.publish()
|
|
for anchors in ((None, 'a'*64), ('x'*64, 'b'*64)):
|
|
with self.assertRaises(ValueError): p.publish(*anchors)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory).resolve(); _, anchors = fixture(root)
|
|
with self.assertRaisesRegex(ValueError, 'authority mismatch'): p.publish(*anchors, root=root)
|
|
self.assertFalse((root / p.REGISTRY_REL).exists())
|
|
|
|
def test_wrong_external_manifest_or_final_is_rejected_before_writes(self):
|
|
for position in (0, 1):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory).resolve(); prereg, anchors = fixture(root); bad = list(anchors); bad[position] = 'c'*64
|
|
with patch.object(p, 'PREREG_SHA', prereg), self.assertRaises(ValueError): p.publish(*bad, root=root)
|
|
self.assertFalse((root / p.REGISTRY_REL).exists())
|
|
|
|
def test_source_mutation_and_missing_source_are_rejected(self):
|
|
for mutation in ('changed', 'missing'):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory).resolve(); prereg, anchors = fixture(root)
|
|
source = root / p.MODEL_REL / 'runtime_light.py'
|
|
if mutation == 'changed': source.write_bytes(b'changed')
|
|
else: source.unlink()
|
|
with patch.object(p, 'PREREG_SHA', prereg), self.assertRaises(ValueError): p.publish(*anchors, root=root)
|
|
self.assertFalse((root / p.REGISTRY_REL).exists())
|
|
|
|
def test_snapshot_mutation_rejected_even_if_new_experiment_seal_is_valid(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory).resolve(); prereg, _ = fixture(root); exp = root / p.EXP_REL
|
|
(exp / 'protocol/source-snapshot/runtime_light.py').write_bytes(b'different')
|
|
anchors = seal(exp)
|
|
with patch.object(p, 'PREREG_SHA', prereg), self.assertRaisesRegex(ValueError, 'snapshot mismatch'):
|
|
p.publish(*anchors, root=root)
|
|
|
|
def test_partial_or_unlisted_seal_payload_is_not_ignored(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory).resolve(); prereg, anchors = fixture(root)
|
|
(root / p.EXP_REL / 'unlisted').write_bytes(b'unlisted')
|
|
with patch.object(p, 'PREREG_SHA', prereg), self.assertRaises(ValueError): p.publish(*anchors, root=root)
|
|
|
|
def test_helper_pin_and_integrity_authority_are_checked(self):
|
|
for mutation in ('helper', 'old-authority'):
|
|
with self.subTest(mutation=mutation), tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory).resolve(); prereg, anchors = fixture(root); exp = root / p.EXP_REL
|
|
if mutation == 'helper':
|
|
(root / p.HELPER_REL).write_bytes(b'# not the pinned verifier')
|
|
else:
|
|
for point, value in (('before', 'a'), ('after', 'b')):
|
|
path = exp / ('regression/old-integrity-' + point + '.json')
|
|
record = read(path); record['0']['manifest_sha256'] = value * 64; write(path, record)
|
|
anchors = seal(exp)
|
|
with patch.object(p, 'PREREG_SHA', prereg), self.assertRaises(ValueError): p.publish(*anchors, root=root)
|
|
self.assertFalse((root / p.REGISTRY_REL).exists())
|
|
|
|
def test_final_false_success_learning_and_bad_count_rejected(self):
|
|
for key, value in (('status', 'FAIL-NUMERICAL'), ('learned_candidate', True), ('completed_calls', 31), ('completed_calls', True),
|
|
('analysis_recomputed_exactly', False), ('model_directory_sealed', True)):
|
|
with self.subTest(key=key, value=value), tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory).resolve(); prereg, _ = fixture(root); exp = root / p.EXP_REL
|
|
final = read(exp / 'audit/final-status.json'); final[key] = value; write(exp / 'audit/final-status.json', final)
|
|
with patch.object(p, 'PREREG_SHA', prereg), self.assertRaises(ValueError): p.publish(*seal(exp), root=root)
|
|
|
|
def test_changed_execution_order_or_checkpoint_link_rejected(self):
|
|
for mutation in ('order', 'checkpoint'):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory).resolve(); prereg, _ = fixture(root); exp = root / p.EXP_REL
|
|
if mutation == 'order':
|
|
name = exp / 'audit/execution-complete.json'; data = read(name); data['completed'].reverse(); write(name, data)
|
|
else:
|
|
name = exp / 'runs/synthetic-0/checkpoint-manifest.json'; write(name, {'synthetic': False})
|
|
with patch.object(p, 'PREREG_SHA', prereg), self.assertRaises(ValueError): p.publish(*seal(exp), root=root)
|
|
|
|
def test_policy_limits_also_exclude_small_build_json_and_all_jsonl(self):
|
|
for name in ('extension-build/info.json', 'x/queue.jsonl', 'x/state.npz', 'x/trace.npy', 'x/random.py'):
|
|
self.assertIsNone(p.selection(dict(path=name, bytes=1))[0])
|
|
self.assertIsNone(p.selection(dict(path='large.json', bytes=p.MAX_BYTES + 1))[0])
|
|
self.assertEqual(p.selection(dict(path='protocol/source-snapshot/runtime_light.py', bytes=1))[0], 'source-snapshot/runtime_light.py')
|
|
|
|
def test_existing_registry_is_never_overwritten_or_retried(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory).resolve(); prereg, anchors = fixture(root)
|
|
destination = root / p.REGISTRY_REL; destination.mkdir(parents=True); (destination / 'partial').write_bytes(b'preserve')
|
|
with patch.object(p, 'PREREG_SHA', prereg), self.assertRaisesRegex(ValueError, 'no overwrite'):
|
|
p.publish(*anchors, root=root)
|
|
self.assertEqual((destination / 'partial').read_bytes(), b'preserve')
|
|
|
|
def test_copy_failure_preserves_partial_and_disallows_retry(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory).resolve(); prereg, anchors = fixture(root)
|
|
helper = p.load_helper(root / p.HELPER_REL); original = helper._copy_verified; calls = []
|
|
def fail_second(*args):
|
|
calls.append(args)
|
|
if len(calls) == 2: raise OSError('synthetic write failure')
|
|
return original(*args)
|
|
with patch.object(p, 'PREREG_SHA', prereg), patch.object(p, 'load_helper', return_value=helper), patch.object(helper, '_copy_verified', side_effect=fail_second):
|
|
with self.assertRaises(OSError): p.publish(*anchors, root=root)
|
|
registry = root / p.REGISTRY_REL
|
|
self.assertTrue((registry / 'publication-failure.json').is_file())
|
|
self.assertTrue(calls[0][1].is_file())
|
|
self.assertFalse((registry / 'publication-receipt.json').exists())
|
|
with patch.object(p, 'PREREG_SHA', prereg), self.assertRaisesRegex(ValueError, 'no overwrite'):
|
|
p.publish(*anchors, root=root)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|