Add re-auth-flow to fyta integration (#114972)

* add re-auth-flow to fyta integration

* add strings for reauth-flow

* resolve typing error

* update based on review comments

* Update homeassistant/components/fyta/config_flow.py

Co-authored-by: G Johansson <goran.johansson@shiftit.se>

* add async_auth

* adjustment based on review commet

* Update test_config_flow.py

* remove credentials

* Update homeassistant/components/fyta/config_flow.py

Co-authored-by: G Johansson <goran.johansson@shiftit.se>

* Update tests/components/fyta/test_config_flow.py

Co-authored-by: G Johansson <goran.johansson@shiftit.se>

* Update tests/components/fyta/test_config_flow.py

Co-authored-by: G Johansson <goran.johansson@shiftit.se>

* Update conftest.py

* Update test_config_flow.py

* Aktualisieren von conftest.py

* Update test_config_flow.py

---------

Co-authored-by: G Johansson <goran.johansson@shiftit.se>
This commit is contained in:
dontinelli 2024-04-12 20:33:24 +02:00 committed by GitHub
parent 77d1e2c812
commit 6eaf3402c6
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 129 additions and 29 deletions

View file

@ -2,6 +2,7 @@
from __future__ import annotations
from collections.abc import Mapping
import logging
from typing import Any
@ -13,7 +14,7 @@ from fyta_cli.fyta_exceptions import (
)
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.config_entries import ConfigEntry, ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from .const import DOMAIN
@ -30,36 +31,70 @@ class FytaConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Fyta."""
VERSION = 1
_entry: ConfigEntry | None = None
async def async_auth(self, user_input: Mapping[str, Any]) -> dict[str, str]:
"""Reusable Auth Helper."""
fyta = FytaConnector(user_input[CONF_USERNAME], user_input[CONF_PASSWORD])
try:
await fyta.login()
except FytaConnectionError:
return {"base": "cannot_connect"}
except FytaAuthentificationError:
return {"base": "invalid_auth"}
except FytaPasswordError:
return {"base": "invalid_auth", CONF_PASSWORD: "password_error"}
except Exception as e: # pylint: disable=broad-except
_LOGGER.error(e)
return {"base": "unknown"}
finally:
await fyta.client.close()
return {}
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the initial step."""
errors = {}
if user_input:
self._async_abort_entries_match({CONF_USERNAME: user_input[CONF_USERNAME]})
fyta = FytaConnector(user_input[CONF_USERNAME], user_input[CONF_PASSWORD])
try:
await fyta.login()
except FytaConnectionError:
errors["base"] = "cannot_connect"
except FytaAuthentificationError:
errors["base"] = "invalid_auth"
except FytaPasswordError:
errors["base"] = "invalid_auth"
errors[CONF_PASSWORD] = "password_error"
except Exception: # pylint: disable=broad-except
errors["base"] = "unknown"
else:
if not (errors := await self.async_auth(user_input)):
return self.async_create_entry(
title=user_input[CONF_USERNAME], data=user_input
)
finally:
await fyta.client.close()
return self.async_show_form(
step_id="user", data_schema=DATA_SCHEMA, errors=errors
)
async def async_step_reauth(
self, entry_data: Mapping[str, Any]
) -> ConfigFlowResult:
"""Handle flow upon an API authentication error."""
self._entry = self.hass.config_entries.async_get_entry(self.context["entry_id"])
return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle reauthorization flow."""
errors = {}
assert self._entry is not None
if user_input and not (errors := await self.async_auth(user_input)):
return self.async_update_reload_and_abort(
self._entry, data={**self._entry.data, **user_input}
)
data_schema = self.add_suggested_values_to_schema(
DATA_SCHEMA,
{CONF_USERNAME: self._entry.data[CONF_USERNAME], **(user_input or {})},
)
return self.async_show_form(
step_id="reauth_confirm",
data_schema=data_schema,
errors=errors,
)

View file

@ -13,7 +13,7 @@ from fyta_cli.fyta_exceptions import (
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryError, ConfigEntryNotReady
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
_LOGGER = logging.getLogger(__name__)
@ -52,4 +52,4 @@ class FytaCoordinator(DataUpdateCoordinator[dict[int, dict[str, Any]]]):
except FytaConnectionError as ex:
raise ConfigEntryNotReady from ex
except (FytaAuthentificationError, FytaPasswordError) as ex:
raise ConfigEntryError from ex
raise ConfigEntryAuthFailed from ex

View file

@ -8,8 +8,19 @@
"username": "[%key:common::config_flow::data::username%]",
"password": "[%key:common::config_flow::data::password%]"
}
},
"reauth_confirm": {
"description": "Update your credentials for FYTA API",
"data": {
"username": "[%key:common::config_flow::data::username%]",
"password": "[%key:common::config_flow::data::password%]"
}
}
},
"abort": {
"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%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",

View file

@ -5,8 +5,6 @@ from unittest.mock import AsyncMock, patch
import pytest
from .test_config_flow import ACCESS_TOKEN, EXPIRATION
@pytest.fixture
def mock_fyta():
@ -17,10 +15,7 @@ def mock_fyta():
"homeassistant.components.fyta.config_flow.FytaConnector",
return_value=mock_fyta_api,
) as mock_fyta_api:
mock_fyta_api.return_value.login.return_value = {
"access_token": ACCESS_TOKEN,
"expiration": EXPIRATION,
}
mock_fyta_api.return_value.login.return_value = {}
yield mock_fyta_api

View file

@ -1,6 +1,5 @@
"""Test the fyta config flow."""
from datetime import datetime
from unittest.mock import AsyncMock
from fyta_cli.fyta_exceptions import (
@ -20,8 +19,6 @@ from tests.common import MockConfigEntry
USERNAME = "fyta_user"
PASSWORD = "fyta_pass"
ACCESS_TOKEN = "123xyz"
EXPIRATION = datetime.now()
async def test_user_flow(
@ -121,3 +118,65 @@ async def test_duplicate_entry(hass: HomeAssistant, mock_fyta: AsyncMock) -> Non
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
@pytest.mark.parametrize(
("exception", "error"),
[
(FytaConnectionError, {"base": "cannot_connect"}),
(FytaAuthentificationError, {"base": "invalid_auth"}),
(FytaPasswordError, {"base": "invalid_auth", CONF_PASSWORD: "password_error"}),
(Exception, {"base": "unknown"}),
],
)
async def test_reauth(
hass: HomeAssistant,
exception: Exception,
error: dict[str, str],
mock_fyta: AsyncMock,
mock_setup_entry,
) -> None:
"""Test reauth-flow works."""
entry = MockConfigEntry(
domain=DOMAIN,
title=USERNAME,
data={CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD},
)
entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_REAUTH, "entry_id": entry.entry_id},
data=entry.data,
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
mock_fyta.return_value.login.side_effect = exception
# tests with connection error
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD}
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
assert result["errors"] == error
mock_fyta.return_value.login.side_effect = None
# tests with all information provided
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_USERNAME: "other_username", CONF_PASSWORD: "other_password"},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reauth_successful"
assert entry.data[CONF_USERNAME] == "other_username"
assert entry.data[CONF_PASSWORD] == "other_password"
assert len(mock_setup_entry.mock_calls) == 1