Digital Drosophila Archive
208 lines
12 KiB
Python
208 lines
12 KiB
Python
import copy
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import types
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
import numpy as np
|
|
|
|
import engineering_workflow as w
|
|
from test_engineering_inputs import spec
|
|
|
|
|
|
def template(row):
|
|
return dict(schema='phase4an-engineering-case-template-v1', row_id=row['row_id'], original_spec=spec(),
|
|
start=row['start'], stop=row['stop'], budget_root='/project/new', output_budget_bytes=8 * 1024**3)
|
|
|
|
|
|
class Workflow(unittest.TestCase):
|
|
def test_only_exact_plan_row_and_safe_output_identity(self):
|
|
reg = {'plan': w.PLAN.build_plan()}
|
|
row = reg['plan']['rows'][0]
|
|
self.assertEqual(w.row_from(reg, row['row_id']), row)
|
|
for bad in ('unknown', '../escape', '/absolute'):
|
|
with self.assertRaises(ValueError): w.row_from(reg, bad)
|
|
for bad in ('../escape', '/absolute', 'a/b', 'a\\b', '', 'a:b'):
|
|
with self.assertRaises(ValueError): w.locations(bad)
|
|
reg['plan']['rows'][0]['steps'] = True
|
|
with self.assertRaises(ValueError): w.row_from(reg, row['row_id'])
|
|
|
|
def test_real_parent_is_required_only_for_later_segments(self):
|
|
for row in w.PLAN.build_plan()['rows']:
|
|
before = template(row)
|
|
pin = None if row['parent_row_id'] is None else 'a' * 64
|
|
derived, audit = w.derive_case(before, row, pin)
|
|
self.assertEqual(derived['steps'], row['steps'])
|
|
self.assertEqual(before, template(row))
|
|
self.assertEqual(audit['input_interval'], [row['start'], row['stop']])
|
|
with self.assertRaises(ValueError):
|
|
w.derive_case(before, row, 'a' * 64 if pin is None else None)
|
|
|
|
def test_forged_template_and_boolean_interval_rejected(self):
|
|
row = w.PLAN.build_plan()['rows'][0]
|
|
for key, value in (('schema', 'wrong'), ('row_id', 'wrong'), ('start', False), ('stop', 18)):
|
|
changed = template(row); changed[key] = value
|
|
with self.subTest(key=key), self.assertRaises(ValueError): w.derive_case(changed, row, None)
|
|
changed = template(row); changed['extra'] = True
|
|
with self.assertRaises(ValueError): w.derive_case(changed, row, None)
|
|
|
|
def test_json_comparison_does_not_conflate_booleans_and_numbers(self):
|
|
self.assertFalse(w.same_json({'n': True}, {'n': 1}))
|
|
self.assertFalse(w.same_json({'n': 1.}, {'n': 1}))
|
|
with self.assertRaises(ValueError): w.canonical({'n': float('nan')})
|
|
|
|
def test_actual_precision_flags_are_checked_not_merely_recorded(self):
|
|
component = dict(cuda_matmul_allow_tf32=False, cudnn_allow_tf32=True, deterministic_algorithms=False,
|
|
cudnn_deterministic=False, cudnn_benchmark=False, float32_matmul_precision='highest',
|
|
CUBLAS_WORKSPACE_CONFIG=None, device_name='NVIDIA GeForce RTX 4090', torch='2.test', cuda='13.0',
|
|
numpy='2.test', precision_configuration_changed_by_component=False)
|
|
recorded = dict(gpu_uuid=w.GPU, settings_changed_by_environment_recorder=False,
|
|
matmul_allow_tf32=False, cudnn_allow_tf32=True, deterministic_algorithms=False,
|
|
cudnn_deterministic=False, cudnn_benchmark=False, float32_matmul_precision='highest',
|
|
CUBLAS_WORKSPACE_CONFIG=None, device_name='NVIDIA GeForce RTX 4090', torch_version='2.test',
|
|
torch_cuda_version='13.0', numpy_version='2.test')
|
|
w.verify_environment(recorded, component)
|
|
for key, value in [('cuda_matmul_allow_tf32', True), ('float32_matmul_precision', 'high'),
|
|
('precision_configuration_changed_by_component', True), ('cudnn_allow_tf32', False)]:
|
|
altered = dict(component, **{key: value})
|
|
with self.subTest(key=key), self.assertRaises(ValueError): w.verify_environment(recorded, altered)
|
|
with self.assertRaises(ValueError): w.verify_environment(dict(recorded, matmul_allow_tf32=0), component)
|
|
|
|
def test_npz_real_compression_measured_not_assumed(self):
|
|
with tempfile.TemporaryDirectory() as name:
|
|
root = Path(name)
|
|
with (root / 'compressed.npz').open('xb') as stream:
|
|
np.savez_compressed(stream, v=np.zeros(50, np.float64))
|
|
measured = w.npz_storage(root / 'compressed.npz')
|
|
self.assertGreater(measured['uncompressed_entry_bytes'], measured['compressed_entry_bytes'])
|
|
self.assertEqual(measured['entries'][0]['compression'], 8)
|
|
with (root / 'stored.npz').open('xb') as stream:
|
|
np.savez(stream, v=np.zeros(50, np.float64))
|
|
with self.assertRaises(ValueError): w.npz_storage(root / 'stored.npz')
|
|
|
|
def test_resource_limit_preserves_existing_payload(self):
|
|
with tempfile.TemporaryDirectory() as name, patch.object(w, 'EXP', Path(name)):
|
|
path = Path(name) / 'payload.bin'
|
|
path.write_bytes(b'not deleted')
|
|
reg = {'budget': {'output_budget_bytes': 1, 'minimum_free_reserve_bytes': 1}}
|
|
with self.assertRaises(ValueError): w.disk_budget(reg)
|
|
self.assertEqual(path.read_bytes(), b'not deleted')
|
|
|
|
def _temporary_gate(self, root):
|
|
for name in ('audit', 'logs', 'receipts', 'runs', 'cases'):
|
|
(root / name).mkdir()
|
|
reg = {'plan': w.PLAN.build_plan(), 'budget': {'output_budget_bytes': 1 << 20,
|
|
'minimum_free_reserve_bytes': 1}}
|
|
return reg, types.SimpleNamespace(read_registration=lambda sha: copy.deepcopy(reg))
|
|
|
|
def test_controller_refuses_partial_without_rerun_or_deletion(self):
|
|
with tempfile.TemporaryDirectory() as name, patch.object(w, 'EXP', Path(name)):
|
|
root = Path(name); reg, prep = self._temporary_gate(root)
|
|
paths = w.locations(reg['plan']['rows'][0]['row_id'])
|
|
paths['stdout'].write_bytes(b'original evidence')
|
|
with (patch.object(w, 'preparation', return_value=prep), patch.object(w, 'preflight') as pre,
|
|
patch.object(w.subprocess, 'run') as runner):
|
|
with self.assertRaisesRegex(ValueError, 'Partial row'):
|
|
w.run_registered('a' * 64)
|
|
runner.assert_not_called(); pre.assert_not_called()
|
|
self.assertEqual(paths['stdout'].read_bytes(), b'original evidence')
|
|
self.assertFalse(paths['execution'].exists())
|
|
self.assertTrue((root / 'audit/stopped.json').exists())
|
|
|
|
def test_nonzero_worker_exit_cannot_become_completed_receipt(self):
|
|
with tempfile.TemporaryDirectory() as name, patch.object(w, 'EXP', Path(name)):
|
|
root = Path(name); reg, prep = self._temporary_gate(root)
|
|
paths = w.locations(reg['plan']['rows'][0]['row_id'])
|
|
failure = types.SimpleNamespace(returncode=1)
|
|
with (patch.object(w, 'preparation', return_value=prep), patch.object(w, 'preflight'),
|
|
patch.object(w.subprocess, 'run', return_value=failure) as runner):
|
|
with self.assertRaisesRegex(ValueError, 'Worker failed'):
|
|
w.run_registered('a' * 64)
|
|
self.assertEqual(runner.call_count, 1)
|
|
self.assertTrue(paths['stdout'].exists() and paths['stderr'].exists())
|
|
self.assertFalse(paths['execution'].exists())
|
|
self.assertTrue((root / 'audit/stopped.json').exists())
|
|
|
|
def test_stopped_inventory_cannot_auto_resume(self):
|
|
with tempfile.TemporaryDirectory() as name, patch.object(w, 'EXP', Path(name)):
|
|
root = Path(name); _, prep = self._temporary_gate(root)
|
|
(root / 'audit/stopped.json').write_text('{}')
|
|
with patch.object(w, 'preparation', return_value=prep), patch.object(w.subprocess, 'run') as runner:
|
|
with self.assertRaisesRegex(ValueError, 'Stopped/completed'):
|
|
w.run_registered('a' * 64)
|
|
runner.assert_not_called()
|
|
|
|
def _completion(self, reg, root):
|
|
entries = []
|
|
for row in reg['plan']['rows']:
|
|
path = w.locations(row['row_id'])['execution']
|
|
path.write_bytes(row['row_id'].encode())
|
|
entries.append(dict(row_id=row['row_id'], execution_sha256=w.R.digest(path), reverified_existing=False))
|
|
return dict(complete=True, registration_sha256='a' * 64, expected_calls=32, executed_ticks=11074,
|
|
training=False, learning_pass_claimed=False, completed=entries)
|
|
|
|
def test_inventory_rechecks_every_actual_receipt(self):
|
|
with tempfile.TemporaryDirectory() as name, patch.object(w, 'EXP', Path(name)):
|
|
root = Path(name); reg, _ = self._temporary_gate(root)
|
|
complete = self._completion(reg, root)
|
|
w.R._save(root / 'audit/execution-complete.json', complete)
|
|
with patch.object(w, 'verify_execution', side_effect=lambda r, s, row: dict(row=row)) as verify:
|
|
records = w.verified_inventory(reg, 'a' * 64)
|
|
self.assertEqual(verify.call_count, 32)
|
|
self.assertEqual(list(records), [row['row_id'] for row in reg['plan']['rows']])
|
|
|
|
def test_inventory_rejects_reorder_missing_and_changed_receipt(self):
|
|
for corruption in ('reorder', 'missing', 'changed', 'claimed_training', 'wrong_ticks'):
|
|
with self.subTest(corruption=corruption), tempfile.TemporaryDirectory() as name, patch.object(w, 'EXP', Path(name)):
|
|
root = Path(name); reg, _ = self._temporary_gate(root)
|
|
complete = self._completion(reg, root)
|
|
if corruption == 'reorder': complete['completed'].reverse()
|
|
if corruption == 'missing': complete['completed'].pop()
|
|
if corruption == 'changed': w.locations(reg['plan']['rows'][0]['row_id'])['execution'].write_bytes(b'changed')
|
|
if corruption == 'claimed_training': complete['training'] = True
|
|
if corruption == 'wrong_ticks': complete['executed_ticks'] += 1
|
|
w.R._save(root / 'audit/execution-complete.json', complete)
|
|
with patch.object(w, 'verify_execution'), self.assertRaises(ValueError):
|
|
w.verified_inventory(reg, 'a' * 64)
|
|
|
|
def test_seal_contains_nested_manifests_and_self_checksum(self):
|
|
with tempfile.TemporaryDirectory() as name, patch.object(w, 'EXP', Path(name)):
|
|
root = Path(name)
|
|
(root / 'runs' / 'one').mkdir(parents=True)
|
|
(root / 'runs/one/manifest.json').write_bytes(b'nested evidence')
|
|
(root / 'payload').write_bytes(b'payload')
|
|
result = w.seal_experiment()
|
|
manifest = json.loads((root / 'manifest.json').read_text())
|
|
self.assertEqual([row['path'] for row in manifest['files']], ['payload', 'runs/one/manifest.json'])
|
|
lines = (root / 'checksums.sha256').read_text().splitlines()
|
|
self.assertIn(result['manifest_sha256'] + ' manifest.json', lines)
|
|
self.assertEqual(len(lines), 3)
|
|
self.assertEqual(result['checksums_sha256'], hashlib.sha256((root / 'checksums.sha256').read_bytes()).hexdigest())
|
|
before = (root / 'manifest.json').read_bytes()
|
|
with self.assertRaises(ValueError): w.seal_experiment()
|
|
self.assertEqual((root / 'manifest.json').read_bytes(), before)
|
|
|
|
def test_seal_rejects_bytecode_instead_of_silently_losing_it(self):
|
|
with tempfile.TemporaryDirectory() as name, patch.object(w, 'EXP', Path(name)):
|
|
root = Path(name); (root / '__pycache__').mkdir()
|
|
(root / '__pycache__/a.pyc').write_bytes(b'cache')
|
|
with self.assertRaises(ValueError): w.seal_experiment()
|
|
self.assertFalse((root / 'manifest.json').exists())
|
|
|
|
def test_finalize_rejects_recomputed_difference_before_sealing(self):
|
|
with tempfile.TemporaryDirectory() as name, patch.object(w, 'EXP', Path(name)):
|
|
root = Path(name); _, prep = self._temporary_gate(root)
|
|
w.R._save(root / 'audit/engineering-analysis.json', dict(status='saved'))
|
|
with patch.object(w, 'evaluate', return_value=({}, {'status': 'different'})), patch.object(w, 'seal_experiment') as seal:
|
|
with self.assertRaisesRegex(ValueError, 'Recomputed analysis'):
|
|
w.finalize('a' * 64)
|
|
seal.assert_not_called()
|
|
self.assertFalse((root / 'audit/final-status.json').exists())
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|