Add re-authemtication flow to AVM FRITZ!Box Call Monitor (#127497)

homeassistant/components/fritzbox_callmonitor/__init__.py

@@ -8,7 +8,7 @@ from requests.exceptions import ConnectionError as RequestsConnectionError
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME
 from homeassistant.core import HomeAssistant
-from homeassistant.exceptions import ConfigEntryNotReady
+from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
 
 from .base import FritzBoxPhonebook
 from .const import CONF_PHONEBOOK, CONF_PREFIXES, PLATFORMS
@@ -42,8 +42,7 @@ async def async_setup_entry(
         )
         return False
     except FritzConnectionException as ex:
-        _LOGGER.error("Invalid authentication: %s", ex)
+        raise ConfigEntryAuthFailed from ex
-        return False
     except RequestsConnectionError as ex:
         _LOGGER.error("Unable to connect to AVM FRITZ!Box call monitor: %s", ex)
         raise ConfigEntryNotReady from ex

homeassistant/components/fritzbox_callmonitor/config_flow.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+from collections.abc import Mapping
 from enum import StrEnum
 from typing import Any, cast
 
@@ -65,6 +66,7 @@ class FritzBoxCallMonitorConfigFlow(ConfigFlow, domain=DOMAIN):
 
     VERSION = 1
 
+    _entry: ConfigEntry
     _host: str
     _port: int
     _username: str
@@ -209,6 +211,69 @@ class FritzBoxCallMonitorConfigFlow(ConfigFlow, domain=DOMAIN):
 
         return self._get_config_entry()
 
+    async def async_step_reauth(
+        self, entry_data: Mapping[str, Any]
+    ) -> ConfigFlowResult:
+        """Handle flow upon an API authentication error."""
+        self._entry = self._get_reauth_entry()
+        self._host = entry_data[CONF_HOST]
+        self._port = entry_data[CONF_PORT]
+        self._username = entry_data[CONF_USERNAME]
+        self._password = entry_data[CONF_PASSWORD]
+        self._phonebook_id = entry_data[CONF_PHONEBOOK]
+
+        return await self.async_step_reauth_confirm()
+
+    def _show_setup_form_reauth_confirm(
+        self, user_input: dict[str, Any], errors: dict[str, str] | None = None
+    ) -> ConfigFlowResult:
+        """Show the reauth form to the user."""
+        default_username = user_input.get(CONF_USERNAME)
+        return self.async_show_form(
+            step_id="reauth_confirm",
+            data_schema=vol.Schema(
+                {
+                    vol.Required(CONF_USERNAME, default=default_username): str,
+                    vol.Required(CONF_PASSWORD): str,
+                }
+            ),
+            description_placeholders={"host": self._host},
+            errors=errors or {},
+        )
+
+    async def async_step_reauth_confirm(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Dialog that informs the user that reauth is required."""
+        if user_input is None:
+            return self._show_setup_form_reauth_confirm(
+                user_input={CONF_USERNAME: self._username}
+            )
+
+        self._username = user_input[CONF_USERNAME]
+        self._password = user_input[CONF_PASSWORD]
+
+        if (
+            error := await self.hass.async_add_executor_job(self._try_connect)
+        ) is not ConnectResult.SUCCESS:
+            return self._show_setup_form_reauth_confirm(
+                user_input=user_input, errors={"base": error}
+            )
+
+        self.hass.config_entries.async_update_entry(
+            self._entry,
+            data={
+                CONF_HOST: self._host,
+                CONF_PORT: self._port,
+                CONF_USERNAME: self._username,
+                CONF_PASSWORD: self._password,
+                CONF_PHONEBOOK: self._phonebook_id,
+                SERIAL_NUMBER: self._serial_number,
+            },
+        )
+        await self.hass.config_entries.async_reload(self._entry.entry_id)
+        return self.async_abort(reason="reauth_successful")
+
 
 class FritzBoxCallMonitorOptionsFlowHandler(OptionsFlow):
     """Handle a fritzbox_callmonitor options flow."""

homeassistant/components/fritzbox_callmonitor/strings.json

@@ -17,14 +17,22 @@
         "data": {
           "phonebook": "Phonebook"
         }
+      },
+      "reauth_confirm": {
+        "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%]",
       "no_devices_found": "[%key:common::config_flow::abort::no_devices_found%]",
-      "insufficient_permissions": "User has insufficient permissions to access AVM FRITZ!Box settings and its phonebooks."
+      "insufficient_permissions": "User has insufficient permissions to access AVM FRITZ!Box settings and its phonebooks.",
+      "reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
     },
     "error": {
+      "insufficient_permissions": "[%key:component::fritzbox_callmonitor::config::abort::insufficient_permissions%]",
       "invalid_auth": "[%key:common::config_flow::error::invalid_auth%]"
     }
   },

tests/components/fritzbox_callmonitor/test_config_flow.py

@@ -264,6 +264,97 @@ async def test_setup_invalid_auth(
     assert result["errors"] == {"base": ConnectResult.INVALID_AUTH}
 
 
+async def test_reauth_successful(hass: HomeAssistant) -> None:
+    """Test starting a reauthentication flow."""
+    mock_config = MockConfigEntry(domain=DOMAIN, data=MOCK_CONFIG_ENTRY)
+    mock_config.add_to_hass(hass)
+    result = await mock_config.start_reauth_flow(hass)
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "reauth_confirm"
+
+    with (
+        patch(
+            "homeassistant.components.fritzbox_callmonitor.base.FritzPhonebook.__init__",
+            return_value=None,
+        ),
+        patch(
+            "homeassistant.components.fritzbox_callmonitor.base.FritzPhonebook.phonebook_ids",
+            new_callable=PropertyMock,
+            return_value=[0],
+        ),
+        patch(
+            "homeassistant.components.fritzbox_callmonitor.base.FritzPhonebook.phonebook_info",
+            return_value=MOCK_PHONEBOOK_INFO_1,
+        ),
+        patch(
+            "homeassistant.components.fritzbox_callmonitor.base.FritzPhonebook.modelname",
+            return_value=MOCK_PHONEBOOK_NAME_1,
+        ),
+        patch(
+            "homeassistant.components.fritzbox_callmonitor.config_flow.FritzConnection.__init__",
+            return_value=None,
+        ),
+        patch(
+            "homeassistant.components.fritzbox_callmonitor.config_flow.FritzConnection.updatecheck",
+            new_callable=PropertyMock,
+            return_value=MOCK_DEVICE_INFO,
+        ),
+        patch(
+            "homeassistant.components.fritzbox_callmonitor.async_setup_entry",
+            return_value=True,
+        ) as mock_setup_entry,
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            user_input={
+                CONF_USERNAME: "other_fake_user",
+                CONF_PASSWORD: "other_fake_password",
+            },
+        )
+
+        assert result["type"] is FlowResultType.ABORT
+        assert result["reason"] == "reauth_successful"
+        assert mock_config.data == {
+            **MOCK_CONFIG_ENTRY,
+            CONF_USERNAME: "other_fake_user",
+            CONF_PASSWORD: "other_fake_password",
+        }
+        assert len(mock_setup_entry.mock_calls) == 1
+
+
+@pytest.mark.parametrize(
+    ("side_effect", "error"),
+    [
+        (FritzConnectionException, ConnectResult.INVALID_AUTH),
+        (FritzSecurityError, ConnectResult.INSUFFICIENT_PERMISSIONS),
+    ],
+)
+async def test_reauth_not_successful(
+    hass: HomeAssistant, side_effect: Exception, error: str
+) -> None:
+    """Test starting a reauthentication flow but no connection found."""
+    mock_config = MockConfigEntry(domain=DOMAIN, data=MOCK_CONFIG_ENTRY)
+    mock_config.add_to_hass(hass)
+    result = await mock_config.start_reauth_flow(hass)
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "reauth_confirm"
+
+    with patch(
+        "homeassistant.components.fritzbox_callmonitor.base.FritzPhonebook.__init__",
+        side_effect=side_effect,
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            user_input={
+                CONF_USERNAME: "other_fake_user",
+                CONF_PASSWORD: "other_fake_password",
+            },
+        )
+        assert result["type"] is FlowResultType.FORM
+        assert result["step_id"] == "reauth_confirm"
+        assert result["errors"]["base"] == error
+
+
 async def test_options_flow_correct_prefixes(hass: HomeAssistant) -> None:
     """Test config flow options."""