Digital Drosophila Archive
37 lines
1.4 KiB
Python
37 lines
1.4 KiB
Python
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
import numpy as np
|
|
import pyarrow.parquet as pq
|
|
from run_audit import first_ticks, temporal_relation, min_observed, EventSink
|
|
|
|
|
|
class HarnessTests(unittest.TestCase):
|
|
def test_first_tick_uses_minimum_not_order(self):
|
|
self.assertEqual(first_ticks(np.array([4, 2, 3], np.int32), np.array([7, 7, 8], np.int32),
|
|
np.array([7, 8, 9], np.int32)).tolist(), [2, 3, -1])
|
|
|
|
def test_strict_temporal_boundary(self):
|
|
self.assertEqual(temporal_relation(17, 18), 'before')
|
|
self.assertEqual(temporal_relation(18, 18), 'same_tick')
|
|
self.assertEqual(temporal_relation(19, 18), 'after')
|
|
self.assertEqual(temporal_relation(-1, 18), 'unobserved_source')
|
|
self.assertEqual(temporal_relation(18, -1), 'unobserved_target')
|
|
self.assertEqual(temporal_relation(-1, -1), 'both_unobserved')
|
|
|
|
def test_min_observed(self):
|
|
self.assertEqual(min_observed([-1, 0, 3]), 0)
|
|
self.assertEqual(min_observed([-1, -1]), -1)
|
|
|
|
def test_empty_event_sink_has_schema(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
path = Path(tmp) / 'events.parquet'
|
|
sink = EventSink(path)
|
|
sink.close()
|
|
self.assertEqual(pq.read_table(path).num_rows, 0)
|
|
self.assertIn('source_spike_tick', pq.read_schema(path).names)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|