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.providers.wiim.constants import SOURCE_NETWORK
11from music_assistant.providers.wiim.player import SDK_TO_MA_STATE, WiimPlayer
12
13
14@pytest.fixture
15def mock_wiim_device() -> MagicMock:
16    """Create a mock WiimDevice."""
17    device = MagicMock()
18    device.name = "Test WiiM Pro"
19    device.udn = "uuid:test-wiim-001"
20    device.available = True
21    device.volume = 50
22    device.is_muted = False
23    device.playing_status = None
24    device.play_mode = None
25    device.current_media = None
26    device.model_name = "WiiM Pro"
27    device.manufacturer = "Linkplay"
28    device.firmware_version = "4.8.1"
29    device.ip_address = "192.168.1.100"
30    device.supports_http_api = True
31    device.supported_input_modes = ("Network", "Bluetooth", "Line In", "Optical In")
32    device.async_play = AsyncMock()
33    device.async_pause = AsyncMock()
34    device.async_stop = AsyncMock()
35    device.async_set_volume = AsyncMock()
36    device.async_set_mute = AsyncMock()
37    device.async_set_play_mode = AsyncMock()
38    device.sync_device_duration_and_position = AsyncMock()
39    device.disconnect = AsyncMock()
40    device.ensure_subscriptions = AsyncMock()
41    device.general_event_callback = None
42    device.rendering_control_event_callback = None
43    device.av_transport_event_callback = None
44    device.play_queue_event_callback = None
45    return device
46
47
48@pytest.fixture
49def mock_controller() -> MagicMock:
50    """Create a mock WiimController."""
51    controller = MagicMock()
52    snapshot = MagicMock()
53    snapshot.role = "standalone"
54    snapshot.leader_udn = "uuid:test-wiim-001"
55    snapshot.member_udns = ("uuid:test-wiim-001",)
56    controller.get_group_snapshot.return_value = snapshot
57    controller.get_group_members.return_value = []
58    controller.get_device.return_value = MagicMock()
59    controller.async_join_group = AsyncMock()
60    controller.async_ungroup_device = AsyncMock()
61    return controller
62
63
64@pytest.fixture
65def mock_provider(mock_controller: MagicMock) -> MagicMock:
66    """Create a mock WiimProvider."""
67    provider = MagicMock()
68    provider.wiim_controller = mock_controller
69    provider.instance_id = "wiim_test"
70    provider.domain = "wiim"
71    provider.manifest = MagicMock()
72    provider.manifest.domain = "wiim"
73    provider.mass = MagicMock()
74    provider.mass.players = MagicMock()
75
76    config = MagicMock()
77    config.name = None
78    config.default_name = "Test WiiM Pro"
79    config.enabled = True
80    config.player_type = None
81    config.get_value = MagicMock(return_value=None)
82    provider.mass.config.get_base_player_config.return_value = config
83    return provider
84
85
86class TestSDKStateMapping:
87    """Test SDK to MA state mapping."""
88
89    def test_playing_maps_to_playing(self) -> None:
90        """PLAYING should map to PlaybackState.PLAYING."""
91        assert SDK_TO_MA_STATE[PlayingStatus.PLAYING] == PlaybackState.PLAYING
92
93    def test_paused_maps_to_paused(self) -> None:
94        """PAUSED should map to PlaybackState.PAUSED."""
95        assert SDK_TO_MA_STATE[PlayingStatus.PAUSED] == PlaybackState.PAUSED
96
97    def test_stopped_maps_to_idle(self) -> None:
98        """STOPPED should map to PlaybackState.IDLE."""
99        assert SDK_TO_MA_STATE[PlayingStatus.STOPPED] == PlaybackState.IDLE
100
101    def test_loading_maps_to_playing(self) -> None:
102        """LOADING should map to PlaybackState.PLAYING."""
103        assert SDK_TO_MA_STATE[PlayingStatus.LOADING] == PlaybackState.PLAYING
104
105    def test_all_sdk_states_mapped(self) -> None:
106        """All non-UNKNOWN SDK states should have a mapping."""
107        for status in PlayingStatus:
108            if status != PlayingStatus.UNKNOWN:
109                assert status in SDK_TO_MA_STATE, f"{status} not mapped"
110
111
112class TestFalsePlayingFilter:
113    """A uri-less PLAYING report in network mode must not become PLAYING state."""
114
115    def _make_player(self, provider: MagicMock, device: MagicMock) -> WiimPlayer:
116        player = WiimPlayer(provider=provider, player_id="uuid:test", device=device)
117        player.update_state = MagicMock()  # type: ignore[misc,method-assign]
118        return player
119
120    def test_false_playing_ack_is_suppressed(
121        self, mock_provider: MagicMock, mock_wiim_device: MagicMock
122    ) -> None:
123        """
124        The transient PLAYING ack without media loaded must keep the previous state.
125
126        The device acks (group) transport commands with a short false PLAYING
127        report before any track is loaded; propagating it causes a
128        PLAYING->IDLE->PLAYING flicker downstream.
129        """
130        mock_wiim_device.play_mode = SOURCE_NETWORK
131        mock_wiim_device.current_media = None
132        mock_wiim_device.playing_status = PlayingStatus.PLAYING
133        player = self._make_player(mock_provider, mock_wiim_device)
134        player._attr_playback_state = PlaybackState.IDLE
135
136        player._update_ma_state_from_sdk_cache()
137
138        assert player._attr_playback_state == PlaybackState.IDLE
139
140    def test_loading_without_uri_is_suppressed(
141        self, mock_provider: MagicMock, mock_wiim_device: MagicMock
142    ) -> None:
143        """LOADING maps to PLAYING and gets the same uri-less filter."""
144        mock_wiim_device.play_mode = SOURCE_NETWORK
145        mock_wiim_device.current_media = None
146        mock_wiim_device.playing_status = PlayingStatus.LOADING
147        player = self._make_player(mock_provider, mock_wiim_device)
148        player._attr_playback_state = PlaybackState.IDLE
149
150        player._update_ma_state_from_sdk_cache()
151
152        assert player._attr_playback_state == PlaybackState.IDLE
153
154    def test_playing_kept_when_uri_drops_mid_playback(
155        self, mock_provider: MagicMock, mock_wiim_device: MagicMock
156    ) -> None:
157        """The filter keeps the previous state; it never forces a playing player idle."""
158        mock_wiim_device.play_mode = SOURCE_NETWORK
159        mock_wiim_device.current_media = None
160        mock_wiim_device.playing_status = PlayingStatus.PLAYING
161        player = self._make_player(mock_provider, mock_wiim_device)
162        player._attr_playback_state = PlaybackState.PLAYING
163
164        player._update_ma_state_from_sdk_cache()
165
166        assert player._attr_playback_state == PlaybackState.PLAYING
167
168    def test_playing_accepted_once_uri_present(
169        self, mock_provider: MagicMock, mock_wiim_device: MagicMock
170    ) -> None:
171        """A PLAYING report with media loaded is a real start and passes through."""
172        media = MagicMock()
173        media.uri = "http://192.168.1.80:8097/single/abc/queue/item/uuid:test.flac"
174        mock_wiim_device.play_mode = SOURCE_NETWORK
175        mock_wiim_device.current_media = media
176        mock_wiim_device.playing_status = PlayingStatus.PLAYING
177        player = self._make_player(mock_provider, mock_wiim_device)
178        player._attr_playback_state = PlaybackState.IDLE
179
180        player._update_ma_state_from_sdk_cache()
181
182        assert player._attr_playback_state == PlaybackState.PLAYING
183
184    def test_external_input_playing_without_uri_accepted(
185        self, mock_provider: MagicMock, mock_wiim_device: MagicMock
186    ) -> None:
187        """External inputs legitimately play without a URI and must not be filtered."""
188        mock_wiim_device.play_mode = "Line In"
189        mock_wiim_device.current_media = None
190        mock_wiim_device.playing_status = PlayingStatus.PLAYING
191        player = self._make_player(mock_provider, mock_wiim_device)
192        player._attr_playback_state = PlaybackState.IDLE
193
194        player._update_ma_state_from_sdk_cache()
195
196        assert player._attr_playback_state == PlaybackState.PLAYING
197
198    def test_unknown_play_mode_trusts_device(
199        self, mock_provider: MagicMock, mock_wiim_device: MagicMock
200    ) -> None:
201        """Without a known play mode the device report is trusted (no suppression)."""
202        mock_wiim_device.play_mode = None
203        mock_wiim_device.current_media = None
204        mock_wiim_device.playing_status = PlayingStatus.PLAYING
205        player = self._make_player(mock_provider, mock_wiim_device)
206        player._attr_playback_state = PlaybackState.IDLE
207
208        player._update_ma_state_from_sdk_cache()
209
210        assert player._attr_playback_state == PlaybackState.PLAYING
211
212    def test_stopped_report_unaffected(
213        self, mock_provider: MagicMock, mock_wiim_device: MagicMock
214    ) -> None:
215        """The filter only guards PLAYING-mapped reports; STOPPED passes through."""
216        mock_wiim_device.play_mode = SOURCE_NETWORK
217        mock_wiim_device.current_media = None
218        mock_wiim_device.playing_status = PlayingStatus.STOPPED
219        player = self._make_player(mock_provider, mock_wiim_device)
220        player._attr_playback_state = PlaybackState.PLAYING
221
222        player._update_ma_state_from_sdk_cache()
223
224        assert player._attr_playback_state == PlaybackState.IDLE
225
226
227class TestSupportedFeatures:
228    """Test that required features are declared."""
229
230    def test_play_media_in_features(
231        self, mock_provider: MagicMock, mock_wiim_device: MagicMock
232    ) -> None:
233        """PLAY_MEDIA should be in supported features."""
234        player = WiimPlayer(provider=mock_provider, player_id="uuid:test", device=mock_wiim_device)
235        assert PlayerFeature.PLAY_MEDIA in player._attr_supported_features
236
237    def test_volume_features(self, mock_provider: MagicMock, mock_wiim_device: MagicMock) -> None:
238        """VOLUME_SET and VOLUME_MUTE should be in supported features."""
239        player = WiimPlayer(provider=mock_provider, player_id="uuid:test", device=mock_wiim_device)
240        assert PlayerFeature.VOLUME_SET in player._attr_supported_features
241        assert PlayerFeature.VOLUME_MUTE in player._attr_supported_features
242
243    def test_select_source_in_features(
244        self, mock_provider: MagicMock, mock_wiim_device: MagicMock
245    ) -> None:
246        """SELECT_SOURCE should be in supported features."""
247        player = WiimPlayer(provider=mock_provider, player_id="uuid:test", device=mock_wiim_device)
248        assert PlayerFeature.SELECT_SOURCE in player._attr_supported_features
249
250
251class TestSourceList:
252    """Test dynamic source list construction."""
253
254    @pytest.mark.asyncio
255    async def test_setup_adds_device_input_modes(
256        self, mock_provider: MagicMock, mock_wiim_device: MagicMock
257    ) -> None:
258        """setup() should add sources for device-supported input modes."""
259        player = WiimPlayer(provider=mock_provider, player_id="uuid:test", device=mock_wiim_device)
260        await player.setup()
261        source_ids = [s.id for s in player._attr_source_list]
262        assert "bluetooth" in source_ids
263        assert "line_in" in source_ids
264        assert "optical" in source_ids
265
266    @pytest.mark.asyncio
267    async def test_setup_adds_passive_sources(
268        self, mock_provider: MagicMock, mock_wiim_device: MagicMock
269    ) -> None:
270        """setup() should add passive sources (AirPlay, Spotify)."""
271        player = WiimPlayer(provider=mock_provider, player_id="uuid:test", device=mock_wiim_device)
272        await player.setup()
273        source_ids = [s.id for s in player._attr_source_list]
274        assert "airplay" in source_ids
275        assert "spotify" in source_ids
276
277    @pytest.mark.asyncio
278    async def test_setup_skips_unknown_input_modes(
279        self, mock_provider: MagicMock, mock_wiim_device: MagicMock
280    ) -> None:
281        """setup() should skip input modes not in INPUT_MODE_SOURCES."""
282        mock_wiim_device.supported_input_modes = ("Network", "FutureMode")
283        player = WiimPlayer(provider=mock_provider, player_id="uuid:test", device=mock_wiim_device)
284        await player.setup()
285        source_ids = [s.id for s in player._attr_source_list]
286        assert "futuremode" not in source_ids
287
288
289class TestErrorHandling:
290    """Test that command errors mark device unavailable."""
291
292    @pytest.mark.asyncio
293    async def test_play_error_refreshes_state(
294        self, mock_provider: MagicMock, mock_wiim_device: MagicMock
295    ) -> None:
296        """Play command error should refresh state without marking unavailable."""
297        mock_wiim_device.async_play.side_effect = WiimRequestException("timeout")
298        player = WiimPlayer(provider=mock_provider, player_id="uuid:test", device=mock_wiim_device)
299        player.update_state = MagicMock()  # type: ignore[misc,method-assign]
300        await player.play()
301        assert player._attr_available is True
302        player.update_state.assert_called()
303
304    @pytest.mark.skip(
305        reason="volume_set inlines the HTTP fix from wiim PR#18 and no longer calls "
306        "async_set_volume; re-enable when wiim-sdk 0.1.5+ has been released"
307    )
308    @pytest.mark.asyncio
309    async def test_volume_set_error_refreshes_state(
310        self, mock_provider: MagicMock, mock_wiim_device: MagicMock
311    ) -> None:
312        """Volume set error should refresh state without marking unavailable."""
313        mock_wiim_device.async_set_volume.side_effect = WiimDeviceException("disconnected")
314        player = WiimPlayer(provider=mock_provider, player_id="uuid:test", device=mock_wiim_device)
315        player.update_state = MagicMock()  # type: ignore[misc,method-assign]
316        await player.volume_set(50)
317        assert player._attr_available is True
318        player.update_state.assert_called()
319
320    @pytest.mark.asyncio
321    async def test_stop_error_refreshes_state(
322        self, mock_provider: MagicMock, mock_wiim_device: MagicMock
323    ) -> None:
324        """Stop command error should refresh state without marking unavailable."""
325        mock_wiim_device.async_stop.side_effect = WiimRequestException("connection lost")
326        player = WiimPlayer(provider=mock_provider, player_id="uuid:test", device=mock_wiim_device)
327        player.update_state = MagicMock()  # type: ignore[misc,method-assign]
328        await player.stop()
329        assert player._attr_available is True
330        player.update_state.assert_called()
331
332    @pytest.mark.asyncio
333    async def test_pause_error_refreshes_state(
334        self, mock_provider: MagicMock, mock_wiim_device: MagicMock
335    ) -> None:
336        """Pause command error should refresh state without marking unavailable."""
337        mock_wiim_device.async_pause.side_effect = WiimDeviceException("timeout")
338        player = WiimPlayer(provider=mock_provider, player_id="uuid:test", device=mock_wiim_device)
339        player.update_state = MagicMock()  # type: ignore[misc,method-assign]
340        await player.pause()
341        assert player._attr_available is True
342        player.update_state.assert_called()
343