Explain why Teslemetry Bluetooth pairing failed (#184172)

homeassistant/components/teslemetry/config_flow.py

@@ -9,6 +9,8 @@ from typing import TYPE_CHECKING, Any, cast, override
 from aiohttp import ClientError
 from aiopowerwall import PowerwallAuthenticationError, PowerwallError
 from bleak.exc import BleakError
+from bleak_retry_connector import BleakNotFoundError, BleakOutOfConnectionSlotsError
+from habluetooth.const import STRONG_OWNER_STALE_RSSI
 import probatio
 from tesla_fleet_api.const import (
     AuthorizedClientKeyType,
@@ -42,6 +44,7 @@ from homeassistant.components.application_credentials import (
 )
 from homeassistant.components.bluetooth import (
     async_discovered_service_info,
+    async_last_service_info,
     async_request_active_scan,
     async_scanner_count,
 )
@@ -304,7 +307,27 @@ class VehicleSubentryFlowHandler(ConfigSubentryFlow):
                     except (BleakError, TeslaFleetError, TimeoutError) as err:
                         LOGGER.error("Failed to connect over Bluetooth: %s", err)
                         await self._async_disconnect()
-                        errors["base"] = "cannot_connect"
+                        cause = err.__cause__
+                        last_info = async_last_service_info(
+                            self.hass, device.address, connectable=True
+                        )
+                        if not isinstance(
+                            cause, BleakNotFoundError | BleakOutOfConnectionSlotsError
+                        ):
+                            errors["base"] = "cannot_connect"
+                        # bleak-retry-connector also raises BleakNotFoundError from a final connect timeout.
+                        elif last_info is None or (
+                            isinstance(cause, BleakNotFoundError)
+                            and not isinstance(cause.__cause__, TimeoutError)
+                        ):
+                            errors["base"] = "device_not_found"
+                        elif isinstance(cause, BleakOutOfConnectionSlotsError):
+                            errors["base"] = "no_connection_slot"
+                        # habluetooth treats this signal as a close device, so the timeout is not about range.
+                        elif last_info.rssi >= STRONG_OWNER_STALE_RSSI:
+                            errors["base"] = "vehicle_busy"
+                        else:
+                            errors["base"] = "weak_signal"
                     else:
                         return await self.async_step_pair()
 

homeassistant/components/teslemetry/strings.json

@@ -105,10 +105,13 @@
         "confirm_timeout": "The key was not confirmed on the vehicle's touchscreen in time. Select Submit, tap a key card, then confirm the prompt on the vehicle's touchscreen.",
         "device_not_found": "No matching Tesla vehicle was found nearby over Bluetooth. Make sure the vehicle is awake and within range, then try again.",
         "key_unverified": "The key was added, but the vehicle's reply could not be verified. Select Submit to try again.",
+        "no_connection_slot": "Every Bluetooth adapter or proxy that can reach the vehicle has no free connection slot. Free a connection slot on a Bluetooth proxy, or add another proxy near the vehicle, then try again.",
         "pair_denied": "The key was declined on the vehicle's touchscreen. Select Submit and tap a key card to try again.",
         "pair_failed": "The vehicle rejected the key. Make sure it is not in valet mode and does not already have the maximum number of keys, then try again.",
         "tap_timeout": "The key card was not tapped in time. Select Submit and tap a key card within 30 seconds.",
-        "timeout": "Timed out waiting for the vehicle to add the key. Select Submit and tap a key card on the center console card reader."
+        "timeout": "Timed out waiting for the vehicle to add the key. Select Submit and tap a key card on the center console card reader.",
+        "vehicle_busy": "The vehicle is busy and may already have its maximum number of Bluetooth connections. Disconnect other phones or devices from the vehicle, then try again.",
+        "weak_signal": "The vehicle's Bluetooth signal is too weak to connect. Move the vehicle closer to a Bluetooth adapter, or add a Bluetooth proxy near it, then try again."
       },
       "initiate_flow": {
         "user": "Add local vehicle"

tests/components/teslemetry/test_config_flow.py

@@ -14,7 +14,12 @@ from aiopowerwall import (
     PowerwallConnectionError,
     PowerwallFaultError,
 )
-from bleak.exc import BleakError
+from bleak.exc import BleakDeviceNotFoundError, BleakError
+from bleak_retry_connector import (
+    BleakConnectionError,
+    BleakNotFoundError,
+    BleakOutOfConnectionSlotsError,
+)
 from cryptography.hazmat.primitives import serialization
 from cryptography.hazmat.primitives.asymmetric import rsa
 import probatio
@@ -78,6 +83,11 @@ from . import mock_config_entry, setup_platform
 from .const import CONFIG_V1, METADATA, PRODUCTS, UNIQUE_ID
 
 from tests.common import MockConfigEntry
+from tests.components.bluetooth import (
+    generate_advertisement_data,
+    generate_ble_device,
+    inject_advertisement,
+)
 from tests.test_util.aiohttp import AiohttpClientMocker
 from tests.typing import ClientSessionGenerator
 
@@ -730,7 +740,7 @@ def _discovered_info() -> MagicMock:
     info = MagicMock()
     info.name = TeslaBluetooth().get_name(VIN)
     info.address = ADDRESS
-    info.device = MagicMock()
+    info.device = generate_ble_device(ADDRESS, info.name)
     return info
 
 
@@ -1003,12 +1013,23 @@ async def test_subentry_handshake_fails_after_pairing(hass: HomeAssistant) -> No
     vehicle.pair.assert_awaited_once()
 
 
-@pytest.mark.usefixtures("enable_bluetooth")
-async def test_subentry_scan_connect_fails(hass: HomeAssistant) -> None:
-    """The scan step re-shows the form with an error when BLE connect fails."""
+def _connect_failure(*chain: BaseException) -> BluetoothTransportError:
+    """Return the connect error tesla_fleet_api raises, caused by the given chain."""
+    error = BluetoothTransportError()
+    outer: BaseException = error
+    for cause in chain:
+        outer.__cause__ = cause
+        outer = cause
+    return error
+
+
+async def _connect_fails_then_pairs(
+    hass: HomeAssistant, error: BaseException, expected: str
+) -> None:
+    """Fail the first Bluetooth connect, check the error, then pair on retry."""
     entry = await _setup_account_entry(hass)
     vehicle = _mock_vehicle()
-    vehicle.connect = AsyncMock(side_effect=BleakError("nope"))
+    vehicle.connect = AsyncMock(side_effect=[error, None])
 
     with (
         patch(
@@ -1019,18 +1040,113 @@ async def test_subentry_scan_connect_fails(hass: HomeAssistant) -> None:
             "homeassistant.components.teslemetry.config_flow.async_get_ble_parent",
             return_value=_mock_ble_parent(vehicle),
         ),
+        patch.object(hass.config_entries, "async_schedule_reload"),
     ):
         result = await _start_pairing_at_scan(hass, entry)
         result = await hass.config_entries.subentries.async_configure(
             result["flow_id"], {}
         )
+        assert result["type"] is FlowResultType.FORM
+        assert result["step_id"] == "scan"
+        assert result["errors"] == {"base": expected}
+        # A failed pairing never creates a subentry.
+        assert not entry.get_subentries_of_type(SUBENTRY_TYPE_VEHICLE)
+        vehicle.disconnect.assert_awaited_once()
 
-    assert result["type"] is FlowResultType.FORM
-    assert result["step_id"] == "scan"
-    assert result["errors"] == {"base": "cannot_connect"}
-    # A failed pairing never creates a subentry.
-    assert not entry.get_subentries_of_type(SUBENTRY_TYPE_VEHICLE)
-    vehicle.disconnect.assert_awaited_once()
+        result = await hass.config_entries.subentries.async_configure(
+            result["flow_id"], {}
+        )
+        await hass.async_block_till_done()
+
+    assert result["type"] is FlowResultType.CREATE_ENTRY
+    subentries = entry.get_subentries_of_type(SUBENTRY_TYPE_VEHICLE)
+    assert len(subentries) == 1
+    assert subentries[0].data == {CONF_VIN: VIN, CONF_ADDRESS: ADDRESS}
+
+
+@pytest.mark.parametrize(
+    ("error", "rssi", "expected"),
+    [
+        pytest.param(
+            _connect_failure(
+                BleakOutOfConnectionSlotsError("no slot"), BleakError("no slot")
+            ),
+            -60,
+            "no_connection_slot",
+            id="out_of_slots",
+        ),
+        pytest.param(
+            _connect_failure(BleakNotFoundError("timeout"), TimeoutError()),
+            -62,
+            "vehicle_busy",
+            id="timeout_strong_signal",
+        ),
+        pytest.param(
+            _connect_failure(BleakNotFoundError("timeout"), TimeoutError()),
+            -70,
+            "vehicle_busy",
+            id="timeout_at_strong_threshold",
+        ),
+        pytest.param(
+            _connect_failure(BleakNotFoundError("timeout"), TimeoutError()),
+            -71,
+            "weak_signal",
+            id="timeout_weak_signal",
+        ),
+        pytest.param(
+            _connect_failure(
+                BleakNotFoundError("missing"), BleakDeviceNotFoundError(ADDRESS)
+            ),
+            -60,
+            "device_not_found",
+            id="device_vanished",
+        ),
+        pytest.param(
+            _connect_failure(BleakConnectionError("failed")),
+            -60,
+            "cannot_connect",
+            id="other_connection_error",
+        ),
+        pytest.param(BleakError("nope"), -60, "cannot_connect", id="unwrapped_error"),
+    ],
+)
+@pytest.mark.usefixtures("enable_bluetooth")
+async def test_subentry_scan_connect_fails(
+    hass: HomeAssistant,
+    error: BleakError | BluetoothTransportError,
+    rssi: int,
+    expected: str,
+) -> None:
+    """The scan step explains why the Bluetooth connect failed, then pairs on retry."""
+    inject_advertisement(
+        hass,
+        generate_ble_device(ADDRESS, TeslaBluetooth().get_name(VIN)),
+        generate_advertisement_data(rssi=rssi),
+    )
+    await _connect_fails_then_pairs(hass, error, expected)
+
+
+@pytest.mark.parametrize(
+    "error",
+    [
+        pytest.param(
+            _connect_failure(
+                BleakOutOfConnectionSlotsError("no slot"), BleakError("no slot")
+            ),
+            id="out_of_slots",
+        ),
+        pytest.param(
+            _connect_failure(BleakNotFoundError("timeout"), TimeoutError()),
+            id="timeout",
+        ),
+    ],
+)
+@pytest.mark.usefixtures("enable_bluetooth")
+async def test_subentry_scan_connect_fails_vehicle_gone(
+    hass: HomeAssistant, error: BluetoothTransportError
+) -> None:
+    """A vehicle no longer advertising when the connect fails is reported not found."""
+    await _connect_fails_then_pairs(hass, error, "device_not_found")
 
 
 @pytest.mark.parametrize(