/
/
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 _request(session_id: str, queue_id: str = "queue-1", command: str = "next") -> MagicMock:
23 """Return a request for the command route."""
24 request = MagicMock()
25 request.method = "GET"
26 request.headers = {}
27 request.match_info = {
28 "session_id": session_id,
29 "queue_id": queue_id,
30 "command": command,
31 }
32 return request
33
34
35def test_command_url_is_bound_to_the_playback_session(
36 streams_controller: StreamsController,
37) -> None:
38 """The url carries the session, so it stops working once playback moves on."""
39 streams_controller.mass.player_queues = MagicMock()
40 streams_controller.mass.player_queues.queue_data_or_none.return_value = _queue_data("session-1")
41 url = streams_controller.get_command_url("queue-1", "next")
42 assert url is not None
43 assert url.endswith("/command/session-1/queue-1/next.mp3")
44
45
46@pytest.mark.parametrize(
47 ("queue_data", "reason"),
48 [(None, "unknown queue"), (_queue_data(None), "queue not playing")],
49)
50def test_command_url_is_absent_without_a_session(
51 streams_controller: StreamsController, queue_data: MagicMock | None, reason: str
52) -> None:
53 """Without an active session there is no url to hand to a player."""
54 streams_controller.mass.player_queues = MagicMock()
55 streams_controller.mass.player_queues.queue_data_or_none.return_value = queue_data
56 assert streams_controller.get_command_url("queue-1", "next") is None, reason
57
58
59async def test_command_request_advances_the_queue(
60 streams_controller: StreamsController, monkeypatch: pytest.MonkeyPatch
61) -> None:
62 """A request carrying the current session skips to the next item."""
63 streams_controller.mass.player_queues = MagicMock()
64 streams_controller.mass.player_queues.queue_data_or_none.return_value = _queue_data("session-1")
65 create_task = MagicMock()
66 monkeypatch.setattr(streams_controller.mass, "create_task", create_task)
67 resp = await streams_controller.serve_command_request(_request("session-1"))
68 assert isinstance(resp, web.FileResponse)
69 create_task.assert_called_once()
70
71
72@pytest.mark.parametrize("session_id", ["session-2", ""])
73async def test_command_request_refuses_a_foreign_session(
74 streams_controller: StreamsController, session_id: str, monkeypatch: pytest.MonkeyPatch
75) -> None:
76 """A caller that does not hold the current session cannot drive the queue."""
77 streams_controller.mass.player_queues = MagicMock()
78 streams_controller.mass.player_queues.queue_data_or_none.return_value = _queue_data("session-1")
79 create_task = MagicMock()
80 monkeypatch.setattr(streams_controller.mass, "create_task", create_task)
81 with pytest.raises(web.HTTPNotFound):
82 await streams_controller.serve_command_request(_request(session_id))
83 create_task.assert_not_called()
84
85
86async def test_command_request_refuses_an_unknown_queue(
87 streams_controller: StreamsController, monkeypatch: pytest.MonkeyPatch
88) -> None:
89 """An unregistered queue is refused rather than raising out of the handler."""
90 streams_controller.mass.player_queues = MagicMock()
91 streams_controller.mass.player_queues.queue_data_or_none.return_value = None
92 create_task = MagicMock()
93 monkeypatch.setattr(streams_controller.mass, "create_task", create_task)
94 with pytest.raises(web.HTTPNotFound):
95 await streams_controller.serve_command_request(_request("session-1"))
96 create_task.assert_not_called()
97