/
/
/
1"""Tests for the guest access helpers."""
2
3from __future__ import annotations
4
5from unittest.mock import AsyncMock, MagicMock
6
7import pytest
8from music_assistant_models.auth import UserRole
9from music_assistant_models.errors import InvalidDataError
10
11from music_assistant.helpers import guest_access
12
13
14def _create_mock_mass() -> MagicMock:
15 """Create a mock MusicAssistant with a mocked auth controller."""
16 mass = MagicMock()
17 auth = mass.webserver.auth
18 auth.get_user_by_username = AsyncMock(return_value=None)
19 auth.create_user = AsyncMock()
20 auth.get_active_join_code = AsyncMock(return_value=None)
21 auth.generate_join_code = AsyncMock()
22 auth.revoke_join_codes = AsyncMock(return_value=0)
23 auth.revoke_tokens_for_user = AsyncMock(return_value=0)
24 return mass
25
26
27async def test_get_or_create_guest_user_returns_existing() -> None:
28 """An existing guest user is returned without creating a new one."""
29 mass = _create_mock_mass()
30 existing_user = MagicMock(role=UserRole.GUEST)
31 mass.webserver.auth.get_user_by_username.return_value = existing_user
32
33 user = await guest_access.get_or_create_guest_user(mass, "party_guest", "Party Guest")
34
35 assert user is existing_user
36 mass.webserver.auth.create_user.assert_not_awaited()
37
38
39async def test_get_or_create_guest_user_rejects_non_guest() -> None:
40 """An existing user with a non-guest role is never reused for guest access."""
41 mass = _create_mock_mass()
42 mass.webserver.auth.get_user_by_username.return_value = MagicMock(role=UserRole.ADMIN)
43
44 with pytest.raises(InvalidDataError):
45 await guest_access.get_or_create_guest_user(mass, "party_guest", "Party Guest")
46 mass.webserver.auth.create_user.assert_not_awaited()
47
48
49async def test_get_or_create_guest_user_creates_guest() -> None:
50 """A missing guest user is created with the GUEST role."""
51 mass = _create_mock_mass()
52 created_user = MagicMock()
53 mass.webserver.auth.create_user.return_value = created_user
54
55 user = await guest_access.get_or_create_guest_user(mass, "party_guest", "Party Guest")
56
57 assert user is created_user
58 mass.webserver.auth.create_user.assert_awaited_once_with(
59 username="party_guest",
60 role=UserRole.GUEST,
61 display_name="Party Guest",
62 )
63
64
65async def test_get_or_create_join_code_reuses_active_code() -> None:
66 """An active join code is reused instead of generating a new one."""
67 mass = _create_mock_mass()
68 mass.webserver.auth.get_active_join_code.return_value = "ABC123"
69
70 code = await guest_access.get_or_create_join_code(mass, MagicMock())
71
72 assert code == "ABC123"
73 mass.webserver.auth.generate_join_code.assert_not_awaited()
74
75
76async def test_get_or_create_join_code_generates_new_code() -> None:
77 """A new join code is generated when no active code exists."""
78 mass = _create_mock_mass()
79 mass.webserver.auth.generate_join_code.return_value = ("XYZ789", MagicMock())
80 user = MagicMock()
81
82 code = await guest_access.get_or_create_join_code(
83 mass, user, expires_in_hours=4, max_uses=5, device_name="Quiz Guest"
84 )
85
86 assert code == "XYZ789"
87 mass.webserver.auth.generate_join_code.assert_awaited_once_with(
88 user=user,
89 expires_in_hours=4,
90 max_uses=5,
91 device_name="Quiz Guest",
92 )
93
94
95def test_build_join_url_remote_access() -> None:
96 """The remote access URL is used when remote access is enabled."""
97 mass = _create_mock_mass()
98 mass.webserver.remote_access.is_enabled = True
99 mass.webserver.remote_access.remote_id = "remote123"
100
101 url = guest_access.build_join_url(mass, "ABC123")
102
103 assert url == "https://app.music-assistant.io/?remote_id=remote123&join=ABC123"
104
105
106def test_build_join_url_local() -> None:
107 """The local base URL is used when remote access is disabled."""
108 mass = _create_mock_mass()
109 mass.webserver.remote_access.is_enabled = False
110 mass.webserver.base_url = "http://192.168.1.2:8095"
111
112 url = guest_access.build_join_url(mass, "ABC123")
113
114 assert url == "http://192.168.1.2:8095/?join=ABC123"
115
116
117async def test_revoke_guest_access() -> None:
118 """Join codes and tokens of the guest user are revoked."""
119 mass = _create_mock_mass()
120 user = MagicMock()
121 mass.webserver.auth.get_user_by_username.return_value = user
122 mass.webserver.auth.revoke_join_codes.return_value = 2
123 mass.webserver.auth.revoke_tokens_for_user.return_value = 3
124
125 result = await guest_access.revoke_guest_access(mass, "party_guest")
126
127 assert result == (2, 3)
128 mass.webserver.auth.revoke_join_codes.assert_awaited_once_with(user)
129 mass.webserver.auth.revoke_tokens_for_user.assert_awaited_once_with(user)
130
131
132async def test_revoke_guest_access_no_user() -> None:
133 """Revoking access for an unknown user is a no-op."""
134 mass = _create_mock_mass()
135
136 result = await guest_access.revoke_guest_access(mass, "unknown_guest")
137
138 assert result == (0, 0)
139 mass.webserver.auth.revoke_join_codes.assert_not_awaited()
140
141
142def test_credential_owner_encodes_the_lifetime_policy() -> None:
143 """The owner prefix says whether a credential is session-scoped or account-bound."""
144 guest = MagicMock(role=UserRole.GUEST, user_id="g1")
145 user = MagicMock(role=UserRole.USER, user_id="u1")
146 assert guest_access.credential_owner(guest) == "guest-g1"
147 assert guest_access.credential_owner(user) == "user-u1"
148 assert guest_access.is_session_scoped_owner("guest-g1")
149 assert not guest_access.is_session_scoped_owner("user-u1")
150 assert guest_access.credential_owners_for_user_id("x") == ("guest-x", "user-x")
151
152
153def test_credential_owner_user_id_resolves_both_prefixes() -> None:
154 """An owner id resolves back to its account, and other owner kinds resolve to nothing."""
155 assert guest_access.credential_owner_user_id("guest-g1") == "g1"
156 assert guest_access.credential_owner_user_id("user-u1") == "u1"
157 assert guest_access.credential_owner_user_id("token-t1") is None
158