From bed5794b24b0aa5eec151b212279e20a92959a77 Mon Sep 17 00:00:00 2001 From: Tobias Brox Date: Fri, 15 Nov 2024 20:51:15 +0100 Subject: [PATCH] Tests: Move Xandikos/Radicale setup/teardown Test framework has been refactored a bit. Code for setting up and rigging down xandikos/radicale servers have been moved from `tests/test_caldav.py` to `tests/conf.py`. This allows for: * Adding code (including system calls or remote API calls) for Setting up and tearing down calendar servers in `conf_private.py` * Creating a local xandikos or radicale server in the `tests.client`-method, which is also used in the `examples`-section. * Allows offline testing of my upcoming `check_server_compatibility`-script --- CHANGELOG.md | 7 ++ tests/conf.py | 194 ++++++++++++++++++++++++++-------- tests/conf_private.py.EXAMPLE | 8 +- tests/test_caldav.py | 184 ++++---------------------------- 4 files changed, 184 insertions(+), 209 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e9a8a3b2..9ea76e74 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,13 @@ This project should more or less adhere to [Semantic Versioning](https://semver. ## [Unreleased] +### Changed + +* Test framework has been refactored a bit. Code for setting up and rigging down xandikos/radicale servers have been moved from `tests/test_caldav.py` to `tests/conf.py`. This allows for: + * Adding code (including system calls or remote API calls) for Setting up and tearing down calendar servers in `conf_private.py` + * Creating a local xandikos or radicale server in the `tests.client`-method, which is also used in the `examples`-section. + * Allows offline testing of my upcoming `check_server_compatibility`-script + ### Added * By now `calendar.search(..., sort_keys=("DTSTART")` will work. Sort keys expects a list or a tuple, but it's easy to send an attribute by mistake. diff --git a/tests/conf.py b/tests/conf.py index 01b9754e..b34f35e7 100644 --- a/tests/conf.py +++ b/tests/conf.py @@ -4,10 +4,14 @@ ## Make a conf_private.py for personal configuration. ## Check conf_private.py.EXAMPLE import logging +import tempfile +import threading +import time -from caldav.davclient import DAVClient +import requests -# from .compatibility_issues import bedework, xandikos +from . import compatibility_issues +from caldav.davclient import DAVClient #################################### # Import personal test server config @@ -79,63 +83,149 @@ ##################### # Public test servers ##################### -## As of 2019-09, all of those are down. Will try to fix Real Soon ... possibly before 2029 even. -if False: - # if test_public_test_servers: - ## TODO: this one is set up on emphemeral storage on OpenShift and - ## then configured manually through the webui installer, it will - ## most likely only work for some few days until it's down again. - ## It's needed to hard-code the configuration into - ## https://github.com/python-caldav/baikal +## Currently I'm not aware of any publically available test servers, and my +## own attempts on maintaining any has been canned. - caldav_servers.append( - { - "url": "http://baikal-caldav-servers.cloudapps.bitbit.net/html/cal.php/", - "username": "baikaluser", - "password": "asdf", - } - ) +# if test_public_test_servers: +# caldav_servers.append( ... ) - # bedework: - # * todos and journals are not properly supported - - # ref https://github.com/Bedework/bedework/issues/5 - # * propfind fails to return resourcetype, - # ref https://github.com/Bedework/bedework/issues/110 - # * date search on recurrences of recurring events doesn't work - # (not reported yet - TODO) - caldav_servers.append( - { - "url": "http://bedework-caldav-servers.cloudapps.bitbit.net/ucaldav/", - "username": "vbede", - "password": "bedework", - "incompatibilities": compatibility_issues.bedework, - } - ) +####################### +# Internal test servers +####################### + +if test_radicale: + import radicale.config + import radicale + import radicale.server + import socket + + def setup_radicale(self): + self.serverdir = tempfile.TemporaryDirectory() + self.serverdir.__enter__() + self.configuration = radicale.config.load("") + self.configuration.update( + {"storage": {"filesystem_folder": self.serverdir.name}} + ) + self.server = radicale.server + self.shutdown_socket, self.shutdown_socket_out = socket.socketpair() + self.radicale_thread = threading.Thread( + target=self.server.serve, + args=(self.configuration, self.shutdown_socket_out), + ) + self.radicale_thread.start() + i = 0 + while True: + try: + requests.get(self.url) + break + except: + time.sleep(0.05) + i += 1 + assert i < 100 + def teardown_radicale(self): + self.shutdown_socket.close() + i = 0 + self.serverdir.__exit__(None, None, None) + + url = "http://%s:%i/" % (radicale_host, radicale_port) caldav_servers.append( { - "url": "http://xandikos-caldav-servers.cloudapps.bitbit.net/", + "url": url, "username": "user1", - "password": "password1", - "incompatibilities": compatibility_issues.xandikos, + "password": "any-password-seems-to-work", + "backwards_compatibility_url": url + "user1", + "incompatibilities": compatibility_issues.radicale, + "setup": setup_radicale, + "teardown": teardown_radicale, } ) - # radicale +if test_xandikos: + import asyncio + + import aiohttp + import aiohttp.web + from xandikos.web import XandikosApp, XandikosBackend + + def setup_xandikos(self): + ## TODO: https://github.com/jelmer/xandikos/issues/131#issuecomment-1054805270 suggests a simpler way to launch the xandikos server + + self.serverdir = tempfile.TemporaryDirectory() + self.serverdir.__enter__() + ## Most of the stuff below is cargo-cult-copied from xandikos.web.main + ## Later jelmer created some API that could be used for this + ## Threshold put high due to https://github.com/jelmer/xandikos/issues/235 + ## index_threshold not supported in latest release yet + # self.backend = XandikosBackend(path=self.serverdir.name, index_threshold=0, paranoid=True) + # self.backend = XandikosBackend(path=self.serverdir.name, index_threshold=9999, paranoid=True) + self.backend = XandikosBackend(path=self.serverdir.name) + self.backend._mark_as_principal("/sometestuser/") + self.backend.create_principal("/sometestuser/", create_defaults=True) + mainapp = XandikosApp( + self.backend, current_user_principal="sometestuser", strict=True + ) + + async def xandikos_handler(request): + return await mainapp.aiohttp_handler(request, "/") + + self.xapp = aiohttp.web.Application() + self.xapp.router.add_route("*", "/{path_info:.*}", xandikos_handler) + ## https://stackoverflow.com/questions/51610074/how-to-run-an-aiohttp-server-in-a-thread + self.xapp_loop = asyncio.new_event_loop() + self.xapp_runner = aiohttp.web.AppRunner(self.xapp) + asyncio.set_event_loop(self.xapp_loop) + self.xapp_loop.run_until_complete(self.xapp_runner.setup()) + self.xapp_site = aiohttp.web.TCPSite( + self.xapp_runner, host=xandikos_host, port=xandikos_port + ) + self.xapp_loop.run_until_complete(self.xapp_site.start()) + + def aiohttp_server(): + self.xapp_loop.run_forever() + + self.xandikos_thread = threading.Thread(target=aiohttp_server) + self.xandikos_thread.start() + + def teardown_xandikos(self): + if not test_xandikos: + return + self.xapp_loop.stop() + + ## ... but the thread may be stuck waiting for a request ... + def silly_request(): + try: + requests.get(self.url) + except: + pass + + threading.Thread(target=silly_request).start() + i = 0 + while self.xapp_loop.is_running(): + time.sleep(0.05) + i += 1 + assert i < 100 + self.xapp_loop.run_until_complete(self.xapp_runner.cleanup()) + i = 0 + while self.xandikos_thread.is_alive(): + time.sleep(0.05) + i += 1 + assert i < 100 + + self.serverdir.__exit__(None, None, None) + + url = "http://%s:%i/" % (xandikos_host, xandikos_port) caldav_servers.append( { - "url": "http://radicale-caldav-servers.cloudapps.bitbit.net/", - "username": "testuser", - "password": "123", - "nofreebusy": True, - "nodefaultcalendar": True, - "noproxy": True, + "url": url, + "backwards_compatibility_url": url + "sometestuser", + "incompatibilities": compatibility_issues.xandikos, + "setup": setup_xandikos, + "teardown": teardown_xandikos, } ) -caldav_servers = [x for x in caldav_servers if x.get("enable", True)] - ################################################################### # Convenience - get a DAVClient object from the caldav_servers list ################################################################### @@ -144,9 +234,13 @@ ) -def client(idx=None, **kwargs): +def client(idx=None, setup=lambda conn: None, teardown=lambda conn: None, **kwargs): + ## No parameters given - find the first server in caldav_servers list if idx is None and not kwargs: - return client(0) + idx = 0 + while idx < len(caldav_servers) and not caldav_servers[idx].get("enable", True): + idx += 1 + return client(idx=idx) elif idx is not None and not kwargs and caldav_servers: return client(**caldav_servers[idx]) elif not kwargs: @@ -166,4 +260,10 @@ def client(idx=None, **kwargs): % kw ) kwargs.pop(kw) - return DAVClient(**kwargs) + conn = DAVClient(**kwargs) + setup(conn) + conn.teardown = teardown + return conn + + +caldav_servers = [x for x in caldav_servers if x.get("enable", True)] diff --git a/tests/conf_private.py.EXAMPLE b/tests/conf_private.py.EXAMPLE index 994882a6..39ecac92 100644 --- a/tests/conf_private.py.EXAMPLE +++ b/tests/conf_private.py.EXAMPLE @@ -10,7 +10,7 @@ from tests import compatibility_issues ## Define your primary caldav server here caldav_servers = [ { - ## Set enable to False if you don't want to use a server + ## Set enable to False if you don't want to use a server 'enable': True, ## This is all that is really needed - url, username and @@ -27,6 +27,11 @@ caldav_servers = [ ## tests/compatibility_issues.py for premade lists #'incompatibilities': compatibility_issues.nextcloud 'incompatibilities': [], + + ## You may even add setup and teardown methods to set up + ## and rig down the calendar server + #setup = lambda self: ... + #teardown = lambda self: ... } ] @@ -71,4 +76,5 @@ test_radicale = True ## For usage by ../examples/scheduling_examples.py. Should typically ## be three different users on the same caldav server. +## (beware of dragons - there is some half-done work in the caldav_test that is likely to break if this is set) #rfc6638_users = [ caldav_servers[0], caldav_servers[1], caldav_servers[2] ] diff --git a/tests/test_caldav.py b/tests/test_caldav.py index b3c6254b..2e788748 100644 --- a/tests/test_caldav.py +++ b/tests/test_caldav.py @@ -11,7 +11,6 @@ import logging import random import sys -import tempfile import threading import time import uuid @@ -20,10 +19,10 @@ from datetime import datetime from datetime import timedelta from datetime import timezone +from urllib.parse import urlparse import icalendar import pytest -import requests import vobject from requests.packages import urllib3 @@ -59,21 +58,6 @@ from caldav.objects import Principal from caldav.objects import Todo -if test_xandikos: - import asyncio - - import aiohttp - import aiohttp.web - from xandikos.web import XandikosApp, XandikosBackend - -if test_radicale: - import radicale.config - import radicale - import radicale.server - import socket - -from urllib.parse import urlparse - log = logging.getLogger("caldav") urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) @@ -505,6 +489,8 @@ def teardown_method(self): self.principals[i].calendar(name=calendar_name).delete() except error.NotFoundError: pass + for c in self.clients: + c.teardown() ## TODO # def testFreeBusy(self): @@ -649,7 +635,8 @@ def teardown_method(self): logging.debug("############## test teardown_method") logging.debug("############################") self._cleanup("post") - logging.debug("############## test teardown_method done") + logging.debug("############## test teardown_method almost done") + self.caldav.teardown(self.caldav) def _cleanup(self, mode=None): if self.cleanup_regime in ("pre", "post") and self.cleanup_regime != mode: @@ -860,10 +847,13 @@ def testProxy(self): pytest.skip("Unable to set up proxy server") threadobj = threading.Thread(target=proxy_httpd.serve_forever) + conn_params = self.server_params.copy() + for special in ("setup", "teardown"): + if special in conn_params: + conn_params.pop(special) try: threadobj.start() assert threadobj.is_alive() - conn_params = self.server_params.copy() conn_params["proxy"] = proxy c = client(**conn_params) p = c.principal() @@ -879,7 +869,6 @@ def testProxy(self): try: threadobj.start() assert threadobj.is_alive() - conn_params = self.server_params.copy() conn_params["proxy"] = proxy_noport c = client(**conn_params) p = c.principal() @@ -1698,12 +1687,15 @@ def testWrongPassword(self): pytest.skip( "Testing with wrong password skipped as calendar server does not require a password" ) - server_params = self.server_params.copy() - server_params["password"] = ( - codecs.encode(server_params["password"], "rot13") + "!" + connect_params = self.server_params.copy() + for delme in ("url", "setup", "teardown", "name"): + if delme in connect_params: + connect_params.pop(delme) + connect_params["password"] = ( + codecs.encode(connect_params["password"], "rot13") + "!" ) with pytest.raises(error.AuthorizationError): - client(**server_params).principal() + client(**connect_params).principal() def testCreateChildParent(self): self.skip_on_compatibility_flag("read_only") @@ -2821,7 +2813,9 @@ def testOffsetURL(self): """ urls = [self.principal.url, self._fixCalendar().url] connect_params = self.server_params.copy() - connect_params.pop("url") + for delme in ("url", "setup", "teardown", "name"): + if delme in connect_params: + connect_params.pop(delme) for url in urls: conn = client(**connect_params, url=url) principal = conn.principal() @@ -2842,6 +2836,10 @@ def testObjects(self): # (maybe a custom nose test loader really would be the better option?) # -- Tobias Brox , 2013-10-10 +## TODO: can we use @pytest.mark.parametrize to run the collection of tests +## above on each of the servers defined in caldav_servers? +## -- Tobias Brox , 2024-11-15 + _servernames = set() for _caldav_server in caldav_servers: # create a unique identifier out of the server domain name @@ -2862,139 +2860,3 @@ def testObjects(self): (RepeatedFunctionalTestsBaseClass,), {"server_params": _caldav_server}, ) - - -class TestLocalRadicale(RepeatedFunctionalTestsBaseClass): - """ - Sets up a local Radicale server and runs the functional tests towards it - """ - - def setup_method(self): - if not test_radicale: - pytest.skip("Skipping Radicale test due to configuration") - self.serverdir = tempfile.TemporaryDirectory() - self.serverdir.__enter__() - self.configuration = radicale.config.load("") - self.configuration.update( - {"storage": {"filesystem_folder": self.serverdir.name}} - ) - self.server = radicale.server - self.server_params = { - "url": "http://%s:%i/" % (radicale_host, radicale_port), - "username": "user1", - "password": "any-password-seems-to-work", - } - self.server_params["backwards_compatibility_url"] = ( - self.server_params["url"] + "user1" - ) - self.server_params["incompatibilities"] = compatibility_issues.radicale - self.shutdown_socket, self.shutdown_socket_out = socket.socketpair() - self.radicale_thread = threading.Thread( - target=self.server.serve, - args=(self.configuration, self.shutdown_socket_out), - ) - self.radicale_thread.start() - i = 0 - while True: - try: - requests.get(self.server_params["url"]) - break - except: - time.sleep(0.05) - i += 1 - assert i < 100 - try: - RepeatedFunctionalTestsBaseClass.setup_method(self) - except: - logging.critical("something bad happened in setup", exc_info=True) - self.teardown_method() - - def teardown_method(self): - if not test_radicale: - return - self.shutdown_socket.close() - i = 0 - self.serverdir.__exit__(None, None, None) - RepeatedFunctionalTestsBaseClass.teardown_method(self) - - -class TestLocalXandikos(RepeatedFunctionalTestsBaseClass): - """ - Sets up a local Xandikos server and runs the functional tests towards it - """ - - def setup_method(self): - if not test_xandikos: - pytest.skip("Skipping Xadikos test due to configuration") - - ## TODO: https://github.com/jelmer/xandikos/issues/131#issuecomment-1054805270 suggests a simpler way to launch the xandikos server - - self.serverdir = tempfile.TemporaryDirectory() - self.serverdir.__enter__() - ## Most of the stuff below is cargo-cult-copied from xandikos.web.main - ## Later jelmer created some API that could be used for this - ## Threshold put high due to https://github.com/jelmer/xandikos/issues/235 - ## index_threshold not supported in latest release yet - # self.backend = XandikosBackend(path=self.serverdir.name, index_threshold=0, paranoid=True) - # self.backend = XandikosBackend(path=self.serverdir.name, index_threshold=9999, paranoid=True) - self.backend = XandikosBackend(path=self.serverdir.name) - self.backend._mark_as_principal("/sometestuser/") - self.backend.create_principal("/sometestuser/", create_defaults=True) - mainapp = XandikosApp( - self.backend, current_user_principal="sometestuser", strict=True - ) - - async def xandikos_handler(request): - return await mainapp.aiohttp_handler(request, "/") - - self.xapp = aiohttp.web.Application() - self.xapp.router.add_route("*", "/{path_info:.*}", xandikos_handler) - ## https://stackoverflow.com/questions/51610074/how-to-run-an-aiohttp-server-in-a-thread - self.xapp_loop = asyncio.new_event_loop() - self.xapp_runner = aiohttp.web.AppRunner(self.xapp) - asyncio.set_event_loop(self.xapp_loop) - self.xapp_loop.run_until_complete(self.xapp_runner.setup()) - self.xapp_site = aiohttp.web.TCPSite( - self.xapp_runner, host=xandikos_host, port=xandikos_port - ) - self.xapp_loop.run_until_complete(self.xapp_site.start()) - - def aiohttp_server(): - self.xapp_loop.run_forever() - - self.xandikos_thread = threading.Thread(target=aiohttp_server) - self.xandikos_thread.start() - self.server_params = {"url": "http://%s:%i/" % (xandikos_host, xandikos_port)} - self.server_params["backwards_compatibility_url"] = ( - self.server_params["url"] + "sometestuser" - ) - self.server_params["incompatibilities"] = compatibility_issues.xandikos - RepeatedFunctionalTestsBaseClass.setup_method(self) - - def teardown_method(self): - if not test_xandikos: - return - self.xapp_loop.stop() - - ## ... but the thread may be stuck waiting for a request ... - def silly_request(): - try: - requests.get(self.server_params["url"]) - except: - pass - - threading.Thread(target=silly_request).start() - i = 0 - while self.xapp_loop.is_running(): - time.sleep(0.05) - i += 1 - assert i < 100 - self.xapp_loop.run_until_complete(self.xapp_runner.cleanup()) - i = 0 - while self.xandikos_thread.is_alive(): - time.sleep(0.05) - i += 1 - assert i < 100 - - self.serverdir.__exit__(None, None, None) - RepeatedFunctionalTestsBaseClass.teardown_method(self)