Use dataclass properties in dlna_dmr discovery (#60693)

Co-authored-by: epenet <epenet@users.noreply.github.com>

homeassistant/components/dlna_dmr/config_flow.py

@@ -216,7 +216,7 @@ class DlnaDmrFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
         # Abort if the device doesn't support all services required for a DmrDevice.
         # Use the discovery_info instead of DmrDevice.is_profile_device to avoid
         # contacting the device again.
-        discovery_service_list = discovery_info.get(ssdp.ATTR_UPNP_SERVICE_LIST)
+        discovery_service_list = discovery_info.upnp.get(ssdp.ATTR_UPNP_SERVICE_LIST)
         if not discovery_service_list:
             return self.async_abort(reason="not_dmr")
         discovery_service_ids = {
@@ -334,15 +334,15 @@ class DlnaDmrFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
         """Set information required for a config entry from the SSDP discovery."""
         LOGGER.debug(
             "_async_set_info_from_discovery: location: %s, UDN: %s",
-            discovery_info[ssdp.ATTR_SSDP_LOCATION],
-            discovery_info[ssdp.ATTR_SSDP_UDN],
+            discovery_info.ssdp_location,
+            discovery_info.ssdp_udn,
         )
 
         if not self._location:
-            self._location = discovery_info[ssdp.ATTR_SSDP_LOCATION]
+            self._location = discovery_info.ssdp_location
             assert isinstance(self._location, str)
 
-        self._udn = discovery_info[ssdp.ATTR_SSDP_UDN]
+        self._udn = discovery_info.ssdp_udn
         await self.async_set_unique_id(self._udn)
 
         if abort_if_configured:
@@ -351,11 +351,9 @@ class DlnaDmrFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
                 updates={CONF_URL: self._location}, reload_on_update=False
             )
 
-        self._device_type = (
-            discovery_info.get(ssdp.ATTR_SSDP_NT) or discovery_info[ssdp.ATTR_SSDP_ST]
-        )
+        self._device_type = discovery_info.ssdp_nt or discovery_info.ssdp_st
         self._name = (
-            discovery_info.get(ssdp.ATTR_UPNP_FRIENDLY_NAME)
+            discovery_info.upnp.get(ssdp.ATTR_UPNP_FRIENDLY_NAME)
             or urlparse(self._location).hostname
             or DEFAULT_NAME
         )
@@ -461,22 +459,25 @@ def _is_ignored_device(discovery_info: ssdp.SsdpServiceInfo) -> bool:
     flow, which will list all discovered but unconfigured devices.
     """
     # Did the discovery trigger more than just this flow?
-    if len(discovery_info.get(ssdp.ATTR_HA_MATCHING_DOMAINS, set())) > 1:
+    if len(discovery_info.x_homeassistant_matching_domains) > 1:
         LOGGER.debug(
             "Ignoring device supported by multiple integrations: %s",
-            discovery_info[ssdp.ATTR_HA_MATCHING_DOMAINS],
+            discovery_info.x_homeassistant_matching_domains,
         )
         return True
 
     # Is the root device not a DMR?
-    if discovery_info.get(ssdp.ATTR_UPNP_DEVICE_TYPE) not in DmrDevice.DEVICE_TYPES:
+    if (
+        discovery_info.upnp.get(ssdp.ATTR_UPNP_DEVICE_TYPE)
+        not in DmrDevice.DEVICE_TYPES
+    ):
         return True
 
     # Special cases for devices with other discovery methods (e.g. mDNS), or
     # that advertise multiple unrelated (sent in separate discovery packets)
     # UPnP devices.
-    manufacturer = discovery_info.get(ssdp.ATTR_UPNP_MANUFACTURER, "").lower()
-    model = discovery_info.get(ssdp.ATTR_UPNP_MODEL_NAME, "").lower()
+    manufacturer = discovery_info.upnp.get(ssdp.ATTR_UPNP_MANUFACTURER, "").lower()
+    model = discovery_info.upnp.get(ssdp.ATTR_UPNP_MODEL_NAME, "").lower()
 
     if manufacturer.startswith("xbmc") or model == "kodi":
         # kodi

homeassistant/components/dlna_dmr/media_player.py

@@ -247,12 +247,12 @@ class DlnaDmrEntity(MediaPlayerEntity):
         _LOGGER.debug(
             "SSDP %s notification of device %s at %s",
             change,
-            info[ssdp.ATTR_SSDP_USN],
-            info.get(ssdp.ATTR_SSDP_LOCATION),
+            info.ssdp_usn,
+            info.ssdp_location,
         )
 
         try:
-            bootid_str = info[ssdp.ATTR_SSDP_BOOTID]
+            bootid_str = info.ssdp_headers[ssdp.ATTR_SSDP_BOOTID]
             bootid: int | None = int(bootid_str, 10)
         except (KeyError, ValueError):
             bootid = None
@@ -263,7 +263,7 @@ class DlnaDmrEntity(MediaPlayerEntity):
                 # Store the new value (because our old value matches) so that we
                 # can ignore subsequent ssdp:alive messages
                 with contextlib.suppress(KeyError, ValueError):
-                    next_bootid_str = info[ssdp.ATTR_SSDP_NEXTBOOTID]
+                    next_bootid_str = info.ssdp_headers[ssdp.ATTR_SSDP_NEXTBOOTID]
                     self._bootid = int(next_bootid_str, 10)
             # Nothing left to do until ssdp:alive comes through
             return
@@ -278,7 +278,7 @@ class DlnaDmrEntity(MediaPlayerEntity):
             await self._device_disconnect()
 
         if change == ssdp.SsdpChange.ALIVE and not self._device:
-            location = info[ssdp.ATTR_SSDP_LOCATION]
+            location = info.ssdp_location or ""
             try:
                 await self._device_connect(location)
             except UpnpError as err:

tests/components/dlna_dmr/test_media_player.py

@@ -51,6 +51,8 @@ from .conftest import (
 
 from tests.common import MockConfigEntry
 
+MOCK_DEVICE_ST = "mock_st"
+
 # Auto-use the domain_data_mock fixture for every test in this module
 pytestmark = pytest.mark.usefixtures("domain_data_mock")
 
@@ -1052,10 +1054,12 @@ async def test_become_available(
     # Send an SSDP notification from the now alive device
     ssdp_callback = ssdp_scanner_mock.async_register_callback.call_args.args[0]
     await ssdp_callback(
-        {
-            ssdp.ATTR_SSDP_USN: MOCK_DEVICE_USN,
-            ssdp.ATTR_SSDP_LOCATION: NEW_DEVICE_LOCATION,
-        },
+        ssdp.SsdpServiceInfo(
+            ssdp_usn=MOCK_DEVICE_USN,
+            ssdp_location=NEW_DEVICE_LOCATION,
+            ssdp_st=MOCK_DEVICE_ST,
+            upnp={},
+        ),
         ssdp.SsdpChange.ALIVE,
     )
     await hass.async_block_till_done()
@@ -1114,10 +1118,12 @@ async def test_alive_but_gone(
     # Send an SSDP notification from the still missing device
     ssdp_callback = ssdp_scanner_mock.async_register_callback.call_args.args[0]
     await ssdp_callback(
-        {
-            ssdp.ATTR_SSDP_USN: MOCK_DEVICE_USN,
-            ssdp.ATTR_SSDP_LOCATION: NEW_DEVICE_LOCATION,
-        },
+        ssdp.SsdpServiceInfo(
+            ssdp_usn=MOCK_DEVICE_USN,
+            ssdp_location=NEW_DEVICE_LOCATION,
+            ssdp_st=MOCK_DEVICE_ST,
+            upnp={},
+        ),
         ssdp.SsdpChange.ALIVE,
     )
     await hass.async_block_till_done()
@@ -1153,17 +1159,21 @@ async def test_multiple_ssdp_alive(
     # Send two SSDP notifications with the new device URL
     ssdp_callback = ssdp_scanner_mock.async_register_callback.call_args.args[0]
     await ssdp_callback(
-        {
-            ssdp.ATTR_SSDP_USN: MOCK_DEVICE_USN,
-            ssdp.ATTR_SSDP_LOCATION: NEW_DEVICE_LOCATION,
-        },
+        ssdp.SsdpServiceInfo(
+            ssdp_usn=MOCK_DEVICE_USN,
+            ssdp_location=NEW_DEVICE_LOCATION,
+            ssdp_st=MOCK_DEVICE_ST,
+            upnp={},
+        ),
         ssdp.SsdpChange.ALIVE,
     )
     await ssdp_callback(
-        {
-            ssdp.ATTR_SSDP_USN: MOCK_DEVICE_USN,
-            ssdp.ATTR_SSDP_LOCATION: NEW_DEVICE_LOCATION,
-        },
+        ssdp.SsdpServiceInfo(
+            ssdp_usn=MOCK_DEVICE_USN,
+            ssdp_location=NEW_DEVICE_LOCATION,
+            ssdp_st=MOCK_DEVICE_ST,
+            upnp={},
+        ),
         ssdp.SsdpChange.ALIVE,
     )
     await hass.async_block_till_done()
@@ -1189,11 +1199,13 @@ async def test_ssdp_byebye(
     # First byebye will cause a disconnect
     ssdp_callback = ssdp_scanner_mock.async_register_callback.call_args.args[0]
     await ssdp_callback(
-        {
-            ssdp.ATTR_SSDP_USN: MOCK_DEVICE_USN,
-            "_udn": MOCK_DEVICE_UDN,
-            "NTS": "ssdp:byebye",
-        },
+        ssdp.SsdpServiceInfo(
+            ssdp_usn=MOCK_DEVICE_USN,
+            ssdp_udn=MOCK_DEVICE_UDN,
+            ssdp_headers={"NTS": "ssdp:byebye"},
+            ssdp_st=MOCK_DEVICE_ST,
+            upnp={},
+        ),
         ssdp.SsdpChange.BYEBYE,
     )
 
@@ -1206,11 +1218,13 @@ async def test_ssdp_byebye(
 
     # Second byebye will do nothing
     await ssdp_callback(
-        {
-            ssdp.ATTR_SSDP_USN: MOCK_DEVICE_USN,
-            "_udn": MOCK_DEVICE_UDN,
-            "NTS": "ssdp:byebye",
-        },
+        ssdp.SsdpServiceInfo(
+            ssdp_usn=MOCK_DEVICE_USN,
+            ssdp_udn=MOCK_DEVICE_UDN,
+            ssdp_headers={"NTS": "ssdp:byebye"},
+            ssdp_st=MOCK_DEVICE_ST,
+            upnp={},
+        ),
         ssdp.SsdpChange.BYEBYE,
     )
 
@@ -1237,24 +1251,30 @@ async def test_ssdp_update_seen_bootid(
     # Send SSDP alive with boot ID
     ssdp_callback = ssdp_scanner_mock.async_register_callback.call_args.args[0]
     await ssdp_callback(
-        {
-            ssdp.ATTR_SSDP_USN: MOCK_DEVICE_USN,
-            ssdp.ATTR_SSDP_LOCATION: MOCK_DEVICE_LOCATION,
-            ssdp.ATTR_SSDP_BOOTID: "1",
-        },
+        ssdp.SsdpServiceInfo(
+            ssdp_usn=MOCK_DEVICE_USN,
+            ssdp_location=MOCK_DEVICE_LOCATION,
+            ssdp_headers={ssdp.ATTR_SSDP_BOOTID: "1"},
+            ssdp_st=MOCK_DEVICE_ST,
+            upnp={},
+        ),
         ssdp.SsdpChange.ALIVE,
     )
     await hass.async_block_till_done()
 
     # Send SSDP update with next boot ID
     await ssdp_callback(
-        {
-            ssdp.ATTR_SSDP_USN: MOCK_DEVICE_USN,
-            "_udn": MOCK_DEVICE_UDN,
-            "NTS": "ssdp:update",
-            ssdp.ATTR_SSDP_BOOTID: "1",
-            ssdp.ATTR_SSDP_NEXTBOOTID: "2",
-        },
+        ssdp.SsdpServiceInfo(
+            ssdp_usn=MOCK_DEVICE_USN,
+            ssdp_udn=MOCK_DEVICE_UDN,
+            ssdp_headers={
+                "NTS": "ssdp:update",
+                ssdp.ATTR_SSDP_BOOTID: "1",
+                ssdp.ATTR_SSDP_NEXTBOOTID: "2",
+            },
+            ssdp_st=MOCK_DEVICE_ST,
+            upnp={},
+        ),
         ssdp.SsdpChange.UPDATE,
     )
     await hass.async_block_till_done()
@@ -1269,13 +1289,17 @@ async def test_ssdp_update_seen_bootid(
 
     # Send SSDP update with same next boot ID, again
     await ssdp_callback(
-        {
-            ssdp.ATTR_SSDP_USN: MOCK_DEVICE_USN,
-            "_udn": MOCK_DEVICE_UDN,
-            "NTS": "ssdp:update",
-            ssdp.ATTR_SSDP_BOOTID: "1",
-            ssdp.ATTR_SSDP_NEXTBOOTID: "2",
-        },
+        ssdp.SsdpServiceInfo(
+            ssdp_usn=MOCK_DEVICE_USN,
+            ssdp_udn=MOCK_DEVICE_UDN,
+            ssdp_headers={
+                "NTS": "ssdp:update",
+                ssdp.ATTR_SSDP_BOOTID: "1",
+                ssdp.ATTR_SSDP_NEXTBOOTID: "2",
+            },
+            ssdp_st=MOCK_DEVICE_ST,
+            upnp={},
+        ),
         ssdp.SsdpChange.UPDATE,
     )
     await hass.async_block_till_done()
@@ -1290,13 +1314,17 @@ async def test_ssdp_update_seen_bootid(
 
     # Send SSDP update with bad next boot ID
     await ssdp_callback(
-        {
-            ssdp.ATTR_SSDP_USN: MOCK_DEVICE_USN,
-            "_udn": MOCK_DEVICE_UDN,
-            "NTS": "ssdp:update",
-            ssdp.ATTR_SSDP_BOOTID: "2",
-            ssdp.ATTR_SSDP_NEXTBOOTID: "7c848375-a106-4bd1-ac3c-8e50427c8e4f",
-        },
+        ssdp.SsdpServiceInfo(
+            ssdp_usn=MOCK_DEVICE_USN,
+            ssdp_udn=MOCK_DEVICE_UDN,
+            ssdp_headers={
+                "NTS": "ssdp:update",
+                ssdp.ATTR_SSDP_BOOTID: "2",
+                ssdp.ATTR_SSDP_NEXTBOOTID: "7c848375-a106-4bd1-ac3c-8e50427c8e4f",
+            },
+            ssdp_st=MOCK_DEVICE_ST,
+            upnp={},
+        ),
         ssdp.SsdpChange.UPDATE,
     )
     await hass.async_block_till_done()
@@ -1311,11 +1339,13 @@ async def test_ssdp_update_seen_bootid(
 
     # Send a new SSDP alive with the new boot ID, device should not reconnect
     await ssdp_callback(
-        {
-            ssdp.ATTR_SSDP_USN: MOCK_DEVICE_USN,
-            ssdp.ATTR_SSDP_LOCATION: MOCK_DEVICE_LOCATION,
-            ssdp.ATTR_SSDP_BOOTID: "2",
-        },
+        ssdp.SsdpServiceInfo(
+            ssdp_usn=MOCK_DEVICE_USN,
+            ssdp_location=MOCK_DEVICE_LOCATION,
+            ssdp_headers={ssdp.ATTR_SSDP_BOOTID: "2"},
+            ssdp_st=MOCK_DEVICE_ST,
+            upnp={},
+        ),
         ssdp.SsdpChange.ALIVE,
     )
     await hass.async_block_till_done()
@@ -1348,24 +1378,30 @@ async def test_ssdp_update_missed_bootid(
     # Send SSDP alive with boot ID
     ssdp_callback = ssdp_scanner_mock.async_register_callback.call_args.args[0]
     await ssdp_callback(
-        {
-            ssdp.ATTR_SSDP_USN: MOCK_DEVICE_USN,
-            ssdp.ATTR_SSDP_LOCATION: MOCK_DEVICE_LOCATION,
-            ssdp.ATTR_SSDP_BOOTID: "1",
-        },
+        ssdp.SsdpServiceInfo(
+            ssdp_usn=MOCK_DEVICE_USN,
+            ssdp_location=MOCK_DEVICE_LOCATION,
+            ssdp_headers={ssdp.ATTR_SSDP_BOOTID: "1"},
+            ssdp_st=MOCK_DEVICE_ST,
+            upnp={},
+        ),
         ssdp.SsdpChange.ALIVE,
     )
     await hass.async_block_till_done()
 
     # Send SSDP update with skipped boot ID (not previously seen)
     await ssdp_callback(
-        {
-            ssdp.ATTR_SSDP_USN: MOCK_DEVICE_USN,
-            "_udn": MOCK_DEVICE_UDN,
-            "NTS": "ssdp:update",
-            ssdp.ATTR_SSDP_BOOTID: "2",
-            ssdp.ATTR_SSDP_NEXTBOOTID: "3",
-        },
+        ssdp.SsdpServiceInfo(
+            ssdp_usn=MOCK_DEVICE_USN,
+            ssdp_udn=MOCK_DEVICE_UDN,
+            ssdp_headers={
+                "NTS": "ssdp:update",
+                ssdp.ATTR_SSDP_BOOTID: "2",
+                ssdp.ATTR_SSDP_NEXTBOOTID: "3",
+            },
+            ssdp_st=MOCK_DEVICE_ST,
+            upnp={},
+        ),
         ssdp.SsdpChange.UPDATE,
     )
     await hass.async_block_till_done()
@@ -1380,11 +1416,13 @@ async def test_ssdp_update_missed_bootid(
 
     # Send a new SSDP alive with the new boot ID, device should reconnect
     await ssdp_callback(
-        {
-            ssdp.ATTR_SSDP_USN: MOCK_DEVICE_USN,
-            ssdp.ATTR_SSDP_LOCATION: MOCK_DEVICE_LOCATION,
-            ssdp.ATTR_SSDP_BOOTID: "3",
-        },
+        ssdp.SsdpServiceInfo(
+            ssdp_usn=MOCK_DEVICE_USN,
+            ssdp_location=MOCK_DEVICE_LOCATION,
+            ssdp_headers={ssdp.ATTR_SSDP_BOOTID: "3"},
+            ssdp_st=MOCK_DEVICE_ST,
+            upnp={},
+        ),
         ssdp.SsdpChange.ALIVE,
     )
     await hass.async_block_till_done()
@@ -1417,11 +1455,13 @@ async def test_ssdp_bootid(
     # Send SSDP alive with boot ID
     ssdp_callback = ssdp_scanner_mock.async_register_callback.call_args.args[0]
     await ssdp_callback(
-        {
-            ssdp.ATTR_SSDP_USN: MOCK_DEVICE_USN,
-            ssdp.ATTR_SSDP_LOCATION: MOCK_DEVICE_LOCATION,
-            ssdp.ATTR_SSDP_BOOTID: "1",
-        },
+        ssdp.SsdpServiceInfo(
+            ssdp_usn=MOCK_DEVICE_USN,
+            ssdp_location=MOCK_DEVICE_LOCATION,
+            ssdp_headers={ssdp.ATTR_SSDP_BOOTID: "1"},
+            ssdp_st=MOCK_DEVICE_ST,
+            upnp={},
+        ),
         ssdp.SsdpChange.ALIVE,
     )
     await hass.async_block_till_done()
@@ -1435,11 +1475,13 @@ async def test_ssdp_bootid(
 
     # Send SSDP alive with same boot ID, nothing should happen
     await ssdp_callback(
-        {
-            ssdp.ATTR_SSDP_USN: MOCK_DEVICE_USN,
-            ssdp.ATTR_SSDP_LOCATION: MOCK_DEVICE_LOCATION,
-            ssdp.ATTR_SSDP_BOOTID: "1",
-        },
+        ssdp.SsdpServiceInfo(
+            ssdp_usn=MOCK_DEVICE_USN,
+            ssdp_location=MOCK_DEVICE_LOCATION,
+            ssdp_headers={ssdp.ATTR_SSDP_BOOTID: "1"},
+            ssdp_st=MOCK_DEVICE_ST,
+            upnp={},
+        ),
         ssdp.SsdpChange.ALIVE,
     )
     await hass.async_block_till_done()
@@ -1453,11 +1495,13 @@ async def test_ssdp_bootid(
 
     # Send a new SSDP alive with an incremented boot ID, device should be dis/reconnected
     await ssdp_callback(
-        {
-            ssdp.ATTR_SSDP_USN: MOCK_DEVICE_USN,
-            ssdp.ATTR_SSDP_LOCATION: MOCK_DEVICE_LOCATION,
-            ssdp.ATTR_SSDP_BOOTID: "2",
-        },
+        ssdp.SsdpServiceInfo(
+            ssdp_usn=MOCK_DEVICE_USN,
+            ssdp_location=MOCK_DEVICE_LOCATION,
+            ssdp_headers={ssdp.ATTR_SSDP_BOOTID: "2"},
+            ssdp_st=MOCK_DEVICE_ST,
+            upnp={},
+        ),
         ssdp.SsdpChange.ALIVE,
     )
     await hass.async_block_till_done()