music-assistant-server

1"""Tests for music_assistant.providers.snapcast.ma_stream._register_tcp_server_source."""
2
3from __future__ import annotations
4
5import asyncio
6from typing import TYPE_CHECKING
7from unittest.mock import MagicMock
8
9import pytest
10from music_assistant_models.enums import ContentType
11
12from music_assistant.providers.snapcast.ma_stream import SnapcastMAStream
13
14if TYPE_CHECKING:
15    from .conftest import FakeSnapserver
16
17
18def _make_stream(
19    provider: MagicMock, name: str = "Music Assistant - testhash (announcement)"
20) -> SnapcastMAStream:
21    """
22    Build a SnapcastMAStream instance directly.
23
24    Bypasses the constructor's media handling (we only test the snapserver
25    source registration).
26    """
27    media = MagicMock()
28    return SnapcastMAStream(
29        provider=provider,
30        media=media,
31        stream_name=name,
32    )
33
34
35def test_transport_format_uses_snapserver_codec() -> None:
36    """The reported final format follows the codec configured on Snapserver."""
37    provider = MagicMock()
38    provider._use_builtin_server = True
39    provider._snapcast_server_transport_codec = "opus"
40    stream = _make_stream(provider)
41
42    output_format = stream._get_transport_format()
43
44    assert output_format.content_type == ContentType.OPUS
45    assert output_format.codec_type == ContentType.OPUS
46    assert output_format.sample_rate == 48000
47    assert output_format.bit_depth == 16
48
49
50def test_external_transport_format_reads_uri_codec() -> None:
51    """External Snapserver codec is read from the stream URI query."""
52    provider = MagicMock()
53    provider._use_builtin_server = False
54    stream = _make_stream(provider)
55    stream.snap_stream = MagicMock(
56        _stream={"uri": {"query": {"codec": "flac"}}},
57    )
58
59    output_format = stream._get_transport_format()
60
61    assert output_format.content_type == ContentType.FLAC
62    assert output_format.codec_type == ContentType.FLAC
63
64
65def test_output_plan_is_registered_for_all_snapcast_members() -> None:
66    """Every client consuming a shared Snapcast stream gets the same output path."""
67    provider = MagicMock()
68    stream = _make_stream(provider)
69    stream.media.source_id = "queue-1"
70    stream.media.custom_data = {"session_id": "session-1"}
71    stream.snap_stream = MagicMock(identifier="stream-1")
72    group = MagicMock(
73        stream="stream-1",
74        clients=["snap-child"],
75    )
76    provider._snapserver.groups = [group]
77    provider._get_ma_id.return_value = "child"
78    output_details = MagicMock(player_ids=["leader"])
79    stream._output_plan = MagicMock(output_details=output_details)
80
81    stream._register_output_plan()
82
83    registered_ids = {
84        call.args[0] for call in provider.mass.streams.audio_processing.update_output.call_args_list
85    }
86    assert registered_ids == {"leader", "child"}
87
88
89@pytest.mark.asyncio
90async def test_happy_path_register_succeeds_first_attempt(
91    fake_provider: MagicMock, fake_snapserver: FakeSnapserver
92) -> None:
93    """Regression: a clean snapserver returns id on first try, MA registers the stream."""
94    fake_snapserver.queue_success(stream_id="ok-1")
95    stream = _make_stream(fake_provider)
96
97    await stream._register_tcp_server_source()
98
99    assert stream.snap_stream is not None
100    assert stream.snap_stream.identifier == "ok-1"
101    assert len(fake_snapserver.add_stream_calls) == 1
102
103
104@pytest.mark.asyncio
105async def test_real_port_conflict_retries_with_different_port(
106    fake_provider: MagicMock, fake_snapserver: FakeSnapserver
107) -> None:
108    """Regression: a non-name error (e.g. port already bound) keeps the loop going."""
109    fake_snapserver.queue_other_error("bind: Address already in use")
110    fake_snapserver.queue_success(stream_id="ok-after-retry")
111
112    stream = _make_stream(fake_provider)
113    await stream._register_tcp_server_source()
114
115    assert stream.snap_stream is not None
116    assert stream.snap_stream.identifier == "ok-after-retry"
117    # Two add_stream calls were made — first failed, second succeeded
118    assert len(fake_snapserver.add_stream_calls) == 2
119    # The two URIs must use different ports
120    port_1 = fake_snapserver.add_stream_calls[0].split("0.0.0.0:")[1].split("?")[0]
121    port_2 = fake_snapserver.add_stream_calls[1].split("0.0.0.0:")[1].split("?")[0]
122    assert port_1 != port_2
123
124
125@pytest.mark.asyncio
126async def test_all_attempts_exhausted_raises_with_honest_message(
127    fake_provider: MagicMock, fake_snapserver: FakeSnapserver
128) -> None:
129    """
130    When all 50 retries fail the error must reference 'after retries' or similar.
131
132    Must NOT say 'No free port found' which lies about the cause.
133    """
134    # Queue 51 errors so even after 50 attempts the loop hits the raise
135    for _ in range(51):
136        fake_snapserver.queue_other_error("Some persistent error")
137
138    stream = _make_stream(fake_provider)
139
140    with pytest.raises(RuntimeError) as exc_info:
141        await stream._register_tcp_server_source()
142
143    message = str(exc_info.value)
144    assert "No free port" not in message, (
145        f"Error message still claims port shortage when the real cause may differ: {message}"
146    )
147    assert "attempts" in message.lower() or "register" in message.lower()
148
149
150@pytest.mark.parametrize(
151    ("result", "expected"),
152    [
153        (
154            {
155                "code": -32603,
156                "data": 'Stream with name "x" already exists',
157                "message": "Internal error",
158            },
159            True,
160        ),
161        (
162            {
163                "code": -32603,
164                "data": 'Stream with name "abc" already exists in registry',
165                "message": "Internal error",
166            },
167            True,
168        ),
169        # negative cases — these must NOT match
170        (
171            {"code": -32603, "data": "bind: Address already in use", "message": "Internal error"},
172            False,
173        ),
174        (
175            {
176                "code": -32603,
177                "data": "Some other error already exists somewhere",
178                "message": "Internal error",
179            },
180            False,
181        ),
182        ({}, False),
183        (None, False),
184        ("plain string", False),
185        ({"data": 12345}, False),
186    ],
187)
188def test_is_name_collision_error_matches_only_specific_pattern(
189    result: object, expected: bool
190) -> None:
191    """_is_name_collision_error must NOT false-positive on unrelated errors."""
192    assert SnapcastMAStream._is_name_collision_error(result) is expected
193
194
195@pytest.mark.asyncio
196async def test_name_collision_with_local_stream_cached_adopts_it(
197    fake_provider: MagicMock, fake_snapserver: FakeSnapserver
198) -> None:
199    """
200    When snapserver reports the name as already-registered MA must adopt the orphan.
201
202    The stream must be in the local snapserver cache; MA must adopt it instead of
203    looping until retries exhaust.
204    """
205    target_name = "Music Assistant - 590b15 (announcement)"
206    fake_snapserver.cache_stream_directly("orphan-id", target_name)
207    fake_snapserver.queue_name_collision()
208
209    stream = _make_stream(fake_provider, name=target_name)
210    await stream._register_tcp_server_source()
211
212    assert stream.snap_stream is not None
213    assert stream.snap_stream.identifier == "orphan-id"
214    # Adoption must NOT spend further retries
215    assert len(fake_snapserver.add_stream_calls) == 1
216
217
218@pytest.mark.asyncio
219async def test_orphan_stream_after_ma_restart_gets_adopted_via_resync(
220    fake_provider: MagicMock, fake_snapserver: FakeSnapserver
221) -> None:
222    """
223    Bug B core scenario: MA restarted while a stream was registered.
224
225    The local snapserver cache is empty, but the server still holds the orphan.
226    MA must call status() + synchronize() to re-discover it before adopting.
227    """
228    target_name = "Music Assistant - 590b15 (announcement)"
229    # Stream visible only via status(), NOT in the local cache yet
230    fake_snapserver.stage_orphan_stream("orphan-id", target_name)
231    fake_snapserver.queue_name_collision()
232
233    # Sanity: the local cache is empty before
234    assert fake_snapserver._streams_by_id == {}
235
236    stream = _make_stream(fake_provider, name=target_name)
237    await stream._register_tcp_server_source()
238
239    assert stream.snap_stream is not None
240    assert stream.snap_stream.identifier == "orphan-id"
241    # The status() round-trip must have happened
242    assert fake_snapserver.status.await_count >= 1
243
244
245@pytest.mark.asyncio
246async def test_second_register_call_is_idempotent_when_stream_already_set(
247    fake_provider: MagicMock, fake_snapserver: FakeSnapserver
248) -> None:
249    """
250    Calling _register_tcp_server_source after self.snap_stream is set is a no-op.
251
252    This is the early-return guard at the top of the method. Real concurrency
253    safety is provided by `_lifecycle_lock` in setup(); that path is exercised
254    by integration tests, not unit tests.
255    """
256    fake_snapserver.queue_success(stream_id="single-stream")
257    fake_snapserver.queue_success(stream_id="should-not-happen")
258
259    stream = _make_stream(fake_provider)
260
261    await asyncio.gather(
262        stream._register_tcp_server_source(),
263        stream._register_tcp_server_source(),
264    )
265
266    assert stream.snap_stream is not None
267    assert stream.snap_stream.identifier == "single-stream"
268    # Only ONE add_stream call was issued
269    assert len(fake_snapserver.add_stream_calls) == 1
270
271
272def test_pick_port_avoids_already_tried() -> None:
273    """Within a single retry loop, _pick_port_avoiding must not return a tried port."""
274    # Build a stream stub that exposes only what _pick_port_avoiding needs
275    stream = SnapcastMAStream.__new__(SnapcastMAStream)
276
277    used: set[int] = set()
278    for _ in range(100):
279        port = stream._pick_port_avoiding(used)
280        assert port is not None, "Helper unexpectedly returned None when ports remain"
281        assert 4953 <= port <= 4953 + 200
282        assert port not in used
283        used.add(port)
284