diora-web/tui/tools/smoke_server.py

"""End-to-end check of the sync against a LOCAL diora server (never a real account):
login, protected credentials, book list, lazy download, a position going out, another device
reading later, an offline catch-up that must lose. Throwaway state in a temp folder.

  python tools/smoke_server.py http://127.0.0.1:8001 <username> <password>

It can run again on the same database (the furthest place only ever grows).

The server is the diora-web checkout run with its own throwaway database, seeded like
diora-android's tools/seed_testserver.py does.
"""
import os
import sys
import tempfile
import time
from pathlib import Path

home = tempfile.mkdtemp(prefix="diora-tui-smoke-")
for var in ("LOCALAPPDATA", "APPDATA", "XDG_DATA_HOME", "XDG_CONFIG_HOME", "XDG_CACHE_HOME"):
    os.environ[var] = home
os.environ["HOME"] = home
os.environ["DIORA_TUI_HOME"] = home

from diora_tui import api, blocks, config, crypto, epub, login, progress, remote  # noqa: E402

server, username, password = sys.argv[1].rstrip("/"), sys.argv[2], sys.argv[3]
if "127.0.0.1" not in server and "localhost" not in server:
    sys.exit("only against a local test server")


def check(what, ok, detail=""):
    print(("OK   " if ok else "FAIL ") + what + (f"  [{detail}]" if detail else ""))
    if not ok:
        sys.exit(1)


token, name = login.web_login(server, username, password)
key_b64 = crypto.derive_key_b64(name, password)
cfg = config.RemoteConfig(server, token, key_b64, name)
config.save(cfg)
back = config.load()
check("credentials stored protected and read back",
      back == cfg and (sys.platform != "win32" or "dpapi:" in Path(config._CONFIG_FILE).read_text()))
check("no token or key in the config file in the clear",
      token not in Path(config._CONFIG_FILE).read_text() and key_b64 not in Path(config._CONFIG_FILE).read_text())

lib = Path(home) / "lib"
res = remote.sync_library(lib, cfg)
check("book list pulled, nothing downloaded", res.books >= 1 and not list(lib.glob("*.epub")), f"{res.books} books")
store = progress.default()
check("server clock offset learnt", abs(store.offset) < 5000, f"{store.offset} ms")

entry = sorted(__import__("diora_tui.library_meta", fromlist=["x"]).remote_entries().items())[0]
skey, info = entry
path = remote.download_book(lib, cfg, info["server_id"], info["title"])
check("book downloaded on demand and keyed by the server id", path.exists() and epub._book_id(path) == skey, path.name)
book = blocks.load_flat_book(path)

# the reader arrives at block 12 now
now = progress.now_ms()
a1 = blocks.format_anchor(12, 0.5)
store.write(skey, a1, blocks.fraction_for_anchor(book, 12, 0.5), now)
check("position queued", store.get(skey).pending is not None)
check("flushed", remote.flush_pending(cfg) == 1 and store.get(skey).pending is None)
row = [b for b in api.fetch_books(server, token)[0] if b["id"] == info["server_id"]][0]
check("server holds it with device and a real fraction",
      row["position_anchor"] == a1 and row["device"] == "diora TUI" and 0 < row["scroll_fraction"] < 1
      and abs(row["read_at"] - now) < 10_000, f"fraction {row['scroll_fraction']:.4f}, read_at diff {row['read_at'] - now} ms")

# another device reads somewhere else, later
time.sleep(0.05)
other = blocks.format_anchor(3, 0.0)
api.post_progress(server, token, info["server_id"], anchor=other, fraction=0.02, read_at=progress.now_ms(), device="Handy")
remote.sync_library(lib, cfg)
st = store.get(skey)
check("a pull shows the later position of the other device", st.anchor == other and st.device == "Handy")
check("the furthest place is kept (never behind what was read)",
      progress.cmp_progress(st.furthest_anchor, 0, a1, 0) >= 0, st.furthest_anchor)

# an offline catch-up from before that must not become the continue point
store.write(skey, blocks.format_anchor(30, 0.0), 0.5, progress.now_ms() - 120_000)
remote.flush_pending(cfg)
row = [b for b in api.fetch_books(server, token)[0] if b["id"] == info["server_id"]][0]
check("an older write loses on the server", row["position_anchor"] == other and row["device"] == "Handy")
check("...and locally", store.get(skey).anchor == other and store.get(skey).pending is None)
check("...but raises the furthest place",
      progress.cmp_progress(row["furthest_anchor"], 0, blocks.format_anchor(30, 0.0), 0) >= 0, row["furthest_anchor"])
print("all good")