"""Pure synthetic selection, temporal joining, and streaming observer tests.""" import unittest from types import SimpleNamespace import importlib.util import ast from pathlib import Path import numpy as np import pandas as pd from al_origin_audit import (select_cohort, predecessor_counts, first_ticks, source_labels, ALLNObserver, _window_add, bit_equal) class MemorySink: def __init__(self): self.rows = 0 self.calls = [] def write(self, data): self.rows += len(data['edge_id']) self.calls.append({k: np.asarray(v).copy() for k, v in data.items()}) class OriginTests(unittest.TestCase): def test_main_does_not_shadow_callable_helpers(self): tree = ast.parse((Path(__file__).parent / 'al_origin_audit.py').read_text(encoding='utf-8')) helpers = {node.name for node in tree.body if isinstance(node, (ast.FunctionDef, ast.ClassDef))} main = next(node for node in tree.body if isinstance(node, ast.FunctionDef) and node.name == 'main') assigned = {node.id for node in ast.walk(main) if isinstance(node, ast.Name) and isinstance(node.ctx, ast.Store)} self.assertEqual(helpers & assigned, set()) def test_all_earliest_ties_and_static_design(self): ids = np.arange(5, dtype=np.int32) result = select_cohort(ids, np.array([1, 5, 5, 8, -1], np.int32), np.array([0], np.int32), 20, 30) np.testing.assert_array_equal(result['cohort'], [1, 2]) self.assertEqual(result['matched_stop_exclusive'], 5) self.assertEqual(result['cohort_relation_to_first_KC'], 'before') def test_no_non_design_activity_does_not_choose_never_active(self): result = select_cohort(np.arange(3, dtype=np.int32), np.array([0, -1, -1], np.int32), np.array([0], np.int32), 20, 30) self.assertEqual(len(result['cohort']), 0) self.assertEqual(result['earliest_tick'], -1) def test_same_tick_is_not_before(self): result = select_cohort(np.arange(2, dtype=np.int32), np.array([0, 20], np.int32), np.array([0], np.int32), 20, 30) self.assertFalse(result['cohort_before_first_KC']) self.assertEqual(result['cohort_relation_to_first_KC'], 'same_tick') def test_KC_absent_is_censored(self): result = select_cohort(np.arange(2, dtype=np.int32), np.array([0, 20], np.int32), np.array([0], np.int32), -1, 30) self.assertTrue(result['pre_kc_window_censored']) self.assertEqual(result['pre_kc_stop_exclusive'], 30) self.assertFalse(result['cohort_before_first_KC']) def test_after_KC_retains_cohort_without_selecting_a_different_one(self): result = select_cohort(np.arange(2, dtype=np.int32), np.array([0, 20], np.int32), np.array([0], np.int32), 10, 30) self.assertEqual(result['cohort'].tolist(), [1]) self.assertEqual(result['cohort_relation_to_first_KC'], 'after') self.assertEqual(result['matched_stop_exclusive'], 10) def test_temporal_join_excludes_both_boundaries_and_prior_reset(self): result = predecessor_counts(np.array([3, 5, 6, 9, 10, 11]), np.array([1., 2., -3., 4., 5., 6.]), 10, 5) self.assertEqual(result['all_prior_design_PN_events'], 4) self.assertEqual(result['since_last_reset_design_PN_events'], 2) self.assertEqual(result['prior_reset_excluded_events'], 2) self.assertEqual(result['since_last_reset_positive_weight_SI'], 4.) self.assertEqual(result['since_last_reset_negative_abs_weight_SI'], 3.) def test_zero_weight_not_transmission(self): result = predecessor_counts(np.array([2]), np.array([0.]), 3, -1) self.assertEqual(result['since_last_reset_design_PN_events'], 1) self.assertEqual(result['since_last_reset_nonzero_design_PN_events'], 0) def test_no_direct_antecedent_valid_empty(self): result = predecessor_counts(np.array([], np.int32), np.array([], np.float64), 20, -1) self.assertEqual(result['since_last_reset_design_PN_events'], 0) def test_reject_bad_temporal_order(self): with self.assertRaises(ValueError): predecessor_counts(np.array([3, 1]), np.ones(2), 10, 5) with self.assertRaises(ValueError): predecessor_counts(np.array([3]), np.ones(1), 10, 10) def test_first_ticks_missing_and_repeated_events(self): values = first_ticks(np.array([1, 3, 3], np.int32), np.array([2, 1, 2], np.int32), np.arange(4, dtype=np.int32), n=4) np.testing.assert_array_equal(values, [-1, 3, 1, -1]) def test_exact_role_mapping_unknown_stays_unknown(self): labels = source_labels(np.arange(4), np.array(['ALPN', 'ALPN', 'ALLN', 'unclassified']), np.array([True, False, False, False])) self.assertEqual(labels.tolist(), ['design_ALPN', 'non_design_ALPN', 'ALLN', 'unclassified']) def test_window_aggregation_preserves_sign_and_exclusion(self): stats = {} _window_add(stats, pd.DataFrame({'weight_SI': [1., -2., 9.]}), np.array([True, True, False]), np.array(['ALLN'] * 3)) self.assertEqual(stats['ALLN']['events'], 2) self.assertEqual(stats['ALLN']['positive_weight_SI'], 1.) self.assertEqual(stats['ALLN']['negative_abs_weight_SI'], 2.) def test_bit_equal_rejects_signed_zero(self): self.assertFalse(bit_equal(np.array([0.]), np.array([-0.]))) @staticmethod def observer(): sink = MemorySink() incoming = SimpleNamespace(src_global=np.array([0, 1, 2], np.int32), dst_local=np.array([0, 0, 1], np.int32), edge_id=np.arange(3, dtype=np.int64), weights=np.array([.1, 0., -.2], np.float64)) observer = ALLNObserver(np.array([3, 4], np.int32), incoming, np.array(['ALPN', 'APL', 'ALLN', 'ALLN', 'ALLN']), np.array([True, False, False, False, False]), np.arange(5, dtype=np.uint64), 19, 19, sink) return observer, sink @staticmethod def stage(due, accepted, ref=None): return dict(h32=np.zeros(2, np.float32), delta_si=np.zeros(2), request=np.zeros(2, bool), g=np.zeros(2), g_pre_apl=np.zeros(2), due_edge_indices=np.asarray(due, np.int64), arrival_accepted=np.asarray(accepted, bool), ref_after_reset=np.zeros(2, np.int32) if ref is None else ref, fired=np.zeros(2, bool)) def test_observer_all_accepted_signs_and_zeros(self): observer, sink = self.observer() for tick in range(18): observer(tick, self.stage([], [])) observer(18, self.stage([0, 1, 2], [True, True, True])) report = observer.report() self.assertEqual(sink.rows, 3) self.assertEqual(report['source_groups']['APL']['zero'], 1) self.assertEqual(sink.calls[-1]['source_spike_tick'].tolist(), [0, 0, 0]) self.assertEqual(observer.design_ticks[3], [18]) self.assertEqual(sink.calls[-1]['weight_sign'].tolist(), [1, 0, -1]) def test_observer_ref_guard_and_no_rejected_records(self): observer, sink = self.observer() for tick in range(18): observer(tick, self.stage([], [])) stage = self.stage([0, 1, 2], [False, False, True], np.array([22, 0], np.int32)) stage['fired'][0] = True observer(18, stage) self.assertEqual(sink.rows, 1) self.assertEqual(observer.report()['rejected_events'], 2) self.assertEqual(observer.last.tolist(), [18, -1]) def test_observer_bad_acceptance_raises(self): observer, _ = self.observer() for tick in range(18): observer(tick, self.stage([], [])) with self.assertRaises(ValueError): observer(18, self.stage([0], [True], np.array([22, 0], np.int32))) def test_observer_rejects_unexpected_input_and_graded_APL(self): observer, _ = self.observer() stage = self.stage([], []) stage['request'][0] = True with self.assertRaises(ValueError): observer(0, stage) stage['request'][0] = False stage['h32'][0] = 1. with self.assertRaises(ValueError): observer(0, stage) def test_frozen_replay_core_synthetic_integration(self): module_spec = importlib.util.spec_from_file_location('origin_test_frozen_core', Path(__file__).resolve().parent.parent / 'kc-drive-audit-v1/replay_core.py') core = importlib.util.module_from_spec(module_spec) module_spec.loader.exec_module(core) incoming = core.IncomingIndex(np.array([0], np.int32), np.array([0], np.int32), np.array([0], np.int64), np.array([1.], np.float64), 2) sink = MemorySink() observer = ALLNObserver(np.array([3, 4], np.int32), incoming, np.array(['ALPN', 'APL', 'ALLN', 'ALLN', 'ALLN']), np.array([True, False, False, False, False]), np.arange(5, dtype=np.uint64), 40, 30, sink) coeff = dict(gd=.98, c0=-.00052, cg=.005, cv=.99, threshold=-.045, reset=-.052, seed_volt=-.044) result = core.replay(np.full(2, -.052), np.zeros(2), coeff, 40, np.array([3, 4], np.int32), dict(tick=np.array([], np.int32), neuron_idx=np.array([], np.int32)), dict(step=np.array([0], np.int32), neuron_idx=np.array([0], np.int32)), incoming, np.zeros((40, 2), np.float32), observer) self.assertEqual(observer.report()['accepted_events'], 1) self.assertGreater(int(result['first_spike'][0]), 18) self.assertEqual(int(result['first_spike'][1]), -1) self.assertEqual(observer.design_ticks[3], [18]) nonempty = [call for call in sink.calls if len(call['edge_id'])] self.assertEqual(nonempty[0]['target_previous_spike_tick'].tolist(), [-1]) if __name__ == '__main__': unittest.main()