/
/
/
1"""Tests for the streamserver publish IP resolution."""
2
3from unittest.mock import AsyncMock, patch
4
5from aiohttp.test_utils import unused_port
6
7from music_assistant.constants import (
8 CONF_BIND_IP,
9 CONF_BIND_PORT,
10 CONF_PUBLISH_IP,
11 CONF_VALUE_AUTO,
12)
13from music_assistant.controllers.streams.controller import StreamsController
14from music_assistant.mass import MusicAssistant
15
16# TEST-NET-3 (RFC 5737) represents an unavailable configured address
17UNBINDABLE_IP = "203.0.113.7"
18# a publish IP does not have to exist on this host at all (NAT/port forward setups)
19EXTERNAL_PUBLISH_IP = "203.0.113.9"
20LOOPBACK_IP = "127.0.0.1"
21FALLBACK_PORT = 8097
22# the address the tests bind is deliberately absent here, so a publish IP that follows
23# the bind address can never be mistaken for the one auto-detection would have picked
24ALL_ADDRESSES = ("192.168.1.10", "10.0.0.5", "fd00::10")
25
26
27async def _setup_streams(
28 controller: StreamsController,
29 mass: MusicAssistant,
30 bind_ip: str,
31 publish_ip: str = CONF_VALUE_AUTO,
32 all_ip_addresses: tuple[str, ...] = ALL_ADDRESSES,
33 bind_port: int | None = None,
34) -> None:
35 """
36 Run the streamserver's setup against the given bind and publish configuration.
37
38 :param controller: The StreamsController to set up.
39 :param mass: The MusicAssistant instance owning the controller.
40 :param bind_ip: Address to configure as bind IP.
41 :param publish_ip: Address to configure as publish IP ("auto" to have it resolved).
42 :param all_ip_addresses: Host addresses the setup detects, in ranked order.
43 :param bind_port: Port to configure as bind port (0 to have the OS assign one).
44 """
45 config = await mass.config.get_core_config(controller.domain)
46 config.update(
47 {
48 CONF_BIND_IP: bind_ip,
49 CONF_PUBLISH_IP: publish_ip,
50 CONF_BIND_PORT: unused_port() if bind_port is None else bind_port,
51 }
52 )
53 with (
54 patch(
55 "music_assistant.controllers.streams.controller.get_publish_ip_candidates",
56 AsyncMock(return_value=all_ip_addresses),
57 ),
58 patch(
59 "music_assistant.controllers.streams.controller.check_ffmpeg_version",
60 AsyncMock(),
61 ),
62 ):
63 await controller.setup(config)
64
65
66async def test_specific_bind_ip_is_published(
67 streams_controller: StreamsController, mass_minimal: MusicAssistant
68) -> None:
69 """A streamserver pinned to one interface hands players that same address."""
70 await _setup_streams(streams_controller, mass_minimal, bind_ip=LOOPBACK_IP)
71
72 assert streams_controller.publish_ip == LOOPBACK_IP
73 assert streams_controller.base_url == f"http://{LOOPBACK_IP}:{streams_controller.publish_port}"
74
75
76async def test_wildcard_bind_publishes_primary_address(
77 streams_controller: StreamsController, mass_minimal: MusicAssistant
78) -> None:
79 """A streamserver on all interfaces advertises the primary detected address."""
80 await _setup_streams(streams_controller, mass_minimal, bind_ip="0.0.0.0")
81
82 assert streams_controller.publish_ip == ALL_ADDRESSES[0]
83 assert (
84 streams_controller.base_url
85 == f"http://{ALL_ADDRESSES[0]}:{streams_controller.publish_port}"
86 )
87
88
89async def test_configured_publish_ip_outranks_bind_ip(
90 streams_controller: StreamsController, mass_minimal: MusicAssistant
91) -> None:
92 """An explicitly configured publish IP stays authoritative over the bind address."""
93 await _setup_streams(
94 streams_controller, mass_minimal, bind_ip=LOOPBACK_IP, publish_ip=EXTERNAL_PUBLISH_IP
95 )
96
97 assert streams_controller.publish_ip == EXTERNAL_PUBLISH_IP
98 assert (
99 streams_controller.base_url
100 == f"http://{EXTERNAL_PUBLISH_IP}:{streams_controller.publish_port}"
101 )
102
103
104async def test_unavailable_bind_ip_publishes_dialable_address(
105 streams_controller: StreamsController,
106 mass_minimal: MusicAssistant,
107 streamserver_fallback: AsyncMock,
108) -> None:
109 """A bind that fell back to all interfaces advertises a reachable address, not the failed one."""
110 await _setup_streams(
111 streams_controller,
112 mass_minimal,
113 bind_ip=UNBINDABLE_IP,
114 bind_port=FALLBACK_PORT,
115 )
116
117 streamserver_fallback.assert_awaited_once()
118 assert (setup_call := streamserver_fallback.await_args) is not None
119 assert setup_call.kwargs["bind_ip"] == UNBINDABLE_IP
120 assert setup_call.kwargs["bind_port"] == FALLBACK_PORT
121 assert streams_controller.bind_ip == "0.0.0.0"
122 assert streams_controller.publish_ip == ALL_ADDRESSES[0]
123 assert (
124 streams_controller.base_url
125 == f"http://{ALL_ADDRESSES[0]}:{streams_controller.publish_port}"
126 )
127
128
129async def test_os_assigned_port_lands_in_base_url(
130 streams_controller: StreamsController, mass_minimal: MusicAssistant
131) -> None:
132 """A port the OS only assigns at bind time ends up in the URLs handed to players."""
133 await _setup_streams(streams_controller, mass_minimal, bind_ip=LOOPBACK_IP, bind_port=0)
134
135 assert streams_controller.publish_port != 0
136 assert streams_controller.base_url == f"http://{LOOPBACK_IP}:{streams_controller.publish_port}"
137
138
139async def test_ipv6_publish_ip_is_bracketed_in_base_url(
140 streams_controller: StreamsController, mass_minimal: MusicAssistant
141) -> None:
142 """An IPv6 publish IP is bracketed in the stream URLs handed to players."""
143 await _setup_streams(
144 streams_controller, mass_minimal, bind_ip="::", all_ip_addresses=("fd00::10",)
145 )
146
147 assert streams_controller.publish_ip == "fd00::10"
148 assert streams_controller.base_url == f"http://[fd00::10]:{streams_controller.publish_port}"
149