"""Synthetic CPU-only contract tests; no real trajectories, GPU, or file writes.""" import sys import unittest import numpy as np from replay_core import IncomingIndex, replay, state_step def coefficients(**changes): result = dict(gd=1., c0=0., cg=0., cv=1., threshold=1., reset=0., seed_volt=2.) result.update(changes) return result def record(rows=(), key="tick"): values = np.array(rows, np.int32).reshape(-1, 2) return {key: values[:, 0], "neuron_idx": values[:, 1]} def empty_index(count=1): return IncomingIndex(np.empty(0, np.int32), np.empty(0, np.int32), np.empty(0, np.int64), np.empty(0, np.float64), count) def one_edge(weight=2.): return IncomingIndex(np.array([10], np.int32), np.array([0], np.int32), np.array([81], np.int64), np.array([weight], np.float64), 1) def shadow(steps=25, *, initial_v=None, initial_g=None, coeff=None, requests=None, events=None, incoming=None, h32=None, callback=None, targets=None): targets = np.array([100], np.int32) if targets is None else targets n = len(targets) return replay(np.zeros(n, np.float64) if initial_v is None else initial_v, np.zeros(n, np.float64) if initial_g is None else initial_g, coefficients() if coeff is None else coeff, steps, targets, record() if requests is None else requests, record() if events is None else events, empty_index(n) if incoming is None else incoming, np.zeros((steps, n), np.float32) if h32 is None else h32, callback) class StepTests(unittest.TestCase): def test_strict_threshold_and_positive_zero_reset(self): v = np.array([1., np.nextafter(1., np.inf)], np.float64) result = state_step(v, np.array([3., -3.]), np.zeros(2, np.int32), np.zeros(2, bool), coefficients()) np.testing.assert_array_equal(result["fired"], [False, True]) np.testing.assert_array_equal(result["ref"], [0, 22]) self.assertEqual(result["g"].view(np.uint64)[1], 0) self.assertEqual(result["v_threshold"][1], v[1]) def test_voltage_uses_old_g_and_separate_expression_order(self): c = coefficients(gd=.4, cg=.2, cv=.3, c0=.1, threshold=100., seed_volt=101.) result = state_step(np.array([3.]), np.array([2.]), np.array([0], np.int32), np.array([False]), c) expected = np.float64(c["c0"]) + ((np.float64(c["cg"]) * np.float64(2.)) + (np.float64(c["cv"]) * np.float64(3.))) self.assertEqual(result["v"][0].hex(), expected.hex()) self.assertEqual(result["g"][0].hex(), (np.float64(.4) * np.float64(2.)).hex()) self.assertNotEqual(result["v"][0], .1 + (.2 * .8 + .3 * 3.)) def test_negative_g_and_voltage_freeze_when_refractory(self): result = state_step(np.array([-.052]), np.array([-.123]), np.array([22], np.int32), np.array([True]), coefficients(gd=.5, cg=1., c0=.3)) self.assertEqual(result["v"][0], -.052) self.assertEqual(result["g"][0], -.123) self.assertEqual(result["ref"][0], 21) self.assertFalse(result["fired"][0]) self.assertTrue(result["request_rejected"][0]) def test_ref_one_releases_before_request_and_reset(self): result = state_step(np.array([-.052]), np.array([-.123]), np.array([1], np.int32), np.array([True]), coefficients()) self.assertTrue(result["available"][0]) self.assertTrue(result["request_accepted"][0]) self.assertEqual(result["v_threshold"][0], 2.) self.assertTrue(result["fired"][0]) self.assertEqual(result["ref"][0], 22) self.assertEqual(result["g"][0], 0.) def test_state_inputs_unchanged(self): v, g, ref, request = np.array([2.]), np.array([5.]), np.array([0], np.int32), np.array([False]) inputs = [x.copy() for x in (v, g, ref, request)] state_step(v, g, ref, request, coefficients()) for actual, expected in zip((v, g, ref, request), inputs): np.testing.assert_array_equal(actual, expected) class ReplayTests(unittest.TestCase): def test_logged_target_spikes_are_not_forced(self): result = shadow(events=record([(0, 100), (22, 100)])) self.assertEqual(len(result["step"]), 0) np.testing.assert_array_equal(result["first_spike"], [-1]) def test_predicted_events_exist_without_logged_target_spikes(self): result = shadow(requests=record([(0, 100), (1, 100), (21, 100), (22, 100), (23, 100)])) np.testing.assert_array_equal(result["step"], [0, 22]) np.testing.assert_array_equal(result["neuron_idx"], [100, 100]) np.testing.assert_array_equal(result["counts"], [2]) np.testing.assert_array_equal(result["first_spike"], [0]) def test_anatomical_arrival_is_tick18_then_voltage_tick19(self): history = [] shadow(steps=20, coeff=coefficients(cg=1., threshold=100., seed_volt=101.), events=record([(0, 10)]), incoming=one_edge(), callback=lambda t, s: history.append(s)) self.assertEqual(history[17]["g"][0], 0.) self.assertEqual(history[18]["g"][0], 2.) self.assertEqual(history[18]["v"][0], 0.) self.assertEqual(history[19]["v"][0], 2.) np.testing.assert_array_equal(history[18]["due_edge_id"], [81]) def test_arrival_rejected_on_same_tick_target_fire(self): history = [] result = shadow(steps=19, events=record([(0, 10)]), incoming=one_edge(), requests=record([(18, 100)]), callback=lambda t, s: history.append(s)) np.testing.assert_array_equal(result["step"], [18]) self.assertFalse(history[18]["arrival_accepted"][0]) self.assertEqual(history[18]["g_pre_apl"][0], 0.) def test_refractory_rejects18_and_release22_accepts(self): history = [] shadow(steps=23, events=record([(0, 10), (4, 10)]), incoming=one_edge(), requests=record([(0, 100)]), callback=lambda t, s: history.append(s)) self.assertFalse(history[18]["arrival_accepted"][0]) self.assertTrue(history[22]["arrival_accepted"][0]) self.assertEqual(history[22]["ref"][0], 0) self.assertEqual(history[22]["g_pre_apl"][0], 2.) def test_apl_unconditional_during_freeze_and_release_decay(self): history = [] h = np.ones((23, 1), np.float32) shadow(steps=23, coeff=coefficients(gd=.5), requests=record([(0, 100)]), h32=h, callback=lambda t, s: history.append(s)) self.assertEqual(history[0]["g_after_reset"][0], 0.) self.assertEqual(history[0]["g"][0], -.001) for tick in range(1, 22): self.assertEqual(history[tick]["g_decayed"][0].hex(), history[tick]["old_g"][0].hex()) self.assertEqual(history[tick]["g"][0].hex(), (history[tick]["old_g"][0] - np.float64(.001)).hex()) self.assertEqual(history[22]["g_decayed"][0].hex(), (.5 * history[22]["old_g"][0]).hex()) def test_si_promotion_order_is_not_early_float32_scaling(self): h = np.array([[.01]], np.float32) result = shadow(steps=1, h32=h) expected = -(np.float64(h[0, 0]) * np.float64(.001)) wrong = -np.float64(h[0, 0] * np.float32(.001)) self.assertEqual(result["g"][0].hex(), expected.hex()) self.assertNotEqual(result["g"][0].hex(), wrong.hex()) def test_arrivals_start_at_existing_g_in_original_edge_order(self): incoming = IncomingIndex(np.array([10, 11], np.int32), np.array([0, 0], np.int32), np.array([2, 81], np.int64), np.array([-1e16, 1.], np.float64), 1) result = shadow(steps=19, initial_g=np.array([1e16]), incoming=incoming, events=record([(0, 10), (0, 11)])) self.assertEqual(result["g"][0], 1.) self.assertNotEqual(result["g"][0], np.float64(1e16) + np.sum(incoming.weights)) def test_zero_weight_edges_still_present_in_due_trace(self): history = [] shadow(steps=19, events=record([(0, 10)]), incoming=one_edge(0.), callback=lambda t, s: history.append(s)) np.testing.assert_array_equal(history[18]["due_edge_id"], [81]) np.testing.assert_array_equal(history[18]["arrival_accepted"], [True]) def test_non_target_requests_ignored_and_step_alias_supported(self): result = shadow(requests=record([(0, 10), (0, 100)], key="step"), events=record(key="step")) np.testing.assert_array_equal(result["step"], [0]) np.testing.assert_array_equal(result["neuron_idx"], [100]) def test_callback_mutation_cannot_change_replay(self): args = dict(steps=23, requests=record([(0, 100)]), h32=np.ones((23, 1), np.float32)) expected = shadow(**args) def mutate(tick, stage): for value in stage.values(): value[...] = 0 actual = shadow(**args, callback=mutate) for name in expected: np.testing.assert_array_equal(actual[name].view(np.uint8), expected[name].view(np.uint8)) def test_callback_can_fail_fast_without_later_ticks(self): visited = [] def stop(tick, stage): visited.append(tick) if tick == 2: raise RuntimeError("intentional mismatch") with self.assertRaisesRegex(RuntimeError, "intentional mismatch"): shadow(callback=stop) self.assertEqual(visited, [0, 1, 2]) def test_no_dense_history_and_exact_output_dtypes(self): result = shadow() self.assertFalse(any(value.ndim > 1 for value in result.values())) for name in ("v", "g"): self.assertEqual(result[name].dtype, np.float64) for name in ("ref", "counts", "step", "neuron_idx", "first_spike", "target_globals"): self.assertEqual(result[name].dtype, np.int32) class ValidationTests(unittest.TestCase): def test_index_copies_inputs_and_reuses_source_lookup(self): src, dst, ids, w = np.array([3, 3, 8], np.int32), np.array([1, 0, 1], np.int32), np.array([8, 20, 99], np.int64), np.array([1., 0., -2.]) index = IncomingIndex(src, dst, ids, w, 2) w[:] = 99 np.testing.assert_array_equal(index.weights, [1., 0., -2.]) np.testing.assert_array_equal(index.due(np.array([3, 7, 8], np.int32)), [0, 1, 2]) self.assertFalse(index.weights.flags.writeable) def test_rejects_bad_index_dtype_range_order_and_nonfinite(self): base = dict(src_global=np.array([10, 11], np.int32), dst_local=np.array([0, 0], np.int32), edge_id=np.array([3, 8], np.int64), weights=np.array([1., 2.]), target_count=1) changes = [dict(src_global=np.array([10, 11], np.int64)), dict(src_global=np.array([11, 10], np.int32)), dict(src_global=np.array([10, 166700], np.int32)), dict(dst_local=np.array([0, 1], np.int32)), dict(edge_id=np.array([3, 3], np.int64)), dict(weights=np.array([1., np.nan])), dict(weights=np.array([1., 2.], np.float32))] for change in changes: with self.subTest(change=change), self.assertRaises(ValueError): IncomingIndex(**(base | change)) def test_rejects_invalid_replay_arrays(self): variants = [dict(initial_v=np.array([0.], np.float32)), dict(initial_g=np.array([np.inf])), dict(targets=np.array([100], np.int64)), dict(targets=np.array([100, 100], np.int32)), dict(h32=np.zeros((25, 1), np.float64)), dict(h32=np.full((25, 1), np.nan, np.float32)), dict(h32=np.full((25, 1), -1, np.float32)), dict(coeff=coefficients(gd=np.float32(.9))), dict(requests=record([(0, 100), (0, 100)])), dict(events=record([(25, 10)])), dict(events=record([(0, 166700)])), dict(steps=0)] for change in variants: with self.subTest(change=change), self.assertRaises(ValueError): shadow(**change) def test_no_scientific_runtime_import(self): self.assertNotIn("torch", sys.modules) self.assertNotIn("brian2", sys.modules) if __name__ == "__main__": unittest.main(verbosity=2)