Digital Drosophila Archive
110 lines
5.3 KiB
Python
110 lines
5.3 KiB
Python
"""CPU-only coordinator contract tests; no real brain/input sampling."""
|
|
import ast
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
import numpy as np
|
|
|
|
import conditioning_preflight_prepare as P
|
|
import conditioning_preflight_workflow as W
|
|
|
|
|
|
class CoordinatorTests(unittest.TestCase):
|
|
def test_npz_exclusive_roundtrip(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
path = Path(d) / 'x.npz'
|
|
a = np.asarray([0, 3], np.int32)
|
|
P.save_npz(path, tick=a)
|
|
self.assertEqual(W.npz(path)['tick'].tobytes(), a.tobytes())
|
|
with self.assertRaises(FileExistsError): P.save_npz(path, tick=a)
|
|
|
|
def test_exact_pins_and_mutation(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
p = Path(d) / 'input'
|
|
p.write_bytes(b'original')
|
|
pins = {str(p): P.digest(p)}
|
|
P.verify_pins(pins)
|
|
p.write_bytes(b'changed')
|
|
with self.assertRaisesRegex(ValueError, 'Pinned file'): P.verify_pins(pins)
|
|
|
|
def test_rows_exact_unique(self):
|
|
row = dict(row_id='D1--reference--epoch-000')
|
|
self.assertEqual(W.row_by({'plan': {'calls': [row]}}, row['row_id']), row)
|
|
for rows in ([], [row, row]):
|
|
with self.assertRaises(ValueError): W.row_by({'plan': {'calls': rows}}, row['row_id'])
|
|
|
|
def test_locations_reject_path_traversal(self):
|
|
for value in ('../x', '/x', 'a/b', 'a\\b', '', None):
|
|
with self.assertRaises(ValueError): W.locations(value)
|
|
|
|
def test_plan_descriptor_avoids_huge_payload_duplication(self):
|
|
value = dict(full_trial_plan={'large': [1]}, payload_sha256='x', calls=[{'row_id': 'a'}])
|
|
self.assertEqual(P.plan_descriptor(value), {'calls': [{'row_id': 'a'}]})
|
|
self.assertIn('full_trial_plan', value)
|
|
|
|
def test_original_AST_rejects_unanchored_bytes(self):
|
|
with self.assertRaisesRegex(ValueError, 'L source changed'):
|
|
W.original_sampler_nodes(b'def run_rates(): pass')
|
|
|
|
def test_original_AST_if_local_frozen_source_available(self):
|
|
# Both repository checkout and local source mirror are supported; this
|
|
# test never reads historical dynamics or executes the extracted code.
|
|
candidates = [Path(__file__).resolve().parents[1] / 'phase4l.py',
|
|
Path(__file__).resolve().parents[1] / 'models/male-cns-lif/phase4l.py']
|
|
source = next((p for p in candidates if p.is_file()), None)
|
|
self.assertIsNotNone(source, 'Original L source required for actual AST extraction test')
|
|
nodes = W.original_sampler_nodes(source.read_bytes())
|
|
self.assertEqual(len(nodes), 4)
|
|
self.assertNotIn('reset', ast.unparse(ast.Module(body=nodes, type_ignores=[])))
|
|
|
|
def test_registration_external_hash_rejection_before_parse(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
exp = Path(d); (exp / 'protocol').mkdir()
|
|
P.save(exp / 'protocol/preregistered.json', {'fake': True})
|
|
with patch.object(P, 'EXP', exp):
|
|
with self.assertRaisesRegex(ValueError, 'external anchor'):
|
|
P.read_registration('0' * 64)
|
|
|
|
def test_initial_overlay_has_no_parent_process_lookup(self):
|
|
expected = dict(overlay_file='/immutable/zero.npz', overlay_manifest='/immutable/zero.json')
|
|
self.assertIs(W.before_overlay({'initial_overlay': expected}, 'unused', {'previous_delta_row_id': None}), expected)
|
|
|
|
def test_wrong_parent_arm_rejected_before_runtime_read(self):
|
|
parent = dict(row_id='parent', arm='D0', diagnostic_mode='light', input_epoch_index=0)
|
|
child = dict(previous_delta_row_id='parent', arm='D1', diagnostic_mode='light', input_epoch_index=1)
|
|
with self.assertRaisesRegex(ValueError, 'cross-epoch'):
|
|
W.before_overlay({'plan': {'calls': [parent]}}, 'unused', child)
|
|
|
|
def test_budget_refuses_overrun_without_delete(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
exp = Path(d); p = exp / 'keep'
|
|
p.write_bytes(b'1234')
|
|
with patch.object(W, 'EXP', exp):
|
|
with self.assertRaisesRegex(ValueError, 'Resource cap'):
|
|
W.budget({'budget': {'output_budget_bytes': 1, 'minimum_free_reserve_bytes': 1}})
|
|
self.assertEqual(p.read_bytes(), b'1234')
|
|
|
|
def test_partial_log_is_not_retried(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
exp = Path(d); (exp / 'logs').mkdir()
|
|
(exp / 'logs/input-generation.stdout.log').write_bytes(b'partial')
|
|
with patch.object(W, 'EXP', exp), patch.object(W.subprocess, 'run') as run:
|
|
with self.assertRaisesRegex(ValueError, 'never retry'): W.child_process({}, 'sha', 'materialize')
|
|
run.assert_not_called()
|
|
|
|
def test_admission_blocks_stopped_complete_or_sealed(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
exp = Path(d); (exp / 'audit').mkdir()
|
|
with patch.object(W, 'EXP', exp):
|
|
W.admission()
|
|
for name in ('audit/stopped.json', 'audit/execution-complete.json', 'manifest.json'):
|
|
p = exp / name; p.write_bytes(b'{}')
|
|
with self.assertRaisesRegex(ValueError, 'Stopped/completed'): W.admission()
|
|
p.unlink()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|