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
|
"""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 MusicBrainzClient, _MIN_INTERVAL_SECS
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()
@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()
|