/
/
/
1"""Tests for the SynchronizerRole."""
2
3from __future__ import annotations
4
5from unittest.mock import MagicMock, patch
6
7import pytest
8
9from music_assistant.providers.sendspin.synchronizer_role import (
10 SYNC_ROLE_ID,
11 SynchronizerRole,
12)
13
14
15@pytest.fixture
16def mock_client() -> MagicMock:
17 """Return a mock SendspinClient."""
18 client = MagicMock()
19 client.client_id = "test-client-123"
20 client.info.visualizer_support = MagicMock()
21 client.info.visualizer_support.to_dict.return_value = {
22 "buffer_capacity": 4096,
23 "types": ["loudness", "f_peak"],
24 "batch_max": 1,
25 }
26 client.group.group_role.return_value = MagicMock()
27 return client
28
29
30class TestSynchronizerRoleProperties:
31 """Test role properties and identity."""
32
33 def test_role_id(self, mock_client: MagicMock) -> None:
34 """Test that role ID is correct."""
35 role = SynchronizerRole(mock_client)
36 assert role.role_id == SYNC_ROLE_ID
37 assert role.role_id == "visualizer@_sync"
38
39 def test_role_family(self, mock_client: MagicMock) -> None:
40 """Test that role family is visualizer."""
41 role = SynchronizerRole(mock_client)
42 assert role.role_family == "visualizer"
43
44 def test_has_connection_returns_true(self, mock_client: MagicMock) -> None:
45 """Test that has_connection always returns True for bridge roles."""
46 role = SynchronizerRole(mock_client)
47 assert role.has_connection() is True
48
49 def test_supports_preconnect_audio(self, mock_client: MagicMock) -> None:
50 """Test that preconnect audio is supported."""
51 role = SynchronizerRole(mock_client)
52 assert role.supports_preconnect_audio() is True
53
54
55class TestSynchronizerRoleAudioRequirements:
56 """Test audio requirement setup."""
57
58 def test_initial_requirements_none(self, mock_client: MagicMock) -> None:
59 """Test that audio requirements are None before setup."""
60 role = SynchronizerRole(mock_client)
61 assert role.get_audio_requirements() is None
62
63 def test_setup_audio_requirements(self, mock_client: MagicMock) -> None:
64 """Test that setup creates proper audio requirements."""
65 role = SynchronizerRole(mock_client)
66 role.setup_audio_requirements()
67 reqs = role.get_audio_requirements()
68 assert reqs is not None
69 assert reqs.sample_rate == 48_000
70 assert reqs.bit_depth == 16
71 assert reqs.channels == 2
72 assert reqs.frame_duration_us == 25_000
73
74
75class TestSynchronizerRoleCallbacks:
76 """Test callback wiring and invocation."""
77
78 def test_callbacks_not_set_initially(self, mock_client: MagicMock) -> None:
79 """Test that callbacks are None before set_callbacks."""
80 role = SynchronizerRole(mock_client)
81 assert role._on_visualization_data_cb is None
82 assert role._on_stream_start_cb is None
83 assert role._on_stream_end_cb is None
84
85 def test_set_callbacks(self, mock_client: MagicMock) -> None:
86 """Test that set_callbacks wires up all callbacks."""
87 role = SynchronizerRole(mock_client)
88 viz_cb = MagicMock()
89 start_cb = MagicMock()
90 end_cb = MagicMock()
91 role.set_callbacks(
92 on_visualization_data=viz_cb,
93 on_stream_start=start_cb,
94 on_stream_end=end_cb,
95 )
96 assert role._on_visualization_data_cb is viz_cb
97 assert role._on_stream_start_cb is start_cb
98 assert role._on_stream_end_cb is end_cb
99
100 def test_on_stream_start_invokes_callback(self, mock_client: MagicMock) -> None:
101 """Test that on_stream_start creates extractor and calls callback."""
102 role = SynchronizerRole(mock_client)
103 start_cb = MagicMock()
104 role.set_callbacks(
105 on_visualization_data=MagicMock(),
106 on_stream_start=start_cb,
107 on_stream_end=MagicMock(),
108 )
109 role.setup_audio_requirements()
110 role.setup_stream_config()
111
112 with patch(
113 "music_assistant.providers.sendspin.synchronizer_role.VisualizerFeatureExtractor"
114 ):
115 role.on_stream_start()
116
117 start_cb.assert_called_once()
118
119 def test_on_stream_end_invokes_callback(self, mock_client: MagicMock) -> None:
120 """Test that on_stream_end clears extractor and calls callback."""
121 role = SynchronizerRole(mock_client)
122 end_cb = MagicMock()
123 role.set_callbacks(
124 on_visualization_data=MagicMock(),
125 on_stream_start=MagicMock(),
126 on_stream_end=end_cb,
127 )
128 role.on_stream_end()
129 end_cb.assert_called_once()
130 assert role._extractor is None
131
132 def test_on_audio_chunk_without_extractor_is_noop(self, mock_client: MagicMock) -> None:
133 """Test that audio chunks are ignored before stream starts."""
134 role = SynchronizerRole(mock_client)
135 viz_cb = MagicMock()
136 role.set_callbacks(
137 on_visualization_data=viz_cb,
138 on_stream_start=MagicMock(),
139 on_stream_end=MagicMock(),
140 )
141 chunk = MagicMock()
142 chunk.data = b"\x00" * 100
143 chunk.timestamp_us = 0
144 role.on_audio_chunk(chunk)
145 viz_cb.assert_not_called()
146
147 def test_on_audio_chunk_with_extractor_calls_callback(self, mock_client: MagicMock) -> None:
148 """Test that audio chunks produce visualization data callback."""
149 role = SynchronizerRole(mock_client)
150 viz_cb = MagicMock()
151 role.set_callbacks(
152 on_visualization_data=viz_cb,
153 on_stream_start=MagicMock(),
154 on_stream_end=MagicMock(),
155 )
156 role.setup_audio_requirements()
157 role.setup_stream_config()
158
159 mock_extractor = MagicMock()
160 mock_frame = MagicMock()
161 mock_extractor.process_chunk.return_value = [mock_frame]
162 role._extractor = mock_extractor
163
164 chunk = MagicMock()
165 chunk.data = b"\x00" * 100
166 chunk.timestamp_us = 1000000
167 role.on_audio_chunk(chunk)
168
169 mock_extractor.process_chunk.assert_called_once_with(chunk.data, chunk.timestamp_us)
170 viz_cb.assert_called_once_with(mock_frame)
171