/
/
1"""Test YouTube Music Provider."""
2
3from unittest.mock import AsyncMock, MagicMock, patch
4
5import pytest
6from aiohttp import ClientError, ServerDisconnectedError
7from music_assistant_models.enums import MediaType
8from music_assistant_models.errors import LoginFailed
9
10from music_assistant.models.music_provider import MusicProvider
11from music_assistant.providers.ytmusic import YoutubeMusicProvider
12
13
14@pytest.fixture
15def provider() -> YoutubeMusicProvider:
16 """Return a YoutubeMusicProvider instance with mocked dependencies."""
17 mass = AsyncMock()
18 mass.http_session = MagicMock()
19 manifest = MagicMock()
20 manifest.domain = "ytmusic"
21 config = MagicMock()
22 config.get_value.return_value = "GLOBAL"
23 prov = YoutubeMusicProvider(mass, manifest, config)
24 prov._po_token_server_url = "http://localhost:4416"
25 return prov
26
27
28def _ping_context_manager(
29 *, response: MagicMock | None = None, exc: Exception | None = None
30) -> MagicMock:
31 """Build a fake async context manager mimicking aiohttp's session.get()."""
32 cm = MagicMock()
33 cm.__aenter__ = AsyncMock(return_value=response, side_effect=exc)
34 cm.__aexit__ = AsyncMock(return_value=False)
35 return cm
36
37
38async def test_verify_po_token_url_success(provider: YoutubeMusicProvider) -> None:
39 """A healthy PO Token server (HTTP 200) verifies successfully."""
40 response = MagicMock()
41 response.status = 200
42 response.raise_for_status = MagicMock()
43 provider.mass.http_session.get = MagicMock( # type: ignore[method-assign]
44 return_value=_ping_context_manager(response=response)
45 )
46 assert await provider._verify_po_token_url() is True
47
48
49@pytest.mark.parametrize(
50 "exc",
51 [
52 # boot race: the POT container's port accepts TCP before the server is serving,
53 # so the ping fails with ServerDisconnectedError (a ClientError, but NOT a
54 # ClientConnectorError, which is all the provider used to catch).
55 ServerDisconnectedError(),
56 # connection refused / DNS failure etc. (the originally-handled family).
57 ClientError("connection error"),
58 # an explicit/implicit request timeout.
59 TimeoutError(),
60 ],
61)
62async def test_verify_po_token_url_transient_failure(
63 provider: YoutubeMusicProvider, exc: Exception
64) -> None:
65 """Transient PO Token server errors return False instead of escaping uncaught."""
66 provider.mass.http_session.get = MagicMock( # type: ignore[method-assign]
67 return_value=_ping_context_manager(exc=exc)
68 )
69 assert await provider._verify_po_token_url() is False
70
71
72async def test_sync_library_unloads_on_invalid_session(provider: YoutubeMusicProvider) -> None:
73 """A library sync that hits an invalid session unloads the provider for re-auth."""
74 provider.available = True
75 provider.unload_with_error = MagicMock() # type: ignore[method-assign]
76 err = LoginFailed("Your YouTube Music session is no longer valid.")
77 with (
78 patch.object(MusicProvider, "sync_library", AsyncMock(side_effect=err)),
79 pytest.raises(LoginFailed),
80 ):
81 await provider.sync_library(MediaType.PLAYLIST)
82 provider.unload_with_error.assert_called_once_with(err)
83
84
85async def test_sync_library_keeps_other_errors_silent(provider: YoutubeMusicProvider) -> None:
86 """Any other sync failure must not unload the provider."""
87 provider.available = True
88 provider.unload_with_error = MagicMock() # type: ignore[method-assign]
89 with (
90 patch.object(MusicProvider, "sync_library", AsyncMock(side_effect=KeyError("boom"))),
91 pytest.raises(KeyError),
92 ):
93 await provider.sync_library(MediaType.PLAYLIST)
94 provider.unload_with_error.assert_not_called()
95
96
97def test_parse_owned_playlist_is_editable_without_privacy(
98 provider: YoutubeMusicProvider,
99) -> None:
100 """An owned playlist is editable even when the library response omits privacy."""
101 playlist = provider._parse_playlist(
102 {
103 "id": "PL_owned",
104 "title": "Owned playlist",
105 "owned": True,
106 }
107 )
108
109 assert playlist.is_editable is True
110