/
/
/
1"""Fixtures for DACP replay tests."""
2
3from __future__ import annotations
4
5from unittest.mock import AsyncMock, MagicMock
6
7import pytest
8from music_assistant_models.enums import PlaybackState
9
10from music_assistant.providers.airplay.provider import AirPlayProvider
11
12_PLAYER_CMDS = (
13 "cmd_play",
14 "cmd_pause",
15 "cmd_play_pause",
16 "cmd_stop",
17 "cmd_next_track",
18 "cmd_previous_track",
19 "cmd_volume_up",
20 "cmd_volume_down",
21 "cmd_ungroup",
22)
23
24
25@pytest.fixture
26def mock_mass() -> MagicMock:
27 """Create a mock MusicAssistant instance for the AirPlay provider."""
28 mass = MagicMock()
29 mass.players = MagicMock()
30 for name in _PLAYER_CMDS:
31 setattr(mass.players, name, AsyncMock())
32 mass.players.get_active_queue = MagicMock(
33 return_value=MagicMock(queue_id="q1", shuffle_enabled=False)
34 )
35 mass.players.iter_players = MagicMock(return_value=[])
36 mass.player_queues = MagicMock()
37 mass.player_queues.set_shuffle = AsyncMock()
38 mass.call_later = MagicMock()
39 mass.cancel_timer = MagicMock()
40 mass.config.get_raw_player_config_value = MagicMock(return_value=False)
41 return mass
42
43
44@pytest.fixture
45def manifest_mock() -> MagicMock:
46 """Return a mock provider manifest."""
47 manifest = MagicMock()
48 manifest.domain = "airplay"
49 manifest.name = "AirPlay"
50 return manifest
51
52
53@pytest.fixture
54def config_mock() -> MagicMock:
55 """Return a mock provider config (GLOBAL log level so construction doesn't fail)."""
56 config = MagicMock()
57 config.instance_id = "airplay_test"
58 config.get_value = MagicMock(return_value="GLOBAL")
59 return config
60
61
62@pytest.fixture
63def airplay_provider(
64 mock_mass: MagicMock, manifest_mock: MagicMock, config_mock: MagicMock
65) -> AirPlayProvider:
66 """
67 Build a real AirPlayProvider backed by the mock mass.
68
69 handle_async_init (DACP server / zeroconf) is intentionally not called; only the
70 request handler is exercised. The default logger level keeps VERBOSE capture off.
71 """
72 provider = AirPlayProvider(mock_mass, manifest_mock, config_mock)
73 provider._bridge_manager = MagicMock()
74 provider._bridge_manager.get_transport_command_target.return_value = None
75 return provider
76
77
78def make_player( # noqa: PLR0913
79 mass: MagicMock,
80 *,
81 active_remote: str = "123",
82 player_id: str = "ap123",
83 playing: bool = True,
84 manufacturer: str = "Test",
85 synced_to: str | None = None,
86 transitioning: bool = False,
87 stream_connected: bool = True,
88 prevent_playback: bool = False,
89 has_session: bool = True,
90 has_stream: bool = True,
91) -> MagicMock:
92 """
93 Make the mock mass return a single mock AirPlay player, and return it.
94
95 Each call replaces the mass player list, so calling it twice does not yield two
96 players.
97
98 :param mass: The mock_mass fixture the provider looks players up through.
99 :param active_remote: Active-Remote id the player's stream answers to.
100 :param player_id: Player id used in dispatched commands.
101 :param playing: Whether the player reports PLAYING (else IDLE).
102 :param manufacturer: Device manufacturer ("Apple" triggers ignore-volume).
103 :param synced_to: Parent player id when synced, else None.
104 :param transitioning: Value of player._transitioning.
105 :param stream_connected: Whether the stream reports connected.
106 :param prevent_playback: Initial stream.prevent_playback flag.
107 :param has_session: Whether the stream has an active session.
108 :param has_stream: Whether the player has a stream at all.
109 """
110 player = MagicMock()
111 player.player_id = player_id
112 player.name = "Test Player"
113 player.protocol_parent_id = None
114 player.synced_to = synced_to
115 player._transitioning = transitioning
116 state = PlaybackState.PLAYING if playing else PlaybackState.IDLE
117 player.playback_state = state
118 player.state.playback_state = state
119 player.state.active_group = None
120 player.device_info.manufacturer = manufacturer
121 player.ignore_volume_reports = manufacturer.lower() == "apple"
122 player.volume_muted = False
123 player.volume_level = 50
124 if has_stream:
125 stream = MagicMock()
126 stream.active_remote_id = active_remote
127 stream.running = True
128 stream.connected = stream_connected
129 stream.prevent_playback = prevent_playback
130 stream.session = MagicMock() if has_session else None
131 stream.stop = AsyncMock()
132 stream.send_cli_command = AsyncMock()
133 player.stream = stream
134 else:
135 player.stream = None
136 mass.players.iter_players.return_value = [player]
137 return player
138