93 lines
3.8 KiB
Python
93 lines
3.8 KiB
Python
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from altera_logic.stream import NativeStream
|
|
|
|
|
|
class StreamTests(unittest.TestCase):
|
|
def test_can_uart_identical_and_duplicates(self):
|
|
source, can, uart = NativeStream(), NativeStream(), NativeStream()
|
|
self.assertEqual((source.device_type, source.device_id, source.device_name), (6, 14, "Altera Logic"))
|
|
ident, data = source.metadata_packet(7, 1000000)
|
|
self.assertEqual(ident, 0x1EE3FE00)
|
|
self.assertEqual(data, bytes.fromhex("01 4C 07 00 40 42 0F 00"))
|
|
can.feed_can(ident, data)
|
|
_, raw = source.metadata_packet(7, 1000000, uart=True)
|
|
for byte in raw:
|
|
uart.feed_uart(bytes([byte]))
|
|
for i in range(100):
|
|
ident, data = source.data_packet(7, i, i)
|
|
can.feed_can(ident, data)
|
|
_, raw = source.data_packet(7, i, i, uart=True)
|
|
uart.feed_uart(raw)
|
|
can.feed_uart(raw) # Duplicate publication through the second transport.
|
|
self.assertEqual(can.snapshot(), uart.snapshot())
|
|
self.assertEqual(can.stats["duplicates"], 100)
|
|
self.assertEqual(can.stats["received"], 100)
|
|
|
|
def test_gap_old_session_and_counter_wrap(self):
|
|
s = NativeStream()
|
|
s.feed_can(*s.metadata_packet(10, 1000000))
|
|
s.feed_can(*s.data_packet(10, 0xfffffffe, 1))
|
|
s.feed_can(*s.data_packet(10, 1, 2))
|
|
snap = s.snapshot()
|
|
self.assertEqual(snap.indices, (0xfffffffe, 0x100000001))
|
|
self.assertEqual(s.stats["missing"], 2)
|
|
self.assertEqual(snap.breaks, (1, 1))
|
|
s.feed_can(*s.metadata_packet(11, 2000000))
|
|
s.feed_can(*s.data_packet(10, 5, 3))
|
|
s.feed_can(*s.metadata_packet(10, 1000000))
|
|
self.assertEqual(s.stats["session"], 11)
|
|
self.assertEqual(s.snapshot().samples, ())
|
|
s.feed_can(*s.data_packet(11, 0, 0x8001))
|
|
self.assertEqual(s.snapshot().samples, (0x8001,))
|
|
|
|
def test_malformed_filter_crc_and_recovery(self):
|
|
s = NativeStream()
|
|
ident, meta = s.metadata_packet(1, 1000)
|
|
s.feed_can(ident, meta, extended=False)
|
|
s.feed_can(ident, meta, remote=True)
|
|
s.feed_can(ident ^ (1 << 20), meta)
|
|
self.assertFalse(s.stats["has_meta"])
|
|
raw = bytearray(s.metadata_packet(1, 1000, uart=True)[1])
|
|
raw[-1] ^= 1
|
|
s.feed_uart(bytes(raw))
|
|
self.assertEqual(s.stats["crc_errors"], 1)
|
|
s.feed_uart(s.metadata_packet(1, 1000, uart=True)[1])
|
|
self.assertTrue(s.stats["has_meta"])
|
|
ident, data = s.data_packet(1, 0, 5)
|
|
s.feed_can(ident, data[:-1])
|
|
self.assertEqual(s.stats["invalid"], 1)
|
|
self.assertEqual(s.stats["count"], 0)
|
|
|
|
def test_bounded_history_and_export(self):
|
|
s = NativeStream()
|
|
for _ in range(3):
|
|
s.demo_step(4096)
|
|
snap = s.snapshot(demo=True)
|
|
self.assertEqual(len(snap.samples), 8192)
|
|
self.assertEqual(snap.indices[0], 4096)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory)/"stream.csv"
|
|
snap.save_csv(path)
|
|
rows = path.read_text().splitlines()
|
|
self.assertEqual(len(rows), 8193)
|
|
self.assertIn("gap_before", rows[0])
|
|
self.assertEqual(rows[1].split(",")[1], "4096")
|
|
|
|
def test_sparse_trace_does_not_connect_across_lost_samples(self):
|
|
s = NativeStream()
|
|
s.feed_can(*s.metadata_packet(1, 1000000))
|
|
for index, value in ((0, 0), (1, 1), (4, 0)):
|
|
s.feed_can(*s.data_packet(1, index, value))
|
|
trace = s.snapshot().traces[0]
|
|
self.assertEqual(trace[0], (0, 0, 1))
|
|
self.assertIn((1, 0, 0), trace)
|
|
self.assertIn((1, 1, 0), trace)
|
|
self.assertIn((4, 0, 1), trace)
|
|
self.assertNotIn((4, 1, 0), trace)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|