/
/
/
1"""Tests for queue move / reorder MCP tools."""
2
3from __future__ import annotations
4
5from types import SimpleNamespace
6from unittest.mock import MagicMock
7
8import pytest
9from fastmcp import Client, FastMCP
10from fastmcp.exceptions import ToolError
11from music_assistant_models.errors import InvalidDataError
12
13
14def _queue_item(*, item_id: str, name: str) -> SimpleNamespace:
15 return SimpleNamespace(queue_item_id=item_id, name=name, duration=180, media_item=None)
16
17
18def _mock_queue(mock_mass: MagicMock, *, queue_id: str, items: list[SimpleNamespace]) -> None:
19 queue = SimpleNamespace(
20 queue_id=queue_id,
21 current_index=0,
22 items=len(items),
23 shuffle_enabled=False,
24 repeat_mode=None,
25 available=True,
26 )
27 mock_mass.player_queues.get = MagicMock(return_value=queue)
28 mock_mass.player_queues.items = MagicMock(return_value=items)
29
30
31async def test_move_item_forwards_pos_shift(mounted_queue: FastMCP, mock_mass: MagicMock) -> None:
32 """pos_shift is forwarded to player_queues.move_item."""
33 _mock_queue(mock_mass, queue_id="q1", items=[_queue_item(item_id="item-1", name="One")])
34 async with Client(mounted_queue) as client:
35 result = await client.call_tool(
36 "queue_move_item",
37 {"queue_id": "q1", "item_id": "item-1", "pos_shift": -1},
38 )
39 mock_mass.player_queues.move_item.assert_called_once_with("q1", "item-1", -1)
40 assert result.data.queue_id == "q1"
41 assert [i.item_id for i in result.data.items] == ["item-1"]
42
43
44async def test_move_item_defaults_pos_shift(mounted_queue: FastMCP, mock_mass: MagicMock) -> None:
45 """pos_shift defaults to +1 (move down one slot)."""
46 _mock_queue(mock_mass, queue_id="q1", items=[_queue_item(item_id="item-1", name="One")])
47 async with Client(mounted_queue) as client:
48 await client.call_tool(
49 "queue_move_item",
50 {"queue_id": "q1", "item_id": "item-1"},
51 )
52 mock_mass.player_queues.move_item.assert_called_once_with("q1", "item-1", 1)
53
54
55async def test_move_item_returns_queue_brief(mounted_queue: FastMCP, mock_mass: MagicMock) -> None:
56 """A successful move returns the reordered queue order inline."""
57 _mock_queue(
58 mock_mass,
59 queue_id="q1",
60 items=[
61 _queue_item(item_id="item-2", name="Two"),
62 _queue_item(item_id="item-1", name="One"),
63 ],
64 )
65 async with Client(mounted_queue) as client:
66 result = await client.call_tool(
67 "queue_move_item",
68 {"queue_id": "q1", "item_id": "item-2", "pos_shift": -1},
69 )
70 assert [i.item_id for i in result.data.items] == ["item-2", "item-1"]
71 assert [i.name for i in result.data.items] == ["Two", "One"]
72
73
74async def test_move_item_raises_tool_error_on_buffered_item(
75 mounted_queue: FastMCP, mock_mass: MagicMock
76) -> None:
77 """Buffered/played items surface as ToolError for the agent."""
78 mock_mass.player_queues.get = MagicMock(return_value=SimpleNamespace(queue_id="q1"))
79 mock_mass.player_queues.move_item = MagicMock(
80 side_effect=IndexError("0 is already played/buffered")
81 )
82 async with Client(mounted_queue) as client:
83 with pytest.raises(ToolError, match="buffered"):
84 await client.call_tool(
85 "queue_move_item",
86 {"queue_id": "q1", "item_id": "item-1", "pos_shift": 0},
87 )
88
89
90async def test_move_item_raises_tool_error_on_missing_item(
91 mounted_queue: FastMCP, mock_mass: MagicMock
92) -> None:
93 """Unknown item_id surfaces as ToolError."""
94 mock_mass.player_queues.get = MagicMock(return_value=SimpleNamespace(queue_id="q1"))
95 mock_mass.player_queues.move_item = MagicMock(
96 side_effect=InvalidDataError("Item missing not found in queue")
97 )
98 async with Client(mounted_queue) as client:
99 with pytest.raises(ToolError, match="not found"):
100 await client.call_tool(
101 "queue_move_item",
102 {"queue_id": "q1", "item_id": "missing", "pos_shift": 1},
103 )
104
105
106async def test_move_item_raises_tool_error_on_unknown_queue(
107 mounted_queue: FastMCP, mock_mass: MagicMock
108) -> None:
109 """Unknown queue_id surfaces as ToolError before the move call."""
110 mock_mass.player_queues.get = MagicMock(return_value=None)
111 async with Client(mounted_queue) as client:
112 with pytest.raises(ToolError, match="Queue 'q-missing' not found"):
113 await client.call_tool(
114 "queue_move_item",
115 {"queue_id": "q-missing", "item_id": "item-1", "pos_shift": 1},
116 )
117 mock_mass.player_queues.move_item.assert_not_called()
118
119
120async def test_move_item_raises_when_queue_missing_after_move(
121 mounted_queue: FastMCP, mock_mass: MagicMock
122) -> None:
123 """A vanished queue after a successful move surfaces as ToolError."""
124 queue = SimpleNamespace(queue_id="q1")
125 mock_mass.player_queues.get = MagicMock(side_effect=[queue, None])
126 async with Client(mounted_queue) as client:
127 with pytest.raises(ToolError, match="not found after move"):
128 await client.call_tool(
129 "queue_move_item",
130 {"queue_id": "q1", "item_id": "item-1", "pos_shift": 1},
131 )
132
133
134async def test_move_item_to_end_forwards_call(mounted_queue: FastMCP, mock_mass: MagicMock) -> None:
135 """move_item_to_end delegates to player_queues.move_item_end and returns the brief."""
136 _mock_queue(mock_mass, queue_id="q1", items=[_queue_item(item_id="item-1", name="One")])
137 async with Client(mounted_queue) as client:
138 result = await client.call_tool(
139 "queue_move_item_to_end",
140 {"queue_id": "q1", "item_id": "item-1"},
141 )
142 mock_mass.player_queues.move_item_end.assert_called_once_with("q1", "item-1")
143 assert result.data.queue_id == "q1"
144 assert [i.item_id for i in result.data.items] == ["item-1"]
145
146
147async def test_move_item_to_end_raises_tool_error_on_unknown_queue(
148 mounted_queue: FastMCP, mock_mass: MagicMock
149) -> None:
150 """Unknown queue_id surfaces as ToolError before move_item_end."""
151 mock_mass.player_queues.get = MagicMock(return_value=None)
152 async with Client(mounted_queue) as client:
153 with pytest.raises(ToolError, match="Queue 'q-missing' not found"):
154 await client.call_tool(
155 "queue_move_item_to_end",
156 {"queue_id": "q-missing", "item_id": "item-1"},
157 )
158 mock_mass.player_queues.move_item_end.assert_not_called()
159