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

Update cover tests (#16832)

Browse source 
* Update test_group
* Update test_command_line
* Update test_demo
* Update test_mqtt
* Update test_template
* Remove cover service call helpers
This commit is contained in:
cdce8p 2018-09-25 22:32:05 +02:00 • committed by GitHub
parent 8e311686d0
commit 7eaf8640d0
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 595 additions and 589 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

58
homeassistant/components/cover/__init__.py
View file

@ -81,64 +81,6 @@ def is_closed(hass, entity_id=None):
return hass.states.is_state(entity_id, STATE_CLOSED) return hass.states.is_state(entity_id, STATE_CLOSED)
@bind_hass
def open_cover(hass, entity_id=None):
"""Open all or specified cover."""
data = {ATTR_ENTITY_ID: entity_id} if entity_id else None
hass.services.call(DOMAIN, SERVICE_OPEN_COVER, data)
@bind_hass
def close_cover(hass, entity_id=None):
"""Close all or specified cover."""
data = {ATTR_ENTITY_ID: entity_id} if entity_id else None
hass.services.call(DOMAIN, SERVICE_CLOSE_COVER, data)
@bind_hass
def set_cover_position(hass, position, entity_id=None):
"""Move to specific position all or specified cover."""
data = {ATTR_ENTITY_ID: entity_id} if entity_id else {}
data[ATTR_POSITION] = position
hass.services.call(DOMAIN, SERVICE_SET_COVER_POSITION, data)
@bind_hass
def stop_cover(hass, entity_id=None):
"""Stop all or specified cover."""
data = {ATTR_ENTITY_ID: entity_id} if entity_id else None
hass.services.call(DOMAIN, SERVICE_STOP_COVER, data)
@bind_hass
def open_cover_tilt(hass, entity_id=None):
"""Open all or specified cover tilt."""
data = {ATTR_ENTITY_ID: entity_id} if entity_id else None
hass.services.call(DOMAIN, SERVICE_OPEN_COVER_TILT, data)
@bind_hass
def close_cover_tilt(hass, entity_id=None):
"""Close all or specified cover tilt."""
data = {ATTR_ENTITY_ID: entity_id} if entity_id else None
hass.services.call(DOMAIN, SERVICE_CLOSE_COVER_TILT, data)
@bind_hass
def set_cover_tilt_position(hass, tilt_position, entity_id=None):
"""Move to specific tilt position all or specified cover."""
data = {ATTR_ENTITY_ID: entity_id} if entity_id else {}
data[ATTR_TILT_POSITION] = tilt_position
hass.services.call(DOMAIN, SERVICE_SET_COVER_TILT_POSITION, data)
@bind_hass
def stop_cover_tilt(hass, entity_id=None):
"""Stop all or specified cover tilt."""
data = {ATTR_ENTITY_ID: entity_id} if entity_id else None
hass.services.call(DOMAIN, SERVICE_STOP_COVER_TILT, data)
async def async_setup(hass, config): async def async_setup(hass, config):
"""Track states and offer events for covers.""" """Track states and offer events for covers."""
component = hass.data[DOMAIN] = EntityComponent( component = hass.data[DOMAIN] = EntityComponent(

126
tests/components/cover/test_command_line.py
View file

@ -1,87 +1,75 @@
"""The tests the cover command line platform.""" """The tests the cover command line platform."""
import os import os
import tempfile import tempfile
import unittest
from unittest import mock from unittest import mock
from homeassistant.setup import setup_component import pytest
import homeassistant.components.cover as cover
from homeassistant.components.cover import (
command_line as cmd_rs)
from tests.common import get_test_home_assistant from homeassistant.components.cover import DOMAIN
import homeassistant.components.cover.command_line as cmd_rs
from homeassistant.const import (
ATTR_ENTITY_ID, SERVICE_CLOSE_COVER, SERVICE_OPEN_COVER,
SERVICE_STOP_COVER)
from homeassistant.setup import async_setup_component
class TestCommandCover(unittest.TestCase): @pytest.fixture
"""Test the cover command line platform.""" def rs(hass):
"""Return CommandCover instance."""
return cmd_rs.CommandCover(hass, 'foo', 'command_open', 'command_close',
'command_stop', 'command_state', None)
def setup_method(self, method):
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
self.rs = cmd_rs.CommandCover(self.hass, 'foo',
'command_open', 'command_close',
'command_stop', 'command_state',
None)
def teardown_method(self, method): def test_should_poll_new(rs):
"""Stop down everything that was started.""" """Test the setting of polling."""
self.hass.stop() assert rs.should_poll is True
rs._command_state = None
assert rs.should_poll is False
def test_should_poll(self):
"""Test the setting of polling."""
self.assertTrue(self.rs.should_poll)
self.rs._command_state = None
self.assertFalse(self.rs.should_poll)
def test_query_state_value(self): def test_query_state_value(rs):
"""Test with state value.""" """Test with state value."""
with mock.patch('subprocess.check_output') as mock_run: with mock.patch('subprocess.check_output') as mock_run:
mock_run.return_value = b' foo bar ' mock_run.return_value = b' foo bar '
result = self.rs._query_state_value('runme') result = rs._query_state_value('runme')
self.assertEqual('foo bar', result) assert 'foo bar' == result
self.assertEqual(mock_run.call_count, 1) assert mock_run.call_count == 1
self.assertEqual( assert mock_run.call_args == mock.call('runme', shell=True)
mock_run.call_args, mock.call('runme', shell=True)
)
def test_state_value(self):
"""Test with state value.""" async def test_state_value(hass):
with tempfile.TemporaryDirectory() as tempdirname: """Test with state value."""
path = os.path.join(tempdirname, 'cover_status') with tempfile.TemporaryDirectory() as tempdirname:
test_cover = { path = os.path.join(tempdirname, 'cover_status')
'command_state': 'cat {}'.format(path), test_cover = {
'command_open': 'echo 1 > {}'.format(path), 'command_state': 'cat {}'.format(path),
'command_close': 'echo 1 > {}'.format(path), 'command_open': 'echo 1 > {}'.format(path),
'command_stop': 'echo 0 > {}'.format(path), 'command_close': 'echo 1 > {}'.format(path),
'value_template': '{{ value }}' 'command_stop': 'echo 0 > {}'.format(path),
} 'value_template': '{{ value }}'
self.assertTrue(setup_component(self.hass, cover.DOMAIN, { }
'cover': { assert await async_setup_component(hass, DOMAIN, {
'platform': 'command_line', 'cover': {
'covers': { 'platform': 'command_line',
'test': test_cover 'covers': {
} 'test': test_cover
} }
})) }
}) is True
state = self.hass.states.get('cover.test') assert 'unknown' == hass.states.get('cover.test').state
self.assertEqual('unknown', state.state)
cover.open_cover(self.hass, 'cover.test') await hass.services.async_call(
self.hass.block_till_done() DOMAIN, SERVICE_OPEN_COVER,
{ATTR_ENTITY_ID: 'cover.test'}, blocking=True)
assert 'open' == hass.states.get('cover.test').state
state = self.hass.states.get('cover.test') await hass.services.async_call(
self.assertEqual('open', state.state) DOMAIN, SERVICE_CLOSE_COVER,
{ATTR_ENTITY_ID: 'cover.test'}, blocking=True)
assert 'open' == hass.states.get('cover.test').state
cover.close_cover(self.hass, 'cover.test') await hass.services.async_call(
self.hass.block_till_done() DOMAIN, SERVICE_STOP_COVER,
{ATTR_ENTITY_ID: 'cover.test'}, blocking=True)
state = self.hass.states.get('cover.test') assert 'closed' == hass.states.get('cover.test').state
self.assertEqual('open', state.state)
cover.stop_cover(self.hass, 'cover.test')
self.hass.block_till_done()
state = self.hass.states.get('cover.test')
self.assertEqual('closed', state.state)

303
tests/components/cover/test_demo.py
View file

@ -1,158 +1,181 @@
"""The tests for the Demo cover platform.""" """The tests for the Demo cover platform."""
import unittest
from datetime import timedelta from datetime import timedelta
import pytest
from homeassistant.components.cover import (
ATTR_POSITION, ATTR_TILT_POSITION, DOMAIN)
from homeassistant.const import (
ATTR_ENTITY_ID, SERVICE_CLOSE_COVER, SERVICE_CLOSE_COVER_TILT,
SERVICE_OPEN_COVER, SERVICE_OPEN_COVER_TILT, SERVICE_SET_COVER_POSITION,
SERVICE_SET_COVER_TILT_POSITION, SERVICE_STOP_COVER,
SERVICE_STOP_COVER_TILT)
from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util import homeassistant.util.dt as dt_util
from homeassistant.setup import setup_component from tests.common import assert_setup_component, async_fire_time_changed
from homeassistant.components import cover
from tests.common import get_test_home_assistant, fire_time_changed
CONFIG = {'cover': {'platform': 'demo'}}
ENTITY_COVER = 'cover.living_room_window' ENTITY_COVER = 'cover.living_room_window'
class TestCoverDemo(unittest.TestCase): @pytest.fixture
"""Test the Demo cover.""" async def setup_comp(hass):
"""Set up demo cover component."""
with assert_setup_component(1, DOMAIN):
await async_setup_component(hass, DOMAIN, CONFIG)
def setUp(self): # pylint: disable=invalid-name
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
self.assertTrue(setup_component(self.hass, cover.DOMAIN, {'cover': {
'platform': 'demo',
}}))
def tearDown(self): # pylint: disable=invalid-name async def test_supported_features(hass, setup_comp):
"""Stop down everything that was started.""" """Test cover supported features."""
self.hass.stop() state = hass.states.get('cover.garage_door')
assert 3 == state.attributes.get('supported_features')
state = hass.states.get('cover.kitchen_window')
assert 11 == state.attributes.get('supported_features')
state = hass.states.get('cover.hall_window')
assert 15 == state.attributes.get('supported_features')
state = hass.states.get('cover.living_room_window')
assert 255 == state.attributes.get('supported_features')
def test_supported_features(self):
"""Test cover supported features."""
state = self.hass.states.get('cover.garage_door')
self.assertEqual(3, state.attributes.get('supported_features'))
state = self.hass.states.get('cover.kitchen_window')
self.assertEqual(11, state.attributes.get('supported_features'))
state = self.hass.states.get('cover.hall_window')
self.assertEqual(15, state.attributes.get('supported_features'))
state = self.hass.states.get('cover.living_room_window')
self.assertEqual(255, state.attributes.get('supported_features'))
def test_close_cover(self): async def test_close_cover(hass, setup_comp):
"""Test closing the cover.""" """Test closing the cover."""
state = self.hass.states.get(ENTITY_COVER) state = hass.states.get(ENTITY_COVER)
self.assertEqual(state.state, 'open') assert state.state == 'open'
self.assertEqual(70, state.attributes.get('current_position')) assert 70 == state.attributes.get('current_position')
cover.close_cover(self.hass, ENTITY_COVER)
self.hass.block_till_done()
state = self.hass.states.get(ENTITY_COVER)
self.assertEqual(state.state, 'closing')
for _ in range(7):
future = dt_util.utcnow() + timedelta(seconds=1)
fire_time_changed(self.hass, future)
self.hass.block_till_done()
state = self.hass.states.get(ENTITY_COVER) await hass.services.async_call(
self.assertEqual(state.state, 'closed') DOMAIN, SERVICE_CLOSE_COVER,
self.assertEqual(0, state.attributes.get('current_position')) {ATTR_ENTITY_ID: ENTITY_COVER}, blocking=True)
state = hass.states.get(ENTITY_COVER)
def test_open_cover(self): assert state.state == 'closing'
"""Test opening the cover.""" for _ in range(7):
state = self.hass.states.get(ENTITY_COVER)
self.assertEqual(state.state, 'open')
self.assertEqual(70, state.attributes.get('current_position'))
cover.open_cover(self.hass, ENTITY_COVER)
self.hass.block_till_done()
state = self.hass.states.get(ENTITY_COVER)
self.assertEqual(state.state, 'opening')
for _ in range(7):
future = dt_util.utcnow() + timedelta(seconds=1)
fire_time_changed(self.hass, future)
self.hass.block_till_done()
state = self.hass.states.get(ENTITY_COVER)
self.assertEqual(state.state, 'open')
self.assertEqual(100, state.attributes.get('current_position'))
def test_set_cover_position(self):
"""Test moving the cover to a specific position."""
state = self.hass.states.get(ENTITY_COVER)
self.assertEqual(70, state.attributes.get('current_position'))
cover.set_cover_position(self.hass, 10, ENTITY_COVER)
self.hass.block_till_done()
for _ in range(6):
future = dt_util.utcnow() + timedelta(seconds=1)
fire_time_changed(self.hass, future)
self.hass.block_till_done()
state = self.hass.states.get(ENTITY_COVER)
self.assertEqual(10, state.attributes.get('current_position'))
def test_stop_cover(self):
"""Test stopping the cover."""
state = self.hass.states.get(ENTITY_COVER)
self.assertEqual(70, state.attributes.get('current_position'))
cover.open_cover(self.hass, ENTITY_COVER)
self.hass.block_till_done()
future = dt_util.utcnow() + timedelta(seconds=1) future = dt_util.utcnow() + timedelta(seconds=1)
fire_time_changed(self.hass, future) async_fire_time_changed(hass, future)
self.hass.block_till_done() await hass.async_block_till_done()
cover.stop_cover(self.hass, ENTITY_COVER)
self.hass.block_till_done()
fire_time_changed(self.hass, future)
state = self.hass.states.get(ENTITY_COVER)
self.assertEqual(80, state.attributes.get('current_position'))
def test_close_cover_tilt(self): state = hass.states.get(ENTITY_COVER)
"""Test closing the cover tilt.""" assert state.state == 'closed'
state = self.hass.states.get(ENTITY_COVER) assert 0 == state.attributes.get('current_position')
self.assertEqual(50, state.attributes.get('current_tilt_position'))
cover.close_cover_tilt(self.hass, ENTITY_COVER)
self.hass.block_till_done()
for _ in range(7):
future = dt_util.utcnow() + timedelta(seconds=1)
fire_time_changed(self.hass, future)
self.hass.block_till_done()
state = self.hass.states.get(ENTITY_COVER)
self.assertEqual(0, state.attributes.get('current_tilt_position'))
def test_open_cover_tilt(self): async def test_open_cover(hass, setup_comp):
"""Test opening the cover tilt.""" """Test opening the cover."""
state = self.hass.states.get(ENTITY_COVER) state = hass.states.get(ENTITY_COVER)
self.assertEqual(50, state.attributes.get('current_tilt_position')) assert state.state == 'open'
cover.open_cover_tilt(self.hass, ENTITY_COVER) assert 70 == state.attributes.get('current_position')
self.hass.block_till_done() await hass.services.async_call(
for _ in range(7): DOMAIN, SERVICE_OPEN_COVER,
future = dt_util.utcnow() + timedelta(seconds=1) {ATTR_ENTITY_ID: ENTITY_COVER}, blocking=True)
fire_time_changed(self.hass, future) state = hass.states.get(ENTITY_COVER)
self.hass.block_till_done() assert state.state == 'opening'
for _ in range(7):
state = self.hass.states.get(ENTITY_COVER)
self.assertEqual(100, state.attributes.get('current_tilt_position'))
def test_set_cover_tilt_position(self):
"""Test moving the cover til to a specific position."""
state = self.hass.states.get(ENTITY_COVER)
self.assertEqual(50, state.attributes.get('current_tilt_position'))
cover.set_cover_tilt_position(self.hass, 90, ENTITY_COVER)
self.hass.block_till_done()
for _ in range(7):
future = dt_util.utcnow() + timedelta(seconds=1)
fire_time_changed(self.hass, future)
self.hass.block_till_done()
state = self.hass.states.get(ENTITY_COVER)
self.assertEqual(90, state.attributes.get('current_tilt_position'))
def test_stop_cover_tilt(self):
"""Test stopping the cover tilt."""
state = self.hass.states.get(ENTITY_COVER)
self.assertEqual(50, state.attributes.get('current_tilt_position'))
cover.close_cover_tilt(self.hass, ENTITY_COVER)
self.hass.block_till_done()
future = dt_util.utcnow() + timedelta(seconds=1) future = dt_util.utcnow() + timedelta(seconds=1)
fire_time_changed(self.hass, future) async_fire_time_changed(hass, future)
self.hass.block_till_done() await hass.async_block_till_done()
cover.stop_cover_tilt(self.hass, ENTITY_COVER)
self.hass.block_till_done() state = hass.states.get(ENTITY_COVER)
fire_time_changed(self.hass, future) assert state.state == 'open'
state = self.hass.states.get(ENTITY_COVER) assert 100 == state.attributes.get('current_position')
self.assertEqual(40, state.attributes.get('current_tilt_position'))
async def test_set_cover_position(hass, setup_comp):
"""Test moving the cover to a specific position."""
state = hass.states.get(ENTITY_COVER)
assert 70 == state.attributes.get('current_position')
await hass.services.async_call(
DOMAIN, SERVICE_SET_COVER_POSITION,
{ATTR_ENTITY_ID: ENTITY_COVER, ATTR_POSITION: 10}, blocking=True)
for _ in range(6):
future = dt_util.utcnow() + timedelta(seconds=1)
async_fire_time_changed(hass, future)
await hass.async_block_till_done()
state = hass.states.get(ENTITY_COVER)
assert 10 == state.attributes.get('current_position')
async def test_stop_cover(hass, setup_comp):
"""Test stopping the cover."""
state = hass.states.get(ENTITY_COVER)
assert 70 == state.attributes.get('current_position')
await hass.services.async_call(
DOMAIN, SERVICE_OPEN_COVER,
{ATTR_ENTITY_ID: ENTITY_COVER}, blocking=True)
future = dt_util.utcnow() + timedelta(seconds=1)
async_fire_time_changed(hass, future)
await hass.async_block_till_done()
await hass.services.async_call(
DOMAIN, SERVICE_STOP_COVER,
{ATTR_ENTITY_ID: ENTITY_COVER}, blocking=True)
async_fire_time_changed(hass, future)
await hass.async_block_till_done()
state = hass.states.get(ENTITY_COVER)
assert 80 == state.attributes.get('current_position')
async def test_close_cover_tilt(hass, setup_comp):
"""Test closing the cover tilt."""
state = hass.states.get(ENTITY_COVER)
assert 50 == state.attributes.get('current_tilt_position')
await hass.services.async_call(
DOMAIN, SERVICE_CLOSE_COVER_TILT,
{ATTR_ENTITY_ID: ENTITY_COVER}, blocking=True)
for _ in range(7):
future = dt_util.utcnow() + timedelta(seconds=1)
async_fire_time_changed(hass, future)
await hass.async_block_till_done()
state = hass.states.get(ENTITY_COVER)
assert 0 == state.attributes.get('current_tilt_position')
async def test_open_cover_tilt(hass, setup_comp):
"""Test opening the cover tilt."""
state = hass.states.get(ENTITY_COVER)
assert 50 == state.attributes.get('current_tilt_position')
await hass.services.async_call(
DOMAIN, SERVICE_OPEN_COVER_TILT,
{ATTR_ENTITY_ID: ENTITY_COVER}, blocking=True)
for _ in range(7):
future = dt_util.utcnow() + timedelta(seconds=1)
async_fire_time_changed(hass, future)
await hass.async_block_till_done()
state = hass.states.get(ENTITY_COVER)
assert 100 == state.attributes.get('current_tilt_position')
async def test_set_cover_tilt_position(hass, setup_comp):
"""Test moving the cover til to a specific position."""
state = hass.states.get(ENTITY_COVER)
assert 50 == state.attributes.get('current_tilt_position')
await hass.services.async_call(
DOMAIN, SERVICE_SET_COVER_TILT_POSITION,
{ATTR_ENTITY_ID: ENTITY_COVER, ATTR_TILT_POSITION: 90}, blocking=True)
for _ in range(7):
future = dt_util.utcnow() + timedelta(seconds=1)
async_fire_time_changed(hass, future)
await hass.async_block_till_done()
state = hass.states.get(ENTITY_COVER)
assert 90 == state.attributes.get('current_tilt_position')
async def test_stop_cover_tilt(hass, setup_comp):
"""Test stopping the cover tilt."""
state = hass.states.get(ENTITY_COVER)
assert 50 == state.attributes.get('current_tilt_position')
await hass.services.async_call(
DOMAIN, SERVICE_CLOSE_COVER_TILT,
{ATTR_ENTITY_ID: ENTITY_COVER}, blocking=True)
future = dt_util.utcnow() + timedelta(seconds=1)
async_fire_time_changed(hass, future)
await hass.async_block_till_done()
await hass.services.async_call(
DOMAIN, SERVICE_STOP_COVER_TILT,
{ATTR_ENTITY_ID: ENTITY_COVER}, blocking=True)
async_fire_time_changed(hass, future)
await hass.async_block_till_done()
state = hass.states.get(ENTITY_COVER)
assert 40 == state.attributes.get('current_tilt_position')

551
tests/components/cover/test_group.py
View file

@ -1,19 +1,23 @@
"""The tests for the group cover platform.""" """The tests for the group cover platform."""
import unittest
from datetime import timedelta from datetime import timedelta
import homeassistant.util.dt as dt_util
from homeassistant import setup import pytest
from homeassistant.components import cover
from homeassistant.components.cover import ( from homeassistant.components.cover import (
ATTR_CURRENT_POSITION, ATTR_CURRENT_TILT_POSITION, DOMAIN) ATTR_CURRENT_POSITION, ATTR_CURRENT_TILT_POSITION, ATTR_POSITION,
ATTR_TILT_POSITION, DOMAIN)
from homeassistant.components.cover.group import DEFAULT_NAME from homeassistant.components.cover.group import DEFAULT_NAME
from homeassistant.const import ( from homeassistant.const import (
ATTR_ASSUMED_STATE, ATTR_FRIENDLY_NAME, ATTR_SUPPORTED_FEATURES, ATTR_ASSUMED_STATE, ATTR_ENTITY_ID, ATTR_FRIENDLY_NAME,
CONF_ENTITIES, STATE_OPEN, STATE_CLOSED) ATTR_SUPPORTED_FEATURES, CONF_ENTITIES,
from tests.common import ( SERVICE_CLOSE_COVER, SERVICE_CLOSE_COVER_TILT,
assert_setup_component, get_test_home_assistant, fire_time_changed) SERVICE_OPEN_COVER, SERVICE_OPEN_COVER_TILT, SERVICE_SET_COVER_POSITION,
SERVICE_SET_COVER_TILT_POSITION, SERVICE_STOP_COVER,
SERVICE_STOP_COVER_TILT, STATE_OPEN, STATE_CLOSED)
from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util
from tests.common import assert_setup_component, async_fire_time_changed
COVER_GROUP = 'cover.cover_group' COVER_GROUP = 'cover.cover_group'
DEMO_COVER = 'cover.kitchen_window' DEMO_COVER = 'cover.kitchen_window'
@ -31,320 +35,301 @@ CONFIG = {
} }
class TestMultiCover(unittest.TestCase): @pytest.fixture
"""Test the group cover platform.""" async def setup_comp(hass):
"""Set up group cover component."""
with assert_setup_component(2, DOMAIN):
await async_setup_component(hass, DOMAIN, CONFIG)
def setUp(self):
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
def tearDown(self): async def test_attributes(hass):
"""Stop down everything that was started.""" """Test handling of state attributes."""
self.hass.stop() config = {DOMAIN: {'platform': 'group', CONF_ENTITIES: [
DEMO_COVER, DEMO_COVER_POS, DEMO_COVER_TILT, DEMO_TILT]}}
def test_attributes(self): with assert_setup_component(1, DOMAIN):
"""Test handling of state attributes.""" await async_setup_component(hass, DOMAIN, config)
config = {DOMAIN: {'platform': 'group', CONF_ENTITIES: [
DEMO_COVER, DEMO_COVER_POS, DEMO_COVER_TILT, DEMO_TILT]}}
with assert_setup_component(1, DOMAIN): state = hass.states.get(COVER_GROUP)
assert setup.setup_component(self.hass, DOMAIN, config) assert state.state == STATE_CLOSED
assert state.attributes.get(ATTR_FRIENDLY_NAME) == DEFAULT_NAME
assert state.attributes.get(ATTR_ASSUMED_STATE) is None
assert state.attributes.get(ATTR_SUPPORTED_FEATURES) == 0
assert state.attributes.get(ATTR_CURRENT_POSITION) is None
assert state.attributes.get(ATTR_CURRENT_TILT_POSITION) is None
state = self.hass.states.get(COVER_GROUP) # Add Entity that supports open / close / stop
attr = state.attributes hass.states.async_set(
self.assertEqual(state.state, STATE_CLOSED) DEMO_COVER, STATE_OPEN, {ATTR_SUPPORTED_FEATURES: 11})
self.assertEqual(attr.get(ATTR_FRIENDLY_NAME), DEFAULT_NAME) await hass.async_block_till_done()
self.assertEqual(attr.get(ATTR_ASSUMED_STATE), None)
self.assertEqual(attr.get(ATTR_SUPPORTED_FEATURES), 0)
self.assertEqual(attr.get(ATTR_CURRENT_POSITION), None)
self.assertEqual(attr.get(ATTR_CURRENT_TILT_POSITION), None)
# Add Entity that supports open / close / stop state = hass.states.get(COVER_GROUP)
self.hass.states.set( assert state.state == STATE_OPEN
DEMO_COVER, STATE_OPEN, {ATTR_SUPPORTED_FEATURES: 11}) assert state.attributes.get(ATTR_ASSUMED_STATE) is None
self.hass.block_till_done() assert state.attributes.get(ATTR_SUPPORTED_FEATURES) == 11
assert state.attributes.get(ATTR_CURRENT_POSITION) is None
assert state.attributes.get(ATTR_CURRENT_TILT_POSITION) is None
state = self.hass.states.get(COVER_GROUP) # Add Entity that supports set_cover_position
attr = state.attributes hass.states.async_set(
self.assertEqual(state.state, STATE_OPEN) DEMO_COVER_POS, STATE_OPEN,
self.assertEqual(attr.get(ATTR_ASSUMED_STATE), None) {ATTR_SUPPORTED_FEATURES: 4, ATTR_CURRENT_POSITION: 70})
self.assertEqual(attr.get(ATTR_SUPPORTED_FEATURES), 11) await hass.async_block_till_done()
self.assertEqual(attr.get(ATTR_CURRENT_POSITION), None)
self.assertEqual(attr.get(ATTR_CURRENT_TILT_POSITION), None)
# Add Entity that supports set_cover_position state = hass.states.get(COVER_GROUP)
self.hass.states.set( assert state.state == STATE_OPEN
DEMO_COVER_POS, STATE_OPEN, assert state.attributes.get(ATTR_ASSUMED_STATE) is None
{ATTR_SUPPORTED_FEATURES: 4, ATTR_CURRENT_POSITION: 70}) assert state.attributes.get(ATTR_SUPPORTED_FEATURES) == 15
self.hass.block_till_done() assert state.attributes.get(ATTR_CURRENT_POSITION) == 70
assert state.attributes.get(ATTR_CURRENT_TILT_POSITION) is None
state = self.hass.states.get(COVER_GROUP) # Add Entity that supports open tilt / close tilt / stop tilt
attr = state.attributes hass.states.async_set(
self.assertEqual(state.state, STATE_OPEN) DEMO_TILT, STATE_OPEN, {ATTR_SUPPORTED_FEATURES: 112})
self.assertEqual(attr.get(ATTR_ASSUMED_STATE), None) await hass.async_block_till_done()
self.assertEqual(attr.get(ATTR_SUPPORTED_FEATURES), 15)
self.assertEqual(attr.get(ATTR_CURRENT_POSITION), 70)
self.assertEqual(attr.get(ATTR_CURRENT_TILT_POSITION), None)
# Add Entity that supports open tilt / close tilt / stop tilt state = hass.states.get(COVER_GROUP)
self.hass.states.set( assert state.state == STATE_OPEN
DEMO_TILT, STATE_OPEN, {ATTR_SUPPORTED_FEATURES: 112}) assert state.attributes.get(ATTR_ASSUMED_STATE) is None
self.hass.block_till_done() assert state.attributes.get(ATTR_SUPPORTED_FEATURES) == 127
assert state.attributes.get(ATTR_CURRENT_POSITION) == 70
assert state.attributes.get(ATTR_CURRENT_TILT_POSITION) is None
state = self.hass.states.get(COVER_GROUP) # Add Entity that supports set_tilt_position
attr = state.attributes hass.states.async_set(
self.assertEqual(state.state, STATE_OPEN) DEMO_COVER_TILT, STATE_OPEN,
self.assertEqual(attr.get(ATTR_ASSUMED_STATE), None) {ATTR_SUPPORTED_FEATURES: 128, ATTR_CURRENT_TILT_POSITION: 60})
self.assertEqual(attr.get(ATTR_SUPPORTED_FEATURES), 127) await hass.async_block_till_done()
self.assertEqual(attr.get(ATTR_CURRENT_POSITION), 70)
self.assertEqual(attr.get(ATTR_CURRENT_TILT_POSITION), None)
# Add Entity that supports set_tilt_position state = hass.states.get(COVER_GROUP)
self.hass.states.set( assert state.state == STATE_OPEN
DEMO_COVER_TILT, STATE_OPEN, assert state.attributes.get(ATTR_ASSUMED_STATE) is None
{ATTR_SUPPORTED_FEATURES: 128, ATTR_CURRENT_TILT_POSITION: 60}) assert state.attributes.get(ATTR_SUPPORTED_FEATURES) == 255
self.hass.block_till_done() assert state.attributes.get(ATTR_CURRENT_POSITION) == 70
assert state.attributes.get(ATTR_CURRENT_TILT_POSITION) == 60
state = self.hass.states.get(COVER_GROUP) # ### Test assumed state ###
attr = state.attributes # ##########################
self.assertEqual(state.state, STATE_OPEN)
self.assertEqual(attr.get(ATTR_ASSUMED_STATE), None)
self.assertEqual(attr.get(ATTR_SUPPORTED_FEATURES), 255)
self.assertEqual(attr.get(ATTR_CURRENT_POSITION), 70)
self.assertEqual(attr.get(ATTR_CURRENT_TILT_POSITION), 60)
# ### Test assumed state ### # For covers
# ########################## hass.states.async_set(
DEMO_COVER, STATE_OPEN,
{ATTR_SUPPORTED_FEATURES: 4, ATTR_CURRENT_POSITION: 100})
await hass.async_block_till_done()
# For covers state = hass.states.get(COVER_GROUP)
self.hass.states.set( assert state.state == STATE_OPEN
DEMO_COVER, STATE_OPEN, assert state.attributes.get(ATTR_ASSUMED_STATE) is True
{ATTR_SUPPORTED_FEATURES: 4, ATTR_CURRENT_POSITION: 100}) assert state.attributes.get(ATTR_SUPPORTED_FEATURES) == 244
self.hass.block_till_done() assert state.attributes.get(ATTR_CURRENT_POSITION) == 100
assert state.attributes.get(ATTR_CURRENT_TILT_POSITION) == 60
state = self.hass.states.get(COVER_GROUP) hass.states.async_remove(DEMO_COVER)
attr = state.attributes hass.states.async_remove(DEMO_COVER_POS)
self.assertEqual(state.state, STATE_OPEN) await hass.async_block_till_done()
self.assertEqual(attr.get(ATTR_ASSUMED_STATE), True)
self.assertEqual(attr.get(ATTR_SUPPORTED_FEATURES), 244)
self.assertEqual(attr.get(ATTR_CURRENT_POSITION), 100)
self.assertEqual(attr.get(ATTR_CURRENT_TILT_POSITION), 60)
self.hass.states.remove(DEMO_COVER) state = hass.states.get(COVER_GROUP)
self.hass.block_till_done() assert state.state == STATE_OPEN
self.hass.states.remove(DEMO_COVER_POS) assert state.attributes.get(ATTR_ASSUMED_STATE) is None
self.hass.block_till_done() assert state.attributes.get(ATTR_SUPPORTED_FEATURES) == 240
assert state.attributes.get(ATTR_CURRENT_POSITION) is None
assert state.attributes.get(ATTR_CURRENT_TILT_POSITION) == 60
state = self.hass.states.get(COVER_GROUP) # For tilts
attr = state.attributes hass.states.async_set(
self.assertEqual(state.state, STATE_OPEN) DEMO_TILT, STATE_OPEN,
self.assertEqual(attr.get(ATTR_ASSUMED_STATE), None) {ATTR_SUPPORTED_FEATURES: 128, ATTR_CURRENT_TILT_POSITION: 100})
self.assertEqual(attr.get(ATTR_SUPPORTED_FEATURES), 240) await hass.async_block_till_done()
self.assertEqual(attr.get(ATTR_CURRENT_POSITION), None)
self.assertEqual(attr.get(ATTR_CURRENT_TILT_POSITION), 60)
# For tilts state = hass.states.get(COVER_GROUP)
self.hass.states.set( assert state.state == STATE_OPEN
DEMO_TILT, STATE_OPEN, assert state.attributes.get(ATTR_ASSUMED_STATE) is True
{ATTR_SUPPORTED_FEATURES: 128, ATTR_CURRENT_TILT_POSITION: 100}) assert state.attributes.get(ATTR_SUPPORTED_FEATURES) == 128
self.hass.block_till_done() assert state.attributes.get(ATTR_CURRENT_POSITION) is None
assert state.attributes.get(ATTR_CURRENT_TILT_POSITION) == 100
state = self.hass.states.get(COVER_GROUP) hass.states.async_remove(DEMO_COVER_TILT)
attr = state.attributes hass.states.async_set(DEMO_TILT, STATE_CLOSED)
self.assertEqual(state.state, STATE_OPEN) await hass.async_block_till_done()
self.assertEqual(attr.get(ATTR_ASSUMED_STATE), True)
self.assertEqual(attr.get(ATTR_SUPPORTED_FEATURES), 128)
self.assertEqual(attr.get(ATTR_CURRENT_POSITION), None)
self.assertEqual(attr.get(ATTR_CURRENT_TILT_POSITION), 100)
self.hass.states.remove(DEMO_COVER_TILT) state = hass.states.get(COVER_GROUP)
self.hass.states.set(DEMO_TILT, STATE_CLOSED) assert state.state == STATE_CLOSED
self.hass.block_till_done() assert state.attributes.get(ATTR_ASSUMED_STATE) is None
assert state.attributes.get(ATTR_SUPPORTED_FEATURES) == 0
assert state.attributes.get(ATTR_CURRENT_POSITION) is None
assert state.attributes.get(ATTR_CURRENT_TILT_POSITION) is None
state = self.hass.states.get(COVER_GROUP) hass.states.async_set(
attr = state.attributes DEMO_TILT, STATE_CLOSED, {ATTR_ASSUMED_STATE: True})
self.assertEqual(state.state, STATE_CLOSED) await hass.async_block_till_done()
self.assertEqual(attr.get(ATTR_ASSUMED_STATE), None)
self.assertEqual(attr.get(ATTR_SUPPORTED_FEATURES), 0)
self.assertEqual(attr.get(ATTR_CURRENT_POSITION), None)
self.assertEqual(attr.get(ATTR_CURRENT_TILT_POSITION), None)
self.hass.states.set( state = hass.states.get(COVER_GROUP)
DEMO_TILT, STATE_CLOSED, {ATTR_ASSUMED_STATE: True}) assert state.attributes.get(ATTR_ASSUMED_STATE) is True
self.hass.block_till_done()
state = self.hass.states.get(COVER_GROUP)
attr = state.attributes
self.assertEqual(attr.get(ATTR_ASSUMED_STATE), True)
def test_open_covers(self): async def test_open_covers(hass, setup_comp):
"""Test open cover function.""" """Test open cover function."""
with assert_setup_component(2, DOMAIN): await hass.services.async_call(
assert setup.setup_component(self.hass, DOMAIN, CONFIG) DOMAIN, SERVICE_OPEN_COVER,
{ATTR_ENTITY_ID: COVER_GROUP}, blocking=True)
cover.open_cover(self.hass, COVER_GROUP) for _ in range(10):
self.hass.block_till_done()
for _ in range(10):
future = dt_util.utcnow() + timedelta(seconds=1)
fire_time_changed(self.hass, future)
self.hass.block_till_done()
state = self.hass.states.get(COVER_GROUP)
self.assertEqual(state.state, STATE_OPEN)
self.assertEqual(state.attributes.get(ATTR_CURRENT_POSITION), 100)
self.assertEqual(self.hass.states.get(DEMO_COVER).state, STATE_OPEN)
self.assertEqual(self.hass.states.get(DEMO_COVER_POS)
.attributes.get(ATTR_CURRENT_POSITION), 100)
self.assertEqual(self.hass.states.get(DEMO_COVER_TILT)
.attributes.get(ATTR_CURRENT_POSITION), 100)
def test_close_covers(self):
"""Test close cover function."""
with assert_setup_component(2, DOMAIN):
assert setup.setup_component(self.hass, DOMAIN, CONFIG)
cover.close_cover(self.hass, COVER_GROUP)
self.hass.block_till_done()
for _ in range(10):
future = dt_util.utcnow() + timedelta(seconds=1)
fire_time_changed(self.hass, future)
self.hass.block_till_done()
state = self.hass.states.get(COVER_GROUP)
self.assertEqual(state.state, STATE_CLOSED)
self.assertEqual(state.attributes.get(ATTR_CURRENT_POSITION), 0)
self.assertEqual(self.hass.states.get(DEMO_COVER).state, STATE_CLOSED)
self.assertEqual(self.hass.states.get(DEMO_COVER_POS)
.attributes.get(ATTR_CURRENT_POSITION), 0)
self.assertEqual(self.hass.states.get(DEMO_COVER_TILT)
.attributes.get(ATTR_CURRENT_POSITION), 0)
def test_stop_covers(self):
"""Test stop cover function."""
with assert_setup_component(2, DOMAIN):
assert setup.setup_component(self.hass, DOMAIN, CONFIG)
cover.open_cover(self.hass, COVER_GROUP)
self.hass.block_till_done()
future = dt_util.utcnow() + timedelta(seconds=1) future = dt_util.utcnow() + timedelta(seconds=1)
fire_time_changed(self.hass, future) async_fire_time_changed(hass, future)
self.hass.block_till_done() await hass.async_block_till_done()
cover.stop_cover(self.hass, COVER_GROUP)
self.hass.block_till_done() state = hass.states.get(COVER_GROUP)
assert state.state == STATE_OPEN
assert state.attributes.get(ATTR_CURRENT_POSITION) == 100
assert hass.states.get(DEMO_COVER).state == STATE_OPEN
assert hass.states.get(DEMO_COVER_POS) \
.attributes.get(ATTR_CURRENT_POSITION) == 100
assert hass.states.get(DEMO_COVER_TILT) \
.attributes.get(ATTR_CURRENT_POSITION) == 100
async def test_close_covers(hass, setup_comp):
"""Test close cover function."""
await hass.services.async_call(
DOMAIN, SERVICE_CLOSE_COVER,
{ATTR_ENTITY_ID: COVER_GROUP}, blocking=True)
for _ in range(10):
future = dt_util.utcnow() + timedelta(seconds=1) future = dt_util.utcnow() + timedelta(seconds=1)
fire_time_changed(self.hass, future) async_fire_time_changed(hass, future)
self.hass.block_till_done() await hass.async_block_till_done()
state = self.hass.states.get(COVER_GROUP) state = hass.states.get(COVER_GROUP)
self.assertEqual(state.state, STATE_OPEN) assert state.state == STATE_CLOSED
self.assertEqual(state.attributes.get(ATTR_CURRENT_POSITION), 100) assert state.attributes.get(ATTR_CURRENT_POSITION) == 0
self.assertEqual(self.hass.states.get(DEMO_COVER).state, STATE_OPEN) assert hass.states.get(DEMO_COVER).state == STATE_CLOSED
self.assertEqual(self.hass.states.get(DEMO_COVER_POS) assert hass.states.get(DEMO_COVER_POS) \
.attributes.get(ATTR_CURRENT_POSITION), 20) .attributes.get(ATTR_CURRENT_POSITION) == 0
self.assertEqual(self.hass.states.get(DEMO_COVER_TILT) assert hass.states.get(DEMO_COVER_TILT) \
.attributes.get(ATTR_CURRENT_POSITION), 80) .attributes.get(ATTR_CURRENT_POSITION) == 0
def test_set_cover_position(self):
"""Test set cover position function."""
with assert_setup_component(2, DOMAIN):
assert setup.setup_component(self.hass, DOMAIN, CONFIG)
cover.set_cover_position(self.hass, 50, COVER_GROUP) async def test_stop_covers(hass, setup_comp):
self.hass.block_till_done() """Test stop cover function."""
for _ in range(4): await hass.services.async_call(
future = dt_util.utcnow() + timedelta(seconds=1) DOMAIN, SERVICE_OPEN_COVER,
fire_time_changed(self.hass, future) {ATTR_ENTITY_ID: COVER_GROUP}, blocking=True)
self.hass.block_till_done() future = dt_util.utcnow() + timedelta(seconds=1)
async_fire_time_changed(hass, future)
await hass.async_block_till_done()
state = self.hass.states.get(COVER_GROUP) await hass.services.async_call(
self.assertEqual(state.state, STATE_OPEN) DOMAIN, SERVICE_STOP_COVER,
self.assertEqual(state.attributes.get(ATTR_CURRENT_POSITION), 50) {ATTR_ENTITY_ID: COVER_GROUP}, blocking=True)
future = dt_util.utcnow() + timedelta(seconds=1)
async_fire_time_changed(hass, future)
await hass.async_block_till_done()
self.assertEqual(self.hass.states.get(DEMO_COVER).state, STATE_CLOSED) state = hass.states.get(COVER_GROUP)
self.assertEqual(self.hass.states.get(DEMO_COVER_POS) assert state.state == STATE_OPEN
.attributes.get(ATTR_CURRENT_POSITION), 50) assert state.attributes.get(ATTR_CURRENT_POSITION) == 100
self.assertEqual(self.hass.states.get(DEMO_COVER_TILT)
.attributes.get(ATTR_CURRENT_POSITION), 50)
def test_open_tilts(self): assert hass.states.get(DEMO_COVER).state == STATE_OPEN
"""Test open tilt function.""" assert hass.states.get(DEMO_COVER_POS) \
with assert_setup_component(2, DOMAIN): .attributes.get(ATTR_CURRENT_POSITION) == 20
assert setup.setup_component(self.hass, DOMAIN, CONFIG) assert hass.states.get(DEMO_COVER_TILT) \
.attributes.get(ATTR_CURRENT_POSITION) == 80
cover.open_cover_tilt(self.hass, COVER_GROUP)
self.hass.block_till_done()
for _ in range(5):
future = dt_util.utcnow() + timedelta(seconds=1)
fire_time_changed(self.hass, future)
self.hass.block_till_done()
state = self.hass.states.get(COVER_GROUP) async def test_set_cover_position(hass, setup_comp):
self.assertEqual(state.state, STATE_OPEN) """Test set cover position function."""
self.assertEqual(state.attributes.get(ATTR_CURRENT_TILT_POSITION), 100) await hass.services.async_call(
DOMAIN, SERVICE_SET_COVER_POSITION,
self.assertEqual(self.hass.states.get(DEMO_COVER_TILT) {ATTR_ENTITY_ID: COVER_GROUP, ATTR_POSITION: 50}, blocking=True)
.attributes.get(ATTR_CURRENT_TILT_POSITION), 100) for _ in range(4):
def test_close_tilts(self):
"""Test close tilt function."""
with assert_setup_component(2, DOMAIN):
assert setup.setup_component(self.hass, DOMAIN, CONFIG)
cover.close_cover_tilt(self.hass, COVER_GROUP)
self.hass.block_till_done()
for _ in range(5):
future = dt_util.utcnow() + timedelta(seconds=1)
fire_time_changed(self.hass, future)
self.hass.block_till_done()
state = self.hass.states.get(COVER_GROUP)
self.assertEqual(state.state, STATE_OPEN)
self.assertEqual(state.attributes.get(ATTR_CURRENT_TILT_POSITION), 0)
self.assertEqual(self.hass.states.get(DEMO_COVER_TILT)
.attributes.get(ATTR_CURRENT_TILT_POSITION), 0)
def test_stop_tilts(self):
"""Test stop tilts function."""
with assert_setup_component(2, DOMAIN):
assert setup.setup_component(self.hass, DOMAIN, CONFIG)
cover.open_cover_tilt(self.hass, COVER_GROUP)
self.hass.block_till_done()
future = dt_util.utcnow() + timedelta(seconds=1) future = dt_util.utcnow() + timedelta(seconds=1)
fire_time_changed(self.hass, future) async_fire_time_changed(hass, future)
self.hass.block_till_done() await hass.async_block_till_done()
cover.stop_cover_tilt(self.hass, COVER_GROUP)
self.hass.block_till_done() state = hass.states.get(COVER_GROUP)
assert state.state == STATE_OPEN
assert state.attributes.get(ATTR_CURRENT_POSITION) == 50
assert hass.states.get(DEMO_COVER).state == STATE_CLOSED
assert hass.states.get(DEMO_COVER_POS) \
.attributes.get(ATTR_CURRENT_POSITION) == 50
assert hass.states.get(DEMO_COVER_TILT) \
.attributes.get(ATTR_CURRENT_POSITION) == 50
async def test_open_tilts(hass, setup_comp):
"""Test open tilt function."""
await hass.services.async_call(
DOMAIN, SERVICE_OPEN_COVER_TILT,
{ATTR_ENTITY_ID: COVER_GROUP}, blocking=True)
for _ in range(5):
future = dt_util.utcnow() + timedelta(seconds=1) future = dt_util.utcnow() + timedelta(seconds=1)
fire_time_changed(self.hass, future) async_fire_time_changed(hass, future)
self.hass.block_till_done() await hass.async_block_till_done()
state = self.hass.states.get(COVER_GROUP) state = hass.states.get(COVER_GROUP)
self.assertEqual(state.state, STATE_OPEN) assert state.state == STATE_OPEN
self.assertEqual(state.attributes.get(ATTR_CURRENT_TILT_POSITION), 60) assert state.attributes.get(ATTR_CURRENT_TILT_POSITION) == 100
self.assertEqual(self.hass.states.get(DEMO_COVER_TILT) assert hass.states.get(DEMO_COVER_TILT) \
.attributes.get(ATTR_CURRENT_TILT_POSITION), 60) .attributes.get(ATTR_CURRENT_TILT_POSITION) == 100
def test_set_tilt_positions(self):
"""Test set tilt position function."""
with assert_setup_component(2, DOMAIN):
assert setup.setup_component(self.hass, DOMAIN, CONFIG)
cover.set_cover_tilt_position(self.hass, 80, COVER_GROUP) async def test_close_tilts(hass, setup_comp):
self.hass.block_till_done() """Test close tilt function."""
for _ in range(3): await hass.services.async_call(
future = dt_util.utcnow() + timedelta(seconds=1) DOMAIN, SERVICE_CLOSE_COVER_TILT,
fire_time_changed(self.hass, future) {ATTR_ENTITY_ID: COVER_GROUP}, blocking=True)
self.hass.block_till_done() for _ in range(5):
future = dt_util.utcnow() + timedelta(seconds=1)
async_fire_time_changed(hass, future)
await hass.async_block_till_done()
state = self.hass.states.get(COVER_GROUP) state = hass.states.get(COVER_GROUP)
self.assertEqual(state.state, STATE_OPEN) assert state.state == STATE_OPEN
self.assertEqual(state.attributes.get(ATTR_CURRENT_TILT_POSITION), 80) assert state.attributes.get(ATTR_CURRENT_TILT_POSITION) == 0
self.assertEqual(self.hass.states.get(DEMO_COVER_TILT) assert hass.states.get(DEMO_COVER_TILT) \
.attributes.get(ATTR_CURRENT_TILT_POSITION), 80) .attributes.get(ATTR_CURRENT_TILT_POSITION) == 0
async def test_stop_tilts(hass, setup_comp):
"""Test stop tilts function."""
await hass.services.async_call(
DOMAIN, SERVICE_OPEN_COVER_TILT,
{ATTR_ENTITY_ID: COVER_GROUP}, blocking=True)
future = dt_util.utcnow() + timedelta(seconds=1)
async_fire_time_changed(hass, future)
await hass.async_block_till_done()
await hass.services.async_call(
DOMAIN, SERVICE_STOP_COVER_TILT,
{ATTR_ENTITY_ID: COVER_GROUP}, blocking=True)
future = dt_util.utcnow() + timedelta(seconds=1)
async_fire_time_changed(hass, future)
await hass.async_block_till_done()
state = hass.states.get(COVER_GROUP)
assert state.state == STATE_OPEN
assert state.attributes.get(ATTR_CURRENT_TILT_POSITION) == 60
assert hass.states.get(DEMO_COVER_TILT) \
.attributes.get(ATTR_CURRENT_TILT_POSITION) == 60
async def test_set_tilt_positions(hass, setup_comp):
"""Test set tilt position function."""
await hass.services.async_call(
DOMAIN, SERVICE_SET_COVER_TILT_POSITION,
{ATTR_ENTITY_ID: COVER_GROUP, ATTR_TILT_POSITION: 80}, blocking=True)
for _ in range(3):
future = dt_util.utcnow() + timedelta(seconds=1)
async_fire_time_changed(hass, future)
await hass.async_block_till_done()
state = hass.states.get(COVER_GROUP)
assert state.state == STATE_OPEN
assert state.attributes.get(ATTR_CURRENT_TILT_POSITION) == 80
assert hass.states.get(DEMO_COVER_TILT) \
.attributes.get(ATTR_CURRENT_TILT_POSITION) == 80

65
tests/components/cover/test_mqtt.py
View file

@ -1,12 +1,17 @@
"""The tests for the MQTT cover platform.""" """The tests for the MQTT cover platform."""
import unittest import unittest
from homeassistant.setup import setup_component
from homeassistant.const import STATE_OPEN, STATE_CLOSED, STATE_UNKNOWN, \
STATE_UNAVAILABLE, ATTR_ASSUMED_STATE
import homeassistant.components.cover as cover import homeassistant.components.cover as cover
from homeassistant.components.cover import (ATTR_POSITION, ATTR_TILT_POSITION)
from homeassistant.components.cover.mqtt import MqttCover from homeassistant.components.cover.mqtt import MqttCover
from homeassistant.components.mqtt.discovery import async_start from homeassistant.components.mqtt.discovery import async_start
from homeassistant.const import (
ATTR_ASSUMED_STATE, ATTR_ENTITY_ID,
SERVICE_CLOSE_COVER, SERVICE_CLOSE_COVER_TILT, SERVICE_OPEN_COVER,
SERVICE_OPEN_COVER_TILT, SERVICE_SET_COVER_POSITION,
SERVICE_SET_COVER_TILT_POSITION, SERVICE_STOP_COVER,
STATE_CLOSED, STATE_OPEN, STATE_UNAVAILABLE, STATE_UNKNOWN)
from homeassistant.setup import setup_component
from tests.common import ( from tests.common import (
get_test_home_assistant, mock_mqtt_component, async_fire_mqtt_message, get_test_home_assistant, mock_mqtt_component, async_fire_mqtt_message,
@ -117,7 +122,9 @@ class TestCoverMQTT(unittest.TestCase):
self.assertEqual(STATE_UNKNOWN, state.state) self.assertEqual(STATE_UNKNOWN, state.state)
self.assertTrue(state.attributes.get(ATTR_ASSUMED_STATE)) self.assertTrue(state.attributes.get(ATTR_ASSUMED_STATE))
cover.open_cover(self.hass, 'cover.test') self.hass.services.call(
cover.DOMAIN, SERVICE_OPEN_COVER,
{ATTR_ENTITY_ID: 'cover.test'}, blocking=True)
self.hass.block_till_done() self.hass.block_till_done()
self.mock_publish.async_publish.assert_called_once_with( self.mock_publish.async_publish.assert_called_once_with(
@ -126,7 +133,9 @@ class TestCoverMQTT(unittest.TestCase):
state = self.hass.states.get('cover.test') state = self.hass.states.get('cover.test')
self.assertEqual(STATE_OPEN, state.state) self.assertEqual(STATE_OPEN, state.state)
cover.close_cover(self.hass, 'cover.test') self.hass.services.call(
cover.DOMAIN, SERVICE_CLOSE_COVER,
{ATTR_ENTITY_ID: 'cover.test'}, blocking=True)
self.hass.block_till_done() self.hass.block_till_done()
self.mock_publish.async_publish.assert_called_once_with( self.mock_publish.async_publish.assert_called_once_with(
@ -149,7 +158,9 @@ class TestCoverMQTT(unittest.TestCase):
state = self.hass.states.get('cover.test') state = self.hass.states.get('cover.test')
self.assertEqual(STATE_UNKNOWN, state.state) self.assertEqual(STATE_UNKNOWN, state.state)
cover.open_cover(self.hass, 'cover.test') self.hass.services.call(
cover.DOMAIN, SERVICE_OPEN_COVER,
{ATTR_ENTITY_ID: 'cover.test'}, blocking=True)
self.hass.block_till_done() self.hass.block_till_done()
self.mock_publish.async_publish.assert_called_once_with( self.mock_publish.async_publish.assert_called_once_with(
@ -172,7 +183,9 @@ class TestCoverMQTT(unittest.TestCase):
state = self.hass.states.get('cover.test') state = self.hass.states.get('cover.test')
self.assertEqual(STATE_UNKNOWN, state.state) self.assertEqual(STATE_UNKNOWN, state.state)
cover.close_cover(self.hass, 'cover.test') self.hass.services.call(
cover.DOMAIN, SERVICE_CLOSE_COVER,
{ATTR_ENTITY_ID: 'cover.test'}, blocking=True)
self.hass.block_till_done() self.hass.block_till_done()
self.mock_publish.async_publish.assert_called_once_with( self.mock_publish.async_publish.assert_called_once_with(
@ -195,7 +208,9 @@ class TestCoverMQTT(unittest.TestCase):
state = self.hass.states.get('cover.test') state = self.hass.states.get('cover.test')
self.assertEqual(STATE_UNKNOWN, state.state) self.assertEqual(STATE_UNKNOWN, state.state)
cover.stop_cover(self.hass, 'cover.test') self.hass.services.call(
cover.DOMAIN, SERVICE_STOP_COVER,
{ATTR_ENTITY_ID: 'cover.test'}, blocking=True)
self.hass.block_till_done() self.hass.block_till_done()
self.mock_publish.async_publish.assert_called_once_with( self.mock_publish.async_publish.assert_called_once_with(
@ -297,7 +312,9 @@ class TestCoverMQTT(unittest.TestCase):
} }
})) }))
cover.set_cover_position(self.hass, 100, 'cover.test') self.hass.services.call(
cover.DOMAIN, SERVICE_SET_COVER_POSITION,
{ATTR_ENTITY_ID: 'cover.test', ATTR_POSITION: 100}, blocking=True)
self.hass.block_till_done() self.hass.block_till_done()
self.mock_publish.async_publish.assert_called_once_with( self.mock_publish.async_publish.assert_called_once_with(
@ -318,7 +335,9 @@ class TestCoverMQTT(unittest.TestCase):
} }
})) }))
cover.set_cover_position(self.hass, 62, 'cover.test') self.hass.services.call(
cover.DOMAIN, SERVICE_SET_COVER_POSITION,
{ATTR_ENTITY_ID: 'cover.test', ATTR_POSITION: 62}, blocking=True)
self.hass.block_till_done() self.hass.block_till_done()
self.mock_publish.async_publish.assert_called_once_with( self.mock_publish.async_publish.assert_called_once_with(
@ -403,14 +422,18 @@ class TestCoverMQTT(unittest.TestCase):
} }
})) }))
cover.open_cover_tilt(self.hass, 'cover.test') self.hass.services.call(
cover.DOMAIN, SERVICE_OPEN_COVER_TILT,
{ATTR_ENTITY_ID: 'cover.test'}, blocking=True)
self.hass.block_till_done() self.hass.block_till_done()
self.mock_publish.async_publish.assert_called_once_with( self.mock_publish.async_publish.assert_called_once_with(
'tilt-command-topic', 100, 0, False) 'tilt-command-topic', 100, 0, False)
self.mock_publish.async_publish.reset_mock() self.mock_publish.async_publish.reset_mock()
cover.close_cover_tilt(self.hass, 'cover.test') self.hass.services.call(
cover.DOMAIN, SERVICE_CLOSE_COVER_TILT,
{ATTR_ENTITY_ID: 'cover.test'}, blocking=True)
self.hass.block_till_done() self.hass.block_till_done()
self.mock_publish.async_publish.assert_called_once_with( self.mock_publish.async_publish.assert_called_once_with(
@ -435,14 +458,18 @@ class TestCoverMQTT(unittest.TestCase):
} }
})) }))
cover.open_cover_tilt(self.hass, 'cover.test') self.hass.services.call(
cover.DOMAIN, SERVICE_OPEN_COVER_TILT,
{ATTR_ENTITY_ID: 'cover.test'}, blocking=True)
self.hass.block_till_done() self.hass.block_till_done()
self.mock_publish.async_publish.assert_called_once_with( self.mock_publish.async_publish.assert_called_once_with(
'tilt-command-topic', 400, 0, False) 'tilt-command-topic', 400, 0, False)
self.mock_publish.async_publish.reset_mock() self.mock_publish.async_publish.reset_mock()
cover.close_cover_tilt(self.hass, 'cover.test') self.hass.services.call(
cover.DOMAIN, SERVICE_CLOSE_COVER_TILT,
{ATTR_ENTITY_ID: 'cover.test'}, blocking=True)
self.hass.block_till_done() self.hass.block_till_done()
self.mock_publish.async_publish.assert_called_once_with( self.mock_publish.async_publish.assert_called_once_with(
@ -542,7 +569,10 @@ class TestCoverMQTT(unittest.TestCase):
} }
})) }))
cover.set_cover_tilt_position(self.hass, 50, 'cover.test') self.hass.services.call(
cover.DOMAIN, SERVICE_SET_COVER_TILT_POSITION,
{ATTR_ENTITY_ID: 'cover.test', ATTR_TILT_POSITION: 50},
blocking=True)
self.hass.block_till_done() self.hass.block_till_done()
self.mock_publish.async_publish.assert_called_once_with( self.mock_publish.async_publish.assert_called_once_with(
@ -569,7 +599,10 @@ class TestCoverMQTT(unittest.TestCase):
} }
})) }))
cover.set_cover_tilt_position(self.hass, 50, 'cover.test') self.hass.services.call(
cover.DOMAIN, SERVICE_SET_COVER_TILT_POSITION,
{ATTR_ENTITY_ID: 'cover.test', ATTR_TILT_POSITION: 50},
blocking=True)
self.hass.block_till_done() self.hass.block_till_done()
self.mock_publish.async_publish.assert_called_once_with( self.mock_publish.async_publish.assert_called_once_with(

81
tests/components/cover/test_template.py
View file

@ -1,17 +1,24 @@
"""The tests the cover command line platform.""" """The tests the cover command line platform."""
import logging import logging
import unittest import unittest
from homeassistant.core import callback
from homeassistant import setup from homeassistant import setup
import homeassistant.components.cover as cover from homeassistant.core import callback
from homeassistant.const import STATE_OPEN, STATE_CLOSED from homeassistant.components.cover import (
ATTR_POSITION, ATTR_TILT_POSITION, DOMAIN)
from homeassistant.const import (
ATTR_ENTITY_ID, SERVICE_CLOSE_COVER, SERVICE_CLOSE_COVER_TILT,
SERVICE_OPEN_COVER, SERVICE_OPEN_COVER_TILT, SERVICE_SET_COVER_POSITION,
SERVICE_SET_COVER_TILT_POSITION, SERVICE_STOP_COVER,
STATE_CLOSED, STATE_OPEN)
from tests.common import ( from tests.common import (
get_test_home_assistant, assert_setup_component) get_test_home_assistant, assert_setup_component)
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
ENTITY_COVER = 'cover.test_template_cover'
class TestTemplateCover(unittest.TestCase): class TestTemplateCover(unittest.TestCase):
"""Test the cover command line platform.""" """Test the cover command line platform."""
@ -362,7 +369,9 @@ class TestTemplateCover(unittest.TestCase):
state = self.hass.states.get('cover.test_template_cover') state = self.hass.states.get('cover.test_template_cover')
assert state.state == STATE_CLOSED assert state.state == STATE_CLOSED
cover.open_cover(self.hass, 'cover.test_template_cover') self.hass.services.call(
DOMAIN, SERVICE_OPEN_COVER,
{ATTR_ENTITY_ID: ENTITY_COVER}, blocking=True)
self.hass.block_till_done() self.hass.block_till_done()
assert len(self.calls) == 1 assert len(self.calls) == 1
@ -398,10 +407,14 @@ class TestTemplateCover(unittest.TestCase):
state = self.hass.states.get('cover.test_template_cover') state = self.hass.states.get('cover.test_template_cover')
assert state.state == STATE_OPEN assert state.state == STATE_OPEN
cover.close_cover(self.hass, 'cover.test_template_cover') self.hass.services.call(
DOMAIN, SERVICE_CLOSE_COVER,
{ATTR_ENTITY_ID: ENTITY_COVER}, blocking=True)
self.hass.block_till_done() self.hass.block_till_done()
cover.stop_cover(self.hass, 'cover.test_template_cover') self.hass.services.call(
DOMAIN, SERVICE_STOP_COVER,
{ATTR_ENTITY_ID: ENTITY_COVER}, blocking=True)
self.hass.block_till_done() self.hass.block_till_done()
assert len(self.calls) == 2 assert len(self.calls) == 2
@ -445,18 +458,23 @@ class TestTemplateCover(unittest.TestCase):
state = self.hass.states.get('cover.test_template_cover') state = self.hass.states.get('cover.test_template_cover')
assert state.state == STATE_OPEN assert state.state == STATE_OPEN
cover.open_cover(self.hass, 'cover.test_template_cover') self.hass.services.call(
DOMAIN, SERVICE_OPEN_COVER,
{ATTR_ENTITY_ID: ENTITY_COVER}, blocking=True)
self.hass.block_till_done() self.hass.block_till_done()
state = self.hass.states.get('cover.test_template_cover') state = self.hass.states.get('cover.test_template_cover')
assert state.attributes.get('current_position') == 100.0 assert state.attributes.get('current_position') == 100.0
cover.close_cover(self.hass, 'cover.test_template_cover') self.hass.services.call(
DOMAIN, SERVICE_CLOSE_COVER,
{ATTR_ENTITY_ID: ENTITY_COVER}, blocking=True)
self.hass.block_till_done() self.hass.block_till_done()
state = self.hass.states.get('cover.test_template_cover') state = self.hass.states.get('cover.test_template_cover')
assert state.attributes.get('current_position') == 0.0 assert state.attributes.get('current_position') == 0.0
cover.set_cover_position(self.hass, 25, self.hass.services.call(
'cover.test_template_cover') DOMAIN, SERVICE_SET_COVER_POSITION,
{ATTR_ENTITY_ID: ENTITY_COVER, ATTR_POSITION: 25}, blocking=True)
self.hass.block_till_done() self.hass.block_till_done()
state = self.hass.states.get('cover.test_template_cover') state = self.hass.states.get('cover.test_template_cover')
assert state.attributes.get('current_position') == 25.0 assert state.attributes.get('current_position') == 25.0
@ -490,8 +508,10 @@ class TestTemplateCover(unittest.TestCase):
self.hass.start() self.hass.start()
self.hass.block_till_done() self.hass.block_till_done()
cover.set_cover_tilt_position(self.hass, 42, self.hass.services.call(
'cover.test_template_cover') DOMAIN, SERVICE_SET_COVER_TILT_POSITION,
{ATTR_ENTITY_ID: ENTITY_COVER, ATTR_TILT_POSITION: 42},
blocking=True)
self.hass.block_till_done() self.hass.block_till_done()
assert len(self.calls) == 1 assert len(self.calls) == 1
@ -525,7 +545,9 @@ class TestTemplateCover(unittest.TestCase):
self.hass.start() self.hass.start()
self.hass.block_till_done() self.hass.block_till_done()
cover.open_cover_tilt(self.hass, 'cover.test_template_cover') self.hass.services.call(
DOMAIN, SERVICE_OPEN_COVER_TILT,
{ATTR_ENTITY_ID: ENTITY_COVER}, blocking=True)
self.hass.block_till_done() self.hass.block_till_done()
assert len(self.calls) == 1 assert len(self.calls) == 1
@ -559,7 +581,9 @@ class TestTemplateCover(unittest.TestCase):
self.hass.start() self.hass.start()
self.hass.block_till_done() self.hass.block_till_done()
cover.close_cover_tilt(self.hass, 'cover.test_template_cover') self.hass.services.call(
DOMAIN, SERVICE_CLOSE_COVER_TILT,
{ATTR_ENTITY_ID: ENTITY_COVER}, blocking=True)
self.hass.block_till_done() self.hass.block_till_done()
assert len(self.calls) == 1 assert len(self.calls) == 1
@ -585,18 +609,23 @@ class TestTemplateCover(unittest.TestCase):
state = self.hass.states.get('cover.test_template_cover') state = self.hass.states.get('cover.test_template_cover')
assert state.attributes.get('current_position') is None assert state.attributes.get('current_position') is None
cover.set_cover_position(self.hass, 42, self.hass.services.call(
'cover.test_template_cover') DOMAIN, SERVICE_SET_COVER_POSITION,
{ATTR_ENTITY_ID: ENTITY_COVER, ATTR_POSITION: 42}, blocking=True)
self.hass.block_till_done() self.hass.block_till_done()
state = self.hass.states.get('cover.test_template_cover') state = self.hass.states.get('cover.test_template_cover')
assert state.attributes.get('current_position') == 42.0 assert state.attributes.get('current_position') == 42.0
cover.close_cover(self.hass, 'cover.test_template_cover') self.hass.services.call(
DOMAIN, SERVICE_CLOSE_COVER,
{ATTR_ENTITY_ID: ENTITY_COVER}, blocking=True)
self.hass.block_till_done() self.hass.block_till_done()
state = self.hass.states.get('cover.test_template_cover') state = self.hass.states.get('cover.test_template_cover')
assert state.state == STATE_CLOSED assert state.state == STATE_CLOSED
cover.open_cover(self.hass, 'cover.test_template_cover') self.hass.services.call(
DOMAIN, SERVICE_OPEN_COVER,
{ATTR_ENTITY_ID: ENTITY_COVER}, blocking=True)
self.hass.block_till_done() self.hass.block_till_done()
state = self.hass.states.get('cover.test_template_cover') state = self.hass.states.get('cover.test_template_cover')
assert state.state == STATE_OPEN assert state.state == STATE_OPEN
@ -627,18 +656,24 @@ class TestTemplateCover(unittest.TestCase):
state = self.hass.states.get('cover.test_template_cover') state = self.hass.states.get('cover.test_template_cover')
assert state.attributes.get('current_tilt_position') is None assert state.attributes.get('current_tilt_position') is None
cover.set_cover_tilt_position(self.hass, 42, self.hass.services.call(
'cover.test_template_cover') DOMAIN, SERVICE_SET_COVER_TILT_POSITION,
{ATTR_ENTITY_ID: ENTITY_COVER, ATTR_TILT_POSITION: 42},
blocking=True)
self.hass.block_till_done() self.hass.block_till_done()
state = self.hass.states.get('cover.test_template_cover') state = self.hass.states.get('cover.test_template_cover')
assert state.attributes.get('current_tilt_position') == 42.0 assert state.attributes.get('current_tilt_position') == 42.0
cover.close_cover_tilt(self.hass, 'cover.test_template_cover') self.hass.services.call(
DOMAIN, SERVICE_CLOSE_COVER_TILT,
{ATTR_ENTITY_ID: ENTITY_COVER}, blocking=True)
self.hass.block_till_done() self.hass.block_till_done()
state = self.hass.states.get('cover.test_template_cover') state = self.hass.states.get('cover.test_template_cover')
assert state.attributes.get('current_tilt_position') == 0.0 assert state.attributes.get('current_tilt_position') == 0.0
cover.open_cover_tilt(self.hass, 'cover.test_template_cover') self.hass.services.call(
DOMAIN, SERVICE_OPEN_COVER_TILT,
{ATTR_ENTITY_ID: ENTITY_COVER}, blocking=True)
self.hass.block_till_done() self.hass.block_till_done()
state = self.hass.states.get('cover.test_template_cover') state = self.hass.states.get('cover.test_template_cover')
assert state.attributes.get('current_tilt_position') == 100.0 assert state.attributes.get('current_tilt_position') == 100.0