""" `WebRTCPeerSession._tmdb_search` — the movie retry ladder. Regression cover for a batch of wrong poster-grid matches found live on a real library (2026-08-29), all one mechanism: the ladder used to return the first candidate query whose title-similarity ratio merely cleared 0.6, so a wrong film that scored ~0.7 against guessit's weak bare title won before `alternative_title` or the Roman-numeral variant was ever tried. * A two-volume film's second part matched the *first* — guessit peels "Volume 2" into its own field, collapsing both parts onto one query, and the more popular first part is TMDB's top result for both. * A numbered sequel matched a same-year making-of documentary — TMDB's real entry uses a Roman numeral, so the "3" query surfaces the doc. * Two entries of one franchise matched a single early entry whose *localized* TMDB title is itself the parsed franchise name. * One entry matched nothing — every candidate query missed on text (the filename's spelling of the subtitle differs from TMDB's by one letter). The canned `(result, ratio)` tuples stand in for what the live TMDB API + the real `_best_match` return for each query; the ratios are the ones those queries actually produced when the mechanism was traced against the API. """ import pytest from meshbay_common.protocol import IndexEntry from meshbay_node.indexer.title_parse import naive_title, parse_movie_filename from meshbay_node.transport.webrtc_server import WebRTCPeerSession pytestmark = pytest.mark.asyncio # Stand-ins for real TMDB rows: a two-part film, a numbered sequel vs a # same-year documentary, and a franchise whose localized lead entry's title # is the franchise name. PART1 = {"id": "101", "title": "Some Saga: Volume 1", "release_date": "2003-10-10"} PART2 = {"id": "102", "title": "Some Saga: Volume 2", "release_date": "2004-04-16"} DOC = {"id": "201", "title": "Beyond Old Frontier", "release_date": "2001-11-01"} SEQUEL3 = {"id": "202", "title": "Old Frontier III", "release_date": "2001-07-18"} FRANCHISE_LEAD = {"id": "301", "title": "Some Agent 42 vs. Doctor X", "release_date": "1962-10-05"} ENTRY_2002 = {"id": "302", "title": "Second Errand", "release_date": "2002-11-20"} ENTRY_1997 = {"id": "303", "title": "First Errand", "release_date": "1997-12-12"} STANDALONE = {"id": "401", "title": "A Quiet Film", "release_date": "2010-07-15"} class LadderTmdb: """Canned `(result, ratio)` keyed by `(query, year)` — a hit for `(query, None)` also answers a year-constrained lookup, mirroring the real ladder's unconstrained retry.""" def __init__(self, table: dict): self._table = table self.calls: list[tuple] = [] async def search_movie(self, title, year=None): self.calls.append((title, year)) if (title, year) in self._table: return self._table[(title, year)] if (title, None) in self._table: return self._table[(title, None)] return None, 0.0 async def search_tv(self, title, year=None): self.calls.append((title, year)) if (title, year) in self._table: return self._table[(title, year)] if (title, None) in self._table: return self._table[(title, None)] return None, 0.0 async def _run(name: str, table: dict): """Drive the real `_tmdb_search` for a movie filename, deriving `display_title` exactly as enrich.py would from the current parser.""" parsed = parse_movie_filename(name) entry = IndexEntry( id="x", name=name, path="movies", size=1, type="video", added_at=0, display_title=parsed.display_title or parsed.naive_title, ) client = LadderTmdb(table) session = WebRTCPeerSession.__new__(WebRTCPeerSession) result, ratio = await session._tmdb_search(client, entry, is_show=False) return result, ratio, client async def _run_show(display_title: str, name: str, table: dict): entry = IndexEntry( id="s", name=name, path="Show/S1", size=1, type="video", added_at=0, display_title=display_title, season=1, episode=1, ) client = LadderTmdb(table) session = WebRTCPeerSession.__new__(WebRTCPeerSession) result, ratio = await session._tmdb_search(client, entry, is_show=True) return result, ratio, client # ── a two-part film: the parts must resolve to different entries ──────────── async def test_second_volume_resolves_to_the_second_volume(): result, ratio, client = await _run( "Some.Saga.Volume.2.2004.mkv", {("Some Saga 2", 2004): (PART2, 0.92)}, ) assert result["id"] == "102" # a strong first hit still costs exactly one request assert client.calls == [("Some Saga 2", 2004)] async def test_first_volume_still_resolves_to_the_first_volume(): result, _, _ = await _run( "Some.Saga.Volume.1.2003.mkv", {("Some Saga 1", 2003): (PART1, 0.92)}, ) assert result["id"] == "101" # ── numbered sequel: the Roman-numeral variant beats a same-year documentary ─ async def test_numbered_sequel_prefers_the_real_film_over_a_documentary(): naive = naive_title("Old.Frontier.3.2001.mkv") result, _, _ = await _run( "Old.Frontier.3.2001.mkv", { ("Old Frontier 3", 2001): (DOC, 0.80), # weak bare-title hit ("Old Frontier", 2001): (DOC, 0.76), ("Old Frontier III", 2001): (SEQUEL3, 1.0), # the sequel variant (naive, 2001): (DOC, 0.50), }, ) assert result["id"] == "202" # ── franchise: alternative_title beats the localized franchise name ──────── async def test_franchise_entry_uses_its_subtitle_not_the_franchise_name(): result, _, _ = await _run( "Some.Agent.42.-.2002.-.Second.Errand.mkv", { ("Some Agent 42", 2002): (FRANCHISE_LEAD, 0.70), # localized lead, popularity #1 ("Second Errand", 2002): (ENTRY_2002, 1.0), }, ) assert result["id"] == "302" async def test_franchise_alternative_title_still_wins_over_the_localized_lead(): # PASS 1 lands on the localized lead at 0.70 (not strong enough to # short-circuit); the ladder must go on to try the subtitle. result, _, _ = await _run( "Some.Agent.42.-.1995.-.Third.Errand.mkv", { ("Some Agent 42", 1995): (FRANCHISE_LEAD, 0.70), ("Third Errand", 1995): ({"id": "305", "title": "Third Errand", "release_date": "1995-11-16"}, 1.0), }, ) assert result["id"] == "305" # ── the year-exact rescue: every candidate query misses on text ──────────── async def test_rescued_by_exact_release_year_when_every_candidate_misses(): # PASS 1's own top hit IS the right film (TMDB year-filtered the search) # but scores far below 0.6 against the bare franchise title; the # alt-title and naive queries all miss because the filename spells the # subtitle differently from TMDB. result, ratio, _ = await _run( "Some.Agent.42.1997.First.Errand.mkv", {("Some Agent 42", 1997): (ENTRY_1997, 0.20)}, ) assert result["id"] == "303" # rescued to exactly the confidence floor — _do_media_meta_request keeps # a match at ratio >= 0.6, rejects one below it. assert ratio >= 0.6 async def test_year_rescue_does_not_fire_without_a_year_match(): # Same weak PASS 1, but the top hit's year does NOT match the filename's # → no rescue, stays sub-0.6, and _do_media_meta_request reports # confidence 0 rather than pinning a wrong film. result, ratio, _ = await _run( "Some.Agent.42.1997.First.Errand.mkv", {("Some Agent 42", 1997): ({"id": "999", "title": "An Old Film", "release_date": "1962-01-01"}, 0.20)}, ) assert ratio < 0.6 # ── the ladder must not let a weaker later candidate override a good hit ──── async def test_a_strong_first_hit_is_not_overridden_by_a_weaker_variant(): result, _, _ = await _run( "The.Thing.-.2011.-.Wrong.Subtitle.mkv", { ("The Thing", 2011): ({"id": "1", "title": "The Thing", "release_date": "2011-10-14"}, 0.80), ("Wrong Subtitle", 2011): ({"id": "999", "title": "Wrong Subtitle", "release_date": "2011-01-01"}, 0.75), }, ) assert result["id"] == "1" async def test_strong_direct_match_costs_a_single_request(): _, _, client = await _run("A.Quiet.Film.2010.mkv", {("A Quiet Film", 2010): (STANDALONE, 1.0)}) assert client.calls == [("A Quiet Film", 2010)] # ── §10.1/V11: a decent primary hit with nothing more specific to try ────── async def test_decent_primary_with_no_stronger_candidate_costs_one_call(): result, _, client = await _run( "A.Quiet.Film.2010.mkv", {("A Quiet Film", 2010): ({"id": "77", "title": "A Quiet Movie", "release_date": "2010-01-01"}, 0.70)}, ) assert result["id"] == "77" assert client.calls == [("A Quiet Film", 2010)], ( "no alt title, no sequel index → the punctuation-restatement fallback " "is not worth a second request once the primary hit is decent") # ── §10.1/V8: the show branch uses the same scored ladder ───────────────── async def test_show_scored_ladder_beats_a_weak_primary_hit(): result, _, _ = await _run_show( "Some.Show.Name.S01", "s01e01.mkv", { ("Some.Show.Name.S01", None): ({"id": "10", "name": "Promo Special", "first_air_date": "2019-01-01"}, 0.55), ("Some Show Name S01", None): ({"id": "20", "name": "Some Show Name", "first_air_date": "2017-01-01"}, 0.95), }, ) assert result["id"] == "20" async def test_show_year_in_folder_name_rescues_a_weak_hit(): result, ratio, _ = await _run_show( "Some.Show.2019.S01", "s01e01.mkv", {("Some.Show.2019.S01", 2019): ({"id": "30", "name": "Some Show", "first_air_date": "2019-05-01"}, 0.30)}, ) assert result["id"] == "30" assert ratio >= 0.6