Fix Shelly BLE rediscovery after factory reset (home-assistant#157113)

Co-authored-by: Copilot <[email protected]>

Author: bdraco

homeassistant/components/shelly/config_flow.py

@@ -33,6 +33,7 @@
 from homeassistant.components.bluetooth import (
     BluetoothServiceInfoBleak,
     async_ble_device_from_address,
+    async_clear_address_from_match_history,
 )
 from homeassistant.config_entries import ConfigFlow, ConfigFlowResult, OptionsFlow
 from homeassistant.const import (
@@ -395,6 +396,14 @@ async def async_step_bluetooth(
         if not mac:
             return self.async_abort(reason="invalid_discovery_info")
 
+        # Clear match history at the start of discovery flow.
+        # This ensures that if the user never provisions the device and it
+        # disappears (powers down), the discovery flow gets cleaned up,
+        # and then the device comes back later, it can be rediscovered.
+        # Also handles factory reset scenarios where the device may reappear
+        # with different advertisement content (RPC-over-BLE re-enabled).
+        async_clear_address_from_match_history(self.hass, discovery_info.address)
+
         # Check if RPC-over-BLE is enabled - required for WiFi provisioning
         if not has_rpc_over_ble(discovery_info.manufacturer_data):
             LOGGER.debug(
@@ -685,6 +694,13 @@ async def _async_provision_wifi_and_wait_for_zeroconf(
         # Secure device after provisioning if requested (disable AP/BLE)
         await self._async_secure_device_after_provision(self.host, self.port)
 
+        # Clear match history so device can be rediscovered if factory reset
+        # This ensures that if the device is factory reset in the future
+        # (re-enabling BLE provisioning), it will trigger a new discovery flow
+        if TYPE_CHECKING:
+            assert self.ble_device is not None
+        async_clear_address_from_match_history(self.hass, self.ble_device.address)
+
         # User just provisioned this device - create entry directly without confirmation
         return self.async_create_entry(
             title=device_info["title"],

tests/components/shelly/test_config_flow.py

@@ -101,6 +101,11 @@
     0x0BA9: bytes.fromhex("0105000b99990a70d6c297bacc")
 }  # Flags (0x01, 0x05, 0x00), Model (0x0b, 0x99, 0x99) - unknown model ID, MAC (0x0a, 0x70, 0xd6, 0xc2, 0x97, 0xba, 0xcc)
 
+BLE_MANUFACTURER_DATA_FOR_CLEAR_TEST = {
+    0x0BA9: bytes.fromhex("0105000b30100a00eeddccbbaa")
+}  # Flags (0x01, 0x05, 0x00), Model (0x0b, 0x30, 0x10), MAC (0x0a, 0x00, 0xee, 0xdd, 0xcc, 0xbb, 0xaa)
+# Device WiFi MAC: 00eeddccbbaa (little-endian) -> AABBCCDDEE00 (reversed to big-endian)
+
 BLE_DISCOVERY_INFO = BluetoothServiceInfoBleak(
     name="ShellyPlus2PM-C049EF8873E8",
     address="AA:BB:CC:DD:EE:FF",
@@ -121,6 +126,26 @@
     tx_power=-127,
 )
 
+BLE_DISCOVERY_INFO_FOR_CLEAR_TEST = BluetoothServiceInfoBleak(
+    name="ShellyPlus2PM-AABBCCDDEE00",
+    address="AA:BB:CC:DD:EE:00",
+    rssi=-60,
+    manufacturer_data=BLE_MANUFACTURER_DATA_FOR_CLEAR_TEST,
+    service_uuids=[],
+    service_data={},
+    source="local",
+    device=generate_ble_device(
+        address="AA:BB:CC:DD:EE:00",
+        name="ShellyPlus2PM-AABBCCDDEE00",
+    ),
+    advertisement=generate_advertisement_data(
+        manufacturer_data=BLE_MANUFACTURER_DATA_FOR_CLEAR_TEST,
+    ),
+    time=0,
+    connectable=True,
+    tx_power=-127,
+)
+
 BLE_DISCOVERY_INFO_NO_RPC = BluetoothServiceInfoBleak(
     name="ShellyPlus2PM-C049EF8873E8",
     address="AA:BB:CC:DD:EE:FF",
@@ -2077,6 +2102,87 @@ async def test_bluetooth_discovery(
     assert len(mock_setup_entry.mock_calls) == 1
 
 
+async def test_bluetooth_provisioning_clears_match_history(
+    hass: HomeAssistant,
+    mock_setup_entry: AsyncMock,
+    mock_setup: AsyncMock,
+) -> None:
+    """Test bluetooth provisioning clears match history at discovery start and after successful provisioning."""
+    # Inject BLE device so it's available in the bluetooth scanner
+    inject_bluetooth_service_info_bleak(hass, BLE_DISCOVERY_INFO_FOR_CLEAR_TEST)
+
+    with patch(
+        "homeassistant.components.shelly.config_flow.async_clear_address_from_match_history",
+    ) as mock_clear:
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN,
+            data=BLE_DISCOVERY_INFO_FOR_CLEAR_TEST,
+            context={"source": config_entries.SOURCE_BLUETOOTH},
+        )
+
+        assert result["type"] is FlowResultType.FORM
+        assert result["step_id"] == "bluetooth_confirm"
+
+        # Confirm
+        with patch(
+            "homeassistant.components.shelly.config_flow.async_scan_wifi_networks",
+            return_value=[{"ssid": "MyNetwork", "rssi": -50, "auth": 2}],
+        ):
+            result = await hass.config_entries.flow.async_configure(
+                result["flow_id"], {}
+            )
+
+        # Select network
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {CONF_SSID: "MyNetwork"},
+        )
+
+        # Reset mock to only count calls during provisioning
+        mock_clear.reset_mock()
+
+        # Enter password and provision
+        with (
+            patch(
+                "homeassistant.components.shelly.config_flow.async_provision_wifi",
+            ),
+            patch(
+                "homeassistant.components.shelly.config_flow.async_lookup_device_by_name",
+                return_value=("1.1.1.1", 80),
+            ),
+            patch(
+                "homeassistant.components.shelly.config_flow.get_info",
+                return_value=MOCK_DEVICE_INFO,
+            ),
+        ):
+            result = await hass.config_entries.flow.async_configure(
+                result["flow_id"],
+                {CONF_PASSWORD: "my_password"},
+            )
+
+            # Provisioning happens in background, shows progress
+            assert result["type"] is FlowResultType.SHOW_PROGRESS
+            await hass.async_block_till_done()
+
+            # Complete provisioning by configuring the progress step
+            result = await hass.config_entries.flow.async_configure(result["flow_id"])
+
+        # Provisioning should complete and create entry
+        assert result["type"] is FlowResultType.CREATE_ENTRY
+        assert result["result"].unique_id == "AABBCCDDEE00"
+
+        # Verify match history was cleared once during provisioning
+        # Only count calls with our test device's address to avoid interference from other tests
+        our_device_calls = [
+            call
+            for call in mock_clear.call_args_list
+            if len(call.args) > 1
+            and call.args[1] == BLE_DISCOVERY_INFO_FOR_CLEAR_TEST.address
+        ]
+        assert our_device_calls
+        mock_clear.assert_called_with(hass, BLE_DISCOVERY_INFO_FOR_CLEAR_TEST.address)
+
+
 @pytest.mark.usefixtures("mock_zeroconf")
 async def test_bluetooth_discovery_no_rpc_over_ble(
     hass: HomeAssistant,
@@ -2092,6 +2198,88 @@ async def test_bluetooth_discovery_no_rpc_over_ble(
     assert result["reason"] == "invalid_discovery_info"
 
 
+async def test_bluetooth_factory_reset_rediscovery(
+    hass: HomeAssistant,
+    mock_setup_entry: AsyncMock,
+    mock_setup: AsyncMock,
+) -> None:
+    """Test device can be rediscovered after factory reset when RPC-over-BLE is re-enabled."""
+    # First discovery: device is already provisioned (no RPC-over-BLE)
+    # Inject the device without RPC so it's in the bluetooth scanner
+    inject_bluetooth_service_info_bleak(hass, BLE_DISCOVERY_INFO_NO_RPC)
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        data=BLE_DISCOVERY_INFO_NO_RPC,
+        context={"source": config_entries.SOURCE_BLUETOOTH},
+    )
+
+    # Should abort because RPC-over-BLE is not enabled
+    assert result["type"] is FlowResultType.ABORT
+    assert result["reason"] == "invalid_discovery_info"
+
+    # Simulate factory reset: device now advertises with RPC-over-BLE enabled
+    # Inject the updated advertisement
+    inject_bluetooth_service_info_bleak(hass, BLE_DISCOVERY_INFO)
+
+    # Second discovery: device after factory reset (RPC-over-BLE now enabled)
+    # Wait for automatic discovery to happen
+    await hass.async_block_till_done()
+
+    # Find the flow that was automatically created
+    flows = hass.config_entries.flow.async_progress()
+    assert len(flows) == 1
+    result = flows[0]
+
+    # Should successfully start config flow since match history was cleared
+    assert result["step_id"] == "bluetooth_confirm"
+    assert (
+        result["context"]["title_placeholders"]["name"] == "ShellyPlus2PM-C049EF8873E8"
+    )
+
+    with patch(
+        "homeassistant.components.shelly.config_flow.async_scan_wifi_networks",
+        return_value=[{"ssid": "MyNetwork", "rssi": -50, "auth": 2}],
+    ):
+        result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
+
+    # Select network
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        {CONF_SSID: "MyNetwork"},
+    )
+
+    # Enter password and provision
+    with (
+        patch(
+            "homeassistant.components.shelly.config_flow.async_provision_wifi",
+        ),
+        patch(
+            "homeassistant.components.shelly.config_flow.async_lookup_device_by_name",
+            return_value=("1.1.1.1", 80),
+        ),
+        patch(
+            "homeassistant.components.shelly.config_flow.get_info",
+            return_value=MOCK_DEVICE_INFO,
+        ),
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {CONF_PASSWORD: "my_password"},
+        )
+
+        # Provisioning happens in background
+        assert result["type"] is FlowResultType.SHOW_PROGRESS
+        await hass.async_block_till_done()
+
+        # Complete provisioning
+        result = await hass.config_entries.flow.async_configure(result["flow_id"])
+
+    # Provisioning should complete and create entry
+    assert result["type"] is FlowResultType.CREATE_ENTRY
+    assert result["result"].unique_id == "C049EF8873E8"
+
+
 @pytest.mark.usefixtures("mock_zeroconf")
 async def test_bluetooth_discovery_invalid_name(
     hass: HomeAssistant,
@@ -2184,6 +2372,41 @@ async def test_bluetooth_discovery_already_configured(
     assert result["reason"] == "already_configured"
 
 
+async def test_bluetooth_discovery_already_configured_clears_match_history(
+    hass: HomeAssistant,
+) -> None:
+    """Test bluetooth discovery clears match history when device already configured."""
+    # Inject BLE device so it's available in the bluetooth scanner
+    inject_bluetooth_service_info_bleak(hass, BLE_DISCOVERY_INFO)
+
+    entry = MockConfigEntry(
+        domain=DOMAIN,
+        unique_id="C049EF8873E8",  # MAC from device name - uppercase no colons
+        data={
+            CONF_HOST: "1.1.1.1",
+            CONF_MODEL: MODEL_PLUS_2PM,
+            CONF_SLEEP_PERIOD: 0,
+            CONF_GEN: 2,
+        },
+    )
+    entry.add_to_hass(hass)
+
+    with patch(
+        "homeassistant.components.shelly.config_flow.async_clear_address_from_match_history"
+    ) as mock_clear:
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN,
+            data=BLE_DISCOVERY_INFO,
+            context={"source": config_entries.SOURCE_BLUETOOTH},
+        )
+
+    assert result["type"] is FlowResultType.ABORT
+    assert result["reason"] == "already_configured"
+
+    # Verify match history was cleared to allow rediscovery if factory reset
+    mock_clear.assert_called_once_with(hass, BLE_DISCOVERY_INFO.address)
+
+
 @pytest.mark.usefixtures("mock_zeroconf")
 async def test_bluetooth_discovery_no_ble_device(
     hass: HomeAssistant,