jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-23 08:47:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add device automation condition (#26313)

Browse Source 
* Add support for device conditions

* Lint

* Update test case

* Make and+or conditions async, adjust tests

* Cleanup tests

* Remove non callback versions of conditions, correct typing

* Correct typing

* Update light/strings.json

* Address review comments

* Make device automation lists simple lists, not dicts

* Add device_automation/const.py

* Use IS_ON/IS_OFF everywhere for conditions
This commit is contained in:
Erik Montnemery 2019-09-05 16:49:32 +02:00 committed by GitHub
parent c50faaef3c
commit f7dc537275
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 480 additions and 263 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
homeassistant/components/automation/__init__.py
View File

@ -386,7 +386,7 @@ async def _async_process_config(hass, config, component):
action = _async_get_action(hass, config_block.get(CONF_ACTION, {}), name)
if CONF_CONDITION in config_block:
cond_func = _async_process_if(hass, config, config_block)
cond_func = await _async_process_if(hass, config, config_block)
if cond_func is None:
continue
@ -437,14 +437,14 @@ def _async_get_action(hass, config, name):
return action
def _async_process_if(hass, config, p_config):
async def _async_process_if(hass, config, p_config):
"""Process if checks."""
if_configs = p_config.get(CONF_CONDITION)
checks = []
for if_config in if_configs:
try:
checks.append(condition.async_from_config(if_config, False))
checks.append(await condition.async_from_config(hass, if_config, False))
except HomeAssistantError as ex:
_LOGGER.warning("Invalid condition: %s", ex)
return None

71
homeassistant/components/device_automation/__init__.py
View File

@ -1,12 +1,16 @@
"""Helpers for device automations."""
import asyncio
import logging
from typing import Callable, cast
import voluptuous as vol
from homeassistant.components import websocket_api
from homeassistant.core import split_entity_id
from homeassistant.const import CONF_DOMAIN
from homeassistant.core import split_entity_id, HomeAssistant
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity_registry import async_entries_for_device
from homeassistant.helpers.typing import ConfigType
from homeassistant.loader import async_get_integration, IntegrationNotFound
DOMAIN = "device_automation"
@ -16,14 +20,31 @@ _LOGGER = logging.getLogger(__name__)
async def async_setup(hass, config):
"""Set up device automation."""
hass.components.websocket_api.async_register_command(
websocket_device_automation_list_conditions
)
hass.components.websocket_api.async_register_command(
websocket_device_automation_list_triggers
)
return True
async def _async_get_device_automation_triggers(hass, domain, device_id):
"""List device triggers."""
async def async_device_condition_from_config(
hass: HomeAssistant, config: ConfigType, config_validation: bool = True
) -> Callable[..., bool]:
"""Wrap action method with state based condition."""
if config_validation:
config = cv.DEVICE_CONDITION_SCHEMA(config)
integration = await async_get_integration(hass, config[CONF_DOMAIN])
platform = integration.get_platform("device_automation")
return cast(
Callable[..., bool],
platform.async_condition_from_config(config, config_validation), # type: ignore
)
async def _async_get_device_automations_from_domain(hass, domain, fname, device_id):
"""List device automations."""
integration = None
try:
integration = await async_get_integration(hass, domain)
@ -37,19 +58,19 @@ async def _async_get_device_automation_triggers(hass, domain, device_id):
# The domain does not have device automations
return None
if hasattr(platform, "async_get_triggers"):
return await platform.async_get_triggers(hass, device_id)
if hasattr(platform, fname):
return await getattr(platform, fname)(hass, device_id)
async def async_get_device_automation_triggers(hass, device_id):
"""List device triggers."""
async def _async_get_device_automations(hass, fname, device_id):
"""List device automations."""
device_registry, entity_registry = await asyncio.gather(
hass.helpers.device_registry.async_get_registry(),
hass.helpers.entity_registry.async_get_registry(),
)
domains = set()
triggers = []
automations = []
device = device_registry.async_get(device_id)
for entry_id in device.config_entries:
config_entry = hass.config_entries.async_get_entry(entry_id)
@ -59,17 +80,33 @@ async def async_get_device_automation_triggers(hass, device_id):
for entity in entities:
domains.add(split_entity_id(entity.entity_id)[0])
device_triggers = await asyncio.gather(
device_automations = await asyncio.gather(
*(
_async_get_device_automation_triggers(hass, domain, device_id)
_async_get_device_automations_from_domain(hass, domain, fname, device_id)
for domain in domains
)
)
for device_trigger in device_triggers:
if device_trigger is not None:
triggers.extend(device_trigger)
for device_automation in device_automations:
if device_automation is not None:
automations.extend(device_automation)
return triggers
return automations
@websocket_api.async_response
@websocket_api.websocket_command(
{
vol.Required("type"): "device_automation/condition/list",
vol.Required("device_id"): str,
}
)
async def websocket_device_automation_list_conditions(hass, connection, msg):
"""Handle request for device conditions."""
device_id = msg["device_id"]
conditions = await _async_get_device_automations(
hass, "async_get_conditions", device_id
)
connection.send_result(msg["id"], conditions)
@websocket_api.async_response
@ -82,5 +119,7 @@ async def async_get_device_automation_triggers(hass, device_id):
async def websocket_device_automation_list_triggers(hass, connection, msg):
"""Handle request for device triggers."""
device_id = msg["device_id"]
triggers = await async_get_device_automation_triggers(hass, device_id)
connection.send_result(msg["id"], {"triggers": triggers})
triggers = await _async_get_device_automations(
hass, "async_get_triggers", device_id
)
connection.send_result(msg["id"], triggers)

5
homeassistant/components/device_automation/const.py Normal file
View File

@ -0,0 +1,5 @@
"""Constants for device automations."""
CONF_IS_OFF = "is_off"
CONF_IS_ON = "is_on"
CONF_TURN_OFF = "turn_off"
CONF_TURN_ON = "turn_on"

89
homeassistant/components/light/device_automation.py
View File

@ -2,39 +2,70 @@
import voluptuous as vol
import homeassistant.components.automation.state as state
from homeassistant.components.device_automation.const import (
CONF_IS_OFF,
CONF_IS_ON,
CONF_TURN_OFF,
CONF_TURN_ON,
)
from homeassistant.core import split_entity_id
from homeassistant.const import (
CONF_CONDITION,
CONF_DEVICE_ID,
CONF_DOMAIN,
CONF_ENTITY_ID,
CONF_PLATFORM,
CONF_TYPE,
)
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers import condition, config_validation as cv
from homeassistant.helpers.entity_registry import async_entries_for_device
from . import DOMAIN
# mypy: allow-untyped-defs, no-check-untyped-defs
CONF_TURN_OFF = "turn_off"
CONF_TURN_ON = "turn_on"
ENTITY_CONDITIONS = [
{
# True when light is turned off
CONF_CONDITION: "device",
CONF_DOMAIN: DOMAIN,
CONF_TYPE: CONF_IS_OFF,
},
{
# True when light is turned on
CONF_CONDITION: "device",
CONF_DOMAIN: DOMAIN,
CONF_TYPE: CONF_IS_ON,
},
]
ENTITY_TRIGGERS = [
{
# Trigger when light is turned on
# Trigger when light is turned off
CONF_PLATFORM: "device",
CONF_DOMAIN: DOMAIN,
CONF_TYPE: CONF_TURN_OFF,
},
{
# Trigger when light is turned off
# Trigger when light is turned on
CONF_PLATFORM: "device",
CONF_DOMAIN: DOMAIN,
CONF_TYPE: CONF_TURN_ON,
},
]
CONDITION_SCHEMA = vol.All(
vol.Schema(
{
vol.Required(CONF_CONDITION): "device",
vol.Optional(CONF_DEVICE_ID): str,
vol.Required(CONF_DOMAIN): DOMAIN,
vol.Required(CONF_ENTITY_ID): cv.entity_id,
vol.Required(CONF_TYPE): vol.In([CONF_IS_OFF, CONF_IS_ON]),
}
)
)
TRIGGER_SCHEMA = vol.All(
vol.Schema(
{
@ -42,7 +73,7 @@ TRIGGER_SCHEMA = vol.All(
vol.Optional(CONF_DEVICE_ID): str,
vol.Required(CONF_DOMAIN): DOMAIN,
vol.Required(CONF_ENTITY_ID): cv.entity_id,
vol.Required(CONF_TYPE): str,
vol.Required(CONF_TYPE): vol.In([CONF_TURN_OFF, CONF_TURN_ON]),
}
)
)
@ -52,9 +83,27 @@ def _is_domain(entity, domain):
return split_entity_id(entity.entity_id)[0] == domain
def async_condition_from_config(config, config_validation):
"""Evaluate state based on configuration."""
config = CONDITION_SCHEMA(config)
condition_type = config[CONF_TYPE]
if condition_type == CONF_IS_ON:
stat = "on"
else:
stat = "off"
state_config = {
condition.CONF_CONDITION: "state",
condition.CONF_ENTITY_ID: config[CONF_ENTITY_ID],
condition.CONF_STATE: stat,
}
return condition.state_from_config(state_config, config_validation)
async def async_attach_trigger(hass, config, action, automation_info):
"""Listen for state changes based on configuration."""
trigger_type = config.get(CONF_TYPE)
config = TRIGGER_SCHEMA(config)
trigger_type = config[CONF_TYPE]
if trigger_type == CONF_TURN_ON:
from_state = "off"
to_state = "on"
@ -75,17 +124,27 @@ async def async_trigger(hass, config, action, automation_info):
return await async_attach_trigger(hass, config, action, automation_info)
async def async_get_triggers(hass, device_id):
"""List device triggers."""
triggers = []
async def _async_get_automations(hass, device_id, automation_templates):
"""List device automations."""
automations = []
entity_registry = await hass.helpers.entity_registry.async_get_registry()
entities = async_entries_for_device(entity_registry, device_id)
domain_entities = [x for x in entities if _is_domain(x, DOMAIN)]
for entity in domain_entities:
for trigger in ENTITY_TRIGGERS:
trigger = dict(trigger)
trigger.update(device_id=device_id, entity_id=entity.entity_id)
triggers.append(trigger)
for automation in automation_templates:
automation = dict(automation)
automation.update(device_id=device_id, entity_id=entity.entity_id)
automations.append(automation)
return triggers
return automations
async def async_get_conditions(hass, device_id):
"""List device conditions."""
return await _async_get_automations(hass, device_id, ENTITY_CONDITIONS)
async def async_get_triggers(hass, device_id):
"""List device triggers."""
return await _async_get_automations(hass, device_id, ENTITY_TRIGGERS)

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

@ -1,5 +1,9 @@
{
"device_automation": {
"condition_type": {
"is_on": "{name} is on",
"is_off": "{name} is off"
},
"trigger_type": {
"turn_on": "{name} turned on",
"turn_off": "{name} turned off"

87
homeassistant/helpers/condition.py
View File

@ -1,4 +1,5 @@
"""Offer reusable conditions."""
import asyncio
from datetime import datetime, timedelta
import functools as ft
import logging
@ -10,6 +11,9 @@ from homeassistant.helpers.typing import ConfigType, TemplateVarsType
from homeassistant.core import HomeAssistant, State
from homeassistant.components import zone as zone_cmp
from homeassistant.components.device_automation import ( # noqa: F401 pylint: disable=unused-import
async_device_condition_from_config as async_device_from_config,
)
from homeassistant.const import (
ATTR_GPS_ACCURACY,
ATTR_LATITUDE,
@ -41,40 +45,9 @@ ASYNC_FROM_CONFIG_FORMAT = "async_{}_from_config"
_LOGGER = logging.getLogger(__name__)
# PyLint does not like the use of _threaded_factory
# pylint: disable=invalid-name
def _threaded_factory(
async_factory: Callable[[ConfigType, bool], Callable[..., bool]]
) -> Callable[[ConfigType, bool], Callable[..., bool]]:
"""Create threaded versions of async factories."""
@ft.wraps(async_factory)
def factory(
config: ConfigType, config_validation: bool = True
) -> Callable[..., bool]:
"""Threaded factory."""
async_check = async_factory(config, config_validation)
def condition_if(
hass: HomeAssistant, variables: TemplateVarsType = None
) -> bool:
"""Validate condition."""
return cast(
bool,
run_callback_threadsafe(
hass.loop, async_check, hass, variables
).result(),
)
return condition_if
return factory
def async_from_config(
config: ConfigType, config_validation: bool = True
async def async_from_config(
hass: HomeAssistant, config: ConfigType, config_validation: bool = True
) -> Callable[..., bool]:
"""Turn a condition configuration into a method.
@ -95,29 +68,30 @@ def async_from_config(
)
)
# Check for partials to properly determine if coroutine function
check_factory = factory
while isinstance(check_factory, ft.partial):
check_factory = check_factory.func
if asyncio.iscoroutinefunction(check_factory):
return cast(Callable[..., bool], await factory(hass, config, config_validation))
return cast(Callable[..., bool], factory(config, config_validation))
from_config = _threaded_factory(async_from_config)
def async_and_from_config(
config: ConfigType, config_validation: bool = True
async def async_and_from_config(
hass: HomeAssistant, config: ConfigType, config_validation: bool = True
) -> Callable[..., bool]:
"""Create multi condition matcher using 'AND'."""
if config_validation:
config = cv.AND_CONDITION_SCHEMA(config)
checks = None
checks = [
await async_from_config(hass, entry, False) for entry in config["conditions"]
]
def if_and_condition(
hass: HomeAssistant, variables: TemplateVarsType = None
) -> bool:
"""Test and condition."""
nonlocal checks
if checks is None:
checks = [async_from_config(entry, False) for entry in config["conditions"]]
try:
for check in checks:
if not check(hass, variables):
@ -131,26 +105,20 @@ def async_and_from_config(
return if_and_condition
and_from_config = _threaded_factory(async_and_from_config)
def async_or_from_config(
config: ConfigType, config_validation: bool = True
async def async_or_from_config(
hass: HomeAssistant, config: ConfigType, config_validation: bool = True
) -> Callable[..., bool]:
"""Create multi condition matcher using 'OR'."""
if config_validation:
config = cv.OR_CONDITION_SCHEMA(config)
checks = None
checks = [
await async_from_config(hass, entry, False) for entry in config["conditions"]
]
def if_or_condition(
hass: HomeAssistant, variables: TemplateVarsType = None
) -> bool:
"""Test and condition."""
nonlocal checks
if checks is None:
checks = [async_from_config(entry, False) for entry in config["conditions"]]
try:
for check in checks:
if check(hass, variables):
@ -163,9 +131,6 @@ def async_or_from_config(
return if_or_condition
or_from_config = _threaded_factory(async_or_from_config)
def numeric_state(
hass: HomeAssistant,
entity: Union[None, str, State],
@ -263,9 +228,6 @@ def async_numeric_state_from_config(
return if_numeric_state
numeric_state_from_config = _threaded_factory(async_numeric_state_from_config)
def state(
hass: HomeAssistant,
entity: Union[None, str, State],
@ -423,9 +385,6 @@ def async_template_from_config(
return template_if
template_from_config = _threaded_factory(async_template_from_config)
def time(
before: Optional[dt_util.dt.time] = None,
after: Optional[dt_util.dt.time] = None,

13
homeassistant/helpers/config_validation.py
View File

@ -24,10 +24,13 @@ from homeassistant.const import (
CONF_ALIAS,
CONF_BELOW,
CONF_CONDITION,
CONF_DOMAIN,
CONF_ENTITY_ID,
CONF_ENTITY_NAMESPACE,
CONF_FOR,
CONF_PLATFORM,
CONF_SCAN_INTERVAL,
CONF_STATE,
CONF_UNIT_SYSTEM_IMPERIAL,
CONF_UNIT_SYSTEM_METRIC,
CONF_VALUE_TEMPLATE,
@ -746,8 +749,8 @@ STATE_CONDITION_SCHEMA = vol.All(
{
vol.Required(CONF_CONDITION): "state",
vol.Required(CONF_ENTITY_ID): entity_id,
vol.Required("state"): str,
vol.Optional("for"): vol.All(time_period, positive_timedelta),
vol.Required(CONF_STATE): str,
vol.Optional(CONF_FOR): vol.All(time_period, positive_timedelta),
# To support use_trigger_value in automation
# Deprecated 2016/04/25
vol.Optional("from"): str,
@ -823,6 +826,11 @@ OR_CONDITION_SCHEMA = vol.Schema(
}
)
DEVICE_CONDITION_SCHEMA = vol.Schema(
{vol.Required(CONF_CONDITION): "device", vol.Required(CONF_DOMAIN): str},
extra=vol.ALLOW_EXTRA,
)
CONDITION_SCHEMA: vol.Schema = vol.Any(
NUMERIC_STATE_CONDITION_SCHEMA,
STATE_CONDITION_SCHEMA,
@ -832,6 +840,7 @@ CONDITION_SCHEMA: vol.Schema = vol.Any(
ZONE_CONDITION_SCHEMA,
AND_CONDITION_SCHEMA,
OR_CONDITION_SCHEMA,
DEVICE_CONDITION_SCHEMA,
)
_SCRIPT_DELAY_SCHEMA = vol.Schema(

2
homeassistant/helpers/script.py
View File

@ -338,7 +338,7 @@ class Script:
config_cache_key = frozenset((k, str(v)) for k, v in action.items())
config = self._config_cache.get(config_cache_key)
if not config:
config = condition.async_from_config(action, False)
config = await condition.async_from_config(self.hass, action, False)
self._config_cache[config_cache_key] = config
self.last_action = action.get(CONF_ALIAS, action[CONF_CONDITION])

50
tests/components/device_automation/test_init.py
View File

@ -21,7 +21,7 @@ def entity_reg(hass):
return mock_registry(hass)
def _same_triggers(a, b):
def _same_lists(a, b):
if len(a) != len(b):
return False
@ -31,6 +31,50 @@ def _same_triggers(a, b):
return True
async def test_websocket_get_conditions(hass, hass_ws_client, device_reg, entity_reg):
"""Test we get the expected conditions from a light through websocket."""
await async_setup_component(hass, "device_automation", {})
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("light", "test", "5678", device_id=device_entry.id)
expected_conditions = [
{
"condition": "device",
"domain": "light",
"type": "is_off",
"device_id": device_entry.id,
"entity_id": "light.test_5678",
},
{
"condition": "device",
"domain": "light",
"type": "is_on",
"device_id": device_entry.id,
"entity_id": "light.test_5678",
},
]
client = await hass_ws_client(hass)
await client.send_json(
{
"id": 1,
"type": "device_automation/condition/list",
"device_id": device_entry.id,
}
)
msg = await client.receive_json()
assert msg["id"] == 1
assert msg["type"] == TYPE_RESULT
assert msg["success"]
conditions = msg["result"]
assert _same_lists(conditions, expected_conditions)
async def test_websocket_get_triggers(hass, hass_ws_client, device_reg, entity_reg):
"""Test we get the expected triggers from a light through websocket."""
await async_setup_component(hass, "device_automation", {})
@ -71,5 +115,5 @@ async def test_websocket_get_triggers(hass, hass_ws_client, device_reg, entity_r
assert msg["id"] == 1
assert msg["type"] == TYPE_RESULT
assert msg["success"]
triggers = msg["result"]["triggers"]
assert _same_triggers(triggers, expected_triggers)
triggers = msg["result"]
assert _same_lists(triggers, expected_triggers)

115
tests/components/light/test_device_automation.py
View File

@ -6,11 +6,10 @@ from homeassistant.const import STATE_ON, STATE_OFF, CONF_PLATFORM
from homeassistant.setup import async_setup_component
import homeassistant.components.automation as automation
from homeassistant.components.device_automation import (
async_get_device_automation_triggers,
_async_get_device_automations as async_get_device_automations,
)
from homeassistant.helpers import device_registry
from tests.common import (
MockConfigEntry,
async_mock_service,
@ -37,7 +36,7 @@ def calls(hass):
return async_mock_service(hass, "test", "automation")
def _same_triggers(a, b):
def _same_lists(a, b):
if len(a) != len(b):
return False
@ -47,6 +46,37 @@ def _same_triggers(a, b):
return True
async def test_get_conditions(hass, device_reg, entity_reg):
"""Test we get the expected conditions from a light."""
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("light", "test", "5678", device_id=device_entry.id)
expected_conditions = [
{
"condition": "device",
"domain": "light",
"type": "is_off",
"device_id": device_entry.id,
"entity_id": "light.test_5678",
},
{
"condition": "device",
"domain": "light",
"type": "is_on",
"device_id": device_entry.id,
"entity_id": "light.test_5678",
},
]
conditions = await async_get_device_automations(
hass, "async_get_conditions", device_entry.id
)
assert _same_lists(conditions, expected_conditions)
async def test_get_triggers(hass, device_reg, entity_reg):
"""Test we get the expected triggers from a light."""
config_entry = MockConfigEntry(domain="test", data={})
@ -72,8 +102,10 @@ async def test_get_triggers(hass, device_reg, entity_reg):
"entity_id": "light.test_5678",
},
]
triggers = await async_get_device_automation_triggers(hass, device_entry.id)
assert _same_triggers(triggers, expected_triggers)
triggers = await async_get_device_automations(
hass, "async_get_triggers", device_entry.id
)
assert _same_lists(triggers, expected_triggers)
async def test_if_fires_on_state_change(hass, calls):
@ -158,3 +190,76 @@ async def test_if_fires_on_state_change(hass, calls):
assert calls[1].data["some"] == "turn_on state - {} - off - on - None".format(
dev1.entity_id
)
async def test_if_state(hass, calls):
"""Test for turn_on and turn_off conditions."""
platform = getattr(hass.components, "test.light")
platform.init()
assert await async_setup_component(
hass, light.DOMAIN, {light.DOMAIN: {CONF_PLATFORM: "test"}}
)
dev1, dev2, dev3 = platform.DEVICES
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: [
{
"trigger": {"platform": "event", "event_type": "test_event1"},
"condition": [
{
"condition": "device",
"domain": "light",
"entity_id": dev1.entity_id,
"type": "is_on",
}
],
"action": {
"service": "test.automation",
"data_template": {
"some": "is_on {{ trigger.%s }}"
% "}} - {{ trigger.".join(("platform", "event.event_type"))
},
},
},
{
"trigger": {"platform": "event", "event_type": "test_event2"},
"condition": [
{
"condition": "device",
"domain": "light",
"entity_id": dev1.entity_id,
"type": "is_off",
}
],
"action": {
"service": "test.automation",
"data_template": {
"some": "is_off {{ trigger.%s }}"
% "}} - {{ trigger.".join(("platform", "event.event_type"))
},
},
},
]
},
)
await hass.async_block_till_done()
assert hass.states.get(dev1.entity_id).state == STATE_ON
assert len(calls) == 0
hass.bus.async_fire("test_event1")
hass.bus.async_fire("test_event2")
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["some"] == "is_on event - test_event1"
hass.states.async_set(dev1.entity_id, STATE_OFF)
hass.bus.async_fire("test_event1")
hass.bus.async_fire("test_event2")
await hass.async_block_till_done()
assert len(calls) == 2
assert calls[1].data["some"] == "is_off event - test_event2"

301
tests/helpers/test_condition.py
View File

@ -4,182 +4,175 @@ from unittest.mock import patch
from homeassistant.helpers import condition
from homeassistant.util import dt
from tests.common import get_test_home_assistant
async def test_and_condition(hass):
"""Test the 'and' condition."""
test = await condition.async_from_config(
hass,
{
"condition": "and",
"conditions": [
{
"condition": "state",
"entity_id": "sensor.temperature",
"state": "100",
},
{
"condition": "numeric_state",
"entity_id": "sensor.temperature",
"below": 110,
},
],
},
)
hass.states.async_set("sensor.temperature", 120)
assert not test(hass)
hass.states.async_set("sensor.temperature", 105)
assert not test(hass)
hass.states.async_set("sensor.temperature", 100)
assert test(hass)
class TestConditionHelper:
"""Test condition helpers."""
async def test_and_condition_with_template(hass):
"""Test the 'and' condition."""
test = await condition.async_from_config(
hass,
{
"condition": "and",
"conditions": [
{
"condition": "template",
"value_template": '{{ states.sensor.temperature.state == "100" }}',
},
{
"condition": "numeric_state",
"entity_id": "sensor.temperature",
"below": 110,
},
],
},
)
def setup_method(self, method):
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
hass.states.async_set("sensor.temperature", 120)
assert not test(hass)
def teardown_method(self, method):
"""Stop everything that was started."""
self.hass.stop()
hass.states.async_set("sensor.temperature", 105)
assert not test(hass)
def test_and_condition(self):
"""Test the 'and' condition."""
test = condition.from_config(
{
"condition": "and",
"conditions": [
{
"condition": "state",
"entity_id": "sensor.temperature",
"state": "100",
},
{
"condition": "numeric_state",
"entity_id": "sensor.temperature",
"below": 110,
},
],
}
)
hass.states.async_set("sensor.temperature", 100)
assert test(hass)
self.hass.states.set("sensor.temperature", 120)
assert not test(self.hass)
self.hass.states.set("sensor.temperature", 105)
assert not test(self.hass)
async def test_or_condition(hass):
"""Test the 'or' condition."""
test = await condition.async_from_config(
hass,
{
"condition": "or",
"conditions": [
{
"condition": "state",
"entity_id": "sensor.temperature",
"state": "100",
},
{
"condition": "numeric_state",
"entity_id": "sensor.temperature",
"below": 110,
},
],
},
)
self.hass.states.set("sensor.temperature", 100)
assert test(self.hass)
hass.states.async_set("sensor.temperature", 120)
assert not test(hass)
def test_and_condition_with_template(self):
"""Test the 'and' condition."""
test = condition.from_config(
{
"condition": "and",
"conditions": [
{
"condition": "template",
"value_template": '{{ states.sensor.temperature.state == "100" }}',
},
{
"condition": "numeric_state",
"entity_id": "sensor.temperature",
"below": 110,
},
],
}
)
hass.states.async_set("sensor.temperature", 105)
assert test(hass)
self.hass.states.set("sensor.temperature", 120)
assert not test(self.hass)
hass.states.async_set("sensor.temperature", 100)
assert test(hass)
self.hass.states.set("sensor.temperature", 105)
assert not test(self.hass)
self.hass.states.set("sensor.temperature", 100)
assert test(self.hass)
async def test_or_condition_with_template(hass):
"""Test the 'or' condition."""
test = await condition.async_from_config(
hass,
{
"condition": "or",
"conditions": [
{
"condition": "template",
"value_template": '{{ states.sensor.temperature.state == "100" }}',
},
{
"condition": "numeric_state",
"entity_id": "sensor.temperature",
"below": 110,
},
],
},
)
def test_or_condition(self):
"""Test the 'or' condition."""
test = condition.from_config(
{
"condition": "or",
"conditions": [
{
"condition": "state",
"entity_id": "sensor.temperature",
"state": "100",
},
{
"condition": "numeric_state",
"entity_id": "sensor.temperature",
"below": 110,
},
],
}
)
hass.states.async_set("sensor.temperature", 120)
assert not test(hass)
self.hass.states.set("sensor.temperature", 120)
assert not test(self.hass)
hass.states.async_set("sensor.temperature", 105)
assert test(hass)
self.hass.states.set("sensor.temperature", 105)
assert test(self.hass)
hass.states.async_set("sensor.temperature", 100)
assert test(hass)
self.hass.states.set("sensor.temperature", 100)
assert test(self.hass)
def test_or_condition_with_template(self):
"""Test the 'or' condition."""
test = condition.from_config(
{
"condition": "or",
"conditions": [
{
"condition": "template",
"value_template": '{{ states.sensor.temperature.state == "100" }}',
},
{
"condition": "numeric_state",
"entity_id": "sensor.temperature",
"below": 110,
},
],
}
)
async def test_time_window(hass):
"""Test time condition windows."""
sixam = dt.parse_time("06:00:00")
sixpm = dt.parse_time("18:00:00")
self.hass.states.set("sensor.temperature", 120)
assert not test(self.hass)
with patch(
"homeassistant.helpers.condition.dt_util.now",
return_value=dt.now().replace(hour=3),
):
assert not condition.time(after=sixam, before=sixpm)
assert condition.time(after=sixpm, before=sixam)
self.hass.states.set("sensor.temperature", 105)
assert test(self.hass)
with patch(
"homeassistant.helpers.condition.dt_util.now",
return_value=dt.now().replace(hour=9),
):
assert condition.time(after=sixam, before=sixpm)
assert not condition.time(after=sixpm, before=sixam)
self.hass.states.set("sensor.temperature", 100)
assert test(self.hass)
with patch(
"homeassistant.helpers.condition.dt_util.now",
return_value=dt.now().replace(hour=15),
):
assert condition.time(after=sixam, before=sixpm)
assert not condition.time(after=sixpm, before=sixam)
def test_time_window(self):
"""Test time condition windows."""
sixam = dt.parse_time("06:00:00")
sixpm = dt.parse_time("18:00:00")
with patch(
"homeassistant.helpers.condition.dt_util.now",
return_value=dt.now().replace(hour=21),
):
assert not condition.time(after=sixam, before=sixpm)
assert condition.time(after=sixpm, before=sixam)
with patch(
"homeassistant.helpers.condition.dt_util.now",
return_value=dt.now().replace(hour=3),
):
assert not condition.time(after=sixam, before=sixpm)
assert condition.time(after=sixpm, before=sixam)
with patch(
"homeassistant.helpers.condition.dt_util.now",
return_value=dt.now().replace(hour=9),
):
assert condition.time(after=sixam, before=sixpm)
assert not condition.time(after=sixpm, before=sixam)
async def test_if_numeric_state_not_raise_on_unavailable(hass):
"""Test numeric_state doesn't raise on unavailable/unknown state."""
test = await condition.async_from_config(
hass,
{"condition": "numeric_state", "entity_id": "sensor.temperature", "below": 42},
)
with patch(
"homeassistant.helpers.condition.dt_util.now",
return_value=dt.now().replace(hour=15),
):
assert condition.time(after=sixam, before=sixpm)
assert not condition.time(after=sixpm, before=sixam)
with patch("homeassistant.helpers.condition._LOGGER.warning") as logwarn:
hass.states.async_set("sensor.temperature", "unavailable")
assert not test(hass)
assert len(logwarn.mock_calls) == 0
with patch(
"homeassistant.helpers.condition.dt_util.now",
return_value=dt.now().replace(hour=21),
):
assert not condition.time(after=sixam, before=sixpm)
assert condition.time(after=sixpm, before=sixam)
def test_if_numeric_state_not_raise_on_unavailable(self):
"""Test numeric_state doesn't raise on unavailable/unknown state."""
test = condition.from_config(
{
"condition": "numeric_state",
"entity_id": "sensor.temperature",
"below": 42,
}
)
with patch("homeassistant.helpers.condition._LOGGER.warning") as logwarn:
self.hass.states.set("sensor.temperature", "unavailable")
assert not test(self.hass)
assert len(logwarn.mock_calls) == 0
self.hass.states.set("sensor.temperature", "unknown")
assert not test(self.hass)
assert len(logwarn.mock_calls) == 0
hass.states.async_set("sensor.temperature", "unknown")
assert not test(hass)
assert len(logwarn.mock_calls) == 0