github
/
ha-core
mirror of https://github.com/home-assistant/core
1
Fork 0
Code Releases Activity

Local calendar integration (#79601)

This commit is contained in:
Allen Porter 2022-11-30 12:20:21 -08:00 committed by GitHub
parent 8acc114cd9
commit 532ab12a48
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
21 changed files with 1152 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CODEOWNERS
View File

@ -639,6 +639,8 @@ build.json @home-assistant/supervisor
/tests/components/litterrobot/ @natekspencer @tkdrob
/homeassistant/components/livisi/ @StefanIacobLivisi
/tests/components/livisi/ @StefanIacobLivisi
/homeassistant/components/local_calendar/ @allenporter
/tests/components/local_calendar/ @allenporter
/homeassistant/components/local_ip/ @issacg
/tests/components/local_ip/ @issacg
/homeassistant/components/lock/ @home-assistant/core

154
homeassistant/components/calendar/__init__.py
View File

@ -10,12 +10,17 @@ import re
from typing import Any, cast, final
from aiohttp import web
from dateutil.rrule import rrulestr
import voluptuous as vol
from homeassistant.components import frontend, http
from homeassistant.components import frontend, http, websocket_api
from homeassistant.components.websocket_api import ERR_NOT_FOUND, ERR_NOT_SUPPORTED
from homeassistant.components.websocket_api.connection import ActiveConnection
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import STATE_OFF, STATE_ON
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.config_validation import ( # noqa: F401
PLATFORM_SCHEMA,
PLATFORM_SCHEMA_BASE,
@ -27,12 +32,29 @@ from homeassistant.helpers.template import DATE_STR_FORMAT
from homeassistant.helpers.typing import ConfigType
from homeassistant.util import dt
from .const import (
CONF_EVENT,
EVENT_DESCRIPTION,
EVENT_END,
EVENT_RECURRENCE_ID,
EVENT_RECURRENCE_RANGE,
EVENT_RRULE,
EVENT_START,
EVENT_SUMMARY,
EVENT_UID,
CalendarEntityFeature,
)
_LOGGER = logging.getLogger(__name__)
DOMAIN = "calendar"
ENTITY_ID_FORMAT = DOMAIN + ".{}"
SCAN_INTERVAL = datetime.timedelta(seconds=60)
# Don't support rrules more often than daily
VALID_FREQS = {"DAILY", "WEEKLY", "MONTHLY", "YEARLY"}
# mypy: disallow-any-generics
@ -49,6 +71,9 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
hass, "calendar", "calendar", "hass:calendar"
)
websocket_api.async_register_command(hass, handle_calendar_event_create)
websocket_api.async_register_command(hass, handle_calendar_event_delete)
await component.async_setup(config)
return True
@ -88,6 +113,10 @@ class CalendarEvent:
description: str | None = None
location: str | None = None
uid: str | None = None
recurrence_id: str | None = None
rrule: str | None = None
@property
def start_datetime_local(self) -> datetime.datetime:
"""Return event start time as a local datetime."""
@ -183,6 +212,30 @@ def is_offset_reached(
return start + offset_time <= dt.now(start.tzinfo)
def _validate_rrule(value: Any) -> str:
"""Validate a recurrence rule string."""
if value is None:
raise vol.Invalid("rrule value is None")
if not isinstance(value, str):
raise vol.Invalid("rrule value expected a string")
try:
rrulestr(value)
except ValueError as err:
raise vol.Invalid(f"Invalid rrule: {str(err)}") from err
# Example format: FREQ=DAILY;UNTIL=...
rule_parts = dict(s.split("=", 1) for s in value.split(";"))
if not (freq := rule_parts.get("FREQ")):
raise vol.Invalid("rrule did not contain FREQ")
if freq not in VALID_FREQS:
raise vol.Invalid(f"Invalid frequency for rule: {value}")
return str(value)
class CalendarEntity(Entity):
"""Base class for calendar event entities."""
@ -230,6 +283,19 @@ class CalendarEntity(Entity):
"""Return calendar events within a datetime range."""
raise NotImplementedError()
async def async_create_event(self, **kwargs: Any) -> None:
"""Add a new event to calendar."""
raise NotImplementedError()
async def async_delete_event(
self,
uid: str,
recurrence_id: str | None = None,
recurrence_range: str | None = None,
) -> None:
"""Delete an event on the calendar."""
raise NotImplementedError()
class CalendarEventView(http.HomeAssistantView):
"""View to retrieve calendar content."""
@ -297,3 +363,89 @@ class CalendarListView(http.HomeAssistantView):
calendar_list.append({"name": state.name, "entity_id": entity.entity_id})
return self.json(sorted(calendar_list, key=lambda x: cast(str, x["name"])))
@websocket_api.websocket_command(
{
vol.Required("type"): "calendar/event/create",
vol.Required("entity_id"): cv.entity_id,
vol.Required(CONF_EVENT): {
vol.Required(EVENT_START): vol.Any(cv.date, cv.datetime),
vol.Required(EVENT_END): vol.Any(cv.date, cv.datetime),
vol.Required(EVENT_SUMMARY): cv.string,
vol.Optional(EVENT_DESCRIPTION): cv.string,
vol.Optional(EVENT_RRULE): _validate_rrule,
},
}
)
@websocket_api.async_response
async def handle_calendar_event_create(
hass: HomeAssistant, connection: ActiveConnection, msg: dict[str, Any]
) -> None:
"""Handle creation of a calendar event."""
component: EntityComponent[CalendarEntity] = hass.data[DOMAIN]
if not (entity := component.get_entity(msg["entity_id"])):
connection.send_error(msg["id"], ERR_NOT_FOUND, "Entity not found")
return
if (
not entity.supported_features
or not entity.supported_features & CalendarEntityFeature.CREATE_EVENT
):
connection.send_message(
websocket_api.error_message(
msg["id"], ERR_NOT_SUPPORTED, "Calendar does not support event creation"
)
)
return
try:
await entity.async_create_event(**msg[CONF_EVENT])
except HomeAssistantError as ex:
connection.send_error(msg["id"], "failed", str(ex))
else:
connection.send_result(msg["id"])
@websocket_api.websocket_command(
{
vol.Required("type"): "calendar/event/delete",
vol.Required("entity_id"): cv.entity_id,
vol.Required(EVENT_UID): cv.string,
vol.Optional(EVENT_RECURRENCE_ID): cv.string,
vol.Optional(EVENT_RECURRENCE_RANGE): cv.string,
}
)
@websocket_api.async_response
async def handle_calendar_event_delete(
hass: HomeAssistant, connection: ActiveConnection, msg: dict[str, Any]
) -> None:
"""Handle delete of a calendar event."""
component: EntityComponent[CalendarEntity] = hass.data[DOMAIN]
if not (entity := component.get_entity(msg["entity_id"])):
connection.send_error(msg["id"], ERR_NOT_FOUND, "Entity not found")
return
if (
not entity.supported_features
or not entity.supported_features & CalendarEntityFeature.DELETE_EVENT
):
connection.send_message(
websocket_api.error_message(
msg["id"], ERR_NOT_SUPPORTED, "Calendar does not support event deletion"
)
)
return
try:
await entity.async_delete_event(
msg[EVENT_UID],
recurrence_id=msg.get(EVENT_RECURRENCE_ID),
recurrence_range=msg.get(EVENT_RECURRENCE_RANGE),
)
except (HomeAssistantError, ValueError) as ex:
_LOGGER.error("Error handling Calendar Event call: %s", ex)
connection.send_error(msg["id"], "failed", str(ex))
else:
connection.send_result(msg["id"])

24
homeassistant/components/calendar/const.py Normal file
View File

@ -0,0 +1,24 @@
"""Constants for calendar components."""
from enum import IntEnum
CONF_EVENT = "event"
class CalendarEntityFeature(IntEnum):
"""Supported features of the calendar entity."""
CREATE_EVENT = 1
DELETE_EVENT = 2
# rfc5545 fields
EVENT_UID = "uid"
EVENT_START = "dtstart"
EVENT_END = "dtend"
EVENT_SUMMARY = "summary"
EVENT_DESCRIPTION = "description"
EVENT_LOCATION = "location"
EVENT_RECURRENCE_ID = "recurrence_id"
EVENT_RECURRENCE_RANGE = "recurrence_range"
EVENT_RRULE = "rrule"

41
homeassistant/components/local_calendar/__init__.py Normal file
View File

@ -0,0 +1,41 @@
"""The Local Calendar integration."""
from __future__ import annotations
import logging
from pathlib import Path
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.util import slugify
from .const import CONF_CALENDAR_NAME, DOMAIN
from .store import LocalCalendarStore
_LOGGER = logging.getLogger(__name__)
PLATFORMS: list[Platform] = [Platform.CALENDAR]
STORAGE_PATH = ".storage/local_calendar.{key}.ics"
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Local Calendar from a config entry."""
hass.data.setdefault(DOMAIN, {})
key = slugify(entry.data[CONF_CALENDAR_NAME])
path = Path(hass.config.path(STORAGE_PATH.format(key=key)))
hass.data[DOMAIN][entry.entry_id] = LocalCalendarStore(hass, path)
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."""
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok

149
homeassistant/components/local_calendar/calendar.py Normal file
View File

@ -0,0 +1,149 @@
"""Calendar platform for a Local Calendar."""
from __future__ import annotations
from datetime import datetime
import logging
from typing import Any
from ical.calendar import Calendar
from ical.calendar_stream import IcsCalendarStream
from ical.event import Event
from ical.store import EventStore
from ical.types import Range, Recur
from homeassistant.components.calendar import (
EVENT_DESCRIPTION,
EVENT_END,
EVENT_RRULE,
EVENT_START,
EVENT_SUMMARY,
CalendarEntity,
CalendarEntityFeature,
CalendarEvent,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.util import dt as dt_util
from .const import CONF_CALENDAR_NAME, DOMAIN
from .store import LocalCalendarStore
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the local calendar platform."""
store = hass.data[DOMAIN][config_entry.entry_id]
ics = await store.async_load()
calendar = IcsCalendarStream.calendar_from_ics(ics)
name = config_entry.data[CONF_CALENDAR_NAME]
entity = LocalCalendarEntity(store, calendar, name, unique_id=config_entry.entry_id)
async_add_entities([entity], True)
class LocalCalendarEntity(CalendarEntity):
"""A calendar entity backed by a local iCalendar file."""
_attr_has_entity_name = True
_attr_supported_features = (
CalendarEntityFeature.CREATE_EVENT | CalendarEntityFeature.DELETE_EVENT
)
def __init__(
self,
store: LocalCalendarStore,
calendar: Calendar,
name: str,
unique_id: str,
) -> None:
"""Initialize LocalCalendarEntity."""
self._store = store
self._calendar = calendar
self._event: CalendarEvent | None = None
self._attr_name = name.capitalize()
self._attr_unique_id = unique_id
@property
def event(self) -> CalendarEvent | None:
"""Return the next upcoming event."""
return self._event
async def async_get_events(
self, hass: HomeAssistant, start_date: datetime, end_date: datetime
) -> list[CalendarEvent]:
"""Get all events in a specific time frame."""
events = self._calendar.timeline_tz(dt_util.DEFAULT_TIME_ZONE).overlapping(
dt_util.as_local(start_date),
dt_util.as_local(end_date),
)
return [_get_calendar_event(event) for event in events]
async def async_update(self) -> None:
"""Update entity state with the next upcoming event."""
events = self._calendar.timeline_tz(dt_util.DEFAULT_TIME_ZONE).active_after(
dt_util.now()
)
if event := next(events, None):
self._event = _get_calendar_event(event)
else:
self._event = None
async def _async_store(self) -> None:
"""Persist the calendar to disk."""
content = IcsCalendarStream.calendar_to_ics(self._calendar)
await self._store.async_store(content)
async def async_create_event(self, **kwargs: Any) -> None:
"""Add a new event to calendar."""
event = Event.parse_obj(
{
EVENT_SUMMARY: kwargs[EVENT_SUMMARY],
EVENT_START: kwargs[EVENT_START],
EVENT_END: kwargs[EVENT_END],
EVENT_DESCRIPTION: kwargs.get(EVENT_DESCRIPTION),
}
)
if rrule := kwargs.get(EVENT_RRULE):
event.rrule = Recur.from_rrule(rrule)
EventStore(self._calendar).add(event)
await self._async_store()
await self.async_update_ha_state(force_refresh=True)
async def async_delete_event(
self,
uid: str,
recurrence_id: str | None = None,
recurrence_range: str | None = None,
) -> None:
"""Delete an event on the calendar."""
range_value: Range = Range.NONE
if recurrence_range == Range.THIS_AND_FUTURE:
range_value = Range.THIS_AND_FUTURE
EventStore(self._calendar).delete(
uid,
recurrence_id=recurrence_id,
recurrence_range=range_value,
)
await self._async_store()
await self.async_update_ha_state(force_refresh=True)
def _get_calendar_event(event: Event) -> CalendarEvent:
"""Return a CalendarEvent from an API event."""
return CalendarEvent(
summary=event.summary,
start=event.start,
end=event.end,
description=event.description,
uid=event.uid,
rrule=event.rrule.as_rrule_str() if event.rrule else None,
recurrence_id=event.recurrence_id,
)

36
homeassistant/components/local_calendar/config_flow.py Normal file
View File

@ -0,0 +1,36 @@
"""Config flow for Local Calendar integration."""
from __future__ import annotations
from typing import Any
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.data_entry_flow import FlowResult
from .const import CONF_CALENDAR_NAME, DOMAIN
STEP_USER_DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_CALENDAR_NAME): str,
}
)
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Local Calendar."""
VERSION = 1
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the initial step."""
if user_input is None:
return self.async_show_form(
step_id="user", data_schema=STEP_USER_DATA_SCHEMA
)
return self.async_create_entry(
title=user_input[CONF_CALENDAR_NAME], data=user_input
)

5
homeassistant/components/local_calendar/const.py Normal file
View File

@ -0,0 +1,5 @@
"""Constants for the Local Calendar integration."""
DOMAIN = "local_calendar"
CONF_CALENDAR_NAME = "calendar_name"

10
homeassistant/components/local_calendar/manifest.json Normal file
View File

@ -0,0 +1,10 @@
{
"domain": "local_calendar",
"name": "Local Calendar",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/local_calendar",
"requirements": ["ical==4.1.1"],
"codeowners": ["@allenporter"],
"iot_class": "local_polling",
"loggers": ["ical"]
}

38
homeassistant/components/local_calendar/store.py Normal file
View File

@ -0,0 +1,38 @@
"""Local storage for the Local Calendar integration."""
import asyncio
from pathlib import Path
from homeassistant.core import HomeAssistant
STORAGE_PATH = ".storage/{key}.ics"
class LocalCalendarStore:
"""Local calendar storage."""
def __init__(self, hass: HomeAssistant, path: Path) -> None:
"""Initialize LocalCalendarStore."""
self._hass = hass
self._path = path
self._lock = asyncio.Lock()
async def async_load(self) -> str:
"""Load the calendar from disk."""
async with self._lock:
return await self._hass.async_add_executor_job(self._load)
def _load(self) -> str:
"""Load the calendar from disk."""
if not self._path.exists():
return ""
return self._path.read_text()
async def async_store(self, ics_content: str) -> None:
"""Persist the calendar to storage."""
async with self._lock:
await self._hass.async_add_executor_job(self._store, ics_content)
def _store(self, ics_content: str) -> None:
"""Persist the calendar to storage."""
self._path.write_text(ics_content)

12
homeassistant/components/local_calendar/strings.json Normal file
View File

@ -0,0 +1,12 @@
{
"config": {
"step": {
"user": {
"description": "Please choose a name for your new calendar",
"data": {
"calendar_name": "Calendar Name"
}
}
}
}
}

12
homeassistant/components/local_calendar/translations/en.json Normal file
View File

@ -0,0 +1,12 @@
{
"config": {
"step": {
"user": {
"data": {
"calendar_name": "Calendar Name"
},
"description": "Please choose a name for your new calendar"
}
}
}
}

1
homeassistant/generated/config_flows.py
View File

@ -226,6 +226,7 @@ FLOWS = {
"litejet",
"litterrobot",
"livisi",
"local_calendar",
"local_ip",
"locative",
"logi_circle",

6
homeassistant/generated/integrations.json
View File

@ -2878,6 +2878,12 @@
"config_flow": false,
"iot_class": "cloud_push"
},
"local_calendar": {
"name": "Local Calendar",
"integration_type": "hub",
"config_flow": true,
"iot_class": "local_polling"
},
"local_file": {
"name": "Local File",
"integration_type": "hub",

3
requirements_all.txt
View File

@ -925,6 +925,9 @@ ibm-watson==5.2.2
# homeassistant.components.watson_iot
ibmiotf==0.3.4
# homeassistant.components.local_calendar
ical==4.1.1
# homeassistant.components.ping
icmplib==3.0

1
requirements_test.txt
View File

@ -40,6 +40,7 @@ types-decorator==0.1.7
types-enum34==0.1.8
types-ipaddress==0.1.5
types-pkg-resources==0.1.3
types-python-dateutil==2.8.19.2
types-python-slugify==0.1.2
types-pytz==2021.1.2
types-PyYAML==5.4.6

3
requirements_test_all.txt
View File

@ -690,6 +690,9 @@ iaqualink==0.5.0
# homeassistant.components.ibeacon
ibeacon_ble==1.0.1
# homeassistant.components.local_calendar
ical==4.1.1
# homeassistant.components.ping
icmplib==3.0

3
tests/components/caldav/test_calendar.py
View File

@ -939,5 +939,8 @@ async def test_get_events_custom_calendars(hass, calendar, get_api_events):
"summary": "This is a normal event",
"location": "Hamburg",
"description": "Surprisingly rainy",
"uid": None,
"recurrence_id": None,
"rrule": None,
}
]

1
tests/components/local_calendar/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Tests for the Local Calendar integration."""

614
tests/components/local_calendar/test_calendar.py Normal file
View File

@ -0,0 +1,614 @@
"""Tests for calendar platform of local calendar."""
from collections.abc import Awaitable, Callable
import datetime
from http import HTTPStatus
from pathlib import Path
from typing import Any
from unittest.mock import patch
import urllib
from aiohttp import ClientSession, ClientWebSocketResponse
import pytest
from homeassistant.components.local_calendar import LocalCalendarStore
from homeassistant.components.local_calendar.const import CONF_CALENDAR_NAME, DOMAIN
from homeassistant.const import STATE_OFF, STATE_ON
from homeassistant.core import HomeAssistant
from homeassistant.helpers.template import DATE_STR_FORMAT
from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util
from tests.common import MockConfigEntry
CALENDAR_NAME = "Light Schedule"
FRIENDLY_NAME = "Light schedule"
TEST_ENTITY = "calendar.light_schedule"
class FakeStore(LocalCalendarStore):
"""Mock storage implementation."""
def __init__(self, hass: HomeAssistant, path: Path) -> None:
"""Initialize FakeStore."""
super().__init__(hass, path)
self._content = ""
def _load(self) -> str:
"""Read from calendar storage."""
return self._content
def _store(self, ics_content: str) -> None:
"""Persist the calendar storage."""
self._content = ics_content
@pytest.fixture(name="store", autouse=True)
def mock_store() -> None:
"""Test cleanup, remove any media storage persisted during the test."""
def new_store(hass: HomeAssistant, path: Path) -> FakeStore:
return FakeStore(hass, path)
with patch(
"homeassistant.components.local_calendar.LocalCalendarStore", new=new_store
):
yield
@pytest.fixture(name="time_zone")
def mock_time_zone() -> str:
"""Fixture for time zone to use in tests."""
# Set our timezone to CST/Regina so we can check calculations
# This keeps UTC-6 all year round
return "America/Regina"
@pytest.fixture(autouse=True)
def set_time_zone(hass: HomeAssistant, time_zone: str):
"""Set the time zone for the tests."""
# Set our timezone to CST/Regina so we can check calculations
# This keeps UTC-6 all year round
hass.config.set_time_zone(time_zone)
@pytest.fixture(name="config_entry")
def mock_config_entry() -> MockConfigEntry:
"""Fixture for mock configuration entry."""
return MockConfigEntry(domain=DOMAIN, data={CONF_CALENDAR_NAME: CALENDAR_NAME})
@pytest.fixture(name="setup_integration")
async def setup_integration(hass: HomeAssistant, config_entry: MockConfigEntry) -> None:
"""Set up the integration."""
config_entry.add_to_hass(hass)
assert await async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()
GetEventsFn = Callable[[str, str], Awaitable[dict[str, Any]]]
@pytest.fixture(name="get_events")
def get_events_fixture(
hass_client: Callable[..., Awaitable[ClientSession]]
) -> GetEventsFn:
"""Fetch calendar events from the HTTP API."""
async def _fetch(start: str, end: str) -> None:
client = await hass_client()
response = await client.get(
f"/api/calendars/{TEST_ENTITY}?start={urllib.parse.quote(start)}&end={urllib.parse.quote(end)}"
)
assert response.status == HTTPStatus.OK
return await response.json()
return _fetch
def event_fields(data: dict[str, str]) -> dict[str, str]:
"""Filter event API response to minimum fields."""
return {
k: data.get(k)
for k in ["summary", "start", "end", "recurrence_id"]
if data.get(k)
}
class Client:
"""Test client with helper methods for calendar websocket."""
def __init__(self, client):
"""Initialize Client."""
self.client = client
self.id = 0
async def cmd(self, cmd: str, payload: dict[str, Any] = None) -> dict[str, Any]:
"""Send a command and receive the json result."""
self.id += 1
await self.client.send_json(
{
"id": self.id,
"type": f"calendar/event/{cmd}",
**(payload if payload is not None else {}),
}
)
resp = await self.client.receive_json()
assert resp.get("id") == self.id
return resp
async def cmd_result(self, cmd: str, payload: dict[str, Any] = None) -> Any:
"""Send a command and parse the result."""
resp = await self.cmd(cmd, payload)
assert resp.get("success")
assert resp.get("type") == "result"
return resp.get("result")
ClientFixture = Callable[[], Awaitable[Client]]
@pytest.fixture
async def ws_client(
hass: HomeAssistant,
hass_ws_client: Callable[[HomeAssistant], Awaitable[ClientWebSocketResponse]],
) -> ClientFixture:
"""Fixture for creating the test websocket client."""
async def create_client() -> Client:
ws_client = await hass_ws_client(hass)
return Client(ws_client)
return create_client
async def test_empty_calendar(
hass: HomeAssistant, setup_integration: None, get_events: GetEventsFn
):
"""Test querying the API and fetching events."""
events = await get_events("1997-07-14T00:00:00", "1997-07-16T00:00:00")
assert len(events) == 0
state = hass.states.get(TEST_ENTITY)
assert state.name == FRIENDLY_NAME
assert state.state == STATE_OFF
assert dict(state.attributes) == {
"friendly_name": FRIENDLY_NAME,
"supported_features": 3,
}
async def test_api_date_time_event(
ws_client: ClientFixture, setup_integration: None, get_events: GetEventsFn
):
"""Test an event with a start/end date time."""
client = await ws_client()
await client.cmd_result(
"create",
{
"entity_id": TEST_ENTITY,
"event": {
"summary": "Bastille Day Party",
"dtstart": "1997-07-14T17:00:00+00:00",
"dtend": "1997-07-15T04:00:00+00:00",
},
},
)
events = await get_events("1997-07-14T00:00:00Z", "1997-07-16T00:00:00Z")
assert list(map(event_fields, events)) == [
{
"summary": "Bastille Day Party",
"start": {"dateTime": "1997-07-14T11:00:00-06:00"},
"end": {"dateTime": "1997-07-14T22:00:00-06:00"},
}
]
# Time range before event
events = await get_events("1997-07-13T00:00:00Z", "1997-07-14T16:00:00Z")
assert len(events) == 0
# Time range after event
events = await get_events("1997-07-15T05:00:00Z", "1997-07-15T06:00:00Z")
assert len(events) == 0
# Overlap with event start
events = await get_events("1997-07-13T00:00:00Z", "1997-07-14T18:00:00Z")
assert len(events) == 1
# Overlap with event end
events = await get_events("1997-07-15T03:00:00Z", "1997-07-15T06:00:00Z")
assert len(events) == 1
async def test_api_date_event(
ws_client: ClientFixture, setup_integration: None, get_events: GetEventsFn
):
"""Test an event with a start/end date all day event."""
client = await ws_client()
await client.cmd_result(
"create",
{
"entity_id": TEST_ENTITY,
"event": {
"summary": "Festival International de Jazz de Montreal",
"dtstart": "2007-06-28",
"dtend": "2007-07-09",
},
},
)
events = await get_events("2007-06-20T00:00:00", "2007-07-20T00:00:00")
assert list(map(event_fields, events)) == [
{
"summary": "Festival International de Jazz de Montreal",
"start": {"date": "2007-06-28"},
"end": {"date": "2007-07-09"},
}
]
# Time range before event (timezone is -6)
events = await get_events("2007-06-26T00:00:00Z", "2007-06-28T01:00:00Z")
assert len(events) == 0
# Time range after event
events = await get_events("2007-07-10T00:00:00Z", "2007-07-11T00:00:00Z")
assert len(events) == 0
# Overlap with event start (timezone is -6)
events = await get_events("2007-06-26T00:00:00Z", "2007-06-28T08:00:00Z")
assert len(events) == 1
# Overlap with event end
events = await get_events("2007-07-09T00:00:00Z", "2007-07-11T00:00:00Z")
assert len(events) == 1
async def test_active_event(
hass: HomeAssistant,
ws_client: ClientFixture,
setup_integration: None,
):
"""Test an event with a start/end date time."""
start = dt_util.now() - datetime.timedelta(minutes=30)
end = dt_util.now() + datetime.timedelta(minutes=30)
client = await ws_client()
await client.cmd_result(
"create",
{
"entity_id": TEST_ENTITY,
"event": {
"summary": "Evening lights",
"dtstart": start.isoformat(),
"dtend": end.isoformat(),
},
},
)
state = hass.states.get(TEST_ENTITY)
assert state.name == FRIENDLY_NAME
assert state.state == STATE_ON
assert dict(state.attributes) == {
"friendly_name": FRIENDLY_NAME,
"message": "Evening lights",
"all_day": False,
"description": "",
"location": "",
"start_time": start.strftime(DATE_STR_FORMAT),
"end_time": end.strftime(DATE_STR_FORMAT),
"supported_features": 3,
}
async def test_upcoming_event(
hass: HomeAssistant,
ws_client: ClientFixture,
setup_integration: None,
):
"""Test an event with a start/end date time."""
start = dt_util.now() + datetime.timedelta(days=1)
end = dt_util.now() + datetime.timedelta(days=1, hours=1)
client = await ws_client()
await client.cmd_result(
"create",
{
"entity_id": TEST_ENTITY,
"event": {
"summary": "Evening lights",
"dtstart": start.isoformat(),
"dtend": end.isoformat(),
},
},
)
state = hass.states.get(TEST_ENTITY)
assert state.name == FRIENDLY_NAME
assert state.state == STATE_OFF
assert dict(state.attributes) == {
"friendly_name": FRIENDLY_NAME,
"message": "Evening lights",
"all_day": False,
"description": "",
"location": "",
"message": "Evening lights",
"start_time": start.strftime(DATE_STR_FORMAT),
"end_time": end.strftime(DATE_STR_FORMAT),
"supported_features": 3,
}
async def test_recurring_event(
ws_client: ClientFixture,
setup_integration: None,
hass: HomeAssistant,
get_events: GetEventsFn,
):
"""Test an event with a recurrence rule."""
client = await ws_client()
await client.cmd_result(
"create",
{
"entity_id": TEST_ENTITY,
"event": {
"summary": "Monday meeting",
"dtstart": "2022-08-29T09:00:00",
"dtend": "2022-08-29T10:00:00",
"rrule": "FREQ=WEEKLY",
},
},
)
events = await get_events("2022-08-20T00:00:00", "2022-09-20T00:00:00")
assert list(map(event_fields, events)) == [
{
"summary": "Monday meeting",
"start": {"dateTime": "2022-08-29T09:00:00-06:00"},
"end": {"dateTime": "2022-08-29T10:00:00-06:00"},
"recurrence_id": "20220829T090000",
},
{
"summary": "Monday meeting",
"start": {"dateTime": "2022-09-05T09:00:00-06:00"},
"end": {"dateTime": "2022-09-05T10:00:00-06:00"},
"recurrence_id": "20220905T090000",
},
{
"summary": "Monday meeting",
"start": {"dateTime": "2022-09-12T09:00:00-06:00"},
"end": {"dateTime": "2022-09-12T10:00:00-06:00"},
"recurrence_id": "20220912T090000",
},
{
"summary": "Monday meeting",
"start": {"dateTime": "2022-09-19T09:00:00-06:00"},
"end": {"dateTime": "2022-09-19T10:00:00-06:00"},
"recurrence_id": "20220919T090000",
},
]
async def test_websocket_delete(
ws_client: ClientFixture, setup_integration: None, get_events: GetEventsFn
):
"""Test websocket delete command."""
client = await ws_client()
await client.cmd_result(
"create",
{
"entity_id": TEST_ENTITY,
"event": {
"summary": "Bastille Day Party",
"dtstart": "1997-07-14T17:00:00+00:00",
"dtend": "1997-07-15T04:00:00+00:00",
},
},
)
events = await get_events("1997-07-14T00:00:00", "1997-07-16T00:00:00")
assert list(map(event_fields, events)) == [
{
"summary": "Bastille Day Party",
"start": {"dateTime": "1997-07-14T11:00:00-06:00"},
"end": {"dateTime": "1997-07-14T22:00:00-06:00"},
}
]
uid = events[0]["uid"]
# Delete the event
await client.cmd_result(
"delete",
{
"entity_id": TEST_ENTITY,
"uid": uid,
},
)
events = await get_events("1997-07-14T00:00:00", "1997-07-16T00:00:00")
assert list(map(event_fields, events)) == []
async def test_websocket_delete_recurring(
ws_client: ClientFixture, setup_integration: None, get_events: GetEventsFn
):
"""Test deleting a recurring event."""
client = await ws_client()
await client.cmd_result(
"create",
{
"entity_id": TEST_ENTITY,
"event": {
"summary": "Morning Routine",
"dtstart": "2022-08-22T08:30:00",
"dtend": "2022-08-22T09:00:00",
"rrule": "FREQ=DAILY",
},
},
)
events = await get_events("2022-08-22T00:00:00", "2022-08-26T00:00:00")
assert list(map(event_fields, events)) == [
{
"summary": "Morning Routine",
"start": {"dateTime": "2022-08-22T08:30:00-06:00"},
"end": {"dateTime": "2022-08-22T09:00:00-06:00"},
"recurrence_id": "20220822T083000",
},
{
"summary": "Morning Routine",
"start": {"dateTime": "2022-08-23T08:30:00-06:00"},
"end": {"dateTime": "2022-08-23T09:00:00-06:00"},
"recurrence_id": "20220823T083000",
},
{
"summary": "Morning Routine",
"start": {"dateTime": "2022-08-24T08:30:00-06:00"},
"end": {"dateTime": "2022-08-24T09:00:00-06:00"},
"recurrence_id": "20220824T083000",
},
{
"summary": "Morning Routine",
"start": {"dateTime": "2022-08-25T08:30:00-06:00"},
"end": {"dateTime": "2022-08-25T09:00:00-06:00"},
"recurrence_id": "20220825T083000",
},
]
uid = events[0]["uid"]
assert [event["uid"] for event in events] == [uid] * 4
# Cancel a single instance and confirm it was removed
await client.cmd_result(
"delete",
{
"entity_id": TEST_ENTITY,
"uid": uid,
"recurrence_id": "20220824T083000",
},
)
events = await get_events("2022-08-22T00:00:00", "2022-08-26T00:00:00")
assert list(map(event_fields, events)) == [
{
"summary": "Morning Routine",
"start": {"dateTime": "2022-08-22T08:30:00-06:00"},
"end": {"dateTime": "2022-08-22T09:00:00-06:00"},
"recurrence_id": "20220822T083000",
},
{
"summary": "Morning Routine",
"start": {"dateTime": "2022-08-23T08:30:00-06:00"},
"end": {"dateTime": "2022-08-23T09:00:00-06:00"},
"recurrence_id": "20220823T083000",
},
{
"summary": "Morning Routine",
"start": {"dateTime": "2022-08-25T08:30:00-06:00"},
"end": {"dateTime": "2022-08-25T09:00:00-06:00"},
"recurrence_id": "20220825T083000",
},
]
# Delete all and future and confirm multiple were removed
await client.cmd_result(
"delete",
{
"entity_id": TEST_ENTITY,
"uid": uid,
"recurrence_id": "20220823T083000",
"recurrence_range": "THISANDFUTURE",
},
)
events = await get_events("2022-08-22T00:00:00", "2022-08-26T00:00:00")
assert list(map(event_fields, events)) == [
{
"summary": "Morning Routine",
"start": {"dateTime": "2022-08-22T08:30:00-06:00"},
"end": {"dateTime": "2022-08-22T09:00:00-06:00"},
"recurrence_id": "20220822T083000",
},
]
@pytest.mark.parametrize(
"rrule",
[
"FREQ=SECONDLY",
"FREQ=MINUTELY",
"FREQ=HOURLY",
"invalid",
"",
],
)
async def test_invalid_rrule(
ws_client: ClientFixture,
setup_integration: None,
hass: HomeAssistant,
get_events: GetEventsFn,
rrule: str,
):
"""Test an event with a recurrence rule."""
client = await ws_client()
resp = await client.cmd(
"create",
{
"entity_id": TEST_ENTITY,
"event": {
"summary": "Monday meeting",
"dtstart": "2022-08-29T09:00:00",
"dtend": "2022-08-29T10:00:00",
"rrule": rrule,
},
},
)
assert not resp.get("success")
assert "error" in resp
assert resp.get("error").get("code") == "invalid_format"
@pytest.mark.parametrize(
"time_zone,event_order",
[
("America/Los_Angeles", ["One", "Two", "All Day Event"]),
("America/Regina", ["One", "Two", "All Day Event"]),
("UTC", ["One", "All Day Event", "Two"]),
("Asia/Tokyo", ["All Day Event", "One", "Two"]),
],
)
async def test_all_day_iter_order(
hass: HomeAssistant,
ws_client: ClientFixture,
setup_integration: None,
get_events: GetEventsFn,
event_order: list[str],
):
"""Test the sort order of an all day events depending on the time zone."""
client = await ws_client()
await client.cmd_result(
"create",
{
"entity_id": TEST_ENTITY,
"event": {
"summary": "All Day Event",
"dtstart": "2022-10-08",
"dtend": "2022-10-09",
},
},
)
await client.cmd_result(
"create",
{
"entity_id": TEST_ENTITY,
"event": {
"summary": "One",
"dtstart": "2022-10-07T23:00:00+00:00",
"dtend": "2022-10-07T23:30:00+00:00",
},
},
)
await client.cmd_result(
"create",
{
"entity_id": TEST_ENTITY,
"event": {
"summary": "Two",
"dtstart": "2022-10-08T01:00:00+00:00",
"dtend": "2022-10-08T02:00:00+00:00",
},
},
)
events = await get_events("2022-10-06T00:00:00Z", "2022-10-09T00:00:00Z")
assert [event["summary"] for event in events] == event_order

35
tests/components/local_calendar/test_config_flow.py Normal file
View File

@ -0,0 +1,35 @@
"""Test the Local Calendar config flow."""
from unittest.mock import patch
from homeassistant import config_entries
from homeassistant.components.local_calendar.const import CONF_CALENDAR_NAME, DOMAIN
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
async def test_form(hass: HomeAssistant) -> None:
"""Test we get the form."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["errors"] is None
with patch(
"homeassistant.components.local_calendar.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_CALENDAR_NAME: "My Calendar",
},
)
await hass.async_block_till_done()
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "My Calendar"
assert result2["data"] == {
CONF_CALENDAR_NAME: "My Calendar",
}
assert len(mock_setup_entry.mock_calls) == 1

3
tests/components/twentemilieu/test_calendar.py
View File

@ -81,4 +81,7 @@ async def test_api_events(
"summary": "Christmas tree pickup",
"description": None,
"location": None,
"uid": None,
"recurrence_id": None,
"rrule": None,
}