Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-05-08 17:49:37 +01:00
Code Activity
Files
fd76693d6e24a1aa705a45b18656d31b86b72951
core/tests/components/timer/test_trigger.py
T
Erik Montnemery 10084c8c0c Add trigger timer.time_remaining (#169763)
2026-05-05 23:54:49 -04:00

761 lines
26 KiB
Python
Raw Blame History

"""Test timer triggers."""
from datetime import timedelta
import logging
from typing import Any
from freezegun.api import FrozenDateTimeFactory
import pytest
import voluptuous as vol
from homeassistant.components.timer import (
ATTR_FINISHES_AT,
ATTR_LAST_TRANSITION,
DOMAIN,
STATUS_ACTIVE,
STATUS_IDLE,
STATUS_PAUSED,
)
from homeassistant.const import (
ATTR_LABEL_ID,
CONF_ENTITY_ID,
CONF_OPTIONS,
CONF_PLATFORM,
CONF_TARGET,
)
from homeassistant.core import Context, HomeAssistant, callback
from homeassistant.helpers import entity_registry as er, label_registry as lr
from homeassistant.helpers.trigger import (
async_initialize_triggers,
async_validate_trigger_config,
)
from homeassistant.helpers.typing import TemplateVarsType
from homeassistant.util import dt as dt_util
from tests.common import async_fire_time_changed
from tests.components.common import (
TriggerStateDescription,
assert_trigger_behavior_any,
assert_trigger_behavior_first,
assert_trigger_behavior_last,
assert_trigger_gated_by_labs_flag,
assert_trigger_options_supported,
parametrize_target_entities,
parametrize_trigger_states,
target_entities,
)
@pytest.fixture
async def target_timers(hass: HomeAssistant) -> dict[str, list[str]]:
"""Create multiple timer entities associated with different targets."""
return await target_entities(hass, DOMAIN)
@pytest.mark.parametrize(
"trigger_key",
[
"timer.cancelled",
"timer.finished",
"timer.paused",
"timer.restarted",
"timer.started",
"timer.time_remaining",
],
)
async def test_timer_triggers_gated_by_labs_flag(
hass: HomeAssistant, caplog: pytest.LogCaptureFixture, trigger_key: str
) -> None:
"""Test the timer triggers are gated by the labs flag."""
await assert_trigger_gated_by_labs_flag(hass, caplog, trigger_key)
@pytest.mark.usefixtures("enable_labs_preview_features")
@pytest.mark.parametrize(
("trigger_key", "base_options", "supports_behavior", "supports_duration"),
[
("timer.cancelled", {}, True, True),
("timer.finished", {}, True, True),
("timer.paused", {}, True, True),
("timer.restarted", {}, True, True),
("timer.started", {}, True, True),
("timer.time_remaining", {"remaining": {"hours": 1}}, False, False),
],
)
async def test_timer_trigger_options_validation(
hass: HomeAssistant,
trigger_key: str,
base_options: dict[str, Any] | None,
supports_behavior: bool,
supports_duration: bool,
) -> None:
"""Test that timer triggers support the expected options."""
await assert_trigger_options_supported(
hass,
trigger_key,
base_options,
supports_behavior=supports_behavior,
supports_duration=supports_duration,
)
@pytest.mark.usefixtures("enable_labs_preview_features")
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities(DOMAIN),
)
@pytest.mark.parametrize(
("trigger", "trigger_options", "states"),
[
*parametrize_trigger_states(
trigger="timer.cancelled",
target_states=[(STATUS_IDLE, {ATTR_LAST_TRANSITION: "cancelled"})],
other_states=[(STATUS_ACTIVE, {ATTR_LAST_TRANSITION: "started"})],
),
*parametrize_trigger_states(
trigger="timer.finished",
target_states=[(STATUS_IDLE, {ATTR_LAST_TRANSITION: "finished"})],
other_states=[(STATUS_ACTIVE, {ATTR_LAST_TRANSITION: "started"})],
),
*parametrize_trigger_states(
trigger="timer.paused",
target_states=[(STATUS_PAUSED, {ATTR_LAST_TRANSITION: "paused"})],
other_states=[(STATUS_ACTIVE, {ATTR_LAST_TRANSITION: "started"})],
),
*parametrize_trigger_states(
trigger="timer.restarted",
target_states=[(STATUS_ACTIVE, {ATTR_LAST_TRANSITION: "restarted"})],
other_states=[(STATUS_ACTIVE, {ATTR_LAST_TRANSITION: "started"})],
),
*parametrize_trigger_states(
trigger="timer.started",
target_states=[(STATUS_ACTIVE, {ATTR_LAST_TRANSITION: "started"})],
other_states=[(STATUS_IDLE, {ATTR_LAST_TRANSITION: "cancelled"})],
),
],
)
async def test_timer_trigger_behavior_any(
hass: HomeAssistant,
target_timers: dict[str, list[str]],
trigger_target_config: dict[str, Any],
entity_id: str,
entities_in_target: int,
trigger: str,
trigger_options: dict[str, Any],
states: list[TriggerStateDescription],
) -> None:
"""Test that the timer trigger fires when any timer's last_transition changes to a specific value."""
await assert_trigger_behavior_any(
hass,
target_entities=target_timers,
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(DOMAIN),
)
@pytest.mark.parametrize(
("trigger", "trigger_options", "states"),
[
*parametrize_trigger_states(
trigger="timer.cancelled",
target_states=[(STATUS_IDLE, {ATTR_LAST_TRANSITION: "cancelled"})],
other_states=[(STATUS_ACTIVE, {ATTR_LAST_TRANSITION: "started"})],
),
*parametrize_trigger_states(
trigger="timer.finished",
target_states=[(STATUS_IDLE, {ATTR_LAST_TRANSITION: "finished"})],
other_states=[(STATUS_ACTIVE, {ATTR_LAST_TRANSITION: "started"})],
),
*parametrize_trigger_states(
trigger="timer.paused",
target_states=[(STATUS_PAUSED, {ATTR_LAST_TRANSITION: "paused"})],
other_states=[(STATUS_ACTIVE, {ATTR_LAST_TRANSITION: "started"})],
),
*parametrize_trigger_states(
trigger="timer.restarted",
target_states=[(STATUS_ACTIVE, {ATTR_LAST_TRANSITION: "restarted"})],
other_states=[(STATUS_ACTIVE, {ATTR_LAST_TRANSITION: "started"})],
),
*parametrize_trigger_states(
trigger="timer.started",
target_states=[(STATUS_ACTIVE, {ATTR_LAST_TRANSITION: "started"})],
other_states=[(STATUS_IDLE, {ATTR_LAST_TRANSITION: "cancelled"})],
),
],
)
async def test_timer_trigger_behavior_first(
hass: HomeAssistant,
target_timers: dict[str, list[str]],
trigger_target_config: dict[str, Any],
entity_id: str,
entities_in_target: int,
trigger: str,
trigger_options: dict[str, Any],
states: list[TriggerStateDescription],
) -> None:
"""Test that the timer trigger fires when the first timer's last_transition changes to a specific value."""
await assert_trigger_behavior_first(
hass,
target_entities=target_timers,
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(DOMAIN),
)
@pytest.mark.parametrize(
("trigger", "trigger_options", "states"),
[
*parametrize_trigger_states(
trigger="timer.cancelled",
target_states=[(STATUS_IDLE, {ATTR_LAST_TRANSITION: "cancelled"})],
other_states=[(STATUS_ACTIVE, {ATTR_LAST_TRANSITION: "started"})],
),
*parametrize_trigger_states(
trigger="timer.finished",
target_states=[(STATUS_IDLE, {ATTR_LAST_TRANSITION: "finished"})],
other_states=[(STATUS_ACTIVE, {ATTR_LAST_TRANSITION: "started"})],
),
*parametrize_trigger_states(
trigger="timer.paused",
target_states=[(STATUS_PAUSED, {ATTR_LAST_TRANSITION: "paused"})],
other_states=[(STATUS_ACTIVE, {ATTR_LAST_TRANSITION: "started"})],
),
*parametrize_trigger_states(
trigger="timer.restarted",
target_states=[(STATUS_ACTIVE, {ATTR_LAST_TRANSITION: "restarted"})],
other_states=[(STATUS_ACTIVE, {ATTR_LAST_TRANSITION: "started"})],
),
*parametrize_trigger_states(
trigger="timer.started",
target_states=[(STATUS_ACTIVE, {ATTR_LAST_TRANSITION: "started"})],
other_states=[(STATUS_IDLE, {ATTR_LAST_TRANSITION: "cancelled"})],
),
],
)
async def test_timer_trigger_behavior_last(
hass: HomeAssistant,
target_timers: dict[str, list[str]],
trigger_target_config: dict[str, Any],
entity_id: str,
entities_in_target: int,
trigger: str,
trigger_options: dict[str, Any],
states: list[TriggerStateDescription],
) -> None:
"""Test that the timer trigger fires when the last timer's last_transition changes to a specific value."""
await assert_trigger_behavior_last(
hass,
target_entities=target_timers,
trigger_target_config=trigger_target_config,
entity_id=entity_id,
entities_in_target=entities_in_target,
trigger=trigger,
trigger_options=trigger_options,
states=states,
)
# --- time_remaining trigger tests ---
async def _arm_time_remaining_trigger(
hass: HomeAssistant,
entity_id: str,
remaining: dict[str, int],
calls: list[dict[str, Any]],
*,
target: dict[str, Any] | None = None,
) -> None:
"""Arm the time_remaining trigger."""
trigger_config = await async_validate_trigger_config(
hass,
[
{
CONF_PLATFORM: "timer.time_remaining",
CONF_TARGET: target or {CONF_ENTITY_ID: entity_id},
CONF_OPTIONS: {"remaining": remaining},
}
],
)
@callback
def action(run_variables: TemplateVarsType, context: Context | None = None) -> None:
calls.append(run_variables["trigger"])
logger = logging.getLogger(__name__)
def log_cb(level: int, msg: str, **kwargs: Any) -> None:
logger._log(level, "%s", msg, **kwargs)
await async_initialize_triggers(
hass,
trigger_config,
action,
domain="test",
name="test_trigger",
log_cb=log_cb,
)
@pytest.mark.usefixtures("enable_labs_preview_features")
async def test_time_remaining_trigger_validation(hass: HomeAssistant) -> None:
"""Test time_remaining trigger config validation."""
# Valid config
await async_validate_trigger_config(
hass,
[
{
CONF_PLATFORM: "timer.time_remaining",
CONF_TARGET: {CONF_ENTITY_ID: "timer.test"},
CONF_OPTIONS: {"remaining": {"seconds": 30}},
}
],
)
# Missing remaining option
with pytest.raises(vol.Invalid):
await async_validate_trigger_config(
hass,
[
{
CONF_PLATFORM: "timer.time_remaining",
CONF_TARGET: {CONF_ENTITY_ID: "timer.test"},
CONF_OPTIONS: {},
}
],
)
@pytest.mark.usefixtures("enable_labs_preview_features")
async def test_time_remaining_trigger_fires(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test time_remaining trigger fires at the right time."""
now = dt_util.utcnow()
calls: list[dict[str, Any]] = []
hass.states.async_set("timer.test", STATUS_IDLE, {ATTR_LAST_TRANSITION: None})
await hass.async_block_till_done()
await _arm_time_remaining_trigger(hass, "timer.test", {"seconds": 30}, calls)
# Start timer with 60 second duration
finishes_at = now + timedelta(seconds=60)
hass.states.async_set(
"timer.test",
STATUS_ACTIVE,
{ATTR_LAST_TRANSITION: "started", ATTR_FINISHES_AT: finishes_at.isoformat()},
)
await hass.async_block_till_done()
assert len(calls) == 0
# Advance to 25 seconds - 35 seconds remaining, should not fire
freezer.move_to(now + timedelta(seconds=25))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 0
# Advance to 30 seconds - 30 seconds remaining, should fire
freezer.move_to(now + timedelta(seconds=30))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0]["entity_id"] == "timer.test"
assert calls[0]["remaining"] == timedelta(seconds=30)
@pytest.mark.usefixtures("enable_labs_preview_features")
async def test_time_remaining_trigger_paused_before_threshold(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test time_remaining trigger does not fire when timer is paused before threshold."""
now = dt_util.utcnow()
calls: list[dict[str, Any]] = []
hass.states.async_set("timer.test", STATUS_IDLE, {ATTR_LAST_TRANSITION: None})
await hass.async_block_till_done()
await _arm_time_remaining_trigger(hass, "timer.test", {"seconds": 30}, calls)
# Start timer with 60 second duration
finishes_at = now + timedelta(seconds=60)
hass.states.async_set(
"timer.test",
STATUS_ACTIVE,
{ATTR_LAST_TRANSITION: "started", ATTR_FINISHES_AT: finishes_at.isoformat()},
)
await hass.async_block_till_done()
# Pause timer at 10 seconds (before the 30-second threshold)
freezer.move_to(now + timedelta(seconds=10))
hass.states.async_set(
"timer.test",
STATUS_PAUSED,
{ATTR_LAST_TRANSITION: "paused"},
)
await hass.async_block_till_done()
# Advance past the original fire time - should not fire since paused
freezer.move_to(now + timedelta(seconds=35))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 0
@pytest.mark.usefixtures("enable_labs_preview_features")
async def test_time_remaining_trigger_cancelled_before_threshold(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test time_remaining trigger does not fire when timer is cancelled before threshold."""
now = dt_util.utcnow()
calls: list[dict[str, Any]] = []
hass.states.async_set("timer.test", STATUS_IDLE, {ATTR_LAST_TRANSITION: None})
await hass.async_block_till_done()
await _arm_time_remaining_trigger(hass, "timer.test", {"seconds": 30}, calls)
# Start timer with 60 second duration
finishes_at = now + timedelta(seconds=60)
hass.states.async_set(
"timer.test",
STATUS_ACTIVE,
{ATTR_LAST_TRANSITION: "started", ATTR_FINISHES_AT: finishes_at.isoformat()},
)
await hass.async_block_till_done()
# Cancel timer at 10 seconds
freezer.move_to(now + timedelta(seconds=10))
hass.states.async_set(
"timer.test",
STATUS_IDLE,
{ATTR_LAST_TRANSITION: "cancelled"},
)
await hass.async_block_till_done()
# Advance past the original fire time - should not fire since cancelled
freezer.move_to(now + timedelta(seconds=35))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 0
@pytest.mark.usefixtures("enable_labs_preview_features")
async def test_time_remaining_trigger_restarted(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test time_remaining trigger reschedules when timer is restarted."""
now = dt_util.utcnow()
calls: list[dict[str, Any]] = []
hass.states.async_set("timer.test", STATUS_IDLE, {ATTR_LAST_TRANSITION: None})
await hass.async_block_till_done()
await _arm_time_remaining_trigger(hass, "timer.test", {"seconds": 30}, calls)
# Start timer with 60 second duration
finishes_at = now + timedelta(seconds=60)
hass.states.async_set(
"timer.test",
STATUS_ACTIVE,
{ATTR_LAST_TRANSITION: "started", ATTR_FINISHES_AT: finishes_at.isoformat()},
)
await hass.async_block_till_done()
# Restart timer at 10 seconds with a new 60-second duration
freezer.move_to(now + timedelta(seconds=10))
new_finishes_at = now + timedelta(seconds=70) # 10s elapsed + 60s new
hass.states.async_set(
"timer.test",
STATUS_ACTIVE,
{
ATTR_LAST_TRANSITION: "restarted",
ATTR_FINISHES_AT: new_finishes_at.isoformat(),
},
)
await hass.async_block_till_done()
assert len(calls) == 0
# Original fire time (30s) should not fire since rescheduled
freezer.move_to(now + timedelta(seconds=30))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 0
# New fire time: new_finishes_at - 30s = now + 40s
freezer.move_to(now + timedelta(seconds=40))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 1
@pytest.mark.usefixtures("enable_labs_preview_features")
async def test_time_remaining_trigger_short_timer(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test time_remaining trigger does not fire when timer duration is shorter than remaining threshold."""
now = dt_util.utcnow()
calls: list[dict[str, Any]] = []
hass.states.async_set("timer.test", STATUS_IDLE, {ATTR_LAST_TRANSITION: None})
await hass.async_block_till_done()
await _arm_time_remaining_trigger(hass, "timer.test", {"seconds": 30}, calls)
# Start timer with only 20 second duration (less than 30s threshold)
finishes_at = now + timedelta(seconds=20)
hass.states.async_set(
"timer.test",
STATUS_ACTIVE,
{ATTR_LAST_TRANSITION: "started", ATTR_FINISHES_AT: finishes_at.isoformat()},
)
await hass.async_block_till_done()
# fire_at = now + 20 - 30 = now - 10 (in the past), should not schedule
# Advance past the timer's end time
freezer.move_to(now + timedelta(seconds=25))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 0
@pytest.mark.usefixtures("enable_labs_preview_features")
async def test_time_remaining_trigger_already_active_at_attach(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test trigger schedules for timers already active when the trigger attaches."""
now = dt_util.utcnow()
calls: list[dict[str, Any]] = []
# Timer is already active before the trigger is armed
finishes_at = now + timedelta(seconds=60)
hass.states.async_set(
"timer.test",
STATUS_ACTIVE,
{ATTR_LAST_TRANSITION: "started", ATTR_FINISHES_AT: finishes_at.isoformat()},
)
await hass.async_block_till_done()
await _arm_time_remaining_trigger(hass, "timer.test", {"seconds": 30}, calls)
# No fire yet
assert len(calls) == 0
# Before fire_at (finishes_at - 30s = now + 30s) — should not fire
freezer.move_to(now + timedelta(seconds=25))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 0
# At fire_at — should fire even though no state change occurred
freezer.move_to(now + timedelta(seconds=30))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0]["entity_id"] == "timer.test"
assert calls[0]["remaining"] == timedelta(seconds=30)
@pytest.mark.usefixtures("enable_labs_preview_features")
async def test_time_remaining_trigger_already_active_past_threshold_at_attach(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test trigger does not schedule for timers already past the fire point at attach."""
now = dt_util.utcnow()
calls: list[dict[str, Any]] = []
# Timer is active but only 20 seconds remain — past the 30s threshold already
finishes_at = now + timedelta(seconds=20)
hass.states.async_set(
"timer.test",
STATUS_ACTIVE,
{ATTR_LAST_TRANSITION: "started", ATTR_FINISHES_AT: finishes_at.isoformat()},
)
await hass.async_block_till_done()
await _arm_time_remaining_trigger(hass, "timer.test", {"seconds": 30}, calls)
# Advance past the timer's finishing time — should never fire
freezer.move_to(now + timedelta(seconds=25))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 0
@pytest.mark.usefixtures("enable_labs_preview_features")
async def test_time_remaining_trigger_idle_at_attach(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test trigger does not schedule for non-active timers at attach time."""
now = dt_util.utcnow()
calls: list[dict[str, Any]] = []
hass.states.async_set("timer.test", STATUS_IDLE, {ATTR_LAST_TRANSITION: None})
await hass.async_block_till_done()
await _arm_time_remaining_trigger(hass, "timer.test", {"seconds": 30}, calls)
# Even far in the future, no fire because timer never started
freezer.move_to(now + timedelta(seconds=120))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 0
@pytest.mark.usefixtures("enable_labs_preview_features")
async def test_time_remaining_trigger_active_on_first_state_event(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test trigger schedules when first observed state event has no from_state.
This simulates a timer entity that is created/restored after the trigger
is attached and appears directly in active state (e.g., RestoreEntity on
restart), where the initial state-change event has from_state=None.
"""
now = dt_util.utcnow()
calls: list[dict[str, Any]] = []
await _arm_time_remaining_trigger(hass, "timer.test", {"seconds": 30}, calls)
# First state event for the entity has no old_state
finishes_at = now + timedelta(seconds=60)
hass.states.async_set(
"timer.test",
STATUS_ACTIVE,
{ATTR_LAST_TRANSITION: "started", ATTR_FINISHES_AT: finishes_at.isoformat()},
)
await hass.async_block_till_done()
assert len(calls) == 0
# Advance to fire time — should still fire even though from_state was None
freezer.move_to(now + timedelta(seconds=30))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0]["entity_id"] == "timer.test"
assert calls[0]["remaining"] == timedelta(seconds=30)
@pytest.mark.usefixtures("enable_labs_preview_features")
async def test_time_remaining_trigger_entity_removed_from_target(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
entity_registry: er.EntityRegistry,
) -> None:
"""Test trigger cancels scheduled fire when entity is removed from the target."""
now = dt_util.utcnow()
calls: list[dict[str, Any]] = []
label_reg = lr.async_get(hass)
label = label_reg.async_create("Test Time Remaining")
entry = entity_registry.async_get_or_create(
domain=DOMAIN, platform="test", unique_id="time_remaining_remove"
)
entity_registry.async_update_entity(entry.entity_id, labels={label.label_id})
hass.states.async_set(entry.entity_id, STATUS_IDLE, {ATTR_LAST_TRANSITION: None})
await hass.async_block_till_done()
await _arm_time_remaining_trigger(
hass,
entry.entity_id,
{"seconds": 30},
calls,
target={ATTR_LABEL_ID: label.label_id},
)
# Start the timer — this schedules a fire via the state-change path
finishes_at = now + timedelta(seconds=60)
hass.states.async_set(
entry.entity_id,
STATUS_ACTIVE,
{ATTR_LAST_TRANSITION: "started", ATTR_FINISHES_AT: finishes_at.isoformat()},
)
await hass.async_block_till_done()
# Remove the entity from the target by stripping its label
freezer.move_to(now + timedelta(seconds=10))
entity_registry.async_update_entity(entry.entity_id, labels=set())
await hass.async_block_till_done()
# Advance past the original fire time — should not fire since cancelled
freezer.move_to(now + timedelta(seconds=35))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 0
@pytest.mark.usefixtures("enable_labs_preview_features")
async def test_time_remaining_trigger_entity_added_to_target(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
entity_registry: er.EntityRegistry,
) -> None:
"""Test trigger schedules a fire for an active timer added to the target later."""
now = dt_util.utcnow()
calls: list[dict[str, Any]] = []
label_reg = lr.async_get(hass)
label = label_reg.async_create("Test Time Remaining Add")
entry = entity_registry.async_get_or_create(
domain=DOMAIN, platform="test", unique_id="time_remaining_add"
)
# Timer is active, but not in the target yet
finishes_at = now + timedelta(seconds=60)
hass.states.async_set(
entry.entity_id,
STATUS_ACTIVE,
{ATTR_LAST_TRANSITION: "started", ATTR_FINISHES_AT: finishes_at.isoformat()},
)
await hass.async_block_till_done()
await _arm_time_remaining_trigger(
hass,
entry.entity_id,
{"seconds": 30},
calls,
target={ATTR_LABEL_ID: label.label_id},
)
# Now label the entity so it joins the target
entity_registry.async_update_entity(entry.entity_id, labels={label.label_id})
await hass.async_block_till_done()
# Advance to the fire time — should fire even though no state change occurred
freezer.move_to(now + timedelta(seconds=30))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0]["entity_id"] == entry.entity_id
assert calls[0]["remaining"] == timedelta(seconds=30)
View Git Blame Copy Permalink