c17138bf7d
Isolated enrichment subsystem mirroring the music worker pattern, on video.db.
- VideoEnrichmentWorker: daemon loop pulling enrichment_next -> client.match ->
enrichment_apply; pause/resume/stop; get_stats in the same shape the music
enrichment API returns (enabled/running/paused/idle/current_item/stats/
progress/breakdown). Client injected -> loop fully unit-tested with a fake.
- TMDB/TVDB clients (thin adapters, keys from video_settings): .enabled +
match(kind,title,year) -> {id, metadata}; validated live, worker logic tested.
- VideoEnrichmentEngine registry + lazy get_video_enrichment_engine() singleton.
8 tests (match/not_found/error-resilience/stats/disabled/engine/isolation).
105 lines
3.7 KiB
Python
105 lines
3.7 KiB
Python
"""Seam tests for the video enrichment workers (experimental branch).
|
|
|
|
The worker's loop/queue/status logic is driven by a FAKE client so it's tested
|
|
without hitting TMDB/TVDB. Also guards that the enrichment package imports
|
|
nothing from the music side.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from database.video_database import VideoDatabase
|
|
from core.video.enrichment.worker import VideoEnrichmentWorker
|
|
from core.video.enrichment.engine import VideoEnrichmentEngine
|
|
|
|
|
|
class FakeClient:
|
|
enabled = True
|
|
|
|
def __init__(self, result):
|
|
self._result = result
|
|
self.calls = []
|
|
|
|
def match(self, kind, title, year):
|
|
self.calls.append((kind, title, year))
|
|
return self._result
|
|
|
|
|
|
@pytest.fixture()
|
|
def db(tmp_path):
|
|
return VideoDatabase(database_path=str(tmp_path / "video_library.db"))
|
|
|
|
|
|
def test_worker_process_one_matches(db):
|
|
db.upsert_movie("plex", {"server_id": "m1", "title": "Dune", "year": 2021})
|
|
client = FakeClient({"id": 438631, "metadata": {"overview": "O", "backdrop_url": "/b.jpg"}})
|
|
w = VideoEnrichmentWorker(db, "tmdb", client)
|
|
assert w.process_one() is True
|
|
assert client.calls == [("movie", "Dune", 2021)]
|
|
with db.connect() as c:
|
|
row = c.execute("SELECT tmdb_id, tmdb_match_status, overview FROM movies").fetchone()
|
|
assert (row["tmdb_id"], row["tmdb_match_status"], row["overview"]) == (438631, "matched", "O")
|
|
assert w.stats["matched"] == 1
|
|
|
|
|
|
def test_worker_process_one_not_found(db):
|
|
db.upsert_movie("plex", {"server_id": "m1", "title": "X"})
|
|
w = VideoEnrichmentWorker(db, "tmdb", FakeClient(None))
|
|
assert w.process_one() is True
|
|
with db.connect() as c:
|
|
assert c.execute("SELECT tmdb_match_status FROM movies").fetchone()["tmdb_match_status"] == "not_found"
|
|
assert w.stats["not_found"] == 1
|
|
|
|
|
|
def test_worker_process_one_no_items_returns_false(db):
|
|
assert VideoEnrichmentWorker(db, "tmdb", FakeClient(None)).process_one() is False
|
|
|
|
|
|
def test_worker_match_exception_marks_not_found_and_counts_error(db):
|
|
db.upsert_movie("plex", {"server_id": "m1", "title": "X"})
|
|
|
|
class Boom:
|
|
enabled = True
|
|
def match(self, *a): raise RuntimeError("api down")
|
|
|
|
w = VideoEnrichmentWorker(db, "tmdb", Boom())
|
|
assert w.process_one() is True # doesn't crash the loop
|
|
assert w.stats["errors"] == 1
|
|
with db.connect() as c:
|
|
assert c.execute("SELECT tmdb_match_status FROM movies").fetchone()["tmdb_match_status"] == "not_found"
|
|
|
|
|
|
def test_worker_get_stats_shape(db):
|
|
db.upsert_movie("plex", {"server_id": "m1", "title": "X"})
|
|
s = VideoEnrichmentWorker(db, "tmdb", FakeClient(None)).get_stats()
|
|
assert {"enabled", "running", "paused", "idle", "current_item", "stats",
|
|
"progress", "breakdown"} <= set(s)
|
|
assert s["enabled"] is True
|
|
assert s["stats"]["pending"] == 1
|
|
|
|
|
|
def test_worker_disabled_without_key(db):
|
|
class NoKey:
|
|
enabled = False
|
|
|
|
w = VideoEnrichmentWorker(db, "tmdb", NoKey())
|
|
assert w.enabled is False
|
|
assert w.get_stats()["enabled"] is False
|
|
|
|
|
|
def test_engine_builds_and_lists_workers(db):
|
|
eng = VideoEnrichmentEngine(db, {"tmdb": FakeClient(None), "tvdb": FakeClient(None)})
|
|
assert {s["id"] for s in eng.services()} == {"tmdb", "tvdb"}
|
|
assert eng.worker("tmdb") is not None and eng.worker("nope") is None
|
|
|
|
|
|
def test_enrichment_package_imports_nothing_from_music():
|
|
base = Path(__file__).resolve().parent.parent / "core" / "video" / "enrichment"
|
|
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}"
|