/
/
/
1"""
2Tests for the inbound half of the Music Assistant Cast namespace.
3
4The receiver app forwards the device's own next/previous button presses as
5custom messages (see https://github.com/music-assistant/support/issues/2969);
6this controller turns them back into MA queue commands.
7"""
8
9from __future__ import annotations
10
11from typing import cast
12from unittest.mock import MagicMock
13
14import pytest
15from music_assistant_models.enums import PlaybackState
16
17from music_assistant.providers.chromecast.constants import DASHBOARD_NAMESPACE
18from music_assistant.providers.chromecast.player import ChromecastPlayer
19from music_assistant.providers.chromecast.receiver_commands import MassCastCommandController
20
21
22def test_controller_uses_the_mass_cast_namespace() -> None:
23 """The controller listens on the same namespace the receiver replies on."""
24 controller = MassCastCommandController(MagicMock())
25
26 assert controller.namespace == DASHBOARD_NAMESPACE
27
28
29@pytest.mark.parametrize("command", ["next", "previous"])
30def test_player_command_is_forwarded(command: str) -> None:
31 """A player_command message invokes the callback and is marked handled."""
32 on_command = MagicMock()
33 controller = MassCastCommandController(on_command)
34
35 handled = controller.receive_message(
36 MagicMock(), {"type": "player_command", "command": command}
37 )
38
39 assert handled is True
40 on_command.assert_called_once_with(command)
41
42
43@pytest.mark.parametrize(
44 "data",
45 [
46 {"type": "receiver_status", "connected": True},
47 {"type": "player_command", "command": "shuffle"},
48 {"type": "player_command"},
49 {},
50 ],
51)
52def test_unhandled_messages_are_ignored(data: dict[str, object]) -> None:
53 """Other namespace traffic is left for other handlers and never dispatched."""
54 on_command = MagicMock()
55 controller = MassCastCommandController(on_command)
56
57 assert controller.receive_message(MagicMock(), data) is False
58 on_command.assert_not_called()
59
60
61### Dispatch into the queue controller
62
63
64def _fake_cast(
65 *,
66 queue_id: str | None = "cast_queue",
67 queue_active: bool = True,
68 queue_state: PlaybackState = PlaybackState.PLAYING,
69) -> MagicMock:
70 """Build a MagicMock Cast whose call_soon_threadsafe hop runs the callback inline."""
71 fake = MagicMock()
72 fake.display_name = "Fake Cast"
73 fake.mass.closing = False
74 # call_soon_threadsafe runs the callback inline so the dispatch is observable
75 fake.mass.loop.call_soon_threadsafe = MagicMock(side_effect=lambda func, *args: func(*args))
76 if queue_id is None:
77 fake.mass.players.get_active_queue.return_value = None
78 else:
79 queue = MagicMock()
80 queue.queue_id = queue_id
81 queue.active = queue_active
82 queue.state = queue_state
83 fake.mass.players.get_active_queue.return_value = queue
84 return fake
85
86
87def _dispatch(fake: MagicMock, command: str) -> None:
88 ChromecastPlayer._handle_receiver_command(cast("ChromecastPlayer", fake), command)
89
90
91def test_next_command_targets_the_active_queue() -> None:
92 """Next is dispatched to the queue that get_active_queue resolves for this player."""
93 fake = _fake_cast(queue_id="up_universal")
94
95 _dispatch(fake, "next")
96
97 fake.mass.loop.call_soon_threadsafe.assert_called_once()
98 fake.mass.players.get_active_queue.assert_called_once_with(fake)
99 fake.mass.create_task.assert_called_once()
100 fake.mass.player_queues.next.assert_called_once_with("up_universal")
101
102
103def test_previous_command_targets_the_active_queue() -> None:
104 """Previous is routed to the same resolved queue id."""
105 fake = _fake_cast()
106
107 _dispatch(fake, "previous")
108
109 fake.mass.loop.call_soon_threadsafe.assert_called_once()
110 fake.mass.player_queues.previous.assert_called_once_with("cast_queue")
111
112
113def test_no_active_queue_is_ignored() -> None:
114 """A command without any resolvable queue (dashboard-only session) is dropped."""
115 fake = _fake_cast(queue_id=None)
116
117 _dispatch(fake, "next")
118
119 fake.mass.player_queues.next.assert_not_called()
120 fake.mass.create_task.assert_not_called()
121
122
123def test_inactive_queue_is_ignored() -> None:
124 """A command whose resolved queue is not active is dropped, not dispatched."""
125 fake = _fake_cast(queue_active=False)
126
127 _dispatch(fake, "next")
128
129 fake.mass.player_queues.next.assert_not_called()
130 fake.mass.create_task.assert_not_called()
131
132
133def test_idle_queue_is_ignored() -> None:
134 """An idle queue still reports active=True; a press must not start playback."""
135 fake = _fake_cast(queue_state=PlaybackState.IDLE)
136
137 _dispatch(fake, "next")
138
139 fake.mass.player_queues.next.assert_not_called()
140 fake.mass.create_task.assert_not_called()
141
142
143def test_commands_are_ignored_during_shutdown() -> None:
144 """A press racing MusicAssistant.stop() must not schedule a new queue task."""
145 fake = _fake_cast()
146 fake.mass.closing = True
147
148 _dispatch(fake, "next")
149
150 fake.mass.loop.call_soon_threadsafe.assert_not_called()
151 fake.mass.create_task.assert_not_called()
152
153
154def test_paused_queue_is_dispatched() -> None:
155 """Next while paused is a legitimate command and goes through."""
156 fake = _fake_cast(queue_state=PlaybackState.PAUSED)
157
158 _dispatch(fake, "next")
159
160 fake.mass.player_queues.next.assert_called_once_with("cast_queue")
161