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

Convert automation tests to async (#17794)

Browse source 
* Convert automation tests to async

* Fix 8 last tests

* Lint
This commit is contained in:
Paulus Schoutsen 2018-10-26 11:31:14 +02:00 • committed by GitHub
parent 3f4798b5c3
commit e276e899cf
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
11 changed files with 3544 additions and 3514 deletions
Download patch file Download diff file Expand all files Collapse all files

29
tests/components/automation/common.py
View file

@ -11,43 +11,34 @@ from homeassistant.loader import bind_hass
@bind_hass
def turn_on(hass, entity_id=None):
async def async_turn_on(hass, entity_id=None):
"""Turn on specified automation or all."""
data = {ATTR_ENTITY_ID: entity_id} if entity_id else {}
hass.services.call(DOMAIN, SERVICE_TURN_ON, data)
await hass.services.async_call(DOMAIN, SERVICE_TURN_ON, data)
@bind_hass
def turn_off(hass, entity_id=None):
async def async_turn_off(hass, entity_id=None):
"""Turn off specified automation or all."""
data = {ATTR_ENTITY_ID: entity_id} if entity_id else {}
hass.services.call(DOMAIN, SERVICE_TURN_OFF, data)
await hass.services.async_call(DOMAIN, SERVICE_TURN_OFF, data)
@bind_hass
def toggle(hass, entity_id=None):
async def async_toggle(hass, entity_id=None):
"""Toggle specified automation or all."""
data = {ATTR_ENTITY_ID: entity_id} if entity_id else {}
hass.services.call(DOMAIN, SERVICE_TOGGLE, data)
await hass.services.async_call(DOMAIN, SERVICE_TOGGLE, data)
@bind_hass
def trigger(hass, entity_id=None):
async def async_trigger(hass, entity_id=None):
"""Trigger specified automation or all."""
data = {ATTR_ENTITY_ID: entity_id} if entity_id else {}
hass.services.call(DOMAIN, SERVICE_TRIGGER, data)
await hass.services.async_call(DOMAIN, SERVICE_TRIGGER, data)
@bind_hass
def reload(hass):
async def async_reload(hass):
"""Reload the automation from config."""
hass.services.call(DOMAIN, SERVICE_RELOAD)
@bind_hass
def async_reload(hass):
"""Reload the automation from config.
Returns a coroutine object.
"""
return hass.services.async_call(DOMAIN, SERVICE_RELOAD)
await hass.services.async_call(DOMAIN, SERVICE_RELOAD)

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

@ -1,175 +1,172 @@
"""The tests for the Event automation."""
import unittest
import pytest
from homeassistant.core import Context, callback
from homeassistant.setup import setup_component
from homeassistant.core import Context
from homeassistant.setup import async_setup_component
import homeassistant.components.automation as automation
from tests.common import get_test_home_assistant, mock_component
from tests.common import mock_component
from tests.components.automation import common
from tests.common import async_mock_service
# pylint: disable=invalid-name
class TestAutomationEvent(unittest.TestCase):
"""Test the event automation."""
@pytest.fixture
def calls(hass):
"""Track calls to a mock serivce."""
return async_mock_service(hass, 'test', 'automation')
def setUp(self):
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
mock_component(self.hass, 'group')
self.calls = []
@callback
def record_call(service):
"""Record the call."""
self.calls.append(service)
@pytest.fixture(autouse=True)
def setup_comp(hass):
"""Initialize components."""
mock_component(hass, 'group')
self.hass.services.register('test', 'automation', record_call)
def tearDown(self):
"""Stop everything that was started."""
self.hass.stop()
async def test_if_fires_on_event(hass, calls):
"""Test the firing of events."""
context = Context()
def test_if_fires_on_event(self):
"""Test the firing of events."""
context = Context()
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'event',
'event_type': 'test_event',
},
'action': {
'service': 'test.automation',
}
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'event',
'event_type': 'test_event',
},
'action': {
'service': 'test.automation',
}
})
}
})
self.hass.bus.fire('test_event', context=context)
self.hass.block_till_done()
assert 1 == len(self.calls)
assert self.calls[0].context is context
hass.bus.async_fire('test_event', context=context)
await hass.async_block_till_done()
assert 1 == len(calls)
assert calls[0].context is context
common.turn_off(self.hass)
self.hass.block_till_done()
await common.async_turn_off(hass)
await hass.async_block_till_done()
self.hass.bus.fire('test_event')
self.hass.block_till_done()
assert 1 == len(self.calls)
hass.bus.async_fire('test_event')
await hass.async_block_till_done()
assert 1 == len(calls)
def test_if_fires_on_event_extra_data(self):
"""Test the firing of events still matches with event data."""
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'event',
'event_type': 'test_event',
},
'action': {
'service': 'test.automation',
}
async def test_if_fires_on_event_extra_data(hass, calls):
"""Test the firing of events still matches with event data."""
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'event',
'event_type': 'test_event',
},
'action': {
'service': 'test.automation',
}
})
}
})
self.hass.bus.fire('test_event', {'extra_key': 'extra_data'})
self.hass.block_till_done()
assert 1 == len(self.calls)
hass.bus.async_fire('test_event', {'extra_key': 'extra_data'})
await hass.async_block_till_done()
assert 1 == len(calls)
common.turn_off(self.hass)
self.hass.block_till_done()
await common.async_turn_off(hass)
await hass.async_block_till_done()
self.hass.bus.fire('test_event')
self.hass.block_till_done()
assert 1 == len(self.calls)
hass.bus.async_fire('test_event')
await hass.async_block_till_done()
assert 1 == len(calls)
def test_if_fires_on_event_with_data(self):
"""Test the firing of events with data."""
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'event',
'event_type': 'test_event',
'event_data': {'some_attr': 'some_value'}
},
'action': {
'service': 'test.automation',
}
async def test_if_fires_on_event_with_data(hass, calls):
"""Test the firing of events with data."""
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'event',
'event_type': 'test_event',
'event_data': {'some_attr': 'some_value'}
},
'action': {
'service': 'test.automation',
}
})
}
})
self.hass.bus.fire('test_event', {'some_attr': 'some_value',
'another': 'value'})
self.hass.block_till_done()
assert 1 == len(self.calls)
hass.bus.async_fire('test_event', {'some_attr': 'some_value',
'another': 'value'})
await hass.async_block_till_done()
assert 1 == len(calls)
def test_if_fires_on_event_with_empty_data_config(self):
"""Test the firing of events with empty data config.
The frontend automation editor can produce configurations with an
empty dict for event_data instead of no key.
"""
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'event',
'event_type': 'test_event',
'event_data': {}
},
'action': {
'service': 'test.automation',
}
async def test_if_fires_on_event_with_empty_data_config(hass, calls):
"""Test the firing of events with empty data config.
The frontend automation editor can produce configurations with an
empty dict for event_data instead of no key.
"""
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'event',
'event_type': 'test_event',
'event_data': {}
},
'action': {
'service': 'test.automation',
}
})
}
})
self.hass.bus.fire('test_event', {'some_attr': 'some_value',
'another': 'value'})
self.hass.block_till_done()
assert 1 == len(self.calls)
hass.bus.async_fire('test_event', {'some_attr': 'some_value',
'another': 'value'})
await hass.async_block_till_done()
assert 1 == len(calls)
def test_if_fires_on_event_with_nested_data(self):
"""Test the firing of events with nested data."""
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'event',
'event_type': 'test_event',
'event_data': {
'parent_attr': {
'some_attr': 'some_value'
}
async def test_if_fires_on_event_with_nested_data(hass, calls):
"""Test the firing of events with nested data."""
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'event',
'event_type': 'test_event',
'event_data': {
'parent_attr': {
'some_attr': 'some_value'
}
},
'action': {
'service': 'test.automation',
}
},
'action': {
'service': 'test.automation',
}
})
}
})
self.hass.bus.fire('test_event', {
'parent_attr': {
'some_attr': 'some_value',
'another': 'value'
hass.bus.async_fire('test_event', {
'parent_attr': {
'some_attr': 'some_value',
'another': 'value'
}
})
await hass.async_block_till_done()
assert 1 == len(calls)
async def test_if_not_fires_if_event_data_not_matches(hass, calls):
"""Test firing of event if no match."""
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'event',
'event_type': 'test_event',
'event_data': {'some_attr': 'some_value'}
},
'action': {
'service': 'test.automation',
}
})
self.hass.block_till_done()
assert 1 == len(self.calls)
}
})
def test_if_not_fires_if_event_data_not_matches(self):
"""Test firing of event if no match."""
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'event',
'event_type': 'test_event',
'event_data': {'some_attr': 'some_value'}
},
'action': {
'service': 'test.automation',
}
}
})
self.hass.bus.fire('test_event', {'some_attr': 'some_other_value'})
self.hass.block_till_done()
assert 0 == len(self.calls)
hass.bus.async_fire('test_event', {'some_attr': 'some_other_value'})
await hass.async_block_till_done()
assert 0 == len(calls)

457
tests/components/automation/test_geo_location.py
View file

@ -1,268 +1,265 @@
"""The tests for the geo location trigger."""
import unittest
import pytest
from homeassistant.components import automation, zone
from homeassistant.core import callback, Context
from homeassistant.setup import setup_component
from homeassistant.core import Context
from homeassistant.setup import async_setup_component
from tests.common import get_test_home_assistant, mock_component
from tests.common import mock_component
from tests.components.automation import common
from tests.common import async_mock_service
class TestAutomationGeoLocation(unittest.TestCase):
"""Test the geo location trigger."""
@pytest.fixture
def calls(hass):
"""Track calls to a mock serivce."""
return async_mock_service(hass, 'test', 'automation')
def setUp(self):
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
mock_component(self.hass, 'group')
assert setup_component(self.hass, zone.DOMAIN, {
@pytest.fixture(autouse=True)
def setup_comp(hass):
"""Initialize components."""
mock_component(hass, 'group')
hass.loop.run_until_complete(async_setup_component(hass, zone.DOMAIN, {
'zone': {
'name': 'test',
'latitude': 32.880837,
'longitude': -117.237561,
'radius': 250,
}
})
}))
self.calls = []
@callback
def record_call(service):
"""Record calls."""
self.calls.append(service)
async def test_if_fires_on_zone_enter(hass, calls):
"""Test for firing on zone enter."""
context = Context()
hass.states.async_set('geo_location.entity', 'hello', {
'latitude': 32.881011,
'longitude': -117.234758,
'source': 'test_source'
})
await hass.async_block_till_done()
self.hass.services.register('test', 'automation', record_call)
def tearDown(self):
"""Stop everything that was started."""
self.hass.stop()
def test_if_fires_on_zone_enter(self):
"""Test for firing on zone enter."""
context = Context()
self.hass.states.set('geo_location.entity', 'hello', {
'latitude': 32.881011,
'longitude': -117.234758,
'source': 'test_source'
})
self.hass.block_till_done()
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'geo_location',
'source': 'test_source',
'zone': 'zone.test',
'event': 'enter',
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'geo_location',
'source': 'test_source',
'zone': 'zone.test',
'event': 'enter',
},
'action': {
'service': 'test.automation',
'data_template': {
'some': '{{ trigger.%s }}' % '}} - {{ trigger.'.join((
'platform', 'entity_id',
'from_state.state', 'to_state.state',
'zone.name'))
},
'action': {
'service': 'test.automation',
'data_template': {
'some': '{{ trigger.%s }}' % '}} - {{ trigger.'.join((
'platform', 'entity_id',
'from_state.state', 'to_state.state',
'zone.name'))
},
}
}
})
}
})
self.hass.states.set('geo_location.entity', 'hello', {
'latitude': 32.880586,
'longitude': -117.237564
}, context=context)
self.hass.block_till_done()
hass.states.async_set('geo_location.entity', 'hello', {
'latitude': 32.880586,
'longitude': -117.237564
}, context=context)
await hass.async_block_till_done()
assert 1 == len(self.calls)
assert self.calls[0].context is context
assert 'geo_location - geo_location.entity - hello - hello - test' == \
self.calls[0].data['some']
assert 1 == len(calls)
assert calls[0].context is context
assert 'geo_location - geo_location.entity - hello - hello - test' == \
calls[0].data['some']
# Set out of zone again so we can trigger call
self.hass.states.set('geo_location.entity', 'hello', {
'latitude': 32.881011,
'longitude': -117.234758
})
self.hass.block_till_done()
# Set out of zone again so we can trigger call
hass.states.async_set('geo_location.entity', 'hello', {
'latitude': 32.881011,
'longitude': -117.234758
})
await hass.async_block_till_done()
common.turn_off(self.hass)
self.hass.block_till_done()
await common.async_turn_off(hass)
await hass.async_block_till_done()
self.hass.states.set('geo_location.entity', 'hello', {
'latitude': 32.880586,
'longitude': -117.237564
})
self.hass.block_till_done()
hass.states.async_set('geo_location.entity', 'hello', {
'latitude': 32.880586,
'longitude': -117.237564
})
await hass.async_block_till_done()
assert 1 == len(self.calls)
assert 1 == len(calls)
def test_if_not_fires_for_enter_on_zone_leave(self):
"""Test for not firing on zone leave."""
self.hass.states.set('geo_location.entity', 'hello', {
'latitude': 32.880586,
'longitude': -117.237564,
'source': 'test_source'
})
self.hass.block_till_done()
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'geo_location',
'source': 'test_source',
'zone': 'zone.test',
'event': 'enter',
async def test_if_not_fires_for_enter_on_zone_leave(hass, calls):
"""Test for not firing on zone leave."""
hass.states.async_set('geo_location.entity', 'hello', {
'latitude': 32.880586,
'longitude': -117.237564,
'source': 'test_source'
})
await hass.async_block_till_done()
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'geo_location',
'source': 'test_source',
'zone': 'zone.test',
'event': 'enter',
},
'action': {
'service': 'test.automation',
}
}
})
hass.states.async_set('geo_location.entity', 'hello', {
'latitude': 32.881011,
'longitude': -117.234758
})
await hass.async_block_till_done()
assert 0 == len(calls)
async def test_if_fires_on_zone_leave(hass, calls):
"""Test for firing on zone leave."""
hass.states.async_set('geo_location.entity', 'hello', {
'latitude': 32.880586,
'longitude': -117.237564,
'source': 'test_source'
})
await hass.async_block_till_done()
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'geo_location',
'source': 'test_source',
'zone': 'zone.test',
'event': 'leave',
},
'action': {
'service': 'test.automation',
}
}
})
hass.states.async_set('geo_location.entity', 'hello', {
'latitude': 32.881011,
'longitude': -117.234758,
'source': 'test_source'
})
await hass.async_block_till_done()
assert 1 == len(calls)
async def test_if_not_fires_for_leave_on_zone_enter(hass, calls):
"""Test for not firing on zone enter."""
hass.states.async_set('geo_location.entity', 'hello', {
'latitude': 32.881011,
'longitude': -117.234758,
'source': 'test_source'
})
await hass.async_block_till_done()
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'geo_location',
'source': 'test_source',
'zone': 'zone.test',
'event': 'leave',
},
'action': {
'service': 'test.automation',
}
}
})
hass.states.async_set('geo_location.entity', 'hello', {
'latitude': 32.880586,
'longitude': -117.237564
})
await hass.async_block_till_done()
assert 0 == len(calls)
async def test_if_fires_on_zone_appear(hass, calls):
"""Test for firing if entity appears in zone."""
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'geo_location',
'source': 'test_source',
'zone': 'zone.test',
'event': 'enter',
},
'action': {
'service': 'test.automation',
'data_template': {
'some': '{{ trigger.%s }}' % '}} - {{ trigger.'.join((
'platform', 'entity_id',
'from_state.state', 'to_state.state',
'zone.name'))
},
'action': {
'service': 'test.automation',
}
}
})
}
})
self.hass.states.set('geo_location.entity', 'hello', {
'latitude': 32.881011,
'longitude': -117.234758
})
self.hass.block_till_done()
# Entity appears in zone without previously existing outside the zone.
context = Context()
hass.states.async_set('geo_location.entity', 'hello', {
'latitude': 32.880586,
'longitude': -117.237564,
'source': 'test_source'
}, context=context)
await hass.async_block_till_done()
assert 0 == len(self.calls)
assert 1 == len(calls)
assert calls[0].context is context
assert 'geo_location - geo_location.entity - - hello - test' == \
calls[0].data['some']
def test_if_fires_on_zone_leave(self):
"""Test for firing on zone leave."""
self.hass.states.set('geo_location.entity', 'hello', {
'latitude': 32.880586,
'longitude': -117.237564,
'source': 'test_source'
})
self.hass.block_till_done()
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'geo_location',
'source': 'test_source',
'zone': 'zone.test',
'event': 'leave',
async def test_if_fires_on_zone_disappear(hass, calls):
"""Test for firing if entity disappears from zone."""
hass.states.async_set('geo_location.entity', 'hello', {
'latitude': 32.880586,
'longitude': -117.237564,
'source': 'test_source'
})
await hass.async_block_till_done()
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'geo_location',
'source': 'test_source',
'zone': 'zone.test',
'event': 'leave',
},
'action': {
'service': 'test.automation',
'data_template': {
'some': '{{ trigger.%s }}' % '}} - {{ trigger.'.join((
'platform', 'entity_id',
'from_state.state', 'to_state.state',
'zone.name'))
},
'action': {
'service': 'test.automation',
}
}
})
}
})
self.hass.states.set('geo_location.entity', 'hello', {
'latitude': 32.881011,
'longitude': -117.234758,
'source': 'test_source'
})
self.hass.block_till_done()
# Entity disappears from zone without new coordinates outside the zone.
hass.states.async_remove('geo_location.entity')
await hass.async_block_till_done()
assert 1 == len(self.calls)
def test_if_not_fires_for_leave_on_zone_enter(self):
"""Test for not firing on zone enter."""
self.hass.states.set('geo_location.entity', 'hello', {
'latitude': 32.881011,
'longitude': -117.234758,
'source': 'test_source'
})
self.hass.block_till_done()
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'geo_location',
'source': 'test_source',
'zone': 'zone.test',
'event': 'leave',
},
'action': {
'service': 'test.automation',
}
}
})
self.hass.states.set('geo_location.entity', 'hello', {
'latitude': 32.880586,
'longitude': -117.237564
})
self.hass.block_till_done()
assert 0 == len(self.calls)
def test_if_fires_on_zone_appear(self):
"""Test for firing if entity appears in zone."""
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'geo_location',
'source': 'test_source',
'zone': 'zone.test',
'event': 'enter',
},
'action': {
'service': 'test.automation',
'data_template': {
'some': '{{ trigger.%s }}' % '}} - {{ trigger.'.join((
'platform', 'entity_id',
'from_state.state', 'to_state.state',
'zone.name'))
},
}
}
})
# Entity appears in zone without previously existing outside the zone.
context = Context()
self.hass.states.set('geo_location.entity', 'hello', {
'latitude': 32.880586,
'longitude': -117.237564,
'source': 'test_source'
}, context=context)
self.hass.block_till_done()
assert 1 == len(self.calls)
assert self.calls[0].context is context
assert 'geo_location - geo_location.entity - - hello - test' == \
self.calls[0].data['some']
def test_if_fires_on_zone_disappear(self):
"""Test for firing if entity disappears from zone."""
self.hass.states.set('geo_location.entity', 'hello', {
'latitude': 32.880586,
'longitude': -117.237564,
'source': 'test_source'
})
self.hass.block_till_done()
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'geo_location',
'source': 'test_source',
'zone': 'zone.test',
'event': 'leave',
},
'action': {
'service': 'test.automation',
'data_template': {
'some': '{{ trigger.%s }}' % '}} - {{ trigger.'.join((
'platform', 'entity_id',
'from_state.state', 'to_state.state',
'zone.name'))
},
}
}
})
# Entity disappears from zone without new coordinates outside the zone.
self.hass.states.async_remove('geo_location.entity')
self.hass.block_till_done()
assert 1 == len(self.calls)
assert 'geo_location - geo_location.entity - hello - - test' == \
self.calls[0].data['some']
assert 1 == len(calls)
assert 'geo_location - geo_location.entity - hello - - test' == \
calls[0].data['some']

940
tests/components/automation/test_init.py
View file

File diff suppressed because it is too large Load diff

154
tests/components/automation/test_mqtt.py
View file

@ -1,102 +1,94 @@
"""The tests for the MQTT automation."""
import unittest
import pytest
from homeassistant.core import callback
from homeassistant.setup import setup_component
from homeassistant.setup import async_setup_component
import homeassistant.components.automation as automation
from tests.common import (
mock_mqtt_component, fire_mqtt_message, get_test_home_assistant,
mock_component)
async_fire_mqtt_message,
mock_component, async_mock_service, async_mock_mqtt_component)
from tests.components.automation import common
# pylint: disable=invalid-name
class TestAutomationMQTT(unittest.TestCase):
"""Test the event automation."""
@pytest.fixture
def calls(hass):
"""Track calls to a mock serivce."""
return async_mock_service(hass, 'test', 'automation')
def setUp(self):
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
mock_component(self.hass, 'group')
mock_mqtt_component(self.hass)
self.calls = []
@callback
def record_call(service):
"""Record calls."""
self.calls.append(service)
@pytest.fixture(autouse=True)
def setup_comp(hass):
"""Initialize components."""
mock_component(hass, 'group')
hass.loop.run_until_complete(async_mock_mqtt_component(hass))
self.hass.services.register('test', 'automation', record_call)
def tearDown(self):
"""Stop everything that was started."""
self.hass.stop()
def test_if_fires_on_topic_match(self):
"""Test if message is fired on topic match."""
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'mqtt',
'topic': 'test-topic'
async def test_if_fires_on_topic_match(hass, calls):
"""Test if message is fired on topic match."""
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'mqtt',
'topic': 'test-topic'
},
'action': {
'service': 'test.automation',
'data_template': {
'some': '{{ trigger.platform }} - {{ trigger.topic }}'
' - {{ trigger.payload }} - '
'{{ trigger.payload_json.hello }}'
},
'action': {
'service': 'test.automation',
'data_template': {
'some': '{{ trigger.platform }} - {{ trigger.topic }}'
' - {{ trigger.payload }} - '
'{{ trigger.payload_json.hello }}'
},
}
}
})
}
})
fire_mqtt_message(self.hass, 'test-topic', '{ "hello": "world" }')
self.hass.block_till_done()
assert 1 == len(self.calls)
assert 'mqtt - test-topic - { "hello": "world" } - world' == \
self.calls[0].data['some']
async_fire_mqtt_message(hass, 'test-topic', '{ "hello": "world" }')
await hass.async_block_till_done()
assert 1 == len(calls)
assert 'mqtt - test-topic - { "hello": "world" } - world' == \
calls[0].data['some']
common.turn_off(self.hass)
self.hass.block_till_done()
fire_mqtt_message(self.hass, 'test-topic', 'test_payload')
self.hass.block_till_done()
assert 1 == len(self.calls)
await common.async_turn_off(hass)
await hass.async_block_till_done()
async_fire_mqtt_message(hass, 'test-topic', 'test_payload')
await hass.async_block_till_done()
assert 1 == len(calls)
def test_if_fires_on_topic_and_payload_match(self):
"""Test if message is fired on topic and payload match."""
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'mqtt',
'topic': 'test-topic',
'payload': 'hello'
},
'action': {
'service': 'test.automation'
}
async def test_if_fires_on_topic_and_payload_match(hass, calls):
"""Test if message is fired on topic and payload match."""
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'mqtt',
'topic': 'test-topic',
'payload': 'hello'
},
'action': {
'service': 'test.automation'
}
})
}
})
fire_mqtt_message(self.hass, 'test-topic', 'hello')
self.hass.block_till_done()
assert 1 == len(self.calls)
async_fire_mqtt_message(hass, 'test-topic', 'hello')
await hass.async_block_till_done()
assert 1 == len(calls)
def test_if_not_fires_on_topic_but_no_payload_match(self):
"""Test if message is not fired on topic but no payload."""
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'mqtt',
'topic': 'test-topic',
'payload': 'hello'
},
'action': {
'service': 'test.automation'
}
async def test_if_not_fires_on_topic_but_no_payload_match(hass, calls):
"""Test if message is not fired on topic but no payload."""
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'mqtt',
'topic': 'test-topic',
'payload': 'hello'
},
'action': {
'service': 'test.automation'
}
})
}
})
fire_mqtt_message(self.hass, 'test-topic', 'no-hello')
self.hass.block_till_done()
assert 0 == len(self.calls)
async_fire_mqtt_message(hass, 'test-topic', 'no-hello')
await hass.async_block_till_done()
assert 0 == len(calls)

1736
tests/components/automation/test_numeric_state.py
View file

File diff suppressed because it is too large Load diff

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

File diff suppressed because it is too large Load diff

543
tests/components/automation/test_sun.py
View file

@ -1,322 +1,319 @@
"""The tests for the sun automation."""
from datetime import datetime
import unittest
import pytest
from unittest.mock import patch
from homeassistant.core import callback
from homeassistant.setup import setup_component
from homeassistant.setup import async_setup_component
from homeassistant.components import sun
import homeassistant.components.automation as automation
import homeassistant.util.dt as dt_util
from tests.common import (
fire_time_changed, get_test_home_assistant, mock_component)
async_fire_time_changed, mock_component, async_mock_service)
from tests.components.automation import common
# pylint: disable=invalid-name
class TestAutomationSun(unittest.TestCase):
"""Test the sun automation."""
@pytest.fixture
def calls(hass):
"""Track calls to a mock serivce."""
return async_mock_service(hass, 'test', 'automation')
def setUp(self):
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
mock_component(self.hass, 'group')
setup_component(self.hass, sun.DOMAIN, {
sun.DOMAIN: {sun.CONF_ELEVATION: 0}})
self.calls = []
@pytest.fixture(autouse=True)
def setup_comp(hass):
"""Initialize components."""
mock_component(hass, 'group')
hass.loop.run_until_complete(async_setup_component(hass, sun.DOMAIN, {
sun.DOMAIN: {sun.CONF_ELEVATION: 0}}))
@callback
def record_call(service):
"""Call recorder."""
self.calls.append(service)
self.hass.services.register('test', 'automation', record_call)
async def test_sunset_trigger(hass, calls):
"""Test the sunset trigger."""
now = datetime(2015, 9, 15, 23, tzinfo=dt_util.UTC)
trigger_time = datetime(2015, 9, 16, 2, tzinfo=dt_util.UTC)
def tearDown(self):
"""Stop everything that was started."""
self.hass.stop()
with patch('homeassistant.util.dt.utcnow',
return_value=now):
await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'sun',
'event': 'sunset',
},
'action': {
'service': 'test.automation',
}
}
})
def test_sunset_trigger(self):
"""Test the sunset trigger."""
now = datetime(2015, 9, 15, 23, tzinfo=dt_util.UTC)
trigger_time = datetime(2015, 9, 16, 2, tzinfo=dt_util.UTC)
await common.async_turn_off(hass)
await hass.async_block_till_done()
with patch('homeassistant.util.dt.utcnow',
return_value=now):
setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'sun',
'event': 'sunset',
async_fire_time_changed(hass, trigger_time)
await hass.async_block_till_done()
assert 0 == len(calls)
with patch('homeassistant.util.dt.utcnow',
return_value=now):
await common.async_turn_on(hass)
await hass.async_block_till_done()
async_fire_time_changed(hass, trigger_time)
await hass.async_block_till_done()
assert 1 == len(calls)
async def test_sunrise_trigger(hass, calls):
"""Test the sunrise trigger."""
now = datetime(2015, 9, 13, 23, tzinfo=dt_util.UTC)
trigger_time = datetime(2015, 9, 16, 14, tzinfo=dt_util.UTC)
with patch('homeassistant.util.dt.utcnow',
return_value=now):
await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'sun',
'event': 'sunrise',
},
'action': {
'service': 'test.automation',
}
}
})
async_fire_time_changed(hass, trigger_time)
await hass.async_block_till_done()
assert 1 == len(calls)
async def test_sunset_trigger_with_offset(hass, calls):
"""Test the sunset trigger with offset."""
now = datetime(2015, 9, 15, 23, tzinfo=dt_util.UTC)
trigger_time = datetime(2015, 9, 16, 2, 30, tzinfo=dt_util.UTC)
with patch('homeassistant.util.dt.utcnow',
return_value=now):
await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'sun',
'event': 'sunset',
'offset': '0:30:00'
},
'action': {
'service': 'test.automation',
'data_template': {
'some':
'{{ trigger.%s }}' % '}} - {{ trigger.'.join((
'platform', 'event', 'offset'))
},
'action': {
'service': 'test.automation',
}
}
})
common.turn_off(self.hass)
self.hass.block_till_done()
fire_time_changed(self.hass, trigger_time)
self.hass.block_till_done()
assert 0 == len(self.calls)
with patch('homeassistant.util.dt.utcnow',
return_value=now):
common.turn_on(self.hass)
self.hass.block_till_done()
fire_time_changed(self.hass, trigger_time)
self.hass.block_till_done()
assert 1 == len(self.calls)
def test_sunrise_trigger(self):
"""Test the sunrise trigger."""
now = datetime(2015, 9, 13, 23, tzinfo=dt_util.UTC)
trigger_time = datetime(2015, 9, 16, 14, tzinfo=dt_util.UTC)
with patch('homeassistant.util.dt.utcnow',
return_value=now):
setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'sun',
'event': 'sunrise',
},
'action': {
'service': 'test.automation',
}
}
})
fire_time_changed(self.hass, trigger_time)
self.hass.block_till_done()
assert 1 == len(self.calls)
def test_sunset_trigger_with_offset(self):
"""Test the sunset trigger with offset."""
now = datetime(2015, 9, 15, 23, tzinfo=dt_util.UTC)
trigger_time = datetime(2015, 9, 16, 2, 30, tzinfo=dt_util.UTC)
with patch('homeassistant.util.dt.utcnow',
return_value=now):
setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'sun',
'event': 'sunset',
'offset': '0:30:00'
},
'action': {
'service': 'test.automation',
'data_template': {
'some':
'{{ trigger.%s }}' % '}} - {{ trigger.'.join((
'platform', 'event', 'offset'))
},
}
}
})
fire_time_changed(self.hass, trigger_time)
self.hass.block_till_done()
assert 1 == len(self.calls)
assert 'sun - sunset - 0:30:00' == self.calls[0].data['some']
def test_sunrise_trigger_with_offset(self):
"""Test the sunrise trigger with offset."""
now = datetime(2015, 9, 13, 23, tzinfo=dt_util.UTC)
trigger_time = datetime(2015, 9, 16, 13, 30, tzinfo=dt_util.UTC)
with patch('homeassistant.util.dt.utcnow',
return_value=now):
setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'sun',
'event': 'sunrise',
'offset': '-0:30:00'
},
'action': {
'service': 'test.automation',
}
}
})
fire_time_changed(self.hass, trigger_time)
self.hass.block_till_done()
assert 1 == len(self.calls)
def test_if_action_before(self):
"""Test if action was before."""
setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'event',
'event_type': 'test_event',
},
'condition': {
'condition': 'sun',
'before': 'sunrise',
},
'action': {
'service': 'test.automation'
}
}
})
now = datetime(2015, 9, 16, 15, tzinfo=dt_util.UTC)
with patch('homeassistant.util.dt.utcnow',
return_value=now):
self.hass.bus.fire('test_event')
self.hass.block_till_done()
assert 0 == len(self.calls)
async_fire_time_changed(hass, trigger_time)
await hass.async_block_till_done()
assert 1 == len(calls)
assert 'sun - sunset - 0:30:00' == calls[0].data['some']
now = datetime(2015, 9, 16, 10, tzinfo=dt_util.UTC)
with patch('homeassistant.util.dt.utcnow',
return_value=now):
self.hass.bus.fire('test_event')
self.hass.block_till_done()
assert 1 == len(self.calls)
def test_if_action_after(self):
"""Test if action was after."""
setup_component(self.hass, automation.DOMAIN, {
async def test_sunrise_trigger_with_offset(hass, calls):
"""Test the sunrise trigger with offset."""
now = datetime(2015, 9, 13, 23, tzinfo=dt_util.UTC)
trigger_time = datetime(2015, 9, 16, 13, 30, tzinfo=dt_util.UTC)
with patch('homeassistant.util.dt.utcnow',
return_value=now):
await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'event',
'event_type': 'test_event',
},
'condition': {
'condition': 'sun',
'after': 'sunrise',
'platform': 'sun',
'event': 'sunrise',
'offset': '-0:30:00'
},
'action': {
'service': 'test.automation'
'service': 'test.automation',
}
}
})
now = datetime(2015, 9, 16, 13, tzinfo=dt_util.UTC)
with patch('homeassistant.util.dt.utcnow',
return_value=now):
self.hass.bus.fire('test_event')
self.hass.block_till_done()
assert 0 == len(self.calls)
async_fire_time_changed(hass, trigger_time)
await hass.async_block_till_done()
assert 1 == len(calls)
now = datetime(2015, 9, 16, 15, tzinfo=dt_util.UTC)
with patch('homeassistant.util.dt.utcnow',
return_value=now):
self.hass.bus.fire('test_event')
self.hass.block_till_done()
assert 1 == len(self.calls)
def test_if_action_before_with_offset(self):
"""Test if action was before offset."""
setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'event',
'event_type': 'test_event',
},
'condition': {
'condition': 'sun',
'before': 'sunrise',
'before_offset': '+1:00:00'
},
'action': {
'service': 'test.automation'
}
async def test_if_action_before(hass, calls):
"""Test if action was before."""
await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'event',
'event_type': 'test_event',
},
'condition': {
'condition': 'sun',
'before': 'sunrise',
},
'action': {
'service': 'test.automation'
}
})
}
})
now = datetime(2015, 9, 16, 14, 32, 44, tzinfo=dt_util.UTC)
with patch('homeassistant.util.dt.utcnow',
return_value=now):
self.hass.bus.fire('test_event')
self.hass.block_till_done()
assert 0 == len(self.calls)
now = datetime(2015, 9, 16, 15, tzinfo=dt_util.UTC)
with patch('homeassistant.util.dt.utcnow',
return_value=now):
hass.bus.async_fire('test_event')
await hass.async_block_till_done()
assert 0 == len(calls)
now = datetime(2015, 9, 16, 14, 32, 43, tzinfo=dt_util.UTC)
with patch('homeassistant.util.dt.utcnow',
return_value=now):
self.hass.bus.fire('test_event')
self.hass.block_till_done()
assert 1 == len(self.calls)
now = datetime(2015, 9, 16, 10, tzinfo=dt_util.UTC)
with patch('homeassistant.util.dt.utcnow',
return_value=now):
hass.bus.async_fire('test_event')
await hass.async_block_till_done()
assert 1 == len(calls)
def test_if_action_after_with_offset(self):
"""Test if action was after offset."""
setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'event',
'event_type': 'test_event',
},
'condition': {
'condition': 'sun',
'after': 'sunrise',
'after_offset': '+1:00:00'
},
'action': {
'service': 'test.automation'
}
async def test_if_action_after(hass, calls):
"""Test if action was after."""
await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'event',
'event_type': 'test_event',
},
'condition': {
'condition': 'sun',
'after': 'sunrise',
},
'action': {
'service': 'test.automation'
}
})
}
})
now = datetime(2015, 9, 16, 14, 32, 42, tzinfo=dt_util.UTC)
with patch('homeassistant.util.dt.utcnow',
return_value=now):
self.hass.bus.fire('test_event')
self.hass.block_till_done()
assert 0 == len(self.calls)
now = datetime(2015, 9, 16, 13, tzinfo=dt_util.UTC)
with patch('homeassistant.util.dt.utcnow',
return_value=now):
hass.bus.async_fire('test_event')
await hass.async_block_till_done()
assert 0 == len(calls)
now = datetime(2015, 9, 16, 14, 32, 43, tzinfo=dt_util.UTC)
with patch('homeassistant.util.dt.utcnow',
return_value=now):
self.hass.bus.fire('test_event')
self.hass.block_till_done()
assert 1 == len(self.calls)
now = datetime(2015, 9, 16, 15, tzinfo=dt_util.UTC)
with patch('homeassistant.util.dt.utcnow',
return_value=now):
hass.bus.async_fire('test_event')
await hass.async_block_till_done()
assert 1 == len(calls)
def test_if_action_before_and_after_during(self):
"""Test if action was before and after during."""
setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'event',
'event_type': 'test_event',
},
'condition': {
'condition': 'sun',
'after': 'sunrise',
'before': 'sunset'
},
'action': {
'service': 'test.automation'
}
async def test_if_action_before_with_offset(hass, calls):
"""Test if action was before offset."""
await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'event',
'event_type': 'test_event',
},
'condition': {
'condition': 'sun',
'before': 'sunrise',
'before_offset': '+1:00:00'
},
'action': {
'service': 'test.automation'
}
})
}
})
now = datetime(2015, 9, 16, 13, 8, 51, tzinfo=dt_util.UTC)
with patch('homeassistant.util.dt.utcnow',
return_value=now):
self.hass.bus.fire('test_event')
self.hass.block_till_done()
assert 0 == len(self.calls)
now = datetime(2015, 9, 16, 14, 32, 44, tzinfo=dt_util.UTC)
with patch('homeassistant.util.dt.utcnow',
return_value=now):
hass.bus.async_fire('test_event')
await hass.async_block_till_done()
assert 0 == len(calls)
now = datetime(2015, 9, 17, 2, 25, 18, tzinfo=dt_util.UTC)
with patch('homeassistant.util.dt.utcnow',
return_value=now):
self.hass.bus.fire('test_event')
self.hass.block_till_done()
assert 0 == len(self.calls)
now = datetime(2015, 9, 16, 14, 32, 43, tzinfo=dt_util.UTC)
with patch('homeassistant.util.dt.utcnow',
return_value=now):
hass.bus.async_fire('test_event')
await hass.async_block_till_done()
assert 1 == len(calls)
now = datetime(2015, 9, 16, 16, tzinfo=dt_util.UTC)
with patch('homeassistant.util.dt.utcnow',
return_value=now):
self.hass.bus.fire('test_event')
self.hass.block_till_done()
assert 1 == len(self.calls)
async def test_if_action_after_with_offset(hass, calls):
"""Test if action was after offset."""
await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'event',
'event_type': 'test_event',
},
'condition': {
'condition': 'sun',
'after': 'sunrise',
'after_offset': '+1:00:00'
},
'action': {
'service': 'test.automation'
}
}
})
now = datetime(2015, 9, 16, 14, 32, 42, tzinfo=dt_util.UTC)
with patch('homeassistant.util.dt.utcnow',
return_value=now):
hass.bus.async_fire('test_event')
await hass.async_block_till_done()
assert 0 == len(calls)
now = datetime(2015, 9, 16, 14, 32, 43, tzinfo=dt_util.UTC)
with patch('homeassistant.util.dt.utcnow',
return_value=now):
hass.bus.async_fire('test_event')
await hass.async_block_till_done()
assert 1 == len(calls)
async def test_if_action_before_and_after_during(hass, calls):
"""Test if action was before and after during."""
await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'event',
'event_type': 'test_event',
},
'condition': {
'condition': 'sun',
'after': 'sunrise',
'before': 'sunset'
},
'action': {
'service': 'test.automation'
}
}
})
now = datetime(2015, 9, 16, 13, 8, 51, tzinfo=dt_util.UTC)
with patch('homeassistant.util.dt.utcnow',
return_value=now):
hass.bus.async_fire('test_event')
await hass.async_block_till_done()
assert 0 == len(calls)
now = datetime(2015, 9, 17, 2, 25, 18, tzinfo=dt_util.UTC)
with patch('homeassistant.util.dt.utcnow',
return_value=now):
hass.bus.async_fire('test_event')
await hass.async_block_till_done()
assert 0 == len(calls)
now = datetime(2015, 9, 16, 16, tzinfo=dt_util.UTC)
with patch('homeassistant.util.dt.utcnow',
return_value=now):
hass.bus.async_fire('test_event')
await hass.async_block_till_done()
assert 1 == len(calls)

811
tests/components/automation/test_template.py
View file

@ -1,44 +1,380 @@
"""The tests for the Template automation."""
import unittest
import pytest
from homeassistant.core import Context, callback
from homeassistant.setup import setup_component
from homeassistant.core import Context
from homeassistant.setup import async_setup_component
import homeassistant.components.automation as automation
from tests.common import (
get_test_home_assistant, assert_setup_component, mock_component)
from tests.common import (assert_setup_component, mock_component)
from tests.components.automation import common
from tests.common import async_mock_service
# pylint: disable=invalid-name
class TestAutomationTemplate(unittest.TestCase):
"""Test the event automation."""
@pytest.fixture
def calls(hass):
"""Track calls to a mock serivce."""
return async_mock_service(hass, 'test', 'automation')
def setUp(self):
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
mock_component(self.hass, 'group')
self.hass.states.set('test.entity', 'hello')
self.calls = []
@callback
def record_call(service):
"""Record calls."""
self.calls.append(service)
@pytest.fixture(autouse=True)
def setup_comp(hass):
"""Initialize components."""
mock_component(hass, 'group')
hass.states.async_set('test.entity', 'hello')
self.hass.services.register('test', 'automation', record_call)
def tearDown(self):
"""Stop everything that was started."""
self.hass.stop()
async def test_if_fires_on_change_bool(hass, calls):
"""Test for firing on boolean change."""
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'template',
'value_template': '{{ true }}',
},
'action': {
'service': 'test.automation'
}
}
})
def test_if_fires_on_change_bool(self):
"""Test for firing on boolean change."""
assert setup_component(self.hass, automation.DOMAIN, {
hass.states.async_set('test.entity', 'world')
await hass.async_block_till_done()
assert 1 == len(calls)
await common.async_turn_off(hass)
await hass.async_block_till_done()
hass.states.async_set('test.entity', 'planet')
await hass.async_block_till_done()
assert 1 == len(calls)
async def test_if_fires_on_change_str(hass, calls):
"""Test for firing on change."""
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'template',
'value_template': 'true',
},
'action': {
'service': 'test.automation'
}
}
})
hass.states.async_set('test.entity', 'world')
await hass.async_block_till_done()
assert 1 == len(calls)
async def test_if_fires_on_change_str_crazy(hass, calls):
"""Test for firing on change."""
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'template',
'value_template': 'TrUE',
},
'action': {
'service': 'test.automation'
}
}
})
hass.states.async_set('test.entity', 'world')
await hass.async_block_till_done()
assert 1 == len(calls)
async def test_if_not_fires_on_change_bool(hass, calls):
"""Test for not firing on boolean change."""
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'template',
'value_template': '{{ false }}',
},
'action': {
'service': 'test.automation'
}
}
})
hass.states.async_set('test.entity', 'world')
await hass.async_block_till_done()
assert 0 == len(calls)
async def test_if_not_fires_on_change_str(hass, calls):
"""Test for not firing on string change."""
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'template',
'value_template': 'False',
},
'action': {
'service': 'test.automation'
}
}
})
hass.states.async_set('test.entity', 'world')
await hass.async_block_till_done()
assert 0 == len(calls)
async def test_if_not_fires_on_change_str_crazy(hass, calls):
"""Test for not firing on string change."""
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'template',
'value_template': 'Anything other than "true" is false.',
},
'action': {
'service': 'test.automation'
}
}
})
hass.states.async_set('test.entity', 'world')
await hass.async_block_till_done()
assert 0 == len(calls)
async def test_if_fires_on_no_change(hass, calls):
"""Test for firing on no change."""
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'template',
'value_template': '{{ true }}',
},
'action': {
'service': 'test.automation'
}
}
})
await hass.async_block_till_done()
cur_len = len(calls)
hass.states.async_set('test.entity', 'hello')
await hass.async_block_till_done()
assert cur_len == len(calls)
async def test_if_fires_on_two_change(hass, calls):
"""Test for firing on two changes."""
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'template',
'value_template': '{{ true }}',
},
'action': {
'service': 'test.automation'
}
}
})
# Trigger once
hass.states.async_set('test.entity', 'world')
await hass.async_block_till_done()
assert 1 == len(calls)
# Trigger again
hass.states.async_set('test.entity', 'world')
await hass.async_block_till_done()
assert 1 == len(calls)
async def test_if_fires_on_change_with_template(hass, calls):
"""Test for firing on change with template."""
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'template',
'value_template': '{{ is_state("test.entity", "world") }}',
},
'action': {
'service': 'test.automation'
}
}
})
hass.states.async_set('test.entity', 'world')
await hass.async_block_till_done()
assert 1 == len(calls)
async def test_if_not_fires_on_change_with_template(hass, calls):
"""Test for not firing on change with template."""
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'template',
'value_template': '{{ is_state("test.entity", "hello") }}',
},
'action': {
'service': 'test.automation'
}
}
})
await hass.async_block_till_done()
hass.states.async_set('test.entity', 'world')
await hass.async_block_till_done()
assert len(calls) == 0
async def test_if_fires_on_change_with_template_advanced(hass, calls):
"""Test for firing on change with template advanced."""
context = Context()
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'template',
'value_template': '{{ is_state("test.entity", "world") }}'
},
'action': {
'service': 'test.automation',
'data_template': {
'some':
'{{ trigger.%s }}' % '}} - {{ trigger.'.join((
'platform', 'entity_id', 'from_state.state',
'to_state.state'))
},
}
}
})
await hass.async_block_till_done()
hass.states.async_set('test.entity', 'world', context=context)
await hass.async_block_till_done()
assert 1 == len(calls)
assert calls[0].context is context
assert 'template - test.entity - hello - world' == \
calls[0].data['some']
async def test_if_fires_on_no_change_with_template_advanced(hass, calls):
"""Test for firing on no change with template advanced."""
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'template',
'value_template': '''{%- if is_state("test.entity", "world") -%}
true
{%- else -%}
false
{%- endif -%}''',
},
'action': {
'service': 'test.automation'
}
}
})
# Different state
hass.states.async_set('test.entity', 'worldz')
await hass.async_block_till_done()
assert 0 == len(calls)
# Different state
hass.states.async_set('test.entity', 'hello')
await hass.async_block_till_done()
assert 0 == len(calls)
async def test_if_fires_on_change_with_template_2(hass, calls):
"""Test for firing on change with template."""
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'template',
'value_template':
'{{ not is_state("test.entity", "world") }}',
},
'action': {
'service': 'test.automation'
}
}
})
await hass.async_block_till_done()
hass.states.async_set('test.entity', 'world')
await hass.async_block_till_done()
assert len(calls) == 0
hass.states.async_set('test.entity', 'home')
await hass.async_block_till_done()
assert len(calls) == 1
hass.states.async_set('test.entity', 'work')
await hass.async_block_till_done()
assert len(calls) == 1
hass.states.async_set('test.entity', 'not_home')
await hass.async_block_till_done()
assert len(calls) == 1
hass.states.async_set('test.entity', 'world')
await hass.async_block_till_done()
assert len(calls) == 1
hass.states.async_set('test.entity', 'home')
await hass.async_block_till_done()
assert len(calls) == 2
async def test_if_action(hass, calls):
"""Test for firing if action."""
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'event',
'event_type': 'test_event',
},
'condition': [{
'condition': 'template',
'value_template': '{{ is_state("test.entity", "world") }}'
}],
'action': {
'service': 'test.automation'
}
}
})
# Condition is not true yet
hass.bus.async_fire('test_event')
await hass.async_block_till_done()
assert 0 == len(calls)
# Change condition to true, but it shouldn't be triggered yet
hass.states.async_set('test.entity', 'world')
await hass.async_block_till_done()
assert 0 == len(calls)
# Condition is true and event is triggered
hass.bus.async_fire('test_event')
await hass.async_block_till_done()
assert 1 == len(calls)
async def test_if_fires_on_change_with_bad_template(hass, calls):
"""Test for firing on change with bad template."""
with assert_setup_component(0):
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'template',
'value_template': '{{ true }}',
'value_template': '{{ ',
},
'action': {
'service': 'test.automation'
@ -46,388 +382,55 @@ class TestAutomationTemplate(unittest.TestCase):
}
})
self.hass.states.set('test.entity', 'world')
self.hass.block_till_done()
assert 1 == len(self.calls)
common.turn_off(self.hass)
self.hass.block_till_done()
self.hass.states.set('test.entity', 'planet')
self.hass.block_till_done()
assert 1 == len(self.calls)
def test_if_fires_on_change_str(self):
"""Test for firing on change."""
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'template',
'value_template': 'true',
},
'action': {
'service': 'test.automation'
}
async def test_if_fires_on_change_with_bad_template_2(hass, calls):
"""Test for firing on change with bad template."""
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'template',
'value_template': '{{ xyz | round(0) }}',
},
'action': {
'service': 'test.automation'
}
})
}
})
self.hass.states.set('test.entity', 'world')
self.hass.block_till_done()
assert 1 == len(self.calls)
hass.states.async_set('test.entity', 'world')
await hass.async_block_till_done()
assert 0 == len(calls)
def test_if_fires_on_change_str_crazy(self):
"""Test for firing on change."""
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'template',
'value_template': 'TrUE',
},
'action': {
'service': 'test.automation'
}
}
})
self.hass.states.set('test.entity', 'world')
self.hass.block_till_done()
assert 1 == len(self.calls)
async def test_wait_template_with_trigger(hass, calls):
"""Test using wait template with 'trigger.entity_id'."""
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'template',
'value_template':
"{{ states.test.entity.state == 'world' }}",
},
'action': [
{'wait_template':
"{{ is_state(trigger.entity_id, 'hello') }}"},
{'service': 'test.automation',
'data_template': {
'some':
'{{ trigger.%s }}' % '}} - {{ trigger.'.join((
'platform', 'entity_id', 'from_state.state',
'to_state.state'))
}}
],
}
})
def test_if_not_fires_on_change_bool(self):
"""Test for not firing on boolean change."""
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'template',
'value_template': '{{ false }}',
},
'action': {
'service': 'test.automation'
}
}
})
await hass.async_block_till_done()
self.hass.states.set('test.entity', 'world')
self.hass.block_till_done()
assert 0 == len(self.calls)
def test_if_not_fires_on_change_str(self):
"""Test for not firing on string change."""
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'template',
'value_template': 'False',
},
'action': {
'service': 'test.automation'
}
}
})
self.hass.states.set('test.entity', 'world')
self.hass.block_till_done()
assert 0 == len(self.calls)
def test_if_not_fires_on_change_str_crazy(self):
"""Test for not firing on string change."""
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'template',
'value_template': 'Anything other than "true" is false.',
},
'action': {
'service': 'test.automation'
}
}
})
self.hass.states.set('test.entity', 'world')
self.hass.block_till_done()
assert 0 == len(self.calls)
def test_if_fires_on_no_change(self):
"""Test for firing on no change."""
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'template',
'value_template': '{{ true }}',
},
'action': {
'service': 'test.automation'
}
}
})
self.hass.block_till_done()
self.calls = []
self.hass.states.set('test.entity', 'hello')
self.hass.block_till_done()
assert 0 == len(self.calls)
def test_if_fires_on_two_change(self):
"""Test for firing on two changes."""
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'template',
'value_template': '{{ true }}',
},
'action': {
'service': 'test.automation'
}
}
})
# Trigger once
self.hass.states.set('test.entity', 'world')
self.hass.block_till_done()
assert 1 == len(self.calls)
# Trigger again
self.hass.states.set('test.entity', 'world')
self.hass.block_till_done()
assert 1 == len(self.calls)
def test_if_fires_on_change_with_template(self):
"""Test for firing on change with template."""
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'template',
'value_template': '{{ is_state("test.entity", "world") }}',
},
'action': {
'service': 'test.automation'
}
}
})
self.hass.states.set('test.entity', 'world')
self.hass.block_till_done()
assert 1 == len(self.calls)
def test_if_not_fires_on_change_with_template(self):
"""Test for not firing on change with template."""
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'template',
'value_template': '{{ is_state("test.entity", "hello") }}',
},
'action': {
'service': 'test.automation'
}
}
})
self.hass.block_till_done()
self.calls = []
self.hass.states.set('test.entity', 'world')
self.hass.block_till_done()
assert len(self.calls) == 0
def test_if_fires_on_change_with_template_advanced(self):
"""Test for firing on change with template advanced."""
context = Context()
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'template',
'value_template': '{{ is_state("test.entity", "world") }}'
},
'action': {
'service': 'test.automation',
'data_template': {
'some':
'{{ trigger.%s }}' % '}} - {{ trigger.'.join((
'platform', 'entity_id', 'from_state.state',
'to_state.state'))
},
}
}
})
self.hass.block_till_done()
self.calls = []
self.hass.states.set('test.entity', 'world', context=context)
self.hass.block_till_done()
assert 1 == len(self.calls)
assert self.calls[0].context is context
assert 'template - test.entity - hello - world' == \
self.calls[0].data['some']
def test_if_fires_on_no_change_with_template_advanced(self):
"""Test for firing on no change with template advanced."""
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'template',
'value_template': '''{%- if is_state("test.entity", "world") -%}
true
{%- else -%}
false
{%- endif -%}''',
},
'action': {
'service': 'test.automation'
}
}
})
# Different state
self.hass.states.set('test.entity', 'worldz')
self.hass.block_till_done()
assert 0 == len(self.calls)
# Different state
self.hass.states.set('test.entity', 'hello')
self.hass.block_till_done()
assert 0 == len(self.calls)
def test_if_fires_on_change_with_template_2(self):
"""Test for firing on change with template."""
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'template',
'value_template':
'{{ not is_state("test.entity", "world") }}',
},
'action': {
'service': 'test.automation'
}
}
})
self.hass.block_till_done()
self.calls = []
self.hass.states.set('test.entity', 'world')
self.hass.block_till_done()
assert len(self.calls) == 0
self.hass.states.set('test.entity', 'home')
self.hass.block_till_done()
assert len(self.calls) == 1
self.hass.states.set('test.entity', 'work')
self.hass.block_till_done()
assert len(self.calls) == 1
self.hass.states.set('test.entity', 'not_home')
self.hass.block_till_done()
assert len(self.calls) == 1
self.hass.states.set('test.entity', 'world')
self.hass.block_till_done()
assert len(self.calls) == 1
self.hass.states.set('test.entity', 'home')
self.hass.block_till_done()
assert len(self.calls) == 2
def test_if_action(self):
"""Test for firing if action."""
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'event',
'event_type': 'test_event',
},
'condition': [{
'condition': 'template',
'value_template': '{{ is_state("test.entity", "world") }}'
}],
'action': {
'service': 'test.automation'
}
}
})
# Condition is not true yet
self.hass.bus.fire('test_event')
self.hass.block_till_done()
assert 0 == len(self.calls)
# Change condition to true, but it shouldn't be triggered yet
self.hass.states.set('test.entity', 'world')
self.hass.block_till_done()
assert 0 == len(self.calls)
# Condition is true and event is triggered
self.hass.bus.fire('test_event')
self.hass.block_till_done()
assert 1 == len(self.calls)
def test_if_fires_on_change_with_bad_template(self):
"""Test for firing on change with bad template."""
with assert_setup_component(0):
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'template',
'value_template': '{{ ',
},
'action': {
'service': 'test.automation'
}
}
})
def test_if_fires_on_change_with_bad_template_2(self):
"""Test for firing on change with bad template."""
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'template',
'value_template': '{{ xyz | round(0) }}',
},
'action': {
'service': 'test.automation'
}
}
})
self.hass.states.set('test.entity', 'world')
self.hass.block_till_done()
assert 0 == len(self.calls)
def test_wait_template_with_trigger(self):
"""Test using wait template with 'trigger.entity_id'."""
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'template',
'value_template':
"{{ states.test.entity.state == 'world' }}",
},
'action': [
{'wait_template':
"{{ is_state(trigger.entity_id, 'hello') }}"},
{'service': 'test.automation',
'data_template': {
'some':
'{{ trigger.%s }}' % '}} - {{ trigger.'.join((
'platform', 'entity_id', 'from_state.state',
'to_state.state'))
}}
],
}
})
self.hass.block_till_done()
self.calls = []
self.hass.states.set('test.entity', 'world')
self.hass.block_till_done()
self.hass.states.set('test.entity', 'hello')
self.hass.block_till_done()
assert 1 == len(self.calls)
assert 'template - test.entity - hello - world' == \
self.calls[0].data['some']
hass.states.async_set('test.entity', 'world')
await hass.async_block_till_done()
hass.states.async_set('test.entity', 'hello')
await hass.async_block_till_done()
assert 1 == len(calls)
assert 'template - test.entity - hello - world' == \
calls[0].data['some']

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

@ -1,47 +1,216 @@
"""The tests for the time automation."""
from datetime import timedelta
import unittest
from unittest.mock import patch
from homeassistant.core import callback
from homeassistant.setup import setup_component
import pytest
from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util
import homeassistant.components.automation as automation
from tests.common import (
fire_time_changed, get_test_home_assistant, assert_setup_component,
mock_component)
async_fire_time_changed, assert_setup_component, mock_component)
from tests.components.automation import common
from tests.common import async_mock_service
# pylint: disable=invalid-name
class TestAutomationTime(unittest.TestCase):
"""Test the event automation."""
@pytest.fixture
def calls(hass):
"""Track calls to a mock serivce."""
return async_mock_service(hass, 'test', 'automation')
def setUp(self):
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
mock_component(self.hass, 'group')
self.calls = []
@callback
def record_call(service):
"""Record calls."""
self.calls.append(service)
@pytest.fixture(autouse=True)
def setup_comp(hass):
"""Initialize components."""
mock_component(hass, 'group')
self.hass.services.register('test', 'automation', record_call)
def tearDown(self):
"""Stop everything that was started."""
self.hass.stop()
async def test_if_fires_when_hour_matches(hass, calls):
"""Test for firing if hour is matching."""
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'time',
'hours': 0,
},
'action': {
'service': 'test.automation'
}
}
})
def test_if_fires_when_hour_matches(self):
"""Test for firing if hour is matching."""
assert setup_component(self.hass, automation.DOMAIN, {
async_fire_time_changed(hass, dt_util.utcnow().replace(hour=0))
await hass.async_block_till_done()
assert 1 == len(calls)
await common.async_turn_off(hass)
await hass.async_block_till_done()
async_fire_time_changed(hass, dt_util.utcnow().replace(hour=0))
await hass.async_block_till_done()
assert 1 == len(calls)
async def test_if_fires_when_minute_matches(hass, calls):
"""Test for firing if minutes are matching."""
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'time',
'minutes': 0,
},
'action': {
'service': 'test.automation'
}
}
})
async_fire_time_changed(hass, dt_util.utcnow().replace(minute=0))
await hass.async_block_till_done()
assert 1 == len(calls)
async def test_if_fires_when_second_matches(hass, calls):
"""Test for firing if seconds are matching."""
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'time',
'seconds': 0,
},
'action': {
'service': 'test.automation'
}
}
})
async_fire_time_changed(hass, dt_util.utcnow().replace(second=0))
await hass.async_block_till_done()
assert 1 == len(calls)
async def test_if_fires_when_all_matches(hass, calls):
"""Test for firing if everything matches."""
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'time',
'hours': 1,
'minutes': 2,
'seconds': 3,
},
'action': {
'service': 'test.automation'
}
}
})
async_fire_time_changed(hass, dt_util.utcnow().replace(
hour=1, minute=2, second=3))
await hass.async_block_till_done()
assert 1 == len(calls)
async def test_if_fires_periodic_seconds(hass, calls):
"""Test for firing periodically every second."""
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'time',
'seconds': "/2",
},
'action': {
'service': 'test.automation'
}
}
})
async_fire_time_changed(hass, dt_util.utcnow().replace(
hour=0, minute=0, second=2))
await hass.async_block_till_done()
assert 1 == len(calls)
async def test_if_fires_periodic_minutes(hass, calls):
"""Test for firing periodically every minute."""
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'time',
'minutes': "/2",
},
'action': {
'service': 'test.automation'
}
}
})
async_fire_time_changed(hass, dt_util.utcnow().replace(
hour=0, minute=2, second=0))
await hass.async_block_till_done()
assert 1 == len(calls)
async def test_if_fires_periodic_hours(hass, calls):
"""Test for firing periodically every hour."""
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'time',
'hours': "/2",
},
'action': {
'service': 'test.automation'
}
}
})
async_fire_time_changed(hass, dt_util.utcnow().replace(
hour=2, minute=0, second=0))
await hass.async_block_till_done()
assert 1 == len(calls)
async def test_if_fires_using_at(hass, calls):
"""Test for firing at."""
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'time',
'at': '5:00:00',
},
'action': {
'service': 'test.automation',
'data_template': {
'some': '{{ trigger.platform }} - '
'{{ trigger.now.hour }}'
},
}
}
})
async_fire_time_changed(hass, dt_util.utcnow().replace(
hour=5, minute=0, second=0))
await hass.async_block_till_done()
assert 1 == len(calls)
assert 'time - 5' == calls[0].data['some']
async def test_if_not_working_if_no_values_in_conf_provided(hass, calls):
"""Test for failure if no configuration."""
with assert_setup_component(0):
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'time',
'hours': 0,
},
'action': {
'service': 'test.automation'
@ -49,24 +218,25 @@ class TestAutomationTime(unittest.TestCase):
}
})
fire_time_changed(self.hass, dt_util.utcnow().replace(hour=0))
self.hass.block_till_done()
assert 1 == len(self.calls)
async_fire_time_changed(hass, dt_util.utcnow().replace(
hour=5, minute=0, second=0))
common.turn_off(self.hass)
self.hass.block_till_done()
await hass.async_block_till_done()
assert 0 == len(calls)
fire_time_changed(self.hass, dt_util.utcnow().replace(hour=0))
self.hass.block_till_done()
assert 1 == len(self.calls)
def test_if_fires_when_minute_matches(self):
"""Test for firing if minutes are matching."""
assert setup_component(self.hass, automation.DOMAIN, {
async def test_if_not_fires_using_wrong_at(hass, calls):
"""YAML translates time values to total seconds.
This should break the before rule.
"""
with assert_setup_component(0):
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'time',
'minutes': 0,
'at': 3605,
# Total seconds. Hour = 3600 second
},
'action': {
'service': 'test.automation'
@ -74,328 +244,162 @@ class TestAutomationTime(unittest.TestCase):
}
})
fire_time_changed(self.hass, dt_util.utcnow().replace(minute=0))
async_fire_time_changed(hass, dt_util.utcnow().replace(
hour=1, minute=0, second=5))
self.hass.block_till_done()
assert 1 == len(self.calls)
await hass.async_block_till_done()
assert 0 == len(calls)
def test_if_fires_when_second_matches(self):
"""Test for firing if seconds are matching."""
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'time',
'seconds': 0,
},
'action': {
'service': 'test.automation'
}
async def test_if_action_before(hass, calls):
"""Test for if action before."""
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'event',
'event_type': 'test_event'
},
'condition': {
'condition': 'time',
'before': '10:00',
},
'action': {
'service': 'test.automation'
}
})
}
})
fire_time_changed(self.hass, dt_util.utcnow().replace(second=0))
before_10 = dt_util.now().replace(hour=8)
after_10 = dt_util.now().replace(hour=14)
self.hass.block_till_done()
assert 1 == len(self.calls)
with patch('homeassistant.helpers.condition.dt_util.now',
return_value=before_10):
hass.bus.async_fire('test_event')
await hass.async_block_till_done()
def test_if_fires_when_all_matches(self):
"""Test for firing if everything matches."""
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'time',
'hours': 1,
'minutes': 2,
'seconds': 3,
},
'action': {
'service': 'test.automation'
}
assert 1 == len(calls)
with patch('homeassistant.helpers.condition.dt_util.now',
return_value=after_10):
hass.bus.async_fire('test_event')
await hass.async_block_till_done()
assert 1 == len(calls)
async def test_if_action_after(hass, calls):
"""Test for if action after."""
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'event',
'event_type': 'test_event'
},
'condition': {
'condition': 'time',
'after': '10:00',
},
'action': {
'service': 'test.automation'
}
})
}
})
fire_time_changed(self.hass, dt_util.utcnow().replace(
hour=1, minute=2, second=3))
before_10 = dt_util.now().replace(hour=8)
after_10 = dt_util.now().replace(hour=14)
self.hass.block_till_done()
assert 1 == len(self.calls)
with patch('homeassistant.helpers.condition.dt_util.now',
return_value=before_10):
hass.bus.async_fire('test_event')
await hass.async_block_till_done()
def test_if_fires_periodic_seconds(self):
"""Test for firing periodically every second."""
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'time',
'seconds': "/2",
},
'action': {
'service': 'test.automation'
}
assert 0 == len(calls)
with patch('homeassistant.helpers.condition.dt_util.now',
return_value=after_10):
hass.bus.async_fire('test_event')
await hass.async_block_till_done()
assert 1 == len(calls)
async def test_if_action_one_weekday(hass, calls):
"""Test for if action with one weekday."""
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'event',
'event_type': 'test_event'
},
'condition': {
'condition': 'time',
'weekday': 'mon',
},
'action': {
'service': 'test.automation'
}
})
}
})
fire_time_changed(self.hass, dt_util.utcnow().replace(
hour=0, minute=0, second=2))
days_past_monday = dt_util.now().weekday()
monday = dt_util.now() - timedelta(days=days_past_monday)
tuesday = monday + timedelta(days=1)
self.hass.block_till_done()
assert 1 == len(self.calls)
with patch('homeassistant.helpers.condition.dt_util.now',
return_value=monday):
hass.bus.async_fire('test_event')
await hass.async_block_till_done()
def test_if_fires_periodic_minutes(self):
"""Test for firing periodically every minute."""
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'time',
'minutes': "/2",
},
'action': {
'service': 'test.automation'
}
assert 1 == len(calls)
with patch('homeassistant.helpers.condition.dt_util.now',
return_value=tuesday):
hass.bus.async_fire('test_event')
await hass.async_block_till_done()
assert 1 == len(calls)
async def test_if_action_list_weekday(hass, calls):
"""Test for action with a list of weekdays."""
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'event',
'event_type': 'test_event'
},
'condition': {
'condition': 'time',
'weekday': ['mon', 'tue'],
},
'action': {
'service': 'test.automation'
}
})
}
})
fire_time_changed(self.hass, dt_util.utcnow().replace(
hour=0, minute=2, second=0))
days_past_monday = dt_util.now().weekday()
monday = dt_util.now() - timedelta(days=days_past_monday)
tuesday = monday + timedelta(days=1)
wednesday = tuesday + timedelta(days=1)
self.hass.block_till_done()
assert 1 == len(self.calls)
with patch('homeassistant.helpers.condition.dt_util.now',
return_value=monday):
hass.bus.async_fire('test_event')
await hass.async_block_till_done()
def test_if_fires_periodic_hours(self):
"""Test for firing periodically every hour."""
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'time',
'hours': "/2",
},
'action': {
'service': 'test.automation'
}
}
})
assert 1 == len(calls)
fire_time_changed(self.hass, dt_util.utcnow().replace(
hour=2, minute=0, second=0))
with patch('homeassistant.helpers.condition.dt_util.now',
return_value=tuesday):
hass.bus.async_fire('test_event')
await hass.async_block_till_done()
self.hass.block_till_done()
assert 1 == len(self.calls)
assert 2 == len(calls)
def test_if_fires_using_at(self):
"""Test for firing at."""
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'time',
'at': '5:00:00',
},
'action': {
'service': 'test.automation',
'data_template': {
'some': '{{ trigger.platform }} - '
'{{ trigger.now.hour }}'
},
}
}
})
with patch('homeassistant.helpers.condition.dt_util.now',
return_value=wednesday):
hass.bus.async_fire('test_event')
await hass.async_block_till_done()
fire_time_changed(self.hass, dt_util.utcnow().replace(
hour=5, minute=0, second=0))
self.hass.block_till_done()
assert 1 == len(self.calls)
assert 'time - 5' == self.calls[0].data['some']
def test_if_not_working_if_no_values_in_conf_provided(self):
"""Test for failure if no configuration."""
with assert_setup_component(0):
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'time',
},
'action': {
'service': 'test.automation'
}
}
})
fire_time_changed(self.hass, dt_util.utcnow().replace(
hour=5, minute=0, second=0))
self.hass.block_till_done()
assert 0 == len(self.calls)
def test_if_not_fires_using_wrong_at(self):
"""YAML translates time values to total seconds.
This should break the before rule.
"""
with assert_setup_component(0):
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'time',
'at': 3605,
# Total seconds. Hour = 3600 second
},
'action': {
'service': 'test.automation'
}
}
})
fire_time_changed(self.hass, dt_util.utcnow().replace(
hour=1, minute=0, second=5))
self.hass.block_till_done()
assert 0 == len(self.calls)
def test_if_action_before(self):
"""Test for if action before."""
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'event',
'event_type': 'test_event'
},
'condition': {
'condition': 'time',
'before': '10:00',
},
'action': {
'service': 'test.automation'
}
}
})
before_10 = dt_util.now().replace(hour=8)
after_10 = dt_util.now().replace(hour=14)
with patch('homeassistant.helpers.condition.dt_util.now',
return_value=before_10):
self.hass.bus.fire('test_event')
self.hass.block_till_done()
assert 1 == len(self.calls)
with patch('homeassistant.helpers.condition.dt_util.now',
return_value=after_10):
self.hass.bus.fire('test_event')
self.hass.block_till_done()
assert 1 == len(self.calls)
def test_if_action_after(self):
"""Test for if action after."""
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'event',
'event_type': 'test_event'
},
'condition': {
'condition': 'time',
'after': '10:00',
},
'action': {
'service': 'test.automation'
}
}
})
before_10 = dt_util.now().replace(hour=8)
after_10 = dt_util.now().replace(hour=14)
with patch('homeassistant.helpers.condition.dt_util.now',
return_value=before_10):
self.hass.bus.fire('test_event')
self.hass.block_till_done()
assert 0 == len(self.calls)
with patch('homeassistant.helpers.condition.dt_util.now',
return_value=after_10):
self.hass.bus.fire('test_event')
self.hass.block_till_done()
assert 1 == len(self.calls)
def test_if_action_one_weekday(self):
"""Test for if action with one weekday."""
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'event',
'event_type': 'test_event'
},
'condition': {
'condition': 'time',
'weekday': 'mon',
},
'action': {
'service': 'test.automation'
}
}
})
days_past_monday = dt_util.now().weekday()
monday = dt_util.now() - timedelta(days=days_past_monday)
tuesday = monday + timedelta(days=1)
with patch('homeassistant.helpers.condition.dt_util.now',
return_value=monday):
self.hass.bus.fire('test_event')
self.hass.block_till_done()
assert 1 == len(self.calls)
with patch('homeassistant.helpers.condition.dt_util.now',
return_value=tuesday):
self.hass.bus.fire('test_event')
self.hass.block_till_done()
assert 1 == len(self.calls)
def test_if_action_list_weekday(self):
"""Test for action with a list of weekdays."""
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'event',
'event_type': 'test_event'
},
'condition': {
'condition': 'time',
'weekday': ['mon', 'tue'],
},
'action': {
'service': 'test.automation'
}
}
})
days_past_monday = dt_util.now().weekday()
monday = dt_util.now() - timedelta(days=days_past_monday)
tuesday = monday + timedelta(days=1)
wednesday = tuesday + timedelta(days=1)
with patch('homeassistant.helpers.condition.dt_util.now',
return_value=monday):
self.hass.bus.fire('test_event')
self.hass.block_till_done()
assert 1 == len(self.calls)
with patch('homeassistant.helpers.condition.dt_util.now',
return_value=tuesday):
self.hass.bus.fire('test_event')
self.hass.block_till_done()
assert 2 == len(self.calls)
with patch('homeassistant.helpers.condition.dt_util.now',
return_value=wednesday):
self.hass.bus.fire('test_event')
self.hass.block_till_done()
assert 2 == len(self.calls)
assert 2 == len(calls)

340
tests/components/automation/test_zone.py
View file

@ -1,218 +1,212 @@
"""The tests for the location automation."""
import unittest
import pytest
from homeassistant.core import Context, callback
from homeassistant.setup import setup_component
from homeassistant.core import Context
from homeassistant.setup import async_setup_component
from homeassistant.components import automation, zone
from tests.common import get_test_home_assistant, mock_component
from tests.components.automation import common
from tests.common import async_mock_service, mock_component
# pylint: disable=invalid-name
class TestAutomationZone(unittest.TestCase):
"""Test the event automation."""
@pytest.fixture
def calls(hass):
"""Track calls to a mock serivce."""
return async_mock_service(hass, 'test', 'automation')
def setUp(self):
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
mock_component(self.hass, 'group')
assert setup_component(self.hass, zone.DOMAIN, {
@pytest.fixture(autouse=True)
def setup_comp(hass):
"""Initialize components."""
mock_component(hass, 'group')
hass.loop.run_until_complete(async_setup_component(hass, zone.DOMAIN, {
'zone': {
'name': 'test',
'latitude': 32.880837,
'longitude': -117.237561,
'radius': 250,
}
})
}))
self.calls = []
@callback
def record_call(service):
"""Record calls."""
self.calls.append(service)
async def test_if_fires_on_zone_enter(hass, calls):
"""Test for firing on zone enter."""
context = Context()
hass.states.async_set('test.entity', 'hello', {
'latitude': 32.881011,
'longitude': -117.234758
})
await hass.async_block_till_done()
self.hass.services.register('test', 'automation', record_call)
def tearDown(self):
"""Stop everything that was started."""
self.hass.stop()
def test_if_fires_on_zone_enter(self):
"""Test for firing on zone enter."""
context = Context()
self.hass.states.set('test.entity', 'hello', {
'latitude': 32.881011,
'longitude': -117.234758
})
self.hass.block_till_done()
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'zone',
'entity_id': 'test.entity',
'zone': 'zone.test',
'event': 'enter',
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'zone',
'entity_id': 'test.entity',
'zone': 'zone.test',
'event': 'enter',
},
'action': {
'service': 'test.automation',
'data_template': {
'some': '{{ trigger.%s }}' % '}} - {{ trigger.'.join((
'platform', 'entity_id',
'from_state.state', 'to_state.state',
'zone.name'))
},
'action': {
'service': 'test.automation',
'data_template': {
'some': '{{ trigger.%s }}' % '}} - {{ trigger.'.join((
'platform', 'entity_id',
'from_state.state', 'to_state.state',
'zone.name'))
},
}
}
})
}
})
self.hass.states.set('test.entity', 'hello', {
'latitude': 32.880586,
'longitude': -117.237564
}, context=context)
self.hass.block_till_done()
hass.states.async_set('test.entity', 'hello', {
'latitude': 32.880586,
'longitude': -117.237564
}, context=context)
await hass.async_block_till_done()
assert 1 == len(self.calls)
assert self.calls[0].context is context
assert 'zone - test.entity - hello - hello - test' == \
self.calls[0].data['some']
assert 1 == len(calls)
assert calls[0].context is context
assert 'zone - test.entity - hello - hello - test' == \
calls[0].data['some']
# Set out of zone again so we can trigger call
self.hass.states.set('test.entity', 'hello', {
'latitude': 32.881011,
'longitude': -117.234758
})
self.hass.block_till_done()
# Set out of zone again so we can trigger call
hass.states.async_set('test.entity', 'hello', {
'latitude': 32.881011,
'longitude': -117.234758
})
await hass.async_block_till_done()
common.turn_off(self.hass)
self.hass.block_till_done()
await common.async_turn_off(hass)
await hass.async_block_till_done()
self.hass.states.set('test.entity', 'hello', {
'latitude': 32.880586,
'longitude': -117.237564
})
self.hass.block_till_done()
hass.states.async_set('test.entity', 'hello', {
'latitude': 32.880586,
'longitude': -117.237564
})
await hass.async_block_till_done()
assert 1 == len(self.calls)
assert 1 == len(calls)
def test_if_not_fires_for_enter_on_zone_leave(self):
"""Test for not firing on zone leave."""
self.hass.states.set('test.entity', 'hello', {
'latitude': 32.880586,
'longitude': -117.237564
})
self.hass.block_till_done()
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'zone',
'entity_id': 'test.entity',
'zone': 'zone.test',
'event': 'enter',
},
'action': {
'service': 'test.automation',
}
async def test_if_not_fires_for_enter_on_zone_leave(hass, calls):
"""Test for not firing on zone leave."""
hass.states.async_set('test.entity', 'hello', {
'latitude': 32.880586,
'longitude': -117.237564
})
await hass.async_block_till_done()
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'zone',
'entity_id': 'test.entity',
'zone': 'zone.test',
'event': 'enter',
},
'action': {
'service': 'test.automation',
}
})
}
})
self.hass.states.set('test.entity', 'hello', {
'latitude': 32.881011,
'longitude': -117.234758
})
self.hass.block_till_done()
hass.states.async_set('test.entity', 'hello', {
'latitude': 32.881011,
'longitude': -117.234758
})
await hass.async_block_till_done()
assert 0 == len(self.calls)
assert 0 == len(calls)
def test_if_fires_on_zone_leave(self):
"""Test for firing on zone leave."""
self.hass.states.set('test.entity', 'hello', {
'latitude': 32.880586,
'longitude': -117.237564
})
self.hass.block_till_done()
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'zone',
'entity_id': 'test.entity',
'zone': 'zone.test',
'event': 'leave',
},
'action': {
'service': 'test.automation',
}
async def test_if_fires_on_zone_leave(hass, calls):
"""Test for firing on zone leave."""
hass.states.async_set('test.entity', 'hello', {
'latitude': 32.880586,
'longitude': -117.237564
})
await hass.async_block_till_done()
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'zone',
'entity_id': 'test.entity',
'zone': 'zone.test',
'event': 'leave',
},
'action': {
'service': 'test.automation',
}
})
}
})
self.hass.states.set('test.entity', 'hello', {
'latitude': 32.881011,
'longitude': -117.234758
})
self.hass.block_till_done()
hass.states.async_set('test.entity', 'hello', {
'latitude': 32.881011,
'longitude': -117.234758
})
await hass.async_block_till_done()
assert 1 == len(self.calls)
assert 1 == len(calls)
def test_if_not_fires_for_leave_on_zone_enter(self):
"""Test for not firing on zone enter."""
self.hass.states.set('test.entity', 'hello', {
'latitude': 32.881011,
'longitude': -117.234758
})
self.hass.block_till_done()
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'zone',
'entity_id': 'test.entity',
'zone': 'zone.test',
'event': 'leave',
},
'action': {
'service': 'test.automation',
}
async def test_if_not_fires_for_leave_on_zone_enter(hass, calls):
"""Test for not firing on zone enter."""
hass.states.async_set('test.entity', 'hello', {
'latitude': 32.881011,
'longitude': -117.234758
})
await hass.async_block_till_done()
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'zone',
'entity_id': 'test.entity',
'zone': 'zone.test',
'event': 'leave',
},
'action': {
'service': 'test.automation',
}
})
}
})
self.hass.states.set('test.entity', 'hello', {
'latitude': 32.880586,
'longitude': -117.237564
})
self.hass.block_till_done()
hass.states.async_set('test.entity', 'hello', {
'latitude': 32.880586,
'longitude': -117.237564
})
await hass.async_block_till_done()
assert 0 == len(self.calls)
assert 0 == len(calls)
def test_zone_condition(self):
"""Test for zone condition."""
self.hass.states.set('test.entity', 'hello', {
'latitude': 32.880586,
'longitude': -117.237564
})
self.hass.block_till_done()
assert setup_component(self.hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'event',
'event_type': 'test_event'
},
'condition': {
'condition': 'zone',
'entity_id': 'test.entity',
'zone': 'zone.test',
},
'action': {
'service': 'test.automation',
}
async def test_zone_condition(hass, calls):
"""Test for zone condition."""
hass.states.async_set('test.entity', 'hello', {
'latitude': 32.880586,
'longitude': -117.237564
})
await hass.async_block_till_done()
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: {
'trigger': {
'platform': 'event',
'event_type': 'test_event'
},
'condition': {
'condition': 'zone',
'entity_id': 'test.entity',
'zone': 'zone.test',
},
'action': {
'service': 'test.automation',
}
})
}
})
self.hass.bus.fire('test_event')
self.hass.block_till_done()
assert 1 == len(self.calls)
hass.bus.async_fire('test_event')
await hass.async_block_till_done()
assert 1 == len(calls)