1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
|
"""
`_do_media_meta_request`, keyed by `file_id` (2026-08-25 fix) — same
regression as test_music_meta_request.py, one app over: `IndexEntry.path`
is the *folder* a file is in, not the file itself, so a lookup by path
alone (the pre-fix `GroupIndex.get_entry_by_path`) silently resolved to
whichever entry the index happened to return first for that folder — a
real risk here too, since a season folder routinely holds many episodes.
"""
import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from meshbay_common.protocol import 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
class FakeTmdbClient:
"""Returns a distinct, deterministic match per entry — real enough to
prove the server searched using the *right* entry's own fields."""
def __init__(self):
self.searched = []
async def search_movie(self, title):
self.searched.append(("movie", title))
return {"id": 1000 + len(self.searched), "title": title,
"release_date": "2001-01-01"}, 1.0
async def search_tv(self, title):
self.searched.append(("tv", title))
return {"id": 2000 + len(self.searched), "name": title,
"first_air_date": "2001-01-01"}, 1.0
async def fetch_image(self, url):
return f"image-bytes-for-{url}".encode()
@staticmethod
def poster_url(path):
return f"https://image.tmdb.org/t/p/w500{path}"
def _entry(path: str, name: str, file_id: str, display_title: str) -> IndexEntry:
return IndexEntry(
id=file_id, name=name, path=path, size=1, type="video", added_at=0,
display_title=display_title,
)
@pytest.fixture
async def media_cache(tmp_path):
c = MediaCache(db_path=tmp_path / "media_cache.db")
await c.open()
yield c
await c.close()
def _session(index, media_cache, tmdb_client):
session = WebRTCPeerSession.__new__(WebRTCPeerSession)
session._ctx = {
"index": index,
"media_cache": media_cache,
"tmdb_client": tmdb_client,
"tmdb_enabled": True,
}
session._group_id = None
session.sent = []
session._send = session.sent.append
# _tmdb_search/_tmdb_build_meta are the real methods (not part of this
# regression) — stub the search ladder to a single direct call by
# display_title so this test is about routing, not TMDB matching.
async def _search(tmdb_client_, entry, is_show):
return await (tmdb_client_.search_tv(entry.display_title) if is_show
else tmdb_client_.search_movie(entry.display_title))
session._tmdb_search = lambda *a: _search(*a)
async def _build_meta(tmdb_client_, tmdb_id, media_type, result):
return {
"title": result.get("title") or result.get("name"),
"original_title": result.get("title") or result.get("name"),
"release_date": result.get("release_date"),
"first_air_date": result.get("first_air_date"),
"confidence": 1.0,
}
session._tmdb_build_meta = lambda *a: _build_meta(*a)
return session
async def test_two_episodes_in_the_same_season_folder_each_get_their_own_metadata(media_cache):
"""The Videos-side analogue of the Music bug: two episodes share a
season folder, and each must resolve against its own entry — not
whichever one the index happens to return first for that folder."""
index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate())
ep1 = _entry("shows/Show/Season 1", "s01e01.mkv", "id-1", "War of the Worlds")
ep2 = _entry("shows/Show/Season 1", "s01e02.mkv", "id-2", "A Different Show")
index.add_entry(ep1)
index.add_entry(ep2)
client = FakeTmdbClient()
session = _session(index, media_cache, client)
await session._do_media_meta_request({"file_id": "id-1"})
await session._do_media_meta_request({"file_id": "id-2"})
resp_1, resp_2 = session.sent
assert resp_1["file_id"] == "id-1"
assert resp_1["title"] == "War of the Worlds"
assert resp_2["file_id"] == "id-2"
assert resp_2["title"] == "A Different Show"
assert resp_1["tmdb_id"] != resp_2["tmdb_id"], (
"two different shows sharing a season folder must not resolve to the same match")
async def test_missing_file_id_is_refused(media_cache):
index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate())
session = _session(index, media_cache, FakeTmdbClient())
await session._do_media_meta_request({})
assert session.sent == [{"type": "error", "detail": "Missing file_id"}]
async def test_unknown_file_id_is_refused(media_cache):
index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate())
session = _session(index, media_cache, FakeTmdbClient())
await session._do_media_meta_request({"file_id": "nope"})
assert session.sent == [{"type": "error", "detail": "File not found"}]
|