/
/
/
1"""
2Tests for the Spotify setup flow's account checks.
3
4Right after the sign-in the flow refuses accounts that cannot work: one without
5Spotify Premium (librespot refuses to stream for a free account) and one that is
6already set up on another provider instance.
7"""
8
9from __future__ import annotations
10
11from typing import Any
12from unittest import mock
13
14import pytest
15from aiohttp import ClientError
16
17from music_assistant.models.setup_flow import AbortFlow, SetupFlowContext, SetupSession
18from music_assistant.providers.spotify import setup_flow as spotify_flow
19from music_assistant.providers.spotify.provider import SpotifyProvider
20
21
22def _make_session(*, instance_id: str | None = None) -> SetupSession:
23 """Return a setup session for a fresh setup (or a reconfigure of the given instance)."""
24 mass = mock.Mock()
25 mass.providers = []
26 mass.config.get_provider_configs = mock.AsyncMock(return_value=[])
27 mass.config.get_provider_setup_value = mock.Mock(return_value=None)
28 mass.get_provider = mock.Mock(return_value=None)
29
30 async def finish(_session: SetupSession, _submitted: dict[str, Any]) -> dict[str, str]:
31 return {"instance_id": "spotify--test"}
32
33 context = SetupFlowContext(
34 kind="reconfigure" if instance_id else "setup",
35 reason="user",
36 domain="spotify",
37 instance_id=instance_id,
38 )
39 return SetupSession(mass, "flow-test", context, finish)
40
41
42def _stub_configs(session: SetupSession, accounts: dict[str, str | None]) -> None:
43 """Point the session at the given configured Spotify instances and their stored accounts."""
44 session.mass.config.get_provider_configs = mock.AsyncMock( # type: ignore[method-assign]
45 return_value=[mock.Mock(instance_id=instance_id) for instance_id in accounts]
46 )
47 session.mass.config.get_provider_setup_value = mock.Mock( # type: ignore[method-assign]
48 side_effect=lambda instance_id, _key: accounts.get(instance_id)
49 )
50
51
52def _stub_me(session: SetupSession, *, status: int = 200, payload: Any = None) -> None:
53 """Point the session's http_session at a canned GET /me response."""
54 response = mock.MagicMock()
55 response.status = status
56 response.json = mock.AsyncMock(return_value=payload)
57 session.mass.http_session.get = mock.MagicMock( # type: ignore[method-assign]
58 return_value=mock.MagicMock(
59 __aenter__=mock.AsyncMock(return_value=response), __aexit__=mock.AsyncMock()
60 )
61 )
62
63
64@pytest.mark.parametrize(
65 ("product", "aborts"),
66 [("premium", False), ("free", True), ("open", True), ("", False), (None, False)],
67)
68async def test_non_premium_accounts_are_turned_away(product: str | None, aborts: bool) -> None:
69 """Only a non-Premium answer aborts; an absent product field is not held against the user."""
70 session = _make_session()
71 payload = {"id": "u1"} if product is None else {"id": "u1", "product": product}
72 _stub_me(session, payload=payload)
73
74 if aborts:
75 with pytest.raises(AbortFlow, match="premium_required"):
76 await spotify_flow._verify_account(session, "at-test")
77 else:
78 await spotify_flow._verify_account(session, "at-test")
79
80
81async def test_a_failing_lookup_does_not_block_the_setup() -> None:
82 """A lookup Spotify answers with an error must not stop the user from setting up."""
83 session = _make_session()
84 _stub_me(session, status=503)
85
86 await spotify_flow._verify_account(session, "at-test")
87
88
89async def test_an_unreachable_lookup_does_not_block_the_setup() -> None:
90 """A lookup that never completes (transport error/timeout) must not stop the setup."""
91 session = _make_session()
92 session.mass.http_session.get = mock.MagicMock( # type: ignore[method-assign]
93 side_effect=ClientError("boom")
94 )
95
96 await spotify_flow._verify_account(session, "at-test")
97
98
99@pytest.mark.parametrize(
100 ("setup_instance_id", "other_instance_id", "aborts"),
101 [
102 # a fresh setup adding an account another instance already serves
103 (None, "spotify--other", True),
104 # a reconfigure of a different instance
105 ("spotify--test", "spotify--other", True),
106 # a reconfigure of the very instance that owns the account
107 ("spotify--test", "spotify--test", False),
108 ],
109)
110async def test_an_already_configured_account_is_refused(
111 setup_instance_id: str | None, other_instance_id: str, aborts: bool
112) -> None:
113 """An account another instance already serves is refused; a reconfigure keeps its own."""
114 session = _make_session(instance_id=setup_instance_id)
115 _stub_configs(session, {other_instance_id: "u1"})
116 _stub_me(session, payload={"id": "u1", "product": "premium"})
117
118 if aborts:
119 with pytest.raises(AbortFlow, match="account_already_configured"):
120 await spotify_flow._verify_account(session, "at-test")
121 else:
122 assert await spotify_flow._verify_account(session, "at-test") == "u1"
123
124
125async def test_a_disabled_instance_still_holds_its_account() -> None:
126 """An instance that is not running is still found through its stored account id."""
127 session = _make_session()
128 # configured but absent from mass.providers, as a disabled or failed instance is
129 _stub_configs(session, {"spotify--disabled": "u1"})
130 _stub_me(session, payload={"id": "u1", "product": "premium"})
131
132 with pytest.raises(AbortFlow, match="account_already_configured"):
133 await spotify_flow._verify_account(session, "at-test")
134
135
136async def test_a_config_without_a_stored_account_falls_back_to_the_instance() -> None:
137 """A configuration predating the stored account id is compared via its running instance."""
138 session = _make_session()
139 _stub_configs(session, {"spotify--legacy": None})
140 legacy = mock.MagicMock(spec=SpotifyProvider)
141 legacy.account_id = "u1"
142 session.mass.get_provider = mock.Mock(return_value=legacy) # type: ignore[method-assign]
143 _stub_me(session, payload={"id": "u1", "product": "premium"})
144
145 with pytest.raises(AbortFlow, match="account_already_configured"):
146 await spotify_flow._verify_account(session, "at-test")
147
148
149async def test_a_different_account_is_accepted() -> None:
150 """A second account alongside an existing instance is allowed and returned."""
151 session = _make_session()
152 _stub_configs(session, {"spotify--other": "u1"})
153 _stub_me(session, payload={"id": "u2", "product": "premium"})
154
155 assert await spotify_flow._verify_account(session, "at-test") == "u2"
156
157
158@pytest.mark.parametrize("payload", [None, [], "nope"])
159async def test_a_malformed_account_response_does_not_block_the_setup(payload: Any) -> None:
160 """A 200 whose body is not an object must fail open like any other bad lookup."""
161 session = _make_session()
162 _stub_me(session, payload=payload)
163
164 assert await spotify_flow._verify_account(session, "at-test") is None
165