Compare commits

...

2 Commits

Author SHA1 Message Date
Michael Hansen 1993031364 Update bootstrap snapshot for timer_list base platform
timer_list is a base platform, so it is set up by default with the other
base platforms and now appears in the bootstrap components snapshot.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-25 15:13:02 -05:00
Michael Hansen fdd5da3f3b Add timer_list integration
A timer_list entity holds many in-memory countdown timers (its items),
mirroring how a to-do list holds many to-do items; its state is the number
of active timers. Adds the base entity platform with start/pause/unpause/
cancel/add_time/remove_time/clear/get services, automation triggers
(timer_started/updated/finished/cancelled), and a websocket subscribe
command. Also adds a local_timer_list helper so timer lists can be created
from the UI.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-25 14:25:48 -05:00
29 changed files with 1914 additions and 0 deletions
+1
View File
@@ -51,6 +51,7 @@ base_platforms: &base_platforms
- homeassistant/components/switch/**
- homeassistant/components/text/**
- homeassistant/components/time/**
- homeassistant/components/timer_list/**
- homeassistant/components/todo/**
- homeassistant/components/tts/**
- homeassistant/components/update/**
Generated
+4
View File
@@ -1031,6 +1031,8 @@ CLAUDE.md @home-assistant/core
/tests/components/local_calendar/ @allenporter
/homeassistant/components/local_ip/ @issacg
/tests/components/local_ip/ @issacg
/homeassistant/components/local_timer_list/ @home-assistant/core @synesthesiam
/tests/components/local_timer_list/ @home-assistant/core @synesthesiam
/homeassistant/components/local_todo/ @allenporter
/tests/components/local_todo/ @allenporter
/homeassistant/components/lock/ @home-assistant/core
@@ -1832,6 +1834,8 @@ CLAUDE.md @home-assistant/core
/tests/components/time/ @home-assistant/core
/homeassistant/components/time_date/ @fabaff
/tests/components/time_date/ @fabaff
/homeassistant/components/timer_list/ @home-assistant/core @synesthesiam
/tests/components/timer_list/ @home-assistant/core @synesthesiam
/homeassistant/components/tmb/ @alemuro
/homeassistant/components/todo/ @home-assistant/core
/tests/components/todo/ @home-assistant/core
@@ -0,0 +1,22 @@
"""The Local Timer list integration.
Creates an in-memory timer list entity (provided by the ``timer_list`` base
platform) for each config entry, so users can create timer lists from the UI.
"""
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
PLATFORMS = [Platform.TIMER_LIST]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Local Timer list from a config entry."""
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
@@ -0,0 +1,33 @@
"""Config flow for the Local Timer list integration."""
from typing import Any, override
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from .const import CONF_TIMER_LIST_NAME, DOMAIN
STEP_USER_DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_TIMER_LIST_NAME): str,
}
)
class LocalTimerListConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Local Timer list."""
VERSION = 1
@override
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the initial step."""
if user_input is not None:
return self.async_create_entry(
title=user_input[CONF_TIMER_LIST_NAME], data=user_input
)
return self.async_show_form(step_id="user", data_schema=STEP_USER_DATA_SCHEMA)
@@ -0,0 +1,5 @@
"""Constants for the Local Timer list integration."""
DOMAIN = "local_timer_list"
CONF_TIMER_LIST_NAME = "timer_list_name"
@@ -0,0 +1,10 @@
{
"domain": "local_timer_list",
"name": "Local Timer list",
"codeowners": ["@home-assistant/core", "@synesthesiam"],
"config_flow": true,
"dependencies": ["timer_list"],
"documentation": "https://www.home-assistant.io/integrations/local_timer_list",
"integration_type": "helper",
"iot_class": "local_push"
}
@@ -0,0 +1,13 @@
{
"config": {
"step": {
"user": {
"data": {
"timer_list_name": "Name"
},
"description": "Choose a name for the new timer list. The entity ID is derived from this name.",
"submit": "Create"
}
}
}
}
@@ -0,0 +1,34 @@
"""Local timer list platform."""
from homeassistant.components.timer_list import TimerListEntity, TimerListEntityFeature
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import CONF_TIMER_LIST_NAME
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the local timer list entity from a config entry."""
async_add_entities([LocalTimerListEntity(config_entry)])
class LocalTimerListEntity(TimerListEntity):
"""A local, in-memory timer list."""
_attr_supported_features = (
TimerListEntityFeature.START_TIMER
| TimerListEntityFeature.PAUSE_TIMER
| TimerListEntityFeature.CANCEL_TIMER
| TimerListEntityFeature.ADD_TIME
)
def __init__(self, config_entry: ConfigEntry) -> None:
"""Initialize the timer list."""
super().__init__()
self._attr_name = config_entry.data[CONF_TIMER_LIST_NAME]
self._attr_unique_id = config_entry.entry_id
@@ -0,0 +1,547 @@
"""The Timer list integration.
A timer list entity holds many independent countdown timers (its *items*),
mirroring how a to-do list holds many to-do items. The entity state is the
number of active timers. Timers are kept in memory only: they do not survive a
restart of Home Assistant in this first version (see the module-level notes on
``async_will_remove_from_hass``).
"""
from collections.abc import Callable
import copy
import dataclasses
from datetime import datetime, timedelta
from functools import partial
import logging
from typing import Any, final, override
import voluptuous as vol
from homeassistant.components import websocket_api
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_NAME
from homeassistant.core import (
CALLBACK_TYPE,
HomeAssistant,
ServiceCall,
SupportsResponse,
callback,
)
from homeassistant.exceptions import ServiceValidationError
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.entity_component import EntityComponent
from homeassistant.helpers.event import async_track_point_in_utc_time
from homeassistant.helpers.typing import ConfigType
from homeassistant.util import dt as dt_util, ulid as ulid_util
from .const import (
ATTR_DURATION,
ATTR_FINISH_ACTION,
ATTR_STATUS,
ATTR_TIMER_ID,
DATA_COMPONENT,
DOMAIN,
TimerFinishAction,
TimerListEntityFeature,
TimerListEventType,
TimerListServices,
TimerStatus,
)
_LOGGER = logging.getLogger(__name__)
ENTITY_ID_FORMAT = DOMAIN + ".{}"
PLATFORM_SCHEMA = cv.PLATFORM_SCHEMA
PLATFORM_SCHEMA_BASE = cv.PLATFORM_SCHEMA_BASE
_FINISHED_STATUSES = (TimerStatus.FINISHED, TimerStatus.CANCELLED)
@dataclasses.dataclass
class TimerItem:
"""A single timer within a timer list."""
timer_id: str
"""Generated unique id of the timer."""
name: str | None
"""Optional user-provided name."""
status: TimerStatus
"""Current status of the timer."""
finish_action: TimerFinishAction
"""What happens to the timer once it finishes."""
duration: timedelta
"""Original duration the timer was created with (used by ``restart``)."""
created_at: datetime
"""When the timer was (re)started, in UTC."""
finishes_at: datetime | None = None
"""Absolute time the timer will finish, in UTC. ``None`` unless active."""
remaining: timedelta | None = None
"""Remaining time captured while paused. ``None`` unless paused."""
finished_at: datetime | None = None
"""When the timer finished or was cancelled, in UTC."""
def remaining_at(self, now: datetime) -> timedelta:
"""Return the time left on the timer relative to ``now``."""
if self.status == TimerStatus.ACTIVE and self.finishes_at is not None:
return max(timedelta(0), self.finishes_at - now)
if self.status == TimerStatus.PAUSED and self.remaining is not None:
return self.remaining
return timedelta(0)
@dataclasses.dataclass(frozen=True)
class TimerListEvent:
"""A change to a timer, pushed to subscribers and triggers."""
event_type: TimerListEventType
item: TimerItem
@callback
def timer_to_dict(item: TimerItem, now: datetime) -> dict[str, Any]:
"""Serialize a timer item for the websocket API and triggers."""
return {
ATTR_TIMER_ID: item.timer_id,
ATTR_NAME: item.name,
ATTR_STATUS: item.status.value,
ATTR_FINISH_ACTION: item.finish_action.value,
"duration": item.duration.total_seconds(),
"created_at": item.created_at.isoformat(),
"finishes_at": item.finishes_at.isoformat() if item.finishes_at else None,
"finished_at": item.finished_at.isoformat() if item.finished_at else None,
"remaining": item.remaining_at(now).total_seconds(),
}
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the Timer list component."""
component = hass.data[DATA_COMPONENT] = EntityComponent[TimerListEntity](
_LOGGER, DOMAIN, hass
)
websocket_api.async_register_command(hass, websocket_handle_subscribe)
websocket_api.async_register_command(hass, websocket_handle_list)
component.async_register_entity_service(
TimerListServices.START_TIMER,
{
vol.Optional(ATTR_NAME): cv.string,
vol.Required(ATTR_DURATION): cv.positive_time_period,
vol.Optional(
ATTR_FINISH_ACTION, default=TimerFinishAction.REMOVE
): vol.Coerce(TimerFinishAction),
},
_async_start_timer,
required_features=[TimerListEntityFeature.START_TIMER],
supports_response=SupportsResponse.OPTIONAL,
)
component.async_register_entity_service(
TimerListServices.PAUSE_TIMER,
{vol.Required(ATTR_TIMER_ID): cv.string},
"async_pause_timer",
required_features=[TimerListEntityFeature.PAUSE_TIMER],
)
component.async_register_entity_service(
TimerListServices.UNPAUSE_TIMER,
{vol.Required(ATTR_TIMER_ID): cv.string},
"async_unpause_timer",
required_features=[TimerListEntityFeature.PAUSE_TIMER],
)
component.async_register_entity_service(
TimerListServices.CANCEL_TIMER,
{vol.Required(ATTR_TIMER_ID): cv.string},
"async_cancel_timer",
required_features=[TimerListEntityFeature.CANCEL_TIMER],
)
component.async_register_entity_service(
TimerListServices.CANCEL_ALL_TIMERS,
None,
"async_cancel_all_timers",
required_features=[TimerListEntityFeature.CANCEL_TIMER],
)
component.async_register_entity_service(
TimerListServices.ADD_TIME,
{
vol.Required(ATTR_TIMER_ID): cv.string,
vol.Required(ATTR_DURATION): cv.positive_time_period,
},
_async_add_time,
required_features=[TimerListEntityFeature.ADD_TIME],
)
component.async_register_entity_service(
TimerListServices.REMOVE_TIME,
{
vol.Required(ATTR_TIMER_ID): cv.string,
vol.Required(ATTR_DURATION): cv.positive_time_period,
},
_async_remove_time,
required_features=[TimerListEntityFeature.ADD_TIME],
)
component.async_register_entity_service(
TimerListServices.REMOVE_TIMER,
{vol.Required(ATTR_TIMER_ID): cv.string},
"async_remove_timer",
)
component.async_register_entity_service(
TimerListServices.CLEAR_FINISHED_TIMERS,
None,
"async_clear_finished_timers",
)
component.async_register_entity_service(
TimerListServices.GET_TIMERS,
{vol.Optional(ATTR_STATUS): vol.All(cv.ensure_list, [vol.Coerce(TimerStatus)])},
_async_get_timers,
supports_response=SupportsResponse.ONLY,
)
await component.async_setup(config)
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up a config entry."""
return await hass.data[DATA_COMPONENT].async_setup_entry(entry)
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
return await hass.data[DATA_COMPONENT].async_unload_entry(entry)
class TimerListEntity(Entity):
"""An entity that holds a list of independent countdown timers."""
_attr_should_poll = False
def __init__(self) -> None:
"""Initialize the timer list."""
self._timers: dict[str, TimerItem] = {}
self._cancel_callbacks: dict[str, CALLBACK_TYPE] = {}
self._update_listeners: list[Callable[[TimerListEvent], None]] = []
@property
@override
def state(self) -> int:
"""Return the number of active timers."""
return sum(
timer.status == TimerStatus.ACTIVE for timer in self._timers.values()
)
@property
def timers(self) -> list[TimerItem]:
"""Return the timers in the list."""
return list(self._timers.values())
async def async_start_timer(
self,
*,
name: str | None,
duration: timedelta,
finish_action: TimerFinishAction,
) -> str:
"""Create and start a new timer, returning its id."""
now = dt_util.utcnow()
timer_id = ulid_util.ulid_now()
timer = TimerItem(
timer_id=timer_id,
name=name,
status=TimerStatus.ACTIVE,
finish_action=finish_action,
duration=duration,
created_at=now,
finishes_at=now + duration,
)
self._timers[timer_id] = timer
self._schedule(timer)
self._notify(TimerListEventType.STARTED, timer)
return timer_id
async def async_pause_timer(self, timer_id: str) -> None:
"""Pause an active timer."""
timer = self._get_timer(timer_id)
if timer.status != TimerStatus.ACTIVE or timer.finishes_at is None:
return
timer.remaining = max(timedelta(0), timer.finishes_at - dt_util.utcnow())
timer.finishes_at = None
timer.status = TimerStatus.PAUSED
self._unschedule(timer_id)
self._notify(TimerListEventType.UPDATED, timer)
async def async_unpause_timer(self, timer_id: str) -> None:
"""Resume a paused timer."""
timer = self._get_timer(timer_id)
if timer.status != TimerStatus.PAUSED or timer.remaining is None:
return
timer.finishes_at = dt_util.utcnow() + timer.remaining
timer.remaining = None
timer.status = TimerStatus.ACTIVE
self._schedule(timer)
self._notify(TimerListEventType.UPDATED, timer)
async def async_cancel_timer(self, timer_id: str) -> None:
"""Cancel a timer.
The timer is retained in the ``cancelled`` state only when its finish
action is ``archive``; otherwise it is removed.
"""
timer = self._get_timer(timer_id)
self._unschedule(timer_id)
timer.status = TimerStatus.CANCELLED
timer.finishes_at = None
timer.remaining = None
timer.finished_at = dt_util.utcnow()
self._notify(TimerListEventType.CANCELLED, timer)
if timer.finish_action != TimerFinishAction.ARCHIVE:
del self._timers[timer_id]
self._notify(TimerListEventType.REMOVED, timer)
async def async_cancel_all_timers(self) -> None:
"""Cancel every active or paused timer."""
for timer_id in [
timer.timer_id
for timer in self._timers.values()
if timer.status in (TimerStatus.ACTIVE, TimerStatus.PAUSED)
]:
await self.async_cancel_timer(timer_id)
async def async_add_time(self, timer_id: str, duration: timedelta) -> None:
"""Add (or, with a negative duration, remove) time on a timer."""
timer = self._get_timer(timer_id)
if timer.status == TimerStatus.ACTIVE and timer.finishes_at is not None:
now = dt_util.utcnow()
finishes_at = timer.finishes_at + duration
if finishes_at <= now:
self._unschedule(timer_id)
self._async_timer_finished(timer_id, now)
return
timer.finishes_at = finishes_at
self._schedule(timer)
elif timer.status == TimerStatus.PAUSED and timer.remaining is not None:
timer.remaining = max(timedelta(0), timer.remaining + duration)
else:
return
self._notify(TimerListEventType.UPDATED, timer)
async def async_remove_timer(self, timer_id: str) -> None:
"""Remove a timer from the list regardless of its status."""
timer = self._get_timer(timer_id)
self._unschedule(timer_id)
del self._timers[timer_id]
self._notify(TimerListEventType.REMOVED, timer)
async def async_clear_finished_timers(self) -> None:
"""Remove all finished and cancelled (archived) timers."""
for timer_id in [
timer.timer_id
for timer in self._timers.values()
if timer.status in _FINISHED_STATUSES
]:
timer = self._timers.pop(timer_id)
self._notify(TimerListEventType.REMOVED, timer)
def _get_timer(self, timer_id: str) -> TimerItem:
"""Return a timer by id or raise if it does not exist."""
if (timer := self._timers.get(timer_id)) is None:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="timer_not_found",
translation_placeholders={"timer_id": timer_id},
)
return timer
@callback
def _schedule(self, timer: TimerItem) -> None:
"""Schedule (or reschedule) the finish callback for a timer."""
self._unschedule(timer.timer_id)
assert timer.finishes_at is not None
self._cancel_callbacks[timer.timer_id] = async_track_point_in_utc_time(
self.hass,
partial(self._async_timer_finished, timer.timer_id),
timer.finishes_at,
)
@callback
def _unschedule(self, timer_id: str) -> None:
"""Cancel a pending finish callback, if any."""
if cancel := self._cancel_callbacks.pop(timer_id, None):
cancel()
@callback
def _async_timer_finished(self, timer_id: str, now: datetime) -> None:
"""Handle a timer reaching its finish time."""
self._cancel_callbacks.pop(timer_id, None)
if (timer := self._timers.get(timer_id)) is None:
return
timer.status = TimerStatus.FINISHED
timer.finishes_at = None
timer.remaining = None
timer.finished_at = dt_util.utcnow()
self._notify(TimerListEventType.FINISHED, timer)
if timer.finish_action == TimerFinishAction.REMOVE:
self._timers.pop(timer_id, None)
self._notify(TimerListEventType.REMOVED, timer)
elif timer.finish_action == TimerFinishAction.RESTART:
restarted_at = dt_util.utcnow()
timer.status = TimerStatus.ACTIVE
timer.created_at = restarted_at
timer.finishes_at = restarted_at + timer.duration
timer.finished_at = None
self._schedule(timer)
self._notify(TimerListEventType.STARTED, timer)
@final
@callback
def async_subscribe_updates(
self, listener: Callable[[TimerListEvent], None]
) -> CALLBACK_TYPE:
"""Subscribe to timer change events.
Only future changes are pushed; the current set of timers is not
replayed on subscribe.
"""
self._update_listeners.append(listener)
@callback
def unsubscribe() -> None:
self._update_listeners.remove(listener)
return unsubscribe
@callback
def _notify(self, event_type: TimerListEventType, timer: TimerItem) -> None:
"""Push a change event to subscribers and write entity state."""
event = TimerListEvent(event_type=event_type, item=copy.copy(timer))
for listener in list(self._update_listeners):
listener(event)
self.async_write_ha_state()
@override
async def async_will_remove_from_hass(self) -> None:
"""Cancel all pending finish callbacks."""
for cancel in self._cancel_callbacks.values():
cancel()
self._cancel_callbacks.clear()
async def _async_start_timer(
entity: TimerListEntity, call: ServiceCall
) -> dict[str, Any]:
"""Handle the start_timer service."""
timer_id = await entity.async_start_timer(
name=call.data.get(ATTR_NAME),
duration=call.data[ATTR_DURATION],
finish_action=call.data[ATTR_FINISH_ACTION],
)
return {ATTR_TIMER_ID: timer_id}
async def _async_add_time(entity: TimerListEntity, call: ServiceCall) -> None:
"""Handle the add_time service."""
await entity.async_add_time(call.data[ATTR_TIMER_ID], call.data[ATTR_DURATION])
async def _async_remove_time(entity: TimerListEntity, call: ServiceCall) -> None:
"""Handle the remove_time service."""
await entity.async_add_time(call.data[ATTR_TIMER_ID], -call.data[ATTR_DURATION])
async def _async_get_timers(
entity: TimerListEntity, call: ServiceCall
) -> dict[str, Any]:
"""Handle the get_timers service."""
now = dt_util.utcnow()
statuses: list[TimerStatus] | None = call.data.get(ATTR_STATUS)
return {
"timers": [
timer_to_dict(timer, now)
for timer in entity.timers
if not statuses or timer.status in statuses
]
}
@websocket_api.websocket_command(
{
vol.Required("type"): "timer_list/item/subscribe",
vol.Required("entity_id"): cv.entity_domain(DOMAIN),
}
)
@websocket_api.async_response
async def websocket_handle_subscribe(
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict[str, Any]
) -> None:
"""Subscribe to timer changes for a timer list, with an initial snapshot."""
entity_id: str = msg["entity_id"]
if not (entity := hass.data[DATA_COMPONENT].get_entity(entity_id)):
connection.send_error(
msg["id"],
websocket_api.ERR_NOT_FOUND,
f"Timer list entity not found: {entity_id}",
)
return
@callback
def forward_event(event: TimerListEvent) -> None:
"""Forward a timer change event to the websocket connection."""
connection.send_message(
websocket_api.event_message(
msg["id"],
{
"type": "change",
"event_type": event.event_type.value,
"timer": timer_to_dict(event.item, dt_util.utcnow()),
},
)
)
connection.subscriptions[msg["id"]] = entity.async_subscribe_updates(forward_event)
connection.send_result(msg["id"])
now = dt_util.utcnow()
connection.send_message(
websocket_api.event_message(
msg["id"],
{
"type": "timers",
"timers": [timer_to_dict(timer, now) for timer in entity.timers],
},
)
)
@websocket_api.websocket_command(
{
vol.Required("type"): "timer_list/item/list",
vol.Required("entity_id"): cv.entity_domain(DOMAIN),
}
)
@websocket_api.async_response
async def websocket_handle_list(
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict[str, Any]
) -> None:
"""Return the current timers for a timer list."""
entity_id: str = msg["entity_id"]
if not (entity := hass.data[DATA_COMPONENT].get_entity(entity_id)):
connection.send_error(
msg["id"],
websocket_api.ERR_NOT_FOUND,
f"Timer list entity not found: {entity_id}",
)
return
now = dt_util.utcnow()
connection.send_result(
msg["id"],
{"timers": [timer_to_dict(timer, now) for timer in entity.timers]},
)
@@ -0,0 +1,76 @@
"""Constants for the Timer list integration."""
from enum import IntFlag, StrEnum
from typing import TYPE_CHECKING
from homeassistant.util.hass_dict import HassKey
if TYPE_CHECKING:
from homeassistant.helpers.entity_component import EntityComponent
from . import TimerListEntity
DOMAIN = "timer_list"
DATA_COMPONENT: HassKey[EntityComponent[TimerListEntity]] = HassKey(DOMAIN)
ATTR_TIMER_ID = "timer_id"
ATTR_DURATION = "duration"
ATTR_FINISH_ACTION = "finish_action"
ATTR_FINISHES_AT = "finishes_at"
ATTR_CREATED_AT = "created_at"
ATTR_FINISHED_AT = "finished_at"
ATTR_REMAINING = "remaining"
ATTR_STATUS = "status"
ATTR_TIMER = "timer"
ATTR_TIMERS = "timers"
class TimerListServices(StrEnum):
"""Services for the Timer list integration."""
START_TIMER = "start_timer"
PAUSE_TIMER = "pause_timer"
UNPAUSE_TIMER = "unpause_timer"
CANCEL_TIMER = "cancel_timer"
CANCEL_ALL_TIMERS = "cancel_all_timers"
ADD_TIME = "add_time"
REMOVE_TIME = "remove_time"
REMOVE_TIMER = "remove_timer"
CLEAR_FINISHED_TIMERS = "clear_finished_timers"
GET_TIMERS = "get_timers"
class TimerStatus(StrEnum):
"""Status of a single timer in a timer list."""
ACTIVE = "active"
PAUSED = "paused"
FINISHED = "finished"
CANCELLED = "cancelled"
class TimerFinishAction(StrEnum):
"""What happens to a timer once it finishes."""
REMOVE = "remove"
ARCHIVE = "archive"
RESTART = "restart"
class TimerListEventType(StrEnum):
"""Type of change pushed to timer list subscribers."""
STARTED = "started"
UPDATED = "updated"
FINISHED = "finished"
CANCELLED = "cancelled"
REMOVED = "removed"
class TimerListEntityFeature(IntFlag):
"""Supported features of a timer list entity."""
START_TIMER = 1
PAUSE_TIMER = 2
CANCEL_TIMER = 4
ADD_TIME = 8
@@ -0,0 +1,53 @@
{
"entity_component": {
"_": {
"default": "mdi:timer-outline"
}
},
"services": {
"add_time": {
"service": "mdi:timer-plus-outline"
},
"cancel_all_timers": {
"service": "mdi:timer-cancel-outline"
},
"cancel_timer": {
"service": "mdi:timer-cancel-outline"
},
"clear_finished_timers": {
"service": "mdi:timer-remove-outline"
},
"get_timers": {
"service": "mdi:timer-outline"
},
"pause_timer": {
"service": "mdi:timer-pause-outline"
},
"remove_time": {
"service": "mdi:timer-minus-outline"
},
"remove_timer": {
"service": "mdi:timer-remove-outline"
},
"start_timer": {
"service": "mdi:timer-plus-outline"
},
"unpause_timer": {
"service": "mdi:timer-play-outline"
}
},
"triggers": {
"timer_cancelled": {
"trigger": "mdi:timer-cancel-outline"
},
"timer_finished": {
"trigger": "mdi:timer-check-outline"
},
"timer_started": {
"trigger": "mdi:timer-play-outline"
},
"timer_updated": {
"trigger": "mdi:timer-edit-outline"
}
}
}
@@ -0,0 +1,9 @@
{
"domain": "timer_list",
"name": "Timer list",
"codeowners": ["@home-assistant/core", "@synesthesiam"],
"dependencies": ["websocket_api"],
"documentation": "https://www.home-assistant.io/integrations/timer_list",
"integration_type": "entity",
"quality_scale": "internal"
}
@@ -0,0 +1,131 @@
start_timer:
target:
entity:
domain: timer_list
supported_features:
- timer_list.TimerListEntityFeature.START_TIMER
fields:
name:
example: "Pasta"
selector:
text:
duration:
required: true
example: "00:05:00"
selector:
duration:
finish_action:
default: remove
selector:
select:
translation_key: finish_action
options:
- remove
- archive
- restart
pause_timer:
target:
entity:
domain: timer_list
supported_features:
- timer_list.TimerListEntityFeature.PAUSE_TIMER
fields:
timer_id:
required: true
example: "01HZ8ABCDEF0123456789ABCDE"
selector:
text:
unpause_timer:
target:
entity:
domain: timer_list
supported_features:
- timer_list.TimerListEntityFeature.PAUSE_TIMER
fields:
timer_id:
required: true
example: "01HZ8ABCDEF0123456789ABCDE"
selector:
text:
cancel_timer:
target:
entity:
domain: timer_list
supported_features:
- timer_list.TimerListEntityFeature.CANCEL_TIMER
fields:
timer_id:
required: true
example: "01HZ8ABCDEF0123456789ABCDE"
selector:
text:
cancel_all_timers:
target:
entity:
domain: timer_list
supported_features:
- timer_list.TimerListEntityFeature.CANCEL_TIMER
add_time:
target:
entity:
domain: timer_list
supported_features:
- timer_list.TimerListEntityFeature.ADD_TIME
fields:
timer_id:
required: true
example: "01HZ8ABCDEF0123456789ABCDE"
selector:
text:
duration:
required: true
example: "00:01:00"
selector:
duration:
remove_time:
target:
entity:
domain: timer_list
supported_features:
- timer_list.TimerListEntityFeature.ADD_TIME
fields:
timer_id:
required: true
example: "01HZ8ABCDEF0123456789ABCDE"
selector:
text:
duration:
required: true
example: "00:01:00"
selector:
duration:
remove_timer:
target:
entity:
domain: timer_list
fields:
timer_id:
required: true
example: "01HZ8ABCDEF0123456789ABCDE"
selector:
text:
clear_finished_timers:
target:
entity:
domain: timer_list
get_timers:
target:
entity:
domain: timer_list
fields:
status:
example: "active"
selector:
select:
translation_key: status
multiple: true
options:
- active
- paused
- finished
- cancelled
@@ -0,0 +1,154 @@
{
"entity_component": {
"_": {
"name": "[%key:component::timer_list::title%]"
}
},
"exceptions": {
"timer_not_found": {
"message": "Unable to find timer with id: {timer_id}"
}
},
"selector": {
"finish_action": {
"options": {
"archive": "Archive",
"remove": "Remove",
"restart": "Restart"
}
},
"status": {
"options": {
"active": "Active",
"cancelled": "Cancelled",
"finished": "Finished",
"paused": "Paused"
}
}
},
"services": {
"add_time": {
"description": "Adds time to a timer.",
"fields": {
"duration": {
"description": "How much time to add to the timer.",
"name": "Duration"
},
"timer_id": {
"description": "The id of the timer to add time to.",
"name": "Timer ID"
}
},
"name": "Add time"
},
"cancel_all_timers": {
"description": "Cancels every active and paused timer on a timer list.",
"name": "Cancel all timers"
},
"cancel_timer": {
"description": "Cancels a timer.",
"fields": {
"timer_id": {
"description": "The id of the timer to cancel.",
"name": "Timer ID"
}
},
"name": "Cancel timer"
},
"clear_finished_timers": {
"description": "Removes all finished and cancelled timers from a timer list.",
"name": "Clear finished timers"
},
"get_timers": {
"description": "Gets the timers on a timer list.",
"fields": {
"status": {
"description": "Only return timers with the specified statuses.",
"name": "Status"
}
},
"name": "Get timers"
},
"pause_timer": {
"description": "Pauses an active timer.",
"fields": {
"timer_id": {
"description": "The id of the timer to pause.",
"name": "Timer ID"
}
},
"name": "Pause timer"
},
"remove_time": {
"description": "Removes time from a timer.",
"fields": {
"duration": {
"description": "How much time to remove from the timer.",
"name": "Duration"
},
"timer_id": {
"description": "The id of the timer to remove time from.",
"name": "Timer ID"
}
},
"name": "Remove time"
},
"remove_timer": {
"description": "Removes a timer from a timer list.",
"fields": {
"timer_id": {
"description": "The id of the timer to remove.",
"name": "Timer ID"
}
},
"name": "Remove timer"
},
"start_timer": {
"description": "Creates and starts a new timer on a timer list.",
"fields": {
"duration": {
"description": "How long the timer should run for.",
"name": "Duration"
},
"finish_action": {
"description": "What happens to the timer once it finishes.",
"name": "Finish action"
},
"name": {
"description": "Optional name for the timer.",
"name": "Name"
}
},
"name": "Start timer"
},
"unpause_timer": {
"description": "Resumes a paused timer.",
"fields": {
"timer_id": {
"description": "The id of the timer to resume.",
"name": "Timer ID"
}
},
"name": "Resume timer"
}
},
"title": "Timer list",
"triggers": {
"timer_cancelled": {
"description": "Triggers when a timer is cancelled on a timer list.",
"name": "Timer cancelled"
},
"timer_finished": {
"description": "Triggers when a timer finishes on a timer list.",
"name": "Timer finished"
},
"timer_started": {
"description": "Triggers when a timer is started on a timer list.",
"name": "Timer started"
},
"timer_updated": {
"description": "Triggers when a timer is paused, resumed, or has time added or removed.",
"name": "Timer updated"
}
}
}
@@ -0,0 +1,163 @@
"""Provides triggers for timer lists."""
from collections.abc import Callable
from functools import partial
from typing import TYPE_CHECKING, cast, override
import voluptuous as vol
from homeassistant.const import ATTR_ENTITY_ID, CONF_OPTIONS, CONF_TARGET
from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback, split_entity_id
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.target import TargetEntityChangeTracker, TargetSelection
from homeassistant.helpers.trigger import (
Trigger,
TriggerActionRunner,
TriggerConfig,
TriggerNotTriggeredReporter,
)
from homeassistant.helpers.typing import ConfigType
from homeassistant.util import dt as dt_util
from . import TimerListEvent, timer_to_dict
from .const import ATTR_TIMER, DATA_COMPONENT, DOMAIN, TimerListEventType
TRIGGER_SCHEMA = vol.Schema(
{
vol.Required(CONF_TARGET): cv.TARGET_FIELDS,
vol.Required(CONF_OPTIONS, default={}): {},
}
)
class TimerEventListener(TargetEntityChangeTracker):
"""Subscribe to timer change events for the targeted timer list entities."""
def __init__(
self,
hass: HomeAssistant,
target_selection: TargetSelection,
listener: Callable[[str, TimerListEvent], None],
) -> None:
"""Initialize the listener."""
def entity_filter(entities: set[str]) -> set[str]:
return {
entity_id
for entity_id in entities
if split_entity_id(entity_id)[0] == DOMAIN
}
super().__init__(hass, target_selection, entity_filter)
self._listener = listener
self._unsubscribe_listeners: list[CALLBACK_TYPE] = []
@override
@callback
def _handle_entities_update(self, tracked_entities: set[str]) -> None:
"""Resubscribe when the set of tracked entities changes."""
for unsub in self._unsubscribe_listeners:
unsub()
self._unsubscribe_listeners = []
component = self._hass.data[DATA_COMPONENT]
for entity_id in tracked_entities:
if (entity := component.get_entity(entity_id)) is None:
continue
self._unsubscribe_listeners.append(
entity.async_subscribe_updates(partial(self._listener, entity_id))
)
@override
@callback
def _unsubscribe(self) -> None:
"""Unsubscribe from all events."""
super()._unsubscribe()
for unsub in self._unsubscribe_listeners:
unsub()
self._unsubscribe_listeners = []
class TimerEventTrigger(Trigger):
"""Trigger that fires on a specific timer change event type."""
_event_type: TimerListEventType
@override
@classmethod
async def async_validate_config(
cls, hass: HomeAssistant, config: ConfigType
) -> ConfigType:
"""Validate config."""
return cast(ConfigType, TRIGGER_SCHEMA(config))
def __init__(self, hass: HomeAssistant, config: TriggerConfig) -> None:
"""Initialize the trigger."""
super().__init__(hass, config)
if TYPE_CHECKING:
assert config.target is not None
self._target = config.target
@override
async def async_attach_runner(
self,
run_action: TriggerActionRunner,
did_not_trigger: TriggerNotTriggeredReporter | None = None,
) -> CALLBACK_TYPE:
"""Attach the trigger to an action runner."""
target_selection = TargetSelection(self._target)
if not target_selection.has_any_target:
raise HomeAssistantError(f"No target defined in {self._target}")
@callback
def handle_event(entity_id: str, event: TimerListEvent) -> None:
if event.event_type != self._event_type:
return
run_action(
{
ATTR_ENTITY_ID: entity_id,
ATTR_TIMER: timer_to_dict(event.item, dt_util.utcnow()),
},
f"timer {self._event_type.value} on {entity_id}",
)
listener = TimerEventListener(self._hass, target_selection, handle_event)
return await listener.async_setup()
class TimerStartedTrigger(TimerEventTrigger):
"""Trigger when a timer starts."""
_event_type = TimerListEventType.STARTED
class TimerUpdatedTrigger(TimerEventTrigger):
"""Trigger when a timer is paused, resumed, or has time added/removed."""
_event_type = TimerListEventType.UPDATED
class TimerFinishedTrigger(TimerEventTrigger):
"""Trigger when a timer finishes."""
_event_type = TimerListEventType.FINISHED
class TimerCancelledTrigger(TimerEventTrigger):
"""Trigger when a timer is cancelled."""
_event_type = TimerListEventType.CANCELLED
TRIGGERS: dict[str, type[Trigger]] = {
"timer_started": TimerStartedTrigger,
"timer_updated": TimerUpdatedTrigger,
"timer_finished": TimerFinishedTrigger,
"timer_cancelled": TimerCancelledTrigger,
}
async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
"""Return the triggers for timer lists."""
return TRIGGERS
@@ -0,0 +1,9 @@
.trigger_common: &trigger_common
target:
entity:
domain: timer_list
timer_started: *trigger_common
timer_updated: *trigger_common
timer_finished: *trigger_common
timer_cancelled: *trigger_common
+1
View File
@@ -12,6 +12,7 @@ FLOWS = {
"group",
"history_stats",
"integration",
"local_timer_list",
"min_max",
"mold_indicator",
"otp",
+1
View File
@@ -46,6 +46,7 @@ class EntityPlatforms(StrEnum):
SWITCH = "switch"
TEXT = "text"
TIME = "time"
TIMER_LIST = "timer_list"
TODO = "todo"
TTS = "tts"
UPDATE = "update"
@@ -8448,6 +8448,12 @@
"config_flow": true,
"iot_class": "local_push"
},
"local_timer_list": {
"name": "Local Timer list",
"integration_type": "helper",
"config_flow": true,
"iot_class": "local_push"
},
"manual": {
"name": "Manual Alarm Control Panel",
"integration_type": "helper",
@@ -101,6 +101,7 @@ _ENTITY_COMPONENTS: set[str] = set(ENTITY_COMPONENTS).union(
"script",
"tag",
"timer",
"timer_list",
}
)
+1
View File
@@ -125,6 +125,7 @@ NO_IOT_CLASS = [
"tag",
"temperature",
"timer",
"timer_list",
"trace",
"web_rtc",
"webhook",
+2
View File
@@ -2109,6 +2109,8 @@ NO_QUALITY_SCALE = [
"tag",
"temperature",
"timer",
"timer_list",
"local_timer_list",
"trace",
"usage_prediction",
"web_rtc",
@@ -0,0 +1 @@
"""Tests for the Local Timer list integration."""
@@ -0,0 +1,28 @@
"""Test the Local Timer list config flow."""
from homeassistant.components.local_timer_list.const import CONF_TIMER_LIST_NAME, DOMAIN
from homeassistant.config_entries import SOURCE_USER
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
async def test_user_flow_creates_entity(hass: HomeAssistant) -> None:
"""Test the user config flow creates an entry and a named timer list entity."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_TIMER_LIST_NAME: "Kitchen"}
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "Kitchen"
assert result["data"] == {CONF_TIMER_LIST_NAME: "Kitchen"}
state = hass.states.get("timer_list.kitchen")
assert state is not None
assert state.state == "0"
+61
View File
@@ -0,0 +1,61 @@
"""Tests for the Timer list integration."""
from homeassistant.components.timer_list import TimerListEntity
from homeassistant.components.timer_list.const import DOMAIN, TimerListEntityFeature
from homeassistant.config_entries import ConfigEntry, ConfigFlow
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from tests.common import MockConfigEntry, MockPlatform, mock_platform
TEST_DOMAIN = "test"
ALL_FEATURES = (
TimerListEntityFeature.START_TIMER
| TimerListEntityFeature.PAUSE_TIMER
| TimerListEntityFeature.CANCEL_TIMER
| TimerListEntityFeature.ADD_TIME
)
class MockFlow(ConfigFlow):
"""Test flow."""
class MockTimerListEntity(TimerListEntity):
"""Test timer list entity."""
_attr_supported_features = ALL_FEATURES
def __init__(self, name: str = "Timers") -> None:
"""Initialize entity."""
super().__init__()
self._attr_name = name
async def create_mock_platform(
hass: HomeAssistant,
entities: list[TimerListEntity],
) -> MockConfigEntry:
"""Create a timer_list platform with the specified entities."""
async def async_setup_entry_platform(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up test timer_list platform via config entry."""
async_add_entities(entities)
mock_platform(
hass,
f"{TEST_DOMAIN}.{DOMAIN}",
MockPlatform(async_setup_entry=async_setup_entry_platform),
)
config_entry = MockConfigEntry(domain=TEST_DOMAIN)
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
return config_entry
+64
View File
@@ -0,0 +1,64 @@
"""Fixtures for the Timer list component tests."""
from collections.abc import Generator
import pytest
from homeassistant.components.timer_list import TimerListEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from . import TEST_DOMAIN, MockFlow, MockTimerListEntity, create_mock_platform
from tests.common import MockModule, mock_config_flow, mock_integration, mock_platform
@pytest.fixture(autouse=True)
def config_flow_fixture(hass: HomeAssistant) -> Generator[None]:
"""Mock config flow."""
mock_platform(hass, f"{TEST_DOMAIN}.config_flow")
with mock_config_flow(TEST_DOMAIN, MockFlow):
yield
@pytest.fixture(autouse=True)
def mock_setup_integration(hass: HomeAssistant) -> None:
"""Fixture to set up a mock integration."""
async def async_setup_entry_init(
hass: HomeAssistant, config_entry: ConfigEntry
) -> bool:
"""Set up test config entry."""
await hass.config_entries.async_forward_entry_setups(
config_entry, [Platform.TIMER_LIST]
)
return True
async def async_unload_entry_init(
hass: HomeAssistant, config_entry: ConfigEntry
) -> bool:
await hass.config_entries.async_unload_platforms(
config_entry, [Platform.TIMER_LIST]
)
return True
mock_platform(hass, f"{TEST_DOMAIN}.config_flow")
mock_integration(
hass,
MockModule(
TEST_DOMAIN,
async_setup_entry=async_setup_entry_init,
async_unload_entry=async_unload_entry_init,
),
)
@pytest.fixture(name="test_entity")
async def mock_test_entity(hass: HomeAssistant) -> TimerListEntity:
"""Fixture that creates a test timer list entity."""
entity = MockTimerListEntity()
entity.entity_id = "timer_list.timers"
await create_mock_platform(hass, [entity])
return entity
+329
View File
@@ -0,0 +1,329 @@
"""Tests for the Timer list integration."""
from datetime import timedelta
from typing import Any
from freezegun.api import FrozenDateTimeFactory
import pytest
from homeassistant.components.timer_list import TimerListEntity
from homeassistant.components.timer_list.const import DOMAIN
from homeassistant.const import ATTR_ENTITY_ID, ATTR_NAME
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ServiceValidationError
from tests.common import async_fire_time_changed
from tests.typing import WebSocketGenerator
TEST_ENTITY_ID = "timer_list.timers"
async def _start_timer(
hass: HomeAssistant,
*,
duration: int = 60,
name: str | None = None,
finish_action: str = "remove",
) -> str:
"""Start a timer and return its id."""
data: dict[str, Any] = {
"duration": {"seconds": duration},
"finish_action": finish_action,
}
if name is not None:
data[ATTR_NAME] = name
result = await hass.services.async_call(
DOMAIN,
"start_timer",
data,
target={ATTR_ENTITY_ID: TEST_ENTITY_ID},
blocking=True,
return_response=True,
)
return result[TEST_ENTITY_ID]["timer_id"]
async def _get_timers(
hass: HomeAssistant, status: list[str] | None = None
) -> list[dict[str, Any]]:
"""Return the timers via the get_timers service."""
data: dict[str, Any] = {}
if status is not None:
data["status"] = status
result = await hass.services.async_call(
DOMAIN,
"get_timers",
data,
target={ATTR_ENTITY_ID: TEST_ENTITY_ID},
blocking=True,
return_response=True,
)
return result[TEST_ENTITY_ID]["timers"]
async def _call(hass: HomeAssistant, service: str, **fields: Any) -> None:
"""Call an entity service targeting the test entity."""
await hass.services.async_call(
DOMAIN,
service,
fields,
target={ATTR_ENTITY_ID: TEST_ENTITY_ID},
blocking=True,
)
@pytest.mark.usefixtures("test_entity")
async def test_start_timer_sets_state_and_returns_id(hass: HomeAssistant) -> None:
"""Test starting timers updates the state and returns an id."""
assert hass.states.get(TEST_ENTITY_ID).state == "0"
timer_id = await _start_timer(hass, name="Pasta")
assert timer_id
assert hass.states.get(TEST_ENTITY_ID).state == "1"
await _start_timer(hass)
assert hass.states.get(TEST_ENTITY_ID).state == "2"
timers = await _get_timers(hass)
assert len(timers) == 2
assert {timer["status"] for timer in timers} == {"active"}
assert timers[0]["name"] == "Pasta"
@pytest.mark.usefixtures("test_entity")
async def test_get_timers_status_filter(hass: HomeAssistant) -> None:
"""Test the get_timers status filter."""
await _start_timer(hass)
paused_id = await _start_timer(hass)
await _call(hass, "pause_timer", timer_id=paused_id)
assert len(await _get_timers(hass, status=["active"])) == 1
assert len(await _get_timers(hass, status=["paused"])) == 1
assert len(await _get_timers(hass, status=["active", "paused"])) == 2
@pytest.mark.usefixtures("test_entity")
async def test_finish_action_remove(
hass: HomeAssistant, freezer: FrozenDateTimeFactory
) -> None:
"""Test a timer is removed after finishing with the remove action."""
await _start_timer(hass, duration=60, finish_action="remove")
assert hass.states.get(TEST_ENTITY_ID).state == "1"
freezer.tick(timedelta(seconds=61))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert hass.states.get(TEST_ENTITY_ID).state == "0"
assert await _get_timers(hass) == []
@pytest.mark.usefixtures("test_entity")
async def test_finish_action_archive(
hass: HomeAssistant, freezer: FrozenDateTimeFactory
) -> None:
"""Test a timer is retained as finished with the archive action."""
await _start_timer(hass, duration=60, finish_action="archive")
freezer.tick(timedelta(seconds=61))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert hass.states.get(TEST_ENTITY_ID).state == "0"
timers = await _get_timers(hass)
assert len(timers) == 1
assert timers[0]["status"] == "finished"
assert timers[0]["finished_at"] is not None
@pytest.mark.usefixtures("test_entity")
async def test_finish_action_restart(
hass: HomeAssistant, freezer: FrozenDateTimeFactory
) -> None:
"""Test a timer restarts itself with the restart action."""
timer_id = await _start_timer(hass, duration=60, finish_action="restart")
freezer.tick(timedelta(seconds=61))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert hass.states.get(TEST_ENTITY_ID).state == "1"
timers = await _get_timers(hass)
assert len(timers) == 1
assert timers[0]["timer_id"] == timer_id
assert timers[0]["status"] == "active"
@pytest.mark.usefixtures("test_entity")
async def test_pause_and_unpause(hass: HomeAssistant) -> None:
"""Test pausing and resuming a timer."""
timer_id = await _start_timer(hass)
await _call(hass, "pause_timer", timer_id=timer_id)
assert hass.states.get(TEST_ENTITY_ID).state == "0"
timers = await _get_timers(hass)
assert timers[0]["status"] == "paused"
assert timers[0]["finishes_at"] is None
await _call(hass, "unpause_timer", timer_id=timer_id)
assert hass.states.get(TEST_ENTITY_ID).state == "1"
timers = await _get_timers(hass)
assert timers[0]["status"] == "active"
assert timers[0]["finishes_at"] is not None
@pytest.mark.usefixtures("test_entity")
async def test_add_and_remove_time(
hass: HomeAssistant, freezer: FrozenDateTimeFactory
) -> None:
"""Test adding and removing time on a timer."""
timer_id = await _start_timer(hass, duration=60)
await _call(hass, "add_time", timer_id=timer_id, duration={"seconds": 60})
assert (await _get_timers(hass))[0]["remaining"] == pytest.approx(120, abs=1)
await _call(hass, "remove_time", timer_id=timer_id, duration={"seconds": 90})
assert (await _get_timers(hass))[0]["remaining"] == pytest.approx(30, abs=1)
@pytest.mark.usefixtures("test_entity")
async def test_remove_time_finishes_timer(hass: HomeAssistant) -> None:
"""Test removing more time than remaining finishes the timer immediately."""
timer_id = await _start_timer(hass, duration=60, finish_action="archive")
await _call(hass, "remove_time", timer_id=timer_id, duration={"seconds": 120})
assert hass.states.get(TEST_ENTITY_ID).state == "0"
assert (await _get_timers(hass))[0]["status"] == "finished"
@pytest.mark.usefixtures("test_entity")
async def test_cancel_timer_remove(hass: HomeAssistant) -> None:
"""Test cancelling a remove-action timer deletes it."""
timer_id = await _start_timer(hass, finish_action="remove")
await _call(hass, "cancel_timer", timer_id=timer_id)
assert hass.states.get(TEST_ENTITY_ID).state == "0"
assert await _get_timers(hass) == []
@pytest.mark.usefixtures("test_entity")
async def test_cancel_timer_archive(hass: HomeAssistant) -> None:
"""Test cancelling an archive-action timer retains it as cancelled."""
timer_id = await _start_timer(hass, finish_action="archive")
await _call(hass, "cancel_timer", timer_id=timer_id)
assert hass.states.get(TEST_ENTITY_ID).state == "0"
timers = await _get_timers(hass)
assert len(timers) == 1
assert timers[0]["status"] == "cancelled"
@pytest.mark.usefixtures("test_entity")
async def test_cancel_all_timers(hass: HomeAssistant) -> None:
"""Test cancelling all timers."""
await _start_timer(hass)
await _start_timer(hass, finish_action="archive")
await _call(hass, "cancel_all_timers")
assert hass.states.get(TEST_ENTITY_ID).state == "0"
# The archived timer is retained as cancelled, the remove timer is deleted.
assert len(await _get_timers(hass)) == 1
@pytest.mark.usefixtures("test_entity")
async def test_clear_finished_timers(hass: HomeAssistant) -> None:
"""Test clearing finished and cancelled timers."""
timer_id = await _start_timer(hass, finish_action="archive")
await _call(hass, "cancel_timer", timer_id=timer_id)
await _start_timer(hass)
assert len(await _get_timers(hass)) == 2
await _call(hass, "clear_finished_timers")
timers = await _get_timers(hass)
assert len(timers) == 1
assert timers[0]["status"] == "active"
@pytest.mark.usefixtures("test_entity")
async def test_remove_timer(hass: HomeAssistant) -> None:
"""Test removing a single timer regardless of status."""
timer_id = await _start_timer(hass)
await _call(hass, "remove_timer", timer_id=timer_id)
assert hass.states.get(TEST_ENTITY_ID).state == "0"
assert await _get_timers(hass) == []
@pytest.mark.usefixtures("test_entity")
async def test_timer_not_found(hass: HomeAssistant) -> None:
"""Test acting on an unknown timer id raises."""
with pytest.raises(ServiceValidationError):
await _call(hass, "pause_timer", timer_id="does-not-exist")
@pytest.mark.usefixtures("test_entity")
async def test_websocket_subscribe(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test subscribing to timer changes with an initial snapshot."""
client = await hass_ws_client(hass)
await client.send_json_auto_id(
{"type": "timer_list/item/subscribe", "entity_id": TEST_ENTITY_ID}
)
msg = await client.receive_json()
assert msg["success"]
msg = await client.receive_json()
assert msg["event"] == {"type": "timers", "timers": []}
timer_id = await _start_timer(hass, name="Pasta")
msg = await client.receive_json()
assert msg["event"]["type"] == "change"
assert msg["event"]["event_type"] == "started"
assert msg["event"]["timer"]["timer_id"] == timer_id
assert msg["event"]["timer"]["name"] == "Pasta"
await _call(hass, "cancel_timer", timer_id=timer_id)
msg = await client.receive_json()
assert msg["event"]["event_type"] == "cancelled"
# remove-action timers also emit a removed event after cancellation.
msg = await client.receive_json()
assert msg["event"]["event_type"] == "removed"
@pytest.mark.usefixtures("test_entity")
async def test_websocket_list(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test the one-shot websocket list command."""
await _start_timer(hass, name="Pasta")
client = await hass_ws_client(hass)
await client.send_json_auto_id(
{"type": "timer_list/item/list", "entity_id": TEST_ENTITY_ID}
)
msg = await client.receive_json()
assert msg["success"]
assert len(msg["result"]["timers"]) == 1
assert msg["result"]["timers"][0]["name"] == "Pasta"
async def test_websocket_subscribe_unknown_entity(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
test_entity: TimerListEntity,
) -> None:
"""Test subscribing to an entity that does not exist."""
client = await hass_ws_client(hass)
await client.send_json_auto_id(
{"type": "timer_list/item/subscribe", "entity_id": "timer_list.unknown"}
)
msg = await client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "not_found"
+153
View File
@@ -0,0 +1,153 @@
"""Tests for the Timer list triggers."""
from datetime import timedelta
from freezegun.api import FrozenDateTimeFactory
import pytest
from homeassistant.components import automation
from homeassistant.components.timer_list.const import DOMAIN
from homeassistant.const import (
ATTR_ENTITY_ID,
CONF_ENTITY_ID,
CONF_PLATFORM,
CONF_TARGET,
)
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.setup import async_setup_component
from . import MockTimerListEntity, create_mock_platform
from tests.common import async_fire_time_changed, async_mock_service
from tests.components.common import assert_trigger_options_supported
TEST_ENTITY_ID = "timer_list.timers"
@pytest.fixture
def service_calls(hass: HomeAssistant) -> list[ServiceCall]:
"""Track calls to a mock service."""
return async_mock_service(hass, "test", "automation")
@pytest.fixture(autouse=True)
async def setup_entity(hass: HomeAssistant) -> None:
"""Create a timer list entity via the mock platform."""
entity = MockTimerListEntity()
entity.entity_id = TEST_ENTITY_ID
entity._attr_unique_id = "timers"
await create_mock_platform(hass, [entity])
async def _setup_automation(hass: HomeAssistant, trigger_type: str) -> None:
"""Set up an automation for the given timer list trigger."""
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: {
"triggers": [
{
CONF_PLATFORM: f"{DOMAIN}.{trigger_type}",
CONF_TARGET: {CONF_ENTITY_ID: TEST_ENTITY_ID},
}
],
"action": {
"service": "test.automation",
"data": {
"entity_id": "{{ trigger.entity_id }}",
"timer_id": "{{ trigger.timer.timer_id }}",
"status": "{{ trigger.timer.status }}",
},
},
}
},
)
await hass.async_block_till_done()
async def _start_timer(hass: HomeAssistant, finish_action: str = "remove") -> str:
"""Start a timer and return its id."""
result = await hass.services.async_call(
DOMAIN,
"start_timer",
{"duration": {"seconds": 60}, "finish_action": finish_action},
target={ATTR_ENTITY_ID: TEST_ENTITY_ID},
blocking=True,
return_response=True,
)
return result[TEST_ENTITY_ID]["timer_id"]
async def test_timer_finished_trigger(
hass: HomeAssistant,
service_calls: list[ServiceCall],
freezer: FrozenDateTimeFactory,
) -> None:
"""Test the timer_finished trigger fires when a timer finishes."""
await _setup_automation(hass, "timer_finished")
timer_id = await _start_timer(hass)
assert len(service_calls) == 0
freezer.tick(timedelta(seconds=61))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(service_calls) == 1
assert service_calls[0].data == {
"entity_id": TEST_ENTITY_ID,
"timer_id": timer_id,
"status": "finished",
}
async def test_timer_started_trigger(
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test the timer_started trigger fires when a timer starts."""
await _setup_automation(hass, "timer_started")
timer_id = await _start_timer(hass)
await hass.async_block_till_done()
assert len(service_calls) == 1
assert service_calls[0].data["timer_id"] == timer_id
assert service_calls[0].data["status"] == "active"
async def test_timer_cancelled_trigger(
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test the timer_cancelled trigger fires and ignores other events."""
await _setup_automation(hass, "timer_cancelled")
timer_id = await _start_timer(hass)
await hass.async_block_till_done()
assert len(service_calls) == 0
await hass.services.async_call(
DOMAIN,
"cancel_timer",
{"timer_id": timer_id},
target={ATTR_ENTITY_ID: TEST_ENTITY_ID},
blocking=True,
)
assert len(service_calls) == 1
assert service_calls[0].data["status"] == "cancelled"
async def test_trigger_options_supported(hass: HomeAssistant) -> None:
"""Test the timer list triggers do not advertise behavior or duration."""
for trigger_type in (
"timer_started",
"timer_updated",
"timer_finished",
"timer_cancelled",
):
await assert_trigger_options_supported(
hass,
f"{DOMAIN}.{trigger_type}",
None,
supports_behavior=False,
supports_duration=False,
)
+2
View File
@@ -94,6 +94,7 @@
'text',
'time',
'timer',
'timer_list',
'trace',
'tts',
'update',
@@ -202,6 +203,7 @@
'text',
'time',
'timer',
'timer_list',
'trace',
'tts',
'update',