Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Update switchbot to be local push (#75645) 

* Update switchbot to be local push

* fixes

* fixes

* fixes

* fixes

* adjust

* cover is not assumed anymore

* cleanups

* adjust

* adjust

* add missing cover

* import compat

* fixes

* uses lower

* uses lower

* bleak users upper case addresses

* fixes

* bump

* keep conf_mac and deprecated options for rollback

* reuse coordinator

* adjust

* move around

* move around

* move around

* move around

* refactor fixes

* compat with DataUpdateCoordinator

* fix available

* Update homeassistant/components/bluetooth/passive_update_processor.py

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

* Update homeassistant/components/bluetooth/passive_update_coordinator.py

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

* Update homeassistant/components/bluetooth/update_coordinator.py

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

* Split bluetooth coordinator into PassiveBluetoothDataUpdateCoordinator and PassiveBluetoothProcessorCoordinator

The PassiveBluetoothDataUpdateCoordinator is now used to replace instances
of DataUpdateCoordinator where the data is coming from bluetooth
advertisements, and the integration may also mix in active updates

The PassiveBluetoothProcessorCoordinator is used for integrations that
want to process each bluetooth advertisement with multiple processors
which can be dispatched to individual platforms or areas or the integration
as it chooes

* change connections

* reduce code churn to reduce review overhead

* reduce code churn to reduce review overhead

* Update homeassistant/components/bluetooth/passive_update_coordinator.py

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

* add basic test

* add basic test

* complete coverage

* Update homeassistant/components/switchbot/coordinator.py

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

* Update homeassistant/components/switchbot/coordinator.py

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

* Update homeassistant/components/switchbot/__init__.py

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

* Update homeassistant/components/switchbot/__init__.py

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

* lint

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
pull/75691/head
J. Nick Koston 2022-07-24 11:38:45 -05:00 committed by GitHub
parent 79be87f9ce
commit 198167a2c8
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
17 changed files with 542 additions and 462 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

120
homeassistant/components/switchbot/__init__.py
View File

@ -1,10 +1,24 @@
"""Support for Switchbot devices."""
from collections.abc import Mapping
import logging
from types import MappingProxyType
from typing import Any
import switchbot
from homeassistant.components import bluetooth
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_SENSOR_TYPE, Platform
from homeassistant.const import (
CONF_ADDRESS,
CONF_MAC,
CONF_PASSWORD,
CONF_SENSOR_TYPE,
Platform,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import device_registry as dr
from .const import (
ATTR_BOT,
@ -13,13 +27,8 @@ from .const import (
COMMON_OPTIONS,
CONF_RETRY_COUNT,
CONF_RETRY_TIMEOUT,
CONF_SCAN_TIMEOUT,
CONF_TIME_BETWEEN_UPDATE_COMMAND,
DATA_COORDINATOR,
DEFAULT_RETRY_COUNT,
DEFAULT_RETRY_TIMEOUT,
DEFAULT_SCAN_TIMEOUT,
DEFAULT_TIME_BETWEEN_UPDATE_COMMAND,
DOMAIN,
)
from .coordinator import SwitchbotDataUpdateCoordinator
@ -29,57 +38,67 @@ PLATFORMS_BY_TYPE = {
ATTR_CURTAIN: [Platform.COVER, Platform.BINARY_SENSOR, Platform.SENSOR],
ATTR_HYGROMETER: [Platform.SENSOR],
}
CLASS_BY_DEVICE = {
ATTR_CURTAIN: switchbot.SwitchbotCurtain,
ATTR_BOT: switchbot.Switchbot,
}
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Switchbot from a config entry."""
hass.data.setdefault(DOMAIN, {})
domain_data = hass.data[DOMAIN]
if not entry.options:
options = {
CONF_TIME_BETWEEN_UPDATE_COMMAND: DEFAULT_TIME_BETWEEN_UPDATE_COMMAND,
CONF_RETRY_COUNT: DEFAULT_RETRY_COUNT,
CONF_RETRY_TIMEOUT: DEFAULT_RETRY_TIMEOUT,
CONF_SCAN_TIMEOUT: DEFAULT_SCAN_TIMEOUT,
}
hass.config_entries.async_update_entry(entry, options=options)
# Use same coordinator instance for all entities.
# Uses BTLE advertisement data, all Switchbot devices in range is stored here.
if DATA_COORDINATOR not in hass.data[DOMAIN]:
if COMMON_OPTIONS not in hass.data[DOMAIN]:
hass.data[DOMAIN][COMMON_OPTIONS] = {**entry.options}
switchbot.DEFAULT_RETRY_TIMEOUT = hass.data[DOMAIN][COMMON_OPTIONS][
CONF_RETRY_TIMEOUT
]
# Store api in coordinator.
coordinator = SwitchbotDataUpdateCoordinator(
hass,
update_interval=hass.data[DOMAIN][COMMON_OPTIONS][
CONF_TIME_BETWEEN_UPDATE_COMMAND
],
api=switchbot,
retry_count=hass.data[DOMAIN][COMMON_OPTIONS][CONF_RETRY_COUNT],
scan_timeout=hass.data[DOMAIN][COMMON_OPTIONS][CONF_SCAN_TIMEOUT],
if CONF_ADDRESS not in entry.data and CONF_MAC in entry.data:
# Bleak uses addresses not mac addresses which are are actually
# UUIDs on some platforms (MacOS).
mac = entry.data[CONF_MAC]
if "-" not in mac:
mac = dr.format_mac(mac)
hass.config_entries.async_update_entry(
entry,
data={**entry.data, CONF_ADDRESS: mac},
)
hass.data[DOMAIN][DATA_COORDINATOR] = coordinator
if not entry.options:
hass.config_entries.async_update_entry(
entry,
options={
CONF_RETRY_COUNT: DEFAULT_RETRY_COUNT,
CONF_RETRY_TIMEOUT: DEFAULT_RETRY_TIMEOUT,
},
)
else:
coordinator = hass.data[DOMAIN][DATA_COORDINATOR]
sensor_type: str = entry.data[CONF_SENSOR_TYPE]
address: str = entry.data[CONF_ADDRESS]
ble_device = bluetooth.async_ble_device_from_address(hass, address.upper())
if not ble_device:
raise ConfigEntryNotReady(
f"Could not find Switchbot {sensor_type} with address {address}"
)
await coordinator.async_config_entry_first_refresh()
if COMMON_OPTIONS not in domain_data:
domain_data[COMMON_OPTIONS] = entry.options
common_options: Mapping[str, int] = domain_data[COMMON_OPTIONS]
switchbot.DEFAULT_RETRY_TIMEOUT = common_options[CONF_RETRY_TIMEOUT]
cls = CLASS_BY_DEVICE.get(sensor_type, switchbot.SwitchbotDevice)
device = cls(
device=ble_device,
password=entry.data.get(CONF_PASSWORD),
retry_count=entry.options[CONF_RETRY_COUNT],
)
coordinator = hass.data[DOMAIN][entry.entry_id] = SwitchbotDataUpdateCoordinator(
hass, _LOGGER, ble_device, device, common_options
)
entry.async_on_unload(coordinator.async_start())
if not await coordinator.async_wait_ready():
raise ConfigEntryNotReady(f"Switchbot {sensor_type} with {address} not ready")
entry.async_on_unload(entry.add_update_listener(_async_update_listener))
hass.data[DOMAIN][entry.entry_id] = {DATA_COORDINATOR: coordinator}
sensor_type = entry.data[CONF_SENSOR_TYPE]
await hass.config_entries.async_forward_entry_setups(
entry, PLATFORMS_BY_TYPE[sensor_type]
)
@ -96,8 +115,7 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
if unload_ok:
hass.data[DOMAIN].pop(entry.entry_id)
if len(hass.config_entries.async_entries(DOMAIN)) == 0:
if not hass.config_entries.async_entries(DOMAIN):
hass.data.pop(DOMAIN)
return unload_ok
@ -106,8 +124,6 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def _async_update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Handle options update."""
# Update entity options stored in hass.
if {**entry.options} != hass.data[DOMAIN][COMMON_OPTIONS]:
hass.data[DOMAIN][COMMON_OPTIONS] = {**entry.options}
hass.data[DOMAIN].pop(DATA_COORDINATOR)
await hass.config_entries.async_reload(entry.entry_id)
common_options: MappingProxyType[str, Any] = hass.data[DOMAIN][COMMON_OPTIONS]
if entry.options != common_options:
await hass.config_entries.async_reload(entry.entry_id)

27
homeassistant/components/switchbot/binary_sensor.py
View File

@ -6,13 +6,12 @@ from homeassistant.components.binary_sensor import (
BinarySensorEntityDescription,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_MAC, CONF_NAME
from homeassistant.const import CONF_ADDRESS, CONF_NAME
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import PlatformNotReady
from homeassistant.helpers.entity import EntityCategory
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DATA_COORDINATOR, DOMAIN
from .const import DOMAIN
from .coordinator import SwitchbotDataUpdateCoordinator
from .entity import SwitchbotEntity
@ -30,23 +29,19 @@ async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up Switchbot curtain based on a config entry."""
coordinator: SwitchbotDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id][
DATA_COORDINATOR
]
if not coordinator.data.get(entry.unique_id):
raise PlatformNotReady
coordinator: SwitchbotDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
unique_id = entry.unique_id
assert unique_id is not None
async_add_entities(
[
SwitchBotBinarySensor(
coordinator,
entry.unique_id,
unique_id,
binary_sensor,
entry.data[CONF_MAC],
entry.data[CONF_ADDRESS],
entry.data[CONF_NAME],
)
for binary_sensor in coordinator.data[entry.unique_id]["data"]
for binary_sensor in coordinator.data["data"]
if binary_sensor in BINARY_SENSOR_TYPES
]
)
@ -58,15 +53,15 @@ class SwitchBotBinarySensor(SwitchbotEntity, BinarySensorEntity):
def __init__(
self,
coordinator: SwitchbotDataUpdateCoordinator,
idx: str | None,
unique_id: str,
binary_sensor: str,
mac: str,
switchbot_name: str,
) -> None:
"""Initialize the Switchbot sensor."""
super().__init__(coordinator, idx, mac, name=switchbot_name)
super().__init__(coordinator, unique_id, mac, name=switchbot_name)
self._sensor = binary_sensor
self._attr_unique_id = f"{idx}-{binary_sensor}"
self._attr_unique_id = f"{unique_id}-{binary_sensor}"
self._attr_name = f"{switchbot_name} {binary_sensor.title()}"
self.entity_description = BINARY_SENSOR_TYPES[binary_sensor]

143
homeassistant/components/switchbot/config_flow.py
View File

@ -2,25 +2,26 @@
from __future__ import annotations
import logging
from typing import Any
from typing import Any, cast
from switchbot import GetSwitchbotDevices
from switchbot import SwitchBotAdvertisement, parse_advertisement_data
import voluptuous as vol
from homeassistant.components.bluetooth import (
BluetoothServiceInfoBleak,
async_discovered_service_info,
)
from homeassistant.config_entries import ConfigEntry, ConfigFlow, OptionsFlow
from homeassistant.const import CONF_MAC, CONF_NAME, CONF_PASSWORD, CONF_SENSOR_TYPE
from homeassistant.const import CONF_ADDRESS, CONF_NAME, CONF_PASSWORD, CONF_SENSOR_TYPE
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers.service_info.bluetooth import BluetoothServiceInfo
from .const import (
CONF_RETRY_COUNT,
CONF_RETRY_TIMEOUT,
CONF_SCAN_TIMEOUT,
CONF_TIME_BETWEEN_UPDATE_COMMAND,
DEFAULT_RETRY_COUNT,
DEFAULT_RETRY_TIMEOUT,
DEFAULT_SCAN_TIMEOUT,
DEFAULT_TIME_BETWEEN_UPDATE_COMMAND,
DOMAIN,
SUPPORTED_MODEL_TYPES,
)
@ -28,15 +29,9 @@ from .const import (
_LOGGER = logging.getLogger(__name__)
async def _btle_connect() -> dict:
"""Scan for BTLE advertisement data."""
switchbot_devices = await GetSwitchbotDevices().discover()
if not switchbot_devices:
raise NotConnectedError("Failed to discover switchbot")
return switchbot_devices
def format_unique_id(address: str) -> str:
"""Format the unique ID for a switchbot."""
return address.replace(":", "").lower()
class SwitchbotConfigFlow(ConfigFlow, domain=DOMAIN):
@ -44,18 +39,6 @@ class SwitchbotConfigFlow(ConfigFlow, domain=DOMAIN):
VERSION = 1
async def _get_switchbots(self) -> dict:
"""Try to discover nearby Switchbot devices."""
# asyncio.lock prevents btle adapter exceptions if there are multiple calls to this method.
# store asyncio.lock in hass data if not present.
if DOMAIN not in self.hass.data:
self.hass.data.setdefault(DOMAIN, {})
# Discover switchbots nearby.
_btle_adv_data = await _btle_connect()
return _btle_adv_data
@staticmethod
@callback
def async_get_options_flow(
@ -66,62 +49,79 @@ class SwitchbotConfigFlow(ConfigFlow, domain=DOMAIN):
def __init__(self):
"""Initialize the config flow."""
self._discovered_devices = {}
self._discovered_adv: SwitchBotAdvertisement | None = None
self._discovered_advs: dict[str, SwitchBotAdvertisement] = {}
async def async_step_bluetooth(
self, discovery_info: BluetoothServiceInfo
) -> FlowResult:
"""Handle the bluetooth discovery step."""
_LOGGER.debug("Discovered bluetooth device: %s", discovery_info)
await self.async_set_unique_id(format_unique_id(discovery_info.address))
self._abort_if_unique_id_configured()
discovery_info_bleak = cast(BluetoothServiceInfoBleak, discovery_info)
parsed = parse_advertisement_data(
discovery_info_bleak.device, discovery_info_bleak.advertisement
)
if not parsed or parsed.data.get("modelName") not in SUPPORTED_MODEL_TYPES:
return self.async_abort(reason="not_supported")
self._discovered_adv = parsed
data = parsed.data
self.context["title_placeholders"] = {
"name": data["modelName"],
"address": discovery_info.address,
}
return await self.async_step_user()
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle a flow initiated by the user."""
"""Handle the user step to pick discovered device."""
errors: dict[str, str] = {}
if user_input is not None:
await self.async_set_unique_id(user_input[CONF_MAC].replace(":", ""))
address = user_input[CONF_ADDRESS]
await self.async_set_unique_id(
format_unique_id(address), raise_on_progress=False
)
self._abort_if_unique_id_configured()
user_input[CONF_SENSOR_TYPE] = SUPPORTED_MODEL_TYPES[
self._discovered_devices[self.unique_id]["modelName"]
self._discovered_advs[address].data["modelName"]
]
return self.async_create_entry(title=user_input[CONF_NAME], data=user_input)
try:
self._discovered_devices = await self._get_switchbots()
for device in self._discovered_devices.values():
_LOGGER.debug("Found %s", device)
if discovery := self._discovered_adv:
self._discovered_advs[discovery.address] = discovery
else:
current_addresses = self._async_current_ids()
for discovery_info in async_discovered_service_info(self.hass):
address = discovery_info.address
if (
format_unique_id(address) in current_addresses
or address in self._discovered_advs
):
continue
parsed = parse_advertisement_data(
discovery_info.device, discovery_info.advertisement
)
if parsed and parsed.data.get("modelName") in SUPPORTED_MODEL_TYPES:
self._discovered_advs[address] = parsed
except NotConnectedError:
return self.async_abort(reason="cannot_connect")
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unexpected exception")
return self.async_abort(reason="unknown")
# Get devices already configured.
configured_devices = {
item.data[CONF_MAC]
for item in self._async_current_entries(include_ignore=False)
}
# Get supported devices not yet configured.
unconfigured_devices = {
device["mac_address"]: f"{device['mac_address']} {device['modelName']}"
for device in self._discovered_devices.values()
if device.get("modelName") in SUPPORTED_MODEL_TYPES
and device["mac_address"] not in configured_devices
}
if not unconfigured_devices:
if not self._discovered_advs:
return self.async_abort(reason="no_unconfigured_devices")
data_schema = vol.Schema(
{
vol.Required(CONF_MAC): vol.In(unconfigured_devices),
vol.Required(CONF_ADDRESS): vol.In(
{
address: f"{parsed.data['modelName']} ({address})"
for address, parsed in self._discovered_advs.items()
}
),
vol.Required(CONF_NAME): str,
vol.Optional(CONF_PASSWORD): str,
}
)
return self.async_show_form(
step_id="user", data_schema=data_schema, errors=errors
)
@ -148,13 +148,6 @@ class SwitchbotOptionsFlowHandler(OptionsFlow):
return self.async_create_entry(title="", data=user_input)
options = {
vol.Optional(
CONF_TIME_BETWEEN_UPDATE_COMMAND,
default=self.config_entry.options.get(
CONF_TIME_BETWEEN_UPDATE_COMMAND,
DEFAULT_TIME_BETWEEN_UPDATE_COMMAND,
),
): int,
vol.Optional(
CONF_RETRY_COUNT,
default=self.config_entry.options.get(
@ -167,16 +160,6 @@ class SwitchbotOptionsFlowHandler(OptionsFlow):
CONF_RETRY_TIMEOUT, DEFAULT_RETRY_TIMEOUT
),
): int,
vol.Optional(
CONF_SCAN_TIMEOUT,
default=self.config_entry.options.get(
CONF_SCAN_TIMEOUT, DEFAULT_SCAN_TIMEOUT
),
): int,
}
return self.async_show_form(step_id="init", data_schema=vol.Schema(options))
class NotConnectedError(Exception):
"""Exception for unable to find device."""

9
homeassistant/components/switchbot/const.py
View File

@ -16,15 +16,14 @@ SUPPORTED_MODEL_TYPES = {
# Config Defaults
DEFAULT_RETRY_COUNT = 3
DEFAULT_RETRY_TIMEOUT = 5
DEFAULT_TIME_BETWEEN_UPDATE_COMMAND = 60
DEFAULT_SCAN_TIMEOUT = 5
# Config Options
CONF_TIME_BETWEEN_UPDATE_COMMAND = "update_time"
CONF_RETRY_COUNT = "retry_count"
CONF_RETRY_TIMEOUT = "retry_timeout"
CONF_SCAN_TIMEOUT = "scan_timeout"
# Deprecated config Entry Options to be removed in 2023.4
CONF_TIME_BETWEEN_UPDATE_COMMAND = "update_time"
CONF_RETRY_TIMEOUT = "retry_timeout"
# Data
DATA_COORDINATOR = "coordinator"
COMMON_OPTIONS = "common_options"

82
homeassistant/components/switchbot/coordinator.py
View File

@ -1,15 +1,22 @@
"""Provides the switchbot DataUpdateCoordinator."""
from __future__ import annotations
from datetime import timedelta
import asyncio
from collections.abc import Mapping
import logging
from typing import Any, cast
from bleak.backends.device import BLEDevice
import switchbot
from switchbot import parse_advertisement_data
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from homeassistant.components import bluetooth
from homeassistant.components.bluetooth.passive_update_coordinator import (
PassiveBluetoothDataUpdateCoordinator,
)
from homeassistant.core import HomeAssistant, callback
from .const import DOMAIN
from .const import CONF_RETRY_COUNT
_LOGGER = logging.getLogger(__name__)
@ -22,40 +29,53 @@ def flatten_sensors_data(sensor):
return sensor
class SwitchbotDataUpdateCoordinator(DataUpdateCoordinator):
class SwitchbotDataUpdateCoordinator(PassiveBluetoothDataUpdateCoordinator):
"""Class to manage fetching switchbot data."""
def __init__(
self,
hass: HomeAssistant,
*,
update_interval: int,
api: switchbot,
retry_count: int,
scan_timeout: int,
logger: logging.Logger,
ble_device: BLEDevice,
device: switchbot.SwitchbotDevice,
common_options: Mapping[str, int],
) -> None:
"""Initialize global switchbot data updater."""
self.switchbot_api = api
self.switchbot_data = self.switchbot_api.GetSwitchbotDevices()
self.retry_count = retry_count
self.scan_timeout = scan_timeout
self.update_interval = timedelta(seconds=update_interval)
super().__init__(hass, logger, ble_device.address)
self.ble_device = ble_device
self.device = device
self.common_options = common_options
self.data: dict[str, Any] = {}
self._ready_event = asyncio.Event()
super().__init__(
hass, _LOGGER, name=DOMAIN, update_interval=self.update_interval
)
@property
def retry_count(self) -> int:
"""Return retry count."""
return self.common_options[CONF_RETRY_COUNT]
async def _async_update_data(self) -> dict | None:
"""Fetch data from switchbot."""
@callback
def _async_handle_bluetooth_event(
self,
service_info: bluetooth.BluetoothServiceInfo,
change: bluetooth.BluetoothChange,
) -> None:
"""Handle a Bluetooth event."""
super()._async_handle_bluetooth_event(service_info, change)
discovery_info_bleak = cast(bluetooth.BluetoothServiceInfoBleak, service_info)
if adv := parse_advertisement_data(
discovery_info_bleak.device, discovery_info_bleak.advertisement
):
self.data = flatten_sensors_data(adv.data)
if "modelName" in self.data:
self._ready_event.set()
_LOGGER.debug("%s: Switchbot data: %s", self.ble_device.address, self.data)
self.device.update_from_advertisement(adv)
self.async_update_listeners()
switchbot_data = await self.switchbot_data.discover(
retry=self.retry_count, scan_timeout=self.scan_timeout
)
if not switchbot_data:
raise UpdateFailed("Unable to fetch switchbot services data")
return {
identifier: flatten_sensors_data(sensor)
for identifier, sensor in switchbot_data.items()
}
async def async_wait_ready(self) -> bool:
"""Wait for the device to be ready."""
try:
await asyncio.wait_for(self._ready_event.wait(), timeout=55)
except asyncio.TimeoutError:
return False
return True

43
homeassistant/components/switchbot/cover.py
View File

@ -14,13 +14,12 @@ from homeassistant.components.cover import (
CoverEntityFeature,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_MAC, CONF_NAME, CONF_PASSWORD
from homeassistant.const import CONF_ADDRESS, CONF_NAME
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import PlatformNotReady
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.restore_state import RestoreEntity
from .const import CONF_RETRY_COUNT, DATA_COORDINATOR, DOMAIN
from .const import DOMAIN
from .coordinator import SwitchbotDataUpdateCoordinator
from .entity import SwitchbotEntity
@ -33,25 +32,17 @@ async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up Switchbot curtain based on a config entry."""
coordinator: SwitchbotDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id][
DATA_COORDINATOR
]
if not coordinator.data.get(entry.unique_id):
raise PlatformNotReady
coordinator: SwitchbotDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
unique_id = entry.unique_id
assert unique_id is not None
async_add_entities(
[
SwitchBotCurtainEntity(
coordinator,
entry.unique_id,
entry.data[CONF_MAC],
unique_id,
entry.data[CONF_ADDRESS],
entry.data[CONF_NAME],
coordinator.switchbot_api.SwitchbotCurtain(
mac=entry.data[CONF_MAC],
password=entry.data.get(CONF_PASSWORD),
retry_count=entry.options[CONF_RETRY_COUNT],
),
coordinator.device,
)
]
)
@ -67,19 +58,18 @@ class SwitchBotCurtainEntity(SwitchbotEntity, CoverEntity, RestoreEntity):
| CoverEntityFeature.STOP
| CoverEntityFeature.SET_POSITION
)
_attr_assumed_state = True
def __init__(
self,
coordinator: SwitchbotDataUpdateCoordinator,
idx: str | None,
mac: str,
unique_id: str,
address: str,
name: str,
device: SwitchbotCurtain,
) -> None:
"""Initialize the Switchbot."""
super().__init__(coordinator, idx, mac, name)
self._attr_unique_id = idx
super().__init__(coordinator, unique_id, address, name)
self._attr_unique_id = unique_id
self._attr_is_closed = None
self._device = device
@ -97,21 +87,21 @@ class SwitchBotCurtainEntity(SwitchbotEntity, CoverEntity, RestoreEntity):
async def async_open_cover(self, **kwargs: Any) -> None:
"""Open the curtain."""
_LOGGER.debug("Switchbot to open curtain %s", self._mac)
_LOGGER.debug("Switchbot to open curtain %s", self._address)
self._last_run_success = bool(await self._device.open())
self.async_write_ha_state()
async def async_close_cover(self, **kwargs: Any) -> None:
"""Close the curtain."""
_LOGGER.debug("Switchbot to close the curtain %s", self._mac)
_LOGGER.debug("Switchbot to close the curtain %s", self._address)
self._last_run_success = bool(await self._device.close())
self.async_write_ha_state()
async def async_stop_cover(self, **kwargs: Any) -> None:
"""Stop the moving of this device."""
_LOGGER.debug("Switchbot to stop %s", self._mac)
_LOGGER.debug("Switchbot to stop %s", self._address)
self._last_run_success = bool(await self._device.stop())
self.async_write_ha_state()
@ -119,7 +109,7 @@ class SwitchBotCurtainEntity(SwitchbotEntity, CoverEntity, RestoreEntity):
"""Move the cover shutter to a specific position."""
position = kwargs.get(ATTR_POSITION)
_LOGGER.debug("Switchbot to move at %d %s", position, self._mac)
_LOGGER.debug("Switchbot to move at %d %s", position, self._address)
self._last_run_success = bool(await self._device.set_position(position))
self.async_write_ha_state()
@ -128,4 +118,5 @@ class SwitchBotCurtainEntity(SwitchbotEntity, CoverEntity, RestoreEntity):
"""Handle updated data from the coordinator."""
self._attr_current_cover_position = self.data["data"]["position"]
self._attr_is_closed = self.data["data"]["position"] <= 20
self._attr_is_opening = self.data["data"]["inMotion"]
self.async_write_ha_state()

35
homeassistant/components/switchbot/entity.py
View File

@ -4,43 +4,58 @@ from __future__ import annotations
from collections.abc import Mapping
from typing import Any
from homeassistant.components.bluetooth.passive_update_coordinator import (
PassiveBluetoothCoordinatorEntity,
)
from homeassistant.const import ATTR_CONNECTIONS
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.entity import DeviceInfo, Entity
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from homeassistant.helpers.entity import DeviceInfo
from .const import MANUFACTURER
from .coordinator import SwitchbotDataUpdateCoordinator
class SwitchbotEntity(CoordinatorEntity[SwitchbotDataUpdateCoordinator], Entity):
class SwitchbotEntity(PassiveBluetoothCoordinatorEntity):
"""Generic entity encapsulating common features of Switchbot device."""
coordinator: SwitchbotDataUpdateCoordinator
def __init__(
self,
coordinator: SwitchbotDataUpdateCoordinator,
idx: str | None,
mac: str,
unique_id: str,
address: str,
name: str,
) -> None:
"""Initialize the entity."""
super().__init__(coordinator)
self._last_run_success: bool | None = None
self._idx = idx
self._mac = mac
self._unique_id = unique_id
self._address = address
self._attr_name = name
self._attr_device_info = DeviceInfo(
connections={(dr.CONNECTION_NETWORK_MAC, self._mac)},
connections={(dr.CONNECTION_BLUETOOTH, self._address)},
manufacturer=MANUFACTURER,
model=self.data["modelName"],
name=name,
)
if ":" not in self._address:
# MacOS Bluetooth addresses are not mac addresses
return
# If the bluetooth address is also a mac address,
# add this connection as well to prevent a new device
# entry from being created when upgrading from a previous
# version of the integration.
self._attr_device_info[ATTR_CONNECTIONS].add(
(dr.CONNECTION_NETWORK_MAC, self._address)
)
@property
def data(self) -> dict[str, Any]:
"""Return coordinator data for this entity."""
return self.coordinator.data[self._idx]
return self.coordinator.data
@property
def extra_state_attributes(self) -> Mapping[Any, Any]:
"""Return the state attributes."""
return {"last_run_success": self._last_run_success, "mac_address": self._mac}
return {"last_run_success": self._last_run_success}

5
homeassistant/components/switchbot/manifest.json
View File

@ -2,10 +2,11 @@
"domain": "switchbot",
"name": "SwitchBot",
"documentation": "https://www.home-assistant.io/integrations/switchbot",
"requirements": ["PySwitchbot==0.14.1"],
"requirements": ["PySwitchbot==0.15.0"],
"config_flow": true,
"dependencies": ["bluetooth"],
"codeowners": ["@danielhiversen", "@RenierM26", "@murtas"],
"bluetooth": [{ "service_uuid": "cba20d00-224d-11e6-9fb8-0002a5d5c51b" }],
"iot_class": "local_polling",
"iot_class": "local_push",
"loggers": ["switchbot"]
}

29
homeassistant/components/switchbot/sensor.py
View File

@ -8,18 +8,17 @@ from homeassistant.components.sensor import (
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_MAC,
CONF_ADDRESS,
CONF_NAME,
PERCENTAGE,
SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
TEMP_CELSIUS,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import PlatformNotReady
from homeassistant.helpers.entity import EntityCategory
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DATA_COORDINATOR, DOMAIN
from .const import DOMAIN
from .coordinator import SwitchbotDataUpdateCoordinator
from .entity import SwitchbotEntity
@ -61,23 +60,19 @@ async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up Switchbot sensor based on a config entry."""
coordinator: SwitchbotDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id][
DATA_COORDINATOR
]
if not coordinator.data.get(entry.unique_id):
raise PlatformNotReady
coordinator: SwitchbotDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
unique_id = entry.unique_id
assert unique_id is not None
async_add_entities(
[
SwitchBotSensor(
coordinator,
entry.unique_id,
unique_id,
sensor,
entry.data[CONF_MAC],
entry.data[CONF_ADDRESS],
entry.data[CONF_NAME],
)
for sensor in coordinator.data[entry.unique_id]["data"]
for sensor in coordinator.data["data"]
if sensor in SENSOR_TYPES
]
)
@ -89,15 +84,15 @@ class SwitchBotSensor(SwitchbotEntity, SensorEntity):
def __init__(
self,
coordinator: SwitchbotDataUpdateCoordinator,
idx: str | None,
unique_id: str,
sensor: str,
mac: str,
address: str,
switchbot_name: str,
) -> None:
"""Initialize the Switchbot sensor."""
super().__init__(coordinator, idx, mac, name=switchbot_name)
super().__init__(coordinator, unique_id, address, name=switchbot_name)
self._sensor = sensor
self._attr_unique_id = f"{idx}-{sensor}"
self._attr_unique_id = f"{unique_id}-{sensor}"
self._attr_name = f"{switchbot_name} {sensor.title()}"
self.entity_description = SENSOR_TYPES[sensor]

8
homeassistant/components/switchbot/strings.json
View File

@ -1,11 +1,11 @@
{
"config": {
"flow_title": "{name}",
"flow_title": "{name} ({address})",
"step": {
"user": {
"title": "Setup Switchbot device",
"data": {
"mac": "Device MAC address",
"address": "Device address",
"name": "[%key:common::config_flow::data::name%]",
"password": "[%key:common::config_flow::data::password%]"
}
@ -24,10 +24,8 @@
"step": {
"init": {
"data": {
"update_time": "Time between updates (seconds)",
"retry_count": "Retry count",
"retry_timeout": "Timeout between retries",
"scan_timeout": "How long to scan for advertisement data"
"retry_timeout": "Timeout between retries"
}
}
}

37
homeassistant/components/switchbot/switch.py
View File

@ -8,13 +8,12 @@ from switchbot import Switchbot
from homeassistant.components.switch import SwitchDeviceClass, SwitchEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_MAC, CONF_NAME, CONF_PASSWORD, STATE_ON
from homeassistant.const import CONF_ADDRESS, CONF_NAME, STATE_ON
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import PlatformNotReady
from homeassistant.helpers import entity_platform
from homeassistant.helpers.restore_state import RestoreEntity
from .const import CONF_RETRY_COUNT, DATA_COORDINATOR, DOMAIN
from .const import DOMAIN
from .coordinator import SwitchbotDataUpdateCoordinator
from .entity import SwitchbotEntity
@ -29,25 +28,17 @@ async def async_setup_entry(
async_add_entities: entity_platform.AddEntitiesCallback,
) -> None:
"""Set up Switchbot based on a config entry."""
coordinator: SwitchbotDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id][
DATA_COORDINATOR
]
if not coordinator.data.get(entry.unique_id):
raise PlatformNotReady
coordinator: SwitchbotDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
unique_id = entry.unique_id
assert unique_id is not None
async_add_entities(
[
SwitchBotBotEntity(
coordinator,
entry.unique_id,
entry.data[CONF_MAC],
unique_id,
entry.data[CONF_ADDRESS],
entry.data[CONF_NAME],
coordinator.switchbot_api.Switchbot(
mac=entry.data[CONF_MAC],
password=entry.data.get(CONF_PASSWORD),
retry_count=entry.options[CONF_RETRY_COUNT],
),
coordinator.device,
)
]
)
@ -61,14 +52,14 @@ class SwitchBotBotEntity(SwitchbotEntity, SwitchEntity, RestoreEntity):
def __init__(
self,
coordinator: SwitchbotDataUpdateCoordinator,
idx: str | None,
mac: str,
unique_id: str,
address: str,
name: str,
device: Switchbot,
) -> None:
"""Initialize the Switchbot."""
super().__init__(coordinator, idx, mac, name)
self._attr_unique_id = idx
super().__init__(coordinator, unique_id, address, name)
self._attr_unique_id = unique_id
self._device = device
self._attr_is_on = False
@ -82,7 +73,7 @@ class SwitchBotBotEntity(SwitchbotEntity, SwitchEntity, RestoreEntity):
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn device on."""
_LOGGER.info("Turn Switchbot bot on %s", self._mac)
_LOGGER.info("Turn Switchbot bot on %s", self._address)
self._last_run_success = bool(await self._device.turn_on())
if self._last_run_success:
@ -91,7 +82,7 @@ class SwitchBotBotEntity(SwitchbotEntity, SwitchEntity, RestoreEntity):
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn device off."""
_LOGGER.info("Turn Switchbot bot off %s", self._mac)
_LOGGER.info("Turn Switchbot bot off %s", self._address)
self._last_run_success = bool(await self._device.turn_off())
if self._last_run_success:

9
homeassistant/components/switchbot/translations/en.json
View File

@ -7,11 +7,12 @@
"switchbot_unsupported_type": "Unsupported Switchbot Type.",
"unknown": "Unexpected error"
},
"flow_title": "{name}",
"error": {},
"flow_title": "{name} ({address})",
"step": {
"user": {
"data": {
"mac": "Device MAC address",
"address": "Device address",
"name": "Name",
"password": "Password"
},
@ -24,9 +25,7 @@
"init": {
"data": {
"retry_count": "Retry count",
"retry_timeout": "Timeout between retries",
"scan_timeout": "How long to scan for advertisement data",
"update_time": "Time between updates (seconds)"
"retry_timeout": "Timeout between retries"
}
}
}

2
requirements_all.txt
View File

@ -37,7 +37,7 @@ PyRMVtransport==0.3.3
PySocks==1.7.1
# homeassistant.components.switchbot
PySwitchbot==0.14.1
PySwitchbot==0.15.0
# homeassistant.components.transport_nsw
PyTransportNSW==0.1.1

2
requirements_test_all.txt
View File

@ -33,7 +33,7 @@ PyRMVtransport==0.3.3
PySocks==1.7.1
# homeassistant.components.switchbot
PySwitchbot==0.14.1
PySwitchbot==0.15.0
# homeassistant.components.transport_nsw
PyTransportNSW==0.1.1

82
tests/components/switchbot/__init__.py
View File

@ -1,7 +1,11 @@
"""Tests for the switchbot integration."""
from unittest.mock import patch
from homeassistant.const import CONF_MAC, CONF_NAME, CONF_PASSWORD
from bleak.backends.device import BLEDevice
from bleak.backends.scanner import AdvertisementData
from homeassistant.components.bluetooth import BluetoothServiceInfoBleak
from homeassistant.const import CONF_ADDRESS, CONF_NAME, CONF_PASSWORD
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
@ -11,37 +15,37 @@ DOMAIN = "switchbot"
ENTRY_CONFIG = {
CONF_NAME: "test-name",
CONF_PASSWORD: "test-password",
CONF_MAC: "e7:89:43:99:99:99",
CONF_ADDRESS: "e7:89:43:99:99:99",
}
USER_INPUT = {
CONF_NAME: "test-name",
CONF_PASSWORD: "test-password",
CONF_MAC: "e7:89:43:99:99:99",
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
}
USER_INPUT_CURTAIN = {
CONF_NAME: "test-name",
CONF_PASSWORD: "test-password",
CONF_MAC: "e7:89:43:90:90:90",
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
}
USER_INPUT_SENSOR = {
CONF_NAME: "test-name",
CONF_PASSWORD: "test-password",
CONF_MAC: "c0:ce:b0:d4:26:be",
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
}
USER_INPUT_UNSUPPORTED_DEVICE = {
CONF_NAME: "test-name",
CONF_PASSWORD: "test-password",
CONF_MAC: "test",
CONF_ADDRESS: "test",
}
USER_INPUT_INVALID = {
CONF_NAME: "test-name",
CONF_PASSWORD: "test-password",
CONF_MAC: "invalid-mac",
CONF_ADDRESS: "invalid-mac",
}
@ -68,3 +72,67 @@ async def init_integration(
await hass.async_block_till_done()
return entry
WOHAND_SERVICE_INFO = BluetoothServiceInfoBleak(
name="WoHand",
manufacturer_data={89: b"\xfd`0U\x92W"},
service_data={"00000d00-0000-1000-8000-00805f9b34fb": b"H\x90\xd9"},
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
address="aa:bb:cc:dd:ee:ff",
rssi=-60,
source="local",
advertisement=AdvertisementData(
local_name="WoHand",
manufacturer_data={89: b"\xfd`0U\x92W"},
service_data={"00000d00-0000-1000-8000-00805f9b34fb": b"H\x90\xd9"},
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
),
device=BLEDevice("aa:bb:cc:dd:ee:ff", "WoHand"),
)
WOCURTAIN_SERVICE_INFO = BluetoothServiceInfoBleak(
name="WoCurtain",
address="aa:bb:cc:dd:ee:ff",
manufacturer_data={89: b"\xc1\xc7'}U\xab"},
service_data={"00000d00-0000-1000-8000-00805f9b34fb": b"c\xd0Y\x00\x11\x04"},
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
rssi=-60,
source="local",
advertisement=AdvertisementData(
local_name="WoCurtain",
manufacturer_data={89: b"\xc1\xc7'}U\xab"},
service_data={"00000d00-0000-1000-8000-00805f9b34fb": b"c\xd0Y\x00\x11\x04"},
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
),
device=BLEDevice("aa:bb:cc:dd:ee:ff", "WoCurtain"),
)
WOSENSORTH_SERVICE_INFO = BluetoothServiceInfoBleak(
name="WoSensorTH",
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
address="aa:bb:cc:dd:ee:ff",
manufacturer_data={2409: b"\xda,\x1e\xb1\x86Au\x03\x00\x96\xac"},
service_data={"0000fd3d-0000-1000-8000-00805f9b34fb": b"T\x00d\x00\x96\xac"},
rssi=-60,
source="local",
advertisement=AdvertisementData(
manufacturer_data={2409: b"\xda,\x1e\xb1\x86Au\x03\x00\x96\xac"},
service_data={"0000fd3d-0000-1000-8000-00805f9b34fb": b"T\x00d\x00\x96\xac"},
),
device=BLEDevice("aa:bb:cc:dd:ee:ff", "WoSensorTH"),
)
NOT_SWITCHBOT_INFO = BluetoothServiceInfoBleak(
name="unknown",
service_uuids=[],
address="aa:bb:cc:dd:ee:ff",
manufacturer_data={},
service_data={},
rssi=-60,
source="local",
advertisement=AdvertisementData(
manufacturer_data={},
service_data={},
),
device=BLEDevice("aa:bb:cc:dd:ee:ff", "unknown"),
)

139
tests/components/switchbot/conftest.py
View File

@ -1,139 +1,8 @@
"""Define fixtures available for all tests."""
import sys
from unittest.mock import MagicMock, patch
from pytest import fixture
import pytest
class MocGetSwitchbotDevices:
"""Scan for all Switchbot devices and return by type."""
def __init__(self, interface=None) -> None:
"""Get switchbot devices class constructor."""
self._interface = interface
self._all_services_data = {
"e78943999999": {
"mac_address": "e7:89:43:99:99:99",
"isEncrypted": False,
"model": "H",
"data": {
"switchMode": "true",
"isOn": "true",
"battery": 91,
"rssi": -71,
},
"modelName": "WoHand",
},
"e78943909090": {
"mac_address": "e7:89:43:90:90:90",
"isEncrypted": False,
"model": "c",
"data": {
"calibration": True,
"battery": 74,
"inMotion": False,
"position": 100,
"lightLevel": 2,
"deviceChain": 1,
"rssi": -73,
},
"modelName": "WoCurtain",
},
"ffffff19ffff": {
"mac_address": "ff:ff:ff:19:ff:ff",
"isEncrypted": False,
"model": "m",
"rawAdvData": "000d6d00",
},
"c0ceb0d426be": {
"mac_address": "c0:ce:b0:d4:26:be",
"isEncrypted": False,
"data": {
"temp": {"c": 21.6, "f": 70.88},
"fahrenheit": False,
"humidity": 73,
"battery": 100,
"rssi": -58,
},
"model": "T",
"modelName": "WoSensorTH",
},
}
self._curtain_all_services_data = {
"mac_address": "e7:89:43:90:90:90",
"isEncrypted": False,
"model": "c",
"data": {
"calibration": True,
"battery": 74,
"position": 100,
"lightLevel": 2,
"rssi": -73,
},
"modelName": "WoCurtain",
}
self._sensor_data = {
"mac_address": "c0:ce:b0:d4:26:be",
"isEncrypted": False,
"data": {
"temp": {"c": 21.6, "f": 70.88},
"fahrenheit": False,
"humidity": 73,
"battery": 100,
"rssi": -58,
},
"model": "T",
"modelName": "WoSensorTH",
}
self._unsupported_device = {
"mac_address": "test",
"isEncrypted": False,
"model": "HoN",
"data": {
"switchMode": "true",
"isOn": "true",
"battery": 91,
"rssi": -71,
},
"modelName": "WoOther",
}
async def discover(self, retry=0, scan_timeout=0):
"""Mock discover."""
return self._all_services_data
async def get_device_data(self, mac=None):
"""Return data for specific device."""
if mac == "e7:89:43:99:99:99":
return self._all_services_data
if mac == "test":
return self._unsupported_device
if mac == "e7:89:43:90:90:90":
return self._curtain_all_services_data
if mac == "c0:ce:b0:d4:26:be":
return self._sensor_data
return None
class MocNotConnectedError(Exception):
"""Mock exception."""
module = type(sys)("switchbot")
module.GetSwitchbotDevices = MocGetSwitchbotDevices
module.NotConnectedError = MocNotConnectedError
sys.modules["switchbot"] = module
@fixture
def switchbot_config_flow(hass):
"""Mock the bluepy api for easier config flow testing."""
with patch.object(MocGetSwitchbotDevices, "discover", return_value=True), patch(
"homeassistant.components.switchbot.config_flow.GetSwitchbotDevices"
) as mock_switchbot:
instance = mock_switchbot.return_value
instance.discover = MagicMock(return_value=True)
yield mock_switchbot
@pytest.fixture(autouse=True)
def mock_bluetooth(enable_bluetooth):
"""Auto mock bluetooth."""

232
tests/components/switchbot/test_config_flow.py
View File

@ -1,33 +1,104 @@
"""Test the switchbot config flow."""
from homeassistant.components.switchbot.config_flow import NotConnectedError
from unittest.mock import patch
from homeassistant.components.switchbot.const import (
CONF_RETRY_COUNT,
CONF_RETRY_TIMEOUT,
CONF_SCAN_TIMEOUT,
CONF_TIME_BETWEEN_UPDATE_COMMAND,
)
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_MAC, CONF_NAME, CONF_PASSWORD, CONF_SENSOR_TYPE
from homeassistant.config_entries import SOURCE_BLUETOOTH, SOURCE_USER
from homeassistant.const import CONF_ADDRESS, CONF_NAME, CONF_PASSWORD, CONF_SENSOR_TYPE
from homeassistant.data_entry_flow import FlowResultType
from . import (
NOT_SWITCHBOT_INFO,
USER_INPUT,
USER_INPUT_CURTAIN,
USER_INPUT_SENSOR,
WOCURTAIN_SERVICE_INFO,
WOHAND_SERVICE_INFO,
WOSENSORTH_SERVICE_INFO,
init_integration,
patch_async_setup_entry,
)
from tests.common import MockConfigEntry
DOMAIN = "switchbot"
async def test_user_form_valid_mac(hass):
async def test_bluetooth_discovery(hass):
"""Test discovery via bluetooth with a valid device."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_BLUETOOTH},
data=WOHAND_SERVICE_INFO,
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
with patch_async_setup_entry() as mock_setup_entry:
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
USER_INPUT,
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "test-name"
assert result["data"] == {
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
CONF_NAME: "test-name",
CONF_PASSWORD: "test-password",
CONF_SENSOR_TYPE: "bot",
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_bluetooth_discovery_already_setup(hass):
"""Test discovery via bluetooth with a valid device when already setup."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
CONF_NAME: "test-name",
CONF_PASSWORD: "test-password",
CONF_SENSOR_TYPE: "bot",
},
unique_id="aabbccddeeff",
)
entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_BLUETOOTH},
data=WOHAND_SERVICE_INFO,
)
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "already_configured"
async def test_async_step_bluetooth_not_switchbot(hass):
"""Test discovery via bluetooth not switchbot."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_BLUETOOTH},
data=NOT_SWITCHBOT_INFO,
)
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "not_supported"
async def test_user_setup_wohand(hass):
"""Test the user initiated form with password and valid mac."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
with patch(
"homeassistant.components.switchbot.config_flow.async_discovered_service_info",
return_value=[WOHAND_SERVICE_INFO],
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {}
@ -42,7 +113,7 @@ async def test_user_form_valid_mac(hass):
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "test-name"
assert result["data"] == {
CONF_MAC: "e7:89:43:99:99:99",
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
CONF_NAME: "test-name",
CONF_PASSWORD: "test-password",
CONF_SENSOR_TYPE: "bot",
@ -50,11 +121,41 @@ async def test_user_form_valid_mac(hass):
assert len(mock_setup_entry.mock_calls) == 1
# test curtain device creation.
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
async def test_user_setup_wohand_already_configured(hass):
"""Test the user initiated form with password and valid mac."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
CONF_NAME: "test-name",
CONF_PASSWORD: "test-password",
CONF_SENSOR_TYPE: "bot",
},
unique_id="aabbccddeeff",
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.switchbot.config_flow.async_discovered_service_info",
return_value=[WOHAND_SERVICE_INFO],
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "no_unconfigured_devices"
async def test_user_setup_wocurtain(hass):
"""Test the user initiated form with password and valid mac."""
with patch(
"homeassistant.components.switchbot.config_flow.async_discovered_service_info",
return_value=[WOCURTAIN_SERVICE_INFO],
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {}
@ -69,7 +170,7 @@ async def test_user_form_valid_mac(hass):
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "test-name"
assert result["data"] == {
CONF_MAC: "e7:89:43:90:90:90",
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
CONF_NAME: "test-name",
CONF_PASSWORD: "test-password",
CONF_SENSOR_TYPE: "curtain",
@ -77,11 +178,16 @@ async def test_user_form_valid_mac(hass):
assert len(mock_setup_entry.mock_calls) == 1
# test sensor device creation.
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
async def test_user_setup_wosensor(hass):
"""Test the user initiated form with password and valid mac."""
with patch(
"homeassistant.components.switchbot.config_flow.async_discovered_service_info",
return_value=[WOSENSORTH_SERVICE_INFO],
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {}
@ -96,7 +202,7 @@ async def test_user_form_valid_mac(hass):
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "test-name"
assert result["data"] == {
CONF_MAC: "c0:ce:b0:d4:26:be",
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
CONF_NAME: "test-name",
CONF_PASSWORD: "test-password",
CONF_SENSOR_TYPE: "hygrometer",
@ -104,39 +210,78 @@ async def test_user_form_valid_mac(hass):
assert len(mock_setup_entry.mock_calls) == 1
# tests abort if no unconfigured devices are found.
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
async def test_user_no_devices(hass):
"""Test the user initiated form with password and valid mac."""
with patch(
"homeassistant.components.switchbot.config_flow.async_discovered_service_info",
return_value=[],
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "no_unconfigured_devices"
async def test_user_form_exception(hass, switchbot_config_flow):
"""Test we handle exception on user form."""
switchbot_config_flow.side_effect = NotConnectedError
async def test_async_step_user_takes_precedence_over_discovery(hass):
"""Test manual setup takes precedence over discovery."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
DOMAIN,
context={"source": SOURCE_BLUETOOTH},
data=WOCURTAIN_SERVICE_INFO,
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "cannot_connect"
with patch(
"homeassistant.components.switchbot.config_flow.async_discovered_service_info",
return_value=[WOCURTAIN_SERVICE_INFO],
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
)
assert result["type"] == FlowResultType.FORM
switchbot_config_flow.side_effect = Exception
with patch_async_setup_entry() as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=USER_INPUT,
)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "test-name"
assert result2["data"] == {
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
CONF_NAME: "test-name",
CONF_PASSWORD: "test-password",
CONF_SENSOR_TYPE: "curtain",
}
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "unknown"
assert len(mock_setup_entry.mock_calls) == 1
# Verify the original one was aborted
assert not hass.config_entries.flow.async_progress(DOMAIN)
async def test_options_flow(hass):
"""Test updating options."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
CONF_NAME: "test-name",
CONF_PASSWORD: "test-password",
CONF_SENSOR_TYPE: "bot",
},
options={
CONF_RETRY_COUNT: 10,
CONF_RETRY_TIMEOUT: 10,
},
unique_id="aabbccddeeff",
)
entry.add_to_hass(hass)
with patch_async_setup_entry() as mock_setup_entry:
entry = await init_integration(hass)
@ -148,21 +293,17 @@ async def test_options_flow(hass):
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONF_TIME_BETWEEN_UPDATE_COMMAND: 60,
CONF_RETRY_COUNT: 3,
CONF_RETRY_TIMEOUT: 5,
CONF_SCAN_TIMEOUT: 5,
},
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["data"][CONF_TIME_BETWEEN_UPDATE_COMMAND] == 60
assert result["data"][CONF_RETRY_COUNT] == 3
assert result["data"][CONF_RETRY_TIMEOUT] == 5
assert result["data"][CONF_SCAN_TIMEOUT] == 5
assert len(mock_setup_entry.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 2
# Test changing of entry options.
@ -177,18 +318,17 @@ async def test_options_flow(hass):
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONF_TIME_BETWEEN_UPDATE_COMMAND: 66,
CONF_RETRY_COUNT: 6,
CONF_RETRY_TIMEOUT: 6,
CONF_SCAN_TIMEOUT: 6,
},
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["data"][CONF_TIME_BETWEEN_UPDATE_COMMAND] == 66
assert result["data"][CONF_RETRY_COUNT] == 6
assert result["data"][CONF_RETRY_TIMEOUT] == 6
assert result["data"][CONF_SCAN_TIMEOUT] == 6
assert len(mock_setup_entry.mock_calls) == 1
assert entry.options[CONF_RETRY_COUNT] == 6
assert entry.options[CONF_RETRY_TIMEOUT] == 6