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
|
"""
`_do_music_meta_request`, keyed by `file_id` (2026-08-25 fix).
Regression found live: `IndexEntry.path` is the *folder* a track is in
(indexer.py's `_virtual_dir`), not the track itself — an album is one
folder with many tracks in it, so looking a track up by `.path` alone
(the pre-fix behaviour, `GroupIndex.get_entry_by_path`) silently resolved
every track in that folder to whichever entry the index happened to
return first. Three unrelated albums showed the same wrong MusicBrainz
cover in production before this was found and fixed.
"""
import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from meshbay_common.protocol import MNP, IndexEntry
from meshbay_node.indexer.group_index import GroupIndex
from meshbay_node.media_cache import MediaCache
from meshbay_node.transport.webrtc_server import WebRTCPeerSession
pytestmark = pytest.mark.asyncio
class FakeMusicBrainzClient:
"""Returns a distinct, deterministic match per (artist, album) pair —
real enough to prove the server routed to the *right* track's own tags,
not a stand-in for musicbrainz.py's own search-quality tests."""
def __init__(self):
self.calls = []
async def search_release(self, artist, album):
self.calls.append((artist, album))
return (
{"id": f"mbid-for-{artist}-{album}", "title": album,
"artist-credit": [{"name": artist}]},
1.0,
)
async def fetch_cover_art(self, mbid):
return f"cover-bytes-for-{mbid}".encode()
def _entry(path: str, name: str, file_id: str, artist: str, album: str) -> IndexEntry:
return IndexEntry(
id=file_id, name=name, path=path, size=1, type="audio", added_at=0,
artist=artist, album=album, display_title=name,
)
@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()
def _session(index, media_cache, musicbrainz_client):
session = WebRTCPeerSession.__new__(WebRTCPeerSession)
session._ctx = {
"index": index,
"media_cache": media_cache,
"musicbrainz_client": musicbrainz_client,
"musicbrainz_enabled": True,
}
session._group_id = None
session.sent = []
session._send = session.sent.append
return session
async def test_two_tracks_in_the_same_folder_each_get_their_own_metadata(media_cache):
"""The exact production scenario: two tracks share a folder (an album),
with different artist/album tags of their own (one mistagged, sitting
in the wrong physical folder — a real, if messy, real-world case). Each
must resolve against its *own* tags, not whichever track the index
happens to return first for that shared folder path."""
index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate())
track_a = _entry("music/high_tone", "a.mp3", "id-a", "High Tone", "Future Dub (1)")
track_b = _entry("music/high_tone", "b.mp3", "id-b", "Le Peuple de l'Herbe", "Triple Zero")
index.add_entry(track_a)
index.add_entry(track_b)
client = FakeMusicBrainzClient()
session = _session(index, media_cache, client)
await session._do_music_meta_request({"file_id": "id-a"})
await session._do_music_meta_request({"file_id": "id-b"})
resp_a, resp_b = session.sent
assert resp_a["file_id"] == "id-a"
assert resp_a["artist"] == "High Tone"
assert resp_a["album"] == "Future Dub (1)"
assert resp_b["file_id"] == "id-b"
assert resp_b["artist"] == "Le Peuple de l'Herbe"
assert resp_b["album"] == "Triple Zero"
assert resp_a["cover_thumb_hash"] != resp_b["cover_thumb_hash"], (
"two different tracks sharing a folder must not end up with the same cover")
assert set(client.calls) == {("High Tone", "Future Dub (1)"),
("Le Peuple de l'Herbe", "Triple Zero")}
async def test_missing_file_id_is_refused(media_cache):
index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate())
session = _session(index, media_cache, FakeMusicBrainzClient())
await session._do_music_meta_request({})
assert session.sent == [{"type": "error", "detail": "Missing file_id"}]
async def test_unknown_file_id_is_refused(media_cache):
index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate())
session = _session(index, media_cache, FakeMusicBrainzClient())
await session._do_music_meta_request({"file_id": "nope"})
assert session.sent == [{"type": "error", "detail": "File not found"}]
async def test_no_confidence_below_threshold_still_answers_by_file_id(media_cache):
index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate())
index.add_entry(_entry("music/x", "a.mp3", "id-a", "Some Artist", "Some Album"))
class LowConfidenceClient(FakeMusicBrainzClient):
async def search_release(self, artist, album):
return {"id": "mbid", "title": album,
"artist-credit": [{"name": artist}]}, 0.1
session = _session(index, media_cache, LowConfidenceClient())
await session._do_music_meta_request({"file_id": "id-a"})
assert session.sent == [{
"type": MNP.MUSIC_META_RESP, "v": session.sent[0]["v"],
"file_id": "id-a", "confidence": 0,
}]
|