/
/
/
1"""
2End-to-end test for enqueueing a dynamic radio station through the real queue controller.
3
4Boots the hermetic `e2e_mass` fixture (fake `test` provider + demo players) plus a small
5fake provider exposing one dynamic ("Pandora-style") station, and exercises the same
6dynamic-mode transition covered for dynamic playlists in test_dynamic_mode_e2e.py: enqueueing
7the station records it as a dynamic source, puts the queue into dynamic mode, and fills the
8managed pool from the provider's ``get_dynamic_radio_tracks``.
9"""
10
11from __future__ import annotations
12
13from typing import TYPE_CHECKING, cast
14
15import pytest
16from music_assistant_models.config_entries import ProviderConfig
17from music_assistant_models.enums import MediaType, ProviderType, QueueOption
18from music_assistant_models.media_items import ProviderMapping, Radio, Track
19from music_assistant_models.provider import ProviderManifest
20
21from music_assistant.mass import MusicAssistant
22from music_assistant.models.music_provider import MusicProvider
23
24from .conftest import demo_players, wait_for
25
26if TYPE_CHECKING:
27 from collections.abc import AsyncGenerator
28
29FAKE_RADIO_DOMAIN = "fake_dynamic_radio_e2e"
30FAKE_RADIO_INSTANCE = "fake_dynamic_radio_e2e--instance"
31DYNAMIC_STATION_ID = "station-1"
32STATION_BATCH_SIZE = 10
33
34
35class FakeDynamicRadioProvider(MusicProvider):
36 """Provider owning a single dynamic ("Pandora-style") radio station."""
37
38 async def sync_library(self, media_type: MediaType) -> None:
39 """No-op sync implementation for tests."""
40
41 async def get_radio(self, prov_radio_id: str) -> Radio:
42 """Return the fake dynamic station."""
43 return Radio(
44 item_id=prov_radio_id,
45 provider=self.instance_id,
46 name="Dynamic Station",
47 is_dynamic=True,
48 provider_mappings={
49 ProviderMapping(
50 item_id=prov_radio_id,
51 provider_domain=self.domain,
52 provider_instance=self.instance_id,
53 )
54 },
55 )
56
57 async def get_dynamic_radio_tracks(self, prov_radio_id: str) -> list[Track]:
58 """Return a fixed batch of fake tracks for the dynamic station."""
59 return [
60 Track(
61 item_id=f"{prov_radio_id}-t{i}",
62 provider=self.instance_id,
63 name=f"Station Track {i}",
64 duration=60,
65 provider_mappings={
66 ProviderMapping(
67 item_id=f"{prov_radio_id}-t{i}",
68 provider_domain=self.domain,
69 provider_instance=self.instance_id,
70 )
71 },
72 )
73 for i in range(STATION_BATCH_SIZE)
74 ]
75
76
77@pytest.fixture
78async def e2e_mass_with_dynamic_radio(
79 e2e_mass: MusicAssistant,
80) -> AsyncGenerator[MusicAssistant]:
81 """Register the fake dynamic-radio provider on top of the hermetic e2e instance."""
82 config = ProviderConfig(
83 values={},
84 type=ProviderType.MUSIC,
85 domain=FAKE_RADIO_DOMAIN,
86 instance_id=FAKE_RADIO_INSTANCE,
87 name="Fake Dynamic Radio",
88 )
89 provider = FakeDynamicRadioProvider(
90 e2e_mass,
91 manifest=ProviderManifest(
92 type=ProviderType.MUSIC,
93 domain=FAKE_RADIO_DOMAIN,
94 name="Fake Dynamic Radio",
95 description="Fake dynamic radio provider",
96 codeowners=["@music-assistant"],
97 ),
98 config=config,
99 supported_features=set(),
100 )
101 provider.available = True
102 e2e_mass._providers[FAKE_RADIO_INSTANCE] = provider
103 # the global "available providers" cache drives MediaItem.available; refresh it so the
104 # fake station's tracks are not filtered out of the managed pool as unavailable
105 await e2e_mass._update_available_providers_cache()
106 try:
107 yield e2e_mass
108 finally:
109 e2e_mass._providers.pop(FAKE_RADIO_INSTANCE, None)
110 await e2e_mass._update_available_providers_cache()
111
112
113@pytest.mark.asyncio
114async def test_enqueue_dynamic_radio_enters_dynamic_mode(
115 e2e_mass_with_dynamic_radio: MusicAssistant,
116) -> None:
117 """Enqueueing a dynamic radio station puts the queue in dynamic mode and fills the pool."""
118 mass = e2e_mass_with_dynamic_radio
119 queue_id = demo_players(mass)[0].player_id
120 provider = cast("MusicProvider", mass.get_provider(FAKE_RADIO_INSTANCE))
121 assert provider is not None
122 station = await provider.get_radio(DYNAMIC_STATION_ID)
123
124 # ADD keeps this off the playback path (no streamdetails needed for the fake tracks),
125 # mirroring how test_dynamic_mode_e2e.py adds a dynamic playlist without starting playback.
126 await mass.player_queues.play_media(queue_id, station, option=QueueOption.ADD)
127
128 assert await wait_for(lambda: len(mass.player_queues.items(queue_id)) > 0), (
129 "dynamic radio pool never populated"
130 )
131 queue = mass.player_queues.get(queue_id)
132 assert queue is not None
133 assert queue.is_dynamic is True
134
135 data = mass.player_queues._queue_data[queue_id]
136 assert any(
137 isinstance(item, Radio) and item.is_dynamic and item.item_id == DYNAMIC_STATION_ID
138 for item in data.source_items
139 )
140
141 # the pool was filled straight from the provider's get_dynamic_radio_tracks batch
142 items = mass.player_queues.items(queue_id)
143 item_ids = {item.media_item.item_id for item in items if item.media_item}
144 assert item_ids <= {f"{DYNAMIC_STATION_ID}-t{i}" for i in range(STATION_BATCH_SIZE)}
145 assert item_ids
146