"""Self-running tests. No pytest: `python3 test_mini_asyncio.py`.

These pin the exact numbers the commentary quotes, so a change to the loop
that moves them fails loudly instead of quietly invalidating the page.
"""

import math

from mini_asyncio import (Loop, Task, Future, burn, sleep, deliveries,
                          periods)

PASSED = []


def test(fn):
    PASSED.append(fn)
    return fn


def heartbeat(period):
    while True:
        yield sleep(period)


def hog(chunk):
    while True:
        yield burn(chunk)


def run(hops=2, chunk=200, period=1000, n_hogs=1, until=8000):
    loop = Loop(hops=hops)
    loop.spawn("beat", heartbeat(period))
    for i in range(n_hogs):
        loop.spawn(f"hog{i}", hog(chunk))
    loop.run(until)
    return loop


# --------------------------------------------------------------- the clock

@test
def test_clock_starts_at_zero_and_is_an_integer():
    loop = Loop()
    assert loop.clock == 0
    assert isinstance(loop.clock, int)


@test
def test_a_lone_sleeper_is_never_late():
    loop = Loop()
    loop.spawn("beat", heartbeat(1000))
    loop.run(8000)
    assert periods(loop, "beat") == [1000] * 6
    for w in deliveries(loop, "beat"):
        assert w.ran_at == w.deadline, w


@test
def test_idle_jump_is_the_only_free_clock_movement():
    """With no work to do the loop lands exactly on the deadline, never past
    it -- so the clock cannot drift forward on its own."""
    loop = Loop()
    loop.spawn("beat", heartbeat(3500))
    loop.run(8000)
    assert [w.ran_at for w in deliveries(loop, "beat")] == [3500, 7000]


@test
def test_burn_is_the_only_other_clock_movement():
    def once():
        yield burn(37)
        yield burn(5)
    loop = Loop()
    loop.spawn("t", once())
    loop.run(1000)
    assert loop.clock == 42


# --------------------------------------------------------- the headline aha

@test
def test_headline_period_is_1400_not_1000():
    assert periods(run(), "beat") == [1400] * 4


@test
def test_the_timer_itself_is_never_late():
    """The load-bearing observation: every timer expired on the exact
    millisecond it was due. The lateness is all downstream of the heap."""
    for w in deliveries(run(), "beat"):
        assert w.expired_at == w.deadline, w
        assert w.fired_at - w.deadline == 200, w
        assert w.ran_at - w.deadline == 400, w


@test
def test_nothing_raises_and_no_delivery_is_dropped():
    loop = run(until=20000)
    assert len(deliveries(loop, "beat")) == 14
    assert all(w.ran_at > w.deadline for w in deliveries(loop, "beat"))


# ------------------------------------------------------------- the two hops

@test
def test_one_hop_is_1200_and_two_hops_is_1400():
    assert periods(run(hops=1), "beat") == [1200] * 5
    assert periods(run(hops=2), "beat") == [1400] * 4


@test
def test_the_hop_difference_is_exactly_one_chunk():
    for chunk in (50, 100, 200, 300, 400, 700):
        one = periods(run(hops=1, chunk=chunk, until=40000), "beat")[0]
        two = periods(run(hops=2, chunk=chunk, until=40000), "beat")[0]
        assert two - one == chunk, (chunk, one, two)


@test
def test_hop_two_resolves_a_future_and_hop_one_does_not():
    """hops=1 delivers inside the timer callback; hops=2 needs another pass,
    so the callback and the resumption land at different clock readings."""
    for w in deliveries(run(hops=1), "beat"):
        assert w.fired_at == w.ran_at, w
    for w in deliveries(run(hops=2), "beat"):
        assert w.ran_at - w.fired_at == 200, w


@test
def test_future_set_result_schedules_rather_than_calls():
    loop = Loop()
    fut = Future(loop)
    calls = []
    fut.waiter = lambda: calls.append(loop.clock)
    fut.set_result()
    assert fut.done is True
    assert calls == []                    # NOT resumed
    assert len(loop.ready) == 1           # ...only queued
    loop.run_once()
    assert calls == [0]


# ------------------------------------------------------------ the closed form

@test
def test_closed_form_predicts_every_combination():
    """observed = (ceil(P/R) + hops) * R, with R = n_hogs * chunk."""
    checked = 0
    for period in (900, 1000, 1100, 1250, 2000):
        for chunk in (100, 200, 250, 300, 400, 700):
            for hops in (1, 2):
                for n_hogs in (1, 2, 3):
                    R = n_hogs * chunk
                    predicted = (math.ceil(period / R) + hops) * R
                    observed = set(periods(
                        run(hops=hops, chunk=chunk, period=period,
                            n_hogs=n_hogs, until=40000), "beat"))
                    assert observed == {predicted}, (period, chunk, hops,
                                                     n_hogs, observed,
                                                     predicted)
                    checked += 1
    assert checked == 180


@test
def test_lateness_floor_is_hops_times_R():
    for chunk, expected in ((200, 1400), (100, 1200), (50, 1100),
                            (10, 1020), (1, 1002)):
        obs = periods(run(chunk=chunk, until=20000), "beat")[0]
        assert obs == expected, (chunk, obs)
        assert obs - 1000 == 2 * chunk


# ------------------------------------------- the snapshot, and what it saves

@test
def test_snapshot_bounds_a_pass_to_the_queue_it_started_with():
    """run_once must not run work that this pass created. That single
    property is what keeps a self-rescheduling task from starving timers."""
    loop = Loop()
    seen = []

    def a():
        seen.append("a")
        loop.call_soon(b)

    def b():
        seen.append("b")

    loop.call_soon(a)
    loop.run_once()
    assert seen == ["a"]        # b was scheduled by this pass, so it waits
    loop.run_once()
    assert seen == ["a", "b"]


@test
def test_timers_are_queued_behind_existing_ready_work():
    loop = Loop()
    order = []
    loop.call_soon(lambda: order.append("already ready"))
    loop.call_later(0, lambda: order.append("timer"))
    loop.run_once()
    assert order == ["already ready", "timer"]


@test
def test_equal_deadlines_stay_fifo():
    loop = Loop()
    order = []
    for i in range(4):
        loop.call_later(500, lambda i=i: order.append(i))
    loop.clock = 500
    loop.run_once()
    assert order == [0, 1, 2, 3]


# ----------------------------------------------------------- determinism

@test
def test_runs_are_bit_identical():
    a = [tuple(w) for w in run(until=20000).trace]
    b = [tuple(w) for w in run(until=20000).trace]
    assert a == b
    assert all(isinstance(x, int)
               for w in a for x in w[1:])       # no floats anywhere


@test
def test_spawn_order_does_not_change_the_steady_state():
    loop = Loop()
    loop.spawn("hog0", hog(200))
    loop.spawn("beat", heartbeat(1000))
    loop.run(20000)
    assert set(periods(loop, "beat")) == {1400}


if __name__ == "__main__":
    for fn in PASSED:
        fn()
        print(f"ok  {fn.__name__}")
    print(f"\n{len(PASSED)} tests passed")