mirror of
https://github.com/home-assistant/core.git
synced 2026-08-24 10:13:52 -05:00
Add websocket subscription support for calendar events (#156340)
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
@@ -17,6 +17,7 @@ import voluptuous as vol
|
||||
|
||||
from homeassistant.components import frontend, http, websocket_api
|
||||
from homeassistant.components.websocket_api import (
|
||||
ERR_INVALID_FORMAT,
|
||||
ERR_NOT_FOUND,
|
||||
ERR_NOT_SUPPORTED,
|
||||
ActiveConnection,
|
||||
@@ -33,6 +34,7 @@ from homeassistant.core import (
|
||||
)
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
from homeassistant.helpers import config_validation as cv, entity_registry as er
|
||||
from homeassistant.helpers.debounce import Debouncer
|
||||
from homeassistant.helpers.entity import Entity, EntityDescription
|
||||
from homeassistant.helpers.entity_component import EntityComponent
|
||||
from homeassistant.helpers.event import async_track_point_in_time
|
||||
@@ -76,6 +78,7 @@ ENTITY_ID_FORMAT = DOMAIN + ".{}"
|
||||
PLATFORM_SCHEMA = cv.PLATFORM_SCHEMA
|
||||
PLATFORM_SCHEMA_BASE = cv.PLATFORM_SCHEMA_BASE
|
||||
SCAN_INTERVAL = datetime.timedelta(seconds=60)
|
||||
EVENT_LISTENER_DEBOUNCE_COOLDOWN = 1.0 # seconds
|
||||
|
||||
# Don't support rrules more often than daily
|
||||
VALID_FREQS = {"DAILY", "WEEKLY", "MONTHLY", "YEARLY"}
|
||||
@@ -320,6 +323,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
|
||||
websocket_api.async_register_command(hass, handle_calendar_event_create)
|
||||
websocket_api.async_register_command(hass, handle_calendar_event_delete)
|
||||
websocket_api.async_register_command(hass, handle_calendar_event_update)
|
||||
websocket_api.async_register_command(hass, handle_calendar_event_subscribe)
|
||||
|
||||
component.async_register_entity_service(
|
||||
CREATE_EVENT_SERVICE,
|
||||
@@ -517,6 +521,17 @@ class CalendarEntity(Entity):
|
||||
_entity_component_unrecorded_attributes = frozenset({"description"})
|
||||
|
||||
_alarm_unsubs: list[CALLBACK_TYPE] | None = None
|
||||
_event_listeners: (
|
||||
list[
|
||||
tuple[
|
||||
datetime.datetime,
|
||||
datetime.datetime,
|
||||
Callable[[list[JsonValueType] | None], None],
|
||||
]
|
||||
]
|
||||
| None
|
||||
) = None
|
||||
_event_listener_debouncer: Debouncer[None] | None = None
|
||||
|
||||
_attr_initial_color: str | None
|
||||
|
||||
@@ -585,6 +600,10 @@ class CalendarEntity(Entity):
|
||||
the current or upcoming event.
|
||||
"""
|
||||
super()._async_write_ha_state()
|
||||
|
||||
# Notify websocket subscribers of event changes (debounced)
|
||||
if self._event_listeners and self._event_listener_debouncer:
|
||||
self._event_listener_debouncer.async_schedule_call()
|
||||
if self._alarm_unsubs is None:
|
||||
self._alarm_unsubs = []
|
||||
_LOGGER.debug(
|
||||
@@ -625,6 +644,13 @@ class CalendarEntity(Entity):
|
||||
event.end_datetime_local,
|
||||
)
|
||||
|
||||
@callback
|
||||
def _async_cancel_event_listener_debouncer(self) -> None:
|
||||
"""Cancel and clear the event listener debouncer."""
|
||||
if self._event_listener_debouncer:
|
||||
self._event_listener_debouncer.async_cancel()
|
||||
self._event_listener_debouncer = None
|
||||
|
||||
async def async_will_remove_from_hass(self) -> None:
|
||||
"""Run when entity will be removed from hass.
|
||||
|
||||
@@ -633,6 +659,90 @@ class CalendarEntity(Entity):
|
||||
for unsub in self._alarm_unsubs or ():
|
||||
unsub()
|
||||
self._alarm_unsubs = None
|
||||
self._async_cancel_event_listener_debouncer()
|
||||
|
||||
@final
|
||||
@callback
|
||||
def async_subscribe_events(
|
||||
self,
|
||||
start_date: datetime.datetime,
|
||||
end_date: datetime.datetime,
|
||||
event_listener: Callable[[list[JsonValueType] | None], None],
|
||||
) -> CALLBACK_TYPE:
|
||||
"""Subscribe to calendar event updates.
|
||||
|
||||
Called by websocket API.
|
||||
"""
|
||||
if self._event_listeners is None:
|
||||
self._event_listeners = []
|
||||
|
||||
if self._event_listener_debouncer is None:
|
||||
self._event_listener_debouncer = Debouncer(
|
||||
self.hass,
|
||||
_LOGGER,
|
||||
cooldown=EVENT_LISTENER_DEBOUNCE_COOLDOWN,
|
||||
immediate=True,
|
||||
function=self.async_update_event_listeners,
|
||||
)
|
||||
|
||||
listener_data = (start_date, end_date, event_listener)
|
||||
self._event_listeners.append(listener_data)
|
||||
|
||||
@callback
|
||||
def unsubscribe() -> None:
|
||||
if self._event_listeners:
|
||||
self._event_listeners.remove(listener_data)
|
||||
if not self._event_listeners:
|
||||
self._async_cancel_event_listener_debouncer()
|
||||
|
||||
return unsubscribe
|
||||
|
||||
@final
|
||||
@callback
|
||||
def async_update_event_listeners(self) -> None:
|
||||
"""Push updated calendar events to all listeners."""
|
||||
if not self._event_listeners:
|
||||
return
|
||||
|
||||
for start_date, end_date, listener in self._event_listeners:
|
||||
self.async_update_single_event_listener(start_date, end_date, listener)
|
||||
|
||||
@final
|
||||
@callback
|
||||
def async_update_single_event_listener(
|
||||
self,
|
||||
start_date: datetime.datetime,
|
||||
end_date: datetime.datetime,
|
||||
listener: Callable[[list[JsonValueType] | None], None],
|
||||
) -> None:
|
||||
"""Schedule an event fetch and push to a single listener."""
|
||||
self.hass.async_create_task(
|
||||
self._async_update_listener(start_date, end_date, listener)
|
||||
)
|
||||
|
||||
async def _async_update_listener(
|
||||
self,
|
||||
start_date: datetime.datetime,
|
||||
end_date: datetime.datetime,
|
||||
listener: Callable[[list[JsonValueType] | None], None],
|
||||
) -> None:
|
||||
"""Fetch events and push to a single listener."""
|
||||
try:
|
||||
events = await self.async_get_events(self.hass, start_date, end_date)
|
||||
except HomeAssistantError as err:
|
||||
_LOGGER.debug(
|
||||
"Error fetching calendar events for %s: %s",
|
||||
self.entity_id,
|
||||
err,
|
||||
)
|
||||
listener(None)
|
||||
return
|
||||
|
||||
event_list: list[JsonValueType] = [
|
||||
dataclasses.asdict(event, dict_factory=_list_events_dict_factory)
|
||||
for event in events
|
||||
]
|
||||
listener(event_list)
|
||||
|
||||
async def async_get_events(
|
||||
self,
|
||||
@@ -867,6 +977,65 @@ async def handle_calendar_event_update(
|
||||
connection.send_result(msg["id"])
|
||||
|
||||
|
||||
@websocket_api.websocket_command(
|
||||
{
|
||||
vol.Required("type"): "calendar/event/subscribe",
|
||||
vol.Required("entity_id"): cv.entity_domain(DOMAIN),
|
||||
vol.Required("start"): cv.datetime,
|
||||
vol.Required("end"): cv.datetime,
|
||||
}
|
||||
)
|
||||
@websocket_api.async_response
|
||||
async def handle_calendar_event_subscribe(
|
||||
hass: HomeAssistant, connection: ActiveConnection, msg: dict[str, Any]
|
||||
) -> None:
|
||||
"""Subscribe to calendar event updates."""
|
||||
entity_id: str = msg["entity_id"]
|
||||
|
||||
if not (entity := hass.data[DATA_COMPONENT].get_entity(entity_id)):
|
||||
connection.send_error(
|
||||
msg["id"],
|
||||
ERR_NOT_FOUND,
|
||||
f"Calendar entity not found: {entity_id}",
|
||||
)
|
||||
return
|
||||
|
||||
start_date = dt_util.as_local(msg["start"])
|
||||
end_date = dt_util.as_local(msg["end"])
|
||||
|
||||
if start_date >= end_date:
|
||||
connection.send_error(
|
||||
msg["id"],
|
||||
ERR_INVALID_FORMAT,
|
||||
"Start must be before end",
|
||||
)
|
||||
return
|
||||
|
||||
subscription_id = msg["id"]
|
||||
|
||||
@callback
|
||||
def event_listener(events: list[JsonValueType] | None) -> None:
|
||||
"""Push updated calendar events to websocket."""
|
||||
if subscription_id not in connection.subscriptions:
|
||||
return
|
||||
connection.send_message(
|
||||
websocket_api.event_message(
|
||||
subscription_id,
|
||||
{
|
||||
"events": events,
|
||||
},
|
||||
)
|
||||
)
|
||||
|
||||
connection.subscriptions[subscription_id] = entity.async_subscribe_events(
|
||||
start_date, end_date, event_listener
|
||||
)
|
||||
connection.send_result(subscription_id)
|
||||
|
||||
# Push initial events only to the new subscriber
|
||||
entity.async_update_single_event_listener(start_date, end_date, event_listener)
|
||||
|
||||
|
||||
def _validate_timespan(
|
||||
values: dict[str, Any],
|
||||
) -> tuple[datetime.datetime | datetime.date, datetime.datetime | datetime.date]:
|
||||
|
||||
@@ -28,6 +28,7 @@ from homeassistant.util import dt as dt_util
|
||||
|
||||
from .conftest import MockCalendarEntity, MockConfigEntry
|
||||
|
||||
from tests.common import async_fire_time_changed
|
||||
from tests.typing import ClientSessionGenerator, WebSocketGenerator
|
||||
|
||||
|
||||
@@ -715,3 +716,236 @@ async def test_calendar_initial_color_precedence(
|
||||
|
||||
entity = TestCalendarEntity(description_color, attr_color)
|
||||
assert entity.initial_color == expected_color
|
||||
|
||||
|
||||
async def test_websocket_handle_subscribe_calendar_events(
|
||||
hass: HomeAssistant,
|
||||
hass_ws_client: WebSocketGenerator,
|
||||
test_entities: list[MockCalendarEntity],
|
||||
) -> None:
|
||||
"""Test subscribing to calendar event updates via websocket."""
|
||||
client = await hass_ws_client(hass)
|
||||
|
||||
start = dt_util.now()
|
||||
end = start + timedelta(days=1)
|
||||
|
||||
await client.send_json_auto_id(
|
||||
{
|
||||
"type": "calendar/event/subscribe",
|
||||
"entity_id": "calendar.calendar_1",
|
||||
"start": start.isoformat(),
|
||||
"end": end.isoformat(),
|
||||
}
|
||||
)
|
||||
msg = await client.receive_json()
|
||||
assert msg["success"]
|
||||
subscription_id = msg["id"]
|
||||
|
||||
# Should receive initial event list
|
||||
msg = await client.receive_json()
|
||||
assert msg["id"] == subscription_id
|
||||
assert msg["type"] == "event"
|
||||
assert "events" in msg["event"]
|
||||
events = msg["event"]["events"]
|
||||
assert len(events) == 1
|
||||
assert events[0]["summary"] == "Future Event"
|
||||
|
||||
|
||||
async def test_websocket_subscribe_updates_on_state_change(
|
||||
hass: HomeAssistant,
|
||||
hass_ws_client: WebSocketGenerator,
|
||||
test_entities: list[MockCalendarEntity],
|
||||
) -> None:
|
||||
"""Test that subscribers receive updates when calendar state changes."""
|
||||
client = await hass_ws_client(hass)
|
||||
|
||||
start = dt_util.now()
|
||||
end = start + timedelta(days=1)
|
||||
|
||||
await client.send_json_auto_id(
|
||||
{
|
||||
"type": "calendar/event/subscribe",
|
||||
"entity_id": "calendar.calendar_1",
|
||||
"start": start.isoformat(),
|
||||
"end": end.isoformat(),
|
||||
}
|
||||
)
|
||||
msg = await client.receive_json()
|
||||
assert msg["success"]
|
||||
subscription_id = msg["id"]
|
||||
|
||||
# Receive initial event list
|
||||
msg = await client.receive_json()
|
||||
assert msg["id"] == subscription_id
|
||||
|
||||
# Add a new event and trigger state update
|
||||
entity = test_entities[0]
|
||||
entity.create_event(
|
||||
start=start + timedelta(hours=2),
|
||||
end=start + timedelta(hours=3),
|
||||
summary="New Event",
|
||||
)
|
||||
entity.async_write_ha_state()
|
||||
await hass.async_block_till_done()
|
||||
|
||||
# Should receive updated event list
|
||||
msg = await client.receive_json()
|
||||
assert msg["id"] == subscription_id
|
||||
assert msg["type"] == "event"
|
||||
events = msg["event"]["events"]
|
||||
assert len(events) == 2
|
||||
summaries = {event["summary"] for event in events}
|
||||
assert "Future Event" in summaries
|
||||
assert "New Event" in summaries
|
||||
|
||||
|
||||
async def test_websocket_subscribe_entity_not_found(
|
||||
hass: HomeAssistant,
|
||||
hass_ws_client: WebSocketGenerator,
|
||||
) -> None:
|
||||
"""Test subscribing to a non-existent calendar entity."""
|
||||
client = await hass_ws_client(hass)
|
||||
|
||||
start = dt_util.now()
|
||||
end = start + timedelta(days=1)
|
||||
|
||||
await client.send_json_auto_id(
|
||||
{
|
||||
"type": "calendar/event/subscribe",
|
||||
"entity_id": "calendar.nonexistent",
|
||||
"start": start.isoformat(),
|
||||
"end": end.isoformat(),
|
||||
}
|
||||
)
|
||||
msg = await client.receive_json()
|
||||
assert not msg["success"]
|
||||
assert msg["error"]["code"] == "not_found"
|
||||
assert "Calendar entity not found" in msg["error"]["message"]
|
||||
|
||||
|
||||
async def test_websocket_subscribe_event_fetch_error(
|
||||
hass: HomeAssistant,
|
||||
hass_ws_client: WebSocketGenerator,
|
||||
test_entities: list[MockCalendarEntity],
|
||||
) -> None:
|
||||
"""Test subscription handles event fetch errors gracefully."""
|
||||
client = await hass_ws_client(hass)
|
||||
|
||||
start = dt_util.now()
|
||||
end = start + timedelta(days=1)
|
||||
|
||||
# Set up entity to fail on async_get_events
|
||||
test_entities[0].async_get_events.side_effect = HomeAssistantError("API Error")
|
||||
|
||||
await client.send_json_auto_id(
|
||||
{
|
||||
"type": "calendar/event/subscribe",
|
||||
"entity_id": "calendar.calendar_1",
|
||||
"start": start.isoformat(),
|
||||
"end": end.isoformat(),
|
||||
}
|
||||
)
|
||||
msg = await client.receive_json()
|
||||
assert msg["success"]
|
||||
subscription_id = msg["id"]
|
||||
|
||||
# Should receive None for events due to error
|
||||
msg = await client.receive_json()
|
||||
assert msg["id"] == subscription_id
|
||||
assert msg["type"] == "event"
|
||||
assert msg["event"]["events"] is None
|
||||
|
||||
|
||||
async def test_websocket_subscribe_invalid_timespan(
|
||||
hass: HomeAssistant,
|
||||
hass_ws_client: WebSocketGenerator,
|
||||
test_entities: list[MockCalendarEntity],
|
||||
) -> None:
|
||||
"""Test subscribing with start after end returns an error."""
|
||||
client = await hass_ws_client(hass)
|
||||
|
||||
now = dt_util.now()
|
||||
start = now + timedelta(days=1)
|
||||
end = now
|
||||
|
||||
await client.send_json_auto_id(
|
||||
{
|
||||
"type": "calendar/event/subscribe",
|
||||
"entity_id": "calendar.calendar_1",
|
||||
"start": start.isoformat(),
|
||||
"end": end.isoformat(),
|
||||
}
|
||||
)
|
||||
msg = await client.receive_json()
|
||||
assert not msg["success"]
|
||||
assert msg["error"]["code"] == "invalid_format"
|
||||
assert "Start must be before end" in msg["error"]["message"]
|
||||
|
||||
|
||||
async def test_websocket_subscribe_debounces_rapid_updates(
|
||||
hass: HomeAssistant,
|
||||
hass_ws_client: WebSocketGenerator,
|
||||
test_entities: list[MockCalendarEntity],
|
||||
) -> None:
|
||||
"""Test that rapid state writes are debounced for event listeners."""
|
||||
client = await hass_ws_client(hass)
|
||||
|
||||
start = dt_util.now()
|
||||
end = start + timedelta(days=1)
|
||||
|
||||
await client.send_json_auto_id(
|
||||
{
|
||||
"type": "calendar/event/subscribe",
|
||||
"entity_id": "calendar.calendar_1",
|
||||
"start": start.isoformat(),
|
||||
"end": end.isoformat(),
|
||||
}
|
||||
)
|
||||
msg = await client.receive_json()
|
||||
assert msg["success"]
|
||||
subscription_id = msg["id"]
|
||||
|
||||
# Receive initial event list
|
||||
msg = await client.receive_json()
|
||||
assert msg["id"] == subscription_id
|
||||
|
||||
entity = test_entities[0]
|
||||
entity.async_get_events.reset_mock()
|
||||
|
||||
# Rapidly write state multiple times
|
||||
for i in range(5):
|
||||
entity.create_event(
|
||||
start=start + timedelta(hours=i + 2),
|
||||
end=start + timedelta(hours=i + 3),
|
||||
summary=f"Rapid Event {i}",
|
||||
)
|
||||
entity.async_write_ha_state()
|
||||
|
||||
await hass.async_block_till_done()
|
||||
|
||||
# The debouncer with immediate=True fires the first call immediately
|
||||
# and coalesces the rest into one call after the cooldown.
|
||||
# Without debouncing this would be 5 calls.
|
||||
assert entity.async_get_events.call_count == 1
|
||||
|
||||
# Advance time past the debounce cooldown to fire the trailing call
|
||||
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=2))
|
||||
await hass.async_block_till_done()
|
||||
|
||||
# Should be exactly 2 total: immediate + one coalesced trailing call
|
||||
assert entity.async_get_events.call_count == 2
|
||||
|
||||
# Drain messages: immediate update + trailing debounced update
|
||||
messages: list[dict] = []
|
||||
for _ in range(10):
|
||||
msg = await client.receive_json()
|
||||
assert msg["id"] == subscription_id
|
||||
assert msg["type"] == "event"
|
||||
messages.append(msg)
|
||||
if len(msg["event"]["events"]) == 6: # 1 original + 5 rapid
|
||||
break
|
||||
else:
|
||||
pytest.fail("Did not receive expected calendar event list with 6 events")
|
||||
|
||||
# The final message has all events
|
||||
assert len(messages[-1]["event"]["events"]) == 6
|
||||
|
||||
Reference in New Issue
Block a user