Add reconfiguration flow to NUT (#142127)

* Add reconfiguration flow

* Check host/port/alias without comparing strings

* Replace repeat strings with references
This commit is contained in:
tdfountain 2025-04-05 14:02:46 -07:00 committed by GitHub
parent 33cbebc727
commit cd7d7cd35c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 948 additions and 20 deletions

View File

@ -4,6 +4,7 @@ from __future__ import annotations
from collections.abc import Mapping from collections.abc import Mapping
import logging import logging
from types import MappingProxyType
from typing import Any from typing import Any
from aionut import NUTError, NUTLoginError from aionut import NUTError, NUTLoginError
@ -27,16 +28,26 @@ from .const import DEFAULT_HOST, DEFAULT_PORT, DOMAIN
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
AUTH_SCHEMA = {vol.Optional(CONF_USERNAME): str, vol.Optional(CONF_PASSWORD): str} REAUTH_SCHEMA = {vol.Optional(CONF_USERNAME): str, vol.Optional(CONF_PASSWORD): str}
PASSWORD_NOT_CHANGED = "__**password_not_changed**__"
def _base_schema(nut_config: dict[str, Any]) -> vol.Schema: def _base_schema(
nut_config: dict[str, Any] | MappingProxyType[str, Any],
use_password_not_changed: bool = False,
) -> vol.Schema:
"""Generate base schema.""" """Generate base schema."""
base_schema = { base_schema = {
vol.Optional(CONF_HOST, default=nut_config.get(CONF_HOST) or DEFAULT_HOST): str, vol.Optional(CONF_HOST, default=nut_config.get(CONF_HOST) or DEFAULT_HOST): str,
vol.Optional(CONF_PORT, default=nut_config.get(CONF_PORT) or DEFAULT_PORT): int, vol.Optional(CONF_PORT, default=nut_config.get(CONF_PORT) or DEFAULT_PORT): int,
vol.Optional(CONF_USERNAME, default=nut_config.get(CONF_USERNAME) or ""): str,
vol.Optional(
CONF_PASSWORD,
default=PASSWORD_NOT_CHANGED if use_password_not_changed else "",
): str,
} }
base_schema.update(AUTH_SCHEMA)
return vol.Schema(base_schema) return vol.Schema(base_schema)
@ -66,6 +77,26 @@ async def validate_input(hass: HomeAssistant, data: dict[str, Any]) -> dict[str,
return {"ups_list": nut_data.ups_list, "available_resources": status} return {"ups_list": nut_data.ups_list, "available_resources": status}
def _check_host_port_alias_match(
first: Mapping[str, Any], second: Mapping[str, Any]
) -> bool:
"""Check if first and second have the same host, port and alias."""
if first[CONF_HOST] != second[CONF_HOST] or first[CONF_PORT] != second[CONF_PORT]:
return False
first_alias = first.get(CONF_ALIAS)
second_alias = second.get(CONF_ALIAS)
if (first_alias is None and second_alias is None) or (
first_alias is not None
and second_alias is not None
and first_alias == second_alias
):
return True
return False
def _format_host_port_alias(user_input: Mapping[str, Any]) -> str: def _format_host_port_alias(user_input: Mapping[str, Any]) -> str:
"""Format a host, port, and alias so it can be used for comparison or display.""" """Format a host, port, and alias so it can be used for comparison or display."""
host = user_input[CONF_HOST] host = user_input[CONF_HOST]
@ -137,7 +168,7 @@ class NutConfigFlow(ConfigFlow, domain=DOMAIN):
async def async_step_ups( async def async_step_ups(
self, user_input: dict[str, Any] | None = None self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult: ) -> ConfigFlowResult:
"""Handle the picking the ups.""" """Handle selecting the NUT device alias."""
errors: dict[str, str] = {} errors: dict[str, str] = {}
placeholders: dict[str, str] = {} placeholders: dict[str, str] = {}
nut_config = self.nut_config nut_config = self.nut_config
@ -163,6 +194,99 @@ class NutConfigFlow(ConfigFlow, domain=DOMAIN):
description_placeholders=placeholders, description_placeholders=placeholders,
) )
async def async_step_reconfigure(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle reconfiguration of the integration."""
errors: dict[str, str] = {}
placeholders: dict[str, str] = {}
reconfigure_entry = self._get_reconfigure_entry()
nut_config = self.nut_config
if user_input is not None:
nut_config.update(user_input)
info, errors, placeholders = await self._async_validate_or_error(nut_config)
if not errors:
if len(info["ups_list"]) > 1:
self.ups_list = info["ups_list"]
return await self.async_step_reconfigure_ups()
if not _check_host_port_alias_match(
reconfigure_entry.data,
nut_config,
) and (self._host_port_alias_already_configured(nut_config)):
return self.async_abort(reason="already_configured")
if unique_id := _unique_id_from_status(info["available_resources"]):
await self.async_set_unique_id(unique_id)
self._abort_if_unique_id_mismatch(reason="unique_id_mismatch")
if nut_config[CONF_PASSWORD] == PASSWORD_NOT_CHANGED:
nut_config.pop(CONF_PASSWORD)
new_title = _format_host_port_alias(nut_config)
return self.async_update_reload_and_abort(
self._get_reconfigure_entry(),
unique_id=unique_id,
title=new_title,
data_updates=nut_config,
)
return self.async_show_form(
step_id="reconfigure",
data_schema=_base_schema(
reconfigure_entry.data,
use_password_not_changed=True,
),
errors=errors,
description_placeholders=placeholders,
)
async def async_step_reconfigure_ups(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle selecting the NUT device alias."""
errors: dict[str, str] = {}
placeholders: dict[str, str] = {}
reconfigure_entry = self._get_reconfigure_entry()
nut_config = self.nut_config
if user_input is not None:
self.nut_config.update(user_input)
if not _check_host_port_alias_match(
reconfigure_entry.data,
nut_config,
) and (self._host_port_alias_already_configured(nut_config)):
return self.async_abort(reason="already_configured")
info, errors, placeholders = await self._async_validate_or_error(nut_config)
if not errors:
if unique_id := _unique_id_from_status(info["available_resources"]):
await self.async_set_unique_id(unique_id)
self._abort_if_unique_id_mismatch(reason="unique_id_mismatch")
if nut_config[CONF_PASSWORD] == PASSWORD_NOT_CHANGED:
nut_config.pop(CONF_PASSWORD)
new_title = _format_host_port_alias(nut_config)
return self.async_update_reload_and_abort(
self._get_reconfigure_entry(),
unique_id=unique_id,
title=new_title,
data_updates=nut_config,
)
return self.async_show_form(
step_id="reconfigure_ups",
data_schema=_ups_schema(self.ups_list or {}),
errors=errors,
description_placeholders=placeholders,
)
def _host_port_alias_already_configured(self, user_input: dict[str, Any]) -> bool: def _host_port_alias_already_configured(self, user_input: dict[str, Any]) -> bool:
"""See if we already have a nut entry matching user input configured.""" """See if we already have a nut entry matching user input configured."""
existing_host_port_aliases = { existing_host_port_aliases = {
@ -204,6 +328,7 @@ class NutConfigFlow(ConfigFlow, domain=DOMAIN):
self, user_input: dict[str, Any] | None = None self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult: ) -> ConfigFlowResult:
"""Handle reauth input.""" """Handle reauth input."""
errors: dict[str, str] = {} errors: dict[str, str] = {}
existing_entry = self.reauth_entry existing_entry = self.reauth_entry
assert existing_entry assert existing_entry
@ -212,6 +337,7 @@ class NutConfigFlow(ConfigFlow, domain=DOMAIN):
CONF_HOST: existing_data[CONF_HOST], CONF_HOST: existing_data[CONF_HOST],
CONF_PORT: existing_data[CONF_PORT], CONF_PORT: existing_data[CONF_PORT],
} }
if user_input is not None: if user_input is not None:
new_config = { new_config = {
**existing_data, **existing_data,
@ -229,8 +355,8 @@ class NutConfigFlow(ConfigFlow, domain=DOMAIN):
description_placeholders.update(placeholders) description_placeholders.update(placeholders)
return self.async_show_form( return self.async_show_form(
description_placeholders=description_placeholders,
step_id="reauth_confirm", step_id="reauth_confirm",
data_schema=vol.Schema(AUTH_SCHEMA), data_schema=vol.Schema(REAUTH_SCHEMA),
errors=errors, errors=errors,
description_placeholders=description_placeholders,
) )

View File

@ -28,6 +28,27 @@
"username": "[%key:common::config_flow::data::username%]", "username": "[%key:common::config_flow::data::username%]",
"password": "[%key:common::config_flow::data::password%]" "password": "[%key:common::config_flow::data::password%]"
} }
},
"reconfigure": {
"description": "[%key:component::nut::config::step::user::description%]",
"data": {
"host": "[%key:common::config_flow::data::host%]",
"port": "[%key:common::config_flow::data::port%]",
"username": "[%key:common::config_flow::data::username%]",
"password": "[%key:common::config_flow::data::password%]"
},
"data_description": {
"host": "[%key:component::nut::config::step::user::data_description::host%]",
"port": "[%key:component::nut::config::step::user::data_description::port%]",
"username": "[%key:component::nut::config::step::user::data_description::username%]",
"password": "[%key:component::nut::config::step::user::data_description::password%]"
}
},
"reconfigure_ups": {
"title": "[%key:component::nut::config::step::ups::title%]",
"data": {
"alias": "[%key:component::nut::config::step::ups::data::alias%]"
}
} }
}, },
"error": { "error": {
@ -38,7 +59,9 @@
"abort": { "abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]", "already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"no_ups_found": "There are no UPS devices available on the NUT server.", "no_ups_found": "There are no UPS devices available on the NUT server.",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]" "reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]",
"reconfigure_successful": "[%key:common::config_flow::abort::reconfigure_successful%]",
"unique_id_mismatch": "The device's manufacturer, model and serial number identifier does not match the previous identifier."
} }
}, },
"device_automation": { "device_automation": {

View File

@ -6,6 +6,7 @@ from unittest.mock import patch
from aionut import NUTError, NUTLoginError from aionut import NUTError, NUTLoginError
from homeassistant import config_entries, setup from homeassistant import config_entries, setup
from homeassistant.components.nut.config_flow import PASSWORD_NOT_CHANGED
from homeassistant.components.nut.const import DOMAIN from homeassistant.components.nut.const import DOMAIN
from homeassistant.const import ( from homeassistant.const import (
CONF_ALIAS, CONF_ALIAS,
@ -83,8 +84,8 @@ async def test_form_zeroconf(hass: HomeAssistant) -> None:
assert len(mock_setup_entry.mock_calls) == 1 assert len(mock_setup_entry.mock_calls) == 1
async def test_form_user_one_ups(hass: HomeAssistant) -> None: async def test_form_user_one_alias(hass: HomeAssistant) -> None:
"""Test we get the form.""" """Test we can configure a device with one alias."""
await setup.async_setup_component(hass, "persistent_notification", {}) await setup.async_setup_component(hass, "persistent_notification", {})
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER} DOMAIN, context={"source": config_entries.SOURCE_USER}
@ -128,8 +129,8 @@ async def test_form_user_one_ups(hass: HomeAssistant) -> None:
assert len(mock_setup_entry.mock_calls) == 1 assert len(mock_setup_entry.mock_calls) == 1
async def test_form_user_multiple_ups(hass: HomeAssistant) -> None: async def test_form_user_multiple_aliases(hass: HomeAssistant) -> None:
"""Test we get the form.""" """Test we can configure device with multiple aliases."""
await setup.async_setup_component(hass, "persistent_notification", {}) await setup.async_setup_component(hass, "persistent_notification", {})
config_entry = MockConfigEntry( config_entry = MockConfigEntry(
@ -194,7 +195,7 @@ async def test_form_user_multiple_ups(hass: HomeAssistant) -> None:
assert len(mock_setup_entry.mock_calls) == 2 assert len(mock_setup_entry.mock_calls) == 2
async def test_form_user_one_ups_with_ignored_entry(hass: HomeAssistant) -> None: async def test_form_user_one_alias_with_ignored_entry(hass: HomeAssistant) -> None:
"""Test we can setup a new one when there is an ignored one.""" """Test we can setup a new one when there is an ignored one."""
ignored_entry = MockConfigEntry( ignored_entry = MockConfigEntry(
domain=DOMAIN, data={}, source=config_entries.SOURCE_IGNORE domain=DOMAIN, data={}, source=config_entries.SOURCE_IGNORE
@ -244,8 +245,8 @@ async def test_form_user_one_ups_with_ignored_entry(hass: HomeAssistant) -> None
assert len(mock_setup_entry.mock_calls) == 1 assert len(mock_setup_entry.mock_calls) == 1
async def test_form_no_upses_found(hass: HomeAssistant) -> None: async def test_form_no_aliases_found(hass: HomeAssistant) -> None:
"""Test we abort when the NUT server has not UPSes.""" """Test we abort when the NUT server has no aliases."""
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER} DOMAIN, context={"source": config_entries.SOURCE_USER}
) )
@ -561,8 +562,8 @@ async def test_abort_duplicate_unique_ids(hass: HomeAssistant) -> None:
assert result2["reason"] == "already_configured" assert result2["reason"] == "already_configured"
async def test_abort_multiple_ups_duplicate_unique_ids(hass: HomeAssistant) -> None: async def test_abort_multiple_aliases_duplicate_unique_ids(hass: HomeAssistant) -> None:
"""Test we abort on multiple devices if unique_id is already setup.""" """Test we abort on multiple aliases if unique_id is already setup."""
list_vars = { list_vars = {
"device.mfr": "Some manufacturer", "device.mfr": "Some manufacturer",
@ -670,3 +671,762 @@ async def test_abort_if_already_setup_alias(hass: HomeAssistant) -> None:
assert result3["type"] is FlowResultType.ABORT assert result3["type"] is FlowResultType.ABORT
assert result3["reason"] == "already_configured" assert result3["reason"] == "already_configured"
async def test_reconfigure_one_alias_successful(hass: HomeAssistant) -> None:
"""Test reconfigure one alias successful."""
entry = await async_init_integration(
hass,
host="1.1.1.1",
port=123,
username="test-username",
password="test-password",
list_ups={"ups1": "UPS 1"},
list_vars={"battery.voltage": "voltage"},
)
result = await entry.start_reconfigure_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reconfigure"
mock_pynut = _get_mock_nutclient(
list_vars={"battery.voltage": "voltage"},
list_ups={"ups1": "UPS 1"},
)
with patch(
"homeassistant.components.nut.AIONUTClient",
return_value=mock_pynut,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_HOST: "2.2.2.2",
CONF_PORT: 456,
CONF_USERNAME: "test-new-username",
CONF_PASSWORD: "test-new-password",
},
)
assert result2["type"] is FlowResultType.ABORT
assert result2["reason"] == "reconfigure_successful"
assert entry.data[CONF_HOST] == "2.2.2.2"
assert entry.data[CONF_PORT] == 456
assert entry.data[CONF_USERNAME] == "test-new-username"
assert entry.data[CONF_PASSWORD] == "test-new-password"
async def test_reconfigure_one_alias_nochange(hass: HomeAssistant) -> None:
"""Test reconfigure one alias when there is no change."""
entry = await async_init_integration(
hass,
host="1.1.1.1",
port=123,
username="test-username",
password="test-password",
list_ups={"ups1": "UPS 1"},
list_vars={"battery.voltage": "voltage"},
)
result = await entry.start_reconfigure_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reconfigure"
mock_pynut = _get_mock_nutclient(
list_ups={"ups1": "UPS 1"},
list_vars={"battery.voltage": "voltage"},
)
with patch(
"homeassistant.components.nut.AIONUTClient",
return_value=mock_pynut,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_HOST: entry.data[CONF_HOST],
CONF_PORT: int(entry.data[CONF_PORT]),
CONF_USERNAME: entry.data[CONF_USERNAME],
CONF_PASSWORD: entry.data[CONF_PASSWORD],
},
)
assert result2["type"] is FlowResultType.ABORT
assert result2["reason"] == "reconfigure_successful"
assert entry.data[CONF_HOST] == "1.1.1.1"
assert entry.data[CONF_PORT] == 123
assert entry.data[CONF_USERNAME] == "test-username"
assert entry.data[CONF_PASSWORD] == "test-password"
async def test_reconfigure_one_alias_password_nochange(hass: HomeAssistant) -> None:
"""Test reconfigure one alias when there is no password change."""
entry = await async_init_integration(
hass,
host="1.1.1.1",
port=123,
username="test-username",
password="test-password",
list_ups={"ups1": "UPS 1"},
list_vars={"battery.voltage": "voltage"},
)
result = await entry.start_reconfigure_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reconfigure"
mock_pynut = _get_mock_nutclient(
list_vars={"battery.voltage": "voltage"},
list_ups={"ups1": "UPS 1"},
)
with patch(
"homeassistant.components.nut.AIONUTClient",
return_value=mock_pynut,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_HOST: "2.2.2.2",
CONF_PORT: 456,
CONF_USERNAME: "test-new-username",
CONF_PASSWORD: PASSWORD_NOT_CHANGED,
},
)
assert result2["type"] is FlowResultType.ABORT
assert result2["reason"] == "reconfigure_successful"
assert entry.data[CONF_HOST] == "2.2.2.2"
assert entry.data[CONF_PORT] == 456
assert entry.data[CONF_USERNAME] == "test-new-username"
assert entry.data[CONF_PASSWORD] == "test-password"
async def test_reconfigure_one_alias_already_configured(hass: HomeAssistant) -> None:
"""Test reconfigure when config changed to an existing host/port/alias."""
entry = await async_init_integration(
hass,
host="1.1.1.1",
port=123,
username="test-username",
password="test-password",
list_ups={"ups1": "UPS 1"},
list_vars={"battery.voltage": "voltage"},
)
entry2 = await async_init_integration(
hass,
host="2.2.2.2",
port=456,
username="test-username",
password="test-password",
list_ups={"ups1": "UPS 1"},
list_vars={"battery.voltage": "voltage"},
)
result = await entry2.start_reconfigure_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reconfigure"
mock_pynut = _get_mock_nutclient(
list_ups={"ups1": "UPS 1"},
list_vars={"battery.voltage": "voltage"},
)
with patch(
"homeassistant.components.nut.AIONUTClient",
return_value=mock_pynut,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_HOST: entry.data[CONF_HOST],
CONF_PORT: int(entry.data[CONF_PORT]),
CONF_USERNAME: entry.data[CONF_USERNAME],
CONF_PASSWORD: entry.data[CONF_PASSWORD],
},
)
assert result2["type"] is FlowResultType.ABORT
assert result2["reason"] == "already_configured"
assert entry.data[CONF_HOST] == "1.1.1.1"
assert entry.data[CONF_PORT] == 123
assert entry.data[CONF_USERNAME] == "test-username"
assert entry.data[CONF_PASSWORD] == "test-password"
assert entry2.data[CONF_HOST] == "2.2.2.2"
assert entry2.data[CONF_PORT] == 456
assert entry2.data[CONF_USERNAME] == "test-username"
assert entry2.data[CONF_PASSWORD] == "test-password"
async def test_reconfigure_one_alias_unique_id_change(hass: HomeAssistant) -> None:
"""Test reconfigure when the unique ID is changed."""
entry = await async_init_integration(
hass,
host="1.1.1.1",
port=123,
username="test-username",
password="test-password",
list_ups={"ups1": "UPS 1"},
list_vars={
"device.mfr": "Some manufacturer",
"device.model": "Some model",
"device.serial": "0000-1",
},
)
result = await entry.start_reconfigure_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reconfigure"
mock_pynut = _get_mock_nutclient(
list_ups={"ups1": "UPS 1"},
list_vars={
"device.mfr": "Another manufacturer",
"device.model": "Another model",
"device.serial": "0000-2",
},
)
with patch(
"homeassistant.components.nut.AIONUTClient",
return_value=mock_pynut,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_HOST: entry.data[CONF_HOST],
CONF_PORT: entry.data[CONF_PORT],
CONF_USERNAME: entry.data[CONF_USERNAME],
CONF_PASSWORD: entry.data[CONF_PASSWORD],
},
)
assert result2["type"] is FlowResultType.ABORT
assert result2["reason"] == "unique_id_mismatch"
async def test_reconfigure_one_alias_duplicate_unique_ids(hass: HomeAssistant) -> None:
"""Test reconfigure that results in a duplicate unique ID."""
list_vars = {
"device.mfr": "Some manufacturer",
"device.model": "Some model",
"device.serial": "0000-1",
}
await async_init_integration(
hass,
host="1.1.1.1",
port=123,
username="test-username",
password="test-password",
list_ups={"ups1": "UPS 1"},
list_vars=list_vars,
)
entry2 = await async_init_integration(
hass,
host="2.2.2.2",
port=456,
username="test-username",
password="test-password",
list_ups={"ups2": "UPS 2"},
list_vars={
"device.mfr": "Another manufacturer",
"device.model": "Another model",
"device.serial": "0000-2",
},
)
result = await entry2.start_reconfigure_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reconfigure"
mock_pynut = _get_mock_nutclient(
list_ups={"ups2": "UPS 2"},
list_vars=list_vars,
)
with patch(
"homeassistant.components.nut.AIONUTClient",
return_value=mock_pynut,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_HOST: "3.3.3.3",
CONF_PORT: 789,
CONF_USERNAME: "test-new-username",
CONF_PASSWORD: "test-new-password",
},
)
assert result2["type"] is FlowResultType.ABORT
assert result2["reason"] == "unique_id_mismatch"
async def test_reconfigure_multiple_aliases_successful(hass: HomeAssistant) -> None:
"""Test reconfigure with multiple aliases is successful."""
entry = await async_init_integration(
hass,
host="1.1.1.1",
port=123,
username="test-username",
password="test-password",
list_ups={"ups1": "UPS 1"},
list_vars={"battery.voltage": "voltage"},
)
result = await entry.start_reconfigure_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reconfigure"
mock_pynut = _get_mock_nutclient(
list_ups={
"ups1": "UPS 1",
"ups2": "UPS 2",
},
list_vars={"battery.voltage": "voltage"},
)
with patch(
"homeassistant.components.nut.AIONUTClient",
return_value=mock_pynut,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_HOST: "2.2.2.2",
CONF_PORT: 456,
CONF_USERNAME: "test-new-username",
CONF_PASSWORD: "test-new-password",
},
)
assert result2["type"] is FlowResultType.FORM
assert result2["step_id"] == "reconfigure_ups"
with (
patch(
"homeassistant.components.nut.AIONUTClient",
return_value=mock_pynut,
),
patch(
"homeassistant.components.nut.async_setup_entry",
return_value=True,
),
):
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
{CONF_ALIAS: "ups2"},
)
await hass.async_block_till_done()
assert result3["type"] is FlowResultType.ABORT
assert result3["reason"] == "reconfigure_successful"
assert entry.data[CONF_HOST] == "2.2.2.2"
assert entry.data[CONF_PORT] == 456
assert entry.data[CONF_USERNAME] == "test-new-username"
assert entry.data[CONF_PASSWORD] == "test-new-password"
assert entry.data[CONF_ALIAS] == "ups2"
async def test_reconfigure_multiple_aliases_nochange(hass: HomeAssistant) -> None:
"""Test reconfigure with multiple aliases and no change."""
entry = await async_init_integration(
hass,
host="1.1.1.1",
port=123,
username="test-username",
password="test-password",
list_ups={"ups1": "UPS 1"},
list_vars={"battery.voltage": "voltage"},
)
result = await entry.start_reconfigure_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reconfigure"
mock_pynut = _get_mock_nutclient(
list_ups={
"ups1": "UPS 1",
"ups2": "UPS 2",
},
list_vars={"battery.voltage": "voltage"},
)
with patch(
"homeassistant.components.nut.AIONUTClient",
return_value=mock_pynut,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_HOST: entry.data[CONF_HOST],
CONF_PORT: entry.data[CONF_PORT],
CONF_USERNAME: entry.data[CONF_USERNAME],
CONF_PASSWORD: entry.data[CONF_PASSWORD],
},
)
assert result2["type"] is FlowResultType.FORM
assert result2["step_id"] == "reconfigure_ups"
with (
patch(
"homeassistant.components.nut.AIONUTClient",
return_value=mock_pynut,
),
patch(
"homeassistant.components.nut.async_setup_entry",
return_value=True,
),
):
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
{CONF_ALIAS: "ups1"},
)
await hass.async_block_till_done()
assert result3["type"] is FlowResultType.ABORT
assert result3["reason"] == "reconfigure_successful"
assert entry.data[CONF_HOST] == "1.1.1.1"
assert entry.data[CONF_PORT] == 123
assert entry.data[CONF_USERNAME] == "test-username"
assert entry.data[CONF_PASSWORD] == "test-password"
assert entry.data[CONF_ALIAS] == "ups1"
async def test_reconfigure_multiple_aliases_password_nochange(
hass: HomeAssistant,
) -> None:
"""Test reconfigure with multiple aliases when no password change."""
entry = await async_init_integration(
hass,
host="1.1.1.1",
port=123,
username="test-username",
password="test-password",
list_ups={"ups1": "UPS 1"},
list_vars={"battery.voltage": "voltage"},
)
result = await entry.start_reconfigure_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reconfigure"
mock_pynut = _get_mock_nutclient(
list_ups={
"ups1": "UPS 1",
"ups2": "UPS 2",
},
list_vars={"battery.voltage": "voltage"},
)
with patch(
"homeassistant.components.nut.AIONUTClient",
return_value=mock_pynut,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_HOST: "2.2.2.2",
CONF_PORT: 456,
CONF_USERNAME: "test-new-username",
CONF_PASSWORD: PASSWORD_NOT_CHANGED,
},
)
assert result2["type"] is FlowResultType.FORM
assert result2["step_id"] == "reconfigure_ups"
with (
patch(
"homeassistant.components.nut.AIONUTClient",
return_value=mock_pynut,
),
patch(
"homeassistant.components.nut.async_setup_entry",
return_value=True,
),
):
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
{CONF_ALIAS: "ups2"},
)
await hass.async_block_till_done()
assert result3["type"] is FlowResultType.ABORT
assert result3["reason"] == "reconfigure_successful"
assert entry.data[CONF_HOST] == "2.2.2.2"
assert entry.data[CONF_PORT] == 456
assert entry.data[CONF_USERNAME] == "test-new-username"
assert entry.data[CONF_PASSWORD] == "test-password"
assert entry.data[CONF_ALIAS] == "ups2"
async def test_reconfigure_multiple_aliases_already_configured(
hass: HomeAssistant,
) -> None:
"""Test reconfigure multi aliases changed to existing host/port/alias."""
entry = await async_init_integration(
hass,
host="1.1.1.1",
port=123,
alias="ups1",
username="test-username",
password="test-password",
list_ups={"ups1": "UPS 1", "ups2": "UPS 2"},
list_vars={"battery.voltage": "voltage"},
)
entry2 = await async_init_integration(
hass,
host="2.2.2.2",
port=456,
alias="ups2",
username="test-username",
password="test-password",
list_ups={"ups1": "UPS 1"},
list_vars={"battery.voltage": "voltage"},
)
assert entry2.data[CONF_HOST] == "2.2.2.2"
assert entry2.data[CONF_PORT] == 456
assert entry2.data[CONF_USERNAME] == "test-username"
assert entry2.data[CONF_PASSWORD] == "test-password"
result = await entry2.start_reconfigure_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reconfigure"
mock_pynut = _get_mock_nutclient(
list_ups={
"ups1": "UPS 1",
"ups2": "UPS 2",
},
list_vars={"battery.voltage": "voltage"},
)
with patch(
"homeassistant.components.nut.AIONUTClient",
return_value=mock_pynut,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_HOST: entry.data[CONF_HOST],
CONF_PORT: entry.data[CONF_PORT],
CONF_USERNAME: entry.data[CONF_USERNAME],
CONF_PASSWORD: entry.data[CONF_PASSWORD],
},
)
assert result2["type"] is FlowResultType.FORM
assert result2["step_id"] == "reconfigure_ups"
with (
patch(
"homeassistant.components.nut.AIONUTClient",
return_value=mock_pynut,
),
patch(
"homeassistant.components.nut.async_setup_entry",
return_value=True,
),
):
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
{CONF_ALIAS: entry.data[CONF_ALIAS]},
)
await hass.async_block_till_done()
assert result3["type"] is FlowResultType.ABORT
assert result3["reason"] == "already_configured"
assert entry.data[CONF_HOST] == "1.1.1.1"
assert entry.data[CONF_PORT] == 123
assert entry.data[CONF_USERNAME] == "test-username"
assert entry.data[CONF_PASSWORD] == "test-password"
assert entry.data[CONF_ALIAS] == "ups1"
assert entry2.data[CONF_HOST] == "2.2.2.2"
assert entry2.data[CONF_PORT] == 456
assert entry2.data[CONF_USERNAME] == "test-username"
assert entry2.data[CONF_PASSWORD] == "test-password"
assert entry2.data[CONF_ALIAS] == "ups2"
async def test_reconfigure_multiple_aliases_unique_id_change(
hass: HomeAssistant,
) -> None:
"""Test reconfigure with multiple aliases and the unique ID is changed."""
entry = await async_init_integration(
hass,
host="1.1.1.1",
port=123,
alias="ups1",
username="test-username",
password="test-password",
list_ups={"ups1": "UPS 1", "ups2": "UPS 2"},
list_vars={"battery.voltage": "voltage"},
)
result = await entry.start_reconfigure_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reconfigure"
mock_pynut = _get_mock_nutclient(
list_ups={
"ups1": "UPS 1",
"ups2": "UPS 2",
},
list_vars={
"device.mfr": "Another manufacturer",
"device.model": "Another model",
"device.serial": "0000-2",
},
)
with patch(
"homeassistant.components.nut.AIONUTClient",
return_value=mock_pynut,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_HOST: entry.data[CONF_HOST],
CONF_PORT: entry.data[CONF_PORT],
CONF_USERNAME: entry.data[CONF_USERNAME],
CONF_PASSWORD: entry.data[CONF_PASSWORD],
},
)
assert result2["type"] is FlowResultType.FORM
assert result2["step_id"] == "reconfigure_ups"
with (
patch(
"homeassistant.components.nut.AIONUTClient",
return_value=mock_pynut,
),
patch(
"homeassistant.components.nut.async_setup_entry",
return_value=True,
),
):
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
{CONF_ALIAS: entry.data[CONF_ALIAS]},
)
await hass.async_block_till_done()
assert result3["type"] is FlowResultType.ABORT
assert result3["reason"] == "unique_id_mismatch"
async def test_reconfigure_multiple_aliases_duplicate_unique_ids(
hass: HomeAssistant,
) -> None:
"""Test reconfigure multi aliases that results in duplicate unique ID."""
list_vars = {
"device.mfr": "Some manufacturer",
"device.model": "Some model",
"device.serial": "0000-1",
}
entry = await async_init_integration(
hass,
host="1.1.1.1",
port=123,
alias="ups1",
username="test-username",
password="test-password",
list_ups={"ups1": "UPS 1", "ups2": "UPS 2"},
list_vars=list_vars,
)
entry2 = await async_init_integration(
hass,
host="2.2.2.2",
port=456,
alias="ups2",
username="test-username",
password="test-password",
list_ups={"ups1": "UPS 1"},
list_vars={
"device.mfr": "Another manufacturer",
"device.model": "Another model",
"device.serial": "0000-2",
},
)
result = await entry2.start_reconfigure_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reconfigure"
mock_pynut = _get_mock_nutclient(
list_ups={
"ups1": "UPS 1",
"ups2": "UPS 2",
},
list_vars=list_vars,
)
with patch(
"homeassistant.components.nut.AIONUTClient",
return_value=mock_pynut,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_HOST: "3.3.3.3",
CONF_PORT: 789,
CONF_USERNAME: "test-new-username",
CONF_PASSWORD: "test-new-password",
},
)
assert result2["type"] is FlowResultType.FORM
assert result2["step_id"] == "reconfigure_ups"
with (
patch(
"homeassistant.components.nut.AIONUTClient",
return_value=mock_pynut,
),
patch(
"homeassistant.components.nut.async_setup_entry",
return_value=True,
),
):
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
{CONF_ALIAS: entry.data[CONF_ALIAS]},
)
await hass.async_block_till_done()
assert result3["type"] is FlowResultType.ABORT
assert result3["reason"] == "unique_id_mismatch"

View File

@ -1,10 +1,17 @@
"""Tests for the nut integration.""" """Tests for the nut integration."""
import json import json
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch from unittest.mock import AsyncMock, MagicMock, patch
from homeassistant.components.nut.const import DOMAIN from homeassistant.components.nut.const import DOMAIN
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_PORT, CONF_USERNAME from homeassistant.const import (
CONF_ALIAS,
CONF_HOST,
CONF_PASSWORD,
CONF_PORT,
CONF_USERNAME,
)
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er from homeassistant.helpers import entity_registry as er
@ -35,8 +42,11 @@ def _get_mock_nutclient(
async def async_init_integration( async def async_init_integration(
hass: HomeAssistant, hass: HomeAssistant,
ups_fixture: str | None = None, ups_fixture: str | None = None,
host: str = "mock",
port: str = "mock",
username: str = "mock", username: str = "mock",
password: str = "mock", password: str = "mock",
alias: str | None = None,
list_ups: dict[str, str] | None = None, list_ups: dict[str, str] | None = None,
list_vars: dict[str, str] | None = None, list_vars: dict[str, str] | None = None,
list_commands_return_value: dict[str, str] | None = None, list_commands_return_value: dict[str, str] | None = None,
@ -65,15 +75,24 @@ async def async_init_integration(
"homeassistant.components.nut.AIONUTClient", "homeassistant.components.nut.AIONUTClient",
return_value=mock_pynut, return_value=mock_pynut,
): ):
extra_config_entry_data: dict[str, Any] = {}
if alias is not None:
extra_config_entry_data = {
CONF_ALIAS: alias,
}
entry = MockConfigEntry( entry = MockConfigEntry(
domain=DOMAIN, domain=DOMAIN,
data={ data={
CONF_HOST: "mock", CONF_HOST: host,
CONF_PASSWORD: password, CONF_PASSWORD: password,
CONF_PORT: "mock", CONF_PORT: port,
CONF_USERNAME: username, CONF_USERNAME: username,
}, }
| extra_config_entry_data,
) )
entry.add_to_hass(hass) entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id) await hass.config_entries.async_setup(entry.entry_id)