/
/
/
1"""Tests for queue_get_active_queue index visibility."""
2
3from __future__ import annotations
4
5from types import SimpleNamespace
6from unittest.mock import MagicMock
7
8from fastmcp import Client, FastMCP
9
10from music_assistant.providers.fastmcp_server.tools.queue import build_queue_server
11
12
13def _queue(*, current_index: int = 2, index_in_buffer: int | None = 4) -> SimpleNamespace:
14 return SimpleNamespace(
15 queue_id="q1",
16 current_index=current_index,
17 index_in_buffer=index_in_buffer,
18 items=10,
19 shuffle_enabled=False,
20 repeat_mode=SimpleNamespace(value="off"),
21 )
22
23
24def _items(count: int, *, start: int = 0) -> list[SimpleNamespace]:
25 return [
26 SimpleNamespace(
27 queue_item_id=f"id-{i}",
28 name=f"track-{i}",
29 duration=180,
30 media_item=None,
31 )
32 for i in range(start, start + count)
33 ]
34
35
36async def test_get_active_queue_exposes_insert_index_fields(mock_mass: MagicMock) -> None:
37 """get_active_queue returns next_insertable_index, index_in_buffer, and item.index."""
38 queue = _queue(current_index=2, index_in_buffer=4)
39 mock_mass.player_queues.get_active_queue = MagicMock(return_value=queue)
40 mock_mass.player_queues.items = MagicMock(return_value=_items(3))
41
42 mcp = FastMCP(name="test")
43 mcp.mount(build_queue_server(mock_mass), namespace="queue")
44 async with Client(mcp) as client:
45 result = await client.call_tool(
46 "queue_get_active_queue", {"player_id": "p1", "include_items": 3}
47 )
48
49 assert result.data.index_in_buffer == 4
50 assert result.data.next_insertable_index == 5
51 assert result.data.items_start_index == 0
52 assert [item.index for item in result.data.items] == [0, 1, 2]
53 mock_mass.player_queues.items.assert_called_once_with("q1", limit=3, offset=0)
54
55
56async def test_get_active_queue_items_from_current_uses_offset(mock_mass: MagicMock) -> None:
57 """items_from_current fetches from current_index and sets items_start_index."""
58 queue = _queue(current_index=7, index_in_buffer=9)
59 mock_mass.player_queues.get_active_queue = MagicMock(return_value=queue)
60 mock_mass.player_queues.items = MagicMock(return_value=_items(2, start=7))
61
62 mcp = FastMCP(name="test")
63 mcp.mount(build_queue_server(mock_mass), namespace="queue")
64 async with Client(mcp) as client:
65 result = await client.call_tool(
66 "queue_get_active_queue",
67 {"player_id": "p1", "include_items": 2, "items_from_current": True},
68 )
69
70 assert result.data.items_start_index == 7
71 assert [item.index for item in result.data.items] == [7, 8]
72 assert result.data.next_insertable_index == 10
73 mock_mass.player_queues.items.assert_called_once_with("q1", limit=2, offset=7)
74