Add TotalConnect option to require alarm code (#122270)

* add config option

* use code_required option in alarm

* test code_required options

* only use code for disarm

* change tests to disarm with code

* remove unneeded code variable

* Update homeassistant/components/totalconnect/alarm_control_panel.py

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>

* use ServiceValidationError

* translate ServiceValidationError

* complete typing

* Update tests/components/totalconnect/test_alarm_control_panel.py

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>

* use ServiceValidationError in test

* grab usercode from correct spot

* use client code instead of unfilled location code

* Revert "remove unneeded code variable"

This reverts commit 220de0e698e5779fcd7c45bee999a60ad186ab7f.

* remove unneeded code variable

* improve usercode checking

* use freezer

* fix usercode test data

* Update homeassistant/components/totalconnect/strings.json

Co-authored-by: G Johansson <goran.johansson@shiftit.se>

* Update homeassistant/components/totalconnect/strings.json

Co-authored-by: G Johansson <goran.johansson@shiftit.se>

* update test with new message

---------

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
Co-authored-by: G Johansson <goran.johansson@shiftit.se>
pull/125771/head
Austin Mroczek 2024-09-11 09:23:19 -07:00 committed by GitHub
parent 393181df20
commit 0c1a605693
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 115 additions and 29 deletions

View File

@ -9,6 +9,7 @@ from total_connect_client.location import TotalConnectLocation
from homeassistant.components.alarm_control_panel import (
AlarmControlPanelEntity,
AlarmControlPanelEntityFeature,
CodeFormat,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
@ -22,11 +23,11 @@ from homeassistant.const import (
STATE_ALARM_TRIGGERED,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.helpers import entity_platform
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DOMAIN
from .const import CODE_REQUIRED, DOMAIN
from .coordinator import TotalConnectDataUpdateCoordinator
from .entity import TotalConnectLocationEntity
@ -39,13 +40,10 @@ async def async_setup_entry(
) -> None:
"""Set up TotalConnect alarm panels based on a config entry."""
coordinator: TotalConnectDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
code_required = entry.options.get(CODE_REQUIRED, False)
async_add_entities(
TotalConnectAlarm(
coordinator,
location,
partition_id,
)
TotalConnectAlarm(coordinator, location, partition_id, code_required)
for location in coordinator.client.locations.values()
for partition_id in location.partitions
)
@ -74,13 +72,13 @@ class TotalConnectAlarm(TotalConnectLocationEntity, AlarmControlPanelEntity):
| AlarmControlPanelEntityFeature.ARM_AWAY
| AlarmControlPanelEntityFeature.ARM_NIGHT
)
_attr_code_arm_required = False
def __init__(
self,
coordinator: TotalConnectDataUpdateCoordinator,
location: TotalConnectLocation,
partition_id: int,
require_code: bool,
) -> None:
"""Initialize the TotalConnect status."""
super().__init__(coordinator, location)
@ -100,6 +98,10 @@ class TotalConnectAlarm(TotalConnectLocationEntity, AlarmControlPanelEntity):
self._attr_translation_placeholders = {"partition_id": str(partition_id)}
self._attr_unique_id = f"{location.location_id}_{partition_id}"
self._attr_code_arm_required = require_code
if require_code:
self._attr_code_format = CodeFormat.NUMBER
@property
def state(self) -> str | None:
"""Return the state of the device."""
@ -150,6 +152,7 @@ class TotalConnectAlarm(TotalConnectLocationEntity, AlarmControlPanelEntity):
async def async_alarm_disarm(self, code: str | None = None) -> None:
"""Send disarm command."""
self._check_usercode(code)
try:
await self.hass.async_add_executor_job(self._disarm)
except UsercodeInvalid as error:
@ -163,12 +166,13 @@ class TotalConnectAlarm(TotalConnectLocationEntity, AlarmControlPanelEntity):
) from error
await self.coordinator.async_request_refresh()
def _disarm(self, code=None):
def _disarm(self) -> None:
"""Disarm synchronous."""
ArmingHelper(self._partition).disarm()
async def async_alarm_arm_home(self, code: str | None = None) -> None:
"""Send arm home command."""
self._check_usercode(code)
try:
await self.hass.async_add_executor_job(self._arm_home)
except UsercodeInvalid as error:
@ -182,12 +186,13 @@ class TotalConnectAlarm(TotalConnectLocationEntity, AlarmControlPanelEntity):
) from error
await self.coordinator.async_request_refresh()
def _arm_home(self):
def _arm_home(self) -> None:
"""Arm home synchronous."""
ArmingHelper(self._partition).arm_stay()
async def async_alarm_arm_away(self, code: str | None = None) -> None:
"""Send arm away command."""
self._check_usercode(code)
try:
await self.hass.async_add_executor_job(self._arm_away)
except UsercodeInvalid as error:
@ -201,12 +206,13 @@ class TotalConnectAlarm(TotalConnectLocationEntity, AlarmControlPanelEntity):
) from error
await self.coordinator.async_request_refresh()
def _arm_away(self, code=None):
def _arm_away(self) -> None:
"""Arm away synchronous."""
ArmingHelper(self._partition).arm_away()
async def async_alarm_arm_night(self, code: str | None = None) -> None:
"""Send arm night command."""
self._check_usercode(code)
try:
await self.hass.async_add_executor_job(self._arm_night)
except UsercodeInvalid as error:
@ -220,11 +226,11 @@ class TotalConnectAlarm(TotalConnectLocationEntity, AlarmControlPanelEntity):
) from error
await self.coordinator.async_request_refresh()
def _arm_night(self, code=None):
def _arm_night(self) -> None:
"""Arm night synchronous."""
ArmingHelper(self._partition).arm_stay_night()
async def async_alarm_arm_home_instant(self, code: str | None = None) -> None:
async def async_alarm_arm_home_instant(self) -> None:
"""Send arm home instant command."""
try:
await self.hass.async_add_executor_job(self._arm_home_instant)
@ -243,7 +249,7 @@ class TotalConnectAlarm(TotalConnectLocationEntity, AlarmControlPanelEntity):
"""Arm home instant synchronous."""
ArmingHelper(self._partition).arm_stay_instant()
async def async_alarm_arm_away_instant(self, code: str | None = None) -> None:
async def async_alarm_arm_away_instant(self) -> None:
"""Send arm away instant command."""
try:
await self.hass.async_add_executor_job(self._arm_away_instant)
@ -258,6 +264,16 @@ class TotalConnectAlarm(TotalConnectLocationEntity, AlarmControlPanelEntity):
) from error
await self.coordinator.async_request_refresh()
def _arm_away_instant(self, code=None):
def _arm_away_instant(self):
"""Arm away instant synchronous."""
ArmingHelper(self._partition).arm_away_instant()
def _check_usercode(self, code):
"""Check if the run-time entered code matches configured code."""
if (
self._attr_code_arm_required
and self.coordinator.client.usercodes[self._location.location_id] != code
):
raise ServiceValidationError(
translation_domain=DOMAIN, translation_key="invalid_pin"
)

View File

@ -19,7 +19,7 @@ from homeassistant.const import CONF_LOCATION, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import callback
from homeassistant.helpers.typing import VolDictType
from .const import AUTO_BYPASS, CONF_USERCODES, DOMAIN
from .const import AUTO_BYPASS, CODE_REQUIRED, CONF_USERCODES, DOMAIN
PASSWORD_DATA_SCHEMA = vol.Schema({vol.Required(CONF_PASSWORD): str})
@ -217,7 +217,11 @@ class TotalConnectOptionsFlowHandler(OptionsFlow):
vol.Required(
AUTO_BYPASS,
default=self.config_entry.options.get(AUTO_BYPASS, False),
): bool
): bool,
vol.Required(
CODE_REQUIRED,
default=self.config_entry.options.get(CODE_REQUIRED, False),
): bool,
}
),
)

View File

@ -3,6 +3,7 @@
DOMAIN = "totalconnect"
CONF_USERCODES = "usercodes"
AUTO_BYPASS = "auto_bypass_low_battery"
CODE_REQUIRED = "code_required"
# Most TotalConnect alarms will work passing '-1' as usercode
DEFAULT_USERCODE = "-1"

View File

@ -33,9 +33,9 @@
"step": {
"init": {
"title": "TotalConnect Options",
"description": "Automatically bypass zones the moment they report a low battery.",
"data": {
"auto_bypass_low_battery": "Auto bypass low battery"
"auto_bypass_low_battery": "Auto bypass low battery",
"code_required": "Require user to enter code for alarm actions"
}
}
}
@ -76,5 +76,10 @@
"name": "Bypass"
}
}
},
"exceptions": {
"invalid_pin": {
"message": "Incorrect code entered"
}
}
}

View File

@ -1,11 +1,17 @@
"""Common methods used across tests for TotalConnect."""
from typing import Any
from unittest.mock import patch
from total_connect_client import ArmingState, ResultCode, ZoneStatus, ZoneType
from homeassistant.components.totalconnect.const import CONF_USERCODES, DOMAIN
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME, Platform
from homeassistant.components.totalconnect.const import (
AUTO_BYPASS,
CODE_REQUIRED,
CONF_USERCODES,
DOMAIN,
)
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
@ -341,7 +347,7 @@ RESPONSE_ZONE_BYPASS_FAILURE = {
USERNAME = "username@me.com"
PASSWORD = "password"
USERCODES = {123456: "7890"}
USERCODES = {LOCATION_ID: "7890"}
CONFIG_DATA = {
CONF_USERNAME: USERNAME,
CONF_PASSWORD: PASSWORD,
@ -349,6 +355,9 @@ CONFIG_DATA = {
}
CONFIG_DATA_NO_USERCODES = {CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD}
OPTIONS_DATA = {AUTO_BYPASS: False, CODE_REQUIRED: False}
OPTIONS_DATA_CODE_REQUIRED = {AUTO_BYPASS: False, CODE_REQUIRED: True}
PARTITION_DETAILS_1 = {
"PartitionID": 1,
"ArmingState": ArmingState.DISARMED.value,
@ -395,10 +404,19 @@ TOTALCONNECT_REQUEST = (
)
async def setup_platform(hass: HomeAssistant, platform: Platform) -> MockConfigEntry:
async def setup_platform(
hass: HomeAssistant, platform: Any, code_required: bool = False
) -> MockConfigEntry:
"""Set up the TotalConnect platform."""
# first set up a config entry and add it to hass
mock_entry = MockConfigEntry(domain=DOMAIN, data=CONFIG_DATA)
if code_required:
mock_entry = MockConfigEntry(
domain=DOMAIN, data=CONFIG_DATA, options=OPTIONS_DATA_CODE_REQUIRED
)
else:
mock_entry = MockConfigEntry(
domain=DOMAIN, data=CONFIG_DATA, options=OPTIONS_DATA
)
mock_entry.add_to_hass(hass)
responses = [
@ -426,7 +444,7 @@ async def setup_platform(hass: HomeAssistant, platform: Platform) -> MockConfigE
async def init_integration(hass: HomeAssistant) -> MockConfigEntry:
"""Set up the TotalConnect integration."""
# first set up a config entry and add it to hass
mock_entry = MockConfigEntry(domain=DOMAIN, data=CONFIG_DATA)
mock_entry = MockConfigEntry(domain=DOMAIN, data=CONFIG_DATA, options=OPTIONS_DATA)
mock_entry.add_to_hass(hass)
responses = [

View File

@ -3,6 +3,7 @@
from datetime import timedelta
from unittest.mock import patch
from freezegun.api import FrozenDateTimeFactory
import pytest
from syrupy import SnapshotAssertion
from total_connect_client.exceptions import (
@ -36,12 +37,13 @@ from homeassistant.const import (
STATE_UNAVAILABLE,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers.entity_component import async_update_entity
from homeassistant.util import dt as dt_util
from .common import (
LOCATION_ID,
RESPONSE_ARM_FAILURE,
RESPONSE_ARM_SUCCESS,
RESPONSE_ARMED_AWAY,
@ -60,6 +62,7 @@ from .common import (
RESPONSE_UNKNOWN,
RESPONSE_USER_CODE_INVALID,
TOTALCONNECT_REQUEST,
USERCODES,
setup_platform,
)
@ -132,7 +135,7 @@ async def test_arm_home_failure(hass: HomeAssistant) -> None:
assert hass.states.get(ENTITY_ID).state == STATE_ALARM_DISARMED
assert mock_request.call_count == 2
# usercode is invalid
# config entry usercode is invalid
with pytest.raises(HomeAssistantError) as err:
await hass.services.async_call(
ALARM_DOMAIN, SERVICE_ALARM_ARM_HOME, DATA, blocking=True
@ -369,6 +372,44 @@ async def test_disarm_failure(hass: HomeAssistant) -> None:
assert mock_request.call_count == 3
async def test_disarm_code_required(
hass: HomeAssistant, freezer: FrozenDateTimeFactory
) -> None:
"""Test disarm with code."""
responses = [RESPONSE_ARMED_AWAY, RESPONSE_DISARM_SUCCESS, RESPONSE_DISARMED]
await setup_platform(hass, ALARM_DOMAIN, code_required=True)
with patch(TOTALCONNECT_REQUEST, side_effect=responses) as mock_request:
await async_update_entity(hass, ENTITY_ID)
await hass.async_block_till_done()
assert hass.states.get(ENTITY_ID).state == STATE_ALARM_ARMED_AWAY
assert mock_request.call_count == 1
# runtime user entered code is bad
DATA_WITH_CODE = DATA.copy()
DATA_WITH_CODE["code"] = "666"
with pytest.raises(ServiceValidationError, match="Incorrect code entered"):
await hass.services.async_call(
ALARM_DOMAIN, SERVICE_ALARM_DISARM, DATA_WITH_CODE, blocking=True
)
assert hass.states.get(ENTITY_ID).state == STATE_ALARM_ARMED_AWAY
# code check means the call to total_connect never happens
assert mock_request.call_count == 1
# runtime user entered code that is in config
DATA_WITH_CODE["code"] = USERCODES[LOCATION_ID]
await hass.services.async_call(
ALARM_DOMAIN, SERVICE_ALARM_DISARM, DATA_WITH_CODE, blocking=True
)
await hass.async_block_till_done()
assert mock_request.call_count == 2
freezer.tick(DELAY)
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert mock_request.call_count == 3
assert hass.states.get(ENTITY_ID).state == STATE_ALARM_DISARMED
async def test_arm_night_success(hass: HomeAssistant) -> None:
"""Test arm night method success."""
responses = [RESPONSE_DISARMED, RESPONSE_ARM_SUCCESS, RESPONSE_ARMED_NIGHT]

View File

@ -6,6 +6,7 @@ from total_connect_client.exceptions import AuthenticationError
from homeassistant.components.totalconnect.const import (
AUTO_BYPASS,
CODE_REQUIRED,
CONF_USERCODES,
DOMAIN,
)
@ -238,11 +239,11 @@ async def test_options_flow(hass: HomeAssistant) -> None:
assert result["step_id"] == "init"
result = await hass.config_entries.options.async_configure(
result["flow_id"], user_input={AUTO_BYPASS: True}
result["flow_id"], user_input={AUTO_BYPASS: True, CODE_REQUIRED: False}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert config_entry.options == {AUTO_BYPASS: True}
assert config_entry.options == {AUTO_BYPASS: True, CODE_REQUIRED: False}
await hass.async_block_till_done()
assert await hass.config_entries.async_unload(config_entry.entry_id)