TestClock

__init__(start=0.0, *, warning_delay=1.0, live=None) -> None

def __init__(
    self,
    start: float = 0.0,
    *,
    warning_delay: Duration = 1.0,
    live: ClockService | None = None,
) -> None:
    if not math.isfinite(start):
        raise ValueError(f"TestClock needs a finite start time, got {start}")
    start_ns = to_nanos(start)
    self._timestamp_ns: float = start_ns
    self._wall_ns = start_ns
    self._monotonic_ns = start_ns
    self._sleepers: list[tuple[float, int, _Sleeper]] = []
    self._seq = itertools.count()
    self._warning_delay = to_seconds(warning_delay)
    self._live = live or LiveClock()
    self._not_advanced = _NotAdvanced()
    self._moving = Semaphore(1)

layer(*, warning_delay=1.0) -> Layer[Never]

@staticmethod
def layer(*, warning_delay: Duration = 1.0) -> Layer[Never]:
    """Provide a fresh ``TestClock`` as the ``Clock``. Its ``with_live``
    uses the clock that was current when the layer was built."""

    def make(live: ClockService) -> Effect[TestClock, Never, Scope]:
        base = live._live if isinstance(live, TestClock) else live
        clock = TestClock(warning_delay=warning_delay, live=base)
        return add_finalizer(lambda _: _Sync(clock.stop_warning)).as_(clock)

    return layer_effect(Clock, Clock().flat_map(make))

unsafe_now() -> float

def unsafe_now(self) -> float:
    # As Effect's ``currentTimeMillis``: infinite once time is moved to
    # infinity (the nanosecond clocks stay at the last finite time).
    if math.isinf(self._timestamp_ns):
        return self._timestamp_ns
    return self._wall_ns / _NANOS

unsafe_monotonic() -> float

def unsafe_monotonic(self) -> float:
    return self._monotonic_ns / _NANOS

unsafe_now_nanos() -> int

def unsafe_now_nanos(self) -> int:
    return self._wall_ns

unsafe_monotonic_nanos() -> int

def unsafe_monotonic_nanos(self) -> int:
    return self._monotonic_ns

sleep(seconds) -> Effect[None]

def sleep(self, seconds: Seconds) -> Effect[None]:
    def body(fiber: Fiber[Any, Any]) -> Effect[None]:
        def register(resume: Callable[[Effect[None]], None]) -> Effect[None] | None:
            end = self._timestamp_ns + _nanos(seconds)
            if end <= self._timestamp_ns:
                resume(_Succeed(None))
                return None
            self._not_advanced.arm(fiber._scheduler, self._warning_delay)
            sleeper = _Sleeper(lambda: resume(_Succeed(None)))
            heapq.heappush(self._sleepers, (end, next(self._seq), sleeper))

            def cancel() -> None:
                sleeper.alive = False

            return _Sync(cancel)

        return _Callback(register)

    return _WithFiber(body)

adjust(duration) -> Effect[None]

def adjust(self, duration: Duration) -> Effect[None]:
    """Move time forward by ``duration``, waking sleepers in order.
    After each wake-up, wait for the woken fiber and the work it starts
    to suspend."""
    delta = _nanos(to_seconds(duration))
    return self._move(lambda now: now + delta)

set_time(at) -> Effect[None]

def set_time(self, at: float) -> Effect[None]:
    """Set wall time to ``at`` (seconds since the epoch), waking every
    sleeper due by then. Moving backwards leaves monotonic time alone."""
    target = _nanos(at)
    return self._move(lambda _: target)

with_live(eff) -> Effect[A, E, R]

def with_live[A, E, R](self, eff: Effect[A, E, R]) -> Effect[A, E, R]:
    """Run ``eff`` with the live clock instead of this one."""
    require_effect(eff, "eff")
    return eff.pipe(provide_service(Clock, self._live))

stop_warning() -> None

def stop_warning(self) -> None:
    """Never warn that time is not advancing; runners call it when the
    test ends."""
    self._not_advanced.disarm()