siftlode/backend/tests/test_plex_overlay.py

"""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
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]