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

Address late review of SwitchBee (#78412)

pull/78586/head
Jafar Atili 2022-09-16 15:19:50 +03:00 committed by GitHub
parent 383c83d15f
commit 491177e5d3
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 313 additions and 428 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.coveragerc
View File

@ -1215,6 +1215,7 @@ omit =
homeassistant/components/swisscom/device_tracker.py
homeassistant/components/switchbee/__init__.py
homeassistant/components/switchbee/const.py
homeassistant/components/switchbee/coordinator.py
homeassistant/components/switchbee/switch.py
homeassistant/components/switchbot/__init__.py
homeassistant/components/switchbot/binary_sensor.py

115
homeassistant/components/switchbee/__init__.py
View File

@ -2,29 +2,16 @@
from __future__ import annotations
from datetime import timedelta
import logging
from switchbee.api import CentralUnitAPI, SwitchBeeError
from switchbee.device import DeviceType
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.device_registry import format_mac
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import (
CONF_DEFUALT_ALLOWED,
CONF_DEVICES,
CONF_SWITCHES_AS_LIGHTS,
DOMAIN,
SCAN_INTERVAL_SEC,
)
_LOGGER = logging.getLogger(__name__)
from .const import DOMAIN
from .coordinator import SwitchBeeCoordinator
PLATFORMS: list[Platform] = [Platform.SWITCH]
@ -35,30 +22,24 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
central_unit = entry.data[CONF_HOST]
user = entry.data[CONF_USERNAME]
password = entry.data[CONF_PASSWORD]
devices_map: dict[str, DeviceType] = {s.display: s for s in DeviceType}
allowed_devices = [
devices_map[device]
for device in entry.options.get(CONF_DEVICES, CONF_DEFUALT_ALLOWED)
]
websession = async_get_clientsession(hass, verify_ssl=False)
api = CentralUnitAPI(central_unit, user, password, websession)
try:
await api.connect()
except SwitchBeeError:
return False
except SwitchBeeError as exp:
raise ConfigEntryNotReady("Failed to connect to the Central Unit") from exp
coordinator = SwitchBeeCoordinator(
hass,
api,
SCAN_INTERVAL_SEC,
allowed_devices,
entry.data[CONF_SWITCHES_AS_LIGHTS],
)
await coordinator.async_config_entry_first_refresh()
entry.async_on_unload(entry.add_update_listener(update_listener))
hass.data[DOMAIN][entry.entry_id] = coordinator
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
@ -74,83 +55,3 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def update_listener(hass: HomeAssistant, config_entry: ConfigEntry) -> None:
"""Update listener."""
await hass.config_entries.async_reload(config_entry.entry_id)
class SwitchBeeCoordinator(DataUpdateCoordinator):
"""Class to manage fetching Freedompro data API."""
def __init__(
self,
hass,
swb_api,
scan_interval,
devices: list[DeviceType],
switch_as_light: bool,
):
"""Initialize."""
self._api: CentralUnitAPI = swb_api
self._reconnect_counts: int = 0
self._devices_to_include: list[DeviceType] = devices
self._prev_devices_to_include_to_include: list[DeviceType] = []
self._mac_addr_fmt: str = format_mac(swb_api.mac)
self._switch_as_light = switch_as_light
super().__init__(
hass,
_LOGGER,
name=DOMAIN,
update_interval=timedelta(seconds=scan_interval),
)
@property
def api(self) -> CentralUnitAPI:
"""Return SwitchBee API object."""
return self._api
@property
def mac_formated(self) -> str:
"""Return formatted MAC address."""
return self._mac_addr_fmt
@property
def switch_as_light(self) -> bool:
"""Return switch_as_ligh config."""
return self._switch_as_light
async def _async_update_data(self):
if self._reconnect_counts != self._api.reconnect_count:
self._reconnect_counts = self._api.reconnect_count
_LOGGER.debug(
"Central Unit re-connected again due to invalid token, total %i",
self._reconnect_counts,
)
config_changed = False
if set(self._prev_devices_to_include_to_include) != set(
self._devices_to_include
):
self._prev_devices_to_include_to_include = self._devices_to_include
config_changed = True
# The devices are loaded once during the config_entry
if not self._api.devices or config_changed:
# Try to load the devices from the CU for the first time
try:
await self._api.fetch_configuration(self._devices_to_include)
except SwitchBeeError as exp:
raise UpdateFailed(
f"Error communicating with API: {exp}"
) from SwitchBeeError
else:
_LOGGER.debug("Loaded devices")
# Get the state of the devices
try:
await self._api.fetch_states()
except SwitchBeeError as exp:
raise UpdateFailed(
f"Error communicating with API: {exp}"
) from SwitchBeeError
else:
return self._api.devices

51
homeassistant/components/switchbee/config_flow.py
View File

@ -5,19 +5,18 @@ import logging
from typing import Any
from switchbee.api import CentralUnitAPI, SwitchBeeError
from switchbee.device import DeviceType
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant, callback
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResult
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.aiohttp_client import async_get_clientsession
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.device_registry import format_mac
from .const import CONF_DEFUALT_ALLOWED, CONF_DEVICES, CONF_SWITCHES_AS_LIGHTS, DOMAIN
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
@ -26,7 +25,6 @@ STEP_USER_DATA_SCHEMA = vol.Schema(
vol.Required(CONF_HOST): cv.string,
vol.Required(CONF_USERNAME): cv.string,
vol.Required(CONF_PASSWORD): cv.string,
vol.Required(CONF_SWITCHES_AS_LIGHTS, default=False): cv.boolean,
}
)
@ -43,9 +41,9 @@ async def validate_input(hass: HomeAssistant, data: dict[str, Any]):
except SwitchBeeError as exp:
_LOGGER.error(exp)
if "LOGIN_FAILED" in str(exp):
raise InvalidAuth from SwitchBeeError
raise InvalidAuth from exp
raise CannotConnect from SwitchBeeError
raise CannotConnect from exp
return format_mac(api.mac)
@ -83,47 +81,6 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
)
@staticmethod
@callback
def async_get_options_flow(
config_entry: config_entries.ConfigEntry,
) -> OptionsFlowHandler:
"""Get the options flow for this handler."""
return OptionsFlowHandler(config_entry)
class OptionsFlowHandler(config_entries.OptionsFlow):
"""Handle a option flow for AEMET."""
def __init__(self, config_entry: config_entries.ConfigEntry) -> None:
"""Initialize options flow."""
self.config_entry = config_entry
async def async_step_init(self, user_input=None) -> FlowResult:
"""Handle options flow."""
if user_input is not None:
return self.async_create_entry(title="", data=user_input)
all_devices = [
DeviceType.Switch,
DeviceType.TimedSwitch,
DeviceType.GroupSwitch,
DeviceType.TimedPowerSwitch,
]
data_schema = {
vol.Required(
CONF_DEVICES,
default=self.config_entry.options.get(
CONF_DEVICES,
CONF_DEFUALT_ALLOWED,
),
): cv.multi_select([device.display for device in all_devices]),
}
return self.async_show_form(step_id="init", data_schema=vol.Schema(data_schema))
class CannotConnect(HomeAssistantError):
"""Error to indicate we cannot connect."""

10
homeassistant/components/switchbee/const.py
View File

@ -1,14 +1,4 @@
"""Constants for the SwitchBee Smart Home integration."""
from switchbee.device import DeviceType
DOMAIN = "switchbee"
SCAN_INTERVAL_SEC = 5
CONF_SCAN_INTERVAL = "scan_interval"
CONF_SWITCHES_AS_LIGHTS = "switch_as_light"
CONF_DEVICES = "devices"
CONF_DEFUALT_ALLOWED = [
DeviceType.Switch.display,
DeviceType.TimedPowerSwitch.display,
DeviceType.TimedSwitch.display,
]

74
homeassistant/components/switchbee/coordinator.py Normal file
View File

@ -0,0 +1,74 @@
"""SwitchBee integration Coordinator."""
from datetime import timedelta
import logging
from switchbee.api import CentralUnitAPI, SwitchBeeError
from switchbee.device import DeviceType, SwitchBeeBaseDevice
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import format_mac
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN, SCAN_INTERVAL_SEC
_LOGGER = logging.getLogger(__name__)
class SwitchBeeCoordinator(DataUpdateCoordinator[dict[int, SwitchBeeBaseDevice]]):
"""Class to manage fetching Freedompro data API."""
def __init__(
self,
hass: HomeAssistant,
swb_api: CentralUnitAPI,
) -> None:
"""Initialize."""
self.api: CentralUnitAPI = swb_api
self._reconnect_counts: int = 0
self.mac_formated: str = format_mac(swb_api.mac)
super().__init__(
hass,
_LOGGER,
name=DOMAIN,
update_interval=timedelta(seconds=SCAN_INTERVAL_SEC),
)
async def _async_update_data(self) -> dict[int, SwitchBeeBaseDevice]:
"""Update data via library."""
if self._reconnect_counts != self.api.reconnect_count:
self._reconnect_counts = self.api.reconnect_count
_LOGGER.debug(
"Central Unit re-connected again due to invalid token, total %i",
self._reconnect_counts,
)
# The devices are loaded once during the config_entry
if not self.api.devices:
# Try to load the devices from the CU for the first time
try:
await self.api.fetch_configuration(
[
DeviceType.Switch,
DeviceType.TimedSwitch,
DeviceType.GroupSwitch,
DeviceType.TimedPowerSwitch,
]
)
except SwitchBeeError as exp:
raise UpdateFailed(
f"Error communicating with API: {exp}"
) from SwitchBeeError
else:
_LOGGER.debug("Loaded devices")
# Get the state of the devices
try:
await self.api.fetch_states()
except SwitchBeeError as exp:
raise UpdateFailed(
f"Error communicating with API: {exp}"
) from SwitchBeeError
return self.api.devices

2
homeassistant/components/switchbee/manifest.json
View File

@ -3,7 +3,7 @@
"name": "SwitchBee",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/switchbee",
"requirements": ["pyswitchbee==1.4.7"],
"requirements": ["pyswitchbee==1.4.8"],
"codeowners": ["@jafar-atili"],
"iot_class": "local_polling"
}

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

@ -6,8 +6,7 @@
"data": {
"host": "[%key:common::config_flow::data::host%]",
"username": "[%key:common::config_flow::data::username%]",
"password": "[%key:common::config_flow::data::password%]",
"switch_as_light": "Initialize switches as light entities"
"password": "[%key:common::config_flow::data::password%]"
}
}
},
@ -19,14 +18,5 @@
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
}
},
"options": {
"step": {
"init": {
"data": {
"devices": "Devices to include"
}
}
}
}
}

90
homeassistant/components/switchbee/switch.py
View File

@ -9,13 +9,13 @@ from switchbee.device import ApiStateCommand, DeviceType, SwitchBeeBaseDevice
from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import aiohttp_client
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import SwitchBeeCoordinator
from .const import DOMAIN
from .coordinator import SwitchBeeCoordinator
_LOGGER = logging.getLogger(__name__)
@ -25,37 +25,35 @@ async def async_setup_entry(
) -> None:
"""Set up Switchbee switch."""
coordinator: SwitchBeeCoordinator = hass.data[DOMAIN][entry.entry_id]
device_types = (
[DeviceType.TimedPowerSwitch]
if coordinator.switch_as_light
else [
async_add_entities(
SwitchBeeSwitchEntity(device, coordinator)
for device in coordinator.data.values()
if device.type
in [
DeviceType.TimedPowerSwitch,
DeviceType.GroupSwitch,
DeviceType.Switch,
DeviceType.TimedSwitch,
DeviceType.TwoWay,
]
)
async_add_entities(
Device(hass, device, coordinator)
for device in coordinator.data.values()
if device.type in device_types
)
class Device(CoordinatorEntity, SwitchEntity):
class SwitchBeeSwitchEntity(CoordinatorEntity[SwitchBeeCoordinator], SwitchEntity):
"""Representation of an Switchbee switch."""
def __init__(self, hass, device: SwitchBeeBaseDevice, coordinator):
def __init__(
self,
device: SwitchBeeBaseDevice,
coordinator: SwitchBeeCoordinator,
) -> None:
"""Initialize the Switchbee switch."""
super().__init__(coordinator)
self._session = aiohttp_client.async_get_clientsession(hass)
self._attr_name = f"{device.name}"
self._device_id = device.id
self._attr_unique_id = f"{coordinator.mac_formated}-{device.id}"
self._attr_is_on = False
self._attr_available = True
self._is_online = True
self._attr_has_entity_name = True
self._device = device
self._attr_device_info = DeviceInfo(
@ -75,11 +73,32 @@ class Device(CoordinatorEntity, SwitchEntity):
),
)
@property
def available(self) -> bool:
"""Return True if entity is available."""
return self._is_online and self.coordinator.last_update_success
@callback
def _handle_coordinator_update(self) -> None:
"""Handle updated data from the coordinator."""
self._update_from_coordinator()
super()._handle_coordinator_update()
def _update_from_coordinator(self) -> None:
"""Update the entity attributes from the coordinator data."""
async def async_refresh_state():
"""Refresh the device state in the Central Unit.
This function addresses issue of a device that came online back but still report
unavailable state (-1).
Such device (offline device) will keep reporting unavailable state (-1)
until it has been actuated by the user (state changed to on/off).
With this code we keep trying setting dummy state for the device
in order for it to start reporting its real state back (assuming it came back online)
"""
try:
await self.coordinator.api.set_state(self._device_id, "dummy")
@ -92,35 +111,30 @@ class Device(CoordinatorEntity, SwitchEntity):
# This specific call will refresh the state of the device in the CU
self.hass.async_create_task(async_refresh_state())
if self.available:
# if the device was online (now offline), log message and mark it as Unavailable
if self._is_online:
_LOGGER.error(
"%s switch is not responding, check the status in the SwitchBee mobile app",
self.name,
)
self._attr_available = False
self.async_write_ha_state()
return None
self._is_online = False
if not self.available:
return
# check if the device was offline (now online) and bring it back
if not self._is_online:
_LOGGER.info(
"%s switch is now responding",
self.name,
)
self._attr_available = True
self._is_online = True
# timed power switch state will represent a number of minutes until it goes off
# regulare switches state is ON/OFF
# timed power switch state is an integer representing the number of minutes left until it goes off
# regulare switches state is ON/OFF (1/0 respectively)
self._attr_is_on = (
self.coordinator.data[self._device_id].state != ApiStateCommand.OFF
)
super()._handle_coordinator_update()
async def async_added_to_hass(self) -> None:
"""When entity is added to hass."""
await super().async_added_to_hass()
self._handle_coordinator_update()
async def async_turn_on(self, **kwargs: Any) -> None:
"""Async function to set on to switch."""
return await self._async_set_state(ApiStateCommand.ON)
@ -129,13 +143,13 @@ class Device(CoordinatorEntity, SwitchEntity):
"""Async function to set off to switch."""
return await self._async_set_state(ApiStateCommand.OFF)
async def _async_set_state(self, state):
async def _async_set_state(self, state: ApiStateCommand) -> None:
try:
await self.coordinator.api.set_state(self._device_id, state)
except (SwitchBeeError, SwitchBeeDeviceOfflineError) as exp:
_LOGGER.error(
"Failed to set %s state %s, error: %s", self._attr_name, state, exp
)
self._async_write_ha_state()
else:
await self.coordinator.async_refresh()
raise HomeAssistantError(
f"Failed to set {self._attr_name} state {state}, {str(exp)}"
) from exp
await self.coordinator.async_refresh()

48
homeassistant/components/switchbee/translations/en.json
View File

@ -1,32 +1,22 @@
{
"config": {
"abort": {
"already_configured": "Device is already configured"
},
"error": {
"cannot_connect": "Failed to connect",
"invalid_auth": "Invalid authentication",
"unknown": "Unexpected error"
},
"step": {
"user": {
"data": {
"host": "Host",
"password": "Password",
"switch_as_light": "Initialize switches as light entities",
"username": "Username"
},
"description": "Setup SwitchBee integration with Home Assistant."
"config": {
"abort": {
"already_configured_device": "Device is already configured"
},
"error": {
"cannot_connect": "Failed to connect",
"invalid_auth": "Failed to Authenticate with the Central Unit",
"unknown": "Unexpected error"
},
"step": {
"user": {
"description": "Setup SwitchBee integration with Home Assistant.",
"data": {
"host": "Central Unit IP address",
"username": "User (e-mail)",
"password": "Password"
}
}
},
"options": {
"step": {
"init": {
"data": {
"devices": "Devices to include"
}
}
}
}
}
}
}
}

2
requirements_all.txt
View File

@ -1912,7 +1912,7 @@ pystiebeleltron==0.0.1.dev2
pysuez==0.1.19
# homeassistant.components.switchbee
pyswitchbee==1.4.7
pyswitchbee==1.4.8
# homeassistant.components.syncthru
pysyncthru==0.7.10

2
requirements_test_all.txt
View File

@ -1335,7 +1335,7 @@ pyspcwebgw==0.4.0
pysqueezebox==0.6.0
# homeassistant.components.switchbee
pyswitchbee==1.4.7
pyswitchbee==1.4.8
# homeassistant.components.syncthru
pysyncthru==0.7.10

135
tests/components/switchbee/__init__.py
View File

@ -1,140 +1,5 @@
"""Tests for the SwitchBee Smart Home integration."""
MOCK_GET_CONFIGURATION = {
"status": "OK",
"data": {
"mac": "A8-21-08-E7-67-B6",
"name": "Residence",
"version": "1.4.4(4)",
"lastConfChange": 1661856874511,
"zones": [
{
"name": "Sensor Setting",
"items": [
{
"id": 200000,
"name": "home",
"hw": "VIRTUAL",
"type": "ALARM_SYSTEM",
},
{
"id": 200010,
"name": "away",
"hw": "VIRTUAL",
"type": "ALARM_SYSTEM",
},
],
},
{
"name": "General",
"items": [
{
"operations": [113],
"id": 100080,
"name": "All Lights",
"hw": "VIRTUAL",
"type": "GROUP_SWITCH",
},
{
"operations": [
{"itemId": 21, "value": 100},
{"itemId": 333, "value": 100},
],
"id": 100160,
"name": "Sunrise",
"hw": "VIRTUAL",
"type": "SCENARIO",
},
],
},
{
"name": "Entrance",
"items": [
{
"id": 113,
"name": "Staircase Lights",
"hw": "DIMMABLE_SWITCH",
"type": "TIMED_SWITCH",
},
{
"id": 222,
"name": "Front Door",
"hw": "REGULAR_SWITCH",
"type": "TIMED_SWITCH",
},
],
},
{
"name": "Kitchen",
"items": [
{"id": 21, "name": "Shutter ", "hw": "SHUTTER", "type": "SHUTTER"},
{
"operations": [593, 581, 171],
"id": 481,
"name": "Leds",
"hw": "DIMMABLE_SWITCH",
"type": "GROUP_SWITCH",
},
{
"id": 12,
"name": "Walls",
"hw": "DIMMABLE_SWITCH",
"type": "DIMMER",
},
],
},
{
"name": "Two Way Zone",
"items": [
{
"operations": [113],
"id": 72,
"name": "Staircase Lights",
"hw": "DIMMABLE_SWITCH",
"type": "TWO_WAY",
}
],
},
{
"name": "Facilities ",
"items": [
{
"id": 321,
"name": "Boiler",
"hw": "TIMED_POWER_SWITCH",
"type": "TIMED_POWER",
},
{
"modes": ["COOL", "HEAT", "FAN"],
"temperatureUnits": "CELSIUS",
"id": 271,
"name": "HVAC",
"hw": "THERMOSTAT",
"type": "THERMOSTAT",
},
{
"id": 571,
"name": "Repeater",
"hw": "REPEATER",
"type": "REPEATER",
},
],
},
{
"name": "Alarm",
"items": [
{
"operations": [{"itemId": 113, "value": 100}],
"id": 81,
"name": "Open Home",
"hw": "STIKER_SWITCH",
"type": "SCENARIO",
}
],
},
],
},
}
MOCK_FAILED_TO_LOGIN_MSG = (
"Central Unit replied with failure: {'status': 'LOGIN_FAILED'}"
)

135
tests/components/switchbee/fixtures/switchbee.json Normal file
View File

@ -0,0 +1,135 @@
{
"status": "OK",
"data": {
"mac": "A8-21-08-E7-67-B6",
"name": "Residence",
"version": "1.4.4(4)",
"lastConfChange": 1661856874511,
"zones": [
{
"name": "Sensor Setting",
"items": [
{
"id": 200000,
"name": "home",
"hw": "VIRTUAL",
"type": "ALARM_SYSTEM"
},
{
"id": 200010,
"name": "away",
"hw": "VIRTUAL",
"type": "ALARM_SYSTEM"
}
]
},
{
"name": "General",
"items": [
{
"operations": [113],
"id": 100080,
"name": "All Lights",
"hw": "VIRTUAL",
"type": "GROUP_SWITCH"
},
{
"operations": [
{ "itemId": 21, "value": 100 },
{ "itemId": 333, "value": 100 }
],
"id": 100160,
"name": "Sunrise",
"hw": "VIRTUAL",
"type": "SCENARIO"
}
]
},
{
"name": "Entrance",
"items": [
{
"id": 113,
"name": "Staircase Lights",
"hw": "DIMMABLE_SWITCH",
"type": "TIMED_SWITCH"
},
{
"id": 222,
"name": "Front Door",
"hw": "REGULAR_SWITCH",
"type": "TIMED_SWITCH"
}
]
},
{
"name": "Kitchen",
"items": [
{ "id": 21, "name": "Shutter ", "hw": "SHUTTER", "type": "SHUTTER" },
{
"operations": [593, 581, 171],
"id": 481,
"name": "Leds",
"hw": "DIMMABLE_SWITCH",
"type": "GROUP_SWITCH"
},
{
"id": 12,
"name": "Walls",
"hw": "DIMMABLE_SWITCH",
"type": "DIMMER"
}
]
},
{
"name": "Two Way Zone",
"items": [
{
"operations": [113],
"id": 72,
"name": "Staircase Lights",
"hw": "DIMMABLE_SWITCH",
"type": "TWO_WAY"
}
]
},
{
"name": "Facilities ",
"items": [
{
"id": 321,
"name": "Boiler",
"hw": "TIMED_POWER_SWITCH",
"type": "TIMED_POWER"
},
{
"modes": ["COOL", "HEAT", "FAN"],
"temperatureUnits": "CELSIUS",
"id": 271,
"name": "HVAC",
"hw": "THERMOSTAT",
"type": "THERMOSTAT"
},
{
"id": 571,
"name": "Repeater",
"hw": "REPEATER",
"type": "REPEATER"
}
]
},
{
"name": "Alarm",
"items": [
{
"operations": [{ "itemId": 113, "value": 100 }],
"id": 81,
"name": "Open Home",
"hw": "STIKER_SWITCH",
"type": "SCENARIO"
}
]
}
]
}
}

64
tests/components/switchbee/test_config_flow.py
View File

@ -1,21 +1,23 @@
"""Test the SwitchBee Smart Home config flow."""
import json
from unittest.mock import patch
from homeassistant import config_entries
from homeassistant.components.switchbee.config_flow import DeviceType, SwitchBeeError
from homeassistant.components.switchbee.const import CONF_SWITCHES_AS_LIGHTS, DOMAIN
from homeassistant.const import CONF_DEVICES, CONF_HOST, CONF_PASSWORD, CONF_USERNAME
from homeassistant.components.switchbee.config_flow import SwitchBeeError
from homeassistant.components.switchbee.const import DOMAIN
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import RESULT_TYPE_FORM, FlowResultType
from homeassistant.data_entry_flow import FlowResultType
from . import MOCK_FAILED_TO_LOGIN_MSG, MOCK_GET_CONFIGURATION, MOCK_INVALID_TOKEN_MGS
from . import MOCK_FAILED_TO_LOGIN_MSG, MOCK_INVALID_TOKEN_MGS
from tests.common import MockConfigEntry
from tests.common import MockConfigEntry, load_fixture
async def test_form(hass):
"""Test we get the form."""
coordinator_data = json.loads(load_fixture("switchbee.json", "switchbee"))
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
@ -24,7 +26,7 @@ async def test_form(hass):
with patch(
"switchbee.api.CentralUnitAPI.get_configuration",
return_value=MOCK_GET_CONFIGURATION,
return_value=coordinator_data,
), patch(
"homeassistant.components.switchbee.async_setup_entry",
return_value=True,
@ -40,7 +42,6 @@ async def test_form(hass):
CONF_HOST: "1.1.1.1",
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
CONF_SWITCHES_AS_LIGHTS: False,
},
)
await hass.async_block_till_done()
@ -51,7 +52,6 @@ async def test_form(hass):
CONF_HOST: "1.1.1.1",
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
CONF_SWITCHES_AS_LIGHTS: False,
}
@ -71,16 +71,16 @@ async def test_form_invalid_auth(hass: HomeAssistant) -> None:
CONF_HOST: "1.1.1.1",
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
CONF_SWITCHES_AS_LIGHTS: False,
},
)
assert result2["type"] == RESULT_TYPE_FORM
assert result2["type"] == FlowResultType.FORM
assert result2["errors"] == {"base": "invalid_auth"}
async def test_form_cannot_connect(hass: HomeAssistant) -> None:
"""Test we handle cannot connect error."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
@ -95,11 +95,10 @@ async def test_form_cannot_connect(hass: HomeAssistant) -> None:
CONF_HOST: "1.1.1.1",
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
CONF_SWITCHES_AS_LIGHTS: False,
},
)
assert result2["type"] == RESULT_TYPE_FORM
assert result2["type"] == FlowResultType.FORM
assert result2["errors"] == {"base": "cannot_connect"}
@ -119,16 +118,17 @@ async def test_form_unknown_error(hass):
CONF_HOST: "1.1.1.1",
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
CONF_SWITCHES_AS_LIGHTS: False,
},
)
assert form_result["type"] == RESULT_TYPE_FORM
assert form_result["type"] == FlowResultType.FORM
assert form_result["errors"] == {"base": "unknown"}
async def test_form_entry_exists(hass):
"""Test we handle an already existing entry."""
coordinator_data = json.loads(load_fixture("switchbee.json", "switchbee"))
MockConfigEntry(
unique_id="a8:21:08:e7:67:b6",
domain=DOMAIN,
@ -136,7 +136,6 @@ async def test_form_entry_exists(hass):
CONF_HOST: "1.1.1.1",
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
CONF_SWITCHES_AS_LIGHTS: False,
},
title="1.1.1.1",
).add_to_hass(hass)
@ -150,7 +149,7 @@ async def test_form_entry_exists(hass):
return_value=True,
), patch(
"switchbee.api.CentralUnitAPI.get_configuration",
return_value=MOCK_GET_CONFIGURATION,
return_value=coordinator_data,
), patch(
"switchbee.api.CentralUnitAPI.fetch_states", return_value=None
):
@ -160,39 +159,8 @@ async def test_form_entry_exists(hass):
CONF_HOST: "1.2.2.2",
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
CONF_SWITCHES_AS_LIGHTS: False,
},
)
assert form_result["type"] == FlowResultType.ABORT
assert form_result["reason"] == "already_configured"
async def test_option_flow(hass):
"""Test config flow options."""
entry = MockConfigEntry(
unique_id="a8:21:08:e7:67:b6",
domain=DOMAIN,
data={
CONF_HOST: "1.1.1.1",
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
CONF_SWITCHES_AS_LIGHTS: False,
},
title="1.1.1.1",
)
entry.add_to_hass(hass)
result = await hass.config_entries.options.async_init(entry.entry_id)
assert result["type"] == "form"
assert result["step_id"] == "init"
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONF_DEVICES: [DeviceType.Switch.display, DeviceType.GroupSwitch.display],
},
)
assert result["type"] == "create_entry"
assert result["data"] == {
CONF_DEVICES: [DeviceType.Switch.display, DeviceType.GroupSwitch.display],
}