"""E30 S6: the people are made local ahead of time. The people_enrich job fetches linked people's TMDB bios (the most-credited first) and caches their portraits resized through Plex; the person page says when a bio is still on its way (`enriching`), so it can ask again instead of staying empty. DB lane (Postgres); the portrait cache runs in a temp directory. """ from datetime import datetime, timezone import pytest from app import scheduler from app.models import PlexItem, PlexLibrary, PlexPerson, User from app.plex import imgcache from app.plex import people as plex_people from app.plex import tmdb as plex_tmdb CDN = plex_people.PERSON_IMG_HOST + "people/" DONE = datetime(2024, 1, 1, tzinfo=timezone.utc) def _movie(db, lib, key, cast): db.add(PlexItem(rating_key=key, library_id=lib.id, kind="movie", title=key, cast_names=cast)) def _library(db): lib = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True) db.add(lib) db.flush() return lib class _Tmdb: """A TmdbClient stand-in: records the ids asked for, fails on `fail_on`.""" def __init__(self, fail_on=None): self.asked: list[int] = [] self.fail_on = fail_on def fetch_person(self, pid): self.asked.append(pid) if pid == self.fail_on: raise plex_tmdb.TmdbError("429") return {"biography": f"Bio {pid}."} def test_the_batch_takes_the_most_credited_linked_people_first(db): lib = _library(db) _movie(db, lib, "a", ["Often", "Sometimes"]) _movie(db, lib, "b", ["Often", "Sometimes"]) _movie(db, lib, "c", ["Often", "Unlinked"]) db.add_all( [ PlexPerson(name="Sometimes", tmdb_person_id=2), PlexPerson(name="Often", tmdb_person_id=1), PlexPerson(name="Unlinked"), # no TMDB id: nothing to fetch PlexPerson(name="Done", tmdb_person_id=3, tmdb_enriched_at=DONE), PlexPerson(name="Rare", tmdb_person_id=4), # credited nowhere: last ] ) db.commit() tmdb = _Tmdb() assert plex_tmdb.enrich_people_batch(db, tmdb, 2) == {"enriched": 2} # type: ignore[arg-type] assert tmdb.asked == [1, 2] db.expire_all() often = db.query(PlexPerson).filter_by(name="Often").one() assert often.biography == "Bio 1." and often.tmdb_enriched_at is not None assert db.query(PlexPerson).filter_by(name="Rare").one().tmdb_enriched_at is None # next run def test_a_failed_lookup_ends_the_batch_and_keeps_what_landed(db): db.add_all([PlexPerson(name="A", tmdb_person_id=1), PlexPerson(name="B", tmdb_person_id=2)]) db.add(PlexPerson(name="C", tmdb_person_id=3)) db.commit() tmdb = _Tmdb(fail_on=2) assert plex_tmdb.enrich_people_batch(db, tmdb, 10) == {"enriched": 1, "failed": 1} # type: ignore[arg-type] assert tmdb.asked == [1, 2] # C is not asked: TMDB is struggling db.expire_all() stamped = {p.name: p.tmdb_enriched_at is not None for p in db.query(PlexPerson)} assert stamped == {"A": True, "B": False, "C": False} class _Plex: def __init__(self, fail=False): self.asked: list[tuple[str, int | None]] = [] self.fail = fail def __enter__(self): return self def __exit__(self, *a): return False def image_bytes(self, url, width=None, height=None): self.asked.append((url, width)) if self.fail: raise plex_people.PlexError("down") return b"small", "image/jpeg" @pytest.fixture def cache_dir(tmp_path, monkeypatch): monkeypatch.setattr(imgcache, "CACHE_DIR", tmp_path) return tmp_path def test_portraits_are_cached_small_and_replace_the_full_size_original(db, cache_dir, monkeypatch): lib = _library(db) _movie(db, lib, "a", ["Often"]) _movie(db, lib, "b", ["Often", "Cached"]) db.add_all( [ PlexPerson(name="Often", photo_thumb=plex_people.person_thumb_url(CDN + "often.jpg")), PlexPerson(name="Cached", photo_thumb=plex_people.person_thumb_url(CDN + "cached.jpg")), PlexPerson(name="Faceless"), ] ) db.commit() small, legacy = plex_people._portrait_keys(CDN + "often.jpg") imgcache.write(legacy, b"a big original", "image/jpeg") # from before E30 S6 imgcache.write(plex_people._portrait_keys(CDN + "cached.jpg")[0], b"x", "image/jpeg") plex = _Plex() monkeypatch.setattr(plex_people, "PlexClient", lambda d: plex) assert plex_people.warm_portraits(db, 10) == {"cached": 1} assert plex.asked == [ (CDN + "often.jpg", plex_people.PORTRAIT_BOX) ] # resized, the cached one skipped assert imgcache.read(small) == (b"small", "image/jpeg") assert not imgcache.has(legacy) def test_portraits_wait_for_plex(db, cache_dir, monkeypatch): db.add(PlexPerson(name="P", photo_thumb=plex_people.person_thumb_url(CDN + "p.jpg"))) db.commit() def not_configured(d): raise plex_people.PlexNotConfigured() monkeypatch.setattr(plex_people, "PlexClient", not_configured) assert plex_people.warm_portraits(db, 10) == {"cached": 0, "skipped": "disabled"} assert list(cache_dir.iterdir()) == [] # no full-size originals behind Plex's back def test_a_failing_plex_ends_the_portrait_run(db, cache_dir, monkeypatch): db.add_all( [ PlexPerson(name=f"P{i}", photo_thumb=plex_people.person_thumb_url(CDN + f"{i}.jpg")) for i in range(8) ] ) db.commit() plex = _Plex(fail=True) monkeypatch.setattr(plex_people, "PlexClient", lambda d: plex) assert plex_people.warm_portraits(db, 100) == {"cached": 0, "failed": 5} assert len(plex.asked) == 5 def test_a_phase_count_is_a_change_worth_an_audit_row(): assert scheduler._run_changed({"bios": {"enriched": 3}, "portraits": {"cached": 0}}) assert scheduler._run_changed({"recent": {"videos_added": 2}, "deep": {}}) assert not scheduler._run_changed({"bios": {"skipped": "x"}, "portraits": {"cached": 0}}) def test_the_person_page_says_a_bio_is_on_its_way(db, monkeypatch, override_dep): 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.routes.plex import people db.add_all( [ PlexPerson(name="Pending", tmdb_person_id=1), PlexPerson(name="Known", tmdb_person_id=2, tmdb_enriched_at=DONE, biography="Bio."), PlexPerson(name="Nobody"), ] ) user = User(email="enriching@example.com") db.add(user) db.commit() monkeypatch.setattr(state, "_configured_cache", True) override_dep(get_db, db.info["request"]) override_dep(current_user, lambda: user) monkeypatch.setattr(people, "_enrich_person_bg", lambda pid: None) key = {"value": "key"} monkeypatch.setattr(people.sysconfig, "get_str", lambda d, k: key["value"]) def enriching(name): resp = TestClient(app).get(f"/api/plex/person/{name}") assert resp.status_code == 200, resp.text return resp.json()["enriching"] assert enriching("Pending") is True assert enriching("Known") is False assert enriching("Nobody") is False key["value"] = "" # no TMDB key: nothing will ever arrive, so the page must not wait assert enriching("Pending") is False def test_known_for_is_the_most_voted_roles_not_guest_visits(): payload = { "combined_credits": { "cast": [ # Popular this week, but a visit as himself: not what he is known for. { "id": 1, "media_type": "tv", "name": "Late Show", "character": "Himself - Guest", "popularity": 900, "vote_count": 50, }, { "id": 2, "media_type": "tv", "name": "Talk Hour", "character": "", "genre_ids": [10767], "popularity": 800, "vote_count": 400, }, { "id": 3, "media_type": "movie", "title": "Small Film", "character": "Bob", "popularity": 90, "vote_count": 300, }, { "id": 4, "media_type": "movie", "title": "Big Film", "character": "Hero", "popularity": 10, "vote_count": 9000, }, ], "crew": [ { "id": 5, "media_type": "movie", "title": "His Own Film", "job": "Director", "vote_count": 1000, }, { "id": 6, "media_type": "movie", "title": "Thanked In", "job": "Thanks", "vote_count": 99999, }, ], } } kf = plex_tmdb.parse_person(payload)["known_for"] assert [k["title"] for k in kf] == ["Big Film", "His Own Film", "Small Film"] def test_the_page_shows_at_most_fifteen_known_for_titles(db): from app.routes.plex import people lib = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True) db.add(lib) db.flush() for i in range(1, 21): # TMDB ids start at 1 db.add( PlexItem( rating_key=f"k{i}", library_id=lib.id, kind="movie", title=f"T{i}", guid=f"tmdb:{i}" ) ) db.commit() pool = [{"tmdb_id": i, "media_type": "movie", "title": f"T{i}"} for i in range(1, 21)] cards = people._resolve_known_for(db, pool, 0, [lib.id], [], "en") assert [c["title"] for c in cards] == [f"T{i}" for i in range(1, 16)]