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
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
|
"""Tests for indexer/enrich.py — season/title corroboration and the end-to-end pool."""
import asyncio
import shutil
import subprocess
from pathlib import Path
import pytest
from meshbay_common.protocol import IndexEntry
from meshbay_node.indexer.enrich import Enricher, _season_from_ancestors, _title_from_siblings
from meshbay_node.media_cache import MediaCache
_HAVE_FFMPEG = shutil.which("ffmpeg") and shutil.which("ffprobe")
# ── pure helpers, no ffmpeg needed ───────────────────────────────────────────
def test_season_from_ancestors_finds_season_folder(tmp_path):
folder = tmp_path / "Some Show" / "Season 2"
folder.mkdir(parents=True)
ep = folder / "01 - Episode Title.mkv"
ep.touch()
assert _season_from_ancestors(ep) == 2
def test_season_from_ancestors_none_when_no_season_folder(tmp_path):
folder = tmp_path / "Movies"
folder.mkdir()
f = folder / "Some Movie 2015.mkv"
f.touch()
assert _season_from_ancestors(f) is None
def test_title_from_siblings_borrows_from_a_titled_sibling(tmp_path):
folder = tmp_path / "Acronym Show"
folder.mkdir()
titled = folder / "Some.Show.Name.S01E01.720p.mkv"
untitled = folder / "S01E02.SUBFRENCH.720p.mkv"
titled.touch()
untitled.touch()
assert _title_from_siblings(untitled) == "Some Show Name"
def test_title_from_siblings_none_when_no_titled_sibling(tmp_path):
folder = tmp_path / "Acronym Show"
folder.mkdir()
(folder / "S01E02.mkv").touch()
assert _title_from_siblings(folder / "S01E02.mkv") is None
# ── end-to-end against a real (tiny, synthetic) video file ──────────────────
pytestmark_ffmpeg = pytest.mark.skipif(not _HAVE_FFMPEG, reason="ffmpeg/ffprobe not installed")
def _make_clip(path: Path) -> None:
subprocess.run(
["ffmpeg", "-hide_banner", "-loglevel", "error", "-y",
"-f", "lavfi", "-i", "testsrc=size=320x240:rate=25:duration=1",
"-c:v", "libx264", "-preset", "ultrafast", "-an", str(path)],
check=True, capture_output=True,
)
@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()
@pytestmark_ffmpeg
@pytest.mark.asyncio
async def test_enricher_populates_fields_and_stores_thumbnail(tmp_path, media_cache):
clip = tmp_path / "Some.Movie.2015.1080p.mkv"
_make_clip(clip)
entry = IndexEntry(id="fileid1", name=clip.name, path=clip.name,
size=clip.stat().st_size, type="video", added_at=0)
enricher = Enricher(media_cache)
done = asyncio.get_event_loop().create_future()
async def on_done(file_id, fields):
done.set_result((file_id, fields))
enricher.spawn(entry, clip, on_done)
file_id, fields = await asyncio.wait_for(done, timeout=30)
assert file_id == "fileid1"
assert fields["width"] == 320
assert fields["height"] == 240
assert fields["display_title"] == "Some Movie"
assert fields.get("thumb_hash")
stored = await media_cache.get_thumb(fields["thumb_hash"])
assert stored is not None and len(stored) > 0
async def _run(enricher, entry, path):
done = asyncio.get_event_loop().create_future()
async def on_done(file_id, fields):
done.set_result((file_id, fields))
enricher.spawn(entry, path, on_done)
return await asyncio.wait_for(done, timeout=30)
@pytestmark_ffmpeg
@pytest.mark.asyncio
async def test_enricher_reuses_cached_probe_and_thumbnail_without_touching_the_file_again(
tmp_path, media_cache):
"""
The gap this closes: duration/width/height/thumb_hash only ever lived
on the in-memory GroupIndex entry, so every daemon restart re-ran
ffprobe and ffmpeg over every video in every group from scratch, even
though media_cache.db already had the answer. Proven strongly: the
source file is deleted between the two runs, so a real second probe or
frame grab would fail outright rather than merely being redundant.
"""
clip = tmp_path / "Some.Movie.2015.1080p.mkv"
_make_clip(clip)
entry = IndexEntry(id="fileid_reuse", name=clip.name, path=clip.name,
size=clip.stat().st_size, type="video", added_at=0)
first_enricher = Enricher(media_cache)
_, first_fields = await _run(first_enricher, entry, clip)
assert first_fields.get("thumb_hash")
clip.unlink()
second_enricher = Enricher(media_cache)
_, second_fields = await _run(second_enricher, entry, clip)
assert second_fields == first_fields
@pytestmark_ffmpeg
@pytest.mark.asyncio
async def test_a_renamed_file_re_parses_its_title_even_with_a_cached_probe(tmp_path, media_cache):
"""
A cached probe/thumbnail must never leak into a stale title parse.
display_title/season/episode come from guessit against the *filename*,
which is exactly what a rename needs re-derived
(test_rename_reenrichment.py covers the scheduling half of this) — the
new cache-hit path added alongside them must not accidentally reuse a
stale parse just because it took the same shortcut for duration/width/
height. Proven the same strong way: the renamed file never exists on
disk at all, so a correct implementation still succeeds (guessit only
reads entry.name) while a regression that tried to re-probe or
re-thumbnail the "new" path would fail outright.
"""
clip = tmp_path / "Old.Name.2015.mkv"
_make_clip(clip)
entry = IndexEntry(id="fileid_rename", name=clip.name, path=clip.name,
size=clip.stat().st_size, type="video", added_at=0)
enricher = Enricher(media_cache)
_, first_fields = await _run(enricher, entry, clip)
assert first_fields["display_title"] == "Old Name"
renamed_path = tmp_path / "New.Name.2020.mkv" # never created — proves nothing re-reads it
renamed_entry = IndexEntry(id="fileid_rename", name=renamed_path.name,
path=renamed_path.name, size=entry.size, type="video", added_at=0)
_, second_fields = await _run(enricher, renamed_entry, renamed_path)
assert second_fields["display_title"] == "New Name"
assert second_fields["width"] == first_fields["width"]
assert second_fields["height"] == first_fields["height"]
assert second_fields["thumb_hash"] == first_fields["thumb_hash"]
@pytestmark_ffmpeg
@pytest.mark.asyncio
async def test_enricher_handles_episode_with_season_from_folder(tmp_path, media_cache):
# Filename carries only a bare episode number, no SxxExx token — guessit
# confirmed (separately) not to find a season here at all — so the
# season must come from the ancestor folder (§3.4's non-standard case).
folder = tmp_path / "Some Show" / "Saison 3"
folder.mkdir(parents=True)
titled_sibling = folder / "Some.Show.Episode.06.mkv"
titled_sibling.touch()
clip = folder / "Episode.07.720p.mkv"
_make_clip(clip)
entry = IndexEntry(id="fileid2", name=clip.name, path=str(clip.relative_to(tmp_path)),
size=clip.stat().st_size, type="video", added_at=0)
enricher = Enricher(media_cache)
done = asyncio.get_event_loop().create_future()
async def on_done(file_id, fields):
done.set_result((file_id, fields))
enricher.spawn(entry, clip, on_done)
file_id, fields = await asyncio.wait_for(done, timeout=30)
assert fields["display_title"] == "Some Show"
assert fields["season"] == 3
assert fields["episode"] == 7
|