evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add tests for MQTT alarm

Browse source
This commit is contained in:
Paulus Schoutsen 2015-10-13 23:08:12 -07:00
parent d37b70556d
commit 716376081d
4 changed files with 209 additions and 22 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

45
homeassistant/components/alarm_control_panel/mqtt.py
View file

@ -10,7 +10,9 @@ import logging
import homeassistant.components.mqtt as mqtt
import homeassistant.components.alarm_control_panel as alarm
from homeassistant.const import (STATE_UNKNOWN)
from homeassistant.const import (
STATE_ALARM_DISARMED, STATE_ALARM_ARMED_HOME, STATE_ALARM_ARMED_AWAY,
STATE_ALARM_PENDING, STATE_ALARM_TRIGGERED, STATE_UNKNOWN)
_LOGGER = logging.getLogger(__name__)
@ -62,10 +64,15 @@ class MqttAlarm(alarm.AlarmControlPanel):
self._payload_disarm = payload_disarm
self._payload_arm_home = payload_arm_home
self._payload_arm_away = payload_arm_away
self._code = code
self._code = str(code) if code else None
def message_received(topic, payload, qos):
""" A new MQTT message has been received. """
if payload not in (STATE_ALARM_DISARMED, STATE_ALARM_ARMED_HOME,
STATE_ALARM_ARMED_AWAY, STATE_ALARM_PENDING,
STATE_ALARM_TRIGGERED):
_LOGGER.warning('Received unexpected payload: %s', payload)
return
self._state = payload
self.update_ha_state()
@ -93,24 +100,28 @@ class MqttAlarm(alarm.AlarmControlPanel):
def alarm_disarm(self, code=None):
""" Send disarm command. """
if code == str(self._code) or self.code_format is None:
mqtt.publish(self.hass, self._command_topic,
self._payload_disarm, self._qos)
else:
_LOGGER.warning("Wrong code entered while disarming!")
if not self._validate_code(code, 'disarming'):
return
mqtt.publish(self.hass, self._command_topic,
self._payload_disarm, self._qos)
def alarm_arm_home(self, code=None):
""" Send arm home command. """
if code == str(self._code) or self.code_format is None:
mqtt.publish(self.hass, self._command_topic,
self._payload_arm_home, self._qos)
else:
_LOGGER.warning("Wrong code entered while arming home!")
if not self._validate_code(code, 'arming home'):
return
mqtt.publish(self.hass, self._command_topic,
self._payload_arm_home, self._qos)
def alarm_arm_away(self, code=None):
""" Send arm away command. """
if code == str(self._code) or self.code_format is None:
mqtt.publish(self.hass, self._command_topic,
self._payload_arm_away, self._qos)
else:
_LOGGER.warning("Wrong code entered while arming away!")
if not self._validate_code(code, 'arming away'):
return
mqtt.publish(self.hass, self._command_topic,
self._payload_arm_away, self._qos)
def _validate_code(self, code, state):
""" Validate given code. """
check = self._code is None or code == self._code
if not check:
_LOGGER.warning('Wrong code entered for %s', state)
return check

6
tests/common.py
View file

@ -125,16 +125,14 @@ def mock_http_component(hass):
@mock.patch('homeassistant.components.mqtt.MQTT')
@mock.patch('homeassistant.components.mqtt.MQTT.publish')
def mock_mqtt_component(hass, mock_mqtt, mock_mqtt_publish):
def mock_mqtt_component(hass, mock_mqtt):
mqtt.setup(hass, {
mqtt.DOMAIN: {
mqtt.CONF_BROKER: 'mock-broker',
}
})
hass.config.components.append(mqtt.DOMAIN)
return mock_mqtt_publish
return mock_mqtt
class MockHTTP(object):

2
tests/components/alarm_control_panel/test_manual.py
View file

@ -21,7 +21,7 @@ CODE = 'HELLO_CODE'
class TestAlarmControlPanelManual(unittest.TestCase):
""" Test the demo module. """
""" Test the manual alarm module. """
def setUp(self): # pylint: disable=invalid-name
self.hass = ha.HomeAssistant()

178
tests/components/alarm_control_panel/test_mqtt.py Normal file
View file

@ -0,0 +1,178 @@
"""
tests.components.alarm_control_panel.test_manual
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Tests manual alarm control panel component.
"""
import unittest
from unittest.mock import patch
import homeassistant.core as ha
from homeassistant.const import (
STATE_ALARM_DISARMED, STATE_ALARM_ARMED_HOME, STATE_ALARM_ARMED_AWAY,
STATE_ALARM_PENDING, STATE_ALARM_TRIGGERED, STATE_UNKNOWN)
from homeassistant.components import alarm_control_panel
from tests.common import mock_mqtt_component, fire_mqtt_message
CODE = 'HELLO_CODE'
class TestAlarmControlPanelMQTT(unittest.TestCase):
""" Test the manual alarm module. """
def setUp(self): # pylint: disable=invalid-name
self.hass = ha.HomeAssistant()
self.mock_publish = mock_mqtt_component(self.hass)
def tearDown(self): # pylint: disable=invalid-name
""" Stop down stuff we started. """
self.hass.stop()
@patch('homeassistant.components.alarm_control_panel.mqtt._LOGGER.error')
def test_fail_setup_without_state_topic(self, mock_error):
self.assertTrue(alarm_control_panel.setup(self.hass, {
'alarm_control_panel': {
'platform': 'mqtt',
'command_topic': 'alarm/command'
}}))
self.assertEqual(1, mock_error.call_count)
@patch('homeassistant.components.alarm_control_panel.mqtt._LOGGER.error')
def test_fail_setup_without_command_topic(self, mock_error):
self.assertTrue(alarm_control_panel.setup(self.hass, {
'alarm_control_panel': {
'platform': 'mqtt',
'state_topic': 'alarm/state'
}}))
self.assertEqual(1, mock_error.call_count)
def test_update_state_via_state_topic(self):
""" Test arm home method. """
self.assertTrue(alarm_control_panel.setup(self.hass, {
'alarm_control_panel': {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'alarm/state',
'command_topic': 'alarm/command',
}}))
entity_id = 'alarm_control_panel.test'
self.assertEqual(STATE_UNKNOWN,
self.hass.states.get(entity_id).state)
for state in (STATE_ALARM_DISARMED, STATE_ALARM_ARMED_HOME,
STATE_ALARM_ARMED_AWAY, STATE_ALARM_PENDING,
STATE_ALARM_TRIGGERED):
fire_mqtt_message(self.hass, 'alarm/state', state)
self.hass.pool.block_till_done()
self.assertEqual(state, self.hass.states.get(entity_id).state)
def test_ignore_update_state_if_unknown_via_state_topic(self):
""" Test arm home method. """
self.assertTrue(alarm_control_panel.setup(self.hass, {
'alarm_control_panel': {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'alarm/state',
'command_topic': 'alarm/command',
}}))
entity_id = 'alarm_control_panel.test'
self.assertEqual(STATE_UNKNOWN,
self.hass.states.get(entity_id).state)
fire_mqtt_message(self.hass, 'alarm/state', 'unsupported state')
self.hass.pool.block_till_done()
self.assertEqual(STATE_UNKNOWN, self.hass.states.get(entity_id).state)
def test_arm_home_publishes_mqtt(self):
self.assertTrue(alarm_control_panel.setup(self.hass, {
'alarm_control_panel': {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'alarm/state',
'command_topic': 'alarm/command',
}}))
alarm_control_panel.alarm_arm_home(self.hass)
self.hass.pool.block_till_done()
self.assertEqual(('alarm/command', 'ARM_HOME', 0),
self.mock_publish.mock_calls[-1][1])
def test_arm_home_not_publishes_mqtt_with_invalid_code(self):
self.assertTrue(alarm_control_panel.setup(self.hass, {
'alarm_control_panel': {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'alarm/state',
'command_topic': 'alarm/command',
'code': '1234'
}}))
call_count = self.mock_publish.call_count
alarm_control_panel.alarm_arm_home(self.hass, 'abcd')
self.hass.pool.block_till_done()
self.assertEqual(call_count, self.mock_publish.call_count)
def test_arm_away_publishes_mqtt(self):
self.assertTrue(alarm_control_panel.setup(self.hass, {
'alarm_control_panel': {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'alarm/state',
'command_topic': 'alarm/command',
}}))
alarm_control_panel.alarm_arm_away(self.hass)
self.hass.pool.block_till_done()
self.assertEqual(('alarm/command', 'ARM_AWAY', 0),
self.mock_publish.mock_calls[-1][1])
def test_arm_away_not_publishes_mqtt_with_invalid_code(self):
self.assertTrue(alarm_control_panel.setup(self.hass, {
'alarm_control_panel': {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'alarm/state',
'command_topic': 'alarm/command',
'code': '1234'
}}))
call_count = self.mock_publish.call_count
alarm_control_panel.alarm_arm_away(self.hass, 'abcd')
self.hass.pool.block_till_done()
self.assertEqual(call_count, self.mock_publish.call_count)
def test_disarm_publishes_mqtt(self):
self.assertTrue(alarm_control_panel.setup(self.hass, {
'alarm_control_panel': {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'alarm/state',
'command_topic': 'alarm/command',
}}))
alarm_control_panel.alarm_disarm(self.hass)
self.hass.pool.block_till_done()
self.assertEqual(('alarm/command', 'DISARM', 0),
self.mock_publish.mock_calls[-1][1])
def test_disarm_not_publishes_mqtt_with_invalid_code(self):
self.assertTrue(alarm_control_panel.setup(self.hass, {
'alarm_control_panel': {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'alarm/state',
'command_topic': 'alarm/command',
'code': '1234'
}}))
call_count = self.mock_publish.call_count
alarm_control_panel.alarm_disarm(self.hass, 'abcd')
self.hass.pool.block_till_done()
self.assertEqual(call_count, self.mock_publish.call_count)