jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-25 01:38:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add a caldav calendar component (#10842)

Browse Source 
* Add caldav component

* Code review - 1

* Code review - 2

* Sort imports
This commit is contained in:
maxlaverse 2017-12-10 17:44:28 +01:00 committed by Pascal Vizeli
parent 3b228c78c0
commit 04cb893d10
6 changed files with 540 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.coveragerc
View File

@ -284,6 +284,7 @@ omit =
homeassistant/components/binary_sensor/rest.py
homeassistant/components/binary_sensor/tapsaff.py
homeassistant/components/browser.py
homeassistant/components/calendar/caldav.py
homeassistant/components/calendar/todoist.py
homeassistant/components/camera/bloomsky.py
homeassistant/components/camera/canary.py

230
homeassistant/components/calendar/caldav.py Normal file
View File

@ -0,0 +1,230 @@
"""
Support for WebDav Calendar.
For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/calendar.caldav/
"""
import logging
import re
from datetime import datetime, timedelta
import voluptuous as vol
import homeassistant.helpers.config_validation as cv
from homeassistant.components.calendar import (
CalendarEventDevice, PLATFORM_SCHEMA)
from homeassistant.const import (
CONF_NAME, CONF_PASSWORD, CONF_URL, CONF_USERNAME)
from homeassistant.util import dt, Throttle
REQUIREMENTS = ['caldav==0.5.0']
_LOGGER = logging.getLogger(__name__)
CONF_DEVICE_ID = 'device_id'
CONF_CALENDARS = 'calendars'
CONF_CUSTOM_CALENDARS = 'custom_calendars'
CONF_CALENDAR = 'calendar'
CONF_ALL_DAY = 'all_day'
CONF_SEARCH = 'search'
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(CONF_URL): vol.Url,
vol.Optional(CONF_CALENDARS, default=[]):
vol.All(cv.ensure_list, vol.Schema([
cv.string
])),
vol.Inclusive(CONF_USERNAME, 'authentication'): cv.string,
vol.Inclusive(CONF_PASSWORD, 'authentication'): cv.string,
vol.Optional(CONF_CUSTOM_CALENDARS, default=[]):
vol.All(cv.ensure_list, vol.Schema([
vol.Schema({
vol.Required(CONF_NAME): cv.string,
vol.Required(CONF_CALENDAR): cv.string,
vol.Required(CONF_SEARCH): cv.string
})
]))
})
MIN_TIME_BETWEEN_UPDATES = timedelta(minutes=15)
def setup_platform(hass, config, add_devices, disc_info=None):
"""Set up the WebDav Calendar platform."""
import caldav
client = caldav.DAVClient(config.get(CONF_URL),
None,
config.get(CONF_USERNAME),
config.get(CONF_PASSWORD))
# Retrieve all the remote calendars
calendars = client.principal().calendars()
calendar_devices = []
for calendar in list(calendars):
# If a calendar name was given in the configuration,
# ignore all the others
if (config.get(CONF_CALENDARS)
and calendar.name not in config.get(CONF_CALENDARS)):
_LOGGER.debug("Ignoring calendar '%s'", calendar.name)
continue
# Create additional calendars based on custom filtering
# rules
for cust_calendar in config.get(CONF_CUSTOM_CALENDARS):
# Check that the base calendar matches
if cust_calendar.get(CONF_CALENDAR) != calendar.name:
continue
device_data = {
CONF_NAME: cust_calendar.get(CONF_NAME),
CONF_DEVICE_ID: "{} {}".format(
cust_calendar.get(CONF_CALENDAR),
cust_calendar.get(CONF_NAME)),
}
calendar_devices.append(
WebDavCalendarEventDevice(hass,
device_data,
calendar,
cust_calendar.get(CONF_ALL_DAY),
cust_calendar.get(CONF_SEARCH))
)
# Create a default calendar if there was no custom one
if not config.get(CONF_CUSTOM_CALENDARS):
device_data = {
CONF_NAME: calendar.name,
CONF_DEVICE_ID: calendar.name
}
calendar_devices.append(
WebDavCalendarEventDevice(hass, device_data, calendar)
)
# Finally add all the calendars we've created
add_devices(calendar_devices)
class WebDavCalendarEventDevice(CalendarEventDevice):
"""A device for getting the next Task from a WebDav Calendar."""
def __init__(self,
hass,
device_data,
calendar,
all_day=False,
search=None):
"""Create the WebDav Calendar Event Device."""
self.data = WebDavCalendarData(calendar, all_day, search)
super().__init__(hass, device_data)
@property
def device_state_attributes(self):
"""Return the device state attributes."""
if self.data.event is None:
# No tasks, we don't REALLY need to show anything.
return {}
attributes = super().device_state_attributes
return attributes
class WebDavCalendarData(object):
"""Class to utilize the calendar dav client object to get next event."""
def __init__(self, calendar, include_all_day, search):
"""Set up how we are going to search the WebDav calendar."""
self.calendar = calendar
self.include_all_day = include_all_day
self.search = search
self.event = None
@Throttle(MIN_TIME_BETWEEN_UPDATES)
def update(self):
"""Get the latest data."""
# We have to retrieve the results for the whole day as the server
# won't return events that have already started
results = self.calendar.date_search(
dt.start_of_local_day(),
dt.start_of_local_day() + timedelta(days=1)
)
# dtstart can be a date or datetime depending if the event lasts a
# whole day. Convert everything to datetime to be able to sort it
results.sort(key=lambda x: self.to_datetime(
x.instance.vevent.dtstart.value
))
vevent = next((
event.instance.vevent for event in results
if (self.is_matching(event.instance.vevent, self.search)
and (not self.is_all_day(event.instance.vevent)
or self.include_all_day)
and not self.is_over(event.instance.vevent))), None)
# If no matching event could be found
if vevent is None:
_LOGGER.debug(
"No matching event found in the %d results for %s",
len(results),
self.calendar.name,
)
self.event = None
return True
# Populate the entity attributes with the event values
self.event = {
"summary": vevent.summary.value,
"start": self.get_hass_date(vevent.dtstart.value),
"end": self.get_hass_date(vevent.dtend.value),
"location": self.get_attr_value(vevent, "location"),
"description": self.get_attr_value(vevent, "description")
}
return True
@staticmethod
def is_matching(vevent, search):
"""Return if the event matches the filter critera."""
if search is None:
return True
pattern = re.compile(search)
return (hasattr(vevent, "summary")
and pattern.match(vevent.summary.value)
or hasattr(vevent, "location")
and pattern.match(vevent.location.value)
or hasattr(vevent, "description")
and pattern.match(vevent.description.value))
@staticmethod
def is_all_day(vevent):
"""Return if the event last the whole day."""
return not isinstance(vevent.dtstart.value, datetime)
@staticmethod
def is_over(vevent):
"""Return if the event is over."""
return dt.now() > WebDavCalendarData.to_datetime(vevent.dtend.value)
@staticmethod
def get_hass_date(obj):
"""Return if the event matches."""
if isinstance(obj, datetime):
return {"dateTime": obj.isoformat()}
return {"date": obj.isoformat()}
@staticmethod
def to_datetime(obj):
"""Return a datetime."""
if isinstance(obj, datetime):
return obj
return dt.as_local(dt.dt.datetime.combine(obj, dt.dt.time.min))
@staticmethod
def get_attr_value(obj, attribute):
"""Return the value of the attribute if defined."""
if hasattr(obj, attribute):
return getattr(obj, attribute).value
return None

3
requirements_all.txt
View File

@ -160,6 +160,9 @@ broadlink==0.5
# homeassistant.components.weather.buienradar
buienradar==0.9
# homeassistant.components.calendar.caldav
caldav==0.5.0
# homeassistant.components.notify.ciscospark
ciscosparkapi==0.4.2

3
requirements_test_all.txt
View File

@ -36,6 +36,9 @@ aiohttp_cors==0.5.3
# homeassistant.components.notify.apns
apns2==0.3.0
# homeassistant.components.calendar.caldav
caldav==0.5.0
# homeassistant.components.sensor.coinmarketcap
coinmarketcap==4.1.1

1
script/gen_requirements_all.py
View File

@ -38,6 +38,7 @@ TEST_REQUIREMENTS = (
'aioautomatic',
'aiohttp_cors',
'apns2',
'caldav',
'coinmarketcap',
'defusedxml',
'dsmr_parser',

302
tests/components/calendar/test_caldav.py Normal file
View File

@ -0,0 +1,302 @@
"""The tests for the webdav calendar component."""
# pylint: disable=protected-access
import datetime
import logging
import unittest
from unittest.mock import (patch, Mock, MagicMock)
import homeassistant.components.calendar as calendar_base
import homeassistant.components.calendar.caldav as caldav
from caldav.objects import Event
from homeassistant.const import CONF_PLATFORM, STATE_OFF, STATE_ON
from homeassistant.util import dt
from tests.common import get_test_home_assistant
TEST_PLATFORM = {calendar_base.DOMAIN: {CONF_PLATFORM: 'test'}}
_LOGGER = logging.getLogger(__name__)
DEVICE_DATA = {
"name": "Private Calendar",
"device_id": "Private Calendar"
}
EVENTS = [
"""BEGIN:VCALENDAR
VERSION:2.0
PRODID:-//E-Corp.//CalDAV Client//EN
BEGIN:VEVENT
UID:1
DTSTAMP:20171125T000000Z
DTSTART:20171127T170000Z
DTEND:20171127T180000Z
SUMMARY:This is a normal event
LOCATION:Hamburg
DESCRIPTION:Surprisingly rainy
END:VEVENT
END:VCALENDAR
""",
"""BEGIN:VCALENDAR
VERSION:2.0
PRODID:-//Global Dynamics.//CalDAV Client//EN
BEGIN:VEVENT
UID:2
DTSTAMP:20171125T000000Z
DTSTART:20171127T100000Z
DTEND:20171127T110000Z
SUMMARY:This is an offset event !!-02:00
LOCATION:Hamburg
DESCRIPTION:Surprisingly shiny
END:VEVENT
END:VCALENDAR
""",
"""BEGIN:VCALENDAR
VERSION:2.0
PRODID:-//Global Corp.//CalDAV Client//EN
BEGIN:VEVENT
UID:3
DTSTAMP:20171125T000000Z
DTSTART:20171127
DTEND:20171128
SUMMARY:This is an all day event
LOCATION:Hamburg
DESCRIPTION:What a beautiful day
END:VEVENT
END:VCALENDAR
"""
]
def _local_datetime(hours, minutes):
"""Build a datetime object for testing in the correct timezone."""
return dt.as_local(datetime.datetime(2017, 11, 27, hours, minutes, 0))
def _mocked_dav_client(*args, **kwargs):
"""Mock requests.get invocations."""
calendars = [
_mock_calendar("First"),
_mock_calendar("Second")
]
principal = Mock()
principal.calendars = MagicMock(return_value=calendars)
client = Mock()
client.principal = MagicMock(return_value=principal)
return client
def _mock_calendar(name):
events = []
for idx, event in enumerate(EVENTS):
events.append(Event(None, "%d.ics" % idx, event, None, str(idx)))
calendar = Mock()
calendar.date_search = MagicMock(return_value=events)
calendar.name = name
return calendar
class TestComponentsWebDavCalendar(unittest.TestCase):
"""Test the WebDav calendar."""
hass = None # HomeAssistant
# pylint: disable=invalid-name
def setUp(self):
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
self.calendar = _mock_calendar("Private")
# pylint: disable=invalid-name
def tearDown(self):
"""Stop everything that was started."""
self.hass.stop()
@patch('caldav.DAVClient', side_effect=_mocked_dav_client)
def test_setup_component(self, req_mock):
"""Test setup component with calendars."""
def _add_device(devices):
assert len(devices) == 2
assert devices[0].name == "First"
assert devices[0].dev_id == "First"
assert devices[1].name == "Second"
assert devices[1].dev_id == "Second"
caldav.setup_platform(self.hass,
{
"url": "http://test.local",
"custom_calendars": []
},
_add_device)
@patch('caldav.DAVClient', side_effect=_mocked_dav_client)
def test_setup_component_with_no_calendar_matching(self, req_mock):
"""Test setup component with wrong calendar."""
def _add_device(devices):
assert not devices
caldav.setup_platform(self.hass,
{
"url": "http://test.local",
"calendars": ["none"],
"custom_calendars": []
},
_add_device)
@patch('caldav.DAVClient', side_effect=_mocked_dav_client)
def test_setup_component_with_a_calendar_match(self, req_mock):
"""Test setup component with right calendar."""
def _add_device(devices):
assert len(devices) == 1
assert devices[0].name == "Second"
caldav.setup_platform(self.hass,
{
"url": "http://test.local",
"calendars": ["Second"],
"custom_calendars": []
},
_add_device)
@patch('caldav.DAVClient', side_effect=_mocked_dav_client)
def test_setup_component_with_one_custom_calendar(self, req_mock):
"""Test setup component with custom calendars."""
def _add_device(devices):
assert len(devices) == 1
assert devices[0].name == "HomeOffice"
assert devices[0].dev_id == "Second HomeOffice"
caldav.setup_platform(self.hass,
{
"url": "http://test.local",
"custom_calendars": [
{
"name": "HomeOffice",
"calendar": "Second",
"filter": "HomeOffice"
}]
},
_add_device)
@patch('homeassistant.util.dt.now', return_value=_local_datetime(17, 30))
def test_ongoing_event(self, mock_now):
"""Test that the ongoing event is returned."""
cal = caldav.WebDavCalendarEventDevice(self.hass,
DEVICE_DATA,
self.calendar)
self.assertEqual(cal.name, DEVICE_DATA["name"])
self.assertEqual(cal.state, STATE_ON)
self.assertEqual(cal.device_state_attributes, {
"message": "This is a normal event",
"all_day": False,
"offset_reached": False,
"start_time": "2017-11-27 17:00:00",
"end_time": "2017-11-27 18:00:00",
"location": "Hamburg",
"description": "Surprisingly rainy"
})
@patch('homeassistant.util.dt.now', return_value=_local_datetime(8, 30))
def test_ongoing_event_with_offset(self, mock_now):
"""Test that the offset is taken into account."""
cal = caldav.WebDavCalendarEventDevice(self.hass,
DEVICE_DATA,
self.calendar)
self.assertEqual(cal.state, STATE_OFF)
self.assertEqual(cal.device_state_attributes, {
"message": "This is an offset event",
"all_day": False,
"offset_reached": True,
"start_time": "2017-11-27 10:00:00",
"end_time": "2017-11-27 11:00:00",
"location": "Hamburg",
"description": "Surprisingly shiny"
})
@patch('homeassistant.util.dt.now', return_value=_local_datetime(12, 00))
def test_matching_filter(self, mock_now):
"""Test that the matching event is returned."""
cal = caldav.WebDavCalendarEventDevice(self.hass,
DEVICE_DATA,
self.calendar,
False,
"This is a normal event")
self.assertEqual(cal.state, STATE_OFF)
self.assertFalse(cal.offset_reached())
self.assertEqual(cal.device_state_attributes, {
"message": "This is a normal event",
"all_day": False,
"offset_reached": False,
"start_time": "2017-11-27 17:00:00",
"end_time": "2017-11-27 18:00:00",
"location": "Hamburg",
"description": "Surprisingly rainy"
})
@patch('homeassistant.util.dt.now', return_value=_local_datetime(12, 00))
def test_matching_filter_real_regexp(self, mock_now):
"""Test that the event matching the regexp is returned."""
cal = caldav.WebDavCalendarEventDevice(self.hass,
DEVICE_DATA,
self.calendar,
False,
"^This.*event")
self.assertEqual(cal.state, STATE_OFF)
self.assertFalse(cal.offset_reached())
self.assertEqual(cal.device_state_attributes, {
"message": "This is a normal event",
"all_day": False,
"offset_reached": False,
"start_time": "2017-11-27 17:00:00",
"end_time": "2017-11-27 18:00:00",
"location": "Hamburg",
"description": "Surprisingly rainy"
})
@patch('homeassistant.util.dt.now', return_value=_local_datetime(20, 00))
def test_filter_matching_past_event(self, mock_now):
"""Test that the matching past event is not returned."""
cal = caldav.WebDavCalendarEventDevice(self.hass,
DEVICE_DATA,
self.calendar,
False,
"This is a normal event")
self.assertEqual(cal.data.event, None)
@patch('homeassistant.util.dt.now', return_value=_local_datetime(12, 00))
def test_no_result_with_filtering(self, mock_now):
"""Test that nothing is returned since nothing matches."""
cal = caldav.WebDavCalendarEventDevice(self.hass,
DEVICE_DATA,
self.calendar,
False,
"This is a non-existing event")
self.assertEqual(cal.data.event, None)
@patch('homeassistant.util.dt.now', return_value=_local_datetime(17, 30))
def test_all_day_event_returned(self, mock_now):
"""Test that the event lasting the whole day is returned."""
cal = caldav.WebDavCalendarEventDevice(self.hass,
DEVICE_DATA,
self.calendar,
True)
self.assertEqual(cal.name, DEVICE_DATA["name"])
self.assertEqual(cal.state, STATE_ON)
self.assertEqual(cal.device_state_attributes, {
"message": "This is an all day event",
"all_day": True,
"offset_reached": False,
"start_time": "2017-11-27 00:00:00",
"end_time": "2017-11-28 00:00:00",
"location": "Hamburg",
"description": "What a beautiful day"
})