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

Improve Elmax code quality (#61273) 

Co-authored-by: Marvin Wichmann <marvin@fam-wichmann.de>
Co-authored-by: Franck Nijhof <frenck@frenck.nl>
Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
pull/62868/head
Alberto Geniola 2021-12-27 19:50:43 +01:00 committed by GitHub
parent 0d957ad93b
commit 1f425b1942
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 426 additions and 319 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

121
homeassistant/components/elmax/common.py
View File

@ -1,11 +1,9 @@
"""Elmax integration common classes and utilities."""
from __future__ import annotations
from collections.abc import Mapping
from datetime import timedelta
import logging
from logging import Logger
from typing import Any
import async_timeout
from elmax_api.exceptions import (
@ -15,20 +13,24 @@ from elmax_api.exceptions import (
ElmaxNetworkError,
)
from elmax_api.http import Elmax
from elmax_api.model.actuator import Actuator
from elmax_api.model.endpoint import DeviceEndpoint
from elmax_api.model.panel import PanelEntry, PanelStatus
from homeassistant.exceptions import ConfigEntryAuthFailed, HomeAssistantError
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.typing import HomeAssistantType
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
UpdateFailed,
)
from .const import DEFAULT_TIMEOUT, DOMAIN
_LOGGER = logging.getLogger(__name__)
class ElmaxCoordinator(DataUpdateCoordinator):
class ElmaxCoordinator(DataUpdateCoordinator[PanelStatus]):
"""Coordinator helper to handle Elmax API polling."""
def __init__(
@ -57,16 +59,11 @@ class ElmaxCoordinator(DataUpdateCoordinator):
"""Return the panel entry."""
return self._panel_entry
@property
def panel_status(self) -> PanelStatus | None:
"""Return the last fetched panel status."""
return self.data
def get_endpoint_state(self, endpoint_id: str) -> DeviceEndpoint | None:
"""Return the last fetched status for the given endpoint-id."""
def get_actuator_state(self, actuator_id: str) -> Actuator:
"""Return state of a specific actuator."""
if self._state_by_endpoint is not None:
return self._state_by_endpoint.get(endpoint_id)
return None
return self._state_by_endpoint.get(actuator_id)
raise HomeAssistantError("Unknown actuator")
@property
def http_client(self):
@ -78,18 +75,17 @@ class ElmaxCoordinator(DataUpdateCoordinator):
async with async_timeout.timeout(DEFAULT_TIMEOUT):
# Retrieve the panel online status first
panels = await self._client.list_control_panels()
panels = list(filter(lambda x: x.hash == self._panel_id, panels))
panel = next(
(panel for panel in panels if panel.hash == self._panel_id), None
)
# If the panel is no more available within the given. Raise config error as the user must
# reconfigure it in order to make it work again
if len(panels) < 1:
_LOGGER.error(
"Panel ID %s is no more linked to this user account",
self._panel_id,
if not panel:
raise ConfigEntryAuthFailed(
f"Panel ID {self._panel_id} is no more linked to this user account"
)
raise ConfigEntryAuthFailed()
panel = panels[0]
self._panel_entry = panel
# If the panel is online, proceed with fetching its state
@ -109,27 +105,22 @@ class ElmaxCoordinator(DataUpdateCoordinator):
return None
except ElmaxBadPinError as err:
_LOGGER.error("Control panel pin was refused")
raise ConfigEntryAuthFailed from err
raise ConfigEntryAuthFailed("Control panel pin was refused") from err
except ElmaxBadLoginError as err:
_LOGGER.error("Refused username/password")
raise ConfigEntryAuthFailed from err
raise ConfigEntryAuthFailed("Refused username/password") from err
except ElmaxApiError as err:
raise HomeAssistantError(
f"Error communicating with ELMAX API: {err}"
) from err
raise UpdateFailed(f"Error communicating with ELMAX API: {err}") from err
except ElmaxNetworkError as err:
raise HomeAssistantError(
"Network error occurred while contacting ELMAX cloud"
raise UpdateFailed(
"A network error occurred while communicating with Elmax cloud."
) from err
except Exception as err:
_LOGGER.exception("Unexpected exception")
raise HomeAssistantError("An unexpected error occurred") from err
class ElmaxEntity(Entity):
class ElmaxEntity(CoordinatorEntity):
"""Wrapper for Elmax entities."""
coordinator: ElmaxCoordinator
def __init__(
self,
panel: PanelEntry,
@ -138,21 +129,11 @@ class ElmaxEntity(Entity):
coordinator: ElmaxCoordinator,
) -> None:
"""Construct the object."""
super().__init__(coordinator=coordinator)
self._panel = panel
self._device = elmax_device
self._panel_version = panel_version
self._coordinator = coordinator
self._transitory_state = None
@property
def transitory_state(self) -> Any | None:
"""Return the transitory state for this entity."""
return self._transitory_state
@transitory_state.setter
def transitory_state(self, value: Any) -> None:
"""Set the transitory state value."""
self._transitory_state = value
self._client = coordinator.http_client
@property
def panel_id(self) -> str:
@ -169,21 +150,13 @@ class ElmaxEntity(Entity):
"""Return the entity name."""
return self._device.name
@property
def extra_state_attributes(self) -> Mapping[str, Any] | None:
"""Return extra attributes."""
return {
"index": self._device.index,
"visible": self._device.visible,
}
@property
def device_info(self):
"""Return device specific attributes."""
return {
"identifiers": {(DOMAIN, self._panel.hash)},
"name": self._panel.get_name_by_user(
self._coordinator.http_client.get_authenticated_username()
self.coordinator.http_client.get_authenticated_username()
),
"manufacturer": "Elmax",
"model": self._panel_version,
@ -192,39 +165,5 @@ class ElmaxEntity(Entity):
@property
def available(self) -> bool:
"""Return True if entity is available."""
return self._panel.online
def _http_data_changed(self) -> None:
# Whenever new HTTP data is received from the coordinator we extract the stat of this
# device and store it locally for later use
device_state = self._coordinator.get_endpoint_state(self._device.endpoint_id)
if self._device is None or device_state.__dict__ != self._device.__dict__:
# If HTTP data has changed, we need to schedule a forced refresh
self._device = device_state
self.async_schedule_update_ha_state(force_refresh=True)
# Reset the transitory state as we did receive a fresh state
self._transitory_state = None
@property
def should_poll(self) -> bool:
"""Return True if entity has to be polled for state.
False if entity pushes its state to HA.
"""
return False
async def async_added_to_hass(self) -> None:
"""Run when entity about to be added to hass.
To be extended by integrations.
"""
self._coordinator.async_add_listener(self._http_data_changed)
async def async_will_remove_from_hass(self) -> None:
"""Run when entity will be removed from hass.
To be extended by integrations.
"""
self._coordinator.async_remove_listener(self._http_data_changed)
"""Return if entity is available."""
return super().available and self._panel.online

135
homeassistant/components/elmax/config_flow.py
View File

@ -50,16 +50,13 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for elmax-cloud."""
VERSION = 1
def __init__(self):
"""Initialize."""
self._client: Elmax = None
self._username: str = None
self._password: str = None
self._panels_schema = None
self._panel_names = None
self._reauth_username = None
self._reauth_panelid = None
_client: Elmax
_username: str
_password: str
_panels_schema: vol.Schema
_panel_names: dict
_reauth_username: str | None
_reauth_panelid: str | None
async def async_step_user(
self, user_input: dict[str, Any] | None = None
@ -69,69 +66,73 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
if user_input is None:
return self.async_show_form(step_id="user", data_schema=LOGIN_FORM_SCHEMA)
errors: dict[str, str] = {}
username = user_input[CONF_ELMAX_USERNAME]
password = user_input[CONF_ELMAX_PASSWORD]
# Otherwise, it means we are handling now the "submission" of the user form.
# In this case, let's try to log in to the Elmax cloud and retrieve the available panels.
try:
client = Elmax(username=username, password=password)
await client.login()
# If the login succeeded, retrieve the list of available panels and filter the online ones
online_panels = [x for x in await client.list_control_panels() if x.online]
# If no online panel was found, we display an error in the next UI.
panels = list(online_panels)
if len(panels) < 1:
raise NoOnlinePanelsError()
# Show the panel selection.
# We want the user to choose the panel using the associated name, we set up a mapping
# dictionary to handle that case.
panel_names: dict[str, str] = {}
username = client.get_authenticated_username()
for panel in panels:
_store_panel_by_name(
panel=panel, username=username, panel_names=panel_names
)
self._client = client
self._panel_names = panel_names
schema = vol.Schema(
{
vol.Required(CONF_ELMAX_PANEL_NAME): vol.In(
self._panel_names.keys()
),
vol.Required(CONF_ELMAX_PANEL_PIN, default="000000"): str,
}
)
self._panels_schema = schema
self._username = username
self._password = password
return self.async_show_form(
step_id="panels", data_schema=schema, errors=errors
)
client = await self._async_login(username=username, password=password)
except ElmaxBadLoginError:
_LOGGER.error("Wrong credentials or failed login")
errors["base"] = "bad_auth"
except NoOnlinePanelsError:
_LOGGER.warning("No online device panel was found")
errors["base"] = "no_panel_online"
return self.async_show_form(
step_id="user",
data_schema=LOGIN_FORM_SCHEMA,
errors={"base": "invalid_auth"},
)
except ElmaxNetworkError:
_LOGGER.exception("A network error occurred")
errors["base"] = "network_error"
return self.async_show_form(
step_id="user",
data_schema=LOGIN_FORM_SCHEMA,
errors={"base": "network_error"},
)
# If an error occurred, show back the login form.
return self.async_show_form(
step_id="user", data_schema=LOGIN_FORM_SCHEMA, errors=errors
# If the login succeeded, retrieve the list of available panels and filter the online ones
online_panels = [x for x in await client.list_control_panels() if x.online]
# If no online panel was found, we display an error in the next UI.
if not online_panels:
return self.async_show_form(
step_id="user",
data_schema=LOGIN_FORM_SCHEMA,
errors={"base": "no_panel_online"},
)
# Show the panel selection.
# We want the user to choose the panel using the associated name, we set up a mapping
# dictionary to handle that case.
panel_names: dict[str, str] = {}
username = client.get_authenticated_username()
for panel in online_panels:
_store_panel_by_name(
panel=panel, username=username, panel_names=panel_names
)
self._client = client
self._panel_names = panel_names
schema = vol.Schema(
{
vol.Required(CONF_ELMAX_PANEL_NAME): vol.In(self._panel_names.keys()),
vol.Required(CONF_ELMAX_PANEL_PIN, default="000000"): str,
}
)
self._panels_schema = schema
self._username = username
self._password = password
# If everything went OK, proceed to panel selection.
return await self.async_step_panels(user_input=None)
async def async_step_panels(self, user_input: dict[str, Any]) -> FlowResult:
async def async_step_panels(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle Panel selection step."""
errors = {}
errors: dict[str, Any] = {}
if user_input is None:
return self.async_show_form(
step_id="panels", data_schema=self._panels_schema, errors=errors
)
panel_name = user_input[CONF_ELMAX_PANEL_NAME]
panel_pin = user_input[CONF_ELMAX_PANEL_PIN]
@ -160,7 +161,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
errors["base"] = "invalid_pin"
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Error occurred")
errors["base"] = "unknown_error"
errors["base"] = "unknown"
return self.async_show_form(
step_id="panels", data_schema=self._panels_schema, errors=errors
@ -184,8 +185,9 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
# and verify its pin is correct.
try:
# Test login.
client = Elmax(username=self._reauth_username, password=password)
await client.login()
client = await self._async_login(
username=self._reauth_username, password=password
)
# Make sure the panel we are authenticating to is still available.
panels = [
@ -220,7 +222,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
_LOGGER.error(
"Wrong credentials or failed login while re-authenticating"
)
errors["base"] = "bad_auth"
errors["base"] = "invalid_auth"
except NoOnlinePanelsError:
_LOGGER.warning(
"Panel ID %s is no longer associated to this user",
@ -245,6 +247,13 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
step_id="reauth_confirm", data_schema=schema, errors=errors
)
@staticmethod
async def _async_login(username: str, password: str) -> Elmax:
"""Log in to the Elmax cloud and return the http client."""
client = Elmax(username=username, password=password)
await client.login()
return client
class NoOnlinePanelsError(HomeAssistantError):
"""Error occurring when no online panel was found."""

7
homeassistant/components/elmax/strings.json
View File

@ -1,9 +1,7 @@
{
"title": "Elmax Cloud Setup",
"config": {
"step": {
"user": {
"title": "Account Login",
"description": "Please login to the Elmax cloud using your credentials",
"data": {
"password": "[%key:common::config_flow::data::password%]",
@ -11,7 +9,6 @@
}
},
"panels": {
"title": "Panel selection",
"description": "Select which panel you would like to control with this integration. Please note that the panel must be ON in order to be configured.",
"data": {
"panel_name": "Panel Name",
@ -22,10 +19,10 @@
},
"error": {
"no_panel_online": "No online Elmax control panel was found.",
"bad_auth": "Invalid authentication",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"network_error": "A network error occurred",
"invalid_pin": "The provided pin is invalid",
"unknown_error": "An unexpected error occurred"
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"

83
homeassistant/components/elmax/switch.py
View File

@ -1,4 +1,6 @@
"""Elmax switch platform."""
import asyncio
import logging
from typing import Any
from elmax_api.model.command import SwitchCommand
@ -13,39 +15,7 @@ from . import ElmaxCoordinator
from .common import ElmaxEntity
from .const import DOMAIN
class ElmaxSwitch(ElmaxEntity, SwitchEntity):
"""Implement the Elmax switch entity."""
@property
def is_on(self) -> bool:
"""Return True if entity is on."""
if self.transitory_state is not None:
return self.transitory_state
return self._device.opened
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the entity on."""
client = self._coordinator.http_client
await client.execute_command(
endpoint_id=self._device.endpoint_id, command=SwitchCommand.TURN_ON
)
self.transitory_state = True
await self.async_update_ha_state()
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the entity off."""
client = self._coordinator.http_client
await client.execute_command(
endpoint_id=self._device.endpoint_id, command=SwitchCommand.TURN_OFF
)
self.transitory_state = False
await self.async_update_ha_state()
@property
def assumed_state(self) -> bool:
"""Return True if unable to access real state of the entity."""
return False
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
@ -58,7 +28,7 @@ async def async_setup_entry(
known_devices = set()
def _discover_new_devices():
panel_status = coordinator.panel_status # type: PanelStatus
panel_status: PanelStatus = coordinator.data
# In case the panel is offline, its status will be None. In that case, simply do nothing
if panel_status is None:
return
@ -82,3 +52,48 @@ async def async_setup_entry(
# Immediately run a discovery, so we don't need to wait for the next update
_discover_new_devices()
class ElmaxSwitch(ElmaxEntity, SwitchEntity):
"""Implement the Elmax switch entity."""
@property
def is_on(self) -> bool:
"""Return True if entity is on."""
return self.coordinator.get_actuator_state(self._device.endpoint_id).opened
async def _wait_for_state_change(self) -> bool:
"""Refresh data and wait until the state state changes."""
old_state = self.coordinator.get_actuator_state(self._device.endpoint_id).opened
# Wait a bit at first to let Elmax cloud assimilate the new state.
await asyncio.sleep(2.0)
await self.coordinator.async_refresh()
new_state = self.coordinator.get_actuator_state(self._device.endpoint_id).opened
# First check attempt.
if new_state == old_state:
# Otherwise sleep a bit more and then trigger a final update.
await asyncio.sleep(5.0)
await self.coordinator.async_refresh()
new_state = self.coordinator.get_actuator_state(
self._device.endpoint_id
).opened
return new_state != old_state
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the entity on."""
await self.coordinator.http_client.execute_command(
endpoint_id=self._device.endpoint_id, command=SwitchCommand.TURN_ON
)
if await self._wait_for_state_change():
self.async_write_ha_state()
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the entity off."""
await self.coordinator.http_client.execute_command(
endpoint_id=self._device.endpoint_id, command=SwitchCommand.TURN_OFF
)
if await self._wait_for_state_change():
self.async_write_ha_state()

17
homeassistant/components/elmax/translations/en.json
View File

@ -1,14 +1,15 @@
{
"config": {
"abort": {
"already_configured": "Device is already configured"
"single_instance_allowed": "There already is an integration for that Elmaxc panel."
},
"error": {
"bad_auth": "Invalid authentication",
"invalid_auth": "Invalid authentication",
"invalid_pin": "The provided pin is invalid",
"network_error": "A network error occurred",
"no_panel_online": "No online Elmax control panel was found.",
"unknown_error": "An unexpected error occurred"
"reauth_panel_disappeared": "The panel is no longer associated to your account.",
"unknown": "Unexpected error"
},
"step": {
"panels": {
@ -20,6 +21,16 @@
"description": "Select which panel you would like to control with this integration. Please note that the panel must be ON in order to be configured.",
"title": "Panel selection"
},
"reauth_confirm": {
"data": {
"username": "Username",
"password": "Password",
"panel_id": "Panel ID",
"panel_pin": "PIN Code"
},
"description": "Please authenticate again to the Elmax cloud.",
"title": "Re-Authenticate"
},
"user": {
"data": {
"password": "Password",

382
tests/components/elmax/test_config_flow.py
View File

@ -1,4 +1,4 @@
"""Tests for the Abode config flow."""
"""Tests for the Elmax config flow."""
from unittest.mock import patch
from elmax_api.exceptions import ElmaxBadLoginError, ElmaxBadPinError, ElmaxNetworkError
@ -13,8 +13,8 @@ from homeassistant.components.elmax.const import (
DOMAIN,
)
from homeassistant.config_entries import SOURCE_REAUTH
from homeassistant.data_entry_flow import FlowResult
from tests.common import MockConfigEntry
from tests.components.elmax import (
MOCK_PANEL_ID,
MOCK_PANEL_NAME,
@ -26,78 +26,6 @@ from tests.components.elmax import (
CONF_POLLING = "polling"
def _has_error(errors):
return errors is not None and len(errors.keys()) > 0
async def _bootstrap(
hass,
source=config_entries.SOURCE_USER,
username=MOCK_USERNAME,
password=MOCK_PASSWORD,
panel_name=MOCK_PANEL_NAME,
panel_pin=MOCK_PANEL_PIN,
) -> FlowResult:
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": source}
)
if result["type"] != data_entry_flow.RESULT_TYPE_FORM or _has_error(
result["errors"]
):
return result
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_ELMAX_USERNAME: username,
CONF_ELMAX_PASSWORD: password,
},
)
if result2["type"] != data_entry_flow.RESULT_TYPE_FORM or _has_error(
result2["errors"]
):
return result2
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
{
CONF_ELMAX_PANEL_NAME: panel_name,
CONF_ELMAX_PANEL_PIN: panel_pin,
},
)
return result3
async def _reauth(hass):
# Trigger reauth
result2 = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_REAUTH},
data={
CONF_ELMAX_PANEL_ID: MOCK_PANEL_ID,
CONF_ELMAX_PANEL_PIN: MOCK_PANEL_PIN,
CONF_ELMAX_USERNAME: MOCK_USERNAME,
CONF_ELMAX_PASSWORD: MOCK_PASSWORD,
},
)
if result2["type"] != data_entry_flow.RESULT_TYPE_FORM or _has_error(
result2["errors"]
):
return result2
# Perform reauth confirm step
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
{
CONF_ELMAX_PANEL_ID: MOCK_PANEL_ID,
CONF_ELMAX_PANEL_PIN: MOCK_PANEL_PIN,
CONF_ELMAX_USERNAME: MOCK_USERNAME,
CONF_ELMAX_PASSWORD: MOCK_PASSWORD,
},
)
return result3
async def test_show_form(hass):
"""Test that the form is served with no input."""
result = await hass.config_entries.flow.async_init(
@ -107,16 +35,67 @@ async def test_show_form(hass):
assert result["step_id"] == "user"
async def test_standard_setup(hass):
"""Test the standard setup case."""
# Setup once.
show_form_result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch(
"homeassistant.components.elmax.async_setup_entry",
return_value=True,
):
login_result = await hass.config_entries.flow.async_configure(
show_form_result["flow_id"],
{
CONF_ELMAX_USERNAME: MOCK_USERNAME,
CONF_ELMAX_PASSWORD: MOCK_PASSWORD,
},
)
result = await hass.config_entries.flow.async_configure(
login_result["flow_id"],
{
CONF_ELMAX_PANEL_NAME: MOCK_PANEL_NAME,
CONF_ELMAX_PANEL_PIN: MOCK_PANEL_PIN,
},
)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
async def test_one_config_allowed(hass):
"""Test that only one Elmax configuration is allowed for each panel."""
# Setup once.
attempt1 = await _bootstrap(hass)
assert attempt1["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
MockConfigEntry(
domain=DOMAIN,
data={
CONF_ELMAX_PANEL_ID: MOCK_PANEL_ID,
CONF_ELMAX_USERNAME: MOCK_USERNAME,
CONF_ELMAX_PASSWORD: MOCK_PASSWORD,
CONF_ELMAX_PANEL_PIN: MOCK_PANEL_PIN,
},
unique_id=MOCK_PANEL_ID,
).add_to_hass(hass)
# Attempt to add another instance of the integration for the very same panel, it must fail.
attempt2 = await _bootstrap(hass)
assert attempt2["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert attempt2["reason"] == "already_configured"
show_form_result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
login_result = await hass.config_entries.flow.async_configure(
show_form_result["flow_id"],
{
CONF_ELMAX_USERNAME: MOCK_USERNAME,
CONF_ELMAX_PASSWORD: MOCK_PASSWORD,
},
)
result = await hass.config_entries.flow.async_configure(
login_result["flow_id"],
{
CONF_ELMAX_PANEL_NAME: MOCK_PANEL_NAME,
CONF_ELMAX_PANEL_PIN: MOCK_PANEL_PIN,
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
async def test_invalid_credentials(hass):
@ -125,12 +104,19 @@ async def test_invalid_credentials(hass):
"elmax_api.http.Elmax.login",
side_effect=ElmaxBadLoginError(),
):
result = await _bootstrap(
hass, username="wrong_user_name@email.com", password="incorrect_password"
show_form_result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["step_id"] == "user"
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {"base": "bad_auth"}
login_result = await hass.config_entries.flow.async_configure(
show_form_result["flow_id"],
{
CONF_ELMAX_USERNAME: "wrong_user_name@email.com",
CONF_ELMAX_PASSWORD: "incorrect_password",
},
)
assert login_result["step_id"] == "user"
assert login_result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert login_result["errors"] == {"base": "invalid_auth"}
async def test_connection_error(hass):
@ -139,12 +125,19 @@ async def test_connection_error(hass):
"elmax_api.http.Elmax.login",
side_effect=ElmaxNetworkError(),
):
result = await _bootstrap(
hass, username="wrong_user_name@email.com", password="incorrect_password"
show_form_result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["step_id"] == "user"
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {"base": "network_error"}
login_result = await hass.config_entries.flow.async_configure(
show_form_result["flow_id"],
{
CONF_ELMAX_USERNAME: MOCK_USERNAME,
CONF_ELMAX_PASSWORD: MOCK_PASSWORD,
},
)
assert login_result["step_id"] == "user"
assert login_result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert login_result["errors"] == {"base": "network_error"}
async def test_unhandled_error(hass):
@ -153,10 +146,26 @@ async def test_unhandled_error(hass):
"elmax_api.http.Elmax.get_panel_status",
side_effect=Exception(),
):
result = await _bootstrap(hass)
show_form_result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
login_result = await hass.config_entries.flow.async_configure(
show_form_result["flow_id"],
{
CONF_ELMAX_USERNAME: MOCK_USERNAME,
CONF_ELMAX_PASSWORD: MOCK_PASSWORD,
},
)
result = await hass.config_entries.flow.async_configure(
login_result["flow_id"],
{
CONF_ELMAX_PANEL_NAME: MOCK_PANEL_NAME,
CONF_ELMAX_PANEL_PIN: MOCK_PANEL_PIN,
},
)
assert result["step_id"] == "panels"
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {"base": "unknown_error"}
assert result["errors"] == {"base": "unknown"}
async def test_invalid_pin(hass):
@ -166,7 +175,23 @@ async def test_invalid_pin(hass):
"elmax_api.http.Elmax.get_panel_status",
side_effect=ElmaxBadPinError(),
):
result = await _bootstrap(hass)
show_form_result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
login_result = await hass.config_entries.flow.async_configure(
show_form_result["flow_id"],
{
CONF_ELMAX_USERNAME: MOCK_USERNAME,
CONF_ELMAX_PASSWORD: MOCK_PASSWORD,
},
)
result = await hass.config_entries.flow.async_configure(
login_result["flow_id"],
{
CONF_ELMAX_PANEL_NAME: MOCK_PANEL_NAME,
CONF_ELMAX_PANEL_PIN: MOCK_PANEL_PIN,
},
)
assert result["step_id"] == "panels"
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {"base": "invalid_pin"}
@ -179,22 +204,19 @@ async def test_no_online_panel(hass):
"elmax_api.http.Elmax.list_control_panels",
return_value=[],
):
result = await _bootstrap(hass)
assert result["step_id"] == "user"
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {"base": "no_panel_online"}
async def test_step_user(hass):
"""Test that the user step works."""
result = await _bootstrap(hass)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["data"] == {
CONF_ELMAX_PANEL_ID: MOCK_PANEL_ID,
CONF_ELMAX_PANEL_PIN: MOCK_PANEL_PIN,
CONF_ELMAX_USERNAME: MOCK_USERNAME,
CONF_ELMAX_PASSWORD: MOCK_PASSWORD,
}
show_form_result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
login_result = await hass.config_entries.flow.async_configure(
show_form_result["flow_id"],
{
CONF_ELMAX_USERNAME: MOCK_USERNAME,
CONF_ELMAX_PASSWORD: MOCK_PASSWORD,
},
)
assert login_result["step_id"] == "user"
assert login_result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert login_result["errors"] == {"base": "no_panel_online"}
async def test_show_reauth(hass):
@ -215,24 +237,84 @@ async def test_show_reauth(hass):
async def test_reauth_flow(hass):
"""Test that the reauth flow works."""
# Simulate a first setup
await _bootstrap(hass)
MockConfigEntry(
domain=DOMAIN,
data={
CONF_ELMAX_PANEL_ID: MOCK_PANEL_ID,
CONF_ELMAX_USERNAME: MOCK_USERNAME,
CONF_ELMAX_PASSWORD: MOCK_PASSWORD,
CONF_ELMAX_PANEL_PIN: MOCK_PANEL_PIN,
},
unique_id=MOCK_PANEL_ID,
).add_to_hass(hass)
# Trigger reauth
result = await _reauth(hass)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "reauth_successful"
with patch(
"homeassistant.components.elmax.async_setup_entry",
return_value=True,
):
reauth_result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_REAUTH},
data={
CONF_ELMAX_PANEL_ID: MOCK_PANEL_ID,
CONF_ELMAX_PANEL_PIN: MOCK_PANEL_PIN,
CONF_ELMAX_USERNAME: MOCK_USERNAME,
CONF_ELMAX_PASSWORD: MOCK_PASSWORD,
},
)
result = await hass.config_entries.flow.async_configure(
reauth_result["flow_id"],
{
CONF_ELMAX_PANEL_ID: MOCK_PANEL_ID,
CONF_ELMAX_PANEL_PIN: MOCK_PANEL_PIN,
CONF_ELMAX_USERNAME: MOCK_USERNAME,
CONF_ELMAX_PASSWORD: MOCK_PASSWORD,
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
await hass.async_block_till_done()
assert result["reason"] == "reauth_successful"
async def test_reauth_panel_disappeared(hass):
"""Test that the case where panel is no longer associated with the user."""
# Simulate a first setup
await _bootstrap(hass)
MockConfigEntry(
domain=DOMAIN,
data={
CONF_ELMAX_PANEL_ID: MOCK_PANEL_ID,
CONF_ELMAX_USERNAME: MOCK_USERNAME,
CONF_ELMAX_PASSWORD: MOCK_PASSWORD,
CONF_ELMAX_PANEL_PIN: MOCK_PANEL_PIN,
},
unique_id=MOCK_PANEL_ID,
).add_to_hass(hass)
# Trigger reauth
with patch(
"elmax_api.http.Elmax.list_control_panels",
return_value=[],
):
result = await _reauth(hass)
reauth_result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_REAUTH},
data={
CONF_ELMAX_PANEL_ID: MOCK_PANEL_ID,
CONF_ELMAX_PANEL_PIN: MOCK_PANEL_PIN,
CONF_ELMAX_USERNAME: MOCK_USERNAME,
CONF_ELMAX_PASSWORD: MOCK_PASSWORD,
},
)
result = await hass.config_entries.flow.async_configure(
reauth_result["flow_id"],
{
CONF_ELMAX_PANEL_ID: MOCK_PANEL_ID,
CONF_ELMAX_PANEL_PIN: MOCK_PANEL_PIN,
CONF_ELMAX_USERNAME: MOCK_USERNAME,
CONF_ELMAX_PASSWORD: MOCK_PASSWORD,
},
)
assert result["step_id"] == "reauth_confirm"
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {"base": "reauth_panel_disappeared"}
@ -240,14 +322,41 @@ async def test_reauth_panel_disappeared(hass):
async def test_reauth_invalid_pin(hass):
"""Test that the case where panel is no longer associated with the user."""
# Simulate a first setup
await _bootstrap(hass)
MockConfigEntry(
domain=DOMAIN,
data={
CONF_ELMAX_PANEL_ID: MOCK_PANEL_ID,
CONF_ELMAX_USERNAME: MOCK_USERNAME,
CONF_ELMAX_PASSWORD: MOCK_PASSWORD,
CONF_ELMAX_PANEL_PIN: MOCK_PANEL_PIN,
},
unique_id=MOCK_PANEL_ID,
).add_to_hass(hass)
# Trigger reauth
with patch(
"elmax_api.http.Elmax.get_panel_status",
side_effect=ElmaxBadPinError(),
):
result = await _reauth(hass)
reauth_result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_REAUTH},
data={
CONF_ELMAX_PANEL_ID: MOCK_PANEL_ID,
CONF_ELMAX_PANEL_PIN: MOCK_PANEL_PIN,
CONF_ELMAX_USERNAME: MOCK_USERNAME,
CONF_ELMAX_PASSWORD: MOCK_PASSWORD,
},
)
result = await hass.config_entries.flow.async_configure(
reauth_result["flow_id"],
{
CONF_ELMAX_PANEL_ID: MOCK_PANEL_ID,
CONF_ELMAX_PANEL_PIN: MOCK_PANEL_PIN,
CONF_ELMAX_USERNAME: MOCK_USERNAME,
CONF_ELMAX_PASSWORD: MOCK_PASSWORD,
},
)
assert result["step_id"] == "reauth_confirm"
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {"base": "invalid_pin"}
@ -255,14 +364,41 @@ async def test_reauth_invalid_pin(hass):
async def test_reauth_bad_login(hass):
"""Test bad login attempt at reauth time."""
# Simulate a first setup
await _bootstrap(hass)
MockConfigEntry(
domain=DOMAIN,
data={
CONF_ELMAX_PANEL_ID: MOCK_PANEL_ID,
CONF_ELMAX_USERNAME: MOCK_USERNAME,
CONF_ELMAX_PASSWORD: MOCK_PASSWORD,
CONF_ELMAX_PANEL_PIN: MOCK_PANEL_PIN,
},
unique_id=MOCK_PANEL_ID,
).add_to_hass(hass)
# Trigger reauth
with patch(
"elmax_api.http.Elmax.login",
side_effect=ElmaxBadLoginError(),
):
result = await _reauth(hass)
reauth_result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_REAUTH},
data={
CONF_ELMAX_PANEL_ID: MOCK_PANEL_ID,
CONF_ELMAX_PANEL_PIN: MOCK_PANEL_PIN,
CONF_ELMAX_USERNAME: MOCK_USERNAME,
CONF_ELMAX_PASSWORD: MOCK_PASSWORD,
},
)
result = await hass.config_entries.flow.async_configure(
reauth_result["flow_id"],
{
CONF_ELMAX_PANEL_ID: MOCK_PANEL_ID,
CONF_ELMAX_PANEL_PIN: MOCK_PANEL_PIN,
CONF_ELMAX_USERNAME: MOCK_USERNAME,
CONF_ELMAX_PASSWORD: MOCK_PASSWORD,
},
)
assert result["step_id"] == "reauth_confirm"
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {"base": "bad_auth"}
assert result["errors"] == {"base": "invalid_auth"}