Added support for ARM_NIGHT for manual_mqtt alarm (#9358)

* - Added support for ARM_NIGHT for manual_mqtt alarm

* - port "Add post_pending_state attribute to manual alarm_control_panel #9291" to manuql_mqtt

* - port "Fixed manual alarm not re-arm after 2nd trigger #9249" to manuql_mqtt

* - port "Add manual alarm_control_panel pending time per state #9264" to manuql_mqtt

* - Updated test_trigger_with_specific_pending to simulate real scenario e.g. arm the system then trigger
pull/9561/head
Joe Lu 2017-09-23 23:57:37 -07:00 committed by Martin Hjelmare
parent 84524e0712
commit bbf6e9ea47
2 changed files with 453 additions and 36 deletions

View File

@ -5,6 +5,7 @@ For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/alarm_control_panel.manual_mqtt/
"""
import asyncio
import copy
import datetime
import logging
@ -13,9 +14,9 @@ import voluptuous as vol
import homeassistant.components.alarm_control_panel as alarm
import homeassistant.util.dt as dt_util
from homeassistant.const import (
STATE_ALARM_ARMED_AWAY, STATE_ALARM_ARMED_HOME, STATE_ALARM_DISARMED,
STATE_ALARM_PENDING, STATE_ALARM_TRIGGERED, CONF_PLATFORM,
CONF_NAME, CONF_CODE, CONF_PENDING_TIME, CONF_TRIGGER_TIME,
STATE_ALARM_ARMED_AWAY, STATE_ALARM_ARMED_HOME, STATE_ALARM_ARMED_NIGHT,
STATE_ALARM_DISARMED, STATE_ALARM_PENDING, STATE_ALARM_TRIGGERED,
CONF_PLATFORM, CONF_NAME, CONF_CODE, CONF_PENDING_TIME, CONF_TRIGGER_TIME,
CONF_DISARM_AFTER_TRIGGER)
import homeassistant.components.mqtt as mqtt
@ -28,6 +29,7 @@ from homeassistant.helpers.event import track_point_in_time
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'
DEFAULT_ALARM_NAME = 'HA Alarm'
DEFAULT_PENDING_TIME = 60
@ -35,11 +37,32 @@ DEFAULT_TRIGGER_TIME = 120
DEFAULT_DISARM_AFTER_TRIGGER = False
DEFAULT_ARM_AWAY = 'ARM_AWAY'
DEFAULT_ARM_HOME = 'ARM_HOME'
DEFAULT_ARM_NIGHT = 'ARM_NIGHT'
DEFAULT_DISARM = 'DISARM'
SUPPORTED_PENDING_STATES = [STATE_ALARM_ARMED_AWAY, STATE_ALARM_ARMED_HOME,
STATE_ALARM_ARMED_NIGHT, STATE_ALARM_TRIGGERED]
ATTR_POST_PENDING_STATE = 'post_pending_state'
def _state_validator(config):
config = copy.deepcopy(config)
for state in SUPPORTED_PENDING_STATES:
if CONF_PENDING_TIME not in config[state]:
config[state][CONF_PENDING_TIME] = config[CONF_PENDING_TIME]
return config
STATE_SETTING_SCHEMA = vol.Schema({
vol.Optional(CONF_PENDING_TIME):
vol.All(vol.Coerce(int), vol.Range(min=0))
})
DEPENDENCIES = ['mqtt']
PLATFORM_SCHEMA = mqtt.MQTT_BASE_PLATFORM_SCHEMA.extend({
PLATFORM_SCHEMA = vol.Schema(vol.All(mqtt.MQTT_BASE_PLATFORM_SCHEMA.extend({
vol.Required(CONF_PLATFORM): 'manual_mqtt',
vol.Optional(CONF_NAME, default=DEFAULT_ALARM_NAME): cv.string,
vol.Optional(CONF_CODE): cv.string,
@ -49,12 +72,17 @@ PLATFORM_SCHEMA = mqtt.MQTT_BASE_PLATFORM_SCHEMA.extend({
vol.All(vol.Coerce(int), vol.Range(min=1)),
vol.Optional(CONF_DISARM_AFTER_TRIGGER,
default=DEFAULT_DISARM_AFTER_TRIGGER): cv.boolean,
vol.Optional(STATE_ALARM_ARMED_AWAY, default={}): STATE_SETTING_SCHEMA,
vol.Optional(STATE_ALARM_ARMED_HOME, default={}): STATE_SETTING_SCHEMA,
vol.Optional(STATE_ALARM_ARMED_NIGHT, default={}): STATE_SETTING_SCHEMA,
vol.Optional(STATE_ALARM_TRIGGERED, default={}): STATE_SETTING_SCHEMA,
vol.Required(mqtt.CONF_COMMAND_TOPIC): mqtt.valid_publish_topic,
vol.Required(mqtt.CONF_STATE_TOPIC): mqtt.valid_subscribe_topic,
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_DISARM, default=DEFAULT_DISARM): cv.string,
})
}), _state_validator))
_LOGGER = logging.getLogger(__name__)
@ -73,7 +101,9 @@ def setup_platform(hass, config, add_devices, discovery_info=None):
config.get(mqtt.CONF_QOS),
config.get(CONF_PAYLOAD_DISARM),
config.get(CONF_PAYLOAD_ARM_HOME),
config.get(CONF_PAYLOAD_ARM_AWAY))])
config.get(CONF_PAYLOAD_ARM_AWAY),
config.get(CONF_PAYLOAD_ARM_NIGHT),
config)])
class ManualMQTTAlarm(alarm.AlarmControlPanel):
@ -89,7 +119,8 @@ class ManualMQTTAlarm(alarm.AlarmControlPanel):
def __init__(self, hass, name, code, pending_time,
trigger_time, disarm_after_trigger,
state_topic, command_topic, qos,
payload_disarm, payload_arm_home, payload_arm_away):
payload_disarm, payload_arm_home, payload_arm_away,
payload_arm_night, config):
"""Init the manual MQTT alarm panel."""
self._state = STATE_ALARM_DISARMED
self._hass = hass
@ -101,12 +132,18 @@ class ManualMQTTAlarm(alarm.AlarmControlPanel):
self._pre_trigger_state = self._state
self._state_ts = None
self._pending_time_by_state = {}
for state in SUPPORTED_PENDING_STATES:
self._pending_time_by_state[state] = datetime.timedelta(
seconds=config[state][CONF_PENDING_TIME])
self._state_topic = state_topic
self._command_topic = command_topic
self._qos = qos
self._payload_disarm = payload_disarm
self._payload_arm_home = payload_arm_home
self._payload_arm_away = payload_arm_away
self._payload_arm_night = payload_arm_night
@property
def should_poll(self):
@ -121,23 +158,27 @@ class ManualMQTTAlarm(alarm.AlarmControlPanel):
@property
def state(self):
"""Return the state of the device."""
if self._state in (STATE_ALARM_ARMED_HOME,
STATE_ALARM_ARMED_AWAY) and \
self._pending_time and self._state_ts + self._pending_time > \
dt_util.utcnow():
return STATE_ALARM_PENDING
if self._state == STATE_ALARM_TRIGGERED and self._trigger_time:
if self._state_ts + self._pending_time > dt_util.utcnow():
if self._within_pending_time(self._state):
return STATE_ALARM_PENDING
elif (self._state_ts + self._pending_time +
elif (self._state_ts + self._pending_time_by_state[self._state] +
self._trigger_time) < dt_util.utcnow():
if self._disarm_after_trigger:
return STATE_ALARM_DISARMED
return self._pre_trigger_state
else:
self._state = self._pre_trigger_state
return self._state
if self._state in SUPPORTED_PENDING_STATES and \
self._within_pending_time(self._state):
return STATE_ALARM_PENDING
return self._state
def _within_pending_time(self, state):
pending_time = self._pending_time_by_state[state]
return self._state_ts + pending_time > dt_util.utcnow()
@property
def code_format(self):
"""One or more characters."""
@ -157,44 +198,47 @@ class ManualMQTTAlarm(alarm.AlarmControlPanel):
if not self._validate_code(code, STATE_ALARM_ARMED_HOME):
return
self._state = STATE_ALARM_ARMED_HOME
self._state_ts = dt_util.utcnow()
self.schedule_update_ha_state()
if self._pending_time:
track_point_in_time(
self._hass, self.async_update_ha_state,
self._state_ts + self._pending_time)
self._update_state(STATE_ALARM_ARMED_HOME)
def alarm_arm_away(self, code=None):
"""Send arm away command."""
if not self._validate_code(code, STATE_ALARM_ARMED_AWAY):
return
self._state = STATE_ALARM_ARMED_AWAY
self._state_ts = dt_util.utcnow()
self.schedule_update_ha_state()
self._update_state(STATE_ALARM_ARMED_AWAY)
if self._pending_time:
track_point_in_time(
self._hass, self.async_update_ha_state,
self._state_ts + self._pending_time)
def alarm_arm_night(self, code=None):
"""Send arm night command."""
if not self._validate_code(code, STATE_ALARM_ARMED_NIGHT):
return
self._update_state(STATE_ALARM_ARMED_NIGHT)
def alarm_trigger(self, code=None):
"""Send alarm trigger command. No code needed."""
self._pre_trigger_state = self._state
self._state = STATE_ALARM_TRIGGERED
self._update_state(STATE_ALARM_TRIGGERED)
def _update_state(self, state):
self._state = state
self._state_ts = dt_util.utcnow()
self.schedule_update_ha_state()
if self._trigger_time:
pending_time = self._pending_time_by_state[state]
if state == STATE_ALARM_TRIGGERED and self._trigger_time:
track_point_in_time(
self._hass, self.async_update_ha_state,
self._state_ts + self._pending_time)
self._state_ts + pending_time)
track_point_in_time(
self._hass, self.async_update_ha_state,
self._state_ts + self._pending_time + self._trigger_time)
self._state_ts + self._trigger_time + pending_time)
elif state in SUPPORTED_PENDING_STATES and pending_time:
track_point_in_time(
self._hass, self.async_update_ha_state,
self._state_ts + pending_time)
def _validate_code(self, code, state):
"""Validate given code."""
@ -203,6 +247,16 @@ class ManualMQTTAlarm(alarm.AlarmControlPanel):
_LOGGER.warning("Invalid code given for %s", state)
return check
@property
def device_state_attributes(self):
"""Return the state attributes."""
state_attr = {}
if self.state == STATE_ALARM_PENDING:
state_attr[ATTR_POST_PENDING_STATE] = self._state
return state_attr
def async_added_to_hass(self):
"""Subscribe mqtt events.
@ -221,6 +275,8 @@ class ManualMQTTAlarm(alarm.AlarmControlPanel):
self.async_alarm_arm_home(self._code)
elif payload == self._payload_arm_away:
self.async_alarm_arm_away(self._code)
elif payload == self._payload_arm_night:
self.async_alarm_arm_night(self._code)
else:
_LOGGER.warning("Received unexpected payload: %s", payload)
return

View File

@ -6,7 +6,7 @@ from unittest.mock import patch
from homeassistant.setup import setup_component
from homeassistant.const import (
STATE_ALARM_DISARMED, STATE_ALARM_ARMED_HOME, STATE_ALARM_ARMED_AWAY,
STATE_ALARM_PENDING, STATE_ALARM_TRIGGERED)
STATE_ALARM_ARMED_NIGHT, STATE_ALARM_PENDING, STATE_ALARM_TRIGGERED)
from homeassistant.components import alarm_control_panel
import homeassistant.util.dt as dt_util
@ -100,6 +100,11 @@ class TestAlarmControlPanelManualMqtt(unittest.TestCase):
self.assertEqual(STATE_ALARM_PENDING,
self.hass.states.get(entity_id).state)
self.assertTrue(
self.hass.states.is_state_attr(entity_id,
'post_pending_state',
STATE_ALARM_ARMED_HOME))
future = dt_util.utcnow() + timedelta(seconds=1)
with patch(('homeassistant.components.alarm_control_panel.manual_mqtt.'
'dt_util.utcnow'), return_value=future):
@ -184,6 +189,11 @@ class TestAlarmControlPanelManualMqtt(unittest.TestCase):
self.assertEqual(STATE_ALARM_PENDING,
self.hass.states.get(entity_id).state)
self.assertTrue(
self.hass.states.is_state_attr(entity_id,
'post_pending_state',
STATE_ALARM_ARMED_AWAY))
future = dt_util.utcnow() + timedelta(seconds=1)
with patch(('homeassistant.components.alarm_control_panel.manual_mqtt.'
'dt_util.utcnow'), return_value=future):
@ -218,6 +228,95 @@ class TestAlarmControlPanelManualMqtt(unittest.TestCase):
self.assertEqual(STATE_ALARM_DISARMED,
self.hass.states.get(entity_id).state)
def test_arm_night_no_pending(self):
"""Test arm night method."""
self.assertTrue(setup_component(
self.hass, alarm_control_panel.DOMAIN,
{'alarm_control_panel': {
'platform': 'manual_mqtt',
'name': 'test',
'code': CODE,
'pending_time': 0,
'disarm_after_trigger': False,
'command_topic': 'alarm/command',
'state_topic': 'alarm/state',
}}))
entity_id = 'alarm_control_panel.test'
self.assertEqual(STATE_ALARM_DISARMED,
self.hass.states.get(entity_id).state)
alarm_control_panel.alarm_arm_night(self.hass, CODE, entity_id)
self.hass.block_till_done()
self.assertEqual(STATE_ALARM_ARMED_NIGHT,
self.hass.states.get(entity_id).state)
def test_arm_night_with_pending(self):
"""Test arm night method."""
self.assertTrue(setup_component(
self.hass, alarm_control_panel.DOMAIN,
{'alarm_control_panel': {
'platform': 'manual_mqtt',
'name': 'test',
'code': CODE,
'pending_time': 1,
'disarm_after_trigger': False,
'command_topic': 'alarm/command',
'state_topic': 'alarm/state',
}}))
entity_id = 'alarm_control_panel.test'
self.assertEqual(STATE_ALARM_DISARMED,
self.hass.states.get(entity_id).state)
alarm_control_panel.alarm_arm_night(self.hass, CODE)
self.hass.block_till_done()
self.assertEqual(STATE_ALARM_PENDING,
self.hass.states.get(entity_id).state)
self.assertTrue(
self.hass.states.is_state_attr(entity_id,
'post_pending_state',
STATE_ALARM_ARMED_NIGHT))
future = dt_util.utcnow() + timedelta(seconds=1)
with patch(('homeassistant.components.alarm_control_panel.manual_mqtt.'
'dt_util.utcnow'), return_value=future):
fire_time_changed(self.hass, future)
self.hass.block_till_done()
self.assertEqual(STATE_ALARM_ARMED_NIGHT,
self.hass.states.get(entity_id).state)
def test_arm_night_with_invalid_code(self):
"""Attempt to arm night without a valid code."""
self.assertTrue(setup_component(
self.hass, alarm_control_panel.DOMAIN,
{'alarm_control_panel': {
'platform': 'manual_mqtt',
'name': 'test',
'code': CODE,
'pending_time': 1,
'disarm_after_trigger': False,
'command_topic': 'alarm/command',
'state_topic': 'alarm/state',
}}))
entity_id = 'alarm_control_panel.test'
self.assertEqual(STATE_ALARM_DISARMED,
self.hass.states.get(entity_id).state)
alarm_control_panel.alarm_arm_night(self.hass, CODE + '2')
self.hass.block_till_done()
self.assertEqual(STATE_ALARM_DISARMED,
self.hass.states.get(entity_id).state)
def test_trigger_no_pending(self):
"""Test triggering when no pending submitted method."""
self.assertTrue(setup_component(
@ -276,6 +375,11 @@ class TestAlarmControlPanelManualMqtt(unittest.TestCase):
self.assertEqual(STATE_ALARM_PENDING,
self.hass.states.get(entity_id).state)
self.assertTrue(
self.hass.states.is_state_attr(entity_id,
'post_pending_state',
STATE_ALARM_TRIGGERED))
future = dt_util.utcnow() + timedelta(seconds=2)
with patch(('homeassistant.components.alarm_control_panel.manual_mqtt.'
'dt_util.utcnow'), return_value=future):
@ -328,6 +432,61 @@ class TestAlarmControlPanelManualMqtt(unittest.TestCase):
self.assertEqual(STATE_ALARM_DISARMED,
self.hass.states.get(entity_id).state)
def test_back_to_back_trigger_with_no_disarm_after_trigger(self):
"""Test no disarm after back to back trigger."""
self.assertTrue(setup_component(
self.hass, alarm_control_panel.DOMAIN,
{'alarm_control_panel': {
'platform': 'manual_mqtt',
'name': 'test',
'trigger_time': 5,
'pending_time': 0,
'disarm_after_trigger': False,
'command_topic': 'alarm/command',
'state_topic': 'alarm/state',
}}))
entity_id = 'alarm_control_panel.test'
self.assertEqual(STATE_ALARM_DISARMED,
self.hass.states.get(entity_id).state)
alarm_control_panel.alarm_arm_away(self.hass, CODE, entity_id)
self.hass.block_till_done()
self.assertEqual(STATE_ALARM_ARMED_AWAY,
self.hass.states.get(entity_id).state)
alarm_control_panel.alarm_trigger(self.hass, entity_id=entity_id)
self.hass.block_till_done()
self.assertEqual(STATE_ALARM_TRIGGERED,
self.hass.states.get(entity_id).state)
future = dt_util.utcnow() + timedelta(seconds=5)
with patch(('homeassistant.components.alarm_control_panel.manual_mqtt.'
'dt_util.utcnow'), return_value=future):
fire_time_changed(self.hass, future)
self.hass.block_till_done()
self.assertEqual(STATE_ALARM_ARMED_AWAY,
self.hass.states.get(entity_id).state)
alarm_control_panel.alarm_trigger(self.hass, entity_id=entity_id)
self.hass.block_till_done()
self.assertEqual(STATE_ALARM_TRIGGERED,
self.hass.states.get(entity_id).state)
future = dt_util.utcnow() + timedelta(seconds=5)
with patch(('homeassistant.components.alarm_control_panel.manual_mqtt.'
'dt_util.utcnow'), return_value=future):
fire_time_changed(self.hass, future)
self.hass.block_till_done()
self.assertEqual(STATE_ALARM_ARMED_AWAY,
self.hass.states.get(entity_id).state)
def test_disarm_while_pending_trigger(self):
"""Test disarming while pending state."""
self.assertTrue(setup_component(
@ -407,6 +566,160 @@ class TestAlarmControlPanelManualMqtt(unittest.TestCase):
self.assertEqual(STATE_ALARM_TRIGGERED,
self.hass.states.get(entity_id).state)
def test_armed_home_with_specific_pending(self):
"""Test arm home method."""
self.assertTrue(setup_component(
self.hass, alarm_control_panel.DOMAIN,
{'alarm_control_panel': {
'platform': 'manual_mqtt',
'name': 'test',
'pending_time': 10,
'armed_home': {
'pending_time': 2
},
'command_topic': 'alarm/command',
'state_topic': 'alarm/state',
}}))
entity_id = 'alarm_control_panel.test'
alarm_control_panel.alarm_arm_home(self.hass)
self.hass.block_till_done()
self.assertEqual(STATE_ALARM_PENDING,
self.hass.states.get(entity_id).state)
future = dt_util.utcnow() + timedelta(seconds=2)
with patch(('homeassistant.components.alarm_control_panel.manual_mqtt.'
'dt_util.utcnow'), return_value=future):
fire_time_changed(self.hass, future)
self.hass.block_till_done()
self.assertEqual(STATE_ALARM_ARMED_HOME,
self.hass.states.get(entity_id).state)
def test_armed_away_with_specific_pending(self):
"""Test arm home method."""
self.assertTrue(setup_component(
self.hass, alarm_control_panel.DOMAIN,
{'alarm_control_panel': {
'platform': 'manual_mqtt',
'name': 'test',
'pending_time': 10,
'armed_away': {
'pending_time': 2
},
'command_topic': 'alarm/command',
'state_topic': 'alarm/state',
}}))
entity_id = 'alarm_control_panel.test'
alarm_control_panel.alarm_arm_away(self.hass)
self.hass.block_till_done()
self.assertEqual(STATE_ALARM_PENDING,
self.hass.states.get(entity_id).state)
future = dt_util.utcnow() + timedelta(seconds=2)
with patch(('homeassistant.components.alarm_control_panel.manual_mqtt.'
'dt_util.utcnow'), return_value=future):
fire_time_changed(self.hass, future)
self.hass.block_till_done()
self.assertEqual(STATE_ALARM_ARMED_AWAY,
self.hass.states.get(entity_id).state)
def test_armed_night_with_specific_pending(self):
"""Test arm home method."""
self.assertTrue(setup_component(
self.hass, alarm_control_panel.DOMAIN,
{'alarm_control_panel': {
'platform': 'manual_mqtt',
'name': 'test',
'pending_time': 10,
'armed_night': {
'pending_time': 2
},
'command_topic': 'alarm/command',
'state_topic': 'alarm/state',
}}))
entity_id = 'alarm_control_panel.test'
alarm_control_panel.alarm_arm_night(self.hass)
self.hass.block_till_done()
self.assertEqual(STATE_ALARM_PENDING,
self.hass.states.get(entity_id).state)
future = dt_util.utcnow() + timedelta(seconds=2)
with patch(('homeassistant.components.alarm_control_panel.manual_mqtt.'
'dt_util.utcnow'), return_value=future):
fire_time_changed(self.hass, future)
self.hass.block_till_done()
self.assertEqual(STATE_ALARM_ARMED_NIGHT,
self.hass.states.get(entity_id).state)
def test_trigger_with_specific_pending(self):
"""Test arm home method."""
self.assertTrue(setup_component(
self.hass, alarm_control_panel.DOMAIN,
{'alarm_control_panel': {
'platform': 'manual_mqtt',
'name': 'test',
'pending_time': 10,
'triggered': {
'pending_time': 2
},
'trigger_time': 3,
'disarm_after_trigger': False,
'command_topic': 'alarm/command',
'state_topic': 'alarm/state',
}}))
entity_id = 'alarm_control_panel.test'
alarm_control_panel.alarm_arm_home(self.hass)
self.hass.block_till_done()
self.assertEqual(STATE_ALARM_PENDING,
self.hass.states.get(entity_id).state)
future = dt_util.utcnow() + timedelta(seconds=10)
with patch(('homeassistant.components.alarm_control_panel.manual_mqtt.'
'dt_util.utcnow'), return_value=future):
fire_time_changed(self.hass, future)
self.hass.block_till_done()
self.assertEqual(STATE_ALARM_ARMED_HOME,
self.hass.states.get(entity_id).state)
alarm_control_panel.alarm_trigger(self.hass)
self.hass.block_till_done()
self.assertEqual(STATE_ALARM_PENDING,
self.hass.states.get(entity_id).state)
future = dt_util.utcnow() + timedelta(seconds=2)
with patch(('homeassistant.components.alarm_control_panel.manual_mqtt.'
'dt_util.utcnow'), return_value=future):
fire_time_changed(self.hass, future)
self.hass.block_till_done()
self.assertEqual(STATE_ALARM_TRIGGERED,
self.hass.states.get(entity_id).state)
future = dt_util.utcnow() + timedelta(seconds=5)
with patch(('homeassistant.components.alarm_control_panel.manual_mqtt.'
'dt_util.utcnow'), return_value=future):
fire_time_changed(self.hass, future)
self.hass.block_till_done()
self.assertEqual(STATE_ALARM_ARMED_HOME,
self.hass.states.get(entity_id).state)
def test_arm_home_via_command_topic(self):
"""Test arming home via command topic."""
assert setup_component(self.hass, alarm_control_panel.DOMAIN, {
@ -475,6 +788,40 @@ class TestAlarmControlPanelManualMqtt(unittest.TestCase):
self.assertEqual(STATE_ALARM_ARMED_AWAY,
self.hass.states.get(entity_id).state)
def test_arm_night_via_command_topic(self):
"""Test arming night via command topic."""
assert setup_component(self.hass, alarm_control_panel.DOMAIN, {
alarm_control_panel.DOMAIN: {
'platform': 'manual_mqtt',
'name': 'test',
'pending_time': 1,
'state_topic': 'alarm/state',
'command_topic': 'alarm/command',
'payload_arm_night': 'ARM_NIGHT',
}
})
entity_id = 'alarm_control_panel.test'
self.assertEqual(STATE_ALARM_DISARMED,
self.hass.states.get(entity_id).state)
# Fire the arm command via MQTT; ensure state changes to pending
fire_mqtt_message(self.hass, 'alarm/command', 'ARM_NIGHT')
self.hass.block_till_done()
self.assertEqual(STATE_ALARM_PENDING,
self.hass.states.get(entity_id).state)
# Fast-forward a little bit
future = dt_util.utcnow() + timedelta(seconds=1)
with patch(('homeassistant.components.alarm_control_panel.manual_mqtt.'
'dt_util.utcnow'), return_value=future):
fire_time_changed(self.hass, future)
self.hass.block_till_done()
self.assertEqual(STATE_ALARM_ARMED_NIGHT,
self.hass.states.get(entity_id).state)
def test_disarm_pending_via_command_topic(self):
"""Test disarming pending alarm via command topic."""
assert setup_component(self.hass, alarm_control_panel.DOMAIN, {
@ -552,6 +899,20 @@ class TestAlarmControlPanelManualMqtt(unittest.TestCase):
self.assertEqual(('alarm/state', STATE_ALARM_ARMED_AWAY, 0, True),
self.mock_publish.mock_calls[-2][1])
# Arm in night mode
alarm_control_panel.alarm_arm_night(self.hass)
self.hass.block_till_done()
self.assertEqual(('alarm/state', STATE_ALARM_PENDING, 0, True),
self.mock_publish.mock_calls[-2][1])
# Fast-forward a little bit
future = dt_util.utcnow() + timedelta(seconds=1)
with patch(('homeassistant.components.alarm_control_panel.manual_mqtt.'
'dt_util.utcnow'), return_value=future):
fire_time_changed(self.hass, future)
self.hass.block_till_done()
self.assertEqual(('alarm/state', STATE_ALARM_ARMED_NIGHT, 0, True),
self.mock_publish.mock_calls[-2][1])
# Disarm
alarm_control_panel.alarm_disarm(self.hass)
self.hass.block_till_done()