Add reauth flow to Ohme (#133275)

* Add reauth flow to ohme

* Reuse config flow user step for reauth

* Tidying up

* Add common _validate_account method for reauth and user config flow steps

* Add reauth fail test
This commit is contained in:
Dan Raper 2024-12-16 22:54:33 +00:00 committed by GitHub
parent 9cdc36681a
commit a374c7e4ca
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 150 additions and 13 deletions

View File

@ -7,7 +7,7 @@ from ohme import ApiException, AuthException, OhmeApiClient
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryError, ConfigEntryNotReady
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from .const import DOMAIN, PLATFORMS
from .coordinator import OhmeAdvancedSettingsCoordinator, OhmeChargeSessionCoordinator
@ -36,7 +36,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: OhmeConfigEntry) -> bool
translation_key="device_info_failed", translation_domain=DOMAIN
)
except AuthException as e:
raise ConfigEntryError(
raise ConfigEntryAuthFailed(
translation_key="auth_failed", translation_domain=DOMAIN
) from e
except ApiException as e:

View File

@ -1,5 +1,6 @@
"""Config flow for ohme integration."""
from collections.abc import Mapping
from typing import Any
from ohme import ApiException, AuthException, OhmeApiClient
@ -32,6 +33,17 @@ USER_SCHEMA = vol.Schema(
}
)
REAUTH_SCHEMA = vol.Schema(
{
vol.Required(CONF_PASSWORD): TextSelector(
TextSelectorConfig(
type=TextSelectorType.PASSWORD,
autocomplete="current-password",
),
),
}
)
class OhmeConfigFlow(ConfigFlow, domain=DOMAIN):
"""Config flow."""
@ -46,14 +58,9 @@ class OhmeConfigFlow(ConfigFlow, domain=DOMAIN):
if user_input is not None:
self._async_abort_entries_match({CONF_EMAIL: user_input[CONF_EMAIL]})
instance = OhmeApiClient(user_input[CONF_EMAIL], user_input[CONF_PASSWORD])
try:
await instance.async_login()
except AuthException:
errors["base"] = "invalid_auth"
except ApiException:
errors["base"] = "unknown"
errors = await self._validate_account(
user_input[CONF_EMAIL], user_input[CONF_PASSWORD]
)
if not errors:
return self.async_create_entry(
title=user_input[CONF_EMAIL], data=user_input
@ -62,3 +69,48 @@ class OhmeConfigFlow(ConfigFlow, domain=DOMAIN):
return self.async_show_form(
step_id="user", data_schema=USER_SCHEMA, errors=errors
)
async def async_step_reauth(
self, entry_data: Mapping[str, Any]
) -> ConfigFlowResult:
"""Handle re-authentication."""
return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle re-authentication confirmation."""
errors: dict[str, str] = {}
reauth_entry = self._get_reauth_entry()
if user_input is not None:
errors = await self._validate_account(
reauth_entry.data[CONF_EMAIL],
user_input[CONF_PASSWORD],
)
if not errors:
return self.async_update_reload_and_abort(
reauth_entry,
data_updates=user_input,
)
return self.async_show_form(
step_id="reauth_confirm",
data_schema=REAUTH_SCHEMA,
description_placeholders={"email": reauth_entry.data[CONF_EMAIL]},
errors=errors,
)
async def _validate_account(self, email: str, password: str) -> dict[str, str]:
"""Validate Ohme account and return dict of errors."""
errors: dict[str, str] = {}
client = OhmeApiClient(
email,
password,
)
try:
await client.async_login()
except AuthException:
errors["base"] = "invalid_auth"
except ApiException:
errors["base"] = "unknown"
return errors

View File

@ -6,6 +6,6 @@
"documentation": "https://www.home-assistant.io/integrations/ohme/",
"integration_type": "device",
"iot_class": "cloud_polling",
"quality_scale": "bronze",
"quality_scale": "silver",
"requirements": ["ohme==1.1.1"]
}

View File

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

View File

@ -11,6 +11,16 @@
"email": "Enter the email address associated with your Ohme account.",
"password": "Enter the password for your Ohme account"
}
},
"reauth_confirm": {
"description": "Please update your password for {email}",
"title": "[%key:common::config_flow::title::reauth%]",
"data": {
"password": "[%key:common::config_flow::data::password%]"
},
"data_description": {
"password": "Enter the password for your Ohme account"
}
}
},
"error": {
@ -18,7 +28,8 @@
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"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%]"
}
},
"entity": {

View File

@ -108,3 +108,77 @@ async def test_already_configured(
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
async def test_reauth_form(hass: HomeAssistant, mock_client: MagicMock) -> None:
"""Test reauth form."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_EMAIL: "test@example.com",
CONF_PASSWORD: "hunter1",
},
)
entry.add_to_hass(hass)
result = await entry.start_reauth_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
assert not result["errors"]
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_PASSWORD: "hunter2"},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reauth_successful"
@pytest.mark.parametrize(
("test_exception", "expected_error"),
[(AuthException, "invalid_auth"), (ApiException, "unknown")],
)
async def test_reauth_fail(
hass: HomeAssistant,
mock_client: MagicMock,
test_exception: Exception,
expected_error: str,
) -> None:
"""Test reauth errors."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_EMAIL: "test@example.com",
CONF_PASSWORD: "hunter1",
},
)
entry.add_to_hass(hass)
# Initial form load
result = await entry.start_reauth_flow(hass)
assert result["step_id"] == "reauth_confirm"
assert result["type"] is FlowResultType.FORM
assert not result["errors"]
# Failed login
mock_client.async_login.side_effect = test_exception
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_PASSWORD: "hunter1"},
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": expected_error}
# End with success
mock_client.async_login.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_PASSWORD: "hunter2"},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reauth_successful"