/
/
/
1"""Tests for PlayerQueuesController.set_overlay."""
2
3from __future__ import annotations
4
5from unittest.mock import AsyncMock, MagicMock, Mock
6
7import pytest
8from music_assistant_models.enums import PlaybackState
9from music_assistant_models.errors import InvalidCommand, InvalidDataError
10from music_assistant_models.media_items import (
11 ItemMapping,
12 MediaItemType,
13 ProviderMapping,
14 SoundEffect,
15 Track,
16)
17from music_assistant_models.player_queue import PlayerQueue
18
19from music_assistant.controllers.player_queues import PlayerQueuesController
20from music_assistant.controllers.player_queues.state import PlayerQueueData
21
22QUEUE_ID = "q1"
23
24
25def _sound_effect(item_id: str = "rain") -> SoundEffect:
26 """Build a SoundEffect on the 'ambient' provider."""
27 return SoundEffect(
28 item_id=item_id,
29 provider="ambient",
30 name=f"Sound {item_id}",
31 provider_mappings={
32 ProviderMapping(item_id=item_id, provider_domain="ambient", provider_instance="ambient")
33 },
34 )
35
36
37def _track() -> Track:
38 """Build a minimal Track (an invalid overlay source)."""
39 return Track(
40 item_id="track1",
41 provider="test",
42 name="Track",
43 provider_mappings={
44 ProviderMapping(item_id="track1", provider_domain="test", provider_instance="test")
45 },
46 )
47
48
49def _controller(resolved_item: MediaItemType | None = None) -> PlayerQueuesController:
50 """Build a bare controller with one queue and the collaborators stubbed out."""
51 ctrl = PlayerQueuesController.__new__(PlayerQueuesController)
52 ctrl.signal_update = Mock() # type: ignore[method-assign]
53 ctrl.resume = AsyncMock() # type: ignore[method-assign]
54 ctrl.mass = MagicMock()
55 ctrl.mass.music.get_item_by_uri = AsyncMock(return_value=resolved_item)
56 queue = PlayerQueue(
57 queue_id=QUEUE_ID, active=True, display_name="Queue", available=True, items=0
58 )
59 ctrl._queue_data = {QUEUE_ID: PlayerQueueData(queue=queue)}
60 return ctrl
61
62
63async def test_set_overlay_source_and_enable() -> None:
64 """Setting a source and enabling stores the mapping and signals an update."""
65 ctrl = _controller(resolved_item=_sound_effect())
66 await ctrl.set_overlay(QUEUE_ID, enabled=True, source="ambient://sound_effect/rain")
67 queue = ctrl._queue_data[QUEUE_ID].queue
68 assert queue.overlay_enabled is True
69 assert queue.overlay_source == ItemMapping.from_item(_sound_effect())
70 ctrl.signal_update.assert_called_once_with(QUEUE_ID) # type: ignore[attr-defined]
71 # queue is idle: no playback restart
72 ctrl.resume.assert_not_awaited() # type: ignore[attr-defined]
73
74
75async def test_set_overlay_enable_without_source_raises() -> None:
76 """Enabling the overlay without a source selected is rejected."""
77 ctrl = _controller()
78 with pytest.raises(InvalidCommand):
79 await ctrl.set_overlay(QUEUE_ID, enabled=True)
80
81
82async def test_set_overlay_rejects_non_sound_effect_source() -> None:
83 """Only sound effect items are accepted as overlay source."""
84 ctrl = _controller(resolved_item=_track())
85 with pytest.raises(InvalidDataError):
86 await ctrl.set_overlay(QUEUE_ID, source="library://track/1")
87
88
89@pytest.mark.parametrize("volume", [-1, 201])
90async def test_set_overlay_rejects_out_of_range_volume(volume: int) -> None:
91 """Overlay volume outside 0-200 is rejected."""
92 ctrl = _controller()
93 with pytest.raises(InvalidDataError):
94 await ctrl.set_overlay(QUEUE_ID, volume=volume)
95
96
97async def test_set_overlay_restarts_playback_when_playing() -> None:
98 """An audible change while playing restarts playback from the current position."""
99 ctrl = _controller(resolved_item=_sound_effect())
100 queue = ctrl._queue_data[QUEUE_ID].queue
101 queue.state = PlaybackState.PLAYING
102 await ctrl.set_overlay(QUEUE_ID, enabled=True, source="ambient://sound_effect/rain")
103 ctrl.resume.assert_awaited_once_with(QUEUE_ID) # type: ignore[attr-defined]
104
105
106async def test_set_overlay_inaudible_change_does_not_restart() -> None:
107 """Changing volume while the overlay is disabled never restarts playback."""
108 ctrl = _controller()
109 queue = ctrl._queue_data[QUEUE_ID].queue
110 queue.state = PlaybackState.PLAYING
111 await ctrl.set_overlay(QUEUE_ID, volume=50)
112 assert queue.overlay_volume == 50
113 ctrl.signal_update.assert_called_once_with(QUEUE_ID) # type: ignore[attr-defined]
114 ctrl.resume.assert_not_awaited() # type: ignore[attr-defined]
115
116
117async def test_set_overlay_no_change_is_noop() -> None:
118 """A call that changes nothing does not signal or restart anything."""
119 ctrl = _controller()
120 await ctrl.set_overlay(QUEUE_ID, enabled=False, volume=100)
121 ctrl.signal_update.assert_not_called() # type: ignore[attr-defined]
122 ctrl.resume.assert_not_awaited() # type: ignore[attr-defined]
123
124
125async def test_set_overlay_disable_keeps_source() -> None:
126 """Disabling the overlay keeps the selected source for later re-enabling."""
127 ctrl = _controller(resolved_item=_sound_effect())
128 await ctrl.set_overlay(QUEUE_ID, enabled=True, source="ambient://sound_effect/rain")
129 await ctrl.set_overlay(QUEUE_ID, enabled=False)
130 queue = ctrl._queue_data[QUEUE_ID].queue
131 assert queue.overlay_enabled is False
132 assert queue.overlay_source is not None
133