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
|
"""Tests for media_cache.py — TMDB/MusicBrainz/thumbnail cache and its pruning obligation."""
import time
import pytest
from meshbay_node.media_cache import MUSICBRAINZ_META_TTL_SECS, TMDB_META_TTL_SECS, MediaCache
@pytest.fixture
async def cache(tmp_path):
c = MediaCache(db_path=tmp_path / "media_cache.db")
await c.open()
yield c
await c.close()
@pytest.mark.asyncio
async def test_file_tmdb_round_trip(cache):
assert await cache.get_file_tmdb("file1") is None
await cache.set_file_tmdb("file1", "12345", "movie")
assert await cache.get_file_tmdb("file1") == ("12345", "movie")
@pytest.mark.asyncio
async def test_tmdb_meta_round_trip(cache):
assert await cache.get_tmdb_meta("12345", "movie") is None
await cache.set_tmdb_meta("12345", "movie", {"title": "A Movie", "vote_average": 7.5})
meta = await cache.get_tmdb_meta("12345", "movie")
assert meta == {"title": "A Movie", "vote_average": 7.5}
@pytest.mark.asyncio
async def test_tmdb_meta_expires_after_ttl(cache):
await cache._db.execute(
"INSERT INTO tmdb_meta (tmdb_id, media_type, json, fetched_at) VALUES (?, ?, ?, ?)",
("999", "movie", '{"title": "Old"}', time.time() - TMDB_META_TTL_SECS - 1),
)
await cache._db.commit()
assert await cache.get_tmdb_meta("999", "movie") is None
@pytest.mark.asyncio
async def test_thumb_round_trip(cache):
assert await cache.get_thumb("thumbhash1") is None
await cache.put_thumb("thumbhash1", "file1", b"\xff\xd8fakejpeg")
assert await cache.get_thumb("thumbhash1") == b"\xff\xd8fakejpeg"
@pytest.mark.asyncio
async def test_prune_file_removes_thumb_and_mapping_but_not_shared_meta(cache):
# Two episodes of the same show share one tmdb_meta row (§2's stated case).
await cache.set_file_tmdb("ep1", "555", "tv")
await cache.set_file_tmdb("ep2", "555", "tv")
await cache.set_tmdb_meta("555", "tv", {"name": "A Show"})
await cache.put_thumb("thumb-ep1", "ep1", b"jpeg-bytes-1")
await cache.prune_file("ep1")
assert await cache.get_file_tmdb("ep1") is None
assert await cache.get_thumb("thumb-ep1") is None
# ep2's own mapping and the shared show metadata both survive
assert await cache.get_file_tmdb("ep2") == ("555", "tv")
assert await cache.get_tmdb_meta("555", "tv") == {"name": "A Show"}
# ── auto-resolved matches vs manual "Fix match" corrections ─────────────────
@pytest.mark.asyncio
async def test_clear_tmdb_matches_drops_auto_matches_but_keeps_overrides(cache):
await cache.set_file_tmdb("auto1", "10", "movie")
await cache.set_file_tmdb("auto2", "20", "movie")
await cache.set_file_tmdb("fixed", "30", "movie")
await cache.mark_tmdb_override("fixed")
removed = await cache.clear_tmdb_matches(["auto1", "auto2", "fixed", "never-seen"])
assert removed == 2
assert await cache.get_file_tmdb("auto1") is None
assert await cache.get_file_tmdb("auto2") is None
assert await cache.get_file_tmdb("fixed") == ("30", "movie"), (
"a manual Fix match correction must survive ops.rematch_video")
@pytest.mark.asyncio
async def test_clear_tmdb_matches_empty_list_is_a_noop(cache):
assert await cache.clear_tmdb_matches([]) == 0
@pytest.mark.asyncio
async def test_clear_file_tmdb_drops_one_auto_match_but_keeps_an_override(cache):
await cache.set_file_tmdb("renamed", "10", "movie")
await cache.clear_file_tmdb("renamed")
assert await cache.get_file_tmdb("renamed") is None
await cache.set_file_tmdb("renamed-fixed", "11", "movie")
await cache.mark_tmdb_override("renamed-fixed")
await cache.clear_file_tmdb("renamed-fixed")
assert await cache.get_file_tmdb("renamed-fixed") == ("11", "movie")
@pytest.mark.asyncio
async def test_drop_tmdb_match_forgets_both_the_match_and_the_override(cache):
await cache.set_file_tmdb("f", "10", "movie")
await cache.mark_tmdb_override("f")
await cache.drop_tmdb_match("f")
assert await cache.get_file_tmdb("f") is None
# override marker is gone: a fresh match is now treated as ordinary
await cache.set_file_tmdb("f", "20", "movie")
assert await cache.clear_tmdb_matches(["f"]) == 1
@pytest.mark.asyncio
async def test_prune_file_also_clears_the_override_marker(cache):
await cache.set_file_tmdb("gone", "10", "movie")
await cache.mark_tmdb_override("gone")
await cache.prune_file("gone")
# the marker must not linger after the file leaves the index and then
# shield a later match on the same id from ops.rematch_video
assert await cache.get_file_tmdb("gone") is None
await cache.set_file_tmdb("gone", "99", "movie")
assert await cache.clear_tmdb_matches(["gone"]) == 1
# ── Music app (docs/MESHBAY_DESIGN.md §9.8) — file_mbid/mbid_meta ────────────
@pytest.mark.asyncio
async def test_file_mbid_round_trip(cache):
assert await cache.get_file_mbid("file1") is None
await cache.set_file_mbid("file1", "release-mbid-1")
assert await cache.get_file_mbid("file1") == "release-mbid-1"
@pytest.mark.asyncio
async def test_mbid_meta_round_trip(cache):
assert await cache.get_mbid_meta("release-mbid-1") is None
await cache.set_mbid_meta("release-mbid-1", {"artist": "Some Artist", "album": "An Album"})
meta = await cache.get_mbid_meta("release-mbid-1")
assert meta == {"artist": "Some Artist", "album": "An Album"}
@pytest.mark.asyncio
async def test_mbid_meta_expires_after_ttl(cache):
await cache._db.execute(
"INSERT INTO mbid_meta (mbid, json, fetched_at) VALUES (?, ?, ?)",
("old-mbid", '{"album": "Old"}', time.time() - MUSICBRAINZ_META_TTL_SECS - 1),
)
await cache._db.commit()
assert await cache.get_mbid_meta("old-mbid") is None
@pytest.mark.asyncio
async def test_prune_file_removes_mbid_mapping_but_not_shared_meta(cache):
# Two tracks of the same release share one mbid_meta row.
await cache.set_file_mbid("track1", "release-mbid-1")
await cache.set_file_mbid("track2", "release-mbid-1")
await cache.set_mbid_meta("release-mbid-1", {"album": "An Album"})
await cache.put_thumb("cover-hash", "track1", b"cover-bytes")
await cache.prune_file("track1")
assert await cache.get_file_mbid("track1") is None
assert await cache.get_thumb("cover-hash") is None
# track2's own mapping and the shared release metadata both survive
assert await cache.get_file_mbid("track2") == "release-mbid-1"
assert await cache.get_mbid_meta("release-mbid-1") == {"album": "An Album"}
|