/
/
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) is never assigned to a host, so binding it always fails
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"
21# the address the tests bind is deliberately absent here, so a publish IP that follows
22# the bind address can never be mistaken for the one auto-detection would have picked
23ALL_ADDRESSES = ("192.168.1.10", "10.0.0.5", "fd00::10")
24
25
26async def _setup_streams(
27 controller: StreamsController,
28 mass: MusicAssistant,
29 bind_ip: str,
30 publish_ip: str = CONF_VALUE_AUTO,
31 all_ip_addresses: tuple[str, ...] = ALL_ADDRESSES,
32 bind_port: int | None = None,
33) -> None:
34 """
35 Run the streamserver's setup against the given bind and publish configuration.
36
37 :param controller: The StreamsController to set up.
38 :param mass: The MusicAssistant instance owning the controller.
39 :param bind_ip: Address to configure as bind IP.
40 :param publish_ip: Address to configure as publish IP ("auto" to have it resolved).
41 :param all_ip_addresses: Host addresses the setup detects, in ranked order.
42 :param bind_port: Port to configure as bind port (0 to have the OS assign one).
43 """
44 config = await mass.config.get_core_config(controller.domain)
45 config.update(
46 {
47 CONF_BIND_IP: bind_ip,
48 CONF_PUBLISH_IP: publish_ip,
49 CONF_BIND_PORT: unused_port() if bind_port is None else bind_port,
50 }
51 )
52 with (
53 patch(
54 "music_assistant.controllers.streams.controller.get_publish_ip_candidates",
55 AsyncMock(return_value=all_ip_addresses),
56 ),
57 patch(
58 "music_assistant.controllers.streams.controller.check_ffmpeg_version",
59 AsyncMock(),
60 ),
61 ):
62 await controller.setup(config)
63
64
65async def test_specific_bind_ip_is_published(
66 streams_controller: StreamsController, mass_minimal: MusicAssistant
67) -> None:
68 """A streamserver pinned to one interface hands players that same address."""
69 await _setup_streams(streams_controller, mass_minimal, bind_ip=LOOPBACK_IP)
70
71 assert streams_controller.publish_ip == LOOPBACK_IP
72 assert streams_controller.base_url == f"http://{LOOPBACK_IP}:{streams_controller.publish_port}"
73
74
75async def test_wildcard_bind_publishes_primary_address(
76 streams_controller: StreamsController, mass_minimal: MusicAssistant
77) -> None:
78 """A streamserver on all interfaces advertises the primary detected address."""
79 await _setup_streams(streams_controller, mass_minimal, bind_ip="0.0.0.0")
80
81 assert streams_controller.publish_ip == ALL_ADDRESSES[0]
82 assert (
83 streams_controller.base_url
84 == f"http://{ALL_ADDRESSES[0]}:{streams_controller.publish_port}"
85 )
86
87
88async def test_configured_publish_ip_outranks_bind_ip(
89 streams_controller: StreamsController, mass_minimal: MusicAssistant
90) -> None:
91 """An explicitly configured publish IP stays authoritative over the bind address."""
92 await _setup_streams(
93 streams_controller, mass_minimal, bind_ip=LOOPBACK_IP, publish_ip=EXTERNAL_PUBLISH_IP
94 )
95
96 assert streams_controller.publish_ip == EXTERNAL_PUBLISH_IP
97 assert (
98 streams_controller.base_url
99 == f"http://{EXTERNAL_PUBLISH_IP}:{streams_controller.publish_port}"
100 )
101
102
103async def test_unavailable_bind_ip_publishes_dialable_address(
104 streams_controller: StreamsController, mass_minimal: MusicAssistant
105) -> None:
106 """A bind that fell back to all interfaces advertises a reachable address, not the failed one."""
107 await _setup_streams(streams_controller, mass_minimal, bind_ip=UNBINDABLE_IP)
108
109 assert streams_controller.publish_ip == ALL_ADDRESSES[0]
110 assert (
111 streams_controller.base_url
112 == f"http://{ALL_ADDRESSES[0]}:{streams_controller.publish_port}"
113 )
114
115
116async def test_os_assigned_port_lands_in_base_url(
117 streams_controller: StreamsController, mass_minimal: MusicAssistant
118) -> None:
119 """A port the OS only assigns at bind time ends up in the URLs handed to players."""
120 await _setup_streams(streams_controller, mass_minimal, bind_ip=LOOPBACK_IP, bind_port=0)
121
122 assert streams_controller.publish_port != 0
123 assert streams_controller.base_url == f"http://{LOOPBACK_IP}:{streams_controller.publish_port}"
124
125
126async def test_ipv6_publish_ip_is_bracketed_in_base_url(
127 streams_controller: StreamsController, mass_minimal: MusicAssistant
128) -> None:
129 """An IPv6 publish IP is bracketed in the stream URLs handed to players."""
130 await _setup_streams(
131 streams_controller, mass_minimal, bind_ip="::", all_ip_addresses=("fd00::10",)
132 )
133
134 assert streams_controller.publish_ip == "fd00::10"
135 assert streams_controller.base_url == f"http://[fd00::10]:{streams_controller.publish_port}"
136