path: root/packages/meshbay-node/tests/test_tmdb_override_policy.py

"""
An operator correcting a wrong automatic TMDB match (found live: a real
show's search consistently matched a season-3-specific promotional TMDB
entry instead of the show itself). Signed like video_root/tmdb_config —
it changes what every member sees, node-wide (media_cache is shared, not
per-viewer) — and, once authorized, applies to every index entry sharing
the representative file's display_title, the same grouping the poster
grid itself uses (§3.4/§V6), not just the one file the operator happened
to be looking at.
"""

import hashlib
from pathlib import Path

import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

from meshbay_common.adminop import OP_TMDB_OVERRIDE
from meshbay_common.protocol import IndexEntry, MNP
from meshbay_node.indexer.group_index import GroupIndex
from meshbay_node.media_cache import MediaCache
from meshbay_node.transport.webrtc_server import WebRTCPeerSession

from conftest import one_root

pytestmark = pytest.mark.asyncio


def _session(tmp_path: Path, user_id: str, *, operator: str | None = None) -> WebRTCPeerSession:
    shared_root = tmp_path / "shared"
    shared_root.mkdir(exist_ok=True)
    index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate())
    ctx = {
        "roots": one_root(shared_root),
        "index": index,
        "sk_node": index.sk_node,
        "node_user_id": operator,
    }
    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
    session._ctx = ctx
    session._group_id = None
    session._user_id = user_id
    session._pk_user = ""
    session.sent = []
    session._send = session.sent.append
    session._audit = lambda *a, **k: None
    return session


def _entry(path: str, name: str, display_title: str) -> IndexEntry:
    # A real id is a blake3 content hash; sha256 here is just a stand-in with
    # the same property that matters for these tests — deterministic and
    # effectively collision-free across the handful of entries a test builds.
    # (`hash((path, name)) % 10` was tried here before and is NOT that: it's
    # randomized per-process by PYTHONHASHSEED and collides constantly across
    # only 10 possible values, silently dropping entries in GroupIndex's
    # id-keyed dict.)
    digest = hashlib.sha256(f"{path}/{name}".encode()).hexdigest()
    return IndexEntry(
        id=digest, name=name, path=path,
        size=1, type="video", added_at=0, display_title=display_title,
        season=1, episode=1,
    )


# ── Refused before a challenge is even issued ───────────────────────────────

async def test_missing_file_id_is_refused(tmp_path):
    session = _session(tmp_path, "op", operator="op")
    session._has_admin_authority = lambda: True
    issued = []
    session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))

    session._do_tmdb_override({"tmdb_id": "123", "media_type": "tv"})

    assert not issued
    assert [m for m in session.sent if m.get("type") == "error"]


async def test_missing_tmdb_id_is_refused(tmp_path):
    session = _session(tmp_path, "op", operator="op")
    entry = _entry("shared", "ep.mkv", "Show")
    session._ctx["index"].add_entry(entry)
    session._has_admin_authority = lambda: True
    issued = []
    session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))

    session._do_tmdb_override({"file_id": entry.id, "media_type": "tv"})

    assert not issued
    assert [m for m in session.sent if m.get("type") == "error"]


async def test_unknown_file_id_is_refused(tmp_path):
    session = _session(tmp_path, "op", operator="op")
    session._has_admin_authority = lambda: True
    issued = []
    session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))

    session._do_tmdb_override({"file_id": "nope", "tmdb_id": "123", "media_type": "tv"})

    assert not issued
    assert [m for m in session.sent if m.get("type") == "error"]


async def test_a_request_with_nobody_to_authorize_it_is_refused(tmp_path):
    session = _session(tmp_path, "member-1", operator="the-operator")
    entry = _entry("shared", "ep.mkv", "Show")
    session._ctx["index"].add_entry(entry)
    session._has_admin_authority = lambda: False

    session._do_tmdb_override({"file_id": entry.id, "tmdb_id": "123", "media_type": "tv"})

    assert [m for m in session.sent if m.get("type") == "error"]


async def test_a_valid_request_is_signed(tmp_path):
    session = _session(tmp_path, "op", operator="op")
    entry = _entry("shared", "ep.mkv", "War of the Worlds")
    session._ctx["index"].add_entry(entry)
    session._has_admin_authority = lambda: True
    issued = []
    session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))

    session._do_tmdb_override({"file_id": entry.id, "tmdb_id": "2255", "media_type": "tv"})

    assert issued == [(OP_TMDB_OVERRIDE,
                       f"file_id={entry.id},tmdb_id=2255,media_type=tv")]


async def test_two_files_in_the_same_folder_are_told_apart(tmp_path):
    """
    Regression (found live, 2026-08-25): `IndexEntry.path` is the *folder* a
    file is in, not the file itself — two files in the same folder (any
    multi-episode season) used to collide when looked up by path, silently
    resolving to whichever entry the index happened to return first. Keyed
    by `file_id` now, so two entries sharing a folder must resolve to their
    own, distinct entries.
    """
    session = _session(tmp_path, "op", operator="op")
    e1 = _entry("shared/Season 1", "s01e01.mkv", "War of the Worlds")
    e2 = _entry("shared/Season 1", "s01e02.mkv", "War of the Worlds")
    session._ctx["index"].add_entry(e1)
    session._ctx["index"].add_entry(e2)
    session._has_admin_authority = lambda: True
    issued = []
    session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))

    session._do_tmdb_override({"file_id": e2.id, "tmdb_id": "2255", "media_type": "tv"})

    assert issued == [(OP_TMDB_OVERRIDE,
                       f"file_id={e2.id},tmdb_id=2255,media_type=tv")]


# ── Applying the override ───────────────────────────────────────────────────

async def test_override_updates_every_entry_sharing_the_display_title(tmp_path):
    session = _session(tmp_path, "op", operator="op")
    index = session._ctx["index"]
    s1 = _entry("shared/S1", "s01e01.mkv", "War of the Worlds")
    s2 = _entry("shared/S2", "s02e01.mkv", "War of the Worlds")
    s3 = _entry("shared/S3", "s03e02.mkv", "War of the Worlds")
    other_show = _entry("shared/Other", "ep.mkv", "A Different Show")
    for e in (s1, s2, s3, other_show):
        index.add_entry(e)

    media_cache = MediaCache(db_path=tmp_path / "media_cache.db")
    await media_cache.open()
    try:
        session._ctx["media_cache"] = media_cache
        # Signature verification itself is exercised generically elsewhere
        # (test_roster_pairing.py) — this test is about the policy once a
        # signature is known good: which entries actually get updated, and
        # who is told about it.
        session._verify_admin_sig = lambda transcript, sig: _true()
        peer = type("Peer", (), {"sent": []})()
        peer._send = peer.sent.append
        session._peer_registry = lambda: {"peer-1": peer}

        await session._admin_exec_tmdb_override(
            {"subject": f"file_id={s1.id},tmdb_id=999,media_type=tv"},
            b"transcript", b"sig")

        for e in (s1, s2, s3):
            assert await media_cache.get_file_tmdb(e.id) == ("999", "tv"), (
                "every entry sharing the representative file's display_title "
                "must be corrected, not just the one the operator clicked on")
        assert await media_cache.get_file_tmdb(other_show.id) is None, (
            "a different show's own match must be left alone")
        # The ack is broadcast to other connected peers, never echoed onto
        # the requester's own `sent` — see the loop in
        # _admin_exec_tmdb_override, which sends via each peer's own _send.
        assert [m for m in peer.sent if m.get("type") == MNP.TMDB_OVERRIDE_ACK]
    finally:
        await media_cache.close()


class _FakeTmdbClient:
    """Just enough for _tmdb_build_meta to run end to end — a fixed,
    deterministic response, not a search stub (the override already has a
    chosen tmdb_id; nothing here should need to search for anything)."""

    async def movie_details(self, tmdb_id, language=None):
        return {"title": "The Corrected Title", "overview": "A correct overview.",
                "poster_path": "/poster.jpg", "genres": [{"name": "Drama"}]}

    async def tv_details(self, tmdb_id, language=None):
        return await self.movie_details(tmdb_id, language)

    async def movie_credits(self, tmdb_id):
        return {"cast": [], "crew": []}

    async def tv_credits(self, tmdb_id):
        return {"cast": [], "crew": []}


async def test_override_stores_the_chosen_matchs_metadata_not_just_its_id(tmp_path):
    """
    The real bug this guards: only the file->tmdb_id mapping ever got
    recorded, never the metadata the chosen id actually names.
    _do_media_meta_request's cache check agrees the mapping is fresh (same
    media_type) but finds nothing under that id in tmdb_meta — nothing had
    ever fetched it — and falls through to a brand new search using the
    file's own title, reproducing the very match the override was meant to
    replace. Confirmed live: this stayed invisible for shows whose own
    title happened to be enough for that fallback search to land on the
    right answer anyway, and surfaced on a movie whose own title kept
    landing on the same wrong match regardless of the override.
    """
    session = _session(tmp_path, "op", operator="op")
    index = session._ctx["index"]
    entry = _entry("shared", "movie.mkv", "Some Movie's Own Wrong Title")
    index.add_entry(entry)

    media_cache = MediaCache(db_path=tmp_path / "media_cache.db")
    await media_cache.open()
    try:
        session._ctx["media_cache"] = media_cache
        session._ctx["tmdb_client"] = _FakeTmdbClient()
        session._verify_admin_sig = lambda transcript, sig: _true()
        session._peer_registry = lambda: {}

        await session._admin_exec_tmdb_override(
            {"subject": f"file_id={entry.id},tmdb_id=999,media_type=movie"},
            b"transcript", b"sig")

        meta = await media_cache.get_tmdb_meta("999", "movie")
        assert meta is not None, (
            "the override must store the metadata its chosen id actually names, "
            "not just the file->tmdb_id mapping")
        assert meta["title"] == "The Corrected Title"
    finally:
        await media_cache.close()


async def _true():
    return True