1598448993

Build Type

push

github

Committed by Unrud

Commit Message

Change type of stack_frame to Optional[...]

Run Details

1443 of 1821 branches covered (79.24%)

Branch coverage included in aggregate %.

1 of 1 new or added line in 1 file covered. (100.0%)

24 existing lines in 2 files now uncovered.

3316 of 3876 relevant lines covered (85.55%)

12.8 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

84.81

/radicale/item/filter.py

# This file is part of Radicale - CalDAV and CardDAV server
# Copyright © 2008 Nicolas Kandel
# Copyright © 2008 Pascal Halter
# Copyright © 2008-2015 Guillaume Ayoub
# Copyright © 2017-2018 Unrud <unrud@outlook.com>
#
# This library is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Radicale.  If not, see <http://www.gnu.org/licenses/>.


import math
import xml.etree.ElementTree as ET
from datetime import date, datetime, timedelta, timezone
from itertools import chain
from typing import (Callable, Iterable, Iterator, List, Optional, Sequence,
                    Tuple)

import vobject

from radicale import item, xmlutils
from radicale.log import logger

DAY: timedelta = timedelta(days=1)
SECOND: timedelta = timedelta(seconds=1)
DATETIME_MIN: datetime = datetime.min.replace(tzinfo=timezone.utc)
DATETIME_MAX: datetime = datetime.max.replace(tzinfo=timezone.utc)
TIMESTAMP_MIN: int = math.floor(DATETIME_MIN.timestamp())
TIMESTAMP_MAX: int = math.ceil(DATETIME_MAX.timestamp())


def date_to_datetime(d: date) -> datetime:
    """Transform any date to a UTC datetime.

    If ``d`` is a datetime without timezone, return as UTC datetime. If ``d``
    is already a datetime with timezone, return as is.

    """
    if not isinstance(d, datetime):
        d = datetime.combine(d, datetime.min.time())
    if not d.tzinfo:
        d = d.replace(tzinfo=timezone.utc)
    return d


def comp_match(item: "item.Item", filter_: ET.Element, level: int = 0) -> bool:
    """Check whether the ``item`` matches the comp ``filter_``.

    If ``level`` is ``0``, the filter is applied on the
    item's collection. Otherwise, it's applied on the item.

    See rfc4791-9.7.1.

    """

    # TODO: Filtering VALARM and VFREEBUSY is not implemented
    # HACK: the filters are tested separately against all components

    if level == 0:
        tag = item.name
    elif level == 1:
        tag = item.component_name
    else:
        logger.warning(
            "Filters with three levels of comp-filter are not supported")
        return True
    if not tag:
        return False
    name = filter_.get("name", "").upper()
    if len(filter_) == 0:
        # Point #1 of rfc4791-9.7.1
        return name == tag
    if len(filter_) == 1:
        if filter_[0].tag == xmlutils.make_clark("C:is-not-defined"):
            # Point #2 of rfc4791-9.7.1
            return name != tag
    if name != tag:
        return False
    if (level == 0 and name != "VCALENDAR" or
            level == 1 and name not in ("VTODO", "VEVENT", "VJOURNAL")):
        logger.warning("Filtering %s is not supported", name)
        return True
    # Point #3 and #4 of rfc4791-9.7.1
    components = ([item.vobject_item] if level == 0
                  else list(getattr(item.vobject_item,
                                    "%s_list" % tag.lower())))
    for child in filter_:
        if child.tag == xmlutils.make_clark("C:prop-filter"):
            if not any(prop_match(comp, child, "C")
                       for comp in components):
                return False
        elif child.tag == xmlutils.make_clark("C:time-range"):
            if not time_range_match(item.vobject_item, filter_[0], tag):
                return False
        elif child.tag == xmlutils.make_clark("C:comp-filter"):
            if not comp_match(item, child, level=level + 1):
                return False
        else:
            raise ValueError("Unexpected %r in comp-filter" % child.tag)
    return True


def prop_match(vobject_item: vobject.base.Component,
               filter_: ET.Element, ns: str) -> bool:
    """Check whether the ``item`` matches the prop ``filter_``.

    See rfc4791-9.7.2 and rfc6352-10.5.1.

    """
    name = filter_.get("name", "").lower()
    if len(filter_) == 0:
        # Point #1 of rfc4791-9.7.2
        return name in vobject_item.contents
    if len(filter_) == 1:
        if filter_[0].tag == xmlutils.make_clark("%s:is-not-defined" % ns):
            # Point #2 of rfc4791-9.7.2
            return name not in vobject_item.contents
    if name not in vobject_item.contents:
        return False
    # Point #3 and #4 of rfc4791-9.7.2
    for child in filter_:
        if ns == "C" and child.tag == xmlutils.make_clark("C:time-range"):
            if not time_range_match(vobject_item, child, name):
                return False
        elif child.tag == xmlutils.make_clark("%s:text-match" % ns):
            if not text_match(vobject_item, child, name, ns):
                return False
        elif child.tag == xmlutils.make_clark("%s:param-filter" % ns):
            if not param_filter_match(vobject_item, child, name, ns):
                return False
        else:
            raise ValueError("Unexpected %r in prop-filter" % child.tag)
    return True


def time_range_match(vobject_item: vobject.base.Component,
                     filter_: ET.Element, child_name: str) -> bool:
    """Check whether the component/property ``child_name`` of
       ``vobject_item`` matches the time-range ``filter_``."""

    start_text = filter_.get("start")
    end_text = filter_.get("end")
    if not start_text and not end_text:
        return False
    if start_text:
        start = datetime.strptime(start_text, "%Y%m%dT%H%M%SZ")
    else:
        start = datetime.min
    if end_text:
        end = datetime.strptime(end_text, "%Y%m%dT%H%M%SZ")
    else:
        end = datetime.max
    start = start.replace(tzinfo=timezone.utc)
    end = end.replace(tzinfo=timezone.utc)

    matched = False

    def range_fn(range_start: datetime, range_end: datetime,
                 is_recurrence: bool) -> bool:
        nonlocal matched
        if start < range_end and range_start < end:
            matched = True
            return True
        if end < range_start and not is_recurrence:
            return True
        return False

    def infinity_fn(start: datetime) -> bool:
        return False

    visit_time_ranges(vobject_item, child_name, range_fn, infinity_fn)
    return matched


def visit_time_ranges(vobject_item: vobject.base.Component, child_name: str,
                      range_fn: Callable[[datetime, datetime, bool], bool],
                      infinity_fn: Callable[[datetime], bool]) -> None:
    """Visit all time ranges in the component/property ``child_name`` of
    `vobject_item`` with visitors ``range_fn`` and ``infinity_fn``.

    ``range_fn`` gets called for every time_range with ``start`` and ``end``
    datetimes and ``is_recurrence`` as arguments. If the function returns True,
    the operation is cancelled.

    ``infinity_fn`` gets called when an infinite recurrence rule is detected
    with ``start`` datetime as argument. If the function returns True, the
    operation is cancelled.

    See rfc4791-9.9.

    """

    # HACK: According to rfc5545-3.8.4.4 an recurrance that is resheduled
    # with Recurrence ID affects the recurrence itself and all following
    # recurrences too. This is not respected and client don't seem to bother
    # either.

    def getrruleset(child: vobject.base.Component, ignore: Sequence[date]
                    ) -> Tuple[Iterable[date], bool]:
        infinite = False
        for rrule in child.contents.get("rrule", []):
            if (";UNTIL=" not in rrule.value.upper() and
                    ";COUNT=" not in rrule.value.upper()):
                infinite = True
                break
        if infinite:
            for dtstart in child.getrruleset(addRDate=True):
                if dtstart in ignore:
                    continue
                if infinity_fn(date_to_datetime(dtstart)):
                    return (), True
                break
        return filter(lambda dtstart: dtstart not in ignore,
                      child.getrruleset(addRDate=True)), False

    def get_children(components: Iterable[vobject.base.Component]) -> Iterator[
                         Tuple[vobject.base.Component, bool, List[date]]]:
        main = None
        recurrences = []
        for comp in components:
            if hasattr(comp, "recurrence_id") and comp.recurrence_id.value:
                recurrences.append(comp.recurrence_id.value)
                if comp.rruleset:
                    # Prevent possible infinite loop
                    raise ValueError("Overwritten recurrence with RRULESET")
                yield comp, True, []
            else:
                if main is not None:
                    raise ValueError("Multiple main components")
                main = comp
        if main is None:
            raise ValueError("Main component missing")
        yield main, False, recurrences

    # Comments give the lines in the tables of the specification
    if child_name == "VEVENT":
        for child, is_recurrence, recurrences in get_children(
                vobject_item.vevent_list):
            # TODO: check if there's a timezone
            dtstart = child.dtstart.value

            if child.rruleset:
                dtstarts, infinity = getrruleset(child, recurrences)
                if infinity:
                    return
            else:
                dtstarts = (dtstart,)

            dtend = getattr(child, "dtend", None)
            if dtend is not None:
                dtend = dtend.value
                original_duration = (dtend - dtstart).total_seconds()
                dtend = date_to_datetime(dtend)

            duration = getattr(child, "duration", None)
            if duration is not None:
                original_duration = duration = duration.value

            for dtstart in dtstarts:
                dtstart_is_datetime = isinstance(dtstart, datetime)
                dtstart = date_to_datetime(dtstart)

                if dtend is not None:
                    # Line 1
                    dtend = dtstart + timedelta(seconds=original_duration)
                    if range_fn(dtstart, dtend, is_recurrence):
                        return
                elif duration is not None:
                    if original_duration is None:
                        original_duration = duration.seconds
                    if duration.seconds > 0:
                        # Line 2
                        if range_fn(dtstart, dtstart + duration,
                                    is_recurrence):
                            return
                    else:
                        # Line 3
                        if range_fn(dtstart, dtstart + SECOND, is_recurrence):
                            return
                elif dtstart_is_datetime:
                    # Line 4
                    if range_fn(dtstart, dtstart + SECOND, is_recurrence):
                        return
                else:
                    # Line 5
                    if range_fn(dtstart, dtstart + DAY, is_recurrence):
                        return

    elif child_name == "VTODO":
        for child, is_recurrence, recurrences in get_children(
                vobject_item.vtodo_list):
            dtstart = getattr(child, "dtstart", None)
            duration = getattr(child, "duration", None)
            due = getattr(child, "due", None)
            completed = getattr(child, "completed", None)
            created = getattr(child, "created", None)

            if dtstart is not None:
                dtstart = date_to_datetime(dtstart.value)
            if duration is not None:
                duration = duration.value
            if due is not None:
                due = date_to_datetime(due.value)
                if dtstart is not None:
                    original_duration = (due - dtstart).total_seconds()
            if completed is not None:
                completed = date_to_datetime(completed.value)
                if created is not None:
                    created = date_to_datetime(created.value)
                    original_duration = (completed - created).total_seconds()
            elif created is not None:
                created = date_to_datetime(created.value)

            if child.rruleset:
                reference_dates, infinity = getrruleset(child, recurrences)
                if infinity:
                    return
            else:
                if dtstart is not None:
                    reference_dates = (dtstart,)
                elif due is not None:
                    reference_dates = (due,)
                elif completed is not None:
                    reference_dates = (completed,)
                elif created is not None:
                    reference_dates = (created,)
                else:
                    # Line 8
                    if range_fn(DATETIME_MIN, DATETIME_MAX, is_recurrence):
                        return
                    reference_dates = ()

            for reference_date in reference_dates:
                reference_date = date_to_datetime(reference_date)

                if dtstart is not None and duration is not None:
                    # Line 1
                    if range_fn(reference_date,
                                reference_date + duration + SECOND,
                                is_recurrence):
                        return
                    if range_fn(reference_date + duration - SECOND,
                                reference_date + duration + SECOND,
                                is_recurrence):
                        return
                elif dtstart is not None and due is not None:
                    # Line 2
                    due = reference_date + timedelta(seconds=original_duration)
                    if (range_fn(reference_date, due, is_recurrence) or
                            range_fn(reference_date,
                                     reference_date + SECOND, is_recurrence) or
                            range_fn(due - SECOND, due, is_recurrence) or
                            range_fn(due - SECOND, reference_date + SECOND,
                                     is_recurrence)):
                        return
                elif dtstart is not None:
                    if range_fn(reference_date, reference_date + SECOND,
                                is_recurrence):
                        return
                elif due is not None:
                    # Line 4
                    if range_fn(reference_date - SECOND, reference_date,
                                is_recurrence):
                        return
                elif completed is not None and created is not None:
                    # Line 5
                    completed = reference_date + timedelta(
                        seconds=original_duration)
                    if (range_fn(reference_date - SECOND,
                                 reference_date + SECOND,
                                 is_recurrence) or
                            range_fn(completed - SECOND, completed + SECOND,
                                     is_recurrence) or
                            range_fn(reference_date - SECOND,
                                     reference_date + SECOND, is_recurrence) or
                            range_fn(completed - SECOND, completed + SECOND,
                                     is_recurrence)):
                        return
                elif completed is not None:
                    # Line 6
                    if range_fn(reference_date - SECOND,
                                reference_date + SECOND, is_recurrence):
                        return
                elif created is not None:
                    # Line 7
                    if range_fn(reference_date, DATETIME_MAX, is_recurrence):
                        return

    elif child_name == "VJOURNAL":
        for child, is_recurrence, recurrences in get_children(
                vobject_item.vjournal_list):
            dtstart = getattr(child, "dtstart", None)

            if dtstart is not None:
                dtstart = dtstart.value
                if child.rruleset:
                    dtstarts, infinity = getrruleset(child, recurrences)
                    if infinity:
                        return
                else:
                    dtstarts = (dtstart,)

                for dtstart in dtstarts:
                    dtstart_is_datetime = isinstance(dtstart, datetime)
                    dtstart = date_to_datetime(dtstart)

                    if dtstart_is_datetime:
                        # Line 1
                        if range_fn(dtstart, dtstart + SECOND, is_recurrence):
                            return
                    else:
                        # Line 2
                        if range_fn(dtstart, dtstart + DAY, is_recurrence):
                            return

    else:
        # Match a property
        child = getattr(vobject_item, child_name.lower())
        if isinstance(child, date):
            child_is_datetime = isinstance(child, datetime)
            child = date_to_datetime(child)
            if child_is_datetime:
                range_fn(child, child + SECOND, False)
            else:
                range_fn(child, child + DAY, False)


def text_match(vobject_item: vobject.base.Component,
               filter_: ET.Element, child_name: str, ns: str,
               attrib_name: Optional[str] = None) -> bool:
    """Check whether the ``item`` matches the text-match ``filter_``.

    See rfc4791-9.7.5.

    """
    # TODO: collations are not supported, but the default ones needed
    # for DAV servers are actually pretty useless. Texts are lowered to
    # be case-insensitive, almost as the "i;ascii-casemap" value.
    text = next(filter_.itertext()).lower()
    match_type = "contains"
    if ns == "CR":
        match_type = filter_.get("match-type", match_type)

    def match(value: str) -> bool:
        value = value.lower()
        if match_type == "equals":
            return value == text
        if match_type == "contains":
            return text in value
        if match_type == "starts-with":
            return value.startswith(text)
        if match_type == "ends-with":
            return value.endswith(text)
        raise ValueError("Unexpected text-match match-type: %r" % match_type)

    children = getattr(vobject_item, "%s_list" % child_name, [])
    if attrib_name is not None:
        condition = any(
            match(attrib) for child in children
            for attrib in child.params.get(attrib_name, []))
    else:
        condition = any(match(child.value) for child in children)
    if filter_.get("negate-condition") == "yes":
        return not condition
    return condition


def param_filter_match(vobject_item: vobject.base.Component,
                       filter_: ET.Element, parent_name: str, ns: str) -> bool:
    """Check whether the ``item`` matches the param-filter ``filter_``.

    See rfc4791-9.7.3.

    """
    name = filter_.get("name", "").upper()
    children = getattr(vobject_item, "%s_list" % parent_name, [])
    condition = any(name in child.params for child in children)
    if len(filter_) > 0:
        if filter_[0].tag == xmlutils.make_clark("%s:text-match" % ns):
            return condition and text_match(
                vobject_item, filter_[0], parent_name, ns, name)
        if filter_[0].tag == xmlutils.make_clark("%s:is-not-defined" % ns):
            return not condition
    return condition


def simplify_prefilters(filters: Iterable[ET.Element], collection_tag: str
                        ) -> Tuple[Optional[str], int, int, bool]:
    """Creates a simplified condition from ``filters``.

    Returns a tuple (``tag``, ``start``, ``end``, ``simple``) where ``tag`` is
    a string or None (match all) and ``start`` and ``end`` are POSIX
    timestamps (as int). ``simple`` is a bool that indicates that ``filters``
    and the simplified condition are identical.

    """
    flat_filters = list(chain.from_iterable(filters))
    simple = len(flat_filters) <= 1
    for col_filter in flat_filters:
        if collection_tag != "VCALENDAR":
            simple = False
            break
        if (col_filter.tag != xmlutils.make_clark("C:comp-filter") or
                col_filter.get("name", "").upper() != "VCALENDAR"):
            simple = False
            continue
        simple &= len(col_filter) <= 1
        for comp_filter in col_filter:
            if comp_filter.tag != xmlutils.make_clark("C:comp-filter"):
                simple = False
                continue
            tag = comp_filter.get("name", "").upper()
            if comp_filter.find(
                    xmlutils.make_clark("C:is-not-defined")) is not None:
                simple = False
                continue
            simple &= len(comp_filter) <= 1
            for time_filter in comp_filter:
                if tag not in ("VTODO", "VEVENT", "VJOURNAL"):
                    simple = False
                    break
                if time_filter.tag != xmlutils.make_clark("C:time-range"):
                    simple = False
                    continue
                start_text = time_filter.get("start")
                end_text = time_filter.get("end")
                if start_text:
                    start = math.floor(datetime.strptime(
                        start_text, "%Y%m%dT%H%M%SZ").replace(
                            tzinfo=timezone.utc).timestamp())
                else:
                    start = TIMESTAMP_MIN
                if end_text:
                    end = math.ceil(datetime.strptime(
                        end_text, "%Y%m%dT%H%M%SZ").replace(
                            tzinfo=timezone.utc).timestamp())
                else:
                    end = TIMESTAMP_MAX
                return tag, start, end, simple
            return tag, TIMESTAMP_MIN, TIMESTAMP_MAX, simple
    return None, TIMESTAMP_MIN, TIMESTAMP_MAX, simple

1	# This file is part of Radicale - CalDAV and CardDAV server
2	# Copyright © 2008 Nicolas Kandel
3	# Copyright © 2008 Pascal Halter
4	# Copyright © 2008-2015 Guillaume Ayoub
5	# Copyright © 2017-2018 Unrud <unrud@outlook.com>
6	#
7	# This library is free software: you can redistribute it and/or modify
8	# it under the terms of the GNU General Public License as published by
9	# the Free Software Foundation, either version 3 of the License, or
10	# (at your option) any later version.
11	#
12	# This library is distributed in the hope that it will be useful,
13	# but WITHOUT ANY WARRANTY; without even the implied warranty of
14	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15	# GNU General Public License for more details.
16	#
17	# You should have received a copy of the GNU General Public License
18	# along with Radicale. If not, see <http://www.gnu.org/licenses/>.
19
20
21	import math	15✔
22	import xml.etree.ElementTree as ET	15✔
23	from datetime import date, datetime, timedelta, timezone	15✔
24	from itertools import chain	15✔
25	from typing import (Callable, Iterable, Iterator, List, Optional, Sequence,	15✔
26	Tuple)
27
28	import vobject	15✔
29
30	from radicale import item, xmlutils	15✔
31	from radicale.log import logger	15✔
32
33	DAY: timedelta = timedelta(days=1)	15✔
34	SECOND: timedelta = timedelta(seconds=1)	15✔
35	DATETIME_MIN: datetime = datetime.min.replace(tzinfo=timezone.utc)	15✔
36	DATETIME_MAX: datetime = datetime.max.replace(tzinfo=timezone.utc)	15✔
37	TIMESTAMP_MIN: int = math.floor(DATETIME_MIN.timestamp())	15✔
38	TIMESTAMP_MAX: int = math.ceil(DATETIME_MAX.timestamp())	15✔
39
40
41	def date_to_datetime(d: date) -> datetime:	15✔
42	"""Transform any date to a UTC datetime.
43
44	If ``d`` is a datetime without timezone, return as UTC datetime. If ``d``
45	is already a datetime with timezone, return as is.
46
47	"""
48	if not isinstance(d, datetime):	15✔
49	d = datetime.combine(d, datetime.min.time())	15✔
50	if not d.tzinfo:	15✔
51	d = d.replace(tzinfo=timezone.utc)	15✔
52	return d	15✔
53
54
55	def comp_match(item: "item.Item", filter_: ET.Element, level: int = 0) -> bool:	15✔
56	"""Check whether the ``item`` matches the comp ``filter_``.
57
58	If ``level`` is ``0``, the filter is applied on the
59	item's collection. Otherwise, it's applied on the item.
60
61	See rfc4791-9.7.1.
62
63	"""
64
65	# TODO: Filtering VALARM and VFREEBUSY is not implemented
66	# HACK: the filters are tested separately against all components
67
68	if level == 0:	15✔
69	tag = item.name	15✔
70	elif level == 1:	15!
71	tag = item.component_name	15✔
72	else:
73	logger.warning(	×
74	"Filters with three levels of comp-filter are not supported")
75	return True	×
76	if not tag:	15!
77	return False	×
78	name = filter_.get("name", "").upper()	15✔
79	if len(filter_) == 0:	15!
80	# Point #1 of rfc4791-9.7.1
81	return name == tag	×
82	if len(filter_) == 1:	15✔
83	if filter_[0].tag == xmlutils.make_clark("C:is-not-defined"):	15✔
84	# Point #2 of rfc4791-9.7.1
85	return name != tag	15✔
86	if name != tag:	15!
87	return False	×
88	if (level == 0 and name != "VCALENDAR" or	15!
89	level == 1 and name not in ("VTODO", "VEVENT", "VJOURNAL")):
90	logger.warning("Filtering %s is not supported", name)	×
91	return True	×
92	# Point #3 and #4 of rfc4791-9.7.1
93	components = ([item.vobject_item] if level == 0	15✔
94	else list(getattr(item.vobject_item,
95	"%s_list" % tag.lower())))
96	for child in filter_:	15✔
97	if child.tag == xmlutils.make_clark("C:prop-filter"):	15✔
98	if not any(prop_match(comp, child, "C")	15✔
99	for comp in components):
100	return False	15✔
101	elif child.tag == xmlutils.make_clark("C:time-range"):	15✔
102	if not time_range_match(item.vobject_item, filter_[0], tag):	15✔
103	return False	15✔
104	elif child.tag == xmlutils.make_clark("C:comp-filter"):	15!
105	if not comp_match(item, child, level=level + 1):	15✔
106	return False	15✔
107	else:
108	raise ValueError("Unexpected %r in comp-filter" % child.tag)	×
109	return True	15✔
110
111
112	def prop_match(vobject_item: vobject.base.Component,	15✔
113	filter_: ET.Element, ns: str) -> bool:
114	"""Check whether the ``item`` matches the prop ``filter_``.
115
116	See rfc4791-9.7.2 and rfc6352-10.5.1.
117
118	"""
119	name = filter_.get("name", "").lower()	15✔
120	if len(filter_) == 0:	15✔
121	# Point #1 of rfc4791-9.7.2
122	return name in vobject_item.contents	15✔
123	if len(filter_) == 1:	15!
124	if filter_[0].tag == xmlutils.make_clark("%s:is-not-defined" % ns):	15✔
125	# Point #2 of rfc4791-9.7.2
126	return name not in vobject_item.contents	15✔
127	if name not in vobject_item.contents:	15✔
128	return False	15✔
129	# Point #3 and #4 of rfc4791-9.7.2
130	for child in filter_:	15✔
131	if ns == "C" and child.tag == xmlutils.make_clark("C:time-range"):	15!
132	if not time_range_match(vobject_item, child, name):	×
133	return False	×
134	elif child.tag == xmlutils.make_clark("%s:text-match" % ns):	15✔
135	if not text_match(vobject_item, child, name, ns):	15✔
136	return False	15✔
137	elif child.tag == xmlutils.make_clark("%s:param-filter" % ns):	15!
138	if not param_filter_match(vobject_item, child, name, ns):	15✔
139	return False	15✔
140	else:
141	raise ValueError("Unexpected %r in prop-filter" % child.tag)	×
142	return True	15✔
143
144
145	def time_range_match(vobject_item: vobject.base.Component,	15✔
146	filter_: ET.Element, child_name: str) -> bool:
147	"""Check whether the component/property ``child_name`` of
148	``vobject_item`` matches the time-range ``filter_``."""
149
150	start_text = filter_.get("start")	15✔
151	end_text = filter_.get("end")	15✔
152	if not start_text and not end_text:	15!
153	return False	×
154	if start_text:	15!
155	start = datetime.strptime(start_text, "%Y%m%dT%H%M%SZ")	15✔
156	else:
157	start = datetime.min	×
158	if end_text:	15!
159	end = datetime.strptime(end_text, "%Y%m%dT%H%M%SZ")	15✔
160	else:
161	end = datetime.max	×
162	start = start.replace(tzinfo=timezone.utc)	15✔
163	end = end.replace(tzinfo=timezone.utc)	15✔
164
165	matched = False	15✔
166
167	def range_fn(range_start: datetime, range_end: datetime,	15✔
168	is_recurrence: bool) -> bool:
169	nonlocal matched
170	if start < range_end and range_start < end:	15✔
171	matched = True	15✔
172	return True	15✔
173	if end < range_start and not is_recurrence:	15✔
174	return True	15✔
175	return False	15✔
176
177	def infinity_fn(start: datetime) -> bool:	15✔
178	return False	15✔
179
180	visit_time_ranges(vobject_item, child_name, range_fn, infinity_fn)	15✔
181	return matched	15✔
182
183
184	def visit_time_ranges(vobject_item: vobject.base.Component, child_name: str,	15✔
185	range_fn: Callable[[datetime, datetime, bool], bool],
186	infinity_fn: Callable[[datetime], bool]) -> None:
187	"""Visit all time ranges in the component/property ``child_name`` of
188	`vobject_item`` with visitors ``range_fn`` and ``infinity_fn``.
189
190	``range_fn`` gets called for every time_range with ``start`` and ``end``
191	datetimes and ``is_recurrence`` as arguments. If the function returns True,
192	the operation is cancelled.
193
194	``infinity_fn`` gets called when an infinite recurrence rule is detected
195	with ``start`` datetime as argument. If the function returns True, the
196	operation is cancelled.
197
198	See rfc4791-9.9.
199
200	"""
201
202	# HACK: According to rfc5545-3.8.4.4 an recurrance that is resheduled
203	# with Recurrence ID affects the recurrence itself and all following
204	# recurrences too. This is not respected and client don't seem to bother
205	# either.
206
207	def getrruleset(child: vobject.base.Component, ignore: Sequence[date]	15✔
208	) -> Tuple[Iterable[date], bool]:
209	infinite = False	15✔
210	for rrule in child.contents.get("rrule", []):	15✔
211	if (";UNTIL=" not in rrule.value.upper() and	15✔
212	";COUNT=" not in rrule.value.upper()):
213	infinite = True	15✔
214	break	15✔
215	if infinite:	15✔
216	for dtstart in child.getrruleset(addRDate=True):	15!
217	if dtstart in ignore:	15!
UNCOV 218	continue	×
219	if infinity_fn(date_to_datetime(dtstart)):	15✔
220	return (), True	15✔
221	break	15✔
222	return filter(lambda dtstart: dtstart not in ignore,	15✔
223	child.getrruleset(addRDate=True)), False
224
225	def get_children(components: Iterable[vobject.base.Component]) -> Iterator[	15✔
226	Tuple[vobject.base.Component, bool, List[date]]]:
227	main = None	15✔
228	recurrences = []	15✔
229	for comp in components:	15✔
230	if hasattr(comp, "recurrence_id") and comp.recurrence_id.value:	15✔
231	recurrences.append(comp.recurrence_id.value)	15✔
232	if comp.rruleset:	15!
233	# Prevent possible infinite loop
234	raise ValueError("Overwritten recurrence with RRULESET")	×
235	yield comp, True, []	15✔
236	else:
237	if main is not None:	15!
UNCOV 238	raise ValueError("Multiple main components")	×
239	main = comp	15✔
240	if main is None:	15!
UNCOV 241	raise ValueError("Main component missing")	×
242	yield main, False, recurrences	15✔
243
244	# Comments give the lines in the tables of the specification
245	if child_name == "VEVENT":	15✔
246	for child, is_recurrence, recurrences in get_children(	15✔
247	vobject_item.vevent_list):
248	# TODO: check if there's a timezone
249	dtstart = child.dtstart.value	15✔
250
251	if child.rruleset:	15✔
252	dtstarts, infinity = getrruleset(child, recurrences)	15✔
253	if infinity:	15✔
254	return	15✔
255	else:
256	dtstarts = (dtstart,)	15✔
257
258	dtend = getattr(child, "dtend", None)	15✔
259	if dtend is not None:	15✔
260	dtend = dtend.value	15✔
261	original_duration = (dtend - dtstart).total_seconds()	15✔
262	dtend = date_to_datetime(dtend)	15✔
263
264	duration = getattr(child, "duration", None)	15✔
265	if duration is not None:	15✔
266	original_duration = duration = duration.value	15✔
267
268	for dtstart in dtstarts:	15✔
269	dtstart_is_datetime = isinstance(dtstart, datetime)	15✔
270	dtstart = date_to_datetime(dtstart)	15✔
271
272	if dtend is not None:	15✔
273	# Line 1
274	dtend = dtstart + timedelta(seconds=original_duration)	15✔
275	if range_fn(dtstart, dtend, is_recurrence):	15✔
276	return	15✔
277	elif duration is not None:	15✔
278	if original_duration is None:	15!
UNCOV 279	original_duration = duration.seconds	×
280	if duration.seconds > 0:	15!
281	# Line 2
282	if range_fn(dtstart, dtstart + duration,	15!
283	is_recurrence):
284	return	×
285	else:
286	# Line 3
UNCOV 287	if range_fn(dtstart, dtstart + SECOND, is_recurrence):	×
288	return	×
289	elif dtstart_is_datetime:	15✔
290	# Line 4
291	if range_fn(dtstart, dtstart + SECOND, is_recurrence):	15!
292	return	×
293	else:
294	# Line 5
295	if range_fn(dtstart, dtstart + DAY, is_recurrence):	15!
UNCOV 296	return	×
297
298	elif child_name == "VTODO":	15✔
299	for child, is_recurrence, recurrences in get_children(	15✔
300	vobject_item.vtodo_list):
301	dtstart = getattr(child, "dtstart", None)	15✔
302	duration = getattr(child, "duration", None)	15✔
303	due = getattr(child, "due", None)	15✔
304	completed = getattr(child, "completed", None)	15✔
305	created = getattr(child, "created", None)	15✔
306
307	if dtstart is not None:	15✔
308	dtstart = date_to_datetime(dtstart.value)	15✔
309	if duration is not None:	15✔
310	duration = duration.value	15✔
311	if due is not None:	15✔
312	due = date_to_datetime(due.value)	15✔
313	if dtstart is not None:	15✔
314	original_duration = (due - dtstart).total_seconds()	15✔
315	if completed is not None:	15✔
316	completed = date_to_datetime(completed.value)	15✔
317	if created is not None:	15✔
318	created = date_to_datetime(created.value)	15✔
319	original_duration = (completed - created).total_seconds()	15✔
320	elif created is not None:	15✔
321	created = date_to_datetime(created.value)	15✔
322
323	if child.rruleset:	15✔
324	reference_dates, infinity = getrruleset(child, recurrences)	15✔
325	if infinity:	15✔
326	return	15✔
327	else:
328	if dtstart is not None:	15✔
329	reference_dates = (dtstart,)	15✔
330	elif due is not None:	15✔
331	reference_dates = (due,)	15✔
332	elif completed is not None:	15✔
333	reference_dates = (completed,)	15✔
334	elif created is not None:	15✔
335	reference_dates = (created,)	15✔
336	else:
337	# Line 8
338	if range_fn(DATETIME_MIN, DATETIME_MAX, is_recurrence):	15✔
339	return	15✔
340	reference_dates = ()	15✔
341
342	for reference_date in reference_dates:	15✔
343	reference_date = date_to_datetime(reference_date)	15✔
344
345	if dtstart is not None and duration is not None:	15✔
346	# Line 1
347	if range_fn(reference_date,	15!
348	reference_date + duration + SECOND,
349	is_recurrence):
350	return	×
351	if range_fn(reference_date + duration - SECOND,	15!
352	reference_date + duration + SECOND,
353	is_recurrence):
UNCOV 354	return	×
355	elif dtstart is not None and due is not None:	15✔
356	# Line 2
357	due = reference_date + timedelta(seconds=original_duration)	15✔
358	if (range_fn(reference_date, due, is_recurrence) or	15✔
359	range_fn(reference_date,
360	reference_date + SECOND, is_recurrence) or
361	range_fn(due - SECOND, due, is_recurrence) or
362	range_fn(due - SECOND, reference_date + SECOND,
363	is_recurrence)):
364	return	15✔
365	elif dtstart is not None:	15✔
366	if range_fn(reference_date, reference_date + SECOND,	15!
367	is_recurrence):
UNCOV 368	return	×
369	elif due is not None:	15✔
370	# Line 4
371	if range_fn(reference_date - SECOND, reference_date,	15✔
372	is_recurrence):
373	return	15✔
374	elif completed is not None and created is not None:	15✔
375	# Line 5
376	completed = reference_date + timedelta(	15✔
377	seconds=original_duration)
378	if (range_fn(reference_date - SECOND,	15!
379	reference_date + SECOND,
380	is_recurrence) or
381	range_fn(completed - SECOND, completed + SECOND,
382	is_recurrence) or
383	range_fn(reference_date - SECOND,
384	reference_date + SECOND, is_recurrence) or
385	range_fn(completed - SECOND, completed + SECOND,
386	is_recurrence)):
UNCOV 387	return	×
388	elif completed is not None:	15✔
389	# Line 6
390	if range_fn(reference_date - SECOND,	15!
391	reference_date + SECOND, is_recurrence):
UNCOV 392	return	×
393	elif created is not None:	15!
394	# Line 7
395	if range_fn(reference_date, DATETIME_MAX, is_recurrence):	15✔
396	return	15✔
397
398	elif child_name == "VJOURNAL":	15!
399	for child, is_recurrence, recurrences in get_children(	15✔
400	vobject_item.vjournal_list):
401	dtstart = getattr(child, "dtstart", None)	15✔
402
403	if dtstart is not None:	15✔
404	dtstart = dtstart.value	15✔
405	if child.rruleset:	15✔
406	dtstarts, infinity = getrruleset(child, recurrences)	15✔
407	if infinity:	15✔
408	return	15✔
409	else:
410	dtstarts = (dtstart,)	15✔
411
412	for dtstart in dtstarts:	15✔
413	dtstart_is_datetime = isinstance(dtstart, datetime)	15✔
414	dtstart = date_to_datetime(dtstart)	15✔
415
416	if dtstart_is_datetime:	15✔
417	# Line 1
418	if range_fn(dtstart, dtstart + SECOND, is_recurrence):	15✔
419	return	15✔
420	else:
421	# Line 2
422	if range_fn(dtstart, dtstart + DAY, is_recurrence):	15!
423	return	×
424
425	else:
426	# Match a property
427	child = getattr(vobject_item, child_name.lower())	×
428	if isinstance(child, date):	×
UNCOV 429	child_is_datetime = isinstance(child, datetime)	×
430	child = date_to_datetime(child)	×
UNCOV 431	if child_is_datetime:	×
UNCOV 432	range_fn(child, child + SECOND, False)	×
433	else:
UNCOV 434	range_fn(child, child + DAY, False)	×
435
436
437	def text_match(vobject_item: vobject.base.Component,	15✔
438	filter_: ET.Element, child_name: str, ns: str,
439	attrib_name: Optional[str] = None) -> bool:
440	"""Check whether the ``item`` matches the text-match ``filter_``.
441
442	See rfc4791-9.7.5.
443
444	"""
445	# TODO: collations are not supported, but the default ones needed
446	# for DAV servers are actually pretty useless. Texts are lowered to
447	# be case-insensitive, almost as the "i;ascii-casemap" value.
448	text = next(filter_.itertext()).lower()	15✔
449	match_type = "contains"	15✔
450	if ns == "CR":	15✔
451	match_type = filter_.get("match-type", match_type)	15✔
452
453	def match(value: str) -> bool:	15✔
454	value = value.lower()	15✔
455	if match_type == "equals":	15✔
456	return value == text	15✔
457	if match_type == "contains":	15✔
458	return text in value	15✔
459	if match_type == "starts-with":	15✔
460	return value.startswith(text)	15✔
461	if match_type == "ends-with":	15!
462	return value.endswith(text)	15✔
UNCOV 463	raise ValueError("Unexpected text-match match-type: %r" % match_type)	×
464
465	children = getattr(vobject_item, "%s_list" % child_name, [])	15✔
466	if attrib_name is not None:	15✔
467	condition = any(	15✔
468	match(attrib) for child in children
469	for attrib in child.params.get(attrib_name, []))
470	else:
471	condition = any(match(child.value) for child in children)	15✔
472	if filter_.get("negate-condition") == "yes":	15✔
473	return not condition	15✔
474	return condition	15✔
475
476
477	def param_filter_match(vobject_item: vobject.base.Component,	15✔
478	filter_: ET.Element, parent_name: str, ns: str) -> bool:
479	"""Check whether the ``item`` matches the param-filter ``filter_``.
480
481	See rfc4791-9.7.3.
482
483	"""
484	name = filter_.get("name", "").upper()	15✔
485	children = getattr(vobject_item, "%s_list" % parent_name, [])	15✔
486	condition = any(name in child.params for child in children)	15✔
487	if len(filter_) > 0:	15!
488	if filter_[0].tag == xmlutils.make_clark("%s:text-match" % ns):	15✔
489	return condition and text_match(	15✔
490	vobject_item, filter_[0], parent_name, ns, name)
491	if filter_[0].tag == xmlutils.make_clark("%s:is-not-defined" % ns):	15!
492	return not condition	15✔
UNCOV 493	return condition	×
494
495
496	def simplify_prefilters(filters: Iterable[ET.Element], collection_tag: str	15✔
497	) -> Tuple[Optional[str], int, int, bool]:
498	"""Creates a simplified condition from ``filters``.
499
500	Returns a tuple (``tag``, ``start``, ``end``, ``simple``) where ``tag`` is
501	a string or None (match all) and ``start`` and ``end`` are POSIX
502	timestamps (as int). ``simple`` is a bool that indicates that ``filters``
503	and the simplified condition are identical.
504
505	"""
506	flat_filters = list(chain.from_iterable(filters))	15✔
507	simple = len(flat_filters) <= 1	15✔
508	for col_filter in flat_filters:	15✔
509	if collection_tag != "VCALENDAR":	15✔
510	simple = False	15✔
511	break	15✔
512	if (col_filter.tag != xmlutils.make_clark("C:comp-filter") or	15!
513	col_filter.get("name", "").upper() != "VCALENDAR"):
UNCOV 514	simple = False	×
515	continue	×
516	simple &= len(col_filter) <= 1	15✔
517	for comp_filter in col_filter:	15✔
518	if comp_filter.tag != xmlutils.make_clark("C:comp-filter"):	15!
UNCOV 519	simple = False	×
UNCOV 520	continue	×
521	tag = comp_filter.get("name", "").upper()	15✔
522	if comp_filter.find(	15✔
523	xmlutils.make_clark("C:is-not-defined")) is not None:
524	simple = False	15✔
525	continue	15✔
526	simple &= len(comp_filter) <= 1	15✔
527	for time_filter in comp_filter:	15✔
528	if tag not in ("VTODO", "VEVENT", "VJOURNAL"):	15!
UNCOV 529	simple = False	×
UNCOV 530	break	×
531	if time_filter.tag != xmlutils.make_clark("C:time-range"):	15✔
532	simple = False	15✔
533	continue	15✔
534	start_text = time_filter.get("start")	15✔
535	end_text = time_filter.get("end")	15✔
536	if start_text:	15!
537	start = math.floor(datetime.strptime(	15✔
538	start_text, "%Y%m%dT%H%M%SZ").replace(
539	tzinfo=timezone.utc).timestamp())
540	else:
UNCOV 541	start = TIMESTAMP_MIN	×
542	if end_text:	15✔
543	end = math.ceil(datetime.strptime(	15✔
544	end_text, "%Y%m%dT%H%M%SZ").replace(
545	tzinfo=timezone.utc).timestamp())
546	else:
547	end = TIMESTAMP_MAX	15✔
548	return tag, start, end, simple	15✔
549	return tag, TIMESTAMP_MIN, TIMESTAMP_MAX, simple	15✔
550	return None, TIMESTAMP_MIN, TIMESTAMP_MAX, simple	15✔

Kozea / Radicale / 1598448993

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous