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
|
"""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
@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
|