Add battery conditions (#165208)

Co-authored-by: Erik Montnemery <erik@montnemery.com>
This commit is contained in:
Ariel Ebersberger
2026-03-23 22:18:48 +01:00
committed by GitHub
co-authored by Erik Montnemery
parent 66b5a3755c
commit 03c672a4f3
19 changed files with 1049 additions and 45 deletions
Generated
+2
View File
@@ -214,6 +214,8 @@ build.json @home-assistant/supervisor
/tests/components/balboa/ @garbled1 @natekspencer
/homeassistant/components/bang_olufsen/ @mj23000
/tests/components/bang_olufsen/ @mj23000
/homeassistant/components/battery/ @home-assistant/core
/tests/components/battery/ @home-assistant/core
/homeassistant/components/bayesian/ @HarvsG
/tests/components/bayesian/ @HarvsG
/homeassistant/components/beewi_smartclim/ @alemuro
+1
View File
@@ -242,6 +242,7 @@ DEFAULT_INTEGRATIONS = {
#
# Integrations providing triggers and conditions for base platforms:
"air_quality",
"battery",
"door",
"garage_door",
"gate",
@@ -120,6 +120,7 @@ NEW_TRIGGERS_CONDITIONS_FEATURE_FLAG = "new_triggers_conditions"
_EXPERIMENTAL_CONDITION_PLATFORMS = {
"alarm_control_panel",
"assist_satellite",
"battery",
"climate",
"cover",
"device_tracker",
@@ -0,0 +1,17 @@
"""Integration for battery conditions."""
from __future__ import annotations
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.typing import ConfigType
DOMAIN = "battery"
CONFIG_SCHEMA = cv.empty_config_schema(DOMAIN)
__all__ = []
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the component."""
return True
@@ -0,0 +1,48 @@
"""Provides conditions for batteries."""
from __future__ import annotations
from homeassistant.components.binary_sensor import (
DOMAIN as BINARY_SENSOR_DOMAIN,
BinarySensorDeviceClass,
)
from homeassistant.components.number import DOMAIN as NUMBER_DOMAIN, NumberDeviceClass
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN, SensorDeviceClass
from homeassistant.const import PERCENTAGE, STATE_OFF, STATE_ON
from homeassistant.core import HomeAssistant
from homeassistant.helpers.automation import DomainSpec
from homeassistant.helpers.condition import (
Condition,
make_entity_numerical_condition,
make_entity_state_condition,
)
BATTERY_DOMAIN_SPECS = {
BINARY_SENSOR_DOMAIN: DomainSpec(device_class=BinarySensorDeviceClass.BATTERY)
}
BATTERY_CHARGING_DOMAIN_SPECS = {
BINARY_SENSOR_DOMAIN: DomainSpec(
device_class=BinarySensorDeviceClass.BATTERY_CHARGING
)
}
BATTERY_PERCENTAGE_DOMAIN_SPECS = {
SENSOR_DOMAIN: DomainSpec(device_class=SensorDeviceClass.BATTERY),
NUMBER_DOMAIN: DomainSpec(device_class=NumberDeviceClass.BATTERY),
}
CONDITIONS: dict[str, type[Condition]] = {
"is_low": make_entity_state_condition(BATTERY_DOMAIN_SPECS, STATE_ON),
"is_not_low": make_entity_state_condition(BATTERY_DOMAIN_SPECS, STATE_OFF),
"is_charging": make_entity_state_condition(BATTERY_CHARGING_DOMAIN_SPECS, STATE_ON),
"is_not_charging": make_entity_state_condition(
BATTERY_CHARGING_DOMAIN_SPECS, STATE_OFF
),
"percentage": make_entity_numerical_condition(
BATTERY_PERCENTAGE_DOMAIN_SPECS, PERCENTAGE
),
}
async def async_get_conditions(hass: HomeAssistant) -> dict[str, type[Condition]]:
"""Return the conditions for batteries."""
return CONDITIONS
@@ -0,0 +1,66 @@
.condition_common: &condition_common
target: &target_battery_binary_sensor
entity:
- domain: binary_sensor
device_class: battery
fields:
behavior: &condition_behavior
required: true
default: any
selector:
select:
translation_key: condition_behavior
options:
- all
- any
.number_or_entity: &number_or_entity
required: false
selector:
choose:
choices:
number:
selector:
number:
unit_of_measurement: "%"
entity:
selector:
entity:
filter:
domain:
- input_number
- number
- sensor
translation_key: number_or_entity
is_low: *condition_common
is_not_low: *condition_common
is_charging:
target:
entity:
- domain: binary_sensor
device_class: battery_charging
fields:
behavior: *condition_behavior
is_not_charging:
target:
entity:
- domain: binary_sensor
device_class: battery_charging
fields:
behavior: *condition_behavior
percentage:
target:
entity:
- domain: sensor
device_class: battery
- domain: number
device_class: battery
fields:
behavior: *condition_behavior
above: *number_or_entity
below: *number_or_entity
@@ -0,0 +1,19 @@
{
"conditions": {
"is_charging": {
"condition": "mdi:battery-charging"
},
"is_low": {
"condition": "mdi:battery-alert"
},
"is_not_charging": {
"condition": "mdi:battery"
},
"is_not_low": {
"condition": "mdi:battery"
},
"percentage": {
"condition": "mdi:battery-unknown"
}
}
}
@@ -0,0 +1,8 @@
{
"domain": "battery",
"name": "Battery",
"codeowners": ["@home-assistant/core"],
"documentation": "https://www.home-assistant.io/integrations/battery",
"integration_type": "system",
"quality_scale": "internal"
}
@@ -0,0 +1,81 @@
{
"common": {
"condition_behavior_description": "How the state should match on the targeted batteries.",
"condition_behavior_name": "Behavior"
},
"conditions": {
"is_charging": {
"description": "Tests if one or more batteries are charging.",
"fields": {
"behavior": {
"description": "[%key:component::battery::common::condition_behavior_description%]",
"name": "[%key:component::battery::common::condition_behavior_name%]"
}
},
"name": "Battery is charging"
},
"is_low": {
"description": "Tests if one or more batteries are low.",
"fields": {
"behavior": {
"description": "[%key:component::battery::common::condition_behavior_description%]",
"name": "[%key:component::battery::common::condition_behavior_name%]"
}
},
"name": "Battery is low"
},
"is_not_charging": {
"description": "Tests if one or more batteries are not charging.",
"fields": {
"behavior": {
"description": "[%key:component::battery::common::condition_behavior_description%]",
"name": "[%key:component::battery::common::condition_behavior_name%]"
}
},
"name": "Battery is not charging"
},
"is_not_low": {
"description": "Tests if one or more batteries are not low.",
"fields": {
"behavior": {
"description": "[%key:component::battery::common::condition_behavior_description%]",
"name": "[%key:component::battery::common::condition_behavior_name%]"
}
},
"name": "Battery is not low"
},
"percentage": {
"description": "Tests the percentage of one or more batteries.",
"fields": {
"above": {
"description": "Require the percentage to be above this value.",
"name": "Above"
},
"behavior": {
"description": "[%key:component::battery::common::condition_behavior_description%]",
"name": "[%key:component::battery::common::condition_behavior_name%]"
},
"below": {
"description": "Require the percentage to be below this value.",
"name": "Below"
}
},
"name": "Battery percentage"
}
},
"selector": {
"condition_behavior": {
"options": {
"all": "All",
"any": "Any"
}
},
"number_or_entity": {
"choices": {
"entity": "Entity",
"number": "Number"
}
}
},
"title": "Battery"
}
+23
View File
@@ -10,6 +10,7 @@ import voluptuous as vol
from homeassistant.const import CONF_OPTIONS
from homeassistant.core import HomeAssistant, split_entity_id
from . import config_validation as cv
from .entity import get_device_class_or_undefined
from .typing import ConfigType
@@ -140,3 +141,25 @@ def move_options_fields_to_top_level(
new_config.update(options)
return new_config
_NUMBER_OR_ENTITY_CHOOSE_SCHEMA = vol.Schema(
{
vol.Required("active_choice"): vol.In(["number", "entity"]),
vol.Optional("entity"): cv.entity_id,
vol.Optional("number"): vol.Coerce(float),
}
)
def _validate_number_or_entity(value: dict | float | str) -> float | str:
"""Validate number or entity selector result."""
if isinstance(value, dict):
_NUMBER_OR_ENTITY_CHOOSE_SCHEMA(value)
return value[value["active_choice"]] # type: ignore[no-any-return]
return value
number_or_entity = vol.All(
_validate_number_or_entity, vol.Any(vol.Coerce(float), cv.entity_id)
)
+118 -7
View File
@@ -30,6 +30,7 @@ import voluptuous as vol
from homeassistant.const import (
ATTR_DEVICE_CLASS,
ATTR_UNIT_OF_MEASUREMENT,
CONF_ABOVE,
CONF_AFTER,
CONF_ATTRIBUTE,
@@ -81,6 +82,7 @@ from .automation import (
get_absolute_description_key,
get_relative_description_key,
move_options_fields_to_top_level,
number_or_entity,
)
from .integration_platform import async_process_integration_platforms
from .selector import TargetSelector
@@ -96,7 +98,7 @@ from .trace import (
trace_stack_push,
trace_stack_top,
)
from .typing import ConfigType, TemplateVarsType
from .typing import UNDEFINED, ConfigType, TemplateVarsType, UndefinedType
ASYNC_FROM_CONFIG_FORMAT = "async_{}_from_config"
FROM_CONFIG_FORMAT = "{}_from_config"
@@ -321,15 +323,14 @@ ATTR_BEHAVIOR: Final = "behavior"
BEHAVIOR_ANY: Final = "any"
BEHAVIOR_ALL: Final = "all"
STATE_CONDITION_OPTIONS_SCHEMA: dict[vol.Marker, Any] = {
vol.Required(ATTR_BEHAVIOR, default=BEHAVIOR_ANY): vol.In(
[BEHAVIOR_ANY, BEHAVIOR_ALL]
),
}
ENTITY_STATE_CONDITION_SCHEMA_ANY_ALL = vol.Schema(
{
vol.Required(CONF_TARGET): cv.TARGET_FIELDS,
vol.Required(CONF_OPTIONS): STATE_CONDITION_OPTIONS_SCHEMA,
vol.Required(CONF_OPTIONS): {
vol.Required(ATTR_BEHAVIOR, default=BEHAVIOR_ANY): vol.In(
[BEHAVIOR_ANY, BEHAVIOR_ALL]
),
},
}
)
@@ -454,6 +455,116 @@ def make_entity_state_condition(
return CustomCondition
def _validate_above_below(config: dict[str, Any]) -> dict[str, Any]:
"""Validate that above < below when both are set."""
above = config.get(CONF_ABOVE)
below = config.get(CONF_BELOW)
if above is None or below is None:
return config
if isinstance(above, str) or isinstance(below, str):
return config
if above >= below:
raise vol.Invalid(
f"A value can never be above {above} and below {below} at the same"
" time. You probably want two different conditions."
)
return config
NUMERICAL_CONDITION_SCHEMA = vol.Schema(
{
vol.Required(CONF_TARGET): cv.TARGET_FIELDS,
vol.Required(CONF_OPTIONS): vol.All(
{
vol.Required(ATTR_BEHAVIOR, default=BEHAVIOR_ANY): vol.In(
[BEHAVIOR_ANY, BEHAVIOR_ALL]
),
vol.Optional(CONF_ABOVE): number_or_entity,
vol.Optional(CONF_BELOW): number_or_entity,
},
cv.has_at_least_one_key(CONF_ABOVE, CONF_BELOW),
_validate_above_below,
),
}
)
class EntityNumericalConditionBase(EntityConditionBase):
"""Condition for numerical state comparisons with above/below thresholds."""
_schema = NUMERICAL_CONDITION_SCHEMA
_valid_unit: str | None | UndefinedType = UNDEFINED
def __init__(self, hass: HomeAssistant, config: ConditionConfig) -> None:
"""Initialize the numerical condition."""
super().__init__(hass, config)
if TYPE_CHECKING:
assert config.options is not None
self._above: float | str | None = config.options.get(CONF_ABOVE)
self._below: float | str | None = config.options.get(CONF_BELOW)
def _is_valid_unit(self, unit: str | None) -> bool:
"""Check if the given unit is valid for this condition."""
if isinstance(self._valid_unit, UndefinedType):
return True
return unit == self._valid_unit
def _get_numerical_value(self, entity_or_float: float | str) -> float | None:
"""Get numerical value from float or entity state."""
if isinstance(entity_or_float, str):
if not (ref_state := self._hass.states.get(entity_or_float)):
return None
if not self._is_valid_unit(
ref_state.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
):
return None
try:
return float(ref_state.state)
except TypeError, ValueError:
return None
return entity_or_float
def is_valid_state(self, entity_state: State) -> bool:
"""Check if the state is within the specified range."""
if not self._is_valid_unit(
entity_state.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
):
return False
try:
value = float(self._get_tracked_value(entity_state))
except TypeError, ValueError:
return False
if self._above is not None:
if (above := self._get_numerical_value(self._above)) is None:
return False
if value <= above:
return False
if self._below is not None:
if (below := self._get_numerical_value(self._below)) is None:
return False
if value >= below:
return False
return True
def make_entity_numerical_condition(
domain_specs: Mapping[str, DomainSpec] | str,
valid_unit: str | None | UndefinedType = UNDEFINED,
) -> type[EntityNumericalConditionBase]:
"""Create a condition for numerical state comparisons."""
specs = _normalize_domain_specs(domain_specs)
class CustomCondition(EntityNumericalConditionBase):
"""Condition for numerical state."""
_domain_specs = specs
_valid_unit = valid_unit
return CustomCondition
class ConditionProtocol(Protocol):
"""Define the format of condition modules."""
+9 -29
View File
@@ -76,6 +76,7 @@ from .automation import (
get_absolute_description_key,
get_relative_description_key,
move_options_fields_to_top_level,
number_or_entity,
)
from .integration_platform import async_process_integration_platforms
from .selector import TargetSelector
@@ -565,33 +566,12 @@ def _validate_unit_set_if_range_numerical[_T: dict[str, Any]](
return _validate_unit_set_if_range_numerical_impl
_NUMBER_OR_ENTITY_CHOOSE_SCHEMA = vol.Schema(
{
vol.Required("active_choice"): vol.In(["number", "entity"]),
vol.Optional("entity"): cv.entity_id,
vol.Optional("number"): vol.Coerce(float),
}
)
def _validate_number_or_entity(value: dict | float | str) -> float | str:
"""Validate number or entity selector result."""
if isinstance(value, dict):
_NUMBER_OR_ENTITY_CHOOSE_SCHEMA(value)
return value[value["active_choice"]] # type: ignore[no-any-return]
return value
_number_or_entity = vol.All(
_validate_number_or_entity, vol.Any(vol.Coerce(float), cv.entity_id)
)
NUMERICAL_ATTRIBUTE_CHANGED_TRIGGER_SCHEMA = ENTITY_STATE_TRIGGER_SCHEMA.extend(
{
vol.Required(CONF_OPTIONS, default={}): vol.All(
{
vol.Optional(CONF_ABOVE): _number_or_entity,
vol.Optional(CONF_BELOW): _number_or_entity,
vol.Optional(CONF_ABOVE): number_or_entity,
vol.Optional(CONF_BELOW): number_or_entity,
},
_validate_range(CONF_ABOVE, CONF_BELOW),
)
@@ -771,8 +751,8 @@ def make_numerical_state_changed_with_unit_schema(
{
vol.Required(CONF_OPTIONS, default={}): vol.All(
{
vol.Optional(CONF_ABOVE): _number_or_entity,
vol.Optional(CONF_BELOW): _number_or_entity,
vol.Optional(CONF_ABOVE): number_or_entity,
vol.Optional(CONF_BELOW): number_or_entity,
vol.Optional(CONF_UNIT): vol.In(unit_converter.VALID_UNITS),
},
_validate_range(CONF_ABOVE, CONF_BELOW),
@@ -835,8 +815,8 @@ NUMERICAL_ATTRIBUTE_CROSSED_THRESHOLD_SCHEMA = ENTITY_STATE_TRIGGER_SCHEMA.exten
vol.Required(ATTR_BEHAVIOR, default=BEHAVIOR_ANY): vol.In(
[BEHAVIOR_FIRST, BEHAVIOR_LAST, BEHAVIOR_ANY]
),
vol.Optional(CONF_LOWER_LIMIT): _number_or_entity,
vol.Optional(CONF_UPPER_LIMIT): _number_or_entity,
vol.Optional(CONF_LOWER_LIMIT): number_or_entity,
vol.Optional(CONF_UPPER_LIMIT): number_or_entity,
vol.Required(CONF_THRESHOLD_TYPE): vol.Coerce(ThresholdType),
},
_validate_range(CONF_LOWER_LIMIT, CONF_UPPER_LIMIT),
@@ -920,8 +900,8 @@ def make_numerical_state_crossed_threshold_with_unit_schema(
vol.Required(ATTR_BEHAVIOR, default=BEHAVIOR_ANY): vol.In(
[BEHAVIOR_FIRST, BEHAVIOR_LAST, BEHAVIOR_ANY]
),
vol.Optional(CONF_LOWER_LIMIT): _number_or_entity,
vol.Optional(CONF_UPPER_LIMIT): _number_or_entity,
vol.Optional(CONF_LOWER_LIMIT): number_or_entity,
vol.Optional(CONF_UPPER_LIMIT): number_or_entity,
vol.Required(CONF_THRESHOLD_TYPE): vol.Coerce(ThresholdType),
vol.Optional(CONF_UNIT): vol.In(unit_converter.VALID_UNITS),
},
+1
View File
@@ -61,6 +61,7 @@ NO_IOT_CLASS = [
"application_credentials",
"auth",
"automation",
"battery",
"blueprint",
"brands",
"color_extractor",
+1
View File
@@ -2094,6 +2094,7 @@ NO_QUALITY_SCALE = [
"application_credentials",
"auth",
"automation",
"battery",
"blueprint",
"brands",
"config",
+1
View File
@@ -0,0 +1 @@
"""Tests for the battery integration."""
+249
View File
@@ -0,0 +1,249 @@
"""Test battery conditions."""
from typing import Any
import pytest
from homeassistant.const import (
ATTR_DEVICE_CLASS,
ATTR_UNIT_OF_MEASUREMENT,
STATE_OFF,
STATE_ON,
)
from homeassistant.core import HomeAssistant
from tests.components.common import (
ConditionStateDescription,
assert_condition_behavior_all,
assert_condition_behavior_any,
assert_condition_gated_by_labs_flag,
parametrize_condition_states_all,
parametrize_condition_states_any,
parametrize_numerical_condition_above_below_all,
parametrize_numerical_condition_above_below_any,
parametrize_target_entities,
target_entities,
)
_BATTERY_UNIT_ATTRS = {ATTR_UNIT_OF_MEASUREMENT: "%"}
@pytest.fixture
async def target_binary_sensors(hass: HomeAssistant) -> dict[str, list[str]]:
"""Create multiple binary sensor entities associated with different targets."""
return await target_entities(hass, "binary_sensor")
@pytest.fixture
async def target_sensors(hass: HomeAssistant) -> dict[str, list[str]]:
"""Create multiple sensor entities associated with different targets."""
return await target_entities(hass, "sensor")
@pytest.fixture
async def target_numbers(hass: HomeAssistant) -> dict[str, list[str]]:
"""Create multiple number entities associated with different targets."""
return await target_entities(hass, "number")
@pytest.mark.parametrize(
"condition",
[
"battery.is_low",
"battery.is_not_low",
"battery.is_charging",
"battery.is_not_charging",
"battery.percentage",
],
)
async def test_battery_conditions_gated_by_labs_flag(
hass: HomeAssistant, caplog: pytest.LogCaptureFixture, condition: str
) -> None:
"""Test the battery conditions are gated by the labs flag."""
await assert_condition_gated_by_labs_flag(hass, caplog, condition)
@pytest.mark.usefixtures("enable_labs_preview_features")
@pytest.mark.parametrize(
("condition_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("binary_sensor"),
)
@pytest.mark.parametrize(
("condition", "condition_options", "states"),
[
*parametrize_condition_states_any(
condition="battery.is_low",
target_states=[STATE_ON],
other_states=[STATE_OFF],
required_filter_attributes={ATTR_DEVICE_CLASS: "battery"},
),
*parametrize_condition_states_any(
condition="battery.is_not_low",
target_states=[STATE_OFF],
other_states=[STATE_ON],
required_filter_attributes={ATTR_DEVICE_CLASS: "battery"},
),
*parametrize_condition_states_any(
condition="battery.is_charging",
target_states=[STATE_ON],
other_states=[STATE_OFF],
required_filter_attributes={ATTR_DEVICE_CLASS: "battery_charging"},
),
*parametrize_condition_states_any(
condition="battery.is_not_charging",
target_states=[STATE_OFF],
other_states=[STATE_ON],
required_filter_attributes={ATTR_DEVICE_CLASS: "battery_charging"},
),
],
)
async def test_battery_binary_condition_behavior_any(
hass: HomeAssistant,
target_binary_sensors: dict[str, list[str]],
condition_target_config: dict,
entity_id: str,
entities_in_target: int,
condition: str,
condition_options: dict[str, Any],
states: list[ConditionStateDescription],
) -> None:
"""Test the battery binary conditions with 'any' behavior."""
await assert_condition_behavior_any(
hass,
target_entities=target_binary_sensors,
condition_target_config=condition_target_config,
entity_id=entity_id,
entities_in_target=entities_in_target,
condition=condition,
condition_options=condition_options,
states=states,
)
@pytest.mark.usefixtures("enable_labs_preview_features")
@pytest.mark.parametrize(
("condition_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("binary_sensor"),
)
@pytest.mark.parametrize(
("condition", "condition_options", "states"),
[
*parametrize_condition_states_all(
condition="battery.is_low",
target_states=[STATE_ON],
other_states=[STATE_OFF],
required_filter_attributes={ATTR_DEVICE_CLASS: "battery"},
),
*parametrize_condition_states_all(
condition="battery.is_not_low",
target_states=[STATE_OFF],
other_states=[STATE_ON],
required_filter_attributes={ATTR_DEVICE_CLASS: "battery"},
),
*parametrize_condition_states_all(
condition="battery.is_charging",
target_states=[STATE_ON],
other_states=[STATE_OFF],
required_filter_attributes={ATTR_DEVICE_CLASS: "battery_charging"},
),
*parametrize_condition_states_all(
condition="battery.is_not_charging",
target_states=[STATE_OFF],
other_states=[STATE_ON],
required_filter_attributes={ATTR_DEVICE_CLASS: "battery_charging"},
),
],
)
async def test_battery_binary_condition_behavior_all(
hass: HomeAssistant,
target_binary_sensors: dict[str, list[str]],
condition_target_config: dict,
entity_id: str,
entities_in_target: int,
condition: str,
condition_options: dict[str, Any],
states: list[ConditionStateDescription],
) -> None:
"""Test the battery binary conditions with 'all' behavior."""
await assert_condition_behavior_all(
hass,
target_entities=target_binary_sensors,
condition_target_config=condition_target_config,
entity_id=entity_id,
entities_in_target=entities_in_target,
condition=condition,
condition_options=condition_options,
states=states,
)
@pytest.mark.usefixtures("enable_labs_preview_features")
@pytest.mark.parametrize(
("condition_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("sensor"),
)
@pytest.mark.parametrize(
("condition", "condition_options", "states"),
parametrize_numerical_condition_above_below_any(
"battery.percentage",
device_class="battery",
unit_attributes=_BATTERY_UNIT_ATTRS,
),
)
async def test_battery_percentage_condition_behavior_any(
hass: HomeAssistant,
target_sensors: dict[str, list[str]],
condition_target_config: dict,
entity_id: str,
entities_in_target: int,
condition: str,
condition_options: dict[str, Any],
states: list[ConditionStateDescription],
) -> None:
"""Test the battery percentage condition with 'any' behavior."""
await assert_condition_behavior_any(
hass,
target_entities=target_sensors,
condition_target_config=condition_target_config,
entity_id=entity_id,
entities_in_target=entities_in_target,
condition=condition,
condition_options=condition_options,
states=states,
)
@pytest.mark.usefixtures("enable_labs_preview_features")
@pytest.mark.parametrize(
("condition_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("sensor"),
)
@pytest.mark.parametrize(
("condition", "condition_options", "states"),
parametrize_numerical_condition_above_below_all(
"battery.percentage",
device_class="battery",
unit_attributes=_BATTERY_UNIT_ATTRS,
),
)
async def test_battery_percentage_condition_behavior_all(
hass: HomeAssistant,
target_sensors: dict[str, list[str]],
condition_target_config: dict,
entity_id: str,
entities_in_target: int,
condition: str,
condition_options: dict[str, Any],
states: list[ConditionStateDescription],
) -> None:
"""Test the battery percentage condition with 'all' behavior."""
await assert_condition_behavior_all(
hass,
target_entities=target_sensors,
condition_target_config=condition_target_config,
entity_id=entity_id,
entities_in_target=entities_in_target,
condition=condition,
condition_options=condition_options,
states=states,
)
+141 -7
View File
@@ -948,7 +948,7 @@ async def assert_condition_behavior_any(
set_or_remove_state(hass, eid, states[0]["excluded_state"])
await hass.async_block_till_done()
condition = await create_target_condition(
cond = await create_target_condition(
hass,
condition=condition,
target=condition_target_config,
@@ -965,18 +965,18 @@ async def assert_condition_behavior_any(
for excluded_entity_id in excluded_entity_ids:
set_or_remove_state(hass, excluded_entity_id, excluded_state)
await hass.async_block_till_done()
assert condition(hass) is False
assert cond(hass) is False
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert condition(hass) == state["condition_true"]
assert cond(hass) == state["condition_true"]
# Set other included entities to the included state to verify that
# they don't change the condition evaluation
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
assert condition(hass) == state["condition_true"]
assert cond(hass) == state["condition_true"]
async def assert_condition_behavior_all(
@@ -1001,7 +1001,7 @@ async def assert_condition_behavior_all(
set_or_remove_state(hass, eid, states[0]["excluded_state"])
await hass.async_block_till_done()
condition = await create_target_condition(
cond = await create_target_condition(
hass,
condition=condition,
target=condition_target_config,
@@ -1015,7 +1015,7 @@ async def assert_condition_behavior_all(
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert condition(hass) == state["condition_true_first_entity"]
assert cond(hass) == state["condition_true_first_entity"]
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, included_state)
@@ -1024,7 +1024,7 @@ async def assert_condition_behavior_all(
set_or_remove_state(hass, excluded_entity_id, excluded_state)
await hass.async_block_till_done()
assert condition(hass) == state["condition_true"]
assert cond(hass) == state["condition_true"]
async def assert_trigger_behavior_any(
@@ -1166,6 +1166,140 @@ async def assert_trigger_behavior_last(
assert len(service_calls) == 0
def parametrize_numerical_condition_above_below_any(
condition: str,
*,
device_class: str,
condition_options: dict[str, Any] | None = None,
unit_attributes: dict | None = None,
) -> list[tuple[str, dict[str, Any], list[ConditionStateDescription]]]:
"""Parametrize above/below threshold test cases for numerical conditions.
Returns a list of tuples with (condition, condition_options, states).
"""
from homeassistant.const import ATTR_DEVICE_CLASS # noqa: PLC0415
required_filter_attributes = {ATTR_DEVICE_CLASS: device_class}
condition_options = condition_options or {}
unit_attributes = unit_attributes or {}
return [
*parametrize_condition_states_any(
condition=condition,
condition_options={CONF_ABOVE: 20, **condition_options},
target_states=[
("21", unit_attributes),
("50", unit_attributes),
("100", unit_attributes),
],
other_states=[
("0", unit_attributes),
("10", unit_attributes),
("20", unit_attributes),
],
required_filter_attributes=required_filter_attributes,
),
*parametrize_condition_states_any(
condition=condition,
condition_options={CONF_BELOW: 80, **condition_options},
target_states=[
("0", unit_attributes),
("50", unit_attributes),
("79", unit_attributes),
],
other_states=[
("80", unit_attributes),
("90", unit_attributes),
("100", unit_attributes),
],
required_filter_attributes=required_filter_attributes,
),
*parametrize_condition_states_any(
condition=condition,
condition_options={CONF_ABOVE: 20, CONF_BELOW: 80, **condition_options},
target_states=[
("21", unit_attributes),
("50", unit_attributes),
("79", unit_attributes),
],
other_states=[
("0", unit_attributes),
("20", unit_attributes),
("80", unit_attributes),
("100", unit_attributes),
],
required_filter_attributes=required_filter_attributes,
),
]
def parametrize_numerical_condition_above_below_all(
condition: str,
*,
device_class: str,
condition_options: dict[str, Any] | None = None,
unit_attributes: dict | None = None,
) -> list[tuple[str, dict[str, Any], list[ConditionStateDescription]]]:
"""Parametrize above/below threshold test cases for numerical conditions with 'all' behavior.
Returns a list of tuples with (condition, condition_options, states).
"""
from homeassistant.const import ATTR_DEVICE_CLASS # noqa: PLC0415
required_filter_attributes = {ATTR_DEVICE_CLASS: device_class}
condition_options = condition_options or {}
unit_attributes = unit_attributes or {}
return [
*parametrize_condition_states_all(
condition=condition,
condition_options={CONF_ABOVE: 20, **condition_options},
target_states=[
("21", unit_attributes),
("50", unit_attributes),
("100", unit_attributes),
],
other_states=[
("0", unit_attributes),
("10", unit_attributes),
("20", unit_attributes),
],
required_filter_attributes=required_filter_attributes,
),
*parametrize_condition_states_all(
condition=condition,
condition_options={CONF_BELOW: 80, **condition_options},
target_states=[
("0", unit_attributes),
("50", unit_attributes),
("79", unit_attributes),
],
other_states=[
("80", unit_attributes),
("90", unit_attributes),
("100", unit_attributes),
],
required_filter_attributes=required_filter_attributes,
),
*parametrize_condition_states_all(
condition=condition,
condition_options={CONF_ABOVE: 20, CONF_BELOW: 80, **condition_options},
target_states=[
("21", unit_attributes),
("50", unit_attributes),
("79", unit_attributes),
],
other_states=[
("0", unit_attributes),
("20", unit_attributes),
("80", unit_attributes),
("100", unit_attributes),
],
required_filter_attributes=required_filter_attributes,
),
]
async def assert_trigger_ignores_limit_entities_with_wrong_unit(
hass: HomeAssistant,
*,
+261 -2
View File
@@ -1,5 +1,6 @@
"""Test the condition helper."""
from collections.abc import Mapping
from datetime import timedelta
import io
from typing import Any
@@ -19,9 +20,15 @@ from homeassistant.components.sun import DOMAIN as SUN_DOMAIN
from homeassistant.components.system_health import DOMAIN as SYSTEM_HEALTH_DOMAIN
from homeassistant.const import (
ATTR_DEVICE_CLASS,
ATTR_UNIT_OF_MEASUREMENT,
CONF_ABOVE,
CONF_BELOW,
CONF_CONDITION,
CONF_DEVICE_ID,
CONF_DOMAIN,
CONF_ENTITY_ID,
CONF_OPTIONS,
CONF_TARGET,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
)
@@ -33,14 +40,21 @@ from homeassistant.helpers import (
entity_registry as er,
trace,
)
from homeassistant.helpers.automation import move_top_level_schema_fields_to_options
from homeassistant.helpers.automation import (
DomainSpec,
move_top_level_schema_fields_to_options,
)
from homeassistant.helpers.condition import (
ATTR_BEHAVIOR,
BEHAVIOR_ALL,
BEHAVIOR_ANY,
Condition,
ConditionChecker,
async_validate_condition_config,
make_entity_numerical_condition,
)
from homeassistant.helpers.template import Template
from homeassistant.helpers.typing import ConfigType
from homeassistant.helpers.typing import UNDEFINED, ConfigType, UndefinedType
from homeassistant.loader import Integration, async_get_integration
from homeassistant.setup import async_setup_component
from homeassistant.util import dt as dt_util
@@ -2993,3 +3007,248 @@ async def test_subscribe_conditions_no_conditions(
assert await async_setup_component(hass, "light", {})
await hass.async_block_till_done()
assert condition_events == []
_DEFAULT_DOMAIN_SPECS = {"test": DomainSpec()}
async def _setup_numerical_condition(
hass: HomeAssistant,
condition_options: dict[str, Any],
entity_ids: str | list[str],
domain_specs: Mapping[str, DomainSpec] | None = None,
valid_unit: str | None | UndefinedType = UNDEFINED,
) -> condition.ConditionCheckerType:
"""Set up a numerical condition via a mock platform and return the test."""
condition_cls = make_entity_numerical_condition(
domain_specs or _DEFAULT_DOMAIN_SPECS, valid_unit
)
async def async_get_conditions(
hass: HomeAssistant,
) -> dict[str, type[Condition]]:
return {"_": condition_cls}
mock_integration(hass, MockModule("test"))
mock_platform(
hass, "test.condition", Mock(async_get_conditions=async_get_conditions)
)
if isinstance(entity_ids, str):
entity_ids = [entity_ids]
config: dict[str, Any] = {
CONF_CONDITION: "test",
CONF_TARGET: {CONF_ENTITY_ID: entity_ids},
CONF_OPTIONS: condition_options,
}
config = await async_validate_condition_config(hass, config)
test = await condition.async_from_config(hass, config)
assert test is not None
return test
@pytest.mark.parametrize(
("condition_options", "state_value", "expected"),
[
# above only
({CONF_ABOVE: 50}, "75", True),
({CONF_ABOVE: 50}, "50", False),
({CONF_ABOVE: 50}, "25", False),
# below only
({CONF_BELOW: 50}, "25", True),
({CONF_BELOW: 50}, "50", False),
({CONF_BELOW: 50}, "75", False),
# above and below (range)
({CONF_ABOVE: 20, CONF_BELOW: 80}, "50", True),
({CONF_ABOVE: 20, CONF_BELOW: 80}, "20", False),
({CONF_ABOVE: 20, CONF_BELOW: 80}, "80", False),
({CONF_ABOVE: 20, CONF_BELOW: 80}, "10", False),
({CONF_ABOVE: 20, CONF_BELOW: 80}, "90", False),
],
)
async def test_numerical_condition_thresholds(
hass: HomeAssistant,
condition_options: dict[str, Any],
state_value: str,
expected: bool,
) -> None:
"""Test numerical condition above/below thresholds."""
test = await _setup_numerical_condition(
hass,
condition_options=condition_options,
entity_ids="test.entity_1",
)
hass.states.async_set("test.entity_1", state_value)
assert test(hass) is expected
@pytest.mark.parametrize(
"state_value",
["cat", STATE_UNAVAILABLE, STATE_UNKNOWN],
)
async def test_numerical_condition_invalid_state(
hass: HomeAssistant, state_value: str
) -> None:
"""Test numerical condition with non-numeric or unavailable state values."""
test = await _setup_numerical_condition(
hass,
condition_options={CONF_ABOVE: 50},
entity_ids="test.entity_1",
)
hass.states.async_set("test.entity_1", state_value)
assert test(hass) is False
async def test_numerical_condition_attribute_value_source(
hass: HomeAssistant,
) -> None:
"""Test numerical condition reads from attribute when value_source is set."""
test = await _setup_numerical_condition(
hass,
domain_specs={"test": DomainSpec(value_source="brightness")},
condition_options={CONF_ABOVE: 100},
entity_ids="test.entity_1",
)
# Attribute above threshold -> True
hass.states.async_set("test.entity_1", "on", {"brightness": 200})
assert test(hass) is True
# Attribute below threshold -> False
hass.states.async_set("test.entity_1", "on", {"brightness": 50})
assert test(hass) is False
# Missing attribute -> False
hass.states.async_set("test.entity_1", "on", {})
assert test(hass) is False
@pytest.mark.parametrize(
("valid_unit", "entity_unit", "expected"),
[
# valid_unit="%" — only matching unit passes
("%", "%", True),
("%", "°C", False),
("%", None, False),
# valid_unit=None — only entities without unit pass
(None, None, True),
(None, "%", False),
# valid_unit=UNDEFINED (default) — any unit passes
(UNDEFINED, None, True),
(UNDEFINED, "%", True),
(UNDEFINED, "°C", True),
],
)
async def test_numerical_condition_valid_unit(
hass: HomeAssistant,
valid_unit: str | None | UndefinedType,
entity_unit: str | None,
expected: bool,
) -> None:
"""Test numerical condition valid_unit filtering."""
test = await _setup_numerical_condition(
hass,
condition_options={CONF_ABOVE: 50},
entity_ids="test.entity_1",
valid_unit=valid_unit,
)
attrs = {ATTR_UNIT_OF_MEASUREMENT: entity_unit} if entity_unit else {}
hass.states.async_set("test.entity_1", "75", attrs)
assert test(hass) is expected
@pytest.mark.parametrize(
("behavior", "one_match_expected"),
[
(BEHAVIOR_ANY, True),
(BEHAVIOR_ALL, False),
],
)
async def test_numerical_condition_behavior(
hass: HomeAssistant,
behavior: str,
one_match_expected: bool,
) -> None:
"""Test numerical condition with behavior any/all."""
test = await _setup_numerical_condition(
hass,
condition_options={CONF_ABOVE: 50, ATTR_BEHAVIOR: behavior},
entity_ids=["test.entity_1", "test.entity_2"],
)
# Both above -> True for any and all
hass.states.async_set("test.entity_1", "75")
hass.states.async_set("test.entity_2", "80")
assert test(hass) is True
# Only one above -> depends on behavior
hass.states.async_set("test.entity_2", "25")
assert test(hass) is one_match_expected
# Neither above -> False for any and all
hass.states.async_set("test.entity_1", "25")
assert test(hass) is False
async def test_numerical_condition_schema_requires_above_or_below(
hass: HomeAssistant,
) -> None:
"""Test numerical condition schema requires at least above or below."""
condition_cls = make_entity_numerical_condition({"test": DomainSpec()})
async def async_get_conditions(
hass: HomeAssistant,
) -> dict[str, type[Condition]]:
return {"_": condition_cls}
mock_integration(hass, MockModule("test"))
mock_platform(
hass, "test.condition", Mock(async_get_conditions=async_get_conditions)
)
config: dict[str, Any] = {
CONF_CONDITION: "test",
CONF_TARGET: {CONF_ENTITY_ID: "test.entity_1"},
CONF_OPTIONS: {},
}
with pytest.raises(vol.Invalid):
await async_validate_condition_config(hass, config)
@pytest.mark.parametrize(
("above", "below"),
[
(10.0, 10.0),
(20.0, 10.0),
],
)
async def test_numerical_condition_schema_above_must_be_less_than_below(
hass: HomeAssistant,
above: float,
below: float,
) -> None:
"""Test numerical condition schema rejects above >= below."""
condition_cls = make_entity_numerical_condition({"test": DomainSpec()})
async def async_get_conditions(
hass: HomeAssistant,
) -> dict[str, type[Condition]]:
return {"_": condition_cls}
mock_integration(hass, MockModule("test"))
mock_platform(
hass, "test.condition", Mock(async_get_conditions=async_get_conditions)
)
config: dict[str, Any] = {
CONF_CONDITION: "test",
CONF_TARGET: {CONF_ENTITY_ID: "test.entity_1"},
CONF_OPTIONS: {CONF_ABOVE: above, CONF_BELOW: below},
}
with pytest.raises(vol.Invalid, match="can never be above"):
await async_validate_condition_config(hass, config)
+2
View File
@@ -14,6 +14,7 @@
'backup',
'backup.event',
'backup.sensor',
'battery',
'binary_sensor',
'blueprint',
'brands',
@@ -120,6 +121,7 @@
'backup',
'backup.event',
'backup.sensor',
'battery',
'binary_sensor',
'blueprint',
'brands',