1 files changed, 172 insertions, 0 deletions
diff --git a/packages/meshbay-node/tests/test_tmdb_override_policy.py b/packages/meshbay-node/tests/test_tmdb_override_policy.py
new file mode 100644
index 0000000..b8fa6f9
--- /dev/null
+++ b/packages/meshbay-node/tests/test_tmdb_override_policy.py
@@ -0,0 +1,172 @@
+"""
+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_path_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")
+    session._ctx["index"].add_entry(_entry("shared", "ep.mkv", "Show"))
+    session._has_admin_authority = lambda: True
+    issued = []
+    session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
+
+    session._do_tmdb_override({"path": "shared", "media_type": "tv"})
+
+    assert not issued
+    assert [m for m in session.sent if m.get("type") == "error"]
+
+
+async def test_unknown_path_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({"path": "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")
+    session._ctx["index"].add_entry(_entry("shared", "ep.mkv", "Show"))
+    session._has_admin_authority = lambda: False
+
+    session._do_tmdb_override({"path": "shared", "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")
+    session._ctx["index"].add_entry(_entry("shared", "ep.mkv", "War of the Worlds"))
+    session._has_admin_authority = lambda: True
+    issued = []
+    session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
+
+    session._do_tmdb_override({"path": "shared", "tmdb_id": "2255", "media_type": "tv"})
+
+    assert issued == [(OP_TMDB_OVERRIDE,
+                       "path=shared,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": "path=shared/S1,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()
+
+
+async def _true():
+    return True