starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-09-26 09:23:17 -04:00
Code Issues Packages Projects Releases Wiki Activity

Unifi/reconfiguration (#182201)

Browse Source 
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
Robin Siep
2026-09-18 10:38:38 +02:00
committed by GitHub
co-authored by Copilot Autofix powered by AI
parent f5f0f8ce4f
commit 73f79aa5a2
4 changed files with 266 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/unifi/config_flow.py
+124 -83
View File
@@ -6,7 +6,8 @@ Reauthentication when issue with credentials are reported.
Configuration of options through options flow.
"""
from collections.abc import Mapping
from collections.abc import Iterator, Mapping
from contextlib import contextmanager
import operator
import socket
from types import MappingProxyType
@@ -17,7 +18,7 @@ import probatio
from homeassistant.config_entries import (
SOURCE_REAUTH,
ConfigEntryState,
ConfigEntry,
ConfigFlow,
ConfigFlowResult,
OptionsFlow,
@@ -58,6 +59,7 @@ from .const import (
from .errors import AuthenticationRequired, CannotConnect
from .hub import UnifiHub, get_unifi_api
DEFAULT_HOST = "unifi"
DEFAULT_PORT = 443
DEFAULT_SITE_ID = "default"
DEFAULT_VERIFY_SSL = False
@@ -89,63 +91,34 @@ class UnifiFlowHandler(ConfigFlow, domain=DOMAIN):
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle a flow initialized by the user."""
errors = {}
errors: dict[str, str] = {}
if user_input is not None:
self.config = {
CONF_HOST: user_input[CONF_HOST],
CONF_USERNAME: user_input[CONF_USERNAME],
CONF_PASSWORD: user_input[CONF_PASSWORD],
CONF_PORT: user_input.get(CONF_PORT),
CONF_VERIFY_SSL: user_input.get(CONF_VERIFY_SSL),
CONF_SITE_ID: DEFAULT_SITE_ID,
}
try:
hub = await get_unifi_api(self.hass, MappingProxyType(self.config))
await hub.sites.update()
self.sites = hub.sites
except AuthenticationRequired:
errors["base"] = "faulty_credentials"
except CannotConnect:
errors["base"] = "service_unavailable"
else:
if self.source == SOURCE_REAUTH:
if (
(reauth_unique_id := self._get_reauth_entry().unique_id)
is not None
) and reauth_unique_id in self.sites:
return await self.async_step_site(
{CONF_SITE_ID: reauth_unique_id}
)
raise AbortFlow("unknown_site_id")
self.config = _config_from_input(user_input)
data_schema = self._build_form_schema(
self.config[CONF_HOST],
self.config[CONF_USERNAME],
self.config[CONF_PORT],
self.config[CONF_VERIFY_SSL],
)
with _catch_unifi_api_flow_errors(errors):
self.sites = await self._async_update_sites(self.config)
return await self.async_step_site()
if not (host := self.config.get(CONF_HOST, "")) and await _async_discover_unifi(
self.hass
):
host = "unifi"
data = self.reauth_schema or {
probatio.Required(CONF_HOST, default=host): str,
probatio.Required(CONF_USERNAME): str,
probatio.Required(CONF_PASSWORD): str,
probatio.Optional(
CONF_PORT, default=self.config.get(CONF_PORT, DEFAULT_PORT)
): int,
probatio.Optional(
CONF_VERIFY_SSL,
default=self.config.get(CONF_VERIFY_SSL, DEFAULT_VERIFY_SSL),
): bool,
}
else:
host = self.config.get(CONF_HOST)
if not host:
host = await _async_discover_unifi(self.hass)
if not host:
host = DEFAULT_HOST
data_schema = self._build_form_schema(
host=host,
verify_ssl=self.config.get(CONF_VERIFY_SSL, DEFAULT_VERIFY_SSL),
)
return self.async_show_form(
step_id="user",
data_schema=probatio.Schema(data),
data_schema=data_schema,
errors=errors,
)
@@ -157,24 +130,8 @@ class UnifiFlowHandler(ConfigFlow, domain=DOMAIN):
unique_id = user_input[CONF_SITE_ID]
self.config[CONF_SITE_ID] = self.sites[unique_id].name
config_entry = await self.async_set_unique_id(unique_id)
abort_reason = "configuration_updated"
if self.source == SOURCE_REAUTH:
config_entry = self._get_reauth_entry()
abort_reason = "reauth_successful"
if config_entry:
if (
config_entry.state is ConfigEntryState.LOADED
and (hub := config_entry.runtime_data)
and hub.available
):
return self.async_abort(reason="already_configured")
return self.async_update_and_abort(
config_entry, data=self.config, reason=abort_reason
)
await self.async_set_unique_id(unique_id)
self._abort_if_unique_id_configured()
site_nice_name = self.sites[unique_id].description
return self.async_create_entry(title=site_nice_name, data=self.config)
@@ -195,25 +152,53 @@ class UnifiFlowHandler(ConfigFlow, domain=DOMAIN):
) -> ConfigFlowResult:
"""Trigger a reauthentication flow."""
reauth_entry = self._get_reauth_entry()
self.context["title_placeholders"] = {
CONF_HOST: reauth_entry.data[CONF_HOST],
CONF_NAME: reauth_entry.title,
}
self.reauth_schema = {
probatio.Required(CONF_HOST, default=reauth_entry.data[CONF_HOST]): str,
probatio.Required(
CONF_USERNAME, default=reauth_entry.data[CONF_USERNAME]
): str,
probatio.Required(CONF_PASSWORD): str,
probatio.Required(CONF_PORT, default=reauth_entry.data[CONF_PORT]): int,
probatio.Required(
CONF_VERIFY_SSL, default=reauth_entry.data[CONF_VERIFY_SSL]
): bool,
}
return await self.async_step_reconfigure()
return await self.async_step_user()
async def async_step_reconfigure(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle a reconfiguration flow."""
config_entry = self._get_reauth_or_reconfigure_entry()
errors: dict[str, str] = {}
if user_input is not None:
config_data = _config_from_input(user_input)
data_schema = self._build_form_schema(
config_data[CONF_HOST],
config_data[CONF_USERNAME],
config_data[CONF_PORT],
config_data[CONF_VERIFY_SSL],
)
with _catch_unifi_api_flow_errors(errors):
sites = await self._async_update_sites(config_data)
if (
(unique_id := config_entry.unique_id) is not None
) and unique_id in sites:
config_data[CONF_SITE_ID] = sites[unique_id].name
return self.async_update_reload_and_abort(
config_entry, data_updates=config_data
)
raise AbortFlow("unknown_site_id")
else:
data_schema = self._build_form_schema(
config_entry.data[CONF_HOST],
config_entry.data[CONF_USERNAME],
config_entry.data[CONF_PORT],
config_entry.data[CONF_VERIFY_SSL],
)
return self.async_show_form(
step_id="reconfigure",
data_schema=data_schema,
errors=errors,
)
@override
async def async_step_integration_discovery(
@@ -259,6 +244,39 @@ class UnifiFlowHandler(ConfigFlow, domain=DOMAIN):
return await self.async_step_user()
def _build_form_schema(
self,
host: str = DEFAULT_HOST,
username: str = "",
port: int = DEFAULT_PORT,
verify_ssl: bool = DEFAULT_VERIFY_SSL,
) -> probatio.Schema:
return probatio.Schema(
{
probatio.Required(CONF_HOST, default=host): str,
probatio.Required(CONF_USERNAME, default=username): str,
probatio.Required(CONF_PASSWORD): str,
probatio.Optional(CONF_PORT, default=port): int,
probatio.Optional(
CONF_VERIFY_SSL,
default=verify_ssl,
): bool,
}
)
async def _async_update_sites(self, data: Mapping[str, Any]) -> Sites:
"""Get updated sites through UniFi API."""
hub = await get_unifi_api(self.hass, MappingProxyType(data))
await hub.sites.update()
return hub.sites
@callback
def _get_reauth_or_reconfigure_entry(self) -> ConfigEntry:
"""Return the config entry the current flow is modifying."""
if self.source == SOURCE_REAUTH:
return self._get_reauth_entry()
return self._get_reconfigure_entry()
class UnifiOptionsFlowHandler(OptionsFlow):
"""Handle Unifi Network options."""
@@ -407,3 +425,26 @@ async def _async_discover_unifi(hass: HomeAssistant) -> str | None:
return await hass.async_add_executor_job(socket.gethostbyname, "unifi")
except socket.gaierror:
return None
def _config_from_input(user_input: dict[str, Any]) -> dict[str, Any]:
"""Build config entry data from user input."""
return {
CONF_HOST: user_input[CONF_HOST],
CONF_USERNAME: user_input[CONF_USERNAME],
CONF_PASSWORD: user_input[CONF_PASSWORD],
CONF_PORT: user_input.get(CONF_PORT),
CONF_VERIFY_SSL: user_input.get(CONF_VERIFY_SSL),
CONF_SITE_ID: DEFAULT_SITE_ID,
}
@contextmanager
def _catch_unifi_api_flow_errors(errors: dict[str, str]) -> Iterator[None]:
"""Map UniFi API exceptions to config flow form errors."""
try:
yield
except AuthenticationRequired:
errors["base"] = "faulty_credentials"
except CannotConnect:
errors["base"] = "service_unavailable"
homeassistant/components/unifi/quality_scale.yaml
+1 -6
View File
@@ -58,12 +58,7 @@ rules:
entity-translations: done
exception-translations: todo
icon-translations: done
reconfiguration-flow:
status: todo
comment: |
The user flow currently allows updating existing config entry data
(host/credentials), which should be handled by a dedicated
async_step_reconfigure instead.
reconfiguration-flow: done
repair-issues: todo
stale-devices:
status: todo
homeassistant/components/unifi/strings.json
+19
View File
@@ -5,6 +5,7 @@
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"configuration_updated": "Configuration updated",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]",
"reconfigure_successful": "[%key:common::config_flow::abort::reconfigure_successful%]",
"unknown_site_id": "Previously configured UniFi Network site can no longer be found"
},
"error": {
@@ -14,6 +15,24 @@
},
"flow_title": "{name} ({host})",
"step": {
"reconfigure": {
"data": {
"host": "[%key:common::config_flow::data::host%]",
"password": "[%key:common::config_flow::data::password%]",
"port": "[%key:common::config_flow::data::port%]",
"site": "[%key:component::unifi::config::step::user::data::site%]",
"username": "[%key:common::config_flow::data::username%]",
"verify_ssl": "[%key:common::config_flow::data::verify_ssl%]"
},
"data_description": {
"host": "[%key:component::unifi::config::step::user::data_description::host%]",
"password": "[%key:component::unifi::config::step::user::data_description::password%]",
"port": "[%key:component::unifi::config::step::user::data_description::port%]",
"username": "[%key:component::unifi::config::step::user::data_description::username%]",
"verify_ssl": "[%key:component::unifi::config::step::user::data_description::verify_ssl%]"
},
"title": "[%key:component::unifi::config::step::user::title%]"
},
"site": {
"data": {
"site": "Site ID"
tests/components/unifi/test_config_flow.py
+122 -75
View File
@@ -3,7 +3,7 @@
from collections.abc import Callable
import socket
from typing import Any
from unittest.mock import PropertyMock, patch
from unittest.mock import patch
import pytest
@@ -38,8 +38,6 @@ from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers.device_registry import format_mac
from .conftest import ConfigEntryFactoryType
from tests.common import MockConfigEntry
CLIENTS = [{"mac": "00:00:00:00:00:01"}]
@@ -109,8 +107,8 @@ async def test_flow_works(hass: HomeAssistant, mock_discovery) -> None:
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["data_schema"]({CONF_USERNAME: "", CONF_PASSWORD: ""}) == {
CONF_HOST: "unifi",
assert result["data_schema"]({CONF_PASSWORD: ""}) == {
CONF_HOST: "1",
CONF_USERNAME: "",
CONF_PASSWORD: "",
CONF_PORT: 443,
@@ -149,8 +147,8 @@ async def test_flow_works_negative_discovery(hass: HomeAssistant) -> None:
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["data_schema"]({CONF_USERNAME: "", CONF_PASSWORD: ""}) == {
CONF_HOST: "",
assert result["data_schema"]({CONF_PASSWORD: ""}) == {
CONF_HOST: "unifi",
CONF_USERNAME: "",
CONF_PASSWORD: "",
CONF_PORT: 443,
@@ -219,35 +217,6 @@ async def test_flow_raise_already_configured(hass: HomeAssistant) -> None:
assert result["reason"] == "already_configured"
@pytest.mark.usefixtures("config_entry_setup")
async def test_flow_aborts_configuration_updated(hass: HomeAssistant) -> None:
"""Test config flow aborts since a connected config entry already exists."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
with patch("homeassistant.components.unifi.async_setup_entry") and patch(
"homeassistant.components.unifi.UnifiHub.available", new_callable=PropertyMock
) as ws_mock:
ws_mock.return_value = False
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_HOST: "1.2.3.4",
CONF_USERNAME: "username",
CONF_PASSWORD: "password",
CONF_PORT: 12345,
CONF_VERIFY_SSL: True,
},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "configuration_updated"
@pytest.mark.parametrize(
("side_effect", "error"),
[
@@ -320,7 +289,7 @@ async def test_reauth_flow_update_configuration(
result = await config_entry.start_reauth_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["step_id"] == "reconfigure"
context = next(
flow["context"]
@@ -332,43 +301,6 @@ async def test_reauth_flow_update_configuration(
"name": config_entry.title,
}
with patch(
"homeassistant.components.unifi.UnifiHub.available", new_callable=PropertyMock
) as ws_mock:
ws_mock.return_value = False
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_HOST: "1.2.3.4",
CONF_USERNAME: "new_name",
CONF_PASSWORD: "new_pass",
CONF_PORT: 1234,
CONF_VERIFY_SSL: True,
},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reauth_successful"
assert config_entry.data[CONF_HOST] == "1.2.3.4"
assert config_entry.data[CONF_USERNAME] == "new_name"
assert config_entry.data[CONF_PASSWORD] == "new_pass"
async def test_reauth_flow_update_configuration_on_not_loaded_entry(
hass: HomeAssistant, config_entry_factory: ConfigEntryFactoryType
) -> None:
"""Verify reauth flow can update hub configuration on a not loaded entry."""
with patch(
"homeassistant.components.unifi.get_unifi_api",
side_effect=CannotConnect,
):
config_entry = await config_entry_factory()
result = await config_entry.start_reauth_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
@@ -406,7 +338,122 @@ async def test_abort_reauth_flow_on_site_id_mismatch(
result = await config_entry.start_reauth_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["step_id"] == "reconfigure"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_HOST: "1.2.3.4",
CONF_USERNAME: "new_name",
CONF_PASSWORD: "new_pass",
CONF_PORT: 1234,
CONF_VERIFY_SSL: True,
},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "unknown_site_id"
assert config_entry.data[CONF_SITE_ID] == "site_id"
async def test_reconfigure_flow_update_configuration(
hass: HomeAssistant, config_entry_setup: MockConfigEntry
) -> None:
"""Verify reconfigure flow can update hub configuration."""
config_entry = config_entry_setup
result = await config_entry.start_reconfigure_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reconfigure"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_HOST: "1.2.3.4",
CONF_USERNAME: "new_name",
CONF_PASSWORD: "new_pass",
CONF_PORT: 1234,
CONF_VERIFY_SSL: True,
},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reconfigure_successful"
assert config_entry.data[CONF_HOST] == "1.2.3.4"
assert config_entry.data[CONF_USERNAME] == "new_name"
assert config_entry.data[CONF_PASSWORD] == "new_pass"
@pytest.mark.parametrize(
("side_effect", "error"),
[
(AuthenticationRequired, "faulty_credentials"),
(CannotConnect, "service_unavailable"),
],
)
@pytest.mark.usefixtures("mock_default_requests")
async def test_reconfigure_flow_retains_user_input_on_error(
hass: HomeAssistant,
config_entry_setup: MockConfigEntry,
side_effect: type[Exception],
error: str,
) -> None:
"""Verify reconfigure flow can update hub configuration."""
config_entry = config_entry_setup
result = await config_entry.start_reconfigure_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reconfigure"
user_input = {
CONF_HOST: "4.3.2.1",
CONF_USERNAME: "new_name",
CONF_PASSWORD: "new_pass",
CONF_PORT: 4321,
CONF_VERIFY_SSL: True,
}
with patch(
"homeassistant.components.unifi.config_flow.get_unifi_api",
side_effect=side_effect,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=user_input,
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": error}
assert result["data_schema"]({CONF_PASSWORD: ""}) == {
CONF_HOST: user_input[CONF_HOST],
CONF_USERNAME: user_input[CONF_USERNAME],
CONF_PASSWORD: "",
CONF_PORT: user_input[CONF_PORT],
CONF_VERIFY_SSL: user_input[CONF_VERIFY_SSL],
}
@pytest.mark.parametrize(
"site_payload",
[
[
{"name": "site2", "role": "admin", "desc": "site2 name", "_id": "2"},
]
],
)
async def test_abort_reconfigure_flow_on_site_id_mismatch(
hass: HomeAssistant,
config_entry: MockConfigEntry,
mock_requests: Callable[[str, str], None],
) -> None:
"""Verify reconfigure flow aborts when original site can no longer be found."""
mock_requests(config_entry.data[CONF_HOST], config_entry.data[CONF_SITE_ID])
result = await config_entry.start_reconfigure_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reconfigure"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],