/
/
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) is never assigned to a host, so binding it always fails
12UNBINDABLE_IP = "203.0.113.7"
13ALL_ADDRESSES = ("192.168.1.10", "fd00::10")
14
15
16async def _setup_with_bind_ip(
17 controller: StreamsController, mass: MusicAssistant, bind_ip: str
18) -> None:
19 """
20 Run the streamserver's setup against a config that binds the given address.
21
22 :param controller: The StreamsController to set up.
23 :param mass: The MusicAssistant instance owning the controller.
24 :param bind_ip: Address to configure as bind IP.
25 """
26 config = await mass.config.get_core_config(controller.domain)
27 config.update({CONF_BIND_IP: bind_ip, CONF_BIND_PORT: unused_port()})
28 with (
29 patch(
30 "music_assistant.controllers.streams.controller.get_publish_ip_candidates",
31 AsyncMock(return_value=ALL_ADDRESSES),
32 ),
33 patch(
34 "music_assistant.controllers.streams.controller.check_ffmpeg_version",
35 AsyncMock(),
36 ),
37 ):
38 await controller.setup(config)
39
40
41async def test_unavailable_bind_ip_publishes_dialable_addresses(
42 streams_controller: StreamsController, mass_minimal: MusicAssistant
43) -> None:
44 """An address that cannot be bound leaves the streamserver advertising reachable addresses."""
45 await _setup_with_bind_ip(streams_controller, mass_minimal, UNBINDABLE_IP)
46
47 assert streams_controller.bind_ip == "0.0.0.0"
48 assert streams_controller._publish_addresses == list(ALL_ADDRESSES)
49
50
51async def test_available_bind_ip_publishes_only_that_address(
52 streams_controller: StreamsController, mass_minimal: MusicAssistant
53) -> None:
54 """A bind that succeeded advertises exactly the interface the streamserver is pinned to."""
55 await _setup_with_bind_ip(streams_controller, mass_minimal, "127.0.0.1")
56
57 assert streams_controller.bind_ip == "127.0.0.1"
58 assert streams_controller._publish_addresses == ["127.0.0.1"]
59