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
|
"""
Tests for ops.rematch_video — drops a group's *auto-resolved* video->TMDB
matches so they re-resolve against the current matcher (run by the operator
after a matcher/parser fix; `media_cache.file_tmdb` is keyed by content
hash and is otherwise only pruned on deletion). Manual "Fix match"
corrections are kept.
"""
import types
import pytest
from meshbay_node import ops
from meshbay_node.media_cache import MediaCache
pytestmark = pytest.mark.asyncio
def _indexer(*entries):
return types.SimpleNamespace(index=types.SimpleNamespace(entries=list(entries)))
def _entry(file_id, type_="video"):
return types.SimpleNamespace(id=file_id, type=type_)
@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()
async def test_no_media_cache_raises():
with pytest.raises(ops.OpError):
await ops.rematch_video({}, "g" * 32)
async def test_unknown_group_raises(media_cache):
with pytest.raises(ops.OpError):
await ops.rematch_video({"media_cache": media_cache, "indexers": {}}, "nope")
async def test_clears_auto_matches_keeps_overrides_ignores_non_video(media_cache):
await media_cache.set_file_tmdb("v-auto-1", "10", "movie")
await media_cache.set_file_tmdb("v-auto-2", "20", "movie")
await media_cache.set_file_tmdb("v-fixed", "30", "movie")
await media_cache.mark_tmdb_override("v-fixed")
await media_cache.set_file_mbid("a-track", "some-mbid") # audio, untouched
state = {
"media_cache": media_cache,
"indexers": {"g": _indexer(
_entry("v-auto-1"), _entry("v-auto-2"), _entry("v-fixed"),
_entry("a-track", "audio"),
)},
}
result = await ops.rematch_video(state, "g")
assert result == {"status": "cleared", "removed": 2, "videos": 3, "group_id": "g"}
assert await media_cache.get_file_tmdb("v-auto-1") is None
assert await media_cache.get_file_tmdb("v-auto-2") is None
assert await media_cache.get_file_tmdb("v-fixed") == ("30", "movie")
assert await media_cache.get_file_mbid("a-track") == "some-mbid"
async def test_group_with_no_videos_is_a_clean_noop(media_cache):
state = {"media_cache": media_cache, "indexers": {"g": _indexer()}}
result = await ops.rematch_video(state, "g")
assert result == {"status": "cleared", "removed": 0, "videos": 0, "group_id": "g"}
|