core/tests/helpers/test_service.py

"""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