Improve discovering upnp/igd device by always using the SSDP-discovery for the Unique Device Name (#111487)

* Always use the UDN found in the SSDP discovery, instead of the device description

* Ensure existing DeviceEntries are still matched
pull/113164/head
Steven Looman 2024-03-12 17:38:09 +01:00 committed by GitHub
parent 3b1ab6436d
commit cd4e8707ea
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 103 additions and 12 deletions

View File

@ -80,6 +80,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
# Create device.
assert discovery_info is not None
assert discovery_info.ssdp_udn
assert discovery_info.ssdp_all_locations
location = get_preferred_location(discovery_info.ssdp_all_locations)
try:
@ -118,7 +119,9 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
if device.serial_number:
identifiers.add((IDENTIFIER_SERIAL_NUMBER, device.serial_number))
connections = {(dr.CONNECTION_UPNP, device.udn)}
connections = {(dr.CONNECTION_UPNP, discovery_info.ssdp_udn)}
if discovery_info.ssdp_udn != device.udn:
connections.add((dr.CONNECTION_UPNP, device.udn))
if device_mac_address:
connections.add((dr.CONNECTION_NETWORK_MAC, device_mac_address))

View File

@ -42,7 +42,7 @@ def _friendly_name_from_discovery(discovery_info: ssdp.SsdpServiceInfo) -> str:
def _is_complete_discovery(discovery_info: ssdp.SsdpServiceInfo) -> bool:
"""Test if discovery is complete and usable."""
return bool(
ssdp.ATTR_UPNP_UDN in discovery_info.upnp
discovery_info.ssdp_udn
and discovery_info.ssdp_st
and discovery_info.ssdp_all_locations
and discovery_info.ssdp_usn
@ -80,9 +80,8 @@ class UpnpFlowHandler(ConfigFlow, domain=DOMAIN):
VERSION = 1
# Paths:
# - ssdp(discovery_info) --> ssdp_confirm(None)
# --> ssdp_confirm({}) --> create_entry()
# - user(None): scan --> user({...}) --> create_entry()
# 1: ssdp(discovery_info) --> ssdp_confirm(None) --> ssdp_confirm({}) --> create_entry()
# 2: user(None): scan --> user({...}) --> create_entry()
@property
def _discoveries(self) -> dict[str, SsdpServiceInfo]:
@ -243,9 +242,9 @@ class UpnpFlowHandler(ConfigFlow, domain=DOMAIN):
discovery = self._remove_discovery(usn)
mac_address = await _async_mac_address_from_discovery(self.hass, discovery)
data = {
CONFIG_ENTRY_UDN: discovery.upnp[ssdp.ATTR_UPNP_UDN],
CONFIG_ENTRY_UDN: discovery.ssdp_udn,
CONFIG_ENTRY_ST: discovery.ssdp_st,
CONFIG_ENTRY_ORIGINAL_UDN: discovery.upnp[ssdp.ATTR_UPNP_UDN],
CONFIG_ENTRY_ORIGINAL_UDN: discovery.ssdp_udn,
CONFIG_ENTRY_MAC_ADDRESS: mac_address,
CONFIG_ENTRY_HOST: discovery.ssdp_headers["_host"],
CONFIG_ENTRY_LOCATION: get_preferred_location(discovery.ssdp_all_locations),
@ -267,9 +266,9 @@ class UpnpFlowHandler(ConfigFlow, domain=DOMAIN):
title = _friendly_name_from_discovery(discovery)
mac_address = await _async_mac_address_from_discovery(self.hass, discovery)
data = {
CONFIG_ENTRY_UDN: discovery.upnp[ssdp.ATTR_UPNP_UDN],
CONFIG_ENTRY_UDN: discovery.ssdp_udn,
CONFIG_ENTRY_ST: discovery.ssdp_st,
CONFIG_ENTRY_ORIGINAL_UDN: discovery.upnp[ssdp.ATTR_UPNP_UDN],
CONFIG_ENTRY_ORIGINAL_UDN: discovery.ssdp_udn,
CONFIG_ENTRY_LOCATION: get_preferred_location(discovery.ssdp_all_locations),
CONFIG_ENTRY_MAC_ADDRESS: mac_address,
CONFIG_ENTRY_HOST: discovery.ssdp_headers["_host"],

View File

@ -1,5 +1,6 @@
"""Test UPnP/IGD config flow."""
import copy
from copy import deepcopy
from unittest.mock import patch
@ -111,6 +112,7 @@ async def test_flow_ssdp_incomplete_discovery(hass: HomeAssistant) -> None:
context={"source": config_entries.SOURCE_SSDP},
data=ssdp.SsdpServiceInfo(
ssdp_usn=TEST_USN,
# ssdp_udn=TEST_UDN, # Not provided.
ssdp_st=TEST_ST,
ssdp_location=TEST_LOCATION,
upnp={
@ -132,12 +134,12 @@ async def test_flow_ssdp_non_igd_device(hass: HomeAssistant) -> None:
context={"source": config_entries.SOURCE_SSDP},
data=ssdp.SsdpServiceInfo(
ssdp_usn=TEST_USN,
ssdp_udn=TEST_UDN,
ssdp_st=TEST_ST,
ssdp_location=TEST_LOCATION,
ssdp_all_locations=[TEST_LOCATION],
upnp={
ssdp.ATTR_UPNP_DEVICE_TYPE: "urn:schemas-upnp-org:device:WFADevice:1", # Non-IGD
ssdp.ATTR_UPNP_UDN: TEST_UDN,
},
),
)
@ -442,3 +444,40 @@ async def test_flow_user_no_discovery(hass: HomeAssistant) -> None:
)
assert result["type"] == data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "no_devices_found"
@pytest.mark.usefixtures(
"ssdp_instant_discovery",
"mock_setup_entry",
"mock_get_source_ip",
"mock_mac_address_from_host",
)
async def test_flow_ssdp_with_mismatched_udn(hass: HomeAssistant) -> None:
"""Test config flow: discovered + configured through ssdp, where the UDN differs in the SSDP-discovery vs device description."""
# Discovered via step ssdp.
test_discovery = copy.deepcopy(TEST_DISCOVERY)
test_discovery.upnp[ssdp.ATTR_UPNP_UDN] = "uuid:another_udn"
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data=test_discovery,
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "ssdp_confirm"
# Confirm via step ssdp_confirm.
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={},
)
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert result["title"] == TEST_FRIENDLY_NAME
assert result["data"] == {
CONFIG_ENTRY_ST: TEST_ST,
CONFIG_ENTRY_UDN: TEST_UDN,
CONFIG_ENTRY_ORIGINAL_UDN: TEST_UDN,
CONFIG_ENTRY_LOCATION: TEST_LOCATION,
CONFIG_ENTRY_MAC_ADDRESS: TEST_MAC_ADDRESS,
CONFIG_ENTRY_HOST: TEST_HOST,
}

View File

@ -2,10 +2,12 @@
from __future__ import annotations
from unittest.mock import AsyncMock
import copy
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from homeassistant.components import ssdp
from homeassistant.components.upnp.const import (
CONFIG_ENTRY_LOCATION,
CONFIG_ENTRY_MAC_ADDRESS,
@ -16,7 +18,14 @@ from homeassistant.components.upnp.const import (
)
from homeassistant.core import HomeAssistant
from .conftest import TEST_LOCATION, TEST_MAC_ADDRESS, TEST_ST, TEST_UDN, TEST_USN
from .conftest import (
TEST_DISCOVERY,
TEST_LOCATION,
TEST_MAC_ADDRESS,
TEST_ST,
TEST_UDN,
TEST_USN,
)
from tests.common import MockConfigEntry
@ -95,3 +104,44 @@ async def test_async_setup_entry_multi_location(
# Ensure that the IPv4 location is used.
mock_async_create_device.assert_called_once_with(TEST_LOCATION)
@pytest.mark.usefixtures("mock_get_source_ip", "mock_mac_address_from_host")
async def test_async_setup_udn_mismatch(
hass: HomeAssistant, mock_async_create_device: AsyncMock
) -> None:
"""Test async_setup_entry for a device which reports a different UDN from SSDP-discovery and device description."""
test_discovery = copy.deepcopy(TEST_DISCOVERY)
test_discovery.upnp[ssdp.ATTR_UPNP_UDN] = "uuid:another_udn"
entry = MockConfigEntry(
domain=DOMAIN,
unique_id=TEST_USN,
data={
CONFIG_ENTRY_ST: TEST_ST,
CONFIG_ENTRY_UDN: TEST_UDN,
CONFIG_ENTRY_ORIGINAL_UDN: TEST_UDN,
CONFIG_ENTRY_LOCATION: TEST_LOCATION,
CONFIG_ENTRY_MAC_ADDRESS: TEST_MAC_ADDRESS,
},
)
# Set up device discovery callback.
async def register_callback(hass, callback, match_dict):
"""Immediately do callback."""
await callback(test_discovery, ssdp.SsdpChange.ALIVE)
return MagicMock()
with patch(
"homeassistant.components.ssdp.async_register_callback",
side_effect=register_callback,
), patch(
"homeassistant.components.ssdp.async_get_discovery_info_by_st",
return_value=[test_discovery],
):
# Load config_entry.
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id) is True
# Ensure that the IPv4 location is used.
mock_async_create_device.assert_called_once_with(TEST_LOCATION)