"""E5 S6: person page TMDB bio enrichment. `parse_person` shapes the /person payload (bio + a popularity-ranked, deduped `known_for`); `enrich_person` lazily fetches + persists once (guarded on a key + a tmdb_person_id); and /person/{name} serves the bio + resolves `known_for` to library rows. DB lane for the endpoint/enrich tests; parse_person is pure. Skipped without Postgres where DB-backed. """ from fastapi.testclient import TestClient 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, PlexPerson, User from app.plex import tmdb as plex_tmdb PAYLOAD = { "biography": " A celebrated actor. ", "birthday": "1970-01-01", "deathday": None, "place_of_birth": "Springfield", "known_for_department": "Acting", "also_known_as": ["AKA One", "AKA Two"], "combined_credits": { "cast": [ {"id": 10, "media_type": "movie", "title": "Mid Movie", "popularity": 50}, {"id": 20, "media_type": "tv", "name": "Top Show", "popularity": 80}, {"id": 10, "media_type": "movie", "title": "dup", "popularity": 5}, # dup id {"id": 30, "media_type": "movie", "title": "Low Movie", "popularity": 1}, {"id": None, "media_type": "movie", "title": "No id", "popularity": 99}, # dropped ] }, } def test_parse_person_shapes_bio_and_ranked_known_for(): p = plex_tmdb.parse_person(PAYLOAD) assert p["biography"] == "A celebrated actor." # stripped assert p["birthday"] == "1970-01-01" assert p["known_for_department"] == "Acting" assert p["also_known_as"] == ["AKA One", "AKA Two"] # Popularity desc, deduped by id, None-id dropped: Top Show (80) → Mid Movie (50) → Low Movie (1). assert [k["tmdb_id"] for k in p["known_for"]] == [20, 10, 30] assert p["known_for"][0]["media_type"] == "tv" def test_parse_person_empty_payload(): p = plex_tmdb.parse_person({}) assert p["biography"] is None and p["known_for"] is None class _FakeTmdb: def __init__(self, payload): self.payload = payload def __enter__(self): return self def __exit__(self, *a): return False def fetch_person(self, pid): return self.payload def test_enrich_person_persists_and_stamps(db, monkeypatch): person = PlexPerson(name="Actor X", tmdb_person_id=555) db.add(person) db.commit() monkeypatch.setattr(plex_tmdb.sysconfig, "get_str", lambda d, k: "key" if k == "tmdb_api_key" else "") monkeypatch.setattr(plex_tmdb, "TmdbClient", lambda d, timeout=None: _FakeTmdb(PAYLOAD)) assert plex_tmdb.enrich_person(db, person) is True assert person.biography == "A celebrated actor." assert person.tmdb_enriched_at is not None assert [k["tmdb_id"] for k in person.known_for] == [20, 10, 30] def test_enrich_person_noop_without_key(db, monkeypatch): person = PlexPerson(name="Actor Y", tmdb_person_id=555) db.add(person) db.commit() monkeypatch.setattr(plex_tmdb.sysconfig, "get_str", lambda d, k: "") # no key assert plex_tmdb.enrich_person(db, person) is False assert person.tmdb_enriched_at is None def test_enrich_person_noop_without_tmdb_id(db, monkeypatch): person = PlexPerson(name="Actor Z", tmdb_person_id=None) db.add(person) db.commit() monkeypatch.setattr(plex_tmdb.sysconfig, "get_str", lambda d, k: "key") assert plex_tmdb.enrich_person(db, person) is False 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 test_person_detail_survives_poisoned_enrich(db, monkeypatch): # If the lazy enrich_person leaves the session in a failed-transaction state (e.g. a commit-time # flush error), person_detail's except must roll it back so the following SELECTs still run — the # page renders despite the enrich failure ("never fail the page"). Without the db.rollback() fix the # poisoned session would raise PendingRollbackError on the next query and 500. from sqlalchemy import text lib = PlexLibrary(plex_key="9", title="Movies", kind="movie", enabled=True) db.add(lib) db.flush() # Un-enriched person (tmdb_person_id set + tmdb_enriched_at None) → triggers the lazy enrich path. db.add(PlexPerson(name="Poison P", tmdb_person_id=777)) user = User(email="poison@example.com") db.add(user) db.commit() def _poison_then_raise(db_, person_): try: db_.execute(text("SELECT 1 / 0")) # aborts the transaction (like a failed enrich commit) except Exception: # noqa: BLE001 — swallow so the tx stays poisoned, then raise like the real path pass raise RuntimeError("enrich commit blew up") monkeypatch.setattr(plex_tmdb, "enrich_person", _poison_then_raise) resp = _get(db, user, monkeypatch, "/api/plex/person/Poison P") assert resp.status_code == 200, resp.text # page still renders (session was rolled back) assert resp.json()["biography"] is None # enrich failed, so no bio def test_person_detail_serves_bio_and_resolves_known_for(db, monkeypatch): lib = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True) db.add(lib) db.flush() # A library movie that IS one of the person's known_for credits (matched by guid tmdb:100). db.add(PlexItem(rating_key="km", library_id=lib.id, kind="movie", title="Known Movie", guid="tmdb:100", cast_names=["Actor X"])) # Already-enriched person (tmdb_enriched_at set → no lazy fetch), with a known_for pool. from datetime import datetime, timezone db.add(PlexPerson(name="Actor X", tmdb_person_id=555, tmdb_enriched_at=datetime(2024, 1, 1, tzinfo=timezone.utc), biography="Bio here.", birthday="1970-01-01", known_for_department="Acting", also_known_as=["Alt Name"], known_for=[{"tmdb_id": 100, "media_type": "movie", "title": "Known Movie"}, {"tmdb_id": 999, "media_type": "movie", "title": "Not In Library"}])) user = User(email="personbio@example.com") db.add(user) db.commit() resp = _get(db, user, monkeypatch, "/api/plex/person/Actor X") assert resp.status_code == 200, resp.text j = resp.json() assert j["biography"] == "Bio here." assert j["birthday"] == "1970-01-01" assert j["also_known_as"] == ["Alt Name"] # Only the in-library known_for credit resolves to a card; the out-of-library one is dropped. assert [c["title"] for c in j["known_for"]] == ["Known Movie"]