/
test_service.py
308 lines (249 loc) · 11 KB
/
test_service.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
"""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
}
}
})))):
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
}
}
})))):
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({})))):
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']]