/
/
/
1"""Tests for the media resolver's get_tracks_for_playback umbrella dispatch."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING, cast
6from unittest.mock import AsyncMock, MagicMock
7
8import pytest
9from music_assistant_models.enums import MediaType
10from music_assistant_models.media_items import Audiobook, MediaCollection, Track, UniqueList
11
12from music_assistant.controllers.player_queues.media_resolver import MediaResolver
13
14if TYPE_CHECKING:
15 from music_assistant_models.media_items import MediaItemType
16
17
18def _trk(item_id: str) -> Track:
19 return Track(item_id=item_id, provider="library", name=item_id, provider_mappings=set())
20
21
22def _resolver() -> MediaResolver:
23 resolver = MediaResolver.__new__(MediaResolver)
24 resolver.get_artist_tracks = AsyncMock(return_value=[_trk("artist")]) # type: ignore[method-assign]
25 resolver.get_album_tracks = AsyncMock(return_value=[_trk("album")]) # type: ignore[method-assign]
26 resolver.get_genre_tracks = AsyncMock(return_value=[_trk("genre")]) # type: ignore[method-assign]
27 resolver.get_playlist_tracks = AsyncMock(return_value=[]) # type: ignore[method-assign]
28 return resolver
29
30
31def _item(media_type: MediaType) -> MediaItemType:
32 item = MagicMock()
33 item.media_type = media_type
34 return cast("MediaItemType", item)
35
36
37@pytest.mark.asyncio
38async def test_track_resolves_to_itself() -> None:
39 """A track resolves to itself without touching the per-type resolvers."""
40 resolver = _resolver()
41 track = _trk("solo")
42 assert await resolver.get_tracks_for_playback(track) == [track]
43
44
45@pytest.mark.asyncio
46async def test_album_routes_to_get_album_tracks() -> None:
47 """An album is resolved through the config-driven get_album_tracks."""
48 resolver = _resolver()
49 result = await resolver.get_tracks_for_playback(_item(MediaType.ALBUM))
50 assert [t.item_id for t in result] == ["album"]
51
52
53@pytest.mark.asyncio
54async def test_artist_routes_to_get_artist_tracks() -> None:
55 """An artist is resolved through the config-driven get_artist_tracks."""
56 resolver = _resolver()
57 result = await resolver.get_tracks_for_playback(_item(MediaType.ARTIST))
58 assert [t.item_id for t in result] == ["artist"]
59
60
61@pytest.mark.asyncio
62async def test_genre_routes_to_get_genre_tracks() -> None:
63 """A genre is resolved through get_genre_tracks."""
64 resolver = _resolver()
65 result = await resolver.get_tracks_for_playback(_item(MediaType.GENRE))
66 assert [t.item_id for t in result] == ["genre"]
67
68
69@pytest.mark.asyncio
70async def test_genre_samples_artists_via_top_tracks() -> None:
71 """Genre resolution samples artists' top tracks, never the full artist-tracks resolution."""
72 resolver = MediaResolver.__new__(MediaResolver)
73 resolver.mass = MagicMock()
74 resolver.logger = MagicMock()
75 resolver.get_artist_tracks = AsyncMock() # type: ignore[method-assign]
76 artist = MagicMock()
77 artist.item_id = "a1"
78 artist.provider = "test"
79 resolver.mass.music.genres.mapped_media = AsyncMock(return_value=([], [], [artist]))
80 resolver.mass.music.artists.top_tracks = AsyncMock(return_value=[_trk("top1"), _trk("top2")])
81 genre = MagicMock()
82 genre.name = "Rock"
83
84 result = await resolver.get_genre_tracks(genre, None)
85
86 resolver.mass.music.artists.top_tracks.assert_awaited_once_with("a1", "test")
87 resolver.get_artist_tracks.assert_not_awaited()
88 assert {t.item_id for t in result} == {"top1", "top2"}
89
90
91@pytest.mark.asyncio
92async def test_genre_artist_without_top_tracks_falls_back() -> None:
93 """An empty top tracks listing falls back to the plain, preference-independent listing."""
94 resolver = MediaResolver.__new__(MediaResolver)
95 resolver.mass = MagicMock()
96 resolver.logger = MagicMock()
97 resolver.get_artist_tracks = AsyncMock() # type: ignore[method-assign]
98 artist = MagicMock()
99 artist.item_id = "a1"
100 artist.provider = "test"
101 resolver.mass.music.genres.mapped_media = AsyncMock(return_value=([], [], [artist]))
102 resolver.mass.music.artists.top_tracks = AsyncMock(return_value=[])
103 resolver.mass.music.artists.tracks = AsyncMock(return_value=[_trk("full1")])
104 genre = MagicMock()
105 genre.name = "Rock"
106
107 result = await resolver.get_genre_tracks(genre, None)
108
109 resolver.mass.music.artists.tracks.assert_awaited_once_with("a1", "test")
110 resolver.get_artist_tracks.assert_not_awaited()
111 assert {t.item_id for t in result} == {"full1"}
112
113
114@pytest.mark.asyncio
115async def test_playlist_filters_to_tracks() -> None:
116 """A playlist resolves to its tracks, dropping any non-track playable items."""
117 resolver = _resolver()
118 track = _trk("p1")
119 resolver.get_playlist_tracks = AsyncMock(return_value=[track, MagicMock()]) # type: ignore[method-assign]
120 assert await resolver.get_tracks_for_playback(_item(MediaType.PLAYLIST)) == [track]
121
122
123@pytest.mark.asyncio
124async def test_unsupported_type_is_empty() -> None:
125 """A type with no playable-track resolution yields an empty list."""
126 resolver = _resolver()
127 assert await resolver.get_tracks_for_playback(_item(MediaType.AUDIOBOOK)) == []
128
129
130@pytest.mark.asyncio
131async def test_audiobook_collection_resolves_existing_items() -> None:
132 """An audiobook collection resolves its model items without deserializing them again."""
133 resolver = _resolver()
134 resolver.mass = MagicMock()
135 resolver.mass.music.get_resume_position = AsyncMock(side_effect=[(True, 0), (False, 12_000)])
136 first = Audiobook(
137 item_id="book-1",
138 provider="library",
139 name="Book 1",
140 provider_mappings=set(),
141 )
142 second = Audiobook(
143 item_id="book-2",
144 provider="library",
145 name="Book 2",
146 provider_mappings=set(),
147 )
148 collection = MediaCollection[Audiobook](
149 item_id="audiobook___Series",
150 provider="library",
151 name="Series",
152 provider_mappings=set(),
153 items=UniqueList([first, second]),
154 )
155
156 result = await resolver._resolve_media_items(collection, userid="user")
157
158 assert result == [second]
159 assert result[0] is second
160 assert second.resume_position_ms == 12_000
161