Elmax integration (#59321)
* Add elmax integration. * Run hassfest and generate requirements_all * Remove secondary platforms from elmax integration as per first component integration. * Move ElmaxCoordinator and ElmaxEntity into external file Linting review * Remove useless variables * Fix wrong indentation. * Remove unecessary platforms. * Remove unnecessary attributes from manifest. * Rely on property getters/setters rathern than private attribute from parent. Update internal entity state just after transitory state update. * Update homeassistant/components/elmax/const.py Reference Platform constant Co-authored-by: Marvin Wichmann <marvin@fam-wichmann.de> * Update username/password values Rely on already-present templating constants Co-authored-by: Marvin Wichmann <marvin@fam-wichmann.de> * Add missing constant import. * Remove unnecessary test_unhandled_error() callback implementation. * Add common.py to coverage ignore list. * Improve coverage of config_flow. * Rename the integration. Co-authored-by: Franck Nijhof <frenck@frenck.nl> * Fix reauth bug and improve testing. * Refactor lambdas into generators. Co-authored-by: Marvin Wichmann <marvin@fam-wichmann.de> Co-authored-by: Franck Nijhof <frenck@frenck.nl>pull/61192/head
Alberto Geniola
GitHub
parent
7c7df5bb51
commit
b0affe7bfb
|
|
@ -257,6 +257,10 @@ omit =
|
|||
homeassistant/components/eight_sleep/*
|
||||
homeassistant/components/eliqonline/sensor.py
|
||||
homeassistant/components/elkm1/*
|
||||
homeassistant/components/elmax/__init__.py
|
||||
homeassistant/components/elmax/common.py
|
||||
homeassistant/components/elmax/const.py
|
||||
homeassistant/components/elmax/switch.py
|
||||
homeassistant/components/elv/*
|
||||
homeassistant/components/emby/media_player.py
|
||||
homeassistant/components/emoncms/sensor.py
|
||||
|
|
|
|||
|
|
@ -146,6 +146,7 @@ homeassistant/components/egardia/* @jeroenterheerdt
|
|||
homeassistant/components/eight_sleep/* @mezz64 @raman325
|
||||
homeassistant/components/elgato/* @frenck
|
||||
homeassistant/components/elkm1/* @gwww @bdraco
|
||||
homeassistant/components/elmax/* @albertogeniola
|
||||
homeassistant/components/elv/* @majuss
|
||||
homeassistant/components/emby/* @mezz64
|
||||
homeassistant/components/emoncms/* @borpin
|
||||
|
|
|
|||
|
|
@ -0,0 +1,56 @@
|
|||
"""The elmax-cloud integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.helpers.typing import HomeAssistantType
|
||||
|
||||
from .common import ElmaxCoordinator
|
||||
from .const import (
|
||||
CONF_ELMAX_PANEL_ID,
|
||||
CONF_ELMAX_PANEL_PIN,
|
||||
CONF_ELMAX_PASSWORD,
|
||||
CONF_ELMAX_USERNAME,
|
||||
DOMAIN,
|
||||
ELMAX_PLATFORMS,
|
||||
POLLING_SECONDS,
|
||||
)
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistantType, entry: ConfigEntry) -> bool:
|
||||
"""Set up elmax-cloud from a config entry."""
|
||||
# Create the API client object and attempt a login, so that we immediately know
|
||||
# if there is something wrong with user credentials
|
||||
coordinator = ElmaxCoordinator(
|
||||
hass=hass,
|
||||
logger=_LOGGER,
|
||||
username=entry.data[CONF_ELMAX_USERNAME],
|
||||
password=entry.data[CONF_ELMAX_PASSWORD],
|
||||
panel_id=entry.data[CONF_ELMAX_PANEL_ID],
|
||||
panel_pin=entry.data[CONF_ELMAX_PANEL_PIN],
|
||||
name=f"Elmax Cloud {entry.entry_id}",
|
||||
update_interval=timedelta(seconds=POLLING_SECONDS),
|
||||
)
|
||||
|
||||
# Issue a first refresh, so that we trigger a re-auth flow if necessary
|
||||
await coordinator.async_config_entry_first_refresh()
|
||||
|
||||
# Store a global reference to the coordinator for later use
|
||||
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
|
||||
|
||||
# Perform platform initialization.
|
||||
hass.config_entries.async_setup_platforms(entry, ELMAX_PLATFORMS)
|
||||
return True
|
||||
|
||||
|
||||
async def async_unload_entry(hass: HomeAssistantType, entry: ConfigEntry) -> bool:
|
||||
"""Unload a config entry."""
|
||||
unload_ok = await hass.config_entries.async_unload_platforms(entry, ELMAX_PLATFORMS)
|
||||
if unload_ok:
|
||||
hass.data[DOMAIN].pop(entry.entry_id)
|
||||
|
||||
return unload_ok
|
||||
|
|
@ -0,0 +1,229 @@
|
|||
"""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 (
|
||||
ElmaxApiError,
|
||||
ElmaxBadLoginError,
|
||||
ElmaxBadPinError,
|
||||
ElmaxNetworkError,
|
||||
)
|
||||
from elmax_api.http import Elmax
|
||||
from elmax_api.model.endpoint import DeviceEndpoint
|
||||
from elmax_api.model.panel import PanelEntry, PanelStatus
|
||||
|
||||
from homeassistant.components.elmax.const import DEFAULT_TIMEOUT, DOMAIN
|
||||
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
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class ElmaxCoordinator(DataUpdateCoordinator):
|
||||
"""Coordinator helper to handle Elmax API polling."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
hass: HomeAssistantType,
|
||||
logger: Logger,
|
||||
username: str,
|
||||
password: str,
|
||||
panel_id: str,
|
||||
panel_pin: str,
|
||||
name: str,
|
||||
update_interval: timedelta,
|
||||
) -> None:
|
||||
"""Instantiate the object."""
|
||||
self._client = Elmax(username=username, password=password)
|
||||
self._panel_id = panel_id
|
||||
self._panel_pin = panel_pin
|
||||
self._panel_entry = None
|
||||
self._state_by_endpoint = None
|
||||
super().__init__(
|
||||
hass=hass, logger=logger, name=name, update_interval=update_interval
|
||||
)
|
||||
|
||||
@property
|
||||
def panel_entry(self) -> PanelEntry | None:
|
||||
"""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."""
|
||||
if self._state_by_endpoint is not None:
|
||||
return self._state_by_endpoint.get(endpoint_id)
|
||||
return None
|
||||
|
||||
@property
|
||||
def http_client(self):
|
||||
"""Return the current http client being used by this instance."""
|
||||
return self._client
|
||||
|
||||
async def _async_update_data(self):
|
||||
try:
|
||||
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))
|
||||
|
||||
# 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,
|
||||
)
|
||||
raise ConfigEntryAuthFailed()
|
||||
|
||||
panel = panels[0]
|
||||
self._panel_entry = panel
|
||||
|
||||
# If the panel is online, proceed with fetching its state
|
||||
# and return it right away
|
||||
if panel.online:
|
||||
status = await self._client.get_panel_status(
|
||||
control_panel_id=panel.hash, pin=self._panel_pin
|
||||
) # type: PanelStatus
|
||||
|
||||
# Store a dictionary for fast endpoint state access
|
||||
self._state_by_endpoint = {
|
||||
k.endpoint_id: k for k in status.all_endpoints
|
||||
}
|
||||
return status
|
||||
|
||||
# Otherwise, return None. Listeners will know that this means the device is offline
|
||||
return None
|
||||
|
||||
except ElmaxBadPinError as err:
|
||||
_LOGGER.error("Control panel pin was refused")
|
||||
raise ConfigEntryAuthFailed from err
|
||||
except ElmaxBadLoginError as err:
|
||||
_LOGGER.error("Refused username/password")
|
||||
raise ConfigEntryAuthFailed from err
|
||||
except ElmaxApiError as err:
|
||||
raise HomeAssistantError(
|
||||
f"Error communicating with ELMAX API: {err}"
|
||||
) from err
|
||||
except ElmaxNetworkError as err:
|
||||
raise HomeAssistantError(
|
||||
"Network error occurred while contacting ELMAX cloud"
|
||||
) from err
|
||||
except Exception as err:
|
||||
_LOGGER.exception("Unexpected exception")
|
||||
raise HomeAssistantError("An unexpected error occurred") from err
|
||||
|
||||
|
||||
class ElmaxEntity(Entity):
|
||||
"""Wrapper for Elmax entities."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
panel: PanelEntry,
|
||||
elmax_device: DeviceEndpoint,
|
||||
panel_version: str,
|
||||
coordinator: ElmaxCoordinator,
|
||||
) -> None:
|
||||
"""Construct the object."""
|
||||
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
|
||||
|
||||
@property
|
||||
def panel_id(self) -> str:
|
||||
"""Retrieve the panel id."""
|
||||
return self._panel.hash
|
||||
|
||||
@property
|
||||
def unique_id(self) -> str | None:
|
||||
"""Provide a unique id for this entity."""
|
||||
return self._device.endpoint_id
|
||||
|
||||
@property
|
||||
def name(self) -> str | None:
|
||||
"""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()
|
||||
),
|
||||
"manufacturer": "Elmax",
|
||||
"model": self._panel_version,
|
||||
"sw_version": self._panel_version,
|
||||
}
|
||||
|
||||
@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)
|
||||
|
|
@ -0,0 +1,249 @@
|
|||
"""Config flow for elmax-cloud integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from elmax_api.exceptions import ElmaxBadLoginError, ElmaxBadPinError, ElmaxNetworkError
|
||||
from elmax_api.http import Elmax
|
||||
from elmax_api.model.panel import PanelEntry
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.components.elmax.const import (
|
||||
CONF_ELMAX_PANEL_ID,
|
||||
CONF_ELMAX_PANEL_NAME,
|
||||
CONF_ELMAX_PANEL_PIN,
|
||||
CONF_ELMAX_PASSWORD,
|
||||
CONF_ELMAX_USERNAME,
|
||||
DOMAIN,
|
||||
)
|
||||
from homeassistant.data_entry_flow import FlowResult
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
LOGIN_FORM_SCHEMA = vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_ELMAX_USERNAME): str,
|
||||
vol.Required(CONF_ELMAX_PASSWORD): str,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def _store_panel_by_name(
|
||||
panel: PanelEntry, username: str, panel_names: dict[str, str]
|
||||
) -> None:
|
||||
original_panel_name = panel.get_name_by_user(username=username)
|
||||
panel_id = panel.hash
|
||||
collisions_count = 0
|
||||
panel_name = original_panel_name
|
||||
while panel_name in panel_names:
|
||||
# Handle same-name collision.
|
||||
collisions_count += 1
|
||||
panel_name = f"{original_panel_name} ({collisions_count})"
|
||||
panel_names[panel_name] = panel_id
|
||||
|
||||
|
||||
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
|
||||
|
||||
async def async_step_user(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Handle a flow initialized by the user."""
|
||||
# When invokes without parameters, show the login form.
|
||||
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
|
||||
)
|
||||
|
||||
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"
|
||||
except ElmaxNetworkError:
|
||||
_LOGGER.exception("A network error occurred")
|
||||
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
|
||||
)
|
||||
|
||||
async def async_step_panels(self, user_input: dict[str, Any]) -> FlowResult:
|
||||
"""Handle Panel selection step."""
|
||||
errors = {}
|
||||
panel_name = user_input[CONF_ELMAX_PANEL_NAME]
|
||||
panel_pin = user_input[CONF_ELMAX_PANEL_PIN]
|
||||
|
||||
# Lookup the panel id from the panel name.
|
||||
panel_id = self._panel_names[panel_name]
|
||||
|
||||
# Make sure this is the only elmax integration for this specific panel id.
|
||||
await self.async_set_unique_id(panel_id)
|
||||
self._abort_if_unique_id_configured()
|
||||
|
||||
# Try to list all the devices using the given PIN.
|
||||
try:
|
||||
await self._client.get_panel_status(
|
||||
control_panel_id=panel_id, pin=panel_pin
|
||||
)
|
||||
return self.async_create_entry(
|
||||
title=f"Elmax {panel_name}",
|
||||
data={
|
||||
CONF_ELMAX_PANEL_ID: panel_id,
|
||||
CONF_ELMAX_PANEL_PIN: panel_pin,
|
||||
CONF_ELMAX_USERNAME: self._username,
|
||||
CONF_ELMAX_PASSWORD: self._password,
|
||||
},
|
||||
)
|
||||
except ElmaxBadPinError:
|
||||
errors["base"] = "invalid_pin"
|
||||
except Exception: # pylint: disable=broad-except
|
||||
_LOGGER.exception("Error occurred")
|
||||
errors["base"] = "unknown_error"
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="panels", data_schema=self._panels_schema, errors=errors
|
||||
)
|
||||
|
||||
async def async_step_reauth(self, user_input=None):
|
||||
"""Perform reauth upon an API authentication error."""
|
||||
self._reauth_username = user_input.get(CONF_ELMAX_USERNAME)
|
||||
self._reauth_panelid = user_input.get(CONF_ELMAX_PANEL_ID)
|
||||
return await self.async_step_reauth_confirm()
|
||||
|
||||
async def async_step_reauth_confirm(self, user_input=None):
|
||||
"""Handle reauthorization flow."""
|
||||
errors = {}
|
||||
if user_input is not None:
|
||||
panel_pin = user_input.get(CONF_ELMAX_PANEL_PIN)
|
||||
password = user_input.get(CONF_ELMAX_PASSWORD)
|
||||
entry = await self.async_set_unique_id(self._reauth_panelid)
|
||||
|
||||
# Handle authentication, make sure the panel we are re-authenticating against is listed among results
|
||||
# and verify its pin is correct.
|
||||
try:
|
||||
# Test login.
|
||||
client = Elmax(username=self._reauth_username, password=password)
|
||||
await client.login()
|
||||
|
||||
# Make sure the panel we are authenticating to is still available.
|
||||
panels = [
|
||||
p
|
||||
for p in await client.list_control_panels()
|
||||
if p.hash == self._reauth_panelid
|
||||
]
|
||||
if len(panels) < 1:
|
||||
raise NoOnlinePanelsError()
|
||||
|
||||
# Verify the pin is still valid.from
|
||||
await client.get_panel_status(
|
||||
control_panel_id=self._reauth_panelid, pin=panel_pin
|
||||
)
|
||||
|
||||
# If it is, proceed with configuration update.
|
||||
self.hass.config_entries.async_update_entry(
|
||||
entry,
|
||||
data={
|
||||
CONF_ELMAX_PANEL_ID: self._reauth_panelid,
|
||||
CONF_ELMAX_PANEL_PIN: panel_pin,
|
||||
CONF_ELMAX_USERNAME: self._reauth_username,
|
||||
CONF_ELMAX_PASSWORD: password,
|
||||
},
|
||||
)
|
||||
await self.hass.config_entries.async_reload(entry.entry_id)
|
||||
self._reauth_username = None
|
||||
self._reauth_panelid = None
|
||||
return self.async_abort(reason="reauth_successful")
|
||||
|
||||
except ElmaxBadLoginError:
|
||||
_LOGGER.error(
|
||||
"Wrong credentials or failed login while re-authenticating"
|
||||
)
|
||||
errors["base"] = "bad_auth"
|
||||
except NoOnlinePanelsError:
|
||||
_LOGGER.warning(
|
||||
"Panel ID %s is no longer associated to this user",
|
||||
self._reauth_panelid,
|
||||
)
|
||||
errors["base"] = "reauth_panel_disappeared"
|
||||
except ElmaxBadPinError:
|
||||
errors["base"] = "invalid_pin"
|
||||
|
||||
# We want the user to re-authenticate only for the given panel id using the same login.
|
||||
# We pin them to the UI, so the user realizes she must log in with the appropriate credentials
|
||||
# for the that specific panel.
|
||||
schema = vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_ELMAX_USERNAME): self._reauth_username,
|
||||
vol.Required(CONF_ELMAX_PASSWORD): str,
|
||||
vol.Required(CONF_ELMAX_PANEL_ID): self._reauth_panelid,
|
||||
vol.Required(CONF_ELMAX_PANEL_PIN): str,
|
||||
}
|
||||
)
|
||||
return self.async_show_form(
|
||||
step_id="reauth_confirm", data_schema=schema, errors=errors
|
||||
)
|
||||
|
||||
|
||||
class NoOnlinePanelsError(HomeAssistantError):
|
||||
"""Error occurring when no online panel was found."""
|
||||
|
|
@ -0,0 +1,17 @@
|
|||
"""Constants for the elmax-cloud integration."""
|
||||
from homeassistant.const import Platform
|
||||
|
||||
DOMAIN = "elmax"
|
||||
CONF_ELMAX_USERNAME = "username"
|
||||
CONF_ELMAX_PASSWORD = "password"
|
||||
CONF_ELMAX_PANEL_ID = "panel_id"
|
||||
CONF_ELMAX_PANEL_PIN = "panel_pin"
|
||||
CONF_ELMAX_PANEL_NAME = "panel_name"
|
||||
|
||||
CONF_CONFIG_ENTRY_ID = "config_entry_id"
|
||||
CONF_ENDPOINT_ID = "endpoint_id"
|
||||
|
||||
ELMAX_PLATFORMS = [Platform.SWITCH]
|
||||
|
||||
POLLING_SECONDS = 30
|
||||
DEFAULT_TIMEOUT = 10.0
|
||||
|
|
@ -0,0 +1,11 @@
|
|||
{
|
||||
"domain": "elmax",
|
||||
"name": "Elmax",
|
||||
"config_flow": true,
|
||||
"documentation": "https://www.home-assistant.io/integrations/elmax",
|
||||
"requirements": ["elmax_api==0.0.2"],
|
||||
"codeowners": [
|
||||
"@albertogeniola"
|
||||
],
|
||||
"iot_class": "cloud_polling"
|
||||
}
|
||||
|
|
@ -0,0 +1,34 @@
|
|||
{
|
||||
"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%]",
|
||||
"username": "[%key:common::config_flow::data::username%]"
|
||||
}
|
||||
},
|
||||
"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",
|
||||
"panel_id": "Panel ID",
|
||||
"panel_pin": "PIN Code"
|
||||
}
|
||||
}
|
||||
},
|
||||
"error": {
|
||||
"no_panel_online": "No online Elmax control panel was found.",
|
||||
"bad_auth": "Invalid authentication",
|
||||
"network_error": "A network error occurred",
|
||||
"invalid_pin": "The provided pin is invalid",
|
||||
"unknown_error": "An unexpected error occurred"
|
||||
},
|
||||
"abort": {
|
||||
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,83 @@
|
|||
"""Elmax switch platform."""
|
||||
from typing import Any
|
||||
|
||||
from elmax_api.model.command import SwitchCommand
|
||||
from elmax_api.model.panel import PanelStatus
|
||||
|
||||
from homeassistant.components.elmax import ElmaxCoordinator
|
||||
from homeassistant.components.elmax.common import ElmaxEntity
|
||||
from homeassistant.components.elmax.const import DOMAIN
|
||||
from homeassistant.components.switch import SwitchEntity
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.helpers.typing import HomeAssistantType
|
||||
|
||||
|
||||
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
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistantType,
|
||||
config_entry: ConfigEntry,
|
||||
async_add_entities: AddEntitiesCallback,
|
||||
) -> None:
|
||||
"""Set up the Elmax switch platform."""
|
||||
coordinator: ElmaxCoordinator = hass.data[DOMAIN][config_entry.entry_id]
|
||||
known_devices = set()
|
||||
|
||||
def _discover_new_devices():
|
||||
panel_status = coordinator.panel_status # type: PanelStatus
|
||||
# In case the panel is offline, its status will be None. In that case, simply do nothing
|
||||
if panel_status is None:
|
||||
return
|
||||
|
||||
# Otherwise, add all the entities we found
|
||||
entities = []
|
||||
for actuator in panel_status.actuators:
|
||||
entity = ElmaxSwitch(
|
||||
panel=coordinator.panel_entry,
|
||||
elmax_device=actuator,
|
||||
panel_version=panel_status.release,
|
||||
coordinator=coordinator,
|
||||
)
|
||||
if entity.unique_id not in known_devices:
|
||||
entities.append(entity)
|
||||
async_add_entities(entities, True)
|
||||
known_devices.update([entity.unique_id for entity in entities])
|
||||
|
||||
# Register a listener for the discovery of new devices
|
||||
coordinator.async_add_listener(_discover_new_devices)
|
||||
|
||||
# Immediately run a discovery, so we don't need to wait for the next update
|
||||
_discover_new_devices()
|
||||
|
|
@ -0,0 +1,45 @@
|
|||
{
|
||||
"config": {
|
||||
"abort": {
|
||||
"single_instance_allowed": "There already is an integration for that Elmaxc panel."
|
||||
},
|
||||
"error": {
|
||||
"bad_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."
|
||||
},
|
||||
"step": {
|
||||
"panels": {
|
||||
"data": {
|
||||
"panel_id": "Panel ID",
|
||||
"panel_name": "Panel Name",
|
||||
"panel_pin": "PIN Code"
|
||||
},
|
||||
"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",
|
||||
"username": "Username"
|
||||
},
|
||||
"description": "Please login to the Elmax cloud using your credentials",
|
||||
"title": "Account Login"
|
||||
}
|
||||
}
|
||||
},
|
||||
"title": "Elmax Cloud Setup"
|
||||
}
|
||||
|
|
@ -78,6 +78,7 @@ FLOWS = [
|
|||
"efergy",
|
||||
"elgato",
|
||||
"elkm1",
|
||||
"elmax",
|
||||
"emonitor",
|
||||
"emulated_roku",
|
||||
"enocean",
|
||||
|
|
|
|||
|
|
@ -590,6 +590,9 @@ eliqonline==1.2.2
|
|||
# homeassistant.components.elkm1
|
||||
elkm1-lib==1.0.0
|
||||
|
||||
# homeassistant.components.elmax
|
||||
elmax_api==0.0.2
|
||||
|
||||
# homeassistant.components.mobile_app
|
||||
emoji==1.5.0
|
||||
|
||||
|
|
|
|||
|
|
@ -368,6 +368,9 @@ elgato==2.2.0
|
|||
# homeassistant.components.elkm1
|
||||
elkm1-lib==1.0.0
|
||||
|
||||
# homeassistant.components.elmax
|
||||
elmax_api==0.0.2
|
||||
|
||||
# homeassistant.components.mobile_app
|
||||
emoji==1.5.0
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,15 @@
|
|||
"""Tests for the Elmax component."""
|
||||
|
||||
MOCK_USER_JWT = (
|
||||
"JWT eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9"
|
||||
".eyJfaWQiOiIxYjExYmIxMWJiYjExMTExYjFiMTFiMWIiLCJlbWFpbCI6InRoaXMuaXNAdGVzdC5jb20iLCJyb2xlIjoid"
|
||||
"XNlciIsImlhdCI6MTYzNjE5OTk5OCwiZXhwIjoxNjM2MjM1OTk4fQ.1C7lXuKyX1HEGOfMxNwxJ2n-CjoW4rwvNRITQxLI"
|
||||
"Cv0"
|
||||
)
|
||||
MOCK_USERNAME = "this.is@test.com"
|
||||
MOCK_USER_ROLE = "user"
|
||||
MOCK_USER_ID = "1b11bb11bbb11111b1b11b1b"
|
||||
MOCK_PANEL_ID = "2db3dae30b9102de4d078706f94d0708"
|
||||
MOCK_PANEL_NAME = "Test Panel Name"
|
||||
MOCK_PANEL_PIN = "000000"
|
||||
MOCK_PASSWORD = "password"
|
||||
|
|
@ -0,0 +1,42 @@
|
|||
"""Configuration for Elmax tests."""
|
||||
import json
|
||||
|
||||
from elmax_api.constants import (
|
||||
BASE_URL,
|
||||
ENDPOINT_DEVICES,
|
||||
ENDPOINT_DISCOVERY,
|
||||
ENDPOINT_LOGIN,
|
||||
)
|
||||
from httpx import Response
|
||||
import pytest
|
||||
import respx
|
||||
|
||||
from tests.common import load_fixture
|
||||
from tests.components.elmax import MOCK_PANEL_ID, MOCK_PANEL_PIN
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def httpx_mock_fixture(requests_mock):
|
||||
"""Configure httpx fixture."""
|
||||
with respx.mock(base_url=BASE_URL, assert_all_called=False) as respx_mock:
|
||||
# Mock Login POST.
|
||||
login_route = respx_mock.post(f"/{ENDPOINT_LOGIN}", name="login")
|
||||
login_route.return_value = Response(
|
||||
200, json=json.loads(load_fixture("login.json", "elmax"))
|
||||
)
|
||||
|
||||
# Mock Device list GET.
|
||||
list_devices_route = respx_mock.get(f"/{ENDPOINT_DEVICES}", name="list_devices")
|
||||
list_devices_route.return_value = Response(
|
||||
200, json=json.loads(load_fixture("list_devices.json", "elmax"))
|
||||
)
|
||||
|
||||
# Mock Panel GET.
|
||||
get_panel_route = respx_mock.get(
|
||||
f"/{ENDPOINT_DISCOVERY}/{MOCK_PANEL_ID}/{MOCK_PANEL_PIN}", name="get_panel"
|
||||
)
|
||||
get_panel_route.return_value = Response(
|
||||
200, json=json.loads(load_fixture("get_panel.json", "elmax"))
|
||||
)
|
||||
|
||||
yield respx_mock
|
||||
|
|
@ -0,0 +1,126 @@
|
|||
{
|
||||
"release": 11.7,
|
||||
"tappFeature": true,
|
||||
"sceneFeature": true,
|
||||
"zone": [
|
||||
{
|
||||
"endpointId": "2db3dae30b9102de4d078706f94d0708-zona-0",
|
||||
"visibile": true,
|
||||
"indice": 0,
|
||||
"nome": "Feed zone 0",
|
||||
"aperta": false,
|
||||
"esclusa": false
|
||||
},
|
||||
{
|
||||
"endpointId": "2db3dae30b9102de4d078706f94d0708-zona-1",
|
||||
"visibile": true,
|
||||
"indice": 1,
|
||||
"nome": "Feed Zone 1",
|
||||
"aperta": false,
|
||||
"esclusa": false
|
||||
},
|
||||
{
|
||||
"endpointId": "2db3dae30b9102de4d078706f94d0708-zona-2",
|
||||
"visibile": true,
|
||||
"indice": 2,
|
||||
"nome": "Feed Zone 2",
|
||||
"aperta": false,
|
||||
"esclusa": false
|
||||
}
|
||||
],
|
||||
"uscite": [
|
||||
{
|
||||
"endpointId": "2db3dae30b9102de4d078706f94d0708-uscita-0",
|
||||
"visibile": true,
|
||||
"indice": 0,
|
||||
"nome": "Actuator 0",
|
||||
"aperta": false
|
||||
},
|
||||
{
|
||||
"endpointId": "2db3dae30b9102de4d078706f94d0708-uscita-1",
|
||||
"visibile": true,
|
||||
"indice": 1,
|
||||
"nome": "Actuator 1",
|
||||
"aperta": false
|
||||
},
|
||||
{
|
||||
"endpointId": "2db3dae30b9102de4d078706f94d0708-uscita-2",
|
||||
"visibile": true,
|
||||
"indice": 2,
|
||||
"nome": "Actuator 2",
|
||||
"aperta": true
|
||||
}
|
||||
],
|
||||
"aree": [
|
||||
{
|
||||
"endpointId": "2db3dae30b9102de4d078706f94d0708-area-0",
|
||||
"visibile": true,
|
||||
"indice": 0,
|
||||
"nome": "AREA 0",
|
||||
"statiDisponibili": [0, 1, 2, 3, 4],
|
||||
"statiSessioneDisponibili": [0, 1, 2, 3],
|
||||
"stato": 0,
|
||||
"statoSessione": 0
|
||||
},
|
||||
{
|
||||
"endpointId": "2db3dae30b9102de4d078706f94d0708-area-1",
|
||||
"visibile": true,
|
||||
"indice": 1,
|
||||
"nome": "AREA 1",
|
||||
"statiDisponibili": [0, 1, 2, 3, 4],
|
||||
"statiSessioneDisponibili": [0, 1, 2, 3],
|
||||
"stato": 0,
|
||||
"statoSessione": 0
|
||||
},
|
||||
{
|
||||
"endpointId": "2db3dae30b9102de4d078706f94d0708-area-2",
|
||||
"visibile": false,
|
||||
"indice": 2,
|
||||
"nome": "AREA 2",
|
||||
"statiDisponibili": [0, 1, 2, 3, 4],
|
||||
"statiSessioneDisponibili": [0, 1, 2, 3],
|
||||
"stato": 0,
|
||||
"statoSessione": 0
|
||||
}
|
||||
],
|
||||
"tapparelle": [
|
||||
{
|
||||
"endpointId": "2db3dae30b9102de4d078706f94d0708-tapparella-0",
|
||||
"visibile": true,
|
||||
"indice": 0,
|
||||
"stato": "stop",
|
||||
"posizione": 100,
|
||||
"nome": "Cover 0"
|
||||
}
|
||||
],
|
||||
"gruppi": [
|
||||
{
|
||||
"endpointId": "2db3dae30b9102de4d078706f94d0708-gruppo-0",
|
||||
"visibile": true,
|
||||
"indice": 0,
|
||||
"nome": "Group 0"
|
||||
},
|
||||
{
|
||||
"endpointId": "2db3dae30b9102de4d078706f94d0708-gruppo-1",
|
||||
"visibile": false,
|
||||
"indice": 1,
|
||||
"nome": "Group 1"
|
||||
}
|
||||
],
|
||||
"scenari": [
|
||||
{
|
||||
"endpointId": "2db3dae30b9102de4d078706f94d0708-scenario-0",
|
||||
"visibile": true,
|
||||
"indice": 0,
|
||||
"nome": "Automation 0"
|
||||
},
|
||||
{
|
||||
"endpointId": "2db3dae30b9102de4d078706f94d0708-scenario-2",
|
||||
"visibile": true,
|
||||
"indice": 2,
|
||||
"nome": "Automation 2"
|
||||
}
|
||||
],
|
||||
"utente": "this.is@test.com",
|
||||
"centrale": "2db3dae30b9102de4d078706f94d0708"
|
||||
}
|
||||
|
|
@ -0,0 +1,11 @@
|
|||
[
|
||||
{
|
||||
"centrale_online": true,
|
||||
"hash": "2db3dae30b9102de4d078706f94d0708",
|
||||
"username": [{"name": "this.is@test.com", "label": "Test Panel Name"}]
|
||||
},{
|
||||
"centrale_online": true,
|
||||
"hash": "d8e8fca2dc0f896fd7cb4cb0031ba249",
|
||||
"username": [{"name": "this.is@test.com", "label": "Test Panel Name"}]
|
||||
}
|
||||
]
|
||||
|
|
@ -0,0 +1,8 @@
|
|||
{
|
||||
"token": "JWT eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJfaWQiOiIxYjExYmIxMWJiYjExMTExYjFiMTFiMWIiLCJlbWFpbCI6InRoaXMuaXNAdGVzdC5jb20iLCJyb2xlIjoidXNlciIsImlhdCI6MTYzNjE5OTk5OCwiZXhwIjoxNjM2MjM1OTk4fQ.1C7lXuKyX1HEGOfMxNwxJ2n-CjoW4rwvNRITQxLICv0",
|
||||
"user": {
|
||||
"_id": "1b11bb11bbb11111b1b11b1b",
|
||||
"email": "this.is@test.com",
|
||||
"role": "user"
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,268 @@
|
|||
"""Tests for the Abode config flow."""
|
||||
from unittest.mock import patch
|
||||
|
||||
from elmax_api.exceptions import ElmaxBadLoginError, ElmaxBadPinError, ElmaxNetworkError
|
||||
|
||||
from homeassistant import config_entries, data_entry_flow
|
||||
from homeassistant.components.elmax.const import (
|
||||
CONF_ELMAX_PANEL_ID,
|
||||
CONF_ELMAX_PANEL_NAME,
|
||||
CONF_ELMAX_PANEL_PIN,
|
||||
CONF_ELMAX_PASSWORD,
|
||||
CONF_ELMAX_USERNAME,
|
||||
DOMAIN,
|
||||
)
|
||||
from homeassistant.config_entries import SOURCE_REAUTH
|
||||
from homeassistant.data_entry_flow import FlowResult
|
||||
|
||||
from tests.components.elmax import (
|
||||
MOCK_PANEL_ID,
|
||||
MOCK_PANEL_NAME,
|
||||
MOCK_PANEL_PIN,
|
||||
MOCK_PASSWORD,
|
||||
MOCK_USERNAME,
|
||||
)
|
||||
|
||||
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(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "user"
|
||||
|
||||
|
||||
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
|
||||
|
||||
# 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"
|
||||
|
||||
|
||||
async def test_invalid_credentials(hass):
|
||||
"""Test that invalid credentials throws an error."""
|
||||
with patch(
|
||||
"elmax_api.http.Elmax.login",
|
||||
side_effect=ElmaxBadLoginError(),
|
||||
):
|
||||
result = await _bootstrap(
|
||||
hass, username="wrong_user_name@email.com", password="incorrect_password"
|
||||
)
|
||||
assert result["step_id"] == "user"
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["errors"] == {"base": "bad_auth"}
|
||||
|
||||
|
||||
async def test_connection_error(hass):
|
||||
"""Test other than invalid credentials throws an error."""
|
||||
with patch(
|
||||
"elmax_api.http.Elmax.login",
|
||||
side_effect=ElmaxNetworkError(),
|
||||
):
|
||||
result = await _bootstrap(
|
||||
hass, username="wrong_user_name@email.com", password="incorrect_password"
|
||||
)
|
||||
assert result["step_id"] == "user"
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["errors"] == {"base": "network_error"}
|
||||
|
||||
|
||||
async def test_unhandled_error(hass):
|
||||
"""Test unhandled exceptions."""
|
||||
with patch(
|
||||
"elmax_api.http.Elmax.get_panel_status",
|
||||
side_effect=Exception(),
|
||||
):
|
||||
result = await _bootstrap(hass)
|
||||
assert result["step_id"] == "panels"
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["errors"] == {"base": "unknown_error"}
|
||||
|
||||
|
||||
async def test_invalid_pin(hass):
|
||||
"""Test error is thrown when a wrong pin is used to pair a panel."""
|
||||
# Simulate bad pin response.
|
||||
with patch(
|
||||
"elmax_api.http.Elmax.get_panel_status",
|
||||
side_effect=ElmaxBadPinError(),
|
||||
):
|
||||
result = await _bootstrap(hass)
|
||||
assert result["step_id"] == "panels"
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["errors"] == {"base": "invalid_pin"}
|
||||
|
||||
|
||||
async def test_no_online_panel(hass):
|
||||
"""Test no-online panel is available."""
|
||||
# Simulate low-level api returns no panels.
|
||||
with patch(
|
||||
"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,
|
||||
}
|
||||
|
||||
|
||||
async def test_show_reauth(hass):
|
||||
"""Test that the reauth form shows."""
|
||||
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,
|
||||
},
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "reauth_confirm"
|
||||
|
||||
|
||||
async def test_reauth_flow(hass):
|
||||
"""Test that the reauth flow works."""
|
||||
# Simulate a first setup
|
||||
await _bootstrap(hass)
|
||||
# Trigger reauth
|
||||
result = await _reauth(hass)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
|
||||
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)
|
||||
# Trigger reauth
|
||||
with patch(
|
||||
"elmax_api.http.Elmax.list_control_panels",
|
||||
return_value=[],
|
||||
):
|
||||
result = await _reauth(hass)
|
||||
assert result["step_id"] == "reauth_confirm"
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["errors"] == {"base": "reauth_panel_disappeared"}
|
||||
|
||||
|
||||
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)
|
||||
# Trigger reauth
|
||||
with patch(
|
||||
"elmax_api.http.Elmax.get_panel_status",
|
||||
side_effect=ElmaxBadPinError(),
|
||||
):
|
||||
result = await _reauth(hass)
|
||||
assert result["step_id"] == "reauth_confirm"
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["errors"] == {"base": "invalid_pin"}
|
||||
|
||||
|
||||
async def test_reauth_bad_login(hass):
|
||||
"""Test bad login attempt at reauth time."""
|
||||
# Simulate a first setup
|
||||
await _bootstrap(hass)
|
||||
# Trigger reauth
|
||||
with patch(
|
||||
"elmax_api.http.Elmax.login",
|
||||
side_effect=ElmaxBadLoginError(),
|
||||
):
|
||||
result = await _reauth(hass)
|
||||
assert result["step_id"] == "reauth_confirm"
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["errors"] == {"base": "bad_auth"}
|
||||
Loading…
Reference in New Issue