Merge pull request #1293 from metallerok/processing-expand-property

Processing expand property

radicale/app/report.py

@@ -18,11 +18,17 @@
 # along with Radicale.  If not, see <http://www.gnu.org/licenses/>.
 
 import contextlib
+import datetime
 import posixpath
 import socket
+import copy
 import xml.etree.ElementTree as ET
+from vobject.base import ContentLine
 from http import client
-from typing import Callable, Iterable, Iterator, Optional, Sequence, Tuple
+from typing import (
+    Callable, Iterable, Iterator,
+    Optional, Sequence, Tuple,
+)
 from urllib.parse import unquote, urlparse
 
 import radicale.item as radicale_item
@@ -64,9 +70,8 @@ def xml_report(base_prefix: str, path: str, xml_request: Optional[ET.Element],
         logger.warning("Invalid REPORT method %r on %r requested",
                        xmlutils.make_human_tag(root.tag), path)
         return client.FORBIDDEN, xmlutils.webdav_error("D:supported-report")
-    prop_element = root.find(xmlutils.make_clark("D:prop"))
+
-    props = ([prop.tag for prop in prop_element]
+    props = root.find(xmlutils.make_clark("D:prop")) or []
-             if prop_element is not None else [])
 
     hreferences: Iterable[str]
     if root.tag in (
@@ -138,18 +143,39 @@ def xml_report(base_prefix: str, path: str, xml_request: Optional[ET.Element],
         found_props = []
         not_found_props = []
 
-        for tag in props:
+        for prop in props:
-            element = ET.Element(tag)
+            element = ET.Element(prop.tag)
-            if tag == xmlutils.make_clark("D:getetag"):
+            if prop.tag == xmlutils.make_clark("D:getetag"):
                 element.text = item.etag
                 found_props.append(element)
-            elif tag == xmlutils.make_clark("D:getcontenttype"):
+            elif prop.tag == xmlutils.make_clark("D:getcontenttype"):
                 element.text = xmlutils.get_content_type(item, encoding)
                 found_props.append(element)
-            elif tag in (
+            elif prop.tag in (
                     xmlutils.make_clark("C:calendar-data"),
                     xmlutils.make_clark("CR:address-data")):
                 element.text = item.serialize()
+
+                expand = prop.find(xmlutils.make_clark("C:expand"))
+                if expand is not None:
+                    start = expand.get('start')
+                    end = expand.get('end')
+
+                    if (start is None) or (end is None):
+                        return client.FORBIDDEN, \
+                            xmlutils.webdav_error("C:expand")
+
+                    start = datetime.datetime.strptime(
+                        start, '%Y%m%dT%H%M%SZ'
+                    ).replace(tzinfo=datetime.timezone.utc)
+                    end = datetime.datetime.strptime(
+                        end, '%Y%m%dT%H%M%SZ'
+                    ).replace(tzinfo=datetime.timezone.utc)
+
+                    expanded_element = _expand(
+                        element, copy.copy(item), start, end)
+                    found_props.append(expanded_element)
+                else:
                     found_props.append(element)
             else:
                 not_found_props.append(element)
@@ -164,6 +190,90 @@ def xml_report(base_prefix: str, path: str, xml_request: Optional[ET.Element],
     return client.MULTI_STATUS, multistatus
 
 
+def _expand(
+        element: ET.Element,
+        item: radicale_item.Item,
+        start: datetime.datetime,
+        end: datetime.datetime,
+) -> ET.Element:
+    rruleset = None
+    if hasattr(item.vobject_item.vevent, 'rrule'):
+        rruleset = item.vobject_item.vevent.getrruleset()
+
+    expanded_item = _make_vobject_expanded_item(item)
+
+    if rruleset:
+        recurrences = rruleset.between(start, end)
+
+        expanded = None
+        for recurrence_dt in recurrences:
+            vobject_item = copy.copy(expanded_item.vobject_item)
+
+            recurrence_utc = recurrence_dt.astimezone(datetime.timezone.utc)
+
+            vevent = copy.deepcopy(vobject_item.vevent)
+            vevent.recurrence_id = ContentLine(
+                name='RECURRENCE-ID',
+                value=recurrence_utc.strftime('%Y%m%dT%H%M%SZ'), params={}
+            )
+
+            if expanded is None:
+                vobject_item.vevent = vevent
+                expanded = vobject_item
+            else:
+                expanded.add(vevent)
+
+        element.text = expanded.serialize()
+    else:
+        element.text = expanded_item.vobject_item.serialize()
+
+    return element
+
+
+def _make_vobject_expanded_item(
+        item: radicale_item.Item
+) -> radicale_item.Item:
+    # https://www.rfc-editor.org/rfc/rfc4791#section-9.6.5
+    # The returned calendar components MUST NOT use recurrence
+    #       properties (i.e., EXDATE, EXRULE, RDATE, and RRULE) and MUST NOT
+    #       have reference to or include VTIMEZONE components.  Date and local
+    #       time with reference to time zone information MUST be converted
+    #       into date with UTC time.
+
+    item = copy.copy(item)
+    vevent = item.vobject_item.vevent
+
+    start_utc = vevent.dtstart.value.astimezone(datetime.timezone.utc)
+    vevent.dtstart = ContentLine(
+        name='DTSTART',
+        value=start_utc.strftime('%Y%m%dT%H%M%SZ'), params={})
+
+    dt_end = getattr(vevent, 'dtend', None)
+    if dt_end is not None:
+        end_utc = dt_end.value.astimezone(datetime.timezone.utc)
+        vevent.dtend = ContentLine(
+            name='DTEND',
+            value=end_utc.strftime('%Y%m%dT%H%M%SZ'), params={})
+
+    timezones_to_remove = []
+    for component in item.vobject_item.components():
+        if component.name == 'VTIMEZONE':
+            timezones_to_remove.append(component)
+
+    for timezone in timezones_to_remove:
+        item.vobject_item.remove(timezone)
+
+    try:
+        delattr(item.vobject_item.vevent, 'rrule')
+        delattr(item.vobject_item.vevent, 'exdate')
+        delattr(item.vobject_item.vevent, 'exrule')
+        delattr(item.vobject_item.vevent, 'rdate')
+    except AttributeError:
+        pass
+
+    return item
+
+
 def xml_item_response(base_prefix: str, href: str,
                       found_props: Sequence[ET.Element] = (),
                       not_found_props: Sequence[ET.Element] = (),

radicale/tests/static/event_daily_rrule.ics

@@ -0,0 +1,28 @@
+BEGIN:VCALENDAR
+VERSION:2.0
+BEGIN:VTIMEZONE
+LAST-MODIFIED:20040110T032845Z
+TZID:US/Eastern
+BEGIN:DAYLIGHT
+DTSTART:20000404T020000
+RRULE:FREQ=YEARLY;BYDAY=1SU;BYMONTH=4
+TZNAME:EDT
+TZOFFSETFROM:-0500
+TZOFFSETTO:-0400
+END:DAYLIGHT
+BEGIN:STANDARD
+DTSTART:20001026T020000
+RRULE:FREQ=YEARLY;BYDAY=-1SU;BYMONTH=10
+TZNAME:EST
+TZOFFSETFROM:-0400
+TZOFFSETTO:-0500
+END:STANDARD
+END:VTIMEZONE
+BEGIN:VEVENT
+DTSTART;TZID=US/Eastern:20060102T120000
+DURATION:PT1H
+RRULE:FREQ=DAILY;COUNT=5
+SUMMARY:Recurring event
+UID:event_daily_rrule
+END:VEVENT
+END:VCALENDAR

radicale/tests/test_base.py

@@ -1525,6 +1525,87 @@ permissions: RrWw""")
             calendar_path, "http://radicale.org/ns/sync/INVALID")
         assert not sync_token
 
+    def test_report_with_expand_property(self) -> None:
+        self.put("/calendar.ics/", get_file_content("event_daily_rrule.ics"))
+        req_body_without_expand = \
+            """<?xml version="1.0" encoding="utf-8" ?>
+            <C:calendar-query xmlns:D="DAV:" xmlns:C="urn:ietf:params:xml:ns:caldav">
+                <D:prop>
+                    <C:calendar-data>
+                    </C:calendar-data>
+                </D:prop>
+                <C:filter>
+                    <C:comp-filter name="VCALENDAR">
+                        <C:comp-filter name="VEVENT">
+                            <C:time-range start="20060103T000000Z" end="20060105T000000Z"/>
+                        </C:comp-filter>
+                    </C:comp-filter>
+                </C:filter>
+            </C:calendar-query>
+            """
+        _, responses = self.report("/calendar.ics/", req_body_without_expand)
+        assert len(responses) == 1
+
+        response = responses['/calendar.ics/event_daily_rrule.ics']
+        status, element = list(response.values())[0]
+
+        assert status == 200
+
+        assert "RRULE" in element.text
+        assert "BEGIN:VTIMEZONE" in element.text
+        assert "RECURRENCE-ID" not in element.text
+
+        uids = []
+        for line in element.text.split("\n"):
+            if line.startswith("UID:"):
+                uid = line[len("UID:"):]
+                assert uid == "event_daily_rrule"
+                uids.append(uids)
+
+        assert len(uids) == 1
+
+        req_body_with_expand = \
+            """<?xml version="1.0" encoding="utf-8" ?>
+            <C:calendar-query xmlns:D="DAV:" xmlns:C="urn:ietf:params:xml:ns:caldav">
+                <D:prop>
+                    <C:calendar-data>
+                        <C:expand start="20060103T000000Z" end="20060105T000000Z"/>
+                    </C:calendar-data>
+                </D:prop>
+                <C:filter>
+                    <C:comp-filter name="VCALENDAR">
+                        <C:comp-filter name="VEVENT">
+                            <C:time-range start="20060103T000000Z" end="20060105T000000Z"/>
+                        </C:comp-filter>
+                    </C:comp-filter>
+                </C:filter>
+            </C:calendar-query>
+            """
+
+        _, responses = self.report("/calendar.ics/", req_body_with_expand)
+
+        assert len(responses) == 1
+
+        response = responses['/calendar.ics/event_daily_rrule.ics']
+        status, element = list(response.values())[0]
+
+        assert status == 200
+        assert "RRULE" not in element.text
+        assert "BEGIN:VTIMEZONE" not in element.text
+
+        uids = []
+        recurrence_ids = []
+        for line in element.text.split("\n"):
+            if line.startswith("UID:"):
+                assert line == "UID:event_daily_rrule"
+                uids.append(uids)
+
+            if line.startswith("RECURRENCE-ID:"):
+                recurrence_ids.append(line)
+
+        assert len(uids) == 2
+        assert len(set(recurrence_ids)) == 2
+
     def test_propfind_sync_token(self) -> None:
         """Retrieve the sync-token with a propfind request"""
         calendar_path = "/calendar.ics/"