/
/
/
1"""Tests for the guess-the-song quiz type distractor sourcing."""
2
3from __future__ import annotations
4
5import asyncio
6import json
7from types import SimpleNamespace
8from unittest.mock import AsyncMock, MagicMock, patch
9
10import pytest
11from music_assistant_models.enums import MediaType
12from music_assistant_models.errors import InvalidDataError
13from music_assistant_models.media_items import ItemMapping, ProviderMapping, Track
14from music_assistant_models.unique_list import UniqueList
15
16from music_assistant.controllers.music.recency import RecencySnapshot, RecencyWindows
17from music_assistant.models.plugin import AIEngine, PluginProvider
18from music_assistant.providers.music_quiz.models import (
19 MultipleChoiceRoundState,
20 MusicQuizConfig,
21)
22from music_assistant.providers.music_quiz.quiz_types.base import QUIZ_TRACK_RECENCY_SECONDS
23from music_assistant.providers.music_quiz.quiz_types.guess_the_song import (
24 GuessTheSongQuizType,
25 _track_to_candidate,
26)
27from music_assistant.providers.music_quiz.suggestions import SuggestionCandidate
28
29CORRECT_LABEL = "Daft Punk - Around the World"
30
31
32def test_config_normalization_preserves_similar_music() -> None:
33 """Keep the shared source-expansion setting for Guess the Song."""
34 config = MusicQuizConfig(
35 source_uris=["prov://playlist/1"],
36 include_similar_music=True,
37 )
38
39 normalized = GuessTheSongQuizType.normalize_config(config)
40
41 assert normalized.include_similar_music is True
42
43
44def _track(item_id: str, name: str, artist: str | None, provider: str = "prov") -> Track:
45 """Return a minimal Track with a single artist mapping."""
46 return Track(
47 item_id=item_id,
48 provider=provider,
49 name=name,
50 artists=UniqueList(
51 [
52 ItemMapping(
53 media_type=MediaType.ARTIST,
54 item_id=f"a_{item_id}",
55 provider=provider,
56 name=artist,
57 )
58 ]
59 if artist
60 else []
61 ),
62 provider_mappings={
63 ProviderMapping(item_id=item_id, provider_domain=provider, provider_instance=provider)
64 },
65 )
66
67
68def _artist(item_id: str, name: str, provider: str = "prov") -> ItemMapping:
69 """Return a minimal artist mapping."""
70 return ItemMapping(media_type=MediaType.ARTIST, item_id=item_id, provider=provider, name=name)
71
72
73def _pool(tracks: list[Track]) -> dict[str, Track]:
74 """Return a source-track pool keyed by URI."""
75 pool: dict[str, Track] = {}
76 for track in tracks:
77 assert track.uri is not None
78 pool[track.uri] = track
79 return pool
80
81
82def _correct() -> tuple[Track, SuggestionCandidate]:
83 """Return the correct source track and its answer candidate."""
84 track = _track("c1", "Around the World", "Daft Punk")
85 return track, _track_to_candidate(track)
86
87
88def _quiz_type(
89 difficulty: str = "normal",
90 use_ai: bool = False,
91 suggestion_count: int = 4,
92) -> tuple[GuessTheSongQuizType, MagicMock]:
93 """Return a quiz type with a mock MusicAssistant and empty music lookups."""
94 mass = MagicMock()
95 mass.music.get_item = AsyncMock()
96 mass.music.search = AsyncMock(return_value=SimpleNamespace(tracks=[]))
97 mass.music.tracks.get_provider_item = AsyncMock()
98 mass.music.tracks.similar_tracks = AsyncMock(return_value=[])
99 mass.music.artists.similar_artists = AsyncMock(return_value=[])
100 mass.music.artists.top_tracks = AsyncMock(return_value=[])
101 mass.music.recency.snapshot = AsyncMock(return_value=RecencySnapshot(now=0))
102 mass.metadata.get_image_url_for_item = AsyncMock(return_value=None)
103 mass.get_providers_supporting_feature = MagicMock(return_value=[])
104 config = MusicQuizConfig(
105 suggestion_count=suggestion_count,
106 source_uris=["prov://playlist/1"],
107 difficulty=difficulty,
108 use_ai_distractors=use_ai,
109 ai_engine="ai--1/engine" if use_ai else None,
110 )
111 quiz_type = GuessTheSongQuizType(mass, config)
112 quiz_type._source_track_pool = {}
113 return quiz_type, mass
114
115
116def _ai_provider(response: object = None, error: Exception | None = None) -> MagicMock:
117 """Return a mock AI_QUERY-capable plugin provider exposing a single engine."""
118 provider = MagicMock(spec=PluginProvider)
119 provider.instance_id = "ai--1"
120 provider.ai_query = AsyncMock(return_value=response, side_effect=error)
121 provider.get_ai_engines = AsyncMock(
122 return_value=[AIEngine(id="engine", name="ai--1", provider=provider)]
123 )
124 return provider
125
126
127def _ai_response(
128 ranked_ids: list[str],
129 synthetic: list[tuple[str, str]],
130 **extra: object,
131) -> str:
132 """Return a structured AI distractor response."""
133 return json.dumps(
134 {
135 "ranked_ids": ranked_ids,
136 "synthetic": [{"kind": kind, "label": label} for kind, label in synthetic],
137 **extra,
138 }
139 )
140
141
142def test_reject_track_removes_it_from_the_source_pool() -> None:
143 """Exclude failed playback tracks from later Guess rounds."""
144 quiz_type, _ = _quiz_type()
145 failed = _track("failed", "Unavailable", "Artist")
146 available = _track("available", "Playable", "Artist")
147 quiz_type._source_track_pool = _pool([failed, available])
148 assert failed.uri is not None
149
150 quiz_type.reject_track(failed.uri)
151
152 assert quiz_type._source_track_pool == _pool([available])
153
154
155@pytest.mark.asyncio
156async def test_initialize_reads_partial_play_recency() -> None:
157 """Load interrupted playback history for quiz track selection."""
158 quiz_type, mass = _quiz_type()
159 snapshot = RecencySnapshot(now=100)
160 mass.music.recency.snapshot.return_value = snapshot
161
162 await quiz_type.initialize()
163
164 assert quiz_type._recency_snapshot is snapshot
165 mass.music.recency.snapshot.assert_awaited_once_with(
166 RecencyWindows(song_seconds=QUIZ_TRACK_RECENCY_SECONDS),
167 include_partially_played=True,
168 )
169
170
171@pytest.mark.asyncio
172async def test_source_selection_prefers_tracks_not_played_recently() -> None:
173 """Choose a fresh source track while retaining recent tracks as fallback candidates."""
174 quiz_type, _ = _quiz_type()
175 recent = _track("recent", "Teardrop", "Massive Attack")
176 fresh = _track("fresh", "Genesis", "Justice")
177 quiz_type._source_track_pool = _pool([recent, fresh])
178 quiz_type._recency_snapshot = RecencySnapshot(
179 now=100,
180 song_ts={(recent.provider, recent.item_id): 100},
181 )
182
183 with patch(
184 "music_assistant.providers.music_quiz.quiz_types.guess_the_song.secrets.choice",
185 side_effect=lambda candidates: candidates[0],
186 ) as choose:
187 selected = await quiz_type._get_next_source_track(set())
188 assert quiz_type._recency_snapshot is not None
189 quiz_type._recency_snapshot.song_ts[(fresh.provider, fresh.item_id)] = 100
190 fallback = await quiz_type._get_next_source_track(set())
191
192 assert selected is fresh
193 assert fallback is recent
194 assert choose.call_args_list[0].args[0] == [fresh]
195 assert choose.call_args_list[1].args[0] == [recent, fresh]
196
197
198@pytest.mark.asyncio
199async def test_normal_difficulty_prefers_search_before_source_pool() -> None:
200 """Normal difficulty keeps catalog results ahead of the source-pool fallback."""
201 quiz_type, mass = _quiz_type("normal", use_ai=True)
202 mass.music.search.return_value = SimpleNamespace(
203 tracks=[_track("s1", "One More Time", "Daft Punk"), _track("s2", "Genesis", "Justice")]
204 )
205 correct_track, correct = _correct()
206 quiz_type._source_track_pool = _pool([correct_track, _track("pool", "Lisztomania", "Phoenix")])
207
208 result = list(await quiz_type._gather_distractors(correct_track, correct))
209
210 assert [item.label for item in result] == [
211 "Daft Punk - One More Time",
212 "Justice - Genesis",
213 "Phoenix - Lisztomania",
214 ]
215 mass.music.tracks.similar_tracks.assert_not_awaited()
216 mass.get_providers_supporting_feature.assert_not_called()
217
218
219@pytest.mark.asyncio
220async def test_hard_difficulty_prefers_similar_tracks() -> None:
221 """Hard difficulty offers similar tracks first, with the search kept as a fallback tail."""
222 quiz_type, mass = _quiz_type("hard")
223 mass.music.tracks.similar_tracks.return_value = [
224 _track("st1", "Digital Love", "Daft Punk"),
225 _track("st2", "D.A.N.C.E.", "Justice"),
226 _track("st3", "Sexy Boy", "Air"),
227 ]
228 mass.music.search.return_value = SimpleNamespace(tracks=[_track("s1", "Fallback", "Someone")])
229 correct_track, correct = _correct()
230 pool_fallback = _track("pool", "Lisztomania", "Phoenix")
231 quiz_type._source_track_pool = _pool([correct_track, pool_fallback])
232
233 result = list(await quiz_type._gather_distractors(correct_track, correct))
234
235 labels = [item.label for item in result]
236 assert labels[:3] == [
237 "Daft Punk - Digital Love",
238 "Justice - D.A.N.C.E.",
239 "Air - Sexy Boy",
240 ]
241 assert labels[-2:] == ["Someone - Fallback", "Phoenix - Lisztomania"]
242 mass.music.tracks.similar_tracks.assert_awaited_once()
243
244
245@pytest.mark.asyncio
246async def test_hard_difficulty_enriches_with_similar_artists_when_tracks_sparse() -> None:
247 """When similar tracks are sparse, top tracks of similar artists are added."""
248 quiz_type, mass = _quiz_type("hard")
249 mass.music.artists.similar_artists.return_value = [
250 _artist("a2", "Justice"),
251 _artist("a3", "Air"),
252 ]
253
254 async def _top_tracks(item_id: str, **_kwargs: str) -> list[Track]:
255 return {
256 "a2": [_track("j1", "Genesis", "Justice")],
257 "a3": [_track("air1", "Sexy Boy", "Air")],
258 }.get(item_id, [])
259
260 mass.music.artists.top_tracks = AsyncMock(side_effect=_top_tracks)
261 correct_track, correct = _correct()
262
263 result = list(await quiz_type._gather_distractors(correct_track, correct))
264
265 labels = {item.label for item in result}
266 assert "Justice - Genesis" in labels
267 assert "Air - Sexy Boy" in labels
268 mass.music.artists.similar_artists.assert_awaited_once()
269
270
271@pytest.mark.asyncio
272async def test_hard_filters_unusable_similar_tracks_before_artist_enrichment() -> None:
273 """Enrich through similar artists when raw track results are only answer variants."""
274 quiz_type, mass = _quiz_type("hard", use_ai=True)
275 correct_track, correct = _correct()
276 mass.music.tracks.similar_tracks.return_value = [
277 correct_track,
278 _track("remix", "Around the World (Remix)", "Daft Punk"),
279 _track("radio", "Around the World [Radio Edit]", "Daft Punk"),
280 ]
281 mass.music.artists.similar_artists.return_value = [
282 _artist("justice", "Justice"),
283 _artist("air", "Air"),
284 _artist("phoenix", "Phoenix"),
285 ]
286 top_tracks = {
287 "justice": [_track("j1", "Genesis", "Justice")],
288 "air": [_track("a1", "Sexy Boy", "Air")],
289 "phoenix": [_track("p1", "Lisztomania", "Phoenix")],
290 }
291 mass.music.artists.top_tracks.side_effect = lambda item_id, **_kwargs: top_tracks[item_id]
292 provider = _ai_provider(
293 _ai_response(
294 ["candidate_0", "candidate_1", "candidate_2"],
295 [
296 ("same_artist_title", "Daft Punk - Neon Horizon"),
297 ("context_track", "Lunar Circuit - Chrome Reverie"),
298 ],
299 )
300 )
301 mass.get_providers_supporting_feature.return_value = [provider]
302
303 result = list(await quiz_type._gather_distractors(correct_track, correct))
304
305 assert [item.label for item in result[:3]] == [
306 "Justice - Genesis",
307 "Daft Punk - Neon Horizon",
308 "Lunar Circuit - Chrome Reverie",
309 ]
310 mass.music.artists.similar_artists.assert_awaited_once()
311 provider.ai_query.assert_awaited_once()
312
313
314@pytest.mark.asyncio
315async def test_hard_difficulty_falls_back_to_search_on_error() -> None:
316 """A failing similar-tracks/artists lookup falls through to the search distractors."""
317 quiz_type, mass = _quiz_type("hard")
318 mass.music.tracks.similar_tracks.side_effect = Exception("boom")
319 mass.music.artists.similar_artists.side_effect = Exception("boom")
320 mass.music.search.return_value = SimpleNamespace(
321 tracks=[
322 _track("s1", "One More Time", "Daft Punk"),
323 _track("s2", "Genesis", "Justice"),
324 _track("s3", "Sexy Boy", "Air"),
325 ]
326 )
327 correct_track, correct = _correct()
328
329 result = list(await quiz_type._gather_distractors(correct_track, correct))
330
331 assert "Daft Punk - One More Time" in {item.label for item in result}
332
333
334@pytest.mark.asyncio
335async def test_easy_difficulty_uses_source_pool() -> None:
336 """Easy difficulty traverses the source pool once before search fallback."""
337 quiz_type, mass = _quiz_type("easy", use_ai=True)
338 correct_track, correct = _correct()
339 pool_tracks = [
340 _track("p2", "Genesis", "Justice"),
341 _track("p3", "Sexy Boy", "Air"),
342 _track("p4", "1901", "Phoenix"),
343 ]
344 quiz_type._source_track_pool = _pool([correct_track, *pool_tracks])
345 mass.music.search.return_value = SimpleNamespace(
346 tracks=[_track("search", "Fallback", "Someone")]
347 )
348
349 with (
350 patch(
351 "music_assistant.providers.music_quiz.quiz_types.guess_the_song.SYSTEM_RANDOM.shuffle"
352 ) as shuffle_pool,
353 patch(
354 "music_assistant.providers.music_quiz.quiz_types.guess_the_song._track_to_candidate",
355 wraps=_track_to_candidate,
356 ) as convert_candidate,
357 ):
358 result = list(await quiz_type._gather_distractors(correct_track, correct))
359
360 assert [item.label for item in result] == [
361 "Justice - Genesis",
362 "Air - Sexy Boy",
363 "Phoenix - 1901",
364 "Someone - Fallback",
365 ]
366 assert correct_track.uri not in {item.uri for item in result}
367 shuffle_pool.assert_called_once()
368 assert convert_candidate.call_count == 4
369 mass.music.search.assert_awaited_once()
370 mass.music.tracks.similar_tracks.assert_not_awaited()
371 mass.get_providers_supporting_feature.assert_not_called()
372
373
374@pytest.mark.asyncio
375async def test_hard_ai_distractors_mix_real_and_synthetic_context() -> None:
376 """Build the default hard mix from one real and two contextual synthetic choices."""
377 quiz_type, mass = _quiz_type("hard", use_ai=True)
378 mass.music.tracks.similar_tracks.return_value = [
379 _track("st1", "Digital Love", "Daft Punk"),
380 _track("st2", "D.A.N.C.E.", "Justice"),
381 _track("st3", "Sexy Boy", "Air"),
382 ]
383 provider = _ai_provider(
384 _ai_response(
385 ["candidate_2", "candidate_0", "candidate_1"],
386 [
387 ("same_artist_title", "Daft Punk - Neon Horizon"),
388 ("context_track", "Lunar Circuit - Chrome Reverie"),
389 ],
390 )
391 )
392 mass.get_providers_supporting_feature.return_value = [provider]
393 correct_track, correct = _correct()
394
395 result = list(await quiz_type._gather_distractors(correct_track, correct))
396
397 assert [(item.label, item.uri) for item in result[:3]] == [
398 ("Air - Sexy Boy", mass.music.tracks.similar_tracks.return_value[2].uri),
399 ("Daft Punk - Neon Horizon", None),
400 ("Lunar Circuit - Chrome Reverie", None),
401 ]
402 assert [item.artist_names for item in result[1:3]] == [
403 ("Daft Punk",),
404 ("Lunar Circuit",),
405 ]
406 provider.ai_query.assert_awaited_once()
407 prompt = provider.ai_query.await_args.args[0]
408 assert CORRECT_LABEL in prompt
409 assert "Daft Punk - Digital Love" in prompt
410 assert "Justice - D.A.N.C.E." in prompt
411
412
413@pytest.mark.asyncio
414async def test_hard_ai_composition_scales_with_real_catalog_dominance() -> None:
415 """Retain two bounded synthetic roles while larger option sets add real tracks."""
416 quiz_type, mass = _quiz_type("hard", use_ai=True, suggestion_count=6)
417 similar_tracks = [
418 _track("st0", "Genesis", "Justice"),
419 _track("st1", "Sexy Boy", "Air"),
420 _track("st2", "Lisztomania", "Phoenix"),
421 _track("st3", "Teardrop", "Massive Attack"),
422 _track("st4", "Midnight City", "M83"),
423 ]
424 mass.music.tracks.similar_tracks.return_value = similar_tracks
425 provider = _ai_provider(
426 _ai_response(
427 [f"candidate_{index}" for index in reversed(range(5))],
428 [
429 ("same_artist_title", "Daft Punk - Neon Horizon"),
430 ("context_track", "Lunar Circuit - Chrome Reverie"),
431 ],
432 )
433 )
434 mass.get_providers_supporting_feature.return_value = [provider]
435 correct_track, correct = _correct()
436
437 result = list(await quiz_type._gather_distractors(correct_track, correct))
438
439 selected = result[:5]
440 assert sum(item.uri is not None for item in selected) == 3
441 assert sum(item.uri is None for item in selected) == 2
442 assert {item.label for item in selected if item.uri is None} == {
443 "Daft Punk - Neon Horizon",
444 "Lunar Circuit - Chrome Reverie",
445 }
446 provider.ai_query.assert_awaited_once()
447
448
449@pytest.mark.asyncio
450async def test_hard_ai_context_preserves_non_english_grounding() -> None:
451 """Pass non-English source/catalog data through the bounded untrusted context."""
452 quiz_type, mass = _quiz_type("hard", use_ai=True)
453 source = _track("source", "Zoutelande", "BLØF")
454 correct = _track_to_candidate(source)
455 mass.music.tracks.similar_tracks.return_value = [
456 _track("st1", "Het Is Een Nacht", "Guus Meeuwis"),
457 _track("st2", "Rood", "Marco Borsato"),
458 _track("st3", "Iedereen Is Van De Wereld", "The Scene"),
459 ]
460 provider = _ai_provider(
461 _ai_response(
462 ["candidate_0", "candidate_1", "candidate_2"],
463 [
464 ("same_artist_title", "BLØF - Mooie Dagen"),
465 ("context_track", "Noorderlicht - Aan Zee"),
466 ],
467 )
468 )
469 mass.get_providers_supporting_feature.return_value = [provider]
470
471 result = list(await quiz_type._gather_distractors(source, correct))
472
473 assert {item.label for item in result[:3]} == {
474 "Guus Meeuwis - Het Is Een Nacht",
475 "BLØF - Mooie Dagen",
476 "Noorderlicht - Aan Zee",
477 }
478 prompt = provider.ai_query.await_args.args[0]
479 assert "BLØF" in prompt
480 assert "Zoutelande" in prompt
481 assert "Guus Meeuwis - Het Is Een Nacht" in prompt
482
483
484@pytest.mark.asyncio
485@pytest.mark.parametrize(
486 "synthetic",
487 [
488 [
489 ("same_artist_title", "Daft Punk - Around the World (Remix)"),
490 ("context_track", "Lunar Circuit - Chrome Reverie"),
491 ],
492 [
493 ("same_artist_title", "Neon Horizon"),
494 ("context_track", "Lunar Circuit - Chrome Reverie"),
495 ],
496 [
497 ("same_artist_title", "Daft Punk - Neon Horizon"),
498 ("context_track", "Daft Punk - Chrome Reverie"),
499 ],
500 [
501 ("same_artist_title", "Daft Punk - Neon Horizon"),
502 ("context_track", "Justice - Chrome Reverie"),
503 ],
504 ],
505)
506async def test_invalid_hard_ai_semantics_fall_back_to_real_catalog(
507 synthetic: list[tuple[str, str]],
508) -> None:
509 """Reject correct-title leakage and malformed synthetic track roles."""
510 quiz_type, mass = _quiz_type("hard", use_ai=True)
511 real_tracks = [
512 _track("st1", "Digital Love", "Daft Punk"),
513 _track("st2", "D.A.N.C.E.", "Justice"),
514 _track("st3", "Sexy Boy", "Air"),
515 ]
516 mass.music.tracks.similar_tracks.return_value = real_tracks
517 provider = _ai_provider(
518 _ai_response(
519 ["candidate_0", "candidate_1", "candidate_2"],
520 synthetic,
521 )
522 )
523 mass.get_providers_supporting_feature.return_value = [provider]
524 correct_track, correct = _correct()
525
526 result = list(await quiz_type._gather_distractors(correct_track, correct))
527
528 assert [item.label for item in result[:3]] == [
529 "Daft Punk - Digital Love",
530 "Justice - D.A.N.C.E.",
531 "Air - Sexy Boy",
532 ]
533 assert all(item.uri is not None for item in result[:3])
534 provider.ai_query.assert_awaited_once()
535
536
537@pytest.mark.asyncio
538async def test_context_track_cannot_reuse_an_individual_source_contributor() -> None:
539 """Reject a context artist that is already a source-track contributor."""
540 quiz_type, mass = _quiz_type("hard", use_ai=True)
541 correct_track, _ = _correct()
542 correct_track.artists.append(_artist("pharrell", "Pharrell Williams"))
543 correct = _track_to_candidate(correct_track)
544 real_tracks = [
545 _track("st1", "Digital Love", "Daft Punk"),
546 _track("st2", "D.A.N.C.E.", "Justice"),
547 _track("st3", "Sexy Boy", "Air"),
548 ]
549 mass.music.tracks.similar_tracks.return_value = real_tracks
550 provider = _ai_provider(
551 _ai_response(
552 ["candidate_0", "candidate_1", "candidate_2"],
553 [
554 (
555 "same_artist_title",
556 f"{correct_track.artist_str} - Neon Horizon",
557 ),
558 ("context_track", "Pharrell Williams - Chrome Reverie"),
559 ],
560 )
561 )
562 mass.get_providers_supporting_feature.return_value = [provider]
563
564 result = list(await quiz_type._gather_distractors(correct_track, correct))
565
566 assert [item.label for item in result[:3]] == [
567 "Daft Punk - Digital Love",
568 "Justice - D.A.N.C.E.",
569 "Air - Sexy Boy",
570 ]
571 provider.ai_query.assert_awaited_once()
572
573
574@pytest.mark.asyncio
575async def test_invalid_primary_ai_response_does_not_try_another_provider() -> None:
576 """Use one deterministic provider request before falling back to real tracks."""
577 quiz_type, mass = _quiz_type("hard", use_ai=True)
578 real_tracks = [
579 _track("st1", "Digital Love", "Daft Punk"),
580 _track("st2", "D.A.N.C.E.", "Justice"),
581 _track("st3", "Sexy Boy", "Air"),
582 ]
583 mass.music.tracks.similar_tracks.return_value = real_tracks
584 invalid = _ai_provider(
585 _ai_response(
586 ["candidate_0", "candidate_1", "candidate_2"],
587 [
588 ("same_artist_title", "Daft Punk - Neon Horizon"),
589 ("context_track", "Lunar Circuit - Chrome Reverie"),
590 ],
591 extra=True,
592 )
593 )
594 later = _ai_provider()
595 later.instance_id = "ai--b"
596 mass.get_providers_supporting_feature.return_value = [invalid, later]
597 correct_track, correct = _correct()
598
599 result = list(await quiz_type._gather_distractors(correct_track, correct))
600
601 assert [item.label for item in result[:3]] == [
602 "Daft Punk - Digital Love",
603 "Justice - D.A.N.C.E.",
604 "Air - Sexy Boy",
605 ]
606 invalid.ai_query.assert_awaited_once()
607 later.ai_query.assert_not_awaited()
608
609
610@pytest.mark.asyncio
611async def test_hard_ai_timeout_falls_back_to_real_catalog() -> None:
612 """Use only real candidates when the bounded AI request times out."""
613 quiz_type, mass = _quiz_type("hard", use_ai=True)
614 real_tracks = [
615 _track("st1", "Digital Love", "Daft Punk"),
616 _track("st2", "D.A.N.C.E.", "Justice"),
617 _track("st3", "Sexy Boy", "Air"),
618 ]
619 mass.music.tracks.similar_tracks.return_value = real_tracks
620 provider = _ai_provider()
621
622 async def _stall(_prompt: str) -> str:
623 await asyncio.sleep(1)
624 return ""
625
626 provider.ai_query.side_effect = _stall
627 mass.get_providers_supporting_feature.return_value = [provider]
628 correct_track, correct = _correct()
629
630 with patch(
631 "music_assistant.providers.music_quiz.quiz_types.guess_the_song.AI_QUERY_TIMEOUT_SECONDS",
632 0.001,
633 ):
634 result = list(await quiz_type._gather_distractors(correct_track, correct))
635
636 assert [item.label for item in result[:3]] == [
637 "Daft Punk - Digital Love",
638 "Justice - D.A.N.C.E.",
639 "Air - Sexy Boy",
640 ]
641 provider.ai_query.assert_awaited_once()
642
643
644@pytest.mark.asyncio
645async def test_hard_ai_provider_failure_falls_back_to_real_catalog() -> None:
646 """Use only real candidates when the selected AI provider fails."""
647 quiz_type, mass = _quiz_type("hard", use_ai=True)
648 real_tracks = [
649 _track("st1", "Digital Love", "Daft Punk"),
650 _track("st2", "D.A.N.C.E.", "Justice"),
651 _track("st3", "Sexy Boy", "Air"),
652 ]
653 mass.music.tracks.similar_tracks.return_value = real_tracks
654 provider = _ai_provider(error=RuntimeError("provider unavailable"))
655 mass.get_providers_supporting_feature.return_value = [provider]
656 correct_track, correct = _correct()
657
658 result = list(await quiz_type._gather_distractors(correct_track, correct))
659
660 assert [item.label for item in result[:3]] == [
661 "Daft Punk - Digital Love",
662 "Justice - D.A.N.C.E.",
663 "Air - Sexy Boy",
664 ]
665 provider.ai_query.assert_awaited_once()
666
667
668@pytest.mark.asyncio
669async def test_prepare_round_builds_suggestions_from_similar_tracks() -> None:
670 """A hard-mode round is assembled with exactly one correct answer from similar tracks."""
671 quiz_type, mass = _quiz_type("hard")
672 correct_track, _ = _correct()
673 quiz_type._source_track_pool = _pool([correct_track])
674 mass.music.tracks.similar_tracks.return_value = [
675 _track("st1", "Digital Love", "Daft Punk"),
676 _track("st2", "Genesis", "Justice"),
677 _track("st3", "Sexy Boy", "Air"),
678 ]
679 mass.metadata.get_image_url_for_item = AsyncMock(return_value="http://img/1")
680
681 with patch(
682 "music_assistant.providers.music_quiz.quiz_types.guess_the_song.SYSTEM_RANDOM.shuffle"
683 ) as shuffle_pool:
684 game_round = await quiz_type.prepare_round(0, [])
685
686 assert game_round.track_uri == correct_track.uri
687 assert isinstance(game_round.answer_state, MultipleChoiceRoundState)
688 suggestions = game_round.answer_state.suggestions
689 assert len(suggestions) == 4
690 assert sum(item.is_correct for item in suggestions) == 1
691 assert [item.label for item in suggestions if item.is_correct] == [CORRECT_LABEL]
692 assert all(" - " in item.label for item in suggestions)
693 shuffle_pool.assert_not_called()
694 mass.music.tracks.get_provider_item.assert_not_awaited()
695
696
697@pytest.mark.parametrize("difficulty", ["easy", "normal"])
698@pytest.mark.asyncio
699async def test_artist_round_excludes_title_only_distractors(difficulty: str) -> None:
700 """Keep every option artist-title when the correct track has an artist."""
701 quiz_type, mass = _quiz_type(difficulty)
702 correct_track, _ = _correct()
703 missing_artist = _track("missing", "Hey There Delilah", None)
704 pool_distractor = _track("pool", "Digital Love", "Daft Punk")
705 quiz_type._source_track_pool = _pool([correct_track, missing_artist, pool_distractor])
706 mass.music.search.return_value = SimpleNamespace(
707 tracks=[
708 missing_artist,
709 _track("s1", "Genesis", "Justice"),
710 _track("s2", "Sexy Boy", "Air"),
711 _track("s3", "Lisztomania", "Phoenix"),
712 ]
713 )
714
715 with patch(
716 "music_assistant.providers.music_quiz.quiz_types.guess_the_song.secrets.choice",
717 return_value=correct_track,
718 ):
719 game_round = await quiz_type.prepare_round(0, [])
720
721 assert isinstance(game_round.answer_state, MultipleChoiceRoundState)
722 suggestions = game_round.answer_state.suggestions
723 assert len(suggestions) == 4
724 assert all(" - " in suggestion.label for suggestion in suggestions)
725 assert missing_artist.uri not in {suggestion.uri for suggestion in suggestions}
726 mass.music.tracks.get_provider_item.assert_not_awaited()
727
728
729@pytest.mark.asyncio
730async def test_missing_artist_round_projects_every_option_to_title_only() -> None:
731 """Prevent the screenshot-like title-only correct answer from standing out."""
732 quiz_type, mass = _quiz_type()
733 correct_track = _track("source", "Hey There Delilah", None)
734 quiz_type._source_track_pool = _pool([correct_track])
735 mass.music.tracks.get_provider_item.return_value = correct_track
736 mass.music.search.return_value = SimpleNamespace(
737 tracks=[
738 _track("s1", "Use Somebody", "Kings of Leon"),
739 _track("s2", "Apologize", "Timbaland"),
740 _track("s3", "Chasing Cars", "Snow Patrol"),
741 ]
742 )
743
744 with patch(
745 "music_assistant.providers.music_quiz.suggestions.secrets.token_hex",
746 side_effect=["id-a", "id-b", "id-c", "id-d"],
747 ):
748 game_round = await quiz_type.prepare_round(0, [])
749
750 assert isinstance(game_round.answer_state, MultipleChoiceRoundState)
751 suggestions = game_round.answer_state.suggestions
752 assert {suggestion.label for suggestion in suggestions} == {
753 "Hey There Delilah",
754 "Use Somebody",
755 "Apologize",
756 "Chasing Cars",
757 }
758 assert all(" - " not in suggestion.label for suggestion in suggestions)
759 assert {suggestion.suggestion_id for suggestion in suggestions} == {
760 "id-a",
761 "id-b",
762 "id-c",
763 "id-d",
764 }
765 correct = next(suggestion for suggestion in suggestions if suggestion.is_correct)
766 assert correct.uri == correct_track.uri
767 assert correct.label == game_round.answer_label == "Hey There Delilah"
768 mass.music.tracks.get_provider_item.assert_awaited_once_with(
769 correct_track.item_id,
770 correct_track.provider,
771 )
772
773
774@pytest.mark.asyncio
775async def test_selected_track_recovers_artist_with_one_full_item_lookup() -> None:
776 """Use recovered artist metadata without changing the selected source identity."""
777 quiz_type, mass = _quiz_type()
778 correct_track = _track("source", "Hey There Delilah", None)
779 full_track = _track("source", "Hey There Delilah", "Plain White T's")
780 quiz_type._source_track_pool = _pool([correct_track])
781 mass.music.tracks.get_provider_item.return_value = full_track
782 mass.music.search.return_value = SimpleNamespace(
783 tracks=[
784 _track("s1", "Use Somebody", "Kings of Leon"),
785 _track("s2", "Apologize", "Timbaland"),
786 _track("s3", "Chasing Cars", "Snow Patrol"),
787 ]
788 )
789
790 game_round = await quiz_type.prepare_round(0, [])
791
792 assert isinstance(game_round.answer_state, MultipleChoiceRoundState)
793 assert game_round.track_uri == correct_track.uri
794 assert game_round.answer_label == "Plain White T's - Hey There Delilah"
795 assert all(" - " in suggestion.label for suggestion in game_round.answer_state.suggestions)
796 correct = next(
797 suggestion for suggestion in game_round.answer_state.suggestions if suggestion.is_correct
798 )
799 assert correct.uri == correct_track.uri
800 mass.music.tracks.get_provider_item.assert_awaited_once_with(
801 correct_track.item_id,
802 correct_track.provider,
803 )
804
805
806@pytest.mark.asyncio
807async def test_selected_track_artist_lookup_error_falls_back_to_title_only() -> None:
808 """Keep a playable source when its optional artist lookup fails."""
809 quiz_type, mass = _quiz_type(suggestion_count=2)
810 correct_track = _track("source", "Hey There Delilah", None)
811 quiz_type._source_track_pool = _pool([correct_track])
812 mass.music.tracks.get_provider_item.side_effect = RuntimeError("unavailable")
813 mass.music.search.return_value = SimpleNamespace(
814 tracks=[_track("s1", "Use Somebody", "Kings of Leon")]
815 )
816
817 game_round = await quiz_type.prepare_round(0, [])
818
819 assert isinstance(game_round.answer_state, MultipleChoiceRoundState)
820 assert {suggestion.label for suggestion in game_round.answer_state.suggestions} == {
821 "Hey There Delilah",
822 "Use Somebody",
823 }
824 mass.music.tracks.get_provider_item.assert_awaited_once_with(
825 correct_track.item_id,
826 correct_track.provider,
827 )
828
829
830@pytest.mark.asyncio
831async def test_normal_pool_fallback_scans_large_cached_pool_across_rounds() -> None:
832 """Fill collapsed search results lazily from one cached large source pool."""
833 quiz_type, mass = _quiz_type()
834 first_title = "Zij W\u0069l Mij"
835 first_correct = _track("first", first_title, "FLEMMING")
836 second_correct = _track("second", "Teardrop", "Massive Attack")
837 alternatives = [
838 _track("a1", "Genesis", "Justice"),
839 _track("a2", "Lisztomania", "Phoenix"),
840 _track("a3", "Chasing Cars", "Snow Patrol"),
841 _track("a4", "Midnight City", "M83"),
842 ]
843 source_pool = _pool(
844 [
845 first_correct,
846 second_correct,
847 *alternatives,
848 *[
849 _track(f"bulk-{index}", f"Bulk Song {index}", f"Bulk Artist {index}")
850 for index in range(1000)
851 ],
852 ]
853 )
854 quiz_type._source_track_pool = source_pool
855 mass.music.search.side_effect = [
856 SimpleNamespace(
857 tracks=[
858 first_correct,
859 _track("first-copy", first_title, "FLEMMING"),
860 _track("first-remix", f"{first_title} (Remix)", "FLEMMING"),
861 ]
862 ),
863 SimpleNamespace(
864 tracks=[
865 second_correct,
866 _track("second-copy", "Teardrop", "Massive Attack"),
867 _track("second-remix", "Teardrop (Remastered)", "Massive Attack"),
868 ]
869 ),
870 ]
871
872 with (
873 patch(
874 "music_assistant.providers.music_quiz.quiz_types.guess_the_song.secrets.choice",
875 side_effect=[first_correct, second_correct],
876 ),
877 patch(
878 "music_assistant.providers.music_quiz.quiz_types.guess_the_song.SYSTEM_RANDOM.shuffle"
879 ),
880 patch(
881 "music_assistant.providers.music_quiz.quiz_types.guess_the_song._track_to_candidate",
882 wraps=_track_to_candidate,
883 ) as convert_candidate,
884 ):
885 first_round = await quiz_type.prepare_round(0, [])
886 second_round = await quiz_type.prepare_round(1, [first_round])
887
888 for game_round in (first_round, second_round):
889 assert isinstance(game_round.answer_state, MultipleChoiceRoundState)
890 assert len(game_round.answer_state.suggestions) == 4
891 assert all(" - " in item.label for item in game_round.answer_state.suggestions)
892 assert first_round.answer_label == f"FLEMMING - {first_title}"
893 assert second_round.answer_label == "Massive Attack - Teardrop"
894 assert convert_candidate.call_count == 14
895 assert mass.music.search.await_count == 2
896 assert all(item.kwargs["limit"] == 32 for item in mass.music.search.await_args_list)
897 mass.music.get_item.assert_not_awaited()
898 mass.music.tracks.get_provider_item.assert_not_awaited()
899 mass.music.tracks.similar_tracks.assert_not_awaited()
900 assert quiz_type._source_track_pool is source_pool
901
902
903@pytest.mark.asyncio
904async def test_title_only_projection_refilters_duplicate_and_close_titles() -> None:
905 """Continue through pool collisions until enough projected titles survive."""
906 quiz_type, mass = _quiz_type()
907 correct_track = _track("source", "Hey There Delilah", None)
908 quiz_type._source_track_pool = _pool(
909 [
910 correct_track,
911 _track("s1", "Electric Feel", "Artist One"),
912 _track("s2", "Electric Feel", "Artist Two"),
913 _track("s3", "Electric Feel (Radio Edit)", "Artist Three"),
914 _track("s4", "Genesis", "Justice"),
915 _track("s5", "Lisztomania", "Phoenix"),
916 _track("unused", "Teardrop", "Massive Attack"),
917 ]
918 )
919 mass.music.tracks.get_provider_item.return_value = correct_track
920
921 with (
922 patch(
923 "music_assistant.providers.music_quiz.quiz_types.guess_the_song.secrets.choice",
924 return_value=correct_track,
925 ),
926 patch(
927 "music_assistant.providers.music_quiz.quiz_types.guess_the_song.SYSTEM_RANDOM.shuffle"
928 ),
929 patch(
930 "music_assistant.providers.music_quiz.quiz_types.guess_the_song._track_to_candidate",
931 wraps=_track_to_candidate,
932 ) as convert_candidate,
933 ):
934 game_round = await quiz_type.prepare_round(0, [])
935
936 assert isinstance(game_round.answer_state, MultipleChoiceRoundState)
937 assert {suggestion.label for suggestion in game_round.answer_state.suggestions} == {
938 "Hey There Delilah",
939 "Electric Feel",
940 "Genesis",
941 "Lisztomania",
942 }
943 assert convert_candidate.call_count == 6
944 mass.music.search.assert_awaited_once()
945
946
947@pytest.mark.asyncio
948async def test_title_only_projection_reports_insufficient_candidates() -> None:
949 """Return the localized distractor error when projected titles all overlap."""
950 quiz_type, mass = _quiz_type(suggestion_count=3)
951 correct_track = _track("source", "Hey There Delilah", None)
952 quiz_type._source_track_pool = _pool(
953 [
954 correct_track,
955 _track("s1", "Electric Feel", "Artist One"),
956 _track("s2", "Electric Feel", "Artist Two"),
957 _track("s3", "Electric Feel (Radio Edit)", "Artist Three"),
958 ]
959 )
960 mass.music.tracks.get_provider_item.return_value = correct_track
961
962 with (
963 patch(
964 "music_assistant.providers.music_quiz.quiz_types.guess_the_song.secrets.choice",
965 return_value=correct_track,
966 ),
967 patch(
968 "music_assistant.providers.music_quiz.quiz_types.guess_the_song.SYSTEM_RANDOM.shuffle"
969 ),
970 pytest.raises(InvalidDataError) as err,
971 ):
972 await quiz_type.prepare_round(0, [])
973
974 assert err.value.translation_key == "music_quiz_not_enough_distractors"
975 mass.music.search.assert_awaited_once()
976
977
978@pytest.mark.asyncio
979async def test_hard_ai_uses_pool_to_fill_real_candidate_slots() -> None:
980 """Keep the hard AI composition when preferred real candidates are sparse."""
981 quiz_type, mass = _quiz_type("hard", use_ai=True, suggestion_count=6)
982 correct_track, _ = _correct()
983 pool_tracks = [
984 _track("p1", "Genesis", "Justice"),
985 _track("p2", "Lisztomania", "Phoenix"),
986 _track("p3", "Chasing Cars", "Snow Patrol"),
987 ]
988 quiz_type._source_track_pool = _pool([correct_track, *pool_tracks])
989 similar = _track("similar", "Digital Love", "Daft Punk")
990 mass.music.tracks.similar_tracks.return_value = [similar]
991 mass.music.search.return_value = SimpleNamespace(
992 tracks=[
993 correct_track,
994 _track("copy", "Around the World", "Daft Punk"),
995 _track("remix", "Around the World (Remix)", "Daft Punk"),
996 ]
997 )
998 provider = _ai_provider(
999 _ai_response(
1000 ["candidate_0", "candidate_1", "candidate_2"],
1001 [
1002 ("same_artist_title", "Daft Punk - Neon Horizon"),
1003 ("context_track", "Lunar Circuit - Chrome Reverie"),
1004 ],
1005 )
1006 )
1007 mass.get_providers_supporting_feature.return_value = [provider]
1008
1009 with (
1010 patch(
1011 "music_assistant.providers.music_quiz.quiz_types.guess_the_song.secrets.choice",
1012 return_value=correct_track,
1013 ),
1014 patch(
1015 "music_assistant.providers.music_quiz.quiz_types.guess_the_song.SYSTEM_RANDOM.shuffle"
1016 ),
1017 ):
1018 game_round = await quiz_type.prepare_round(0, [])
1019
1020 assert isinstance(game_round.answer_state, MultipleChoiceRoundState)
1021 suggestions = game_round.answer_state.suggestions
1022 assert len(suggestions) == 6
1023 assert sum(item.is_correct for item in suggestions) == 1
1024 assert sum(item.uri is None for item in suggestions) == 2
1025 assert len({item.uri for item in suggestions} & {item.uri for item in pool_tracks}) == 2
1026 assert all(" - " in item.label for item in suggestions)
1027 provider.ai_query.assert_awaited_once()
1028 mass.music.tracks.similar_tracks.assert_awaited_once_with(
1029 item_id=correct_track.item_id,
1030 provider_instance_id_or_domain=correct_track.provider,
1031 limit=24,
1032 )
1033 assert mass.music.search.await_args.kwargs["limit"] == 48
1034 mass.music.get_item.assert_not_awaited()
1035
1036
1037@pytest.mark.asyncio
1038async def test_prepare_round_keeps_hard_ai_artist_title_shape() -> None:
1039 """Retain artist-title formatting for real and synthetic hard-mode options."""
1040 quiz_type, mass = _quiz_type("hard", use_ai=True)
1041 correct_track = _track("c1", "Around the World", None)
1042 quiz_type._source_track_pool = _pool([correct_track])
1043 mass.music.tracks.get_provider_item.return_value = _track(
1044 "c1",
1045 "Around the World",
1046 "Daft Punk",
1047 )
1048 mass.music.tracks.similar_tracks.return_value = [
1049 _track("st1", "Digital Love", "Daft Punk"),
1050 _track("st2", "D.A.N.C.E.", "Justice"),
1051 _track("st3", "Sexy Boy", "Air"),
1052 ]
1053 provider = _ai_provider(
1054 _ai_response(
1055 ["candidate_2", "candidate_0", "candidate_1"],
1056 [
1057 ("same_artist_title", "Daft Punk - Neon Horizon"),
1058 ("context_track", "Lunar Circuit - Chrome Reverie"),
1059 ],
1060 )
1061 )
1062 mass.get_providers_supporting_feature.return_value = [provider]
1063
1064 game_round = await quiz_type.prepare_round(0, [])
1065
1066 assert isinstance(game_round.answer_state, MultipleChoiceRoundState)
1067 suggestions = game_round.answer_state.suggestions
1068 assert len(suggestions) == 4
1069 assert all(" - " in suggestion.label for suggestion in suggestions)
1070 assert game_round.answer_label == CORRECT_LABEL
1071 assert sum(suggestion.is_correct for suggestion in suggestions) == 1
1072 provider.ai_query.assert_awaited_once()
1073 mass.music.tracks.get_provider_item.assert_awaited_once_with(
1074 correct_track.item_id,
1075 correct_track.provider,
1076 )
1077