Digital Drosophila Archive
194 lines
12 KiB
Python
194 lines
12 KiB
Python
"""Pure temporary sealed trees; no real AQ publication or scientific run."""
|
|
import copy
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
import publish_phase4aq as P
|
|
|
|
HELPER_CANDIDATES = (Path(__file__).resolve().parents[1] / 'pn-recruitment-audit-v1/publication.py',
|
|
Path(__file__).resolve().parents[2] / P.HELPER_REL)
|
|
HELPER = next((p for p in HELPER_CANDIDATES if p.is_file()), HELPER_CANDIDATES[0])
|
|
|
|
|
|
def put(path, value):
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
raw = value if type(value) is bytes else (json.dumps(value, sort_keys=True) + '\n').encode()
|
|
path.write_bytes(raw)
|
|
return P.digest(path)
|
|
|
|
|
|
def read(path):
|
|
return json.loads(path.read_text(encoding='utf-8'))
|
|
|
|
|
|
def seal(root):
|
|
rows = [r for r in P.file_rows(root) if r['path'] not in ('manifest.json', 'checksums.sha256')]
|
|
put(root / 'manifest.json', dict(root=str(root), files=rows))
|
|
raw = ''.join(r['sha256'] + ' ' + r['path'] + '\n' for r in rows)
|
|
put(root / 'checksums.sha256', (raw + P.digest(root / 'manifest.json') + ' manifest.json\n').encode())
|
|
|
|
|
|
class PublicationTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory(); self.addCleanup(self.temp.cleanup)
|
|
self.root = Path(self.temp.name).resolve()
|
|
self.exp, self.model, self.registry = (self.root / x for x in (P.EXP_REL, P.MODEL_REL, P.REGISTRY_REL))
|
|
put(self.root / P.HELPER_REL, HELPER.read_bytes())
|
|
|
|
def fixture(self, status='PASS-EXACT-UPDATED-BOUNDARY-ENGINEERING-NOT-LEARNING', source_count=3):
|
|
sources = {}; prepared = {}
|
|
for i in range(source_count):
|
|
name = f'new_source_{i}.py'; raw = ('# synthetic source ' + str(i)).encode()
|
|
sha = put(self.model / name, raw); sources[str(self.model / name)] = sha
|
|
prepared[str(self.exp / P.SNAPSHOT / name)] = put(self.exp / P.SNAPSHOT / name, raw)
|
|
self.reg = dict(schema='conditioning-execution-registration-v1', engineering_only=True,
|
|
scientific_acquisition_authorized=False, source_sha256=sources, prepared_file_sha256=prepared,
|
|
source_code_commit='a'*40, plan=dict(calls=[dict(call_id=f'call-{i}',steps=1000 if i<22 else 103200) for i in range(23)],
|
|
counts=dict(new_brain_calls=23,new_brain_ticks=125200,cpu_children=7,gpu_sampler_children=1,total_children=31)))
|
|
self.execution = dict(complete=True, completed_calls=[c['call_id'] for c in self.reg['plan']['calls']],
|
|
scientific_learning_pass_claimed=False,candidate_generated=False,processes={})
|
|
evidence = {}
|
|
for i in range(31):
|
|
name = f'process-{i}'; row = {'recorded_exit_code':0}
|
|
for field, rel in (('process_sha256',f'receipts/{name}.process.json'),
|
|
('worker_sha256',f'receipts/{name}.worker.json'),('job_sha256',f'jobs/{name}.json')):
|
|
row[field] = put(self.exp / rel, {'synthetic':name,'role':field})
|
|
evidence[name] = row; self.execution['processes'][name] = str(self.exp / 'receipts' / (name+'.process.json'))
|
|
self.numerical = dict(schema='conditioning-execution-numerical-audit-v1',complete=True,status=status,
|
|
pass_all=status.startswith('PASS-'),learning_pass_claimed=False,candidate_generated=False,
|
|
actual_new_brain_calls=23,actual_new_brain_ticks=125200,actual_processes=31,actual_CPU_processes=7,
|
|
actual_GPU_sampler_processes=1,process_evidence=evidence)
|
|
self.budget = dict(schema='conditioning-actual-budget-report-v1', status='INSUFFICIENT-MEASUREMENTS',
|
|
learning_pass_claimed=False, engineering_scope=True,normal_budget_authorized=False,execution_authorized=False)
|
|
self.integrity = dict(complete=True,AP={'pass_all':True},prior={'synthetic':'prior checks'},anchors={'synthetic':'b'*64},
|
|
old_experiments_reexecuted=False,old_files_modified=False,five_frozen_reference_trials_rerun_here=False,
|
|
scientific_learning_pass_claimed=False)
|
|
self.final = dict(schema='conditioning-execution-final-v1',status=status,source_code_commit='a'*40,
|
|
numerical_and_budget_recomputed_at_seal=True,actual_new_brain_calls=23,actual_new_brain_ticks=125200,
|
|
actual_execution_child_processes=31,historical_integrity_after_pass=True,
|
|
historical_five_trial_regression='AM sealed evidence preserved; not rerun here',
|
|
scientific_acquisition_executed=False,scientific_acquisition_authorized=False,candidate_generated=False,
|
|
learning_pass_claimed=False,entire_model_directory_sealed=False,bulk_data_git_backup_claimed=False)
|
|
put(self.exp / 'runs/one/state.npz', b'binary-not-exported')
|
|
put(self.exp / 'extension-build/info.json', {'not':'small evidence despite suffix'})
|
|
put(self.exp / 'protocol/full-trial-plan.json', {'not':'exported despite small size'})
|
|
put(self.exp / 'logs/one.stdout.log', b'actual fixture log')
|
|
self.refresh()
|
|
|
|
def refresh(self):
|
|
regsha = put(self.exp / P.REGISTRATION,self.reg)
|
|
for value in (self.final,self.numerical,self.budget,self.execution): value['registration_sha256']=regsha
|
|
self.numerical['execution_complete_sha256']=put(self.exp/'audit/execution-complete.json',self.execution)
|
|
self.final['numerical_sha256']=put(self.exp/P.REPORTS['numerical_sha256'],self.numerical)
|
|
self.final['budget_sha256']=put(self.exp/P.REPORTS['budget_sha256'],self.budget)
|
|
self.final['integrity_after_sha256']=put(self.exp/P.REPORTS['integrity_after_sha256'],self.integrity)
|
|
self.final['integrity_at_seal_sha256']=put(self.exp/P.REPORTS['integrity_at_seal_sha256'],self.integrity)
|
|
finalsha=put(self.exp/P.FINAL,self.final);seal(self.exp)
|
|
self.anchors=(regsha,P.digest(self.exp/'manifest.json'),finalsha)
|
|
|
|
def publish(self, anchors=None):
|
|
return P.publish(*(self.anchors if anchors is None else anchors),root=self.root)
|
|
|
|
def test_complete_small_registry_dynamic_sources_omissions_and_seal(self):
|
|
self.fixture(source_count=5); result=self.publish()
|
|
self.assertEqual(result['source_count'],5); self.assertTrue(result['publication_complete'])
|
|
receipt=read(self.registry/'publication-receipt.json')
|
|
self.assertFalse(receipt['complete_bulk_backup']);self.assertFalse(receipt['learning_achieved'])
|
|
self.assertEqual(receipt['declared_budget_status'],'INSUFFICIENT-MEASUREMENTS')
|
|
self.assertFalse(receipt['normal_training_or_budget_authorized'])
|
|
self.assertTrue(P.load_helper(self.root/P.HELPER_REL).verify_seal(self.registry)['pass_all'])
|
|
inv=read(self.registry/'artifact-inventory.json')['files']
|
|
omitted={r['path']:r for r in inv if r['disposition']=='omitted'}
|
|
self.assertTrue({'runs/one/state.npz','extension-build/info.json','protocol/full-trial-plan.json'} <= set(omitted))
|
|
self.assertTrue(all('sha256' in r and 'bytes' in r for r in omitted.values()))
|
|
self.assertFalse((self.registry/'evidence/runs/one/state.npz').exists())
|
|
with self.assertRaisesRegex(ValueError,'never overwrite'): self.publish()
|
|
|
|
def test_FAIL_and_INCOMPLETE_preserved_without_promotion(self):
|
|
for status in ('FAIL-NUMERICAL-OR-LINEAGE','INCOMPLETE-CHANGED-EDGE-COVERAGE'):
|
|
with self.subTest(status=status),tempfile.TemporaryDirectory() as name:
|
|
saved=(self.root,self.exp,self.model,self.registry)
|
|
self.root=Path(name).resolve();self.exp,self.model,self.registry=(self.root/x for x in (P.EXP_REL,P.MODEL_REL,P.REGISTRY_REL))
|
|
put(self.root/P.HELPER_REL,HELPER.read_bytes());self.fixture(status)
|
|
result=self.publish();self.assertEqual(result['declared_audit_status'],status)
|
|
self.assertFalse(result['scientific_gate_certified_by_publication'])
|
|
self.root,self.exp,self.model,self.registry=saved
|
|
|
|
def test_three_external_anchors_and_helper_pin_are_required(self):
|
|
self.fixture()
|
|
for index in range(3):
|
|
bad=list(self.anchors);bad[index]='0'*64
|
|
with self.subTest(index=index),self.assertRaisesRegex(ValueError,'External anchor'): self.publish(bad)
|
|
put(self.root/P.HELPER_REL,b'not original')
|
|
with self.assertRaisesRegex(ValueError,'helper SHA'):self.publish()
|
|
|
|
def test_missing_final_seal_controls_and_partial_workflow_rejected(self):
|
|
self.fixture();(self.exp/'checksums.sha256').unlink()
|
|
with self.assertRaisesRegex(ValueError,'Missing/nonregular'):self.publish()
|
|
self.execution['complete']=False;self.refresh()
|
|
with self.assertRaisesRegex(ValueError,'complete recorded execution'):self.publish()
|
|
self.assertFalse(self.registry.exists())
|
|
|
|
def test_current_source_and_snapshot_mutations_rejected(self):
|
|
self.fixture();source=self.model/'new_source_0.py';old=source.read_bytes();source.write_bytes(b'changed')
|
|
with self.assertRaisesRegex(ValueError,'bytes differ'):self.publish()
|
|
source.write_bytes(old);put(self.exp/P.SNAPSHOT/'new_source_0.py',b'changed snapshot');seal(self.exp)
|
|
anchors=(self.anchors[0],P.digest(self.exp/'manifest.json'),self.anchors[2])
|
|
with self.assertRaisesRegex(ValueError,'bytes differ'):self.publish(anchors)
|
|
|
|
def test_source_escape_or_extra_snapshot_rejected_even_resealed(self):
|
|
self.fixture();key=str(self.model/'new_source_0.py')
|
|
self.reg['source_sha256'][str(self.root/'elsewhere/new_source_0.py')]=self.reg['source_sha256'].pop(key)
|
|
self.refresh()
|
|
with self.assertRaisesRegex(ValueError,'escapes exact AQ'):self.publish()
|
|
self.fixture();put(self.exp/P.SNAPSHOT/'unregistered.py',b'extra');seal(self.exp)
|
|
with self.assertRaisesRegex(ValueError,'snapshot inventory'):
|
|
self.publish((self.anchors[0],P.digest(self.exp/'manifest.json'),self.anchors[2]))
|
|
|
|
def test_all_linked_report_SHAs_checked(self):
|
|
self.fixture()
|
|
for field in P.REPORTS:
|
|
original=self.final[field];self.final[field]='0'*64;put(self.exp/P.FINAL,self.final);seal(self.exp)
|
|
with self.subTest(field=field),self.assertRaisesRegex(ValueError,'linked report SHA'):
|
|
self.publish((self.anchors[0],P.digest(self.exp/'manifest.json'),P.digest(self.exp/P.FINAL)))
|
|
self.final[field]=original
|
|
|
|
def test_historical_prior_disagreement_and_learning_overclaim_rejected(self):
|
|
self.fixture();changed=copy.deepcopy(self.integrity);changed['prior']['synthetic']='different'
|
|
self.final['integrity_at_seal_sha256']=put(self.exp/P.REPORTS['integrity_at_seal_sha256'],changed)
|
|
put(self.exp/P.FINAL,self.final);seal(self.exp)
|
|
with self.assertRaisesRegex(ValueError,'integrity reports disagree'):
|
|
self.publish((self.anchors[0],P.digest(self.exp/'manifest.json'),P.digest(self.exp/P.FINAL)))
|
|
self.final['learning_pass_claimed']=True;self.refresh()
|
|
with self.assertRaisesRegex(ValueError,'scientific/backup claim'):self.publish()
|
|
|
|
def test_process_reference_rebinding_rejected_even_resealed(self):
|
|
self.fixture();self.numerical['process_evidence']['process-0']['worker_sha256']='0'*64;self.refresh()
|
|
with self.assertRaisesRegex(ValueError,'process file reference'):self.publish()
|
|
|
|
def test_oversize_required_evidence_and_unlisted_payload_rejected(self):
|
|
self.fixture();put(self.exp/'extra.json',{} )
|
|
with self.assertRaisesRegex(ValueError,'unlisted'):self.publish()
|
|
(self.exp/'extra.json').unlink()
|
|
with patch.object(P,'MAX_BYTES',100):
|
|
with self.assertRaisesRegex(ValueError,'oversize'):self.publish()
|
|
|
|
def test_postcopy_omitted_bulk_mutation_preserves_partial_registry(self):
|
|
self.fixture();helper=P.load_helper(self.root/P.HELPER_REL);original=helper._copy_verified;changed=False
|
|
def copying(*args):
|
|
nonlocal changed
|
|
original(*args)
|
|
if not changed:
|
|
changed=True;put(self.exp/'runs/one/state.npz',b'mutated omitted file')
|
|
with patch.object(P,'load_helper',return_value=helper),patch.object(helper,'_copy_verified',side_effect=copying):
|
|
with self.assertRaisesRegex(ValueError,'Payload byte count|Payload SHA'):self.publish()
|
|
self.assertTrue(read(self.registry/'publication-failure.json')['partial_outputs_preserved'])
|
|
self.assertFalse((self.registry/'publication-receipt.json').exists())
|
|
with self.assertRaisesRegex(ValueError,'never overwrite'):self.publish()
|
|
|
|
|
|
if __name__=='__main__':unittest.main()
|