From 49db75dae8fa2a43271dc4b1a8c012159191c172 Mon Sep 17 00:00:00 2001 From: Peter Bieringer Date: Thu, 2 Apr 2026 13:11:24 +0200 Subject: [PATCH] delay and sharing/token: align+fix --- radicale/tests/test_auth.py | 29 ++++++++---- radicale/tests/test_base.py | 21 ++++++++- radicale/tests/test_sharing.py | 86 +++++++++++++++++++++++++--------- 3 files changed, 104 insertions(+), 32 deletions(-) diff --git a/radicale/tests/test_auth.py b/radicale/tests/test_auth.py index 65e277b3..d75e9868 100644 --- a/radicale/tests/test_auth.py +++ b/radicale/tests/test_auth.py @@ -23,10 +23,10 @@ Radicale tests with simple requests and authentication. """ import base64 +import datetime import logging import os import sys -import time from typing import Iterable, Tuple, Union import pytest @@ -61,7 +61,7 @@ class TestBaseAuthRequests(BaseTest): def _test_htpasswd(self, htpasswd_encryption: str, htpasswd_content: str, test_matrix: Union[str, Iterable[Tuple[str, str, bool]]] - = "ascii", delay: int = 0) -> None: + = "ascii", delay: float = 0) -> None: """Test htpasswd authentication with user "tmp" and password "bepo" for ``test_matrix`` "ascii" or user "😀" and password "🔑" for ``test_matrix`` "unicode".""" @@ -223,16 +223,25 @@ class TestBaseAuthRequests(BaseTest): def test_htpasswd_login_cache_failed_delay_plain(self, caplog) -> None: caplog.set_level(logging.INFO) self.configure({"auth": {"cache_logins": "True"}}) - delay = 1 - delay_ns = delay * 10**9 * 0.5 # delay minimum jitter - time_ns_begin1 = time.time_ns() + delay = .3 + delay_min = delay * 0.9 # no random jitter during test + delay_max = delay + 0.2 # no random jitter during test + if sys.platform == "darwin": # no reliable sleep times + delay_max = delay_max * 1.5 + + time_begin = datetime.datetime.now() self._test_htpasswd("plain", "tmp:bepo", [("tmp", "bepo1", False)], delay=delay) - time_ns_end1 = time.time_ns() - time_ns_begin2 = time.time_ns() + time_end = datetime.datetime.now() + time_delta = (time_end - time_begin).total_seconds() + assert time_delta > delay_min + assert time_delta < delay_max + + time_begin = datetime.datetime.now() self._test_htpasswd("plain", "tmp:bepo", [("tmp", "bepo1", False)], delay=delay) - time_ns_end2 = time.time_ns() - assert (time_ns_end1 - time_ns_begin1) > delay_ns - assert (time_ns_end2 - time_ns_begin2) > delay_ns + time_end = datetime.datetime.now() + time_delta = (time_end - time_begin).total_seconds() + assert time_delta > delay_min + assert time_delta < delay_max # htpasswd file cache def test_htpasswd_file_cache(self, caplog) -> None: diff --git a/radicale/tests/test_base.py b/radicale/tests/test_base.py index 787364dc..d17f3d15 100644 --- a/radicale/tests/test_base.py +++ b/radicale/tests/test_base.py @@ -21,9 +21,11 @@ Radicale tests with simple requests. """ +import datetime import logging import os import posixpath +import sys import urllib from typing import Any, Callable, ClassVar, Iterable, List, Optional, Tuple @@ -727,7 +729,7 @@ permissions: RrWw""") assert responses["/calendar.ics/"] == 200 self.get("/calendar.ics/", check=404) - def test_delete_collection_global_forbid(self) -> None: + def test_delete_collection_global_forbid_base(self) -> None: """Delete a collection (expect forbidden).""" self.configure({"rights": {"permit_delete_collection": False}}) self.mkcalendar("/calendar.ics/") @@ -736,6 +738,23 @@ permissions: RrWw""") _, responses = self.delete("/calendar.ics/", check=401) self.get("/calendar.ics/", check=200) + def test_delete_collection_global_forbid_delay(self) -> None: + """Delete a collection (expect forbidden, check delay).""" + delay = .3 + delay_min = delay * 0.9 # no random jitter during test + delay_max = delay + 0.2 # no random jitter during test + if sys.platform == "darwin": # no reliable sleep times + delay_max = delay_max * 1.5 + + self.configure({"rights": {"permit_delete_collection": False}, "auth": {"delay": delay}}) + self.mkcalendar("/calendar.ics/") + time_begin = datetime.datetime.now() + _, responses = self.delete("/calendar.ics/", check=401) + time_end = datetime.datetime.now() + time_delta = (time_end - time_begin).total_seconds() + assert time_delta > delay_min + assert time_delta < delay_max + def test_delete_collection_global_forbid_explicit_permit(self) -> None: """Delete a collection with permitted path (expect permit).""" self.configure({"rights": {"permit_delete_collection": False}}) diff --git a/radicale/tests/test_sharing.py b/radicale/tests/test_sharing.py index 4450e685..e3581f77 100644 --- a/radicale/tests/test_sharing.py +++ b/radicale/tests/test_sharing.py @@ -20,11 +20,12 @@ Radicale tests related to sharing. """ +import datetime import json import logging import os import re -import time +import sys from typing import Dict, Sequence, Tuple, Union from radicale import sharing, xmlutils @@ -202,7 +203,7 @@ class TestSharingApiSanity(BaseTest): "collection_by_token": "False"} }) - def test_sharing_api_base_no_auth(self) -> None: + def test_sharing_api_base_no_auth_basic(self) -> None: """POST request at '/.sharing' without authentication.""" # disabled for path in ["/.sharing", "/.sharing/"]: @@ -258,6 +259,39 @@ class TestSharingApiSanity(BaseTest): }) _, headers, _ = self.request("POST", path, check=401) + def test_sharing_api_base_no_auth_delay(self) -> None: + delay = .3 + delay_min = delay * 0.9 # no random jitter during test + delay_max = delay + 0.2 # no random jitter during test + if sys.platform == "darwin": # no reliable sleep times + delay_max = delay_max * 1.5 + + for path in ["/.sharing", "/.sharing/"]: + time_begin = datetime.datetime.now() + _, headers, _ = self.request("POST", path, check=404) + time_end = datetime.datetime.now() + time_delta = (time_end - time_begin).total_seconds() + assert time_delta < delay_min # 404 should have no delay + + path = "/.sharing/" + + for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)): + logging.info("\n*** test: %s", db_type) + self.configure({"sharing": {"type": db_type}}) + + # no database is active + logging.info("\n*** check API hook base: map=True token=False (incl. delay)") + self.configure({"sharing": { + "collection_by_map": "True", + "collection_by_token": "False"}, + "auth": {"delay": delay}}) + time_begin = datetime.datetime.now() + _, headers, _ = self.request("POST", path, check=401) + time_end = datetime.datetime.now() + time_delta = (time_end - time_begin).total_seconds() + assert time_delta > delay_min + assert time_delta < delay_max + def test_sharing_api_base_with_auth(self) -> None: """POST request at '/.sharing' with authentication.""" self.configure({"auth": {"type": "htpasswd", @@ -834,7 +868,7 @@ class TestSharingApiSanity(BaseTest): assert "Status='success'" in answer logging.info("\n*** fetch collection using invalid token") - _, headers, answer = self.request("GET", "/.token/v1/invalidtoken/", check=401) + _, headers, answer = self.request("GET", "/.token/v1/invalidtoken/", check=403) logging.info("\n*** fetch collection using token") _, headers, answer = self.request("GET", token, check=200) @@ -846,7 +880,7 @@ class TestSharingApiSanity(BaseTest): assert "Status='success'" in answer logging.info("\n*** fetch collection using disabled token") - _, headers, answer = self.request("GET", token, check=401) + _, headers, answer = self.request("GET", token, check=403) logging.info("\n*** enable token (form->text)") form_array = ["PathOrToken=" + token] @@ -877,12 +911,15 @@ class TestSharingApiSanity(BaseTest): _, headers, answer = self._sharing_api_form("token", "delete", check=404, login="owner:ownerpw", form_array=form_array) logging.info("\n*** fetch collection using deleted token") - _, headers, answer = self.request("GET", token, check=401) + _, headers, answer = self.request("GET", token, check=403) def test_sharing_api_token_usage_delay(self) -> None: """share-by-token API tests - real usage.""" delay = .3 - delay_ns = delay * 10**9 * 0.5 # delay minimum jitter + delay_min = delay * 0.9 # no random jitter during test + delay_max = delay + 0.2 # no random jitter during test + if sys.platform == "darwin": # no reliable sleep times + delay_max = delay_max * 1.5 self.configure({"auth": {"type": "htpasswd", "delay": delay, @@ -930,16 +967,19 @@ class TestSharingApiSanity(BaseTest): assert False logging.info("\n*** fetch collection using invalid token") - time_ns_begin = time.time_ns() - _, headers, answer = self.request("GET", "/.token/v1/invalidtoken/", check=401) - time_ns_end = time.time_ns() - assert (time_ns_end - time_ns_begin) > delay_ns + time_begin = datetime.datetime.now() + _, headers, answer = self.request("GET", "/.token/v1/invalidtoken/", check=403) + time_end = datetime.datetime.now() + time_delta = (time_end - time_begin).total_seconds() + assert time_delta > delay_min + assert time_delta < delay_max logging.info("\n*** fetch collection using token") - time_ns_begin = time.time_ns() + time_begin = datetime.datetime.now() _, headers, answer = self.request("GET", token, check=200) - time_ns_end = time.time_ns() - assert (time_ns_end - time_ns_begin) < delay_ns + time_end = datetime.datetime.now() + time_delta = (time_end - time_begin).total_seconds() + assert time_delta < delay_min # no delay assert "UID:event" in answer logging.info("\n*** disable token (form->text)") @@ -948,10 +988,12 @@ class TestSharingApiSanity(BaseTest): assert "Status='success'" in answer logging.info("\n*** fetch collection using disabled token") - time_ns_begin = time.time_ns() - _, headers, answer = self.request("GET", token, check=401) - time_ns_end = time.time_ns() - assert (time_ns_end - time_ns_begin) > delay_ns + time_begin = datetime.datetime.now() + _, headers, answer = self.request("GET", token, check=403) + time_end = datetime.datetime.now() + time_delta = (time_end - time_begin).total_seconds() + assert time_delta > delay_min + assert time_delta < delay_max logging.info("\n*** delete token (json->json)") json_dict = {'PathOrToken': token} @@ -961,10 +1003,12 @@ class TestSharingApiSanity(BaseTest): assert answer_dict['Status'] == "success" logging.info("\n*** fetch collection using deleted token with delay") - time_ns_begin = time.time_ns() - _, headers, answer = self.request("GET", token, check=401) - time_ns_end = time.time_ns() - assert (time_ns_end - time_ns_begin) > delay_ns + time_begin = datetime.datetime.now() + _, headers, answer = self.request("GET", token, check=403) + time_end = datetime.datetime.now() + time_delta = (time_end - time_begin).total_seconds() + assert time_delta > delay_min + assert time_delta < delay_max def test_sharing_api_map_basic(self) -> None: """share-by-map API basic tests."""