diff --git a/radicale/tests/__init__.py b/radicale/tests/__init__.py index 2a12a11..0833fc4 100644 --- a/radicale/tests/__init__.py +++ b/radicale/tests/__init__.py @@ -32,7 +32,7 @@ from typing import Any, Dict, List, Optional, Tuple, Union import defusedxml.ElementTree as DefusedET import radicale -from radicale import app, config, xmlutils +from radicale import app, config, types, xmlutils RESPONSES = Dict[str, Union[int, Dict[str, Tuple[int, ET.Element]]]] @@ -50,12 +50,15 @@ class BaseTest: def setup(self) -> None: self.configuration = config.load() self.colpath = tempfile.mkdtemp() - self.configuration.update({ + 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"}}, "test", privileged=True) + "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: diff --git a/radicale/tests/test_auth.py b/radicale/tests/test_auth.py index 1282970..516a6f8 100644 --- a/radicale/tests/test_auth.py +++ b/radicale/tests/test_auth.py @@ -27,7 +27,7 @@ from typing import Iterable, Tuple, Union import pytest -from radicale import Application, xmlutils +from radicale import xmlutils from radicale.tests import BaseTest @@ -58,11 +58,9 @@ class TestBaseAuthRequests(BaseTest): 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), @@ -121,8 +119,7 @@ class TestBaseAuthRequests(BaseTest): self._test_htpasswd("plain", "#comment\n #comment\n \ntmp:bepo\n\n") def test_remote_user(self) -> None: - self.configuration.update({"auth": {"type": "remote_user"}}, "test") - self.application = Application(self.configuration) + self.configure({"auth": {"type": "remote_user"}}) _, responses = self.propfind("/", """\ @@ -139,9 +136,7 @@ class TestBaseAuthRequests(BaseTest): assert href_element is not None and href_element.text == "/test/" def test_http_x_remote_user(self) -> None: - self.configuration.update( - {"auth": {"type": "http_x_remote_user"}}, "test") - self.application = Application(self.configuration) + self.configure({"auth": {"type": "http_x_remote_user"}}) _, responses = self.propfind("/", """\ @@ -159,7 +154,5 @@ class TestBaseAuthRequests(BaseTest): 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:") diff --git a/radicale/tests/test_base.py b/radicale/tests/test_base.py index d317d7c..bf85130 100644 --- a/radicale/tests/test_base.py +++ b/radicale/tests/test_base.py @@ -31,7 +31,7 @@ import defusedxml.ElementTree as DefusedET import pytest import radicale.tests.custom.storage_simple_sync -from radicale import Application, config, storage, xmlutils +from radicale import config, storage, xmlutils from radicale.tests import RESPONSES, BaseTest from radicale.tests.helpers import get_file_content @@ -1544,12 +1544,10 @@ class BaseRequestsMixIn(BaseTest): def test_authentication(self) -> None: """Test if server sends authentication request.""" - self.configuration.update({ - "auth": {"type": "htpasswd", - "htpasswd_filename": os.devnull, - "htpasswd_encryption": "plain"}, - "rights": {"type": "owner_only"}}, "test") - self.application = Application(self.configuration) + self.configure({"auth": {"type": "htpasswd", + "htpasswd_filename": os.devnull, + "htpasswd_encryption": "plain"}, + "rights": {"type": "owner_only"}}) status, headers, _ = self.request("MKCOL", "/user/") assert status in (401, 403) assert headers.get("WWW-Authenticate") @@ -1580,8 +1578,7 @@ class BaseRequestsMixIn(BaseTest): self.propfind("/") def test_custom_headers(self) -> None: - self.configuration.update({"headers": {"test": "123"}}, "test") - self.application = Application(self.configuration) + self.configure({"headers": {"test": "123"}}) # Test if header is set on success status, headers, _ = self.request("OPTIONS", "/") assert status == 200 @@ -1615,11 +1612,9 @@ class BaseStorageTest(BaseTest): user: .* collection: .* permissions: RrWw""") - self.configuration.update({ - "storage": {"type": self.storage_type}, - "rights": {"file": rights_file_path, - "type": "from_file"}}, "test", privileged=True) - self.application = Application(self.configuration) + self.configure({"storage": {"type": self.storage_type}, + "rights": {"file": rights_file_path, + "type": "from_file"}}) class TestMultiFileSystem(BaseStorageTest, BaseRequestsMixIn): @@ -1630,33 +1625,25 @@ class TestMultiFileSystem(BaseStorageTest, BaseRequestsMixIn): def test_folder_creation(self) -> None: """Verify that the folder is created.""" folder = os.path.join(self.colpath, "subfolder") - self.configuration.update( - {"storage": {"filesystem_folder": folder}}, "test") - self.application = Application(self.configuration) + 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.configuration.update({"storage": {"_filesystem_fsync": "True"}}, - "test", privileged=True) - self.application = Application(self.configuration) + self.configure({"storage": {"_filesystem_fsync": "True"}}) self.mkcalendar("/calendar.ics/") def test_hook(self) -> None: """Run hook.""" - self.configuration.update({"storage": { - "hook": ("mkdir %s" % os.path.join( - "collection-root", "created_by_hook"))}}, "test") - self.application = Application(self.configuration) + 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.configuration.update({"storage": { - "hook": ("mkdir %s" % os.path.join( - "collection-root", "created_by_hook"))}}, "test") - self.application = Application(self.configuration) + self.configure({"storage": {"hook": "mkdir %s" % os.path.join( + "collection-root", "created_by_hook")}}) self.propfind("/") self.propfind("/created_by_hook/", check=404) @@ -1664,24 +1651,20 @@ class TestMultiFileSystem(BaseStorageTest, BaseRequestsMixIn): reason="flock command not found") def test_hook_storage_locked(self) -> None: """Verify that the storage is locked when the hook runs.""" - self.configuration.update({"storage": {"hook": ( - "flock -n .Radicale.lock || exit 0; exit 1")}}, "test") - self.application = Application(self.configuration) + 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.configuration.update({"storage": { - "hook": ("mkdir %s" % os.path.join( - "collection-root", "created_by_hook"))}}, "test") - self.application = Application(self.configuration) + 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.configuration.update({"storage": {"hook": "exit 1"}}, "test") - self.application = Application(self.configuration) + self.configure({"storage": {"hook": "exit 1"}}) self.mkcalendar("/calendar.ics/", check=500) def test_item_cache_rebuild(self) -> None: diff --git a/radicale/tests/test_rights.py b/radicale/tests/test_rights.py index c1a4d43..c8efa4b 100644 --- a/radicale/tests/test_rights.py +++ b/radicale/tests/test_rights.py @@ -20,7 +20,6 @@ Radicale tests with simple requests and rights. import os -from radicale import Application from radicale.tests import BaseTest from radicale.tests.helpers import get_file_content @@ -35,12 +34,11 @@ class TestBaseRightsRequests(BaseTest): 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) @@ -113,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) @@ -134,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,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/") @@ -171,7 +165,6 @@ permissions: i""") 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") diff --git a/radicale/tests/test_server.py b/radicale/tests/test_server.py index c6555f1..e2f3b13 100644 --- a/radicale/tests/test_server.py +++ b/radicale/tests/test_server.py @@ -76,11 +76,9 @@ class TestBaseServerRequests(BaseTest): # Find available port sock.bind(("127.0.0.1", 0)) self.sockname = sock.getsockname() - self.configuration.update({ - "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() @@ -141,10 +139,9 @@ class TestBaseServerRequests(BaseTest): self.get("/", check=302) def test_ssl(self) -> None: - self.configuration.update({ - "server": {"ssl": "True", - "certificate": get_file_path("cert.pem"), - "key": get_file_path("key.pem")}}, "test") + self.configure({"server": {"ssl": "True", + "certificate": get_file_path("cert.pem"), + "key": get_file_path("key.pem")}}) self.thread.start() self.get("/", check=302) @@ -182,13 +179,12 @@ class TestBaseServerRequests(BaseTest): 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, with_bool_options=False) -> None: - self.configuration.update({"headers": {"Test-Server": "test"}}) + self.configure({"headers": {"Test-Server": "test"}}) config_args = [] for section in self.configuration.sections(): if section.startswith("_"): diff --git a/radicale/tests/test_web.py b/radicale/tests/test_web.py index bfa2c72..9432166 100644 --- a/radicale/tests/test_web.py +++ b/radicale/tests/test_web.py @@ -19,7 +19,6 @@ Test web plugin. """ -from radicale import Application from radicale.tests import BaseTest @@ -35,8 +34,7 @@ class TestBaseWebRequests(BaseTest): self.post("/.web", check=405) def test_none(self) -> None: - self.configuration.update({"web": {"type": "none"}}, "test") - self.application = Application(self.configuration) + self.configure({"web": {"type": "none"}}) _, answer = self.get("/.web") assert answer self.get("/.web/", check=404) @@ -44,9 +42,7 @@ class TestBaseWebRequests(BaseTest): 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")