pyGUI/tests/test_stream_reader.py

import time

from pygui.backend.models.frame import Frame
from pygui.serialcomm.stream_reader import StreamReader, _crc16


class FakeTransport:
    """Yield canned data then blocks (returns b'')."""
    def __init__(self, lines):
        self._data = b"".join(lines)
        self._pos = 0
        self.closed = False
    def read(self, n=1):
        time.sleep(0.001)
        if self._pos >= len(self._data):
            return b""
        res = self._data[self._pos:self._pos+n]
        self._pos += n
        return res
    def close(self): self.closed = True


def make_packet(payload: bytes, seq: int = 1) -> bytes:
    """Build a wire packet exactly as the device does: AA 88, msg type,
    seq (LE), payload len (LE), payload, CRC16/MODBUS (LE) over the rest."""
    body = (b"\xaa\x88\x01" + seq.to_bytes(2, "little")
            + len(payload).to_bytes(2, "little") + payload)
    return body + _crc16(body).to_bytes(2, "little")


def parse_payload(payload: bytes, seq: int) -> Frame:
    return Frame(seq=seq, time=0.0, values=[float(b) for b in payload])


def test_crc16_known_vector():
    # CRC16/MODBUS check value for "123456789" is 0x4B37.
    assert _crc16(b"123456789") == 0x4B37


def test_binary_stream_parses_crc_valid_packets():
    got = []
    data = make_packet(b"\x01\x02", seq=1) + make_packet(b"\x03\x04", seq=2)
    transport = FakeTransport([data])
    r = StreamReader(transport, parse_payload,
                    on_frame=got.append, on_error=lambda e: None)
    r.start()
    time.sleep(0.2)
    r.stop()
    assert [f.seq for f in got] == [1, 2]
    assert r.error_count == 0
    assert transport.closed


def test_corrupt_crc_packet_dropped_and_stream_continues():
    got, errors = [], []
    bad = bytearray(make_packet(b"\x09\x09", seq=1))
    bad[-1] ^= 0xFF  # break the CRC
    data = bytes(bad) + make_packet(b"\x01\x02", seq=2)
    r = StreamReader(FakeTransport([data]), parse_payload,
                    on_frame=got.append, on_error=errors.append)
    r.start()
    time.sleep(0.2)
    r.stop()
    assert [f.seq for f in got] == [2]
    assert r.error_count == 1


def test_bogus_giant_payload_len_does_not_stall_the_stream():
    """A false 0xAA 0x88 inside data yields a garbage payload_len (up to
    64KB). Without a sanity cap the reader waits forever accumulating a
    packet that doesn't exist — the 'frames stop, no error' symptom."""
    got = []
    # False sync + header claiming a 0xFFFF-byte payload, then a real packet.
    bogus = b"\xaa\x88\x01\x00\x00\xff\xff"
    data = bogus + make_packet(b"\x01\x02", seq=7)
    r = StreamReader(FakeTransport([data]), parse_payload,
                    on_frame=got.append, on_error=lambda e: None)
    r.start()
    time.sleep(0.2)
    r.stop()
    assert [f.seq for f in got] == [7]


def test_quiet_port_between_scans_is_not_a_resync_failure():
    """Device pauses between measurement scans. Empty reads must not count
    toward the resync give-up cap (C#'s findMessageStart waits forever) —
    the old code died with TimeoutError after ~50 empty reads."""
    errors = []
    got = []
    r = StreamReader(FakeTransport([make_packet(b"\x01\x02", seq=1)]),
                    parse_payload, on_frame=got.append, on_error=errors.append)
    r.start()
    time.sleep(0.5)  # long idle stretch after the single packet drains
    r.stop()
    assert [f.seq for f in got] == [1]
    assert errors == []


def test_resync_giveup_debug_log_has_bytes_error_log_does_not(caplog):
    """Regression: the give-up log used to dump `buf` after it had already
    been cleared for the new read, so the hex-diagnostic came back empty —
    useless for diagnosing what was actually on the wire. Fix: real bytes
    go to DEBUG only; the ERROR line the terminal/Activity Log show stays
    free of raw internal buffer content."""
    import logging
    errors = []
    # 0xAA/0x88 excluded so this exercises _run_binary's resync path even
    # though it's called directly (bypassing the ASCII/binary dispatch,
    # which is exercised separately — see test_ascii_hex_dump_stream_*).
    garbage = b"\x42" * 20000
    r = StreamReader(FakeTransport([garbage]), parse_payload,
                    on_frame=lambda f: None, on_error=errors.append)
    with caplog.at_level(logging.DEBUG, logger="pygui.serialcomm.stream_reader"):
        r._run_binary(bytearray())
    assert len(errors) == 1
    error_lines = [r.message for r in caplog.records if r.levelname == "ERROR"]
    debug_lines = [r.message for r in caplog.records if r.levelname == "DEBUG"]
    assert any("giving up" in m for m in error_lines)
    assert not any("42424242" in m for m in error_lines), (
        "raw wire bytes must not appear in the ERROR line users see"
    )
    assert any("42424242" in m for m in debug_lines)


def test_resync_gives_up_after_16kb_of_garbage():
    """Give-up cap counts discarded BYTES, not read attempts, so it fires
    at a fixed data volume regardless of transfer rate."""
    errors = []
    got = []
    garbage = b"\x01" * 20000  # never contains 0xAA 0x88; well past the cap
    r = StreamReader(FakeTransport([garbage]), parse_payload,
                    on_frame=got.append, on_error=errors.append)
    r._run_binary(bytearray())  # deterministic give-up — no thread needed
    assert got == []
    assert len(errors) == 1
    assert "resync" in str(errors[0]).lower()


def test_ascii_hex_dump_stream_decodes_real_wafer_capture():
    """Regression using an exact byte capture from real hardware (family A):
    the wire is plain ASCII hex text, 4 chars/word, no 0xAA 0x88 framing at
    all — see docs/adr/0007. _run() must route to the ASCII path and the
    words must decode to plausible room-temperature values, not stall."""
    got = []
    words = ["280B", "1C0B", "220B", "1B0B", "200B", "1F0B", "190B", "1D0B",
            "1B0B", "240B", "180B", "210B", "2B0B", "260B", "290B", "2B0B"]
    # aepwafer.yaml's family "A" has 48 real sensors; pad to one full block
    # (256 words) the way the real device does — extra slots are unpopulated.
    block = "".join(words) + "0000" * (256 - len(words))
    r = StreamReader(FakeTransport([block.encode()]), parse_payload,
                    on_frame=got.append, on_error=lambda e: None,
                    family_code="A")
    r.start()
    time.sleep(0.2)
    r.stop()
    assert len(got) == 1
    assert len(got[0].values) == 48  # sensor_count_for("A")
    # First word "280B" byte-swaps to "0B28" -> ~22.31C via the AEP formula.
    assert 22.0 < got[0].values[0] < 22.5


class WedgedTransport:
    """Simulates a worker thread stuck in a blocking read() that ignores
    the stop event — read() never returns on its own. stop() must close
    the port to unblock it rather than relying on the read loop to notice."""
    def __init__(self):
        self.closed = False

    def read(self, n=1):
        while not self.closed:
            time.sleep(0.001)
        raise OSError("port closed")

    def close(self):
        self.closed = True


def test_stop_closes_transport_even_if_worker_thread_is_wedged():
    r = StreamReader(WedgedTransport(), parse_payload,
                    on_frame=lambda f: None, on_error=lambda e: None)
    r.start()
    time.sleep(0.05)  # let the thread enter the blocking read
    r.stop()
    assert r._transport.closed, (
        "stop() must close the transport itself — otherwise a wedged "
        "worker thread keeps the port open and a subsequent startStream() "
        "opens a second competing handle to the same port"
    )


def test_stop_does_not_report_the_close_it_caused_as_an_error():
    """stop()'s own transport.close() unblocks the worker thread's read()
    by making it raise — that's expected teardown, not a stream fault, and
    must not reach on_error (Activity Log would otherwise show a fake
    "Live stream error" on every ordinary Stop)."""
    errors = []
    r = StreamReader(WedgedTransport(), parse_payload,
                    on_frame=lambda f: None, on_error=errors.append)
    r.start()
    time.sleep(0.05)
    r.stop()
    time.sleep(0.05)  # let the worker thread's except-handler run
    assert errors == []