jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-23 13:17:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add additional device conditions to cover (#27830)

Browse Source 
* Add additional device conditions to cover

* Add default value

* Add test

* Use numeric_state condition instead of template condition
This commit is contained in:
Erik Montnemery 2019-10-26 04:50:46 +08:00 committed by Paulus Schoutsen
parent d44bfa8e88
commit 7fee44b8c5
4 changed files with 689 additions and 60 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

208
homeassistant/components/cover/device_condition.py
View File

@ -4,6 +4,9 @@ import voluptuous as vol
from homeassistant.const import (
ATTR_ENTITY_ID,
ATTR_SUPPORTED_FEATURES,
CONF_ABOVE,
CONF_BELOW,
CONF_CONDITION,
CONF_DOMAIN,
CONF_TYPE,
@ -15,20 +18,50 @@ from homeassistant.const import (
STATE_CLOSING,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import condition, config_validation as cv, entity_registry
from homeassistant.helpers import (
condition,
config_validation as cv,
entity_registry,
template,
)
from homeassistant.helpers.typing import ConfigType, TemplateVarsType
from homeassistant.helpers.config_validation import DEVICE_CONDITION_BASE_SCHEMA
from . import DOMAIN
from . import (
DOMAIN,
SUPPORT_CLOSE,
SUPPORT_OPEN,
SUPPORT_SET_POSITION,
SUPPORT_SET_TILT_POSITION,
)
CONDITION_TYPES = {"is_open", "is_closed", "is_opening", "is_closing"}
POSITION_CONDITION_TYPES = {"is_position", "is_tilt_position"}
STATE_CONDITION_TYPES = {"is_open", "is_closed", "is_opening", "is_closing"}
CONDITION_SCHEMA = DEVICE_CONDITION_BASE_SCHEMA.extend(
POSITION_CONDITION_SCHEMA = vol.All(
DEVICE_CONDITION_BASE_SCHEMA.extend(
{
vol.Required(CONF_ENTITY_ID): cv.entity_id,
vol.Required(CONF_TYPE): vol.In(POSITION_CONDITION_TYPES),
vol.Optional(CONF_ABOVE): vol.All(
vol.Coerce(int), vol.Range(min=0, max=100)
),
vol.Optional(CONF_BELOW): vol.All(
vol.Coerce(int), vol.Range(min=0, max=100)
),
}
),
cv.has_at_least_one_key(CONF_BELOW, CONF_ABOVE),
)
STATE_CONDITION_SCHEMA = DEVICE_CONDITION_BASE_SCHEMA.extend(
{
vol.Required(CONF_ENTITY_ID): cv.entity_id,
vol.Required(CONF_TYPE): vol.In(CONDITION_TYPES),
vol.Required(CONF_TYPE): vol.In(STATE_CONDITION_TYPES),
}
)
CONDITION_SCHEMA = vol.Any(POSITION_CONDITION_SCHEMA, STATE_CONDITION_SCHEMA)
async def async_get_conditions(hass: HomeAssistant, device_id: str) -> List[dict]:
"""List device conditions for Cover devices."""
@ -40,64 +73,133 @@ async def async_get_conditions(hass: HomeAssistant, device_id: str) -> List[dict
if entry.domain != DOMAIN:
continue
state = hass.states.get(entry.entity_id)
if not state or ATTR_SUPPORTED_FEATURES not in state.attributes:
continue
supported_features = state.attributes[ATTR_SUPPORTED_FEATURES]
supports_open_close = supported_features & (SUPPORT_OPEN | SUPPORT_CLOSE)
# Add conditions for each entity that belongs to this integration
conditions.append(
{
CONF_CONDITION: "device",
CONF_DEVICE_ID: device_id,
CONF_DOMAIN: DOMAIN,
CONF_ENTITY_ID: entry.entity_id,
CONF_TYPE: "is_open",
}
)
conditions.append(
{
CONF_CONDITION: "device",
CONF_DEVICE_ID: device_id,
CONF_DOMAIN: DOMAIN,
CONF_ENTITY_ID: entry.entity_id,
CONF_TYPE: "is_closed",
}
)
conditions.append(
{
CONF_CONDITION: "device",
CONF_DEVICE_ID: device_id,
CONF_DOMAIN: DOMAIN,
CONF_ENTITY_ID: entry.entity_id,
CONF_TYPE: "is_opening",
}
)
conditions.append(
{
CONF_CONDITION: "device",
CONF_DEVICE_ID: device_id,
CONF_DOMAIN: DOMAIN,
CONF_ENTITY_ID: entry.entity_id,
CONF_TYPE: "is_closing",
}
)
if supports_open_close:
conditions.append(
{
CONF_CONDITION: "device",
CONF_DEVICE_ID: device_id,
CONF_DOMAIN: DOMAIN,
CONF_ENTITY_ID: entry.entity_id,
CONF_TYPE: "is_open",
}
)
conditions.append(
{
CONF_CONDITION: "device",
CONF_DEVICE_ID: device_id,
CONF_DOMAIN: DOMAIN,
CONF_ENTITY_ID: entry.entity_id,
CONF_TYPE: "is_closed",
}
)
conditions.append(
{
CONF_CONDITION: "device",
CONF_DEVICE_ID: device_id,
CONF_DOMAIN: DOMAIN,
CONF_ENTITY_ID: entry.entity_id,
CONF_TYPE: "is_opening",
}
)
conditions.append(
{
CONF_CONDITION: "device",
CONF_DEVICE_ID: device_id,
CONF_DOMAIN: DOMAIN,
CONF_ENTITY_ID: entry.entity_id,
CONF_TYPE: "is_closing",
}
)
if supported_features & SUPPORT_SET_POSITION:
conditions.append(
{
CONF_CONDITION: "device",
CONF_DEVICE_ID: device_id,
CONF_DOMAIN: DOMAIN,
CONF_ENTITY_ID: entry.entity_id,
CONF_TYPE: "is_position",
}
)
if supported_features & SUPPORT_SET_TILT_POSITION:
conditions.append(
{
CONF_CONDITION: "device",
CONF_DEVICE_ID: device_id,
CONF_DOMAIN: DOMAIN,
CONF_ENTITY_ID: entry.entity_id,
CONF_TYPE: "is_tilt_position",
}
)
return conditions
async def async_get_condition_capabilities(hass: HomeAssistant, config: dict) -> dict:
"""List condition capabilities."""
if config[CONF_TYPE] not in ["is_position", "is_tilt_position"]:
return {}
return {
"extra_fields": vol.Schema(
{
vol.Optional(CONF_ABOVE, default=0): vol.All(
vol.Coerce(int), vol.Range(min=0, max=100)
),
vol.Optional(CONF_BELOW, default=100): vol.All(
vol.Coerce(int), vol.Range(min=0, max=100)
),
}
)
}
def async_condition_from_config(
config: ConfigType, config_validation: bool
) -> condition.ConditionCheckerType:
"""Create a function to test a device condition."""
if config_validation:
config = CONDITION_SCHEMA(config)
if config[CONF_TYPE] == "is_open":
state = STATE_OPEN
elif config[CONF_TYPE] == "is_closed":
state = STATE_CLOSED
elif config[CONF_TYPE] == "is_opening":
state = STATE_OPENING
elif config[CONF_TYPE] == "is_closing":
state = STATE_CLOSING
def test_is_state(hass: HomeAssistant, variables: TemplateVarsType) -> bool:
"""Test if an entity is a certain state."""
return condition.state(hass, config[ATTR_ENTITY_ID], state)
if config[CONF_TYPE] in STATE_CONDITION_TYPES:
if config[CONF_TYPE] == "is_open":
state = STATE_OPEN
elif config[CONF_TYPE] == "is_closed":
state = STATE_CLOSED
elif config[CONF_TYPE] == "is_opening":
state = STATE_OPENING
elif config[CONF_TYPE] == "is_closing":
state = STATE_CLOSING
return test_is_state
def test_is_state(hass: HomeAssistant, variables: TemplateVarsType) -> bool:
"""Test if an entity is a certain state."""
return condition.state(hass, config[ATTR_ENTITY_ID], state)
return test_is_state
if config[CONF_TYPE] == "is_position":
position = "current_position"
if config[CONF_TYPE] == "is_tilt_position":
position = "current_tilt_position"
min_pos = config.get(CONF_ABOVE, None)
max_pos = config.get(CONF_BELOW, None)
value_template = template.Template( # type: ignore
f"{{{{ state.attributes.{position} }}}}"
)
def template_if(hass: HomeAssistant, variables: TemplateVarsType = None) -> bool:
"""Validate template based if-condition."""
value_template.hass = hass
return condition.async_numeric_state(
hass, config[ATTR_ENTITY_ID], max_pos, min_pos, value_template
)
return template_if

4
homeassistant/components/cover/strings.json
View File

@ -4,7 +4,9 @@
"is_open": "{entity_name} is open",
"is_closed": "{entity_name} is closed",
"is_opening": "{entity_name} is opening",
"is_closing": "{entity_name} is closing"
"is_closing": "{entity_name} is closing",
"is_position": "{entity_name} position is",
"is_tilt_position": "{entity_name} tilt position is"
}
}
}

476
tests/components/cover/test_device_condition.py
View File

@ -2,7 +2,13 @@
import pytest
from homeassistant.components.cover import DOMAIN
from homeassistant.const import STATE_OPEN, STATE_CLOSED, STATE_OPENING, STATE_CLOSING
from homeassistant.const import (
CONF_PLATFORM,
STATE_OPEN,
STATE_CLOSED,
STATE_OPENING,
STATE_CLOSING,
)
from homeassistant.setup import async_setup_component
import homeassistant.components.automation as automation
from homeassistant.helpers import device_registry
@ -14,6 +20,7 @@ from tests.common import (
mock_device_registry,
mock_registry,
async_get_device_automations,
async_get_device_automation_capabilities,
)
@ -37,47 +44,298 @@ def calls(hass):
async def test_get_conditions(hass, device_reg, entity_reg):
"""Test we get the expected conditions from a cover."""
platform = getattr(hass.components, f"test.{DOMAIN}")
platform.init()
ent = platform.ENTITIES[0]
config_entry = MockConfigEntry(domain="test", data={})
config_entry.add_to_hass(hass)
device_entry = device_reg.async_get_or_create(
config_entry_id=config_entry.entry_id,
connections={(device_registry.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
)
entity_reg.async_get_or_create(DOMAIN, "test", "5678", device_id=device_entry.id)
entity_reg.async_get_or_create(
DOMAIN, "test", ent.unique_id, device_id=device_entry.id
)
assert await async_setup_component(hass, DOMAIN, {DOMAIN: {CONF_PLATFORM: "test"}})
expected_conditions = [
{
"condition": "device",
"domain": DOMAIN,
"type": "is_open",
"device_id": device_entry.id,
"entity_id": f"{DOMAIN}.test_5678",
"entity_id": f"{DOMAIN}.test_{ent.unique_id}",
},
{
"condition": "device",
"domain": DOMAIN,
"type": "is_closed",
"device_id": device_entry.id,
"entity_id": f"{DOMAIN}.test_5678",
"entity_id": f"{DOMAIN}.test_{ent.unique_id}",
},
{
"condition": "device",
"domain": DOMAIN,
"type": "is_opening",
"device_id": device_entry.id,
"entity_id": f"{DOMAIN}.test_5678",
"entity_id": f"{DOMAIN}.test_{ent.unique_id}",
},
{
"condition": "device",
"domain": DOMAIN,
"type": "is_closing",
"device_id": device_entry.id,
"entity_id": f"{DOMAIN}.test_5678",
"entity_id": f"{DOMAIN}.test_{ent.unique_id}",
},
]
conditions = await async_get_device_automations(hass, "condition", device_entry.id)
assert_lists_same(conditions, expected_conditions)
async def test_get_conditions_set_pos(hass, device_reg, entity_reg):
"""Test we get the expected conditions from a cover."""
platform = getattr(hass.components, f"test.{DOMAIN}")
platform.init()
ent = platform.ENTITIES[1]
config_entry = MockConfigEntry(domain="test", data={})
config_entry.add_to_hass(hass)
device_entry = device_reg.async_get_or_create(
config_entry_id=config_entry.entry_id,
connections={(device_registry.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
)
entity_reg.async_get_or_create(
DOMAIN, "test", ent.unique_id, device_id=device_entry.id
)
assert await async_setup_component(hass, DOMAIN, {DOMAIN: {CONF_PLATFORM: "test"}})
expected_conditions = [
{
"condition": "device",
"domain": DOMAIN,
"type": "is_open",
"device_id": device_entry.id,
"entity_id": f"{DOMAIN}.test_{ent.unique_id}",
},
{
"condition": "device",
"domain": DOMAIN,
"type": "is_closed",
"device_id": device_entry.id,
"entity_id": f"{DOMAIN}.test_{ent.unique_id}",
},
{
"condition": "device",
"domain": DOMAIN,
"type": "is_opening",
"device_id": device_entry.id,
"entity_id": f"{DOMAIN}.test_{ent.unique_id}",
},
{
"condition": "device",
"domain": DOMAIN,
"type": "is_closing",
"device_id": device_entry.id,
"entity_id": f"{DOMAIN}.test_{ent.unique_id}",
},
{
"condition": "device",
"domain": DOMAIN,
"type": "is_position",
"device_id": device_entry.id,
"entity_id": f"{DOMAIN}.test_{ent.unique_id}",
},
]
conditions = await async_get_device_automations(hass, "condition", device_entry.id)
assert_lists_same(conditions, expected_conditions)
async def test_get_conditions_set_tilt_pos(hass, device_reg, entity_reg):
"""Test we get the expected conditions from a cover."""
platform = getattr(hass.components, f"test.{DOMAIN}")
platform.init()
ent = platform.ENTITIES[2]
config_entry = MockConfigEntry(domain="test", data={})
config_entry.add_to_hass(hass)
device_entry = device_reg.async_get_or_create(
config_entry_id=config_entry.entry_id,
connections={(device_registry.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
)
entity_reg.async_get_or_create(
DOMAIN, "test", ent.unique_id, device_id=device_entry.id
)
assert await async_setup_component(hass, DOMAIN, {DOMAIN: {CONF_PLATFORM: "test"}})
expected_conditions = [
{
"condition": "device",
"domain": DOMAIN,
"type": "is_open",
"device_id": device_entry.id,
"entity_id": f"{DOMAIN}.test_{ent.unique_id}",
},
{
"condition": "device",
"domain": DOMAIN,
"type": "is_closed",
"device_id": device_entry.id,
"entity_id": f"{DOMAIN}.test_{ent.unique_id}",
},
{
"condition": "device",
"domain": DOMAIN,
"type": "is_opening",
"device_id": device_entry.id,
"entity_id": f"{DOMAIN}.test_{ent.unique_id}",
},
{
"condition": "device",
"domain": DOMAIN,
"type": "is_closing",
"device_id": device_entry.id,
"entity_id": f"{DOMAIN}.test_{ent.unique_id}",
},
{
"condition": "device",
"domain": DOMAIN,
"type": "is_tilt_position",
"device_id": device_entry.id,
"entity_id": f"{DOMAIN}.test_{ent.unique_id}",
},
]
conditions = await async_get_device_automations(hass, "condition", device_entry.id)
assert_lists_same(conditions, expected_conditions)
async def test_get_condition_capabilities(hass, device_reg, entity_reg):
"""Test we get the expected capabilities from a cover condition."""
platform = getattr(hass.components, f"test.{DOMAIN}")
platform.init()
ent = platform.ENTITIES[0]
config_entry = MockConfigEntry(domain="test", data={})
config_entry.add_to_hass(hass)
device_entry = device_reg.async_get_or_create(
config_entry_id=config_entry.entry_id,
connections={(device_registry.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
)
entity_reg.async_get_or_create(
DOMAIN, "test", ent.unique_id, device_id=device_entry.id
)
assert await async_setup_component(hass, DOMAIN, {DOMAIN: {CONF_PLATFORM: "test"}})
conditions = await async_get_device_automations(hass, "condition", device_entry.id)
assert len(conditions) == 4
for condition in conditions:
capabilities = await async_get_device_automation_capabilities(
hass, "condition", condition
)
assert capabilities == {"extra_fields": []}
async def test_get_condition_capabilities_set_pos(hass, device_reg, entity_reg):
"""Test we get the expected capabilities from a cover condition."""
platform = getattr(hass.components, f"test.{DOMAIN}")
platform.init()
ent = platform.ENTITIES[1]
config_entry = MockConfigEntry(domain="test", data={})
config_entry.add_to_hass(hass)
device_entry = device_reg.async_get_or_create(
config_entry_id=config_entry.entry_id,
connections={(device_registry.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
)
entity_reg.async_get_or_create(
DOMAIN, "test", ent.unique_id, device_id=device_entry.id
)
assert await async_setup_component(hass, DOMAIN, {DOMAIN: {CONF_PLATFORM: "test"}})
expected_capabilities = {
"extra_fields": [
{
"name": "above",
"optional": True,
"type": "integer",
"default": 0,
"valueMax": 100,
"valueMin": 0,
},
{
"name": "below",
"optional": True,
"type": "integer",
"default": 100,
"valueMax": 100,
"valueMin": 0,
},
]
}
conditions = await async_get_device_automations(hass, "condition", device_entry.id)
assert len(conditions) == 5
for condition in conditions:
capabilities = await async_get_device_automation_capabilities(
hass, "condition", condition
)
if condition["type"] == "is_position":
assert capabilities == expected_capabilities
else:
assert capabilities == {"extra_fields": []}
async def test_get_condition_capabilities_set_tilt_pos(hass, device_reg, entity_reg):
"""Test we get the expected capabilities from a cover condition."""
platform = getattr(hass.components, f"test.{DOMAIN}")
platform.init()
ent = platform.ENTITIES[2]
config_entry = MockConfigEntry(domain="test", data={})
config_entry.add_to_hass(hass)
device_entry = device_reg.async_get_or_create(
config_entry_id=config_entry.entry_id,
connections={(device_registry.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
)
entity_reg.async_get_or_create(
DOMAIN, "test", ent.unique_id, device_id=device_entry.id
)
assert await async_setup_component(hass, DOMAIN, {DOMAIN: {CONF_PLATFORM: "test"}})
expected_capabilities = {
"extra_fields": [
{
"name": "above",
"optional": True,
"type": "integer",
"default": 0,
"valueMax": 100,
"valueMin": 0,
},
{
"name": "below",
"optional": True,
"type": "integer",
"default": 100,
"valueMax": 100,
"valueMin": 0,
},
]
}
conditions = await async_get_device_automations(hass, "condition", device_entry.id)
assert len(conditions) == 5
for condition in conditions:
capabilities = await async_get_device_automation_capabilities(
hass, "condition", condition
)
if condition["type"] == "is_tilt_position":
assert capabilities == expected_capabilities
else:
assert capabilities == {"extra_fields": []}
async def test_if_state(hass, calls):
"""Test for turn_on and turn_off conditions."""
hass.states.async_set("cover.entity", STATE_OPEN)
@ -188,3 +446,209 @@ async def test_if_state(hass, calls):
await hass.async_block_till_done()
assert len(calls) == 4
assert calls[3].data["some"] == "is_closing - event - test_event4"
async def test_if_position(hass, calls):
"""Test for position conditions."""
platform = getattr(hass.components, f"test.{DOMAIN}")
platform.init()
ent = platform.ENTITIES[1]
assert await async_setup_component(hass, DOMAIN, {DOMAIN: {CONF_PLATFORM: "test"}})
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: [
{
"trigger": {"platform": "event", "event_type": "test_event1"},
"condition": [
{
"condition": "device",
"domain": DOMAIN,
"device_id": "",
"entity_id": ent.entity_id,
"type": "is_position",
"above": 45,
}
],
"action": {
"service": "test.automation",
"data_template": {
"some": "is_pos_gt_45 - {{ trigger.platform }} - {{ trigger.event.event_type }}"
},
},
},
{
"trigger": {"platform": "event", "event_type": "test_event2"},
"condition": [
{
"condition": "device",
"domain": DOMAIN,
"device_id": "",
"entity_id": ent.entity_id,
"type": "is_position",
"below": 90,
}
],
"action": {
"service": "test.automation",
"data_template": {
"some": "is_pos_lt_90 - {{ trigger.platform }} - {{ trigger.event.event_type }}"
},
},
},
{
"trigger": {"platform": "event", "event_type": "test_event3"},
"condition": [
{
"condition": "device",
"domain": DOMAIN,
"device_id": "",
"entity_id": ent.entity_id,
"type": "is_position",
"above": 45,
"below": 90,
}
],
"action": {
"service": "test.automation",
"data_template": {
"some": "is_pos_gt_45_lt_90 - {{ trigger.platform }} - {{ trigger.event.event_type }}"
},
},
},
]
},
)
hass.bus.async_fire("test_event1")
hass.bus.async_fire("test_event2")
hass.bus.async_fire("test_event3")
await hass.async_block_till_done()
assert len(calls) == 3
assert calls[0].data["some"] == "is_pos_gt_45 - event - test_event1"
assert calls[1].data["some"] == "is_pos_lt_90 - event - test_event2"
assert calls[2].data["some"] == "is_pos_gt_45_lt_90 - event - test_event3"
hass.states.async_set(
ent.entity_id, STATE_CLOSED, attributes={"current_position": 45}
)
hass.bus.async_fire("test_event1")
hass.bus.async_fire("test_event2")
hass.bus.async_fire("test_event3")
await hass.async_block_till_done()
assert len(calls) == 4
assert calls[3].data["some"] == "is_pos_lt_90 - event - test_event2"
hass.states.async_set(
ent.entity_id, STATE_CLOSED, attributes={"current_position": 90}
)
hass.bus.async_fire("test_event1")
hass.bus.async_fire("test_event2")
hass.bus.async_fire("test_event3")
await hass.async_block_till_done()
assert len(calls) == 5
assert calls[4].data["some"] == "is_pos_gt_45 - event - test_event1"
async def test_if_tilt_position(hass, calls):
"""Test for tilt position conditions."""
platform = getattr(hass.components, f"test.{DOMAIN}")
platform.init()
ent = platform.ENTITIES[2]
assert await async_setup_component(hass, DOMAIN, {DOMAIN: {CONF_PLATFORM: "test"}})
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: [
{
"trigger": {"platform": "event", "event_type": "test_event1"},
"condition": [
{
"condition": "device",
"domain": DOMAIN,
"device_id": "",
"entity_id": ent.entity_id,
"type": "is_tilt_position",
"above": 45,
}
],
"action": {
"service": "test.automation",
"data_template": {
"some": "is_pos_gt_45 - {{ trigger.platform }} - {{ trigger.event.event_type }}"
},
},
},
{
"trigger": {"platform": "event", "event_type": "test_event2"},
"condition": [
{
"condition": "device",
"domain": DOMAIN,
"device_id": "",
"entity_id": ent.entity_id,
"type": "is_tilt_position",
"below": 90,
}
],
"action": {
"service": "test.automation",
"data_template": {
"some": "is_pos_lt_90 - {{ trigger.platform }} - {{ trigger.event.event_type }}"
},
},
},
{
"trigger": {"platform": "event", "event_type": "test_event3"},
"condition": [
{
"condition": "device",
"domain": DOMAIN,
"device_id": "",
"entity_id": ent.entity_id,
"type": "is_tilt_position",
"above": 45,
"below": 90,
}
],
"action": {
"service": "test.automation",
"data_template": {
"some": "is_pos_gt_45_lt_90 - {{ trigger.platform }} - {{ trigger.event.event_type }}"
},
},
},
]
},
)
hass.bus.async_fire("test_event1")
hass.bus.async_fire("test_event2")
hass.bus.async_fire("test_event3")
await hass.async_block_till_done()
assert len(calls) == 3
assert calls[0].data["some"] == "is_pos_gt_45 - event - test_event1"
assert calls[1].data["some"] == "is_pos_lt_90 - event - test_event2"
assert calls[2].data["some"] == "is_pos_gt_45_lt_90 - event - test_event3"
hass.states.async_set(
ent.entity_id, STATE_CLOSED, attributes={"current_tilt_position": 45}
)
hass.bus.async_fire("test_event1")
hass.bus.async_fire("test_event2")
hass.bus.async_fire("test_event3")
await hass.async_block_till_done()
assert len(calls) == 4
assert calls[3].data["some"] == "is_pos_lt_90 - event - test_event2"
hass.states.async_set(
ent.entity_id, STATE_CLOSED, attributes={"current_tilt_position": 90}
)
hass.bus.async_fire("test_event1")
hass.bus.async_fire("test_event2")
hass.bus.async_fire("test_event3")
await hass.async_block_till_done()
assert len(calls) == 5
assert calls[4].data["some"] == "is_pos_gt_45 - event - test_event1"

61
tests/testing_config/custom_components/test/cover.py Normal file
View File

@ -0,0 +1,61 @@
"""
Provide a mock cover platform.
Call init before using it in your tests to ensure clean test data.
"""
from homeassistant.components.cover import CoverDevice
from tests.common import MockEntity
ENTITIES = {}
def init(empty=False):
"""Initialize the platform with entities."""
global ENTITIES
ENTITIES = (
[]
if empty
else [
MockCover(name=f"Simple cover", is_on=True, unique_id=f"unique_cover"),
MockCover(
name=f"Set position cover",
is_on=True,
unique_id=f"unique_set_pos_cover",
current_cover_position=50,
),
MockCover(
name=f"Set tilt position cover",
is_on=True,
unique_id=f"unique_set_pos_tilt_cover",
current_cover_tilt_position=50,
),
]
)
async def async_setup_platform(
hass, config, async_add_entities_callback, discovery_info=None
):
"""Return mock entities."""
async_add_entities_callback(ENTITIES)
class MockCover(MockEntity, CoverDevice):
"""Mock Cover class."""
@property
def is_closed(self):
"""Return if the cover is closed or not."""
return False
@property
def current_cover_position(self):
"""Return current position of cover."""
return self._handle("current_cover_position")
@property
def current_cover_tilt_position(self):
"""Return current position of cover tilt."""
return self._handle("current_cover_tilt_position")