From f7ce9b1688a9bc9cb575b411c0d62e62826c016f Mon Sep 17 00:00:00 2001 From: Joakim Plate Date: Wed, 12 Jul 2023 16:08:15 +0200 Subject: [PATCH] Add support for gardena bluetooth (#95179) Add support for gardena bluetooth based water computers. --- .coveragerc | 4 + CODEOWNERS | 2 + .../components/gardena_bluetooth/__init__.py | 86 ++++++ .../gardena_bluetooth/config_flow.py | 138 ++++++++++ .../components/gardena_bluetooth/const.py | 3 + .../gardena_bluetooth/coordinator.py | 121 ++++++++ .../gardena_bluetooth/manifest.json | 17 ++ .../components/gardena_bluetooth/strings.json | 28 ++ .../components/gardena_bluetooth/switch.py | 74 +++++ homeassistant/generated/bluetooth.py | 6 + homeassistant/generated/config_flows.py | 1 + homeassistant/generated/integrations.json | 6 + requirements_all.txt | 3 + requirements_test_all.txt | 3 + .../components/gardena_bluetooth/__init__.py | 61 +++++ .../components/gardena_bluetooth/conftest.py | 30 ++ .../snapshots/test_config_flow.ambr | 258 ++++++++++++++++++ .../gardena_bluetooth/test_config_flow.py | 134 +++++++++ 18 files changed, 975 insertions(+) create mode 100644 homeassistant/components/gardena_bluetooth/__init__.py create mode 100644 homeassistant/components/gardena_bluetooth/config_flow.py create mode 100644 homeassistant/components/gardena_bluetooth/const.py create mode 100644 homeassistant/components/gardena_bluetooth/coordinator.py create mode 100644 homeassistant/components/gardena_bluetooth/manifest.json create mode 100644 homeassistant/components/gardena_bluetooth/strings.json create mode 100644 homeassistant/components/gardena_bluetooth/switch.py create mode 100644 tests/components/gardena_bluetooth/__init__.py create mode 100644 tests/components/gardena_bluetooth/conftest.py create mode 100644 tests/components/gardena_bluetooth/snapshots/test_config_flow.ambr create mode 100644 tests/components/gardena_bluetooth/test_config_flow.py diff --git a/.coveragerc b/.coveragerc index 912c472de3e..e10a23e9c31 100644 --- a/.coveragerc +++ b/.coveragerc @@ -404,6 +404,10 @@ omit = homeassistant/components/garages_amsterdam/__init__.py homeassistant/components/garages_amsterdam/binary_sensor.py homeassistant/components/garages_amsterdam/sensor.py + homeassistant/components/gardena_bluetooth/__init__.py + homeassistant/components/gardena_bluetooth/const.py + homeassistant/components/gardena_bluetooth/coordinator.py + homeassistant/components/gardena_bluetooth/switch.py homeassistant/components/gc100/* homeassistant/components/geniushub/* homeassistant/components/geocaching/__init__.py diff --git a/CODEOWNERS b/CODEOWNERS index 16c0426d87f..01f486e2704 100644 --- a/CODEOWNERS +++ b/CODEOWNERS @@ -425,6 +425,8 @@ build.json @home-assistant/supervisor /tests/components/fully_kiosk/ @cgarwood /homeassistant/components/garages_amsterdam/ @klaasnicolaas /tests/components/garages_amsterdam/ @klaasnicolaas +/homeassistant/components/gardena_bluetooth/ @elupus +/tests/components/gardena_bluetooth/ @elupus /homeassistant/components/gdacs/ @exxamalte /tests/components/gdacs/ @exxamalte /homeassistant/components/generic/ @davet2001 diff --git a/homeassistant/components/gardena_bluetooth/__init__.py b/homeassistant/components/gardena_bluetooth/__init__.py new file mode 100644 index 00000000000..05ac16381d1 --- /dev/null +++ b/homeassistant/components/gardena_bluetooth/__init__.py @@ -0,0 +1,86 @@ +"""The Gardena Bluetooth integration.""" +from __future__ import annotations + +import asyncio +import logging + +from bleak.backends.device import BLEDevice +from gardena_bluetooth.client import CachedConnection, Client +from gardena_bluetooth.const import DeviceConfiguration, DeviceInformation +from gardena_bluetooth.exceptions import CommunicationFailure + +from homeassistant.components import bluetooth +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import CONF_ADDRESS, Platform +from homeassistant.core import HomeAssistant +from homeassistant.exceptions import ConfigEntryNotReady +from homeassistant.helpers.entity import DeviceInfo +import homeassistant.util.dt as dt_util + +from .const import DOMAIN +from .coordinator import Coordinator, DeviceUnavailable + +PLATFORMS: list[Platform] = [Platform.SWITCH] +LOGGER = logging.getLogger(__name__) +TIMEOUT = 20.0 +DISCONNECT_DELAY = 5 + + +def get_connection(hass: HomeAssistant, address: str) -> CachedConnection: + """Set up a cached client that keeps connection after last use.""" + + def _device_lookup() -> BLEDevice: + device = bluetooth.async_ble_device_from_address( + hass, address, connectable=True + ) + if not device: + raise DeviceUnavailable("Unable to find device") + return device + + return CachedConnection(DISCONNECT_DELAY, _device_lookup) + + +async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: + """Set up Gardena Bluetooth from a config entry.""" + + address = entry.data[CONF_ADDRESS] + client = Client(get_connection(hass, address)) + try: + sw_version = await client.read_char(DeviceInformation.firmware_version, None) + manufacturer = await client.read_char(DeviceInformation.manufacturer_name, None) + model = await client.read_char(DeviceInformation.model_number, None) + name = await client.read_char( + DeviceConfiguration.custom_device_name, entry.title + ) + uuids = await client.get_all_characteristics_uuid() + await client.update_timestamp(dt_util.now()) + except (asyncio.TimeoutError, CommunicationFailure, DeviceUnavailable) as exception: + await client.disconnect() + raise ConfigEntryNotReady( + f"Unable to connect to device {address} due to {exception}" + ) from exception + + device = DeviceInfo( + identifiers={(DOMAIN, address)}, + name=name, + sw_version=sw_version, + manufacturer=manufacturer, + model=model, + ) + + coordinator = Coordinator(hass, LOGGER, client, uuids, device, address) + + hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator + await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) + await coordinator.async_refresh() + + return True + + +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): + coordinator: Coordinator = hass.data[DOMAIN].pop(entry.entry_id) + await coordinator.async_shutdown() + + return unload_ok diff --git a/homeassistant/components/gardena_bluetooth/config_flow.py b/homeassistant/components/gardena_bluetooth/config_flow.py new file mode 100644 index 00000000000..3e981675057 --- /dev/null +++ b/homeassistant/components/gardena_bluetooth/config_flow.py @@ -0,0 +1,138 @@ +"""Config flow for Gardena Bluetooth integration.""" +from __future__ import annotations + +import logging +from typing import Any + +from gardena_bluetooth.client import Client +from gardena_bluetooth.const import DeviceInformation, ScanService +from gardena_bluetooth.exceptions import CharacteristicNotFound, CommunicationFailure +from gardena_bluetooth.parse import ManufacturerData, ProductGroup +import voluptuous as vol + +from homeassistant import config_entries +from homeassistant.components.bluetooth import ( + BluetoothServiceInfo, + async_discovered_service_info, +) +from homeassistant.const import CONF_ADDRESS +from homeassistant.data_entry_flow import AbortFlow, FlowResult + +from . import get_connection +from .const import DOMAIN + +_LOGGER = logging.getLogger(__name__) + + +def _is_supported(discovery_info: BluetoothServiceInfo): + """Check if device is supported.""" + if ScanService not in discovery_info.service_uuids: + return False + + if not (data := discovery_info.manufacturer_data.get(ManufacturerData.company)): + _LOGGER.debug("Missing manufacturer data: %s", discovery_info) + return False + + manufacturer_data = ManufacturerData.decode(data) + if manufacturer_data.group != ProductGroup.WATER_CONTROL: + _LOGGER.debug("Unsupported device: %s", manufacturer_data) + return False + + return True + + +def _get_name(discovery_info: BluetoothServiceInfo): + if discovery_info.name and discovery_info.name != discovery_info.address: + return discovery_info.name + return "Gardena Device" + + +class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): + """Handle a config flow for Gardena Bluetooth.""" + + VERSION = 1 + + def __init__(self) -> None: + """Initialize the config flow.""" + self.devices: dict[str, str] = {} + self.address: str | None + + async def async_read_data(self): + """Try to connect to device and extract information.""" + client = Client(get_connection(self.hass, self.address)) + try: + model = await client.read_char(DeviceInformation.model_number) + _LOGGER.debug("Found device with model: %s", model) + except (CharacteristicNotFound, CommunicationFailure) as exception: + raise AbortFlow( + "cannot_connect", description_placeholders={"error": str(exception)} + ) from exception + finally: + await client.disconnect() + + return {CONF_ADDRESS: self.address} + + async def async_step_bluetooth( + self, discovery_info: BluetoothServiceInfo + ) -> FlowResult: + """Handle the bluetooth discovery step.""" + _LOGGER.debug("Discovered device: %s", discovery_info) + if not _is_supported(discovery_info): + return self.async_abort(reason="no_devices_found") + + self.address = discovery_info.address + self.devices = {discovery_info.address: _get_name(discovery_info)} + await self.async_set_unique_id(self.address) + self._abort_if_unique_id_configured() + return await self.async_step_confirm() + + async def async_step_confirm( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: + """Confirm discovery.""" + assert self.address + title = self.devices[self.address] + + if user_input is not None: + data = await self.async_read_data() + return self.async_create_entry(title=title, data=data) + + self.context["title_placeholders"] = { + "name": title, + } + + self._set_confirm_only() + return self.async_show_form( + step_id="confirm", + description_placeholders=self.context["title_placeholders"], + ) + + async def async_step_user( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: + """Handle the initial step.""" + if user_input is not None: + self.address = user_input[CONF_ADDRESS] + await self.async_set_unique_id(self.address, raise_on_progress=False) + self._abort_if_unique_id_configured() + return await self.async_step_confirm() + + current_addresses = self._async_current_ids() + for discovery_info in async_discovered_service_info(self.hass): + address = discovery_info.address + if address in current_addresses or not _is_supported(discovery_info): + continue + + self.devices[address] = _get_name(discovery_info) + + if not self.devices: + return self.async_abort(reason="no_devices_found") + + return self.async_show_form( + step_id="user", + data_schema=vol.Schema( + { + vol.Required(CONF_ADDRESS): vol.In(self.devices), + }, + ), + ) diff --git a/homeassistant/components/gardena_bluetooth/const.py b/homeassistant/components/gardena_bluetooth/const.py new file mode 100644 index 00000000000..7de4c15b5fa --- /dev/null +++ b/homeassistant/components/gardena_bluetooth/const.py @@ -0,0 +1,3 @@ +"""Constants for the Gardena Bluetooth integration.""" + +DOMAIN = "gardena_bluetooth" diff --git a/homeassistant/components/gardena_bluetooth/coordinator.py b/homeassistant/components/gardena_bluetooth/coordinator.py new file mode 100644 index 00000000000..fa7639dece0 --- /dev/null +++ b/homeassistant/components/gardena_bluetooth/coordinator.py @@ -0,0 +1,121 @@ +"""Provides the DataUpdateCoordinator.""" +from __future__ import annotations + +from datetime import timedelta +import logging +from typing import Any + +from gardena_bluetooth.client import Client +from gardena_bluetooth.exceptions import ( + CharacteristicNoAccess, + GardenaBluetoothException, +) +from gardena_bluetooth.parse import Characteristic, CharacteristicType + +from homeassistant.components import bluetooth +from homeassistant.core import HomeAssistant +from homeassistant.exceptions import HomeAssistantError +from homeassistant.helpers.entity import DeviceInfo +from homeassistant.helpers.update_coordinator import ( + CoordinatorEntity, + DataUpdateCoordinator, + UpdateFailed, +) + +SCAN_INTERVAL = timedelta(seconds=60) +LOGGER = logging.getLogger(__name__) + + +class DeviceUnavailable(HomeAssistantError): + """Raised if device can't be found.""" + + +class Coordinator(DataUpdateCoordinator[dict[str, bytes]]): + """Class to manage fetching data.""" + + def __init__( + self, + hass: HomeAssistant, + logger: logging.Logger, + client: Client, + characteristics: set[str], + device_info: DeviceInfo, + address: str, + ) -> None: + """Initialize global data updater.""" + super().__init__( + hass=hass, + logger=logger, + name="Gardena Bluetooth Data Update Coordinator", + update_interval=SCAN_INTERVAL, + ) + self.address = address + self.data = {} + self.client = client + self.characteristics = characteristics + self.device_info = device_info + + async def async_shutdown(self) -> None: + """Shutdown coordinator and any connection.""" + await super().async_shutdown() + await self.client.disconnect() + + async def _async_update_data(self) -> dict[str, bytes]: + """Poll the device.""" + uuids: set[str] = { + uuid for context in self.async_contexts() for uuid in context + } + if not uuids: + return {} + + data: dict[str, bytes] = {} + for uuid in uuids: + try: + data[uuid] = await self.client.read_char_raw(uuid) + except CharacteristicNoAccess as exception: + LOGGER.debug("Unable to get data for %s due to %s", uuid, exception) + except (GardenaBluetoothException, DeviceUnavailable) as exception: + raise UpdateFailed( + f"Unable to update data for {uuid} due to {exception}" + ) from exception + return data + + def read_cached( + self, char: Characteristic[CharacteristicType] + ) -> CharacteristicType | None: + """Read cached characteristic.""" + if data := self.data.get(char.uuid): + return char.decode(data) + return None + + async def write( + self, char: Characteristic[CharacteristicType], value: CharacteristicType + ) -> None: + """Write characteristic to device.""" + try: + await self.client.write_char(char, value) + except (GardenaBluetoothException, DeviceUnavailable) as exception: + raise HomeAssistantError( + f"Unable to write characteristic {char} dur to {exception}" + ) from exception + + self.data[char.uuid] = char.encode(value) + await self.async_refresh() + + +class GardenaBluetoothEntity(CoordinatorEntity[Coordinator]): + """Coordinator entity for Gardena Bluetooth.""" + + _attr_has_entity_name = True + + def __init__(self, coordinator: Coordinator, context: Any = None) -> None: + """Initialize coordinator entity.""" + super().__init__(coordinator, context) + self._attr_device_info = coordinator.device_info + + @property + def available(self) -> bool: + """Return if entity is available.""" + return super().available and bluetooth.async_address_present( + self.hass, self.coordinator.address, True + ) diff --git a/homeassistant/components/gardena_bluetooth/manifest.json b/homeassistant/components/gardena_bluetooth/manifest.json new file mode 100644 index 00000000000..cdc43a802c9 --- /dev/null +++ b/homeassistant/components/gardena_bluetooth/manifest.json @@ -0,0 +1,17 @@ +{ + "domain": "gardena_bluetooth", + "name": "Gardena Bluetooth", + "bluetooth": [ + { + "manufacturer_id": 1062, + "service_uuid": "98bd0001-0b0e-421a-84e5-ddbf75dc6de4", + "connectable": true + } + ], + "codeowners": ["@elupus"], + "config_flow": true, + "dependencies": ["bluetooth_adapters"], + "documentation": "https://www.home-assistant.io/integrations/gardena_bluetooth", + "iot_class": "local_polling", + "requirements": ["gardena_bluetooth==1.0.1"] +} diff --git a/homeassistant/components/gardena_bluetooth/strings.json b/homeassistant/components/gardena_bluetooth/strings.json new file mode 100644 index 00000000000..165e336bbec --- /dev/null +++ b/homeassistant/components/gardena_bluetooth/strings.json @@ -0,0 +1,28 @@ +{ + "config": { + "step": { + "user": { + "description": "[%key:component::bluetooth::config::step::user::description%]", + "data": { + "address": "[%key:component::bluetooth::config::step::user::data::address%]" + } + }, + "confirm": { + "description": "[%key:component::bluetooth::config::step::bluetooth_confirm::description%]" + } + }, + "error": { + "cannot_connect": "Failed to connect: {error}" + }, + "abort": { + "already_configured": "[%key:common::config_flow::abort::already_configured_device%]" + } + }, + "entity": { + "switch": { + "state": { + "name": "Open" + } + } + } +} diff --git a/homeassistant/components/gardena_bluetooth/switch.py b/homeassistant/components/gardena_bluetooth/switch.py new file mode 100644 index 00000000000..e3fcc8978c7 --- /dev/null +++ b/homeassistant/components/gardena_bluetooth/switch.py @@ -0,0 +1,74 @@ +"""Support for switch entities.""" +from __future__ import annotations + +from typing import Any + +from gardena_bluetooth.const import Valve + +from homeassistant.components.switch import SwitchEntity +from homeassistant.config_entries import ConfigEntry +from homeassistant.core import HomeAssistant +from homeassistant.exceptions import HomeAssistantError +from homeassistant.helpers.entity_platform import AddEntitiesCallback + +from .const import DOMAIN +from .coordinator import Coordinator, GardenaBluetoothEntity + + +async def async_setup_entry( + hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback +) -> None: + """Set up switch based on a config entry.""" + coordinator: Coordinator = hass.data[DOMAIN][entry.entry_id] + entities = [] + if GardenaBluetoothValveSwitch.characteristics.issubset( + coordinator.characteristics + ): + entities.append(GardenaBluetoothValveSwitch(coordinator)) + + async_add_entities(entities) + + +class GardenaBluetoothValveSwitch(GardenaBluetoothEntity, SwitchEntity): + """Representation of a valve switch.""" + + characteristics = { + Valve.state.uuid, + Valve.manual_watering_time.uuid, + Valve.manual_watering_time.uuid, + } + + def __init__( + self, + coordinator: Coordinator, + ) -> None: + """Initialize the switch.""" + super().__init__( + coordinator, {Valve.state.uuid, Valve.manual_watering_time.uuid} + ) + self._attr_unique_id = f"{coordinator.address}-{Valve.state.uuid}" + self._attr_translation_key = "state" + self._attr_is_on = None + + def _handle_coordinator_update(self) -> None: + if data := self.coordinator.data.get(Valve.state.uuid): + self._attr_is_on = Valve.state.decode(data) + else: + self._attr_is_on = None + super()._handle_coordinator_update() + + async def async_turn_on(self, **kwargs: Any) -> None: + """Turn the entity on.""" + if not (data := self.coordinator.data.get(Valve.manual_watering_time.uuid)): + raise HomeAssistantError("Unable to get manual activation time.") + + value = Valve.manual_watering_time.decode(data) + await self.coordinator.write(Valve.remaining_open_time, value) + self._attr_is_on = True + self.async_write_ha_state() + + async def async_turn_off(self, **kwargs: Any) -> None: + """Turn the entity off.""" + await self.coordinator.write(Valve.remaining_open_time, 0) + self._attr_is_on = False + self.async_write_ha_state() diff --git a/homeassistant/generated/bluetooth.py b/homeassistant/generated/bluetooth.py index 24215a8a0c4..64fae252975 100644 --- a/homeassistant/generated/bluetooth.py +++ b/homeassistant/generated/bluetooth.py @@ -83,6 +83,12 @@ BLUETOOTH: list[dict[str, bool | str | int | list[int]]] = [ ], "manufacturer_id": 20296, }, + { + "connectable": True, + "domain": "gardena_bluetooth", + "manufacturer_id": 1062, + "service_uuid": "98bd0001-0b0e-421a-84e5-ddbf75dc6de4", + }, { "connectable": False, "domain": "govee_ble", diff --git a/homeassistant/generated/config_flows.py b/homeassistant/generated/config_flows.py index d8ffa25b765..6d9a132a29a 100644 --- a/homeassistant/generated/config_flows.py +++ b/homeassistant/generated/config_flows.py @@ -155,6 +155,7 @@ FLOWS = { "frontier_silicon", "fully_kiosk", "garages_amsterdam", + "gardena_bluetooth", "gdacs", "generic", "geo_json_events", diff --git a/homeassistant/generated/integrations.json b/homeassistant/generated/integrations.json index d1f68271c55..21f7acd59e3 100644 --- a/homeassistant/generated/integrations.json +++ b/homeassistant/generated/integrations.json @@ -1884,6 +1884,12 @@ "config_flow": true, "iot_class": "cloud_polling" }, + "gardena_bluetooth": { + "name": "Gardena Bluetooth", + "integration_type": "hub", + "config_flow": true, + "iot_class": "local_polling" + }, "gaviota": { "name": "Gaviota", "integration_type": "virtual", diff --git a/requirements_all.txt b/requirements_all.txt index 0cb289d748f..d33f01b34be 100644 --- a/requirements_all.txt +++ b/requirements_all.txt @@ -819,6 +819,9 @@ fritzconnection[qr]==1.12.2 # homeassistant.components.google_translate gTTS==2.2.4 +# homeassistant.components.gardena_bluetooth +gardena_bluetooth==1.0.1 + # homeassistant.components.google_assistant_sdk gassist-text==0.0.10 diff --git a/requirements_test_all.txt b/requirements_test_all.txt index 04065df19c1..f35021c3e00 100644 --- a/requirements_test_all.txt +++ b/requirements_test_all.txt @@ -641,6 +641,9 @@ fritzconnection[qr]==1.12.2 # homeassistant.components.google_translate gTTS==2.2.4 +# homeassistant.components.gardena_bluetooth +gardena_bluetooth==1.0.1 + # homeassistant.components.google_assistant_sdk gassist-text==0.0.10 diff --git a/tests/components/gardena_bluetooth/__init__.py b/tests/components/gardena_bluetooth/__init__.py new file mode 100644 index 00000000000..6a064409e9e --- /dev/null +++ b/tests/components/gardena_bluetooth/__init__.py @@ -0,0 +1,61 @@ +"""Tests for the Gardena Bluetooth integration.""" + +from homeassistant.helpers.service_info.bluetooth import BluetoothServiceInfo + +WATER_TIMER_SERVICE_INFO = BluetoothServiceInfo( + name="Timer", + address="00000000-0000-0000-0000-000000000001", + rssi=-63, + service_data={}, + manufacturer_data={ + 1062: b"\x02\x07d\x02\x05\x01\x02\x08\x00\x02\t\x01\x04\x06\x12\x00\x01" + }, + service_uuids=["98bd0001-0b0e-421a-84e5-ddbf75dc6de4"], + source="local", +) + +WATER_TIMER_UNNAMED_SERVICE_INFO = BluetoothServiceInfo( + name=None, + address="00000000-0000-0000-0000-000000000002", + rssi=-63, + service_data={}, + manufacturer_data={ + 1062: b"\x02\x07d\x02\x05\x01\x02\x08\x00\x02\t\x01\x04\x06\x12\x00\x01" + }, + service_uuids=["98bd0001-0b0e-421a-84e5-ddbf75dc6de4"], + source="local", +) + +MISSING_SERVICE_SERVICE_INFO = BluetoothServiceInfo( + name="Missing Service Info", + address="00000000-0000-0000-0001-000000000000", + rssi=-63, + service_data={}, + manufacturer_data={ + 1062: b"\x02\x07d\x02\x05\x01\x02\x08\x00\x02\t\x01\x04\x06\x12\x00\x01" + }, + service_uuids=[], + source="local", +) + +MISSING_MANUFACTURER_DATA_SERVICE_INFO = BluetoothServiceInfo( + name="Missing Manufacturer Data", + address="00000000-0000-0000-0001-000000000001", + rssi=-63, + service_data={}, + manufacturer_data={}, + service_uuids=["98bd0001-0b0e-421a-84e5-ddbf75dc6de4"], + source="local", +) + +UNSUPPORTED_GROUP_SERVICE_INFO = BluetoothServiceInfo( + name="Unsupported Group", + address="00000000-0000-0000-0001-000000000002", + rssi=-63, + service_data={}, + manufacturer_data={ + 1062: b"\x02\x07d\x02\x05\x01\x02\x08\x00\x02\t\x01\x04\x06\x10\x00\x01" + }, + service_uuids=["98bd0001-0b0e-421a-84e5-ddbf75dc6de4"], + source="local", +) diff --git a/tests/components/gardena_bluetooth/conftest.py b/tests/components/gardena_bluetooth/conftest.py new file mode 100644 index 00000000000..f09a274742f --- /dev/null +++ b/tests/components/gardena_bluetooth/conftest.py @@ -0,0 +1,30 @@ +"""Common fixtures for the Gardena Bluetooth tests.""" +from collections.abc import Generator +from unittest.mock import AsyncMock, Mock, patch + +from gardena_bluetooth.client import Client +import pytest + + +@pytest.fixture +def mock_setup_entry() -> Generator[AsyncMock, None, None]: + """Override async_setup_entry.""" + with patch( + "homeassistant.components.gardena_bluetooth.async_setup_entry", + return_value=True, + ) as mock_setup_entry: + yield mock_setup_entry + + +@pytest.fixture(autouse=True) +def mock_client(enable_bluetooth): + """Auto mock bluetooth.""" + + client = Mock(spec_set=Client) + client.get_all_characteristics_uuid.return_value = set() + + with patch( + "homeassistant.components.gardena_bluetooth.config_flow.Client", + return_value=client, + ): + yield client diff --git a/tests/components/gardena_bluetooth/snapshots/test_config_flow.ambr b/tests/components/gardena_bluetooth/snapshots/test_config_flow.ambr new file mode 100644 index 00000000000..fde70b60a01 --- /dev/null +++ b/tests/components/gardena_bluetooth/snapshots/test_config_flow.ambr @@ -0,0 +1,258 @@ +# serializer version: 1 +# name: test_bluetooth + FlowResultSnapshot({ + 'data_schema': None, + 'description_placeholders': dict({ + 'name': 'Timer', + }), + 'errors': None, + 'flow_id': , + 'handler': 'gardena_bluetooth', + 'last_step': None, + 'step_id': 'confirm', + 'type': , + }) +# --- +# name: test_bluetooth.1 + FlowResultSnapshot({ + 'context': dict({ + 'confirm_only': True, + 'source': 'bluetooth', + 'title_placeholders': dict({ + 'name': 'Timer', + }), + 'unique_id': '00000000-0000-0000-0000-000000000001', + }), + 'data': dict({ + 'address': '00000000-0000-0000-0000-000000000001', + }), + 'description': None, + 'description_placeholders': None, + 'flow_id': , + 'handler': 'gardena_bluetooth', + 'options': dict({ + }), + 'result': ConfigEntrySnapshot({ + 'data': dict({ + 'address': '00000000-0000-0000-0000-000000000001', + }), + 'disabled_by': None, + 'domain': 'gardena_bluetooth', + 'entry_id': , + 'options': dict({ + }), + 'pref_disable_new_entities': False, + 'pref_disable_polling': False, + 'source': 'bluetooth', + 'title': 'Timer', + 'unique_id': '00000000-0000-0000-0000-000000000001', + 'version': 1, + }), + 'title': 'Timer', + 'type': , + 'version': 1, + }) +# --- +# name: test_bluetooth_invalid + FlowResultSnapshot({ + 'description_placeholders': None, + 'flow_id': , + 'handler': 'gardena_bluetooth', + 'reason': 'no_devices_found', + 'type': , + }) +# --- +# name: test_bluetooth_lost + FlowResultSnapshot({ + 'data_schema': None, + 'description_placeholders': dict({ + 'name': 'Timer', + }), + 'errors': None, + 'flow_id': , + 'handler': 'gardena_bluetooth', + 'last_step': None, + 'step_id': 'confirm', + 'type': , + }) +# --- +# name: test_bluetooth_lost.1 + FlowResultSnapshot({ + 'context': dict({ + 'confirm_only': True, + 'source': 'bluetooth', + 'title_placeholders': dict({ + 'name': 'Timer', + }), + 'unique_id': '00000000-0000-0000-0000-000000000001', + }), + 'data': dict({ + 'address': '00000000-0000-0000-0000-000000000001', + }), + 'description': None, + 'description_placeholders': None, + 'flow_id': , + 'handler': 'gardena_bluetooth', + 'options': dict({ + }), + 'result': ConfigEntrySnapshot({ + 'data': dict({ + 'address': '00000000-0000-0000-0000-000000000001', + }), + 'disabled_by': None, + 'domain': 'gardena_bluetooth', + 'entry_id': , + 'options': dict({ + }), + 'pref_disable_new_entities': False, + 'pref_disable_polling': False, + 'source': 'bluetooth', + 'title': 'Timer', + 'unique_id': '00000000-0000-0000-0000-000000000001', + 'version': 1, + }), + 'title': 'Timer', + 'type': , + 'version': 1, + }) +# --- +# name: test_failed_connect + FlowResultSnapshot({ + 'data_schema': list([ + dict({ + 'name': 'address', + 'options': list([ + tuple( + '00000000-0000-0000-0000-000000000001', + 'Timer', + ), + ]), + 'required': True, + 'type': 'select', + }), + ]), + 'description_placeholders': None, + 'errors': None, + 'flow_id': , + 'handler': 'gardena_bluetooth', + 'last_step': None, + 'step_id': 'user', + 'type': , + }) +# --- +# name: test_failed_connect.1 + FlowResultSnapshot({ + 'data_schema': None, + 'description_placeholders': dict({ + 'name': 'Timer', + }), + 'errors': None, + 'flow_id': , + 'handler': 'gardena_bluetooth', + 'last_step': None, + 'step_id': 'confirm', + 'type': , + }) +# --- +# name: test_failed_connect.2 + FlowResultSnapshot({ + 'description_placeholders': dict({ + 'error': 'something went wrong', + }), + 'flow_id': , + 'handler': 'gardena_bluetooth', + 'reason': 'cannot_connect', + 'type': , + }) +# --- +# name: test_no_devices + FlowResultSnapshot({ + 'description_placeholders': None, + 'flow_id': , + 'handler': 'gardena_bluetooth', + 'reason': 'no_devices_found', + 'type': , + }) +# --- +# name: test_user_selection + FlowResultSnapshot({ + 'data_schema': list([ + dict({ + 'name': 'address', + 'options': list([ + tuple( + '00000000-0000-0000-0000-000000000001', + 'Timer', + ), + tuple( + '00000000-0000-0000-0000-000000000002', + 'Gardena Device', + ), + ]), + 'required': True, + 'type': 'select', + }), + ]), + 'description_placeholders': None, + 'errors': None, + 'flow_id': , + 'handler': 'gardena_bluetooth', + 'last_step': None, + 'step_id': 'user', + 'type': , + }) +# --- +# name: test_user_selection.1 + FlowResultSnapshot({ + 'data_schema': None, + 'description_placeholders': dict({ + 'name': 'Timer', + }), + 'errors': None, + 'flow_id': , + 'handler': 'gardena_bluetooth', + 'last_step': None, + 'step_id': 'confirm', + 'type': , + }) +# --- +# name: test_user_selection.2 + FlowResultSnapshot({ + 'context': dict({ + 'confirm_only': True, + 'source': 'user', + 'title_placeholders': dict({ + 'name': 'Timer', + }), + 'unique_id': '00000000-0000-0000-0000-000000000001', + }), + 'data': dict({ + 'address': '00000000-0000-0000-0000-000000000001', + }), + 'description': None, + 'description_placeholders': None, + 'flow_id': , + 'handler': 'gardena_bluetooth', + 'options': dict({ + }), + 'result': ConfigEntrySnapshot({ + 'data': dict({ + 'address': '00000000-0000-0000-0000-000000000001', + }), + 'disabled_by': None, + 'domain': 'gardena_bluetooth', + 'entry_id': , + 'options': dict({ + }), + 'pref_disable_new_entities': False, + 'pref_disable_polling': False, + 'source': 'user', + 'title': 'Timer', + 'unique_id': '00000000-0000-0000-0000-000000000001', + 'version': 1, + }), + 'title': 'Timer', + 'type': , + 'version': 1, + }) +# --- diff --git a/tests/components/gardena_bluetooth/test_config_flow.py b/tests/components/gardena_bluetooth/test_config_flow.py new file mode 100644 index 00000000000..0f0e297c4d7 --- /dev/null +++ b/tests/components/gardena_bluetooth/test_config_flow.py @@ -0,0 +1,134 @@ +"""Test the Gardena Bluetooth config flow.""" +from unittest.mock import Mock + +from gardena_bluetooth.exceptions import CharacteristicNotFound +import pytest +from syrupy.assertion import SnapshotAssertion + +from homeassistant import config_entries +from homeassistant.components.gardena_bluetooth.const import DOMAIN +from homeassistant.core import HomeAssistant + +from . import ( + MISSING_MANUFACTURER_DATA_SERVICE_INFO, + MISSING_SERVICE_SERVICE_INFO, + UNSUPPORTED_GROUP_SERVICE_INFO, + WATER_TIMER_SERVICE_INFO, + WATER_TIMER_UNNAMED_SERVICE_INFO, +) + +from tests.components.bluetooth import ( + inject_bluetooth_service_info, +) + +pytestmark = pytest.mark.usefixtures("mock_setup_entry") + + +async def test_user_selection( + hass: HomeAssistant, + snapshot: SnapshotAssertion, +) -> None: + """Test we can select a device.""" + + inject_bluetooth_service_info(hass, WATER_TIMER_SERVICE_INFO) + inject_bluetooth_service_info(hass, WATER_TIMER_UNNAMED_SERVICE_INFO) + + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + assert result == snapshot + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={"address": "00000000-0000-0000-0000-000000000001"}, + ) + assert result == snapshot + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={}, + ) + assert result == snapshot + + +async def test_failed_connect( + hass: HomeAssistant, + mock_client: Mock, + snapshot: SnapshotAssertion, +) -> None: + """Test we can select a device.""" + + inject_bluetooth_service_info(hass, WATER_TIMER_SERVICE_INFO) + + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + assert result == snapshot + + mock_client.read_char.side_effect = CharacteristicNotFound("something went wrong") + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={"address": "00000000-0000-0000-0000-000000000001"}, + ) + assert result == snapshot + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={}, + ) + assert result == snapshot + + +async def test_no_devices( + hass: HomeAssistant, + snapshot: SnapshotAssertion, +) -> None: + """Test missing device.""" + + inject_bluetooth_service_info(hass, MISSING_MANUFACTURER_DATA_SERVICE_INFO) + inject_bluetooth_service_info(hass, MISSING_SERVICE_SERVICE_INFO) + inject_bluetooth_service_info(hass, UNSUPPORTED_GROUP_SERVICE_INFO) + + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + assert result == snapshot + + +async def test_bluetooth( + hass: HomeAssistant, + snapshot: SnapshotAssertion, +) -> None: + """Test bluetooth device discovery.""" + + inject_bluetooth_service_info(hass, WATER_TIMER_SERVICE_INFO) + + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_BLUETOOTH}, + data=WATER_TIMER_SERVICE_INFO, + ) + assert result == snapshot + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={}, + ) + assert result == snapshot + + +async def test_bluetooth_invalid( + hass: HomeAssistant, + snapshot: SnapshotAssertion, +) -> None: + """Test bluetooth device discovery with invalid data.""" + + inject_bluetooth_service_info(hass, UNSUPPORTED_GROUP_SERVICE_INFO) + + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_BLUETOOTH}, + data=UNSUPPORTED_GROUP_SERVICE_INFO, + ) + assert result == snapshot