fix(plex): address S5a review — overlay reconcile, dup-guid, guid-wipe, tests

- Correctness (stale HU): _sync_overlay is now a full reconcile — it clears every guid-bearing EN
  row's *_hu first, then repopulates the matches, so a title whose HU twin vanished or was re-guid'd
  falls back to EN instead of keeping stale Hungarian metadata (only rows that held HU data go dirty).
- Correctness (dup guid): by_guid maps a guid to the LIST of EN rows carrying it, so two editions of
  one film both receive the overlay instead of just the last-iterated one.
- Robustness (guid wipe): _apply_item/_sync_shows keep a prior guid when a listing carries none
  (`_canonical_guid(meta) or row.guid`), so a server ignoring includeGuids can't erase the match key.
- Reuse: merged _sync_overlay's movie/show branches (only the select + item_type differ).
- Test-coverage: a full-sync test drives sync() over a stub Plex (sections + section_items +
  overlay_map) asserting pass-2 designation (metadata_lang/overlay_of), guid capture, the HU write,
  and browse exclusion; plus a reconcile test that a vanished twin clears stale *_hu.

Full backend suite 274 passed / 2 skipped.
This commit is contained in:
2026-08-16 22:31:08 +02:00
parent c959859d29
commit 704c6c35d4
2 changed files with 128 additions and 28 deletions
+88
View File
@@ -7,6 +7,8 @@ Uses the DB lane (`db` fixture) — skipped when no Postgres is reachable. Runs
"""
from sqlalchemy import func, select
import app.plex.sync as syncmod
from app import sysconfig
from app.models import PlexItem, PlexLibrary
from app.plex.sync import _canonical_guid, _sync_overlay
from app.routes.plex.browse import _content_libs
@@ -82,6 +84,92 @@ def test_sync_overlay_writes_hu_onto_matched_en_rows(db):
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_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")