Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-12-24 04:50:05 +00:00
Code Activity

Migrate tests to pytest (#23544)

Browse Source 
* Migrate tests to pytest

* Fixup

* Use loop fixture in test_check_config

* Lint
This commit is contained in:
Erik Montnemery
2019-04-30 18:20:38 +02:00
committed by Paulus Schoutsen
parent d71424f285
commit 407e0c58f9
25 changed files with 4744 additions and 4910 deletions
Download Patch File Download Diff File Expand all files Collapse all files

367
tests/helpers/test_entity_component.py
View File

@@ -1,25 +1,24 @@
"""The tests for the Entity component helper."""
# pylint: disable=protected-access
import asyncio
from collections import OrderedDict
import logging
import unittest
from unittest.mock import patch, Mock
from datetime import timedelta
import asynctest
import pytest
import homeassistant.core as ha
from homeassistant.exceptions import PlatformNotReady
from homeassistant.components import group
from homeassistant.helpers.entity_component import EntityComponent
from homeassistant.setup import setup_component, async_setup_component
from homeassistant.setup import async_setup_component
from homeassistant.helpers import discovery
import homeassistant.util.dt as dt_util
from tests.common import (
get_test_home_assistant, MockPlatform, MockModule, mock_coro,
MockPlatform, MockModule, mock_coro,
async_fire_time_changed, MockEntity, MockConfigEntry,
mock_entity_platform, mock_integration)
@@ -27,178 +26,169 @@ _LOGGER = logging.getLogger(__name__)
DOMAIN = "test_domain"
class TestHelpersEntityComponent(unittest.TestCase):
"""Test homeassistant.helpers.entity_component module."""
async def test_setting_up_group(hass):
"""Set up the setting of a group."""
assert await async_setup_component(hass, 'group', {'group': {}})
component = EntityComponent(_LOGGER, DOMAIN, hass,
group_name='everyone')
def setUp(self): # pylint: disable=invalid-name
"""Initialize a test Home Assistant instance."""
self.hass = get_test_home_assistant()
# No group after setup
assert len(hass.states.async_entity_ids()) == 0
def tearDown(self): # pylint: disable=invalid-name
"""Clean up the test Home Assistant instance."""
self.hass.stop()
await component.async_add_entities([MockEntity()])
await hass.async_block_till_done()
def test_setting_up_group(self):
"""Set up the setting of a group."""
setup_component(self.hass, 'group', {'group': {}})
component = EntityComponent(_LOGGER, DOMAIN, self.hass,
group_name='everyone')
# group exists
assert len(hass.states.async_entity_ids()) == 2
assert hass.states.async_entity_ids('group') == ['group.everyone']
# No group after setup
assert len(self.hass.states.entity_ids()) == 0
grp = hass.states.get('group.everyone')
component.add_entities([MockEntity()])
self.hass.block_till_done()
assert grp.attributes.get('entity_id') == \
('test_domain.unnamed_device',)
# group exists
assert len(self.hass.states.entity_ids()) == 2
assert self.hass.states.entity_ids('group') == ['group.everyone']
# group extended
await component.async_add_entities([MockEntity(name='goodbye')])
await hass.async_block_till_done()
group = self.hass.states.get('group.everyone')
assert len(hass.states.async_entity_ids()) == 3
grp = hass.states.get('group.everyone')
assert group.attributes.get('entity_id') == \
('test_domain.unnamed_device',)
# group extended
component.add_entities([MockEntity(name='goodbye')])
self.hass.block_till_done()
assert len(self.hass.states.entity_ids()) == 3
group = self.hass.states.get('group.everyone')
# Ordered in order of added to the group
assert group.attributes.get('entity_id') == \
('test_domain.goodbye', 'test_domain.unnamed_device')
def test_setup_loads_platforms(self):
"""Test the loading of the platforms."""
component_setup = Mock(return_value=True)
platform_setup = Mock(return_value=None)
mock_integration(self.hass,
MockModule('test_component', setup=component_setup))
# mock the dependencies
mock_integration(self.hass,
MockModule('mod2', dependencies=['test_component']))
mock_entity_platform(self.hass, 'test_domain.mod2',
MockPlatform(platform_setup))
component = EntityComponent(_LOGGER, DOMAIN, self.hass)
assert not component_setup.called
assert not platform_setup.called
component.setup({
DOMAIN: {
'platform': 'mod2',
}
})
self.hass.block_till_done()
assert component_setup.called
assert platform_setup.called
def test_setup_recovers_when_setup_raises(self):
"""Test the setup if exceptions are happening."""
platform1_setup = Mock(side_effect=Exception('Broken'))
platform2_setup = Mock(return_value=None)
mock_entity_platform(self.hass, 'test_domain.mod1',
MockPlatform(platform1_setup))
mock_entity_platform(self.hass, 'test_domain.mod2',
MockPlatform(platform2_setup))
component = EntityComponent(_LOGGER, DOMAIN, self.hass)
assert not platform1_setup.called
assert not platform2_setup.called
component.setup(OrderedDict([
(DOMAIN, {'platform': 'mod1'}),
("{} 2".format(DOMAIN), {'platform': 'non_exist'}),
("{} 3".format(DOMAIN), {'platform': 'mod2'}),
]))
self.hass.block_till_done()
assert platform1_setup.called
assert platform2_setup.called
@patch('homeassistant.helpers.entity_component.EntityComponent'
'._async_setup_platform', return_value=mock_coro())
@patch('homeassistant.setup.async_setup_component',
return_value=mock_coro(True))
def test_setup_does_discovery(self, mock_setup_component, mock_setup):
"""Test setup for discovery."""
component = EntityComponent(_LOGGER, DOMAIN, self.hass)
component.setup({})
discovery.load_platform(self.hass, DOMAIN, 'platform_test',
{'msg': 'discovery_info'}, {DOMAIN: {}})
self.hass.block_till_done()
assert mock_setup.called
assert ('platform_test', {}, {'msg': 'discovery_info'}) == \
mock_setup.call_args[0]
@patch('homeassistant.helpers.entity_platform.'
'async_track_time_interval')
def test_set_scan_interval_via_config(self, mock_track):
"""Test the setting of the scan interval via configuration."""
def platform_setup(hass, config, add_entities, discovery_info=None):
"""Test the platform setup."""
add_entities([MockEntity(should_poll=True)])
mock_entity_platform(self.hass, 'test_domain.platform',
MockPlatform(platform_setup))
component = EntityComponent(_LOGGER, DOMAIN, self.hass)
component.setup({
DOMAIN: {
'platform': 'platform',
'scan_interval': timedelta(seconds=30),
}
})
self.hass.block_till_done()
assert mock_track.called
assert timedelta(seconds=30) == mock_track.call_args[0][2]
def test_set_entity_namespace_via_config(self):
"""Test setting an entity namespace."""
def platform_setup(hass, config, add_entities, discovery_info=None):
"""Test the platform setup."""
add_entities([
MockEntity(name='beer'),
MockEntity(name=None),
])
platform = MockPlatform(platform_setup)
mock_entity_platform(self.hass, 'test_domain.platform', platform)
component = EntityComponent(_LOGGER, DOMAIN, self.hass)
component.setup({
DOMAIN: {
'platform': 'platform',
'entity_namespace': 'yummy'
}
})
self.hass.block_till_done()
assert sorted(self.hass.states.entity_ids()) == \
['test_domain.yummy_beer', 'test_domain.yummy_unnamed_device']
# Ordered in order of added to the group
assert grp.attributes.get('entity_id') == \
('test_domain.goodbye', 'test_domain.unnamed_device')
@asyncio.coroutine
def test_extract_from_service_available_device(hass):
async def test_setup_loads_platforms(hass):
"""Test the loading of the platforms."""
component_setup = Mock(return_value=True)
platform_setup = Mock(return_value=None)
mock_integration(hass, MockModule('test_component', setup=component_setup))
# mock the dependencies
mock_integration(hass, MockModule('mod2', dependencies=['test_component']))
mock_entity_platform(hass, 'test_domain.mod2',
MockPlatform(platform_setup))
component = EntityComponent(_LOGGER, DOMAIN, hass)
assert not component_setup.called
assert not platform_setup.called
component.setup({
DOMAIN: {
'platform': 'mod2',
}
})
await hass.async_block_till_done()
assert component_setup.called
assert platform_setup.called
async def test_setup_recovers_when_setup_raises(hass):
"""Test the setup if exceptions are happening."""
platform1_setup = Mock(side_effect=Exception('Broken'))
platform2_setup = Mock(return_value=None)
mock_entity_platform(hass, 'test_domain.mod1',
MockPlatform(platform1_setup))
mock_entity_platform(hass, 'test_domain.mod2',
MockPlatform(platform2_setup))
component = EntityComponent(_LOGGER, DOMAIN, hass)
assert not platform1_setup.called
assert not platform2_setup.called
component.setup(OrderedDict([
(DOMAIN, {'platform': 'mod1'}),
("{} 2".format(DOMAIN), {'platform': 'non_exist'}),
("{} 3".format(DOMAIN), {'platform': 'mod2'}),
]))
await hass.async_block_till_done()
assert platform1_setup.called
assert platform2_setup.called
@asynctest.patch('homeassistant.helpers.entity_component.EntityComponent'
'._async_setup_platform', return_value=mock_coro())
@asynctest.patch('homeassistant.setup.async_setup_component',
return_value=mock_coro(True))
async def test_setup_does_discovery(mock_setup_component, mock_setup, hass):
"""Test setup for discovery."""
component = EntityComponent(_LOGGER, DOMAIN, hass)
component.setup({})
discovery.load_platform(hass, DOMAIN, 'platform_test',
{'msg': 'discovery_info'}, {DOMAIN: {}})
await hass.async_block_till_done()
assert mock_setup.called
assert ('platform_test', {}, {'msg': 'discovery_info'}) == \
mock_setup.call_args[0]
@asynctest.patch('homeassistant.helpers.entity_platform.'
'async_track_time_interval')
async def test_set_scan_interval_via_config(mock_track, hass):
"""Test the setting of the scan interval via configuration."""
def platform_setup(hass, config, add_entities, discovery_info=None):
"""Test the platform setup."""
add_entities([MockEntity(should_poll=True)])
mock_entity_platform(hass, 'test_domain.platform',
MockPlatform(platform_setup))
component = EntityComponent(_LOGGER, DOMAIN, hass)
component.setup({
DOMAIN: {
'platform': 'platform',
'scan_interval': timedelta(seconds=30),
}
})
await hass.async_block_till_done()
assert mock_track.called
assert timedelta(seconds=30) == mock_track.call_args[0][2]
async def test_set_entity_namespace_via_config(hass):
"""Test setting an entity namespace."""
def platform_setup(hass, config, add_entities, discovery_info=None):
"""Test the platform setup."""
add_entities([
MockEntity(name='beer'),
MockEntity(name=None),
])
platform = MockPlatform(platform_setup)
mock_entity_platform(hass, 'test_domain.platform', platform)
component = EntityComponent(_LOGGER, DOMAIN, hass)
component.setup({
DOMAIN: {
'platform': 'platform',
'entity_namespace': 'yummy'
}
})
await hass.async_block_till_done()
assert sorted(hass.states.async_entity_ids()) == \
['test_domain.yummy_beer', 'test_domain.yummy_unnamed_device']
async def test_extract_from_service_available_device(hass):
"""Test the extraction of entity from service and device is available."""
component = EntityComponent(_LOGGER, DOMAIN, hass)
yield from component.async_add_entities([
await component.async_add_entities([
MockEntity(name='test_1'),
MockEntity(name='test_2', available=False),
MockEntity(name='test_3'),
@@ -209,7 +199,7 @@ def test_extract_from_service_available_device(hass):
assert ['test_domain.test_1', 'test_domain.test_3'] == \
sorted(ent.entity_id for ent in
(yield from component.async_extract_from_service(call_1)))
(await component.async_extract_from_service(call_1)))
call_2 = ha.ServiceCall('test', 'service', data={
'entity_id': ['test_domain.test_3', 'test_domain.test_4'],
@@ -217,11 +207,10 @@ def test_extract_from_service_available_device(hass):
assert ['test_domain.test_3'] == \
sorted(ent.entity_id for ent in
(yield from component.async_extract_from_service(call_2)))
(await component.async_extract_from_service(call_2)))
@asyncio.coroutine
def test_platform_not_ready(hass):
async def test_platform_not_ready(hass):
"""Test that we retry when platform not ready."""
platform1_setup = Mock(side_effect=[PlatformNotReady, PlatformNotReady,
None])
@@ -231,7 +220,7 @@ def test_platform_not_ready(hass):
component = EntityComponent(_LOGGER, DOMAIN, hass)
yield from component.async_setup({
await component.async_setup({
DOMAIN: {
'platform': 'mod1'
}
@@ -245,32 +234,31 @@ def test_platform_not_ready(hass):
with patch('homeassistant.util.dt.utcnow', return_value=utcnow):
# Should not trigger attempt 2
async_fire_time_changed(hass, utcnow + timedelta(seconds=29))
yield from hass.async_block_till_done()
await hass.async_block_till_done()
assert len(platform1_setup.mock_calls) == 1
# Should trigger attempt 2
async_fire_time_changed(hass, utcnow + timedelta(seconds=30))
yield from hass.async_block_till_done()
await hass.async_block_till_done()
assert len(platform1_setup.mock_calls) == 2
assert 'test_domain.mod1' not in hass.config.components
# This should not trigger attempt 3
async_fire_time_changed(hass, utcnow + timedelta(seconds=59))
yield from hass.async_block_till_done()
await hass.async_block_till_done()
assert len(platform1_setup.mock_calls) == 2
# Trigger attempt 3, which succeeds
async_fire_time_changed(hass, utcnow + timedelta(seconds=60))
yield from hass.async_block_till_done()
await hass.async_block_till_done()
assert len(platform1_setup.mock_calls) == 3
assert 'test_domain.mod1' in hass.config.components
@asyncio.coroutine
def test_extract_from_service_returns_all_if_no_entity_id(hass):
async def test_extract_from_service_returns_all_if_no_entity_id(hass):
"""Test the extraction of everything from service."""
component = EntityComponent(_LOGGER, DOMAIN, hass)
yield from component.async_add_entities([
await component.async_add_entities([
MockEntity(name='test_1'),
MockEntity(name='test_2'),
])
@@ -279,14 +267,13 @@ def test_extract_from_service_returns_all_if_no_entity_id(hass):
assert ['test_domain.test_1', 'test_domain.test_2'] == \
sorted(ent.entity_id for ent in
(yield from component.async_extract_from_service(call)))
(await component.async_extract_from_service(call)))
@asyncio.coroutine
def test_extract_from_service_filter_out_non_existing_entities(hass):
async def test_extract_from_service_filter_out_non_existing_entities(hass):
"""Test the extraction of non existing entities from service."""
component = EntityComponent(_LOGGER, DOMAIN, hass)
yield from component.async_add_entities([
await component.async_add_entities([
MockEntity(name='test_1'),
MockEntity(name='test_2'),
])
@@ -297,28 +284,26 @@ def test_extract_from_service_filter_out_non_existing_entities(hass):
assert ['test_domain.test_2'] == \
[ent.entity_id for ent
in (yield from component.async_extract_from_service(call))]
in await component.async_extract_from_service(call)]
@asyncio.coroutine
def test_extract_from_service_no_group_expand(hass):
async def test_extract_from_service_no_group_expand(hass):
"""Test not expanding a group."""
component = EntityComponent(_LOGGER, DOMAIN, hass)
test_group = yield from group.Group.async_create_group(
test_group = await group.Group.async_create_group(
hass, 'test_group', ['light.Ceiling', 'light.Kitchen'])
yield from component.async_add_entities([test_group])
await component.async_add_entities([test_group])
call = ha.ServiceCall('test', 'service', {
'entity_id': ['group.test_group']
})
extracted = yield from component.async_extract_from_service(
extracted = await component.async_extract_from_service(
call, expand_group=False)
assert extracted == [test_group]
@asyncio.coroutine
def test_setup_dependencies_platform(hass):
async def test_setup_dependencies_platform(hass):
"""Test we setup the dependencies of a platform.
We're explictely testing that we process dependencies even if a component
@@ -331,7 +316,7 @@ def test_setup_dependencies_platform(hass):
component = EntityComponent(_LOGGER, DOMAIN, hass)
yield from component.async_setup({
await component.async_setup({
DOMAIN: {
'platform': 'test_component',
}
@@ -355,7 +340,7 @@ async def test_setup_entry(hass):
assert await component.async_setup_entry(entry)
assert len(mock_setup_entry.mock_calls) == 1
p_hass, p_entry, p_add_entities = mock_setup_entry.mock_calls[0][1]
p_hass, p_entry, _ = mock_setup_entry.mock_calls[0][1]
assert p_hass is hass
assert p_entry is entry
@@ -448,7 +433,7 @@ async def test_set_service_race(hass):
await async_setup_component(hass, 'group', {})
component = EntityComponent(_LOGGER, DOMAIN, hass, group_name='yo')
for i in range(2):
for _ in range(2):
hass.async_create_task(component.async_add_entities([MockEntity()]))
await hass.async_block_till_done()