mirror of https://github.com/home-assistant/core
Create iAlarmXR integration (#67817)
* Creating iAlarmXR integration * fixing after review code * fixing remaining review hints * fixing remaining review hints * updating underlying pyialarm library * Creating iAlarmXR integration * fixing after review code * fixing remaining review hints * fixing remaining review hints * updating underlying pyialarm library * fixing after iMicknl review * Improving exception handling * Updating pyialarmxr library * fixing after merge dev * fixing after iMicknl review * Update CODEOWNERS Co-authored-by: Ludovico de Nittis <git@denittis.one> * fixing iot_class * Update homeassistant/components/ialarmxr/config_flow.py Co-authored-by: J. Nick Koston <nick@koston.org> * fixing after bdraco review * Update homeassistant/components/ialarmxr/config_flow.py Co-authored-by: J. Nick Koston <nick@koston.org> * reverting catching exception in setup step * Update homeassistant/components/ialarmxr/__init__.py Co-authored-by: J. Nick Koston <nick@koston.org> * Update homeassistant/components/ialarmxr/__init__.py Co-authored-by: J. Nick Koston <nick@koston.org> * fixing after bdraco suggestions * Update homeassistant/components/ialarmxr/alarm_control_panel.py Co-authored-by: J. Nick Koston <nick@koston.org> * Update homeassistant/components/ialarmxr/alarm_control_panel.py Co-authored-by: Mick Vleeshouwer <mick@imick.nl> * Update homeassistant/components/ialarmxr/config_flow.py Co-authored-by: J. Nick Koston <nick@koston.org> * Update homeassistant/components/ialarmxr/config_flow.py Co-authored-by: J. Nick Koston <nick@koston.org> * Update homeassistant/components/ialarmxr/__init__.py Co-authored-by: J. Nick Koston <nick@koston.org> * Update homeassistant/components/ialarmxr/__init__.py Co-authored-by: J. Nick Koston <nick@koston.org> * Update homeassistant/components/ialarmxr/utils.py Co-authored-by: J. Nick Koston <nick@koston.org> * regenerate translation and rename function to async_get_ialarmxr_mac * removing and collapsing unused error messages * fixing tests * improve code coverage in tests * improve code coverage in tests * improve code coverage in tests * fixing retry policy with new pyalarmxr library * snake case fix * renaming integration in ialarm_xr * renaming control panel name Co-authored-by: Ludovico de Nittis <git@denittis.one> Co-authored-by: J. Nick Koston <nick@koston.org> Co-authored-by: Mick Vleeshouwer <mick@imick.nl>
This commit is contained in:
BigMoby
GitHub
parent
5b896b315e
commit
42c80dda85
|
|
@ -514,6 +514,7 @@ omit =
|
|||
homeassistant/components/hvv_departures/__init__.py
|
||||
homeassistant/components/hydrawise/*
|
||||
homeassistant/components/ialarm/alarm_control_panel.py
|
||||
homeassistant/components/ialarm_xr/alarm_control_panel.py
|
||||
homeassistant/components/iammeter/sensor.py
|
||||
homeassistant/components/iaqualink/binary_sensor.py
|
||||
homeassistant/components/iaqualink/climate.py
|
||||
|
|
|
|||
|
|
@ -127,6 +127,7 @@ homeassistant.components.homewizard.*
|
|||
homeassistant.components.http.*
|
||||
homeassistant.components.huawei_lte.*
|
||||
homeassistant.components.hyperion.*
|
||||
homeassistant.components.ialarm_xr.*
|
||||
homeassistant.components.image_processing.*
|
||||
homeassistant.components.input_button.*
|
||||
homeassistant.components.input_select.*
|
||||
|
|
|
|||
|
|
@ -470,6 +470,8 @@ build.json @home-assistant/supervisor
|
|||
/tests/components/hyperion/ @dermotduffy
|
||||
/homeassistant/components/ialarm/ @RyuzakiKK
|
||||
/tests/components/ialarm/ @RyuzakiKK
|
||||
/homeassistant/components/ialarm_xr/ @bigmoby
|
||||
/tests/components/ialarm_xr/ @bigmoby
|
||||
/homeassistant/components/iammeter/ @lewei50
|
||||
/homeassistant/components/iaqualink/ @flz
|
||||
/tests/components/iaqualink/ @flz
|
||||
|
|
|
|||
|
|
@ -0,0 +1,101 @@
|
|||
"""iAlarmXR integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
|
||||
from async_timeout import timeout
|
||||
from pyialarmxr import (
|
||||
IAlarmXR,
|
||||
IAlarmXRGenericException,
|
||||
IAlarmXRSocketTimeoutException,
|
||||
)
|
||||
|
||||
from homeassistant.components.alarm_control_panel import SCAN_INTERVAL
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import (
|
||||
CONF_HOST,
|
||||
CONF_PASSWORD,
|
||||
CONF_PORT,
|
||||
CONF_USERNAME,
|
||||
Platform,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import ConfigEntryNotReady
|
||||
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
|
||||
|
||||
from .const import DOMAIN, IALARMXR_TO_HASS
|
||||
from .utils import async_get_ialarmxr_mac
|
||||
|
||||
PLATFORMS = [Platform.ALARM_CONTROL_PANEL]
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Set up iAlarmXR config."""
|
||||
host = entry.data[CONF_HOST]
|
||||
port = entry.data[CONF_PORT]
|
||||
username = entry.data[CONF_USERNAME]
|
||||
password = entry.data[CONF_PASSWORD]
|
||||
|
||||
ialarmxr = IAlarmXR(username, password, host, port)
|
||||
|
||||
try:
|
||||
async with timeout(10):
|
||||
ialarmxr_mac = await async_get_ialarmxr_mac(hass, ialarmxr)
|
||||
except (
|
||||
asyncio.TimeoutError,
|
||||
ConnectionError,
|
||||
IAlarmXRGenericException,
|
||||
IAlarmXRSocketTimeoutException,
|
||||
) as ex:
|
||||
raise ConfigEntryNotReady from ex
|
||||
|
||||
coordinator = IAlarmXRDataUpdateCoordinator(hass, ialarmxr, ialarmxr_mac)
|
||||
await coordinator.async_config_entry_first_refresh()
|
||||
|
||||
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
|
||||
|
||||
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Unload iAlarmXR config."""
|
||||
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
|
||||
hass.data[DOMAIN].pop(entry.entry_id)
|
||||
return unload_ok
|
||||
|
||||
|
||||
class IAlarmXRDataUpdateCoordinator(DataUpdateCoordinator):
|
||||
"""Class to manage fetching iAlarmXR data."""
|
||||
|
||||
def __init__(self, hass: HomeAssistant, ialarmxr: IAlarmXR, mac: str) -> None:
|
||||
"""Initialize global iAlarm data updater."""
|
||||
self.ialarmxr: IAlarmXR = ialarmxr
|
||||
self.state: str | None = None
|
||||
self.host: str = ialarmxr.host
|
||||
self.mac: str = mac
|
||||
|
||||
super().__init__(
|
||||
hass,
|
||||
_LOGGER,
|
||||
name=DOMAIN,
|
||||
update_interval=SCAN_INTERVAL,
|
||||
)
|
||||
|
||||
def _update_data(self) -> None:
|
||||
"""Fetch data from iAlarmXR via sync functions."""
|
||||
status: int = self.ialarmxr.get_status()
|
||||
_LOGGER.debug("iAlarmXR status: %s", status)
|
||||
|
||||
self.state = IALARMXR_TO_HASS.get(status)
|
||||
|
||||
async def _async_update_data(self) -> None:
|
||||
"""Fetch data from iAlarmXR."""
|
||||
try:
|
||||
async with timeout(10):
|
||||
await self.hass.async_add_executor_job(self._update_data)
|
||||
except ConnectionError as error:
|
||||
raise UpdateFailed(error) from error
|
||||
|
|
@ -0,0 +1,63 @@
|
|||
"""Interfaces with iAlarmXR control panels."""
|
||||
from __future__ import annotations
|
||||
|
||||
from homeassistant.components.alarm_control_panel import (
|
||||
AlarmControlPanelEntity,
|
||||
AlarmControlPanelEntityFeature,
|
||||
)
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers import device_registry
|
||||
from homeassistant.helpers.entity import DeviceInfo
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
||||
|
||||
from . import IAlarmXRDataUpdateCoordinator
|
||||
from .const import DOMAIN
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
|
||||
) -> None:
|
||||
"""Set up a iAlarmXR alarm control panel based on a config entry."""
|
||||
coordinator = hass.data[DOMAIN][entry.entry_id]
|
||||
async_add_entities([IAlarmXRPanel(coordinator)])
|
||||
|
||||
|
||||
class IAlarmXRPanel(CoordinatorEntity, AlarmControlPanelEntity):
|
||||
"""Representation of an iAlarmXR device."""
|
||||
|
||||
_attr_supported_features = (
|
||||
AlarmControlPanelEntityFeature.ARM_HOME
|
||||
| AlarmControlPanelEntityFeature.ARM_AWAY
|
||||
)
|
||||
_attr_name = "iAlarm_XR"
|
||||
_attr_icon = "mdi:security"
|
||||
|
||||
def __init__(self, coordinator: IAlarmXRDataUpdateCoordinator) -> None:
|
||||
"""Initialize the alarm panel."""
|
||||
super().__init__(coordinator)
|
||||
self.coordinator: IAlarmXRDataUpdateCoordinator = coordinator
|
||||
self._attr_unique_id = coordinator.mac
|
||||
self._attr_device_info = DeviceInfo(
|
||||
manufacturer="Antifurto365 - Meian",
|
||||
name=self.name,
|
||||
connections={(device_registry.CONNECTION_NETWORK_MAC, coordinator.mac)},
|
||||
)
|
||||
|
||||
@property
|
||||
def state(self) -> str | None:
|
||||
"""Return the state of the device."""
|
||||
return self.coordinator.state
|
||||
|
||||
def alarm_disarm(self, code: str | None = None) -> None:
|
||||
"""Send disarm command."""
|
||||
self.coordinator.ialarmxr.disarm()
|
||||
|
||||
def alarm_arm_home(self, code: str | None = None) -> None:
|
||||
"""Send arm home command."""
|
||||
self.coordinator.ialarmxr.arm_stay()
|
||||
|
||||
def alarm_arm_away(self, code: str | None = None) -> None:
|
||||
"""Send arm away command."""
|
||||
self.coordinator.ialarmxr.arm_away()
|
||||
|
|
@ -0,0 +1,94 @@
|
|||
"""Config flow for Antifurto365 iAlarmXR integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from logging import Logger
|
||||
from typing import Any
|
||||
|
||||
from pyialarmxr import (
|
||||
IAlarmXR,
|
||||
IAlarmXRGenericException,
|
||||
IAlarmXRSocketTimeoutException,
|
||||
)
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant import config_entries, core
|
||||
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_PORT, CONF_USERNAME
|
||||
from homeassistant.data_entry_flow import FlowResult
|
||||
|
||||
from .const import DOMAIN
|
||||
from .utils import async_get_ialarmxr_mac
|
||||
|
||||
_LOGGER: Logger = logging.getLogger(__name__)
|
||||
|
||||
DATA_SCHEMA = vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_HOST, default=IAlarmXR.IALARM_P2P_DEFAULT_HOST): str,
|
||||
vol.Required(CONF_PORT, default=IAlarmXR.IALARM_P2P_DEFAULT_PORT): int,
|
||||
vol.Required(CONF_USERNAME): str,
|
||||
vol.Required(CONF_PASSWORD): str,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
async def _async_get_device_formatted_mac(
|
||||
hass: core.HomeAssistant, username: str, password: str, host: str, port: int
|
||||
) -> str:
|
||||
"""Return iAlarmXR mac address."""
|
||||
|
||||
ialarmxr = IAlarmXR(username, password, host, port)
|
||||
return await async_get_ialarmxr_mac(hass, ialarmxr)
|
||||
|
||||
|
||||
class IAlarmConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
"""Handle a config flow for Antifurto365 iAlarmXR."""
|
||||
|
||||
VERSION = 1
|
||||
|
||||
async def async_step_user(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Handle the initial step."""
|
||||
|
||||
errors = {}
|
||||
|
||||
if user_input is not None:
|
||||
mac = None
|
||||
host = user_input[CONF_HOST]
|
||||
port = user_input[CONF_PORT]
|
||||
username = user_input[CONF_USERNAME]
|
||||
password = user_input[CONF_PASSWORD]
|
||||
|
||||
try:
|
||||
# If we are able to get the MAC address, we are able to establish
|
||||
# a connection to the device.
|
||||
mac = await _async_get_device_formatted_mac(
|
||||
self.hass, username, password, host, port
|
||||
)
|
||||
except ConnectionError:
|
||||
errors["base"] = "cannot_connect"
|
||||
except IAlarmXRGenericException as ialarmxr_exception:
|
||||
_LOGGER.debug(
|
||||
"IAlarmXRGenericException with message: [ %s ]",
|
||||
ialarmxr_exception.message,
|
||||
)
|
||||
errors["base"] = "unknown"
|
||||
except IAlarmXRSocketTimeoutException as ialarmxr_socket_timeout_exception:
|
||||
_LOGGER.debug(
|
||||
"IAlarmXRSocketTimeoutException with message: [ %s ]",
|
||||
ialarmxr_socket_timeout_exception.message,
|
||||
)
|
||||
errors["base"] = "unknown"
|
||||
except Exception: # pylint: disable=broad-except
|
||||
_LOGGER.exception("Unexpected exception")
|
||||
errors["base"] = "unknown"
|
||||
|
||||
if not errors:
|
||||
await self.async_set_unique_id(mac)
|
||||
self._abort_if_unique_id_configured()
|
||||
return self.async_create_entry(
|
||||
title=user_input[CONF_HOST], data=user_input
|
||||
)
|
||||
return self.async_show_form(
|
||||
step_id="user", data_schema=DATA_SCHEMA, errors=errors
|
||||
)
|
||||
|
|
@ -0,0 +1,18 @@
|
|||
"""Constants for the iAlarmXR integration."""
|
||||
from pyialarmxr import IAlarmXR
|
||||
|
||||
from homeassistant.const import (
|
||||
STATE_ALARM_ARMED_AWAY,
|
||||
STATE_ALARM_ARMED_HOME,
|
||||
STATE_ALARM_DISARMED,
|
||||
STATE_ALARM_TRIGGERED,
|
||||
)
|
||||
|
||||
DOMAIN = "ialarm_xr"
|
||||
|
||||
IALARMXR_TO_HASS = {
|
||||
IAlarmXR.ARMED_AWAY: STATE_ALARM_ARMED_AWAY,
|
||||
IAlarmXR.ARMED_STAY: STATE_ALARM_ARMED_HOME,
|
||||
IAlarmXR.DISARMED: STATE_ALARM_DISARMED,
|
||||
IAlarmXR.TRIGGERED: STATE_ALARM_TRIGGERED,
|
||||
}
|
||||
|
|
@ -0,0 +1,10 @@
|
|||
{
|
||||
"domain": "ialarm_xr",
|
||||
"name": "Antifurto365 iAlarmXR",
|
||||
"documentation": "https://www.home-assistant.io/integrations/ialarmxr",
|
||||
"requirements": ["pyialarmxr==1.0.13"],
|
||||
"codeowners": ["@bigmoby"],
|
||||
"config_flow": true,
|
||||
"iot_class": "cloud_polling",
|
||||
"loggers": ["pyialarmxr"]
|
||||
}
|
||||
|
|
@ -0,0 +1,21 @@
|
|||
{
|
||||
"config": {
|
||||
"step": {
|
||||
"user": {
|
||||
"data": {
|
||||
"host": "[%key:common::config_flow::data::host%]",
|
||||
"port": "[%key:common::config_flow::data::port%]",
|
||||
"username": "[%key:common::config_flow::data::username%]",
|
||||
"password": "[%key:common::config_flow::data::password%]"
|
||||
}
|
||||
}
|
||||
},
|
||||
"error": {
|
||||
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
|
||||
"unknown": "[%key:common::config_flow::error::unknown%]"
|
||||
},
|
||||
"abort": {
|
||||
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,21 @@
|
|||
{
|
||||
"config": {
|
||||
"abort": {
|
||||
"already_configured": "Device is already configured"
|
||||
},
|
||||
"error": {
|
||||
"cannot_connect": "Failed to connect",
|
||||
"unknown": "Unexpected error"
|
||||
},
|
||||
"step": {
|
||||
"user": {
|
||||
"data": {
|
||||
"host": "Host",
|
||||
"password": "Password",
|
||||
"port": "Port",
|
||||
"username": "Username"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,18 @@
|
|||
"""iAlarmXR utils."""
|
||||
import logging
|
||||
|
||||
from pyialarmxr import IAlarmXR
|
||||
|
||||
from homeassistant import core
|
||||
from homeassistant.helpers.device_registry import format_mac
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def async_get_ialarmxr_mac(hass: core.HomeAssistant, ialarmxr: IAlarmXR) -> str:
|
||||
"""Retrieve iAlarmXR MAC address."""
|
||||
_LOGGER.debug("Retrieving ialarmxr mac address")
|
||||
|
||||
mac = await hass.async_add_executor_job(ialarmxr.get_mac)
|
||||
|
||||
return format_mac(mac)
|
||||
|
|
@ -161,6 +161,7 @@ FLOWS = {
|
|||
"hvv_departures",
|
||||
"hyperion",
|
||||
"ialarm",
|
||||
"ialarm_xr",
|
||||
"iaqualink",
|
||||
"icloud",
|
||||
"ifttt",
|
||||
|
|
|
|||
11
mypy.ini
11
mypy.ini
|
|
@ -1160,6 +1160,17 @@ no_implicit_optional = true
|
|||
warn_return_any = true
|
||||
warn_unreachable = true
|
||||
|
||||
[mypy-homeassistant.components.ialarm_xr.*]
|
||||
check_untyped_defs = true
|
||||
disallow_incomplete_defs = true
|
||||
disallow_subclassing_any = true
|
||||
disallow_untyped_calls = true
|
||||
disallow_untyped_decorators = true
|
||||
disallow_untyped_defs = true
|
||||
no_implicit_optional = true
|
||||
warn_return_any = true
|
||||
warn_unreachable = true
|
||||
|
||||
[mypy-homeassistant.components.image_processing.*]
|
||||
check_untyped_defs = true
|
||||
disallow_incomplete_defs = true
|
||||
|
|
|
|||
|
|
@ -1549,6 +1549,9 @@ pyhomeworks==0.0.6
|
|||
# homeassistant.components.ialarm
|
||||
pyialarm==1.9.0
|
||||
|
||||
# homeassistant.components.ialarm_xr
|
||||
pyialarmxr==1.0.13
|
||||
|
||||
# homeassistant.components.icloud
|
||||
pyicloud==1.0.0
|
||||
|
||||
|
|
|
|||
|
|
@ -1037,6 +1037,9 @@ pyhomematic==0.1.77
|
|||
# homeassistant.components.ialarm
|
||||
pyialarm==1.9.0
|
||||
|
||||
# homeassistant.components.ialarm_xr
|
||||
pyialarmxr==1.0.13
|
||||
|
||||
# homeassistant.components.icloud
|
||||
pyicloud==1.0.0
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1 @@
|
|||
"""Tests for the Antifurto365 iAlarmXR integration."""
|
||||
|
|
@ -0,0 +1,185 @@
|
|||
"""Test the Antifurto365 iAlarmXR config flow."""
|
||||
|
||||
from unittest.mock import patch
|
||||
|
||||
from pyialarmxr import IAlarmXRGenericException, IAlarmXRSocketTimeoutException
|
||||
|
||||
from homeassistant import config_entries, data_entry_flow
|
||||
from homeassistant.components.ialarm_xr.const import DOMAIN
|
||||
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_PORT, CONF_USERNAME
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
TEST_DATA = {
|
||||
CONF_HOST: "1.1.1.1",
|
||||
CONF_PORT: 18034,
|
||||
CONF_USERNAME: "000ZZZ0Z00",
|
||||
CONF_PASSWORD: "00000000",
|
||||
}
|
||||
|
||||
TEST_MAC = "00:00:54:12:34:56"
|
||||
|
||||
|
||||
async def test_form(hass):
|
||||
"""Test we get the form."""
|
||||
|
||||
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["handler"] == "ialarm_xr"
|
||||
assert result["data_schema"].schema.get("host") == str
|
||||
assert result["data_schema"].schema.get("port") == int
|
||||
assert result["data_schema"].schema.get("password") == str
|
||||
assert result["data_schema"].schema.get("username") == str
|
||||
assert result["errors"] == {}
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.ialarm_xr.config_flow.IAlarmXR.get_status",
|
||||
return_value=1,
|
||||
), patch(
|
||||
"homeassistant.components.ialarm_xr.config_flow.IAlarmXR.get_mac",
|
||||
return_value=TEST_MAC,
|
||||
), patch(
|
||||
"homeassistant.components.ialarm_xr.async_setup_entry",
|
||||
return_value=True,
|
||||
) as mock_setup_entry:
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], TEST_DATA
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result2["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
|
||||
assert result2["title"] == TEST_DATA["host"]
|
||||
assert result2["data"] == TEST_DATA
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
|
||||
|
||||
async def test_form_cannot_connect(hass):
|
||||
"""Test we handle cannot connect error."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.ialarm_xr.config_flow.IAlarmXR.get_mac",
|
||||
side_effect=ConnectionError,
|
||||
):
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], TEST_DATA
|
||||
)
|
||||
|
||||
assert result2["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result2["errors"] == {"base": "cannot_connect"}
|
||||
|
||||
|
||||
async def test_form_exception(hass):
|
||||
"""Test we handle unknown exception."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.ialarm_xr.config_flow.IAlarmXR.get_mac",
|
||||
side_effect=Exception,
|
||||
):
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], TEST_DATA
|
||||
)
|
||||
|
||||
assert result2["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result2["errors"] == {"base": "unknown"}
|
||||
|
||||
|
||||
async def test_form_cannot_connect_throwing_connection_error(hass):
|
||||
"""Test we handle cannot connect error."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.ialarm_xr.config_flow.IAlarmXR.get_mac",
|
||||
side_effect=ConnectionError,
|
||||
):
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], TEST_DATA
|
||||
)
|
||||
|
||||
assert result2["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result2["errors"] == {"base": "cannot_connect"}
|
||||
|
||||
|
||||
async def test_form_cannot_connect_throwing_socket_timeout_exception(hass):
|
||||
"""Test we handle cannot connect error because of socket timeout."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.ialarm_xr.config_flow.IAlarmXR.get_mac",
|
||||
side_effect=IAlarmXRSocketTimeoutException,
|
||||
):
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], TEST_DATA
|
||||
)
|
||||
|
||||
assert result2["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result2["errors"] == {"base": "unknown"}
|
||||
|
||||
|
||||
async def test_form_cannot_connect_throwing_generic_exception(hass):
|
||||
"""Test we handle cannot connect error."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.ialarm_xr.config_flow.IAlarmXR.get_mac",
|
||||
side_effect=IAlarmXRGenericException,
|
||||
):
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], TEST_DATA
|
||||
)
|
||||
|
||||
assert result2["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result2["errors"] == {"base": "unknown"}
|
||||
|
||||
|
||||
async def test_form_already_exists(hass):
|
||||
"""Test that a flow with an existing host aborts."""
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
unique_id=TEST_MAC,
|
||||
data=TEST_DATA,
|
||||
)
|
||||
|
||||
entry.add_to_hass(hass)
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.ialarm_xr.config_flow.IAlarmXR.get_mac",
|
||||
return_value=TEST_MAC,
|
||||
):
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], TEST_DATA
|
||||
)
|
||||
|
||||
assert result2["type"] == data_entry_flow.RESULT_TYPE_ABORT
|
||||
assert result2["reason"] == "already_configured"
|
||||
|
||||
|
||||
async def test_flow_user_step_no_input(hass):
|
||||
"""Test appropriate error when no input is provided."""
|
||||
_result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
_result["flow_id"], user_input=None
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["step_id"] == config_entries.SOURCE_USER
|
||||
assert result["errors"] == {}
|
||||
|
|
@ -0,0 +1,120 @@
|
|||
"""Test the Antifurto365 iAlarmXR init."""
|
||||
import asyncio
|
||||
from datetime import timedelta
|
||||
from unittest.mock import Mock, patch
|
||||
from uuid import uuid4
|
||||
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.ialarm_xr.const import DOMAIN
|
||||
from homeassistant.config_entries import ConfigEntryState
|
||||
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_PORT, CONF_USERNAME
|
||||
from homeassistant.util.dt import utcnow
|
||||
|
||||
from tests.common import MockConfigEntry, async_fire_time_changed
|
||||
|
||||
|
||||
@pytest.fixture(name="ialarmxr_api")
|
||||
def ialarmxr_api_fixture():
|
||||
"""Set up IAlarmXR API fixture."""
|
||||
with patch("homeassistant.components.ialarm_xr.IAlarmXR") as mock_ialarm_api:
|
||||
yield mock_ialarm_api
|
||||
|
||||
|
||||
@pytest.fixture(name="mock_config_entry")
|
||||
def mock_config_fixture():
|
||||
"""Return a fake config entry."""
|
||||
return MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data={
|
||||
CONF_HOST: "192.168.10.20",
|
||||
CONF_PORT: 18034,
|
||||
CONF_USERNAME: "000ZZZ0Z00",
|
||||
CONF_PASSWORD: "00000000",
|
||||
},
|
||||
entry_id=str(uuid4()),
|
||||
)
|
||||
|
||||
|
||||
async def test_setup_entry(hass, ialarmxr_api, mock_config_entry):
|
||||
"""Test setup entry."""
|
||||
ialarmxr_api.return_value.get_mac = Mock(return_value="00:00:54:12:34:56")
|
||||
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
ialarmxr_api.return_value.get_mac.assert_called_once()
|
||||
assert mock_config_entry.state is ConfigEntryState.LOADED
|
||||
|
||||
|
||||
async def test_setup_not_ready(hass, ialarmxr_api, mock_config_entry):
|
||||
"""Test setup failed because we can't connect to the alarm system."""
|
||||
ialarmxr_api.return_value.get_mac = Mock(side_effect=ConnectionError)
|
||||
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
assert not await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
|
||||
|
||||
|
||||
async def test_unload_entry(hass, ialarmxr_api, mock_config_entry):
|
||||
"""Test being able to unload an entry."""
|
||||
ialarmxr_api.return_value.get_mac = Mock(return_value="00:00:54:12:34:56")
|
||||
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_config_entry.state is ConfigEntryState.LOADED
|
||||
assert await hass.config_entries.async_unload(mock_config_entry.entry_id)
|
||||
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
|
||||
|
||||
|
||||
async def test_setup_not_ready_connection_error(hass, ialarmxr_api, mock_config_entry):
|
||||
"""Test setup failed because we can't connect to the alarm system."""
|
||||
ialarmxr_api.return_value.get_status = Mock(side_effect=ConnectionError)
|
||||
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
assert not await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
future = utcnow() + timedelta(seconds=30)
|
||||
async_fire_time_changed(hass, future)
|
||||
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
|
||||
|
||||
|
||||
async def test_setup_not_ready_timeout(hass, ialarmxr_api, mock_config_entry):
|
||||
"""Test setup failed because we can't connect to the alarm system."""
|
||||
ialarmxr_api.return_value.get_status = Mock(side_effect=asyncio.TimeoutError)
|
||||
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
assert not await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
future = utcnow() + timedelta(seconds=30)
|
||||
async_fire_time_changed(hass, future)
|
||||
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
|
||||
|
||||
|
||||
async def test_setup_entry_and_then_fail_on_update(
|
||||
hass, ialarmxr_api, mock_config_entry
|
||||
):
|
||||
"""Test setup entry."""
|
||||
ialarmxr_api.return_value.get_mac = Mock(return_value="00:00:54:12:34:56")
|
||||
ialarmxr_api.return_value.get_status = Mock(value=ialarmxr_api.DISARMED)
|
||||
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
ialarmxr_api.return_value.get_mac.assert_called_once()
|
||||
ialarmxr_api.return_value.get_status.assert_called_once()
|
||||
assert mock_config_entry.state is ConfigEntryState.LOADED
|
||||
|
||||
ialarmxr_api.return_value.get_status = Mock(side_effect=asyncio.TimeoutError)
|
||||
future = utcnow() + timedelta(seconds=60)
|
||||
async_fire_time_changed(hass, future)
|
||||
await hass.async_block_till_done()
|
||||
ialarmxr_api.return_value.get_status.assert_called_once()
|
||||
assert hass.states.get("alarm_control_panel.ialarm_xr").state == "unavailable"
|
||||
Loading…
Reference in New Issue