diff --git a/radicale/tests/__init__.py b/radicale/tests/__init__.py
index 984aa1a..713028e 100644
--- a/radicale/tests/__init__.py
+++ b/radicale/tests/__init__.py
@@ -25,10 +25,11 @@ import os
import sys
from io import BytesIO
+import defusedxml.ElementTree as DefusedET
from pytest_cov import embed
import radicale
-from radicale import server
+from radicale import server, xmlutils
# Measure coverage of forked processes
finish_request = server.ParallelHTTPServer.finish_request
@@ -76,3 +77,103 @@ class BaseTest:
return (int(status.split()[0]), dict(headers),
answer[0].decode() if answer else None)
+
+ @staticmethod
+ def parse_responses(text):
+ xml = DefusedET.fromstring(text)
+ assert xml.tag == xmlutils.make_clark("D:multistatus")
+ path_responses = {}
+ for response in xml.findall(xmlutils.make_clark("D:response")):
+ href = response.find(xmlutils.make_clark("D:href"))
+ assert href.text not in path_responses
+ prop_respones = {}
+ for propstat in response.findall(
+ xmlutils.make_clark("D:propstat")):
+ status = propstat.find(xmlutils.make_clark("D:status"))
+ assert status.text.startswith("HTTP/1.1 ")
+ status_code = int(status.text.split(" ")[1])
+ for prop in propstat.findall(xmlutils.make_clark("D:prop")):
+ for element in prop:
+ human_tag = xmlutils.make_human_tag(element.tag)
+ assert human_tag not in prop_respones
+ prop_respones[human_tag] = (status_code, element)
+ status = response.find(xmlutils.make_clark("D:status"))
+ if status is not None:
+ assert not prop_respones
+ assert status.text.startswith("HTTP/1.1 ")
+ status_code = int(status.text.split(" ")[1])
+ path_responses[href.text] = status_code
+ else:
+ path_responses[href.text] = prop_respones
+ return path_responses
+
+ @staticmethod
+ def _check_status(status, good_status, check=True):
+ if check is not False:
+ assert status in (good_status, check)
+ return status == good_status
+
+ def get(self, path, check=True, **args):
+ status, _, answer = self.request("GET", path, **args)
+ self._check_status(status, 200, check)
+ return status, answer
+
+ def put(self, path, data, check=True, **args):
+ status, _, answer = self.request("PUT", path, data, **args)
+ self._check_status(status, 201, check)
+ return status
+
+ def propfind(self, path, data=None, check=True, **args):
+ status, _, answer = self.request("PROPFIND", path, data, **args)
+ if not self._check_status(status, 207, check):
+ return status, None
+ responses = self.parse_responses(answer)
+ if args.get("HTTP_DEPTH", 0) == 0:
+ assert len(responses) == 1 and path in responses
+ return status, responses
+
+ def proppatch(self, path, data=None, check=True, **args):
+ status, _, answer = self.request("PROPPATCH", path, data, **args)
+ if not self._check_status(status, 207, check):
+ return status, None
+ responses = self.parse_responses(answer)
+ assert len(responses) == 1 and path in responses
+ return status, responses
+
+ def report(self, path, data, check=True, **args):
+ status, _, answer = self.request("REPORT", path, data, **args)
+ if not self._check_status(status, 207, check):
+ return status, None
+ return status, self.parse_responses(answer)
+
+ def delete(self, path, check=True, **args):
+ status, _, answer = self.request("DELETE", path, **args)
+ if not self._check_status(status, 200, check):
+ return status, None
+ responses = self.parse_responses(answer)
+ assert len(responses) == 1 and path in responses
+ return status, responses
+
+ def mkcalendar(self, path, data=None, check=True, **args):
+ status, _, _ = self.request("MKCALENDAR", path, data, **args)
+ self._check_status(status, 201, check)
+ return status
+
+ def mkcol(self, path, data=None, check=True, **args):
+ status, _, _ = self.request("MKCOL", path, data, **args)
+ self._check_status(status, 201, check)
+ return status
+
+ def create_addressbook(self, path, check=True, **args):
+ return self.mkcol(path, """\
+
+
+
+
+
+
+
+
+
+
+""", check=check, **args)
diff --git a/radicale/tests/test_auth.py b/radicale/tests/test_auth.py
index bd8f112..635ab5e 100644
--- a/radicale/tests/test_auth.py
+++ b/radicale/tests/test_auth.py
@@ -28,8 +28,8 @@ import tempfile
import pytest
-from radicale import Application, config
-from radicale.tests.test_base import BaseTest
+from radicale import Application, config, xmlutils
+from radicale.tests import BaseTest
class TestBaseAuthRequests(BaseTest):
@@ -84,11 +84,10 @@ class TestBaseAuthRequests(BaseTest):
("😁", "🔑", False), ("😀", "", False),
("", "🔑", False), ("", "", False))
for user, password, valid in test_matrix:
- status, _, _ = self.request(
- "PROPFIND", "/",
- HTTP_AUTHORIZATION="Basic %s" % base64.b64encode(
- ("%s:%s" % (user, password)).encode()).decode())
- assert status == (207 if valid else 401)
+ status, _ = self.propfind(
+ "/", check=207 if valid else 401, HTTP_AUTHORIZATION=(
+ "Basic %s" % base64.b64encode(
+ ("%s:%s" % (user, password)).encode()).decode()))
def test_htpasswd_plain(self):
self._test_htpasswd("plain", "tmp:bepo")
@@ -136,38 +135,36 @@ class TestBaseAuthRequests(BaseTest):
def test_remote_user(self):
self.configuration.update({"auth": {"type": "remote_user"}}, "test")
self.application = Application(self.configuration)
- status, _, answer = self.request(
- "PROPFIND", "/",
- """
-
-
-
-
- """, REMOTE_USER="test")
- assert status == 207
- assert ">/test/<" in answer
+ _, responses = self.propfind("/", """\
+
+
+
+
+
+""", REMOTE_USER="test")
+ status, prop = responses["/"]["D:current-user-principal"]
+ assert status == 200
+ assert prop.find(xmlutils.make_clark("D:href")).text == "/test/"
def test_http_x_remote_user(self):
self.configuration.update(
{"auth": {"type": "http_x_remote_user"}}, "test")
self.application = Application(self.configuration)
- status, _, answer = self.request(
- "PROPFIND", "/",
- """
-
-
-
-
- """, HTTP_X_REMOTE_USER="test")
- assert status == 207
- assert ">/test/<" in answer
+ _, responses = self.propfind("/", """\
+
+
+
+
+
+""", HTTP_X_REMOTE_USER="test")
+ status, prop = responses["/"]["D:current-user-principal"]
+ assert status == 200
+ assert prop.find(xmlutils.make_clark("D:href")).text == "/test/"
def test_custom(self):
"""Custom authentication."""
self.configuration.update(
{"auth": {"type": "radicale.tests.custom.auth"}}, "test")
self.application = Application(self.configuration)
- status, _, _ = self.request(
- "PROPFIND", "/tmp", HTTP_AUTHORIZATION="Basic %s" %
- base64.b64encode(("tmp:").encode()).decode())
- assert status == 207
+ self.propfind("/tmp/", HTTP_AUTHORIZATION="Basic %s" %
+ base64.b64encode(("tmp:").encode()).decode())
diff --git a/radicale/tests/test_base.py b/radicale/tests/test_base.py
index ca69a3f..b460f6a 100644
--- a/radicale/tests/test_base.py
+++ b/radicale/tests/test_base.py
@@ -26,12 +26,11 @@ import posixpath
import shutil
import sys
import tempfile
-from functools import partial
import defusedxml.ElementTree as DefusedET
import pytest
-from radicale import Application, config, storage
+from radicale import Application, config, storage, xmlutils
from radicale.tests import BaseTest
from radicale.tests.helpers import get_file_content
@@ -44,27 +43,22 @@ class BaseRequestsMixIn:
def test_root(self):
"""GET request at "/"."""
- status, _, answer = self.request("GET", "/")
- assert status == 302
+ _, answer = self.get("/", check=302)
assert answer == "Redirected to .web"
def test_script_name(self):
"""GET request at "/" with SCRIPT_NAME."""
- status, _, answer = self.request("GET", "/", SCRIPT_NAME="/radicale")
- assert status == 302
+ _, answer = self.get("/", check=302, SCRIPT_NAME="/radicale")
assert answer == "Redirected to .web"
- status, _, answer = self.request("GET", "", SCRIPT_NAME="/radicale")
- assert status == 302
+ _, answer = self.get("", check=302, SCRIPT_NAME="/radicale")
assert answer == "Redirected to radicale/.web"
def test_add_event(self):
"""Add an event."""
- status, _, _ = self.request("MKCALENDAR", "/calendar.ics/")
- assert status == 201
+ self.mkcalendar("/calendar.ics/")
event = get_file_content("event1.ics")
path = "/calendar.ics/event1.ics"
- status, _, _ = self.request("PUT", path, event)
- assert status == 201
+ self.put(path, event)
status, headers, answer = self.request("GET", path)
assert status == 200
assert "ETag" in headers
@@ -75,22 +69,18 @@ class BaseRequestsMixIn:
def test_add_event_without_uid(self):
"""Add an event without UID."""
- status, _, _ = self.request("MKCALENDAR", "/calendar.ics/")
- assert status == 201
+ self.mkcalendar("/calendar.ics/")
event = get_file_content("event1.ics").replace("UID:event1\n", "")
assert "\nUID:" not in event
path = "/calendar.ics/event.ics"
- status, _, _ = self.request("PUT", path, event)
- assert status == 400
+ self.put(path, event, check=400)
def test_add_todo(self):
"""Add a todo."""
- status, _, _ = self.request("MKCALENDAR", "/calendar.ics/")
- assert status == 201
+ self.mkcalendar("/calendar.ics/")
todo = get_file_content("todo1.ics")
path = "/calendar.ics/todo1.ics"
- status, _, _ = self.request("PUT", path, todo)
- assert status == 201
+ self.put(path, todo)
status, headers, answer = self.request("GET", path)
assert status == 200
assert "ETag" in headers
@@ -99,58 +89,36 @@ class BaseRequestsMixIn:
assert "Todo" in answer
assert "UID:todo" in answer
- def _create_addressbook(self, path):
- return self.request(
- "MKCOL", path, """\
-
-
-
-
-
-
-
-
-
-
-""")
-
def test_add_contact(self):
"""Add a contact."""
- status, _, _ = self._create_addressbook("/contacts.vcf/")
- assert status == 201
+ self.create_addressbook("/contacts.vcf/")
contact = get_file_content("contact1.vcf")
path = "/contacts.vcf/contact.vcf"
- status, _, _ = self.request("PUT", path, contact)
- assert status == 201
+ self.put(path, contact)
status, headers, answer = self.request("GET", path)
assert status == 200
assert "ETag" in headers
assert headers["Content-Type"] == "text/vcard; charset=utf-8"
assert "VCARD" in answer
assert "UID:contact1" in answer
- status, _, answer = self.request("GET", path)
- assert status == 200
+ _, answer = self.get(path)
assert "UID:contact1" in answer
def test_add_contact_without_uid(self):
"""Add a contact without UID."""
- status, _, _ = self._create_addressbook("/contacts.vcf/")
- assert status == 201
+ self.create_addressbook("/contacts.vcf/")
contact = get_file_content("contact1.vcf").replace("UID:contact1\n",
"")
assert "\nUID" not in contact
path = "/contacts.vcf/contact.vcf"
- status, _, _ = self.request("PUT", path, contact)
- assert status == 400
+ self.put(path, contact, check=400)
def test_update(self):
"""Update an event."""
- status, _, _ = self.request("MKCALENDAR", "/calendar.ics/")
- assert status == 201
+ self.mkcalendar("/calendar.ics/")
event = get_file_content("event1.ics")
path = "/calendar.ics/event1.ics"
- status, _, _ = self.request("PUT", path, event)
- assert status == 201
+ self.put(path, event)
status, headers, answer = self.request("GET", path)
assert "ETag" in headers
assert status == 200
@@ -162,10 +130,8 @@ class BaseRequestsMixIn:
# Then we send another PUT request
event = get_file_content("event1-prime.ics")
- status, _, _ = self.request("PUT", path, event)
- assert status == 201
- status, _, answer = self.request("GET", "/calendar.ics/")
- assert status == 200
+ self.put(path, event)
+ _, answer = self.get("/calendar.ics/")
assert answer.count("BEGIN:VEVENT") == 1
status, headers, answer = self.request("GET", path)
@@ -181,21 +147,14 @@ class BaseRequestsMixIn:
def test_put_whole_calendar(self):
"""Create and overwrite a whole calendar."""
- status, _, _ = self.request(
- "PUT", "/calendar.ics/", "BEGIN:VCALENDAR\r\nEND:VCALENDAR")
- assert status == 201
+ self.put("/calendar.ics/", "BEGIN:VCALENDAR\r\nEND:VCALENDAR")
event1 = get_file_content("event1.ics")
- status, _, _ = self.request(
- "PUT", "/calendar.ics/test_event.ics", event1)
- assert status == 201
+ self.put("/calendar.ics/test_event.ics", event1)
# Overwrite
events = get_file_content("event_multiple.ics")
- status, _, _ = self.request("PUT", "/calendar.ics/", events)
- assert status == 201
- status, _, _ = self.request("GET", "/calendar.ics/test_event.ics")
- assert status == 404
- status, _, answer = self.request("GET", "/calendar.ics/")
- assert status == 200
+ self.put("/calendar.ics/", events)
+ self.get("/calendar.ics/test_event.ics", check=404)
+ _, answer = self.get("/calendar.ics/")
assert "\r\nUID:event\r\n" in answer and "\r\nUID:todo\r\n" in answer
assert "\r\nUID:event1\r\n" not in answer
@@ -204,10 +163,8 @@ class BaseRequestsMixIn:
event = get_file_content("event_multiple.ics")
event = event.replace("UID:event\n", "").replace("UID:todo\n", "")
assert "\nUID:" not in event
- status, _, _ = self.request("PUT", "/calendar.ics/", event)
- assert status == 201
- status, _, answer = self.request("GET", "/calendar.ics")
- assert status == 200
+ self.put("/calendar.ics/", event)
+ _, answer = self.get("/calendar.ics")
uids = []
for line in answer.split("\r\n"):
if line.startswith("UID:"):
@@ -221,10 +178,8 @@ class BaseRequestsMixIn:
def test_put_whole_addressbook(self):
"""Create and overwrite a whole addressbook."""
contacts = get_file_content("contact_multiple.vcf")
- status, _, _ = self.request("PUT", "/contacts.vcf/", contacts)
- assert status == 201
- status, _, answer = self.request("GET", "/contacts.vcf/")
- assert status == 200
+ self.put("/contacts.vcf/", contacts)
+ _, answer = self.get("/contacts.vcf/")
assert ("\r\nUID:contact1\r\n" in answer and
"\r\nUID:contact2\r\n" in answer)
@@ -234,10 +189,8 @@ class BaseRequestsMixIn:
contacts = contacts.replace("UID:contact1\n", "").replace(
"UID:contact2\n", "")
assert "\nUID:" not in contacts
- status, _, _ = self.request("PUT", "/contacts.vcf/", contacts)
- assert status == 201
- status, _, answer = self.request("GET", "/contacts.vcf")
- assert status == 200
+ self.put("/contacts.vcf/", contacts)
+ _, answer = self.get("/contacts.vcf")
uids = []
for line in answer.split("\r\n"):
if line.startswith("UID:"):
@@ -251,74 +204,57 @@ class BaseRequestsMixIn:
def test_verify(self):
"""Verify the storage."""
contacts = get_file_content("contact_multiple.vcf")
- status, _, _ = self.request("PUT", "/contacts.vcf/", contacts)
- assert status == 201
+ self.put("/contacts.vcf/", contacts)
events = get_file_content("event_multiple.ics")
- status, _, _ = self.request("PUT", "/calendar.ics/", events)
- assert status == 201
+ self.put("/calendar.ics/", events)
s = storage.load(self.configuration)
assert s.verify()
def test_delete(self):
"""Delete an event."""
- status, _, _ = self.request("MKCALENDAR", "/calendar.ics/")
- assert status == 201
+ self.mkcalendar("/calendar.ics/")
event = get_file_content("event1.ics")
path = "/calendar.ics/event1.ics"
- status, _, _ = self.request("PUT", path, event)
- assert status == 201
+ self.put(path, event)
# Then we send a DELETE request
- status, _, answer = self.request("DELETE", path)
- assert status == 200
- assert "href>%s/calendar.ics///%s%s%s" in answer
- status, _, answer = self.request(
- "PROPFIND", "/calendar.ics/event.ics", propfind)
- assert "" in answer
+ _, responses = self.propfind("/calendar.ics/", propfind)
+ status, prop = responses["/calendar.ics/"]["D:sync-token"]
+ assert status == 200 and not prop.text
+ _, responses = self.propfind("/calendar.ics/event.ics", propfind)
+ status, prop = responses["/calendar.ics/event.ics"]["D:getetag"]
+ assert status == 200 and not prop.text
def test_propfind_allprop(self):
- status, _, _ = self.request("MKCALENDAR", "/calendar.ics/")
- assert status == 201
+ self.mkcalendar("/calendar.ics/")
event = get_file_content("event1.ics")
- status, _, _ = self.request("PUT", "/calendar.ics/event.ics", event)
- assert status == 201
+ self.put("/calendar.ics/event.ics", event)
propfind = get_file_content("allprop.xml")
- status, _, answer = self.request(
- "PROPFIND", "/calendar.ics/", propfind)
- assert "" in answer
- status, _, answer = self.request(
- "PROPFIND", "/calendar.ics/event.ics", propfind)
- assert "" in answer
+ _, responses = self.propfind("/calendar.ics/", propfind)
+ status, prop = responses["/calendar.ics/"]["D:sync-token"]
+ assert status == 200 and prop.text
+ _, responses = self.propfind("/calendar.ics/event.ics", propfind)
+ status, prop = responses["/calendar.ics/event.ics"]["D:getetag"]
+ assert status == 200 and prop.text
+
+ def test_propfind_nonexistent(self):
+ """Read a property that does not exist."""
+ self.mkcalendar("/calendar.ics/")
+ propfind = get_file_content("propfind1.xml")
+ _, responses = self.propfind("/calendar.ics/", propfind)
+ assert len(responses["/calendar.ics/"]) == 1
+ status, prop = responses["/calendar.ics/"]["ICAL:calendar-color"]
+ assert status == 404 and not prop.text
def test_proppatch(self):
"""Write a property and read it back."""
- status, _, _ = self.request("MKCALENDAR", "/calendar.ics/")
- assert status == 201
+ self.mkcalendar("/calendar.ics/")
proppatch = get_file_content("proppatch1.xml")
- status, _, answer = self.request(
- "PROPPATCH", "/calendar.ics/", proppatch)
- assert status == 207
- assert "calendar-color" in answer
- assert "200 OK#BADA55" in answer
+ _, responses = self.propfind("/calendar.ics/", propfind)
+ status, prop = responses["/calendar.ics/"]["ICAL:calendar-color"]
+ assert status == 200 and prop.text == "#BADA55"
def test_put_whole_calendar_multiple_events_with_same_uid(self):
"""Add two events with the same UID."""
- status, _, _ = self.request(
- "PUT", "/calendar.ics/", get_file_content("event2.ics"))
- assert status == 201
- status, _, answer = self.request(
- "REPORT", "/calendar.ics/",
- """
-
-
- """)
- assert status == 207
- assert answer.count("") == 1
- status, _, answer = self.request("GET", "/calendar.ics/")
- assert status == 200
+ self.put("/calendar.ics/", get_file_content("event2.ics"))
+ _, responses = self.report("/calendar.ics/", """\
+
+
+
+
+
+""")
+ assert len(responses) == 1
+ status, prop = responses["/calendar.ics/event2.ics"]["D:getetag"]
+ assert status == 200 and prop.text
+ _, answer = self.get("/calendar.ics/")
assert answer.count("BEGIN:VEVENT") == 2
def _test_filter(self, filters, kind="event", test=None, items=(1,)):
filter_template = "%s"
if kind in ("event", "journal", "todo"):
- create_collection_fn = partial(self.request, "MKCALENDAR")
+ create_collection_fn = self.mkcalendar
path = "/calendar.ics/"
filename_template = "%s%d.ics"
namespace = "urn:ietf:params:xml:ns:caldav"
report = "calendar-query"
elif kind == "contact":
- create_collection_fn = self._create_addressbook
+ create_collection_fn = self.create_addressbook
if test:
filter_template = '%%s' % test
path = "/contacts.vcf/"
@@ -461,860 +384,824 @@ class BaseRequestsMixIn:
report = "addressbook-query"
else:
raise ValueError("Unsupported kind: %r" % kind)
- status, _, _ = self.request("DELETE", path)
+ status, _, = self.delete(path, check=False)
assert status in (200, 404)
- status, _, _ = create_collection_fn(path)
- assert status == 201
+ create_collection_fn(path)
for i in items:
filename = filename_template % (kind, i)
event = get_file_content(filename)
- status, _, _ = self.request(
- "PUT", posixpath.join(path, filename), event)
- assert status == 201
+ self.put(posixpath.join(path, filename), event)
filters_text = "".join(filter_template % f for f in filters)
- status, _, answer = self.request(
- "REPORT", path,
- """
-
-
-
-
- {2}
- """.format(namespace, report, filters_text))
- assert status == 207
- return answer
+ _, responses = self.report(path, """\
+
+
+
+
+
+ {2}
+""".format(namespace, report, filters_text))
+ paths = []
+ for path, props in responses.items():
+ assert len(props) == 1
+ status, prop = props["D:getetag"]
+ assert status == 200 and prop.text
+ paths.append(path)
+ return paths
def test_addressbook_empty_filter(self):
self._test_filter([""], kind="contact")
def test_addressbook_prop_filter(self):
- assert "href>/contacts.vcf/contact1.vcf
- es
- """], "contact")
- assert "href>/contacts.vcf/contact1.vcf
- es
- """], "contact")
- assert "href>/contacts.vcf/contact1.vcf
- a
- """], "contact")
- assert "href>/contacts.vcf/contact1.vcf
- test
- """], "contact")
- assert "href>/contacts.vcf/contact1.vcf
- tes
- """], "contact")
- assert "href>/contacts.vcf/contact1.vcf
- est
- """], "contact")
- assert "href>/contacts.vcf/contact1.vcf
- tes
- """], "contact")
- assert "href>/contacts.vcf/contact1.vcf
- est
- """], "contact")
- assert "href>/contacts.vcf/contact1.vcf
- est
- """], "contact")
- assert "href>/contacts.vcf/contact1.vcf
- tes
- """], "contact")
+ assert "/contacts.vcf/contact1.vcf" in self._test_filter(["""\
+
+ es
+"""], "contact")
+ assert "/contacts.vcf/contact1.vcf" in self._test_filter(["""\
+
+ es
+"""], "contact")
+ assert "/contacts.vcf/contact1.vcf" not in self._test_filter(["""\
+
+ a
+"""], "contact")
+ assert "/contacts.vcf/contact1.vcf" in self._test_filter(["""\
+
+ test
+"""], "contact")
+ assert "/contacts.vcf/contact1.vcf" not in self._test_filter(["""\
+
+ tes
+"""], "contact")
+ assert "/contacts.vcf/contact1.vcf" not in self._test_filter(["""\
+
+ est
+"""], "contact")
+ assert "/contacts.vcf/contact1.vcf" in self._test_filter(["""\
+
+ tes
+"""], "contact")
+ assert "/contacts.vcf/contact1.vcf" not in self._test_filter(["""\
+
+ est
+"""], "contact")
+ assert "/contacts.vcf/contact1.vcf" in self._test_filter(["""\
+
+ est
+"""], "contact")
+ assert "/contacts.vcf/contact1.vcf" not in self._test_filter(["""\
+
+ tes
+"""], "contact")
def test_addressbook_prop_filter_any(self):
- assert "href>/contacts.vcf/contact1.vcf
- test
-
-
- test
- """], "contact", test="anyof")
- assert "href>/contacts.vcf/contact1.vcf
- a
-
-
- test
- """], "contact", test="anyof")
- assert "href>/contacts.vcf/contact1.vcf
- test
-
-
- test
- """], "contact")
+ assert "/contacts.vcf/contact1.vcf" in self._test_filter(["""\
+
+ test
+
+
+ test
+"""], "contact", test="anyof")
+ assert "/contacts.vcf/contact1.vcf" not in self._test_filter(["""\
+
+ a
+
+
+ test
+"""], "contact", test="anyof")
+ assert "/contacts.vcf/contact1.vcf" in self._test_filter(["""\
+
+ test
+
+
+ test
+"""], "contact")
def test_addressbook_prop_filter_all(self):
- assert "href>/contacts.vcf/contact1.vcf
- tes
-
-
- est
- """], "contact", test="allof")
- assert "href>/contacts.vcf/contact1.vcf
- test
-
-
- test
- """], "contact", test="allof")
+ assert "/contacts.vcf/contact1.vcf" in self._test_filter(["""\
+
+ tes
+
+
+ est
+"""], "contact", test="allof")
+ assert "/contacts.vcf/contact1.vcf" not in self._test_filter(["""\
+
+ test
+
+
+ test
+"""], "contact", test="allof")
def test_calendar_empty_filter(self):
self._test_filter([""])
def test_calendar_tag_filter(self):
"""Report request with tag-based filter on calendar."""
- assert "href>/calendar.ics/event1.ics"""])
+ assert "/calendar.ics/event1.ics" in self._test_filter(["""\
+"""])
def test_item_tag_filter(self):
"""Report request with tag-based filter on an item."""
- assert "href>/calendar.ics/event1.ics
-
- """])
- assert "href>/calendar.ics/event1.ics
-
- """])
+ assert "/calendar.ics/event1.ics" in self._test_filter(["""\
+
+
+"""])
+ assert "/calendar.ics/event1.ics" not in self._test_filter(["""\
+
+
+"""])
def test_item_not_tag_filter(self):
"""Report request with tag-based is-not filter on an item."""
- assert "href>/calendar.ics/event1.ics
-
-
-
- """])
- assert "href>/calendar.ics/event1.ics
-
-
-
- """])
+ assert "/calendar.ics/event1.ics" not in self._test_filter(["""\
+
+
+
+
+"""])
+ assert "/calendar.ics/event1.ics" in self._test_filter(["""\
+
+
+
+
+"""])
def test_item_prop_filter(self):
"""Report request with prop-based filter on an item."""
- assert "href>/calendar.ics/event1.ics
-
-
-
- """])
- assert "href>/calendar.ics/event1.ics
-
-
-
- """])
+ assert "/calendar.ics/event1.ics" in self._test_filter(["""\
+
+
+
+
+"""])
+ assert "/calendar.ics/event1.ics" not in self._test_filter(["""\
+
+
+
+
+"""])
def test_item_not_prop_filter(self):
"""Report request with prop-based is-not filter on an item."""
- assert "href>/calendar.ics/event1.ics
-
-
-
-
-
- """])
- assert "href>/calendar.ics/event1.ics
-
-
-
-
-
- """])
+ assert "/calendar.ics/event1.ics" not in self._test_filter(["""\
+
+
+
+
+
+
+"""])
+ assert "/calendar.ics/event1.ics" in self._test_filter(["""\
+
+
+
+
+
+
+"""])
def test_mutiple_filters(self):
"""Report request with multiple filters on an item."""
- assert "href>/calendar.ics/event1.ics
-
-
-
-
-
- """, """
-
-
-
-
-
-
- """])
- assert "href>/calendar.ics/event1.ics
-
-
-
- """, """
-
-
-
-
-
-
- """])
- assert "href>/calendar.ics/event1.ics
-
-
-
-
-
-
- """])
+ assert "/calendar.ics/event1.ics" not in self._test_filter(["""\
+
+
+
+
+
+
+""", """
+
+
+
+
+
+
+"""])
+ assert "/calendar.ics/event1.ics" in self._test_filter(["""\
+
+
+
+
+""", """
+
+
+
+
+
+
+"""])
+ assert "/calendar.ics/event1.ics" in self._test_filter(["""\
+
+
+
+
+
+
+
+"""])
def test_text_match_filter(self):
"""Report request with text-match filter on calendar."""
- assert "href>/calendar.ics/event1.ics
-
-
- event
-
-
- """])
- assert "href>/calendar.ics/event1.ics
-
-
- event
-
-
- """])
- assert "href>/calendar.ics/event1.ics
-
-
- unknown
-
-
- """])
- assert "href>/calendar.ics/event1.ics
-
-
- event
-
-
- """])
+ assert "/calendar.ics/event1.ics" in self._test_filter(["""\
+
+
+
+ event
+
+
+"""])
+ assert "/calendar.ics/event1.ics" not in self._test_filter(["""\
+
+
+
+ event
+
+
+"""])
+ assert "/calendar.ics/event1.ics" not in self._test_filter(["""\
+
+
+
+ unknown
+
+
+"""])
+ assert "/calendar.ics/event1.ics" not in self._test_filter(["""\
+
+
+
+ event
+
+
+"""])
def test_param_filter(self):
"""Report request with param-filter on calendar."""
- assert "href>/calendar.ics/event1.ics
-
-
-
-
+
+
+
+ ACCEPTED
-
-
-
- """])
- assert "href>/calendar.ics/event1.ics
-
-
-
-
+
+
+"""])
+ assert "/calendar.ics/event1.ics" not in self._test_filter(["""\
+
+
+
+
+ UNKNOWN
-
-
-
- """])
- assert "href>/calendar.ics/event1.ics
-
-
-
-
-
-
-
- """])
- assert "href>/calendar.ics/event1.ics
-
-
-
-
-
-
-
- """])
+
+
+
+"""])
+ assert "/calendar.ics/event1.ics" not in self._test_filter(["""\
+
+
+
+
+
+
+
+
+"""])
+ assert "/calendar.ics/event1.ics" in self._test_filter(["""\
+
+
+
+
+
+
+
+
+"""])
def test_time_range_filter_events(self):
"""Report request with time-range filter on events."""
- answer = self._test_filter(["""
-
-
-
-
- """], "event", items=range(1, 6))
- assert "href>/calendar.ics/event1.ics/calendar.ics/event2.ics/calendar.ics/event3.ics/calendar.ics/event4.ics/calendar.ics/event5.ics
-
-
-
- """], "event", items=range(1, 6))
- assert "href>/calendar.ics/event1.ics
-
-
-
-
-
-
-
-
- """], items=range(1, 6))
- assert "href>/calendar.ics/event1.ics/calendar.ics/event2.ics/calendar.ics/event3.ics/calendar.ics/event4.ics/calendar.ics/event5.ics
-
-
-
- """], items=range(1, 6))
- assert "href>/calendar.ics/event1.ics/calendar.ics/event2.ics/calendar.ics/event3.ics/calendar.ics/event4.ics/calendar.ics/event5.ics
-
-
-
- """], items=range(1, 6))
- assert "href>/calendar.ics/event1.ics/calendar.ics/event2.ics/calendar.ics/event3.ics/calendar.ics/event4.ics/calendar.ics/event5.ics
-
-
-
- """], items=range(1, 6))
- assert "href>/calendar.ics/event1.ics/calendar.ics/event2.ics/calendar.ics/event3.ics/calendar.ics/event4.ics/calendar.ics/event5.ics
-
-
-
- """], items=range(1, 6))
- assert "href>/calendar.ics/event1.ics/calendar.ics/event2.ics/calendar.ics/event3.ics/calendar.ics/event4.ics/calendar.ics/event5.ics
+
+
+
+"""], "event", items=range(1, 6))
+ assert "/calendar.ics/event1.ics" in answer
+ assert "/calendar.ics/event2.ics" in answer
+ assert "/calendar.ics/event3.ics" in answer
+ assert "/calendar.ics/event4.ics" in answer
+ assert "/calendar.ics/event5.ics" in answer
+ answer = self._test_filter(["""\
+
+
+
+
+"""], "event", items=range(1, 6))
+ assert "/calendar.ics/event1.ics" not in answer
+ answer = self._test_filter(["""\
+
+
+
+
+
+
+
+
+
+"""], items=range(1, 6))
+ assert "/calendar.ics/event1.ics" not in answer
+ assert "/calendar.ics/event2.ics" not in answer
+ assert "/calendar.ics/event3.ics" not in answer
+ assert "/calendar.ics/event4.ics" not in answer
+ assert "/calendar.ics/event5.ics" not in answer
+ answer = self._test_filter(["""\
+
+
+
+
+"""], items=range(1, 6))
+ assert "/calendar.ics/event1.ics" not in answer
+ assert "/calendar.ics/event2.ics" in answer
+ assert "/calendar.ics/event3.ics" in answer
+ assert "/calendar.ics/event4.ics" in answer
+ assert "/calendar.ics/event5.ics" in answer
+ answer = self._test_filter(["""\
+
+
+
+
+"""], items=range(1, 6))
+ assert "/calendar.ics/event1.ics" not in answer
+ assert "/calendar.ics/event2.ics" not in answer
+ assert "/calendar.ics/event3.ics" in answer
+ assert "/calendar.ics/event4.ics" in answer
+ assert "/calendar.ics/event5.ics" in answer
+ answer = self._test_filter(["""\
+
+
+
+
+"""], items=range(1, 6))
+ assert "/calendar.ics/event1.ics" not in answer
+ assert "/calendar.ics/event2.ics" not in answer
+ assert "/calendar.ics/event3.ics" in answer
+ assert "/calendar.ics/event4.ics" not in answer
+ assert "/calendar.ics/event5.ics" not in answer
+ answer = self._test_filter(["""\
+
+
+
+
+"""], items=range(1, 6))
+ assert "/calendar.ics/event1.ics" not in answer
+ assert "/calendar.ics/event2.ics" not in answer
+ assert "/calendar.ics/event3.ics" not in answer
+ assert "/calendar.ics/event4.ics" not in answer
+ assert "/calendar.ics/event5.ics" not in answer
# HACK: VObject doesn't match RECURRENCE-ID to recurrences, the
# overwritten recurrence is still used for filtering.
- answer = self._test_filter(["""
-
-
-
-
- """], items=(6, 7, 8, 9))
- assert "href>/calendar.ics/event6.ics/calendar.ics/event7.ics/calendar.ics/event8.ics/calendar.ics/event9.ics
-
-
-
- """], items=(6, 7, 8, 9))
- assert "href>/calendar.ics/event6.ics/calendar.ics/event7.ics/calendar.ics/event8.ics/calendar.ics/event9.ics
-
-
-
- """], items=(6, 7, 8, 9))
- assert "href>/calendar.ics/event6.ics/calendar.ics/event7.ics/calendar.ics/event8.ics/calendar.ics/event9.ics
-
-
-
- """], items=(9,))
- assert "href>/calendar.ics/event9.ics
-
-
-
- """], items=(9,))
- assert "href>/calendar.ics/event9.ics
+
+
+
+"""], items=(6, 7, 8, 9))
+ assert "/calendar.ics/event6.ics" in answer
+ assert "/calendar.ics/event7.ics" in answer
+ assert "/calendar.ics/event8.ics" in answer
+ assert "/calendar.ics/event9.ics" in answer
+ answer = self._test_filter(["""\
+
+
+
+
+"""], items=(6, 7, 8, 9))
+ assert "/calendar.ics/event6.ics" in answer
+ assert "/calendar.ics/event7.ics" in answer
+ assert "/calendar.ics/event8.ics" in answer
+ assert "/calendar.ics/event9.ics" not in answer
+ answer = self._test_filter(["""\
+
+
+
+
+"""], items=(6, 7, 8, 9))
+ assert "/calendar.ics/event6.ics" not in answer
+ assert "/calendar.ics/event7.ics" not in answer
+ assert "/calendar.ics/event8.ics" not in answer
+ assert "/calendar.ics/event9.ics" not in answer
+ answer = self._test_filter(["""\
+
+
+
+
+"""], items=(9,))
+ assert "/calendar.ics/event9.ics" in answer
+ answer = self._test_filter(["""\
+
+
+
+
+"""], items=(9,))
+ assert "/calendar.ics/event9.ics" not in answer
def test_time_range_filter_events_rrule(self):
"""Report request with time-range filter on events with rrules."""
- answer = self._test_filter(["""
-
-
-
-
- """], "event", items=(1, 2))
- assert "href>/calendar.ics/event1.ics/calendar.ics/event2.ics
-
-
-
- """], "event", items=(1, 2))
- assert "href>/calendar.ics/event1.ics/calendar.ics/event2.ics
-
-
-
- """], "event", items=(1, 2))
- assert "href>/calendar.ics/event1.ics/calendar.ics/event2.ics
-
-
-
- """], "event", items=(1, 2))
- assert "href>/calendar.ics/event1.ics/calendar.ics/event2.ics
+
+
+
+"""], "event", items=(1, 2))
+ assert "/calendar.ics/event1.ics" in answer
+ assert "/calendar.ics/event2.ics" in answer
+ answer = self._test_filter(["""\
+
+
+
+
+"""], "event", items=(1, 2))
+ assert "/calendar.ics/event1.ics" not in answer
+ assert "/calendar.ics/event2.ics" in answer
+ answer = self._test_filter(["""\
+
+
+
+
+"""], "event", items=(1, 2))
+ assert "/calendar.ics/event1.ics" not in answer
+ assert "/calendar.ics/event2.ics" not in answer
+ answer = self._test_filter(["""\
+
+
+
+
+"""], "event", items=(1, 2))
+ assert "/calendar.ics/event1.ics" not in answer
+ assert "/calendar.ics/event2.ics" not in answer
def test_time_range_filter_todos(self):
"""Report request with time-range filter on todos."""
- answer = self._test_filter(["""
-
-
-
-
- """], "todo", items=range(1, 9))
- assert "href>/calendar.ics/todo1.ics/calendar.ics/todo2.ics/calendar.ics/todo3.ics/calendar.ics/todo4.ics/calendar.ics/todo5.ics/calendar.ics/todo6.ics/calendar.ics/todo7.ics/calendar.ics/todo8.ics
-
-
-
- """], "todo", items=range(1, 9))
- assert "href>/calendar.ics/todo1.ics/calendar.ics/todo2.ics/calendar.ics/todo3.ics/calendar.ics/todo4.ics/calendar.ics/todo5.ics/calendar.ics/todo6.ics/calendar.ics/todo7.ics/calendar.ics/todo8.ics
-
-
-
- """], "todo", items=range(1, 9))
- assert "href>/calendar.ics/todo2.ics
-
-
-
- """], "todo", items=range(1, 9))
- assert "href>/calendar.ics/todo2.ics
-
-
-
- """], "todo", items=range(1, 9))
- assert "href>/calendar.ics/todo3.ics
-
-
-
- """], "todo", items=range(1, 9))
- assert "href>/calendar.ics/todo7.ics
+
+
+
+"""], "todo", items=range(1, 9))
+ assert "/calendar.ics/todo1.ics" in answer
+ assert "/calendar.ics/todo2.ics" in answer
+ assert "/calendar.ics/todo3.ics" in answer
+ assert "/calendar.ics/todo4.ics" in answer
+ assert "/calendar.ics/todo5.ics" in answer
+ assert "/calendar.ics/todo6.ics" in answer
+ assert "/calendar.ics/todo7.ics" in answer
+ assert "/calendar.ics/todo8.ics" in answer
+ answer = self._test_filter(["""\
+
+
+
+
+"""], "todo", items=range(1, 9))
+ assert "/calendar.ics/todo1.ics" not in answer
+ assert "/calendar.ics/todo2.ics" in answer
+ assert "/calendar.ics/todo3.ics" in answer
+ assert "/calendar.ics/todo4.ics" not in answer
+ assert "/calendar.ics/todo5.ics" not in answer
+ assert "/calendar.ics/todo6.ics" not in answer
+ assert "/calendar.ics/todo7.ics" in answer
+ assert "/calendar.ics/todo8.ics" in answer
+ answer = self._test_filter(["""\
+
+
+
+
+"""], "todo", items=range(1, 9))
+ assert "/calendar.ics/todo2.ics" not in answer
+ answer = self._test_filter(["""\
+
+
+
+
+"""], "todo", items=range(1, 9))
+ assert "/calendar.ics/todo2.ics" not in answer
+ answer = self._test_filter(["""\
+
+
+
+
+"""], "todo", items=range(1, 9))
+ assert "/calendar.ics/todo3.ics" not in answer
+ answer = self._test_filter(["""\
+
+
+
+
+"""], "todo", items=range(1, 9))
+ assert "/calendar.ics/todo7.ics" in answer
def test_time_range_filter_todos_rrule(self):
"""Report request with time-range filter on todos with rrules."""
- answer = self._test_filter(["""
-
-
-
-
- """], "todo", items=(1, 2, 9))
- assert "href>/calendar.ics/todo1.ics/calendar.ics/todo2.ics/calendar.ics/todo9.ics
-
-
-
- """], "todo", items=(1, 2, 9))
- assert "href>/calendar.ics/todo1.ics/calendar.ics/todo2.ics/calendar.ics/todo9.ics
-
-
-
- """], "todo", items=(1, 2))
- assert "href>/calendar.ics/todo1.ics/calendar.ics/todo2.ics
-
-
-
- """], "todo", items=(1, 2))
- assert "href>/calendar.ics/todo1.ics/calendar.ics/todo2.ics
-
-
-
- """], "todo", items=(9,))
- assert "href>/calendar.ics/todo9.ics
+
+
+
+"""], "todo", items=(1, 2, 9))
+ assert "/calendar.ics/todo1.ics" in answer
+ assert "/calendar.ics/todo2.ics" in answer
+ assert "/calendar.ics/todo9.ics" in answer
+ answer = self._test_filter(["""\
+
+
+
+
+"""], "todo", items=(1, 2, 9))
+ assert "/calendar.ics/todo1.ics" not in answer
+ assert "/calendar.ics/todo2.ics" in answer
+ assert "/calendar.ics/todo9.ics" in answer
+ answer = self._test_filter(["""\
+
+
+
+
+"""], "todo", items=(1, 2))
+ assert "/calendar.ics/todo1.ics" not in answer
+ assert "/calendar.ics/todo2.ics" in answer
+ answer = self._test_filter(["""\
+
+
+
+
+"""], "todo", items=(1, 2))
+ assert "/calendar.ics/todo1.ics" not in answer
+ assert "/calendar.ics/todo2.ics" not in answer
+ answer = self._test_filter(["""\
+
+
+
+
+"""], "todo", items=(9,))
+ assert "/calendar.ics/todo9.ics" not in answer
def test_time_range_filter_journals(self):
"""Report request with time-range filter on journals."""
- answer = self._test_filter(["""
-
-
-
-
- """], "journal", items=(1, 2, 3))
- assert "href>/calendar.ics/journal1.ics/calendar.ics/journal2.ics/calendar.ics/journal3.ics
-
-
-
- """], "journal", items=(1, 2, 3))
- assert "href>/calendar.ics/journal1.ics/calendar.ics/journal2.ics/calendar.ics/journal3.ics
-
-
-
- """], "journal", items=(1, 2, 3))
- assert "href>/calendar.ics/journal1.ics/calendar.ics/journal2.ics/calendar.ics/journal3.ics
-
-
-
- """], "journal", items=(1, 2, 3))
- assert "href>/calendar.ics/journal1.ics/calendar.ics/journal2.ics/calendar.ics/journal3.ics
-
-
-
- """], "journal", items=(1, 2, 3))
- assert "href>/calendar.ics/journal1.ics/calendar.ics/journal2.ics/calendar.ics/journal3.ics
+
+
+
+"""], "journal", items=(1, 2, 3))
+ assert "/calendar.ics/journal1.ics" not in answer
+ assert "/calendar.ics/journal2.ics" in answer
+ assert "/calendar.ics/journal3.ics" in answer
+ answer = self._test_filter(["""\
+
+
+
+
+"""], "journal", items=(1, 2, 3))
+ assert "/calendar.ics/journal1.ics" not in answer
+ assert "/calendar.ics/journal2.ics" in answer
+ assert "/calendar.ics/journal3.ics" in answer
+ answer = self._test_filter(["""\
+
+
+
+
+"""], "journal", items=(1, 2, 3))
+ assert "/calendar.ics/journal1.ics" not in answer
+ assert "/calendar.ics/journal2.ics" not in answer
+ assert "/calendar.ics/journal3.ics" not in answer
+ answer = self._test_filter(["""\
+
+
+
+
+"""], "journal", items=(1, 2, 3))
+ assert "/calendar.ics/journal1.ics" not in answer
+ assert "/calendar.ics/journal2.ics" in answer
+ assert "/calendar.ics/journal3.ics" not in answer
+ answer = self._test_filter(["""\
+
+
+
+
+"""], "journal", items=(1, 2, 3))
+ assert "/calendar.ics/journal1.ics" not in answer
+ assert "/calendar.ics/journal2.ics" in answer
+ assert "/calendar.ics/journal3.ics" in answer
def test_time_range_filter_journals_rrule(self):
"""Report request with time-range filter on journals with rrules."""
- answer = self._test_filter(["""
-
-
-
-
- """], "journal", items=(1, 2))
- assert "href>/calendar.ics/journal1.ics/calendar.ics/journal2.ics
-
-
-
- """], "journal", items=(1, 2))
- assert "href>/calendar.ics/journal1.ics/calendar.ics/journal2.ics
-
-
-
- """], "journal", items=(1, 2))
- assert "href>/calendar.ics/journal1.ics/calendar.ics/journal2.ics
+
+
+
+"""], "journal", items=(1, 2))
+ assert "/calendar.ics/journal1.ics" not in answer
+ assert "/calendar.ics/journal2.ics" in answer
+ answer = self._test_filter(["""\
+
+
+
+
+"""], "journal", items=(1, 2))
+ assert "/calendar.ics/journal1.ics" not in answer
+ assert "/calendar.ics/journal2.ics" in answer
+ answer = self._test_filter(["""\
+
+
+
+
+"""], "journal", items=(1, 2))
+ assert "/calendar.ics/journal1.ics" not in answer
+ assert "/calendar.ics/journal2.ics" not in answer
def test_report_item(self):
"""Test report request on an item"""
calendar_path = "/calendar.ics/"
- status, _, _ = self.request("MKCALENDAR", calendar_path)
- assert status == 201
+ self.mkcalendar(calendar_path)
event = get_file_content("event1.ics")
event_path = posixpath.join(calendar_path, "event.ics")
- status, _, _ = self.request("PUT", event_path, event)
- assert status == 201
- status, _, answer = self.request(
- "REPORT", event_path,
- """
-
-
-
-
- """)
- assert status == 207
- assert "href>%s<" % event_path in answer
+ self.put(event_path, event)
+ _, responses = self.report(event_path, """\
+
+
+
+
+
+""")
+ assert len(responses) == 1
+ status, prop = responses[event_path]["D:getetag"]
+ assert status == 200 and prop.text
def _report_sync_token(self, calendar_path, sync_token=None):
sync_token_xml = (
"" % sync_token
if sync_token else "")
- status, _, answer = self.request(
- "REPORT", calendar_path,
- """
-
-
-
-
- %s
- """ % sync_token_xml)
+ status, _, answer = self.request("REPORT", calendar_path, """\
+
+
+
+
+
+ %s
+""" % sync_token_xml)
if sync_token and status == 409:
return None, None
assert status == 207
xml = DefusedET.fromstring(answer)
- sync_token = xml.find("{DAV:}sync-token").text.strip()
+ assert xml.tag == xmlutils.make_clark("D:multistatus")
+ sync_token = xml.find(xmlutils.make_clark("D:sync-token")).text.strip()
assert sync_token
- return sync_token, xml
+ responses = self.parse_responses(answer)
+ for href, response in responses.items():
+ if not isinstance(response, int):
+ status, prop = response["D:getetag"]
+ assert status == 200 and prop.text and len(response) == 1
+ responses[href] = response = 200
+ assert response in (200, 404)
+ return sync_token, responses
def test_report_sync_collection_no_change(self):
"""Test sync-collection report without modifying the collection"""
calendar_path = "/calendar.ics/"
- status, _, _ = self.request("MKCALENDAR", calendar_path)
- assert status == 201
+ self.mkcalendar(calendar_path)
event = get_file_content("event1.ics")
event_path = posixpath.join(calendar_path, "event.ics")
- status, _, _ = self.request("PUT", event_path, event)
- assert status == 201
- sync_token, xml = self._report_sync_token(calendar_path)
- assert xml.find("{DAV:}response") is not None
- new_sync_token, xml = self._report_sync_token(calendar_path,
- sync_token)
+ self.put(event_path, event)
+ sync_token, responses = self._report_sync_token(calendar_path)
+ assert len(responses) == 1 and responses[event_path] == 200
+ new_sync_token, responses = self._report_sync_token(
+ calendar_path, sync_token)
if not self.full_sync_token_support and not new_sync_token:
return
- assert sync_token == new_sync_token
- assert xml.find("{DAV:}response") is None
+ assert sync_token == new_sync_token and len(responses) == 0
def test_report_sync_collection_add(self):
"""Test sync-collection report with an added item"""
calendar_path = "/calendar.ics/"
- status, _, _ = self.request("MKCALENDAR", calendar_path)
- assert status == 201
- sync_token, xml = self._report_sync_token(calendar_path)
+ self.mkcalendar(calendar_path)
+ sync_token, responses = self._report_sync_token(calendar_path)
+ assert len(responses) == 0
event = get_file_content("event1.ics")
event_path = posixpath.join(calendar_path, "event.ics")
- status, _, _ = self.request("PUT", event_path, event)
- assert status == 201
- sync_token, xml = self._report_sync_token(calendar_path, sync_token)
+ self.put(event_path, event)
+ sync_token, responses = self._report_sync_token(
+ calendar_path, sync_token)
if not self.full_sync_token_support and not sync_token:
return
- assert xml.find("{DAV:}response") is not None
- assert xml.find("{DAV:}response/{DAV:}status") is None
+ assert len(responses) == 1 and responses[event_path] == 200
def test_report_sync_collection_delete(self):
"""Test sync-collection report with a deleted item"""
calendar_path = "/calendar.ics/"
- status, _, _ = self.request("MKCALENDAR", calendar_path)
- assert status == 201
+ self.mkcalendar(calendar_path)
event = get_file_content("event1.ics")
event_path = posixpath.join(calendar_path, "event.ics")
- status, _, _ = self.request("PUT", event_path, event)
- assert status == 201
- sync_token, xml = self._report_sync_token(calendar_path)
- status, _, _ = self.request("DELETE", event_path)
- assert status == 200
- sync_token, xml = self._report_sync_token(calendar_path, sync_token)
+ self.put(event_path, event)
+ sync_token, responses = self._report_sync_token(calendar_path)
+ assert len(responses) == 1 and responses[event_path] == 200
+ self.delete(event_path)
+ sync_token, responses = self._report_sync_token(
+ calendar_path, sync_token)
if not self.full_sync_token_support and not sync_token:
return
- assert "404" in xml.find("{DAV:}response/{DAV:}status").text
+ assert len(responses) == 1 and responses[event_path] == 404
def test_report_sync_collection_create_delete(self):
"""Test sync-collection report with a created and deleted item"""
calendar_path = "/calendar.ics/"
- status, _, _ = self.request("MKCALENDAR", calendar_path)
- assert status == 201
- sync_token, xml = self._report_sync_token(calendar_path)
+ self.mkcalendar(calendar_path)
+ sync_token, responses = self._report_sync_token(calendar_path)
+ assert len(responses) == 0
event = get_file_content("event1.ics")
event_path = posixpath.join(calendar_path, "event.ics")
- status, _, _ = self.request("PUT", event_path, event)
- assert status == 201
- status, _, _ = self.request("DELETE", event_path)
- assert status == 200
- sync_token, xml = self._report_sync_token(calendar_path, sync_token)
+ self.put(event_path, event)
+ self.delete(event_path)
+ sync_token, responses = self._report_sync_token(
+ calendar_path, sync_token)
if not self.full_sync_token_support and not sync_token:
return
- assert "404" in xml.find("{DAV:}response/{DAV:}status").text
+ assert len(responses) == 1 and responses[event_path] == 404
def test_report_sync_collection_modify_undo(self):
"""Test sync-collection report with a modified and changed back item"""
calendar_path = "/calendar.ics/"
- status, _, _ = self.request("MKCALENDAR", calendar_path)
- assert status == 201
+ self.mkcalendar(calendar_path)
event1 = get_file_content("event1.ics")
event2 = get_file_content("event1_modified.ics")
- event_path = posixpath.join(calendar_path, "event1.ics")
- status, _, _ = self.request("PUT", event_path, event1)
- assert status == 201
- sync_token, xml = self._report_sync_token(calendar_path)
- status, _, _ = self.request("PUT", event_path, event2)
- assert status == 201
- status, _, _ = self.request("PUT", event_path, event1)
- assert status == 201
- sync_token, xml = self._report_sync_token(calendar_path, sync_token)
+ event_path = posixpath.join(calendar_path, "event.ics")
+ self.put(event_path, event1)
+ sync_token, responses = self._report_sync_token(calendar_path)
+ assert len(responses) == 1 and responses[event_path] == 200
+ self.put(event_path, event2)
+ self.put(event_path, event1)
+ sync_token, responses = self._report_sync_token(
+ calendar_path, sync_token)
if not self.full_sync_token_support and not sync_token:
return
- assert xml.find("{DAV:}response") is not None
- assert xml.find("{DAV:}response/{DAV:}status") is None
+ assert len(responses) == 1 and responses[event_path] == 200
def test_report_sync_collection_move(self):
"""Test sync-collection report a moved item"""
calendar_path = "/calendar.ics/"
- status, _, _ = self.request("MKCALENDAR", calendar_path)
- assert status == 201
+ self.mkcalendar(calendar_path)
event = get_file_content("event1.ics")
event1_path = posixpath.join(calendar_path, "event1.ics")
event2_path = posixpath.join(calendar_path, "event2.ics")
- status, _, _ = self.request("PUT", event1_path, event)
- assert status == 201
- sync_token, xml = self._report_sync_token(calendar_path)
+ self.put(event1_path, event)
+ sync_token, responses = self._report_sync_token(calendar_path)
+ assert len(responses) == 1 and responses[event1_path] == 200
status, _, _ = self.request(
"MOVE", event1_path, HTTP_DESTINATION=event2_path, HTTP_HOST="")
assert status == 201
- sync_token, xml = self._report_sync_token(calendar_path, sync_token)
+ sync_token, responses = self._report_sync_token(
+ calendar_path, sync_token)
if not self.full_sync_token_support and not sync_token:
return
- for response in xml.findall("{DAV:}response"):
- if response.find("{DAV:}status") is None:
- assert response.find("{DAV:}href").text == event2_path
- else:
- assert "404" in response.find("{DAV:}status").text
- assert response.find("{DAV:}href").text == event1_path
+ assert len(responses) == 2 and (responses[event1_path] == 404 and
+ responses[event2_path] == 200)
def test_report_sync_collection_move_undo(self):
"""Test sync-collection report with a moved and moved back item"""
calendar_path = "/calendar.ics/"
- status, _, _ = self.request("MKCALENDAR", calendar_path)
- assert status == 201
+ self.mkcalendar(calendar_path)
event = get_file_content("event1.ics")
event1_path = posixpath.join(calendar_path, "event1.ics")
event2_path = posixpath.join(calendar_path, "event2.ics")
- status, _, _ = self.request("PUT", event1_path, event)
- assert status == 201
- sync_token, xml = self._report_sync_token(calendar_path)
+ self.put(event1_path, event)
+ sync_token, responses = self._report_sync_token(calendar_path)
+ assert len(responses) == 1 and responses[event1_path] == 200
status, _, _ = self.request(
"MOVE", event1_path, HTTP_DESTINATION=event2_path, HTTP_HOST="")
assert status == 201
status, _, _ = self.request(
"MOVE", event2_path, HTTP_DESTINATION=event1_path, HTTP_HOST="")
assert status == 201
- sync_token, xml = self._report_sync_token(calendar_path, sync_token)
+ sync_token, responses = self._report_sync_token(
+ calendar_path, sync_token)
if not self.full_sync_token_support and not sync_token:
return
- created = deleted = 0
- for response in xml.findall("{DAV:}response"):
- if response.find("{DAV:}status") is None:
- assert response.find("{DAV:}href").text == event1_path
- created += 1
- else:
- assert "404" in response.find("{DAV:}status").text
- assert response.find("{DAV:}href").text == event2_path
- deleted += 1
- assert created == 1 and deleted == 1
+ assert len(responses) == 2 and (responses[event1_path] == 200 and
+ responses[event2_path] == 404)
def test_report_sync_collection_invalid_sync_token(self):
"""Test sync-collection report with an invalid sync token"""
calendar_path = "/calendar.ics/"
- status, _, _ = self.request("MKCALENDAR", calendar_path)
- assert status == 201
+ self.mkcalendar(calendar_path)
sync_token, _ = self._report_sync_token(
calendar_path, "http://radicale.org/ns/sync/INVALID")
assert not sync_token
@@ -1322,80 +1209,81 @@ class BaseRequestsMixIn:
def test_propfind_sync_token(self):
"""Retrieve the sync-token with a propfind request"""
calendar_path = "/calendar.ics/"
- status, _, _ = self.request("MKCALENDAR", calendar_path)
- assert status == 201
- sync_token, _ = self._report_sync_token(calendar_path)
+ self.mkcalendar(calendar_path)
+ propfind = get_file_content("allprop.xml")
+ _, responses = self.propfind(calendar_path, propfind)
+ status, sync_token = responses[calendar_path]["D:sync-token"]
+ assert status == 200 and sync_token.text
event = get_file_content("event1.ics")
event_path = posixpath.join(calendar_path, "event.ics")
- status, _, _ = self.request("PUT", event_path, event)
- assert status == 201
- new_sync_token, _ = self._report_sync_token(calendar_path, sync_token)
- assert sync_token != new_sync_token
+ self.put(event_path, event)
+ _, responses = self.propfind(calendar_path, propfind)
+ status, new_sync_token = responses[calendar_path]["D:sync-token"]
+ assert status == 200 and new_sync_token.text
+ assert sync_token.text != new_sync_token.text
def test_propfind_same_as_sync_collection_sync_token(self):
"""Compare sync-token property with sync-collection sync-token"""
calendar_path = "/calendar.ics/"
- status, _, _ = self.request("MKCALENDAR", calendar_path)
- assert status == 201
- sync_token, _ = self._report_sync_token(calendar_path)
- new_sync_token, _ = self._report_sync_token(calendar_path, sync_token)
- if not self.full_sync_token_support and not new_sync_token:
- return
- assert sync_token == new_sync_token
+ self.mkcalendar(calendar_path)
+ propfind = get_file_content("allprop.xml")
+ _, responses = self.propfind(calendar_path, propfind)
+ status, sync_token = responses[calendar_path]["D:sync-token"]
+ assert status == 200 and sync_token.text
+ report_sync_token, _ = self._report_sync_token(calendar_path)
+ assert sync_token.text == report_sync_token
def test_calendar_getcontenttype(self):
"""Test report request on an item"""
- status, _, _ = self.request("MKCALENDAR", "/test/")
- assert status == 201
+ self.mkcalendar("/test/")
for component in ("event", "todo", "journal"):
event = get_file_content("%s1.ics" % component)
- status, _, _ = self.request("DELETE", "/test/test.ics")
+ status, _ = self.delete("/test/test.ics", check=False)
assert status in (200, 404)
- status, _, _ = self.request("PUT", "/test/test.ics", event)
- assert status == 201
- status, _, answer = self.request(
- "REPORT", "/test/",
- """
-
-
-
-
- """)
- assert status == 207
- assert ">text/calendar;charset=utf-8;component=V%s<" % (
- component.upper()) in answer
+ self.put("/test/test.ics", event)
+ _, responses = self.report("/test/", """\
+
+
+
+
+
+""")
+ assert len(responses) == 1 and len(
+ responses["/test/test.ics"]) == 1
+ status, prop = responses["/test/test.ics"]["D:getcontenttype"]
+ assert status == 200 and prop.text == (
+ "text/calendar;charset=utf-8;component=V%s" %
+ component.upper())
def test_addressbook_getcontenttype(self):
"""Test report request on an item"""
- status, _, _ = self._create_addressbook("/test/")
- assert status == 201
+ self.create_addressbook("/test/")
contact = get_file_content("contact1.vcf")
- status, _, _ = self.request("PUT", "/test/test.vcf", contact)
- assert status == 201
- status, _, answer = self.request(
- "REPORT", "/test/",
- """
-
-
-
-
- """)
- assert status == 207
- assert ">text/vcard;charset=utf-8<" in answer
+ self.put("/test/test.vcf", contact)
+ _, responses = self.report("/test/", """\
+
+
+
+
+
+""")
+ assert len(responses) == 1 and len(responses["/test/test.vcf"]) == 1
+ status, prop = responses["/test/test.vcf"]["D:getcontenttype"]
+ assert status == 200 and prop.text == "text/vcard;charset=utf-8"
def test_authorization(self):
authorization = "Basic " + base64.b64encode(b"user:").decode()
- status, _, answer = self.request(
- "PROPFIND", "/",
- """
-
-
-
-
- """,
- HTTP_AUTHORIZATION=authorization)
- assert status == 207
- assert "href>/user/<" in answer
+ _, responses = self.propfind("/", """\
+
+
+
+
+
+""", HTTP_AUTHORIZATION=authorization)
+ assert len(responses["/"]) == 1
+ status, prop = responses["/"]["D:current-user-principal"]
+ assert status == 200 and len(prop) == 1
+ assert prop.find(xmlutils.make_clark("D:href")).text == "/user/"
def test_authentication(self):
"""Test if server sends authentication request."""
@@ -1411,20 +1299,16 @@ class BaseRequestsMixIn:
def test_principal_collection_creation(self):
"""Verify existence of the principal collection."""
- status, _, _ = self.request("PROPFIND", "/user/", HTTP_AUTHORIZATION=(
+ self.propfind("/user/", HTTP_AUTHORIZATION=(
"Basic " + base64.b64encode(b"user:").decode()))
- assert status == 207
def test_existence_of_root_collections(self):
"""Verify that the root collection always exists."""
# Use PROPFIND because GET returns message
- status, _, _ = self.request("PROPFIND", "/")
- assert status == 207
+ self.propfind("/")
# it should still exist after deletion
- status, _, _ = self.request("DELETE", "/")
- assert status == 200
- status, _, _ = self.request("PROPFIND", "/")
- assert status == 207
+ self.delete("/")
+ self.propfind("/")
def test_custom_headers(self):
self.configuration.update({"headers": {"test": "123"}}, "test")
@@ -1434,8 +1318,7 @@ class BaseRequestsMixIn:
assert status == 200
assert headers.get("test") == "123"
# Test if header is set on failure
- status, headers, _ = self.request(
- "GET", "/.well-known/does not exist")
+ status, headers, _ = self.request("GET", "/.well-known/does not exist")
assert status == 404
assert headers.get("test") == "123"
@@ -1443,11 +1326,9 @@ class BaseRequestsMixIn:
reason="Unsupported in Python < 3.6")
def test_timezone_seconds(self):
"""Verify that timezones with minutes and seconds work."""
- status, _, _ = self.request("MKCALENDAR", "/calendar.ics/")
- assert status == 201
+ self.mkcalendar("/calendar.ics/")
event = get_file_content("event_timezone_seconds.ics")
- status, _, _ = self.request("PUT", "/calendar.ics/event.ics", event)
- assert status == 201
+ self.put("/calendar.ics/event.ics", event)
class BaseFileSystemTest(BaseTest):
@@ -1485,8 +1366,8 @@ class TestMultiFileSystem(BaseFileSystemTest, BaseRequestsMixIn):
def test_folder_creation(self):
"""Verify that the folder is created."""
folder = os.path.join(self.colpath, "subfolder")
- self.configuration.update({"storage": {"filesystem_folder": folder}},
- "test")
+ self.configuration.update(
+ {"storage": {"filesystem_folder": folder}}, "test")
self.application = Application(self.configuration)
assert os.path.isdir(folder)
@@ -1495,8 +1376,7 @@ class TestMultiFileSystem(BaseFileSystemTest, BaseRequestsMixIn):
self.configuration.update({
"internal": {"filesystem_fsync": "True"}}, "test", internal=True)
self.application = Application(self.configuration)
- status, _, _ = self.request("MKCALENDAR", "/calendar.ics/")
- assert status == 201
+ self.mkcalendar("/calendar.ics/")
def test_hook(self):
"""Run hook."""
@@ -1504,10 +1384,8 @@ class TestMultiFileSystem(BaseFileSystemTest, BaseRequestsMixIn):
"hook": ("mkdir %s" % os.path.join(
"collection-root", "created_by_hook"))}}, "test")
self.application = Application(self.configuration)
- status, _, _ = self.request("MKCALENDAR", "/calendar.ics/")
- assert status == 201
- status, _, _ = self.request("PROPFIND", "/created_by_hook/")
- assert status == 207
+ self.mkcalendar("/calendar.ics/")
+ self.propfind("/created_by_hook/")
def test_hook_read_access(self):
"""Verify that hook is not run for read accesses."""
@@ -1515,10 +1393,8 @@ class TestMultiFileSystem(BaseFileSystemTest, BaseRequestsMixIn):
"hook": ("mkdir %s" % os.path.join(
"collection-root", "created_by_hook"))}}, "test")
self.application = Application(self.configuration)
- status, _, _ = self.request("PROPFIND", "/")
- assert status == 207
- status, _, _ = self.request("PROPFIND", "/created_by_hook/")
- assert status == 404
+ self.propfind("/")
+ self.propfind("/created_by_hook/", check=404)
@pytest.mark.skipif(os.system("type flock") != 0,
reason="flock command not found")
@@ -1527,8 +1403,7 @@ class TestMultiFileSystem(BaseFileSystemTest, BaseRequestsMixIn):
self.configuration.update({"storage": {"hook": (
"flock -n .Radicale.lock || exit 0; exit 1")}}, "test")
self.application = Application(self.configuration)
- status, _, _ = self.request("MKCALENDAR", "/calendar.ics/")
- assert status == 201
+ self.mkcalendar("/calendar.ics/")
def test_hook_principal_collection_creation(self):
"""Verify that the hooks runs when a new user is created."""
@@ -1536,35 +1411,29 @@ class TestMultiFileSystem(BaseFileSystemTest, BaseRequestsMixIn):
"hook": ("mkdir %s" % os.path.join(
"collection-root", "created_by_hook"))}}, "test")
self.application = Application(self.configuration)
- status, _, _ = self.request("PROPFIND", "/", HTTP_AUTHORIZATION=(
+ self.propfind("/", HTTP_AUTHORIZATION=(
"Basic " + base64.b64encode(b"user:").decode()))
- assert status == 207
- status, _, _ = self.request("PROPFIND", "/created_by_hook/")
- assert status == 207
+ self.propfind("/created_by_hook/")
def test_hook_fail(self):
"""Verify that a request fails if the hook fails."""
self.configuration.update({"storage": {"hook": "exit 1"}}, "test")
self.application = Application(self.configuration)
- status, _, _ = self.request("MKCALENDAR", "/calendar.ics/")
+ status = self.mkcalendar("/calendar.ics/", check=False)
assert status != 201
def test_item_cache_rebuild(self):
"""Delete the item cache and verify that it is rebuild."""
- status, _, _ = self.request("MKCALENDAR", "/calendar.ics/")
- assert status == 201
+ self.mkcalendar("/calendar.ics/")
event = get_file_content("event1.ics")
path = "/calendar.ics/event1.ics"
- status, _, _ = self.request("PUT", path, event)
- assert status == 201
- status, _, answer1 = self.request("GET", path)
- assert status == 200
+ self.put(path, event)
+ _, answer1 = self.get(path)
cache_folder = os.path.join(self.colpath, "collection-root",
"calendar.ics", ".Radicale.cache", "item")
assert os.path.exists(os.path.join(cache_folder, "event1.ics"))
shutil.rmtree(cache_folder)
- status, _, answer2 = self.request("GET", path)
- assert status == 200
+ _, answer2 = self.get(path)
assert answer1 == answer2
assert os.path.exists(os.path.join(cache_folder, "event1.ics"))
@@ -1574,9 +1443,7 @@ class TestMultiFileSystem(BaseFileSystemTest, BaseRequestsMixIn):
"""Test if UIDs are used as file names."""
BaseRequestsMixIn.test_put_whole_calendar(self)
for uid in ("todo", "event"):
- status, _, answer = self.request(
- "GET", "/calendar.ics/%s.ics" % uid)
- assert status == 200
+ _, answer = self.get("/calendar.ics/%s.ics" % uid)
assert "\r\nUID:%s\r\n" % uid in answer
@pytest.mark.skipif(os.name not in ("nt", "posix"),
@@ -1584,16 +1451,13 @@ class TestMultiFileSystem(BaseFileSystemTest, BaseRequestsMixIn):
def test_put_whole_calendar_random_uids_used_as_file_names(self):
"""Test if UIDs are used as file names."""
BaseRequestsMixIn.test_put_whole_calendar_without_uids(self)
- status, _, answer = self.request("GET", "/calendar.ics")
- assert status == 200
+ _, answer = self.get("/calendar.ics")
uids = []
for line in answer.split("\r\n"):
if line.startswith("UID:"):
uids.append(line[len("UID:"):])
for uid in uids:
- status, _, answer = self.request(
- "GET", "/calendar.ics/%s.ics" % uid)
- assert status == 200
+ _, answer = self.get("/calendar.ics/%s.ics" % uid)
assert "\r\nUID:%s\r\n" % uid in answer
@pytest.mark.skipif(os.name not in ("nt", "posix"),
@@ -1602,9 +1466,7 @@ class TestMultiFileSystem(BaseFileSystemTest, BaseRequestsMixIn):
"""Test if UIDs are used as file names."""
BaseRequestsMixIn.test_put_whole_addressbook(self)
for uid in ("contact1", "contact2"):
- status, _, answer = self.request(
- "GET", "/contacts.vcf/%s.vcf" % uid)
- assert status == 200
+ _, answer = self.get("/contacts.vcf/%s.vcf" % uid)
assert "\r\nUID:%s\r\n" % uid in answer
@pytest.mark.skipif(os.name not in ("nt", "posix"),
@@ -1612,16 +1474,13 @@ class TestMultiFileSystem(BaseFileSystemTest, BaseRequestsMixIn):
def test_put_whole_addressbook_random_uids_used_as_file_names(self):
"""Test if UIDs are used as file names."""
BaseRequestsMixIn.test_put_whole_addressbook_without_uids(self)
- status, _, answer = self.request("GET", "/contacts.vcf")
- assert status == 200
+ _, answer = self.get("/contacts.vcf")
uids = []
for line in answer.split("\r\n"):
if line.startswith("UID:"):
uids.append(line[len("UID:"):])
for uid in uids:
- status, _, answer = self.request(
- "GET", "/contacts.vcf/%s.vcf" % uid)
- assert status == 200
+ _, answer = self.get("/contacts.vcf/%s.vcf" % uid)
assert "\r\nUID:%s\r\n" % uid in answer
diff --git a/radicale/tests/test_rights.py b/radicale/tests/test_rights.py
index 75aa0e2..7d07039 100644
--- a/radicale/tests/test_rights.py
+++ b/radicale/tests/test_rights.py
@@ -24,8 +24,8 @@ import shutil
import tempfile
from radicale import Application, config
+from radicale.tests import BaseTest
from radicale.tests.helpers import get_file_content
-from radicale.tests.test_base import BaseTest
class TestBaseRightsRequests(BaseTest):
@@ -56,69 +56,67 @@ class TestBaseRightsRequests(BaseTest):
"htpasswd_encryption": "plain"}}, "test")
self.application = Application(self.configuration)
for u in ("tmp", "other"):
- status, _, _ = self.request(
- "PROPFIND", "/%s" % u, HTTP_AUTHORIZATION="Basic %s" %
+ status, _ = self.propfind(
+ "/%s/" % u, HTTP_AUTHORIZATION="Basic %s" %
base64.b64encode(("%s:bepo" % u).encode()).decode())
- assert status == 207
- status, _, _ = self.request(
- "PROPFIND" if mode == "r" else "PROPPATCH", path,
- HTTP_AUTHORIZATION="Basic %s" % base64.b64encode(
- ("tmp:bepo").encode()).decode() if user else "")
+ status, _ = (self.propfind if mode == "r" else self.proppatch)(
+ path, check=False, HTTP_AUTHORIZATION="Basic %s" %
+ base64.b64encode(("tmp:bepo").encode()).decode() if user else "")
assert status == expected_status
def test_owner_only(self):
self._test_rights("owner_only", "", "/", "r", 401)
self._test_rights("owner_only", "", "/", "w", 401)
- self._test_rights("owner_only", "", "/tmp", "r", 401)
- self._test_rights("owner_only", "", "/tmp", "w", 401)
+ self._test_rights("owner_only", "", "/tmp/", "r", 401)
+ self._test_rights("owner_only", "", "/tmp/", "w", 401)
self._test_rights("owner_only", "tmp", "/", "r", 207)
self._test_rights("owner_only", "tmp", "/", "w", 403)
- self._test_rights("owner_only", "tmp", "/tmp", "r", 207)
- self._test_rights("owner_only", "tmp", "/tmp", "w", 207)
- self._test_rights("owner_only", "tmp", "/other", "r", 403)
- self._test_rights("owner_only", "tmp", "/other", "w", 403)
+ self._test_rights("owner_only", "tmp", "/tmp/", "r", 207)
+ self._test_rights("owner_only", "tmp", "/tmp/", "w", 207)
+ self._test_rights("owner_only", "tmp", "/other/", "r", 403)
+ self._test_rights("owner_only", "tmp", "/other/", "w", 403)
def test_owner_only_without_auth(self):
self._test_rights("owner_only", "", "/", "r", 207, False)
self._test_rights("owner_only", "", "/", "w", 401, False)
- self._test_rights("owner_only", "", "/tmp", "r", 207, False)
- self._test_rights("owner_only", "", "/tmp", "w", 207, False)
+ self._test_rights("owner_only", "", "/tmp/", "r", 207, False)
+ self._test_rights("owner_only", "", "/tmp/", "w", 207, False)
def test_owner_write(self):
self._test_rights("owner_write", "", "/", "r", 401)
self._test_rights("owner_write", "", "/", "w", 401)
- self._test_rights("owner_write", "", "/tmp", "r", 401)
- self._test_rights("owner_write", "", "/tmp", "w", 401)
+ self._test_rights("owner_write", "", "/tmp/", "r", 401)
+ self._test_rights("owner_write", "", "/tmp/", "w", 401)
self._test_rights("owner_write", "tmp", "/", "r", 207)
self._test_rights("owner_write", "tmp", "/", "w", 403)
- self._test_rights("owner_write", "tmp", "/tmp", "r", 207)
- self._test_rights("owner_write", "tmp", "/tmp", "w", 207)
- self._test_rights("owner_write", "tmp", "/other", "r", 207)
- self._test_rights("owner_write", "tmp", "/other", "w", 403)
+ self._test_rights("owner_write", "tmp", "/tmp/", "r", 207)
+ self._test_rights("owner_write", "tmp", "/tmp/", "w", 207)
+ self._test_rights("owner_write", "tmp", "/other/", "r", 207)
+ self._test_rights("owner_write", "tmp", "/other/", "w", 403)
def test_owner_write_without_auth(self):
self._test_rights("owner_write", "", "/", "r", 207, False)
self._test_rights("owner_write", "", "/", "w", 401, False)
- self._test_rights("owner_write", "", "/tmp", "r", 207, False)
- self._test_rights("owner_write", "", "/tmp", "w", 207, False)
+ self._test_rights("owner_write", "", "/tmp/", "r", 207, False)
+ self._test_rights("owner_write", "", "/tmp/", "w", 207, False)
def test_authenticated(self):
self._test_rights("authenticated", "", "/", "r", 401)
self._test_rights("authenticated", "", "/", "w", 401)
- self._test_rights("authenticated", "", "/tmp", "r", 401)
- self._test_rights("authenticated", "", "/tmp", "w", 401)
+ self._test_rights("authenticated", "", "/tmp/", "r", 401)
+ self._test_rights("authenticated", "", "/tmp/", "w", 401)
self._test_rights("authenticated", "tmp", "/", "r", 207)
self._test_rights("authenticated", "tmp", "/", "w", 207)
- self._test_rights("authenticated", "tmp", "/tmp", "r", 207)
- self._test_rights("authenticated", "tmp", "/tmp", "w", 207)
- self._test_rights("authenticated", "tmp", "/other", "r", 207)
- self._test_rights("authenticated", "tmp", "/other", "w", 207)
+ self._test_rights("authenticated", "tmp", "/tmp/", "r", 207)
+ self._test_rights("authenticated", "tmp", "/tmp/", "w", 207)
+ self._test_rights("authenticated", "tmp", "/other/", "r", 207)
+ self._test_rights("authenticated", "tmp", "/other/", "w", 207)
def test_authenticated_without_auth(self):
self._test_rights("authenticated", "", "/", "r", 207, False)
self._test_rights("authenticated", "", "/", "w", 207, False)
- self._test_rights("authenticated", "", "/tmp", "r", 207, False)
- self._test_rights("authenticated", "", "/tmp", "w", 207, False)
+ self._test_rights("authenticated", "", "/tmp/", "r", 207, False)
+ self._test_rights("authenticated", "", "/tmp/", "w", 207, False)
def test_from_file(self):
rights_file_path = os.path.join(self.colpath, "rights")
@@ -134,8 +132,8 @@ collection: custom(/.*)?
permissions: Rr""")
self.configuration.update(
{"rights": {"file": rights_file_path}}, "test")
- self._test_rights("from_file", "", "/other", "r", 401)
- self._test_rights("from_file", "tmp", "/other", "r", 403)
+ self._test_rights("from_file", "", "/other/", "r", 401)
+ self._test_rights("from_file", "tmp", "/other/", "r", 403)
self._test_rights("from_file", "", "/custom/sub", "r", 404)
self._test_rights("from_file", "tmp", "/custom/sub", "r", 404)
self._test_rights("from_file", "", "/custom/sub", "w", 401)
@@ -144,7 +142,8 @@ permissions: Rr""")
def test_custom(self):
"""Custom rights management."""
self._test_rights("radicale.tests.custom.rights", "", "/", "r", 401)
- self._test_rights("radicale.tests.custom.rights", "", "/tmp", "r", 207)
+ self._test_rights(
+ "radicale.tests.custom.rights", "", "/tmp/", "r", 207)
def test_collections_and_items(self):
"""Test rights for creation of collections, calendars and items.
@@ -155,33 +154,19 @@ permissions: Rr""")
"""
self.application = Application(self.configuration)
- status, _, _ = self.request("MKCALENDAR", "/")
- assert status == 401
- status, _, _ = self.request("MKCALENDAR", "/user/")
- assert status == 401
- status, _, _ = self.request("MKCOL", "/user/")
- assert status == 201
- status, _, _ = self.request("MKCOL", "/user/calendar/")
- assert status == 401
- status, _, _ = self.request("MKCALENDAR", "/user/calendar/")
- assert status == 201
- status, _, _ = self.request("MKCOL", "/user/calendar/item")
- assert status == 401
- status, _, _ = self.request("MKCALENDAR", "/user/calendar/item")
- assert status == 401
+ self.mkcalendar("/", check=401)
+ self.mkcalendar("/user/", check=401)
+ self.mkcol("/user/")
+ self.mkcol("/user/calendar/", check=401)
+ self.mkcalendar("/user/calendar/")
+ self.mkcol("/user/calendar/item", check=401)
+ self.mkcalendar("/user/calendar/item", check=401)
def test_put_collections_and_items(self):
"""Test rights for creation of calendars and items with PUT."""
self.application = Application(self.configuration)
- status, _, _ = self.request(
- "PUT", "/user/", "BEGIN:VCALENDAR\r\nEND:VCALENDAR")
- assert status == 401
- status, _, _ = self.request("MKCOL", "/user/")
- assert status == 201
- status, _, _ = self.request(
- "PUT", "/user/calendar/", "BEGIN:VCALENDAR\r\nEND:VCALENDAR")
- assert status == 201
+ self.put("/user/", "BEGIN:VCALENDAR\r\nEND:VCALENDAR", check=401)
+ self.mkcol("/user/")
+ self.put("/user/calendar/", "BEGIN:VCALENDAR\r\nEND:VCALENDAR")
event1 = get_file_content("event1.ics")
- status, _, _ = self.request(
- "PUT", "/user/calendar/event1.ics", event1)
- assert status == 201
+ self.put("/user/calendar/event1.ics", event1)
diff --git a/radicale/tests/test_server.py b/radicale/tests/test_server.py
index df858a8..317c8a1 100644
--- a/radicale/tests/test_server.py
+++ b/radicale/tests/test_server.py
@@ -35,6 +35,7 @@ from urllib.error import HTTPError, URLError
import pytest
from radicale import config, server
+from radicale.tests import BaseTest
from radicale.tests.helpers import configuration_to_dict, get_file_path
try:
@@ -50,7 +51,7 @@ class DisabledRedirectHandler(request.HTTPRedirectHandler):
http_error_301 = http_error_303 = http_error_307 = http_error_302
-class TestBaseServerRequests:
+class TestBaseServerRequests(BaseTest):
"""Test the internal server."""
def setup(self):
@@ -108,8 +109,7 @@ class TestBaseServerRequests:
def test_root(self):
self.thread.start()
- status, _, _ = self.request("GET", "/")
- assert status == 302
+ self.get("/", check=302)
def test_ssl(self):
self.configuration.update({
@@ -117,8 +117,7 @@ class TestBaseServerRequests:
"certificate": get_file_path("cert.pem"),
"key": get_file_path("key.pem")}}, "test")
self.thread.start()
- status, _, _ = self.request("GET", "/")
- assert status == 302
+ self.get("/", check=302)
@pytest.mark.skipif(not server.HAS_IPV6, reason="IPv6 not supported")
def test_ipv6(self):
@@ -132,16 +131,8 @@ class TestBaseServerRequests:
self.sockname = sock.getsockname()[:2]
self.configuration.update({
"server": {"hosts": "[%s]:%d" % self.sockname}}, "test")
- original_eai_addrfamily = server.EAI_ADDRFAMILY
- if os.name == "nt" and server.EAI_ADDRFAMILY is None:
- # HACK: incomplete errno conversion in WINE
- server.EAI_ADDRFAMILY = -9
- try:
- self.thread.start()
- status, _, _ = self.request("GET", "/")
- finally:
- server.EAI_ADDRFAMILY = original_eai_addrfamily
- assert status == 302
+ self.thread.start()
+ self.get("/", check=302)
def test_command_line_interface(self):
config_args = []
@@ -165,9 +156,7 @@ class TestBaseServerRequests:
p = subprocess.Popen(
[sys.executable, "-m", "radicale"] + config_args, env=env)
try:
- status, _, _ = self.request(
- "GET", "/", is_alive_fn=lambda: p.poll() is None)
- assert status == 302
+ self.get("/", is_alive_fn=lambda: p.poll() is None, check=302)
finally:
p.terminate()
p.wait()
@@ -189,9 +178,7 @@ class TestBaseServerRequests:
"--bind", self.configuration.get_raw("server", "hosts"),
"--env", "RADICALE_CONFIG=%s" % config_path, "radicale"], env=env)
try:
- status, _, _ = self.request(
- "GET", "/", is_alive_fn=lambda: p.poll() is None)
- assert status == 302
+ self.get("/", is_alive_fn=lambda: p.poll() is None, check=302)
finally:
p.terminate()
p.wait()
diff --git a/radicale/tests/test_web.py b/radicale/tests/test_web.py
index 516e381..a19d387 100644
--- a/radicale/tests/test_web.py
+++ b/radicale/tests/test_web.py
@@ -23,7 +23,7 @@ import shutil
import tempfile
from radicale import Application, config
-from radicale.tests.test_base import BaseTest
+from radicale.tests import BaseTest
class TestBaseWebRequests(BaseTest):
@@ -45,24 +45,20 @@ class TestBaseWebRequests(BaseTest):
status, headers, _ = self.request("GET", "/.web")
assert status == 302
assert headers.get("Location") == ".web/"
- status, _, answer = self.request("GET", "/.web/")
- assert status == 200
+ _, answer = self.get("/.web/")
assert answer
def test_none(self):
self.configuration.update({"web": {"type": "none"}}, "test")
self.application = Application(self.configuration)
- status, _, answer = self.request("GET", "/.web")
- assert status == 200
+ _, answer = self.get("/.web")
assert answer
- status, _, answer = self.request("GET", "/.web/")
- assert status == 404
+ self.get("/.web/", check=404)
def test_custom(self):
"""Custom web plugin."""
self.configuration.update({
"web": {"type": "radicale.tests.custom.web"}}, "test")
self.application = Application(self.configuration)
- status, _, answer = self.request("GET", "/.web")
- assert status == 200
+ _, answer = self.get("/.web")
assert answer == "custom"