Add reauth to SFR Box (#86511)

This commit is contained in:
epenet 2023-01-24 11:00:22 +01:00 committed by GitHub
parent e084fe4903
commit 22dee1f92b
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
7 changed files with 144 additions and 9 deletions

View file

@ -4,10 +4,12 @@ from __future__ import annotations
import asyncio
from sfrbox_api.bridge import SFRBox
from sfrbox_api.exceptions import SFRBoxAuthenticationError, SFRBoxError
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.httpx_client import get_async_client
@ -19,6 +21,16 @@ from .models import DomainData
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up SFR box as config entry."""
box = SFRBox(ip=entry.data[CONF_HOST], client=get_async_client(hass))
if (username := entry.data.get(CONF_USERNAME)) and (
password := entry.data.get(CONF_PASSWORD)
):
try:
await box.authenticate(username=username, password=password)
except SFRBoxAuthenticationError as err:
raise ConfigEntryAuthFailed() from err
except SFRBoxError as err:
raise ConfigEntryNotReady() from err
data = DomainData(
dsl=SFRDataUpdateCoordinator(hass, box, "dsl", lambda b: b.dsl_get_info()),
system=SFRDataUpdateCoordinator(

View file

@ -1,13 +1,14 @@
"""SFR Box config flow."""
from __future__ import annotations
from collections.abc import Mapping
from typing import Any
from sfrbox_api.bridge import SFRBox
from sfrbox_api.exceptions import SFRBoxAuthenticationError, SFRBoxError
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow
from homeassistant.config_entries import ConfigEntry, ConfigFlow
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers import selector
@ -34,8 +35,9 @@ class SFRBoxFlowHandler(ConfigFlow, domain=DOMAIN):
"""SFR Box config flow."""
VERSION = 1
_config: dict[str, Any] = {}
_box: SFRBox
_config: dict[str, Any] = {}
_reauth_entry: ConfigEntry | None = None
async def async_step_user(
self, user_input: dict[str, str] | None = None
@ -84,10 +86,21 @@ class SFRBoxFlowHandler(ConfigFlow, domain=DOMAIN):
except SFRBoxAuthenticationError:
errors["base"] = "invalid_auth"
else:
if reauth_entry := self._reauth_entry:
data = {**reauth_entry.data, **user_input}
self.hass.config_entries.async_update_entry(reauth_entry, data=data)
self.hass.async_create_task(
self.hass.config_entries.async_reload(reauth_entry.entry_id)
)
return self.async_abort(reason="reauth_successful")
self._config.update(user_input)
return self.async_create_entry(title="SFR Box", data=self._config)
data_schema = self.add_suggested_values_to_schema(AUTH_SCHEMA, user_input)
suggested_values: Mapping[str, Any] | None = user_input
if self._reauth_entry and not suggested_values:
suggested_values = self._reauth_entry.data
data_schema = self.add_suggested_values_to_schema(AUTH_SCHEMA, suggested_values)
return self.async_show_form(
step_id="auth", data_schema=data_schema, errors=errors
)
@ -97,3 +110,11 @@ class SFRBoxFlowHandler(ConfigFlow, domain=DOMAIN):
) -> FlowResult:
"""Skip authentication."""
return self.async_create_entry(title="SFR Box", data=self._config)
async def async_step_reauth(self, entry_data: Mapping[str, Any]) -> FlowResult:
"""Handle failed credentials."""
self._reauth_entry = self.hass.config_entries.async_get_entry(
self.context["entry_id"]
)
self._box = SFRBox(ip=entry_data[CONF_HOST], client=get_async_client(self.hass))
return await self.async_step_auth()

View file

@ -1,7 +1,8 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",

View file

@ -1,7 +1,8 @@
{
"config": {
"abort": {
"already_configured": "Device is already configured"
"already_configured": "Device is already configured",
"reauth_successful": "Re-authentication was successful"
},
"error": {
"cannot_connect": "Failed to connect",

View file

@ -8,7 +8,7 @@ from sfrbox_api.models import DslInfo, SystemInfo
from homeassistant.components.sfr_box.const import DOMAIN
from homeassistant.config_entries import SOURCE_USER, ConfigEntry
from homeassistant.const import CONF_HOST
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry, load_fixture
@ -29,6 +29,25 @@ def get_config_entry(hass: HomeAssistant) -> ConfigEntry:
return config_entry
@pytest.fixture(name="config_entry_with_auth")
def get_config_entry_with_auth(hass: HomeAssistant) -> ConfigEntry:
"""Create and register mock config entry."""
config_entry_with_auth = MockConfigEntry(
domain=DOMAIN,
source=SOURCE_USER,
data={
CONF_HOST: "192.168.0.1",
CONF_USERNAME: "admin",
CONF_PASSWORD: "password",
},
unique_id="e4:5d:51:00:11:23",
options={},
entry_id="1234567",
)
config_entry_with_auth.add_to_hass(hass)
return config_entry_with_auth
@pytest.fixture()
def system_get_info() -> Generator[SystemInfo, None, None]:
"""Fixture for SFRBox.system_get_info."""

View file

@ -1,4 +1,5 @@
"""Test the SFR Box config flow."""
from collections.abc import Generator
import json
from unittest.mock import AsyncMock, patch
@ -8,6 +9,7 @@ from sfrbox_api.models import SystemInfo
from homeassistant import config_entries, data_entry_flow
from homeassistant.components.sfr_box.const import DOMAIN
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
@ -15,7 +17,7 @@ from tests.common import load_fixture
@pytest.fixture(autouse=True, name="mock_setup_entry")
def override_async_setup_entry() -> AsyncMock:
def override_async_setup_entry() -> Generator[AsyncMock, None, None]:
"""Override async_setup_entry."""
with patch(
"homeassistant.components.sfr_box.async_setup_entry", return_value=True
@ -201,3 +203,50 @@ async def test_config_flow_duplicate_mac(
await hass.async_block_till_done()
assert len(mock_setup_entry.mock_calls) == 0
async def test_reauth(hass: HomeAssistant, config_entry_with_auth: ConfigEntry) -> None:
"""Test the start of the config flow."""
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={
"source": config_entries.SOURCE_REAUTH,
"entry_id": config_entry_with_auth.entry_id,
"unique_id": config_entry_with_auth.unique_id,
},
data=config_entry_with_auth.data,
)
assert result.get("type") == data_entry_flow.FlowResultType.FORM
assert result.get("errors") == {}
# Failed credentials
with patch(
"homeassistant.components.sfr_box.config_flow.SFRBox.authenticate",
side_effect=SFRBoxAuthenticationError,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_USERNAME: "admin",
CONF_PASSWORD: "invalid",
},
)
assert result.get("type") == data_entry_flow.FlowResultType.FORM
assert result.get("errors") == {"base": "invalid_auth"}
# Valid credentials
with patch("homeassistant.components.sfr_box.config_flow.SFRBox.authenticate"):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_USERNAME: "admin",
CONF_PASSWORD: "new_password",
},
)
assert result.get("type") == data_entry_flow.FlowResultType.ABORT
assert result.get("reason") == "reauth_successful"

View file

@ -3,7 +3,7 @@ from collections.abc import Generator
from unittest.mock import patch
import pytest
from sfrbox_api.exceptions import SFRBoxError
from sfrbox_api.exceptions import SFRBoxAuthenticationError, SFRBoxError
from homeassistant.components.sfr_box.const import DOMAIN
from homeassistant.config_entries import ConfigEntry, ConfigEntryState
@ -48,3 +48,35 @@ async def test_setup_entry_exception(
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
assert config_entry.state is ConfigEntryState.SETUP_RETRY
assert not hass.data.get(DOMAIN)
async def test_setup_entry_auth_exception(
hass: HomeAssistant, config_entry_with_auth: ConfigEntry
) -> None:
"""Test ConfigEntryNotReady when API raises an exception during authentication."""
with patch(
"homeassistant.components.sfr_box.coordinator.SFRBox.authenticate",
side_effect=SFRBoxError,
):
await hass.config_entries.async_setup(config_entry_with_auth.entry_id)
await hass.async_block_till_done()
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
assert config_entry_with_auth.state is ConfigEntryState.SETUP_RETRY
assert not hass.data.get(DOMAIN)
async def test_setup_entry_invalid_auth(
hass: HomeAssistant, config_entry_with_auth: ConfigEntry
) -> None:
"""Test ConfigEntryAuthFailed when API raises an exception during authentication."""
with patch(
"homeassistant.components.sfr_box.coordinator.SFRBox.authenticate",
side_effect=SFRBoxAuthenticationError,
):
await hass.config_entries.async_setup(config_entry_with_auth.entry_id)
await hass.async_block_till_done()
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
assert config_entry_with_auth.state is ConfigEntryState.SETUP_ERROR
assert not hass.data.get(DOMAIN)