/
/
/
1"""
2Unit tests for KionMusicProvider (provider.py).
3
4These tests construct a partial provider instance via ``__new__`` (no
5``__init__``), attach the attributes the method-under-test reads, and
6exercise it directly, so the upstream provider-init machinery does not run.
7The cache decorator does need a server, so those tests attach the minimal
8``MusicAssistant`` instance from the ``mass_minimal`` fixture.
9"""
10
11from __future__ import annotations
12
13import asyncio
14from typing import TYPE_CHECKING, Any
15from unittest import mock
16
17import pytest
18
19from music_assistant.providers.kion_music.constants import MY_WAVE_PLAYLIST_ID
20from music_assistant.providers.kion_music.provider import KionMusicProvider
21
22if TYPE_CHECKING:
23 from collections.abc import Callable
24
25 from music_assistant.mass import MusicAssistant
26
27
28class _StubConfig:
29 """Minimal provider config for the @use_cache decorator."""
30
31 instance_id = "kion_music_test"
32
33 def get_value(self, key: str, default: Any = None) -> Any:
34 """Return the default for every config key."""
35 return default
36
37
38@pytest.fixture
39async def cached_provider(
40 mass_minimal: MusicAssistant,
41) -> tuple[KionMusicProvider, mock.AsyncMock]:
42 """Return a provider with a mocked API client, backed by a real (empty) cache."""
43 await mass_minimal.cache._setup_database()
44 provider = KionMusicProvider.__new__(KionMusicProvider)
45 mock_client = mock.AsyncMock()
46 mock_client.user_id = 12345
47 provider._client = mock_client
48 provider.logger = mock.MagicMock()
49 provider.mass = mass_minimal
50 provider.config = _StubConfig() # type: ignore[assignment]
51 provider.manifest = mock.MagicMock(domain="kion_music")
52 provider._my_wave_lock = asyncio.Lock()
53 provider._my_wave_seen_track_ids = set()
54 provider._my_wave_radio_started_sent = False
55 provider._my_wave_playlist_next_cursor = None
56 return provider, mock_client
57
58
59async def _wait_for_gated_fetch(started: Callable[[], bool]) -> None:
60 """Wait until the gated fetch runs, then let the other callers catch up with it."""
61 for _ in range(200):
62 if started():
63 break
64 await asyncio.sleep(0.01)
65 else:
66 pytest.fail("gated fetch never started")
67 # a caller arriving after the gate is released would start a second fetch,
68 # which the await-count assertions below catch
69 await asyncio.sleep(0.05)
70
71
72async def test_regular_playlist_fetch_is_shared_between_callers(
73 cached_provider: tuple[KionMusicProvider, mock.AsyncMock],
74) -> None:
75 """Concurrent callers for the same regular playlist share one provider fetch."""
76 provider, mock_client = cached_provider
77 gate = asyncio.Event()
78
79 async def _get_playlist(*_args: Any, **_kwargs: Any) -> Any:
80 await gate.wait()
81 return type("PL", (), {"tracks": [], "track_count": 0})()
82
83 mock_client.get_playlist = mock.AsyncMock(side_effect=_get_playlist)
84
85 tasks = [asyncio.create_task(provider.get_playlist_tracks("12345:67")) for _ in range(3)]
86 await _wait_for_gated_fetch(lambda: mock_client.get_playlist.await_count > 0)
87 gate.set()
88
89 assert await asyncio.gather(*tasks) == [[], [], []]
90 assert mock_client.get_playlist.await_count == 1
91
92
93async def test_my_mix_fetch_is_shared_between_callers(
94 cached_provider: tuple[KionMusicProvider, mock.AsyncMock],
95) -> None:
96 """Concurrent My Mix callers share one fetch, so the rotor advances once."""
97 provider, mock_client = cached_provider
98 gate = asyncio.Event()
99
100 async def _get_my_wave_tracks(*_args: Any, **_kwargs: Any) -> tuple[list[Any], None]:
101 await gate.wait()
102 return [], None
103
104 mock_client.get_my_wave_tracks = mock.AsyncMock(side_effect=_get_my_wave_tracks)
105
106 tasks = [
107 asyncio.create_task(provider.get_playlist_tracks(MY_WAVE_PLAYLIST_ID)) for _ in range(3)
108 ]
109 await _wait_for_gated_fetch(lambda: mock_client.get_my_wave_tracks.await_count > 0)
110 gate.set()
111
112 assert await asyncio.gather(*tasks) == [[], [], []]
113 assert mock_client.get_my_wave_tracks.await_count == 1
114