Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions CHANGELOG.rst
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,8 @@ Changelog
Next
----

- Add ``sleep_fn`` parameter to ``MockVWS`` for injecting a custom delay strategy, enabling deterministic and fast tests without monkey-patching.

2026.02.15.3
------------

Expand Down
12 changes: 10 additions & 2 deletions src/mock_vws/_requests_mock_server/decorators.py
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,7 @@ def __init__(
target_tracking_rater: TargetTrackingRater = _BRISQUE_TRACKING_RATER,
real_http: bool = False,
response_delay_seconds: float = 0.0,
sleep_fn: Callable[[float], None] = time.sleep,
) -> None:
"""Route requests to Vuforia's Web Service APIs to fakes of those
APIs.
Expand All @@ -91,13 +92,18 @@ def __init__(
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.
sleep_fn: The function to use for sleeping during response
delays. Defaults to ``time.sleep``. Inject a custom
function to control virtual time in tests without
monkey-patching.

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._sleep_fn = sleep_fn
self._mock: RequestsMock
self._target_manager = TargetManager()

Expand Down Expand Up @@ -136,6 +142,7 @@ def add_database(self, database: VuforiaDatabase) -> None:
def _wrap_callback(
callback: _Callback,
delay_seconds: float,
sleep_fn: Callable[[float], None],
) -> _Callback:
"""Wrap a callback to add a response delay."""

Expand All @@ -159,11 +166,11 @@ def wrapped(
effective = float(timeout)

if effective is not None and delay_seconds > effective:
time.sleep(effective)
sleep_fn(effective)
raise requests.exceptions.Timeout

result = callback(request)
time.sleep(delay_seconds)
sleep_fn(delay_seconds)
return result

return wrapped
Expand Down Expand Up @@ -195,6 +202,7 @@ def __enter__(self) -> Self:
callback=self._wrap_callback(
callback=original_callback,
delay_seconds=self._response_delay_seconds,
sleep_fn=self._sleep_fn,
),
content_type=None,
)
Expand Down
48 changes: 48 additions & 0 deletions tests/mock_vws/test_requests_mock_usage.py
Original file line number Diff line number Diff line change
Expand Up @@ -195,6 +195,54 @@ def test_delay_with_tuple_timeout() -> None:
timeout=(5.0, 0.1),
)

@staticmethod
def test_custom_sleep_fn_called_on_delay() -> None:
"""
When a custom ``sleep_fn`` is provided, it is called instead of
``time.sleep`` for the non-timeout delay path.
"""
calls: list[float] = []
with MockVWS(
response_delay_seconds=5.0,
sleep_fn=calls.append,
):
requests.get(
url="https://vws.vuforia.com/summary",
headers={
"Date": rfc_1123_date(),
"Authorization": "bad_auth_token",
},
data=b"",
timeout=30,
)
assert calls == [5.0]

@staticmethod
def test_custom_sleep_fn_called_on_timeout() -> None:
"""
When a custom ``sleep_fn`` is provided, it is called instead of
``time.sleep`` for the timeout path.
"""
calls: list[float] = []
with (
MockVWS(
response_delay_seconds=5.0,
sleep_fn=calls.append,
),
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=1.0,
)
# sleep_fn should have been called with the effective timeout
assert calls == [1.0]


class TestProcessingTime:
"""Tests for the time taken to process targets in the mock."""
Expand Down