mirror of
https://github.com/home-assistant/core.git
synced 2025-07-19 11:17:21 +00:00
Add alarm platform to Comelit (#104718)
* initial work on alarm * final work on alarm * coveragerc * add tests * add code validation * remove sensor changes for a dedicated PR * code optimization and cleanup * tweaks * tweak #2 * apply suggestion * code quality * code quality #2 * fix cover.py * api typing * use base classes where possibile * apply const as per review comment * cleanup unload entry * apply review comments
This commit is contained in:
parent
c6d1f1ccc8
commit
2cd6c2b6bf
@ -173,6 +173,7 @@ omit =
|
||||
homeassistant/components/coinbase/sensor.py
|
||||
homeassistant/components/comed_hourly_pricing/sensor.py
|
||||
homeassistant/components/comelit/__init__.py
|
||||
homeassistant/components/comelit/alarm_control_panel.py
|
||||
homeassistant/components/comelit/const.py
|
||||
homeassistant/components/comelit/cover.py
|
||||
homeassistant/components/comelit/coordinator.py
|
||||
|
@ -1,38 +1,66 @@
|
||||
"""Comelit integration."""
|
||||
|
||||
|
||||
from aiocomelit.const import BRIDGE
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import CONF_HOST, CONF_PIN, CONF_PORT, Platform
|
||||
from homeassistant.const import CONF_HOST, CONF_PIN, CONF_PORT, CONF_TYPE, Platform
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from .const import DEFAULT_PORT, DOMAIN
|
||||
from .coordinator import ComelitSerialBridge
|
||||
from .coordinator import ComelitBaseCoordinator, ComelitSerialBridge, ComelitVedoSystem
|
||||
|
||||
PLATFORMS = [Platform.COVER, Platform.LIGHT, Platform.SENSOR, Platform.SWITCH]
|
||||
BRIDGE_PLATFORMS = [
|
||||
Platform.COVER,
|
||||
Platform.LIGHT,
|
||||
Platform.SENSOR,
|
||||
Platform.SWITCH,
|
||||
]
|
||||
VEDO_PLATFORMS = [
|
||||
Platform.ALARM_CONTROL_PANEL,
|
||||
]
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Set up Comelit platform."""
|
||||
coordinator = ComelitSerialBridge(
|
||||
hass,
|
||||
entry.data[CONF_HOST],
|
||||
entry.data.get(CONF_PORT, DEFAULT_PORT),
|
||||
entry.data[CONF_PIN],
|
||||
)
|
||||
|
||||
coordinator: ComelitBaseCoordinator
|
||||
if entry.data.get(CONF_TYPE, BRIDGE) == BRIDGE:
|
||||
coordinator = ComelitSerialBridge(
|
||||
hass,
|
||||
entry.data[CONF_HOST],
|
||||
entry.data.get(CONF_PORT, DEFAULT_PORT),
|
||||
entry.data[CONF_PIN],
|
||||
)
|
||||
platforms = BRIDGE_PLATFORMS
|
||||
else:
|
||||
coordinator = ComelitVedoSystem(
|
||||
hass,
|
||||
entry.data[CONF_HOST],
|
||||
entry.data.get(CONF_PORT, DEFAULT_PORT),
|
||||
entry.data[CONF_PIN],
|
||||
)
|
||||
platforms = VEDO_PLATFORMS
|
||||
|
||||
await coordinator.async_config_entry_first_refresh()
|
||||
|
||||
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
|
||||
|
||||
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
|
||||
await hass.config_entries.async_forward_entry_setups(entry, platforms)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Unload a config entry."""
|
||||
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
|
||||
coordinator: ComelitSerialBridge = hass.data[DOMAIN][entry.entry_id]
|
||||
|
||||
if entry.data.get(CONF_TYPE, BRIDGE) == BRIDGE:
|
||||
platforms = BRIDGE_PLATFORMS
|
||||
else:
|
||||
platforms = VEDO_PLATFORMS
|
||||
|
||||
coordinator: ComelitBaseCoordinator = hass.data[DOMAIN][entry.entry_id]
|
||||
if unload_ok := await hass.config_entries.async_unload_platforms(entry, platforms):
|
||||
await coordinator.api.logout()
|
||||
await coordinator.api.close()
|
||||
hass.data[DOMAIN].pop(entry.entry_id)
|
||||
|
155
homeassistant/components/comelit/alarm_control_panel.py
Normal file
155
homeassistant/components/comelit/alarm_control_panel.py
Normal file
@ -0,0 +1,155 @@
|
||||
"""Support for Comelit VEDO system."""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
from aiocomelit.api import ComelitVedoAreaObject
|
||||
from aiocomelit.const import ALARM_AREAS, AlarmAreaState
|
||||
|
||||
from homeassistant.components.alarm_control_panel import (
|
||||
AlarmControlPanelEntity,
|
||||
AlarmControlPanelEntityFeature,
|
||||
CodeFormat,
|
||||
)
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import (
|
||||
STATE_ALARM_ARMED_AWAY,
|
||||
STATE_ALARM_ARMED_HOME,
|
||||
STATE_ALARM_ARMED_NIGHT,
|
||||
STATE_ALARM_ARMING,
|
||||
STATE_ALARM_DISARMED,
|
||||
STATE_ALARM_DISARMING,
|
||||
STATE_ALARM_TRIGGERED,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.helpers.typing import StateType
|
||||
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
||||
|
||||
from .const import DOMAIN
|
||||
from .coordinator import ComelitVedoSystem
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
AWAY = "away"
|
||||
DISABLE = "disable"
|
||||
HOME = "home"
|
||||
HOME_P1 = "home_p1"
|
||||
HOME_P2 = "home_p2"
|
||||
NIGHT = "night"
|
||||
|
||||
ALARM_ACTIONS: dict[str, str] = {
|
||||
DISABLE: "dis", # Disarm
|
||||
HOME: "p1", # Arm P1
|
||||
NIGHT: "p12", # Arm P1+P2
|
||||
AWAY: "tot", # Arm P1+P2 + IR / volumetric
|
||||
}
|
||||
|
||||
|
||||
ALARM_AREA_ARMED_STATUS: dict[str, int] = {
|
||||
HOME_P1: 1,
|
||||
HOME_P2: 2,
|
||||
NIGHT: 3,
|
||||
AWAY: 4,
|
||||
}
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant,
|
||||
config_entry: ConfigEntry,
|
||||
async_add_entities: AddEntitiesCallback,
|
||||
) -> None:
|
||||
"""Set up the Comelit VEDO system alarm control panel devices."""
|
||||
|
||||
coordinator: ComelitVedoSystem = hass.data[DOMAIN][config_entry.entry_id]
|
||||
|
||||
async_add_entities(
|
||||
ComelitAlarmEntity(coordinator, device, config_entry.entry_id)
|
||||
for device in coordinator.data[ALARM_AREAS].values()
|
||||
)
|
||||
|
||||
|
||||
class ComelitAlarmEntity(CoordinatorEntity[ComelitVedoSystem], AlarmControlPanelEntity):
|
||||
"""Representation of a Ness alarm panel."""
|
||||
|
||||
_attr_has_entity_name = True
|
||||
_attr_name = None
|
||||
_attr_code_format = CodeFormat.NUMBER
|
||||
_attr_code_arm_required = False
|
||||
_attr_supported_features = (
|
||||
AlarmControlPanelEntityFeature.ARM_AWAY
|
||||
| AlarmControlPanelEntityFeature.ARM_HOME
|
||||
)
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
coordinator: ComelitVedoSystem,
|
||||
area: ComelitVedoAreaObject,
|
||||
config_entry_entry_id: str,
|
||||
) -> None:
|
||||
"""Initialize the alarm panel."""
|
||||
self._api = coordinator.api
|
||||
self._area_index = area.index
|
||||
super().__init__(coordinator)
|
||||
# Use config_entry.entry_id as base for unique_id
|
||||
# because no serial number or mac is available
|
||||
self._attr_unique_id = f"{config_entry_entry_id}-{area.index}"
|
||||
self._attr_device_info = coordinator.platform_device_info(area, "area")
|
||||
if area.p2:
|
||||
self._attr_supported_features |= AlarmControlPanelEntityFeature.ARM_NIGHT
|
||||
|
||||
@property
|
||||
def _area(self) -> ComelitVedoAreaObject:
|
||||
"""Return area object."""
|
||||
return self.coordinator.data[ALARM_AREAS][self._area_index]
|
||||
|
||||
@property
|
||||
def available(self) -> bool:
|
||||
"""Return True if alarm is available."""
|
||||
if self._area.human_status in [AlarmAreaState.ANOMALY, AlarmAreaState.UNKNOWN]:
|
||||
return False
|
||||
return super().available
|
||||
|
||||
@property
|
||||
def state(self) -> StateType:
|
||||
"""Return the state of the alarm."""
|
||||
|
||||
_LOGGER.debug(
|
||||
"Area %s status is: %s. Armed is %s",
|
||||
self._area.name,
|
||||
self._area.human_status,
|
||||
self._area.armed,
|
||||
)
|
||||
if self._area.human_status == AlarmAreaState.ARMED:
|
||||
if self._area.armed == ALARM_AREA_ARMED_STATUS[AWAY]:
|
||||
return STATE_ALARM_ARMED_AWAY
|
||||
if self._area.armed == ALARM_AREA_ARMED_STATUS[NIGHT]:
|
||||
return STATE_ALARM_ARMED_NIGHT
|
||||
return STATE_ALARM_ARMED_HOME
|
||||
|
||||
{
|
||||
AlarmAreaState.DISARMED: STATE_ALARM_DISARMED,
|
||||
AlarmAreaState.ENTRY_DELAY: STATE_ALARM_DISARMING,
|
||||
AlarmAreaState.EXIT_DELAY: STATE_ALARM_ARMING,
|
||||
AlarmAreaState.TRIGGERED: STATE_ALARM_TRIGGERED,
|
||||
}.get(self._area.human_status)
|
||||
|
||||
return None
|
||||
|
||||
async def async_alarm_disarm(self, code: str | None = None) -> None:
|
||||
"""Send disarm command."""
|
||||
if code != str(self._api.device_pin):
|
||||
return
|
||||
await self._api.set_zone_status(self._area.index, ALARM_ACTIONS[DISABLE])
|
||||
|
||||
async def async_alarm_arm_away(self, code: str | None = None) -> None:
|
||||
"""Send arm away command."""
|
||||
await self._api.set_zone_status(self._area.index, ALARM_ACTIONS[AWAY])
|
||||
|
||||
async def async_alarm_arm_home(self, code: str | None = None) -> None:
|
||||
"""Send arm home command."""
|
||||
await self._api.set_zone_status(self._area.index, ALARM_ACTIONS[HOME])
|
||||
|
||||
async def async_alarm_arm_night(self, code: str | None = None) -> None:
|
||||
"""Send arm night command."""
|
||||
await self._api.set_zone_status(self._area.index, ALARM_ACTIONS[NIGHT])
|
@ -4,16 +4,22 @@ from __future__ import annotations
|
||||
from collections.abc import Mapping
|
||||
from typing import Any
|
||||
|
||||
from aiocomelit import ComeliteSerialBridgeApi, exceptions as aiocomelit_exceptions
|
||||
from aiocomelit import (
|
||||
ComeliteSerialBridgeApi,
|
||||
ComelitVedoApi,
|
||||
exceptions as aiocomelit_exceptions,
|
||||
)
|
||||
from aiocomelit.api import ComelitCommonApi
|
||||
from aiocomelit.const import BRIDGE
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant import core, exceptions
|
||||
from homeassistant.config_entries import ConfigEntry, ConfigFlow
|
||||
from homeassistant.const import CONF_HOST, CONF_PIN, CONF_PORT
|
||||
from homeassistant.const import CONF_HOST, CONF_PIN, CONF_PORT, CONF_TYPE
|
||||
from homeassistant.data_entry_flow import FlowResult
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
|
||||
from .const import _LOGGER, DEFAULT_PORT, DOMAIN
|
||||
from .const import _LOGGER, DEFAULT_PORT, DEVICE_TYPE_LIST, DOMAIN
|
||||
|
||||
DEFAULT_HOST = "192.168.1.252"
|
||||
DEFAULT_PIN = 111111
|
||||
@ -27,6 +33,7 @@ def user_form_schema(user_input: dict[str, Any] | None) -> vol.Schema:
|
||||
vol.Required(CONF_HOST, default=DEFAULT_HOST): cv.string,
|
||||
vol.Required(CONF_PORT, default=DEFAULT_PORT): cv.port,
|
||||
vol.Optional(CONF_PIN, default=DEFAULT_PIN): cv.positive_int,
|
||||
vol.Required(CONF_TYPE, default=BRIDGE): vol.In(DEVICE_TYPE_LIST),
|
||||
}
|
||||
)
|
||||
|
||||
@ -39,7 +46,11 @@ async def validate_input(
|
||||
) -> dict[str, str]:
|
||||
"""Validate the user input allows us to connect."""
|
||||
|
||||
api = ComeliteSerialBridgeApi(data[CONF_HOST], data[CONF_PORT], data[CONF_PIN])
|
||||
api: ComelitCommonApi
|
||||
if data.get(CONF_TYPE, BRIDGE) == BRIDGE:
|
||||
api = ComeliteSerialBridgeApi(data[CONF_HOST], data[CONF_PORT], data[CONF_PIN])
|
||||
else:
|
||||
api = ComelitVedoApi(data[CONF_HOST], data[CONF_PORT], data[CONF_PIN])
|
||||
|
||||
try:
|
||||
await api.login()
|
||||
|
@ -1,7 +1,10 @@
|
||||
"""Comelit constants."""
|
||||
import logging
|
||||
|
||||
from aiocomelit.const import BRIDGE, VEDO
|
||||
|
||||
_LOGGER = logging.getLogger(__package__)
|
||||
|
||||
DOMAIN = "comelit"
|
||||
DEFAULT_PORT = 80
|
||||
DEVICE_TYPE_LIST = [BRIDGE, VEDO]
|
||||
|
@ -1,9 +1,17 @@
|
||||
"""Support for Comelit."""
|
||||
from abc import abstractmethod
|
||||
from datetime import timedelta
|
||||
from typing import Any
|
||||
|
||||
from aiocomelit import ComeliteSerialBridgeApi, ComelitSerialBridgeObject, exceptions
|
||||
from aiocomelit.const import BRIDGE
|
||||
from aiocomelit import (
|
||||
ComeliteSerialBridgeApi,
|
||||
ComelitSerialBridgeObject,
|
||||
ComelitVedoApi,
|
||||
ComelitVedoAreaObject,
|
||||
exceptions,
|
||||
)
|
||||
from aiocomelit.api import ComelitCommonApi
|
||||
from aiocomelit.const import BRIDGE, VEDO
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.core import HomeAssistant
|
||||
@ -14,19 +22,18 @@ from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, Upda
|
||||
from .const import _LOGGER, DOMAIN
|
||||
|
||||
|
||||
class ComelitSerialBridge(DataUpdateCoordinator):
|
||||
"""Queries Comelit Serial Bridge."""
|
||||
class ComelitBaseCoordinator(DataUpdateCoordinator[dict[str, Any]]):
|
||||
"""Base coordinator for Comelit Devices."""
|
||||
|
||||
_hw_version: str
|
||||
config_entry: ConfigEntry
|
||||
api: ComelitCommonApi
|
||||
|
||||
def __init__(self, hass: HomeAssistant, host: str, port: int, pin: int) -> None:
|
||||
def __init__(self, hass: HomeAssistant, device: str, host: str) -> None:
|
||||
"""Initialize the scanner."""
|
||||
|
||||
self._device = device
|
||||
self._host = host
|
||||
self._port = port
|
||||
self._pin = pin
|
||||
|
||||
self.api = ComeliteSerialBridgeApi(host, port, pin)
|
||||
|
||||
super().__init__(
|
||||
hass=hass,
|
||||
@ -38,43 +45,80 @@ class ComelitSerialBridge(DataUpdateCoordinator):
|
||||
device_registry.async_get_or_create(
|
||||
config_entry_id=self.config_entry.entry_id,
|
||||
identifiers={(DOMAIN, self.config_entry.entry_id)},
|
||||
model=BRIDGE,
|
||||
name=f"{BRIDGE} ({self.api.host})",
|
||||
**self.basic_device_info,
|
||||
model=device,
|
||||
name=f"{device} ({self._host})",
|
||||
manufacturer="Comelit",
|
||||
hw_version=self._hw_version,
|
||||
)
|
||||
|
||||
@property
|
||||
def basic_device_info(self) -> dict:
|
||||
"""Set basic device info."""
|
||||
|
||||
return {
|
||||
"manufacturer": "Comelit",
|
||||
"hw_version": "20003101",
|
||||
}
|
||||
|
||||
def platform_device_info(self, device: ComelitSerialBridgeObject) -> dr.DeviceInfo:
|
||||
def platform_device_info(
|
||||
self,
|
||||
object_class: ComelitVedoAreaObject | ComelitSerialBridgeObject,
|
||||
object_type: str,
|
||||
) -> dr.DeviceInfo:
|
||||
"""Set platform device info."""
|
||||
|
||||
return dr.DeviceInfo(
|
||||
identifiers={
|
||||
(DOMAIN, f"{self.config_entry.entry_id}-{device.type}-{device.index}")
|
||||
(
|
||||
DOMAIN,
|
||||
f"{self.config_entry.entry_id}-{object_type}-{object_class.index}",
|
||||
)
|
||||
},
|
||||
via_device=(DOMAIN, self.config_entry.entry_id),
|
||||
name=device.name,
|
||||
model=f"{BRIDGE} {device.type}",
|
||||
**self.basic_device_info,
|
||||
name=object_class.name,
|
||||
model=f"{self._device} {object_type}",
|
||||
manufacturer="Comelit",
|
||||
hw_version=self._hw_version,
|
||||
)
|
||||
|
||||
async def _async_update_data(self) -> dict[str, Any]:
|
||||
"""Update device data."""
|
||||
_LOGGER.debug("Polling Comelit Serial Bridge host: %s", self._host)
|
||||
|
||||
_LOGGER.debug("Polling Comelit %s host: %s", self._device, self._host)
|
||||
try:
|
||||
await self.api.login()
|
||||
return await self.api.get_all_devices()
|
||||
return await self._async_update_system_data()
|
||||
except exceptions.CannotConnect as err:
|
||||
_LOGGER.warning("Connection error for %s", self._host)
|
||||
await self.api.close()
|
||||
raise UpdateFailed(f"Error fetching data: {repr(err)}") from err
|
||||
except exceptions.CannotAuthenticate:
|
||||
raise ConfigEntryAuthFailed
|
||||
|
||||
return {}
|
||||
|
||||
@abstractmethod
|
||||
async def _async_update_system_data(self) -> dict[str, Any]:
|
||||
"""Class method for updating data."""
|
||||
|
||||
|
||||
class ComelitSerialBridge(ComelitBaseCoordinator):
|
||||
"""Queries Comelit Serial Bridge."""
|
||||
|
||||
_hw_version = "20003101"
|
||||
api: ComeliteSerialBridgeApi
|
||||
|
||||
def __init__(self, hass: HomeAssistant, host: str, port: int, pin: int) -> None:
|
||||
"""Initialize the scanner."""
|
||||
self.api = ComeliteSerialBridgeApi(host, port, pin)
|
||||
super().__init__(hass, BRIDGE, host)
|
||||
|
||||
async def _async_update_system_data(self) -> dict[str, Any]:
|
||||
"""Specific method for updating data."""
|
||||
return await self.api.get_all_devices()
|
||||
|
||||
|
||||
class ComelitVedoSystem(ComelitBaseCoordinator):
|
||||
"""Queries Comelit VEDO system."""
|
||||
|
||||
_hw_version = "VEDO IP"
|
||||
api: ComelitVedoApi
|
||||
|
||||
def __init__(self, hass: HomeAssistant, host: str, port: int, pin: int) -> None:
|
||||
"""Initialize the scanner."""
|
||||
self.api = ComelitVedoApi(host, port, pin)
|
||||
super().__init__(hass, VEDO, host)
|
||||
|
||||
async def _async_update_system_data(self) -> dict[str, Any]:
|
||||
"""Specific method for updating data."""
|
||||
return await self.api.get_all_areas_and_zones()
|
||||
|
@ -54,7 +54,7 @@ class ComelitCoverEntity(
|
||||
# Use config_entry.entry_id as base for unique_id
|
||||
# because no serial number or mac is available
|
||||
self._attr_unique_id = f"{config_entry_entry_id}-{device.index}"
|
||||
self._attr_device_info = coordinator.platform_device_info(device)
|
||||
self._attr_device_info = coordinator.platform_device_info(device, device.type)
|
||||
# Device doesn't provide a status so we assume UNKNOWN at first startup
|
||||
self._last_action: int | None = None
|
||||
self._last_state: str | None = None
|
||||
|
@ -50,7 +50,7 @@ class ComelitLightEntity(CoordinatorEntity[ComelitSerialBridge], LightEntity):
|
||||
# Use config_entry.entry_id as base for unique_id
|
||||
# because no serial number or mac is available
|
||||
self._attr_unique_id = f"{config_entry_entry_id}-{device.index}"
|
||||
self._attr_device_info = coordinator.platform_device_info(device)
|
||||
self._attr_device_info = coordinator.platform_device_info(device, device.type)
|
||||
|
||||
async def _light_set_state(self, state: int) -> None:
|
||||
"""Set desired light state."""
|
||||
|
@ -6,5 +6,5 @@
|
||||
"documentation": "https://www.home-assistant.io/integrations/comelit",
|
||||
"iot_class": "local_polling",
|
||||
"loggers": ["aiocomelit"],
|
||||
"requirements": ["aiocomelit==0.6.2"]
|
||||
"requirements": ["aiocomelit==0.7.0"]
|
||||
}
|
||||
|
@ -69,7 +69,7 @@ class ComelitSensorEntity(CoordinatorEntity[ComelitSerialBridge], SensorEntity):
|
||||
# Use config_entry.entry_id as base for unique_id
|
||||
# because no serial number or mac is available
|
||||
self._attr_unique_id = f"{config_entry_entry_id}-{device.index}"
|
||||
self._attr_device_info = coordinator.platform_device_info(device)
|
||||
self._attr_device_info = coordinator.platform_device_info(device, device.type)
|
||||
|
||||
self.entity_description = description
|
||||
|
||||
|
@ -56,7 +56,7 @@ class ComelitSwitchEntity(CoordinatorEntity[ComelitSerialBridge], SwitchEntity):
|
||||
# Use config_entry.entry_id as base for unique_id
|
||||
# because no serial number or mac is available
|
||||
self._attr_unique_id = f"{config_entry_entry_id}-{device.type}-{device.index}"
|
||||
self._attr_device_info = coordinator.platform_device_info(device)
|
||||
self._attr_device_info = coordinator.platform_device_info(device, device.type)
|
||||
if device.type == OTHER:
|
||||
self._attr_device_class = SwitchDeviceClass.OUTLET
|
||||
|
||||
|
@ -215,7 +215,7 @@ aiobafi6==0.9.0
|
||||
aiobotocore==2.6.0
|
||||
|
||||
# homeassistant.components.comelit
|
||||
aiocomelit==0.6.2
|
||||
aiocomelit==0.7.0
|
||||
|
||||
# homeassistant.components.dhcp
|
||||
aiodiscover==1.6.0
|
||||
|
@ -194,7 +194,7 @@ aiobafi6==0.9.0
|
||||
aiobotocore==2.6.0
|
||||
|
||||
# homeassistant.components.comelit
|
||||
aiocomelit==0.6.2
|
||||
aiocomelit==0.7.0
|
||||
|
||||
# homeassistant.components.dhcp
|
||||
aiodiscover==1.6.0
|
||||
|
@ -1,7 +1,9 @@
|
||||
"""Common stuff for Comelit SimpleHome tests."""
|
||||
|
||||
from aiocomelit.const import VEDO
|
||||
|
||||
from homeassistant.components.comelit.const import DOMAIN
|
||||
from homeassistant.const import CONF_DEVICES, CONF_HOST, CONF_PIN, CONF_PORT
|
||||
from homeassistant.const import CONF_DEVICES, CONF_HOST, CONF_PIN, CONF_PORT, CONF_TYPE
|
||||
|
||||
MOCK_CONFIG = {
|
||||
DOMAIN: {
|
||||
@ -10,11 +12,18 @@ MOCK_CONFIG = {
|
||||
CONF_HOST: "fake_host",
|
||||
CONF_PORT: 80,
|
||||
CONF_PIN: 1234,
|
||||
}
|
||||
},
|
||||
{
|
||||
CONF_HOST: "fake_vedo_host",
|
||||
CONF_PORT: 8080,
|
||||
CONF_PIN: 1234,
|
||||
CONF_TYPE: VEDO,
|
||||
},
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
MOCK_USER_DATA = MOCK_CONFIG[DOMAIN][CONF_DEVICES][0]
|
||||
MOCK_USER_BRIDGE_DATA = MOCK_CONFIG[DOMAIN][CONF_DEVICES][0]
|
||||
MOCK_USER_VEDO_DATA = MOCK_CONFIG[DOMAIN][CONF_DEVICES][1]
|
||||
|
||||
FAKE_PIN = 5678
|
||||
|
@ -1,4 +1,5 @@
|
||||
"""Tests for Comelit SimpleHome config flow."""
|
||||
from typing import Any
|
||||
from unittest.mock import patch
|
||||
|
||||
from aiocomelit import CannotAuthenticate, CannotConnect
|
||||
@ -10,24 +11,27 @@ from homeassistant.const import CONF_HOST, CONF_PIN, CONF_PORT
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.data_entry_flow import FlowResultType
|
||||
|
||||
from .const import FAKE_PIN, MOCK_USER_DATA
|
||||
from .const import FAKE_PIN, MOCK_USER_BRIDGE_DATA, MOCK_USER_VEDO_DATA
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
async def test_user(hass: HomeAssistant) -> None:
|
||||
@pytest.mark.parametrize(
|
||||
("class_api", "user_input"),
|
||||
[
|
||||
("ComeliteSerialBridgeApi", MOCK_USER_BRIDGE_DATA),
|
||||
("ComelitVedoApi", MOCK_USER_VEDO_DATA),
|
||||
],
|
||||
)
|
||||
async def test_full_flow(
|
||||
hass: HomeAssistant, class_api: str, user_input: dict[str, Any]
|
||||
) -> None:
|
||||
"""Test starting a flow by user."""
|
||||
with patch(
|
||||
"aiocomelit.api.ComeliteSerialBridgeApi.login",
|
||||
f"aiocomelit.api.{class_api}.login",
|
||||
), patch(
|
||||
"aiocomelit.api.ComeliteSerialBridgeApi.logout",
|
||||
), patch(
|
||||
"homeassistant.components.comelit.async_setup_entry"
|
||||
) as mock_setup_entry, patch(
|
||||
"requests.get",
|
||||
) as mock_request_get:
|
||||
mock_request_get.return_value.status_code = 200
|
||||
|
||||
f"aiocomelit.api.{class_api}.logout",
|
||||
), patch("homeassistant.components.comelit.async_setup_entry") as mock_setup_entry:
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
@ -35,12 +39,12 @@ async def test_user(hass: HomeAssistant) -> None:
|
||||
assert result["step_id"] == "user"
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input=MOCK_USER_DATA
|
||||
result["flow_id"], user_input=user_input
|
||||
)
|
||||
assert result["type"] == FlowResultType.CREATE_ENTRY
|
||||
assert result["data"][CONF_HOST] == "fake_host"
|
||||
assert result["data"][CONF_PORT] == 80
|
||||
assert result["data"][CONF_PIN] == 1234
|
||||
assert result["data"][CONF_HOST] == user_input[CONF_HOST]
|
||||
assert result["data"][CONF_PORT] == user_input[CONF_PORT]
|
||||
assert result["data"][CONF_PIN] == user_input[CONF_PIN]
|
||||
assert not result["result"].unique_id
|
||||
await hass.async_block_till_done()
|
||||
|
||||
@ -73,7 +77,7 @@ async def test_exception_connection(hass: HomeAssistant, side_effect, error) ->
|
||||
"homeassistant.components.comelit.async_setup_entry",
|
||||
):
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input=MOCK_USER_DATA
|
||||
result["flow_id"], user_input=MOCK_USER_BRIDGE_DATA
|
||||
)
|
||||
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
@ -84,7 +88,7 @@ async def test_exception_connection(hass: HomeAssistant, side_effect, error) ->
|
||||
async def test_reauth_successful(hass: HomeAssistant) -> None:
|
||||
"""Test starting a reauthentication flow."""
|
||||
|
||||
mock_config = MockConfigEntry(domain=DOMAIN, data=MOCK_USER_DATA)
|
||||
mock_config = MockConfigEntry(domain=DOMAIN, data=MOCK_USER_BRIDGE_DATA)
|
||||
mock_config.add_to_hass(hass)
|
||||
|
||||
with patch(
|
||||
@ -128,7 +132,7 @@ async def test_reauth_successful(hass: HomeAssistant) -> None:
|
||||
async def test_reauth_not_successful(hass: HomeAssistant, side_effect, error) -> None:
|
||||
"""Test starting a reauthentication flow but no connection found."""
|
||||
|
||||
mock_config = MockConfigEntry(domain=DOMAIN, data=MOCK_USER_DATA)
|
||||
mock_config = MockConfigEntry(domain=DOMAIN, data=MOCK_USER_BRIDGE_DATA)
|
||||
mock_config.add_to_hass(hass)
|
||||
|
||||
with patch(
|
||||
|
Loading…
x
Reference in New Issue
Block a user