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

Add local API support to elmax (#94392)

Browse Source 
* Add support for local (lan) panel integration

* Fix merge conflicts

* Remove executable flag from non-executable files

* Fix tests

* Update homeassistant/components/elmax/__init__.py

Shorten comment

Co-authored-by: Erik Montnemery <erik@montnemery.com>

* Fix typehint

* Rename DummyPanel into DirectPanel

* Update homeassistant/components/elmax/__init__.py

Rewording

Co-authored-by: Erik Montnemery <erik@montnemery.com>

* Update homeassistant/components/elmax/__init__.py

Rewording

Co-authored-by: Erik Montnemery <erik@montnemery.com>

* Refactor option step into menu step

* Change requirement statement

* Refactor dictionary key entries to use existing constants

* Align step names to new constants

* Align step names to new constants amd align tests

* Align step names to new constants amd align tests

* Align step names to new constants

* Simplify logic to handle entire entry instead of a portion of the state

* Simplify working mode checks

* Add data_description dictionary to better explain SSL and FOLLOW_MDSN options

* Add support for local (lan) panel integration

* Fix merge conflicts

* Remove executable flag from non-executable files

* Fix tests

* Update homeassistant/components/elmax/__init__.py

Shorten comment

Co-authored-by: Erik Montnemery <erik@montnemery.com>

* Fix typehint

* Rename DummyPanel into DirectPanel

* Update homeassistant/components/elmax/__init__.py

Rewording

Co-authored-by: Erik Montnemery <erik@montnemery.com>

* Update homeassistant/components/elmax/__init__.py

Rewording

Co-authored-by: Erik Montnemery <erik@montnemery.com>

* Refactor option step into menu step

* Change requirement statement

* Refactor dictionary key entries to use existing constants

* Align step names to new constants

* Align step names to new constants amd align tests

* Align step names to new constants amd align tests

* Align step names to new constants

* Simplify logic to handle entire entry instead of a portion of the state

* Simplify working mode checks

* Add data_description dictionary to better explain SSL and FOLLOW_MDSN options

* Add newline at end of file

* Remove CONF_ELMAX_MODE_DIRECT_FOLLOW_MDNS option

* Fix Ruff pre-check

---------

Co-authored-by: Erik Montnemery <erik@montnemery.com>
This commit is contained in:
Alberto Geniola 2024-03-04 11:39:13 +01:00 committed by GitHub
parent 4c67670566
commit 86039de3cd
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
20 changed files with 1242 additions and 106 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

88
homeassistant/components/elmax/__init__.py
View File

@ -4,11 +4,28 @@ from __future__ import annotations
from datetime import timedelta
import logging
from elmax_api.exceptions import ElmaxBadLoginError
from elmax_api.http import Elmax, ElmaxLocal, GenericElmax
from elmax_api.model.panel import PanelEntry
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from .common import ElmaxCoordinator
from .common import (
DirectPanel,
ElmaxCoordinator,
build_direct_ssl_context,
get_direct_api_url,
)
from .const import (
CONF_ELMAX_MODE,
CONF_ELMAX_MODE_CLOUD,
CONF_ELMAX_MODE_DIRECT,
CONF_ELMAX_MODE_DIRECT_HOST,
CONF_ELMAX_MODE_DIRECT_PORT,
CONF_ELMAX_MODE_DIRECT_SSL,
CONF_ELMAX_MODE_DIRECT_SSL_CERT,
CONF_ELMAX_PANEL_ID,
CONF_ELMAX_PANEL_PIN,
CONF_ELMAX_PASSWORD,
@ -21,17 +38,71 @@ from .const import (
_LOGGER = logging.getLogger(__name__)
async def _load_elmax_panel_client(
entry: ConfigEntry,
) -> tuple[GenericElmax, PanelEntry]:
# Connection mode was not present in initial version, default to cloud if not set
mode = entry.data.get(CONF_ELMAX_MODE, CONF_ELMAX_MODE_CLOUD)
if mode == CONF_ELMAX_MODE_DIRECT:
client_api_url = get_direct_api_url(
host=entry.data[CONF_ELMAX_MODE_DIRECT_HOST],
port=entry.data[CONF_ELMAX_MODE_DIRECT_PORT],
use_ssl=entry.data[CONF_ELMAX_MODE_DIRECT_SSL],
)
custom_ssl_context = None
custom_ssl_cert = entry.data.get(CONF_ELMAX_MODE_DIRECT_SSL_CERT)
if custom_ssl_cert:
custom_ssl_context = build_direct_ssl_context(cadata=custom_ssl_cert)
client = ElmaxLocal(
panel_api_url=client_api_url,
panel_code=entry.data[CONF_ELMAX_PANEL_PIN],
ssl_context=custom_ssl_context,
)
panel = DirectPanel(panel_uri=client_api_url)
else:
client = Elmax(
username=entry.data[CONF_ELMAX_USERNAME],
password=entry.data[CONF_ELMAX_PASSWORD],
)
client.set_current_panel(
entry.data[CONF_ELMAX_PANEL_ID], entry.data[CONF_ELMAX_PANEL_PIN]
)
# Make sure the panel is online and assigned to the current user
panel = await _check_cloud_panel_status(client, entry.data[CONF_ELMAX_PANEL_ID])
return client, panel
async def _check_cloud_panel_status(client: Elmax, panel_id: str) -> PanelEntry:
"""Perform integrity checks against the cloud for panel-user association."""
# Retrieve the panel online status first
panels = await client.list_control_panels()
panel = next((panel for panel in panels if panel.hash == panel_id), None)
# If the panel is no longer available within the ones associated to that client, raise
# a config error as the user must reconfigure it in order to make it work again
if not panel:
raise ConfigEntryAuthFailed(
f"Panel ID {panel_id} is no longer linked to this user account"
)
return panel
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up elmax-cloud from a config entry."""
try:
client, panel = await _load_elmax_panel_client(entry)
except ElmaxBadLoginError as err:
raise ConfigEntryAuthFailed() from err
# 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],
elmax_api_client=client,
panel=panel,
name=f"Elmax Cloud {entry.entry_id}",
update_interval=timedelta(seconds=POLLING_SECONDS),
)
@ -42,11 +113,18 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
# Store a global reference to the coordinator for later use
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
entry.async_on_unload(entry.add_update_listener(async_reload_entry))
# Perform platform initialization.
await hass.config_entries.async_forward_entry_setups(entry, ELMAX_PLATFORMS)
return True
async def async_reload_entry(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Handle an options update."""
await hass.config_entries.async_reload(entry.entry_id)
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, ELMAX_PLATFORMS)

1
homeassistant/components/elmax/alarm_control_panel.py
View File

@ -39,7 +39,6 @@ async def async_setup_entry(
# Otherwise, add all the entities we found
entities = [
ElmaxArea(
panel=coordinator.panel_entry,
elmax_device=area,
panel_version=panel_status.release,
coordinator=coordinator,

1
homeassistant/components/elmax/binary_sensor.py
View File

@ -38,7 +38,6 @@ async def async_setup_entry(
if zone.endpoint_id in known_devices:
continue
entity = ElmaxSensor(
panel=coordinator.panel_entry,
elmax_device=zone,
panel_version=panel_status.release,
coordinator=coordinator,

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

@ -1,10 +1,11 @@
"""Elmax integration common classes and utilities."""
from __future__ import annotations
import asyncio
from asyncio import timeout
from datetime import timedelta
import logging
from logging import Logger
import ssl
from elmax_api.exceptions import (
ElmaxApiError,
@ -13,12 +14,14 @@ from elmax_api.exceptions import (
ElmaxNetworkError,
ElmaxPanelBusyError,
)
from elmax_api.http import Elmax
from elmax_api.http import Elmax, GenericElmax
from elmax_api.model.actuator import Actuator
from elmax_api.model.area import Area
from elmax_api.model.cover import Cover
from elmax_api.model.endpoint import DeviceEndpoint
from elmax_api.model.panel import PanelEntry, PanelStatus
from httpx import ConnectError, ConnectTimeout
from packaging import version
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, HomeAssistantError
@ -29,11 +32,50 @@ from homeassistant.helpers.update_coordinator import (
UpdateFailed,
)
from .const import DEFAULT_TIMEOUT, DOMAIN
from .const import (
DEFAULT_TIMEOUT,
DOMAIN,
ELMAX_LOCAL_API_PATH,
MIN_APIV2_SUPPORTED_VERSION,
)
_LOGGER = logging.getLogger(__name__)
def get_direct_api_url(host: str, port: int, use_ssl: bool) -> str:
"""Return the direct API url given the base URI."""
schema = "https" if use_ssl else "http"
return f"{schema}://{host}:{port}/{ELMAX_LOCAL_API_PATH}"
def build_direct_ssl_context(cadata: str) -> ssl.SSLContext:
"""Create a custom SSL context for direct-api verification."""
context = ssl.SSLContext(protocol=ssl.PROTOCOL_TLS_CLIENT)
context.check_hostname = False
context.verify_mode = ssl.CERT_REQUIRED
context.load_verify_locations(cadata=cadata)
return context
def check_local_version_supported(api_version: str | None) -> bool:
"""Check whether the given API version is supported."""
if api_version is None:
return False
return version.parse(api_version) >= version.parse(MIN_APIV2_SUPPORTED_VERSION)
class DirectPanel(PanelEntry):
"""Helper class for wrapping a directly accessed Elmax Panel."""
def __init__(self, panel_uri):
"""Construct the object."""
super().__init__(panel_uri, True, {})
def get_name_by_user(self, username: str) -> str:
"""Return the panel name."""
return f"Direct Panel {self.hash}"
class ElmaxCoordinator(DataUpdateCoordinator[PanelStatus]): # pylint: disable=hass-enforce-coordinator-module
"""Coordinator helper to handle Elmax API polling."""
@ -41,25 +83,21 @@ class ElmaxCoordinator(DataUpdateCoordinator[PanelStatus]): # pylint: disable=h
self,
hass: HomeAssistant,
logger: Logger,
username: str,
password: str,
panel_id: str,
panel_pin: str,
elmax_api_client: GenericElmax,
panel: PanelEntry,
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._client = elmax_api_client
self._panel_entry = panel
self._state_by_endpoint = None
super().__init__(
hass=hass, logger=logger, name=name, update_interval=update_interval
)
@property
def panel_entry(self) -> PanelEntry | None:
def panel_entry(self) -> PanelEntry:
"""Return the panel entry."""
return self._panel_entry
@ -92,54 +130,46 @@ class ElmaxCoordinator(DataUpdateCoordinator[PanelStatus]): # pylint: disable=h
"""Return the current http client being used by this instance."""
return self._client
@http_client.setter
def http_client(self, client: GenericElmax):
"""Set the client library instance for Elmax API."""
self._client = client
async def _async_update_data(self):
try:
async with asyncio.timeout(DEFAULT_TIMEOUT):
# Retrieve the panel online status first
panels = await self._client.list_control_panels()
panel = next(
(panel for panel in panels if panel.hash == self._panel_id), None
)
async with timeout(DEFAULT_TIMEOUT):
# The following command might fail in case of the panel is offline.
# We handle this case in the following exception blocks.
status = await self._client.get_current_panel_status()
# 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 not panel:
raise ConfigEntryAuthFailed(
f"Panel ID {self._panel_id} is no more linked to this user"
" account"
)
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
# Store a dictionary for fast endpoint state access
self._state_by_endpoint = {
k.endpoint_id: k for k in status.all_endpoints
}
return status
except ElmaxBadPinError as err:
raise ConfigEntryAuthFailed("Control panel pin was refused") from err
except ElmaxBadLoginError as err:
raise ConfigEntryAuthFailed("Refused username/password") from err
raise ConfigEntryAuthFailed("Refused username/password/pin") from err
except ElmaxApiError as err:
raise UpdateFailed(f"Error communicating with ELMAX API: {err}") from err
except ElmaxPanelBusyError as err:
raise UpdateFailed(
"Communication with the panel failed, as it is currently busy"
) from err
except ElmaxNetworkError as err:
except (ConnectError, ConnectTimeout, ElmaxNetworkError) as err:
if isinstance(self._client, Elmax):
raise UpdateFailed(
"A communication error has occurred. "
"Make sure HA can reach the internet and that "
"your firewall allows communication with the Meross Cloud."
) from err
raise UpdateFailed(
"A network error occurred while communicating with Elmax cloud."
"A communication error has occurred. "
"Make sure the panel is online and that "
"your firewall allows communication with it."
) from err
@ -148,20 +178,18 @@ class ElmaxEntity(CoordinatorEntity[ElmaxCoordinator]):
def __init__(
self,
panel: PanelEntry,
elmax_device: DeviceEndpoint,
panel_version: str,
coordinator: ElmaxCoordinator,
) -> None:
"""Construct the object."""
super().__init__(coordinator=coordinator)
self._panel = panel
self._device = elmax_device
self._attr_unique_id = elmax_device.endpoint_id
self._attr_name = elmax_device.name
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, panel.hash)},
name=panel.get_name_by_user(
identifiers={(DOMAIN, coordinator.panel_entry.hash)},
name=coordinator.panel_entry.get_name_by_user(
coordinator.http_client.get_authenticated_username()
),
manufacturer="Elmax",
@ -172,4 +200,4 @@ class ElmaxEntity(CoordinatorEntity[ElmaxCoordinator]):
@property
def available(self) -> bool:
"""Return if entity is available."""
return super().available and self._panel.online
return super().available and self.coordinator.panel_entry.online

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

@ -6,20 +6,37 @@ 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
from elmax_api.http import Elmax, ElmaxLocal, GenericElmax
from elmax_api.model.panel import PanelEntry, PanelStatus
import httpx
import voluptuous as vol
from homeassistant.components.zeroconf import ZeroconfServiceInfo
from homeassistant.config_entries import ConfigEntry, ConfigFlow, ConfigFlowResult
from homeassistant.data_entry_flow import FlowResult
from homeassistant.exceptions import HomeAssistantError
from .common import (
build_direct_ssl_context,
check_local_version_supported,
get_direct_api_url,
)
from .const import (
CONF_ELMAX_MODE,
CONF_ELMAX_MODE_CLOUD,
CONF_ELMAX_MODE_DIRECT,
CONF_ELMAX_MODE_DIRECT_HOST,
CONF_ELMAX_MODE_DIRECT_PORT,
CONF_ELMAX_MODE_DIRECT_SSL,
CONF_ELMAX_MODE_DIRECT_SSL_CERT,
CONF_ELMAX_PANEL_ID,
CONF_ELMAX_PANEL_NAME,
CONF_ELMAX_PANEL_PIN,
CONF_ELMAX_PASSWORD,
CONF_ELMAX_USERNAME,
DOMAIN,
ELMAX_MODE_DIRECT_DEFAULT_HTTP_PORT,
ELMAX_MODE_DIRECT_DEFAULT_HTTPS_PORT,
)
_LOGGER = logging.getLogger(__name__)
@ -39,6 +56,22 @@ REAUTH_FORM_SCHEMA = vol.Schema(
}
)
DIRECT_SETUP_SCHEMA = vol.Schema(
{
vol.Required(CONF_ELMAX_MODE_DIRECT_HOST): str,
vol.Required(CONF_ELMAX_MODE_DIRECT_PORT, default=443): int,
vol.Required(CONF_ELMAX_MODE_DIRECT_SSL, default=True): bool,
vol.Required(CONF_ELMAX_PANEL_PIN): str,
}
)
ZEROCONF_SETUP_SCHEMA = vol.Schema(
{
vol.Required(CONF_ELMAX_PANEL_PIN): str,
vol.Required(CONF_ELMAX_MODE_DIRECT_SSL, default=True): bool,
}
)
def _store_panel_by_name(
panel: PanelEntry, username: str, panel_names: dict[str, str]
@ -59,38 +92,216 @@ class ElmaxConfigFlow(ConfigFlow, domain=DOMAIN):
VERSION = 1
_client: Elmax
_username: str
_password: str
_selected_mode: str
_panel_pin: str
_panel_id: str
# Direct API variables
_panel_direct_use_ssl: bool
_panel_direct_hostname: str
_panel_direct_port: int
_panel_direct_follow_mdns: bool
_panel_direct_ssl_cert: str | None
_panel_direct_http_port: int
_panel_direct_https_port: int
# Cloud API variables
_cloud_username: str
_cloud_password: str
_reauth_cloud_username: str | None
_reauth_cloud_panelid: str | None
# Panel selection variables
_panels_schema: vol.Schema
_panel_names: dict
_entry: ConfigEntry | None
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle a flow initialized by the user."""
) -> FlowResult:
"""Handle the flow initiated by the user."""
return await self.async_step_choose_mode(user_input=user_input)
async def async_step_choose_mode(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle local vs cloud mode selection step."""
return self.async_show_menu(
step_id="choose_mode",
menu_options={
CONF_ELMAX_MODE_CLOUD: "Connect to Elmax Panel via Elmax Cloud APIs",
CONF_ELMAX_MODE_DIRECT: "Connect to Elmax Panel via local/direct IP",
},
)
async def _handle_direct_and_create_entry(
self, fallback_step_id: str, schema: vol.Schema
) -> FlowResult:
return await self._test_direct_and_create_entry()
async def _test_direct_and_create_entry(self):
"""Test the direct connection to the Elmax panel and create and entry if successful."""
ssl_context = None
self._panel_direct_ssl_cert = None
if self._panel_direct_use_ssl:
# Fetch the remote certificate.
# Local API is exposed via a self-signed SSL that we must add to our trust store.
self._panel_direct_ssl_cert = (
await GenericElmax.retrieve_server_certificate(
hostname=self._panel_direct_hostname,
port=self._panel_direct_port,
)
)
ssl_context = build_direct_ssl_context(cadata=self._panel_direct_ssl_cert)
# Attempt the connection to make sure the pin works. Also, take the chance to retrieve the panel ID via APIs.
client_api_url = get_direct_api_url(
host=self._panel_direct_hostname,
port=self._panel_direct_port,
use_ssl=self._panel_direct_use_ssl,
)
client = ElmaxLocal(
panel_api_url=client_api_url,
panel_code=self._panel_pin,
ssl_context=ssl_context,
)
try:
await client.login()
except (ElmaxNetworkError, httpx.ConnectError, httpx.ConnectTimeout):
return self.async_show_form(
step_id=CONF_ELMAX_MODE_DIRECT,
data_schema=DIRECT_SETUP_SCHEMA,
errors={"base": "network_error"},
)
except ElmaxBadLoginError:
return self.async_show_form(
step_id=CONF_ELMAX_MODE_DIRECT,
data_schema=DIRECT_SETUP_SCHEMA,
errors={"base": "invalid_auth"},
)
# Retrieve the current panel status. If this succeeds, it means the
# setup did complete successfully.
panel_status: PanelStatus = await client.get_current_panel_status()
# Make sure this is the only Elmax integration for this specific panel id.
await self.async_set_unique_id(panel_status.panel_id)
self._abort_if_unique_id_configured()
return await self._check_unique_and_create_entry(
unique_id=panel_status.panel_id,
title=f"Elmax Direct {panel_status.panel_id}",
data={
CONF_ELMAX_MODE: self._selected_mode,
CONF_ELMAX_MODE_DIRECT_HOST: self._panel_direct_hostname,
CONF_ELMAX_MODE_DIRECT_PORT: self._panel_direct_port,
CONF_ELMAX_MODE_DIRECT_SSL: self._panel_direct_use_ssl,
CONF_ELMAX_PANEL_PIN: self._panel_pin,
CONF_ELMAX_PANEL_ID: panel_status.panel_id,
CONF_ELMAX_MODE_DIRECT_SSL_CERT: self._panel_direct_ssl_cert,
},
)
async def async_step_direct(self, user_input: dict[str, Any]) -> FlowResult:
"""Handle the direct setup step."""
self._selected_mode = CONF_ELMAX_MODE_CLOUD
if user_input is None:
return self.async_show_form(
step_id=CONF_ELMAX_MODE_DIRECT,
data_schema=DIRECT_SETUP_SCHEMA,
errors=None,
)
self._panel_direct_hostname = user_input[CONF_ELMAX_MODE_DIRECT_HOST]
self._panel_direct_port = user_input[CONF_ELMAX_MODE_DIRECT_PORT]
self._panel_direct_use_ssl = user_input[CONF_ELMAX_MODE_DIRECT_SSL]
self._panel_pin = user_input[CONF_ELMAX_PANEL_PIN]
self._panel_direct_follow_mdns = True
tmp_schema = vol.Schema(
{
vol.Required(
CONF_ELMAX_MODE_DIRECT_HOST, default=self._panel_direct_hostname
): str,
vol.Required(
CONF_ELMAX_MODE_DIRECT_PORT, default=self._panel_direct_port
): int,
vol.Required(
CONF_ELMAX_MODE_DIRECT_SSL, default=self._panel_direct_use_ssl
): bool,
vol.Required(CONF_ELMAX_PANEL_PIN, default=self._panel_pin): str,
}
)
return await self._handle_direct_and_create_entry(
fallback_step_id=CONF_ELMAX_MODE_DIRECT, schema=tmp_schema
)
async def async_step_zeroconf_setup(self, user_input: dict[str, Any]) -> FlowResult:
"""Handle the direct setup step triggered via zeroconf."""
if user_input is None:
return self.async_show_form(
step_id="zeroconf_setup",
data_schema=ZEROCONF_SETUP_SCHEMA,
errors=None,
)
self._panel_direct_use_ssl = user_input[CONF_ELMAX_MODE_DIRECT_SSL]
self._panel_direct_port = (
self._panel_direct_https_port
if self._panel_direct_use_ssl
else self._panel_direct_http_port
)
self._panel_pin = user_input[CONF_ELMAX_PANEL_PIN]
tmp_schema = vol.Schema(
{
vol.Required(CONF_ELMAX_PANEL_PIN, default=self._panel_pin): str,
vol.Required(
CONF_ELMAX_MODE_DIRECT_SSL, default=self._panel_direct_use_ssl
): bool,
}
)
return await self._handle_direct_and_create_entry(
fallback_step_id="zeroconf_setup", schema=tmp_schema
)
async def _check_unique_and_create_entry(
self, unique_id: str, title: str, data: Mapping[str, Any]
) -> FlowResult:
# Make sure this is the only Elmax integration for this specific panel id.
await self.async_set_unique_id(unique_id)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=title,
data=data,
)
async def async_step_cloud(self, user_input: dict[str, Any]) -> FlowResult:
"""Handle the cloud setup flow."""
self._selected_mode = CONF_ELMAX_MODE_CLOUD
# 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)
username = user_input[CONF_ELMAX_USERNAME]
password = user_input[CONF_ELMAX_PASSWORD]
return self.async_show_form(
step_id=CONF_ELMAX_MODE_CLOUD, data_schema=LOGIN_FORM_SCHEMA, errors={}
)
# 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.
username = user_input[CONF_ELMAX_USERNAME]
password = user_input[CONF_ELMAX_PASSWORD]
try:
client = await self._async_login(username=username, password=password)
except ElmaxBadLoginError:
return self.async_show_form(
step_id="user",
step_id=CONF_ELMAX_MODE_CLOUD,
data_schema=LOGIN_FORM_SCHEMA,
errors={"base": "invalid_auth"},
)
except ElmaxNetworkError:
_LOGGER.exception("A network error occurred")
return self.async_show_form(
step_id="user",
step_id=CONF_ELMAX_MODE_CLOUD,
data_schema=LOGIN_FORM_SCHEMA,
errors={"base": "network_error"},
)
@ -101,7 +312,7 @@ class ElmaxConfigFlow(ConfigFlow, domain=DOMAIN):
# 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",
step_id=CONF_ELMAX_MODE_CLOUD,
data_schema=LOGIN_FORM_SCHEMA,
errors={"base": "no_panel_online"},
)
@ -125,8 +336,8 @@ class ElmaxConfigFlow(ConfigFlow, domain=DOMAIN):
}
)
self._panels_schema = schema
self._username = username
self._password = password
self._cloud_username = username
self._cloud_password = password
# If everything went OK, proceed to panel selection.
return await self.async_step_panels(user_input=None)
@ -155,23 +366,27 @@ class ElmaxConfigFlow(ConfigFlow, domain=DOMAIN):
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"
return self.async_show_form(
step_id="panels", data_schema=self._panels_schema, errors=errors
if errors:
return self.async_show_form(
step_id="panels", data_schema=self._panels_schema, errors=errors
)
return await self._check_unique_and_create_entry(
unique_id=panel_id,
title=f"Elmax cloud {panel_name}",
data={
CONF_ELMAX_MODE: CONF_ELMAX_MODE_CLOUD,
CONF_ELMAX_PANEL_ID: panel_id,
CONF_ELMAX_PANEL_PIN: panel_pin,
CONF_ELMAX_USERNAME: self._cloud_username,
CONF_ELMAX_PASSWORD: self._cloud_password,
},
)
async def async_step_reauth(
@ -179,6 +394,8 @@ class ElmaxConfigFlow(ConfigFlow, domain=DOMAIN):
) -> ConfigFlowResult:
"""Perform reauth upon an API authentication error."""
self._entry = self.hass.config_entries.async_get_entry(self.context["entry_id"])
self._reauth_cloud_username = entry_data.get(CONF_ELMAX_USERNAME)
self._reauth_cloud_panelid = entry_data.get(CONF_ELMAX_PANEL_ID)
return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(
@ -190,6 +407,7 @@ class ElmaxConfigFlow(ConfigFlow, domain=DOMAIN):
username = user_input[CONF_ELMAX_USERNAME]
password = user_input[CONF_ELMAX_PASSWORD]
panel_pin = user_input[CONF_ELMAX_PANEL_PIN]
await self.async_set_unique_id(self._reauth_cloud_panelid)
# Handle authentication, make sure the panel we are re-authenticating against is listed among results
# and verify its pin is correct.
@ -238,6 +456,89 @@ class ElmaxConfigFlow(ConfigFlow, domain=DOMAIN):
step_id="reauth_confirm", data_schema=REAUTH_FORM_SCHEMA, errors=errors
)
async def _async_handle_entry_match(
self,
local_id: str,
remote_id: str | None,
host: str,
https_port: int,
http_port: int,
) -> FlowResult | None:
# Look for another entry with the same PANEL_ID (local or remote).
# If there already is a matching panel, take the change to notify the Coordinator
# so that it uses the newly discovered IP address. This mitigates the issues
# arising with DHCP and IP changes of the panels.
for entry in self._async_current_entries(include_ignore=False):
if entry.data[CONF_ELMAX_PANEL_ID] in (local_id, remote_id):
# If the discovery finds another entry with the same ID, skip the notification.
# However, if the discovery finds a new host for a panel that was already registered
# for a given host (leave PORT comparison aside as we don't want to get notified twice
# for HTTP and HTTPS), update the entry so that the integration "follows" the DHCP IP.
if (
entry.data.get(CONF_ELMAX_MODE, CONF_ELMAX_MODE_CLOUD)
== CONF_ELMAX_MODE_DIRECT
and entry.data[CONF_ELMAX_MODE_DIRECT_HOST] != host
):
new_data: dict[str, Any] = {}
new_data.update(entry.data)
new_data[CONF_ELMAX_MODE_DIRECT_HOST] = host
new_data[CONF_ELMAX_MODE_DIRECT_PORT] = (
https_port
if entry.data[CONF_ELMAX_MODE_DIRECT_SSL]
else http_port
)
self.hass.config_entries.async_update_entry(
entry, unique_id=entry.unique_id, data=new_data
)
# Abort the configuration, as there already is an entry for this PANEL-ID.
return self.async_abort(reason="already_configured")
return None
async def async_step_zeroconf(
self, discovery_info: ZeroconfServiceInfo
) -> FlowResult:
"""Handle device found via zeroconf."""
host = discovery_info.host
https_port = (
int(discovery_info.port)
if discovery_info.port is not None
else ELMAX_MODE_DIRECT_DEFAULT_HTTPS_PORT
)
plain_http_port = discovery_info.properties.get(
"http_port", ELMAX_MODE_DIRECT_DEFAULT_HTTP_PORT
)
plain_http_port = int(plain_http_port)
local_id = discovery_info.properties.get("idl")
remote_id = discovery_info.properties.get("idr")
v2api_version = discovery_info.properties.get("v2")
# Only deal with panels exposing v2 version
if not check_local_version_supported(v2api_version):
return self.async_abort(reason="not_supported")
# Handle the discovered panel info. This is useful especially if the panel
# changes its IP address while remaining perfectly configured.
if (
local_id is not None
and (
abort_result := await self._async_handle_entry_match(
local_id, remote_id, host, https_port, plain_http_port
)
)
is not None
):
return abort_result
self._selected_mode = CONF_ELMAX_MODE_DIRECT
self._panel_direct_hostname = host
self._panel_direct_https_port = https_port
self._panel_direct_http_port = plain_http_port
self._panel_direct_follow_mdns = True
return self.async_show_form(
step_id="zeroconf_setup", data_schema=ZEROCONF_SETUP_SCHEMA
)
@staticmethod
async def _async_login(username: str, password: str) -> Elmax:
"""Log in to the Elmax cloud and return the http client."""

15
homeassistant/components/elmax/const.py
View File

@ -5,9 +5,21 @@ DOMAIN = "elmax"
CONF_ELMAX_USERNAME = "username"
CONF_ELMAX_PASSWORD = "password"
CONF_ELMAX_PANEL_ID = "panel_id"
CONF_ELMAX_PANEL_LOCAL_ID = "panel_local_id"
CONF_ELMAX_PANEL_REMOTE_ID = "panel_remote_id"
CONF_ELMAX_PANEL_PIN = "panel_pin"
CONF_ELMAX_PANEL_NAME = "panel_name"
CONF_ELMAX_MODE = "mode"
CONF_ELMAX_MODE_CLOUD = "cloud"
CONF_ELMAX_MODE_DIRECT = "direct"
CONF_ELMAX_MODE_DIRECT_HOST = "panel_api_host"
CONF_ELMAX_MODE_DIRECT_PORT = "panel_api_port"
CONF_ELMAX_MODE_DIRECT_SSL = "use_ssl"
CONF_ELMAX_MODE_DIRECT_SSL_CERT = "ssl_cert"
ELMAX_LOCAL_API_PATH = "api/v2"
CONF_CONFIG_ENTRY_ID = "config_entry_id"
CONF_ENDPOINT_ID = "endpoint_id"
@ -18,5 +30,8 @@ ELMAX_PLATFORMS = [
Platform.COVER,
]
ELMAX_MODE_DIRECT_DEFAULT_HTTPS_PORT = 443
ELMAX_MODE_DIRECT_DEFAULT_HTTP_PORT = 80
POLLING_SECONDS = 30
DEFAULT_TIMEOUT = 10.0
MIN_APIV2_SUPPORTED_VERSION = "4.9.13"

1
homeassistant/components/elmax/cover.py
View File

@ -49,7 +49,6 @@ async def async_setup_entry(
if cover.endpoint_id in known_devices:
continue
entity = ElmaxCover(
panel=coordinator.panel_entry,
elmax_device=cover,
panel_version=panel_status.release,
coordinator=coordinator,

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

@ -6,5 +6,10 @@
"documentation": "https://www.home-assistant.io/integrations/elmax",
"iot_class": "cloud_polling",
"loggers": ["elmax_api"],
"requirements": ["elmax-api==0.0.4"]
"requirements": ["elmax-api==0.0.4"],
"zeroconf": [
{
"type": "_elmax-ssl._tcp.local."
}
]
}

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

@ -1,13 +1,42 @@
{
"config": {
"step": {
"user": {
"choose_mode": {
"description": "Please choose the connection mode to Elmax panels.",
"menu_options": {
"cloud": "Connect to Elmax Panel via Elmax Cloud APIs",
"direct": "Connect to Elmax Panel via local/direct IP"
}
},
"cloud": {
"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%]"
}
},
"zeroconf_setup": {
"description": "Configure discovered local Elmax panel",
"data": {
"panel_pin": "Panel PIN code",
"use_ssl": "Use SSL"
},
"data_description": {
"use_ssl": "Whether or not using strict SSL checks. Disable if the panel does not expose a valid SSL certificate or if SSL communication is unsupported by the panel you are connecting to."
}
},
"direct": {
"description": "Specify the Elmax panel connection parameters below.",
"data": {
"panel_api_host": "Panel API Hostname or IP",
"panel_api_port": "Panel API port",
"use_ssl": "Use SSL",
"panel_pin": "Panel PIN code"
},
"data_description": {
"use_ssl": "Whether or not using strict SSL checks. Disable if the panel does not expose a valid SSL certificate or if SSL communication is unsupported by the panel you are connecting to."
}
},
"panels": {
"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": {
@ -30,6 +59,7 @@
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"network_error": "A network error occurred",
"invalid_pin": "The provided pin is invalid",
"invalid_mode": "Invalid or unsupported mode",
"reauth_panel_disappeared": "The given panel is no longer associated to this user. Please log in using an account associated to this panel.",
"unknown": "[%key:common::config_flow::error::unknown%]"
},

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

@ -40,7 +40,6 @@ async def async_setup_entry(
if actuator.endpoint_id in known_devices:
continue
entity = ElmaxSwitch(
panel=coordinator.panel_entry,
elmax_device=actuator,
panel_version=panel_status.release,
coordinator=coordinator,

5
homeassistant/generated/zeroconf.py
View File

@ -411,6 +411,11 @@ ZEROCONF = {
"domain": "elgato",
},
],
"_elmax-ssl._tcp.local.": [
{
"domain": "elmax",
},
],
"_enphase-envoy._tcp.local.": [
{
"domain": "enphase_envoy",

8
tests/components/elmax/__init__.py
View File

@ -1,4 +1,5 @@
"""Tests for the Elmax component."""
from tests.common import load_fixture
MOCK_USER_JWT = (
"JWT eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9"
@ -12,4 +13,11 @@ MOCK_USER_ID = "1b11bb11bbb11111b1b11b1b"
MOCK_PANEL_ID = "2db3dae30b9102de4d078706f94d0708"
MOCK_PANEL_NAME = "Test Panel Name"
MOCK_PANEL_PIN = "000000"
MOCK_WRONG_PANEL_PIN = "000000"
MOCK_PASSWORD = "password"
MOCK_DIRECT_HOST = "1.1.1.1"
MOCK_DIRECT_HOST_CHANGED = "2.2.2.2"
MOCK_DIRECT_PORT = 443
MOCK_DIRECT_SSL = True
MOCK_DIRECT_CERT = load_fixture("direct/cert.pem", "elmax")
MOCK_DIRECT_FOLLOW_MDNS = True

56
tests/components/elmax/conftest.py
View File

@ -1,5 +1,6 @@
"""Configuration for Elmax tests."""
import json
from unittest.mock import patch
from elmax_api.constants import (
BASE_URL,
@ -11,25 +12,35 @@ from httpx import Response
import pytest
import respx
from . import MOCK_PANEL_ID, MOCK_PANEL_PIN
from . import (
MOCK_DIRECT_HOST,
MOCK_DIRECT_PORT,
MOCK_DIRECT_SSL,
MOCK_PANEL_ID,
MOCK_PANEL_PIN,
)
from tests.common import load_fixture
MOCK_DIRECT_BASE_URI = (
f"{'https' if MOCK_DIRECT_SSL else 'http'}://{MOCK_DIRECT_HOST}:{MOCK_DIRECT_PORT}"
)
@pytest.fixture(autouse=True)
def httpx_mock_fixture(requests_mock):
"""Configure httpx fixture."""
def httpx_mock_cloud_fixture(requests_mock):
"""Configure httpx fixture for cloud API communication."""
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"))
200, json=json.loads(load_fixture("cloud/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"))
200, json=json.loads(load_fixture("cloud/list_devices.json", "elmax"))
)
# Mock Panel GET.
@ -37,7 +48,40 @@ def httpx_mock_fixture(requests_mock):
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"))
200, json=json.loads(load_fixture("cloud/get_panel.json", "elmax"))
)
yield respx_mock
@pytest.fixture(autouse=True)
def httpx_mock_direct_fixture(requests_mock):
"""Configure httpx fixture for direct Panel-API communication."""
with respx.mock(
base_url=MOCK_DIRECT_BASE_URI, assert_all_called=False
) as respx_mock:
# Mock Login POST.
login_route = respx_mock.post(f"/api/v2/{ENDPOINT_LOGIN}", name="login")
login_route.return_value = Response(
200, json=json.loads(load_fixture("direct/login.json", "elmax"))
)
# Mock Device list GET.
list_devices_route = respx_mock.get(
f"/api/v2/{ENDPOINT_DISCOVERY}", name="discovery_panel"
)
list_devices_route.return_value = Response(
200, json=json.loads(load_fixture("direct/discovery_panel.json", "elmax"))
)
yield respx_mock
@pytest.fixture(autouse=True)
def elmax_mock_direct_cert(requests_mock):
"""Patch elmax library to return a specific PEM for SSL communication."""
with patch(
"elmax_api.http.GenericElmax.retrieve_server_certificate",
return_value=load_fixture("direct/cert.pem", "elmax"),
) as patched_ssl_get_cert:
yield patched_ssl_get_cert

126
tests/components/elmax/fixtures/cloud/get_panel.json Normal file
View File

@ -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"
}

12
tests/components/elmax/fixtures/cloud/list_devices.json Normal file
View File

@ -0,0 +1,12 @@
[
{
"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" }]
}
]

8
tests/components/elmax/fixtures/cloud/login.json Normal file
View File

@ -0,0 +1,8 @@
{
"token": "JWT eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJfaWQiOiIxYjExYmIxMWJiYjExMTExYjFiMTFiMWIiLCJlbWFpbCI6InRoaXMuaXNAdGVzdC5jb20iLCJyb2xlIjoidXNlciIsImlhdCI6MTYzNjE5OTk5OCwiZXhwIjoxNjM2MjM1OTk4fQ.1C7lXuKyX1HEGOfMxNwxJ2n-CjoW4rwvNRITQxLICv0",
"user": {
"_id": "1b11bb11bbb11111b1b11b1b",
"email": "this.is@test.com",
"role": "user"
}
}

22
tests/components/elmax/fixtures/direct/cert.pem Normal file
View File

@ -0,0 +1,22 @@
-----BEGIN CERTIFICATE-----
MIIDozCCAougAwIBAgIUQw+nCBfAnisI86E/KS24OpJl6oQwDQYJKoZIhvcNAQEL
BQAwYTELMAkGA1UEBhMCSVQxDzANBgNVBAgMBk1pbGFubzEPMA0GA1UEBwwGTWls
YW5vMRcwFQYDVQQKDA5BbGJlcnRvR2VuaW9sYTEXMBUGA1UEAwwOYWxiZXJ0b2dl
bmlvbGEwHhcNMjIxMTEyMTc1MzE0WhcNMjMxMTEyMTc1MzE0WjBhMQswCQYDVQQG
EwJJVDEPMA0GA1UECAwGTWlsYW5vMQ8wDQYDVQQHDAZNaWxhbm8xFzAVBgNVBAoM
DkFsYmVydG9HZW5pb2xhMRcwFQYDVQQDDA5hbGJlcnRvZ2VuaW9sYTCCASIwDQYJ
KoZIhvcNAQEBBQADggEPADCCAQoCggEBAKKlyvbpb3IqrKsmNe3WLscXWOm1zWuZ
OECukWl9NFep1v7H2VgH8Vc5/Lz8GyFIK6f4auq3E8Cv3AuDH3Z9q8sxN4E4vh6Q
zkFqyBa4yAXyaN6AT/ZgmZTbd4KUg+AHecGeDdedbDRc7s8bPDILcQM/S49RSnnS
IYivDnf3uByCPjvU2+/JRUvrB+rlL3tvUt/H+In8uRd01cBAx3GQLN/eqmkgUhiy
/eI3r7g5goxyCZpy6uyQEfN1CYWOpoIdL9rAEwwvrT+zK7iPqxCN3N0xQNvVpNzE
ifTONUPq+JPxO0SIP3Ro7rSeNSoe1O309qb7kpi5G/Zt7u3nRoiL1zUCAwEAAaNT
MFEwHQYDVR0OBBYEFIceFAyZ62kgZZqVJ4cLQCMbproRMB8GA1UdIwQYMBaAFIce
FAyZ62kgZZqVJ4cLQCMbproRMA8GA1UdEwEB/wQFMAMBAf8wDQYJKoZIhvcNAQEL
BQADggEBACHt8GY2FAruQ4Xa1JIMFHzGsqW8OMg6M1Ntrclu7MmST7T35VyCzBfB
QUwasn8bp/XHTZSmJB8RndYRKslATHSMyWhqjGCBNoSI8ljJgn9YLopRnAwEKQ6P
sqRl6/uMhwC587nIG1GJGWx2SbxrqoTcy39QmK0vHZfCVctzREZLaWGI2rtiQkVZ
mt3A0aQck+KQPlCIac14Z6lZJLJ4wSq6x7gjQAFA9uQfbnTaerOGgDTnhwsIrON3
TkSZ0dFgvt3lnbpO7fa8nPhdKFgzWZzymAjv3vsAxiboDE3LPn9BulxEkC+IFf5Q
6n+BK6Ogu16GZX4zUKeVdF089opux64=
-----END CERTIFICATE-----

148
tests/components/elmax/fixtures/direct/discovery_panel.json Normal file
View File

@ -0,0 +1,148 @@
{
"release": "PHANTOM64PRO_GSM 11.9.844",
"tappFeature": true,
"sceneFeature": true,
"zone": [
{
"endpointId": "13762559c53cd093171-zona-0",
"visibile": true,
"indice": 0,
"aperta": true,
"esclusa": false,
"nome": "ZONA 01"
},
{
"endpointId": "13762559c53cd093171-zona-1",
"visibile": true,
"indice": 1,
"aperta": true,
"esclusa": false,
"nome": "ZONA 02e"
},
{
"endpointId": "13762559c53cd093171-zona-2",
"visibile": true,
"indice": 2,
"aperta": true,
"esclusa": false,
"nome": "ZONA 03a"
},
{
"endpointId": "13762559c53cd093171-zona-3",
"visibile": true,
"indice": 3,
"aperta": true,
"esclusa": false,
"nome": "ZONA 04"
},
{
"endpointId": "13762559c53cd093171-zona-4",
"visibile": true,
"indice": 4,
"aperta": true,
"esclusa": false,
"nome": "ZONA 05"
},
{
"endpointId": "13762559c53cd093171-zona-5",
"visibile": true,
"indice": 5,
"aperta": true,
"esclusa": false,
"nome": "ZONA 06"
},
{
"endpointId": "13762559c53cd093171-zona-6",
"visibile": true,
"indice": 6,
"aperta": true,
"esclusa": false,
"nome": "ZONA 07"
},
{
"endpointId": "13762559c53cd093171-zona-7",
"visibile": true,
"indice": 7,
"aperta": true,
"esclusa": false,
"nome": "ZONA 08"
}
],
"uscite": [
{
"endpointId": "13762559c53cd093171-uscita-1",
"visibile": true,
"indice": 1,
"aperta": true,
"nome": "USCITA 02"
}
],
"aree": [
{
"endpointId": "13762559c53cd093171-area-0",
"visibile": true,
"indice": 0,
"statiDisponibili": [0, 1, 2, 3, 4],
"statiSessioneDisponibili": [0, 1, 2, 3],
"stato": 0,
"statoSessione": 1,
"zoneBmask": "0700000000000000",
"nome": "AREA 1"
},
{
"endpointId": "13762559c53cd093171-area-1",
"visibile": true,
"indice": 1,
"statiDisponibili": [0, 1, 2, 3, 4],
"statiSessioneDisponibili": [0, 1, 2, 3],
"stato": 0,
"statoSessione": 1,
"zoneBmask": "3800000000000000",
"nome": "AREA 2"
},
{
"endpointId": "13762559c53cd093171-area-2",
"visibile": true,
"indice": 2,
"statiDisponibili": [0, 1, 2, 3, 4],
"statiSessioneDisponibili": [0, 1, 2, 3],
"stato": 0,
"statoSessione": 1,
"zoneBmask": "C000000000000000",
"nome": "AREA 3"
}
],
"tapparelle": [
{
"endpointId": "13762559c53cd093171-tapparella-0",
"visibile": true,
"indice": 0,
"stato": "stop",
"posizione": 0,
"nome": "ESPAN.DOM.01"
}
],
"gruppi": [
{
"endpointId": "13762559c53cd093171-gruppo-1",
"visibile": true,
"indice": 1,
"nome": "GRUPPOUSC02"
}
],
"scenari": [
{
"endpointId": "13762559c53cd093171-scenario-1",
"visibile": true,
"indice": 1,
"nome": "SCENARIO02"
},
{
"endpointId": "13762559c53cd093171-scenario-2",
"visibile": true,
"indice": 2,
"nome": "SCENARIO03"
}
],
"datetime": "19:16:44 23/10/2022"
}

3
tests/components/elmax/fixtures/direct/login.json Normal file
View File

@ -0,0 +1,3 @@
{
"token": "JWT eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJyb2xlIjoidXNlciIsImNhcGFiaWxpdGllcyI6eyJ6b25lIjoiMTExMTExMTEwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMCIsInVzYyI6IjAxMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwIiwiaW5kaWNlIjowLCJhcmVlIjo3LCJjYW0iOjAsInRhcHAiOjEsImdydXBwaSI6Miwic2NlbmFyaSI6Nn0sImlhdCI6MTY2NjU0NDYzMywiZXhwIjoxNTY2NTQ4MjM0fQ.0N50aK8VrCBvVZuLf2AzLxH96PFES7gql69URKb50cA"
}

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

@ -4,7 +4,15 @@ from unittest.mock import patch
from elmax_api.exceptions import ElmaxBadLoginError, ElmaxBadPinError, ElmaxNetworkError
from homeassistant import config_entries, data_entry_flow
from homeassistant.components import zeroconf
from homeassistant.components.elmax.const import (
CONF_ELMAX_MODE,
CONF_ELMAX_MODE_CLOUD,
CONF_ELMAX_MODE_DIRECT,
CONF_ELMAX_MODE_DIRECT_HOST,
CONF_ELMAX_MODE_DIRECT_PORT,
CONF_ELMAX_MODE_DIRECT_SSL,
CONF_ELMAX_MODE_DIRECT_SSL_CERT,
CONF_ELMAX_PANEL_ID,
CONF_ELMAX_PANEL_NAME,
CONF_ELMAX_PANEL_PIN,
@ -16,29 +24,122 @@ from homeassistant.config_entries import SOURCE_REAUTH
from homeassistant.core import HomeAssistant
from . import (
MOCK_DIRECT_CERT,
MOCK_DIRECT_HOST,
MOCK_DIRECT_HOST_CHANGED,
MOCK_DIRECT_PORT,
MOCK_DIRECT_SSL,
MOCK_PANEL_ID,
MOCK_PANEL_NAME,
MOCK_PANEL_PIN,
MOCK_PASSWORD,
MOCK_USERNAME,
MOCK_WRONG_PANEL_PIN,
)
from tests.common import MockConfigEntry
MOCK_ZEROCONF_DISCOVERY_INFO = zeroconf.ZeroconfServiceInfo(
ip_address=MOCK_DIRECT_HOST,
ip_addresses=[MOCK_DIRECT_HOST],
hostname="VideoBox.local",
name="VideoBox",
port=443,
properties={
"idl": MOCK_PANEL_ID,
"idr": MOCK_PANEL_ID,
"v1": "PHANTOM64PRO_GSM 11.9.844",
"v2": "4.9.13",
},
type="_elmax-ssl._tcp",
)
MOCK_ZEROCONF_DISCOVERY_CHANGED_INFO = zeroconf.ZeroconfServiceInfo(
ip_address=MOCK_DIRECT_HOST_CHANGED,
ip_addresses=[MOCK_DIRECT_HOST_CHANGED],
hostname="VideoBox.local",
name="VideoBox",
port=443,
properties={
"idl": MOCK_PANEL_ID,
"idr": MOCK_PANEL_ID,
"v1": "PHANTOM64PRO_GSM 11.9.844",
"v2": "4.9.13",
},
type="_elmax-ssl._tcp",
)
MOCK_ZEROCONF_DISCOVERY_INFO_NOT_SUPPORTED = zeroconf.ZeroconfServiceInfo(
ip_address=MOCK_DIRECT_HOST,
ip_addresses=[MOCK_DIRECT_HOST],
hostname="VideoBox.local",
name="VideoBox",
port=443,
properties={
"idl": MOCK_PANEL_ID,
"idr": MOCK_PANEL_ID,
"v1": "PHANTOM64PRO_GSM 11.9.844",
},
type="_elmax-ssl._tcp",
)
CONF_POLLING = "polling"
async def test_show_form(hass: HomeAssistant) -> None:
async def test_show_menu(hass: HomeAssistant) -> None:
"""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.FlowResultType.FORM
assert result["step_id"] == "user"
assert result["type"] == data_entry_flow.FlowResultType.MENU
assert result["step_id"] == "choose_mode"
async def test_standard_setup(hass: HomeAssistant) -> None:
"""Test the standard setup case."""
async def test_direct_setup(hass: HomeAssistant) -> None:
"""Test the standard direct setup case."""
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,
):
set_mode_result = await hass.config_entries.flow.async_configure(
show_form_result["flow_id"],
{"next_step_id": CONF_ELMAX_MODE_DIRECT},
)
result = await hass.config_entries.flow.async_configure(
set_mode_result["flow_id"],
{
CONF_ELMAX_MODE_DIRECT_HOST: MOCK_DIRECT_HOST,
CONF_ELMAX_MODE_DIRECT_PORT: MOCK_DIRECT_PORT,
CONF_ELMAX_MODE_DIRECT_SSL: MOCK_DIRECT_SSL,
CONF_ELMAX_PANEL_PIN: MOCK_PANEL_PIN,
},
)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
async def test_direct_show_form(hass: HomeAssistant) -> None:
"""Test the standard direct show form case."""
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,
):
set_mode_result = await hass.config_entries.flow.async_configure(
show_form_result["flow_id"],
)
result = await hass.config_entries.flow.async_configure(
set_mode_result["flow_id"], {"next_step_id": CONF_ELMAX_MODE_DIRECT}
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == CONF_ELMAX_MODE_DIRECT
assert result["errors"] is None
async def test_cloud_setup(hass: HomeAssistant) -> None:
"""Test the standard cloud setup case."""
# Setup once.
show_form_result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
@ -47,6 +148,10 @@ async def test_standard_setup(hass: HomeAssistant) -> None:
"homeassistant.components.elmax.async_setup_entry",
return_value=True,
):
show_form_result = await hass.config_entries.flow.async_configure(
show_form_result["flow_id"],
{"next_step_id": CONF_ELMAX_MODE_CLOUD},
)
login_result = await hass.config_entries.flow.async_configure(
show_form_result["flow_id"],
{
@ -65,7 +170,131 @@ async def test_standard_setup(hass: HomeAssistant) -> None:
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
async def test_one_config_allowed(hass: HomeAssistant) -> None:
async def test_zeroconf_form_setup_api_not_supported(hass):
"""Test the zeroconf setup case."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
data=MOCK_ZEROCONF_DISCOVERY_INFO_NOT_SUPPORTED,
)
assert result["type"] == data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "not_supported"
async def test_zeroconf_discovery(hass):
"""Test discovery of Elmax local api panel."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
data=MOCK_ZEROCONF_DISCOVERY_INFO,
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "zeroconf_setup"
assert result["errors"] is None
async def test_zeroconf_setup_show_form(hass):
"""Test discovery shows a form when activated."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
data=MOCK_ZEROCONF_DISCOVERY_INFO,
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "zeroconf_setup"
async def test_zeroconf_setup(hass):
"""Test the successful creation of config entry via discovery flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
data=MOCK_ZEROCONF_DISCOVERY_INFO,
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_ELMAX_PANEL_PIN: MOCK_PANEL_PIN,
CONF_ELMAX_MODE_DIRECT_SSL: MOCK_DIRECT_SSL,
},
)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
async def test_zeroconf_already_configured(hass):
"""Ensure local discovery aborts when same panel is already added to ha."""
MockConfigEntry(
domain=DOMAIN,
title=f"Elmax Direct ({MOCK_PANEL_ID})",
data={
CONF_ELMAX_MODE: CONF_ELMAX_MODE_DIRECT,
CONF_ELMAX_MODE_DIRECT_HOST: MOCK_DIRECT_HOST,
CONF_ELMAX_MODE_DIRECT_PORT: MOCK_DIRECT_PORT,
CONF_ELMAX_MODE_DIRECT_SSL: MOCK_DIRECT_SSL,
CONF_ELMAX_PANEL_PIN: MOCK_PANEL_PIN,
CONF_ELMAX_PANEL_ID: MOCK_PANEL_ID,
CONF_ELMAX_MODE_DIRECT_SSL_CERT: MOCK_DIRECT_CERT,
},
unique_id=MOCK_PANEL_ID,
).add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
data=MOCK_ZEROCONF_DISCOVERY_INFO,
)
assert result["type"] == data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "already_configured"
async def test_zeroconf_panel_changed_ip(hass):
"""Ensure local discovery updates the panel data when a the panel changes its IP."""
# Simulate an entry already exists for ip MOCK_DIRECT_HOST.
config_entry = MockConfigEntry(
domain=DOMAIN,
title=f"Elmax Direct ({MOCK_PANEL_ID})",
data={
CONF_ELMAX_MODE: CONF_ELMAX_MODE_DIRECT,
CONF_ELMAX_MODE_DIRECT_HOST: MOCK_DIRECT_HOST,
CONF_ELMAX_MODE_DIRECT_PORT: MOCK_DIRECT_PORT,
CONF_ELMAX_MODE_DIRECT_SSL: MOCK_DIRECT_SSL,
CONF_ELMAX_PANEL_PIN: MOCK_PANEL_PIN,
CONF_ELMAX_PANEL_ID: MOCK_PANEL_ID,
CONF_ELMAX_MODE_DIRECT_SSL_CERT: MOCK_DIRECT_CERT,
},
unique_id=MOCK_PANEL_ID,
)
config_entry.add_to_hass(hass)
# Simulate a MDNS discovery finds the same panel with a different IP (MOCK_ZEROCONF_DISCOVERY_CHANGED_INFO).
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
data=MOCK_ZEROCONF_DISCOVERY_CHANGED_INFO,
)
# Expect we abort the configuration as "already configured"
assert result["type"] == data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "already_configured"
# Expect the panel ip has been updated.
assert (
hass.config_entries.async_get_entry(config_entry.entry_id).data[
CONF_ELMAX_MODE_DIRECT_HOST
]
== MOCK_ZEROCONF_DISCOVERY_CHANGED_INFO.host
)
async def test_one_config_allowed_cloud(hass: HomeAssistant) -> None:
"""Test that only one Elmax configuration is allowed for each panel."""
MockConfigEntry(
domain=DOMAIN,
@ -82,8 +311,12 @@ async def test_one_config_allowed(hass: HomeAssistant) -> None:
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(
user_result = await hass.config_entries.flow.async_configure(
show_form_result["flow_id"],
{"next_step_id": CONF_ELMAX_MODE_CLOUD},
)
login_result = await hass.config_entries.flow.async_configure(
user_result["flow_id"],
{
CONF_ELMAX_USERNAME: MOCK_USERNAME,
CONF_ELMAX_PASSWORD: MOCK_PASSWORD,
@ -100,7 +333,7 @@ async def test_one_config_allowed(hass: HomeAssistant) -> None:
assert result["reason"] == "already_configured"
async def test_invalid_credentials(hass: HomeAssistant) -> None:
async def test_cloud_invalid_credentials(hass: HomeAssistant) -> None:
"""Test that invalid credentials throws an error."""
with patch(
"elmax_api.http.Elmax.login",
@ -109,6 +342,10 @@ async def test_invalid_credentials(hass: HomeAssistant) -> None:
show_form_result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
show_form_result = await hass.config_entries.flow.async_configure(
show_form_result["flow_id"],
{"next_step_id": CONF_ELMAX_MODE_CLOUD},
)
login_result = await hass.config_entries.flow.async_configure(
show_form_result["flow_id"],
{
@ -116,12 +353,12 @@ async def test_invalid_credentials(hass: HomeAssistant) -> None:
CONF_ELMAX_PASSWORD: "incorrect_password",
},
)
assert login_result["step_id"] == "user"
assert login_result["step_id"] == CONF_ELMAX_MODE_CLOUD
assert login_result["type"] == data_entry_flow.FlowResultType.FORM
assert login_result["errors"] == {"base": "invalid_auth"}
async def test_connection_error(hass: HomeAssistant) -> None:
async def test_cloud_connection_error(hass: HomeAssistant) -> None:
"""Test other than invalid credentials throws an error."""
with patch(
"elmax_api.http.Elmax.login",
@ -130,6 +367,10 @@ async def test_connection_error(hass: HomeAssistant) -> None:
show_form_result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
show_form_result = await hass.config_entries.flow.async_configure(
show_form_result["flow_id"],
{"next_step_id": CONF_ELMAX_MODE_CLOUD},
)
login_result = await hass.config_entries.flow.async_configure(
show_form_result["flow_id"],
{
@ -137,11 +378,65 @@ async def test_connection_error(hass: HomeAssistant) -> None:
CONF_ELMAX_PASSWORD: MOCK_PASSWORD,
},
)
assert login_result["step_id"] == "user"
assert login_result["step_id"] == CONF_ELMAX_MODE_CLOUD
assert login_result["type"] == data_entry_flow.FlowResultType.FORM
assert login_result["errors"] == {"base": "network_error"}
async def test_direct_connection_error(hass: HomeAssistant) -> None:
"""Test network error while dealing with direct panel APIs."""
with patch(
"elmax_api.http.ElmaxLocal.login",
side_effect=ElmaxNetworkError(),
):
show_form_result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
set_mode_result = await hass.config_entries.flow.async_configure(
show_form_result["flow_id"],
{"next_step_id": CONF_ELMAX_MODE_DIRECT},
)
result = await hass.config_entries.flow.async_configure(
set_mode_result["flow_id"],
{
CONF_ELMAX_MODE_DIRECT_HOST: MOCK_DIRECT_HOST,
CONF_ELMAX_MODE_DIRECT_PORT: MOCK_DIRECT_PORT,
CONF_ELMAX_MODE_DIRECT_SSL: MOCK_DIRECT_SSL,
CONF_ELMAX_PANEL_PIN: MOCK_PANEL_PIN,
},
)
assert result["step_id"] == CONF_ELMAX_MODE_DIRECT
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["errors"] == {"base": "network_error"}
async def test_direct_wrong_panel_code(hass: HomeAssistant) -> None:
"""Test wrong code being specified while dealing with direct panel APIs."""
with patch(
"elmax_api.http.ElmaxLocal.login",
side_effect=ElmaxBadLoginError(),
):
show_form_result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
set_mode_result = await hass.config_entries.flow.async_configure(
show_form_result["flow_id"],
{"next_step_id": CONF_ELMAX_MODE_DIRECT},
)
result = await hass.config_entries.flow.async_configure(
set_mode_result["flow_id"],
{
CONF_ELMAX_MODE_DIRECT_HOST: MOCK_DIRECT_HOST,
CONF_ELMAX_MODE_DIRECT_PORT: MOCK_DIRECT_PORT,
CONF_ELMAX_MODE_DIRECT_SSL: MOCK_DIRECT_SSL,
CONF_ELMAX_PANEL_PIN: MOCK_WRONG_PANEL_PIN,
},
)
assert result["step_id"] == CONF_ELMAX_MODE_DIRECT
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["errors"] == {"base": "invalid_auth"}
async def test_unhandled_error(hass: HomeAssistant) -> None:
"""Test unhandled exceptions."""
with patch(
@ -151,6 +446,10 @@ async def test_unhandled_error(hass: HomeAssistant) -> None:
show_form_result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
show_form_result = await hass.config_entries.flow.async_configure(
show_form_result["flow_id"],
{"next_step_id": CONF_ELMAX_MODE_CLOUD},
)
login_result = await hass.config_entries.flow.async_configure(
show_form_result["flow_id"],
{
@ -180,6 +479,10 @@ async def test_invalid_pin(hass: HomeAssistant) -> None:
show_form_result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
show_form_result = await hass.config_entries.flow.async_configure(
show_form_result["flow_id"],
{"next_step_id": CONF_ELMAX_MODE_CLOUD},
)
login_result = await hass.config_entries.flow.async_configure(
show_form_result["flow_id"],
{
@ -209,6 +512,10 @@ async def test_no_online_panel(hass: HomeAssistant) -> None:
show_form_result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
show_form_result = await hass.config_entries.flow.async_configure(
show_form_result["flow_id"],
{"next_step_id": CONF_ELMAX_MODE_CLOUD},
)
login_result = await hass.config_entries.flow.async_configure(
show_form_result["flow_id"],
{
@ -216,7 +523,7 @@ async def test_no_online_panel(hass: HomeAssistant) -> None:
CONF_ELMAX_PASSWORD: MOCK_PASSWORD,
},
)
assert login_result["step_id"] == "user"
assert login_result["step_id"] == CONF_ELMAX_MODE_CLOUD
assert login_result["type"] == data_entry_flow.FlowResultType.FORM
assert login_result["errors"] == {"base": "no_panel_online"}