/
/
/
1"""Tests for the state the DLNA player reads from its device."""
2
3from __future__ import annotations
4
5import asyncio
6import time
7from collections.abc import Coroutine
8from datetime import UTC, datetime, timedelta
9from typing import Any
10from unittest.mock import MagicMock
11
12import pytest
13from async_upnp_client.profiles.dlna import TransportState
14from music_assistant_models.enums import PlaybackState
15
16from music_assistant.providers.dlna.player import DLNAPlayer
17from tests.common import MockProvider
18
19REPORTED_AT = datetime(2026, 8, 7, 12, 0, 0, tzinfo=UTC)
20STALE_POSITION = 120.0
21STALE_REPORTED_AT = 1000.0
22
23
24def _mock_device(**device_state: Any) -> MagicMock:
25 """
26 Return a mocked device that fully reports its state.
27
28 :param device_state: Attributes to override on the device.
29 """
30 device = MagicMock()
31 device.profile_device.available = True
32 device.name = "Living Room Renderer"
33 device.volume_level = 0.5
34 device.is_volume_muted = False
35 device.transport_state = TransportState.PLAYING
36 device.current_track_uri = "http://192.168.1.10/stream.mp3"
37 device.media_title = "Test Title"
38 device.media_artist = "Test Artist"
39 device.media_album_name = "Test Album"
40 device.media_image_url = "http://192.168.1.10/cover.jpg"
41 device.media_duration = 240
42 device.media_position = 42
43 device.media_position_updated_at = REPORTED_AT
44 for name, value in device_state.items():
45 setattr(device, name, value)
46 return device
47
48
49def _player(device: MagicMock) -> DLNAPlayer:
50 """
51 Return a player for the given device that already knows a position.
52
53 :param device: The mocked device to attach to the player.
54 """
55 provider = MockProvider("dlna", instance_id="dlna_test")
56 provider.mass.streams.base_url = "http://192.168.1.2:8097"
57
58 player = DLNAPlayer(
59 provider, # type: ignore[arg-type]
60 "uuid:dlna-player",
61 "http://192.168.1.10/description.xml",
62 device=device,
63 )
64 player._attr_elapsed_time = STALE_POSITION
65 player._attr_elapsed_time_last_updated = STALE_REPORTED_AT
66 return player
67
68
69async def _updated_player(**device_state: Any) -> DLNAPlayer:
70 """
71 Run a state update on a player that already knows a position, and return it.
72
73 :param device_state: Attributes to override on the fully reporting mocked device.
74 """
75 player = _player(_mock_device(**device_state))
76 await player.set_dynamic_attributes()
77 return player
78
79
80async def test_zero_position_replaces_the_previous_one() -> None:
81 """A device restarting a track reports 0, which must not be read as 'unknown'."""
82 player = await _updated_player(media_position=0)
83
84 assert player.elapsed_time == 0.0
85 assert player.elapsed_time_last_updated == REPORTED_AT.timestamp()
86
87
88async def test_known_position_is_applied() -> None:
89 """A position reported by the device is adopted together with its timestamp."""
90 player = await _updated_player(media_position=42)
91
92 assert player.elapsed_time == 42.0
93 assert player.elapsed_time_last_updated == REPORTED_AT.timestamp()
94
95
96async def test_missing_position_keeps_the_previous_one() -> None:
97 """A device that reports no position at all leaves the known position alone."""
98 player = await _updated_player(media_position=None)
99
100 assert player.elapsed_time == STALE_POSITION
101 assert player.elapsed_time_last_updated == STALE_REPORTED_AT
102
103
104async def test_position_from_before_a_resume_is_not_extrapolated() -> None:
105 """
106 A position reported from before a resume is anchored at the resume.
107
108 A device does not re-stamp a position that did not change, so the timestamp it
109 reports right after a resume still dates from before the pause.
110 """
111 paused_at = datetime.now(UTC) - timedelta(minutes=10)
112 device = _mock_device(
113 transport_state=TransportState.PAUSED_PLAYBACK,
114 media_position=200,
115 media_position_updated_at=paused_at,
116 )
117 player = _player(device)
118 await player.set_dynamic_attributes()
119
120 device.transport_state = TransportState.PLAYING
121 resumed_at = time.time()
122 await player.set_dynamic_attributes()
123
124 assert player.elapsed_time == 200.0
125 assert player.elapsed_time_last_updated is not None
126 assert player.elapsed_time_last_updated >= resumed_at
127
128
129async def test_position_of_a_player_found_while_playing_keeps_its_own_anchor() -> None:
130 """Without having seen playback start, the timestamp the device reports is all there is."""
131 player = await _updated_player(media_position=200, media_position_updated_at=REPORTED_AT)
132
133 assert player.elapsed_time == 200.0
134 assert player.elapsed_time_last_updated == REPORTED_AT.timestamp()
135
136
137async def test_optimistic_playing_state_does_not_hide_the_start_of_playback() -> None:
138 """
139 Playback starting is tracked from what the device reports, not from what MA assumes.
140
141 A play command marks the player as playing before the device confirms it, which must
142 not be mistaken for the player having been playing all along.
143 """
144 device = _mock_device(
145 transport_state=TransportState.STOPPED,
146 media_position=200,
147 media_position_updated_at=datetime.now(UTC) - timedelta(minutes=10),
148 )
149 player = _player(device)
150 await player.set_dynamic_attributes()
151
152 player._attr_playback_state = PlaybackState.PLAYING
153 device.transport_state = TransportState.PLAYING
154 started_at = time.time()
155 await player.set_dynamic_attributes()
156
157 assert player.elapsed_time_last_updated is not None
158 assert player.elapsed_time_last_updated >= started_at
159
160
161async def test_transport_state_event_polls_before_reading_the_position() -> None:
162 """A player that starts playing reports the position of the track it just started."""
163 device = _mock_device(
164 transport_state=TransportState.STOPPED,
165 media_position=200,
166 media_position_updated_at=datetime.now(UTC) - timedelta(minutes=10),
167 )
168
169 async def _async_update(**_kwargs: Any) -> None:
170 """Answer the poll a round trip later with the position of the new track."""
171 await asyncio.sleep(0)
172 device.transport_state = TransportState.PLAYING
173 device.media_position = 0
174 device.media_position_updated_at = datetime.now(UTC)
175
176 device.async_update = _async_update
177
178 player = _player(device)
179 tasks: list[asyncio.Task[Any]] = []
180
181 def _create_task(target: Coroutine[Any, Any, Any], **_kwargs: Any) -> asyncio.Task[Any]:
182 # eager, like the real helper: the stale read happened because the update task
183 # ran up to its first suspension before the poll task got its answer
184 task: asyncio.Task[Any] = asyncio.Task(
185 target, loop=asyncio.get_running_loop(), eager_start=True
186 )
187 tasks.append(task)
188 return task
189
190 player.mass.create_task = _create_task # type: ignore[assignment]
191
192 service = MagicMock()
193 service.service_id = "urn:upnp-org:serviceId:AVTransport"
194 state_variable = MagicMock()
195 state_variable.name = "TransportState"
196 state_variable.value = TransportState.PLAYING
197
198 started_at = time.time()
199 player._handle_event(service, [state_variable])
200 await asyncio.gather(*tasks)
201
202 assert player.elapsed_time == 0.0
203 assert player.elapsed_time_last_updated is not None
204 assert player.elapsed_time_last_updated >= started_at
205
206
207@pytest.mark.parametrize(("reported", "expected"), [(0.5, 50), (0.0, 0), (1.0, 100)])
208async def test_volume_level_is_scaled_to_percent(reported: float, expected: int) -> None:
209 """The device reports volume as a 0..1 fraction, the player as a percentage."""
210 player = await _updated_player(volume_level=reported)
211
212 assert player.volume_level == expected
213
214
215async def test_unknown_volume_level_stays_unknown() -> None:
216 """A device that does not report its volume must not read as volume 0."""
217 player = await _updated_player(volume_level=None)
218
219 assert player.volume_level is None
220
221
222@pytest.mark.parametrize("reported", [True, False])
223async def test_mute_state_is_applied(reported: bool) -> None:
224 """A mute state reported by the device is adopted as-is."""
225 player = await _updated_player(is_volume_muted=reported)
226
227 assert player.volume_muted is reported
228
229
230async def test_unknown_mute_state_stays_unknown() -> None:
231 """A device that does not report its mute state must not read as unmuted."""
232 player = await _updated_player(is_volume_muted=None)
233
234 assert player.volume_muted is None
235