Digital Drosophila Archive
200 lines
10 KiB
Python
200 lines
10 KiB
Python
"""Synthetic orchestration/producer-contract checks; never launch a brain."""
|
|
import ast
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
from types import ModuleType
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
import numpy as np
|
|
|
|
import compare_timing
|
|
import oracle_timing
|
|
import run_suite
|
|
from shift_audit import transport
|
|
from timing_queue import audit_sequence
|
|
|
|
|
|
HERE = Path(__file__).resolve().parent
|
|
|
|
|
|
def fake_calibration_records():
|
|
return [dict(name=f'cal-{mode}-{odor:02d}-s{seed}', mode=mode, odor_id='odor'+str(odor), seed=seed)
|
|
for mode in ('physiological', 'matched_energy') for odor in range(6)
|
|
for seed in range(51101, 51106)]
|
|
|
|
|
|
def oracle_timing_dict(delay, generated, pending, final_state):
|
|
"""Evaluate only the actual producer's explicit metadata dict, no run call."""
|
|
tree = ast.parse((HERE / 'oracle_timing.py').read_text(encoding='utf-8'))
|
|
calls = [n for n in ast.walk(tree) if isinstance(n, ast.Call) and isinstance(n.func, ast.Name)
|
|
and n.func.id == '_save' and len(n.args) > 1 and isinstance(n.args[1], ast.Dict)
|
|
and any(isinstance(s, ast.Constant) and s.value == 'apl-timing-audit.json'
|
|
for s in ast.walk(n.args[0]))]
|
|
if len(calls) != 1:
|
|
raise AssertionError('Expected exactly one explicit oracle timing-metadata producer')
|
|
steps = len(generated)
|
|
namespace = dict(np=np, min=min, max=max, bool=bool, apl_delay_steps=delay, steps=steps,
|
|
consumed=max(steps-delay, 0), initial_zero_prefix=max(delay-steps, 0), pending_tail=pending,
|
|
apl_history={'state': np.asarray([final_state], np.float32)}, out_dir=Path('.'),
|
|
_array_hash=oracle_timing._array_hash, _digest=lambda _: '0'*64)
|
|
return eval(compile(ast.Expression(calls[0].args[1]), '<only metadata literal>', 'eval'), namespace)
|
|
|
|
|
|
class ProtocolTests(unittest.TestCase):
|
|
def test_two_independent_queues_match_both_arms_and_short_windows(self):
|
|
for delay in (0, 18):
|
|
for steps in (0, 1, 17, 18, 19, 37):
|
|
with self.subTest(delay=delay, steps=steps):
|
|
generated = ((np.arange(steps * 4064).reshape(steps, 4064) % 31)
|
|
.astype(np.float32) * np.float32(.001))
|
|
if steps:
|
|
generated[0, 0] = np.float32(-0.)
|
|
applied, pending, report = transport(generated, delay)
|
|
independent = oracle_timing._APLDelayFIFO(delay)
|
|
expected = np.empty_like(generated)
|
|
for tick, row in enumerate(generated):
|
|
expected[tick] = independent.step(tick, row)
|
|
tail = independent.finish(steps)
|
|
np.testing.assert_array_equal(applied.view(np.uint32), expected.view(np.uint32))
|
|
np.testing.assert_array_equal(pending.view(np.uint32), tail.view(np.uint32))
|
|
self.assertTrue(report['pass_all'])
|
|
|
|
def test_queue_rejects_off_by_one_and_lost_pending(self):
|
|
values = np.ones((19, 2), np.float32)
|
|
applied, pending, _ = transport(values, 18)
|
|
corrupt = applied.copy()
|
|
corrupt[17] = 1
|
|
with self.assertRaises(ValueError):
|
|
audit_sequence(values, corrupt, pending, 18)
|
|
with self.assertRaises(ValueError):
|
|
audit_sequence(values, applied, pending[:-1], 18)
|
|
|
|
def test_actual_oracle_metadata_matches_compare_contract(self):
|
|
for delay in (0, 18):
|
|
for steps in (1, 17, 18, 19, 37):
|
|
with self.subTest(delay=delay, steps=steps):
|
|
generated = np.full((steps, 4064), .25, np.float32)
|
|
_, pending, _ = transport(generated, delay)
|
|
final = np.array([.125, -0.], np.float32)
|
|
timing = oracle_timing_dict(delay, generated, pending, final)
|
|
summary = dict(pending_generated_count=min(delay, steps),
|
|
pending_initial_zero_prefix=max(delay-steps, 0), pending_tail_shape=[delay, 4064],
|
|
future_generated_decay_tail_included=False, APL_worker_ticks=steps)
|
|
run = dict(timing=timing, summary=summary, pending=pending,
|
|
apl={'state': np.asarray([final], np.float32)})
|
|
checks = {}
|
|
compare_timing._timing_metadata_checks(run, 'cpu', delay, steps, checks)
|
|
self.assertTrue(all(checks.values()), checks)
|
|
|
|
def test_bad_metadata_must_fail_without_fitted_tolerance(self):
|
|
generated = np.ones((3, 4064), np.float32)
|
|
_, pending, _ = transport(generated, 18)
|
|
final = np.array([.125, 0.], np.float32)
|
|
timing = oracle_timing_dict(18, generated, pending, final)
|
|
timing['pending_initial_zero_prefix'] += 1
|
|
run = dict(timing=timing, summary={}, pending=pending, apl={'state': np.asarray([final])})
|
|
checks = {}
|
|
compare_timing._timing_metadata_checks(run, 'cpu', 18, 3, checks)
|
|
self.assertFalse(checks['cpu.timing_integer_metadata'])
|
|
|
|
def _suite(self, stage, *, failed_read=None, execution=True, representation=False,
|
|
control=True, repeat=True):
|
|
records = fake_calibration_records()
|
|
saved, children = {}, []
|
|
def read(path):
|
|
path = str(path).replace('\\', '/')
|
|
if path.endswith('calibration-cases.json'):
|
|
return records
|
|
return {'pass_all': not (failed_read and path.endswith(failed_read))}
|
|
fake = ModuleType('calibration')
|
|
fake.summarize = lambda: dict(execution_gate_pass=execution, pass_all=representation)
|
|
fake.control_identity = lambda *args: dict(pass_all=control)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
with patch.object(run_suite, 'EXP', Path(directory)), patch.object(run_suite, 'check'), \
|
|
patch.object(run_suite, 'budget'), patch.object(run_suite, 'read', side_effect=read), \
|
|
patch.object(run_suite, 'save', side_effect=lambda path, result: saved.update({Path(path).name: result})), \
|
|
patch.object(run_suite, 'child', side_effect=lambda *args: children.append(args)), \
|
|
patch.object(run_suite, 'gate_case', side_effect=lambda arm, name, repeats:
|
|
dict(arm=arm, case=name, pass_all=True, gpu_repeats=repeats)), \
|
|
patch.object(compare_timing, 'compare_gpu_repeats', return_value=dict(pass_all=repeat)), \
|
|
patch.dict(sys.modules, {'calibration': fake}), patch.object(sys, 'argv', ['run_suite.py', stage]):
|
|
error = None
|
|
try:
|
|
run_suite.main()
|
|
except AssertionError as caught:
|
|
error = caught
|
|
return saved, children, error
|
|
|
|
def test_fixture_gate_covers_both_arms_all_cases(self):
|
|
saved, children, error = self._suite('fixtures')
|
|
self.assertIsNone(error)
|
|
self.assertEqual(children, [])
|
|
rows = saved['fixtures-gate.json']['cases']
|
|
self.assertEqual(len(rows), 16)
|
|
self.assertEqual({r['arm'] for r in rows}, {'delay0', 'delay18'})
|
|
self.assertTrue(all(r['gpu_repeats'] == 3 for r in rows))
|
|
|
|
def test_probe_stage_cannot_bypass_fixture_failure(self):
|
|
saved, children, error = self._suite('probes', failed_read='fixtures-gate.json')
|
|
self.assertIsNotNone(error)
|
|
self.assertEqual(children, [])
|
|
self.assertNotIn('probes-gate.json', saved)
|
|
|
|
def test_CAL_cannot_bypass_probe_failure(self):
|
|
saved, children, error = self._suite('calibration', failed_read='probes-gate.json')
|
|
self.assertIsNotNone(error)
|
|
self.assertEqual(children, [])
|
|
self.assertNotIn('calibration-gate.json', saved)
|
|
|
|
def test_representation_failure_does_not_select_or_stop_an_arm(self):
|
|
saved, children, error = self._suite('calibration', execution=True, representation=False)
|
|
self.assertIsNone(error)
|
|
self.assertEqual(len(children), 140) # (60 - 2 already-probed + 12 repeat) per arm.
|
|
self.assertEqual({row[1] for row in children}, {'delay0', 'delay18'})
|
|
self.assertTrue(saved['calibration-gate.json']['pass_all'])
|
|
self.assertFalse(saved['calibration-gate.json']['representation_pass_all'])
|
|
for arm in ('delay0', 'delay18'):
|
|
self.assertEqual(sum(row[1] == arm and len(row) == 4 and row[3] == 1 for row in children), 12)
|
|
|
|
def test_CAL_numerical_failure_stops_later_execution(self):
|
|
saved, _, error = self._suite('calibration', execution=False, representation=False)
|
|
self.assertIsNotNone(error)
|
|
self.assertFalse(saved['calibration-gate.json']['pass_all'])
|
|
self.assertTrue(saved['stopped-calibration.json']['later_stages_not_permitted'])
|
|
|
|
def test_control_divergence_stops_before_next_trial_or_other_arm(self):
|
|
saved, children, error = self._suite('calibration', control=False)
|
|
self.assertIsNotNone(error)
|
|
# The first r0 is a completed fullgraph probe; do not even run its r1.
|
|
self.assertEqual(children, [])
|
|
self.assertNotIn('calibration-gate.json', saved)
|
|
self.assertTrue(saved['stopped-calibration.json']['later_stages_not_permitted'])
|
|
|
|
def test_repeat_divergence_stops_before_next_trial_or_other_arm(self):
|
|
saved, children, error = self._suite('calibration', repeat=False)
|
|
self.assertIsNotNone(error)
|
|
self.assertEqual(len(children), 1)
|
|
self.assertEqual(children[0][1], 'delay0')
|
|
self.assertEqual(children[0][-1], 1)
|
|
self.assertNotIn('calibration-gate.json', saved)
|
|
self.assertTrue(saved['stopped-calibration.json']['later_stages_not_permitted'])
|
|
|
|
def test_origin_failure_prevents_any_new_fixture_run(self):
|
|
saved, children, error = self._suite('fixtures', failed_read='origin-audit/summary.json')
|
|
self.assertIsNotNone(error)
|
|
self.assertEqual(saved, {})
|
|
self.assertEqual(children, [])
|
|
|
|
def test_shift_failure_prevents_any_new_fixture_run(self):
|
|
saved, children, error = self._suite('fixtures', failed_read='shift-gate.json')
|
|
self.assertIsNotNone(error)
|
|
self.assertEqual(saved, {})
|
|
self.assertEqual(children, [])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|