Digital Drosophila Archive
130 lines
5.7 KiB
Python
130 lines
5.7 KiB
Python
"""Synthetic CPU-only checks; no Torch, data bundle, or scientific trial."""
|
|
from pathlib import Path
|
|
import sys
|
|
import unittest
|
|
|
|
import numpy as np
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
from timing_queue import TimingQueue, audit_sequence
|
|
|
|
|
|
class TimingQueueTests(unittest.TestCase):
|
|
def sequence(self, delay, steps, width=3):
|
|
generated = np.arange(1, steps * width + 1, dtype=np.float32).reshape(steps, width)
|
|
queue = TimingQueue(delay, width)
|
|
applied = np.empty_like(generated)
|
|
for tick, row in enumerate(generated):
|
|
applied[tick] = queue.push(row, tick=tick)
|
|
return queue, generated, applied, queue.pending()
|
|
|
|
def test_delay_zero_identity(self):
|
|
queue, generated, applied, pending = self.sequence(0, 40)
|
|
self.assertTrue(np.array_equal(generated.view(np.uint32), applied.view(np.uint32)))
|
|
self.assertEqual(pending.shape, (0, 3))
|
|
self.assertTrue(audit_sequence(generated, applied, pending, 0)["pass_all"])
|
|
self.assertEqual(queue.metadata()["pending_generated_count"], 0)
|
|
|
|
def test_delay_18_boundary(self):
|
|
_, generated, applied, pending = self.sequence(18, 39)
|
|
self.assertTrue(np.all(applied[:18].view(np.uint32) == 0))
|
|
np.testing.assert_array_equal(applied[18], generated[0])
|
|
np.testing.assert_array_equal(applied[36], generated[18])
|
|
np.testing.assert_array_equal(pending, generated[-18:])
|
|
audit_sequence(generated, applied, pending, 18)
|
|
|
|
def test_short_sequence_keeps_initial_pending_slots(self):
|
|
for steps in (0, 1, 7, 17, 18, 19):
|
|
with self.subTest(steps=steps):
|
|
queue, generated, applied, pending = self.sequence(18, steps)
|
|
meta = audit_sequence(generated, applied, pending, 18)
|
|
self.assertEqual(pending.shape, (18, 3))
|
|
self.assertEqual(meta["pending_initial_zero_prefix"], max(18 - steps, 0))
|
|
self.assertEqual(meta["pending_generated_count"], min(18, steps))
|
|
self.assertEqual(queue.steps, steps)
|
|
self.assertTrue(np.all(pending[:max(18 - steps, 0)].view(np.uint32) == 0))
|
|
|
|
def test_no_alias_to_input_or_return_or_pending(self):
|
|
queue = TimingQueue(18, 1)
|
|
row = np.array([3], dtype=np.float32)
|
|
first = queue.push(row)
|
|
row[0] = 9
|
|
first[0] = 8
|
|
pending = queue.pending()
|
|
pending[:] = 12
|
|
for _ in range(17):
|
|
queue.push(np.zeros(1, dtype=np.float32))
|
|
self.assertEqual(queue.push(np.zeros(1, dtype=np.float32))[0], 3)
|
|
|
|
def test_zero_delay_return_owned(self):
|
|
row = np.array([2], dtype=np.float32)
|
|
value = TimingQueue(0, 1).push(row)
|
|
value[0] = 99
|
|
self.assertEqual(row[0], 2)
|
|
|
|
def test_signed_zero_subnormal_and_extreme_bits(self):
|
|
row = np.array([0x80000000, 1, 0x7f7fffff], dtype=np.uint32).view(np.float32)
|
|
for delay in (0, 18):
|
|
queue = TimingQueue(delay, 3)
|
|
generated = np.stack([row.copy() for _ in range(23)])
|
|
applied = np.stack([queue.push(x) for x in generated])
|
|
audit_sequence(generated, applied, queue.pending(), delay)
|
|
|
|
def test_reject_invalid_configuration(self):
|
|
for delay in (True, 1, -18, 18.0, np.int32(18)):
|
|
with self.assertRaises(ValueError):
|
|
TimingQueue(delay)
|
|
for width in (0, -1, True, 2.0):
|
|
with self.assertRaises(ValueError):
|
|
TimingQueue(18, width)
|
|
|
|
def test_reject_invalid_sample_without_advance(self):
|
|
queue = TimingQueue(18, 2)
|
|
bad = [np.ones(2, dtype=np.float64), np.ones((1, 2), dtype=np.float32),
|
|
np.ones(3, dtype=np.float32), [1, 2],
|
|
np.array([np.nan, 1], dtype=np.float32), np.array([np.inf, 1], dtype=np.float32),
|
|
np.array([-1, 1], dtype=np.float32)]
|
|
for row in bad:
|
|
with self.assertRaises(ValueError):
|
|
queue.push(row)
|
|
self.assertEqual(queue.steps, 0)
|
|
|
|
def test_tick_must_be_consecutive(self):
|
|
queue = TimingQueue(0, 1)
|
|
row = np.zeros(1, dtype=np.float32)
|
|
for tick in (1, True, -1, .0):
|
|
with self.assertRaises(ValueError):
|
|
queue.push(row, tick=tick)
|
|
queue.push(row, tick=0)
|
|
with self.assertRaises(ValueError):
|
|
queue.push(row, tick=0)
|
|
self.assertEqual(queue.steps, 1)
|
|
|
|
def test_audit_detects_shift_or_initial_negative_zero(self):
|
|
_, generated, applied, pending = self.sequence(18, 40)
|
|
applied[18, 0] += 1
|
|
with self.assertRaises(ValueError):
|
|
audit_sequence(generated, applied, pending, 18)
|
|
_, generated, applied, pending = self.sequence(18, 40)
|
|
applied.view(np.uint32)[0, 0] = 0x80000000
|
|
with self.assertRaises(ValueError):
|
|
audit_sequence(generated, applied, pending, 18)
|
|
|
|
def test_audit_rejects_invalid_schema(self):
|
|
_, generated, applied, pending = self.sequence(18, 2)
|
|
for g, a, p in ((generated.astype(np.float64), applied, pending),
|
|
(generated, applied[:1], pending), (generated, applied, pending[:2])):
|
|
with self.assertRaises(ValueError):
|
|
audit_sequence(g, a, p, 18)
|
|
|
|
def test_finite_sequence_sum_preserved_without_brain_tail(self):
|
|
queue, generated, applied, pending = self.sequence(18, 53)
|
|
np.testing.assert_array_equal(applied.sum(axis=0, dtype=np.float64) + pending.sum(axis=0, dtype=np.float64),
|
|
generated.sum(axis=0, dtype=np.float64))
|
|
self.assertEqual(queue.steps, 53)
|
|
self.assertFalse(queue.metadata()["brain_tail_advanced"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|