From c771f209221619c45e6a786db4b840e2cdb7fc9c Mon Sep 17 00:00:00 2001 From: Ingo Fischer Date: Fri, 22 May 2026 20:33:05 +0200 Subject: [PATCH] feat(ble-proxy): pluggable connect_strategy hook for MatterBleProxy Adds an optional connect_strategy callable to MatterBleProxy so embedders can plug in bleak_retry_connector.establish_connection (or any other connection helper) without making it a hard dependency of the library. - Public API: MatterBleProxy(connect_strategy=...) and the default matter_ble_proxy.default_connect_strategy implementation - _handle_connect delegates to the strategy; the existing TimeoutError / generic-Exception mapping to "connection_failed" is preserved - Tests cover constructor wiring, success path, device-not-found short-circuit, and timeout/exception mapping Background: Home Assistant's habluetooth.wrappers warns that BleakClient.connect() was called without bleak-retry-connector during BLE proxy pairing. With this hook HA can pass establish_connection and get adapter path scoring + retry for free. --- python_ble_proxy/matter_ble_proxy/__init__.py | 9 +- python_ble_proxy/matter_ble_proxy/client.py | 41 ++++- .../tests/test_connect_strategy.py | 151 ++++++++++++++++++ 3 files changed, 195 insertions(+), 6 deletions(-) create mode 100644 python_ble_proxy/tests/test_connect_strategy.py diff --git a/python_ble_proxy/matter_ble_proxy/__init__.py b/python_ble_proxy/matter_ble_proxy/__init__.py index d024c66d..8c91ff04 100644 --- a/python_ble_proxy/matter_ble_proxy/__init__.py +++ b/python_ble_proxy/matter_ble_proxy/__init__.py @@ -14,7 +14,13 @@ """ from .bleak_backend import BleakDeviceResolver, BleakScanSource -from .client import BleDeviceResolver, BleScanSource, ConnectionState, MatterBleProxy +from .client import ( + BleDeviceResolver, + BleScanSource, + ConnectionState, + MatterBleProxy, + default_connect_strategy, +) from .protocol import ( BINARY_FRAME_HEADER, BLE_PROXY_PROTOCOL_VERSION, @@ -37,4 +43,5 @@ "BleakScanSource", "ConnectionState", "MatterBleProxy", + "default_connect_strategy", ] diff --git a/python_ble_proxy/matter_ble_proxy/client.py b/python_ble_proxy/matter_ble_proxy/client.py index af6bb109..950d2998 100644 --- a/python_ble_proxy/matter_ble_proxy/client.py +++ b/python_ble_proxy/matter_ble_proxy/client.py @@ -38,13 +38,18 @@ ) if TYPE_CHECKING: - from collections.abc import Callable, Coroutine + from collections.abc import Awaitable, Callable, Coroutine from bleak import BleakClient from bleak.backends.characteristic import BleakGATTCharacteristic from bleak.backends.device import BLEDevice from bleak.backends.service import BleakGATTServiceCollection + ConnectStrategy = Callable[ + ["BLEDevice | str", Callable[[BleakClient], None], int], + Awaitable[BleakClient], + ] + _LOGGER = logging.getLogger(__name__) # Trailing 24 hex chars of the Bluetooth SIG Base UUID. A 128-bit UUID whose @@ -88,6 +93,26 @@ async def stop(self) -> None: +async def default_connect_strategy( + target: BLEDevice | str, + disconnected_callback: Callable[[BleakClient], None], + timeout_ms: int, +) -> BleakClient: + """Connect with plain `BleakClient` under a single timeout. + + Embedded hosts (Home Assistant) can pass their own strategy via + `MatterBleProxy(connect_strategy=...)` to plug in `bleak-retry-connector` + or other connection-establishment helpers without making them a hard + dependency of the library. + """ + from bleak import BleakClient + + client = BleakClient(target, disconnected_callback=disconnected_callback) + async with asyncio.timeout(timeout_ms / 1000): + await client.connect() + return client + + class BleDeviceResolver(ABC): """Pluggable lookup from `address` to a Bleak connect target.""" @@ -143,6 +168,7 @@ def __init__( *, session: aiohttp.ClientSession | None = None, task_factory: Callable[[Coroutine[Any, Any, Any]], asyncio.Task[Any]] | None = None, + connect_strategy: ConnectStrategy | None = None, ) -> None: """Initialize the proxy. @@ -156,6 +182,13 @@ def __init__( work (e.g. event forwarding). Defaults to `asyncio.get_event_loop().create_task`. Pass `hass.async_create_task` from Home Assistant. + connect_strategy: Optional coroutine called to establish each + BLE connection. Receives `(target, disconnected_callback, + timeout_ms)` and must return a connected `BleakClient`. + Defaults to `default_connect_strategy`. Home Assistant + supplies an implementation backed by + `bleak_retry_connector.establish_connection` for adapter + path scoring and retry. """ self._ws_url = ws_url @@ -171,6 +204,7 @@ def __init__( self._closed_event = asyncio.Event() self._loop: asyncio.AbstractEventLoop | None = None self._task_factory = task_factory + self._connect_strategy: ConnectStrategy = connect_strategy or default_connect_strategy async def connect(self) -> None: """Connect to the matter-server `/ble` endpoint and perform handshake.""" @@ -399,7 +433,6 @@ async def _handle_stop_scan(self, cmd_id: int, _args: dict[str, Any]) -> None: await self._send_success(cmd_id) async def _handle_connect(self, cmd_id: int, args: dict[str, Any]) -> None: - from bleak import BleakClient address: str = args["address"] timeout_ms: int = args.get("timeout", DEFAULT_CONNECT_TIMEOUT_MS) @@ -425,10 +458,8 @@ def _on_disconnect(_client: BleakClient) -> None: self._send_event("disconnected", {"connection_handle": handle}), ) - client = BleakClient(target, disconnected_callback=_on_disconnect) try: - async with asyncio.timeout(timeout_ms / 1000): - await client.connect() + client = await self._connect_strategy(target, _on_disconnect, timeout_ms) except TimeoutError: await self._send_error(cmd_id, "connection_failed", f"Timeout connecting to {address}") return diff --git a/python_ble_proxy/tests/test_connect_strategy.py b/python_ble_proxy/tests/test_connect_strategy.py new file mode 100644 index 00000000..9d746105 --- /dev/null +++ b/python_ble_proxy/tests/test_connect_strategy.py @@ -0,0 +1,151 @@ +"""Unit tests for the pluggable connect_strategy hook on MatterBleProxy.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from matter_ble_proxy.client import ( + BleDeviceResolver, + BleScanSource, + MatterBleProxy, + default_connect_strategy, +) + +if TYPE_CHECKING: + from bleak.backends.device import BLEDevice + + from matter_ble_proxy.client import ConnectStrategy + + +class _StubScanSource(BleScanSource): + async def start(self, callback: object) -> None: + pass + + async def stop(self) -> None: + pass + + +class _StubResolver(BleDeviceResolver): + def __init__(self, target: BLEDevice | str | None) -> None: + self._target = target + + async def resolve(self, address: str) -> BLEDevice | str | None: + return self._target + + +def _make_proxy( + *, + resolver: BleDeviceResolver, + connect_strategy: ConnectStrategy | None = None, +) -> tuple[MatterBleProxy, AsyncMock, AsyncMock]: + """Build a MatterBleProxy with WS plumbing mocked out.""" + proxy = MatterBleProxy( + ws_url="ws://test/ble", + scan_source=_StubScanSource(), + device_resolver=resolver, + connect_strategy=connect_strategy, + ) + send_success = AsyncMock() + send_error = AsyncMock() + proxy._send_success = send_success # type: ignore[method-assign] + proxy._send_error = send_error # type: ignore[method-assign] + return proxy, send_success, send_error + + +def test_default_connect_strategy_used_when_none_provided() -> None: + """Constructor wires `default_connect_strategy` when caller passes None.""" + proxy, _, _ = _make_proxy(resolver=_StubResolver(None)) + assert proxy._connect_strategy is default_connect_strategy + + +def test_custom_connect_strategy_replaces_default() -> None: + """Constructor stores the supplied strategy verbatim.""" + sentinel = AsyncMock() + proxy, _, _ = _make_proxy( + resolver=_StubResolver(None), connect_strategy=sentinel + ) + assert proxy._connect_strategy is sentinel + + +@pytest.mark.asyncio +async def test_handle_connect_invokes_strategy_with_target_and_timeout() -> None: + """`_handle_connect` resolves the address then calls the strategy.""" + target = "AA:BB:CC:DD:EE:FF" + client = MagicMock() + client.mtu_size = 247 + strategy = AsyncMock(return_value=client) + + proxy, send_success, _ = _make_proxy( + resolver=_StubResolver(target), connect_strategy=strategy + ) + await proxy._handle_connect( + cmd_id=1, args={"address": "AA:BB:CC:DD:EE:FF", "timeout": 8000} + ) + + strategy.assert_awaited_once() + strategy_args = strategy.await_args + assert strategy_args is not None + call_target, _on_disconnect, timeout_ms = strategy_args.args + assert call_target is target + assert callable(_on_disconnect) + assert timeout_ms == 8000 + send_success.assert_awaited_once() + success_args = send_success.await_args + assert success_args is not None + assert success_args.args[1]["mtu"] == 247 + assert 1 in proxy._connections + + +@pytest.mark.asyncio +async def test_handle_connect_reports_device_not_found_without_calling_strategy() -> ( + None +): + """When the resolver returns None the strategy is never invoked.""" + strategy = AsyncMock() + proxy, _, send_error = _make_proxy( + resolver=_StubResolver(None), connect_strategy=strategy + ) + await proxy._handle_connect(cmd_id=2, args={"address": "AA:BB:CC:DD:EE:FF"}) + + strategy.assert_not_awaited() + send_error.assert_awaited_once() + await_args = send_error.await_args + assert await_args is not None + assert await_args.args[1] == "device_not_found" + + +@pytest.mark.asyncio +async def test_handle_connect_maps_timeout_error_to_connection_failed() -> None: + """A `TimeoutError` from the strategy surfaces as `connection_failed`.""" + strategy = AsyncMock(side_effect=TimeoutError()) + proxy, _, send_error = _make_proxy( + resolver=_StubResolver("AA:BB:CC:DD:EE:FF"), connect_strategy=strategy + ) + await proxy._handle_connect(cmd_id=3, args={"address": "AA:BB:CC:DD:EE:FF"}) + + send_error.assert_awaited_once() + await_args = send_error.await_args + assert await_args is not None + code = await_args.args[1] + message = await_args.args[2] + assert code == "connection_failed" + assert "Timeout" in message + + +@pytest.mark.asyncio +async def test_handle_connect_maps_arbitrary_exception_to_connection_failed() -> None: + """Any other exception from the strategy is reported as `connection_failed`.""" + strategy = AsyncMock(side_effect=RuntimeError("boom")) + proxy, _, send_error = _make_proxy( + resolver=_StubResolver("AA:BB:CC:DD:EE:FF"), connect_strategy=strategy + ) + await proxy._handle_connect(cmd_id=4, args={"address": "AA:BB:CC:DD:EE:FF"}) + + send_error.assert_awaited_once() + await_args = send_error.await_args + assert await_args is not None + assert await_args.args[1] == "connection_failed" + assert "boom" in await_args.args[2]