siftlode/backend/tests/test_plex_watch_reconcile.py

"""The daily full watch reconcile (`run_plex_watch_reconcile`) — the one pass that DELETES watch state:
a mirrored row Plex no longer has a record for is read as "Plex un-watched it" and cleared. A review
found it trusting a scan a Plex hiccup had cut short; these hold what it may and may not clear.
"""

import pytest
from sqlalchemy import select

from app import sysconfig
from app.models import PlexItem, PlexLibrary, PlexLink, PlexState, User
from app.plex import watch_sync


class _FakePlex:
    """A PlexClient stand-in: one movie section whose listing reports `total` (more than it hands
    over = a walk cut short), plus any `more` sections as `(type, leaves, total)` — listed AFTER
    the movies — and a record of every write the reconcile sends back to Plex."""

    def __init__(self, leaves: list[dict], total: int | None = None, more=()):
        self.listings = {"1": ("movie", leaves, len(leaves) if total is None else total)}
        for i, (stype, extra, extra_total) in enumerate(more, start=2):
            self.listings[str(i)] = (stype, extra, extra_total)
        self.calls: list[tuple] = []

    def __enter__(self):
        return self

    def __exit__(self, *exc):
        return False

    def sections(self):
        return [{"type": stype, "key": key} for key, (stype, _l, _t) in self.listings.items()]

    def section_items(self, key, item_type, start, size):
        _stype, leaves, total = self.listings[key]
        return leaves[start : start + size], total

    def scrobble(self, rk):
        self.calls.append(("scrobble", rk))

    def unscrobble(self, rk):
        self.calls.append(("unscrobble", rk))

    def set_timeline(self, rk, time_ms, duration_ms):
        self.calls.append(("timeline", rk, time_ms))


@pytest.fixture
def world(db):
    """Five mirrored movies and the owner's link; each test sets the local states it needs."""
    sysconfig.set_value(db, "plex_enabled", True)
    lib = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True)
    user = User(email="reconcile@example.com")
    db.add_all([lib, user])
    db.flush()
    items = {}
    for rk in ("m1", "m2", "m3", "m4", "m5"):
        items[rk] = PlexItem(
            rating_key=rk, library_id=lib.id, kind="movie", title=rk, duration_s=6000
        )
        db.add(items[rk])
    db.add(
        PlexLink(
            user_id=user.id,
            uses_admin=True,
            sync_enabled=True,
            initial_import_done=True,
            plex_account_id=1,
        )
    )
    db.flush()

    def state(rk, **kw):
        db.add(PlexState(user_id=user.id, item_id=items[rk].id, **kw))

    def rows():
        db.expire_all()
        by_id = {i.id: rk for rk, i in items.items()}
        return {
            by_id[s.item_id]: (s.status, s.position_seconds, s.synced_to_plex)
            for s in db.execute(select(PlexState).where(PlexState.user_id == user.id)).scalars()
            if s.item_id in by_id  # the five movies; a test's own extra items it reads itself
        }

    return state, rows


def _run(db, monkeypatch, plex):
    monkeypatch.setattr(watch_sync, "PlexClient", lambda _db: plex)
    return watch_sync.run_plex_watch_reconcile(db)


def _local_states(state):
    state("m1", status="watched", synced_to_plex=True)  # mirrored; Plex has no record now
    state("m2", status="watched", synced_to_plex=False)  # local change that never reached Plex
    state("m3", status="hidden", synced_to_plex=True)  # Siftlode-only, never cleared
    state("m5", status="new", position_seconds=300, synced_to_plex=True)  # Plex says otherwise


def test_a_complete_scan_clears_what_plex_unwatched_and_keeps_the_rest(db, monkeypatch, world):
    state, rows = world
    _local_states(state)
    db.commit()
    plex = _FakePlex(
        [
            {"ratingKey": "m3"},  # listed, no watch signal
            {"ratingKey": "m4", "viewCount": 1, "lastViewedAt": 1_700_000_000},
            {"ratingKey": "m5", "viewOffset": 120_000, "lastViewedAt": 1_700_000_000},
        ]
    )
    out = _run(db, monkeypatch, plex)
    assert rows() == {
        "m2": ("watched", 0, True),  # re-pushed, then settled
        "m3": ("hidden", 0, True),
        "m4": ("watched", 0, True),  # pulled in
        "m5": ("new", 120, True),  # Plex's position wins
    }
    assert ("scrobble", "m2") in plex.calls
    assert out["cleared"] == 1 and out["repushed"] == 1 and "incomplete" not in out


def test_a_scan_cut_short_clears_nothing(db, monkeypatch, world):
    state, rows = world
    _local_states(state)
    db.commit()
    # The server said 5, the walk got 3: m1 may simply not have been reached.
    plex = _FakePlex(
        [
            {"ratingKey": "m3"},
            {"ratingKey": "m4", "viewCount": 1, "lastViewedAt": 1_700_000_000},
            {"ratingKey": "m5", "viewOffset": 120_000, "lastViewedAt": 1_700_000_000},
        ],
        total=5,
    )
    out = _run(db, monkeypatch, plex)
    after = rows()
    assert after["m1"] == ("watched", 0, True)  # NOT un-watched on a partial picture
    assert after["m4"] == ("watched", 0, True) and after["m5"] == ("new", 120, True)  # pulls stand
    assert after["m2"] == ("watched", 0, True)  # the dirty row is still pushed up
    assert out["cleared"] == 0 and out["incomplete"] == 1


_LISTED = [
    {"ratingKey": "m3"},
    {"ratingKey": "m4", "viewCount": 1, "lastViewedAt": 1_700_000_000},
    {"ratingKey": "m5", "viewOffset": 120_000, "lastViewedAt": 1_700_000_000},
]


def test_a_genuinely_empty_second_section_still_lets_the_reconcile_clear(db, monkeypatch, world):
    # The server SAYS the shows section holds 0: that is a complete listing, not a hiccup — before
    # a review it read as "incomplete" forever and un-watches never synced again.
    state, rows = world
    _local_states(state)
    db.commit()
    out = _run(db, monkeypatch, _FakePlex(_LISTED, more=[("show", [], 0)]))
    assert "m1" not in rows()
    assert out["cleared"] == 1 and "incomplete" not in out


def _watched_episode_in_a_shows_section(db) -> int:
    """A second mirrored section (key "2") holding one episode the owner has watched, synced."""
    shows = PlexLibrary(plex_key="2", title="Shows", kind="show", enabled=True)
    db.add(shows)
    db.flush()
    ep = PlexItem(rating_key="e1", library_id=shows.id, kind="episode", title="e1")
    db.add(ep)
    db.flush()
    user_id = db.execute(select(PlexLink.user_id)).scalar_one()
    db.add(PlexState(user_id=user_id, item_id=ep.id, status="watched", synced_to_plex=True))
    return ep.id


def _state_of(db, item_id: int) -> str | None:
    db.expire_all()
    st = db.execute(select(PlexState).where(PlexState.item_id == item_id)).scalar_one_or_none()
    return st.status if st else None


def test_a_section_we_mirror_that_says_0_keeps_its_rows_and_does_not_hold_up_the_rest(
    db, monkeypatch, world
):
    # A total of 0 from a section our items come from is a Plex that lost the library for a moment,
    # not one that un-watched all of it: its rows stay. And it is decided per
    # section: at first one such section (a library emptied on Plex, its mirrored items kept)
    # made EVERY run incomplete, and un-watches in every other section never synced again.
    state, rows = world
    _local_states(state)
    ep_id = _watched_episode_in_a_shows_section(db)
    db.commit()
    out = _run(db, monkeypatch, _FakePlex(_LISTED, more=[("show", [], 0)]))
    assert _state_of(db, ep_id) == "watched"
    assert "m1" not in rows()  # the movies section was walked whole: Plex un-watched m1
    assert out["cleared"] == 1 and out["incomplete"] == 1


def test_a_library_the_allowlist_leaves_out_keeps_its_watch_states(db, monkeypatch, world):
    # Its items are absent from every scan because the scan never walks it — not because Plex
    # un-watched them. It cleared them on every run.
    state, _rows = world
    other = PlexLibrary(plex_key="2", title="Shows", kind="show", enabled=True)
    db.add(other)
    db.flush()
    ep = PlexItem(rating_key="e1", library_id=other.id, kind="episode", title="e1")
    db.add(ep)
    db.flush()
    user_id = db.execute(select(PlexLink.user_id)).scalar_one()
    db.add(PlexState(user_id=user_id, item_id=ep.id, status="watched", synced_to_plex=True))
    _local_states(state)
    sysconfig.set_value(db, "plex_libraries", "1")
    db.commit()
    out = _run(db, monkeypatch, _FakePlex(_LISTED, more=[("show", [], 0)]))
    db.expire_all()
    left = {
        rk
        for rk in db.execute(
            select(PlexItem.rating_key).join(PlexState, PlexState.item_id == PlexItem.id)
        ).scalars()
    }
    assert "m1" not in left  # the walked section still clears
    kept = db.execute(select(PlexState).where(PlexState.item_id == ep.id)).scalar_one_or_none()
    assert kept is not None and kept.status == "watched"
    assert out["cleared"] == 1


def test_a_cut_short_section_is_not_rescued_by_a_complete_one_after_it(db, monkeypatch, world):
    # Each section settles on its own walk: movies cut short (5 said, 3 handed), then a complete
    # shows listing — m1 may simply not have been reached.
    state, rows = world
    _local_states(state)
    db.commit()
    shows = [{"ratingKey": "s1"}]
    out = _run(db, monkeypatch, _FakePlex(_LISTED, total=5, more=[("show", shows, 1)]))
    assert rows()["m1"] == ("watched", 0, True)
    assert out["cleared"] == 0 and out["incomplete"] == 1


def test_a_listing_without_a_total_is_never_complete(db, monkeypatch, world):
    state, rows = world
    _local_states(state)
    ep_id = _watched_episode_in_a_shows_section(db)
    db.commit()
    out = _run(db, monkeypatch, _FakePlex(_LISTED, more=[("show", [], None)]))
    assert _state_of(db, ep_id) == "watched"  # its section never said how much it holds
    assert "m1" not in rows()
    assert out["cleared"] == 1 and out["incomplete"] == 1


def test_an_empty_listing_is_not_a_library_of_unwatched_titles(db, monkeypatch, world):
    state, rows = world
    state("m1", status="watched", synced_to_plex=True)
    db.commit()
    out = _run(db, monkeypatch, _FakePlex([]))
    assert rows() == {"m1": ("watched", 0, True)}
    assert out["cleared"] == 0


def test_a_dirty_resume_is_re_pushed_as_a_timeline_and_a_dirty_reset_as_an_unscrobble(
    db, monkeypatch, world
):
    state, rows = world
    state("m1", status="new", position_seconds=300, synced_to_plex=False)
    state("m2", status="new", position_seconds=2, synced_to_plex=False)  # under the 5 s floor
    db.commit()
    plex = _FakePlex([{"ratingKey": rk} for rk in ("m1", "m2", "m3", "m4", "m5")])
    _run(db, monkeypatch, plex)
    assert ("timeline", "m1", 300_000) in plex.calls
    assert ("unscrobble", "m2") in plex.calls
    assert {rk: synced for rk, (_s, _p, synced) in rows().items()} == {"m1": True, "m2": True}


def test_watch_fields_map_to_a_state_only_when_plex_holds_a_signal():
    to = watch_sync._plex_watch_to_state
    assert to(None, None, None) is None
    assert to(0, 4_000, None) is None  # a stray few seconds is not "in progress"
    assert to(0, 90_000, 1_700_000_000)[:2] == ("new", 90)
    assert to("2", 90_000, 1_700_000_000)[:2] == ("watched", 0)  # watched wins over an offset