"""Plain asserts, no pytest:  python3 test_dlock.py

Every number this toy's commentary claims is pinned here. The tests that
matter most are `test_lock_is_innocent` (no two leases ever overlap) and
`test_boundary_formula` (the corruption boundary is an inequality, not a
coincidence -- it predicts all 140 cells of the sweep grid).
"""

from dlock import (LockServer, LockService, Resource, scenario,
                   overlapping_grants, stale_overwrites, unlocked_writes)

TESTS = []


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


DEMO = dict(ttl=10, pause_from=2, pause_to=16, b_arrives=12, n=5)


# ---------------------------------------------------------------- lock server

@test
def test_grant_when_free():
    s = LockServer(1)
    assert s.acquire("A", 0, 10) == 1
    assert s.holder == "A" and s.expires == 10


@test
def test_deny_while_lease_live():
    s = LockServer(1)
    s.acquire("A", 0, 10)
    assert s.acquire("B", 9, 10) is None       # 9 < 10, still A's
    assert s.acquire("B", 10, 10) == 2         # expiry is half-open


@test
def test_tokens_are_monotonic():
    s = LockServer(1)
    assert [s.acquire(c, t, 5) for c, t in (("A", 0), ("B", 5), ("A", 10))] \
        == [1, 2, 3]


@test
def test_release_needs_a_live_lease():
    s = LockServer(1)
    s.acquire("A", 0, 10)
    assert s.release("B", 3) is False          # not the holder
    assert s.release("A", 10) is False         # lease already expired
    assert s.release("A", 3) is True
    assert s.grants[0]["end"] == 3             # audit record truncated


@test
def test_holder_at_is_half_open():
    s = LockServer(1)
    s.acquire("A", 0, 10)
    assert s.holder_at(0)["client"] == "A"
    assert s.holder_at(9)["client"] == "A"
    assert s.holder_at(10) is None
    assert s.holder_at(-1) is None


@test
def test_quorum_sizes():
    assert [LockService(n).quorum for n in (1, 3, 5, 7, 9, 51)] \
        == [1, 2, 3, 4, 5, 26]


@test
def test_partial_grant_is_handed_back():
    """B cannot reach a quorum while A holds 3 of 5, and must not keep the
    two it did get -- otherwise a later acquirer is blocked by a phantom."""
    svc = LockService(5)
    assert svc.acquire("A", 0, 10) == 1
    for s in svc.servers[:2]:                  # expire A on two servers only
        s.expires = 0
    assert svc.acquire("B", 1, 10) is None     # 2 of 5 is not a quorum
    assert all(s.holder in (None, "A") for s in svc.servers)


# ------------------------------------------------------------------- resource

@test
def test_unfenced_resource_accepts_anything():
    r = Resource(fencing=False)
    assert r.write("B", 0, "B", 2) is True
    assert r.write("A", 1, "A", 1) is True
    assert r.value == "A"


@test
def test_fenced_resource_rejects_a_lower_token():
    r = Resource(fencing=True)
    assert r.write("B", 0, "B", 2) is True
    assert r.write("A", 1, "A", 1) is False
    assert r.value == "B" and r.max_token == 2


@test
def test_one_lease_may_write_twice():
    """The fencing test is `<`, not `<=`, so a holder is not limited to a
    single write per lease. Tokens are unique per grant, so a stale writer is
    always strictly lower and is still caught."""
    r = Resource(fencing=True)
    assert r.write("B", 14, "B1", 2) is True
    assert r.write("B", 15, "B2", 2) is True
    assert r.write("A", 16, "A", 1) is False
    assert r.value == "B2"


# ----------------------------------------------------------------- the schedule

@test
def test_run1_corrupts_the_resource():
    lock, res, st, _ = scenario(**DEMO)
    assert res.value == "A"                    # B's completed work, erased
    assert stale_overwrites(res) == [
        dict(t=16, client="A", token=1, applied=2)]


@test
def test_lock_is_innocent():
    """THE assertion. Every lease every one of the five servers ever issued,
    compared pairwise: no two are valid at the same instant."""
    lock, res, st, _ = scenario(**DEMO)
    assert overlapping_grants(lock) == []
    grants = [(g["client"], g["token"], g["start"], g["end"])
              for s in lock.servers for g in s.grants]
    assert grants == [("A", 1, 0, 10), ("B", 2, 12, 15)] * 5


@test
def test_a_held_nothing_when_it_wrote():
    lock, res, st, _ = scenario(**DEMO)
    assert unlocked_writes(lock, res) == [dict(t=16, client="A")]
    assert lock.held_by("A", 16) is False
    assert lock.held_by("B", 14) is True       # B's write was legitimate
    assert lock.servers[0].holder_at(16) is None


@test
def test_fencing_fixes_the_identical_schedule():
    lock, res, st, _ = scenario(**DEMO, fencing=True)
    assert res.value == "B"
    assert stale_overwrites(res) == []
    assert unlocked_writes(lock, res) == []
    assert res.log[-1] == dict(t=16, client="A", value="A", token=1, ok=False)


@test
def test_fencing_changes_nothing_about_the_lock():
    """The fix is entirely in the resource. Same leases, same tokens, same
    trace up to the one word that differs on A's write."""
    a = scenario(**DEMO)
    b = scenario(**DEMO, fencing=True)
    assert [s.grants for s in a[0].servers] == [s.grants for s in b[0].servers]
    assert a[3][:-1] == b[3][:-1]
    assert a[3][-1].endswith("ACCEPTED")
    assert b[3][-1].endswith("REJECTED (stale token)")


@test
def test_more_lock_servers_change_nothing():
    for n in (1, 3, 5, 7, 9, 51):
        lock, res, st, _ = scenario(**{**DEMO, "n": n})
        assert res.value == "A", n
        assert len(stale_overwrites(res)) == 1, n
        assert overlapping_grants(lock) == [], n


@test
def test_deterministic():
    a = scenario(**DEMO)
    b = scenario(**DEMO)
    assert a[3] == b[3]
    assert a[1].value == b[1].value


# ------------------------------------------------------- the boundary, derived

PAUSES = list(range(2, 42, 3))
TTLS = (1, 2, 4, 8, 10, 16, 25, 40, 80, 1000)


@test
def test_boundary_formula():
    """A writes at pause_from + pause_len. B is blocked until A's lease
    expires (t = ttl) or A releases, and cannot start before it arrives, so it
    acquires at max(b_arrives, min(ttl, release)) and writes two ticks later.
    Corruption is exactly "B's write lands first", which reduces to
    pause_len > max(ttl, b_arrives). Checked against all 140 cells."""
    checked = 0
    for ttl in TTLS:
        for pl in PAUSES:
            _, res, _, _ = scenario(ttl=ttl, pause_from=2, pause_to=2 + pl,
                                    b_arrives=12, n=5)
            predicted = pl > max(ttl, 12)
            assert bool(stale_overwrites(res)) == predicted, (ttl, pl)
            checked += 1
    assert checked == 140


@test
def test_fencing_is_clean_on_every_cell():
    for ttl in TTLS:
        for pl in PAUSES:
            _, res, _, _ = scenario(ttl=ttl, pause_from=2, pause_to=2 + pl,
                                    b_arrives=12, n=5, fencing=True)
            assert stale_overwrites(res) == [], (ttl, pl)


@test
def test_longer_lease_buys_safety_with_blocked_time():
    """A longer lease does not remove the failure, it moves the threshold --
    and B waits one extra tick for every tick added, until A finally releases
    at t=17 and the wait stops growing."""
    for ttl in (1, 5, 10, 12, 13, 14, 15, 16, 20, 40, 1000):
        lock, res, st, _ = scenario(**{**DEMO, "ttl": ttl})
        assert st["b_at"] == max(12, min(ttl, 17)), ttl


if __name__ == "__main__":
    for t in TESTS:
        t()
        print(f"  ok  {t.__name__}")
    print(f"All {len(TESTS)} tests PASSED")