From 0f8e48c26df1ceb03c030b988a1d3d9bc3492f5e Mon Sep 17 00:00:00 2001 From: Paolo Bonzini Date: Sun, 3 Dec 2017 13:52:31 +0100 Subject: [PATCH] More declarative timeout syntax for manual alarm control panel. (#10738) More declarative timeout syntax for manual alarm control panel --- .../components/alarm_control_panel/demo.py | 31 +- .../components/alarm_control_panel/manual.py | 161 +++-- .../alarm_control_panel/manual_mqtt.py | 161 +++-- homeassistant/const.py | 1 + .../alarm_control_panel/test_manual.py | 523 ++++++++++++++++ .../alarm_control_panel/test_manual_mqtt.py | 566 +++++++++++++++++- 6 files changed, 1329 insertions(+), 114 deletions(-) diff --git a/homeassistant/components/alarm_control_panel/demo.py b/homeassistant/components/alarm_control_panel/demo.py index aa90fe1f889..c080a136c08 100644 --- a/homeassistant/components/alarm_control_panel/demo.py +++ b/homeassistant/components/alarm_control_panel/demo.py @@ -4,30 +4,45 @@ Demo platform that has two fake alarm control panels. For more details about this platform, please refer to the documentation https://home-assistant.io/components/demo/ """ +import datetime import homeassistant.components.alarm_control_panel.manual as manual from homeassistant.const import ( - STATE_ALARM_ARMED_AWAY, STATE_ALARM_ARMED_HOME, STATE_ALARM_ARMED_NIGHT, - STATE_ALARM_ARMED_CUSTOM_BYPASS, STATE_ALARM_TRIGGERED, CONF_PENDING_TIME) + STATE_ALARM_ARMED_AWAY, STATE_ALARM_ARMED_CUSTOM_BYPASS, + STATE_ALARM_ARMED_HOME, STATE_ALARM_ARMED_NIGHT, + STATE_ALARM_DISARMED, STATE_ALARM_TRIGGERED, CONF_DELAY_TIME, + CONF_PENDING_TIME, CONF_TRIGGER_TIME) def setup_platform(hass, config, add_devices, discovery_info=None): """Set up the Demo alarm control panel platform.""" add_devices([ - manual.ManualAlarm(hass, 'Alarm', '1234', 5, 10, False, { + manual.ManualAlarm(hass, 'Alarm', '1234', None, False, { STATE_ALARM_ARMED_AWAY: { - CONF_PENDING_TIME: 5 + CONF_DELAY_TIME: datetime.timedelta(seconds=0), + CONF_PENDING_TIME: datetime.timedelta(seconds=5), + CONF_TRIGGER_TIME: datetime.timedelta(seconds=10), }, STATE_ALARM_ARMED_HOME: { - CONF_PENDING_TIME: 5 + CONF_DELAY_TIME: datetime.timedelta(seconds=0), + CONF_PENDING_TIME: datetime.timedelta(seconds=5), + CONF_TRIGGER_TIME: datetime.timedelta(seconds=10), }, STATE_ALARM_ARMED_NIGHT: { - CONF_PENDING_TIME: 5 + CONF_DELAY_TIME: datetime.timedelta(seconds=0), + CONF_PENDING_TIME: datetime.timedelta(seconds=5), + CONF_TRIGGER_TIME: datetime.timedelta(seconds=10), + }, + STATE_ALARM_DISARMED: { + CONF_DELAY_TIME: datetime.timedelta(seconds=0), + CONF_TRIGGER_TIME: datetime.timedelta(seconds=10), }, STATE_ALARM_ARMED_CUSTOM_BYPASS: { - CONF_PENDING_TIME: 5 + CONF_DELAY_TIME: datetime.timedelta(seconds=0), + CONF_PENDING_TIME: datetime.timedelta(seconds=5), + CONF_TRIGGER_TIME: datetime.timedelta(seconds=10), }, STATE_ALARM_TRIGGERED: { - CONF_PENDING_TIME: 5 + CONF_PENDING_TIME: datetime.timedelta(seconds=5), }, }), ]) diff --git a/homeassistant/components/alarm_control_panel/manual.py b/homeassistant/components/alarm_control_panel/manual.py index 55f3834c06a..5ff6092493b 100644 --- a/homeassistant/components/alarm_control_panel/manual.py +++ b/homeassistant/components/alarm_control_panel/manual.py @@ -16,24 +16,40 @@ from homeassistant.const import ( STATE_ALARM_ARMED_AWAY, STATE_ALARM_ARMED_HOME, STATE_ALARM_ARMED_NIGHT, STATE_ALARM_ARMED_CUSTOM_BYPASS, 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) + CONF_DELAY_TIME, CONF_PENDING_TIME, CONF_TRIGGER_TIME, + CONF_DISARM_AFTER_TRIGGER) import homeassistant.helpers.config_validation as cv from homeassistant.helpers.event import track_point_in_time +CONF_CODE_TEMPLATE = 'code_template' + DEFAULT_ALARM_NAME = 'HA Alarm' -DEFAULT_PENDING_TIME = 60 -DEFAULT_TRIGGER_TIME = 120 +DEFAULT_DELAY_TIME = datetime.timedelta(seconds=0) +DEFAULT_PENDING_TIME = datetime.timedelta(seconds=60) +DEFAULT_TRIGGER_TIME = datetime.timedelta(seconds=120) DEFAULT_DISARM_AFTER_TRIGGER = False -SUPPORTED_PENDING_STATES = [STATE_ALARM_ARMED_AWAY, STATE_ALARM_ARMED_HOME, - STATE_ALARM_ARMED_NIGHT, STATE_ALARM_TRIGGERED, - STATE_ALARM_ARMED_CUSTOM_BYPASS] +SUPPORTED_STATES = [STATE_ALARM_DISARMED, STATE_ALARM_ARMED_AWAY, + STATE_ALARM_ARMED_HOME, STATE_ALARM_ARMED_NIGHT, + STATE_ALARM_ARMED_CUSTOM_BYPASS, STATE_ALARM_TRIGGERED] +SUPPORTED_PRETRIGGER_STATES = [state for state in SUPPORTED_STATES + if state != STATE_ALARM_TRIGGERED] + +SUPPORTED_PENDING_STATES = [state for state in SUPPORTED_STATES + if state != STATE_ALARM_DISARMED] + +ATTR_PRE_PENDING_STATE = 'pre_pending_state' ATTR_POST_PENDING_STATE = 'post_pending_state' def _state_validator(config): config = copy.deepcopy(config) + for state in SUPPORTED_PRETRIGGER_STATES: + if CONF_DELAY_TIME not in config[state]: + config[state][CONF_DELAY_TIME] = config[CONF_DELAY_TIME] + if CONF_TRIGGER_TIME not in config[state]: + config[state][CONF_TRIGGER_TIME] = config[CONF_TRIGGER_TIME] for state in SUPPORTED_PENDING_STATES: if CONF_PENDING_TIME not in config[state]: config[state][CONF_PENDING_TIME] = config[CONF_PENDING_TIME] @@ -41,28 +57,44 @@ def _state_validator(config): return config -STATE_SETTING_SCHEMA = vol.Schema({ - vol.Optional(CONF_PENDING_TIME): - vol.All(vol.Coerce(int), vol.Range(min=0)) -}) +def _state_schema(state): + schema = {} + if state in SUPPORTED_PRETRIGGER_STATES: + schema[vol.Optional(CONF_DELAY_TIME)] = vol.All( + cv.time_period, cv.positive_timedelta) + schema[vol.Optional(CONF_TRIGGER_TIME)] = vol.All( + cv.time_period, cv.positive_timedelta) + if state in SUPPORTED_PENDING_STATES: + schema[vol.Optional(CONF_PENDING_TIME)] = vol.All( + cv.time_period, cv.positive_timedelta) + return vol.Schema(schema) PLATFORM_SCHEMA = vol.Schema(vol.All({ vol.Required(CONF_PLATFORM): 'manual', vol.Optional(CONF_NAME, default=DEFAULT_ALARM_NAME): cv.string, - vol.Optional(CONF_CODE): cv.string, + vol.Exclusive(CONF_CODE, 'code validation'): cv.string, + vol.Exclusive(CONF_CODE_TEMPLATE, 'code validation'): cv.template, + vol.Optional(CONF_DELAY_TIME, default=DEFAULT_DELAY_TIME): + vol.All(cv.time_period, cv.positive_timedelta), vol.Optional(CONF_PENDING_TIME, default=DEFAULT_PENDING_TIME): - vol.All(vol.Coerce(int), vol.Range(min=0)), + vol.All(cv.time_period, cv.positive_timedelta), vol.Optional(CONF_TRIGGER_TIME, default=DEFAULT_TRIGGER_TIME): - vol.All(vol.Coerce(int), vol.Range(min=1)), + vol.All(cv.time_period, cv.positive_timedelta), 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_ARMED_CUSTOM_BYPASS, - default={}): STATE_SETTING_SCHEMA, - vol.Optional(STATE_ALARM_TRIGGERED, default={}): STATE_SETTING_SCHEMA, + vol.Optional(STATE_ALARM_ARMED_AWAY, default={}): + _state_schema(STATE_ALARM_ARMED_AWAY), + vol.Optional(STATE_ALARM_ARMED_HOME, default={}): + _state_schema(STATE_ALARM_ARMED_HOME), + vol.Optional(STATE_ALARM_ARMED_NIGHT, default={}): + _state_schema(STATE_ALARM_ARMED_NIGHT), + vol.Optional(STATE_ALARM_ARMED_CUSTOM_BYPASS, default={}): + _state_schema(STATE_ALARM_ARMED_CUSTOM_BYPASS), + vol.Optional(STATE_ALARM_DISARMED, default={}): + _state_schema(STATE_ALARM_DISARMED), + vol.Optional(STATE_ALARM_TRIGGERED, default={}): + _state_schema(STATE_ALARM_TRIGGERED), }, _state_validator)) _LOGGER = logging.getLogger(__name__) @@ -74,8 +106,7 @@ def setup_platform(hass, config, add_devices, discovery_info=None): hass, config[CONF_NAME], config.get(CONF_CODE), - config.get(CONF_PENDING_TIME, DEFAULT_PENDING_TIME), - config.get(CONF_TRIGGER_TIME, DEFAULT_TRIGGER_TIME), + config.get(CONF_CODE_TEMPLATE), config.get(CONF_DISARM_AFTER_TRIGGER, DEFAULT_DISARM_AFTER_TRIGGER), config )]) @@ -86,27 +117,37 @@ class ManualAlarm(alarm.AlarmControlPanel): Representation of an alarm status. When armed, will be pending for 'pending_time', after that armed. - When triggered, will be pending for 'trigger_time'. After that will be - triggered for 'trigger_time', after that we return to the previous state - or disarm if `disarm_after_trigger` is true. + When triggered, will be pending for the triggering state's 'delay_time' + plus the triggered state's 'pending_time'. + After that will be triggered for 'trigger_time', after that we return to + the previous state or disarm if `disarm_after_trigger` is true. + A trigger_time of zero disables the alarm_trigger service. """ - def __init__(self, hass, name, code, pending_time, trigger_time, + def __init__(self, hass, name, code, code_template, disarm_after_trigger, config): """Init the manual alarm panel.""" self._state = STATE_ALARM_DISARMED self._hass = hass self._name = name - self._code = str(code) if code else None - self._trigger_time = datetime.timedelta(seconds=trigger_time) + if code_template: + self._code = code_template + self._code.hass = hass + else: + self._code = code or None self._disarm_after_trigger = disarm_after_trigger - self._pre_trigger_state = self._state + self._previous_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._delay_time_by_state = { + state: config[state][CONF_DELAY_TIME] + for state in SUPPORTED_PRETRIGGER_STATES} + self._trigger_time_by_state = { + state: config[state][CONF_TRIGGER_TIME] + for state in SUPPORTED_PRETRIGGER_STATES} + self._pending_time_by_state = { + state: config[state][CONF_PENDING_TIME] + for state in SUPPORTED_PENDING_STATES} @property def should_poll(self): @@ -121,15 +162,16 @@ class ManualAlarm(alarm.AlarmControlPanel): @property def state(self): """Return the state of the device.""" - if self._state == STATE_ALARM_TRIGGERED and self._trigger_time: + if self._state == STATE_ALARM_TRIGGERED: if self._within_pending_time(self._state): return STATE_ALARM_PENDING - elif (self._state_ts + self._pending_time_by_state[self._state] + - self._trigger_time) < dt_util.utcnow(): + trigger_time = self._trigger_time_by_state[self._previous_state] + if (self._state_ts + self._pending_time(self._state) + + trigger_time) < dt_util.utcnow(): if self._disarm_after_trigger: return STATE_ALARM_DISARMED else: - self._state = self._pre_trigger_state + self._state = self._previous_state return self._state if self._state in SUPPORTED_PENDING_STATES and \ @@ -138,9 +180,21 @@ class ManualAlarm(alarm.AlarmControlPanel): return self._state - def _within_pending_time(self, state): + @property + def _active_state(self): + if self.state == STATE_ALARM_PENDING: + return self._previous_state + else: + return self._state + + def _pending_time(self, state): pending_time = self._pending_time_by_state[state] - return self._state_ts + pending_time > dt_util.utcnow() + if state == STATE_ALARM_TRIGGERED: + pending_time += self._delay_time_by_state[self._previous_state] + return pending_time + + def _within_pending_time(self, state): + return self._state_ts + self._pending_time(state) > dt_util.utcnow() @property def code_format(self): @@ -185,26 +239,35 @@ class ManualAlarm(alarm.AlarmControlPanel): self._update_state(STATE_ALARM_ARMED_CUSTOM_BYPASS) def alarm_trigger(self, code=None): - """Send alarm trigger command. No code needed.""" - self._pre_trigger_state = self._state + """ + Send alarm trigger command. + No code needed, a trigger time of zero for the current state + disables the alarm. + """ + if not self._trigger_time_by_state[self._active_state]: + return self._update_state(STATE_ALARM_TRIGGERED) def _update_state(self, state): + if self._state == state: + return + + self._previous_state = self._state self._state = state self._state_ts = dt_util.utcnow() self.schedule_update_ha_state() - pending_time = self._pending_time_by_state[state] - - if state == STATE_ALARM_TRIGGERED and self._trigger_time: + pending_time = self._pending_time(state) + if state == STATE_ALARM_TRIGGERED: track_point_in_time( self._hass, self.async_update_ha_state, self._state_ts + pending_time) + trigger_time = self._trigger_time_by_state[self._previous_state] track_point_in_time( self._hass, self.async_update_ha_state, - self._state_ts + self._trigger_time + pending_time) + self._state_ts + pending_time + trigger_time) elif state in SUPPORTED_PENDING_STATES and pending_time: track_point_in_time( self._hass, self.async_update_ha_state, @@ -212,7 +275,14 @@ class ManualAlarm(alarm.AlarmControlPanel): def _validate_code(self, code, state): """Validate given code.""" - check = self._code is None or code == self._code + if self._code is None: + return True + if isinstance(self._code, str): + alarm_code = self._code + else: + alarm_code = self._code.render(from_state=self._state, + to_state=state) + check = not alarm_code or code == alarm_code if not check: _LOGGER.warning("Invalid code given for %s", state) return check @@ -223,6 +293,7 @@ class ManualAlarm(alarm.AlarmControlPanel): state_attr = {} if self.state == STATE_ALARM_PENDING: + state_attr[ATTR_PRE_PENDING_STATE] = self._previous_state state_attr[ATTR_POST_PENDING_STATE] = self._state return state_attr diff --git a/homeassistant/components/alarm_control_panel/manual_mqtt.py b/homeassistant/components/alarm_control_panel/manual_mqtt.py index 44247616b59..9e388806e73 100644 --- a/homeassistant/components/alarm_control_panel/manual_mqtt.py +++ b/homeassistant/components/alarm_control_panel/manual_mqtt.py @@ -16,8 +16,8 @@ import homeassistant.util.dt as dt_util from homeassistant.const import ( 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) + CONF_PLATFORM, CONF_NAME, CONF_CODE, CONF_DELAY_TIME, CONF_PENDING_TIME, + CONF_TRIGGER_TIME, CONF_DISARM_AFTER_TRIGGER) import homeassistant.components.mqtt as mqtt from homeassistant.helpers.event import async_track_state_change @@ -26,28 +26,44 @@ from homeassistant.core import callback import homeassistant.helpers.config_validation as cv from homeassistant.helpers.event import track_point_in_time +CONF_CODE_TEMPLATE = 'code_template' + 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 -DEFAULT_TRIGGER_TIME = 120 +DEFAULT_DELAY_TIME = datetime.timedelta(seconds=0) +DEFAULT_PENDING_TIME = datetime.timedelta(seconds=60) +DEFAULT_TRIGGER_TIME = datetime.timedelta(seconds=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] +SUPPORTED_STATES = [STATE_ALARM_DISARMED, STATE_ALARM_ARMED_AWAY, + STATE_ALARM_ARMED_HOME, STATE_ALARM_ARMED_NIGHT, + STATE_ALARM_TRIGGERED] +SUPPORTED_PRETRIGGER_STATES = [state for state in SUPPORTED_STATES + if state != STATE_ALARM_TRIGGERED] + +SUPPORTED_PENDING_STATES = [state for state in SUPPORTED_STATES + if state != STATE_ALARM_DISARMED] + +ATTR_PRE_PENDING_STATE = 'pre_pending_state' ATTR_POST_PENDING_STATE = 'post_pending_state' def _state_validator(config): config = copy.deepcopy(config) + for state in SUPPORTED_PRETRIGGER_STATES: + if CONF_DELAY_TIME not in config[state]: + config[state][CONF_DELAY_TIME] = config[CONF_DELAY_TIME] + if CONF_TRIGGER_TIME not in config[state]: + config[state][CONF_TRIGGER_TIME] = config[CONF_TRIGGER_TIME] for state in SUPPORTED_PENDING_STATES: if CONF_PENDING_TIME not in config[state]: config[state][CONF_PENDING_TIME] = config[CONF_PENDING_TIME] @@ -55,27 +71,44 @@ def _state_validator(config): return config -STATE_SETTING_SCHEMA = vol.Schema({ - vol.Optional(CONF_PENDING_TIME): - vol.All(vol.Coerce(int), vol.Range(min=0)) -}) +def _state_schema(state): + schema = {} + if state in SUPPORTED_PRETRIGGER_STATES: + schema[vol.Optional(CONF_DELAY_TIME)] = vol.All( + cv.time_period, cv.positive_timedelta) + schema[vol.Optional(CONF_TRIGGER_TIME)] = vol.All( + cv.time_period, cv.positive_timedelta) + if state in SUPPORTED_PENDING_STATES: + schema[vol.Optional(CONF_PENDING_TIME)] = vol.All( + cv.time_period, cv.positive_timedelta) + return vol.Schema(schema) + DEPENDENCIES = ['mqtt'] 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, + vol.Exclusive(CONF_CODE, 'code validation'): cv.string, + vol.Exclusive(CONF_CODE_TEMPLATE, 'code validation'): cv.template, + vol.Optional(CONF_DELAY_TIME, default=DEFAULT_DELAY_TIME): + vol.All(cv.time_period, cv.positive_timedelta), vol.Optional(CONF_PENDING_TIME, default=DEFAULT_PENDING_TIME): - vol.All(vol.Coerce(int), vol.Range(min=0)), + vol.All(cv.time_period, cv.positive_timedelta), vol.Optional(CONF_TRIGGER_TIME, default=DEFAULT_TRIGGER_TIME): - vol.All(vol.Coerce(int), vol.Range(min=1)), + vol.All(cv.time_period, cv.positive_timedelta), 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.Optional(STATE_ALARM_ARMED_AWAY, default={}): + _state_schema(STATE_ALARM_ARMED_AWAY), + vol.Optional(STATE_ALARM_ARMED_HOME, default={}): + _state_schema(STATE_ALARM_ARMED_HOME), + vol.Optional(STATE_ALARM_ARMED_NIGHT, default={}): + _state_schema(STATE_ALARM_ARMED_NIGHT), + vol.Optional(STATE_ALARM_DISARMED, default={}): + _state_schema(STATE_ALARM_DISARMED), + vol.Optional(STATE_ALARM_TRIGGERED, default={}): + _state_schema(STATE_ALARM_TRIGGERED), 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, @@ -93,8 +126,7 @@ def setup_platform(hass, config, add_devices, discovery_info=None): hass, config[CONF_NAME], config.get(CONF_CODE), - config.get(CONF_PENDING_TIME, DEFAULT_PENDING_TIME), - config.get(CONF_TRIGGER_TIME, DEFAULT_TRIGGER_TIME), + config.get(CONF_CODE_TEMPLATE), config.get(CONF_DISARM_AFTER_TRIGGER, DEFAULT_DISARM_AFTER_TRIGGER), config.get(mqtt.CONF_STATE_TOPIC), config.get(mqtt.CONF_COMMAND_TOPIC), @@ -111,13 +143,15 @@ class ManualMQTTAlarm(alarm.AlarmControlPanel): Representation of an alarm status. When armed, will be pending for 'pending_time', after that armed. - When triggered, will be pending for 'trigger_time'. After that will be - triggered for 'trigger_time', after that we return to the previous state - or disarm if `disarm_after_trigger` is true. + When triggered, will be pending for the triggering state's 'delay_time' + plus the triggered state's 'pending_time'. + After that will be triggered for 'trigger_time', after that we return to + the previous state or disarm if `disarm_after_trigger` is true. + A trigger_time of zero disables the alarm_trigger service. """ - def __init__(self, hass, name, code, pending_time, - trigger_time, disarm_after_trigger, + def __init__(self, hass, name, code, code_template, + disarm_after_trigger, state_topic, command_topic, qos, payload_disarm, payload_arm_home, payload_arm_away, payload_arm_night, config): @@ -125,17 +159,24 @@ class ManualMQTTAlarm(alarm.AlarmControlPanel): self._state = STATE_ALARM_DISARMED self._hass = hass self._name = name - self._code = str(code) if code else None - self._pending_time = datetime.timedelta(seconds=pending_time) - self._trigger_time = datetime.timedelta(seconds=trigger_time) + if code_template: + self._code = code_template + self._code.hass = hass + else: + self._code = code or None self._disarm_after_trigger = disarm_after_trigger - self._pre_trigger_state = self._state + self._previous_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._delay_time_by_state = { + state: config[state][CONF_DELAY_TIME] + for state in SUPPORTED_PRETRIGGER_STATES} + self._trigger_time_by_state = { + state: config[state][CONF_TRIGGER_TIME] + for state in SUPPORTED_PRETRIGGER_STATES} + self._pending_time_by_state = { + state: config[state][CONF_PENDING_TIME] + for state in SUPPORTED_PENDING_STATES} self._state_topic = state_topic self._command_topic = command_topic @@ -158,15 +199,16 @@ class ManualMQTTAlarm(alarm.AlarmControlPanel): @property def state(self): """Return the state of the device.""" - if self._state == STATE_ALARM_TRIGGERED and self._trigger_time: + if self._state == STATE_ALARM_TRIGGERED: if self._within_pending_time(self._state): return STATE_ALARM_PENDING - elif (self._state_ts + self._pending_time_by_state[self._state] + - self._trigger_time) < dt_util.utcnow(): + trigger_time = self._trigger_time_by_state[self._previous_state] + if (self._state_ts + self._pending_time(self._state) + + trigger_time) < dt_util.utcnow(): if self._disarm_after_trigger: return STATE_ALARM_DISARMED else: - self._state = self._pre_trigger_state + self._state = self._previous_state return self._state if self._state in SUPPORTED_PENDING_STATES and \ @@ -175,9 +217,21 @@ class ManualMQTTAlarm(alarm.AlarmControlPanel): return self._state - def _within_pending_time(self, state): + @property + def _active_state(self): + if self.state == STATE_ALARM_PENDING: + return self._previous_state + else: + return self._state + + def _pending_time(self, state): pending_time = self._pending_time_by_state[state] - return self._state_ts + pending_time > dt_util.utcnow() + if state == STATE_ALARM_TRIGGERED: + pending_time += self._delay_time_by_state[self._previous_state] + return pending_time + + def _within_pending_time(self, state): + return self._state_ts + self._pending_time(state) > dt_util.utcnow() @property def code_format(self): @@ -215,26 +269,35 @@ class ManualMQTTAlarm(alarm.AlarmControlPanel): 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 + """ + Send alarm trigger command. + No code needed, a trigger time of zero for the current state + disables the alarm. + """ + if not self._trigger_time_by_state[self._active_state]: + return self._update_state(STATE_ALARM_TRIGGERED) def _update_state(self, state): + if self._state == state: + return + + self._previous_state = self._state self._state = state self._state_ts = dt_util.utcnow() self.schedule_update_ha_state() - pending_time = self._pending_time_by_state[state] - - if state == STATE_ALARM_TRIGGERED and self._trigger_time: + pending_time = self._pending_time(state) + if state == STATE_ALARM_TRIGGERED: track_point_in_time( self._hass, self.async_update_ha_state, self._state_ts + pending_time) + trigger_time = self._trigger_time_by_state[self._previous_state] track_point_in_time( self._hass, self.async_update_ha_state, - self._state_ts + self._trigger_time + pending_time) + self._state_ts + pending_time + trigger_time) elif state in SUPPORTED_PENDING_STATES and pending_time: track_point_in_time( self._hass, self.async_update_ha_state, @@ -242,7 +305,14 @@ class ManualMQTTAlarm(alarm.AlarmControlPanel): def _validate_code(self, code, state): """Validate given code.""" - check = self._code is None or code == self._code + if self._code is None: + return True + if isinstance(self._code, str): + alarm_code = self._code + else: + alarm_code = self._code.render(from_state=self._state, + to_state=state) + check = not alarm_code or code == alarm_code if not check: _LOGGER.warning("Invalid code given for %s", state) return check @@ -253,6 +323,7 @@ class ManualMQTTAlarm(alarm.AlarmControlPanel): state_attr = {} if self.state == STATE_ALARM_PENDING: + state_attr[ATTR_PRE_PENDING_STATE] = self._previous_state state_attr[ATTR_POST_PENDING_STATE] = self._state return state_attr diff --git a/homeassistant/const.py b/homeassistant/const.py index beb34146e70..9d394bb4a99 100644 --- a/homeassistant/const.py +++ b/homeassistant/const.py @@ -52,6 +52,7 @@ CONF_CURRENCY = 'currency' CONF_CUSTOMIZE = 'customize' CONF_CUSTOMIZE_DOMAIN = 'customize_domain' CONF_CUSTOMIZE_GLOB = 'customize_glob' +CONF_DELAY_TIME = 'delay_time' CONF_DEVICE = 'device' CONF_DEVICE_CLASS = 'device_class' CONF_DEVICES = 'devices' diff --git a/tests/components/alarm_control_panel/test_manual.py b/tests/components/alarm_control_panel/test_manual.py index d65568b0844..c47ed941b65 100644 --- a/tests/components/alarm_control_panel/test_manual.py +++ b/tests/components/alarm_control_panel/test_manual.py @@ -140,6 +140,32 @@ class TestAlarmControlPanelManual(unittest.TestCase): self.assertEqual(STATE_ALARM_ARMED_AWAY, self.hass.states.get(entity_id).state) + def test_arm_home_with_template_code(self): + """Attempt to arm with a template-based code.""" + self.assertTrue(setup_component( + self.hass, alarm_control_panel.DOMAIN, + {'alarm_control_panel': { + 'platform': 'manual', + 'name': 'test', + 'code_template': '{{ "abc" }}', + 'pending_time': 0, + 'disarm_after_trigger': False + }})) + + entity_id = 'alarm_control_panel.test' + + self.hass.start() + self.hass.block_till_done() + + self.assertEqual(STATE_ALARM_DISARMED, + self.hass.states.get(entity_id).state) + + alarm_control_panel.alarm_arm_home(self.hass, 'abc') + self.hass.block_till_done() + + state = self.hass.states.get(entity_id) + self.assertEqual(STATE_ALARM_ARMED_HOME, state.state) + def test_arm_away_with_pending(self): """Test arm home method.""" self.assertTrue(setup_component( @@ -257,6 +283,13 @@ class TestAlarmControlPanelManual(unittest.TestCase): state = self.hass.states.get(entity_id) assert state.state == STATE_ALARM_ARMED_NIGHT + # Do not go to the pending state when updating to the same 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_invalid_code(self): """Attempt to night home without a valid code.""" self.assertTrue(setup_component( @@ -311,6 +344,93 @@ class TestAlarmControlPanelManual(unittest.TestCase): self.assertEqual(STATE_ALARM_TRIGGERED, self.hass.states.get(entity_id).state) + def test_trigger_with_delay(self): + """Test trigger method and switch from pending to triggered.""" + self.assertTrue(setup_component( + self.hass, alarm_control_panel.DOMAIN, + {'alarm_control_panel': { + 'platform': 'manual', + 'name': 'test', + 'code': CODE, + 'delay_time': 1, + 'pending_time': 0, + 'disarm_after_trigger': False + }})) + + 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) + 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() + + state = self.hass.states.get(entity_id) + self.assertEqual(STATE_ALARM_PENDING, state.state) + self.assertEqual(STATE_ALARM_TRIGGERED, + state.attributes['post_pending_state']) + + future = dt_util.utcnow() + timedelta(seconds=1) + with patch(('homeassistant.components.alarm_control_panel.manual.' + 'dt_util.utcnow'), return_value=future): + fire_time_changed(self.hass, future) + self.hass.block_till_done() + + state = self.hass.states.get(entity_id) + self.assertEqual(STATE_ALARM_TRIGGERED, state.state) + + def test_trigger_zero_trigger_time(self): + """Test disabled trigger.""" + self.assertTrue(setup_component( + self.hass, alarm_control_panel.DOMAIN, + {'alarm_control_panel': { + 'platform': 'manual', + 'name': 'test', + 'pending_time': 0, + 'trigger_time': 0, + 'disarm_after_trigger': False + }})) + + entity_id = 'alarm_control_panel.test' + + self.assertEqual(STATE_ALARM_DISARMED, + self.hass.states.get(entity_id).state) + + alarm_control_panel.alarm_trigger(self.hass) + self.hass.block_till_done() + + self.assertEqual(STATE_ALARM_DISARMED, + self.hass.states.get(entity_id).state) + + def test_trigger_zero_trigger_time_with_pending(self): + """Test disabled trigger.""" + self.assertTrue(setup_component( + self.hass, alarm_control_panel.DOMAIN, + {'alarm_control_panel': { + 'platform': 'manual', + 'name': 'test', + 'pending_time': 2, + 'trigger_time': 0, + 'disarm_after_trigger': False + }})) + + entity_id = 'alarm_control_panel.test' + + self.assertEqual(STATE_ALARM_DISARMED, + self.hass.states.get(entity_id).state) + + alarm_control_panel.alarm_trigger(self.hass) + self.hass.block_till_done() + + self.assertEqual(STATE_ALARM_DISARMED, + self.hass.states.get(entity_id).state) + def test_trigger_with_pending(self): """Test arm home method.""" self.assertTrue(setup_component( @@ -355,6 +475,203 @@ class TestAlarmControlPanelManual(unittest.TestCase): state = self.hass.states.get(entity_id) assert state.state == STATE_ALARM_DISARMED + def test_trigger_with_unused_specific_delay(self): + """Test trigger method and switch from pending to triggered.""" + self.assertTrue(setup_component( + self.hass, alarm_control_panel.DOMAIN, + {'alarm_control_panel': { + 'platform': 'manual', + 'name': 'test', + 'code': CODE, + 'delay_time': 5, + 'pending_time': 0, + 'armed_home': { + 'delay_time': 10 + }, + 'disarm_after_trigger': False + }})) + + 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) + 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() + + state = self.hass.states.get(entity_id) + self.assertEqual(STATE_ALARM_PENDING, state.state) + self.assertEqual(STATE_ALARM_TRIGGERED, + state.attributes['post_pending_state']) + + future = dt_util.utcnow() + timedelta(seconds=5) + with patch(('homeassistant.components.alarm_control_panel.manual.' + 'dt_util.utcnow'), return_value=future): + fire_time_changed(self.hass, future) + self.hass.block_till_done() + + state = self.hass.states.get(entity_id) + assert state.state == STATE_ALARM_TRIGGERED + + def test_trigger_with_specific_delay(self): + """Test trigger method and switch from pending to triggered.""" + self.assertTrue(setup_component( + self.hass, alarm_control_panel.DOMAIN, + {'alarm_control_panel': { + 'platform': 'manual', + 'name': 'test', + 'code': CODE, + 'delay_time': 10, + 'pending_time': 0, + 'armed_away': { + 'delay_time': 1 + }, + 'disarm_after_trigger': False + }})) + + 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) + 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() + + state = self.hass.states.get(entity_id) + self.assertEqual(STATE_ALARM_PENDING, state.state) + self.assertEqual(STATE_ALARM_TRIGGERED, + state.attributes['post_pending_state']) + + future = dt_util.utcnow() + timedelta(seconds=1) + with patch(('homeassistant.components.alarm_control_panel.manual.' + 'dt_util.utcnow'), return_value=future): + fire_time_changed(self.hass, future) + self.hass.block_till_done() + + state = self.hass.states.get(entity_id) + assert state.state == STATE_ALARM_TRIGGERED + + def test_trigger_with_pending_and_delay(self): + """Test trigger method and switch from pending to triggered.""" + self.assertTrue(setup_component( + self.hass, alarm_control_panel.DOMAIN, + {'alarm_control_panel': { + 'platform': 'manual', + 'name': 'test', + 'code': CODE, + 'delay_time': 1, + 'pending_time': 0, + 'triggered': { + 'pending_time': 1 + }, + 'disarm_after_trigger': False + }})) + + 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) + 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() + + state = self.hass.states.get(entity_id) + assert state.state == STATE_ALARM_PENDING + assert state.attributes['post_pending_state'] == STATE_ALARM_TRIGGERED + + future = dt_util.utcnow() + timedelta(seconds=1) + with patch(('homeassistant.components.alarm_control_panel.manual.' + 'dt_util.utcnow'), return_value=future): + fire_time_changed(self.hass, future) + self.hass.block_till_done() + + state = self.hass.states.get(entity_id) + assert state.state == STATE_ALARM_PENDING + assert state.attributes['post_pending_state'] == STATE_ALARM_TRIGGERED + + future += timedelta(seconds=1) + with patch(('homeassistant.components.alarm_control_panel.manual.' + 'dt_util.utcnow'), return_value=future): + fire_time_changed(self.hass, future) + self.hass.block_till_done() + + state = self.hass.states.get(entity_id) + assert state.state == STATE_ALARM_TRIGGERED + + def test_trigger_with_pending_and_specific_delay(self): + """Test trigger method and switch from pending to triggered.""" + self.assertTrue(setup_component( + self.hass, alarm_control_panel.DOMAIN, + {'alarm_control_panel': { + 'platform': 'manual', + 'name': 'test', + 'code': CODE, + 'delay_time': 10, + 'pending_time': 0, + 'armed_away': { + 'delay_time': 1 + }, + 'triggered': { + 'pending_time': 1 + }, + 'disarm_after_trigger': False + }})) + + 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) + 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() + + state = self.hass.states.get(entity_id) + assert state.state == STATE_ALARM_PENDING + assert state.attributes['post_pending_state'] == STATE_ALARM_TRIGGERED + + future = dt_util.utcnow() + timedelta(seconds=1) + with patch(('homeassistant.components.alarm_control_panel.manual.' + 'dt_util.utcnow'), return_value=future): + fire_time_changed(self.hass, future) + self.hass.block_till_done() + + state = self.hass.states.get(entity_id) + assert state.state == STATE_ALARM_PENDING + assert state.attributes['post_pending_state'] == STATE_ALARM_TRIGGERED + + future += timedelta(seconds=1) + with patch(('homeassistant.components.alarm_control_panel.manual.' + 'dt_util.utcnow'), return_value=future): + fire_time_changed(self.hass, future) + self.hass.block_till_done() + + state = self.hass.states.get(entity_id) + assert state.state == STATE_ALARM_TRIGGERED + def test_armed_home_with_specific_pending(self): """Test arm home method.""" self.assertTrue(setup_component( @@ -518,6 +835,101 @@ class TestAlarmControlPanelManual(unittest.TestCase): self.assertEqual(STATE_ALARM_DISARMED, self.hass.states.get(entity_id).state) + def test_trigger_with_zero_specific_trigger_time(self): + """Test trigger method.""" + self.assertTrue(setup_component( + self.hass, alarm_control_panel.DOMAIN, + {'alarm_control_panel': { + 'platform': 'manual', + 'name': 'test', + 'trigger_time': 5, + 'disarmed': { + 'trigger_time': 0 + }, + 'pending_time': 0, + 'disarm_after_trigger': True + }})) + + entity_id = 'alarm_control_panel.test' + + self.assertEqual(STATE_ALARM_DISARMED, + 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_DISARMED, + self.hass.states.get(entity_id).state) + + def test_trigger_with_unused_zero_specific_trigger_time(self): + """Test disarm after trigger.""" + self.assertTrue(setup_component( + self.hass, alarm_control_panel.DOMAIN, + {'alarm_control_panel': { + 'platform': 'manual', + 'name': 'test', + 'trigger_time': 5, + 'armed_home': { + 'trigger_time': 0 + }, + 'pending_time': 0, + 'disarm_after_trigger': True + }})) + + entity_id = 'alarm_control_panel.test' + + self.assertEqual(STATE_ALARM_DISARMED, + 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.' + 'dt_util.utcnow'), return_value=future): + fire_time_changed(self.hass, future) + self.hass.block_till_done() + + self.assertEqual(STATE_ALARM_DISARMED, + self.hass.states.get(entity_id).state) + + def test_trigger_with_specific_trigger_time(self): + """Test disarm after trigger.""" + self.assertTrue(setup_component( + self.hass, alarm_control_panel.DOMAIN, + {'alarm_control_panel': { + 'platform': 'manual', + 'name': 'test', + 'disarmed': { + 'trigger_time': 5 + }, + 'pending_time': 0, + 'disarm_after_trigger': True + }})) + + entity_id = 'alarm_control_panel.test' + + self.assertEqual(STATE_ALARM_DISARMED, + 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.' + 'dt_util.utcnow'), return_value=future): + fire_time_changed(self.hass, future) + self.hass.block_till_done() + + self.assertEqual(STATE_ALARM_DISARMED, + self.hass.states.get(entity_id).state) + def test_trigger_with_no_disarm_after_trigger(self): """Test disarm after trigger.""" self.assertTrue(setup_component( @@ -684,6 +1096,45 @@ class TestAlarmControlPanelManual(unittest.TestCase): self.assertEqual(STATE_ALARM_TRIGGERED, self.hass.states.get(entity_id).state) + def test_disarm_with_template_code(self): + """Attempt to disarm with a valid or invalid template-based code.""" + self.assertTrue(setup_component( + self.hass, alarm_control_panel.DOMAIN, + {'alarm_control_panel': { + 'platform': 'manual', + 'name': 'test', + 'code_template': + '{{ "" if from_state == "disarmed" else "abc" }}', + 'pending_time': 0, + 'disarm_after_trigger': False + }})) + + entity_id = 'alarm_control_panel.test' + + self.hass.start() + self.hass.block_till_done() + + self.assertEqual(STATE_ALARM_DISARMED, + self.hass.states.get(entity_id).state) + + alarm_control_panel.alarm_arm_home(self.hass, 'def') + self.hass.block_till_done() + + state = self.hass.states.get(entity_id) + self.assertEqual(STATE_ALARM_ARMED_HOME, state.state) + + alarm_control_panel.alarm_disarm(self.hass, 'def') + self.hass.block_till_done() + + state = self.hass.states.get(entity_id) + self.assertEqual(STATE_ALARM_ARMED_HOME, state.state) + + alarm_control_panel.alarm_disarm(self.hass, 'abc') + self.hass.block_till_done() + + state = self.hass.states.get(entity_id) + self.assertEqual(STATE_ALARM_DISARMED, state.state) + def test_arm_custom_bypass_no_pending(self): """Test arm custom bypass method.""" self.assertTrue(setup_component( @@ -795,3 +1246,75 @@ class TestAlarmControlPanelManual(unittest.TestCase): self.assertEqual(STATE_ALARM_ARMED_CUSTOM_BYPASS, self.hass.states.get(entity_id).state) + + def test_arm_away_after_disabled_disarmed(self): + """Test pending state with and without zero trigger time.""" + self.assertTrue(setup_component( + self.hass, alarm_control_panel.DOMAIN, + {'alarm_control_panel': { + 'platform': 'manual', + 'name': 'test', + 'code': CODE, + 'pending_time': 0, + 'delay_time': 1, + 'armed_away': { + 'pending_time': 1, + }, + 'disarmed': { + 'trigger_time': 0 + }, + 'disarm_after_trigger': False + }})) + + 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) + self.hass.block_till_done() + + state = self.hass.states.get(entity_id) + self.assertEqual(STATE_ALARM_PENDING, state.state) + self.assertEqual(STATE_ALARM_DISARMED, + state.attributes['pre_pending_state']) + self.assertEqual(STATE_ALARM_ARMED_AWAY, + state.attributes['post_pending_state']) + + alarm_control_panel.alarm_trigger(self.hass, entity_id=entity_id) + self.hass.block_till_done() + + state = self.hass.states.get(entity_id) + self.assertEqual(STATE_ALARM_PENDING, state.state) + self.assertEqual(STATE_ALARM_DISARMED, + state.attributes['pre_pending_state']) + self.assertEqual(STATE_ALARM_ARMED_AWAY, + state.attributes['post_pending_state']) + + future = dt_util.utcnow() + timedelta(seconds=1) + with patch(('homeassistant.components.alarm_control_panel.manual.' + 'dt_util.utcnow'), return_value=future): + fire_time_changed(self.hass, future) + self.hass.block_till_done() + + state = self.hass.states.get(entity_id) + self.assertEqual(STATE_ALARM_ARMED_AWAY, state.state) + + alarm_control_panel.alarm_trigger(self.hass, entity_id=entity_id) + self.hass.block_till_done() + + state = self.hass.states.get(entity_id) + self.assertEqual(STATE_ALARM_PENDING, state.state) + self.assertEqual(STATE_ALARM_ARMED_AWAY, + state.attributes['pre_pending_state']) + self.assertEqual(STATE_ALARM_TRIGGERED, + state.attributes['post_pending_state']) + + future += timedelta(seconds=1) + with patch(('homeassistant.components.alarm_control_panel.manual.' + 'dt_util.utcnow'), return_value=future): + fire_time_changed(self.hass, future) + self.hass.block_till_done() + + state = self.hass.states.get(entity_id) + self.assertEqual(STATE_ALARM_TRIGGERED, state.state) diff --git a/tests/components/alarm_control_panel/test_manual_mqtt.py b/tests/components/alarm_control_panel/test_manual_mqtt.py index e56b6865e6e..83254d9104f 100644 --- a/tests/components/alarm_control_panel/test_manual_mqtt.py +++ b/tests/components/alarm_control_panel/test_manual_mqtt.py @@ -162,6 +162,34 @@ class TestAlarmControlPanelManualMqtt(unittest.TestCase): self.assertEqual(STATE_ALARM_ARMED_AWAY, self.hass.states.get(entity_id).state) + def test_arm_home_with_template_code(self): + """Attempt to arm with a template-based code.""" + self.assertTrue(setup_component( + self.hass, alarm_control_panel.DOMAIN, + {'alarm_control_panel': { + 'platform': 'manual_mqtt', + 'name': 'test', + 'code_template': '{{ "abc" }}', + 'pending_time': 0, + 'disarm_after_trigger': False, + 'command_topic': 'alarm/command', + 'state_topic': 'alarm/state', + }})) + + entity_id = 'alarm_control_panel.test' + + self.hass.start() + self.hass.block_till_done() + + self.assertEqual(STATE_ALARM_DISARMED, + self.hass.states.get(entity_id).state) + + alarm_control_panel.alarm_arm_home(self.hass, 'abc') + self.hass.block_till_done() + + state = self.hass.states.get(entity_id) + self.assertEqual(STATE_ALARM_ARMED_HOME, state.state) + def test_arm_away_with_pending(self): """Test arm home method.""" self.assertTrue(setup_component( @@ -287,6 +315,13 @@ class TestAlarmControlPanelManualMqtt(unittest.TestCase): self.assertEqual(STATE_ALARM_ARMED_NIGHT, self.hass.states.get(entity_id).state) + # Do not go to the pending state when updating to the same 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_invalid_code(self): """Attempt to arm night without a valid code.""" self.assertTrue(setup_component( @@ -345,6 +380,99 @@ class TestAlarmControlPanelManualMqtt(unittest.TestCase): self.assertEqual(STATE_ALARM_TRIGGERED, self.hass.states.get(entity_id).state) + def test_trigger_with_delay(self): + """Test trigger method and switch from pending to triggered.""" + self.assertTrue(setup_component( + self.hass, alarm_control_panel.DOMAIN, + {'alarm_control_panel': { + 'platform': 'manual_mqtt', + 'name': 'test', + 'code': CODE, + 'delay_time': 1, + '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) + 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() + + state = self.hass.states.get(entity_id) + self.assertEqual(STATE_ALARM_PENDING, state.state) + self.assertEqual(STATE_ALARM_TRIGGERED, + state.attributes['post_pending_state']) + + 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() + + state = self.hass.states.get(entity_id) + self.assertEqual(STATE_ALARM_TRIGGERED, state.state) + + def test_trigger_zero_trigger_time(self): + """Test disabled trigger.""" + self.assertTrue(setup_component( + self.hass, alarm_control_panel.DOMAIN, + {'alarm_control_panel': { + 'platform': 'manual_mqtt', + 'name': 'test', + 'pending_time': 0, + 'trigger_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_trigger(self.hass) + self.hass.block_till_done() + + self.assertEqual(STATE_ALARM_DISARMED, + self.hass.states.get(entity_id).state) + + def test_trigger_zero_trigger_time_with_pending(self): + """Test disabled trigger.""" + self.assertTrue(setup_component( + self.hass, alarm_control_panel.DOMAIN, + {'alarm_control_panel': { + 'platform': 'manual_mqtt', + 'name': 'test', + 'pending_time': 2, + 'trigger_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_trigger(self.hass) + self.hass.block_till_done() + + self.assertEqual(STATE_ALARM_DISARMED, + self.hass.states.get(entity_id).state) + def test_trigger_with_pending(self): """Test arm home method.""" self.assertTrue(setup_component( @@ -425,6 +553,107 @@ class TestAlarmControlPanelManualMqtt(unittest.TestCase): self.assertEqual(STATE_ALARM_DISARMED, self.hass.states.get(entity_id).state) + def test_trigger_with_zero_specific_trigger_time(self): + """Test trigger method.""" + self.assertTrue(setup_component( + self.hass, alarm_control_panel.DOMAIN, + {'alarm_control_panel': { + 'platform': 'manual_mqtt', + 'name': 'test', + 'trigger_time': 5, + 'disarmed': { + 'trigger_time': 0 + }, + 'pending_time': 0, + 'disarm_after_trigger': True, + '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_trigger(self.hass, entity_id=entity_id) + self.hass.block_till_done() + + self.assertEqual(STATE_ALARM_DISARMED, + self.hass.states.get(entity_id).state) + + def test_trigger_with_unused_zero_specific_trigger_time(self): + """Test disarm after trigger.""" + self.assertTrue(setup_component( + self.hass, alarm_control_panel.DOMAIN, + {'alarm_control_panel': { + 'platform': 'manual_mqtt', + 'name': 'test', + 'trigger_time': 5, + 'armed_home': { + 'trigger_time': 0 + }, + 'pending_time': 0, + 'disarm_after_trigger': True, + '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_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_DISARMED, + self.hass.states.get(entity_id).state) + + def test_trigger_with_specific_trigger_time(self): + """Test disarm after trigger.""" + self.assertTrue(setup_component( + self.hass, alarm_control_panel.DOMAIN, + {'alarm_control_panel': { + 'platform': 'manual_mqtt', + 'name': 'test', + 'disarmed': { + 'trigger_time': 5 + }, + 'pending_time': 0, + 'disarm_after_trigger': True, + '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_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_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( @@ -559,6 +788,211 @@ class TestAlarmControlPanelManualMqtt(unittest.TestCase): self.assertEqual(STATE_ALARM_TRIGGERED, self.hass.states.get(entity_id).state) + def test_trigger_with_unused_specific_delay(self): + """Test trigger method and switch from pending to triggered.""" + self.assertTrue(setup_component( + self.hass, alarm_control_panel.DOMAIN, + {'alarm_control_panel': { + 'platform': 'manual_mqtt', + 'name': 'test', + 'code': CODE, + 'delay_time': 5, + 'pending_time': 0, + 'armed_home': { + 'delay_time': 10 + }, + '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) + 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() + + state = self.hass.states.get(entity_id) + self.assertEqual(STATE_ALARM_PENDING, state.state) + self.assertEqual(STATE_ALARM_TRIGGERED, + state.attributes['post_pending_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() + + state = self.hass.states.get(entity_id) + assert state.state == STATE_ALARM_TRIGGERED + + def test_trigger_with_specific_delay(self): + """Test trigger method and switch from pending to triggered.""" + self.assertTrue(setup_component( + self.hass, alarm_control_panel.DOMAIN, + {'alarm_control_panel': { + 'platform': 'manual_mqtt', + 'name': 'test', + 'code': CODE, + 'delay_time': 10, + 'pending_time': 0, + 'armed_away': { + 'delay_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_away(self.hass, CODE) + 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() + + state = self.hass.states.get(entity_id) + self.assertEqual(STATE_ALARM_PENDING, state.state) + self.assertEqual(STATE_ALARM_TRIGGERED, + state.attributes['post_pending_state']) + + 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() + + state = self.hass.states.get(entity_id) + assert state.state == STATE_ALARM_TRIGGERED + + def test_trigger_with_pending_and_delay(self): + """Test trigger method and switch from pending to triggered.""" + self.assertTrue(setup_component( + self.hass, alarm_control_panel.DOMAIN, + {'alarm_control_panel': { + 'platform': 'manual_mqtt', + 'name': 'test', + 'code': CODE, + 'delay_time': 1, + 'pending_time': 0, + 'triggered': { + '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_away(self.hass, CODE) + 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() + + state = self.hass.states.get(entity_id) + assert state.state == STATE_ALARM_PENDING + assert state.attributes['post_pending_state'] == STATE_ALARM_TRIGGERED + + 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() + + state = self.hass.states.get(entity_id) + assert state.state == STATE_ALARM_PENDING + assert state.attributes['post_pending_state'] == STATE_ALARM_TRIGGERED + + future += 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() + + state = self.hass.states.get(entity_id) + assert state.state == STATE_ALARM_TRIGGERED + + def test_trigger_with_pending_and_specific_delay(self): + """Test trigger method and switch from pending to triggered.""" + self.assertTrue(setup_component( + self.hass, alarm_control_panel.DOMAIN, + {'alarm_control_panel': { + 'platform': 'manual_mqtt', + 'name': 'test', + 'code': CODE, + 'delay_time': 10, + 'pending_time': 0, + 'armed_away': { + 'delay_time': 1 + }, + 'triggered': { + '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_away(self.hass, CODE) + 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() + + state = self.hass.states.get(entity_id) + assert state.state == STATE_ALARM_PENDING + assert state.attributes['post_pending_state'] == STATE_ALARM_TRIGGERED + + 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() + + state = self.hass.states.get(entity_id) + assert state.state == STATE_ALARM_PENDING + assert state.attributes['post_pending_state'] == STATE_ALARM_TRIGGERED + + future += 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() + + state = self.hass.states.get(entity_id) + assert state.state == STATE_ALARM_TRIGGERED + def test_armed_home_with_specific_pending(self): """Test arm home method.""" self.assertTrue(setup_component( @@ -674,21 +1108,6 @@ class TestAlarmControlPanelManualMqtt(unittest.TestCase): 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() @@ -710,9 +1129,124 @@ class TestAlarmControlPanelManualMqtt(unittest.TestCase): fire_time_changed(self.hass, future) self.hass.block_till_done() - self.assertEqual(STATE_ALARM_ARMED_HOME, + self.assertEqual(STATE_ALARM_DISARMED, self.hass.states.get(entity_id).state) + def test_arm_away_after_disabled_disarmed(self): + """Test pending state with and without zero trigger time.""" + self.assertTrue(setup_component( + self.hass, alarm_control_panel.DOMAIN, + {'alarm_control_panel': { + 'platform': 'manual_mqtt', + 'name': 'test', + 'code': CODE, + 'pending_time': 0, + 'delay_time': 1, + 'armed_away': { + 'pending_time': 1, + }, + 'disarmed': { + 'trigger_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) + self.hass.block_till_done() + + state = self.hass.states.get(entity_id) + self.assertEqual(STATE_ALARM_PENDING, state.state) + self.assertEqual(STATE_ALARM_DISARMED, + state.attributes['pre_pending_state']) + self.assertEqual(STATE_ALARM_ARMED_AWAY, + state.attributes['post_pending_state']) + + alarm_control_panel.alarm_trigger(self.hass, entity_id=entity_id) + self.hass.block_till_done() + + state = self.hass.states.get(entity_id) + self.assertEqual(STATE_ALARM_PENDING, state.state) + self.assertEqual(STATE_ALARM_DISARMED, + state.attributes['pre_pending_state']) + self.assertEqual(STATE_ALARM_ARMED_AWAY, + state.attributes['post_pending_state']) + + 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() + + state = self.hass.states.get(entity_id) + self.assertEqual(STATE_ALARM_ARMED_AWAY, state.state) + + alarm_control_panel.alarm_trigger(self.hass, entity_id=entity_id) + self.hass.block_till_done() + + state = self.hass.states.get(entity_id) + self.assertEqual(STATE_ALARM_PENDING, state.state) + self.assertEqual(STATE_ALARM_ARMED_AWAY, + state.attributes['pre_pending_state']) + self.assertEqual(STATE_ALARM_TRIGGERED, + state.attributes['post_pending_state']) + + future += 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() + + state = self.hass.states.get(entity_id) + self.assertEqual(STATE_ALARM_TRIGGERED, state.state) + + def test_disarm_with_template_code(self): + """Attempt to disarm with a valid or invalid template-based code.""" + self.assertTrue(setup_component( + self.hass, alarm_control_panel.DOMAIN, + {'alarm_control_panel': { + 'platform': 'manual_mqtt', + 'name': 'test', + 'code_template': + '{{ "" if from_state == "disarmed" else "abc" }}', + 'pending_time': 0, + 'disarm_after_trigger': False, + 'command_topic': 'alarm/command', + 'state_topic': 'alarm/state', + }})) + + entity_id = 'alarm_control_panel.test' + + self.hass.start() + self.hass.block_till_done() + + self.assertEqual(STATE_ALARM_DISARMED, + self.hass.states.get(entity_id).state) + + alarm_control_panel.alarm_arm_home(self.hass, 'def') + self.hass.block_till_done() + + state = self.hass.states.get(entity_id) + self.assertEqual(STATE_ALARM_ARMED_HOME, state.state) + + alarm_control_panel.alarm_disarm(self.hass, 'def') + self.hass.block_till_done() + + state = self.hass.states.get(entity_id) + self.assertEqual(STATE_ALARM_ARMED_HOME, state.state) + + alarm_control_panel.alarm_disarm(self.hass, 'abc') + self.hass.block_till_done() + + state = self.hass.states.get(entity_id) + self.assertEqual(STATE_ALARM_DISARMED, state.state) + def test_arm_home_via_command_topic(self): """Test arming home via command topic.""" assert setup_component(self.hass, alarm_control_panel.DOMAIN, {