jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-24 17:27:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add reconfigure support to ESPHome (#143132)

Browse Source
This commit is contained in:
J. Nick Koston 2025-04-19 00:40:07 -10:00 committed by GitHub
parent ff1ab1da37
commit 3e3697dc7a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 536 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

170
homeassistant/components/esphome/config_flow.py
View File

@ -23,6 +23,7 @@ import voluptuous as vol
from homeassistant.components import zeroconf
from homeassistant.config_entries import (
SOURCE_REAUTH,
SOURCE_RECONFIGURE,
ConfigEntry,
ConfigFlow,
ConfigFlowResult,
@ -44,6 +45,7 @@ from .const import (
CONF_SUBSCRIBE_LOGS,
DEFAULT_ALLOW_SERVICE_CALLS,
DEFAULT_NEW_CONFIG_ALLOW_ALLOW_SERVICE_CALLS,
DEFAULT_PORT,
DOMAIN,
)
from .dashboard import async_get_or_create_dashboard_manager, async_set_dashboard_info
@ -63,6 +65,7 @@ class EsphomeFlowHandler(ConfigFlow, domain=DOMAIN):
VERSION = 1
_reauth_entry: ConfigEntry
_reconfig_entry: ConfigEntry
def __init__(self) -> None:
"""Initialize flow."""
@ -88,7 +91,7 @@ class EsphomeFlowHandler(ConfigFlow, domain=DOMAIN):
fields: dict[Any, type] = OrderedDict()
fields[vol.Required(CONF_HOST, default=self._host or vol.UNDEFINED)] = str
fields[vol.Optional(CONF_PORT, default=self._port or 6053)] = int
fields[vol.Optional(CONF_PORT, default=self._port or DEFAULT_PORT)] = int
errors = {}
if error is not None:
@ -140,7 +143,7 @@ class EsphomeFlowHandler(ConfigFlow, domain=DOMAIN):
"""Handle reauthorization flow when encryption was removed."""
if user_input is not None:
self._noise_psk = None
return await self._async_get_entry_or_resolve_conflict()
return await self._async_validated_connection()
return self.async_show_form(
step_id="reauth_encryption_removed_confirm",
@ -172,6 +175,18 @@ class EsphomeFlowHandler(ConfigFlow, domain=DOMAIN):
description_placeholders={"name": self._name},
)
async def async_step_reconfigure(
self, entry_data: Mapping[str, Any]
) -> ConfigFlowResult:
"""Handle a flow initialized by a reconfig request."""
self._reconfig_entry = self._get_reconfigure_entry()
data = self._reconfig_entry.data
self._host = data[CONF_HOST]
self._port = data.get(CONF_PORT, DEFAULT_PORT)
self._noise_psk = data.get(CONF_NOISE_PSK)
self._device_name = data.get(CONF_DEVICE_NAME)
return await self._async_step_user_base()
@property
def _name(self) -> str:
return self.__name or "ESPHome"
@ -230,7 +245,7 @@ class EsphomeFlowHandler(ConfigFlow, domain=DOMAIN):
return await self.async_step_authenticate()
self._password = ""
return await self._async_get_entry_or_resolve_conflict()
return await self._async_validated_connection()
async def async_step_discovery_confirm(
self, user_input: dict[str, Any] | None = None
@ -270,13 +285,13 @@ class EsphomeFlowHandler(ConfigFlow, domain=DOMAIN):
await self._async_validate_mac_abort_configured(
mac_address, self._host, self._port
)
return await self.async_step_discovery_confirm()
async def _async_validate_mac_abort_configured(
self, formatted_mac: str, host: str, port: int | None
) -> None:
"""Validate if the MAC address is already configured."""
assert self.unique_id is not None
if not (
entry := self.hass.config_entries.async_entry_for_domain_unique_id(
self.handler, formatted_mac
@ -393,7 +408,7 @@ class EsphomeFlowHandler(ConfigFlow, domain=DOMAIN):
data={
**self._entry_with_name_conflict.data,
CONF_HOST: self._host,
CONF_PORT: self._port or 6053,
CONF_PORT: self._port or DEFAULT_PORT,
CONF_PASSWORD: self._password or "",
CONF_NOISE_PSK: self._noise_psk or "",
},
@ -417,20 +432,24 @@ class EsphomeFlowHandler(ConfigFlow, domain=DOMAIN):
await self.hass.config_entries.async_remove(
self._entry_with_name_conflict.entry_id
)
return self._async_get_entry()
async def _async_get_entry_or_resolve_conflict(self) -> ConfigFlowResult:
"""Return the entry or resolve a conflict."""
if self.source != SOURCE_REAUTH:
for entry in self._async_current_entries(include_ignore=False):
if entry.data.get(CONF_DEVICE_NAME) == self._device_name:
self._entry_with_name_conflict = entry
return await self.async_step_name_conflict()
return self._async_get_entry()
return self._async_create_entry()
@callback
def _async_get_entry(self) -> ConfigFlowResult:
config_data = {
def _async_create_entry(self) -> ConfigFlowResult:
"""Create the config entry."""
assert self._name is not None
return self.async_create_entry(
title=self._name,
data=self._async_make_config_data(),
options={
CONF_ALLOW_SERVICE_CALLS: DEFAULT_NEW_CONFIG_ALLOW_ALLOW_SERVICE_CALLS,
},
)
@callback
def _async_make_config_data(self) -> dict[str, Any]:
"""Return config data for the entry."""
return {
CONF_HOST: self._host,
CONF_PORT: self._port,
# The API uses protobuf, so empty string denotes absence
@ -438,19 +457,99 @@ class EsphomeFlowHandler(ConfigFlow, domain=DOMAIN):
CONF_NOISE_PSK: self._noise_psk or "",
CONF_DEVICE_NAME: self._device_name,
}
config_options = {
CONF_ALLOW_SERVICE_CALLS: DEFAULT_NEW_CONFIG_ALLOW_ALLOW_SERVICE_CALLS,
}
if self.source == SOURCE_REAUTH:
return self.async_update_reload_and_abort(
self._reauth_entry, data=self._reauth_entry.data | config_data
)
assert self._name is not None
return self.async_create_entry(
title=self._name,
data=config_data,
options=config_options,
async def _async_validated_connection(self) -> ConfigFlowResult:
"""Handle validated connection."""
if self.source == SOURCE_RECONFIGURE:
return await self._async_reconfig_validated_connection()
if self.source == SOURCE_REAUTH:
return await self._async_reauth_validated_connection()
for entry in self._async_current_entries(include_ignore=False):
if entry.data.get(CONF_DEVICE_NAME) == self._device_name:
self._entry_with_name_conflict = entry
return await self.async_step_name_conflict()
return self._async_create_entry()
async def _async_reauth_validated_connection(self) -> ConfigFlowResult:
"""Handle reauth validated connection."""
assert self._reauth_entry.unique_id is not None
if self.unique_id == self._reauth_entry.unique_id:
return self.async_update_reload_and_abort(
self._reauth_entry,
data=self._reauth_entry.data | self._async_make_config_data(),
)
assert self._host is not None
self._abort_if_unique_id_configured(
updates={
CONF_HOST: self._host,
CONF_PORT: self._port,
CONF_NOISE_PSK: self._noise_psk,
}
)
# Reauth was triggered a while ago, and since than
# a new device resides at the same IP address.
assert self._device_name is not None
return self.async_abort(
reason="reauth_unique_id_changed",
description_placeholders={
"name": self._reauth_entry.data.get(
CONF_DEVICE_NAME, self._reauth_entry.title
),
"host": self._host,
"expected_mac": format_mac(self._reauth_entry.unique_id),
"unexpected_mac": format_mac(self.unique_id),
"unexpected_device_name": self._device_name,
},
)
async def _async_reconfig_validated_connection(self) -> ConfigFlowResult:
"""Handle reconfigure validated connection."""
assert self._reconfig_entry.unique_id is not None
assert self._host is not None
assert self._device_name is not None
if not (
unique_id_matches := (self.unique_id == self._reconfig_entry.unique_id)
):
self._abort_if_unique_id_configured(
updates={
CONF_HOST: self._host,
CONF_PORT: self._port,
CONF_NOISE_PSK: self._noise_psk,
}
)
for entry in self._async_current_entries(include_ignore=False):
if (
entry.entry_id != self._reconfig_entry.entry_id
and entry.data.get(CONF_DEVICE_NAME) == self._device_name
):
return self.async_abort(
reason="reconfigure_name_conflict",
description_placeholders={
"name": self._reconfig_entry.data[CONF_DEVICE_NAME],
"host": self._host,
"expected_mac": format_mac(self._reconfig_entry.unique_id),
"existing_title": entry.title,
},
)
if unique_id_matches:
return self.async_update_reload_and_abort(
self._reconfig_entry,
data=self._reconfig_entry.data | self._async_make_config_data(),
)
if self._reconfig_entry.data.get(CONF_DEVICE_NAME) == self._device_name:
self._entry_with_name_conflict = self._reconfig_entry
return await self.async_step_name_conflict()
return self.async_abort(
reason="reconfigure_unique_id_changed",
description_placeholders={
"name": self._reconfig_entry.data.get(
CONF_DEVICE_NAME, self._reconfig_entry.title
),
"host": self._host,
"expected_mac": format_mac(self._reconfig_entry.unique_id),
"unexpected_mac": format_mac(self.unique_id),
"unexpected_device_name": self._device_name,
},
)
async def async_step_encryption_key(
@ -481,7 +580,7 @@ class EsphomeFlowHandler(ConfigFlow, domain=DOMAIN):
error = await self.try_login()
if error:
return await self.async_step_authenticate(error=error)
return await self._async_get_entry_or_resolve_conflict()
return await self._async_validated_connection()
errors = {}
if error is not None:
@ -501,12 +600,11 @@ class EsphomeFlowHandler(ConfigFlow, domain=DOMAIN):
zeroconf_instance = await zeroconf.async_get_instance(self.hass)
cli = APIClient(
host,
port or 6053,
port or DEFAULT_PORT,
"",
zeroconf_instance=zeroconf_instance,
noise_psk=noise_psk,
)
try:
await cli.connect()
self._device_info = await cli.device_info()
@ -541,9 +639,13 @@ class EsphomeFlowHandler(ConfigFlow, domain=DOMAIN):
assert self._device_info is not None
mac_address = format_mac(self._device_info.mac_address)
await self.async_set_unique_id(mac_address, raise_on_progress=False)
if self.source != SOURCE_REAUTH:
if self.source not in (SOURCE_REAUTH, SOURCE_RECONFIGURE):
self._abort_if_unique_id_configured(
updates={CONF_HOST: self._host, CONF_PORT: self._port}
updates={
CONF_HOST: self._host,
CONF_PORT: self._port,
CONF_NOISE_PSK: self._noise_psk,
}
)
return None

3
homeassistant/components/esphome/const.py
View File

@ -1,5 +1,7 @@
"""ESPHome constants."""
from typing import Final
from awesomeversion import AwesomeVersion
DOMAIN = "esphome"
@ -13,6 +15,7 @@ CONF_BLUETOOTH_MAC_ADDRESS = "bluetooth_mac_address"
DEFAULT_ALLOW_SERVICE_CALLS = True
DEFAULT_NEW_CONFIG_ALLOW_ALLOW_SERVICE_CALLS = False
DEFAULT_PORT: Final = 6053
STABLE_BLE_VERSION_STR = "2025.2.2"
STABLE_BLE_VERSION = AwesomeVersion(STABLE_BLE_VERSION_STR)

6
homeassistant/components/esphome/strings.json
View File

@ -10,7 +10,11 @@
"mqtt_missing_api": "Missing API port in MQTT properties.",
"mqtt_missing_ip": "Missing IP address in MQTT properties.",
"mqtt_missing_payload": "Missing MQTT Payload.",
"name_conflict_migrated": "The configuration for `{name}` has been migrated to a new device with MAC address `{mac}` from `{existing_mac}`."
"name_conflict_migrated": "The configuration for `{name}` has been migrated to a new device with MAC address `{mac}` from `{existing_mac}`.",
"reconfigure_successful": "[%key:common::config_flow::abort::reconfigure_successful%]",
"reauth_unique_id_changed": "**Re-authentication of `{name}` was aborted** because the address `{host}` points to a different device: `{unexpected_device_name}` (MAC: `{unexpected_mac}`) instead of the expected one (MAC: `{expected_mac}`).",
"reconfigure_name_conflict": "**Reconfiguration of `{name}` was aborted** because the address `{host}` points to a device named `{name}` (MAC: `{expected_mac}`), which is already in use by another configuration entry: `{existing_title}`.",
"reconfigure_unique_id_changed": "**Reconfiguration of `{name}` was aborted** because the address `{host}` points to a different device: `{unexpected_device_name}` (MAC: `{unexpected_mac}`) instead of the expected one (MAC: `{expected_mac}`)."
},
"error": {
"resolve_error": "Can't resolve address of the ESP. If this error persists, please set a static IP address",

402
tests/components/esphome/test_config_flow.py
View File

@ -813,12 +813,15 @@ async def test_reauth_confirm_valid(
entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_HOST: "127.0.0.1", CONF_PORT: 6053, CONF_PASSWORD: ""},
unique_id="11:22:33:44:55:aa",
)
entry.add_to_hass(hass)
result = await entry.start_reauth_flow(hass)
mock_client.device_info.return_value = DeviceInfo(uses_password=False, name="test")
mock_client.device_info.return_value = DeviceInfo(
uses_password=False, name="test", mac_address="11:22:33:44:55:aa"
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_NOISE_PSK: VALID_NOISE_PSK}
)
@ -828,6 +831,48 @@ async def test_reauth_confirm_valid(
assert entry.data[CONF_NOISE_PSK] == VALID_NOISE_PSK
@pytest.mark.usefixtures("mock_zeroconf", "mock_setup_entry")
async def test_reauth_attempt_to_change_mac_aborts(
hass: HomeAssistant, mock_client: APIClient
) -> None:
"""Test reauth initiation with valid PSK attempting to change mac.
This can happen if reauth starts, but they don't finish it before
a new device takes the place of the old one at the same IP.
"""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "127.0.0.1",
CONF_PORT: 6053,
CONF_PASSWORD: "",
CONF_DEVICE_NAME: "test",
},
unique_id="11:22:33:44:55:aa",
)
entry.add_to_hass(hass)
result = await entry.start_reauth_flow(hass)
mock_client.device_info.return_value = DeviceInfo(
uses_password=False, name="test", mac_address="11:22:33:44:55:bb"
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_NOISE_PSK: VALID_NOISE_PSK}
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reauth_unique_id_changed"
assert CONF_NOISE_PSK not in entry.data
assert result["description_placeholders"] == {
"expected_mac": "11:22:33:44:55:aa",
"host": "127.0.0.1",
"name": "test",
"unexpected_device_name": "test",
"unexpected_mac": "11:22:33:44:55:bb",
}
@pytest.mark.usefixtures("mock_zeroconf")
async def test_reauth_fixed_via_dashboard(
hass: HomeAssistant,
@ -845,10 +890,13 @@ async def test_reauth_fixed_via_dashboard(
CONF_PASSWORD: "",
CONF_DEVICE_NAME: "test",
},
unique_id="11:22:33:44:55:aa",
)
entry.add_to_hass(hass)
mock_client.device_info.return_value = DeviceInfo(uses_password=False, name="test")
mock_client.device_info.return_value = DeviceInfo(
uses_password=False, name="test", mac_address="11:22:33:44:55:aa"
)
mock_dashboard["configured"].append(
{
@ -883,7 +931,7 @@ async def test_reauth_fixed_via_dashboard_add_encryption_remove_password(
"""Test reauth fixed automatically via dashboard with password removed."""
mock_client.device_info.side_effect = (
InvalidAuthAPIError,
DeviceInfo(uses_password=False, name="test"),
DeviceInfo(uses_password=False, name="test", mac_address="11:22:33:44:55:aa"),
)
mock_dashboard["configured"].append(
@ -917,7 +965,9 @@ async def test_reauth_fixed_via_remove_password(
mock_setup_entry: None,
) -> None:
"""Test reauth fixed automatically by seeing password removed."""
mock_client.device_info.return_value = DeviceInfo(uses_password=False, name="test")
mock_client.device_info.return_value = DeviceInfo(
uses_password=False, name="test", mac_address="11:22:33:44:55:aa"
)
result = await mock_config_entry.start_reauth_flow(hass)
@ -943,10 +993,13 @@ async def test_reauth_fixed_via_dashboard_at_confirm(
CONF_PASSWORD: "",
CONF_DEVICE_NAME: "test",
},
unique_id="11:22:33:44:55:aa",
)
entry.add_to_hass(hass)
mock_client.device_info.return_value = DeviceInfo(uses_password=False, name="test")
mock_client.device_info.return_value = DeviceInfo(
uses_password=False, name="test", mac_address="11:22:33:44:55:aa"
)
result = await entry.start_reauth_flow(hass)
@ -984,6 +1037,7 @@ async def test_reauth_confirm_invalid(
entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_HOST: "127.0.0.1", CONF_PORT: 6053, CONF_PASSWORD: ""},
unique_id="11:22:33:44:55:aa",
)
entry.add_to_hass(hass)
@ -1000,7 +1054,9 @@ async def test_reauth_confirm_invalid(
assert result["errors"]["base"] == "invalid_psk"
mock_client.device_info = AsyncMock(
return_value=DeviceInfo(uses_password=False, name="test")
return_value=DeviceInfo(
uses_password=False, name="test", mac_address="11:22:33:44:55:aa"
)
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_NOISE_PSK: VALID_NOISE_PSK}
@ -1019,7 +1075,7 @@ async def test_reauth_confirm_invalid_with_unique_id(
entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_HOST: "127.0.0.1", CONF_PORT: 6053, CONF_PASSWORD: ""},
unique_id="test",
unique_id="11:22:33:44:55:aa",
)
entry.add_to_hass(hass)
@ -1036,7 +1092,9 @@ async def test_reauth_confirm_invalid_with_unique_id(
assert result["errors"]["base"] == "invalid_psk"
mock_client.device_info = AsyncMock(
return_value=DeviceInfo(uses_password=False, name="test")
return_value=DeviceInfo(
uses_password=False, name="test", mac_address="11:22:33:44:55:aa"
)
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_NOISE_PSK: VALID_NOISE_PSK}
@ -1049,7 +1107,7 @@ async def test_reauth_confirm_invalid_with_unique_id(
@pytest.mark.usefixtures("mock_zeroconf")
async def test_reauth_encryption_key_removed(
hass: HomeAssistant, mock_client, mock_setup_entry: None
hass: HomeAssistant, mock_client: APIClient, mock_setup_entry: None
) -> None:
"""Test reauth when the encryption key was removed."""
entry = MockConfigEntry(
@ -1060,7 +1118,7 @@ async def test_reauth_encryption_key_removed(
CONF_PASSWORD: "",
CONF_NOISE_PSK: VALID_NOISE_PSK,
},
unique_id="test",
unique_id="11:22:33:44:55:aa",
)
entry.add_to_hass(hass)
@ -1660,7 +1718,11 @@ async def test_user_flow_name_conflict_migrate(
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "name_conflict_migrated"
assert result["description_placeholders"] == {
"existing_mac": "11:22:33:44:55:cc",
"mac": "11:22:33:44:55:aa",
"name": "test",
}
assert existing_entry.data == {
CONF_HOST: "127.0.0.1",
CONF_PORT: 6053,
@ -1715,3 +1777,321 @@ async def test_user_flow_name_conflict_overwrite(
CONF_DEVICE_NAME: "test",
}
assert result["context"]["unique_id"] == "11:22:33:44:55:aa"
@pytest.mark.usefixtures("mock_zeroconf", "mock_setup_entry")
async def test_reconfig_success_with_same_ip_new_name(
hass: HomeAssistant, mock_client: APIClient
) -> None:
"""Test reconfig initiation with same ip and new name."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "127.0.0.1",
CONF_PORT: 6053,
CONF_PASSWORD: "",
CONF_DEVICE_NAME: "test",
},
unique_id="11:22:33:44:55:aa",
)
entry.add_to_hass(hass)
result = await entry.start_reconfigure_flow(hass)
mock_client.device_info.return_value = DeviceInfo(
uses_password=False, name="other", mac_address="11:22:33:44:55:aa"
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_HOST: "127.0.0.1", CONF_PORT: 6053}
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reconfigure_successful"
assert entry.data[CONF_HOST] == "127.0.0.1"
assert entry.data[CONF_DEVICE_NAME] == "other"
@pytest.mark.usefixtures("mock_zeroconf", "mock_setup_entry")
async def test_reconfig_success_with_new_ip_new_name(
hass: HomeAssistant, mock_client: APIClient
) -> None:
"""Test reconfig initiation with new ip and new name."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "127.0.0.1",
CONF_PORT: 6053,
CONF_PASSWORD: "",
CONF_DEVICE_NAME: "test",
},
unique_id="11:22:33:44:55:aa",
)
entry.add_to_hass(hass)
result = await entry.start_reconfigure_flow(hass)
mock_client.device_info.return_value = DeviceInfo(
uses_password=False, name="other", mac_address="11:22:33:44:55:aa"
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_HOST: "127.0.0.2", CONF_PORT: 6053}
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reconfigure_successful"
assert entry.data[CONF_HOST] == "127.0.0.2"
assert entry.data[CONF_DEVICE_NAME] == "other"
@pytest.mark.usefixtures("mock_zeroconf", "mock_setup_entry")
async def test_reconfig_success_with_new_ip_same_name(
hass: HomeAssistant, mock_client: APIClient
) -> None:
"""Test reconfig initiation with new ip and same name."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "127.0.0.1",
CONF_PORT: 6053,
CONF_PASSWORD: "",
CONF_DEVICE_NAME: "test",
CONF_NOISE_PSK: VALID_NOISE_PSK,
},
unique_id="11:22:33:44:55:aa",
)
entry.add_to_hass(hass)
result = await entry.start_reconfigure_flow(hass)
mock_client.device_info.return_value = DeviceInfo(
uses_password=False, name="test", mac_address="11:22:33:44:55:aa"
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_HOST: "127.0.0.1", CONF_PORT: 6053}
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reconfigure_successful"
assert entry.data[CONF_HOST] == "127.0.0.1"
assert entry.data[CONF_DEVICE_NAME] == "test"
assert entry.data[CONF_NOISE_PSK] == VALID_NOISE_PSK
@pytest.mark.usefixtures("mock_zeroconf", "mock_setup_entry")
async def test_reconfig_name_conflict_with_existing_entry(
hass: HomeAssistant, mock_client: APIClient
) -> None:
"""Test reconfig with a name conflict with an existing entry."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "127.0.0.1",
CONF_PORT: 6053,
CONF_PASSWORD: "",
CONF_DEVICE_NAME: "test",
},
unique_id="11:22:33:44:55:aa",
)
entry.add_to_hass(hass)
entry2 = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "127.0.0.2",
CONF_PORT: 6053,
CONF_PASSWORD: "",
CONF_DEVICE_NAME: "other",
},
unique_id="11:22:33:44:55:bb",
)
entry2.add_to_hass(hass)
result = await entry.start_reconfigure_flow(hass)
mock_client.device_info.return_value = DeviceInfo(
uses_password=False, name="other", mac_address="11:22:33:44:55:aa"
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_HOST: "127.0.0.3", CONF_PORT: 6053}
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reconfigure_name_conflict"
assert result["description_placeholders"] == {
"existing_title": "Mock Title",
"expected_mac": "11:22:33:44:55:aa",
"host": "127.0.0.3",
"name": "test",
}
@pytest.mark.usefixtures("mock_zeroconf", "mock_setup_entry")
async def test_reconfig_attempt_to_change_mac_aborts(
hass: HomeAssistant, mock_client: APIClient
) -> None:
"""Test reconfig initiation with valid PSK attempting to change mac."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "127.0.0.1",
CONF_PORT: 6053,
CONF_PASSWORD: "",
CONF_DEVICE_NAME: "test",
},
unique_id="11:22:33:44:55:aa",
)
entry.add_to_hass(hass)
result = await entry.start_reconfigure_flow(hass)
mock_client.device_info.return_value = DeviceInfo(
uses_password=False, name="other", mac_address="11:22:33:44:55:bb"
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_HOST: "127.0.0.2", CONF_PORT: 6053}
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reconfigure_unique_id_changed"
assert CONF_NOISE_PSK not in entry.data
assert result["description_placeholders"] == {
"expected_mac": "11:22:33:44:55:aa",
"host": "127.0.0.2",
"name": "test",
"unexpected_device_name": "other",
"unexpected_mac": "11:22:33:44:55:bb",
}
@pytest.mark.usefixtures("mock_zeroconf", "mock_setup_entry")
async def test_reconfig_mac_used_by_other_entry(
hass: HomeAssistant, mock_client: APIClient
) -> None:
"""Test reconfig when there is another entry for the mac."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "127.0.0.1",
CONF_PORT: 6053,
CONF_PASSWORD: "",
CONF_DEVICE_NAME: "test",
},
unique_id="11:22:33:44:55:aa",
)
entry.add_to_hass(hass)
entry2 = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "127.0.0.2",
CONF_PORT: 6053,
CONF_PASSWORD: "",
CONF_DEVICE_NAME: "test4",
},
unique_id="11:22:33:44:55:bb",
)
entry2.add_to_hass(hass)
result = await entry.start_reconfigure_flow(hass)
mock_client.device_info.return_value = DeviceInfo(
uses_password=False, name="test", mac_address="11:22:33:44:55:bb"
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_HOST: "127.0.0.2", CONF_PORT: 6053}
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
@pytest.mark.usefixtures("mock_zeroconf", "mock_setup_entry")
async def test_reconfig_name_conflict_migrate(
hass: HomeAssistant, mock_client: APIClient
) -> None:
"""Test reconfig initiation when device has been replaced."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "127.0.0.1",
CONF_PORT: 6053,
CONF_PASSWORD: "",
CONF_DEVICE_NAME: "test",
},
unique_id="11:22:33:44:55:aa",
)
entry.add_to_hass(hass)
result = await entry.start_reconfigure_flow(hass)
mock_client.device_info.return_value = DeviceInfo(
uses_password=False, name="test", mac_address="11:22:33:44:55:bb"
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_HOST: "127.0.0.2", CONF_PORT: 6053}
)
assert result["type"] is FlowResultType.MENU
assert result["step_id"] == "name_conflict"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={"next_step_id": "name_conflict_migrate"}
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "name_conflict_migrated"
assert entry.data == {
CONF_HOST: "127.0.0.2",
CONF_PORT: 6053,
CONF_PASSWORD: "",
CONF_NOISE_PSK: "",
CONF_DEVICE_NAME: "test",
}
assert entry.unique_id == "11:22:33:44:55:bb"
@pytest.mark.usefixtures("mock_zeroconf", "mock_setup_entry")
async def test_reconfig_name_conflict_overwrite(
hass: HomeAssistant, mock_client: APIClient
) -> None:
"""Test reconfig initiation when device has been replaced."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "127.0.0.1",
CONF_PORT: 6053,
CONF_PASSWORD: "",
CONF_DEVICE_NAME: "test",
},
unique_id="11:22:33:44:55:aa",
)
entry.add_to_hass(hass)
result = await entry.start_reconfigure_flow(hass)
mock_client.device_info.return_value = DeviceInfo(
uses_password=False, name="test", mac_address="11:22:33:44:55:bb"
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_HOST: "127.0.0.2", CONF_PORT: 6053}
)
assert result["type"] is FlowResultType.MENU
assert result["step_id"] == "name_conflict"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={"next_step_id": "name_conflict_overwrite"}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"] == {
CONF_HOST: "127.0.0.2",
CONF_PORT: 6053,
CONF_PASSWORD: "",
CONF_NOISE_PSK: "",
CONF_DEVICE_NAME: "test",
}
assert result["context"]["unique_id"] == "11:22:33:44:55:bb"
assert (
hass.config_entries.async_entry_for_domain_unique_id(
DOMAIN, "11:22:33:44:55:aa"
)
is None
)

2
tests/components/esphome/test_dashboard.py
View File

@ -193,7 +193,7 @@ async def test_new_dashboard_fix_reauth(
"""Test config entries waiting for reauth are triggered."""
mock_client.device_info.side_effect = (
InvalidAuthAPIError,
DeviceInfo(uses_password=False, name="test"),
DeviceInfo(uses_password=False, name="test", mac_address="11:22:33:44:55:AA"),
)
with patch(