diora-web/tui/tests/test_progress.py

"""The reading-position rules of diora-web docs/progress-sync.md (the same cases the server's
books/tests.py, Ops.kt and ops.js hold): the place last read wins."""
import json

import pytest

from diora_tui import progress
from diora_tui.progress import ProgressStore, cmp_progress

T = 1_760_000_000_000  # a plausible ms time


@pytest.fixture
def store(tmp_path):
    return ProgressStore(tmp_path / "progress.json")


def test_newer_reading_wins_even_if_behind(store):
    store.write("s1", "50:0.0", 0.5, T, now=T)
    store.write("s1", "10:0.0", 0.1, T + 1000, now=T + 1000)  # read back earlier in the book, later in time
    st = store.get("s1")
    assert (st.anchor, st.read_at) == ("10:0.0", T + 1000)
    assert st.furthest_anchor == "50:0.0"  # furthest only grows


def test_older_write_is_queued_but_not_the_continue_point(store):
    store.write("s1", "50:0.0", 0.5, T, now=T)
    took = store.write("s1", "10:0.0", 0.1, T - 5000, now=T)  # an offline catch-up from earlier
    assert not took
    assert store.get("s1").anchor == "50:0.0"
    assert store.get("s1").pending["anchor"] == "50:0.0"  # the later read stays the one to send


def test_clock_offset_and_nothing_from_the_future(store):
    store.offset = 5000  # the server is 5 s ahead of this machine
    store.write("s1", "3:0.0", 0.1, T, now=T)
    assert store.get("s1").read_at == T + 5000
    store.write("s1", "4:0.0", 0.2, T + 10 * 3600_000, now=T + 100)  # a read_at hours ahead of the clock
    assert store.get("s1").read_at == T + 100 + 5000


def test_force_takes_the_place_with_the_time_of_now(store):
    store.write("s1", "9:0.0", 0.9, T, now=T)
    store.write("s1", "2:0.0", 0.2, T - 1000, force=True, now=T + 50)
    st = store.get("s1")
    assert (st.anchor, st.read_at) == ("2:0.0", T + 50)
    assert st.pending["force"] is True


def test_force_stays_forced_when_reading_goes_on(store):
    store.write("s1", "2:0.0", 0.2, T, force=True, now=T)
    store.write("s1", "3:0.0", 0.3, T + 1000, now=T + 1000)
    assert store.get("s1").pending["force"] is True
    assert store.get("s1").pending["anchor"] == "3:0.0"


def test_not_a_server_book_queues_nothing(store):
    store.write("abc", "3:0.0", 0.3, T, queue=False, now=T)
    assert store.get("abc").pending is None and store.get("abc").anchor == "3:0.0"
    assert store.pending_keys() == []


def server_entry(anchor, read_at, device="Browser", furthest=None):
    return {"position_anchor": anchor, "scroll_fraction": 0.5, "read_at": read_at, "device": device,
            "furthest_anchor": furthest or anchor, "furthest_fraction": 0.5}


def test_pull_shows_the_servers_place(store):
    store.apply_server("s1", server_entry("20:0.5", T, "Handy"))
    st = store.get("s1")
    assert (st.anchor, st.read_at, st.device) == ("20:0.5", T, "Handy")


def test_pull_applies_a_newer_pending_on_top_and_drops_an_older_one(store):
    store.write("s1", "7:0.0", 0.1, T + 2000, now=T + 2000)
    store.apply_server("s1", server_entry("20:0.5", T))  # the server's is older: ours shows
    assert store.get("s1").anchor == "7:0.0"
    store.apply_server("s1", server_entry("30:0.0", T + 9000))  # the server's is newer: it shows, ours is still sent
    st = store.get("s1")
    assert st.anchor == "30:0.0" and st.pending["anchor"] == "7:0.0"


def test_sent_clears_the_queue_unless_something_newer_queued(store):
    store.write("s1", "7:0.0", 0.1, T, now=T)
    body = dict(store.get("s1").pending)
    store.write("s1", "8:0.0", 0.2, T + 1000, now=T + 1000)  # read on while the request was out
    store.sent("s1", body, {"stored": True, "read_at": T, "position_anchor": "7:0.0", "server_now": T})
    assert store.get("s1").pending["anchor"] == "8:0.0"
    body = dict(store.get("s1").pending)
    store.sent("s1", body, {"stored": True, "read_at": T + 1000, "position_anchor": "8:0.0", "scroll_fraction": 0.2})
    st = store.get("s1")
    assert st.pending is None and st.echo_read_at == T + 1000 and st.anchor == "8:0.0"


def test_refused_write_takes_the_servers_newer_place(store):
    store.write("s1", "7:0.0", 0.1, T, now=T)
    body = dict(store.get("s1").pending)
    store.sent("s1", body, {"stored": False, "read_at": T + 5000, "position_anchor": "40:0.0", "device": "Handy"})
    st = store.get("s1")
    assert (st.anchor, st.device, st.echo_read_at) == ("40:0.0", "Handy", None)


def test_server_clock_sets_the_offset(store):
    store.set_server_now(T + 7000, T)
    assert store.offset == 7000
    store.set_server_now(None)
    store.set_server_now(12, T)  # not a plausible time
    assert store.offset == 7000


def test_survives_a_restart(tmp_path):
    a = ProgressStore(tmp_path / "p.json")
    a.set_server_now(T + 100, T)
    a.write("s1", "7:0.250000", 0.3, T, now=T)
    b = ProgressStore(tmp_path / "p.json")
    assert b.offset == 100 and b.get("s1").anchor == "7:0.250000" and b.get("s1").pending["read_at"] == T


def test_reads_the_old_furthest_wins_file(tmp_path):
    (tmp_path / "p.json").write_text(json.dumps({"abc": {"anchor": "5:0.5", "updated_at": 1760000000.0}}))
    st = ProgressStore(tmp_path / "p.json").get("abc")
    assert st.anchor == "5:0.5" and st.read_at == 1760000000000 and st.pending is None


@pytest.mark.parametrize("a,b,sign", [
    (("5:0.1", 0), ("4:0.9", 0), 1),
    (("5:0.5", 0), ("5:0.4", 0), 1),
    (("4:0.9", 0), ("5:0.1", 0), -1),
    (("", 0.7), ("5:0.1", 0.2), 1),   # no anchor on one side: the fraction decides
    (("", 0.1), ("", 0.1), 0),
])
def test_cmp_progress(a, b, sign):
    r = cmp_progress(a[0], a[1], b[0], b[1])
    assert (r > 0) - (r < 0) == sign


def test_plausible_ms():
    assert progress.plausible_ms(T) == T
    assert progress.plausible_ms(5) is None
    assert progress.plausible_ms(True) is None
    assert progress.plausible_ms(float("nan")) is None
    assert progress.plausible_ms(None) is None