Digital Drosophila Archive
275 lines
13 KiB
Python
275 lines
13 KiB
Python
"""Temporary-directory, CPU-only publication mutations; no real experiment."""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
import publish_phase4ao as P
|
|
|
|
|
|
_KNOWN_HELPERS = (P.ROOT / P.HELPER_REL,
|
|
Path(__file__).resolve().parents[1] / 'pn-recruitment-audit-v1/publication.py')
|
|
HELPER = next((p for p in _KNOWN_HELPERS if p.is_file()), None)
|
|
if HELPER is None or P.digest(HELPER) != P.HELPER_SHA:
|
|
raise ValueError('Pinned publication helper absent or changed in server/local source layout')
|
|
|
|
|
|
def put(path, value):
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
data = value if isinstance(value, bytes) else (json.dumps(value, sort_keys=True) + '\n').encode()
|
|
path.write_bytes(data)
|
|
return hashlib.sha256(data).hexdigest()
|
|
|
|
|
|
def read(path):
|
|
return json.loads(path.read_text(encoding='utf-8'))
|
|
|
|
|
|
def seal(exp):
|
|
rows = [dict(path=p.relative_to(exp).as_posix(), bytes=p.stat().st_size, sha256=P.digest(p))
|
|
for p in sorted(exp.rglob('*')) if p.is_file() and p.relative_to(exp).as_posix()
|
|
not in ('manifest.json', 'checksums.sha256')]
|
|
put(exp / 'manifest.json', dict(root=str(exp), files=rows))
|
|
data = ''.join(r['sha256'] + ' ' + r['path'] + '\n' for r in rows)
|
|
data += P.digest(exp / 'manifest.json') + ' manifest.json\n'
|
|
put(exp / 'checksums.sha256', data.encode())
|
|
|
|
|
|
class PublisherTest(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 / P.EXP_REL
|
|
self.registry = self.root / P.REGISTRY_REL
|
|
self.model = self.root / P.MODEL_REL
|
|
put(self.root / P.HELPER_REL, HELPER.read_bytes())
|
|
|
|
def fixture(self, status='PASS-EXACT-TWO-EPOCH-DELTA-INTERFACE-NOT-LEARNING', count=3):
|
|
sources, prepared = {}, {}
|
|
names = ['worker.py', 'test_worker.py', 'PROTOCOL.md'] + [f'source_{i:03d}.py' for i in range(max(0, count - 3))]
|
|
for name in names[:count]:
|
|
raw = ('# synthetic registered source ' + name + '\n').encode()
|
|
original = self.model / name
|
|
snapshot = self.exp / P.SNAPSHOT / name
|
|
sha = put(original, raw)
|
|
put(snapshot, raw)
|
|
sources[str(original)] = sha
|
|
prepared[str(snapshot)] = sha
|
|
full_plan = put(self.exp / P.FULL_PLAN, dict(draft_only=True, no_brain_inputs=True))
|
|
prepared[str(self.exp / P.FULL_PLAN)] = full_plan
|
|
self.registration = dict(schema='phase4ao-conditioning-interface-registration-v1',
|
|
source_sha256=sources, prepared_file_sha256=prepared,
|
|
full_trial_plan_file_sha256=full_plan, git_code_commit='4' * 40,
|
|
committed_source_blobs_verified=len(sources))
|
|
self.reg_sha = put(self.exp / P.REGISTRATION, self.registration)
|
|
analysis_path = P.FAILURE_ANALYSIS if status.startswith('FAIL') else P.ANALYSIS
|
|
analysis_sha = put(self.exp / analysis_path, dict(status=status, registration_sha256=self.reg_sha,
|
|
complete_conditioning_executed=False, completed_rows=[] if status.startswith('FAIL') else ['synthetic']))
|
|
self.final = dict(status=status, registration_sha256=self.reg_sha, analysis_sha256=analysis_sha,
|
|
learned_candidate=False, learning_success_claimed=False, full_conditioning_executed=False)
|
|
put(self.exp / P.FINAL, self.final)
|
|
put(self.exp / 'logs/worker.log', b'partial failures are retained\n')
|
|
put(self.exp / 'runs/first/summary.json', dict(complete=True))
|
|
put(self.exp / 'runs/first/state.npz', b'fake binary')
|
|
put(self.exp / 'runs/first/hashes.jsonl', b'{}\n')
|
|
put(self.exp / 'extension-build/compiler.log', b'build output not exported')
|
|
put(self.exp / 'reference.npy', b'not a text file')
|
|
put(self.exp / 'large.json', b' ' * (P.MAX_BYTES + 1))
|
|
self.reseal()
|
|
|
|
def reseal(self):
|
|
seal(self.exp)
|
|
self.anchors = (P.digest(self.exp / P.REGISTRATION), P.digest(self.exp / 'manifest.json'),
|
|
P.digest(self.exp / P.FINAL))
|
|
|
|
def publish(self, *anchors):
|
|
return P.publish(*(anchors or self.anchors), root=self.root)
|
|
|
|
def test_success_complete_small_seal_inventory_and_51_dynamic_sources(self):
|
|
self.fixture(count=51)
|
|
original = {p: P.digest(p) for p in self.exp.rglob('*') if p.is_file()}
|
|
result = self.publish()
|
|
self.assertTrue(result['publication_complete'])
|
|
self.assertEqual(result['source_count'], 51)
|
|
self.assertFalse(result['complete_bulk_backup'])
|
|
helper = P.load_helper(self.root / P.HELPER_REL)
|
|
verified = helper.verify_seal(self.registry)
|
|
self.assertEqual(verified['manifest_sha256'], result['registry_manifest_sha256'])
|
|
receipt = read(self.registry / 'publication-receipt.json')
|
|
self.assertEqual(result['receipt_sha256'], P.digest(self.registry / 'publication-receipt.json'))
|
|
self.assertEqual(receipt['registration_sha256'], self.anchors[0])
|
|
self.assertFalse(receipt['scientific_gate_certified_by_publication'])
|
|
for entry in receipt['files']:
|
|
self.assertEqual(P.digest(self.registry / entry['path']), entry['sha256'])
|
|
inventory = read(self.registry / 'artifact-inventory.json')['files']
|
|
rows = {r['path']: r for r in inventory}
|
|
self.assertEqual(set(rows), {p.relative_to(self.exp).as_posix() for p in original})
|
|
self.assertEqual(rows[P.FULL_PLAN]['reason'], 'full_trial_plan_intentionally_not_exported')
|
|
self.assertEqual(rows['large.json']['reason'], 'exceeds_4_MiB')
|
|
for name in ('runs/first/state.npz', 'reference.npy', 'runs/first/hashes.jsonl', 'extension-build/compiler.log'):
|
|
self.assertEqual(rows[name]['disposition'], 'omitted')
|
|
self.assertEqual(rows[name]['sha256'], P.digest(self.exp / name))
|
|
self.assertEqual((self.registry / 'evidence/manifest.json').read_bytes(), (self.exp / 'manifest.json').read_bytes())
|
|
self.assertEqual((self.registry / 'evidence/checksums.sha256').read_bytes(), (self.exp / 'checksums.sha256').read_bytes())
|
|
self.assertTrue(all(p.stat().st_size <= P.MAX_BYTES for p in self.registry.rglob('*') if p.is_file()))
|
|
self.assertEqual(original, {p: P.digest(p) for p in self.exp.rglob('*') if p.is_file()})
|
|
|
|
def test_fail_with_partial_run_no_execution_complete_is_preserved(self):
|
|
self.fixture('FAIL-POSTPROCESS-MODULE-CACHE')
|
|
result = self.publish()
|
|
self.assertEqual(result['declared_audit_status'], 'FAIL-POSTPROCESS-MODULE-CACHE')
|
|
receipt = read(self.registry / 'publication-receipt.json')
|
|
self.assertEqual(receipt['analysis_path'], P.FAILURE_ANALYSIS)
|
|
self.assertFalse((self.registry / 'evidence/audit/execution-complete.json').exists())
|
|
self.assertFalse(receipt['learning_achieved'])
|
|
|
|
def test_incomplete_status_is_not_upgraded(self):
|
|
self.fixture('INCOMPLETE-NO-NONZERO-UPDATE-COVERAGE')
|
|
self.assertEqual(self.publish()['declared_audit_status'], 'INCOMPLETE-NO-NONZERO-UPDATE-COVERAGE')
|
|
|
|
def test_three_external_anchors_required_and_checked(self):
|
|
self.fixture()
|
|
for index in range(3):
|
|
values = list(self.anchors)
|
|
values[index] = '0' * 64
|
|
with self.subTest(index=index), self.assertRaisesRegex(ValueError, 'External anchor mismatch'):
|
|
self.publish(*values)
|
|
with self.assertRaises(ValueError):
|
|
self.publish('', self.anchors[1], self.anchors[2])
|
|
self.assertFalse(self.registry.exists())
|
|
|
|
def test_existing_registry_is_never_touched(self):
|
|
self.fixture()
|
|
marker = self.registry / 'do-not-touch.txt'
|
|
put(marker, b'keep')
|
|
with self.assertRaisesRegex(ValueError, 'already exists'):
|
|
self.publish()
|
|
self.assertEqual(marker.read_bytes(), b'keep')
|
|
self.assertEqual(len(list(self.registry.iterdir())), 1)
|
|
|
|
def test_original_source_mutation_rejected_even_with_valid_experiment_seal(self):
|
|
self.fixture()
|
|
put(self.model / 'worker.py', b'# other source')
|
|
with self.assertRaisesRegex(ValueError, 'differ'):
|
|
self.publish()
|
|
self.assertFalse(self.registry.exists())
|
|
|
|
def test_snapshot_mutation_resealed_still_rejected_by_registration(self):
|
|
self.fixture()
|
|
put(self.exp / P.SNAPSHOT / 'worker.py', b'# changed')
|
|
self.reseal()
|
|
with self.assertRaisesRegex(ValueError, 'differ'):
|
|
self.publish()
|
|
|
|
def test_extra_snapshot_is_not_silently_exported(self):
|
|
self.fixture()
|
|
put(self.exp / P.SNAPSHOT / 'unregistered.py', b'# not registered')
|
|
self.reseal()
|
|
with self.assertRaisesRegex(ValueError, 'snapshot inventory'):
|
|
self.publish()
|
|
|
|
def test_source_escape_rejected_after_self_consistent_resealing(self):
|
|
self.fixture()
|
|
old = str(self.model / 'worker.py')
|
|
checksum = self.registration['source_sha256'].pop(old)
|
|
external = self.root / 'unrelated/worker.py'
|
|
put(external, (self.model / 'worker.py').read_bytes())
|
|
self.registration['source_sha256'][str(external)] = checksum
|
|
self.reg_sha = put(self.exp / P.REGISTRATION, self.registration)
|
|
analysis = read(self.exp / P.ANALYSIS)
|
|
analysis['registration_sha256'] = self.reg_sha
|
|
self.final['analysis_sha256'] = put(self.exp / P.ANALYSIS, analysis)
|
|
self.final['registration_sha256'] = self.reg_sha
|
|
put(self.exp / P.FINAL, self.final)
|
|
self.reseal()
|
|
with self.assertRaisesRegex(ValueError, 'escapes the fixed model'):
|
|
self.publish()
|
|
|
|
def test_unlisted_file_or_bad_checksum_control_rejected(self):
|
|
self.fixture()
|
|
put(self.exp / 'unlisted.txt', b'not sealed')
|
|
with self.assertRaisesRegex(ValueError, 'unlisted'):
|
|
self.publish()
|
|
(self.exp / 'unlisted.txt').unlink()
|
|
with (self.exp / 'checksums.sha256').open('ab') as handle:
|
|
handle.write(b'bad line\n')
|
|
with self.assertRaisesRegex(ValueError, 'Malformed checksums'):
|
|
self.publish()
|
|
|
|
def test_final_analysis_link_and_status_must_agree(self):
|
|
self.fixture()
|
|
self.final['analysis_sha256'] = '0' * 64
|
|
put(self.exp / P.FINAL, self.final)
|
|
self.reseal()
|
|
with self.assertRaisesRegex(ValueError, 'analysis link'):
|
|
self.publish()
|
|
self.final['analysis_sha256'] = P.digest(self.exp / P.ANALYSIS)
|
|
self.final['status'] = 'FAIL-OTHER'
|
|
put(self.exp / P.FINAL, self.final)
|
|
self.reseal()
|
|
with self.assertRaisesRegex(ValueError, 'identities differ'):
|
|
self.publish()
|
|
|
|
def test_no_learning_claim_and_helper_pin_cannot_be_bypassed(self):
|
|
self.fixture()
|
|
self.final['learning_success_claimed'] = True
|
|
put(self.exp / P.FINAL, self.final)
|
|
self.reseal()
|
|
with self.assertRaisesRegex(ValueError, 'scope flag'):
|
|
self.publish()
|
|
put(self.root / P.HELPER_REL, b'# alternate helper')
|
|
with self.assertRaisesRegex(ValueError, 'helper SHA mismatch'):
|
|
self.publish()
|
|
|
|
def test_postcopy_full_seal_detects_change_in_omitted_bulk_and_preserves_partial(self):
|
|
self.fixture()
|
|
helper = P.load_helper(self.root / P.HELPER_REL)
|
|
original_copy = helper._copy_verified
|
|
changed = False
|
|
def interrupted_copy(*args):
|
|
nonlocal changed
|
|
original_copy(*args)
|
|
if not changed:
|
|
changed = True
|
|
put(self.exp / 'reference.npy', b'changed omitted bytes')
|
|
# Test-local injected failure only; production never changes helper globals.
|
|
with mock.patch.object(helper, '_copy_verified', side_effect=interrupted_copy), mock.patch.object(P, 'load_helper', return_value=helper):
|
|
with self.assertRaisesRegex(ValueError, 'Payload byte count|Payload SHA'):
|
|
self.publish()
|
|
self.assertTrue(self.registry.exists())
|
|
self.assertTrue(read(self.registry / 'publication-failure.json')['partial_outputs_preserved'])
|
|
self.assertFalse((self.registry / 'publication-receipt.json').exists())
|
|
with self.assertRaisesRegex(ValueError, 'already exists'):
|
|
self.publish()
|
|
|
|
def test_small_registry_seal_catches_copied_payload_mutation(self):
|
|
self.fixture()
|
|
self.publish()
|
|
path = self.registry / 'evidence/logs/worker.log'
|
|
put(path, b'tampered after publication')
|
|
helper = P.load_helper(self.root / P.HELPER_REL)
|
|
with self.assertRaisesRegex(ValueError, 'Payload byte count|Payload SHA'):
|
|
helper.verify_seal(self.registry)
|
|
|
|
def test_valid_reseal_between_anchor_read_and_verification_is_rejected(self):
|
|
self.fixture()
|
|
helper = P.load_helper(self.root / P.HELPER_REL)
|
|
original_verify = helper.verify_seal
|
|
def resealed_verify(directory):
|
|
put(self.exp / 'logs/worker.log', b'changed after external anchor read')
|
|
seal(self.exp)
|
|
return original_verify(directory)
|
|
with mock.patch.object(helper, 'verify_seal', side_effect=resealed_verify), mock.patch.object(P, 'load_helper', return_value=helper):
|
|
with self.assertRaisesRegex(ValueError, 'differs from external'):
|
|
self.publish()
|
|
self.assertFalse(self.registry.exists())
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|