looptime package

class looptime.Chronometer(clock: Callable[[], float]=<built-in function perf_counter>)[source][source]

Bases: Numeric

A helper context manager to measure the time of the code-blocks.

Usage:

import time

def test_chronometer():
    with Chronometer() as chronometer:
        time.sleep(1.23)  # do something slow
        print(f"Executing for {chronometer.seconds}s already.")
        time.sleep(2.34)  # do something slow again

    print(f"Executed in {chronometer.seconds}s.")
    assert chronometer.seconds < 5.0  # 3.57s or slightly more
property seconds: float | None[source]

The elapsed time in seconds (fractional).

exception looptime.IdleTimeoutError[source][source]

Bases: TimeoutError

A special kind of timeout when the loop idles with no external I/O.

class looptime.LoopTimeEventLoop(*args: Any, start: float | None | Callable[[], float | None] = None, end: float | None | Callable[[], float | None] = None, resolution: float = 1e-06, idle_step: float | None = None, idle_timeout: float | None = None, noop_cycles: int = 42, **kwargs: Any)[source][source]

Bases: BaseEventLoop

An event loop with time compaction. Either a class or a mixin.

looptime_enabled() Iterator[None][source][source]

A context manager to temporarily enable the time compaction.

property looptime_on: bool[source]

Whether the time compaction is enabled at the moment.

run_in_executor(executor: Any, func: Any, *args: Any) Future[source][source]
setup_looptime(*, start: float | None | Callable[[], float | None] = None, end: float | None | Callable[[], float | None] = None, resolution: float = 1e-06, idle_step: float | None = None, idle_timeout: float | None = None, noop_cycles: int = 42, _enabled: bool | None = None) None[source][source]

Set all the fake-time fields and patch the i/o selector.

This is the same as __init__(...), except that it can be used when the mixin/class is injected into the existing event loop object. In that case, the object is already initialised except for these fields.

time() float[source][source]

Return the time according to the event loop’s clock.

This is a float expressed in seconds since an epoch, but the epoch, precision, accuracy and drift are unspecified and may differ per event loop.

class looptime.LoopTimeProxy(loop: AbstractEventLoop | None = None, *, resolution: float = 1e-09)[source][source]

Bases: Numeric

A numeric-compatible proxy to the time of the current/specific event loop.

It is mainly represented by the looptime fixture in pytest.

exception looptime.LoopTimeoutError[source][source]

Bases: TimeoutError

A special kind of timeout when the loop’s time reaches its end.

exception looptime.TimeWarning[source][source]

Bases: UserWarning

Issued when the loop time moves backwards, violating its monotonicity.

class looptime.enabled(*, strict: bool = False, loop: AbstractEventLoop | None = None)[source][source]

Bases: ContextManager[None, bool | None]

Enable the looptime time compaction temporarily.

If used as a context manager, enables the time compaction for the wrapped code block only:

import asyncio
import looptime

async def main() -> None:
    with looptime.enabled(strict=True):
        await asyncio.sleep(10)

if __name__ == '__main__':
    asuncio.run(main())

If used as a function/fixture decorator, enables the time compaction for the duration of the function/fixture:

import asyncio
import looptime

@looptime.enabled(strict=True)
async def main() -> None:
    await asyncio.sleep(10)

if __name__ == '__main__':
    asuncio.run(main())

In both cases, the event loop must be pre-patched (usually at creation). In strict mode, if the event loop is not patched, the call will fail. In non-strict mode (the default), it will issue a warning and continue with the real time flow (i.e. with no time compaction).

Use it, for example, for fixtures or finalizers of fixtures where the fast time flow is required despite fixtures are normally excluded from the time compaction magic (because it is impossible or difficult to infer which event loop is being used in the multi-scoped setup of pytest-asyncio), and because of the structure of pytest hooks for fixture finalizing (no finalizer hook, only the post-finalizer hook, when it is too late).

Beware of a caveat: if used as a decorator on a yield-based fixture, it will enable the looptime magic for the whole duration of the test, including all its fixtures (even undecorated ones), until the decorated fixture reaches its finalizer. This might have unexpected side effects.

strict: bool[source]
looptime.make_event_loop_class(cls: Type[BaseEventLoop], *, prefix: str = 'Looptime') Type[LoopTimeEventLoop][source][source]

Create a new looptime-enabled event loop class from the original class.

Technically, it is equivalent to creating a new class that inherits from the original class and looptime.LoopTimeEventLoop as a mixin, with no content (methods or fields) of its own:

# Not the actual code, just the idea of what happens under the hood.
class NewEventLoop(loops.LoopTimeEventLoop, cls):
    pass

New classes are cached, so the same original class always produces the same derived class, not a new one on every call.

looptime.new_event_loop(**kwargs: Any) LoopTimeEventLoop[source][source]

Create a new event loop as asyncio.new_event_loop(), but patched.

looptime.patch_event_loop(loop: BaseEventLoop, **kwargs: Any) LoopTimeEventLoop[source][source]

Patch an existing event loop to be looptime-ready.

This operation is idempotent and can be safely called multiple times.

Internally, it takes the existing class of the event loop and replaces it with the new class, which is a mix of the original class and looptime.LoopTimeEventLoop as a mixin. The new classes are cached.

looptime.reset_caches() None[source][source]

Purge all caches populated by the patching function of looptime.

The classes themselves are not destroyed, so if there are event loops that were created before the caches are cleared, they will continue to work.