Add reauth flow to Ituran (#132755)

This commit is contained in:
Assaf Inbal 2024-12-16 15:00:06 +02:00 committed by GitHub
parent 38fdfba169
commit a953abf5c3
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 86 additions and 10 deletions

View File

@ -2,6 +2,7 @@
from __future__ import annotations
from collections.abc import Mapping
import logging
from typing import Any
@ -9,7 +10,7 @@ from pyituran import Ituran
from pyituran.exceptions import IturanApiError, IturanAuthError
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.config_entries import SOURCE_REAUTH, ConfigFlow, ConfigFlowResult
from .const import (
CONF_ID_OR_PASSPORT,
@ -43,11 +44,12 @@ class IturanConfigFlow(ConfigFlow, domain=DOMAIN):
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the inial step."""
"""Handle the initial step."""
errors: dict[str, str] = {}
if user_input is not None:
await self.async_set_unique_id(user_input[CONF_ID_OR_PASSPORT])
self._abort_if_unique_id_configured()
if self.source != SOURCE_REAUTH:
self._abort_if_unique_id_configured()
ituran = Ituran(
user_input[CONF_ID_OR_PASSPORT],
@ -81,7 +83,7 @@ class IturanConfigFlow(ConfigFlow, domain=DOMAIN):
async def async_step_otp(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the inial step."""
"""Handle the OTP step."""
errors: dict[str, str] = {}
if user_input is not None:
ituran = Ituran(
@ -99,6 +101,10 @@ class IturanConfigFlow(ConfigFlow, domain=DOMAIN):
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
if self.source == SOURCE_REAUTH:
return self.async_update_reload_and_abort(
self._get_reauth_entry(), data=self._user_info
)
return self.async_create_entry(
title=f"Ituran {self._user_info[CONF_ID_OR_PASSPORT]}",
data=self._user_info,
@ -107,3 +113,25 @@ class IturanConfigFlow(ConfigFlow, domain=DOMAIN):
return self.async_show_form(
step_id="otp", data_schema=STEP_OTP_DATA_SCHEMA, errors=errors
)
async def async_step_reauth(
self, entry_data: Mapping[str, Any]
) -> ConfigFlowResult:
"""Handle configuration by re-auth."""
self._user_info = dict(entry_data)
return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle reauth confirmation message."""
if user_input is not None:
return await self.async_step_user(self._user_info)
return self.async_show_form(
step_id="reauth_confirm",
data_schema=vol.Schema({}),
description_placeholders={
"phone_number": self._user_info[CONF_PHONE_NUMBER]
},
)

View File

@ -7,7 +7,7 @@ from pyituran.exceptions import IturanApiError, IturanAuthError
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryError
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
@ -54,7 +54,7 @@ class IturanDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Vehicle]]):
translation_domain=DOMAIN, translation_key="api_error"
) from e
except IturanAuthError as e:
raise ConfigEntryError(
raise ConfigEntryAuthFailed(
translation_domain=DOMAIN, translation_key="auth_error"
) from e

View File

@ -35,7 +35,7 @@ rules:
status: exempt
comment: |
This integration does not provide additional actions.
reauthentication-flow: todo
reauthentication-flow: done
parallel-updates:
status: exempt
comment: |

View File

@ -7,7 +7,7 @@
"phone_number": "Mobile phone number"
},
"data_description": {
"id_or_passport": "The goverment ID or passport number provided when registering with Ituran.",
"id_or_passport": "The government ID or passport number provided when registering with Ituran.",
"phone_number": "The mobile phone number provided when registering with Ituran. A one-time password will be sent to this mobile number."
}
},
@ -18,6 +18,10 @@
"data_description": {
"otp": "A one-time-password sent as a text message to the mobile phone number provided before."
}
},
"reauth_confirm": {
"title": "[%key:common::config_flow::title::reauth%]",
"description": "A new one-time password will be sent to {phone_number}."
}
},
"error": {
@ -27,15 +31,16 @@
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"abort": {
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]",
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]"
}
},
"exceptions": {
"api_error": {
"message": "An error occured while communicating with the Ituran service."
"message": "An error occurred while communicating with the Ituran service."
},
"auth_error": {
"message": "Failed authenticating with the Ituran service, please remove and re-add integration."
"message": "Failed authenticating with the Ituran service, please reauthenticate the integration."
}
}
}

View File

@ -16,8 +16,11 @@ from homeassistant.config_entries import SOURCE_USER, ConfigFlowResult
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from . import setup_integration
from .const import MOCK_CONFIG_DATA
from tests.common import MockConfigEntry
async def __do_successful_user_step(
hass: HomeAssistant, result: ConfigFlowResult, mock_ituran: AsyncMock
@ -209,3 +212,43 @@ async def test_already_authenticated(
assert result["data"][CONF_PHONE_NUMBER] == MOCK_CONFIG_DATA[CONF_PHONE_NUMBER]
assert result["data"][CONF_MOBILE_ID] == MOCK_CONFIG_DATA[CONF_MOBILE_ID]
assert result["result"].unique_id == MOCK_CONFIG_DATA[CONF_ID_OR_PASSPORT]
async def test_reauth(
hass: HomeAssistant,
mock_ituran: AsyncMock,
mock_setup_entry: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test reauthenticating."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await __do_successful_user_step(hass, result, mock_ituran)
await __do_successful_otp_step(hass, result, mock_ituran)
await setup_integration(hass, mock_config_entry)
result = await mock_config_entry.start_reauth_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
assert result["errors"] is None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "otp"
assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_OTP: "123456",
},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reauth_successful"