Skip to content

Controlled

provide.testkit.time.controlled

Controlled time utilities for testing.

Injectable time sources and sleep functions that don't rely on global mocking.

Functions

advance_time 

advance_time(mock_time: Mock, seconds: float) -> None

Advance a mocked time by specified seconds.

Parameters:

Name Type Description Default
mock_time Mock

The mock time object

 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 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
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

make_controlled_time 

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 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
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