Add MQTT Alarm Control Panel custom bypass state (#32541)

* MQTT Alarm Control Panel to have all available states

* MQTT Alarm Control Panel to have all available states

* test_arm_custom_bypass_* tests added

* MQTT payload_arm_custom_bypass abbreviation
This commit is contained in:
Jevgeni Kiski 2020-04-06 12:45:37 +03:00 committed by GitHub
parent ffa111deb9
commit a1aebe904e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 110 additions and 1 deletions

View File

@ -85,6 +85,7 @@ ABBREVIATIONS = {
"pl_arm_away": "payload_arm_away",
"pl_arm_home": "payload_arm_home",
"pl_arm_nite": "payload_arm_night",
"pl_arm_custom_b": "payload_arm_custom_bypass",
"pl_avail": "payload_available",
"pl_cln_sp": "payload_clean_spot",
"pl_cls": "payload_close",

View File

@ -8,6 +8,7 @@ from homeassistant.components import mqtt
import homeassistant.components.alarm_control_panel as alarm
from homeassistant.components.alarm_control_panel.const import (
SUPPORT_ALARM_ARM_AWAY,
SUPPORT_ALARM_ARM_CUSTOM_BYPASS,
SUPPORT_ALARM_ARM_HOME,
SUPPORT_ALARM_ARM_NIGHT,
)
@ -17,9 +18,12 @@ from homeassistant.const import (
CONF_NAME,
CONF_VALUE_TEMPLATE,
STATE_ALARM_ARMED_AWAY,
STATE_ALARM_ARMED_CUSTOM_BYPASS,
STATE_ALARM_ARMED_HOME,
STATE_ALARM_ARMED_NIGHT,
STATE_ALARM_ARMING,
STATE_ALARM_DISARMED,
STATE_ALARM_DISARMING,
STATE_ALARM_PENDING,
STATE_ALARM_TRIGGERED,
)
@ -52,12 +56,14 @@ CONF_PAYLOAD_DISARM = "payload_disarm"
CONF_PAYLOAD_ARM_HOME = "payload_arm_home"
CONF_PAYLOAD_ARM_AWAY = "payload_arm_away"
CONF_PAYLOAD_ARM_NIGHT = "payload_arm_night"
CONF_PAYLOAD_ARM_CUSTOM_BYPASS = "payload_arm_custom_bypass"
CONF_COMMAND_TEMPLATE = "command_template"
DEFAULT_COMMAND_TEMPLATE = "{{action}}"
DEFAULT_ARM_NIGHT = "ARM_NIGHT"
DEFAULT_ARM_AWAY = "ARM_AWAY"
DEFAULT_ARM_HOME = "ARM_HOME"
DEFAULT_ARM_CUSTOM_BYPASS = "ARM_CUSTOM_BYPASS"
DEFAULT_DISARM = "DISARM"
DEFAULT_NAME = "MQTT Alarm"
PLATFORM_SCHEMA = (
@ -75,6 +81,9 @@ PLATFORM_SCHEMA = (
vol.Optional(CONF_PAYLOAD_ARM_AWAY, default=DEFAULT_ARM_AWAY): cv.string,
vol.Optional(CONF_PAYLOAD_ARM_HOME, default=DEFAULT_ARM_HOME): cv.string,
vol.Optional(CONF_PAYLOAD_ARM_NIGHT, default=DEFAULT_ARM_NIGHT): cv.string,
vol.Optional(
CONF_PAYLOAD_ARM_CUSTOM_BYPASS, default=DEFAULT_ARM_CUSTOM_BYPASS
): cv.string,
vol.Optional(CONF_PAYLOAD_DISARM, default=DEFAULT_DISARM): cv.string,
vol.Optional(CONF_RETAIN, default=mqtt.DEFAULT_RETAIN): cv.boolean,
vol.Required(CONF_STATE_TOPIC): mqtt.valid_subscribe_topic,
@ -181,7 +190,10 @@ class MqttAlarm(
STATE_ALARM_ARMED_HOME,
STATE_ALARM_ARMED_AWAY,
STATE_ALARM_ARMED_NIGHT,
STATE_ALARM_ARMED_CUSTOM_BYPASS,
STATE_ALARM_PENDING,
STATE_ALARM_ARMING,
STATE_ALARM_DISARMING,
STATE_ALARM_TRIGGERED,
):
_LOGGER.warning("Received unexpected payload: %s", msg.payload)
@ -233,7 +245,12 @@ class MqttAlarm(
@property
def supported_features(self) -> int:
"""Return the list of supported features."""
return SUPPORT_ALARM_ARM_HOME | SUPPORT_ALARM_ARM_AWAY | SUPPORT_ALARM_ARM_NIGHT
return (
SUPPORT_ALARM_ARM_HOME
| SUPPORT_ALARM_ARM_AWAY
| SUPPORT_ALARM_ARM_NIGHT
| SUPPORT_ALARM_ARM_CUSTOM_BYPASS
)
@property
def code_format(self):
@ -295,6 +312,17 @@ class MqttAlarm(
action = self._config[CONF_PAYLOAD_ARM_NIGHT]
self._publish(code, action)
async def async_alarm_arm_custom_bypass(self, code=None):
"""Send arm custom bypass command.
This method is a coroutine.
"""
code_required = self._config[CONF_CODE_ARM_REQUIRED]
if code_required and not self._validate_code(code, "arming custom bypass"):
return
action = self._config[CONF_PAYLOAD_ARM_CUSTOM_BYPASS]
self._publish(code, action)
def _publish(self, code, action):
"""Publish via mqtt."""
command_template = self._config[CONF_COMMAND_TEMPLATE]

View File

@ -5,9 +5,12 @@ import json
from homeassistant.components import alarm_control_panel
from homeassistant.const import (
STATE_ALARM_ARMED_AWAY,
STATE_ALARM_ARMED_CUSTOM_BYPASS,
STATE_ALARM_ARMED_HOME,
STATE_ALARM_ARMED_NIGHT,
STATE_ALARM_ARMING,
STATE_ALARM_DISARMED,
STATE_ALARM_DISARMING,
STATE_ALARM_PENDING,
STATE_ALARM_TRIGGERED,
STATE_UNKNOWN,
@ -112,7 +115,10 @@ async def test_update_state_via_state_topic(hass, mqtt_mock):
STATE_ALARM_ARMED_HOME,
STATE_ALARM_ARMED_AWAY,
STATE_ALARM_ARMED_NIGHT,
STATE_ALARM_ARMED_CUSTOM_BYPASS,
STATE_ALARM_PENDING,
STATE_ALARM_ARMING,
STATE_ALARM_DISARMING,
STATE_ALARM_TRIGGERED,
):
async_fire_mqtt_message(hass, "alarm/state", state)
@ -256,6 +262,80 @@ async def test_arm_night_publishes_mqtt_when_code_not_req(hass, mqtt_mock):
)
async def test_arm_custom_bypass_publishes_mqtt(hass, mqtt_mock):
"""Test publishing of MQTT messages while armed."""
assert await async_setup_component(
hass,
alarm_control_panel.DOMAIN,
{
alarm_control_panel.DOMAIN: {
"platform": "mqtt",
"name": "test",
"state_topic": "alarm/state",
"command_topic": "alarm/command",
}
},
)
await common.async_alarm_arm_custom_bypass(hass)
mqtt_mock.async_publish.assert_called_once_with(
"alarm/command", "ARM_CUSTOM_BYPASS", 0, False
)
async def test_arm_custom_bypass_not_publishes_mqtt_with_invalid_code_when_req(
hass, mqtt_mock
):
"""Test not publishing of MQTT messages with invalid code.
When code_arm_required = True
"""
assert await async_setup_component(
hass,
alarm_control_panel.DOMAIN,
{
alarm_control_panel.DOMAIN: {
"platform": "mqtt",
"name": "test",
"state_topic": "alarm/state",
"command_topic": "alarm/command",
"code": "1234",
"code_arm_required": True,
}
},
)
call_count = mqtt_mock.async_publish.call_count
await common.async_alarm_arm_custom_bypass(hass, "abcd")
assert mqtt_mock.async_publish.call_count == call_count
async def test_arm_custom_bypass_publishes_mqtt_when_code_not_req(hass, mqtt_mock):
"""Test publishing of MQTT messages.
When code_arm_required = False
"""
assert await async_setup_component(
hass,
alarm_control_panel.DOMAIN,
{
alarm_control_panel.DOMAIN: {
"platform": "mqtt",
"name": "test",
"state_topic": "alarm/state",
"command_topic": "alarm/command",
"code": "1234",
"code_arm_required": False,
}
},
)
await common.async_alarm_arm_custom_bypass(hass)
mqtt_mock.async_publish.assert_called_once_with(
"alarm/command", "ARM_CUSTOM_BYPASS", 0, False
)
async def test_disarm_publishes_mqtt(hass, mqtt_mock):
"""Test publishing of MQTT messages while disarmed."""
assert await async_setup_component(