/
/
/
1"""Tests for the Smart Playlist provider recommendations."""
2
3from __future__ import annotations
4
5from unittest.mock import AsyncMock, MagicMock
6
7from music_assistant_models.enums import EventType
8from music_assistant_models.event import MassEvent
9from music_assistant_models.media_items import ItemMapping, Playlist, ProviderMapping
10
11from music_assistant.providers.smart_playlist import SmartPlaylistProvider
12from music_assistant.providers.smart_playlist.helpers import SmartPlaylistRules
13
14
15def _make_plugin() -> tuple[SmartPlaylistProvider, MagicMock]:
16 """Create a SmartPlaylistProvider with mocked mass and empty stores."""
17 mass = MagicMock()
18 manifest = MagicMock()
19 manifest.domain = "smart_playlist"
20 config = MagicMock()
21 config.get_value.return_value = "GLOBAL"
22 plugin = SmartPlaylistProvider(mass, manifest, config, set())
23 plugin._rules_store = {}
24 plugin._names_store = {}
25 plugin._descriptions_store = {}
26 # the only backend touchpoint of the items path (playlist artwork lookup)
27 mass.music.playlists.get_library_item_by_prov_id = AsyncMock(return_value=None)
28 return plugin, mass
29
30
31async def test_get_recommendations_returns_static_row() -> None:
32 """A populated rules store yields the single static row, without items or backend I/O."""
33 plugin, mass = _make_plugin()
34 plugin._rules_store["abc"] = SmartPlaylistRules(limit=10)
35
36 result = await plugin.get_recommendations()
37
38 assert len(result) == 1
39 folder = result[0]
40 assert folder.item_id == "smart_playlists"
41 # rows must carry the instance_id: the items API resolves the provider from it
42 assert folder.provider == plugin.instance_id
43 assert folder.name == "Smart Playlists"
44 assert folder.translation_key == "smart_playlists"
45 assert len(folder.items) == 0
46 mass.music.playlists.get_library_item_by_prov_id.assert_not_awaited()
47
48
49async def test_get_recommendations_empty_store_returns_no_rows() -> None:
50 """An empty rules store yields no rows."""
51 plugin, _mass = _make_plugin()
52
53 assert await plugin.get_recommendations() == []
54
55
56async def test_get_recommendation_items_builds_playlists() -> None:
57 """The smart_playlists row builds one playlist per stored rule set."""
58 plugin, mass = _make_plugin()
59 plugin._rules_store["abc"] = SmartPlaylistRules(limit=10, is_dynamic=True)
60 plugin._rules_store["def"] = SmartPlaylistRules(limit=20, is_dynamic=False)
61 plugin._names_store["abc"] = "Playlist A"
62 plugin._names_store["def"] = "Playlist B"
63
64 result = await plugin.get_recommendation_items("smart_playlists")
65
66 assert [item.item_id for item in result] == ["abc", "def"]
67 assert all(isinstance(item, Playlist) for item in result)
68 assert [item.name for item in result] == ["Playlist A", "Playlist B"]
69 assert mass.music.playlists.get_library_item_by_prov_id.await_count == 2
70
71
72async def test_get_recommendation_items_unknown_id_returns_empty() -> None:
73 """An unknown row item_id returns empty without building any playlist."""
74 plugin, mass = _make_plugin()
75 plugin._rules_store["abc"] = SmartPlaylistRules(limit=10)
76
77 result = await plugin.get_recommendation_items("unknown_row")
78
79 assert list(result) == []
80 mass.music.playlists.get_library_item_by_prov_id.assert_not_awaited()
81
82
83async def test_get_recommendation_items_returns_item_mapping_for_library_playlists() -> None:
84 """Library-backed playlists are returned as ItemMapping with library DB ID."""
85 plugin, mass = _make_plugin()
86 plugin._rules_store["abc"] = SmartPlaylistRules(limit=10)
87 plugin._names_store["abc"] = "Playlist A"
88
89 library_playlist = Playlist(
90 item_id="42",
91 provider="library",
92 name="Playlist A (Library)",
93 owner="Smart Playlist",
94 provider_mappings=set(),
95 )
96 mass.music.playlists.get_library_item_by_prov_id = AsyncMock(return_value=library_playlist)
97
98 result = await plugin.get_recommendation_items("smart_playlists")
99
100 assert len(result) == 1
101 item = result[0]
102 assert isinstance(item, ItemMapping)
103 assert item.item_id == "42"
104 assert item.provider == "library"
105 assert item.name == "Playlist A (Library)"
106
107
108async def test_delete_smart_playlist_signals_recommendations_updated() -> None:
109 """Deleting a smart playlist emits a recommendations_updated event."""
110 plugin, _mass = _make_plugin()
111 plugin._rules_store["abc123"] = SmartPlaylistRules(limit=10)
112 plugin._names_store["abc123"] = "Test Playlist"
113 plugin._flush_rules_to_disk = AsyncMock() # type: ignore[method-assign]
114 plugin._invalidate_dynamic_sample_cache = AsyncMock() # type: ignore[method-assign]
115 plugin.signal_provider_event = MagicMock() # type: ignore[method-assign, misc]
116
117 library_playlist = Playlist(
118 item_id="999",
119 provider="library",
120 name="Test Playlist",
121 owner="Smart Playlist",
122 provider_mappings={
123 ProviderMapping(
124 item_id="abc123",
125 provider_domain="smart_playlist",
126 provider_instance=plugin.instance_id,
127 )
128 },
129 )
130 event = MassEvent(
131 event=EventType.MEDIA_ITEM_DELETED,
132 object_id="999",
133 data=library_playlist,
134 )
135
136 await plugin._on_media_item_deleted(event)
137
138 assert "abc123" not in plugin._rules_store
139 assert "abc123" not in plugin._names_store
140 payload = plugin.signal_provider_event.call_args.args[0]
141 assert payload["event"] == "recommendations_updated"
142
143
144async def test_delete_unrelated_playlist_does_not_signal() -> None:
145 """Deleting a playlist from a different provider does not emit an event."""
146 plugin, _mass = _make_plugin()
147 plugin._rules_store["abc123"] = SmartPlaylistRules(limit=10)
148 plugin.signal_provider_event = MagicMock() # type: ignore[method-assign, misc]
149
150 unrelated_playlist = Playlist(
151 item_id="999",
152 provider="library",
153 name="Unrelated Playlist",
154 owner="Other",
155 provider_mappings={
156 ProviderMapping(
157 item_id="other_id",
158 provider_domain="other_provider",
159 provider_instance="other_instance",
160 )
161 },
162 )
163 event = MassEvent(
164 event=EventType.MEDIA_ITEM_DELETED,
165 object_id="999",
166 data=unrelated_playlist,
167 )
168
169 await plugin._on_media_item_deleted(event)
170
171 assert "abc123" in plugin._rules_store
172 plugin.signal_provider_event.assert_not_called()
173
174
175async def test_create_smart_playlist_signals_recommendations_updated() -> None:
176 """Creating a smart playlist emits a recommendations_updated event."""
177 plugin, mass = _make_plugin()
178 plugin._validate_rules = MagicMock() # type: ignore[method-assign]
179 plugin._save_rules = AsyncMock() # type: ignore[method-assign]
180 plugin._schedule_ai_description_refresh = MagicMock() # type: ignore[method-assign]
181 plugin.signal_provider_event = MagicMock() # type: ignore[method-assign, misc]
182
183 library_playlist = Playlist(
184 item_id="42",
185 provider="library",
186 name="New Playlist",
187 owner="Smart Playlist",
188 provider_mappings=set(),
189 )
190 mass.music.playlists.add_item_to_library = AsyncMock(return_value=library_playlist)
191 mass.metadata.schedule_update_metadata = MagicMock()
192
193 await plugin.create_smart_playlist(
194 name="New Playlist",
195 rules={"limit": 10},
196 is_dynamic=False,
197 )
198
199 payload = plugin.signal_provider_event.call_args.args[0]
200 assert payload["event"] == "recommendations_updated"
201