Change the Collection API
The new API used comes from vdirsyncer, as proposed by @untitaker in issue #130. The code has been tested and works with the (too simple) unit tests, and with Lightning and DAVdroid. Many things are broken and a good part of the code has not be ported to the new API yet. TODOs have been added where the application is known to be broken.
This commit is contained in:
Guillaume Ayoub
parent
8102926148
commit
406027f3c9
@ -33,7 +33,6 @@ import socket
|
||||
import ssl
|
||||
import wsgiref.simple_server
|
||||
import re
|
||||
|
||||
from http import client
|
||||
from urllib.parse import unquote, urlparse
|
||||
|
||||
@ -163,12 +162,6 @@ class Application(object):
|
||||
pass
|
||||
raise UnicodeDecodeError
|
||||
|
||||
@staticmethod
|
||||
def sanitize_uri(uri):
|
||||
"""Unquote and make absolute to prevent access to other data."""
|
||||
uri = unquote(uri)
|
||||
return storage.sanitize_path(uri)
|
||||
|
||||
def collect_allowed_items(self, items, user):
|
||||
"""Get items from request that user is allowed to access."""
|
||||
read_last_collection_allowed = None
|
||||
@ -181,25 +174,25 @@ class Application(object):
|
||||
if rights.authorized(user, item, "r"):
|
||||
log.LOGGER.debug(
|
||||
"%s has read access to collection %s" %
|
||||
(user or "Anonymous", item.url or "/"))
|
||||
(user or "Anonymous", item.path or "/"))
|
||||
read_last_collection_allowed = True
|
||||
read_allowed_items.append(item)
|
||||
else:
|
||||
log.LOGGER.debug(
|
||||
"%s has NO read access to collection %s" %
|
||||
(user or "Anonymous", item.url or "/"))
|
||||
(user or "Anonymous", item.path or "/"))
|
||||
read_last_collection_allowed = False
|
||||
|
||||
if rights.authorized(user, item, "w"):
|
||||
log.LOGGER.debug(
|
||||
"%s has write access to collection %s" %
|
||||
(user or "Anonymous", item.url or "/"))
|
||||
(user or "Anonymous", item.path or "/"))
|
||||
write_last_collection_allowed = True
|
||||
write_allowed_items.append(item)
|
||||
else:
|
||||
log.LOGGER.debug(
|
||||
"%s has NO write access to collection %s" %
|
||||
(user or "Anonymous", item.url or "/"))
|
||||
(user or "Anonymous", item.path or "/"))
|
||||
write_last_collection_allowed = False
|
||||
else:
|
||||
# item is not a collection, it's the child of the last
|
||||
@ -208,22 +201,22 @@ class Application(object):
|
||||
if read_last_collection_allowed:
|
||||
log.LOGGER.debug(
|
||||
"%s has read access to item %s" %
|
||||
(user or "Anonymous", item.name))
|
||||
(user or "Anonymous", item.href))
|
||||
read_allowed_items.append(item)
|
||||
else:
|
||||
log.LOGGER.debug(
|
||||
"%s has NO read access to item %s" %
|
||||
(user or "Anonymous", item.name))
|
||||
(user or "Anonymous", item.href))
|
||||
|
||||
if write_last_collection_allowed:
|
||||
log.LOGGER.debug(
|
||||
"%s has write access to item %s" %
|
||||
(user or "Anonymous", item.name))
|
||||
(user or "Anonymous", item.href))
|
||||
write_allowed_items.append(item)
|
||||
else:
|
||||
log.LOGGER.debug(
|
||||
"%s has NO write access to item %s" %
|
||||
(user or "Anonymous", item.name))
|
||||
(user or "Anonymous", item.href))
|
||||
|
||||
return read_allowed_items, write_allowed_items
|
||||
|
||||
@ -250,7 +243,8 @@ class Application(object):
|
||||
return []
|
||||
|
||||
# Sanitize request URI
|
||||
environ["PATH_INFO"] = self.sanitize_uri(environ["PATH_INFO"])
|
||||
environ["PATH_INFO"] = storage.sanitize_path(
|
||||
unquote(environ["PATH_INFO"]))
|
||||
log.LOGGER.debug("Sanitized path: %s", environ["PATH_INFO"])
|
||||
|
||||
path = environ["PATH_INFO"]
|
||||
@ -294,7 +288,7 @@ class Application(object):
|
||||
is_valid_user = is_authenticated or not user
|
||||
|
||||
if is_valid_user:
|
||||
items = storage.Collection.from_path(
|
||||
items = storage.Collection.discover(
|
||||
path, environ.get("HTTP_DEPTH", "0"))
|
||||
read_allowed_items, write_allowed_items = (
|
||||
self.collect_allowed_items(items, user))
|
||||
@ -366,7 +360,7 @@ class Application(object):
|
||||
else:
|
||||
# Try to get an item matching the path
|
||||
name = xmlutils.name_from_path(environ["PATH_INFO"], collection)
|
||||
item = collection.items.get(name)
|
||||
item = collection.get(name)
|
||||
|
||||
if item:
|
||||
if_match = environ.get("HTTP_IF_MATCH", "*")
|
||||
@ -396,26 +390,22 @@ class Application(object):
|
||||
|
||||
if item_name:
|
||||
# Get collection item
|
||||
item = collection.items.get(item_name)
|
||||
item = collection.get(item_name)
|
||||
if item:
|
||||
items = [item]
|
||||
if collection.resource_type == "calendar":
|
||||
items.extend(collection.timezones)
|
||||
answer_text = storage.serialize(
|
||||
collection.tag, collection.headers, items)
|
||||
answer_text = item.serialize()
|
||||
etag = item.etag
|
||||
else:
|
||||
return client.NOT_FOUND, {}, None
|
||||
elif not collection.exists:
|
||||
log.LOGGER.debug("Collection at %s unknown" % environ["PATH_INFO"])
|
||||
return client.NOT_FOUND, {}, None
|
||||
else:
|
||||
# Get whole collection
|
||||
answer_text = collection.text
|
||||
answer_text = collection.serialize()
|
||||
if not answer_text:
|
||||
log.LOGGER.debug("Collection at %s unknown" % environ["PATH_INFO"])
|
||||
return client.NOT_FOUND, {}, None
|
||||
etag = collection.etag
|
||||
|
||||
headers = {
|
||||
"Content-Type": collection.mimetype,
|
||||
"Content-Type": storage.MIMETYPES[collection.get_meta("tag")],
|
||||
"Last-Modified": collection.last_modified,
|
||||
"ETag": etag}
|
||||
answer = answer_text.encode(self.encoding)
|
||||
@ -437,14 +427,12 @@ class Application(object):
|
||||
collection = write_collections[0]
|
||||
|
||||
props = xmlutils.props_from_request(content)
|
||||
timezone = props.get("C:calendar-timezone")
|
||||
if timezone:
|
||||
collection.replace("", timezone)
|
||||
del props["C:calendar-timezone"]
|
||||
with collection.props as collection_props:
|
||||
for key, value in props.items():
|
||||
collection_props[key] = value
|
||||
collection.write()
|
||||
# TODO: use this?
|
||||
# timezone = props.get("C:calendar-timezone")
|
||||
collection = storage.create_collection(
|
||||
collection.path, tag="VCALENDAR")
|
||||
for key, value in props.items():
|
||||
collection.set_meta(key, value)
|
||||
return client.CREATED, {}, None
|
||||
|
||||
def do_MKCOL(self, environ, read_collections, write_collections, content,
|
||||
@ -456,10 +444,9 @@ class Application(object):
|
||||
collection = write_collections[0]
|
||||
|
||||
props = xmlutils.props_from_request(content)
|
||||
with collection.props as collection_props:
|
||||
for key, value in props.items():
|
||||
collection_props[key] = value
|
||||
collection.write()
|
||||
collection = storage.create_collection(collection.path)
|
||||
for key, value in props.items():
|
||||
collection.set_meta(key, value)
|
||||
return client.CREATED, {}, None
|
||||
|
||||
def do_MOVE(self, environ, read_collections, write_collections, content,
|
||||
@ -469,34 +456,29 @@ class Application(object):
|
||||
return NOT_ALLOWED
|
||||
|
||||
from_collection = write_collections[0]
|
||||
|
||||
from_name = xmlutils.name_from_path(
|
||||
environ["PATH_INFO"], from_collection)
|
||||
if from_name:
|
||||
item = from_collection.items.get(from_name)
|
||||
if item:
|
||||
# Move the item
|
||||
to_url_parts = urlparse(environ["HTTP_DESTINATION"])
|
||||
if to_url_parts.netloc == environ["HTTP_HOST"]:
|
||||
to_url = to_url_parts.path
|
||||
to_path, to_name = to_url.rstrip("/").rsplit("/", 1)
|
||||
to_collection = storage.Collection.from_path(
|
||||
to_path, depth="0")[0]
|
||||
item = from_collection.get(from_name)
|
||||
if item:
|
||||
# Move the item
|
||||
to_url_parts = urlparse(environ["HTTP_DESTINATION"])
|
||||
if to_url_parts.netloc == environ["HTTP_HOST"]:
|
||||
to_url = to_url_parts.path
|
||||
to_path, to_name = to_url.rstrip("/").rsplit("/", 1)
|
||||
for to_collection in storage.Collection.discover(
|
||||
to_path, depth="0"):
|
||||
if to_collection in write_collections:
|
||||
to_collection.append(to_name, item.text)
|
||||
from_collection.remove(from_name)
|
||||
to_collection.upload(to_name, item)
|
||||
from_collection.delete(from_name)
|
||||
return client.CREATED, {}, None
|
||||
else:
|
||||
return NOT_ALLOWED
|
||||
else:
|
||||
# Remote destination server, not supported
|
||||
return client.BAD_GATEWAY, {}, None
|
||||
else:
|
||||
# No item found
|
||||
return client.GONE, {}, None
|
||||
# Remote destination server, not supported
|
||||
return client.BAD_GATEWAY, {}, None
|
||||
else:
|
||||
# Moving collections, not supported
|
||||
return client.FORBIDDEN, {}, None
|
||||
# No item found
|
||||
return client.GONE, {}, None
|
||||
|
||||
def do_OPTIONS(self, environ, read_collections, write_collections,
|
||||
content, user):
|
||||
@ -510,7 +492,7 @@ class Application(object):
|
||||
def do_PROPFIND(self, environ, read_collections, write_collections,
|
||||
content, user):
|
||||
"""Manage PROPFIND request."""
|
||||
if not any(collection.exists for collection in read_collections):
|
||||
if not read_collections:
|
||||
return client.NOT_FOUND, {}, None
|
||||
headers = {
|
||||
"DAV": "1, 2, 3, calendar-access, addressbook, extended-mkcol",
|
||||
@ -542,10 +524,13 @@ class Application(object):
|
||||
|
||||
collection = write_collections[0]
|
||||
|
||||
collection.set_mimetype(environ.get("CONTENT_TYPE"))
|
||||
content_type = environ.get("CONTENT_TYPE")
|
||||
if content_type:
|
||||
tags = {value: key for key, value in storage.MIMETYPES.items()}
|
||||
collection.set_meta("tag", tags[content_type.split(";")[0]])
|
||||
headers = {}
|
||||
item_name = xmlutils.name_from_path(environ["PATH_INFO"], collection)
|
||||
item = collection.items.get(item_name)
|
||||
item = collection.get(item_name)
|
||||
|
||||
etag = environ.get("HTTP_IF_MATCH", "")
|
||||
match = environ.get("HTTP_IF_NONE_MATCH", "") == "*"
|
||||
@ -561,7 +546,7 @@ class Application(object):
|
||||
# If the added item doesn't have the same name as the one given
|
||||
# by the client, then there's no obvious way to generate an
|
||||
# etag, we can safely ignore it.
|
||||
new_item = collection.items.get(item_name)
|
||||
new_item = collection.get(item_name)
|
||||
if new_item:
|
||||
headers["ETag"] = new_item.etag
|
||||
else:
|
||||
|
||||
@ -33,7 +33,6 @@ import sys
|
||||
import time
|
||||
from contextlib import contextmanager
|
||||
from hashlib import md5
|
||||
from random import randint
|
||||
from uuid import uuid4
|
||||
|
||||
import vobject
|
||||
@ -55,26 +54,14 @@ def _load():
|
||||
FOLDER = os.path.expanduser(config.get("storage", "filesystem_folder"))
|
||||
FILESYSTEM_ENCODING = sys.getfilesystemencoding()
|
||||
STORAGE_ENCODING = config.get("encoding", "stock")
|
||||
MIMETYPES = {"VADDRESSBOOK": "text/vcard", "VCALENDAR": "text/calendar"}
|
||||
|
||||
|
||||
def serialize(tag, headers=(), items=()):
|
||||
"""Return a text corresponding to given collection ``tag``.
|
||||
|
||||
The text may have the given ``headers`` and ``items`` added around the
|
||||
items if needed (ie. for calendars).
|
||||
|
||||
"""
|
||||
items = sorted(items, key=lambda x: x.name)
|
||||
if tag == "VADDRESSBOOK":
|
||||
lines = [item.text.strip() for item in items]
|
||||
else:
|
||||
lines = ["BEGIN:%s" % tag]
|
||||
for part in (headers, items):
|
||||
if part:
|
||||
lines.append("\r\n".join(item.text.strip() for item in part))
|
||||
lines.append("END:%s" % tag)
|
||||
lines.append("")
|
||||
return "\r\n".join(lines)
|
||||
def get_etag(text):
|
||||
"""Etag from collection or item."""
|
||||
etag = md5()
|
||||
etag.update(text.encode("utf-8"))
|
||||
return '"%s"' % etag.hexdigest()
|
||||
|
||||
|
||||
def sanitize_path(path):
|
||||
@ -105,129 +92,35 @@ def is_safe_filesystem_path_component(path):
|
||||
not os.path.split(path)[0] and path not in (os.curdir, os.pardir))
|
||||
|
||||
|
||||
def path_to_filesystem(path):
|
||||
def path_to_filesystem(root, *paths):
|
||||
"""Convert path to a local filesystem path relative to base_folder.
|
||||
|
||||
Conversion is done in a secure manner, or raises ``ValueError``.
|
||||
|
||||
"""
|
||||
sane_path = sanitize_path(path).strip("/")
|
||||
safe_path = FOLDER
|
||||
if not sane_path:
|
||||
return safe_path
|
||||
for part in sane_path.split("/"):
|
||||
if not is_safe_filesystem_path_component(part):
|
||||
log.LOGGER.debug(
|
||||
"Can't translate path safely to filesystem: %s", path)
|
||||
raise ValueError("Unsafe path")
|
||||
safe_path = os.path.join(safe_path, part)
|
||||
root = sanitize_path(root)
|
||||
paths = [sanitize_path(path).strip("/") for path in paths]
|
||||
safe_path = root
|
||||
for path in paths:
|
||||
if not path:
|
||||
continue
|
||||
for part in path.split("/"):
|
||||
if not is_safe_filesystem_path_component(part):
|
||||
log.LOGGER.debug(
|
||||
"Can't translate path safely to filesystem: %s", path)
|
||||
raise ValueError("Unsafe path")
|
||||
safe_path = os.path.join(safe_path, part)
|
||||
return safe_path
|
||||
|
||||
|
||||
class Item(object):
|
||||
"""Internal iCal item."""
|
||||
def __init__(self, text, name=None):
|
||||
"""Initialize object from ``text`` and different ``kwargs``."""
|
||||
self.component = vobject.readOne(text)
|
||||
self._name = name
|
||||
class Item:
|
||||
def __init__(self, item, href, etag):
|
||||
self.item = item
|
||||
self.href = href
|
||||
self.etag = etag
|
||||
|
||||
if not self.component.name:
|
||||
# Header
|
||||
self._name = next(self.component.lines()).name.lower()
|
||||
return
|
||||
|
||||
# We must synchronize the name in the text and in the object.
|
||||
# An item must have a name, determined in order by:
|
||||
#
|
||||
# - the ``name`` parameter
|
||||
# - the ``X-RADICALE-NAME`` iCal property (for Events, Todos, Journals)
|
||||
# - the ``UID`` iCal property (for Events, Todos, Journals)
|
||||
# - the ``TZID`` iCal property (for Timezones)
|
||||
if not self._name:
|
||||
for line in self.component.lines():
|
||||
if line.name in ("X-RADICALE-NAME", "UID", "TZID"):
|
||||
self._name = line.value
|
||||
if line.name == "X-RADICALE-NAME":
|
||||
break
|
||||
|
||||
if self._name:
|
||||
# Leading and ending brackets that may have been put by Outlook.
|
||||
# Slashes are mostly unwanted when saving collections on disk.
|
||||
self._name = self._name.strip("{}").replace("/", "_")
|
||||
else:
|
||||
self._name = uuid4().hex
|
||||
|
||||
if not hasattr(self.component, "x_radicale_name"):
|
||||
self.component.add("X-RADICALE-NAME")
|
||||
self.component.x_radicale_name.value = self._name
|
||||
|
||||
def __hash__(self):
|
||||
return hash(self.text)
|
||||
|
||||
def __eq__(self, item):
|
||||
return isinstance(item, Item) and self.text == item.text
|
||||
|
||||
@property
|
||||
def etag(self):
|
||||
"""Item etag.
|
||||
|
||||
Etag is mainly used to know if an item has changed.
|
||||
|
||||
"""
|
||||
etag = md5()
|
||||
etag.update(self.text.encode("utf-8"))
|
||||
return '"%s"' % etag.hexdigest()
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
"""Item name.
|
||||
|
||||
Name is mainly used to give an URL to the item.
|
||||
|
||||
"""
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def text(self):
|
||||
"""Item serialized text."""
|
||||
return self.component.serialize()
|
||||
|
||||
|
||||
class Header(Item):
|
||||
"""Internal header class."""
|
||||
|
||||
|
||||
class Timezone(Item):
|
||||
"""Internal timezone class."""
|
||||
tag = "VTIMEZONE"
|
||||
|
||||
|
||||
class Component(Item):
|
||||
"""Internal main component of a collection."""
|
||||
|
||||
|
||||
class Event(Component):
|
||||
"""Internal event class."""
|
||||
tag = "VEVENT"
|
||||
mimetype = "text/calendar"
|
||||
|
||||
|
||||
class Todo(Component):
|
||||
"""Internal todo class."""
|
||||
tag = "VTODO" # pylint: disable=W0511
|
||||
mimetype = "text/calendar"
|
||||
|
||||
|
||||
class Journal(Component):
|
||||
"""Internal journal class."""
|
||||
tag = "VJOURNAL"
|
||||
mimetype = "text/calendar"
|
||||
|
||||
|
||||
class Card(Component):
|
||||
"""Internal card class."""
|
||||
tag = "VCARD"
|
||||
mimetype = "text/vcard"
|
||||
def __getattr__(self, attr):
|
||||
return getattr(self.item, attr)
|
||||
|
||||
|
||||
class Collection:
|
||||
@ -242,21 +135,18 @@ class Collection:
|
||||
self.encoding = "utf-8"
|
||||
# path should already be sanitized
|
||||
self.path = sanitize_path(path).strip("/")
|
||||
self._filesystem_path = path_to_filesystem(FOLDER, self.path)
|
||||
split_path = self.path.split("/")
|
||||
if principal and split_path and self.is_node(self.path):
|
||||
# Already existing principal collection
|
||||
self.owner = split_path[0]
|
||||
elif len(split_path) > 1:
|
||||
if len(split_path) > 1:
|
||||
# URL with at least one folder
|
||||
self.owner = split_path[0]
|
||||
else:
|
||||
self.owner = None
|
||||
self.is_principal = principal
|
||||
self._items = None
|
||||
|
||||
@classmethod
|
||||
def from_path(cls, path, depth="1", include_container=True):
|
||||
"""Return a list of collections and items under the given ``path``.
|
||||
def discover(cls, path, depth="1"):
|
||||
"""Discover a list of collections under the given ``path``.
|
||||
|
||||
If ``depth`` is "0", only the actual object under ``path`` is
|
||||
returned.
|
||||
@ -271,314 +161,223 @@ class Collection:
|
||||
"""
|
||||
# path == None means wrong URL
|
||||
if path is None:
|
||||
return []
|
||||
return
|
||||
|
||||
# path should already be sanitized
|
||||
sane_path = sanitize_path(path).strip("/")
|
||||
attributes = sane_path.split("/")
|
||||
if not attributes:
|
||||
return []
|
||||
return
|
||||
|
||||
# Try to guess if the path leads to a collection or an item
|
||||
if cls.is_leaf("/".join(attributes[:-1])):
|
||||
if os.path.exists(path_to_filesystem(
|
||||
FOLDER, *attributes[:-1]) + ".props"):
|
||||
attributes.pop()
|
||||
|
||||
result = []
|
||||
path = "/".join(attributes)
|
||||
|
||||
principal = len(attributes) <= 1
|
||||
if cls.is_node(path):
|
||||
if depth == "0":
|
||||
result.append(cls(path, principal))
|
||||
collection = cls(path, principal)
|
||||
yield collection
|
||||
if depth != "0":
|
||||
items = list(collection.list())
|
||||
if items:
|
||||
for item in items:
|
||||
yield collection.get(item[0])
|
||||
else:
|
||||
if include_container:
|
||||
result.append(cls(path, principal))
|
||||
for child in cls.children(path):
|
||||
result.append(child)
|
||||
else:
|
||||
if depth == "0":
|
||||
result.append(cls(path))
|
||||
else:
|
||||
collection = cls(path, principal)
|
||||
if include_container:
|
||||
result.append(collection)
|
||||
result.extend(collection.components)
|
||||
return result
|
||||
_, directories, files = next(os.walk(collection._filesystem_path))
|
||||
for sub_path in directories + files:
|
||||
full_path = os.path.join(collection._filesystem_path, sub_path)
|
||||
if os.path.exists(path_to_filesystem(full_path)):
|
||||
collection = cls(posixpath.join(path, sub_path))
|
||||
yield collection
|
||||
|
||||
@property
|
||||
def _filesystem_path(self):
|
||||
"""Absolute path of the file at local ``path``."""
|
||||
return path_to_filesystem(self.path)
|
||||
@classmethod
|
||||
def create_collection(cls, href, collection=None, tag=None):
|
||||
"""Create a collection.
|
||||
|
||||
@property
|
||||
def _props_path(self):
|
||||
"""Absolute path of the file storing the collection properties."""
|
||||
return self._filesystem_path + ".props"
|
||||
``collection`` is a list of vobject components.
|
||||
|
||||
def _create_dirs(self):
|
||||
"""Create folder storing the collection if absent."""
|
||||
if not os.path.exists(self._filesystem_path):
|
||||
os.makedirs(self._filesystem_path)
|
||||
|
||||
def set_mimetype(self, mimetype):
|
||||
self._create_dirs()
|
||||
with self.props as props:
|
||||
if "tag" not in props:
|
||||
if mimetype == "text/vcard":
|
||||
props["tag"] = "VADDRESSBOOK"
|
||||
else:
|
||||
props["tag"] = "VCALENDAR"
|
||||
|
||||
@property
|
||||
def exists(self):
|
||||
"""``True`` if the collection exists on the storage, else ``False``."""
|
||||
return self.is_node(self.path) or self.is_leaf(self.path)
|
||||
|
||||
@staticmethod
|
||||
def _parse(text, item_types, name=None):
|
||||
"""Find items with type in ``item_types`` in ``text``.
|
||||
|
||||
If ``name`` is given, give this name to new items in ``text``.
|
||||
|
||||
Return a dict of items.
|
||||
``tag`` is the type of collection (VCALENDAR or VADDRESSBOOK). If
|
||||
``tag`` is not given, it is guessed from the collection.
|
||||
|
||||
"""
|
||||
item_tags = {item_type.tag: item_type for item_type in item_types}
|
||||
items = {}
|
||||
root = next(vobject.readComponents(text))
|
||||
components = (
|
||||
root.components() if root.name in ("VADDRESSBOOK", "VCALENDAR")
|
||||
else (root,))
|
||||
for component in components:
|
||||
item_name = None if component.name == "VTIMEZONE" else name
|
||||
item_type = item_tags[component.name]
|
||||
item = item_type(component.serialize(), item_name)
|
||||
if item.name in items:
|
||||
text = "\r\n".join((item.text, items[item.name].text))
|
||||
items[item.name] = item_type(text, item.name)
|
||||
else:
|
||||
items[item.name] = item
|
||||
path = path_to_filesystem(FOLDER, href)
|
||||
if not os.path.exists(path):
|
||||
os.makedirs(path)
|
||||
if not tag and collection:
|
||||
tag = collection[0].name
|
||||
self = cls(href)
|
||||
if tag == "VCALENDAR":
|
||||
self.set_meta("tag", "VCALENDAR")
|
||||
if collection:
|
||||
collection, = collection
|
||||
for content in ("vevent", "vtodo", "vjournal"):
|
||||
if content in collection.contents:
|
||||
for item in getattr(collection, "%s_list" % content):
|
||||
new_collection = vobject.iCalendar()
|
||||
new_collection.add(item)
|
||||
self.upload(uuid4().hex, new_collection)
|
||||
elif tag == "VCARD":
|
||||
self.set_meta("tag", "VADDRESSBOOK")
|
||||
if collection:
|
||||
for card in collection:
|
||||
self.upload(uuid4().hex, card)
|
||||
return self
|
||||
|
||||
return items
|
||||
def list(self):
|
||||
"""List collection items."""
|
||||
for href in os.listdir(self._filesystem_path):
|
||||
path = os.path.join(self._filesystem_path, href)
|
||||
if not href.endswith(".props") and os.path.isfile(path):
|
||||
with open(path, encoding=STORAGE_ENCODING) as fd:
|
||||
yield href, get_etag(fd.read())
|
||||
|
||||
def save(self, text):
|
||||
self._create_dirs()
|
||||
item_types = (Timezone, Event, Todo, Journal, Card)
|
||||
for name, component in self._parse(text, item_types).items():
|
||||
if not is_safe_filesystem_path_component(name):
|
||||
log.LOGGER.debug(
|
||||
"Can't tranlate name safely to filesystem, "
|
||||
"skipping component: %s", name)
|
||||
continue
|
||||
filename = os.path.join(self._filesystem_path, name)
|
||||
with open(filename, "w", encoding=STORAGE_ENCODING) as fd:
|
||||
fd.write(component.text)
|
||||
|
||||
@property
|
||||
def headers(self):
|
||||
return (
|
||||
Header("PRODID:-//Radicale//NONSGML Radicale Server//EN"),
|
||||
Header("VERSION:%s" % self.version))
|
||||
|
||||
def delete(self):
|
||||
shutil.rmtree(self._filesystem_path)
|
||||
os.remove(self._props_path)
|
||||
|
||||
def remove(self, name):
|
||||
if not is_safe_filesystem_path_component(name):
|
||||
def get(self, href):
|
||||
"""Fetch a single item."""
|
||||
if not href:
|
||||
return
|
||||
href = href.strip("{}").replace("/", "_")
|
||||
if is_safe_filesystem_path_component(href):
|
||||
path = os.path.join(self._filesystem_path, href)
|
||||
if os.path.isfile(path):
|
||||
with open(path, encoding=STORAGE_ENCODING) as fd:
|
||||
text = fd.read()
|
||||
return Item(vobject.readOne(text), href, get_etag(text))
|
||||
else:
|
||||
log.LOGGER.debug(
|
||||
"Can't tranlate name safely to filesystem, "
|
||||
"skipping component: %s", name)
|
||||
return
|
||||
if name in self.items:
|
||||
del self.items[name]
|
||||
filesystem_path = os.path.join(self._filesystem_path, name)
|
||||
if os.path.exists(filesystem_path):
|
||||
os.remove(filesystem_path)
|
||||
"skipping component: %s", href)
|
||||
|
||||
@property
|
||||
def text(self):
|
||||
components = (Timezone, Event, Todo, Journal, Card)
|
||||
items = {}
|
||||
try:
|
||||
filenames = os.listdir(self._filesystem_path)
|
||||
except (OSError, IOError) as e:
|
||||
log.LOGGER.info(
|
||||
"Error while reading collection %r: %r" % (
|
||||
self._filesystem_path, e))
|
||||
return ""
|
||||
def get_multi(self, hrefs):
|
||||
"""Fetch multiple items. Duplicate hrefs must be ignored.
|
||||
|
||||
for filename in filenames:
|
||||
path = os.path.join(self._filesystem_path, filename)
|
||||
try:
|
||||
Functionally similar to ``get``, but might bring performance benefits
|
||||
on some storages when used cleverly.
|
||||
|
||||
"""
|
||||
for href in set(hrefs):
|
||||
yield self.get(href)
|
||||
|
||||
def has(self, href):
|
||||
"""Check if an item exists by its href."""
|
||||
return self.get(href) is not None
|
||||
|
||||
def upload(self, href, item):
|
||||
"""Upload a new item."""
|
||||
# TODO: use returned object in code
|
||||
if is_safe_filesystem_path_component(href):
|
||||
path = path_to_filesystem(self._filesystem_path, href)
|
||||
if not os.path.exists(path):
|
||||
text = item.serialize()
|
||||
with open(path, "w", encoding=STORAGE_ENCODING) as fd:
|
||||
fd.write(text)
|
||||
return href, get_etag(text)
|
||||
else:
|
||||
log.LOGGER.debug(
|
||||
"Can't tranlate name safely to filesystem, "
|
||||
"skipping component: %s", href)
|
||||
|
||||
def update(self, href, item, etag=None):
|
||||
"""Update an item."""
|
||||
# TODO: use etag in code and test it here
|
||||
# TODO: use returned object in code
|
||||
if is_safe_filesystem_path_component(href):
|
||||
path = path_to_filesystem(self._filesystem_path, href)
|
||||
if os.path.exists(path):
|
||||
with open(path, encoding=STORAGE_ENCODING) as fd:
|
||||
items.update(self._parse(fd.read(), components))
|
||||
except (OSError, IOError) as e:
|
||||
log.LOGGER.warning(
|
||||
"Error while reading item %r: %r" % (path, e))
|
||||
text = fd.read()
|
||||
if not etag or etag == get_etag(text):
|
||||
new_text = item.serialize()
|
||||
with open(path, "w", encoding=STORAGE_ENCODING) as fd:
|
||||
fd.write(new_text)
|
||||
return get_etag(new_text)
|
||||
else:
|
||||
log.LOGGER.debug(
|
||||
"Can't tranlate name safely to filesystem, "
|
||||
"skipping component: %s", href)
|
||||
|
||||
return serialize(
|
||||
self.tag, self.headers, sorted(items.values(), key=lambda x: x.name))
|
||||
def delete(self, href=None, etag=None):
|
||||
"""Delete an item.
|
||||
|
||||
@classmethod
|
||||
def children(cls, path):
|
||||
filesystem_path = path_to_filesystem(path)
|
||||
_, directories, files = next(os.walk(filesystem_path))
|
||||
for path in directories + files:
|
||||
# Check that the local path can be translated into an internal path
|
||||
if not path or posixpath.split(path)[0] or path in (".", ".."):
|
||||
log.LOGGER.debug("Skipping unsupported filename: %s", path)
|
||||
continue
|
||||
relative_path = posixpath.join(path, path)
|
||||
if cls.is_node(relative_path) or cls.is_leaf(relative_path):
|
||||
yield cls(relative_path)
|
||||
When ``href`` is ``None``, delete the collection.
|
||||
|
||||
@classmethod
|
||||
def is_node(cls, path):
|
||||
filesystem_path = path_to_filesystem(path)
|
||||
return (
|
||||
os.path.isdir(filesystem_path) and
|
||||
not os.path.exists(filesystem_path + ".props"))
|
||||
"""
|
||||
# TODO: use etag in code and test it here
|
||||
# TODO: use returned object in code
|
||||
if href is None:
|
||||
# Delete the collection
|
||||
if os.path.isdir(self._filesystem_path):
|
||||
shutil.rmtree(self._filesystem_path)
|
||||
props_path = self._filesystem_path + ".props"
|
||||
if os.path.isfile(props_path):
|
||||
os.remove(props_path)
|
||||
return
|
||||
elif is_safe_filesystem_path_component(href):
|
||||
# Delete an item
|
||||
path = path_to_filesystem(self._filesystem_path, href)
|
||||
if os.path.isfile(path):
|
||||
with open(path, encoding=STORAGE_ENCODING) as fd:
|
||||
text = fd.read()
|
||||
if not etag or etag == get_etag(text):
|
||||
os.remove(path)
|
||||
return
|
||||
else:
|
||||
log.LOGGER.debug(
|
||||
"Can't tranlate name safely to filesystem, "
|
||||
"skipping component: %s", href)
|
||||
|
||||
@classmethod
|
||||
def is_leaf(cls, path):
|
||||
filesystem_path = path_to_filesystem(path)
|
||||
return (
|
||||
os.path.isdir(filesystem_path) and
|
||||
os.path.exists(filesystem_path + ".props"))
|
||||
@contextmanager
|
||||
def at_once(self):
|
||||
"""Set a context manager buffering the reads and writes."""
|
||||
# TODO: use in code
|
||||
# TODO: use a file locker
|
||||
yield
|
||||
|
||||
def get_meta(self, key):
|
||||
"""Get metadata value for collection."""
|
||||
props_path = self._filesystem_path + ".props"
|
||||
if os.path.exists(props_path):
|
||||
with open(props_path, encoding=STORAGE_ENCODING) as prop_file:
|
||||
return json.load(prop_file).get(key)
|
||||
|
||||
def set_meta(self, key, value):
|
||||
"""Get metadata value for collection."""
|
||||
props_path = self._filesystem_path + ".props"
|
||||
properties = {}
|
||||
if os.path.exists(props_path):
|
||||
with open(props_path, encoding=STORAGE_ENCODING) as prop_file:
|
||||
properties.update(json.load(prop_file))
|
||||
properties[key] = value
|
||||
with open(props_path, "w", encoding=STORAGE_ENCODING) as prop_file:
|
||||
json.dump(properties, prop_file)
|
||||
|
||||
@property
|
||||
def last_modified(self):
|
||||
"""Get the HTTP-datetime of when the collection was modified."""
|
||||
last = max([
|
||||
os.path.getmtime(os.path.join(self._filesystem_path, filename))
|
||||
for filename in os.listdir(self._filesystem_path)] or [0])
|
||||
return time.strftime("%a, %d %b %Y %H:%M:%S +0000", time.gmtime(last))
|
||||
|
||||
@property
|
||||
@contextmanager
|
||||
def props(self):
|
||||
# On enter
|
||||
properties = {}
|
||||
if os.path.exists(self._props_path):
|
||||
with open(self._props_path) as prop_file:
|
||||
properties.update(json.load(prop_file))
|
||||
old_properties = properties.copy()
|
||||
yield properties
|
||||
# On exit
|
||||
if old_properties != properties:
|
||||
with open(self._props_path, "w") as prop_file:
|
||||
json.dump(properties, prop_file)
|
||||
|
||||
def append(self, name, text):
|
||||
"""Append items from ``text`` to collection.
|
||||
|
||||
If ``name`` is given, give this name to new items in ``text``.
|
||||
|
||||
"""
|
||||
new_items = self._parse(
|
||||
text, (Timezone, Event, Todo, Journal, Card), name)
|
||||
for new_item in new_items.values():
|
||||
if new_item.name not in self.items:
|
||||
self.items[new_item.name] = new_item
|
||||
self.write()
|
||||
|
||||
def replace(self, name, text):
|
||||
"""Replace content by ``text`` in collection objet called ``name``."""
|
||||
self.remove(name)
|
||||
self.append(name, text)
|
||||
|
||||
def write(self):
|
||||
"""Write collection with given parameters."""
|
||||
text = serialize(self.tag, self.headers, self.items.values())
|
||||
self.save(text)
|
||||
|
||||
@property
|
||||
def tag(self):
|
||||
"""Type of the collection."""
|
||||
with self.props as props:
|
||||
if "tag" not in props:
|
||||
try:
|
||||
tag = open(self.path).readlines()[0][6:].rstrip()
|
||||
except IOError:
|
||||
if self.path.endswith((".vcf", "/carddav")):
|
||||
props["tag"] = "VADDRESSBOOK"
|
||||
else:
|
||||
props["tag"] = "VCALENDAR"
|
||||
else:
|
||||
if tag in ("VADDRESSBOOK", "VCARD"):
|
||||
props["tag"] = "VADDRESSBOOK"
|
||||
else:
|
||||
props["tag"] = "VCALENDAR"
|
||||
return props["tag"]
|
||||
|
||||
@property
|
||||
def mimetype(self):
|
||||
"""Mimetype of the collection."""
|
||||
if self.tag == "VADDRESSBOOK":
|
||||
return "text/vcard"
|
||||
elif self.tag == "VCALENDAR":
|
||||
return "text/calendar"
|
||||
|
||||
@property
|
||||
def resource_type(self):
|
||||
"""Resource type of the collection."""
|
||||
if self.tag == "VADDRESSBOOK":
|
||||
return "addressbook"
|
||||
elif self.tag == "VCALENDAR":
|
||||
return "calendar"
|
||||
def serialize(self):
|
||||
items = []
|
||||
for href in os.listdir(self._filesystem_path):
|
||||
path = os.path.join(self._filesystem_path, href)
|
||||
if os.path.isfile(path):
|
||||
with open(path, encoding=STORAGE_ENCODING) as fd:
|
||||
items.append(vobject.readOne(fd.read()))
|
||||
if self.get_meta("tag") == "VCALENDAR":
|
||||
collection = vobject.iCalendar()
|
||||
for item in items:
|
||||
for content in ("vevent", "vtodo", "vjournal"):
|
||||
if content in item.contents:
|
||||
collection.add(getattr(item, content))
|
||||
break
|
||||
return collection.serialize()
|
||||
elif self.get_meta("tag") == "VADDRESSBOOK":
|
||||
return "".join([item.serialize() for item in items])
|
||||
|
||||
@property
|
||||
def etag(self):
|
||||
"""Etag from collection."""
|
||||
etag = md5()
|
||||
etag.update(self.text.encode("utf-8"))
|
||||
return '"%s"' % etag.hexdigest()
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
"""Collection name."""
|
||||
with self.props as props:
|
||||
return props.get("D:displayname", self.path.split(os.path.sep)[-1])
|
||||
|
||||
@property
|
||||
def color(self):
|
||||
"""Collection color."""
|
||||
with self.props as props:
|
||||
if "ICAL:calendar-color" not in props:
|
||||
props["ICAL:calendar-color"] = "#%x" % randint(0, 255 ** 3 - 1)
|
||||
return props["ICAL:calendar-color"]
|
||||
|
||||
@property
|
||||
def items(self):
|
||||
"""Get list of all items in collection."""
|
||||
if self._items is None:
|
||||
self._items = self._parse(
|
||||
self.text, (Event, Todo, Journal, Card, Timezone))
|
||||
return self._items
|
||||
|
||||
@property
|
||||
def timezones(self):
|
||||
"""Get list of all timezones in collection."""
|
||||
return [
|
||||
item for item in self.items.values() if item.tag == Timezone.tag]
|
||||
|
||||
@property
|
||||
def components(self):
|
||||
"""Get list of all components in collection."""
|
||||
tags = [item_type.tag for item_type in (Event, Todo, Journal, Card)]
|
||||
return [item for item in self.items.values() if item.tag in tags]
|
||||
|
||||
@property
|
||||
def owner_url(self):
|
||||
"""Get the collection URL according to its owner."""
|
||||
return "/%s/" % self.owner if self.owner else None
|
||||
|
||||
@property
|
||||
def url(self):
|
||||
"""Get the standard collection URL."""
|
||||
return "%s/" % self.path
|
||||
|
||||
@property
|
||||
def version(self):
|
||||
"""Get the version of the collection type."""
|
||||
return "3.0" if self.tag == "VADDRESSBOOK" else "2.0"
|
||||
return get_etag(self.serialize())
|
||||
|
||||
@ -25,11 +25,14 @@ in them for XML requests (all but PUT).
|
||||
|
||||
"""
|
||||
|
||||
import posixpath
|
||||
import re
|
||||
import xml.etree.ElementTree as ET
|
||||
from collections import OrderedDict
|
||||
from urllib.parse import unquote, urlparse
|
||||
|
||||
import vobject
|
||||
|
||||
from . import client, config, storage
|
||||
|
||||
|
||||
@ -169,7 +172,7 @@ def delete(path, collection):
|
||||
collection.delete()
|
||||
else:
|
||||
# Remove an item from the collection
|
||||
collection.remove(name_from_path(path, collection))
|
||||
collection.delete(name_from_path(path, collection))
|
||||
|
||||
# Writing answer
|
||||
multistatus = ET.Element(_tag("D", "multistatus"))
|
||||
@ -230,14 +233,13 @@ def _propfind_response(path, item, props, user, write=False):
|
||||
"""Build and return a PROPFIND response."""
|
||||
is_collection = isinstance(item, storage.Collection)
|
||||
if is_collection:
|
||||
is_leaf = item.is_leaf(item.path)
|
||||
with item.props as properties:
|
||||
collection_props = properties
|
||||
# TODO: fix this
|
||||
is_leaf = bool(item.list())
|
||||
|
||||
response = ET.Element(_tag("D", "response"))
|
||||
|
||||
href = ET.Element(_tag("D", "href"))
|
||||
uri = item.url if is_collection else "%s/%s" % (path, item.name)
|
||||
uri = item.path if is_collection else "%s/%s" % (path, item.href)
|
||||
href.text = _href(uri.replace("//", "/"))
|
||||
response.append(href)
|
||||
|
||||
@ -255,7 +257,7 @@ def _propfind_response(path, item, props, user, write=False):
|
||||
element = ET.Element(tag)
|
||||
is404 = False
|
||||
if tag == _tag("D", "getetag"):
|
||||
element.text = item.etag
|
||||
element.text = storage.get_etag(item.serialize())
|
||||
elif tag == _tag("D", "principal-URL"):
|
||||
tag = ET.Element(_tag("D", "href"))
|
||||
tag.text = _href(path)
|
||||
@ -272,8 +274,9 @@ def _propfind_response(path, item, props, user, write=False):
|
||||
# pylint: disable=W0511
|
||||
human_tag = _tag_from_clark(tag)
|
||||
if is_collection and is_leaf:
|
||||
if human_tag in collection_props:
|
||||
components = collection_props[human_tag].split(",")
|
||||
meta = item.get_meta(human_tag)
|
||||
if meta:
|
||||
components = meta.split(",")
|
||||
else:
|
||||
components = ("VTODO", "VEVENT", "VJOURNAL")
|
||||
for component in components:
|
||||
@ -307,46 +310,56 @@ def _propfind_response(path, item, props, user, write=False):
|
||||
element.append(supported)
|
||||
elif is_collection:
|
||||
if tag == _tag("D", "getcontenttype"):
|
||||
element.text = item.mimetype
|
||||
element.text = storage.MIMETYPES[item.get_meta("tag")]
|
||||
elif tag == _tag("D", "resourcetype"):
|
||||
if item.is_principal:
|
||||
tag = ET.Element(_tag("D", "principal"))
|
||||
element.append(tag)
|
||||
if is_leaf or (
|
||||
not item.exists and item.resource_type):
|
||||
item_tag = item.get_meta("tag")
|
||||
if is_leaf or item_tag:
|
||||
# 2nd case happens when the collection is not stored yet,
|
||||
# but the resource type is guessed
|
||||
if item.resource_type == "addressbook":
|
||||
tag = ET.Element(_tag("CR", item.resource_type))
|
||||
else:
|
||||
tag = ET.Element(_tag("C", item.resource_type))
|
||||
element.append(tag)
|
||||
if item.get_meta("tag") == "VADDRESSBOOK":
|
||||
tag = ET.Element(_tag("CR", "addressbook"))
|
||||
element.append(tag)
|
||||
elif item.get_meta("tag") == "VCALENDAR":
|
||||
tag = ET.Element(_tag("C", "calendar"))
|
||||
element.append(tag)
|
||||
tag = ET.Element(_tag("D", "collection"))
|
||||
element.append(tag)
|
||||
elif is_leaf:
|
||||
if tag == _tag("D", "owner") and item.owner_url:
|
||||
element.text = item.owner_url
|
||||
if tag == _tag("D", "owner") and item.owner:
|
||||
element.text = "/%s/" % item.owner
|
||||
elif tag == _tag("CS", "getctag"):
|
||||
element.text = item.etag
|
||||
elif tag == _tag("C", "calendar-timezone"):
|
||||
element.text = storage.serialize(
|
||||
item.tag, item.headers, item.timezones)
|
||||
timezones = {}
|
||||
for event in item.list():
|
||||
if "vtimezone" in event.content:
|
||||
for timezone in event.vtimezone_list:
|
||||
timezones.add(timezone)
|
||||
collection = vobject.iCalendar()
|
||||
for timezone in timezones:
|
||||
collection.add(timezone)
|
||||
element.text = collection.serialize()
|
||||
elif tag == _tag("D", "displayname"):
|
||||
element.text = item.name
|
||||
element.text = item.get_meta("D:displayname") or item.path
|
||||
elif tag == _tag("ICAL", "calendar-color"):
|
||||
element.text = item.color
|
||||
element.text = item.get_meta("ICAL:calendar-color")
|
||||
else:
|
||||
human_tag = _tag_from_clark(tag)
|
||||
if human_tag in collection_props:
|
||||
element.text = collection_props[human_tag]
|
||||
meta = item.get_meta(human_tag)
|
||||
if meta:
|
||||
element.text = meta
|
||||
else:
|
||||
is404 = True
|
||||
else:
|
||||
is404 = True
|
||||
# Not for collections
|
||||
elif tag == _tag("D", "getcontenttype"):
|
||||
element.text = "%s; component=%s" % (
|
||||
item.mimetype, item.tag.lower())
|
||||
name = item.name.lower()
|
||||
mimetype = "text/vcard" if name == "vcard" else "text/calendar"
|
||||
element.text = "%s; component=%s" % (mimetype, name)
|
||||
elif tag == _tag("D", "resourcetype"):
|
||||
# resourcetype must be returned empty for non-collection elements
|
||||
pass
|
||||
@ -438,15 +451,18 @@ def proppatch(path, xml_request, collection):
|
||||
def put(path, ical_request, collection):
|
||||
"""Read PUT requests."""
|
||||
name = name_from_path(path, collection)
|
||||
if name in collection.items:
|
||||
# PUT is modifying an existing item
|
||||
collection.replace(name, ical_request)
|
||||
elif name:
|
||||
# PUT is adding a new item
|
||||
collection.append(name, ical_request)
|
||||
else:
|
||||
# PUT is replacing the whole collection
|
||||
collection.save(ical_request)
|
||||
items = list(vobject.readComponents(ical_request))
|
||||
if items:
|
||||
if collection.has(name):
|
||||
# PUT is modifying an existing item
|
||||
return collection.update(name, items[0])
|
||||
elif name:
|
||||
# PUT is adding a new item
|
||||
return collection.upload(name, items[0])
|
||||
else:
|
||||
# PUT is replacing the whole collection
|
||||
collection.delete()
|
||||
return storage.Collection.create_collection(path, items)
|
||||
|
||||
|
||||
def report(path, xml_request, collection):
|
||||
@ -481,7 +497,6 @@ def report(path, xml_request, collection):
|
||||
tag_filters = set(
|
||||
element.get("name") for element
|
||||
in root.findall(".//%s" % _tag("C", "comp-filter")))
|
||||
tag_filters.discard('VCALENDAR')
|
||||
else:
|
||||
hreferences = ()
|
||||
tag_filters = None
|
||||
@ -489,18 +504,15 @@ def report(path, xml_request, collection):
|
||||
# Writing answer
|
||||
multistatus = ET.Element(_tag("D", "multistatus"))
|
||||
|
||||
collection_tag = collection.tag
|
||||
collection_headers = collection.headers
|
||||
collection_timezones = collection.timezones
|
||||
|
||||
for hreference in hreferences:
|
||||
# Check if the reference is an item or a collection
|
||||
name = name_from_path(hreference, collection)
|
||||
|
||||
if name:
|
||||
# Reference is an item
|
||||
path = "/".join(hreference.split("/")[:-1]) + "/"
|
||||
try:
|
||||
items = [collection.items[name]]
|
||||
items = [collection.get(name)]
|
||||
except KeyError:
|
||||
multistatus.append(
|
||||
_item_response(hreference, found_item=False))
|
||||
@ -509,11 +521,10 @@ def report(path, xml_request, collection):
|
||||
else:
|
||||
# Reference is a collection
|
||||
path = hreference
|
||||
items = collection.components
|
||||
items = [collection.get(href) for href, etag in collection.list()]
|
||||
|
||||
for item in items:
|
||||
href = _href("%s/%s" % (path.rstrip("/"), item.name))
|
||||
if tag_filters and item.tag not in tag_filters:
|
||||
if tag_filters and item.name not in tag_filters:
|
||||
continue
|
||||
|
||||
found_props = []
|
||||
@ -525,22 +536,22 @@ def report(path, xml_request, collection):
|
||||
element.text = item.etag
|
||||
found_props.append(element)
|
||||
elif tag == _tag("D", "getcontenttype"):
|
||||
element.text = "%s; component=%s" % (
|
||||
item.mimetype, item.tag.lower())
|
||||
name = item.name.lower()
|
||||
mimetype = (
|
||||
"text/vcard" if name == "vcard" else "text/calendar")
|
||||
element.text = "%s; component=%s" % (mimetype, name)
|
||||
found_props.append(element)
|
||||
elif tag in (_tag("C", "calendar-data"),
|
||||
_tag("CR", "address-data")):
|
||||
if isinstance(item, storage.Component):
|
||||
element.text = storage.serialize(
|
||||
collection_tag, collection_headers,
|
||||
collection_timezones + [item])
|
||||
if isinstance(item, (storage.Item, storage.Collection)):
|
||||
element.text = item.serialize()
|
||||
found_props.append(element)
|
||||
else:
|
||||
not_found_props.append(element)
|
||||
|
||||
multistatus.append(_item_response(
|
||||
href, found_props=found_props, not_found_props=not_found_props,
|
||||
found_item=True))
|
||||
posixpath.join(hreference, item.href), found_props=found_props,
|
||||
not_found_props=not_found_props, found_item=True))
|
||||
|
||||
return _pretty_xml(multistatus)
|
||||
|
||||
|
||||
@ -117,7 +117,6 @@ class TestCustomStorageSystem(BaseRequests, BaseTest):
|
||||
radicale.config.set("storage", "type", "tests.custom.storage")
|
||||
from tests.custom import storage
|
||||
storage.FOLDER = self.colpath
|
||||
storage.GIT_REPOSITORY = None
|
||||
self.application = radicale.Application()
|
||||
|
||||
def teardown(self):
|
||||
|
||||
Loading…
Reference in New Issue
Block a user