danvics/soulsync
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1
soulsync/tests/test_video_database.py
BoulderBadgeDad ba7562c5f2 Watchlist cards: show the channel/playlist video count, not '0 videos' 
The channel card's 'N videos' was actually the count of WISHED videos (0 until you
wish some) — so every channel read '0 videos'. It now shows the REMEMBERED catalog
size (from youtube_channel_videos, which fills in as a channel is enriched/opened),
falling back to 'Channel' when nothing's cached yet. Wished count is kept as a
separate wished_count field. Playlists likewise show their video count (cached when
the playlist is followed-with-videos or opened) instead of a static 'Playlist'.
list_watchlist_channels/playlists return the remembered count; the playlist detail +
follow endpoints cache the list so the count is known. 138 tests green.
2026-06-17 21:13:33 -07:00

1234 lines
67 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""Seam tests for the isolated video database (experimental branch).
These pin the contract that matters: the schema builds with all tables/views,
the no-polymorphic-id CHECK constraints actually reject bad rows, cascades fire,
the derived Watchlist/Wishlist/Calendar views return the right membership, the
settings KV roundtrips — and, critically, that the video DB layer imports
NOTHING from the music database (isolation).
"""
from __future__ import annotations
import sqlite3
from pathlib import Path
import pytest
from database.video_database import VideoDatabase, SCHEMA_VERSION
_MODULE = Path(__file__).resolve().parent.parent / "database" / "video_database.py"
EXPECTED_TABLES = {
"meta", "root_folders", "quality_profiles", "video_settings",
"movies", "shows", "seasons", "episodes", "channels", "channel_videos",
"media_files", "downloads", "activity",
"genres", "movie_genres", "show_genres", "people", "credits",
}
EXPECTED_VIEWS = {"v_watchlist", "v_wishlist", "v_calendar"}
@pytest.fixture()
def db(tmp_path):
return VideoDatabase(database_path=str(tmp_path / "video_library.db"))
# ── schema ──────────────────────────────────────────────────────────────────
def test_schema_builds_with_all_tables_and_views(db):
with db.connect() as conn:
tables = {r[0] for r in conn.execute(
"SELECT name FROM sqlite_master WHERE type='table'")}
views = {r[0] for r in conn.execute(
"SELECT name FROM sqlite_master WHERE type='view'")}
assert EXPECTED_TABLES <= tables, f"missing tables: {EXPECTED_TABLES - tables}"
assert views == EXPECTED_VIEWS
def test_schema_version_recorded(db):
assert db.schema_version == SCHEMA_VERSION
def test_init_is_idempotent(tmp_path):
path = str(tmp_path / "video_library.db")
VideoDatabase(database_path=path)
# Second construction on the same path must not raise or wipe data.
db2 = VideoDatabase(database_path=path)
assert db2.health_check()
def test_health_check_ok(db):
assert db.health_check() is True
# ── no-polymorphic-id CHECK constraints ──────────────────────────────────────
def test_media_file_requires_exactly_one_owner(db):
with db.connect() as conn:
conn.execute("INSERT INTO movies(id,title) VALUES (1,'M')")
conn.execute("INSERT INTO shows(id,title) VALUES (1,'S')")
conn.execute("INSERT INTO seasons(id,show_id,season_number) VALUES (1,1,1)")
conn.execute("INSERT INTO episodes(id,show_id,season_id,season_number,episode_number) "
"VALUES (1,1,1,1,1)")
# zero owners -> reject
with pytest.raises(sqlite3.IntegrityError):
conn.execute("INSERT INTO media_files(relative_path) VALUES ('x.mkv')")
# two owners -> reject
with pytest.raises(sqlite3.IntegrityError):
conn.execute("INSERT INTO media_files(movie_id,episode_id,relative_path) "
"VALUES (1,1,'x.mkv')")
# exactly one -> ok
conn.execute("INSERT INTO media_files(movie_id,relative_path) VALUES (1,'x.mkv')")
def test_download_requires_exactly_one_target(db):
with db.connect() as conn:
conn.execute("INSERT INTO movies(id,title) VALUES (1,'M')")
with pytest.raises(sqlite3.IntegrityError):
conn.execute("INSERT INTO downloads(title) VALUES ('no target')")
conn.execute("INSERT INTO downloads(movie_id,title) VALUES (1,'ok')")
def test_download_status_is_constrained(db):
with db.connect() as conn:
conn.execute("INSERT INTO movies(id,title) VALUES (1,'M')")
with pytest.raises(sqlite3.IntegrityError):
conn.execute("INSERT INTO downloads(movie_id,title,status) VALUES (1,'x','bogus')")
# ── cascades ──────────────────────────────────────────────────────────────────
def test_deleting_show_cascades_to_seasons_and_episodes(db):
with db.connect() as conn:
conn.execute("INSERT INTO shows(id,title) VALUES (1,'S')")
conn.execute("INSERT INTO seasons(id,show_id,season_number) VALUES (1,1,1)")
conn.execute("INSERT INTO episodes(id,show_id,season_id,season_number,episode_number) "
"VALUES (1,1,1,1,1)")
conn.execute("DELETE FROM shows WHERE id=1")
assert conn.execute("SELECT COUNT(*) FROM seasons").fetchone()[0] == 0
assert conn.execute("SELECT COUNT(*) FROM episodes").fetchone()[0] == 0
# ── derived views ─────────────────────────────────────────────────────────────
def test_watchlist_view_is_monitored_shows_and_channels(db):
with db.connect() as conn:
conn.execute("INSERT INTO shows(id,title,monitored) VALUES (1,'Followed',1)")
conn.execute("INSERT INTO shows(id,title,monitored) VALUES (2,'Unfollowed',0)")
conn.execute("INSERT INTO channels(id,youtube_id,title,monitored) VALUES (1,'yt1','Chan',1)")
rows = {(r["kind"], r["title"]) for r in conn.execute("SELECT kind,title FROM v_watchlist")}
assert rows == {("show", "Followed"), ("channel", "Chan")}
def test_wishlist_view_is_wanted_but_missing(db):
with db.connect() as conn:
# wanted movie (monitored, no file) -> in wishlist
conn.execute("INSERT INTO movies(id,title,monitored,has_file) VALUES (1,'Want',1,0)")
# owned movie -> not in wishlist
conn.execute("INSERT INTO movies(id,title,monitored,has_file) VALUES (2,'Own',1,1)")
# aired, monitored, missing episode -> in wishlist; future episode -> not
conn.execute("INSERT INTO shows(id,title) VALUES (1,'S')")
conn.execute("INSERT INTO seasons(id,show_id,season_number) VALUES (1,1,1)")
conn.execute("INSERT INTO episodes(id,show_id,season_id,season_number,episode_number,"
"monitored,has_file,air_date) VALUES (1,1,1,1,1,1,0,'2000-01-01')")
conn.execute("INSERT INTO episodes(id,show_id,season_id,season_number,episode_number,"
"monitored,has_file,air_date) VALUES (2,1,1,1,2,1,0,'2999-01-01')")
rows = [(r["kind"], r["ref_id"]) for r in conn.execute("SELECT kind,ref_id FROM v_wishlist")]
assert ("movie", 1) in rows # wanted movie present
assert ("movie", 2) not in rows # owned movie absent
assert ("episode", 1) in rows # aired+missing episode present
assert ("episode", 2) not in rows # future episode absent
def test_dashboard_stats_empty_is_all_zero(db):
s = db.dashboard_stats()
assert s["library"] == {"movies": 0, "shows": 0, "episodes": 0, "size_bytes": 0}
assert s["downloads"] == {"active": 0, "finished": 0, "speed_bps": 0}
assert s["watchlist"] == 0 and s["wishlist"] == 0
def test_dashboard_stats_counts_content_and_downloads(db):
with db.connect() as conn:
conn.execute("INSERT INTO movies(id,title,monitored,has_file) VALUES (1,'M',1,0)")
# airing library show → on the curated watchlist by default
conn.execute("INSERT INTO shows(id,title,monitored,tmdb_id,status) VALUES (1,'S',1,555,'continuing')")
conn.execute("INSERT INTO seasons(id,show_id,season_number) VALUES (1,1,1)")
conn.execute("INSERT INTO episodes(id,show_id,season_id,season_number,episode_number) "
"VALUES (1,1,1,1,1)")
conn.execute("INSERT INTO media_files(movie_id,relative_path,size_bytes) VALUES (1,'m.mkv',1000)")
conn.execute("INSERT INTO downloads(movie_id,title,status,download_speed_bps) "
"VALUES (1,'d','downloading',500)")
conn.commit()
s = db.dashboard_stats()
assert s["library"] == {"movies": 1, "shows": 1, "episodes": 1, "size_bytes": 1000}
assert s["downloads"]["active"] == 1 and s["downloads"]["speed_bps"] == 500
# watchlist = the airing show (curated default); wishlist is cleared for now
assert s["watchlist"] == 1 and s["wishlist"] == 0
def test_library_selection_roundtrip(db):
assert db.get_library_selection("plex") == {"movies": None, "tv": None}
db.set_library_selection("plex", "Movies", "TV Shows")
assert db.get_library_selection("plex") == {"movies": "Movies", "tv": "TV Shows"}
# Per-server keys don't collide.
assert db.get_library_selection("jellyfin") == {"movies": None, "tv": None}
def test_settings_kv_roundtrip(db):
assert db.get_setting("download_dir", "unset") == "unset"
db.set_setting("download_dir", "/data/video")
assert db.get_setting("download_dir") == "/data/video"
db.set_setting("download_dir", "/data/video2") # upsert
assert db.get_setting("download_dir") == "/data/video2"
# ── scan upserts (server source of truth) ────────────────────────────────────
def test_upsert_movie_inserts_updates_and_attaches_file(db):
mid = db.upsert_movie("plex", {
"server_id": "p1", "title": "Dune", "year": 2021,
"file": {"relative_path": "Dune.mkv", "size_bytes": 2000, "resolution": "2160p"}})
with db.connect() as c:
row = c.execute("SELECT title,year,has_file FROM movies WHERE id=?", (mid,)).fetchone()
assert (row["title"], row["year"], row["has_file"]) == ("Dune", 2021, 1)
f = c.execute("SELECT relative_path,size_bytes,resolution FROM media_files WHERE movie_id=?",
(mid,)).fetchone()
assert (f["relative_path"], f["size_bytes"], f["resolution"]) == ("Dune.mkv", 2000, "2160p")
# Re-scan same server id -> same row, updated fields, file replaced (not duplicated).
mid2 = db.upsert_movie("plex", {
"server_id": "p1", "title": "Dune: Part One", "year": 2021,
"file": {"relative_path": "Dune1.mkv", "size_bytes": 3000}})
assert mid2 == mid
with db.connect() as c:
assert c.execute("SELECT title FROM movies WHERE id=?", (mid,)).fetchone()["title"] == "Dune: Part One"
files = c.execute("SELECT relative_path FROM media_files WHERE movie_id=?", (mid,)).fetchall()
assert [r["relative_path"] for r in files] == ["Dune1.mkv"]
assert db.dashboard_stats()["library"]["movies"] == 1
def test_upsert_show_tree_builds_seasons_episodes_and_prunes(db):
item = {"server_id": "s1", "title": "Show", "seasons": [
{"season_number": 1, "server_id": "se1", "episodes": [
{"episode_number": 1, "title": "E1", "server_id": "ep1", "air_date": "2020-01-01",
"file": {"relative_path": "e1.mkv", "size_bytes": 10}},
{"episode_number": 2, "title": "E2", "server_id": "ep2", "air_date": "2020-01-08"}]}]}
sid = db.upsert_show_tree("plex", item)
with db.connect() as c:
assert c.execute("SELECT COUNT(*) FROM episodes WHERE show_id=?", (sid,)).fetchone()[0] == 2
assert c.execute("SELECT has_file FROM episodes WHERE show_id=? AND episode_number=1",
(sid,)).fetchone()["has_file"] == 1
assert c.execute("SELECT has_file FROM episodes WHERE show_id=? AND episode_number=2",
(sid,)).fetchone()["has_file"] == 0
# Re-scan with E2 removed from the server -> it gets pruned.
item["seasons"][0]["episodes"] = item["seasons"][0]["episodes"][:1]
assert db.upsert_show_tree("plex", item) == sid
with db.connect() as c:
eps = [r["episode_number"] for r in c.execute(
"SELECT episode_number FROM episodes WHERE show_id=?", (sid,)).fetchall()]
assert eps == [1]
def test_upsert_stores_provider_ids(db):
mid = db.upsert_movie("plex", {"server_id": "m1", "title": "Inception",
"tmdb_id": 27205, "imdb_id": "tt1375666"})
with db.connect() as c:
row = c.execute("SELECT tmdb_id, imdb_id FROM movies WHERE id=?", (mid,)).fetchone()
assert (row["tmdb_id"], row["imdb_id"]) == (27205, "tt1375666")
sid = db.upsert_show_tree("plex", {"server_id": "s1", "title": "Show", "tvdb_id": 121361,
"tmdb_id": 1396, "imdb_id": "tt0903747", "seasons": [
{"season_number": 1, "episodes": [{"episode_number": 1, "tvdb_id": 349232}]}]})
with db.connect() as c:
srow = c.execute("SELECT tvdb_id, tmdb_id, imdb_id FROM shows WHERE id=?", (sid,)).fetchone()
erow = c.execute("SELECT tvdb_id FROM episodes WHERE show_id=?", (sid,)).fetchone()
assert (srow["tvdb_id"], srow["tmdb_id"], srow["imdb_id"]) == (121361, 1396, "tt0903747")
assert erow["tvdb_id"] == 349232
def test_show_detail_builds_season_episode_tree_with_rollups(db):
sid = db.upsert_show_tree("plex", {
"server_id": "s1", "title": "Show", "year": 2019, "overview": "A show",
"network": "HBO", "content_rating": "TV-MA", "status": "ended",
"poster_url": "/p.jpg", "seasons": [
{"season_number": 0, "episodes": [{"episode_number": 1, "title": "Special"}]},
{"season_number": 1, "title": "Season One", "episodes": [
{"episode_number": 1, "title": "Pilot", "air_date": "2019-01-01",
"file": {"relative_path": "e1.mkv", "size_bytes": 5}},
{"episode_number": 2, "title": "Two", "air_date": "2019-01-08"}]}]})
# backdrop_url is filled by TMDB enrichment, not the scan — simulate that.
with db.connect() as c:
c.execute("UPDATE shows SET backdrop_url='/b.jpg' WHERE id=?", (sid,))
c.commit()
d = db.show_detail(sid)
assert d["title"] == "Show" and d["network"] == "HBO" and d["status"] == "ended"
assert d["has_poster"] and d["has_backdrop"]
assert (d["episode_total"], d["episode_owned"], d["season_count"]) == (3, 1, 2)
# Season 0 renders as "Specials"; season 1 keeps its title; ordered by number.
assert [s["season_number"] for s in d["seasons"]] == [0, 1]
assert d["seasons"][0]["title"] == "Specials"
s1 = d["seasons"][1]
assert s1["title"] == "Season One"
assert isinstance(s1["id"], int) # season id present for the poster proxy
assert (s1["episode_total"], s1["episode_owned"]) == (2, 1)
assert s1["episodes"][0]["owned"] is True and s1["episodes"][1]["owned"] is False
def test_show_detail_returns_none_for_missing(db):
assert db.show_detail(999999) is None
def test_capture_everything_movie(db):
mid = db.upsert_movie("plex", {
"server_id": "m1", "title": "Dune", "tagline": "Fear is the mind-killer",
"rating": 8.4, "rating_critic": 83, "genres": ["Sci-Fi", "Adventure", "Sci-Fi"]})
d = db.movie_detail(mid)
assert d["tagline"] == "Fear is the mind-killer" and d["rating"] == 8.4 and d["rating_critic"] == 83
assert d["genres"] == ["Adventure", "Sci-Fi"] # deduped + sorted
# Re-upsert with different genres replaces the links (no stale rows).
db.upsert_movie("plex", {"server_id": "m1", "title": "Dune", "genres": ["Drama"]})
assert db.movie_detail(mid)["genres"] == ["Drama"]
def test_capture_everything_show_and_episode_still(db):
sid = db.upsert_show_tree("plex", {
"server_id": "s1", "title": "Show", "tagline": "Tick tock", "rating": 9.1,
"first_air_date": "2015-01-16", "genres": ["Drama", "Mystery"], "seasons": [
{"season_number": 1, "episodes": [
{"episode_number": 1, "title": "Pilot", "still_url": "/ep1.jpg", "rating": 8.0}]}]})
d = db.show_detail(sid)
assert d["tagline"] == "Tick tock" and d["rating"] == 9.1 and d["first_air_date"] == "2015-01-16"
assert d["genres"] == ["Drama", "Mystery"]
ep = d["seasons"][0]["episodes"][0]
assert ep["has_still"] is True and ep["rating"] == 8.0
# Episode still resolves through the image proxy ref (server source from the episode row).
with db.connect() as c:
eid = c.execute("SELECT id FROM episodes WHERE title='Pilot'").fetchone()["id"]
ref = db.get_art_ref("episode", eid, "poster")
assert ref["poster_url"] == "/ep1.jpg" and ref["server_source"] == "plex"
def test_movie_detail_includes_owned_and_file(db):
mid = db.upsert_movie("plex", {
"server_id": "m1", "title": "Dune", "year": 2021, "overview": "Sand",
"tmdb_id": 438631, "poster_url": "/p.jpg",
"file": {"relative_path": "dune.mkv", "size_bytes": 99, "resolution": "2160p"}})
d = db.movie_detail(mid)
assert d["title"] == "Dune" and d["owned"] is True and d["tmdb_id"] == 438631
assert d["file"] and d["file"]["resolution"] == "2160p"
# Full media specs surface for the owned-media block.
assert "video_codec" in d["file"] and "release_source" in d["file"]
assert d["files"] == [d["file"]] # all versions (one here)
# A wishlist movie with no file reports owned False, file None.
mid2 = db.upsert_movie("plex", {"server_id": "m2", "title": "Wanted"})
d2 = db.movie_detail(mid2)
assert d2["owned"] is False and d2["file"] is None and d2["files"] == []
def test_get_art_ref_poster_vs_backdrop(db):
sid = db.upsert_show_tree("plex", {"server_id": "s1", "title": "S", "poster_url": "/p.jpg"})
with db.connect() as c: # backdrop is enrichment-filled, not scanned
c.execute("UPDATE shows SET backdrop_url='/b.jpg' WHERE id=?", (sid,))
c.commit()
assert db.get_art_ref("show", sid, "poster")["poster_url"] == "/p.jpg"
assert db.get_art_ref("show", sid, "backdrop")["poster_url"] == "/b.jpg"
assert db.get_art_ref("show", sid, "bogus") is None
def test_get_art_ref_season_inherits_show_server_source(db):
db.upsert_show_tree("plex", {"server_id": "s1", "title": "S", "seasons": [
{"season_number": 1, "server_id": "se1", "poster_url": "/sp.jpg",
"episodes": [{"episode_number": 1}]}]})
with db.connect() as c:
seid = c.execute("SELECT id FROM seasons WHERE season_number=1").fetchone()["id"]
ref = db.get_art_ref("season", seid, "poster")
# Season carries its own poster + server_id, but inherits the show's source.
assert ref["poster_url"] == "/sp.jpg" and ref["server_source"] == "plex"
assert ref["server_id"] == "se1"
def test_prune_missing_skips_when_over_half_would_be_removed(db):
# >100 movies; a scan that "sees" only a couple must NOT wipe the rest
# (mirrors music's deep-scan 50% safety against partial server failures).
for i in range(150):
db.upsert_movie("plex", {"server_id": "m%d" % i, "title": "M%d" % i})
assert db.prune_missing("movies", "plex", {"m0", "m1"}) == 0
assert db.table_count("movies") == 150
def test_upsert_show_tree_skips_episodes_without_number(db):
sid = db.upsert_show_tree("plex", {"server_id": "s1", "title": "S", "seasons": [
{"season_number": 1, "episodes": [
{"episode_number": 1, "title": "E1"},
{"episode_number": None, "title": "Unmatched special"}]}]})
with db.connect() as c:
eps = [r["episode_number"] for r in c.execute(
"SELECT episode_number FROM episodes WHERE show_id=?", (sid,)).fetchall()]
assert eps == [1]
def test_prune_missing_removes_unseen_top_level(db):
db.upsert_movie("plex", {"server_id": "a", "title": "A"})
db.upsert_movie("plex", {"server_id": "b", "title": "B"})
assert db.prune_missing("movies", "plex", {"a"}) == 1
with db.connect() as c:
ids = {r["server_id"] for r in c.execute("SELECT server_id FROM movies").fetchall()}
assert ids == {"a"}
def test_list_movies_and_shows(db):
db.upsert_movie("plex", {"server_id": "m1", "title": "Zardoz", "year": 1974})
db.upsert_movie("plex", {"server_id": "m2", "title": "Akira", "year": 1988})
db.upsert_show_tree("plex", {"server_id": "s1", "title": "Show", "seasons": [
{"season_number": 1, "episodes": [
{"episode_number": 1, "file": {"relative_path": "e1.mkv"}},
{"episode_number": 2}]}]})
assert [m["title"] for m in db.list_movies()] == ["Akira", "Zardoz"] # title NOCASE sort
shows = db.list_shows()
assert len(shows) == 1
assert (shows[0]["episode_count"], shows[0]["owned_count"]) == (2, 1)
def test_query_library_search_letter_sort_status_pagination(db):
db.upsert_movie("plex", {"server_id": "1", "title": "The Matrix", "year": 1999,
"file": {"relative_path": "x.mkv", "resolution": "1080p"}})
db.upsert_movie("plex", {"server_id": "2", "title": "Akira", "year": 1988}) # wanted
db.upsert_movie("plex", {"server_id": "3", "title": "Avatar", "year": 2009,
"file": {"relative_path": "y.mkv"}})
# Article-aware sort: "The Matrix" files under M.
res = db.query_library("movies")
assert [i["title"] for i in res["items"]] == ["Akira", "Avatar", "The Matrix"]
assert res["pagination"]["total_count"] == 3
assert [i["title"] for i in db.query_library("movies", letter="m")["items"]] == ["The Matrix"]
assert [i["title"] for i in db.query_library("movies", search="aki")["items"]] == ["Akira"]
assert {i["title"] for i in db.query_library("movies", status="owned")["items"]} == {"The Matrix", "Avatar"}
assert [i["title"] for i in db.query_library("movies", status="wanted")["items"]] == ["Akira"]
assert [i["title"] for i in db.query_library("movies", sort="year")["items"]] == ["Avatar", "The Matrix", "Akira"]
# Resolution badge field comes through.
assert db.query_library("movies", search="matrix")["items"][0]["resolution"] == "1080p"
# Pagination.
p1 = db.query_library("movies", limit=2, page=1)
assert len(p1["items"]) == 2 and p1["pagination"]["total_pages"] == 2 and p1["pagination"]["has_next"]
p2 = db.query_library("movies", limit=2, page=2)
assert len(p2["items"]) == 1 and p2["pagination"]["has_prev"]
def test_query_library_shows_status_and_counts(db):
db.upsert_show_tree("plex", {"server_id": "s1", "title": "Owned Show", "seasons": [
{"season_number": 1, "episodes": [{"episode_number": 1, "file": {"relative_path": "e.mkv"}}]}]})
db.upsert_show_tree("plex", {"server_id": "s2", "title": "Wanted Show", "seasons": [
{"season_number": 1, "episodes": [{"episode_number": 1}]}]})
assert [i["title"] for i in db.query_library("shows", status="owned")["items"]] == ["Owned Show"]
assert [i["title"] for i in db.query_library("shows", status="wanted")["items"]] == ["Wanted Show"]
owned = db.query_library("shows", search="Owned")["items"][0]
assert (owned["episode_count"], owned["owned_count"]) == (1, 1)
# ── enrichment plumbing ───────────────────────────────────────────────────────
def test_enrichment_columns_present(db):
with db.connect() as c:
mcols = {r[1] for r in c.execute("PRAGMA table_info(movies)").fetchall()}
scols = {r[1] for r in c.execute("PRAGMA table_info(shows)").fetchall()}
assert {"tmdb_match_status", "tmdb_last_attempted"} <= mcols
assert {"tmdb_match_status", "tvdb_match_status", "tvdb_last_attempted"} <= scols
def test_ensure_columns_is_idempotent(db):
# Running the migration again on an already-migrated DB must not error.
with db.connect() as c:
db._ensure_columns(c)
c.commit()
assert db.enrichment_breakdown("tmdb")["movie"]["pending"] == 0
def test_enrichment_next_pending_then_none_when_fresh(db):
db.upsert_movie("plex", {"server_id": "m1", "title": "A"})
db.upsert_movie("plex", {"server_id": "m2", "title": "B", "tmdb_id": 5})
nxt = db.enrichment_next("tmdb")
assert nxt and nxt["kind"] == "movie"
db.enrichment_apply("tmdb", "movie", nxt["id"], matched=True, external_id=1)
nxt2 = db.enrichment_next("tmdb")
assert nxt2 and nxt2["id"] != nxt["id"]
db.enrichment_apply("tmdb", "movie", nxt2["id"], matched=False)
# both attempted; not_found is fresh (<30d) so nothing is due
assert db.enrichment_next("tmdb") is None
def test_enrichment_apply_matched_sets_id_status_and_metadata(db):
mid = db.upsert_movie("plex", {"server_id": "m1", "title": "A"})
db.enrichment_apply("tmdb", "movie", mid, matched=True, external_id=27205,
metadata={"overview": "O", "backdrop_url": "/b.jpg", "imdb_id": "tt1",
"bogus_col": "x"})
with db.connect() as c:
row = c.execute("SELECT tmdb_id, tmdb_match_status, overview, backdrop_url, imdb_id "
"FROM movies WHERE id=?", (mid,)).fetchone()
assert (row["tmdb_id"], row["tmdb_match_status"]) == (27205, "matched")
assert (row["overview"], row["backdrop_url"], row["imdb_id"]) == ("O", "/b.jpg", "tt1")
def test_enrichment_apply_survives_legacy_unique(db):
# Simulate a pre-existing DB where tvdb_id still carries a UNIQUE index.
with db.connect() as c:
c.execute("CREATE UNIQUE INDEX ux_legacy_shows_tvdb ON shows(tvdb_id)")
c.commit()
a = db.upsert_show_tree("plex", {"server_id": "s1", "title": "A", "seasons": []})
b = db.upsert_show_tree("plex", {"server_id": "s2", "title": "B", "seasons": []})
db.enrichment_apply("tvdb", "show", a, matched=True, external_id=555, metadata={"overview": "OA"})
# b would collide on tvdb_id=555 — must NOT crash; keeps existing id, records the rest.
db.enrichment_apply("tvdb", "show", b, matched=True, external_id=555, metadata={"overview": "OB"})
with db.connect() as c:
ra = c.execute("SELECT tvdb_id, tvdb_match_status FROM shows WHERE id=?", (a,)).fetchone()
rb = c.execute("SELECT tvdb_id, tvdb_match_status, overview FROM shows WHERE id=?", (b,)).fetchone()
assert (ra["tvdb_id"], ra["tvdb_match_status"]) == (555, "matched")
assert rb["tvdb_id"] is None and rb["tvdb_match_status"] == "matched" and rb["overview"] == "OB"
def test_upsert_movie_survives_legacy_unique_tmdb(db):
# Old DBs created movies.tmdb_id UNIQUE; the new model allows the same film in
# >1 library. The scan must STORE the second movie (dropping the colliding id),
# not skip it with an IntegrityError.
with db.connect() as c:
c.execute("CREATE UNIQUE INDEX ux_legacy_movies_tmdb ON movies(tmdb_id)")
c.commit()
a = db.upsert_movie("plex", {"server_id": "m1", "title": "A", "tmdb_id": 548522})
b = db.upsert_movie("plex", {"server_id": "m2", "title": "B", "tmdb_id": 548522})
assert a != b # both rows exist (not skipped)
with db.connect() as c:
ra = c.execute("SELECT tmdb_id FROM movies WHERE id=?", (a,)).fetchone()
rb = c.execute("SELECT title, tmdb_id FROM movies WHERE id=?", (b,)).fetchone()
assert ra["tmdb_id"] == 548522
assert rb["title"] == "B" and rb["tmdb_id"] is None # kept the row, dropped the dup id
def test_upsert_show_survives_legacy_unique_tvdb(db):
with db.connect() as c:
c.execute("CREATE UNIQUE INDEX ux_legacy_shows_tvdb2 ON shows(tvdb_id)")
c.commit()
a = db.upsert_show_tree("plex", {"server_id": "s1", "title": "A", "tvdb_id": 9000, "seasons": []})
b = db.upsert_show_tree("plex", {"server_id": "s2", "title": "B", "tvdb_id": 9000, "seasons": []})
assert a != b
with db.connect() as c:
rb = c.execute("SELECT title, tvdb_id FROM shows WHERE id=?", (b,)).fetchone()
assert rb["title"] == "B" and rb["tvdb_id"] is None
def test_enrichment_breakdown_unmatched_retry(db):
a = db.upsert_movie("plex", {"server_id": "m1", "title": "A"})
b = db.upsert_movie("plex", {"server_id": "m2", "title": "B"})
db.enrichment_apply("tmdb", "movie", a, matched=True, external_id=1)
db.enrichment_apply("tmdb", "movie", b, matched=False)
assert db.enrichment_breakdown("tmdb")["movie"] == {"matched": 1, "not_found": 1, "errors": 0, "pending": 0}
un = db.enrichment_unmatched("tmdb", "movie", status="not_found")
assert [i["title"] for i in un["items"]] == ["B"] and un["total"] == 1
assert db.enrichment_retry("tmdb", "movie", scope="failed") == 1
assert db.enrichment_breakdown("tmdb")["movie"]["pending"] == 1
def test_enrichment_backfills_only_gaps_never_clobbers_server(db):
# Server gave overview + a genre; enrichment must fill the EMPTY fields
# (tagline/rating) but leave the server's overview + genres untouched.
mid = db.upsert_movie("plex", {"server_id": "m1", "title": "Dune",
"overview": "server overview", "genres": ["Sci-Fi"]})
db.enrichment_apply("tmdb", "movie", mid, matched=True, external_id=438631, metadata={
"overview": "tmdb overview", "tagline": "Fear is the mind-killer",
"rating": 8.4, "genres": ["Drama"]})
d = db.movie_detail(mid)
assert d["overview"] == "server overview" # NOT clobbered
assert d["tagline"] == "Fear is the mind-killer" and d["rating"] == 8.4 # gaps filled
assert d["genres"] == ["Sci-Fi"] # had genres → enrichment left them
def test_enrichment_backfills_genres_when_item_has_none(db):
mid = db.upsert_movie("plex", {"server_id": "m1", "title": "X"}) # no genres
db.enrichment_apply("tmdb", "movie", mid, matched=True, external_id=1,
metadata={"genres": ["Drama", "Comedy"]})
assert db.movie_detail(mid)["genres"] == ["Comedy", "Drama"]
def test_enrichment_backfills_cast_and_crew(db):
sid = db.upsert_show_tree("plex", {"server_id": "s1", "title": "S", "seasons": []})
db.enrichment_apply("tmdb", "show", sid, matched=True, external_id=1, metadata={
"cast": [
{"name": "Aidan Gillen", "tmdb_id": 39388, "character": "James Cole",
"photo_url": "https://img/ag.jpg"},
{"name": "Amanda Schull", "tmdb_id": 84223, "character": "Cassie"}],
"crew": [{"name": "Terry Matalas", "tmdb_id": 1, "job": "Creator"}]})
d = db.show_detail(sid)
assert [c["name"] for c in d["cast"]] == ["Aidan Gillen", "Amanda Schull"] # billing order
assert d["cast"][0]["character"] == "James Cole" and d["cast"][0]["photo"] == "https://img/ag.jpg"
assert d["crew"] == [{"name": "Terry Matalas", "job": "Creator", "tmdb_id": 1}]
# Clearlogo backfills like the other art (gap-only) and rides in the payload.
db.enrichment_apply("tmdb", "show", sid, matched=True, external_id=1,
metadata={"logo_url": "https://img/logo.png"})
assert db.show_detail(sid)["logo"] == "https://img/logo.png"
# People are deduped across titles by tmdb_id.
mid = db.upsert_movie("plex", {"server_id": "m1", "title": "M"})
db.enrichment_apply("tmdb", "movie", mid, matched=True, external_id=2,
metadata={"cast": [{"name": "Aidan Gillen", "tmdb_id": 39388, "character": "X"}]})
with db.connect() as c:
assert c.execute("SELECT COUNT(*) FROM people WHERE tmdb_id=39388").fetchone()[0] == 1
def test_enrichment_does_not_clobber_existing_credits(db):
sid = db.upsert_show_tree("plex", {"server_id": "s1", "title": "S", "seasons": []})
db.enrichment_apply("tmdb", "show", sid, matched=True, external_id=1,
metadata={"cast": [{"name": "First", "tmdb_id": 1}]})
db.enrichment_apply("tmdb", "show", sid, matched=True, external_id=1,
metadata={"cast": [{"name": "Second", "tmdb_id": 2}]})
assert [c["name"] for c in db.show_detail(sid)["cast"]] == ["First"] # kept, not replaced
def test_enrichment_backfills_season_posters_only_when_missing(db):
# Season 1 has no art (server didn't provide it); season 2 has server art.
sid = db.upsert_show_tree("plex", {"server_id": "s1", "title": "S", "seasons": [
{"season_number": 1, "episodes": [{"episode_number": 1}]},
{"season_number": 2, "poster_url": "/server.jpg", "episodes": [{"episode_number": 1}]}]})
db.enrichment_apply("tmdb", "show", sid, matched=True, external_id=1, metadata={"seasons": [
{"season_number": 1, "poster_url": "https://image.tmdb.org/s1.jpg"},
{"season_number": 2, "poster_url": "https://image.tmdb.org/s2.jpg"}]})
with db.connect() as c:
rows = {r["season_number"]: r["poster_url"] for r in c.execute(
"SELECT season_number, poster_url FROM seasons WHERE show_id=?", (sid,)).fetchall()}
assert rows[1] == "https://image.tmdb.org/s1.jpg" # gap filled from TMDB
assert rows[2] == "/server.jpg" # server art kept, not clobbered
assert db.show_detail(sid)["seasons"][0]["has_poster"] is True
def test_backfill_episodes_gap_only_and_season_overview(db):
sid = db.upsert_show_tree("plex", {"server_id": "s1", "title": "S", "seasons": [
{"season_number": 1, "episodes": [
{"episode_number": 1, "overview": "server overview"}, # has overview already
{"episode_number": 2}]}]}) # bare
n = db.backfill_episodes(sid, 1, [
{"episode_number": 1, "still_url": "/e1.jpg", "overview": "tmdb overview", "rating": 8.0},
{"episode_number": 2, "still_url": "/e2.jpg", "overview": "tmdb e2", "rating": 7.0},
], season_overview="Season one")
assert n == 2
eps = {e["episode_number"]: e for e in db.show_detail(sid)["seasons"][0]["episodes"]}
assert eps[1]["has_still"] is True and eps[1]["overview"] == "server overview" # overview kept
assert eps[2]["has_still"] is True and eps[2]["overview"] == "tmdb e2" # gap filled
with db.connect() as c:
assert c.execute("SELECT overview FROM seasons WHERE show_id=? AND season_number=1",
(sid,)).fetchone()["overview"] == "Season one"
def test_backfill_inserts_missing_episodes_as_unowned(db):
# Server has only E1; the provider's season has E1-E3 → E2/E3 are inserted
# MISSING (has_file=0) so the page shows what we have AND what we need.
sid = db.upsert_show_tree("plex", {"server_id": "s1", "title": "S", "seasons": [
{"season_number": 1, "episodes": [{"episode_number": 1, "file": {"relative_path": "e1.mkv"}}]}]})
db.backfill_episodes(sid, 1, [
{"episode_number": 1, "title": "One"},
{"episode_number": 2, "title": "Two", "air_date": "2020-01-08"},
{"episode_number": 3, "title": "Three"}])
s1 = db.show_detail(sid)["seasons"][0]
assert (s1["episode_total"], s1["episode_owned"]) == (3, 1) # full list, 1 owned
by = {e["episode_number"]: e for e in s1["episodes"]}
assert by[1]["owned"] is True and by[2]["owned"] is False and by[3]["owned"] is False
assert by[2]["title"] == "Two" and by[2]["air_date"] == "2020-01-08"
def test_apply_ratings_and_payload(db):
mid = db.upsert_movie("plex", {"server_id": "m1", "title": "Dune"})
db.apply_ratings("movie", mid, {"imdb_rating": 8.4, "rt_rating": 95, "metacritic": 74})
d = db.movie_detail(mid)
assert (d["imdb_rating"], d["rt_rating"], d["metacritic"]) == (8.4, 95, 74)
sid = db.upsert_show_tree("plex", {"server_id": "s1", "title": "S", "seasons": []})
db.apply_ratings("show", sid, {"imdb_rating": 9.1}) # partial is fine
sd = db.show_detail(sid)
assert sd["imdb_rating"] == 9.1 and sd["rt_rating"] is None
def test_episodes_synced_flag_drives_lazy_refresh(db):
sid = db.upsert_show_tree("plex", {"server_id": "s1", "title": "S", "seasons": []})
assert db.show_detail(sid)["episodes_synced"] is False # → lazy refresh will run
db.mark_episodes_synced(sid)
assert db.show_detail(sid)["episodes_synced"] is True # → won't re-cascade
def test_backfill_creates_fully_missing_season(db):
sid = db.upsert_show_tree("plex", {"server_id": "s1", "title": "S", "seasons": [
{"season_number": 1, "episodes": [{"episode_number": 1}]}]})
db.backfill_episodes(sid, 2, [{"episode_number": 1, "title": "S2E1"}],
season_poster="https://img/s2.jpg") # season 2 not on the server
d = db.show_detail(sid)
s2 = [s for s in d["seasons"] if s["season_number"] == 2][0]
assert (s2["episode_total"], s2["episode_owned"]) == (1, 0) and s2["has_poster"] is True
def test_missing_episodes_survive_a_rescan_prune(db):
sid = db.upsert_show_tree("plex", {"server_id": "s1", "title": "S", "seasons": [
{"season_number": 1, "episodes": [{"episode_number": 1, "file": {"relative_path": "e1.mkv"}}]}]})
db.backfill_episodes(sid, 1, [{"episode_number": 1}, {"episode_number": 2}, {"episode_number": 3}])
# Re-scan: the server still only reports E1; the prune must NOT remove the
# enrichment-added missing episodes (server_id NULL).
db.upsert_show_tree("plex", {"server_id": "s1", "title": "S", "seasons": [
{"season_number": 1, "episodes": [{"episode_number": 1, "file": {"relative_path": "e1.mkv"}}]}]})
with db.connect() as c:
nums = [r["episode_number"] for r in c.execute(
"SELECT episode_number FROM episodes WHERE show_id=? ORDER BY episode_number", (sid,)).fetchall()]
assert nums == [1, 2, 3]
def test_breakdown_reports_episode_art_coverage_for_tmdb(db):
db.upsert_show_tree("plex", {"server_id": "s1", "title": "S", "seasons": [
{"season_number": 1, "episodes": [{"episode_number": 1, "still_url": "/e1.jpg"},
{"episode_number": 2}]}]})
bd = db.enrichment_breakdown("tmdb")
assert bd["episode"]["matched"] == 1 and bd["episode"]["pending"] == 1
assert bd["episode"].get("coverage_only") is True
# TVDB doesn't cascade episodes → no episode coverage entry.
assert "episode" not in db.enrichment_breakdown("tvdb")
def test_unmatched_lists_episodes_missing_art(db):
db.upsert_show_tree("plex", {"server_id": "s1", "title": "Show", "seasons": [
{"season_number": 1, "episodes": [{"episode_number": 1, "title": "Has", "still_url": "/e1.jpg"},
{"episode_number": 2, "title": "Missing"}]}]})
res = db.enrichment_unmatched("tmdb", "episode", status="unmatched")
assert res["total"] == 1 and "Missing" in res["items"][0]["title"]
def test_error_status_is_distinct_and_retryable_in_ui(db):
a = db.upsert_movie("plex", {"server_id": "m1", "title": "A"})
db.enrichment_apply("tmdb", "movie", a, matched=False, error=True)
bd = db.enrichment_breakdown("tmdb")["movie"]
assert bd == {"matched": 0, "not_found": 0, "errors": 1, "pending": 0}
# Errored items surface in the modal's "unmatched" list so they can be retried.
assert db.enrichment_unmatched("tmdb", "movie", status="unmatched")["total"] == 1
# ...but NOT in the strict 'not_found'-only view.
assert db.enrichment_unmatched("tmdb", "movie", status="not_found")["total"] == 0
# "Retry all failed" re-queues errors too (back to pending/NULL).
assert db.enrichment_retry("tmdb", "movie", scope="failed") == 1
assert db.enrichment_breakdown("tmdb")["movie"]["pending"] == 1
def test_tvdb_is_shows_only(db):
# TVDB enriches shows, never movies. The breakdown must not advertise a
# movie bucket, and asking for tvdb movies returns empty (never garbage) —
# this is what keeps the Manage-Workers modal from showing a Movies view
# for TVDB.
assert "movie" not in db.enrichment_breakdown("tvdb")
assert "show" in db.enrichment_breakdown("tvdb")
db.upsert_movie("plex", {"server_id": "m1", "title": "A"})
assert db.enrichment_unmatched("tvdb", "movie") == {"items": [], "total": 0}
# ── isolation: the video DB imports nothing from music ───────────────────────
def test_video_db_module_imports_nothing_from_music():
src = _MODULE.read_text(encoding="utf-8")
for line in src.splitlines():
stripped = line.strip()
if stripped.startswith("import ") or stripped.startswith("from "):
assert "music" not in stripped.lower(), f"music import leaked in: {stripped!r}"
def test_video_db_uses_distinct_default_path_and_env():
src = _MODULE.read_text(encoding="utf-8")
assert "video_library.db" in src
assert "VIDEO_DATABASE_PATH" in src # distinct env var from music's DATABASE_PATH
assert "music_library.db" not in src
def test_servers_do_not_commingle_in_reads(db):
"""Per-server isolation (mirrors the music side): the SAME movie/show on Plex
and Jellyfin are separate rows, and scoped reads only ever return the active
server's data — so a Jellyfin scan never shows up alongside Plex."""
# Same title on both servers → two distinct rows.
db.upsert_movie("plex", {"server_id": "p1", "title": "Dune", "tmdb_id": 438631, "year": 2021})
db.upsert_movie("jellyfin", {"server_id": "j1", "title": "Dune", "tmdb_id": 438631, "year": 2021})
db.upsert_show_tree("plex", {"server_id": "ps", "title": "Severance", "tmdb_id": 95396, "seasons": []})
db.upsert_show_tree("jellyfin", {"server_id": "js", "title": "Severance", "tmdb_id": 95396, "seasons": []})
# query_library is scoped per server (and unscoped == all).
assert db.query_library("movies", server_source="plex")["pagination"]["total_count"] == 1
assert db.query_library("movies", server_source="jellyfin")["pagination"]["total_count"] == 1
assert db.query_library("movies")["pagination"]["total_count"] == 2
assert db.query_library("shows", server_source="plex")["pagination"]["total_count"] == 1
# dashboard counts are scoped; None counts everything.
assert db.dashboard_stats("plex")["library"]["movies"] == 1
assert db.dashboard_stats("jellyfin")["library"]["movies"] == 1
assert db.dashboard_stats()["library"]["movies"] == 2
# ownership lookup resolves to the row on the ACTIVE server, not the other.
plex_id = db.library_id_for_tmdb("movie", 438631, "plex")
jelly_id = db.library_id_for_tmdb("movie", 438631, "jellyfin")
assert plex_id is not None and jelly_id is not None and plex_id != jelly_id
def test_prune_is_scoped_to_one_server(db):
"""A deep scan of one server must never prune the other server's rows."""
db.upsert_movie("plex", {"server_id": "p1", "title": "Dune"})
db.upsert_movie("jellyfin", {"server_id": "j1", "title": "Arrival"})
# Deep-scan Plex sees nothing → prunes only Plex rows, leaves Jellyfin intact.
db.prune_missing("movies", "plex", seen_ids=[])
assert db.query_library("movies", server_source="plex")["pagination"]["total_count"] == 0
assert db.query_library("movies", server_source="jellyfin")["pagination"]["total_count"] == 1
# ── user watchlist (shows + people) ─────────────────────────────────────
def test_watchlist_add_list_remove(db):
assert db.add_to_watchlist("show", 1399, "Game of Thrones", poster_url="/p.jpg", library_id=7) is True
assert db.add_to_watchlist("person", 287, "Brad Pitt") is True
assert db.add_to_watchlist("movie", 1, "nope") is False # wrong kind
assert db.add_to_watchlist("show", 0, "no id") is False # no tmdb id
rows = db.list_watchlist()
assert {r["kind"] for r in rows} == {"show", "person"}
assert db.list_watchlist("person")[0]["title"] == "Brad Pitt"
assert db.remove_from_watchlist("show", 1399) is True
assert db.remove_from_watchlist("show", 1399) is True # idempotent (mute tombstone)
assert all(r["tmdb_id"] != 1399 for r in db.list_watchlist("show")) # gone from the list
def test_watchlist_airing_library_shows_are_watched_by_default(db):
# Two library shows: one still airing, one ended. No explicit watchlist rows.
with db.connect() as conn:
conn.execute("INSERT INTO shows(id,title,tmdb_id,status) VALUES (1,'Severance',95396,'continuing')")
conn.execute("INSERT INTO shows(id,title,tmdb_id,status) VALUES (2,'The Wire',1438,'ended')")
conn.commit()
# The airing show is on the watchlist by default; the ended one is not.
shows = db.list_watchlist("show")
assert [s["tmdb_id"] for s in shows] == [95396]
assert shows[0].get("auto") is True and shows[0]["library_id"] == 1
assert db.watchlist_state("show", [95396, 1438]) == {95396: True} # ended one absent
# Un-following the airing show (mute tombstone) drops it from the default…
db.remove_from_watchlist("show", 95396)
assert db.list_watchlist("show") == []
assert db.watchlist_state("show", [95396]) == {}
# …and explicitly re-following brings it back (clears the mute).
db.add_to_watchlist("show", 95396, "Severance")
assert db.watchlist_state("show", [95396]) == {95396: True}
def test_watchlist_reAdd_is_upsert_and_coalesces_library_id(db):
db.add_to_watchlist("show", 1399, "GoT", poster_url="/a.jpg", library_id=7)
# Re-add WITHOUT library_id/poster (e.g. from a TMDB search card) must not
# wipe the previously-known library_id/poster, but should refresh the title.
db.add_to_watchlist("show", 1399, "Game of Thrones")
rows = db.list_watchlist("show")
assert len(rows) == 1 # no duplicate
assert rows[0]["title"] == "Game of Thrones" # refreshed
assert rows[0]["library_id"] == 7 and rows[0]["poster_url"] == "/a.jpg" # preserved
def test_watchlist_state_and_counts(db):
db.add_to_watchlist("show", 1399, "GoT")
db.add_to_watchlist("show", 1396, "Breaking Bad")
db.add_to_watchlist("person", 287, "Brad Pitt")
assert db.watchlist_state("show", [1399, 1396, 9999]) == {1399: True, 1396: True}
assert db.watchlist_state("show", []) == {}
assert db.watchlist_state("person", [287]) == {287: True}
assert db.watchlist_counts() == {"show": 2, "person": 1, "total": 3}
def test_query_watchlist_paginates_and_searches(db):
for i in range(1, 8):
db.add_to_watchlist("person", 100 + i, "Person %d" % i)
db.add_to_watchlist("person", 200, "Brad Pitt")
res = db.query_watchlist("person", page=1, limit=3)
assert len(res["items"]) == 3
assert res["pagination"]["total_count"] == 8 and res["pagination"]["total_pages"] == 3
assert res["pagination"]["has_next"] is True and res["pagination"]["has_prev"] is False
# search (case-insensitive title contains)
res2 = db.query_watchlist("person", search="brad", limit=60)
assert len(res2["items"]) == 1 and res2["items"][0]["title"] == "Brad Pitt"
assert res2["pagination"]["total_count"] == 1
# page beyond range clamps to the last page
assert db.query_watchlist("person", page=99, limit=3)["pagination"]["page"] == 3
# ── wishlist (movies + episodes; show/season are bulk ops over episodes) ──────
def test_wishlist_movie_add_is_idempotent(db):
assert db.add_movie_to_wishlist(603, "The Matrix", year=1999, poster_url="/m.jpg") is True
db.add_movie_to_wishlist(603, "The Matrix", year=1999) # re-add → upsert, not dup
res = db.query_wishlist("movie")
assert len(res["items"]) == 1
m = res["items"][0]
assert m["tmdb_id"] == 603 and m["year"] == 1999 and m["poster_url"] == "/m.jpg" # poster kept
assert db.wishlist_counts()["movie"] == 1
def test_wishlist_episodes_group_into_show_tree(db):
n = db.add_episodes_to_wishlist(1396, "Breaking Bad", [
{"season_number": 1, "episode_number": 1, "title": "Pilot", "air_date": "2008-01-20"},
{"season_number": 1, "episode_number": 2, "title": "Cat's in the Bag"},
{"season_number": 2, "episode_number": 1, "title": "Seven Thirty-Seven"}],
poster_url="/bb.jpg")
assert n == 3
db.add_episodes_to_wishlist(1396, "Breaking Bad", [
{"season_number": 1, "episode_number": 1, "title": "Pilot"}]) # re-add → no dup
res = db.query_wishlist("show")
assert res["pagination"]["total_count"] == 1 # one show
show = res["items"][0]
assert show["tmdb_id"] == 1396 and show["wanted"] == 3 and show["done"] == 0
assert [s["season_number"] for s in show["seasons"]] == [1, 2]
assert [e["episode_number"] for e in show["seasons"][0]["episodes"]] == [1, 2]
counts = db.wishlist_counts()
assert counts == {"movie": 0, "show": 1, "episode": 3, "total": 3}
def test_wishlist_remove_scopes(db):
db.add_movie_to_wishlist(603, "The Matrix")
db.add_episodes_to_wishlist(1396, "Breaking Bad", [
{"season_number": 1, "episode_number": 1}, {"season_number": 1, "episode_number": 2},
{"season_number": 2, "episode_number": 1}])
# remove one episode
assert db.remove_from_wishlist("episode", tmdb_id=1396, season_number=1, episode_number=2) == 1
assert db.wishlist_counts()["episode"] == 2
# remove a whole season
assert db.remove_from_wishlist("season", tmdb_id=1396, season_number=1) == 1 # only S1E1 left in S1
assert db.wishlist_counts()["episode"] == 1
# remove the whole show (its remaining episodes)
assert db.remove_from_wishlist("show", tmdb_id=1396) == 1
assert db.wishlist_counts()["show"] == 0
# movie still there; remove it
assert db.remove_from_wishlist("movie", tmdb_id=603) == 1
assert db.wishlist_counts()["total"] == 0
def test_wishlist_movie_and_episode_same_tmdb_dont_collide(db):
# A movie and a show could share a tmdb id across namespaces — the partial
# uniques must keep them independent.
db.add_movie_to_wishlist(42, "Some Movie")
db.add_episodes_to_wishlist(42, "Some Show", [{"season_number": 1, "episode_number": 1}])
c = db.wishlist_counts()
assert c["movie"] == 1 and c["episode"] == 1
def test_wishlist_state_hydration(db):
db.add_movie_to_wishlist(603, "The Matrix")
db.add_episodes_to_wishlist(1396, "Breaking Bad", [
{"season_number": 1, "episode_number": 1}, {"season_number": 2, "episode_number": 3}])
st = db.wishlist_state(movie_ids=[603, 604], show_tmdb_id=1396)
assert st["movies"] == {603}
assert st["episodes"] == {"1_1", "2_3"}
def test_wishlist_query_search_and_paging(db):
for i in range(5):
db.add_movie_to_wishlist(100 + i, "Movie %d" % i)
db.add_movie_to_wishlist(900, "Zebra")
assert db.query_wishlist("movie", search="zeb")["pagination"]["total_count"] == 1
p1 = db.query_wishlist("movie", page=1, limit=2)
assert len(p1["items"]) == 2 and p1["pagination"]["total_pages"] == 3
assert db.query_wishlist("movie", page=99, limit=2)["pagination"]["page"] == 3 # clamps
def test_wishlist_keys_for_shows(db):
db.add_episodes_to_wishlist(1396, "BB", [{"season_number": 1, "episode_number": 1},
{"season_number": 2, "episode_number": 3}])
db.add_episodes_to_wishlist(1399, "GoT", [{"season_number": 1, "episode_number": 1}])
keys = db.wishlist_keys_for_shows([1396, 1399, 9999])
assert keys[1396] == {"1_1", "2_3"} and keys[1399] == {"1_1"} and 9999 not in keys
assert db.wishlist_keys_for_shows([]) == {}
def test_wishlist_query_sort(db):
db.add_episodes_to_wishlist(1, "Alpha", [{"season_number": 1, "episode_number": 1}]) # 1 ep
db.add_episodes_to_wishlist(2, "Zeta", [{"season_number": 1, "episode_number": i} for i in range(5)]) # 5 eps
# CURRENT_TIMESTAMP collides within a second — pin distinct add-times so the
# FIFO vs newest ordering is deterministic.
with db.connect() as c:
c.execute("UPDATE video_wishlist SET date_added='2024-01-01 00:00:00' WHERE tmdb_id=1")
c.execute("UPDATE video_wishlist SET date_added='2024-01-02 00:00:00' WHERE tmdb_id=2")
c.commit()
# most-wanted first
w = [s["title"] for s in db.query_wishlist("show", sort="wanted")["items"]]
assert w[0] == "Zeta"
# FIFO (oldest first) vs recently-added (newest first)
assert [s["title"] for s in db.query_wishlist("show", sort="oldest")["items"]] == ["Alpha", "Zeta"]
assert [s["title"] for s in db.query_wishlist("show", sort="added")["items"]] == ["Zeta", "Alpha"]
# A–Z
az = [s["title"] for s in db.query_wishlist("show", sort="title")["items"]]
assert az == ["Alpha", "Zeta"]
# movies A–Z
db.add_movie_to_wishlist(10, "Banana"); db.add_movie_to_wishlist(11, "Apple")
m = [x["title"] for x in db.query_wishlist("movie", sort="title")["items"]]
assert m == ["Apple", "Banana"]
def test_wishlist_episode_still_roundtrips(db):
db.add_episodes_to_wishlist(1396, "BB", [
{"season_number": 1, "episode_number": 1, "title": "Pilot", "still_url": "https://img/e1.jpg"},
{"season_number": 1, "episode_number": 2}]) # no still
eps = db.query_wishlist("show")["items"][0]["seasons"][0]["episodes"]
by = {e["episode_number"]: e for e in eps}
assert by[1]["still_url"] == "https://img/e1.jpg" and by[2]["still_url"] is None
def test_wishlist_art_backfill_targets_and_set(db):
db.add_episodes_to_wishlist(1396, "BB", [
{"season_number": 1, "episode_number": 1, "still_url": "/have.jpg", "season_poster_url": "/s1.jpg"},
{"season_number": 1, "episode_number": 2},
{"season_number": 2, "episode_number": 1}])
tset = {(t["tmdb_id"], t["season_number"]) for t in db.wishlist_art_backfill_targets()}
assert (1396, 1) in tset and (1396, 2) in tset # S1 missing a still, S2 missing both
assert db.set_wishlist_still(1396, 1, 2, "/new.jpg") is True
assert db.set_wishlist_still(1396, 1, 1, "/x.jpg") is False # won't clobber an existing still
assert db.set_wishlist_season_poster(1396, 2, "/s2.jpg") == 1 # fills the season's episodes
by_season = {s["season_number"]: s for s in db.query_wishlist("show")["items"][0]["seasons"]}
assert by_season[1]["poster_url"] == "/s1.jpg" and by_season[2]["poster_url"] == "/s2.jpg"
def test_wishlist_episode_overview_roundtrips_and_backfills(db):
db.add_episodes_to_wishlist(1396, "BB", [
{"season_number": 1, "episode_number": 1, "overview": "The one where it begins."},
{"season_number": 1, "episode_number": 2}])
eps = {e["episode_number"]: e for e in db.query_wishlist("show")["items"][0]["seasons"][0]["episodes"]}
assert eps[1]["overview"] == "The one where it begins." and eps[2]["overview"] is None
# episode missing an overview shows up as a backfill target
assert (1396, 1) in {(t["tmdb_id"], t["season_number"]) for t in db.wishlist_art_backfill_targets()}
assert db.set_wishlist_episode_overview(1396, 1, 2, "Filled in.") is True
assert db.set_wishlist_episode_overview(1396, 1, 1, "nope") is False # won't clobber
# ── YouTube channels (bridged onto watchlist/wishlist) ───────────────────────
from database.video_database import youtube_surrogate_id
def test_youtube_surrogate_id_is_stable_and_distinct():
a = youtube_surrogate_id("UCPlayStation")
assert a == youtube_surrogate_id("UCPlayStation") # deterministic
assert a > 0 and a < (1 << 63) # fits SQLite INTEGER
assert a != youtube_surrogate_id("UCGoodMythical") # distinct ids → distinct
def test_follow_channel_lists_and_hydrates(db):
ch = {"youtube_id": "UCPlay", "title": "PlayStation", "avatar_url": "http://a/p.jpg"}
assert db.add_channel_to_watchlist(ch) is True
chans = db.list_watchlist_channels()
assert len(chans) == 1
assert chans[0]["youtube_id"] == "UCPlay" and chans[0]["title"] == "PlayStation"
assert chans[0]["poster_url"] == "http://a/p.jpg" and chans[0]["video_count"] == 0
# hydration
assert db.channel_watch_state(["UCPlay", "UCnope"]) == {"UCPlay": True}
# idempotent re-follow refreshes title, no duplicate
db.add_channel_to_watchlist({"youtube_id": "UCPlay", "title": "PlayStation US"})
chans = db.list_watchlist_channels()
assert len(chans) == 1 and chans[0]["title"] == "PlayStation US"
def test_unfollow_channel_removes_row(db):
db.add_channel_to_watchlist({"youtube_id": "UCPlay", "title": "PlayStation"})
assert db.remove_channel_from_watchlist("UCPlay") is True
assert db.list_watchlist_channels() == []
assert db.channel_watch_state(["UCPlay"]) == {}
def test_youtube_wishlist_nebula_shape_year_as_season(db):
"""Channel = show, YEAR = season, video = episode — exact TV-nebula shape."""
ch = {"youtube_id": "UCPlay", "title": "PlayStation", "avatar_url": "http://a/p.jpg"}
vids = [
{"youtube_id": "v1", "title": "Old Trailer", "published_at": "2023-01-01",
"thumbnail_url": "http://t/1.jpg", "description": "older"},
{"youtube_id": "v2", "title": "New State of Play", "published_at": "2024-06-01",
"thumbnail_url": "http://t/2.jpg", "description": "newer"},
{"youtube_id": "v2b", "title": "Mid 2024", "published_at": "2024-02-01", "thumbnail_url": "http://t/2b.jpg"},
{"youtube_id": "v3", "title": "Undated", "thumbnail_url": "http://t/3.jpg"},
]
assert db.add_videos_to_wishlist(ch, vids) == 4
res = db.query_youtube_wishlist()
assert res["pagination"]["total_count"] == 1
grp = res["items"][0]
assert grp["kind"] == "channel" and grp["source"] == "youtube"
assert grp["youtube_id"] == "UCPlay" and grp["title"] == "PlayStation"
assert grp["poster_url"] == "http://a/p.jpg" and grp["wanted"] == 4
assert isinstance(grp["tmdb_id"], int) # surrogate the nebula keys on
# seasons = years, newest first: 2024, 2023, then 0 (undated)
assert [se["season_number"] for se in grp["seasons"]] == [2024, 2023, 0]
y2024 = grp["seasons"][0]
# newest video in the year is episode 1; its still seeds the season poster
assert [e["title"] for e in y2024["episodes"]] == ["New State of Play", "Mid 2024"]
assert [e["episode_number"] for e in y2024["episodes"]] == [1, 2]
assert y2024["poster_url"] == "http://t/2.jpg"
e1 = y2024["episodes"][0]
assert e1["source_id"] == "v2" and e1["overview"] == "newer" and e1["air_date"] == "2024-06-01"
def test_add_videos_is_idempotent_per_video(db):
ch = {"youtube_id": "UCPlay", "title": "PlayStation"}
db.add_videos_to_wishlist(ch, [{"youtube_id": "v1", "title": "A", "published_at": "2024-01-01"}])
db.add_videos_to_wishlist(ch, [{"youtube_id": "v1", "title": "A (updated)", "published_at": "2024-01-01"},
{"youtube_id": "v2", "title": "B", "published_at": "2024-02-01"}])
grp = db.query_youtube_wishlist()["items"][0]
assert grp["wanted"] == 2
titles = {}
for se in grp["seasons"]:
for e in se["episodes"]:
titles[e["source_id"]] = e["title"]
assert titles == {"v1": "A (updated)", "v2": "B"}
def test_youtube_video_wish_state_hydrates(db):
db.add_videos_to_wishlist({"youtube_id": "UCx", "title": "X"},
[{"youtube_id": "a", "title": "A"}, {"youtube_id": "b", "title": "B"}])
assert db.youtube_video_wish_state(["a", "b", "c"]) == {"a", "b"}
assert db.remove_one_video_from_wishlist("a") == 1
assert db.youtube_video_wish_state(["a", "b"]) == {"b"}
def test_youtube_counts_and_removal_scopes(db):
a = {"youtube_id": "UCa", "title": "Chan A"}
b = {"youtube_id": "UCb", "title": "Chan B"}
db.add_videos_to_wishlist(a, [{"youtube_id": "a1", "title": "A1"},
{"youtube_id": "a2", "title": "A2"}])
db.add_videos_to_wishlist(b, [{"youtube_id": "b1", "title": "B1"}])
assert db.youtube_wishlist_counts() == {"channel": 2, "video": 3}
# remove one video
assert db.remove_youtube_from_wishlist("video", "a1") == 1
assert db.youtube_wishlist_counts() == {"channel": 2, "video": 2}
# remove a whole channel
assert db.remove_youtube_from_wishlist("channel", "UCa") == 1 # only a2 left
assert db.youtube_wishlist_counts() == {"channel": 1, "video": 1}
def test_youtube_rows_do_not_disturb_tmdb_counts(db):
"""The bridge must not leak into the existing movie/episode + show/person counts."""
db.add_movie_to_wishlist(101, "A Movie", year=2020)
db.add_episodes_to_wishlist(202, "A Show", [{"season_number": 1, "episode_number": 1}])
db.add_to_watchlist("show", 303, "Watched Show")
db.add_channel_to_watchlist({"youtube_id": "UCx", "title": "Chan"})
db.add_videos_to_wishlist({"youtube_id": "UCx", "title": "Chan"},
[{"youtube_id": "x1", "title": "X1"}])
# existing shapes unchanged
assert db.wishlist_counts() == {"movie": 1, "show": 1, "episode": 1, "total": 2}
assert db.watchlist_counts() == {"show": 1, "person": 0, "total": 1}
# youtube counts live on their own surface
assert db.youtube_wishlist_counts() == {"channel": 1, "video": 1}
assert db.list_watchlist_channels()[0]["wished_count"] == 1 # 1 wished video
assert db.list_watchlist_channels()[0]["video_count"] == 0 # remembered catalog (nothing cached)
def test_upgrade_from_pre_source_schema(tmp_path):
"""Regression: an existing pre-bridge DB (no source/source_id/parent_source_id
columns) must upgrade cleanly. The source_id partial indexes can't live in the
schema executescript (it runs BEFORE the column ALTERs), or init blows up with
'no such column: source_id' and the whole video DB fails to initialize."""
path = str(tmp_path / "video_library.db")
conn = sqlite3.connect(path)
conn.executescript(
"""
CREATE TABLE video_watchlist (
id INTEGER PRIMARY KEY AUTOINCREMENT, kind TEXT NOT NULL, tmdb_id INTEGER NOT NULL,
title TEXT NOT NULL, poster_url TEXT, library_id INTEGER,
state TEXT NOT NULL DEFAULT 'follow', date_added TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
UNIQUE(kind, tmdb_id));
CREATE TABLE video_wishlist (
id INTEGER PRIMARY KEY AUTOINCREMENT, kind TEXT NOT NULL, tmdb_id INTEGER NOT NULL,
title TEXT NOT NULL, poster_url TEXT, year INTEGER, season_number INTEGER,
episode_number INTEGER, episode_title TEXT, still_url TEXT, episode_overview TEXT,
season_poster_url TEXT, air_date TEXT, status TEXT NOT NULL DEFAULT 'wanted',
library_id INTEGER, server_source TEXT, date_added TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP);
""")
conn.commit()
conn.close()
db = VideoDatabase(database_path=path) # must upgrade in place, no raise
with db.connect() as c:
cols = {r[1] for r in c.execute("PRAGMA table_info(video_wishlist)")}
assert {"source", "source_id", "parent_source_id"} <= cols
wcols = {r[1] for r in c.execute("PRAGMA table_info(video_watchlist)")}
assert {"source", "source_id"} <= wcols
idx = {r[0] for r in c.execute("SELECT name FROM sqlite_master WHERE type='index'")}
assert "idx_video_wishlist_video" in idx and "idx_video_wishlist_channel" in idx
# the youtube path actually works on the upgraded DB
assert db.add_videos_to_wishlist({"youtube_id": "UCx", "title": "Chan"},
[{"youtube_id": "x1", "title": "X1"}]) == 1
assert db.youtube_wishlist_counts() == {"channel": 1, "video": 1}
def test_set_wishlist_channel_poster_backfills_avatar(db):
# videos added without an avatar (flat listing omitted it) → poster_url null
db.add_videos_to_wishlist({"youtube_id": "UCx", "title": "X"},
[{"youtube_id": "a", "title": "A", "published_at": "2024-01-01"},
{"youtube_id": "b", "title": "B", "published_at": "2024-02-01"}])
grp = db.query_youtube_wishlist()["items"][0]
assert not grp["poster_url"]
# backfilling the resolved avatar fills every row → the orb gets its poster
assert db.set_wishlist_channel_poster("UCx", "http://yt/avatar.jpg") == 2
grp = db.query_youtube_wishlist()["items"][0]
assert grp["poster_url"] == "http://yt/avatar.jpg"
def test_video_date_cache_roundtrip(db):
assert db.cache_video_dates([{"youtube_id": "a", "published_at": "2024-06-01"},
{"youtube_id": "b", "published_at": "2023-01-01"},
{"youtube_id": "c", "published_at": ""}]) == 2 # blank skipped
assert db.get_video_dates(["a", "b", "c", "d"]) == {"a": "2024-06-01", "b": "2023-01-01"}
# upsert refreshes
db.cache_video_dates([{"youtube_id": "a", "published_at": "2025-12-31"}])
assert db.get_video_dates(["a"]) == {"a": "2025-12-31"}
def test_channel_enrichment_tracking(db):
db.add_videos_to_wishlist({"youtube_id": "UCx", "title": "X"},
[{"youtube_id": "a", "title": "A"}, {"youtube_id": "b", "title": "B"}])
assert set(db.wishlisted_video_ids_for_channel("UCx")) == {"a", "b"}
assert db.channel_dates_enriched_recently("UCx") is False
# A THIN run (few dates → proxies were down) is still 'recent' briefly, but
# retries soon (15-min window) regardless of the full within_hours param.
db.mark_channel_dates_enriched("UCx", date_count=2)
assert db.channel_dates_enriched_recently("UCx") is True
# A GOOD run (>=15 dates) honours the full window param (so within_hours=0 → not recent).
db.mark_channel_dates_enriched("UCx", date_count=50)
assert db.channel_dates_enriched_recently("UCx", within_hours=24) is True
assert db.channel_dates_enriched_recently("UCx", within_hours=0) is False # window respected
def test_legacy_enrichment_rows_upgrade(db):
# A pre-InnerTube row (method NULL) with otherwise-good coverage must NOT be
# locked for 24h — it re-enriches once so it upgrades to the full catalog.
db.mark_channel_dates_enriched("UCleg", date_count=50, method=None)
assert db.channel_dates_enriched_recently("UCleg") is False
# After the InnerTube re-run, the normal window applies again (no churn).
db.mark_channel_dates_enriched("UCleg", date_count=240, method="innertube")
assert db.channel_dates_enriched_recently("UCleg") is True
def test_playlist_watchlist_roundtrip(db):
assert db.add_playlist_to_watchlist({"playlist_id": "PLx", "title": "Mix", "thumbnail_url": "t"}) is True
assert db.playlist_watch_state(["PLx", "PLy"]) == {"PLx": True}
pls = db.list_watchlist_playlists()
assert len(pls) == 1 and pls[0]["playlist_id"] == "PLx" and pls[0]["title"] == "Mix"
# channels and playlists are separate kinds — they don't bleed into each other
db.add_channel_to_watchlist({"youtube_id": "UCx", "title": "Chan"})
assert {p["playlist_id"] for p in db.list_watchlist_playlists()} == {"PLx"}
assert db.channel_watch_state(["UCx"]) == {"UCx": True} and db.playlist_watch_state(["UCx"]) == {}
assert db.remove_playlist_from_watchlist("PLx") is True
assert db.list_watchlist_playlists() == [] and db.playlist_watch_state(["PLx"]) == {}
# name/title required
assert db.add_playlist_to_watchlist({"playlist_id": "PLn"}) is False
def test_remembered_channel_videos_and_meta(db):
# Cache a list (out of date order) + a date for one of them.
db.cache_video_dates([{"youtube_id": "b", "published_at": "2020-05-01"}])
db.cache_channel_videos("UCc", [
{"youtube_id": "a", "title": "A", "thumbnail_url": "ta"}, # undated → sorts last
{"youtube_id": "b", "title": "B", "thumbnail_url": "tb"}])
got = db.get_channel_videos("UCc")
assert [v["youtube_id"] for v in got] == ["b", "a"] # dated first, undated last
assert got[0]["published_at"] == "2020-05-01" and got[1]["published_at"] is None
assert got[0]["title"] == "B" and got[0]["thumbnail_url"] == "tb"
# Upsert refreshes fields without dropping the row; COALESCE keeps a non-null title.
db.cache_channel_videos("UCc", [{"youtube_id": "a", "title": "A2", "thumbnail_url": None}])
a = next(v for v in db.get_channel_videos("UCc") if v["youtube_id"] == "a")
assert a["title"] == "A2" and a["thumbnail_url"] == "ta" # title updated, thumb kept
assert db.get_channel_videos("UNKNOWN") == []
# Metadata round-trips with tags decoded.
assert db.get_channel_meta("UCc") is None
db.cache_channel_meta("UCc", {"title": "Chan", "handle": "@c", "avatar_url": "av",
"subscriber_count": 1234, "tags": ["x", "y"]})
m = db.get_channel_meta("UCc")
assert m["title"] == "Chan" and m["subscriber_count"] == 1234 and m["tags"] == ["x", "y"]
Reference in a new issue View git blame Copy permalink