/
/
/
1"""
2Tests for the ``keep_preceding_items`` handling of ``start_item`` in the media resolver.
3
4Playing a playlist/album from a chosen track normally drops everything before that track
5("play from here onwards"). With shuffle on that has no meaning, so the preceding tracks are
6moved behind the rest instead of being dropped and the chosen track is returned first.
7"""
8
9from __future__ import annotations
10
11from collections.abc import AsyncIterator
12from typing import Any
13from unittest.mock import AsyncMock, MagicMock
14
15from music_assistant_models.media_items import Album, Playlist, ProviderMapping, Track
16
17from music_assistant.controllers.player_queues.media_resolver import MediaResolver
18
19
20def _track(item_id: str) -> Track:
21 """Build an available Track on the 'test' provider."""
22 return Track(
23 item_id=item_id,
24 provider="test",
25 name=f"Track {item_id}",
26 provider_mappings={
27 ProviderMapping(item_id=item_id, provider_domain="test", provider_instance="test")
28 },
29 )
30
31
32def _resolver() -> tuple[MediaResolver, MagicMock]:
33 """Create a bare resolver, plus the mock standing in for the `mass` instance."""
34 resolver = MediaResolver.__new__(MediaResolver)
35 mass = MagicMock()
36 mass.config.get_raw_core_config_value = MagicMock(return_value="all_tracks")
37 resolver.mass = mass
38 resolver.queues = MagicMock()
39 resolver.logger = MagicMock()
40 return resolver, mass
41
42
43def _album_resolver(tracks: list[Track]) -> MediaResolver:
44 """Create a resolver whose album track listing returns the given tracks."""
45 resolver, mass = _resolver()
46 mass.music.albums.tracks = AsyncMock(return_value=tracks)
47 return resolver
48
49
50def _playlist_resolver(tracks: list[Track]) -> MediaResolver:
51 """Create a resolver whose playlist track listing yields the given tracks."""
52 resolver, mass = _resolver()
53
54 async def _tracks(*_args: Any, **_kwargs: Any) -> AsyncIterator[Track]:
55 for track in tracks:
56 yield track
57
58 mass.music.playlists.tracks = _tracks
59 return resolver
60
61
62def _album() -> Album:
63 """Build an Album on the 'test' provider."""
64 return Album(
65 item_id="al",
66 provider="test",
67 name="Album",
68 provider_mappings={
69 ProviderMapping(item_id="al", provider_domain="test", provider_instance="test")
70 },
71 )
72
73
74def _playlist() -> Playlist:
75 """Build a non-dynamic Playlist on the 'test' provider."""
76 return Playlist(
77 item_id="pl",
78 provider="test",
79 name="Playlist",
80 provider_mappings={
81 ProviderMapping(item_id="pl", provider_domain="test", provider_instance="test")
82 },
83 )
84
85
86async def test_album_start_item_drops_preceding_tracks_by_default() -> None:
87 """Without the flag an album starts at the chosen track and the earlier ones are dropped."""
88 tracks = [_track(x) for x in ("a", "b", "c", "d")]
89 resolver = _album_resolver(tracks)
90
91 result = await resolver.get_album_tracks(_album(), start_item="c")
92
93 assert [track.item_id for track in result] == ["c", "d"]
94
95
96async def test_album_start_item_keeps_preceding_tracks_behind_the_rest() -> None:
97 """With the flag the whole album is returned, rotated so the chosen track comes first."""
98 tracks = [_track(x) for x in ("a", "b", "c", "d")]
99 resolver = _album_resolver(tracks)
100
101 result = await resolver.get_album_tracks(_album(), start_item="c", keep_preceding_items=True)
102
103 assert [track.item_id for track in result] == ["c", "d", "a", "b"]
104
105
106async def test_playlist_start_item_drops_preceding_tracks_by_default() -> None:
107 """Without the flag a playlist starts at the chosen track and the earlier ones are dropped."""
108 tracks = [_track(x) for x in ("a", "b", "c", "d")]
109 resolver = _playlist_resolver(tracks)
110
111 result = await resolver.get_playlist_tracks(_playlist(), start_item="c")
112
113 assert [track.item_id for track in result] == ["c", "d"]
114
115
116async def test_playlist_start_item_keeps_preceding_tracks_behind_the_rest() -> None:
117 """With the flag the whole playlist is returned, rotated so the chosen track comes first."""
118 tracks = [_track(x) for x in ("a", "b", "c", "d")]
119 resolver = _playlist_resolver(tracks)
120
121 result = await resolver.get_playlist_tracks(
122 _playlist(), start_item="c", keep_preceding_items=True
123 )
124
125 assert [track.item_id for track in result] == ["c", "d", "a", "b"]
126
127
128async def test_playlist_without_start_item_keeps_full_order() -> None:
129 """The flag is a no-op when no start item is given: the playlist keeps its own order."""
130 tracks = [_track(x) for x in ("a", "b", "c")]
131 resolver = _playlist_resolver(tracks)
132
133 result = await resolver.get_playlist_tracks(
134 _playlist(), start_item=None, keep_preceding_items=True
135 )
136
137 assert [track.item_id for track in result] == ["a", "b", "c"]
138
139
140async def test_playlist_unknown_start_item_yields_nothing() -> None:
141 """An unmatched start item still yields no tracks, so the caller can fall back."""
142 tracks = [_track(x) for x in ("a", "b", "c")]
143 resolver = _playlist_resolver(tracks)
144
145 result = await resolver.get_playlist_tracks(
146 _playlist(), start_item="zz", keep_preceding_items=True
147 )
148
149 assert result == []
150