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

  python3 test_quorum.py
"""

from quorum import Cluster, Replica

N, W, R, KEY = 5, 3, 3, "cart"
tests = []


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


def staged(w=W, r=R, reach=(3,), repair=True):
    c = Cluster(N, w, r, repair=repair)
    c.write(KEY, 1, "A", reach=list(range(1, N + 1)))
    b = c.write(KEY, 2, "B", reach=list(reach))
    return c, b


# ---- the ratchet ---------------------------------------------------------

@test
def put_accepts_a_newer_version():
    rep = Replica(1)
    assert rep.put(KEY, 1, "A", "t") is True
    assert rep.put(KEY, 2, "B", "t") is True
    assert rep.get(KEY) == (2, "B")


@test
def put_rejects_an_older_version():
    rep = Replica(1)
    rep.put(KEY, 2, "B", "t")
    assert rep.put(KEY, 1, "A", "t") is False
    assert rep.get(KEY) == (2, "B")


@test
def put_rejects_the_version_it_already_has():
    """`>=`, not `>`: re-applying the same version is a no-op, which is what
    makes `repaired` in section 6 report [2, 4] and not [2, 3, 4]."""
    rep = Replica(1)
    rep.put(KEY, 2, "B", "t")
    assert rep.put(KEY, 2, "B", "t") is False
    assert len(rep.applied) == 1


@test
def force_writes_past_the_ratchet():
    rep = Replica(1)
    rep.put(KEY, 2, "B", "t")
    rep.force(KEY, 1, "A", "rollback")
    assert rep.get(KEY) == (1, "A")


# ---- the staged run ------------------------------------------------------

@test
def the_partial_write_is_reported_failed():
    _, b = staged()
    assert b["ok"] is False and b["acked"] == [3]


@test
def the_partial_write_is_kept_by_the_replica_that_took_it():
    """There is no undo: FAILED does not roll r3 back to v1."""
    c, _ = staged()
    assert c.rep(3).get(KEY) == (2, "B")
    assert c.holders(KEY, 2) == [3]


@test
def six_of_ten_read_sets_see_it_before_any_repair():
    """C(4, 2) = 6 -- the read sets that contain r3."""
    c, _ = staged()
    assert c.count_returning(KEY, "B") == (6, 10)
    assert sum(1 for s, v in c.read_sets(KEY) if v == "B" and 3 not in s) == 0


@test
def a_read_missing_r3_returns_the_old_value_and_repairs_nothing():
    c, _ = staged()
    g = c.read(KEY, [1, 2, 4])
    assert g["value"] == "A" and g["repaired"] == []


@test
def a_read_touching_r3_returns_it_and_repairs_outward():
    c, _ = staged()
    g = c.read(KEY, [2, 3, 4])
    assert g["version"] == 2 and g["value"] == "B" and g["repaired"] == [2, 4]


@test
def one_read_makes_the_failed_write_unanimous():
    """THE AHA. 6/10 -> 10/10, and the client was told FAILED."""
    c, b = staged()
    assert b["ok"] is False
    assert c.count_returning(KEY, "B") == (6, 10)
    c.read(KEY, [2, 3, 4])
    assert c.count_returning(KEY, "B") == (10, 10)
    assert c.holders(KEY, 2) == [2, 3, 4]


@test
def three_holders_of_five_is_exactly_what_makes_it_unanimous():
    """3 + 3 > 5: any R=3 set and any 3 holders must intersect. Two holders
    would not be enough, and the audit agrees."""
    c, _ = staged()
    c.read(KEY, [2, 3])          # R=3, so this read is refused, not served
    assert c.count_returning(KEY, "B") == (6, 10)
    c.rep(2).put(KEY, 2, "B", "manual")
    assert c.count_returning(KEY, "B") == (9, 10)


@test
def a_read_below_R_is_refused():
    c, _ = staged()
    assert c.read(KEY, [3])["ok"] is False


# ---- the three refutations ----------------------------------------------

@test
def the_outcome_depends_on_r_alone_and_not_on_w_at_all():
    """Sweep 1, stated as an invariant: down each R column of that table the
    before/after counts are identical for all five W, so the W+R > N line
    runs straight through the middle of a block of equal outcomes."""
    for r in range(1, N + 1):
        column = []
        for w in range(1, N + 1):
            c, _ = staged(w=w, r=r)
            before = c.count_returning(KEY, "B")
            s = next(t for t, _ in c.read_sets(KEY, r) if 3 in t)
            c.read(KEY, s)
            column.append((before, c.count_returning(KEY, "B")))
        assert len(set(column)) == 1, (r, column)
    assert column and column[0][0] == (1, 1)          # r = 5


@test
def one_repairing_read_leaves_exactly_r_holders():
    """Which is why the promotion is unanimous iff 2R > N, and why R=2 gets
    only to 7/10 while R=3 gets to 10/10."""
    for r in range(1, N + 1):
        c, _ = staged(r=r)
        s = next(t for t, _ in c.read_sets(KEY, r) if 3 in t)
        c.read(KEY, s)
        assert len(c.holders(KEY, 2)) == r
        n, total = c.count_returning(KEY, "B")
        assert (n == total) == (2 * r > N), (r, n, total)


@test
def w_plus_r_greater_than_n_does_not_prevent_it():
    """W=3, R=3 satisfies W+R > N and promotes the failed write anyway;
    W=1, R=3 violates it and behaves identically."""
    for w in (1, 3, 5):
        c, b = staged(w=w, r=3)
        assert c.count_returning(KEY, "B") == (6, 10)
        c.read(KEY, [2, 3, 4])
        assert c.count_returning(KEY, "B") == (10, 10)


@test
def raising_w_only_moves_the_verdict_never_the_residue():
    """The inversion: for a write that reached one replica, every W from 2 up
    reports FAILED and every one of them ends with 'B' everywhere."""
    verdicts = []
    for w in range(1, N + 1):
        c, b = staged(w=w)
        c.read(KEY, [2, 3, 4])
        verdicts.append(b["ok"])
        assert c.count_returning(KEY, "B") == (10, 10)
    assert verdicts == [True, False, False, False, False]


@test
def read_repair_off_leaves_it_flapping_rather_than_gone():
    c, _ = staged(repair=False)
    seen = [c.read(KEY, list(s))["value"]
            for s in ([1, 2, 4], [2, 3, 4], [1, 2, 4], [1, 4, 5], [1, 2, 5])]
    assert seen == ["A", "B", "A", "A", "A"]
    assert c.count_returning(KEY, "B") == (6, 10)


@test
def anti_entropy_finishes_the_job_read_repair_was_not_doing():
    c, _ = staged(repair=False)
    c.anti_entropy(KEY, [(1, 3)])
    assert c.count_returning(KEY, "B") == (9, 10)
    c.anti_entropy(KEY, [(1, 4)])
    assert c.count_returning(KEY, "B") == (10, 10)


@test
def anti_entropy_skips_pairs_that_already_agree():
    c, _ = staged(repair=False)
    assert c.anti_entropy(KEY, [(1, 2)]) == []


# ---- rollback ------------------------------------------------------------

@test
def rollback_works_if_the_coordinator_survives():
    c, b = staged()
    c.rollback(KEY, b["acked"], 1, "A")
    c.read(KEY, [2, 3, 4])
    assert c.holders(KEY, 2) == []
    assert c.count_returning(KEY, "B") == (0, 10)


@test
def rollback_is_worth_nothing_if_the_coordinator_dies_first():
    c, _ = staged()
    c.read(KEY, [2, 3, 4])          # the crash: no rollback ever runs
    assert c.count_returning(KEY, "B") == (10, 10)


# ---- the boundaries ------------------------------------------------------

@test
def a_write_that_reached_nobody_has_nothing_to_promote():
    c, b = staged(reach=())
    assert b["ok"] is False and b["acked"] == []
    c.read(KEY, [2, 3, 4])
    assert c.count_returning(KEY, "B") == (0, 10)


@test
def at_r_equals_one_repair_has_nowhere_to_push():
    c, _ = staged(r=1)
    assert c.count_returning(KEY, "B", r=1) == (1, 5)
    g = c.read(KEY, [3])
    assert g["value"] == "B" and g["repaired"] == []
    assert c.count_returning(KEY, "B", r=1) == (1, 5)


@test
def the_w1_mirror_case_is_told_ok_and_is_almost_invisible():
    c, b = staged(w=1, r=1)
    assert b["ok"] is True
    assert c.count_returning(KEY, "B", r=1) == (1, 5)


# ---- audit invariants ----------------------------------------------------

@test
def read_sets_does_not_mutate_the_cluster():
    """The audit must not repair, or it would change what it is measuring."""
    c, _ = staged()
    before = c.state(KEY)
    c.read_sets(KEY)
    c.count_returning(KEY, "B")
    assert c.state(KEY) == before


@test
def the_told_log_records_a_failure_the_replicas_disagree_with():
    c, _ = staged()
    c.read(KEY, [2, 3, 4])
    v2 = [t for t in c.told if t["version"] == 2][0]
    assert v2["ok"] is False
    assert len(c.holders(KEY, 2)) == 3


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