Allow discovery to update invalid unique_id in SamsungTV (#67859)

Co-authored-by: epenet <epenet@users.noreply.github.com>
pull/60465/head^2
epenet 2022-03-08 22:27:53 +01:00 committed by GitHub
parent 8adcd10f55
commit 9ab56bd5c8
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 90 additions and 11 deletions

View File

@ -215,15 +215,23 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return self.async_show_form(step_id="user", data_schema=DATA_SCHEMA)
@callback
def _async_get_existing_matching_entry(self) -> config_entries.ConfigEntry | None:
"""Get first existing matching entry."""
def _async_get_existing_matching_entry(
self,
) -> tuple[config_entries.ConfigEntry | None, bool]:
"""Get first existing matching entry (prefer unique id)."""
matching_host_entry: config_entries.ConfigEntry | None = None
for entry in self._async_current_entries(include_ignore=False):
mac = entry.data.get(CONF_MAC)
mac_match = mac and self._mac and mac == self._mac
upnp_udn_match = self._upnp_udn and self._upnp_udn == entry.unique_id
if entry.data[CONF_HOST] == self._host or mac_match or upnp_udn_match:
return entry
return None
if (self._mac and self._mac == entry.data.get(CONF_MAC)) or (
self._upnp_udn and self._upnp_udn == entry.unique_id
):
LOGGER.debug("Found entry matching unique_id for %s", self._host)
return entry, True
if entry.data[CONF_HOST] == self._host:
LOGGER.debug("Found entry matching host for %s", self._host)
matching_host_entry = entry
return matching_host_entry, False
@callback
def _async_update_existing_matching_entry(
@ -233,15 +241,18 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
Returns the existing entry if it was updated.
"""
if entry := self._async_get_existing_matching_entry():
entry, is_unique_match = self._async_get_existing_matching_entry()
if entry:
entry_kw_args: dict = {}
if (self._udn and self._upnp_udn and self._upnp_udn != self._udn) or (
self.unique_id and entry.unique_id is None
if self.unique_id and (
entry.unique_id is None
or (is_unique_match and self.unique_id != entry.unique_id)
):
entry_kw_args["unique_id"] = self.unique_id
if self._mac and not entry.data.get(CONF_MAC):
entry_kw_args["data"] = {**entry.data, CONF_MAC: self._mac}
if entry_kw_args:
LOGGER.debug("Updating existing config entry with %s", entry_kw_args)
self.hass.config_entries.async_update_entry(entry, **entry_kw_args)
self.hass.async_create_task(
self.hass.config_entries.async_reload(entry.entry_id)

View File

@ -1403,3 +1403,71 @@ async def test_update_incorrect_udn_matching_mac_unique_id_added_from_ssdp(
assert result["reason"] == "already_configured"
assert entry.data[CONF_MAC] == "aa:bb:ww:ii:ff:ii"
assert entry.unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4"
@pytest.mark.usefixtures("remotews")
async def test_update_incorrect_udn_matching_mac_from_dhcp(
hass: HomeAssistant,
) -> None:
"""Test that DHCP updates the wrong udn from ssdp via mac match."""
entry = MockConfigEntry(
domain=DOMAIN,
data={**MOCK_WS_ENTRY, CONF_MAC: "aa:bb:ww:ii:ff:ii"},
source=config_entries.SOURCE_SSDP,
unique_id="0d1cef00-00dc-1000-9c80-4844f7b172de",
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.samsungtv.async_setup",
return_value=True,
) as mock_setup, patch(
"homeassistant.components.samsungtv.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_DHCP},
data=MOCK_DHCP_DATA,
)
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
assert result["type"] == "abort"
assert result["reason"] == "already_configured"
assert entry.data[CONF_MAC] == "aa:bb:ww:ii:ff:ii"
assert entry.unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4"
@pytest.mark.usefixtures("remotews")
async def test_no_update_incorrect_udn_not_matching_mac_from_dhcp(
hass: HomeAssistant,
) -> None:
"""Test that DHCP does not update the wrong udn from ssdp via host match."""
entry = MockConfigEntry(
domain=DOMAIN,
data={**MOCK_WS_ENTRY, CONF_MAC: "aa:bb:ss:ss:dd:pp"},
source=config_entries.SOURCE_SSDP,
unique_id="0d1cef00-00dc-1000-9c80-4844f7b172de",
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.samsungtv.async_setup",
return_value=True,
) as mock_setup, patch(
"homeassistant.components.samsungtv.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_DHCP},
data=MOCK_DHCP_DATA,
)
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 0
assert len(mock_setup_entry.mock_calls) == 0
assert result["type"] == "form"
assert result["step_id"] == "confirm"
assert entry.data[CONF_MAC] == "aa:bb:ss:ss:dd:pp"
assert entry.unique_id == "0d1cef00-00dc-1000-9c80-4844f7b172de"