Classes

provide.testkit.time.classes

Time testing classes for fixtures.

Core classes used by time testing fixtures. Extracted from fixtures.py for better organization and discoverability.

Classes

BenchmarkTimer

BenchmarkTimer()

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

avg_time property

avg_time: float

Get average execution time.

max_time property

max_time: float

Get maximum execution time.

min_time property

min_time: float

Get minimum execution time.

Functions

assert_faster_than

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")

measure

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]	Function to measure	required
*args	Any	Function arguments	()
**kwargs	Any	Function keyword arguments	{}

Returns:

Type	Description
tuple[Any, float]	Tuple of (result, duration)

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

FrozenTime

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

__enter__

__enter__() -> FrozenTime

Enter frozen time context.

Source code in provide/testkit/time/classes.py

def __enter__(self) -> FrozenTime:
    """Enter frozen time context."""
    # Patch time.time()
    time_patch = patch("time.time", return_value=self.frozen_time.timestamp())
    self.patches.append(time_patch)
    time_patch.start()

    # Patch datetime.datetime.now()
    datetime_patch = patch("datetime.datetime", wraps=datetime.datetime)
    mock_datetime = datetime_patch.start()
    mock_datetime.now.return_value = self.frozen_time
    mock_datetime.utcnow.return_value = self.frozen_time
    self.patches.append(datetime_patch)

    return self

__exit__

__exit__(
    exc_type: type[BaseException] | None,
    exc: BaseException | None,
    tb: TracebackType | None,
) -> None

Exit frozen time context.

Source code in provide/testkit/time/classes.py

def __exit__(
    self,
    exc_type: type[BaseException] | None,
    exc: BaseException | None,
    tb: TracebackType | None,
) -> None:
    """Exit frozen time context."""
    for p in self.patches:
        p.stop()

tick

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()

MockRateLimiter

MockRateLimiter()

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

check() -> bool

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

reset() -> None

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

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

TimeMachine

TimeMachine()

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

cleanup

cleanup() -> None

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

freeze

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

jump

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)

slow_down

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

speed_up

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

unfreeze

unfreeze() -> None

Unfreeze time.

Source code in provide/testkit/time/classes.py

def unfreeze(self) -> None:
    """Unfreeze time."""
    self.is_frozen = False
    self._stop_all_patches()

Timer

Timer()

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

average property

average: float

Get average duration from all measurements.

elapsed property

elapsed: float

Get elapsed time since start.

Functions

__enter__

__enter__() -> Timer

Context manager entry.

Source code in provide/testkit/time/classes.py

def __enter__(self) -> Timer:
    """Context manager entry."""
    self.start()
    return self

__exit__

__exit__(
    exc_type: type[BaseException] | None,
    exc: BaseException | None,
    tb: TracebackType | None,
) -> None

Context manager exit.

Source code in provide/testkit/time/classes.py

def __exit__(
    self,
    exc_type: type[BaseException] | None,
    exc: BaseException | None,
    tb: TracebackType | None,
) -> None:
    """Context manager exit."""
    self.stop()

reset

reset() -> None

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()

start

start() -> Timer

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

stop() -> float

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

Functions

get_active_time_machines

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