From 3b821f76d048919b61f72414a91df6dc4678388a Mon Sep 17 00:00:00 2001 From: Yash Anand <164618230+YashAnand69@users.noreply.github.com> Date: Mon, 12 Oct 2026 13:39:27 +0530 Subject: [PATCH 1/2] fix: retain UART readline newline terminators --- pslab/bus/busio.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/pslab/bus/busio.py b/pslab/bus/busio.py index 97771dc..3c2c37e 100644 --- a/pslab/bus/busio.py +++ b/pslab/bus/busio.py @@ -493,12 +493,12 @@ def _read_with_timeout(self, nbytes: int = None, *, line=False): char = self._read_byte() start_time = time.time() - if line and char == 0xA: - break - data.append(char) total_read += 1 + if line and char == 0xA: + break + if nbytes and total_read == nbytes: break From 51bc003a1a70a914bbe778dbf42185806d7c6d64 Mon Sep 17 00:00:00 2001 From: Yash Anand <164618230+YashAnand69@users.noreply.github.com> Date: Mon, 12 Oct 2026 13:39:27 +0530 Subject: [PATCH 2/2] test: cover UART complete and partial line framing --- tests/test_busio_uart_readline.py | 117 ++++++++++++++++++++++++++++++ 1 file changed, 117 insertions(+) create mode 100644 tests/test_busio_uart_readline.py diff --git a/tests/test_busio_uart_readline.py b/tests/test_busio_uart_readline.py new file mode 100644 index 0000000..9523193 --- /dev/null +++ b/tests/test_busio_uart_readline.py @@ -0,0 +1,117 @@ +"""Exercise UART line framing with controlled firmware replies, without hardware.""" + +from collections import deque + +import pytest + +import pslab.bus.busio as busio +import pslab.protocol as CP +from pslab.bus.uart import _UARTPrimitive + + +class Clock: + """Advance the read timeout only when the poller sleeps.""" + + def __init__(self): + self.now = 0.0 + + def time(self): + return self.now + + def sleep(self, seconds): + self.now += seconds + + +class UARTTransport: + """Answer actual UART status/read commands from a receive buffer.""" + + version = "PSLab V6" + + def __init__(self, data): + self.received = deque(data) + self.command = [] + self.read_commands = [] + + def send_byte(self, value): + self.command.append(value) + + def send_int(self, value): + self.command.append(value) + + def get_ack(self): + self.command.clear() + return 1 + + def get_byte(self): + command = tuple(self.command) + self.read_commands.append(command) + self.command.clear() + assert command[0] == CP.UART_2 + if command[1] == CP.READ_UART2_STATUS: + return int(bool(self.received)) + assert command[1] == CP.READ_BYTE + return self.received.popleft() + + +@pytest.fixture +def uart_factory(monkeypatch): + """Use an explicit parity to isolate reads from the constructor defect.""" + monkeypatch.setattr(busio, "time", Clock()) + monkeypatch.setattr(_UARTPrimitive, "_brgval", _UARTPrimitive._brgval) + monkeypatch.setattr(_UARTPrimitive, "_mode", _UARTPrimitive._mode) + + def create(data): + device = UARTTransport(data) + uart = busio.UART(device, parity=busio.Parity.EVEN, timeout=0.03) + return uart, device + + return create + + +@pytest.mark.parametrize("line", [b"\n", b"hello\n", b"hello\r\n", b"\x00\n"]) +def test_readline_retains_terminator_without_consuming_next_line(uart_factory, line): + uart, device = uart_factory(line + b"next\n") + assert uart.readline() == line + assert bytes(device.received) == b"next\n" + assert device.read_commands.count((CP.UART_2, CP.READ_BYTE)) == len(line) + + +def test_successive_lines_distinguish_blank_line_from_timeout(uart_factory): + uart, device = uart_factory(b"\nnext\nlast") + assert uart.readline() == b"\n" + assert uart.readline() == b"next\n" + assert uart.readline() == b"last" + assert uart.readline() is None + assert not device.received + + +def test_readline_no_data_returns_none(uart_factory): + uart, device = uart_factory(b"") + assert uart.readline() is None + assert all(command[1] == CP.READ_UART2_STATUS for command in device.read_commands) + + +def test_readline_returns_partial_line_on_timeout(uart_factory): + uart, _ = uart_factory(b"partial\r") + assert uart.readline() == b"partial\r" + + +def test_raw_read_retains_newline_and_respects_byte_limit(uart_factory): + uart, device = uart_factory(b"x\nytail") + assert uart.read(3) == b"x\ny" + assert bytes(device.received) == b"tail" + + +def test_readinto_retains_newline_and_binary_data(uart_factory): + uart, device = uart_factory(b"\n\x00atail") + buffer = bytearray(3) + assert uart.readinto(buffer) == 3 + assert buffer == b"\n\x00a" + assert bytes(device.received) == b"tail" + + +def test_zero_length_read_does_not_touch_receive_buffer(uart_factory): + uart, device = uart_factory(b"\n") + assert uart.read(0) is None + assert bytes(device.received) == b"\n" + assert not device.read_commands