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

These pin the exact numbers the commentary quotes. Two of them
(`test_real_asyncio_*`, `test_real_threads_*`) run against real asyncio and
real OS threads rather than against the toy, because the page's sharpest
claim is a comparison with the real things.
"""
import asyncio
import inspect
import threading

import runtimes as rt
from runtimes import (DEPOSITS, START, TELLERS, Green, Stackful, chain,
                      run_chain, run_green, run_stackful)

EXPECT = START + TELLERS * DEPOSITS          # 200
LOCKSTEP = START + DEPOSITS                  # 150

PASSED = []


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


# ------------------------------------------------ the headline: 200 and 150

@test
def test_library_v1_is_correct_in_both_runtimes():
    assert run_stackful("none")[0] == EXPECT == 200
    assert run_green(rt.g1_teller)[0] == EXPECT


@test
def test_one_added_line_makes_the_stackful_run_return_150():
    balance, switches = run_stackful("after_read")
    assert balance == LOCKSTEP == 150
    assert switches == 102              # 100 suspensions + 2 task exits


@test
def test_the_same_line_makes_the_stackless_run_refuse():
    try:
        run_green(rt.g2_teller)
    except TypeError as e:
        assert str(e) == ("unsupported operand type(s) for +: "
                          "'generator' and 'int'")
    else:
        raise AssertionError("expected the uncoloured caller to blow up")


@test
def test_colouring_all_four_frames_makes_it_run_and_it_returns_150():
    balance, switches = run_green(rt.g3_teller)
    assert balance == LOCKSTEP == 150
    assert switches == 100


@test
def test_colouring_is_inert():
    """Four frames edited, and the two columns still agree exactly."""
    assert run_stackful("after_read")[0] == run_green(rt.g3_teller)[0] == 150


# -------------------------------------------------------- the arithmetic

@test
def test_a_round_advances_the_balance_by_one_whatever_the_teller_count():
    for tellers in (2, 3, 4, 8):
        for deposits in (10, 50):
            got = run_stackful("after_read", tellers, deposits)[0]
            assert got == START + deposits, (tellers, deposits, got)


@test
def test_exactly_half_the_increments_are_lost_at_two_tellers():
    performed = TELLERS * DEPOSITS
    applied = run_stackful("after_read")[0] - START
    assert performed == 100 and applied == 50
    assert performed - applied == 50


# --------------------------------------------------- the load-bearing line

@test
def test_moving_the_suspension_one_line_up_restores_200():
    assert run_stackful("before_read")[0] == 200
    assert run_stackful("after_read")[0] == 150


@test
def test_no_suspension_inside_the_critical_section_is_correct():
    assert run_stackful("none")[0] == 200


# --------------------------------------------------------------- boundaries

@test
def test_one_teller_loses_nothing():
    got = run_stackful("after_read", 1, DEPOSITS)[0]
    assert got == START + DEPOSITS == 150       # its own expected value
    assert (START + 1 * DEPOSITS) - got == 0


@test
def test_colouring_costs_exactly_D_edits_at_depth_D():
    for d in range(5):
        out, edits, err = run_chain(d, "coloured")
        assert edits == d and err is None and out == [8], (d, edits, err)


@test
def test_a_stackful_runtime_costs_zero_edits_at_every_depth():
    for d in range(5):
        out, edits, err = run_chain(d, "stackful")
        assert edits == 0 and err is None and out == [8], (d, edits, err)
        assert "yield" not in chain(d, "stackful")[0]


@test
def test_an_uncoloured_caller_breaks_at_every_depth_above_zero():
    for d in range(1, 5):
        out, edits, err = run_chain(d, "uncoloured")
        assert err is not None and err.startswith("TypeError"), (d, err)
        assert out == []


@test
def test_depth_zero_makes_the_two_runtimes_indistinguishable():
    """The boundary. Nothing is coloured, nothing is forced to change, and
    the uncoloured stackless column runs as happily as the stackful one."""
    for style in ("stackful", "coloured", "uncoloured"):
        out, edits, err = run_chain(0, style)
        assert edits == 0 and err is None and out == [8], (style, err)
    assert "yield from" not in chain(0, "coloured")[0]


@test
def test_the_bank_workload_is_depth_four():
    """The four frames the commentary names, counted out of the source."""
    frames = ("db_fetch", "repo_get", "deposit", "teller")
    v2 = [inspect.getsource(getattr(rt, f"g2_{f}")) for f in frames]
    v3 = [inspect.getsource(getattr(rt, f"g3_{f}")) for f in frames]
    assert sum("yield from" in s for s in v2) == 0
    assert sum("yield from" in s for s in v3) == 4
    assert sum(a != b for a, b in zip(v2, v3)) == 4
    # ...and the stackful column has no marker on any of them.
    sf = [inspect.getsource(getattr(rt, f"sf_{f}")) for f in frames]
    assert sum("yield" in s or "suspend" in s for s in sf) == 0


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

@test
def test_the_stackful_race_is_the_same_value_on_every_run():
    assert {run_stackful("after_read")[0] for _ in range(20)} == {150}


@test
def test_the_stackless_race_is_the_same_value_on_every_run():
    assert {run_green(rt.g3_teller)[0] for _ in range(20)} == {150}


# -------------------------------------- against real asyncio and real threads

def _coop_trial():
    acct = {"bal": START}

    async def io_read():
        v = acct["bal"]
        await asyncio.sleep(0)
        return v

    async def teller():
        for _ in range(DEPOSITS):
            acct["bal"] = await io_read() + 1

    async def main():
        await asyncio.gather(*[teller() for _ in range(TELLERS)])

    asyncio.run(main())
    return acct["bal"]


def _preempt_trial():
    acct = {"bal": START}

    def teller():
        for _ in range(DEPOSITS):
            b = acct["bal"]
            acct["bal"] = b + 1

    ts = [threading.Thread(target=teller) for _ in range(TELLERS)]
    for t in ts:
        t.start()
    for t in ts:
        t.join()
    return acct["bal"]


@test
def test_real_asyncio_is_wrong_on_20_of_20_trials():
    """Cooperative: the suspension point is a fixed source location, so the
    lost update is not a probability. It is the program's only behaviour."""
    trials = [_coop_trial() for _ in range(20)]
    assert set(trials) == {150}
    assert sum(t != EXPECT for t in trials) == 20


@test
def test_real_threads_are_wrong_on_0_of_20_trials():
    """Preemptive: the switch has to land inside a two-bytecode window.

    This is the one assertion on the page that is a probability rather than a
    certainty -- which is exactly the lesson. If it ever fails, the page's
    claim ("0/20 on this machine") is still the honest report of the run it
    came from.
    """
    trials = [_preempt_trial() for _ in range(20)]
    assert sum(t != EXPECT for t in trials) == 0


# ------------------------------------------ the limit of the safety net

@test
def test_an_uncoloured_generator_only_blows_up_when_it_is_consumed():
    def io_read():
        yield
        return {"bal": 7}

    log = []
    log.append(io_read())                       # forwarded: silent
    assert len(log) == 1 and not log[0].gi_running
    assert bool(io_read()) is True              # truth-tested: always True
    try:
        io_read()["bal"]                        # consumed: caught
    except TypeError:
        pass
    else:
        raise AssertionError("expected the consuming caller to blow up")


@test
def test_bool_of_an_unawaited_coroutine_is_unconditionally_true():
    async def user_exists():
        return False                            # the answer is False!

    c = user_exists()
    assert bool(c) is True
    c.close()


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