evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Use identifiers host and serial number to match device (#75657)

Browse source
This commit is contained in:
Steven Looman 2022-09-06 17:33:16 +02:00 • committed by GitHub
parent b2f86ddf76
commit d550b17bd9
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 75 additions and 18 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

29
homeassistant/components/upnp/__init__.py
View file

@ -25,15 +25,18 @@ from homeassistant.helpers.update_coordinator import (
)
from .const import (
CONFIG_ENTRY_HOST,
CONFIG_ENTRY_MAC_ADDRESS,
CONFIG_ENTRY_ORIGINAL_UDN,
CONFIG_ENTRY_ST,
CONFIG_ENTRY_UDN,
DEFAULT_SCAN_INTERVAL,
DOMAIN,
IDENTIFIER_HOST,
IDENTIFIER_SERIAL_NUMBER,
LOGGER,
)
from .device import Device, async_create_device, async_get_mac_address_from_host
from .device import Device, async_create_device
NOTIFICATION_ID = "upnp_notification"
NOTIFICATION_TITLE = "UPnP/IGD Setup"
@ -106,24 +109,30 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
device.original_udn = entry.data[CONFIG_ENTRY_ORIGINAL_UDN]
# Store mac address for changed UDN matching.
if device.host:
device.mac_address = await async_get_mac_address_from_host(hass, device.host)
if device.mac_address and not entry.data.get("CONFIG_ENTRY_MAC_ADDRESS"):
device_mac_address = await device.async_get_mac_address()
if device_mac_address and not entry.data.get(CONFIG_ENTRY_MAC_ADDRESS):
hass.config_entries.async_update_entry(
entry=entry,
data={
**entry.data,
CONFIG_ENTRY_MAC_ADDRESS: device.mac_address,
CONFIG_ENTRY_MAC_ADDRESS: device_mac_address,
CONFIG_ENTRY_HOST: device.host,
},
)
identifiers = {(DOMAIN, device.usn)}
if device.host:
identifiers.add((IDENTIFIER_HOST, device.host))
if device.serial_number:
identifiers.add((IDENTIFIER_SERIAL_NUMBER, device.serial_number))
connections = {(dr.CONNECTION_UPNP, device.udn)}
if device.mac_address:
connections.add((dr.CONNECTION_NETWORK_MAC, device.mac_address))
if device_mac_address:
connections.add((dr.CONNECTION_NETWORK_MAC, device_mac_address))
device_registry = dr.async_get(hass)
device_entry = device_registry.async_get_device(
identifiers=set(), connections=connections
identifiers=identifiers, connections=connections
)
if device_entry:
LOGGER.debug(
@ -136,7 +145,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
device_entry = device_registry.async_get_or_create(
config_entry_id=entry.entry_id,
connections=connections,
identifiers={(DOMAIN, device.usn)},
identifiers=identifiers,
name=device.name,
manufacturer=device.manufacturer,
model=device.model_name,
@ -148,7 +157,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
# Update identifier.
device_entry = device_registry.async_update_device(
device_entry.id,
new_identifiers={(DOMAIN, device.usn)},
new_identifiers=identifiers,
)
assert device_entry

8
homeassistant/components/upnp/config_flow.py
View file

@ -13,6 +13,7 @@ from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResult
from .const import (
CONFIG_ENTRY_HOST,
CONFIG_ENTRY_LOCATION,
CONFIG_ENTRY_MAC_ADDRESS,
CONFIG_ENTRY_ORIGINAL_UDN,
@ -161,22 +162,25 @@ class UpnpFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
unique_id = discovery_info.ssdp_usn
await self.async_set_unique_id(unique_id)
mac_address = await _async_mac_address_from_discovery(self.hass, discovery_info)
host = discovery_info.ssdp_headers["_host"]
self._abort_if_unique_id_configured(
# Store mac address for older entries.
# The location is stored in the config entry such that when the location changes, the entry is reloaded.
updates={
CONFIG_ENTRY_MAC_ADDRESS: mac_address,
CONFIG_ENTRY_LOCATION: discovery_info.ssdp_location,
CONFIG_ENTRY_HOST: host,
},
)
# Handle devices changing their UDN, only allow a single host.
for entry in self._async_current_entries(include_ignore=True):
entry_mac_address = entry.data.get(CONFIG_ENTRY_MAC_ADDRESS)
entry_st = entry.data.get(CONFIG_ENTRY_ST)
if entry_mac_address != mac_address:
entry_host = entry.data.get(CONFIG_ENTRY_HOST)
if entry_mac_address != mac_address and entry_host != host:
continue
entry_st = entry.data.get(CONFIG_ENTRY_ST)
if discovery_info.ssdp_st != entry_st:
# Check ssdp_st to prevent swapping between IGDv1 and IGDv2.
continue

5
homeassistant/components/upnp/const.py
View file

@ -6,7 +6,6 @@ from homeassistant.const import TIME_SECONDS
LOGGER = logging.getLogger(__package__)
CONF_LOCAL_IP = "local_ip"
DOMAIN = "upnp"
BYTES_RECEIVED = "bytes_received"
BYTES_SENT = "bytes_sent"
@ -24,7 +23,9 @@ CONFIG_ENTRY_UDN = "udn"
CONFIG_ENTRY_ORIGINAL_UDN = "original_udn"
CONFIG_ENTRY_MAC_ADDRESS = "mac_address"
CONFIG_ENTRY_LOCATION = "location"
CONFIG_ENTRY_HOST = "host"
IDENTIFIER_HOST = "upnp_host"
IDENTIFIER_SERIAL_NUMBER = "upnp_serial_number"
DEFAULT_SCAN_INTERVAL = timedelta(seconds=30).total_seconds()
ST_IGD_V1 = "urn:schemas-upnp-org:device:InternetGatewayDevice:1"
ST_IGD_V2 = "urn:schemas-upnp-org:device:InternetGatewayDevice:2"
SSDP_SEARCH_TIMEOUT = 4

8
homeassistant/components/upnp/device.py
View file

@ -69,9 +69,15 @@ class Device:
self.hass = hass
self._igd_device = igd_device
self.coordinator: DataUpdateCoordinator | None = None
self.mac_address: str | None = None
self.original_udn: str | None = None
async def async_get_mac_address(self) -> str | None:
"""Get mac address."""
if not self.host:
return None
return await async_get_mac_address_from_host(self.hass, self.host)
@property
def udn(self) -> str:
"""Get the UDN."""

7
tests/components/upnp/conftest.py
View file

@ -25,7 +25,7 @@ TEST_UDN = "uuid:device"
TEST_ST = "urn:schemas-upnp-org:device:InternetGatewayDevice:1"
TEST_USN = f"{TEST_UDN}::{TEST_ST}"
TEST_LOCATION = "http://192.168.1.1/desc.xml"
TEST_HOSTNAME = urlparse(TEST_LOCATION).hostname
TEST_HOST = urlparse(TEST_LOCATION).hostname
TEST_FRIENDLY_NAME = "mock-name"
TEST_MAC_ADDRESS = "00:11:22:33:44:55"
TEST_DISCOVERY = ssdp.SsdpServiceInfo(
@ -41,10 +41,11 @@ TEST_DISCOVERY = ssdp.SsdpServiceInfo(
ssdp.ATTR_UPNP_FRIENDLY_NAME: TEST_FRIENDLY_NAME,
ssdp.ATTR_UPNP_MANUFACTURER: "mock-manufacturer",
ssdp.ATTR_UPNP_MODEL_NAME: "mock-model-name",
ssdp.ATTR_UPNP_SERIAL: "mock-serial",
ssdp.ATTR_UPNP_UDN: TEST_UDN,
},
ssdp_headers={
"_host": TEST_HOSTNAME,
"_host": TEST_HOST,
},
)
@ -54,8 +55,10 @@ def mock_igd_device() -> IgdDevice:
"""Mock async_upnp_client device."""
mock_upnp_device = create_autospec(UpnpDevice, instance=True)
mock_upnp_device.device_url = TEST_DISCOVERY.ssdp_location
mock_upnp_device.serial_number = TEST_DISCOVERY.upnp[ssdp.ATTR_UPNP_SERIAL]
mock_igd_device = create_autospec(IgdDevice)
mock_igd_device.device_type = TEST_DISCOVERY.ssdp_st
mock_igd_device.name = TEST_DISCOVERY.upnp[ssdp.ATTR_UPNP_FRIENDLY_NAME]
mock_igd_device.manufacturer = TEST_DISCOVERY.upnp[ssdp.ATTR_UPNP_MANUFACTURER]
mock_igd_device.model_name = TEST_DISCOVERY.upnp[ssdp.ATTR_UPNP_MODEL_NAME]

36
tests/components/upnp/test_config_flow.py
View file

@ -8,6 +8,7 @@ import pytest
from homeassistant import config_entries, data_entry_flow
from homeassistant.components import ssdp
from homeassistant.components.upnp.const import (
CONFIG_ENTRY_HOST,
CONFIG_ENTRY_LOCATION,
CONFIG_ENTRY_MAC_ADDRESS,
CONFIG_ENTRY_ORIGINAL_UDN,
@ -21,6 +22,7 @@ from homeassistant.core import HomeAssistant
from .conftest import (
TEST_DISCOVERY,
TEST_FRIENDLY_NAME,
TEST_HOST,
TEST_LOCATION,
TEST_MAC_ADDRESS,
TEST_ST,
@ -140,7 +142,7 @@ async def test_flow_ssdp_no_mac_address(hass: HomeAssistant):
@pytest.mark.usefixtures("mock_mac_address_from_host")
async def test_flow_ssdp_discovery_changed_udn(hass: HomeAssistant):
async def test_flow_ssdp_discovery_changed_udn_match_mac(hass: HomeAssistant):
"""Test config flow: discovery through ssdp, same device, but new UDN, matched on mac address."""
entry = MockConfigEntry(
domain=DOMAIN,
@ -171,6 +173,38 @@ async def test_flow_ssdp_discovery_changed_udn(hass: HomeAssistant):
assert result["reason"] == "config_entry_updated"
@pytest.mark.usefixtures("mock_mac_address_from_host")
async def test_flow_ssdp_discovery_changed_udn_match_host(hass: HomeAssistant):
"""Test config flow: discovery through ssdp, same device, but new UDN, matched on mac address."""
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_HOST: TEST_HOST,
},
source=config_entries.SOURCE_SSDP,
state=config_entries.ConfigEntryState.LOADED,
)
entry.add_to_hass(hass)
# New discovery via step ssdp.
new_udn = TEST_UDN + "2"
new_discovery = deepcopy(TEST_DISCOVERY)
new_discovery.ssdp_usn = f"{new_udn}::{TEST_ST}"
new_discovery.upnp["_udn"] = new_udn
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data=new_discovery,
)
assert result["type"] == data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "config_entry_updated"
@pytest.mark.usefixtures(
"ssdp_instant_discovery",
"mock_setup_entry",