From 1e07c4aea202ebaf4894c8a6fe0442f4911c179a Mon Sep 17 00:00:00 2001 From: VivienP Date: Sat, 12 Sep 2026 09:44:55 +0200 Subject: [PATCH] test(io): share buffered Serial fixtures --- .../azenta/fluidx/intellixcap96_tests.py | 294 ++++++++++++------ .../line_barcode_scanner/serial_tests.py | 96 ++---- pylabrobot/io/testing.py | 107 +++++++ pylabrobot/io/testing_tests.py | 218 +++++++++++++ pylabrobot/kbiosystems/sealer_tests.py | 155 +++++---- 5 files changed, 644 insertions(+), 226 deletions(-) create mode 100644 pylabrobot/io/testing.py create mode 100644 pylabrobot/io/testing_tests.py diff --git a/pylabrobot/azenta/fluidx/intellixcap96_tests.py b/pylabrobot/azenta/fluidx/intellixcap96_tests.py index 4bc38bf3506..9cdecda3563 100644 --- a/pylabrobot/azenta/fluidx/intellixcap96_tests.py +++ b/pylabrobot/azenta/fluidx/intellixcap96_tests.py @@ -1,7 +1,6 @@ -import contextlib import unittest -from typing import Iterator, List -from unittest.mock import AsyncMock, patch +from typing import List, cast +from unittest.mock import AsyncMock, NonCallableMagicMock, call, patch from pylabrobot.azenta.fluidx import ( CartridgeProfile, @@ -11,55 +10,26 @@ get_error_message, is_recoverable_error, ) +from pylabrobot.io.testing import fake_serial ACK = "\x06" -class FakeXcapSerial: - """In-memory serial stand-in for the STX/ETX framed decapper protocol. +def xcap_io(script: List[List[str]]) -> NonCallableMagicMock: + """Return a serial fake with STX/ETX framed replies for each awaited write. ``script`` is a list of turns, one per ``write``. Each turn is the list of frame payloads the device emits in response to that write; every payload is queued wrapped in STX (0x02) .. ETX (0x03), exactly as the firmware frames it. """ - def __init__(self, script: List[List[str]]) -> None: - self.port = "FAKE" - self.written: List[str] = [] - self._script = list(script) - self._rx = bytearray() + turns = iter(list(script)) - async def setup(self) -> None: - pass + def on_write(data: bytes) -> bytes: + """Frame the next scripted response, or return no bytes after the final turn.""" + return b"".join(b"\x02" + payload.encode("ascii") + b"\x03" for payload in next(turns, [])) - async def stop(self) -> None: - pass - - async def reset_input_buffer(self) -> None: - self._rx.clear() - - def get_read_timeout(self) -> float: - return 5.0 - - def set_read_timeout(self, timeout: float) -> None: - pass - - @contextlib.contextmanager - def temporary_timeout(self, timeout: float) -> Iterator[None]: - yield - - async def write(self, data: bytes) -> None: - self.written.append(data.decode("ascii").rstrip("\x03")) - turn = self._script.pop(0) if self._script else [] - for payload in turn: - self._rx += b"\x02" + payload.encode("ascii") + b"\x03" - - async def read(self, num_bytes: int = 1) -> bytes: - if not self._rx: - return b"" - out = bytes(self._rx[:num_bytes]) - del self._rx[:num_bytes] - return out + return fake_serial(on_write=on_write, read_timeout=5.0) def status(word: str, echo: str = "a") -> List[str]: @@ -92,19 +62,19 @@ async def asyncSetUp(self) -> None: self.addCleanup(patcher.stop) def _make(self, script: List[List[str]], auto_recover: bool = True) -> FluidXIntelliXcap96: + """Create a decapper with scripted serial replies and the requested recovery behavior.""" device = FluidXIntelliXcap96(port="FAKE", auto_recover=auto_recover) - device.io = FakeXcapSerial(script) # type: ignore[assignment] + device.io = xcap_io(script) # type: ignore[assignment] return device - def _written(self, device: FluidXIntelliXcap96) -> List[str]: - return device.io.written # type: ignore[attr-defined,no-any-return] - # === Connection === async def test_setup_ok(self): device = self._make([status("StatusOK")]) await device.setup() - self.assertEqual(self._written(device), ["a"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual(write.await_args_list, [call(b"a\x03")]) + self.assertEqual(write.call_count, write.await_count) async def test_setup_busy_reports_engaged_estop(self): device = self._make([status("StatusBUSY"), extended(estop_active=True)]) @@ -169,7 +139,9 @@ def test_recoverable_errors_are_the_ones_homing_clears(self): async def test_request_error_code_returns_latched_code(self): device = self._make([query("8", "142")]) self.assertEqual(await device.request_error_code(), 142) - self.assertEqual(self._written(device), ["8"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual(write.await_args_list, [call(b"8\x03")]) + self.assertEqual(write.call_count, write.await_count) async def test_request_error_code_is_none_when_clear(self): device = self._make([query("8", "000")]) @@ -243,32 +215,51 @@ async def test_open_tray_moves_then_idle(self): ] ) await device.open_tray() - self.assertEqual(self._written(device), ["a", "f", "a", "a"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual( + write.await_args_list, + [call(b"a\x03"), call(b"f\x03"), call(b"a\x03"), call(b"a\x03")], + ) + self.assertEqual(write.call_count, write.await_count) async def test_open_tray_finishes_on_its_own_done_frame(self): device = self._make([status("StatusOK"), [ACK, "fOK"], [ACK, "aOK", "OpenDONE"]]) await device.open_tray() - self.assertEqual(self._written(device), ["a", "f", "a"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual(write.await_args_list, [call(b"a\x03"), call(b"f\x03"), call(b"a\x03")]) + self.assertEqual(write.call_count, write.await_count) async def test_open_tray_already_open_is_noop(self): # CommandIgnore = tray already open = success; no status wait afterwards. device = self._make([status("StatusOK"), [ACK, "fOK", "CommandIgnore"]]) await device.open_tray() - self.assertEqual(self._written(device), ["a", "f"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual(write.await_args_list, [call(b"a\x03"), call(b"f\x03")]) + self.assertEqual(write.call_count, write.await_count) async def test_close_tray_moves_then_idle(self): device = self._make( [status("StatusOK"), [ACK, "gOK"], status("StatusBUSY"), status("StatusOK")] ) await device.close_tray() - self.assertEqual(self._written(device), ["a", "g", "a", "a"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual( + write.await_args_list, + [call(b"a\x03"), call(b"g\x03"), call(b"a\x03"), call(b"a\x03")], + ) + self.assertEqual(write.call_count, write.await_count) async def test_close_tray_preserves_caps_held_state(self): device = self._make( [status("StatusRECAP"), [ACK, "gOK"], status("StatusBUSY"), status("StatusRECAP")] ) await device.close_tray() - self.assertEqual(self._written(device), ["a", "g", "a", "a"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual( + write.await_args_list, + [call(b"a\x03"), call(b"g\x03"), call(b"a\x03"), call(b"a\x03")], + ) + self.assertEqual(write.call_count, write.await_count) async def test_step_tray_out_and_in(self): device = self._make( @@ -281,7 +272,12 @@ async def test_step_tray_out_and_in(self): ) await device.step_tray_out() await device.step_tray_in() - self.assertEqual(self._written(device), ["a", "s", "a", "S"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual( + write.await_args_list, + [call(b"a\x03"), call(b"s\x03"), call(b"a\x03"), call(b"S\x03")], + ) + self.assertEqual(write.call_count, write.await_count) async def test_tray_move_out_of_range_raises(self): device = self._make([status("StatusOK"), [ACK, "sOK", "sERROR"], query("8", "148")]) @@ -300,7 +296,9 @@ async def test_tray_move_times_out_without_a_second_echo(self): async def test_home_moves_then_idle(self): device = self._make([[ACK, "ZOK"], status("StatusBUSY"), status("StatusOK")]) await device.home() - self.assertEqual(self._written(device), ["Z", "a", "a"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual(write.await_args_list, [call(b"Z\x03"), call(b"a\x03"), call(b"a\x03")]) + self.assertEqual(write.call_count, write.await_count) async def test_reset_error_recovers_manual_state_by_homing(self): device = self._make( @@ -312,7 +310,12 @@ async def test_reset_error_recovers_manual_state_by_homing(self): ] ) await device.reset_error() - self.assertEqual(self._written(device), ["a", "Z", "a", "a"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual( + write.await_args_list, + [call(b"a\x03"), call(b"Z\x03"), call(b"a\x03"), call(b"a\x03")], + ) + self.assertEqual(write.call_count, write.await_count) async def test_reset_error_recovers_error_state_by_homing(self): device = self._make( @@ -324,17 +327,26 @@ async def test_reset_error_recovers_error_state_by_homing(self): ] ) await device.reset_error() - self.assertEqual(self._written(device), ["a", "Z", "a", "a"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual( + write.await_args_list, + [call(b"a\x03"), call(b"Z\x03"), call(b"a\x03"), call(b"a\x03")], + ) + self.assertEqual(write.call_count, write.await_count) async def test_reset_error_is_noop_when_healthy(self): device = self._make([status("StatusOK")]) await device.reset_error() - self.assertEqual(self._written(device), ["a"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual(write.await_args_list, [call(b"a\x03")]) + self.assertEqual(write.call_count, write.await_count) async def test_initialize_keeping_caps_on_pins(self): device = self._make([status("StatusMANUAL"), [ACK, "zOK", "zDONE"]]) await device.initialize_keeping_caps_on_pins() - self.assertEqual(self._written(device), ["a", "z"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual(write.await_args_list, [call(b"a\x03"), call(b"z\x03")]) + self.assertEqual(write.call_count, write.await_count) async def test_initialize_keeping_caps_on_pins_reports_a_home_failure(self): device = self._make([status("StatusMANUAL"), [ACK, "zOK", "HomeERROR"], query("8", "156")]) @@ -347,7 +359,9 @@ async def test_initialize_keeping_caps_on_pins_requires_manual_mode(self): with self.assertRaises(FluidXError) as ctx: await device.initialize_keeping_caps_on_pins() self.assertIn("manual recovery mode", str(ctx.exception)) - self.assertEqual(self._written(device), ["a"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual(write.await_args_list, [call(b"a\x03")]) + self.assertEqual(write.call_count, write.await_count) async def test_operation_auto_recovers_from_latched_error(self): device = self._make( @@ -363,24 +377,39 @@ async def test_operation_auto_recovers_from_latched_error(self): ] ) await device.open_tray() + write = cast(AsyncMock, device.io.write) self.assertEqual( - self._written(device), - ["a", "Z", "a", "a", "a", "f", "a", "a"], + write.await_args_list, + [ + call(b"a\x03"), + call(b"Z\x03"), + call(b"a\x03"), + call(b"a\x03"), + call(b"a\x03"), + call(b"f\x03"), + call(b"a\x03"), + call(b"a\x03"), + ], ) + self.assertEqual(write.call_count, write.await_count) async def test_latched_error_raises_when_auto_recover_disabled(self): device = self._make([status("StatusError"), query("8", "113")], auto_recover=False) with self.assertRaises(FluidXError) as ctx: await device.open_tray() self.assertEqual(ctx.exception.error_code, 113) - self.assertEqual(self._written(device), ["a", "8"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual(write.await_args_list, [call(b"a\x03"), call(b"8\x03")]) + self.assertEqual(write.call_count, write.await_count) async def test_operation_blocked_during_manual_recovery(self): device = self._make([status("StatusMANUAL")]) with self.assertRaises(FluidXError) as ctx: await device.open_tray() self.assertIn("manual recovery", str(ctx.exception)) - self.assertEqual(self._written(device), ["a", "8"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual(write.await_args_list, [call(b"a\x03"), call(b"8\x03")]) + self.assertEqual(write.call_count, write.await_count) # === Decap / recap === @@ -394,7 +423,12 @@ async def test_decap_success(self): ] ) await device.decap() - self.assertEqual(self._written(device), ["a", "h", "a", "a"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual( + write.await_args_list, + [call(b"a\x03"), call(b"h\x03"), call(b"a\x03"), call(b"a\x03")], + ) + self.assertEqual(write.call_count, write.await_count) async def test_decap_blocked_when_already_decapped(self): device = self._make([status("StatusRecap")]) @@ -427,7 +461,12 @@ async def test_decap_queries_the_error_code_when_the_reply_omits_it(self): self.assertEqual(ctx.exception.error_code, 114) self.assertTrue(ctx.exception.recoverable) self.assertIn("Invalid tube height", str(ctx.exception)) - self.assertEqual(self._written(device), ["a", "h", "a", "8"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual( + write.await_args_list, + [call(b"a\x03"), call(b"h\x03"), call(b"a\x03"), call(b"8\x03")], + ) + self.assertEqual(write.call_count, write.await_count) async def test_manual_halt_during_motion_fails_immediately(self): device = self._make( @@ -448,7 +487,12 @@ async def test_retry_decap_finishes_with_caps_held(self): ] ) await device.retry_decap() - self.assertEqual(self._written(device), ["a", "Q", "a", "a"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual( + write.await_args_list, + [call(b"a\x03"), call(b"Q\x03"), call(b"a\x03"), call(b"a\x03")], + ) + self.assertEqual(write.call_count, write.await_count) async def test_recap_blocked_when_not_decapped(self): device = self._make([status("StatusDecap")]) @@ -465,13 +509,20 @@ async def test_recap_success(self): ] ) await device.recap() - self.assertEqual(self._written(device), ["a", "i", "a", "a"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual( + write.await_args_list, + [call(b"a\x03"), call(b"i\x03"), call(b"a\x03"), call(b"a\x03")], + ) + self.assertEqual(write.call_count, write.await_count) async def test_waste_blocked_when_no_caps_are_held(self): device = self._make([status("StatusOK")]) with self.assertRaises(FluidXError): await device.waste() - self.assertEqual(self._written(device), ["a"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual(write.await_args_list, [call(b"a\x03")]) + self.assertEqual(write.call_count, write.await_count) async def test_waste_success(self): device = self._make( @@ -483,24 +534,35 @@ async def test_waste_success(self): ] ) await device.waste() - self.assertEqual(self._written(device), ["a", "b", "a", "a"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual( + write.await_args_list, + [call(b"a\x03"), call(b"b\x03"), call(b"a\x03"), call(b"a\x03")], + ) + self.assertEqual(write.call_count, write.await_count) # === Standby === async def test_standby_reaches_sleep(self): device = self._make([status("StatusOK"), [ACK, "jOK"], status("StatusSLEEP")]) await device.standby() - self.assertEqual(self._written(device), ["a", "j", "a"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual(write.await_args_list, [call(b"a\x03"), call(b"j\x03"), call(b"a\x03")]) + self.assertEqual(write.call_count, write.await_count) async def test_standby_is_noop_when_asleep(self): device = self._make([status("StatusSLEEP")]) await device.standby() - self.assertEqual(self._written(device), ["a"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual(write.await_args_list, [call(b"a\x03")]) + self.assertEqual(write.call_count, write.await_count) async def test_ready_noop_when_awake(self): device = self._make([status("StatusOK")]) await device.ready() - self.assertEqual(self._written(device), ["a"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual(write.await_args_list, [call(b"a\x03")]) + self.assertEqual(write.call_count, write.await_count) async def test_ready_wakes_from_sleep(self): device = self._make( @@ -512,7 +574,12 @@ async def test_ready_wakes_from_sleep(self): ] ) await device.ready() - self.assertEqual(self._written(device), ["a", "k", "a", "a"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual( + write.await_args_list, + [call(b"a\x03"), call(b"k\x03"), call(b"a\x03"), call(b"a\x03")], + ) + self.assertEqual(write.call_count, write.await_count) # === Cartridge === @@ -526,35 +593,50 @@ async def test_eject_cartridge(self): ] ) await device.eject_cartridge() - self.assertEqual(self._written(device), ["e", "a", "c", "a"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual( + write.await_args_list, + [call(b"e\x03"), call(b"a\x03"), call(b"c\x03"), call(b"a\x03")], + ) + self.assertEqual(write.call_count, write.await_count) async def test_eject_cartridge_is_noop_without_a_cartridge(self): device = self._make([extended(cartridge_installed=False)]) await device.eject_cartridge() - self.assertEqual(self._written(device), ["e"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual(write.await_args_list, [call(b"e\x03")]) + self.assertEqual(write.call_count, write.await_count) async def test_eject_cartridge_blocked_while_caps_are_held(self): device = self._make([extended(cartridge_installed=True, caps_on_pins=True)]) with self.assertRaises(FluidXError): await device.eject_cartridge() - self.assertEqual(self._written(device), ["e"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual(write.await_args_list, [call(b"e\x03")]) + self.assertEqual(write.call_count, write.await_count) async def test_load_cartridge_returns_the_profile(self): device = self._make( [extended(cartridge_installed=False), [ACK, "COK", "o16OK"], status("StatusOK")] ) self.assertEqual(await device.load_cartridge(), 16) - self.assertEqual(self._written(device), ["e", "C", "a"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual(write.await_args_list, [call(b"e\x03"), call(b"C\x03"), call(b"a\x03")]) + self.assertEqual(write.call_count, write.await_count) async def test_load_cartridge_is_noop_when_already_loaded(self): device = self._make([extended(cartridge_installed=True)]) self.assertIsNone(await device.load_cartridge()) - self.assertEqual(self._written(device), ["e"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual(write.await_args_list, [call(b"e\x03")]) + self.assertEqual(write.call_count, write.await_count) async def test_reset_cartridge_counter(self): device = self._make([[ACK, "XOK", "CarResetDONE"]]) await device.reset_cartridge_counter() - self.assertEqual(self._written(device), ["X"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual(write.await_args_list, [call(b"X\x03")]) + self.assertEqual(write.call_count, write.await_count) async def test_reset_cartridge_counter_requires_its_answer(self): device = self._make([[ACK, "XOK"]]) @@ -564,7 +646,9 @@ async def test_reset_cartridge_counter_requires_its_answer(self): async def test_reset_cartridge_counter_is_idempotent(self): device = self._make([[ACK, "XOK", "CommandIgnore"]]) await device.reset_cartridge_counter() - self.assertEqual(self._written(device), ["X"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual(write.await_args_list, [call(b"X\x03")]) + self.assertEqual(write.call_count, write.await_count) async def test_load_cartridge_reads_the_profile_from_the_end_of_the_motion(self): # onnOK is documented as an end-of-motion answer, not part of the ack. @@ -590,32 +674,44 @@ async def test_load_cartridge_reports_a_missing_profile(self): async def test_set_error_detection_off_is_not_read_as_a_fault(self): device = self._make([[ACK, "lOK", "ErrorDetectOFF"]]) await device.set_error_detection_enabled(False) - self.assertEqual(self._written(device), ["l"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual(write.await_args_list, [call(b"l\x03")]) + self.assertEqual(write.call_count, write.await_count) async def test_set_error_detection_on(self): device = self._make([[ACK, "LOK", "ErrorDetectON"]]) await device.set_error_detection_enabled(True) - self.assertEqual(self._written(device), ["L"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual(write.await_args_list, [call(b"L\x03")]) + self.assertEqual(write.call_count, write.await_count) async def test_set_error_detection_is_idempotent(self): device = self._make([[ACK, "LOK", "CommandIgnore"]]) await device.set_error_detection_enabled(True) - self.assertEqual(self._written(device), ["L"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual(write.await_args_list, [call(b"L\x03")]) + self.assertEqual(write.call_count, write.await_count) async def test_set_dry_run_enabled(self): device = self._make([[ACK, "dOK", "DryRunON"]]) await device.set_dry_run_enabled(True) - self.assertEqual(self._written(device), ["d"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual(write.await_args_list, [call(b"d\x03")]) + self.assertEqual(write.call_count, write.await_count) async def test_set_dry_run_disabled(self): device = self._make([[ACK, "DOK", "DryRunOFF"]]) await device.set_dry_run_enabled(False) - self.assertEqual(self._written(device), ["D"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual(write.await_args_list, [call(b"D\x03")]) + self.assertEqual(write.call_count, write.await_count) async def test_set_dry_run_is_idempotent(self): device = self._make([[ACK, "DOK", "CommandIgnore"]]) await device.set_dry_run_enabled(False) - self.assertEqual(self._written(device), ["D"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual(write.await_args_list, [call(b"D\x03")]) + self.assertEqual(write.call_count, write.await_count) async def test_setting_raises_when_the_device_does_not_confirm(self): device = self._make([[ACK, "dOK"]]) @@ -625,17 +721,23 @@ async def test_setting_raises_when_the_device_does_not_confirm(self): async def test_set_safety_door_enabled(self): device = self._make([[ACK, "+OK", "DoorONDONE"]]) await device.set_safety_door_enabled(True) - self.assertEqual(self._written(device), ["+"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual(write.await_args_list, [call(b"+\x03")]) + self.assertEqual(write.call_count, write.await_count) async def test_set_safety_door_disabled(self): device = self._make([[ACK, "-OK", "DoorOFFDONE"]]) await device.set_safety_door_enabled(False) - self.assertEqual(self._written(device), ["-"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual(write.await_args_list, [call(b"-\x03")]) + self.assertEqual(write.call_count, write.await_count) async def test_set_safety_door_is_idempotent(self): device = self._make([[ACK, "+OK", "CommandIgnore"]]) await device.set_safety_door_enabled(True) - self.assertEqual(self._written(device), ["+"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual(write.await_args_list, [call(b"+\x03")]) + self.assertEqual(write.call_count, write.await_count) # === Manual recovery commands === @@ -643,14 +745,18 @@ async def test_eject_caps_requires_manual_mode(self): device = self._make([status("StatusOK")]) with self.assertRaises(FluidXError): await device.eject_caps() - self.assertEqual(self._written(device), ["a"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual(write.await_args_list, [call(b"a\x03")]) + self.assertEqual(write.call_count, write.await_count) async def test_eject_caps(self): # Manual recovery pins the status word to StatusMANUAL, so the answer frame # is the only completion signal: no status polling here. device = self._make([status("StatusMANUAL"), [ACK, "5OK", "EjectDONE"]]) await device.eject_caps() - self.assertEqual(self._written(device), ["a", "5"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual(write.await_args_list, [call(b"a\x03"), call(b"5\x03")]) + self.assertEqual(write.call_count, write.await_count) async def test_eject_caps_reports_a_failure(self): device = self._make([status("StatusMANUAL"), [ACK, "5OK", "EjectERROR"], query("8", "133")]) @@ -661,12 +767,16 @@ async def test_eject_caps_reports_a_failure(self): async def test_head_up(self): device = self._make([status("StatusMANUAL"), [ACK, "6OK", "HeadDONE"]]) await device.head_up() - self.assertEqual(self._written(device), ["a", "6"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual(write.await_args_list, [call(b"a\x03"), call(b"6\x03")]) + self.assertEqual(write.call_count, write.await_count) async def test_open_safety_door(self): device = self._make([status("StatusMANUAL"), [ACK, "7OK", "DoorDONE"]]) await device.open_safety_door() - self.assertEqual(self._written(device), ["a", "7"]) + write = cast(AsyncMock, device.io.write) + self.assertEqual(write.await_args_list, [call(b"a\x03"), call(b"7\x03")]) + self.assertEqual(write.call_count, write.await_count) async def test_manual_command_times_out_without_an_answer(self): device = self._make([status("StatusMANUAL"), [ACK, "6OK"]]) diff --git a/pylabrobot/generic/line_barcode_scanner/serial_tests.py b/pylabrobot/generic/line_barcode_scanner/serial_tests.py index a783d0d736d..97cecd82000 100644 --- a/pylabrobot/generic/line_barcode_scanner/serial_tests.py +++ b/pylabrobot/generic/line_barcode_scanner/serial_tests.py @@ -1,93 +1,48 @@ import unittest -from typing import List +from typing import cast +from unittest.mock import AsyncMock, call from pylabrobot.generic import SerialBarcodeScanner +from pylabrobot.io.testing import fake_serial -class FakeSerialIO: - def __init__(self, chunks: List[bytes]): - self.chunks = chunks - self.writes: List[bytes] = [] - self.port = "COM_TEST" - self.timeout: float = 1 - self.setup_called = False - self.stop_called = False - self.reset_input_buffer_called = False - - async def setup(self): - self.setup_called = True - - async def stop(self): - self.stop_called = True - - async def read(self, num_bytes: int = 1) -> bytes: - del num_bytes - if len(self.chunks) == 0: - return b"" - return self.chunks.pop(0) - - async def write(self, data: bytes): - self.writes.append(data) - - async def reset_input_buffer(self): - self.reset_input_buffer_called = True - - def get_read_timeout(self) -> float: - return self.timeout - - def set_read_timeout(self, timeout: float) -> None: - self.timeout = timeout - - def temporary_timeout(self, timeout: float): - fake = self - - class TemporaryTimeout: - def __enter__(self): - self.original_timeout = fake.timeout - fake.timeout = timeout - - def __exit__(self, exc_type, exc_value, traceback): - fake.timeout = self.original_timeout - - return TemporaryTimeout() - - -def make_scanner(chunks: List[bytes]) -> SerialBarcodeScanner: +def make_scanner(incoming: bytes) -> SerialBarcodeScanner: + """Create a scanner with the supplied bytes waiting on its serial transport.""" scanner = SerialBarcodeScanner(port="COM_TEST") - scanner.io = FakeSerialIO(chunks) # type: ignore[assignment] + scanner.io = fake_serial(incoming=incoming, port="COM_TEST") # type: ignore[assignment] return scanner class TestSerialBarcodeScanner(unittest.IsolatedAsyncioTestCase): async def test_read_line_carriage_return(self): - scanner = make_scanner([b"1", b"2", b"3", b"\r"]) + scanner = make_scanner(b"123\r") self.assertEqual(await scanner.read_line(timeout=1), "123") async def test_read_line_newline(self): - scanner = make_scanner([b"A", b"B", b"C", b"\n"]) + scanner = make_scanner(b"ABC\n") self.assertEqual(await scanner.read_line(timeout=1), "ABC") async def test_read_line_timeout_before_data(self): - scanner = make_scanner([]) + scanner = make_scanner(b"") self.assertEqual(await scanner.read_line(timeout=0), "") async def test_read_line_rejects_negative_timeout(self): - scanner = make_scanner([]) + scanner = make_scanner(b"") with self.assertRaises(ValueError): await scanner.read_line(timeout=-1) async def test_reset_input_buffer(self): - scanner = make_scanner([]) + scanner = make_scanner(b"") await scanner.reset_input_buffer() - fake_io = scanner.io - assert isinstance(fake_io, FakeSerialIO) - self.assertTrue(fake_io.reset_input_buffer_called) + reset_input_buffer = cast(AsyncMock, scanner.io.reset_input_buffer) + reset_input_buffer.assert_awaited_once_with() + self.assertEqual(reset_input_buffer.call_count, reset_input_buffer.await_count) def test_rejects_empty_terminators(self): with self.assertRaises(ValueError): @@ -102,8 +57,7 @@ def test_rejects_non_positive_max_line_length(self): SerialBarcodeScanner(port="COM_TEST", max_line_length=0) async def test_scan_barcode(self): - scanner = SerialBarcodeScanner(port="COM_TEST") - scanner.io = FakeSerialIO([b"2", b"2", b"6", b"\r"]) # type: ignore[assignment] + scanner = make_scanner(b"226\r") barcode = await scanner.scan_barcode( read_time=1, @@ -117,7 +71,7 @@ async def test_scan_barcode(self): self.assertEqual(barcode.position_on_resource, "right") async def test_scan_barcode_returns_none_on_timeout(self): - scanner = make_scanner([]) + scanner = make_scanner(b"") self.assertIsNone(await scanner.scan_barcode(read_time=0)) @@ -127,25 +81,25 @@ async def test_scan_barcode_with_trigger_command(self): trigger_command=b"TRIGGER\r", untrigger_command=b"UNTRIGGER\r", ) - scanner.io = FakeSerialIO([b"1", b"2", b"3", b"\r"]) # type: ignore[assignment] + scanner.io = fake_serial(incoming=b"123\r", port="COM_TEST") # type: ignore[assignment] barcode = await scanner.scan_barcode(read_time=1) assert barcode is not None self.assertEqual(barcode.data, "123") - fake_io = scanner.io - assert isinstance(fake_io, FakeSerialIO) - self.assertEqual(fake_io.writes, [b"TRIGGER\r", b"UNTRIGGER\r"]) + write = cast(AsyncMock, scanner.io.write) + self.assertEqual(write.await_args_list, [call(b"TRIGGER\r"), call(b"UNTRIGGER\r")]) + self.assertEqual(write.call_count, write.await_count) async def test_scan_barcode_rejects_negative_read_time(self): - scanner = make_scanner([]) + scanner = make_scanner(b"") with self.assertRaises(ValueError): await scanner.scan_barcode(read_time=-1) async def test_setup_scan_stop(self): scanner = SerialBarcodeScanner(port="COM_TEST") - fake_io = FakeSerialIO([b"X", b"Y", b"Z", b"\r"]) + fake_io = fake_serial(incoming=b"XYZ\r", port="COM_TEST") scanner.io = fake_io # type: ignore[assignment] await scanner.setup() @@ -156,8 +110,10 @@ async def test_setup_scan_stop(self): self.assertEqual(barcode.data, "XYZ") self.assertEqual(barcode.symbology, "Code 39") self.assertEqual(barcode.position_on_resource, "bottom") - self.assertTrue(fake_io.setup_called) - self.assertTrue(fake_io.stop_called) + fake_io.setup.assert_awaited_once_with() + self.assertEqual(fake_io.setup.call_count, fake_io.setup.await_count) + fake_io.stop.assert_awaited_once_with() + self.assertEqual(fake_io.stop.call_count, fake_io.stop.await_count) if __name__ == "__main__": diff --git a/pylabrobot/io/testing.py b/pylabrobot/io/testing.py new file mode 100644 index 00000000000..9cad7a754da --- /dev/null +++ b/pylabrobot/io/testing.py @@ -0,0 +1,107 @@ +"""Transport doubles for driver tests using native unittest.mock assertions.""" + +from contextlib import contextmanager +from typing import Callable, Iterator, Optional, cast +from unittest.mock import Mock, NonCallableMagicMock, create_autospec + +from pylabrobot.io.serial import Serial + + +def fake_serial( + *, + incoming: bytes = b"", + on_write: Optional[Callable[[bytes], Optional[bytes]]] = None, + port: str = "FAKE", + read_timeout: float = 1.0, +) -> NonCallableMagicMock: + """Create a buffered Serial mock without constructing or opening a transport. + + Reads consume at most the requested number of bytes and return immediately, even when the + buffer is empty. Non-positive read sizes return empty bytes without consuming input, matching + pyserial. Setup and stop are async no-ops; timeouts are fixture-local values, without timing + or connection-state simulation. Protocol encoding and response framing belong in the test. + + Args: + incoming: Bytes already buffered when the fixture is created. + on_write: Synchronous callback invoked when a write is awaited. Its returned bytes are + appended to the receive buffer; None appends nothing. Exceptions propagate to the caller. + port: Value exposed by the mock's port property. + read_timeout: Initial value returned by get_read_timeout(). + + Returns: + An autospecced Serial instance mock with native call and await assertions. Public methods + outside read, write, setup, stop, reset_input_buffer and the three timeout methods raise + NotImplementedError. To configure one, replace its side_effect, or clear side_effect before + setting return_value (for example, io.readline.side_effect = None). + """ + io = cast(NonCallableMagicMock, create_autospec(Serial, instance=True, spec_set=True)) + rx = bytearray(incoming) + + # Serial also exposes file methods inherited from the standard library's IOBase. + unsupported_methods: dict[str, Mock] = { + "readline": io.readline, + "send_break": io.send_break, + "reset_output_buffer": io.reset_output_buffer, + "serialize": io.serialize, + "close": io.close, + "fileno": io.fileno, + "flush": io.flush, + "isatty": io.isatty, + "readable": io.readable, + "readlines": io.readlines, + "seek": io.seek, + "seekable": io.seekable, + "tell": io.tell, + "truncate": io.truncate, + "writable": io.writable, + "writelines": io.writelines, + } + for name, method in unsupported_methods.items(): + method.side_effect = NotImplementedError( + f"fake_serial does not support Serial.{name}(); configure its side_effect explicitly." + ) + + def read(num_bytes: int = 1) -> bytes: + """Consume at most num_bytes from the front of the receive buffer.""" + if num_bytes <= 0: + return b"" + data = bytes(rx[:num_bytes]) + del rx[:num_bytes] + return data + + def write(data: bytes) -> None: + """Append a test-local response after the AsyncMock records the awaited write.""" + if on_write is not None: + response = on_write(data) + if response is not None: + rx.extend(response) + + def get_read_timeout() -> float: + """Return the fixture-local read timeout.""" + return read_timeout + + def set_read_timeout(timeout: float) -> None: + """Update the fixture-local read timeout without simulating elapsed time.""" + nonlocal read_timeout + read_timeout = timeout + + @contextmanager + def temporary_timeout(timeout: float) -> Iterator[None]: + """Restore the previous timeout on exit, including when the block raises.""" + original = get_read_timeout() + set_read_timeout(timeout) + try: + yield + finally: + set_read_timeout(original) + + io.port = port + io.setup.return_value = None + io.stop.return_value = None + io.read.side_effect = read + io.write.side_effect = write + io.reset_input_buffer.side_effect = rx.clear + io.get_read_timeout.side_effect = get_read_timeout + io.set_read_timeout.side_effect = set_read_timeout + io.temporary_timeout.side_effect = temporary_timeout + return io diff --git a/pylabrobot/io/testing_tests.py b/pylabrobot/io/testing_tests.py new file mode 100644 index 00000000000..dcb92899765 --- /dev/null +++ b/pylabrobot/io/testing_tests.py @@ -0,0 +1,218 @@ +"""Tests for buffered Serial doubles used by driver tests.""" + +import unittest +from unittest.mock import AsyncMock, Mock, call, patch + +from pylabrobot.io.serial import Serial +from pylabrobot.io.testing import fake_serial + + +class TestFakeSerial(unittest.IsolatedAsyncioTestCase): + """Check transport behavior and native mock assertions without hardware.""" + + async def test_read_slices_and_retains_remainder(self) -> None: + """Reads consume only the requested prefix, including the default size.""" + io = fake_serial(incoming=b"ABCDE") + self.assertEqual(await io.read(), b"A") + self.assertEqual(await io.read(num_bytes=2), b"BC") + self.assertEqual(await io.read(10), b"DE") + self.assertEqual(await io.read(1), b"") + + async def test_non_positive_reads_leave_buffer_unchanged(self) -> None: + """Pyserial's Windows and POSIX reads return empty for non-positive sizes.""" + io = fake_serial(incoming=b"ABC") + self.assertEqual(await io.read(0), b"") + self.assertEqual(await io.read(-1), b"") + self.assertEqual(await io.read(3), b"ABC") + + async def test_empty_buffer_returns_immediately(self) -> None: + """An empty read does not schedule a sleep or wait for the configured timeout.""" + io = fake_serial(read_timeout=60) + with patch("asyncio.sleep", side_effect=AssertionError("unexpected sleep")): + self.assertEqual(await io.read(1), b"") + + async def test_instances_have_independent_state(self) -> None: + """Buffer contents, timeout values and call histories belong to one fixture.""" + first = fake_serial(incoming=b"ABC", read_timeout=2) + second = fake_serial(incoming=b"XYZ", read_timeout=3) + await first.read(1) + await first.reset_input_buffer() + first.set_read_timeout(4) + self.assertEqual(await first.read(3), b"") + self.assertEqual(await second.read(3), b"XYZ") + self.assertEqual(second.get_read_timeout(), 3) + second.reset_input_buffer.assert_not_called() + + async def test_write_appends_callback_response_when_awaited(self) -> None: + """The callback sees the awaited write and appends behind unread input.""" + callback = Mock(side_effect=[b"BC", b"DE"]) + io = fake_serial(incoming=b"A", on_write=callback) + self.assertIsNone(await io.write(b"first")) + self.assertEqual(io.write.await_args_list, [call(b"first")]) + callback.assert_called_once_with(b"first") + self.assertEqual(await io.read(2), b"AB") + self.assertIsNone(await io.write(b"second")) + self.assertEqual(await io.read(10), b"CDE") + + async def test_write_without_response_returns_none(self) -> None: + """An absent callback and a callback returning None both leave input alone.""" + for callback in (None, Mock(return_value=None)): + with self.subTest(callback=callback): + io = fake_serial(incoming=b"A", on_write=callback) + self.assertIsNone(await io.write(b"command")) + self.assertEqual(await io.read(10), b"A") + io.write.assert_awaited_once_with(b"command") + if callback is not None: + callback.assert_called_once_with(b"command") + + async def test_callback_exception_propagates_after_write_is_recorded(self) -> None: + """Callback failures retain their identity and the native await history.""" + error = ValueError("invalid command") + io = fake_serial(incoming=b"A", on_write=Mock(side_effect=error)) + with self.assertRaises(ValueError) as caught: + await io.write(b"bad") + self.assertIs(caught.exception, error) + io.write.assert_awaited_once_with(b"bad") + self.assertEqual(await io.read(10), b"A") + + async def test_reset_clears_current_input_but_preserves_future_responses(self) -> None: + """Reset discards queued bytes without replacing the response callback.""" + io = fake_serial(incoming=b"stale", on_write=Mock(return_value=b"reply")) + await io.write(b"first") + self.assertIsNone(await io.reset_input_buffer()) + self.assertEqual(await io.read(10), b"") + await io.write(b"second") + self.assertEqual(await io.read(10), b"reply") + + def test_timeout_getter_and_setter(self) -> None: + """Synchronous timeout access uses the fixture's current value.""" + io = fake_serial() + self.assertEqual(io.get_read_timeout(), 1.0) + self.assertIsNone(io.set_read_timeout(0.25)) + self.assertEqual(io.get_read_timeout(), 0.25) + + def test_temporary_timeout_restores_nested_values(self) -> None: + """Each context restores the value that was active on entry.""" + io = fake_serial(read_timeout=5) + with io.temporary_timeout(2): + self.assertEqual(io.get_read_timeout(), 2) + with io.temporary_timeout(0.5): + self.assertEqual(io.get_read_timeout(), 0.5) + self.assertEqual(io.get_read_timeout(), 2) + self.assertEqual(io.get_read_timeout(), 5) + + def test_temporary_timeout_restores_after_exception(self) -> None: + """An exception inside the context is propagated after timeout restoration.""" + io = fake_serial(read_timeout=5) + with self.assertRaisesRegex(ValueError, "failure"): + with io.temporary_timeout(0.5): + io.set_read_timeout(0.1) + raise ValueError("failure") + self.assertEqual(io.get_read_timeout(), 5) + + async def test_autospec_shape_and_lifecycle_without_construction(self) -> None: + """The concrete class supplies the spec without constructing a transport.""" + with patch.object(Serial, "__init__", side_effect=AssertionError("real constructor")): + io = fake_serial(port="COM_TEST") + self.assertIsNone(await io.setup()) + self.assertIsNone(await io.stop()) + self.assertIsInstance(io, Serial) + self.assertEqual(io.port, "COM_TEST") + self.assertEqual(fake_serial().port, "FAKE") + for method in (io.read, io.write, io.setup, io.stop, io.reset_input_buffer, io.readline): + self.assertIsInstance(method, AsyncMock) + for method in (io.get_read_timeout, io.set_read_timeout, io.temporary_timeout, io.serialize): + self.assertIsInstance(method, Mock) + self.assertNotIsInstance(method, AsyncMock) + with self.assertRaises(TypeError): + io.read(size=1) + with self.assertRaises(TypeError): + io.write() + with self.assertRaises(TypeError): + io.set_read_timeout() + with self.assertRaises(AttributeError): + io.missing_method() + with self.assertRaises(AttributeError): + io.missing_attribute = 1 + + async def test_unsupported_public_methods_raise_named_errors(self) -> None: + """Unconfigured Serial and inherited file methods cannot return silent mocks.""" + io = fake_serial() + for name, method, args in ( + ("readline", io.readline, ()), + ("send_break", io.send_break, (0.1,)), + ("reset_output_buffer", io.reset_output_buffer, ()), + ): + with self.subTest(method=name): + with self.assertRaisesRegex(NotImplementedError, "Serial." + name): + await method(*args) + for name, method, sync_args in ( + ("serialize", io.serialize, ()), + ("close", io.close, ()), + ("fileno", io.fileno, ()), + ("flush", io.flush, ()), + ("isatty", io.isatty, ()), + ("readable", io.readable, ()), + ("readlines", io.readlines, ()), + ("seek", io.seek, (0,)), + ("seekable", io.seekable, ()), + ("tell", io.tell, ()), + ("truncate", io.truncate, ()), + ("writable", io.writable, ()), + ("writelines", io.writelines, ([],)), + ): + with self.subTest(method=name): + with self.assertRaisesRegex(NotImplementedError, "Serial." + name): + method(*sync_args) + + async def test_unsupported_method_can_be_configured_with_native_mock_api(self) -> None: + """Tests can replace the default failure using side_effect or return_value.""" + io = fake_serial() + io.readline.side_effect = None + io.readline.return_value = b"line\n" + self.assertEqual(await io.readline(), b"line\n") + io.readline.assert_awaited_once_with() + io.serialize.side_effect = lambda: {"port": "test"} + self.assertEqual(io.serialize(), {"port": "test"}) + + async def test_exact_write_assertions_detect_command_regressions(self) -> None: + """Full await lists detect wrong bytes, terminators, extra and missing writes.""" + expected = [call(b"?\r"), call(b"A100\r")] + for commands in ( + [b"?\r", b"A100\r"], + [b"!\r", b"A100\r"], + [b"?\n", b"A100\r"], + [b"?\r", b"A100\r", b"S\r"], + [b"?\r"], + ): + with self.subTest(commands=commands): + io = fake_serial(incoming=b"reply") + for command in commands: + await io.write(command) + await io.read(2) + await io.read(10) + self.assertEqual(io.write.call_count, io.write.await_count) + if commands == [b"?\r", b"A100\r"]: + self.assertEqual(io.write.await_args_list, expected) + else: + with self.assertRaises(AssertionError): + self.assertEqual(io.write.await_args_list, expected) + + async def test_called_but_unawaited_write_is_detectable(self) -> None: + """Calling write records a call but neither invokes the callback nor buffers bytes.""" + callback = Mock(return_value=b"reply") + io = fake_serial(on_write=callback) + pending = io.write(b"command") + try: + callback.assert_not_called() + io.write.assert_called_once_with(b"command") + io.write.assert_not_awaited() + self.assertEqual(await io.read(10), b"") + with self.assertRaises(AssertionError): + self.assertEqual(io.write.call_count, io.write.await_count) + finally: + pending.close() + + +if __name__ == "__main__": + unittest.main() diff --git a/pylabrobot/kbiosystems/sealer_tests.py b/pylabrobot/kbiosystems/sealer_tests.py index c100bf96201..f1071ad9691 100644 --- a/pylabrobot/kbiosystems/sealer_tests.py +++ b/pylabrobot/kbiosystems/sealer_tests.py @@ -1,7 +1,8 @@ import unittest -from typing import Dict, List -from unittest.mock import AsyncMock, patch +from typing import Dict, Optional, cast +from unittest.mock import AsyncMock, NonCallableMagicMock, call, patch +from pylabrobot.io.testing import fake_serial from pylabrobot.kbiosystems import ( KBiosystemsError, KBiosystemsUltrasealEPRO, @@ -13,41 +14,17 @@ ) -class FakeSealerSerial: - """In-memory serial stand-in that mimics the sealer's echo protocol. +def sealer_io(responses: Dict[str, str]) -> NonCallableMagicMock: + """Create a serial transport that echoes commands before their configured replies.""" - On ``write(command + "\\r")`` it records the command and, if a reply body is - configured for it, queues ``command + body + "\\r"`` to be read back - exactly - like the device, which echoes the command in front of its reply. - """ - - def __init__(self, responses: Dict[str, str]) -> None: - self.responses = responses - self.port = "FAKE" - self.written: List[str] = [] - self._rx = bytearray() - - async def setup(self) -> None: - pass - - async def stop(self) -> None: - pass - - async def reset_input_buffer(self) -> None: - self._rx.clear() - - async def write(self, data: bytes) -> None: + def reply(data: bytes) -> Optional[bytes]: + """Return an echoed ASCII reply, or no response for an unmapped command.""" command = data.decode("ascii").rstrip("\r") - self.written.append(command) - if command in self.responses: - self._rx += (command + self.responses[command] + "\r").encode("ascii") + if command in responses: + return (command + responses[command] + "\r").encode("ascii") + return None - async def read(self, num_bytes: int = 1) -> bytes: - if not self._rx: - return b"" - out = bytes(self._rx[:num_bytes]) - del self._rx[:num_bytes] - return out + return fake_serial(on_write=reply) class KBiosystemsSealerTestBase(unittest.IsolatedAsyncioTestCase): @@ -60,17 +37,20 @@ async def asyncSetUp(self) -> None: class TestUltrasealEPRO(KBiosystemsSealerTestBase): def _make(self, responses: Dict[str, str]) -> KBiosystemsUltrasealEPRO: + """Create an ePRO with configured serial replies.""" sealer = KBiosystemsUltrasealEPRO(port="FAKE") - sealer.io = FakeSealerSerial(responses) # type: ignore[assignment] + sealer.io = sealer_io(responses) # type: ignore[assignment] return sealer async def test_setup_sequence(self): sealer = self._make({"I": "ok", "?": "00", "V": "1.2\rboardA", "A100": "ok"}) await sealer.setup() self.assertEqual(sealer.firmware_version, "1.2 | boardA") - self.assertIn("I", sealer.io.written) # type: ignore[attr-defined] - self.assertIn("V", sealer.io.written) # type: ignore[attr-defined] - self.assertIn("A100", sealer.io.written) # type: ignore[attr-defined] + write = cast(AsyncMock, sealer.io.write) + self.assertEqual( + write.await_args_list, [call(b"I\r"), call(b"?\r"), call(b"V\r"), call(b"A100\r")] + ) + self.assertEqual(write.call_count, write.await_count) async def test_seal_distance_mode_wire_bytes(self): sealer = self._make( @@ -87,11 +67,24 @@ async def test_seal_distance_mode_wire_bytes(self): } ) await sealer.seal(temperature=170, duration=2.5) - written = sealer.io.written # type: ignore[attr-defined] - for expected in ["ECO_OFF", "B25", "A170", "L=120", "FS=0", "DO=25", "S", "A100"]: - self.assertIn(expected, written) - # Distance mode must not touch the force commands. - self.assertFalse(any(w.startswith("PS=") or w == "FS=1" for w in written)) + write = cast(AsyncMock, sealer.io.write) + self.assertEqual( + write.await_args_list, + [ + call(b"ECO_OFF\r"), + call(b"?\r"), + call(b"B25\r"), + call(b"A170\r"), + call(b"L=120\r"), + call(b"FS=0\r"), + call(b"DO=25\r"), + call(b"?\r"), + call(b"S\r"), + call(b"?\r"), + call(b"A100\r"), + ], + ) + self.assertEqual(write.call_count, write.await_count) async def test_seal_force_mode_wire_bytes(self): sealer = self._make( @@ -107,10 +100,24 @@ async def test_seal_force_mode_wire_bytes(self): } ) await sealer.seal(temperature=170, duration=2.5, force_mode=True, sealing_force=30) - written = sealer.io.written # type: ignore[attr-defined] - self.assertIn("FS=1", written) - self.assertIn("PS=30", written) # sent with no reply - self.assertFalse(any(w.startswith("DO=") for w in written)) + write = cast(AsyncMock, sealer.io.write) + self.assertEqual( + write.await_args_list, + [ + call(b"ECO_OFF\r"), + call(b"?\r"), + call(b"B25\r"), + call(b"A170\r"), + call(b"L=120\r"), + call(b"FS=1\r"), + call(b"PS=30\r"), + call(b"?\r"), + call(b"S\r"), + call(b"?\r"), + call(b"A100\r"), + ], + ) + self.assertEqual(write.call_count, write.await_count) async def test_status_decode(self): sealer = self._make({"?": "a4"}) @@ -129,23 +136,35 @@ async def test_temperature_floor_is_5(self): class TestUltrasealXTPro(KBiosystemsSealerTestBase): def _make(self, responses: Dict[str, str]) -> KBiosystemsUltrasealXTPro: + """Create an XT Pro with configured serial replies.""" sealer = KBiosystemsUltrasealXTPro(port="FAKE") - sealer.io = FakeSealerSerial(responses) # type: ignore[assignment] + sealer.io = sealer_io(responses) # type: ignore[assignment] return sealer async def test_setup_sequence(self): sealer = self._make({"?": "00", "A100": "ok"}) await sealer.setup() - self.assertEqual(sealer.io.written, ["?", "A100"]) # type: ignore[attr-defined] + write = cast(AsyncMock, sealer.io.write) + self.assertEqual(write.await_args_list, [call(b"?\r"), call(b"A100\r")]) + self.assertEqual(write.call_count, write.await_count) async def test_seal_wire_bytes(self): sealer = self._make({"?": "00", "B30": "ok", "A180": "ok", "A100": "ok", "S": "ok"}) await sealer.seal(temperature=180, duration=3.0) - written = sealer.io.written # type: ignore[attr-defined] - for expected in ["B30", "A180", "S", "A100"]: - self.assertIn(expected, written) - # The XT Pro has no foil/force/distance/eco commands. - self.assertFalse(any(w.startswith(("L=", "DO=", "PS=", "FS=", "ECO_")) for w in written)) + write = cast(AsyncMock, sealer.io.write) + self.assertEqual( + write.await_args_list, + [ + call(b"?\r"), + call(b"B30\r"), + call(b"A180\r"), + call(b"?\r"), + call(b"S\r"), + call(b"?\r"), + call(b"A100\r"), + ], + ) + self.assertEqual(write.call_count, write.await_count) async def test_shuttle_commands(self): sealer = self._make({"P": "ok", "U": "ok", "R": "ok"}) @@ -174,27 +193,35 @@ async def test_error_status_raises_with_code(self): class TestUltrasealPRO(KBiosystemsSealerTestBase): def _make(self, responses: Dict[str, str]) -> KBiosystemsUltrasealPRO: + """Create a PRO with configured serial replies.""" sealer = KBiosystemsUltrasealPRO(port="FAKE") - sealer.io = FakeSealerSerial(responses) # type: ignore[assignment] + sealer.io = sealer_io(responses) # type: ignore[assignment] return sealer async def test_setup_sequence(self): sealer = self._make({"?": "00", "A100": "ok"}) await sealer.setup() - self.assertEqual(sealer.io.written, ["?", "A100"]) # type: ignore[attr-defined] + write = cast(AsyncMock, sealer.io.write) + self.assertEqual(write.await_args_list, [call(b"?\r"), call(b"A100\r")]) + self.assertEqual(write.call_count, write.await_count) async def test_seal_wire_bytes(self): sealer = self._make({"?": "00", "B30": "ok", "A180": "ok", "A100": "ok", "S": "ok"}) await sealer.seal(temperature=180, duration=3.0) - written = sealer.io.written # type: ignore[attr-defined] - for expected in ["B30", "A180", "S", "A100"]: - self.assertIn(expected, written) - # The Ultraseal PRO has no foil/force/distance/eco or shuttle commands. - self.assertFalse( - any( - w.startswith(("L=", "DO=", "PS=", "FS=", "ECO_")) or w in ("P", "U", "R") for w in written - ) + write = cast(AsyncMock, sealer.io.write) + self.assertEqual( + write.await_args_list, + [ + call(b"?\r"), + call(b"B30\r"), + call(b"A180\r"), + call(b"?\r"), + call(b"S\r"), + call(b"?\r"), + call(b"A100\r"), + ], ) + self.assertEqual(write.call_count, write.await_count) async def test_park_mode_is_bit_7(self): # The Ultraseal PRO reports Park Mode at 0x80 (bit 6 is spare) - unlike the