home-assistant/core
1
0
Fork 1
mirror of https://github.com/home-assistant/core.git synced 2026-01-25 09:02:38 +01:00
Code Issues Packages Projects Releases Wiki Activity
Files
dev
core/tests/components/calendar/test_trigger.py

1177 lines
39 KiB
Python
Raw Permalink Blame History

"""Tests for the calendar automation.
The tests create calendar based automations, set up a fake set of calendar
events, then advance time to exercise that the automation is called. The
tests use a fixture that mocks out events returned by the calendar entity,
and create events using a relative time offset and then advance the clock
forward exercising the triggers.
"""
from __future__ import annotations
from collections.abc import AsyncIterator, Callable, Generator
from contextlib import asynccontextmanager
from dataclasses import dataclass
import datetime
import logging
from typing import Any
from unittest.mock import patch
import zoneinfo
from freezegun.api import FrozenDateTimeFactory
import pytest
from homeassistant.components import automation, calendar
from homeassistant.components.calendar.trigger import (
CONF_OFFSET_TYPE,
EVENT_END,
EVENT_START,
OFFSET_TYPE_AFTER,
OFFSET_TYPE_BEFORE,
)
from homeassistant.const import (
ATTR_AREA_ID,
ATTR_DEVICE_ID,
ATTR_ENTITY_ID,
ATTR_LABEL_ID,
CONF_OFFSET,
CONF_OPTIONS,
CONF_PLATFORM,
CONF_TARGET,
SERVICE_TURN_OFF,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import (
area_registry as ar,
device_registry as dr,
entity_registry as er,
label_registry as lr,
)
from homeassistant.setup import async_setup_component
from homeassistant.util import dt as dt_util
from .conftest import MockCalendarEntity
from tests.common import (
MockConfigEntry,
async_fire_time_changed,
async_mock_service,
mock_device_registry,
)
_LOGGER = logging.getLogger(__name__)
@dataclass
class TriggerFormat:
"""Abstraction for different trigger configuration formats."""
id: str
def get_platform(self, event_type: str) -> str:
"""Get the platform string for trigger payload assertions."""
raise NotImplementedError
def get_trigger_data(
self, entity_id: str, event_type: str, offset: datetime.timedelta | None = None
) -> dict[str, Any]:
"""Get the trigger configuration data."""
raise NotImplementedError
def get_expected_call_data(
self, entity_id: str, event_type: str, calendar_event: dict[str, Any]
) -> dict[str, Any]:
"""Get the expected call data for assertion."""
return {
"platform": self.get_platform(event_type),
"event": event_type,
"entity_id": entity_id,
"calendar_event": calendar_event,
}
@dataclass
class LegacyTriggerFormat(TriggerFormat):
"""Legacy trigger format using platform: calendar with entity_id and event."""
id: str = "legacy"
def get_platform(self, event_type: str) -> str:
"""Get the platform string for trigger payload assertions."""
return calendar.DOMAIN
def get_trigger_data(
self, entity_id: str, event_type: str, offset: datetime.timedelta | None = None
) -> dict[str, Any]:
"""Get the trigger configuration data."""
trigger_data: dict[str, Any] = {
CONF_PLATFORM: calendar.DOMAIN,
"entity_id": entity_id,
"event": event_type,
}
if offset:
trigger_data[CONF_OFFSET] = offset
return trigger_data
@dataclass
class TargetTriggerFormat(TriggerFormat):
"""Target trigger format using platform: calendar.event_started/ended with target."""
id: str = "target"
def get_platform(self, event_type: str) -> str:
"""Get the platform string for trigger payload assertions."""
trigger_type = "event_started" if event_type == EVENT_START else "event_ended"
return f"{calendar.DOMAIN}.{trigger_type}"
def get_trigger_data(
self, entity_id: str, event_type: str, offset: datetime.timedelta | None = None
) -> dict[str, Any]:
"""Get the trigger configuration data."""
trigger_type = "event_started" if event_type == EVENT_START else "event_ended"
trigger_data: dict[str, Any] = {
CONF_PLATFORM: f"{calendar.DOMAIN}.{trigger_type}",
CONF_TARGET: {"entity_id": entity_id},
}
if offset:
options: dict[str, Any] = {}
# Convert signed offset to offset + offset_type
if offset < datetime.timedelta(0):
options[CONF_OFFSET] = -offset
options[CONF_OFFSET_TYPE] = OFFSET_TYPE_BEFORE
else:
options[CONF_OFFSET] = offset
options[CONF_OFFSET_TYPE] = OFFSET_TYPE_AFTER
trigger_data[CONF_OPTIONS] = options
return trigger_data
TRIGGER_FORMATS = [LegacyTriggerFormat(), TargetTriggerFormat()]
TRIGGER_FORMAT_IDS = [fmt.id for fmt in TRIGGER_FORMATS]
@pytest.fixture(params=TRIGGER_FORMATS, ids=TRIGGER_FORMAT_IDS)
def trigger_format(request: pytest.FixtureRequest) -> TriggerFormat:
"""Fixture providing both trigger formats for parameterized tests."""
return request.param
CALENDAR_ENTITY_ID = "calendar.calendar_2"
TEST_AUTOMATION_ACTION = {
"service": "test.automation",
"data": {
"platform": "{{ trigger.platform }}",
"event": "{{ trigger.event }}",
"entity_id": "{{ trigger.entity_id }}",
"calendar_event": "{{ trigger.calendar_event }}",
},
}
# The trigger sets two alarms: One based on the next event and one
# to refresh the schedule. The test advances the time an arbitrary
# amount to trigger either type of event with a small jitter.
TEST_TIME_ADVANCE_INTERVAL = datetime.timedelta(minutes=1)
TEST_UPDATE_INTERVAL = datetime.timedelta(minutes=7)
TARGET_TEST_FIRST_START_CALL_DATA = [
{
"platform": "calendar.event_started",
"event": "start",
"entity_id": "calendar.calendar_1",
"calendar_event": {
"start": "2022-04-19T11:00:00+00:00",
"end": "2022-04-19T11:30:00+00:00",
"summary": "Event on Calendar 1",
"all_day": False,
},
}
]
TARGET_TEST_SECOND_START_CALL_DATA = [
{
"platform": "calendar.event_started",
"event": "start",
"entity_id": "calendar.calendar_2",
"calendar_event": {
"start": "2022-04-19T11:15:00+00:00",
"end": "2022-04-19T11:45:00+00:00",
"summary": "Event on Calendar 2",
"all_day": False,
},
}
]
TARGET_TEST_FIRST_END_CALL_DATA = [
{
"platform": "calendar.event_ended",
"event": "end",
"entity_id": "calendar.calendar_1",
"calendar_event": {
"start": "2022-04-19T11:00:00+00:00",
"end": "2022-04-19T11:30:00+00:00",
"summary": "Event on Calendar 1",
"all_day": False,
},
}
]
TARGET_TEST_SECOND_END_CALL_DATA = [
{
"platform": "calendar.event_ended",
"event": "end",
"entity_id": "calendar.calendar_2",
"calendar_event": {
"start": "2022-04-19T11:15:00+00:00",
"end": "2022-04-19T11:45:00+00:00",
"summary": "Event on Calendar 2",
"all_day": False,
},
}
]
class FakeSchedule:
"""Test fixture class for return events in a specific date range."""
def __init__(self, hass: HomeAssistant, freezer: FrozenDateTimeFactory) -> None:
"""Initialize FakeSchedule."""
self.hass = hass
self.freezer = freezer
async def fire_time(self, trigger_time: datetime.datetime) -> None:
"""Fire an alarm and wait."""
_LOGGER.debug("Firing alarm @ %s", dt_util.as_local(trigger_time))
self.freezer.move_to(trigger_time)
async_fire_time_changed(self.hass, trigger_time)
await self.hass.async_block_till_done()
async def fire_until(self, end: datetime.datetime) -> None:
"""Simulate the passage of time by firing alarms until the time is reached."""
current_time = dt_util.as_utc(self.freezer())
if (end - current_time) > (TEST_UPDATE_INTERVAL * 2):
# Jump ahead to right before the target alarm them to remove
# unnecessary waiting, before advancing in smaller increments below.
# This leaves time for multiple update intervals to refresh the set
# of upcoming events
await self.fire_time(end - TEST_UPDATE_INTERVAL * 2)
while dt_util.utcnow() < end:
self.freezer.tick(TEST_TIME_ADVANCE_INTERVAL)
await self.fire_time(dt_util.utcnow())
@pytest.fixture
def fake_schedule(hass: HomeAssistant, freezer: FrozenDateTimeFactory) -> FakeSchedule:
"""Fixture that tests can use to make fake events."""
# Setup start time for all tests
freezer.move_to("2022-04-19 10:31:02+00:00")
return FakeSchedule(hass, freezer)
@pytest.fixture(name="test_entity")
def mock_test_entity(test_entities: list[MockCalendarEntity]) -> MockCalendarEntity:
"""Fixture to expose the calendar entity used in tests."""
return test_entities[1]
@pytest.fixture(name="setup_platform", autouse=True)
async def mock_setup_platform(
hass: HomeAssistant,
mock_setup_integration: None,
config_entry: MockConfigEntry,
) -> None:
"""Fixture to setup platforms used in the test."""
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
@pytest.fixture
def target_calendars(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
area_registry: ar.AreaRegistry,
label_registry: lr.LabelRegistry,
):
"""Associate calendar entities with different targets.
Sets up the following target structure:
- area_both: An area containing both calendar entities
- label_calendar_1: A label assigned to calendar 1 only
- device_calendar_1: A device associated with calendar 1
- device_calendar_2: A device associated with calendar 2
- area_devices: An area containing both devices
"""
area_both = area_registry.async_get_or_create("area_both_calendars")
label_calendar_1 = label_registry.async_create("calendar_1_label")
label_on_devices = label_registry.async_create("label_on_devices")
device_calendar_1 = dr.DeviceEntry(
id="device_calendar_1", labels=[label_on_devices.label_id]
)
device_calendar_2 = dr.DeviceEntry(
id="device_calendar_2", labels=[label_on_devices.label_id]
)
mock_device_registry(
hass,
{
device_calendar_1.id: device_calendar_1,
device_calendar_2.id: device_calendar_2,
},
)
# Associate calendar entities with targets
entity_registry.async_update_entity(
"calendar.calendar_1",
area_id=area_both.id,
labels={label_calendar_1.label_id},
device_id=device_calendar_1.id,
)
entity_registry.async_update_entity(
"calendar.calendar_2",
area_id=area_both.id,
device_id=device_calendar_2.id,
)
@asynccontextmanager
async def create_automation(
hass: HomeAssistant,
trigger_format: TriggerFormat,
event_type: str,
offset: datetime.timedelta | None = None,
) -> AsyncIterator[None]:
"""Register an automation using the specified trigger format."""
trigger_data = trigger_format.get_trigger_data(
CALENDAR_ENTITY_ID, event_type, offset
)
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: {
"alias": event_type,
"trigger": trigger_data,
"action": TEST_AUTOMATION_ACTION,
"mode": "queued",
}
},
)
await hass.async_block_till_done()
yield
# Disable automation to cleanup lingering timers
await hass.services.async_call(
automation.DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: f"automation.{event_type}"},
blocking=True,
)
@pytest.fixture
def calls_data(hass: HomeAssistant) -> Callable[[], list[dict[str, Any]]]:
"""Fixture to return payload data for automation calls."""
service_calls = async_mock_service(hass, "test", "automation")
def get_trigger_data() -> list[dict[str, Any]]:
return [c.data for c in service_calls]
return get_trigger_data
@pytest.fixture(autouse=True)
def mock_update_interval() -> Generator[None]:
"""Fixture to override the update interval for refreshing events."""
with patch(
"homeassistant.components.calendar.trigger.UPDATE_INTERVAL",
new=TEST_UPDATE_INTERVAL,
):
yield
async def test_event_start_trigger(
hass: HomeAssistant,
calls_data: Callable[[], list[dict[str, Any]]],
fake_schedule: FakeSchedule,
test_entity: MockCalendarEntity,
trigger_format: TriggerFormat,
) -> None:
"""Test the a calendar trigger based on start time."""
event_data = test_entity.create_event(
start=datetime.datetime.fromisoformat("2022-04-19 11:00:00+00:00"),
end=datetime.datetime.fromisoformat("2022-04-19 11:30:00+00:00"),
)
async with create_automation(hass, trigger_format, EVENT_START):
assert len(calls_data()) == 0
await fake_schedule.fire_until(
datetime.datetime.fromisoformat("2022-04-19 11:15:00+00:00"),
)
assert calls_data() == [
trigger_format.get_expected_call_data(
CALENDAR_ENTITY_ID, EVENT_START, event_data
)
]
@pytest.mark.parametrize(
("offset_delta"),
[
datetime.timedelta(hours=-1),
datetime.timedelta(hours=1),
],
)
async def test_event_start_trigger_with_offset(
hass: HomeAssistant,
calls_data: Callable[[], list[dict[str, Any]]],
fake_schedule: FakeSchedule,
test_entity: MockCalendarEntity,
trigger_format: TriggerFormat,
offset_delta: datetime.timedelta,
) -> None:
"""Test the a calendar trigger based on start time with an offset."""
event_data = test_entity.create_event(
start=datetime.datetime.fromisoformat("2022-04-19 12:00:00+00:00"),
end=datetime.datetime.fromisoformat("2022-04-19 12:30:00+00:00"),
)
async with create_automation(
hass, trigger_format, EVENT_START, offset=offset_delta
):
# No calls yet
await fake_schedule.fire_until(
datetime.datetime.fromisoformat("2022-04-19 11:55:00+00:00") + offset_delta,
)
assert len(calls_data()) == 0
# Event has started w/ offset
await fake_schedule.fire_until(
datetime.datetime.fromisoformat("2022-04-19 12:05:00+00:00") + offset_delta,
)
assert calls_data() == [
trigger_format.get_expected_call_data(
CALENDAR_ENTITY_ID, EVENT_START, event_data
)
]
async def test_event_end_trigger(
hass: HomeAssistant,
calls_data: Callable[[], list[dict[str, Any]]],
fake_schedule: FakeSchedule,
test_entity: MockCalendarEntity,
trigger_format: TriggerFormat,
) -> None:
"""Test the a calendar trigger based on end time."""
event_data = test_entity.create_event(
start=datetime.datetime.fromisoformat("2022-04-19 11:00:00+00:00"),
end=datetime.datetime.fromisoformat("2022-04-19 12:00:00+00:00"),
)
async with create_automation(hass, trigger_format, EVENT_END):
# Event started, nothing should fire yet
await fake_schedule.fire_until(
datetime.datetime.fromisoformat("2022-04-19 11:10:00+00:00")
)
assert len(calls_data()) == 0
# Event ends
await fake_schedule.fire_until(
datetime.datetime.fromisoformat("2022-04-19 12:10:00+00:00")
)
assert calls_data() == [
trigger_format.get_expected_call_data(
CALENDAR_ENTITY_ID, EVENT_END, event_data
)
]
@pytest.mark.parametrize(
("offset_delta"),
[
datetime.timedelta(hours=-1),
datetime.timedelta(hours=1),
],
)
async def test_event_end_trigger_with_offset(
hass: HomeAssistant,
calls_data: Callable[[], list[dict[str, Any]]],
fake_schedule: FakeSchedule,
test_entity: MockCalendarEntity,
trigger_format: TriggerFormat,
offset_delta: datetime.timedelta,
) -> None:
"""Test the a calendar trigger based on end time with an offset."""
event_data = test_entity.create_event(
start=datetime.datetime.fromisoformat("2022-04-19 12:00:00+00:00"),
end=datetime.datetime.fromisoformat("2022-04-19 12:30:00+00:00"),
)
async with create_automation(hass, trigger_format, EVENT_END, offset=offset_delta):
# No calls yet
await fake_schedule.fire_until(
datetime.datetime.fromisoformat("2022-04-19 12:05:00+00:00") + offset_delta,
)
assert len(calls_data()) == 0
# Event has started w/ offset
await fake_schedule.fire_until(
datetime.datetime.fromisoformat("2022-04-19 12:35:00+00:00") + offset_delta,
)
assert calls_data() == [
trigger_format.get_expected_call_data(
CALENDAR_ENTITY_ID, EVENT_END, event_data
)
]
async def test_calendar_trigger_with_no_events(
hass: HomeAssistant,
calls_data: Callable[[], list[dict[str, Any]]],
fake_schedule: FakeSchedule,
trigger_format: TriggerFormat,
) -> None:
"""Test a calendar trigger setup with no events."""
async with (
create_automation(hass, trigger_format, EVENT_START),
create_automation(hass, trigger_format, EVENT_END),
):
# No calls, at arbitrary times
await fake_schedule.fire_until(
datetime.datetime.fromisoformat("2022-04-19 11:00:00+00:00")
)
assert len(calls_data()) == 0
async def test_multiple_start_events(
hass: HomeAssistant,
calls_data: Callable[[], list[dict[str, Any]]],
fake_schedule: FakeSchedule,
test_entity: MockCalendarEntity,
trigger_format: TriggerFormat,
) -> None:
"""Test that a trigger fires for multiple events."""
event_data1 = test_entity.create_event(
start=datetime.datetime.fromisoformat("2022-04-19 10:45:00+00:00"),
end=datetime.datetime.fromisoformat("2022-04-19 11:00:00+00:00"),
)
event_data2 = test_entity.create_event(
start=datetime.datetime.fromisoformat("2022-04-19 11:00:00+00:00"),
end=datetime.datetime.fromisoformat("2022-04-19 11:15:00+00:00"),
)
async with create_automation(hass, trigger_format, EVENT_START):
await fake_schedule.fire_until(
datetime.datetime.fromisoformat("2022-04-19 11:30:00+00:00")
)
assert calls_data() == [
trigger_format.get_expected_call_data(
CALENDAR_ENTITY_ID, EVENT_START, event_data1
),
trigger_format.get_expected_call_data(
CALENDAR_ENTITY_ID, EVENT_START, event_data2
),
]
async def test_multiple_end_events(
hass: HomeAssistant,
calls_data: Callable[[], list[dict[str, Any]]],
fake_schedule: FakeSchedule,
test_entity: MockCalendarEntity,
trigger_format: TriggerFormat,
) -> None:
"""Test that a trigger fires for multiple events."""
event_data1 = test_entity.create_event(
start=datetime.datetime.fromisoformat("2022-04-19 10:45:00+00:00"),
end=datetime.datetime.fromisoformat("2022-04-19 11:00:00+00:00"),
)
event_data2 = test_entity.create_event(
start=datetime.datetime.fromisoformat("2022-04-19 11:00:00+00:00"),
end=datetime.datetime.fromisoformat("2022-04-19 11:15:00+00:00"),
)
async with create_automation(hass, trigger_format, EVENT_END):
await fake_schedule.fire_until(
datetime.datetime.fromisoformat("2022-04-19 11:30:00+00:00")
)
assert calls_data() == [
trigger_format.get_expected_call_data(
CALENDAR_ENTITY_ID, EVENT_END, event_data1
),
trigger_format.get_expected_call_data(
CALENDAR_ENTITY_ID, EVENT_END, event_data2
),
]
async def test_multiple_events_sharing_start_time(
hass: HomeAssistant,
calls_data: Callable[[], list[dict[str, Any]]],
fake_schedule: FakeSchedule,
test_entity: MockCalendarEntity,
trigger_format: TriggerFormat,
) -> None:
"""Test that a trigger fires for every event sharing a start time."""
event_data1 = test_entity.create_event(
start=datetime.datetime.fromisoformat("2022-04-19 11:00:00+00:00"),
end=datetime.datetime.fromisoformat("2022-04-19 11:30:00+00:00"),
)
event_data2 = test_entity.create_event(
start=datetime.datetime.fromisoformat("2022-04-19 11:00:00+00:00"),
end=datetime.datetime.fromisoformat("2022-04-19 11:30:00+00:00"),
)
async with create_automation(hass, trigger_format, EVENT_START):
await fake_schedule.fire_until(
datetime.datetime.fromisoformat("2022-04-19 11:35:00+00:00")
)
assert calls_data() == [
trigger_format.get_expected_call_data(
CALENDAR_ENTITY_ID, EVENT_START, event_data1
),
trigger_format.get_expected_call_data(
CALENDAR_ENTITY_ID, EVENT_START, event_data2
),
]
async def test_overlap_events(
hass: HomeAssistant,
calls_data: Callable[[], list[dict[str, Any]]],
fake_schedule: FakeSchedule,
test_entity: MockCalendarEntity,
trigger_format: TriggerFormat,
) -> None:
"""Test that a trigger fires for events that overlap."""
event_data1 = test_entity.create_event(
start=datetime.datetime.fromisoformat("2022-04-19 11:00:00+00:00"),
end=datetime.datetime.fromisoformat("2022-04-19 11:30:00+00:00"),
)
event_data2 = test_entity.create_event(
start=datetime.datetime.fromisoformat("2022-04-19 11:15:00+00:00"),
end=datetime.datetime.fromisoformat("2022-04-19 11:45:00+00:00"),
)
async with create_automation(hass, trigger_format, EVENT_START):
await fake_schedule.fire_until(
datetime.datetime.fromisoformat("2022-04-19 11:20:00+00:00")
)
assert calls_data() == [
trigger_format.get_expected_call_data(
CALENDAR_ENTITY_ID, EVENT_START, event_data1
),
trigger_format.get_expected_call_data(
CALENDAR_ENTITY_ID, EVENT_START, event_data2
),
]
async def test_invalid_calendar_id(
hass: HomeAssistant, caplog: pytest.LogCaptureFixture
) -> None:
"""Test creating a trigger with an invalid calendar id."""
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: {
"action": TEST_AUTOMATION_ACTION,
"trigger": {
"platform": calendar.DOMAIN,
"entity_id": "invalid-calendar-id",
},
}
},
)
await hass.async_block_till_done()
assert "Entity ID invalid-calendar-id is an invalid entity ID" in caplog.text
async def test_legacy_entity_type(
hass: HomeAssistant, caplog: pytest.LogCaptureFixture
) -> None:
"""Test creating a trigger with an invalid calendar id."""
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: {
"action": TEST_AUTOMATION_ACTION,
"trigger": {
"platform": calendar.DOMAIN,
"entity_id": "calendar.calendar_3",
},
}
},
)
await hass.async_block_till_done()
assert "is not a calendar entity" in caplog.text
async def test_update_next_event(
hass: HomeAssistant,
calls_data: Callable[[], list[dict[str, Any]]],
fake_schedule: FakeSchedule,
test_entity: MockCalendarEntity,
trigger_format: TriggerFormat,
) -> None:
"""Test detection of a new event after initial trigger is setup."""
event_data1 = test_entity.create_event(
start=datetime.datetime.fromisoformat("2022-04-19 11:00:00+00:00"),
end=datetime.datetime.fromisoformat("2022-04-19 11:15:00+00:00"),
)
async with create_automation(hass, trigger_format, EVENT_START):
# No calls before event start
await fake_schedule.fire_until(
datetime.datetime.fromisoformat("2022-04-19 10:45:00+00:00")
)
assert len(calls_data()) == 0
# Create a new event between now and when the event fires
event_data2 = test_entity.create_event(
start=datetime.datetime.fromisoformat("2022-04-19 10:55:00+00:00"),
end=datetime.datetime.fromisoformat("2022-04-19 11:05:00+00:00"),
)
# Advance past the end of the events
await fake_schedule.fire_until(
datetime.datetime.fromisoformat("2022-04-19 11:30:00+00:00")
)
assert calls_data() == [
trigger_format.get_expected_call_data(
CALENDAR_ENTITY_ID, EVENT_START, event_data2
),
trigger_format.get_expected_call_data(
CALENDAR_ENTITY_ID, EVENT_START, event_data1
),
]
async def test_update_missed(
hass: HomeAssistant,
calls_data: Callable[[], list[dict[str, Any]]],
fake_schedule: FakeSchedule,
test_entity: MockCalendarEntity,
trigger_format: TriggerFormat,
) -> None:
"""Test that new events are missed if they arrive outside the update interval."""
event_data1 = test_entity.create_event(
start=datetime.datetime.fromisoformat("2022-04-19 11:00:00+00:00"),
end=datetime.datetime.fromisoformat("2022-04-19 11:30:00+00:00"),
)
async with create_automation(hass, trigger_format, EVENT_START):
# Events are refreshed at t+TEST_UPDATE_INTERVAL minutes. A new event is
# added, but the next update happens after the event is already over.
await fake_schedule.fire_until(
datetime.datetime.fromisoformat("2022-04-19 10:38:00+00:00")
)
assert len(calls_data()) == 0
test_entity.create_event(
start=datetime.datetime.fromisoformat("2022-04-19 10:40:00+00:00"),
end=datetime.datetime.fromisoformat("2022-04-19 10:55:00+00:00"),
)
# Only the first event is returned
await fake_schedule.fire_until(
datetime.datetime.fromisoformat("2022-04-19 11:05:00+00:00")
)
assert calls_data() == [
trigger_format.get_expected_call_data(
CALENDAR_ENTITY_ID, EVENT_START, event_data1
),
]
@pytest.mark.parametrize(
("create_data", "fire_time", "payload_data"),
[
(
{
"start": datetime.datetime.fromisoformat("2022-04-19 11:00:00+00:00"),
"end": datetime.datetime.fromisoformat("2022-04-19 11:30:00+00:00"),
"summary": "Summary",
},
datetime.datetime.fromisoformat("2022-04-19 11:15:00+00:00"),
{
"summary": "Summary",
"start": "2022-04-19T11:00:00+00:00",
"end": "2022-04-19T11:30:00+00:00",
"all_day": False,
},
),
(
{
"start": datetime.datetime.fromisoformat("2022-04-19 11:00:00+00:00"),
"end": datetime.datetime.fromisoformat("2022-04-19 11:30:00+00:00"),
"summary": "Summary",
"description": "Description",
"location": "Location",
},
datetime.datetime.fromisoformat("2022-04-19 11:15:00+00:00"),
{
"summary": "Summary",
"start": "2022-04-19T11:00:00+00:00",
"end": "2022-04-19T11:30:00+00:00",
"all_day": False,
"description": "Description",
"location": "Location",
},
),
(
{
"summary": "Summary",
"start": datetime.date.fromisoformat("2022-04-20"),
"end": datetime.date.fromisoformat("2022-04-21"),
},
datetime.datetime.fromisoformat("2022-04-20 00:00:01-06:00"),
{
"summary": "Summary",
"start": "2022-04-20",
"end": "2022-04-21",
"all_day": True,
},
),
],
ids=["basic", "more-fields", "all-day"],
)
async def test_event_payload(
hass: HomeAssistant,
calls_data: Callable[[], list[dict[str, Any]]],
fake_schedule: FakeSchedule,
test_entity: MockCalendarEntity,
set_time_zone: None,
trigger_format: TriggerFormat,
create_data,
fire_time,
payload_data,
) -> None:
"""Test the fields in the calendar event payload are set."""
test_entity.create_event(**create_data)
async with create_automation(hass, trigger_format, EVENT_START):
assert len(calls_data()) == 0
await fake_schedule.fire_until(fire_time)
assert calls_data() == [
trigger_format.get_expected_call_data(
CALENDAR_ENTITY_ID, EVENT_START, payload_data
)
]
async def test_trigger_timestamp_window_edge(
hass: HomeAssistant,
calls_data: Callable[[], list[dict[str, Any]]],
fake_schedule: FakeSchedule,
test_entity: MockCalendarEntity,
freezer: FrozenDateTimeFactory,
trigger_format: TriggerFormat,
) -> None:
"""Test that events in the edge of a scan are included."""
freezer.move_to("2022-04-19 11:00:00+00:00")
# Exactly at a TEST_UPDATE_INTERVAL boundary the start time,
# making this excluded from the first window.
event_data = test_entity.create_event(
start=datetime.datetime.fromisoformat("2022-04-19 11:14:00+00:00"),
end=datetime.datetime.fromisoformat("2022-04-19 11:30:00+00:00"),
)
async with create_automation(hass, trigger_format, EVENT_START):
assert len(calls_data()) == 0
await fake_schedule.fire_until(
datetime.datetime.fromisoformat("2022-04-19 11:20:00+00:00")
)
assert calls_data() == [
trigger_format.get_expected_call_data(
CALENDAR_ENTITY_ID, EVENT_START, event_data
)
]
async def test_event_start_trigger_dst(
hass: HomeAssistant,
calls_data: Callable[[], list[dict[str, Any]]],
fake_schedule: FakeSchedule,
test_entity: MockCalendarEntity,
freezer: FrozenDateTimeFactory,
trigger_format: TriggerFormat,
) -> None:
"""Test a calendar event trigger happening at the start of daylight savings time."""
await hass.config.async_set_time_zone("America/Los_Angeles")
tzinfo = zoneinfo.ZoneInfo("America/Los_Angeles")
freezer.move_to("2023-03-12 01:00:00-08:00")
# Before DST transition starts
event1_data = test_entity.create_event(
summary="Event 1",
start=datetime.datetime(2023, 3, 12, 1, 30, tzinfo=tzinfo),
end=datetime.datetime(2023, 3, 12, 1, 45, tzinfo=tzinfo),
)
# During DST transition (Clocks are turned forward at 2am to 3am)
event2_data = test_entity.create_event(
summary="Event 2",
start=datetime.datetime(2023, 3, 12, 2, 30, tzinfo=tzinfo),
end=datetime.datetime(2023, 3, 12, 2, 45, tzinfo=tzinfo),
)
# After DST transition has ended
event3_data = test_entity.create_event(
summary="Event 3",
start=datetime.datetime(2023, 3, 12, 3, 30, tzinfo=tzinfo),
end=datetime.datetime(2023, 3, 12, 3, 45, tzinfo=tzinfo),
)
async with create_automation(hass, trigger_format, EVENT_START):
assert len(calls_data()) == 0
await fake_schedule.fire_until(
datetime.datetime.fromisoformat("2023-03-12 05:00:00-08:00"),
)
assert calls_data() == [
trigger_format.get_expected_call_data(
CALENDAR_ENTITY_ID, EVENT_START, event1_data
),
trigger_format.get_expected_call_data(
CALENDAR_ENTITY_ID, EVENT_START, event2_data
),
trigger_format.get_expected_call_data(
CALENDAR_ENTITY_ID, EVENT_START, event3_data
),
]
async def test_config_entry_reload(
hass: HomeAssistant,
calls_data: Callable[[], list[dict[str, Any]]],
fake_schedule: FakeSchedule,
test_entities: list[MockCalendarEntity],
config_entry: MockConfigEntry,
trigger_format: TriggerFormat,
) -> None:
"""Test the a calendar trigger after a config entry reload.
This sets ups a config entry, sets up an automation for an entity in that
config entry, then reloads the config entry. This reproduces a bug where
the automation kept a reference to the specific entity which would be
invalid after a config entry was reloaded.
"""
async with create_automation(hass, trigger_format, EVENT_START):
assert len(calls_data()) == 0
assert await hass.config_entries.async_reload(config_entry.entry_id)
# Ensure the reloaded entity has events upcoming.
test_entity = test_entities[1]
event_data = test_entity.create_event(
start=datetime.datetime.fromisoformat("2022-04-19 11:00:00+00:00"),
end=datetime.datetime.fromisoformat("2022-04-19 11:30:00+00:00"),
)
await fake_schedule.fire_until(
datetime.datetime.fromisoformat("2022-04-19 11:15:00+00:00"),
)
assert calls_data() == [
trigger_format.get_expected_call_data(
CALENDAR_ENTITY_ID, EVENT_START, event_data
)
]
async def test_config_entry_unload(
hass: HomeAssistant,
calls_data: Callable[[], list[dict[str, Any]]],
fake_schedule: FakeSchedule,
test_entities: list[MockCalendarEntity],
config_entry: MockConfigEntry,
caplog: pytest.LogCaptureFixture,
trigger_format: TriggerFormat,
) -> None:
"""Test an automation that references a calendar entity that is unloaded."""
async with create_automation(hass, trigger_format, EVENT_START):
assert len(calls_data()) == 0
assert await hass.config_entries.async_unload(config_entry.entry_id)
await fake_schedule.fire_until(
datetime.datetime.fromisoformat("2022-04-19 11:15:00+00:00"),
)
assert "Entity does not exist calendar.calendar_2" in caplog.text
@pytest.mark.usefixtures("target_calendars")
@pytest.mark.parametrize(
(
"trigger_target_conf",
"first_start_call_data",
"first_end_call_data",
"second_start_call_data",
"second_end_call_data",
),
[
({}, [], [], [], []),
(
{ATTR_ENTITY_ID: "calendar.calendar_2"},
[],
[],
TARGET_TEST_SECOND_START_CALL_DATA,
TARGET_TEST_SECOND_END_CALL_DATA,
),
(
{ATTR_ENTITY_ID: ["calendar.calendar_1", "calendar.calendar_2"]},
TARGET_TEST_FIRST_START_CALL_DATA,
TARGET_TEST_FIRST_END_CALL_DATA,
TARGET_TEST_SECOND_START_CALL_DATA,
TARGET_TEST_SECOND_END_CALL_DATA,
),
(
{ATTR_AREA_ID: "area_both_calendars"},
TARGET_TEST_FIRST_START_CALL_DATA,
TARGET_TEST_FIRST_END_CALL_DATA,
TARGET_TEST_SECOND_START_CALL_DATA,
TARGET_TEST_SECOND_END_CALL_DATA,
),
(
{ATTR_LABEL_ID: "calendar_1_label"},
TARGET_TEST_FIRST_START_CALL_DATA,
TARGET_TEST_FIRST_END_CALL_DATA,
[],
[],
),
(
{ATTR_DEVICE_ID: "device_calendar_1"},
TARGET_TEST_FIRST_START_CALL_DATA,
TARGET_TEST_FIRST_END_CALL_DATA,
[],
[],
),
(
{ATTR_DEVICE_ID: "device_calendar_2"},
[],
[],
TARGET_TEST_SECOND_START_CALL_DATA,
TARGET_TEST_SECOND_END_CALL_DATA,
),
(
{ATTR_LABEL_ID: "label_on_devices"},
TARGET_TEST_FIRST_START_CALL_DATA,
TARGET_TEST_FIRST_END_CALL_DATA,
TARGET_TEST_SECOND_START_CALL_DATA,
TARGET_TEST_SECOND_END_CALL_DATA,
),
],
)
async def test_trigger_with_targets(
hass: HomeAssistant,
calls_data: Callable[[], list[dict[str, Any]]],
fake_schedule: FakeSchedule,
test_entities: list[MockCalendarEntity],
trigger_target_conf: dict[str, Any],
first_start_call_data: list[dict[str, Any]],
first_end_call_data: list[dict[str, Any]],
second_start_call_data: list[dict[str, Any]],
second_end_call_data: list[dict[str, Any]],
) -> None:
"""Test that triggers fire for multiple calendar entities with target selector."""
calendar_1 = test_entities[0]
calendar_2 = test_entities[1]
calendar_1.create_event(
start=datetime.datetime.fromisoformat("2022-04-19 11:00:00+00:00"),
end=datetime.datetime.fromisoformat("2022-04-19 11:30:00+00:00"),
summary="Event on Calendar 1",
)
calendar_2.create_event(
start=datetime.datetime.fromisoformat("2022-04-19 11:15:00+00:00"),
end=datetime.datetime.fromisoformat("2022-04-19 11:45:00+00:00"),
summary="Event on Calendar 2",
)
trigger_start = {
CONF_PLATFORM: "calendar.event_started",
CONF_TARGET: {**trigger_target_conf},
}
trigger_end = {
CONF_PLATFORM: "calendar.event_ended",
CONF_TARGET: {**trigger_target_conf},
}
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: [
{
"alias": "start_trigger",
"trigger": trigger_start,
"action": TEST_AUTOMATION_ACTION,
"mode": "queued",
},
{
"alias": "end_trigger",
"trigger": trigger_end,
"action": TEST_AUTOMATION_ACTION,
"mode": "queued",
},
]
},
)
await hass.async_block_till_done()
assert len(calls_data()) == 0
# Advance past first event start
await fake_schedule.fire_until(
datetime.datetime.fromisoformat("2022-04-19 11:10:00+00:00")
)
assert calls_data() == first_start_call_data
# Advance past second event start
await fake_schedule.fire_until(
datetime.datetime.fromisoformat("2022-04-19 11:20:00+00:00")
)
assert calls_data() == first_start_call_data + second_start_call_data
# Advance past first event end
await fake_schedule.fire_until(
datetime.datetime.fromisoformat("2022-04-19 11:40:00+00:00")
)
assert (
calls_data()
== first_start_call_data + second_start_call_data + first_end_call_data
)
# Advance past second event end
await fake_schedule.fire_until(
datetime.datetime.fromisoformat("2022-04-19 11:50:00+00:00")
)
assert (
calls_data()
== first_start_call_data
+ second_start_call_data
+ first_end_call_data
+ second_end_call_data
)
# Disable automations to cleanup lingering timers
await hass.services.async_call(
automation.DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: "automation.start_trigger"},
blocking=True,
)
await hass.services.async_call(
automation.DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: "automation.end_trigger"},
blocking=True,
)
Reference in New Issue View Git Blame Copy Permalink