mirror of
https://github.com/home-assistant/core.git
synced 2025-07-27 23:27:37 +00:00
Add weekdays to time trigger (#147505)
Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
parent
6a7f4953cd
commit
c0368f2448
@ -16,6 +16,7 @@ from homeassistant.const import (
|
||||
CONF_PLATFORM,
|
||||
STATE_UNAVAILABLE,
|
||||
STATE_UNKNOWN,
|
||||
WEEKDAYS,
|
||||
)
|
||||
from homeassistant.core import (
|
||||
CALLBACK_TYPE,
|
||||
@ -37,6 +38,8 @@ from homeassistant.helpers.trigger import TriggerActionType, TriggerInfo
|
||||
from homeassistant.helpers.typing import ConfigType
|
||||
from homeassistant.util import dt as dt_util
|
||||
|
||||
CONF_WEEKDAY = "weekday"
|
||||
|
||||
_TIME_TRIGGER_ENTITY = vol.All(str, cv.entity_domain(["input_datetime", "sensor"]))
|
||||
_TIME_AT_SCHEMA = vol.Any(cv.time, _TIME_TRIGGER_ENTITY)
|
||||
|
||||
@ -74,6 +77,10 @@ TRIGGER_SCHEMA = cv.TRIGGER_BASE_SCHEMA.extend(
|
||||
{
|
||||
vol.Required(CONF_PLATFORM): "time",
|
||||
vol.Required(CONF_AT): vol.All(cv.ensure_list, [_TIME_TRIGGER_SCHEMA]),
|
||||
vol.Optional(CONF_WEEKDAY): vol.Any(
|
||||
vol.In(WEEKDAYS),
|
||||
vol.All(cv.ensure_list, [vol.In(WEEKDAYS)]),
|
||||
),
|
||||
}
|
||||
)
|
||||
|
||||
@ -85,7 +92,7 @@ class TrackEntity(NamedTuple):
|
||||
callback: Callable
|
||||
|
||||
|
||||
async def async_attach_trigger(
|
||||
async def async_attach_trigger( # noqa: C901
|
||||
hass: HomeAssistant,
|
||||
config: ConfigType,
|
||||
action: TriggerActionType,
|
||||
@ -103,6 +110,18 @@ async def async_attach_trigger(
|
||||
description: str, now: datetime, *, entity_id: str | None = None
|
||||
) -> None:
|
||||
"""Listen for time changes and calls action."""
|
||||
# Check weekday filter if configured
|
||||
if CONF_WEEKDAY in config:
|
||||
weekday_config = config[CONF_WEEKDAY]
|
||||
current_weekday = WEEKDAYS[now.weekday()]
|
||||
|
||||
# Check if current weekday matches the configuration
|
||||
if isinstance(weekday_config, str):
|
||||
if current_weekday != weekday_config:
|
||||
return
|
||||
elif current_weekday not in weekday_config:
|
||||
return
|
||||
|
||||
hass.async_run_hass_job(
|
||||
job,
|
||||
{
|
||||
|
@ -1,6 +1,6 @@
|
||||
"""The tests for the time automation."""
|
||||
|
||||
from datetime import timedelta
|
||||
from datetime import datetime, timedelta
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
from freezegun.api import FrozenDateTimeFactory
|
||||
@ -877,3 +877,200 @@ async def test_if_at_template_limited_template(
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert "is not supported in limited templates" in caplog.text
|
||||
|
||||
|
||||
async def test_if_fires_using_weekday_single(
|
||||
hass: HomeAssistant,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
service_calls: list[ServiceCall],
|
||||
) -> None:
|
||||
"""Test for firing on a specific weekday."""
|
||||
# Freeze time to Monday, January 2, 2023 at 5:00:00
|
||||
monday_trigger = dt_util.as_utc(datetime(2023, 1, 2, 5, 0, 0, 0))
|
||||
|
||||
freezer.move_to(monday_trigger)
|
||||
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
automation.DOMAIN,
|
||||
{
|
||||
automation.DOMAIN: {
|
||||
"trigger": {"platform": "time", "at": "5:00:00", "weekday": "mon"},
|
||||
"action": {
|
||||
"service": "test.automation",
|
||||
"data_template": {
|
||||
"some": "{{ trigger.platform }} - {{ trigger.now.strftime('%A') }}",
|
||||
},
|
||||
},
|
||||
}
|
||||
},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
# Fire the trigger on Monday
|
||||
async_fire_time_changed(hass, monday_trigger + timedelta(seconds=1))
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert len(service_calls) == 1
|
||||
assert service_calls[0].data["some"] == "time - Monday"
|
||||
|
||||
# Fire on Tuesday at the same time - should not trigger
|
||||
tuesday_trigger = dt_util.as_utc(datetime(2023, 1, 3, 5, 0, 0, 0))
|
||||
async_fire_time_changed(hass, tuesday_trigger)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
# Should still be only 1 call
|
||||
assert len(service_calls) == 1
|
||||
|
||||
|
||||
async def test_if_fires_using_weekday_multiple(
|
||||
hass: HomeAssistant,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
service_calls: list[ServiceCall],
|
||||
) -> None:
|
||||
"""Test for firing on multiple weekdays."""
|
||||
# Freeze time to Monday, January 2, 2023 at 5:00:00
|
||||
monday_trigger = dt_util.as_utc(datetime(2023, 1, 2, 5, 0, 0, 0))
|
||||
|
||||
freezer.move_to(monday_trigger)
|
||||
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
automation.DOMAIN,
|
||||
{
|
||||
automation.DOMAIN: {
|
||||
"trigger": {
|
||||
"platform": "time",
|
||||
"at": "5:00:00",
|
||||
"weekday": ["mon", "wed", "fri"],
|
||||
},
|
||||
"action": {
|
||||
"service": "test.automation",
|
||||
"data_template": {
|
||||
"some": "{{ trigger.platform }} - {{ trigger.now.strftime('%A') }}",
|
||||
},
|
||||
},
|
||||
}
|
||||
},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
# Fire on Monday - should trigger
|
||||
async_fire_time_changed(hass, monday_trigger + timedelta(seconds=1))
|
||||
await hass.async_block_till_done()
|
||||
assert len(service_calls) == 1
|
||||
assert "Monday" in service_calls[0].data["some"]
|
||||
|
||||
# Fire on Tuesday - should not trigger
|
||||
tuesday_trigger = dt_util.as_utc(datetime(2023, 1, 3, 5, 0, 0, 0))
|
||||
async_fire_time_changed(hass, tuesday_trigger)
|
||||
await hass.async_block_till_done()
|
||||
assert len(service_calls) == 1
|
||||
|
||||
# Fire on Wednesday - should trigger
|
||||
wednesday_trigger = dt_util.as_utc(datetime(2023, 1, 4, 5, 0, 0, 0))
|
||||
async_fire_time_changed(hass, wednesday_trigger)
|
||||
await hass.async_block_till_done()
|
||||
assert len(service_calls) == 2
|
||||
assert "Wednesday" in service_calls[1].data["some"]
|
||||
|
||||
# Fire on Friday - should trigger
|
||||
friday_trigger = dt_util.as_utc(datetime(2023, 1, 6, 5, 0, 0, 0))
|
||||
async_fire_time_changed(hass, friday_trigger)
|
||||
await hass.async_block_till_done()
|
||||
assert len(service_calls) == 3
|
||||
assert "Friday" in service_calls[2].data["some"]
|
||||
|
||||
|
||||
async def test_if_fires_using_weekday_with_entity(
|
||||
hass: HomeAssistant,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
service_calls: list[ServiceCall],
|
||||
) -> None:
|
||||
"""Test for firing on weekday with input_datetime entity."""
|
||||
await async_setup_component(
|
||||
hass,
|
||||
"input_datetime",
|
||||
{"input_datetime": {"trigger": {"has_date": False, "has_time": True}}},
|
||||
)
|
||||
|
||||
# Freeze time to Monday, January 2, 2023 at 5:00:00
|
||||
monday_trigger = dt_util.as_utc(datetime(2023, 1, 2, 5, 0, 0, 0))
|
||||
|
||||
await hass.services.async_call(
|
||||
"input_datetime",
|
||||
"set_datetime",
|
||||
{
|
||||
ATTR_ENTITY_ID: "input_datetime.trigger",
|
||||
"time": "05:00:00",
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
freezer.move_to(monday_trigger)
|
||||
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
automation.DOMAIN,
|
||||
{
|
||||
automation.DOMAIN: {
|
||||
"trigger": {
|
||||
"platform": "time",
|
||||
"at": "input_datetime.trigger",
|
||||
"weekday": "mon",
|
||||
},
|
||||
"action": {
|
||||
"service": "test.automation",
|
||||
"data_template": {
|
||||
"some": "{{ trigger.platform }} - {{ trigger.now.strftime('%A') }}",
|
||||
"entity": "{{ trigger.entity_id }}",
|
||||
},
|
||||
},
|
||||
}
|
||||
},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
# Fire on Monday - should trigger
|
||||
async_fire_time_changed(hass, monday_trigger + timedelta(seconds=1))
|
||||
await hass.async_block_till_done()
|
||||
automation_calls = [call for call in service_calls if call.domain == "test"]
|
||||
assert len(automation_calls) == 1
|
||||
assert "Monday" in automation_calls[0].data["some"]
|
||||
assert automation_calls[0].data["entity"] == "input_datetime.trigger"
|
||||
|
||||
# Fire on Tuesday - should not trigger
|
||||
tuesday_trigger = dt_util.as_utc(datetime(2023, 1, 3, 5, 0, 0, 0))
|
||||
async_fire_time_changed(hass, tuesday_trigger)
|
||||
await hass.async_block_till_done()
|
||||
automation_calls = [call for call in service_calls if call.domain == "test"]
|
||||
assert len(automation_calls) == 1
|
||||
|
||||
|
||||
def test_weekday_validation() -> None:
|
||||
"""Test weekday validation in trigger schema."""
|
||||
# Valid single weekday
|
||||
valid_config = {"platform": "time", "at": "5:00:00", "weekday": "mon"}
|
||||
time.TRIGGER_SCHEMA(valid_config)
|
||||
|
||||
# Valid multiple weekdays
|
||||
valid_config = {
|
||||
"platform": "time",
|
||||
"at": "5:00:00",
|
||||
"weekday": ["mon", "wed", "fri"],
|
||||
}
|
||||
time.TRIGGER_SCHEMA(valid_config)
|
||||
|
||||
# Invalid weekday
|
||||
invalid_config = {"platform": "time", "at": "5:00:00", "weekday": "invalid"}
|
||||
with pytest.raises(vol.Invalid):
|
||||
time.TRIGGER_SCHEMA(invalid_config)
|
||||
|
||||
# Invalid weekday in list
|
||||
invalid_config = {
|
||||
"platform": "time",
|
||||
"at": "5:00:00",
|
||||
"weekday": ["mon", "invalid"],
|
||||
}
|
||||
with pytest.raises(vol.Invalid):
|
||||
time.TRIGGER_SCHEMA(invalid_config)
|
||||
|
Loading…
x
Reference in New Issue
Block a user