peter
13d95ef70e
The HU TV section lists every episode with its Hungarian title and summary, but the overlay read only movies and shows, so every episode stayed English. It now overlays the episodes too, and the HU tagline (new tagline_hu column, served on ?lang=hu). Plex's HU agent returns a mixed genre vocabulary (Action beside Dráma), and a title with no HU twin kept its English list, so the HU genre list showed Drama and Dráma. plex/genres.py maps every genre word to one Hungarian word; the overlay writes it into genres_hu for every row, and a HU genre filter is read through the same map, so a saved filter still holding Action keeps matching.
345 lines
13 KiB
Python
345 lines
13 KiB
Python
"""DB-backed regression (E5 S5a): the HU-overlay sync writes Hungarian metadata onto the matched EN
|
|
rows (paired by guid), never creating browsable HU rows, and overlay libraries are excluded from the
|
|
browse scope. Also covers the canonical-guid parse and that the generated search_vector_hu populates.
|
|
|
|
Uses the DB lane (`db` fixture) — skipped when no Postgres is reachable. Runs against the real schema
|
|
(migrations 0063/0064) so the JSONB + STORED generated columns are exercised.
|
|
"""
|
|
|
|
from sqlalchemy import func, select
|
|
|
|
import app.plex.sync as syncmod
|
|
from app import sysconfig
|
|
from app.models import PlexItem, PlexLibrary, PlexShow
|
|
from app.plex.genres import hu_genre, hu_genres
|
|
from app.plex.sync import _apply_item, _canonical_guid, _sync_overlay
|
|
from app.plex.query import content_libs
|
|
|
|
|
|
class _StubPlex:
|
|
"""Minimal PlexClient stand-in: _sync_overlay only calls section_items via _paginate."""
|
|
|
|
def __init__(self, items: list[dict]):
|
|
self._items = items
|
|
|
|
def section_items(self, key, item_type, start, size):
|
|
return self._items[start : start + size], len(self._items)
|
|
|
|
|
|
def _hu_movie(rk: str, title: str, summary: str, genres: list[str], guid: str | None) -> dict:
|
|
meta = {
|
|
"ratingKey": rk,
|
|
"title": title,
|
|
"summary": summary,
|
|
"Genre": [{"tag": g} for g in genres],
|
|
}
|
|
if guid:
|
|
meta["Guid"] = [{"id": guid}]
|
|
return meta
|
|
|
|
|
|
def test_canonical_guid_prefers_tmdb_then_imdb():
|
|
assert _canonical_guid({"Guid": [{"id": "imdb://tt1"}, {"id": "tmdb://55"}]}) == "tmdb:55"
|
|
assert _canonical_guid({"Guid": [{"id": "imdb://tt9"}, {"id": "tvdb://7"}]}) == "imdb:tt9"
|
|
assert _canonical_guid({"Guid": []}) is None
|
|
assert _canonical_guid({}) is None
|
|
|
|
|
|
def test_sync_overlay_writes_hu_onto_matched_en_rows(db):
|
|
en_lib = PlexLibrary(plex_key="4", title="Movies", kind="movie")
|
|
hu_lib = PlexLibrary(plex_key="9", title="Filmek", kind="movie", metadata_lang="hu")
|
|
db.add_all([en_lib, hu_lib])
|
|
db.flush()
|
|
hu_lib.overlay_of = en_lib.id
|
|
matched = PlexItem(
|
|
rating_key="m1", library_id=en_lib.id, kind="movie", title="Heat", guid="tmdb:1"
|
|
)
|
|
unmatched = PlexItem(
|
|
rating_key="m2", library_id=en_lib.id, kind="movie", title="Solo", guid="tmdb:2"
|
|
)
|
|
db.add_all([matched, unmatched])
|
|
db.commit()
|
|
|
|
stub = _StubPlex(
|
|
[
|
|
_hu_movie("h1", "Öldöklés", "Egy rablás terve.", ["Akció", "Krimi"], "tmdb://1"),
|
|
_hu_movie("h2", "Ismeretlen", "Nincs EN párja.", ["Dráma"], "tmdb://999"),
|
|
]
|
|
)
|
|
stats = {"overlaid": 0}
|
|
_sync_overlay(db, stub, hu_lib, en_lib.id, stats)
|
|
db.commit()
|
|
|
|
db.refresh(matched)
|
|
db.refresh(unmatched)
|
|
assert matched.title_hu == "Öldöklés"
|
|
assert matched.summary_hu == "Egy rablás terve."
|
|
assert matched.genres_hu == ["Akció", "Krimi"]
|
|
assert (
|
|
matched.search_vector_hu is not None
|
|
) # the STORED generated column populated from title_hu
|
|
# The unmatched EN row keeps NULL HU fields (serving falls back to EN).
|
|
assert unmatched.title_hu is None
|
|
assert stats["overlaid"] == 1
|
|
# No browsable rows were created — the HU items never became plex_items of the overlay library.
|
|
hu_row_count = db.scalar(
|
|
select(func.count(PlexItem.id)).where(PlexItem.library_id == hu_lib.id)
|
|
)
|
|
assert hu_row_count == 0
|
|
# The EN library still holds exactly its two movies (no dup from the overlay pass).
|
|
en_row_count = db.scalar(
|
|
select(func.count(PlexItem.id)).where(PlexItem.library_id == en_lib.id)
|
|
)
|
|
assert en_row_count == 2
|
|
|
|
|
|
class _StubClient:
|
|
"""Full-sync stand-in: one EN movie section (key 4) + its HU twin (key 9), paired by guid."""
|
|
|
|
def __init__(self, db=None):
|
|
pass
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *exc):
|
|
return False
|
|
|
|
def sections(self):
|
|
return [
|
|
{"key": "4", "title": "Movies", "type": "movie"},
|
|
{"key": "9", "title": "Filmek", "type": "movie"},
|
|
]
|
|
|
|
def section_items(self, key, item_type, start, size):
|
|
if item_type != syncmod._TYPE_MOVIE:
|
|
return [], 0
|
|
if key == "4":
|
|
items = [{"ratingKey": "en1", "title": "Heat", "Guid": [{"id": "tmdb://1"}]}]
|
|
elif key == "9":
|
|
items = [
|
|
{
|
|
"ratingKey": "hu1",
|
|
"title": "Öldöklés",
|
|
"summary": "Rablás.",
|
|
"Genre": [{"tag": "Akció"}],
|
|
"Guid": [{"id": "tmdb://1"}],
|
|
}
|
|
]
|
|
else:
|
|
items = []
|
|
return items[start : start + size], len(items)
|
|
|
|
def collections(self, key):
|
|
return []
|
|
|
|
def metadata(self, rk, markers=False):
|
|
return {}
|
|
|
|
|
|
def test_sync_designates_overlay_and_writes_hu(db, monkeypatch):
|
|
sysconfig.set_value(db, "plex_enabled", True)
|
|
sysconfig.set_value(db, "plex_libraries", "4") # only the EN section is browsable
|
|
sysconfig.set_value(db, "plex_overlay_map", "9:4") # HU section 9 overlays EN section 4
|
|
db.commit()
|
|
monkeypatch.setattr(syncmod, "PlexClient", _StubClient)
|
|
|
|
stats = syncmod.sync(db)
|
|
|
|
en = db.execute(select(PlexLibrary).where(PlexLibrary.plex_key == "4")).scalars().one()
|
|
hu = db.execute(select(PlexLibrary).where(PlexLibrary.plex_key == "9")).scalars().one()
|
|
assert en.metadata_lang == "en" and en.overlay_of is None
|
|
assert hu.metadata_lang == "hu" and hu.overlay_of == en.id # pass-2 designation
|
|
# The EN movie captured its guid (from includeGuids listing) and received the HU overlay.
|
|
m = db.execute(select(PlexItem).where(PlexItem.rating_key == "en1")).scalars().one()
|
|
assert m.guid == "tmdb:1"
|
|
assert m.title_hu == "Öldöklés"
|
|
assert m.genres_hu == ["Akció"]
|
|
# The HU section is not browsable and mirrored no rows of its own.
|
|
movie_ids, _ = content_libs(db)
|
|
assert en.id in movie_ids and hu.id not in movie_ids
|
|
assert db.scalar(select(func.count(PlexItem.id)).where(PlexItem.library_id == hu.id)) == 0
|
|
assert stats["overlaid"] == 1
|
|
|
|
|
|
def test_overlay_reconcile_clears_stale_hu_when_twin_vanishes(db):
|
|
"""A re-run whose HU section no longer carries a title's guid must drop that EN row's stale HU."""
|
|
en_lib = PlexLibrary(plex_key="4", title="Movies", kind="movie")
|
|
hu_lib = PlexLibrary(plex_key="9", title="Filmek", kind="movie", metadata_lang="hu")
|
|
db.add_all([en_lib, hu_lib])
|
|
db.flush()
|
|
hu_lib.overlay_of = en_lib.id
|
|
row = PlexItem(
|
|
rating_key="m1",
|
|
library_id=en_lib.id,
|
|
kind="movie",
|
|
title="Heat",
|
|
guid="tmdb:1",
|
|
title_hu="Öldöklés",
|
|
summary_hu="régi",
|
|
genres_hu=["Akció"],
|
|
)
|
|
db.add(row)
|
|
db.commit()
|
|
# The HU section no longer lists this film (empty overlay).
|
|
_sync_overlay(db, _StubPlex([]), hu_lib, en_lib.id, {"overlaid": 0})
|
|
db.commit()
|
|
db.refresh(row)
|
|
assert row.title_hu is None and row.summary_hu is None and row.genres_hu is None
|
|
|
|
|
|
def test_apply_item_keeps_prior_guid_when_listing_lacks_guid():
|
|
row = PlexItem(rating_key="m1", kind="movie", guid="tmdb:1")
|
|
_apply_item(row, lib_id=7, kind="movie", meta={"title": "Heat"}) # no Guid array
|
|
assert row.guid == "tmdb:1" # the prior guid survives a guid-less listing
|
|
_apply_item(row, lib_id=7, kind="movie", meta={"title": "Heat", "Guid": [{"id": "tmdb://2"}]})
|
|
assert row.guid == "tmdb:2" # a real guid in the listing still updates it
|
|
|
|
|
|
def test_overlay_applies_to_all_editions_sharing_a_guid(db):
|
|
en_lib = PlexLibrary(plex_key="4", title="Movies", kind="movie")
|
|
hu_lib = PlexLibrary(plex_key="9", title="Filmek", kind="movie", metadata_lang="hu")
|
|
db.add_all([en_lib, hu_lib])
|
|
db.flush()
|
|
hu_lib.overlay_of = en_lib.id
|
|
ed1 = PlexItem(rating_key="a", library_id=en_lib.id, kind="movie", title="Heat", guid="tmdb:1")
|
|
ed2 = PlexItem(
|
|
rating_key="b",
|
|
library_id=en_lib.id,
|
|
kind="movie",
|
|
title="Heat (Director's Cut)",
|
|
guid="tmdb:1",
|
|
)
|
|
db.add_all([ed1, ed2])
|
|
db.commit()
|
|
|
|
# Two HU entries share the guid — first wins, no double count.
|
|
stub = _StubPlex(
|
|
[
|
|
_hu_movie("h1", "Öldöklés", "első", ["Akció"], "tmdb://1"),
|
|
_hu_movie("h2", "Öldöklés (dup)", "második", ["Krimi"], "tmdb://1"),
|
|
]
|
|
)
|
|
stats = {"overlaid": 0}
|
|
_sync_overlay(db, stub, hu_lib, en_lib.id, stats)
|
|
db.commit()
|
|
db.refresh(ed1)
|
|
db.refresh(ed2)
|
|
assert ed1.title_hu == "Öldöklés" and ed2.title_hu == "Öldöklés" # both editions overlaid
|
|
assert ed1.genres_hu == ["Akció"] and ed2.genres_hu == ["Akció"] # first HU entry won
|
|
assert stats["overlaid"] == 1 # counted once for the guid, not per edition or per dup HU item
|
|
|
|
|
|
def test_content_libs_excludes_overlay_libraries(db):
|
|
en_lib = PlexLibrary(plex_key="4", title="Movies", kind="movie")
|
|
en_shows = PlexLibrary(plex_key="1", title="TV Shows", kind="show")
|
|
db.add_all([en_lib, en_shows])
|
|
db.flush()
|
|
overlay = PlexLibrary(
|
|
plex_key="9", title="Filmek", kind="movie", metadata_lang="hu", overlay_of=en_lib.id
|
|
)
|
|
db.add(overlay)
|
|
db.commit()
|
|
|
|
movie_ids, show_ids = content_libs(db)
|
|
assert en_lib.id in movie_ids
|
|
assert overlay.id not in movie_ids # the HU overlay is never browsable
|
|
assert show_ids == [en_shows.id]
|
|
|
|
|
|
class _KindStub:
|
|
"""An overlay section listing per Plex type (2 = show, 4 = episode), like a HU TV section."""
|
|
|
|
def __init__(self, by_type: dict[int, list[dict]]):
|
|
self._by = by_type
|
|
|
|
def section_items(self, key, item_type, start, size):
|
|
items = self._by.get(item_type, [])
|
|
return items[start : start + size], len(items)
|
|
|
|
|
|
def test_a_show_overlay_writes_hungarian_episode_titles_and_summaries(db):
|
|
"""E30: the HU TV section lists every episode in Hungarian; they were never read, so a Hungarian
|
|
reader saw every episode title and summary in English."""
|
|
en_lib = PlexLibrary(plex_key="1", title="TV Shows", kind="show")
|
|
hu_lib = PlexLibrary(plex_key="9", title="Sorozatok", kind="show", metadata_lang="hu")
|
|
db.add_all([en_lib, hu_lib])
|
|
db.flush()
|
|
show = PlexShow(rating_key="s1", library_id=en_lib.id, title="3 Body Problem", guid="tmdb:10")
|
|
db.add(show)
|
|
db.flush()
|
|
ep = PlexItem(
|
|
rating_key="e1",
|
|
library_id=en_lib.id,
|
|
kind="episode",
|
|
show_id=show.id,
|
|
title="Countdown",
|
|
summary="EN.",
|
|
guid="tmdb:501",
|
|
)
|
|
db.add(ep)
|
|
db.commit()
|
|
stub = _KindStub(
|
|
{
|
|
syncmod._TYPE_SHOW: [_hu_movie("hs", "A 3-test-probléma", "HU.", [], "tmdb://10")],
|
|
syncmod._TYPE_EPISODE: [
|
|
_hu_movie("he", "Visszaszámlálás", "Magyar epizód.", [], "tmdb://501")
|
|
],
|
|
}
|
|
)
|
|
_sync_overlay(db, stub, hu_lib, en_lib.id, {})
|
|
db.commit()
|
|
db.refresh(ep)
|
|
db.refresh(show)
|
|
assert (ep.title_hu, ep.summary_hu) == ("Visszaszámlálás", "Magyar epizód.")
|
|
assert show.title_hu == "A 3-test-probléma"
|
|
|
|
|
|
def test_the_overlay_gives_every_row_one_hungarian_word_per_genre(db):
|
|
"""E30: Plex's HU agent mixes "Action" with "Dráma", and a title with no HU twin kept only its
|
|
English list — so the HU genre list offered "Drama" beside "Dráma"."""
|
|
en_lib = PlexLibrary(plex_key="4", title="Movies", kind="movie")
|
|
hu_lib = PlexLibrary(plex_key="9", title="Filmek", kind="movie", metadata_lang="hu")
|
|
db.add_all([en_lib, hu_lib])
|
|
db.flush()
|
|
twin = PlexItem(
|
|
rating_key="m1",
|
|
library_id=en_lib.id,
|
|
kind="movie",
|
|
title="Heat",
|
|
guid="tmdb:1",
|
|
genres=["Action", "Crime"],
|
|
)
|
|
lone = PlexItem(
|
|
rating_key="m2",
|
|
library_id=en_lib.id,
|
|
kind="movie",
|
|
title="Solo",
|
|
guid="tmdb:2",
|
|
genres=["Drama", "Science Fiction"],
|
|
)
|
|
db.add_all([twin, lone])
|
|
db.commit()
|
|
meta = _hu_movie(
|
|
"h1", "Szemtől szemben", "HU.", ["Action", "Bűnügyi", "Drama", "Dráma"], "tmdb://1"
|
|
)
|
|
meta["tagline"] = "Magyar mottó."
|
|
_sync_overlay(db, _StubPlex([meta]), hu_lib, en_lib.id, {})
|
|
db.commit()
|
|
db.refresh(twin)
|
|
db.refresh(lone)
|
|
assert twin.genres_hu == ["Akció", "Bűnügyi", "Dráma"]
|
|
assert twin.tagline_hu == "Magyar mottó."
|
|
assert lone.genres_hu == ["Dráma", "Sci-Fi"] # no twin: its own English list, in Hungarian
|
|
assert lone.tagline_hu is None
|
|
|
|
|
|
def test_the_genre_map_is_idempotent_and_keeps_an_unknown_word():
|
|
assert hu_genre("science fiction") == "Sci-Fi" # case-insensitive
|
|
assert hu_genre("Dráma") == "Dráma"
|
|
assert hu_genre("Space Opera") == "Space Opera" # unknown: shown as Plex gave it, never dropped
|
|
once = hu_genres(["Drama", "Action", "Dráma"])
|
|
assert once == ["Dráma", "Akció"]
|
|
assert hu_genres(once) == once
|
|
assert hu_genres(None) is None and hu_genres([]) is None
|