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
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
|
"""
`_do_season_meta_request` (per-season TMDB overview/poster/air_date, for the
season-tab view — docs/mediacenter.md §5.4's fix for a 3-season show whose
overview read as season-3-specific for every season) and
`_do_tmdb_search_request` (raw TMDB candidates for an operator correcting a
wrong automatic match). Neither is a signed admin op — see each handler's own
docstring for why — so these tests only exercise the read path, unlike
test_tmdb_override_policy.py.
"""
import pytest
from meshbay_common.protocol import MNP
from meshbay_node.media_cache import MediaCache
from meshbay_node.transport.webrtc_server import WebRTCPeerSession
pytestmark = pytest.mark.asyncio
def _session(media_cache=None, tmdb_client=None) -> WebRTCPeerSession:
session = WebRTCPeerSession.__new__(WebRTCPeerSession)
session._ctx = {"media_cache": media_cache, "tmdb_client": tmdb_client}
session._group_id = None
session.sent = []
session._send = session.sent.append
return session
@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()
class FakeTmdbClient:
def __init__(self, season_json=None):
self.season_json = season_json
self.tv_season_calls = []
self.movie_search_calls = []
self.tv_search_calls = []
@staticmethod
def poster_url(path):
return f"https://image.tmdb.org/t/p/w500{path}"
async def fetch_image(self, url):
return b"jpeg-bytes-for-" + url.encode()
async def tv_season(self, tmdb_id, season, language=None):
self.tv_season_calls.append((tmdb_id, season, language))
return self.season_json
async def search_movie_results(self, title):
self.movie_search_calls.append(title)
return [{"id": 111, "title": title, "release_date": "2019-05-01", "poster_path": "/m.jpg"}]
async def search_tv_results(self, title):
self.tv_search_calls.append(title)
return [{"id": 222, "name": title, "first_air_date": "2021-03-01", "poster_path": "/t.jpg"}]
# ── season_meta_req ──────────────────────────────────────────────────────────
async def test_season_meta_missing_tmdb_id_is_refused():
session = _session()
await session._do_season_meta_request({"season": 1})
assert session.sent == [{"type": "error", "detail": "Missing tmdb_id or season"}]
async def test_season_meta_non_int_season_is_refused():
session = _session()
await session._do_season_meta_request({"tmdb_id": "42", "season": "1"})
assert session.sent == [{"type": "error", "detail": "Missing tmdb_id or season"}]
async def test_season_meta_with_no_cache_or_client_reports_zero_confidence():
session = _session(media_cache=None, tmdb_client=None)
await session._do_season_meta_request({"tmdb_id": "42", "season": 1})
assert session.sent == [{
"type": MNP.SEASON_META_RESP, "v": session.sent[0]["v"],
"tmdb_id": "42", "season": 1, "confidence": 0,
}]
async def test_season_meta_cache_hit_skips_the_tmdb_call(media_cache):
await media_cache.set_season_meta("42", 3, {
"name": "Season 3", "overview": "cached overview", "air_date": "2023-01-01",
"poster_path": "/cached.jpg",
})
client = FakeTmdbClient()
session = _session(media_cache=media_cache, tmdb_client=client)
await session._do_season_meta_request({"tmdb_id": "42", "season": 3})
assert client.tv_season_calls == [], "a cached season must not be re-fetched"
resp = session.sent[0]
assert resp["type"] == MNP.SEASON_META_RESP
assert resp["confidence"] == 1.0
assert resp["overview"] == "cached overview"
async def test_season_meta_cache_miss_fetches_and_caches(media_cache):
client = FakeTmdbClient(season_json={
"name": "Season 1", "overview": "fresh overview", "air_date": "2020-01-01",
"poster_path": "/fresh.jpg",
})
session = _session(media_cache=media_cache, tmdb_client=client)
await session._do_season_meta_request({"tmdb_id": "7", "season": 1})
assert client.tv_season_calls == [("7", 1, None)]
resp = session.sent[0]
assert resp["overview"] == "fresh overview"
assert resp["poster_thumb_hash"] is not None
cached = await media_cache.get_season_meta("7", 1)
assert cached["overview"] == "fresh overview", "a fetched season must be cached for next time"
async def test_season_meta_empty_overview_falls_back_to_english(media_cache):
async def tv_season(tmdb_id, season, language=None):
if language == "en-US":
return {"name": "S1", "overview": "English overview", "air_date": "2020-01-01",
"poster_path": "/p.jpg"}
return {"name": "S1", "overview": "", "air_date": "2020-01-01", "poster_path": "/p.jpg"}
client = FakeTmdbClient()
client.tv_season = tv_season
session = _session(media_cache=media_cache, tmdb_client=client)
await session._do_season_meta_request({"tmdb_id": "9", "season": 1})
assert session.sent[0]["overview"] == "English overview"
# ── tmdb_search_req ──────────────────────────────────────────────────────────
async def test_search_missing_query_is_refused():
session = _session()
await session._do_tmdb_search_request({"media_type": "movie"})
assert session.sent == [{"type": "error", "detail": "Missing query or media_type"}]
async def test_search_bad_media_type_is_refused():
session = _session()
await session._do_tmdb_search_request({"query": "war", "media_type": "album"})
assert session.sent == [{"type": "error", "detail": "Missing query or media_type"}]
async def test_search_with_no_cache_or_client_returns_empty_results():
session = _session(media_cache=None, tmdb_client=None)
await session._do_tmdb_search_request({"query": "war", "media_type": "tv"})
assert session.sent == [{
"type": MNP.TMDB_SEARCH_RESP, "v": session.sent[0]["v"],
"query": "war", "media_type": "tv", "results": [],
}]
async def test_search_movie_calls_movie_search_and_echoes_media_type(media_cache):
client = FakeTmdbClient()
session = _session(media_cache=media_cache, tmdb_client=client)
await session._do_tmdb_search_request({"query": "Some Show", "media_type": "movie"})
assert client.movie_search_calls == ["Some Show"]
assert client.tv_search_calls == []
resp = session.sent[0]
assert resp["type"] == MNP.TMDB_SEARCH_RESP
assert resp["media_type"] == "movie", (
"media_type must be echoed back — otherwise a movie search and a tv "
"search for the same query are indistinguishable to the client's "
"keyed response matching (transport.js tmdb_search_resp handler)")
assert resp["results"] == [{
"tmdb_id": "111", "title": "Some Show", "year": "2019",
"poster_thumb_hash": resp["results"][0]["poster_thumb_hash"],
}]
async def test_search_tv_calls_tv_search(media_cache):
client = FakeTmdbClient()
session = _session(media_cache=media_cache, tmdb_client=client)
await session._do_tmdb_search_request({"query": "Some Show", "media_type": "tv"})
assert client.tv_search_calls == ["Some Show"]
assert client.movie_search_calls == []
assert session.sent[0]["media_type"] == "tv"
|