/
/
/
1"""Tests for the OneDrive OAuth helper (token refresh and rotation)."""
2
3from __future__ import annotations
4
5import time
6from unittest.mock import AsyncMock, MagicMock
7
8import pytest
9from aiohttp import ClientError
10from music_assistant_models.errors import LoginFailed, ProviderUnavailableError
11
12from music_assistant.providers.filesystem_cloud.base import CONF_REFRESH_TOKEN
13from music_assistant.providers.filesystem_onedrive.auth import MAOneDriveAuth, _friendly_auth_error
14
15
16def _make_auth() -> tuple[MAOneDriveAuth, MagicMock]:
17 """Return an auth helper plus the mocked mass behind it."""
18 mass = MagicMock()
19 # the store keeps values encrypted; encrypt is an identity map for the test
20 mass.config.encrypt_string = MagicMock(side_effect=lambda value: value)
21 return MAOneDriveAuth(mass, "onedrive--test", "client-id", "client-secret", "refresh-1"), mass
22
23
24def _response_cm(response: MagicMock) -> MagicMock:
25 """Build a fake async context manager mimicking aiohttp's session.post()."""
26 cm = MagicMock()
27 cm.__aenter__ = AsyncMock(return_value=response)
28 cm.__aexit__ = AsyncMock(return_value=False)
29 return cm
30
31
32def _token_response(payload: dict[str, object]) -> MagicMock:
33 """Build a fake 200 token response with the given JSON payload."""
34 response = MagicMock()
35 response.status = 200
36 response.json = AsyncMock(return_value=payload)
37 return response
38
39
40async def test_cached_access_token_is_reused() -> None:
41 """A still-valid access token is returned without calling Microsoft."""
42 auth, mass = _make_auth()
43 auth._access_token = "cached-token"
44 auth._expires_at = time.time() + 3600
45
46 assert await auth.async_get_access_token() == "cached-token"
47 mass.http_session.post.assert_not_called()
48
49
50async def test_refresh_persists_rotated_token_immediately() -> None:
51 """A rotated refresh token is stored encrypted, bypassing the debounced save."""
52 auth, mass = _make_auth()
53 mass.http_session.post = MagicMock(
54 return_value=_response_cm(
55 _token_response(
56 {"access_token": "access-1", "expires_in": 3600, "refresh_token": "refresh-2"}
57 )
58 )
59 )
60
61 assert await auth.async_get_access_token() == "access-1"
62
63 assert auth._refresh_token == "refresh-2"
64 # the rotated token is persisted into setup_data (encrypt is identity here)
65 mass.config.set.assert_called_once_with(
66 f"providers/onedrive--test/setup_data/{CONF_REFRESH_TOKEN}", "refresh-2", immediate=True
67 )
68
69
70async def test_refresh_skips_persisting_unchanged_token() -> None:
71 """No setup_data write happens when Microsoft returns the same refresh token."""
72 auth, mass = _make_auth()
73 mass.http_session.post = MagicMock(
74 return_value=_response_cm(
75 _token_response(
76 {"access_token": "access-1", "expires_in": 3600, "refresh_token": "refresh-1"}
77 )
78 )
79 )
80
81 await auth.async_get_access_token()
82
83 mass.config.set.assert_not_called()
84
85
86async def test_refresh_keeps_token_when_none_returned() -> None:
87 """A response without a refresh token keeps the current one."""
88 auth, mass = _make_auth()
89 mass.http_session.post = MagicMock(
90 return_value=_response_cm(_token_response({"access_token": "access-1", "expires_in": 60}))
91 )
92
93 await auth.async_get_access_token()
94
95 assert auth._refresh_token == "refresh-1"
96 mass.config.set.assert_not_called()
97
98
99async def test_refresh_auth_error_raises_login_failed() -> None:
100 """A 400 from the token endpoint surfaces as LoginFailed with a friendly message."""
101 auth, mass = _make_auth()
102 response = MagicMock()
103 response.status = 400
104 response.text = AsyncMock(return_value='{"error": "invalid_grant"}')
105 mass.http_session.post = MagicMock(return_value=_response_cm(response))
106
107 with pytest.raises(LoginFailed, match="re-authorize"):
108 await auth.async_get_access_token()
109
110
111async def test_refresh_network_error_raises_provider_unavailable() -> None:
112 """Transport failures are not treated as auth problems."""
113 auth, mass = _make_auth()
114 mass.http_session.post = MagicMock(side_effect=ClientError("connection reset"))
115
116 with pytest.raises(ProviderUnavailableError):
117 await auth.async_get_access_token()
118
119
120def test_friendly_auth_error_mapping() -> None:
121 """Well-known Microsoft error codes map to actionable messages."""
122 assert "secret has expired" in str(_friendly_auth_error("AADSTS7000222: expired"))
123 assert "Secret ID" in str(_friendly_auth_error("AADSTS7000215: invalid secret"))
124 assert "re-authorize" in str(_friendly_auth_error("invalid_grant: revoked"))
125 assert _friendly_auth_error("something else entirely") is None
126