/
/
/
1"""Tests for sidebar shortcut cleanup on provider removal and library pruning."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7import pathlib
8import threading
9from collections.abc import AsyncGenerator
10from typing import Any
11
12import pytest
13from music_assistant_models.auth import UserRole
14
15from music_assistant.controllers.config import ConfigController
16from music_assistant.controllers.webserver.auth import (
17 PREF_SIDEBAR_SHORTCUTS,
18 AuthenticationManager,
19)
20from music_assistant.controllers.webserver.controller import WebserverController
21from music_assistant.helpers.json import json_dumps, json_loads
22from music_assistant.mass import MusicAssistant
23
24
25@pytest.fixture
26async def mass_minimal(tmp_path: pathlib.Path) -> AsyncGenerator[MusicAssistant]:
27 """Create a minimal Music Assistant instance for shortcut cleanup testing."""
28 storage_path = tmp_path / "data"
29 cache_path = tmp_path / "cache"
30 storage_path.mkdir(parents=True)
31 cache_path.mkdir(parents=True)
32
33 logging.getLogger("aiosqlite").level = logging.INFO
34
35 mass_instance = MusicAssistant(str(storage_path), str(cache_path))
36 mass_instance.loop = asyncio.get_running_loop()
37 mass_instance.loop_thread_id = threading.get_ident()
38
39 mass_instance.config = ConfigController(mass_instance)
40 await mass_instance.config.setup()
41
42 webserver = WebserverController(mass_instance)
43 mass_instance.webserver = webserver
44
45 webserver_config = await mass_instance.config.get_core_config("webserver")
46 webserver.config = webserver_config
47
48 await webserver.auth.setup()
49
50 try:
51 yield mass_instance
52 finally:
53 await webserver.auth.close()
54 await mass_instance.config.close()
55
56
57@pytest.fixture
58async def auth(mass_minimal: MusicAssistant) -> AuthenticationManager:
59 """Get the auth manager."""
60 return mass_minimal.webserver.auth
61
62
63async def _set_shortcuts(auth: AuthenticationManager, user_id: str, uris: list[str]) -> None:
64 """Write sidebar shortcuts directly into the user's preferences."""
65 prefs = {PREF_SIDEBAR_SHORTCUTS: uris}
66 await auth.database.update(
67 "users",
68 {"user_id": user_id},
69 {"preferences": json_dumps(prefs)},
70 )
71
72
73async def _get_shortcuts(auth: AuthenticationManager, user_id: str) -> list[str]:
74 """Read sidebar shortcuts from a user's preferences."""
75 row = await auth.database.get_row("users", {"user_id": user_id})
76 assert row is not None
77 prefs: dict[str, list[str]] = json_loads(row["preferences"]) if row["preferences"] else {}
78 return prefs.get(PREF_SIDEBAR_SHORTCUTS, [])
79
80
81async def test_cleanup_drops_shortcuts(auth: AuthenticationManager) -> None:
82 """Shortcuts whose rewrite callback returns None are removed."""
83 user = await auth.create_user(username="alice", role=UserRole.USER)
84 await _set_shortcuts(
85 auth,
86 user.user_id,
87 [
88 "spotify_1://track/abc",
89 "library://track/42",
90 "tidal_1://album/xyz",
91 ],
92 )
93
94 async def _drop_spotify(uri: str) -> str | None:
95 if uri.startswith("spotify_1://"):
96 return None
97 return uri
98
99 await auth.cleanup_user_shortcuts(_drop_spotify)
100
101 result = await _get_shortcuts(auth, user.user_id)
102 assert result == ["library://track/42", "tidal_1://album/xyz"]
103
104
105async def test_cleanup_rewrites_shortcuts(auth: AuthenticationManager) -> None:
106 """Shortcuts whose rewrite callback returns a new URI are rewritten."""
107 user = await auth.create_user(username="bob", role=UserRole.USER)
108 await _set_shortcuts(
109 auth,
110 user.user_id,
111 [
112 "spotify_1://track/abc",
113 "library://album/10",
114 ],
115 )
116
117 async def _rewrite_to_library(uri: str) -> str | None:
118 if uri == "spotify_1://track/abc":
119 return "library://track/99"
120 return uri
121
122 await auth.cleanup_user_shortcuts(_rewrite_to_library)
123
124 result = await _get_shortcuts(auth, user.user_id)
125 assert result == ["library://track/99", "library://album/10"]
126
127
128async def test_cleanup_no_change_skips_write(auth: AuthenticationManager) -> None:
129 """When no shortcuts change, the database row is not updated."""
130 user = await auth.create_user(username="carol", role=UserRole.USER)
131 uris = ["library://track/1", "library://track/2"]
132 await _set_shortcuts(auth, user.user_id, uris)
133
134 async def _keep_all(uri: str) -> str | None:
135 return uri
136
137 await auth.cleanup_user_shortcuts(_keep_all)
138
139 result = await _get_shortcuts(auth, user.user_id)
140 assert result == uris
141
142
143async def test_cleanup_empty_shortcuts_skipped(auth: AuthenticationManager) -> None:
144 """Users with no shortcuts are not touched."""
145 await auth.create_user(username="dave", role=UserRole.USER)
146
147 call_count = 0
148
149 async def _counter(uri: str) -> str | None:
150 nonlocal call_count
151 call_count += 1
152 return uri
153
154 await auth.cleanup_user_shortcuts(_counter)
155 assert call_count == 0
156
157
158async def test_cleanup_multiple_users(auth: AuthenticationManager) -> None:
159 """Cleanup iterates over all users."""
160 user_a = await auth.create_user(username="eve", role=UserRole.USER)
161 user_b = await auth.create_user(username="frank", role=UserRole.USER)
162 await _set_shortcuts(auth, user_a.user_id, ["old_provider://track/1"])
163 await _set_shortcuts(auth, user_b.user_id, ["old_provider://track/2", "library://track/5"])
164
165 async def _drop_old(uri: str) -> str | None:
166 if uri.startswith("old_provider://"):
167 return None
168 return uri
169
170 await auth.cleanup_user_shortcuts(_drop_old)
171
172 assert await _get_shortcuts(auth, user_a.user_id) == []
173 assert await _get_shortcuts(auth, user_b.user_id) == ["library://track/5"]
174
175
176async def test_cleanup_preserves_other_preferences(auth: AuthenticationManager) -> None:
177 """Shortcut cleanup does not clobber unrelated preference keys."""
178 user = await auth.create_user(username="grace", role=UserRole.USER)
179 prefs = {
180 "frontend.settings.theme": "dark",
181 PREF_SIDEBAR_SHORTCUTS: ["old://track/1"],
182 }
183 await auth.database.update(
184 "users",
185 {"user_id": user.user_id},
186 {"preferences": json_dumps(prefs)},
187 )
188
189 async def _drop_all(_uri: str) -> str | None:
190 return None
191
192 await auth.cleanup_user_shortcuts(_drop_all)
193
194 row = await auth.database.get_row("users", {"user_id": user.user_id})
195 assert row is not None
196 result_prefs: dict[str, Any] = json_loads(row["preferences"])
197 assert result_prefs["frontend.settings.theme"] == "dark"
198 assert result_prefs[PREF_SIDEBAR_SHORTCUTS] == []
199