Digital Drosophila Archive
177 lines
11 KiB
Python
177 lines
11 KiB
Python
"""CPU/path/admission tests only; no subprocess brain or scientific PASS."""
|
|
import copy
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest.mock import Mock, patch
|
|
|
|
import isolated_execution as W
|
|
|
|
H = W.ap_helpers()
|
|
|
|
|
|
class WorkerTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.root = Path(self.temp.name).resolve() / 'phase4aq-fixture'
|
|
for name in ('inputs', 'updates', 'jobs', 'runs', 'receipts'):
|
|
(self.root / name).mkdir(parents=True)
|
|
self.authority = self.root / 'inputs' / 'authority.json'
|
|
self.chain = self.root / 'updates' / 'chain-receipt.json'
|
|
self.authority.write_text('{}'); self.chain.write_text('{}')
|
|
self.jobfile = self.root / 'jobs' / 'one.json'
|
|
self.job = dict(schema=W.SCHEMA, job_id='cpu-one', kind='validate_cpu',
|
|
registration_sha256='1'*64, worker_sha256=H.digest(W.__file__),
|
|
source_sha256={'validate_update_cpu.py': H.digest(W.MODEL/'validate_update_cpu.py')},
|
|
input_sha256={str(p): H.digest(p) for p in (self.authority, self.chain)},
|
|
output_dir=str(self.root/'runs'/'one'), receipt_file=str(self.root/'receipts'/'one.json'),
|
|
budget=dict(root=str(self.root), output_cap_bytes=1024**2, minimum_free_reserve_bytes=1024**2),
|
|
authority_file=str(self.authority), authority_sha256=H.digest(self.authority),
|
|
update_dir=str(self.chain.parent), chain_sha256=H.digest(self.chain))
|
|
|
|
def tearDown(self):
|
|
self.temp.cleanup()
|
|
|
|
def check(self, job=None):
|
|
self.jobfile.write_text(json.dumps(self.job if job is None else job))
|
|
return W.validate_job(H, self.jobfile, H.digest(self.jobfile))
|
|
|
|
def test_real_pinned_AP_helper_and_CPU_job(self):
|
|
self.assertEqual(H.digest(W.AP), W.AP_SHA)
|
|
job, paths, pins = self.check()
|
|
self.assertEqual(job['kind'], 'validate_cpu')
|
|
self.assertEqual(pins[W.AP], W.AP_SHA)
|
|
self.assertEqual(paths['output'], self.root/'runs'/'one')
|
|
|
|
def test_wrong_or_missing_fields(self):
|
|
changes = [('kind', 'brain'), ('job_id', '../x'), ('job_id', 'a/b'),
|
|
('registration_sha256', ''), ('worker_sha256', '0'*64),
|
|
('source_sha256', {}), ('input_sha256', {})]
|
|
for key, value in changes:
|
|
with self.subTest(key=key, value=value):
|
|
job = copy.deepcopy(self.job); job[key] = value
|
|
with self.assertRaises(ValueError): self.check(job)
|
|
job = copy.deepcopy(self.job); job['extra'] = True
|
|
with self.assertRaises(ValueError): self.check(job)
|
|
|
|
def test_source_bytes_are_checked_before_import(self):
|
|
job = copy.deepcopy(self.job); job['source_sha256']['validate_update_cpu.py'] = '0'*64
|
|
with self.assertRaisesRegex(ValueError, 'Pinned file changed'): self.check(job)
|
|
|
|
def test_parent_input_pin_and_chain_are_mandatory(self):
|
|
for key in ('authority_sha256', 'chain_sha256'):
|
|
job = copy.deepcopy(self.job); job[key] = '0'*64
|
|
with self.subTest(key=key), self.assertRaises(ValueError): self.check(job)
|
|
job = copy.deepcopy(self.job); del job['input_sha256'][str(self.chain)]
|
|
with self.assertRaises(ValueError): self.check(job)
|
|
|
|
def test_partial_output_and_attempt_never_retried(self):
|
|
output = Path(self.job['output_dir']); output.mkdir()
|
|
with self.assertRaises(ValueError): self.check()
|
|
output.rmdir()
|
|
attempt = Path(self.job['receipt_file']+'.attempt.json'); attempt.write_text('{}')
|
|
with self.assertRaises(ValueError): self.check()
|
|
|
|
def test_output_cannot_overlap_input_update_or_receipt(self):
|
|
for path in (self.authority.parent/'child', self.chain.parent/'child',
|
|
self.root/'receipts'/'child', self.root.parent/'escape'):
|
|
job = copy.deepcopy(self.job); job['output_dir'] = str(path)
|
|
with self.subTest(path=path), self.assertRaises(ValueError): self.check(job)
|
|
|
|
def test_linked_output_parent_rejected_when_supported(self):
|
|
target = self.root/'runs'; linked = self.root/'linked'
|
|
try: linked.symlink_to(target, target_is_directory=True)
|
|
except OSError: self.skipTest('Platform does not allow test symlinks')
|
|
job = copy.deepcopy(self.job); job['output_dir'] = str(linked/'one')
|
|
with self.assertRaisesRegex(ValueError, 'Linked'): self.check(job)
|
|
|
|
def test_explicit_resource_limits(self):
|
|
job, paths, _ = self.check()
|
|
result = W.budget_check(H, job, paths)
|
|
self.assertGreater(result['output_bytes'], 0)
|
|
job['budget']['output_cap_bytes'] = 1
|
|
with self.assertRaisesRegex(ValueError, 'Resource budget exceeded'): W.budget_check(H, job, paths)
|
|
for field in ('output_cap_bytes', 'minimum_free_reserve_bytes'):
|
|
job = copy.deepcopy(self.job); job['budget'][field] = 0
|
|
with self.subTest(field=field), self.assertRaises(ValueError): self.check(job)
|
|
|
|
def test_fresh_process_loaders_are_not_reused(self):
|
|
with self.assertRaisesRegex(ValueError, 'Fresh exec helper'): W.ap_helpers()
|
|
|
|
def unchanged(self):
|
|
case=self.root/'inputs'/'case';case.mkdir()
|
|
parent=self.root/'inputs'/'checkpoint';parent.mkdir()
|
|
for p in (case/'spec.json',case/'requests.npz',parent/'checkpoint-manifest.json'):
|
|
p.write_text('{}')
|
|
before=self.root/'inputs'/'before.npz';bm=self.root/'inputs'/'before.json'
|
|
after=self.root/'inputs'/'after.npz';am=self.root/'inputs'/'after.json'
|
|
before.write_bytes(b'zero-values');after.write_bytes(b'zero-values')
|
|
bm.write_text('{"phase":"before"}');am.write_text('{"phase":"after"}')
|
|
files=(case/'spec.json',case/'requests.npz',parent/'checkpoint-manifest.json',before,bm,after,am,W.C_PATH)
|
|
authority=dict(schema='conditioning-updated-boundary-v1',transition='epoch_update',diagnostic_mode='reference',
|
|
registration_sha256='1'*64,expected_effective_weight_before_sha256='a'*64,expected_effective_weight_after_sha256='a'*64,
|
|
parent=dict(directory=str(parent),checkpoint_manifest_sha256=H.digest(parent/'checkpoint-manifest.json')),
|
|
case={key:dict(path=str(case/name),sha256=H.digest(case/name)) for key,name in (('spec','spec.json'),('requests','requests.npz'))},
|
|
overlay_before={key:dict(path=str(p),sha256=H.digest(p)) for key,p in (('file',before),('manifest',bm))},
|
|
overlay_after={key:dict(path=str(p),sha256=H.digest(p)) for key,p in (('file',after),('manifest',am))},frozen_inputs={})
|
|
self.authority.write_text(json.dumps(authority))
|
|
self.job=dict(schema=W.SCHEMA,job_id='unchanged-one',kind='unchanged',registration_sha256='1'*64,
|
|
worker_sha256=H.digest(W.__file__),source_sha256={},input_sha256={str(p):H.digest(p) for p in (*files,self.authority)},
|
|
output_dir=str(self.root/'runs'/'one'),receipt_file=str(self.root/'receipts'/'one.json'),
|
|
budget=dict(root=str(self.root),output_cap_bytes=1024**2,minimum_free_reserve_bytes=1024**2),
|
|
authority_file=str(self.authority),authority_sha256=H.digest(self.authority),case_dir=str(case),
|
|
build_root=str(self.root/'build'),diagnostic_mode='reference',overlay_file=str(before),overlay_manifest=str(bm),
|
|
checkpoint_dir=str(parent),checkpoint_sha256=H.digest(parent/'checkpoint-manifest.json'))
|
|
return authority
|
|
|
|
def test_unchanged_binds_exact_before_not_equal_after_manifest(self):
|
|
self.unchanged();job,paths,pins=self.check()
|
|
self.assertEqual(W.bind_unchanged_authority(H,job,paths,pins)['transition'],'epoch_update')
|
|
self.job['overlay_file']=str(self.root/'inputs'/'after.npz')
|
|
self.job['overlay_manifest']=str(self.root/'inputs'/'after.json')
|
|
job,paths,pins=self.check()
|
|
with self.assertRaisesRegex(ValueError,'actual before'):W.bind_unchanged_authority(H,job,paths,pins)
|
|
|
|
def test_rehashed_wrong_authority_parent_mode_case_registration_rejected(self):
|
|
original=self.unchanged()
|
|
for field in ('parent','diagnostic_mode','registration_sha256','case'):
|
|
value=copy.deepcopy(original)
|
|
if field=='parent':value[field]['checkpoint_manifest_sha256']='f'*64
|
|
elif field=='diagnostic_mode':value[field]='light'
|
|
elif field=='registration_sha256':value[field]='f'*64
|
|
else:value[field]['requests']['sha256']='f'*64
|
|
self.authority.write_text(json.dumps(value))
|
|
self.job['authority_sha256']=self.job['input_sha256'][str(self.authority)]=H.digest(self.authority)
|
|
job,paths,pins=self.check()
|
|
with self.subTest(field=field),self.assertRaises(ValueError):W.bind_unchanged_authority(H,job,paths,pins)
|
|
|
|
def test_unchanged_missing_nested_after_authority_pin_rejected(self):
|
|
self.unchanged();del self.job['input_sha256'][str(self.root/'inputs'/'after.json')]
|
|
job,paths,pins=self.check()
|
|
with self.assertRaisesRegex(ValueError,'external input pin'):W.bind_unchanged_authority(H,job,paths,pins)
|
|
|
|
def test_unchanged_original_entry_already_seals_no_double_seal(self):
|
|
self.unchanged();self.check()
|
|
summary={'apl_component':{'environment':{'synthetic':True}},'complete':True,'steps':5}
|
|
def simulated_entry(case,out,build,overlay,manifest,parent,*,mode):
|
|
out.mkdir();H.save_json(out/'summary.json',summary)
|
|
H.save_json(out/'checkpoint-manifest.json',{'synthetic_seal':True})
|
|
return dict(summary,checkpoint_manifest_sha256=H.digest(out/'checkpoint-manifest.json'))
|
|
runtime=SimpleNamespace(run_continued=Mock(side_effect=simulated_entry),
|
|
load_checkpoint=Mock(return_value={'summary':summary}),
|
|
seal_checkpoint=Mock(side_effect=AssertionError('Double seal would be a real blocker')))
|
|
with patch.object(W,'ap_helpers',return_value=H),patch.object(H,'_clean_process') as clean, \
|
|
patch.object(H,'_admit_brain',return_value={'synthetic_no_GPU':True}), \
|
|
patch.object(H,'_load_source',return_value=runtime),patch.object(H,'_environment',return_value={'synthetic_no_GPU':True}), \
|
|
patch.object(H,'_verify_precision'):
|
|
result=W.execute(self.jobfile,H.digest(self.jobfile))
|
|
runtime.run_continued.assert_called_once();runtime.load_checkpoint.assert_called_once();runtime.seal_checkpoint.assert_not_called()
|
|
clean.assert_called_once_with('brain');self.assertTrue(result['complete'])
|
|
self.assertFalse(result['process_exit_independently_observed'])
|
|
self.assertFalse(result['scientific_learning_pass_claimed'])
|
|
|
|
|
|
if __name__ == '__main__': unittest.main()
|