b2dbd6dec5
Settings → Video Source shows which server video uses (✓ Plex/Jellyfin), a Plex/Jellyfin picker when both are connected, or a clear 'connect Plex or Jellyfin' message when neither (Navidrome/Standalone are music-only and not offered). The Library shows a non-breaking 'no video server' banner + disables Scan until one is connected. Detail Pages prefs moved into their own 'Video Preferences' group. /api/video/server GET+POST drives it.
293 lines
14 KiB
Python
293 lines
14 KiB
Python
"""Seam tests for the isolated /api/video blueprint (experimental branch).
|
|
|
|
Verifies the blueprint builds with its route, the dashboard endpoint returns
|
|
real (zeroed) JSON against an empty video.db, and that the video API package
|
|
imports nothing from the music side.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from flask import Flask
|
|
|
|
|
|
def _make_client(tmp_path):
|
|
# Inject a tmp-backed DB directly so the endpoint never falls back to the
|
|
# real default path (no stray database/video_library.db in the repo).
|
|
import api.video as videoapi
|
|
from database.video_database import VideoDatabase
|
|
videoapi._video_db = VideoDatabase(database_path=str(tmp_path / "video_library.db"))
|
|
app = Flask(__name__)
|
|
app.register_blueprint(videoapi.create_video_blueprint(), url_prefix="/api/video")
|
|
return app.test_client(), videoapi
|
|
|
|
|
|
def test_blueprint_exposes_dashboard_route():
|
|
from api.video import create_video_blueprint
|
|
app = Flask(__name__)
|
|
app.register_blueprint(create_video_blueprint(), url_prefix="/api/video")
|
|
rules = {r.rule for r in app.url_map.iter_rules()}
|
|
assert "/api/video/dashboard" in rules
|
|
assert "/api/video/scan/request" in rules
|
|
assert "/api/video/scan/status" in rules
|
|
assert "/api/video/scan/stop" in rules
|
|
assert "/api/video/library" in rules
|
|
assert "/api/video/libraries" in rules
|
|
assert "/api/video/server" in rules
|
|
assert any(r.startswith("/api/video/poster/") for r in rules)
|
|
assert "/api/video/enrichment/services" in rules
|
|
assert "/api/video/enrichment/<service>/status" in rules
|
|
assert "/api/video/enrichment/<service>/unmatched" in rules
|
|
assert "/api/video/enrichment/config" in rules
|
|
assert "/api/video/enrichment/<service>/test" in rules
|
|
assert "/api/video/detail/show/<int:show_id>" in rules
|
|
assert "/api/video/detail/movie/<int:movie_id>" in rules
|
|
assert "/api/video/detail/show/<int:show_id>/refresh-art" in rules
|
|
assert "/api/video/detail/movie/<int:movie_id>/refresh-art" in rules
|
|
assert "/api/video/detail/<kind>/<int:item_id>/extras" in rules
|
|
assert "/api/video/search" in rules
|
|
assert "/api/video/trending" in rules
|
|
assert "/api/video/tmdb/<kind>/<int:tmdb_id>" in rules
|
|
assert "/api/video/tmdb/show/<int:tv_id>/season/<int:season_number>" in rules
|
|
assert "/api/video/person/<int:tmdb_id>" in rules
|
|
assert "/api/video/episode/<int:tmdb_id>/<int:season>/<int:episode>" in rules
|
|
assert any(r.startswith("/api/video/backdrop/") for r in rules)
|
|
assert "/api/video/img" in rules
|
|
|
|
|
|
def test_img_proxy_rejects_non_tmdb(tmp_path):
|
|
client, _ = _make_client(tmp_path)
|
|
assert client.get("/api/video/img?u=https://evil.example.com/x.jpg").status_code == 404
|
|
assert client.get("/api/video/img").status_code == 404
|
|
|
|
|
|
def test_search_endpoint_empty_query(tmp_path):
|
|
client, _ = _make_client(tmp_path)
|
|
resp = client.get("/api/video/search?q=")
|
|
assert resp.status_code == 200
|
|
assert resp.get_json() == {"results": [], "query": ""}
|
|
|
|
|
|
def test_search_endpoint_uses_engine(tmp_path, monkeypatch):
|
|
client, _ = _make_client(tmp_path)
|
|
|
|
class FakeEngine:
|
|
def search(self, q): return [{"kind": "movie", "tmdb_id": 1, "title": "Dune", "library_id": None}]
|
|
monkeypatch.setattr("core.video.enrichment.engine.get_video_enrichment_engine",
|
|
lambda: FakeEngine())
|
|
body = client.get("/api/video/search?q=dune").get_json()
|
|
assert body["query"] == "dune" and body["results"][0]["title"] == "Dune"
|
|
|
|
|
|
def test_tmdb_detail_endpoint(tmp_path, monkeypatch):
|
|
client, _ = _make_client(tmp_path)
|
|
|
|
class FakeEngine:
|
|
def tmdb_detail(self, kind, tid): return {"source": "tmdb", "kind": kind, "id": tid, "title": "X"}
|
|
monkeypatch.setattr("core.video.enrichment.engine.get_video_enrichment_engine",
|
|
lambda: FakeEngine())
|
|
resp = client.get("/api/video/tmdb/movie/438631")
|
|
assert resp.status_code == 200 and resp.get_json()["source"] == "tmdb"
|
|
assert client.get("/api/video/tmdb/bogus/1").status_code == 400
|
|
|
|
|
|
def test_omdb_key_change_retries_unrated(tmp_path, monkeypatch):
|
|
client, videoapi = _make_client(tmp_path)
|
|
db = videoapi._video_db
|
|
mid = db.upsert_movie("plex", {"server_id": "m1", "title": "A", "imdb_id": "tt1"})
|
|
db.apply_ratings("movie", mid, {}) # burned: synced, but no rating
|
|
assert db.ratings_next() is None # not pending
|
|
monkeypatch.setattr("core.video.enrichment.engine.rebuild_video_enrichment_engine", lambda: None)
|
|
resp = client.post("/api/video/enrichment/config", json={"omdb_api_key": "NEWKEY"})
|
|
assert resp.status_code == 200
|
|
assert db.ratings_next() is not None # new key → re-queued for rating
|
|
|
|
|
|
def test_show_detail_endpoint(tmp_path):
|
|
client, videoapi = _make_client(tmp_path)
|
|
try:
|
|
sid = videoapi._video_db.upsert_show_tree("plex", {
|
|
"server_id": "s1", "title": "Show", "seasons": [
|
|
{"season_number": 1, "episodes": [
|
|
{"episode_number": 1, "title": "Pilot",
|
|
"file": {"relative_path": "e1.mkv", "size_bytes": 5}}]}]})
|
|
resp = client.get("/api/video/detail/show/%d" % sid)
|
|
assert resp.status_code == 200
|
|
d = resp.get_json()
|
|
assert d["kind"] == "show" and d["episode_total"] == 1 and d["episode_owned"] == 1
|
|
assert d["seasons"][0]["episodes"][0]["title"] == "Pilot"
|
|
assert client.get("/api/video/detail/show/999999").status_code == 404
|
|
finally:
|
|
videoapi._video_db = None
|
|
|
|
|
|
def test_enrichment_priority_endpoint(tmp_path):
|
|
client, videoapi = _make_client(tmp_path)
|
|
try:
|
|
assert client.get("/api/video/enrichment/priority").get_json()["priority"] == ""
|
|
r = client.post("/api/video/enrichment/priority", json={"priority": "show"})
|
|
assert r.status_code == 200 and r.get_json()["priority"] == "show"
|
|
assert client.get("/api/video/enrichment/priority").get_json()["priority"] == "show"
|
|
assert client.post("/api/video/enrichment/priority", json={"priority": "bogus"}).status_code == 400
|
|
finally:
|
|
videoapi._video_db = None
|
|
|
|
|
|
def test_monitor_toggle_endpoint(tmp_path):
|
|
client, videoapi = _make_client(tmp_path)
|
|
try:
|
|
sid = videoapi._video_db.upsert_show_tree("plex", {"server_id": "s1", "title": "S"})
|
|
r = client.post("/api/video/monitor", json={"kind": "show", "id": sid, "monitored": False})
|
|
assert r.status_code == 200 and r.get_json()["monitored"] is False
|
|
assert videoapi._video_db.show_detail(sid)["monitored"] is False
|
|
r2 = client.post("/api/video/monitor", json={"kind": "show", "id": sid, "monitored": True})
|
|
assert r2.status_code == 200 and videoapi._video_db.show_detail(sid)["monitored"] is True
|
|
# bad inputs
|
|
assert client.post("/api/video/monitor", json={"kind": "bogus", "id": sid}).status_code == 400
|
|
assert client.post("/api/video/monitor", json={"kind": "show", "id": 999999, "monitored": True}).status_code == 404
|
|
finally:
|
|
videoapi._video_db = None
|
|
|
|
|
|
def test_movie_detail_endpoint(tmp_path):
|
|
client, videoapi = _make_client(tmp_path)
|
|
try:
|
|
mid = videoapi._video_db.upsert_movie("plex", {"server_id": "m1", "title": "Dune"})
|
|
resp = client.get("/api/video/detail/movie/%d" % mid)
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()["title"] == "Dune"
|
|
assert client.get("/api/video/detail/movie/999999").status_code == 404
|
|
finally:
|
|
videoapi._video_db = None
|
|
|
|
|
|
def test_dashboard_endpoint_returns_zeroed_json(tmp_path):
|
|
client, videoapi = _make_client(tmp_path)
|
|
try:
|
|
resp = client.get("/api/video/dashboard")
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
assert data["library"]["movies"] == 0
|
|
assert data["downloads"]["active"] == 0
|
|
assert data["watchlist"] == 0 and data["wishlist"] == 0
|
|
finally:
|
|
videoapi._video_db = None # don't leak the tmp DB to other tests
|
|
|
|
|
|
def test_library_endpoint_lists_content(tmp_path):
|
|
client, videoapi = _make_client(tmp_path)
|
|
try:
|
|
videoapi._video_db.upsert_movie("plex", {"server_id": "m1", "title": "A",
|
|
"poster_url": "/library/metadata/1/thumb/9"})
|
|
resp = client.get("/api/video/library?kind=movies")
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
assert [m["title"] for m in data["items"]] == ["A"]
|
|
assert data["items"][0]["has_poster"] is True # flag, not the raw path
|
|
assert "poster_url" not in data["items"][0] # don't leak server paths
|
|
assert data["pagination"]["total_count"] == 1
|
|
finally:
|
|
videoapi._video_db = None
|
|
|
|
|
|
def test_libraries_endpoint_lists_and_saves(tmp_path, monkeypatch):
|
|
client, videoapi = _make_client(tmp_path)
|
|
try:
|
|
import core.video.sources as vs
|
|
monkeypatch.setattr(vs, "list_video_libraries", lambda: {
|
|
"server": "plex", "movies": [{"title": "Movies"}], "tv": [{"title": "TV"}]})
|
|
import config.settings as cs
|
|
monkeypatch.setattr(cs.config_manager, "get_active_media_server", lambda: "plex")
|
|
|
|
data = client.get("/api/video/libraries").get_json()
|
|
assert data["server"] == "plex"
|
|
assert [m["title"] for m in data["movies"]] == ["Movies"]
|
|
assert data["selected"]["movies"] is None
|
|
|
|
assert client.post("/api/video/libraries", json={"movies": "Movies", "tv": "TV"}).status_code == 200
|
|
data2 = client.get("/api/video/libraries").get_json()
|
|
assert data2["selected"] == {"movies": "Movies", "tv": "TV"}
|
|
finally:
|
|
videoapi._video_db = None
|
|
|
|
|
|
def test_enrichment_endpoints(tmp_path):
|
|
import api.video as videoapi
|
|
from database.video_database import VideoDatabase
|
|
import core.video.enrichment.engine as eng_mod
|
|
from core.video.enrichment.engine import VideoEnrichmentEngine
|
|
|
|
class FakeClient:
|
|
enabled = True
|
|
def match(self, *a, **k): return None
|
|
def test(self): return (True, "ok")
|
|
|
|
db = VideoDatabase(database_path=str(tmp_path / "video_library.db"))
|
|
videoapi._video_db = db
|
|
eng_mod._engine = VideoEnrichmentEngine(db, {"tmdb": FakeClient(), "tvdb": FakeClient()})
|
|
app = Flask(__name__)
|
|
app.register_blueprint(videoapi.create_video_blueprint(), url_prefix="/api/video")
|
|
client = app.test_client()
|
|
try:
|
|
svc = client.get("/api/video/enrichment/services").get_json()
|
|
assert {s["id"] for s in svc["services"]} == {"tmdb", "tvdb"}
|
|
|
|
mid = db.upsert_movie("plex", {"server_id": "m1", "title": "X"})
|
|
st = client.get("/api/video/enrichment/tmdb/status").get_json()
|
|
assert st["enabled"] is True and st["stats"]["pending"] == 1
|
|
|
|
db.enrichment_apply("tmdb", "movie", mid, matched=False)
|
|
bd = client.get("/api/video/enrichment/tmdb/breakdown").get_json()
|
|
assert bd["breakdown"]["movie"]["not_found"] == 1
|
|
un = client.get("/api/video/enrichment/tmdb/unmatched?kind=movie&status=not_found").get_json()
|
|
assert un["total"] == 1 and un["kind"] == "movie"
|
|
|
|
assert client.post("/api/video/enrichment/tmdb/pause").get_json()["status"] == "paused"
|
|
assert client.post("/api/video/enrichment/tmdb/resume").get_json()["status"] == "running"
|
|
assert client.post("/api/video/enrichment/tmdb/retry",
|
|
json={"kind": "movie", "scope": "failed"}).get_json()["reset"] == 1
|
|
assert client.post("/api/video/enrichment/tmdb/test").get_json()["success"] is True
|
|
assert client.post("/api/video/enrichment/nope/test").status_code == 404
|
|
assert client.get("/api/video/enrichment/nope/status").status_code == 404
|
|
finally:
|
|
videoapi._video_db = None
|
|
eng_mod._engine = None
|
|
|
|
|
|
def test_enrichment_config_save_load(tmp_path, monkeypatch):
|
|
import api.video as videoapi
|
|
from database.video_database import VideoDatabase
|
|
import core.video.enrichment.engine as eng_mod
|
|
# Don't build a real engine (would open the default-path DB + start threads).
|
|
monkeypatch.setattr(eng_mod, "rebuild_video_enrichment_engine", lambda: None)
|
|
|
|
db = VideoDatabase(database_path=str(tmp_path / "video_library.db"))
|
|
videoapi._video_db = db
|
|
app = Flask(__name__)
|
|
app.register_blueprint(videoapi.create_video_blueprint(), url_prefix="/api/video")
|
|
client = app.test_client()
|
|
try:
|
|
assert client.get("/api/video/enrichment/config").get_json() == {
|
|
"tmdb_api_key": "", "tvdb_api_key": "", "omdb_api_key": "",
|
|
"billboard_autoplay": True, "watch_region": "US"}
|
|
client.post("/api/video/enrichment/config",
|
|
json={"tmdb_api_key": "abc", "tvdb_api_key": "xyz", "omdb_api_key": "om",
|
|
"billboard_autoplay": False, "watch_region": "gb"})
|
|
assert client.get("/api/video/enrichment/config").get_json() == {
|
|
"tmdb_api_key": "abc", "tvdb_api_key": "xyz", "omdb_api_key": "om",
|
|
"billboard_autoplay": False, "watch_region": "GB"}
|
|
assert db.get_setting("tmdb_api_key") == "abc" and db.get_setting("omdb_api_key") == "om"
|
|
assert client.get("/api/video/prefs").get_json() == {
|
|
"billboard_autoplay": False, "watch_region": "GB"}
|
|
finally:
|
|
videoapi._video_db = None
|
|
|
|
|
|
def test_video_api_imports_nothing_from_music():
|
|
base = Path(__file__).resolve().parent.parent / "api" / "video"
|
|
for py in base.glob("*.py"):
|
|
for line in py.read_text(encoding="utf-8").splitlines():
|
|
s = line.strip()
|
|
if s.startswith("import ") or s.startswith("from "):
|
|
assert "music" not in s.lower(), f"{py.name}: music import leaked: {s!r}"
|