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

Add config flow to Switchbot (#50653) 

Co-authored-by: Daniel Hjelseth Høyer <mail@dahoiv.net>
Co-authored-by: J. Nick Koston <nick@koston.org>
pull/56406/head
RenierM26 2021-09-18 21:25:05 +02:00 committed by GitHub
parent f31b9eae61
commit 3ce8109e5e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
16 changed files with 978 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
.coveragerc
View File

@ -1015,6 +1015,9 @@ omit =
homeassistant/components/swiss_public_transport/sensor.py
homeassistant/components/swisscom/device_tracker.py
homeassistant/components/switchbot/switch.py
homeassistant/components/switchbot/__init__.py
homeassistant/components/switchbot/const.py
homeassistant/components/switchbot/coordinator.py
homeassistant/components/switchmate/switch.py
homeassistant/components/syncthing/__init__.py
homeassistant/components/syncthing/sensor.py

2
CODEOWNERS
View File

@ -503,7 +503,7 @@ homeassistant/components/supla/* @mwegrzynek
homeassistant/components/surepetcare/* @benleb @danielhiversen
homeassistant/components/swiss_hydrological_data/* @fabaff
homeassistant/components/swiss_public_transport/* @fabaff
homeassistant/components/switchbot/* @danielhiversen
homeassistant/components/switchbot/* @danielhiversen @RenierM26
homeassistant/components/switcher_kis/* @tomerfi @thecode
homeassistant/components/switchmate/* @danielhiversen
homeassistant/components/syncthing/* @zhulik

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

@ -1 +1,111 @@
"""The switchbot component."""
"""Support for Switchbot devices."""
from asyncio import Lock
import switchbot # pylint: disable=import-error
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from .const import (
BTLE_LOCK,
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
PLATFORMS = ["switch"]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Switchbot from a config entry."""
hass.data.setdefault(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]:
# Check if asyncio.lock is stored in hass data.
# BTLE has issues with multiple connections,
# so we use a lock to ensure that only one API request is reaching it at a time:
if BTLE_LOCK not in hass.data[DOMAIN]:
hass.data[DOMAIN][BTLE_LOCK] = Lock()
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],
api_lock=hass.data[DOMAIN][BTLE_LOCK],
)
hass.data[DOMAIN][DATA_COORDINATOR] = coordinator
else:
coordinator = hass.data[DOMAIN][DATA_COORDINATOR]
await coordinator.async_config_entry_first_refresh()
if not coordinator.last_update_success:
raise ConfigEntryNotReady
entry.async_on_unload(entry.add_update_listener(_async_update_listener))
hass.data[DOMAIN][entry.entry_id] = {DATA_COORDINATOR: coordinator}
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if unload_ok:
hass.data[DOMAIN].pop(entry.entry_id)
if len(hass.config_entries.async_entries(DOMAIN)) == 0:
hass.data.pop(DOMAIN)
return unload_ok
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)

190
homeassistant/components/switchbot/config_flow.py Normal file
View File

@ -0,0 +1,190 @@
"""Config flow for Switchbot."""
from __future__ import annotations
from asyncio import Lock
import logging
from typing import Any
from switchbot import GetSwitchbotDevices # pylint: disable=import-error
import voluptuous as vol
from homeassistant.config_entries import ConfigEntry, ConfigFlow, OptionsFlow
from homeassistant.const import CONF_MAC, CONF_NAME, CONF_PASSWORD, CONF_SENSOR_TYPE
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
from .const import (
ATTR_BOT,
BTLE_LOCK,
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,
)
_LOGGER = logging.getLogger(__name__)
def _btle_connect(mac: str) -> dict:
"""Scan for BTLE advertisement data."""
# Try to find switchbot mac in nearby devices,
# by scanning for btle devices.
switchbots = GetSwitchbotDevices()
switchbots.discover()
switchbot_device = switchbots.get_device_data(mac=mac)
if not switchbot_device:
raise NotConnectedError("Failed to discover switchbot")
return switchbot_device
class SwitchbotConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Switchbot."""
VERSION = 1
async def _validate_mac(self, data: dict) -> FlowResult:
"""Try to connect to Switchbot device and create entry if successful."""
await self.async_set_unique_id(data[CONF_MAC].replace(":", ""))
self._abort_if_unique_id_configured()
# 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, {})
if BTLE_LOCK not in self.hass.data[DOMAIN]:
self.hass.data[DOMAIN][BTLE_LOCK] = Lock()
connect_lock = self.hass.data[DOMAIN][BTLE_LOCK]
# Validate bluetooth device mac.
async with connect_lock:
_btle_adv_data = await self.hass.async_add_executor_job(
_btle_connect, data[CONF_MAC]
)
if _btle_adv_data["modelName"] == "WoHand":
data[CONF_SENSOR_TYPE] = ATTR_BOT
return self.async_create_entry(title=data[CONF_NAME], data=data)
return self.async_abort(reason="switchbot_unsupported_type")
@staticmethod
@callback
def async_get_options_flow(
config_entry: ConfigEntry,
) -> SwitchbotOptionsFlowHandler:
"""Get the options flow for this handler."""
return SwitchbotOptionsFlowHandler(config_entry)
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle a flow initiated by the user."""
errors = {}
if user_input is not None:
user_input[CONF_MAC] = user_input[CONF_MAC].replace("-", ":").lower()
# abort if already configured.
for item in self._async_current_entries():
if item.data.get(CONF_MAC) == user_input[CONF_MAC]:
return self.async_abort(reason="already_configured_device")
try:
return await self._validate_mac(user_input)
except NotConnectedError:
errors["base"] = "cannot_connect"
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unexpected exception")
return self.async_abort(reason="unknown")
data_schema = vol.Schema(
{
vol.Required(CONF_NAME): str,
vol.Optional(CONF_PASSWORD): str,
vol.Required(CONF_MAC): str,
}
)
return self.async_show_form(
step_id="user", data_schema=data_schema, errors=errors
)
async def async_step_import(self, import_config: dict[str, Any]) -> FlowResult:
"""Handle config import from yaml."""
_LOGGER.debug("import config: %s", import_config)
import_config[CONF_MAC] = import_config[CONF_MAC].replace("-", ":").lower()
await self.async_set_unique_id(import_config[CONF_MAC].replace(":", ""))
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=import_config[CONF_NAME], data=import_config
)
class SwitchbotOptionsFlowHandler(OptionsFlow):
"""Handle Switchbot options."""
def __init__(self, config_entry: ConfigEntry) -> None:
"""Initialize options flow."""
self.config_entry = config_entry
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Manage Switchbot options."""
if user_input is not None:
# Update common entity options for all other entities.
for entry in self.hass.config_entries.async_entries(DOMAIN):
if entry.unique_id != self.config_entry.unique_id:
self.hass.config_entries.async_update_entry(
entry, options=user_input
)
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(
CONF_RETRY_COUNT, DEFAULT_RETRY_COUNT
),
): int,
vol.Optional(
CONF_RETRY_TIMEOUT,
default=self.config_entry.options.get(
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."""

24
homeassistant/components/switchbot/const.py Normal file
View File

@ -0,0 +1,24 @@
"""Constants for the switchbot integration."""
DOMAIN = "switchbot"
MANUFACTURER = "switchbot"
# Config Attributes
ATTR_BOT = "bot"
DEFAULT_NAME = "Switchbot"
# 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"
# Data
DATA_COORDINATOR = "coordinator"
BTLE_LOCK = "btle_lock"
COMMON_OPTIONS = "common_options"

59
homeassistant/components/switchbot/coordinator.py Normal file
View File

@ -0,0 +1,59 @@
"""Provides the switchbot DataUpdateCoordinator."""
from __future__ import annotations
from asyncio import Lock
from datetime import timedelta
import logging
import switchbot # pylint: disable=import-error
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
class SwitchbotDataUpdateCoordinator(DataUpdateCoordinator):
"""Class to manage fetching switchbot data."""
def __init__(
self,
hass: HomeAssistant,
*,
update_interval: int,
api: switchbot,
retry_count: int,
scan_timeout: int,
api_lock: Lock,
) -> None:
"""Initialize global switchbot data updater."""
self.switchbot_api = api
self.retry_count = retry_count
self.scan_timeout = scan_timeout
self.update_interval = timedelta(seconds=update_interval)
super().__init__(
hass, _LOGGER, name=DOMAIN, update_interval=self.update_interval
)
self.api_lock = api_lock
def _update_data(self) -> dict | None:
"""Fetch device states from switchbot api."""
return self.switchbot_api.GetSwitchbotDevices().discover(
retry=self.retry_count, scan_timeout=self.scan_timeout
)
async def _async_update_data(self) -> dict | None:
"""Fetch data from switchbot."""
async with self.api_lock:
switchbot_data = await self.hass.async_add_executor_job(self._update_data)
if not switchbot_data:
raise UpdateFailed("Unable to fetch switchbot services data")
return switchbot_data

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

@ -2,7 +2,8 @@
"domain": "switchbot",
"name": "SwitchBot",
"documentation": "https://www.home-assistant.io/integrations/switchbot",
"requirements": ["PySwitchbot==0.8.0"],
"codeowners": ["@danielhiversen"],
"requirements": ["PySwitchbot==0.11.0"],
"config_flow": true,
"codeowners": ["@danielhiversen", "@RenierM26"],
"iot_class": "local_polling"
}

35
homeassistant/components/switchbot/strings.json Normal file
View File

@ -0,0 +1,35 @@
{
"config": {
"flow_title": "{name}",
"step": {
"user": {
"title": "Setup Switchbot device",
"data": {
"name": "[%key:common::config_flow::data::name%]",
"password": "[%key:common::config_flow::data::password%]",
"mac": "Device MAC address"
}
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]"
},
"abort": {
"already_configured_device": "[%key:common::config_flow::abort::already_configured_device%]",
"unknown": "[%key:common::config_flow::error::unknown%]",
"switchbot_unsupported_type": "Unsupported Switchbot Type."
}
},
"options": {
"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"
}
}
}
}
}

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

@ -1,18 +1,48 @@
"""Support for Switchbot."""
"""Support for Switchbot bot."""
from __future__ import annotations
import logging
from typing import Any
# pylint: disable=import-error
import switchbot
import voluptuous as vol
from homeassistant.components.switch import PLATFORM_SCHEMA, SwitchEntity
from homeassistant.const import CONF_MAC, CONF_NAME, CONF_PASSWORD
import homeassistant.helpers.config_validation as cv
from homeassistant.components.switch import (
DEVICE_CLASS_SWITCH,
PLATFORM_SCHEMA,
SwitchEntity,
)
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import (
CONF_MAC,
CONF_NAME,
CONF_PASSWORD,
CONF_SENSOR_TYPE,
STATE_ON,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import (
config_validation as cv,
device_registry as dr,
entity_platform,
)
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.restore_state import RestoreEntity
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.helpers.update_coordinator import CoordinatorEntity
DEFAULT_NAME = "Switchbot"
from .const import (
ATTR_BOT,
CONF_RETRY_COUNT,
DATA_COORDINATOR,
DEFAULT_NAME,
DOMAIN,
MANUFACTURER,
)
from .coordinator import SwitchbotDataUpdateCoordinator
# Initialize the logger
_LOGGER = logging.getLogger(__name__)
PARALLEL_UPDATES = 1
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
@ -23,46 +53,120 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
)
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Perform the setup for Switchbot devices."""
name = config.get(CONF_NAME)
mac_addr = config[CONF_MAC]
password = config.get(CONF_PASSWORD)
add_entities([SwitchBot(mac_addr, name, password)])
async def async_setup_platform(
hass: HomeAssistant,
config: ConfigType,
async_add_entities: entity_platform.AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Import yaml config and initiates config flow for Switchbot devices."""
# Check if entry config exists and skips import if it does.
if hass.config_entries.async_entries(DOMAIN):
return
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data={
CONF_NAME: config[CONF_NAME],
CONF_PASSWORD: config.get(CONF_PASSWORD, None),
CONF_MAC: config[CONF_MAC].replace("-", ":").lower(),
CONF_SENSOR_TYPE: ATTR_BOT,
},
)
)
class SwitchBot(SwitchEntity, RestoreEntity):
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
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 entry.data[CONF_SENSOR_TYPE] != ATTR_BOT:
return
async_add_entities(
[
SwitchBot(
coordinator,
entry.unique_id,
entry.data[CONF_MAC],
entry.data[CONF_NAME],
entry.data.get(CONF_PASSWORD, None),
entry.options[CONF_RETRY_COUNT],
)
]
)
class SwitchBot(CoordinatorEntity, SwitchEntity, RestoreEntity):
"""Representation of a Switchbot."""
def __init__(self, mac, name, password) -> None:
coordinator: SwitchbotDataUpdateCoordinator
_attr_device_class = DEVICE_CLASS_SWITCH
def __init__(
self,
coordinator: SwitchbotDataUpdateCoordinator,
idx: str | None,
mac: str,
name: str,
password: str,
retry_count: int,
) -> None:
"""Initialize the Switchbot."""
self._state: bool | None = None
super().__init__(coordinator)
self._idx = idx
self._last_run_success: bool | None = None
self._name = name
self._mac = mac
self._device = switchbot.Switchbot(mac=mac, password=password)
self._device = self.coordinator.switchbot_api.Switchbot(
mac=mac, password=password, retry_count=retry_count
)
self._attr_unique_id = self._mac.replace(":", "")
self._attr_name = name
self._attr_device_info: DeviceInfo = {
"connections": {(dr.CONNECTION_NETWORK_MAC, self._mac)},
"name": name,
"model": self.coordinator.data[self._idx]["modelName"],
"manufacturer": MANUFACTURER,
}
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Run when entity about to be added."""
await super().async_added_to_hass()
state = await self.async_get_last_state()
if not state:
last_state = await self.async_get_last_state()
if not last_state:
return
self._state = state.state == "on"
self._attr_is_on = last_state.state == STATE_ON
self._last_run_success = last_state.attributes["last_run_success"]
def turn_on(self, **kwargs) -> None:
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn device on."""
if self._device.turn_on():
self._state = True
_LOGGER.info("Turn Switchbot bot on %s", self._mac)
async with self.coordinator.api_lock:
update_ok = await self.hass.async_add_executor_job(self._device.turn_on)
if update_ok:
self._last_run_success = True
else:
self._last_run_success = False
def turn_off(self, **kwargs) -> None:
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn device off."""
if self._device.turn_off():
self._state = False
_LOGGER.info("Turn Switchbot bot off %s", self._mac)
async with self.coordinator.api_lock:
update_ok = await self.hass.async_add_executor_job(self._device.turn_off)
if update_ok:
self._last_run_success = True
else:
self._last_run_success = False
@ -70,24 +174,20 @@ class SwitchBot(SwitchEntity, RestoreEntity):
@property
def assumed_state(self) -> bool:
"""Return true if unable to access real state of entity."""
return True
if not self.coordinator.data[self._idx]["data"]["switchMode"]:
return True
return False
@property
def is_on(self) -> bool:
"""Return true if device is on."""
return bool(self._state)
return self.coordinator.data[self._idx]["data"]["isOn"]
@property
def unique_id(self) -> str:
"""Return a unique, Home Assistant friendly identifier for this entity."""
return self._mac.replace(":", "")
@property
def name(self) -> str:
"""Return the name of the switch."""
return self._name
@property
def extra_state_attributes(self) -> dict[str, Any]:
def device_state_attributes(self) -> dict:
"""Return the state attributes."""
return {"last_run_success": self._last_run_success}
return {
"last_run_success": self._last_run_success,
"mac_address": self._mac,
"switch_mode": self.coordinator.data[self._idx]["data"]["switchMode"],
}

36
homeassistant/components/switchbot/translations/en.json Normal file
View File

@ -0,0 +1,36 @@
{
"config": {
"abort": {
"already_configured_device": "Device is already configured",
"unknown": "Unexpected error",
"switchbot_unsupported_type": "Unsupported Switchbot Type."
},
"error": {
"cannot_connect": "Failed to connect"
},
"flow_title": "{name}",
"step": {
"user": {
"data": {
"name": "Name",
"password": "Password",
"mac": "Mac"
},
"title": "Setup Switchbot device"
}
}
},
"options": {
"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"
}
}
}
}
}

1
homeassistant/generated/config_flows.py
View File

@ -268,6 +268,7 @@ FLOWS = [
"starline",
"subaru",
"surepetcare",
"switchbot",
"switcher_kis",
"syncthing",
"syncthru",

2
requirements_all.txt
View File

@ -49,7 +49,7 @@ PyRMVtransport==0.3.2
PySocks==1.7.1
# homeassistant.components.switchbot
# PySwitchbot==0.8.0
# PySwitchbot==0.11.0
# homeassistant.components.transport_nsw
PyTransportNSW==0.1.1

3
requirements_test_all.txt
View File

@ -23,6 +23,9 @@ PyQRCode==1.2.1
# homeassistant.components.rmvtransport
PyRMVtransport==0.3.2
# homeassistant.components.switchbot
# PySwitchbot==0.11.0
# homeassistant.components.transport_nsw
PyTransportNSW==0.1.1

64
tests/components/switchbot/__init__.py Normal file
View File

@ -0,0 +1,64 @@
"""Tests for the switchbot integration."""
from unittest.mock import patch
from homeassistant.const import CONF_MAC, CONF_NAME, CONF_PASSWORD, CONF_SENSOR_TYPE
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
DOMAIN = "switchbot"
ENTRY_CONFIG = {
CONF_NAME: "test-name",
CONF_PASSWORD: "test-password",
CONF_MAC: "e7:89:43:99:99:99",
}
USER_INPUT = {
CONF_NAME: "test-name",
CONF_PASSWORD: "test-password",
CONF_MAC: "e7:89:43:99:99:99",
}
USER_INPUT_UNSUPPORTED_DEVICE = {
CONF_NAME: "test-name",
CONF_PASSWORD: "test-password",
CONF_MAC: "test",
}
USER_INPUT_INVALID = {
CONF_NAME: "test-name",
CONF_PASSWORD: "test-password",
CONF_MAC: "invalid-mac",
}
YAML_CONFIG = {
CONF_NAME: "test-name",
CONF_PASSWORD: "test-password",
CONF_MAC: "e7:89:43:99:99:99",
CONF_SENSOR_TYPE: "bot",
}
def _patch_async_setup_entry(return_value=True):
return patch(
"homeassistant.components.switchbot.async_setup_entry",
return_value=return_value,
)
async def init_integration(
hass: HomeAssistant,
*,
data: dict = ENTRY_CONFIG,
skip_entry_setup: bool = False,
) -> MockConfigEntry:
"""Set up the Switchbot integration in Home Assistant."""
entry = MockConfigEntry(domain=DOMAIN, data=data)
entry.add_to_hass(hass)
if not skip_entry_setup:
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
return entry

78
tests/components/switchbot/conftest.py Normal file
View File

@ -0,0 +1,78 @@
"""Define fixtures available for all tests."""
import sys
from unittest.mock import MagicMock, patch
from pytest import fixture
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 = {
"mac_address": "e7:89:43:99:99:99",
"Flags": "06",
"Manufacturer": "5900e78943d9fe7c",
"Complete 128b Services": "cba20d00-224d-11e6-9fb8-0002a5d5c51b",
"data": {
"switchMode": "true",
"isOn": "true",
"battery": 91,
"rssi": -71,
},
"model": "H",
"modelName": "WoHand",
}
self._unsupported_device = {
"mac_address": "test",
"Flags": "06",
"Manufacturer": "5900e78943d9fe7c",
"Complete 128b Services": "cba20d00-224d-11e6-9fb8-0002a5d5c51b",
"data": {
"switchMode": "true",
"isOn": "true",
"battery": 91,
"rssi": -71,
},
"model": "HoN",
"modelName": "WoOther",
}
def discover(self, retry=0, scan_timeout=0):
"""Mock discover."""
return self._all_services_data
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
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)
instance.get_device_data = MagicMock(return_value=True)
yield mock_switchbot

226
tests/components/switchbot/test_config_flow.py Normal file
View File

@ -0,0 +1,226 @@
"""Test the switchbot config flow."""
from unittest.mock import patch
from homeassistant.components.switchbot.config_flow import NotConnectedError
from homeassistant.config_entries import SOURCE_IMPORT, SOURCE_USER
from homeassistant.const import CONF_MAC, CONF_NAME, CONF_PASSWORD, CONF_SENSOR_TYPE
from homeassistant.data_entry_flow import (
RESULT_TYPE_ABORT,
RESULT_TYPE_CREATE_ENTRY,
RESULT_TYPE_FORM,
)
from homeassistant.setup import async_setup_component
from . import (
USER_INPUT,
USER_INPUT_INVALID,
USER_INPUT_UNSUPPORTED_DEVICE,
YAML_CONFIG,
_patch_async_setup_entry,
init_integration,
)
DOMAIN = "switchbot"
async def test_user_form_valid_mac(hass):
"""Test the user initiated form with password and valid mac."""
await async_setup_component(hass, "persistent_notification", {})
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] == RESULT_TYPE_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,
)
await hass.async_block_till_done()
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["title"] == "test-name"
assert result["data"] == {
CONF_MAC: "e7:89:43:99:99:99",
CONF_NAME: "test-name",
CONF_PASSWORD: "test-password",
CONF_SENSOR_TYPE: "bot",
}
assert len(mock_setup_entry.mock_calls) == 1
# test duplicate device creation fails.
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "user"
assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
USER_INPUT,
)
await hass.async_block_till_done()
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "already_configured_device"
async def test_user_form_unsupported_device(hass):
"""Test the user initiated form for unsupported device type."""
await async_setup_component(hass, "persistent_notification", {})
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "user"
assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
USER_INPUT_UNSUPPORTED_DEVICE,
)
await hass.async_block_till_done()
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "switchbot_unsupported_type"
async def test_user_form_invalid_device(hass):
"""Test the user initiated form for invalid device type."""
await async_setup_component(hass, "persistent_notification", {})
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "user"
assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
USER_INPUT_INVALID,
)
await hass.async_block_till_done()
assert result["type"] == RESULT_TYPE_FORM
assert result["errors"] == {"base": "cannot_connect"}
async def test_async_step_import(hass):
"""Test the config import flow."""
await async_setup_component(hass, "persistent_notification", {})
with _patch_async_setup_entry() as mock_setup_entry:
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=YAML_CONFIG
)
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["data"] == {
CONF_MAC: "e7:89:43:99:99:99",
CONF_NAME: "test-name",
CONF_PASSWORD: "test-password",
CONF_SENSOR_TYPE: "bot",
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_user_form_exception(hass, switchbot_config_flow):
"""Test we handle exception on user form."""
await async_setup_component(hass, "persistent_notification", {})
switchbot_config_flow.side_effect = NotConnectedError
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
USER_INPUT,
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "user"
assert result["errors"] == {"base": "cannot_connect"}
switchbot_config_flow.side_effect = Exception
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
USER_INPUT,
)
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "unknown"
async def test_options_flow(hass):
"""Test updating options."""
with patch("homeassistant.components.switchbot.PLATFORMS", []):
entry = await init_integration(hass)
assert entry.options["update_time"] == 60
assert entry.options["retry_count"] == 3
assert entry.options["retry_timeout"] == 5
assert entry.options["scan_timeout"] == 5
result = await hass.config_entries.options.async_init(entry.entry_id)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "init"
assert result["errors"] is None
with _patch_async_setup_entry() as mock_setup_entry:
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
"update_time": 60,
"retry_count": 3,
"retry_timeout": 5,
"scan_timeout": 5,
},
)
await hass.async_block_till_done()
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["data"]["update_time"] == 60
assert result["data"]["retry_count"] == 3
assert result["data"]["retry_timeout"] == 5
assert result["data"]["scan_timeout"] == 5
assert len(mock_setup_entry.mock_calls) == 0
# Test changing of entry options.
result = await hass.config_entries.options.async_init(entry.entry_id)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "init"
assert result["errors"] is None
with _patch_async_setup_entry() as mock_setup_entry:
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
"update_time": 60,
"retry_count": 3,
"retry_timeout": 5,
"scan_timeout": 5,
},
)
await hass.async_block_till_done()
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["data"]["update_time"] == 60
assert result["data"]["retry_count"] == 3
assert result["data"]["retry_timeout"] == 5
assert result["data"]["scan_timeout"] == 5
assert len(mock_setup_entry.mock_calls) == 0