/
/
/
1"""Tests for the streamserver adopting the address it actually bound to."""
2
3from unittest.mock import AsyncMock, patch
4
5from aiohttp.test_utils import unused_port
6
7from music_assistant.constants import CONF_BIND_IP, CONF_BIND_PORT
8from music_assistant.controllers.streams.controller import StreamsController
9from music_assistant.mass import MusicAssistant
10
11# TEST-NET-3 (RFC 5737) represents an unavailable configured address
12UNBINDABLE_IP = "203.0.113.7"
13ALL_ADDRESSES = ("192.168.1.10", "fd00::10")
14FALLBACK_PORT = 8097
15
16
17async def _setup_with_bind_ip(
18 controller: StreamsController,
19 mass: MusicAssistant,
20 bind_ip: str,
21 bind_port: int | None = None,
22) -> None:
23 """
24 Run the streamserver's setup against a config that binds the given address.
25
26 :param controller: The StreamsController to set up.
27 :param mass: The MusicAssistant instance owning the controller.
28 :param bind_ip: Address to configure as bind IP.
29 :param bind_port: Port to configure as bind port.
30 """
31 config = await mass.config.get_core_config(controller.domain)
32 config.update(
33 {
34 CONF_BIND_IP: bind_ip,
35 CONF_BIND_PORT: unused_port() if bind_port is None else bind_port,
36 }
37 )
38 with (
39 patch(
40 "music_assistant.controllers.streams.controller.get_publish_ip_candidates",
41 AsyncMock(return_value=ALL_ADDRESSES),
42 ),
43 patch(
44 "music_assistant.controllers.streams.controller.check_ffmpeg_version",
45 AsyncMock(),
46 ),
47 ):
48 await controller.setup(config)
49
50
51async def test_unavailable_bind_ip_publishes_dialable_addresses(
52 streams_controller: StreamsController,
53 mass_minimal: MusicAssistant,
54 streamserver_fallback: AsyncMock,
55) -> None:
56 """An address that cannot be bound leaves the streamserver advertising reachable addresses."""
57 await _setup_with_bind_ip(
58 streams_controller, mass_minimal, UNBINDABLE_IP, bind_port=FALLBACK_PORT
59 )
60
61 streamserver_fallback.assert_awaited_once()
62 assert (setup_call := streamserver_fallback.await_args) is not None
63 assert setup_call.kwargs["bind_ip"] == UNBINDABLE_IP
64 assert setup_call.kwargs["bind_port"] == FALLBACK_PORT
65 assert streams_controller.bind_ip == "0.0.0.0"
66 assert streams_controller._publish_addresses == list(ALL_ADDRESSES)
67
68
69async def test_available_bind_ip_publishes_only_that_address(
70 streams_controller: StreamsController, mass_minimal: MusicAssistant
71) -> None:
72 """A bind that succeeded advertises exactly the interface the streamserver is pinned to."""
73 await _setup_with_bind_ip(streams_controller, mass_minimal, "127.0.0.1")
74
75 assert streams_controller.bind_ip == "127.0.0.1"
76 assert streams_controller._publish_addresses == ["127.0.0.1"]
77