Rework configuration

radicale/tests/helpers.py

@@ -2,7 +2,7 @@
 # Copyright © 2008 Nicolas Kandel
 # Copyright © 2008 Pascal Halter
 # Copyright © 2008-2017 Guillaume Ayoub
-# Copyright © 2017-2018 Unrud <unrud@outlook.com>
+# 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
@@ -39,3 +39,14 @@ def get_file_content(file_name):
             return fd.read()
     except IOError:
         print("Couldn't open the file %s" % file_name)
+
+
+def configuration_to_dict(configuration):
+    d = {}
+    for section in configuration.sections():
+        if configuration._schema[section].get("_internal", False):
+            continue
+        d[section] = {}
+        for option in configuration.options(section):
+            d[section][option] = configuration.get_raw(section, option)
+    return d

radicale/tests/test_auth.py

@@ -1,7 +1,7 @@
 # This file is part of Radicale Server - Calendar Server
 # Copyright © 2012-2016 Jean-Marc Martins
 # Copyright © 2012-2017 Guillaume Ayoub
-# Copyright © 2017-2018 Unrud <unrud@outlook.com>
+# 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
@@ -42,11 +42,12 @@ class TestBaseAuthRequests(BaseTest):
     def setup(self):
         self.configuration = config.load()
         self.colpath = tempfile.mkdtemp()
-        self.configuration["storage"]["filesystem_folder"] = self.colpath
-        # Disable syncing to disk for better performance
-        self.configuration["internal"]["filesystem_fsync"] = "False"
-        # Set incorrect authentication delay to a very low value
-        self.configuration["auth"]["delay"] = "0.002"
+        self.configuration.update({
+            "storage": {"filesystem_folder": self.colpath},
+            # Disable syncing to disk for better performance
+            "internal": {"filesystem_fsync": "False"},
+            # Set incorrect authentication delay to a very low value
+            "auth": {"delay": "0.002"}}, "test")
 
     def teardown(self):
         shutil.rmtree(self.colpath)
@@ -57,9 +58,10 @@ class TestBaseAuthRequests(BaseTest):
         htpasswd_file_path = os.path.join(self.colpath, ".htpasswd")
         with open(htpasswd_file_path, "w") as f:
             f.write(htpasswd_content)
-        self.configuration["auth"]["type"] = "htpasswd"
-        self.configuration["auth"]["htpasswd_filename"] = htpasswd_file_path
-        self.configuration["auth"]["htpasswd_encryption"] = htpasswd_encryption
+        self.configuration.update({
+            "auth": {"type": "htpasswd",
+                     "htpasswd_filename": htpasswd_file_path,
+                     "htpasswd_encryption": htpasswd_encryption}}, "test")
         self.application = Application(self.configuration)
         if test_matrix is None:
             test_matrix = (
@@ -129,7 +131,7 @@ class TestBaseAuthRequests(BaseTest):
         self._test_htpasswd("plain", "#comment\n #comment\n \ntmp:bepo\n\n")
 
     def test_remote_user(self):
-        self.configuration["auth"]["type"] = "remote_user"
+        self.configuration.update({"auth": {"type": "remote_user"}}, "test")
         self.application = Application(self.configuration)
         status, _, answer = self.request(
             "PROPFIND", "/",
@@ -143,7 +145,8 @@ class TestBaseAuthRequests(BaseTest):
         assert ">/test/<" in answer
 
     def test_http_x_remote_user(self):
-        self.configuration["auth"]["type"] = "http_x_remote_user"
+        self.configuration.update(
+            {"auth": {"type": "http_x_remote_user"}}, "test")
         self.application = Application(self.configuration)
         status, _, answer = self.request(
             "PROPFIND", "/",
@@ -158,7 +161,8 @@ class TestBaseAuthRequests(BaseTest):
 
     def test_custom(self):
         """Custom authentication."""
-        self.configuration["auth"]["type"] = "tests.custom.auth"
+        self.configuration.update(
+            {"auth": {"type": "tests.custom.auth"}}, "test")
         self.application = Application(self.configuration)
         status, _, answer = self.request(
             "PROPFIND", "/tmp", HTTP_AUTHORIZATION="Basic %s" %

radicale/tests/test_base.py

@@ -1,6 +1,6 @@
 # This file is part of Radicale Server - Calendar Server
 # Copyright © 2012-2017 Guillaume Ayoub
-# Copyright © 2017-2018 Unrud <unrud@outlook.com>
+# 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
@@ -1404,10 +1404,11 @@ class BaseRequestsMixIn:
 
     def test_authentication(self):
         """Test if server sends authentication request."""
-        self.configuration["auth"]["type"] = "htpasswd"
-        self.configuration["auth"]["htpasswd_filename"] = os.devnull
-        self.configuration["auth"]["htpasswd_encryption"] = "plain"
-        self.configuration["rights"]["type"] = "owner_only"
+        self.configuration.update({
+            "auth": {"type": "htpasswd",
+                     "htpasswd_filename": os.devnull,
+                     "htpasswd_encryption": "plain"},
+            "rights": {"type": "owner_only"}}, "test")
         self.application = Application(self.configuration)
         status, headers, _ = self.request("MKCOL", "/user/")
         assert status in (401, 403)
@@ -1431,9 +1432,8 @@ class BaseRequestsMixIn:
         assert status == 207
 
     def test_custom_headers(self):
-        if not self.configuration.has_section("headers"):
-            self.configuration.add_section("headers")
-        self.configuration.set("headers", "test", "123")
+        self.configuration.update({"headers": {"test": "123"}}, "test")
+        self.application = Application(self.configuration)
         # Test if header is set on success
         status, headers, _ = self.request("OPTIONS", "/")
         assert status == 200
@@ -1461,11 +1461,7 @@ class BaseFileSystemTest(BaseTest):
 
     def setup(self):
         self.configuration = config.load()
-        self.configuration["storage"]["type"] = self.storage_type
         self.colpath = tempfile.mkdtemp()
-        self.configuration["storage"]["filesystem_folder"] = self.colpath
-        # Disable syncing to disk for better performance
-        self.configuration["internal"]["filesystem_fsync"] = "False"
         # Allow access to anything for tests
         rights_file_path = os.path.join(self.colpath, "rights")
         with open(rights_file_path, "w") as f:
@@ -1474,8 +1470,13 @@ class BaseFileSystemTest(BaseTest):
 user: .*
 collection: .*
 permissions: RrWw""")
-        self.configuration["rights"]["file"] = rights_file_path
-        self.configuration["rights"]["type"] = "from_file"
+        self.configuration.update({
+            "storage": {"type": self.storage_type,
+                        "filesystem_folder": self.colpath},
+            # Disable syncing to disk for better performance
+            "internal": {"filesystem_fsync": "False"},
+            "rights": {"file": rights_file_path,
+                       "type": "from_file"}}, "test")
         self.application = Application(self.configuration)
 
     def teardown(self):
@@ -1488,14 +1489,18 @@ class TestMultiFileSystem(BaseFileSystemTest, BaseRequestsMixIn):
 
     def test_fsync(self):
         """Create a directory and file with syncing enabled."""
-        self.configuration["internal"]["filesystem_fsync"] = "True"
+        self.configuration.update({
+            "internal": {"filesystem_fsync": "True"}}, "test")
+        self.application = Application(self.configuration)
         status, _, _ = self.request("MKCALENDAR", "/calendar.ics/")
         assert status == 201
 
     def test_hook(self):
         """Run hook."""
-        self.configuration["storage"]["hook"] = (
+        self.configuration.update({"storage": {"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/")
@@ -1503,8 +1508,10 @@ class TestMultiFileSystem(BaseFileSystemTest, BaseRequestsMixIn):
 
     def test_hook_read_access(self):
         """Verify that hook is not run for read accesses."""
-        self.configuration["storage"]["hook"] = (
+        self.configuration.update({"storage": {"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/")
@@ -1514,15 +1521,18 @@ class TestMultiFileSystem(BaseFileSystemTest, BaseRequestsMixIn):
                         reason="flock command not found")
     def test_hook_storage_locked(self):
         """Verify that the storage is locked when the hook runs."""
-        self.configuration["storage"]["hook"] = (
-            "flock -n .Radicale.lock || exit 0; exit 1")
+        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
 
     def test_hook_principal_collection_creation(self):
         """Verify that the hooks runs when a new user is created."""
-        self.configuration["storage"]["hook"] = (
+        self.configuration.update({"storage": {"hook": (
             "mkdir %s" % os.path.join("collection-root", "created_by_hook"))
+        }}, "test")
+        self.application = Application(self.configuration)
         status, _, _ = self.request("PROPFIND", "/", HTTP_AUTHORIZATION=(
             "Basic " + base64.b64encode(b"user:").decode()))
         assert status == 207
@@ -1531,7 +1541,8 @@ class TestMultiFileSystem(BaseFileSystemTest, BaseRequestsMixIn):
 
     def test_hook_fail(self):
         """Verify that a request fails if the hook fails."""
-        self.configuration["storage"]["hook"] = "exit 1"
+        self.configuration.update({"storage": {"hook": "exit 1"}}, "test")
+        self.application = Application(self.configuration)
         status, _, _ = self.request("MKCALENDAR", "/calendar.ics/")
         assert status != 201
 

radicale/tests/test_config.py

@@ -0,0 +1,182 @@
+# This file is part of Radicale Server - Calendar Server
+# Copyright © 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/>.
+
+import os
+import shutil
+import tempfile
+from configparser import RawConfigParser
+
+import pytest
+
+from radicale import config
+
+from .helpers import configuration_to_dict
+
+
+class TestConfig:
+    """Test the configuration."""
+
+    def setup(self):
+        self.colpath = tempfile.mkdtemp()
+
+    def teardown(self):
+        shutil.rmtree(self.colpath)
+
+    def _write_config(self, config_dict, name):
+        parser = RawConfigParser()
+        parser.read_dict(config_dict)
+        config_path = os.path.join(self.colpath, name)
+        with open(config_path, "w") as f:
+            parser.write(f)
+        return config_path
+
+    def test_parse_compound_paths(self):
+        assert len(config.parse_compound_paths()) == 0
+        assert len(config.parse_compound_paths("")) == 0
+        assert len(config.parse_compound_paths(None, "")) == 0
+        assert len(config.parse_compound_paths("config", "")) == 0
+        assert len(config.parse_compound_paths("config", None)) == 1
+
+        assert len(config.parse_compound_paths(os.pathsep.join(["", ""]))) == 0
+        assert len(config.parse_compound_paths(os.pathsep.join([
+            "", "config", ""]))) == 1
+
+        paths = config.parse_compound_paths(os.pathsep.join([
+            "config1", "?config2", "config3"]))
+        assert len(paths) == 3
+        for i, (name, ignore_if_missing) in enumerate([
+                ("config1", False), ("config2", True), ("config3", False)]):
+            assert os.path.isabs(paths[i][0])
+            assert os.path.basename(paths[i][0]) == name
+            assert paths[i][1] is ignore_if_missing
+
+    def test_load_empty(self):
+        config_path = self._write_config({}, "config")
+        config.load([(config_path, False)])
+
+    def test_load_full(self):
+        config_path = self._write_config(
+            configuration_to_dict(config.load()), "config")
+        config.load([(config_path, False)])
+
+    def test_load_missing(self):
+        config_path = os.path.join(self.colpath, "does_not_exist")
+        config.load([(config_path, True)])
+        with pytest.raises(Exception) as exc_info:
+            config.load([(config_path, False)])
+        e = exc_info.value
+        assert ("Failed to load config file %r" % config_path) in str(e)
+
+    def test_load_multiple(self):
+        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)
+        assert configuration.get("server", "max_connections") == 1111
+
+    def test_copy(self):
+        configuration1 = config.load()
+        configuration1.update({"server": {"max_connections": "1111"}}, "test")
+        configuration2 = configuration1.copy()
+        configuration2.update({"server": {"max_connections": "1112"}}, "test")
+        assert configuration1.get("server", "max_connections") == 1111
+        assert configuration2.get("server", "max_connections") == 1112
+
+    def test_invalid_section(self):
+        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):
+        configuration = config.load()
+        with pytest.raises(Exception) as exc_info:
+            configuration.update({"server": {"x": "x"}}, "test")
+        e = exc_info.value
+        assert "Invalid option 'x'" in str(e)
+        assert "section 'server'" in str(e)
+
+    def test_invalid_option_plugin(self):
+        configuration = config.load()
+        with pytest.raises(Exception) as exc_info:
+            configuration.update({"auth": {"x": "x"}}, "test")
+        e = exc_info.value
+        assert "Invalid option 'x'" in str(e)
+        assert "section 'auth'" in str(e)
+
+    def test_invalid_value(self):
+        configuration = config.load()
+        with pytest.raises(Exception) as exc_info:
+            configuration.update({"server": {"max_connections": "x"}}, "test")
+        e = exc_info.value
+        assert "Invalid positive_int" in str(e)
+        assert "option 'max_connections" in str(e)
+        assert "section 'server" in str(e)
+        assert "'x'" in str(e)
+
+    def test_internal(self):
+        configuration = config.load()
+        configuration.update({"internal": {"internal_server": "True"}}, "test")
+        with pytest.raises(Exception) as exc_info:
+            configuration.update({"internal": {"internal_server": "True"}},
+                                 "test", internal=False)
+        e = exc_info.value
+        assert "Invalid section 'internal'" in str(e)
+
+    def test_plugin_schema(self):
+        PLUGIN_SCHEMA = {"auth": {"new_option": {"value": "False",
+                                                 "type": bool}}}
+        configuration = config.load()
+        configuration.update({"auth": {"type": "new_plugin"}}, "test")
+        plugin_configuration = configuration.copy(PLUGIN_SCHEMA)
+        assert plugin_configuration.get("auth", "new_option") is False
+        configuration.update({"auth": {"new_option": "True"}}, "test")
+        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}}}
+        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}}}
+        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": {}}
+        configuration = config.load()
+        configuration.update({"auth": {"type": "new_plugin",
+                                       "new_option": False}}, "test")
+        with pytest.raises(Exception) as exc_info:
+            configuration.copy(PLUGIN_SCHEMA)
+        e = exc_info.value
+        assert "Invalid option 'new_option'" in str(e)
+        assert "section 'auth'" in str(e)

radicale/tests/test_rights.py

@@ -1,5 +1,5 @@
 # This file is part of Radicale Server - Calendar Server
-# Copyright © 2017-2018 Unrud <unrud@outlook.com>
+# 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
@@ -35,9 +35,10 @@ class TestBaseRightsRequests(BaseTest):
     def setup(self):
         self.configuration = config.load()
         self.colpath = tempfile.mkdtemp()
-        self.configuration["storage"]["filesystem_folder"] = self.colpath
-        # Disable syncing to disk for better performance
-        self.configuration["internal"]["filesystem_fsync"] = "False"
+        self.configuration.update({
+            "storage": {"filesystem_folder": self.colpath},
+            # Disable syncing to disk for better performance
+            "internal": {"filesystem_fsync": "False"}}, "test")
 
     def teardown(self):
         shutil.rmtree(self.colpath)
@@ -49,11 +50,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["rights"]["type"] = rights_type
-        if with_auth:
-            self.configuration["auth"]["type"] = "htpasswd"
-        self.configuration["auth"]["htpasswd_filename"] = htpasswd_file_path
-        self.configuration["auth"]["htpasswd_encryption"] = "plain"
+        self.configuration.update({
+            "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)
         for u in ("tmp", "other"):
             status, _, _ = self.request(
@@ -132,7 +133,8 @@ permissions: RrWw
 user: .*
 collection: custom(/.*)?
 permissions: Rr""")
-        self.configuration["rights"]["file"] = rights_file_path
+        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", "", "/custom/sub", "r", 404)

radicale/tests/test_server.py

@@ -1,5 +1,5 @@
 # This file is part of Radicale Server - Calendar Server
-# Copyright © 2018 Unrud <unrud@outlook.com>
+# Copyright © 2018-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
@@ -28,7 +28,7 @@ import sys
 import tempfile
 import threading
 import time
-from configparser import ConfigParser
+from configparser import RawConfigParser
 from urllib import request
 from urllib.error import HTTPError, URLError
 
@@ -36,7 +36,7 @@ import pytest
 
 from radicale import config, server
 
-from .helpers import get_file_path
+from .helpers import configuration_to_dict, get_file_path
 
 try:
     import gunicorn
@@ -57,17 +57,18 @@ class TestBaseServerRequests:
     def setup(self):
         self.configuration = config.load()
         self.colpath = tempfile.mkdtemp()
-        self.configuration["storage"]["filesystem_folder"] = self.colpath
-        # Enable debugging for new processes
-        self.configuration["logging"]["level"] = "debug"
-        # Disable syncing to disk for better performance
-        self.configuration["internal"]["filesystem_fsync"] = "False"
         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["server"]["hosts"] = "[%s]:%d" % self.sockname
+        self.configuration.update({
+            "storage": {"filesystem_folder": self.colpath},
+            "server": {"hosts": "[%s]:%d" % self.sockname},
+            # Enable debugging for new processes
+            "logging": {"level": "debug"},
+            # Disable syncing to disk for better performance
+            "internal": {"filesystem_fsync": "False"}}, "test")
         self.thread = threading.Thread(target=server.serve, args=(
             self.configuration, shutdown_socket_out))
         ssl_context = ssl.create_default_context()
@@ -89,8 +90,8 @@ class TestBaseServerRequests:
         """Send a request."""
         if is_alive_fn is None:
             is_alive_fn = self.thread.is_alive
-        scheme = ("https" if self.configuration.getboolean("server", "ssl")
-                  else "http")
+        scheme = ("https" if self.configuration.get("server", "ssl") else
+                  "http")
         req = request.Request(
             "%s://[%s]:%d%s" % (scheme, *self.sockname, path),
             data=data, headers=headers, method=method)
@@ -112,9 +113,10 @@ class TestBaseServerRequests:
         assert status == 302
 
     def test_ssl(self):
-        self.configuration["server"]["ssl"] = "True"
-        self.configuration["server"]["certificate"] = get_file_path("cert.pem")
-        self.configuration["server"]["key"] = get_file_path("key.pem")
+        self.configuration.update({
+            "server": {"ssl": "True",
+                       "certificate": get_file_path("cert.pem"),
+                       "key": get_file_path("key.pem")}}, "test")
         self.thread.start()
         status, _, _ = self.request("GET", "/")
         assert status == 302
@@ -129,7 +131,8 @@ class TestBaseServerRequests:
             except OSError:
                 pytest.skip("IPv6 not supported")
             self.sockname = sock.getsockname()[:2]
-            self.configuration["server"]["hosts"] = "[%s]:%d" % self.sockname
+        self.configuration.update({
+            "server": {"hosts": "[%s]:%d" % self.sockname}}, "test")
         savedEaiAddrfamily = server.EAI_ADDRFAMILY
         if os.name == "nt" and server.EAI_ADDRFAMILY is None:
             # HACK: incomplete errno conversion in WINE
@@ -143,17 +146,22 @@ class TestBaseServerRequests:
 
     def test_command_line_interface(self):
         config_args = []
-        for section, values in config.INITIAL_CONFIG.items():
+        for section, values in config.DEFAULT_CONFIG_SCHEMA.items():
+            if values.get("_internal", False):
+                continue
             for option, data in values.items():
+                if option.startswith("_"):
+                    continue
                 long_name = "--{0}-{1}".format(
                     section, option.replace("_", "-"))
                 if data["type"] == bool:
-                    if not self.configuration.getboolean(section, option):
+                    if not self.configuration.get(section, option):
                         long_name = "--no{0}".format(long_name[1:])
                     config_args.append(long_name)
                 else:
                     config_args.append(long_name)
-                    config_args.append(self.configuration.get(section, option))
+                    config_args.append(
+                        self.configuration.get_raw(section, option))
         env = os.environ.copy()
         env["PYTHONPATH"] = os.pathsep.join(sys.path)
         p = subprocess.Popen(
@@ -170,18 +178,17 @@ class TestBaseServerRequests:
 
     @pytest.mark.skipif(not gunicorn, reason="gunicorn module not found")
     def test_wsgi_server(self):
-        config = ConfigParser()
-        config.read_dict(self.configuration)
-        assert config.remove_section("internal")
         config_path = os.path.join(self.colpath, "config")
+        parser = RawConfigParser()
+        parser.read_dict(configuration_to_dict(self.configuration))
         with open(config_path, "w") as f:
-            config.write(f)
+            parser.write(f)
         env = os.environ.copy()
         env["PYTHONPATH"] = os.pathsep.join(sys.path)
         p = subprocess.Popen([
             sys.executable,
             "-c", "from gunicorn.app.wsgiapp import run; run()",
-            "--bind", self.configuration["server"]["hosts"],
+            "--bind", self.configuration.get_raw("server", "hosts"),
             "--env", "RADICALE_CONFIG=%s" % config_path, "radicale"], env=env)
         try:
             status, _, _ = self.request(

radicale/tests/test_web.py

@@ -1,5 +1,5 @@
 # This file is part of Radicale Server - Calendar Server
-# Copyright © 2018 Unrud <unrud@outlook.com>
+# Copyright © 2018-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
@@ -33,9 +33,10 @@ class TestBaseWebRequests(BaseTest):
     def setup(self):
         self.configuration = config.load()
         self.colpath = tempfile.mkdtemp()
-        self.configuration["storage"]["filesystem_folder"] = self.colpath
-        # Disable syncing to disk for better performance
-        self.configuration["internal"]["filesystem_fsync"] = "False"
+        self.configuration.update({
+            "storage": {"filesystem_folder": self.colpath},
+            # Disable syncing to disk for better performance
+            "internal": {"filesystem_fsync": "False"}}, "test")
         self.application = Application(self.configuration)
 
     def teardown(self):
@@ -50,7 +51,7 @@ class TestBaseWebRequests(BaseTest):
         assert answer
 
     def test_none(self):
-        self.configuration["web"]["type"] = "none"
+        self.configuration.update({"web": {"type": "none"}}, "test")
         self.application = Application(self.configuration)
         status, _, answer = self.request("GET", "/.web")
         assert status == 200
@@ -60,7 +61,8 @@ class TestBaseWebRequests(BaseTest):
 
     def test_custom(self):
         """Custom web plugin."""
-        self.configuration["web"]["type"] = "tests.custom.web"
+        self.configuration.update({
+            "web": {"type": "tests.custom.web"}}, "test")
         self.application = Application(self.configuration)
         status, _, answer = self.request("GET", "/.web")
         assert status == 200