music-assistant-server

1"""
2End-to-end tests for the bounded managed radio pool through the real queue controller + DB.
3
4Boots a hermetic MusicAssistant (fake `test` music provider + demo players) and exercises:
5- a radio-flagged enqueue turning the whole queue into a small bounded pool (not the whole library);
6- a refill drawing from the queue's dynamic sources, weighted per source (multiplicity) and
7  hard-gated against the playlog recency windows.
8"""
9
10from __future__ import annotations
11
12import time
13from typing import TYPE_CHECKING, cast
14
15import pytest
16from music_assistant_models.enums import MediaType
17from music_assistant_models.media_items import Playlist
18from music_assistant_models.queue_item import QueueItem
19
20from music_assistant.constants import DB_TABLE_PLAYLOG
21from music_assistant.controllers.player_queues import managed_pool
22from music_assistant.controllers.player_queues.constants import (
23    MANAGED_POOL_MAX,
24    MANAGED_POOL_TARGET,
25)
26from music_assistant.mass import MusicAssistant
27from music_assistant.models.music_provider import MusicProvider
28from music_assistant.providers.radio_playlist import radio_playlist_uri
29
30from .conftest import demo_players, wait_for
31
32if TYPE_CHECKING:
33    from music_assistant_models.media_items import ItemMapping, MediaItemType
34
35HOUR = 3600
36DAY = 24 * HOUR
37TEST_USER = "e2e-user"
38
39
40async def _insert_play(mass: MusicAssistant, item_id: str, timestamp: int, userid: str) -> None:
41    """Insert a fully-played track row into the playlog for the given user."""
42    await mass.music.database.insert(
43        DB_TABLE_PLAYLOG,
44        {
45            "item_id": item_id,
46            "provider": "test",
47            "media_type": MediaType.TRACK.value,
48            "name": f"Track {item_id}",
49            "timestamp": timestamp,
50            "fully_played": True,
51            "seconds_played": 60,
52            "userid": userid,
53            "user_initiated": True,
54        },
55    )
56
57
58@pytest.mark.asyncio
59async def test_radio_enqueue_builds_bounded_pool(e2e_mass: MusicAssistant) -> None:
60    """A radio-flagged enqueue turns the queue into a small bounded pool, not the whole library."""
61    queue_id = demo_players(e2e_mass)[0].player_id
62    test_prov = cast("MusicProvider", e2e_mass.get_provider("test"))
63    assert test_prov is not None
64    seed = await test_prov.get_track("0_0_0")
65
66    await e2e_mass.player_queues.play_media(
67        queue_id, cast("MediaItemType | ItemMapping | str", seed), radio_mode=True
68    )
69
70    assert await wait_for(lambda: len(e2e_mass.player_queues.items(queue_id)) > 1), (
71        "radio pool never populated"
72    )
73    items = e2e_mass.player_queues.items(queue_id)
74    # bounded: a single-track radio yields ~target items, never the full 500-track library
75    assert 1 < len(items) <= MANAGED_POOL_MAX
76    queue = e2e_mass.player_queues.get(queue_id)
77    assert queue is not None
78    assert queue.sources  # the seed is kept as a dynamic source
79
80
81@pytest.mark.asyncio
82async def test_refill_gates_recent_and_weights_by_multiplicity(e2e_mass: MusicAssistant) -> None:
83    """A managed-pool refill excludes recently-played tracks and favours the higher-multiplicity source."""
84    queue_id = demo_players(e2e_mass)[0].player_id
85    test_prov = cast("MusicProvider", e2e_mass.get_provider("test"))
86    assert test_prov is not None
87    queue = e2e_mass.player_queues.get(queue_id)
88    assert queue is not None
89    e2e_mass.player_queues.queue_data(queue_id).userid = TEST_USER
90
91    # two albums (20 tracks each) as TRACKS sources; the second is added twice (multiplicity 2)
92    album_single = await test_prov.get_album("0_0")  # tracks 0_0_0 .. 0_0_19
93    album_double = await test_prov.get_album("1_0")  # tracks 1_0_0 .. 1_0_19
94    sources = cast("list[MediaItemType]", [album_single, album_double, album_double])
95    e2e_mass.player_queues.queue_data(queue_id).source_items = sources
96
97    # singleton album: 5 tracks played a day ago -> within the 1-week song window -> hard-gated
98    now = int(time.time())
99    gated = [f"0_0_{i}" for i in range(5)]
100    for item_id in gated:
101        await _insert_play(e2e_mass, item_id, now - DAY, TEST_USER)
102    # duplicated album: 5 tracks played 5h ago -> outside the 3h repeat-gap -> allowed
103    for i in range(5):
104        await _insert_play(e2e_mass, f"1_0_{i}", now - 5 * HOUR, TEST_USER)
105
106    pool = await e2e_mass.player_queues._managed_pool.fill(queue_id, is_initial=True)
107    ids = [track.item_id for track in pool]
108
109    assert 0 < len(pool) <= MANAGED_POOL_TARGET
110    # recency hard gate: the within-window singleton tracks are excluded entirely
111    assert not any(item_id in ids for item_id in gated)
112    # per-base quota: the album added twice contributes more than the one added once
113    single_count = sum(1 for tid in ids if tid.startswith("0_0_"))
114    double_count = sum(1 for tid in ids if tid.startswith("1_0_"))
115    assert double_count > single_count
116
117
118@pytest.mark.asyncio
119async def test_refill_dedupes_only_active_tail_not_played_history(e2e_mass: MusicAssistant) -> None:
120    """A track in the played history can return on refill; only the current+unplayed tail is deduped."""
121    queue_id = demo_players(e2e_mass)[0].player_id
122    test_prov = cast("MusicProvider", e2e_mass.get_provider("test"))
123    assert test_prov is not None
124    queue = e2e_mass.player_queues.get(queue_id)
125    assert queue is not None
126    e2e_mass.player_queues.queue_data(queue_id).userid = TEST_USER
127
128    album = await test_prov.get_album("0_0")  # tracks 0_0_0 .. 0_0_19
129    sources = cast("list[MediaItemType]", [album])
130    e2e_mass.player_queues.queue_data(queue_id).source_items = sources
131
132    # simulate a queue with played history: 0_0_0..0_0_4 already played, 0_0_5 is the current track
133    queue_items = [
134        QueueItem.from_media_item(queue_id, await test_prov.get_track(f"0_0_{i}")) for i in range(6)
135    ]
136    e2e_mass.player_queues.queue_data(queue_id).items = queue_items
137    queue.current_index = 5
138
139    pool = await e2e_mass.player_queues._managed_pool.fill(queue_id, is_initial=False)
140    ids = [track.item_id for track in pool]
141
142    # the active (current) track isn't duplicated into the upcoming pool
143    assert "0_0_5" not in ids
144    # but already-played history is not permanently excluded (no recency block in play here)
145    assert any(f"0_0_{i}" in ids for i in range(5))
146
147
148@pytest.mark.asyncio
149async def test_finite_source_materializes_and_plays_through_once(
150    e2e_mass: MusicAssistant, monkeypatch: pytest.MonkeyPatch
151) -> None:
152    """A finite source is materialized once and dequeued progressively until it is exhausted."""
153    # shrink the per-refill target so a single 20-track album is dequeued over several refills
154    monkeypatch.setattr(managed_pool, "MANAGED_POOL_TARGET", 5)
155    queue_id = demo_players(e2e_mass)[0].player_id
156    test_prov = cast("MusicProvider", e2e_mass.get_provider("test"))
157    assert test_prov is not None
158    queue = e2e_mass.player_queues.get(queue_id)
159    assert queue is not None
160    e2e_mass.player_queues.queue_data(queue_id).userid = TEST_USER
161
162    album = await test_prov.get_album("0_0")  # tracks 0_0_0 .. 0_0_19
163    e2e_mass.player_queues.queue_data(queue_id).source_items = cast("list[MediaItemType]", [album])
164
165    played: list[str] = []
166    fills = 0
167    # generous guard; a 20-track album drains in ~4 refills at the patched target of 5
168    while e2e_mass.player_queues.queue_data(queue_id).source_items and fills < 20:
169        pool = await e2e_mass.player_queues._managed_pool.fill(queue_id, is_initial=(fills == 0))
170        played += [track.item_id for track in pool]
171        fills += 1
172
173    # progressive: it took several refills, not one dump of the whole album
174    assert fills > 1
175    # plays through once: every album track exactly once, never recycled as tracks age out
176    assert sorted(played) == sorted(f"0_0_{i}" for i in range(20))
177    # exhausted: the source is retired from the queue and its materialized state is released
178    assert not e2e_mass.player_queues.queue_data(queue_id).source_items
179    assert queue_id not in e2e_mass.player_queues._managed_pool._materialized
180
181
182@pytest.mark.asyncio
183async def test_recency_denied_track_rotates_to_back_not_dropped(e2e_mass: MusicAssistant) -> None:
184    """A recency-denied finite-source track is kept for a fair second chance, not dropped."""
185    queue_id = demo_players(e2e_mass)[0].player_id
186    test_prov = cast("MusicProvider", e2e_mass.get_provider("test"))
187    assert test_prov is not None
188    queue = e2e_mass.player_queues.get(queue_id)
189    assert queue is not None
190    e2e_mass.player_queues.queue_data(queue_id).userid = TEST_USER
191
192    album = await test_prov.get_album("0_0")  # tracks 0_0_0 .. 0_0_19
193    uri = album.uri
194    assert uri is not None
195    e2e_mass.player_queues.queue_data(queue_id).source_items = cast("list[MediaItemType]", [album])
196
197    # 5 tracks played a day ago -> within the 1-week song window -> recency-denied this round
198    now = int(time.time())
199    denied = [f"0_0_{i}" for i in range(5)]
200    for item_id in denied:
201        await _insert_play(e2e_mass, item_id, now - DAY, TEST_USER)
202
203    pool = await e2e_mass.player_queues._managed_pool.fill(queue_id, is_initial=True)
204    ids = [track.item_id for track in pool]
205
206    # the denied tracks are not handed to the queue this round ...
207    assert not any(item_id in ids for item_id in denied)
208    mat = e2e_mass.player_queues._managed_pool._materialized[queue_id][uri]
209    remaining = [track.item_id for track in mat.tracks]
210    # ... but they are kept (rotated to the back of the deque), not dropped
211    assert remaining[-len(denied) :] == denied
212    # the 15 playable tracks were dispatched; the source is not yet exhausted
213    assert mat.dispatched == {f"0_0_{i}" for i in range(5, 20)}
214    assert e2e_mass.player_queues.queue_data(queue_id).source_items
215
216    # once the recency block lifts, the held tracks get their second chance and the source exhausts
217    await e2e_mass.music.database.delete(
218        DB_TABLE_PLAYLOG, {"provider": "test", "userid": TEST_USER}
219    )
220    pool2 = await e2e_mass.player_queues._managed_pool.fill(queue_id, is_initial=False)
221    ids2 = sorted(track.item_id for track in pool2)
222
223    assert ids2 == sorted(denied)
224    assert not e2e_mass.player_queues.queue_data(queue_id).source_items
225
226
227@pytest.mark.asyncio
228async def test_large_finite_source_is_paged_and_bounded(
229    e2e_mass: MusicAssistant, monkeypatch: pytest.MonkeyPatch
230) -> None:
231    """A source larger than the per-source cap is paged in as it drains, bounding internal state."""
232    # a cap far below the album size forces paging: the deque must never hold the whole album
233    monkeypatch.setattr(managed_pool, "MANAGED_POOL_SOURCE_CAP", 5)
234    queue_id = demo_players(e2e_mass)[0].player_id
235    test_prov = cast("MusicProvider", e2e_mass.get_provider("test"))
236    assert test_prov is not None
237    queue = e2e_mass.player_queues.get(queue_id)
238    assert queue is not None
239    e2e_mass.player_queues.queue_data(queue_id).userid = TEST_USER
240
241    album = await test_prov.get_album("0_0")  # 20 tracks, well over the patched cap of 5
242    uri = album.uri
243    assert uri is not None
244    e2e_mass.player_queues.queue_data(queue_id).source_items = cast("list[MediaItemType]", [album])
245
246    played: list[str] = []
247    max_held = 0
248    fills = 0
249    while e2e_mass.player_queues.queue_data(queue_id).source_items and fills < 20:
250        pool = await e2e_mass.player_queues._managed_pool.fill(queue_id, is_initial=False)
251        played += [track.item_id for track in pool]
252        if mat := e2e_mass.player_queues._managed_pool._materialized.get(queue_id, {}).get(uri):
253            max_held = max(max_held, len(mat.tracks))
254        fills += 1
255
256    # bounded: the materialized deque never held more than the per-source cap at once
257    assert max_held <= 5
258    # paged correctly: every track played exactly once despite being fetched in chunks
259    assert sorted(played) == sorted(f"0_0_{i}" for i in range(20))
260    assert not e2e_mass.player_queues.queue_data(queue_id).source_items
261
262
263@pytest.mark.asyncio
264async def test_dynamic_source_is_not_materialized(e2e_mass: MusicAssistant) -> None:
265    """A dynamic-playlist source self-manages and is never materialized; only finite ones are."""
266    queue_id = demo_players(e2e_mass)[0].player_id
267    test_prov = cast("MusicProvider", e2e_mass.get_provider("test"))
268    assert test_prov is not None
269    queue = e2e_mass.player_queues.get(queue_id)
270    assert queue is not None
271    e2e_mass.player_queues.queue_data(queue_id).userid = TEST_USER
272
273    # a dynamic radio playlist (self-managing) mixed with a finite album (materialized)
274    seed = await test_prov.get_track("4_4_0")
275    radio = await e2e_mass.music.get_item_by_uri(radio_playlist_uri(seed))
276    assert isinstance(radio, Playlist)
277    assert radio.is_dynamic
278    album = await test_prov.get_album("0_0")
279    e2e_mass.player_queues.queue_data(queue_id).source_items = cast(
280        "list[MediaItemType]", [radio, album]
281    )
282
283    pool = await e2e_mass.player_queues._managed_pool.fill(queue_id, is_initial=False)
284
285    assert pool  # both sources feed the bounded pool
286    materialized = e2e_mass.player_queues._managed_pool._materialized.get(queue_id, {})
287    assert album.uri in materialized  # the finite source is materialized into a deque
288    assert radio.uri not in materialized  # the dynamic playlist is left to self-manage
289