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 User
from app.plex.models import PlexItem, PlexLibrary, PlexSeason, PlexShow, PlexState
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