mirror of
https://github.com/home-assistant/core
synced 2024-10-08 14:48:22 +00:00
Manual alarm with MQTT control (#8257)
* Manual alarm with MQTT control * Duplicate manual control panel code instead of extending it * Duplicate manual alarm test as well; modify for manual_mqtt * Add MQTT-specific tests for manual_mqtt alarm
This commit is contained in:
Colin O'Dell
Paulus Schoutsen
parent
ecc1429453
commit
cc03f7ee6a
|
|
@ -211,6 +211,7 @@ omit =
|
|||
|
||||
homeassistant/components/alarm_control_panel/alarmdotcom.py
|
||||
homeassistant/components/alarm_control_panel/concord232.py
|
||||
homeassistant/components/alarm_control_panel/manual_mqtt.py
|
||||
homeassistant/components/alarm_control_panel/nx584.py
|
||||
homeassistant/components/alarm_control_panel/simplisafe.py
|
||||
homeassistant/components/alarm_control_panel/totalconnect.py
|
||||
|
|
|
|||
235
homeassistant/components/alarm_control_panel/manual_mqtt.py
Normal file
235
homeassistant/components/alarm_control_panel/manual_mqtt.py
Normal file
|
|
@ -0,0 +1,235 @@
|
|||
"""
|
||||
Support for manual alarms controllable via MQTT.
|
||||
|
||||
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 datetime
|
||||
import logging
|
||||
|
||||
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,
|
||||
CONF_DISARM_AFTER_TRIGGER)
|
||||
import homeassistant.components.mqtt as mqtt
|
||||
|
||||
from homeassistant.helpers.event import async_track_state_change
|
||||
from homeassistant.core import callback
|
||||
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
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'
|
||||
|
||||
DEFAULT_ALARM_NAME = 'HA Alarm'
|
||||
DEFAULT_PENDING_TIME = 60
|
||||
DEFAULT_TRIGGER_TIME = 120
|
||||
DEFAULT_DISARM_AFTER_TRIGGER = False
|
||||
DEFAULT_ARM_AWAY = 'ARM_AWAY'
|
||||
DEFAULT_ARM_HOME = 'ARM_HOME'
|
||||
DEFAULT_DISARM = 'DISARM'
|
||||
|
||||
DEPENDENCIES = ['mqtt']
|
||||
|
||||
PLATFORM_SCHEMA = 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.Optional(CONF_PENDING_TIME, default=DEFAULT_PENDING_TIME):
|
||||
vol.All(vol.Coerce(int), vol.Range(min=0)),
|
||||
vol.Optional(CONF_TRIGGER_TIME, default=DEFAULT_TRIGGER_TIME):
|
||||
vol.All(vol.Coerce(int), vol.Range(min=1)),
|
||||
vol.Optional(CONF_DISARM_AFTER_TRIGGER,
|
||||
default=DEFAULT_DISARM_AFTER_TRIGGER): cv.boolean,
|
||||
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_DISARM, default=DEFAULT_DISARM): cv.string,
|
||||
})
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def setup_platform(hass, config, add_devices, discovery_info=None):
|
||||
"""Set up the manual MQTT alarm platform."""
|
||||
add_devices([ManualMQTTAlarm(
|
||||
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_DISARM_AFTER_TRIGGER, DEFAULT_DISARM_AFTER_TRIGGER),
|
||||
config.get(mqtt.CONF_STATE_TOPIC),
|
||||
config.get(mqtt.CONF_COMMAND_TOPIC),
|
||||
config.get(mqtt.CONF_QOS),
|
||||
config.get(CONF_PAYLOAD_DISARM),
|
||||
config.get(CONF_PAYLOAD_ARM_HOME),
|
||||
config.get(CONF_PAYLOAD_ARM_AWAY))])
|
||||
|
||||
|
||||
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.
|
||||
"""
|
||||
|
||||
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):
|
||||
"""Init the manual MQTT alarm panel."""
|
||||
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)
|
||||
self._disarm_after_trigger = disarm_after_trigger
|
||||
self._pre_trigger_state = self._state
|
||||
self._state_ts = None
|
||||
|
||||
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
|
||||
|
||||
@property
|
||||
def should_poll(self):
|
||||
"""Return the polling state."""
|
||||
return False
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
"""Return the name of the device."""
|
||||
return self._name
|
||||
|
||||
@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():
|
||||
return STATE_ALARM_PENDING
|
||||
elif (self._state_ts + self._pending_time +
|
||||
self._trigger_time) < dt_util.utcnow():
|
||||
if self._disarm_after_trigger:
|
||||
return STATE_ALARM_DISARMED
|
||||
return self._pre_trigger_state
|
||||
|
||||
return self._state
|
||||
|
||||
@property
|
||||
def code_format(self):
|
||||
"""One or more characters."""
|
||||
return None if self._code is None else '.+'
|
||||
|
||||
def alarm_disarm(self, code=None):
|
||||
"""Send disarm command."""
|
||||
if not self._validate_code(code, STATE_ALARM_DISARMED):
|
||||
return
|
||||
|
||||
self._state = STATE_ALARM_DISARMED
|
||||
self._state_ts = dt_util.utcnow()
|
||||
self.schedule_update_ha_state()
|
||||
|
||||
def alarm_arm_home(self, code=None):
|
||||
"""Send arm home command."""
|
||||
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)
|
||||
|
||||
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()
|
||||
|
||||
if self._pending_time:
|
||||
track_point_in_time(
|
||||
self._hass, self.async_update_ha_state,
|
||||
self._state_ts + self._pending_time)
|
||||
|
||||
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._state_ts = dt_util.utcnow()
|
||||
self.schedule_update_ha_state()
|
||||
|
||||
if self._trigger_time:
|
||||
track_point_in_time(
|
||||
self._hass, self.async_update_ha_state,
|
||||
self._state_ts + self._pending_time)
|
||||
|
||||
track_point_in_time(
|
||||
self._hass, self.async_update_ha_state,
|
||||
self._state_ts + self._pending_time + self._trigger_time)
|
||||
|
||||
def _validate_code(self, code, state):
|
||||
"""Validate given code."""
|
||||
check = self._code is None or code == self._code
|
||||
if not check:
|
||||
_LOGGER.warning("Invalid code given for %s", state)
|
||||
return check
|
||||
|
||||
def async_added_to_hass(self):
|
||||
"""Subscribe mqtt events.
|
||||
|
||||
This method must be run in the event loop and returns a coroutine.
|
||||
"""
|
||||
async_track_state_change(
|
||||
self.hass, self.entity_id, self._async_state_changed_listener
|
||||
)
|
||||
|
||||
@callback
|
||||
def message_received(topic, payload, qos):
|
||||
"""Run when new MQTT message has been received."""
|
||||
if payload == self._payload_disarm:
|
||||
self.async_alarm_disarm(self._code)
|
||||
elif payload == self._payload_arm_home:
|
||||
self.async_alarm_arm_home(self._code)
|
||||
elif payload == self._payload_arm_away:
|
||||
self.async_alarm_arm_away(self._code)
|
||||
else:
|
||||
_LOGGER.warning("Received unexpected payload: %s", payload)
|
||||
return
|
||||
|
||||
return mqtt.async_subscribe(
|
||||
self.hass, self._command_topic, message_received, self._qos)
|
||||
|
||||
@asyncio.coroutine
|
||||
def _async_state_changed_listener(self, entity_id, old_state, new_state):
|
||||
"""Publish state change to MQTT."""
|
||||
mqtt.async_publish(self.hass, self._state_topic, new_state.state,
|
||||
self._qos, True)
|
||||
559
tests/components/alarm_control_panel/test_manual_mqtt.py
Normal file
559
tests/components/alarm_control_panel/test_manual_mqtt.py
Normal file
|
|
@ -0,0 +1,559 @@
|
|||
"""The tests for the manual_mqtt Alarm Control Panel component."""
|
||||
from datetime import timedelta
|
||||
import unittest
|
||||
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)
|
||||
from homeassistant.components import alarm_control_panel
|
||||
import homeassistant.util.dt as dt_util
|
||||
|
||||
from tests.common import (
|
||||
fire_time_changed, get_test_home_assistant,
|
||||
mock_mqtt_component, fire_mqtt_message, assert_setup_component)
|
||||
|
||||
CODE = 'HELLO_CODE'
|
||||
|
||||
|
||||
class TestAlarmControlPanelManualMqtt(unittest.TestCase):
|
||||
"""Test the manual_mqtt alarm module."""
|
||||
|
||||
def setUp(self): # pylint: disable=invalid-name
|
||||
"""Setup things to be run when tests are started."""
|
||||
self.hass = get_test_home_assistant()
|
||||
self.mock_publish = mock_mqtt_component(self.hass)
|
||||
|
||||
def tearDown(self): # pylint: disable=invalid-name
|
||||
"""Stop down everything that was started."""
|
||||
self.hass.stop()
|
||||
|
||||
def test_fail_setup_without_state_topic(self):
|
||||
"""Test for failing with no state topic."""
|
||||
with assert_setup_component(0) as config:
|
||||
assert setup_component(self.hass, alarm_control_panel.DOMAIN, {
|
||||
alarm_control_panel.DOMAIN: {
|
||||
'platform': 'mqtt_alarm',
|
||||
'command_topic': 'alarm/command'
|
||||
}
|
||||
})
|
||||
assert not config[alarm_control_panel.DOMAIN]
|
||||
|
||||
def test_fail_setup_without_command_topic(self):
|
||||
"""Test failing with no command topic."""
|
||||
with assert_setup_component(0):
|
||||
assert setup_component(self.hass, alarm_control_panel.DOMAIN, {
|
||||
alarm_control_panel.DOMAIN: {
|
||||
'platform': 'mqtt_alarm',
|
||||
'state_topic': 'alarm/state'
|
||||
}
|
||||
})
|
||||
|
||||
def test_arm_home_no_pending(self):
|
||||
"""Test arm home 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_home(self.hass, CODE)
|
||||
self.hass.block_till_done()
|
||||
|
||||
self.assertEqual(STATE_ALARM_ARMED_HOME,
|
||||
self.hass.states.get(entity_id).state)
|
||||
|
||||
def test_arm_home_with_pending(self):
|
||||
"""Test arm home 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_home(self.hass, CODE, entity_id)
|
||||
self.hass.block_till_done()
|
||||
|
||||
self.assertEqual(STATE_ALARM_PENDING,
|
||||
self.hass.states.get(entity_id).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()
|
||||
|
||||
self.assertEqual(STATE_ALARM_ARMED_HOME,
|
||||
self.hass.states.get(entity_id).state)
|
||||
|
||||
def test_arm_home_with_invalid_code(self):
|
||||
"""Attempt to arm home 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_home(self.hass, CODE + '2')
|
||||
self.hass.block_till_done()
|
||||
|
||||
self.assertEqual(STATE_ALARM_DISARMED,
|
||||
self.hass.states.get(entity_id).state)
|
||||
|
||||
def test_arm_away_no_pending(self):
|
||||
"""Test arm home 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_away(self.hass, CODE, entity_id)
|
||||
self.hass.block_till_done()
|
||||
|
||||
self.assertEqual(STATE_ALARM_ARMED_AWAY,
|
||||
self.hass.states.get(entity_id).state)
|
||||
|
||||
def test_arm_away_with_pending(self):
|
||||
"""Test arm home 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_away(self.hass, CODE)
|
||||
self.hass.block_till_done()
|
||||
|
||||
self.assertEqual(STATE_ALARM_PENDING,
|
||||
self.hass.states.get(entity_id).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()
|
||||
|
||||
self.assertEqual(STATE_ALARM_ARMED_AWAY,
|
||||
self.hass.states.get(entity_id).state)
|
||||
|
||||
def test_arm_away_with_invalid_code(self):
|
||||
"""Attempt to arm away 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_away(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(
|
||||
self.hass, alarm_control_panel.DOMAIN,
|
||||
{'alarm_control_panel': {
|
||||
'platform': 'manual_mqtt',
|
||||
'name': 'test',
|
||||
'trigger_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_trigger(self.hass, entity_id=entity_id)
|
||||
self.hass.block_till_done()
|
||||
|
||||
self.assertEqual(STATE_ALARM_PENDING,
|
||||
self.hass.states.get(entity_id).state)
|
||||
|
||||
future = dt_util.utcnow() + timedelta(seconds=60)
|
||||
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)
|
||||
|
||||
def test_trigger_with_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': 2,
|
||||
'trigger_time': 3,
|
||||
'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_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_DISARMED,
|
||||
self.hass.states.get(entity_id).state)
|
||||
|
||||
def test_trigger_with_disarm_after_trigger(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,
|
||||
'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_disarm_while_pending_trigger(self):
|
||||
"""Test disarming while pending state."""
|
||||
self.assertTrue(setup_component(
|
||||
self.hass, alarm_control_panel.DOMAIN,
|
||||
{'alarm_control_panel': {
|
||||
'platform': 'manual_mqtt',
|
||||
'name': 'test',
|
||||
'trigger_time': 5,
|
||||
'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_PENDING,
|
||||
self.hass.states.get(entity_id).state)
|
||||
|
||||
alarm_control_panel.alarm_disarm(self.hass, entity_id=entity_id)
|
||||
self.hass.block_till_done()
|
||||
|
||||
self.assertEqual(STATE_ALARM_DISARMED,
|
||||
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_disarm_during_trigger_with_invalid_code(self):
|
||||
"""Test disarming while code is invalid."""
|
||||
self.assertTrue(setup_component(
|
||||
self.hass, alarm_control_panel.DOMAIN,
|
||||
{'alarm_control_panel': {
|
||||
'platform': 'manual_mqtt',
|
||||
'name': 'test',
|
||||
'pending_time': 5,
|
||||
'code': CODE + '2',
|
||||
'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_PENDING,
|
||||
self.hass.states.get(entity_id).state)
|
||||
|
||||
alarm_control_panel.alarm_disarm(self.hass, entity_id=entity_id)
|
||||
self.hass.block_till_done()
|
||||
|
||||
self.assertEqual(STATE_ALARM_PENDING,
|
||||
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_TRIGGERED,
|
||||
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, {
|
||||
alarm_control_panel.DOMAIN: {
|
||||
'platform': 'manual_mqtt',
|
||||
'name': 'test',
|
||||
'pending_time': 1,
|
||||
'state_topic': 'alarm/state',
|
||||
'command_topic': 'alarm/command',
|
||||
'payload_arm_home': 'ARM_HOME',
|
||||
}
|
||||
})
|
||||
|
||||
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_HOME')
|
||||
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_HOME,
|
||||
self.hass.states.get(entity_id).state)
|
||||
|
||||
def test_arm_away_via_command_topic(self):
|
||||
"""Test arming away 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_away': 'ARM_AWAY',
|
||||
}
|
||||
})
|
||||
|
||||
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_AWAY')
|
||||
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_AWAY,
|
||||
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, {
|
||||
alarm_control_panel.DOMAIN: {
|
||||
'platform': 'manual_mqtt',
|
||||
'name': 'test',
|
||||
'pending_time': 1,
|
||||
'state_topic': 'alarm/state',
|
||||
'command_topic': 'alarm/command',
|
||||
'payload_disarm': 'DISARM',
|
||||
}
|
||||
})
|
||||
|
||||
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_PENDING,
|
||||
self.hass.states.get(entity_id).state)
|
||||
|
||||
# Now that we're pending, receive a command to disarm
|
||||
fire_mqtt_message(self.hass, 'alarm/command', 'DISARM')
|
||||
self.hass.block_till_done()
|
||||
|
||||
self.assertEqual(STATE_ALARM_DISARMED,
|
||||
self.hass.states.get(entity_id).state)
|
||||
|
||||
def test_state_changes_are_published_to_mqtt(self):
|
||||
"""Test publishing of MQTT messages when state changes."""
|
||||
assert setup_component(self.hass, alarm_control_panel.DOMAIN, {
|
||||
alarm_control_panel.DOMAIN: {
|
||||
'platform': 'manual_mqtt',
|
||||
'name': 'test',
|
||||
'pending_time': 1,
|
||||
'trigger_time': 1,
|
||||
'state_topic': 'alarm/state',
|
||||
'command_topic': 'alarm/command',
|
||||
}
|
||||
})
|
||||
|
||||
# Component should send disarmed alarm state on startup
|
||||
self.hass.block_till_done()
|
||||
self.assertEqual(('alarm/state', STATE_ALARM_DISARMED, 0, True),
|
||||
self.mock_publish.mock_calls[-2][1])
|
||||
|
||||
# Arm in home mode
|
||||
alarm_control_panel.alarm_arm_home(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_HOME, 0, True),
|
||||
self.mock_publish.mock_calls[-2][1])
|
||||
|
||||
# Arm in away mode
|
||||
alarm_control_panel.alarm_arm_away(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_AWAY, 0, True),
|
||||
self.mock_publish.mock_calls[-2][1])
|
||||
|
||||
# Disarm
|
||||
alarm_control_panel.alarm_disarm(self.hass)
|
||||
self.hass.block_till_done()
|
||||
self.assertEqual(('alarm/state', STATE_ALARM_DISARMED, 0, True),
|
||||
self.mock_publish.mock_calls[-2][1])
|
||||
Loading…
Reference in a new issue