From 83323f2a00d89b188afe58d4d47ba8eb50e6119f Mon Sep 17 00:00:00 2001 From: Yash Anand <164618230+YashAnand69@users.noreply.github.com> Date: Sat, 10 Oct 2026 16:05:41 +0530 Subject: [PATCH 1/2] fix: reject empty I2C reads before device commands --- pslab/bus/busio.py | 12 ++++++ tests/test_i2c_empty_reads.py | 80 +++++++++++++++++++++++++++++++++++ 2 files changed, 92 insertions(+) create mode 100644 tests/test_i2c_empty_reads.py diff --git a/pslab/bus/busio.py b/pslab/bus/busio.py index 97771dc..7fac7f7 100644 --- a/pslab/bus/busio.py +++ b/pslab/bus/busio.py @@ -105,6 +105,8 @@ def readfrom_into( ) -> None: """Read from a device at specified address into a buffer. + The selected buffer slice must contain at least one byte. + Parameters ---------- address : int @@ -118,6 +120,10 @@ def readfrom_into( """ end = len(buffer) if end is None else end bytes_to_read = end - start + + if bytes_to_read < 1: + raise ValueError("Read buffer must contain at least one byte.") + self._start(address, 1) buffer[start:end] = self._read(bytes_to_read) self._stop() @@ -167,6 +173,8 @@ def writeto_then_readfrom( ): """Write to then read from a device at specified address. + The selected input buffer slice must contain at least one byte. + Parameters ---------- address : int @@ -187,6 +195,10 @@ def writeto_then_readfrom( out_end = len(buffer_out) if out_end is None else out_end in_end = len(buffer_in) if in_end is None else in_end bytes_to_read = in_end - in_start + + if bytes_to_read < 1: + raise ValueError("Read buffer must contain at least one byte.") + self._start(address, 0) self._send(buffer_out[out_start:out_end]) self._restart(address, 1) diff --git a/tests/test_i2c_empty_reads.py b/tests/test_i2c_empty_reads.py new file mode 100644 index 0000000..832e157 --- /dev/null +++ b/tests/test_i2c_empty_reads.py @@ -0,0 +1,80 @@ +"""Empty I2C reads must not consume sensor bytes or resize output buffers.""" + +import pytest + +import pslab.protocol as CP +from pslab.bus.busio import I2C + + +class I2CConnection: + """Record primitive commands and provide deterministic sensor bytes.""" + + def __init__(self): + self.commands = [] + self.bytes_read = 0 + + def send_byte(self, value): + """Record command and parameter bytes.""" + self.commands.append(value) + + def send_int(self, value): + """Accept the constructor's bus frequency configuration.""" + assert 0 <= value <= 65535 + + def get_ack(self): + """Acknowledge commands and the addressed peripheral.""" + return 1 + + def get_byte(self): + """Return a distinct byte for each sensor read.""" + self.bytes_read += 1 + return 0x40 + self.bytes_read + + +@pytest.mark.parametrize("combined", [False, True]) +@pytest.mark.parametrize("view", [False, True]) +@pytest.mark.parametrize("initial, offset", [(b"", 0), (b"abc", 1)]) +def test_empty_read_preserves_buffer_and_does_not_read_sensor( + combined, view, initial, offset +): + connection = I2CConnection() + bus = I2C(connection) + storage = bytearray(initial) + output = memoryview(storage) if view else storage + + connection.commands.clear() + + with pytest.raises(ValueError, match="at least one byte"): + if combined: + bus.writeto_then_readfrom( + 0x40, b"\x10", output, in_start=offset, in_end=offset + ) + else: + bus.readfrom_into(0x40, output, start=offset, end=offset) + + assert storage == initial + assert connection.bytes_read == 0 + assert CP.I2C_READ_MORE not in connection.commands + assert CP.I2C_READ_END not in connection.commands + assert connection.commands == [] + assert bus._running is False + assert bus._mode is None + + +@pytest.mark.parametrize("combined", [False, True]) +@pytest.mark.parametrize("count", [1, 3]) +def test_nonempty_read_keeps_ack_nack_sequence_and_slice(combined, count): + connection = I2CConnection() + bus = I2C(connection) + storage = bytearray(b"z" * (count + 2)) + + if combined: + bus.writeto_then_readfrom(0x40, b"\x10", storage, in_start=1, in_end=count + 1) + else: + bus.readfrom_into(0x40, storage, start=1, end=count + 1) + + assert storage == b"z" + bytes(range(0x41, 0x41 + count)) + b"z" + assert connection.bytes_read == count + assert connection.commands.count(CP.I2C_READ_MORE) == count - 1 + assert connection.commands.count(CP.I2C_READ_END) == 1 + assert connection.commands[-2:] == [CP.I2C_HEADER, CP.I2C_STOP] From 64583671a4ffee39d7b418fe296243b627a84c56 Mon Sep 17 00:00:00 2001 From: Yash Anand <164618230+YashAnand69@users.noreply.github.com> Date: Sat, 10 Oct 2026 21:11:08 +0530 Subject: [PATCH 2/2] fix: normalize I2C read slices before device commands --- pslab/bus/busio.py | 2 ++ tests/test_i2c_empty_reads.py | 31 +++++++++++++++++++++++++++++++ 2 files changed, 33 insertions(+) diff --git a/pslab/bus/busio.py b/pslab/bus/busio.py index 7fac7f7..e637961 100644 --- a/pslab/bus/busio.py +++ b/pslab/bus/busio.py @@ -119,6 +119,7 @@ def readfrom_into( Index to write up to but not include. Defaults to length of `buffer`. """ end = len(buffer) if end is None else end + start, end, _ = slice(start, end).indices(len(buffer)) bytes_to_read = end - start if bytes_to_read < 1: @@ -194,6 +195,7 @@ def writeto_then_readfrom( """ out_end = len(buffer_out) if out_end is None else out_end in_end = len(buffer_in) if in_end is None else in_end + in_start, in_end, _ = slice(in_start, in_end).indices(len(buffer_in)) bytes_to_read = in_end - in_start if bytes_to_read < 1: diff --git a/tests/test_i2c_empty_reads.py b/tests/test_i2c_empty_reads.py index 832e157..f93dfea 100644 --- a/tests/test_i2c_empty_reads.py +++ b/tests/test_i2c_empty_reads.py @@ -78,3 +78,34 @@ def test_nonempty_read_keeps_ack_nack_sequence_and_slice(combined, count): assert connection.commands.count(CP.I2C_READ_MORE) == count - 1 assert connection.commands.count(CP.I2C_READ_END) == 1 assert connection.commands[-2:] == [CP.I2C_HEADER, CP.I2C_STOP] + + +@pytest.mark.parametrize("combined", [False, True]) +@pytest.mark.parametrize("view", [False, True]) +@pytest.mark.parametrize("start, end", [(5, 6), (1, 8), (-2, None), (-8, 2)]) +def test_read_uses_effective_python_slice(combined, view, start, end): + connection = I2CConnection() + bus = I2C(connection) + storage = bytearray(b"abc") + output = memoryview(storage) if view else storage + selected = storage[start:end] + expected = storage.copy() + connection.commands.clear() + + def read(): + if combined: + bus.writeto_then_readfrom(0x40, b"\x10", output, in_start=start, in_end=end) + else: + bus.readfrom_into(0x40, output, start=start, end=end) + + if not selected: + with pytest.raises(ValueError, match="at least one byte"): + read() + assert connection.commands == [] + else: + read() + expected[start:end] = bytes(range(0x41, 0x41 + len(selected))) + assert connection.commands[-2:] == [CP.I2C_HEADER, CP.I2C_STOP] + assert storage == expected + assert connection.bytes_read == len(selected) + assert bus._running is False