"""Test service helpers."""
import asyncio
from collections import OrderedDict
from copy import deepcopy
import unittest
from unittest.mock import Mock, patch
import pytest
# To prevent circular import when running just this file
import homeassistant.components # noqa
from homeassistant import core as ha, loader, exceptions
from homeassistant.const import STATE_ON, STATE_OFF, ATTR_ENTITY_ID
from homeassistant.helpers import service, template
from homeassistant.setup import async_setup_component
import homeassistant.helpers.config_validation as cv
from homeassistant.auth.permissions import PolicyPermissions
from tests.common import get_test_home_assistant, mock_service, mock_coro
@pytest.fixture
def mock_service_platform_call():
"""Mock service platform call."""
with patch('homeassistant.helpers.service._handle_service_platform_call',
side_effect=lambda *args: mock_coro()) as mock_call:
yield mock_call
@pytest.fixture
def mock_entities():
"""Return mock entities in an ordered dict."""
kitchen = Mock(
entity_id='light.kitchen',
available=True,
should_poll=False,
)
living_room = Mock(
entity_id='light.living_room',
available=True,
should_poll=False,
)
entities = OrderedDict()
entities[kitchen.entity_id] = kitchen
entities[living_room.entity_id] = living_room
return entities
class TestServiceHelpers(unittest.TestCase):
"""Test the Home Assistant service helpers."""
def setUp(self): # pylint: disable=invalid-name
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
self.calls = mock_service(self.hass, 'test_domain', 'test_service')
def tearDown(self): # pylint: disable=invalid-name
"""Stop down everything that was started."""
self.hass.stop()
def test_template_service_call(self):
"""Test service call with templating."""
config = {
'service_template': '{{ \'test_domain.test_service\' }}',
'entity_id': 'hello.world',
'data_template': {
'hello': '{{ \'goodbye\' }}',
'data': {
'value': '{{ \'complex\' }}',
'simple': 'simple'
},
'list': ['{{ \'list\' }}', '2'],
},
}
service.call_from_config(self.hass, config)
self.hass.block_till_done()
assert 'goodbye' == self.calls[0].data['hello']
assert 'complex' == self.calls[0].data['data']['value']
assert 'simple' == self.calls[0].data['data']['simple']
assert 'list' == self.calls[0].data['list'][0]
def test_passing_variables_to_templates(self):
"""Test passing variables to templates."""
config = {
'service_template': '{{ var_service }}',
'entity_id': 'hello.world',
'data_template': {
'hello': '{{ var_data }}',
},
}
service.call_from_config(self.hass, config, variables={
'var_service': 'test_domain.test_service',
'var_data': 'goodbye',
})
self.hass.block_till_done()
assert 'goodbye' == self.calls[0].data['hello']
def test_bad_template(self):
"""Test passing bad template."""
config = {
'service_template': '{{ var_service }}',
'entity_id': 'hello.world',
'data_template': {
'hello': '{{ states + unknown_var }}'
}
}
service.call_from_config(self.hass, config, variables={
'var_service': 'test_domain.test_service',
'var_data': 'goodbye',
})
self.hass.block_till_done()
assert len(self.calls) == 0
def test_split_entity_string(self):
"""Test splitting of entity string."""
service.call_from_config(self.hass, {
'service': 'test_domain.test_service',
'entity_id': 'hello.world, sensor.beer'
})
self.hass.block_till_done()
assert ['hello.world', 'sensor.beer'] == \
self.calls[-1].data.get('entity_id')
def test_not_mutate_input(self):
"""Test for immutable input."""
config = cv.SERVICE_SCHEMA({
'service': 'test_domain.test_service',
'entity_id': 'hello.world, sensor.beer',
'data': {
'hello': 1,
},
'data_template': {
'nested': {
'value': '{{ 1 + 1 }}'
}
}
})
orig = deepcopy(config)
# Only change after call is each template getting hass attached
template.attach(self.hass, orig)
service.call_from_config(self.hass, config, validate_config=False)
assert orig == config
@patch('homeassistant.helpers.service._LOGGER.error')
def test_fail_silently_if_no_service(self, mock_log):
"""Test failing if service is missing."""
service.call_from_config(self.hass, None)
assert 1 == mock_log.call_count
service.call_from_config(self.hass, {})
assert 2 == mock_log.call_count
service.call_from_config(self.hass, {
'service': 'invalid'
})
assert 3 == mock_log.call_count
def test_extract_entity_ids(self):
"""Test extract_entity_ids method."""
self.hass.states.set('light.Bowl', STATE_ON)
self.hass.states.set('light.Ceiling', STATE_OFF)
self.hass.states.set('light.Kitchen', STATE_OFF)
loader.get_component(self.hass, 'group').Group.create_group(
self.hass, 'test', ['light.Ceiling', 'light.Kitchen'])
call = ha.ServiceCall('light', 'turn_on',
{ATTR_ENTITY_ID: 'light.Bowl'})
assert ['light.bowl'] == \
service.extract_entity_ids(self.hass, call)
call = ha.ServiceCall('light', 'turn_on',
{ATTR_ENTITY_ID: 'group.test'})
assert ['light.ceiling', 'light.kitchen'] == \
service.extract_entity_ids(self.hass, call)
assert ['group.test'] == service.extract_entity_ids(
self.hass, call, expand_group=False)
@asyncio.coroutine
def test_async_get_all_descriptions(hass):
"""Test async_get_all_descriptions."""
group = loader.get_component(hass, 'group')
group_config = {group.DOMAIN: {}}
yield from async_setup_component(hass, group.DOMAIN, group_config)
descriptions = yield from service.async_get_all_descriptions(hass)
assert len(descriptions) == 1
assert 'description' in descriptions['group']['reload']
assert 'fields' in descriptions['group']['reload']
logger = loader.get_component(hass, 'logger')
logger_config = {logger.DOMAIN: {}}
yield from async_setup_component(hass, logger.DOMAIN, logger_config)
descriptions = yield from service.async_get_all_descriptions(hass)
assert len(descriptions) == 2
assert 'description' in descriptions[logger.DOMAIN]['set_level']
assert 'fields' in descriptions[logger.DOMAIN]['set_level']
async def test_call_context_user_not_exist(hass):
"""Check we don't allow deleted users to do things."""
with pytest.raises(exceptions.UnknownUser) as err:
await service.entity_service_call(hass, [], Mock(), ha.ServiceCall(
'test_domain', 'test_service', context=ha.Context(
user_id='non-existing')))
assert err.value.context.user_id == 'non-existing'
async def test_call_context_target_all(hass, mock_service_platform_call,
mock_entities):
"""Check we only target allowed entities if targetting all."""
with patch('homeassistant.auth.AuthManager.async_get_user',
return_value=mock_coro(Mock(permissions=PolicyPermissions({
'entities': {
'entity_ids': {
'light.kitchen': True
}
}
}, None)))):
await service.entity_service_call(hass, [
Mock(entities=mock_entities)
], Mock(), ha.ServiceCall('test_domain', 'test_service',
context=ha.Context(user_id='mock-id')))
assert len(mock_service_platform_call.mock_calls) == 1
entities = mock_service_platform_call.mock_calls[0][1][2]
assert entities == [mock_entities['light.kitchen']]
async def test_call_context_target_specific(hass, mock_service_platform_call,
mock_entities):
"""Check targeting specific entities."""
with patch('homeassistant.auth.AuthManager.async_get_user',
return_value=mock_coro(Mock(permissions=PolicyPermissions({
'entities': {
'entity_ids': {
'light.kitchen': True
}
}
}, None)))):
await service.entity_service_call(hass, [
Mock(entities=mock_entities)
], Mock(), ha.ServiceCall('test_domain', 'test_service', {
'entity_id': 'light.kitchen'
}, context=ha.Context(user_id='mock-id')))
assert len(mock_service_platform_call.mock_calls) == 1
entities = mock_service_platform_call.mock_calls[0][1][2]
assert entities == [mock_entities['light.kitchen']]
async def test_call_context_target_specific_no_auth(
hass, mock_service_platform_call, mock_entities):
"""Check targeting specific entities without auth."""
with pytest.raises(exceptions.Unauthorized) as err:
with patch('homeassistant.auth.AuthManager.async_get_user',
return_value=mock_coro(Mock(
permissions=PolicyPermissions({}, None)))):
await service.entity_service_call(hass, [
Mock(entities=mock_entities)
], Mock(), ha.ServiceCall('test_domain', 'test_service', {
'entity_id': 'light.kitchen'
}, context=ha.Context(user_id='mock-id')))
assert err.value.context.user_id == 'mock-id'
assert err.value.entity_id == 'light.kitchen'
async def test_call_no_context_target_all(hass, mock_service_platform_call,
mock_entities):
"""Check we target all if no user context given."""
await service.entity_service_call(hass, [
Mock(entities=mock_entities)
], Mock(), ha.ServiceCall('test_domain', 'test_service'))
assert len(mock_service_platform_call.mock_calls) == 1
entities = mock_service_platform_call.mock_calls[0][1][2]
assert entities == list(mock_entities.values())
async def test_call_no_context_target_specific(
hass, mock_service_platform_call, mock_entities):
"""Check we can target specified entities."""
await service.entity_service_call(hass, [
Mock(entities=mock_entities)
], Mock(), ha.ServiceCall('test_domain', 'test_service', {
'entity_id': ['light.kitchen', 'light.non-existing']
}))
assert len(mock_service_platform_call.mock_calls) == 1
entities = mock_service_platform_call.mock_calls[0][1][2]
assert entities == [mock_entities['light.kitchen']]
async def test_call_with_match_all(hass, mock_service_platform_call,
mock_entities, caplog):
"""Check we only target allowed entities if targetting all."""
await service.entity_service_call(hass, [
Mock(entities=mock_entities)
], Mock(), ha.ServiceCall('test_domain', 'test_service', {
'entity_id': 'all'
}))
assert len(mock_service_platform_call.mock_calls) == 1
entities = mock_service_platform_call.mock_calls[0][1][2]
assert entities == [
mock_entities['light.kitchen'], mock_entities['light.living_room']]
assert ('Not passing an entity ID to a service to target '
'all entities is deprecated') not in caplog.text
async def test_call_with_omit_entity_id(hass, mock_service_platform_call,
mock_entities, caplog):
"""Check we only target allowed entities if targetting all."""
await service.entity_service_call(hass, [
Mock(entities=mock_entities)
], Mock(), ha.ServiceCall('test_domain', 'test_service'))
assert len(mock_service_platform_call.mock_calls) == 1
entities = mock_service_platform_call.mock_calls[0][1][2]
assert entities == [
mock_entities['light.kitchen'], mock_entities['light.living_room']]
assert ('Not passing an entity ID to a service to target '
'all entities is deprecated') in caplog.text