siftlode/backend/tests/test_plex_field_scoped_search.py

"""S2 · field-scoped search — the `field` param on /api/plex/library.

The search box's scope-pill narrows a full-text query to ONE facet of the row: Everywhere (`all`,
the untouched weighted `search_vector`), Titles, Cast & crew (`people`), Descriptions (`summary`),
or Episodes (episode rows only, title-only). These lock that each scope is a strict narrowing of
Everywhere, that `all` is unchanged from the pre-scope behaviour, and that the Episode scope empties
the movie/show grid and only applies where shows are in scope. DB lane (`db` fixture) — skipped
without Postgres (the `search_vector` GIN + per-field `to_tsvector` recheck are real Postgres FTS).
"""
from datetime import datetime, timezone

import app.state as state
from app.auth import current_user
from app.db import get_db
from app.main import app
from app.models import (
    PlexItem,
    PlexLibrary,
    PlexSeason,
    PlexShow,
    User,
)
from fastapi.testclient import TestClient


def _dt(day: int) -> datetime:
    return datetime(2024, 2, day, tzinfo=timezone.utc)


def _seed(db):
    """One rare token, `zephyr`, planted in exactly one facet per row so a scope's narrowing is
    unambiguous: mt has it ONLY in the title, mp ONLY in cast (people_text), ms ONLY in the summary,
    and episode ep ONLY in its title (its show carries none). Everywhere matches all four; each
    scoped field matches exactly its one row."""
    movie_lib = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True)
    show_lib = PlexLibrary(plex_key="2", title="Shows", kind="show", enabled=True)
    db.add_all([movie_lib, show_lib])
    db.flush()
    mt = PlexItem(rating_key="mt", library_id=movie_lib.id, kind="movie",
                  title="Zephyr Rising", people_text="Alice Stone", summary="A calm coastal day.",
                  year=2001, added_at=_dt(1))
    mp = PlexItem(rating_key="mp", library_id=movie_lib.id, kind="movie",
                  title="Quiet Harbor", people_text="Zephyr Vance", summary="A calm coastal day.",
                  year=2002, added_at=_dt(2))
    ms = PlexItem(rating_key="ms", library_id=movie_lib.id, kind="movie",
                  title="Quiet Harbor Two", people_text="Alice Stone",
                  summary="A story about zephyr winds.", year=2003, added_at=_dt(3))
    show = PlexShow(rating_key="s1", library_id=show_lib.id, title="Nimbus Show",
                    summary="Weather anthology.", people_text="Bob Cloud",
                    year=2004, added_at=_dt(4), child_count=1)
    db.add_all([mt, mp, ms, show])
    db.flush()
    season = PlexSeason(rating_key="se1", show_id=show.id, season_number=1)
    db.add(season)
    db.flush()
    ep = PlexItem(rating_key="ep", library_id=show_lib.id, kind="episode", title="Zephyr Episode",
                  show_id=show.id, season_id=season.id, season_number=1, episode_number=1,
                  added_at=_dt(5))
    db.add(ep)
    user = User(email="plexfield@example.com")
    db.add(user)
    db.commit()
    return {"user": user}


def _get(db, user, monkeypatch, path):
    monkeypatch.setattr(state, "_configured_cache", True)
    app.dependency_overrides[get_db] = lambda: db
    app.dependency_overrides[current_user] = lambda: user
    try:
        return TestClient(app).get(path)
    finally:
        app.dependency_overrides.clear()


def _ids(body):
    return sorted(i["id"] for i in body["items"])


class TestFieldScopedSearch:
    def test_everywhere_matches_every_facet(self, db, monkeypatch):
        s = _seed(db)
        body = _get(db, s["user"], monkeypatch, "/api/plex/library?q=zephyr&field=all").json()
        # title (mt), cast (mp), summary (ms) all land in the item grid; the episode is grouped out
        assert _ids(body) == ["mp", "ms", "mt"]
        assert [e["title"] for e in body["episodes"]] == ["Zephyr Episode"]

    def test_field_all_is_identical_to_default(self, db, monkeypatch):
        s = _seed(db)
        default = _get(db, s["user"], monkeypatch, "/api/plex/library?q=zephyr").json()
        scoped = _get(db, s["user"], monkeypatch, "/api/plex/library?q=zephyr&field=all").json()
        assert _ids(default) == _ids(scoped) == ["mp", "ms", "mt"]
        assert default["total"] == scoped["total"]
        assert [e["id"] for e in default["episodes"]] == [e["id"] for e in scoped["episodes"]]

    def test_title_scope_narrows_to_title_match(self, db, monkeypatch):
        s = _seed(db)
        body = _get(db, s["user"], monkeypatch, "/api/plex/library?q=zephyr&field=title").json()
        assert _ids(body) == ["mt"]
        assert body["total"] == 1
        assert body["episodes"] == []  # non-episode field scopes suppress the episode section

    def test_people_scope_narrows_to_cast_match(self, db, monkeypatch):
        s = _seed(db)
        body = _get(db, s["user"], monkeypatch, "/api/plex/library?q=zephyr&field=people").json()
        assert _ids(body) == ["mp"]

    def test_summary_scope_narrows_to_description_match(self, db, monkeypatch):
        s = _seed(db)
        body = _get(db, s["user"], monkeypatch, "/api/plex/library?q=zephyr&field=summary").json()
        assert _ids(body) == ["ms"]

    def test_episode_scope_empties_grid_and_returns_episode(self, db, monkeypatch):
        s = _seed(db)
        body = _get(db, s["user"], monkeypatch, "/api/plex/library?q=zephyr&field=episode").json()
        assert body["items"] == []
        assert body["total"] == 0
        assert [e["title"] for e in body["episodes"]] == ["Zephyr Episode"]

    def test_episode_scope_needs_shows_in_scope(self, db, monkeypatch):
        s = _seed(db)
        body = _get(
            db, s["user"], monkeypatch, "/api/plex/library?q=zephyr&field=episode&scope=movie"
        ).json()
        # movie-only scope has no show libraries -> no episodes to surface, and the grid stays empty
        assert body["items"] == []
        assert body["episodes"] == []

    def test_unknown_field_falls_back_to_everywhere(self, db, monkeypatch):
        s = _seed(db)
        body = _get(db, s["user"], monkeypatch, "/api/plex/library?q=zephyr&field=bogus").json()
        assert _ids(body) == ["mp", "ms", "mt"]