Fix config_flow error for UPnP info with single service (#69979)
parent
f8db38c0b6
commit
741252a32d
|
@ -77,10 +77,16 @@ class DlnaDmsFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
|||
discovery_service_list = discovery_info.upnp.get(ssdp.ATTR_UPNP_SERVICE_LIST)
|
||||
if not discovery_service_list:
|
||||
return self.async_abort(reason="not_dms")
|
||||
discovery_service_ids = {
|
||||
service.get("serviceId")
|
||||
for service in discovery_service_list.get("service") or []
|
||||
}
|
||||
|
||||
services = discovery_service_list.get("service")
|
||||
if not services:
|
||||
discovery_service_ids: set[str] = set()
|
||||
elif isinstance(services, list):
|
||||
discovery_service_ids = {service.get("serviceId") for service in services}
|
||||
else:
|
||||
# Only one service defined (etree_to_dict failed to make a list)
|
||||
discovery_service_ids = {services.get("serviceId")}
|
||||
|
||||
if not DmsDevice.SERVICE_IDS.issubset(discovery_service_ids):
|
||||
return self.async_abort(reason="not_dms")
|
||||
|
||||
|
|
|
@ -325,7 +325,7 @@ async def test_ssdp_flow_upnp_udn(
|
|||
|
||||
async def test_ssdp_missing_services(hass: HomeAssistant) -> None:
|
||||
"""Test SSDP ignores devices that are missing required services."""
|
||||
# No services defined at all
|
||||
# No service list at all
|
||||
discovery = dataclasses.replace(MOCK_DISCOVERY)
|
||||
discovery.upnp = dict(discovery.upnp)
|
||||
del discovery.upnp[ssdp.ATTR_UPNP_SERVICE_LIST]
|
||||
|
@ -337,6 +337,18 @@ async def test_ssdp_missing_services(hass: HomeAssistant) -> None:
|
|||
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
|
||||
assert result["reason"] == "not_dms"
|
||||
|
||||
# Service list does not contain services
|
||||
discovery = dataclasses.replace(MOCK_DISCOVERY)
|
||||
discovery.upnp = dict(discovery.upnp)
|
||||
discovery.upnp[ssdp.ATTR_UPNP_SERVICE_LIST] = {"bad_key": "bad_value"}
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": config_entries.SOURCE_SSDP},
|
||||
data=discovery,
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
|
||||
assert result["reason"] == "not_dms"
|
||||
|
||||
# ContentDirectory service is missing
|
||||
discovery = dataclasses.replace(MOCK_DISCOVERY)
|
||||
discovery.upnp = dict(discovery.upnp)
|
||||
|
@ -352,3 +364,25 @@ async def test_ssdp_missing_services(hass: HomeAssistant) -> None:
|
|||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
|
||||
assert result["reason"] == "not_dms"
|
||||
|
||||
|
||||
async def test_ssdp_single_service(hass: HomeAssistant) -> None:
|
||||
"""Test SSDP discovery info with only one service defined.
|
||||
|
||||
THe etree_to_dict function turns multiple services into a list of dicts, but
|
||||
a single service into only a dict.
|
||||
"""
|
||||
discovery = dataclasses.replace(MOCK_DISCOVERY)
|
||||
discovery.upnp = dict(discovery.upnp)
|
||||
service_list = dict(discovery.upnp[ssdp.ATTR_UPNP_SERVICE_LIST])
|
||||
# Turn mock's list of service dicts into a single dict
|
||||
service_list["service"] = service_list["service"][0]
|
||||
discovery.upnp[ssdp.ATTR_UPNP_SERVICE_LIST] = service_list
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": config_entries.SOURCE_SSDP},
|
||||
data=discovery,
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
|
||||
assert result["reason"] == "not_dms"
|
||||
|
|
Loading…
Reference in New Issue