Martin Bauer
d44c24ec97
- event bus systen - all components are independent - preparation for web frontend
109 lines
2.5 KiB
Python
109 lines
2.5 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
from musicmouse.clock import FakeClock, RealClock
|
|
|
|
|
|
async def test_fake_clock_starts_at_zero_and_advances() -> None:
|
|
clock = FakeClock()
|
|
assert clock.now() == 0.0
|
|
|
|
await clock.advance(2.5)
|
|
|
|
assert clock.now() == 2.5
|
|
|
|
|
|
async def test_sleep_blocks_until_time_is_advanced_past_the_deadline() -> None:
|
|
clock = FakeClock()
|
|
woken = asyncio.Event()
|
|
|
|
async def sleeper() -> None:
|
|
await clock.sleep(10)
|
|
woken.set()
|
|
|
|
task = asyncio.create_task(sleeper())
|
|
await clock.advance(9)
|
|
assert not woken.is_set()
|
|
|
|
await clock.advance(2)
|
|
assert woken.is_set()
|
|
await task
|
|
|
|
|
|
async def test_sleepers_wake_in_deadline_order_regardless_of_start_order() -> None:
|
|
clock = FakeClock()
|
|
woken: list[str] = []
|
|
|
|
async def sleeper(name: str, delay: float) -> None:
|
|
await clock.sleep(delay)
|
|
woken.append(name)
|
|
|
|
tasks = [
|
|
asyncio.create_task(sleeper("late", 30)),
|
|
asyncio.create_task(sleeper("early", 10)),
|
|
asyncio.create_task(sleeper("middle", 20)),
|
|
]
|
|
await clock.advance(60)
|
|
|
|
assert woken == ["early", "middle", "late"]
|
|
await asyncio.gather(*tasks)
|
|
|
|
|
|
async def test_time_at_wake_up_is_the_deadline_not_the_target() -> None:
|
|
clock = FakeClock()
|
|
observed: list[float] = []
|
|
|
|
async def sleeper() -> None:
|
|
await clock.sleep(5)
|
|
observed.append(clock.now())
|
|
|
|
task = asyncio.create_task(sleeper())
|
|
await clock.advance(100)
|
|
|
|
assert observed == [5.0]
|
|
assert clock.now() == 100.0
|
|
await task
|
|
|
|
|
|
async def test_idle_hook_runs_after_each_wake_up() -> None:
|
|
calls: list[int] = []
|
|
|
|
async def idle() -> None:
|
|
calls.append(1)
|
|
|
|
clock = FakeClock(idle=idle)
|
|
|
|
async def sleeper() -> None:
|
|
await clock.sleep(1)
|
|
|
|
task = asyncio.create_task(sleeper())
|
|
await clock.advance(2)
|
|
|
|
assert calls # the bus got a chance to drain between virtual ticks
|
|
await task
|
|
|
|
|
|
async def test_pending_timers_is_visible() -> None:
|
|
clock = FakeClock()
|
|
task = asyncio.create_task(clock.sleep(5))
|
|
await asyncio.sleep(0)
|
|
|
|
assert clock.pending_timers == 1
|
|
await clock.advance(5)
|
|
assert clock.pending_timers == 0
|
|
await task
|
|
|
|
|
|
async def test_zero_sleep_just_yields() -> None:
|
|
clock = FakeClock()
|
|
await clock.sleep(0)
|
|
assert clock.now() == 0.0
|
|
|
|
|
|
async def test_real_clock_measures_real_elapsed_time() -> None:
|
|
clock = RealClock()
|
|
before = clock.now()
|
|
await clock.advance(0.01)
|
|
assert clock.now() - before >= 0.005
|