siftlode/backend/tests/test_plex_sync_prune.py

"""Catalog prune: a sync now deletes rows for titles that disappeared from Plex,
diffing each section's freshly-listed rating_keys against the stored ones. The prune is GUARDED on a
non-empty listing so a transient Plex glitch (an empty page) can never wipe a whole library.

DB lane (`db` fixture) — skipped without Postgres. Drives `_sync_movies` / `_sync_shows` directly
with a stub Plex (the same seam the overlay test uses), so no live server is needed.
"""

from sqlalchemy import select

from app.models import (
    PlexItem,
    PlexLibrary,
    PlexSeason,
    PlexShow,
    PlexState,
    User,
)
from app.plex.sync import (
    _TYPE_EPISODE,
    _TYPE_MOVIE,
    _TYPE_SEASON,
    _TYPE_SHOW,
    _sync_movies,
    _sync_shows,
)


class _StubPlex:
    """Returns canned section listings per Plex item type — the current 'truth' the sync reconciles
    the stored rows against. `_paginate` is the only method the sync calls on it."""

    def __init__(self, by_type: dict[int, list[dict]]):
        self.by_type = by_type

    def section_items(self, key, item_type, start, size):
        items = self.by_type.get(item_type, [])
        return items[start : start + size], len(items)


class _TruncatedPlex:
    """Simulates a truncated listing (a transient Plex hiccup): it reports a large `total` but returns
    only one short page, so pagination stops BEFORE reaching total. Prune must treat this as an
    incomplete fetch and delete nothing — the un-listed rows are present, just not mentioned."""

    def __init__(self, items: list[dict], claimed_total: int):
        self._items = items
        self._total = claimed_total

    def section_items(self, key, item_type, start, size):
        if start == 0:
            return list(self._items), self._total  # one short page, total says there's more
        return [], self._total


def _seed_movies(db):
    lib = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True)
    db.add(lib)
    db.flush()
    m1 = PlexItem(rating_key="m1", library_id=lib.id, kind="movie", title="Kept Movie")
    m2 = PlexItem(rating_key="m2", library_id=lib.id, kind="movie", title="Gone Movie")
    db.add_all([m1, m2])
    user = User(email="prune@example.com")
    db.add(user)
    db.flush()
    # A watch-state on EACH movie: the kept one's must survive, the pruned one's must cascade away.
    db.add_all(
        [
            PlexState(user_id=user.id, item_id=m1.id, status="watched"),
            PlexState(user_id=user.id, item_id=m2.id, status="watched"),
        ]
    )
    db.commit()
    return lib, m1.id, m2.id, user.id


def test_movie_prune_deletes_disappeared_titles_and_cascades_state(db):
    lib, m1_id, _m2_id, _user_id = _seed_movies(db)
    # Plex now lists only m1 — m2 disappeared.
    stub = _StubPlex({_TYPE_MOVIE: [{"ratingKey": "m1", "title": "Kept Movie"}]})
    stats = {"movies": 0, "pruned": 0}

    _sync_movies(db, stub, lib, stats)
    db.commit()

    live = {r.rating_key for r in db.execute(select(PlexItem)).scalars()}
    assert live == {"m1"}, "the disappeared movie must be pruned"
    assert stats["pruned"] == 1
    # The kept movie's watch-state survives; the pruned movie's cascaded away.
    states = {s.item_id for s in db.execute(select(PlexState)).scalars()}
    assert states == {m1_id}


def test_empty_listing_prunes_nothing(db):
    """The safety guard: a section that lists ZERO items (a real empty library, or a transient Plex
    glitch we can't distinguish) must leave the stored rows untouched — never mass-delete."""
    lib, _m1_id, _m2_id, _user_id = _seed_movies(db)
    stub = _StubPlex({_TYPE_MOVIE: []})  # Plex returned nothing this run
    stats = {"movies": 0, "pruned": 0}

    _sync_movies(db, stub, lib, stats)
    db.commit()

    live = {r.rating_key for r in db.execute(select(PlexItem)).scalars()}
    assert live == {"m1", "m2"}, "an empty listing must not prune anything"
    assert stats["pruned"] == 0


def test_truncated_listing_prunes_nothing(db):
    """A partial fetch (Plex reports 5 items but the listing returns only 1 then stops) must NOT prune
    the 4 it failed to mention — the completeness guard, not just the empty-listing guard."""
    lib, m1_id, _m2_id, _user_id = _seed_movies(db)
    # Only m1 is listed, but total claims 5 → the walk never reaches total → incomplete → no prune.
    stub = _TruncatedPlex([{"ratingKey": "m1", "title": "Kept Movie"}], claimed_total=5)
    stats = {"movies": 0, "pruned": 0}

    _sync_movies(db, stub, lib, stats)
    db.commit()

    live = {r.rating_key for r in db.execute(select(PlexItem)).scalars()}
    assert live == {"m1", "m2"}, "a truncated listing must not prune the un-listed rows"
    assert stats["pruned"] == 0


def test_show_prune_removes_stale_show_season_and_episode(db):
    lib = PlexLibrary(plex_key="2", title="Shows", kind="show", enabled=True)
    db.add(lib)
    db.flush()
    # Stored: one kept show (s1 -> se1 -> e1) and one stale show (s2 -> se2 -> e2) that Plex dropped.
    s1 = PlexShow(rating_key="s1", library_id=lib.id, title="Kept Show")
    s2 = PlexShow(rating_key="s2", library_id=lib.id, title="Gone Show")
    db.add_all([s1, s2])
    db.flush()
    se1 = PlexSeason(rating_key="se1", show_id=s1.id, season_number=1)
    se2 = PlexSeason(rating_key="se2", show_id=s2.id, season_number=1)
    db.add_all([se1, se2])
    db.flush()
    e1 = PlexItem(
        rating_key="e1",
        library_id=lib.id,
        kind="episode",
        title="Kept Ep",
        show_id=s1.id,
        season_id=se1.id,
        season_number=1,
        episode_number=1,
    )
    e2 = PlexItem(
        rating_key="e2",
        library_id=lib.id,
        kind="episode",
        title="Gone Ep",
        show_id=s2.id,
        season_id=se2.id,
        season_number=1,
        episode_number=1,
    )
    db.add_all([e1, e2])
    db.commit()

    # Plex now lists only s1's subtree.
    stub = _StubPlex(
        {
            _TYPE_SHOW: [{"ratingKey": "s1", "title": "Kept Show"}],
            _TYPE_SEASON: [
                {"ratingKey": "se1", "parentRatingKey": "s1", "index": 1, "title": "Season 1"}
            ],
            _TYPE_EPISODE: [
                {
                    "ratingKey": "e1",
                    "grandparentRatingKey": "s1",
                    "parentRatingKey": "se1",
                    "parentIndex": 1,
                    "index": 1,
                    "title": "Kept Ep",
                }
            ],
        }
    )
    stats = {"shows": 0, "seasons": 0, "episodes": 0, "pruned": 0}

    _sync_shows(db, stub, lib, stats)
    db.commit()

    shows = {r.rating_key for r in db.execute(select(PlexShow)).scalars()}
    seasons = {r.rating_key for r in db.execute(select(PlexSeason)).scalars()}
    eps = {r.rating_key for r in db.execute(select(PlexItem)).scalars()}
    assert shows == {"s1"}, "the stale show must be pruned"
    assert seasons == {"se1"}, "the stale season must be pruned"
    assert eps == {"e1"}, "the stale episode must be pruned"
    assert stats["pruned"] == 3  # s2 + se2 + e2