music-assistant-server

1"""Tests for WiiM player provider."""
2
3from unittest.mock import AsyncMock, MagicMock
4
5import pytest
6from music_assistant_models.enums import PlaybackState, PlayerFeature
7from wiim import PlayingStatus
8from wiim.exceptions import WiimDeviceException, WiimRequestException
9
10from music_assistant.models.player import PlayerMedia
11from music_assistant.providers.wiim.constants import SOURCE_NETWORK
12from music_assistant.providers.wiim.player import SDK_TO_MA_STATE, WiimPlayer
13
14
15@pytest.fixture
16def mock_wiim_device() -> MagicMock:
17    """Create a mock WiimDevice."""
18    device = MagicMock()
19    device.name = "Test WiiM Pro"
20    device.udn = "uuid:test-wiim-001"
21    device.available = True
22    device.volume = 50
23    device.is_muted = False
24    device.playing_status = None
25    device.play_mode = None
26    device.current_media = None
27    device.model_name = "WiiM Pro"
28    device.manufacturer = "Linkplay"
29    device.firmware_version = "4.8.1"
30    device.ip_address = "192.168.1.100"
31    device.supports_http_api = True
32    device.supported_input_modes = ("Network", "Bluetooth", "Line In", "Optical In")
33    device.async_play = AsyncMock()
34    device.async_pause = AsyncMock()
35    device.async_stop = AsyncMock()
36    device.async_set_volume = AsyncMock()
37    device.async_set_mute = AsyncMock()
38    device.async_set_play_mode = AsyncMock()
39    device.sync_device_duration_and_position = AsyncMock()
40    device.disconnect = AsyncMock()
41    device.ensure_subscriptions = AsyncMock()
42    device.general_event_callback = None
43    device.rendering_control_event_callback = None
44    device.av_transport_event_callback = None
45    device.play_queue_event_callback = None
46    return device
47
48
49@pytest.fixture
50def mock_controller() -> MagicMock:
51    """Create a mock WiimController."""
52    controller = MagicMock()
53    snapshot = MagicMock()
54    snapshot.role = "standalone"
55    snapshot.leader_udn = "uuid:test-wiim-001"
56    snapshot.member_udns = ("uuid:test-wiim-001",)
57    controller.get_group_snapshot.return_value = snapshot
58    controller.get_group_members.return_value = []
59    controller.get_device.return_value = MagicMock()
60    controller.async_join_group = AsyncMock()
61    controller.async_ungroup_device = AsyncMock()
62    return controller
63
64
65@pytest.fixture
66def mock_provider(mock_controller: MagicMock) -> MagicMock:
67    """Create a mock WiimProvider."""
68    provider = MagicMock()
69    provider.wiim_controller = mock_controller
70    provider.instance_id = "wiim_test"
71    provider.domain = "wiim"
72    provider.manifest = MagicMock()
73    provider.manifest.domain = "wiim"
74    provider.mass = MagicMock()
75    provider.mass.players = MagicMock()
76
77    config = MagicMock()
78    config.name = None
79    config.default_name = "Test WiiM Pro"
80    config.enabled = True
81    config.player_type = None
82    config.get_value = MagicMock(return_value=None)
83    provider.mass.config.get_base_player_config.return_value = config
84    return provider
85
86
87class TestSDKStateMapping:
88    """Test SDK to MA state mapping."""
89
90    def test_playing_maps_to_playing(self) -> None:
91        """PLAYING should map to PlaybackState.PLAYING."""
92        assert SDK_TO_MA_STATE[PlayingStatus.PLAYING] == PlaybackState.PLAYING
93
94    def test_paused_maps_to_paused(self) -> None:
95        """PAUSED should map to PlaybackState.PAUSED."""
96        assert SDK_TO_MA_STATE[PlayingStatus.PAUSED] == PlaybackState.PAUSED
97
98    def test_stopped_maps_to_idle(self) -> None:
99        """STOPPED should map to PlaybackState.IDLE."""
100        assert SDK_TO_MA_STATE[PlayingStatus.STOPPED] == PlaybackState.IDLE
101
102    def test_loading_maps_to_playing(self) -> None:
103        """LOADING should map to PlaybackState.PLAYING."""
104        assert SDK_TO_MA_STATE[PlayingStatus.LOADING] == PlaybackState.PLAYING
105
106    def test_all_sdk_states_mapped(self) -> None:
107        """All non-UNKNOWN SDK states should have a mapping."""
108        for status in PlayingStatus:
109            if status != PlayingStatus.UNKNOWN:
110                assert status in SDK_TO_MA_STATE, f"{status} not mapped"
111
112
113class TestFalsePlayingFilter:
114    """A uri-less PLAYING report in network mode must not become PLAYING state."""
115
116    def _make_player(self, provider: MagicMock, device: MagicMock) -> WiimPlayer:
117        player = WiimPlayer(provider=provider, player_id="uuid:test", device=device)
118        player.update_state = MagicMock()  # type: ignore[misc,method-assign]
119        return player
120
121    def test_false_playing_ack_is_suppressed(
122        self, mock_provider: MagicMock, mock_wiim_device: MagicMock
123    ) -> None:
124        """
125        The transient PLAYING ack without media loaded must keep the previous state.
126
127        The device acks (group) transport commands with a short false PLAYING
128        report before any track is loaded; propagating it causes a
129        PLAYING->IDLE->PLAYING flicker downstream.
130        """
131        mock_wiim_device.play_mode = SOURCE_NETWORK
132        mock_wiim_device.current_media = None
133        mock_wiim_device.playing_status = PlayingStatus.PLAYING
134        player = self._make_player(mock_provider, mock_wiim_device)
135        player._attr_playback_state = PlaybackState.IDLE
136
137        player._update_ma_state_from_sdk_cache()
138
139        assert player._attr_playback_state == PlaybackState.IDLE
140
141    def test_loading_without_uri_is_suppressed(
142        self, mock_provider: MagicMock, mock_wiim_device: MagicMock
143    ) -> None:
144        """LOADING maps to PLAYING and gets the same uri-less filter."""
145        mock_wiim_device.play_mode = SOURCE_NETWORK
146        mock_wiim_device.current_media = None
147        mock_wiim_device.playing_status = PlayingStatus.LOADING
148        player = self._make_player(mock_provider, mock_wiim_device)
149        player._attr_playback_state = PlaybackState.IDLE
150
151        player._update_ma_state_from_sdk_cache()
152
153        assert player._attr_playback_state == PlaybackState.IDLE
154
155    def test_playing_kept_when_uri_drops_mid_playback(
156        self, mock_provider: MagicMock, mock_wiim_device: MagicMock
157    ) -> None:
158        """The filter keeps the previous state; it never forces a playing player idle."""
159        mock_wiim_device.play_mode = SOURCE_NETWORK
160        mock_wiim_device.current_media = None
161        mock_wiim_device.playing_status = PlayingStatus.PLAYING
162        player = self._make_player(mock_provider, mock_wiim_device)
163        player._attr_playback_state = PlaybackState.PLAYING
164
165        player._update_ma_state_from_sdk_cache()
166
167        assert player._attr_playback_state == PlaybackState.PLAYING
168
169    def test_playing_accepted_once_uri_present(
170        self, mock_provider: MagicMock, mock_wiim_device: MagicMock
171    ) -> None:
172        """A PLAYING report with media loaded is a real start and passes through."""
173        media = MagicMock()
174        media.uri = "http://192.168.1.80:8097/single/abc/queue/item/uuid:test.flac"
175        mock_wiim_device.play_mode = SOURCE_NETWORK
176        mock_wiim_device.current_media = media
177        mock_wiim_device.playing_status = PlayingStatus.PLAYING
178        player = self._make_player(mock_provider, mock_wiim_device)
179        player._attr_playback_state = PlaybackState.IDLE
180
181        player._update_ma_state_from_sdk_cache()
182
183        assert player._attr_playback_state == PlaybackState.PLAYING
184
185    def test_external_input_playing_without_uri_accepted(
186        self, mock_provider: MagicMock, mock_wiim_device: MagicMock
187    ) -> None:
188        """External inputs legitimately play without a URI and must not be filtered."""
189        mock_wiim_device.play_mode = "Line In"
190        mock_wiim_device.current_media = None
191        mock_wiim_device.playing_status = PlayingStatus.PLAYING
192        player = self._make_player(mock_provider, mock_wiim_device)
193        player._attr_playback_state = PlaybackState.IDLE
194
195        player._update_ma_state_from_sdk_cache()
196
197        assert player._attr_playback_state == PlaybackState.PLAYING
198
199    def test_unknown_play_mode_trusts_device(
200        self, mock_provider: MagicMock, mock_wiim_device: MagicMock
201    ) -> None:
202        """Without a known play mode the device report is trusted (no suppression)."""
203        mock_wiim_device.play_mode = None
204        mock_wiim_device.current_media = None
205        mock_wiim_device.playing_status = PlayingStatus.PLAYING
206        player = self._make_player(mock_provider, mock_wiim_device)
207        player._attr_playback_state = PlaybackState.IDLE
208
209        player._update_ma_state_from_sdk_cache()
210
211        assert player._attr_playback_state == PlaybackState.PLAYING
212
213    def test_stopped_report_unaffected(
214        self, mock_provider: MagicMock, mock_wiim_device: MagicMock
215    ) -> None:
216        """The filter only guards PLAYING-mapped reports; STOPPED passes through."""
217        mock_wiim_device.play_mode = SOURCE_NETWORK
218        mock_wiim_device.current_media = None
219        mock_wiim_device.playing_status = PlayingStatus.STOPPED
220        player = self._make_player(mock_provider, mock_wiim_device)
221        player._attr_playback_state = PlaybackState.PLAYING
222
223        player._update_ma_state_from_sdk_cache()
224
225        assert player._attr_playback_state == PlaybackState.IDLE
226
227
228class TestSupportedFeatures:
229    """Test that required features are declared."""
230
231    def test_play_media_in_features(
232        self, mock_provider: MagicMock, mock_wiim_device: MagicMock
233    ) -> None:
234        """PLAY_MEDIA should be in supported features."""
235        player = WiimPlayer(provider=mock_provider, player_id="uuid:test", device=mock_wiim_device)
236        assert PlayerFeature.PLAY_MEDIA in player._attr_supported_features
237
238    def test_volume_features(self, mock_provider: MagicMock, mock_wiim_device: MagicMock) -> None:
239        """VOLUME_SET and VOLUME_MUTE should be in supported features."""
240        player = WiimPlayer(provider=mock_provider, player_id="uuid:test", device=mock_wiim_device)
241        assert PlayerFeature.VOLUME_SET in player._attr_supported_features
242        assert PlayerFeature.VOLUME_MUTE in player._attr_supported_features
243
244    def test_select_source_in_features(
245        self, mock_provider: MagicMock, mock_wiim_device: MagicMock
246    ) -> None:
247        """SELECT_SOURCE should be in supported features."""
248        player = WiimPlayer(provider=mock_provider, player_id="uuid:test", device=mock_wiim_device)
249        assert PlayerFeature.SELECT_SOURCE in player._attr_supported_features
250
251
252class TestSourceList:
253    """Test dynamic source list construction."""
254
255    @pytest.mark.asyncio
256    async def test_setup_adds_device_input_modes(
257        self, mock_provider: MagicMock, mock_wiim_device: MagicMock
258    ) -> None:
259        """setup() should add sources for device-supported input modes."""
260        player = WiimPlayer(provider=mock_provider, player_id="uuid:test", device=mock_wiim_device)
261        await player.setup()
262        source_ids = [s.id for s in player._attr_source_list]
263        assert "bluetooth" in source_ids
264        assert "line_in" in source_ids
265        assert "optical" in source_ids
266
267    @pytest.mark.asyncio
268    async def test_setup_adds_passive_sources(
269        self, mock_provider: MagicMock, mock_wiim_device: MagicMock
270    ) -> None:
271        """setup() should add passive sources (AirPlay, Spotify)."""
272        player = WiimPlayer(provider=mock_provider, player_id="uuid:test", device=mock_wiim_device)
273        await player.setup()
274        source_ids = [s.id for s in player._attr_source_list]
275        assert "airplay" in source_ids
276        assert "spotify" in source_ids
277
278    @pytest.mark.asyncio
279    async def test_setup_skips_unknown_input_modes(
280        self, mock_provider: MagicMock, mock_wiim_device: MagicMock
281    ) -> None:
282        """setup() should skip input modes not in INPUT_MODE_SOURCES."""
283        mock_wiim_device.supported_input_modes = ("Network", "FutureMode")
284        player = WiimPlayer(provider=mock_provider, player_id="uuid:test", device=mock_wiim_device)
285        await player.setup()
286        source_ids = [s.id for s in player._attr_source_list]
287        assert "futuremode" not in source_ids
288
289
290class TestErrorHandling:
291    """Test that command errors mark device unavailable."""
292
293    @pytest.mark.asyncio
294    async def test_play_error_refreshes_state(
295        self, mock_provider: MagicMock, mock_wiim_device: MagicMock
296    ) -> None:
297        """Play command error should refresh state without marking unavailable."""
298        mock_wiim_device.async_play.side_effect = WiimRequestException("timeout")
299        player = WiimPlayer(provider=mock_provider, player_id="uuid:test", device=mock_wiim_device)
300        player.update_state = MagicMock()  # type: ignore[misc,method-assign]
301        await player.play()
302        assert player._attr_available is True
303        player.update_state.assert_called()
304
305    @pytest.mark.skip(
306        reason="volume_set inlines the HTTP fix from wiim PR#18 and no longer calls "
307        "async_set_volume; re-enable when wiim-sdk 0.1.5+ has been released"
308    )
309    @pytest.mark.asyncio
310    async def test_volume_set_error_refreshes_state(
311        self, mock_provider: MagicMock, mock_wiim_device: MagicMock
312    ) -> None:
313        """Volume set error should refresh state without marking unavailable."""
314        mock_wiim_device.async_set_volume.side_effect = WiimDeviceException("disconnected")
315        player = WiimPlayer(provider=mock_provider, player_id="uuid:test", device=mock_wiim_device)
316        player.update_state = MagicMock()  # type: ignore[misc,method-assign]
317        await player.volume_set(50)
318        assert player._attr_available is True
319        player.update_state.assert_called()
320
321    @pytest.mark.asyncio
322    async def test_stop_error_refreshes_state(
323        self, mock_provider: MagicMock, mock_wiim_device: MagicMock
324    ) -> None:
325        """Stop command error should refresh state without marking unavailable."""
326        mock_wiim_device.async_stop.side_effect = WiimRequestException("connection lost")
327        player = WiimPlayer(provider=mock_provider, player_id="uuid:test", device=mock_wiim_device)
328        player.update_state = MagicMock()  # type: ignore[misc,method-assign]
329        await player.stop()
330        assert player._attr_available is True
331        player.update_state.assert_called()
332
333    @pytest.mark.asyncio
334    async def test_pause_error_refreshes_state(
335        self, mock_provider: MagicMock, mock_wiim_device: MagicMock
336    ) -> None:
337        """Pause command error should refresh state without marking unavailable."""
338        mock_wiim_device.async_pause.side_effect = WiimDeviceException("timeout")
339        player = WiimPlayer(provider=mock_provider, player_id="uuid:test", device=mock_wiim_device)
340        player.update_state = MagicMock()  # type: ignore[misc,method-assign]
341        await player.pause()
342        assert player._attr_available is True
343        player.update_state.assert_called()
344
345
346class TestStalePositionOnNewStream:
347    """A new stream handed to the device must not inherit the previous position."""
348
349    def _make_player(self, provider: MagicMock, device: MagicMock) -> WiimPlayer:
350        player = WiimPlayer(provider=provider, player_id="uuid:test", device=device)
351        player.update_state = MagicMock()  # type: ignore[misc,method-assign]
352        return player
353
354    @pytest.mark.asyncio
355    async def test_play_media_resets_elapsed_time(
356        self, mock_provider: MagicMock, mock_wiim_device: MagicMock
357    ) -> None:
358        """play_media() must clear the stale elapsed_time anchor from prior content."""
359        stream_url = "http://192.168.1.80:8097/single/abc/queue/item/uuid:test.flac"
360        mock_provider.mass.streams.resolve_stream_url = AsyncMock(return_value=stream_url)
361        player = self._make_player(mock_provider, mock_wiim_device)
362        player._attr_elapsed_time = 273
363        player._attr_elapsed_time_last_updated = 1000.0
364
365        await player.play_media(PlayerMedia(uri="library://track/1", title="Some Track"))
366
367        assert player._attr_elapsed_time == 0
368        assert player._attr_elapsed_time_last_updated is not None
369        assert player._attr_elapsed_time_last_updated > 1000.0
370
371    @pytest.mark.asyncio
372    async def test_play_media_then_sync_position_end_to_end(
373        self, mock_provider: MagicMock, mock_wiim_device: MagicMock
374    ) -> None:
375        """
376        The device still reports the previous AirPlay content after play_media.
377
378        Its metadata makes MA rebuild _attr_current_media from the device's own
379        uri, so the position guard must not key off _attr_current_media.
380        """
381        stream_url = "http://192.168.1.80:8097/single/abc/queue/item/uuid:test.flac"
382        mock_provider.mass.streams.resolve_stream_url = AsyncMock(return_value=stream_url)
383
384        device_media = MagicMock()
385        device_media.uri = "wiimu_airplay"
386        device_media.title = "Previous Song"
387        device_media.artist = "Previous Artist"
388        device_media.album = "Previous Album"
389        device_media.position = 273
390        mock_wiim_device.play_mode = SOURCE_NETWORK
391        mock_wiim_device.current_media = device_media
392        mock_wiim_device.playing_status = PlayingStatus.PLAYING
393
394        player = self._make_player(mock_provider, mock_wiim_device)
395        player._attr_elapsed_time = 273
396
397        await player.play_media(PlayerMedia(uri="library://track/1", title="New Track"))
398        assert player._attr_elapsed_time == 0
399
400        await player._sync_position()
401
402        assert player._attr_elapsed_time == 0
403
404    @pytest.mark.asyncio
405    async def test_failed_play_media_releases_the_guard(
406        self, mock_provider: MagicMock, mock_wiim_device: MagicMock
407    ) -> None:
408        """A device that never took our stream must not stay guarded against its own position."""
409        stream_url = "http://192.168.1.80:8097/single/abc/queue/item/uuid:test.flac"
410        mock_provider.mass.streams.resolve_stream_url = AsyncMock(return_value=stream_url)
411        mock_wiim_device.async_play = AsyncMock(side_effect=WiimDeviceException("boom"))
412        player = self._make_player(mock_provider, mock_wiim_device)
413
414        await player.play_media(PlayerMedia(uri="library://track/1", title="New Track"))
415
416        device_media = MagicMock()
417        device_media.uri = "wiimu_airplay"
418        device_media.position = 42
419        mock_wiim_device.current_media = device_media
420
421        await player._sync_position()
422
423        assert player._attr_elapsed_time == 42
424
425    @pytest.mark.asyncio
426    async def test_sync_position_ignores_position_reported_without_uri(
427        self, mock_provider: MagicMock, mock_wiim_device: MagicMock
428    ) -> None:
429        """The device clears its uri mid-handover while still reporting the old position."""
430        stream_url = "http://192.168.1.80:8097/single/abc/queue/item/uuid:test.flac"
431        mock_provider.mass.streams.resolve_stream_url = AsyncMock(return_value=stream_url)
432
433        device_media = MagicMock()
434        device_media.uri = None
435        device_media.title = None
436        device_media.artist = None
437        device_media.album = None
438        device_media.position = 273
439        mock_wiim_device.play_mode = SOURCE_NETWORK
440        mock_wiim_device.current_media = device_media
441
442        player = self._make_player(mock_provider, mock_wiim_device)
443
444        await player.play_media(PlayerMedia(uri="library://track/1", title="New Track"))
445        await player._sync_position()
446
447        assert player._attr_elapsed_time == 0
448
449    @pytest.mark.asyncio
450    async def test_sync_position_ignores_foreign_uri(
451        self, mock_provider: MagicMock, mock_wiim_device: MagicMock
452    ) -> None:
453        """The device's position is rejected while it hasn't loaded MA's stream uri."""
454        stream_uri = "http://192.168.1.80:8097/single/abc/queue/item/uuid:test.flac"
455        player = self._make_player(mock_provider, mock_wiim_device)
456        player._ma_stream_uri = stream_uri
457        player._attr_elapsed_time = 0
458        player._attr_elapsed_time_last_updated = 1000.0
459
460        device_media = MagicMock()
461        device_media.uri = "wiimu_airplay"
462        device_media.position = 273
463        mock_wiim_device.current_media = device_media
464
465        await player._sync_position()
466
467        assert player._attr_elapsed_time == 0
468        assert player._attr_elapsed_time_last_updated == 1000.0
469
470    @pytest.mark.asyncio
471    async def test_sync_position_accepts_matching_uri_and_clears_guard(
472        self, mock_provider: MagicMock, mock_wiim_device: MagicMock
473    ) -> None:
474        """Once the device reports MA's own uri, its position is trusted and the guard lifts."""
475        stream_uri = "http://192.168.1.80:8097/single/abc/queue/item/uuid:test.flac"
476        mock_provider.mass.streams.resolve_stream_url = AsyncMock(return_value=stream_uri)
477        player = self._make_player(mock_provider, mock_wiim_device)
478
479        device_media = MagicMock()
480        device_media.uri = stream_uri
481        device_media.position = 12
482        mock_wiim_device.current_media = device_media
483
484        await player.play_media(PlayerMedia(uri="library://track/1", title="New Track"))
485        player._attr_elapsed_time_last_updated = 1000.0
486
487        await player._sync_position()
488
489        assert player._attr_elapsed_time == 12
490        assert player._attr_elapsed_time_last_updated > 1000.0
491        assert player._ma_stream_uri is None
492
493        # A later switch to an external source must not be blocked by a stale guard.
494        device_media.uri = "wiimu_airplay"
495        device_media.position = 55
496        await player._sync_position()
497
498        assert player._attr_elapsed_time == 55
499
500    @pytest.mark.asyncio
501    async def test_sync_position_accepts_device_when_ma_not_driving_playback(
502        self, mock_provider: MagicMock, mock_wiim_device: MagicMock
503    ) -> None:
504        """With no stream handed over by MA (external source), the device's position is authoritative."""
505        player = self._make_player(mock_provider, mock_wiim_device)
506        player._ma_stream_uri = None
507        player._attr_elapsed_time = 0
508        player._attr_elapsed_time_last_updated = 1000.0
509
510        device_media = MagicMock()
511        device_media.uri = "wiimu_airplay"
512        device_media.position = 42
513        mock_wiim_device.current_media = device_media
514
515        await player._sync_position()
516
517        assert player._attr_elapsed_time == 42
518        assert player._attr_elapsed_time_last_updated > 1000.0
519