jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-23 00:37:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Discover Switchbot MAC in config flow (#56616)

Browse Source 
* Update config_flow.py

* Switchbot Config_flow discover mac instead of needing to type it.

* Do not show already configured devices in config flow, abort if no unconfigured devices.

* Apply suggestions from code review

Co-authored-by: J. Nick Koston <nick@koston.org>

* Move MAC to top of config flow form dict.

* Update homeassistant/components/switchbot/config_flow.py

Co-authored-by: J. Nick Koston <nick@koston.org>
This commit is contained in:
RenierM26 2021-09-27 22:12:40 +02:00 committed by GitHub
parent b40d229369
commit b15f11f46a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 90 additions and 125 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -30,19 +30,15 @@ from .const import (
_LOGGER = logging.getLogger(__name__)
def _btle_connect(mac: str) -> dict:
def _btle_connect() -> 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)
switchbot_devices = GetSwitchbotDevices().discover()
if not switchbot_device:
if not switchbot_devices:
raise NotConnectedError("Failed to discover switchbot")
return switchbot_device
return switchbot_devices
class SwitchbotConfigFlow(ConfigFlow, domain=DOMAIN):
@ -50,11 +46,8 @@ class SwitchbotConfigFlow(ConfigFlow, domain=DOMAIN):
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()
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:
@ -64,17 +57,11 @@ class SwitchbotConfigFlow(ConfigFlow, domain=DOMAIN):
connect_lock = self.hass.data[DOMAIN][BTLE_LOCK]
# Validate bluetooth device mac.
# Discover switchbots nearby.
async with connect_lock:
_btle_adv_data = await self.hass.async_add_executor_job(
_btle_connect, data[CONF_MAC]
)
_btle_adv_data = await self.hass.async_add_executor_job(_btle_connect)
if _btle_adv_data["modelName"] in SUPPORTED_MODEL_TYPES:
data[CONF_SENSOR_TYPE] = SUPPORTED_MODEL_TYPES[_btle_adv_data["modelName"]]
return self.async_create_entry(title=data[CONF_NAME], data=data)
return self.async_abort(reason="switchbot_unsupported_type")
return _btle_adv_data
@staticmethod
@callback
@ -84,36 +71,59 @@ class SwitchbotConfigFlow(ConfigFlow, domain=DOMAIN):
"""Get the options flow for this handler."""
return SwitchbotOptionsFlowHandler(config_entry)
def __init__(self):
"""Initialize the config flow."""
self._discovered_devices = {}
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle a flow initiated by the user."""
errors = {}
errors: dict[str, str] = {}
if user_input is not None:
user_input[CONF_MAC] = user_input[CONF_MAC].replace("-", ":").lower()
await self.async_set_unique_id(user_input[CONF_MAC].replace(":", ""))
self._abort_if_unique_id_configured()
# 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")
user_input[CONF_SENSOR_TYPE] = SUPPORTED_MODEL_TYPES[
self._discovered_devices[self.unique_id]["modelName"]
]
try:
return await self._validate_mac(user_input)
return self.async_create_entry(title=user_input[CONF_NAME], data=user_input)
except NotConnectedError:
errors["base"] = "cannot_connect"
try:
self._discovered_devices = await self._get_switchbots()
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unexpected exception")
return self.async_abort(reason="unknown")
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["modelName"] in SUPPORTED_MODEL_TYPES
and device["mac_address"] not in configured_devices
}
if not unconfigured_devices:
return self.async_abort(reason="no_unconfigured_devices")
data_schema = vol.Schema(
{
vol.Required(CONF_MAC): vol.In(unconfigured_devices),
vol.Required(CONF_NAME): str,
vol.Optional(CONF_PASSWORD): str,
vol.Required(CONF_MAC): str,
}
)

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

@ -5,18 +5,18 @@
"user": {
"title": "Setup Switchbot device",
"data": {
"mac": "Device MAC address",
"name": "[%key:common::config_flow::data::name%]",
"password": "[%key:common::config_flow::data::password%]",
"mac": "Device MAC address"
"password": "[%key:common::config_flow::data::password%]"
}
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]"
},
"error": {},
"abort": {
"already_configured_device": "[%key:common::config_flow::abort::already_configured_device%]",
"no_unconfigured_devices": "No unconfigured devices found.",
"unknown": "[%key:common::config_flow::error::unknown%]",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"switchbot_unsupported_type": "Unsupported Switchbot Type."
}
},

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

@ -2,20 +2,20 @@
"config": {
"abort": {
"already_configured_device": "Device is already configured",
"no_unconfigured_devices": "No unconfigured devices found.",
"unknown": "Unexpected error",
"cannot_connect": "Failed to connect",
"switchbot_unsupported_type": "Unsupported Switchbot Type."
},
"error": {
"cannot_connect": "Failed to connect"
},
"error": {},
"flow_title": "{name}",
"step": {
"user": {
"data": {
"mac": "Mac",
"name": "Name",
"password": "Password",
"mac": "Mac"
"password": "Password"
},
"title": "Setup Switchbot device"
}

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

@ -12,18 +12,35 @@ class MocGetSwitchbotDevices:
"""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,
"e78943999999": {
"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",
},
"e78943909090": {
"mac_address": "e7:89:43:90:90:90",
"Flags": "06",
"Manufacturer": "5900e78943d9fe7c",
"Complete 128b Services": "cba20d00-224d-11e6-9fb8-0002a5d5c51b",
"data": {
"calibration": True,
"battery": 74,
"position": 100,
"lightLevel": 2,
"rssi": -73,
},
"model": "c",
"modelName": "WoCurtain",
},
"model": "H",
"modelName": "WoHand",
}
self._curtain_all_services_data = {
"mac_address": "e7:89:43:90:90:90",
@ -90,6 +107,5 @@ def switchbot_config_flow(hass):
instance = mock_switchbot.return_value
instance.discover = MagicMock(return_value=True)
instance.get_device_data = MagicMock(return_value=True)
yield mock_switchbot

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

@ -19,8 +19,6 @@ from homeassistant.setup import async_setup_component
from . import (
USER_INPUT,
USER_INPUT_CURTAIN,
USER_INPUT_INVALID,
USER_INPUT_UNSUPPORTED_DEVICE,
YAML_CONFIG,
_patch_async_setup_entry,
init_integration,
@ -58,24 +56,6 @@ async def test_user_form_valid_mac(hass):
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"
# test curtain device creation.
result = await hass.config_entries.flow.async_init(
@ -103,47 +83,13 @@ async def test_user_form_valid_mac(hass):
assert len(mock_setup_entry.mock_calls) == 1
async def test_user_form_unsupported_device(hass):
"""Test the user initiated form for unsupported device type."""
await async_setup_component(hass, "persistent_notification", {})
# tests abort if no unconfigured devices are found.
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"}
assert result["reason"] == "no_unconfigured_devices"
async def test_async_step_import(hass):
@ -175,20 +121,13 @@ async def test_user_form_exception(hass, switchbot_config_flow):
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"}
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "cannot_connect"
switchbot_config_flow.side_effect = Exception
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
USER_INPUT,
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] == RESULT_TYPE_ABORT