/
/
/
1"""Tests for the Music Timeline Music Quiz type."""
2
3from __future__ import annotations
4
5import asyncio
6import json
7from collections.abc import Sequence
8from datetime import UTC, datetime
9from types import SimpleNamespace
10from typing import cast
11from unittest.mock import AsyncMock, MagicMock, patch
12
13import pytest
14from music_assistant_models.enums import AlbumType, ExternalID, MediaType
15from music_assistant_models.errors import InvalidDataError, MediaNotFoundError, RateLimited
16from music_assistant_models.media_items import (
17 Album,
18 Artist,
19 ItemMapping,
20 ProviderMapping,
21 Track,
22)
23from music_assistant_models.unique_list import UniqueList
24
25from music_assistant.constants import VARIOUS_ARTISTS_MBID, VARIOUS_ARTISTS_NAME
26from music_assistant.controllers.music.recency import RecencySnapshot
27from music_assistant.models.plugin import AIEngine, PluginProvider
28from music_assistant.providers.music_quiz.constants import AI_QUERY_TIMEOUT_SECONDS
29from music_assistant.providers.music_quiz.models import (
30 MusicQuizConfig,
31 MusicQuizDifficulty,
32 MusicQuizRound,
33 TimelineBonusMode,
34 TimelineBonusType,
35 TimelineFreeTextBonusDefinition,
36 TimelineMultipleChoiceBonusDefinition,
37 TimelineRoundState,
38)
39from music_assistant.providers.music_quiz.quiz_types.base import PLAYBACK_REPLACEMENT_RESERVE
40from music_assistant.providers.music_quiz.quiz_types.music_timeline import (
41 BONUS_CANDIDATE_LIMIT,
42 DEFAULT_BONUS_OPTION_COUNT,
43 TRACK_ENRICHMENT_CONCURRENCY,
44 MusicTimelineQuizType,
45)
46from music_assistant.providers.music_quiz.suggestions import (
47 SuggestionCandidate,
48 answer_labels_are_too_close,
49)
50
51# album types whose year describes a reissue rather than an original release
52UNTRUSTED_ALBUM_TYPES = frozenset({AlbumType.COMPILATION, AlbumType.LIVE, AlbumType.SOUNDTRACK})
53
54
55def _track(
56 item_id: str,
57 name: str,
58 artist: str,
59 *,
60 album_year: int | None = None,
61 release_year: int | None = None,
62 available: bool = True,
63 is_playable: bool = True,
64 album_mapping: bool = False,
65) -> Track:
66 """Return a track with a resolved or unresolved album and configurable years."""
67 provider = "prov"
68 album: Album | ItemMapping = (
69 ItemMapping(
70 media_type=MediaType.ALBUM,
71 item_id=f"album-{item_id}",
72 provider=provider,
73 name=f"Album {item_id}",
74 year=album_year,
75 )
76 if album_mapping
77 else _full_album(f"album-{item_id}", f"Album {item_id}", year=album_year)
78 )
79 track = Track(
80 item_id=item_id,
81 provider=provider,
82 name=name,
83 is_playable=is_playable,
84 duration=180,
85 artists=UniqueList(
86 [
87 ItemMapping(
88 media_type=MediaType.ARTIST,
89 item_id=f"artist-{item_id}",
90 provider=provider,
91 name=artist,
92 )
93 ]
94 ),
95 album=album,
96 provider_mappings={
97 ProviderMapping(
98 item_id=item_id,
99 provider_domain=provider,
100 provider_instance=provider,
101 available=available,
102 )
103 },
104 )
105 if release_year is not None:
106 track.metadata.release_date = datetime(release_year, 1, 1, tzinfo=UTC)
107 return track
108
109
110def _full_album(
111 item_id: str,
112 name: str,
113 *,
114 album_type: AlbumType = AlbumType.ALBUM,
115 artists: Sequence[Artist | ItemMapping] = (),
116 year: int | None = None,
117 provider: str = "prov",
118) -> Album:
119 """Return a full album with configurable compilation evidence."""
120 return Album(
121 item_id=item_id,
122 provider=provider,
123 name=name,
124 album_type=album_type,
125 artists=UniqueList(artists),
126 year=year,
127 provider_mappings={
128 ProviderMapping(
129 item_id=item_id,
130 provider_domain=provider,
131 provider_instance=provider,
132 )
133 },
134 )
135
136
137def _album_artist(
138 item_id: str,
139 name: str,
140 *,
141 mbid: str | None = None,
142 provider: str = "prov",
143) -> Artist:
144 """Return a full album artist with optional MusicBrainz identity."""
145 return Artist(
146 item_id=item_id,
147 provider=provider,
148 name=name,
149 external_ids={(ExternalID.MB_ARTIST, mbid)} if mbid else set(),
150 provider_mappings=set(),
151 )
152
153
154def _artist(item_id: str, name: str) -> ItemMapping:
155 """Return a catalog artist mapping."""
156 return ItemMapping(
157 media_type=MediaType.ARTIST,
158 item_id=item_id,
159 provider="prov",
160 name=name,
161 )
162
163
164def _ai_provider(
165 response: object = None,
166 error: Exception | None = None,
167) -> MagicMock:
168 """Return a mock AI-query plugin provider exposing a single engine."""
169 provider = MagicMock(spec=PluginProvider)
170 provider.instance_id = "ai--test"
171 provider.ai_query = AsyncMock(return_value=response, side_effect=error)
172 provider.get_ai_engines = AsyncMock(
173 return_value=[AIEngine(id="engine", name="ai--test", provider=provider)]
174 )
175 return provider
176
177
178def _ai_response(
179 ranked_ids: list[str],
180 synthetic: list[tuple[str, str]],
181 **extra: object,
182) -> str:
183 """Return a structured AI distractor response."""
184 return json.dumps(
185 {
186 "ranked_ids": ranked_ids,
187 "synthetic": [{"kind": kind, "label": label} for kind, label in synthetic],
188 **extra,
189 }
190 )
191
192
193def _mass() -> MagicMock:
194 """Return a mock MusicAssistant for Music Timeline tests."""
195 mass = MagicMock()
196 mass.metadata.get_image_url_for_item = AsyncMock(
197 side_effect=lambda track: f"https://img/{track.item_id}"
198 )
199 mass.music.tracks.get = AsyncMock()
200 mass.music.tracks.similar_tracks = AsyncMock(return_value=[])
201 mass.music.artists.similar_artists = AsyncMock(return_value=[])
202 mass.music.artists.top_tracks = AsyncMock(return_value=[])
203 mass.music.artists.tracks = AsyncMock(return_value=[])
204 mass.music.search = AsyncMock(return_value=SimpleNamespace(tracks=[]))
205 mass.music.recency.snapshot = AsyncMock(return_value=RecencySnapshot(now=0))
206 mass.get_providers_supporting_feature = MagicMock(return_value=[])
207 return mass
208
209
210def _quiz(
211 tracks: list[Track],
212 *,
213 round_count: int = 1,
214 artist_mode: TimelineBonusMode = TimelineBonusMode.OFF,
215 title_mode: TimelineBonusMode = TimelineBonusMode.OFF,
216 use_ai: bool = False,
217) -> tuple[MusicTimelineQuizType, MagicMock]:
218 """Return a Music Timeline type backed by a deterministic source pool."""
219 mass = _mass()
220 config = MusicQuizConfig(
221 round_count=round_count,
222 source_uris=["prov://playlist/1"],
223 artist_bonus_mode=artist_mode,
224 title_bonus_mode=title_mode,
225 use_ai_distractors=use_ai,
226 ai_engine="ai--test/engine" if use_ai else None,
227 )
228 quiz = MusicTimelineQuizType(mass, config)
229 quiz._source_track_pool = {track.uri: track for track in tracks if track.uri}
230 return quiz, mass
231
232
233def _with_musicbrainz(
234 mass: MagicMock,
235 years: dict[str, int] | None = None,
236 error: Exception | None = None,
237 name_years: dict[tuple[str, str], int] | None = None,
238) -> MagicMock:
239 """
240 Attach a MusicBrainz provider to the mock MusicAssistant.
241
242 :param mass: Mock MusicAssistant to attach the provider to.
243 :param years: Release year per ISRC.
244 :param error: Raised by both lookups instead of answering.
245 :param name_years: Release year per (artist name, track name), for tracks without an ISRC.
246 """
247 provider = MagicMock()
248 provider.get_release_year_by_isrc = AsyncMock(
249 side_effect=error if error is not None else lambda isrc: (years or {}).get(isrc)
250 )
251 provider.get_release_year_by_track_name = AsyncMock(
252 side_effect=error
253 if error is not None
254 else lambda artist, track: (name_years or {}).get((artist, track))
255 )
256 mass.get_provider = MagicMock(
257 side_effect=lambda domain: provider if domain == "musicbrainz" else None
258 )
259 return provider
260
261
262def _with_isrc(track: Track, isrc: str) -> Track:
263 """Return the given track carrying an ISRC."""
264 track.add_external_id(ExternalID.ISRC, isrc)
265 return track
266
267
268async def _prepare_round_with_tracks(
269 quiz: MusicTimelineQuizType,
270 tracks: list[Track],
271) -> MusicQuizRound:
272 """Prepare the first round with the given tracks selected as anchor and candidate."""
273 with patch(
274 "music_assistant.providers.music_quiz.quiz_types.music_timeline.secrets.choice",
275 side_effect=tracks,
276 ):
277 return await quiz.prepare_round(0, [])
278
279
280def test_config_normalization_and_validation_are_music_timeline_specific() -> None:
281 """Use fixed option defaults without requiring guess-the-song settings."""
282 config = MusicQuizConfig(
283 round_count=2,
284 suggestion_count=9,
285 source_uris=["prov://playlist/1"],
286 include_similar_music=True,
287 difficulty="not-used",
288 use_ai_distractors=True,
289 artist_bonus_mode=TimelineBonusMode.FREE_TEXT,
290 title_bonus_mode=TimelineBonusMode.MULTIPLE_CHOICE,
291 )
292
293 normalized = MusicTimelineQuizType.normalize_config(config)
294 MusicTimelineQuizType.validate_config(normalized)
295
296 assert normalized.suggestion_count == DEFAULT_BONUS_OPTION_COUNT
297 assert normalized.difficulty == MusicQuizDifficulty.NORMAL.value
298 assert normalized.use_ai_distractors is True
299 assert normalized.include_similar_music is True
300 with pytest.raises(InvalidDataError, match="bonus mode"):
301 MusicTimelineQuizType.validate_config(
302 MusicQuizConfig(
303 source_uris=["prov://playlist/1"],
304 artist_bonus_mode=cast("TimelineBonusMode", "invalid"),
305 )
306 )
307
308
309@pytest.mark.asyncio
310async def test_initialize_requires_anchor_plus_unique_scored_tracks() -> None:
311 """Validate the complete pool without retaining an ephemeral track sequence."""
312 tracks = [
313 _track("one", "Teardrop", "Massive Attack", album_year=1998),
314 _track("two", "Genesis", "Justice", album_year=2007),
315 _track("three", "Lisztomania", "Phoenix", album_year=2009),
316 _track("four", "Sexy Boy", "Air", album_year=1998),
317 ]
318 quiz, _ = _quiz(tracks, round_count=3)
319
320 await quiz.initialize()
321
322 assert quiz._eligible_tracks is not None
323 assert len(quiz._eligible_tracks) == 4
324 assert not hasattr(quiz, "_selected_tracks")
325
326
327@pytest.mark.asyncio
328async def test_initialize_rejects_missing_or_insufficient_dated_tracks() -> None:
329 """Raise localized errors when real release years cannot fill the game."""
330 missing = _track("missing", "Unknown", "Artist")
331 quiz, mass = _quiz([missing])
332 mass.music.tracks.get.return_value = missing
333
334 with pytest.raises(InvalidDataError) as no_dated:
335 await quiz.initialize()
336 assert no_dated.value.translation_key == "music_quiz_no_dated_tracks"
337
338 tracks = [
339 _track("one", "Teardrop", "Massive Attack", album_year=1998),
340 _track("two", "Genesis", "Justice", album_year=2007),
341 ]
342 quiz, _ = _quiz(tracks, round_count=2)
343 with pytest.raises(InvalidDataError) as insufficient:
344 await quiz.initialize()
345 assert insufficient.value.translation_key == "music_quiz_not_enough_dated_tracks"
346
347
348@pytest.mark.asyncio
349async def test_initialize_excludes_unavailable_and_unplayable_dated_tracks() -> None:
350 """Never count or select dated tracks that cannot actually be played."""
351 playable = [
352 _track("one", "Teardrop", "Massive Attack", album_year=1998),
353 _track("two", "Genesis", "Justice", album_year=2007),
354 ]
355 unavailable = _track(
356 "unavailable",
357 "Unavailable",
358 "Artist",
359 album_year=2001,
360 available=False,
361 )
362 unplayable = _track(
363 "unplayable",
364 "Unplayable",
365 "Artist",
366 album_year=2002,
367 is_playable=False,
368 )
369 quiz, mass = _quiz([*playable, unavailable, unplayable])
370
371 await quiz.initialize()
372
373 assert quiz._eligible_tracks is not None
374 assert {track.item_id for track in quiz._eligible_tracks} == {"one", "two"}
375 mass.music.tracks.get.assert_not_awaited()
376
377 insufficient_quiz, _ = _quiz([playable[0], unavailable, unplayable])
378 with pytest.raises(InvalidDataError) as insufficient:
379 await insufficient_quiz.initialize()
380 assert insufficient.value.translation_key == "music_quiz_not_enough_dated_tracks"
381
382
383@pytest.mark.asyncio
384async def test_large_eligible_pool_skips_enrichment_and_remains_available() -> None:
385 """Preserve every ready source track without requesting redundant details."""
386 tracks = [
387 _track(
388 f"track-{index}",
389 f"Track {index}",
390 f"Artist {index}",
391 album_year=1980 + index,
392 )
393 for index in range(40)
394 ]
395 quiz, mass = _quiz(tracks, round_count=5)
396
397 await quiz.initialize()
398
399 mass.music.tracks.get.assert_not_awaited()
400 assert quiz._eligible_tracks == tracks
401
402
403@pytest.mark.asyncio
404async def test_partial_playlist_tracks_are_enriched_through_track_api() -> None:
405 """Fetch full track details when a playlist item lacks release metadata."""
406 partial = _track("partial", "Teardrop", "Massive Attack")
407 enriched = _track("partial", "Teardrop", "Massive Attack", album_year=1998)
408 other = _track("other", "Genesis", "Justice", album_year=2007)
409 quiz, mass = _quiz([partial, other])
410 mass.music.tracks.get.return_value = enriched
411
412 await quiz.initialize()
413
414 mass.music.tracks.get.assert_awaited_once_with(partial.item_id, partial.provider)
415 assert quiz._eligible_tracks is not None
416 assert {track.item_id for track in quiz._eligible_tracks} == {"partial", "other"}
417
418
419@pytest.mark.asyncio
420async def test_track_enrichment_stops_after_game_requirement_and_reserve() -> None:
421 """Stop requesting details once the complete game has replacement capacity."""
422 ready = [
423 _track("ready-one", "Ready One", "Artist One", album_year=1990),
424 _track("ready-two", "Ready Two", "Artist Two", album_year=1991),
425 ]
426 partial = [
427 _track(f"partial-{index}", f"Partial {index}", f"Artist {index}") for index in range(20)
428 ]
429 enriched = {
430 track.item_id: _track(
431 track.item_id,
432 track.name,
433 track.artist_str,
434 album_year=2000 + index,
435 )
436 for index, track in enumerate(partial)
437 }
438 round_count = 3
439 target_count = round_count + 1 + PLAYBACK_REPLACEMENT_RESERVE
440 expected_enrichments = target_count - len(ready)
441 quiz, mass = _quiz([*ready, *partial], round_count=round_count)
442 mass.music.tracks.get.side_effect = lambda item_id, _provider: enriched[item_id]
443
444 await quiz.initialize()
445
446 assert mass.music.tracks.get.await_count == expected_enrichments
447 enriched_ids = [await_call.args[0] for await_call in mass.music.tracks.get.await_args_list]
448 assert len(set(enriched_ids)) == expected_enrichments
449 assert set(enriched_ids) <= {track.item_id for track in partial}
450 assert quiz._eligible_tracks is not None
451 assert len(quiz._eligible_tracks) == target_count
452 assert quiz._eligible_tracks[: len(ready)] == ready
453
454
455@pytest.mark.asyncio
456async def test_bounded_enrichment_samples_across_the_whole_source_pool() -> None:
457 """Draw the bounded enrichment sample from anywhere in the source pool."""
458 partial = [
459 _track(f"partial-{index}", f"Partial {index}", f"Artist {index}") for index in range(20)
460 ]
461 enriched = {
462 track.item_id: _track(
463 track.item_id,
464 track.name,
465 track.artist_str,
466 album_year=2000 + index,
467 )
468 for index, track in enumerate(partial)
469 }
470 round_count = 3
471 target_count = round_count + 1 + PLAYBACK_REPLACEMENT_RESERVE
472 quiz, mass = _quiz(partial, round_count=round_count)
473 mass.music.tracks.get.side_effect = lambda item_id, _provider: enriched[item_id]
474
475 with patch(
476 "music_assistant.providers.music_quiz.quiz_types.music_timeline.SYSTEM_RANDOM.shuffle",
477 side_effect=list.reverse,
478 ) as shuffle:
479 await quiz.initialize()
480
481 shuffle.assert_called_once()
482 enriched_ids = [await_call.args[0] for await_call in mass.music.tracks.get.await_args_list]
483 assert enriched_ids == [track.item_id for track in reversed(partial[-target_count:])]
484 assert quiz._eligible_tracks is not None
485 assert len(quiz._eligible_tracks) == target_count
486
487
488def test_only_unresolved_album_mapping_years_require_enrichment() -> None:
489 """Require album resolution only for a year that can come from an album mapping."""
490 resolved = _track("resolved", "Resolved", "Artist", album_year=1998)
491 mapped = _track("mapped", "Mapped", "Artist", album_year=1998, album_mapping=True)
492 mapped_with_track_year = _track(
493 "own-year",
494 "Own Year",
495 "Artist",
496 album_year=1998,
497 release_year=1962,
498 album_mapping=True,
499 )
500 mapped_without_album_year = _track(
501 "undated-album",
502 "Undated Album",
503 "Artist",
504 release_year=1975,
505 album_mapping=True,
506 )
507 undated = _track("undated", "Undated", "Artist")
508
509 assert MusicTimelineQuizType._track_is_resolved(resolved) is True
510 assert MusicTimelineQuizType._track_is_resolved(mapped) is False
511 assert MusicTimelineQuizType._track_is_resolved(mapped_with_track_year) is True
512 assert MusicTimelineQuizType._track_is_resolved(mapped_without_album_year) is True
513 assert MusicTimelineQuizType._track_is_resolved(undated) is False
514
515
516@pytest.mark.asyncio
517async def test_album_mapping_years_are_enriched_before_they_are_trusted() -> None:
518 """Resolve an album mapping before accepting the release year it reports."""
519 mapped = _track("mapped", "Mapped Track", "Artist", album_year=1998, album_mapping=True)
520 enriched = _track("mapped", "Mapped Track", "Artist")
521 enriched.album = _full_album("album-mapped", "Real Album", year=1998)
522 resolved = _track("resolved", "Resolved Track", "Other Artist", album_year=2007)
523 quiz, mass = _quiz([mapped, resolved])
524 mass.music.tracks.get.return_value = enriched
525
526 await quiz.initialize()
527
528 enriched_ids = [await_call.args[0] for await_call in mass.music.tracks.get.await_args_list]
529 assert enriched_ids == [mapped.item_id]
530 assert quiz._eligible_tracks is not None
531 assert {track.item_id for track in quiz._eligible_tracks} == {"mapped", "resolved"}
532
533
534@pytest.mark.asyncio
535async def test_enrichment_excludes_tracks_revealed_as_compilations() -> None:
536 """Drop a track once enrichment reveals its album is a compilation."""
537 mapped = _track("mapped", "Mapped Track", "Artist", album_year=1998, album_mapping=True)
538 enriched = _track("mapped", "Mapped Track", "Artist")
539 enriched.album = _full_album(
540 "album-mapped",
541 "Party Hits",
542 album_type=AlbumType.COMPILATION,
543 year=1998,
544 )
545 resolved = [
546 _track(f"resolved-{index}", f"Resolved {index}", "Artist", album_year=1990 + index)
547 for index in range(2)
548 ]
549 quiz, mass = _quiz([mapped, *resolved])
550 mass.music.tracks.get.return_value = enriched
551
552 await quiz.initialize()
553
554 mass.music.tracks.get.assert_awaited_once_with(mapped.item_id, mapped.provider)
555 assert quiz._eligible_tracks is not None
556 assert {track.item_id for track in quiz._eligible_tracks} == {"resolved-0", "resolved-1"}
557
558
559@pytest.mark.asyncio
560@pytest.mark.parametrize("album_type", [AlbumType.ALBUM, AlbumType.SINGLE])
561async def test_enrichment_keeps_tracks_revealed_as_original_releases(
562 album_type: AlbumType,
563) -> None:
564 """Keep the resolved year of a track whose album dates an original release."""
565 mapped = _track("mapped", "Mapped Track", "Artist", album_year=1998, album_mapping=True)
566 enriched = _track("mapped", "Mapped Track", "Artist")
567 enriched.album = _full_album(
568 "album-mapped",
569 "Real Album",
570 album_type=album_type,
571 year=1971,
572 )
573 resolved = [
574 _track(f"resolved-{index}", f"Resolved {index}", "Artist", album_year=1990 + index)
575 for index in range(2)
576 ]
577 quiz, mass = _quiz([mapped, *resolved])
578 mass.music.tracks.get.return_value = enriched
579
580 await quiz.initialize()
581
582 assert quiz._eligible_tracks is not None
583 eligible_tracks = {track.item_id: track for track in quiz._eligible_tracks}
584 assert set(eligible_tracks) == {"mapped", "resolved-0", "resolved-1"}
585 assert MusicTimelineQuizType._release_year(eligible_tracks["mapped"]) == 1971
586
587
588@pytest.mark.asyncio
589async def test_album_mappings_stay_eligible_when_enrichment_fails() -> None:
590 """Trust an album mapping year when its resolution raises."""
591 mapped = [
592 _track(
593 f"mapped-{index}",
594 f"Mapped {index}",
595 f"Artist {index}",
596 album_year=1990 + index,
597 album_mapping=True,
598 )
599 for index in range(12)
600 ]
601 round_count = 3
602 target_count = round_count + 1 + PLAYBACK_REPLACEMENT_RESERVE
603 quiz, mass = _quiz(mapped, round_count=round_count)
604 mass.music.tracks.get.side_effect = MediaNotFoundError("provider unavailable")
605
606 await quiz.initialize()
607
608 assert quiz._eligible_tracks is not None
609 assert len(quiz._eligible_tracks) == target_count
610
611
612@pytest.mark.asyncio
613async def test_album_mappings_stay_eligible_when_enrichment_cannot_resolve_them() -> None:
614 """Trust an album mapping year once its resolution has been attempted."""
615 mapped = [
616 _track(
617 f"mapped-{index}",
618 f"Mapped {index}",
619 f"Artist {index}",
620 album_year=1990 + index,
621 album_mapping=True,
622 )
623 for index in range(12)
624 ]
625 round_count = 3
626 target_count = round_count + 1 + PLAYBACK_REPLACEMENT_RESERVE
627 tracks_by_id = {track.item_id: track for track in mapped}
628 quiz, mass = _quiz(mapped, round_count=round_count)
629 mass.music.tracks.get.side_effect = lambda item_id, _provider: tracks_by_id[item_id]
630
631 await quiz.initialize()
632
633 assert mass.music.tracks.get.await_count == target_count
634 assert quiz._eligible_tracks is not None
635 assert len(quiz._eligible_tracks) == target_count
636 assert [MusicTimelineQuizType._release_year(track) for track in quiz._eligible_tracks] == [
637 1990 + int(track.item_id.removeprefix("mapped-")) for track in quiz._eligible_tracks
638 ]
639
640
641@pytest.mark.asyncio
642async def test_insufficient_enrichment_exhausts_unresolved_tracks() -> None:
643 """Check every unresolved source track before reporting an insufficient pool."""
644 ready = [
645 _track("ready-one", "Ready One", "Artist One", album_year=1990),
646 _track("ready-two", "Ready Two", "Artist Two", album_year=1991),
647 ]
648 unresolved = [
649 _track(f"partial-{index}", f"Partial {index}", f"Artist {index}") for index in range(12)
650 ]
651 quiz, mass = _quiz([*ready, *unresolved], round_count=4)
652 mass.music.tracks.get.side_effect = lambda item_id, _provider: next(
653 track for track in unresolved if track.item_id == item_id
654 )
655
656 with pytest.raises(InvalidDataError) as insufficient:
657 await quiz.initialize()
658
659 assert insufficient.value.translation_key == "music_quiz_not_enough_dated_tracks"
660 assert mass.music.tracks.get.await_count == len(unresolved)
661
662
663@pytest.mark.asyncio
664@pytest.mark.parametrize(
665 ("available", "is_playable"),
666 [(False, True), (True, False)],
667)
668async def test_enriched_tracks_must_still_be_available_and_playable(
669 available: bool,
670 is_playable: bool,
671) -> None:
672 """Exclude enriched details when they reveal an unusable source track."""
673 partial = _track("partial", "Partial", "Artist")
674 enriched = _track(
675 "partial",
676 "Partial",
677 "Artist",
678 album_year=2000,
679 available=available,
680 is_playable=is_playable,
681 )
682 playable = [
683 _track("one", "Teardrop", "Massive Attack", album_year=1998),
684 _track("two", "Genesis", "Justice", album_year=2007),
685 ]
686 quiz, mass = _quiz([partial, *playable])
687 mass.music.tracks.get.return_value = enriched
688
689 await quiz.initialize()
690
691 assert quiz._eligible_tracks is not None
692 assert {track.item_id for track in quiz._eligible_tracks} == {"one", "two"}
693
694
695@pytest.mark.asyncio
696async def test_track_enrichment_concurrency_is_bounded() -> None:
697 """Process large undated pools in bounded batches."""
698 tracks = [_track(f"track-{index}", f"Track {index}", f"Artist {index}") for index in range(25)]
699 tracks_by_id = {track.item_id: track for track in tracks}
700 quiz, mass = _quiz(tracks)
701 active_calls = 0
702 max_active_calls = 0
703
704 async def _get_track(item_id: str, _provider: str) -> Track:
705 nonlocal active_calls, max_active_calls
706 active_calls += 1
707 max_active_calls = max(max_active_calls, active_calls)
708 await asyncio.sleep(0)
709 active_calls -= 1
710 return tracks_by_id[item_id]
711
712 mass.music.tracks.get.side_effect = _get_track
713
714 with pytest.raises(InvalidDataError):
715 await quiz.initialize()
716
717 assert mass.music.tracks.get.await_count == len(tracks)
718 assert max_active_calls <= TRACK_ENRICHMENT_CONCURRENCY
719
720
721@pytest.mark.asyncio
722async def test_musicbrainz_year_only_wins_when_it_predates_the_library_year() -> None:
723 """Keep the earliest of the library and MusicBrainz release year."""
724 reissue = _with_isrc(
725 _track("reissue", "Immigrant Song", "Led Zeppelin", album_year=2018),
726 "ISRC-REISSUE",
727 )
728 # remasters get their own ISRC, so MusicBrainz can date a song later than the library does
729 remaster = _with_isrc(
730 _track("remaster", "Penny Lane", "The Beatles", album_year=1967),
731 "ISRC-REMASTER",
732 )
733 quiz, mass = _quiz([reissue, remaster])
734 _with_musicbrainz(mass, {"ISRC-REISSUE": 1970, "ISRC-REMASTER": 2017})
735
736 await quiz.initialize()
737 game_round = await _prepare_round_with_tracks(quiz, [reissue, remaster])
738
739 assert isinstance(game_round.answer_state, TimelineRoundState)
740 assert game_round.answer_state.placement_snapshot[0].release_year == 1970
741 assert game_round.answer_state.candidate.entry.release_year == 1967
742
743
744@pytest.mark.asyncio
745async def test_an_earlier_track_release_date_survives_a_later_musicbrainz_year() -> None:
746 """Never push a release date the track already carries forward to a later year."""
747 dated = _with_isrc(
748 _track("dated", "Suspicious Minds", "Elvis Presley", release_year=1969),
749 "ISRC-DATED",
750 )
751 other = _track("other", "Genesis", "Justice", album_year=2007)
752 quiz, mass = _quiz([dated, other])
753 _with_musicbrainz(mass, {"ISRC-DATED": 1990})
754
755 await quiz.initialize()
756 game_round = await _prepare_round_with_tracks(quiz, [dated, other])
757
758 assert isinstance(game_round.answer_state, TimelineRoundState)
759 assert game_round.answer_state.placement_snapshot[0].release_year == 1969
760
761
762@pytest.mark.asyncio
763async def test_musicbrainz_lookups_do_not_scale_with_the_source_pool() -> None:
764 """Date only the tracks that reach the timeline, never the complete eligible pool."""
765 tracks = [
766 _with_isrc(
767 _track(f"track-{index}", f"Song {index}", f"Artist {index}", album_year=1990),
768 f"ISRC-{index}",
769 )
770 for index in range(60)
771 ]
772 quiz, mass = _quiz(tracks)
773 musicbrainz = _with_musicbrainz(mass, {f"ISRC-{index}": 1970 for index in range(60)})
774
775 await quiz.initialize()
776 game_round = await quiz.prepare_round(0, [])
777
778 assert isinstance(game_round.answer_state, TimelineRoundState)
779 assert game_round.answer_state.candidate.entry.release_year == 1970
780 assert musicbrainz.get_release_year_by_isrc.await_count <= quiz.config.round_count + 1
781 # the name search runs alongside the ISRC lookup, so it is bound to the same entries
782 assert musicbrainz.get_release_year_by_track_name.await_count <= quiz.config.round_count + 1
783
784
785@pytest.mark.asyncio
786async def test_musicbrainz_dates_a_track_without_an_isrc_by_name() -> None:
787 """Date tracks from providers that never hand out an ISRC by artist and title."""
788 # a greatest hits set typed as a regular album keeps its (wrong) year without this
789 hits = _track("hits", "Radio Ga Ga", "Queen", album_year=1991)
790 other = _track("other", "Genesis", "Justice", album_year=2007)
791 quiz, mass = _quiz([hits, other])
792 _with_musicbrainz(mass, {}, name_years={("Queen", "Radio Ga Ga"): 1984})
793
794 await quiz.initialize()
795 game_round = await _prepare_round_with_tracks(quiz, [hits, other])
796
797 assert isinstance(game_round.answer_state, TimelineRoundState)
798 assert game_round.answer_state.placement_snapshot[0].release_year == 1984
799
800
801@pytest.mark.asyncio
802async def test_musicbrainz_name_lookup_never_pushes_a_track_to_a_later_year() -> None:
803 """Never let a name lookup overwrite a release date the track already carries."""
804 # the track carries the year itself, so only the "already earlier" guard can keep it
805 dated = _track("dated", "Africa", "Toto", release_year=1982)
806 quiz, mass = _quiz([dated])
807 _with_musicbrainz(mass, {}, name_years={("Toto", "Africa"): 2005})
808
809 dated_track, release_year = await quiz._musicbrainz_dated_track(dated)
810
811 assert release_year == 2005
812 assert MusicTimelineQuizType._release_year(dated_track) == 1982
813
814
815@pytest.mark.asyncio
816async def test_musicbrainz_dates_a_remastered_recording_on_the_name_lookup() -> None:
817 """Date a song by name when its ISRC points at a remaster released after the song."""
818 remastered = _with_isrc(
819 _track("remastered", "Africa", "Toto", album_year=1998), "ISRC-REMASTER"
820 )
821 quiz, mass = _quiz([remastered])
822 musicbrainz = _with_musicbrainz(
823 mass,
824 {"ISRC-REMASTER": 2005},
825 name_years={("Toto", "Africa"): 1982},
826 )
827
828 _, release_year = await quiz._musicbrainz_dated_track(remastered)
829
830 assert release_year == 1982
831 musicbrainz.get_release_year_by_track_name.assert_awaited_once_with("Toto", "Africa")
832
833
834@pytest.mark.asyncio
835async def test_musicbrainz_keeps_the_older_of_the_two_lookups() -> None:
836 """Keep the ISRC year when the name lookup dates the song later than the recording."""
837 late = _with_isrc(_track("late", "Africa", "Toto", album_year=1998), "ISRC-LATE")
838 quiz, mass = _quiz([late])
839 _with_musicbrainz(mass, {"ISRC-LATE": 1999}, name_years={("Toto", "Africa"): 2012})
840
841 _, release_year = await quiz._musicbrainz_dated_track(late)
842
843 assert release_year == 1999
844
845
846@pytest.mark.asyncio
847async def test_musicbrainz_keeps_the_isrc_year_when_the_name_lookup_finds_nothing() -> None:
848 """Never drop a year the ISRC supplied because the name search came back empty."""
849 late = _with_isrc(_track("late", "Africa", "Toto", album_year=1998), "ISRC-LATE")
850 quiz, mass = _quiz([late])
851 musicbrainz = _with_musicbrainz(mass, {"ISRC-LATE": 2005})
852
853 _, release_year = await quiz._musicbrainz_dated_track(late)
854
855 assert release_year == 2005
856 musicbrainz.get_release_year_by_track_name.assert_awaited_once_with("Toto", "Africa")
857
858
859@pytest.mark.asyncio
860async def test_musicbrainz_keeps_the_isrc_year_when_the_name_lookup_fails() -> None:
861 """Never drop a year the ISRC supplied because the name search errored."""
862 late = _with_isrc(_track("late", "Africa", "Toto", album_year=1998), "ISRC-LATE")
863 quiz, mass = _quiz([late])
864 musicbrainz = _with_musicbrainz(mass, {"ISRC-LATE": 2005})
865 musicbrainz.get_release_year_by_track_name.side_effect = RateLimited("rate limited")
866
867 _, release_year = await quiz._musicbrainz_dated_track(late)
868
869 assert release_year == 2005
870
871
872@pytest.mark.asyncio
873async def test_musicbrainz_keeps_the_isrc_year_when_the_name_lookup_never_answers() -> None:
874 """Never drop a year the ISRC supplied because the name search outlasted the budget."""
875
876 async def _never_answer(*_args: object) -> int:
877 await asyncio.Event().wait()
878 return 0
879
880 late = _with_isrc(_track("late", "Africa", "Toto", album_year=1998), "ISRC-LATE")
881 quiz, mass = _quiz([late])
882 musicbrainz = _with_musicbrainz(mass, {"ISRC-LATE": 2005})
883 musicbrainz.get_release_year_by_track_name.side_effect = _never_answer
884
885 with patch(
886 "music_assistant.providers.music_quiz.quiz_types.base.RELEASE_YEAR_LOOKUP_BUDGET_SECONDS",
887 0.01,
888 ):
889 _, release_year = await quiz._musicbrainz_dated_track(late)
890
891 assert release_year == 2005
892
893
894@pytest.mark.asyncio
895async def test_rescuing_undated_tracks_spends_one_request_per_track() -> None:
896 """Keep the shared rescue budget within the MusicBrainz request allowance."""
897 candidates = [
898 _with_isrc(_track(f"undated-{index}", f"Song {index}", f"Artist {index}"), f"ISRC-{index}")
899 for index in range(5)
900 ]
901 quiz, mass = _quiz(candidates)
902 musicbrainz = _with_musicbrainz(
903 mass,
904 {f"ISRC-{index}": 1990 for index in range(5)},
905 name_years={(f"Artist {index}", f"Song {index}"): 1985 for index in range(5)},
906 )
907 eligible_tracks: dict[str, Track] = {}
908
909 await quiz._rescue_undated_tracks(eligible_tracks, candidates)
910
911 assert len(eligible_tracks) == 5
912 assert musicbrainz.get_release_year_by_isrc.await_count == 5
913 musicbrainz.get_release_year_by_track_name.assert_not_awaited()
914
915
916@pytest.mark.asyncio
917async def test_rescuing_undated_tracks_still_dates_them_without_an_isrc() -> None:
918 """Rescue a track by name when it carries no ISRC, on the one lookup it is allowed."""
919 candidate = _track("undated", "Africa", "Toto")
920 quiz, mass = _quiz([candidate])
921 musicbrainz = _with_musicbrainz(mass, {}, name_years={("Toto", "Africa"): 1982})
922 eligible_tracks: dict[str, Track] = {}
923
924 await quiz._rescue_undated_tracks(eligible_tracks, [candidate])
925
926 assert MusicTimelineQuizType._release_year(eligible_tracks[str(candidate.uri)]) == 1982
927 musicbrainz.get_release_year_by_track_name.assert_awaited_once_with("Toto", "Africa")
928
929
930@pytest.mark.asyncio
931async def test_rescuing_undated_tracks_falls_back_to_the_name_for_an_unknown_isrc() -> None:
932 """Rescue a track by name when MusicBrainz has no recording for the ISRC it carries."""
933 candidate = _with_isrc(_track("undated", "Africa", "Toto"), "ISRC-UNKNOWN")
934 quiz, mass = _quiz([candidate])
935 musicbrainz = _with_musicbrainz(mass, {}, name_years={("Toto", "Africa"): 1982})
936 eligible_tracks: dict[str, Track] = {}
937
938 await quiz._rescue_undated_tracks(eligible_tracks, [candidate])
939
940 assert MusicTimelineQuizType._release_year(eligible_tracks[str(candidate.uri)]) == 1982
941 musicbrainz.get_release_year_by_track_name.assert_awaited_once_with("Toto", "Africa")
942
943
944@pytest.mark.asyncio
945async def test_an_implausible_year_does_not_discard_the_other_lookup() -> None:
946 """Judge each MusicBrainz answer on its own so one unusable year cannot annul the other."""
947 track = _with_isrc(_track("track", "Africa", "Toto", album_year=1998), "ISRC-TRACK")
948 quiz, mass = _quiz([track])
949 _with_musicbrainz(mass, {"ISRC-TRACK": 1982}, name_years={("Toto", "Africa"): 999})
950
951 _, release_year = await quiz._musicbrainz_dated_track(track)
952
953 assert release_year == 1982
954
955
956@pytest.mark.asyncio
957async def test_rescued_tracks_still_reach_the_name_lookup_as_timeline_entries() -> None:
958 """Cross-check a rescued track once it becomes an entry and has a budget of its own."""
959 compilation = _with_isrc(_track("compilation", "Africa", "Toto"), "ISRC-COMPILATION")
960 compilation.album = _full_album(
961 "album-compilation",
962 "Party Hits",
963 album_type=AlbumType.COMPILATION,
964 year=1998,
965 )
966 quiz, mass = _quiz([compilation])
967 _with_musicbrainz(
968 mass,
969 {"ISRC-COMPILATION": 2005},
970 name_years={("Toto", "Africa"): 1982},
971 )
972 eligible_tracks: dict[str, Track] = {}
973 await quiz._rescue_undated_tracks(eligible_tracks, [compilation])
974
975 entry = await quiz._create_entry(next(iter(eligible_tracks.values())))
976
977 assert entry.release_year == 1982
978
979
980@pytest.mark.asyncio
981async def test_musicbrainz_falls_back_to_the_name_when_it_does_not_know_the_isrc() -> None:
982 """Date a track by name when MusicBrainz has no recording for its ISRC."""
983 unknown = _with_isrc(_track("unknown", "Africa", "Toto", album_year=1998), "ISRC-UNKNOWN")
984 quiz, mass = _quiz([unknown])
985 musicbrainz = _with_musicbrainz(mass, {}, name_years={("Toto", "Africa"): 1982})
986
987 _, release_year = await quiz._musicbrainz_dated_track(unknown)
988
989 assert release_year == 1982
990 musicbrainz.get_release_year_by_isrc.assert_awaited_once_with("ISRC-UNKNOWN")
991 musicbrainz.get_release_year_by_track_name.assert_awaited_once_with("Toto", "Africa")
992
993
994@pytest.mark.asyncio
995async def test_musicbrainz_skips_the_name_lookup_without_an_artist() -> None:
996 """Never search MusicBrainz for a track that carries no artist to search on."""
997 anonymous = _track("anonymous", "Untitled", "Nobody", album_year=1990)
998 anonymous.artists = UniqueList([])
999 quiz, mass = _quiz([anonymous])
1000 musicbrainz = _with_musicbrainz(mass, {})
1001
1002 _, release_year = await quiz._musicbrainz_dated_track(anonymous)
1003
1004 assert release_year is None
1005 musicbrainz.get_release_year_by_track_name.assert_not_awaited()
1006
1007
1008@pytest.mark.asyncio
1009async def test_musicbrainz_dates_tracks_dropped_as_compilations() -> None:
1010 """Restore a compilation track once MusicBrainz dates its recording."""
1011 compilation = _with_isrc(_track("compilation", "Africa", "Toto"), "ISRC-COMPILATION")
1012 compilation.album = _full_album(
1013 "album-compilation",
1014 "Party Hits",
1015 album_type=AlbumType.COMPILATION,
1016 year=1998,
1017 )
1018 resolved = _track("resolved", "Genesis", "Justice", album_year=2007)
1019 quiz, mass = _quiz([compilation, resolved])
1020 mass.music.tracks.get.return_value = compilation
1021 _with_musicbrainz(mass, {"ISRC-COMPILATION": 1982})
1022
1023 await quiz.initialize()
1024
1025 assert quiz._eligible_tracks is not None
1026 eligible_tracks = {track.item_id: track for track in quiz._eligible_tracks}
1027 assert set(eligible_tracks) == {"compilation", "resolved"}
1028 assert MusicTimelineQuizType._release_year(eligible_tracks["compilation"]) == 1982
1029
1030
1031@pytest.mark.asyncio
1032async def test_library_years_are_kept_without_musicbrainz() -> None:
1033 """Place tracks on the timeline when no MusicBrainz provider is configured."""
1034 tracks = [
1035 _with_isrc(_track("one", "Teardrop", "Massive Attack", album_year=1998), "ISRC-ONE"),
1036 _with_isrc(_track("two", "Genesis", "Justice", album_year=2007), "ISRC-TWO"),
1037 ]
1038 quiz, mass = _quiz(tracks)
1039 mass.get_provider = MagicMock(return_value=None)
1040
1041 await quiz.initialize()
1042 game_round = await _prepare_round_with_tracks(quiz, tracks)
1043
1044 assert isinstance(game_round.answer_state, TimelineRoundState)
1045 assert game_round.answer_state.placement_snapshot[0].release_year == 1998
1046 assert game_round.answer_state.candidate.entry.release_year == 2007
1047
1048
1049@pytest.mark.asyncio
1050async def test_tracks_without_an_isrc_are_not_looked_up() -> None:
1051 """Never ask MusicBrainz about a track that carries no ISRC."""
1052 tracks = [
1053 _track("one", "Teardrop", "Massive Attack", album_year=1998),
1054 _track("two", "Genesis", "Justice", album_year=2007),
1055 ]
1056 quiz, mass = _quiz(tracks)
1057 musicbrainz = _with_musicbrainz(mass, {"ISRC-ONE": 1970})
1058
1059 await quiz.initialize()
1060 game_round = await _prepare_round_with_tracks(quiz, tracks)
1061
1062 musicbrainz.get_release_year_by_isrc.assert_not_awaited()
1063 assert isinstance(game_round.answer_state, TimelineRoundState)
1064 assert game_round.answer_state.placement_snapshot[0].release_year == 1998
1065 assert game_round.answer_state.candidate.entry.release_year == 2007
1066
1067
1068@pytest.mark.asyncio
1069@pytest.mark.parametrize("release_year", [0, 999, 9999])
1070async def test_musicbrainz_years_outside_the_accepted_range_are_ignored(release_year: int) -> None:
1071 """Keep the library year when MusicBrainz reports a year the timeline cannot use."""
1072 tracks = [
1073 _with_isrc(_track("one", "Teardrop", "Massive Attack", album_year=1998), "ISRC-ONE"),
1074 _with_isrc(_track("two", "Genesis", "Justice", album_year=2007), "ISRC-TWO"),
1075 ]
1076 quiz, mass = _quiz(tracks)
1077 _with_musicbrainz(mass, {"ISRC-ONE": release_year, "ISRC-TWO": release_year})
1078
1079 await quiz.initialize()
1080 game_round = await _prepare_round_with_tracks(quiz, tracks)
1081
1082 assert isinstance(game_round.answer_state, TimelineRoundState)
1083 assert game_round.answer_state.placement_snapshot[0].release_year == 1998
1084 assert game_round.answer_state.candidate.entry.release_year == 2007
1085
1086
1087@pytest.mark.asyncio
1088async def test_failing_musicbrainz_lookups_keep_the_library_year() -> None:
1089 """Start the game on library years when the MusicBrainz lookup fails."""
1090 tracks = [
1091 _with_isrc(_track("one", "Teardrop", "Massive Attack", album_year=1998), "ISRC-ONE"),
1092 _with_isrc(_track("two", "Genesis", "Justice", album_year=2007), "ISRC-TWO"),
1093 ]
1094 quiz, mass = _quiz(tracks)
1095 _with_musicbrainz(mass, error=RuntimeError("musicbrainz unavailable"))
1096
1097 await quiz.initialize()
1098 game_round = await _prepare_round_with_tracks(quiz, tracks)
1099
1100 assert isinstance(game_round.answer_state, TimelineRoundState)
1101 assert game_round.answer_state.placement_snapshot[0].release_year == 1998
1102 assert game_round.answer_state.candidate.entry.release_year == 2007
1103
1104
1105@pytest.mark.asyncio
1106async def test_stalled_musicbrainz_lookups_keep_the_library_year() -> None:
1107 """Never let a MusicBrainz lookup that does not answer hold up a round."""
1108 tracks = [
1109 _with_isrc(_track("one", "Teardrop", "Massive Attack", album_year=1998), "ISRC-ONE"),
1110 _with_isrc(_track("two", "Genesis", "Justice", album_year=2007), "ISRC-TWO"),
1111 ]
1112 quiz, mass = _quiz(tracks)
1113 musicbrainz = _with_musicbrainz(mass)
1114
1115 async def _never_answer(_isrc: str) -> int:
1116 await asyncio.sleep(30)
1117 return 1970
1118
1119 musicbrainz.get_release_year_by_isrc.side_effect = _never_answer
1120
1121 await quiz.initialize()
1122 with patch(
1123 "music_assistant.providers.music_quiz.quiz_types.base.RELEASE_YEAR_LOOKUP_BUDGET_SECONDS",
1124 0.01,
1125 ):
1126 game_round = await _prepare_round_with_tracks(quiz, tracks)
1127
1128 assert isinstance(game_round.answer_state, TimelineRoundState)
1129 assert game_round.answer_state.placement_snapshot[0].release_year == 1998
1130 assert game_round.answer_state.candidate.entry.release_year == 2007
1131
1132
1133def test_release_year_uses_earliest_valid_track_or_album_year() -> None:
1134 """Resolve the earliest usable year from album and track metadata."""
1135 track_first = _track(
1136 "track-first",
1137 "Track First",
1138 "Artist",
1139 album_year=2005,
1140 release_year=1999,
1141 )
1142 full_album = Album(
1143 item_id="album",
1144 provider="prov",
1145 name="Full Album",
1146 year=2000,
1147 provider_mappings={
1148 ProviderMapping(
1149 item_id="album",
1150 provider_domain="prov",
1151 provider_instance="prov",
1152 )
1153 },
1154 )
1155 album_first = _track("album-first", "Album First", "Artist", release_year=2002)
1156 album_first.album = full_album
1157 album_only = _track("album-only", "Album Only", "Artist", album_year=2001)
1158 track_only = _track("track-only", "Track Only", "Artist", release_year=2003)
1159 future_album = _track(
1160 "future-album",
1161 "Future Album",
1162 "Artist",
1163 album_year=datetime.now(tz=UTC).year + 1,
1164 release_year=2004,
1165 )
1166 too_old_track = _track(
1167 "too-old-track",
1168 "Too Old Track",
1169 "Artist",
1170 album_year=2006,
1171 release_year=999,
1172 )
1173 invalid = _track(
1174 "invalid",
1175 "Invalid",
1176 "Artist",
1177 album_year=datetime.now(tz=UTC).year + 1,
1178 release_year=999,
1179 )
1180
1181 assert MusicTimelineQuizType._release_year(track_first) == 1999
1182 assert MusicTimelineQuizType._release_year(album_first) == 2000
1183 assert MusicTimelineQuizType._release_year(album_only) == 2001
1184 assert MusicTimelineQuizType._release_year(track_only) == 2003
1185 assert MusicTimelineQuizType._release_year(future_album) == 2004
1186 assert MusicTimelineQuizType._release_year(too_old_track) == 2006
1187 assert MusicTimelineQuizType._release_year(invalid) is None
1188
1189
1190@pytest.mark.parametrize("album_type", list(AlbumType), ids=lambda value: value.value)
1191def test_release_year_trusts_only_original_release_album_types(album_type: AlbumType) -> None:
1192 """Accept album years only from album types that date an original release."""
1193 track = _track("typed", "Typed Track", "Artist")
1194 track.album = _full_album("album", "Typed Album", album_type=album_type, year=1998)
1195 expected_year = None if album_type in UNTRUSTED_ALBUM_TYPES else 1998
1196
1197 assert MusicTimelineQuizType._release_year(track) == expected_year
1198
1199
1200@pytest.mark.parametrize(
1201 "album_artists",
1202 [
1203 [_album_artist("va-mbid", "Artistes divers", mbid=VARIOUS_ARTISTS_MBID)],
1204 [_album_artist("va-name", VARIOUS_ARTISTS_NAME)],
1205 [
1206 _album_artist("primary", "Primary Artist"),
1207 _album_artist("va-multiple", "VARIOUS-ARTISTS"),
1208 ],
1209 ],
1210 ids=["canonical-mbid", "canonical-name", "multiple-artists"],
1211)
1212def test_release_year_rejects_various_artists_album_credits(
1213 album_artists: list[Artist],
1214) -> None:
1215 """Reject the album year of a various-artists set typed as a regular album."""
1216 track = _track("various", "Various Track", "Artist")
1217 track.album = _full_album(
1218 "album",
1219 "Various Album",
1220 album_type=AlbumType.ALBUM,
1221 artists=album_artists,
1222 year=1998,
1223 )
1224
1225 assert MusicTimelineQuizType._release_year(track) is None
1226
1227
1228def test_release_year_trusts_an_album_mapping_without_compilation_evidence() -> None:
1229 """Accept the album year of a mapping that carries no album type or artists."""
1230 track = _track("mapping", "Mapped Track", "Artist", album_year=1998)
1231 track.album = ItemMapping(
1232 media_type=MediaType.ALBUM,
1233 item_id="album-mapping",
1234 provider="prov",
1235 name=VARIOUS_ARTISTS_NAME,
1236 year=1998,
1237 )
1238
1239 assert MusicTimelineQuizType._release_year(track) == 1998
1240
1241
1242def test_release_year_of_a_compilation_falls_back_to_the_track_year() -> None:
1243 """Read the track year when a compilation album supplies the only other year."""
1244 track = _track("compilation", "Compilation Track", "Artist", release_year=1962)
1245 track.album = _full_album(
1246 "album",
1247 "Greatest Hits",
1248 album_type=AlbumType.COMPILATION,
1249 year=1998,
1250 )
1251
1252 assert MusicTimelineQuizType._release_year(track) == 1962
1253
1254
1255@pytest.mark.asyncio
1256async def test_compilation_tracks_are_excluded_from_the_eligible_pool() -> None:
1257 """Build the timeline pool from original releases only."""
1258 dated = [
1259 _track(f"dated-{index}", f"Dated {index}", "Artist", album_year=1990 + index)
1260 for index in range(2)
1261 ]
1262 compilations = [
1263 _track(f"compilation-{index}", f"Compilation {index}", "Artist") for index in range(2)
1264 ]
1265 for index, track in enumerate(compilations):
1266 track.album = _full_album(
1267 f"album-{index}",
1268 f"Party Hits {index}",
1269 album_type=AlbumType.COMPILATION,
1270 year=2005 + index,
1271 )
1272 tracks = [*dated, *compilations]
1273 quiz, mass = _quiz(tracks, round_count=1)
1274 tracks_by_id = {track.item_id: track for track in tracks}
1275 mass.music.tracks.get.side_effect = lambda item_id, _provider: tracks_by_id[item_id]
1276
1277 await quiz.initialize()
1278
1279 assert quiz._eligible_tracks is not None
1280 assert [track.uri for track in quiz._eligible_tracks] == [track.uri for track in dated]
1281
1282
1283def test_reject_track_removes_it_from_music_timeline_caches() -> None:
1284 """Exclude failed playback tracks from future Music Timeline selection."""
1285 failed = _track("failed", "Unavailable", "Artist", album_year=2000)
1286 available = _track("available", "Playable", "Artist", album_year=2001)
1287 quiz, _ = _quiz([failed, available])
1288 assert failed.uri is not None
1289 assert available.uri is not None
1290 quiz._source_track_pool = {failed.uri: failed, available.uri: available}
1291 quiz._eligible_tracks = [failed, available]
1292 artist_key = ("prov", "artist")
1293 failed_candidate = SuggestionCandidate(failed.name, uri=failed.uri)
1294 available_candidate = SuggestionCandidate(available.name, uri=available.uri)
1295 quiz._title_top_track_candidates[artist_key] = [
1296 failed_candidate,
1297 available_candidate,
1298 ]
1299 quiz._title_search_candidates[artist_key] = [
1300 available_candidate,
1301 failed_candidate,
1302 ]
1303
1304 quiz.reject_track(failed.uri)
1305
1306 assert quiz._source_track_pool == {available.uri: available}
1307 assert quiz._eligible_tracks == [available]
1308 assert quiz._title_top_track_candidates[artist_key] == [available_candidate]
1309 assert quiz._title_search_candidates[artist_key] == [available_candidate]
1310
1311
1312@pytest.mark.asyncio
1313async def test_prepare_round_seeds_anchor_and_prefetches_guaranteed_future_snapshot() -> None:
1314 """Build the next snapshot with the current song before that song is revealed."""
1315 anchor = _track("anchor", "Teardrop", "Massive Attack", album_year=1998)
1316 first = _track("first", "Genesis", "Justice", album_year=2007)
1317 second = _track("second", "Lisztomania", "Phoenix", album_year=2009)
1318 quiz, _ = _quiz([anchor, first, second], round_count=2)
1319 quiz._eligible_tracks = [anchor, first, second]
1320
1321 with patch(
1322 "music_assistant.providers.music_quiz.quiz_types.music_timeline.secrets.choice",
1323 side_effect=[anchor, first],
1324 ):
1325 first_round = await quiz.prepare_round(0, [])
1326 reconnected_quiz, _ = _quiz([anchor, first, second], round_count=2)
1327 reconnected_quiz._eligible_tracks = [anchor, first, second]
1328 with patch(
1329 "music_assistant.providers.music_quiz.quiz_types.music_timeline.secrets.choice",
1330 return_value=second,
1331 ):
1332 second_round = await reconnected_quiz.prepare_round(1, [first_round])
1333
1334 assert isinstance(first_round.answer_state, TimelineRoundState)
1335 assert isinstance(second_round.answer_state, TimelineRoundState)
1336 anchor_entry = first_round.answer_state.placement_snapshot[0]
1337 assert anchor_entry.is_anchor is True
1338 assert anchor_entry.track_uri == anchor.uri
1339 assert first_round.track_uri == first.uri
1340 assert [entry.track_uri for entry in second_round.answer_state.placement_snapshot] == [
1341 anchor.uri,
1342 first.uri,
1343 ]
1344 assert second_round.track_uri == second.uri
1345 assert (
1346 len(
1347 {
1348 entry.entry_id
1349 for entry in [
1350 *second_round.answer_state.placement_snapshot,
1351 second_round.answer_state.candidate.entry,
1352 ]
1353 }
1354 )
1355 == 3
1356 )
1357
1358
1359@pytest.mark.asyncio
1360async def test_prepare_round_reserves_fresh_track_for_scored_timeline_round() -> None:
1361 """Use a recent track as the anchor so a fresh track remains for players to place."""
1362 recent_anchor = _track("recent-anchor", "Teardrop", "Massive Attack", album_year=1998)
1363 recent_other = _track("recent-other", "Genesis", "Justice", album_year=2007)
1364 fresh = _track("fresh", "Lisztomania", "Phoenix", album_year=2009)
1365 quiz, _ = _quiz([recent_anchor, recent_other, fresh])
1366 quiz._eligible_tracks = [recent_anchor, recent_other, fresh]
1367 assert recent_anchor.uri is not None
1368 assert recent_other.uri is not None
1369 quiz.add_recent_track_uris([recent_anchor.uri, recent_other.uri])
1370
1371 with patch(
1372 "music_assistant.providers.music_quiz.quiz_types.music_timeline.secrets.choice",
1373 side_effect=lambda candidates: candidates[0],
1374 ):
1375 game_round = await quiz.prepare_round(0, [])
1376
1377 assert isinstance(game_round.answer_state, TimelineRoundState)
1378 assert game_round.answer_state.placement_snapshot[0].track_uri == recent_anchor.uri
1379 assert game_round.track_uri == fresh.uri
1380
1381
1382@pytest.mark.asyncio
1383async def test_prepare_round_builds_free_text_and_opaque_multiple_choice_bonuses() -> None:
1384 """Build independent bonus modes with four opaque, distinct MC options."""
1385 tracks = [
1386 _track("anchor", "Teardrop", "Massive Attack", album_year=1998),
1387 _track("current", "Genesis", "Justice", album_year=2007),
1388 _track("one", "Lisztomania", "Phoenix", album_year=2009),
1389 _track("two", "Sexy Boy", "Air", album_year=1998),
1390 _track("three", "Glory Box", "Portishead", album_year=1994),
1391 ]
1392 quiz, mass = _quiz(
1393 tracks,
1394 artist_mode=TimelineBonusMode.FREE_TEXT,
1395 title_mode=TimelineBonusMode.MULTIPLE_CHOICE,
1396 )
1397 quiz._eligible_tracks = tracks
1398
1399 with patch(
1400 "music_assistant.providers.music_quiz.quiz_types.music_timeline.secrets.choice",
1401 side_effect=tracks[:2],
1402 ):
1403 game_round = await quiz.prepare_round(0, [])
1404
1405 assert isinstance(game_round.answer_state, TimelineRoundState)
1406 artist_definition, title_definition = game_round.answer_state.bonus_definitions
1407 assert isinstance(artist_definition, TimelineFreeTextBonusDefinition)
1408 assert artist_definition.bonus_type == TimelineBonusType.ARTIST
1409 assert game_round.answer_state.candidate.artist_answers == ["Justice"]
1410 assert game_round.answer_state.candidate.title_answers == ["Genesis"]
1411 assert isinstance(title_definition, TimelineMultipleChoiceBonusDefinition)
1412 assert title_definition.bonus_type == TimelineBonusType.TITLE
1413 assert len(title_definition.options) == 4
1414 assert sum(option.is_correct for option in title_definition.options) == 1
1415 assert len({option.option_id for option in title_definition.options}) == 4
1416 assert all("correct" not in option.option_id for option in title_definition.options)
1417 mass.music.artists.top_tracks.assert_awaited_once()
1418 mass.music.artists.tracks.assert_not_awaited()
1419 mass.music.search.assert_awaited_once()
1420
1421
1422@pytest.mark.asyncio
1423async def test_artist_bonus_prefers_similar_artists_and_excludes_aliases() -> None:
1424 """Build artist options from similar artists without accepted aliases."""
1425 current = _track("current", "Get Lucky", "Daft Punk", album_year=2013)
1426 current.artists[0].sort_name = "Punk, Daft"
1427 current.artists.append(_artist("pharrell", "Pharrell Williams"))
1428 fallback_tracks = [
1429 _track("anchor", "Teardrop", "Massive Attack", album_year=1998),
1430 _track("fallback", "Glory Box", "Portishead", album_year=1994),
1431 ]
1432 quiz, mass = _quiz(
1433 [current, *fallback_tracks],
1434 artist_mode=TimelineBonusMode.MULTIPLE_CHOICE,
1435 )
1436 quiz._eligible_tracks = [current, *fallback_tracks]
1437 mass.music.artists.similar_artists.return_value = [
1438 _artist("alias-sort", "Punk, Daft"),
1439 _artist("alias-feature", "Pharrell Williams"),
1440 _artist("justice", "Justice"),
1441 _artist("air", "Air"),
1442 _artist("phoenix", "Phoenix"),
1443 _artist("duplicate-air", "air"),
1444 ]
1445
1446 options = await quiz._create_bonus_options(current, TimelineBonusType.ARTIST)
1447
1448 assert {option.label for option in options if not option.is_correct} == {
1449 "Justice",
1450 "Air",
1451 "Phoenix",
1452 }
1453 assert len({option.option_id for option in options}) == DEFAULT_BONUS_OPTION_COUNT
1454 mass.music.artists.similar_artists.assert_awaited_once_with(
1455 item_id=current.artists[0].item_id,
1456 provider_instance_id_or_domain=current.artists[0].provider,
1457 limit=24,
1458 )
1459 mass.music.tracks.similar_tracks.assert_not_awaited()
1460 mass.get_providers_supporting_feature.assert_not_called()
1461
1462
1463@pytest.mark.asyncio
1464async def test_artist_bonus_uses_similar_tracks_before_source_fallback() -> None:
1465 """Supplement similar artists with related-track artists before source artists."""
1466 current = _track("current", "Genesis", "Justice", album_year=2007)
1467 source_fallback = [
1468 _track("fallback-one", "Teardrop", "Massive Attack", album_year=1998),
1469 _track("fallback-two", "Glory Box", "Portishead", album_year=1994),
1470 ]
1471 quiz, mass = _quiz(
1472 [current, *source_fallback],
1473 artist_mode=TimelineBonusMode.MULTIPLE_CHOICE,
1474 )
1475 quiz._eligible_tracks = [current, *source_fallback]
1476 mass.music.artists.similar_artists.return_value = [_artist("daft-punk", "Daft Punk")]
1477 mass.music.tracks.similar_tracks.return_value = [
1478 _track("similar-track", "Sexy Boy", "Air"),
1479 ]
1480
1481 options = await quiz._create_bonus_options(current, TimelineBonusType.ARTIST)
1482
1483 wrong_labels = {option.label for option in options if not option.is_correct}
1484 assert {"Daft Punk", "Air"} <= wrong_labels
1485 assert len(wrong_labels & {"Massive Attack", "Portishead"}) == 1
1486 mass.music.tracks.similar_tracks.assert_awaited_once()
1487
1488
1489@pytest.mark.asyncio
1490async def test_artist_bonus_supplements_pairwise_close_similar_artists() -> None:
1491 """Do not count mutually ambiguous artist names as distinct options."""
1492 current = _track("current", "Genesis", "Justice", album_year=2007)
1493 source_fallback = _track("fallback", "Glory Box", "Portishead", album_year=1994)
1494 quiz, mass = _quiz(
1495 [current, source_fallback],
1496 artist_mode=TimelineBonusMode.MULTIPLE_CHOICE,
1497 )
1498 quiz._eligible_tracks = [current, source_fallback]
1499 mass.music.artists.similar_artists.return_value = [
1500 _artist("massive-attack", "Massive Attack"),
1501 _artist("massive-attack-collective", "Massive Attack Collective"),
1502 _artist("daft-punk", "Daft Punk"),
1503 ]
1504 mass.music.tracks.similar_tracks.return_value = [
1505 _track("similar-track", "Sexy Boy", "Air"),
1506 ]
1507
1508 options = await quiz._create_bonus_options(current, TimelineBonusType.ARTIST)
1509
1510 wrong_labels = {option.label for option in options if not option.is_correct}
1511 assert "Air" in wrong_labels
1512 assert "Portishead" not in wrong_labels
1513 assert len(wrong_labels & {"Massive Attack", "Massive Attack Collective"}) == 1
1514 mass.music.tracks.similar_tracks.assert_awaited_once()
1515
1516
1517@pytest.mark.asyncio
1518async def test_artist_source_fallback_stops_fuzzy_filtering_after_three_options() -> None:
1519 """Bound pairwise filtering even when the source pool is large."""
1520 current = _track("current", "Genesis", "Justice", album_year=2007)
1521 fallback_tracks = [
1522 _track("one", "Teardrop", "Massive Attack", album_year=1998),
1523 _track("two", "Glory Box", "Portishead", album_year=1994),
1524 _track("three", "Roads", "Tricky", album_year=1995),
1525 *[
1526 _track(
1527 f"unused-{index}",
1528 f"Unused Track {index}",
1529 f"Unused Artist {index}",
1530 album_year=2000,
1531 )
1532 for index in range(1000)
1533 ],
1534 ]
1535 quiz, _ = _quiz(
1536 [current, *fallback_tracks],
1537 artist_mode=TimelineBonusMode.MULTIPLE_CHOICE,
1538 )
1539 quiz._eligible_tracks = [current, *fallback_tracks]
1540
1541 with patch(
1542 "music_assistant.providers.music_quiz.quiz_types.music_timeline."
1543 "answer_labels_are_too_close",
1544 wraps=answer_labels_are_too_close,
1545 ) as compare_labels:
1546 options = await quiz._create_bonus_options(current, TimelineBonusType.ARTIST)
1547
1548 assert {option.label for option in options if not option.is_correct} == {
1549 "Massive Attack",
1550 "Portishead",
1551 "Tricky",
1552 }
1553 assert compare_labels.call_count == 6
1554
1555
1556@pytest.mark.asyncio
1557async def test_title_bonus_prefers_same_artist_source_tracks_and_excludes_versions() -> None:
1558 """Use same-artist source titles while excluding the current song and close versions."""
1559 current = _track("current", "Genesis", "Justice", album_year=2007)
1560 same_artist_tracks = [
1561 _track("close", "Genesis (Remix)", "Justice", album_year=2008),
1562 _track("one", "D.A.N.C.E.", "Justice", album_year=2007),
1563 _track("two", "Phantom", "Justice", album_year=2007),
1564 _track("three", "Audio, Video, Disco", "Justice", album_year=2011),
1565 ]
1566 unrelated = _track("unrelated", "Teardrop", "Massive Attack", album_year=1998)
1567 quiz, mass = _quiz(
1568 [current, *same_artist_tracks, unrelated],
1569 title_mode=TimelineBonusMode.MULTIPLE_CHOICE,
1570 )
1571 quiz._eligible_tracks = [current, *same_artist_tracks, unrelated]
1572
1573 options = await quiz._create_bonus_options(current, TimelineBonusType.TITLE)
1574
1575 assert {option.label for option in options if not option.is_correct} == {
1576 "D.A.N.C.E.",
1577 "Phantom",
1578 "Audio, Video, Disco",
1579 }
1580 mass.music.artists.top_tracks.assert_not_awaited()
1581 mass.music.artists.tracks.assert_not_awaited()
1582 mass.music.search.assert_not_awaited()
1583 mass.get_providers_supporting_feature.assert_not_called()
1584
1585
1586@pytest.mark.asyncio
1587async def test_title_bonus_uses_filtered_top_tracks_before_unrelated_sources() -> None:
1588 """Supplement source titles with real primary-artist top tracks only."""
1589 current = _track("current", "Genesis", "Justice", album_year=2007)
1590 source_same_artist = _track("source-same", "D.A.N.C.E.", "Justice", album_year=2007)
1591 unrelated = _track("unrelated", "Teardrop", "Massive Attack", album_year=1998)
1592 secondary_artist = _track("secondary", "Waters of Nazareth", "Mr. Oizo")
1593 secondary_artist.artists.append(_artist("justice", "Justice"))
1594 quiz, mass = _quiz(
1595 [current, source_same_artist, unrelated],
1596 title_mode=TimelineBonusMode.MULTIPLE_CHOICE,
1597 )
1598 quiz._eligible_tracks = [current, source_same_artist, unrelated]
1599 mass.music.artists.top_tracks.return_value = [
1600 current,
1601 _track("close", "Genesis (Remix)", "Justice"),
1602 _track("duplicate", "D.A.N.C.E.", "Justice"),
1603 secondary_artist,
1604 _artist("not-a-track", "Not a track"),
1605 _track("catalog-one", "Phantom", "Justice"),
1606 _track("catalog-two", "Audio, Video, Disco", "Justice"),
1607 ]
1608
1609 options = await quiz._create_bonus_options(current, TimelineBonusType.TITLE)
1610
1611 assert {option.label for option in options if not option.is_correct} == {
1612 "D.A.N.C.E.",
1613 "Phantom",
1614 "Audio, Video, Disco",
1615 }
1616 assert "Teardrop" not in {option.label for option in options}
1617 mass.music.artists.top_tracks.assert_awaited_once_with(
1618 item_id=current.artists[0].item_id,
1619 provider_instance_id_or_domain=current.artists[0].provider,
1620 )
1621 mass.music.artists.tracks.assert_not_awaited()
1622 mass.music.search.assert_not_awaited()
1623
1624
1625@pytest.mark.asyncio
1626async def test_title_bonus_search_is_bounded_and_cached_per_artist() -> None:
1627 """Reuse one top-track and bounded search result for repeated artist rounds."""
1628 current = _track("current", "Genesis", "Justice", album_year=2007)
1629 other = _track("other", "Stress", "Justice", album_year=2007)
1630 other.artists = UniqueList([current.artists[0]])
1631 unrelated = _track("unrelated", "Teardrop", "Massive Attack", album_year=1998)
1632 quiz, mass = _quiz(
1633 [current, other, unrelated],
1634 title_mode=TimelineBonusMode.MULTIPLE_CHOICE,
1635 )
1636 quiz._eligible_tracks = [current, other, unrelated]
1637 mass.music.artists.top_tracks.return_value = [
1638 _track("top", "Phantom", "Justice"),
1639 ]
1640 mass.music.search.return_value = SimpleNamespace(
1641 tracks=[
1642 _track("wrong-artist", "Sexy Boy", "Air"),
1643 _track("search-one", "Audio, Video, Disco", "Justice"),
1644 _track("search-two", "Civilization", "Justice"),
1645 ]
1646 )
1647
1648 current_options = await quiz._create_bonus_options(current, TimelineBonusType.TITLE)
1649 other_options = await quiz._create_bonus_options(other, TimelineBonusType.TITLE)
1650
1651 assert "Teardrop" not in {option.label for option in current_options}
1652 assert "Teardrop" not in {option.label for option in other_options}
1653 assert "Sexy Boy" not in {option.label for option in current_options}
1654 assert "Sexy Boy" not in {option.label for option in other_options}
1655 mass.music.artists.top_tracks.assert_awaited_once()
1656 mass.music.search.assert_awaited_once_with(
1657 search_query="Justice",
1658 media_types=[MediaType.TRACK],
1659 limit=BONUS_CANDIDATE_LIMIT,
1660 library_only=False,
1661 )
1662 mass.music.artists.tracks.assert_not_awaited()
1663
1664
1665@pytest.mark.asyncio
1666async def test_title_bonus_uses_unrelated_source_only_when_grounded_pool_is_sparse() -> None:
1667 """Use unrelated source titles only after bounded same-artist sources run out."""
1668 current = _track("current", "Genesis", "Justice", album_year=2007)
1669 source_same_artist = _track("source-same", "D.A.N.C.E.", "Justice", album_year=2007)
1670 unrelated = _track("unrelated", "Teardrop", "Massive Attack", album_year=1998)
1671 unused_fallback = _track("other", "Glory Box", "Portishead", album_year=1994)
1672 quiz, mass = _quiz(
1673 [current, source_same_artist, unrelated, unused_fallback],
1674 title_mode=TimelineBonusMode.MULTIPLE_CHOICE,
1675 )
1676 quiz._eligible_tracks = [
1677 current,
1678 source_same_artist,
1679 unrelated,
1680 unused_fallback,
1681 ]
1682 mass.music.artists.top_tracks.return_value = [
1683 _track("top", "Phantom", "Justice"),
1684 ]
1685
1686 options = await quiz._create_bonus_options(current, TimelineBonusType.TITLE)
1687
1688 assert {option.label for option in options if not option.is_correct} == {
1689 "D.A.N.C.E.",
1690 "Phantom",
1691 "Teardrop",
1692 }
1693 mass.music.artists.top_tracks.assert_awaited_once()
1694 mass.music.search.assert_awaited_once()
1695 mass.music.artists.tracks.assert_not_awaited()
1696
1697
1698@pytest.mark.asyncio
1699async def test_title_bonus_uses_same_artist_source_tracks_outside_playback_set() -> None:
1700 """Ground same-artist titles in the complete source pool without extra lookups."""
1701 current = _track("current", "Genesis", "Justice", album_year=2007)
1702 undated_same_artist = [
1703 _track("one", "D.A.N.C.E.", "Justice"),
1704 _track("two", "Phantom", "Justice"),
1705 _track("three", "Audio, Video, Disco", "Justice"),
1706 ]
1707 unrelated = _track("unrelated", "Teardrop", "Massive Attack", album_year=1998)
1708 quiz, mass = _quiz(
1709 [current, *undated_same_artist, unrelated],
1710 title_mode=TimelineBonusMode.MULTIPLE_CHOICE,
1711 )
1712 quiz._eligible_tracks = [current, unrelated]
1713
1714 with patch.object(
1715 quiz,
1716 "_source_pool_bonus_distractors",
1717 wraps=quiz._source_pool_bonus_distractors,
1718 ) as source_pool_fallback:
1719 options = await quiz._create_bonus_options(current, TimelineBonusType.TITLE)
1720
1721 assert {option.label for option in options if not option.is_correct} == {
1722 "D.A.N.C.E.",
1723 "Phantom",
1724 "Audio, Video, Disco",
1725 }
1726 source_pool_fallback.assert_not_awaited()
1727 mass.music.artists.top_tracks.assert_not_awaited()
1728 mass.music.search.assert_not_awaited()
1729
1730
1731@pytest.mark.asyncio
1732async def test_bonus_fallback_uses_source_tracks_outside_playback_set() -> None:
1733 """Fill the last-resort bonus options from source tracks the timeline cannot use."""
1734 current = _track("current", "Genesis", "Justice", album_year=2007)
1735 undated = [
1736 _track("one", "Teardrop", "Massive Attack"),
1737 _track("two", "Glory Box", "Portishead"),
1738 _track("three", "Roads", "Tricky"),
1739 ]
1740 quiz, _ = _quiz(
1741 [current, *undated],
1742 artist_mode=TimelineBonusMode.MULTIPLE_CHOICE,
1743 title_mode=TimelineBonusMode.MULTIPLE_CHOICE,
1744 )
1745 quiz._eligible_tracks = [current]
1746
1747 artist_options = await quiz._create_bonus_options(current, TimelineBonusType.ARTIST)
1748 title_options = await quiz._create_bonus_options(current, TimelineBonusType.TITLE)
1749
1750 assert {option.label for option in artist_options if not option.is_correct} == {
1751 "Massive Attack",
1752 "Portishead",
1753 "Tricky",
1754 }
1755 assert {option.label for option in title_options if not option.is_correct} == {
1756 "Teardrop",
1757 "Glory Box",
1758 "Roads",
1759 }
1760
1761
1762@pytest.mark.asyncio
1763async def test_rejected_track_is_excluded_from_bonus_options() -> None:
1764 """Never reuse a rejected track as a bonus label while the pool can still fill options."""
1765 current = _track("current", "Genesis", "Justice", album_year=2007)
1766 failed = _track("failed", "Teardrop", "Massive Attack", album_year=1998)
1767 remaining = [
1768 _track("one", "Glory Box", "Portishead"),
1769 _track("two", "Roads", "Tricky"),
1770 _track("three", "Sexy Boy", "Air"),
1771 ]
1772 quiz, _ = _quiz(
1773 [current, failed, *remaining],
1774 title_mode=TimelineBonusMode.MULTIPLE_CHOICE,
1775 )
1776 quiz._eligible_tracks = [current, failed]
1777 assert failed.uri is not None
1778
1779 quiz.reject_track(failed.uri)
1780 options = await quiz._create_bonus_options(current, TimelineBonusType.TITLE)
1781
1782 assert {option.label for option in options if not option.is_correct} == {
1783 "Glory Box",
1784 "Roads",
1785 "Sexy Boy",
1786 }
1787
1788
1789@pytest.mark.asyncio
1790async def test_ai_artist_bonus_keeps_real_candidates_dominant() -> None:
1791 """Use two ranked similar artists and at most one synthetic artist."""
1792 current = _track("current", "Genesis", "Justice", album_year=2007)
1793 source_fallback = _track("fallback", "Glory Box", "Portishead", album_year=1994)
1794 quiz, mass = _quiz(
1795 [current, source_fallback],
1796 artist_mode=TimelineBonusMode.MULTIPLE_CHOICE,
1797 use_ai=True,
1798 )
1799 quiz._eligible_tracks = [current, source_fallback]
1800 mass.music.artists.similar_artists.return_value = [
1801 _artist("massive-attack", "Massive Attack"),
1802 _artist("daft-punk", "Daft Punk"),
1803 _artist("air", "Air"),
1804 _artist("phoenix", "Phoenix"),
1805 ]
1806 provider = _ai_provider(
1807 _ai_response(
1808 ["candidate_2", "candidate_1", "candidate_0", "candidate_3"],
1809 [("artist", "Neon Assembly")],
1810 )
1811 )
1812 mass.get_providers_supporting_feature.return_value = [provider]
1813
1814 options = await quiz._create_bonus_options(current, TimelineBonusType.ARTIST)
1815
1816 assert {option.label for option in options if not option.is_correct} == {
1817 "Air",
1818 "Daft Punk",
1819 "Neon Assembly",
1820 }
1821 assert "Portishead" not in {option.label for option in options}
1822 mass.music.tracks.similar_tracks.assert_not_awaited()
1823 provider.ai_query.assert_awaited_once()
1824 prompt = provider.ai_query.await_args.args[0]
1825 assert "Justice" in prompt
1826 assert "Massive Attack" in prompt
1827 assert "Portishead" not in prompt
1828
1829
1830@pytest.mark.asyncio
1831async def test_ai_artist_bonus_fills_only_real_candidate_shortfall() -> None:
1832 """Retain every available real artist and synthesize only missing options."""
1833 current = _track("current", "Genesis", "Justice", album_year=2007)
1834 quiz, mass = _quiz(
1835 [current],
1836 artist_mode=TimelineBonusMode.MULTIPLE_CHOICE,
1837 use_ai=True,
1838 )
1839 quiz._eligible_tracks = [current]
1840 mass.music.artists.similar_artists.return_value = [_artist("daft-punk", "Daft Punk")]
1841 provider = _ai_provider(
1842 _ai_response(
1843 ["candidate_0"],
1844 [
1845 ("artist", "Neon Assembly"),
1846 ("artist", "Lunar Circuit"),
1847 ],
1848 )
1849 )
1850 mass.get_providers_supporting_feature.return_value = [provider]
1851
1852 options = await quiz._create_bonus_options(current, TimelineBonusType.ARTIST)
1853
1854 assert {option.label for option in options if not option.is_correct} == {
1855 "Daft Punk",
1856 "Neon Assembly",
1857 "Lunar Circuit",
1858 }
1859 provider.ai_query.assert_awaited_once()
1860
1861
1862@pytest.mark.asyncio
1863async def test_ai_artist_bonus_uses_synthetic_fill_before_source_fallbacks() -> None:
1864 """Measure AI shortfall against similar artists rather than unrelated sources."""
1865 current = _track("current", "Genesis", "Justice", album_year=2007)
1866 source_fallback = [
1867 _track("fallback-one", "Teardrop", "Massive Attack", album_year=1998),
1868 _track("fallback-two", "Glory Box", "Portishead", album_year=1994),
1869 _track("fallback-three", "Roads", "Tricky", album_year=1995),
1870 ]
1871 quiz, mass = _quiz(
1872 [current, *source_fallback],
1873 artist_mode=TimelineBonusMode.MULTIPLE_CHOICE,
1874 use_ai=True,
1875 )
1876 quiz._eligible_tracks = [current, *source_fallback]
1877 mass.music.artists.similar_artists.return_value = [_artist("daft-punk", "Daft Punk")]
1878 provider = _ai_provider(
1879 _ai_response(
1880 ["candidate_0"],
1881 [
1882 ("artist", "Neon Assembly"),
1883 ("artist", "Lunar Circuit"),
1884 ],
1885 )
1886 )
1887 mass.get_providers_supporting_feature.return_value = [provider]
1888
1889 options = await quiz._create_bonus_options(current, TimelineBonusType.ARTIST)
1890
1891 wrong_labels = {option.label for option in options if not option.is_correct}
1892 assert wrong_labels == {"Daft Punk", "Neon Assembly", "Lunar Circuit"}
1893 prompt = provider.ai_query.await_args.args[0]
1894 assert "Daft Punk" in prompt
1895 assert "Massive Attack" not in prompt
1896 assert "Portishead" not in prompt
1897 assert "Tricky" not in prompt
1898
1899
1900@pytest.mark.asyncio
1901async def test_invalid_ai_fill_falls_back_to_real_source_candidates() -> None:
1902 """Keep catalog-only resilience when a sparse AI response is invalid."""
1903 current = _track("current", "Genesis", "Justice", album_year=2007)
1904 source_fallback = [
1905 _track("fallback-one", "Teardrop", "Massive Attack", album_year=1998),
1906 _track("fallback-two", "Glory Box", "Portishead", album_year=1994),
1907 _track("fallback-three", "Roads", "Tricky", album_year=1995),
1908 ]
1909 quiz, mass = _quiz(
1910 [current, *source_fallback],
1911 artist_mode=TimelineBonusMode.MULTIPLE_CHOICE,
1912 use_ai=True,
1913 )
1914 quiz._eligible_tracks = [current, *source_fallback]
1915 provider = _ai_provider(
1916 _ai_response(
1917 [],
1918 [
1919 ("artist", "Neon Assembly"),
1920 ("artist", "Lunar Circuit"),
1921 ("artist", "Chrome Pilots"),
1922 ],
1923 extra=True,
1924 )
1925 )
1926 mass.get_providers_supporting_feature.return_value = [provider]
1927
1928 options = await quiz._create_bonus_options(current, TimelineBonusType.ARTIST)
1929
1930 assert {option.label for option in options if not option.is_correct} == {
1931 "Massive Attack",
1932 "Portishead",
1933 "Tricky",
1934 }
1935 provider.ai_query.assert_awaited_once()
1936
1937
1938@pytest.mark.asyncio
1939async def test_ai_title_bonus_keeps_same_artist_catalog_candidates_dominant() -> None:
1940 """Use two real same-artist titles and at most one synthetic title."""
1941 current = _track("current", "Genesis", "Justice", album_year=2007)
1942 same_artist = [
1943 _track("one", "D.A.N.C.E.", "Justice", album_year=2007),
1944 _track("two", "Phantom", "Justice", album_year=2007),
1945 _track("three", "Audio, Video, Disco", "Justice", album_year=2011),
1946 ]
1947 quiz, mass = _quiz(
1948 [current, *same_artist],
1949 title_mode=TimelineBonusMode.MULTIPLE_CHOICE,
1950 use_ai=True,
1951 )
1952 quiz._eligible_tracks = [current, *same_artist]
1953 provider = _ai_provider(
1954 _ai_response(
1955 ["candidate_1", "candidate_0", "candidate_2"],
1956 [("title", "Chrome Cathedral")],
1957 )
1958 )
1959 mass.get_providers_supporting_feature.return_value = [provider]
1960
1961 options = await quiz._create_bonus_options(current, TimelineBonusType.TITLE)
1962
1963 assert {option.label for option in options if not option.is_correct} == {
1964 "Phantom",
1965 "D.A.N.C.E.",
1966 "Chrome Cathedral",
1967 }
1968 mass.music.artists.top_tracks.assert_not_awaited()
1969 mass.music.search.assert_not_awaited()
1970 provider.ai_query.assert_awaited_once()
1971
1972
1973@pytest.mark.asyncio
1974async def test_ai_title_bonus_fills_only_missing_same_artist_titles() -> None:
1975 """Fill a sparse grounded title pool without replacing its real candidate."""
1976 current = _track("current", "Genesis", "Justice", album_year=2007)
1977 quiz, mass = _quiz(
1978 [current],
1979 title_mode=TimelineBonusMode.MULTIPLE_CHOICE,
1980 use_ai=True,
1981 )
1982 quiz._eligible_tracks = [current]
1983 mass.music.artists.top_tracks.return_value = [
1984 _track("top", "Phantom", "Justice"),
1985 ]
1986 provider = _ai_provider(
1987 _ai_response(
1988 ["candidate_0"],
1989 [
1990 ("title", "Chrome Cathedral"),
1991 ("title", "Electric Pilgrims"),
1992 ],
1993 )
1994 )
1995 mass.get_providers_supporting_feature.return_value = [provider]
1996
1997 options = await quiz._create_bonus_options(current, TimelineBonusType.TITLE)
1998
1999 assert {option.label for option in options if not option.is_correct} == {
2000 "Phantom",
2001 "Chrome Cathedral",
2002 "Electric Pilgrims",
2003 }
2004 mass.music.artists.top_tracks.assert_awaited_once()
2005 mass.music.search.assert_awaited_once()
2006 provider.ai_query.assert_awaited_once()
2007
2008
2009@pytest.mark.asyncio
2010@pytest.mark.parametrize(
2011 "synthetic",
2012 [
2013 [("artist", "Justice")],
2014 [("artist", "Justice!")],
2015 [("artist", "Fake Artist - Fake Song")],
2016 [("artist", "Fake Artist \u2013 Fake Song")],
2017 ],
2018)
2019async def test_invalid_ai_bonus_labels_fall_back_to_real_catalog(
2020 synthetic: list[tuple[str, str]],
2021) -> None:
2022 """Reject correct, near-correct, and malformed synthetic bonus labels."""
2023 current = _track("current", "Genesis", "Justice", album_year=2007)
2024 quiz, mass = _quiz(
2025 [current],
2026 artist_mode=TimelineBonusMode.MULTIPLE_CHOICE,
2027 use_ai=True,
2028 )
2029 quiz._eligible_tracks = [current]
2030 mass.music.artists.similar_artists.return_value = [
2031 _artist("daft-punk", "Daft Punk"),
2032 _artist("air", "Air"),
2033 _artist("phoenix", "Phoenix"),
2034 ]
2035 provider = _ai_provider(
2036 _ai_response(
2037 ["candidate_0", "candidate_1", "candidate_2"],
2038 synthetic,
2039 )
2040 )
2041 mass.get_providers_supporting_feature.return_value = [provider]
2042
2043 options = await quiz._create_bonus_options(current, TimelineBonusType.ARTIST)
2044
2045 assert {option.label for option in options if not option.is_correct} == {
2046 "Daft Punk",
2047 "Air",
2048 "Phoenix",
2049 }
2050 provider.ai_query.assert_awaited_once()
2051
2052
2053@pytest.mark.asyncio
2054async def test_invalid_primary_ai_provider_does_not_try_another_provider() -> None:
2055 """Make one deterministic AI request before falling back to real candidates."""
2056 current = _track("current", "Genesis", "Justice", album_year=2007)
2057 quiz, mass = _quiz(
2058 [current],
2059 artist_mode=TimelineBonusMode.MULTIPLE_CHOICE,
2060 use_ai=True,
2061 )
2062 quiz._eligible_tracks = [current]
2063 mass.music.artists.similar_artists.return_value = [
2064 _artist("daft-punk", "Daft Punk"),
2065 _artist("air", "Air"),
2066 _artist("phoenix", "Phoenix"),
2067 ]
2068 invalid = _ai_provider(
2069 _ai_response(
2070 ["candidate_0", "candidate_1", "candidate_2"],
2071 [("artist", "Neon Assembly")],
2072 extra=True,
2073 )
2074 )
2075 later = _ai_provider()
2076 later.instance_id = "ai--b"
2077 mass.get_providers_supporting_feature.return_value = [invalid, later]
2078
2079 options = await quiz._create_bonus_options(current, TimelineBonusType.ARTIST)
2080
2081 assert {option.label for option in options if not option.is_correct} == {
2082 "Daft Punk",
2083 "Air",
2084 "Phoenix",
2085 }
2086 invalid.ai_query.assert_awaited_once()
2087 later.ai_query.assert_not_awaited()
2088
2089
2090@pytest.mark.asyncio
2091async def test_ai_bonus_timeout_falls_back_to_real_catalog() -> None:
2092 """Use only catalog candidates when the single AI request times out."""
2093 current = _track("current", "Genesis", "Justice", album_year=2007)
2094 quiz, mass = _quiz(
2095 [current],
2096 artist_mode=TimelineBonusMode.MULTIPLE_CHOICE,
2097 use_ai=True,
2098 )
2099 quiz._eligible_tracks = [current]
2100 mass.music.artists.similar_artists.return_value = [
2101 _artist("daft-punk", "Daft Punk"),
2102 _artist("air", "Air"),
2103 _artist("phoenix", "Phoenix"),
2104 ]
2105 provider = _ai_provider()
2106
2107 async def _stall(_prompt: str) -> str:
2108 await asyncio.sleep(1)
2109 return ""
2110
2111 provider.ai_query.side_effect = _stall
2112 mass.get_providers_supporting_feature.return_value = [provider]
2113
2114 with patch(
2115 "music_assistant.providers.music_quiz.quiz_types.music_timeline.AI_QUERY_TIMEOUT_SECONDS",
2116 AI_QUERY_TIMEOUT_SECONDS / 30_000,
2117 ):
2118 options = await quiz._create_bonus_options(current, TimelineBonusType.ARTIST)
2119
2120 assert {option.label for option in options if not option.is_correct} == {
2121 "Daft Punk",
2122 "Air",
2123 "Phoenix",
2124 }
2125 provider.ai_query.assert_awaited_once()
2126
2127
2128@pytest.mark.asyncio
2129async def test_candidate_persists_display_artist_contributors_and_sort_aliases() -> None:
2130 """Persist justified artist truths while keeping one deterministic display value."""
2131 anchor = _track("anchor", "Teardrop", "Massive Attack", album_year=1998)
2132 current = _track("current", "Get Lucky", "Daft Punk", album_year=2013)
2133 current.artists[0].sort_name = "Punk, Daft"
2134 current.artists.append(
2135 ItemMapping(
2136 media_type=MediaType.ARTIST,
2137 item_id="artist-pharrell",
2138 provider="prov",
2139 name="Pharrell Williams",
2140 )
2141 )
2142 quiz, _ = _quiz([anchor, current])
2143 quiz._eligible_tracks = [anchor, current]
2144
2145 with patch(
2146 "music_assistant.providers.music_quiz.quiz_types.music_timeline.secrets.choice",
2147 side_effect=[anchor, current],
2148 ):
2149 game_round = await quiz.prepare_round(0, [])
2150
2151 assert isinstance(game_round.answer_state, TimelineRoundState)
2152 candidate = game_round.answer_state.candidate
2153 assert candidate.entry.artist == current.artist_str
2154 assert candidate.artist_answers == [
2155 current.artist_str,
2156 "Daft Punk",
2157 "Pharrell Williams",
2158 "Punk, Daft",
2159 ]
2160 assert candidate.title_answers == ["Get Lucky"]
2161