/
/
/
1"""Tests for the Zvuk Music setup flow and setup-backed token."""
2
3from __future__ import annotations
4
5import json
6from pathlib import Path
7from types import SimpleNamespace
8from typing import Any, Self
9from unittest.mock import AsyncMock, Mock, patch
10
11import pytest
12from music_assistant_models.enums import ConfigEntryType
13
14from music_assistant.controllers.config.migrations import migrate_provider_setup_data
15from music_assistant.models.music_provider import MusicProvider
16from music_assistant.models.setup_flow import SetupFlowError
17from music_assistant.providers.zvuk_music.constants import CONF_QUALITY, CONF_TOKEN
18from music_assistant.providers.zvuk_music.provider import ZvukMusicProvider
19from music_assistant.providers.zvuk_music.setup_flow import run_setup
20
21
22def _make_session(*, setup_data: dict[str, Any] | None = None) -> Any:
23 """Create a minimal setup session mock."""
24 session = Mock()
25 session.context = SimpleNamespace(setup_data=setup_data or {})
26 session.form = AsyncMock()
27 session.finish = AsyncMock()
28 return session
29
30
31class _ImageResponse:
32 """Provide the async response protocol used by image resolution."""
33
34 status = 200
35
36 async def read(self) -> bytes:
37 """Return deterministic image content."""
38 return b"image"
39
40 async def __aenter__(self) -> Self:
41 """Enter the response context."""
42 return self
43
44 async def __aexit__(self, *args: object) -> None:
45 """Exit the response context."""
46
47
48@pytest.mark.asyncio
49async def test_run_setup_finishes_with_submitted_token() -> None:
50 """The setup flow persists the submitted secure token."""
51 session = _make_session()
52 session.form.return_value = {CONF_TOKEN: "fresh-token"}
53
54 await run_setup(session)
55
56 entries = session.form.await_args.args[0]
57 assert [entry.key for entry in entries][-1] == CONF_TOKEN
58 assert entries[-1].type == ConfigEntryType.SECURE_STRING
59 session.finish.assert_awaited_once_with({CONF_TOKEN: "fresh-token"})
60
61
62@pytest.mark.asyncio
63async def test_run_setup_retries_with_preserved_token_after_validation_error() -> None:
64 """A rejected token stays visible while the setup form reports its error."""
65 session = _make_session()
66 session.form.side_effect = [
67 {CONF_TOKEN: "rejected-token"},
68 {CONF_TOKEN: "replacement-token"},
69 ]
70 session.finish.side_effect = [
71 SetupFlowError("Rejected", translation_key="login_failed"),
72 None,
73 ]
74
75 await run_setup(session)
76
77 second_entries = session.form.await_args_list[1].args[0]
78 assert second_entries[-1].value == "rejected-token"
79 assert session.form.await_args_list[1].kwargs["errors"] == {"base": "login_failed"}
80 assert session.finish.await_args_list[-1].args[0] == {CONF_TOKEN: "replacement-token"}
81
82
83@pytest.mark.asyncio
84async def test_provider_options_expose_quality_but_not_token() -> None:
85 """Authentication stays out of ordinary provider options."""
86 provider = Mock(spec=ZvukMusicProvider)
87 provider.get_config_entries = ZvukMusicProvider.get_config_entries.__get__(
88 provider, ZvukMusicProvider
89 )
90
91 entries = await provider.get_config_entries()
92
93 keys = {entry.key for entry in entries}
94 assert CONF_QUALITY in keys
95 assert CONF_TOKEN not in keys
96
97
98@pytest.mark.asyncio
99async def test_provider_initialization_uses_setup_token() -> None:
100 """Provider initialization connects with the setup-backed token."""
101 provider = Mock(spec=ZvukMusicProvider)
102 provider.get_setup_value = Mock(return_value="setup-token")
103 provider.logger = Mock()
104 provider.handle_async_init = ZvukMusicProvider.handle_async_init.__get__(
105 provider, ZvukMusicProvider
106 )
107 client = Mock()
108 client.connect = AsyncMock()
109
110 with patch(
111 "music_assistant.providers.zvuk_music.provider.ZvukMusicClient", return_value=client
112 ) as client_cls:
113 await provider.handle_async_init()
114
115 provider.get_setup_value.assert_called_once_with(CONF_TOKEN)
116 client_cls.assert_called_once_with("setup-token")
117 client.connect.assert_awaited_once()
118
119
120@pytest.mark.asyncio
121async def test_provider_initialization_uses_migrated_legacy_token() -> None:
122 """Existing installations migrate their config token into setup data."""
123 instance_id = "zvuk_music--legacy"
124 settings = {
125 "providers": {
126 instance_id: {
127 "domain": "zvuk_music",
128 "values": {CONF_TOKEN: "legacy-token"},
129 }
130 }
131 }
132 assert migrate_provider_setup_data(settings, lambda value: f"encrypted:{value}")
133 migrated_config = settings["providers"][instance_id]
134 assert migrated_config["values"] == {}
135 assert migrated_config["setup_data"] == {CONF_TOKEN: "encrypted:legacy-token"}
136
137 provider = Mock(spec=ZvukMusicProvider)
138 provider.instance_id = instance_id
139 provider.mass = Mock()
140 provider.mass.config.get.return_value = migrated_config["setup_data"]
141 provider.mass.config.decrypt_string.return_value = "legacy-token"
142 provider.get_setup_value = MusicProvider.get_setup_value.__get__(provider, ZvukMusicProvider)
143 provider.logger = Mock()
144 provider.handle_async_init = ZvukMusicProvider.handle_async_init.__get__(
145 provider, ZvukMusicProvider
146 )
147 client = Mock()
148 client.connect = AsyncMock()
149
150 with patch(
151 "music_assistant.providers.zvuk_music.provider.ZvukMusicClient", return_value=client
152 ) as client_cls:
153 await provider.handle_async_init()
154
155 provider.mass.config.get.assert_called_once()
156 provider.mass.config.decrypt_string.assert_called_once_with("encrypted:legacy-token")
157 client_cls.assert_called_once_with("legacy-token")
158 client.connect.assert_awaited_once()
159
160
161@pytest.mark.asyncio
162async def test_image_resolution_uses_setup_token() -> None:
163 """Authenticated image requests use the setup-backed token."""
164 provider = Mock(spec=ZvukMusicProvider)
165 provider.get_setup_value = Mock(return_value="setup-token")
166 provider.mass = Mock()
167 provider.mass.http_session.get = Mock(return_value=_ImageResponse())
168 provider.logger = Mock()
169 provider.resolve_image = ZvukMusicProvider.resolve_image.__get__(provider, ZvukMusicProvider)
170
171 result = await provider.resolve_image("https://zvuk.com/cover.jpg")
172
173 assert result == b"image"
174 provider.get_setup_value.assert_called_once_with(CONF_TOKEN)
175 headers = provider.mass.http_session.get.call_args.kwargs["headers"]
176 assert headers["X-Auth-Token"] == "setup-token"
177
178
179def test_strings_define_setup_flow_and_token_specific_login_error() -> None:
180 """Provider strings describe setup and name the rejected token."""
181 strings_path = Path(run_setup.__code__.co_filename).with_name("strings.json")
182 strings = json.loads(strings_path.read_text())
183
184 assert strings["setup_flow"]["user"]["title"] == "Connect to Zvuk"
185 assert "X-Auth-Token" in strings["errors"]["login_failed"]
186 assert "clear_auth" not in strings["config_entries"]
187