/
/
/
1"""Tests for the network addressing helpers on the streams controller."""
2
3from __future__ import annotations
4
5from unittest.mock import AsyncMock, MagicMock, patch
6
7import pytest
8
9from music_assistant.controllers.streams.controller import StreamsController
10
11DEVICE_IP = "192.168.1.50"
12DEVICE_IP_V6 = "fd00::50"
13
14
15def _streams_controller(
16 bind_ip: str = "0.0.0.0",
17 configured_publish_ip: str | None = None,
18) -> StreamsController:
19 """Build a streams controller with only its addressing state populated."""
20 mass = MagicMock()
21 mass.config.get_raw_core_config_value.return_value = "GLOBAL"
22 controller = StreamsController(mass)
23 controller._bind_ip = bind_ip
24 controller._configured_publish_ip = configured_publish_ip
25 return controller
26
27
28class TestGetSourceIp:
29 """get_source_ip returns a local, bindable address on the player-facing network."""
30
31 @pytest.mark.asyncio
32 async def test_concrete_bind_ip_wins_for_every_device(self) -> None:
33 """An operator pin narrows the whole server, so no routing lookup is needed."""
34 controller = _streams_controller(bind_ip="192.168.1.5")
35 with patch(
36 "music_assistant.controllers.streams.controller.get_source_ip_for_target",
37 new=AsyncMock(),
38 ) as routing_lookup:
39 assert await controller.get_source_ip(DEVICE_IP) == "192.168.1.5"
40 routing_lookup.assert_not_awaited()
41
42 @pytest.mark.asyncio
43 async def test_concrete_bind_ip_rejected_for_other_ip_family(self) -> None:
44 """A bind IP of the wrong family cannot carry traffic to the device."""
45 controller = _streams_controller(bind_ip="192.168.1.5")
46 assert await controller.get_source_ip(DEVICE_IP_V6) is None
47
48 @pytest.mark.asyncio
49 @pytest.mark.parametrize(
50 ("bind_ip", "device_ip", "source_ip"),
51 [("0.0.0.0", DEVICE_IP, "192.168.1.5"), ("::", DEVICE_IP_V6, "fd00::5")],
52 )
53 async def test_wildcard_bind_ip_resolves_per_device_route(
54 self, bind_ip: str, device_ip: str, source_ip: str
55 ) -> None:
56 """
57 Without an operator pin, the interface that routes to the device is used.
58
59 Every wildcard bind IP takes this path, not just the IPv4 one: a wildcard is not
60 itself a bindable address, so handing it to a caller would be a broken result.
61
62 :param bind_ip: Wildcard bind IP the streamserver is bound to.
63 :param device_ip: IP address of the device the traffic is meant for.
64 :param source_ip: Local address the per-device route lookup resolves to.
65 """
66 controller = _streams_controller(bind_ip=bind_ip)
67 with patch(
68 "music_assistant.controllers.streams.controller.get_source_ip_for_target",
69 new=AsyncMock(return_value=source_ip),
70 ) as routing_lookup:
71 assert await controller.get_source_ip(device_ip) == source_ip
72 routing_lookup.assert_awaited_once_with(device_ip)
73
74 @pytest.mark.asyncio
75 async def test_unroutable_target_pins_nothing(self) -> None:
76 """An inconclusive routing lookup leaves the choice to the routing table."""
77 controller = _streams_controller(bind_ip="0.0.0.0")
78 with patch(
79 "music_assistant.controllers.streams.controller.get_source_ip_for_target",
80 new=AsyncMock(return_value=""),
81 ):
82 assert await controller.get_source_ip(DEVICE_IP) is None
83
84 @pytest.mark.asyncio
85 async def test_shared_consumer_without_target_pins_nothing(self) -> None:
86 """A caller serving every player cannot be narrowed without an operator pin."""
87 controller = _streams_controller(bind_ip="0.0.0.0")
88 assert await controller.get_source_ip() is None
89
90 @pytest.mark.asyncio
91 async def test_shared_consumer_honours_concrete_bind_ip(self) -> None:
92 """An operator pin still applies to a caller that serves every player."""
93 controller = _streams_controller(bind_ip="192.168.1.5")
94 assert await controller.get_source_ip() == "192.168.1.5"
95
96
97class TestGetPublishIp:
98 """get_publish_ip only hands out an address the user actually configured."""
99
100 def test_configured_publish_ip_is_returned(self) -> None:
101 """An explicit publish IP is a reachability statement and is honoured verbatim."""
102 controller = _streams_controller(configured_publish_ip="10.45.0.20")
103 assert controller.get_publish_ip(DEVICE_IP) == "10.45.0.20"
104
105 def test_auto_detected_publish_ip_is_not_returned(self) -> None:
106 """An auto-detected publish IP is only a guess and must never be advertised."""
107 controller = _streams_controller()
108 controller.publish_ip = "10.45.0.20"
109 assert controller.get_publish_ip(DEVICE_IP) is None
110
111 def test_configured_publish_ip_of_other_ip_family_is_rejected(self) -> None:
112 """An address the device cannot even parse is worse than none at all."""
113 controller = _streams_controller(configured_publish_ip="10.45.0.20")
114 assert controller.get_publish_ip(DEVICE_IP_V6) is None
115
116
117@pytest.mark.asyncio
118async def test_diagnostics_report_whether_publish_ip_is_configured() -> None:
119 """Diagnostics distinguish a configured publish IP from an auto-detected one."""
120 assert (await _streams_controller().get_diagnostics())["publish_ip_configured"] is False
121 controller = _streams_controller(configured_publish_ip="10.45.0.20")
122 assert (await controller.get_diagnostics())["publish_ip_configured"] is True
123