Add improv_ble integration (#102129)

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

CODEOWNERS

@@ -586,6 +586,8 @@ build.json @home-assistant/supervisor
 /tests/components/image_upload/ @home-assistant/core
 /homeassistant/components/imap/ @jbouwh
 /tests/components/imap/ @jbouwh
+/homeassistant/components/improv_ble/ @emontnemery
+/tests/components/improv_ble/ @emontnemery
 /homeassistant/components/incomfort/ @zxdavb
 /homeassistant/components/influxdb/ @mdegat01
 /tests/components/influxdb/ @mdegat01

homeassistant/components/improv_ble/__init__.py

@@ -0,0 +1 @@
+"""The Improv BLE integration."""

homeassistant/components/improv_ble/config_flow.py

@@ -0,0 +1,387 @@
+"""Config flow for Improv via BLE integration."""
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Awaitable, Callable, Coroutine
+from dataclasses import dataclass
+import logging
+from typing import Any, TypeVar
+
+from bleak import BleakError
+from improv_ble_client import (
+    SERVICE_DATA_UUID,
+    Error,
+    ImprovBLEClient,
+    ImprovServiceData,
+    State,
+    device_filter,
+    errors as improv_ble_errors,
+)
+import voluptuous as vol
+
+from homeassistant import config_entries
+from homeassistant.components.bluetooth import (
+    BluetoothServiceInfoBleak,
+    async_discovered_service_info,
+    async_last_service_info,
+)
+from homeassistant.const import CONF_ADDRESS
+from homeassistant.data_entry_flow import FlowResult
+
+from .const import DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+_T = TypeVar("_T")
+
+STEP_PROVISION_SCHEMA = vol.Schema(
+    {
+        vol.Required("ssid"): str,
+        vol.Required("password"): str,
+    }
+)
+
+
+class AbortFlow(Exception):
+    """Raised when a flow should be aborted."""
+
+    def __init__(self, reason: str) -> None:
+        """Initialize."""
+        self.reason = reason
+
+
+@dataclass
+class Credentials:
+    """Container for WiFi credentials."""
+
+    password: str
+    ssid: str
+
+
+class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for Improv via BLE."""
+
+    VERSION = 1
+
+    _authorize_task: asyncio.Task | None = None
+    _can_identify: bool | None = None
+    _credentials: Credentials | None = None
+    _provision_result: FlowResult | None = None
+    _provision_task: asyncio.Task | None = None
+    _reauth_entry: config_entries.ConfigEntry | None = None
+    _unsub: Callable[[], None] | None = None
+
+    def __init__(self) -> None:
+        """Initialize the config flow."""
+        self._device: ImprovBLEClient | None = None
+        # Populated by user step
+        self._discovered_devices: dict[str, BluetoothServiceInfoBleak] = {}
+        # Populated by bluetooth, reauth_confirm and user steps
+        self._discovery_info: BluetoothServiceInfoBleak | None = None
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Handle the user step to pick discovered device."""
+        errors: dict[str, str] = {}
+
+        if user_input is not None:
+            address = user_input[CONF_ADDRESS]
+            await self.async_set_unique_id(address, raise_on_progress=False)
+            # Guard against the user selecting a device which has been configured by
+            # another flow.
+            self._abort_if_unique_id_configured()
+            self._discovery_info = self._discovered_devices[address]
+            return await self.async_step_start_improv()
+
+        current_addresses = self._async_current_ids()
+        for discovery in async_discovered_service_info(self.hass):
+            if (
+                discovery.address in current_addresses
+                or discovery.address in self._discovered_devices
+                or not device_filter(discovery.advertisement)
+            ):
+                continue
+            self._discovered_devices[discovery.address] = discovery
+
+        if not self._discovered_devices:
+            return self.async_abort(reason="no_devices_found")
+
+        data_schema = vol.Schema(
+            {
+                vol.Required(CONF_ADDRESS): vol.In(
+                    {
+                        service_info.address: (
+                            f"{service_info.name} ({service_info.address})"
+                        )
+                        for service_info in self._discovered_devices.values()
+                    }
+                ),
+            }
+        )
+        return self.async_show_form(
+            step_id="user",
+            data_schema=data_schema,
+            errors=errors,
+        )
+
+    async def async_step_bluetooth(
+        self, discovery_info: BluetoothServiceInfoBleak
+    ) -> FlowResult:
+        """Handle the Bluetooth discovery step."""
+        await self.async_set_unique_id(discovery_info.address)
+        self._abort_if_unique_id_configured()
+        service_data = discovery_info.service_data
+        improv_service_data = ImprovServiceData.from_bytes(
+            service_data[SERVICE_DATA_UUID]
+        )
+        if improv_service_data.state in (State.PROVISIONING, State.PROVISIONED):
+            _LOGGER.debug(
+                "Device is already provisioned: %s", improv_service_data.state
+            )
+            return self.async_abort(reason="already_provisioned")
+        self._discovery_info = discovery_info
+        name = self._discovery_info.name or self._discovery_info.address
+        self.context["title_placeholders"] = {"name": name}
+        return await self.async_step_bluetooth_confirm()
+
+    async def async_step_bluetooth_confirm(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Handle bluetooth confirm step."""
+        # mypy is not aware that we can't get here without having these set already
+        assert self._discovery_info is not None
+
+        if user_input is None:
+            name = self._discovery_info.name or self._discovery_info.address
+            return self.async_show_form(
+                step_id="bluetooth_confirm",
+                description_placeholders={"name": name},
+            )
+
+        return await self.async_step_start_improv()
+
+    async def async_step_start_improv(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Start improv flow.
+
+        If the device supports identification, show a menu, if it does not,
+        ask for WiFi credentials.
+        """
+        # mypy is not aware that we can't get here without having these set already
+        assert self._discovery_info is not None
+        discovery_info = self._discovery_info = async_last_service_info(
+            self.hass, self._discovery_info.address
+        )
+        if not discovery_info:
+            return self.async_abort(reason="cannot_connect")
+        service_data = discovery_info.service_data
+        improv_service_data = ImprovServiceData.from_bytes(
+            service_data[SERVICE_DATA_UUID]
+        )
+        if improv_service_data.state in (State.PROVISIONING, State.PROVISIONED):
+            _LOGGER.debug(
+                "Device is already provisioned: %s", improv_service_data.state
+            )
+            return self.async_abort(reason="already_provisioned")
+
+        if not self._device:
+            self._device = ImprovBLEClient(discovery_info.device)
+        device = self._device
+
+        if self._can_identify is None:
+            try:
+                self._can_identify = await self._try_call(device.can_identify())
+            except AbortFlow as err:
+                return self.async_abort(reason=err.reason)
+        if self._can_identify:
+            return await self.async_step_main_menu()
+        return await self.async_step_provision()
+
+    async def async_step_main_menu(self, _: None = None) -> FlowResult:
+        """Show the main menu."""
+        return self.async_show_menu(
+            step_id="main_menu",
+            menu_options=[
+                "identify",
+                "provision",
+            ],
+        )
+
+    async def async_step_identify(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Handle identify step."""
+        # mypy is not aware that we can't get here without having these set already
+        assert self._device is not None
+
+        if user_input is None:
+            try:
+                await self._try_call(self._device.identify())
+            except AbortFlow as err:
+                return self.async_abort(reason=err.reason)
+            return self.async_show_form(step_id="identify")
+        return await self.async_step_start_improv()
+
+    async def async_step_provision(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Handle provision step."""
+        # mypy is not aware that we can't get here without having these set already
+        assert self._device is not None
+
+        if user_input is None and self._credentials is None:
+            return self.async_show_form(
+                step_id="provision", data_schema=STEP_PROVISION_SCHEMA
+            )
+        if user_input is not None:
+            self._credentials = Credentials(user_input["password"], user_input["ssid"])
+
+        try:
+            need_authorization = await self._try_call(self._device.need_authorization())
+        except AbortFlow as err:
+            return self.async_abort(reason=err.reason)
+        _LOGGER.debug("Need authorization: %s", need_authorization)
+        if need_authorization:
+            return await self.async_step_authorize()
+        return await self.async_step_do_provision()
+
+    async def async_step_do_provision(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Execute provisioning."""
+
+        async def _do_provision() -> None:
+            # mypy is not aware that we can't get here without having these set already
+            assert self._credentials is not None
+            assert self._device is not None
+
+            errors = {}
+            try:
+                redirect_url = await self._try_call(
+                    self._device.provision(
+                        self._credentials.ssid, self._credentials.password, None
+                    )
+                )
+            except AbortFlow as err:
+                self._provision_result = self.async_abort(reason=err.reason)
+                return
+            except improv_ble_errors.ProvisioningFailed as err:
+                if err.error == Error.NOT_AUTHORIZED:
+                    _LOGGER.debug("Need authorization when calling provision")
+                    self._provision_result = await self.async_step_authorize()
+                    return
+                if err.error == Error.UNABLE_TO_CONNECT:
+                    self._credentials = None
+                    errors["base"] = "unable_to_connect"
+                else:
+                    self._provision_result = self.async_abort(reason="unknown")
+                    return
+            else:
+                _LOGGER.debug("Provision successful, redirect URL: %s", redirect_url)
+                # Abort all flows in progress with same unique ID
+                for flow in self._async_in_progress(include_uninitialized=True):
+                    flow_unique_id = flow["context"].get("unique_id")
+                    if (
+                        flow["flow_id"] != self.flow_id
+                        and self.unique_id == flow_unique_id
+                    ):
+                        self.hass.config_entries.flow.async_abort(flow["flow_id"])
+                if redirect_url:
+                    self._provision_result = self.async_abort(
+                        reason="provision_successful_url",
+                        description_placeholders={"url": redirect_url},
+                    )
+                    return
+                self._provision_result = self.async_abort(reason="provision_successful")
+                return
+            self._provision_result = self.async_show_form(
+                step_id="provision", data_schema=STEP_PROVISION_SCHEMA, errors=errors
+            )
+            return
+
+        if not self._provision_task:
+            self._provision_task = self.hass.async_create_task(
+                self._resume_flow_when_done(_do_provision())
+            )
+            return self.async_show_progress(
+                step_id="do_provision", progress_action="provisioning"
+            )
+
+        await self._provision_task
+        self._provision_task = None
+        return self.async_show_progress_done(next_step_id="provision_done")
+
+    async def async_step_provision_done(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Show the result of the provision step."""
+        # mypy is not aware that we can't get here without having these set already
+        assert self._provision_result is not None
+
+        result = self._provision_result
+        self._provision_result = None
+        return result
+
+    async def _resume_flow_when_done(self, awaitable: Awaitable) -> None:
+        try:
+            await awaitable
+        finally:
+            self.hass.async_create_task(
+                self.hass.config_entries.flow.async_configure(flow_id=self.flow_id)
+            )
+
+    async def async_step_authorize(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Handle authorize step."""
+        # mypy is not aware that we can't get here without having these set already
+        assert self._device is not None
+
+        _LOGGER.debug("Wait for authorization")
+        if not self._authorize_task:
+            authorized_event = asyncio.Event()
+
+            def on_state_update(state: State) -> None:
+                _LOGGER.debug("State update: %s", state.name)
+                if state != State.AUTHORIZATION_REQUIRED:
+                    authorized_event.set()
+
+            try:
+                self._unsub = await self._try_call(
+                    self._device.subscribe_state_updates(on_state_update)
+                )
+            except AbortFlow as err:
+                return self.async_abort(reason=err.reason)
+
+            self._authorize_task = self.hass.async_create_task(
+                self._resume_flow_when_done(authorized_event.wait())
+            )
+            return self.async_show_progress(
+                step_id="authorize", progress_action="authorize"
+            )
+
+        await self._authorize_task
+        self._authorize_task = None
+        if self._unsub:
+            self._unsub()
+            self._unsub = None
+        return self.async_show_progress_done(next_step_id="provision")
+
+    @staticmethod
+    async def _try_call(func: Coroutine[Any, Any, _T]) -> _T:
+        """Call the library and abort flow on common errors."""
+        try:
+            return await func
+        except BleakError as err:
+            _LOGGER.warning("BleakError", exc_info=err)
+            raise AbortFlow("cannot_connect") from err
+        except improv_ble_errors.CharacteristicMissingError as err:
+            _LOGGER.warning("CharacteristicMissing", exc_info=err)
+            raise AbortFlow("characteristic_missing") from err
+        except improv_ble_errors.CommandFailed:
+            raise
+        except Exception as err:  # pylint: disable=broad-except
+            _LOGGER.exception("Unexpected exception")
+            raise AbortFlow("unknown") from err

homeassistant/components/improv_ble/const.py

@@ -0,0 +1,3 @@
+"""Constants for the Improv BLE integration."""
+
+DOMAIN = "improv_ble"

homeassistant/components/improv_ble/manifest.json

@@ -0,0 +1,17 @@
+{
+  "domain": "improv_ble",
+  "name": "Improv via BLE",
+  "bluetooth": [
+    {
+      "service_uuid": "00467768-6228-2272-4663-277478268000",
+      "service_data_uuid": "00004677-0000-1000-8000-00805f9b34fb"
+    }
+  ],
+  "codeowners": ["@emontnemery"],
+  "config_flow": true,
+  "dependencies": ["bluetooth_adapters"],
+  "documentation": "https://www.home-assistant.io/integrations/improv_ble",
+  "integration_type": "device",
+  "iot_class": "local_polling",
+  "requirements": ["py-improv-ble-client==1.0.2"]
+}

homeassistant/components/improv_ble/strings.json

@@ -0,0 +1,50 @@
+{
+  "config": {
+    "flow_title": "[%key:component::bluetooth::config::flow_title%]",
+    "step": {
+      "user": {
+        "description": "[%key:component::bluetooth::config::step::user::description%]",
+        "data": {
+          "address": "[%key:common::config_flow::data::device%]"
+        }
+      },
+      "bluetooth_confirm": {
+        "description": "[%key:component::bluetooth::config::step::bluetooth_confirm::description%]"
+      },
+      "identify": {
+        "description": "The device is now identifying itself, for example by blinking or beeping."
+      },
+      "main_menu": {
+        "description": "Choose next step.",
+        "menu_options": {
+          "identify": "Identify device",
+          "provision": "Connect device to a Wi-Fi network"
+        }
+      },
+      "provision": {
+        "description": "Enter Wi-Fi credentials to connect the device to your network.",
+        "data": {
+          "password": "Password",
+          "ssid": "SSID"
+        }
+      }
+    },
+    "progress": {
+      "authorize": "The device requires authorization, please press its authorization button or consult the device's manual for how to proceed.",
+      "provisioning": "The device is connecting to the Wi-Fi network."
+    },
+    "error": {
+      "unable_to_connect": "The device could not connect to the Wi-Fi network. Check that the SSID and password are correct and try again."
+    },
+    "abort": {
+      "already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
+      "already_provisioned": "The device is already connected to Wi-Fi. If you want to connect it to another network, try factory resetting it first.",
+      "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
+      "characteristic_missing": "The device is either already connected to Wi-Fi, or no longer able to connect to Wi-Fi. If you want to connect it to another network, try factory resetting it first.",
+      "no_devices_found": "[%key:common::config_flow::abort::no_devices_found%]",
+      "provision_successful": "The device has successfully connected to the Wi-Fi network.",
+      "provision_successful_url": "The device has successfully connected to the Wi-Fi network.\n\nPlease visit {url} to finish setup.",
+      "unknown": "[%key:common::config_flow::error::unknown%]"
+    }
+  }
+}

homeassistant/generated/bluetooth.py

@@ -217,6 +217,11 @@ BLUETOOTH: list[dict[str, bool | str | int | list[int]]] = [
         "domain": "idasen_desk",
         "service_uuid": "99fa0001-338a-1024-8a49-009c0215f78a",
     },
+    {
+        "domain": "improv_ble",
+        "service_data_uuid": "00004677-0000-1000-8000-00805f9b34fb",
+        "service_uuid": "00467768-6228-2272-4663-277478268000",
+    },
     {
         "connectable": False,
         "domain": "inkbird",

homeassistant/generated/config_flows.py

@@ -218,6 +218,7 @@ FLOWS = {
         "idasen_desk",
         "ifttt",
         "imap",
+        "improv_ble",
         "inkbird",
         "insteon",
         "intellifire",

homeassistant/generated/integrations.json

@@ -2614,6 +2614,12 @@
       "config_flow": true,
       "iot_class": "cloud_push"
     },
+    "improv_ble": {
+      "name": "Improv via BLE",
+      "integration_type": "device",
+      "config_flow": true,
+      "iot_class": "local_polling"
+    },
     "incomfort": {
       "name": "Intergas InComfort/Intouch Lan2RF gateway",
       "integration_type": "hub",

requirements_all.txt

@@ -1512,6 +1512,9 @@ py-cpuinfo==9.0.0
 # homeassistant.components.dormakaba_dkey
 py-dormakaba-dkey==1.0.5
 
+# homeassistant.components.improv_ble
+py-improv-ble-client==1.0.2
+
 # homeassistant.components.melissa
 py-melissa-climate==2.1.4
 

requirements_test_all.txt

@@ -1157,6 +1157,9 @@ py-cpuinfo==9.0.0
 # homeassistant.components.dormakaba_dkey
 py-dormakaba-dkey==1.0.5
 
+# homeassistant.components.improv_ble
+py-improv-ble-client==1.0.2
+
 # homeassistant.components.melissa
 py-melissa-climate==2.1.4
 

tests/components/improv_ble/__init__.py

@@ -0,0 +1,60 @@
+"""Tests for the Improv via BLE integration."""
+
+from improv_ble_client import SERVICE_DATA_UUID, SERVICE_UUID
+
+from homeassistant.components.bluetooth import BluetoothServiceInfoBleak
+
+from tests.components.bluetooth import generate_advertisement_data, generate_ble_device
+
+IMPROV_BLE_DISCOVERY_INFO = BluetoothServiceInfoBleak(
+    name="00123456",
+    address="AA:BB:CC:DD:EE:F0",
+    rssi=-60,
+    manufacturer_data={},
+    service_uuids=[SERVICE_UUID],
+    service_data={SERVICE_DATA_UUID: b"\x01\x00\x00\x00\x00\x00"},
+    source="local",
+    device=generate_ble_device(address="AA:BB:CC:DD:EE:F0", name="00123456"),
+    advertisement=generate_advertisement_data(
+        service_uuids=[SERVICE_UUID],
+        service_data={SERVICE_DATA_UUID: b"\x01\x00\x00\x00\x00\x00"},
+    ),
+    time=0,
+    connectable=True,
+)
+
+
+PROVISIONED_IMPROV_BLE_DISCOVERY_INFO = BluetoothServiceInfoBleak(
+    name="00123456",
+    address="AA:BB:CC:DD:EE:F0",
+    rssi=-60,
+    manufacturer_data={},
+    service_uuids=[SERVICE_UUID],
+    service_data={SERVICE_DATA_UUID: b"\x04\x00\x00\x00\x00\x00"},
+    source="local",
+    device=generate_ble_device(address="AA:BB:CC:DD:EE:F0", name="00123456"),
+    advertisement=generate_advertisement_data(
+        service_uuids=[SERVICE_UUID],
+        service_data={SERVICE_DATA_UUID: b"\x04\x00\x00\x00\x00\x00"},
+    ),
+    time=0,
+    connectable=True,
+)
+
+
+NOT_IMPROV_BLE_DISCOVERY_INFO = BluetoothServiceInfoBleak(
+    name="Not",
+    address="AA:BB:CC:DD:EE:F2",
+    rssi=-60,
+    manufacturer_data={
+        33: b"\x00\x00\xd1\xf0b;\xd8\x1dE\xd6\xba\xeeL\xdd]\xf5\xb2\xe9",
+        21: b"\x061\x00Z\x8f\x93\xb2\xec\x85\x06\x00i\x00\x02\x02Q\xed\x1d\xf0",
+    },
+    service_uuids=[],
+    service_data={},
+    source="local",
+    device=generate_ble_device(address="AA:BB:CC:DD:EE:F2", name="Aug"),
+    advertisement=generate_advertisement_data(),
+    time=0,
+    connectable=True,
+)

tests/components/improv_ble/conftest.py

@@ -0,0 +1,8 @@
+"""Improv via BLE test fixtures."""
+
+import pytest
+
+
+@pytest.fixture(autouse=True)
+def mock_bluetooth(enable_bluetooth):
+    """Auto mock bluetooth."""

tests/components/improv_ble/test_config_flow.py

@@ -0,0 +1,717 @@
+"""Test the Improv via BLE config flow."""
+from collections.abc import Callable
+from unittest.mock import patch
+
+from bleak.exc import BleakError
+from improv_ble_client import Error, State, errors as improv_ble_errors
+import pytest
+
+from homeassistant import config_entries
+from homeassistant.components.improv_ble.const import DOMAIN
+from homeassistant.const import CONF_ADDRESS
+from homeassistant.core import HomeAssistant
+from homeassistant.data_entry_flow import FlowResult, FlowResultType
+
+from . import (
+    IMPROV_BLE_DISCOVERY_INFO,
+    NOT_IMPROV_BLE_DISCOVERY_INFO,
+    PROVISIONED_IMPROV_BLE_DISCOVERY_INFO,
+)
+
+IMPROV_BLE = "homeassistant.components.improv_ble"
+
+
+@pytest.mark.parametrize(
+    ("url", "abort_reason", "placeholders"),
+    [
+        ("http://bla.local", "provision_successful_url", {"url": "http://bla.local"}),
+        (None, "provision_successful", None),
+    ],
+)
+async def test_user_step_success(
+    hass: HomeAssistant,
+    url: str | None,
+    abort_reason: str | None,
+    placeholders: dict[str, str] | None,
+) -> None:
+    """Test user step success path."""
+    with patch(
+        f"{IMPROV_BLE}.config_flow.async_discovered_service_info",
+        return_value=[NOT_IMPROV_BLE_DISCOVERY_INFO, IMPROV_BLE_DISCOVERY_INFO],
+    ):
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN, context={"source": config_entries.SOURCE_USER}
+        )
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "user"
+    assert result["errors"] == {}
+
+    with patch(
+        f"{IMPROV_BLE}.config_flow.async_last_service_info",
+        return_value=IMPROV_BLE_DISCOVERY_INFO,
+    ):
+        await _test_common_success_wo_identify(
+            hass,
+            result,
+            IMPROV_BLE_DISCOVERY_INFO.address,
+            url,
+            abort_reason,
+            placeholders,
+        )
+
+
+async def test_user_step_success_authorize(hass: HomeAssistant) -> None:
+    """Test user step success path."""
+    with patch(
+        f"{IMPROV_BLE}.config_flow.async_discovered_service_info",
+        return_value=[NOT_IMPROV_BLE_DISCOVERY_INFO, IMPROV_BLE_DISCOVERY_INFO],
+    ):
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN, context={"source": config_entries.SOURCE_USER}
+        )
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "user"
+    assert result["errors"] == {}
+
+    with patch(
+        f"{IMPROV_BLE}.config_flow.async_last_service_info",
+        return_value=IMPROV_BLE_DISCOVERY_INFO,
+    ):
+        await _test_common_success_wo_identify_w_authorize(
+            hass, result, IMPROV_BLE_DISCOVERY_INFO.address
+        )
+
+
+async def test_user_step_no_devices_found(hass: HomeAssistant) -> None:
+    """Test user step with no devices found."""
+    with patch(
+        f"{IMPROV_BLE}.config_flow.async_discovered_service_info",
+        return_value=[
+            PROVISIONED_IMPROV_BLE_DISCOVERY_INFO,
+            NOT_IMPROV_BLE_DISCOVERY_INFO,
+        ],
+    ):
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN, context={"source": config_entries.SOURCE_USER}
+        )
+    assert result["type"] == FlowResultType.ABORT
+    assert result["reason"] == "no_devices_found"
+
+
+async def test_async_step_user_takes_precedence_over_discovery(
+    hass: HomeAssistant,
+) -> None:
+    """Test manual setup takes precedence over discovery."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": config_entries.SOURCE_BLUETOOTH},
+        data=IMPROV_BLE_DISCOVERY_INFO,
+    )
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "bluetooth_confirm"
+
+    with patch(
+        f"{IMPROV_BLE}.config_flow.async_discovered_service_info",
+        return_value=[IMPROV_BLE_DISCOVERY_INFO],
+    ):
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN,
+            context={"source": config_entries.SOURCE_USER},
+        )
+        assert result["type"] == FlowResultType.FORM
+
+    with patch(
+        f"{IMPROV_BLE}.config_flow.async_last_service_info",
+        return_value=IMPROV_BLE_DISCOVERY_INFO,
+    ):
+        await _test_common_success_wo_identify(
+            hass, result, IMPROV_BLE_DISCOVERY_INFO.address
+        )
+
+    # Verify the discovery flow was aborted
+    assert not hass.config_entries.flow.async_progress(DOMAIN)
+
+
+async def test_bluetooth_step_provisioned_device(hass: HomeAssistant) -> None:
+    """Test bluetooth step when device is already provisioned."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": config_entries.SOURCE_BLUETOOTH},
+        data=PROVISIONED_IMPROV_BLE_DISCOVERY_INFO,
+    )
+    assert result["type"] == FlowResultType.ABORT
+    assert result["reason"] == "already_provisioned"
+
+
+async def test_bluetooth_confirm_provisioned_device(hass: HomeAssistant) -> None:
+    """Test bluetooth confirm step when device is already provisioned."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": config_entries.SOURCE_BLUETOOTH},
+        data=IMPROV_BLE_DISCOVERY_INFO,
+    )
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "bluetooth_confirm"
+    assert result["errors"] is None
+
+    with patch(
+        f"{IMPROV_BLE}.config_flow.ImprovBLEClient.can_identify", return_value=False
+    ), patch(
+        f"{IMPROV_BLE}.config_flow.async_last_service_info",
+        return_value=PROVISIONED_IMPROV_BLE_DISCOVERY_INFO,
+    ):
+        result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
+        assert result["type"] == FlowResultType.ABORT
+        assert result["reason"] == "already_provisioned"
+
+
+async def test_bluetooth_confirm_lost_device(hass: HomeAssistant) -> None:
+    """Test bluetooth confirm step when device can no longer be connected to."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": config_entries.SOURCE_BLUETOOTH},
+        data=IMPROV_BLE_DISCOVERY_INFO,
+    )
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "bluetooth_confirm"
+    assert result["errors"] is None
+
+    with patch(
+        f"{IMPROV_BLE}.config_flow.ImprovBLEClient.can_identify", return_value=False
+    ), patch(
+        f"{IMPROV_BLE}.config_flow.async_last_service_info",
+        return_value=None,
+    ):
+        result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
+        assert result["type"] == FlowResultType.ABORT
+        assert result["reason"] == "cannot_connect"
+
+
+async def test_bluetooth_step_success(hass: HomeAssistant) -> None:
+    """Test bluetooth step success path."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": config_entries.SOURCE_BLUETOOTH},
+        data=IMPROV_BLE_DISCOVERY_INFO,
+    )
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "bluetooth_confirm"
+    assert result["errors"] is None
+
+    with patch(
+        f"{IMPROV_BLE}.config_flow.async_last_service_info",
+        return_value=IMPROV_BLE_DISCOVERY_INFO,
+    ):
+        await _test_common_success_wo_identify(
+            hass, result, IMPROV_BLE_DISCOVERY_INFO.address
+        )
+
+
+async def test_bluetooth_step_success_identify(hass: HomeAssistant) -> None:
+    """Test bluetooth step success path."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": config_entries.SOURCE_BLUETOOTH},
+        data=IMPROV_BLE_DISCOVERY_INFO,
+    )
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "bluetooth_confirm"
+    assert result["errors"] is None
+
+    with patch(
+        f"{IMPROV_BLE}.config_flow.async_last_service_info",
+        return_value=IMPROV_BLE_DISCOVERY_INFO,
+    ):
+        await _test_common_success_with_identify(
+            hass, result, IMPROV_BLE_DISCOVERY_INFO.address
+        )
+
+
+async def _test_common_success_with_identify(
+    hass: HomeAssistant, result: FlowResult, address: str
+) -> None:
+    """Test bluetooth and user flow success paths."""
+    with patch(
+        f"{IMPROV_BLE}.config_flow.ImprovBLEClient.can_identify", return_value=True
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {CONF_ADDRESS: address},
+        )
+    assert result["type"] == FlowResultType.MENU
+    assert result["menu_options"] == ["identify", "provision"]
+    assert result["step_id"] == "main_menu"
+
+    with patch(f"{IMPROV_BLE}.config_flow.ImprovBLEClient.identify"):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {"next_step_id": "identify"},
+        )
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "identify"
+    assert result["errors"] is None
+
+    result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
+    assert result["type"] == FlowResultType.MENU
+    assert result["menu_options"] == ["identify", "provision"]
+    assert result["step_id"] == "main_menu"
+
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        {"next_step_id": "provision"},
+    )
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "provision"
+    assert result["errors"] is None
+
+    await _test_common_success(hass, result)
+
+
+async def _test_common_success_wo_identify(
+    hass: HomeAssistant,
+    result: FlowResult,
+    address: str,
+    url: str | None = None,
+    abort_reason: str = "provision_successful",
+    placeholders: dict[str, str] | None = None,
+) -> None:
+    """Test bluetooth and user flow success paths."""
+    with patch(
+        f"{IMPROV_BLE}.config_flow.ImprovBLEClient.can_identify", return_value=False
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {CONF_ADDRESS: address},
+        )
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "provision"
+    assert result["errors"] is None
+
+    await _test_common_success(hass, result)
+
+
+async def _test_common_success(
+    hass: HomeAssistant,
+    result: FlowResult,
+    url: str | None = None,
+    abort_reason: str = "provision_successful",
+    placeholders: dict[str, str] | None = None,
+) -> None:
+    """Test bluetooth and user flow success paths."""
+
+    with patch(
+        f"{IMPROV_BLE}.config_flow.ImprovBLEClient.need_authorization",
+        return_value=False,
+    ), patch(
+        f"{IMPROV_BLE}.config_flow.ImprovBLEClient.provision",
+        return_value=url,
+    ) as mock_provision:
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"], {"ssid": "MyWIFI", "password": "secret"}
+        )
+        assert result["type"] == FlowResultType.SHOW_PROGRESS
+        assert result["progress_action"] == "provisioning"
+        assert result["step_id"] == "do_provision"
+
+        result = await hass.config_entries.flow.async_configure(result["flow_id"])
+        assert result["type"] == FlowResultType.SHOW_PROGRESS_DONE
+        assert result["step_id"] == "provision_done"
+
+    result = await hass.config_entries.flow.async_configure(result["flow_id"])
+    assert result.get("description_placeholders") == placeholders
+    assert result["type"] == FlowResultType.ABORT
+    assert result["reason"] == abort_reason
+
+    mock_provision.assert_awaited_once_with("MyWIFI", "secret", None)
+
+
+async def _test_common_success_wo_identify_w_authorize(
+    hass: HomeAssistant, result: FlowResult, address: str
+) -> None:
+    """Test bluetooth and user flow success paths."""
+    with patch(
+        f"{IMPROV_BLE}.config_flow.ImprovBLEClient.can_identify", return_value=False
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {CONF_ADDRESS: address},
+        )
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "provision"
+    assert result["errors"] is None
+
+    await _test_common_success_w_authorize(hass, result)
+
+
+async def _test_common_success_w_authorize(
+    hass: HomeAssistant, result: FlowResult
+) -> None:
+    """Test bluetooth and user flow success paths."""
+
+    async def subscribe_state_updates(
+        state_callback: Callable[[State], None]
+    ) -> Callable[[], None]:
+        state_callback(State.AUTHORIZED)
+        return lambda: None
+
+    with patch(
+        f"{IMPROV_BLE}.config_flow.ImprovBLEClient.need_authorization",
+        return_value=True,
+    ), patch(
+        f"{IMPROV_BLE}.config_flow.ImprovBLEClient.subscribe_state_updates",
+        side_effect=subscribe_state_updates,
+    ) as mock_subscribe_state_updates:
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"], {"ssid": "MyWIFI", "password": "secret"}
+        )
+        assert result["type"] == FlowResultType.SHOW_PROGRESS
+        assert result["progress_action"] == "authorize"
+        assert result["step_id"] == "authorize"
+        mock_subscribe_state_updates.assert_awaited_once()
+
+        result = await hass.config_entries.flow.async_configure(result["flow_id"])
+        assert result["type"] == FlowResultType.SHOW_PROGRESS_DONE
+        assert result["step_id"] == "provision"
+
+    with patch(
+        f"{IMPROV_BLE}.config_flow.ImprovBLEClient.need_authorization",
+        return_value=False,
+    ), patch(
+        f"{IMPROV_BLE}.config_flow.ImprovBLEClient.provision",
+        return_value="http://blabla.local",
+    ) as mock_provision:
+        result = await hass.config_entries.flow.async_configure(result["flow_id"])
+        assert result["type"] == FlowResultType.SHOW_PROGRESS
+        assert result["progress_action"] == "provisioning"
+        assert result["step_id"] == "do_provision"
+
+        result = await hass.config_entries.flow.async_configure(result["flow_id"])
+        assert result["type"] == FlowResultType.SHOW_PROGRESS_DONE
+        assert result["step_id"] == "provision_done"
+
+    result = await hass.config_entries.flow.async_configure(result["flow_id"])
+    assert result["description_placeholders"] == {"url": "http://blabla.local"}
+    assert result["type"] == FlowResultType.ABORT
+    assert result["reason"] == "provision_successful_url"
+
+    mock_provision.assert_awaited_once_with("MyWIFI", "secret", None)
+
+
+async def test_bluetooth_step_already_in_progress(hass: HomeAssistant) -> None:
+    """Test we can't start a flow for the same device twice."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": config_entries.SOURCE_BLUETOOTH},
+        data=IMPROV_BLE_DISCOVERY_INFO,
+    )
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "bluetooth_confirm"
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": config_entries.SOURCE_BLUETOOTH},
+        data=IMPROV_BLE_DISCOVERY_INFO,
+    )
+    assert result["type"] == FlowResultType.ABORT
+    assert result["reason"] == "already_in_progress"
+
+
+@pytest.mark.parametrize(
+    ("exc", "error"),
+    (
+        (BleakError, "cannot_connect"),
+        (Exception, "unknown"),
+        (improv_ble_errors.CharacteristicMissingError, "characteristic_missing"),
+    ),
+)
+async def test_can_identify_fails(hass: HomeAssistant, exc, error) -> None:
+    """Test bluetooth flow with error."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": config_entries.SOURCE_BLUETOOTH},
+        data=IMPROV_BLE_DISCOVERY_INFO,
+    )
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "bluetooth_confirm"
+    assert result["errors"] is None
+
+    result = await hass.config_entries.flow.async_configure(result["flow_id"])
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "bluetooth_confirm"
+    assert result["errors"] is None
+
+    with patch(
+        f"{IMPROV_BLE}.config_flow.ImprovBLEClient.can_identify", side_effect=exc
+    ), patch(
+        f"{IMPROV_BLE}.config_flow.async_last_service_info",
+        return_value=IMPROV_BLE_DISCOVERY_INFO,
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {CONF_ADDRESS: IMPROV_BLE_DISCOVERY_INFO.address},
+        )
+    assert result["type"] == FlowResultType.ABORT
+    assert result["reason"] == error
+
+
+@pytest.mark.parametrize(
+    ("exc", "error"),
+    (
+        (BleakError, "cannot_connect"),
+        (Exception, "unknown"),
+        (improv_ble_errors.CharacteristicMissingError, "characteristic_missing"),
+    ),
+)
+async def test_identify_fails(hass: HomeAssistant, exc, error) -> None:
+    """Test bluetooth flow with error."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": config_entries.SOURCE_BLUETOOTH},
+        data=IMPROV_BLE_DISCOVERY_INFO,
+    )
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "bluetooth_confirm"
+    assert result["errors"] is None
+
+    result = await hass.config_entries.flow.async_configure(result["flow_id"])
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "bluetooth_confirm"
+    assert result["errors"] is None
+
+    with patch(
+        f"{IMPROV_BLE}.config_flow.ImprovBLEClient.can_identify", return_value=True
+    ), patch(
+        f"{IMPROV_BLE}.config_flow.async_last_service_info",
+        return_value=IMPROV_BLE_DISCOVERY_INFO,
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {CONF_ADDRESS: IMPROV_BLE_DISCOVERY_INFO.address},
+        )
+    assert result["type"] == FlowResultType.MENU
+    assert result["step_id"] == "main_menu"
+
+    with patch(f"{IMPROV_BLE}.config_flow.ImprovBLEClient.identify", side_effect=exc):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {"next_step_id": "identify"},
+        )
+    assert result["type"] == FlowResultType.ABORT
+    assert result["reason"] == error
+
+
+@pytest.mark.parametrize(
+    ("exc", "error"),
+    (
+        (BleakError, "cannot_connect"),
+        (Exception, "unknown"),
+        (improv_ble_errors.CharacteristicMissingError, "characteristic_missing"),
+    ),
+)
+async def test_need_authorization_fails(hass: HomeAssistant, exc, error) -> None:
+    """Test bluetooth flow with error."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": config_entries.SOURCE_BLUETOOTH},
+        data=IMPROV_BLE_DISCOVERY_INFO,
+    )
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "bluetooth_confirm"
+    assert result["errors"] is None
+
+    result = await hass.config_entries.flow.async_configure(result["flow_id"])
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "bluetooth_confirm"
+    assert result["errors"] is None
+
+    with patch(
+        f"{IMPROV_BLE}.config_flow.ImprovBLEClient.can_identify", return_value=False
+    ), patch(
+        f"{IMPROV_BLE}.config_flow.async_last_service_info",
+        return_value=IMPROV_BLE_DISCOVERY_INFO,
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {CONF_ADDRESS: IMPROV_BLE_DISCOVERY_INFO.address},
+        )
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "provision"
+
+    with patch(
+        f"{IMPROV_BLE}.config_flow.ImprovBLEClient.need_authorization", side_effect=exc
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"], {"ssid": "MyWIFI", "password": "secret"}
+        )
+    assert result["type"] == FlowResultType.ABORT
+    assert result["reason"] == error
+
+
+@pytest.mark.parametrize(
+    ("exc", "error"),
+    (
+        (BleakError, "cannot_connect"),
+        (Exception, "unknown"),
+        (improv_ble_errors.CharacteristicMissingError, "characteristic_missing"),
+    ),
+)
+async def test_authorize_fails(hass: HomeAssistant, exc, error) -> None:
+    """Test bluetooth flow with error."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": config_entries.SOURCE_BLUETOOTH},
+        data=IMPROV_BLE_DISCOVERY_INFO,
+    )
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "bluetooth_confirm"
+    assert result["errors"] is None
+
+    result = await hass.config_entries.flow.async_configure(result["flow_id"])
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "bluetooth_confirm"
+    assert result["errors"] is None
+
+    with patch(
+        f"{IMPROV_BLE}.config_flow.ImprovBLEClient.can_identify", return_value=False
+    ), patch(
+        f"{IMPROV_BLE}.config_flow.async_last_service_info",
+        return_value=IMPROV_BLE_DISCOVERY_INFO,
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {CONF_ADDRESS: IMPROV_BLE_DISCOVERY_INFO.address},
+        )
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "provision"
+
+    with patch(
+        f"{IMPROV_BLE}.config_flow.ImprovBLEClient.need_authorization",
+        return_value=True,
+    ), patch(
+        f"{IMPROV_BLE}.config_flow.ImprovBLEClient.subscribe_state_updates",
+        side_effect=exc,
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"], {"ssid": "MyWIFI", "password": "secret"}
+        )
+    assert result["type"] == FlowResultType.ABORT
+    assert result["reason"] == error
+
+
+async def _test_provision_error(hass: HomeAssistant, exc) -> None:
+    """Test bluetooth flow with error."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": config_entries.SOURCE_BLUETOOTH},
+        data=IMPROV_BLE_DISCOVERY_INFO,
+    )
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "bluetooth_confirm"
+    assert result["errors"] is None
+
+    result = await hass.config_entries.flow.async_configure(result["flow_id"])
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "bluetooth_confirm"
+    assert result["errors"] is None
+
+    with patch(
+        f"{IMPROV_BLE}.config_flow.ImprovBLEClient.can_identify", return_value=False
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {CONF_ADDRESS: IMPROV_BLE_DISCOVERY_INFO.address},
+        )
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "provision"
+
+    with patch(
+        f"{IMPROV_BLE}.config_flow.ImprovBLEClient.need_authorization",
+        return_value=False,
+    ), patch(
+        f"{IMPROV_BLE}.config_flow.ImprovBLEClient.provision",
+        side_effect=exc,
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"], {"ssid": "MyWIFI", "password": "secret"}
+        )
+        assert result["type"] == FlowResultType.SHOW_PROGRESS
+        assert result["progress_action"] == "provisioning"
+        assert result["step_id"] == "do_provision"
+
+        result = await hass.config_entries.flow.async_configure(result["flow_id"])
+        assert result["type"] == FlowResultType.SHOW_PROGRESS_DONE
+        assert result["step_id"] == "provision_done"
+
+    return result["flow_id"]
+
+
+@pytest.mark.parametrize(
+    ("exc", "error"),
+    (
+        (BleakError, "cannot_connect"),
+        (Exception, "unknown"),
+        (improv_ble_errors.CharacteristicMissingError, "characteristic_missing"),
+        (improv_ble_errors.ProvisioningFailed(Error.UNKNOWN_ERROR), "unknown"),
+    ),
+)
+async def test_provision_fails(hass: HomeAssistant, exc, error) -> None:
+    """Test bluetooth flow with error."""
+    with patch(
+        f"{IMPROV_BLE}.config_flow.async_last_service_info",
+        return_value=IMPROV_BLE_DISCOVERY_INFO,
+    ):
+        flow_id = await _test_provision_error(hass, exc)
+
+    result = await hass.config_entries.flow.async_configure(flow_id)
+    assert result["type"] == FlowResultType.ABORT
+    assert result["reason"] == error
+
+
+@pytest.mark.parametrize(
+    ("exc", "error"),
+    ((improv_ble_errors.ProvisioningFailed(Error.NOT_AUTHORIZED), "unknown"),),
+)
+async def test_provision_not_authorized(hass: HomeAssistant, exc, error) -> None:
+    """Test bluetooth flow with error."""
+
+    async def subscribe_state_updates(
+        state_callback: Callable[[State], None]
+    ) -> Callable[[], None]:
+        state_callback(State.AUTHORIZED)
+        return lambda: None
+
+    with patch(
+        f"{IMPROV_BLE}.config_flow.ImprovBLEClient.subscribe_state_updates",
+        side_effect=subscribe_state_updates,
+    ), patch(
+        f"{IMPROV_BLE}.config_flow.async_last_service_info",
+        return_value=IMPROV_BLE_DISCOVERY_INFO,
+    ):
+        flow_id = await _test_provision_error(hass, exc)
+    result = await hass.config_entries.flow.async_configure(flow_id)
+    assert result["type"] == FlowResultType.SHOW_PROGRESS
+    assert result["progress_action"] == "authorize"
+    assert result["step_id"] == "authorize"
+
+
+@pytest.mark.parametrize(
+    ("exc", "error"),
+    (
+        (
+            improv_ble_errors.ProvisioningFailed(Error.UNABLE_TO_CONNECT),
+            "unable_to_connect",
+        ),
+    ),
+)
+async def test_provision_retry(hass: HomeAssistant, exc, error) -> None:
+    """Test bluetooth flow with error."""
+    with patch(
+        f"{IMPROV_BLE}.config_flow.async_last_service_info",
+        return_value=IMPROV_BLE_DISCOVERY_INFO,
+    ):
+        flow_id = await _test_provision_error(hass, exc)
+
+    result = await hass.config_entries.flow.async_configure(flow_id)
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "provision"
+    assert result["errors"] == {"base": error}