From 8e0c26bf8697d5dae554efc2285129bd2aae47ef Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Mon, 29 Aug 2022 11:38:18 -0500 Subject: [PATCH] Add LED BLE integration (#77489) Co-authored-by: Paulus Schoutsen --- .coveragerc | 3 + CODEOWNERS | 2 + homeassistant/components/led_ble/__init__.py | 119 +++++++++ .../components/led_ble/config_flow.py | 117 +++++++++ homeassistant/components/led_ble/const.py | 10 + homeassistant/components/led_ble/light.py | 99 ++++++++ .../components/led_ble/manifest.json | 17 ++ homeassistant/components/led_ble/models.py | 17 ++ homeassistant/components/led_ble/strings.json | 23 ++ .../components/led_ble/translations/en.json | 23 ++ homeassistant/components/led_ble/util.py | 51 ++++ homeassistant/generated/bluetooth.py | 20 ++ homeassistant/generated/config_flows.py | 1 + requirements_all.txt | 3 + requirements_test_all.txt | 3 + tests/components/led_ble/__init__.py | 51 ++++ tests/components/led_ble/conftest.py | 8 + tests/components/led_ble/test_config_flow.py | 229 ++++++++++++++++++ 18 files changed, 796 insertions(+) create mode 100644 homeassistant/components/led_ble/__init__.py create mode 100644 homeassistant/components/led_ble/config_flow.py create mode 100644 homeassistant/components/led_ble/const.py create mode 100644 homeassistant/components/led_ble/light.py create mode 100644 homeassistant/components/led_ble/manifest.json create mode 100644 homeassistant/components/led_ble/models.py create mode 100644 homeassistant/components/led_ble/strings.json create mode 100644 homeassistant/components/led_ble/translations/en.json create mode 100644 homeassistant/components/led_ble/util.py create mode 100644 tests/components/led_ble/__init__.py create mode 100644 tests/components/led_ble/conftest.py create mode 100644 tests/components/led_ble/test_config_flow.py diff --git a/.coveragerc b/.coveragerc index 365f64076b3..9301edcec52 100644 --- a/.coveragerc +++ b/.coveragerc @@ -659,6 +659,9 @@ omit = homeassistant/components/lcn/helpers.py homeassistant/components/lcn/scene.py homeassistant/components/lcn/services.py + homeassistant/components/led_ble/__init__.py + homeassistant/components/led_ble/light.py + homeassistant/components/led_ble/util.py homeassistant/components/lg_netcast/media_player.py homeassistant/components/lg_soundbar/media_player.py homeassistant/components/life360/__init__.py diff --git a/CODEOWNERS b/CODEOWNERS index 004bc365d89..fe634225124 100644 --- a/CODEOWNERS +++ b/CODEOWNERS @@ -601,6 +601,8 @@ build.json @home-assistant/supervisor /tests/components/laundrify/ @xLarry /homeassistant/components/lcn/ @alengwenus /tests/components/lcn/ @alengwenus +/homeassistant/components/led_ble/ @bdraco +/tests/components/led_ble/ @bdraco /homeassistant/components/lg_netcast/ @Drafteed /homeassistant/components/life360/ @pnbruckner /tests/components/life360/ @pnbruckner diff --git a/homeassistant/components/led_ble/__init__.py b/homeassistant/components/led_ble/__init__.py new file mode 100644 index 00000000000..d885b3eb950 --- /dev/null +++ b/homeassistant/components/led_ble/__init__.py @@ -0,0 +1,119 @@ +"""The LED BLE integration.""" +from __future__ import annotations + +import asyncio +from datetime import timedelta +import logging + +import async_timeout +from led_ble import BLEAK_EXCEPTIONS, LEDBLE + +from homeassistant.components import bluetooth +from homeassistant.components.bluetooth.match import ADDRESS, BluetoothCallbackMatcher +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import CONF_ADDRESS, EVENT_HOMEASSISTANT_STOP, Platform +from homeassistant.core import Event, HomeAssistant, callback +from homeassistant.exceptions import ConfigEntryNotReady +from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed + +from .const import DEVICE_TIMEOUT, DOMAIN, UPDATE_SECONDS +from .models import LEDBLEData + +PLATFORMS: list[Platform] = [Platform.LIGHT] + +_LOGGER = logging.getLogger(__name__) + + +async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: + """Set up LED BLE from a config entry.""" + address: str = entry.data[CONF_ADDRESS] + ble_device = bluetooth.async_ble_device_from_address(hass, address.upper(), True) + if not ble_device: + raise ConfigEntryNotReady( + f"Could not find LED BLE device with address {address}" + ) + + led_ble = LEDBLE(ble_device) + + @callback + def _async_update_ble( + service_info: bluetooth.BluetoothServiceInfoBleak, + change: bluetooth.BluetoothChange, + ) -> None: + """Update from a ble callback.""" + led_ble.set_ble_device(service_info.device) + + entry.async_on_unload( + bluetooth.async_register_callback( + hass, + _async_update_ble, + BluetoothCallbackMatcher({ADDRESS: address}), + bluetooth.BluetoothScanningMode.PASSIVE, + ) + ) + + async def _async_update(): + """Update the device state.""" + try: + await led_ble.update() + except BLEAK_EXCEPTIONS as ex: + raise UpdateFailed(str(ex)) from ex + + startup_event = asyncio.Event() + cancel_first_update = led_ble.register_callback(lambda *_: startup_event.set()) + coordinator = DataUpdateCoordinator( + hass, + _LOGGER, + name=led_ble.name, + update_method=_async_update, + update_interval=timedelta(seconds=UPDATE_SECONDS), + ) + + try: + await coordinator.async_config_entry_first_refresh() + except ConfigEntryNotReady: + cancel_first_update() + raise + + try: + async with async_timeout.timeout(DEVICE_TIMEOUT): + await startup_event.wait() + except asyncio.TimeoutError as ex: + raise ConfigEntryNotReady( + "Unable to communicate with the device; " + f"Try moving the Bluetooth adapter closer to {led_ble.name}" + ) from ex + finally: + cancel_first_update() + + hass.data.setdefault(DOMAIN, {})[entry.entry_id] = LEDBLEData( + entry.title, led_ble, coordinator + ) + + await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) + entry.async_on_unload(entry.add_update_listener(_async_update_listener)) + + async def _async_stop(event: Event) -> None: + """Close the connection.""" + await led_ble.stop() + + entry.async_on_unload( + hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, _async_stop) + ) + return True + + +async def _async_update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None: + """Handle options update.""" + data: LEDBLEData = hass.data[DOMAIN][entry.entry_id] + if entry.title != data.title: + await hass.config_entries.async_reload(entry.entry_id) + + +async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: + """Unload a config entry.""" + if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS): + data: LEDBLEData = hass.data[DOMAIN].pop(entry.entry_id) + await data.device.stop() + + return unload_ok diff --git a/homeassistant/components/led_ble/config_flow.py b/homeassistant/components/led_ble/config_flow.py new file mode 100644 index 00000000000..19be92f6647 --- /dev/null +++ b/homeassistant/components/led_ble/config_flow.py @@ -0,0 +1,117 @@ +"""Config flow for LEDBLE integration.""" +from __future__ import annotations + +import logging +from typing import Any + +from led_ble import BLEAK_EXCEPTIONS, LEDBLE +import voluptuous as vol + +from homeassistant import config_entries +from homeassistant.components.bluetooth import ( + BluetoothServiceInfoBleak, + async_discovered_service_info, +) +from homeassistant.const import CONF_ADDRESS +from homeassistant.data_entry_flow import FlowResult + +from .const import DOMAIN, LOCAL_NAMES, UNSUPPORTED_SUB_MODEL +from .util import human_readable_name + +_LOGGER = logging.getLogger(__name__) + + +class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): + """Handle a config flow for Yale Access Bluetooth.""" + + VERSION = 1 + + def __init__(self) -> None: + """Initialize the config flow.""" + self._discovery_info: BluetoothServiceInfoBleak | None = None + self._discovered_devices: dict[str, BluetoothServiceInfoBleak] = {} + + async def async_step_bluetooth( + self, discovery_info: BluetoothServiceInfoBleak + ) -> FlowResult: + """Handle the bluetooth discovery step.""" + if discovery_info.name.startswith(UNSUPPORTED_SUB_MODEL): + # These versions speak a different protocol + # that we do not support yet. + return self.async_abort(reason="not_supported") + await self.async_set_unique_id(discovery_info.address) + self._abort_if_unique_id_configured() + self._discovery_info = discovery_info + self.context["title_placeholders"] = { + "name": human_readable_name( + None, discovery_info.name, discovery_info.address + ) + } + return await self.async_step_user() + + 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] + discovery_info = self._discovered_devices[address] + local_name = discovery_info.name + await self.async_set_unique_id( + discovery_info.address, raise_on_progress=False + ) + self._abort_if_unique_id_configured() + led_ble = LEDBLE(discovery_info.device) + try: + await led_ble.update() + except BLEAK_EXCEPTIONS: + errors["base"] = "cannot_connect" + except Exception: # pylint: disable=broad-except + _LOGGER.exception("Unexpected error") + errors["base"] = "unknown" + else: + await led_ble.stop() + return self.async_create_entry( + title=local_name, + data={ + CONF_ADDRESS: discovery_info.address, + }, + ) + + if discovery := self._discovery_info: + self._discovered_devices[discovery.address] = discovery + else: + 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 any( + discovery.name.startswith(local_name) + and not discovery.name.startswith(UNSUPPORTED_SUB_MODEL) + for local_name in LOCAL_NAMES + ) + ): + continue + self._discovered_devices[discovery.address] = discovery + + if not self._discovered_devices: + return self.async_abort(reason="no_unconfigured_devices") + + 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, + ) diff --git a/homeassistant/components/led_ble/const.py b/homeassistant/components/led_ble/const.py new file mode 100644 index 00000000000..ad3ea8f6707 --- /dev/null +++ b/homeassistant/components/led_ble/const.py @@ -0,0 +1,10 @@ +"""Constants for the LED BLE integration.""" + +DOMAIN = "led_ble" + +DEVICE_TIMEOUT = 30 +LOCAL_NAMES = {"LEDnet", "BLE-LED", "LEDBLE", "Triones", "LEDBlue"} + +UNSUPPORTED_SUB_MODEL = "LEDnetWF" + +UPDATE_SECONDS = 15 diff --git a/homeassistant/components/led_ble/light.py b/homeassistant/components/led_ble/light.py new file mode 100644 index 00000000000..a18ab812b19 --- /dev/null +++ b/homeassistant/components/led_ble/light.py @@ -0,0 +1,99 @@ +"""LED BLE integration light platform.""" +from __future__ import annotations + +from typing import Any + +from led_ble import LEDBLE + +from homeassistant.components.light import ( + ATTR_BRIGHTNESS, + ATTR_RGB_COLOR, + ATTR_WHITE, + ColorMode, + LightEntity, +) +from homeassistant.config_entries import ConfigEntry +from homeassistant.core import HomeAssistant, callback +from homeassistant.helpers import device_registry as dr +from homeassistant.helpers.entity import DeviceInfo +from homeassistant.helpers.entity_platform import AddEntitiesCallback +from homeassistant.helpers.update_coordinator import ( + CoordinatorEntity, + DataUpdateCoordinator, +) + +from .const import DOMAIN +from .models import LEDBLEData + + +async def async_setup_entry( + hass: HomeAssistant, + entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, +) -> None: + """Set up the light platform for LEDBLE.""" + data: LEDBLEData = hass.data[DOMAIN][entry.entry_id] + async_add_entities([LEDBLEEntity(data.coordinator, data.device, entry.title)]) + + +class LEDBLEEntity(CoordinatorEntity, LightEntity): + """Representation of LEDBLE device.""" + + _attr_supported_color_modes = {ColorMode.RGB, ColorMode.WHITE} + _attr_has_entity_name = True + + def __init__( + self, coordinator: DataUpdateCoordinator, device: LEDBLE, name: str + ) -> None: + """Initialize an ledble light.""" + super().__init__(coordinator) + self._device = device + self._attr_unique_id = device._address + self._attr_device_info = DeviceInfo( + name=name, + model=hex(device.model_num), + sw_version=hex(device.version_num), + connections={(dr.CONNECTION_BLUETOOTH, device._address)}, + ) + self._async_update_attrs() + + @callback + def _async_update_attrs(self) -> None: + """Handle updating _attr values.""" + device = self._device + self._attr_color_mode = ColorMode.WHITE if device.w else ColorMode.RGB + self._attr_brightness = device.brightness + self._attr_rgb_color = device.rgb_unscaled + self._attr_is_on = device.on + + async def async_turn_on(self, **kwargs: Any) -> None: + """Instruct the light to turn on.""" + brightness = kwargs.get(ATTR_BRIGHTNESS, self.brightness) + if ATTR_RGB_COLOR in kwargs: + rgb = kwargs[ATTR_RGB_COLOR] + await self._device.set_rgb(rgb, brightness) + return + if ATTR_BRIGHTNESS in kwargs: + await self._device.set_brightness(brightness) + return + if ATTR_WHITE in kwargs: + await self._device.set_white(kwargs[ATTR_WHITE]) + return + await self._device.turn_on() + + async def async_turn_off(self, **kwargs: Any) -> None: + """Instruct the light to turn off.""" + await self._device.turn_off() + + @callback + def _handle_coordinator_update(self, *args: Any) -> None: + """Handle data update.""" + self._async_update_attrs() + self.async_write_ha_state() + + async def async_added_to_hass(self) -> None: + """Register callbacks.""" + self.async_on_remove( + self._device.register_callback(self._handle_coordinator_update) + ) + return await super().async_added_to_hass() diff --git a/homeassistant/components/led_ble/manifest.json b/homeassistant/components/led_ble/manifest.json new file mode 100644 index 00000000000..376fadcb3be --- /dev/null +++ b/homeassistant/components/led_ble/manifest.json @@ -0,0 +1,17 @@ +{ + "domain": "led_ble", + "name": "LED BLE", + "config_flow": true, + "documentation": "https://www.home-assistant.io/integrations/ble_ble", + "requirements": ["led-ble==0.5.4"], + "dependencies": ["bluetooth"], + "codeowners": ["@bdraco"], + "bluetooth": [ + { "local_name": "LEDnet*" }, + { "local_name": "BLE-LED*" }, + { "local_name": "LEDBLE*" }, + { "local_name": "Triones*" }, + { "local_name": "LEDBlue*" } + ], + "iot_class": "local_polling" +} diff --git a/homeassistant/components/led_ble/models.py b/homeassistant/components/led_ble/models.py new file mode 100644 index 00000000000..611d484ea61 --- /dev/null +++ b/homeassistant/components/led_ble/models.py @@ -0,0 +1,17 @@ +"""The led ble integration models.""" +from __future__ import annotations + +from dataclasses import dataclass + +from led_ble import LEDBLE + +from homeassistant.helpers.update_coordinator import DataUpdateCoordinator + + +@dataclass +class LEDBLEData: + """Data for the led ble integration.""" + + title: str + device: LEDBLE + coordinator: DataUpdateCoordinator diff --git a/homeassistant/components/led_ble/strings.json b/homeassistant/components/led_ble/strings.json new file mode 100644 index 00000000000..79540552575 --- /dev/null +++ b/homeassistant/components/led_ble/strings.json @@ -0,0 +1,23 @@ +{ + "config": { + "flow_title": "{name}", + "step": { + "user": { + "data": { + "address": "Bluetooth address" + } + } + }, + "error": { + "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]", + "unknown": "[%key:common::config_flow::error::unknown%]" + }, + "abort": { + "not_supported": "Device not supported", + "already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]", + "already_configured": "[%key:common::config_flow::abort::already_configured_device%]", + "no_unconfigured_devices": "No unconfigured devices found.", + "no_devices_found": "[%key:common::config_flow::abort::no_devices_found%]" + } + } +} diff --git a/homeassistant/components/led_ble/translations/en.json b/homeassistant/components/led_ble/translations/en.json new file mode 100644 index 00000000000..75356b78460 --- /dev/null +++ b/homeassistant/components/led_ble/translations/en.json @@ -0,0 +1,23 @@ +{ + "config": { + "abort": { + "already_configured": "Device is already configured", + "already_in_progress": "Configuration flow is already in progress", + "no_devices_found": "No devices found on the network", + "no_unconfigured_devices": "No unconfigured devices found.", + "not_supported": "Device not supported" + }, + "error": { + "cannot_connect": "Failed to connect", + "unknown": "Unexpected error" + }, + "flow_title": "{name}", + "step": { + "user": { + "data": { + "address": "Bluetooth address" + } + } + } + } +} \ No newline at end of file diff --git a/homeassistant/components/led_ble/util.py b/homeassistant/components/led_ble/util.py new file mode 100644 index 00000000000..e43655e2905 --- /dev/null +++ b/homeassistant/components/led_ble/util.py @@ -0,0 +1,51 @@ +"""The yalexs_ble integration models.""" +from __future__ import annotations + +from homeassistant.components.bluetooth import ( + BluetoothScanningMode, + BluetoothServiceInfoBleak, + async_discovered_service_info, + async_process_advertisements, +) +from homeassistant.components.bluetooth.match import ADDRESS, BluetoothCallbackMatcher +from homeassistant.core import HomeAssistant, callback + +from .const import DEVICE_TIMEOUT + + +@callback +def async_find_existing_service_info( + hass: HomeAssistant, local_name: str, address: str +) -> BluetoothServiceInfoBleak | None: + """Return the service info for the given local_name and address.""" + for service_info in async_discovered_service_info(hass): + device = service_info.device + if device.address == address: + return service_info + return None + + +async def async_get_service_info( + hass: HomeAssistant, local_name: str, address: str +) -> BluetoothServiceInfoBleak: + """Wait for the service info for the given local_name and address.""" + if service_info := async_find_existing_service_info(hass, local_name, address): + return service_info + return await async_process_advertisements( + hass, + lambda service_info: True, + BluetoothCallbackMatcher({ADDRESS: address}), + BluetoothScanningMode.ACTIVE, + DEVICE_TIMEOUT, + ) + + +def short_address(address: str) -> str: + """Convert a Bluetooth address to a short address.""" + split_address = address.replace("-", ":").split(":") + return f"{split_address[-2].upper()}{split_address[-1].upper()}"[-4:] + + +def human_readable_name(name: str | None, local_name: str, address: str) -> str: + """Return a human readable name for the given name, local_name, and address.""" + return f"{name or local_name} ({short_address(address)})" diff --git a/homeassistant/generated/bluetooth.py b/homeassistant/generated/bluetooth.py index bda76859688..320c4c296da 100644 --- a/homeassistant/generated/bluetooth.py +++ b/homeassistant/generated/bluetooth.py @@ -118,6 +118,26 @@ BLUETOOTH: list[dict[str, bool | str | int | list[int]]] = [ "local_name": "tps", "connectable": False }, + { + "domain": "led_ble", + "local_name": "LEDnet*" + }, + { + "domain": "led_ble", + "local_name": "BLE-LED*" + }, + { + "domain": "led_ble", + "local_name": "LEDBLE*" + }, + { + "domain": "led_ble", + "local_name": "Triones*" + }, + { + "domain": "led_ble", + "local_name": "LEDBlue*" + }, { "domain": "moat", "local_name": "Moat_S*", diff --git a/homeassistant/generated/config_flows.py b/homeassistant/generated/config_flows.py index 07a5cdce04f..a9b303eabea 100644 --- a/homeassistant/generated/config_flows.py +++ b/homeassistant/generated/config_flows.py @@ -201,6 +201,7 @@ FLOWS = { "landisgyr_heat_meter", "launch_library", "laundrify", + "led_ble", "lg_soundbar", "life360", "lifx", diff --git a/requirements_all.txt b/requirements_all.txt index 3a52b694167..8cb1e2fcf63 100644 --- a/requirements_all.txt +++ b/requirements_all.txt @@ -961,6 +961,9 @@ lakeside==0.12 # homeassistant.components.laundrify laundrify_aio==1.1.2 +# homeassistant.components.led_ble +led-ble==0.5.4 + # homeassistant.components.foscam libpyfoscam==1.0 diff --git a/requirements_test_all.txt b/requirements_test_all.txt index 244adebff53..c9ee9666ff2 100644 --- a/requirements_test_all.txt +++ b/requirements_test_all.txt @@ -699,6 +699,9 @@ lacrosse-view==0.0.9 # homeassistant.components.laundrify laundrify_aio==1.1.2 +# homeassistant.components.led_ble +led-ble==0.5.4 + # homeassistant.components.foscam libpyfoscam==1.0 diff --git a/tests/components/led_ble/__init__.py b/tests/components/led_ble/__init__.py new file mode 100644 index 00000000000..702b793f57a --- /dev/null +++ b/tests/components/led_ble/__init__.py @@ -0,0 +1,51 @@ +"""Tests for the LED BLE Bluetooth integration.""" +from bleak.backends.device import BLEDevice +from bleak.backends.scanner import AdvertisementData + +from homeassistant.components.bluetooth import BluetoothServiceInfoBleak + +LED_BLE_DISCOVERY_INFO = BluetoothServiceInfoBleak( + name="Triones:F30200000152C", + address="AA:BB:CC:DD:EE:FF", + rssi=-60, + manufacturer_data={}, + service_uuids=[], + service_data={}, + source="local", + device=BLEDevice(address="AA:BB:CC:DD:EE:FF", name="Triones:F30200000152C"), + advertisement=AdvertisementData(), + time=0, + connectable=True, +) + +UNSUPPORTED_LED_BLE_DISCOVERY_INFO = BluetoothServiceInfoBleak( + name="LEDnetWFF30200000152C", + address="AA:BB:CC:DD:EE:FF", + rssi=-60, + manufacturer_data={}, + service_uuids=[], + service_data={}, + source="local", + device=BLEDevice(address="AA:BB:CC:DD:EE:FF", name="LEDnetWFF30200000152C"), + advertisement=AdvertisementData(), + time=0, + connectable=True, +) + + +NOT_LED_BLE_DISCOVERY_INFO = BluetoothServiceInfoBleak( + name="Not", + address="AA:BB:CC:DD:EE:FF", + 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=BLEDevice(address="AA:BB:CC:DD:EE:FF", name="Aug"), + advertisement=AdvertisementData(), + time=0, + connectable=True, +) diff --git a/tests/components/led_ble/conftest.py b/tests/components/led_ble/conftest.py new file mode 100644 index 00000000000..280eb0d6f17 --- /dev/null +++ b/tests/components/led_ble/conftest.py @@ -0,0 +1,8 @@ +"""led_ble session fixtures.""" + +import pytest + + +@pytest.fixture(autouse=True) +def mock_bluetooth(enable_bluetooth): + """Auto mock bluetooth.""" diff --git a/tests/components/led_ble/test_config_flow.py b/tests/components/led_ble/test_config_flow.py new file mode 100644 index 00000000000..6767302af50 --- /dev/null +++ b/tests/components/led_ble/test_config_flow.py @@ -0,0 +1,229 @@ +"""Test the LED BLE Bluetooth config flow.""" +from unittest.mock import patch + +from bleak import BleakError + +from homeassistant import config_entries +from homeassistant.components.led_ble.const import DOMAIN +from homeassistant.const import CONF_ADDRESS +from homeassistant.core import HomeAssistant +from homeassistant.data_entry_flow import FlowResultType + +from . import ( + LED_BLE_DISCOVERY_INFO, + NOT_LED_BLE_DISCOVERY_INFO, + UNSUPPORTED_LED_BLE_DISCOVERY_INFO, +) + +from tests.common import MockConfigEntry + + +async def test_user_step_success(hass: HomeAssistant) -> None: + """Test user step success path.""" + with patch( + "homeassistant.components.led_ble.config_flow.async_discovered_service_info", + return_value=[NOT_LED_BLE_DISCOVERY_INFO, LED_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("homeassistant.components.led_ble.config_flow.LEDBLE.update",), patch( + "homeassistant.components.led_ble.async_setup_entry", + return_value=True, + ) as mock_setup_entry: + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + CONF_ADDRESS: LED_BLE_DISCOVERY_INFO.address, + }, + ) + await hass.async_block_till_done() + + assert result2["type"] == FlowResultType.CREATE_ENTRY + assert result2["title"] == LED_BLE_DISCOVERY_INFO.name + assert result2["data"] == { + CONF_ADDRESS: LED_BLE_DISCOVERY_INFO.address, + } + assert result2["result"].unique_id == LED_BLE_DISCOVERY_INFO.address + assert len(mock_setup_entry.mock_calls) == 1 + + +async def test_user_step_no_devices_found(hass: HomeAssistant) -> None: + """Test user step with no devices found.""" + with patch( + "homeassistant.components.led_ble.config_flow.async_discovered_service_info", + return_value=[NOT_LED_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_unconfigured_devices" + + +async def test_user_step_no_new_devices_found(hass: HomeAssistant) -> None: + """Test user step with only existing devices found.""" + entry = MockConfigEntry( + domain=DOMAIN, + data={ + CONF_ADDRESS: LED_BLE_DISCOVERY_INFO.address, + }, + unique_id=LED_BLE_DISCOVERY_INFO.address, + ) + entry.add_to_hass(hass) + with patch( + "homeassistant.components.led_ble.config_flow.async_discovered_service_info", + return_value=[LED_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_unconfigured_devices" + + +async def test_user_step_cannot_connect(hass: HomeAssistant) -> None: + """Test user step and we cannot connect.""" + with patch( + "homeassistant.components.led_ble.config_flow.async_discovered_service_info", + return_value=[LED_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( + "homeassistant.components.led_ble.config_flow.LEDBLE.update", + side_effect=BleakError, + ): + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + CONF_ADDRESS: LED_BLE_DISCOVERY_INFO.address, + }, + ) + await hass.async_block_till_done() + + assert result2["type"] == FlowResultType.FORM + assert result2["step_id"] == "user" + assert result2["errors"] == {"base": "cannot_connect"} + + with patch("homeassistant.components.led_ble.config_flow.LEDBLE.update",), patch( + "homeassistant.components.led_ble.async_setup_entry", + return_value=True, + ) as mock_setup_entry: + result3 = await hass.config_entries.flow.async_configure( + result2["flow_id"], + { + CONF_ADDRESS: LED_BLE_DISCOVERY_INFO.address, + }, + ) + await hass.async_block_till_done() + + assert result3["type"] == FlowResultType.CREATE_ENTRY + assert result3["title"] == LED_BLE_DISCOVERY_INFO.name + assert result3["data"] == { + CONF_ADDRESS: LED_BLE_DISCOVERY_INFO.address, + } + assert result3["result"].unique_id == LED_BLE_DISCOVERY_INFO.address + assert len(mock_setup_entry.mock_calls) == 1 + + +async def test_user_step_unknown_exception(hass: HomeAssistant) -> None: + """Test user step with an unknown exception.""" + with patch( + "homeassistant.components.led_ble.config_flow.async_discovered_service_info", + return_value=[NOT_LED_BLE_DISCOVERY_INFO, LED_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( + "homeassistant.components.led_ble.config_flow.LEDBLE.update", + side_effect=RuntimeError, + ): + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + CONF_ADDRESS: LED_BLE_DISCOVERY_INFO.address, + }, + ) + await hass.async_block_till_done() + + assert result2["type"] == FlowResultType.FORM + assert result2["step_id"] == "user" + assert result2["errors"] == {"base": "unknown"} + + with patch("homeassistant.components.led_ble.config_flow.LEDBLE.update",), patch( + "homeassistant.components.led_ble.async_setup_entry", + return_value=True, + ) as mock_setup_entry: + result3 = await hass.config_entries.flow.async_configure( + result2["flow_id"], + { + CONF_ADDRESS: LED_BLE_DISCOVERY_INFO.address, + }, + ) + await hass.async_block_till_done() + + assert result3["type"] == FlowResultType.CREATE_ENTRY + assert result3["title"] == LED_BLE_DISCOVERY_INFO.name + assert result3["data"] == { + CONF_ADDRESS: LED_BLE_DISCOVERY_INFO.address, + } + assert result3["result"].unique_id == LED_BLE_DISCOVERY_INFO.address + assert len(mock_setup_entry.mock_calls) == 1 + + +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=LED_BLE_DISCOVERY_INFO, + ) + assert result["type"] == FlowResultType.FORM + assert result["step_id"] == "user" + assert result["errors"] == {} + + with patch("homeassistant.components.led_ble.config_flow.LEDBLE.update",), patch( + "homeassistant.components.led_ble.async_setup_entry", + return_value=True, + ) as mock_setup_entry: + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + CONF_ADDRESS: LED_BLE_DISCOVERY_INFO.address, + }, + ) + await hass.async_block_till_done() + + assert result2["type"] == FlowResultType.CREATE_ENTRY + assert result2["title"] == LED_BLE_DISCOVERY_INFO.name + assert result2["data"] == { + CONF_ADDRESS: LED_BLE_DISCOVERY_INFO.address, + } + assert result2["result"].unique_id == LED_BLE_DISCOVERY_INFO.address + assert len(mock_setup_entry.mock_calls) == 1 + + +async def test_bluetooth_unsupported_model(hass: HomeAssistant) -> None: + """Test bluetooth step with an unsupported model path.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_BLUETOOTH}, + data=UNSUPPORTED_LED_BLE_DISCOVERY_INFO, + ) + assert result["type"] == FlowResultType.ABORT + assert result["reason"] == "not_supported"