feat(ble-proxy): pluggable connect_strategy hook for MatterBleProxy

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",
 ]

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

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]