/
/
/
1"""
2Tests for the flow-mode 'command' queue item a Cast player inserts.
3
4In flow mode the whole queue plays as one continuous stream, so the player gets a
5special command item appended to its cast queue: fetching it triggers queue-next on
6the server (the on-player next button, and the recovery path when the flow stream
7dies). Vendor cast stacks reject the item when its declared contentType does not
8match the silence file the url actually serves, so that coupling is asserted here.
9"""
10
11from __future__ import annotations
12
13import mimetypes
14from typing import Any, cast
15from unittest.mock import MagicMock
16
17from music_assistant_models.enums import PlaybackState
18from music_assistant_models.player import PlayerMedia
19
20from music_assistant.constants import SILENCE_FILE
21from music_assistant.providers.chromecast.player import ChromecastPlayer
22
23COMMAND_URL = "http://mass:8097/command/session-1/queue-1/next.mp3"
24
25
26def _fake_flow_player(*, cast_queue_items: list[Any] | None = None) -> Any:
27 """
28 Build a Cast player playing a flow stream, on mocked collaborators.
29
30 The player is seeded mid-flow-playback so on_player_media_updated runs its
31 metadata update for real, with the messages it sends captured on the mocked
32 media controller.
33
34 :param cast_queue_items: Items the receiver reports in its cast queue.
35 """
36 # __init__ is skipped: it needs a provider, cast info and a live Chromecast connection.
37 # Typed as Any because the collaborators below are read back as the mocks they are.
38 fake = cast("Any", ChromecastPlayer.__new__(ChromecastPlayer))
39 fake.mass = MagicMock()
40 fake.mass.streams.get_command_url.return_value = COMMAND_URL
41 fake.logger = MagicMock()
42 fake._player_id = "cast-child-1"
43 fake.cc = MagicMock()
44 fake.cc.media_controller.status.player_is_playing = True
45 fake.cc.media_controller.status.media_session_id = 7
46 fake.cc.media_controller.status.items = cast_queue_items or []
47 fake.active_cast_group = None
48 fake.flow_meta_checksum = None
49 fake._attr_powered = True
50 fake._attr_playback_state = PlaybackState.PLAYING
51 fake._attr_current_media = PlayerMedia(uri="http://mass:8097/flow/s1/q1/i1/p1.flac")
52 fake._state = MagicMock(current_media=PlayerMedia(uri="http://mass:8097/flow/s1/q1/i1/p1.flac"))
53 return fake
54
55
56async def _run_flow_metadata_update(fake: Any) -> list[dict[str, Any]]:
57 """Run the player's flow metadata update and return the cast messages it sent."""
58 fake.on_player_media_updated()
59 await fake.mass.create_task.call_args[0][0]
60 return [call.kwargs["data"] for call in fake.cc.media_controller.send_message.call_args_list]
61
62
63async def test_command_item_declares_the_type_of_the_served_silence_file() -> None:
64 """The inserted item's contentType matches the file its url serves, not the flow codec."""
65 fake = _fake_flow_player()
66 messages = await _run_flow_metadata_update(fake)
67 insert = next(msg for msg in messages if msg["type"] == "QUEUE_INSERT")
68 media = insert["items"][0]["media"]
69 assert media["contentId"] == COMMAND_URL
70 assert media["contentType"] == mimetypes.guess_type(SILENCE_FILE)[0]
71 fake.mass.streams.get_command_url.assert_called_once_with("cast-child-1", "next")
72
73
74async def test_command_item_is_not_inserted_without_a_command_url() -> None:
75 """No command item is queued when there is no active session to build a url for."""
76 fake = _fake_flow_player()
77 fake.mass.streams.get_command_url.return_value = None
78 messages = await _run_flow_metadata_update(fake)
79 assert not [msg for msg in messages if msg["type"] == "QUEUE_INSERT"]
80
81
82async def test_command_item_is_not_inserted_twice() -> None:
83 """A cast queue already holding the command item does not get another one."""
84 fake = _fake_flow_player(cast_queue_items=[MagicMock(), MagicMock()])
85 messages = await _run_flow_metadata_update(fake)
86 assert not [msg for msg in messages if msg["type"] == "QUEUE_INSERT"]
87