PlexLibrary, PlexShow, PlexSeason, PlexItem, PlexPerson, PlexCollection, PlexState, PlexLink, PlexPlaylist and PlexPlaylistItem leave app/models.py for app/plex/models.py; app.tables registers them. Every import is re-pointed; no schema change.
182 lines
6.6 KiB
Python
182 lines
6.6 KiB
Python
"""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 User
|
|
from app.plex.models import PlexItem, PlexLibrary, PlexSeason, PlexShow
|
|
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, override_dep, path):
|
|
monkeypatch.setattr(state, "_configured_cache", True)
|
|
override_dep(get_db, db.info["request"])
|
|
override_dep(current_user, lambda: user)
|
|
return TestClient(app).get(path)
|
|
|
|
|
|
def _ids(body):
|
|
return sorted(i["id"] for i in body["items"])
|
|
|
|
|
|
class TestFieldScopedSearch:
|
|
def test_everywhere_matches_every_facet(self, db, monkeypatch, override_dep):
|
|
s = _seed(db)
|
|
body = _get(
|
|
db, s["user"], monkeypatch, override_dep, "/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, override_dep):
|
|
s = _seed(db)
|
|
default = _get(
|
|
db, s["user"], monkeypatch, override_dep, "/api/plex/library?q=zephyr"
|
|
).json()
|
|
scoped = _get(
|
|
db, s["user"], monkeypatch, override_dep, "/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, override_dep):
|
|
s = _seed(db)
|
|
body = _get(
|
|
db, s["user"], monkeypatch, override_dep, "/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, override_dep):
|
|
s = _seed(db)
|
|
body = _get(
|
|
db, s["user"], monkeypatch, override_dep, "/api/plex/library?q=zephyr&field=people"
|
|
).json()
|
|
assert _ids(body) == ["mp"]
|
|
|
|
def test_summary_scope_narrows_to_description_match(self, db, monkeypatch, override_dep):
|
|
s = _seed(db)
|
|
body = _get(
|
|
db, s["user"], monkeypatch, override_dep, "/api/plex/library?q=zephyr&field=summary"
|
|
).json()
|
|
assert _ids(body) == ["ms"]
|
|
|
|
def test_episode_scope_empties_grid_and_returns_episode(self, db, monkeypatch, override_dep):
|
|
s = _seed(db)
|
|
body = _get(
|
|
db, s["user"], monkeypatch, override_dep, "/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, override_dep):
|
|
s = _seed(db)
|
|
body = _get(
|
|
db,
|
|
s["user"],
|
|
monkeypatch,
|
|
override_dep,
|
|
"/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, override_dep):
|
|
s = _seed(db)
|
|
body = _get(
|
|
db, s["user"], monkeypatch, override_dep, "/api/plex/library?q=zephyr&field=bogus"
|
|
).json()
|
|
assert _ids(body) == ["mp", "ms", "mt"]
|