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

Simplify switchbot config flow (#76272)

pull/76568/head
J. Nick Koston 2022-08-10 09:02:08 -10:00 committed by GitHub
parent 54fc17e10d
commit b1497b0857
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
13 changed files with 412 additions and 246 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -9,6 +9,7 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_ADDRESS,
CONF_MAC,
CONF_NAME,
CONF_PASSWORD,
CONF_SENSOR_TYPE,
Platform,
@ -17,31 +18,26 @@ from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import device_registry as dr
from .const import (
ATTR_BOT,
ATTR_CONTACT,
ATTR_CURTAIN,
ATTR_HYGROMETER,
ATTR_MOTION,
ATTR_PLUG,
CONF_RETRY_COUNT,
DEFAULT_RETRY_COUNT,
DOMAIN,
)
from .const import CONF_RETRY_COUNT, DEFAULT_RETRY_COUNT, DOMAIN, SupportedModels
from .coordinator import SwitchbotDataUpdateCoordinator
PLATFORMS_BY_TYPE = {
ATTR_BOT: [Platform.SWITCH, Platform.SENSOR],
ATTR_PLUG: [Platform.SWITCH, Platform.SENSOR],
ATTR_CURTAIN: [Platform.COVER, Platform.BINARY_SENSOR, Platform.SENSOR],
ATTR_HYGROMETER: [Platform.SENSOR],
ATTR_CONTACT: [Platform.BINARY_SENSOR, Platform.SENSOR],
ATTR_MOTION: [Platform.BINARY_SENSOR, Platform.SENSOR],
SupportedModels.BULB.value: [Platform.SENSOR],
SupportedModels.BOT.value: [Platform.SWITCH, Platform.SENSOR],
SupportedModels.PLUG.value: [Platform.SWITCH, Platform.SENSOR],
SupportedModels.CURTAIN.value: [
Platform.COVER,
Platform.BINARY_SENSOR,
Platform.SENSOR,
],
SupportedModels.HYGROMETER.value: [Platform.SENSOR],
SupportedModels.CONTACT.value: [Platform.BINARY_SENSOR, Platform.SENSOR],
SupportedModels.MOTION.value: [Platform.BINARY_SENSOR, Platform.SENSOR],
}
CLASS_BY_DEVICE = {
ATTR_CURTAIN: switchbot.SwitchbotCurtain,
ATTR_BOT: switchbot.Switchbot,
ATTR_PLUG: switchbot.SwitchbotPlugMini,
SupportedModels.CURTAIN.value: switchbot.SwitchbotCurtain,
SupportedModels.BOT.value: switchbot.Switchbot,
SupportedModels.PLUG.value: switchbot.SwitchbotPlugMini,
}
_LOGGER = logging.getLogger(__name__)
@ -49,6 +45,7 @@ _LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Switchbot from a config entry."""
assert entry.unique_id is not None
hass.data.setdefault(DOMAIN, {})
if CONF_ADDRESS not in entry.data and CONF_MAC in entry.data:
# Bleak uses addresses not mac addresses which are are actually
@ -81,7 +78,12 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
retry_count=entry.options[CONF_RETRY_COUNT],
)
coordinator = hass.data[DOMAIN][entry.entry_id] = SwitchbotDataUpdateCoordinator(
hass, _LOGGER, ble_device, device
hass,
_LOGGER,
ble_device,
device,
entry.unique_id,
entry.data.get(CONF_NAME, entry.title),
)
entry.async_on_unload(coordinator.async_start())
if not await coordinator.async_wait_ready():

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

@ -7,7 +7,6 @@ from homeassistant.components.binary_sensor import (
BinarySensorEntityDescription,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_ADDRESS, CONF_NAME
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import EntityCategory
from homeassistant.helpers.entity_platform import AddEntitiesCallback
@ -53,20 +52,10 @@ async def async_setup_entry(
) -> None:
"""Set up Switchbot curtain based on a config entry."""
coordinator: SwitchbotDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
unique_id = entry.unique_id
assert unique_id is not None
async_add_entities(
[
SwitchBotBinarySensor(
coordinator,
unique_id,
binary_sensor,
entry.data[CONF_ADDRESS],
entry.data[CONF_NAME],
)
for binary_sensor in coordinator.data["data"]
if binary_sensor in BINARY_SENSOR_TYPES
]
SwitchBotBinarySensor(coordinator, binary_sensor)
for binary_sensor in coordinator.data["data"]
if binary_sensor in BINARY_SENSOR_TYPES
)
@ -78,15 +67,12 @@ class SwitchBotBinarySensor(SwitchbotEntity, BinarySensorEntity):
def __init__(
self,
coordinator: SwitchbotDataUpdateCoordinator,
unique_id: str,
binary_sensor: str,
mac: str,
switchbot_name: str,
) -> None:
"""Initialize the Switchbot sensor."""
super().__init__(coordinator, unique_id, mac, name=switchbot_name)
super().__init__(coordinator)
self._sensor = binary_sensor
self._attr_unique_id = f"{unique_id}-{binary_sensor}"
self._attr_unique_id = f"{coordinator.base_unique_id}-{binary_sensor}"
self.entity_description = BINARY_SENSOR_TYPES[binary_sensor]
self._attr_name = self.entity_description.name

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

@ -12,9 +12,9 @@ from homeassistant.components.bluetooth import (
async_discovered_service_info,
)
from homeassistant.config_entries import ConfigEntry, ConfigFlow, OptionsFlow
from homeassistant.const import CONF_ADDRESS, CONF_NAME, CONF_PASSWORD, CONF_SENSOR_TYPE
from homeassistant.const import CONF_ADDRESS, CONF_PASSWORD, CONF_SENSOR_TYPE
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.data_entry_flow import AbortFlow, FlowResult
from .const import CONF_RETRY_COUNT, DEFAULT_RETRY_COUNT, DOMAIN, SUPPORTED_MODEL_TYPES
@ -26,6 +26,17 @@ def format_unique_id(address: str) -> str:
return address.replace(":", "").lower()
def short_address(address: str) -> str:
"""Convert a Bluetooth address to a short address."""
results = address.replace("-", ":").split(":")
return f"{results[-2].upper()}{results[-1].upper()}"[-4:]
def name_from_discovery(discovery: SwitchBotAdvertisement) -> str:
"""Get the name from a discovery."""
return f'{discovery.data["modelFriendlyName"]} {short_address(discovery.address)}'
class SwitchbotConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Switchbot."""
@ -59,62 +70,128 @@ class SwitchbotConfigFlow(ConfigFlow, domain=DOMAIN):
self._discovered_adv = parsed
data = parsed.data
self.context["title_placeholders"] = {
"name": data["modelName"],
"address": discovery_info.address,
"name": data["modelFriendlyName"],
"address": short_address(discovery_info.address),
}
return await self.async_step_user()
if self._discovered_adv.data["isEncrypted"]:
return await self.async_step_password()
return await self.async_step_confirm()
async def _async_create_entry_from_discovery(
self, user_input: dict[str, Any]
) -> FlowResult:
"""Create an entry from a discovery."""
assert self._discovered_adv is not None
discovery = self._discovered_adv
name = name_from_discovery(discovery)
model_name = discovery.data["modelName"]
return self.async_create_entry(
title=name,
data={
**user_input,
CONF_ADDRESS: discovery.address,
CONF_SENSOR_TYPE: str(SUPPORTED_MODEL_TYPES[model_name]),
},
)
async def async_step_confirm(self, user_input: dict[str, Any] = None) -> FlowResult:
"""Confirm a single device."""
assert self._discovered_adv is not None
if user_input is not None:
return await self._async_create_entry_from_discovery(user_input)
self._set_confirm_only()
return self.async_show_form(
step_id="confirm",
data_schema=vol.Schema({}),
description_placeholders={
"name": name_from_discovery(self._discovered_adv)
},
)
async def async_step_password(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the password step."""
assert self._discovered_adv is not None
if user_input is not None:
# There is currently no api to validate the password
# that does not operate the device so we have
# to accept it as-is
return await self._async_create_entry_from_discovery(user_input)
return self.async_show_form(
step_id="password",
data_schema=vol.Schema({vol.Required(CONF_PASSWORD): str}),
description_placeholders={
"name": name_from_discovery(self._discovered_adv)
},
)
@callback
def _async_discover_devices(self) -> None:
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
if not self._discovered_advs:
raise AbortFlow("no_unconfigured_devices")
async def _async_set_device(self, discovery: SwitchBotAdvertisement) -> None:
"""Set the device to work with."""
self._discovered_adv = discovery
address = discovery.address
await self.async_set_unique_id(
format_unique_id(address), raise_on_progress=False
)
self._abort_if_unique_id_configured()
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] = {}
device_adv: SwitchBotAdvertisement | None = None
if user_input is not None:
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_advs[address].data["modelName"]
]
return self.async_create_entry(title=user_input[CONF_NAME], data=user_input)
device_adv = self._discovered_advs[user_input[CONF_ADDRESS]]
await self._async_set_device(device_adv)
if device_adv.data["isEncrypted"]:
return await self.async_step_password()
return await self._async_create_entry_from_discovery(user_input)
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
self._async_discover_devices()
if len(self._discovered_advs) == 1:
# If there is only one device we can ask for a password
# or simply confirm it
device_adv = list(self._discovered_advs.values())[0]
await self._async_set_device(device_adv)
if device_adv.data["isEncrypted"]:
return await self.async_step_password()
return await self.async_step_confirm()
if not self._discovered_advs:
return self.async_abort(reason="no_unconfigured_devices")
data_schema = vol.Schema(
{
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
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_ADDRESS): vol.In(
{
address: name_from_discovery(parsed)
for address, parsed in self._discovered_advs.items()
}
),
}
),
errors=errors,
)

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

@ -1,25 +1,39 @@
"""Constants for the switchbot integration."""
from switchbot import SwitchbotModel
from homeassistant.backports.enum import StrEnum
DOMAIN = "switchbot"
MANUFACTURER = "switchbot"
# Config Attributes
ATTR_BOT = "bot"
ATTR_CURTAIN = "curtain"
ATTR_HYGROMETER = "hygrometer"
ATTR_CONTACT = "contact"
ATTR_PLUG = "plug"
ATTR_MOTION = "motion"
DEFAULT_NAME = "Switchbot"
class SupportedModels(StrEnum):
"""Supported Switchbot models."""
BOT = "bot"
BULB = "bulb"
CURTAIN = "curtain"
HYGROMETER = "hygrometer"
CONTACT = "contact"
PLUG = "plug"
MOTION = "motion"
SUPPORTED_MODEL_TYPES = {
"WoHand": ATTR_BOT,
"WoCurtain": ATTR_CURTAIN,
"WoSensorTH": ATTR_HYGROMETER,
"WoContact": ATTR_CONTACT,
"WoPlug": ATTR_PLUG,
"WoPresence": ATTR_MOTION,
SwitchbotModel.BOT: SupportedModels.BOT,
SwitchbotModel.CURTAIN: SupportedModels.CURTAIN,
SwitchbotModel.METER: SupportedModels.HYGROMETER,
SwitchbotModel.CONTACT_SENSOR: SupportedModels.CONTACT,
SwitchbotModel.PLUG_MINI: SupportedModels.PLUG,
SwitchbotModel.MOTION_SENSOR: SupportedModels.MOTION,
SwitchbotModel.COLOR_BULB: SupportedModels.BULB,
}
# Config Defaults
DEFAULT_RETRY_COUNT = 3

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

@ -37,6 +37,8 @@ class SwitchbotDataUpdateCoordinator(PassiveBluetoothDataUpdateCoordinator):
logger: logging.Logger,
ble_device: BLEDevice,
device: switchbot.SwitchbotDevice,
base_unique_id: str,
device_name: str,
) -> None:
"""Initialize global switchbot data updater."""
super().__init__(
@ -45,6 +47,8 @@ class SwitchbotDataUpdateCoordinator(PassiveBluetoothDataUpdateCoordinator):
self.ble_device = ble_device
self.device = device
self.data: dict[str, Any] = {}
self.device_name = device_name
self.base_unique_id = base_unique_id
self._ready_event = asyncio.Event()
@callback

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

@ -4,8 +4,6 @@ from __future__ import annotations
import logging
from typing import Any
from switchbot import SwitchbotCurtain
from homeassistant.components.cover import (
ATTR_CURRENT_POSITION,
ATTR_POSITION,
@ -14,7 +12,6 @@ from homeassistant.components.cover import (
CoverEntityFeature,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_ADDRESS, CONF_NAME
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.restore_state import RestoreEntity
@ -33,19 +30,7 @@ async def async_setup_entry(
) -> None:
"""Set up Switchbot curtain based on a config entry."""
coordinator: SwitchbotDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
unique_id = entry.unique_id
assert unique_id is not None
async_add_entities(
[
SwitchBotCurtainEntity(
coordinator,
unique_id,
entry.data[CONF_ADDRESS],
entry.data[CONF_NAME],
coordinator.device,
)
]
)
async_add_entities([SwitchBotCurtainEntity(coordinator)])
class SwitchBotCurtainEntity(SwitchbotEntity, CoverEntity, RestoreEntity):
@ -59,19 +44,10 @@ class SwitchBotCurtainEntity(SwitchbotEntity, CoverEntity, RestoreEntity):
| CoverEntityFeature.SET_POSITION
)
def __init__(
self,
coordinator: SwitchbotDataUpdateCoordinator,
unique_id: str,
address: str,
name: str,
device: SwitchbotCurtain,
) -> None:
def __init__(self, coordinator: SwitchbotDataUpdateCoordinator) -> None:
"""Initialize the Switchbot."""
super().__init__(coordinator, unique_id, address, name)
self._attr_unique_id = unique_id
super().__init__(coordinator)
self._attr_is_closed = None
self._device = device
async def async_added_to_hass(self) -> None:
"""Run when entity about to be added."""

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

@ -20,24 +20,19 @@ class SwitchbotEntity(PassiveBluetoothCoordinatorEntity):
coordinator: SwitchbotDataUpdateCoordinator
def __init__(
self,
coordinator: SwitchbotDataUpdateCoordinator,
unique_id: str,
address: str,
name: str,
) -> None:
def __init__(self, coordinator: SwitchbotDataUpdateCoordinator) -> None:
"""Initialize the entity."""
super().__init__(coordinator)
self._device = coordinator.device
self._last_run_success: bool | None = None
self._unique_id = unique_id
self._address = address
self._attr_name = name
self._address = coordinator.ble_device.address
self._attr_unique_id = coordinator.base_unique_id
self._attr_name = coordinator.device_name
self._attr_device_info = DeviceInfo(
connections={(dr.CONNECTION_BLUETOOTH, self._address)},
manufacturer=MANUFACTURER,
model=self.data["modelName"],
name=name,
name=coordinator.device_name,
)
if ":" not in self._address:
# MacOS Bluetooth addresses are not mac addresses

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

@ -9,8 +9,6 @@ from homeassistant.components.sensor import (
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_ADDRESS,
CONF_NAME,
PERCENTAGE,
SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
TEMP_CELSIUS,
@ -73,20 +71,13 @@ async def async_setup_entry(
) -> None:
"""Set up Switchbot sensor based on a config entry."""
coordinator: SwitchbotDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
unique_id = entry.unique_id
assert unique_id is not None
async_add_entities(
[
SwitchBotSensor(
coordinator,
unique_id,
sensor,
entry.data[CONF_ADDRESS],
entry.data[CONF_NAME],
)
for sensor in coordinator.data["data"]
if sensor in SENSOR_TYPES
]
SwitchBotSensor(
coordinator,
sensor,
)
for sensor in coordinator.data["data"]
if sensor in SENSOR_TYPES
)
@ -96,16 +87,14 @@ class SwitchBotSensor(SwitchbotEntity, SensorEntity):
def __init__(
self,
coordinator: SwitchbotDataUpdateCoordinator,
unique_id: str,
sensor: str,
address: str,
switchbot_name: str,
) -> None:
"""Initialize the Switchbot sensor."""
super().__init__(coordinator, unique_id, address, name=switchbot_name)
super().__init__(coordinator)
self._sensor = sensor
self._attr_unique_id = f"{unique_id}-{sensor}"
self._attr_name = f"{switchbot_name} {sensor.replace('_', ' ').title()}"
self._attr_unique_id = f"{coordinator.base_unique_id}-{sensor}"
name = coordinator.device_name
self._attr_name = f"{name} {sensor.replace('_', ' ').title()}"
self.entity_description = SENSOR_TYPES[sensor]
@property

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

@ -3,10 +3,16 @@
"flow_title": "{name} ({address})",
"step": {
"user": {
"title": "Setup Switchbot device",
"data": {
"address": "Device address",
"name": "[%key:common::config_flow::data::name%]",
"address": "Device address"
}
},
"confirm": {
"description": "Do you want to setup {name}?"
},
"password": {
"description": "The {name} device requires a password",
"data": {
"password": "[%key:common::config_flow::data::password%]"
}
}

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

@ -4,11 +4,9 @@ from __future__ import annotations
import logging
from typing import Any
from switchbot import Switchbot
from homeassistant.components.switch import SwitchDeviceClass, SwitchEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_ADDRESS, CONF_NAME, STATE_ON
from homeassistant.const import STATE_ON
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_platform
from homeassistant.helpers.restore_state import RestoreEntity
@ -29,19 +27,7 @@ async def async_setup_entry(
) -> None:
"""Set up Switchbot based on a config entry."""
coordinator: SwitchbotDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
unique_id = entry.unique_id
assert unique_id is not None
async_add_entities(
[
SwitchBotSwitch(
coordinator,
unique_id,
entry.data[CONF_ADDRESS],
entry.data[CONF_NAME],
coordinator.device,
)
]
)
async_add_entities([SwitchBotSwitch(coordinator)])
class SwitchBotSwitch(SwitchbotEntity, SwitchEntity, RestoreEntity):
@ -49,18 +35,9 @@ class SwitchBotSwitch(SwitchbotEntity, SwitchEntity, RestoreEntity):
_attr_device_class = SwitchDeviceClass.SWITCH
def __init__(
self,
coordinator: SwitchbotDataUpdateCoordinator,
unique_id: str,
address: str,
name: str,
device: Switchbot,
) -> None:
def __init__(self, coordinator: SwitchbotDataUpdateCoordinator) -> None:
"""Initialize the Switchbot."""
super().__init__(coordinator, unique_id, address, name)
self._attr_unique_id = unique_id
self._device = device
super().__init__(coordinator)
self._attr_is_on = False
async def async_added_to_hass(self) -> None:

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

@ -7,16 +7,22 @@
"switchbot_unsupported_type": "Unsupported Switchbot Type.",
"unknown": "Unexpected error"
},
"error": {},
"flow_title": "{name} ({address})",
"step": {
"user": {
"confirm": {
"description": "Do you want to setup {name}?"
},
"password": {
"data": {
"address": "Device address",
"mac": "Device MAC address",
"name": "Name",
"password": "Password"
},
"title": "Setup Switchbot device"
"description": "The {name} device requires a password"
},
"user": {
"data": {
"address": "Device address"
}
}
}
},
@ -24,10 +30,7 @@
"step": {
"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_count": "Retry count"
}
}
}

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

@ -5,7 +5,7 @@ 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.const import CONF_ADDRESS
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
@ -13,38 +13,18 @@ from tests.common import MockConfigEntry
DOMAIN = "switchbot"
ENTRY_CONFIG = {
CONF_NAME: "test-name",
CONF_PASSWORD: "test-password",
CONF_ADDRESS: "e7:89:43:99:99:99",
}
USER_INPUT = {
CONF_NAME: "test-name",
CONF_PASSWORD: "test-password",
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
}
USER_INPUT_CURTAIN = {
CONF_NAME: "test-name",
CONF_PASSWORD: "test-password",
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
}
USER_INPUT_SENSOR = {
CONF_NAME: "test-name",
CONF_PASSWORD: "test-password",
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
}
USER_INPUT_UNSUPPORTED_DEVICE = {
CONF_NAME: "test-name",
CONF_PASSWORD: "test-password",
CONF_ADDRESS: "test",
}
USER_INPUT_INVALID = {
CONF_NAME: "test-name",
CONF_PASSWORD: "test-password",
CONF_ADDRESS: "invalid-mac",
}
@ -90,6 +70,42 @@ WOHAND_SERVICE_INFO = BluetoothServiceInfoBleak(
),
device=BLEDevice("aa:bb:cc:dd:ee:ff", "WoHand"),
)
WOHAND_ENCRYPTED_SERVICE_INFO = BluetoothServiceInfoBleak(
name="WoHand",
manufacturer_data={89: b"\xd8.\xad\xcd\r\x85"},
service_data={"00000d00-0000-1000-8000-00805f9b34fb": b"\xc8\x10\xcf"},
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
address="798A8547-2A3D-C609-55FF-73FA824B923B",
rssi=-60,
source="local",
advertisement=AdvertisementData(
local_name="WoHand",
manufacturer_data={89: b"\xd8.\xad\xcd\r\x85"},
service_data={"00000d00-0000-1000-8000-00805f9b34fb": b"\xc8\x10\xcf"},
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
),
device=BLEDevice("798A8547-2A3D-C609-55FF-73FA824B923B", "WoHand"),
)
WOHAND_SERVICE_ALT_ADDRESS_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="cc:cc:cc:cc:cc:cc",
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",

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

@ -10,9 +10,9 @@ from homeassistant.data_entry_flow import FlowResultType
from . import (
NOT_SWITCHBOT_INFO,
USER_INPUT,
USER_INPUT_CURTAIN,
USER_INPUT_SENSOR,
WOCURTAIN_SERVICE_INFO,
WOHAND_ENCRYPTED_SERVICE_INFO,
WOHAND_SERVICE_ALT_ADDRESS_INFO,
WOHAND_SERVICE_INFO,
WOSENSORTH_SERVICE_INFO,
init_integration,
@ -32,27 +32,53 @@ async def test_bluetooth_discovery(hass):
data=WOHAND_SERVICE_INFO,
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
assert result["step_id"] == "confirm"
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["title"] == "Bot EEFF"
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_requires_password(hass):
"""Test discovery via bluetooth with a valid device that needs a password."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_BLUETOOTH},
data=WOHAND_ENCRYPTED_SERVICE_INFO,
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "password"
with patch_async_setup_entry() as mock_setup_entry:
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_PASSWORD: "abc123"},
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "Bot 923B"
assert result["data"] == {
CONF_ADDRESS: "798A8547-2A3D-C609-55FF-73FA824B923B",
CONF_SENSOR_TYPE: "bot",
CONF_PASSWORD: "abc123",
}
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(
@ -97,22 +123,20 @@ async def test_user_setup_wohand(hass):
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {}
assert result["step_id"] == "confirm"
assert result["errors"] is None
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["title"] == "Bot EEFF"
assert result["data"] == {
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
CONF_NAME: "test-name",
CONF_PASSWORD: "test-password",
CONF_SENSOR_TYPE: "bot",
}
@ -154,28 +178,129 @@ async def test_user_setup_wocurtain(hass):
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "confirm"
assert result["errors"] is None
with patch_async_setup_entry() as mock_setup_entry:
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{},
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "Curtain EEFF"
assert result["data"] == {
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
CONF_SENSOR_TYPE: "curtain",
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_user_setup_wocurtain_or_bot(hass):
"""Test the user initiated form with valid address."""
with patch(
"homeassistant.components.switchbot.config_flow.async_discovered_service_info",
return_value=[WOCURTAIN_SERVICE_INFO, WOHAND_SERVICE_ALT_ADDRESS_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"] == {}
with patch_async_setup_entry() as mock_setup_entry:
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
USER_INPUT_CURTAIN,
USER_INPUT,
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "test-name"
assert result["title"] == "Curtain EEFF"
assert result["data"] == {
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
CONF_NAME: "test-name",
CONF_PASSWORD: "test-password",
CONF_SENSOR_TYPE: "curtain",
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_user_setup_wocurtain_or_bot_with_password(hass):
"""Test the user initiated form and valid address and a bot with a password."""
with patch(
"homeassistant.components.switchbot.config_flow.async_discovered_service_info",
return_value=[WOCURTAIN_SERVICE_INFO, WOHAND_ENCRYPTED_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"] == {}
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_ADDRESS: "798A8547-2A3D-C609-55FF-73FA824B923B"},
)
assert result2["type"] == FlowResultType.FORM
assert result2["step_id"] == "password"
assert result2["errors"] is None
with patch_async_setup_entry() as mock_setup_entry:
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
{CONF_PASSWORD: "abc123"},
)
await hass.async_block_till_done()
assert result3["type"] == FlowResultType.CREATE_ENTRY
assert result3["title"] == "Bot 923B"
assert result3["data"] == {
CONF_ADDRESS: "798A8547-2A3D-C609-55FF-73FA824B923B",
CONF_PASSWORD: "abc123",
CONF_SENSOR_TYPE: "bot",
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_user_setup_single_bot_with_password(hass):
"""Test the user initiated form for a bot with a password."""
with patch(
"homeassistant.components.switchbot.config_flow.async_discovered_service_info",
return_value=[WOHAND_ENCRYPTED_SERVICE_INFO],
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "password"
assert result["errors"] is None
with patch_async_setup_entry() as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_PASSWORD: "abc123"},
)
await hass.async_block_till_done()
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "Bot 923B"
assert result2["data"] == {
CONF_ADDRESS: "798A8547-2A3D-C609-55FF-73FA824B923B",
CONF_PASSWORD: "abc123",
CONF_SENSOR_TYPE: "bot",
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_user_setup_wosensor(hass):
"""Test the user initiated form with password and valid mac."""
with patch(
@ -186,22 +311,20 @@ async def test_user_setup_wosensor(hass):
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {}
assert result["step_id"] == "confirm"
assert result["errors"] is None
with patch_async_setup_entry() as mock_setup_entry:
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
USER_INPUT_SENSOR,
{},
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "test-name"
assert result["title"] == "Meter EEFF"
assert result["data"] == {
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
CONF_NAME: "test-name",
CONF_PASSWORD: "test-password",
CONF_SENSOR_TYPE: "hygrometer",
}
@ -229,7 +352,7 @@ async def test_async_step_user_takes_precedence_over_discovery(hass):
data=WOCURTAIN_SERVICE_INFO,
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
assert result["step_id"] == "confirm"
with patch(
"homeassistant.components.switchbot.config_flow.async_discovered_service_info",
@ -244,15 +367,13 @@ async def test_async_step_user_takes_precedence_over_discovery(hass):
with patch_async_setup_entry() as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=USER_INPUT,
user_input={},
)
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "test-name"
assert result2["title"] == "Curtain EEFF"
assert result2["data"] == {
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
CONF_NAME: "test-name",
CONF_PASSWORD: "test-password",
CONF_SENSOR_TYPE: "curtain",
}