"""Every number the commentary claims, pinned. Plain asserts, no pytest.

  python3 test_broker.py
"""

from broker import BLOCK, DROP_NEWEST, DROP_OLDEST, Broker, Subscriber, run

TICKS, RATE, CAP, SLOW, FAST = 200, 5, 10, 1, 5

tests = []


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


# ---- the queue mechanics -------------------------------------------------

@test
def offer_accepts_until_capacity():
    s = Subscriber("s", capacity=3)
    assert [s.offer(i, 0) for i in range(3)] == [True, True, True]
    assert s.full
    assert s.offer(99, 0) is False
    assert s.dropped == 1


@test
def drop_newest_keeps_what_it_had():
    s = Subscriber("s", capacity=3, policy=DROP_NEWEST)
    for i in range(3):
        s.offer(i, 0)
    s.offer(99, 0)
    assert [seq for seq, _ in s.queue] == [0, 1, 2]


@test
def drop_oldest_evicts_the_head():
    s = Subscriber("s", capacity=3, policy=DROP_OLDEST)
    for i in range(3):
        s.offer(i, 0)
    s.offer(99, 0)
    assert [seq for seq, _ in s.queue] == [1, 2, 99]


@test
def both_policies_leave_the_queue_at_exactly_capacity():
    """This is why the counts cannot differ: identical occupancy."""
    for policy in (DROP_NEWEST, DROP_OLDEST):
        s = Subscriber("s", capacity=4, policy=policy)
        for i in range(20):
            s.offer(i, 0)
        assert len(s.queue) == 4, (policy, len(s.queue))


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

def slow_of(policy, ticks=TICKS, rate=RATE, cap=CAP, rates=(SLOW, FAST)):
    broker, subs = run(policy, ticks, rate, cap, rates)
    return broker, subs[0]


@test
def the_two_drop_policies_lose_the_identical_791():
    newest_broker, newest = slow_of(DROP_NEWEST)
    oldest_broker, oldest = slow_of(DROP_OLDEST)
    assert newest_broker.published == oldest_broker.published == 1000
    assert newest.dropped == oldest.dropped == 791
    assert len(newest.delivered) == len(oldest.delivered) == 199


def predict(rate, drain, cap, ticks=TICKS):
    return max(0, (rate - drain) * ticks - (cap - drain))


@test
def the_loss_formula_is_exact_wherever_a_tick_burst_fits():
    """Exact across every (rate, capacity, drain) with capacity >= rate.
    The domain matters: with capacity < rate a single tick's burst cannot fit
    even into an empty queue, and the steady-state argument does not apply."""
    checked = 0
    for rate in range(1, 13):
        for cap in (1, 2, 3, 5, 7, 10, 33, 100, 250):
            if cap < rate:
                continue
            for drain in (1, 2, 3, 4, 6):
                predicted = predict(rate, drain, cap)
                for policy in (DROP_NEWEST, DROP_OLDEST):
                    _, slow = slow_of(policy, rate=rate, cap=cap, rates=(drain,))
                    assert slow.dropped == predicted, (policy, rate, cap, drain,
                                                       slow.dropped, predicted)
                checked += 1
    assert checked == 320, checked


@test
def the_policies_agree_even_outside_the_formulas_domain():
    """The equality is the headline and it does not need the formula."""
    checked = 0
    for rate in range(1, 13):
        for cap in (1, 2, 3, 5, 7):
            if cap >= rate:
                continue
            for drain in (1, 2, 3, 4, 6):
                _, n = slow_of(DROP_NEWEST, rate=rate, cap=cap, rates=(drain,))
                _, o = slow_of(DROP_OLDEST, rate=rate, cap=cap, rates=(drain,))
                assert n.dropped == o.dropped, (rate, cap, drain)
                checked += 1
    assert checked == 210, checked


@test
def nothing_is_lost_when_the_subscriber_keeps_up():
    """The boundary: no saturation, no loss, and the policy is inert."""
    for policy in (DROP_NEWEST, DROP_OLDEST):
        _, slow = slow_of(policy, rate=1, rates=(SLOW,))
        assert slow.dropped == 0
        assert len(slow.delivered) == 199  # one still in flight at the end


@test
def the_policy_decides_which_messages_survive():
    _, newest = slow_of(DROP_NEWEST)
    _, oldest = slow_of(DROP_OLDEST)
    n_seqs = [s for s, _, _ in newest.delivered]
    o_seqs = [s for s, _, _ in oldest.delivered]
    assert n_seqs[:8] == [0, 1, 2, 3, 4, 5, 6, 7]
    assert o_seqs[:8] == [0, 1, 5, 10, 15, 20, 25, 30]
    assert n_seqs[-1] == 945 and o_seqs[-1] == 985


@test
def drop_newest_runs_five_times_staler():
    _, newest = slow_of(DROP_NEWEST)
    _, oldest = slow_of(DROP_OLDEST)
    n_ages, o_ages = newest.ages(), oldest.ages()
    assert round(sum(n_ages) / len(n_ages), 1) == 9.7
    assert round(sum(o_ages) / len(o_ages), 1) == 2.0
    assert n_ages[-1] == 10 and o_ages[-1] == 2
    # staleness is bounded by how long the queue takes to drain
    assert max(n_ages) <= CAP / SLOW


# ---- fan-out -------------------------------------------------------------

@test
def the_fast_subscriber_is_untouched_by_the_slow_one():
    for policy in (DROP_NEWEST, DROP_OLDEST):
        _, subs = run(policy, TICKS, RATE, CAP, (SLOW, FAST))
        fast = subs[1]
        assert fast.dropped == 0
        assert len(fast.delivered) == 995


@test
def every_subscriber_gets_its_own_copy_of_a_publish():
    broker = Broker()
    a = broker.subscribe("t", Subscriber("a", capacity=5))
    b = broker.subscribe("t", Subscriber("b", capacity=5))
    broker.publish("t", 0)
    assert [s for s, _ in a.queue] == [s for s, _ in b.queue] == [0]


@test
def a_publish_to_a_topic_with_no_subscribers_is_not_an_error():
    broker = Broker()
    assert broker.publish("nobody-listening", 0) is True
    assert broker.published == 1


# ---- backpressure --------------------------------------------------------

@test
def blocking_refuses_exactly_the_same_791():
    """The deficit is conserved. All three policies fail to move 791
    messages; they differ only in where the loss is taken."""
    broker, subs = run(BLOCK, TICKS, RATE, CAP, (SLOW, FAST))
    assert broker.refused == 791
    assert broker.published == 209
    assert broker.published + broker.refused == RATE * TICKS
    assert subs[0].dropped == 0  # nothing dropped: it was never published


@test
def blocking_makes_the_slow_subscribers_problem_everybodys():
    _, blocked = run(BLOCK, TICKS, RATE, CAP, (SLOW, FAST))
    _, dropping = run(DROP_NEWEST, TICKS, RATE, CAP, (SLOW, FAST))
    assert len(dropping[1].delivered) == 995
    assert len(blocked[1].delivered) == 208
    assert len(blocked[0].delivered) == len(dropping[0].delivered) == 199


@test
def a_blocking_subscriber_is_never_offered_a_message_it_cannot_hold():
    """Subscriber.offer raises if it ever happens; the broker checks first."""
    broker = Broker()
    s = broker.subscribe("t", Subscriber("s", drain_rate=0, capacity=2,
                                         policy=BLOCK))
    for now in range(10):
        broker.publish("t", now)
    assert len(s.queue) == 2
    assert broker.published == 2 and broker.refused == 8


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