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()