From d3735c8f7bfcdfdeca1b1cab336525cb891fcf17 Mon Sep 17 00:00:00 2001 From: Jean-Pierre Martineau Date: Mon, 24 Aug 2026 01:32:42 -0400 Subject: [PATCH] BLE: a connect that fails must put the link back down MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit MeshCore.create_ble builds its client, calls connect() on it, and hands it back only on success. A connect that *raises* therefore left that client — and the bleak link it had already opened — orphaned inside the library with no reference we could close. Our _mc was never assigned, so disconnect() was a no-op and the probe's _safe_disconnect closed nothing. That is not a corner: a PIN-protected companion answers the unbonded notify-subscribe with a GATT authentication error, and the picker's first attempt is always PIN-less. So selecting an unbonded companion leaked the link, Windows held it for the life of the process, and the peripheral — still believing it had a peer — stopped advertising. The PIN dialog then asked for a code it could no longer deliver: the retry paired at the OS level (the bond really is created) but could not form a GATT session, and the user was told their correct PIN belonged to something that "isn't a MeshCore companion". Since Unpair & quit is what leaves you unbonded, the whole thing read as unpairing having changed the PIN. Assemble the two objects here instead. _connect_owned_ble keeps the handle, so both failure shapes close it: a raise (closed, then re-raised) and a None handshake (closed, then reported as "not a companion"). The teardown is shielded — the other way in is a probe whose wait_for expired and cancelled the handshake, where a plain await would be cancelled the moment it suspended and abandon the very close it was called to perform. The picker's accepted PIN now also rides on the context, so a mid-session reconnect doesn't re-run the whole "needs a PIN" dance against a companion we already authenticated to. Session-lifetime only; nothing reaches disk. Tests stop naming a real companion: the addresses and pairing code are stand-ins now, and the fakes model the class we construct rather than the create_ble we no longer call. Co-Authored-By: Claude Opus 5 (1M context) --- meshterm/core/connection.py | 92 ++++++++++++-- meshterm/ui/menu.py | 13 +- tests/test_ble_pairing.py | 231 ++++++++++++++++++++++++++++++++++++ tests/test_disconnect.py | 209 +++++++++++++++++++------------- tests/test_discovery.py | 2 +- tests/test_tui.py | 10 +- 6 files changed, 457 insertions(+), 100 deletions(-) create mode 100644 tests/test_ble_pairing.py diff --git a/meshterm/core/connection.py b/meshterm/core/connection.py index 7286c33..4953c5d 100644 --- a/meshterm/core/connection.py +++ b/meshterm/core/connection.py @@ -116,6 +116,11 @@ _DISCONNECT_TIMEOUT_S = 2.0 #: ``_DISCONNECT_TIMEOUT_S + _FORCE_DISCONNECT_TIMEOUT_S`` stays under the exit watchdog. _FORCE_DISCONNECT_TIMEOUT_S = 1.5 +#: Bound on closing a meshcore client whose own ``connect`` failed (seconds). Deliberately +#: short: this runs on a failure path the user is waiting through — the startup probe that is +#: about to raise "needs a PIN" — and a half-open client has no session state worth draining. +_DISCARD_TIMEOUT_S = 3.0 + #: Total attempts at opening the BLE link before its failure is surfaced. Opening a BLE #: connection on Windows is intermittently flaky (a slow-advertising peripheral is missed by #: bleak's internal lookup, or the link-layer connect races the just-finished discovery scan); @@ -1001,7 +1006,7 @@ class MeshCoreDevice(Device): raise DeviceAuthenticationError(self._ble_auth_message()) from exc async def _create_ble_with_retry(self, mesh_core): # type: ignore[no-untyped-def] - """Call ``create_ble``, retrying the transport-level failures that are transient. + """Open the owned BLE client, retrying the transport-level failures that are transient. The meshcore client raises a bare ``ConnectionError`` when the *link itself* could not be opened — the peripheral wasn't found during bleak's internal lookup, or the @@ -1015,7 +1020,9 @@ class MeshCoreDevice(Device): Only ``ConnectionError`` is retried: a PIN/bond rejection or any other GATT failure propagates unchanged on the first attempt so the auth handling in :meth:`_open_ble` - (and a genuine wrong-PIN) is never looped. + (and a genuine wrong-PIN) is never looped. Each attempt builds its own client through + :meth:`_connect_owned_ble`, which closes it before letting any failure out — a + retried attempt therefore starts from a released link, never a leaked one. Args: mesh_core: The imported ``meshcore.MeshCore`` class. @@ -1032,13 +1039,7 @@ class MeshCoreDevice(Device): if attempt: await asyncio.sleep(_BLE_CONNECT_RETRY_DELAY_S) try: - return await mesh_core.create_ble( - address=self._address, - device=self._ble_device, - pin=self._pin, - default_timeout=self._connect_timeout, - auto_reconnect=False, - ) + return await self._connect_owned_ble(mesh_core) except ConnectionError as exc: _log.debug( "BLE link to %s failed to open (attempt %d/%d): %s", @@ -1051,6 +1052,79 @@ class MeshCoreDevice(Device): assert last_exc is not None # the loop always runs; only ConnectionError falls through raise last_exc + async def _connect_owned_ble(self, mesh_core): # type: ignore[no-untyped-def] + """Build the meshcore BLE client here and connect it, so we own its teardown. + + ``MeshCore.create_ble`` assembles a client, calls ``connect()`` on it, and hands it + back *only on success* — so a connect that **raises** leaves that client, and the + bleak link it has already opened, orphaned inside the library with no reference we + could close. That is not a hypothetical: a PIN-protected companion answers the + unbonded notify-subscribe with a GATT authentication error, which is raised from + deep inside ``connect()`` after bleak has brought the link up. Our own + :meth:`disconnect` then does nothing (``_mc`` was never assigned), Windows holds the + ACL link for the life of the process, and the peripheral — still believing it has a + peer — **stops advertising**. The PIN dialog that opens next therefore asks for a + code it can no longer deliver: the retry can't find the device, and the user reads a + correct PIN being refused. Assembling the same two objects here costs three lines and + keeps the handle, so every exit puts the link back down. + + Both public failure shapes are covered, matching what ``create_ble`` does on the one + it bothers to handle: a raise (closed, then re-raised) and a ``None`` from the + identity handshake (closed, then reported as "not a companion"). + + Args: + mesh_core: The imported ``meshcore.MeshCore`` class. + + Returns: + The connected ``MeshCore`` client, or ``None`` if the transport opened but the + peripheral never answered the identity handshake. + + Raises: + Exception: Whatever ``connect`` raised — but not before the link is closed. + """ + from meshcore import BLEConnection + + connection = BLEConnection( + address=self._address, device=self._ble_device, pin=self._pin + ) + mc = mesh_core( + connection, + default_timeout=self._connect_timeout, + auto_reconnect=False, + ) + try: + started = await mc.connect() + except BaseException: + await MeshCoreDevice._discard_meshcore(mc) + raise + if started is None: + await MeshCoreDevice._discard_meshcore(mc) + return None + return mc + + @staticmethod + async def _discard_meshcore(mc) -> None: # type: ignore[no-untyped-def] + """Close a client whose ``connect`` didn't complete. Bounded, and never raises. + + Shielded on purpose. The other way into this method is a probe whose ``wait_for`` + expired and cancelled the handshake mid-flight; a plain ``await`` would then be + cancelled itself the moment it suspended, abandoning the very teardown it was called + to perform and leaking exactly the link this exists to close. Shielding lets the + close finish on its own while the cancellation continues to propagate to our caller. + + Args: + mc: A half-open ``meshcore.MeshCore`` client. + """ + closing = asyncio.ensure_future( + asyncio.wait_for(mc.disconnect(), timeout=_DISCARD_TIMEOUT_S) + ) + try: + await asyncio.shield(closing) + except BaseException as exc: # noqa: BLE001 - teardown of a doomed client + # Includes CancelledError: swallowed here only so the caller's own exception + # (or cancellation) is the one that propagates. ``closing`` runs on regardless. + _log.debug("discarded a half-open BLE client: %s", exc) + async def _pair_ble_windows(self, *, force: bool) -> bool: """Establish an authenticated BLE bond via the WinRT ProvidePin ceremony (Windows only). diff --git a/meshterm/ui/menu.py b/meshterm/ui/menu.py index 8f0a897..5132d2a 100644 --- a/meshterm/ui/menu.py +++ b/meshterm/ui/menu.py @@ -594,14 +594,19 @@ async def _startup(ctx: AppContext) -> bool: async def verify(device: DiscoveredDevice, pin: Optional[str] = None): # ``pin`` is what the picker's PIN dialog collected on a retry; fall back to any # ``--ble-pin`` supplied on the CLI for the first attempt. - result = await probe_device( - device, baudrate=baudrate, pin=pin if pin is not None else ctx.ble_pin - ) + used = pin if pin is not None else ctx.ble_pin + result = await probe_device(device, baudrate=baudrate, pin=used) if result is None: return None connection, info = result probed["device_id"] = device.stable_id probed["device"] = connection + # Remember the PIN that actually worked. The adopted connection carries its own + # copy, but a mid-session reconnect builds a *fresh* device from the context's + # resolved endpoint — which, without this, would go back unarmed and re-run the + # whole "needs a PIN" dance against a companion we already authenticated to. + # Session-lifetime only: nothing writes a pairing code to disk. + probed["pin"] = used return info chosen = await prompt_device( @@ -624,6 +629,8 @@ async def _startup(ctx: AppContext) -> bool: ctx.tcp_override = None if probed.get("device_id") == chosen.stable_id: ctx.adopt_device(probed["device"]) + if probed.get("pin"): + ctx.ble_pin = probed["pin"] # Between the device splash and the first menu paint, resuming background listening opens # the radio — a slow, silent step that would otherwise leave the screen blank for a beat. # Float the skeleton card across that gap on any real link (see _busy_over_link). diff --git a/tests/test_ble_pairing.py b/tests/test_ble_pairing.py new file mode 100644 index 0000000..f23c573 --- /dev/null +++ b/tests/test_ble_pairing.py @@ -0,0 +1,231 @@ +"""Tests for the BLE connect path's ownership of its meshcore client. + +The bug these pin down: ``MeshCore.create_ble`` builds a client, calls ``connect()`` on it, +and returns it *only on success*. A connect that raises — which is exactly how a +PIN-protected companion answers an unbonded notify-subscribe — therefore left the client, +and the bleak link it had already opened, orphaned inside the library. MeshTerm's own +``disconnect`` was a no-op (``_mc`` was never assigned), Windows held the ACL link for the +life of the process, and the peripheral, still believing it had a peer, stopped advertising. +The PIN dialog that opened next then asked for a code it could no longer deliver. + +Everything here runs against fakes standing in for ``meshcore.MeshCore`` and +``meshcore.BLEConnection``; no radio, no address, and no pairing code is involved. +""" + +from __future__ import annotations + +import asyncio +from typing import Optional + +import pytest + +from meshterm.core import connection as conn_mod +from meshterm.core.connection import MeshCoreDevice + +# A stand-in address. Deliberately in the documentation-style ``00:`` space so no real +# companion is named here, and nothing in this file needs hardware to run. +_ADDR = "00:11:22:33:44:55" + + +class _FakeBLEConnection: + """Stands in for ``meshcore.BLEConnection`` — records how it was constructed.""" + + last: Optional["_FakeBLEConnection"] = None + + def __init__(self, address=None, device=None, pin=None) -> None: + self.address = address + self.device = device + self.pin = pin + _FakeBLEConnection.last = self + + +class _FakeMeshCore: + """Stands in for the ``meshcore.MeshCore`` *class*, one instance per connect attempt. + + ``outcome`` drives what ``connect()`` does: ``"ok"`` returns a truthy handshake result, + ``"none"`` returns ``None`` (transport up, no identity reply), and an exception instance + is raised — the GATT authentication failure being the case that mattered. + """ + + #: Every instance built during a test, in order (a retry builds a second one). + built: list["_FakeMeshCore"] = [] + #: What the next ``connect()`` should do; a list is consumed one entry per attempt. + outcomes: list = [] + + def __init__(self, cx, *, default_timeout=None, auto_reconnect=False, **kwargs) -> None: + self.cx = cx + self.default_timeout = default_timeout + self.auto_reconnect = auto_reconnect + self.connected = False + self.disconnect_calls = 0 + self.commands = None + type(self).built.append(self) + + async def connect(self): + outcome = type(self).outcomes.pop(0) + if isinstance(outcome, BaseException): + # The link is up by the time the subscribe fails — that is the whole point: + # something must close it, and only the holder of this object can. + self.connected = True + raise outcome + if outcome == "slow": + self.connected = True + await asyncio.sleep(30) # a handshake the caller's wait_for will cancel + self.connected = True + return None if outcome == "none" else {"ok": True} + + async def disconnect(self): + self.disconnect_calls += 1 + self.connected = False + + @classmethod + def reset(cls, *outcomes) -> None: + cls.built = [] + cls.outcomes = list(outcomes) + + +@pytest.fixture(autouse=True) +def _fake_meshcore(monkeypatch: pytest.MonkeyPatch): + """Make ``from meshcore import BLEConnection`` inside the connect path resolve to a fake.""" + import sys + import types + + module = types.ModuleType("meshcore") + module.BLEConnection = _FakeBLEConnection + module.MeshCore = _FakeMeshCore + monkeypatch.setitem(sys.modules, "meshcore", module) + _FakeBLEConnection.last = None + yield + + +def _device(pin: Optional[str] = None) -> MeshCoreDevice: + return MeshCoreDevice(transport="ble", address=_ADDR, pin=pin, connect_timeout=5.0) + + +class _AuthError(Exception): + """A stand-in for the GATT failure a PIN-protected companion answers with.""" + + +def test_a_raising_connect_closes_the_client_it_leaves_behind() -> None: + """THE bug: a connect that raises must not strand an open link. + + ``create_ble`` would have swallowed the reference here. Owning the client means the + failure path can put the link down before the exception continues on its way — so the + peripheral stops holding a phantom peer and goes on advertising for the PIN retry. + """ + _FakeMeshCore.reset(_AuthError("Insufficient Authentication")) + dev = _device() + + with pytest.raises(_AuthError): + asyncio.run(dev._connect_owned_ble(_FakeMeshCore)) + + assert len(_FakeMeshCore.built) == 1 + client = _FakeMeshCore.built[0] + assert client.disconnect_calls == 1, "the half-open client was left open" + assert client.connected is False + + +def test_an_unanswered_handshake_closes_the_client_too() -> None: + """Transport up, no identity reply: reported as "not a companion", link still closed.""" + _FakeMeshCore.reset("none") + dev = _device() + + assert asyncio.run(dev._connect_owned_ble(_FakeMeshCore)) is None + assert _FakeMeshCore.built[0].disconnect_calls == 1 + + +def test_a_good_connect_hands_the_client_over_still_open() -> None: + """The success path must not close anything — the caller owns the live client.""" + _FakeMeshCore.reset("ok") + dev = _device() + + client = asyncio.run(dev._connect_owned_ble(_FakeMeshCore)) + + assert client is _FakeMeshCore.built[0] + assert client.disconnect_calls == 0 + assert client.connected is True + + +def test_the_connection_is_built_from_this_device_s_own_endpoint() -> None: + """Address, discovered ``BLEDevice`` and PIN all reach the connection we construct.""" + _FakeMeshCore.reset("ok") + sentinel = object() + dev = MeshCoreDevice( + transport="ble", address=_ADDR, pin="000000", ble_device=sentinel, + connect_timeout=7.5, + ) + + asyncio.run(dev._connect_owned_ble(_FakeMeshCore)) + + cx = _FakeBLEConnection.last + assert (cx.address, cx.device, cx.pin) == (_ADDR, sentinel, "000000") + client = _FakeMeshCore.built[0] + # auto_reconnect stays off: MeshTerm drives reconnection itself. + assert (client.default_timeout, client.auto_reconnect) == (7.5, False) + + +def test_a_cancelled_handshake_still_closes_the_link() -> None: + """A probe's ``wait_for`` expiring mid-handshake must not leak what it cancelled. + + This is the second way in, and the reason the teardown is shielded: a plain ``await`` + would itself be cancelled the moment it suspended, abandoning the close. + """ + _FakeMeshCore.reset("slow") + dev = _device() + + async def scenario(): + with pytest.raises(asyncio.TimeoutError): + await asyncio.wait_for(dev._connect_owned_ble(_FakeMeshCore), timeout=0.1) + # The shielded close runs on after the cancellation propagates; give it a tick. + await asyncio.sleep(0.1) + return _FakeMeshCore.built[0] + + client = asyncio.run(scenario()) + assert client.disconnect_calls == 1, "a cancelled handshake stranded its link" + + +def test_a_failing_disconnect_never_masks_the_real_error() -> None: + """Teardown is best-effort: the connect's own failure is what the caller must see.""" + + class _Stubborn(_FakeMeshCore): + async def disconnect(self): + self.disconnect_calls += 1 + raise RuntimeError("teardown exploded") + + _Stubborn.reset(_AuthError("Insufficient Authentication")) + dev = _device() + + with pytest.raises(_AuthError): + asyncio.run(dev._connect_owned_ble(_Stubborn)) + assert _Stubborn.built[0].disconnect_calls == 1 + + +def test_the_retry_loop_never_reuses_a_torn_down_client() -> None: + """A retried link-open builds a fresh client, and the abandoned one was closed.""" + _FakeMeshCore.reset(ConnectionError("Failed to connect to device"), "ok") + dev = _device() + + async def scenario(): + return await dev._create_ble_with_retry(_FakeMeshCore) + + client = asyncio.run(scenario()) + + assert len(_FakeMeshCore.built) == 2, "the retry did not build its own client" + assert _FakeMeshCore.built[0].disconnect_calls == 1, "the failed attempt leaked" + assert client is _FakeMeshCore.built[1] + assert client.disconnect_calls == 0 + + +def test_every_link_open_attempt_failing_raises_the_last_error() -> None: + """Exhausting the retries surfaces the link failure — with nothing left open.""" + _FakeMeshCore.reset( + ConnectionError("Failed to connect to device"), + ConnectionError("Failed to connect to device"), + ) + dev = _device() + + with pytest.raises(ConnectionError): + asyncio.run(dev._create_ble_with_retry(_FakeMeshCore)) + + assert len(_FakeMeshCore.built) == conn_mod._BLE_CONNECT_ATTEMPTS + assert all(c.disconnect_calls == 1 for c in _FakeMeshCore.built) diff --git a/tests/test_disconnect.py b/tests/test_disconnect.py index f17f08f..43ec13d 100644 --- a/tests/test_disconnect.py +++ b/tests/test_disconnect.py @@ -115,6 +115,16 @@ def test_ble_auth_error_walks_the_exception_chain() -> None: assert connection._is_ble_auth_error(exc) + +#: Stand-in Bluetooth addresses. Never a real companion's: a test that names one turns a +#: personal device into repository content, and any hardware coupling here would be a lie — +#: nothing in this file touches a radio. +_BONDED_ADDR = "00:11:22:33:44:55" +_OPEN_ADDR = "AA:BB:CC:DD:EE:FF" +#: A stand-in pairing code, for the same reason. What matters is only that one was supplied. +_A_PIN = "123456" + + async def _disable_windows_pairing(dev: "connection.MeshCoreDevice") -> None: """Stub out the WinRT ProvidePin step so ``_create_ble`` stays hermetic in tests. @@ -129,53 +139,92 @@ async def _disable_windows_pairing(dev: "connection.MeshCoreDevice") -> None: dev._pair_ble_windows = _never_pairs # type: ignore[method-assign] -async def test_create_ble_translates_auth_error_to_pin_guidance() -> None: - """A raw GATT auth rejection becomes a DeviceAuthenticationError that names the PIN fix.""" +def _fake_ble_stack(monkeypatch: pytest.MonkeyPatch, *outcomes): + """A stand-in ``MeshCore`` class, scripted one entry per connect attempt. + + MeshTerm builds the meshcore client itself (:meth:`~MeshCoreDevice._connect_owned_ble`) + rather than letting ``create_ble`` construct and then orphan it, so the fake here is a + *class* that gets instantiated per attempt — and each instance records whether it was + closed, which is the property that matters: a connect that fails must never leave its + link open, or the peripheral goes on believing it has a peer and stops advertising. + + Args: + monkeypatch: Used to stub the lazily-imported ``meshcore`` module, so no real + ``BLEConnection`` (and therefore no ``bleak``) is needed. + outcomes: What each successive ``connect()`` does — an exception instance to raise, + or a value to return (``None`` standing for an unanswered handshake). + + Returns: + The fake class; its ``built`` list holds the instances it made, in order. + """ + import sys + import types + + class _FakeBLEConnection: + def __init__(self, address=None, device=None, pin=None) -> None: + self.address, self.device, self.pin = address, device, pin class _FakeMeshCore: - @staticmethod - async def create_ble(**kwargs): - raise BleakGATTProtocolError("Insufficient Authentication") + built: list = [] + script: list = list(outcomes) - # No PIN supplied → tell the user to pass one. The subclass lets the interactive picker + def __init__(self, cx, **kwargs) -> None: + self.cx = cx + self.kwargs = kwargs + self.disconnect_calls = 0 + type(self).built.append(self) + + async def connect(self): + outcome = type(self).script.pop(0) + if isinstance(outcome, BaseException): + raise outcome + return outcome + + async def disconnect(self) -> None: + self.disconnect_calls += 1 + + module = types.ModuleType("meshcore") + module.BLEConnection = _FakeBLEConnection + module.MeshCore = _FakeMeshCore + monkeypatch.setitem(sys.modules, "meshcore", module) + return _FakeMeshCore + + +async def test_create_ble_translates_auth_error_to_pin_guidance(monkeypatch) -> None: + """A raw GATT auth rejection becomes a DeviceAuthenticationError that names the PIN fix.""" + # No PIN supplied -> tell the user to pass one. The subclass lets the interactive picker # catch "needs a PIN" specifically, while the CLI still catches it as DeviceCommandError. - dev = connection.MeshCoreDevice(transport="ble", address="00:11:22:33:44:55") + fake = _fake_ble_stack(monkeypatch, BleakGATTProtocolError("Insufficient Authentication")) + dev = connection.MeshCoreDevice(transport="ble", address=_BONDED_ADDR) await _disable_windows_pairing(dev) with pytest.raises(connection.DeviceAuthenticationError) as excinfo: - await dev._create_ble(_FakeMeshCore) + await dev._create_ble(fake) assert isinstance(excinfo.value, DeviceCommandError) # so the scripted CLI catches it too assert "--ble-pin" in str(excinfo.value) + assert fake.built[0].disconnect_calls == 1 # the doomed link was released, not stranded - # PIN supplied but rejected → say it was wrong, not that none was given. - dev_pin = connection.MeshCoreDevice( - transport="ble", address="00:11:22:33:44:55", pin="123456" + # PIN supplied but rejected -> say it was wrong, not that none was given. + fake_pin = _fake_ble_stack( + monkeypatch, BleakGATTProtocolError("Insufficient Authentication") ) + dev_pin = connection.MeshCoreDevice(transport="ble", address=_BONDED_ADDR, pin=_A_PIN) await _disable_windows_pairing(dev_pin) with pytest.raises(connection.DeviceAuthenticationError) as excinfo_pin: - await dev_pin._create_ble(_FakeMeshCore) + await dev_pin._create_ble(fake_pin) assert "rejected" in str(excinfo_pin.value).lower() -async def test_create_ble_repairs_stale_bond_and_retries_once() -> None: +async def test_create_ble_repairs_stale_bond_and_retries_once(monkeypatch) -> None: """A first auth failure triggers one unpair-and-re-pair, then the retried connect succeeds. Models the Windows upgrade case: a leftover unauthenticated "Just Works" bond makes the first connect fail even with the right PIN, so ``_pair_ble_windows(force=True)`` clears it and the second connect goes through. """ - attempts: list[int] = [] - - class _FakeMeshCore: - @staticmethod - async def create_ble(**kwargs): - attempts.append(1) - if len(attempts) == 1: - raise BleakGATTProtocolError("Insufficient Authentication") - return "connected-client" - - dev = connection.MeshCoreDevice( - transport="ble", address="00:11:22:33:44:55", pin="123456" + fake = _fake_ble_stack( + monkeypatch, BleakGATTProtocolError("Insufficient Authentication"), "handshake-ok" ) + dev = connection.MeshCoreDevice(transport="ble", address=_BONDED_ADDR, pin=_A_PIN) repairs: list[bool] = [] async def _pair(*, force: bool) -> bool: @@ -183,23 +232,22 @@ async def test_create_ble_repairs_stale_bond_and_retries_once() -> None: return force # the pre-connect pass (force=False) no-ops; the repair (force=True) works dev._pair_ble_windows = _pair # type: ignore[method-assign] - result = await dev._create_ble(_FakeMeshCore) - assert result == "connected-client" - assert len(attempts) == 2 # failed once, retried once + result = await dev._create_ble(fake) + assert result is fake.built[1] # the retry's client is what the caller gets + assert len(fake.built) == 2 # failed once, retried once assert repairs == [False, True] # pre-connect attempt, then the healing re-pair + assert fake.built[0].disconnect_calls == 1 # ...and the failed one was closed first + assert fake.built[1].disconnect_calls == 0 # the live one is handed over open -async def test_create_ble_gives_up_after_one_repair() -> None: +async def test_create_ble_gives_up_after_one_repair(monkeypatch) -> None: """A wrong PIN that never bonds fails cleanly rather than looping on the repair retry.""" - - class _FakeMeshCore: - @staticmethod - async def create_ble(**kwargs): - raise BleakGATTProtocolError("Insufficient Authentication") - - dev = connection.MeshCoreDevice( - transport="ble", address="00:11:22:33:44:55", pin="000000" + fake = _fake_ble_stack( + monkeypatch, + BleakGATTProtocolError("Insufficient Authentication"), + BleakGATTProtocolError("Insufficient Authentication"), ) + dev = connection.MeshCoreDevice(transport="ble", address=_BONDED_ADDR, pin=_A_PIN) calls: list[bool] = [] async def _pair(*, force: bool) -> bool: @@ -208,10 +256,11 @@ async def test_create_ble_gives_up_after_one_repair() -> None: dev._pair_ble_windows = _pair # type: ignore[method-assign] with pytest.raises(connection.DeviceAuthenticationError): - await dev._create_ble(_FakeMeshCore) + await dev._create_ble(fake) # force=False (pre-connect), then force=True (repair). The repair's retry passes # allow_repair=False, so there is no third pairing attempt even though it keeps failing. assert calls == [False, True] + assert all(client.disconnect_calls == 1 for client in fake.built) async def test_create_ble_retries_a_transient_link_failure(monkeypatch) -> None: @@ -223,72 +272,68 @@ async def test_create_ble_retries_a_transient_link_failure(monkeypatch) -> None: itself before surfacing the failure. """ scanned_device = object() # the BLEDevice the discovery scan produced - attempts: list[dict] = [] - - class _FakeMeshCore: - @staticmethod - async def create_ble(**kwargs): - attempts.append(kwargs) - if len(attempts) == 1: - raise ConnectionError("Failed to connect to device") - return "connected-client" - + fake = _fake_ble_stack( + monkeypatch, ConnectionError("Failed to connect to device"), "handshake-ok" + ) monkeypatch.setattr(connection, "_BLE_CONNECT_RETRY_DELAY_S", 0.0) dev = connection.MeshCoreDevice( - transport="ble", address="AA:BB:CC:DD:EE:FF", ble_device=scanned_device + transport="ble", address=_OPEN_ADDR, ble_device=scanned_device ) await _disable_windows_pairing(dev) - assert await dev._create_ble(_FakeMeshCore) == "connected-client" - assert len(attempts) == 2 # failed once, retried once + assert await dev._create_ble(fake) is fake.built[1] + assert len(fake.built) == 2 # failed once, retried once # Every attempt connects through the already-discovered BLEDevice, never a bare address. - assert all(kw["device"] is scanned_device for kw in attempts) + assert all(client.cx.device is scanned_device for client in fake.built) async def test_create_ble_gives_up_after_the_retry(monkeypatch) -> None: """A link that never opens surfaces its ConnectionError after the bounded retries.""" - attempts: list[int] = [] - - class _FakeMeshCore: - @staticmethod - async def create_ble(**kwargs): - attempts.append(1) - raise ConnectionError("Failed to connect to device") - + fake = _fake_ble_stack( + monkeypatch, + ConnectionError("Failed to connect to device"), + ConnectionError("Failed to connect to device"), + ) monkeypatch.setattr(connection, "_BLE_CONNECT_RETRY_DELAY_S", 0.0) - dev = connection.MeshCoreDevice(transport="ble", address="AA:BB:CC:DD:EE:FF") + dev = connection.MeshCoreDevice(transport="ble", address=_OPEN_ADDR) await _disable_windows_pairing(dev) with pytest.raises(ConnectionError): - await dev._create_ble(_FakeMeshCore) - assert len(attempts) == connection._BLE_CONNECT_ATTEMPTS + await dev._create_ble(fake) + assert len(fake.built) == connection._BLE_CONNECT_ATTEMPTS + assert all(client.disconnect_calls == 1 for client in fake.built) -async def test_create_ble_never_retries_an_auth_rejection() -> None: +async def test_create_ble_never_retries_an_auth_rejection(monkeypatch) -> None: """A PIN/bond rejection is translated on the first attempt — never looped by the retry.""" - attempts: list[int] = [] - - class _FakeMeshCore: - @staticmethod - async def create_ble(**kwargs): - attempts.append(1) - raise BleakGATTProtocolError("Insufficient Authentication") - - dev = connection.MeshCoreDevice(transport="ble", address="00:11:22:33:44:55") + fake = _fake_ble_stack(monkeypatch, BleakGATTProtocolError("Insufficient Authentication")) + dev = connection.MeshCoreDevice(transport="ble", address=_BONDED_ADDR) await _disable_windows_pairing(dev) with pytest.raises(connection.DeviceAuthenticationError): - await dev._create_ble(_FakeMeshCore) - assert len(attempts) == 1 + await dev._create_ble(fake) + assert len(fake.built) == 1 + + +async def test_connect_reports_an_unanswered_handshake_and_closes_it(monkeypatch) -> None: + """Transport up but no identity reply: a clean "not a companion", link still released.""" + fake = _fake_ble_stack(monkeypatch, None) + dev = connection.MeshCoreDevice(transport="ble", address=_OPEN_ADDR) + await _disable_windows_pairing(dev) + with pytest.raises(DeviceCommandError): + await dev.connect() + assert fake.built[0].disconnect_calls == 1 async def test_pair_ble_windows_noops_without_pin() -> None: """Pairing is skipped (no WinRT touched) when no PIN is set — the fast, hermetic path.""" - dev = connection.MeshCoreDevice(transport="ble", address="00:11:22:33:44:55") + dev = connection.MeshCoreDevice(transport="ble", address=_BONDED_ADDR) assert await dev._pair_ble_windows(force=False) is False + + def test_ble_address_int_parses_macs_and_rejects_others() -> None: """A colon/dash MAC becomes a 48-bit int; a non-MAC (e.g. a macOS UUID) yields None.""" parse = connection.MeshCoreDevice._ble_address_int - assert parse("00:11:22:33:44:55") == 0xDAC5216B7C7C + assert parse(_BONDED_ADDR) == 0x001122334455 assert parse("da-c5-21-6b-7c-7c") == 0xDAC5216B7C7C assert parse("not-a-mac") is None assert parse("550e8400-e29b-41d4-a716-446655440000") is None # CoreBluetooth UUID @@ -307,7 +352,7 @@ async def test_can_unpair_only_for_bonded_ble(monkeypatch: pytest.MonkeyPatch) - """The quit dialog offers unpair only on a BLE link that Windows actually holds a bond for.""" async def _is_paired(address: str) -> bool: - return address == "00:11:22:33:44:55" + return address == _BONDED_ADDR monkeypatch.setattr( connection.MeshCoreDevice, "is_ble_paired", staticmethod(_is_paired) @@ -319,10 +364,10 @@ async def test_can_unpair_only_for_bonded_ble(monkeypatch: pytest.MonkeyPatch) - ble_no_addr = SimpleNamespace(active_transport="ble", active_address=None) assert await menu._can_unpair(ble_no_addr) is False # BLE with a live bond: offered. - bonded = SimpleNamespace(active_transport="ble", active_address="00:11:22:33:44:55") + bonded = SimpleNamespace(active_transport="ble", active_address=_BONDED_ADDR) assert await menu._can_unpair(bonded) is True # BLE but open (no bond, e.g. the PIN-less companion): not offered. - open_ble = SimpleNamespace(active_transport="ble", active_address="AA:BB:CC:DD:EE:FF") + open_ble = SimpleNamespace(active_transport="ble", active_address=_OPEN_ADDR) assert await menu._can_unpair(open_ble) is False @@ -341,11 +386,11 @@ async def test_unpair_on_exit_disconnects_before_unpairing( return True monkeypatch.setattr(connection.MeshCoreDevice, "unpair_ble", staticmethod(_unpair)) - ctx = SimpleNamespace(active_address="00:11:22:33:44:55", _device=_Dev()) + ctx = SimpleNamespace(active_address=_BONDED_ADDR, _device=_Dev()) await menu._unpair_on_exit(ctx) # Disconnect precedes unpair (a bond can't be dropped while in use), and the device handle # is released. The device_store is never touched — the remembered record survives. - assert order == ["disconnect", "unpair:00:11:22:33:44:55"] + assert order == ["disconnect", f"unpair:{_BONDED_ADDR}"] assert ctx._device is None diff --git a/tests/test_discovery.py b/tests/test_discovery.py index b09f76b..57e173a 100644 --- a/tests/test_discovery.py +++ b/tests/test_discovery.py @@ -298,7 +298,7 @@ def test_remember_keeps_known_node_name_when_none_supplied(tmp_path: Path) -> No def test_device_store_round_trips_hardware_model(tmp_path: Path) -> None: """The firmware model learned at connect time writes and reads back.""" store = DeviceStore(tmp_path / "devices.json") - dev = DiscoveredDevice(transport="ble", address="AA:BB:CC:DD:EE:FF", name="MeshCore-Waymarker") + dev = DiscoveredDevice(transport="ble", address="AA:BB:CC:DD:EE:FF", name="MeshCore-Testbench") store.remember(dev, node_name="Waymarker", hardware_model="Seeed Tracker T1000-E") loaded = store.load() diff --git a/tests/test_tui.py b/tests/test_tui.py index d073ef8..a4a5808 100644 --- a/tests/test_tui.py +++ b/tests/test_tui.py @@ -998,21 +998,21 @@ def test_pin_dialog_shows_six_slots_with_dots_for_blanks() -> None: """The PIN field is six fixed slots: bullets for typed digits, centre dots for blanks.""" from meshterm.ui.tui.prompt import PinDialog - dialog = PinDialog("MeshCore-Homestead") + dialog = PinDialog("MeshCore-Testbench") # Empty: six blank centre dots, no bullets yet. field = dialog._editor.render(slots=PinDialog.PIN_LENGTH).plain assert field.count("·") == 6 assert "•" not in field # After three digits: three bullets, three remaining centre dots. - for ch in "701": + for ch in "123": dialog.handle("text", ch) field = dialog._editor.render(slots=PinDialog.PIN_LENGTH).plain assert field.count("•") == 3 assert field.count("·") == 3 # A full six-digit PIN fills every slot; typing more is capped at six. - for ch in "307999": + for ch in "456999": dialog.handle("text", ch) assert dialog._editor.text == "123456" field = dialog._editor.render(slots=PinDialog.PIN_LENGTH).plain @@ -1605,7 +1605,7 @@ def test_device_picker_prompts_and_retries_ble_pin(tmp_path) -> None: from meshterm.ui.device_picker import prompt_device devices = [ - DiscoveredDevice(transport="ble", address="00:11:22:33:44:55", name="MeshCore-Pinned") + DiscoveredDevice(transport="ble", address="00:11:22:33:44:55", name="MeshCore-Testbench") ] store = DeviceStore(tmp_path / "devices.json") @@ -1652,7 +1652,7 @@ def test_device_picker_pin_cancel_returns_to_list(tmp_path) -> None: from meshterm.ui.device_picker import _QUIT, prompt_device devices = [ - DiscoveredDevice(transport="ble", address="00:11:22:33:44:55", name="MeshCore-Pinned") + DiscoveredDevice(transport="ble", address="00:11:22:33:44:55", name="MeshCore-Testbench") ] store = DeviceStore(tmp_path / "devices.json") picks = iter([0, "quit"]) # pick the device once, then quit the re-opened list