Electric-Special/ha-core
1
0
Fork 0
mirror of https://github.com/Electric-Special/ha-core.git synced 2026-03-21 06:05:26 +01:00
Code Issues Packages Projects Releases Wiki Activity

Add Satel Integra config flow (#138946)

Browse Source 
Co-authored-by: Shay Levy <levyshay1@gmail.com>
This commit is contained in:
Tom Matheussen
2025-09-16 17:24:15 +02:00
committed by GitHub
parent d65e704823
commit 6e4258c8a9
15 changed files with 1647 additions and 168 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
CODEOWNERS generated
View File

@@ -1350,6 +1350,8 @@ build.json @home-assistant/supervisor
/tests/components/samsungtv/ @chemelli74 @epenet
/homeassistant/components/sanix/ @tomaszsluszniak
/tests/components/sanix/ @tomaszsluszniak
/homeassistant/components/satel_integra/ @Tommatheussen
/tests/components/satel_integra/ @Tommatheussen
/homeassistant/components/scene/ @home-assistant/core
/tests/components/scene/ @home-assistant/core
/homeassistant/components/schedule/ @home-assistant/core

218
homeassistant/components/satel_integra/__init__.py
View File

@@ -1,59 +1,67 @@
"""Support for Satel Integra devices."""
import collections
import logging
from satel_integra.satel_integra import AsyncSatel
import voluptuous as vol
from homeassistant.const import CONF_HOST, CONF_PORT, EVENT_HOMEASSISTANT_STOP, Platform
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.discovery import async_load_platform
from homeassistant.config_entries import SOURCE_IMPORT
from homeassistant.const import (
CONF_CODE,
CONF_HOST,
CONF_NAME,
CONF_PORT,
EVENT_HOMEASSISTANT_STOP,
Platform,
)
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant, callback
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import config_validation as cv, issue_registry as ir
from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.helpers.typing import ConfigType
DEFAULT_ALARM_NAME = "satel_integra"
DEFAULT_PORT = 7094
DEFAULT_CONF_ARM_HOME_MODE = 1
DEFAULT_DEVICE_PARTITION = 1
DEFAULT_ZONE_TYPE = "motion"
from .const import (
CONF_ARM_HOME_MODE,
CONF_DEVICE_PARTITIONS,
CONF_OUTPUT_NUMBER,
CONF_OUTPUTS,
CONF_PARTITION_NUMBER,
CONF_SWITCHABLE_OUTPUT_NUMBER,
CONF_SWITCHABLE_OUTPUTS,
CONF_ZONE_NUMBER,
CONF_ZONE_TYPE,
CONF_ZONES,
DEFAULT_CONF_ARM_HOME_MODE,
DEFAULT_PORT,
DEFAULT_ZONE_TYPE,
DOMAIN,
SIGNAL_OUTPUTS_UPDATED,
SIGNAL_PANEL_MESSAGE,
SIGNAL_ZONES_UPDATED,
SUBENTRY_TYPE_OUTPUT,
SUBENTRY_TYPE_PARTITION,
SUBENTRY_TYPE_SWITCHABLE_OUTPUT,
SUBENTRY_TYPE_ZONE,
ZONES,
SatelConfigEntry,
)
_LOGGER = logging.getLogger(__name__)
DOMAIN = "satel_integra"
PLATFORMS = [Platform.ALARM_CONTROL_PANEL, Platform.BINARY_SENSOR, Platform.SWITCH]
DATA_SATEL = "satel_integra"
CONF_DEVICE_CODE = "code"
CONF_DEVICE_PARTITIONS = "partitions"
CONF_ARM_HOME_MODE = "arm_home_mode"
CONF_ZONE_NAME = "name"
CONF_ZONE_TYPE = "type"
CONF_ZONES = "zones"
CONF_OUTPUTS = "outputs"
CONF_SWITCHABLE_OUTPUTS = "switchable_outputs"
ZONES = "zones"
SIGNAL_PANEL_MESSAGE = "satel_integra.panel_message"
SIGNAL_PANEL_ARM_AWAY = "satel_integra.panel_arm_away"
SIGNAL_PANEL_ARM_HOME = "satel_integra.panel_arm_home"
SIGNAL_PANEL_DISARM = "satel_integra.panel_disarm"
SIGNAL_ZONES_UPDATED = "satel_integra.zones_updated"
SIGNAL_OUTPUTS_UPDATED = "satel_integra.outputs_updated"
ZONE_SCHEMA = vol.Schema(
{
vol.Required(CONF_ZONE_NAME): cv.string,
vol.Required(CONF_NAME): cv.string,
vol.Optional(CONF_ZONE_TYPE, default=DEFAULT_ZONE_TYPE): cv.string,
}
)
EDITABLE_OUTPUT_SCHEMA = vol.Schema({vol.Required(CONF_ZONE_NAME): cv.string})
EDITABLE_OUTPUT_SCHEMA = vol.Schema({vol.Required(CONF_NAME): cv.string})
PARTITION_SCHEMA = vol.Schema(
{
vol.Required(CONF_ZONE_NAME): cv.string,
vol.Required(CONF_NAME): cv.string,
vol.Optional(CONF_ARM_HOME_MODE, default=DEFAULT_CONF_ARM_HOME_MODE): vol.In(
[1, 2, 3]
),
@@ -63,7 +71,7 @@ PARTITION_SCHEMA = vol.Schema(
def is_alarm_code_necessary(value):
"""Check if alarm code must be configured."""
if value.get(CONF_SWITCHABLE_OUTPUTS) and CONF_DEVICE_CODE not in value:
if value.get(CONF_SWITCHABLE_OUTPUTS) and CONF_CODE not in value:
raise vol.Invalid("You need to specify alarm code to use switchable_outputs")
return value
@@ -75,7 +83,7 @@ CONFIG_SCHEMA = vol.Schema(
{
vol.Required(CONF_HOST): cv.string,
vol.Optional(CONF_PORT, default=DEFAULT_PORT): cv.port,
vol.Optional(CONF_DEVICE_CODE): cv.string,
vol.Optional(CONF_CODE): cv.string,
vol.Optional(CONF_DEVICE_PARTITIONS, default={}): {
vol.Coerce(int): PARTITION_SCHEMA
},
@@ -92,64 +100,106 @@ CONFIG_SCHEMA = vol.Schema(
)
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the Satel Integra component."""
conf = config[DOMAIN]
async def async_setup(hass: HomeAssistant, hass_config: ConfigType) -> bool:
"""Set up Satel Integra from YAML."""
zones = conf.get(CONF_ZONES)
outputs = conf.get(CONF_OUTPUTS)
switchable_outputs = conf.get(CONF_SWITCHABLE_OUTPUTS)
host = conf.get(CONF_HOST)
port = conf.get(CONF_PORT)
partitions = conf.get(CONF_DEVICE_PARTITIONS)
if config := hass_config.get(DOMAIN):
hass.async_create_task(_async_import(hass, config))
monitored_outputs = collections.OrderedDict(
list(outputs.items()) + list(switchable_outputs.items())
return True
async def _async_import(hass: HomeAssistant, config: ConfigType) -> None:
"""Process YAML import."""
if not hass.config_entries.async_entries(DOMAIN):
# Start import flow
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=config
)
if result.get("type") == FlowResultType.ABORT:
ir.async_create_issue(
hass,
DOMAIN,
"deprecated_yaml_import_issue_cannot_connect",
breaks_in_ha_version="2026.4.0",
is_fixable=False,
issue_domain=DOMAIN,
severity=ir.IssueSeverity.WARNING,
translation_key="deprecated_yaml_import_issue_cannot_connect",
translation_placeholders={
"domain": DOMAIN,
"integration_title": "Satel Integra",
},
)
return
ir.async_create_issue(
hass,
HOMEASSISTANT_DOMAIN,
f"deprecated_yaml_{DOMAIN}",
breaks_in_ha_version="2026.4.0",
is_fixable=False,
issue_domain=DOMAIN,
severity=ir.IssueSeverity.WARNING,
translation_key="deprecated_yaml",
translation_placeholders={
"domain": DOMAIN,
"integration_title": "Satel Integra",
},
)
controller = AsyncSatel(host, port, hass.loop, zones, monitored_outputs, partitions)
hass.data[DATA_SATEL] = controller
async def async_setup_entry(hass: HomeAssistant, entry: SatelConfigEntry) -> bool:
"""Set up Satel Integra from a config entry."""
host = entry.data[CONF_HOST]
port = entry.data[CONF_PORT]
# Make sure we initialize the Satel controller with the configured entries to monitor
partitions = [
subentry.data[CONF_PARTITION_NUMBER]
for subentry in entry.subentries.values()
if subentry.subentry_type == SUBENTRY_TYPE_PARTITION
]
zones = [
subentry.data[CONF_ZONE_NUMBER]
for subentry in entry.subentries.values()
if subentry.subentry_type == SUBENTRY_TYPE_ZONE
]
outputs = [
subentry.data[CONF_OUTPUT_NUMBER]
for subentry in entry.subentries.values()
if subentry.subentry_type == SUBENTRY_TYPE_OUTPUT
]
switchable_outputs = [
subentry.data[CONF_SWITCHABLE_OUTPUT_NUMBER]
for subentry in entry.subentries.values()
if subentry.subentry_type == SUBENTRY_TYPE_SWITCHABLE_OUTPUT
]
monitored_outputs = outputs + switchable_outputs
controller = AsyncSatel(host, port, hass.loop, zones, monitored_outputs, partitions)
result = await controller.connect()
if not result:
return False
raise ConfigEntryNotReady("Controller failed to connect")
entry.runtime_data = controller
@callback
def _close(*_):
controller.close()
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, _close)
entry.async_on_unload(hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, _close))
_LOGGER.debug("Arm home config: %s, mode: %s ", conf, conf.get(CONF_ARM_HOME_MODE))
hass.async_create_task(
async_load_platform(hass, Platform.ALARM_CONTROL_PANEL, DOMAIN, conf, config)
)
hass.async_create_task(
async_load_platform(
hass,
Platform.BINARY_SENSOR,
DOMAIN,
{CONF_ZONES: zones, CONF_OUTPUTS: outputs},
config,
)
)
hass.async_create_task(
async_load_platform(
hass,
Platform.SWITCH,
DOMAIN,
{
CONF_SWITCHABLE_OUTPUTS: switchable_outputs,
CONF_DEVICE_CODE: conf.get(CONF_DEVICE_CODE),
},
config,
)
)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
@callback
def alarm_status_update_callback():
@@ -179,3 +229,13 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
)
return True
async def async_unload_entry(hass: HomeAssistant, entry: SatelConfigEntry) -> bool:
"""Unloading the Satel platforms."""
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
controller = entry.runtime_data
controller.close()
return unload_ok

51
homeassistant/components/satel_integra/alarm_control_panel.py
View File

@@ -14,46 +14,49 @@ from homeassistant.components.alarm_control_panel import (
AlarmControlPanelState,
CodeFormat,
)
from homeassistant.const import CONF_NAME
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import (
from .const import (
CONF_ARM_HOME_MODE,
CONF_DEVICE_PARTITIONS,
CONF_ZONE_NAME,
DATA_SATEL,
CONF_PARTITION_NUMBER,
SIGNAL_PANEL_MESSAGE,
SUBENTRY_TYPE_PARTITION,
SatelConfigEntry,
)
_LOGGER = logging.getLogger(__name__)
async def async_setup_platform(
async def async_setup_entry(
hass: HomeAssistant,
config: ConfigType,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
config_entry: SatelConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up for Satel Integra alarm panels."""
if not discovery_info:
return
configured_partitions = discovery_info[CONF_DEVICE_PARTITIONS]
controller = hass.data[DATA_SATEL]
controller = config_entry.runtime_data
devices = []
partition_subentries = filter(
lambda entry: entry.subentry_type == SUBENTRY_TYPE_PARTITION,
config_entry.subentries.values(),
)
for partition_num, device_config_data in configured_partitions.items():
zone_name = device_config_data[CONF_ZONE_NAME]
arm_home_mode = device_config_data.get(CONF_ARM_HOME_MODE)
device = SatelIntegraAlarmPanel(
controller, zone_name, arm_home_mode, partition_num
for subentry in partition_subentries:
partition_num = subentry.data[CONF_PARTITION_NUMBER]
zone_name = subentry.data[CONF_NAME]
arm_home_mode = subentry.data[CONF_ARM_HOME_MODE]
async_add_entities(
[
SatelIntegraAlarmPanel(
controller, zone_name, arm_home_mode, partition_num
)
],
config_subentry_id=subentry.subentry_id,
)
devices.append(device)
async_add_entities(devices)
class SatelIntegraAlarmPanel(AlarmControlPanelEntity):
@@ -66,7 +69,7 @@ class SatelIntegraAlarmPanel(AlarmControlPanelEntity):
| AlarmControlPanelEntityFeature.ARM_AWAY
)
def __init__(self, controller, name, arm_home_mode, partition_id):
def __init__(self, controller, name, arm_home_mode, partition_id) -> None:
"""Initialize the alarm panel."""
self._attr_name = name
self._attr_unique_id = f"satel_alarm_panel_{partition_id}"

90
homeassistant/components/satel_integra/binary_sensor.py
View File

@@ -6,61 +6,79 @@ from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
BinarySensorEntity,
)
from homeassistant.const import CONF_NAME
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import (
CONF_OUTPUTS,
CONF_ZONE_NAME,
from .const import (
CONF_OUTPUT_NUMBER,
CONF_ZONE_NUMBER,
CONF_ZONE_TYPE,
CONF_ZONES,
DATA_SATEL,
SIGNAL_OUTPUTS_UPDATED,
SIGNAL_ZONES_UPDATED,
SUBENTRY_TYPE_OUTPUT,
SUBENTRY_TYPE_ZONE,
SatelConfigEntry,
)
async def async_setup_platform(
async def async_setup_entry(
hass: HomeAssistant,
config: ConfigType,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
config_entry: SatelConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the Satel Integra binary sensor devices."""
if not discovery_info:
return
configured_zones = discovery_info[CONF_ZONES]
controller = hass.data[DATA_SATEL]
controller = config_entry.runtime_data
devices = []
zone_subentries = filter(
lambda entry: entry.subentry_type == SUBENTRY_TYPE_ZONE,
config_entry.subentries.values(),
)
for zone_num, device_config_data in configured_zones.items():
zone_type = device_config_data[CONF_ZONE_TYPE]
zone_name = device_config_data[CONF_ZONE_NAME]
device = SatelIntegraBinarySensor(
controller, zone_num, zone_name, zone_type, CONF_ZONES, SIGNAL_ZONES_UPDATED
for subentry in zone_subentries:
zone_num = subentry.data[CONF_ZONE_NUMBER]
zone_type = subentry.data[CONF_ZONE_TYPE]
zone_name = subentry.data[CONF_NAME]
async_add_entities(
[
SatelIntegraBinarySensor(
controller,
zone_num,
zone_name,
zone_type,
SUBENTRY_TYPE_ZONE,
SIGNAL_ZONES_UPDATED,
)
],
config_subentry_id=subentry.subentry_id,
)
devices.append(device)
configured_outputs = discovery_info[CONF_OUTPUTS]
output_subentries = filter(
lambda entry: entry.subentry_type == SUBENTRY_TYPE_OUTPUT,
config_entry.subentries.values(),
)
for zone_num, device_config_data in configured_outputs.items():
zone_type = device_config_data[CONF_ZONE_TYPE]
zone_name = device_config_data[CONF_ZONE_NAME]
device = SatelIntegraBinarySensor(
controller,
zone_num,
zone_name,
zone_type,
CONF_OUTPUTS,
SIGNAL_OUTPUTS_UPDATED,
for subentry in output_subentries:
output_num = subentry.data[CONF_OUTPUT_NUMBER]
ouput_type = subentry.data[CONF_ZONE_TYPE]
output_name = subentry.data[CONF_NAME]
async_add_entities(
[
SatelIntegraBinarySensor(
controller,
output_num,
output_name,
ouput_type,
SUBENTRY_TYPE_OUTPUT,
SIGNAL_OUTPUTS_UPDATED,
)
],
config_subentry_id=subentry.subentry_id,
)
devices.append(device)
async_add_entities(devices)
class SatelIntegraBinarySensor(BinarySensorEntity):

496
homeassistant/components/satel_integra/config_flow.py Normal file
View File

@@ -0,0 +1,496 @@
"""Config flow for Satel Integra."""
from __future__ import annotations
import logging
from typing import Any
from satel_integra.satel_integra import AsyncSatel
import voluptuous as vol
from homeassistant.components.binary_sensor import BinarySensorDeviceClass
from homeassistant.config_entries import (
ConfigEntry,
ConfigFlow,
ConfigFlowResult,
ConfigSubentryData,
ConfigSubentryFlow,
OptionsFlowWithReload,
SubentryFlowResult,
)
from homeassistant.const import CONF_CODE, CONF_HOST, CONF_NAME, CONF_PORT
from homeassistant.core import callback
from homeassistant.helpers import config_validation as cv, selector
from .const import (
CONF_ARM_HOME_MODE,
CONF_DEVICE_PARTITIONS,
CONF_OUTPUT_NUMBER,
CONF_OUTPUTS,
CONF_PARTITION_NUMBER,
CONF_SWITCHABLE_OUTPUT_NUMBER,
CONF_SWITCHABLE_OUTPUTS,
CONF_ZONE_NUMBER,
CONF_ZONE_TYPE,
CONF_ZONES,
DEFAULT_CONF_ARM_HOME_MODE,
DEFAULT_PORT,
DOMAIN,
SUBENTRY_TYPE_OUTPUT,
SUBENTRY_TYPE_PARTITION,
SUBENTRY_TYPE_SWITCHABLE_OUTPUT,
SUBENTRY_TYPE_ZONE,
SatelConfigEntry,
)
_LOGGER = logging.getLogger(__package__)
CONNECTION_SCHEMA = vol.Schema(
{
vol.Required(CONF_HOST): str,
vol.Required(CONF_PORT, default=DEFAULT_PORT): cv.port,
vol.Optional(CONF_CODE): cv.string,
}
)
CODE_SCHEMA = vol.Schema(
{
vol.Optional(CONF_CODE): cv.string,
}
)
PARTITION_SCHEMA = vol.Schema(
{
vol.Required(CONF_NAME): cv.string,
vol.Required(CONF_ARM_HOME_MODE, default=DEFAULT_CONF_ARM_HOME_MODE): vol.In(
[1, 2, 3]
),
}
)
ZONE_AND_OUTPUT_SCHEMA = vol.Schema(
{
vol.Required(CONF_NAME): cv.string,
vol.Required(
CONF_ZONE_TYPE, default=BinarySensorDeviceClass.MOTION
): selector.SelectSelector(
selector.SelectSelectorConfig(
options=[cls.value for cls in BinarySensorDeviceClass],
mode=selector.SelectSelectorMode.DROPDOWN,
translation_key="binary_sensor_device_class",
sort=True,
),
),
}
)
SWITCHABLE_OUTPUT_SCHEMA = vol.Schema({vol.Required(CONF_NAME): cv.string})
class SatelConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a Satel Integra config flow."""
VERSION = 1
@staticmethod
@callback
def async_get_options_flow(
config_entry: SatelConfigEntry,
) -> SatelOptionsFlow:
"""Create the options flow."""
return SatelOptionsFlow()
@classmethod
@callback
def async_get_supported_subentry_types(
cls, config_entry: ConfigEntry
) -> dict[str, type[ConfigSubentryFlow]]:
"""Return subentries supported by this integration."""
return {
SUBENTRY_TYPE_PARTITION: PartitionSubentryFlowHandler,
SUBENTRY_TYPE_ZONE: ZoneSubentryFlowHandler,
SUBENTRY_TYPE_OUTPUT: OutputSubentryFlowHandler,
SUBENTRY_TYPE_SWITCHABLE_OUTPUT: SwitchableOutputSubentryFlowHandler,
}
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle a flow initialized by the user."""
errors: dict[str, str] = {}
if user_input is not None:
valid = await self.test_connection(
user_input[CONF_HOST], user_input[CONF_PORT]
)
if valid:
return self.async_create_entry(
title=user_input[CONF_HOST],
data={
CONF_HOST: user_input[CONF_HOST],
CONF_PORT: user_input[CONF_PORT],
},
options={CONF_CODE: user_input.get(CONF_CODE)},
)
errors["base"] = "cannot_connect"
return self.async_show_form(
step_id="user", data_schema=CONNECTION_SCHEMA, errors=errors
)
async def async_step_import(
self, import_config: dict[str, Any]
) -> ConfigFlowResult:
"""Handle a flow initialized by import."""
valid = await self.test_connection(
import_config[CONF_HOST], import_config.get(CONF_PORT, DEFAULT_PORT)
)
if valid:
subentries: list[ConfigSubentryData] = []
for partition_number, partition_data in import_config.get(
CONF_DEVICE_PARTITIONS, {}
).items():
subentries.append(
{
"subentry_type": SUBENTRY_TYPE_PARTITION,
"title": partition_data[CONF_NAME],
"unique_id": f"{SUBENTRY_TYPE_PARTITION}_{partition_number}",
"data": {
CONF_NAME: partition_data[CONF_NAME],
CONF_ARM_HOME_MODE: partition_data.get(
CONF_ARM_HOME_MODE, DEFAULT_CONF_ARM_HOME_MODE
),
CONF_PARTITION_NUMBER: partition_number,
},
}
)
for zone_number, zone_data in import_config.get(CONF_ZONES, {}).items():
subentries.append(
{
"subentry_type": SUBENTRY_TYPE_ZONE,
"title": zone_data[CONF_NAME],
"unique_id": f"{SUBENTRY_TYPE_ZONE}_{zone_number}",
"data": {
CONF_NAME: zone_data[CONF_NAME],
CONF_ZONE_NUMBER: zone_number,
CONF_ZONE_TYPE: zone_data.get(
CONF_ZONE_TYPE, BinarySensorDeviceClass.MOTION
),
},
}
)
for output_number, output_data in import_config.get(
CONF_OUTPUTS, {}
).items():
subentries.append(
{
"subentry_type": SUBENTRY_TYPE_OUTPUT,
"title": output_data[CONF_NAME],
"unique_id": f"{SUBENTRY_TYPE_OUTPUT}_{output_number}",
"data": {
CONF_NAME: output_data[CONF_NAME],
CONF_OUTPUT_NUMBER: output_number,
CONF_ZONE_TYPE: output_data.get(
CONF_ZONE_TYPE, BinarySensorDeviceClass.MOTION
),
},
}
)
for switchable_output_number, switchable_output_data in import_config.get(
CONF_SWITCHABLE_OUTPUTS, {}
).items():
subentries.append(
{
"subentry_type": SUBENTRY_TYPE_SWITCHABLE_OUTPUT,
"title": switchable_output_data[CONF_NAME],
"unique_id": f"{SUBENTRY_TYPE_SWITCHABLE_OUTPUT}_{switchable_output_number}",
"data": {
CONF_NAME: switchable_output_data[CONF_NAME],
CONF_SWITCHABLE_OUTPUT_NUMBER: switchable_output_number,
},
}
)
return self.async_create_entry(
title=import_config[CONF_HOST],
data={
CONF_HOST: import_config[CONF_HOST],
CONF_PORT: import_config.get(CONF_PORT, DEFAULT_PORT),
},
options={CONF_CODE: import_config.get(CONF_CODE)},
subentries=subentries,
)
return self.async_abort(reason="cannot_connect")
async def test_connection(self, host: str, port: int) -> bool:
"""Test a connection to the Satel alarm."""
controller = AsyncSatel(host, port, self.hass.loop)
result = await controller.connect()
# Make sure we close the connection again
controller.close()
return result
class SatelOptionsFlow(OptionsFlowWithReload):
"""Handle Satel options flow."""
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Init step."""
if user_input is not None:
return self.async_create_entry(data={CONF_CODE: user_input.get(CONF_CODE)})
return self.async_show_form(
step_id="init",
data_schema=self.add_suggested_values_to_schema(
CODE_SCHEMA, self.config_entry.options
),
)
class PartitionSubentryFlowHandler(ConfigSubentryFlow):
"""Handle subentry flow for adding and modifying a partition."""
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> SubentryFlowResult:
"""User flow to add new partition."""
errors: dict[str, str] = {}
if user_input is not None:
unique_id = f"{SUBENTRY_TYPE_PARTITION}_{user_input[CONF_PARTITION_NUMBER]}"
for existing_subentry in self._get_entry().subentries.values():
if existing_subentry.unique_id == unique_id:
errors[CONF_PARTITION_NUMBER] = "already_configured"
if not errors:
return self.async_create_entry(
title=user_input[CONF_NAME], data=user_input, unique_id=unique_id
)
return self.async_show_form(
step_id="user",
errors=errors,
data_schema=vol.Schema(
{
vol.Required(CONF_PARTITION_NUMBER): vol.All(
vol.Coerce(int), vol.Range(min=1)
),
}
).extend(PARTITION_SCHEMA.schema),
)
async def async_step_reconfigure(
self, user_input: dict[str, Any] | None = None
) -> SubentryFlowResult:
"""Reconfigure existing partition."""
subconfig_entry = self._get_reconfigure_subentry()
if user_input is not None:
return self.async_update_and_abort(
self._get_entry(),
subconfig_entry,
title=user_input[CONF_NAME],
data_updates=user_input,
)
return self.async_show_form(
step_id="reconfigure",
data_schema=self.add_suggested_values_to_schema(
PARTITION_SCHEMA,
subconfig_entry.data,
),
description_placeholders={
CONF_PARTITION_NUMBER: subconfig_entry.data[CONF_PARTITION_NUMBER]
},
)
class ZoneSubentryFlowHandler(ConfigSubentryFlow):
"""Handle subentry flow for adding and modifying a zone."""
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> SubentryFlowResult:
"""User flow to add new zone."""
errors: dict[str, str] = {}
if user_input is not None:
unique_id = f"{SUBENTRY_TYPE_ZONE}_{user_input[CONF_ZONE_NUMBER]}"
for existing_subentry in self._get_entry().subentries.values():
if existing_subentry.unique_id == unique_id:
errors[CONF_ZONE_NUMBER] = "already_configured"
if not errors:
return self.async_create_entry(
title=user_input[CONF_NAME], data=user_input, unique_id=unique_id
)
return self.async_show_form(
step_id="user",
errors=errors,
data_schema=vol.Schema(
{
vol.Required(CONF_ZONE_NUMBER): vol.All(
vol.Coerce(int), vol.Range(min=1)
),
}
).extend(ZONE_AND_OUTPUT_SCHEMA.schema),
)
async def async_step_reconfigure(
self, user_input: dict[str, Any] | None = None
) -> SubentryFlowResult:
"""Reconfigure existing zone."""
subconfig_entry = self._get_reconfigure_subentry()
if user_input is not None:
return self.async_update_and_abort(
self._get_entry(),
subconfig_entry,
title=user_input[CONF_NAME],
data_updates=user_input,
)
return self.async_show_form(
step_id="reconfigure",
data_schema=self.add_suggested_values_to_schema(
ZONE_AND_OUTPUT_SCHEMA, subconfig_entry.data
),
description_placeholders={
CONF_ZONE_NUMBER: subconfig_entry.data[CONF_ZONE_NUMBER]
},
)
class OutputSubentryFlowHandler(ConfigSubentryFlow):
"""Handle subentry flow for adding and modifying a output."""
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> SubentryFlowResult:
"""User flow to add new output."""
errors: dict[str, str] = {}
if user_input is not None:
unique_id = f"{SUBENTRY_TYPE_OUTPUT}_{user_input[CONF_OUTPUT_NUMBER]}"
for existing_subentry in self._get_entry().subentries.values():
if existing_subentry.unique_id == unique_id:
errors[CONF_OUTPUT_NUMBER] = "already_configured"
if not errors:
return self.async_create_entry(
title=user_input[CONF_NAME], data=user_input, unique_id=unique_id
)
return self.async_show_form(
step_id="user",
errors=errors,
data_schema=vol.Schema(
{
vol.Required(CONF_OUTPUT_NUMBER): vol.All(
vol.Coerce(int), vol.Range(min=1)
),
}
).extend(ZONE_AND_OUTPUT_SCHEMA.schema),
)
async def async_step_reconfigure(
self, user_input: dict[str, Any] | None = None
) -> SubentryFlowResult:
"""Reconfigure existing output."""
subconfig_entry = self._get_reconfigure_subentry()
if user_input is not None:
return self.async_update_and_abort(
self._get_entry(),
subconfig_entry,
title=user_input[CONF_NAME],
data_updates=user_input,
)
return self.async_show_form(
step_id="reconfigure",
data_schema=self.add_suggested_values_to_schema(
ZONE_AND_OUTPUT_SCHEMA, subconfig_entry.data
),
description_placeholders={
CONF_OUTPUT_NUMBER: subconfig_entry.data[CONF_OUTPUT_NUMBER]
},
)
class SwitchableOutputSubentryFlowHandler(ConfigSubentryFlow):
"""Handle subentry flow for adding and modifying a switchable output."""
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> SubentryFlowResult:
"""User flow to add new switchable output."""
errors: dict[str, str] = {}
if user_input is not None:
unique_id = f"{SUBENTRY_TYPE_SWITCHABLE_OUTPUT}_{user_input[CONF_SWITCHABLE_OUTPUT_NUMBER]}"
for existing_subentry in self._get_entry().subentries.values():
if existing_subentry.unique_id == unique_id:
errors[CONF_SWITCHABLE_OUTPUT_NUMBER] = "already_configured"
if not errors:
return self.async_create_entry(
title=user_input[CONF_NAME], data=user_input, unique_id=unique_id
)
return self.async_show_form(
step_id="user",
errors=errors,
data_schema=vol.Schema(
{
vol.Required(CONF_SWITCHABLE_OUTPUT_NUMBER): vol.All(
vol.Coerce(int), vol.Range(min=1)
),
}
).extend(SWITCHABLE_OUTPUT_SCHEMA.schema),
)
async def async_step_reconfigure(
self, user_input: dict[str, Any] | None = None
) -> SubentryFlowResult:
"""Reconfigure existing switchable output."""
subconfig_entry = self._get_reconfigure_subentry()
if user_input is not None:
return self.async_update_and_abort(
self._get_entry(),
subconfig_entry,
title=user_input[CONF_NAME],
data_updates=user_input,
)
return self.async_show_form(
step_id="reconfigure",
data_schema=self.add_suggested_values_to_schema(
SWITCHABLE_OUTPUT_SCHEMA, subconfig_entry.data
),
description_placeholders={
CONF_SWITCHABLE_OUTPUT_NUMBER: subconfig_entry.data[
CONF_SWITCHABLE_OUTPUT_NUMBER
]
},
)

38
homeassistant/components/satel_integra/const.py Normal file
View File

@@ -0,0 +1,38 @@
"""Constants for the Satel Integra integration."""
from satel_integra.satel_integra import AsyncSatel
from homeassistant.config_entries import ConfigEntry
DEFAULT_CONF_ARM_HOME_MODE = 1
DEFAULT_PORT = 7094
DEFAULT_ZONE_TYPE = "motion"
DOMAIN = "satel_integra"
SUBENTRY_TYPE_PARTITION = "partition"
SUBENTRY_TYPE_ZONE = "zone"
SUBENTRY_TYPE_OUTPUT = "output"
SUBENTRY_TYPE_SWITCHABLE_OUTPUT = "switchable_output"
CONF_PARTITION_NUMBER = "partition_number"
CONF_ZONE_NUMBER = "zone_number"
CONF_OUTPUT_NUMBER = "output_number"
CONF_SWITCHABLE_OUTPUT_NUMBER = "switchable_output_number"
CONF_DEVICE_PARTITIONS = "partitions"
CONF_ARM_HOME_MODE = "arm_home_mode"
CONF_ZONE_TYPE = "type"
CONF_ZONES = "zones"
CONF_OUTPUTS = "outputs"
CONF_SWITCHABLE_OUTPUTS = "switchable_outputs"
ZONES = "zones"
SIGNAL_PANEL_MESSAGE = "satel_integra.panel_message"
SIGNAL_ZONES_UPDATED = "satel_integra.zones_updated"
SIGNAL_OUTPUTS_UPDATED = "satel_integra.outputs_updated"
type SatelConfigEntry = ConfigEntry[AsyncSatel]

6
homeassistant/components/satel_integra/manifest.json
View File

@@ -1,10 +1,12 @@
{
"domain": "satel_integra",
"name": "Satel Integra",
"codeowners": [],
"codeowners": ["@Tommatheussen"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/satel_integra",
"iot_class": "local_push",
"loggers": ["satel_integra"],
"quality_scale": "legacy",
"requirements": ["satel-integra==0.3.7"]
"requirements": ["satel-integra==0.3.7"],
"single_config_entry": true
}

210
homeassistant/components/satel_integra/strings.json Normal file
View File

@@ -0,0 +1,210 @@
{
"common": {
"code_input_description": "Code to toggle switchable outputs",
"code": "Access code"
},
"config": {
"step": {
"user": {
"data": {
"host": "[%key:common::config_flow::data::host%]",
"port": "[%key:common::config_flow::data::port%]",
"code": "[%key:component::satel_integra::common::code%]"
},
"data_description": {
"host": "The IP address of the alarm panel",
"port": "The port of the alarm panel",
"code": "[%key:component::satel_integra::common::code_input_description%]"
}
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]"
}
},
"config_subentries": {
"partition": {
"initiate_flow": {
"user": "Add partition"
},
"step": {
"user": {
"title": "Configure partition",
"data": {
"partition_number": "Partition number",
"name": "[%key:common::config_flow::data::name%]",
"arm_home_mode": "Arm home mode"
},
"data_description": {
"partition_number": "Enter partition number to configure",
"name": "The name to give to the alarm panel",
"arm_home_mode": "The mode in which the partition is armed when 'arm home' is used. For more information on what the differences are between them, please refer to Satel Integra manual."
}
},
"reconfigure": {
"title": "Reconfigure partition {partition_number}",
"data": {
"name": "[%key:common::config_flow::data::name%]",
"arm_home_mode": "[%key:component::satel_integra::config_subentries::partition::step::user::data::arm_home_mode%]"
},
"data_description": {
"arm_home_mode": "[%key:component::satel_integra::config_subentries::partition::step::user::data_description::arm_home_mode%]"
}
}
},
"error": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
}
},
"zone": {
"initiate_flow": {
"user": "Add zone"
},
"step": {
"user": {
"title": "Configure zone",
"data": {
"zone_number": "Zone number",
"name": "[%key:common::config_flow::data::name%]",
"type": "Zone type"
},
"data_description": {
"zone_number": "Enter zone number to configure",
"name": "The name to give to the sensor",
"type": "Choose the device class you would like the sensor to show as"
}
},
"reconfigure": {
"title": "Reconfigure zone {zone_number}",
"data": {
"name": "[%key:common::config_flow::data::name%]",
"type": "[%key:component::satel_integra::config_subentries::zone::step::user::data::type%]"
},
"data_description": {
"name": "[%key:component::satel_integra::config_subentries::zone::step::user::data_description::name%]",
"type": "[%key:component::satel_integra::config_subentries::zone::step::user::data_description::type%]"
}
}
},
"error": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
}
},
"output": {
"initiate_flow": {
"user": "Add output"
},
"step": {
"user": {
"title": "Configure output",
"data": {
"output_number": "Output number",
"name": "[%key:common::config_flow::data::name%]",
"type": "Output type"
},
"data_description": {
"output_number": "Enter output number to configure",
"name": "[%key:component::satel_integra::config_subentries::zone::step::user::data_description::name%]",
"type": "[%key:component::satel_integra::config_subentries::zone::step::user::data_description::type%]"
}
},
"reconfigure": {
"title": "Reconfigure output {output_number}",
"data": {
"name": "[%key:common::config_flow::data::name%]",
"type": "[%key:component::satel_integra::config_subentries::output::step::user::data::type%]"
},
"data_description": {
"name": "[%key:component::satel_integra::config_subentries::zone::step::user::data_description::name%]",
"type": "[%key:component::satel_integra::config_subentries::zone::step::user::data_description::type%]"
}
}
},
"error": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
}
},
"switchable_output": {
"initiate_flow": {
"user": "Add switchable output"
},
"step": {
"user": {
"title": "Configure switchable output",
"data": {
"switchable_output_number": "Switchable output number",
"name": "[%key:common::config_flow::data::name%]"
},
"data_description": {
"switchable_output_number": "Enter switchable output number to configure",
"name": "The name to give to the switch"
}
},
"reconfigure": {
"title": "Reconfigure switchable output {switchable_output_number}",
"data": {
"name": "[%key:common::config_flow::data::name%]"
},
"data_description": {
"name": "[%key:component::satel_integra::config_subentries::switchable_output::step::user::data_description::name%]"
}
}
},
"error": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
}
}
},
"options": {
"step": {
"init": {
"data": {
"code": "[%key:component::satel_integra::common::code%]"
},
"data_description": {
"code": "[%key:component::satel_integra::common::code_input_description%]"
}
}
}
},
"issues": {
"deprecated_yaml_import_issue_cannot_connect": {
"title": "YAML import failed due to a connection error",
"description": "Configuring {integration_title} using YAML is being removed but there was an connection error importing your existing configuration.\n\nEnsure connection to {integration_title} works and restart Home Assistant to try again or remove the `{domain}` YAML configuration from your configuration.yaml file and add the {integration_title} integration manually."
}
},
"selector": {
"binary_sensor_device_class": {
"options": {
"battery": "[%key:component::binary_sensor::entity_component::battery::name%]",
"battery_charging": "[%key:component::binary_sensor::entity_component::battery_charging::name%]",
"carbon_monoxide": "[%key:component::binary_sensor::entity_component::carbon_monoxide::name%]",
"cold": "[%key:component::binary_sensor::entity_component::cold::name%]",
"connectivity": "[%key:component::binary_sensor::entity_component::connectivity::name%]",
"door": "[%key:component::binary_sensor::entity_component::door::name%]",
"garage_door": "[%key:component::binary_sensor::entity_component::garage_door::name%]",
"gas": "[%key:component::binary_sensor::entity_component::gas::name%]",
"heat": "[%key:component::binary_sensor::entity_component::heat::name%]",
"light": "[%key:component::binary_sensor::entity_component::light::name%]",
"lock": "[%key:component::binary_sensor::entity_component::lock::name%]",
"moisture": "[%key:component::binary_sensor::entity_component::moisture::name%]",
"motion": "[%key:component::binary_sensor::entity_component::motion::name%]",
"moving": "[%key:component::binary_sensor::entity_component::moving::name%]",
"occupancy": "[%key:component::binary_sensor::entity_component::occupancy::name%]",
"opening": "[%key:component::binary_sensor::entity_component::opening::name%]",
"plug": "[%key:component::binary_sensor::entity_component::plug::name%]",
"power": "[%key:component::binary_sensor::entity_component::power::name%]",
"presence": "[%key:component::binary_sensor::entity_component::presence::name%]",
"problem": "[%key:component::binary_sensor::entity_component::problem::name%]",
"running": "[%key:component::binary_sensor::entity_component::running::name%]",
"safety": "[%key:component::binary_sensor::entity_component::safety::name%]",
"smoke": "[%key:component::binary_sensor::entity_component::smoke::name%]",
"sound": "[%key:component::binary_sensor::entity_component::sound::name%]",
"tamper": "[%key:component::binary_sensor::entity_component::tamper::name%]",
"update": "[%key:component::binary_sensor::entity_component::update::name%]",
"vibration": "[%key:component::binary_sensor::entity_component::vibration::name%]",
"window": "[%key:component::binary_sensor::entity_component::window::name%]"
}
}
}
}

52
homeassistant/components/satel_integra/switch.py
View File

@@ -6,48 +6,50 @@ import logging
from typing import Any
from homeassistant.components.switch import SwitchEntity
from homeassistant.const import CONF_CODE, CONF_NAME
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import (
CONF_DEVICE_CODE,
CONF_SWITCHABLE_OUTPUTS,
CONF_ZONE_NAME,
DATA_SATEL,
from .const import (
CONF_SWITCHABLE_OUTPUT_NUMBER,
SIGNAL_OUTPUTS_UPDATED,
SUBENTRY_TYPE_SWITCHABLE_OUTPUT,
SatelConfigEntry,
)
_LOGGER = logging.getLogger(__name__)
DEPENDENCIES = ["satel_integra"]
async def async_setup_platform(
async def async_setup_entry(
hass: HomeAssistant,
config: ConfigType,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
config_entry: SatelConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the Satel Integra switch devices."""
if not discovery_info:
return
configured_zones = discovery_info[CONF_SWITCHABLE_OUTPUTS]
controller = hass.data[DATA_SATEL]
controller = config_entry.runtime_data
devices = []
switchable_output_subentries = filter(
lambda entry: entry.subentry_type == SUBENTRY_TYPE_SWITCHABLE_OUTPUT,
config_entry.subentries.values(),
)
for zone_num, device_config_data in configured_zones.items():
zone_name = device_config_data[CONF_ZONE_NAME]
for subentry in switchable_output_subentries:
switchable_output_num = subentry.data[CONF_SWITCHABLE_OUTPUT_NUMBER]
switchable_output_name = subentry.data[CONF_NAME]
device = SatelIntegraSwitch(
controller, zone_num, zone_name, discovery_info[CONF_DEVICE_CODE]
async_add_entities(
[
SatelIntegraSwitch(
controller,
switchable_output_num,
switchable_output_name,
config_entry.options.get(CONF_CODE),
),
],
config_subentry_id=subentry.subentry_id,
)
devices.append(device)
async_add_entities(devices)
class SatelIntegraSwitch(SwitchEntity):

1
homeassistant/generated/config_flows.py generated
View File

@@ -559,6 +559,7 @@ FLOWS = {
"sabnzbd",
"samsungtv",
"sanix",
"satel_integra",
"schlage",
"scrape",
"screenlogic",

5
homeassistant/generated/integrations.json
View File

@@ -5728,8 +5728,9 @@
"satel_integra": {
"name": "Satel Integra",
"integration_type": "hub",
"config_flow": false,
"iot_class": "local_push"
"config_flow": true,
"iot_class": "local_push",
"single_config_entry": true
},
"schlage": {
"name": "Schlage",

3
requirements_test_all.txt generated
View File

@@ -2272,6 +2272,9 @@ samsungtvws[async,encrypted]==2.7.2
# homeassistant.components.sanix
sanix==1.0.6
# homeassistant.components.satel_integra
satel-integra==0.3.7
# homeassistant.components.screenlogic
screenlogicpy==0.10.2

1
tests/components/satel_integra/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""The tests for Satel Integra integration."""

49
tests/components/satel_integra/conftest.py Normal file
View File

@@ -0,0 +1,49 @@
"""Satel Integra tests configuration."""
from collections.abc import Generator
from unittest.mock import AsyncMock, patch
import pytest
from homeassistant.components.satel_integra.const import DEFAULT_PORT, DOMAIN
from homeassistant.const import CONF_HOST, CONF_PORT
from tests.common import MockConfigEntry
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Override integration setup."""
with patch(
"homeassistant.components.satel_integra.async_setup_entry",
return_value=True,
) as mock_setup_entry:
yield mock_setup_entry
@pytest.fixture
def mock_satel() -> Generator[AsyncMock]:
"""Override the satel test."""
with (
patch(
"homeassistant.components.satel_integra.AsyncSatel",
autospec=True,
) as mock_client,
patch(
"homeassistant.components.satel_integra.config_flow.AsyncSatel",
new=mock_client,
),
):
client = mock_client.return_value
yield client
@pytest.fixture(name="config_entry")
def mock_config_entry() -> MockConfigEntry:
"""Mock satel configuration entry."""
return MockConfigEntry(
domain=DOMAIN,
title="192.168.0.2",
data={CONF_HOST: "192.168.0.2", CONF_PORT: DEFAULT_PORT},
)

593
tests/components/satel_integra/test_config_flow.py Normal file
View File

@@ -0,0 +1,593 @@
"""Test the satel integra config flow."""
from typing import Any
from unittest.mock import AsyncMock
import pytest
from homeassistant.components.binary_sensor import BinarySensorDeviceClass
from homeassistant.components.satel_integra.const import (
CONF_ARM_HOME_MODE,
CONF_DEVICE_PARTITIONS,
CONF_OUTPUT_NUMBER,
CONF_OUTPUTS,
CONF_PARTITION_NUMBER,
CONF_SWITCHABLE_OUTPUT_NUMBER,
CONF_SWITCHABLE_OUTPUTS,
CONF_ZONE_NUMBER,
CONF_ZONE_TYPE,
CONF_ZONES,
DEFAULT_PORT,
DOMAIN,
SUBENTRY_TYPE_OUTPUT,
SUBENTRY_TYPE_PARTITION,
SUBENTRY_TYPE_SWITCHABLE_OUTPUT,
SUBENTRY_TYPE_ZONE,
)
from homeassistant.config_entries import (
SOURCE_IMPORT,
SOURCE_RECONFIGURE,
SOURCE_USER,
ConfigSubentry,
ConfigSubentryData,
)
from homeassistant.const import CONF_CODE, CONF_HOST, CONF_NAME, CONF_PORT
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from tests.common import MockConfigEntry
CONST_HOST = "192.168.0.2"
CONST_PORT = 7095
CONST_CODE = "1234"
@pytest.mark.parametrize(
("user_input", "entry_data", "entry_options"),
[
(
{CONF_HOST: CONST_HOST, CONF_PORT: CONST_PORT, CONF_CODE: CONST_CODE},
{CONF_HOST: CONST_HOST, CONF_PORT: CONST_PORT},
{CONF_CODE: CONST_CODE},
),
(
{
CONF_HOST: CONST_HOST,
},
{CONF_HOST: CONST_HOST, CONF_PORT: DEFAULT_PORT},
{CONF_CODE: None},
),
],
)
async def test_setup_flow(
hass: HomeAssistant,
mock_satel: AsyncMock,
mock_setup_entry: AsyncMock,
user_input: dict[str, Any],
entry_data: dict[str, Any],
entry_options: dict[str, Any],
) -> None:
"""Test the setup flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input,
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == CONST_HOST
assert result["data"] == entry_data
assert result["options"] == entry_options
assert len(mock_setup_entry.mock_calls) == 1
async def test_setup_connection_failed(
hass: HomeAssistant, mock_satel: AsyncMock, mock_setup_entry: AsyncMock
) -> None:
"""Test the setup flow when connection fails."""
user_input = {CONF_HOST: CONST_HOST, CONF_PORT: CONST_PORT, CONF_CODE: CONST_CODE}
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
mock_satel.connect.return_value = False
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input,
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "cannot_connect"}
mock_satel.connect.return_value = True
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input,
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert len(mock_setup_entry.mock_calls) == 1
@pytest.mark.parametrize(
("import_input", "entry_data", "entry_options"),
[
(
{
CONF_HOST: CONST_HOST,
CONF_PORT: CONST_PORT,
CONF_CODE: CONST_CODE,
CONF_DEVICE_PARTITIONS: {
"1": {CONF_NAME: "Partition Import 1", CONF_ARM_HOME_MODE: 1}
},
CONF_ZONES: {
"1": {CONF_NAME: "Zone Import 1", CONF_ZONE_TYPE: "motion"},
"2": {CONF_NAME: "Zone Import 2", CONF_ZONE_TYPE: "door"},
},
CONF_OUTPUTS: {
"1": {CONF_NAME: "Output Import 1", CONF_ZONE_TYPE: "light"},
"2": {CONF_NAME: "Output Import 2", CONF_ZONE_TYPE: "safety"},
},
CONF_SWITCHABLE_OUTPUTS: {
"1": {CONF_NAME: "Switchable output Import 1"},
"2": {CONF_NAME: "Switchable output Import 2"},
},
},
{CONF_HOST: CONST_HOST, CONF_PORT: CONST_PORT},
{CONF_CODE: CONST_CODE},
)
],
)
async def test_import_flow(
hass: HomeAssistant,
mock_satel: AsyncMock,
mock_setup_entry: AsyncMock,
import_input: dict[str, Any],
entry_data: dict[str, Any],
entry_options: dict[str, Any],
) -> None:
"""Test the import flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=import_input
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == CONST_HOST
assert result["data"] == entry_data
assert result["options"] == entry_options
assert len(result["subentries"]) == 7
assert len(mock_setup_entry.mock_calls) == 1
async def test_import_flow_connection_failure(
hass: HomeAssistant, mock_satel: AsyncMock
) -> None:
"""Test the import flow."""
mock_satel.connect.return_value = False
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data={CONF_HOST: CONST_HOST, CONF_PORT: CONST_PORT, CONF_CODE: CONST_CODE},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "cannot_connect"
@pytest.mark.parametrize(
("user_input", "entry_options"),
[
(
{CONF_CODE: CONST_CODE},
{CONF_CODE: CONST_CODE},
),
({}, {CONF_CODE: None}),
],
)
async def test_options_flow(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
user_input: dict[str, Any],
entry_options: dict[str, Any],
) -> None:
"""Test general options flow."""
entry = MockConfigEntry(domain=DOMAIN)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.options.async_init(entry.entry_id)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "init"
result = await hass.config_entries.options.async_configure(
result["flow_id"], user_input
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert entry.options == entry_options
@pytest.mark.parametrize(
("subentry_type", "user_input", "subentry"),
[
(
SUBENTRY_TYPE_PARTITION,
{CONF_NAME: "Home", CONF_PARTITION_NUMBER: 1, CONF_ARM_HOME_MODE: 1},
{
"data": {
CONF_NAME: "Home",
CONF_ARM_HOME_MODE: 1,
CONF_PARTITION_NUMBER: 1,
},
"subentry_type": SUBENTRY_TYPE_PARTITION,
"title": "Home",
"unique_id": "partition_1",
},
),
(
SUBENTRY_TYPE_ZONE,
{
CONF_NAME: "Backdoor",
CONF_ZONE_TYPE: BinarySensorDeviceClass.DOOR,
CONF_ZONE_NUMBER: 2,
},
{
"data": {
CONF_NAME: "Backdoor",
CONF_ZONE_TYPE: BinarySensorDeviceClass.DOOR,
CONF_ZONE_NUMBER: 2,
},
"subentry_type": SUBENTRY_TYPE_ZONE,
"title": "Backdoor",
"unique_id": "zone_2",
},
),
(
SUBENTRY_TYPE_OUTPUT,
{
CONF_NAME: "Power outage",
CONF_ZONE_TYPE: BinarySensorDeviceClass.SAFETY,
CONF_OUTPUT_NUMBER: 1,
},
{
"data": {
CONF_NAME: "Power outage",
CONF_ZONE_TYPE: BinarySensorDeviceClass.SAFETY,
CONF_OUTPUT_NUMBER: 1,
},
"subentry_type": SUBENTRY_TYPE_OUTPUT,
"title": "Power outage",
"unique_id": "output_1",
},
),
(
SUBENTRY_TYPE_SWITCHABLE_OUTPUT,
{
CONF_NAME: "Gate",
CONF_SWITCHABLE_OUTPUT_NUMBER: 3,
},
{
"data": {
CONF_NAME: "Gate",
CONF_SWITCHABLE_OUTPUT_NUMBER: 3,
},
"subentry_type": SUBENTRY_TYPE_SWITCHABLE_OUTPUT,
"title": "Gate",
"unique_id": "switchable_output_3",
},
),
],
)
async def test_subentry_creation(
hass: HomeAssistant,
mock_satel: AsyncMock,
config_entry: MockConfigEntry,
subentry_type: str,
user_input: dict[str, Any],
subentry: dict[str, Any],
) -> None:
"""Test partitions options flow."""
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.subentries.async_init(
(config_entry.entry_id, subentry_type),
context={"source": SOURCE_USER},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
result = await hass.config_entries.subentries.async_configure(
result["flow_id"],
user_input,
)
assert len(config_entry.subentries) == 1
subentry_id = list(config_entry.subentries)[0]
subentry["subentry_id"] = subentry_id
assert config_entry.subentries == {subentry_id: ConfigSubentry(**subentry)}
@pytest.mark.parametrize(
(
"user_input",
"default_subentry_info",
"subentry",
"updated_subentry",
),
[
(
{CONF_NAME: "New Home", CONF_ARM_HOME_MODE: 3},
{
"subentry_id": "ABCD",
"subentry_type": SUBENTRY_TYPE_PARTITION,
"unique_id": "partition_1",
},
ConfigSubentryData(
data={
CONF_NAME: "Home",
CONF_ARM_HOME_MODE: 1,
CONF_PARTITION_NUMBER: 1,
},
title="Home",
),
ConfigSubentryData(
data={
CONF_NAME: "New Home",
CONF_ARM_HOME_MODE: 3,
CONF_PARTITION_NUMBER: 1,
},
title="New Home",
),
),
(
{CONF_NAME: "Backdoor", CONF_ZONE_TYPE: BinarySensorDeviceClass.DOOR},
{
"subentry_id": "ABCD",
"subentry_type": SUBENTRY_TYPE_ZONE,
"unique_id": "zone_1",
},
ConfigSubentryData(
data={
CONF_NAME: "Zone 1",
CONF_ZONE_TYPE: BinarySensorDeviceClass.MOTION,
CONF_ZONE_NUMBER: 1,
},
title="Zone 1",
),
ConfigSubentryData(
data={
CONF_NAME: "Backdoor",
CONF_ZONE_TYPE: BinarySensorDeviceClass.DOOR,
CONF_ZONE_NUMBER: 1,
},
title="Backdoor",
),
),
(
{
CONF_NAME: "Alarm Triggered",
CONF_ZONE_TYPE: BinarySensorDeviceClass.PROBLEM,
},
{
"subentry_id": "ABCD",
"subentry_type": SUBENTRY_TYPE_OUTPUT,
"unique_id": "output_1",
},
ConfigSubentryData(
data={
CONF_NAME: "Output 1",
CONF_ZONE_TYPE: BinarySensorDeviceClass.SAFETY,
CONF_OUTPUT_NUMBER: 1,
},
title="Output 1",
),
ConfigSubentryData(
data={
CONF_NAME: "Alarm Triggered",
CONF_ZONE_TYPE: BinarySensorDeviceClass.PROBLEM,
CONF_OUTPUT_NUMBER: 1,
},
title="Alarm Triggered",
),
),
(
{CONF_NAME: "Gate Lock"},
{
"subentry_id": "ABCD",
"subentry_type": SUBENTRY_TYPE_SWITCHABLE_OUTPUT,
"unique_id": "switchable_output_1",
},
ConfigSubentryData(
data={
CONF_NAME: "Switchable Output 1",
CONF_SWITCHABLE_OUTPUT_NUMBER: 1,
},
title="Switchable Output 1",
),
ConfigSubentryData(
data={
CONF_NAME: "Gate Lock",
CONF_SWITCHABLE_OUTPUT_NUMBER: 1,
},
title="Gate Lock",
),
),
],
)
async def test_subentry_reconfigure(
hass: HomeAssistant,
mock_satel: AsyncMock,
mock_setup_entry: AsyncMock,
config_entry: MockConfigEntry,
user_input: dict[str, Any],
default_subentry_info: dict[str, Any],
subentry: ConfigSubentryData,
updated_subentry: ConfigSubentryData,
) -> None:
"""Test subentry reconfiguration."""
config_entry.add_to_hass(hass)
config_entry.subentries = {
default_subentry_info["subentry_id"]: ConfigSubentry(
**default_subentry_info, **subentry
)
}
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.subentries.async_init(
(config_entry.entry_id, default_subentry_info["subentry_type"]),
context={
"source": SOURCE_RECONFIGURE,
"subentry_id": default_subentry_info["subentry_id"],
},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reconfigure"
result = await hass.config_entries.subentries.async_configure(
result["flow_id"],
user_input,
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reconfigure_successful"
assert len(config_entry.subentries) == 1
assert config_entry.subentries == {
default_subentry_info["subentry_id"]: ConfigSubentry(
**default_subentry_info, **updated_subentry
)
}
@pytest.mark.parametrize(
("subentry", "user_input", "error_field"),
[
(
{
"subentry_type": SUBENTRY_TYPE_PARTITION,
"unique_id": "partition_1",
"title": "Home",
},
{
CONF_NAME: "Home",
CONF_ARM_HOME_MODE: 1,
CONF_PARTITION_NUMBER: 1,
},
CONF_PARTITION_NUMBER,
),
(
{
"subentry_type": SUBENTRY_TYPE_ZONE,
"unique_id": "zone_1",
"title": "Zone 1",
},
{
CONF_NAME: "Zone 1",
CONF_ZONE_TYPE: BinarySensorDeviceClass.MOTION,
CONF_ZONE_NUMBER: 1,
},
CONF_ZONE_NUMBER,
),
(
{
"subentry_type": SUBENTRY_TYPE_OUTPUT,
"unique_id": "output_1",
"title": "Output 1",
},
{
CONF_NAME: "Output 1",
CONF_ZONE_TYPE: BinarySensorDeviceClass.SAFETY,
CONF_OUTPUT_NUMBER: 1,
},
CONF_OUTPUT_NUMBER,
),
(
{
"subentry_type": SUBENTRY_TYPE_SWITCHABLE_OUTPUT,
"unique_id": "switchable_output_1",
"title": "Switchable Output 1",
},
{
CONF_NAME: "Switchable Output 1",
CONF_SWITCHABLE_OUTPUT_NUMBER: 1,
},
CONF_SWITCHABLE_OUTPUT_NUMBER,
),
],
)
async def test_cannot_create_same_subentry(
hass: HomeAssistant,
mock_satel: AsyncMock,
mock_setup_entry: AsyncMock,
config_entry: MockConfigEntry,
subentry: dict[str, any],
user_input: dict[str, any],
error_field: str,
) -> None:
"""Test subentry reconfiguration."""
config_entry.add_to_hass(hass)
config_entry.subentries = {
"ABCD": ConfigSubentry(**subentry, **ConfigSubentryData({"data": user_input}))
}
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.subentries.async_init(
(config_entry.entry_id, subentry["subentry_type"]),
context={"source": SOURCE_USER},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
result = await hass.config_entries.subentries.async_configure(
result["flow_id"],
user_input,
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {error_field: "already_configured"}
assert len(config_entry.subentries) == 1
async def test_one_config_allowed(
hass: HomeAssistant, config_entry: MockConfigEntry
) -> None:
"""Test that only one Satel Integra configuration is allowed."""
config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "single_instance_allowed"