From d85562e5fff2a6b82162456d5fb80427da82c4a1 Mon Sep 17 00:00:00 2001 From: Tobias Brox Date: Sun, 15 Mar 2026 20:11:57 +0100 Subject: [PATCH 01/10] test: normalize comparison in testCreateJournalListAndJournalEntry Stalwart (and RFC-compliant servers in general) may return the same iCal object with different line folding depending on the fetch method (get_journal_by_uid vs get_journals). The old test relied on a side effect of property access on icalendar_instance to force re-serialization through the icalendar library (which normalizes folding), then compared .data. After the ruff B018 cleanup replaced that with the explicit get_icalendar_instance() call (which has no side effects on .data), the raw differently-folded bytes were compared and the assertion failed. Fix: compare to_ical() output directly, which is both side-effect-free and normalizes line folding. Co-Authored-By: Claude Sonnet 4.6 --- tests/test_caldav.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/tests/test_caldav.py b/tests/test_caldav.py index 44df8838..a8ee0856 100644 --- a/tests/test_caldav.py +++ b/tests/test_caldav.py @@ -2362,9 +2362,8 @@ def testCreateJournalListAndJournalEntry(self): assert len(journals) == 1 self.skip_unless_support("search.text.by-uid") j1_ = c.get_journal_by_uid(j1.id) - j1_.get_icalendar_instance() - journals[0].get_icalendar_instance() - assert j1_.data == journals[0].data + ## Direct comparison handles different line folding from different fetch methods + assert j1_.get_icalendar_instance() == journals[0].get_icalendar_instance() j2 = c.add_journal( dtstart=date(2011, 11, 11), summary="A childbirth in a hospital in Kupchino", From f1fa38128f031adcbe5e7954889f96d6040e6971 Mon Sep 17 00:00:00 2001 From: Tobias Brox Date: Mon, 16 Mar 2026 11:13:40 +0100 Subject: [PATCH 02/10] feat: PYTHON_CALDAV_USE_TEST_SERVER=1 auto-starts xandikos via registry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When PYTHON_CALDAV_USE_TEST_SERVER=1 is set (or testconfig=True), and no testing_allowed config-file section is found, get_davclient() now falls back to spinning up the first enabled server from the test-server registry. Registry changes: - Xandikos is now discovered before Radicale in _discover_embedded_servers(), giving it higher default priority - all_servers() and enabled_servers() now return servers sorted by priority (lower number = first); defaults: embedded=10, docker=20, external=30 - Per-server priority can be overridden via priority: in either caldav_test_servers.yaml or a calendar config file section - Three new env vars filter which server types are included: PYTHON_CALDAV_TEST_EMBEDDED (default: enabled) PYTHON_CALDAV_TEST_DOCKER (default: enabled) PYTHON_CALDAV_TEST_EXTERNAL (default: enabled) Set any to 0/false/no/off to exclude that category. caldav/config.py changes: - _get_test_server_config() falls back to _try_start_registry_server() when no testing_allowed config section is found - _try_start_registry_server() iterates enabled_servers() in priority order, starts the first one that succeeds, and returns conn params Tests added: - testAutoEmbeddedServer: verifies that testconfig=True auto-starts xandikos when no config file server is found (skipped if xandikos absent) - test_get_davclient_returns_none_without_env_or_config: verifies that get_davclient() returns None when no env var and no config file are present - Add caldav/testing.py (shipped with package): EmbeddedServer base class, XandikosServer, RadicaleServer — so pip-installed users can use PYTHON_CALDAV_USE_TEST_SERVER=1 without the source tree. - Rewrite caldav/config.py _get_test_server_config() to use the registry as the primary authority (not a fallback): priority-ordered servers win over config-file entries unless those entries have an explicit better priority. Adds _collect_test_servers(), _get_pip_test_servers(), _ConfiguredServer, _test_server_to_params(). Removes _try_start_registry_server() and _registry_server_to_params() which duplicated client_context() logic. - Rewrite tests/test_servers/helpers.py client_context(): no longer writes a temporary config file; just starts the server directly and sets PYTHON_CALDAV_USE_TEST_SERVER=1 for the duration. - Rewrite tests/test_servers/embedded.py: XandikosTestServer and RadicaleTestServer now delegate to caldav.testing.XandikosServer / RadicaleServer, eliminating duplicated startup/teardown code. - tests/test_caldav.py: add testAutoEmbeddedServer and test_get_davclient_returns_none_without_env_or_config; fix caldav_servers[-1] → caldav_servers[0] (embedded servers now sort first). Co-Authored-By: Claude Sonnet 4.6 --- caldav/config.py | 170 +++++++++++++--- caldav/testing.py | 347 +++++++++++++++++++++++++++++++++ tests/test_caldav.py | 26 ++- tests/test_servers/base.py | 16 ++ tests/test_servers/embedded.py | 295 ++++------------------------ tests/test_servers/helpers.py | 102 ++++------ tests/test_servers/registry.py | 92 +++++++-- 7 files changed, 680 insertions(+), 368 deletions(-) create mode 100644 caldav/testing.py diff --git a/caldav/config.py b/caldav/config.py index 9ee375a2..4df8f44f 100644 --- a/caldav/config.py +++ b/caldav/config.py @@ -337,45 +337,161 @@ def _get_test_server_config( name: str | None, environment: bool, config_file: str | None = None ) -> dict[str, Any] | None: """ - Get connection parameters for test server. + Get connection parameters for a test server. - Priority: - 1. Config file sections with 'testing_allowed: true' + Uses the priority-ordered server list as the single source of truth. + Embedded servers (xandikos priority 10, radicale priority 10) beat docker + (20) and configured / testing_allowed servers (30) by default. Any server + can override its priority via ``priority: `` in its config. + + When running from the source tree the full registry is used (embedded + + docker + external). When caldav is pip-installed without the test + infrastructure, only embedded servers from ``caldav.testing`` plus any + ``testing_allowed`` config-file sections are available. Args: - name: Specific config section or test server name/index to use. - Can be a config section name, test server name, or numeric index. - environment: Whether to check environment variables for server selection. - config_file: Explicit config file path to check. + name: Optional server name or server_type to restrict selection. + environment: Whether to read PYTHON_CALDAV_TEST_SERVER_NAME. + config_file: Explicit config file path (for testing_allowed fallback). Returns: - Connection parameters dict, or None if no test server configured. + Connection parameters dict, or None if no server could be started. """ - # Check environment for server name if environment and name is None: name = os.environ.get("PYTHON_CALDAV_TEST_SERVER_NAME") - # 1. Try config file with testing_allowed flag - cfg = read_config(config_file) # Use explicit file or default locations - if cfg: - # If name is specified, check if it's a config section with testing_allowed - if name is not None and not isinstance(name, int): - section_data = config_section(cfg, str(name)) - if section_data.get("testing_allowed"): - return _extract_conn_params_from_section(section_data) - - # Find first section with testing_allowed=true (if no name specified) - if name is None: - for section_name in cfg: - section_data = config_section(cfg, section_name) - if section_data.get("testing_allowed"): - logging.info(f"Using test server from config section: {section_name}") - return _extract_conn_params_from_section(section_data) - - # No built-in test server fallback - use config files or environment variables + for server in _collect_test_servers(name, config_file): + was_already_started = server._started + try: + server.start() + except Exception as e: + logging.warning("Failed to start test server %s: %s", server.name, e) + continue + logging.info("Using test server: %s at %s", server.name, server.url) + return _test_server_to_params(server, was_already_started) + + logging.info( + "PYTHON_CALDAV_USE_TEST_SERVER is set but no test server is available. " + "Install xandikos or radicale, or add 'testing_allowed: true' to a config section." + ) return None +def _collect_test_servers(name: str | None, config_file: str | None) -> list[Any]: + """ + Return a priority-ordered list of available test servers. + + Tries the full registry first (source tree); falls back to embedded + servers from ``caldav.testing`` plus ``testing_allowed`` config sections + (pip-installed users). + """ + try: + from tests.test_servers.registry import get_registry + + servers: list[Any] = list(get_registry().enabled_servers()) + except ImportError: + servers = _get_pip_test_servers(config_file) + + if name is not None: + servers = [s for s in servers if name.lower() in (s.name.lower(), s.server_type.lower())] + return servers + + +def _get_pip_test_servers(config_file: str | None) -> list[Any]: + """ + Build a server list for pip-installed users (no tests/ available). + + Includes embedded servers from caldav.testing and any testing_allowed + sections from config files, all sorted by priority. + """ + from caldav.testing import RadicaleServer, XandikosServer + + servers: list[Any] = [] + + try: + import xandikos # noqa: F401 + + servers.append(XandikosServer()) + except ImportError: + pass + + try: + import radicale # noqa: F401 + + servers.append(RadicaleServer()) + except ImportError: + pass + + for section_name, params in get_all_test_servers(config_file).items(): + servers.append(_ConfiguredServer(section_name, params)) + + return sorted(servers, key=lambda s: s.priority) + + +class _ConfiguredServer: + """ + Thin wrapper around a ``testing_allowed`` config-file section. + + Used only in pip-installed mode; in the source tree these sections are + loaded into the registry as ``ExternalTestServer`` instances instead. + """ + + server_type = "external" + + def __init__(self, name: str, params: dict[str, Any]) -> None: + self.name = name + self._params = params + self._started = False + + @property + def priority(self) -> int: + return int(self._params.get("priority", 30)) + + @property + def url(self) -> str: + return self._params.get("url", "") + + @property + def username(self) -> str | None: + return self._params.get("username") + + @property + def password(self) -> str | None: + return self._params.get("password") + + @property + def features(self) -> Any: + return self._params.get("features") + + def start(self) -> None: + self._started = True # external — assumed already running + + def stop(self) -> None: + self._started = False + + def is_accessible(self) -> bool: + return bool(self.url) + + +def _test_server_to_params(server: Any, was_already_started: bool) -> dict[str, Any]: + """Build a ``get_davclient``-compatible params dict from a server object.""" + params: dict[str, Any] = { + "url": server.url, + "_server_name": server.name, + } + if server.username: + params["username"] = server.username + if server.password: + params["password"] = server.password + if server.features: + params["features"] = server.features + # Only attach teardown if we started the server; if it was already running + # (e.g. started by client_context) leave lifecycle management to the caller. + if not was_already_started: + params["_teardown"] = lambda _client=None, s=server: s.stop() + return params + + def _extract_conn_params_from_section(section_data: dict[str, Any]) -> dict[str, Any] | None: """Extract connection parameters from a config section dict.""" conn_params: dict[str, Any] = {} diff --git a/caldav/testing.py b/caldav/testing.py new file mode 100644 index 00000000..bc7ef032 --- /dev/null +++ b/caldav/testing.py @@ -0,0 +1,347 @@ +""" +Lightweight embedded CalDAV test servers. + +Provides XandikosServer and RadicaleServer that are part of the installed +package so that pip-installed users can use get_davclient() with +PYTHON_CALDAV_USE_TEST_SERVER=1 without needing the full test infrastructure +from the source tree. + +Docker and external server support lives in tests/test_servers/ (source only). +""" + +import socket +import tempfile +import threading +import time +from typing import Any + +try: + import niquests as requests +except ImportError: + import requests # type: ignore[no-redef] + +# ── Constants ──────────────────────────────────────────────────────────────── + +MAX_STARTUP_WAIT_SECONDS = 60 +STARTUP_POLL_INTERVAL = 0.05 + + +# ── Base class ─────────────────────────────────────────────────────────────── + + +class EmbeddedServer: + """ + Base class for lightweight embedded CalDAV test servers. + + Subclasses must implement ``start()``, ``stop()``, and ``is_accessible()``. + + Priority + -------- + Both embedded servers default to priority **10**, which beats docker (20) + and external / configured servers (30). Override via ``priority: `` + in the config dict or by setting ``_default_priority`` on the subclass. + """ + + name: str = "EmbeddedServer" + server_type: str = "embedded" + _default_priority: int = 10 + + def __init__(self, config: dict[str, Any] | None = None) -> None: + self.config: dict[str, Any] = config or {} + self.host: str = self.config.get("host", "localhost") + self.port: int = self.config.get("port", self._default_port()) + self._started: bool = False + self._was_stopped: bool = False + + def _default_port(self) -> int: + return 5232 + + # ── Properties read by caldav/config.py ────────────────────────────────── + + @property + def priority(self) -> int: + return int(self.config.get("priority", self._default_priority)) + + @property + def username(self) -> str | None: + return ( + self.config.get("username") + or self.config.get("caldav_username") + or self.config.get("caldav_user") + ) + + @property + def password(self) -> str | None: + for key in ("password", "caldav_password", "caldav_pass"): + if key in self.config: + return self.config[key] + return None + + @property + def features(self) -> Any: + from caldav.config import resolve_features + + return resolve_features(self.config.get("features", [])) + + @property + def url(self) -> str: + return f"http://{self.host}:{self.port}/{self.username or ''}" + + # ── Lifecycle ───────────────────────────────────────────────────────────── + + def start(self) -> None: + raise NotImplementedError + + def stop(self) -> None: + raise NotImplementedError + + def is_accessible(self) -> bool: + raise NotImplementedError + + def _wait_for_startup(self) -> None: + attempts = int(MAX_STARTUP_WAIT_SECONDS / STARTUP_POLL_INTERVAL) + for _ in range(attempts): + if self.is_accessible(): + return + time.sleep(STARTUP_POLL_INTERVAL) + raise RuntimeError(f"{self.name} failed to start after {MAX_STARTUP_WAIT_SECONDS} seconds") + + +# ── XandikosServer ──────────────────────────────────────────────────────────── + + +class XandikosServer(EmbeddedServer): + """Xandikos CalDAV server running in a background aiohttp thread.""" + + name = "Xandikos" + + def __init__(self, config: dict[str, Any] | None = None) -> None: + config = config or {} + config.setdefault("host", "localhost") + config.setdefault("port", 8993) + config.setdefault("username", "sometestuser") + if "features" not in config: + from caldav import compatibility_hints + + features = compatibility_hints.xandikos.copy() + features["auto-connect.url"]["domain"] = f"{config['host']}:{config['port']}" + config["features"] = features + super().__init__(config) + + self.serverdir: tempfile.TemporaryDirectory | None = None + self.xapp_loop: Any | None = None + self.xapp_runner: Any | None = None + self.xapp: Any | None = None + self.thread: threading.Thread | None = None + + def _default_port(self) -> int: + return 8993 + + @property + def url(self) -> str: + return f"http://{self.host}:{self.port}/{self.username}" + + def is_accessible(self) -> bool: + try: + response = requests.request( + "PROPFIND", + f"http://{self.host}:{self.port}", + timeout=2, + ) + return response.status_code in (200, 207, 401, 403, 404) + except Exception: + return False + + def start(self) -> None: + if self._started: + return + if not self._was_stopped and self.is_accessible(): + self._started = True + return + + try: + from xandikos.web import XandikosApp + + try: + from xandikos.web import SingleUserFilesystemBackend as XandikosBackend + except ImportError: + from xandikos.web import XandikosBackend # type: ignore[no-redef] + except ImportError as e: + raise RuntimeError("Xandikos is not installed") from e + + import asyncio + + from aiohttp import web + + self.serverdir = tempfile.TemporaryDirectory() + self.serverdir.__enter__() + + backend = XandikosBackend(self.serverdir.name) + backend._mark_as_principal(f"/{self.username}/") + backend.create_principal(f"/{self.username}/", create_defaults=True) + + mainapp = XandikosApp(backend, current_user_principal=self.username, strict=True) + + async def xandikos_handler(request: web.Request) -> web.Response: + return await mainapp.aiohttp_handler(request, "/") + + self.xapp = web.Application() + self.xapp.router.add_route("*", "/{path_info:.*}", xandikos_handler) + + def run_in_thread() -> None: + self.xapp_loop = asyncio.new_event_loop() + asyncio.set_event_loop(self.xapp_loop) + + async def start_app() -> None: + self.xapp_runner = web.AppRunner(self.xapp) + await self.xapp_runner.setup() + site = web.TCPSite(self.xapp_runner, self.host, self.port) + await site.start() + + self.xapp_loop.run_until_complete(start_app()) + self.xapp_loop.run_forever() + + self.thread = threading.Thread(target=run_in_thread, daemon=True) + self.thread.start() + self._wait_for_startup() + self._started = True + + def stop(self) -> None: + import asyncio + + if self.xapp_loop and self.xapp_runner: + + async def cleanup_and_stop() -> None: + await self.xapp_runner.cleanup() + self.xapp_loop.stop() + + try: + asyncio.run_coroutine_threadsafe(cleanup_and_stop(), self.xapp_loop).result( + timeout=10 + ) + except Exception: + if self.xapp_loop: + self.xapp_loop.call_soon_threadsafe(self.xapp_loop.stop) + elif self.xapp_loop: + self.xapp_loop.call_soon_threadsafe(self.xapp_loop.stop) + + if self.thread: + self.thread.join(timeout=5) + self.thread = None + + if self.serverdir: + self.serverdir.__exit__(None, None, None) + self.serverdir = None + + self.xapp_loop = None + self.xapp_runner = None + self.xapp = None + self._started = False + self._was_stopped = True + + +# ── RadicaleServer ──────────────────────────────────────────────────────────── + + +class RadicaleServer(EmbeddedServer): + """Radicale CalDAV server running in a background thread.""" + + name = "Radicale" + + def __init__(self, config: dict[str, Any] | None = None) -> None: + config = config or {} + config.setdefault("host", "localhost") + config.setdefault("port", 5232) + config.setdefault("username", "user1") + config.setdefault("password", "") + if "features" not in config: + from caldav import compatibility_hints + + features = compatibility_hints.radicale.copy() + features["auto-connect.url"]["domain"] = f"{config['host']}:{config['port']}" + config["features"] = features + super().__init__(config) + + self.serverdir: tempfile.TemporaryDirectory | None = None + self.shutdown_socket: socket.socket | None = None + self.shutdown_socket_out: socket.socket | None = None + self.thread: threading.Thread | None = None + + def _default_port(self) -> int: + return 5232 + + @property + def url(self) -> str: + return f"http://{self.host}:{self.port}/{self.username}" + + def is_accessible(self) -> bool: + try: + response = requests.get( + f"http://{self.host}:{self.port}/{self.username}", + timeout=2, + ) + return response.status_code in (200, 401, 403, 404) + except Exception: + return False + + def start(self) -> None: + if self._started: + return + if not self._was_stopped and self.is_accessible(): + self._started = True + return + + try: + import radicale + import radicale.config + import radicale.server + except ImportError as e: + raise RuntimeError("Radicale is not installed") from e + + self.serverdir = tempfile.TemporaryDirectory() + self.serverdir.__enter__() + + configuration = radicale.config.load("") + configuration.update( + { + "storage": {"filesystem_folder": self.serverdir.name}, + "auth": {"type": "none"}, + } + ) + + self.shutdown_socket, self.shutdown_socket_out = socket.socketpair() + self.thread = threading.Thread( + target=radicale.server.serve, + args=(configuration, self.shutdown_socket_out), + daemon=True, + ) + self.thread.start() + self._wait_for_startup() + + # Create the user principal collection (Radicale needs it before MKCALENDAR) + user_url = f"http://{self.host}:{self.port}/{self.username}/" + try: + r = requests.request("MKCOL", user_url, timeout=5) + if r.status_code not in (200, 201, 204, 405): + requests.request("MKCOL", user_url.rstrip("/"), timeout=5) + except Exception: + pass + + self._started = True + + def stop(self) -> None: + if self.shutdown_socket: + self.shutdown_socket.close() + self.shutdown_socket = None + self.shutdown_socket_out = None + + if self.thread: + self.thread.join(timeout=5) + self.thread = None + + if self.serverdir: + self.serverdir.__exit__(None, None, None) + self.serverdir = None + + self._started = False + self._was_stopped = True diff --git a/tests/test_caldav.py b/tests/test_caldav.py index a8ee0856..9f3e3e8d 100644 --- a/tests/test_caldav.py +++ b/tests/test_caldav.py @@ -464,7 +464,7 @@ def _make_client( @pytest.mark.skipif( not caldav_servers, - reason="Requirement: at least one working server in conf.py. The tail object of the server list will be chosen, that is typically the LocalRadicale or LocalXandikos server.", + reason="Requirement: at least one working server configured. The highest-priority server (index 0) will be used, typically Xandikos or Radicale.", ) class TestGetDAVClient: """ @@ -475,7 +475,7 @@ class TestGetDAVClient: def testTestConfig(self): """Test that get_davclient(testconfig=True) finds config with testing_allowed.""" # Start a test server using test_servers framework - server_params = caldav_servers[-1] + server_params = caldav_servers[0] with client(**server_params) as conn: # Create a config file with testing_allowed: true config = {"testing_allowed": True} @@ -499,7 +499,7 @@ def testTestConfig(self): def testEnvironment(self): """Test that get_davclient() reads from environment variables.""" # Start a test server using test_servers framework - server_params = caldav_servers[-1] + server_params = caldav_servers[0] with client(**server_params) as conn: # Set environment variables (only if value is not None) for key in ("username", "password", "proxy"): @@ -520,7 +520,7 @@ def testEnvironment(self): def testConfigfile(self): """Test that get_davclient() reads from config file.""" # Start a test server using test_servers framework - server_params = caldav_servers[-1] + server_params = caldav_servers[0] with client(**server_params) as conn: config = {} for key in ("username", "password", "proxy"): @@ -537,6 +537,14 @@ def testConfigfile(self): ) as conn2: assert conn2.principal() + def testAutoEmbeddedServer(self) -> None: + """Test that get_davclient(testconfig=True) auto-starts xandikos when no config file server found.""" + pytest.importorskip("xandikos") + with get_davclient(testconfig=True, check_config_file=False, environment=False) as conn: + assert hasattr(conn, "server_name") + assert "xandikos" in conn.server_name.lower() + conn.principal() + def testNoConfigfile(self, fs): """This is actually a unit test, not a functional test. Should move it to another file probably, and make more unit @@ -561,6 +569,16 @@ def testNoConfigfile(self, fs): assert client.url == "https://caldav.example.com/dav" +def test_get_davclient_returns_none_without_env_or_config(fs) -> None: + """get_davclient() must return None when PYTHON_CALDAV_USE_TEST_SERVER is not set + and no config file is present.""" + from unittest.mock import patch + + with patch.dict(os.environ, {}, clear=True): + result = get_davclient() + assert result is None + + @pytest.mark.skipif( not rfc6638_users, reason="need rfc6638_users to be set in order to run this test" ) diff --git a/tests/test_servers/base.py b/tests/test_servers/base.py index f1805055..5c1edbde 100644 --- a/tests/test_servers/base.py +++ b/tests/test_servers/base.py @@ -78,6 +78,19 @@ def password(self) -> str | None: return self.config[key] return None + # Default priority per server type. Lower number = higher priority. + # Subclasses (embedded=10, docker=20, external=30) override this. + _default_priority: int = 50 + + @property + def priority(self) -> int: + """Return the sort priority for this server. + + Lower values are returned first by the registry. Can be overridden + per-server via ``priority: `` in the server config. + """ + return int(self.config.get("priority", self._default_priority)) + @property def features(self) -> Any: """ @@ -218,6 +231,7 @@ class EmbeddedTestServer(TestServer): """ server_type = "embedded" + _default_priority = 10 def __init__(self, config: dict[str, Any] | None = None) -> None: super().__init__(config) @@ -266,6 +280,7 @@ class DockerTestServer(TestServer): """ server_type = "docker" + _default_priority = 20 def __init__(self, config: dict[str, Any] | None = None) -> None: super().__init__(config) @@ -406,6 +421,7 @@ class ExternalTestServer(TestServer): """ server_type = "external" + _default_priority = 30 def __init__(self, config: dict[str, Any] | None = None) -> None: super().__init__(config) diff --git a/tests/test_servers/embedded.py b/tests/test_servers/embedded.py index 1adf9d3a..fac684d3 100644 --- a/tests/test_servers/embedded.py +++ b/tests/test_servers/embedded.py @@ -1,310 +1,85 @@ """ -Embedded test server implementations. +Embedded test server implementations for the test infrastructure. -This module provides test server implementations for servers that run -in-process: Radicale and Xandikos. +The actual server logic lives in caldav/testing.py (part of the installed +package). This module wraps those classes so they fit into the +EmbeddedTestServer hierarchy and are registered with the server registry. """ -import socket -import tempfile -import threading from typing import Any -try: - import niquests as requests -except ImportError: - import requests # type: ignore - -from caldav import compatibility_hints +from caldav.testing import RadicaleServer as _RadicaleCore +from caldav.testing import XandikosServer as _XandikosCore from .base import EmbeddedTestServer from .registry import register_server_class -class RadicaleTestServer(EmbeddedTestServer): - """ - Radicale CalDAV server running in a thread. - - Radicale is a lightweight CalDAV server that's easy to embed - for testing purposes. - """ +class XandikosTestServer(EmbeddedTestServer): + """Xandikos server wrapped for the test infrastructure.""" - name = "LocalRadicale" + name = "Xandikos" def __init__(self, config: dict[str, Any] | None = None) -> None: config = config or {} config.setdefault("host", "localhost") - config.setdefault("port", 5232) - config.setdefault("username", "user1") - config.setdefault("password", "") - # Set up Radicale-specific compatibility hints - if "features" not in config: - features = compatibility_hints.radicale.copy() - host = config.get("host", "localhost") - port = config.get("port", 5232) - features["auto-connect.url"]["domain"] = f"{host}:{port}" - config["features"] = features + config.setdefault("port", 8993) + config.setdefault("username", "sometestuser") super().__init__(config) - - # Server state - self.serverdir: tempfile.TemporaryDirectory | None = None - self.shutdown_socket: socket.socket | None = None - self.shutdown_socket_out: socket.socket | None = None - self.thread: threading.Thread | None = None + self._core = _XandikosCore(config) def _default_port(self) -> int: - return 5232 + return 8993 @property def url(self) -> str: - return f"http://{self.host}:{self.port}/{self.username}" + return self._core.url def is_accessible(self) -> bool: - try: - # Check the user URL to ensure the server is ready - # and to auto-create the user collection (Radicale does this on first access) - response = requests.get( - f"http://{self.host}:{self.port}/{self.username}", - timeout=2, - ) - return response.status_code in (200, 401, 403, 404) - except Exception: - return False + return self._core.is_accessible() def start(self) -> None: - """Start the Radicale server in a background thread.""" - # Only check is_accessible() if we haven't been started before. - # After stop() is called, the port might still respond briefly, - # so we can't trust is_accessible() in that case. - if self._started: - return - if not hasattr(self, "_was_stopped") and self.is_accessible(): - return - - try: - import radicale - import radicale.config - import radicale.server - except ImportError as e: - raise RuntimeError("Radicale is not installed") from e - - # Create temporary storage directory - self.serverdir = tempfile.TemporaryDirectory() - self.serverdir.__enter__() - - # Configure Radicale - configuration = radicale.config.load("") - configuration.update( - { - "storage": {"filesystem_folder": self.serverdir.name}, - "auth": {"type": "none"}, - } - ) - - # Create shutdown socket pair - self.shutdown_socket, self.shutdown_socket_out = socket.socketpair() - - # Start server thread - self.thread = threading.Thread( - target=radicale.server.serve, - args=(configuration, self.shutdown_socket_out), - ) - self.thread.start() - - # Wait for server to be ready - self._wait_for_startup() - - # Create the user collection with MKCOL - # Radicale requires the parent collection to exist before MKCALENDAR - user_url = f"http://{self.host}:{self.port}/{self.username}/" - try: - response = requests.request( - "MKCOL", - user_url, - timeout=5, - ) - # 201 = created, 405 = already exists (or method not allowed) - if response.status_code not in (200, 201, 204, 405): - # Some servers need a trailing slash, try without - response = requests.request( - "MKCOL", - user_url.rstrip("/"), - timeout=5, - ) - except Exception: - pass # Ignore errors, the collection might already exist - - self._started = True + self._core.start() + self._started = self._core._started def stop(self) -> None: - """Stop the Radicale server and cleanup.""" - if self.shutdown_socket: - self.shutdown_socket.close() - self.shutdown_socket = None - self.shutdown_socket_out = None - - if self.thread: - self.thread.join(timeout=5) - self.thread = None + self._core.stop() + self._started = self._core._started + self._was_stopped = self._core._was_stopped - if self.serverdir: - self.serverdir.__exit__(None, None, None) - self.serverdir = None - self._started = False - self._was_stopped = True # Mark that we've been stopped at least once - - -class XandikosTestServer(EmbeddedTestServer): - """ - Xandikos CalDAV server running with aiohttp. - - Xandikos is an async CalDAV server that uses aiohttp. - We run it in a separate thread with its own event loop. - """ +class RadicaleTestServer(EmbeddedTestServer): + """Radicale server wrapped for the test infrastructure.""" - name = "LocalXandikos" + name = "Radicale" def __init__(self, config: dict[str, Any] | None = None) -> None: config = config or {} config.setdefault("host", "localhost") - config.setdefault("port", 8993) - config.setdefault("username", "sometestuser") - # Set up Xandikos-specific compatibility hints - if "features" not in config: - features = compatibility_hints.xandikos.copy() - host = config.get("host", "localhost") - port = config.get("port", 8993) - features["auto-connect.url"]["domain"] = f"{host}:{port}" - config["features"] = features + config.setdefault("port", 5232) + config.setdefault("username", "user1") + config.setdefault("password", "") super().__init__(config) - - # Server state - self.serverdir: tempfile.TemporaryDirectory | None = None - self.xapp_loop: Any | None = None - self.xapp_runner: Any | None = None - self.xapp: Any | None = None - self.thread: threading.Thread | None = None + self._core = _RadicaleCore(config) def _default_port(self) -> int: - return 8993 + return 5232 @property def url(self) -> str: - return f"http://{self.host}:{self.port}/{self.username}" + return self._core.url def is_accessible(self) -> bool: - try: - response = requests.request( - "PROPFIND", - f"http://{self.host}:{self.port}", - timeout=2, - ) - return response.status_code in (200, 207, 401, 403, 404) - except Exception: - return False + return self._core.is_accessible() def start(self) -> None: - """Start the Xandikos server.""" - # Only check is_accessible() if we haven't been started before. - # After stop() is called, the port might still respond briefly, - # so we can't trust is_accessible() in that case. - if self._started: - return - if not hasattr(self, "_was_stopped") and self.is_accessible(): - return - - try: - from xandikos.web import XandikosApp - - try: - # xandikos master renamed XandikosBackend to SingleUserFilesystemBackend - from xandikos.web import SingleUserFilesystemBackend as XandikosBackend - except ImportError: - from xandikos.web import XandikosBackend # type: ignore[no-redef] - except ImportError as e: - raise RuntimeError("Xandikos is not installed") from e - - import asyncio - - from aiohttp import web - - # Create temporary storage directory - self.serverdir = tempfile.TemporaryDirectory() - self.serverdir.__enter__() - - # Create backend and configure principal (following conf.py pattern) - backend = XandikosBackend(self.serverdir.name) - backend._mark_as_principal(f"/{self.username}/") - backend.create_principal(f"/{self.username}/", create_defaults=True) - - # Create the Xandikos app with the backend - mainapp = XandikosApp(backend, current_user_principal=self.username, strict=True) - - # Create aiohttp handler - async def xandikos_handler(request: web.Request) -> web.Response: - return await mainapp.aiohttp_handler(request, "/") - - self.xapp = web.Application() - self.xapp.router.add_route("*", "/{path_info:.*}", xandikos_handler) - - def run_in_thread() -> None: - self.xapp_loop = asyncio.new_event_loop() - asyncio.set_event_loop(self.xapp_loop) - - async def start_app() -> None: - self.xapp_runner = web.AppRunner(self.xapp) - await self.xapp_runner.setup() - site = web.TCPSite(self.xapp_runner, self.host, self.port) - await site.start() - - self.xapp_loop.run_until_complete(start_app()) - self.xapp_loop.run_forever() - - # Start server in a background thread - self.thread = threading.Thread(target=run_in_thread) - self.thread.start() - - # Wait for server to be ready - self._wait_for_startup() - self._started = True + self._core.start() + self._started = self._core._started def stop(self) -> None: - """Stop the Xandikos server and cleanup.""" - import asyncio - - if self.xapp_loop and self.xapp_runner: - # Clean shutdown: first cleanup the aiohttp runner (stops accepting - # connections and waits for in-flight requests), then stop the loop. - # This must be done from within the event loop thread. - async def cleanup_and_stop() -> None: - await self.xapp_runner.cleanup() - self.xapp_loop.stop() - - try: - asyncio.run_coroutine_threadsafe(cleanup_and_stop(), self.xapp_loop).result( - timeout=10 - ) - except Exception: - # Fallback: force stop if cleanup fails - if self.xapp_loop: - self.xapp_loop.call_soon_threadsafe(self.xapp_loop.stop) - elif self.xapp_loop: - self.xapp_loop.call_soon_threadsafe(self.xapp_loop.stop) - - if self.thread: - self.thread.join(timeout=5) - self.thread = None - - if self.serverdir: - self.serverdir.__exit__(None, None, None) - self.serverdir = None - - self.xapp_loop = None - self.xapp_runner = None - self.xapp = None - self._started = False - self._was_stopped = True # Mark that we've been stopped at least once + self._core.stop() + self._started = self._core._started + self._was_stopped = self._core._was_stopped # Register server classes diff --git a/tests/test_servers/helpers.py b/tests/test_servers/helpers.py index ec2c9fc4..6d783424 100644 --- a/tests/test_servers/helpers.py +++ b/tests/test_servers/helpers.py @@ -5,9 +5,7 @@ with get_davclient() support. """ -import json import os -import tempfile from contextlib import contextmanager from caldav import DAVClient @@ -18,94 +16,76 @@ @contextmanager def client_context(server_index: int = 0, server_name: str | None = None): """ - Context manager that provides a running test server and configured environment. + Context manager that provides a running test server. - This is the recommended way to get a test client when you need: - - A running server - - Environment configured so get_davclient() works - - Automatic cleanup + Starts the highest-priority available server (or the one selected by + ``server_index`` / ``server_name``), yields a connected DAVClient, and + stops the server on exit. + + Sets ``PYTHON_CALDAV_USE_TEST_SERVER=1`` for the duration so that + ``get_davclient()`` calls within the block also find the running server + via the registry (without needing a temporary config file). + + Usage:: - Usage: from tests.test_servers import client_context with client_context() as client: principal = client.principal() - # get_davclient() will also work within this context + # get_davclient() also works within this context Args: - server_index: Index into the caldav_servers list (default: 0, first server) - server_name: Optional server name to use instead of index + server_index: Index into the priority-ordered enabled-servers list + (default: 0 = highest-priority server). + server_name: Optional server name to use instead of index. Yields: - DAVClient: Connected client to the test server + DAVClient: Connected client to the test server. Raises: - RuntimeError: If no test servers are configured + RuntimeError: If no test servers are configured or the requested + server is not found. """ registry = get_registry() - servers = registry.get_caldav_servers_list() + servers = registry.enabled_servers() if not servers: raise RuntimeError("No test servers configured") - # Find the server to use if server_name: - server_params = None - for s in servers: - if s.get("name") == server_name: - server_params = s - break - if not server_params: + server = next((s for s in servers if s.name == server_name), None) + if server is None: raise RuntimeError(f"Server '{server_name}' not found") else: - server_params = servers[server_index] - - # Import here to avoid circular imports - from caldav.davclient import CONNKEYS - - # Create client and start server via setup callback - kwargs = {k: v for k, v in server_params.items() if k in CONNKEYS} - conn = DAVClient(**kwargs) - conn.setup = server_params.get("setup", lambda _: None) - conn.teardown = server_params.get("teardown", lambda _: None) + server = servers[server_index] - # Create temporary config file for get_davclient() - config = {"testing_allowed": True} - for key in ("username", "password", "proxy"): - if key in server_params: - config[f"caldav_{key}"] = server_params[key] - config["caldav_url"] = server_params["url"] + server.start() - config_file = tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) - json.dump({"default": config}, config_file) - config_file.close() - - # Set environment variables - old_config_file = os.environ.get("CALDAV_CONFIG_FILE") - old_test_server = os.environ.get("PYTHON_CALDAV_USE_TEST_SERVER") + from caldav.davclient import CONNKEYS - os.environ["CALDAV_CONFIG_FILE"] = config_file.name + conn_kwargs = { + k: v + for k, v in { + "url": server.url, + "username": server.username, + "password": server.password, + "features": server.features, + }.items() + if v is not None and k in CONNKEYS + } + conn = DAVClient(**conn_kwargs) + conn.server_name = server.name + + old_env = os.environ.get("PYTHON_CALDAV_USE_TEST_SERVER") os.environ["PYTHON_CALDAV_USE_TEST_SERVER"] = "1" try: - # Enter client context (starts server) - conn.__enter__() yield conn finally: - # Exit client context (stops server) - conn.__exit__(None, None, None) - - # Clean up config file - os.unlink(config_file.name) - - # Restore environment - if old_config_file is not None: - os.environ["CALDAV_CONFIG_FILE"] = old_config_file - else: - os.environ.pop("CALDAV_CONFIG_FILE", None) + server.stop() - if old_test_server is not None: - os.environ["PYTHON_CALDAV_USE_TEST_SERVER"] = old_test_server + if old_env is not None: + os.environ["PYTHON_CALDAV_USE_TEST_SERVER"] = old_env else: os.environ.pop("PYTHON_CALDAV_USE_TEST_SERVER", None) @@ -113,4 +93,4 @@ def client_context(server_index: int = 0, server_name: str | None = None): def has_test_servers() -> bool: """Check if any test servers are configured.""" registry = get_registry() - return len(registry.get_caldav_servers_list()) > 0 + return len(registry.enabled_servers()) > 0 diff --git a/tests/test_servers/registry.py b/tests/test_servers/registry.py index cd5a4064..ed794a23 100644 --- a/tests/test_servers/registry.py +++ b/tests/test_servers/registry.py @@ -3,8 +3,36 @@ This module provides a registry for discovering and managing test servers. It supports automatic detection of available servers and lazy initialization. + +Server type env-var filtering +------------------------------ +Three environment variables control which server *types* are included when +the registry reports ``enabled_servers()`` / ``get_caldav_servers_list()``: + +* ``PYTHON_CALDAV_TEST_EMBEDDED`` – embedded (in-process) servers such as + Xandikos and Radicale. Default: **enabled**. +* ``PYTHON_CALDAV_TEST_DOCKER`` – Docker-based servers (Baikal, Nextcloud, + Cyrus, …). Default: **enabled** (skipped automatically when Docker is + not available). +* ``PYTHON_CALDAV_TEST_EXTERNAL`` – externally configured servers loaded + from ``caldav_test_servers.yaml``. Default: **enabled**. + +Set any of these to ``0``, ``false``, ``no``, or ``off`` (case-insensitive) +to disable that category. Any other value (including ``1``, ``true``, etc.) +keeps them enabled. + +Priority order +-------------- +Servers are registered — and therefore returned — in this order: + +1. Xandikos (embedded) +2. Radicale (embedded) +3. Docker servers (in alphabetical directory order) +4. External / config-file servers """ +import os + from .base import TestServer # Server class registry - maps type names to server classes @@ -90,21 +118,50 @@ def get(self, name: str) -> TestServer | None: def all_servers(self) -> list[TestServer]: """ - Get all registered test servers. + Get all registered test servers, sorted by priority (lowest first). Returns: List of all registered servers """ - return list(self._servers.values()) + return sorted(self._servers.values(), key=lambda s: s.priority) + + @staticmethod + def _is_server_type_enabled(server_type: str) -> bool: + """Return True unless the env var for this server type is set to a falsy value. + + Env vars checked: + - ``PYTHON_CALDAV_TEST_EMBEDDED`` for ``"embedded"`` + - ``PYTHON_CALDAV_TEST_DOCKER`` for ``"docker"`` + - ``PYTHON_CALDAV_TEST_EXTERNAL`` for ``"external"`` + """ + env_map = { + "embedded": "PYTHON_CALDAV_TEST_EMBEDDED", + "docker": "PYTHON_CALDAV_TEST_DOCKER", + "external": "PYTHON_CALDAV_TEST_EXTERNAL", + } + env_var = env_map.get(server_type) + if env_var is None: + return True + val = os.environ.get(env_var, "").strip().lower() + return val not in ("0", "false", "no", "off") def enabled_servers(self) -> list[TestServer]: """ - Get all enabled test servers. + Get all enabled test servers, sorted by priority (lowest first) and + respecting per-type env-var overrides. Returns: - List of servers where config.get("enabled", True) is True + List of servers where config.get("enabled", True) is True *and* + the server's type has not been disabled via an environment variable. """ - return [s for s in self._servers.values() if s.config.get("enabled", True)] + return sorted( + ( + s + for s in self._servers.values() + if s.config.get("enabled", True) and self._is_server_type_enabled(s.server_type) + ), + key=lambda s: s.priority, + ) def load_from_config(self, config: dict) -> None: """ @@ -186,18 +243,11 @@ def auto_discover(self) -> None: self._discover_docker_servers() def _discover_embedded_servers(self) -> None: - """Discover available embedded servers.""" - # Check for Radicale - try: - import radicale # noqa: F401 - - radicale_class = get_server_class("radicale") - if radicale_class is not None: - self.register(radicale_class()) - except ImportError: - pass + """Discover available embedded servers. - # Check for Xandikos + Xandikos is registered first (higher default priority than Radicale). + """ + # Check for Xandikos first (preferred embedded server) try: import xandikos # noqa: F401 @@ -207,6 +257,16 @@ def _discover_embedded_servers(self) -> None: except ImportError: pass + # Check for Radicale (fallback embedded server) + try: + import radicale # noqa: F401 + + radicale_class = get_server_class("radicale") + if radicale_class is not None: + self.register(radicale_class()) + except ImportError: + pass + def _discover_docker_servers(self) -> None: """Discover available Docker servers.""" from pathlib import Path From 3a9b542ac42f5314b5ae2fdefe67bd8184d5f119 Mon Sep 17 00:00:00 2001 From: Tobias Brox Date: Mon, 16 Mar 2026 15:37:47 +0100 Subject: [PATCH 03/10] feat: get_calendars() supports multi-server config sections; fix calendar_name/url filtering Two bugs fixed, one new feature: Bug 1: _extract_conn_params_from_section silently dropped calendar_name and calendar_url keys from config sections, so get_calendars(config_section="foo") where foo had calendar_name: "My Cal" would return all calendars instead of filtering. Fixed by extracting those keys alongside the caldav_* ones. Bug 2: expand_config_section was never called when reading the config file, so meta-sections ({"contains": ["work", "personal"]}) had no effect. New feature: get_calendars() now uses expand_config_section on the config file path, expanding a single config_section value ("*", "all", "work_*", ...) to multiple leaf sections. Each expanded section gets its own DAVClient; all calendars are aggregated into one CalendarCollection. CalendarCollection now holds a list of clients and closes all of them on exit. The single-server path (explicit url=, env vars, testconfig=True) is unchanged. Per-section calendar_name / calendar_url are used as filters; function-level arguments override them. New helper _fetch_calendars_for_client() eliminates the duplicate calendar- fetching logic that was inline in get_calendars(). New config.py export: get_all_file_connection_params(config_file, section). Tests added in test_caldav_unit.py (TestExpandConfigSection, TestGetAllFileConnectionParams) and test_caldav.py (TestGetCalendarsConfig). Co-Authored-By: Claude Sonnet 4.6 --- caldav/base_client.py | 243 ++++++++++++++++++++++++++------------ caldav/config.py | 47 +++++++- caldav/davclient.py | 2 +- tests/test_caldav.py | 91 ++++++++++++++ tests/test_caldav_unit.py | 139 ++++++++++++++++++++++ 5 files changed, 442 insertions(+), 80 deletions(-) diff --git a/caldav/base_client.py b/caldav/base_client.py index 5f41c17f..23ddddd7 100644 --- a/caldav/base_client.py +++ b/caldav/base_client.py @@ -291,15 +291,25 @@ class CalendarCollection(list): calendars[0].client.close() """ - def __init__(self, calendars: list | None = None, client: Any = None): + def __init__( + self, + calendars: list | None = None, + client: Any = None, + clients: list | None = None, + ): super().__init__(calendars or []) - self._client = client + if clients is not None: + self._clients: list = list(clients) + elif client is not None: + self._clients = [client] + else: + self._clients = [] @property def client(self): """The underlying DAV client, if available.""" - if self._client: - return self._client + if self._clients: + return self._clients[0] # Fall back to getting client from first calendar if self: return self[0].client @@ -313,10 +323,13 @@ def __exit__(self, exc_type, exc_val, exc_tb): return False def close(self): - """Close the underlying DAV client connection.""" - if self._client: - self._client.close() - elif self: + """Close all underlying DAV client connections.""" + seen: set[int] = set() + for c in self._clients: + if id(c) not in seen: + c.close() + seen.add(id(c)) + if not self._clients and self: self[0].client.close() @@ -394,6 +407,67 @@ def _normalize_to_list(obj: Any) -> list: return list(obj) +def _fetch_calendars_for_client( + client: Any, + calendar_url: Any | None, + calendar_name: Any | None, + raise_errors: bool, +) -> list: + """ + Fetch calendars from a single connected client, optionally filtered. + + Returns a (possibly empty) list of Calendar objects. On error the + behaviour is controlled by ``raise_errors``. + """ + import logging + + log = logging.getLogger("caldav") + + def _try(meth, kwargs, errmsg): + try: + ret = meth(**kwargs) + if ret is None: + raise ValueError(f"Method returned None: {errmsg}") + return ret + except Exception as e: + log.error(f"Problems fetching calendar information: {errmsg} - {e}") + if raise_errors: + raise + return None + + principal = _try(client.principal, {}, "getting principal") + if not principal: + return [] + + calendars = [] + calendar_urls = _normalize_to_list(calendar_url) + calendar_names = _normalize_to_list(calendar_name) + + for cal_url in calendar_urls: + if "/" in str(cal_url): + calendar = principal.calendar(cal_url=cal_url) + else: + calendar = principal.calendar(cal_id=cal_url) + if _try(calendar.get_display_name, {}, f"calendar {cal_url}"): + calendars.append(calendar) + + for cal_name in calendar_names: + calendar = _try( + principal.calendar, + {"name": cal_name}, + f"calendar by name '{cal_name}'", + ) + if calendar: + calendars.append(calendar) + + if not calendars and not calendar_urls and not calendar_names: + all_cals = _try(principal.get_calendars, {}, "getting all calendars") + if all_cals: + calendars = all_cals + + return calendars + + def get_calendars( client_class: type, calendar_url: Any | None = None, @@ -408,68 +482,74 @@ def get_calendars( **config_data, ) -> CalendarCollection: """ - Get calendars from a CalDAV server with configuration from multiple sources. + Get calendars from one or more CalDAV servers. - This function creates a client, connects to the server, and returns - calendar objects based on the specified criteria. Configuration is read - from various sources (explicit parameters, environment variables, config files). + Configuration is read from multiple sources in priority order: - The returned CalendarCollection can be used as a context manager to ensure - the underlying connection is properly closed. + 1. Explicit keyword arguments (``url``, ``username``, ``password``, …) + 2. Test server (``testconfig=True`` or ``PYTHON_CALDAV_USE_TEST_SERVER``) + 3. Environment variables (``CALDAV_URL``, …) + 4. Config file — supports meta-sections so a single ``config_section`` + can expand to multiple servers (see below) + + **Multi-server / meta-sections** + + Sources 1–3 always produce a single connection. When the config file is + used (source 4) the ``config_section`` value is passed through + ``expand_config_section``, which supports: + + * ``"*"`` – every non-disabled section in the file + * ``"all"`` – a meta-section defined as ``{"contains": ["work", "personal"]}`` + * Glob patterns such as ``"work_*"`` + * A plain section name (normal single-server behaviour) + + Each expanded leaf section can carry its own ``calendar_name`` or + ``calendar_url`` to filter which calendars are returned for that server. + Function-level ``calendar_name`` / ``calendar_url`` arguments override + per-section values when provided. + + The returned :class:`CalendarCollection` is a list that can be used as a + context manager; on exit **all** underlying connections are closed. Args: - client_class: The client class to use (DAVClient or AsyncDAVClient). + client_class: The client class to use (``DAVClient`` or ``AsyncDAVClient``). calendar_url: URL(s) or ID(s) of specific calendars to fetch. - Can be a string or list of strings. If the value contains '/', - it's treated as a URL; otherwise as a calendar ID. calendar_name: Name(s) of specific calendars to fetch by display name. - Can be a string or list of strings. check_config_file: Whether to look for config files (default: True). config_file: Explicit path to config file. - config_section: Section name in config file (default: "default"). + config_section: Section name in config file (default: ``"default"``). + Supports ``*``, meta-sections, and glob patterns. testconfig: Whether to use test server configuration. environment: Whether to read from environment variables (default: True). name: Name of test server to use (for testconfig). raise_errors: If True, raise exceptions on errors; if False, log and skip. - **config_data: Connection parameters (url, username, password, etc.) + **config_data: Explicit connection parameters (url, username, password, …). Returns: - CalendarCollection of Calendar objects matching the criteria. - If no calendar_url or calendar_name specified, returns all calendars. + :class:`CalendarCollection` of matching calendars (may be empty). - Example:: + Example — single server:: from caldav import get_calendars - # As context manager (recommended) - with get_calendars(url="https://...", username="...", password="...") as calendars: - for cal in calendars: + with get_calendars(url="https://...", username="...", password="...") as cals: + for cal in cals: print(cal.get_display_name()) - # Without context manager - connection closed on garbage collection - calendars = get_calendars(url="https://...", username="...", password="...") - """ - import logging - - log = logging.getLogger("caldav") + Example — all sections in config file:: - def _try(meth, kwargs, errmsg): - """Try a method call, handling errors based on raise_errors flag.""" - try: - ret = meth(**kwargs) - if ret is None: - raise ValueError(f"Method returned None: {errmsg}") - return ret - except Exception as e: - log.error(f"Problems fetching calendar information: {errmsg} - {e}") - if raise_errors: - raise - return None + with get_calendars(config_section="*") as cals: + for cal in cals: + print(cal.get_display_name()) + """ + from caldav import config as _config - # Get client using existing config infrastructure + # ── Priority 1-3: explicit params / test mode / env vars ────────────── + # Try without config file first; if a client is resolved we stay + # single-server (existing behaviour unchanged). client = get_davclient( client_class=client_class, - check_config_file=check_config_file, + check_config_file=False, config_file=config_file, config_section=config_section, testconfig=testconfig, @@ -478,48 +558,55 @@ def _try(meth, kwargs, errmsg): **config_data, ) - if client is None: + if client is not None: + calendars = _fetch_calendars_for_client(client, calendar_url, calendar_name, raise_errors) + return CalendarCollection(calendars, client=client) + + # ── Priority 4: config file (may expand to multiple sections) ───────── + if not check_config_file: if raise_errors: raise ValueError("Could not create DAV client - no configuration found") return CalendarCollection() - # Get principal - principal = _try(client.principal, {}, "getting principal") - if not principal: - return CalendarCollection(client=client) + # Resolve config_file path from env if not given (mirrors get_connection_params) + resolved_config_file = config_file + if environment and not resolved_config_file: + import os - calendars = [] - calendar_urls = _normalize_to_list(calendar_url) - calendar_names = _normalize_to_list(calendar_name) + resolved_config_file = os.environ.get("CALDAV_CONFIG_FILE") - # Fetch specific calendars by URL/ID - for cal_url in calendar_urls: - if "/" in str(cal_url): - calendar = principal.calendar(cal_url=cal_url) - else: - calendar = principal.calendar(cal_id=cal_url) + resolved_section = config_section + if environment and not resolved_section: + import os - # Verify the calendar exists by trying to get its display name - if _try(calendar.get_display_name, {}, f"calendar {cal_url}"): - calendars.append(calendar) + resolved_section = os.environ.get("CALDAV_CONFIG_SECTION") - # Fetch specific calendars by name - for cal_name in calendar_names: - calendar = _try( - principal.calendar, - {"name": cal_name}, - f"calendar by name '{cal_name}'", - ) - if calendar: - calendars.append(calendar) + all_params = _config.get_all_file_connection_params(resolved_config_file, resolved_section) - # If no specific calendars requested, get all calendars - if not calendars and not calendar_urls and not calendar_names: - all_cals = _try(principal.get_calendars, {}, "getting all calendars") - if all_cals: - calendars = all_cals + if not all_params: + if raise_errors: + raise ValueError("Could not create DAV client - no configuration found") + return CalendarCollection() + + from caldav.config import CONNKEYS + + all_calendars: list = [] + all_clients: list = [] + + for params in all_params: + # Per-section calendar filters — function-level args override them + sec_cal_url = params.pop("calendar_url", None) + sec_cal_name = params.pop("calendar_name", None) + eff_cal_url = calendar_url if calendar_url is not None else sec_cal_url + eff_cal_name = calendar_name if calendar_name is not None else sec_cal_name + + conn_params = {k: v for k, v in params.items() if k in CONNKEYS} + c = client_class(**conn_params) + section_cals = _fetch_calendars_for_client(c, eff_cal_url, eff_cal_name, raise_errors) + all_calendars.extend(section_cals) + all_clients.append(c) - return CalendarCollection(calendars, client=client) + return CalendarCollection(all_calendars, clients=all_clients) def get_davclient( diff --git a/caldav/config.py b/caldav/config.py index 4df8f44f..d0008506 100644 --- a/caldav/config.py +++ b/caldav/config.py @@ -493,7 +493,14 @@ def _test_server_to_params(server: Any, was_already_started: bool) -> dict[str, def _extract_conn_params_from_section(section_data: dict[str, Any]) -> dict[str, Any] | None: - """Extract connection parameters from a config section dict.""" + """Extract connection parameters from a config section dict. + + Returns a dict containing only CONNKEYS entries. Returns ``None`` if no + server URL is present. Calendar filter keys (``calendar_name``, + ``calendar_url``) are intentionally excluded — callers that need them + (e.g. :func:`get_all_file_connection_params`) read ``section_data`` + directly. + """ conn_params: dict[str, Any] = {} for k in section_data: if k.startswith("caldav_"): @@ -514,6 +521,44 @@ def _extract_conn_params_from_section(section_data: dict[str, Any]) -> dict[str, return conn_params if conn_params.get("url") else None +def get_all_file_connection_params( + config_file: str | None, + section_name: str | None = None, +) -> list[dict[str, Any]]: + """Return connection-params dicts for every leaf section that ``section_name`` expands to. + + ``section_name`` follows the same rules as everywhere else in the config + system: a plain name returns ``[section_name]``; a meta-section with + ``contains: [...]`` expands recursively; ``*`` means all non-disabled + sections; glob patterns (``work_*``) are also supported. + + Each dict contains CONNKEYS entries plus optional ``calendar_name`` / + ``calendar_url`` calendar-filter keys read from the config section. + + Returns an empty list when the config file is absent or the section has + no usable URL. + """ + if not section_name: + section_name = "default" + + cfg = read_config(config_file) + if not cfg: + return [] + + sections = expand_config_section(cfg, section_name) + result: list[dict[str, Any]] = [] + for s in sections: + section_data = config_section(cfg, s) + params = _extract_conn_params_from_section(section_data) + if params: + # Add calendar filter keys — these must NOT flow into DAVClient() + for k in ("calendar_name", "calendar_url"): + if section_data.get(k): + params[k] = section_data[k] + result.append(params) + return result + + def get_all_test_servers( config_file: str | None = None, ) -> dict[str, dict[str, Any]]: diff --git a/caldav/davclient.py b/caldav/davclient.py index 3cdb7610..25b6dc99 100644 --- a/caldav/davclient.py +++ b/caldav/davclient.py @@ -945,7 +945,7 @@ def _sync_request( def get_calendars(**kwargs) -> list["Calendar"]: """ - Get calendars from a CalDAV server with configuration from multiple sources. + Get calendars from CalDAV servers with configuration from multiple sources. This is a convenience wrapper around :func:`caldav.base_client.get_calendars` that uses DAVClient. diff --git a/tests/test_caldav.py b/tests/test_caldav.py index 9f3e3e8d..07359431 100644 --- a/tests/test_caldav.py +++ b/tests/test_caldav.py @@ -579,6 +579,97 @@ def test_get_davclient_returns_none_without_env_or_config(fs) -> None: assert result is None +@pytest.mark.skipif(not caldav_servers, reason="need at least one test server") +class TestGetCalendarsConfig: + """ + Tests for caldav.get_calendars() config-file integration: + - calendar_name / calendar_url in a config section filter the returned calendars + - meta-sections (contains: [...]) aggregate calendars from multiple servers + """ + + def _server_config(self, conn, server) -> dict: + """Build a caldav config dict for a running server connection.""" + cfg: dict = {"caldav_url": str(conn.url)} + if server and server.username: + cfg["caldav_username"] = server.username + if server and server.password is not None: + cfg["caldav_password"] = server.password + return cfg + + def test_calendar_name_from_config_filters(self) -> None: + """calendar_name in a config section must restrict which calendar is returned.""" + from caldav import get_calendars + + from .test_servers.helpers import client_context + + with client_context(server_index=0) as conn: + principal = conn.principal() + principal.make_calendar(name="CalConfigA") + principal.make_calendar(name="CalConfigB") + + server = _registry.get(conn.server_name) + cfg_section = self._server_config(conn, server) + cfg_section["calendar_name"] = "CalConfigA" + + with tempfile.NamedTemporaryFile(delete=True, mode="w", suffix=".json") as tmp: + json.dump({"default": cfg_section}, tmp) + tmp.flush() + os.fsync(tmp.fileno()) + with get_calendars( + config_file=tmp.name, + config_section="default", + testconfig=False, + environment=False, + ) as calendars: + names = [c.get_display_name() for c in calendars] + assert "CalConfigA" in names + assert "CalConfigB" not in names + + @pytest.mark.skipif( + not (test_radicale and test_xandikos), + reason="need both Xandikos and Radicale for multi-server test", + ) + def test_multi_server_meta_section(self) -> None: + """A meta-section (contains: [...]) must aggregate calendars from multiple servers.""" + from caldav import get_calendars + + from .test_servers.helpers import client_context + + with ( + client_context(server_name="Xandikos") as x_conn, + client_context(server_name="Radicale") as r_conn, + ): + x_conn.principal().make_calendar(name="XandikosMetaCal") + r_conn.principal().make_calendar(name="RadicaleMetaCal") + + x_cfg = self._server_config(x_conn, _xandikos_server) + x_cfg["calendar_name"] = "XandikosMetaCal" + r_cfg = self._server_config(r_conn, _radicale_server) + r_cfg["calendar_name"] = "RadicaleMetaCal" + + config = { + "xandikos": x_cfg, + "radicale": r_cfg, + "both": {"contains": ["xandikos", "radicale"]}, + } + with tempfile.NamedTemporaryFile(delete=True, mode="w", suffix=".json") as tmp: + json.dump(config, tmp) + tmp.flush() + os.fsync(tmp.fileno()) + with get_calendars( + config_file=tmp.name, + config_section="both", + testconfig=False, + environment=False, + ) as calendars: + names = {c.get_display_name() for c in calendars} + assert "XandikosMetaCal" in names + assert "RadicaleMetaCal" in names + # calendars came from two distinct DAVClient instances + clients = {id(c.client) for c in calendars} + assert len(clients) == 2 + + @pytest.mark.skipif( not rfc6638_users, reason="need rfc6638_users to be set in order to run this test" ) diff --git a/tests/test_caldav_unit.py b/tests/test_caldav_unit.py index 76ecfbae..7b7d604c 100755 --- a/tests/test_caldav_unit.py +++ b/tests/test_caldav_unit.py @@ -2084,3 +2084,142 @@ def test_rate_limit_max_sleep_stops_adaptive_retries(self, mocked): with mock.patch("caldav.davclient.time.sleep"): with pytest.raises(error.RateLimitError): client.request("/") + + +class TestExpandConfigSection: + """Unit tests for caldav.config.expand_config_section.""" + + def test_normal_section_returns_single_item(self): + from caldav.config import expand_config_section + + config = {"default": {"caldav_url": "https://example.com/"}} + assert expand_config_section(config, "default") == ["default"] + + def test_meta_section_with_contains(self): + from caldav.config import expand_config_section + + config = { + "work": {"caldav_url": "https://work.example.com/"}, + "personal": {"caldav_url": "https://personal.example.com/"}, + "all": {"contains": ["work", "personal"]}, + } + assert expand_config_section(config, "all") == ["work", "personal"] + + def test_star_returns_all_non_disabled_sections(self): + from caldav.config import expand_config_section + + config = { + "work": {"caldav_url": "https://work.example.com/"}, + "personal": {"caldav_url": "https://personal.example.com/"}, + "hidden": {"caldav_url": "https://hidden.example.com/", "disable": True}, + } + result = expand_config_section(config, "*") + assert set(result) == {"work", "personal"} + + def test_glob_pattern(self): + from caldav.config import expand_config_section + + config = { + "work_a": {"caldav_url": "https://a.example.com/"}, + "work_b": {"caldav_url": "https://b.example.com/"}, + "personal": {"caldav_url": "https://c.example.com/"}, + } + result = expand_config_section(config, "work_*") + assert set(result) == {"work_a", "work_b"} + + def test_recursive_meta_section(self): + from caldav.config import expand_config_section + + config = { + "a": {"caldav_url": "https://a.example.com/"}, + "b": {"caldav_url": "https://b.example.com/"}, + "ab": {"contains": ["a", "b"]}, + "c": {"caldav_url": "https://c.example.com/"}, + "all": {"contains": ["ab", "c"]}, + } + assert set(expand_config_section(config, "all")) == {"a", "b", "c"} + + +class TestGetAllFileConnectionParams: + """Unit tests for caldav.config.get_all_file_connection_params.""" + + def test_single_section_returns_one_dict(self, tmp_path): + import json + + from caldav.config import get_all_file_connection_params + + config = { + "default": { + "caldav_url": "https://example.com/dav/", + "caldav_username": "user", + "caldav_password": "pass", + } + } + config_file = tmp_path / "calendar.conf" + config_file.write_text(json.dumps(config)) + results = get_all_file_connection_params(str(config_file), "default") + assert len(results) == 1 + assert results[0]["url"] == "https://example.com/dav/" + assert results[0]["username"] == "user" + + def test_calendar_name_extracted_from_section(self, tmp_path): + import json + + from caldav.config import get_all_file_connection_params + + config = { + "default": { + "caldav_url": "https://example.com/dav/", + "caldav_username": "user", + "calendar_name": "My Calendar", + } + } + config_file = tmp_path / "calendar.conf" + config_file.write_text(json.dumps(config)) + results = get_all_file_connection_params(str(config_file), "default") + assert len(results) == 1 + assert results[0]["calendar_name"] == "My Calendar" + + def test_calendar_url_extracted_from_section(self, tmp_path): + import json + + from caldav.config import get_all_file_connection_params + + config = { + "default": { + "caldav_url": "https://example.com/dav/", + "caldav_username": "user", + "calendar_url": "/dav/user/mycalendar/", + } + } + config_file = tmp_path / "calendar.conf" + config_file.write_text(json.dumps(config)) + results = get_all_file_connection_params(str(config_file), "default") + assert len(results) == 1 + assert results[0]["calendar_url"] == "/dav/user/mycalendar/" + + def test_meta_section_returns_multiple_dicts(self, tmp_path): + import json + + from caldav.config import get_all_file_connection_params + + config = { + "work": { + "caldav_url": "https://work.example.com/dav/", + "caldav_username": "wuser", + }, + "personal": { + "caldav_url": "https://personal.example.com/dav/", + "caldav_username": "puser", + }, + "all": {"contains": ["work", "personal"]}, + } + config_file = tmp_path / "calendar.conf" + config_file.write_text(json.dumps(config)) + results = get_all_file_connection_params(str(config_file), "all") + assert len(results) == 2 + urls = {r["url"] for r in results} + assert urls == { + "https://work.example.com/dav/", + "https://personal.example.com/dav/", + } From 920a89e698ccf308eb243d3519e4b26197a36697 Mon Sep 17 00:00:00 2001 From: Tobias Brox Date: Tue, 17 Mar 2026 16:41:44 +0100 Subject: [PATCH 04/10] feat: add get_icalendar_component() and edit_icalendar_component() The component was forgotten earlier when creating the get_* end edit_* patterns. get_icalendar_component() returns a deepcopy of the inner VEVENT/VTODO/VJOURNAL subcomponent for read-only access, consistent with the get_icalendar_instance() naming convention. edit_icalendar_component() is a context manager that yields the inner component directly for editing, delegating to edit_icalendar_instance() so all borrow/state/save machinery is reused. Co-Authored-By: Claude Sonnet 4.6 --- caldav/calendarobjectresource.py | 43 ++++++++++++++++++++++++++++++++ 1 file changed, 43 insertions(+) diff --git a/caldav/calendarobjectresource.py b/caldav/calendarobjectresource.py index 5b3ec7af..08edfd3c 100644 --- a/caldav/calendarobjectresource.py +++ b/caldav/calendarobjectresource.py @@ -1347,6 +1347,23 @@ def get_icalendar_instance(self) -> icalendar.Calendar: """ return self._ensure_state().get_icalendar_copy() + def get_icalendar_component(self) -> "icalendar.Component": + """Get a COPY of the inner icalendar component (VEVENT/VTODO/VJOURNAL) for read-only access. + + This is safe for inspection - modifications to the returned object + will NOT be saved. For editing, use edit_icalendar_component(). + + For recurring events with multiple components, returns the first + non-timezone component (the master RRULE component). Use + ``search(..., expand=True)`` to get individual expanded occurrences. + + Returns: + A copy of the first non-timezone subcomponent. + """ + import copy + + return copy.deepcopy(self.icalendar_component) + def get_vobject_instance(self) -> "vobject.base.Component": """Get a COPY of the vobject object for read-only access. @@ -1401,6 +1418,32 @@ def edit_icalendar_instance(self): finally: self._borrowed = False + @contextmanager + def edit_icalendar_component(self): + """Context manager to borrow the inner icalendar component for editing. + + Like :meth:`edit_icalendar_instance` but yields the first + ``VEVENT`` / ``VTODO`` / ``VJOURNAL`` subcomponent directly, + rather than the ``VCALENDAR`` wrapper. This is convenient when + you only need to modify a single property on the component itself. + + Usage:: + + with event.edit_icalendar_component() as comp: + comp['SUMMARY'] = 'New Summary' + event.save() + + Yields: + The first non-``VTIMEZONE`` subcomponent of the icalendar object. + + Raises: + RuntimeError: If another representation is currently borrowed. + StopIteration: If the calendar contains no non-timezone components. + """ + with self.edit_icalendar_instance() as cal: + component = next(c for c in cal.subcomponents if c.name != "VTIMEZONE") + yield component + @contextmanager def edit_vobject_instance(self): """Context manager to borrow the vobject object for editing. From ef3e0345e2a1f5a1ecd6b258c350ffe9442c063b Mon Sep 17 00:00:00 2001 From: Tobias Brox Date: Tue, 17 Mar 2026 19:38:28 +0100 Subject: [PATCH 05/10] =?UTF-8?q?fix:=20coerce=20date=E2=86=92datetime=20i?= =?UTF-8?q?n=20searcher=20to=20silence=20icalendar=5Fsearcher=20warning?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit RFC 4791 §9.9 requires UTC datetime values in time-range queries. When plain date objects were passed to calendar.search() or calendar.searcher(), they flowed into icalendar_searcher.Searcher unchanged, which emitted a logging.warning("Date-range searches not well supported yet; use datetime rather than dates") for every object checked during client-side filtering. Fix: extract _populate_searcher() helper (eliminating the duplicated loop between searcher() and search()), and coerce any date to datetime(Y, M, D, tzinfo=utc) before setting start/end/alarm_start/alarm_end on the searcher. Co-Authored-By: Claude Sonnet 4.6 --- caldav/collection.py | 85 +++++++++++++++++-------------------- tests/test_caldav_unit.py | 88 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 126 insertions(+), 47 deletions(-) diff --git a/caldav/collection.py b/caldav/collection.py index a21217e2..484806d3 100644 --- a/caldav/collection.py +++ b/caldav/collection.py @@ -13,7 +13,8 @@ import logging import uuid import warnings -from datetime import datetime +from datetime import date as _date +from datetime import datetime, timezone from time import sleep from typing import TYPE_CHECKING, Any, Optional, TypeVar from urllib.parse import ParseResult, SplitResult, quote, unquote @@ -1173,30 +1174,11 @@ async def _async_request_report_build_resultlist( ) return (response, matches) - def searcher(self, **searchargs) -> "CalDAVSearcher": - """Create a searcher object for building complex search queries. - - This is the recommended way to perform advanced searches. The - returned searcher can have filters added, and then be executed: + def _populate_searcher(self, my_searcher, searchargs: dict, sort_reverse: bool) -> None: + """Populate a CalDAVSearcher from a dict of search keyword arguments. - .. code-block:: python - - searcher = calendar.searcher(event=True, start=..., end=...) - searcher.add_property_filter("SUMMARY", "meeting") - results = searcher.search() - - For simple searches, use :meth:`search` directly instead. - - :param searchargs: Search parameters (same as for :meth:`search`) - :return: A CalDAVSearcher bound to this calendar - - See :class:`caldav.search.CalDAVSearcher` for available filter methods. + Shared by :meth:`searcher` and :meth:`search`. """ - from .search import CalDAVSearcher - - my_searcher = CalDAVSearcher() - my_searcher._calendar = self - for key in searchargs: assert key[0] != "_" ## not allowed alias = key @@ -1207,7 +1189,6 @@ def searcher(self, **searchargs) -> "CalDAVSearcher": if key == "no_class_": alias = "no_class" if key == "sort_keys": - sort_reverse = searchargs.get("sort_reverse", False) if isinstance(searchargs["sort_keys"], str): searchargs["sort_keys"] = [searchargs["sort_keys"]] for sortkey in searchargs["sort_keys"]: @@ -1215,12 +1196,43 @@ def searcher(self, **searchargs) -> "CalDAVSearcher": elif key == "sort_reverse": pass # handled with sort_keys elif key == "comp_class" or key in my_searcher.__dataclass_fields__: - setattr(my_searcher, key, searchargs[key]) + value = searchargs[key] + if ( + key in ("start", "end", "alarm_start", "alarm_end") + and isinstance(value, _date) + and not isinstance(value, datetime) + ): + value = datetime(value.year, value.month, value.day, tzinfo=timezone.utc) + setattr(my_searcher, key, value) elif alias.startswith("no_"): my_searcher.add_property_filter(alias[3:], searchargs[key], operator="undef") else: my_searcher.add_property_filter(alias, searchargs[key]) + def searcher(self, **searchargs) -> "CalDAVSearcher": + """Create a searcher object for building complex search queries. + + This is the recommended way to perform advanced searches. The + returned searcher can have filters added, and then be executed: + + .. code-block:: python + + searcher = calendar.searcher(event=True, start=..., end=...) + searcher.add_property_filter("SUMMARY", "meeting") + results = searcher.search() + + For simple searches, use :meth:`search` directly instead. + + :param searchargs: Search parameters (same as for :meth:`search`) + :return: A CalDAVSearcher bound to this calendar + + See :class:`caldav.search.CalDAVSearcher` for available filter methods. + """ + from .search import CalDAVSearcher + + my_searcher = CalDAVSearcher() + my_searcher._calendar = self + self._populate_searcher(my_searcher, searchargs, searchargs.get("sort_reverse", False)) return my_searcher def search( @@ -1337,28 +1349,7 @@ def search( ## Transfer all the arguments to CalDAVSearcher my_searcher = CalDAVSearcher() - for key in searchargs: - assert key[0] != "_" ## not allowed - alias = key - if key == "class_": ## because class is a reserved word - alias = "class" - if key == "no_category": - alias = "no_categories" - if key == "no_class_": - alias = "no_class" - if key == "sort_keys": - if isinstance(searchargs["sort_keys"], str): - searchargs["sort_keys"] = [searchargs["sort_keys"]] - for sortkey in searchargs["sort_keys"]: - my_searcher.add_sort_key(sortkey, sort_reverse) - continue - elif key == "comp_class" or key in my_searcher.__dataclass_fields__: - setattr(my_searcher, key, searchargs[key]) - continue - elif alias.startswith("no_"): - my_searcher.add_property_filter(alias[3:], searchargs[key], operator="undef") - else: - my_searcher.add_property_filter(alias, searchargs[key]) + self._populate_searcher(my_searcher, searchargs, sort_reverse) if not xml and filters: xml = filters diff --git a/tests/test_caldav_unit.py b/tests/test_caldav_unit.py index 7b7d604c..2060930b 100755 --- a/tests/test_caldav_unit.py +++ b/tests/test_caldav_unit.py @@ -2086,6 +2086,94 @@ def test_rate_limit_max_sleep_stops_adaptive_retries(self, mocked): client.request("/") +class TestDateToUtcConversion: + """ + RFC 4791 §9.9: time-range start/end MUST be UTC datetime values. + Plain date objects must be coerced to midnight UTC before use in + REPORT XML and before being stored on the searcher (for client-side + filtering). + """ + + def test_to_utc_date_string_with_date_gives_midnight_utc(self): + """_to_utc_date_string(date) must produce a valid UTC datetime string.""" + from datetime import date + + from caldav.elements.cdav import _to_utc_date_string + + result = _to_utc_date_string(date(2026, 5, 1)) + assert result == "20260501T000000Z" + + def test_to_utc_date_string_with_datetime_unchanged(self): + """_to_utc_date_string(datetime) must still work correctly.""" + from datetime import datetime, timezone + + from caldav.elements.cdav import _to_utc_date_string + + result = _to_utc_date_string(datetime(2026, 5, 1, 12, 0, 0, tzinfo=timezone.utc)) + assert result == "20260501T120000Z" + + def test_time_range_xml_with_date_produces_utc_attrs(self): + """TimeRange built with plain date objects must emit UTC datetime attributes.""" + from datetime import date + + from caldav.elements.cdav import TimeRange + + tr = TimeRange(start=date(2026, 5, 1), end=date(2026, 6, 1)) + assert tr.attributes["start"] == "20260501T000000Z" + assert tr.attributes["end"] == "20260601T000000Z" + + def test_search_with_date_emits_utc_datetime_in_report_xml(self): + """calendar.search(start=date(...)) must send UTC datetime strings in + the REPORT body, not bare date strings.""" + from datetime import date + + class CapturingClient(MockedDAVClient): + def __init__(self): + super().__init__("") + self.report_bodies = [] + + def request(self, url, method="GET", body=None, headers=None): + if method == "REPORT" and body: + self.report_bodies.append(body) + return super().request(url, method, body, headers) + + client = CapturingClient() + calendar = Calendar(client, url="/cal/") + calendar.search(event=True, start=date(2026, 5, 1), end=date(2026, 6, 1)) + + assert client.report_bodies, "expected at least one REPORT request" + body = client.report_bodies[0] + if isinstance(body, bytes): + body = body.decode() + assert "20260501T000000Z" in body, f"UTC start not found in REPORT body:\n{body}" + assert "20260601T000000Z" in body, f"UTC end not found in REPORT body:\n{body}" + # Must NOT contain bare date strings without time component + assert "20260501Z" not in body + assert "20260601Z" not in body + + def test_searcher_coerces_date_to_datetime(self): + """calendar.searcher(start=date(...)) must store datetime on the searcher, + not bare date objects — otherwise icalendar_searcher warns during + client-side filtering (check_component is called per result object).""" + from datetime import date, datetime, timezone + + client = MockedDAVClient("") + calendar = Calendar(client, url="/cal/") + searcher = calendar.searcher( + event=True, + start=date(2026, 5, 1), + end=date(2026, 6, 1), + ) + assert isinstance(searcher.start, datetime), ( + f"searcher.start should be datetime, got {type(searcher.start)}" + ) + assert isinstance(searcher.end, datetime), ( + f"searcher.end should be datetime, got {type(searcher.end)}" + ) + assert searcher.start == datetime(2026, 5, 1, 0, 0, 0, tzinfo=timezone.utc) + assert searcher.end == datetime(2026, 6, 1, 0, 0, 0, tzinfo=timezone.utc) + + class TestExpandConfigSection: """Unit tests for caldav.config.expand_config_section.""" From beceb1aec6466284b7da74a6af96edd31f4affc9 Mon Sep 17 00:00:00 2001 From: Tobias Brox Date: Tue, 17 Mar 2026 19:53:21 +0100 Subject: [PATCH 06/10] fix: send explicit resourcetype PROPFIND in XandikosServer.is_accessible() MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A bodyless PROPFIND is treated as allprop by RFC 4918 §9.1. Xandikos handles allprop on the RootPage by iterating all registered properties, several of which (owner, creationdate, comment) are not implemented on RootPage and raise NotImplementedError, producing spurious ERROR log lines on every liveness poll during test-server startup. Send a minimal PROPFIND body that asks only for {DAV:}resourcetype. This verifies the server is a real WebDAV endpoint (no false positives from non-DAV HTTP servers) without triggering allprop enumeration. The Content-Type: application/xml header is required; without it Xandikos rejects the request with 415 Unsupported Media Type. Co-Authored-By: Claude Sonnet 4.6 --- caldav/testing.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/caldav/testing.py b/caldav/testing.py index bc7ef032..28a552d8 100644 --- a/caldav/testing.py +++ b/caldav/testing.py @@ -146,6 +146,8 @@ def is_accessible(self) -> bool: response = requests.request( "PROPFIND", f"http://{self.host}:{self.port}", + headers={"Depth": "0", "Content-Type": "application/xml"}, + data=b'', timeout=2, ) return response.status_code in (200, 207, 401, 403, 404) From d979e1cd3f3d027587b1e319c3df1a1e1d493fd6 Mon Sep 17 00:00:00 2001 From: Tobias Brox Date: Wed, 18 Mar 2026 20:06:51 +0100 Subject: [PATCH 07/10] docs: rewrite configfile.rst; add tests for inherits and env-var expansion Co-Authored-By: Claude Sonnet 4.6 --- docs/source/configfile.rst | 264 ++++++++++++++++++++++++++++++++++--- docs/source/index.rst | 1 + tests/test_caldav_unit.py | 91 +++++++++++++ 3 files changed, 337 insertions(+), 19 deletions(-) diff --git a/docs/source/configfile.rst b/docs/source/configfile.rst index e6cc9df8..825d0874 100644 --- a/docs/source/configfile.rst +++ b/docs/source/configfile.rst @@ -2,46 +2,272 @@ Config file format ================== - -The :func:`caldav.get_davclient`, :func:`caldav.get_calendar`, and :func:`caldav.get_calendars` functions can read from a config file. It will look for it in the following locations: +The :func:`caldav.davclient.get_davclient`, :func:`caldav.davclient.get_calendar`, and +:func:`caldav.davclient.get_calendars` functions read connection parameters from a config +file. The file is searched in the following locations (first match wins): * ``$HOME/.config/caldav/calendar.conf`` * ``$HOME/.config/caldav/calendar.yaml`` * ``$HOME/.config/caldav/calendar.json`` * ``$HOME/.config/calendar.conf`` +* ``/etc/caldav/calendar.conf`` * ``/etc/calendar.conf`` -The config file has to be valid json or yaml (support for toml and Apple pkl may be considered). +The config file must be valid JSON or YAML. The path can also be given +explicitly via the ``CALDAV_CONFIG_FILE`` environment variable. + +Sections +======== -The config file is expected to be divided in sections, where each section can describe locations and credentials to a CalDAV server, a CalDAV calendar or a collection of calendars/servers. +The file is divided into named sections. Each section contains key/value +pairs describing how to connect to a CalDAV server and optionally which +calendar to select. The section to use is chosen as follows (first match +wins): -A config section can be given either through parameters to :func:`caldav.get_davclient` or by enviornment variable ``CALDAV_CONFIG_SECTION``. If no section is given, the ``default`` section is used. +1. The ``config_section`` parameter passed to :func:`~caldav.davclient.get_davclient` / + :func:`~caldav.davclient.get_calendars`. +2. The ``CALDAV_CONFIG_SECTION`` environment variable. +3. The ``default`` section. Connection parameters ===================== -The section should contain configuration keys and values. All configuration keys starting with ``caldav_`` is considered to be connection parameters and is passed to the DAVClient object. Typically, ``caldav_url``, ``caldav_username`` and ``caldav_password`` should be passed. +All keys starting with ``caldav_`` are connection parameters passed to the +:class:`~caldav.davclient.DAVClient` constructor after stripping the prefix. +The most common ones are: + +.. list-table:: + :header-rows: 1 + + * - Config key + - DAVClient parameter + - Notes + * - ``caldav_url`` + - ``url`` + - CalDAV server URL + * - ``caldav_username`` or ``caldav_user`` + - ``username`` + - Login name + * - ``caldav_password`` or ``caldav_pass`` + - ``password`` + - Login password + * - ``caldav_proxy`` + - ``proxy`` + - HTTP/HTTPS proxy URL + * - ``caldav_timeout`` + - ``timeout`` + - Request timeout in seconds + * - ``caldav_ssl_verify_cert`` + - ``ssl_verify_cert`` + - ``false`` to skip TLS verification + +The special ``features`` key (not prefixed with ``caldav_``) names a +server-compatibility profile — e.g. ``xandikos``, ``radicale``, ``baikal``. +See :mod:`caldav.compatibility_hints` for the full list of known profiles. + +Environment variable expansion +------------------------------- + +Values may reference environment variables using ``${VAR}`` or +``${VAR:-default}`` syntax: + +.. code-block:: yaml + + default: + caldav_url: https://caldav.example.com/ + caldav_username: ${CALDAV_USER:-alice} + caldav_password: ${CALDAV_PASSWORD} Calendar parameters =================== -The :func:`caldav.get_calendar` and :func:`caldav.get_calendars` functions accept -``calendar_name`` and ``calendar_url`` parameters to select a specific calendar. +:func:`~caldav.davclient.get_calendar` and :func:`~caldav.davclient.get_calendars` accept +``calendar_name`` and ``calendar_url`` to select a specific calendar. These +can also be set in a config section so that a named section always refers to +one particular calendar: + +.. code-block:: yaml + + work_inbox: + caldav_url: https://caldav.example.com/ + caldav_username: alice + caldav_password: secret + calendar_name: Inbox + + work_team: + caldav_url: https://caldav.example.com/ + caldav_username: alice + caldav_password: secret + calendar_url: https://caldav.example.com/cal/shared/ + +Calling ``get_calendars(config_section="work_inbox")`` will return only the +calendar named ``Inbox`` on that server. + +Section inheritance +=================== + +A section may declare ``inherits: `` to copy all values from +another section and then override specific keys. This is useful when several +sections share the same server or credentials: + +.. code-block:: yaml + + base_work: + caldav_url: https://caldav.example.com/ + caldav_username: alice + caldav_password: secret -Inheritance and collections -=========================== + work_inbox: + inherits: base_work + calendar_name: Inbox -A section may ``inherit`` another section. This may typically be used if having several sections in the config file corresponding to the same server/user but different calendars, or several sections corresponding to the same calendar server, but different users. + work_tasks: + inherits: base_work + calendar_name: Tasks -If a section ``contains`` different other sections, it's efficiently a collection of calendars. This is not relevant for 2.0 though. +Inheritance is recursive: a section can inherit from a section that itself +inherits from another. -Simple example -============== +Meta-sections (collections) +============================ + +A section with a ``contains`` key is a *meta-section*: it groups other +sections together. :func:`~caldav.davclient.get_calendars` with a meta-section will +aggregate calendars from all listed sections, including across multiple servers: .. code-block:: yaml - --- - default: - caldav_url: http://caldav.example.com/dav/ - caldav_user: tor - caldav_pass: hunter2 + personal: + caldav_url: https://personal.example.com/ + caldav_username: alice + caldav_password: secret1 + + work: + caldav_url: https://work.example.com/ + caldav_username: alice + caldav_password: secret2 + + all: + contains: + - personal + - work + +Calling ``get_calendars(config_section="all")`` returns calendars from both +servers. Meta-sections are resolved recursively, so a meta-section may +contain other meta-sections. Circular references are detected and ignored. + +A section can also be disabled so it is skipped during expansion: + +.. code-block:: yaml + + old_server: + disable: true + caldav_url: https://old.example.com/ + +Glob patterns and wildcards +---------------------------- + +Instead of listing sections explicitly, ``contains`` may use glob patterns, +and the ``config_section`` argument (or ``CALDAV_CONFIG_SECTION``) itself may +be a glob or ``*``: + +.. code-block:: yaml + + work_inbox: + caldav_url: https://work.example.com/ + caldav_username: alice + caldav_password: secret + calendar_name: Inbox + + work_tasks: + caldav_url: https://work.example.com/ + caldav_username: alice + caldav_password: secret + calendar_name: Tasks + + all_work: + contains: + - work_* + +``get_calendars(config_section="work_*")`` and +``get_calendars(config_section="all_work")`` are therefore equivalent. +``get_calendars(config_section="*")`` returns calendars from every +non-disabled section in the file. + +Environment variables +===================== + +Connection parameters can also be passed via environment variables without a +config file. The variables are mapped as follows: + +.. list-table:: + :header-rows: 1 + + * - Environment variable + - Parameter + * - ``CALDAV_URL`` + - ``url`` + * - ``CALDAV_USERNAME`` or ``CALDAV_USER`` + - ``username`` + * - ``CALDAV_PASSWORD`` or ``CALDAV_PASS`` + - ``password`` + * - ``CALDAV_CONFIG_FILE`` + - Path to config file + * - ``CALDAV_CONFIG_SECTION`` + - Section name (may be a glob) + +Examples +======== + +Minimal single-server config +----------------------------- + +.. code-block:: yaml + + --- + default: + caldav_url: https://caldav.example.com/dav/ + caldav_username: alice + caldav_password: secret + +Multiple servers aggregated under one name +------------------------------------------ + +.. code-block:: yaml + + --- + personal: + caldav_url: https://personal.example.com/ + caldav_username: alice + caldav_password: secret1 + features: xandikos + + work: + caldav_url: https://work.example.com/ + caldav_username: alice.work + caldav_password: secret2 + + all: + contains: + - personal + - work + +Shared base with per-calendar overrides +---------------------------------------- + +.. code-block:: yaml + + --- + _server: &server + caldav_url: https://caldav.example.com/ + caldav_username: alice + caldav_password: secret + + inbox: + inherits: _server + calendar_name: Inbox + + tasks: + inherits: _server + calendar_name: Tasks + calendar_url: https://caldav.example.com/cal/tasks/ diff --git a/docs/source/index.rst b/docs/source/index.rst index 89f83ed0..f1213c57 100644 --- a/docs/source/index.rst +++ b/docs/source/index.rst @@ -20,6 +20,7 @@ Contents about v3-migration tutorial + configfile async jmap howtos diff --git a/tests/test_caldav_unit.py b/tests/test_caldav_unit.py index 2060930b..413b53d0 100755 --- a/tests/test_caldav_unit.py +++ b/tests/test_caldav_unit.py @@ -2228,6 +2228,97 @@ def test_recursive_meta_section(self): assert set(expand_config_section(config, "all")) == {"a", "b", "c"} +class TestConfigSectionInheritance: + """Unit tests for caldav.config.config_section (inherits key).""" + + def test_inherits_copies_parent_values(self): + from caldav.config import config_section + + config = { + "base": {"caldav_url": "https://example.com/", "caldav_username": "alice"}, + "child": {"inherits": "base", "caldav_username": "bob"}, + } + result = config_section(config, "child") + assert result["caldav_url"] == "https://example.com/" + # child's own value overrides parent's + assert result["caldav_username"] == "bob" + + def test_inherits_does_not_affect_parent(self): + from caldav.config import config_section + + config = { + "base": {"caldav_url": "https://example.com/", "caldav_username": "alice"}, + "child": {"inherits": "base", "calendar_name": "Inbox"}, + } + parent = config_section(config, "base") + assert "calendar_name" not in parent + + def test_inherits_recursive(self): + from caldav.config import config_section + + config = { + "grandparent": {"caldav_url": "https://example.com/"}, + "parent": {"inherits": "grandparent", "caldav_username": "alice"}, + "child": {"inherits": "parent", "caldav_password": "secret"}, + } + result = config_section(config, "child") + assert result["caldav_url"] == "https://example.com/" + assert result["caldav_username"] == "alice" + assert result["caldav_password"] == "secret" + + +class TestExpandEnvVars: + """Unit tests for caldav.config.expand_env_vars.""" + + def test_simple_var(self, monkeypatch): + from caldav.config import expand_env_vars + + monkeypatch.setenv("TEST_CALDAV_URL", "https://env.example.com/") + assert expand_env_vars("${TEST_CALDAV_URL}") == "https://env.example.com/" + + def test_default_when_missing(self): + import os + + from caldav.config import expand_env_vars + + os.environ.pop("MISSING_CALDAV_VAR", None) + assert expand_env_vars("${MISSING_CALDAV_VAR:-fallback}") == "fallback" + + def test_empty_default_when_missing(self): + import os + + from caldav.config import expand_env_vars + + os.environ.pop("MISSING_CALDAV_VAR", None) + assert expand_env_vars("${MISSING_CALDAV_VAR}") == "" + + def test_recursive_in_dict(self, monkeypatch): + from caldav.config import expand_env_vars + + monkeypatch.setenv("TEST_USER", "alice") + result = expand_env_vars({"caldav_username": "${TEST_USER}"}) + assert result == {"caldav_username": "alice"} + + def test_env_var_in_config_section(self, tmp_path, monkeypatch): + """end-to-end: env var in config file is expanded before use.""" + import json + + from caldav.config import get_all_file_connection_params + + monkeypatch.setenv("MY_CALDAV_PASS", "s3cr3t") + config = { + "default": { + "caldav_url": "https://example.com/", + "caldav_username": "alice", + "caldav_password": "${MY_CALDAV_PASS}", + } + } + config_file = tmp_path / "calendar.conf" + config_file.write_text(json.dumps(config)) + results = get_all_file_connection_params(str(config_file)) + assert results[0]["password"] == "s3cr3t" + + class TestGetAllFileConnectionParams: """Unit tests for caldav.config.get_all_file_connection_params.""" From df5c2e167485a31688d908f1a9d8ff738571fb93 Mon Sep 17 00:00:00 2001 From: Tobias Brox Date: Wed, 18 Mar 2026 20:22:43 +0100 Subject: [PATCH 08/10] ci: deptry and lychee fixups --- .lycheeignore | 1 + pyproject.toml | 1 + 2 files changed, 2 insertions(+) diff --git a/.lycheeignore b/.lycheeignore index 53ea0099..3e4c08e6 100644 --- a/.lycheeignore +++ b/.lycheeignore @@ -31,3 +31,4 @@ file://.*/scheme:.* http://x/ https://ecloud\.global/remote\.php/.* https://tobixen@e\.email/remote\.php/dav +https://dav\.qq\.com/.* diff --git a/pyproject.toml b/pyproject.toml index 87356a93..31cd6111 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -102,6 +102,7 @@ ignore = ["DEP002"] # Test dependencies (pytest, coverage, etc.) are not import [tool.deptry.per_rule_ignores] DEP001 = ["conf", "h2"] # conf: Local test config, h2: Optional HTTP/2 support +DEP003 = ["aiohttp"] # aiohttp: optional dep used only in caldav/testing.py (XandikosServer) [tool.ruff] line-length = 100 From 1171aa45708638f7acf941edac5b6f61f988dcdd Mon Sep 17 00:00:00 2001 From: Tobias Brox Date: Wed, 18 Mar 2026 20:07:03 +0100 Subject: [PATCH 09/10] test: tweak test server spinup/takedown Co-Authored-By: Claude Sonnet 4.6 --- caldav/base_client.py | 12 ++++++++++-- tests/test_docs.py | 17 ++++++++++++----- 2 files changed, 22 insertions(+), 7 deletions(-) diff --git a/caldav/base_client.py b/caldav/base_client.py index 23ddddd7..01e808d4 100644 --- a/caldav/base_client.py +++ b/caldav/base_client.py @@ -319,7 +319,13 @@ def __enter__(self): return self def __exit__(self, exc_type, exc_val, exc_tb): - self.close() + seen: set[int] = set() + for c in self._clients: + if id(c) not in seen: + c.__exit__(exc_type, exc_val, exc_tb) + seen.add(id(c)) + if not self._clients and self: + self[0].client.__exit__(exc_type, exc_val, exc_tb) return False def close(self): @@ -379,7 +385,9 @@ def __enter__(self): return self._calendar def __exit__(self, exc_type, exc_val, exc_tb): - self.close() + client = self.client + if client: + client.__exit__(exc_type, exc_val, exc_tb) return False def close(self): diff --git a/tests/test_docs.py b/tests/test_docs.py index 640160a3..107f52fd 100644 --- a/tests/test_docs.py +++ b/tests/test_docs.py @@ -1,3 +1,4 @@ +import os import unittest import manuel.codeblock @@ -6,7 +7,7 @@ import manuel.testing import pytest -from .test_servers import client_context, has_test_servers +from .test_servers import has_test_servers # manuel.ignore must be the base to process ignore directives first m = manuel.ignore.Manuel() @@ -18,11 +19,17 @@ @pytest.mark.skipif(not has_test_servers(), reason="No test servers configured") class DocTests(unittest.TestCase): def setUp(self): - # Start a test server and configure environment for get_davclient() - self._test_context = client_context() - self._conn = self._test_context.__enter__() + # Set the env var so each with-block in the tutorial starts its own + # ephemeral test server (via get_davclient / get_calendar / get_calendars). + # Do NOT pre-start a server here — that would cause all blocks to share + # state, which is not what the tutorial intends. + self._old_env = os.environ.get("PYTHON_CALDAV_USE_TEST_SERVER") + os.environ["PYTHON_CALDAV_USE_TEST_SERVER"] = "1" def tearDown(self): - self._test_context.__exit__(None, None, None) + if self._old_env is not None: + os.environ["PYTHON_CALDAV_USE_TEST_SERVER"] = self._old_env + else: + os.environ.pop("PYTHON_CALDAV_USE_TEST_SERVER", None) test_tutorial = manueltest("../docs/source/tutorial.rst") From 4db366718b809dbae149748e14b0caee333a633f Mon Sep 17 00:00:00 2001 From: Tobias Brox Date: Wed, 18 Mar 2026 20:07:19 +0100 Subject: [PATCH 10/10] docs: rewrite and fix tutorial Replace the old tutorial with a comprehensive rewrite: add sections on creating/accessing calendars, creating events from icalendar data, searching with expand, investigating and modifying events, and tasks. Fix all cross-references, imports, grammar (seem/may or may not/one calendar/also exist/three kinds), and add Xandikos test-server setup instructions. Co-Authored-By: Claude Sonnet 4.6 --- docs/source/tutorial.rst | 265 ++++++++++++++++++++----------- examples/basic_usage_examples.py | 2 +- 2 files changed, 172 insertions(+), 95 deletions(-) diff --git a/docs/source/tutorial.rst b/docs/source/tutorial.rst index 259e998c..eed6205b 100644 --- a/docs/source/tutorial.rst +++ b/docs/source/tutorial.rst @@ -3,27 +3,31 @@ Tutorial ======== This tutorial covers basic usage of the python CalDAV client library. -Copy code examples into a file and add a ``breakpoint()`` inside the -with-block to inspect return objects. Do not name your file `caldav.py` -or `calendar.py`, as this may break imports. +Copy code examples into a file. You are encouraged to add a +``breakpoint()`` inside the with-block to inspect return objects. Do +not name your file ``caldav.py`` or ``calendar.py``, as this may break +imports. + +Go through the tutorial twice, first against a Xandikos test server, +and then against a server of your own choice. + +This tutorial only covers the sync API. The async API is quite +similar. A tutorial on the async API will come soon. Ad-hoc Configuration -------------------- -To run the tutorial examples against a test server, you need: +This is needed to get this tutorial working with Xandikos: -* The caldav source with tests: ``git clone https://github.com/python-caldav/caldav.git ; cd caldav`` -* Radicale installed: ``pip install radicale`` +* Xandikos installed: ``pip install xandikos`` * Environment variable set: ``export PYTHON_CALDAV_USE_TEST_SERVER=1`` -With this setup, the with-blocks below will spin up a Radicale server. - +With this setup, the with-blocks below will spin up Xandikos servers. The Xandikos server is by default populated with one calendar. Real Configuration ------------------ -The recommended way to configure caldav is through a config file or -environment variables. Create ``~/.config/caldav/calendar.conf``: +Edit ``~/.config/caldav/calendar.conf``: .. code-block:: yaml @@ -33,79 +37,107 @@ environment variables. Create ``~/.config/caldav/calendar.conf``: caldav_url: https://caldav.example.com/ caldav_username: alice caldav_password: secret + features: xandikos -Or set environment variables: +**Caveat:** De-facto, all CalDAV server implementations seem to have their own dialect of the CalDAV standard. This tutorial has been tested with Xandikos. It may or may not work with your server. For instance, the RFC says "Support for MKCALENDAR on the server is only RECOMMENDED and not REQUIRED", so already on the "Creating Calendars"-section you may get into trouble. There are some workarounds in the CalDAV library for some servers, you can try to put the name of your server implementation in the ``features``-field. If it fails, leave the field blank. -.. code-block:: bash +Remember to unset the ``PYTHON_CALDAV_USE_TEST_SERVER`` environment variable, if set. - # export CALDAV_URL=https://caldav.example.com/ - # export CALDAV_USERNAME=alice - # export CALDAV_PASSWORD=secret +See :doc:`configfile` for the full config file format, including multiple +servers, section inheritance, glob patterns, and calendar selection by name or +URL. + +Creating Calendars +------------------ -With configuration in place, you can use caldav without hardcoding credentials: +Many servers will start with a "clean slate", with no calendars - so to get anything at all working, it's needed to first create a calendar. Calendars have to be owned by a principal. As of v3.0, the way to go is to use the :func:`~caldav.davclient.get_davclient` factory function to get a :class:`caldav.davclient.DAVClient` object, from there use the :meth:`~caldav.davclient.DAVClient.get_principal` to get the :class:`caldav.collection.Principal`-object of the logged-in principal (user). .. code-block:: python - from caldav import get_calendars + from caldav import get_davclient - with get_calendars() as calendars: - for cal in calendars: - print(cal.get_display_name()) + ## Get a client ... + with get_davclient() as client: + ## ... from the client get the principal ... + my_principal = client.get_principal() + ## ... from the principal we can create calendar ... + my_new_calendar = my_principal.make_calendar(name="Teest calendar") + ## Enable the debug breakpoint to investigate the calendar object + #breakpoint() + my_new_calendar.delete() + +The delete-step is unimportant when running towards an ephemeral test server. -Getting Calendars ------------------ +**Tip:** In test mode, the with-block ensures the test server is stopped when done. It also ensures that the HTTP-session is terminated. However, for testing things interactively in the python it's a pain. Usage of the with-block is Best Recommended Practice, but the test server will anyway be terminated when the python process exits, and the HTTP-session will be terminated on timeout or when leaving the test server, whatever comes first. Feel free to just use ``client = get_davclient()`` while you're testing things. -Use :func:`caldav.get_calendars` to get all calendars or filter by name: +**Caveat:** In many settings, communication is done lazily when needed. Things will eventually break if password/url/username is wrong, but perhaps not where you expect it to. To test, you may try out: .. code-block:: python - from caldav import get_calendars, get_calendar, get_davclient + from caldav import get_davclient + ## Invalid domain, invalid password ... + ## ... this probably ought to raise an error? + with get_davclient( + username='alice', + password='hunter2', + url='https://calendar.example.com/dav/') as client: + ... - # First create a calendar to work with - with get_davclient() as client: - my_principal = client.get_principal() - my_principal.make_calendar(name="Work") +Accessing calendars +------------------- + +Use the factory function :func:`caldav.davclient.get_calendars` for listing out all available calendars. For the clean-slate Xandikos server, there should be one calendar. You can pass ``url``, ``username`` and ``password`` to the method to test towards your own calendar servers. + +.. code-block:: python + + from caldav import get_calendars - # Get all calendars with get_calendars() as calendars: - for cal in calendars: - print(cal.get_display_name()) + for calendar in calendars: + print(f"Calendar \"{calendar.get_display_name()}\" has URL {calendar.url}") + +The :func:`caldav.davclient.get_calendar` will give you one calendar. **``get_calendar`` should most often be your primary starting point.** Now please go and play with it: - # Get a specific calendar by name - with get_calendar(calendar_name="Work") as work_calendar: - if work_calendar: - events = work_calendar.search(event=True) +.. code-block:: python -Creating Calendars and Events ------------------------------ + from caldav import get_calendar -Create a test calendar and add an event: + with get_calendar() as calendar: + print(f"Calendar \"{calendar.get_display_name()}\" has URL {calendar.url}") + ## You may add a debugger breakpoint and investigate the object + #breakpoint() + +The calendar has a ``.client`` property which gives the client. + +Creating Events +--------------- + +From the :class:`caldav.collection.Calendar` object, it's possible to use :meth:`~caldav.collection.Calendar.add_event` (``add_todo``, ``add_object`` and others also exist) for adding an event: .. code-block:: python - from caldav import get_davclient + from caldav import get_calendar import datetime - with get_davclient() as client: - my_principal = client.get_principal() - my_new_calendar = my_principal.make_calendar(name="Test calendar") - may17 = my_new_calendar.add_event( + with get_calendar() as cal: + ## Add a may 17 event + may17 = cal.add_event( dtstart=datetime.datetime(2020,5,17,8), dtend=datetime.datetime(2020,5,18,1), uid="may17", summary="Do the needful", rrule={'FREQ': 'YEARLY'}) + ## You may want to inspect the event + #breakpoint() -Add an event from icalendar data: +You have icalendar code and want to put it into the calendar? Easy! .. code-block:: python - from caldav import get_davclient + from caldav import get_calendar - with get_davclient() as client: - my_principal = client.get_principal() - my_new_calendar = my_principal.make_calendar(name="Test calendar") - may17 = my_new_calendar.add_event("""BEGIN:VCALENDAR + with get_calendar() as cal: + may17 = cal.add_event("""BEGIN:VCALENDAR VERSION:2.0 PRODID:-//Example Corp.//CalDAV Client//EN BEGIN:VEVENT @@ -118,107 +150,150 @@ Add an event from icalendar data: END:VEVENT END:VCALENDAR """) + #breakpoint() + Searching --------- -Use search to find events, tasks, or journals: +The best way of getting information out from the calendar is to use the search. CalDAV defines a way to construct and send search queries, but in reality there are huge problems with compatibility. With correct configuration of ``features``, the library will work around misbehaving servers, falling back to client-side filtering if needed. .. code-block:: python - from caldav import get_davclient - from datetime import date - import datetime + from caldav import get_calendar + from datetime import datetime, date - with get_davclient() as client: - my_principal = client.get_principal() - my_new_calendar = my_principal.make_calendar(name="Test calendar") - my_new_calendar.add_event( - dtstart=datetime.datetime(2023,5,17,8), - dtend=datetime.datetime(2023,5,18,1), + with get_calendar() as cal: + cal.add_event( + dtstart=datetime(2023,5,17,8), + dtend=datetime(2023,5,18,1), uid="may17", summary="Do the needful", rrule={'FREQ': 'YEARLY'}) - my_events = my_new_calendar.search( + my_events = cal.search( event=True, start=date(2026,5,1), end=date(2026,6,1), expand=True) - assert len(my_events) == 1 print(my_events[0].data) + #breakpoint() + +The ``expand`` parameter matters for recurring objects. When set it returns all *recurrences* within the search time span. Try to set the end to 2028 with and without ``expand`` and you will probably understand. + +``event`` causes the search to only return events. There are three kinds of objects that can be saved to a calendar (but not all servers support all three) - events, journals and tasks (``VEVENT``, ``VJOURNAL`` and ``VTODO``). This is called Calendar Object Resources in the RFC. Now that's quite a mouthful! To ease things, the word "event" is simply used in documentation and communication. So when reading "event", be aware that most of the time it actually means "a CalendarObjectResource objects such as an event, but it could also be a task or a journal" - and if you contribute code, remember to work on objects of type ``CalendarObjectResource`` rather than ``Event``. + +The return type is a list of objects of the type :class:`~caldav.calendarobjectresource.Event` - for tasks and journals there are similar classes :class:`~caldav.calendarobjectresource.Todo` and :class:`~caldav.calendarobjectresource.Journal`. + +Investigating Events +-------------------- + +Above, ``.data`` is used to access the icalendar data directly. There is also :meth:`~caldav.calendarobjectresource.CalendarObjectResource.get_vobject_instance`, :meth:`~caldav.calendarobjectresource.CalendarObjectResource.get_icalendar_instance` and :meth:`~caldav.calendarobjectresource.CalendarObjectResource.get_icalendar_component`, each yielding a copied object. -The ``expand`` parameter expands recurring events into individual -occurrences within the search interval. The ``event=True`` parameter -filters results to events only (excluding tasks and journals). + +.. code-block:: python + + from caldav import get_calendar + from datetime import datetime, date + + with get_calendar() as cal: + cal.add_event( + dtstart=datetime(2023,5,17,8), + dtend=datetime(2023,5,18,1), + uid="may17", + summary="Do the needful", + rrule={'FREQ': 'YEARLY'}) + + my_events = cal.search( + event=True, + start=date(2026,5,1), + end=date(2026,6,1), + expand=True) + + print(my_events[0].get_icalendar_component()['summary']) + print(my_events[0].get_icalendar_component().duration) + #breakpoint() + +``get_icalendar_component()`` is the easiest way of accessing event data, but there is a **big caveat** there. Events may be recurring. The recurring events may have been changed. Say that you have a meeting every Wed at 10:00, this started in 2024, in 2025 the time was changed to 11:00, at one particular Wed in 2026 the time was pushed to 11:30, the next Wed it was cancelled. This will be represented as *four* components. ``.get_icalendar_component`` will only give you access to the original event! The ``get_icalendar_component()`` is safe to use when doing ``.search(..., expand=True)``, as this will ensure every object is one and only one recurrence. Modifying Events ---------------- -The ``data`` property contains icalendar data as a string: +The ``data`` property contains icalendar data as a string, and you can replace it: .. code-block:: python - from caldav import get_davclient + from caldav import get_calendar from datetime import date import datetime - with get_davclient() as client: - my_principal = client.get_principal() - my_new_calendar = my_principal.make_calendar(name="Test calendar") - my_new_calendar.add_event( + with get_calendar() as cal: + ## Create yearly event ... + cal.add_event( dtstart=datetime.datetime(2023,5,17,8), dtend=datetime.datetime(2023,5,18,1), uid="may17", summary="Do the needful", rrule={'FREQ': 'YEARLY'}) - my_events = my_new_calendar.search( + ## Search for a single recurrence + my_events = cal.search( event=True, start=date(2026,5,1), end=date(2026,6,1), expand=True) - assert len(my_events) == 1 + ## Replace the old summary with a new one my_events[0].data = my_events[0].data.replace("Do the needful", "Have fun!") my_events[0].save() + #breakpoint() -Better practice is to use the icalendar library. The ``component`` -property gives access to the :class:`icalendar.cal.Event` object: +This is not best practice - the thing above may even break due to line wrapping, etc. Best practice is to "borrow" an editable icalendar instance through :meth:`~caldav.calendarobjectresource.CalendarObjectResource.edit_icalendar_component` or :meth:`~caldav.calendarobjectresource.CalendarObjectResource.edit_icalendar_instance`. Note that in the example below we're taking out one particular recurrence, so only that recurrence will be changed. .. code-block:: python - from caldav import get_davclient + from caldav import get_calendar from datetime import date import datetime - with get_davclient() as client: - my_principal = client.get_principal() - my_new_calendar = my_principal.make_calendar(name="Test calendar") - my_new_calendar.add_event( + with get_calendar() as cal: + ## Create a recurring event + cal.add_event( dtstart=datetime.datetime(2023,5,17,8), dtend=datetime.datetime(2023,5,18,1), uid="may17", summary="Do the needful", rrule={'FREQ': 'YEARLY'}) - my_events = my_new_calendar.search( + ## Find a particular recurrence + my_events = cal.search( event=True, start=date(2026,5,1), end=date(2026,6,1), expand=True) - assert len(my_events) == 1 - print(f"Event starts at {my_events[0].component.start}") - with my_events[0].edit_icalendar_instance() as cal: - cal.subcomponents[0]['summary'] = "Norwegian national day celebrations" + ## Edit the summary using the "borrowing pattern": + with my_events[0].edit_icalendar_component() as event_ical: + ## "component" is always safe after an expanded search + event_ical['summary'] = "Norwegian national day celebrations" my_events[0].save() + ## Let's take out the event again: + may17 = cal.get_event_by_uid('may17') + + ## Inspect may17 in a debug breakpoint + #breakpoint() + +How does the new may17-event look from a technical point of view, when we're editing only the 2026-edition? Enable the breakpoint and find out! Use ``.get_icalendar_instance()`` or ``.data`` + + Tasks ----- -Create a task list and work with tasks: +Anything you can do with events can also be done with tasks. You may try to use ``get_calendar()`` below instead of creating a calendar. On most servers all calendars can be used for both tasks and events, however some servers (notably, Zimbra) differs between tasklists and calendars, and we need to create (or select) a tasklist and not a calendar. The CalDAV standard allows to define this through the "supported calendar component set" parameter (ignored on most servers though). + +There is some extra functionality around tasks, including the possibility to :meth:`~caldav.calendarobjectresource.Todo.complete` them. .. code-block:: python @@ -227,21 +302,23 @@ Create a task list and work with tasks: with get_davclient() as client: my_principal = client.get_principal() - my_new_calendar = my_principal.make_calendar( - name="Test calendar", supported_calendar_component_set=['VTODO']) - my_new_calendar.add_todo( + ## This can be read as "create me a tasklist" + cal = my_principal.make_calendar( + name="Test tasklist", supported_calendar_component_set=['VTODO']) + ## ... but for most servers it's an ordinary calendar! + cal.add_todo( summary="prepare for the Norwegian national day", due=date(2025,5,16)) - my_tasks = my_new_calendar.search( + my_tasks = cal.search( todo=True) assert len(my_tasks) == 1 my_tasks[0].complete() - my_tasks = my_new_calendar.search( + my_tasks = cal.search( todo=True) assert len(my_tasks) == 0 - my_tasks = my_new_calendar.search( + my_tasks = cal.search( todo=True, include_completed=True) - assert len(my_tasks) == 1 + assert my_tasks Further Reading --------------- @@ -251,8 +328,8 @@ See the :ref:`examples:examples` folder for more code, including and `scheduling examples `_ for invites. -The `test code `_ -covers most features. +The `integration tests `_ +covers most features, but is not much optimized for readability. There is also a `command line interface `_ built around the caldav library. diff --git a/examples/basic_usage_examples.py b/examples/basic_usage_examples.py index 6cfa7a31..7088a42b 100644 --- a/examples/basic_usage_examples.py +++ b/examples/basic_usage_examples.py @@ -88,7 +88,7 @@ def print_calendars_demo(calendars): ## this principal. print("your principal has %i calendars:" % len(calendars)) for c in calendars: - print(" Name: %-36s URL: %s" % (c.name, c.url)) + print(" Name: %-36s URL: %s" % (c.get_display_name(), c.url)) else: print("your principal has no calendars")