Time testing utilities for the provide-io ecosystem.
Fixtures and utilities for mocking time, freezing time, and testing
time-dependent code across any project that depends on provide.foundation.
Classes
Timer specifically for benchmarking code.
Initialize benchmark timer.
Source code in provide/testkit/time/classes.py
| def __init__(self) -> None:
"""Initialize benchmark timer."""
self.measurements: list[float] = []
|
Attributes
property
Get minimum execution time.
property
Get maximum execution time.
property
Get average execution time.
Functions
measure(
func: Callable[..., Any], *args: Any, **kwargs: Any
) -> tuple[Any, float]
Measure execution time of a function.
Parameters:
| Name |
Type |
Description |
Default |
func
|
Callable[..., Any]
|
|
required
|
*args
|
Any
|
|
()
|
**kwargs
|
Any
|
Function keyword arguments
|
{}
|
Returns:
Source code in provide/testkit/time/classes.py
| def measure(self, func: Callable[..., Any], *args: Any, **kwargs: Any) -> tuple[Any, float]:
"""Measure execution time of a function.
Args:
func: Function to measure
*args: Function arguments
**kwargs: Function keyword arguments
Returns:
Tuple of (result, duration)
"""
start = time.perf_counter()
result = func(*args, **kwargs)
duration = time.perf_counter() - start
self.measurements.append(duration)
return result, duration
|
assert_faster_than(seconds: float) -> None
Assert all measurements were faster than threshold.
Source code in provide/testkit/time/classes.py
| def assert_faster_than(self, seconds: float) -> None:
"""Assert all measurements were faster than threshold."""
if not self.measurements:
raise AssertionError("No measurements taken")
if self.max_time > seconds:
raise AssertionError(f"Maximum time {self.max_time:.3f}s exceeded threshold {seconds:.3f}s")
|
FrozenTime(frozen_time: datetime | None = None)
Context manager for freezing time at a specific point.
Initialize frozen time context.
Parameters:
| Name |
Type |
Description |
Default |
frozen_time
|
datetime | None
|
Time to freeze at (defaults to now)
|
None
|
Source code in provide/testkit/time/classes.py
| def __init__(self, frozen_time: datetime.datetime | None = None) -> None:
"""Initialize frozen time context.
Args:
frozen_time: Time to freeze at (defaults to now)
"""
self.frozen_time = frozen_time or datetime.datetime.now()
self.original_time = time.time
self.original_datetime = datetime.datetime
self.patches: list[Any] = []
|
Functions
tick(seconds: float = 1.0) -> None
Advance the frozen time by the specified seconds.
Source code in provide/testkit/time/classes.py
| def tick(self, seconds: float = 1.0) -> None:
"""Advance the frozen time by the specified seconds."""
self.frozen_time += datetime.timedelta(seconds=seconds)
# Update mocks
for p in self.patches:
if hasattr(p, "return_value"):
p.return_value = self.frozen_time.timestamp()
|
Mock for testing rate-limited code.
Initialize mock rate limiter.
Source code in provide/testkit/time/classes.py
| def __init__(self) -> None:
"""Initialize mock rate limiter."""
self.calls = []
self.should_limit = False
self.limit_after = None
self.call_count = 0
|
Functions
Check if rate limit is exceeded.
Source code in provide/testkit/time/classes.py
| def check(self) -> bool:
"""Check if rate limit is exceeded."""
self.call_count += 1
self.calls.append(time.time())
if self.limit_after and self.call_count > self.limit_after:
return False # Rate limited
return not self.should_limit
|
Reset the rate limiter.
Source code in provide/testkit/time/classes.py
| def reset(self) -> None:
"""Reset the rate limiter."""
self.calls.clear()
self.call_count = 0
self.should_limit = False
self.limit_after = None
|
set_limit(after_calls: int) -> None
Set to limit after N calls.
Source code in provide/testkit/time/classes.py
| def set_limit(self, after_calls: int) -> None:
"""Set to limit after N calls."""
self.limit_after = after_calls
|
Advanced time manipulation class for testing.
Provides methods to:
- Freeze time
- Speed up/slow down time
- Jump to specific times
Initialize the TimeMachine.
Source code in provide/testkit/time/classes.py
| def __init__(self) -> None:
"""Initialize the TimeMachine."""
self.current_time = time.time()
self.speed_multiplier = 1.0
self.patches: list[Any] = []
self.is_frozen = False
# Register in global registry for efficient cleanup
_active_time_machines.add(self)
|
Functions
freeze(at: float | None = None) -> TimeMachine
Freeze time at a specific timestamp.
Source code in provide/testkit/time/classes.py
| def freeze(self, at: float | None = None) -> TimeMachine:
"""Freeze time at a specific timestamp."""
self.is_frozen = True
self.current_time = at or time.time()
# Patch global time.time
global_patcher = patch("time.time", return_value=self.current_time)
global_patcher.start()
self.patches.append(global_patcher)
# Patch time.monotonic as well for timing operations
monotonic_patcher = patch("time.monotonic", return_value=self.current_time)
monotonic_patcher.start()
self.patches.append(monotonic_patcher)
# Patch module-specific time imports for provide.foundation modules
module_patches = [
"provide.foundation.state._internal.transitions.time.time",
"provide.foundation.state._internal.transitions.time.monotonic",
"provide.foundation.resilience.retry.time.time",
"provide.foundation.resilience.retry.time.monotonic",
"provide.foundation.resilience.circuit.time.time",
"provide.foundation.resilience.circuit.time.monotonic",
"provide.foundation.utils.rate_limiting.time.time",
"provide.foundation.utils.rate_limiting.time.monotonic",
"provide.foundation.utils.timing.time.time",
"provide.foundation.utils.timing.time.monotonic",
"provide.foundation.transport.middleware.time.time",
"provide.foundation.transport.middleware.time.monotonic",
"provide.foundation.tracer.spans.time.time",
"provide.foundation.tracer.spans.time.monotonic",
]
for module_path in module_patches:
try:
patcher = patch(module_path, return_value=self.current_time)
patcher.start()
self.patches.append(patcher)
except (ImportError, AttributeError):
# Module might not be imported yet or doesn't exist
pass
return self
|
Unfreeze time.
Source code in provide/testkit/time/classes.py
| def unfreeze(self) -> None:
"""Unfreeze time."""
self.is_frozen = False
self._stop_all_patches()
|
jump(seconds: float) -> None
Jump forward or backward in time.
Source code in provide/testkit/time/classes.py
| def jump(self, seconds: float) -> None:
"""Jump forward or backward in time."""
self.current_time += seconds
if self.is_frozen:
# Stop all patches and restart them with the new time
self.unfreeze()
self.freeze(self.current_time)
|
speed_up(factor: float) -> None
Speed up time by a factor.
Source code in provide/testkit/time/classes.py
| def speed_up(self, factor: float) -> None:
"""Speed up time by a factor."""
self.speed_multiplier = factor
|
slow_down(factor: float) -> None
Slow down time by a factor.
Source code in provide/testkit/time/classes.py
| def slow_down(self, factor: float) -> None:
"""Slow down time by a factor."""
self.speed_multiplier = 1.0 / factor
|
Clean up all patches and reset state.
Source code in provide/testkit/time/classes.py
| def cleanup(self) -> None:
"""Clean up all patches and reset state."""
self.is_frozen = False
self._stop_all_patches()
# Unregister from global registry
_active_time_machines.discard(self) # discard() won't raise if not in set
|
Timer for measuring execution time.
Initialize timer.
Source code in provide/testkit/time/classes.py
| def __init__(self) -> None:
"""Initialize timer."""
self.start_time: float | None = None
self.end_time: float | None = None
self.durations: list[float] = []
|
Attributes
property
Get elapsed time since start.
property
Get average duration from all measurements.
Functions
Start the timer.
Source code in provide/testkit/time/classes.py
| def start(self) -> Timer:
"""Start the timer."""
self.start_time = time.perf_counter()
return self
|
Stop the timer and return duration.
Source code in provide/testkit/time/classes.py
| def stop(self) -> float:
"""Stop the timer and return duration."""
self.end_time = time.perf_counter()
if self.start_time is None:
raise RuntimeError("Timer not started")
duration = self.end_time - self.start_time
self.durations.append(duration)
return duration
|
Reset the timer.
Source code in provide/testkit/time/classes.py
| def reset(self) -> None:
"""Reset the timer."""
self.start_time = None
self.end_time = None
self.durations.clear()
|
Functions
advance_time(mock_time: Mock, seconds: float) -> None
Advance a mocked time by specified seconds.
Parameters:
| Name |
Type |
Description |
Default |
mock_time
|
Mock
|
|
required
|
seconds
|
float
|
Number of seconds to advance
|
required
|
Example
from unittest.mock import Mock, patch
with patch("time.time") as mock_time:
... mock_time.return_value = 100.0
... advance_time(mock_time, 50.0)
... assert mock_time.return_value == 150.0
Source code in provide/testkit/time/controlled.py
| def advance_time(mock_time: Mock, seconds: float) -> None:
"""Advance a mocked time by specified seconds.
Args:
mock_time: The mock time object
seconds: Number of seconds to advance
Example:
>>> from unittest.mock import Mock, patch
>>> with patch("time.time") as mock_time:
... mock_time.return_value = 100.0
... advance_time(mock_time, 50.0)
... assert mock_time.return_value == 150.0
"""
if hasattr(mock_time, "return_value"):
mock_time.return_value += seconds
|
benchmark_timer() -> BenchmarkTimer
Timer specifically for benchmarking code.
Returns:
Example
def test_with_benchmark(benchmark_timer):
... result, duration = benchmark_timer.measure(my_function, arg1, arg2)
... benchmark_timer.assert_faster_than(0.1) # Assert < 100ms
Source code in provide/testkit/time/measurement.py
| @pytest.fixture
def benchmark_timer() -> BenchmarkTimer:
"""Timer specifically for benchmarking code.
Returns:
Benchmark timer with statistics.
Example:
>>> def test_with_benchmark(benchmark_timer):
... result, duration = benchmark_timer.measure(my_function, arg1, arg2)
... benchmark_timer.assert_faster_than(0.1) # Assert < 100ms
"""
return BenchmarkTimer()
|
freeze_time() -> (
Callable[[datetime.datetime | None], FrozenTime]
)
Fixture to freeze time at a specific point.
Returns:
Example
def test_with_frozen_time(freeze_time):
... with freeze_time(datetime.datetime(2024, 1, 1)) as frozen:
... # Time is frozen at 2024-01-01
... frozen.tick(seconds=60) # Advance by 60 seconds
Source code in provide/testkit/time/freezing.py
| @pytest.fixture
def freeze_time() -> Callable[[datetime.datetime | None], FrozenTime]:
"""Fixture to freeze time at a specific point.
Returns:
Function that freezes time and returns a context manager.
Example:
>>> def test_with_frozen_time(freeze_time):
... with freeze_time(datetime.datetime(2024, 1, 1)) as frozen:
... # Time is frozen at 2024-01-01
... frozen.tick(seconds=60) # Advance by 60 seconds
"""
def _freeze(at: datetime.datetime | None = None) -> FrozenTime:
"""Freeze time at a specific point.
Args:
at: Optional datetime to freeze at (defaults to now)
Returns:
FrozenTime context manager
"""
return FrozenTime(at)
return _freeze
|
get_active_time_machines() -> set[Any]
Get set of currently active TimeMachine instances.
Returns:
| Type |
Description |
set[Any]
|
Set of TimeMachine instances that are currently active.
|
Note
Thread-safe within a process. pytest-xdist workers are separate processes,
so no cross-process synchronization needed.
Source code in provide/testkit/time/classes.py
| def get_active_time_machines() -> set[Any]:
"""Get set of currently active TimeMachine instances.
Returns:
Set of TimeMachine instances that are currently active.
Note:
Thread-safe within a process. pytest-xdist workers are separate processes,
so no cross-process synchronization needed.
"""
return _active_time_machines.copy() # Return copy to prevent external modification
|
make_controlled_time() -> tuple[
Callable[[], float],
Callable[[float], None],
Callable[[float], None],
Callable[[float], Awaitable[None]],
]
Create controlled time source and sleep functions for testing.
This provides injectable time/sleep functions that don't rely on global mocking,
making tests faster and more reliable. Use these instead of time_machine.freeze()
for retry/circuit breaker tests.
Returns:
| Type |
Description |
tuple[Callable[[], float], Callable[[float], None], Callable[[float], None], Callable[[float], Awaitable[None]]]
|
Tuple of (get_time, advance_time, fake_sleep, fake_async_sleep)
|
Example
get_time, advance_time, fake_sleep, fake_async_sleep = make_controlled_time()
executor = RetryExecutor(
... policy,
... time_source=get_time,
... sleep_func=fake_sleep,
... async_sleep_func=fake_async_sleep,
... )
In tests:
advance_time(5.0) # Simulate 5 seconds passing
assert get_time() == 5.0
Source code in provide/testkit/time/controlled.py
| def make_controlled_time() -> tuple[
Callable[[], float],
Callable[[float], None],
Callable[[float], None],
Callable[[float], Awaitable[None]],
]:
"""Create controlled time source and sleep functions for testing.
This provides injectable time/sleep functions that don't rely on global mocking,
making tests faster and more reliable. Use these instead of time_machine.freeze()
for retry/circuit breaker tests.
Returns:
Tuple of (get_time, advance_time, fake_sleep, fake_async_sleep)
Example:
>>> get_time, advance_time, fake_sleep, fake_async_sleep = make_controlled_time()
>>> executor = RetryExecutor(
... policy,
... time_source=get_time,
... sleep_func=fake_sleep,
... async_sleep_func=fake_async_sleep,
... )
>>> # In tests:
>>> advance_time(5.0) # Simulate 5 seconds passing
>>> assert get_time() == 5.0
"""
current_time = [0.0]
def get_time() -> float:
"""Get current test time."""
return current_time[0]
def advance_time(seconds: float) -> None:
"""Advance test time by seconds."""
current_time[0] += seconds
def fake_sleep(seconds: float) -> None:
"""Fake sleep that advances time instead of blocking."""
advance_time(seconds)
async def fake_async_sleep(seconds: float) -> None:
"""Fake async sleep that advances time instead of blocking."""
advance_time(seconds)
return get_time, advance_time, fake_sleep, fake_async_sleep
|
mock_datetime() -> Generator[Mock, None, None]
Mock datetime module for testing.
Returns:
| Type |
Description |
None
|
Mock datetime module with common methods mocked.
|
Example
def test_with_mock_datetime(mock_datetime):
... now = datetime.datetime.now()
... assert now == datetime.datetime(2024, 1, 1, 12, 0, 0)
Source code in provide/testkit/time/mocking.py
| @pytest.fixture
def mock_datetime() -> Generator[Mock, None, None]:
"""Mock datetime module for testing.
Returns:
Mock datetime module with common methods mocked.
Example:
>>> def test_with_mock_datetime(mock_datetime):
... now = datetime.datetime.now()
... assert now == datetime.datetime(2024, 1, 1, 12, 0, 0)
"""
with patch("datetime.datetime") as mock_dt:
# Set up a fake "now"
fake_now = datetime.datetime(2024, 1, 1, 12, 0, 0)
mock_dt.now.return_value = fake_now
mock_dt.utcnow.return_value = fake_now
mock_dt.today.return_value = fake_now.date()
# Allow normal datetime construction
mock_dt.side_effect = lambda *args, **kwargs: datetime.datetime(*args, **kwargs)
yield mock_dt
|
mock_sleep() -> Generator[Mock, None, None]
Mock time.sleep to speed up tests.
Returns:
| Type |
Description |
None
|
Mock object that replaces time.sleep.
|
Example
def test_with_mock_sleep(mock_sleep):
... time.sleep(10) # Returns instantly
... assert mock_sleep.called
Source code in provide/testkit/time/mocking.py
| @pytest.fixture
def mock_sleep() -> Generator[Mock, None, None]:
"""Mock time.sleep to speed up tests.
Returns:
Mock object that replaces time.sleep.
Example:
>>> def test_with_mock_sleep(mock_sleep):
... time.sleep(10) # Returns instantly
... assert mock_sleep.called
"""
with patch("time.sleep") as mock:
# Make sleep instant by default
mock.return_value = None
yield mock
|
mock_sleep_with_callback() -> (
Callable[[Callable[[float], None] | None], Mock]
)
Mock time.sleep with a callback for each sleep call.
Returns:
| Type |
Description |
Callable[[Callable[[float], None] | None], Mock]
|
Function to set up sleep mock with callback.
|
Example
def test_with_callback(mock_sleep_with_callback):
... total_sleep = []
... sleep_mock = mock_sleep_with_callback(lambda s: total_sleep.append(s))
... with patch("time.sleep", sleep_mock):
... time.sleep(1.5)
... assert total_sleep == [1.5]
Source code in provide/testkit/time/mocking.py
| @pytest.fixture
def mock_sleep_with_callback() -> Callable[[Callable[[float], None] | None], Mock]:
"""Mock time.sleep with a callback for each sleep call.
Returns:
Function to set up sleep mock with callback.
Example:
>>> def test_with_callback(mock_sleep_with_callback):
... total_sleep = []
... sleep_mock = mock_sleep_with_callback(lambda s: total_sleep.append(s))
... with patch("time.sleep", sleep_mock):
... time.sleep(1.5)
... assert total_sleep == [1.5]
"""
def _mock_sleep(callback: Callable[[float], None] | None = None) -> Mock:
"""Create a mock sleep with optional callback.
Args:
callback: Function called with sleep duration
Returns:
Mock sleep object
"""
def sleep_side_effect(seconds: float) -> None:
if callback:
callback(seconds)
return None
mock = Mock(side_effect=sleep_side_effect)
return mock
return _mock_sleep
|
rate_limiter_mock() -> MockRateLimiter
Mock for testing rate-limited code.
Returns:
| Type |
Description |
MockRateLimiter
|
Mock rate limiter that can be controlled in tests.
|
Example
def test_rate_limiting(rate_limiter_mock):
... rate_limiter_mock.set_limit(after_calls=3)
... assert rate_limiter_mock.check() is True # Call 1
... assert rate_limiter_mock.check() is True # Call 2
... assert rate_limiter_mock.check() is True # Call 3
... assert rate_limiter_mock.check() is False # Rate limited!
Source code in provide/testkit/time/rate_limiting.py
| @pytest.fixture
def rate_limiter_mock() -> MockRateLimiter:
"""Mock for testing rate-limited code.
Returns:
Mock rate limiter that can be controlled in tests.
Example:
>>> def test_rate_limiting(rate_limiter_mock):
... rate_limiter_mock.set_limit(after_calls=3)
... assert rate_limiter_mock.check() is True # Call 1
... assert rate_limiter_mock.check() is True # Call 2
... assert rate_limiter_mock.check() is True # Call 3
... assert rate_limiter_mock.check() is False # Rate limited!
"""
return MockRateLimiter()
|
time_machine(request: FixtureRequest) -> TimeMachine
Advanced time manipulation fixture.
Yields:
| Type |
Description |
TimeMachine
|
TimeMachine instance for time manipulation.
|
IMPORTANT: Uses request.addfinalizer() to ensure patches are stopped
BEFORE pytest-asyncio creates event loops for the next test. Also forcibly
closes ALL event loops to prevent cached frozen time.monotonic references.
Example
def test_with_time_machine(time_machine):
... time_machine.freeze(at=time.time())
... # Perform tests with frozen time
... time_machine.jump(seconds=60) # Jump forward
... time_machine.unfreeze()
Source code in provide/testkit/time/freezing.py
| @pytest.fixture
def time_machine(request: pytest.FixtureRequest) -> TimeMachine:
"""Advanced time manipulation fixture.
Yields:
TimeMachine instance for time manipulation.
IMPORTANT: Uses request.addfinalizer() to ensure patches are stopped
BEFORE pytest-asyncio creates event loops for the next test. Also forcibly
closes ALL event loops to prevent cached frozen time.monotonic references.
Example:
>>> def test_with_time_machine(time_machine):
... time_machine.freeze(at=time.time())
... # Perform tests with frozen time
... time_machine.jump(seconds=60) # Jump forward
... time_machine.unfreeze()
"""
machine = TimeMachine()
# Register cleanup with highest priority (runs before standard teardown)
# This ensures time patches are stopped before pytest-asyncio creates
# event loops for the next test
def cleanup_patches() -> None:
machine.cleanup()
# NUCLEAR OPTION: Close ALL event loops to force fresh creation
# This is necessary because event loops cache time.monotonic references
# at creation time, and those cached values persist even after patches stop
try:
import asyncio
with suppress(RuntimeError):
loop = asyncio.get_event_loop()
if not loop.is_running() and not loop.is_closed():
loop.close()
# Also close the running loop if there is one
with suppress(RuntimeError):
asyncio.get_running_loop()
# Can't close running loop, but we can stop it
except Exception:
pass
request.addfinalizer(cleanup_patches)
yield machine
# Also call cleanup here as backup (defensive)
machine.cleanup()
|
time_travel() -> (
Generator[
Callable[[datetime.datetime], None], None, None
]
)
Fixture for traveling through time in tests.
Returns:
| Type |
Description |
None
|
Function to travel to specific time points.
|
Example
def test_with_time_travel(time_travel):
... time_travel(datetime.datetime(2025, 1, 1))
... # time.time() now returns the timestamp for 2025-01-01
Source code in provide/testkit/time/mocking.py
| @pytest.fixture
def time_travel() -> Generator[Callable[[datetime.datetime], None], None, None]:
"""Fixture for traveling through time in tests.
Returns:
Function to travel to specific time points.
Example:
>>> def test_with_time_travel(time_travel):
... time_travel(datetime.datetime(2025, 1, 1))
... # time.time() now returns the timestamp for 2025-01-01
"""
original_time = time.time
current_offset = 0.0
def mock_time() -> float:
return original_time() + current_offset
def _travel_to(target: datetime.datetime) -> None:
"""Travel to a specific point in time.
Args:
target: The datetime to travel to
"""
nonlocal current_offset
current_offset = target.timestamp() - original_time()
with patch("time.time", mock_time):
yield _travel_to
|
Timer fixture for measuring execution time.
Returns:
| Type |
Description |
Timer
|
Timer instance for measuring durations.
|
Example
def test_with_timer(timer):
... with timer:
... # Code to time
... pass
... print(f"Elapsed: {timer.elapsed}s")
Source code in provide/testkit/time/measurement.py
| @pytest.fixture
def timer() -> Timer:
"""Timer fixture for measuring execution time.
Returns:
Timer instance for measuring durations.
Example:
>>> def test_with_timer(timer):
... with timer:
... # Code to time
... pass
... print(f"Elapsed: {timer.elapsed}s")
"""
return Timer()
|