Add reauth flow to Habitica integration (#131676)

* Add reauth flow to Habitica integration

* tests, invalid_credentials string

* test only api_key

* section consts

* test config entry

* test reauth is triggered

* set reauthentication-flow to done

* use consts in tests

* reauth_entry

* changes

* fix import

* changes
This commit is contained in:
Manu 2024-12-29 21:12:36 +01:00 committed by GitHub
parent 53e69af088
commit 9804e8aa98
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 385 additions and 64 deletions

View File

@ -2,13 +2,21 @@
from __future__ import annotations
from collections.abc import Mapping
import logging
from typing import TYPE_CHECKING, Any
from aiohttp import ClientError
from habiticalib import Habitica, HabiticaException, NotAuthorizedError
from habiticalib import (
Habitica,
HabiticaException,
LoginData,
NotAuthorizedError,
UserData,
)
import voluptuous as vol
from homeassistant import data_entry_flow
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import (
CONF_API_KEY,
@ -25,12 +33,15 @@ from homeassistant.helpers.selector import (
TextSelectorType,
)
from . import HabiticaConfigEntry
from .const import (
CONF_API_USER,
DEFAULT_URL,
DOMAIN,
FORGOT_PASSWORD_URL,
HABITICANS_URL,
SECTION_REAUTH_API_KEY,
SECTION_REAUTH_LOGIN,
SIGN_UP_URL,
SITE_DATA_URL,
X_CLIENT,
@ -62,14 +73,44 @@ STEP_LOGIN_DATA_SCHEMA = vol.Schema(
}
)
STEP_REAUTH_DATA_SCHEMA = vol.Schema(
{
vol.Required(SECTION_REAUTH_LOGIN): data_entry_flow.section(
vol.Schema(
{
vol.Optional(CONF_USERNAME): TextSelector(
TextSelectorConfig(
type=TextSelectorType.EMAIL,
autocomplete="email",
)
),
vol.Optional(CONF_PASSWORD): TextSelector(
TextSelectorConfig(
type=TextSelectorType.PASSWORD,
autocomplete="current-password",
)
),
},
),
{"collapsed": False},
),
vol.Required(SECTION_REAUTH_API_KEY): data_entry_flow.section(
vol.Schema(
{
vol.Optional(CONF_API_KEY): str,
},
),
{"collapsed": True},
),
}
)
_LOGGER = logging.getLogger(__name__)
class HabiticaConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for habitica."""
VERSION = 1
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
@ -94,33 +135,20 @@ class HabiticaConfigFlow(ConfigFlow, domain=DOMAIN):
"""
errors: dict[str, str] = {}
if user_input is not None:
session = async_get_clientsession(self.hass)
api = Habitica(session=session, x_client=X_CLIENT)
try:
login = await api.login(
username=user_input[CONF_USERNAME],
password=user_input[CONF_PASSWORD],
)
user = await api.get_user(user_fields="profile")
except NotAuthorizedError:
errors["base"] = "invalid_auth"
except (HabiticaException, ClientError):
errors["base"] = "cannot_connect"
except Exception:
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
await self.async_set_unique_id(str(login.data.id))
errors, login, user = await self.validate_login(
{**user_input, CONF_URL: DEFAULT_URL}
)
if not errors and login is not None and user is not None:
await self.async_set_unique_id(str(login.id))
self._abort_if_unique_id_configured()
if TYPE_CHECKING:
assert user.data.profile.name
assert user.profile.name
return self.async_create_entry(
title=user.data.profile.name,
title=user.profile.name,
data={
CONF_API_USER: str(login.data.id),
CONF_API_KEY: login.data.apiToken,
CONF_NAME: user.data.profile.name, # needed for api_call action
CONF_API_USER: str(login.id),
CONF_API_KEY: login.apiToken,
CONF_NAME: user.profile.name, # needed for api_call action
CONF_URL: DEFAULT_URL,
CONF_VERIFY_SSL: True,
},
@ -145,36 +173,18 @@ class HabiticaConfigFlow(ConfigFlow, domain=DOMAIN):
"""
errors: dict[str, str] = {}
if user_input is not None:
session = async_get_clientsession(
self.hass, verify_ssl=user_input[CONF_VERIFY_SSL]
)
try:
api = Habitica(
session=session,
x_client=X_CLIENT,
api_user=user_input[CONF_API_USER],
api_key=user_input[CONF_API_KEY],
url=user_input.get(CONF_URL, DEFAULT_URL),
)
user = await api.get_user(user_fields="profile")
except NotAuthorizedError:
errors["base"] = "invalid_auth"
except (HabiticaException, ClientError):
errors["base"] = "cannot_connect"
except Exception:
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
await self.async_set_unique_id(user_input[CONF_API_USER])
self._abort_if_unique_id_configured()
await self.async_set_unique_id(user_input[CONF_API_USER])
self._abort_if_unique_id_configured()
errors, user = await self.validate_api_key(user_input)
if not errors and user is not None:
if TYPE_CHECKING:
assert user.data.profile.name
assert user.profile.name
return self.async_create_entry(
title=user.data.profile.name,
title=user.profile.name,
data={
**user_input,
CONF_URL: user_input.get(CONF_URL, DEFAULT_URL),
CONF_NAME: user.data.profile.name, # needed for api_call action
CONF_NAME: user.profile.name, # needed for api_call action
},
)
@ -189,3 +199,120 @@ class HabiticaConfigFlow(ConfigFlow, domain=DOMAIN):
"default_url": DEFAULT_URL,
},
)
async def async_step_reauth(
self, entry_data: Mapping[str, Any]
) -> ConfigFlowResult:
"""Perform reauth upon an API authentication error."""
return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Dialog that informs the user that reauth is required."""
errors: dict[str, str] = {}
reauth_entry: HabiticaConfigEntry = self._get_reauth_entry()
if user_input is not None:
if user_input[SECTION_REAUTH_LOGIN].get(CONF_USERNAME) and user_input[
SECTION_REAUTH_LOGIN
].get(CONF_PASSWORD):
errors, login, _ = await self.validate_login(
{**reauth_entry.data, **user_input[SECTION_REAUTH_LOGIN]}
)
if not errors and login is not None:
await self.async_set_unique_id(str(login.id))
self._abort_if_unique_id_mismatch()
return self.async_update_reload_and_abort(
reauth_entry,
data_updates={CONF_API_KEY: login.apiToken},
)
elif user_input[SECTION_REAUTH_API_KEY].get(CONF_API_KEY):
errors, user = await self.validate_api_key(
{
**reauth_entry.data,
**user_input[SECTION_REAUTH_API_KEY],
}
)
if not errors and user is not None:
return self.async_update_reload_and_abort(
reauth_entry, data_updates=user_input[SECTION_REAUTH_API_KEY]
)
else:
errors["base"] = "invalid_credentials"
return self.async_show_form(
step_id="reauth_confirm",
data_schema=self.add_suggested_values_to_schema(
data_schema=STEP_REAUTH_DATA_SCHEMA,
suggested_values={
CONF_USERNAME: (
user_input[SECTION_REAUTH_LOGIN].get(CONF_USERNAME)
if user_input
else None,
)
},
),
description_placeholders={
CONF_NAME: reauth_entry.title,
"habiticans": HABITICANS_URL,
},
errors=errors,
)
async def validate_login(
self, user_input: Mapping[str, Any]
) -> tuple[dict[str, str], LoginData | None, UserData | None]:
"""Validate login with login credentials."""
errors: dict[str, str] = {}
session = async_get_clientsession(
self.hass, verify_ssl=user_input.get(CONF_VERIFY_SSL, True)
)
api = Habitica(session=session, x_client=X_CLIENT)
try:
login = await api.login(
username=user_input[CONF_USERNAME],
password=user_input[CONF_PASSWORD],
)
user = await api.get_user(user_fields="profile")
except NotAuthorizedError:
errors["base"] = "invalid_auth"
except (HabiticaException, ClientError):
errors["base"] = "cannot_connect"
except Exception:
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
return errors, login.data, user.data
return errors, None, None
async def validate_api_key(
self, user_input: Mapping[str, Any]
) -> tuple[dict[str, str], UserData | None]:
"""Validate authentication with api key."""
errors: dict[str, str] = {}
session = async_get_clientsession(
self.hass, verify_ssl=user_input.get(CONF_VERIFY_SSL, True)
)
api = Habitica(
session=session,
x_client=X_CLIENT,
api_user=user_input[CONF_API_USER],
api_key=user_input[CONF_API_KEY],
url=user_input.get(CONF_URL, DEFAULT_URL),
)
try:
user = await api.get_user(user_fields="profile")
except NotAuthorizedError:
errors["base"] = "invalid_auth"
except (HabiticaException, ClientError):
errors["base"] = "cannot_connect"
except Exception:
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
return errors, user.data
return errors, None

View File

@ -46,3 +46,6 @@ SERVICE_TRANSFORMATION = "transformation"
DEVELOPER_ID = "4c4ca53f-c059-4ffa-966e-9d29dd405daf"
X_CLIENT = f"{DEVELOPER_ID} - {APPLICATION_NAME} {__version__}"
SECTION_REAUTH_LOGIN = "reauth_login"
SECTION_REAUTH_API_KEY = "reauth_api_key"

View File

@ -23,7 +23,11 @@ from habiticalib import (
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_NAME
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady, HomeAssistantError
from homeassistant.exceptions import (
ConfigEntryAuthFailed,
ConfigEntryNotReady,
HomeAssistantError,
)
from homeassistant.helpers.debounce import Debouncer
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
@ -71,10 +75,9 @@ class HabiticaDataUpdateCoordinator(DataUpdateCoordinator[HabiticaData]):
await self.habitica.get_content(user.data.preferences.language)
).data
except NotAuthorizedError as e:
raise ConfigEntryNotReady(
raise ConfigEntryAuthFailed(
translation_domain=DOMAIN,
translation_key="invalid_auth",
translation_placeholders={"account": self.config_entry.title},
translation_key="authentication_failed",
) from e
except TooManyRequestsError as e:
raise ConfigEntryNotReady(

View File

@ -34,7 +34,7 @@ rules:
integration-owner: done
log-when-unavailable: done
parallel-updates: todo
reauthentication-flow: todo
reauthentication-flow: done
test-coverage: done
# Gold

View File

@ -10,12 +10,15 @@
},
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]"
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]",
"unique_id_mismatch": "Hmm, those login details are correct, but they're not for this adventurer. Got another account to try?",
"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%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
"unknown": "[%key:common::config_flow::error::unknown%]",
"invalid_credentials": "Input is incomplete. You must provide either your login details or an API token"
},
"step": {
"user": {
@ -53,6 +56,34 @@
"verify_ssl": "Enable SSL certificate verification for secure connections. Disable only if connecting to a Habitica instance using a self-signed certificate"
},
"description": "You can retrieve your `User ID` and `API Token` from [**Settings -> Site Data**]({site_data}) on Habitica or the instance you want to connect to"
},
"reauth_confirm": {
"title": "Re-authorize {name} with Habitica",
"description": "![Habiticans]({habiticans}) It seems your API token for **{name}** has been reset. To re-authorize the integration, you can either log in with your username or email, and password, or directly provide your new API token.",
"sections": {
"reauth_login": {
"name": "Re-authorize via login",
"description": "Enter your login details below to re-authorize the Home Assistant integration with Habitica",
"data": {
"username": "[%key:component::habitica::config::step::login::data::username%]",
"password": "[%key:common::config_flow::data::password%]"
},
"data_description": {
"username": "[%key:component::habitica::config::step::login::data_description::username%]",
"password": "[%key:component::habitica::config::step::login::data_description::password%]"
}
},
"reauth_api_key": {
"description": "Enter your new API token below. You can find it in Habitica under 'Settings -> Site Data'",
"name": "Re-authorize via API Token",
"data": {
"api_key": "[%key:component::habitica::config::step::advanced::data::api_key%]"
},
"data_description": {
"api_key": "[%key:component::habitica::config::step::advanced::data_description::api_key%]"
}
}
}
}
}
},
@ -367,8 +398,8 @@
"item_not_found": {
"message": "Unable to use {item}, you don't own this item."
},
"invalid_auth": {
"message": "Authentication failed for {account}."
"authentication_failed": {
"message": "Authentication failed. It looks like your API token has been reset. Please re-authenticate using your new token"
}
},
"issues": {

View File

@ -1,10 +1,17 @@
"""Test the habitica config flow."""
from typing import Any
from unittest.mock import AsyncMock
import pytest
from homeassistant.components.habitica.const import CONF_API_USER, DEFAULT_URL, DOMAIN
from homeassistant.components.habitica.const import (
CONF_API_USER,
DEFAULT_URL,
DOMAIN,
SECTION_REAUTH_API_KEY,
SECTION_REAUTH_LOGIN,
)
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import (
CONF_API_KEY,
@ -36,6 +43,18 @@ MOCK_DATA_ADVANCED_STEP = {
CONF_VERIFY_SSL: True,
}
USER_INPUT_REAUTH_LOGIN = {
SECTION_REAUTH_LOGIN: {
CONF_USERNAME: "new-email",
CONF_PASSWORD: "new-password",
},
SECTION_REAUTH_API_KEY: {},
}
USER_INPUT_REAUTH_API_KEY = {
SECTION_REAUTH_LOGIN: {},
SECTION_REAUTH_API_KEY: {CONF_API_KEY: "cd0e5985-17de-4b4f-849e-5d506c5e4382"},
}
@pytest.mark.usefixtures("habitica")
async def test_form_login(hass: HomeAssistant, mock_setup_entry: AsyncMock) -> None:
@ -253,3 +272,118 @@ async def test_form_advanced_already_configured(
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
@pytest.mark.parametrize(
"user_input",
[
(USER_INPUT_REAUTH_LOGIN),
(USER_INPUT_REAUTH_API_KEY),
],
ids=["reauth with login details", "rauth with api key"],
)
@pytest.mark.usefixtures("habitica")
async def test_flow_reauth(
hass: HomeAssistant, config_entry: MockConfigEntry, user_input: dict[str, Any]
) -> None:
"""Test reauth flow."""
config_entry.add_to_hass(hass)
result = await config_entry.start_reauth_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input,
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reauth_successful"
assert config_entry.data[CONF_API_KEY] == "cd0e5985-17de-4b4f-849e-5d506c5e4382"
assert len(hass.config_entries.async_entries()) == 1
@pytest.mark.parametrize(
("raise_error", "user_input", "text_error"),
[
(
ERROR_BAD_REQUEST,
USER_INPUT_REAUTH_LOGIN,
"cannot_connect",
),
(
ERROR_NOT_AUTHORIZED,
USER_INPUT_REAUTH_LOGIN,
"invalid_auth",
),
(IndexError(), USER_INPUT_REAUTH_LOGIN, "unknown"),
(
ERROR_BAD_REQUEST,
USER_INPUT_REAUTH_API_KEY,
"cannot_connect",
),
(
ERROR_NOT_AUTHORIZED,
USER_INPUT_REAUTH_API_KEY,
"invalid_auth",
),
(IndexError(), USER_INPUT_REAUTH_API_KEY, "unknown"),
(
None,
{SECTION_REAUTH_LOGIN: {}, SECTION_REAUTH_API_KEY: {}},
"invalid_credentials",
),
],
ids=[
"login cannot_connect",
"login invalid_auth",
"login unknown",
"api_key cannot_connect",
"api_key invalid_auth",
"api_key unknown",
"invalid_credentials",
],
)
async def test_flow_reauth_errors(
hass: HomeAssistant,
habitica: AsyncMock,
config_entry: MockConfigEntry,
raise_error: Exception,
user_input: dict[str, Any],
text_error: str,
) -> None:
"""Test reauth flow with invalid credentials."""
config_entry.add_to_hass(hass)
result = await config_entry.start_reauth_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
habitica.get_user.side_effect = raise_error
habitica.login.side_effect = raise_error
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": text_error}
habitica.get_user.side_effect = None
habitica.login.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=USER_INPUT_REAUTH_API_KEY,
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reauth_successful"
assert config_entry.data[CONF_API_KEY] == "cd0e5985-17de-4b4f-849e-5d506c5e4382"
assert len(hass.config_entries.async_entries()) == 1

View File

@ -15,7 +15,7 @@ from homeassistant.components.habitica.const import (
EVENT_API_CALL_SUCCESS,
SERVICE_API_CALL,
)
from homeassistant.config_entries import ConfigEntryState
from homeassistant.config_entries import SOURCE_REAUTH, ConfigEntryState
from homeassistant.const import ATTR_NAME
from homeassistant.core import Event, HomeAssistant
@ -88,9 +88,8 @@ async def test_service_call(
[
ERROR_BAD_REQUEST,
ERROR_TOO_MANY_REQUESTS,
ERROR_NOT_AUTHORIZED,
],
ids=["BadRequestError", "TooManyRequestsError", "NotAuthorizedError"],
ids=["BadRequestError", "TooManyRequestsError"],
)
async def test_config_entry_not_ready(
hass: HomeAssistant,
@ -108,6 +107,30 @@ async def test_config_entry_not_ready(
assert config_entry.state is ConfigEntryState.SETUP_RETRY
async def test_config_entry_auth_failed(
hass: HomeAssistant, config_entry: MockConfigEntry, habitica: AsyncMock
) -> None:
"""Test config entry auth failed setup error."""
habitica.get_user.side_effect = ERROR_NOT_AUTHORIZED
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.SETUP_ERROR
flows = hass.config_entries.flow.async_progress()
assert len(flows) == 1
flow = flows[0]
assert flow.get("step_id") == "reauth_confirm"
assert flow.get("handler") == DOMAIN
assert "context" in flow
assert flow["context"].get("source") == SOURCE_REAUTH
assert flow["context"].get("entry_id") == config_entry.entry_id
async def test_coordinator_update_failed(
hass: HomeAssistant,
config_entry: MockConfigEntry,