/
/
/
1"""Tests for the smart shuffle algorithm (player_queues/smart_shuffle.py)."""
2
3from __future__ import annotations
4
5import random
6
7from music_assistant_models.enums import MediaType
8from music_assistant_models.media_items import ItemMapping, ProviderMapping, Track
9from music_assistant_models.queue_item import QueueItem
10from music_assistant_models.unique_list import UniqueList
11
12from music_assistant.controllers.music.recency import RecencySnapshot, RecencyWindows
13from music_assistant.controllers.player_queues.smart_shuffle import _arrange
14
15NOW = 1_000_000_000
16WEEK = 7 * 24 * 3600
17GAP = 3 * 3600
18
19
20def _track(song_id: str, artist: str) -> Track:
21 """Build a library Track with a single artist and provider mapping."""
22 return Track(
23 item_id=song_id,
24 provider="library",
25 name=f"Song {song_id}",
26 duration=180,
27 artists=UniqueList(
28 [
29 ItemMapping(
30 item_id=artist.lower(),
31 provider="library",
32 name=artist,
33 media_type=MediaType.ARTIST,
34 )
35 ]
36 ),
37 provider_mappings={
38 ProviderMapping(item_id=song_id, provider_domain="library", provider_instance="library")
39 },
40 )
41
42
43def _item(song_id: str, *, artist: str | None = None, qiid: str | None = None) -> QueueItem:
44 """Build a QueueItem wrapping a library Track (copies share song_id, differ by qiid)."""
45 track = _track(song_id, artist or f"Artist {song_id}")
46 return QueueItem(
47 queue_id="q1",
48 queue_item_id=qiid or f"qi-{song_id}",
49 name=track.name,
50 duration=180,
51 media_item=track,
52 )
53
54
55def _snapshot(*, song_recent: tuple[str, ...] = (), played_at: int = NOW) -> RecencySnapshot:
56 """Build a snapshot marking the given song ids as played at played_at."""
57 return RecencySnapshot(now=NOW, song_ts={("library", sid): played_at for sid in song_recent})
58
59
60def _ids(items: list[QueueItem]) -> list[str]:
61 return [item.queue_item_id for item in items]
62
63
64def _song_id(item: QueueItem) -> str:
65 assert item.media_item is not None
66 return item.media_item.item_id
67
68
69def test_empty_returns_empty() -> None:
70 """An empty queue arranges to empty."""
71 assert _arrange([], _snapshot(), RecencyWindows()) == []
72
73
74def test_single_item_preserved() -> None:
75 """A single item is returned unchanged."""
76 items = [_item("a")]
77 assert _arrange(items, _snapshot(), RecencyWindows()) == items
78
79
80def test_preserves_all_items_across_seeds() -> None:
81 """Arranging never drops or duplicates items, regardless of the random seed."""
82 items = [_item(f"s{i}", artist=f"A{i}") for i in range(12)]
83 for seed in range(8):
84 random.seed(seed)
85 result = _arrange(list(items), _snapshot(), RecencyWindows())
86 assert sorted(_ids(result)) == sorted(_ids(items))
87
88
89def test_duplicates_spread_not_bursted() -> None:
90 """Copies of an intentionally-duplicated song spread across the queue, never adjacent."""
91 dup = [_item("x", artist="X", qiid=f"x{k}") for k in range(8)]
92 fillers = [_item(f"f{i}", artist=f"A{i}") for i in range(24)]
93 items = dup + fillers
94 for seed in range(8):
95 random.seed(seed)
96 result = _arrange(list(items), _snapshot(), RecencyWindows())
97 positions = [i for i, item in enumerate(result) if _song_id(item) == "x"]
98 # spread across most of the queue rather than clustered in a burst
99 assert positions[-1] - positions[0] >= len(items) // 2
100 # no two copies of the same song are adjacent
101 assert all(positions[i + 1] - positions[i] >= 2 for i in range(len(positions) - 1))
102
103
104def test_duplicate_rounds_have_independent_order() -> None:
105 """Equally frequent songs do not repeat the same sequence on every pass."""
106 song_ids = ("a", "b", "c", "d", "e")
107 items = [
108 _item(song_id, artist=song_id, qiid=f"{song_id}-{copy}")
109 for song_id in song_ids
110 for copy in range(4)
111 ]
112 random.seed(0)
113 result = _arrange(items, _snapshot(), RecencyWindows())
114 sequence = [_song_id(item) for item in result]
115 rounds = [
116 tuple(sequence[offset : offset + len(song_ids)])
117 for offset in range(0, len(sequence), len(song_ids))
118 ]
119 assert all(set(round_) == set(song_ids) for round_ in rounds)
120 assert len(set(rounds)) > 1
121
122
123def test_recent_singletons_pushed_to_back() -> None:
124 """Recently-played singletons sort behind fresh ones."""
125 recent = [_item(f"r{i}", artist=f"R{i}") for i in range(6)]
126 fresh = [_item(f"f{i}", artist=f"F{i}") for i in range(6)]
127 snapshot = _snapshot(song_recent=tuple(f"r{i}" for i in range(6)))
128 windows = RecencyWindows(song_seconds=WEEK, artist_seconds=0, duplicate_gap_seconds=GAP)
129 for seed in range(8):
130 random.seed(seed)
131 result = _arrange(recent + fresh, snapshot, windows)
132 index = {_song_id(item): i for i, item in enumerate(result)}
133 assert max(index[f"f{i}"] for i in range(6)) < min(index[f"r{i}"] for i in range(6))
134
135
136def test_duplicated_song_not_buried_like_singleton() -> None:
137 """A song heard 5h ago is buried as a singleton (week window) but fresh as a duplicate (3h gap)."""
138 snapshot = _snapshot(song_recent=("dup", "single"), played_at=NOW - 5 * 3600)
139 windows = RecencyWindows(song_seconds=WEEK, artist_seconds=0, duplicate_gap_seconds=GAP)
140 dup = [_item("dup", artist="D", qiid=f"d{k}") for k in range(4)]
141 single = _item("single", artist="S", qiid="single")
142 fresh = [_item(f"f{i}", artist=f"F{i}") for i in range(8)]
143 random.seed(1)
144 result = _arrange([*dup, single, *fresh], snapshot, windows)
145 index = {item.queue_item_id: i for i, item in enumerate(result)}
146 # the duplicate (fresh, tier 0) lands ahead of the buried singleton (tier 2)
147 assert max(index[f"d{k}"] for k in range(4)) < index["single"]
148
149
150def test_all_recent_never_drops() -> None:
151 """When every track is recently played, all items are still present (reordered, not dropped)."""
152 items = [_item(f"s{i}", artist=f"A{i}") for i in range(10)]
153 snapshot = _snapshot(song_recent=tuple(f"s{i}" for i in range(10)))
154 windows = RecencyWindows(song_seconds=WEEK)
155 random.seed(0)
156 result = _arrange(list(items), snapshot, windows)
157 assert sorted(_ids(result)) == sorted(_ids(items))
158
159
160def test_deterministic_under_seed() -> None:
161 """Same seed + same inputs yield an identical arrangement."""
162 items = [_item(f"s{i}", artist=f"A{i % 3}") for i in range(15)]
163 random.seed(123)
164 first = _ids(_arrange(list(items), _snapshot(), RecencyWindows()))
165 random.seed(123)
166 second = _ids(_arrange(list(items), _snapshot(), RecencyWindows()))
167 assert first == second
168