Digital Drosophila Archive
57 lines
2.9 KiB
Python
57 lines
2.9 KiB
Python
"""Synthetic interface and fail-closed schedule checks, no scientific records."""
|
|
import ast
|
|
from pathlib import Path
|
|
import unittest
|
|
import numpy as np
|
|
from run_audit import compare_results
|
|
from summarize import metrics, primary_rows
|
|
|
|
|
|
class IntegrationTests(unittest.TestCase):
|
|
def test_per_neuron_count_difference_not_cancelled_by_equal_total(self):
|
|
a = dict(step=np.array([0, 1], np.int32), neuron_idx=np.array([1, 1], np.int32),
|
|
counts=np.array([2, 0], np.int32), first_spike=np.array([0, -1], np.int32),
|
|
v=np.array([-.052, -.052], np.float64), g=np.zeros(2, np.float64))
|
|
b = {k: v.copy() for k, v in a.items()}
|
|
b['neuron_idx'][:] = 2; b['counts'][:] = [0, 2]; b['first_spike'][:] = [-1, 0]
|
|
h = dict(history_v=np.zeros((2, 2)), history_g=np.zeros((2, 2)))
|
|
result = compare_results(a, b, h, h)
|
|
self.assertEqual(result['spike_count_L1'], 4)
|
|
self.assertEqual(result['neurons_with_count_change'], 2)
|
|
self.assertTrue(result['functional_count_or_first_latency_change'])
|
|
|
|
def test_event_difference_with_same_counts_and_first_is_retained(self):
|
|
a = dict(step=np.array([0, 3], np.int32), neuron_idx=np.array([1, 1], np.int32),
|
|
counts=np.array([2], np.int32), first_spike=np.array([0], np.int32),
|
|
v=np.zeros(1), g=np.zeros(1))
|
|
b = {k: v.copy() for k, v in a.items()}; b['step'][1] = 4
|
|
h = dict(history_v=np.zeros((5, 1)), history_g=np.zeros((5, 1)))
|
|
result = compare_results(a, b, h, h)
|
|
self.assertTrue(result['event_sequence_changed'])
|
|
self.assertFalse(result['functional_count_or_first_latency_change'])
|
|
|
|
def test_undefined_denominator_is_not_silently_zero(self):
|
|
self.assertIsNone(metrics([])['mean'])
|
|
|
|
def test_repeat_not_an_independent_primary_trial(self):
|
|
rows = [dict(name=str(i), repeat=0) for i in range(120)] + [dict(name=str(i), repeat=1) for i in range(24)]
|
|
self.assertEqual(len(primary_rows(dict(planned=rows))), 120)
|
|
rows[1]['name'] = '0'
|
|
with self.assertRaises(AssertionError):
|
|
primary_rows(dict(planned=rows))
|
|
|
|
def test_all_baseline_gate_precedes_conditional_schedule(self):
|
|
tree = ast.parse((Path(__file__).parent / 'run_audit.py').read_text())
|
|
main = next(n for n in tree.body if isinstance(n, ast.FunctionDef) and n.name == 'main')
|
|
loops = [n for n in ast.walk(main) if isinstance(n, ast.For) and isinstance(n.target, ast.Name) and n.target.id == 'stage']
|
|
self.assertEqual(len(loops), 1)
|
|
self.assertEqual(ast.literal_eval(loops[0].iter), ('baseline', 'conditional25'))
|
|
audit = next(n for n in tree.body if isinstance(n, ast.FunctionDef) and n.name == 'audit_trial')
|
|
source = ast.unparse(audit)
|
|
self.assertIn("baselines-complete.json", source)
|
|
self.assertLess(source.index("baselines-complete.json"), source.index('core.replay'))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|