From fc180266d5801ba204318f52022686bc042c9d33 Mon Sep 17 00:00:00 2001 From: Unrud Date: Mon, 20 Jan 2020 09:47:51 +0100 Subject: [PATCH] Improve tests - Parse and verify XML responses - Extract methods for common requests --- radicale/tests/__init__.py | 103 +- radicale/tests/test_auth.py | 59 +- radicale/tests/test_base.py | 1883 +++++++++++++++------------------ radicale/tests/test_rights.py | 105 +- radicale/tests/test_server.py | 29 +- radicale/tests/test_web.py | 14 +- 6 files changed, 1059 insertions(+), 1134 deletions(-) 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"