Add response_delay_seconds parameter for testing timeouts

CHANGELOG.rst

@@ -4,6 +4,8 @@ Changelog
 Next
 ----
 
+- Add ``response_delay_seconds`` parameter to ``MockVWS`` for simulating slow server responses and testing timeout handling.
+
 2025.03.10.1
 ------------
 

src/mock_vws/_requests_mock_server/decorators.py

@@ -1,11 +1,15 @@
 """Decorators for using the mock."""
 
 import re
+import time
+from collections.abc import Callable, Mapping
 from contextlib import ContextDecorator
-from typing import Literal, Self
+from typing import Any, Literal, Self
 from urllib.parse import urljoin, urlparse
 
+import requests
 from beartype import BeartypeConf, beartype
+from requests import PreparedRequest
 from responses import RequestsMock
 
 from mock_vws.database import VuforiaDatabase
@@ -22,6 +26,9 @@
 from .mock_web_query_api import MockVuforiaWebQueryAPI
 from .mock_web_services_api import MockVuforiaWebServicesAPI
 
+_ResponseType = tuple[int, Mapping[str, str], str]
+_Callback = Callable[[PreparedRequest], _ResponseType]
+
 _STRUCTURAL_SIMILARITY_MATCHER = StructuralSimilarityMatcher()
 _BRISQUE_TRACKING_RATER = BrisqueTargetTrackingRater()
 
@@ -62,6 +69,7 @@ def __init__(
         processing_time_seconds: float = 2.0,
         target_tracking_rater: TargetTrackingRater = _BRISQUE_TRACKING_RATER,
         real_http: bool = False,
+        response_delay_seconds: float = 0.0,
     ) -> None:
         """Route requests to Vuforia's Web Service APIs to fakes of those
         APIs.
@@ -81,12 +89,15 @@ def __init__(
             duplicate_match_checker: A callable which takes two image values
                 and returns whether they are duplicates.
             target_tracking_rater: A callable for rating targets for tracking.
+            response_delay_seconds: The number of seconds to delay each
+                response by. This can be used to test timeout handling.
 
         Raises:
             MissingSchemeError: There is no scheme in a given URL.
         """
         super().__init__()
         self._real_http = real_http
+        self._response_delay_seconds = response_delay_seconds
         self._mock: RequestsMock
         self._target_manager = TargetManager()
 
@@ -121,42 +132,72 @@ def add_database(self, database: VuforiaDatabase) -> None:
         """
         self._target_manager.add_database(database=database)
 
+    @staticmethod
+    def _wrap_callback(
+        callback: _Callback,
+        delay_seconds: float,
+    ) -> _Callback:
+        """Wrap a callback to add a response delay."""
+
+        def wrapped(
+            request: PreparedRequest,
+        ) -> _ResponseType:
+            """Handle the response delay and timeout logic."""
+            # req_kwargs is added dynamically by the responses
+            # library onto PreparedRequest objects - it is not
+            # in the requests type stubs.
+            req_kwargs: dict[str, Any] = getattr(request, "req_kwargs", {})
+            timeout = req_kwargs.get("timeout")
+            # requests allows timeout as a (connect, read)
+            # tuple. The delay simulates server response
+            # time, so compare against the read timeout.
+            effective: float | None = None
+            if isinstance(timeout, tuple):
+                if isinstance(timeout[1], (int, float)):
+                    effective = float(timeout[1])
+            elif isinstance(timeout, (int, float)):
+                effective = float(timeout)
+
+            if effective is not None and delay_seconds > effective:
+                time.sleep(effective)
+                raise requests.exceptions.Timeout
+
+            result = callback(request)
+            time.sleep(delay_seconds)
+            return result
+
+        return wrapped
+
     def __enter__(self) -> Self:
         """Start an instance of a Vuforia mock.
 
         Returns:
             ``self``.
         """
         mock = RequestsMock(assert_all_requests_are_fired=False)
-        for vws_route in self._mock_vws_api.routes:
-            url_pattern = urljoin(
-                base=self._base_vws_url,
-                url=f"{vws_route.path_pattern}$",
-            )
-            compiled_url_pattern = re.compile(pattern=url_pattern)
-
-            for vws_http_method in vws_route.http_methods:
-                mock.add_callback(
-                    method=vws_http_method,
-                    url=compiled_url_pattern,
-                    callback=getattr(self._mock_vws_api, vws_route.route_name),
-                    content_type=None,
-                )
 
-        for vwq_route in self._mock_vwq_api.routes:
-            url_pattern = urljoin(
-                base=self._base_vwq_url,
-                url=f"{vwq_route.path_pattern}$",
-            )
-            compiled_url_pattern = re.compile(pattern=url_pattern)
-
-            for vwq_http_method in vwq_route.http_methods:
-                mock.add_callback(
-                    method=vwq_http_method,
-                    url=compiled_url_pattern,
-                    callback=getattr(self._mock_vwq_api, vwq_route.route_name),
-                    content_type=None,
+        for api, base_url in (
+            (self._mock_vws_api, self._base_vws_url),
+            (self._mock_vwq_api, self._base_vwq_url),
+        ):
+            for route in api.routes:
+                url_pattern = urljoin(
+                    base=base_url,
+                    url=f"{route.path_pattern}$",
                 )
+                compiled_url_pattern = re.compile(pattern=url_pattern)
+
+                for http_method in route.http_methods:
+                    original_callback = getattr(api, route.route_name)
+                    mock.add_callback(
+                        method=http_method,
+                        url=compiled_url_pattern,
+                        callback=self._wrap_callback(
+                            callback=original_callback,
+                            delay_seconds=self._response_delay_seconds,
+                        ),
+                        content_type=None,
+                    )
 
         if self._real_http:
             all_requests_pattern = re.compile(pattern=".*")

tests/mock_vws/test_requests_mock_usage.py

@@ -111,6 +111,91 @@ def test_real_http() -> None:
             request_unmocked_address()
 
 
+class TestResponseDelay:
+    """Tests for the response delay feature."""
+
+    @staticmethod
+    def test_default_no_delay() -> None:
+        """By default, there is no response delay."""
+        with MockVWS():
+            # With a very short timeout, the request should still succeed
+            # because there is no delay
+            response = requests.get(
+                url="https://vws.vuforia.com/summary",
+                headers={
+                    "Date": rfc_1123_date(),
+                    "Authorization": "bad_auth_token",
+                },
+                data=b"",
+                timeout=0.5,
+            )
+            # We just care that no timeout occurred, not the response content
+            assert response.status_code is not None
+
+    @staticmethod
+    def test_delay_causes_timeout() -> None:
+        """
+        When response_delay_seconds is set higher than the client
+        timeout,
+        a Timeout exception is raised.
+        """
+        with (
+            MockVWS(response_delay_seconds=0.5),
+            pytest.raises(expected_exception=requests.exceptions.Timeout),
+        ):
+            requests.get(
+                url="https://vws.vuforia.com/summary",
+                headers={
+                    "Date": rfc_1123_date(),
+                    "Authorization": "bad_auth_token",
+                },
+                data=b"",
+                timeout=0.1,
+            )
+
+    @staticmethod
+    def test_delay_allows_completion() -> None:
+        """
+        When response_delay_seconds is set lower than the client
+        timeout,
+        the request completes successfully.
+        """
+        with MockVWS(response_delay_seconds=0.1):
+            # This should succeed because timeout > delay
+            response = requests.get(
+                url="https://vws.vuforia.com/summary",
+                headers={
+                    "Date": rfc_1123_date(),
+                    "Authorization": "bad_auth_token",
+                },
+                data=b"",
+                timeout=2.0,
+            )
+            assert response.status_code is not None
+
+    @staticmethod
+    def test_delay_with_tuple_timeout() -> None:
+        """
+        The response delay works correctly with tuple timeouts
+        (connect_timeout, read_timeout).
+        """
+        with (
+            MockVWS(response_delay_seconds=0.5),
+            pytest.raises(expected_exception=requests.exceptions.Timeout),
+        ):
+            # Tuple timeout: (connect_timeout, read_timeout)
+            # The read timeout (0.1) is less than the delay (0.5)
+            requests.get(
+                url="https://vws.vuforia.com/summary",
+                headers={
+                    "Date": rfc_1123_date(),
+                    "Authorization": "bad_auth_token",
+                },
+                data=b"",
+                timeout=(5.0, 0.1),
+            )
+
+
 class TestProcessingTime:
     """Tests for the time taken to process targets in the mock."""