Support IPv6 and hostname addressed devices for UPnP eventing

src/wiim/wiim_device.py

@@ -7,6 +7,7 @@
 from urllib.parse import urlparse, urljoin
 from datetime import timedelta
 import xml.etree.ElementTree as ET
+from ipaddress import IPv4Address, IPv6Address, ip_address
 from html import unescape
 from contextlib import suppress
 import time
@@ -279,13 +280,15 @@ async def async_init_services_and_subscribe(self) -> bool:
                 loop = asyncio.get_event_loop()
                 local_ip = self.local_host
                 device_ip = self.ip_address
-                if device_ip:
-                    last_octet = int(device_ip.split(".")[-1])
+                device_address = self._parse_ip_address(device_ip)
+                if device_address is not None:
+                    last_octet = device_address.packed[-1]
                 else:
                     last_octet = 0
                 base_port = 50000
                 assigned_port = base_port + last_octet
-                source_ip = local_ip or "0.0.0.0"
+                is_ipv6 = device_address is not None and device_address.version == 6
+                source_ip = local_ip or ("::" if is_ipv6 else "0.0.0.0")
                 source = (source_ip, assigned_port)
 
                 if self.av_transport:
@@ -2018,6 +2021,16 @@ def loop_state(self) -> WiimLoopState:
             WiimLoopState(repeat=WiimRepeatMode.OFF, shuffle=False),
         )
 
+    @staticmethod
+    def _parse_ip_address(host: str | None) -> IPv4Address | IPv6Address | None:
+        """Return the device address, or None if it is not a literal address."""
+        if not host:
+            return None
+        try:
+            return ip_address(host)
+        except ValueError:
+            return None
+
     @staticmethod
     def build_loop_mode(repeat: WiimRepeatMode, shuffle: bool) -> LoopMode:
         """Return the SDK loop mode for the given repeat and shuffle settings."""

tests/wiim/test_wiim_device.py

@@ -29,7 +29,8 @@ def _build_upnp_device(
     upnp_device.friendly_name = name
     upnp_device.manufacturer = "Linkplay"
     upnp_device.model_name = model_name
-    upnp_device.device_url = f"http://{ip_address}:49152/description.xml"
+    host = f"[{ip_address}]" if ":" in ip_address else ip_address
+    upnp_device.device_url = f"http://{host}:49152/description.xml"
 
     def _build_service() -> MagicMock:
         service = MagicMock()
@@ -165,6 +166,37 @@ async def test_http_actions_are_called_correctly(
         await device.async_set_volume(75)
         device._http_command_ok.assert_called_with(WiimHttpCommand.SET_VOLUME, "75")
 
+    @pytest.mark.parametrize(
+        ("device_ip", "expected_port", "expected_source_ip"),
+        [
+            pytest.param("192.168.1.100", 50100, "0.0.0.0", id="ipv4"),
+            pytest.param("2001:db8::5", 50005, "::", id="ipv6"),
+            pytest.param("wiim.local", 50000, "0.0.0.0", id="hostname"),
+        ],
+    )
+    @pytest.mark.asyncio
+    async def test_event_listener_source_matches_device_family(
+        self,
+        mock_session,
+        device_ip,
+        expected_port,
+        expected_source_ip,
+    ):
+        """Test the notify server binds an address family the device can reach."""
+        upnp_device = _build_upnp_device(
+            udn="uuid:test", name="WiiM", ip_address=device_ip
+        )
+        device = WiimDevice(upnp_device, mock_session)
+
+        with patch("wiim.wiim_device.AiohttpNotifyServer") as notify_server:
+            notify_server.return_value.async_start_server = AsyncMock()
+            await device.async_init_services_and_subscribe()
+
+        assert notify_server.call_args.kwargs["source"] == (
+            expected_source_ip,
+            expected_port,
+        )
+
     def test_parse_duration(self, mock_upnp_device, mock_session):
         """Test the parsing of various duration string formats."""
         device = WiimDevice(mock_upnp_device, mock_session)