diora-web/tui/tests/test_reader_progress.py

"""The reader saves a position only when the reader moved it, and offers — never applies — a
newer position from another device (docs/progress-sync.md)."""
import asyncio

import pytest

from diora_tui import blocks, config as config_mod, epub as epub_mod, layout, progress
from diora_tui.app import DioraTuiApp
from diora_tui.config import RemoteConfig
from diora_tui.reader_screen import JumpOfferScreen, ReaderScreen

T = 1_760_000_000_000


@pytest.fixture
def store(tmp_path, monkeypatch):
    s = progress.ProgressStore(tmp_path / "p.json")
    monkeypatch.setattr(progress, "_default", s)
    return s


def book(epub, name="Testbuch"):
    body = "<h1>Anfang</h1>" + "".join(f"<p>Absatz {i} " + "wort " * 40 + "</p>" for i in range(80))
    return epub([body], title=name)


async def open_reader(app, pilot, path):
    app.push_screen(ReaderScreen(path))
    for _ in range(80):
        await pilot.pause(0.1)
        s = app.screen
        if isinstance(s, ReaderScreen) and s.book is not None and s._last_anchor is not None:
            await pilot.pause(0.2)
            return s
    raise AssertionError("reader did not load")


def test_only_the_readers_moves_are_saved(epub, store, tmp_path):
    path = book(epub)
    lib = tmp_path / "lib"
    lib.mkdir()

    async def run():
        app = DioraTuiApp(lib)
        async with app.run_test(size=(100, 30)) as pilot:
            reader = await open_reader(app, pilot, path)
            key = epub_mod._book_id(path)
            reader._save_progress()
            assert store.get(key) is None  # opened and looked at: nothing written

            await pilot.press(*["j"] * 40)
            reader._save_progress()
            st = store.get(key)
            assert st is not None and blocks.parse_anchor(st.anchor)[0] > 0
            assert st.pending is None  # not a server book: nothing queued
            assert 0 < st.fraction < 1 and abs(st.read_at - progress.now_ms()) < 5000
            first = st.read_at

            await pilot.resize_terminal(70, 24)  # re-wrapped by the program, not read
            await pilot.pause(0.3)
            reader._save_progress()
            assert store.get(key).read_at == first

            reader._save_progress()  # unchanged: no new time either
            assert store.get(key).read_at == first

    asyncio.run(run())


def test_reopening_returns_to_the_saved_place_without_writing(epub, store, tmp_path):
    path = book(epub)
    key = epub_mod._book_id(path)
    lib = tmp_path / "lib"
    lib.mkdir()

    async def run():
        app = DioraTuiApp(lib)
        async with app.run_test(size=(100, 30)) as pilot:
            reader = await open_reader(app, pilot, path)
            await pilot.press(*["j"] * 60)
            reader._save_progress()
            saved = store.get(key)
            app.pop_screen()
            await pilot.pause()

            reader = await open_reader(app, pilot, path)
            assert reader._book_view.scroll_y > 30
            reader._save_progress()
            assert store.get(key).read_at == saved.read_at  # restored, not "read" again

    asyncio.run(run())


@pytest.mark.parametrize("answer,jumps", [("y", True), ("n", False)])
def test_newer_position_from_elsewhere_is_offered(epub, store, tmp_path, monkeypatch, answer, jumps):
    lib = tmp_path / "lib"
    lib.mkdir()
    path = book(epub, "Server")
    target = lib / "0042 - Server.epub"
    path.rename(target)
    monkeypatch.setattr(config_mod, "load", lambda: RemoteConfig("http://x", "t", "k"))
    store.apply_server("s42", {"position_anchor": "5:0.0", "scroll_fraction": 0.05, "read_at": T, "device": "Handy"})

    async def run():
        app = DioraTuiApp(lib)
        async with app.run_test(size=(100, 30)) as pilot:
            reader = await open_reader(app, pilot, target)
            y0 = reader._book_view.scroll_y

            # another device read further on, later
            store.apply_server("s42", {"position_anchor": "60:0.0", "scroll_fraction": 0.8, "read_at": T + 60_000, "device": "Handy"})
            reader._offer_newer()
            await pilot.pause()
            assert isinstance(app.screen, JumpOfferScreen)
            assert reader._book_view.scroll_y == y0  # asked, not moved
            await pilot.press(answer)
            await pilot.pause(0.3)
            assert isinstance(app.screen, ReaderScreen)
            moved = reader._book_view.scroll_y > y0 + 20
            assert moved == jumps

            reader._offer_newer()  # the same state is not offered twice
            await pilot.pause()
            assert isinstance(app.screen, ReaderScreen)

            reader._save_progress()  # jumping isn't reading: still the server's time
            assert store.get("s42").read_at == T + 60_000

    asyncio.run(run())