1702737770

Build Type

push

github

Committed by Unrud

Commit Message

Workaround for broken contact PHOTO from InfCloud

Run Details

1452 of 1829 branches covered (79.39%)

Branch coverage included in aggregate %.

6 of 6 new or added lines in 3 files covered. (100.0%)

47 existing lines in 2 files now uncovered.

3336 of 3894 relevant lines covered (85.67%)

12.82 hits per line

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

81.65

/radicale/item/__init__.py

# This file is part of Radicale - CalDAV and CardDAV server
# Copyright © 2008 Nicolas Kandel
# Copyright © 2008 Pascal Halter
# Copyright © 2014 Jean-Marc Martins
# Copyright © 2008-2017 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/>.

"""
Module for address books and calendar entries (see ``Item``).

"""

import binascii
import contextlib
import math
import os
import re
import sys
from datetime import datetime, timedelta
from hashlib import sha256
from itertools import chain
from typing import (Any, Callable, List, MutableMapping, Optional, Sequence,
                    Tuple)

import vobject

from radicale import storage  # noqa:F401
from radicale import pathutils
from radicale.item import filter as radicale_filter
from radicale.log import logger


def read_components(s: str) -> List[vobject.base.Component]:
    """Wrapper for vobject.readComponents"""
    # Workaround for bug in InfCloud
    # PHOTO is a data URI
    s = re.sub(r"^(PHOTO(?:;[^:\r\n]*)?;ENCODING=b(?:;[^:\r\n]*)?:)"
               r"data:[^;,\r\n]*;base64,", r"\1", s,
               flags=re.MULTILINE | re.IGNORECASE)
    return list(vobject.readComponents(s))


def predict_tag_of_parent_collection(
        vobject_items: Sequence[vobject.base.Component]) -> Optional[str]:
    """Returns the predicted tag or `None`"""
    if len(vobject_items) != 1:
        return None
    if vobject_items[0].name == "VCALENDAR":
        return "VCALENDAR"
    if vobject_items[0].name in ("VCARD", "VLIST"):
        return "VADDRESSBOOK"
    return None


def predict_tag_of_whole_collection(
        vobject_items: Sequence[vobject.base.Component],
        fallback_tag: Optional[str] = None) -> Optional[str]:
    """Returns the predicted tag or `fallback_tag`"""
    if vobject_items and vobject_items[0].name == "VCALENDAR":
        return "VCALENDAR"
    if vobject_items and vobject_items[0].name in ("VCARD", "VLIST"):
        return "VADDRESSBOOK"
    if not fallback_tag and not vobject_items:
        # Maybe an empty address book
        return "VADDRESSBOOK"
    return fallback_tag


def check_and_sanitize_items(
        vobject_items: List[vobject.base.Component],
        is_collection: bool = False, tag: str = "") -> None:
    """Check vobject items for common errors and add missing UIDs.

    Modifies the list `vobject_items`.

    ``is_collection`` indicates that vobject_item contains unrelated
    components.

    The ``tag`` of the collection.

    """
    if tag and tag not in ("VCALENDAR", "VADDRESSBOOK"):
        raise ValueError("Unsupported collection tag: %r" % tag)
    if not is_collection and len(vobject_items) != 1:
        raise ValueError("Item contains %d components" % len(vobject_items))
    if tag == "VCALENDAR":
        if len(vobject_items) > 1:
            raise RuntimeError("VCALENDAR collection contains %d "
                               "components" % len(vobject_items))
        vobject_item = vobject_items[0]
        if vobject_item.name != "VCALENDAR":
            raise ValueError("Item type %r not supported in %r "
                             "collection" % (vobject_item.name, tag))
        component_uids = set()
        for component in vobject_item.components():
            if component.name in ("VTODO", "VEVENT", "VJOURNAL"):
                component_uid = get_uid(component)
                if component_uid:
                    component_uids.add(component_uid)
        component_name = None
        object_uid = None
        object_uid_set = False
        for component in vobject_item.components():
            # https://tools.ietf.org/html/rfc4791#section-4.1
            if component.name == "VTIMEZONE":
                continue
            if component_name is None or is_collection:
                component_name = component.name
            elif component_name != component.name:
                raise ValueError("Multiple component types in object: %r, %r" %
                                 (component_name, component.name))
            if component_name not in ("VTODO", "VEVENT", "VJOURNAL"):
                continue
            component_uid = get_uid(component)
            if not object_uid_set or is_collection:
                object_uid_set = True
                object_uid = component_uid
                if not component_uid:
                    if not is_collection:
                        raise ValueError("%s component without UID in object" %
                                         component_name)
                    component_uid = find_available_uid(
                        component_uids.__contains__)
                    component_uids.add(component_uid)
                    if hasattr(component, "uid"):
                        component.uid.value = component_uid
                    else:
                        component.add("UID").value = component_uid
            elif not object_uid or not component_uid:
                raise ValueError("Multiple %s components without UID in "
                                 "object" % component_name)
            elif object_uid != component_uid:
                raise ValueError(
                    "Multiple %s components with different UIDs in object: "
                    "%r, %r" % (component_name, object_uid, component_uid))
            # Workaround for bug in Lightning (Thunderbird)
            # Rescheduling a single occurrence from a repeating event creates
            # an event with DTEND and DURATION:PT0S
            if (hasattr(component, "dtend") and
                    hasattr(component, "duration") and
                    component.duration.value == timedelta(0)):
                logger.debug("Quirks: Removing zero duration from %s in "
                             "object %r", component_name, component_uid)
                del component.duration
            # Workaround for Evolution
            # EXDATE has value DATE even if DTSTART/DTEND is DATE-TIME.
            # The RFC is vaguely formulated on the issue.
            # To resolve the issue convert EXDATE and RDATE to
            # the same type as DTDSTART
            if hasattr(component, "dtstart"):
                ref_date = component.dtstart.value
                ref_value_param = component.dtstart.params.get("VALUE")
                for dates in chain(component.contents.get("exdate", []),
                                   component.contents.get("rdate", [])):
                    if all(type(d) == type(ref_date) for d in dates.value):
                        continue
                    for i, date in enumerate(dates.value):
                        dates.value[i] = ref_date.replace(
                            date.year, date.month, date.day)
                    with contextlib.suppress(KeyError):
                        del dates.params["VALUE"]
                    if ref_value_param is not None:
                        dates.params["VALUE"] = ref_value_param
            # vobject interprets recurrence rules on demand
            try:
                component.rruleset
            except Exception as e:
                raise ValueError("Invalid recurrence rules in %s in object %r"
                                 % (component.name, component_uid)) from e
    elif tag == "VADDRESSBOOK":
        # https://tools.ietf.org/html/rfc6352#section-5.1
        object_uids = set()
        for vobject_item in vobject_items:
            if vobject_item.name == "VCARD":
                object_uid = get_uid(vobject_item)
                if object_uid:
                    object_uids.add(object_uid)
        for vobject_item in vobject_items:
            if vobject_item.name == "VLIST":
                # Custom format used by SOGo Connector to store lists of
                # contacts
                continue
            if vobject_item.name != "VCARD":
                raise ValueError("Item type %r not supported in %r "
                                 "collection" % (vobject_item.name, tag))
            object_uid = get_uid(vobject_item)
            if not object_uid:
                if not is_collection:
                    raise ValueError("%s object without UID" %
                                     vobject_item.name)
                object_uid = find_available_uid(object_uids.__contains__)
                object_uids.add(object_uid)
                if hasattr(vobject_item, "uid"):
                    vobject_item.uid.value = object_uid
                else:
                    vobject_item.add("UID").value = object_uid
    else:
        for item in vobject_items:
            raise ValueError("Item type %r not supported in %s collection" %
                             (item.name, repr(tag) if tag else "generic"))


def check_and_sanitize_props(props: MutableMapping[Any, Any]
                             ) -> MutableMapping[str, str]:
    """Check collection properties for common errors.

    Modifies the dict `props`.

    """
    for k, v in list(props.items()):  # Make copy to be able to delete items
        if not isinstance(k, str):
            raise ValueError("Key must be %r not %r: %r" % (
                str.__name__, type(k).__name__, k))
        if not isinstance(v, str):
            if v is None:
                del props[k]
                continue
            raise ValueError("Value of %r must be %r not %r: %r" % (
                k, str.__name__, type(v).__name__, v))
        if k == "tag":
            if v not in ("", "VCALENDAR", "VADDRESSBOOK"):
                raise ValueError("Unsupported collection tag: %r" % v)
    return props


def find_available_uid(exists_fn: Callable[[str], bool], suffix: str = ""
                       ) -> str:
    """Generate a pseudo-random UID"""
    # Prevent infinite loop
    for _ in range(1000):
        r = binascii.hexlify(os.urandom(16)).decode("ascii")
        name = "%s-%s-%s-%s-%s%s" % (
            r[:8], r[8:12], r[12:16], r[16:20], r[20:], suffix)
        if not exists_fn(name):
            return name
    # something is wrong with the PRNG
    raise RuntimeError("No unique random sequence found")


def get_etag(text: str) -> str:
    """Etag from collection or item.

    Encoded as quoted-string (see RFC 2616).

    """
    etag = sha256()
    etag.update(text.encode())
    return '"%s"' % etag.hexdigest()


def get_uid(vobject_component: vobject.base.Component) -> str:
    """UID value of an item if defined."""
    return (vobject_component.uid.value or ""
            if hasattr(vobject_component, "uid") else "")


def get_uid_from_object(vobject_item: vobject.base.Component) -> str:
    """UID value of an calendar/addressbook object."""
    if vobject_item.name == "VCALENDAR":
        if hasattr(vobject_item, "vevent"):
            return get_uid(vobject_item.vevent)
        if hasattr(vobject_item, "vjournal"):
            return get_uid(vobject_item.vjournal)
        if hasattr(vobject_item, "vtodo"):
            return get_uid(vobject_item.vtodo)
    elif vobject_item.name == "VCARD":
        return get_uid(vobject_item)
    return ""


def find_tag(vobject_item: vobject.base.Component) -> str:
    """Find component name from ``vobject_item``."""
    if vobject_item.name == "VCALENDAR":
        for component in vobject_item.components():
            if component.name != "VTIMEZONE":
                return component.name or ""
    return ""


def find_time_range(vobject_item: vobject.base.Component, tag: str
                    ) -> Tuple[int, int]:
    """Find enclosing time range from ``vobject item``.

    ``tag`` must be set to the return value of ``find_tag``.

    Returns a tuple (``start``, ``end``) where ``start`` and ``end`` are
    POSIX timestamps.

    This is intened to be used for matching against simplified prefilters.

    """
    if not tag:
        return radicale_filter.TIMESTAMP_MIN, radicale_filter.TIMESTAMP_MAX
    start = end = None

    def range_fn(range_start: datetime, range_end: datetime,
                 is_recurrence: bool) -> bool:
        nonlocal start, end
        if start is None or range_start < start:
            start = range_start
        if end is None or end < range_end:
            end = range_end
        return False

    def infinity_fn(range_start: datetime) -> bool:
        nonlocal start, end
        if start is None or range_start < start:
            start = range_start
        end = radicale_filter.DATETIME_MAX
        return True

    radicale_filter.visit_time_ranges(vobject_item, tag, range_fn, infinity_fn)
    if start is None:
        start = radicale_filter.DATETIME_MIN
    if end is None:
        end = radicale_filter.DATETIME_MAX
    try:
        return math.floor(start.timestamp()), math.ceil(end.timestamp())
    except ValueError as e:
        if str(e) == ("offset must be a timedelta representing a whole "
                      "number of minutes") and sys.version_info < (3, 6):
            raise RuntimeError("Unsupported in Python < 3.6: %s" % e) from e
        raise


class Item:
    """Class for address book and calendar entries."""

    collection: Optional["storage.BaseCollection"]
    href: Optional[str]
    last_modified: Optional[str]

    _collection_path: str
    _text: Optional[str]
    _vobject_item: Optional[vobject.base.Component]
    _etag: Optional[str]
    _uid: Optional[str]
    _name: Optional[str]
    _component_name: Optional[str]
    _time_range: Optional[Tuple[int, int]]

    def __init__(self,
                 collection_path: Optional[str] = None,
                 collection: Optional["storage.BaseCollection"] = None,
                 vobject_item: Optional[vobject.base.Component] = None,
                 href: Optional[str] = None,
                 last_modified: Optional[str] = None,
                 text: Optional[str] = None,
                 etag: Optional[str] = None,
                 uid: Optional[str] = None,
                 name: Optional[str] = None,
                 component_name: Optional[str] = None,
                 time_range: Optional[Tuple[int, int]] = None):
        """Initialize an item.

        ``collection_path`` the path of the parent collection (optional if
        ``collection`` is set).

        ``collection`` the parent collection (optional).

        ``href`` the href of the item.

        ``last_modified`` the HTTP-datetime of when the item was modified.

        ``text`` the text representation of the item (optional if
        ``vobject_item`` is set).

        ``vobject_item`` the vobject item (optional if ``text`` is set).

        ``etag`` the etag of the item (optional). See ``get_etag``.

        ``uid`` the UID of the object (optional). See ``get_uid_from_object``.

        ``name`` the name of the item (optional). See ``vobject_item.name``.

        ``component_name`` the name of the primary component (optional).
        See ``find_tag``.

        ``time_range`` the enclosing time range. See ``find_time_range``.

        """
        if text is None and vobject_item is None:
            raise ValueError(
                "At least one of 'text' or 'vobject_item' must be set")
        if collection_path is None:
            if collection is None:
                raise ValueError("At least one of 'collection_path' or "
                                 "'collection' must be set")
            collection_path = collection.path
        assert collection_path == pathutils.strip_path(
            pathutils.sanitize_path(collection_path))
        self._collection_path = collection_path
        self.collection = collection
        self.href = href
        self.last_modified = last_modified
        self._text = text
        self._vobject_item = vobject_item
        self._etag = etag
        self._uid = uid
        self._name = name
        self._component_name = component_name
        self._time_range = time_range

    def serialize(self) -> str:
        if self._text is None:
            try:
                self._text = self.vobject_item.serialize()
            except Exception as e:
                raise RuntimeError("Failed to serialize item %r from %r: %s" %
                                   (self.href, self._collection_path,
                                    e)) from e
        return self._text

    @property
    def vobject_item(self):
        if self._vobject_item is None:
            try:
                self._vobject_item = vobject.readOne(self._text)
            except Exception as e:
                raise RuntimeError("Failed to parse item %r from %r: %s" %
                                   (self.href, self._collection_path,
                                    e)) from e
        return self._vobject_item

    @property
    def etag(self) -> str:
        """Encoded as quoted-string (see RFC 2616)."""
        if self._etag is None:
            self._etag = get_etag(self.serialize())
        return self._etag

    @property
    def uid(self) -> str:
        if self._uid is None:
            self._uid = get_uid_from_object(self.vobject_item)
        return self._uid

    @property
    def name(self) -> str:
        if self._name is None:
            self._name = self.vobject_item.name or ""
        return self._name

    @property
    def component_name(self) -> str:
        if self._component_name is None:
            self._component_name = find_tag(self.vobject_item)
        return self._component_name

    @property
    def time_range(self) -> Tuple[int, int]:
        if self._time_range is None:
            self._time_range = find_time_range(
                self.vobject_item, self.component_name)
        return self._time_range

    def prepare(self) -> None:
        """Fill cache with values."""
        orig_vobject_item = self._vobject_item
        self.serialize()
        self.etag
        self.uid
        self.name
        self.time_range
        self.component_name
        self._vobject_item = orig_vobject_item

1	# This file is part of Radicale - CalDAV and CardDAV server
2	# Copyright © 2008 Nicolas Kandel
3	# Copyright © 2008 Pascal Halter
4	# Copyright © 2014 Jean-Marc Martins
5	# Copyright © 2008-2017 Guillaume Ayoub
6	# Copyright © 2017-2018 Unrud <unrud@outlook.com>
7	#
8	# This library is free software: you can redistribute it and/or modify
9	# it under the terms of the GNU General Public License as published by
10	# the Free Software Foundation, either version 3 of the License, or
11	# (at your option) any later version.
12	#
13	# This library is distributed in the hope that it will be useful,
14	# but WITHOUT ANY WARRANTY; without even the implied warranty of
15	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
16	# GNU General Public License for more details.
17	#
18	# You should have received a copy of the GNU General Public License
19	# along with Radicale. If not, see <http://www.gnu.org/licenses/>.
20
21	"""	15✔
22	Module for address books and calendar entries (see ``Item``).
23
24	"""
25
26	import binascii	15✔
27	import contextlib	15✔
28	import math	15✔
29	import os	15✔
30	import re	15✔
31	import sys	15✔
32	from datetime import datetime, timedelta	15✔
33	from hashlib import sha256	15✔
34	from itertools import chain	15✔
35	from typing import (Any, Callable, List, MutableMapping, Optional, Sequence,	15✔
36	Tuple)
37
38	import vobject	15✔
39
40	from radicale import storage # noqa:F401	15✔
41	from radicale import pathutils	15✔
42	from radicale.item import filter as radicale_filter	15✔
43	from radicale.log import logger	15✔
44
45
46	def read_components(s: str) -> List[vobject.base.Component]:	15✔
47	"""Wrapper for vobject.readComponents"""
48	# Workaround for bug in InfCloud
49	# PHOTO is a data URI
50	s = re.sub(r"^(PHOTO(?:;[^:\r\n])?;ENCODING=b(?:;[^:\r\n])?:)"	15✔
51	r"data:[^;,\r\n]*;base64,", r"\1", s,
52	flags=re.MULTILINE \| re.IGNORECASE)
53	return list(vobject.readComponents(s))	15✔
54
55
56	def predict_tag_of_parent_collection(	15✔
57	vobject_items: Sequence[vobject.base.Component]) -> Optional[str]:
58	"""Returns the predicted tag or `None`"""
59	if len(vobject_items) != 1:	15!
UNCOV 60	return None	×
61	if vobject_items[0].name == "VCALENDAR":	15!
62	return "VCALENDAR"	15✔
UNCOV 63	if vobject_items[0].name in ("VCARD", "VLIST"):	×
UNCOV 64	return "VADDRESSBOOK"	×
UNCOV 65	return None	×
66
67
68	def predict_tag_of_whole_collection(	15✔
69	vobject_items: Sequence[vobject.base.Component],
70	fallback_tag: Optional[str] = None) -> Optional[str]:
71	"""Returns the predicted tag or `fallback_tag`"""
72	if vobject_items and vobject_items[0].name == "VCALENDAR":	15✔
73	return "VCALENDAR"	15✔
74	if vobject_items and vobject_items[0].name in ("VCARD", "VLIST"):	15!
75	return "VADDRESSBOOK"	15✔
76	if not fallback_tag and not vobject_items:	×
77	# Maybe an empty address book
UNCOV 78	return "VADDRESSBOOK"	×
UNCOV 79	return fallback_tag	×
80
81
82	def check_and_sanitize_items(	15✔
83	vobject_items: List[vobject.base.Component],
84	is_collection: bool = False, tag: str = "") -> None:
85	"""Check vobject items for common errors and add missing UIDs.
86
87	Modifies the list `vobject_items`.
88
89	``is_collection`` indicates that vobject_item contains unrelated
90	components.
91
92	The ``tag`` of the collection.
93
94	"""
95	if tag and tag not in ("VCALENDAR", "VADDRESSBOOK"):	15!
UNCOV 96	raise ValueError("Unsupported collection tag: %r" % tag)	×
97	if not is_collection and len(vobject_items) != 1:	15!
UNCOV 98	raise ValueError("Item contains %d components" % len(vobject_items))	×
99	if tag == "VCALENDAR":	15✔
100	if len(vobject_items) > 1:	15!
UNCOV 101	raise RuntimeError("VCALENDAR collection contains %d "	×
102	"components" % len(vobject_items))
103	vobject_item = vobject_items[0]	15✔
104	if vobject_item.name != "VCALENDAR":	15!
UNCOV 105	raise ValueError("Item type %r not supported in %r "	×
106	"collection" % (vobject_item.name, tag))
107	component_uids = set()	15✔
108	for component in vobject_item.components():	15✔
109	if component.name in ("VTODO", "VEVENT", "VJOURNAL"):	15✔
110	component_uid = get_uid(component)	15✔
111	if component_uid:	15✔
112	component_uids.add(component_uid)	15✔
113	component_name = None	15✔
114	object_uid = None	15✔
115	object_uid_set = False	15✔
116	for component in vobject_item.components():	15✔
117	# https://tools.ietf.org/html/rfc4791#section-4.1
118	if component.name == "VTIMEZONE":	15✔
119	continue	15✔
120	if component_name is None or is_collection:	15✔
121	component_name = component.name	15✔
122	elif component_name != component.name:	15!
UNCOV 123	raise ValueError("Multiple component types in object: %r, %r" %	×
124	(component_name, component.name))
125	if component_name not in ("VTODO", "VEVENT", "VJOURNAL"):	15!
UNCOV 126	continue	×
127	component_uid = get_uid(component)	15✔
128	if not object_uid_set or is_collection:	15✔
129	object_uid_set = True	15✔
130	object_uid = component_uid	15✔
131	if not component_uid:	15✔
132	if not is_collection:	15✔
133	raise ValueError("%s component without UID in object" %	15✔
134	component_name)
135	component_uid = find_available_uid(	15✔
136	component_uids.__contains__)
137	component_uids.add(component_uid)	15✔
138	if hasattr(component, "uid"):	15!
UNCOV 139	component.uid.value = component_uid	×
140	else:
141	component.add("UID").value = component_uid	15✔
142	elif not object_uid or not component_uid:	15!
UNCOV 143	raise ValueError("Multiple %s components without UID in "	×
144	"object" % component_name)
145	elif object_uid != component_uid:	15!
UNCOV 146	raise ValueError(	×
147	"Multiple %s components with different UIDs in object: "
148	"%r, %r" % (component_name, object_uid, component_uid))
149	# Workaround for bug in Lightning (Thunderbird)
150	# Rescheduling a single occurrence from a repeating event creates
151	# an event with DTEND and DURATION:PT0S
152	if (hasattr(component, "dtend") and	15!
153	hasattr(component, "duration") and
154	component.duration.value == timedelta(0)):
UNCOV 155	logger.debug("Quirks: Removing zero duration from %s in "	×
156	"object %r", component_name, component_uid)
157	del component.duration	×
158	# Workaround for Evolution
159	# EXDATE has value DATE even if DTSTART/DTEND is DATE-TIME.
160	# The RFC is vaguely formulated on the issue.
161	# To resolve the issue convert EXDATE and RDATE to
162	# the same type as DTDSTART
163	if hasattr(component, "dtstart"):	15✔
164	ref_date = component.dtstart.value	15✔
165	ref_value_param = component.dtstart.params.get("VALUE")	15✔
166	for dates in chain(component.contents.get("exdate", []),	15✔
167	component.contents.get("rdate", [])):
168	if all(type(d) == type(ref_date) for d in dates.value):	15✔
169	continue	15✔
170	for i, date in enumerate(dates.value):	15✔
171	dates.value[i] = ref_date.replace(	15✔
172	date.year, date.month, date.day)
173	with contextlib.suppress(KeyError):	15✔
174	del dates.params["VALUE"]	15✔
175	if ref_value_param is not None:	15!
UNCOV 176	dates.params["VALUE"] = ref_value_param	×
177	# vobject interprets recurrence rules on demand
178	try:	15✔
179	component.rruleset	15✔
UNCOV 180	except Exception as e:	×
UNCOV 181	raise ValueError("Invalid recurrence rules in %s in object %r"	×
182	% (component.name, component_uid)) from e
183	elif tag == "VADDRESSBOOK":	15!
184	# https://tools.ietf.org/html/rfc6352#section-5.1
185	object_uids = set()	15✔
186	for vobject_item in vobject_items:	15✔
187	if vobject_item.name == "VCARD":	15!
188	object_uid = get_uid(vobject_item)	15✔
189	if object_uid:	15✔
190	object_uids.add(object_uid)	15✔
191	for vobject_item in vobject_items:	15✔
192	if vobject_item.name == "VLIST":	15!
193	# Custom format used by SOGo Connector to store lists of
194	# contacts
UNCOV 195	continue	×
196	if vobject_item.name != "VCARD":	15!
UNCOV 197	raise ValueError("Item type %r not supported in %r "	×
198	"collection" % (vobject_item.name, tag))
199	object_uid = get_uid(vobject_item)	15✔
200	if not object_uid:	15✔
201	if not is_collection:	15✔
202	raise ValueError("%s object without UID" %	15✔
203	vobject_item.name)
204	object_uid = find_available_uid(object_uids.__contains__)	15✔
205	object_uids.add(object_uid)	15✔
206	if hasattr(vobject_item, "uid"):	15!
UNCOV 207	vobject_item.uid.value = object_uid	×
208	else:
209	vobject_item.add("UID").value = object_uid	15✔
210	else:
UNCOV 211	for item in vobject_items:	×
UNCOV 212	raise ValueError("Item type %r not supported in %s collection" %	×
213	(item.name, repr(tag) if tag else "generic"))
214
215
216	def check_and_sanitize_props(props: MutableMapping[Any, Any]	15✔
217	) -> MutableMapping[str, str]:
218	"""Check collection properties for common errors.
219
220	Modifies the dict `props`.
221
222	"""
223	for k, v in list(props.items()): # Make copy to be able to delete items	15✔
224	if not isinstance(k, str):	15!
UNCOV 225	raise ValueError("Key must be %r not %r: %r" % (	×
226	str.__name__, type(k).__name__, k))
227	if not isinstance(v, str):	15✔
228	if v is None:	15!
229	del props[k]	15✔
230	continue	15✔
UNCOV 231	raise ValueError("Value of %r must be %r not %r: %r" % (	×
232	k, str.__name__, type(v).__name__, v))
233	if k == "tag":	15✔
234	if v not in ("", "VCALENDAR", "VADDRESSBOOK"):	15!
UNCOV 235	raise ValueError("Unsupported collection tag: %r" % v)	×
236	return props	15✔
237
238
239	def find_available_uid(exists_fn: Callable[[str], bool], suffix: str = ""	15✔
240	) -> str:
241	"""Generate a pseudo-random UID"""
242	# Prevent infinite loop
243	for _ in range(1000):	15!
244	r = binascii.hexlify(os.urandom(16)).decode("ascii")	15✔
245	name = "%s-%s-%s-%s-%s%s" % (	15✔
246	r[:8], r[8:12], r[12:16], r[16:20], r[20:], suffix)
247	if not exists_fn(name):	15!
248	return name	15✔
249	# something is wrong with the PRNG
UNCOV 250	raise RuntimeError("No unique random sequence found")	×
251
252
253	def get_etag(text: str) -> str:	15✔
254	"""Etag from collection or item.
255
256	Encoded as quoted-string (see RFC 2616).
257
258	"""
259	etag = sha256()	15✔
260	etag.update(text.encode())	15✔
261	return '"%s"' % etag.hexdigest()	15✔
262
263
264	def get_uid(vobject_component: vobject.base.Component) -> str:	15✔
265	"""UID value of an item if defined."""
266	return (vobject_component.uid.value or ""	15✔
267	if hasattr(vobject_component, "uid") else "")
268
269
270	def get_uid_from_object(vobject_item: vobject.base.Component) -> str:	15✔
271	"""UID value of an calendar/addressbook object."""
272	if vobject_item.name == "VCALENDAR":	15✔
273	if hasattr(vobject_item, "vevent"):	15✔
274	return get_uid(vobject_item.vevent)	15✔
275	if hasattr(vobject_item, "vjournal"):	15✔
276	return get_uid(vobject_item.vjournal)	15✔
277	if hasattr(vobject_item, "vtodo"):	15!
278	return get_uid(vobject_item.vtodo)	15✔
279	elif vobject_item.name == "VCARD":	15!
280	return get_uid(vobject_item)	15✔
UNCOV 281	return ""	×
282
283
284	def find_tag(vobject_item: vobject.base.Component) -> str:	15✔
285	"""Find component name from ``vobject_item``."""
286	if vobject_item.name == "VCALENDAR":	15✔
287	for component in vobject_item.components():	15!
288	if component.name != "VTIMEZONE":	15✔
289	return component.name or ""	15✔
290	return ""	15✔
291
292
293	def find_time_range(vobject_item: vobject.base.Component, tag: str	15✔
294	) -> Tuple[int, int]:
295	"""Find enclosing time range from ``vobject item``.
296
297	``tag`` must be set to the return value of ``find_tag``.
298
299	Returns a tuple (``start``, ``end``) where ``start`` and ``end`` are
300	POSIX timestamps.
301
302	This is intened to be used for matching against simplified prefilters.
303
304	"""
305	if not tag:	15✔
306	return radicale_filter.TIMESTAMP_MIN, radicale_filter.TIMESTAMP_MAX	15✔
307	start = end = None	15✔
308
309	def range_fn(range_start: datetime, range_end: datetime,	15✔
310	is_recurrence: bool) -> bool:
311	nonlocal start, end
312	if start is None or range_start < start:	15✔
313	start = range_start	15✔
314	if end is None or end < range_end:	15✔
315	end = range_end	15✔
316	return False	15✔
317
318	def infinity_fn(range_start: datetime) -> bool:	15✔
319	nonlocal start, end
320	if start is None or range_start < start:	15✔
321	start = range_start	15✔
322	end = radicale_filter.DATETIME_MAX	15✔
323	return True	15✔
324
325	radicale_filter.visit_time_ranges(vobject_item, tag, range_fn, infinity_fn)	15✔
326	if start is None:	15✔
327	start = radicale_filter.DATETIME_MIN	15✔
328	if end is None:	15✔
329	end = radicale_filter.DATETIME_MAX	15✔
330	try:	15✔
331	return math.floor(start.timestamp()), math.ceil(end.timestamp())	15✔
UNCOV 332	except ValueError as e:	×
UNCOV 333	if str(e) == ("offset must be a timedelta representing a whole "	×
334	"number of minutes") and sys.version_info < (3, 6):
UNCOV 335	raise RuntimeError("Unsupported in Python < 3.6: %s" % e) from e	×
UNCOV 336	raise	×
337
338
339	class Item:	15✔
340	"""Class for address book and calendar entries."""
341
342	collection: Optional["storage.BaseCollection"]	15✔
343	href: Optional[str]	15✔
344	last_modified: Optional[str]	15✔
345
346	_collection_path: str	15✔
347	_text: Optional[str]	15✔
348	_vobject_item: Optional[vobject.base.Component]	15✔
349	_etag: Optional[str]	15✔
350	_uid: Optional[str]	15✔
351	_name: Optional[str]	15✔
352	_component_name: Optional[str]	15✔
353	_time_range: Optional[Tuple[int, int]]	15✔
354
355	def __init__(self,	15✔
356	collection_path: Optional[str] = None,
357	collection: Optional["storage.BaseCollection"] = None,
358	vobject_item: Optional[vobject.base.Component] = None,
359	href: Optional[str] = None,
360	last_modified: Optional[str] = None,
361	text: Optional[str] = None,
362	etag: Optional[str] = None,
363	uid: Optional[str] = None,
364	name: Optional[str] = None,
365	component_name: Optional[str] = None,
366	time_range: Optional[Tuple[int, int]] = None):
367	"""Initialize an item.
368
369	``collection_path`` the path of the parent collection (optional if
370	``collection`` is set).
371
372	``collection`` the parent collection (optional).
373
374	``href`` the href of the item.
375
376	``last_modified`` the HTTP-datetime of when the item was modified.
377
378	``text`` the text representation of the item (optional if
379	``vobject_item`` is set).
380
381	``vobject_item`` the vobject item (optional if ``text`` is set).
382
383	``etag`` the etag of the item (optional). See ``get_etag``.
384
385	``uid`` the UID of the object (optional). See ``get_uid_from_object``.
386
387	``name`` the name of the item (optional). See ``vobject_item.name``.
388
389	``component_name`` the name of the primary component (optional).
390	See ``find_tag``.
391
392	``time_range`` the enclosing time range. See ``find_time_range``.
393
394	"""
395	if text is None and vobject_item is None:	15!
UNCOV 396	raise ValueError(	×
397	"At least one of 'text' or 'vobject_item' must be set")
398	if collection_path is None:	15✔
399	if collection is None:	15!
UNCOV 400	raise ValueError("At least one of 'collection_path' or "	×
401	"'collection' must be set")
402	collection_path = collection.path	15✔
403	assert collection_path == pathutils.strip_path(	15✔
404	pathutils.sanitize_path(collection_path))
405	self._collection_path = collection_path	15✔
406	self.collection = collection	15✔
407	self.href = href	15✔
408	self.last_modified = last_modified	15✔
409	self._text = text	15✔
410	self._vobject_item = vobject_item	15✔
411	self._etag = etag	15✔
412	self._uid = uid	15✔
413	self._name = name	15✔
414	self._component_name = component_name	15✔
415	self._time_range = time_range	15✔
416
417	def serialize(self) -> str:	15✔
418	if self._text is None:	15✔
419	try:	15✔
420	self._text = self.vobject_item.serialize()	15✔
UNCOV 421	except Exception as e:	×
UNCOV 422	raise RuntimeError("Failed to serialize item %r from %r: %s" %	×
423	(self.href, self._collection_path,
424	e)) from e
425	return self._text	15✔
426
427	@property	15✔
428	def vobject_item(self):	6✔
429	if self._vobject_item is None:	15✔
430	try:	15✔
431	self._vobject_item = vobject.readOne(self._text)	15✔
432	except Exception as e:	×
433	raise RuntimeError("Failed to parse item %r from %r: %s" %	×
434	(self.href, self._collection_path,
435	e)) from e
436	return self._vobject_item	15✔
437
438	@property	15✔
439	def etag(self) -> str:	15✔
440	"""Encoded as quoted-string (see RFC 2616)."""
441	if self._etag is None:	15✔
442	self._etag = get_etag(self.serialize())	15✔
443	return self._etag	15✔
444
445	@property	15✔
446	def uid(self) -> str:	15✔
447	if self._uid is None:	15✔
448	self._uid = get_uid_from_object(self.vobject_item)	15✔
449	return self._uid	15✔
450
451	@property	15✔
452	def name(self) -> str:	15✔
453	if self._name is None:	15✔
454	self._name = self.vobject_item.name or ""	15✔
455	return self._name	15✔
456
457	@property	15✔
458	def component_name(self) -> str:	15✔
459	if self._component_name is None:	15✔
460	self._component_name = find_tag(self.vobject_item)	15✔
461	return self._component_name	15✔
462
463	@property	15✔
464	def time_range(self) -> Tuple[int, int]:	15✔
465	if self._time_range is None:	15✔
466	self._time_range = find_time_range(	15✔
467	self.vobject_item, self.component_name)
468	return self._time_range	15✔
469
470	def prepare(self) -> None:	15✔
471	"""Fill cache with values."""
472	orig_vobject_item = self._vobject_item	15✔
473	self.serialize()	15✔
474	self.etag	15✔
475	self.uid	15✔
476	self.name	15✔
477	self.time_range	15✔
478	self.component_name	15✔
479	self._vobject_item = orig_vobject_item	15✔

Kozea / Radicale / 1702737770

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