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