From 63610eadc98bc498c1820ad5553245590284e125 Mon Sep 17 00:00:00 2001 From: Ricardo Steijn <61013287+RicArch97@users.noreply.github.com> Date: Thu, 23 Sep 2021 09:23:45 +0200 Subject: [PATCH] Address Crownstone review comments (#56485) --- .../components/crownstone/__init__.py | 2 + .../components/crownstone/config_flow.py | 248 ++++++++---------- homeassistant/components/crownstone/const.py | 3 - .../components/crownstone/devices.py | 8 +- .../components/crownstone/entry_manager.py | 5 +- homeassistant/components/crownstone/light.py | 58 +--- .../components/crownstone/listeners.py | 29 +- .../components/crownstone/manifest.json | 2 +- .../components/crownstone/strings.json | 6 +- .../crownstone/translations/en.json | 6 +- requirements_all.txt | 2 +- requirements_test_all.txt | 2 +- .../components/crownstone/test_config_flow.py | 162 ++++++++---- 13 files changed, 261 insertions(+), 272 deletions(-) diff --git a/homeassistant/components/crownstone/__init__.py b/homeassistant/components/crownstone/__init__.py index bd4aae7966525d..92b2f4de5ca2d2 100644 --- a/homeassistant/components/crownstone/__init__.py +++ b/homeassistant/components/crownstone/__init__.py @@ -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() diff --git a/homeassistant/components/crownstone/config_flow.py b/homeassistant/components/crownstone/config_flow.py index 72edeef7910db9..86826f5f6f8dbd 100644 --- a/homeassistant/components/crownstone/config_flow.py +++ b/homeassistant/components/crownstone/config_flow.py @@ -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.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 @@ 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 - cloud: CrownstoneCloud +class BaseCrownstoneFlowHandler(FlowHandler): + """Represent the base flow for Crownstone.""" - @staticmethod - @callback - def async_get_options_flow( - config_entry: ConfigEntry, - ) -> CrownstoneOptionsFlowHandler: - """Return the Crownstone options.""" - return CrownstoneOptionsFlowHandler(config_entry) + cloud: CrownstoneCloud - 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 @@ async def async_step_usb_config( 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 @@ async def async_step_usb_sphere_config( 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 @@ def async_create_new_entry(self) -> FlowResult: ) -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 @@ async def async_step_init( { 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 @@ async def async_step_init( 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) - - 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] - ] + 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 - return self.async_create_entry(title="", data=self.updated_options) + return super().async_create_entry(title="", data=self.updated_options) diff --git a/homeassistant/components/crownstone/const.py b/homeassistant/components/crownstone/const.py index 2238701dcaf86a..21a14b99e86c35 100644 --- a/homeassistant/components/crownstone/const.py +++ b/homeassistant/components/crownstone/const.py @@ -19,9 +19,6 @@ 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" diff --git a/homeassistant/components/crownstone/devices.py b/homeassistant/components/crownstone/devices.py index 49965bc8fcd6ff..91af18ab15ea69 100644 --- a/homeassistant/components/crownstone/devices.py +++ b/homeassistant/components/crownstone/devices.py @@ -10,13 +10,15 @@ 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.""" diff --git a/homeassistant/components/crownstone/entry_manager.py b/homeassistant/components/crownstone/entry_manager.py index b01316a771acb6..b1963462adc706 100644 --- a/homeassistant/components/crownstone/entry_manager.py +++ b/homeassistant/components/crownstone/entry_manager.py @@ -20,6 +20,7 @@ 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 @@ async def async_setup(self) -> bool: # 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 @@ async def async_process_events(self, sse_client: CrownstoneSSEAsync) -> None: 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.""" diff --git a/homeassistant/components/crownstone/light.py b/homeassistant/components/crownstone/light.py index b2d4d8411b7f2d..ff647b2fc849ac 100644 --- a/homeassistant/components/crownstone/light.py +++ b/homeassistant/components/crownstone/light.py @@ -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.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 @@ def __init__(self, crownstone_data: Crownstone, usb: CrownstoneUart = None) -> N 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 @@ def supported_features(self) -> int: 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 @@ async def async_added_to_hass(self) -> None: 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 @@ async def async_turn_on(self, **kwargs: Any) -> None: 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 @@ async def async_turn_on(self, **kwargs: Any) -> None: 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) ) diff --git a/homeassistant/components/crownstone/listeners.py b/homeassistant/components/crownstone/listeners.py index ae316bc0029161..63891545cab847 100644 --- a/homeassistant/components/crownstone/listeners.py +++ b/homeassistant/components/crownstone/listeners.py @@ -9,6 +9,7 @@ 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_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 @@ @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), ), diff --git a/homeassistant/components/crownstone/manifest.json b/homeassistant/components/crownstone/manifest.json index a7caa6a8d7ff9b..4615d0b03295d1 100644 --- a/homeassistant/components/crownstone/manifest.json +++ b/homeassistant/components/crownstone/manifest.json @@ -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" diff --git a/homeassistant/components/crownstone/strings.json b/homeassistant/components/crownstone/strings.json index 7437d458ea7820..25c9fd10293e43 100644 --- a/homeassistant/components/crownstone/strings.json +++ b/homeassistant/components/crownstone/strings.json @@ -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" }, diff --git a/homeassistant/components/crownstone/translations/en.json b/homeassistant/components/crownstone/translations/en.json index e8b552ba53c86d..09a26b9739c93e 100644 --- a/homeassistant/components/crownstone/translations/en.json +++ b/homeassistant/components/crownstone/translations/en.json @@ -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" }, diff --git a/requirements_all.txt b/requirements_all.txt index 6a071858cf7f05..e0ec52746544d8 100644 --- a/requirements_all.txt +++ b/requirements_all.txt @@ -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 diff --git a/requirements_test_all.txt b/requirements_test_all.txt index 0bef36a9971b39..c38966e64fd4e3 100644 --- a/requirements_test_all.txt +++ b/requirements_test_all.txt @@ -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 diff --git a/tests/components/crownstone/test_config_flow.py b/tests/components/crownstone/test_config_flow.py index 227657a65a20af..7b05c8ba53080b 100644 --- a/tests/components/crownstone/test_config_flow.py +++ b/tests/components/crownstone/test_config_flow.py @@ -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 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