/
/
/
1"""Tests for the command route that lets on-player buttons drive a flow stream."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING
6from unittest.mock import MagicMock
7
8import pytest
9from aiohttp import web
10
11if TYPE_CHECKING:
12 from music_assistant.controllers.streams.controller import StreamsController
13
14
15def _queue_data(session_id: str | None) -> MagicMock:
16 """Return a queue record carrying the given playback session."""
17 queue_data = MagicMock()
18 queue_data.session_id = session_id
19 return queue_data
20
21
22def _mock_player_queues(
23 streams_controller: StreamsController,
24 queue_data: MagicMock | None,
25 active_queue_id: str | None = "queue-1",
26) -> MagicMock:
27 """Install a player_queues mock resolving to the given active queue and record."""
28 player_queues = MagicMock()
29 player_queues.get_active_queue.return_value = (
30 MagicMock(queue_id=active_queue_id) if active_queue_id else None
31 )
32 player_queues.queue_data_or_none.return_value = queue_data
33 streams_controller.mass.player_queues = player_queues
34 return player_queues
35
36
37def _request(session_id: str, queue_id: str = "queue-1", command: str = "next") -> MagicMock:
38 """Return a request for the command route."""
39 request = MagicMock()
40 request.method = "GET"
41 request.headers = {}
42 request.match_info = {
43 "session_id": session_id,
44 "queue_id": queue_id,
45 "command": command,
46 }
47 return request
48
49
50def test_command_url_is_bound_to_the_playback_session(
51 streams_controller: StreamsController,
52) -> None:
53 """The url carries the session, so it stops working once playback moves on."""
54 _mock_player_queues(streams_controller, _queue_data("session-1"))
55 url = streams_controller.get_command_url("queue-1", "next")
56 assert url is not None
57 assert url.endswith("/command/session-1/queue-1/next.mp3")
58
59
60def test_command_url_targets_the_active_queue_of_a_protocol_player(
61 streams_controller: StreamsController,
62) -> None:
63 """A protocol (child) player id resolves to the queue that actually drives it."""
64 player_queues = _mock_player_queues(streams_controller, _queue_data("session-1"))
65 url = streams_controller.get_command_url("cast-child-1", "next")
66 assert url is not None
67 assert url.endswith("/command/session-1/queue-1/next.mp3")
68 player_queues.get_active_queue.assert_called_once_with("cast-child-1")
69 player_queues.queue_data_or_none.assert_called_once_with("queue-1")
70
71
72def test_command_url_falls_back_to_the_given_id_without_an_active_queue(
73 streams_controller: StreamsController,
74) -> None:
75 """Without a resolvable active queue the given id is used as the queue id."""
76 _mock_player_queues(streams_controller, _queue_data("session-1"), active_queue_id=None)
77 url = streams_controller.get_command_url("queue-1", "next")
78 assert url is not None
79 assert url.endswith("/command/session-1/queue-1/next.mp3")
80
81
82@pytest.mark.parametrize(
83 ("queue_data", "reason"),
84 [(None, "unknown queue"), (_queue_data(None), "queue not playing")],
85)
86def test_command_url_is_absent_without_a_session(
87 streams_controller: StreamsController, queue_data: MagicMock | None, reason: str
88) -> None:
89 """Without an active session there is no url to hand to a player."""
90 _mock_player_queues(streams_controller, queue_data)
91 assert streams_controller.get_command_url("queue-1", "next") is None, reason
92
93
94async def test_command_request_advances_the_queue(
95 streams_controller: StreamsController, monkeypatch: pytest.MonkeyPatch
96) -> None:
97 """A request carrying the current session skips to the next item."""
98 _mock_player_queues(streams_controller, _queue_data("session-1"))
99 create_task = MagicMock()
100 monkeypatch.setattr(streams_controller.mass, "create_task", create_task)
101 resp = await streams_controller.serve_command_request(_request("session-1"))
102 assert isinstance(resp, web.FileResponse)
103 create_task.assert_called_once()
104
105
106@pytest.mark.parametrize("session_id", ["session-2", ""])
107async def test_command_request_refuses_a_foreign_session(
108 streams_controller: StreamsController, session_id: str, monkeypatch: pytest.MonkeyPatch
109) -> None:
110 """A caller that does not hold the current session cannot drive the queue."""
111 _mock_player_queues(streams_controller, _queue_data("session-1"))
112 create_task = MagicMock()
113 monkeypatch.setattr(streams_controller.mass, "create_task", create_task)
114 with pytest.raises(web.HTTPNotFound):
115 await streams_controller.serve_command_request(_request(session_id))
116 create_task.assert_not_called()
117
118
119async def test_command_request_refuses_an_unknown_queue(
120 streams_controller: StreamsController, monkeypatch: pytest.MonkeyPatch
121) -> None:
122 """An unregistered queue is refused rather than raising out of the handler."""
123 _mock_player_queues(streams_controller, None)
124 create_task = MagicMock()
125 monkeypatch.setattr(streams_controller.mass, "create_task", create_task)
126 with pytest.raises(web.HTTPNotFound):
127 await streams_controller.serve_command_request(_request("session-1"))
128 create_task.assert_not_called()
129