github
/
ha-core
mirror of https://github.com/home-assistant/core
1
Fork 0
Code Releases Activity

Address Crownstone review comments (#56485)

This commit is contained in:
Ricardo Steijn 2021-09-23 09:23:45 +02:00 committed by GitHub
parent ea8f624f28
commit 63610eadc9
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
13 changed files with 261 additions and 272 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
homeassistant/components/crownstone/__init__.py
View File

@ -12,6 +12,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Initiate setup for a Crownstone config entry."""
manager = CrownstoneEntryManager(hass, entry)
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = manager
return await manager.async_setup()

248
homeassistant/components/crownstone/config_flow.py
View File

@ -1,7 +1,7 @@
"""Flow handler for Crownstone."""
from __future__ import annotations
from typing import Any
from typing import Any, Callable
from crownstone_cloud import CrownstoneCloud
from crownstone_cloud.exceptions import (
@ -16,7 +16,7 @@ from homeassistant.components import usb
from homeassistant.config_entries import ConfigEntry, ConfigFlow, OptionsFlow
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.data_entry_flow import FlowHandler, FlowResult
from homeassistant.helpers import aiohttp_client
from .const import (
@ -30,75 +30,26 @@ from .const import (
MANUAL_PATH,
REFRESH_LIST,
)
from .entry_manager import CrownstoneEntryManager
from .helpers import list_ports_as_str
CONFIG_FLOW = "config_flow"
OPTIONS_FLOW = "options_flow"
class CrownstoneConfigFlowHandler(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Crownstone."""
VERSION = 1
class BaseCrownstoneFlowHandler(FlowHandler):
"""Represent the base flow for Crownstone."""
cloud: CrownstoneCloud
@staticmethod
@callback
def async_get_options_flow(
config_entry: ConfigEntry,
) -> CrownstoneOptionsFlowHandler:
"""Return the Crownstone options."""
return CrownstoneOptionsFlowHandler(config_entry)
def __init__(self) -> None:
"""Initialize the flow."""
self.login_info: dict[str, Any] = {}
def __init__(
self, flow_type: str, create_entry_cb: Callable[..., FlowResult]
) -> None:
"""Set up flow instance."""
self.flow_type = flow_type
self.create_entry_callback = create_entry_cb
self.usb_path: str | None = None
self.usb_sphere_id: str | None = None
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the initial step."""
errors: dict[str, str] = {}
if user_input is None:
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{vol.Required(CONF_EMAIL): str, vol.Required(CONF_PASSWORD): str}
),
)
self.cloud = CrownstoneCloud(
email=user_input[CONF_EMAIL],
password=user_input[CONF_PASSWORD],
clientsession=aiohttp_client.async_get_clientsession(self.hass),
)
# Login & sync all user data
try:
await self.cloud.async_initialize()
except CrownstoneAuthenticationError as auth_error:
if auth_error.type == "LOGIN_FAILED":
errors["base"] = "invalid_auth"
elif auth_error.type == "LOGIN_FAILED_EMAIL_NOT_VERIFIED":
errors["base"] = "account_not_verified"
except CrownstoneUnknownError:
errors["base"] = "unknown_error"
# show form again, with the errors
if errors:
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{vol.Required(CONF_EMAIL): str, vol.Required(CONF_PASSWORD): str}
),
errors=errors,
)
await self.async_set_unique_id(self.cloud.cloud_data.user_id)
self._abort_if_unique_id_configured()
self.login_info = user_input
return await self.async_step_usb_config()
async def async_step_usb_config(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
@ -106,19 +57,25 @@ class CrownstoneConfigFlowHandler(ConfigFlow, domain=DOMAIN):
list_of_ports = await self.hass.async_add_executor_job(
serial.tools.list_ports.comports
)
ports_as_string = list_ports_as_str(list_of_ports)
if self.flow_type == CONFIG_FLOW:
ports_as_string = list_ports_as_str(list_of_ports)
else:
ports_as_string = list_ports_as_str(list_of_ports, False)
if user_input is not None:
selection = user_input[CONF_USB_PATH]
if selection == DONT_USE_USB:
return self.async_create_new_entry()
return self.create_entry_callback()
if selection == MANUAL_PATH:
return await self.async_step_usb_manual_config()
if selection != REFRESH_LIST:
selected_port: ListPortInfo = list_of_ports[
(ports_as_string.index(selection) - 1)
]
if self.flow_type == OPTIONS_FLOW:
index = ports_as_string.index(selection)
else:
index = ports_as_string.index(selection) - 1
selected_port: ListPortInfo = list_of_ports[index]
self.usb_path = await self.hass.async_add_executor_job(
usb.get_serial_by_id, selected_port.device
)
@ -165,11 +122,75 @@ class CrownstoneConfigFlowHandler(ConfigFlow, domain=DOMAIN):
elif user_input:
self.usb_sphere_id = spheres[user_input[CONF_USB_SPHERE]]
return self.async_create_new_entry()
return self.create_entry_callback()
class CrownstoneConfigFlowHandler(BaseCrownstoneFlowHandler, ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Crownstone."""
VERSION = 1
@staticmethod
@callback
def async_get_options_flow(
config_entry: ConfigEntry,
) -> CrownstoneOptionsFlowHandler:
"""Return the Crownstone options."""
return CrownstoneOptionsFlowHandler(config_entry)
def __init__(self) -> None:
"""Initialize the flow."""
super().__init__(CONFIG_FLOW, self.async_create_new_entry)
self.login_info: dict[str, Any] = {}
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the initial step."""
errors: dict[str, str] = {}
if user_input is None:
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{vol.Required(CONF_EMAIL): str, vol.Required(CONF_PASSWORD): str}
),
)
self.cloud = CrownstoneCloud(
email=user_input[CONF_EMAIL],
password=user_input[CONF_PASSWORD],
clientsession=aiohttp_client.async_get_clientsession(self.hass),
)
# Login & sync all user data
try:
await self.cloud.async_initialize()
except CrownstoneAuthenticationError as auth_error:
if auth_error.type == "LOGIN_FAILED":
errors["base"] = "invalid_auth"
elif auth_error.type == "LOGIN_FAILED_EMAIL_NOT_VERIFIED":
errors["base"] = "account_not_verified"
except CrownstoneUnknownError:
errors["base"] = "unknown_error"
# show form again, with the errors
if errors:
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{vol.Required(CONF_EMAIL): str, vol.Required(CONF_PASSWORD): str}
),
errors=errors,
)
await self.async_set_unique_id(self.cloud.cloud_data.user_id)
self._abort_if_unique_id_configured()
self.login_info = user_input
return await self.async_step_usb_config()
def async_create_new_entry(self) -> FlowResult:
"""Create a new entry."""
return self.async_create_entry(
return super().async_create_entry(
title=f"Account: {self.login_info[CONF_EMAIL]}",
data={
CONF_EMAIL: self.login_info[CONF_EMAIL],
@ -179,22 +200,22 @@ class CrownstoneConfigFlowHandler(ConfigFlow, domain=DOMAIN):
)
class CrownstoneOptionsFlowHandler(OptionsFlow):
class CrownstoneOptionsFlowHandler(BaseCrownstoneFlowHandler, OptionsFlow):
"""Handle Crownstone options."""
def __init__(self, config_entry: ConfigEntry) -> None:
"""Initialize Crownstone options."""
super().__init__(OPTIONS_FLOW, self.async_create_new_entry)
self.entry = config_entry
self.updated_options = config_entry.options.copy()
self.spheres: dict[str, str] = {}
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Manage Crownstone options."""
manager: CrownstoneEntryManager = self.hass.data[DOMAIN][self.entry.entry_id]
self.cloud: CrownstoneCloud = self.hass.data[DOMAIN][self.entry.entry_id].cloud
spheres = {sphere.name: sphere.cloud_id for sphere in manager.cloud.cloud_data}
spheres = {sphere.name: sphere.cloud_id for sphere in self.cloud.cloud_data}
usb_path = self.entry.options.get(CONF_USB_PATH)
usb_sphere = self.entry.options.get(CONF_USB_SPHERE)
@ -206,15 +227,14 @@ class CrownstoneOptionsFlowHandler(OptionsFlow):
{
vol.Optional(
CONF_USB_SPHERE_OPTION,
default=manager.cloud.cloud_data.spheres[usb_sphere].name,
default=self.cloud.cloud_data.data[usb_sphere].name,
): vol.In(spheres.keys())
}
)
if user_input is not None:
if user_input[CONF_USE_USB_OPTION] and usb_path is None:
self.spheres = spheres
return await self.async_step_usb_config_option()
return await self.async_step_usb_config()
if not user_input[CONF_USE_USB_OPTION] and usb_path is not None:
self.updated_options[CONF_USB_PATH] = None
self.updated_options[CONF_USB_SPHERE] = None
@ -223,77 +243,17 @@ class CrownstoneOptionsFlowHandler(OptionsFlow):
and spheres[user_input[CONF_USB_SPHERE_OPTION]] != usb_sphere
):
sphere_id = spheres[user_input[CONF_USB_SPHERE_OPTION]]
user_input[CONF_USB_SPHERE_OPTION] = sphere_id
self.updated_options[CONF_USB_SPHERE] = sphere_id
return self.async_create_entry(title="", data=self.updated_options)
return self.async_create_new_entry()
return self.async_show_form(step_id="init", data_schema=options_schema)
async def async_step_usb_config_option(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Set up a Crownstone USB dongle."""
list_of_ports = await self.hass.async_add_executor_job(
serial.tools.list_ports.comports
)
ports_as_string = list_ports_as_str(list_of_ports, False)
def async_create_new_entry(self) -> FlowResult:
"""Create a new entry."""
# these attributes will only change when a usb was configured
if self.usb_path is not None and self.usb_sphere_id is not None:
self.updated_options[CONF_USB_PATH] = self.usb_path
self.updated_options[CONF_USB_SPHERE] = self.usb_sphere_id
if user_input is not None:
selection = user_input[CONF_USB_PATH]
if selection == MANUAL_PATH:
return await self.async_step_usb_manual_config_option()
if selection != REFRESH_LIST:
selected_port: ListPortInfo = list_of_ports[
ports_as_string.index(selection)
]
usb_path = await self.hass.async_add_executor_job(
usb.get_serial_by_id, selected_port.device
)
self.updated_options[CONF_USB_PATH] = usb_path
return await self.async_step_usb_sphere_config_option()
return self.async_show_form(
step_id="usb_config_option",
data_schema=vol.Schema(
{vol.Required(CONF_USB_PATH): vol.In(ports_as_string)}
),
)
async def async_step_usb_manual_config_option(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Manually enter Crownstone USB dongle path."""
if user_input is None:
return self.async_show_form(
step_id="usb_manual_config_option",
data_schema=vol.Schema({vol.Required(CONF_USB_MANUAL_PATH): str}),
)
self.updated_options[CONF_USB_PATH] = user_input[CONF_USB_MANUAL_PATH]
return await self.async_step_usb_sphere_config_option()
async def async_step_usb_sphere_config_option(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Select a Crownstone sphere that the USB operates in."""
# no need to select if there's only 1 option
sphere_id: str | None = None
if len(self.spheres) == 1:
sphere_id = next(iter(self.spheres.values()))
if user_input is None and sphere_id is None:
return self.async_show_form(
step_id="usb_sphere_config_option",
data_schema=vol.Schema({CONF_USB_SPHERE: vol.In(self.spheres.keys())}),
)
if sphere_id:
self.updated_options[CONF_USB_SPHERE] = sphere_id
elif user_input:
self.updated_options[CONF_USB_SPHERE] = self.spheres[
user_input[CONF_USB_SPHERE]
]
return self.async_create_entry(title="", data=self.updated_options)
return super().async_create_entry(title="", data=self.updated_options)

3
homeassistant/components/crownstone/const.py
View File

@ -19,9 +19,6 @@ SIG_CROWNSTONE_STATE_UPDATE: Final = "crownstone.crownstone_state_update"
SIG_CROWNSTONE_UPDATE: Final = "crownstone.crownstone_update"
SIG_UART_STATE_CHANGE: Final = "crownstone.uart_state_change"
# Abilities state
ABILITY_STATE: Final[dict[bool, str]] = {True: "Enabled", False: "Disabled"}
# Config flow
CONF_USB_PATH: Final = "usb_path"
CONF_USB_MANUAL_PATH: Final = "usb_manual_path"

8
homeassistant/components/crownstone/devices.py
View File

@ -10,13 +10,15 @@ from homeassistant.const import (
ATTR_NAME,
ATTR_SW_VERSION,
)
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity import DeviceInfo, Entity
from .const import CROWNSTONE_INCLUDE_TYPES, DOMAIN
class CrownstoneDevice:
"""Representation of a Crownstone device."""
class CrownstoneBaseEntity(Entity):
"""Base entity class for Crownstone devices."""
_attr_should_poll = False
def __init__(self, device: Crownstone) -> None:
"""Initialize the device."""

5
homeassistant/components/crownstone/entry_manager.py
View File

@ -20,6 +20,7 @@ from homeassistant.const import CONF_EMAIL, CONF_PASSWORD, EVENT_HOMEASSISTANT_S
from homeassistant.core import Event, HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import aiohttp_client
from homeassistant.helpers.dispatcher import async_dispatcher_send
from .const import (
CONF_USB_PATH,
@ -96,7 +97,6 @@ class CrownstoneEntryManager:
# Makes HA aware of the Crownstone environment HA is placed in, a user can have multiple
self.usb_sphere_id = self.config_entry.options[CONF_USB_SPHERE]
self.hass.data.setdefault(DOMAIN, {})[self.config_entry.entry_id] = self
self.hass.config_entries.async_setup_platforms(self.config_entry, PLATFORMS)
# HA specific listeners
@ -114,8 +114,7 @@ class CrownstoneEntryManager:
async with sse_client as client:
async for event in client:
if event is not None:
# Make SSE updates, like ability change, available to the user
self.hass.bus.async_fire(f"{DOMAIN}_{event.type}", event.data)
async_dispatcher_send(self.hass, f"{DOMAIN}_{event.type}", event)
async def async_setup_usb(self) -> None:
"""Attempt setup of a Crownstone usb dongle."""

58
homeassistant/components/crownstone/light.py
View File

@ -1,17 +1,11 @@
"""Support for Crownstone devices."""
from __future__ import annotations
from collections.abc import Mapping
from functools import partial
import logging
from typing import TYPE_CHECKING, Any
from crownstone_cloud.cloud_models.crownstones import Crownstone
from crownstone_cloud.const import (
DIMMING_ABILITY,
SWITCHCRAFT_ABILITY,
TAP_TO_TOGGLE_ABILITY,
)
from crownstone_cloud.const import DIMMING_ABILITY
from crownstone_cloud.exceptions import CrownstoneAbilityError
from crownstone_uart import CrownstoneUart
@ -22,25 +16,23 @@ from homeassistant.components.light import (
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import (
ABILITY_STATE,
CROWNSTONE_INCLUDE_TYPES,
CROWNSTONE_SUFFIX,
DOMAIN,
SIG_CROWNSTONE_STATE_UPDATE,
SIG_UART_STATE_CHANGE,
)
from .devices import CrownstoneDevice
from .devices import CrownstoneBaseEntity
from .helpers import map_from_to
if TYPE_CHECKING:
from .entry_manager import CrownstoneEntryManager
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistant,
@ -76,17 +68,18 @@ def hass_to_crownstone_state(value: int) -> int:
return map_from_to(value, 0, 255, 0, 100)
class CrownstoneEntity(CrownstoneDevice, LightEntity):
class CrownstoneEntity(CrownstoneBaseEntity, LightEntity):
"""
Representation of a crownstone.
Light platform is used to support dimming.
"""
_attr_should_poll = False
_attr_icon = "mdi:power-socket-de"
def __init__(self, crownstone_data: Crownstone, usb: CrownstoneUart = None) -> None:
def __init__(
self, crownstone_data: Crownstone, usb: CrownstoneUart | None = None
) -> None:
"""Initialize the crownstone."""
super().__init__(crownstone_data)
self.usb = usb
@ -94,11 +87,6 @@ class CrownstoneEntity(CrownstoneDevice, LightEntity):
self._attr_name = str(self.device.name)
self._attr_unique_id = f"{self.cloud_id}-{CROWNSTONE_SUFFIX}"
@property
def usb_available(self) -> bool:
"""Return if this entity can use a usb dongle."""
return self.usb is not None and self.usb.is_ready()
@property
def brightness(self) -> int | None:
"""Return the brightness if dimming enabled."""
@ -116,29 +104,6 @@ class CrownstoneEntity(CrownstoneDevice, LightEntity):
return SUPPORT_BRIGHTNESS
return 0
@property
def extra_state_attributes(self) -> Mapping[str, Any] | None:
"""State attributes for Crownstone devices."""
attributes: dict[str, Any] = {}
# switch method
if self.usb_available:
attributes["switch_method"] = "Crownstone USB Dongle"
else:
attributes["switch_method"] = "Crownstone Cloud"
# crownstone abilities
attributes["dimming"] = ABILITY_STATE.get(
self.device.abilities.get(DIMMING_ABILITY).is_enabled
)
attributes["tap_to_toggle"] = ABILITY_STATE.get(
self.device.abilities.get(TAP_TO_TOGGLE_ABILITY).is_enabled
)
attributes["switchcraft"] = ABILITY_STATE.get(
self.device.abilities.get(SWITCHCRAFT_ABILITY).is_enabled
)
return attributes
async def async_added_to_hass(self) -> None:
"""Set up a listener when this entity is added to HA."""
# new state received
@ -157,7 +122,7 @@ class CrownstoneEntity(CrownstoneDevice, LightEntity):
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on this light via dongle or cloud."""
if ATTR_BRIGHTNESS in kwargs:
if self.usb_available:
if self.usb is not None and self.usb.is_ready():
await self.hass.async_add_executor_job(
partial(
self.usb.dim_crownstone,
@ -171,14 +136,13 @@ class CrownstoneEntity(CrownstoneDevice, LightEntity):
hass_to_crownstone_state(kwargs[ATTR_BRIGHTNESS])
)
except CrownstoneAbilityError as ability_error:
_LOGGER.error(ability_error)
return
raise HomeAssistantError(ability_error) from ability_error
# assume brightness is set on device
self.device.state = hass_to_crownstone_state(kwargs[ATTR_BRIGHTNESS])
self.async_write_ha_state()
elif self.usb_available:
elif self.usb is not None and self.usb.is_ready():
await self.hass.async_add_executor_job(
partial(self.usb.switch_crownstone, self.device.unique_id, on=True)
)
@ -192,7 +156,7 @@ class CrownstoneEntity(CrownstoneDevice, LightEntity):
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off this device via dongle or cloud."""
if self.usb_available:
if self.usb is not None and self.usb.is_ready():
await self.hass.async_add_executor_job(
partial(self.usb.switch_crownstone, self.device.unique_id, on=False)
)

29
homeassistant/components/crownstone/listeners.py
View File

@ -9,6 +9,7 @@ from __future__ import annotations
from functools import partial
from typing import TYPE_CHECKING, cast
from crownstone_cloud.exceptions import CrownstoneNotFoundError
from crownstone_core.packets.serviceDataParsers.containers.AdvExternalCrownstoneState import (
AdvExternalCrownstoneState,
)
@ -25,8 +26,12 @@ from crownstone_sse.events import AbilityChangeEvent, SwitchStateUpdateEvent
from crownstone_uart import UartEventBus, UartTopics
from crownstone_uart.topics.SystemTopics import SystemTopics
from homeassistant.core import Event, callback
from homeassistant.helpers.dispatcher import async_dispatcher_send, dispatcher_send
from homeassistant.core import callback
from homeassistant.helpers.dispatcher import (
async_dispatcher_connect,
async_dispatcher_send,
dispatcher_send,
)
from .const import (
DOMAIN,
@ -42,13 +47,12 @@ if TYPE_CHECKING:
@callback
def async_update_crwn_state_sse(
manager: CrownstoneEntryManager, ha_event: Event
manager: CrownstoneEntryManager, switch_event: SwitchStateUpdateEvent
) -> None:
"""Update the state of a Crownstone when switched externally."""
switch_event = SwitchStateUpdateEvent(ha_event.data)
try:
updated_crownstone = manager.cloud.get_crownstone_by_id(switch_event.cloud_id)
except KeyError:
except CrownstoneNotFoundError:
return
# only update on change.
@ -58,12 +62,13 @@ def async_update_crwn_state_sse(
@callback
def async_update_crwn_ability(manager: CrownstoneEntryManager, ha_event: Event) -> None:
def async_update_crwn_ability(
manager: CrownstoneEntryManager, ability_event: AbilityChangeEvent
) -> None:
"""Update the ability information of a Crownstone."""
ability_event = AbilityChangeEvent(ha_event.data)
try:
updated_crownstone = manager.cloud.get_crownstone_by_id(ability_event.cloud_id)
except KeyError:
except CrownstoneNotFoundError:
return
ability_type = ability_event.ability_type
@ -100,7 +105,7 @@ def update_crwn_state_uart(
updated_crownstone = manager.cloud.get_crownstone_by_uid(
data.crownstoneId, manager.usb_sphere_id
)
except KeyError:
except CrownstoneNotFoundError:
return
if data.switchState is None:
@ -117,11 +122,13 @@ def setup_sse_listeners(manager: CrownstoneEntryManager) -> None:
"""Set up SSE listeners."""
# save unsub function for when entry removed
manager.listeners[SSE_LISTENERS] = [
manager.hass.bus.async_listen(
async_dispatcher_connect(
manager.hass,
f"{DOMAIN}_{EVENT_SWITCH_STATE_UPDATE}",
partial(async_update_crwn_state_sse, manager),
),
manager.hass.bus.async_listen(
async_dispatcher_connect(
manager.hass,
f"{DOMAIN}_{EVENT_ABILITY_CHANGE}",
partial(async_update_crwn_ability, manager),
),

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

@ -4,7 +4,7 @@
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/crownstone",
"requirements": [
"crownstone-cloud==1.4.7",
"crownstone-cloud==1.4.8",
"crownstone-sse==2.0.2",
"crownstone-uart==2.1.0",
"pyserial==3.5"

6
homeassistant/components/crownstone/strings.json
View File

@ -49,21 +49,21 @@
"usb_sphere_option": "Crownstone Sphere where the USB is located"
}
},
"usb_config_option": {
"usb_config": {
"data": {
"usb_path": "[%key:common::config_flow::data::usb_path%]"
},
"title": "Crownstone USB dongle configuration",
"description": "Select the serial port of the Crownstone USB dongle.\n\nLook for a device with VID 10C4 and PID EA60."
},
"usb_manual_config_option": {
"usb_manual_config": {
"data": {
"usb_manual_path": "[%key:common::config_flow::data::usb_path%]"
},
"title": "Crownstone USB dongle manual path",
"description": "Manually enter the path of a Crownstone USB dongle."
},
"usb_sphere_config_option": {
"usb_sphere_config": {
"data": {
"usb_sphere": "Crownstone Sphere"
},

6
homeassistant/components/crownstone/translations/en.json
View File

@ -49,21 +49,21 @@
"use_usb_option": "Use a Crownstone USB dongle for local data transmission"
}
},
"usb_config_option": {
"usb_config": {
"data": {
"usb_path": "USB Device Path"
},
"description": "Select the serial port of the Crownstone USB dongle.\n\nLook for a device with VID 10C4 and PID EA60.",
"title": "Crownstone USB dongle configuration"
},
"usb_manual_config_option": {
"usb_manual_config": {
"data": {
"usb_manual_path": "USB Device Path"
},
"description": "Manually enter the path of a Crownstone USB dongle.",
"title": "Crownstone USB dongle manual path"
},
"usb_sphere_config_option": {
"usb_sphere_config": {
"data": {
"usb_sphere": "Crownstone Sphere"
},

2
requirements_all.txt
View File

@ -490,7 +490,7 @@ coronavirus==1.1.1
croniter==1.0.6
# homeassistant.components.crownstone
crownstone-cloud==1.4.7
crownstone-cloud==1.4.8
# homeassistant.components.crownstone
crownstone-sse==2.0.2

2
requirements_test_all.txt
View File

@ -292,7 +292,7 @@ coronavirus==1.1.1
croniter==1.0.6
# homeassistant.components.crownstone
crownstone-cloud==1.4.7
crownstone-cloud==1.4.8
# homeassistant.components.crownstone
crownstone-sse==2.0.2

162
tests/components/crownstone/test_config_flow.py
View File

@ -1,6 +1,7 @@
"""Tests for the Crownstone integration."""
from __future__ import annotations
from typing import Generator, Union
from unittest.mock import AsyncMock, MagicMock, patch
from crownstone_cloud.cloud_models.spheres import Spheres
@ -28,14 +29,45 @@ from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
MockFixture = Generator[Union[MagicMock, AsyncMock], None, None]
@pytest.fixture(name="crownstone_setup", autouse=True)
def crownstone_setup():
@pytest.fixture(name="crownstone_setup")
def crownstone_setup() -> MockFixture:
"""Mock Crownstone entry setup."""
with patch(
"homeassistant.components.crownstone.async_setup_entry", return_value=True
):
yield
) as setup_mock:
yield setup_mock
@pytest.fixture(name="pyserial_comports")
def usb_comports() -> MockFixture:
"""Mock pyserial comports."""
with patch(
"serial.tools.list_ports.comports",
MagicMock(return_value=[get_mocked_com_port()]),
) as comports_mock:
yield comports_mock
@pytest.fixture(name="usb_path")
def usb_path() -> MockFixture:
"""Mock usb serial path."""
with patch(
"homeassistant.components.usb.get_serial_by_id",
return_value="/dev/serial/by-id/crownstone-usb",
) as usb_path_mock:
yield usb_path_mock
def get_mocked_crownstone_entry_manager(mocked_cloud: MagicMock):
"""Get a mocked CrownstoneEntryManager instance."""
mocked_entry_manager = MagicMock()
mocked_entry_manager.async_setup = AsyncMock(return_value=True)
mocked_entry_manager.cloud = mocked_cloud
return mocked_entry_manager
def get_mocked_crownstone_cloud(spheres: dict[str, MagicMock] | None = None):
@ -43,7 +75,7 @@ def get_mocked_crownstone_cloud(spheres: dict[str, MagicMock] | None = None):
mock_cloud = MagicMock()
mock_cloud.async_initialize = AsyncMock()
mock_cloud.cloud_data = Spheres(MagicMock(), "account_id")
mock_cloud.cloud_data.spheres = spheres
mock_cloud.cloud_data.data = spheres
return mock_cloud
@ -101,26 +133,26 @@ async def start_config_flow(hass: HomeAssistant, mocked_cloud: MagicMock):
"homeassistant.components.crownstone.config_flow.CrownstoneCloud",
return_value=mocked_cloud,
):
result = await hass.config_entries.flow.async_init(
return await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "user"}, data=mocked_login_input
)
return result
async def start_options_flow(
hass: HomeAssistant, entry_id: str, mocked_cloud: MagicMock
hass: HomeAssistant, entry_id: str, mocked_manager: MagicMock
):
"""Patch CrownstoneEntryManager and start the flow."""
mocked_manager = MagicMock()
mocked_manager.cloud = mocked_cloud
hass.data.setdefault(DOMAIN, {})
hass.data[DOMAIN][entry_id] = mocked_manager
# set up integration
with patch(
"homeassistant.components.crownstone.CrownstoneEntryManager",
return_value=mocked_manager,
):
await hass.config_entries.async_setup(entry_id)
return await hass.config_entries.options.async_init(entry_id)
async def test_no_user_input(hass: HomeAssistant):
async def test_no_user_input(crownstone_setup: MockFixture, hass: HomeAssistant):
"""Test the flow done in the correct way."""
# test if a form is returned if no input is provided
result = await hass.config_entries.flow.async_init(
@ -129,9 +161,10 @@ async def test_no_user_input(hass: HomeAssistant):
# show the login form
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
assert crownstone_setup.call_count == 0
async def test_abort_if_configured(hass: HomeAssistant):
async def test_abort_if_configured(crownstone_setup: MockFixture, hass: HomeAssistant):
"""Test flow with correct login input and abort if sphere already configured."""
# create mock entry conf
configured_entry_data = create_mocked_entry_data_conf(
@ -156,9 +189,12 @@ async def test_abort_if_configured(hass: HomeAssistant):
# test if we abort if we try to configure the same entry
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
assert crownstone_setup.call_count == 0
async def test_authentication_errors(hass: HomeAssistant):
async def test_authentication_errors(
crownstone_setup: MockFixture, hass: HomeAssistant
):
"""Test flow with wrong auth errors."""
cloud = get_mocked_crownstone_cloud()
# side effect: auth error login failed
@ -180,9 +216,10 @@ async def test_authentication_errors(hass: HomeAssistant):
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {"base": "account_not_verified"}
assert crownstone_setup.call_count == 0
async def test_unknown_error(hass: HomeAssistant):
async def test_unknown_error(crownstone_setup: MockFixture, hass: HomeAssistant):
"""Test flow with unknown error."""
cloud = get_mocked_crownstone_cloud()
# side effect: unknown error
@ -192,9 +229,12 @@ async def test_unknown_error(hass: HomeAssistant):
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {"base": "unknown_error"}
assert crownstone_setup.call_count == 0
async def test_successful_login_no_usb(hass: HomeAssistant):
async def test_successful_login_no_usb(
crownstone_setup: MockFixture, hass: HomeAssistant
):
"""Test a successful login without configuring a USB."""
entry_data_without_usb = create_mocked_entry_data_conf(
email="example@homeassistant.com",
@ -217,16 +257,15 @@ async def test_successful_login_no_usb(hass: HomeAssistant):
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["data"] == entry_data_without_usb
assert result["options"] == entry_options_without_usb
assert crownstone_setup.call_count == 1
@patch(
"serial.tools.list_ports.comports", MagicMock(return_value=[get_mocked_com_port()])
)
@patch(
"homeassistant.components.usb.get_serial_by_id",
return_value="/dev/serial/by-id/crownstone-usb",
)
async def test_successful_login_with_usb(serial_mock: MagicMock, hass: HomeAssistant):
async def test_successful_login_with_usb(
crownstone_setup: MockFixture,
pyserial_comports: MockFixture,
usb_path: MockFixture,
hass: HomeAssistant,
):
"""Test flow with correct login and usb configuration."""
entry_data_with_usb = create_mocked_entry_data_conf(
email="example@homeassistant.com",
@ -243,6 +282,7 @@ async def test_successful_login_with_usb(serial_mock: MagicMock, hass: HomeAssis
# should show usb form
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "usb_config"
assert pyserial_comports.call_count == 1
# create a mocked port
port = get_mocked_com_port()
@ -261,7 +301,8 @@ async def test_successful_login_with_usb(serial_mock: MagicMock, hass: HomeAssis
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "usb_sphere_config"
assert serial_mock.call_count == 1
assert pyserial_comports.call_count == 2
assert usb_path.call_count == 1
# select a sphere
result = await hass.config_entries.flow.async_configure(
@ -270,12 +311,12 @@ async def test_successful_login_with_usb(serial_mock: MagicMock, hass: HomeAssis
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["data"] == entry_data_with_usb
assert result["options"] == entry_options_with_usb
assert crownstone_setup.call_count == 1
@patch(
"serial.tools.list_ports.comports", MagicMock(return_value=[get_mocked_com_port()])
)
async def test_successful_login_with_manual_usb_path(hass: HomeAssistant):
async def test_successful_login_with_manual_usb_path(
crownstone_setup: MockFixture, pyserial_comports: MockFixture, hass: HomeAssistant
):
"""Test flow with correct login and usb configuration."""
entry_data_with_manual_usb = create_mocked_entry_data_conf(
email="example@homeassistant.com",
@ -292,6 +333,7 @@ async def test_successful_login_with_manual_usb_path(hass: HomeAssistant):
# should show usb form
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "usb_config"
assert pyserial_comports.call_count == 1
# select manual from the list
result = await hass.config_entries.flow.async_configure(
@ -300,6 +342,7 @@ async def test_successful_login_with_manual_usb_path(hass: HomeAssistant):
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "usb_manual_config"
assert pyserial_comports.call_count == 2
# enter USB path
path = "/dev/crownstone-usb"
@ -312,16 +355,12 @@ async def test_successful_login_with_manual_usb_path(hass: HomeAssistant):
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["data"] == entry_data_with_manual_usb
assert result["options"] == entry_options_with_manual_usb
assert crownstone_setup.call_count == 1
@patch(
"serial.tools.list_ports.comports", MagicMock(return_value=[get_mocked_com_port()])
)
@patch(
"homeassistant.components.usb.get_serial_by_id",
return_value="/dev/serial/by-id/crownstone-usb",
)
async def test_options_flow_setup_usb(serial_mock: MagicMock, hass: HomeAssistant):
async def test_options_flow_setup_usb(
pyserial_comports: MockFixture, usb_path: MockFixture, hass: HomeAssistant
):
"""Test options flow init."""
configured_entry_data = create_mocked_entry_data_conf(
email="example@homeassistant.com",
@ -342,7 +381,11 @@ async def test_options_flow_setup_usb(serial_mock: MagicMock, hass: HomeAssistan
entry.add_to_hass(hass)
result = await start_options_flow(
hass, entry.entry_id, get_mocked_crownstone_cloud(create_mocked_spheres(2))
hass,
entry.entry_id,
get_mocked_crownstone_entry_manager(
get_mocked_crownstone_cloud(create_mocked_spheres(2))
),
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
@ -360,7 +403,8 @@ async def test_options_flow_setup_usb(serial_mock: MagicMock, hass: HomeAssistan
result["flow_id"], user_input={CONF_USE_USB_OPTION: True}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "usb_config_option"
assert result["step_id"] == "usb_config"
assert pyserial_comports.call_count == 1
# create a mocked port
port = get_mocked_com_port()
@ -378,8 +422,9 @@ async def test_options_flow_setup_usb(serial_mock: MagicMock, hass: HomeAssistan
result["flow_id"], user_input={CONF_USB_PATH: port_select}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "usb_sphere_config_option"
assert serial_mock.call_count == 1
assert result["step_id"] == "usb_sphere_config"
assert pyserial_comports.call_count == 2
assert usb_path.call_count == 1
# select a sphere
result = await hass.config_entries.options.async_configure(
@ -412,7 +457,11 @@ async def test_options_flow_remove_usb(hass: HomeAssistant):
entry.add_to_hass(hass)
result = await start_options_flow(
hass, entry.entry_id, get_mocked_crownstone_cloud(create_mocked_spheres(2))
hass,
entry.entry_id,
get_mocked_crownstone_entry_manager(
get_mocked_crownstone_cloud(create_mocked_spheres(2))
),
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
@ -438,10 +487,9 @@ async def test_options_flow_remove_usb(hass: HomeAssistant):
)
@patch(
"serial.tools.list_ports.comports", MagicMock(return_value=[get_mocked_com_port()])
)
async def test_options_flow_manual_usb_path(hass: HomeAssistant):
async def test_options_flow_manual_usb_path(
pyserial_comports: MockFixture, hass: HomeAssistant
):
"""Test flow with correct login and usb configuration."""
configured_entry_data = create_mocked_entry_data_conf(
email="example@homeassistant.com",
@ -462,7 +510,11 @@ async def test_options_flow_manual_usb_path(hass: HomeAssistant):
entry.add_to_hass(hass)
result = await start_options_flow(
hass, entry.entry_id, get_mocked_crownstone_cloud(create_mocked_spheres(1))
hass,
entry.entry_id,
get_mocked_crownstone_entry_manager(
get_mocked_crownstone_cloud(create_mocked_spheres(1))
),
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
@ -472,7 +524,8 @@ async def test_options_flow_manual_usb_path(hass: HomeAssistant):
result["flow_id"], user_input={CONF_USE_USB_OPTION: True}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "usb_config_option"
assert result["step_id"] == "usb_config"
assert pyserial_comports.call_count == 1
# select manual from the list
result = await hass.config_entries.options.async_configure(
@ -480,7 +533,8 @@ async def test_options_flow_manual_usb_path(hass: HomeAssistant):
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "usb_manual_config_option"
assert result["step_id"] == "usb_manual_config"
assert pyserial_comports.call_count == 2
# enter USB path
path = "/dev/crownstone-usb"
@ -515,7 +569,11 @@ async def test_options_flow_change_usb_sphere(hass: HomeAssistant):
entry.add_to_hass(hass)
result = await start_options_flow(
hass, entry.entry_id, get_mocked_crownstone_cloud(create_mocked_spheres(3))
hass,
entry.entry_id,
get_mocked_crownstone_entry_manager(
get_mocked_crownstone_cloud(create_mocked_spheres(3))
),
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM