Avii/Radicale
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

Synced with origin

Browse Source
This commit is contained in:
Tuna Celik
2023-02-10 22:03:33 +01:00
parent 9b3bb2de2b
commit cf81d1f9a7
94 changed files with 5096 additions and 3560 deletions
Download Patch File Download Diff File Expand all files Collapse all files

190
radicale/tests/__init__.py
View File

@@ -1,4 +1,4 @@
# This file is part of Radicale Server - Calendar Server
# This file is part of Radicale - CalDAV and CardDAV server
# Copyright © 2012-2017 Guillaume Ayoub
# Copyright © 2017-2018 Unrud <unrud@outlook.com>
#
@@ -22,13 +22,19 @@ Tests for Radicale.
import base64
import logging
import shutil
import sys
import tempfile
import xml.etree.ElementTree as ET
from io import BytesIO
from typing import Any, Dict, List, Optional, Tuple, Union
import defusedxml.ElementTree as DefusedET
import radicale
from radicale import xmlutils
from radicale import app, config, types, xmlutils
RESPONSES = Dict[str, Union[int, Dict[str, Tuple[int, ET.Element]]]]
# Enable debug output
radicale.log.logger.setLevel(logging.DEBUG)
@@ -37,50 +43,84 @@ radicale.log.logger.setLevel(logging.DEBUG)
class BaseTest:
"""Base class for tests."""
def request(self, method, path, data=None, login=None, **args):
colpath: str
configuration: config.Configuration
application: app.Application
def setup(self) -> None:
self.configuration = config.load()
self.colpath = tempfile.mkdtemp()
self.configure({
"storage": {"filesystem_folder": self.colpath,
# Disable syncing to disk for better performance
"_filesystem_fsync": "False"},
# Set incorrect authentication delay to a short duration
"auth": {"delay": "0.001"}})
def configure(self, config_: types.CONFIG) -> None:
self.configuration.update(config_, "test", privileged=True)
self.application = app.Application(self.configuration)
def teardown(self) -> None:
shutil.rmtree(self.colpath)
def request(self, method: str, path: str, data: Optional[str] = None,
check: Optional[int] = None, **kwargs
) -> Tuple[int, Dict[str, str], str]:
"""Send a request."""
for key in args:
args[key.upper()] = args[key]
login = kwargs.pop("login", None)
if login is not None and not isinstance(login, str):
raise TypeError("login argument must be %r, not %r" %
(str, type(login)))
environ: Dict[str, Any] = {k.upper(): v for k, v in kwargs.items()}
for k, v in environ.items():
if not isinstance(v, str):
raise TypeError("type of %r is %r, expected %r" %
(k, type(v), str))
encoding: str = self.configuration.get("encoding", "request")
if login:
args["HTTP_AUTHORIZATION"] = "Basic " + base64.b64encode(
login.encode()).decode()
args["REQUEST_METHOD"] = method.upper()
args["PATH_INFO"] = path
environ["HTTP_AUTHORIZATION"] = "Basic " + base64.b64encode(
login.encode(encoding)).decode()
environ["REQUEST_METHOD"] = method.upper()
environ["PATH_INFO"] = path
if data:
data = data.encode()
args["wsgi.input"] = BytesIO(data)
args["CONTENT_LENGTH"] = str(len(data))
args["wsgi.errors"] = sys.stderr
data_bytes = data.encode(encoding)
environ["wsgi.input"] = BytesIO(data_bytes)
environ["CONTENT_LENGTH"] = str(len(data_bytes))
environ["wsgi.errors"] = sys.stderr
status = headers = None
def start_response(status_, headers_):
def start_response(status_: str, headers_: List[Tuple[str, str]]
) -> None:
nonlocal status, headers
status = status_
headers = headers_
answer = self.application(args, start_response)
status = int(status_.split()[0])
headers = dict(headers_)
answers = list(self.application(environ, start_response))
assert status is not None and headers is not None
assert check is None or status == check, "%d != %d" % (status, check)
return (int(status.split()[0]), dict(headers),
answer[0].decode() if answer else None)
return status, headers, answers[0].decode() if answers else ""
@staticmethod
def parse_responses(text):
def parse_responses(text: str) -> RESPONSES:
xml = DefusedET.fromstring(text)
assert xml.tag == xmlutils.make_clark("D:multistatus")
path_responses = {}
path_responses: Dict[str, Union[
int, Dict[str, Tuple[int, ET.Element]]]] = {}
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 = {}
prop_respones: Dict[str, Tuple[int, ET.Element]] = {}
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)
for element in propstat.findall(
"./%s/*" % xmlutils.make_clark("D: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
@@ -91,66 +131,84 @@ class BaseTest:
path_responses[href.text] = prop_respones
return path_responses
@staticmethod
def _check_status(status, good_status, check=True):
if check is True:
assert status == good_status
elif check is not False:
assert 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)
def get(self, path: str, check: Optional[int] = 200, **kwargs
) -> Tuple[int, str]:
assert "data" not in kwargs
status, _, answer = self.request("GET", path, check=check, **kwargs)
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)
def post(self, path: str, data: str = None, check: Optional[int] = 200,
**kwargs) -> Tuple[int, str]:
status, _, answer = self.request("POST", path, data, check=check,
**kwargs)
return status, answer
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
def put(self, path: str, data: str, check: Optional[int] = 201,
**kwargs) -> Tuple[int, str]:
status, _, answer = self.request("PUT", path, data, check=check,
**kwargs)
return status, answer
def propfind(self, path: str, data: Optional[str] = None,
check: Optional[int] = 207, **kwargs
) -> Tuple[int, RESPONSES]:
status, _, answer = self.request("PROPFIND", path, data, check=check,
**kwargs)
if status < 200 or 300 <= status:
return status, {}
assert answer is not None
responses = self.parse_responses(answer)
if args.get("HTTP_DEPTH", 0) == 0:
if kwargs.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
def proppatch(self, path: str, data: Optional[str] = None,
check: Optional[int] = 207, **kwargs
) -> Tuple[int, RESPONSES]:
status, _, answer = self.request("PROPPATCH", path, data, check=check,
**kwargs)
if status < 200 or 300 <= status:
return status, {}
assert answer is not 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
def report(self, path: str, data: str, check: Optional[int] = 207,
**kwargs) -> Tuple[int, RESPONSES]:
status, _, answer = self.request("REPORT", path, data, check=check,
**kwargs)
if status < 200 or 300 <= status:
return status, {}
assert answer is not 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
def delete(self, path: str, check: Optional[int] = 200, **kwargs
) -> Tuple[int, RESPONSES]:
assert "data" not in kwargs
status, _, answer = self.request("DELETE", path, check=check, **kwargs)
if status < 200 or 300 <= status:
return status, {}
assert answer is not 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, _, answer = self.request("MKCALENDAR", path, data, **args)
self._check_status(status, 201, check)
def mkcalendar(self, path: str, data: Optional[str] = None,
check: Optional[int] = 201, **kwargs
) -> Tuple[int, str]:
status, _, answer = self.request("MKCALENDAR", path, data, check=check,
**kwargs)
return status, answer
def mkcol(self, path, data=None, check=True, **args):
status, _, _ = self.request("MKCOL", path, data, **args)
self._check_status(status, 201, check)
def mkcol(self, path: str, data: Optional[str] = None,
check: Optional[int] = 201, **kwargs) -> int:
status, _, _ = self.request("MKCOL", path, data, check=check, **kwargs)
return status
def create_addressbook(self, path, check=True, **args):
def create_addressbook(self, path: str, check: Optional[int] = 201,
**kwargs) -> int:
assert "data" not in kwargs
return self.mkcol(path, """\
<?xml version="1.0" encoding="UTF-8" ?>
<create xmlns="DAV:" xmlns:CR="urn:ietf:params:xml:ns:carddav">
@@ -162,4 +220,4 @@ class BaseTest:
</resourcetype>
</prop>
</set>
</create>""", check=check, **args)
</create>""", check=check, **kwargs)

5
radicale/tests/custom/auth.py
View File

@@ -1,4 +1,4 @@
# This file is part of Radicale Server - Calendar Server
# This file is part of Radicale - CalDAV and CardDAV server
# Copyright © 2008 Nicolas Kandel
# Copyright © 2008 Pascal Halter
# Copyright © 2008-2017 Guillaume Ayoub
@@ -28,7 +28,8 @@ from radicale import auth
class Auth(auth.BaseAuth):
def login(self, login, password):
def login(self, login: str, password: str) -> str:
if login == "tmp":
return login
return ""

5
radicale/tests/custom/rights.py
View File

@@ -1,4 +1,4 @@
# This file is part of Radicale Server - Calendar Server
# This file is part of Radicale - CalDAV and CardDAV server
# Copyright © 2017-2018 Unrud <unrud@outlook.com>
#
# This library is free software: you can redistribute it and/or modify
@@ -23,7 +23,8 @@ from radicale import pathutils, rights
class Rights(rights.BaseRights):
def authorization(self, user, path):
def authorization(self, user: str, path: str) -> str:
sane_path = pathutils.strip_path(path)
if sane_path not in ("tmp", "other"):
return ""

4
radicale/tests/custom/storage_simple_sync.py
View File

@@ -1,4 +1,4 @@
# This file is part of Radicale Server - Calendar Server
# This file is part of Radicale - CalDAV and CardDAV server
# Copyright © 2012-2017 Guillaume Ayoub
# Copyright © 2017-2018 Unrud <unrud@outlook.com>
#
@@ -27,8 +27,10 @@ from radicale.storage import BaseCollection, multifilesystem
class Collection(multifilesystem.Collection):
sync = BaseCollection.sync
class Storage(multifilesystem.Storage):
_collection_class = Collection

13
radicale/tests/custom/web.py
View File

@@ -1,4 +1,4 @@
# This file is part of Radicale Server - Calendar Server
# This file is part of Radicale - CalDAV and CardDAV server
# Copyright © 2017-2018 Unrud <unrud@outlook.com>
#
# This library is free software: you can redistribute it and/or modify
@@ -21,9 +21,16 @@ Custom web plugin.
from http import client
from radicale import web
from radicale import httputils, types, web
class Web(web.BaseWeb):
def get(self, environ, base_prefix, path, user):
def get(self, environ: types.WSGIEnviron, base_prefix: str, path: str,
user: str) -> types.WSGIResponse:
return client.OK, {"Content-Type": "text/plain"}, "custom"
def post(self, environ: types.WSGIEnviron, base_prefix: str, path: str,
user: str) -> types.WSGIResponse:
content = httputils.read_request_body(self.configuration, environ)
return client.OK, {"Content-Type": "text/plain"}, "echo:" + content

19
radicale/tests/helpers.py
View File

@@ -1,4 +1,4 @@
# This file is part of Radicale Server - Calendar Server
# This file is part of Radicale - CalDAV and CardDAV server
# Copyright © 2008 Nicolas Kandel
# Copyright © 2008 Pascal Halter
# Copyright © 2008-2017 Guillaume Ayoub
@@ -26,22 +26,21 @@ This module offers helpers to use in tests.
import os
EXAMPLES_FOLDER = os.path.join(os.path.dirname(__file__), "static")
from radicale import config, types
EXAMPLES_FOLDER: str = os.path.join(os.path.dirname(__file__), "static")
def get_file_path(file_name):
def get_file_path(file_name: str) -> str:
return os.path.join(EXAMPLES_FOLDER, file_name)
def get_file_content(file_name):
try:
with open(get_file_path(file_name), encoding="utf-8") as fd:
return fd.read()
except IOError:
print("Couldn't open the file %s" % file_name)
def get_file_content(file_name: str) -> str:
with open(get_file_path(file_name), encoding="utf-8") as f:
return f.read()
def configuration_to_dict(configuration):
def configuration_to_dict(configuration: config.Configuration) -> types.CONFIG:
"""Convert configuration to a dict with raw values."""
return {section: {option: configuration.get_raw(section, option)
for option in configuration.options(section)

8
radicale/tests/static/contact_photo_with_data_uri.vcf Normal file
View File

@@ -0,0 +1,8 @@
BEGIN:VCARD
VERSION:3.0
UID:contact
N:Contact;;;;
FN:Contact
NICKNAME:test
PHOTO;ENCODING=b;TYPE=png:data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAIAAACQd1PeAAAAD0lEQVQIHQEEAPv/AP///wX+Av4DfRnGAAAAAElFTkSuQmCC
END:VCARD

33
radicale/tests/static/event_mixed_datetime_and_date.ics Normal file
View File

@@ -0,0 +1,33 @@
BEGIN:VCALENDAR
PRODID:-//Mozilla.org/NONSGML Mozilla Calendar V1.1//EN
VERSION:2.0
BEGIN:VTIMEZONE
TZID:Europe/Paris
X-LIC-LOCATION:Europe/Paris
BEGIN:DAYLIGHT
TZOFFSETFROM:+0100
TZOFFSETTO:+0200
TZNAME:CEST
DTSTART:19700329T020000
RRULE:FREQ=YEARLY;BYDAY=-1SU;BYMONTH=3
END:DAYLIGHT
BEGIN:STANDARD
TZOFFSETFROM:+0200
TZOFFSETTO:+0100
TZNAME:CET
DTSTART:19701025T030000
RRULE:FREQ=YEARLY;BYDAY=-1SU;BYMONTH=10
END:STANDARD
END:VTIMEZONE
BEGIN:VEVENT
CREATED:20130902T150157Z
LAST-MODIFIED:20130902T150158Z
DTSTAMP:20130902T150158Z
UID:event_mixed_datetime_and_date
SUMMARY:Event
DTSTART;TZID=Europe/Paris:20130901T180000
DTEND;TZID=Europe/Paris:20130901T190000
RRULE:FREQ=DAILY;COUNT=3
EXDATE;VALUE=DATE:20130902
END:VEVENT
END:VCALENDAR

16
radicale/tests/static/event_multiple_case_sensitive_uids.ics Normal file
View File

@@ -0,0 +1,16 @@
BEGIN:VCALENDAR
PRODID:-//Mozilla.org/NONSGML Mozilla Calendar V1.1//EN
VERSION:2.0
BEGIN:VEVENT
UID:event
SUMMARY:Event 1
DTSTART:20130901T190000
DTEND:20130901T200000
END:VEVENT
BEGIN:VEVENT
UID:EVENT
SUMMARY:Event 2
DTSTART:20130901T200000
DTEND:20130901T210000
END:VEVENT
END:VCALENDAR

9
radicale/tests/static/mkcol_make_calendar.xml Normal file
View File

@@ -0,0 +1,9 @@
<?xml version="1.0" encoding="utf-8"?>
<D:mkcol xmlns:D="DAV:" xmlns:C="urn:ietf:params:xml:ns:caldav">
<D:set>
<D:prop>
<D:resourcetype><collection /><C:calendar /></D:resourcetype>
<I:calendar-color xmlns:I="http://apple.com/ns/ical/">#BADA55</I:calendar-color>
</D:prop>
</D:set>
</D:mkcol>

0
radicale/tests/static/propfind1.xml → radicale/tests/static/propfind_calendar_color.xml
View File

7
radicale/tests/static/propfind_multiple.xml Normal file
View File

@@ -0,0 +1,7 @@
<?xml version="1.0" encoding="utf-8"?>
<D:propfind xmlns:D="DAV:">
<D:prop>
<I:calendar-color xmlns:I="http://apple.com/ns/ical/" />
<C:calendar-description xmlns:C="urn:ietf:params:xml:ns:caldav" />
</D:prop>
</D:propfind>

8
radicale/tests/static/proppatch_remove_calendar_color.xml Normal file
View File

@@ -0,0 +1,8 @@
<?xml version="1.0" encoding="utf-8"?>
<D:propertyupdate xmlns:D="DAV:">
<D:remove>
<D:prop>
<I:calendar-color xmlns:I="http://apple.com/ns/ical/" />
</D:prop>
</D:remove>
</D:propertyupdate>

9
radicale/tests/static/proppatch_remove_multiple1.xml Normal file
View File

@@ -0,0 +1,9 @@
<?xml version="1.0" encoding="utf-8"?>
<D:propertyupdate xmlns:D="DAV:">
<D:remove>
<D:prop>
<I:calendar-color xmlns:I="http://apple.com/ns/ical/" />
<C:calendar-description xmlns:C="urn:ietf:params:xml:ns:caldav" />
</D:prop>
</D:remove>
</D:propertyupdate>

13
radicale/tests/static/proppatch_remove_multiple2.xml Normal file
View File

@@ -0,0 +1,13 @@
<?xml version="1.0" encoding="utf-8"?>
<D:propertyupdate xmlns:D="DAV:">
<D:remove>
<D:prop>
<I:calendar-color xmlns:I="http://apple.com/ns/ical/" />
</D:prop>
</D:remove>
<D:remove>
<D:prop>
<C:calendar-description xmlns:C="urn:ietf:params:xml:ns:caldav" />
</D:prop>
</D:remove>
</D:propertyupdate>

13
radicale/tests/static/proppatch_set_and_remove.xml Normal file
View File

@@ -0,0 +1,13 @@
<?xml version="1.0" encoding="utf-8"?>
<D:propertyupdate xmlns:D="DAV:">
<D:remove>
<D:prop>
<I:calendar-color xmlns:I="http://apple.com/ns/ical/" />
</D:prop>
</D:remove>
<D:set>
<D:prop>
<C:calendar-description xmlns:C="urn:ietf:params:xml:ns:caldav">test2</C:calendar-description>
</D:prop>
</D:set>
</D:propertyupdate>

0
radicale/tests/static/proppatch1.xml → radicale/tests/static/proppatch_set_calendar_color.xml
View File

9
radicale/tests/static/proppatch_set_multiple1.xml Normal file
View File

@@ -0,0 +1,9 @@
<?xml version="1.0" encoding="utf-8"?>
<D:propertyupdate xmlns:D="DAV:">
<D:set>
<D:prop>
<I:calendar-color xmlns:I="http://apple.com/ns/ical/">#BADA55</I:calendar-color>
<C:calendar-description xmlns:C="urn:ietf:params:xml:ns:caldav">test</C:calendar-description>
</D:prop>
</D:set>
</D:propertyupdate>

13
radicale/tests/static/proppatch_set_multiple2.xml Normal file
View File

@@ -0,0 +1,13 @@
<?xml version="1.0" encoding="utf-8"?>
<D:propertyupdate xmlns:D="DAV:">
<D:set>
<D:prop>
<I:calendar-color xmlns:I="http://apple.com/ns/ical/">#BADA55</I:calendar-color>
</D:prop>
</D:set>
<D:set>
<D:prop>
<C:calendar-description xmlns:C="urn:ietf:params:xml:ns:caldav">test</C:calendar-description>
</D:prop>
</D:set>
</D:propertyupdate>

102
radicale/tests/test_auth.py
View File

@@ -1,4 +1,4 @@
# This file is part of Radicale Server - Calendar Server
# This file is part of Radicale - CalDAV and CardDAV server
# Copyright © 2012-2016 Jean-Marc Martins
# Copyright © 2012-2017 Guillaume Ayoub
# Copyright © 2017-2019 Unrud <unrud@outlook.com>
@@ -22,12 +22,12 @@ Radicale tests with simple requests and authentication.
"""
import os
import shutil
import tempfile
import sys
from typing import Iterable, Tuple, Union
import pytest
from radicale import Application, config, xmlutils
from radicale import xmlutils
from radicale.tests import BaseTest
@@ -37,43 +37,20 @@ class TestBaseAuthRequests(BaseTest):
We should setup auth for each type before creating the Application object.
"""
def setup(self):
self.configuration = config.load()
self.colpath = tempfile.mkdtemp()
self.configuration.update({
"storage": {"filesystem_folder": self.colpath,
# Disable syncing to disk for better performance
"_filesystem_fsync": "False"},
# Set incorrect authentication delay to a very low value
"auth": {"delay": "0.002"}}, "test", privileged=True)
def teardown(self):
shutil.rmtree(self.colpath)
def _test_htpasswd(self, htpasswd_encryption, htpasswd_content,
test_matrix="ascii"):
def _test_htpasswd(self, htpasswd_encryption: str, htpasswd_content: str,
test_matrix: Union[str, Iterable[Tuple[str, str, bool]]]
= "ascii") -> None:
"""Test htpasswd authentication with user "tmp" and password "bepo" for
``test_matrix`` "ascii" or user "😀" and password "🔑" for
``test_matrix`` "unicode"."""
if htpasswd_encryption == "bcrypt":
try:
from passlib.exc import MissingBackendError
from passlib.hash import bcrypt
except ImportError:
pytest.skip("passlib[bcrypt] is not installed")
try:
bcrypt.hash("test-bcrypt-backend")
except MissingBackendError:
pytest.skip("bcrypt backend for passlib is not installed")
htpasswd_file_path = os.path.join(self.colpath, ".htpasswd")
encoding = self.configuration.get("encoding", "stock")
encoding: str = self.configuration.get("encoding", "stock")
with open(htpasswd_file_path, "w", encoding=encoding) as f:
f.write(htpasswd_content)
self.configuration.update({
"auth": {"type": "htpasswd",
"htpasswd_filename": htpasswd_file_path,
"htpasswd_encryption": htpasswd_encryption}}, "test")
self.application = Application(self.configuration)
self.configure({"auth": {"type": "htpasswd",
"htpasswd_filename": htpasswd_file_path,
"htpasswd_encryption": htpasswd_encryption}})
if test_matrix == "ascii":
test_matrix = (("tmp", "bepo", True), ("tmp", "tmp", False),
("tmp", "", False), ("unk", "unk", False),
@@ -82,56 +59,57 @@ class TestBaseAuthRequests(BaseTest):
test_matrix = (("😀", "🔑", True), ("😀", "🌹", False),
("😁", "🔑", False), ("😀", "", False),
("", "🔑", False), ("", "", False))
elif isinstance(test_matrix, str):
raise ValueError("Unknown test matrix %r" % test_matrix)
for user, password, valid in test_matrix:
self.propfind("/", check=207 if valid else 401,
login="%s:%s" % (user, password))
def test_htpasswd_plain(self):
def test_htpasswd_plain(self) -> None:
self._test_htpasswd("plain", "tmp:bepo")
def test_htpasswd_plain_password_split(self):
def test_htpasswd_plain_password_split(self) -> None:
self._test_htpasswd("plain", "tmp:be:po", (
("tmp", "be:po", True), ("tmp", "bepo", False)))
def test_htpasswd_plain_unicode(self):
def test_htpasswd_plain_unicode(self) -> None:
self._test_htpasswd("plain", "😀:🔑", "unicode")
def test_htpasswd_md5(self):
def test_htpasswd_md5(self) -> None:
self._test_htpasswd("md5", "tmp:$apr1$BI7VKCZh$GKW4vq2hqDINMr8uv7lDY/")
def test_htpasswd_md5_unicode(self):
self._test_htpasswd(
"md5", "😀:$apr1$w4ev89r1$29xO8EvJmS2HEAadQ5qy11", "unicode")
def test_htpasswd_bcrypt(self):
def test_htpasswd_bcrypt(self) -> None:
self._test_htpasswd("bcrypt", "tmp:$2y$05$oD7hbiQFQlvCM7zoalo/T.MssV3V"
"NTRI3w5KDnj8NTUKJNWfVpvRq")
def test_htpasswd_bcrypt_unicode(self):
def test_htpasswd_bcrypt_unicode(self) -> None:
self._test_htpasswd("bcrypt", "😀:$2y$10$Oyz5aHV4MD9eQJbk6GPemOs4T6edK"
"6U9Sqlzr.W1mMVCS8wJUftnW", "unicode")
def test_htpasswd_multi(self):
def test_htpasswd_multi(self) -> None:
self._test_htpasswd("plain", "ign:ign\ntmp:bepo")
@pytest.mark.skipif(os.name == "nt", reason="leading and trailing "
@pytest.mark.skipif(sys.platform == "win32", reason="leading and trailing "
"whitespaces not allowed in file names")
def test_htpasswd_whitespace_user(self):
def test_htpasswd_whitespace_user(self) -> None:
for user in (" tmp", "tmp ", " tmp "):
self._test_htpasswd("plain", "%s:bepo" % user, (
(user, "bepo", True), ("tmp", "bepo", False)))
def test_htpasswd_whitespace_password(self):
def test_htpasswd_whitespace_password(self) -> None:
for password in (" bepo", "bepo ", " bepo "):
self._test_htpasswd("plain", "tmp:%s" % password, (
("tmp", password, True), ("tmp", "bepo", False)))
def test_htpasswd_comment(self):
def test_htpasswd_comment(self) -> None:
self._test_htpasswd("plain", "#comment\n #comment\n \ntmp:bepo\n\n")
def test_remote_user(self):
self.configuration.update({"auth": {"type": "remote_user"}}, "test")
self.application = Application(self.configuration)
def test_remote_user(self) -> None:
self.configure({"auth": {"type": "remote_user"}})
_, responses = self.propfind("/", """\
<?xml version="1.0" encoding="utf-8"?>
<propfind xmlns="DAV:">
@@ -139,14 +117,16 @@ class TestBaseAuthRequests(BaseTest):
<current-user-principal />
</prop>
</propfind>""", REMOTE_USER="test")
status, prop = responses["/"]["D:current-user-principal"]
assert responses is not None
response = responses["/"]
assert not isinstance(response, int)
status, prop = response["D:current-user-principal"]
assert status == 200
assert prop.find(xmlutils.make_clark("D:href")).text == "/test/"
href_element = prop.find(xmlutils.make_clark("D:href"))
assert href_element is not None and href_element.text == "/test/"
def test_http_x_remote_user(self):
self.configuration.update(
{"auth": {"type": "http_x_remote_user"}}, "test")
self.application = Application(self.configuration)
def test_http_x_remote_user(self) -> None:
self.configure({"auth": {"type": "http_x_remote_user"}})
_, responses = self.propfind("/", """\
<?xml version="1.0" encoding="utf-8"?>
<propfind xmlns="DAV:">
@@ -154,13 +134,15 @@ class TestBaseAuthRequests(BaseTest):
<current-user-principal />
</prop>
</propfind>""", HTTP_X_REMOTE_USER="test")
status, prop = responses["/"]["D:current-user-principal"]
assert responses is not None
response = responses["/"]
assert not isinstance(response, int)
status, prop = response["D:current-user-principal"]
assert status == 200
assert prop.find(xmlutils.make_clark("D:href")).text == "/test/"
href_element = prop.find(xmlutils.make_clark("D:href"))
assert href_element is not None and href_element.text == "/test/"
def test_custom(self):
def test_custom(self) -> None:
"""Custom authentication."""
self.configuration.update(
{"auth": {"type": "radicale.tests.custom.auth"}}, "test")
self.application = Application(self.configuration)
self.configure({"auth": {"type": "radicale.tests.custom.auth"}})
self.propfind("/tmp/", login="tmp:")

777
radicale/tests/test_base.py
View File

File diff suppressed because it is too large Load Diff

63
radicale/tests/test_config.py
View File

@@ -1,4 +1,4 @@
# This file is part of Radicale Server - Calendar Server
# This file is part of Radicale - CalDAV and CardDAV server
# Copyright © 2019 Unrud <unrud@outlook.com>
#
# This library is free software: you can redistribute it and/or modify
@@ -18,23 +18,26 @@ import os
import shutil
import tempfile
from configparser import RawConfigParser
from typing import List, Tuple
import pytest
from radicale import config
from radicale import config, types
from radicale.tests.helpers import configuration_to_dict
class TestConfig:
"""Test the configuration."""
def setup(self):
colpath: str
def setup(self) -> None:
self.colpath = tempfile.mkdtemp()
def teardown(self):
def teardown(self) -> None:
shutil.rmtree(self.colpath)
def _write_config(self, config_dict, name):
def _write_config(self, config_dict: types.CONFIG, name: str) -> str:
parser = RawConfigParser()
parser.read_dict(config_dict)
config_path = os.path.join(self.colpath, name)
@@ -42,7 +45,7 @@ class TestConfig:
parser.write(f)
return config_path
def test_parse_compound_paths(self):
def test_parse_compound_paths(self) -> None:
assert len(config.parse_compound_paths()) == 0
assert len(config.parse_compound_paths("")) == 0
assert len(config.parse_compound_paths(None, "")) == 0
@@ -62,16 +65,16 @@ class TestConfig:
assert os.path.basename(paths[i][0]) == name
assert paths[i][1] is ignore_if_missing
def test_load_empty(self):
def test_load_empty(self) -> None:
config_path = self._write_config({}, "config")
config.load([(config_path, False)])
def test_load_full(self):
def test_load_full(self) -> None:
config_path = self._write_config(
configuration_to_dict(config.load()), "config")
config.load([(config_path, False)])
def test_load_missing(self):
def test_load_missing(self) -> None:
config_path = os.path.join(self.colpath, "does_not_exist")
config.load([(config_path, True)])
with pytest.raises(Exception) as exc_info:
@@ -79,18 +82,20 @@ class TestConfig:
e = exc_info.value
assert "Failed to load config file %r" % config_path in str(e)
def test_load_multiple(self):
def test_load_multiple(self) -> None:
config_path1 = self._write_config({
"server": {"hosts": "192.0.2.1:1111"}}, "config1")
config_path2 = self._write_config({
"server": {"max_connections": 1111}}, "config2")
configuration = config.load([(config_path1, False),
(config_path2, False)])
assert len(configuration.get("server", "hosts")) == 1
assert configuration.get("server", "hosts")[0] == ("192.0.2.1", 1111)
server_hosts: List[Tuple[str, int]] = configuration.get(
"server", "hosts")
assert len(server_hosts) == 1
assert server_hosts[0] == ("192.0.2.1", 1111)
assert configuration.get("server", "max_connections") == 1111
def test_copy(self):
def test_copy(self) -> None:
configuration1 = config.load()
configuration1.update({"server": {"max_connections": "1111"}}, "test")
configuration2 = configuration1.copy()
@@ -98,14 +103,14 @@ class TestConfig:
assert configuration1.get("server", "max_connections") == 1111
assert configuration2.get("server", "max_connections") == 1112
def test_invalid_section(self):
def test_invalid_section(self) -> None:
configuration = config.load()
with pytest.raises(Exception) as exc_info:
configuration.update({"does_not_exist": {"x": "x"}}, "test")
e = exc_info.value
assert "Invalid section 'does_not_exist'" in str(e)
def test_invalid_option(self):
def test_invalid_option(self) -> None:
configuration = config.load()
with pytest.raises(Exception) as exc_info:
configuration.update({"server": {"x": "x"}}, "test")
@@ -113,7 +118,7 @@ class TestConfig:
assert "Invalid option 'x'" in str(e)
assert "section 'server'" in str(e)
def test_invalid_option_plugin(self):
def test_invalid_option_plugin(self) -> None:
configuration = config.load()
with pytest.raises(Exception) as exc_info:
configuration.update({"auth": {"x": "x"}}, "test")
@@ -121,7 +126,7 @@ class TestConfig:
assert "Invalid option 'x'" in str(e)
assert "section 'auth'" in str(e)
def test_invalid_value(self):
def test_invalid_value(self) -> None:
configuration = config.load()
with pytest.raises(Exception) as exc_info:
configuration.update({"server": {"max_connections": "x"}}, "test")
@@ -131,7 +136,7 @@ class TestConfig:
assert "section 'server" in str(e)
assert "'x'" in str(e)
def test_privileged(self):
def test_privileged(self) -> None:
configuration = config.load()
configuration.update({"server": {"_internal_server": "True"}},
"test", privileged=True)
@@ -141,9 +146,9 @@ class TestConfig:
e = exc_info.value
assert "Invalid option '_internal_server'" in str(e)
def test_plugin_schema(self):
plugin_schema = {"auth": {"new_option": {"value": "False",
"type": bool}}}
def test_plugin_schema(self) -> None:
plugin_schema: types.CONFIG_SCHEMA = {
"auth": {"new_option": {"value": "False", "type": bool}}}
configuration = config.load()
configuration.update({"auth": {"type": "new_plugin"}}, "test")
plugin_configuration = configuration.copy(plugin_schema)
@@ -152,26 +157,26 @@ class TestConfig:
plugin_configuration = configuration.copy(plugin_schema)
assert plugin_configuration.get("auth", "new_option") is True
def test_plugin_schema_duplicate_option(self):
plugin_schema = {"auth": {"type": {"value": "False",
"type": bool}}}
def test_plugin_schema_duplicate_option(self) -> None:
plugin_schema: types.CONFIG_SCHEMA = {
"auth": {"type": {"value": "False", "type": bool}}}
configuration = config.load()
with pytest.raises(Exception) as exc_info:
configuration.copy(plugin_schema)
e = exc_info.value
assert "option already exists in 'auth': 'type'" in str(e)
def test_plugin_schema_invalid(self):
plugin_schema = {"server": {"new_option": {"value": "False",
"type": bool}}}
def test_plugin_schema_invalid(self) -> None:
plugin_schema: types.CONFIG_SCHEMA = {
"server": {"new_option": {"value": "False", "type": bool}}}
configuration = config.load()
with pytest.raises(Exception) as exc_info:
configuration.copy(plugin_schema)
e = exc_info.value
assert "not a plugin section: 'server" in str(e)
def test_plugin_schema_option_invalid(self):
plugin_schema = {"auth": {}}
def test_plugin_schema_option_invalid(self) -> None:
plugin_schema: types.CONFIG_SCHEMA = {"auth": {}}
configuration = config.load()
configuration.update({"auth": {"type": "new_plugin",
"new_option": False}}, "test")

57
radicale/tests/test_rights.py
View File

@@ -1,4 +1,4 @@
# This file is part of Radicale Server - Calendar Server
# This file is part of Radicale - CalDAV and CardDAV server
# Copyright © 2017-2019 Unrud <unrud@outlook.com>
#
# This library is free software: you can redistribute it and/or modify
@@ -19,10 +19,7 @@ Radicale tests with simple requests and rights.
"""
import os
import shutil
import tempfile
from radicale import Application, config
from radicale.tests import BaseTest
from radicale.tests.helpers import get_file_content
@@ -30,38 +27,25 @@ from radicale.tests.helpers import get_file_content
class TestBaseRightsRequests(BaseTest):
"""Tests basic requests with rights."""
def setup(self):
self.configuration = config.load()
self.colpath = tempfile.mkdtemp()
self.configuration.update({
"storage": {"filesystem_folder": self.colpath,
# Disable syncing to disk for better performance
"_filesystem_fsync": "False"}},
"test", privileged=True)
def teardown(self):
shutil.rmtree(self.colpath)
def _test_rights(self, rights_type, user, path, mode, expected_status,
with_auth=True):
def _test_rights(self, rights_type: str, user: str, path: str, mode: str,
expected_status: int, with_auth: bool = True) -> None:
assert mode in ("r", "w")
assert user in ("", "tmp")
htpasswd_file_path = os.path.join(self.colpath, ".htpasswd")
with open(htpasswd_file_path, "w") as f:
f.write("tmp:bepo\nother:bepo")
self.configuration.update({
self.configure({
"rights": {"type": rights_type},
"auth": {"type": "htpasswd" if with_auth else "none",
"htpasswd_filename": htpasswd_file_path,
"htpasswd_encryption": "plain"}}, "test")
self.application = Application(self.configuration)
"htpasswd_encryption": "plain"}})
for u in ("tmp", "other"):
# Indirect creation of principal collection
self.propfind("/%s/" % u, login="%s:bepo" % u)
(self.propfind if mode == "r" else self.proppatch)(
path, check=expected_status, login="tmp:bepo" if user else None)
def test_owner_only(self):
def test_owner_only(self) -> None:
self._test_rights("owner_only", "", "/", "r", 401)
self._test_rights("owner_only", "", "/", "w", 401)
self._test_rights("owner_only", "", "/tmp/", "r", 401)
@@ -73,13 +57,13 @@ class TestBaseRightsRequests(BaseTest):
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):
def test_owner_only_without_auth(self) -> None:
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)
def test_owner_write(self):
def test_owner_write(self) -> None:
self._test_rights("owner_write", "", "/", "r", 401)
self._test_rights("owner_write", "", "/", "w", 401)
self._test_rights("owner_write", "", "/tmp/", "r", 401)
@@ -91,13 +75,13 @@ class TestBaseRightsRequests(BaseTest):
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):
def test_owner_write_without_auth(self) -> None:
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)
def test_authenticated(self):
def test_authenticated(self) -> None:
self._test_rights("authenticated", "", "/", "r", 401)
self._test_rights("authenticated", "", "/", "w", 401)
self._test_rights("authenticated", "", "/tmp/", "r", 401)
@@ -109,13 +93,13 @@ class TestBaseRightsRequests(BaseTest):
self._test_rights("authenticated", "tmp", "/other/", "r", 207)
self._test_rights("authenticated", "tmp", "/other/", "w", 207)
def test_authenticated_without_auth(self):
def test_authenticated_without_auth(self) -> None:
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)
def test_from_file(self):
def test_from_file(self) -> None:
rights_file_path = os.path.join(self.colpath, "rights")
with open(rights_file_path, "w") as f:
f.write("""\
@@ -127,8 +111,7 @@ permissions: RrWw
user: .*
collection: custom(/.*)?
permissions: Rr""")
self.configuration.update(
{"rights": {"file": rights_file_path}}, "test")
self.configure({"rights": {"file": rights_file_path}})
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)
@@ -148,10 +131,8 @@ permissions: RrWw
user: .*
collection: public/[^/]*
permissions: i""")
self.configuration.update(
{"rights": {"type": "from_file",
"file": rights_file_path}}, "test")
self.application = Application(self.configuration)
self.configure({"rights": {"type": "from_file",
"file": rights_file_path}})
self.mkcalendar("/tmp/calendar", login="tmp:bepo")
self.mkcol("/public", login="tmp:bepo")
self.mkcalendar("/public/calendar", login="tmp:bepo")
@@ -160,13 +141,13 @@ permissions: i""")
self.get("/public/calendar")
self.get("/public/calendar/1.ics", check=401)
def test_custom(self):
def test_custom(self) -> None:
"""Custom rights management."""
self._test_rights("radicale.tests.custom.rights", "", "/", "r", 401)
self._test_rights(
"radicale.tests.custom.rights", "", "/tmp/", "r", 207)
def test_collections_and_items(self):
def test_collections_and_items(self) -> None:
"""Test rights for creation of collections, calendars and items.
Collections are allowed at "/" and "/.../".
@@ -174,7 +155,6 @@ permissions: i""")
Items are allowed at "/.../.../...".
"""
self.application = Application(self.configuration)
self.mkcalendar("/", check=401)
self.mkcalendar("/user/", check=401)
self.mkcol("/user/")
@@ -183,9 +163,8 @@ permissions: i""")
self.mkcol("/user/calendar/item", check=401)
self.mkcalendar("/user/calendar/item", check=401)
def test_put_collections_and_items(self):
def test_put_collections_and_items(self) -> None:
"""Test rights for creation of calendars and items with PUT."""
self.application = Application(self.configuration)
self.put("/user/", "BEGIN:VCALENDAR\r\nEND:VCALENDAR", check=401)
self.mkcol("/user/")
self.put("/user/calendar/", "BEGIN:VCALENDAR\r\nEND:VCALENDAR")

137
radicale/tests/test_server.py
View File

@@ -1,4 +1,4 @@
# This file is part of Radicale Server - Calendar Server
# This file is part of Radicale - CalDAV and CardDAV server
# Copyright © 2018-2019 Unrud <unrud@outlook.com>
#
# This library is free software: you can redistribute it and/or modify
@@ -21,15 +21,15 @@ Test the internal server.
import errno
import os
import shutil
import socket
import ssl
import subprocess
import sys
import tempfile
import threading
import time
from configparser import RawConfigParser
from http.client import HTTPMessage
from typing import IO, Callable, Dict, Optional, Tuple, cast
from urllib import request
from urllib.error import HTTPError, URLError
@@ -41,31 +41,29 @@ from radicale.tests.helpers import configuration_to_dict, get_file_path
class DisabledRedirectHandler(request.HTTPRedirectHandler):
def http_error_302(self, req, fp, code, msg, headers):
raise HTTPError(req.full_url, code, msg, headers, fp)
http_error_301 = http_error_303 = http_error_307 = http_error_302
def redirect_request(
self, req: request.Request, fp: IO[bytes], code: int, msg: str,
headers: HTTPMessage, newurl: str) -> None:
return None
class TestBaseServerRequests(BaseTest):
"""Test the internal server."""
def setup(self):
self.configuration = config.load()
self.colpath = tempfile.mkdtemp()
shutdown_socket: socket.socket
thread: threading.Thread
opener: request.OpenerDirector
def setup(self) -> None:
super().setup()
self.shutdown_socket, shutdown_socket_out = socket.socketpair()
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
# Find available port
sock.bind(("127.0.0.1", 0))
self.sockname = sock.getsockname()
self.configuration.update({
"storage": {"filesystem_folder": self.colpath,
# Disable syncing to disk for better performance
"_filesystem_fsync": "False"},
"server": {"hosts": "[%s]:%d" % self.sockname},
# Enable debugging for new processes
"logging": {"level": "debug"}},
"test", privileged=True)
self.configure({"server": {"hosts": "[%s]:%d" % self.sockname},
# Enable debugging for new processes
"logging": {"level": "debug"}})
self.thread = threading.Thread(target=server.serve, args=(
self.configuration, shutdown_socket_out))
ssl_context = ssl.create_default_context()
@@ -75,48 +73,67 @@ class TestBaseServerRequests(BaseTest):
request.HTTPSHandler(context=ssl_context),
DisabledRedirectHandler)
def teardown(self):
def teardown(self) -> None:
self.shutdown_socket.close()
try:
self.thread.join()
except RuntimeError: # Thread never started
pass
shutil.rmtree(self.colpath)
super().teardown()
def request(self, method, path, data=None, is_alive_fn=None, **headers):
def request(self, method: str, path: str, data: Optional[str] = None,
check: Optional[int] = None, **kwargs
) -> Tuple[int, Dict[str, str], str]:
"""Send a request."""
login = kwargs.pop("login", None)
if login is not None and not isinstance(login, str):
raise TypeError("login argument must be %r, not %r" %
(str, type(login)))
if login:
raise NotImplementedError
is_alive_fn: Optional[Callable[[], bool]] = kwargs.pop(
"is_alive_fn", None)
headers: Dict[str, str] = kwargs
for k, v in headers.items():
if not isinstance(v, str):
raise TypeError("type of %r is %r, expected %r" %
(k, type(v), str))
if is_alive_fn is None:
is_alive_fn = self.thread.is_alive
scheme = ("https" if self.configuration.get("server", "ssl") else
"http")
encoding: str = self.configuration.get("encoding", "request")
scheme = "https" if self.configuration.get("server", "ssl") else "http"
data_bytes = None
if data:
data_bytes = data.encode(encoding)
req = request.Request(
"%s://[%s]:%d%s" % (scheme, *self.sockname, path),
data=data, headers=headers, method=method)
data=data_bytes, headers=headers, method=method)
while True:
assert is_alive_fn()
try:
with self.opener.open(req) as f:
return f.getcode(), f.info(), f.read().decode()
return f.getcode(), dict(f.info()), f.read().decode()
except HTTPError as e:
return e.code, e.headers, e.read().decode()
assert check is None or e.code == check, "%d != %d" % (e.code,
check)
return e.code, dict(e.headers), e.read().decode()
except URLError as e:
if not isinstance(e.reason, ConnectionRefusedError):
raise
time.sleep(0.1)
def test_root(self):
def test_root(self) -> None:
self.thread.start()
self.get("/", check=302)
def test_ssl(self):
self.configuration.update({
"server": {"ssl": "True",
"certificate": get_file_path("cert.pem"),
"key": get_file_path("key.pem")}}, "test")
def test_ssl(self) -> None:
self.configure({"server": {"ssl": "True",
"certificate": get_file_path("cert.pem"),
"key": get_file_path("key.pem")}})
self.thread.start()
self.get("/", check=302)
def test_bind_fail(self):
def test_bind_fail(self) -> None:
for address_family, address in [(socket.AF_INET, "::1"),
(socket.AF_INET6, "127.0.0.1")]:
with socket.socket(address_family, socket.SOCK_STREAM) as sock:
@@ -132,10 +149,11 @@ class TestBaseServerRequests(BaseTest):
socket.EAI_NONAME, server.COMPAT_EAI_ADDRFAMILY,
server.COMPAT_EAI_NODATA) or
str(exc_info.value) == "address family mismatched" or
exc_info.value.errno == errno.EADDRNOTAVAIL or
exc_info.value.errno == errno.EAFNOSUPPORT)
exc_info.value.errno in (
errno.EADDRNOTAVAIL, errno.EAFNOSUPPORT,
errno.EPROTONOSUPPORT))
def test_ipv6(self):
def test_ipv6(self) -> None:
try:
with socket.socket(socket.AF_INET6, socket.SOCK_STREAM) as sock:
# Only allow IPv6 connections to the IPv6 socket
@@ -145,44 +163,54 @@ class TestBaseServerRequests(BaseTest):
sock.bind(("::1", 0))
self.sockname = sock.getsockname()[:2]
except OSError as e:
if e.errno in (errno.EADDRNOTAVAIL, errno.EAFNOSUPPORT):
if e.errno in (errno.EADDRNOTAVAIL, errno.EAFNOSUPPORT,
errno.EPROTONOSUPPORT):
pytest.skip("IPv6 not supported")
raise
self.configuration.update({
"server": {"hosts": "[%s]:%d" % self.sockname}}, "test")
self.configure({"server": {"hosts": "[%s]:%d" % self.sockname}})
self.thread.start()
self.get("/", check=302)
def test_command_line_interface(self):
def test_command_line_interface(self, with_bool_options=False) -> None:
self.configure({"headers": {"Test-Server": "test"}})
config_args = []
for section, values in config.DEFAULT_CONFIG_SCHEMA.items():
for section in self.configuration.sections():
if section.startswith("_"):
continue
for option, data in values.items():
for option in self.configuration.options(section):
if option.startswith("_"):
continue
long_name = "--%s-%s" % (section, option.replace("_", "-"))
if data["type"] == bool:
if not self.configuration.get(section, option):
if with_bool_options and config.DEFAULT_CONFIG_SCHEMA.get(
section, {}).get(option, {}).get("type") == bool:
if not cast(bool, self.configuration.get(section, option)):
long_name = "--no%s" % long_name[1:]
config_args.append(long_name)
else:
config_args.append(long_name)
config_args.append(
self.configuration.get_raw(section, option))
env = os.environ.copy()
env["PYTHONPATH"] = os.pathsep.join(sys.path)
raw_value = self.configuration.get_raw(section, option)
assert isinstance(raw_value, str)
config_args.append(raw_value)
config_args.append("--headers-Test-Header=test")
p = subprocess.Popen(
[sys.executable, "-m", "radicale"] + config_args, env=env)
[sys.executable, "-m", "radicale"] + config_args,
env={**os.environ, "PYTHONPATH": os.pathsep.join(sys.path)})
try:
self.get("/", is_alive_fn=lambda: p.poll() is None, check=302)
status, headers, _ = self.request(
"GET", "/", check=302, is_alive_fn=lambda: p.poll() is None)
for key in self.configuration.options("headers"):
assert headers.get(key) == self.configuration.get(
"headers", key)
finally:
p.terminate()
p.wait()
if os.name == "posix":
if sys.platform != "win32":
assert p.returncode == 0
def test_wsgi_server(self):
def test_command_line_interface_with_bool_options(self) -> None:
self.test_command_line_interface(with_bool_options=True)
def test_wsgi_server(self) -> None:
config_path = os.path.join(self.colpath, "config")
parser = RawConfigParser()
parser.read_dict(configuration_to_dict(self.configuration))
@@ -191,9 +219,10 @@ class TestBaseServerRequests(BaseTest):
env = os.environ.copy()
env["PYTHONPATH"] = os.pathsep.join(sys.path)
env["RADICALE_CONFIG"] = config_path
raw_server_hosts = self.configuration.get_raw("server", "hosts")
assert isinstance(raw_server_hosts, str)
p = subprocess.Popen([
sys.executable, "-m", "waitress",
"--listen", self.configuration.get_raw("server", "hosts"),
sys.executable, "-m", "waitress", "--listen", raw_server_hosts,
"radicale:application"], env=env)
try:
self.get("/", is_alive_fn=lambda: p.poll() is None, check=302)

189
radicale/tests/test_storage.py Normal file
View File

@@ -0,0 +1,189 @@
# This file is part of Radicale - CalDAV and CardDAV server
# Copyright © 2012-2017 Guillaume Ayoub
# Copyright © 2017-2019 Unrud <unrud@outlook.com>
#
# This library is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Radicale. If not, see <http://www.gnu.org/licenses/>.
"""
Tests for storage backends.
"""
import os
import shutil
from typing import ClassVar, cast
import pytest
import radicale.tests.custom.storage_simple_sync
from radicale.tests import BaseTest
from radicale.tests.helpers import get_file_content
from radicale.tests.test_base import TestBaseRequests as _TestBaseRequests
class TestMultiFileSystem(BaseTest):
"""Tests for multifilesystem."""
def setup(self) -> None:
_TestBaseRequests.setup(cast(_TestBaseRequests, self))
self.configure({"storage": {"type": "multifilesystem"}})
def test_folder_creation(self) -> None:
"""Verify that the folder is created."""
folder = os.path.join(self.colpath, "subfolder")
self.configure({"storage": {"filesystem_folder": folder}})
assert os.path.isdir(folder)
def test_fsync(self) -> None:
"""Create a directory and file with syncing enabled."""
self.configure({"storage": {"_filesystem_fsync": "True"}})
self.mkcalendar("/calendar.ics/")
def test_hook(self) -> None:
"""Run hook."""
self.configure({"storage": {"hook": "mkdir %s" % os.path.join(
"collection-root", "created_by_hook")}})
self.mkcalendar("/calendar.ics/")
self.propfind("/created_by_hook/")
def test_hook_read_access(self) -> None:
"""Verify that hook is not run for read accesses."""
self.configure({"storage": {"hook": "mkdir %s" % os.path.join(
"collection-root", "created_by_hook")}})
self.propfind("/")
self.propfind("/created_by_hook/", check=404)
@pytest.mark.skipif(not shutil.which("flock"),
reason="flock command not found")
def test_hook_storage_locked(self) -> None:
"""Verify that the storage is locked when the hook runs."""
self.configure({"storage": {"hook": (
"flock -n .Radicale.lock || exit 0; exit 1")}})
self.mkcalendar("/calendar.ics/")
def test_hook_principal_collection_creation(self) -> None:
"""Verify that the hooks runs when a new user is created."""
self.configure({"storage": {"hook": "mkdir %s" % os.path.join(
"collection-root", "created_by_hook")}})
self.propfind("/", login="user:")
self.propfind("/created_by_hook/")
def test_hook_fail(self) -> None:
"""Verify that a request fails if the hook fails."""
self.configure({"storage": {"hook": "exit 1"}})
self.mkcalendar("/calendar.ics/", check=500)
def test_item_cache_rebuild(self) -> None:
"""Delete the item cache and verify that it is rebuild."""
self.mkcalendar("/calendar.ics/")
event = get_file_content("event1.ics")
path = "/calendar.ics/event1.ics"
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)
_, answer2 = self.get(path)
assert answer1 == answer2
assert os.path.exists(os.path.join(cache_folder, "event1.ics"))
def test_put_whole_calendar_uids_used_as_file_names(self) -> None:
"""Test if UIDs are used as file names."""
_TestBaseRequests.test_put_whole_calendar(
cast(_TestBaseRequests, self))
for uid in ("todo", "event"):
_, answer = self.get("/calendar.ics/%s.ics" % uid)
assert "\r\nUID:%s\r\n" % uid in answer
def test_put_whole_calendar_random_uids_used_as_file_names(self) -> None:
"""Test if UIDs are used as file names."""
_TestBaseRequests.test_put_whole_calendar_without_uids(
cast(_TestBaseRequests, self))
_, answer = self.get("/calendar.ics")
assert answer is not None
uids = []
for line in answer.split("\r\n"):
if line.startswith("UID:"):
uids.append(line[len("UID:"):])
for uid in uids:
_, answer = self.get("/calendar.ics/%s.ics" % uid)
assert answer is not None
assert "\r\nUID:%s\r\n" % uid in answer
def test_put_whole_addressbook_uids_used_as_file_names(self) -> None:
"""Test if UIDs are used as file names."""
_TestBaseRequests.test_put_whole_addressbook(
cast(_TestBaseRequests, self))
for uid in ("contact1", "contact2"):
_, answer = self.get("/contacts.vcf/%s.vcf" % uid)
assert "\r\nUID:%s\r\n" % uid in answer
def test_put_whole_addressbook_random_uids_used_as_file_names(
self) -> None:
"""Test if UIDs are used as file names."""
_TestBaseRequests.test_put_whole_addressbook_without_uids(
cast(_TestBaseRequests, self))
_, answer = self.get("/contacts.vcf")
assert answer is not None
uids = []
for line in answer.split("\r\n"):
if line.startswith("UID:"):
uids.append(line[len("UID:"):])
for uid in uids:
_, answer = self.get("/contacts.vcf/%s.vcf" % uid)
assert answer is not None
assert "\r\nUID:%s\r\n" % uid in answer
class TestMultiFileSystemNoLock(BaseTest):
"""Tests for multifilesystem_nolock."""
def setup(self) -> None:
_TestBaseRequests.setup(cast(_TestBaseRequests, self))
self.configure({"storage": {"type": "multifilesystem_nolock"}})
test_add_event = _TestBaseRequests.test_add_event
test_item_cache_rebuild = TestMultiFileSystem.test_item_cache_rebuild
class TestCustomStorageSystem(BaseTest):
"""Test custom backend loading."""
def setup(self) -> None:
_TestBaseRequests.setup(cast(_TestBaseRequests, self))
self.configure({"storage": {
"type": "radicale.tests.custom.storage_simple_sync"}})
full_sync_token_support: ClassVar[bool] = False
test_add_event = _TestBaseRequests.test_add_event
_report_sync_token = _TestBaseRequests._report_sync_token
# include tests related to sync token
s: str = ""
for s in dir(_TestBaseRequests):
if s.startswith("test_") and "sync" in s.split("_"):
locals()[s] = getattr(_TestBaseRequests, s)
del s
class TestCustomStorageSystemCallable(BaseTest):
"""Test custom backend loading with ``callable``."""
def setup(self) -> None:
_TestBaseRequests.setup(cast(_TestBaseRequests, self))
self.configure({"storage": {
"type": radicale.tests.custom.storage_simple_sync.Storage}})
test_add_event = _TestBaseRequests.test_add_event

44
radicale/tests/test_web.py
View File

@@ -1,4 +1,4 @@
# This file is part of Radicale Server - Calendar Server
# This file is part of Radicale - CalDAV and CardDAV server
# Copyright © 2018-2019 Unrud <unrud@outlook.com>
#
# This library is free software: you can redistribute it and/or modify
@@ -19,47 +19,31 @@ Test web plugin.
"""
import shutil
import tempfile
from radicale import Application, config
from radicale.tests import BaseTest
class TestBaseWebRequests(BaseTest):
"""Test web plugin."""
def setup(self):
self.configuration = config.load()
self.colpath = tempfile.mkdtemp()
self.configuration.update({
"storage": {"filesystem_folder": self.colpath,
# Disable syncing to disk for better performance
"_filesystem_fsync": "False"}},
"test", privileged=True)
self.application = Application(self.configuration)
def teardown(self):
shutil.rmtree(self.colpath)
def test_internal(self):
status, headers, _ = self.request("GET", "/.web")
assert status == 302
assert headers.get("Location") == ".web/"
def test_internal(self) -> None:
_, headers, _ = self.request("GET", "/.web", check=302)
assert headers.get("Location") == "/.web/"
_, answer = self.get("/.web/")
assert answer
self.post("/.web", check=405)
def test_none(self):
self.configuration.update({"web": {"type": "none"}}, "test")
self.application = Application(self.configuration)
def test_none(self) -> None:
self.configure({"web": {"type": "none"}})
_, answer = self.get("/.web")
assert answer
self.get("/.web/", check=404)
_, headers, _ = self.request("GET", "/.web/", check=302)
assert headers.get("Location") == "/.web"
self.post("/.web", check=405)
def test_custom(self):
def test_custom(self) -> None:
"""Custom web plugin."""
self.configuration.update({
"web": {"type": "radicale.tests.custom.web"}}, "test")
self.application = Application(self.configuration)
self.configure({"web": {"type": "radicale.tests.custom.web"}})
_, answer = self.get("/.web")
assert answer == "custom"
_, answer = self.post("/.web", "body content")
assert answer == "echo:body content"