2611 lines
84 KiB
Python
2611 lines
84 KiB
Python
"""Test the condition helper."""
|
|
|
|
from datetime import timedelta
|
|
import io
|
|
from typing import Any
|
|
from unittest.mock import AsyncMock, Mock, patch
|
|
|
|
from freezegun import freeze_time
|
|
import pytest
|
|
from pytest_unordered import unordered
|
|
import voluptuous as vol
|
|
|
|
from homeassistant.components.device_automation import (
|
|
DOMAIN as DOMAIN_DEVICE_AUTOMATION,
|
|
)
|
|
from homeassistant.components.sensor import SensorDeviceClass
|
|
from homeassistant.components.sun import DOMAIN as DOMAIN_SUN
|
|
from homeassistant.components.system_health import DOMAIN as DOMAIN_SYSTEM_HEALTH
|
|
from homeassistant.const import (
|
|
ATTR_DEVICE_CLASS,
|
|
CONF_CONDITION,
|
|
CONF_DEVICE_ID,
|
|
CONF_DOMAIN,
|
|
STATE_UNAVAILABLE,
|
|
STATE_UNKNOWN,
|
|
)
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.exceptions import ConditionError, HomeAssistantError
|
|
from homeassistant.helpers import (
|
|
condition,
|
|
config_validation as cv,
|
|
entity_registry as er,
|
|
trace,
|
|
)
|
|
from homeassistant.helpers.template import Template
|
|
from homeassistant.helpers.typing import ConfigType
|
|
from homeassistant.loader import Integration, async_get_integration
|
|
from homeassistant.setup import async_setup_component
|
|
from homeassistant.util import dt as dt_util
|
|
from homeassistant.util.yaml.loader import parse_yaml
|
|
|
|
from tests.common import MockModule, MockPlatform, mock_integration, mock_platform
|
|
|
|
|
|
def assert_element(trace_element, expected_element, path):
|
|
"""Assert a trace element is as expected.
|
|
|
|
Note: Unused variable 'path' is passed to get helpful errors from pytest.
|
|
"""
|
|
expected_result = expected_element.get("result", {})
|
|
# Check that every item in expected_element is present and equal in trace_element
|
|
# The redundant set operation gives helpful errors from pytest
|
|
assert not set(expected_result) - set(trace_element._result or {})
|
|
for result_key, result in expected_result.items():
|
|
assert trace_element._result[result_key] == result
|
|
|
|
# Check for unexpected items in trace_element
|
|
assert not set(trace_element._result or {}) - set(expected_result)
|
|
|
|
if "error_type" in expected_element:
|
|
assert isinstance(trace_element._error, expected_element["error_type"])
|
|
else:
|
|
assert trace_element._error is None
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def prepare_condition_trace() -> None:
|
|
"""Clear previous trace."""
|
|
trace.trace_clear()
|
|
|
|
|
|
def assert_condition_trace(expected):
|
|
"""Assert a trace condition sequence is as expected."""
|
|
condition_trace = trace.trace_get(clear=False)
|
|
trace.trace_clear()
|
|
expected_trace_keys = list(expected.keys())
|
|
assert list(condition_trace.keys()) == expected_trace_keys
|
|
for trace_key_index, key in enumerate(expected_trace_keys):
|
|
assert len(condition_trace[key]) == len(expected[key])
|
|
for index, element in enumerate(expected[key]):
|
|
path = f"[{trace_key_index}][{index}]"
|
|
assert_element(condition_trace[key][index], element, path)
|
|
|
|
|
|
async def test_invalid_condition(hass: HomeAssistant) -> None:
|
|
"""Test if invalid condition raises."""
|
|
with pytest.raises(HomeAssistantError):
|
|
await condition.async_from_config(
|
|
hass,
|
|
{
|
|
"condition": "invalid",
|
|
"conditions": [
|
|
{
|
|
"condition": "state",
|
|
"entity_id": "sensor.temperature",
|
|
"state": "100",
|
|
},
|
|
],
|
|
},
|
|
)
|
|
|
|
|
|
async def test_and_condition(hass: HomeAssistant) -> None:
|
|
"""Test the 'and' condition."""
|
|
config = {
|
|
"alias": "And Condition",
|
|
"condition": "and",
|
|
"conditions": [
|
|
{
|
|
"condition": "state",
|
|
"entity_id": "sensor.temperature",
|
|
"state": "100",
|
|
},
|
|
{
|
|
"condition": "numeric_state",
|
|
"entity_id": "sensor.temperature",
|
|
"below": 110,
|
|
},
|
|
],
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
with pytest.raises(ConditionError):
|
|
test(hass)
|
|
assert_condition_trace(
|
|
{
|
|
"": [{"error_type": ConditionError}],
|
|
"conditions/0": [{"error_type": ConditionError}],
|
|
"conditions/0/entity_id/0": [{"error_type": ConditionError}],
|
|
"conditions/1": [{"error_type": ConditionError}],
|
|
"conditions/1/entity_id/0": [{"error_type": ConditionError}],
|
|
}
|
|
)
|
|
|
|
hass.states.async_set("sensor.temperature", 120)
|
|
assert not test(hass)
|
|
assert_condition_trace(
|
|
{
|
|
"": [{"result": {"result": False}}],
|
|
"conditions/0": [{"result": {"result": False}}],
|
|
"conditions/0/entity_id/0": [
|
|
{"result": {"result": False, "state": "120", "wanted_state": "100"}}
|
|
],
|
|
}
|
|
)
|
|
|
|
hass.states.async_set("sensor.temperature", 105)
|
|
assert not test(hass)
|
|
assert_condition_trace(
|
|
{
|
|
"": [{"result": {"result": False}}],
|
|
"conditions/0": [{"result": {"result": False}}],
|
|
"conditions/0/entity_id/0": [
|
|
{"result": {"result": False, "state": "105", "wanted_state": "100"}}
|
|
],
|
|
}
|
|
)
|
|
|
|
hass.states.async_set("sensor.temperature", 100)
|
|
assert test(hass)
|
|
assert_condition_trace(
|
|
{
|
|
"": [{"result": {"result": True}}],
|
|
"conditions/0": [{"result": {"result": True}}],
|
|
"conditions/0/entity_id/0": [
|
|
{"result": {"result": True, "state": "100", "wanted_state": "100"}}
|
|
],
|
|
"conditions/1": [{"result": {"result": True}}],
|
|
"conditions/1/entity_id/0": [{"result": {"result": True, "state": 100.0}}],
|
|
}
|
|
)
|
|
|
|
|
|
async def test_and_condition_raises(hass: HomeAssistant) -> None:
|
|
"""Test the 'and' condition."""
|
|
config = {
|
|
"alias": "And Condition",
|
|
"condition": "and",
|
|
"conditions": [
|
|
{
|
|
"condition": "state",
|
|
"entity_id": "sensor.temperature",
|
|
"state": "100",
|
|
},
|
|
{
|
|
"condition": "numeric_state",
|
|
"entity_id": "sensor.temperature2",
|
|
"above": 110,
|
|
},
|
|
],
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
# All subconditions raise, the AND-condition should raise
|
|
with pytest.raises(ConditionError):
|
|
test(hass)
|
|
assert_condition_trace(
|
|
{
|
|
"": [{"error_type": ConditionError}],
|
|
"conditions/0": [{"error_type": ConditionError}],
|
|
"conditions/0/entity_id/0": [{"error_type": ConditionError}],
|
|
"conditions/1": [{"error_type": ConditionError}],
|
|
"conditions/1/entity_id/0": [{"error_type": ConditionError}],
|
|
}
|
|
)
|
|
|
|
# The first subconditions raises, the second returns True, the AND-condition
|
|
# should raise
|
|
hass.states.async_set("sensor.temperature2", 120)
|
|
with pytest.raises(ConditionError):
|
|
test(hass)
|
|
assert_condition_trace(
|
|
{
|
|
"": [{"error_type": ConditionError}],
|
|
"conditions/0": [{"error_type": ConditionError}],
|
|
"conditions/0/entity_id/0": [{"error_type": ConditionError}],
|
|
"conditions/1": [{"result": {"result": True}}],
|
|
"conditions/1/entity_id/0": [{"result": {"result": True, "state": 120.0}}],
|
|
}
|
|
)
|
|
|
|
# The first subconditions raises, the second returns False, the AND-condition
|
|
# should return False
|
|
hass.states.async_set("sensor.temperature2", 90)
|
|
assert not test(hass)
|
|
assert_condition_trace(
|
|
{
|
|
"": [{"result": {"result": False}}],
|
|
"conditions/0": [{"error_type": ConditionError}],
|
|
"conditions/0/entity_id/0": [{"error_type": ConditionError}],
|
|
"conditions/1": [{"result": {"result": False}}],
|
|
"conditions/1/entity_id/0": [
|
|
{
|
|
"result": {
|
|
"result": False,
|
|
"state": 90.0,
|
|
"wanted_state_above": 110.0,
|
|
}
|
|
}
|
|
],
|
|
}
|
|
)
|
|
|
|
|
|
async def test_and_condition_with_template(hass: HomeAssistant) -> None:
|
|
"""Test the 'and' condition."""
|
|
config = {
|
|
"condition": "and",
|
|
"conditions": [
|
|
{
|
|
"alias": "Template Condition",
|
|
"condition": "template",
|
|
"value_template": '{{ states.sensor.temperature.state == "100" }}',
|
|
},
|
|
{
|
|
"condition": "numeric_state",
|
|
"entity_id": "sensor.temperature",
|
|
"below": 110,
|
|
},
|
|
],
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
hass.states.async_set("sensor.temperature", 120)
|
|
assert not test(hass)
|
|
assert_condition_trace(
|
|
{
|
|
"": [{"result": {"result": False}}],
|
|
"conditions/0": [
|
|
{"result": {"entities": ["sensor.temperature"], "result": False}}
|
|
],
|
|
}
|
|
)
|
|
|
|
hass.states.async_set("sensor.temperature", 105)
|
|
assert not test(hass)
|
|
|
|
hass.states.async_set("sensor.temperature", 100)
|
|
assert test(hass)
|
|
|
|
|
|
async def test_and_condition_shorthand(hass: HomeAssistant) -> None:
|
|
"""Test the 'and' condition shorthand."""
|
|
config = {
|
|
"alias": "And Condition Shorthand",
|
|
"and": [
|
|
{
|
|
"alias": "Template Condition",
|
|
"condition": "template",
|
|
"value_template": '{{ states.sensor.temperature.state == "100" }}',
|
|
},
|
|
{
|
|
"condition": "numeric_state",
|
|
"entity_id": "sensor.temperature",
|
|
"below": 110,
|
|
},
|
|
],
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
assert config["alias"] == "And Condition Shorthand"
|
|
assert "and" not in config
|
|
|
|
hass.states.async_set("sensor.temperature", 120)
|
|
assert not test(hass)
|
|
assert_condition_trace(
|
|
{
|
|
"": [{"result": {"result": False}}],
|
|
"conditions/0": [
|
|
{"result": {"entities": ["sensor.temperature"], "result": False}}
|
|
],
|
|
}
|
|
)
|
|
|
|
hass.states.async_set("sensor.temperature", 105)
|
|
assert not test(hass)
|
|
|
|
hass.states.async_set("sensor.temperature", 100)
|
|
assert test(hass)
|
|
|
|
|
|
async def test_and_condition_list_shorthand(hass: HomeAssistant) -> None:
|
|
"""Test the 'and' condition list shorthand."""
|
|
config = {
|
|
"alias": "And Condition List Shorthand",
|
|
"condition": [
|
|
{
|
|
"alias": "Template Condition",
|
|
"condition": "template",
|
|
"value_template": '{{ states.sensor.temperature.state == "100" }}',
|
|
},
|
|
{
|
|
"condition": "numeric_state",
|
|
"entity_id": "sensor.temperature",
|
|
"below": 110,
|
|
},
|
|
],
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
assert config["alias"] == "And Condition List Shorthand"
|
|
assert "and" not in config
|
|
|
|
hass.states.async_set("sensor.temperature", 120)
|
|
assert not test(hass)
|
|
assert_condition_trace(
|
|
{
|
|
"": [{"result": {"result": False}}],
|
|
"conditions/0": [
|
|
{"result": {"entities": ["sensor.temperature"], "result": False}}
|
|
],
|
|
}
|
|
)
|
|
|
|
hass.states.async_set("sensor.temperature", 105)
|
|
assert not test(hass)
|
|
|
|
hass.states.async_set("sensor.temperature", 100)
|
|
assert test(hass)
|
|
|
|
|
|
async def test_malformed_and_condition_list_shorthand(hass: HomeAssistant) -> None:
|
|
"""Test the 'and' condition list shorthand syntax check."""
|
|
config = {
|
|
"alias": "Bad shorthand syntax",
|
|
"condition": ["bad", "syntax"],
|
|
}
|
|
|
|
with pytest.raises(vol.MultipleInvalid):
|
|
cv.CONDITION_SCHEMA(config)
|
|
|
|
|
|
async def test_or_condition(hass: HomeAssistant) -> None:
|
|
"""Test the 'or' condition."""
|
|
config = {
|
|
"alias": "Or Condition",
|
|
"condition": "or",
|
|
"conditions": [
|
|
{
|
|
"condition": "state",
|
|
"entity_id": "sensor.temperature",
|
|
"state": "100",
|
|
},
|
|
{
|
|
"condition": "numeric_state",
|
|
"entity_id": "sensor.temperature",
|
|
"below": 110,
|
|
},
|
|
],
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
with pytest.raises(ConditionError):
|
|
test(hass)
|
|
assert_condition_trace(
|
|
{
|
|
"": [{"error_type": ConditionError}],
|
|
"conditions/0": [{"error_type": ConditionError}],
|
|
"conditions/0/entity_id/0": [{"error_type": ConditionError}],
|
|
"conditions/1": [{"error_type": ConditionError}],
|
|
"conditions/1/entity_id/0": [{"error_type": ConditionError}],
|
|
}
|
|
)
|
|
|
|
hass.states.async_set("sensor.temperature", 120)
|
|
assert not test(hass)
|
|
assert_condition_trace(
|
|
{
|
|
"": [{"result": {"result": False}}],
|
|
"conditions/0": [{"result": {"result": False}}],
|
|
"conditions/0/entity_id/0": [
|
|
{"result": {"result": False, "state": "120", "wanted_state": "100"}}
|
|
],
|
|
"conditions/1": [{"result": {"result": False}}],
|
|
"conditions/1/entity_id/0": [
|
|
{
|
|
"result": {
|
|
"result": False,
|
|
"state": 120.0,
|
|
"wanted_state_below": 110.0,
|
|
}
|
|
}
|
|
],
|
|
}
|
|
)
|
|
|
|
hass.states.async_set("sensor.temperature", 105)
|
|
assert test(hass)
|
|
assert_condition_trace(
|
|
{
|
|
"": [{"result": {"result": True}}],
|
|
"conditions/0": [{"result": {"result": False}}],
|
|
"conditions/0/entity_id/0": [
|
|
{"result": {"result": False, "state": "105", "wanted_state": "100"}}
|
|
],
|
|
"conditions/1": [{"result": {"result": True}}],
|
|
"conditions/1/entity_id/0": [{"result": {"result": True, "state": 105.0}}],
|
|
}
|
|
)
|
|
|
|
hass.states.async_set("sensor.temperature", 100)
|
|
assert test(hass)
|
|
assert_condition_trace(
|
|
{
|
|
"": [{"result": {"result": True}}],
|
|
"conditions/0": [{"result": {"result": True}}],
|
|
"conditions/0/entity_id/0": [
|
|
{"result": {"result": True, "state": "100", "wanted_state": "100"}}
|
|
],
|
|
}
|
|
)
|
|
|
|
|
|
async def test_or_condition_raises(hass: HomeAssistant) -> None:
|
|
"""Test the 'or' condition."""
|
|
config = {
|
|
"alias": "Or Condition",
|
|
"condition": "or",
|
|
"conditions": [
|
|
{
|
|
"condition": "state",
|
|
"entity_id": "sensor.temperature",
|
|
"state": "100",
|
|
},
|
|
{
|
|
"condition": "numeric_state",
|
|
"entity_id": "sensor.temperature2",
|
|
"above": 110,
|
|
},
|
|
],
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
# All subconditions raise, the OR-condition should raise
|
|
with pytest.raises(ConditionError):
|
|
test(hass)
|
|
assert_condition_trace(
|
|
{
|
|
"": [{"error_type": ConditionError}],
|
|
"conditions/0": [{"error_type": ConditionError}],
|
|
"conditions/0/entity_id/0": [{"error_type": ConditionError}],
|
|
"conditions/1": [{"error_type": ConditionError}],
|
|
"conditions/1/entity_id/0": [{"error_type": ConditionError}],
|
|
}
|
|
)
|
|
|
|
# The first subconditions raises, the second returns False, the OR-condition
|
|
# should raise
|
|
hass.states.async_set("sensor.temperature2", 100)
|
|
with pytest.raises(ConditionError):
|
|
test(hass)
|
|
assert_condition_trace(
|
|
{
|
|
"": [{"error_type": ConditionError}],
|
|
"conditions/0": [{"error_type": ConditionError}],
|
|
"conditions/0/entity_id/0": [{"error_type": ConditionError}],
|
|
"conditions/1": [{"result": {"result": False}}],
|
|
"conditions/1/entity_id/0": [
|
|
{
|
|
"result": {
|
|
"result": False,
|
|
"state": 100.0,
|
|
"wanted_state_above": 110.0,
|
|
}
|
|
}
|
|
],
|
|
}
|
|
)
|
|
|
|
# The first subconditions raises, the second returns True, the OR-condition
|
|
# should return True
|
|
hass.states.async_set("sensor.temperature2", 120)
|
|
assert test(hass)
|
|
assert_condition_trace(
|
|
{
|
|
"": [{"result": {"result": True}}],
|
|
"conditions/0": [{"error_type": ConditionError}],
|
|
"conditions/0/entity_id/0": [{"error_type": ConditionError}],
|
|
"conditions/1": [{"result": {"result": True}}],
|
|
"conditions/1/entity_id/0": [{"result": {"result": True, "state": 120.0}}],
|
|
}
|
|
)
|
|
|
|
|
|
async def test_or_condition_with_template(hass: HomeAssistant) -> None:
|
|
"""Test the 'or' condition."""
|
|
config = {
|
|
"condition": "or",
|
|
"conditions": [
|
|
{'{{ states.sensor.temperature.state == "100" }}'},
|
|
{
|
|
"condition": "numeric_state",
|
|
"entity_id": "sensor.temperature",
|
|
"below": 110,
|
|
},
|
|
],
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
hass.states.async_set("sensor.temperature", 120)
|
|
assert not test(hass)
|
|
|
|
hass.states.async_set("sensor.temperature", 105)
|
|
assert test(hass)
|
|
|
|
hass.states.async_set("sensor.temperature", 100)
|
|
assert test(hass)
|
|
|
|
|
|
async def test_or_condition_shorthand(hass: HomeAssistant) -> None:
|
|
"""Test the 'or' condition shorthand."""
|
|
config = {
|
|
"alias": "Or Condition Shorthand",
|
|
"or": [
|
|
{'{{ states.sensor.temperature.state == "100" }}'},
|
|
{
|
|
"condition": "numeric_state",
|
|
"entity_id": "sensor.temperature",
|
|
"below": 110,
|
|
},
|
|
],
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
assert config["alias"] == "Or Condition Shorthand"
|
|
assert "or" not in config
|
|
|
|
hass.states.async_set("sensor.temperature", 120)
|
|
assert not test(hass)
|
|
|
|
hass.states.async_set("sensor.temperature", 105)
|
|
assert test(hass)
|
|
|
|
hass.states.async_set("sensor.temperature", 100)
|
|
assert test(hass)
|
|
|
|
|
|
async def test_not_condition(hass: HomeAssistant) -> None:
|
|
"""Test the 'not' condition."""
|
|
config = {
|
|
"alias": "Not Condition",
|
|
"condition": "not",
|
|
"conditions": [
|
|
{
|
|
"condition": "state",
|
|
"entity_id": "sensor.temperature",
|
|
"state": "100",
|
|
},
|
|
{
|
|
"condition": "numeric_state",
|
|
"entity_id": "sensor.temperature",
|
|
"below": 50,
|
|
},
|
|
],
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
with pytest.raises(ConditionError):
|
|
test(hass)
|
|
assert_condition_trace(
|
|
{
|
|
"": [{"error_type": ConditionError}],
|
|
"conditions/0": [{"error_type": ConditionError}],
|
|
"conditions/0/entity_id/0": [{"error_type": ConditionError}],
|
|
"conditions/1": [{"error_type": ConditionError}],
|
|
"conditions/1/entity_id/0": [{"error_type": ConditionError}],
|
|
}
|
|
)
|
|
|
|
hass.states.async_set("sensor.temperature", 101)
|
|
assert test(hass)
|
|
assert_condition_trace(
|
|
{
|
|
"": [{"result": {"result": True}}],
|
|
"conditions/0": [{"result": {"result": False}}],
|
|
"conditions/0/entity_id/0": [
|
|
{"result": {"result": False, "state": "101", "wanted_state": "100"}}
|
|
],
|
|
"conditions/1": [{"result": {"result": False}}],
|
|
"conditions/1/entity_id/0": [
|
|
{
|
|
"result": {
|
|
"result": False,
|
|
"state": 101.0,
|
|
"wanted_state_below": 50.0,
|
|
}
|
|
}
|
|
],
|
|
}
|
|
)
|
|
|
|
hass.states.async_set("sensor.temperature", 50)
|
|
assert test(hass)
|
|
assert_condition_trace(
|
|
{
|
|
"": [{"result": {"result": True}}],
|
|
"conditions/0": [{"result": {"result": False}}],
|
|
"conditions/0/entity_id/0": [
|
|
{"result": {"result": False, "state": "50", "wanted_state": "100"}}
|
|
],
|
|
"conditions/1": [{"result": {"result": False}}],
|
|
"conditions/1/entity_id/0": [
|
|
{"result": {"result": False, "state": 50.0, "wanted_state_below": 50.0}}
|
|
],
|
|
}
|
|
)
|
|
|
|
hass.states.async_set("sensor.temperature", 49)
|
|
assert not test(hass)
|
|
assert_condition_trace(
|
|
{
|
|
"": [{"result": {"result": False}}],
|
|
"conditions/0": [{"result": {"result": False}}],
|
|
"conditions/0/entity_id/0": [
|
|
{"result": {"result": False, "state": "49", "wanted_state": "100"}}
|
|
],
|
|
"conditions/1": [{"result": {"result": True}}],
|
|
"conditions/1/entity_id/0": [{"result": {"result": True, "state": 49.0}}],
|
|
}
|
|
)
|
|
|
|
hass.states.async_set("sensor.temperature", 100)
|
|
assert not test(hass)
|
|
assert_condition_trace(
|
|
{
|
|
"": [{"result": {"result": False}}],
|
|
"conditions/0": [{"result": {"result": True}}],
|
|
"conditions/0/entity_id/0": [
|
|
{"result": {"result": True, "state": "100", "wanted_state": "100"}}
|
|
],
|
|
}
|
|
)
|
|
|
|
|
|
async def test_not_condition_raises(hass: HomeAssistant) -> None:
|
|
"""Test the 'and' condition."""
|
|
config = {
|
|
"alias": "Not Condition",
|
|
"condition": "not",
|
|
"conditions": [
|
|
{
|
|
"condition": "state",
|
|
"entity_id": "sensor.temperature",
|
|
"state": "100",
|
|
},
|
|
{
|
|
"condition": "numeric_state",
|
|
"entity_id": "sensor.temperature2",
|
|
"below": 50,
|
|
},
|
|
],
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
# All subconditions raise, the NOT-condition should raise
|
|
with pytest.raises(ConditionError):
|
|
test(hass)
|
|
assert_condition_trace(
|
|
{
|
|
"": [{"error_type": ConditionError}],
|
|
"conditions/0": [{"error_type": ConditionError}],
|
|
"conditions/0/entity_id/0": [{"error_type": ConditionError}],
|
|
"conditions/1": [{"error_type": ConditionError}],
|
|
"conditions/1/entity_id/0": [{"error_type": ConditionError}],
|
|
}
|
|
)
|
|
|
|
# The first subconditions raises, the second returns False, the NOT-condition
|
|
# should raise
|
|
hass.states.async_set("sensor.temperature2", 90)
|
|
with pytest.raises(ConditionError):
|
|
test(hass)
|
|
assert_condition_trace(
|
|
{
|
|
"": [{"error_type": ConditionError}],
|
|
"conditions/0": [{"error_type": ConditionError}],
|
|
"conditions/0/entity_id/0": [{"error_type": ConditionError}],
|
|
"conditions/1": [{"result": {"result": False}}],
|
|
"conditions/1/entity_id/0": [
|
|
{"result": {"result": False, "state": 90.0, "wanted_state_below": 50.0}}
|
|
],
|
|
}
|
|
)
|
|
|
|
# The first subconditions raises, the second returns True, the NOT-condition
|
|
# should return False
|
|
hass.states.async_set("sensor.temperature2", 40)
|
|
assert not test(hass)
|
|
assert_condition_trace(
|
|
{
|
|
"": [{"result": {"result": False}}],
|
|
"conditions/0": [{"error_type": ConditionError}],
|
|
"conditions/0/entity_id/0": [{"error_type": ConditionError}],
|
|
"conditions/1": [{"result": {"result": True}}],
|
|
"conditions/1/entity_id/0": [{"result": {"result": True, "state": 40.0}}],
|
|
}
|
|
)
|
|
|
|
|
|
async def test_not_condition_with_template(hass: HomeAssistant) -> None:
|
|
"""Test the 'or' condition."""
|
|
config = {
|
|
"condition": "not",
|
|
"conditions": [
|
|
{
|
|
"condition": "template",
|
|
"value_template": '{{ states.sensor.temperature.state == "100" }}',
|
|
},
|
|
{
|
|
"condition": "numeric_state",
|
|
"entity_id": "sensor.temperature",
|
|
"below": 50,
|
|
},
|
|
],
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
hass.states.async_set("sensor.temperature", 101)
|
|
assert test(hass)
|
|
|
|
hass.states.async_set("sensor.temperature", 50)
|
|
assert test(hass)
|
|
|
|
hass.states.async_set("sensor.temperature", 49)
|
|
assert not test(hass)
|
|
|
|
hass.states.async_set("sensor.temperature", 100)
|
|
assert not test(hass)
|
|
|
|
|
|
async def test_not_condition_shorthand(hass: HomeAssistant) -> None:
|
|
"""Test the 'or' condition shorthand."""
|
|
config = {
|
|
"alias": "Not Condition Shorthand",
|
|
"not": [
|
|
{
|
|
"condition": "template",
|
|
"value_template": '{{ states.sensor.temperature.state == "100" }}',
|
|
},
|
|
{
|
|
"condition": "numeric_state",
|
|
"entity_id": "sensor.temperature",
|
|
"below": 50,
|
|
},
|
|
],
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
assert config["alias"] == "Not Condition Shorthand"
|
|
assert "not" not in config
|
|
|
|
hass.states.async_set("sensor.temperature", 101)
|
|
assert test(hass)
|
|
|
|
hass.states.async_set("sensor.temperature", 50)
|
|
assert test(hass)
|
|
|
|
hass.states.async_set("sensor.temperature", 49)
|
|
assert not test(hass)
|
|
|
|
hass.states.async_set("sensor.temperature", 100)
|
|
assert not test(hass)
|
|
|
|
|
|
async def test_time_window(hass: HomeAssistant) -> None:
|
|
"""Test time condition windows."""
|
|
sixam = "06:00:00"
|
|
sixpm = "18:00:00"
|
|
|
|
config1 = {
|
|
"alias": "Time Cond",
|
|
"condition": "time",
|
|
"after": sixam,
|
|
"before": sixpm,
|
|
}
|
|
config1 = cv.CONDITION_SCHEMA(config1)
|
|
config1 = await condition.async_validate_condition_config(hass, config1)
|
|
config2 = {
|
|
"alias": "Time Cond",
|
|
"condition": "time",
|
|
"after": sixpm,
|
|
"before": sixam,
|
|
}
|
|
config2 = cv.CONDITION_SCHEMA(config2)
|
|
config2 = await condition.async_validate_condition_config(hass, config2)
|
|
test1 = await condition.async_from_config(hass, config1)
|
|
test2 = await condition.async_from_config(hass, config2)
|
|
|
|
with patch(
|
|
"homeassistant.helpers.condition.dt_util.now",
|
|
return_value=dt_util.now().replace(hour=3),
|
|
):
|
|
assert not test1(hass)
|
|
assert test2(hass)
|
|
|
|
with patch(
|
|
"homeassistant.helpers.condition.dt_util.now",
|
|
return_value=dt_util.now().replace(hour=9),
|
|
):
|
|
assert test1(hass)
|
|
assert not test2(hass)
|
|
|
|
with patch(
|
|
"homeassistant.helpers.condition.dt_util.now",
|
|
return_value=dt_util.now().replace(hour=15),
|
|
):
|
|
assert test1(hass)
|
|
assert not test2(hass)
|
|
|
|
with patch(
|
|
"homeassistant.helpers.condition.dt_util.now",
|
|
return_value=dt_util.now().replace(hour=21),
|
|
):
|
|
assert not test1(hass)
|
|
assert test2(hass)
|
|
|
|
|
|
async def test_time_using_input_datetime(hass: HomeAssistant) -> None:
|
|
"""Test time conditions using input_datetime entities."""
|
|
await async_setup_component(
|
|
hass,
|
|
"input_datetime",
|
|
{
|
|
"input_datetime": {
|
|
"am": {"has_date": True, "has_time": True},
|
|
"pm": {"has_date": True, "has_time": True},
|
|
}
|
|
},
|
|
)
|
|
|
|
await hass.services.async_call(
|
|
"input_datetime",
|
|
"set_datetime",
|
|
{
|
|
"entity_id": "input_datetime.am",
|
|
"datetime": str(
|
|
dt_util.now()
|
|
.replace(hour=6, minute=0, second=0, microsecond=0)
|
|
.replace(tzinfo=None)
|
|
),
|
|
},
|
|
blocking=True,
|
|
)
|
|
|
|
await hass.services.async_call(
|
|
"input_datetime",
|
|
"set_datetime",
|
|
{
|
|
"entity_id": "input_datetime.pm",
|
|
"datetime": str(
|
|
dt_util.now()
|
|
.replace(hour=18, minute=0, second=0, microsecond=0)
|
|
.replace(tzinfo=None)
|
|
),
|
|
},
|
|
blocking=True,
|
|
)
|
|
|
|
with patch(
|
|
"homeassistant.helpers.condition.dt_util.now",
|
|
return_value=dt_util.now().replace(hour=3),
|
|
):
|
|
assert not condition.time(
|
|
hass, after="input_datetime.am", before="input_datetime.pm"
|
|
)
|
|
assert condition.time(
|
|
hass, after="input_datetime.pm", before="input_datetime.am"
|
|
)
|
|
|
|
with patch(
|
|
"homeassistant.helpers.condition.dt_util.now",
|
|
return_value=dt_util.now().replace(hour=9),
|
|
):
|
|
assert condition.time(
|
|
hass, after="input_datetime.am", before="input_datetime.pm"
|
|
)
|
|
assert not condition.time(
|
|
hass, after="input_datetime.pm", before="input_datetime.am"
|
|
)
|
|
|
|
with patch(
|
|
"homeassistant.helpers.condition.dt_util.now",
|
|
return_value=dt_util.now().replace(hour=15),
|
|
):
|
|
assert condition.time(
|
|
hass, after="input_datetime.am", before="input_datetime.pm"
|
|
)
|
|
assert not condition.time(
|
|
hass, after="input_datetime.pm", before="input_datetime.am"
|
|
)
|
|
|
|
with patch(
|
|
"homeassistant.helpers.condition.dt_util.now",
|
|
return_value=dt_util.now().replace(hour=21),
|
|
):
|
|
assert not condition.time(
|
|
hass, after="input_datetime.am", before="input_datetime.pm"
|
|
)
|
|
assert condition.time(
|
|
hass, after="input_datetime.pm", before="input_datetime.am"
|
|
)
|
|
|
|
# Trigger on PM time
|
|
with patch(
|
|
"homeassistant.helpers.condition.dt_util.now",
|
|
return_value=dt_util.now().replace(hour=18, minute=0, second=0),
|
|
):
|
|
assert condition.time(
|
|
hass, after="input_datetime.pm", before="input_datetime.am"
|
|
)
|
|
assert not condition.time(
|
|
hass, after="input_datetime.am", before="input_datetime.pm"
|
|
)
|
|
assert condition.time(hass, after="input_datetime.pm")
|
|
assert not condition.time(hass, before="input_datetime.pm")
|
|
|
|
# Trigger on AM time
|
|
with patch(
|
|
"homeassistant.helpers.condition.dt_util.now",
|
|
return_value=dt_util.now().replace(hour=6, minute=0, second=0),
|
|
):
|
|
assert not condition.time(
|
|
hass, after="input_datetime.pm", before="input_datetime.am"
|
|
)
|
|
assert condition.time(
|
|
hass, after="input_datetime.am", before="input_datetime.pm"
|
|
)
|
|
assert condition.time(hass, after="input_datetime.am")
|
|
assert not condition.time(hass, before="input_datetime.am")
|
|
|
|
with pytest.raises(ConditionError):
|
|
condition.time(hass, after="input_datetime.not_existing")
|
|
|
|
with pytest.raises(ConditionError):
|
|
condition.time(hass, before="input_datetime.not_existing")
|
|
|
|
|
|
async def test_time_using_time(hass: HomeAssistant) -> None:
|
|
"""Test time conditions using time entities."""
|
|
hass.states.async_set(
|
|
"time.am",
|
|
"06:00:00", # 6 am local time
|
|
)
|
|
hass.states.async_set(
|
|
"time.pm",
|
|
"18:00:00", # 6 pm local time
|
|
)
|
|
hass.states.async_set(
|
|
"time.unknown_state",
|
|
STATE_UNKNOWN,
|
|
)
|
|
hass.states.async_set(
|
|
"time.unavailable_state",
|
|
STATE_UNAVAILABLE,
|
|
)
|
|
|
|
with patch(
|
|
"homeassistant.helpers.condition.dt_util.now",
|
|
return_value=dt_util.now().replace(hour=3),
|
|
):
|
|
assert not condition.time(hass, after="time.am", before="time.pm")
|
|
assert condition.time(hass, after="time.pm", before="time.am")
|
|
|
|
with patch(
|
|
"homeassistant.helpers.condition.dt_util.now",
|
|
return_value=dt_util.now().replace(hour=9),
|
|
):
|
|
assert condition.time(hass, after="time.am", before="time.pm")
|
|
assert not condition.time(hass, after="time.pm", before="time.am")
|
|
|
|
with patch(
|
|
"homeassistant.helpers.condition.dt_util.now",
|
|
return_value=dt_util.now().replace(hour=15),
|
|
):
|
|
assert condition.time(hass, after="time.am", before="time.pm")
|
|
assert not condition.time(hass, after="time.pm", before="time.am")
|
|
|
|
with patch(
|
|
"homeassistant.helpers.condition.dt_util.now",
|
|
return_value=dt_util.now().replace(hour=21),
|
|
):
|
|
assert not condition.time(hass, after="time.am", before="time.pm")
|
|
assert condition.time(hass, after="time.pm", before="time.am")
|
|
|
|
# Trigger on PM time
|
|
with patch(
|
|
"homeassistant.helpers.condition.dt_util.now",
|
|
return_value=dt_util.now().replace(hour=18, minute=0, second=0),
|
|
):
|
|
assert condition.time(hass, after="time.pm", before="time.am")
|
|
assert not condition.time(hass, after="time.am", before="time.pm")
|
|
assert condition.time(hass, after="time.pm")
|
|
assert not condition.time(hass, before="time.pm")
|
|
|
|
# Trigger on AM time
|
|
with patch(
|
|
"homeassistant.helpers.condition.dt_util.now",
|
|
return_value=dt_util.now().replace(hour=6, minute=0, second=0),
|
|
):
|
|
assert not condition.time(hass, after="time.pm", before="time.am")
|
|
assert condition.time(hass, after="time.am", before="time.pm")
|
|
assert condition.time(hass, after="time.am")
|
|
assert not condition.time(hass, before="time.am")
|
|
|
|
assert not condition.time(hass, after="time.unknown_state")
|
|
assert not condition.time(hass, before="time.unavailable_state")
|
|
|
|
with pytest.raises(ConditionError):
|
|
condition.time(hass, after="time.not_existing")
|
|
|
|
with pytest.raises(ConditionError):
|
|
condition.time(hass, before="time.not_existing")
|
|
|
|
|
|
async def test_time_using_sensor(hass: HomeAssistant) -> None:
|
|
"""Test time conditions using sensor entities."""
|
|
hass.states.async_set(
|
|
"sensor.am",
|
|
"2021-06-03 13:00:00.000000+00:00", # 6 am local time
|
|
{ATTR_DEVICE_CLASS: SensorDeviceClass.TIMESTAMP},
|
|
)
|
|
hass.states.async_set(
|
|
"sensor.pm",
|
|
"2020-06-01 01:00:00.000000+00:00", # 6 pm local time
|
|
{ATTR_DEVICE_CLASS: SensorDeviceClass.TIMESTAMP},
|
|
)
|
|
hass.states.async_set(
|
|
"sensor.no_device_class",
|
|
"2020-06-01 01:00:00.000000+00:00",
|
|
)
|
|
hass.states.async_set(
|
|
"sensor.invalid_timestamp",
|
|
"This is not a timestamp",
|
|
{ATTR_DEVICE_CLASS: SensorDeviceClass.TIMESTAMP},
|
|
)
|
|
|
|
with patch(
|
|
"homeassistant.helpers.condition.dt_util.now",
|
|
return_value=dt_util.now().replace(hour=3),
|
|
):
|
|
assert not condition.time(hass, after="sensor.am", before="sensor.pm")
|
|
assert condition.time(hass, after="sensor.pm", before="sensor.am")
|
|
|
|
with patch(
|
|
"homeassistant.helpers.condition.dt_util.now",
|
|
return_value=dt_util.now().replace(hour=9),
|
|
):
|
|
assert condition.time(hass, after="sensor.am", before="sensor.pm")
|
|
assert not condition.time(hass, after="sensor.pm", before="sensor.am")
|
|
|
|
with patch(
|
|
"homeassistant.helpers.condition.dt_util.now",
|
|
return_value=dt_util.now().replace(hour=15),
|
|
):
|
|
assert condition.time(hass, after="sensor.am", before="sensor.pm")
|
|
assert not condition.time(hass, after="sensor.pm", before="sensor.am")
|
|
|
|
with patch(
|
|
"homeassistant.helpers.condition.dt_util.now",
|
|
return_value=dt_util.now().replace(hour=21),
|
|
):
|
|
assert not condition.time(hass, after="sensor.am", before="sensor.pm")
|
|
assert condition.time(hass, after="sensor.pm", before="sensor.am")
|
|
|
|
# Trigger on PM time
|
|
with patch(
|
|
"homeassistant.helpers.condition.dt_util.now",
|
|
return_value=dt_util.now().replace(hour=18, minute=0, second=0),
|
|
):
|
|
assert condition.time(hass, after="sensor.pm", before="sensor.am")
|
|
assert not condition.time(hass, after="sensor.am", before="sensor.pm")
|
|
assert condition.time(hass, after="sensor.pm")
|
|
assert not condition.time(hass, before="sensor.pm")
|
|
|
|
# Even though valid, the device class is missing
|
|
assert not condition.time(hass, after="sensor.no_device_class")
|
|
assert not condition.time(hass, before="sensor.no_device_class")
|
|
|
|
# Trigger on AM time
|
|
with patch(
|
|
"homeassistant.helpers.condition.dt_util.now",
|
|
return_value=dt_util.now().replace(hour=6, minute=0, second=0),
|
|
):
|
|
assert not condition.time(hass, after="sensor.pm", before="sensor.am")
|
|
assert condition.time(hass, after="sensor.am", before="sensor.pm")
|
|
assert condition.time(hass, after="sensor.am")
|
|
assert not condition.time(hass, before="sensor.am")
|
|
|
|
assert not condition.time(hass, after="sensor.invalid_timestamp")
|
|
assert not condition.time(hass, before="sensor.invalid_timestamp")
|
|
|
|
with pytest.raises(ConditionError):
|
|
condition.time(hass, after="sensor.not_existing")
|
|
|
|
with pytest.raises(ConditionError):
|
|
condition.time(hass, before="sensor.not_existing")
|
|
|
|
|
|
async def test_state_raises(hass: HomeAssistant) -> None:
|
|
"""Test that state raises ConditionError on errors."""
|
|
# No entity
|
|
with pytest.raises(ConditionError, match="no entity"):
|
|
condition.state(hass, entity=None, req_state="missing")
|
|
|
|
# Unknown entities
|
|
config = {
|
|
"condition": "state",
|
|
"entity_id": ["sensor.door_unknown", "sensor.window_unknown"],
|
|
"state": "open",
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
with pytest.raises(ConditionError, match="unknown entity.*door"):
|
|
test(hass)
|
|
with pytest.raises(ConditionError, match="unknown entity.*window"):
|
|
test(hass)
|
|
|
|
# Unknown state entity
|
|
|
|
config = {
|
|
"condition": "state",
|
|
"entity_id": "sensor.door",
|
|
"state": "input_text.missing",
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
hass.states.async_set("sensor.door", "open")
|
|
with pytest.raises(ConditionError, match="input_text.missing"):
|
|
test(hass)
|
|
|
|
|
|
async def test_state_for(hass: HomeAssistant) -> None:
|
|
"""Test state with duration."""
|
|
config = {
|
|
"condition": "and",
|
|
"conditions": [
|
|
{
|
|
"condition": "state",
|
|
"entity_id": ["sensor.temperature"],
|
|
"state": "100",
|
|
"for": {"seconds": 5},
|
|
},
|
|
],
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
hass.states.async_set("sensor.temperature", 100)
|
|
assert not test(hass)
|
|
|
|
now = dt_util.utcnow() + timedelta(seconds=5)
|
|
with freeze_time(now):
|
|
assert test(hass)
|
|
|
|
|
|
async def test_state_for_template(hass: HomeAssistant) -> None:
|
|
"""Test state with templated duration."""
|
|
config = {
|
|
"condition": "and",
|
|
"conditions": [
|
|
{
|
|
"condition": "state",
|
|
"entity_id": ["sensor.temperature"],
|
|
"state": "100",
|
|
"for": {"seconds": "{{ states('input_number.test')|int }}"},
|
|
},
|
|
],
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
hass.states.async_set("sensor.temperature", 100)
|
|
hass.states.async_set("input_number.test", 5)
|
|
assert not test(hass)
|
|
|
|
now = dt_util.utcnow() + timedelta(seconds=5)
|
|
with freeze_time(now):
|
|
assert test(hass)
|
|
|
|
|
|
@pytest.mark.parametrize("for_template", [{"{{invalid}}": 5}, {"hours": "{{ 1/0 }}"}])
|
|
async def test_state_for_invalid_template(
|
|
hass: HomeAssistant, for_template: dict[str, Any]
|
|
) -> None:
|
|
"""Test state with invalid templated duration."""
|
|
config = {
|
|
"condition": "and",
|
|
"conditions": [
|
|
{
|
|
"condition": "state",
|
|
"entity_id": ["sensor.temperature"],
|
|
"state": "100",
|
|
"for": for_template,
|
|
},
|
|
],
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
hass.states.async_set("sensor.temperature", 100)
|
|
hass.states.async_set("input_number.test", 5)
|
|
with pytest.raises(ConditionError):
|
|
assert not test(hass)
|
|
|
|
|
|
async def test_state_unknown_attribute(hass: HomeAssistant) -> None:
|
|
"""Test that state returns False on unknown attribute."""
|
|
# Unknown attribute
|
|
config = {
|
|
"condition": "state",
|
|
"entity_id": "sensor.door",
|
|
"attribute": "model",
|
|
"state": "acme",
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
hass.states.async_set("sensor.door", "open")
|
|
assert not test(hass)
|
|
assert_condition_trace(
|
|
{
|
|
"": [{"result": {"result": False}}],
|
|
"entity_id/0": [
|
|
{
|
|
"result": {
|
|
"result": False,
|
|
"message": (
|
|
"attribute 'model' of entity sensor.door does not exist"
|
|
),
|
|
}
|
|
}
|
|
],
|
|
}
|
|
)
|
|
|
|
|
|
async def test_state_multiple_entities(hass: HomeAssistant) -> None:
|
|
"""Test with multiple entities in condition."""
|
|
config = {
|
|
"condition": "and",
|
|
"conditions": [
|
|
{
|
|
"condition": "state",
|
|
"entity_id": ["sensor.temperature_1", "sensor.temperature_2"],
|
|
"state": "100",
|
|
},
|
|
],
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
hass.states.async_set("sensor.temperature_1", 100)
|
|
hass.states.async_set("sensor.temperature_2", 100)
|
|
assert test(hass)
|
|
|
|
hass.states.async_set("sensor.temperature_1", 101)
|
|
hass.states.async_set("sensor.temperature_2", 100)
|
|
assert not test(hass)
|
|
|
|
hass.states.async_set("sensor.temperature_1", 100)
|
|
hass.states.async_set("sensor.temperature_2", 101)
|
|
assert not test(hass)
|
|
|
|
|
|
async def test_state_multiple_entities_match_any(hass: HomeAssistant) -> None:
|
|
"""Test with multiple entities in condition with match any."""
|
|
config = {
|
|
"condition": "and",
|
|
"conditions": [
|
|
{
|
|
"condition": "state",
|
|
"entity_id": ["sensor.temperature_1", "sensor.temperature_2"],
|
|
"match": "any",
|
|
"state": "100",
|
|
},
|
|
],
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
hass.states.async_set("sensor.temperature_1", 100)
|
|
hass.states.async_set("sensor.temperature_2", 100)
|
|
assert test(hass)
|
|
|
|
hass.states.async_set("sensor.temperature_1", 101)
|
|
hass.states.async_set("sensor.temperature_2", 100)
|
|
assert test(hass)
|
|
|
|
hass.states.async_set("sensor.temperature_1", 100)
|
|
hass.states.async_set("sensor.temperature_2", 101)
|
|
assert test(hass)
|
|
|
|
hass.states.async_set("sensor.temperature_1", 101)
|
|
hass.states.async_set("sensor.temperature_2", 101)
|
|
assert not test(hass)
|
|
|
|
|
|
async def test_multiple_states(hass: HomeAssistant) -> None:
|
|
"""Test with multiple states in condition."""
|
|
config = {
|
|
"condition": "and",
|
|
"conditions": [
|
|
{
|
|
"alias": "State Condition",
|
|
"condition": "state",
|
|
"entity_id": "sensor.temperature",
|
|
"state": ["100", "200"],
|
|
},
|
|
],
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
hass.states.async_set("sensor.temperature", 100)
|
|
assert test(hass)
|
|
|
|
hass.states.async_set("sensor.temperature", 200)
|
|
assert test(hass)
|
|
|
|
hass.states.async_set("sensor.temperature", 42)
|
|
assert not test(hass)
|
|
|
|
|
|
async def test_state_attribute(hass: HomeAssistant) -> None:
|
|
"""Test with state attribute in condition."""
|
|
config = {
|
|
"condition": "and",
|
|
"conditions": [
|
|
{
|
|
"condition": "state",
|
|
"entity_id": "sensor.temperature",
|
|
"attribute": "attribute1",
|
|
"state": 200,
|
|
},
|
|
],
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
hass.states.async_set("sensor.temperature", 100, {"unknown_attr": 200})
|
|
assert not test(hass)
|
|
|
|
hass.states.async_set("sensor.temperature", 100, {"attribute1": 200})
|
|
assert test(hass)
|
|
|
|
hass.states.async_set("sensor.temperature", 100, {"attribute1": "200"})
|
|
assert not test(hass)
|
|
|
|
hass.states.async_set("sensor.temperature", 100, {"attribute1": 201})
|
|
assert not test(hass)
|
|
|
|
hass.states.async_set("sensor.temperature", 100, {"attribute1": None})
|
|
assert not test(hass)
|
|
|
|
|
|
async def test_state_attribute_boolean(hass: HomeAssistant) -> None:
|
|
"""Test with boolean state attribute in condition."""
|
|
config = {
|
|
"condition": "state",
|
|
"entity_id": "sensor.temperature",
|
|
"attribute": "happening",
|
|
"state": False,
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
hass.states.async_set("sensor.temperature", 100, {"happening": 200})
|
|
assert not test(hass)
|
|
|
|
hass.states.async_set("sensor.temperature", 100, {"happening": True})
|
|
assert not test(hass)
|
|
|
|
hass.states.async_set("sensor.temperature", 100, {"no_happening": 201})
|
|
assert not test(hass)
|
|
|
|
hass.states.async_set("sensor.temperature", 100, {"happening": False})
|
|
assert test(hass)
|
|
|
|
|
|
async def test_state_entity_registry_id(
|
|
hass: HomeAssistant, entity_registry: er.EntityRegistry
|
|
) -> None:
|
|
"""Test with entity specified by entity registry id."""
|
|
entry = entity_registry.async_get_or_create(
|
|
"switch", "hue", "1234", suggested_object_id="test"
|
|
)
|
|
assert entry.entity_id == "switch.test"
|
|
config = {
|
|
"condition": "state",
|
|
"entity_id": entry.id,
|
|
"state": "on",
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
hass.states.async_set("switch.test", "on")
|
|
assert test(hass)
|
|
|
|
hass.states.async_set("switch.test", "off")
|
|
assert not test(hass)
|
|
|
|
|
|
async def test_state_using_input_entities(hass: HomeAssistant) -> None:
|
|
"""Test state conditions using input_* entities."""
|
|
await async_setup_component(
|
|
hass,
|
|
"input_text",
|
|
{
|
|
"input_text": {
|
|
"hello": {"initial": "goodbye"},
|
|
}
|
|
},
|
|
)
|
|
|
|
await async_setup_component(
|
|
hass,
|
|
"input_select",
|
|
{
|
|
"input_select": {
|
|
"hello": {"options": ["cya", "goodbye", "welcome"], "initial": "cya"},
|
|
}
|
|
},
|
|
)
|
|
|
|
config = {
|
|
"condition": "and",
|
|
"conditions": [
|
|
{
|
|
"condition": "state",
|
|
"entity_id": "sensor.salut",
|
|
"state": [
|
|
"input_text.hello",
|
|
"input_select.hello",
|
|
"salut",
|
|
],
|
|
},
|
|
],
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
hass.states.async_set("sensor.salut", "goodbye")
|
|
assert test(hass)
|
|
|
|
hass.states.async_set("sensor.salut", "salut")
|
|
assert test(hass)
|
|
|
|
hass.states.async_set("sensor.salut", "hello")
|
|
assert not test(hass)
|
|
|
|
await hass.services.async_call(
|
|
"input_text",
|
|
"set_value",
|
|
{
|
|
"entity_id": "input_text.hello",
|
|
"value": "hi",
|
|
},
|
|
blocking=True,
|
|
)
|
|
assert not test(hass)
|
|
|
|
hass.states.async_set("sensor.salut", "hi")
|
|
assert test(hass)
|
|
|
|
hass.states.async_set("sensor.salut", "cya")
|
|
assert test(hass)
|
|
|
|
await hass.services.async_call(
|
|
"input_select",
|
|
"select_option",
|
|
{
|
|
"entity_id": "input_select.hello",
|
|
"option": "welcome",
|
|
},
|
|
blocking=True,
|
|
)
|
|
assert not test(hass)
|
|
|
|
hass.states.async_set("sensor.salut", "welcome")
|
|
assert test(hass)
|
|
|
|
|
|
async def test_numeric_state_known_non_matching(hass: HomeAssistant) -> None:
|
|
"""Test that numeric_state doesn't match on known non-matching states."""
|
|
hass.states.async_set("sensor.temperature", "unavailable")
|
|
config = {
|
|
"condition": "numeric_state",
|
|
"entity_id": "sensor.temperature",
|
|
"above": 0,
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
# Unavailable state
|
|
assert not test(hass)
|
|
|
|
assert_condition_trace(
|
|
{
|
|
"": [{"result": {"result": False}}],
|
|
"entity_id/0": [
|
|
{
|
|
"result": {
|
|
"result": False,
|
|
"message": (
|
|
"value 'unavailable' is non-numeric and treated as False"
|
|
),
|
|
}
|
|
}
|
|
],
|
|
}
|
|
)
|
|
|
|
# Unknown state
|
|
hass.states.async_set("sensor.temperature", "unknown")
|
|
assert not test(hass)
|
|
|
|
assert_condition_trace(
|
|
{
|
|
"": [{"result": {"result": False}}],
|
|
"entity_id/0": [
|
|
{
|
|
"result": {
|
|
"result": False,
|
|
"message": (
|
|
"value 'unknown' is non-numeric and treated as False"
|
|
),
|
|
}
|
|
}
|
|
],
|
|
}
|
|
)
|
|
|
|
|
|
async def test_numeric_state_raises(hass: HomeAssistant) -> None:
|
|
"""Test that numeric_state raises ConditionError on errors."""
|
|
# Unknown entities
|
|
config = {
|
|
"condition": "numeric_state",
|
|
"entity_id": ["sensor.temperature_unknown", "sensor.humidity_unknown"],
|
|
"above": 0,
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
with pytest.raises(ConditionError, match="unknown entity.*temperature"):
|
|
test(hass)
|
|
with pytest.raises(ConditionError, match="unknown entity.*humidity"):
|
|
test(hass)
|
|
|
|
# Template error
|
|
config = {
|
|
"condition": "numeric_state",
|
|
"entity_id": "sensor.temperature",
|
|
"value_template": "{{ 1 / 0 }}",
|
|
"above": 0,
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
hass.states.async_set("sensor.temperature", 50)
|
|
with pytest.raises(ConditionError, match="ZeroDivisionError"):
|
|
test(hass)
|
|
|
|
# Bad number
|
|
config = {
|
|
"condition": "numeric_state",
|
|
"entity_id": "sensor.temperature",
|
|
"above": 0,
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
hass.states.async_set("sensor.temperature", "fifty")
|
|
with pytest.raises(ConditionError, match="cannot be processed as a number"):
|
|
test(hass)
|
|
|
|
# Below entity missing
|
|
config = {
|
|
"condition": "numeric_state",
|
|
"entity_id": "sensor.temperature",
|
|
"below": "input_number.missing",
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
hass.states.async_set("sensor.temperature", 50)
|
|
with pytest.raises(ConditionError, match="'below' entity"):
|
|
test(hass)
|
|
|
|
# Below entity not a number
|
|
hass.states.async_set("input_number.missing", "number")
|
|
with pytest.raises(
|
|
ConditionError,
|
|
match="'below'.*input_number.missing.*cannot be processed as a number",
|
|
):
|
|
test(hass)
|
|
|
|
# Above entity missing
|
|
config = {
|
|
"condition": "numeric_state",
|
|
"entity_id": "sensor.temperature",
|
|
"above": "input_number.missing",
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
hass.states.async_set("sensor.temperature", 50)
|
|
with pytest.raises(ConditionError, match="'above' entity"):
|
|
test(hass)
|
|
|
|
# Above entity not a number
|
|
hass.states.async_set("input_number.missing", "number")
|
|
with pytest.raises(
|
|
ConditionError,
|
|
match="'above'.*input_number.missing.*cannot be processed as a number",
|
|
):
|
|
test(hass)
|
|
|
|
|
|
async def test_numeric_state_unknown_attribute(hass: HomeAssistant) -> None:
|
|
"""Test that numeric_state returns False on unknown attribute."""
|
|
# Unknown attribute
|
|
config = {
|
|
"condition": "numeric_state",
|
|
"entity_id": "sensor.temperature",
|
|
"attribute": "temperature",
|
|
"above": 0,
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
hass.states.async_set("sensor.temperature", 50)
|
|
assert not test(hass)
|
|
assert_condition_trace(
|
|
{
|
|
"": [{"result": {"result": False}}],
|
|
"entity_id/0": [
|
|
{
|
|
"result": {
|
|
"result": False,
|
|
"message": (
|
|
"attribute 'temperature' of entity sensor.temperature does"
|
|
" not exist"
|
|
),
|
|
}
|
|
}
|
|
],
|
|
}
|
|
)
|
|
|
|
|
|
async def test_numeric_state_multiple_entities(hass: HomeAssistant) -> None:
|
|
"""Test with multiple entities in condition."""
|
|
config = {
|
|
"condition": "and",
|
|
"conditions": [
|
|
{
|
|
"alias": "Numeric State Condition",
|
|
"condition": "numeric_state",
|
|
"entity_id": ["sensor.temperature_1", "sensor.temperature_2"],
|
|
"below": 50,
|
|
},
|
|
],
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
hass.states.async_set("sensor.temperature_1", 49)
|
|
hass.states.async_set("sensor.temperature_2", 49)
|
|
assert test(hass)
|
|
|
|
hass.states.async_set("sensor.temperature_1", 50)
|
|
hass.states.async_set("sensor.temperature_2", 49)
|
|
assert not test(hass)
|
|
|
|
hass.states.async_set("sensor.temperature_1", 49)
|
|
hass.states.async_set("sensor.temperature_2", 50)
|
|
assert not test(hass)
|
|
|
|
|
|
async def test_numeric_state_attribute(hass: HomeAssistant) -> None:
|
|
"""Test with numeric state attribute in condition."""
|
|
config = {
|
|
"condition": "and",
|
|
"conditions": [
|
|
{
|
|
"condition": "numeric_state",
|
|
"entity_id": "sensor.temperature",
|
|
"attribute": "attribute1",
|
|
"below": 50,
|
|
},
|
|
],
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
hass.states.async_set("sensor.temperature", 100, {"unknown_attr": 10})
|
|
assert not test(hass)
|
|
|
|
hass.states.async_set("sensor.temperature", 100, {"attribute1": 49})
|
|
assert test(hass)
|
|
|
|
hass.states.async_set("sensor.temperature", 100, {"attribute1": "49"})
|
|
assert test(hass)
|
|
|
|
hass.states.async_set("sensor.temperature", 100, {"attribute1": 51})
|
|
assert not test(hass)
|
|
|
|
hass.states.async_set("sensor.temperature", 100, {"attribute1": None})
|
|
assert not test(hass)
|
|
|
|
|
|
async def test_numeric_state_entity_registry_id(
|
|
hass: HomeAssistant, entity_registry: er.EntityRegistry
|
|
) -> None:
|
|
"""Test with entity specified by entity registry id."""
|
|
entry = entity_registry.async_get_or_create(
|
|
"sensor", "hue", "1234", suggested_object_id="test"
|
|
)
|
|
assert entry.entity_id == "sensor.test"
|
|
config = {
|
|
"condition": "numeric_state",
|
|
"entity_id": entry.id,
|
|
"above": 100,
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
hass.states.async_set("sensor.test", "110")
|
|
assert test(hass)
|
|
|
|
hass.states.async_set("sensor.test", "90")
|
|
assert not test(hass)
|
|
|
|
|
|
async def test_numeric_state_using_input_number(hass: HomeAssistant) -> None:
|
|
"""Test numeric_state conditions using input_number entities."""
|
|
hass.states.async_set("number.low", 10)
|
|
await async_setup_component(
|
|
hass,
|
|
"input_number",
|
|
{
|
|
"input_number": {
|
|
"high": {"min": 0, "max": 255, "initial": 100},
|
|
}
|
|
},
|
|
)
|
|
|
|
config = {
|
|
"condition": "and",
|
|
"conditions": [
|
|
{
|
|
"condition": "numeric_state",
|
|
"entity_id": "sensor.temperature",
|
|
"below": "input_number.high",
|
|
"above": "number.low",
|
|
},
|
|
],
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
hass.states.async_set("sensor.temperature", 42)
|
|
assert test(hass)
|
|
|
|
hass.states.async_set("sensor.temperature", 10)
|
|
assert not test(hass)
|
|
|
|
hass.states.async_set("sensor.temperature", 100)
|
|
assert not test(hass)
|
|
|
|
hass.states.async_set("input_number.high", "unknown")
|
|
assert not test(hass)
|
|
|
|
hass.states.async_set("input_number.high", "unavailable")
|
|
assert not test(hass)
|
|
|
|
await hass.services.async_call(
|
|
"input_number",
|
|
"set_value",
|
|
{
|
|
"entity_id": "input_number.high",
|
|
"value": 101,
|
|
},
|
|
blocking=True,
|
|
)
|
|
assert test(hass)
|
|
|
|
hass.states.async_set("number.low", "unknown")
|
|
assert not test(hass)
|
|
|
|
hass.states.async_set("number.low", "unavailable")
|
|
assert not test(hass)
|
|
|
|
with pytest.raises(ConditionError):
|
|
condition.async_numeric_state(
|
|
hass, entity="sensor.temperature", below="input_number.not_exist"
|
|
)
|
|
with pytest.raises(ConditionError):
|
|
condition.async_numeric_state(
|
|
hass, entity="sensor.temperature", above="input_number.not_exist"
|
|
)
|
|
|
|
|
|
@pytest.mark.usefixtures("hass")
|
|
async def test_extract_entities() -> None:
|
|
"""Test extracting entities."""
|
|
assert condition.async_extract_entities(
|
|
{
|
|
"condition": "and",
|
|
"conditions": [
|
|
{
|
|
"condition": "state",
|
|
"entity_id": "sensor.temperature",
|
|
"state": "100",
|
|
},
|
|
{
|
|
"condition": "numeric_state",
|
|
"entity_id": "sensor.temperature_2",
|
|
"below": 110,
|
|
},
|
|
{
|
|
"condition": "not",
|
|
"conditions": [
|
|
{
|
|
"condition": "state",
|
|
"entity_id": "sensor.temperature_3",
|
|
"state": "100",
|
|
},
|
|
{
|
|
"condition": "numeric_state",
|
|
"entity_id": "sensor.temperature_4",
|
|
"below": 110,
|
|
},
|
|
],
|
|
},
|
|
{
|
|
"condition": "or",
|
|
"conditions": [
|
|
{
|
|
"condition": "state",
|
|
"entity_id": "sensor.temperature_5",
|
|
"state": "100",
|
|
},
|
|
{
|
|
"condition": "numeric_state",
|
|
"entity_id": "sensor.temperature_6",
|
|
"below": 110,
|
|
},
|
|
],
|
|
},
|
|
{
|
|
"condition": "state",
|
|
"entity_id": ["sensor.temperature_7", "sensor.temperature_8"],
|
|
"state": "100",
|
|
},
|
|
{
|
|
"condition": "numeric_state",
|
|
"entity_id": ["sensor.temperature_9", "sensor.temperature_10"],
|
|
"below": 110,
|
|
},
|
|
Template("{{ is_state('light.example', 'on') }}"),
|
|
],
|
|
}
|
|
) == {
|
|
"sensor.temperature",
|
|
"sensor.temperature_2",
|
|
"sensor.temperature_3",
|
|
"sensor.temperature_4",
|
|
"sensor.temperature_5",
|
|
"sensor.temperature_6",
|
|
"sensor.temperature_7",
|
|
"sensor.temperature_8",
|
|
"sensor.temperature_9",
|
|
"sensor.temperature_10",
|
|
}
|
|
|
|
|
|
@pytest.mark.usefixtures("hass")
|
|
async def test_extract_devices() -> None:
|
|
"""Test extracting devices."""
|
|
assert condition.async_extract_devices(
|
|
{
|
|
"condition": "and",
|
|
"conditions": [
|
|
{"condition": "device", "device_id": "abcd", "domain": "light"},
|
|
{"condition": "device", "device_id": "qwer", "domain": "switch"},
|
|
{
|
|
"condition": "state",
|
|
"entity_id": "sensor.not_a_device",
|
|
"state": "100",
|
|
},
|
|
{
|
|
"condition": "not",
|
|
"conditions": [
|
|
{
|
|
"condition": "device",
|
|
"device_id": "abcd_not",
|
|
"domain": "light",
|
|
},
|
|
{
|
|
"condition": "device",
|
|
"device_id": "qwer_not",
|
|
"domain": "switch",
|
|
},
|
|
],
|
|
},
|
|
{
|
|
"condition": "or",
|
|
"conditions": [
|
|
{
|
|
"condition": "device",
|
|
"device_id": "abcd_or",
|
|
"domain": "light",
|
|
},
|
|
{
|
|
"condition": "device",
|
|
"device_id": "qwer_or",
|
|
"domain": "switch",
|
|
},
|
|
],
|
|
},
|
|
Template("{{ is_state('light.example', 'on') }}"),
|
|
],
|
|
}
|
|
) == {"abcd", "qwer", "abcd_not", "qwer_not", "abcd_or", "qwer_or"}
|
|
|
|
|
|
async def test_condition_template_error(hass: HomeAssistant) -> None:
|
|
"""Test invalid template."""
|
|
config = {"condition": "template", "value_template": "{{ undefined.state }}"}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
with pytest.raises(ConditionError, match="template"):
|
|
test(hass)
|
|
|
|
|
|
async def test_condition_template_invalid_results(hass: HomeAssistant) -> None:
|
|
"""Test template condition render false with invalid results."""
|
|
config = {"condition": "template", "value_template": "{{ 'string' }}"}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
assert not test(hass)
|
|
|
|
config = {"condition": "template", "value_template": "{{ 10.1 }}"}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
assert not test(hass)
|
|
|
|
config = {"condition": "template", "value_template": "{{ 42 }}"}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
assert not test(hass)
|
|
|
|
config = {"condition": "template", "value_template": "{{ [1, 2, 3] }}"}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
assert not test(hass)
|
|
|
|
|
|
async def test_trigger(hass: HomeAssistant) -> None:
|
|
"""Test trigger condition."""
|
|
config = {"alias": "Trigger Cond", "condition": "trigger", "id": "123456"}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
assert not test(hass)
|
|
assert not test(hass, {})
|
|
assert not test(hass, {"other_var": "123456"})
|
|
assert not test(hass, {"trigger": {"trigger_id": "123456"}})
|
|
assert test(hass, {"trigger": {"id": "123456"}})
|
|
|
|
|
|
async def test_platform_async_get_conditions(hass: HomeAssistant) -> None:
|
|
"""Test platform.async_get_conditions will be called if it exists."""
|
|
config = {CONF_DEVICE_ID: "test", CONF_DOMAIN: "test", CONF_CONDITION: "device"}
|
|
with patch(
|
|
"homeassistant.components.device_automation.condition.async_get_conditions",
|
|
AsyncMock(return_value={"device": AsyncMock()}),
|
|
) as device_automation_async_get_conditions_mock:
|
|
await condition.async_validate_condition_config(hass, config)
|
|
device_automation_async_get_conditions_mock.assert_awaited()
|
|
|
|
|
|
async def test_platform_multiple_conditions(hass: HomeAssistant) -> None:
|
|
"""Test a condition platform with multiple conditions."""
|
|
|
|
class MockCondition(condition.Condition):
|
|
"""Mock condition."""
|
|
|
|
def __init__(self, hass: HomeAssistant, config: ConfigType) -> None:
|
|
"""Initialize condition."""
|
|
|
|
@classmethod
|
|
async def async_validate_condition_config(
|
|
cls, hass: HomeAssistant, config: ConfigType
|
|
) -> ConfigType:
|
|
"""Validate config."""
|
|
return config
|
|
|
|
class MockCondition1(MockCondition):
|
|
"""Mock condition 1."""
|
|
|
|
async def async_condition_from_config(self) -> condition.ConditionCheckerType:
|
|
"""Evaluate state based on configuration."""
|
|
return lambda hass, vars: True
|
|
|
|
class MockCondition2(MockCondition):
|
|
"""Mock condition 2."""
|
|
|
|
async def async_condition_from_config(self) -> condition.ConditionCheckerType:
|
|
"""Evaluate state based on configuration."""
|
|
return lambda hass, vars: False
|
|
|
|
async def async_get_conditions(
|
|
hass: HomeAssistant,
|
|
) -> dict[str, type[condition.Condition]]:
|
|
return {
|
|
"test": MockCondition1,
|
|
"test.cond_2": MockCondition2,
|
|
}
|
|
|
|
mock_integration(hass, MockModule("test"))
|
|
mock_platform(
|
|
hass, "test.condition", Mock(async_get_conditions=async_get_conditions)
|
|
)
|
|
|
|
config_1 = {CONF_CONDITION: "test"}
|
|
config_2 = {CONF_CONDITION: "test.cond_2"}
|
|
config_3 = {CONF_CONDITION: "test.unknown_cond"}
|
|
assert await condition.async_validate_condition_config(hass, config_1) == config_1
|
|
assert await condition.async_validate_condition_config(hass, config_2) == config_2
|
|
with pytest.raises(
|
|
vol.Invalid, match="Invalid condition 'test.unknown_cond' specified"
|
|
):
|
|
await condition.async_validate_condition_config(hass, config_3)
|
|
|
|
cond_func = await condition.async_from_config(hass, config_1)
|
|
assert cond_func(hass, {}) is True
|
|
|
|
cond_func = await condition.async_from_config(hass, config_2)
|
|
assert cond_func(hass, {}) is False
|
|
|
|
with pytest.raises(KeyError):
|
|
await condition.async_from_config(hass, config_3)
|
|
|
|
|
|
@pytest.mark.parametrize("enabled_value", [True, "{{ 1 == 1 }}"])
|
|
async def test_enabled_condition(
|
|
hass: HomeAssistant, enabled_value: bool | str
|
|
) -> None:
|
|
"""Test an explicitly enabled condition."""
|
|
config = {
|
|
"enabled": enabled_value,
|
|
"condition": "state",
|
|
"entity_id": "binary_sensor.test",
|
|
"state": "on",
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
hass.states.async_set("binary_sensor.test", "on")
|
|
assert test(hass) is True
|
|
|
|
# Still passes, condition is not enabled
|
|
hass.states.async_set("binary_sensor.test", "off")
|
|
assert test(hass) is False
|
|
|
|
|
|
@pytest.mark.parametrize("enabled_value", [False, "{{ 1 == 9 }}"])
|
|
async def test_disabled_condition(
|
|
hass: HomeAssistant, enabled_value: bool | str
|
|
) -> None:
|
|
"""Test a disabled condition returns none."""
|
|
config = {
|
|
"enabled": enabled_value,
|
|
"condition": "state",
|
|
"entity_id": "binary_sensor.test",
|
|
"state": "on",
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
hass.states.async_set("binary_sensor.test", "on")
|
|
assert test(hass) is None
|
|
|
|
# Still passes, condition is not enabled
|
|
hass.states.async_set("binary_sensor.test", "off")
|
|
assert test(hass) is None
|
|
|
|
|
|
async def test_condition_enabled_template_limited(hass: HomeAssistant) -> None:
|
|
"""Test conditions enabled template raises for non-limited template uses."""
|
|
config = {
|
|
"enabled": "{{ states('sensor.limited') }}",
|
|
"condition": "state",
|
|
"entity_id": "binary_sensor.test",
|
|
"state": "on",
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
|
|
with pytest.raises(HomeAssistantError):
|
|
await condition.async_from_config(hass, config)
|
|
|
|
|
|
async def test_and_condition_with_disabled_condition(hass: HomeAssistant) -> None:
|
|
"""Test the 'and' condition with one of the conditions disabled."""
|
|
config = {
|
|
"alias": "And Condition",
|
|
"condition": "and",
|
|
"conditions": [
|
|
{
|
|
"enabled": False,
|
|
"condition": "state",
|
|
"entity_id": "sensor.temperature",
|
|
"state": "100",
|
|
},
|
|
{
|
|
"condition": "numeric_state",
|
|
"entity_id": "sensor.temperature",
|
|
"below": 110,
|
|
},
|
|
],
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
hass.states.async_set("sensor.temperature", 120)
|
|
assert not test(hass)
|
|
assert_condition_trace(
|
|
{
|
|
"": [{"result": {"result": False}}],
|
|
"conditions/0": [{"result": {"result": None}}],
|
|
"conditions/1": [{"result": {"result": False}}],
|
|
"conditions/1/entity_id/0": [
|
|
{
|
|
"result": {
|
|
"result": False,
|
|
"wanted_state_below": 110.0,
|
|
"state": 120.0,
|
|
}
|
|
}
|
|
],
|
|
}
|
|
)
|
|
|
|
hass.states.async_set("sensor.temperature", 105)
|
|
assert test(hass)
|
|
assert_condition_trace(
|
|
{
|
|
"": [{"result": {"result": True}}],
|
|
"conditions/0": [{"result": {"result": None}}],
|
|
"conditions/1": [{"result": {"result": True}}],
|
|
"conditions/1/entity_id/0": [{"result": {"result": True, "state": 105.0}}],
|
|
}
|
|
)
|
|
|
|
hass.states.async_set("sensor.temperature", 100)
|
|
assert test(hass)
|
|
assert_condition_trace(
|
|
{
|
|
"": [{"result": {"result": True}}],
|
|
"conditions/0": [{"result": {"result": None}}],
|
|
"conditions/1": [{"result": {"result": True}}],
|
|
"conditions/1/entity_id/0": [{"result": {"result": True, "state": 100.0}}],
|
|
}
|
|
)
|
|
|
|
|
|
async def test_or_condition_with_disabled_condition(hass: HomeAssistant) -> None:
|
|
"""Test the 'or' condition with one of the conditions disabled."""
|
|
config = {
|
|
"alias": "Or Condition",
|
|
"condition": "or",
|
|
"conditions": [
|
|
{
|
|
"enabled": False,
|
|
"condition": "state",
|
|
"entity_id": "sensor.temperature",
|
|
"state": "100",
|
|
},
|
|
{
|
|
"condition": "numeric_state",
|
|
"entity_id": "sensor.temperature",
|
|
"below": 110,
|
|
},
|
|
],
|
|
}
|
|
config = cv.CONDITION_SCHEMA(config)
|
|
config = await condition.async_validate_condition_config(hass, config)
|
|
test = await condition.async_from_config(hass, config)
|
|
|
|
hass.states.async_set("sensor.temperature", 120)
|
|
assert not test(hass)
|
|
assert_condition_trace(
|
|
{
|
|
"": [{"result": {"result": False}}],
|
|
"conditions/0": [{"result": {"result": None}}],
|
|
"conditions/1": [{"result": {"result": False}}],
|
|
"conditions/1/entity_id/0": [
|
|
{
|
|
"result": {
|
|
"result": False,
|
|
"state": 120.0,
|
|
"wanted_state_below": 110.0,
|
|
}
|
|
}
|
|
],
|
|
}
|
|
)
|
|
|
|
hass.states.async_set("sensor.temperature", 105)
|
|
assert test(hass)
|
|
assert_condition_trace(
|
|
{
|
|
"": [{"result": {"result": True}}],
|
|
"conditions/0": [{"result": {"result": None}}],
|
|
"conditions/1": [{"result": {"result": True}}],
|
|
"conditions/1/entity_id/0": [{"result": {"result": True, "state": 105.0}}],
|
|
}
|
|
)
|
|
|
|
hass.states.async_set("sensor.temperature", 100)
|
|
assert test(hass)
|
|
assert_condition_trace(
|
|
{
|
|
"": [{"result": {"result": True}}],
|
|
"conditions/0": [{"result": {"result": None}}],
|
|
"conditions/1": [{"result": {"result": True}}],
|
|
"conditions/1/entity_id/0": [{"result": {"result": True, "state": 100.0}}],
|
|
}
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"sun_condition_descriptions",
|
|
[
|
|
"""
|
|
sun:
|
|
fields:
|
|
after:
|
|
example: sunrise
|
|
selector:
|
|
select:
|
|
options:
|
|
- sunrise
|
|
- sunset
|
|
after_offset:
|
|
selector:
|
|
time: null
|
|
before:
|
|
example: sunrise
|
|
selector:
|
|
select:
|
|
options:
|
|
- sunrise
|
|
- sunset
|
|
before_offset:
|
|
selector:
|
|
time: null
|
|
""",
|
|
"""
|
|
.sunrise_sunset_selector: &sunrise_sunset_selector
|
|
example: sunrise
|
|
selector:
|
|
select:
|
|
options:
|
|
- sunrise
|
|
- sunset
|
|
.offset_selector: &offset_selector
|
|
selector:
|
|
time: null
|
|
sun:
|
|
fields:
|
|
after: *sunrise_sunset_selector
|
|
after_offset: *offset_selector
|
|
before: *sunrise_sunset_selector
|
|
before_offset: *offset_selector
|
|
""",
|
|
],
|
|
)
|
|
async def test_async_get_all_descriptions(
|
|
hass: HomeAssistant, sun_condition_descriptions: str
|
|
) -> None:
|
|
"""Test async_get_all_descriptions."""
|
|
device_automation_condition_descriptions = """
|
|
device: {}
|
|
"""
|
|
|
|
assert await async_setup_component(hass, DOMAIN_SUN, {})
|
|
assert await async_setup_component(hass, DOMAIN_SYSTEM_HEALTH, {})
|
|
await hass.async_block_till_done()
|
|
|
|
def _load_yaml(fname, secrets=None):
|
|
if fname.endswith("device_automation/conditions.yaml"):
|
|
condition_descriptions = device_automation_condition_descriptions
|
|
elif fname.endswith("sun/conditions.yaml"):
|
|
condition_descriptions = sun_condition_descriptions
|
|
with io.StringIO(condition_descriptions) as file:
|
|
return parse_yaml(file)
|
|
|
|
with (
|
|
patch(
|
|
"homeassistant.helpers.condition._load_conditions_files",
|
|
side_effect=condition._load_conditions_files,
|
|
) as proxy_load_conditions_files,
|
|
patch(
|
|
"annotatedyaml.loader.load_yaml",
|
|
side_effect=_load_yaml,
|
|
),
|
|
patch.object(Integration, "has_conditions", return_value=True),
|
|
):
|
|
descriptions = await condition.async_get_all_descriptions(hass)
|
|
|
|
# Test we only load conditions.yaml for integrations with conditions,
|
|
# system_health has no conditions
|
|
assert proxy_load_conditions_files.mock_calls[0][1][1] == unordered(
|
|
[
|
|
await async_get_integration(hass, DOMAIN_SUN),
|
|
]
|
|
)
|
|
|
|
# system_health does not have conditions and should not be in descriptions
|
|
assert descriptions == {
|
|
DOMAIN_SUN: {
|
|
"fields": {
|
|
"after": {
|
|
"example": "sunrise",
|
|
"selector": {"select": {"options": ["sunrise", "sunset"]}},
|
|
},
|
|
"after_offset": {"selector": {"time": None}},
|
|
"before": {
|
|
"example": "sunrise",
|
|
"selector": {"select": {"options": ["sunrise", "sunset"]}},
|
|
},
|
|
"before_offset": {"selector": {"time": None}},
|
|
}
|
|
}
|
|
}
|
|
|
|
# Verify the cache returns the same object
|
|
assert await condition.async_get_all_descriptions(hass) is descriptions
|
|
|
|
# Load the device_automation integration and check a new cache object is created
|
|
assert await async_setup_component(hass, DOMAIN_DEVICE_AUTOMATION, {})
|
|
await hass.async_block_till_done()
|
|
|
|
with (
|
|
patch(
|
|
"annotatedyaml.loader.load_yaml",
|
|
side_effect=_load_yaml,
|
|
),
|
|
patch.object(Integration, "has_conditions", return_value=True),
|
|
):
|
|
new_descriptions = await condition.async_get_all_descriptions(hass)
|
|
assert new_descriptions is not descriptions
|
|
assert new_descriptions == {
|
|
"device": {
|
|
"fields": {},
|
|
},
|
|
DOMAIN_SUN: {
|
|
"fields": {
|
|
"after": {
|
|
"example": "sunrise",
|
|
"selector": {"select": {"options": ["sunrise", "sunset"]}},
|
|
},
|
|
"after_offset": {"selector": {"time": None}},
|
|
"before": {
|
|
"example": "sunrise",
|
|
"selector": {"select": {"options": ["sunrise", "sunset"]}},
|
|
},
|
|
"before_offset": {"selector": {"time": None}},
|
|
}
|
|
},
|
|
}
|
|
|
|
# Verify the cache returns the same object
|
|
assert await condition.async_get_all_descriptions(hass) is new_descriptions
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("yaml_error", "expected_message"),
|
|
[
|
|
(
|
|
FileNotFoundError("Blah"),
|
|
"Unable to find conditions.yaml for the sun integration",
|
|
),
|
|
(
|
|
HomeAssistantError("Test error"),
|
|
"Unable to parse conditions.yaml for the sun integration: Test error",
|
|
),
|
|
],
|
|
)
|
|
async def test_async_get_all_descriptions_with_yaml_error(
|
|
hass: HomeAssistant,
|
|
caplog: pytest.LogCaptureFixture,
|
|
yaml_error: Exception,
|
|
expected_message: str,
|
|
) -> None:
|
|
"""Test async_get_all_descriptions."""
|
|
assert await async_setup_component(hass, DOMAIN_SUN, {})
|
|
await hass.async_block_till_done()
|
|
|
|
def _load_yaml_dict(fname, secrets=None):
|
|
raise yaml_error
|
|
|
|
with (
|
|
patch(
|
|
"homeassistant.helpers.condition.load_yaml_dict",
|
|
side_effect=_load_yaml_dict,
|
|
),
|
|
patch.object(Integration, "has_conditions", return_value=True),
|
|
):
|
|
descriptions = await condition.async_get_all_descriptions(hass)
|
|
|
|
assert descriptions == {DOMAIN_SUN: None}
|
|
|
|
assert expected_message in caplog.text
|
|
|
|
|
|
async def test_async_get_all_descriptions_with_bad_description(
|
|
hass: HomeAssistant,
|
|
caplog: pytest.LogCaptureFixture,
|
|
) -> None:
|
|
"""Test async_get_all_descriptions."""
|
|
sun_service_descriptions = """
|
|
sun:
|
|
fields: not_a_dict
|
|
"""
|
|
|
|
assert await async_setup_component(hass, DOMAIN_SUN, {})
|
|
await hass.async_block_till_done()
|
|
|
|
def _load_yaml(fname, secrets=None):
|
|
with io.StringIO(sun_service_descriptions) as file:
|
|
return parse_yaml(file)
|
|
|
|
with (
|
|
patch(
|
|
"annotatedyaml.loader.load_yaml",
|
|
side_effect=_load_yaml,
|
|
),
|
|
patch.object(Integration, "has_conditions", return_value=True),
|
|
):
|
|
descriptions = await condition.async_get_all_descriptions(hass)
|
|
|
|
assert descriptions == {DOMAIN_SUN: None}
|
|
|
|
assert (
|
|
"Unable to parse conditions.yaml for the sun integration: "
|
|
"expected a dictionary for dictionary value @ data['sun']['fields']"
|
|
) in caplog.text
|
|
|
|
|
|
async def test_invalid_condition_platform(
|
|
hass: HomeAssistant,
|
|
caplog: pytest.LogCaptureFixture,
|
|
) -> None:
|
|
"""Test invalid condition platform."""
|
|
mock_integration(hass, MockModule("test", async_setup=AsyncMock(return_value=True)))
|
|
mock_platform(hass, "test.condition", MockPlatform())
|
|
|
|
await async_setup_component(hass, "test", {})
|
|
|
|
assert (
|
|
"Integration test does not provide condition support, skipping" in caplog.text
|
|
)
|
|
|
|
|
|
@patch("annotatedyaml.loader.load_yaml")
|
|
@patch.object(Integration, "has_conditions", return_value=True)
|
|
async def test_subscribe_conditions(
|
|
mock_has_conditions: Mock,
|
|
mock_load_yaml: Mock,
|
|
hass: HomeAssistant,
|
|
caplog: pytest.LogCaptureFixture,
|
|
) -> None:
|
|
"""Test condition.async_subscribe_platform_events."""
|
|
sun_condition_descriptions = """
|
|
sun: {}
|
|
"""
|
|
|
|
def _load_yaml(fname, secrets=None):
|
|
if fname.endswith("sun/conditions.yaml"):
|
|
condition_descriptions = sun_condition_descriptions
|
|
else:
|
|
raise FileNotFoundError
|
|
with io.StringIO(condition_descriptions) as file:
|
|
return parse_yaml(file)
|
|
|
|
mock_load_yaml.side_effect = _load_yaml
|
|
|
|
async def broken_subscriber(_):
|
|
"""Simulate a broken subscriber."""
|
|
raise Exception("Boom!") # noqa: TRY002
|
|
|
|
condition_events = []
|
|
|
|
async def good_subscriber(new_conditions: set[str]):
|
|
"""Simulate a working subscriber."""
|
|
condition_events.append(new_conditions)
|
|
|
|
condition.async_subscribe_platform_events(hass, broken_subscriber)
|
|
condition.async_subscribe_platform_events(hass, good_subscriber)
|
|
|
|
assert await async_setup_component(hass, "sun", {})
|
|
|
|
assert condition_events == [{"sun"}]
|
|
assert "Error while notifying condition platform listener" in caplog.text
|