siftlode/backend/tests/test_plex_plaintext.py

"""Plex summaries and taglines are stored as plain text.

A Hungarian TV agent hands summaries over as HTML (`<p>A műsorban … p&aacute;rosnak …</p>`), and
the info page and the hover card showed the tags. `plain_text` is the one cleaner: the sync's every
summary path and the HU overlay store through it, and migration 0101 cleaned the rows written before.
"""

import pytest
from alembic import command
from alembic.config import Config
from sqlalchemy import select
from sqlalchemy.orm import Session

from app.db import engine
from app.models import PlexCollection, PlexItem, PlexLibrary, PlexShow
from app.plex.detail import rich_detail
from app.plex.overlay import sync_overlay
from app.plex.plaintext import plain_text
from app.plex import sync
from app.plex.sync import _apply_item, _upsert_collection

HU_SUMMARY = "<p>A műsorban minden p&aacute;rosnak stoppal kell eljutnia.</p>"


@pytest.mark.parametrize(
    "raw, clean",
    [
        (HU_SUMMARY, "A műsorban minden párosnak stoppal kell eljutnia."),
        ("<p>One.</p><p>Two.</p>", "One.\n\nTwo."),
        ("Line one<br>line two<BR/>three", "Line one\nline two\nthree"),
        ("<b>Bold</b> &amp; <i>more</i>", "Bold & more"),
        # Prose that only looks like markup stays as written.
        ("a < b and 1 <3 you", "a < b and 1 <3 you"),
        # Only a reference with its semicolon is one: an ampersand in prose stays as written.
        ("AT&notice, R&copy and H&M", "AT&notice, R&copy and H&M"),
        ("&#233;t&#xE9; &copy;", "été ©"),
        ("Plain summary.", "Plain summary."),
        ("  padded  ", "padded"),
        ("<p></p>", None),
        ("", None),
        (None, None),
    ],
)
def test_plain_text(raw, clean):
    assert plain_text(raw) == clean


def test_the_item_sync_stores_a_summary_as_plain_text():
    row = PlexItem()
    _apply_item(row, lib_id=1, kind="episode", meta={"title": "Ep", "summary": HU_SUMMARY})
    assert row.summary == "A műsorban minden párosnak stoppal kell eljutnia."


def test_the_show_sync_stores_a_summary_as_plain_text(db, monkeypatch):
    lib = PlexLibrary(plex_key="2", title="Shows", kind="show", enabled=True)
    db.add(lib)
    db.commit()
    meta = {"ratingKey": "s1", "title": "Show", "summary": HU_SUMMARY}

    def listing(_plex, _key, typ, progress=None):
        if progress is not None:
            progress["complete"] = True
        return [meta] if typ == sync._TYPE_SHOW else []

    monkeypatch.setattr(sync, "_paginate", listing)
    sync._sync_shows(db, object(), lib, {"shows": 0, "seasons": 0, "episodes": 0})
    db.commit()
    show = db.execute(select(PlexShow).where(PlexShow.rating_key == "s1")).scalar_one()
    assert show.summary == "A műsorban minden párosnak stoppal kell eljutnia."


def test_a_collection_stores_its_summary_as_plain_text(db):
    lib = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True)
    db.add(lib)
    db.flush()
    col = _upsert_collection(db, None, "c1", lib.id, {"title": "C", "summary": "<p>A &amp; B</p>"})
    assert col.summary == "A & B"


def test_a_fetched_tagline_is_plain_text():
    assert rich_detail({"tagline": "<i>An offer</i> &amp; more"})["tagline"] == "An offer & more"


class _StubPlex:
    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 test_the_hungarian_overlay_stores_plain_text(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
    row = PlexItem(rating_key="m1", library_id=en_lib.id, kind="movie", title="Heat", guid="tmdb:1")
    db.add(row)
    db.commit()

    meta = {
        "ratingKey": "h1",
        "title": "Szemtől szemben",
        "summary": HU_SUMMARY,
        "tagline": "<i>Egy &eacute;let</i>",
        "Guid": [{"id": "tmdb://1"}],
    }
    sync_overlay(db, _StubPlex([meta]), hu_lib, en_lib.id, {"overlaid": 0})
    db.commit()
    db.refresh(row)
    assert row.summary_hu == "A műsorban minden párosnak stoppal kell eljutnia."
    assert row.tagline_hu == "Egy élet"


def test_0101_cleans_the_stored_rows(db):
    cfg = Config("alembic.ini")
    command.downgrade(cfg, "0100_drop_plex_item_markers")
    upgraded = False
    try:
        with Session(engine) as s:
            lib = PlexLibrary(plex_key="1", title="Shows", kind="show", enabled=True)
            s.add(lib)
            s.flush()
            ep = PlexItem(
                rating_key="e1",
                library_id=lib.id,
                kind="episode",
                title="Ep",
                summary=HU_SUMMARY,
                tagline_hu="<i>x</i>",
            )
            plain = PlexItem(
                rating_key="m1", library_id=lib.id, kind="movie", title="M", summary="a < b"
            )
            show = PlexShow(rating_key="s1", library_id=lib.id, title="S", summary_hu="<p>Hu</p>")
            col = PlexCollection(rating_key="c1", library_id=lib.id, title="C", summary="A &amp; B")
            s.add_all([ep, plain, show, col])
            s.commit()
            ids = (ep.id, plain.id, show.id, col.id)
        command.upgrade(cfg, "head")
        upgraded = True
        with Session(engine) as s:
            ep2, plain2 = s.get(PlexItem, ids[0]), s.get(PlexItem, ids[1])
            show2, col2 = s.get(PlexShow, ids[2]), s.get(PlexCollection, ids[3])
            assert ep2 and plain2 and show2 and col2
            assert ep2.summary == "A műsorban minden párosnak stoppal kell eljutnia."
            assert ep2.tagline_hu == "x"
            assert plain2.summary == "a < b"
            assert show2.summary_hu == "Hu"
            assert col2.summary == "A & B"
    finally:
        if not upgraded:
            command.upgrade(cfg, "head")