Add zeroconf discovery to Sonos (#52655)

homeassistant/components/sonos/__init__.py

@@ -31,6 +31,7 @@ from homeassistant.helpers.dispatcher import async_dispatcher_send, dispatcher_s
 from .alarms import SonosAlarms
 from .const import (
     DATA_SONOS,
+    DATA_SONOS_DISCOVERY_MANAGER,
     DISCOVERY_INTERVAL,
     DOMAIN,
     PLATFORMS,
@@ -91,7 +92,7 @@ class SonosData:
         self.alarms: dict[str, SonosAlarms] = {}
         self.topology_condition = asyncio.Condition()
         self.hosts_heartbeat = None
-        self.ssdp_known: set[str] = set()
+        self.discovery_known: set[str] = set()
         self.boot_counts: dict[str, int] = {}
 
 
@@ -111,9 +112,7 @@ async def async_setup(hass, config):
     return True
 
 
-async def async_setup_entry(  # noqa: C901
-    hass: HomeAssistant, entry: ConfigEntry
-) -> bool:
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     """Set up Sonos from a config entry."""
     pysonos.config.EVENTS_MODULE = events_asyncio
 
@@ -123,7 +122,6 @@ async def async_setup_entry(  # noqa: C901
     data = hass.data[DATA_SONOS]
     config = hass.data[DOMAIN].get("media_player", {})
     hosts = config.get(CONF_HOSTS, [])
-    discovery_lock = asyncio.Lock()
     _LOGGER.debug("Reached async_setup_entry, config=%s", config)
 
     advertise_addr = config.get(CONF_ADVERTISE_ADDR)
@@ -137,153 +135,181 @@ async def async_setup_entry(  # noqa: C901
             deprecated_address,
         )
 
-    async def _async_stop_event_listener(event: Event) -> None:
+    manager = hass.data[DATA_SONOS_DISCOVERY_MANAGER] = SonosDiscoveryManager(
+        hass, entry, data, hosts
+    )
+    hass.async_create_task(manager.setup_platforms_and_discovery())
+    return True
+
+
+def _create_soco(ip_address: str, source: SoCoCreationSource) -> SoCo | None:
+    """Create a soco instance and return if successful."""
+    try:
+        soco = pysonos.SoCo(ip_address)
+        # Ensure that the player is available and UID is cached
+        _ = soco.uid
+        _ = soco.volume
+        return soco
+    except (OSError, SoCoException) as ex:
+        _LOGGER.warning(
+            "Failed to connect to %s player '%s': %s", source.value, ip_address, ex
+        )
+    return None
+
+
+class SonosDiscoveryManager:
+    """Manage sonos discovery."""
+
+    def __init__(
+        self, hass: HomeAssistant, entry: ConfigEntry, data: SonosData, hosts: list[str]
+    ) -> None:
+        """Init discovery manager."""
+        self.hass = hass
+        self.entry = entry
+        self.data = data
+        self.hosts = hosts
+        self.discovery_lock = asyncio.Lock()
+
+    async def _async_stop_event_listener(self, event: Event) -> None:
         await asyncio.gather(
-            *[speaker.async_unsubscribe() for speaker in data.discovered.values()],
+            *[speaker.async_unsubscribe() for speaker in self.data.discovered.values()],
             return_exceptions=True,
         )
         if events_asyncio.event_listener:
             await events_asyncio.event_listener.async_stop()
 
-    def _stop_manual_heartbeat(event: Event) -> None:
-        if data.hosts_heartbeat:
-            data.hosts_heartbeat()
-            data.hosts_heartbeat = None
+    def _stop_manual_heartbeat(self, event: Event) -> None:
+        if self.data.hosts_heartbeat:
+            self.data.hosts_heartbeat()
+            self.data.hosts_heartbeat = None
 
-    def _discovered_player(soco: SoCo) -> None:
+    def _discovered_player(self, soco: SoCo) -> None:
         """Handle a (re)discovered player."""
         try:
             speaker_info = soco.get_speaker_info(True)
             _LOGGER.debug("Adding new speaker: %s", speaker_info)
-            speaker = SonosSpeaker(hass, soco, speaker_info)
-            data.discovered[soco.uid] = speaker
+            speaker = SonosSpeaker(self.hass, soco, speaker_info)
+            self.data.discovered[soco.uid] = speaker
             for coordinator, coord_dict in [
-                (SonosAlarms, data.alarms),
-                (SonosFavorites, data.favorites),
+                (SonosAlarms, self.data.alarms),
+                (SonosFavorites, self.data.favorites),
             ]:
                 if soco.household_id not in coord_dict:
-                    new_coordinator = coordinator(hass, soco.household_id)
+                    new_coordinator = coordinator(self.hass, soco.household_id)
                     new_coordinator.setup(soco)
                     coord_dict[soco.household_id] = new_coordinator
             speaker.setup()
         except (OSError, SoCoException):
             _LOGGER.warning("Failed to add SonosSpeaker using %s", soco, exc_info=True)
 
-    def _create_soco(ip_address: str, source: SoCoCreationSource) -> SoCo | None:
-        """Create a soco instance and return if successful."""
-        try:
-            soco = pysonos.SoCo(ip_address)
-            # Ensure that the player is available and UID is cached
-            _ = soco.uid
-            _ = soco.volume
-            return soco
-        except (OSError, SoCoException) as ex:
-            _LOGGER.warning(
-                "Failed to connect to %s player '%s': %s", source.value, ip_address, ex
-            )
-        return None
-
-    def _manual_hosts(now: datetime.datetime | None = None) -> None:
+    def _manual_hosts(self, now: datetime.datetime | None = None) -> None:
         """Players from network configuration."""
-        for host in hosts:
+        for host in self.hosts:
             ip_addr = socket.gethostbyname(host)
             known_uid = next(
                 (
                     uid
-                    for uid, speaker in data.discovered.items()
+                    for uid, speaker in self.data.discovered.items()
                     if speaker.soco.ip_address == ip_addr
                 ),
                 None,
             )
 
             if known_uid:
-                dispatcher_send(hass, f"{SONOS_SEEN}-{known_uid}")
+                dispatcher_send(self.hass, f"{SONOS_SEEN}-{known_uid}")
             else:
                 soco = _create_soco(ip_addr, SoCoCreationSource.CONFIGURED)
                 if soco and soco.is_visible:
-                    _discovered_player(soco)
+                    self._discovered_player(soco)
 
-        data.hosts_heartbeat = hass.helpers.event.call_later(
-            DISCOVERY_INTERVAL.total_seconds(), _manual_hosts
+        self.data.hosts_heartbeat = self.hass.helpers.event.call_later(
+            DISCOVERY_INTERVAL.total_seconds(), self._manual_hosts
         )
 
     @callback
-    def _async_signal_update_groups(event):
-        async_dispatcher_send(hass, SONOS_GROUP_UPDATE)
+    def _async_signal_update_groups(self, _event):
+        async_dispatcher_send(self.hass, SONOS_GROUP_UPDATE)
 
-    def _discovered_ip(ip_address):
+    def _discovered_ip(self, ip_address):
         soco = _create_soco(ip_address, SoCoCreationSource.DISCOVERED)
         if soco and soco.is_visible:
-            _discovered_player(soco)
+            self._discovered_player(soco)
 
-    async def _async_create_discovered_player(uid, discovered_ip, boot_seqnum):
+    async def _async_create_discovered_player(self, uid, discovered_ip, boot_seqnum):
         """Only create one player at a time."""
-        async with discovery_lock:
-            if uid not in data.discovered:
-                await hass.async_add_executor_job(_discovered_ip, discovered_ip)
+        async with self.discovery_lock:
+            if uid not in self.data.discovered:
+                await self.hass.async_add_executor_job(
+                    self._discovered_ip, discovered_ip
+                )
                 return
 
-            if boot_seqnum and boot_seqnum > data.boot_counts[uid]:
-                data.boot_counts[uid] = boot_seqnum
-                if soco := await hass.async_add_executor_job(
+            if boot_seqnum and boot_seqnum > self.data.boot_counts[uid]:
+                self.data.boot_counts[uid] = boot_seqnum
+                if soco := await self.hass.async_add_executor_job(
                     _create_soco, discovered_ip, SoCoCreationSource.REBOOTED
                 ):
-                    async_dispatcher_send(hass, f"{SONOS_REBOOTED}-{uid}", soco)
+                    async_dispatcher_send(self.hass, f"{SONOS_REBOOTED}-{uid}", soco)
             else:
-                async_dispatcher_send(hass, f"{SONOS_SEEN}-{uid}")
+                async_dispatcher_send(self.hass, f"{SONOS_SEEN}-{uid}")
 
     @callback
-    def _async_discovered_player(info):
-        if info.get("modelName") in DISCOVERY_IGNORED_MODELS:
-            _LOGGER.debug("Ignoring device: %s", info.get("friendlyName"))
-            return
+    def _async_ssdp_discovered_player(self, info):
+        discovered_ip = urlparse(info[ssdp.ATTR_SSDP_LOCATION]).hostname
+        boot_seqnum = info.get("X-RINCON-BOOTSEQ")
         uid = info.get(ssdp.ATTR_UPNP_UDN)
         if uid.startswith("uuid:"):
             uid = uid[5:]
-        if boot_seqnum := info.get("X-RINCON-BOOTSEQ"):
-            boot_seqnum = int(boot_seqnum)
-            data.boot_counts.setdefault(uid, boot_seqnum)
-        if uid not in data.ssdp_known:
-            _LOGGER.debug("New discovery: %s", info)
-            data.ssdp_known.add(uid)
-        discovered_ip = urlparse(info[ssdp.ATTR_SSDP_LOCATION]).hostname
-        asyncio.create_task(
-            _async_create_discovered_player(uid, discovered_ip, boot_seqnum)
+        self.async_discovered_player(
+            info, discovered_ip, uid, boot_seqnum, info.get("modelName")
         )
 
-    async def setup_platforms_and_discovery():
+    @callback
+    def async_discovered_player(self, info, discovered_ip, uid, boot_seqnum, model):
+        """Handle discovery via ssdp or zeroconf."""
+        if model in DISCOVERY_IGNORED_MODELS:
+            _LOGGER.debug("Ignoring device: %s", info)
+            return
+        if boot_seqnum:
+            boot_seqnum = int(boot_seqnum)
+            self.data.boot_counts.setdefault(uid, boot_seqnum)
+        if uid not in self.data.discovery_known:
+            _LOGGER.debug("New discovery uid=%s: %s", uid, info)
+            self.data.discovery_known.add(uid)
+        asyncio.create_task(
+            self._async_create_discovered_player(uid, discovered_ip, boot_seqnum)
+        )
+
+    async def setup_platforms_and_discovery(self):
+        """Set up platforms and discovery."""
         await asyncio.gather(
             *[
-                hass.config_entries.async_forward_entry_setup(entry, platform)
+                self.hass.config_entries.async_forward_entry_setup(self.entry, platform)
                 for platform in PLATFORMS
             ]
         )
-        entry.async_on_unload(
-            hass.bus.async_listen_once(
-                EVENT_HOMEASSISTANT_START, _async_signal_update_groups
+        self.entry.async_on_unload(
+            self.hass.bus.async_listen_once(
+                EVENT_HOMEASSISTANT_START, self._async_signal_update_groups
             )
         )
-        entry.async_on_unload(
-            hass.bus.async_listen_once(
-                EVENT_HOMEASSISTANT_STOP, _async_stop_event_listener
+        self.entry.async_on_unload(
+            self.hass.bus.async_listen_once(
+                EVENT_HOMEASSISTANT_STOP, self._async_stop_event_listener
             )
         )
         _LOGGER.debug("Adding discovery job")
-        if hosts:
-            entry.async_on_unload(
-                hass.bus.async_listen_once(
-                    EVENT_HOMEASSISTANT_STOP, _stop_manual_heartbeat
+        if self.hosts:
+            self.entry.async_on_unload(
+                self.hass.bus.async_listen_once(
+                    EVENT_HOMEASSISTANT_STOP, self._stop_manual_heartbeat
                 )
             )
-            await hass.async_add_executor_job(_manual_hosts)
+            await self.hass.async_add_executor_job(self._manual_hosts)
             return
 
-        entry.async_on_unload(
+        self.entry.async_on_unload(
             ssdp.async_register_callback(
-                hass, _async_discovered_player, {"st": UPNP_ST}
+                self.hass, self._async_ssdp_discovered_player, {"st": UPNP_ST}
             )
         )
-
-    hass.async_create_task(setup_platforms_and_discovery())
-
-    return True

homeassistant/components/sonos/config_flow.py

@@ -1,10 +1,19 @@
 """Config flow for SONOS."""
+import logging
+
 import pysonos
 
+from homeassistant import config_entries
+from homeassistant.const import CONF_HOST
 from homeassistant.core import HomeAssistant
-from homeassistant.helpers import config_entry_flow
+from homeassistant.data_entry_flow import FlowResult
+from homeassistant.helpers.config_entry_flow import DiscoveryFlowHandler
+from homeassistant.helpers.typing import DiscoveryInfoType
 
-from .const import DOMAIN
+from .const import DATA_SONOS_DISCOVERY_MANAGER, DOMAIN
+from .helpers import hostname_to_uid
+
+_LOGGER = logging.getLogger(__name__)
 
 
 async def _async_has_devices(hass: HomeAssistant) -> bool:
@@ -13,4 +22,37 @@ async def _async_has_devices(hass: HomeAssistant) -> bool:
     return bool(result)
 
 
-config_entry_flow.register_discovery_flow(DOMAIN, "Sonos", _async_has_devices)
+class SonosDiscoveryFlowHandler(DiscoveryFlowHandler):
+    """Sonos discovery flow that callsback zeroconf updates."""
+
+    def __init__(self) -> None:
+        """Init discovery flow."""
+        super().__init__(DOMAIN, "Sonos", _async_has_devices)
+
+    async def async_step_zeroconf(
+        self, discovery_info: DiscoveryInfoType
+    ) -> FlowResult:
+        """Handle a flow initialized by zeroconf."""
+        hostname = discovery_info["hostname"]
+        if hostname is None or not hostname.startswith("Sonos-"):
+            return self.async_abort(reason="not_sonos_device")
+        await self.async_set_unique_id(self._domain, raise_on_progress=False)
+        host = discovery_info[CONF_HOST]
+        properties = discovery_info["properties"]
+        boot_seqnum = properties.get("bootseq")
+        model = properties.get("model")
+        uid = hostname_to_uid(hostname)
+        _LOGGER.debug(
+            "Calling async_discovered_player for %s with uid=%s and boot_seqnum=%s",
+            host,
+            uid,
+            boot_seqnum,
+        )
+        if discovery_manager := self.hass.data.get(DATA_SONOS_DISCOVERY_MANAGER):
+            discovery_manager.async_discovered_player(
+                properties, host, uid, boot_seqnum, model
+            )
+        return await self.async_step_discovery(discovery_info)
+
+
+config_entries.HANDLERS.register(DOMAIN)(SonosDiscoveryFlowHandler)

homeassistant/components/sonos/const.py

@@ -26,6 +26,7 @@ UPNP_ST = "urn:schemas-upnp-org:device:ZonePlayer:1"
 
 DOMAIN = "sonos"
 DATA_SONOS = "sonos_media_player"
+DATA_SONOS_DISCOVERY_MANAGER = "sonos_discovery_manager"
 PLATFORMS = {BINARY_SENSOR_DOMAIN, MP_DOMAIN, SENSOR_DOMAIN, SWITCH_DOMAIN}
 
 SONOS_ARTIST = "artists"
@@ -154,3 +155,5 @@ SCAN_INTERVAL = datetime.timedelta(seconds=10)
 DISCOVERY_INTERVAL = datetime.timedelta(seconds=60)
 SEEN_EXPIRE_TIME = 3.5 * DISCOVERY_INTERVAL
 SUBSCRIPTION_TIMEOUT = 1200
+
+MDNS_SERVICE = "_sonos._tcp.local."

homeassistant/components/sonos/helpers.py

@@ -9,6 +9,9 @@ from pysonos.exceptions import SoCoException, SoCoUPnPException
 
 from homeassistant.exceptions import HomeAssistantError
 
+UID_PREFIX = "RINCON_"
+UID_POSTFIX = "01400"
+
 _LOGGER = logging.getLogger(__name__)
 
 
@@ -36,3 +39,19 @@ def soco_error(errorcodes: list[str] | None = None) -> Callable:
         return wrapper
 
     return decorator
+
+
+def uid_to_short_hostname(uid: str) -> str:
+    """Convert a Sonos uid to a short hostname."""
+    hostname_uid = uid
+    if hostname_uid.startswith(UID_PREFIX):
+        hostname_uid = hostname_uid[len(UID_PREFIX) :]
+    if hostname_uid.endswith(UID_POSTFIX):
+        hostname_uid = hostname_uid[: -len(UID_POSTFIX)]
+    return f"Sonos-{hostname_uid}"
+
+
+def hostname_to_uid(hostname: str) -> str:
+    """Convert a Sonos hostname to a uid."""
+    baseuid = hostname.split("-")[1].replace(".local.", "")
+    return f"{UID_PREFIX}{baseuid}{UID_POSTFIX}"

homeassistant/components/sonos/manifest.json

@@ -5,7 +5,8 @@
   "documentation": "https://www.home-assistant.io/integrations/sonos",
   "requirements": ["pysonos==0.0.51"],
   "dependencies": ["ssdp"],
-  "after_dependencies": ["plex"],
+  "after_dependencies": ["plex", "zeroconf"],
+  "zeroconf": ["_sonos._tcp.local."],
   "ssdp": [
     {
       "st": "urn:schemas-upnp-org:device:ZonePlayer:1"

homeassistant/components/sonos/speaker.py

@@ -19,6 +19,7 @@ from pysonos.music_library import MusicLibrary
 from pysonos.plugins.sharelink import ShareLinkPlugin
 from pysonos.snapshot import Snapshot
 
+from homeassistant.components import zeroconf
 from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR_DOMAIN
 from homeassistant.components.media_player import DOMAIN as MP_DOMAIN
 from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
@@ -37,6 +38,7 @@ from .const import (
     BATTERY_SCAN_INTERVAL,
     DATA_SONOS,
     DOMAIN,
+    MDNS_SERVICE,
     PLATFORMS,
     SCAN_INTERVAL,
     SEEN_EXPIRE_TIME,
@@ -56,7 +58,7 @@ from .const import (
     SUBSCRIPTION_TIMEOUT,
 )
 from .favorites import SonosFavorites
-from .helpers import soco_error
+from .helpers import soco_error, uid_to_short_hostname
 
 EVENT_CHARGING = {
     "CHARGING": True,
@@ -498,12 +500,27 @@ class SonosSpeaker:
         self, now: datetime.datetime | None = None, will_reconnect: bool = False
     ) -> None:
         """Make this player unavailable when it was not seen recently."""
-        self._share_link_plugin = None
-
         if self._seen_timer:
             self._seen_timer()
             self._seen_timer = None
 
+        hostname = uid_to_short_hostname(self.soco.uid)
+        zcname = f"{hostname}.{MDNS_SERVICE}"
+        aiozeroconf = await zeroconf.async_get_async_instance(self.hass)
+        if await aiozeroconf.async_get_service_info(MDNS_SERVICE, zcname):
+            # We can still see the speaker via zeroconf check again later.
+            self._seen_timer = self.hass.helpers.event.async_call_later(
+                SEEN_EXPIRE_TIME.total_seconds(), self.async_unseen
+            )
+            return
+
+        _LOGGER.debug(
+            "No activity and could not locate %s on the network. Marking unavailable",
+            zcname,
+        )
+
+        self._share_link_plugin = None
+
         if self._poll_timer:
             self._poll_timer()
             self._poll_timer = None
@@ -511,7 +528,7 @@ class SonosSpeaker:
         await self.async_unsubscribe()
 
         if not will_reconnect:
-            self.hass.data[DATA_SONOS].ssdp_known.discard(self.soco.uid)
+            self.hass.data[DATA_SONOS].discovery_known.discard(self.soco.uid)
             self.async_write_entity_states()
 
     async def async_rebooted(self, soco: SoCo) -> None:

homeassistant/components/sonos/strings.json

@@ -6,6 +6,7 @@
       }
     },
     "abort": {
+      "not_sonos_device": "Discovered device is not a Sonos device",
       "single_instance_allowed": "[%key:common::config_flow::abort::single_instance_allowed%]",
       "no_devices_found": "[%key:common::config_flow::abort::no_devices_found%]"
     }

homeassistant/components/sonos/translations/en.json

@@ -2,6 +2,7 @@
     "config": {
         "abort": {
             "no_devices_found": "No devices found on the network",
+            "not_sonos_device": "Discovered device is not a Sonos device",
             "single_instance_allowed": "Already configured. Only a single configuration possible."
         },
         "step": {

homeassistant/generated/zeroconf.py

@@ -186,6 +186,11 @@ ZEROCONF = {
             "name": "brother*"
         }
     ],
+    "_sonos._tcp.local.": [
+        {
+            "domain": "sonos"
+        }
+    ],
     "_spotify-connect._tcp.local.": [
         {
             "domain": "spotify"

tests/components/sonos/test_config_flow.py

@@ -0,0 +1,92 @@
+"""Test the sonos config flow."""
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+from homeassistant import config_entries, core, setup
+from homeassistant.components.sonos.const import DATA_SONOS_DISCOVERY_MANAGER, DOMAIN
+
+
+@patch("homeassistant.components.sonos.config_flow.pysonos.discover", return_value=True)
+async def test_user_form(discover_mock: MagicMock, hass: core.HomeAssistant):
+    """Test we get the user initiated form."""
+    await setup.async_setup_component(hass, "persistent_notification", {})
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+    assert result["type"] == "form"
+    assert result["errors"] is None
+    with patch(
+        "homeassistant.components.sonos.async_setup",
+        return_value=True,
+    ) as mock_setup, patch(
+        "homeassistant.components.sonos.async_setup_entry",
+        return_value=True,
+    ) as mock_setup_entry:
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {},
+        )
+        await hass.async_block_till_done()
+
+    assert result2["type"] == "create_entry"
+    assert result2["title"] == "Sonos"
+    assert result2["data"] == {}
+    assert len(mock_setup.mock_calls) == 1
+    assert len(mock_setup_entry.mock_calls) == 1
+
+
+async def test_zeroconf_form(hass: core.HomeAssistant):
+    """Test we pass sonos devices to the discovery manager."""
+    await setup.async_setup_component(hass, "persistent_notification", {})
+    mock_manager = hass.data[DATA_SONOS_DISCOVERY_MANAGER] = MagicMock()
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": config_entries.SOURCE_ZEROCONF},
+        data={
+            "host": "192.168.4.2",
+            "hostname": "Sonos-aaa",
+            "properties": {"bootseq": "1234"},
+        },
+    )
+    assert result["type"] == "form"
+    assert result["errors"] is None
+
+    with patch(
+        "homeassistant.components.sonos.async_setup",
+        return_value=True,
+    ) as mock_setup, patch(
+        "homeassistant.components.sonos.async_setup_entry",
+        return_value=True,
+    ) as mock_setup_entry:
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {},
+        )
+        await hass.async_block_till_done()
+
+    assert result2["type"] == "create_entry"
+    assert result2["title"] == "Sonos"
+    assert result2["data"] == {}
+
+    assert len(mock_setup.mock_calls) == 1
+    assert len(mock_setup_entry.mock_calls) == 1
+    assert len(mock_manager.mock_calls) == 2
+
+
+async def test_zeroconf_form_not_sonos(hass: core.HomeAssistant):
+    """Test we abort on non-sonos devices."""
+    mock_manager = hass.data[DATA_SONOS_DISCOVERY_MANAGER] = MagicMock()
+    await setup.async_setup_component(hass, "persistent_notification", {})
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": config_entries.SOURCE_ZEROCONF},
+        data={
+            "host": "192.168.4.2",
+            "hostname": "not-aaa",
+            "properties": {"bootseq": "1234"},
+        },
+    )
+    assert result["type"] == "abort"
+    assert result["reason"] == "not_sonos_device"
+    assert len(mock_manager.mock_calls) == 0

tests/components/sonos/test_helpers.py

@@ -0,0 +1,17 @@
+"""Test the sonos config flow."""
+from __future__ import annotations
+
+from homeassistant.components.sonos.helpers import (
+    hostname_to_uid,
+    uid_to_short_hostname,
+)
+
+
+async def test_uid_to_short_hostname():
+    """Test we can convert a uid to a short hostname."""
+    assert uid_to_short_hostname("RINCON_347E5C0CF1E301400") == "Sonos-347E5C0CF1E3"
+
+
+async def test_uid_to_hostname():
+    """Test we can convert a hostname to a uid."""
+    assert hostname_to_uid("Sonos-347E5C0CF1E3.local.") == "RINCON_347E5C0CF1E301400"