diff --git a/CODEOWNERS b/CODEOWNERS index e967f4d65e4..5441fea97d1 100644 --- a/CODEOWNERS +++ b/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 diff --git a/homeassistant/components/improv_ble/__init__.py b/homeassistant/components/improv_ble/__init__.py new file mode 100644 index 00000000000..985684cb5b8 --- /dev/null +++ b/homeassistant/components/improv_ble/__init__.py @@ -0,0 +1 @@ +"""The Improv BLE integration.""" diff --git a/homeassistant/components/improv_ble/config_flow.py b/homeassistant/components/improv_ble/config_flow.py new file mode 100644 index 00000000000..8776227fc53 --- /dev/null +++ b/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 diff --git a/homeassistant/components/improv_ble/const.py b/homeassistant/components/improv_ble/const.py new file mode 100644 index 00000000000..0641773a055 --- /dev/null +++ b/homeassistant/components/improv_ble/const.py @@ -0,0 +1,3 @@ +"""Constants for the Improv BLE integration.""" + +DOMAIN = "improv_ble" diff --git a/homeassistant/components/improv_ble/manifest.json b/homeassistant/components/improv_ble/manifest.json new file mode 100644 index 00000000000..201bd206490 --- /dev/null +++ b/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"] +} diff --git a/homeassistant/components/improv_ble/strings.json b/homeassistant/components/improv_ble/strings.json new file mode 100644 index 00000000000..48b13f6b782 --- /dev/null +++ b/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%]" + } + } +} diff --git a/homeassistant/generated/bluetooth.py b/homeassistant/generated/bluetooth.py index c2b24b68d29..13700a4521c 100644 --- a/homeassistant/generated/bluetooth.py +++ b/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", diff --git a/homeassistant/generated/config_flows.py b/homeassistant/generated/config_flows.py index fa83d93c87b..64806d8fb86 100644 --- a/homeassistant/generated/config_flows.py +++ b/homeassistant/generated/config_flows.py @@ -218,6 +218,7 @@ FLOWS = { "idasen_desk", "ifttt", "imap", + "improv_ble", "inkbird", "insteon", "intellifire", diff --git a/homeassistant/generated/integrations.json b/homeassistant/generated/integrations.json index fb42c7f0e8e..b89139d7447 100644 --- a/homeassistant/generated/integrations.json +++ b/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", diff --git a/requirements_all.txt b/requirements_all.txt index fa334ab6ea6..f7e9af4484c 100644 --- a/requirements_all.txt +++ b/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 diff --git a/requirements_test_all.txt b/requirements_test_all.txt index c28a50a81ee..fce13c56860 100644 --- a/requirements_test_all.txt +++ b/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 diff --git a/tests/components/improv_ble/__init__.py b/tests/components/improv_ble/__init__.py new file mode 100644 index 00000000000..f1c83bbc0d7 --- /dev/null +++ b/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, +) diff --git a/tests/components/improv_ble/conftest.py b/tests/components/improv_ble/conftest.py new file mode 100644 index 00000000000..ea548efeb15 --- /dev/null +++ b/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.""" diff --git a/tests/components/improv_ble/test_config_flow.py b/tests/components/improv_ble/test_config_flow.py new file mode 100644 index 00000000000..60228b409c2 --- /dev/null +++ b/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}