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 == []