peter
79704b0001
playlistNotFound was mapped to "gone" everywhere: exploring a channel without uploads (its uploads list answers not found) and a list deleted mid bulk pull both said a playlist no longer exists. The reason is now "gone" only on single-playlist routes; explore reads it as an empty back-catalogue, and the bulk pull skips that list and syncs the rest.
260 lines
9.1 KiB
Python
260 lines
9.1 KiB
Python
"""An empty answer is not an emptied library (E32 S2).
|
|
|
|
Every mirror here reconciles: what the remote no longer lists is removed locally. The catalog
|
|
(`_prune_stale`) and the subscriptions (`should_prune`) had long refused to do that on an empty
|
|
listing that contradicts what we hold — a 200 with no items is a blip far more often than a user
|
|
who removed everything. Two mirrors did not: the Plex collections (a row carries the user-set
|
|
`editable` flag, and every member its keys) and the YouTube playlist mirrors. These hold both — and,
|
|
one level down (E32 review), a collection's members and a playlist's items: an empty answer that
|
|
contradicts the remote's own count is not believed either.
|
|
"""
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from app.models import (
|
|
Channel,
|
|
Playlist,
|
|
PlaylistItem,
|
|
PlexCollection,
|
|
PlexItem,
|
|
PlexLibrary,
|
|
Video,
|
|
)
|
|
from app.plex import sync as plex_sync
|
|
from app.sync import playlists as plsync
|
|
from app.youtube.client import YouTubeError
|
|
|
|
|
|
class _Plex:
|
|
def __init__(self, collections: list[dict], children: dict[str, list[dict]]):
|
|
self._collections = collections
|
|
self._children = children
|
|
|
|
def collections(self, section_key):
|
|
return self._collections
|
|
|
|
def collection_children(self, rk):
|
|
return self._children.get(rk, [])
|
|
|
|
def metadata(self, rk):
|
|
return next((c for c in self._collections if c.get("ratingKey") == rk), {})
|
|
|
|
|
|
def _library_with_a_collection(db):
|
|
lib = PlexLibrary(plex_key="4", title="Movies", kind="movie")
|
|
db.add(lib)
|
|
db.flush()
|
|
db.add(PlexCollection(rating_key="c1", library_id=lib.id, title="Bond", editable=True))
|
|
movie = PlexItem(
|
|
rating_key="m1", library_id=lib.id, kind="movie", title="Dr. No", collection_keys=["c1"]
|
|
)
|
|
db.add(movie)
|
|
db.commit()
|
|
return lib, movie
|
|
|
|
|
|
def test_an_empty_collection_listing_keeps_the_collections_and_their_members(db):
|
|
lib, movie = _library_with_a_collection(db)
|
|
stats = {"collections": 0}
|
|
plex_sync._sync_collections(db, _Plex([], {}), lib, stats)
|
|
db.commit()
|
|
db.expire_all()
|
|
col = db.execute(select(PlexCollection).where(PlexCollection.rating_key == "c1")).scalar_one()
|
|
assert col.editable is True
|
|
assert db.get(PlexItem, movie.id).collection_keys == ["c1"]
|
|
|
|
|
|
def test_a_listing_with_something_in_it_still_prunes_what_it_no_longer_has(db):
|
|
lib, movie = _library_with_a_collection(db)
|
|
stats = {"collections": 0}
|
|
other = {"ratingKey": "c2", "title": "Alien"}
|
|
plex_sync._sync_collections(db, _Plex([other], {"c2": []}), lib, stats)
|
|
db.commit()
|
|
db.expire_all()
|
|
keys = set(db.execute(select(PlexCollection.rating_key)).scalars())
|
|
assert keys == {"c2"}
|
|
assert db.get(PlexItem, movie.id).collection_keys is None
|
|
|
|
|
|
def test_no_members_from_a_collection_plex_counts_as_non_empty_keeps_its_members(db):
|
|
lib, movie = _library_with_a_collection(db)
|
|
stats = {"collections": 0}
|
|
bond = {"ratingKey": "c1", "title": "Bond", "childCount": 2}
|
|
plex_sync._sync_collections(db, _Plex([bond], {"c1": []}), lib, stats)
|
|
db.commit()
|
|
db.expire_all()
|
|
assert db.get(PlexItem, movie.id).collection_keys == ["c1"]
|
|
|
|
|
|
def test_a_collection_plex_counts_as_empty_does_lose_its_members(db):
|
|
lib, movie = _library_with_a_collection(db)
|
|
stats = {"collections": 0}
|
|
bond = {"ratingKey": "c1", "title": "Bond", "childCount": 0}
|
|
plex_sync._sync_collections(db, _Plex([bond], {"c1": []}), lib, stats)
|
|
db.commit()
|
|
db.expire_all()
|
|
assert db.get(PlexItem, movie.id).collection_keys is None
|
|
|
|
|
|
@pytest.mark.parametrize(("count", "kept"), [(2, ["c1"]), (0, None)])
|
|
def test_a_collection_resync_believes_an_empty_member_list_only_with_a_zero_count(db, count, kept):
|
|
lib, movie = _library_with_a_collection(db)
|
|
bond = {"ratingKey": "c1", "title": "Bond", "childCount": count}
|
|
plex_sync.resync_collection(db, _Plex([bond], {"c1": []}), lib, "c1")
|
|
db.commit()
|
|
db.expire_all()
|
|
assert db.get(PlexItem, movie.id).collection_keys == kept
|
|
|
|
|
|
class _YT:
|
|
def __init__(self, playlists: list[dict], snippet: dict | None = None):
|
|
self._playlists = playlists
|
|
self._snippet = snippet
|
|
|
|
def __call__(self, *_):
|
|
return self
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_):
|
|
return False
|
|
|
|
def iter_my_playlists(self):
|
|
return self._playlists
|
|
|
|
def iter_playlist_items_with_ids(self, _ytid):
|
|
return []
|
|
|
|
def iter_my_playlist_video_ids(self, _ytid):
|
|
return []
|
|
|
|
def get_playlist_snippet(self, _ytid):
|
|
return self._snippet
|
|
|
|
def get_videos(self, _ids):
|
|
return []
|
|
|
|
|
|
def test_an_empty_playlist_answer_keeps_the_youtube_mirrors(db, make_user, monkeypatch):
|
|
user = make_user()
|
|
db.add(Playlist(user_id=user.id, name="Mixes", source="youtube", yt_playlist_id="PL1"))
|
|
db.commit()
|
|
monkeypatch.setattr(plsync, "YouTubeClient", _YT([]))
|
|
plsync.pull_user_playlists(db, user)
|
|
db.expire_all()
|
|
names = db.execute(select(Playlist.name).where(Playlist.user_id == user.id)).scalars().all()
|
|
assert "Mixes" in names
|
|
|
|
|
|
def test_a_playlist_answer_with_something_in_it_still_drops_a_vanished_mirror(
|
|
db, make_user, monkeypatch
|
|
):
|
|
user = make_user()
|
|
db.add(Playlist(user_id=user.id, name="Mixes", source="youtube", yt_playlist_id="PL1"))
|
|
db.commit()
|
|
monkeypatch.setattr(plsync, "YouTubeClient", _YT([{"id": "PL2", "title": "New"}]))
|
|
plsync.pull_user_playlists(db, user)
|
|
db.expire_all()
|
|
ytids = set(
|
|
db.execute(select(Playlist.yt_playlist_id).where(Playlist.source == "youtube")).scalars()
|
|
)
|
|
assert "PL1" not in ytids
|
|
|
|
|
|
class _YTOneGone(_YT):
|
|
"""A list deleted on YouTube between the listing and the read of its items."""
|
|
|
|
def iter_my_playlist_video_ids(self, ytid):
|
|
if ytid == "PLgone":
|
|
raise YouTubeError("m", reason="playlistNotFound", path="playlistItems")
|
|
return []
|
|
|
|
|
|
def test_a_list_deleted_mid_pull_is_skipped_and_the_rest_still_sync(db, make_user, monkeypatch):
|
|
# Not "this playlist no longer exists" for a bulk pull the user never named a list in, and not
|
|
# a failure for every other list either (E32 review round 2): the next pull drops the mirror.
|
|
user = make_user()
|
|
db.commit()
|
|
listing = [
|
|
{"id": "PLgone", "title": "Gone", "item_count": 3},
|
|
{"id": "PL2", "title": "Kept", "item_count": 0},
|
|
]
|
|
monkeypatch.setattr(plsync, "YouTubeClient", _YTOneGone(listing))
|
|
assert plsync.pull_user_playlists(db, user) == {"synced": 1}
|
|
db.expire_all()
|
|
ytids = set(
|
|
db.execute(select(Playlist.yt_playlist_id).where(Playlist.user_id == user.id)).scalars()
|
|
)
|
|
assert ytids == {"PL2"}
|
|
|
|
|
|
def _mirror_with_an_item(db, user, **kw):
|
|
db.add(Channel(id="UC_pl", title="C"))
|
|
db.flush()
|
|
db.add(Video(id="v1", channel_id="UC_pl", title="One"))
|
|
pl = Playlist(user_id=user.id, name="Mixes", yt_playlist_id="PL1", **kw)
|
|
db.add(pl)
|
|
db.flush()
|
|
db.add(PlaylistItem(playlist_id=pl.id, video_id="v1", position=0))
|
|
db.commit()
|
|
return pl
|
|
|
|
|
|
def _items(db, pl):
|
|
db.expire_all()
|
|
return (
|
|
db.execute(select(PlaylistItem.video_id).where(PlaylistItem.playlist_id == pl.id))
|
|
.scalars()
|
|
.all()
|
|
)
|
|
|
|
|
|
def test_no_items_from_a_playlist_youtube_counts_as_non_empty_keeps_the_mirror_items(
|
|
db, make_user, monkeypatch
|
|
):
|
|
user = make_user()
|
|
pl = _mirror_with_an_item(db, user, source="youtube")
|
|
listing = [{"id": "PL1", "title": "Mixes", "item_count": 1}]
|
|
monkeypatch.setattr(plsync, "YouTubeClient", _YT(listing))
|
|
plsync.pull_user_playlists(db, user)
|
|
assert _items(db, pl) == ["v1"]
|
|
|
|
|
|
def test_a_playlist_youtube_counts_as_empty_is_emptied(db, make_user, monkeypatch):
|
|
user = make_user()
|
|
pl = _mirror_with_an_item(db, user, source="youtube")
|
|
listing = [{"id": "PL1", "title": "Mixes", "item_count": 0}]
|
|
monkeypatch.setattr(plsync, "YouTubeClient", _YT(listing))
|
|
plsync.pull_user_playlists(db, user)
|
|
assert _items(db, pl) == []
|
|
|
|
|
|
def test_a_reset_to_youtube_refuses_a_blank_listing_and_keeps_the_items(db, make_user, monkeypatch):
|
|
user = make_user()
|
|
pl = _mirror_with_an_item(db, user, source="youtube")
|
|
monkeypatch.setattr(plsync, "YouTubeClient", _YT([], {"title": "Mixes", "item_count": 1}))
|
|
with pytest.raises(YouTubeError) as err:
|
|
plsync.repull_playlist(db, user, pl)
|
|
db.rollback()
|
|
assert err.value.reason == "emptyAnswer"
|
|
assert _items(db, pl) == ["v1"]
|
|
|
|
|
|
def test_a_reset_of_a_deleted_playlist_says_it_is_gone(db, make_user, monkeypatch):
|
|
user = make_user()
|
|
pl = _mirror_with_an_item(db, user, source="youtube")
|
|
monkeypatch.setattr(plsync, "YouTubeClient", _YT([], None))
|
|
with pytest.raises(YouTubeError) as err:
|
|
plsync.repull_playlist(db, user, pl)
|
|
assert err.value.reason == "playlistNotFound"
|
|
|
|
|
|
def test_a_push_plan_refuses_a_blank_listing_rather_than_reading_it_as_all_deleted(db, make_user):
|
|
user = make_user()
|
|
pl = _mirror_with_an_item(db, user, source="local")
|
|
with pytest.raises(YouTubeError) as err:
|
|
plsync.plan_merge(db, user, pl, live=([], {"title": "Mixes", "item_count": 1}))
|
|
assert err.value.reason == "emptyAnswer"
|