/
/
/
1"""Tests for MusicController.recently_played playlog queries."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING
6from unittest.mock import Mock, patch
7
8from music_assistant_models.enums import MediaType
9
10from music_assistant.constants import DB_TABLE_PLAYLOG, DB_TABLE_PROVIDER_MAPPINGS
11from music_assistant.mass import MusicAssistant
12
13if TYPE_CHECKING:
14 import pytest
15
16GET_CURRENT_USER = "music_assistant.controllers.music.controller.get_current_user"
17
18
19async def _add_playlog_track(
20 mass: MusicAssistant,
21 item_id: str,
22 timestamp: int,
23 *,
24 fully_played: bool = True,
25 userid: str = "user-a",
26 provider: str = "library",
27) -> None:
28 """Insert a single track row into the playlog."""
29 await mass.music.database.insert(
30 DB_TABLE_PLAYLOG,
31 {
32 "item_id": item_id,
33 "provider": provider,
34 "media_type": MediaType.TRACK.value,
35 "name": f"Track {item_id}",
36 "timestamp": timestamp,
37 "fully_played": fully_played,
38 "seconds_played": 180,
39 "userid": userid,
40 "user_initiated": True,
41 },
42 )
43
44
45async def _add_provider_mapping(
46 mass: MusicAssistant,
47 item_id: int,
48 provider_instance: str,
49 media_type: MediaType = MediaType.TRACK,
50 *,
51 available: bool = True,
52) -> None:
53 """Insert a provider mapping row for a library item."""
54 await mass.music.database.insert(
55 DB_TABLE_PROVIDER_MAPPINGS,
56 {
57 "media_type": media_type.value,
58 "item_id": item_id,
59 "provider_domain": provider_instance.rstrip("0123456789_"),
60 "provider_instance": provider_instance,
61 "provider_item_id": f"{provider_instance}-{item_id}",
62 "available": available,
63 "in_library": True,
64 },
65 )
66
67
68async def test_recently_played_filters_by_played_after_timestamp(mass: MusicAssistant) -> None:
69 """Only entries with timestamp >= played_after_timestamp are returned."""
70 await _add_playlog_track(mass, "recent", timestamp=2000)
71 await _add_playlog_track(mass, "old", timestamp=1000)
72
73 result = await mass.music.recently_played(
74 limit=0,
75 media_types=[MediaType.TRACK],
76 userid="user-a",
77 played_after_timestamp=1500,
78 )
79
80 item_ids = {item.item_id for item in result}
81 assert "recent" in item_ids
82 assert "old" not in item_ids
83
84
85async def test_recently_played_without_timestamp_returns_all(mass: MusicAssistant) -> None:
86 """Existing callers (no played_after_timestamp) still get every entry."""
87 await _add_playlog_track(mass, "recent", timestamp=2000)
88 await _add_playlog_track(mass, "old", timestamp=1000)
89
90 result = await mass.music.recently_played(
91 limit=0,
92 media_types=[MediaType.TRACK],
93 userid="user-a",
94 )
95
96 item_ids = {item.item_id for item in result}
97 assert {"recent", "old"} <= item_ids
98
99
100async def test_recently_played_filters_direct_provider_entries(
101 mass: MusicAssistant, monkeypatch: pytest.MonkeyPatch
102) -> None:
103 """Direct (non-library) playlog entries are matched against their own provider."""
104 monkeypatch.setattr(
105 mass.music, "get_active_provider_instances", lambda: ["spotify_1", "local_1"]
106 )
107 await _add_playlog_track(mass, "spotify-track", timestamp=2000, provider="spotify_1")
108 await _add_playlog_track(mass, "local-track", timestamp=1999, provider="local_1")
109
110 result = await mass.music.recently_played(
111 limit=0, media_types=[MediaType.TRACK], userid="user-a", providers=["local_1"]
112 )
113
114 assert {item.item_id for item in result} == {"local-track"}
115
116
117async def test_recently_played_multiple_providers_or_semantics(
118 mass: MusicAssistant, monkeypatch: pytest.MonkeyPatch
119) -> None:
120 """Requesting several providers matches entries from any one of them."""
121 monkeypatch.setattr(
122 mass.music, "get_active_provider_instances", lambda: ["spotify_1", "local_1", "tidal_1"]
123 )
124 await _add_playlog_track(mass, "spotify-track", timestamp=2000, provider="spotify_1")
125 await _add_playlog_track(mass, "local-track", timestamp=1999, provider="local_1")
126 await _add_playlog_track(mass, "tidal-track", timestamp=1998, provider="tidal_1")
127
128 result = await mass.music.recently_played(
129 limit=0,
130 media_types=[MediaType.TRACK],
131 userid="user-a",
132 providers=["spotify_1", "local_1"],
133 )
134
135 assert {item.item_id for item in result} == {"spotify-track", "local-track"}
136
137
138async def test_recently_played_library_item_or_matches_any_mapping(
139 mass: MusicAssistant, monkeypatch: pytest.MonkeyPatch
140) -> None:
141 """A library item mapped to Spotify and local remains when only local is requested."""
142 monkeypatch.setattr(
143 mass.music, "get_active_provider_instances", lambda: ["spotify_1", "local_1"]
144 )
145 await _add_provider_mapping(mass, item_id=1, provider_instance="spotify_1")
146 await _add_provider_mapping(mass, item_id=1, provider_instance="local_1")
147 await _add_playlog_track(mass, "1", timestamp=2000, provider="library")
148
149 result = await mass.music.recently_played(
150 limit=0, media_types=[MediaType.TRACK], userid="user-a", providers=["local_1"]
151 )
152
153 assert {item.item_id for item in result} == {"1"}
154
155
156async def test_recently_played_excludes_unavailable_mapping(
157 mass: MusicAssistant, monkeypatch: pytest.MonkeyPatch
158) -> None:
159 """A mapping that is no longer available on the requested provider cannot match."""
160 monkeypatch.setattr(
161 mass.music, "get_active_provider_instances", lambda: ["spotify_1", "local_1"]
162 )
163 await _add_provider_mapping(mass, item_id=1, provider_instance="local_1")
164 await _add_provider_mapping(mass, item_id=1, provider_instance="spotify_1", available=False)
165 await _add_playlog_track(mass, "1", timestamp=2000, provider="library")
166
167 result = await mass.music.recently_played(
168 limit=0, media_types=[MediaType.TRACK], userid="user-a", providers=["spotify_1"]
169 )
170
171 assert result == []
172
173
174async def test_recently_played_explicit_empty_providers_returns_no_items(
175 mass: MusicAssistant,
176) -> None:
177 """An explicit empty provider list returns no items, distinct from omitting the filter."""
178 await _add_playlog_track(mass, "recent", timestamp=2000)
179
180 result = await mass.music.recently_played(
181 limit=0, media_types=[MediaType.TRACK], userid="user-a", providers=[]
182 )
183
184 assert result == []
185
186
187async def test_recently_played_combines_explicit_and_user_provider_filter(
188 mass: MusicAssistant, monkeypatch: pytest.MonkeyPatch
189) -> None:
190 """A user's provider_filter narrows an explicit filter that would otherwise pass."""
191 # get_active_provider_instances already applies the user's provider_filter internally, so a
192 # user restricted to local_1 never sees spotify_1 in the "active" provider set.
193 monkeypatch.setattr(mass.music, "get_active_provider_instances", lambda: ["local_1"])
194 await _add_provider_mapping(mass, item_id=1, provider_instance="spotify_1")
195 await _add_provider_mapping(mass, item_id=1, provider_instance="local_1")
196 await _add_playlog_track(mass, "1", timestamp=2000, provider="library")
197
198 with patch(GET_CURRENT_USER, return_value=Mock(user_id="user-a", provider_filter=["local_1"])):
199 result = await mass.music.recently_played(
200 limit=0, media_types=[MediaType.TRACK], providers=["local_1"]
201 )
202 assert {item.item_id for item in result} == {"1"}
203
204 # requesting a provider the user isn't permitted to use must not leak the item back
205 # in, even though the item does have a (permission-restricted) mapping to it.
206 with patch(GET_CURRENT_USER, return_value=Mock(user_id="user-a", provider_filter=["local_1"])):
207 result = await mass.music.recently_played(
208 limit=0, media_types=[MediaType.TRACK], providers=["spotify_1"]
209 )
210 assert result == []
211
212
213async def test_recently_played_provider_filter_applied_before_limit(
214 mass: MusicAssistant, monkeypatch: pytest.MonkeyPatch
215) -> None:
216 """The provider filter runs in SQL before LIMIT, so filtered-out rows can't starve results."""
217 monkeypatch.setattr(
218 mass.music, "get_active_provider_instances", lambda: ["spotify_1", "local_1"]
219 )
220 # newer spotify entries (excluded by the filter) outnumber and outrank the older
221 # local entries; if the filter were applied after LIMIT, they would crowd it out.
222 for i in range(10):
223 await _add_playlog_track(mass, f"spotify-{i}", timestamp=3000 - i, provider="spotify_1")
224 for i in range(5):
225 await _add_playlog_track(mass, f"local-{i}", timestamp=2000 - i, provider="local_1")
226
227 result = await mass.music.recently_played(
228 limit=5, media_types=[MediaType.TRACK], userid="user-a", providers=["local_1"]
229 )
230
231 assert {item.item_id for item in result} == {f"local-{i}" for i in range(5)}
232