Compare commits

..

5 Commits

Author SHA1 Message Date
Paul Bottein
0063dc81d3 Copilot suggestions 2026-03-21 19:09:13 +01:00
Paul Bottein
7463bb79dd Remove expiration 2026-03-21 19:07:13 +01:00
Paul Bottein
d17b681477 Add time sync button to Matter integration 2026-03-21 18:57:50 +01:00
Ingmar Stein
c6c5661b4b Add Identify button to Velux integration (#163893)
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-03-21 18:20:02 +01:00
Joost Lekkerkerker
d0154e5019 Add stick cleaner fixture to SmartThings (#166121) 2026-03-21 16:57:26 +01:00
34 changed files with 1763 additions and 2361 deletions

2
CODEOWNERS generated
View File

@@ -1703,8 +1703,6 @@ build.json @home-assistant/supervisor
/tests/components/tellduslive/ @fredrike
/homeassistant/components/teltonika/ @karlbeecken
/tests/components/teltonika/ @karlbeecken
/homeassistant/components/temperature/ @home-assistant/core
/tests/components/temperature/ @home-assistant/core
/homeassistant/components/template/ @Petro31 @home-assistant/core
/tests/components/template/ @Petro31 @home-assistant/core
/homeassistant/components/tesla_fleet/ @Bre77

View File

@@ -247,7 +247,6 @@ DEFAULT_INTEGRATIONS = {
"humidity",
"motion",
"occupancy",
"temperature",
"window",
}
DEFAULT_INTEGRATIONS_RECOVERY_MODE = {

View File

@@ -169,7 +169,6 @@ _EXPERIMENTAL_TRIGGER_PLATFORMS = {
"select",
"siren",
"switch",
"temperature",
"text",
"update",
"vacuum",

View File

@@ -4,9 +4,11 @@ from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from datetime import UTC, datetime
from typing import TYPE_CHECKING, Any
from chip.clusters import Objects as clusters
from chip.clusters.Types import NullValue
from homeassistant.components.button import (
ButtonDeviceClass,
@@ -17,6 +19,7 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.const import EntityCategory, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.util import dt as dt_util
from .entity import MatterEntity, MatterEntityDescription
from .helpers import get_matter
@@ -52,6 +55,67 @@ class MatterCommandButton(MatterEntity, ButtonEntity):
await self.send_device_command(self.entity_description.command())
# CHIP epoch: 2000-01-01 00:00:00 UTC
CHIP_EPOCH = datetime(2000, 1, 1, tzinfo=UTC)
class MatterTimeSyncButton(MatterEntity, ButtonEntity):
"""Button to synchronize time to a Matter device."""
entity_description: MatterButtonEntityDescription
async def async_press(self) -> None:
"""Sync Home Assistant time to the Matter device."""
now = dt_util.utcnow()
tz = dt_util.get_default_time_zone()
delta = now - CHIP_EPOCH
utc_us = (
(delta.days * 86400 * 1_000_000)
+ (delta.seconds * 1_000_000)
+ delta.microseconds
)
# Compute timezone and DST offsets
local_now = now.astimezone(tz)
utc_offset_delta = local_now.utcoffset()
utc_offset = int(utc_offset_delta.total_seconds()) if utc_offset_delta else 0
dst_offset_delta = local_now.dst()
dst_offset = int(dst_offset_delta.total_seconds()) if dst_offset_delta else 0
standard_offset = utc_offset - dst_offset
# 1. Set timezone
await self.send_device_command(
clusters.TimeSynchronization.Commands.SetTimeZone(
timeZone=[
clusters.TimeSynchronization.Structs.TimeZoneStruct(
offset=standard_offset, validAt=0, name=str(tz)
)
]
)
)
# 2. Set DST offset
await self.send_device_command(
clusters.TimeSynchronization.Commands.SetDSTOffset(
DSTOffset=[
clusters.TimeSynchronization.Structs.DSTOffsetStruct(
offset=dst_offset,
validStarting=0,
validUntil=NullValue,
)
]
)
)
# 3. Set UTC time
await self.send_device_command(
clusters.TimeSynchronization.Commands.SetUTCTime(
UTCTime=utc_us,
granularity=clusters.TimeSynchronization.Enums.GranularityEnum.kMicrosecondsGranularity,
)
)
# Discovery schema(s) to map Matter Attributes to HA entities
DISCOVERY_SCHEMAS = [
MatterDiscoverySchema(
@@ -169,4 +233,16 @@ DISCOVERY_SCHEMAS = [
value_contains=clusters.WaterHeaterManagement.Commands.CancelBoost.command_id,
allow_multi=True, # Also used in water_heater
),
MatterDiscoverySchema(
platform=Platform.BUTTON,
entity_description=MatterButtonEntityDescription(
key="TimeSynchronizationSyncTimeButton",
translation_key="sync_time",
entity_category=EntityCategory.CONFIG,
),
entity_class=MatterTimeSyncButton,
required_attributes=(clusters.TimeSynchronization.Attributes.UTCTime,),
allow_multi=True,
allow_none_value=True,
),
]

View File

@@ -20,6 +20,9 @@
},
"stop": {
"default": "mdi:stop"
},
"sync_time": {
"default": "mdi:clock-check-outline"
}
},
"fan": {

View File

@@ -141,6 +141,9 @@
},
"stop": {
"name": "[%key:common::action::stop%]"
},
"sync_time": {
"name": "Sync time"
}
},
"climate": {

View File

@@ -1,17 +0,0 @@
"""Integration for temperature triggers."""
from __future__ import annotations
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.typing import ConfigType
DOMAIN = "temperature"
CONFIG_SCHEMA = cv.empty_config_schema(DOMAIN)
__all__ = []
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the component."""
return True

View File

@@ -1,10 +0,0 @@
{
"triggers": {
"changed": {
"trigger": "mdi:thermometer"
},
"crossed_threshold": {
"trigger": "mdi:thermometer"
}
}
}

View File

@@ -1,8 +0,0 @@
{
"domain": "temperature",
"name": "Temperature",
"codeowners": ["@home-assistant/core"],
"documentation": "https://www.home-assistant.io/integrations/temperature",
"integration_type": "system",
"quality_scale": "internal"
}

View File

@@ -1,76 +0,0 @@
{
"common": {
"trigger_behavior_description": "The behavior of the targeted entities to trigger on.",
"trigger_behavior_name": "Behavior"
},
"selector": {
"number_or_entity": {
"choices": {
"entity": "Entity",
"number": "Number"
}
},
"trigger_behavior": {
"options": {
"any": "Any",
"first": "First",
"last": "Last"
}
},
"trigger_threshold_type": {
"options": {
"above": "Above",
"below": "Below",
"between": "Between",
"outside": "Outside"
}
}
},
"title": "Temperature",
"triggers": {
"changed": {
"description": "Triggers when the temperature changes.",
"fields": {
"above": {
"description": "Only trigger when temperature is above this value.",
"name": "Above"
},
"below": {
"description": "Only trigger when temperature is below this value.",
"name": "Below"
},
"unit": {
"description": "All values will be converted to this unit when evaluating the trigger.",
"name": "Unit of measurement"
}
},
"name": "Temperature changed"
},
"crossed_threshold": {
"description": "Triggers when the temperature crosses a threshold.",
"fields": {
"behavior": {
"description": "[%key:component::temperature::common::trigger_behavior_description%]",
"name": "[%key:component::temperature::common::trigger_behavior_name%]"
},
"lower_limit": {
"description": "The lower limit of the threshold.",
"name": "Lower limit"
},
"threshold_type": {
"description": "The type of threshold to use.",
"name": "Threshold type"
},
"unit": {
"description": "[%key:component::temperature::triggers::changed::fields::unit::description%]",
"name": "[%key:component::temperature::triggers::changed::fields::unit::name%]"
},
"upper_limit": {
"description": "The upper limit of the threshold.",
"name": "Upper limit"
}
},
"name": "Temperature crossed threshold"
}
}
}

View File

@@ -1,83 +0,0 @@
"""Provides triggers for temperature."""
from __future__ import annotations
from homeassistant.components.climate import (
ATTR_CURRENT_TEMPERATURE as CLIMATE_ATTR_CURRENT_TEMPERATURE,
DOMAIN as CLIMATE_DOMAIN,
)
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN, SensorDeviceClass
from homeassistant.components.water_heater import (
ATTR_CURRENT_TEMPERATURE as WATER_HEATER_ATTR_CURRENT_TEMPERATURE,
DOMAIN as WATER_HEATER_DOMAIN,
)
from homeassistant.components.weather import (
ATTR_WEATHER_TEMPERATURE,
ATTR_WEATHER_TEMPERATURE_UNIT,
DOMAIN as WEATHER_DOMAIN,
)
from homeassistant.const import ATTR_UNIT_OF_MEASUREMENT, UnitOfTemperature
from homeassistant.core import HomeAssistant, State
from homeassistant.helpers.automation import NumericalDomainSpec
from homeassistant.helpers.trigger import (
EntityNumericalStateChangedTriggerWithUnitBase,
EntityNumericalStateCrossedThresholdTriggerWithUnitBase,
EntityNumericalStateTriggerWithUnitBase,
Trigger,
)
from homeassistant.util.unit_conversion import TemperatureConverter
TEMPERATURE_DOMAIN_SPECS = {
CLIMATE_DOMAIN: NumericalDomainSpec(
value_source=CLIMATE_ATTR_CURRENT_TEMPERATURE,
),
SENSOR_DOMAIN: NumericalDomainSpec(
device_class=SensorDeviceClass.TEMPERATURE,
),
WATER_HEATER_DOMAIN: NumericalDomainSpec(
value_source=WATER_HEATER_ATTR_CURRENT_TEMPERATURE
),
WEATHER_DOMAIN: NumericalDomainSpec(
value_source=ATTR_WEATHER_TEMPERATURE,
),
}
class _TemperatureTriggerMixin(EntityNumericalStateTriggerWithUnitBase):
"""Mixin for temperature triggers providing entity filtering, value extraction, and unit conversion."""
_base_unit = UnitOfTemperature.CELSIUS
_domain_specs = TEMPERATURE_DOMAIN_SPECS
_unit_converter = TemperatureConverter
def _get_entity_unit(self, state: State) -> str | None:
"""Get the temperature unit of an entity from its state."""
if state.domain == SENSOR_DOMAIN:
return state.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
if state.domain == WEATHER_DOMAIN:
return state.attributes.get(ATTR_WEATHER_TEMPERATURE_UNIT)
# Climate and water_heater: show_temp converts to system unit
return self._hass.config.units.temperature_unit
class TemperatureChangedTrigger(
_TemperatureTriggerMixin, EntityNumericalStateChangedTriggerWithUnitBase
):
"""Trigger for temperature value changes across multiple domains."""
class TemperatureCrossedThresholdTrigger(
_TemperatureTriggerMixin, EntityNumericalStateCrossedThresholdTriggerWithUnitBase
):
"""Trigger for temperature value crossing a threshold across multiple domains."""
TRIGGERS: dict[str, type[Trigger]] = {
"changed": TemperatureChangedTrigger,
"crossed_threshold": TemperatureCrossedThresholdTrigger,
}
async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
"""Return the triggers for temperature."""
return TRIGGERS

View File

@@ -1,77 +0,0 @@
.trigger_common_fields:
behavior: &trigger_behavior
required: true
default: any
selector:
select:
translation_key: trigger_behavior
options:
- first
- last
- any
.number_or_entity: &number_or_entity
required: false
selector:
choose:
choices:
number:
selector:
number:
mode: box
entity:
selector:
entity:
filter:
- domain: input_number
unit_of_measurement:
- "°C"
- "°F"
- domain: sensor
device_class: temperature
- domain: number
device_class: temperature
translation_key: number_or_entity
.trigger_threshold_type: &trigger_threshold_type
required: true
default: above
selector:
select:
options:
- above
- below
- between
- outside
translation_key: trigger_threshold_type
.trigger_unit: &trigger_unit
required: false
selector:
select:
options:
- "°C"
- "°F"
.trigger_target: &trigger_target
entity:
- domain: sensor
device_class: temperature
- domain: climate
- domain: water_heater
- domain: weather
changed:
target: *trigger_target
fields:
above: *number_or_entity
below: *number_or_entity
unit: *trigger_unit
crossed_threshold:
target: *trigger_target
fields:
behavior: *trigger_behavior
threshold_type: *trigger_threshold_type
lower_limit: *number_or_entity
upper_limit: *number_or_entity
unit: *trigger_unit

View File

@@ -2,7 +2,7 @@
from __future__ import annotations
from pyvlx import PyVLX, PyVLXException
from pyvlx import Node, PyVLX, PyVLXException
from homeassistant.components.button import ButtonDeviceClass, ButtonEntity
from homeassistant.const import EntityCategory
@@ -13,6 +13,7 @@ from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import VeluxConfigEntry
from .const import DOMAIN
from .entity import VeluxEntity, wrap_pyvlx_call_exceptions
PARALLEL_UPDATES = 1
@@ -23,9 +24,32 @@ async def async_setup_entry(
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up button entities for the Velux integration."""
async_add_entities(
[VeluxGatewayRebootButton(config_entry.entry_id, config_entry.runtime_data)]
entities: list[ButtonEntity] = [
VeluxGatewayRebootButton(config_entry.entry_id, config_entry.runtime_data)
]
entities.extend(
VeluxIdentifyButton(node, config_entry.entry_id)
for node in config_entry.runtime_data.nodes
if isinstance(node, Node)
)
async_add_entities(entities)
class VeluxIdentifyButton(VeluxEntity, ButtonEntity):
"""Representation of a Velux identify button."""
_attr_device_class = ButtonDeviceClass.IDENTIFY
_attr_entity_category = EntityCategory.DIAGNOSTIC
def __init__(self, node: Node, config_entry_id: str) -> None:
"""Initialize the Velux identify button."""
super().__init__(node, config_entry_id)
self._attr_unique_id = f"{self._attr_unique_id}_identify"
@wrap_pyvlx_call_exceptions
async def async_press(self) -> None:
"""Identify the physical device."""
await self.node.wink()
class VeluxGatewayRebootButton(ButtonEntity):

View File

@@ -39,7 +39,7 @@ class DomainSpec:
class NumericalDomainSpec(DomainSpec):
"""DomainSpec with an optional value converter for numerical triggers."""
value_converter: Callable[[float], float] | None = None
value_converter: Callable[[Any], float] | None = None
"""Optional converter for numerical values (e.g. uint8 → percentage)."""

View File

@@ -26,7 +26,6 @@ import voluptuous as vol
from homeassistant.const import (
ATTR_ENTITY_ID,
ATTR_UNIT_OF_MEASUREMENT,
CONF_ABOVE,
CONF_ALIAS,
CONF_BELOW,
@@ -65,7 +64,6 @@ from homeassistant.loader import (
)
from homeassistant.util.async_ import create_eager_task
from homeassistant.util.hass_dict import HassKey
from homeassistant.util.unit_conversion import BaseUnitConverter
from homeassistant.util.yaml import load_yaml_dict
from . import config_validation as cv, selector
@@ -521,7 +519,7 @@ def _validate_range[_T: dict[str, Any]](
) -> Callable[[_T], _T]:
"""Generate range validator."""
def _validate_range_impl(value: _T) -> _T:
def _validate_range(value: _T) -> _T:
above = value.get(lower_limit)
below = value.get(upper_limit)
@@ -541,28 +539,7 @@ def _validate_range[_T: dict[str, Any]](
return value
return _validate_range_impl
CONF_UNIT: Final = "unit"
def _validate_unit_set_if_range_numerical[_T: dict[str, Any]](
lower_limit: str, upper_limit: str
) -> Callable[[_T], _T]:
"""Validate that unit is set if upper or lower limit is numerical."""
def _validate_unit_set_if_range_numerical_impl(options: _T) -> _T:
if (
any(
opt in options and not isinstance(options[opt], str)
for opt in (lower_limit, upper_limit)
)
) and options.get(CONF_UNIT) is None:
raise vol.Invalid("Unit must be specified when using numerical thresholds.")
return options
return _validate_unit_set_if_range_numerical_impl
return _validate_range
_NUMBER_OR_ENTITY_CHOOSE_SCHEMA = vol.Schema(
@@ -599,107 +576,38 @@ NUMERICAL_ATTRIBUTE_CHANGED_TRIGGER_SCHEMA = ENTITY_STATE_TRIGGER_SCHEMA.extend(
)
class EntityNumericalStateTriggerBase(EntityTriggerBase[NumericalDomainSpec]):
"""Base class for numerical state and state attribute triggers."""
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 (state := self._hass.states.get(entity_or_float)):
# Entity not found
return None
try:
return float(state.state)
except TypeError, ValueError:
# Entity state is not a valid number
return None
return entity_or_float
def _get_tracked_value(self, state: State) -> float | None:
"""Get the tracked numerical value from a state."""
domain_spec = self._domain_specs[state.domain]
raw_value: Any
if domain_spec.value_source is None:
raw_value = state.state
else:
raw_value = state.attributes.get(domain_spec.value_source)
def _get_numerical_value(
hass: HomeAssistant, entity_or_float: float | str
) -> float | None:
"""Get numerical value from float or entity state."""
if isinstance(entity_or_float, str):
if not (state := hass.states.get(entity_or_float)):
# Entity not found
return None
try:
return float(raw_value)
return float(state.state)
except TypeError, ValueError:
# Entity state is not a valid number
return None
return entity_or_float
def _get_converter(self, state: State) -> Callable[[float], float]:
class EntityNumericalStateTriggerBase(EntityTriggerBase[NumericalDomainSpec]):
"""Base class for numerical state and state attribute triggers."""
def _get_tracked_value(self, state: State) -> Any:
"""Get the tracked numerical value from a state."""
domain_spec = self._domain_specs[state.domain]
if domain_spec.value_source is None:
return state.state
return state.attributes.get(domain_spec.value_source)
def _get_converter(self, state: State) -> Callable[[Any], float]:
"""Get the value converter for an entity."""
domain_spec = self._domain_specs[state.domain]
if domain_spec.value_converter is not None:
return domain_spec.value_converter
return lambda x: x
class EntityNumericalStateTriggerWithUnitBase(EntityNumericalStateTriggerBase):
"""Base class for numerical state and state attribute triggers."""
_base_unit: str # Base unit for the tracked value
_manual_limit_unit: str | None # Unit of above/below limits when numbers
_unit_converter: type[BaseUnitConverter]
def __init__(self, hass: HomeAssistant, config: TriggerConfig) -> None:
"""Initialize the trigger."""
super().__init__(hass, config)
self._manual_limit_unit = self._options.get(CONF_UNIT)
def _get_entity_unit(self, state: State) -> str | None:
"""Get the unit of an entity from its state."""
return state.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
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, (int, float)):
return self._unit_converter.convert(
entity_or_float, self._manual_limit_unit, self._base_unit
)
if not (state := self._hass.states.get(entity_or_float)):
# Entity not found
return None
try:
value = float(state.state)
except TypeError, ValueError:
# Entity state is not a valid number
return None
try:
return self._unit_converter.convert(
value, state.attributes.get(ATTR_UNIT_OF_MEASUREMENT), self._base_unit
)
except HomeAssistantError:
# Unit conversion failed (i.e. incompatible units), treat as invalid number
return None
def _get_tracked_value(self, state: State) -> float | None:
"""Get the tracked numerical value from a state."""
domain_spec = self._domain_specs[state.domain]
raw_value: Any
if domain_spec.value_source is None:
raw_value = state.state
else:
raw_value = state.attributes.get(domain_spec.value_source)
try:
value = float(raw_value)
except TypeError, ValueError:
# Entity state is not a valid number
return None
try:
return self._unit_converter.convert(
value, self._get_entity_unit(state), self._base_unit
)
except HomeAssistantError:
# Unit conversion failed (i.e. incompatible units), treat as invalid number
return None
return float
class EntityNumericalStateChangedTriggerBase(EntityNumericalStateTriggerBase):
@@ -721,7 +629,7 @@ class EntityNumericalStateChangedTriggerBase(EntityNumericalStateTriggerBase):
if from_state.state in (STATE_UNAVAILABLE, STATE_UNKNOWN):
return False
return self._get_tracked_value(from_state) != self._get_tracked_value(to_state)
return self._get_tracked_value(from_state) != self._get_tracked_value(to_state) # type: ignore[no-any-return]
def is_valid_state(self, state: State) -> bool:
"""Check if the new state or state attribute matches the expected one."""
@@ -729,10 +637,14 @@ class EntityNumericalStateChangedTriggerBase(EntityNumericalStateTriggerBase):
if (_attribute_value := self._get_tracked_value(state)) is None:
return False
current_value = self._get_converter(state)(_attribute_value)
try:
current_value = self._get_converter(state)(_attribute_value)
except TypeError, ValueError:
# Value is not a valid number, don't trigger
return False
if self._above is not None:
if (above := self._get_numerical_value(self._above)) is None:
if (above := _get_numerical_value(self._hass, self._above)) is None:
# Entity not found or invalid number, don't trigger
return False
if current_value <= above:
@@ -740,7 +652,7 @@ class EntityNumericalStateChangedTriggerBase(EntityNumericalStateTriggerBase):
return False
if self._below is not None:
if (below := self._get_numerical_value(self._below)) is None:
if (below := _get_numerical_value(self._hass, self._below)) is None:
# Entity not found or invalid number, don't trigger
return False
if current_value >= below:
@@ -750,37 +662,6 @@ class EntityNumericalStateChangedTriggerBase(EntityNumericalStateTriggerBase):
return True
def make_numerical_state_changed_with_unit_schema(
unit_converter: type[BaseUnitConverter],
) -> vol.Schema:
"""Factory for numerical state trigger schema with unit option."""
return 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_UNIT): vol.In(unit_converter.VALID_UNITS),
},
_validate_range(CONF_ABOVE, CONF_BELOW),
_validate_unit_set_if_range_numerical(CONF_ABOVE, CONF_BELOW),
)
}
)
class EntityNumericalStateChangedTriggerWithUnitBase(
EntityNumericalStateChangedTriggerBase,
EntityNumericalStateTriggerWithUnitBase,
):
"""Trigger for numerical state and state attribute changes."""
def __init_subclass__(cls, **kwargs: Any) -> None:
"""Create a schema."""
super().__init_subclass__(**kwargs)
cls._schema = make_numerical_state_changed_with_unit_schema(cls._unit_converter)
CONF_LOWER_LIMIT = "lower_limit"
CONF_UPPER_LIMIT = "upper_limit"
CONF_THRESHOLD_TYPE = "threshold_type"
@@ -863,12 +744,16 @@ class EntityNumericalStateCrossedThresholdTriggerBase(EntityNumericalStateTrigge
def is_valid_state(self, state: State) -> bool:
"""Check if the new state attribute matches the expected one."""
if self._lower_limit is not None:
if (lower_limit := self._get_numerical_value(self._lower_limit)) is None:
if (
lower_limit := _get_numerical_value(self._hass, self._lower_limit)
) is None:
# Entity not found or invalid number, don't trigger
return False
if self._upper_limit is not None:
if (upper_limit := self._get_numerical_value(self._upper_limit)) is None:
if (
upper_limit := _get_numerical_value(self._hass, self._upper_limit)
) is None:
# Entity not found or invalid number, don't trigger
return False
@@ -876,7 +761,11 @@ class EntityNumericalStateCrossedThresholdTriggerBase(EntityNumericalStateTrigge
if (_attribute_value := self._get_tracked_value(state)) is None:
return False
current_value = self._get_converter(state)(_attribute_value)
try:
current_value = self._get_converter(state)(_attribute_value)
except TypeError, ValueError:
# Value is not a valid number, don't trigger
return False
# Note: We do not need to check for lower_limit/upper_limit being None here
# because of the validation done in the schema.
@@ -892,50 +781,6 @@ class EntityNumericalStateCrossedThresholdTriggerBase(EntityNumericalStateTrigge
return not between
def make_numerical_state_crossed_threshold_with_unit_schema(
unit_converter: type[BaseUnitConverter],
) -> vol.Schema:
"""Trigger for numerical state and state attribute changes.
This trigger only fires when the observed attribute changes from not within to within
the defined threshold.
"""
return ENTITY_STATE_TRIGGER_SCHEMA.extend(
{
vol.Required(CONF_OPTIONS, default={}): vol.All(
{
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.Required(CONF_THRESHOLD_TYPE): vol.Coerce(ThresholdType),
vol.Optional(CONF_UNIT): vol.In(unit_converter.VALID_UNITS),
},
_validate_range(CONF_LOWER_LIMIT, CONF_UPPER_LIMIT),
_validate_limits_for_threshold_type,
_validate_unit_set_if_range_numerical(
CONF_LOWER_LIMIT, CONF_UPPER_LIMIT
),
)
}
)
class EntityNumericalStateCrossedThresholdTriggerWithUnitBase(
EntityNumericalStateCrossedThresholdTriggerBase,
EntityNumericalStateTriggerWithUnitBase,
):
"""Trigger for numerical state and state attribute changes."""
def __init_subclass__(cls, **kwargs: Any) -> None:
"""Create a schema."""
super().__init_subclass__(**kwargs)
cls._schema = make_numerical_state_crossed_threshold_with_unit_schema(
cls._unit_converter
)
def _normalize_domain_specs(
domain_specs: Mapping[str, DomainSpec] | str,
) -> Mapping[str, DomainSpec]:

View File

@@ -120,7 +120,6 @@ NO_IOT_CLASS = [
"system_health",
"system_log",
"tag",
"temperature",
"timer",
"trace",
"web_rtc",

View File

@@ -2149,7 +2149,6 @@ NO_QUALITY_SCALE = [
"system_health",
"system_log",
"tag",
"temperature",
"timer",
"trace",
"usage_prediction",

View File

@@ -518,75 +518,54 @@ def parametrize_trigger_states(
def parametrize_numerical_attribute_changed_trigger_states(
trigger: str,
state: str,
attribute: str,
*,
trigger_options: dict[str, Any] | None = None,
required_filter_attributes: dict | None = None,
unit_attributes: dict | None = None,
trigger: str, state: str, attribute: str
) -> list[tuple[str, dict[str, Any], list[TriggerStateDescription]]]:
"""Parametrize states and expected service call counts for numerical changed triggers."""
trigger_options = trigger_options or {}
unit_attributes = unit_attributes or {}
return [
*parametrize_trigger_states(
trigger=trigger,
trigger_options={**trigger_options},
trigger_options={},
target_states=[
(state, {attribute: 0} | unit_attributes),
(state, {attribute: 50} | unit_attributes),
(state, {attribute: 100} | unit_attributes),
(state, {attribute: 0}),
(state, {attribute: 50}),
(state, {attribute: 100}),
],
other_states=[(state, {attribute: None} | unit_attributes)],
required_filter_attributes=required_filter_attributes,
other_states=[(state, {attribute: None})],
retrigger_on_target_state=True,
),
*parametrize_trigger_states(
trigger=trigger,
trigger_options={CONF_ABOVE: 10, **trigger_options},
trigger_options={CONF_ABOVE: 10},
target_states=[
(state, {attribute: 50} | unit_attributes),
(state, {attribute: 100} | unit_attributes),
(state, {attribute: 50}),
(state, {attribute: 100}),
],
other_states=[
(state, {attribute: None} | unit_attributes),
(state, {attribute: 0} | unit_attributes),
(state, {attribute: None}),
(state, {attribute: 0}),
],
required_filter_attributes=required_filter_attributes,
retrigger_on_target_state=True,
),
*parametrize_trigger_states(
trigger=trigger,
trigger_options={CONF_BELOW: 90, **trigger_options},
trigger_options={CONF_BELOW: 90},
target_states=[
(state, {attribute: 0} | unit_attributes),
(state, {attribute: 50} | unit_attributes),
(state, {attribute: 0}),
(state, {attribute: 50}),
],
other_states=[
(state, {attribute: None} | unit_attributes),
(state, {attribute: 100} | unit_attributes),
(state, {attribute: None}),
(state, {attribute: 100}),
],
required_filter_attributes=required_filter_attributes,
retrigger_on_target_state=True,
),
]
def parametrize_numerical_attribute_crossed_threshold_trigger_states(
trigger: str,
state: str,
attribute: str,
*,
trigger_options: dict[str, Any] | None = None,
required_filter_attributes: dict | None = None,
unit_attributes: dict | None = None,
trigger: str, state: str, attribute: str
) -> list[tuple[str, dict[str, Any], list[TriggerStateDescription]]]:
"""Parametrize states and expected service call counts for numerical crossed threshold triggers."""
trigger_options = trigger_options or {}
unit_attributes = unit_attributes or {}
return [
*parametrize_trigger_states(
trigger=trigger,
@@ -594,18 +573,16 @@ def parametrize_numerical_attribute_crossed_threshold_trigger_states(
CONF_THRESHOLD_TYPE: ThresholdType.BETWEEN,
CONF_LOWER_LIMIT: 10,
CONF_UPPER_LIMIT: 90,
**trigger_options,
},
target_states=[
(state, {attribute: 50} | unit_attributes),
(state, {attribute: 60} | unit_attributes),
(state, {attribute: 50}),
(state, {attribute: 60}),
],
other_states=[
(state, {attribute: None} | unit_attributes),
(state, {attribute: 0} | unit_attributes),
(state, {attribute: 100} | unit_attributes),
(state, {attribute: None}),
(state, {attribute: 0}),
(state, {attribute: 100}),
],
required_filter_attributes=required_filter_attributes,
),
*parametrize_trigger_states(
trigger=trigger,
@@ -613,62 +590,52 @@ def parametrize_numerical_attribute_crossed_threshold_trigger_states(
CONF_THRESHOLD_TYPE: ThresholdType.OUTSIDE,
CONF_LOWER_LIMIT: 10,
CONF_UPPER_LIMIT: 90,
**trigger_options,
},
target_states=[
(state, {attribute: 0} | unit_attributes),
(state, {attribute: 100} | unit_attributes),
(state, {attribute: 0}),
(state, {attribute: 100}),
],
other_states=[
(state, {attribute: None} | unit_attributes),
(state, {attribute: 50} | unit_attributes),
(state, {attribute: 60} | unit_attributes),
(state, {attribute: None}),
(state, {attribute: 50}),
(state, {attribute: 60}),
],
required_filter_attributes=required_filter_attributes,
),
*parametrize_trigger_states(
trigger=trigger,
trigger_options={
CONF_THRESHOLD_TYPE: ThresholdType.ABOVE,
CONF_LOWER_LIMIT: 10,
**trigger_options,
},
target_states=[
(state, {attribute: 50} | unit_attributes),
(state, {attribute: 100} | unit_attributes),
(state, {attribute: 50}),
(state, {attribute: 100}),
],
other_states=[
(state, {attribute: None} | unit_attributes),
(state, {attribute: 0} | unit_attributes),
(state, {attribute: None}),
(state, {attribute: 0}),
],
required_filter_attributes=required_filter_attributes,
),
*parametrize_trigger_states(
trigger=trigger,
trigger_options={
CONF_THRESHOLD_TYPE: ThresholdType.BELOW,
CONF_UPPER_LIMIT: 90,
**trigger_options,
},
target_states=[
(state, {attribute: 0} | unit_attributes),
(state, {attribute: 50} | unit_attributes),
(state, {attribute: 0}),
(state, {attribute: 50}),
],
other_states=[
(state, {attribute: None} | unit_attributes),
(state, {attribute: 100} | unit_attributes),
(state, {attribute: None}),
(state, {attribute: 100}),
],
required_filter_attributes=required_filter_attributes,
),
]
def parametrize_numerical_state_value_changed_trigger_states(
trigger: str,
*,
device_class: str,
trigger_options: dict[str, Any] | None = None,
unit_attributes: dict | None = None,
trigger: str, device_class: str
) -> list[tuple[str, dict[str, Any], list[TriggerStateDescription]]]:
"""Parametrize states and expected service call counts for numerical state-value changed triggers.
@@ -679,37 +646,30 @@ def parametrize_numerical_state_value_changed_trigger_states(
from homeassistant.const import ATTR_DEVICE_CLASS # noqa: PLC0415
required_filter_attributes = {ATTR_DEVICE_CLASS: device_class}
trigger_options = trigger_options or {}
unit_attributes = unit_attributes or {}
return [
*parametrize_trigger_states(
trigger=trigger,
trigger_options=trigger_options,
target_states=[
("0", unit_attributes),
("50", unit_attributes),
("100", unit_attributes),
],
other_states=[("none", unit_attributes)],
trigger_options={},
target_states=["0", "50", "100"],
other_states=["none"],
required_filter_attributes=required_filter_attributes,
retrigger_on_target_state=True,
trigger_from_none=False,
),
*parametrize_trigger_states(
trigger=trigger,
trigger_options={CONF_ABOVE: 10} | trigger_options,
target_states=[("50", unit_attributes), ("100", unit_attributes)],
other_states=[("none", unit_attributes), ("0", unit_attributes)],
trigger_options={CONF_ABOVE: 10},
target_states=["50", "100"],
other_states=["none", "0"],
required_filter_attributes=required_filter_attributes,
retrigger_on_target_state=True,
trigger_from_none=False,
),
*parametrize_trigger_states(
trigger=trigger,
trigger_options={CONF_BELOW: 90} | trigger_options,
target_states=[("0", unit_attributes), ("50", unit_attributes)],
other_states=[("none", unit_attributes), ("100", unit_attributes)],
trigger_options={CONF_BELOW: 90},
target_states=["0", "50"],
other_states=["none", "100"],
required_filter_attributes=required_filter_attributes,
retrigger_on_target_state=True,
trigger_from_none=False,
@@ -718,11 +678,7 @@ def parametrize_numerical_state_value_changed_trigger_states(
def parametrize_numerical_state_value_crossed_threshold_trigger_states(
trigger: str,
*,
device_class: str,
trigger_options: dict[str, Any] | None = None,
unit_attributes: dict | None = None,
trigger: str, device_class: str
) -> list[tuple[str, dict[str, Any], list[TriggerStateDescription]]]:
"""Parametrize states and expected service call counts for numerical state-value crossed threshold triggers.
@@ -733,9 +689,6 @@ def parametrize_numerical_state_value_crossed_threshold_trigger_states(
from homeassistant.const import ATTR_DEVICE_CLASS # noqa: PLC0415
required_filter_attributes = {ATTR_DEVICE_CLASS: device_class}
trigger_options = trigger_options or {}
unit_attributes = unit_attributes or {}
return [
*parametrize_trigger_states(
trigger=trigger,
@@ -743,14 +696,9 @@ def parametrize_numerical_state_value_crossed_threshold_trigger_states(
CONF_THRESHOLD_TYPE: ThresholdType.BETWEEN,
CONF_LOWER_LIMIT: 10,
CONF_UPPER_LIMIT: 90,
**trigger_options,
},
target_states=[("50", unit_attributes), ("60", unit_attributes)],
other_states=[
("none", unit_attributes),
("0", unit_attributes),
("100", unit_attributes),
],
target_states=["50", "60"],
other_states=["none", "0", "100"],
required_filter_attributes=required_filter_attributes,
trigger_from_none=False,
),
@@ -760,14 +708,9 @@ def parametrize_numerical_state_value_crossed_threshold_trigger_states(
CONF_THRESHOLD_TYPE: ThresholdType.OUTSIDE,
CONF_LOWER_LIMIT: 10,
CONF_UPPER_LIMIT: 90,
**trigger_options,
},
target_states=[("0", unit_attributes), ("100", unit_attributes)],
other_states=[
("none", unit_attributes),
("50", unit_attributes),
("60", unit_attributes),
],
target_states=["0", "100"],
other_states=["none", "50", "60"],
required_filter_attributes=required_filter_attributes,
trigger_from_none=False,
),
@@ -776,10 +719,9 @@ def parametrize_numerical_state_value_crossed_threshold_trigger_states(
trigger_options={
CONF_THRESHOLD_TYPE: ThresholdType.ABOVE,
CONF_LOWER_LIMIT: 10,
**trigger_options,
},
target_states=[("50", unit_attributes), ("100", unit_attributes)],
other_states=[("none", unit_attributes), ("0", unit_attributes)],
target_states=["50", "100"],
other_states=["none", "0"],
required_filter_attributes=required_filter_attributes,
trigger_from_none=False,
),
@@ -788,10 +730,9 @@ def parametrize_numerical_state_value_crossed_threshold_trigger_states(
trigger_options={
CONF_THRESHOLD_TYPE: ThresholdType.BELOW,
CONF_UPPER_LIMIT: 90,
**trigger_options,
},
target_states=[("0", unit_attributes), ("50", unit_attributes)],
other_states=[("none", unit_attributes), ("100", unit_attributes)],
target_states=["0", "50"],
other_states=["none", "100"],
required_filter_attributes=required_filter_attributes,
trigger_from_none=False,
),

View File

@@ -11,7 +11,6 @@ from homeassistant.components.climate import (
from homeassistant.components.humidifier import (
ATTR_CURRENT_HUMIDITY as HUMIDIFIER_ATTR_CURRENT_HUMIDITY,
)
from homeassistant.components.sensor import SensorDeviceClass
from homeassistant.components.weather import ATTR_WEATHER_HUMIDITY
from homeassistant.const import ATTR_DEVICE_CLASS, CONF_ENTITY_ID, STATE_ON
from homeassistant.core import HomeAssistant, ServiceCall
@@ -82,10 +81,10 @@ async def test_humidity_triggers_gated_by_labs_flag(
("trigger", "trigger_options", "states"),
[
*parametrize_numerical_state_value_changed_trigger_states(
"humidity.changed", device_class=SensorDeviceClass.HUMIDITY
"humidity.changed", "humidity"
),
*parametrize_numerical_state_value_crossed_threshold_trigger_states(
"humidity.crossed_threshold", device_class=SensorDeviceClass.HUMIDITY
"humidity.crossed_threshold", "humidity"
),
],
)
@@ -123,7 +122,7 @@ async def test_humidity_trigger_sensor_behavior_any(
("trigger", "trigger_options", "states"),
[
*parametrize_numerical_state_value_crossed_threshold_trigger_states(
"humidity.crossed_threshold", device_class=SensorDeviceClass.HUMIDITY
"humidity.crossed_threshold", "humidity"
),
],
)
@@ -161,7 +160,7 @@ async def test_humidity_trigger_sensor_crossed_threshold_behavior_first(
("trigger", "trigger_options", "states"),
[
*parametrize_numerical_state_value_crossed_threshold_trigger_states(
"humidity.crossed_threshold", device_class=SensorDeviceClass.HUMIDITY
"humidity.crossed_threshold", "humidity"
),
],
)

View File

@@ -1325,6 +1325,56 @@
'state': 'unknown',
})
# ---
# name: test_buttons[eve_shutter][button.eve_shutter_switch_20eci1701_sync_time-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'button',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'button.eve_shutter_switch_20eci1701_sync_time',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Sync time',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Sync time',
'platform': 'matter',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'sync_time',
'unique_id': '00000000000004D2-0000000000000094-MatterNodeDevice-0-TimeSynchronizationSyncTimeButton-56-0',
'unit_of_measurement': None,
})
# ---
# name: test_buttons[eve_shutter][button.eve_shutter_switch_20eci1701_sync_time-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Eve Shutter Switch 20ECI1701 Sync time',
}),
'context': <ANY>,
'entity_id': 'button.eve_shutter_switch_20eci1701_sync_time',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_buttons[eve_thermo_v4][button.eve_thermo_20ebp1701_identify-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
@@ -1376,6 +1426,56 @@
'state': 'unknown',
})
# ---
# name: test_buttons[eve_thermo_v4][button.eve_thermo_20ebp1701_sync_time-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'button',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'button.eve_thermo_20ebp1701_sync_time',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Sync time',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Sync time',
'platform': 'matter',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'sync_time',
'unique_id': '00000000000004D2-0000000000000021-MatterNodeDevice-0-TimeSynchronizationSyncTimeButton-56-0',
'unit_of_measurement': None,
})
# ---
# name: test_buttons[eve_thermo_v4][button.eve_thermo_20ebp1701_sync_time-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Eve Thermo 20EBP1701 Sync time',
}),
'context': <ANY>,
'entity_id': 'button.eve_thermo_20ebp1701_sync_time',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_buttons[eve_thermo_v5][button.eve_thermo_20ecd1701_identify-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
@@ -1427,6 +1527,56 @@
'state': 'unknown',
})
# ---
# name: test_buttons[eve_thermo_v5][button.eve_thermo_20ecd1701_sync_time-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'button',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'button.eve_thermo_20ecd1701_sync_time',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Sync time',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Sync time',
'platform': 'matter',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'sync_time',
'unique_id': '00000000000004D2-000000000000000C-MatterNodeDevice-0-TimeSynchronizationSyncTimeButton-56-0',
'unit_of_measurement': None,
})
# ---
# name: test_buttons[eve_thermo_v5][button.eve_thermo_20ecd1701_sync_time-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Eve Thermo 20ECD1701 Sync time',
}),
'context': <ANY>,
'entity_id': 'button.eve_thermo_20ecd1701_sync_time',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_buttons[eve_weather_sensor][button.eve_weather_identify_1-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
@@ -1935,6 +2085,56 @@
'state': 'unknown',
})
# ---
# name: test_buttons[ikea_air_quality_monitor][button.alpstuga_air_quality_monitor_sync_time-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'button',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'button.alpstuga_air_quality_monitor_sync_time',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Sync time',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Sync time',
'platform': 'matter',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'sync_time',
'unique_id': '00000000000004D2-0000000000000025-MatterNodeDevice-0-TimeSynchronizationSyncTimeButton-56-0',
'unit_of_measurement': None,
})
# ---
# name: test_buttons[ikea_air_quality_monitor][button.alpstuga_air_quality_monitor_sync_time-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'ALPSTUGA air quality monitor Sync time',
}),
'context': <ANY>,
'entity_id': 'button.alpstuga_air_quality_monitor_sync_time',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_buttons[inovelli_vtm30][button.white_series_onoff_switch_identify_load_control-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
@@ -2845,6 +3045,56 @@
'state': 'unknown',
})
# ---
# name: test_buttons[mock_leak_sensor][button.water_leak_detector_sync_time-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'button',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'button.water_leak_detector_sync_time',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Sync time',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Sync time',
'platform': 'matter',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'sync_time',
'unique_id': '00000000000004D2-0000000000000020-MatterNodeDevice-0-TimeSynchronizationSyncTimeButton-56-0',
'unit_of_measurement': None,
})
# ---
# name: test_buttons[mock_leak_sensor][button.water_leak_detector_sync_time-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Water Leak Detector Sync time',
}),
'context': <ANY>,
'entity_id': 'button.water_leak_detector_sync_time',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_buttons[mock_lock][button.mock_lock_identify-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
@@ -4211,6 +4461,56 @@
'state': 'unknown',
})
# ---
# name: test_buttons[silabs_light_switch][button.light_switch_example_sync_time-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'button',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'button.light_switch_example_sync_time',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Sync time',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Sync time',
'platform': 'matter',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'sync_time',
'unique_id': '00000000000004D2-000000000000008E-MatterNodeDevice-0-TimeSynchronizationSyncTimeButton-56-0',
'unit_of_measurement': None,
})
# ---
# name: test_buttons[silabs_light_switch][button.light_switch_example_sync_time-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Light switch example Sync time',
}),
'context': <ANY>,
'entity_id': 'button.light_switch_example_sync_time',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_buttons[silabs_refrigerator][button.refrigerator_identify-entry]
EntityRegistryEntrySnapshot({
'aliases': list([

View File

@@ -1,8 +1,10 @@
"""Test Matter switches."""
"""Test Matter buttons."""
from datetime import UTC, datetime
from unittest.mock import MagicMock, call
from chip.clusters import Objects as clusters
from chip.clusters.Types import NullValue
from matter_server.client.models.node import MatterNode
import pytest
from syrupy.assertion import SnapshotAssertion
@@ -10,6 +12,7 @@ from syrupy.assertion import SnapshotAssertion
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.util import dt as dt_util
from .common import snapshot_matter_entities
@@ -107,3 +110,82 @@ async def test_smoke_detector_self_test(
endpoint_id=1,
command=clusters.SmokeCoAlarm.Commands.SelfTestRequest(),
)
@pytest.mark.freeze_time("2025-06-15T12:00:00+00:00")
@pytest.mark.parametrize("node_fixture", ["ikea_air_quality_monitor"])
async def test_time_sync_button(
hass: HomeAssistant,
matter_client: MagicMock,
matter_node: MatterNode,
) -> None:
"""Test button entity is created for a Matter TimeSynchronization Cluster."""
entity_id = "button.alpstuga_air_quality_monitor_sync_time"
state = hass.states.get(entity_id)
assert state
assert state.attributes["friendly_name"] == "ALPSTUGA air quality monitor Sync time"
# test press action
await hass.services.async_call(
"button",
"press",
{
"entity_id": entity_id,
},
blocking=True,
)
assert matter_client.send_device_command.call_count == 3
# Compute expected values based on HA's configured timezone
chip_epoch = datetime(2000, 1, 1, tzinfo=UTC)
frozen_now = datetime(2025, 6, 15, 12, 0, 0, tzinfo=UTC)
delta = frozen_now - chip_epoch
expected_utc_us = (
(delta.days * 86400 * 1_000_000)
+ (delta.seconds * 1_000_000)
+ delta.microseconds
)
ha_tz = dt_util.get_default_time_zone()
local_now = frozen_now.astimezone(ha_tz)
utc_offset_delta = local_now.utcoffset()
utc_offset = int(utc_offset_delta.total_seconds()) if utc_offset_delta else 0
dst_offset_delta = local_now.dst()
dst_offset = int(dst_offset_delta.total_seconds()) if dst_offset_delta else 0
standard_offset = utc_offset - dst_offset
# Verify SetTimeZone command
assert matter_client.send_device_command.call_args_list[0] == call(
node_id=matter_node.node_id,
endpoint_id=0,
command=clusters.TimeSynchronization.Commands.SetTimeZone(
timeZone=[
clusters.TimeSynchronization.Structs.TimeZoneStruct(
offset=standard_offset,
validAt=0,
name=str(ha_tz),
)
]
),
)
# Verify SetDSTOffset command
assert matter_client.send_device_command.call_args_list[1] == call(
node_id=matter_node.node_id,
endpoint_id=0,
command=clusters.TimeSynchronization.Commands.SetDSTOffset(
DSTOffset=[
clusters.TimeSynchronization.Structs.DSTOffsetStruct(
offset=dst_offset,
validStarting=0,
validUntil=NullValue,
)
]
),
)
# Verify SetUTCTime command
assert matter_client.send_device_command.call_args_list[2] == call(
node_id=matter_node.node_id,
endpoint_id=0,
command=clusters.TimeSynchronization.Commands.SetUTCTime(
UTCTime=expected_utc_us,
granularity=clusters.TimeSynchronization.Enums.GranularityEnum.kMicrosecondsGranularity,
),
)

View File

@@ -74,6 +74,7 @@ DEVICE_FIXTURES = [
"da_wm_dw_01011",
"da_rvc_normal_000001",
"da_rvc_map_01011",
"da_vc_stick_01001",
"da_ks_microwave_0101x",
"da_ks_cooktop_000001",
"da_ks_cooktop_31001",

View File

@@ -0,0 +1,426 @@
{
"components": {
"station": {
"samsungce.cleanStationStickStatus": {
"status": {
"value": "attached",
"timestamp": "2026-03-21T15:25:21.619Z"
}
},
"custom.disabledCapabilities": {
"disabledCapabilities": {
"value": ["samsungce.cleanStationUvCleaning"],
"timestamp": "2026-03-21T14:44:08.042Z"
}
},
"samsungce.cleanStationUvCleaning": {
"operatingState": {
"value": "ready",
"timestamp": "2026-03-21T14:44:06.339Z"
},
"lastFinishedTime": {
"value": null
},
"uvcIntensive": {
"value": null
},
"operationTime": {
"value": null
},
"remainingTime": {
"value": null
}
},
"samsungce.stickCleanerDustBag": {
"supportedStatus": {
"value": ["full", "normal"],
"timestamp": "2026-03-21T14:44:06.339Z"
},
"usage": {
"value": 343,
"timestamp": "2026-03-21T15:25:35.951Z"
},
"status": {
"value": "normal",
"timestamp": "2026-03-21T14:44:06.339Z"
}
}
},
"main": {
"custom.disabledComponents": {
"disabledComponents": {
"value": [],
"timestamp": "2026-03-21T14:44:06.339Z"
}
},
"powerConsumptionReport": {
"powerConsumption": {
"value": {
"energy": 4,
"deltaEnergy": 3,
"power": 0,
"powerEnergy": 0.0,
"persistedEnergy": 0,
"energySaved": 0,
"start": "2026-03-21T15:35:31Z",
"end": "2026-03-21T15:41:41Z"
},
"timestamp": "2026-03-21T15:41:41.889Z"
}
},
"samsungce.stickCleanerStatus": {
"operatingState": {
"value": "ready",
"timestamp": "2026-03-21T15:25:35.978Z"
}
},
"refresh": {},
"samsungce.notification": {
"supportedActionSettings": {
"value": [
{
"action": "stop",
"supportedSettings": ["on", "off"]
}
],
"timestamp": "2026-03-21T14:44:06.339Z"
},
"actionSetting": {
"value": {
"stop": {
"setting": "on"
}
},
"timestamp": "2026-03-21T14:50:31.556Z"
},
"supportedContexts": {
"value": ["incomingCall", "messageReceived"],
"timestamp": "2026-03-21T14:44:06.339Z"
},
"supportCustomContent": {
"value": false,
"timestamp": "2026-03-21T14:44:06.339Z"
}
},
"samsungce.stickCleanerDustbinStatus": {
"operatingState": {
"value": "ready",
"timestamp": "2026-03-21T15:25:35.978Z"
},
"lastEmptiedTime": {
"value": "2026-03-21T15:25:00Z",
"timestamp": "2026-03-21T15:25:35.978Z"
}
},
"battery": {
"quantity": {
"value": null
},
"battery": {
"value": 80,
"unit": "%",
"timestamp": "2026-03-21T15:41:41.889Z"
},
"type": {
"value": null
}
},
"execute": {
"data": {
"value": null
}
},
"samsungce.deviceIdentification": {
"micomAssayCode": {
"value": "50025842",
"timestamp": "2026-03-21T14:44:06.339Z"
},
"modelName": {
"value": null
},
"serialNumber": {
"value": null
},
"serialNumberExtra": {
"value": null
},
"releaseCountry": {
"value": null
},
"modelClassificationCode": {
"value": "80030200001711000802000000000000",
"timestamp": "2026-03-21T14:44:06.339Z"
},
"description": {
"value": "A-VSWW-TP1-23-VS9700",
"timestamp": "2026-03-21T14:44:06.339Z"
},
"releaseYear": {
"value": null
},
"binaryId": {
"value": "A-VSWW-TP1-23-VS9700",
"timestamp": "2026-03-21T15:41:41.888Z"
}
},
"sec.wifiConfiguration": {
"autoReconnection": {
"value": true,
"timestamp": "2026-03-21T14:44:06.339Z"
},
"minVersion": {
"value": "1.0",
"timestamp": "2026-03-21T14:44:06.339Z"
},
"supportedWiFiFreq": {
"value": ["2.4G"],
"timestamp": "2026-03-21T14:44:06.339Z"
},
"supportedAuthType": {
"value": ["OPEN", "WEP", "WPA-PSK", "WPA2-PSK", "SAE"],
"timestamp": "2026-03-21T14:44:06.339Z"
},
"protocolType": {
"value": ["helper_hotspot", "ble_ocf"],
"timestamp": "2026-03-21T14:44:06.339Z"
}
},
"samsungce.selfCheck": {
"result": {
"value": "failed",
"timestamp": "2026-03-21T15:25:48.370Z"
},
"supportedActions": {
"value": ["start", "cancel"],
"timestamp": "2026-03-21T14:44:06.339Z"
},
"progress": {
"value": 100,
"unit": "%",
"timestamp": "2026-03-21T15:25:48.370Z"
},
"errors": {
"value": [
{
"code": "DA_VCS_E_001"
}
],
"timestamp": "2026-03-21T15:25:48.370Z"
},
"status": {
"value": "ready",
"timestamp": "2026-03-21T15:25:48.370Z"
}
},
"samsungce.softwareVersion": {
"versions": {
"value": [
{
"id": "0",
"swType": "Software",
"versionNumber": "25051400",
"description": "Version"
},
{
"id": "1",
"swType": "Firmware",
"versionNumber": "00258B23110300",
"description": "Version"
},
{
"id": "2",
"swType": "Firmware",
"versionNumber": "00253B23070700",
"description": "Version"
}
],
"timestamp": "2026-03-21T14:50:31.808Z"
}
},
"ocf": {
"st": {
"value": null
},
"mndt": {
"value": null
},
"mnfv": {
"value": "A-VSWW-TP1-23-VS9700_51250514",
"timestamp": "2026-03-21T14:50:07.585Z"
},
"mnhw": {
"value": "Realtek",
"timestamp": "2026-03-21T14:50:07.585Z"
},
"di": {
"value": "e1f93c0c-6fe0-c65a-a314-c8f7b163c86b",
"timestamp": "2026-03-21T14:50:07.585Z"
},
"mnsl": {
"value": "http://www.samsung.com",
"timestamp": "2026-03-21T14:50:07.585Z"
},
"dmv": {
"value": "1.2.1",
"timestamp": "2026-03-21T14:50:07.585Z"
},
"n": {
"value": "[vacuum] Samsung",
"timestamp": "2026-03-21T14:50:07.585Z"
},
"mnmo": {
"value": "A-VSWW-TP1-23-VS9700|50025842|80030200001711000802000000000000",
"timestamp": "2026-03-21T14:50:07.585Z"
},
"vid": {
"value": "DA-VC-STICK-01001",
"timestamp": "2026-03-21T14:50:07.585Z"
},
"mnmn": {
"value": "Samsung Electronics",
"timestamp": "2026-03-21T14:50:07.585Z"
},
"mnml": {
"value": "http://www.samsung.com",
"timestamp": "2026-03-21T14:50:07.585Z"
},
"mnpv": {
"value": "SYSTEM 2.0",
"timestamp": "2026-03-21T14:50:07.585Z"
},
"mnos": {
"value": "TizenRT 4.0",
"timestamp": "2026-03-21T14:50:07.585Z"
},
"pi": {
"value": "e1f93c0c-6fe0-c65a-a314-c8f7b163c86b",
"timestamp": "2026-03-21T14:50:07.585Z"
},
"icv": {
"value": "core.1.1.0",
"timestamp": "2026-03-21T14:50:07.585Z"
}
},
"samsungce.stickCleanerStickStatus": {
"mode": {
"value": "none",
"timestamp": "2026-03-21T15:25:06.593Z"
},
"status": {
"value": "charging",
"timestamp": "2026-03-21T15:25:22.052Z"
},
"bleConnectionState": {
"value": "connected",
"timestamp": "2026-03-21T14:50:29.775Z"
}
},
"custom.disabledCapabilities": {
"disabledCapabilities": {
"value": ["sec.wifiConfiguration"],
"timestamp": "2026-03-21T14:44:06.339Z"
}
},
"samsungce.driverVersion": {
"versionNumber": {
"value": 25040101,
"timestamp": "2026-03-21T14:44:06.339Z"
}
},
"samsungce.softwareUpdate": {
"targetModule": {
"value": {},
"timestamp": "2026-03-21T14:44:08.245Z"
},
"otnDUID": {
"value": "BDCPH4AI7GMCS",
"timestamp": "2026-03-21T14:44:06.339Z"
},
"lastUpdatedDate": {
"value": null
},
"availableModules": {
"value": [],
"timestamp": "2026-03-21T14:44:06.339Z"
},
"newVersionAvailable": {
"value": false,
"timestamp": "2026-03-21T14:44:06.339Z"
},
"operatingState": {
"value": "none",
"timestamp": "2026-03-21T14:44:08.245Z"
},
"progress": {
"value": 0,
"unit": "%",
"timestamp": "2026-03-21T14:50:09.045Z"
}
},
"sec.diagnosticsInformation": {
"logType": {
"value": ["errCode", "dump"],
"timestamp": "2026-03-21T14:44:06.339Z"
},
"endpoint": {
"value": "SSM",
"timestamp": "2026-03-21T14:44:06.339Z"
},
"minVersion": {
"value": "3.0",
"timestamp": "2026-03-21T14:44:06.339Z"
},
"signinPermission": {
"value": null
},
"setupId": {
"value": "VS2",
"timestamp": "2026-03-21T14:44:06.339Z"
},
"protocolType": {
"value": "ble_ocf",
"timestamp": "2026-03-21T14:44:06.339Z"
},
"tsId": {
"value": "DA01",
"timestamp": "2026-03-21T14:44:06.339Z"
},
"mnId": {
"value": "0AJT",
"timestamp": "2026-03-21T14:44:06.339Z"
},
"dumpType": {
"value": "file",
"timestamp": "2026-03-21T14:44:06.339Z"
}
},
"custom.deviceReportStateConfiguration": {
"reportStateRealtimePeriod": {
"value": null
},
"reportStateRealtime": {
"value": {
"state": "enabled",
"duration": 10,
"unit": "minute"
},
"timestamp": "2026-03-21T14:44:42.985Z"
},
"reportStatePeriod": {
"value": "enabled",
"timestamp": "2026-03-21T14:44:06.339Z"
}
},
"samsungce.lamp": {
"brightnessLevel": {
"value": "on",
"timestamp": "2026-03-21T14:44:06.339Z"
},
"supportedBrightnessLevel": {
"value": ["on", "off"],
"timestamp": "2026-03-21T14:44:06.339Z"
}
}
}
}
}

View File

@@ -0,0 +1,168 @@
{
"items": [
{
"deviceId": "e1f93c0c-6fe0-c65a-a314-c8f7b163c86b",
"name": "[vacuum] Samsung",
"label": "Stick vacuum",
"manufacturerName": "Samsung Electronics",
"presentationId": "DA-VC-STICK-01001",
"deviceManufacturerCode": "Samsung Electronics",
"locationId": "03f25476-ce87-4f94-b153-03d40451dee0",
"ownerId": "62619912-9710-ee72-bdf7-6e3910560913",
"roomId": "2f820695-73c1-4d43-8ee9-7c6a07feeb9a",
"deviceTypeName": "x.com.st.d.stickcleaner",
"components": [
{
"id": "main",
"label": "main",
"capabilities": [
{
"id": "ocf",
"version": 1
},
{
"id": "execute",
"version": 1
},
{
"id": "powerConsumptionReport",
"version": 1
},
{
"id": "refresh",
"version": 1
},
{
"id": "samsungce.deviceIdentification",
"version": 1
},
{
"id": "samsungce.driverVersion",
"version": 1
},
{
"id": "samsungce.stickCleanerStickStatus",
"version": 1
},
{
"id": "battery",
"version": 1
},
{
"id": "samsungce.lamp",
"version": 1
},
{
"id": "samsungce.notification",
"version": 1
},
{
"id": "samsungce.selfCheck",
"version": 1
},
{
"id": "samsungce.stickCleanerDustbinStatus",
"version": 1
},
{
"id": "samsungce.softwareUpdate",
"version": 1
},
{
"id": "samsungce.softwareVersion",
"version": 1
},
{
"id": "samsungce.stickCleanerStatus",
"version": 1
},
{
"id": "custom.deviceReportStateConfiguration",
"version": 1
},
{
"id": "custom.disabledComponents",
"version": 1
},
{
"id": "custom.disabledCapabilities",
"version": 1
},
{
"id": "sec.diagnosticsInformation",
"version": 1
},
{
"id": "sec.wifiConfiguration",
"version": 1
}
],
"categories": [
{
"name": "StickVacuumCleaner",
"categoryType": "manufacturer"
}
],
"optional": false
},
{
"id": "station",
"label": "station",
"capabilities": [
{
"id": "samsungce.stickCleanerDustBag",
"version": 1
},
{
"id": "samsungce.cleanStationStickStatus",
"version": 1
},
{
"id": "samsungce.cleanStationUvCleaning",
"version": 1
},
{
"id": "custom.disabledCapabilities",
"version": 1
}
],
"categories": [
{
"name": "Other",
"categoryType": "manufacturer"
}
],
"optional": false
}
],
"createTime": "2026-03-21T14:43:55.855Z",
"profile": {
"id": "21c15481-d69b-34a9-86a6-bcdb478a68cb"
},
"ocf": {
"ocfDeviceType": "x.com.st.d.stickcleaner",
"name": "[vacuum] Samsung",
"specVersion": "core.1.1.0",
"verticalDomainSpecVersion": "1.2.1",
"manufacturerName": "Samsung Electronics",
"modelNumber": "A-VSWW-TP1-23-VS9700|50025842|80030200001711000802000000000000",
"platformVersion": "SYSTEM 2.0",
"platformOS": "TizenRT 4.0",
"hwVersion": "Realtek",
"firmwareVersion": "A-VSWW-TP1-23-VS9700_51250514",
"vendorId": "DA-VC-STICK-01001",
"vendorResourceClientServerVersion": "Realtek Release 250514",
"lastSignupTime": "2026-03-21T14:43:55.796850354Z",
"transferCandidate": true,
"additionalAuthCodeRequired": false,
"modelCode": "VS28C9784QK/WA"
},
"type": "OCF",
"restrictionTier": 0,
"allowed": null,
"executionContext": "CLOUD",
"relationships": []
}
],
"_links": {}
}

View File

@@ -1239,6 +1239,37 @@
'via_device_id': None,
})
# ---
# name: test_devices[da_vc_stick_01001]
DeviceRegistryEntrySnapshot({
'area_id': None,
'config_entries': <ANY>,
'config_entries_subentries': <ANY>,
'configuration_url': 'https://account.smartthings.com',
'connections': set({
}),
'disabled_by': None,
'entry_type': None,
'hw_version': 'Realtek',
'id': <ANY>,
'identifiers': set({
tuple(
'smartthings',
'e1f93c0c-6fe0-c65a-a314-c8f7b163c86b',
),
}),
'labels': set({
}),
'manufacturer': 'Samsung Electronics',
'model': 'A-VSWW-TP1-23-VS9700',
'model_id': 'VS28C9784QK/WA',
'name': 'Stick vacuum',
'name_by_user': None,
'primary_config_entry': <ANY>,
'serial_number': None,
'sw_version': 'A-VSWW-TP1-23-VS9700_51250514',
'via_device_id': None,
})
# ---
# name: test_devices[da_wm_dw_000001]
DeviceRegistryEntrySnapshot({
'area_id': 'theater',

View File

@@ -851,6 +851,65 @@
'state': 'medium',
})
# ---
# name: test_all_entities[da_vc_stick_01001][select.stick_vacuum_lamp-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
'options': list([
'on',
'off',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'select',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'select.stick_vacuum_lamp',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Lamp',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Lamp',
'platform': 'smartthings',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'lamp',
'unique_id': 'e1f93c0c-6fe0-c65a-a314-c8f7b163c86b_main_samsungce.lamp_brightnessLevel_brightnessLevel',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[da_vc_stick_01001][select.stick_vacuum_lamp-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Stick vacuum Lamp',
'options': list([
'on',
'off',
]),
}),
'context': <ANY>,
'entity_id': 'select.stick_vacuum_lamp',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
# name: test_all_entities[da_wm_dw_000001][select.dishwasher-entry]
EntityRegistryEntrySnapshot({
'aliases': list([

View File

@@ -13340,6 +13340,350 @@
'state': 'room',
})
# ---
# name: test_all_entities[da_vc_stick_01001][sensor.stick_vacuum_battery-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.stick_vacuum_battery',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Battery',
'options': dict({
}),
'original_device_class': <SensorDeviceClass.BATTERY: 'battery'>,
'original_icon': None,
'original_name': 'Battery',
'platform': 'smartthings',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'e1f93c0c-6fe0-c65a-a314-c8f7b163c86b_main_battery_battery_battery',
'unit_of_measurement': '%',
})
# ---
# name: test_all_entities[da_vc_stick_01001][sensor.stick_vacuum_battery-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'battery',
'friendly_name': 'Stick vacuum Battery',
'unit_of_measurement': '%',
}),
'context': <ANY>,
'entity_id': 'sensor.stick_vacuum_battery',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '80',
})
# ---
# name: test_all_entities[da_vc_stick_01001][sensor.stick_vacuum_energy-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.TOTAL_INCREASING: 'total_increasing'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.stick_vacuum_energy',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Energy',
'options': dict({
'sensor': dict({
'suggested_display_precision': 2,
}),
}),
'original_device_class': <SensorDeviceClass.ENERGY: 'energy'>,
'original_icon': None,
'original_name': 'Energy',
'platform': 'smartthings',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'e1f93c0c-6fe0-c65a-a314-c8f7b163c86b_main_powerConsumptionReport_powerConsumption_energy_meter',
'unit_of_measurement': <UnitOfEnergy.KILO_WATT_HOUR: 'kWh'>,
})
# ---
# name: test_all_entities[da_vc_stick_01001][sensor.stick_vacuum_energy-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'energy',
'friendly_name': 'Stick vacuum Energy',
'state_class': <SensorStateClass.TOTAL_INCREASING: 'total_increasing'>,
'unit_of_measurement': <UnitOfEnergy.KILO_WATT_HOUR: 'kWh'>,
}),
'context': <ANY>,
'entity_id': 'sensor.stick_vacuum_energy',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0.004',
})
# ---
# name: test_all_entities[da_vc_stick_01001][sensor.stick_vacuum_energy_difference-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.TOTAL: 'total'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.stick_vacuum_energy_difference',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Energy difference',
'options': dict({
'sensor': dict({
'suggested_display_precision': 2,
}),
}),
'original_device_class': <SensorDeviceClass.ENERGY: 'energy'>,
'original_icon': None,
'original_name': 'Energy difference',
'platform': 'smartthings',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'energy_difference',
'unique_id': 'e1f93c0c-6fe0-c65a-a314-c8f7b163c86b_main_powerConsumptionReport_powerConsumption_deltaEnergy_meter',
'unit_of_measurement': <UnitOfEnergy.KILO_WATT_HOUR: 'kWh'>,
})
# ---
# name: test_all_entities[da_vc_stick_01001][sensor.stick_vacuum_energy_difference-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'energy',
'friendly_name': 'Stick vacuum Energy difference',
'state_class': <SensorStateClass.TOTAL: 'total'>,
'unit_of_measurement': <UnitOfEnergy.KILO_WATT_HOUR: 'kWh'>,
}),
'context': <ANY>,
'entity_id': 'sensor.stick_vacuum_energy_difference',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0.003',
})
# ---
# name: test_all_entities[da_vc_stick_01001][sensor.stick_vacuum_energy_saved-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.TOTAL_INCREASING: 'total_increasing'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.stick_vacuum_energy_saved',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Energy saved',
'options': dict({
'sensor': dict({
'suggested_display_precision': 2,
}),
}),
'original_device_class': <SensorDeviceClass.ENERGY: 'energy'>,
'original_icon': None,
'original_name': 'Energy saved',
'platform': 'smartthings',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'energy_saved',
'unique_id': 'e1f93c0c-6fe0-c65a-a314-c8f7b163c86b_main_powerConsumptionReport_powerConsumption_energySaved_meter',
'unit_of_measurement': <UnitOfEnergy.KILO_WATT_HOUR: 'kWh'>,
})
# ---
# name: test_all_entities[da_vc_stick_01001][sensor.stick_vacuum_energy_saved-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'energy',
'friendly_name': 'Stick vacuum Energy saved',
'state_class': <SensorStateClass.TOTAL_INCREASING: 'total_increasing'>,
'unit_of_measurement': <UnitOfEnergy.KILO_WATT_HOUR: 'kWh'>,
}),
'context': <ANY>,
'entity_id': 'sensor.stick_vacuum_energy_saved',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0.0',
})
# ---
# name: test_all_entities[da_vc_stick_01001][sensor.stick_vacuum_power-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.stick_vacuum_power',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Power',
'options': dict({
'sensor': dict({
'suggested_display_precision': 2,
}),
}),
'original_device_class': <SensorDeviceClass.POWER: 'power'>,
'original_icon': None,
'original_name': 'Power',
'platform': 'smartthings',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'e1f93c0c-6fe0-c65a-a314-c8f7b163c86b_main_powerConsumptionReport_powerConsumption_power_meter',
'unit_of_measurement': <UnitOfPower.WATT: 'W'>,
})
# ---
# name: test_all_entities[da_vc_stick_01001][sensor.stick_vacuum_power-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'power',
'friendly_name': 'Stick vacuum Power',
'power_consumption_end': '2026-03-21T15:41:41Z',
'power_consumption_start': '2026-03-21T15:35:31Z',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': <UnitOfPower.WATT: 'W'>,
}),
'context': <ANY>,
'entity_id': 'sensor.stick_vacuum_power',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0',
})
# ---
# name: test_all_entities[da_vc_stick_01001][sensor.stick_vacuum_power_energy-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.TOTAL: 'total'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.stick_vacuum_power_energy',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Power energy',
'options': dict({
'sensor': dict({
'suggested_display_precision': 2,
}),
}),
'original_device_class': <SensorDeviceClass.ENERGY: 'energy'>,
'original_icon': None,
'original_name': 'Power energy',
'platform': 'smartthings',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'power_energy',
'unique_id': 'e1f93c0c-6fe0-c65a-a314-c8f7b163c86b_main_powerConsumptionReport_powerConsumption_powerEnergy_meter',
'unit_of_measurement': <UnitOfEnergy.KILO_WATT_HOUR: 'kWh'>,
})
# ---
# name: test_all_entities[da_vc_stick_01001][sensor.stick_vacuum_power_energy-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'energy',
'friendly_name': 'Stick vacuum Power energy',
'state_class': <SensorStateClass.TOTAL: 'total'>,
'unit_of_measurement': <UnitOfEnergy.KILO_WATT_HOUR: 'kWh'>,
}),
'context': <ANY>,
'entity_id': 'sensor.stick_vacuum_power_energy',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0.0',
})
# ---
# name: test_all_entities[da_wm_dw_000001][sensor.dishwasher_completion_time-entry]
EntityRegistryEntrySnapshot({
'aliases': list([

View File

@@ -1 +0,0 @@
"""Tests for the temperature integration."""

View File

@@ -1,931 +0,0 @@
"""Test temperature trigger."""
from typing import Any
import pytest
from homeassistant.components.climate import (
ATTR_CURRENT_TEMPERATURE as CLIMATE_ATTR_CURRENT_TEMPERATURE,
HVACMode,
)
from homeassistant.components.sensor import SensorDeviceClass
from homeassistant.components.water_heater import (
ATTR_CURRENT_TEMPERATURE as WATER_HEATER_ATTR_CURRENT_TEMPERATURE,
)
from homeassistant.components.weather import (
ATTR_WEATHER_TEMPERATURE,
ATTR_WEATHER_TEMPERATURE_UNIT,
)
from homeassistant.const import (
ATTR_DEVICE_CLASS,
ATTR_UNIT_OF_MEASUREMENT,
CONF_ENTITY_ID,
UnitOfTemperature,
)
from homeassistant.core import HomeAssistant, ServiceCall
from tests.components.common import (
TriggerStateDescription,
arm_trigger,
assert_trigger_behavior_any,
assert_trigger_behavior_first,
assert_trigger_behavior_last,
assert_trigger_gated_by_labs_flag,
parametrize_numerical_attribute_changed_trigger_states,
parametrize_numerical_attribute_crossed_threshold_trigger_states,
parametrize_numerical_state_value_changed_trigger_states,
parametrize_numerical_state_value_crossed_threshold_trigger_states,
parametrize_target_entities,
target_entities,
)
_TEMPERATURE_TRIGGER_OPTIONS = {"unit": UnitOfTemperature.CELSIUS}
_SENSOR_UNIT_ATTRIBUTES = {
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.CELSIUS,
}
_WEATHER_UNIT_ATTRIBUTES = {
ATTR_WEATHER_TEMPERATURE_UNIT: UnitOfTemperature.CELSIUS,
}
@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_climates(hass: HomeAssistant) -> dict[str, list[str]]:
"""Create multiple climate entities associated with different targets."""
return await target_entities(hass, "climate")
@pytest.fixture
async def target_water_heaters(hass: HomeAssistant) -> dict[str, list[str]]:
"""Create multiple water_heater entities associated with different targets."""
return await target_entities(hass, "water_heater")
@pytest.fixture
async def target_weathers(hass: HomeAssistant) -> dict[str, list[str]]:
"""Create multiple weather entities associated with different targets."""
return await target_entities(hass, "weather")
@pytest.mark.parametrize(
"trigger_key",
[
"temperature.changed",
"temperature.crossed_threshold",
],
)
async def test_temperature_triggers_gated_by_labs_flag(
hass: HomeAssistant, caplog: pytest.LogCaptureFixture, trigger_key: str
) -> None:
"""Test the temperature triggers are gated by the labs flag."""
await assert_trigger_gated_by_labs_flag(hass, caplog, trigger_key)
# --- Sensor domain tests (value in state.state) ---
@pytest.mark.usefixtures("enable_labs_preview_features")
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("sensor"),
)
@pytest.mark.parametrize(
("trigger", "trigger_options", "states"),
[
*parametrize_numerical_state_value_changed_trigger_states(
"temperature.changed",
device_class=SensorDeviceClass.TEMPERATURE,
trigger_options=_TEMPERATURE_TRIGGER_OPTIONS,
unit_attributes=_SENSOR_UNIT_ATTRIBUTES,
),
*parametrize_numerical_state_value_crossed_threshold_trigger_states(
"temperature.crossed_threshold",
device_class=SensorDeviceClass.TEMPERATURE,
trigger_options=_TEMPERATURE_TRIGGER_OPTIONS,
unit_attributes=_SENSOR_UNIT_ATTRIBUTES,
),
],
)
async def test_temperature_trigger_sensor_behavior_any(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_sensors: dict[str, list[str]],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
trigger_options: dict[str, Any],
states: list[TriggerStateDescription],
) -> None:
"""Test temperature trigger fires for sensor entities with device_class temperature."""
await assert_trigger_behavior_any(
hass,
service_calls=service_calls,
target_entities=target_sensors,
trigger_target_config=trigger_target_config,
entity_id=entity_id,
entities_in_target=entities_in_target,
trigger=trigger,
trigger_options=trigger_options,
states=states,
)
@pytest.mark.usefixtures("enable_labs_preview_features")
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("sensor"),
)
@pytest.mark.parametrize(
("trigger", "trigger_options", "states"),
[
*parametrize_numerical_state_value_crossed_threshold_trigger_states(
"temperature.crossed_threshold",
device_class=SensorDeviceClass.TEMPERATURE,
trigger_options=_TEMPERATURE_TRIGGER_OPTIONS,
unit_attributes=_SENSOR_UNIT_ATTRIBUTES,
),
],
)
async def test_temperature_trigger_sensor_crossed_threshold_behavior_first(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_sensors: dict[str, list[str]],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
trigger_options: dict[str, Any],
states: list[TriggerStateDescription],
) -> None:
"""Test temperature crossed_threshold trigger fires on the first sensor state change."""
await assert_trigger_behavior_first(
hass,
service_calls=service_calls,
target_entities=target_sensors,
trigger_target_config=trigger_target_config,
entity_id=entity_id,
entities_in_target=entities_in_target,
trigger=trigger,
trigger_options=trigger_options,
states=states,
)
@pytest.mark.usefixtures("enable_labs_preview_features")
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("sensor"),
)
@pytest.mark.parametrize(
("trigger", "trigger_options", "states"),
[
*parametrize_numerical_state_value_crossed_threshold_trigger_states(
"temperature.crossed_threshold",
device_class=SensorDeviceClass.TEMPERATURE,
trigger_options=_TEMPERATURE_TRIGGER_OPTIONS,
unit_attributes=_SENSOR_UNIT_ATTRIBUTES,
),
],
)
async def test_temperature_trigger_sensor_crossed_threshold_behavior_last(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_sensors: dict[str, list[str]],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
trigger_options: dict[str, Any],
states: list[TriggerStateDescription],
) -> None:
"""Test temperature crossed_threshold trigger fires when the last sensor changes state."""
await assert_trigger_behavior_last(
hass,
service_calls=service_calls,
target_entities=target_sensors,
trigger_target_config=trigger_target_config,
entity_id=entity_id,
entities_in_target=entities_in_target,
trigger=trigger,
trigger_options=trigger_options,
states=states,
)
# --- Climate domain tests (value in current_temperature attribute) ---
@pytest.mark.usefixtures("enable_labs_preview_features")
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("climate"),
)
@pytest.mark.parametrize(
("trigger", "trigger_options", "states"),
[
*parametrize_numerical_attribute_changed_trigger_states(
"temperature.changed",
HVACMode.AUTO,
CLIMATE_ATTR_CURRENT_TEMPERATURE,
trigger_options=_TEMPERATURE_TRIGGER_OPTIONS,
),
*parametrize_numerical_attribute_crossed_threshold_trigger_states(
"temperature.crossed_threshold",
HVACMode.AUTO,
CLIMATE_ATTR_CURRENT_TEMPERATURE,
trigger_options=_TEMPERATURE_TRIGGER_OPTIONS,
),
],
)
async def test_temperature_trigger_climate_behavior_any(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_climates: dict[str, list[str]],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
trigger_options: dict[str, Any],
states: list[TriggerStateDescription],
) -> None:
"""Test temperature trigger fires for climate entities."""
await assert_trigger_behavior_any(
hass,
service_calls=service_calls,
target_entities=target_climates,
trigger_target_config=trigger_target_config,
entity_id=entity_id,
entities_in_target=entities_in_target,
trigger=trigger,
trigger_options=trigger_options,
states=states,
)
@pytest.mark.usefixtures("enable_labs_preview_features")
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("climate"),
)
@pytest.mark.parametrize(
("trigger", "trigger_options", "states"),
[
*parametrize_numerical_attribute_crossed_threshold_trigger_states(
"temperature.crossed_threshold",
HVACMode.AUTO,
CLIMATE_ATTR_CURRENT_TEMPERATURE,
trigger_options=_TEMPERATURE_TRIGGER_OPTIONS,
),
],
)
async def test_temperature_trigger_climate_crossed_threshold_behavior_first(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_climates: dict[str, list[str]],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
trigger_options: dict[str, Any],
states: list[TriggerStateDescription],
) -> None:
"""Test temperature crossed_threshold trigger fires on the first climate state change."""
await assert_trigger_behavior_first(
hass,
service_calls=service_calls,
target_entities=target_climates,
trigger_target_config=trigger_target_config,
entity_id=entity_id,
entities_in_target=entities_in_target,
trigger=trigger,
trigger_options=trigger_options,
states=states,
)
@pytest.mark.usefixtures("enable_labs_preview_features")
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("climate"),
)
@pytest.mark.parametrize(
("trigger", "trigger_options", "states"),
[
*parametrize_numerical_attribute_crossed_threshold_trigger_states(
"temperature.crossed_threshold",
HVACMode.AUTO,
CLIMATE_ATTR_CURRENT_TEMPERATURE,
trigger_options=_TEMPERATURE_TRIGGER_OPTIONS,
),
],
)
async def test_temperature_trigger_climate_crossed_threshold_behavior_last(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_climates: dict[str, list[str]],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
trigger_options: dict[str, Any],
states: list[TriggerStateDescription],
) -> None:
"""Test temperature crossed_threshold trigger fires when the last climate changes state."""
await assert_trigger_behavior_last(
hass,
service_calls=service_calls,
target_entities=target_climates,
trigger_target_config=trigger_target_config,
entity_id=entity_id,
entities_in_target=entities_in_target,
trigger=trigger,
trigger_options=trigger_options,
states=states,
)
# --- Water heater domain tests (value in current_temperature attribute) ---
@pytest.mark.usefixtures("enable_labs_preview_features")
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("water_heater"),
)
@pytest.mark.parametrize(
("trigger", "trigger_options", "states"),
[
*parametrize_numerical_attribute_changed_trigger_states(
"temperature.changed",
"eco",
WATER_HEATER_ATTR_CURRENT_TEMPERATURE,
trigger_options=_TEMPERATURE_TRIGGER_OPTIONS,
),
*parametrize_numerical_attribute_crossed_threshold_trigger_states(
"temperature.crossed_threshold",
"eco",
WATER_HEATER_ATTR_CURRENT_TEMPERATURE,
trigger_options=_TEMPERATURE_TRIGGER_OPTIONS,
),
],
)
async def test_temperature_trigger_water_heater_behavior_any(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_water_heaters: dict[str, list[str]],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
trigger_options: dict[str, Any],
states: list[TriggerStateDescription],
) -> None:
"""Test temperature trigger fires for water_heater entities."""
await assert_trigger_behavior_any(
hass,
service_calls=service_calls,
target_entities=target_water_heaters,
trigger_target_config=trigger_target_config,
entity_id=entity_id,
entities_in_target=entities_in_target,
trigger=trigger,
trigger_options=trigger_options,
states=states,
)
@pytest.mark.usefixtures("enable_labs_preview_features")
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("water_heater"),
)
@pytest.mark.parametrize(
("trigger", "trigger_options", "states"),
[
*parametrize_numerical_attribute_crossed_threshold_trigger_states(
"temperature.crossed_threshold",
"eco",
WATER_HEATER_ATTR_CURRENT_TEMPERATURE,
trigger_options=_TEMPERATURE_TRIGGER_OPTIONS,
),
],
)
async def test_temperature_trigger_water_heater_crossed_threshold_behavior_first(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_water_heaters: dict[str, list[str]],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
trigger_options: dict[str, Any],
states: list[TriggerStateDescription],
) -> None:
"""Test temperature crossed_threshold trigger fires on the first water_heater state change."""
await assert_trigger_behavior_first(
hass,
service_calls=service_calls,
target_entities=target_water_heaters,
trigger_target_config=trigger_target_config,
entity_id=entity_id,
entities_in_target=entities_in_target,
trigger=trigger,
trigger_options=trigger_options,
states=states,
)
@pytest.mark.usefixtures("enable_labs_preview_features")
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("water_heater"),
)
@pytest.mark.parametrize(
("trigger", "trigger_options", "states"),
[
*parametrize_numerical_attribute_crossed_threshold_trigger_states(
"temperature.crossed_threshold",
"eco",
WATER_HEATER_ATTR_CURRENT_TEMPERATURE,
trigger_options=_TEMPERATURE_TRIGGER_OPTIONS,
),
],
)
async def test_temperature_trigger_water_heater_crossed_threshold_behavior_last(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_water_heaters: dict[str, list[str]],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
trigger_options: dict[str, Any],
states: list[TriggerStateDescription],
) -> None:
"""Test temperature crossed_threshold trigger fires when the last water_heater changes state."""
await assert_trigger_behavior_last(
hass,
service_calls=service_calls,
target_entities=target_water_heaters,
trigger_target_config=trigger_target_config,
entity_id=entity_id,
entities_in_target=entities_in_target,
trigger=trigger,
trigger_options=trigger_options,
states=states,
)
# --- Weather domain tests (value in temperature attribute) ---
@pytest.mark.usefixtures("enable_labs_preview_features")
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("weather"),
)
@pytest.mark.parametrize(
("trigger", "trigger_options", "states"),
[
*parametrize_numerical_attribute_changed_trigger_states(
"temperature.changed",
"sunny",
ATTR_WEATHER_TEMPERATURE,
trigger_options=_TEMPERATURE_TRIGGER_OPTIONS,
unit_attributes=_WEATHER_UNIT_ATTRIBUTES,
),
*parametrize_numerical_attribute_crossed_threshold_trigger_states(
"temperature.crossed_threshold",
"sunny",
ATTR_WEATHER_TEMPERATURE,
trigger_options=_TEMPERATURE_TRIGGER_OPTIONS,
unit_attributes=_WEATHER_UNIT_ATTRIBUTES,
),
],
)
async def test_temperature_trigger_weather_behavior_any(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_weathers: dict[str, list[str]],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
trigger_options: dict[str, Any],
states: list[TriggerStateDescription],
) -> None:
"""Test temperature trigger fires for weather entities."""
await assert_trigger_behavior_any(
hass,
service_calls=service_calls,
target_entities=target_weathers,
trigger_target_config=trigger_target_config,
entity_id=entity_id,
entities_in_target=entities_in_target,
trigger=trigger,
trigger_options=trigger_options,
states=states,
)
@pytest.mark.usefixtures("enable_labs_preview_features")
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("weather"),
)
@pytest.mark.parametrize(
("trigger", "trigger_options", "states"),
[
*parametrize_numerical_attribute_crossed_threshold_trigger_states(
"temperature.crossed_threshold",
"sunny",
ATTR_WEATHER_TEMPERATURE,
trigger_options=_TEMPERATURE_TRIGGER_OPTIONS,
unit_attributes=_WEATHER_UNIT_ATTRIBUTES,
),
],
)
async def test_temperature_trigger_weather_crossed_threshold_behavior_first(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_weathers: dict[str, list[str]],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
trigger_options: dict[str, Any],
states: list[TriggerStateDescription],
) -> None:
"""Test temperature crossed_threshold trigger fires on the first weather state change."""
await assert_trigger_behavior_first(
hass,
service_calls=service_calls,
target_entities=target_weathers,
trigger_target_config=trigger_target_config,
entity_id=entity_id,
entities_in_target=entities_in_target,
trigger=trigger,
trigger_options=trigger_options,
states=states,
)
@pytest.mark.usefixtures("enable_labs_preview_features")
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("weather"),
)
@pytest.mark.parametrize(
("trigger", "trigger_options", "states"),
[
*parametrize_numerical_attribute_crossed_threshold_trigger_states(
"temperature.crossed_threshold",
"sunny",
ATTR_WEATHER_TEMPERATURE,
trigger_options=_TEMPERATURE_TRIGGER_OPTIONS,
unit_attributes=_WEATHER_UNIT_ATTRIBUTES,
),
],
)
async def test_temperature_trigger_weather_crossed_threshold_behavior_last(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_weathers: dict[str, list[str]],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
trigger_options: dict[str, Any],
states: list[TriggerStateDescription],
) -> None:
"""Test temperature crossed_threshold trigger fires when the last weather changes state."""
await assert_trigger_behavior_last(
hass,
service_calls=service_calls,
target_entities=target_weathers,
trigger_target_config=trigger_target_config,
entity_id=entity_id,
entities_in_target=entities_in_target,
trigger=trigger,
trigger_options=trigger_options,
states=states,
)
# --- Device class exclusion test ---
@pytest.mark.usefixtures("enable_labs_preview_features")
@pytest.mark.parametrize(
(
"trigger_key",
"trigger_options",
"sensor_initial",
"sensor_target",
),
[
(
"temperature.changed",
{},
"20",
"25",
),
(
"temperature.crossed_threshold",
{"threshold_type": "above", "lower_limit": 10, "unit": "°C"},
"5",
"20",
),
],
)
async def test_temperature_trigger_excludes_non_temperature_sensor(
hass: HomeAssistant,
service_calls: list[ServiceCall],
trigger_key: str,
trigger_options: dict[str, Any],
sensor_initial: str,
sensor_target: str,
) -> None:
"""Test temperature trigger does not fire for sensor entities without device_class temperature."""
entity_id_temperature = "sensor.test_temperature"
entity_id_humidity = "sensor.test_humidity"
temp_attrs = {
ATTR_DEVICE_CLASS: "temperature",
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.CELSIUS,
}
humidity_attrs = {ATTR_DEVICE_CLASS: "humidity"}
# Set initial states
hass.states.async_set(entity_id_temperature, sensor_initial, temp_attrs)
hass.states.async_set(entity_id_humidity, sensor_initial, humidity_attrs)
await hass.async_block_till_done()
await arm_trigger(
hass,
trigger_key,
trigger_options,
{
CONF_ENTITY_ID: [
entity_id_temperature,
entity_id_humidity,
]
},
)
# Temperature sensor changes - should trigger
hass.states.async_set(entity_id_temperature, sensor_target, temp_attrs)
await hass.async_block_till_done()
assert len(service_calls) == 1
assert service_calls[0].data[CONF_ENTITY_ID] == entity_id_temperature
service_calls.clear()
# Humidity sensor changes - should NOT trigger (wrong device class)
hass.states.async_set(entity_id_humidity, sensor_target, humidity_attrs)
await hass.async_block_till_done()
assert len(service_calls) == 0
# --- Unit conversion tests ---
@pytest.mark.usefixtures("enable_labs_preview_features")
async def test_temperature_trigger_unit_conversion_sensor_celsius_to_fahrenheit(
hass: HomeAssistant,
service_calls: list[ServiceCall],
) -> None:
"""Test temperature trigger converts sensor value from °C to °F for threshold comparison."""
entity_id = "sensor.test_temp"
# Sensor reports in °C, trigger configured in °F with threshold above 70°F
hass.states.async_set(
entity_id,
"20",
{
ATTR_DEVICE_CLASS: "temperature",
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.CELSIUS,
},
)
await hass.async_block_till_done()
await arm_trigger(
hass,
"temperature.crossed_threshold",
{
"threshold_type": "above",
"lower_limit": 70,
"unit": "°F",
},
{CONF_ENTITY_ID: [entity_id]},
)
# 20°C = 68°F, which is below 70°F - should NOT trigger
hass.states.async_set(
entity_id,
"20",
{
ATTR_DEVICE_CLASS: "temperature",
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.CELSIUS,
},
)
await hass.async_block_till_done()
assert len(service_calls) == 0
# 22°C = 71.6°F, which is above 70°F - should trigger
hass.states.async_set(
entity_id,
"22",
{
ATTR_DEVICE_CLASS: "temperature",
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.CELSIUS,
},
)
await hass.async_block_till_done()
assert len(service_calls) == 1
service_calls.clear()
@pytest.mark.usefixtures("enable_labs_preview_features")
async def test_temperature_trigger_unit_conversion_sensor_fahrenheit_to_celsius(
hass: HomeAssistant,
service_calls: list[ServiceCall],
) -> None:
"""Test temperature trigger converts sensor value from °F to °C for threshold comparison."""
entity_id = "sensor.test_temp"
# Sensor reports in °F, trigger configured in °C with threshold above 25°C
hass.states.async_set(
entity_id,
"70",
{
ATTR_DEVICE_CLASS: "temperature",
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
await hass.async_block_till_done()
await arm_trigger(
hass,
"temperature.crossed_threshold",
{
"threshold_type": "above",
"lower_limit": 25,
"unit": "°C",
},
{CONF_ENTITY_ID: [entity_id]},
)
# 70°F = 21.1°C, which is below 25°C - should NOT trigger
hass.states.async_set(
entity_id,
"70",
{
ATTR_DEVICE_CLASS: "temperature",
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
await hass.async_block_till_done()
assert len(service_calls) == 0
# 80°F = 26.7°C, which is above 25°C - should trigger
hass.states.async_set(
entity_id,
"80",
{
ATTR_DEVICE_CLASS: "temperature",
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
await hass.async_block_till_done()
assert len(service_calls) == 1
service_calls.clear()
@pytest.mark.usefixtures("enable_labs_preview_features")
async def test_temperature_trigger_unit_conversion_changed(
hass: HomeAssistant,
service_calls: list[ServiceCall],
) -> None:
"""Test temperature changed trigger with unit conversion and above/below limits."""
entity_id = "sensor.test_temp"
# Sensor reports in °C, trigger configured in °F: above 68°F (20°C), below 77°F (25°C)
hass.states.async_set(
entity_id,
"18",
{
ATTR_DEVICE_CLASS: "temperature",
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.CELSIUS,
},
)
await hass.async_block_till_done()
await arm_trigger(
hass,
"temperature.changed",
{
"above": 68,
"below": 77,
"unit": "°F",
},
{CONF_ENTITY_ID: [entity_id]},
)
# 18°C = 64.4°F, below 68°F - should NOT trigger
hass.states.async_set(
entity_id,
"19",
{
ATTR_DEVICE_CLASS: "temperature",
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.CELSIUS,
},
)
await hass.async_block_till_done()
assert len(service_calls) == 0
# 22°C = 71.6°F, between 68°F and 77°F - should trigger
hass.states.async_set(
entity_id,
"22",
{
ATTR_DEVICE_CLASS: "temperature",
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.CELSIUS,
},
)
await hass.async_block_till_done()
assert len(service_calls) == 1
service_calls.clear()
# 26°C = 78.8°F, above 77°F - should NOT trigger
hass.states.async_set(
entity_id,
"26",
{
ATTR_DEVICE_CLASS: "temperature",
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.CELSIUS,
},
)
await hass.async_block_till_done()
assert len(service_calls) == 0
@pytest.mark.usefixtures("enable_labs_preview_features")
async def test_temperature_trigger_unit_conversion_weather(
hass: HomeAssistant,
service_calls: list[ServiceCall],
) -> None:
"""Test temperature trigger with unit conversion for weather entities."""
entity_id = "weather.test"
# Weather reports temperature in °F, trigger configured in °C with threshold above 25°C
hass.states.async_set(
entity_id,
"sunny",
{
ATTR_WEATHER_TEMPERATURE: 70,
ATTR_WEATHER_TEMPERATURE_UNIT: UnitOfTemperature.FAHRENHEIT,
},
)
await hass.async_block_till_done()
await arm_trigger(
hass,
"temperature.crossed_threshold",
{
"threshold_type": "above",
"lower_limit": 25,
"unit": "°C",
},
{CONF_ENTITY_ID: [entity_id]},
)
# 70°F = 21.1°C, below 25°C - should NOT trigger
hass.states.async_set(
entity_id,
"sunny",
{
ATTR_WEATHER_TEMPERATURE: 70,
ATTR_WEATHER_TEMPERATURE_UNIT: UnitOfTemperature.FAHRENHEIT,
},
)
await hass.async_block_till_done()
assert len(service_calls) == 0
# 80°F = 26.7°C, above 25°C - should trigger
hass.states.async_set(
entity_id,
"sunny",
{
ATTR_WEATHER_TEMPERATURE: 80,
ATTR_WEATHER_TEMPERATURE_UNIT: UnitOfTemperature.FAHRENHEIT,
},
)
await hass.async_block_till_done()
assert len(service_calls) == 1
service_calls.clear()

View File

@@ -75,6 +75,7 @@ def mock_window() -> AsyncMock:
window.is_opening = False
window.is_closing = False
window.position = MagicMock(position_percent=30, closed=False)
window.wink = AsyncMock()
window.pyvlx = MagicMock()
return window
@@ -213,7 +214,6 @@ def mock_pyvlx(
mock_blind,
mock_window,
mock_exterior_heating,
mock_cover_type,
]
pyvlx.scenes = [mock_scene]

View File

@@ -1,5 +1,5 @@
# serializer version: 1
# name: test_button_snapshot[button.klf_200_gateway_restart-entry]
# name: test_button_snapshot[mock_window][button.klf_200_gateway_restart-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
@@ -36,7 +36,7 @@
'unit_of_measurement': None,
})
# ---
# name: test_button_snapshot[button.klf_200_gateway_restart-state]
# name: test_button_snapshot[mock_window][button.klf_200_gateway_restart-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'restart',
@@ -50,3 +50,54 @@
'state': 'unknown',
})
# ---
# name: test_button_snapshot[mock_window][button.test_window_identify-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'button',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'button.test_window_identify',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Identify',
'options': dict({
}),
'original_device_class': <ButtonDeviceClass.IDENTIFY: 'identify'>,
'original_icon': None,
'original_name': 'Identify',
'platform': 'velux',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': '123456789_identify',
'unit_of_measurement': None,
})
# ---
# name: test_button_snapshot[mock_window][button.test_window_identify-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'identify',
'friendly_name': 'Test Window Identify',
}),
'context': <ANY>,
'entity_id': 'button.test_window_identify',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---

View File

@@ -23,6 +23,7 @@ def platform() -> Platform:
@pytest.mark.usefixtures("setup_integration")
@pytest.mark.parametrize("mock_pyvlx", ["mock_window"], indirect=True)
async def test_button_snapshot(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
@@ -38,18 +39,33 @@ async def test_button_snapshot(
mock_config_entry.entry_id,
)
# Get the button entity setup and test device association
entity_entries = er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
assert len(entity_entries) == 1
entry = entity_entries[0]
assert len(entity_entries) == 2
assert entry.device_id is not None
device_entry = device_registry.async_get(entry.device_id)
assert device_entry is not None
assert (DOMAIN, f"gateway_{mock_config_entry.entry_id}") in device_entry.identifiers
assert device_entry.via_device_id is None
# Check Reboot button is associated with the gateway device
reboot_entry = next(
e for e in entity_entries if e.entity_id == "button.klf_200_gateway_restart"
)
assert reboot_entry.device_id is not None
gateway_device = device_registry.async_get(reboot_entry.device_id)
assert gateway_device is not None
assert (
DOMAIN,
f"gateway_{mock_config_entry.entry_id}",
) in gateway_device.identifiers
assert gateway_device.via_device_id is None
# Check Identify button is associated with the node device via the gateway
identify_entry = next(
e for e in entity_entries if e.entity_id == "button.test_window_identify"
)
assert identify_entry.device_id is not None
node_device = device_registry.async_get(identify_entry.device_id)
assert node_device is not None
assert (DOMAIN, "123456789") in node_device.identifiers
assert node_device.via_device_id == gateway_device.id
@pytest.mark.usefixtures("setup_integration")
@@ -98,3 +114,54 @@ async def test_button_press_failure(
# Verify the reboot method was called
mock_pyvlx.reboot_gateway.assert_called_once()
@pytest.mark.usefixtures("setup_integration")
@pytest.mark.parametrize("mock_pyvlx", ["mock_window"], indirect=True)
async def test_identify_button_press_success(
hass: HomeAssistant,
mock_window: AsyncMock,
) -> None:
"""Test successful identify button press."""
entity_id = "button.test_window_identify"
# Press the button
await hass.services.async_call(
BUTTON_DOMAIN,
SERVICE_PRESS,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
# Verify the wink method was called
mock_window.wink.assert_awaited_once()
@pytest.mark.usefixtures("setup_integration")
@pytest.mark.parametrize("mock_pyvlx", ["mock_window"], indirect=True)
async def test_identify_button_press_failure(
hass: HomeAssistant,
mock_window: AsyncMock,
) -> None:
"""Test identify button press failure handling."""
entity_id = "button.test_window_identify"
# Mock wink failure
mock_window.wink.side_effect = PyVLXException("Connection failed")
# Press the button and expect HomeAssistantError
with pytest.raises(
HomeAssistantError,
match='Failed to communicate with Velux device: <PyVLXException description="Connection failed" />',
):
await hass.services.async_call(
BUTTON_DOMAIN,
SERVICE_PRESS,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
# Verify the wink method was called
mock_window.wink.assert_awaited_once()

View File

@@ -17,7 +17,6 @@ from homeassistant.components.tag import DOMAIN as TAG_DOMAIN
from homeassistant.components.text import DOMAIN as TEXT_DOMAIN
from homeassistant.const import (
ATTR_DEVICE_CLASS,
ATTR_UNIT_OF_MEASUREMENT,
CONF_ABOVE,
CONF_BELOW,
CONF_ENTITY_ID,
@@ -26,7 +25,6 @@ from homeassistant.const import (
CONF_TARGET,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
UnitOfTemperature,
)
from homeassistant.core import (
CALLBACK_TYPE,
@@ -47,11 +45,8 @@ from homeassistant.helpers.automation import (
from homeassistant.helpers.trigger import (
CONF_LOWER_LIMIT,
CONF_THRESHOLD_TYPE,
CONF_UNIT,
CONF_UPPER_LIMIT,
DATA_PLUGGABLE_ACTIONS,
EntityNumericalStateChangedTriggerWithUnitBase,
EntityNumericalStateCrossedThresholdTriggerWithUnitBase,
EntityTriggerBase,
PluggableAction,
Trigger,
@@ -69,7 +64,6 @@ from homeassistant.helpers.trigger import (
from homeassistant.helpers.typing import ConfigType
from homeassistant.loader import Integration, async_get_integration
from homeassistant.setup import async_setup_component
from homeassistant.util.unit_conversion import TemperatureConverter
from homeassistant.util.yaml.loader import parse_yaml
from tests.common import MockModule, MockPlatform, mock_integration, mock_platform
@@ -1180,38 +1174,31 @@ async def test_subscribe_triggers_no_triggers(
("trigger_options", "expected_result"),
[
# Test validating climate.target_temperature_changed
# Valid: no limits at all
# Valid configurations
(
{},
does_not_raise(),
),
# Valid: numerical limits
(
{CONF_ABOVE: 10},
does_not_raise(),
),
(
{CONF_ABOVE: "sensor.test"},
does_not_raise(),
),
(
{CONF_BELOW: 90},
does_not_raise(),
),
(
{CONF_ABOVE: 10, CONF_BELOW: 90},
does_not_raise(),
),
# Valid: entity references
(
{CONF_ABOVE: "sensor.test"},
does_not_raise(),
),
(
{CONF_BELOW: "sensor.test"},
does_not_raise(),
),
(
{CONF_ABOVE: "sensor.test", CONF_BELOW: "sensor.test"},
{CONF_ABOVE: 10, CONF_BELOW: 90},
does_not_raise(),
),
# Valid: Mix of numerical limits and entity references
(
{CONF_ABOVE: "sensor.test", CONF_BELOW: 90},
does_not_raise(),
@@ -1220,6 +1207,10 @@ async def test_subscribe_triggers_no_triggers(
{CONF_ABOVE: 10, CONF_BELOW: "sensor.test"},
does_not_raise(),
),
(
{CONF_ABOVE: "sensor.test", CONF_BELOW: "sensor.test"},
does_not_raise(),
),
# Test verbose choose selector options
(
{CONF_ABOVE: {"active_choice": "entity", "entity": "sensor.test"}},
@@ -1285,147 +1276,6 @@ async def test_numerical_state_attribute_changed_trigger_config_validation(
)
def _make_with_unit_changed_trigger_class() -> type[
EntityNumericalStateChangedTriggerWithUnitBase
]:
"""Create a concrete WithUnit changed trigger class for testing."""
class _TestChangedTrigger(
EntityNumericalStateChangedTriggerWithUnitBase,
):
_base_unit = UnitOfTemperature.CELSIUS
_domain_specs = {"test": NumericalDomainSpec(value_source="test_attribute")}
_unit_converter = TemperatureConverter
return _TestChangedTrigger
@pytest.mark.parametrize(
("trigger_options", "expected_result"),
[
# Valid: no limits at all
(
{},
does_not_raise(),
),
# Valid: unit provided with numerical limits
(
{CONF_ABOVE: 10, CONF_UNIT: UnitOfTemperature.CELSIUS},
does_not_raise(),
),
(
{CONF_BELOW: 90, CONF_UNIT: UnitOfTemperature.FAHRENHEIT},
does_not_raise(),
),
(
{
CONF_ABOVE: 10,
CONF_BELOW: 90,
CONF_UNIT: UnitOfTemperature.CELSIUS,
},
does_not_raise(),
),
# Valid: no unit needed when using entity references
(
{CONF_ABOVE: "sensor.test"},
does_not_raise(),
),
(
{CONF_BELOW: "sensor.test"},
does_not_raise(),
),
(
{CONF_ABOVE: "sensor.test", CONF_BELOW: "sensor.test"},
does_not_raise(),
),
# Valid: unit only needed for numerical limits, not entity references
(
{
CONF_ABOVE: "sensor.test",
CONF_BELOW: 90,
CONF_UNIT: UnitOfTemperature.CELSIUS,
},
does_not_raise(),
),
(
{
CONF_ABOVE: 10,
CONF_BELOW: "sensor.test",
CONF_UNIT: UnitOfTemperature.CELSIUS,
},
does_not_raise(),
),
# Invalid: numerical limit without unit
(
{CONF_ABOVE: 10},
pytest.raises(vol.Invalid),
),
(
{CONF_BELOW: 90},
pytest.raises(vol.Invalid),
),
(
{CONF_ABOVE: 10, CONF_BELOW: 90},
pytest.raises(vol.Invalid),
),
# Invalid: one numerical limit without unit (other is entity)
(
{CONF_ABOVE: 10, CONF_BELOW: "sensor.test"},
pytest.raises(vol.Invalid),
),
(
{CONF_ABOVE: "sensor.test", CONF_BELOW: 90},
pytest.raises(vol.Invalid),
),
# Invalid: invalid unit value
(
{CONF_ABOVE: 10, CONF_UNIT: "invalid_unit"},
pytest.raises(vol.Invalid),
),
# Invalid: Must use valid entity id
(
{CONF_ABOVE: "cat", CONF_BELOW: "dog"},
pytest.raises(vol.Invalid),
),
# Invalid: above must be smaller than below
(
{CONF_ABOVE: 90, CONF_BELOW: 10, CONF_UNIT: UnitOfTemperature.CELSIUS},
pytest.raises(vol.Invalid),
),
# Invalid: invalid choose selector option
(
{CONF_BELOW: {"active_choice": "cat", "cat": 90}},
pytest.raises(vol.Invalid),
),
],
)
async def test_numerical_state_attribute_changed_with_unit_trigger_config_validation(
hass: HomeAssistant,
trigger_options: dict[str, Any],
expected_result: AbstractContextManager,
) -> None:
"""Test numerical state attribute change with unit trigger config validation."""
trigger_cls = _make_with_unit_changed_trigger_class()
async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
return {"test_trigger": trigger_cls}
mock_integration(hass, MockModule("test"))
mock_platform(hass, "test.trigger", Mock(async_get_triggers=async_get_triggers))
with expected_result:
await async_validate_trigger_config(
hass,
[
{
"platform": "test.test_trigger",
CONF_TARGET: {CONF_ENTITY_ID: "test.test_entity"},
CONF_OPTIONS: trigger_options,
}
],
)
async def test_numerical_state_attribute_changed_error_handling(
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
@@ -1539,302 +1389,6 @@ async def test_numerical_state_attribute_changed_error_handling(
assert len(service_calls) == 0
async def test_numerical_state_attribute_changed_with_unit_error_handling(
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test numerical state attribute change with unit conversion error handling."""
trigger_cls = _make_with_unit_changed_trigger_class()
async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
return {"attribute_changed": trigger_cls}
mock_integration(hass, MockModule("test"))
mock_platform(hass, "test.trigger", Mock(async_get_triggers=async_get_triggers))
# Entity reports in °F, trigger configured in °C with above 20°C, below 30°C
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": 68, # 68°F = 20°C
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: [
{
"trigger": {
CONF_PLATFORM: "test.attribute_changed",
CONF_TARGET: {CONF_ENTITY_ID: "test.test_entity"},
CONF_OPTIONS: {
CONF_ABOVE: 20,
CONF_BELOW: 30,
CONF_UNIT: UnitOfTemperature.CELSIUS,
},
},
"action": {
"service": "test.numerical_automation",
"data_template": {CONF_ENTITY_ID: "{{ trigger.entity_id }}"},
},
},
{
"trigger": {
CONF_PLATFORM: "test.attribute_changed",
CONF_TARGET: {CONF_ENTITY_ID: "test.test_entity"},
CONF_OPTIONS: {
CONF_ABOVE: "sensor.above",
CONF_BELOW: "sensor.below",
},
},
"action": {
"service": "test.entity_automation",
"data_template": {CONF_ENTITY_ID: "{{ trigger.entity_id }}"},
},
},
]
},
)
assert len(service_calls) == 0
# 77°F = 25°C, within range (above 20, below 30) - should trigger numerical
# Entity automation won't trigger because sensor.above/below don't exist yet
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": 77,
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
await hass.async_block_till_done()
assert len(service_calls) == 1
assert service_calls[0].service == "numerical_automation"
service_calls.clear()
# 59°F = 15°C, below 20°C - should NOT trigger
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": 59,
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
await hass.async_block_till_done()
assert len(service_calls) == 0
# 95°F = 35°C, above 30°C - should NOT trigger
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": 95,
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
await hass.async_block_till_done()
assert len(service_calls) == 0
# Set up entity limits referencing sensors that report in °F
hass.states.async_set(
"sensor.above",
"68", # 68°F = 20°C
{ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT},
)
hass.states.async_set(
"sensor.below",
"86", # 86°F = 30°C
{ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT},
)
# 77°F = 25°C, between 20°C and 30°C - should trigger both automations
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": 77,
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
await hass.async_block_till_done()
assert len(service_calls) == 2
assert {call.service for call in service_calls} == {
"numerical_automation",
"entity_automation",
}
service_calls.clear()
# Test the trigger does not fire when the attribute value is missing
hass.states.async_set("test.test_entity", "on", {})
await hass.async_block_till_done()
assert len(service_calls) == 0
# Test the trigger does not fire when the attribute value is invalid
for value in ("cat", None):
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": value,
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
await hass.async_block_till_done()
assert len(service_calls) == 0
# Test the trigger does not fire when the unit is incompatible
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": 50,
ATTR_UNIT_OF_MEASUREMENT: "invalid_unit",
},
)
await hass.async_block_till_done()
assert len(service_calls) == 0
# Test the trigger does not fire when the above sensor does not exist
hass.states.async_remove("sensor.above")
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": None,
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
hass.states.async_set(
"test.test_entity",
"on",
{"test_attribute": 50, ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT},
)
await hass.async_block_till_done()
assert len(service_calls) == 0
# Test the trigger does not fire when the above sensor state is not numeric
for invalid_value in ("cat", None):
hass.states.async_set(
"sensor.above",
invalid_value,
{ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT},
)
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": None,
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": 50,
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
await hass.async_block_till_done()
assert len(service_calls) == 0
# Test the trigger does not fire when the above sensor's unit is incompatible
hass.states.async_set(
"sensor.above",
"68", # 68°F = 20°C
{ATTR_UNIT_OF_MEASUREMENT: "invalid_unit"},
)
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": None,
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
hass.states.async_set(
"test.test_entity",
"on",
{"test_attribute": 50, ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT},
)
await hass.async_block_till_done()
assert len(service_calls) == 0
# Reset the above sensor state to a valid numeric value
hass.states.async_set(
"sensor.above",
"68", # 68°F = 20°C
{ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT},
)
# Test the trigger does not fire when the below sensor does not exist
hass.states.async_remove("sensor.below")
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": None,
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
hass.states.async_set(
"test.test_entity",
"on",
{"test_attribute": 50, ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT},
)
await hass.async_block_till_done()
assert len(service_calls) == 0
# Test the trigger does not fire when the below sensor state is not numeric
for invalid_value in ("cat", None):
hass.states.async_set("sensor.below", invalid_value)
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": None,
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": 50,
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
await hass.async_block_till_done()
assert len(service_calls) == 0
# Test the trigger does not fire when the below sensor's unit is incompatible
hass.states.async_set(
"sensor.below",
"68", # 68°F = 20°C
{ATTR_UNIT_OF_MEASUREMENT: "invalid_unit"},
)
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": None,
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
hass.states.async_set(
"test.test_entity",
"on",
{"test_attribute": 50, ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT},
)
await hass.async_block_till_done()
assert len(service_calls) == 0
@pytest.mark.parametrize(
("trigger_options", "expected_result"),
[
@@ -2032,367 +1586,6 @@ async def test_numerical_state_attribute_crossed_threshold_trigger_config_valida
)
def _make_with_unit_crossed_threshold_trigger_class() -> type[
EntityNumericalStateCrossedThresholdTriggerWithUnitBase
]:
"""Create a concrete WithUnit crossed threshold trigger class for testing."""
class _TestCrossedThresholdTrigger(
EntityNumericalStateCrossedThresholdTriggerWithUnitBase,
):
_base_unit = UnitOfTemperature.CELSIUS
_domain_specs = {"test": NumericalDomainSpec(value_source="test_attribute")}
_unit_converter = TemperatureConverter
return _TestCrossedThresholdTrigger
@pytest.mark.parametrize(
("trigger_options", "expected_result"),
[
# Valid: unit provided with numerical limits
(
{
CONF_THRESHOLD_TYPE: "above",
CONF_LOWER_LIMIT: 10,
CONF_UNIT: UnitOfTemperature.CELSIUS,
},
does_not_raise(),
),
(
{
CONF_THRESHOLD_TYPE: "below",
CONF_UPPER_LIMIT: 90,
CONF_UNIT: UnitOfTemperature.FAHRENHEIT,
},
does_not_raise(),
),
(
{
CONF_THRESHOLD_TYPE: "between",
CONF_LOWER_LIMIT: 10,
CONF_UPPER_LIMIT: 90,
CONF_UNIT: UnitOfTemperature.CELSIUS,
},
does_not_raise(),
),
# Valid: no unit needed when using entity references
(
{
CONF_THRESHOLD_TYPE: "above",
CONF_LOWER_LIMIT: "sensor.test",
},
does_not_raise(),
),
(
{
CONF_THRESHOLD_TYPE: "between",
CONF_LOWER_LIMIT: "sensor.test",
CONF_UPPER_LIMIT: "sensor.test",
},
does_not_raise(),
),
# Invalid: numerical limit without unit
(
{CONF_THRESHOLD_TYPE: "above", CONF_LOWER_LIMIT: 10},
pytest.raises(vol.Invalid),
),
(
{
CONF_THRESHOLD_TYPE: "between",
CONF_LOWER_LIMIT: 10,
CONF_UPPER_LIMIT: 90,
},
pytest.raises(vol.Invalid),
),
# Invalid: one numerical limit without unit (other is entity)
(
{
CONF_THRESHOLD_TYPE: "between",
CONF_LOWER_LIMIT: 10,
CONF_UPPER_LIMIT: "sensor.test",
},
pytest.raises(vol.Invalid),
),
# Invalid: invalid unit value
(
{
CONF_THRESHOLD_TYPE: "above",
CONF_LOWER_LIMIT: 10,
CONF_UNIT: "invalid_unit",
},
pytest.raises(vol.Invalid),
),
# Invalid: missing threshold type (shared validation)
(
{},
pytest.raises(vol.Invalid),
),
],
)
async def test_numerical_state_attribute_crossed_threshold_with_unit_trigger_config_validation(
hass: HomeAssistant,
trigger_options: dict[str, Any],
expected_result: AbstractContextManager,
) -> None:
"""Test numerical state attribute crossed threshold with unit trigger config validation."""
trigger_cls = _make_with_unit_crossed_threshold_trigger_class()
async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
return {"test_trigger": trigger_cls}
mock_integration(hass, MockModule("test"))
mock_platform(hass, "test.trigger", Mock(async_get_triggers=async_get_triggers))
with expected_result:
await async_validate_trigger_config(
hass,
[
{
"platform": "test.test_trigger",
CONF_TARGET: {CONF_ENTITY_ID: "test.test_entity"},
CONF_OPTIONS: trigger_options,
}
],
)
async def test_numerical_state_attribute_crossed_threshold_error_handling(
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test numerical state attribute crossed threshold error handling."""
async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
return {
"crossed_threshold": make_entity_numerical_state_crossed_threshold_trigger(
{"test": NumericalDomainSpec(value_source="test_attribute")}
),
}
mock_integration(hass, MockModule("test"))
mock_platform(hass, "test.trigger", Mock(async_get_triggers=async_get_triggers))
hass.states.async_set("test.test_entity", "on", {"test_attribute": 0})
options = {
CONF_OPTIONS: {
CONF_THRESHOLD_TYPE: "between",
CONF_LOWER_LIMIT: "sensor.lower",
CONF_UPPER_LIMIT: "sensor.upper",
},
}
await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: {
"trigger": {
CONF_PLATFORM: "test.crossed_threshold",
CONF_TARGET: {CONF_ENTITY_ID: "test.test_entity"},
}
| options,
"action": {
"service": "test.automation",
"data_template": {CONF_ENTITY_ID: "{{ trigger.entity_id }}"},
},
}
},
)
assert len(service_calls) == 0
# Test the trigger works
hass.states.async_set("sensor.lower", "10")
hass.states.async_set("sensor.upper", "90")
hass.states.async_set("test.test_entity", "on", {"test_attribute": 50})
await hass.async_block_till_done()
assert len(service_calls) == 1
service_calls.clear()
# Test the trigger does not fire again when still within limits
hass.states.async_set("test.test_entity", "on", {"test_attribute": 51})
await hass.async_block_till_done()
assert len(service_calls) == 0
service_calls.clear()
# Test the trigger does not fire when the from-state is unknown or unavailable
for from_state in (STATE_UNKNOWN, STATE_UNAVAILABLE):
hass.states.async_set("test.test_entity", from_state)
hass.states.async_set("test.test_entity", "on", {"test_attribute": 50})
await hass.async_block_till_done()
assert len(service_calls) == 0
# Test the trigger does fire when the attribute value is changing from None
hass.states.async_set("test.test_entity", "on", {"test_attribute": None})
hass.states.async_set("test.test_entity", "on", {"test_attribute": 50})
await hass.async_block_till_done()
assert len(service_calls) == 1
service_calls.clear()
# Test the trigger does not fire when the attribute value is outside the limits
for value in (5, 95):
hass.states.async_set("test.test_entity", "on", {"test_attribute": value})
await hass.async_block_till_done()
assert len(service_calls) == 0
# Test the trigger does not fire when the attribute value is missing
hass.states.async_set("test.test_entity", "on", {})
await hass.async_block_till_done()
assert len(service_calls) == 0
# Test the trigger does not fire when the attribute value is invalid
for value in ("cat", None):
hass.states.async_set("test.test_entity", "on", {"test_attribute": value})
await hass.async_block_till_done()
assert len(service_calls) == 0
# Test the trigger does not fire when the lower sensor does not exist
hass.states.async_remove("sensor.lower")
hass.states.async_set("test.test_entity", "on", {"test_attribute": None})
hass.states.async_set("test.test_entity", "on", {"test_attribute": 50})
await hass.async_block_till_done()
assert len(service_calls) == 0
# Test the trigger does not fire when the lower sensor state is not numeric
for invalid_value in ("cat", None):
hass.states.async_set("sensor.lower", invalid_value)
hass.states.async_set("test.test_entity", "on", {"test_attribute": None})
hass.states.async_set("test.test_entity", "on", {"test_attribute": 50})
await hass.async_block_till_done()
assert len(service_calls) == 0
# Reset the lower sensor state to a valid numeric value
hass.states.async_set("sensor.lower", "10")
# Test the trigger does not fire when the upper sensor does not exist
hass.states.async_remove("sensor.upper")
hass.states.async_set("test.test_entity", "on", {"test_attribute": None})
hass.states.async_set("test.test_entity", "on", {"test_attribute": 50})
await hass.async_block_till_done()
assert len(service_calls) == 0
# Test the trigger does not fire when the upper sensor state is not numeric
for invalid_value in ("cat", None):
hass.states.async_set("sensor.upper", invalid_value)
hass.states.async_set("test.test_entity", "on", {"test_attribute": None})
hass.states.async_set("test.test_entity", "on", {"test_attribute": 50})
await hass.async_block_till_done()
assert len(service_calls) == 0
async def test_numerical_state_attribute_crossed_threshold_with_unit_error_handling(
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test numerical state attribute crossed threshold with unit conversion."""
trigger_cls = _make_with_unit_crossed_threshold_trigger_class()
async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
return {"crossed_threshold": trigger_cls}
mock_integration(hass, MockModule("test"))
mock_platform(hass, "test.trigger", Mock(async_get_triggers=async_get_triggers))
# Entity reports in °F, trigger configured in °C: above 25°C
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": 68, # 68°F = 20°C, below threshold
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
options = {
CONF_OPTIONS: {
CONF_THRESHOLD_TYPE: "above",
CONF_LOWER_LIMIT: 25,
CONF_UNIT: UnitOfTemperature.CELSIUS,
},
}
await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: {
"trigger": {
CONF_PLATFORM: "test.crossed_threshold",
CONF_TARGET: {CONF_ENTITY_ID: "test.test_entity"},
}
| options,
"action": {
"service": "test.automation",
"data_template": {CONF_ENTITY_ID: "{{ trigger.entity_id }}"},
},
}
},
)
assert len(service_calls) == 0
# 80.6°F = 27°C, above 25°C threshold - should trigger
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": 80.6,
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
await hass.async_block_till_done()
assert len(service_calls) == 1
service_calls.clear()
# Still above threshold - should NOT trigger (already crossed)
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": 82,
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
await hass.async_block_till_done()
assert len(service_calls) == 0
# Drop below threshold and cross again
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": 68, # 20°C, below 25°C
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
await hass.async_block_till_done()
assert len(service_calls) == 0
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": 80.6, # 27°C, above 25°C again
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
await hass.async_block_till_done()
assert len(service_calls) == 1
service_calls.clear()
# Test with incompatible unit - should NOT trigger
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": 50,
ATTR_UNIT_OF_MEASUREMENT: "invalid_unit",
},
)
await hass.async_block_till_done()
assert len(service_calls) == 0
def _make_trigger(
hass: HomeAssistant, domain_specs: Mapping[str, DomainSpec]
) -> EntityTriggerBase:

View File

@@ -86,7 +86,6 @@
'system_health',
'system_log',
'tag',
'temperature',
'text',
'time',
'timer',
@@ -191,7 +190,6 @@
'system_health',
'system_log',
'tag',
'temperature',
'text',
'time',
'timer',