/
/
/
1"""Tests for the managed-pool refill allocator (player_queues/managed_pool.py)."""
2
3from __future__ import annotations
4
5import random
6from collections import Counter
7from itertools import groupby
8
9from music_assistant_models.enums import MediaType
10from music_assistant_models.media_items import ItemMapping, ProviderMapping, Track
11from music_assistant_models.unique_list import UniqueList
12
13from music_assistant.controllers.music.recency import RecencySnapshot, RecencyWindows, song_keys
14from music_assistant.controllers.player_queues.managed_pool import (
15 DynamicFillMode,
16 DynamicSource,
17 PoolWeightModel,
18 allocate_refill,
19 gate_tracks,
20)
21
22NOW = 1_000_000_000
23HOUR = 3600
24WEEK = 7 * 24 * HOUR
25GAP = 3 * HOUR
26
27
28def _track(item_id: str) -> Track:
29 """Build a Track on the 'test' provider with a single artist and provider mapping."""
30 return Track(
31 item_id=item_id,
32 provider="test",
33 name=f"Track {item_id}",
34 duration=60,
35 artists=UniqueList(
36 [ItemMapping(item_id="a", provider="test", name="A", media_type=MediaType.ARTIST)]
37 ),
38 provider_mappings={
39 ProviderMapping(item_id=item_id, provider_domain="test", provider_instance="test")
40 },
41 )
42
43
44def _artist_track(item_id: str, artist: str) -> Track:
45 """Build a Track on the 'test' provider with the given single named artist."""
46 return Track(
47 item_id=item_id,
48 provider="test",
49 name=f"Track {item_id}",
50 duration=60,
51 artists=UniqueList(
52 [
53 ItemMapping(
54 item_id=artist.lower(),
55 provider="test",
56 name=artist,
57 media_type=MediaType.ARTIST,
58 )
59 ]
60 ),
61 provider_mappings={
62 ProviderMapping(item_id=item_id, provider_domain="test", provider_instance="test")
63 },
64 )
65
66
67def _version_track(item_id: str, name: str, artist: str) -> Track:
68 """Build a Track with an explicit title and single named artist."""
69 return Track(
70 item_id=item_id,
71 provider="test",
72 name=name,
73 duration=60,
74 artists=UniqueList(
75 [
76 ItemMapping(
77 item_id=artist.lower(),
78 provider="test",
79 name=artist,
80 media_type=MediaType.ARTIST,
81 )
82 ]
83 ),
84 provider_mappings={
85 ProviderMapping(item_id=item_id, provider_domain="test", provider_instance="test")
86 },
87 )
88
89
90def _source(
91 candidate_ids: list[str],
92 *,
93 multiplicity: int = 1,
94 fill_mode: DynamicFillMode = DynamicFillMode.TRACKS,
95) -> DynamicSource:
96 """Build a DynamicSource with the given candidate track ids."""
97 return DynamicSource(
98 media_item=_track("seed"),
99 multiplicity=multiplicity,
100 fill_mode=fill_mode,
101 candidates=[_track(cid) for cid in candidate_ids],
102 )
103
104
105def _artist_source(
106 pairs: list[tuple[str, str]],
107 *,
108 multiplicity: int = 1,
109 fill_mode: DynamicFillMode = DynamicFillMode.TRACKS,
110) -> DynamicSource:
111 """Build a DynamicSource from (track_id, artist) pairs."""
112 return DynamicSource(
113 media_item=_track("seed"),
114 multiplicity=multiplicity,
115 fill_mode=fill_mode,
116 candidates=[_artist_track(item_id, artist) for item_id, artist in pairs],
117 )
118
119
120def _snapshot(
121 played: dict[str, int] | None = None, *, artists_played: dict[str, int] | None = None
122) -> RecencySnapshot:
123 """Build a snapshot marking the given track ids (and artist names) as played."""
124 return RecencySnapshot(
125 now=NOW,
126 song_ts={("test", item_id): ts for item_id, ts in (played or {}).items()},
127 artist_ts={name.lower(): ts for name, ts in (artists_played or {}).items()},
128 )
129
130
131def _artists(tracks: list[Track]) -> list[str]:
132 return [track.artists[0].name for track in tracks]
133
134
135def _ids(tracks: list[Track]) -> list[str]:
136 return [track.item_id for track in tracks]
137
138
139def test_empty_slots_returns_empty() -> None:
140 """Asking for zero (or fewer) slots returns nothing."""
141 source = _source(["a", "b"])
142 assert (
143 allocate_refill(
144 [source], slots=0, pool_keys=set(), snapshot=_snapshot(), windows=RecencyWindows()
145 )
146 == []
147 )
148
149
150def test_no_sources_returns_empty() -> None:
151 """No sources returns nothing."""
152 assert (
153 allocate_refill(
154 [], slots=5, pool_keys=set(), snapshot=_snapshot(), windows=RecencyWindows()
155 )
156 == []
157 )
158
159
160def test_per_base_quota_equal_share() -> None:
161 """Two equally-weighted sources split the slots evenly, regardless of catalogue size."""
162 sources = [
163 _source([f"a{i}" for i in range(20)]),
164 _source([f"b{i}" for i in range(20)]),
165 ]
166 result = allocate_refill(
167 sources, slots=10, pool_keys=set(), snapshot=_snapshot(), windows=RecencyWindows()
168 )
169 counts = Counter("a" if tid.startswith("a") else "b" for tid in _ids(result))
170 assert counts["a"] == 5
171 assert counts["b"] == 5
172
173
174def test_multiplicity_increases_share() -> None:
175 """A source added 3x gets ~3x the slots of one added once (per-base quota)."""
176 sources = [
177 _source([f"a{i}" for i in range(20)], multiplicity=1),
178 _source([f"b{i}" for i in range(20)], multiplicity=3),
179 ]
180 result = allocate_refill(
181 sources, slots=8, pool_keys=set(), snapshot=_snapshot(), windows=RecencyWindows()
182 )
183 counts = Counter("a" if tid.startswith("a") else "b" for tid in _ids(result))
184 assert counts["b"] == 6
185 assert counts["a"] == 2
186
187
188def test_weighted_sources_are_spread_across_batch() -> None:
189 """A higher-weight source is mixed through the batch instead of emitted as one block."""
190 sources = [
191 _source([f"a{i}" for i in range(20)]),
192 _source([f"b{i}" for i in range(20)], multiplicity=2),
193 _source([f"c{i}" for i in range(20)]),
194 _source([f"d{i}" for i in range(20)]),
195 ]
196 random.seed(0)
197 result = allocate_refill(
198 sources, slots=25, pool_keys=set(), snapshot=_snapshot(), windows=RecencyWindows()
199 )
200 source_ids = [track.item_id[0] for track in result]
201 longest_run = max(sum(1 for _ in run) for _, run in groupby(source_ids))
202 assert longest_run <= 2
203
204
205def test_size_multiplicity_weights_by_catalogue_size() -> None:
206 """Under SIZE_MULTIPLICITY, the larger-catalogue source dominates the pool."""
207 sources = [
208 _source(["a0", "a1"], multiplicity=1),
209 _source([f"b{i}" for i in range(20)], multiplicity=1),
210 ]
211 result = allocate_refill(
212 sources,
213 slots=11,
214 pool_keys=set(),
215 snapshot=_snapshot(),
216 windows=RecencyWindows(),
217 weight_model=PoolWeightModel.SIZE_MULTIPLICITY,
218 )
219 counts = Counter("a" if tid.startswith("a") else "b" for tid in _ids(result))
220 # weights 2 vs 20 -> b takes the lion's share; a is capped by its 2 candidates
221 assert counts["b"] > counts["a"]
222 assert counts["a"] <= 2
223
224
225def test_repeat_gap_hard_exclusion() -> None:
226 """A duplicated source's candidate played within the repeat-gap is excluded entirely."""
227 windows = RecencyWindows(song_seconds=WEEK, duplicate_gap_seconds=GAP)
228 sources = [_source(["hot", "cold"], multiplicity=2)]
229 snapshot = _snapshot({"hot": NOW - HOUR, "cold": NOW - 10 * HOUR})
230 result = allocate_refill(sources, slots=10, pool_keys=set(), snapshot=snapshot, windows=windows)
231 assert "hot" not in _ids(result)
232 assert "cold" in _ids(result)
233
234
235def test_singleton_window_vs_duplicate_gap() -> None:
236 """A track 5h old is buried as a singleton (week window) but fresh as a duplicate (3h gap)."""
237 windows = RecencyWindows(song_seconds=WEEK, duplicate_gap_seconds=GAP)
238 singleton = _source(["s"], multiplicity=1)
239 duplicate = _source(["d"], multiplicity=2)
240 snapshot = _snapshot({"s": NOW - 5 * HOUR, "d": NOW - 5 * HOUR})
241 result = allocate_refill(
242 [singleton, duplicate], slots=10, pool_keys=set(), snapshot=snapshot, windows=windows
243 )
244 assert "s" not in _ids(result)
245 assert "d" in _ids(result)
246
247
248def test_least_recently_played_first() -> None:
249 """A dynamic batch is ordered never-played first, then oldest play before most recent."""
250 windows = RecencyWindows(song_seconds=0) # gate off so we only test ordering
251 source = _source(["recent", "old", "never"], fill_mode=DynamicFillMode.DYNAMIC)
252 snapshot = _snapshot({"recent": NOW - 10, "old": NOW - 100_000})
253 result = allocate_refill([source], slots=3, pool_keys=set(), snapshot=snapshot, windows=windows)
254 assert _ids(result) == ["never", "old", "recent"]
255
256
257def test_tracks_mode_preserves_candidate_order() -> None:
258 """A finite (TRACKS) source keeps its materialized order instead of re-sorting by recency."""
259 windows = RecencyWindows(song_seconds=0) # gate off so we only test ordering
260 source = _source(["recent", "old", "never"], fill_mode=DynamicFillMode.TRACKS)
261 snapshot = _snapshot({"recent": NOW - 10, "old": NOW - 100_000})
262 result = allocate_refill([source], slots=3, pool_keys=set(), snapshot=snapshot, windows=windows)
263 assert _ids(result) == ["recent", "old", "never"]
264
265
266def test_pool_keys_excluded() -> None:
267 """A candidate already in the pool is never re-added."""
268 source = _source(["a", "b", "c"])
269 in_pool = _track("b")
270 result = allocate_refill(
271 [source], slots=10, pool_keys={in_pool}, snapshot=_snapshot(), windows=RecencyWindows()
272 )
273 assert "b" not in _ids(result)
274 assert set(_ids(result)) == {"a", "c"}
275
276
277def test_pool_song_keys_exclude_other_version() -> None:
278 """A different catalog version of an already-queued song is skipped too."""
279 queued = _version_track("amber-1", "Amber", "The Thrillseekers")
280 other_version = _version_track("amber-2", "Amber", "The Thrillseekers")
281 fresh = _version_track("other", "Two Bodies", "Flight Facilities")
282 source = DynamicSource(
283 media_item=_track("seed"),
284 multiplicity=1,
285 fill_mode=DynamicFillMode.TRACKS,
286 candidates=[other_version, fresh],
287 )
288 result = allocate_refill(
289 [source],
290 slots=10,
291 pool_keys={queued},
292 pool_song_keys=song_keys(queued),
293 snapshot=_snapshot(),
294 windows=RecencyWindows(),
295 )
296 assert _ids(result) == ["other"]
297
298
299def test_batch_never_contains_two_versions_of_same_song() -> None:
300 """Two catalog versions of the same song offered in one refill yield only one pick."""
301 sources = [
302 DynamicSource(
303 media_item=_track("seed"),
304 multiplicity=1,
305 fill_mode=DynamicFillMode.DYNAMIC,
306 candidates=[
307 _version_track("amber-1", "Amber", "The Thrillseekers"),
308 _version_track("amber-2", "Amber (Remastered 2019)", "The Thrillseekers"),
309 _version_track("other", "Two Bodies", "Flight Facilities"),
310 ],
311 )
312 ]
313 result = allocate_refill(
314 sources, slots=10, pool_keys=set(), snapshot=_snapshot(), windows=RecencyWindows()
315 )
316 assert len([tid for tid in _ids(result) if tid.startswith("amber")]) == 1
317 assert "other" in _ids(result)
318
319
320def test_never_exceeds_slots() -> None:
321 """The result never contains more than the requested number of slots."""
322 source = _source([f"a{i}" for i in range(50)])
323 result = allocate_refill(
324 [source], slots=7, pool_keys=set(), snapshot=_snapshot(), windows=RecencyWindows()
325 )
326 assert len(result) == 7
327
328
329def test_no_duplicates_across_sources() -> None:
330 """A track offered by two sources is added only once."""
331 shared = [f"x{i}" for i in range(10)]
332 sources = [_source(shared), _source(shared)]
333 result = allocate_refill(
334 sources, slots=10, pool_keys=set(), snapshot=_snapshot(), windows=RecencyWindows()
335 )
336 assert len(_ids(result)) == len(set(_ids(result)))
337
338
339def test_all_gated_falls_back_ungated() -> None:
340 """When every candidate is within the window, the ungated least-recently-played set is used."""
341 windows = RecencyWindows(song_seconds=WEEK)
342 source = _source(["a", "b", "c"])
343 snapshot = _snapshot({"a": NOW - HOUR, "b": NOW - 2 * HOUR, "c": NOW - 3 * HOUR})
344 result = allocate_refill([source], slots=2, pool_keys=set(), snapshot=snapshot, windows=windows)
345 # all are recent, but playback must not stall: the two least-recently-played come back
346 assert _ids(result) == ["c", "b"]
347
348
349def test_randomized_order_is_reproducible_under_seed() -> None:
350 """A fixed seed reproduces the interleave while a different seed varies it."""
351 sources = [
352 _source([f"a{i}" for i in range(10)], multiplicity=2),
353 _source([f"b{i}" for i in range(10)]),
354 ]
355 snapshot = _snapshot({"a3": NOW - 10, "b1": NOW - 20})
356 windows = RecencyWindows(song_seconds=WEEK, duplicate_gap_seconds=GAP)
357 random.seed(123)
358 first = _ids(
359 allocate_refill(sources, slots=6, pool_keys=set(), snapshot=snapshot, windows=windows)
360 )
361 random.seed(123)
362 second = _ids(
363 allocate_refill(sources, slots=6, pool_keys=set(), snapshot=snapshot, windows=windows)
364 )
365 random.seed(124)
366 third = _ids(
367 allocate_refill(sources, slots=6, pool_keys=set(), snapshot=snapshot, windows=windows)
368 )
369 assert first == second
370 assert first != third
371
372
373def test_gate_tracks_drops_recent() -> None:
374 """gate_tracks drops tracks played within the song window, keeping order."""
375 windows = RecencyWindows(song_seconds=WEEK)
376 tracks = [_track("a"), _track("b"), _track("c")]
377 snapshot = _snapshot({"b": NOW - HOUR})
378 assert _ids(gate_tracks(tracks, snapshot, windows)) == ["a", "c"]
379
380
381def test_gate_tracks_fallback_when_all_recent() -> None:
382 """gate_tracks returns the ungated list when every track is within the window."""
383 windows = RecencyWindows(song_seconds=WEEK)
384 tracks = [_track("a"), _track("b")]
385 snapshot = _snapshot({"a": NOW - HOUR, "b": NOW - 2 * HOUR})
386 assert _ids(gate_tracks(tracks, snapshot, windows)) == ["a", "b"]
387
388
389def test_spaces_adjacent_same_artist() -> None:
390 """The assembled batch never places two same-artist tracks directly adjacent."""
391 windows = RecencyWindows(song_seconds=0) # gate off; test ordering only
392 source = _artist_source(
393 [("a1", "A"), ("a2", "A"), ("a3", "A"), ("b1", "B"), ("c1", "C")],
394 )
395 result = allocate_refill(
396 [source], slots=5, pool_keys=set(), snapshot=_snapshot(), windows=windows
397 )
398 artists = _artists(result)
399 assert len(result) == 5 # spacing reorders, never drops
400 assert all(artists[i] != artists[i + 1] for i in range(len(artists) - 1))
401
402
403def test_seam_avoids_preceding_artist() -> None:
404 """The first added track is kept clear of the artist that plays right before the batch."""
405 windows = RecencyWindows(song_seconds=0)
406 source = _artist_source([("a1", "A"), ("b1", "B"), ("c1", "C")])
407 result = allocate_refill(
408 [source],
409 slots=3,
410 pool_keys=set(),
411 snapshot=_snapshot(),
412 windows=windows,
413 preceding_artists={"a"},
414 )
415 assert len(result) == 3
416 assert result[0].artists[0].name != "A"
417
418
419def test_artist_recency_deprioritized() -> None:
420 """A dynamic candidate whose artist is within the artist window sorts behind fresh ones."""
421 windows = RecencyWindows(song_seconds=0, artist_seconds=1800)
422 source = _artist_source([("r1", "Recent"), ("f1", "Fresh")], fill_mode=DynamicFillMode.DYNAMIC)
423 snapshot = _snapshot(artists_played={"Recent": NOW - 600})
424 result = allocate_refill([source], slots=2, pool_keys=set(), snapshot=snapshot, windows=windows)
425 # the fresh-artist track leads even though it appears later in the candidate list
426 assert _artists(result) == ["Fresh", "Recent"]
427
428
429def test_artist_recency_not_hard_excluded() -> None:
430 """A within-window artist is only nudged back, never dropped (single-artist stations still play)."""
431 windows = RecencyWindows(song_seconds=0, artist_seconds=1800)
432 source = _artist_source([("a1", "A"), ("a2", "A")], fill_mode=DynamicFillMode.DYNAMIC)
433 snapshot = _snapshot(artists_played={"A": NOW - 600})
434 result = allocate_refill([source], slots=5, pool_keys=set(), snapshot=snapshot, windows=windows)
435 assert len(result) == 2 # both kept despite the artist being recently heard
436