diora-web/tui/tests/test_annotations.py

"""Highlights, notes and bookmarks: the formats of the web reader and the merge the TUI adds
(the server replaces the whole list; nothing another device wrote may be lost)."""
import base64
import os

import pytest
import requests

from diora_tui import annotations as an, api, crypto, remote
from diora_tui.config import RemoteConfig


def hl(i, note="", block=1):
    return {"id": i, "type": "highlight", "anchor": an.make_anchor("c.xhtml", block, 0, block, 5, "quote"), "color": "yellow", "note": note, "createdAt": "x"}


def ids(items):
    return [x["id"] for x in items]


def test_merge_keeps_their_new_items_and_applies_our_changes():
    base = [hl("a"), hl("b"), hl("c")]
    ours = [hl("a", note="edited"), hl("c"), hl("ours")]  # edited a, deleted b, added ours
    theirs = [hl("a"), hl("b"), hl("c"), hl("theirs")]  # added theirs
    out = an.merge(base, ours, theirs)
    assert ids(out) == ["a", "c", "theirs", "ours"]  # b: we deleted it
    assert [x for x in out if x["id"] == "a"][0]["note"] == "edited"


def test_merge_their_deletion_does_not_come_back_unless_we_changed_it():
    base = [hl("a"), hl("b")]
    assert ids(an.merge(base, [hl("a"), hl("b")], [hl("a")])) == ["a"]  # they deleted b, we left it
    assert ids(an.merge(base, [hl("a"), hl("b", note="mine")], [hl("a")])) == ["a", "b"]  # we edited it: kept


def test_merge_without_a_base_deletes_nothing():
    assert ids(an.merge(None, [hl("a")], [hl("a"), hl("b")])) == ["a", "b"]


def test_bookmarks_are_newest_first():
    bm = lambda i: {"id": i, "label": i, "anchor": "c", "scrollFraction": 0, "createdAt": "x"}  # noqa: E731
    out = an.merge([bm("old")], [bm("new"), bm("old")], [bm("old"), bm("theirs")], prepend_added=True)
    assert ids(out) == ["new", "old", "theirs"]


@pytest.fixture
def store(tmp_path):
    return an.AnnotationStore(tmp_path / "ann")


def test_edits_are_queued_and_survive_a_restart(store, tmp_path):
    h = an.new_highlight(an.make_anchor("c", 3, 0, 3, 4, "abcd"), "green", "n")
    store.add("s1", "highlights", h)
    store.add("s1", "bookmarks", an.new_bookmark("Kap. 1", "c", 0.1))
    store.add("s1", "bookmarks", an.new_bookmark("Kap. 2", "c2", 0.2))
    assert store.dirty_keys() == ["s1"]
    again = an.AnnotationStore(tmp_path / "ann").get("s1")
    assert ids(again.highlights) == [h["id"]] and [b["label"] for b in again.bookmarks] == ["Kap. 2", "Kap. 1"]
    assert again.dirty == {"highlights": True, "bookmarks": True}


def test_note_edit_and_delete(store):
    h = an.new_highlight(an.make_anchor("c", 3, 0, 3, 4, "abcd"))
    store.add("s1", "highlights", h)
    assert store.set_note("s1", h["id"], "text") and not store.set_note("s1", h["id"], "text")
    assert store.get("s1").highlights[0]["note"] == "text"
    assert store.delete("s1", "highlights", h["id"]) and not store.delete("s1", "highlights", h["id"])


def test_pull_replaces_when_clean_and_merges_when_dirty(store):
    store.take_remote("s1", "highlights", [hl("a")])
    assert ids(store.get("s1").highlights) == ["a"] and not store.get("s1").dirty["highlights"]
    store.add("s1", "highlights", hl("mine"))
    store.take_remote("s1", "highlights", [hl("a"), hl("b")])
    assert ids(store.get("s1").highlights) == ["a", "b", "mine"] and store.get("s1").dirty["highlights"]


def test_push_flow_with_an_edit_during_the_request(store):
    store.take_remote("s1", "highlights", [hl("a")])
    store.add("s1", "highlights", hl("one"))
    merged, snap = store.for_push("s1", "highlights", [hl("a"), hl("theirs")])
    assert ids(merged) == ["a", "theirs", "one"]
    store.add("s1", "highlights", hl("two"))  # made while the request was out
    store.pushed("s1", "highlights", merged, snap)
    got = store.get("s1")
    assert ids(got.highlights) == ["a", "theirs", "one", "two"] and got.dirty["highlights"]
    assert ids(got.base["highlights"]) == ["a", "theirs", "one"]


def test_blob_roundtrip():
    key = base64.b64encode(os.urandom(32)).decode()
    items = [hl("ä")]
    assert an.decode(key, an.encode(key, items)) == items
    assert an.decode(key, {"ct": None, "iv": None}) == []


CFG = RemoteConfig("http://x", "tok", base64.b64encode(os.urandom(32)).decode())


class FakeServer:
    def __init__(self):
        self.blobs = {}
        self.fail = None

    def fetch(self, server, token, book_id, kind):
        if self.fail:
            raise self.fail
        return self.blobs.get((book_id, kind), {"ct": None, "iv": None})

    def post(self, server, token, book_id, kind, blob):
        if self.fail:
            raise self.fail
        self.blobs[(book_id, kind)] = blob


@pytest.fixture
def server(store, monkeypatch):
    s = FakeServer()
    monkeypatch.setattr(an, "_default", store)
    monkeypatch.setattr(api, "fetch_annotations", s.fetch)
    monkeypatch.setattr(api, "post_annotations", s.post)
    return s


def test_push_merges_with_what_another_device_wrote(server, store):
    server.blobs[(7, "highlights")] = an.encode(CFG.enc_key_b64, [hl("theirs")])
    store.add("s7", "highlights", hl("mine"))
    assert remote.push_annotations(CFG) == 1
    assert ids(an.decode(CFG.enc_key_b64, server.blobs[(7, "highlights")])) == ["theirs", "mine"]
    assert store.dirty_keys() == [] and ids(store.get("s7").highlights) == ["theirs", "mine"]


def test_push_keeps_the_queue_offline_and_pull_survives_it(server, store):
    store.add("s7", "highlights", hl("mine"))
    server.fail = requests.ConnectionError("down")
    assert remote.push_annotations(CFG) == 0 and store.dirty_keys() == ["s7"]
    assert remote.pull_annotations(CFG, 7, "s7") is False and ids(store.get("s7").highlights) == ["mine"]


def test_local_books_are_never_pushed(server, store):
    store.add("0123456789abcdef", "highlights", hl("mine"))
    assert remote.push_annotations(CFG) == 0 and not server.blobs