From 58569a58a93c644e7e391c030d9170df9d58e5a1 Mon Sep 17 00:00:00 2001 From: Jan Bouwhuis Date: Tue, 19 Oct 2021 12:07:38 +0200 Subject: [PATCH] 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 Co-authored-by: Erik Montnemery --- .../components/mqtt/alarm_control_panel.py | 13 +- .../mqtt/test_alarm_control_panel.py | 117 ++++++++++++++++++ 2 files changed, 128 insertions(+), 2 deletions(-) diff --git a/homeassistant/components/mqtt/alarm_control_panel.py b/homeassistant/components/mqtt/alarm_control_panel.py index 0966497c024..3d632baf0f7 100644 --- a/homeassistant/components/mqtt/alarm_control_panel.py +++ b/homeassistant/components/mqtt/alarm_control_panel.py @@ -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 diff --git a/tests/components/mqtt/test_alarm_control_panel.py b/tests/components/mqtt/test_alarm_control_panel.py index e01b246b8af..7b4b8d22168 100644 --- a/tests/components/mqtt/test_alarm_control_panel.py +++ b/tests/components/mqtt/test_alarm_control_panel.py @@ -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)