/
/
/
1"""Test YouTube Music Provider."""
2
3from unittest.mock import AsyncMock, MagicMock
4
5import pytest
6from aiohttp import ClientError, ServerDisconnectedError
7
8from music_assistant.providers.ytmusic import YoutubeMusicProvider
9
10
11@pytest.fixture
12def provider() -> YoutubeMusicProvider:
13 """Return a YoutubeMusicProvider instance with mocked dependencies."""
14 mass = AsyncMock()
15 mass.http_session = MagicMock()
16 manifest = MagicMock()
17 manifest.domain = "ytmusic"
18 config = MagicMock()
19 config.get_value.return_value = "GLOBAL"
20 prov = YoutubeMusicProvider(mass, manifest, config)
21 prov._po_token_server_url = "http://localhost:4416"
22 return prov
23
24
25def _ping_context_manager(
26 *, response: MagicMock | None = None, exc: Exception | None = None
27) -> MagicMock:
28 """Build a fake async context manager mimicking aiohttp's session.get()."""
29 cm = MagicMock()
30 cm.__aenter__ = AsyncMock(return_value=response, side_effect=exc)
31 cm.__aexit__ = AsyncMock(return_value=False)
32 return cm
33
34
35async def test_verify_po_token_url_success(provider: YoutubeMusicProvider) -> None:
36 """A healthy PO Token server (HTTP 200) verifies successfully."""
37 response = MagicMock()
38 response.status = 200
39 response.raise_for_status = MagicMock()
40 provider.mass.http_session.get = MagicMock( # type: ignore[method-assign]
41 return_value=_ping_context_manager(response=response)
42 )
43 assert await provider._verify_po_token_url() is True
44
45
46@pytest.mark.parametrize(
47 "exc",
48 [
49 # boot race: the POT container's port accepts TCP before the server is serving,
50 # so the ping fails with ServerDisconnectedError (a ClientError, but NOT a
51 # ClientConnectorError, which is all the provider used to catch).
52 ServerDisconnectedError(),
53 # connection refused / DNS failure etc. (the originally-handled family).
54 ClientError("connection error"),
55 # an explicit/implicit request timeout.
56 TimeoutError(),
57 ],
58)
59async def test_verify_po_token_url_transient_failure(
60 provider: YoutubeMusicProvider, exc: Exception
61) -> None:
62 """Transient PO Token server errors return False instead of escaping uncaught."""
63 provider.mass.http_session.get = MagicMock( # type: ignore[method-assign]
64 return_value=_ping_context_manager(exc=exc)
65 )
66 assert await provider._verify_po_token_url() is False
67