/
/
/
1"""Tests for the fallback to all interfaces when the configured bind IP is unavailable."""
2
3import logging
4from collections.abc import AsyncGenerator
5from typing import TYPE_CHECKING, cast
6from unittest.mock import AsyncMock, MagicMock, patch
7
8import aiohttp
9import pytest
10from aiohttp.test_utils import unused_port
11
12from music_assistant.constants import CONF_BIND_IP, CONF_BIND_PORT
13from music_assistant.controllers.webserver.controller import WebserverController
14from music_assistant.helpers.webserver import Webserver
15from music_assistant.mass import MusicAssistant
16
17if TYPE_CHECKING:
18 from music_assistant_models.config_entries import CoreConfig
19
20# TEST-NET-3 (RFC 5737) is never assigned to a host, so binding it always fails
21UNBINDABLE_IP = "203.0.113.7"
22ALL_ADDRESSES = ("192.168.1.10", "fd00::10")
23
24
25@pytest.fixture
26async def server() -> AsyncGenerator[Webserver]:
27 """Yield a bare Webserver, guaranteeing its socket is released afterwards."""
28 webserver = Webserver(logging.getLogger(__name__))
29 try:
30 yield webserver
31 finally:
32 await webserver.close()
33
34
35@pytest.fixture
36def controller() -> WebserverController:
37 """Create a WebserverController carrying the state that setup() resolves."""
38 mass = MagicMock()
39 mass.config.get_raw_core_config_value.return_value = "GLOBAL"
40 webserver = WebserverController(mass)
41 config = MagicMock()
42 config.get_value.return_value = False
43 webserver.config = cast("CoreConfig", config)
44 webserver.publish_port = 8095
45 return webserver
46
47
48@pytest.fixture
49async def booted_controller(mass_minimal: MusicAssistant) -> AsyncGenerator[WebserverController]:
50 """
51 Yield a WebserverController attached to a minimal server, closed afterwards.
52
53 :param mass_minimal: Minimal MusicAssistant instance.
54 """
55 controller = WebserverController(mass_minimal)
56 mass_minimal.webserver = controller
57 try:
58 yield controller
59 finally:
60 # close unconditionally: a failed assertion must not leave the socket bound
61 await controller.close()
62
63
64async def test_unavailable_bind_ip_falls_back_to_all_interfaces(server: Webserver) -> None:
65 """An address that cannot be bound starts the server on all interfaces instead."""
66 port = unused_port()
67
68 await server.setup(bind_ip=UNBINDABLE_IP, bind_port=port)
69
70 assert server.bind_ip is None
71 assert server.port == port
72 # no routes are registered on a bare Webserver, so a 404 proves it is really serving
73 async with (
74 aiohttp.ClientSession() as session,
75 session.get(f"http://127.0.0.1:{port}/info") as response,
76 ):
77 assert response.status == 404
78
79
80async def test_available_bind_ip_is_reported(server: Webserver) -> None:
81 """A bind that succeeded reports the address it is pinned to."""
82 port = unused_port()
83
84 await server.setup(bind_ip="127.0.0.1", bind_port=port)
85
86 assert server.bind_ip == "127.0.0.1"
87 assert server.port == port
88
89
90async def test_wildcard_bind_ip_is_reported_as_all_interfaces(server: Webserver) -> None:
91 """A configured wildcard is reported the same way as a fallback: no pinned address."""
92 port = unused_port()
93
94 await server.setup(bind_ip="0.0.0.0", bind_port=port)
95
96 assert server.bind_ip is None
97
98
99def test_fallback_publishes_only_dialable_addresses(controller: WebserverController) -> None:
100 """After a fallback, the un-bindable address is neither advertised nor dialed."""
101 # what setup() resolves from the configured address, before the bind is attempted
102 controller._resolve_publish_state(UNBINDABLE_IP, ALL_ADDRESSES, "http")
103 assert controller.base_url == f"http://{UNBINDABLE_IP}:8095"
104
105 # ...and what it resolves once the webserver reports it fell back to all interfaces
106 controller._resolve_publish_state(None, ALL_ADDRESSES, "http")
107
108 assert controller.bind_ip is None
109 assert controller.internal_base_url == "http://127.0.0.1:8095"
110 assert controller.publish_addresses == ["192.168.1.10", "fd00::10"]
111 assert controller.base_url == "http://192.168.1.10:8095"
112
113
114def test_successful_bind_pins_to_the_configured_address(controller: WebserverController) -> None:
115 """A bind that succeeded keeps advertising exactly the address it is pinned to."""
116 controller._resolve_publish_state("192.168.1.10", ALL_ADDRESSES, "http")
117
118 assert controller.bind_ip == "192.168.1.10"
119 assert controller.publish_ip == "192.168.1.10"
120 assert controller.publish_addresses == ["192.168.1.10"]
121 assert controller.base_url == "http://192.168.1.10:8095"
122 assert controller.internal_base_url == "http://192.168.1.10:8095"
123
124
125async def test_setup_publishes_dialable_addresses_after_fallback(
126 booted_controller: WebserverController, mass_minimal: MusicAssistant
127) -> None:
128 """Setting up on an un-bindable address leaves the webserver advertising what answers."""
129 port = unused_port()
130 config = await mass_minimal.config.get_core_config(booted_controller.domain)
131 config.update({CONF_BIND_IP: UNBINDABLE_IP, CONF_BIND_PORT: port})
132
133 with patch(
134 "music_assistant.controllers.webserver.controller.get_publish_ip_candidates",
135 AsyncMock(return_value=ALL_ADDRESSES),
136 ):
137 await booted_controller.setup(config)
138
139 assert booted_controller.bind_ip is None
140 assert booted_controller.publish_addresses == list(ALL_ADDRESSES)
141 assert booted_controller.base_url == f"http://192.168.1.10:{port}"
142 assert booted_controller.internal_base_url == f"http://127.0.0.1:{port}"
143