Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Update tplink device config during reauth flow (#122089)

pull/119852/head^2
Steven B. 2024-07-17 20:07:53 +01:00 committed by GitHub
parent fa0a5451b9
commit 55cee89392
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 219 additions and 99 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

51
homeassistant/components/tplink/config_flow.py
View File

@ -3,6 +3,7 @@
from __future__ import annotations
from collections.abc import Mapping
import logging
from typing import Any
from kasa import (
@ -52,6 +53,8 @@ from .const import (
DOMAIN,
)
_LOGGER = logging.getLogger(__name__)
STEP_AUTH_DATA_SCHEMA = vol.Schema(
{vol.Required(CONF_USERNAME): str, vol.Required(CONF_PASSWORD): str}
)
@ -88,15 +91,10 @@ class TPLinkConfigFlow(ConfigFlow, domain=DOMAIN):
)
@callback
def _update_config_if_entry_in_setup_error(
def _get_config_updates(
self, entry: ConfigEntry, host: str, config: dict
) -> ConfigFlowResult | None:
"""If discovery encounters a device that is in SETUP_ERROR or SETUP_RETRY update the device config."""
if entry.state not in (
ConfigEntryState.SETUP_ERROR,
ConfigEntryState.SETUP_RETRY,
):
return None
) -> dict | None:
"""Return updates if the host or device config has changed."""
entry_data = entry.data
entry_config_dict = entry_data.get(CONF_DEVICE_CONFIG)
if entry_config_dict == config and entry_data[CONF_HOST] == host:
@ -110,11 +108,31 @@ class TPLinkConfigFlow(ConfigFlow, domain=DOMAIN):
!= config.get(CONF_CONNECTION_TYPE)
):
updates.pop(CONF_CREDENTIALS_HASH, None)
return self.async_update_reload_and_abort(
entry,
data=updates,
reason="already_configured",
)
_LOGGER.debug(
"Connection type changed for %s from %s to: %s",
host,
entry_config_dict.get(CONF_CONNECTION_TYPE),
config.get(CONF_CONNECTION_TYPE),
)
return updates
@callback
def _update_config_if_entry_in_setup_error(
self, entry: ConfigEntry, host: str, config: dict
) -> ConfigFlowResult | None:
"""If discovery encounters a device that is in SETUP_ERROR or SETUP_RETRY update the device config."""
if entry.state not in (
ConfigEntryState.SETUP_ERROR,
ConfigEntryState.SETUP_RETRY,
):
return None
if updates := self._get_config_updates(entry, host, config):
return self.async_update_reload_and_abort(
entry,
data=updates,
reason="already_configured",
)
return None
async def _async_handle_discovery(
self, host: str, formatted_mac: str, config: dict | None = None
@ -454,7 +472,7 @@ class TPLinkConfigFlow(ConfigFlow, domain=DOMAIN):
password = user_input[CONF_PASSWORD]
credentials = Credentials(username, password)
try:
await self._async_try_discover_and_update(
device = await self._async_try_discover_and_update(
host,
credentials=credentials,
raise_on_progress=True,
@ -467,6 +485,11 @@ class TPLinkConfigFlow(ConfigFlow, domain=DOMAIN):
placeholders["error"] = str(ex)
else:
await set_credentials(self.hass, username, password)
config = device.config.to_dict(exclude_credentials=True)
if updates := self._get_config_updates(reauth_entry, host, config):
self.hass.config_entries.async_update_entry(
reauth_entry, data=updates
)
self.hass.async_create_task(
self._async_reload_requires_auth_entries(), eager_start=False
)

46
tests/components/tplink/__init__.py
View File

@ -57,25 +57,26 @@ CREDENTIALS_HASH_LEGACY = ""
DEVICE_CONFIG_LEGACY = DeviceConfig(IP_ADDRESS)
DEVICE_CONFIG_DICT_LEGACY = DEVICE_CONFIG_LEGACY.to_dict(exclude_credentials=True)
CREDENTIALS = Credentials("foo", "bar")
CREDENTIALS_HASH_AUTH = "abcdefghijklmnopqrstuv=="
DEVICE_CONFIG_AUTH = DeviceConfig(
CREDENTIALS_HASH_AES = "AES/abcdefghijklmnopqrstuvabcdefghijklmnopqrstuv=="
CREDENTIALS_HASH_KLAP = "KLAP/abcdefghijklmnopqrstuv=="
DEVICE_CONFIG_KLAP = DeviceConfig(
IP_ADDRESS,
credentials=CREDENTIALS,
connection_type=DeviceConnectionParameters(
DeviceFamily.IotSmartPlugSwitch, DeviceEncryptionType.Klap
DeviceFamily.SmartTapoPlug, DeviceEncryptionType.Klap
),
uses_http=True,
)
DEVICE_CONFIG_AUTH2 = DeviceConfig(
DEVICE_CONFIG_AES = DeviceConfig(
IP_ADDRESS2,
credentials=CREDENTIALS,
connection_type=DeviceConnectionParameters(
DeviceFamily.IotSmartPlugSwitch, DeviceEncryptionType.Klap
DeviceFamily.SmartTapoPlug, DeviceEncryptionType.Aes
),
uses_http=True,
)
DEVICE_CONFIG_DICT_AUTH = DEVICE_CONFIG_AUTH.to_dict(exclude_credentials=True)
DEVICE_CONFIG_DICT_AUTH2 = DEVICE_CONFIG_AUTH2.to_dict(exclude_credentials=True)
DEVICE_CONFIG_DICT_KLAP = DEVICE_CONFIG_KLAP.to_dict(exclude_credentials=True)
DEVICE_CONFIG_DICT_AES = DEVICE_CONFIG_AES.to_dict(exclude_credentials=True)
CREATE_ENTRY_DATA_LEGACY = {
CONF_HOST: IP_ADDRESS,
@ -84,24 +85,28 @@ CREATE_ENTRY_DATA_LEGACY = {
CONF_DEVICE_CONFIG: DEVICE_CONFIG_DICT_LEGACY,
}
CREATE_ENTRY_DATA_AUTH = {
CREATE_ENTRY_DATA_KLAP = {
CONF_HOST: IP_ADDRESS,
CONF_ALIAS: ALIAS,
CONF_MODEL: MODEL,
CONF_CREDENTIALS_HASH: CREDENTIALS_HASH_AUTH,
CONF_DEVICE_CONFIG: DEVICE_CONFIG_DICT_AUTH,
CONF_CREDENTIALS_HASH: CREDENTIALS_HASH_KLAP,
CONF_DEVICE_CONFIG: DEVICE_CONFIG_DICT_KLAP,
}
CREATE_ENTRY_DATA_AUTH2 = {
CREATE_ENTRY_DATA_AES = {
CONF_HOST: IP_ADDRESS2,
CONF_ALIAS: ALIAS,
CONF_MODEL: MODEL,
CONF_CREDENTIALS_HASH: CREDENTIALS_HASH_AUTH,
CONF_DEVICE_CONFIG: DEVICE_CONFIG_DICT_AUTH2,
CONF_CREDENTIALS_HASH: CREDENTIALS_HASH_AES,
CONF_DEVICE_CONFIG: DEVICE_CONFIG_DICT_AES,
}
NEW_CONNECTION_TYPE = DeviceConnectionParameters(
DeviceFamily.IotSmartPlugSwitch, DeviceEncryptionType.Aes
CONNECTION_TYPE_KLAP = DeviceConnectionParameters(
DeviceFamily.SmartTapoPlug, DeviceEncryptionType.Klap
)
NEW_CONNECTION_TYPE_DICT = NEW_CONNECTION_TYPE.to_dict()
CONNECTION_TYPE_KLAP_DICT = CONNECTION_TYPE_KLAP.to_dict()
CONNECTION_TYPE_AES = DeviceConnectionParameters(
DeviceFamily.SmartTapoPlug, DeviceEncryptionType.Aes
)
CONNECTION_TYPE_AES_DICT = CONNECTION_TYPE_AES.to_dict()
def _load_feature_fixtures():
@ -187,7 +192,7 @@ def _mocked_device(
device_id=DEVICE_ID,
alias=ALIAS,
model=MODEL,
ip_address=IP_ADDRESS,
ip_address: str | None = None,
modules: list[str] | None = None,
children: list[Device] | None = None,
features: list[str | Feature] | None = None,
@ -202,12 +207,17 @@ def _mocked_device(
device.mac = mac
device.alias = alias
device.model = model
device.host = ip_address
device.device_id = device_id
device.hw_info = {"sw_ver": "1.0.0", "hw_ver": "1.0.0"}
device.modules = {}
device.features = {}
if not ip_address:
ip_address = IP_ADDRESS
else:
device_config.host = ip_address
device.host = ip_address
if modules:
device.modules = {
module_name: MODULE_TO_MOCK_GEN[module_name](device)

23
tests/components/tplink/conftest.py
View File

@ -11,8 +11,10 @@ from homeassistant.core import HomeAssistant
from . import (
CREATE_ENTRY_DATA_LEGACY,
CREDENTIALS_HASH_AUTH,
DEVICE_CONFIG_AUTH,
CREDENTIALS_HASH_AES,
CREDENTIALS_HASH_KLAP,
DEVICE_CONFIG_AES,
DEVICE_CONFIG_KLAP,
IP_ADDRESS,
IP_ADDRESS2,
MAC_ADDRESS,
@ -32,14 +34,14 @@ def mock_discovery():
discover_single=DEFAULT,
) as mock_discovery:
device = _mocked_device(
device_config=copy.deepcopy(DEVICE_CONFIG_AUTH),
credentials_hash=CREDENTIALS_HASH_AUTH,
device_config=copy.deepcopy(DEVICE_CONFIG_KLAP),
credentials_hash=CREDENTIALS_HASH_KLAP,
alias=None,
)
devices = {
"127.0.0.1": _mocked_device(
device_config=copy.deepcopy(DEVICE_CONFIG_AUTH),
credentials_hash=CREDENTIALS_HASH_AUTH,
device_config=copy.deepcopy(DEVICE_CONFIG_KLAP),
credentials_hash=CREDENTIALS_HASH_KLAP,
alias=None,
)
}
@ -55,12 +57,15 @@ def mock_connect():
with patch("homeassistant.components.tplink.Device.connect") as mock_connect:
devices = {
IP_ADDRESS: _mocked_device(
device_config=DEVICE_CONFIG_AUTH, credentials_hash=CREDENTIALS_HASH_AUTH
device_config=DEVICE_CONFIG_KLAP,
credentials_hash=CREDENTIALS_HASH_KLAP,
ip_address=IP_ADDRESS,
),
IP_ADDRESS2: _mocked_device(
device_config=DEVICE_CONFIG_AUTH,
credentials_hash=CREDENTIALS_HASH_AUTH,
device_config=DEVICE_CONFIG_AES,
credentials_hash=CREDENTIALS_HASH_AES,
mac=MAC_ADDRESS2,
ip_address=IP_ADDRESS2,
),
}

162
tests/components/tplink/test_config_flow.py
View File

@ -1,5 +1,6 @@
"""Test the tplink config flow."""
import logging
from unittest.mock import AsyncMock, patch
from kasa import TimeoutError
@ -11,6 +12,7 @@ from homeassistant.components.tplink import (
DOMAIN,
AuthenticationError,
Credentials,
Device,
DeviceConfig,
KasaException,
)
@ -33,19 +35,21 @@ from homeassistant.data_entry_flow import FlowResultType
from . import (
ALIAS,
CREATE_ENTRY_DATA_AUTH,
CREATE_ENTRY_DATA_AUTH2,
CONNECTION_TYPE_KLAP_DICT,
CREATE_ENTRY_DATA_AES,
CREATE_ENTRY_DATA_KLAP,
CREATE_ENTRY_DATA_LEGACY,
CREDENTIALS_HASH_AUTH,
CREDENTIALS_HASH_AES,
CREDENTIALS_HASH_KLAP,
DEFAULT_ENTRY_TITLE,
DEVICE_CONFIG_DICT_AUTH,
DEVICE_CONFIG_DICT_AES,
DEVICE_CONFIG_DICT_KLAP,
DEVICE_CONFIG_DICT_LEGACY,
DHCP_FORMATTED_MAC_ADDRESS,
IP_ADDRESS,
MAC_ADDRESS,
MAC_ADDRESS2,
MODULE,
NEW_CONNECTION_TYPE_DICT,
_mocked_device,
_patch_connect,
_patch_discovery,
@ -135,7 +139,7 @@ async def test_discovery_auth(
CONF_HOST: IP_ADDRESS,
CONF_MAC: MAC_ADDRESS,
CONF_ALIAS: ALIAS,
CONF_DEVICE_CONFIG: DEVICE_CONFIG_DICT_AUTH,
CONF_DEVICE_CONFIG: DEVICE_CONFIG_DICT_KLAP,
},
)
await hass.async_block_till_done()
@ -154,7 +158,7 @@ async def test_discovery_auth(
assert result2["type"] is FlowResultType.CREATE_ENTRY
assert result2["title"] == DEFAULT_ENTRY_TITLE
assert result2["data"] == CREATE_ENTRY_DATA_AUTH
assert result2["data"] == CREATE_ENTRY_DATA_KLAP
assert result2["context"]["unique_id"] == MAC_ADDRESS
@ -187,7 +191,7 @@ async def test_discovery_auth_errors(
CONF_HOST: IP_ADDRESS,
CONF_MAC: MAC_ADDRESS,
CONF_ALIAS: ALIAS,
CONF_DEVICE_CONFIG: DEVICE_CONFIG_DICT_AUTH,
CONF_DEVICE_CONFIG: DEVICE_CONFIG_DICT_KLAP,
},
)
await hass.async_block_till_done()
@ -218,7 +222,7 @@ async def test_discovery_auth_errors(
},
)
assert result3["type"] is FlowResultType.CREATE_ENTRY
assert result3["data"] == CREATE_ENTRY_DATA_AUTH
assert result3["data"] == CREATE_ENTRY_DATA_KLAP
assert result3["context"]["unique_id"] == MAC_ADDRESS
@ -238,7 +242,7 @@ async def test_discovery_new_credentials(
CONF_HOST: IP_ADDRESS,
CONF_MAC: MAC_ADDRESS,
CONF_ALIAS: ALIAS,
CONF_DEVICE_CONFIG: DEVICE_CONFIG_DICT_AUTH,
CONF_DEVICE_CONFIG: DEVICE_CONFIG_DICT_KLAP,
},
)
await hass.async_block_till_done()
@ -267,7 +271,7 @@ async def test_discovery_new_credentials(
{},
)
assert result3["type"] is FlowResultType.CREATE_ENTRY
assert result3["data"] == CREATE_ENTRY_DATA_AUTH
assert result3["data"] == CREATE_ENTRY_DATA_KLAP
assert result3["context"]["unique_id"] == MAC_ADDRESS
@ -290,7 +294,7 @@ async def test_discovery_new_credentials_invalid(
CONF_HOST: IP_ADDRESS,
CONF_MAC: MAC_ADDRESS,
CONF_ALIAS: ALIAS,
CONF_DEVICE_CONFIG: DEVICE_CONFIG_DICT_AUTH,
CONF_DEVICE_CONFIG: DEVICE_CONFIG_DICT_KLAP,
},
)
await hass.async_block_till_done()
@ -323,7 +327,7 @@ async def test_discovery_new_credentials_invalid(
},
)
assert result3["type"] is FlowResultType.CREATE_ENTRY
assert result3["data"] == CREATE_ENTRY_DATA_AUTH
assert result3["data"] == CREATE_ENTRY_DATA_KLAP
assert result3["context"]["unique_id"] == MAC_ADDRESS
@ -543,7 +547,7 @@ async def test_manual_auth(
await hass.async_block_till_done()
assert result3["type"] is FlowResultType.CREATE_ENTRY
assert result3["title"] == DEFAULT_ENTRY_TITLE
assert result3["data"] == CREATE_ENTRY_DATA_AUTH
assert result3["data"] == CREATE_ENTRY_DATA_KLAP
assert result3["context"]["unique_id"] == MAC_ADDRESS
@ -607,7 +611,7 @@ async def test_manual_auth_errors(
},
)
assert result4["type"] is FlowResultType.CREATE_ENTRY
assert result4["data"] == CREATE_ENTRY_DATA_AUTH
assert result4["data"] == CREATE_ENTRY_DATA_KLAP
assert result4["context"]["unique_id"] == MAC_ADDRESS
await hass.async_block_till_done()
@ -791,16 +795,16 @@ async def test_integration_discovery_with_ip_change(
CONF_HOST: "127.0.0.2",
CONF_MAC: MAC_ADDRESS,
CONF_ALIAS: ALIAS,
CONF_DEVICE_CONFIG: DEVICE_CONFIG_DICT_AUTH,
CONF_DEVICE_CONFIG: DEVICE_CONFIG_DICT_KLAP,
},
)
await hass.async_block_till_done()
assert discovery_result["type"] is FlowResultType.ABORT
assert discovery_result["reason"] == "already_configured"
assert mock_config_entry.data[CONF_DEVICE_CONFIG] == DEVICE_CONFIG_DICT_AUTH
assert mock_config_entry.data[CONF_DEVICE_CONFIG] == DEVICE_CONFIG_DICT_KLAP
assert mock_config_entry.data[CONF_HOST] == "127.0.0.2"
config = DeviceConfig.from_dict(DEVICE_CONFIG_DICT_AUTH)
config = DeviceConfig.from_dict(DEVICE_CONFIG_DICT_KLAP)
mock_connect["connect"].reset_mock(side_effect=True)
bulb = _mocked_device(
@ -832,8 +836,8 @@ async def test_integration_discovery_with_connection_change(
mock_config_entry = MockConfigEntry(
title="TPLink",
domain=DOMAIN,
data=CREATE_ENTRY_DATA_AUTH,
unique_id=MAC_ADDRESS,
data=CREATE_ENTRY_DATA_AES,
unique_id=MAC_ADDRESS2,
)
mock_config_entry.add_to_hass(hass)
with patch("homeassistant.components.tplink.Discover.discover", return_value={}):
@ -849,13 +853,15 @@ async def test_integration_discovery_with_connection_change(
)
== 0
)
assert mock_config_entry.data[CONF_DEVICE_CONFIG] == DEVICE_CONFIG_DICT_AUTH
assert mock_config_entry.data[CONF_DEVICE_CONFIG].get(CONF_HOST) == "127.0.0.1"
assert mock_config_entry.data[CONF_CREDENTIALS_HASH] == CREDENTIALS_HASH_AUTH
assert mock_config_entry.data[CONF_HOST] == "127.0.0.2"
assert mock_config_entry.data[CONF_DEVICE_CONFIG] == DEVICE_CONFIG_DICT_AES
assert mock_config_entry.data[CONF_DEVICE_CONFIG].get(CONF_HOST) == "127.0.0.2"
assert mock_config_entry.data[CONF_CREDENTIALS_HASH] == CREDENTIALS_HASH_AES
NEW_DEVICE_CONFIG = {
**DEVICE_CONFIG_DICT_AUTH,
CONF_CONNECTION_TYPE: NEW_CONNECTION_TYPE_DICT,
**DEVICE_CONFIG_DICT_KLAP,
CONF_CONNECTION_TYPE: CONNECTION_TYPE_KLAP_DICT,
CONF_HOST: "127.0.0.2",
}
config = DeviceConfig.from_dict(NEW_DEVICE_CONFIG)
# Reset the connect mock so when the config flow reloads the entry it succeeds
@ -870,8 +876,8 @@ async def test_integration_discovery_with_connection_change(
DOMAIN,
context={"source": config_entries.SOURCE_INTEGRATION_DISCOVERY},
data={
CONF_HOST: "127.0.0.1",
CONF_MAC: MAC_ADDRESS,
CONF_HOST: "127.0.0.2",
CONF_MAC: MAC_ADDRESS2,
CONF_ALIAS: ALIAS,
CONF_DEVICE_CONFIG: NEW_DEVICE_CONFIG,
},
@ -880,8 +886,8 @@ async def test_integration_discovery_with_connection_change(
assert discovery_result["type"] is FlowResultType.ABORT
assert discovery_result["reason"] == "already_configured"
assert mock_config_entry.data[CONF_DEVICE_CONFIG] == NEW_DEVICE_CONFIG
assert mock_config_entry.data[CONF_HOST] == "127.0.0.1"
assert CREDENTIALS_HASH_AUTH not in mock_config_entry.data
assert mock_config_entry.data[CONF_HOST] == "127.0.0.2"
assert CREDENTIALS_HASH_AES not in mock_config_entry.data
assert mock_config_entry.state is ConfigEntryState.LOADED
@ -953,6 +959,77 @@ async def test_reauth(
await hass.async_block_till_done()
async def test_reauth_update_with_encryption_change(
hass: HomeAssistant,
mock_discovery: AsyncMock,
mock_connect: AsyncMock,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test reauth flow."""
orig_side_effect = mock_connect["connect"].side_effect
mock_connect["connect"].side_effect = AuthenticationError()
mock_config_entry = MockConfigEntry(
title="TPLink",
domain=DOMAIN,
data={**CREATE_ENTRY_DATA_AES},
unique_id=MAC_ADDRESS2,
)
mock_config_entry.add_to_hass(hass)
assert mock_config_entry.data[CONF_DEVICE_CONFIG] == DEVICE_CONFIG_DICT_AES
assert mock_config_entry.data[CONF_CREDENTIALS_HASH] == CREDENTIALS_HASH_AES
with patch("homeassistant.components.tplink.Discover.discover", return_value={}):
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.SETUP_ERROR
caplog.set_level(logging.DEBUG)
flows = hass.config_entries.flow.async_progress()
assert len(flows) == 1
[result] = flows
assert result["step_id"] == "reauth_confirm"
assert mock_config_entry.data[CONF_DEVICE_CONFIG] == DEVICE_CONFIG_DICT_AES
assert CONF_CREDENTIALS_HASH not in mock_config_entry.data
new_config = DeviceConfig(
"127.0.0.2",
credentials=None,
connection_type=Device.ConnectionParameters(
Device.Family.SmartTapoPlug, Device.EncryptionType.Klap
),
uses_http=True,
)
mock_discovery["mock_device"].host = "127.0.0.2"
mock_discovery["mock_device"].config = new_config
mock_discovery["mock_device"].credentials_hash = None
mock_connect["mock_devices"]["127.0.0.2"].config = new_config
mock_connect["mock_devices"]["127.0.0.2"].credentials_hash = CREDENTIALS_HASH_KLAP
mock_connect["connect"].side_effect = orig_side_effect
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_USERNAME: "fake_username",
CONF_PASSWORD: "fake_password",
},
)
await hass.async_block_till_done(wait_background_tasks=True)
assert "Connection type changed for 127.0.0.2" in caplog.text
credentials = Credentials("fake_username", "fake_password")
mock_discovery["discover_single"].assert_called_once_with(
"127.0.0.2", credentials=credentials
)
mock_discovery["mock_device"].update.assert_called_once_with()
assert result2["type"] is FlowResultType.ABORT
assert result2["reason"] == "reauth_successful"
assert mock_config_entry.state is ConfigEntryState.LOADED
assert mock_config_entry.data[CONF_DEVICE_CONFIG] == {
**DEVICE_CONFIG_DICT_KLAP,
CONF_HOST: "127.0.0.2",
}
assert mock_config_entry.data[CONF_CREDENTIALS_HASH] == CREDENTIALS_HASH_KLAP
async def test_reauth_update_from_discovery(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
@ -981,13 +1058,13 @@ async def test_reauth_update_from_discovery(
CONF_HOST: IP_ADDRESS,
CONF_MAC: MAC_ADDRESS,
CONF_ALIAS: ALIAS,
CONF_DEVICE_CONFIG: DEVICE_CONFIG_DICT_AUTH,
CONF_DEVICE_CONFIG: DEVICE_CONFIG_DICT_KLAP,
},
)
await hass.async_block_till_done()
assert discovery_result["type"] is FlowResultType.ABORT
assert discovery_result["reason"] == "already_configured"
assert mock_config_entry.data[CONF_DEVICE_CONFIG] == DEVICE_CONFIG_DICT_AUTH
assert mock_config_entry.data[CONF_DEVICE_CONFIG] == DEVICE_CONFIG_DICT_KLAP
async def test_reauth_update_from_discovery_with_ip_change(
@ -1017,13 +1094,13 @@ async def test_reauth_update_from_discovery_with_ip_change(
CONF_HOST: "127.0.0.2",
CONF_MAC: MAC_ADDRESS,
CONF_ALIAS: ALIAS,
CONF_DEVICE_CONFIG: DEVICE_CONFIG_DICT_AUTH,
CONF_DEVICE_CONFIG: DEVICE_CONFIG_DICT_KLAP,
},
)
await hass.async_block_till_done()
assert discovery_result["type"] is FlowResultType.ABORT
assert discovery_result["reason"] == "already_configured"
assert mock_config_entry.data[CONF_DEVICE_CONFIG] == DEVICE_CONFIG_DICT_AUTH
assert mock_config_entry.data[CONF_DEVICE_CONFIG] == DEVICE_CONFIG_DICT_KLAP
assert mock_config_entry.data[CONF_HOST] == "127.0.0.2"
@ -1040,7 +1117,7 @@ async def test_reauth_no_update_if_config_and_ip_the_same(
mock_config_entry,
data={
**mock_config_entry.data,
CONF_DEVICE_CONFIG: DEVICE_CONFIG_DICT_AUTH,
CONF_DEVICE_CONFIG: DEVICE_CONFIG_DICT_KLAP,
},
)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
@ -1051,7 +1128,7 @@ async def test_reauth_no_update_if_config_and_ip_the_same(
assert len(flows) == 1
[result] = flows
assert result["step_id"] == "reauth_confirm"
assert mock_config_entry.data[CONF_DEVICE_CONFIG] == DEVICE_CONFIG_DICT_AUTH
assert mock_config_entry.data[CONF_DEVICE_CONFIG] == DEVICE_CONFIG_DICT_KLAP
discovery_result = await hass.config_entries.flow.async_init(
DOMAIN,
@ -1060,13 +1137,13 @@ async def test_reauth_no_update_if_config_and_ip_the_same(
CONF_HOST: IP_ADDRESS,
CONF_MAC: MAC_ADDRESS,
CONF_ALIAS: ALIAS,
CONF_DEVICE_CONFIG: DEVICE_CONFIG_DICT_AUTH,
CONF_DEVICE_CONFIG: DEVICE_CONFIG_DICT_KLAP,
},
)
await hass.async_block_till_done()
assert discovery_result["type"] is FlowResultType.ABORT
assert discovery_result["reason"] == "already_configured"
assert mock_config_entry.data[CONF_DEVICE_CONFIG] == DEVICE_CONFIG_DICT_AUTH
assert mock_config_entry.data[CONF_DEVICE_CONFIG] == DEVICE_CONFIG_DICT_KLAP
assert mock_config_entry.data[CONF_HOST] == IP_ADDRESS
@ -1214,15 +1291,20 @@ async def test_discovery_timeout_connect(
async def test_reauth_update_other_flows(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_discovery: AsyncMock,
mock_connect: AsyncMock,
) -> None:
"""Test reauth updates other reauth flows."""
mock_config_entry = MockConfigEntry(
title="TPLink",
domain=DOMAIN,
data={**CREATE_ENTRY_DATA_KLAP},
unique_id=MAC_ADDRESS,
)
mock_config_entry2 = MockConfigEntry(
title="TPLink",
domain=DOMAIN,
data={**CREATE_ENTRY_DATA_AUTH2},
data={**CREATE_ENTRY_DATA_AES},
unique_id=MAC_ADDRESS2,
)
default_side_effect = mock_connect["connect"].side_effect
@ -1244,7 +1326,7 @@ async def test_reauth_update_other_flows(
flows_by_entry_id = {flow["context"]["entry_id"]: flow for flow in flows}
result = flows_by_entry_id[mock_config_entry.entry_id]
assert result["step_id"] == "reauth_confirm"
assert mock_config_entry.data[CONF_DEVICE_CONFIG] == DEVICE_CONFIG_DICT_LEGACY
assert mock_config_entry.data[CONF_DEVICE_CONFIG] == DEVICE_CONFIG_DICT_KLAP
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={

36
tests/components/tplink/test_init.py
View File

@ -33,9 +33,9 @@ from homeassistant.setup import async_setup_component
from homeassistant.util import dt as dt_util
from . import (
CREATE_ENTRY_DATA_AUTH,
CREATE_ENTRY_DATA_KLAP,
CREATE_ENTRY_DATA_LEGACY,
DEVICE_CONFIG_AUTH,
DEVICE_CONFIG_KLAP,
DEVICE_ID,
DEVICE_ID_MAC,
IP_ADDRESS,
@ -178,7 +178,7 @@ async def test_config_entry_device_config(
mock_config_entry = MockConfigEntry(
title="TPLink",
domain=DOMAIN,
data={**CREATE_ENTRY_DATA_AUTH},
data={**CREATE_ENTRY_DATA_KLAP},
unique_id=MAC_ADDRESS,
)
mock_config_entry.add_to_hass(hass)
@ -197,7 +197,7 @@ async def test_config_entry_with_stored_credentials(
mock_config_entry = MockConfigEntry(
title="TPLink",
domain=DOMAIN,
data={**CREATE_ENTRY_DATA_AUTH},
data={**CREATE_ENTRY_DATA_KLAP},
unique_id=MAC_ADDRESS,
)
auth = {
@ -210,7 +210,7 @@ async def test_config_entry_with_stored_credentials(
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.LOADED
config = DEVICE_CONFIG_AUTH
config = DEVICE_CONFIG_KLAP
assert config.credentials != stored_credentials
config.credentials = stored_credentials
mock_connect["connect"].assert_called_once_with(config=config)
@ -223,7 +223,7 @@ async def test_config_entry_device_config_invalid(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test that an invalid device config logs an error and loads the config entry."""
entry_data = copy.deepcopy(CREATE_ENTRY_DATA_AUTH)
entry_data = copy.deepcopy(CREATE_ENTRY_DATA_KLAP)
entry_data[CONF_DEVICE_CONFIG] = {"foo": "bar"}
mock_config_entry = MockConfigEntry(
title="TPLink",
@ -263,7 +263,7 @@ async def test_config_entry_errors(
mock_config_entry = MockConfigEntry(
title="TPLink",
domain=DOMAIN,
data={**CREATE_ENTRY_DATA_AUTH},
data={**CREATE_ENTRY_DATA_KLAP},
unique_id=MAC_ADDRESS,
)
mock_config_entry.add_to_hass(hass)
@ -520,11 +520,11 @@ async def test_move_credentials_hash(
from the device.
"""
device_config = {
**DEVICE_CONFIG_AUTH.to_dict(
**DEVICE_CONFIG_KLAP.to_dict(
exclude_credentials=True, credentials_hash="theHash"
)
}
entry_data = {**CREATE_ENTRY_DATA_AUTH, CONF_DEVICE_CONFIG: device_config}
entry_data = {**CREATE_ENTRY_DATA_KLAP, CONF_DEVICE_CONFIG: device_config}
entry = MockConfigEntry(
title="TPLink",
@ -567,11 +567,11 @@ async def test_move_credentials_hash_auth_error(
in async_setup_entry.
"""
device_config = {
**DEVICE_CONFIG_AUTH.to_dict(
**DEVICE_CONFIG_KLAP.to_dict(
exclude_credentials=True, credentials_hash="theHash"
)
}
entry_data = {**CREATE_ENTRY_DATA_AUTH, CONF_DEVICE_CONFIG: device_config}
entry_data = {**CREATE_ENTRY_DATA_KLAP, CONF_DEVICE_CONFIG: device_config}
entry = MockConfigEntry(
title="TPLink",
@ -610,11 +610,11 @@ async def test_move_credentials_hash_other_error(
at the end of the test.
"""
device_config = {
**DEVICE_CONFIG_AUTH.to_dict(
**DEVICE_CONFIG_KLAP.to_dict(
exclude_credentials=True, credentials_hash="theHash"
)
}
entry_data = {**CREATE_ENTRY_DATA_AUTH, CONF_DEVICE_CONFIG: device_config}
entry_data = {**CREATE_ENTRY_DATA_KLAP, CONF_DEVICE_CONFIG: device_config}
entry = MockConfigEntry(
title="TPLink",
@ -647,9 +647,9 @@ async def test_credentials_hash(
hass: HomeAssistant,
) -> None:
"""Test credentials_hash used to call connect."""
device_config = {**DEVICE_CONFIG_AUTH.to_dict(exclude_credentials=True)}
device_config = {**DEVICE_CONFIG_KLAP.to_dict(exclude_credentials=True)}
entry_data = {
**CREATE_ENTRY_DATA_AUTH,
**CREATE_ENTRY_DATA_KLAP,
CONF_DEVICE_CONFIG: device_config,
CONF_CREDENTIALS_HASH: "theHash",
}
@ -684,9 +684,9 @@ async def test_credentials_hash_auth_error(
hass: HomeAssistant,
) -> None:
"""Test credentials_hash is deleted after an auth failure."""
device_config = {**DEVICE_CONFIG_AUTH.to_dict(exclude_credentials=True)}
device_config = {**DEVICE_CONFIG_KLAP.to_dict(exclude_credentials=True)}
entry_data = {
**CREATE_ENTRY_DATA_AUTH,
**CREATE_ENTRY_DATA_KLAP,
CONF_DEVICE_CONFIG: device_config,
CONF_CREDENTIALS_HASH: "theHash",
}
@ -710,7 +710,7 @@ async def test_credentials_hash_auth_error(
await hass.async_block_till_done()
expected_config = DeviceConfig.from_dict(
DEVICE_CONFIG_AUTH.to_dict(exclude_credentials=True, credentials_hash="theHash")
DEVICE_CONFIG_KLAP.to_dict(exclude_credentials=True, credentials_hash="theHash")
)
connect_mock.assert_called_with(config=expected_config)
assert entry.state is ConfigEntryState.SETUP_ERROR