Add reauth flow to LetPot integration (#135734)

2025-07-19 03:07:37 +00:00 · 2025-01-18 13:11:35 +01:00 · 2025-01-18 13:11:35 +01:00 · d349c47694
commit d349c47694
parent f878465a9a
6 changed files with 252 additions and 29 deletions
--- a/homeassistant/components/letpot/init.py
+++ b/homeassistant/components/letpot/init.py
@ -12,7 +12,7 @@ from letpot.models import AuthenticationInfo
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import CONF_ACCESS_TOKEN, CONF_EMAIL, Platform
 from homeassistant.core import HomeAssistant
-from homeassistant.exceptions import ConfigEntryError, ConfigEntryNotReady
+from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
 from homeassistant.helpers.aiohttp_client import async_get_clientsession

 from .const import (
@ -57,12 +57,12 @@ async def async_setup_entry(hass: HomeAssistant, entry: LetPotConfigEntry) -> bo
                },
            )
        except LetPotAuthenticationException as exc:
-            raise ConfigEntryError from exc
+            raise ConfigEntryAuthFailed from exc

    try:
        devices = await client.get_devices()
    except LetPotAuthenticationException as exc:
-        raise ConfigEntryError from exc
+        raise ConfigEntryAuthFailed from exc
    except LetPotException as exc:
        raise ConfigEntryNotReady from exc

--- a/homeassistant/components/letpot/config_flow.py
+++ b/homeassistant/components/letpot/config_flow.py
@ -2,6 +2,7 @@

 from __future__ import annotations

+from collections.abc import Mapping
 import logging
 from typing import Any

@ -42,6 +43,13 @@ STEP_USER_DATA_SCHEMA = vol.Schema(
        ),
    }
 )
+STEP_REAUTH_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_PASSWORD): TextSelector(
+            TextSelectorConfig(type=TextSelectorType.PASSWORD),
+        ),
+    }
+)


 class LetPotConfigFlow(ConfigFlow, domain=DOMAIN):
@ -51,18 +59,28 @@ class LetPotConfigFlow(ConfigFlow, domain=DOMAIN):

    async def _async_validate_credentials(
        self, email: str, password: str
-    ) -> dict[str, Any]:
+    ) -> tuple[dict[str, str], dict[str, Any] | None]:
+        """Try logging in to the LetPot account and returns credential info."""
        websession = async_get_clientsession(self.hass)
        client = LetPotClient(websession)
-        auth = await client.login(email, password)
-        return {
-            CONF_ACCESS_TOKEN: auth.access_token,
-            CONF_ACCESS_TOKEN_EXPIRES: auth.access_token_expires,
-            CONF_REFRESH_TOKEN: auth.refresh_token,
-            CONF_REFRESH_TOKEN_EXPIRES: auth.refresh_token_expires,
-            CONF_USER_ID: auth.user_id,
-            CONF_EMAIL: auth.email,
-        }
+        try:
+            auth = await client.login(email, password)
+        except LetPotConnectionException:
+            return {"base": "cannot_connect"}, None
+        except LetPotAuthenticationException:
+            return {"base": "invalid_auth"}, None
+        except Exception:
+            _LOGGER.exception("Unexpected exception")
+            return {"base": "unknown"}, None
+        else:
+            return {}, {
+                CONF_ACCESS_TOKEN: auth.access_token,
+                CONF_ACCESS_TOKEN_EXPIRES: auth.access_token_expires,
+                CONF_REFRESH_TOKEN: auth.refresh_token,
+                CONF_REFRESH_TOKEN_EXPIRES: auth.refresh_token_expires,
+                CONF_USER_ID: auth.user_id,
+                CONF_EMAIL: auth.email,
+            }

    async def async_step_user(
        self, user_input: dict[str, Any] | None = None
@ -70,18 +88,10 @@ class LetPotConfigFlow(ConfigFlow, domain=DOMAIN):
        """Handle a flow initialized by the user."""
        errors: dict[str, str] = {}
        if user_input is not None:
-            try:
-                data_dict = await self._async_validate_credentials(
-                    user_input[CONF_EMAIL], user_input[CONF_PASSWORD]
-                )
-            except LetPotConnectionException:
-                errors["base"] = "cannot_connect"
-            except LetPotAuthenticationException:
-                errors["base"] = "invalid_auth"
-            except Exception:
-                _LOGGER.exception("Unexpected exception")
-                errors["base"] = "unknown"
-            else:
+            errors, data_dict = await self._async_validate_credentials(
+                user_input[CONF_EMAIL], user_input[CONF_PASSWORD]
+            )
+            if not errors and data_dict is not None:
                await self.async_set_unique_id(data_dict[CONF_USER_ID])
                self._abort_if_unique_id_configured()
                return self.async_create_entry(
@ -90,3 +100,36 @@ class LetPotConfigFlow(ConfigFlow, domain=DOMAIN):
        return self.async_show_form(
            step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
        )
+
+    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:
+        """Confirm reauth dialog."""
+        errors: dict[str, str] = {}
+        reauth_entry = self._get_reauth_entry()
+        if user_input is not None:
+            errors, data_dict = await self._async_validate_credentials(
+                reauth_entry.data[CONF_EMAIL], user_input[CONF_PASSWORD]
+            )
+            if not errors and data_dict is not None:
+                await self.async_set_unique_id(data_dict[CONF_USER_ID])
+                if reauth_entry.unique_id != data_dict[CONF_USER_ID]:
+                    # Abort if the received account is different and already added
+                    self._abort_if_unique_id_configured()
+                return self.async_update_reload_and_abort(
+                    reauth_entry,
+                    unique_id=self.unique_id,
+                    data_updates=data_dict,
+                )
+        return self.async_show_form(
+            step_id="reauth_confirm",
+            data_schema=STEP_REAUTH_SCHEMA,
+            description_placeholders={"email": reauth_entry.title},
+            errors=errors,
+        )
--- a/homeassistant/components/letpot/coordinator.py
+++ b/homeassistant/components/letpot/coordinator.py
@ -11,7 +11,7 @@ from letpot.exceptions import LetPotAuthenticationException, LetPotException
 from letpot.models import AuthenticationInfo, LetPotDevice, LetPotDeviceStatus

 from homeassistant.core import HomeAssistant
-from homeassistant.exceptions import ConfigEntryError
+from homeassistant.exceptions import ConfigEntryAuthFailed
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed

 from .const import REQUEST_UPDATE_TIMEOUT
@ -52,7 +52,7 @@ class LetPotDeviceCoordinator(DataUpdateCoordinator[LetPotDeviceStatus]):
        try:
            await self.device_client.subscribe(self._handle_status_update)
        except LetPotAuthenticationException as exc:
-            raise ConfigEntryError from exc
+            raise ConfigEntryAuthFailed from exc

    async def _async_update_data(self) -> LetPotDeviceStatus:
        """Request an update from the device and wait for a status update or timeout."""
--- a/homeassistant/components/letpot/quality_scale.yaml
+++ b/homeassistant/components/letpot/quality_scale.yaml
@ -43,7 +43,7 @@ rules:
  integration-owner: done
  log-when-unavailable: todo
  parallel-updates: done
-  reauthentication-flow: todo
+  reauthentication-flow: done
  test-coverage: todo

  # Gold
--- a/homeassistant/components/letpot/strings.json
+++ b/homeassistant/components/letpot/strings.json
@ -10,6 +10,15 @@
          "email": "The email address of your LetPot account.",
          "password": "The password of your LetPot account."
        }
+      },
+      "reauth_confirm": {
+        "description": "The LetPot integration needs to re-authenticate your account {email}.",
+        "data": {
+          "password": "[%key:common::config_flow::data::password%]"
+        },
+        "data_description": {
+          "password": "[%key:component::letpot::config::step::user::data_description::password%]"
+        }
      }
    },
    "error": {
@ -18,7 +27,8 @@
      "unknown": "[%key:common::config_flow::error::unknown%]"
    },
    "abort": {
-      "already_configured": "[%key:common::config_flow::abort::already_configured_account%]"
+      "already_configured": "[%key:common::config_flow::abort::already_configured_account%]",
+      "reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
    }
  },
  "entity": {
--- a/tests/components/letpot/test_config_flow.py
+++ b/tests/components/letpot/test_config_flow.py
@ -1,5 +1,6 @@
 """Test the LetPot config flow."""

+import dataclasses
 from typing import Any
 from unittest.mock import AsyncMock, patch

@ -145,3 +146,172 @@ async def test_flow_duplicate(
    assert result["type"] is FlowResultType.ABORT
    assert result["reason"] == "already_configured"
    assert len(mock_setup_entry.mock_calls) == 0
+
+
+async def test_reauth_flow(
+    hass: HomeAssistant, mock_config_entry: MockConfigEntry
+) -> None:
+    """Test reauth flow with success."""
+    mock_config_entry.add_to_hass(hass)
+
+    result = await mock_config_entry.start_reauth_flow(hass)
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "reauth_confirm"
+
+    updated_auth = dataclasses.replace(
+        AUTHENTICATION,
+        access_token="new_access_token",
+        refresh_token="new_refresh_token",
+    )
+    with patch(
+        "homeassistant.components.letpot.config_flow.LetPotClient.login",
+        return_value=updated_auth,
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {CONF_PASSWORD: "new-password"},
+        )
+        await hass.async_block_till_done()
+
+    assert result["type"] is FlowResultType.ABORT
+    assert result["reason"] == "reauth_successful"
+    assert mock_config_entry.data == {
+        CONF_ACCESS_TOKEN: "new_access_token",
+        CONF_ACCESS_TOKEN_EXPIRES: AUTHENTICATION.access_token_expires,
+        CONF_REFRESH_TOKEN: "new_refresh_token",
+        CONF_REFRESH_TOKEN_EXPIRES: AUTHENTICATION.refresh_token_expires,
+        CONF_USER_ID: AUTHENTICATION.user_id,
+        CONF_EMAIL: AUTHENTICATION.email,
+    }
+    assert len(hass.config_entries.async_entries()) == 1
+
+
+@pytest.mark.parametrize(
+    ("exception", "error"),
+    [
+        (LetPotAuthenticationException, "invalid_auth"),
+        (LetPotConnectionException, "cannot_connect"),
+        (Exception, "unknown"),
+    ],
+)
+async def test_reauth_exceptions(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+    exception: Exception,
+    error: str,
+) -> None:
+    """Test reauth flow with exception during login and recovery."""
+    mock_config_entry.add_to_hass(hass)
+
+    result = await mock_config_entry.start_reauth_flow(hass)
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "reauth_confirm"
+
+    with patch(
+        "homeassistant.components.letpot.config_flow.LetPotClient.login",
+        side_effect=exception,
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {CONF_PASSWORD: "new-password"},
+        )
+
+    assert result["type"] is FlowResultType.FORM
+    assert result["errors"] == {"base": error}
+
+    # Retry to show recovery.
+    updated_auth = dataclasses.replace(
+        AUTHENTICATION,
+        access_token="new_access_token",
+        refresh_token="new_refresh_token",
+    )
+    with patch(
+        "homeassistant.components.letpot.config_flow.LetPotClient.login",
+        return_value=updated_auth,
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {CONF_PASSWORD: "new-password"},
+        )
+        await hass.async_block_till_done()
+
+    assert result["type"] is FlowResultType.ABORT
+    assert result["reason"] == "reauth_successful"
+    assert mock_config_entry.data == {
+        CONF_ACCESS_TOKEN: "new_access_token",
+        CONF_ACCESS_TOKEN_EXPIRES: AUTHENTICATION.access_token_expires,
+        CONF_REFRESH_TOKEN: "new_refresh_token",
+        CONF_REFRESH_TOKEN_EXPIRES: AUTHENTICATION.refresh_token_expires,
+        CONF_USER_ID: AUTHENTICATION.user_id,
+        CONF_EMAIL: AUTHENTICATION.email,
+    }
+    assert len(hass.config_entries.async_entries()) == 1
+
+
+async def test_reauth_different_user_id_new(
+    hass: HomeAssistant, mock_config_entry: MockConfigEntry
+) -> None:
+    """Test reauth flow with different, new user ID updating the existing entry."""
+    mock_config_entry.add_to_hass(hass)
+    config_entries = hass.config_entries.async_entries()
+    assert len(config_entries) == 1
+    assert config_entries[0].unique_id == AUTHENTICATION.user_id
+
+    result = await mock_config_entry.start_reauth_flow(hass)
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "reauth_confirm"
+
+    updated_auth = dataclasses.replace(AUTHENTICATION, user_id="new_user_id")
+    with patch(
+        "homeassistant.components.letpot.config_flow.LetPotClient.login",
+        return_value=updated_auth,
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {CONF_PASSWORD: "new-password"},
+        )
+        await hass.async_block_till_done()
+
+    assert result["type"] is FlowResultType.ABORT
+    assert result["reason"] == "reauth_successful"
+    assert mock_config_entry.data == {
+        CONF_ACCESS_TOKEN: AUTHENTICATION.access_token,
+        CONF_ACCESS_TOKEN_EXPIRES: AUTHENTICATION.access_token_expires,
+        CONF_REFRESH_TOKEN: AUTHENTICATION.refresh_token,
+        CONF_REFRESH_TOKEN_EXPIRES: AUTHENTICATION.refresh_token_expires,
+        CONF_USER_ID: "new_user_id",
+        CONF_EMAIL: AUTHENTICATION.email,
+    }
+    config_entries = hass.config_entries.async_entries()
+    assert len(config_entries) == 1
+    assert config_entries[0].unique_id == "new_user_id"
+
+
+async def test_reauth_different_user_id_existing(
+    hass: HomeAssistant, mock_config_entry: MockConfigEntry
+) -> None:
+    """Test reauth flow with different, existing user ID aborting."""
+    mock_config_entry.add_to_hass(hass)
+    mock_other = MockConfigEntry(
+        domain=DOMAIN, title="email2@example.com", data={}, unique_id="other_user_id"
+    )
+    mock_other.add_to_hass(hass)
+
+    result = await mock_config_entry.start_reauth_flow(hass)
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "reauth_confirm"
+
+    updated_auth = dataclasses.replace(AUTHENTICATION, user_id="other_user_id")
+    with patch(
+        "homeassistant.components.letpot.config_flow.LetPotClient.login",
+        return_value=updated_auth,
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {CONF_PASSWORD: "new-password"},
+        )
+        await hass.async_block_till_done()
+
+    assert result["type"] is FlowResultType.ABORT
+    assert result["reason"] == "already_configured"
+    assert len(hass.config_entries.async_entries()) == 2