SMA add re-authentication flow (#144538)

* Add reauth flow

* Small adjustment

* Update homeassistant/components/sma/config_flow.py

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>

* Review feedback

* Update tests/components/sma/test_config_flow.py

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>

* Update tests/components/sma/test_config_flow.py

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>

* Feedback

---------

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
This commit is contained in:
Erwin Douna 2025-05-09 12:16:49 +02:00 committed by GitHub
parent 9abb4ffc97
commit 47455fee41
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 133 additions and 2 deletions

View File

@ -20,7 +20,7 @@ from homeassistant.const import (
EVENT_HOMEASSISTANT_STOP,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.device_registry import DeviceInfo
@ -63,6 +63,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
pysma.exceptions.SmaConnectionException,
) as exc:
raise ConfigEntryNotReady from exc
except pysma.exceptions.SmaAuthenticationException as exc:
raise ConfigEntryAuthFailed from exc
if TYPE_CHECKING:
assert entry.unique_id

View File

@ -2,6 +2,7 @@
from __future__ import annotations
from collections.abc import Mapping
import logging
from typing import Any
@ -137,6 +138,42 @@ class SmaConfigFlow(ConfigFlow, domain=DOMAIN):
errors=errors,
)
async def async_step_reauth(
self, entry_data: Mapping[str, Any]
) -> ConfigFlowResult:
"""Handle reauth on credential failure."""
return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Prepare reauth."""
errors: dict[str, str] = {}
if user_input is not None:
reauth_entry = self._get_reauth_entry()
errors, device_info = await self._handle_user_input(
user_input={
**reauth_entry.data,
CONF_PASSWORD: user_input[CONF_PASSWORD],
}
)
if not errors:
return self.async_update_reload_and_abort(
reauth_entry,
data_updates={CONF_PASSWORD: user_input[CONF_PASSWORD]},
)
return self.async_show_form(
step_id="reauth_confirm",
data_schema=vol.Schema(
{
vol.Required(CONF_PASSWORD): cv.string,
}
),
errors=errors,
)
async def async_step_dhcp(
self, discovery_info: DhcpServiceInfo
) -> ConfigFlowResult:

View File

@ -2,7 +2,8 @@
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]"
"already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
@ -11,6 +12,13 @@
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"step": {
"reauth_confirm": {
"title": "[%key:common::config_flow::title::reauth%]",
"description": "The SMA integration needs to re-authenticate your connection details",
"data": {
"password": "[%key:common::config_flow::data::password%]"
}
},
"user": {
"data": {
"group": "Group",

View File

@ -27,6 +27,10 @@ MOCK_USER_INPUT = {
CONF_PASSWORD: "password",
}
MOCK_USER_REAUTH = {
CONF_PASSWORD: "new_password",
}
MOCK_DHCP_DISCOVERY_INPUT = {
# CONF_HOST: "1.1.1.2",
CONF_SSL: True,

View File

@ -20,6 +20,7 @@ from . import (
MOCK_DHCP_DISCOVERY,
MOCK_DHCP_DISCOVERY_INPUT,
MOCK_USER_INPUT,
MOCK_USER_REAUTH,
_patch_async_setup_entry,
)
@ -221,3 +222,82 @@ async def test_dhcp_exceptions(
assert result["title"] == MOCK_DHCP_DISCOVERY["host"]
assert result["data"] == MOCK_DHCP_DISCOVERY
assert result["result"].unique_id == DHCP_DISCOVERY.hostname.replace("SMA", "")
async def test_full_flow_reauth(
hass: HomeAssistant, mock_config_entry: MockConfigEntry
) -> None:
"""Test the full flow of the config flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await mock_config_entry.start_reauth_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
# There is no user input
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
with (
patch("pysma.SMA.new_session", return_value=True),
patch("pysma.SMA.device_info", return_value=MOCK_DEVICE),
_patch_async_setup_entry() as mock_setup_entry,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
MOCK_USER_REAUTH,
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reauth_successful"
assert len(mock_setup_entry.mock_calls) == 1
@pytest.mark.parametrize(
("exception", "error"),
[
(SmaConnectionException, "cannot_connect"),
(SmaAuthenticationException, "invalid_auth"),
(SmaReadException, "cannot_retrieve_device_info"),
(Exception, "unknown"),
],
)
async def test_reauth_flow_exceptions(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
exception: Exception,
error: str,
) -> None:
"""Test we handle cannot connect error."""
result = await mock_config_entry.start_reauth_flow(hass)
with (
patch("pysma.SMA.new_session", side_effect=exception),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
MOCK_USER_REAUTH,
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": error}
assert result["step_id"] == "reauth_confirm"
with (
patch("pysma.SMA.new_session", return_value=True),
patch("pysma.SMA.device_info", return_value=MOCK_DEVICE),
_patch_async_setup_entry() as mock_setup_entry,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
MOCK_USER_REAUTH,
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reauth_successful"
assert len(mock_setup_entry.mock_calls) == 1