"""The tests for the Template cover platform."""
import pytest
from homeassistant import setup
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_TOGGLE,
SERVICE_TOGGLE_COVER_TILT,
STATE_CLOSED,
STATE_CLOSING,
STATE_OFF,
STATE_ON,
STATE_OPEN,
STATE_OPENING,
STATE_UNAVAILABLE,
)
from tests.common import assert_setup_component
ENTITY_COVER = "cover.test_template_cover"
OPEN_CLOSE_COVER_CONFIG = {
"open_cover": {
"service": "test.automation",
"data_template": {
"action": "open_cover",
"caller": "{{ this.entity_id }}",
},
},
"close_cover": {
"service": "test.automation",
"data_template": {
"action": "close_cover",
"caller": "{{ this.entity_id }}",
},
},
}
@pytest.mark.parametrize("count,domain", [(1, DOMAIN)])
@pytest.mark.parametrize(
"config, states",
[
(
{
DOMAIN: {
"platform": "template",
"covers": {
"test_template_cover": {
**OPEN_CLOSE_COVER_CONFIG,
"value_template": "{{ states.cover.test_state.state }}",
}
},
}
},
[
("cover.test_state", STATE_OPEN, STATE_OPEN, {}, -1, ""),
("cover.test_state", STATE_CLOSED, STATE_CLOSED, {}, -1, ""),
("cover.test_state", STATE_OPENING, STATE_OPENING, {}, -1, ""),
("cover.test_state", STATE_CLOSING, STATE_CLOSING, {}, -1, ""),
(
"cover.test_state",
"dog",
STATE_CLOSING,
{},
-1,
"Received invalid cover is_on state: dog",
),
("cover.test_state", STATE_OPEN, STATE_OPEN, {}, -1, ""),
(
"cover.test_state",
"cat",
STATE_OPEN,
{},
-1,
"Received invalid cover is_on state: cat",
),
("cover.test_state", STATE_CLOSED, STATE_CLOSED, {}, -1, ""),
(
"cover.test_state",
"bear",
STATE_OPEN,
{},
-1,
"Received invalid cover is_on state: bear",
),
],
),
(
{
DOMAIN: {
"platform": "template",
"covers": {
"test_template_cover": {
**OPEN_CLOSE_COVER_CONFIG,
"position_template": "{{ states.cover.test.attributes.position }}",
"value_template": "{{ states.cover.test_state.state }}",
}
},
}
},
[
("cover.test_state", STATE_OPEN, STATE_OPEN, {}, -1, ""),
("cover.test_state", STATE_CLOSED, STATE_OPEN, {}, -1, ""),
("cover.test_state", STATE_OPENING, STATE_OPENING, {}, -1, ""),
("cover.test_state", STATE_CLOSING, STATE_CLOSING, {}, -1, ""),
("cover.test", STATE_CLOSED, STATE_CLOSING, {"position": 0}, 0, ""),
("cover.test_state", STATE_OPEN, STATE_CLOSED, {}, -1, ""),
("cover.test", STATE_CLOSED, STATE_OPEN, {"position": 10}, 10, ""),
(
"cover.test_state",
"dog",
STATE_OPEN,
{},
-1,
"Received invalid cover is_on state: dog",
),
],
),
],
)
async def test_template_state_text(hass, states, start_ha, caplog):
"""Test the state text of a template."""
state = hass.states.get("cover.test_template_cover")
assert state.state == STATE_OPEN
for entity, set_state, test_state, attr, pos, text in states:
hass.states.async_set(entity, set_state, attributes=attr)
await hass.async_block_till_done()
state = hass.states.get("cover.test_template_cover")
assert state.state == test_state
if pos >= 0:
assert state.attributes.get("current_position") == pos
assert text in caplog.text
@pytest.mark.parametrize("count,domain", [(1, DOMAIN)])
@pytest.mark.parametrize(
"config",
[
{
DOMAIN: {
"platform": "template",
"covers": {
"test_template_cover": {
**OPEN_CLOSE_COVER_CONFIG,
"value_template": "{{ 1 == 1 }}",
}
},
}
},
],
)
async def test_template_state_boolean(hass, start_ha):
"""Test the value_template attribute."""
state = hass.states.get("cover.test_template_cover")
assert state.state == STATE_OPEN
@pytest.mark.parametrize("count,domain", [(1, DOMAIN)])
@pytest.mark.parametrize(
"config",
[
{
DOMAIN: {
"platform": "template",
"covers": {
"test_template_cover": {
**OPEN_CLOSE_COVER_CONFIG,
"position_template": "{{ states.cover.test.attributes.position }}",
}
},
}
},
],
)
async def test_template_position(hass, start_ha):
"""Test the position_template attribute."""
hass.states.async_set("cover.test", STATE_OPEN)
attrs = {}
for set_state, pos, test_state in [
(STATE_CLOSED, 42, STATE_OPEN),
(STATE_OPEN, 0.0, STATE_CLOSED),
]:
attrs["position"] = pos
hass.states.async_set("cover.test", set_state, attributes=attrs)
await hass.async_block_till_done()
state = hass.states.get("cover.test_template_cover")
assert state.attributes.get("current_position") == pos
assert state.state == test_state
@pytest.mark.parametrize("count,domain", [(1, DOMAIN)])
@pytest.mark.parametrize(
"config",
[
{
DOMAIN: {
"platform": "template",
"covers": {
"test_template_cover": {
**OPEN_CLOSE_COVER_CONFIG,
"value_template": "{{ 1 == 1 }}",
"tilt_template": "{{ 42 }}",
}
},
}
},
],
)
async def test_template_tilt(hass, start_ha):
"""Test the tilt_template attribute."""
state = hass.states.get("cover.test_template_cover")
assert state.attributes.get("current_tilt_position") == 42.0
@pytest.mark.parametrize("count,domain", [(1, DOMAIN)])
@pytest.mark.parametrize(
"config",
[
{
DOMAIN: {
"platform": "template",
"covers": {
"test_template_cover": {
**OPEN_CLOSE_COVER_CONFIG,
"position_template": "{{ -1 }}",
"tilt_template": "{{ 110 }}",
}
},
}
},
{
DOMAIN: {
"platform": "template",
"covers": {
"test_template_cover": {
**OPEN_CLOSE_COVER_CONFIG,
"position_template": "{{ on }}",
"tilt_template": "{% if states.cover.test_state.state %}"
"on"
"{% else %}"
"off"
"{% endif %}",
},
},
}
},
],
)
async def test_template_out_of_bounds(hass, start_ha):
"""Test template out-of-bounds condition."""
state = hass.states.get("cover.test_template_cover")
assert state.attributes.get("current_tilt_position") is None
assert state.attributes.get("current_position") is None
@pytest.mark.parametrize("count,domain", [(0, DOMAIN)])
@pytest.mark.parametrize(
"config",
[
{
DOMAIN: {
"platform": "template",
"covers": {"test_template_cover": {"value_template": "{{ 1 == 1 }}"}},
}
},
{
DOMAIN: {
"platform": "template",
"covers": {
"test_template_cover": {
"value_template": "{{ 1 == 1 }}",
"open_cover": {
"service": "test.automation",
"data_template": {
"action": "open_cover",
"caller": "{{ this.entity_id }}",
},
},
}
},
}
},
],
)
async def test_template_open_or_position(hass, start_ha, caplog_setup_text):
"""Test that at least one of open_cover or set_position is used."""
assert hass.states.async_all("cover") == []
assert "Invalid config for [cover.template]" in caplog_setup_text
@pytest.mark.parametrize("count,domain", [(1, DOMAIN)])
@pytest.mark.parametrize(
"config",
[
{
DOMAIN: {
"platform": "template",
"covers": {
"test_template_cover": {
**OPEN_CLOSE_COVER_CONFIG,
"position_template": "{{ 0 }}",
}
},
}
},
],
)
async def test_open_action(hass, start_ha, calls):
"""Test the open_cover command."""
state = hass.states.get("cover.test_template_cover")
assert state.state == STATE_CLOSED
await hass.services.async_call(
DOMAIN, SERVICE_OPEN_COVER, {ATTR_ENTITY_ID: ENTITY_COVER}, blocking=True
)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["action"] == "open_cover"
assert calls[0].data["caller"] == "cover.test_template_cover"
@pytest.mark.parametrize("count,domain", [(1, DOMAIN)])
@pytest.mark.parametrize(
"config",
[
{
DOMAIN: {
"platform": "template",
"covers": {
"test_template_cover": {
**OPEN_CLOSE_COVER_CONFIG,
"position_template": "{{ 100 }}",
"stop_cover": {
"service": "test.automation",
"data_template": {
"action": "stop_cover",
"caller": "{{ this.entity_id }}",
},
},
}
},
}
},
],
)
async def test_close_stop_action(hass, start_ha, calls):
"""Test the close-cover and stop_cover commands."""
state = hass.states.get("cover.test_template_cover")
assert state.state == STATE_OPEN
await hass.services.async_call(
DOMAIN, SERVICE_CLOSE_COVER, {ATTR_ENTITY_ID: ENTITY_COVER}, blocking=True
)
await hass.async_block_till_done()
await hass.services.async_call(
DOMAIN, SERVICE_STOP_COVER, {ATTR_ENTITY_ID: ENTITY_COVER}, blocking=True
)
await hass.async_block_till_done()
assert len(calls) == 2
assert calls[0].data["action"] == "close_cover"
assert calls[0].data["caller"] == "cover.test_template_cover"
assert calls[1].data["action"] == "stop_cover"
assert calls[1].data["caller"] == "cover.test_template_cover"
@pytest.mark.parametrize("count,domain", [(1, "input_number")])
@pytest.mark.parametrize(
"config",
[
{"input_number": {"test": {"min": "0", "max": "100", "initial": "42"}}},
],
)
async def test_set_position(hass, start_ha, calls):
"""Test the set_position command."""
with assert_setup_component(1, "cover"):
assert await setup.async_setup_component(
hass,
"cover",
{
"cover": {
"platform": "template",
"covers": {
"test_template_cover": {
"set_cover_position": {
"service": "test.automation",
"data_template": {
"action": "set_cover_position",
"caller": "{{ this.entity_id }}",
"position": "{{ position }}",
},
},
}
},
}
},
)
await hass.async_block_till_done()
await hass.async_start()
await hass.async_block_till_done()
state = hass.states.async_set("input_number.test", 42)
await hass.async_block_till_done()
state = hass.states.get("cover.test_template_cover")
assert state.state == STATE_OPEN
await hass.services.async_call(
DOMAIN, SERVICE_OPEN_COVER, {ATTR_ENTITY_ID: ENTITY_COVER}, blocking=True
)
await hass.async_block_till_done()
state = hass.states.get("cover.test_template_cover")
assert state.attributes.get("current_position") == 100.0
assert len(calls) == 1
assert calls[-1].data["action"] == "set_cover_position"
assert calls[-1].data["caller"] == "cover.test_template_cover"
assert calls[-1].data["position"] == 100
await hass.services.async_call(
DOMAIN, SERVICE_CLOSE_COVER, {ATTR_ENTITY_ID: ENTITY_COVER}, blocking=True
)
await hass.async_block_till_done()
state = hass.states.get("cover.test_template_cover")
assert state.attributes.get("current_position") == 0.0
assert len(calls) == 2
assert calls[-1].data["action"] == "set_cover_position"
assert calls[-1].data["caller"] == "cover.test_template_cover"
assert calls[-1].data["position"] == 0
await hass.services.async_call(
DOMAIN, SERVICE_TOGGLE, {ATTR_ENTITY_ID: ENTITY_COVER}, blocking=True
)
await hass.async_block_till_done()
state = hass.states.get("cover.test_template_cover")
assert state.attributes.get("current_position") == 100.0
assert len(calls) == 3
assert calls[-1].data["action"] == "set_cover_position"
assert calls[-1].data["caller"] == "cover.test_template_cover"
assert calls[-1].data["position"] == 100
await hass.services.async_call(
DOMAIN, SERVICE_TOGGLE, {ATTR_ENTITY_ID: ENTITY_COVER}, blocking=True
)
await hass.async_block_till_done()
state = hass.states.get("cover.test_template_cover")
assert state.attributes.get("current_position") == 0.0
assert len(calls) == 4
assert calls[-1].data["action"] == "set_cover_position"
assert calls[-1].data["caller"] == "cover.test_template_cover"
assert calls[-1].data["position"] == 0
await hass.services.async_call(
DOMAIN,
SERVICE_SET_COVER_POSITION,
{ATTR_ENTITY_ID: ENTITY_COVER, ATTR_POSITION: 25},
blocking=True,
)
await hass.async_block_till_done()
state = hass.states.get("cover.test_template_cover")
assert state.attributes.get("current_position") == 25.0
assert len(calls) == 5
assert calls[-1].data["action"] == "set_cover_position"
assert calls[-1].data["caller"] == "cover.test_template_cover"
assert calls[-1].data["position"] == 25
@pytest.mark.parametrize("count,domain", [(1, DOMAIN)])
@pytest.mark.parametrize(
"config",
[
{
DOMAIN: {
"platform": "template",
"covers": {
"test_template_cover": {
**OPEN_CLOSE_COVER_CONFIG,
"set_cover_tilt_position": {
"service": "test.automation",
"data_template": {
"action": "set_cover_tilt_position",
"caller": "{{ this.entity_id }}",
"tilt_position": "{{ tilt }}",
},
},
}
},
}
},
],
)
@pytest.mark.parametrize(
"service,attr,tilt_position",
[
(
SERVICE_SET_COVER_TILT_POSITION,
{ATTR_ENTITY_ID: ENTITY_COVER, ATTR_TILT_POSITION: 42},
42,
),
(SERVICE_OPEN_COVER_TILT, {ATTR_ENTITY_ID: ENTITY_COVER}, 100),
(SERVICE_CLOSE_COVER_TILT, {ATTR_ENTITY_ID: ENTITY_COVER}, 0),
],
)
async def test_set_tilt_position(hass, service, attr, start_ha, calls, tilt_position):
"""Test the set_tilt_position command."""
await hass.services.async_call(
DOMAIN,
service,
attr,
blocking=True,
)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[-1].data["action"] == "set_cover_tilt_position"
assert calls[-1].data["caller"] == "cover.test_template_cover"
assert calls[-1].data["tilt_position"] == tilt_position
@pytest.mark.parametrize("count,domain", [(1, DOMAIN)])
@pytest.mark.parametrize(
"config",
[
{
DOMAIN: {
"platform": "template",
"covers": {
"test_template_cover": {
"set_cover_position": {"service": "test.automation"}
}
},
}
},
],
)
async def test_set_position_optimistic(hass, start_ha, calls):
"""Test optimistic position mode."""
state = hass.states.get("cover.test_template_cover")
assert state.attributes.get("current_position") is None
await hass.services.async_call(
DOMAIN,
SERVICE_SET_COVER_POSITION,
{ATTR_ENTITY_ID: ENTITY_COVER, ATTR_POSITION: 42},
blocking=True,
)
await hass.async_block_till_done()
state = hass.states.get("cover.test_template_cover")
assert state.attributes.get("current_position") == 42.0
for service, test_state in [
(SERVICE_CLOSE_COVER, STATE_CLOSED),
(SERVICE_OPEN_COVER, STATE_OPEN),
(SERVICE_TOGGLE, STATE_CLOSED),
(SERVICE_TOGGLE, STATE_OPEN),
]:
await hass.services.async_call(
DOMAIN, service, {ATTR_ENTITY_ID: ENTITY_COVER}, blocking=True
)
await hass.async_block_till_done()
state = hass.states.get("cover.test_template_cover")
assert state.state == test_state
@pytest.mark.parametrize("count,domain", [(1, DOMAIN)])
@pytest.mark.parametrize(
"config",
[
{
DOMAIN: {
"platform": "template",
"covers": {
"test_template_cover": {
"position_template": "{{ 100 }}",
"set_cover_position": {"service": "test.automation"},
"set_cover_tilt_position": {"service": "test.automation"},
}
},
}
},
],
)
async def test_set_tilt_position_optimistic(hass, start_ha, calls):
"""Test the optimistic tilt_position mode."""
state = hass.states.get("cover.test_template_cover")
assert state.attributes.get("current_tilt_position") is None
await hass.services.async_call(
DOMAIN,
SERVICE_SET_COVER_TILT_POSITION,
{ATTR_ENTITY_ID: ENTITY_COVER, ATTR_TILT_POSITION: 42},
blocking=True,
)
await hass.async_block_till_done()
state = hass.states.get("cover.test_template_cover")
assert state.attributes.get("current_tilt_position") == 42.0
for service, pos in [
(SERVICE_CLOSE_COVER_TILT, 0.0),
(SERVICE_OPEN_COVER_TILT, 100.0),
(SERVICE_TOGGLE_COVER_TILT, 0.0),
(SERVICE_TOGGLE_COVER_TILT, 100.0),
]:
await hass.services.async_call(
DOMAIN, service, {ATTR_ENTITY_ID: ENTITY_COVER}, blocking=True
)
await hass.async_block_till_done()
state = hass.states.get("cover.test_template_cover")
assert state.attributes.get("current_tilt_position") == pos
@pytest.mark.parametrize("count,domain", [(1, DOMAIN)])
@pytest.mark.parametrize(
"config",
[
{
DOMAIN: {
"platform": "template",
"covers": {
"test_template_cover": {
**OPEN_CLOSE_COVER_CONFIG,
"value_template": "{{ states.cover.test_state.state }}",
"icon_template": "{% if states.cover.test_state.state %}"
"mdi:check"
"{% endif %}",
}
},
}
},
],
)
async def test_icon_template(hass, start_ha):
"""Test icon template."""
state = hass.states.get("cover.test_template_cover")
assert state.attributes.get("icon") == ""
state = hass.states.async_set("cover.test_state", STATE_OPEN)
await hass.async_block_till_done()
state = hass.states.get("cover.test_template_cover")
assert state.attributes["icon"] == "mdi:check"
@pytest.mark.parametrize("count,domain", [(1, DOMAIN)])
@pytest.mark.parametrize(
"config",
[
{
DOMAIN: {
"platform": "template",
"covers": {
"test_template_cover": {
**OPEN_CLOSE_COVER_CONFIG,
"value_template": "{{ states.cover.test_state.state }}",
"entity_picture_template": "{% if states.cover.test_state.state %}"
"/local/cover.png"
"{% endif %}",
}
},
}
},
],
)
async def test_entity_picture_template(hass, start_ha):
"""Test icon template."""
state = hass.states.get("cover.test_template_cover")
assert state.attributes.get("entity_picture") == ""
state = hass.states.async_set("cover.test_state", STATE_OPEN)
await hass.async_block_till_done()
state = hass.states.get("cover.test_template_cover")
assert state.attributes["entity_picture"] == "/local/cover.png"
@pytest.mark.parametrize("count,domain", [(1, DOMAIN)])
@pytest.mark.parametrize(
"config",
[
{
DOMAIN: {
"platform": "template",
"covers": {
"test_template_cover": {
**OPEN_CLOSE_COVER_CONFIG,
"value_template": "open",
"availability_template": "{{ is_state('availability_state.state','on') }}",
}
},
}
},
],
)
async def test_availability_template(hass, start_ha):
"""Test availability template."""
hass.states.async_set("availability_state.state", STATE_OFF)
await hass.async_block_till_done()
assert hass.states.get("cover.test_template_cover").state == STATE_UNAVAILABLE
hass.states.async_set("availability_state.state", STATE_ON)
await hass.async_block_till_done()
assert hass.states.get("cover.test_template_cover").state != STATE_UNAVAILABLE
@pytest.mark.parametrize("count,domain", [(1, DOMAIN)])
@pytest.mark.parametrize(
"config",
[
{
DOMAIN: {
"platform": "template",
"covers": {
"test_template_cover": {
**OPEN_CLOSE_COVER_CONFIG,
"value_template": "open",
}
},
}
},
],
)
async def test_availability_without_availability_template(hass, start_ha):
"""Test that component is available if there is no."""
state = hass.states.get("cover.test_template_cover")
assert state.state != STATE_UNAVAILABLE
@pytest.mark.parametrize("count,domain", [(1, DOMAIN)])
@pytest.mark.parametrize(
"config",
[
{
DOMAIN: {
"platform": "template",
"covers": {
"test_template_cover": {
**OPEN_CLOSE_COVER_CONFIG,
"availability_template": "{{ x - 12 }}",
"value_template": "open",
}
},
}
},
],
)
async def test_invalid_availability_template_keeps_component_available(
hass, start_ha, caplog_setup_text
):
"""Test that an invalid availability keeps the device available."""
assert hass.states.get("cover.test_template_cover") != STATE_UNAVAILABLE
assert ("UndefinedError: 'x' is undefined") in caplog_setup_text
@pytest.mark.parametrize("count,domain", [(1, DOMAIN)])
@pytest.mark.parametrize(
"config",
[
{
DOMAIN: {
"platform": "template",
"covers": {
"test_template_cover": {
**OPEN_CLOSE_COVER_CONFIG,
"value_template": "{{ states.cover.test_state.state }}",
"device_class": "door",
}
},
}
},
],
)
async def test_device_class(hass, start_ha):
"""Test device class."""
state = hass.states.get("cover.test_template_cover")
assert state.attributes.get("device_class") == "door"
@pytest.mark.parametrize("count,domain", [(0, DOMAIN)])
@pytest.mark.parametrize(
"config",
[
{
DOMAIN: {
"platform": "template",
"covers": {
"test_template_cover": {
**OPEN_CLOSE_COVER_CONFIG,
"value_template": "{{ states.cover.test_state.state }}",
"device_class": "barnacle_bill",
}
},
}
},
],
)
async def test_invalid_device_class(hass, start_ha):
"""Test device class."""
state = hass.states.get("cover.test_template_cover")
assert not state
@pytest.mark.parametrize("count,domain", [(1, DOMAIN)])
@pytest.mark.parametrize(
"config",
[
{
DOMAIN: {
"platform": "template",
"covers": {
"test_template_cover_01": {
**OPEN_CLOSE_COVER_CONFIG,
"unique_id": "not-so-unique-anymore",
"value_template": "{{ true }}",
},
"test_template_cover_02": {
**OPEN_CLOSE_COVER_CONFIG,
"unique_id": "not-so-unique-anymore",
"value_template": "{{ false }}",
},
},
}
},
],
)
async def test_unique_id(hass, start_ha):
"""Test unique_id option only creates one cover per id."""
assert len(hass.states.async_all()) == 1
@pytest.mark.parametrize("count,domain", [(1, DOMAIN)])
@pytest.mark.parametrize(
"config",
[
{
DOMAIN: {
"platform": "template",
"covers": {
"garage_door": {
**OPEN_CLOSE_COVER_CONFIG,
"friendly_name": "Garage Door",
"value_template": "{{ is_state('binary_sensor.garage_door_sensor', 'off') }}",
},
},
}
},
],
)
async def test_state_gets_lowercased(hass, start_ha):
"""Test True/False is lowercased."""
hass.states.async_set("binary_sensor.garage_door_sensor", "off")
await hass.async_block_till_done()
assert len(hass.states.async_all()) == 2
assert hass.states.get("cover.garage_door").state == STATE_OPEN
hass.states.async_set("binary_sensor.garage_door_sensor", "on")
await hass.async_block_till_done()
assert hass.states.get("cover.garage_door").state == STATE_CLOSED
@pytest.mark.parametrize("count,domain", [(1, DOMAIN)])
@pytest.mark.parametrize(
"config",
[
{
DOMAIN: {
"platform": "template",
"covers": {
"office": {
"icon_template": """{% if is_state('cover.office', 'open') %}
mdi:window-shutter-open
{% else %}
mdi:window-shutter
{% endif %}""",
"open_cover": {
"service": "switch.turn_on",
"entity_id": "switch.office_blinds_up",
},
"close_cover": {
"service": "switch.turn_on",
"entity_id": "switch.office_blinds_down",
},
"stop_cover": {
"service": "switch.turn_on",
"entity_id": "switch.office_blinds_up",
},
},
},
}
},
],
)
async def test_self_referencing_icon_with_no_template_is_not_a_loop(
hass, start_ha, caplog
):
"""Test a self referencing icon with no value template is not a loop."""
assert len(hass.states.async_all()) == 1
assert "Template loop detected" not in caplog.text