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
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
|
"""Tests for musicbrainz.py against a mocked httpx transport — no live network in CI."""
import time
import httpx
import pytest
from meshbay_node.musicbrainz import _MIN_INTERVAL_SECS, MusicBrainzClient, _escape_lucene
class FakeRoster:
def __init__(self, contact: str | None = "operator@example.invalid"):
self._contact = contact
async def musicbrainz_contact(self):
return self._contact
def _handler(response_map):
def handle(request: httpx.Request) -> httpx.Response:
path = request.url.path
for prefix, body in response_map.items():
if path.endswith(prefix):
return httpx.Response(200, json=body)
return httpx.Response(404, json={})
return handle
@pytest.mark.asyncio
async def test_search_release_returns_top_result_and_confidence():
body = {"releases": [{"id": "abc-123", "title": "The Great Album",
"artist-credit": [{"name": "Some Artist"}]}]}
client = MusicBrainzClient(
roster=FakeRoster(),
transport=httpx.MockTransport(_handler({"release": body})),
)
result, ratio = await client.search_release("Some Artist", "The Great Album")
assert result is not None
assert result["id"] == "abc-123"
assert ratio > 0.9
await client.close()
@pytest.mark.asyncio
async def test_no_results_returns_none_and_zero_confidence():
client = MusicBrainzClient(
roster=FakeRoster(),
transport=httpx.MockTransport(_handler({"release": {"releases": []}})),
)
result, ratio = await client.search_release("Nobody", "Nonexistent Obscure Album")
assert result is None
assert ratio == 0.0
await client.close()
def test_escape_lucene_escapes_query_syntax_characters():
# Verified live against musicbrainz.org before writing the fallback
# below: a literal "(" inside a quoted phrase is read as query syntax,
# not a character to match, and silently drops the phrase to zero
# results rather than raising.
assert _escape_lucene("Hebron Gate (2003)") == r"Hebron Gate \(2003\)"
assert _escape_lucene('Say "It" Loud') == r"Say \"It\" Loud"
assert _escape_lucene("Rock & Roll: Live") == r"Rock \& Roll\: Live"
assert _escape_lucene("no specials here") == "no specials here"
@pytest.mark.asyncio
async def test_strict_match_does_not_pay_for_a_second_request():
"""A tag that already matches MusicBrainz's own spelling should cost
exactly one request — the fallback below exists for the case that
doesn't, not for every call."""
calls = []
def handle(request: httpx.Request) -> httpx.Response:
calls.append(str(request.url))
return httpx.Response(200, json={
"releases": [{"id": "abc-123", "title": "The Great Album",
"artist-credit": [{"name": "Some Artist"}]}],
})
client = MusicBrainzClient(
roster=FakeRoster(),
transport=httpx.MockTransport(handle),
)
await client.search_release("Some Artist", "The Great Album")
assert len(calls) == 1
await client.close()
@pytest.mark.asyncio
async def test_falls_back_to_a_loose_query_when_the_strict_one_finds_nothing():
"""
Regression for the real failure mode found against the live service:
`artist:"Groundation" AND release:"Hebron Gate"` finds the release
(score 100), but a local album tag/folder carrying a trailing year —
"Hebron Gate (2003)", a common rip-folder shape — drops the strict
exact-phrase query to zero hits, not a low-scored one. The loose,
unscoped query must be tried next instead of stopping at "no match".
"""
calls = []
def handle(request: httpx.Request) -> httpx.Response:
query = request.url.params.get("query", "")
calls.append(query)
if query.startswith("artist:"):
return httpx.Response(200, json={"releases": []})
return httpx.Response(200, json={
"releases": [{"id": "xyz-789", "title": "Hebron Gate",
"artist-credit": [{"name": "Groundation"}]}],
})
client = MusicBrainzClient(
roster=FakeRoster(),
transport=httpx.MockTransport(handle),
)
result, ratio = await client.search_release("Groundation", "Hebron Gate (2003)")
assert result is not None
assert result["id"] == "xyz-789"
assert ratio > 0.5
assert len(calls) == 2, "must retry with a loose query after the strict one finds nothing"
await client.close()
@pytest.mark.asyncio
async def test_confidence_reflects_a_wrong_artist_on_a_same_titled_release():
"""
The loose fallback query has no field scoping at all, so a same-
titled release by an unrelated artist must not read as confidently as
one that also matches on artist — otherwise the fallback trades
"never finds a mismatch" for "sometimes confirms a wrong one".
"""
def handle(request: httpx.Request) -> httpx.Response:
return httpx.Response(200, json={
"releases": [{"id": "wrong-1", "title": "Live",
"artist-credit": [{"name": "An Entirely Different Band"}]}],
})
client = MusicBrainzClient(
roster=FakeRoster(),
transport=httpx.MockTransport(handle),
)
result, ratio = await client.search_release("Groundation", "Live")
assert result is not None
assert ratio < 0.6, "a title-only match against the wrong artist must not read as confident"
await client.close()
@pytest.mark.asyncio
async def test_no_contact_configured_makes_no_request(monkeypatch):
monkeypatch.delenv("MESHBAY_MUSICBRAINZ_CONTACT_DEFAULT", raising=False)
calls = []
def handle(request: httpx.Request) -> httpx.Response:
calls.append(request)
return httpx.Response(200, json={"releases": []})
client = MusicBrainzClient(
roster=FakeRoster(contact=None),
transport=httpx.MockTransport(handle),
)
result, ratio = await client.search_release("Anyone", "Anything")
assert result is None
assert calls == [], "an unidentified client must never be sent — see musicbay.md §3.1"
await client.close()
@pytest.mark.asyncio
async def test_the_configured_contact_is_sent_as_user_agent():
captured = {}
def handle(request: httpx.Request) -> httpx.Response:
captured["ua"] = request.headers.get("user-agent")
return httpx.Response(200, json={"releases": []})
client = MusicBrainzClient(
roster=FakeRoster(contact="operator@example.invalid"),
transport=httpx.MockTransport(handle),
)
await client.search_release("Anyone", "Anything")
assert "operator@example.invalid" in captured["ua"]
await client.close()
@pytest.mark.asyncio
async def test_http_error_returns_none_gracefully():
def handle(request: httpx.Request) -> httpx.Response:
return httpx.Response(500, json={"error": "server error"})
client = MusicBrainzClient(
roster=FakeRoster(),
transport=httpx.MockTransport(handle),
)
result, ratio = await client.search_release("Anyone", "Anything")
assert result is None
assert ratio == 0.0
await client.close()
@pytest.mark.asyncio
async def test_cover_art_missing_returns_none_not_an_error():
def handle(request: httpx.Request) -> httpx.Response:
return httpx.Response(404)
client = MusicBrainzClient(
roster=FakeRoster(),
transport=httpx.MockTransport(handle),
)
content = await client.fetch_cover_art("abc-123")
assert content is None
await client.close()
@pytest.mark.asyncio
async def test_cover_art_found_returns_bytes():
def handle(request: httpx.Request) -> httpx.Response:
return httpx.Response(200, content=b"\xff\xd8fake-jpeg-bytes")
client = MusicBrainzClient(
roster=FakeRoster(),
transport=httpx.MockTransport(handle),
)
content = await client.fetch_cover_art("abc-123")
assert content == b"\xff\xd8fake-jpeg-bytes"
await client.close()
@pytest.mark.asyncio
async def test_calls_are_paced_at_least_min_interval_apart():
"""
docs/musicbay.md §3.2: the ~1 req/s courtesy limit is this node's own
job, not something the server hands out — verified by timing two calls
back to back rather than mocking the clock, so a change to the pacing
implementation that still meets the contract doesn't break this test.
"""
def handle(request: httpx.Request) -> httpx.Response:
return httpx.Response(200, json={"releases": []})
client = MusicBrainzClient(
roster=FakeRoster(),
transport=httpx.MockTransport(handle),
)
start = time.monotonic()
await client.search_release("A", "One")
await client.search_release("B", "Two")
elapsed = time.monotonic() - start
assert elapsed >= _MIN_INTERVAL_SECS * 0.9
await client.close()
|