From 86fd93c1e3b0add80f4e98758fce11e84947cc50 Mon Sep 17 00:00:00 2001 From: andreasgriffin Date: Tue, 31 Mar 2026 15:22:44 +0200 Subject: [PATCH] squashed --- bitcoin_usb/jade_ble_client.py | 4 ++- bitcoin_usb/usb_gui.py | 22 ++++--------- tests/test_bluetooth_scan_support.py | 46 ++++++++++++++-------------- tests/test_jade_ble_client.py | 42 ++++++++++++++++++++++++- 4 files changed, 73 insertions(+), 41 deletions(-) diff --git a/bitcoin_usb/jade_ble_client.py b/bitcoin_usb/jade_ble_client.py index 1a81c9b..cb90014 100644 --- a/bitcoin_usb/jade_ble_client.py +++ b/bitcoin_usb/jade_ble_client.py @@ -12,6 +12,7 @@ import aioitertools import semver +from bitcoin_safe_lib.async_tools.loop_in_thread import LoopInThread from bleak import BleakScanner from hwilib.common import Chain from hwilib.devices.jade import HAS_NETWORKING, JadeClient @@ -59,9 +60,10 @@ def _extract_jade_serial_number(device_name: str) -> str | None: def discover_jade_ble_devices( + loop_in_thread: LoopInThread, scan_timeout: float = DEFAULT_DISCOVERY_SCAN_TIMEOUT_SECONDS, ) -> list[dict[str, Any]]: - devices = asyncio.run(BleakScanner.discover(timeout=max(1.0, scan_timeout))) + devices = loop_in_thread.run_foreground(BleakScanner.discover(timeout=max(1.0, scan_timeout))) discovered: list[dict[str, Any]] = [] seen_addresses: set[str] = set() diff --git a/bitcoin_usb/usb_gui.py b/bitcoin_usb/usb_gui.py index 1d07c88..f63f761 100644 --- a/bitcoin_usb/usb_gui.py +++ b/bitcoin_usb/usb_gui.py @@ -1,10 +1,8 @@ -import asyncio import logging import platform import re import tempfile from collections.abc import Callable -from concurrent.futures import ThreadPoolExecutor from functools import partial from pathlib import Path from typing import Any, TypeVar, cast @@ -38,20 +36,12 @@ def is_ble_available() -> bool: T = TypeVar("T") -def _run_ble_operation(operation: Callable[[], T]) -> T: - with ThreadPoolExecutor(max_workers=1, thread_name_prefix="ble") as executor: - return executor.submit(operation).result() - - -def can_scan_bluetooth_devices(probe_timeout: float = 0.2) -> bool: +def can_scan_bluetooth_devices(loop_in_thread: LoopInThread, probe_timeout: float = 0.2) -> bool: if not is_ble_available(): return False - def _probe_scan() -> list[Any]: - return asyncio.run(BleakScanner.discover(timeout=max(0.1, probe_timeout))) - try: - _run_ble_operation(_probe_scan) + loop_in_thread.run_foreground(BleakScanner.discover(timeout=max(0.1, probe_timeout))) except Exception as e: logger.info("Bluetooth scanning unavailable in this environment: %s", e) return False @@ -146,19 +136,19 @@ def get_bluetooth_devices(self) -> list[dict[str, Any]]: should_retry_probe = platform.system() == "Darwin" if not self._is_bluetooth_scan_supported(force_refresh=should_retry_probe): if should_retry_probe and self._is_bluetooth_scan_supported(force_refresh=True): - return _run_ble_operation(self._discover_bluetooth_devices) + return self._discover_bluetooth_devices() raise RuntimeError(self.tr("Bluetooth scanning is not available in this environment.")) - return _run_ble_operation(self._discover_bluetooth_devices) + return self._discover_bluetooth_devices() def _is_bluetooth_scan_supported(self, force_refresh: bool = False) -> bool: if not self.enable_bluetooth: return False if force_refresh or self._bluetooth_scan_supported is None: - self._bluetooth_scan_supported = can_scan_bluetooth_devices() + self._bluetooth_scan_supported = can_scan_bluetooth_devices(self.loop_in_thread) return self._bluetooth_scan_supported def _discover_bluetooth_devices(self) -> list[dict[str, Any]]: - return discover_jade_ble_devices(scan_timeout=6.0) + return discover_jade_ble_devices(self.loop_in_thread, scan_timeout=6.0) @staticmethod def _is_bluetooth_device(selected_device: dict[str, Any]) -> bool: diff --git a/tests/test_bluetooth_scan_support.py b/tests/test_bluetooth_scan_support.py index ff0ebec..071760b 100644 --- a/tests/test_bluetooth_scan_support.py +++ b/tests/test_bluetooth_scan_support.py @@ -3,59 +3,59 @@ def test_can_scan_bluetooth_devices_uses_ble_operation_wrapper(monkeypatch) -> None: - calls: dict[str, bool] = {"wrapped": False, "run_called": False} + calls: dict[str, object] = {"run_called": False} monkeypatch.setattr(usb_gui, "is_ble_available", lambda: True) - monkeypatch.setattr(usb_gui.BleakScanner, "discover", lambda timeout: "probe") + monkeypatch.setattr(usb_gui.BleakScanner, "discover", lambda timeout: {"timeout": timeout}) - def fake_run_ble_operation(operation): - calls["wrapped"] = True - return operation() + class _LoopInThread: + def run_foreground(self, coroutine): + calls["run_called"] = True + calls["discover_call"] = coroutine + return [] - def fake_asyncio_run(_probe): - calls["run_called"] = True - return [] + loop_in_thread = _LoopInThread() - monkeypatch.setattr(usb_gui, "_run_ble_operation", fake_run_ble_operation) - monkeypatch.setattr(usb_gui.asyncio, "run", fake_asyncio_run) - - assert usb_gui.can_scan_bluetooth_devices() is True - assert calls == {"wrapped": True, "run_called": True} + assert usb_gui.can_scan_bluetooth_devices(loop_in_thread) is True + assert calls == { + "run_called": True, + "discover_call": {"timeout": 0.2}, + } def test_can_scan_bluetooth_devices_returns_false_on_probe_exception(monkeypatch) -> None: monkeypatch.setattr(usb_gui, "is_ble_available", lambda: True) - def raise_runtime_error(_operation): - raise RuntimeError("probe failed") - - monkeypatch.setattr(usb_gui, "_run_ble_operation", raise_runtime_error) + class _LoopInThread: + def run_foreground(self, coroutine): + coroutine.close() + raise RuntimeError("probe failed") - assert usb_gui.can_scan_bluetooth_devices() is False + assert usb_gui.can_scan_bluetooth_devices(_LoopInThread()) is False def test_get_bluetooth_devices_retries_support_probe_on_macos(monkeypatch) -> None: gui = USBGui(network=object(), loop_in_thread=object()) - probe_calls: list[bool] = [] + probe_calls: list[object] = [] - def fake_probe() -> bool: - probe_calls.append(True) + def fake_probe(loop_in_thread, probe_timeout: float = 0.2) -> bool: + probe_calls.append(loop_in_thread) return len(probe_calls) > 1 monkeypatch.setattr(usb_gui.platform, "system", lambda: "Darwin") monkeypatch.setattr(usb_gui, "can_scan_bluetooth_devices", fake_probe) - monkeypatch.setattr(usb_gui, "_run_ble_operation", lambda operation: operation()) monkeypatch.setattr(gui, "_discover_bluetooth_devices", lambda: [{"transport": "bluetooth"}]) assert gui.get_bluetooth_devices() == [{"transport": "bluetooth"}] assert len(probe_calls) == 2 + assert probe_calls == [gui.loop_in_thread, gui.loop_in_thread] assert gui._bluetooth_scan_supported is True def test_get_device_exposes_bluetooth_scan_callback_when_enabled(monkeypatch) -> None: gui = USBGui(network=object(), loop_in_thread=object(), enable_bluetooth=True) captured: dict[str, object] = {} - callback = lambda: [] + callback = lambda: [] # type: ignore monkeypatch.setattr(gui, "get_bluetooth_devices", callback) class _FakeDialog: diff --git a/tests/test_jade_ble_client.py b/tests/test_jade_ble_client.py index 4034895..dbd015f 100644 --- a/tests/test_jade_ble_client.py +++ b/tests/test_jade_ble_client.py @@ -3,7 +3,7 @@ import pytest from hwilib.devices.jadepy.jade_error import JadeError -from bitcoin_usb.jade_ble_client import CompatibleJadeBleImpl +from bitcoin_usb.jade_ble_client import CompatibleJadeBleImpl, discover_jade_ble_devices async def _never_stream(): @@ -45,3 +45,43 @@ def test_write_impl_times_out() -> None: assert client.write_task is None loop.close() + + +def test_discover_jade_ble_devices_uses_existing_loop(monkeypatch) -> None: + class _Device: + def __init__(self, name: str, address: str) -> None: + self.name = name + self.address = address + + captured: dict[str, object] = {} + + monkeypatch.setattr( + "bitcoin_usb.jade_ble_client.BleakScanner.discover", + lambda timeout: {"timeout": timeout}, + ) + + class _LoopInThread: + def run_foreground(self, coroutine): + captured["discover_call"] = coroutine + return [ + _Device("Jade ABC123", "AA:BB"), + _Device("Jade ABC123", "AA:BB"), + _Device("Other Device", "CC:DD"), + ] + + loop_in_thread = _LoopInThread() + + assert discover_jade_ble_devices(loop_in_thread, scan_timeout=2.5) == [ + { + "type": "jade", + "model": "jade_ble", + "path": "ble:AA:BB", + "needs_pin_sent": False, + "needs_passphrase_sent": False, + "transport": "bluetooth", + "bluetooth_name": "Jade ABC123", + "bluetooth_address": "AA:BB", + "bluetooth_serial_number": "ABC123", + } + ] + assert captured["discover_call"] == {"timeout": 2.5}