Use SsdpServiceInfo for ssdp tests (part 1)

tests/components/harmony/test_config_flow.py

@@ -4,6 +4,7 @@
 import aiohttp
 
 from homeassistant import config_entries, data_entry_flow
+from homeassistant.components import ssdp
 from homeassistant.components.harmony.config_flow import CannotConnect
 from homeassistant.components.harmony.const import DOMAIN, PREVIOUS_ACTIVE_ACTIVITY
 from homeassistant.const import CONF_HOST, CONF_NAME
@@ -58,10 +59,14 @@ async def test_form_ssdp(hass):
         result = await hass.config_entries.flow.async_init(
             DOMAIN,
             context={"source": config_entries.SOURCE_SSDP},
-            data={
-                "friendlyName": "Harmony Hub",
-                "ssdp_location": "http://192.168.1.12:8088/description",
-            },
+            data=ssdp.SsdpServiceInfo(
+                ssdp_usn="mock_usn",
+                ssdp_st="mock_st",
+                ssdp_location="http://192.168.1.12:8088/description",
+                upnp={
+                    "friendlyName": "Harmony Hub",
+                },
+            ),
         )
     assert result["type"] == "form"
     assert result["step_id"] == "link"
@@ -106,10 +111,14 @@ async def test_form_ssdp_fails_to_get_remote_id(hass):
         result = await hass.config_entries.flow.async_init(
             DOMAIN,
             context={"source": config_entries.SOURCE_SSDP},
-            data={
-                "friendlyName": "Harmony Hub",
-                "ssdp_location": "http://192.168.1.12:8088/description",
-            },
+            data=ssdp.SsdpServiceInfo(
+                ssdp_usn="mock_usn",
+                ssdp_st="mock_st",
+                ssdp_location="http://192.168.1.12:8088/description",
+                upnp={
+                    "friendlyName": "Harmony Hub",
+                },
+            ),
         )
     assert result["type"] == "abort"
     assert result["reason"] == "cannot_connect"
@@ -139,10 +148,14 @@ async def test_form_ssdp_aborts_before_checking_remoteid_if_host_known(hass):
         result = await hass.config_entries.flow.async_init(
             DOMAIN,
             context={"source": config_entries.SOURCE_SSDP},
-            data={
-                "friendlyName": "Harmony Hub",
-                "ssdp_location": "http://2.2.2.2:8088/description",
-            },
+            data=ssdp.SsdpServiceInfo(
+                ssdp_usn="mock_usn",
+                ssdp_st="mock_st",
+                ssdp_location="http://2.2.2.2:8088/description",
+                upnp={
+                    "friendlyName": "Harmony Hub",
+                },
+            ),
         )
     assert result["type"] == "abort"
 

tests/components/heos/conftest.py

@@ -146,16 +146,20 @@ def dispatcher_fixture() -> Dispatcher:
 @pytest.fixture(name="discovery_data")
 def discovery_data_fixture() -> dict:
     """Return mock discovery data for testing."""
-    return {
-        ssdp.ATTR_SSDP_LOCATION: "http://127.0.0.1:60006/upnp/desc/aios_device/aios_device.xml",
-        ssdp.ATTR_UPNP_DEVICE_TYPE: "urn:schemas-denon-com:device:AiosDevice:1",
-        ssdp.ATTR_UPNP_FRIENDLY_NAME: "Office",
-        ssdp.ATTR_UPNP_MANUFACTURER: "Denon",
-        ssdp.ATTR_UPNP_MODEL_NAME: "HEOS Drive",
-        ssdp.ATTR_UPNP_MODEL_NUMBER: "DWSA-10 4.0",
-        ssdp.ATTR_UPNP_SERIAL: None,
-        ssdp.ATTR_UPNP_UDN: "uuid:e61de70c-2250-1c22-0080-0005cdf512be",
-    }
+    return ssdp.SsdpServiceInfo(
+        ssdp_usn="mock_usn",
+        ssdp_st="mock_st",
+        ssdp_location="http://127.0.0.1:60006/upnp/desc/aios_device/aios_device.xml",
+        upnp={
+            ssdp.ATTR_UPNP_DEVICE_TYPE: "urn:schemas-denon-com:device:AiosDevice:1",
+            ssdp.ATTR_UPNP_FRIENDLY_NAME: "Office",
+            ssdp.ATTR_UPNP_MANUFACTURER: "Denon",
+            ssdp.ATTR_UPNP_MODEL_NAME: "HEOS Drive",
+            ssdp.ATTR_UPNP_MODEL_NUMBER: "DWSA-10 4.0",
+            ssdp.ATTR_UPNP_SERIAL: None,
+            ssdp.ATTR_UPNP_UDN: "uuid:e61de70c-2250-1c22-0080-0005cdf512be",
+        },
+    )
 
 
 @pytest.fixture(name="quick_selects")

tests/components/heos/test_config_flow.py

@@ -79,7 +79,9 @@ async def test_create_entry_when_friendly_name_valid(hass, controller):
         assert DATA_DISCOVERED_HOSTS not in hass.data
 
 
-async def test_discovery_shows_create_form(hass, controller, discovery_data):
+async def test_discovery_shows_create_form(
+    hass, controller, discovery_data: ssdp.SsdpServiceInfo
+):
     """Test discovery shows form to confirm setup and subsequent abort."""
 
     await hass.config_entries.flow.async_init(
@@ -91,9 +93,9 @@ async def test_discovery_shows_create_form(hass, controller, discovery_data):
     assert len(flows_in_progress) == 1
     assert hass.data[DATA_DISCOVERED_HOSTS] == {"Office (127.0.0.1)": "127.0.0.1"}
 
-    port = urlparse(discovery_data[ssdp.ATTR_SSDP_LOCATION]).port
-    discovery_data[ssdp.ATTR_SSDP_LOCATION] = f"http://127.0.0.2:{port}/"
-    discovery_data[ssdp.ATTR_UPNP_FRIENDLY_NAME] = "Bedroom"
+    port = urlparse(discovery_data.ssdp_location).port
+    discovery_data.ssdp_location = f"http://127.0.0.2:{port}/"
+    discovery_data.upnp[ssdp.ATTR_UPNP_FRIENDLY_NAME] = "Bedroom"
 
     await hass.config_entries.flow.async_init(
         heos.DOMAIN, context={"source": SOURCE_SSDP}, data=discovery_data
@@ -109,7 +111,7 @@ async def test_discovery_shows_create_form(hass, controller, discovery_data):
 
 
 async def test_discovery_flow_aborts_already_setup(
-    hass, controller, discovery_data, config_entry
+    hass, controller, discovery_data: ssdp.SsdpServiceInfo, config_entry
 ):
     """Test discovery flow aborts when entry already setup."""
     config_entry.add_to_hass(hass)
@@ -120,12 +122,14 @@ async def test_discovery_flow_aborts_already_setup(
     assert result["reason"] == "single_instance_allowed"
 
 
-async def test_discovery_sets_the_unique_id(hass, controller, discovery_data):
+async def test_discovery_sets_the_unique_id(
+    hass, controller, discovery_data: ssdp.SsdpServiceInfo
+):
     """Test discovery sets the unique id."""
 
-    port = urlparse(discovery_data[ssdp.ATTR_SSDP_LOCATION]).port
-    discovery_data[ssdp.ATTR_SSDP_LOCATION] = f"http://127.0.0.2:{port}/"
-    discovery_data[ssdp.ATTR_UPNP_FRIENDLY_NAME] = "Bedroom"
+    port = urlparse(discovery_data.ssdp_location).port
+    discovery_data.ssdp_location = f"http://127.0.0.2:{port}/"
+    discovery_data.upnp[ssdp.ATTR_UPNP_FRIENDLY_NAME] = "Bedroom"
 
     await hass.config_entries.flow.async_init(
         heos.DOMAIN, context={"source": SOURCE_SSDP}, data=discovery_data

tests/components/isy994/test_config_flow.py

@@ -326,11 +326,15 @@ async def test_form_ssdp_already_configured(hass: HomeAssistant) -> None:
     result = await hass.config_entries.flow.async_init(
         DOMAIN,
         context={"source": SOURCE_SSDP},
-        data={
-            ssdp.ATTR_SSDP_LOCATION: f"http://{MOCK_HOSTNAME}{ISY_URL_POSTFIX}",
-            ssdp.ATTR_UPNP_FRIENDLY_NAME: "myisy",
-            ssdp.ATTR_UPNP_UDN: f"{UDN_UUID_PREFIX}{MOCK_UUID}",
-        },
+        data=ssdp.SsdpServiceInfo(
+            ssdp_usn="mock_usn",
+            ssdp_st="mock_st",
+            ssdp_location=f"http://{MOCK_HOSTNAME}{ISY_URL_POSTFIX}",
+            upnp={
+                ssdp.ATTR_UPNP_FRIENDLY_NAME: "myisy",
+                ssdp.ATTR_UPNP_UDN: f"{UDN_UUID_PREFIX}{MOCK_UUID}",
+            },
+        ),
     )
     assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
 
@@ -341,11 +345,15 @@ async def test_form_ssdp(hass: HomeAssistant):
     result = await hass.config_entries.flow.async_init(
         DOMAIN,
         context={"source": SOURCE_SSDP},
-        data={
-            ssdp.ATTR_SSDP_LOCATION: f"http://{MOCK_HOSTNAME}{ISY_URL_POSTFIX}",
-            ssdp.ATTR_UPNP_FRIENDLY_NAME: "myisy",
-            ssdp.ATTR_UPNP_UDN: f"{UDN_UUID_PREFIX}{MOCK_UUID}",
-        },
+        data=ssdp.SsdpServiceInfo(
+            ssdp_usn="mock_usn",
+            ssdp_st="mock_st",
+            ssdp_location=f"http://{MOCK_HOSTNAME}{ISY_URL_POSTFIX}",
+            upnp={
+                ssdp.ATTR_UPNP_FRIENDLY_NAME: "myisy",
+                ssdp.ATTR_UPNP_UDN: f"{UDN_UUID_PREFIX}{MOCK_UUID}",
+            },
+        ),
     )
     assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
     assert result["step_id"] == "user"
@@ -385,11 +393,15 @@ async def test_form_ssdp_existing_entry(hass: HomeAssistant):
         result = await hass.config_entries.flow.async_init(
             DOMAIN,
             context={"source": SOURCE_SSDP},
-            data={
-                ssdp.ATTR_SSDP_LOCATION: f"http://3.3.3.3{ISY_URL_POSTFIX}",
-                ssdp.ATTR_UPNP_FRIENDLY_NAME: "myisy",
-                ssdp.ATTR_UPNP_UDN: f"{UDN_UUID_PREFIX}{MOCK_UUID}",
-            },
+            data=ssdp.SsdpServiceInfo(
+                ssdp_usn="mock_usn",
+                ssdp_st="mock_st",
+                ssdp_location=f"http://3.3.3.3{ISY_URL_POSTFIX}",
+                upnp={
+                    ssdp.ATTR_UPNP_FRIENDLY_NAME: "myisy",
+                    ssdp.ATTR_UPNP_UDN: f"{UDN_UUID_PREFIX}{MOCK_UUID}",
+                },
+            ),
         )
         await hass.async_block_till_done()
 
@@ -412,11 +424,15 @@ async def test_form_ssdp_existing_entry_with_no_port(hass: HomeAssistant):
         result = await hass.config_entries.flow.async_init(
             DOMAIN,
             context={"source": SOURCE_SSDP},
-            data={
-                ssdp.ATTR_SSDP_LOCATION: f"http://3.3.3.3/{ISY_URL_POSTFIX}",
-                ssdp.ATTR_UPNP_FRIENDLY_NAME: "myisy",
-                ssdp.ATTR_UPNP_UDN: f"{UDN_UUID_PREFIX}{MOCK_UUID}",
-            },
+            data=ssdp.SsdpServiceInfo(
+                ssdp_usn="mock_usn",
+                ssdp_st="mock_st",
+                ssdp_location=f"http://3.3.3.3/{ISY_URL_POSTFIX}",
+                upnp={
+                    ssdp.ATTR_UPNP_FRIENDLY_NAME: "myisy",
+                    ssdp.ATTR_UPNP_UDN: f"{UDN_UUID_PREFIX}{MOCK_UUID}",
+                },
+            ),
         )
         await hass.async_block_till_done()
 
@@ -439,11 +455,15 @@ async def test_form_ssdp_existing_entry_with_alternate_port(hass: HomeAssistant)
         result = await hass.config_entries.flow.async_init(
             DOMAIN,
             context={"source": SOURCE_SSDP},
-            data={
-                ssdp.ATTR_SSDP_LOCATION: f"http://3.3.3.3:1443/{ISY_URL_POSTFIX}",
-                ssdp.ATTR_UPNP_FRIENDLY_NAME: "myisy",
-                ssdp.ATTR_UPNP_UDN: f"{UDN_UUID_PREFIX}{MOCK_UUID}",
-            },
+            data=ssdp.SsdpServiceInfo(
+                ssdp_usn="mock_usn",
+                ssdp_st="mock_st",
+                ssdp_location=f"http://3.3.3.3:1443/{ISY_URL_POSTFIX}",
+                upnp={
+                    ssdp.ATTR_UPNP_FRIENDLY_NAME: "myisy",
+                    ssdp.ATTR_UPNP_UDN: f"{UDN_UUID_PREFIX}{MOCK_UUID}",
+                },
+            ),
         )
         await hass.async_block_till_done()
 
@@ -466,11 +486,15 @@ async def test_form_ssdp_existing_entry_no_port_https(hass: HomeAssistant):
         result = await hass.config_entries.flow.async_init(
             DOMAIN,
             context={"source": SOURCE_SSDP},
-            data={
-                ssdp.ATTR_SSDP_LOCATION: f"https://3.3.3.3/{ISY_URL_POSTFIX}",
-                ssdp.ATTR_UPNP_FRIENDLY_NAME: "myisy",
-                ssdp.ATTR_UPNP_UDN: f"{UDN_UUID_PREFIX}{MOCK_UUID}",
-            },
+            data=ssdp.SsdpServiceInfo(
+                ssdp_usn="mock_usn",
+                ssdp_st="mock_st",
+                ssdp_location=f"https://3.3.3.3/{ISY_URL_POSTFIX}",
+                upnp={
+                    ssdp.ATTR_UPNP_FRIENDLY_NAME: "myisy",
+                    ssdp.ATTR_UPNP_UDN: f"{UDN_UUID_PREFIX}{MOCK_UUID}",
+                },
+            ),
         )
         await hass.async_block_till_done()
 

tests/components/roku/__init__.py

@@ -3,13 +3,9 @@
 import re
 from socket import gaierror as SocketGIAError
 
-from homeassistant.components import zeroconf
+from homeassistant.components import ssdp, zeroconf
 from homeassistant.components.roku.const import DOMAIN
-from homeassistant.components.ssdp import (
-    ATTR_SSDP_LOCATION,
-    ATTR_UPNP_FRIENDLY_NAME,
-    ATTR_UPNP_SERIAL,
-)
+from homeassistant.components.ssdp import ATTR_UPNP_FRIENDLY_NAME, ATTR_UPNP_SERIAL
 from homeassistant.const import CONF_HOST
 from homeassistant.core import HomeAssistant
 
@@ -24,11 +20,15 @@
 UPNP_FRIENDLY_NAME = "My Roku 3"
 UPNP_SERIAL = "1GU48T017973"
 
-MOCK_SSDP_DISCOVERY_INFO = {
-    ATTR_SSDP_LOCATION: SSDP_LOCATION,
-    ATTR_UPNP_FRIENDLY_NAME: UPNP_FRIENDLY_NAME,
-    ATTR_UPNP_SERIAL: UPNP_SERIAL,
-}
+MOCK_SSDP_DISCOVERY_INFO = ssdp.SsdpServiceInfo(
+    ssdp_usn="mock_usn",
+    ssdp_st="mock_st",
+    ssdp_location=SSDP_LOCATION,
+    upnp={
+        ATTR_UPNP_FRIENDLY_NAME: UPNP_FRIENDLY_NAME,
+        ATTR_UPNP_SERIAL: UPNP_SERIAL,
+    },
+)
 
 HOMEKIT_HOST = "192.168.1.161"
 

tests/components/roku/test_config_flow.py

@@ -48,7 +48,7 @@ async def test_duplicate_error(
     assert result["type"] == RESULT_TYPE_ABORT
     assert result["reason"] == "already_configured"
 
-    discovery_info = MOCK_SSDP_DISCOVERY_INFO.copy()
+    discovery_info = dataclasses.replace(MOCK_SSDP_DISCOVERY_INFO)
     result = await hass.config_entries.flow.async_init(
         DOMAIN, context={CONF_SOURCE: SOURCE_SSDP}, data=discovery_info
     )
@@ -216,7 +216,7 @@ async def test_ssdp_cannot_connect(
     """Test we abort SSDP flow on connection error."""
     mock_connection(aioclient_mock, error=True)
 
-    discovery_info = MOCK_SSDP_DISCOVERY_INFO.copy()
+    discovery_info = dataclasses.replace(MOCK_SSDP_DISCOVERY_INFO)
     result = await hass.config_entries.flow.async_init(
         DOMAIN,
         context={CONF_SOURCE: SOURCE_SSDP},
@@ -231,7 +231,7 @@ async def test_ssdp_unknown_error(
     hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
 ) -> None:
     """Test we abort SSDP flow on unknown error."""
-    discovery_info = MOCK_SSDP_DISCOVERY_INFO.copy()
+    discovery_info = dataclasses.replace(MOCK_SSDP_DISCOVERY_INFO)
     with patch(
         "homeassistant.components.roku.config_flow.Roku.update",
         side_effect=Exception,
@@ -252,7 +252,7 @@ async def test_ssdp_discovery(
     """Test the SSDP discovery flow."""
     mock_connection(aioclient_mock)
 
-    discovery_info = MOCK_SSDP_DISCOVERY_INFO.copy()
+    discovery_info = dataclasses.replace(MOCK_SSDP_DISCOVERY_INFO)
     result = await hass.config_entries.flow.async_init(
         DOMAIN, context={CONF_SOURCE: SOURCE_SSDP}, data=discovery_info
     )

tests/components/songpal/test_config_flow.py

@@ -1,5 +1,6 @@
 """Test the songpal config flow."""
 import copy
+import dataclasses
 from unittest.mock import patch
 
 from homeassistant.components import ssdp
@@ -26,17 +27,21 @@
 
 UDN = "uuid:1234"
 
-SSDP_DATA = {
-    ssdp.ATTR_UPNP_UDN: UDN,
-    ssdp.ATTR_UPNP_FRIENDLY_NAME: FRIENDLY_NAME,
-    ssdp.ATTR_SSDP_LOCATION: f"http://{HOST}:52323/dmr.xml",
-    "X_ScalarWebAPI_DeviceInfo": {
-        "X_ScalarWebAPI_BaseURL": ENDPOINT,
-        "X_ScalarWebAPI_ServiceList": {
-            "X_ScalarWebAPI_ServiceType": ["guide", "system", "audio", "avContent"],
+SSDP_DATA = ssdp.SsdpServiceInfo(
+    ssdp_usn="mock_usn",
+    ssdp_st="mock_st",
+    ssdp_location=f"http://{HOST}:52323/dmr.xml",
+    upnp={
+        ssdp.ATTR_UPNP_UDN: UDN,
+        ssdp.ATTR_UPNP_FRIENDLY_NAME: FRIENDLY_NAME,
+        "X_ScalarWebAPI_DeviceInfo": {
+            "X_ScalarWebAPI_BaseURL": ENDPOINT,
+            "X_ScalarWebAPI_ServiceList": {
+                "X_ScalarWebAPI_ServiceType": ["guide", "system", "audio", "avContent"],
+            },
         },
     },
-}
+)
 
 
 def _flow_next(hass, flow_id):
@@ -150,8 +155,9 @@ def _create_mock_config_entry(hass):
 
 async def test_ssdp_bravia(hass):
     """Test discovering a bravia TV."""
-    ssdp_data = copy.deepcopy(SSDP_DATA)
-    ssdp_data["X_ScalarWebAPI_DeviceInfo"]["X_ScalarWebAPI_ServiceList"][
+    ssdp_data = dataclasses.replace(SSDP_DATA)
+    ssdp_data.upnp = copy.deepcopy(ssdp_data.upnp)
+    ssdp_data.upnp["X_ScalarWebAPI_DeviceInfo"]["X_ScalarWebAPI_ServiceList"][
         "X_ScalarWebAPI_ServiceType"
     ].append("videoScreen")
     result = await hass.config_entries.flow.async_init(