diff options
Diffstat (limited to 'packages/meshbay-node/tests')
| -rw-r--r-- | packages/meshbay-node/tests/test_media_cache.py | 13 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_title_parse.py | 44 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_tmdb.py | 53 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_tmdb_rematch_policy.py | 147 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_tmdb_search_ladder.py | 58 |
5 files changed, 313 insertions, 2 deletions
diff --git a/packages/meshbay-node/tests/test_media_cache.py b/packages/meshbay-node/tests/test_media_cache.py index 4e65b24..f12a366 100644 --- a/packages/meshbay-node/tests/test_media_cache.py +++ b/packages/meshbay-node/tests/test_media_cache.py @@ -107,6 +107,19 @@ async def test_clear_file_tmdb_drops_one_auto_match_but_keeps_an_override(cache) @pytest.mark.asyncio +async def test_drop_tmdb_match_forgets_both_the_match_and_the_override(cache): + await cache.set_file_tmdb("f", "10", "movie") + await cache.mark_tmdb_override("f") + + await cache.drop_tmdb_match("f") + + assert await cache.get_file_tmdb("f") is None + # override marker is gone: a fresh match is now treated as ordinary + await cache.set_file_tmdb("f", "20", "movie") + assert await cache.clear_tmdb_matches(["f"]) == 1 + + +@pytest.mark.asyncio async def test_prune_file_also_clears_the_override_marker(cache): await cache.set_file_tmdb("gone", "10", "movie") await cache.mark_tmdb_override("gone") diff --git a/packages/meshbay-node/tests/test_title_parse.py b/packages/meshbay-node/tests/test_title_parse.py index 03fb588..ac4d434 100644 --- a/packages/meshbay-node/tests/test_title_parse.py +++ b/packages/meshbay-node/tests/test_title_parse.py @@ -6,12 +6,14 @@ is a manual acceptance step (§11), not something this repo's corpus holds. from meshbay_node.indexer.title_parse import ( ParsedName, + clean_query, leading_episode_number, naive_title, parse_episode_filename, parse_movie_filename, season_from_folder_name, sequel_variants, + year_in, ) @@ -69,6 +71,48 @@ def test_sequel_variants_empty_when_no_trailing_digit(): assert sequel_variants("Some Movie") == [] +# ── §10.1/V10: wider sequel-index handling ────────────────────────────────── + +def test_sequel_variants_roman_numeral_offers_the_digit_form(): + v = sequel_variants("Old Frontier III") + assert "Old Frontier" in v + assert "Old Frontier 3" in v + + +def test_sequel_variants_strips_a_part_keyword_wrapper(): + v = sequel_variants("Some Saga Part 2") + assert "Some Saga" in v + assert "Some Saga II" in v + + +def test_sequel_variants_reads_a_spelled_out_index(): + v = sequel_variants("Story Chapter Three") + assert "Story" in v + assert "Story 3" in v and "Story III" in v + + +def test_sequel_variants_ignores_a_trailing_word_that_is_not_an_index(): + assert sequel_variants("The Dark Knight") == [] + assert sequel_variants("In Bruges") == [] + + +def test_sequel_variants_ignores_a_four_digit_year_suffix(): + assert sequel_variants("Blade Runner 2049") == [] + + +def test_year_in_lifts_a_year_from_a_show_folder_name(): + assert year_in("Some.Show.2022.S01") == 2022 + assert year_in("Some Show") is None + assert year_in("Episode 100 of 2010") == 2010 + + +def test_clean_query_despaces_a_folder_name_without_eating_the_last_word(): + # naive_title would rsplit on the last dot and drop ".Name" + assert clean_query("Some.Show.Name") == "Some Show Name" + assert clean_query("Some.Show.Name.S01") == "Some Show Name S01" + assert naive_title("Some.Show.Name") != "Some Show Name" # the trap it avoids + + # ── bug 2026-08-29: guessit peels "Volume N" off the title ───────────────── def test_movie_volume_number_is_folded_back_into_the_title(): diff --git a/packages/meshbay-node/tests/test_tmdb.py b/packages/meshbay-node/tests/test_tmdb.py index fcbc2a2..ab452ce 100644 --- a/packages/meshbay-node/tests/test_tmdb.py +++ b/packages/meshbay-node/tests/test_tmdb.py @@ -40,6 +40,59 @@ async def test_search_movie_returns_top_result_and_confidence(): await client.close() +# ── §10.1/V9: year-exact preference, only when the top hit is weak ────────── + +@pytest.mark.asyncio +async def test_year_exact_result_wins_when_the_top_hit_is_low_confidence(): + # results[0] is TMDB's popularity #1 but a poor textual match for the + # query; results[2] is the exact requested year. + body = {"results": [ + {"id": 1, "title": "Franchise vs. The Doctor", "release_date": "1962-10-05"}, + {"id": 2, "title": "Franchise: Goldfinger", "release_date": "1964-09-17"}, + {"id": 3, "title": "Second Errand", "release_date": "2002-11-20"}, + ]} + client = TmdbClient( + roster=FakeRoster(), + transport=httpx.MockTransport(_handler({"search/movie": body})), + ) + result, _ = await client.search_movie("The Franchise", 2002) + + assert result["id"] == 3 + await client.close() + + +@pytest.mark.asyncio +async def test_year_is_ignored_when_the_top_hit_is_already_confident(): + body = {"results": [ + {"id": 1, "title": "The Franchise", "release_date": "1999-01-01"}, + {"id": 2, "title": "Unrelated", "release_date": "2002-01-01"}, + ]} + client = TmdbClient( + roster=FakeRoster(), + transport=httpx.MockTransport(_handler({"search/movie": body})), + ) + result, ratio = await client.search_movie("The Franchise", 2002) + + assert result["id"] == 1 and ratio > 0.9 + await client.close() + + +@pytest.mark.asyncio +async def test_no_year_match_leaves_the_top_result_in_place(): + body = {"results": [ + {"id": 1, "title": "Something Else Entirely", "release_date": "1990-01-01"}, + {"id": 2, "title": "Also Not It", "release_date": "1991-01-01"}, + ]} + client = TmdbClient( + roster=FakeRoster(), + transport=httpx.MockTransport(_handler({"search/movie": body})), + ) + result, _ = await client.search_movie("The Franchise", 2002) + + assert result["id"] == 1 + await client.close() + + @pytest.mark.asyncio async def test_search_tv_returns_top_result(): body = {"results": [{"id": 7, "name": "Some Show"}]} diff --git a/packages/meshbay-node/tests/test_tmdb_rematch_policy.py b/packages/meshbay-node/tests/test_tmdb_rematch_policy.py new file mode 100644 index 0000000..ff259c1 --- /dev/null +++ b/packages/meshbay-node/tests/test_tmdb_rematch_policy.py @@ -0,0 +1,147 @@ +""" +`tmdb_rematch` (§10.1/V13) — an operator dropping one file's cached TMDB +match so it re-resolves with the current matcher. Signed like +`tmdb_override` (media_cache is shared node-wide); unlike `clear_file_tmdb` +it forgets a manual override marker too, since the operator is explicitly +asking for a fresh resolution. +""" + +import hashlib + +import pytest +from conftest import one_root +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from meshbay_common.adminop import OP_TMDB_REMATCH +from meshbay_common.protocol import MNP, IndexEntry +from meshbay_node.indexer.group_index import GroupIndex +from meshbay_node.media_cache import MediaCache +from meshbay_node.transport.webrtc_server import WebRTCPeerSession + +pytestmark = pytest.mark.asyncio + + +def _session(tmp_path, user_id, *, operator=None): + shared = tmp_path / "shared" + shared.mkdir(exist_ok=True) + index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate()) + s = WebRTCPeerSession.__new__(WebRTCPeerSession) + s._ctx = {"roots": one_root(shared), "index": index, "sk_node": index.sk_node, + "node_user_id": operator} + s._group_id = None + s._user_id = user_id + s._pk_user = "" + s.sent = [] + s._send = s.sent.append + s._audit = lambda *a, **k: None + return s + + +def _entry(name): + digest = hashlib.sha256(name.encode()).hexdigest() + return IndexEntry(id=digest, name=name, path="movies", size=1, type="video", + added_at=0, display_title="Some Film") + + +async def _true(): + return True + + +async def test_missing_file_id_is_refused(tmp_path): + s = _session(tmp_path, "op", operator="op") + s._has_admin_authority = lambda: True + issued = [] + s._issue_admin_challenge = lambda op, subject: issued.append((op, subject)) + + s._do_tmdb_rematch({}) + + assert not issued + assert [m for m in s.sent if m.get("type") == "error"] + + +async def test_unknown_file_id_is_refused(tmp_path): + s = _session(tmp_path, "op", operator="op") + s._has_admin_authority = lambda: True + issued = [] + s._issue_admin_challenge = lambda op, subject: issued.append((op, subject)) + + s._do_tmdb_rematch({"file_id": "nope"}) + + assert not issued + assert [m for m in s.sent if m.get("type") == "error"] + + +async def test_no_authorized_key_is_refused(tmp_path): + s = _session(tmp_path, "member", operator="the-operator") + e = _entry("some.film.2001.mkv") + s._ctx["index"].add_entry(e) + s._has_admin_authority = lambda: False + + s._do_tmdb_rematch({"file_id": e.id}) + + assert [m for m in s.sent if m.get("type") == "error"] + + +async def test_a_valid_request_is_signed(tmp_path): + s = _session(tmp_path, "op", operator="op") + e = _entry("some.film.2001.mkv") + s._ctx["index"].add_entry(e) + s._has_admin_authority = lambda: True + issued = [] + s._issue_admin_challenge = lambda op, subject: issued.append((op, subject)) + + s._do_tmdb_rematch({"file_id": e.id}) + + assert issued == [(OP_TMDB_REMATCH, f"file_id={e.id}")] + + +async def test_exec_drops_the_match_and_the_override_marker(tmp_path): + s = _session(tmp_path, "op", operator="op") + e = _entry("some.film.2001.mkv") + s._ctx["index"].add_entry(e) + + media_cache = MediaCache(db_path=tmp_path / "media_cache.db") + await media_cache.open() + try: + s._ctx["media_cache"] = media_cache + await media_cache.set_file_tmdb(e.id, "wrong-id", "movie") + await media_cache.mark_tmdb_override(e.id) + s._verify_admin_sig = lambda transcript, sig: _true() + peer = type("Peer", (), {"sent": []})() + peer._send = peer.sent.append + s._peer_registry = lambda: {"peer-1": peer} + + await s._admin_exec_tmdb_rematch( + {"subject": f"file_id={e.id}"}, b"transcript", b"sig") + + assert await media_cache.get_file_tmdb(e.id) is None + # the override marker is gone too, so a later group-wide rematch + # would treat a fresh match as ordinary + await media_cache.set_file_tmdb(e.id, "fresh", "movie") + assert await media_cache.clear_tmdb_matches([e.id]) == 1 + assert [m for m in peer.sent if m.get("type") == MNP.TMDB_REMATCH_ACK] + finally: + await media_cache.close() + + +async def test_exec_refuses_a_bad_signature(tmp_path): + s = _session(tmp_path, "op", operator="op") + e = _entry("some.film.2001.mkv") + s._ctx["index"].add_entry(e) + + media_cache = MediaCache(db_path=tmp_path / "media_cache.db") + await media_cache.open() + try: + s._ctx["media_cache"] = media_cache + await media_cache.set_file_tmdb(e.id, "keep-me", "movie") + + async def _false(): + return False + s._verify_admin_sig = lambda transcript, sig: _false() + + await s._admin_exec_tmdb_rematch( + {"subject": f"file_id={e.id}"}, b"transcript", b"badsig") + + assert await media_cache.get_file_tmdb(e.id) == ("keep-me", "movie") + assert [m for m in s.sent if m.get("type") == "error"] + finally: + await media_cache.close() diff --git a/packages/meshbay-node/tests/test_tmdb_search_ladder.py b/packages/meshbay-node/tests/test_tmdb_search_ladder.py index 49b71bd..9865c77 100644 --- a/packages/meshbay-node/tests/test_tmdb_search_ladder.py +++ b/packages/meshbay-node/tests/test_tmdb_search_ladder.py @@ -61,8 +61,12 @@ class LadderTmdb: return self._table[(title, None)] return None, 0.0 - async def search_tv(self, title): - self.calls.append(("tv", title)) + 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 @@ -80,6 +84,17 @@ async def _run(name: str, table: dict): 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(): @@ -191,3 +206,42 @@ 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 |