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

Add Config Flow for Ness Alarm (#162414)

Browse Source 
Co-authored-by: Joostlek <joostlek@outlook.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
Joshua Leaper
2026-02-19 09:46:08 +10:30
committed by GitHub
parent 14b147b3f7
commit 0f874f7f03
14 changed files with 1724 additions and 245 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
CODEOWNERS generated
View File

@@ -1098,8 +1098,8 @@ build.json @home-assistant/supervisor
/tests/components/nasweb/ @nasWebio
/homeassistant/components/nederlandse_spoorwegen/ @YarmoM @heindrichpaul
/tests/components/nederlandse_spoorwegen/ @YarmoM @heindrichpaul
/homeassistant/components/ness_alarm/ @nickw444
/tests/components/ness_alarm/ @nickw444
/homeassistant/components/ness_alarm/ @nickw444 @poshy163
/tests/components/ness_alarm/ @nickw444 @poshy163
/homeassistant/components/nest/ @allenporter
/tests/components/nest/ @allenporter
/homeassistant/components/netatmo/ @cgtobi

213
homeassistant/components/ness_alarm/__init__.py
View File

@@ -1,6 +1,7 @@
"""Support for Ness D8X/D16X devices."""
import datetime
from __future__ import annotations
import logging
from typing import NamedTuple
@@ -9,41 +10,41 @@ import voluptuous as vol
from homeassistant.components.binary_sensor import (
DEVICE_CLASSES_SCHEMA as BINARY_SENSOR_DEVICE_CLASSES_SCHEMA,
BinarySensorDeviceClass,
)
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import (
ATTR_CODE,
ATTR_STATE,
CONF_HOST,
CONF_PORT,
CONF_SCAN_INTERVAL,
EVENT_HOMEASSISTANT_STOP,
Platform,
)
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, Event, HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.discovery import async_load_platform
from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.helpers.issue_registry import IssueSeverity, async_create_issue
from homeassistant.helpers.start import async_at_started
from homeassistant.helpers.typing import ConfigType
from homeassistant.util.hass_dict import HassKey
from .const import (
CONF_INFER_ARMING_STATE,
CONF_ZONE_ID,
CONF_ZONE_NAME,
CONF_ZONE_TYPE,
CONF_ZONES,
DEFAULT_SCAN_INTERVAL,
DEFAULT_ZONE_TYPE,
DOMAIN,
PLATFORMS,
SIGNAL_ARMING_STATE_CHANGED,
SIGNAL_ZONE_CHANGED,
)
from .services import async_setup_services
_LOGGER = logging.getLogger(__name__)
DOMAIN = "ness_alarm"
DATA_NESS: HassKey[Client] = HassKey(DOMAIN)
CONF_DEVICE_PORT = "port"
CONF_INFER_ARMING_STATE = "infer_arming_state"
CONF_ZONES = "zones"
CONF_ZONE_NAME = "name"
CONF_ZONE_TYPE = "type"
CONF_ZONE_ID = "id"
ATTR_OUTPUT_ID = "output_id"
DEFAULT_SCAN_INTERVAL = datetime.timedelta(minutes=1)
DEFAULT_INFER_ARMING_STATE = False
SIGNAL_ZONE_CHANGED = "ness_alarm.zone_changed"
SIGNAL_ARMING_STATE_CHANGED = "ness_alarm.arming_state_changed"
type NessAlarmConfigEntry = ConfigEntry[Client]
class ZoneChangedData(NamedTuple):
@@ -53,7 +54,6 @@ class ZoneChangedData(NamedTuple):
state: bool
DEFAULT_ZONE_TYPE = BinarySensorDeviceClass.MOTION
ZONE_SCHEMA = vol.Schema(
{
vol.Required(CONF_ZONE_NAME): cv.string,
@@ -64,88 +64,111 @@ ZONE_SCHEMA = vol.Schema(
}
)
# YAML configuration is deprecated but supported for import
CONFIG_SCHEMA = vol.Schema(
{
DOMAIN: vol.Schema(
{
vol.Required(CONF_HOST): cv.string,
vol.Required(CONF_DEVICE_PORT): cv.port,
vol.Required(CONF_PORT): cv.port,
vol.Optional(
CONF_SCAN_INTERVAL, default=DEFAULT_SCAN_INTERVAL
): cv.positive_time_period,
vol.Optional(CONF_ZONES, default=[]): vol.All(
cv.ensure_list, [ZONE_SCHEMA]
),
vol.Optional(
CONF_INFER_ARMING_STATE, default=DEFAULT_INFER_ARMING_STATE
): cv.boolean,
vol.Optional(CONF_INFER_ARMING_STATE, default=False): cv.boolean,
}
)
},
extra=vol.ALLOW_EXTRA,
)
SERVICE_PANIC = "panic"
SERVICE_AUX = "aux"
SERVICE_SCHEMA_PANIC = vol.Schema({vol.Required(ATTR_CODE): cv.string})
SERVICE_SCHEMA_AUX = vol.Schema(
{
vol.Required(ATTR_OUTPUT_ID): cv.positive_int,
vol.Optional(ATTR_STATE, default=True): cv.boolean,
}
)
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the Ness Alarm platform."""
async_setup_services(hass)
if DOMAIN not in config:
return True
conf = config[DOMAIN]
hass.async_create_task(_async_setup(hass, config))
zones = conf[CONF_ZONES]
host = conf[CONF_HOST]
port = conf[CONF_DEVICE_PORT]
scan_interval = conf[CONF_SCAN_INTERVAL]
infer_arming_state = conf[CONF_INFER_ARMING_STATE]
return True
client = Client(
host=host,
port=port,
update_interval=scan_interval.total_seconds(),
infer_arming_state=infer_arming_state,
async def _async_setup(hass: HomeAssistant, config: ConfigType) -> None:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data=config[DOMAIN],
)
hass.data[DATA_NESS] = client
async def _close(event):
await client.close()
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, _close)
async def _started(event):
# Force update for current arming status and current zone states (once Home Assistant has finished loading required sensors and panel)
_LOGGER.debug("invoking client keepalive() & update()")
hass.loop.create_task(client.keepalive())
hass.loop.create_task(client.update())
async_at_started(hass, _started)
hass.async_create_task(
async_load_platform(
hass, Platform.BINARY_SENSOR, DOMAIN, {CONF_ZONES: zones}, config
if (
result.get("type") is FlowResultType.ABORT
and result.get("reason") != "already_configured"
):
async_create_issue(
hass,
DOMAIN,
f"deprecated_yaml_import_issue_{result.get('reason')}",
breaks_in_ha_version="2026.9.0",
is_fixable=False,
issue_domain=DOMAIN,
severity=IssueSeverity.WARNING,
translation_key=f"deprecated_yaml_import_issue_{result.get('reason')}",
translation_placeholders={
"domain": DOMAIN,
"integration_title": "Ness Alarm",
},
)
)
hass.async_create_task(
async_load_platform(hass, Platform.ALARM_CONTROL_PANEL, DOMAIN, {}, config)
return
async_create_issue(
hass,
HOMEASSISTANT_DOMAIN,
f"deprecated_yaml_{DOMAIN}",
breaks_in_ha_version="2026.9.0",
is_fixable=False,
issue_domain=DOMAIN,
severity=IssueSeverity.WARNING,
translation_key="deprecated_yaml",
translation_placeholders={
"domain": DOMAIN,
"integration_title": "Ness Alarm",
},
)
def on_zone_change(zone_id: int, state: bool):
"""Receives and propagates zone state updates."""
async def async_setup_entry(hass: HomeAssistant, entry: NessAlarmConfigEntry) -> bool:
"""Set up Ness Alarm from a config entry."""
client = Client(
host=entry.data[CONF_HOST],
port=entry.data[CONF_PORT],
update_interval=DEFAULT_SCAN_INTERVAL.total_seconds(),
infer_arming_state=entry.data.get(CONF_INFER_ARMING_STATE, False),
)
# Verify the client can connect to the alarm panel
try:
await client.update()
except OSError as err:
await client.close()
raise ConfigEntryNotReady(
f"Unable to connect to alarm panel at"
f" {entry.data[CONF_HOST]}:{entry.data[CONF_PORT]}"
) from err
entry.runtime_data = client
def on_zone_change(zone_id: int, state: bool) -> None:
"""Receive and propagate zone state updates."""
async_dispatcher_send(
hass, SIGNAL_ZONE_CHANGED, ZoneChangedData(zone_id=zone_id, state=state)
)
def on_state_change(arming_state: ArmingState, arming_mode: ArmingMode | None):
"""Receives and propagates arming state updates."""
def on_state_change(
arming_state: ArmingState, arming_mode: ArmingMode | None
) -> None:
"""Receive and propagate arming state updates."""
async_dispatcher_send(
hass, SIGNAL_ARMING_STATE_CHANGED, arming_state, arming_mode
)
@@ -153,17 +176,37 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
client.on_zone_change(on_zone_change)
client.on_state_change(on_state_change)
async def handle_panic(call: ServiceCall) -> None:
await client.panic(call.data[ATTR_CODE])
async def _close(event: Event) -> None:
await client.close()
async def handle_aux(call: ServiceCall) -> None:
await client.aux(call.data[ATTR_OUTPUT_ID], call.data[ATTR_STATE])
entry.async_on_unload(hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, _close))
hass.services.async_register(
DOMAIN, SERVICE_PANIC, handle_panic, schema=SERVICE_SCHEMA_PANIC
)
hass.services.async_register(
DOMAIN, SERVICE_AUX, handle_aux, schema=SERVICE_SCHEMA_AUX
)
async def _started(hass: HomeAssistant) -> None:
_LOGGER.debug("Invoking client keepalive() & update()")
hass.async_create_task(client.keepalive())
hass.async_create_task(client.update())
async_at_started(hass, _started)
# Forward to platforms
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
# Register update listener for options
entry.async_on_unload(entry.add_update_listener(async_reload_entry))
return True
async def async_unload_entry(hass: HomeAssistant, entry: NessAlarmConfigEntry) -> bool:
"""Unload a config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if unload_ok:
await entry.runtime_data.close()
return unload_ok
async def async_reload_entry(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Reload config entry when options change."""
await hass.config_entries.async_reload(entry.entry_id)

46
homeassistant/components/ness_alarm/alarm_control_panel.py
View File

@@ -13,11 +13,12 @@ from homeassistant.components.alarm_control_panel import (
CodeFormat,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.device_registry import DeviceInfo
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 DATA_NESS, SIGNAL_ARMING_STATE_CHANGED
from . import SIGNAL_ARMING_STATE_CHANGED, NessAlarmConfigEntry
from .const import CONF_SHOW_HOME_MODE, DOMAIN
_LOGGER = logging.getLogger(__name__)
@@ -31,18 +32,18 @@ ARMING_MODE_TO_STATE = {
}
async def async_setup_platform(
async def async_setup_entry(
hass: HomeAssistant,
config: ConfigType,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
entry: NessAlarmConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the Ness Alarm alarm control panel devices."""
if discovery_info is None:
return
"""Set up the Ness Alarm alarm control panel from config entry."""
client = entry.runtime_data
show_home_mode = entry.options.get(CONF_SHOW_HOME_MODE, True)
device = NessAlarmPanel(hass.data[DATA_NESS], "Alarm Panel")
async_add_entities([device])
async_add_entities(
[NessAlarmPanel(client, entry.entry_id, show_home_mode)],
)
class NessAlarmPanel(AlarmControlPanelEntity):
@@ -50,16 +51,23 @@ class NessAlarmPanel(AlarmControlPanelEntity):
_attr_code_format = CodeFormat.NUMBER
_attr_should_poll = False
_attr_supported_features = (
AlarmControlPanelEntityFeature.ARM_HOME
| AlarmControlPanelEntityFeature.ARM_AWAY
| AlarmControlPanelEntityFeature.TRIGGER
)
def __init__(self, client: Client, name: str) -> None:
def __init__(self, client: Client, entry_id: str, show_home_mode: bool) -> None:
"""Initialize the alarm panel."""
self._client = client
self._attr_name = name
self._attr_name = "Alarm Panel"
self._attr_unique_id = f"{entry_id}_alarm_panel"
self._attr_device_info = DeviceInfo(
name="Alarm Panel",
identifiers={(DOMAIN, f"{entry_id}_alarm_panel")},
)
features = (
AlarmControlPanelEntityFeature.ARM_AWAY
| AlarmControlPanelEntityFeature.TRIGGER
)
if show_home_mode:
features |= AlarmControlPanelEntityFeature.ARM_HOME
self._attr_supported_features = features
async def async_added_to_hass(self) -> None:
"""Register callbacks."""

75
homeassistant/components/ness_alarm/binary_sensor.py
View File

@@ -6,42 +6,54 @@ from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
BinarySensorEntity,
)
from homeassistant.const import CONF_TYPE
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.device_registry import DeviceInfo
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_ZONE_ID,
from . import SIGNAL_ZONE_CHANGED, NessAlarmConfigEntry, ZoneChangedData
from .const import (
CONF_ZONE_NAME,
CONF_ZONE_TYPE,
CONF_ZONES,
SIGNAL_ZONE_CHANGED,
ZoneChangedData,
CONF_ZONE_NUMBER,
DEFAULT_ZONE_TYPE,
DOMAIN,
SUBENTRY_TYPE_ZONE,
)
async def async_setup_platform(
async def async_setup_entry(
hass: HomeAssistant,
config: ConfigType,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
entry: NessAlarmConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the Ness Alarm binary sensor devices."""
if not discovery_info:
return
configured_zones = discovery_info[CONF_ZONES]
async_add_entities(
NessZoneBinarySensor(
zone_id=zone_config[CONF_ZONE_ID],
name=zone_config[CONF_ZONE_NAME],
zone_type=zone_config[CONF_ZONE_TYPE],
)
for zone_config in configured_zones
"""Set up the Ness Alarm binary sensor from config entry."""
# Get zone subentries
zone_subentries = filter(
lambda subentry: subentry.subentry_type == SUBENTRY_TYPE_ZONE,
entry.subentries.values(),
)
# Create entities from zone subentries
for subentry in zone_subentries:
zone_num: int = subentry.data[CONF_ZONE_NUMBER]
zone_type: BinarySensorDeviceClass = subentry.data.get(
CONF_TYPE, DEFAULT_ZONE_TYPE
)
zone_name: str | None = subentry.data.get(CONF_ZONE_NAME)
async_add_entities(
[
NessZoneBinarySensor(
zone_id=zone_num,
zone_type=zone_type,
entry_id=entry.entry_id,
zone_name=zone_name,
)
],
config_subentry_id=subentry.subentry_id,
)
class NessZoneBinarySensor(BinarySensorEntity):
"""Representation of an Ness alarm zone as a binary sensor."""
@@ -49,13 +61,22 @@ class NessZoneBinarySensor(BinarySensorEntity):
_attr_should_poll = False
def __init__(
self, zone_id: int, name: str, zone_type: BinarySensorDeviceClass
self,
zone_id: int,
zone_type: BinarySensorDeviceClass,
entry_id: str,
zone_name: str | None = None,
) -> None:
"""Initialize the binary_sensor."""
self._zone_id = zone_id
self._attr_name = name
self._attr_device_class = zone_type
self._attr_is_on = False
self._attr_unique_id = f"{entry_id}_zone_{zone_id}"
self._attr_name = f"Zone {zone_id}"
self._attr_device_info = DeviceInfo(
name=zone_name or f"Zone {zone_id}",
identifiers={(DOMAIN, self._attr_unique_id)},
)
async def async_added_to_hass(self) -> None:
"""Register callbacks."""

294
homeassistant/components/ness_alarm/config_flow.py Normal file
View File

@@ -0,0 +1,294 @@
"""Config flow for Ness Alarm integration."""
from __future__ import annotations
import asyncio
import logging
from types import MappingProxyType
from typing import Any
from nessclient import Client
import voluptuous as vol
from homeassistant.components.binary_sensor import BinarySensorDeviceClass
from homeassistant.config_entries import (
ConfigEntry,
ConfigFlow,
ConfigFlowResult,
ConfigSubentryData,
ConfigSubentryFlow,
OptionsFlow,
SubentryFlowResult,
)
from homeassistant.const import CONF_HOST, CONF_PORT, CONF_TYPE
from homeassistant.core import callback
from homeassistant.helpers import config_validation as cv, selector
from .const import (
CONF_INFER_ARMING_STATE,
CONF_SHOW_HOME_MODE,
CONF_ZONE_ID,
CONF_ZONE_NAME,
CONF_ZONE_NUMBER,
CONF_ZONE_TYPE,
CONF_ZONES,
CONNECTION_TIMEOUT,
DEFAULT_INFER_ARMING_STATE,
DEFAULT_PORT,
DEFAULT_ZONE_TYPE,
DOMAIN,
POST_CONNECTION_DELAY,
SUBENTRY_TYPE_ZONE,
)
_LOGGER = logging.getLogger(__name__)
STEP_USER_DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_HOST): str,
vol.Required(CONF_PORT, default=DEFAULT_PORT): cv.port,
vol.Optional(CONF_INFER_ARMING_STATE, default=DEFAULT_INFER_ARMING_STATE): bool,
}
)
ZONE_SCHEMA = vol.Schema(
{
vol.Required(CONF_TYPE, default=DEFAULT_ZONE_TYPE): selector.SelectSelector(
selector.SelectSelectorConfig(
options=[cls.value for cls in BinarySensorDeviceClass],
mode=selector.SelectSelectorMode.DROPDOWN,
translation_key="binary_sensor_device_class",
sort=True,
),
),
}
)
class NessAlarmConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Ness Alarm."""
VERSION = 1
@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_ZONE: ZoneSubentryFlowHandler,
}
@staticmethod
@callback
def async_get_options_flow(
config_entry: ConfigEntry,
) -> OptionsFlow:
"""Create the options flow."""
return NessAlarmOptionsFlowHandler()
async def _test_connection(self, host: str, port: int) -> None:
"""Test connection to the alarm panel.
Raises OSError on connection failure.
"""
client = Client(host=host, port=port)
try:
await asyncio.wait_for(client.update(), timeout=CONNECTION_TIMEOUT)
except TimeoutError as err:
raise OSError(f"Timed out connecting to {host}:{port}") from err
finally:
await client.close()
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the initial step."""
errors: dict[str, str] = {}
if user_input is not None:
host = user_input[CONF_HOST]
port = user_input[CONF_PORT]
# Check if already configured
self._async_abort_entries_match({CONF_HOST: host})
# Test connection to the alarm panel
try:
await self._test_connection(host, port)
except OSError:
errors["base"] = "cannot_connect"
except Exception:
_LOGGER.exception("Unexpected error connecting to %s:%s", host, port)
errors["base"] = "unknown"
if not errors:
# Brief delay to ensure the panel releases the test connection
await asyncio.sleep(POST_CONNECTION_DELAY)
return self.async_create_entry(
title=f"Ness Alarm {host}:{port}",
data=user_input,
)
return self.async_show_form(
step_id="user",
data_schema=STEP_USER_DATA_SCHEMA,
errors=errors,
)
async def async_step_import(self, import_data: dict[str, Any]) -> ConfigFlowResult:
"""Import YAML configuration."""
host = import_data[CONF_HOST]
port = import_data[CONF_PORT]
# Check if already configured
self._async_abort_entries_match({CONF_HOST: host})
# Test connection to the alarm panel
try:
await self._test_connection(host, port)
except OSError:
return self.async_abort(reason="cannot_connect")
except Exception:
_LOGGER.exception(
"Unexpected error connecting to %s:%s during import", host, port
)
return self.async_abort(reason="unknown")
# Brief delay to ensure the panel releases the test connection
await asyncio.sleep(POST_CONNECTION_DELAY)
# Prepare subentries for zones
subentries: list[ConfigSubentryData] = []
zones = import_data.get(CONF_ZONES, [])
for zone_config in zones:
zone_id = zone_config[CONF_ZONE_ID]
zone_name = zone_config.get(CONF_ZONE_NAME)
zone_type = zone_config.get(CONF_ZONE_TYPE, DEFAULT_ZONE_TYPE)
# Subentry title is always "Zone {zone_id}"
title = f"Zone {zone_id}"
# Build subentry data
subentry_data = {
CONF_ZONE_NUMBER: zone_id,
CONF_TYPE: zone_type,
}
# Include zone name in data if provided (for device naming)
if zone_name:
subentry_data[CONF_ZONE_NAME] = zone_name
subentries.append(
{
"subentry_type": SUBENTRY_TYPE_ZONE,
"title": title,
"unique_id": f"{SUBENTRY_TYPE_ZONE}_{zone_id}",
"data": MappingProxyType(subentry_data),
}
)
return self.async_create_entry(
title=f"Ness Alarm {host}:{port}",
data={
CONF_HOST: host,
CONF_PORT: port,
CONF_INFER_ARMING_STATE: import_data.get(
CONF_INFER_ARMING_STATE, DEFAULT_INFER_ARMING_STATE
),
},
subentries=subentries,
)
class NessAlarmOptionsFlowHandler(OptionsFlow):
"""Handle options flow for Ness Alarm."""
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Manage the options."""
if user_input is not None:
return self.async_create_entry(title="", data=user_input)
return self.async_show_form(
step_id="init",
data_schema=self.add_suggested_values_to_schema(
vol.Schema(
{
vol.Required(CONF_SHOW_HOME_MODE, default=True): bool,
}
),
self.config_entry.options,
),
)
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:
zone_number = int(user_input[CONF_ZONE_NUMBER])
unique_id = f"{SUBENTRY_TYPE_ZONE}_{zone_number}"
# Check if zone already exists
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:
# Store zone_number as int in data
user_input[CONF_ZONE_NUMBER] = zone_number
return self.async_create_entry(
title=f"Zone {zone_number}",
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): selector.NumberSelector(
selector.NumberSelectorConfig(
min=1,
max=32,
mode=selector.NumberSelectorMode.BOX,
)
),
}
).extend(ZONE_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=f"Zone {subconfig_entry.data[CONF_ZONE_NUMBER]}",
data_updates=user_input,
)
return self.async_show_form(
step_id="reconfigure",
data_schema=self.add_suggested_values_to_schema(
ZONE_SCHEMA,
subconfig_entry.data,
),
description_placeholders={
CONF_ZONE_NUMBER: str(subconfig_entry.data[CONF_ZONE_NUMBER])
},
)

42
homeassistant/components/ness_alarm/const.py Normal file
View File

@@ -0,0 +1,42 @@
"""Constants for the Ness Alarm integration."""
from datetime import timedelta
from homeassistant.components.binary_sensor import BinarySensorDeviceClass
from homeassistant.const import Platform
DOMAIN = "ness_alarm"
# Platforms
PLATFORMS = [Platform.ALARM_CONTROL_PANEL, Platform.BINARY_SENSOR]
# Configuration constants
CONF_INFER_ARMING_STATE = "infer_arming_state"
CONF_ZONES = "zones"
CONF_ZONE_NAME = "name"
CONF_ZONE_TYPE = "type"
CONF_ZONE_ID = "id"
CONF_ZONE_NUMBER = "zone_number"
CONF_SHOW_HOME_MODE = "show_home_mode"
# Subentry types
SUBENTRY_TYPE_ZONE = "zone"
# Defaults
DEFAULT_PORT = 4999
DEFAULT_SCAN_INTERVAL = timedelta(minutes=1)
DEFAULT_INFER_ARMING_STATE = False
DEFAULT_ZONE_TYPE = BinarySensorDeviceClass.MOTION
# Connection
CONNECTION_TIMEOUT = 5
POST_CONNECTION_DELAY = 1
# Signals
SIGNAL_ZONE_CHANGED = "ness_alarm.zone_changed"
SIGNAL_ARMING_STATE_CHANGED = "ness_alarm.arming_state_changed"
# Services
SERVICE_PANIC = "panic"
SERVICE_AUX = "aux"
ATTR_OUTPUT_ID = "output_id"

3
homeassistant/components/ness_alarm/manifest.json
View File

@@ -1,7 +1,8 @@
{
"domain": "ness_alarm",
"name": "Ness Alarm",
"codeowners": ["@nickw444"],
"codeowners": ["@nickw444", "@poshy163"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/ness_alarm",
"iot_class": "local_push",
"loggers": ["nessclient"],

53
homeassistant/components/ness_alarm/services.py Normal file
View File

@@ -0,0 +1,53 @@
"""Services for the Ness Alarm integration."""
from __future__ import annotations
import voluptuous as vol
from homeassistant.const import ATTR_CODE, ATTR_STATE
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.exceptions import ServiceValidationError
from homeassistant.helpers import config_validation as cv
from .const import ATTR_OUTPUT_ID, DOMAIN, SERVICE_AUX, SERVICE_PANIC
SERVICE_SCHEMA_PANIC = vol.Schema({vol.Required(ATTR_CODE): cv.string})
SERVICE_SCHEMA_AUX = vol.Schema(
{
vol.Required(ATTR_OUTPUT_ID): cv.positive_int,
vol.Optional(ATTR_STATE, default=True): cv.boolean,
}
)
def async_setup_services(hass: HomeAssistant) -> None:
"""Register Ness Alarm services."""
async def handle_panic(call: ServiceCall) -> None:
"""Handle panic service call."""
entries = call.hass.config_entries.async_loaded_entries(DOMAIN)
if not entries:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="no_config_entry",
)
client = entries[0].runtime_data
await client.panic(call.data[ATTR_CODE])
async def handle_aux(call: ServiceCall) -> None:
"""Handle aux service call."""
entries = call.hass.config_entries.async_loaded_entries(DOMAIN)
if not entries:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="no_config_entry",
)
client = entries[0].runtime_data
await client.aux(call.data[ATTR_OUTPUT_ID], call.data[ATTR_STATE])
hass.services.async_register(
DOMAIN, SERVICE_PANIC, handle_panic, schema=SERVICE_SCHEMA_PANIC
)
hass.services.async_register(
DOMAIN, SERVICE_AUX, handle_aux, schema=SERVICE_SCHEMA_AUX
)

87
homeassistant/components/ness_alarm/strings.json
View File

@@ -1,4 +1,91 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"step": {
"user": {
"data": {
"host": "[%key:common::config_flow::data::host%]",
"infer_arming_state": "Infer arming state",
"port": "[%key:common::config_flow::data::port%]"
},
"data_description": {
"host": "The IP address or hostname of your Ness alarm panel.",
"infer_arming_state": "Attempt to infer the arming state from zone activity.",
"port": "The port on which the Ness alarm panel is accessible."
},
"description": "Configure connection to your Ness D8X/D16X alarm panel.",
"title": "Set up Ness Alarm"
}
}
},
"config_subentries": {
"zone": {
"entry_type": "Zone",
"error": {
"already_configured": "Zone with this number is already configured"
},
"initiate_flow": {
"user": "Add zone"
},
"step": {
"reconfigure": {
"data": {
"type": "[%key:component::ness_alarm::config_subentries::zone::step::user::data::type%]"
},
"data_description": {
"type": "[%key:component::ness_alarm::config_subentries::zone::step::user::data_description::type%]"
},
"title": "Reconfigure zone {zone_number}"
},
"user": {
"data": {
"type": "Zone type",
"zone_number": "Zone number"
},
"data_description": {
"type": "Choose the device class you would like the sensor to show as",
"zone_number": "Enter zone number to configure (1-32)"
},
"title": "Configure zone"
}
}
}
},
"exceptions": {
"no_config_entry": {
"message": "No Ness Alarm configuration entry is loaded"
}
},
"issues": {
"deprecated_yaml_import_issue_cannot_connect": {
"description": "Configuring {integration_title} via YAML is deprecated and will be removed in a future release. While importing your configuration, a connection error occurred. Please correct your YAML configuration and restart Home Assistant, or remove the {domain} key from your configuration and configure the integration via the UI.",
"title": "The {integration_title} YAML configuration is being removed"
},
"deprecated_yaml_import_issue_unknown": {
"description": "Configuring {integration_title} via YAML is deprecated and will be removed in a future release. While importing your configuration, an unknown error occurred. Please correct your YAML configuration and restart Home Assistant, or remove the {domain} key from your configuration and configure the integration via the UI.",
"title": "The {integration_title} YAML configuration is being removed"
}
},
"options": {
"step": {
"init": {
"data": {
"show_home_mode": "Show arm home mode"
},
"data_description": {
"show_home_mode": "Enable this to show the arm home option on the alarm panel."
}
}
}
},
"services": {
"aux": {
"description": "Changes the state of an aux output.",

1
homeassistant/generated/config_flows.py generated
View File

@@ -459,6 +459,7 @@ FLOWS = {
"nasweb",
"neato",
"nederlandse_spoorwegen",
"ness_alarm",
"nest",
"netatmo",
"netgear",

2
homeassistant/generated/integrations.json
View File

@@ -4481,7 +4481,7 @@
"ness_alarm": {
"name": "Ness Alarm",
"integration_type": "hub",
"config_flow": false,
"config_flow": true,
"iot_class": "local_push"
},
"netatmo": {

104
tests/components/ness_alarm/conftest.py Normal file
View File

@@ -0,0 +1,104 @@
"""Test fixtures for ness_alarm."""
from collections.abc import Generator
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from homeassistant.components.ness_alarm.const import DOMAIN
from homeassistant.const import CONF_HOST, CONF_PORT
from tests.common import MockConfigEntry
class MockClient:
"""Mock nessclient.Client stub."""
async def panic(self, code):
"""Handle panic."""
async def disarm(self, code):
"""Handle disarm."""
async def arm_away(self, code):
"""Handle arm_away."""
async def arm_home(self, code):
"""Handle arm_home."""
async def aux(self, output_id, state):
"""Handle auxiliary control."""
async def keepalive(self):
"""Handle keepalive."""
async def update(self):
"""Handle update."""
def on_zone_change(self):
"""Handle on_zone_change."""
def on_state_change(self):
"""Handle on_state_change."""
async def close(self):
"""Handle close."""
@pytest.fixture
def mock_nessclient():
"""Mock the nessclient Client constructor.
Replaces nessclient.Client with a Mock which always returns the same
MagicMock() instance.
"""
_mock_instance = MagicMock(MockClient())
_mock_factory = MagicMock()
_mock_factory.return_value = _mock_instance
with patch(
"homeassistant.components.ness_alarm.Client", new=_mock_factory, create=True
):
yield _mock_instance
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Return a mock config entry."""
return MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "192.168.1.100",
CONF_PORT: 1992,
},
)
@pytest.fixture
def mock_client() -> Generator[AsyncMock]:
"""Mock the nessclient Client for config flow tests."""
with patch(
"homeassistant.components.ness_alarm.config_flow.Client",
return_value=AsyncMock(),
) as mock:
yield mock.return_value
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Mock async_setup_entry."""
with patch(
"homeassistant.components.ness_alarm.async_setup_entry",
return_value=True,
) as mock:
yield mock
@pytest.fixture(autouse=True)
def post_connection_delay() -> Generator[None]:
"""Mock POST_CONNECTION_DELAY to 0 for faster tests."""
with patch(
"homeassistant.components.ness_alarm.config_flow.POST_CONNECTION_DELAY",
0,
):
yield

454
tests/components/ness_alarm/test_config_flow.py Normal file
View File

@@ -0,0 +1,454 @@
"""Test the Ness Alarm config flow."""
from types import MappingProxyType
from unittest.mock import AsyncMock
import pytest
from homeassistant.components.binary_sensor import BinarySensorDeviceClass
from homeassistant.components.ness_alarm.const import (
CONF_INFER_ARMING_STATE,
CONF_SHOW_HOME_MODE,
CONF_ZONE_ID,
CONF_ZONE_NAME,
CONF_ZONE_NUMBER,
CONF_ZONE_TYPE,
CONF_ZONES,
DOMAIN,
SUBENTRY_TYPE_ZONE,
)
from homeassistant.config_entries import SOURCE_IMPORT, SOURCE_USER, ConfigSubentry
from homeassistant.const import CONF_HOST, CONF_PORT, CONF_TYPE
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from tests.common import MockConfigEntry
async def test_user_flow(
hass: HomeAssistant, mock_client: AsyncMock, mock_setup_entry: AsyncMock
) -> None:
"""Test successful user config flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_HOST: "192.168.1.100",
CONF_PORT: 1992,
CONF_INFER_ARMING_STATE: False,
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "Ness Alarm 192.168.1.100:1992"
assert result["data"] == {
CONF_HOST: "192.168.1.100",
CONF_PORT: 1992,
CONF_INFER_ARMING_STATE: False,
}
assert len(mock_setup_entry.mock_calls) == 1
mock_client.close.assert_awaited_once()
async def test_user_flow_with_infer_arming_state(
hass: HomeAssistant, mock_client: AsyncMock, mock_setup_entry: AsyncMock
) -> None:
"""Test user flow with infer_arming_state enabled."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_HOST: "192.168.1.100",
CONF_PORT: 1992,
CONF_INFER_ARMING_STATE: True,
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"][CONF_INFER_ARMING_STATE] is True
async def test_user_flow_already_configured(
hass: HomeAssistant, mock_config_entry: MockConfigEntry
) -> None:
"""Test we abort if already configured."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_HOST: "192.168.1.100",
CONF_PORT: 1992,
CONF_INFER_ARMING_STATE: False,
},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
@pytest.mark.parametrize(
("side_effect", "expected_error"),
[
(OSError("Connection refused"), "cannot_connect"),
(TimeoutError, "cannot_connect"),
(RuntimeError("Unexpected"), "unknown"),
],
)
async def test_user_flow_connection_error_recovery(
hass: HomeAssistant,
mock_client: AsyncMock,
mock_setup_entry: AsyncMock,
side_effect: Exception,
expected_error: str,
) -> None:
"""Test connection error handling and recovery."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
# First attempt fails
mock_client.update.side_effect = side_effect
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_HOST: "192.168.1.100",
CONF_PORT: 1992,
CONF_INFER_ARMING_STATE: False,
},
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": expected_error}
mock_client.close.assert_awaited_once()
# Second attempt succeeds
mock_client.update.side_effect = None
mock_client.close.reset_mock()
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_HOST: "192.168.1.100",
CONF_PORT: 1992,
CONF_INFER_ARMING_STATE: False,
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
async def test_import_yaml_config(
hass: HomeAssistant, mock_client: AsyncMock, mock_setup_entry: AsyncMock
) -> None:
"""Test importing YAML configuration."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data={
CONF_HOST: "192.168.1.72",
CONF_PORT: 4999,
CONF_INFER_ARMING_STATE: False,
CONF_ZONES: [
{CONF_ZONE_NAME: "Garage", CONF_ZONE_ID: 1},
{
CONF_ZONE_NAME: "Front Door",
CONF_ZONE_ID: 5,
CONF_ZONE_TYPE: BinarySensorDeviceClass.DOOR,
},
],
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "Ness Alarm 192.168.1.72:4999"
assert result["data"] == {
CONF_HOST: "192.168.1.72",
CONF_PORT: 4999,
CONF_INFER_ARMING_STATE: False,
}
# Check that subentries were created for zones with names preserved
assert len(result["subentries"]) == 2
assert result["subentries"][0]["title"] == "Zone 1"
assert result["subentries"][0]["unique_id"] == "zone_1"
assert result["subentries"][0]["data"][CONF_TYPE] == BinarySensorDeviceClass.MOTION
assert result["subentries"][0]["data"][CONF_ZONE_NAME] == "Garage"
assert result["subentries"][1]["title"] == "Zone 5"
assert result["subentries"][1]["unique_id"] == "zone_5"
assert result["subentries"][1]["data"][CONF_TYPE] == BinarySensorDeviceClass.DOOR
assert result["subentries"][1]["data"][CONF_ZONE_NAME] == "Front Door"
assert len(mock_setup_entry.mock_calls) == 1
mock_client.close.assert_awaited_once()
@pytest.mark.parametrize(
("side_effect", "expected_reason"),
[
(OSError("Connection refused"), "cannot_connect"),
(TimeoutError, "cannot_connect"),
(RuntimeError("Unexpected"), "unknown"),
],
)
async def test_import_yaml_config_errors(
hass: HomeAssistant,
mock_client: AsyncMock,
mock_setup_entry: AsyncMock,
side_effect: Exception,
expected_reason: str,
) -> None:
"""Test importing YAML configuration."""
mock_client.update.side_effect = side_effect
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data={
CONF_HOST: "192.168.1.72",
CONF_PORT: 4999,
CONF_INFER_ARMING_STATE: False,
CONF_ZONES: [
{CONF_ZONE_NAME: "Garage", CONF_ZONE_ID: 1},
{
CONF_ZONE_NAME: "Front Door",
CONF_ZONE_ID: 5,
CONF_ZONE_TYPE: BinarySensorDeviceClass.DOOR,
},
],
},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == expected_reason
async def test_import_already_configured(
hass: HomeAssistant, mock_config_entry: MockConfigEntry
) -> None:
"""Test we abort import if already configured."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data={
CONF_HOST: "192.168.1.100",
CONF_PORT: 4999,
CONF_ZONES: [],
},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
@pytest.mark.parametrize(
("side_effect", "expected_reason"),
[
(OSError("Connection refused"), "cannot_connect"),
(TimeoutError, "cannot_connect"),
(RuntimeError("Unexpected"), "unknown"),
],
)
async def test_import_connection_errors(
hass: HomeAssistant,
mock_client: AsyncMock,
side_effect: Exception,
expected_reason: str,
) -> None:
"""Test import aborts on connection errors."""
mock_client.update.side_effect = side_effect
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data={
CONF_HOST: "192.168.1.72",
CONF_PORT: 4999,
CONF_ZONES: [],
},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == expected_reason
mock_client.close.assert_awaited_once()
async def test_zone_subentry_flow(hass: HomeAssistant) -> None:
"""Test adding a zone through subentry flow."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "192.168.1.100",
CONF_PORT: 1992,
},
)
entry.add_to_hass(hass)
result = await hass.config_entries.subentries.async_init(
(entry.entry_id, SUBENTRY_TYPE_ZONE),
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"],
{
CONF_ZONE_NUMBER: 1,
CONF_TYPE: BinarySensorDeviceClass.DOOR,
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "Zone 1"
assert result["data"][CONF_ZONE_NUMBER] == 1
assert result["data"][CONF_TYPE] == BinarySensorDeviceClass.DOOR
async def test_zone_subentry_already_configured(hass: HomeAssistant) -> None:
"""Test adding a zone that already exists."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "192.168.1.100",
CONF_PORT: 1992,
},
)
entry.add_to_hass(hass)
entry.subentries = {
"zone_1_id": ConfigSubentry(
subentry_type=SUBENTRY_TYPE_ZONE,
subentry_id="zone_1_id",
unique_id="zone_1",
title="Zone 1",
data=MappingProxyType(
{
CONF_ZONE_NUMBER: 1,
CONF_TYPE: BinarySensorDeviceClass.MOTION,
}
),
)
}
result = await hass.config_entries.subentries.async_init(
(entry.entry_id, SUBENTRY_TYPE_ZONE),
context={"source": SOURCE_USER},
)
result = await hass.config_entries.subentries.async_configure(
result["flow_id"],
{
CONF_ZONE_NUMBER: 1,
CONF_TYPE: BinarySensorDeviceClass.DOOR,
},
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {CONF_ZONE_NUMBER: "already_configured"}
async def test_zone_subentry_reconfigure(hass: HomeAssistant) -> None:
"""Test reconfiguring an existing zone."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "192.168.1.100",
CONF_PORT: 1992,
},
)
entry.add_to_hass(hass)
zone_subentry = ConfigSubentry(
subentry_type=SUBENTRY_TYPE_ZONE,
subentry_id="zone_1_id",
unique_id="zone_1",
title="Zone 1",
data=MappingProxyType(
{
CONF_ZONE_NUMBER: 1,
CONF_TYPE: BinarySensorDeviceClass.MOTION,
}
),
)
entry.subentries = {"zone_1_id": zone_subentry}
result = await entry.start_subentry_reconfigure_flow(hass, "zone_1_id")
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reconfigure"
assert result["description_placeholders"][CONF_ZONE_NUMBER] == "1"
result = await hass.config_entries.subentries.async_configure(
result["flow_id"],
{
CONF_TYPE: BinarySensorDeviceClass.DOOR,
},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reconfigure_successful"
async def test_options_flow(hass: HomeAssistant) -> None:
"""Test options flow to configure alarm panel settings."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "192.168.1.100",
CONF_PORT: 1992,
},
)
entry.add_to_hass(hass)
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"],
{
CONF_SHOW_HOME_MODE: False,
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert entry.options[CONF_SHOW_HOME_MODE] is False
async def test_options_flow_enable_home_mode(hass: HomeAssistant) -> None:
"""Test options flow to enable home mode."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "192.168.1.100",
CONF_PORT: 1992,
},
options={CONF_SHOW_HOME_MODE: False},
)
entry.add_to_hass(hass)
result = await hass.config_entries.options.async_init(entry.entry_id)
result = await hass.config_entries.options.async_configure(
result["flow_id"],
{
CONF_SHOW_HOME_MODE: True,
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert entry.options[CONF_SHOW_HOME_MODE] is True

591
tests/components/ness_alarm/test_init.py
View File

@@ -1,26 +1,33 @@
"""Tests for the ness_alarm component."""
from unittest.mock import MagicMock, patch
from types import MappingProxyType
from unittest.mock import AsyncMock, patch
from nessclient import ArmingMode, ArmingState
import pytest
from homeassistant.components import alarm_control_panel
from homeassistant.components.alarm_control_panel import AlarmControlPanelState
from homeassistant.components.ness_alarm import (
ATTR_CODE,
from homeassistant.components.alarm_control_panel import (
AlarmControlPanelEntityFeature,
AlarmControlPanelState,
)
from homeassistant.components.binary_sensor import BinarySensorDeviceClass
from homeassistant.components.ness_alarm.const import (
ATTR_OUTPUT_ID,
CONF_DEVICE_PORT,
CONF_ZONE_ID,
CONF_ZONE_NAME,
CONF_ZONES,
CONF_SHOW_HOME_MODE,
CONF_ZONE_NUMBER,
DOMAIN,
SERVICE_AUX,
SERVICE_PANIC,
SUBENTRY_TYPE_ZONE,
)
from homeassistant.config_entries import ConfigEntryState, ConfigSubentry
from homeassistant.const import (
ATTR_CODE,
ATTR_ENTITY_ID,
ATTR_STATE,
CONF_HOST,
CONF_PORT,
CONF_TYPE,
SERVICE_ALARM_ARM_AWAY,
SERVICE_ALARM_ARM_HOME,
SERVICE_ALARM_DISARM,
@@ -28,70 +35,234 @@ from homeassistant.const import (
STATE_UNKNOWN,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import issue_registry as ir
from homeassistant.setup import async_setup_component
VALID_CONFIG = {
DOMAIN: {
CONF_HOST: "alarm.local",
CONF_DEVICE_PORT: 1234,
CONF_ZONES: [
{CONF_ZONE_NAME: "Zone 1", CONF_ZONE_ID: 1},
{CONF_ZONE_NAME: "Zone 2", CONF_ZONE_ID: 2},
],
}
}
from tests.common import MockConfigEntry
async def test_setup_platform(hass: HomeAssistant, mock_nessclient) -> None:
"""Test platform setup."""
await async_setup_component(hass, DOMAIN, VALID_CONFIG)
assert hass.services.has_service(DOMAIN, "panic")
assert hass.services.has_service(DOMAIN, "aux")
async def test_config_entry_setup(hass: HomeAssistant, mock_nessclient) -> None:
"""Test config entry setup."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "192.168.1.100",
CONF_PORT: 1992,
},
)
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert hass.states.get("alarm_control_panel.alarm_panel") is not None
assert hass.states.get("binary_sensor.zone_1") is not None
assert hass.states.get("binary_sensor.zone_2") is not None
# Services should be registered
assert hass.services.has_service(DOMAIN, SERVICE_PANIC)
assert hass.services.has_service(DOMAIN, SERVICE_AUX)
# Alarm panel should be created
assert hass.states.get("alarm_control_panel.alarm_panel")
# Client keepalive and update should be called after startup
assert mock_nessclient.keepalive.call_count == 1
assert mock_nessclient.update.call_count == 1
# update is called once during setup (connection test) and once after startup
assert mock_nessclient.update.call_count == 2
async def test_panic_service(hass: HomeAssistant, mock_nessclient) -> None:
"""Test calling panic service."""
await async_setup_component(hass, DOMAIN, VALID_CONFIG)
async def test_config_entry_unload(hass: HomeAssistant, mock_nessclient) -> None:
"""Test config entry unload."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "192.168.1.100",
CONF_PORT: 1992,
},
)
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()
# Client should be closed
mock_nessclient.close.assert_called_once()
async def test_config_entry_not_ready(hass: HomeAssistant, mock_nessclient) -> None:
"""Test config entry raises ConfigEntryNotReady on connection failure."""
mock_nessclient.update.side_effect = OSError("Connection refused")
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "192.168.1.100",
CONF_PORT: 1992,
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.SETUP_RETRY
mock_nessclient.close.assert_called_once()
async def test_config_entry_with_zones(hass: HomeAssistant, mock_nessclient) -> None:
"""Test config entry setup with zones as subentries."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "192.168.1.100",
CONF_PORT: 1992,
},
)
entry.add_to_hass(hass)
# Add zone subentries
entry.subentries = {
"zone_1_id": ConfigSubentry(
subentry_type=SUBENTRY_TYPE_ZONE,
subentry_id="zone_1_id",
unique_id="zone_1",
title="Zone 1",
data={
CONF_ZONE_NUMBER: 1,
CONF_TYPE: BinarySensorDeviceClass.MOTION,
},
),
"zone_2_id": ConfigSubentry(
subentry_type=SUBENTRY_TYPE_ZONE,
subentry_id="zone_2_id",
unique_id="zone_2",
title="Zone 2",
data={
CONF_ZONE_NUMBER: 2,
CONF_TYPE: BinarySensorDeviceClass.DOOR,
},
),
}
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
# Binary sensors should be created for each zone
assert hass.states.get("binary_sensor.zone_1")
assert hass.states.get("binary_sensor.zone_2")
async def test_config_entry_reload_on_subentry_add(
hass: HomeAssistant, mock_nessclient
) -> None:
"""Test config entry with subentries."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "192.168.1.100",
CONF_PORT: 1992,
},
)
entry.add_to_hass(hass)
# Add a zone subentry
entry.subentries = {
"zone_1_id": ConfigSubentry(
subentry_type=SUBENTRY_TYPE_ZONE,
subentry_id="zone_1_id",
unique_id="zone_1",
title="Zone 1",
data={
CONF_ZONE_NUMBER: 1,
CONF_TYPE: BinarySensorDeviceClass.MOTION,
},
),
}
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
# Zone entity should be created
assert hass.states.get("binary_sensor.zone_1")
async def test_panic_service_with_config_entry(
hass: HomeAssistant, mock_nessclient
) -> None:
"""Test calling panic service with config entry."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "192.168.1.100",
CONF_PORT: 1992,
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
await hass.services.async_call(
DOMAIN, SERVICE_PANIC, blocking=True, service_data={ATTR_CODE: "1234"}
)
mock_nessclient.panic.assert_awaited_once_with("1234")
async def test_aux_service(hass: HomeAssistant, mock_nessclient) -> None:
"""Test calling aux service."""
await async_setup_component(hass, DOMAIN, VALID_CONFIG)
async def test_aux_service_with_config_entry(
hass: HomeAssistant, mock_nessclient
) -> None:
"""Test calling aux service with config entry."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "192.168.1.100",
CONF_PORT: 1992,
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
await hass.services.async_call(
DOMAIN, SERVICE_AUX, blocking=True, service_data={ATTR_OUTPUT_ID: 1}
)
mock_nessclient.aux.assert_awaited_once_with(1, True)
async def test_dispatch_state_change(hass: HomeAssistant, mock_nessclient) -> None:
"""Test calling aux service."""
await async_setup_component(hass, DOMAIN, VALID_CONFIG)
await hass.async_block_till_done()
on_state_change = mock_nessclient.on_state_change.call_args[0][0]
on_state_change(ArmingState.ARMING, None)
await hass.async_block_till_done()
assert hass.states.is_state(
"alarm_control_panel.alarm_panel", AlarmControlPanelState.ARMING
async def test_aux_service_with_state_false(
hass: HomeAssistant, mock_nessclient
) -> None:
"""Test calling aux service with state=False."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "192.168.1.100",
CONF_PORT: 1992,
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
await hass.services.async_call(
DOMAIN,
SERVICE_AUX,
blocking=True,
service_data={ATTR_OUTPUT_ID: 2, ATTR_STATE: False},
)
mock_nessclient.aux.assert_awaited_once_with(2, False)
async def test_alarm_disarm(hass: HomeAssistant, mock_nessclient) -> None:
"""Test disarm."""
await async_setup_component(hass, DOMAIN, VALID_CONFIG)
async def test_alarm_panel_disarm(hass: HomeAssistant, mock_nessclient) -> None:
"""Test alarm panel disarm."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "192.168.1.100",
CONF_PORT: 1992,
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
await hass.services.async_call(
@@ -106,9 +277,17 @@ async def test_alarm_disarm(hass: HomeAssistant, mock_nessclient) -> None:
mock_nessclient.disarm.assert_called_once_with("1234")
async def test_alarm_arm_away(hass: HomeAssistant, mock_nessclient) -> None:
"""Test disarm."""
await async_setup_component(hass, DOMAIN, VALID_CONFIG)
async def test_alarm_panel_arm_away(hass: HomeAssistant, mock_nessclient) -> None:
"""Test alarm panel arm away."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "192.168.1.100",
CONF_PORT: 1992,
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
await hass.services.async_call(
@@ -123,9 +302,17 @@ async def test_alarm_arm_away(hass: HomeAssistant, mock_nessclient) -> None:
mock_nessclient.arm_away.assert_called_once_with("1234")
async def test_alarm_arm_home(hass: HomeAssistant, mock_nessclient) -> None:
"""Test disarm."""
await async_setup_component(hass, DOMAIN, VALID_CONFIG)
async def test_alarm_panel_arm_home(hass: HomeAssistant, mock_nessclient) -> None:
"""Test alarm panel arm home."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "192.168.1.100",
CONF_PORT: 1992,
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
await hass.services.async_call(
@@ -140,9 +327,17 @@ async def test_alarm_arm_home(hass: HomeAssistant, mock_nessclient) -> None:
mock_nessclient.arm_home.assert_called_once_with("1234")
async def test_alarm_trigger(hass: HomeAssistant, mock_nessclient) -> None:
"""Test disarm."""
await async_setup_component(hass, DOMAIN, VALID_CONFIG)
async def test_alarm_panel_trigger(hass: HomeAssistant, mock_nessclient) -> None:
"""Test alarm panel trigger."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "192.168.1.100",
CONF_PORT: 1992,
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
await hass.services.async_call(
@@ -157,21 +352,79 @@ async def test_alarm_trigger(hass: HomeAssistant, mock_nessclient) -> None:
mock_nessclient.panic.assert_called_once_with("1234")
async def test_dispatch_zone_change(hass: HomeAssistant, mock_nessclient) -> None:
"""Test zone change events dispatch a signal to subscribers."""
await async_setup_component(hass, DOMAIN, VALID_CONFIG)
async def test_zone_state_change(hass: HomeAssistant, mock_nessclient) -> None:
"""Test zone state change events."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "192.168.1.100",
CONF_PORT: 1992,
},
)
entry.add_to_hass(hass)
# Add zone subentries
entry.subentries = {
"zone_1_id": ConfigSubentry(
subentry_type=SUBENTRY_TYPE_ZONE,
subentry_id="zone_1_id",
unique_id="zone_1",
title="Zone 1",
data={
CONF_ZONE_NUMBER: 1,
CONF_TYPE: BinarySensorDeviceClass.MOTION,
},
),
"zone_2_id": ConfigSubentry(
subentry_type=SUBENTRY_TYPE_ZONE,
subentry_id="zone_2_id",
unique_id="zone_2",
title="Zone 2",
data={
CONF_ZONE_NUMBER: 2,
CONF_TYPE: BinarySensorDeviceClass.DOOR,
},
),
}
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
# Get the zone change callback
on_zone_change = mock_nessclient.on_zone_change.call_args[0][0]
on_zone_change(1, True)
# Trigger zone 1
on_zone_change(1, True)
await hass.async_block_till_done()
assert hass.states.is_state("binary_sensor.zone_1", "on")
assert hass.states.is_state("binary_sensor.zone_2", "off")
# Trigger zone 2
on_zone_change(2, True)
await hass.async_block_till_done()
assert hass.states.is_state("binary_sensor.zone_2", "on")
# Clear zone 1
on_zone_change(1, False)
await hass.async_block_till_done()
assert hass.states.is_state("binary_sensor.zone_1", "off")
async def test_arming_state_change(hass: HomeAssistant, mock_nessclient) -> None:
"""Test arming state change handing."""
async def test_arming_state_changes(hass: HomeAssistant, mock_nessclient) -> None:
"""Test all arming state changes."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "192.168.1.100",
CONF_PORT: 1992,
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
# Get the state change callback
on_state_change = mock_nessclient.on_state_change.call_args[0][0]
states = [
(ArmingState.UNKNOWN, None, STATE_UNKNOWN),
(ArmingState.DISARMED, None, AlarmControlPanelState.DISARMED),
@@ -193,67 +446,185 @@ async def test_arming_state_change(hass: HomeAssistant, mock_nessclient) -> None
ArmingMode.ARMED_NIGHT,
AlarmControlPanelState.ARMED_NIGHT,
),
(
ArmingState.ARMED,
ArmingMode.ARMED_VACATION,
AlarmControlPanelState.ARMED_VACATION,
),
(
ArmingState.ARMED,
ArmingMode.ARMED_DAY,
AlarmControlPanelState.ARMED_AWAY,
),
(
ArmingState.ARMED,
ArmingMode.ARMED_HIGHEST,
AlarmControlPanelState.ARMED_AWAY,
),
(ArmingState.ENTRY_DELAY, None, AlarmControlPanelState.PENDING),
(ArmingState.TRIGGERED, None, AlarmControlPanelState.TRIGGERED),
]
await async_setup_component(hass, DOMAIN, VALID_CONFIG)
await hass.async_block_till_done()
assert hass.states.is_state("alarm_control_panel.alarm_panel", STATE_UNKNOWN)
on_state_change = mock_nessclient.on_state_change.call_args[0][0]
for arming_state, arming_mode, expected_state in states:
on_state_change(arming_state, arming_mode)
await hass.async_block_till_done()
assert hass.states.is_state("alarm_control_panel.alarm_panel", expected_state)
class MockClient:
"""Mock nessclient.Client stub."""
async def test_arming_state_unknown_mode(hass: HomeAssistant, mock_nessclient) -> None:
"""Test arming state with unknown arming mode (for coverage)."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "192.168.1.100",
CONF_PORT: 1992,
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
async def panic(self, code):
"""Handle panic."""
# Get the state change callback
on_state_change = mock_nessclient.on_state_change.call_args[0][0]
async def disarm(self, code):
"""Handle disarm."""
async def arm_away(self, code):
"""Handle arm_away."""
async def arm_home(self, code):
"""Handle arm_home."""
async def aux(self, output_id, state):
"""Handle auxiliary control."""
async def keepalive(self):
"""Handle keepalive."""
async def update(self):
"""Handle update."""
def on_zone_change(self):
"""Handle on_zone_change."""
def on_state_change(self):
"""Handle on_state_change."""
async def close(self):
"""Handle close."""
# Test with unhandled arming state (for coverage of warning log)
on_state_change(999, None) # Invalid state
await hass.async_block_till_done()
@pytest.fixture
def mock_nessclient():
"""Mock the nessclient Client constructor.
async def test_homeassistant_stop_event(hass: HomeAssistant, mock_nessclient) -> None:
"""Test client is closed on homeassistant_stop event."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "192.168.1.100",
CONF_PORT: 1992,
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
Replaces nessclient.Client with a Mock which always returns the same
MagicMock() instance.
"""
_mock_instance = MagicMock(MockClient())
_mock_factory = MagicMock()
_mock_factory.return_value = _mock_instance
# Fire the homeassistant_stop event
hass.bus.async_fire("homeassistant_stop")
await hass.async_block_till_done()
# Client should be closed
mock_nessclient.close.assert_called()
async def test_entry_reload_on_update(hass: HomeAssistant, mock_nessclient) -> None:
"""Test config entry reload when update listener is triggered."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "192.168.1.100",
CONF_PORT: 1992,
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
# Add a zone subentry which should trigger the update listener and reload
zone_subentry = ConfigSubentry(
subentry_type=SUBENTRY_TYPE_ZONE,
subentry_id="zone_1_id",
unique_id="zone_1",
title="Zone 1",
data=MappingProxyType(
{
CONF_ZONE_NUMBER: 1,
CONF_TYPE: BinarySensorDeviceClass.MOTION,
}
),
)
hass.config_entries.async_add_subentry(entry, zone_subentry)
await hass.async_block_till_done()
# Entry should have the new zone subentry
assert len(entry.subentries) == 1
async def test_alarm_panel_home_mode_disabled(
hass: HomeAssistant, mock_nessclient
) -> None:
"""Test alarm panel with home mode disabled via options."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "192.168.1.100",
CONF_PORT: 1992,
},
options={CONF_SHOW_HOME_MODE: False},
)
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
state = hass.states.get("alarm_control_panel.alarm_panel")
assert state is not None
# ARM_HOME should not be in supported features
supported = state.attributes["supported_features"]
assert not supported & AlarmControlPanelEntityFeature.ARM_HOME
assert supported & AlarmControlPanelEntityFeature.ARM_AWAY
assert supported & AlarmControlPanelEntityFeature.TRIGGER
async def test_alarm_panel_home_mode_enabled_by_default(
hass: HomeAssistant, mock_nessclient
) -> None:
"""Test alarm panel has home mode enabled by default."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "192.168.1.100",
CONF_PORT: 1992,
},
)
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
state = hass.states.get("alarm_control_panel.alarm_panel")
assert state is not None
# ARM_HOME should be in supported features by default
supported = state.attributes["supported_features"]
assert supported & AlarmControlPanelEntityFeature.ARM_HOME
assert supported & AlarmControlPanelEntityFeature.ARM_AWAY
assert supported & AlarmControlPanelEntityFeature.TRIGGER
async def test_yaml_import_triggers_flow(
hass: HomeAssistant, mock_setup_entry: AsyncMock, issue_registry: ir.IssueRegistry
) -> None:
"""Test that YAML configuration triggers import flow."""
with patch(
"homeassistant.components.ness_alarm.Client", new=_mock_factory, create=True
"homeassistant.components.ness_alarm.config_flow.Client",
return_value=AsyncMock(),
):
yield _mock_instance
config = {
DOMAIN: {
CONF_HOST: "192.168.1.100",
CONF_PORT: 1992,
}
}
assert await async_setup_component(hass, DOMAIN, config)
await hass.async_block_till_done()
# Check that a config entry was created from the import
entries = hass.config_entries.async_entries(DOMAIN)
assert len(entries) == 1
assert entries[0].data[CONF_HOST] == "192.168.1.100"
assert entries[0].data[CONF_PORT] == 1992
# Check that a deprecation repair issue was created
issue = issue_registry.async_get_issue(
"homeassistant", f"deprecated_yaml_{DOMAIN}"
)
assert issue is not None
assert issue.severity == "warning"