diff --git a/pyproject.toml b/pyproject.toml index 368615c..8487a3f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -52,6 +52,9 @@ dev = [ "tox", ] +[tool.coverage.report] +show_missing = true + [tool.isort] profile = "black" diff --git a/src/pytest_reserial/reserial.py b/src/pytest_reserial/reserial.py index 4ba3e1f..02fdfb6 100644 --- a/src/pytest_reserial/reserial.py +++ b/src/pytest_reserial/reserial.py @@ -9,7 +9,7 @@ from typing import Callable, Dict, Iterator, Literal, Tuple import pytest -from serial import Serial # type: ignore[import-untyped] +from serial import PortNotOpenError, Serial # type: ignore[import-untyped] TrafficLog = Dict[Literal["rx", "tx"], bytes] PatchMethods = Tuple[ @@ -18,6 +18,7 @@ Callable[[Serial], None], # open Callable[[Serial], None], # close Callable[[Serial, bool], None], # _reconfigure_port + Callable[[Serial], int], # in_waiting ] @@ -66,14 +67,20 @@ def reserial( test_name = request.node.name log = get_traffic_log(mode, log_path, test_name) - read_patch, write_patch, open_patch, close_patch, reconfigure_port_patch = ( - get_patched_methods(mode, log) - ) + ( + read_patch, + write_patch, + open_patch, + close_patch, + reconfigure_port_patch, + in_waiting_patch, + ) = get_patched_methods(mode, log) monkeypatch.setattr(Serial, "read", read_patch) monkeypatch.setattr(Serial, "write", write_patch) monkeypatch.setattr(Serial, "open", open_patch) monkeypatch.setattr(Serial, "close", close_patch) monkeypatch.setattr(Serial, "_reconfigure_port", reconfigure_port_patch) + monkeypatch.setattr(Serial, "in_waiting", in_waiting_patch) yield @@ -136,7 +143,7 @@ def get_traffic_log(mode: Mode, log_path: Path, test_name: str) -> TrafficLog: def get_patched_methods(mode: Mode, log: TrafficLog) -> PatchMethods: - """Return patched read, write, open, and closed methods. + """Return patched read, write, open, etc methods. The methods should be monkeypatched over the corresponding `Serial` methods. @@ -158,6 +165,10 @@ def get_patched_methods(mode: Mode, log: TrafficLog) -> PatchMethods: Monkeypatch this over `Serial.open`. close_patch: Callable[[Serial], None] Monkeypatch this over `Serial.close`. + _reconfigure_port_patch: Callable[[Serial, bool], None] + Monkeypatch this over `Serial._reconfigure_port`. + in_waiting_patch: Callable[[Serial], int] + Monkeypatch this over `Serial.in_waiting`. """ if mode == Mode.REPLAY: return get_replay_methods(log) @@ -169,11 +180,12 @@ def get_patched_methods(mode: Mode, log: TrafficLog) -> PatchMethods: Serial.open, Serial.close, Serial._reconfigure_port, # noqa: SLF001 + Serial.in_waiting, ) def get_replay_methods(log: TrafficLog) -> PatchMethods: - """Return patched read, write, open, and close methods for replaying logged traffic. + """Return patched read, write, open, etc methods for replaying logged traffic. Parameters ---------- @@ -190,10 +202,15 @@ def get_replay_methods(log: TrafficLog) -> PatchMethods: Sets `Serial.is_open` to `True`. replay_close: Callable[[Serial], None] Sets `Serial.is_open` to `False`. + replay_reconfigure_port: Callable[[Serial, bool], None] + No-op + record_in_waiting: Callable[[Serial], int] + Return the number of bytes of RX traffic left to replay. + """ def replay_write( - self: Serial, # noqa: ARG001 + self: Serial, data: bytes, ) -> int: """Compare TX data to recording instead of writing to the bus. @@ -206,6 +223,9 @@ def replay_write( _pytest.outcomes.Failed If written data does not match recorded data. """ + if not self.is_open: + raise PortNotOpenError + if data == log["tx"][: len(data)]: log["tx"] = log["tx"][len(data) :] else: @@ -218,7 +238,7 @@ def replay_write( return len(data) def replay_read( - self: Serial, # noqa: ARG001 + self: Serial, size: int = 1, ) -> bytes: """Replay RX data from recording instead of reading from the bus. @@ -226,11 +246,28 @@ def replay_read( Monkeypatch this method over Serial.read to replay traffic. Parameters and return values are identical to Serial.read. """ + if not self.is_open: + raise PortNotOpenError + data = log["rx"][:size] log["rx"] = log["rx"][size:] return bytes(data) - return replay_read, replay_write, replay_open, replay_close, replay_reconfigure_port + @property # type: ignore[misc] + def replay_in_waiting( + self: Serial, # noqa:ARG001 + ) -> int: + """Return the number of bytes in RX data left to replay.""" + return len(log["rx"]) + + return ( + replay_read, + replay_write, + replay_open, + replay_close, + replay_reconfigure_port, + replay_in_waiting, + ) # The open/close method patches don't need access to logs, so they can stay down here. @@ -258,7 +295,7 @@ def replay_reconfigure_port( def get_record_methods(log: TrafficLog) -> PatchMethods: - """Return patched read, write, open, and close methods for recording traffic. + """Return patched read, write, open, etc methods for recording traffic. Parameters ---------- @@ -275,6 +312,12 @@ def get_record_methods(log: TrafficLog) -> PatchMethods: Does not need to be patched when recording, so this is `Serial.open`. record_close: Callable[[Serial], None] Does not need to be patched when recording, so this is `Serial.close`. + record_reconfigure_port: Callable[[Serial, bool], None] + Does not need to be patched when recording, + so this is `Serial._reconfigure_port`. + record_in_waiting: Callable[[Serial], int] + Does not need to be patched when recording, so this is `Serial.in_waiting`. + """ real_read = Serial.read real_write = Serial.write @@ -305,6 +348,7 @@ def record_read(self: Serial, size: int = 1) -> bytes: Serial.open, Serial.close, Serial._reconfigure_port, # noqa: SLF001 + Serial.in_waiting, ) diff --git a/tests/test_reserial.py b/tests/test_reserial.py index 3e9f8d3..7c73acb 100644 --- a/tests/test_reserial.py +++ b/tests/test_reserial.py @@ -11,12 +11,33 @@ def test_reserial(reserial): s = serial.Serial(port="/dev/ttyUSB0") s.write({TEST_TX!r}) + assert s.in_waiting == {len(TEST_RX)} assert s.read() == {TEST_RX!r} def test_reserial2(reserial): s = serial.Serial(port="/dev/ttyUSB0") s.write({TEST_TX!r}) assert s.read() == {TEST_RX!r} """ +TEST_FILE_REPLAY = f""" + import pytest + import serial + def test_reserial(reserial): + s = serial.Serial(port="/dev/ttyUSB0") + s.write({TEST_TX!r}) + assert s.in_waiting == {len(TEST_RX)} + assert s.read() == {TEST_RX!r} + assert s.in_waiting == 0 + s.close() + with pytest.raises(serial.PortNotOpenError): + s.read() + def test_reserial2(reserial): + s = serial.Serial(port="/dev/ttyUSB0") + s.write({TEST_TX!r}) + assert s.read() == {TEST_RX!r} + s.close() + with pytest.raises(serial.PortNotOpenError): + s.write({TEST_TX!r}) + """ TEST_FILE_BAD_TX = f""" import serial def test_reserial(reserial): @@ -39,6 +60,10 @@ def patch_write(self: Serial, data: bytes) -> int: def patch_read(self: Serial, size: int = 1) -> bytes: return TEST_RX + @property + def patch_in_waiting(self: Serial) -> int: + return len(TEST_RX) + def patch_open(self: Serial) -> None: self.is_open = True @@ -49,6 +74,7 @@ def patch_close(self: Serial) -> None: monkeypatch.setattr(Serial, "read", patch_read) monkeypatch.setattr(Serial, "open", patch_open) monkeypatch.setattr(Serial, "close", patch_close) + monkeypatch.setattr(Serial, "in_waiting", patch_in_waiting) result = pytester.runpytest("--record") with open("test_record.jsonl", "r") as f: @@ -108,7 +134,7 @@ def patch_close(self: Serial) -> None: def test_replay(pytester): pytester.makefile(".jsonl", test_replay=TEST_JSONL) - pytester.makepyfile(TEST_FILE) + pytester.makepyfile(TEST_FILE_REPLAY) result = pytester.runpytest("--replay") assert result.ret == 0