/
/
/
1"""Tests that a duration determined while streaming is applied to the item that is playing."""
2
3from __future__ import annotations
4
5from typing import Any, NamedTuple
6from unittest.mock import AsyncMock, MagicMock, Mock
7
8from music_assistant_models.enums import ContentType, MediaType, StreamType
9from music_assistant_models.media_items import (
10 AudioFormat,
11 ItemMapping,
12 PodcastEpisode,
13 ProviderMapping,
14 Radio,
15)
16from music_assistant_models.queue_item import QueueItem
17from music_assistant_models.streamdetails import StreamDetails
18
19from music_assistant.controllers.player_queues import PlayerQueuesController
20
21QUEUE_ID = "q1"
22EPISODE_URI = "overcast--1://podcast_episode/feed1 ep1"
23
24
25def _await_args(mock: AsyncMock) -> Any:
26 """Return the call a mock was last awaited with, narrowed for the type checker."""
27 assert mock.await_args is not None
28 return mock.await_args
29
30
31class _Harness(NamedTuple):
32 """A bare queue controller with its collaborators kept at their mock types."""
33
34 ctrl: PlayerQueuesController
35 signal_update: Mock
36 create_task: Mock
37 cache_get: AsyncMock
38 cache_set: AsyncMock
39
40 async def stored_duration(self) -> int | None:
41 """Run the fire and forget cache write and return the value it stored."""
42 if not self.create_task.called:
43 return None
44 await self.create_task.call_args[0][0]
45 return int(_await_args(self.cache_set).args[1])
46
47
48def _harness(stored: int | None = None) -> _Harness:
49 ctrl = PlayerQueuesController.__new__(PlayerQueuesController)
50 signal_update = Mock()
51 ctrl.signal_update = signal_update # type: ignore[method-assign]
52 mass = MagicMock()
53 mass.cache.get = AsyncMock(return_value=stored)
54 mass.cache.set = AsyncMock()
55 ctrl.mass = mass
56 return _Harness(ctrl, signal_update, mass.create_task, mass.cache.get, mass.cache.set)
57
58
59def _streamdetails(media_type: MediaType, duration: int | None) -> StreamDetails:
60 return StreamDetails(
61 provider="overcast--1",
62 item_id="feed1 ep1",
63 audio_format=AudioFormat(content_type=ContentType.MP3),
64 media_type=media_type,
65 stream_type=StreamType.HTTP,
66 duration=duration,
67 )
68
69
70def _provider_mappings(item_id: str, domain: str) -> set[ProviderMapping]:
71 return {
72 ProviderMapping(
73 item_id=item_id,
74 provider_domain=domain,
75 provider_instance=f"{domain}--1",
76 )
77 }
78
79
80def _episode(duration: int = 0) -> PodcastEpisode:
81 return PodcastEpisode(
82 item_id="feed1 ep1",
83 provider="overcast--1",
84 name="Episode One",
85 uri=EPISODE_URI,
86 duration=duration,
87 position=1,
88 podcast=ItemMapping(
89 item_id="feed1",
90 provider="overcast--1",
91 name="Feed One",
92 media_type=MediaType.PODCAST,
93 ),
94 provider_mappings=_provider_mappings("feed1 ep1", "overcast"),
95 )
96
97
98def _radio() -> Radio:
99 return Radio(
100 item_id="r1",
101 provider="radiobrowser--1",
102 name="Radio One",
103 uri="radio://r1",
104 provider_mappings=_provider_mappings("r1", "radiobrowser"),
105 )
106
107
108def _queue_item(media_item: Any, streamdetails: StreamDetails, duration: int = 0) -> QueueItem:
109 return QueueItem(
110 queue_id=QUEUE_ID,
111 queue_item_id="qi1",
112 name="Episode One",
113 duration=duration,
114 media_item=media_item,
115 streamdetails=streamdetails,
116 )
117
118
119async def test_probed_duration_is_applied_and_stored() -> None:
120 """A duration from the stream lands on the queue item, its media item and the store."""
121 harness = _harness()
122 episode = _episode()
123 queue_item = _queue_item(episode, _streamdetails(MediaType.PODCAST_EPISODE, 2700))
124
125 harness.ctrl._apply_probed_duration(queue_item)
126
127 assert queue_item.duration == 2700
128 assert episode.duration == 2700
129 harness.signal_update.assert_called_once_with(QUEUE_ID, items_changed=True)
130 assert await harness.stored_duration() == 2700
131 assert _await_args(harness.cache_set).kwargs["persistent"] is True
132
133
134async def test_probed_duration_does_not_overwrite_a_known_duration() -> None:
135 """A duration reported by the provider is left alone, so no rounded value replaces it."""
136 harness = _harness()
137 episode = _episode(duration=2712)
138 queue_item = _queue_item(
139 episode, _streamdetails(MediaType.PODCAST_EPISODE, 2700), duration=2712
140 )
141
142 harness.ctrl._apply_probed_duration(queue_item)
143
144 assert queue_item.duration == 2712
145 assert episode.duration == 2712
146 harness.signal_update.assert_not_called()
147 assert await harness.stored_duration() is None
148
149
150async def test_probed_duration_is_ignored_for_radio() -> None:
151 """Radio never gets a duration, which would present it as seekable."""
152 harness = _harness()
153 queue_item = _queue_item(_radio(), _streamdetails(MediaType.RADIO, 2700))
154
155 harness.ctrl._apply_probed_duration(queue_item)
156
157 assert queue_item.duration == 0
158 harness.signal_update.assert_not_called()
159 assert await harness.stored_duration() is None
160
161
162async def test_probed_duration_without_a_result_does_nothing() -> None:
163 """A stream whose duration could not be determined leaves the item as it was."""
164 harness = _harness()
165 episode = _episode()
166 queue_item = _queue_item(episode, _streamdetails(MediaType.PODCAST_EPISODE, None))
167
168 harness.ctrl._apply_probed_duration(queue_item)
169
170 assert queue_item.duration == 0
171 assert episode.duration == 0
172 harness.signal_update.assert_not_called()
173
174
175async def test_stored_duration_is_restored_on_a_stale_item() -> None:
176 """An item played from a listing fetched before its duration was known still gets one."""
177 harness = _harness(stored=2700)
178 episode = _episode()
179 queue_item = _queue_item(episode, _streamdetails(MediaType.PODCAST_EPISODE, None))
180
181 await harness.ctrl._restore_probed_duration(queue_item)
182
183 assert queue_item.duration == 2700
184 assert episode.duration == 2700
185 assert _await_args(harness.cache_get).args[0] == EPISODE_URI
186 harness.signal_update.assert_called_once_with(QUEUE_ID, items_changed=True)
187
188
189async def test_stored_duration_is_not_looked_up_for_radio() -> None:
190 """Radio is never given a stored duration, nor is the store consulted for it."""
191 harness = _harness(stored=2700)
192 queue_item = _queue_item(_radio(), _streamdetails(MediaType.RADIO, None))
193
194 await harness.ctrl._restore_probed_duration(queue_item)
195
196 assert queue_item.duration == 0
197 harness.cache_get.assert_not_awaited()
198