self-hosted/home-assistant-core
self-hosted/home-assistant-core
1
0
Fork 0
mirror of https://github.com/home-assistant/core synced 2024-10-05 15:17:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add automation mqtt tests

Browse source
This commit is contained in:
Paulus Schoutsen 2015-08-10 23:11:46 -07:00
parent eecc51c92d
commit 291cc62381
7 changed files with 228 additions and 34 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
homeassistant/components/automation/__init__.py
View file

@ -25,6 +25,7 @@ _LOGGER = logging.getLogger(__name__)
def setup(hass, config):
""" Sets up automation. """
success = False
for p_type, p_config in config_per_platform(config, DOMAIN, _LOGGER):
platform = prepare_setup_platform(hass, config, DOMAIN, p_type)
@ -36,11 +37,12 @@ def setup(hass, config):
if platform.register(hass, p_config, _get_action(hass, p_config)):
_LOGGER.info(
"Initialized %s rule %s", p_type, p_config.get(CONF_ALIAS, ""))
success = True
else:
_LOGGER.error(
"Error setting up rule %s", p_config.get(CONF_ALIAS, ""))
return True
return success
def _get_action(hass, config):

21
tests/common.py
View file

@ -6,6 +6,7 @@ Helper method for writing tests.
"""
import os
from datetime import timedelta
from unittest import mock
import homeassistant as ha
import homeassistant.util.dt as dt_util
@ -13,7 +14,7 @@ from homeassistant.helpers.entity import ToggleEntity
from homeassistant.const import (
STATE_ON, STATE_OFF, DEVICE_DEFAULT_NAME, EVENT_TIME_CHANGED,
EVENT_STATE_CHANGED)
from homeassistant.components import sun
from homeassistant.components import sun, mqtt
def get_test_config_dir():
@ -52,6 +53,14 @@ def mock_service(hass, domain, service):
return calls
def fire_mqtt_message(hass, topic, payload, qos=0):
hass.bus.fire(mqtt.EVENT_MQTT_MESSAGE_RECEIVED, {
mqtt.ATTR_TOPIC: topic,
mqtt.ATTR_PAYLOAD: payload,
mqtt.ATTR_QOS: qos,
})
def fire_time_changed(hass, time):
hass.bus.fire(EVENT_TIME_CHANGED, {'now': time})
@ -93,6 +102,16 @@ def mock_http_component(hass):
hass.config.components.append('http')
def mock_mqtt_component(hass):
with mock.patch('homeassistant.components.mqtt.MQTT'):
mqtt.setup(hass, {
mqtt.DOMAIN: {
mqtt.CONF_BROKER: 'mock-broker',
}
})
hass.config.components.append(mqtt.DOMAIN)
class MockHTTP(object):
""" Mocks the HTTP module. """

22
tests/components/automation/test_event.py
View file

@ -7,7 +7,6 @@ Tests demo component.
import unittest
import homeassistant as ha
import homeassistant.loader as loader
import homeassistant.components.automation as automation
import homeassistant.components.automation.event as event
from homeassistant.const import CONF_PLATFORM
@ -18,7 +17,6 @@ class TestAutomationEvent(unittest.TestCase):
def setUp(self): # pylint: disable=invalid-name
self.hass = ha.HomeAssistant()
loader.prepare(self.hass)
self.calls = []
def record_call(service):
@ -30,42 +28,50 @@ class TestAutomationEvent(unittest.TestCase):
""" Stop down stuff we started. """
self.hass.stop()
def test_fails_setup_if_no_event_type(self):
self.assertFalse(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'event',
automation.CONF_SERVICE: 'test.automation'
}
}))
def test_if_fires_on_event(self):
automation.setup(self.hass, {
self.assertTrue(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'event',
event.CONF_EVENT_TYPE: 'test_event',
automation.CONF_SERVICE: 'test.automation'
}
})
}))
self.hass.bus.fire('test_event')
self.hass.pool.block_till_done()
self.assertEqual(1, len(self.calls))
def test_if_fires_on_event_with_data(self):
automation.setup(self.hass, {
self.assertTrue(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'event',
event.CONF_EVENT_TYPE: 'test_event',
event.CONF_EVENT_DATA: {'some_attr': 'some_value'},
automation.CONF_SERVICE: 'test.automation'
}
})
}))
self.hass.bus.fire('test_event', {'some_attr': 'some_value'})
self.hass.pool.block_till_done()
self.assertEqual(1, len(self.calls))
def test_if_not_fires_if_event_data_not_matches(self):
automation.setup(self.hass, {
self.assertTrue(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'event',
event.CONF_EVENT_TYPE: 'test_event',
event.CONF_EVENT_DATA: {'some_attr': 'some_value'},
automation.CONF_SERVICE: 'test.automation'
}
})
}))
self.hass.bus.fire('test_event', {'some_attr': 'some_other_value'})
self.hass.pool.block_till_done()

80
tests/components/automation/test_init.py Normal file
View file

@ -0,0 +1,80 @@
"""
tests.test_component_demo
~~~~~~~~~~~~~~~~~~~~~~~~~~~
Tests demo component.
"""
import unittest
import homeassistant as ha
import homeassistant.components.automation as automation
import homeassistant.components.automation.event as event
from homeassistant.const import CONF_PLATFORM, ATTR_ENTITY_ID
class TestAutomationEvent(unittest.TestCase):
""" Test the event automation. """
def setUp(self): # pylint: disable=invalid-name
self.hass = ha.HomeAssistant()
self.calls = []
def record_call(service):
self.calls.append(service)
self.hass.services.register('test', 'automation', record_call)
def tearDown(self): # pylint: disable=invalid-name
""" Stop down stuff we started. """
self.hass.stop()
def test_setup_fails_if_unknown_platform(self):
self.assertFalse(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'i_do_not_exist'
}
}))
def test_service_data_not_a_dict(self):
automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'event',
event.CONF_EVENT_TYPE: 'test_event',
automation.CONF_SERVICE: 'test.automation',
automation.CONF_SERVICE_DATA: 100
}
})
self.hass.bus.fire('test_event')
self.hass.pool.block_till_done()
self.assertEqual(1, len(self.calls))
def test_service_specify_data(self):
automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'event',
event.CONF_EVENT_TYPE: 'test_event',
automation.CONF_SERVICE: 'test.automation',
automation.CONF_SERVICE_DATA: {'some': 'data'}
}
})
self.hass.bus.fire('test_event')
self.hass.pool.block_till_done()
self.assertEqual(1, len(self.calls))
self.assertEqual('data', self.calls[0].data['some'])
def test_service_specify_entity_id(self):
automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'event',
event.CONF_EVENT_TYPE: 'test_event',
automation.CONF_SERVICE: 'test.automation',
automation.CONF_SERVICE_ENTITY_ID: 'hello.world'
}
})
self.hass.bus.fire('test_event')
self.hass.pool.block_till_done()
self.assertEqual(1, len(self.calls))
self.assertEqual(['hello.world'], self.calls[0].data[ATTR_ENTITY_ID])

81
tests/components/automation/test_mqtt.py Normal file
View file

@ -0,0 +1,81 @@
"""
tests.test_component_demo
~~~~~~~~~~~~~~~~~~~~~~~~~~~
Tests demo component.
"""
import unittest
import homeassistant as ha
import homeassistant.components.automation as automation
import homeassistant.components.automation.mqtt as mqtt
from homeassistant.const import CONF_PLATFORM
from tests.common import mock_mqtt_component, fire_mqtt_message
class TestAutomationState(unittest.TestCase):
""" Test the event automation. """
def setUp(self): # pylint: disable=invalid-name
self.hass = ha.HomeAssistant()
mock_mqtt_component(self.hass)
self.calls = []
def record_call(service):
self.calls.append(service)
self.hass.services.register('test', 'automation', record_call)
def tearDown(self): # pylint: disable=invalid-name
""" Stop down stuff we started. """
self.hass.stop()
def test_setup_fails_if_no_topic(self):
self.assertFalse(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'mqtt',
automation.CONF_SERVICE: 'test.automation'
}
}))
def test_if_fires_on_topic_match(self):
self.assertTrue(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'mqtt',
mqtt.CONF_TOPIC: 'test-topic',
automation.CONF_SERVICE: 'test.automation'
}
}))
fire_mqtt_message(self.hass, 'test-topic', '')
self.hass.pool.block_till_done()
self.assertEqual(1, len(self.calls))
def test_if_fires_on_topic_and_payload_match(self):
self.assertTrue(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'mqtt',
mqtt.CONF_TOPIC: 'test-topic',
mqtt.CONF_PAYLOAD: 'hello',
automation.CONF_SERVICE: 'test.automation'
}
}))
fire_mqtt_message(self.hass, 'test-topic', 'hello')
self.hass.pool.block_till_done()
self.assertEqual(1, len(self.calls))
def test_if_not_fires_on_topic_but_no_payload_match(self):
self.assertTrue(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'mqtt',
mqtt.CONF_TOPIC: 'test-topic',
mqtt.CONF_PAYLOAD: 'hello',
automation.CONF_SERVICE: 'test.automation'
}
}))
fire_mqtt_message(self.hass, 'test-topic', 'no-hello')
self.hass.pool.block_till_done()
self.assertEqual(0, len(self.calls))

38
tests/components/automation/test_state.py
View file

@ -7,7 +7,6 @@ Tests demo component.
import unittest
import homeassistant as ha
import homeassistant.loader as loader
import homeassistant.components.automation as automation
import homeassistant.components.automation.state as state
from homeassistant.const import CONF_PLATFORM
@ -18,7 +17,6 @@ class TestAutomationState(unittest.TestCase):
def setUp(self): # pylint: disable=invalid-name
self.hass = ha.HomeAssistant()
loader.prepare(self.hass)
self.hass.states.set('test.entity', 'hello')
self.calls = []
@ -31,49 +29,57 @@ class TestAutomationState(unittest.TestCase):
""" Stop down stuff we started. """
self.hass.stop()
def test_setup_fails_if_no_entity_id(self):
self.assertFalse(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'state',
automation.CONF_SERVICE: 'test.automation'
}
}))
def test_if_fires_on_entity_change(self):
automation.setup(self.hass, {
self.assertTrue(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'state',
state.CONF_ENTITY_ID: 'test.entity',
automation.CONF_SERVICE: 'test.automation'
}
})
}))
self.hass.states.set('test.entity', 'world')
self.hass.pool.block_till_done()
self.assertEqual(1, len(self.calls))
def test_if_fires_on_entity_change_with_from_filter(self):
automation.setup(self.hass, {
self.assertTrue(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'state',
state.CONF_ENTITY_ID: 'test.entity',
state.CONF_FROM: 'hello',
automation.CONF_SERVICE: 'test.automation'
}
})
}))
self.hass.states.set('test.entity', 'world')
self.hass.pool.block_till_done()
self.assertEqual(1, len(self.calls))
def test_if_fires_on_entity_change_with_to_filter(self):
automation.setup(self.hass, {
self.assertTrue(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'state',
state.CONF_ENTITY_ID: 'test.entity',
state.CONF_TO: 'world',
automation.CONF_SERVICE: 'test.automation'
}
})
}))
self.hass.states.set('test.entity', 'world')
self.hass.pool.block_till_done()
self.assertEqual(1, len(self.calls))
def test_if_fires_on_entity_change_with_both_filters(self):
automation.setup(self.hass, {
self.assertTrue(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'state',
state.CONF_ENTITY_ID: 'test.entity',
@ -81,14 +87,14 @@ class TestAutomationState(unittest.TestCase):
state.CONF_TO: 'world',
automation.CONF_SERVICE: 'test.automation'
}
})
}))
self.hass.states.set('test.entity', 'world')
self.hass.pool.block_till_done()
self.assertEqual(1, len(self.calls))
def test_if_not_fires_if_to_filter_not_match(self):
automation.setup(self.hass, {
self.assertTrue(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'state',
state.CONF_ENTITY_ID: 'test.entity',
@ -96,7 +102,7 @@ class TestAutomationState(unittest.TestCase):
state.CONF_TO: 'world',
automation.CONF_SERVICE: 'test.automation'
}
})
}))
self.hass.states.set('test.entity', 'moon')
self.hass.pool.block_till_done()
@ -105,7 +111,7 @@ class TestAutomationState(unittest.TestCase):
def test_if_not_fires_if_from_filter_not_match(self):
self.hass.states.set('test.entity', 'bye')
automation.setup(self.hass, {
self.assertTrue(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'state',
state.CONF_ENTITY_ID: 'test.entity',
@ -113,20 +119,20 @@ class TestAutomationState(unittest.TestCase):
state.CONF_TO: 'world',
automation.CONF_SERVICE: 'test.automation'
}
})
}))
self.hass.states.set('test.entity', 'world')
self.hass.pool.block_till_done()
self.assertEqual(0, len(self.calls))
def test_if_not_fires_if_entity_not_match(self):
automation.setup(self.hass, {
self.assertTrue(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'state',
state.CONF_ENTITY_ID: 'test.another_entity',
automation.CONF_SERVICE: 'test.automation'
}
})
}))
self.hass.states.set('test.entity', 'world')
self.hass.pool.block_till_done()

16
tests/components/automation/test_time.py
View file

@ -34,13 +34,13 @@ class TestAutomationTime(unittest.TestCase):
self.hass.stop()
def test_if_fires_when_hour_matches(self):
automation.setup(self.hass, {
self.assertTrue(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'time',
time.CONF_HOURS: 0,
automation.CONF_SERVICE: 'test.automation'
}
})
}))
fire_time_changed(self.hass, dt_util.utcnow().replace(hour=0))
@ -49,13 +49,13 @@ class TestAutomationTime(unittest.TestCase):
self.assertEqual(1, len(self.calls))
def test_if_fires_when_minute_matches(self):
automation.setup(self.hass, {
self.assertTrue(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'time',
time.CONF_MINUTES: 0,
automation.CONF_SERVICE: 'test.automation'
}
})
}))
fire_time_changed(self.hass, dt_util.utcnow().replace(minute=0))
@ -64,13 +64,13 @@ class TestAutomationTime(unittest.TestCase):
self.assertEqual(1, len(self.calls))
def test_if_fires_when_second_matches(self):
automation.setup(self.hass, {
self.assertTrue(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'time',
time.CONF_SECONDS: 0,
automation.CONF_SERVICE: 'test.automation'
}
})
}))
fire_time_changed(self.hass, dt_util.utcnow().replace(second=0))
@ -79,7 +79,7 @@ class TestAutomationTime(unittest.TestCase):
self.assertEqual(1, len(self.calls))
def test_if_fires_when_all_matches(self):
automation.setup(self.hass, {
self.assertTrue(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'time',
time.CONF_HOURS: 0,
@ -87,7 +87,7 @@ class TestAutomationTime(unittest.TestCase):
time.CONF_SECONDS: 0,
automation.CONF_SERVICE: 'test.automation'
}
})
}))
fire_time_changed(self.hass, dt_util.utcnow().replace(
hour=0, minute=0, second=0))