Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

MQTT Alarm control panel - Enable remote code validation (#57764) 

* Enable remote code validation

* Update homeassistant/components/mqtt/alarm_control_panel.py

Co-authored-by: Erik Montnemery <erik@montnemery.com>

Co-authored-by: Erik Montnemery <erik@montnemery.com>
pull/58031/head
Jan Bouwhuis 2021-10-19 12:07:38 +02:00 committed by GitHub
parent 4f2d313a4a
commit 58569a58a9
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 128 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
homeassistant/components/mqtt/alarm_control_panel.py
View File

@ -67,6 +67,10 @@ DEFAULT_ARM_HOME = "ARM_HOME"
DEFAULT_ARM_CUSTOM_BYPASS = "ARM_CUSTOM_BYPASS"
DEFAULT_DISARM = "DISARM"
DEFAULT_NAME = "MQTT Alarm"
REMOTE_CODE = "REMOTE_CODE"
REMOTE_CODE_TEXT = "REMOTE_CODE_TEXT"
PLATFORM_SCHEMA = mqtt.MQTT_BASE_PLATFORM_SCHEMA.extend(
{
vol.Optional(CONF_CODE): cv.string,
@ -204,7 +208,7 @@ class MqttAlarm(MqttEntity, alarm.AlarmControlPanelEntity):
code = self._config.get(CONF_CODE)
if code is None:
return None
if isinstance(code, str) and re.search("^\\d+$", code):
if code == REMOTE_CODE or (isinstance(code, str) and re.search("^\\d+$", code)):
return alarm.FORMAT_NUMBER
return alarm.FORMAT_TEXT
@ -296,7 +300,12 @@ class MqttAlarm(MqttEntity, alarm.AlarmControlPanelEntity):
def _validate_code(self, code, state):
"""Validate given code."""
conf_code = self._config.get(CONF_CODE)
check = conf_code is None or code == conf_code
check = (
conf_code is None
or code == conf_code
or (conf_code == REMOTE_CODE and code)
or (conf_code == REMOTE_CODE_TEXT and code)
)
if not check:
_LOGGER.warning("Wrong code entered for %s", state)
return check

117
tests/components/mqtt/test_alarm_control_panel.py
View File

@ -83,6 +83,28 @@ DEFAULT_CONFIG_CODE = {
}
}
DEFAULT_CONFIG_REMOTE_CODE = {
alarm_control_panel.DOMAIN: {
"platform": "mqtt",
"name": "test",
"state_topic": "alarm/state",
"command_topic": "alarm/command",
"code": "REMOTE_CODE",
"code_arm_required": True,
}
}
DEFAULT_CONFIG_REMOTE_CODE_TEXT = {
alarm_control_panel.DOMAIN: {
"platform": "mqtt",
"name": "test",
"state_topic": "alarm/state",
"command_topic": "alarm/command",
"code": "REMOTE_CODE_TEXT",
"code_arm_required": True,
}
}
async def test_fail_setup_without_state_topic(hass, mqtt_mock):
"""Test for failing with no state topic."""
@ -240,6 +262,86 @@ async def test_publish_mqtt_with_code(hass, mqtt_mock, service, payload):
mqtt_mock.async_publish.assert_called_once_with("alarm/command", payload, 0, False)
@pytest.mark.parametrize(
"service,payload",
[
(SERVICE_ALARM_ARM_HOME, "ARM_HOME"),
(SERVICE_ALARM_ARM_AWAY, "ARM_AWAY"),
(SERVICE_ALARM_ARM_NIGHT, "ARM_NIGHT"),
(SERVICE_ALARM_ARM_VACATION, "ARM_VACATION"),
(SERVICE_ALARM_ARM_CUSTOM_BYPASS, "ARM_CUSTOM_BYPASS"),
(SERVICE_ALARM_DISARM, "DISARM"),
],
)
async def test_publish_mqtt_with_remote_code(hass, mqtt_mock, service, payload):
"""Test publishing of MQTT messages when remode code is configured."""
assert await async_setup_component(
hass,
alarm_control_panel.DOMAIN,
DEFAULT_CONFIG_REMOTE_CODE,
)
await hass.async_block_till_done()
call_count = mqtt_mock.async_publish.call_count
# No code provided, should not publish
await hass.services.async_call(
alarm_control_panel.DOMAIN,
service,
{ATTR_ENTITY_ID: "alarm_control_panel.test"},
blocking=True,
)
assert mqtt_mock.async_publish.call_count == call_count
# Any code numbered provided, should publish
await hass.services.async_call(
alarm_control_panel.DOMAIN,
service,
{ATTR_ENTITY_ID: "alarm_control_panel.test", ATTR_CODE: "1234"},
blocking=True,
)
mqtt_mock.async_publish.assert_called_once_with("alarm/command", payload, 0, False)
@pytest.mark.parametrize(
"service,payload",
[
(SERVICE_ALARM_ARM_HOME, "ARM_HOME"),
(SERVICE_ALARM_ARM_AWAY, "ARM_AWAY"),
(SERVICE_ALARM_ARM_NIGHT, "ARM_NIGHT"),
(SERVICE_ALARM_ARM_VACATION, "ARM_VACATION"),
(SERVICE_ALARM_ARM_CUSTOM_BYPASS, "ARM_CUSTOM_BYPASS"),
(SERVICE_ALARM_DISARM, "DISARM"),
],
)
async def test_publish_mqtt_with_remote_code_text(hass, mqtt_mock, service, payload):
"""Test publishing of MQTT messages when remote text code is configured."""
assert await async_setup_component(
hass,
alarm_control_panel.DOMAIN,
DEFAULT_CONFIG_REMOTE_CODE_TEXT,
)
await hass.async_block_till_done()
call_count = mqtt_mock.async_publish.call_count
# No code provided, should not publish
await hass.services.async_call(
alarm_control_panel.DOMAIN,
service,
{ATTR_ENTITY_ID: "alarm_control_panel.test"},
blocking=True,
)
assert mqtt_mock.async_publish.call_count == call_count
# Any code numbered provided, should publish
await hass.services.async_call(
alarm_control_panel.DOMAIN,
service,
{ATTR_ENTITY_ID: "alarm_control_panel.test", ATTR_CODE: "any_code"},
blocking=True,
)
mqtt_mock.async_publish.assert_called_once_with("alarm/command", payload, 0, False)
@pytest.mark.parametrize(
"service,payload,disable_code",
[
@ -367,6 +469,21 @@ async def test_attributes_code_number(hass, mqtt_mock):
)
async def test_attributes_remote_code_number(hass, mqtt_mock):
"""Test attributes which are not supported by the vacuum."""
config = copy.deepcopy(DEFAULT_CONFIG_REMOTE_CODE)
config[alarm_control_panel.DOMAIN]["code"] = "REMOTE_CODE"
assert await async_setup_component(hass, alarm_control_panel.DOMAIN, config)
await hass.async_block_till_done()
state = hass.states.get("alarm_control_panel.test")
assert (
state.attributes.get(alarm_control_panel.ATTR_CODE_FORMAT)
== alarm_control_panel.FORMAT_NUMBER
)
async def test_attributes_code_text(hass, mqtt_mock):
"""Test attributes which are not supported by the vacuum."""
config = copy.deepcopy(DEFAULT_CONFIG)