From 680af2cf18924fc5f7edf2c28f3288aedf416681 Mon Sep 17 00:00:00 2001 From: matejcik Date: Thu, 11 Apr 2024 13:49:27 +0200 Subject: [PATCH] fix(tests): restore test functionality on legacy The global layout related changes were wrong for T1 where debuglink behavior is significantly different; in particular, it is not always possible to communicate over debuglink. This change reverts to the old behavior for T1B1 and keeps the new one only for core-based models. --- python/src/trezorlib/debuglink.py | 50 ++++++++++++++++++++++----- tests/device_tests/test_busy_state.py | 28 +++++++++++++-- 2 files changed, 68 insertions(+), 10 deletions(-) diff --git a/python/src/trezorlib/debuglink.py b/python/src/trezorlib/debuglink.py index 677e8aeb2..b29cf0221 100644 --- a/python/src/trezorlib/debuglink.py +++ b/python/src/trezorlib/debuglink.py @@ -624,7 +624,7 @@ class DebugLink: wait_type = DebugWaitType.IMMEDIATE else: wait_type = self.input_wait_type - return self.snapshot(wait_type) + return self._snapshot_core(wait_type) press_yes = _make_input_func(button=messages.DebugButton.YES) """Confirm current layout. See `_decision` for more details.""" @@ -667,10 +667,14 @@ class DebugLink: messages.DebugLinkDecision(x=x, y=y, hold_ms=hold_ms), wait ) - def snapshot( + def _snapshot_core( self, wait_type: DebugWaitType = DebugWaitType.IMMEDIATE ) -> LayoutContent: """Save text and image content of the screen to relevant directories.""" + # skip the snapshot if we are on T1 + if self.model is models.T1B1: + return LayoutContent([]) + # take the snapshot state = self.state(wait_type) layout = LayoutContent(state.tokens) @@ -678,8 +682,6 @@ class DebugLink: if state.tokens and self.layout_dirty: # save it, unless we already did or unless it's empty self.save_debug_screen(layout.visible_screen()) - if state.layout is not None: - self.save_screenshot(state.layout) self.layout_dirty = False # return the layout @@ -748,7 +750,16 @@ class DebugLink: def erase_sd_card(self, format: bool = True) -> messages.Success: return self._call(messages.DebugLinkEraseSdCard(format=format)) - def save_screenshot(self, data: bytes) -> None: + def snapshot_legacy(self) -> None: + """Snapshot the current state of the device.""" + if self.model is not models.T1B1: + return + + state = self.state() + if state.layout is not None: + self._save_screenshot_t1(state.layout) + + def _save_screenshot_t1(self, data: bytes) -> None: if self.t1_screenshot_directory is None: return @@ -833,7 +844,7 @@ class DebugUI: self.debuglink.press_yes() def button_request(self, br: messages.ButtonRequest) -> None: - self.debuglink.snapshot() + self.debuglink.snapshot_legacy() if self.input_flow is None: self._default_input_flow(br) @@ -847,7 +858,7 @@ class DebugUI: self.input_flow = self.INPUT_FLOW_DONE def get_pin(self, code: Optional["PinMatrixRequestType"] = None) -> str: - self.debuglink.snapshot() + self.debuglink.snapshot_legacy() if self.pins is None: raise RuntimeError("PIN requested but no sequence was configured") @@ -858,7 +869,7 @@ class DebugUI: raise AssertionError("PIN sequence ended prematurely") def get_passphrase(self, available_on_device: bool) -> str: - self.debuglink.snapshot() + self.debuglink.snapshot_legacy() return self.passphrase @@ -1254,6 +1265,29 @@ class TrezorClientDebugLink(TrezorClient): output.append(textwrap.indent(protobuf.format_message(act), " ")) raise AssertionError("\n".join(output)) + def sync_responses(self) -> None: + """Synchronize Trezor device receiving with caller. + + When a failed test does not read out the response, the next caller will write + a request, but read the previous response -- while the device had already sent + and placed into queue the new response. + + This function will call `Ping` and read responses until it locates a `Success` + with the expected text. This means that we are reading up-to-date responses. + """ + import secrets + + # Start by canceling whatever is on screen. This will work to cancel T1 PIN + # prompt, which is in TINY mode and does not respond to `Ping`. + # go to super() to avoid message filtering + super()._raw_write(messages.Cancel()) + + message = "SYNC" + secrets.token_hex(8) + super()._raw_write(messages.Ping(message=message)) + resp = None + while resp != messages.Success(message=message): + resp = super()._raw_read() + def mnemonic_callback(self, _) -> str: word, pos = self.debug.read_recovery_word() if word: diff --git a/tests/device_tests/test_busy_state.py b/tests/device_tests/test_busy_state.py index b8de77148..220903084 100644 --- a/tests/device_tests/test_busy_state.py +++ b/tests/device_tests/test_busy_state.py @@ -61,8 +61,8 @@ def test_busy_state(client: Client): assert client.features.unlocked is True -@pytest.mark.flaky(max_runs=5) -def test_busy_expiry(client: Client): +@pytest.mark.skip_t1b1 +def test_busy_expiry_core(client: Client): WAIT_TIME_MS = 1500 TOLERANCE = 1000 @@ -85,3 +85,27 @@ def test_busy_expiry(client: Client): # Also needs to come back to Homescreen (for UI tests). client.refresh_features() _assert_busy(client, False) + + +@pytest.mark.flaky(max_runs=5) +def test_busy_expiry_legacy(client: Client): + if client.model is not models.T1B1: + # TODO better skip markers + pytest.skip("Test only for T1B1") + + _assert_busy(client, False) + # Show the busy dialog. + device.set_busy(client, expiry_ms=1500) + _assert_busy(client, True) + + # Hasn't expired yet. + time.sleep(0.1) + _assert_busy(client, True) + + # Wait for it to expire. Add some tolerance to account for CI/hardware slowness. + time.sleep(4.0) + + # Check that the device is no longer busy. + # Also needs to come back to Homescreen (for UI tests). + client.refresh_features() + _assert_busy(client, False)