Add 2fa support in picnic integration (#167636)

homeassistant/components/picnic/config_flow.py

@@ -7,7 +7,11 @@ import logging
 from typing import Any
 
 from python_picnic_api2 import PicnicAPI
-from python_picnic_api2.session import PicnicAuthError
+from python_picnic_api2.session import (
+    Picnic2FAError,
+    Picnic2FARequired,
+    PicnicAuthError,
+)
 import requests
 import voluptuous as vol
 
@@ -18,13 +22,19 @@ from homeassistant.const import (
     CONF_PASSWORD,
     CONF_USERNAME,
 )
-from homeassistant.core import HomeAssistant
-from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers.selector import (
+    SelectSelector,
+    SelectSelectorConfig,
+    SelectSelectorMode,
+)
 
-from .const import COUNTRY_CODES, DOMAIN
+from .const import COUNTRY_CODES, DOMAIN, TWO_FA_CHANNELS
 
 _LOGGER = logging.getLogger(__name__)
 
+CONF_2FA_CODE = "two_fa_code"
+CONF_2FA_CHANNEL = "two_fa_channel"
+
 STEP_USER_DATA_SCHEMA = vol.Schema(
     {
         vol.Required(CONF_USERNAME): str,
@@ -35,45 +45,23 @@ STEP_USER_DATA_SCHEMA = vol.Schema(
     }
 )
 
-
-class PicnicHub:
-    """Hub class to test user authentication."""
-
-    @staticmethod
-    def authenticate(username, password, country_code) -> tuple[str, dict]:
-        """Test if we can authenticate with the Picnic API."""
-        picnic = PicnicAPI(username, password, country_code)
-        return picnic.session.auth_token, picnic.get_user()
-
-
-async def validate_input(hass: HomeAssistant, data):
-    """Validate the user input allows us to connect.
-
-    Data has the keys from STEP_USER_DATA_SCHEMA with values provided by the user.
-    """
-    hub = PicnicHub()
-
-    try:
-        auth_token, user_data = await hass.async_add_executor_job(
-            hub.authenticate,
-            data[CONF_USERNAME],
-            data[CONF_PASSWORD],
-            data[CONF_COUNTRY_CODE],
-        )
-    except requests.exceptions.ConnectionError as error:
-        raise CannotConnect from error
-    except PicnicAuthError as error:
-        raise InvalidAuth from error
-
-    # Return the validation result
-    address = (
-        f"{user_data['address']['street']} {user_data['address']['house_number']}"
-        f"{user_data['address']['house_number_ext']}"
-    )
-    return auth_token, {
-        "title": address,
-        "unique_id": user_data["user_id"],
+STEP_2FA_CHANNEL_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_2FA_CHANNEL, default=TWO_FA_CHANNELS[0]): SelectSelector(
+            SelectSelectorConfig(
+                options=TWO_FA_CHANNELS,
+                mode=SelectSelectorMode.LIST,
+                translation_key="two_fa_channel",
+            )
+        ),
     }
+)
+
+STEP_2FA_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_2FA_CODE): str,
+    }
+)
 
 
 class PicnicConfigFlow(ConfigFlow, domain=DOMAIN):
@@ -81,6 +69,11 @@ class PicnicConfigFlow(ConfigFlow, domain=DOMAIN):
 
     VERSION = 1
 
+    def __init__(self) -> None:
+        """Initialize the config flow."""
+        self._picnic: PicnicAPI | None = None
+        self._user_input: dict[str, Any] = {}
+
     async def async_step_reauth(
         self, entry_data: Mapping[str, Any]
     ) -> ConfigFlowResult:
@@ -90,7 +83,7 @@ class PicnicConfigFlow(ConfigFlow, domain=DOMAIN):
     async def async_step_user(
         self, user_input: dict[str, Any] | None = None
     ) -> ConfigFlowResult:
-        """Handle the authentication step, this is the generic step for both `step_user` and `step_reauth`."""
+        """Handle the authentication step."""
         if user_input is None:
             return self.async_show_form(
                 step_id="user", data_schema=STEP_USER_DATA_SCHEMA
@@ -99,43 +92,122 @@ class PicnicConfigFlow(ConfigFlow, domain=DOMAIN):
         errors = {}
 
         try:
-            auth_token, info = await validate_input(self.hass, user_input)
-        except CannotConnect:
+            await self.hass.async_add_executor_job(
+                self._start_login,
+                user_input[CONF_USERNAME],
+                user_input[CONF_PASSWORD],
+                user_input[CONF_COUNTRY_CODE],
+            )
+        except Picnic2FARequired:
+            self._user_input = user_input
+            return await self.async_step_2fa_channel()
+        except requests.exceptions.ConnectionError:
             errors["base"] = "cannot_connect"
-        except InvalidAuth:
+        except PicnicAuthError:
             errors["base"] = "invalid_auth"
         except Exception:
             _LOGGER.exception("Unexpected exception")
             errors["base"] = "unknown"
         else:
-            data = {
-                CONF_ACCESS_TOKEN: auth_token,
-                CONF_COUNTRY_CODE: user_input[CONF_COUNTRY_CODE],
-            }
-            existing_entry = await self.async_set_unique_id(info["unique_id"])
-
-            # Abort if we're adding a new config and the unique id is already in use, else create the entry
-            if self.source != SOURCE_REAUTH:
-                self._abort_if_unique_id_configured()
-                return self.async_create_entry(title="Picnic", data=data)
-
-            # In case of re-auth, only continue if an exiting account exists with the same unique id
-            if existing_entry:
-                self.hass.config_entries.async_update_entry(existing_entry, data=data)
-                await self.hass.config_entries.async_reload(existing_entry.entry_id)
-                return self.async_abort(reason="reauth_successful")
-
-            # Set the error because the account is different
-            errors["base"] = "different_account"
+            return await self._async_finish(user_input)
 
         return self.async_show_form(
             step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
         )
 
+    def _start_login(self, username: str, password: str, country_code: str) -> None:
+        self._picnic = PicnicAPI(country_code=country_code)
+        self._picnic.login(username, password)
 
-class CannotConnect(HomeAssistantError):
-    """Error to indicate we cannot connect."""
+    async def async_step_2fa_channel(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Let the user pick the 2FA delivery channel."""
+        assert self._picnic is not None
 
+        if user_input is None:
+            return self.async_show_form(
+                step_id="2fa_channel", data_schema=STEP_2FA_CHANNEL_SCHEMA
+            )
 
-class InvalidAuth(HomeAssistantError):
-    """Error to indicate there is invalid auth."""
+        errors = {}
+        channel = user_input[CONF_2FA_CHANNEL].upper()
+        try:
+            await self.hass.async_add_executor_job(
+                self._picnic.generate_2fa_code, channel
+            )
+        except requests.exceptions.ConnectionError:
+            errors["base"] = "cannot_connect"
+        except Exception:
+            _LOGGER.exception("Failed to request 2FA code via %s", channel)
+            errors["base"] = "unknown"
+        else:
+            return await self.async_step_2fa()
+
+        return self.async_show_form(
+            step_id="2fa_channel",
+            data_schema=STEP_2FA_CHANNEL_SCHEMA,
+            errors=errors,
+        )
+
+    async def async_step_2fa(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle the 2FA verification step."""
+        assert self._picnic is not None
+
+        if user_input is None:
+            return self.async_show_form(step_id="2fa", data_schema=STEP_2FA_SCHEMA)
+
+        errors = {}
+
+        try:
+            await self.hass.async_add_executor_job(
+                self._picnic.verify_2fa_code, user_input[CONF_2FA_CODE]
+            )
+        except Picnic2FAError:
+            errors["base"] = "invalid_2fa_code"
+        except requests.exceptions.ConnectionError:
+            errors["base"] = "cannot_connect"
+        except Exception:
+            _LOGGER.exception("Unexpected exception during 2FA verification")
+            errors["base"] = "unknown"
+        else:
+            return await self._async_finish(self._user_input)
+
+        return self.async_show_form(
+            step_id="2fa", data_schema=STEP_2FA_SCHEMA, errors=errors
+        )
+
+    async def _async_finish(
+        self,
+        user_input: dict[str, Any],
+    ) -> ConfigFlowResult:
+        """Finalize the config entry after successful authentication."""
+        assert self._picnic is not None
+
+        auth_token = self._picnic.session.auth_token
+        user_data = await self.hass.async_add_executor_job(self._picnic.get_user)
+
+        data = {
+            CONF_ACCESS_TOKEN: auth_token,
+            CONF_COUNTRY_CODE: user_input[CONF_COUNTRY_CODE],
+        }
+        existing_entry = await self.async_set_unique_id(user_data["user_id"])
+
+        # Abort if we're adding a new config and the unique id is already in use, else create the entry
+        if self.source != SOURCE_REAUTH:
+            self._abort_if_unique_id_configured()
+            return self.async_create_entry(title="Picnic", data=data)
+
+        # In case of re-auth, only continue if an exiting account exists with the same unique id
+        if existing_entry:
+            self.hass.config_entries.async_update_entry(existing_entry, data=data)
+            await self.hass.config_entries.async_reload(existing_entry.entry_id)
+            return self.async_abort(reason="reauth_successful")
+
+        return self.async_show_form(
+            step_id="user",
+            data_schema=STEP_USER_DATA_SCHEMA,
+            errors={"base": "different_account"},
+        )

homeassistant/components/picnic/const.py

@@ -12,6 +12,7 @@ ATTR_AMOUNT = "amount"
 ATTR_PRODUCT_IDENTIFIERS = "product_identifiers"
 
 COUNTRY_CODES = ["NL", "DE", "BE", "FR"]
+TWO_FA_CHANNELS = ["sms", "email"]
 ATTRIBUTION = "Data provided by Picnic"
 ADDRESS = "address"
 CART_DATA = "cart_data"

homeassistant/components/picnic/strings.json

@@ -7,10 +7,25 @@
     "error": {
       "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
       "different_account": "Account should be the same as used for setting up the integration",
+      "invalid_2fa_code": "The verification code is incorrect or has expired.",
       "invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
       "unknown": "[%key:common::config_flow::error::unknown%]"
     },
     "step": {
+      "2fa": {
+        "data": {
+          "two_fa_code": "Verification code"
+        },
+        "description": "A verification code has been sent to you via your selected channel.",
+        "title": "Two-factor authentication"
+      },
+      "2fa_channel": {
+        "data": {
+          "two_fa_channel": "Channel"
+        },
+        "description": "A second factor is required to complete the login. Select the channel through which you want to receive your second factor.",
+        "title": "Two-factor authentication"
+      },
       "user": {
         "data": {
           "country_code": "Country code",
@@ -77,6 +92,14 @@
       }
     }
   },
+  "selector": {
+    "two_fa_channel": {
+      "options": {
+        "email": "Email",
+        "sms": "Text message (SMS)"
+      }
+    }
+  },
   "services": {
     "add_product": {
       "description": "Adds a product to the cart based on a search string or product ID. The search string and product ID are exclusive.",

tests/components/picnic/test_config_flow.py

@@ -3,7 +3,11 @@
 from unittest.mock import patch
 
 import pytest
-from python_picnic_api2.session import PicnicAuthError
+from python_picnic_api2.session import (
+    Picnic2FAError,
+    Picnic2FARequired,
+    PicnicAuthError,
+)
 import requests
 
 from homeassistant import config_entries
@@ -30,8 +34,12 @@ def picnic_api():
     with patch(
         "homeassistant.components.picnic.config_flow.PicnicAPI",
     ) as picnic_mock:
-        picnic_mock().session.auth_token = auth_token
-        picnic_mock().get_user.return_value = auth_data
+        instance = picnic_mock.return_value
+        instance.session.auth_token = auth_token
+        instance.get_user.return_value = auth_data
+        instance.login.return_value = None  # no 2FA by default
+        instance.generate_2fa_code.return_value = None
+        instance.verify_2fa_code.return_value = None
 
         yield picnic_mock
 
@@ -69,17 +77,19 @@ async def test_form(hass: HomeAssistant, picnic_api) -> None:
     assert len(mock_setup_entry.mock_calls) == 1
 
 
-async def test_form_invalid_auth(hass: HomeAssistant) -> None:
-    """Test we handle invalid authentication."""
+async def test_form_2fa_required(hass: HomeAssistant, picnic_api) -> None:
+    """Test the full 2FA flow."""
+    picnic_api.return_value.login.side_effect = Picnic2FARequired
+
     result = await hass.config_entries.flow.async_init(
         DOMAIN, context={"source": config_entries.SOURCE_USER}
     )
 
     with patch(
-        "homeassistant.components.picnic.config_flow.PicnicHub.authenticate",
-        side_effect=PicnicAuthError,
+        "homeassistant.components.picnic.async_setup_entry",
+        return_value=True,
     ):
-        result2 = await hass.config_entries.flow.async_configure(
+        result_step_user = await hass.config_entries.flow.async_configure(
             result["flow_id"],
             {
                 "username": "test-username",
@@ -87,52 +97,287 @@ async def test_form_invalid_auth(hass: HomeAssistant) -> None:
                 "country_code": "NL",
             },
         )
+        assert result_step_user["type"] is FlowResultType.FORM
+        assert result_step_user["step_id"] == "2fa_channel"
+
+        result_step_2fa_channel = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {"two_fa_channel": "sms"},
+        )
+        assert result_step_2fa_channel["type"] is FlowResultType.FORM
+        assert result_step_2fa_channel["step_id"] == "2fa"
+
+        result_step_2fa_verify = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {"two_fa_code": "123456"},
+        )
+        await hass.async_block_till_done()
+
+    assert result_step_2fa_verify["type"] is FlowResultType.CREATE_ENTRY
+    assert result_step_2fa_verify["title"] == "Picnic"
+    assert result_step_2fa_verify["data"] == {
+        CONF_ACCESS_TOKEN: picnic_api().session.auth_token,
+        CONF_COUNTRY_CODE: "NL",
+    }
+    assert picnic_api.return_value.generate_2fa_code.call_count == 1
+    assert picnic_api.return_value.generate_2fa_code.call_args[0] == ("SMS",)
+    assert picnic_api.return_value.verify_2fa_code.call_count == 1
+    assert picnic_api.return_value.verify_2fa_code.call_args[0] == ("123456",)
+
+
+async def test_form_2fa_channel_cannot_connect(hass: HomeAssistant, picnic_api) -> None:
+    """Test we handle connection errors in the first 2fa step."""
+    picnic_api.return_value.login.side_effect = Picnic2FARequired
+    picnic_api.return_value.generate_2fa_code.side_effect = (
+        requests.exceptions.ConnectionError
+    )
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+
+    with patch(
+        "homeassistant.components.picnic.async_setup_entry",
+        return_value=True,
+    ):
+        result_step_user = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {
+                "username": "test-username",
+                "password": "test-password",
+                "country_code": "NL",
+            },
+        )
+        assert result_step_user["type"] is FlowResultType.FORM
+        assert result_step_user["step_id"] == "2fa_channel"
+
+        result_step_2fa_channel = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {"two_fa_channel": "sms"},
+        )
+        await hass.async_block_till_done()
+
+    assert result_step_2fa_channel["type"] is FlowResultType.FORM
+    assert result_step_2fa_channel["errors"] == {"base": "cannot_connect"}
+
+
+async def test_form_2fa_channel_exception(hass: HomeAssistant, picnic_api) -> None:
+    """Test we handle random exceptions in the first 2fa step."""
+    picnic_api.return_value.login.side_effect = Picnic2FARequired
+    picnic_api.return_value.generate_2fa_code.side_effect = Exception
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+
+    with patch(
+        "homeassistant.components.picnic.async_setup_entry",
+        return_value=True,
+    ):
+        result_step_user = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {
+                "username": "test-username",
+                "password": "test-password",
+                "country_code": "NL",
+            },
+        )
+        assert result_step_user["type"] is FlowResultType.FORM
+        assert result_step_user["step_id"] == "2fa_channel"
+
+        result_step_2fa_channel = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {"two_fa_channel": "sms"},
+        )
+        await hass.async_block_till_done()
+
+    assert result_step_2fa_channel["type"] is FlowResultType.FORM
+    assert result_step_2fa_channel["errors"] == {"base": "unknown"}
+
+
+async def test_form_2fa_wrong_code(hass: HomeAssistant, picnic_api) -> None:
+    """Test the full 2FA flow with incorrect code."""
+    picnic_api.return_value.login.side_effect = Picnic2FARequired
+    picnic_api.return_value.verify_2fa_code.side_effect = Picnic2FAError
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+
+    with patch(
+        "homeassistant.components.picnic.async_setup_entry",
+        return_value=True,
+    ):
+        result_step_user = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {
+                "username": "test-username",
+                "password": "test-password",
+                "country_code": "NL",
+            },
+        )
+        assert result_step_user["type"] is FlowResultType.FORM
+        assert result_step_user["step_id"] == "2fa_channel"
+
+        result_step_2fa_channel = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {"two_fa_channel": "sms"},
+        )
+        assert result_step_2fa_channel["type"] is FlowResultType.FORM
+        assert result_step_2fa_channel["step_id"] == "2fa"
+
+        result_step_2fa_verify = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {"two_fa_code": "654321"},
+        )
+        await hass.async_block_till_done()
+
+    assert result_step_2fa_verify["type"] is FlowResultType.FORM
+    assert result_step_2fa_verify["errors"] == {"base": "invalid_2fa_code"}
+
+
+async def test_form_2fa_cannot_connect(hass: HomeAssistant, picnic_api) -> None:
+    """Test we handle connection errors in the last 2fa step."""
+    picnic_api.return_value.login.side_effect = Picnic2FARequired
+    picnic_api.return_value.verify_2fa_code.side_effect = (
+        requests.exceptions.ConnectionError
+    )
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+
+    with patch(
+        "homeassistant.components.picnic.async_setup_entry",
+        return_value=True,
+    ):
+        result_step_user = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {
+                "username": "test-username",
+                "password": "test-password",
+                "country_code": "NL",
+            },
+        )
+        assert result_step_user["type"] is FlowResultType.FORM
+        assert result_step_user["step_id"] == "2fa_channel"
+
+        result_step_2fa_channel = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {"two_fa_channel": "sms"},
+        )
+        assert result_step_2fa_channel["type"] is FlowResultType.FORM
+        assert result_step_2fa_channel["step_id"] == "2fa"
+
+        result_step_2fa_verify = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {"two_fa_code": "123456"},
+        )
+        await hass.async_block_till_done()
+
+    assert result_step_2fa_verify["type"] is FlowResultType.FORM
+    assert result_step_2fa_verify["errors"] == {"base": "cannot_connect"}
+
+
+async def test_form_2fa_exception(hass: HomeAssistant, picnic_api) -> None:
+    """Test we handle random exceptions in the last 2fa step."""
+    picnic_api.return_value.login.side_effect = Picnic2FARequired
+    picnic_api.return_value.verify_2fa_code.side_effect = Exception
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+
+    with patch(
+        "homeassistant.components.picnic.async_setup_entry",
+        return_value=True,
+    ):
+        result_step_user = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {
+                "username": "test-username",
+                "password": "test-password",
+                "country_code": "NL",
+            },
+        )
+        assert result_step_user["type"] is FlowResultType.FORM
+        assert result_step_user["step_id"] == "2fa_channel"
+
+        result_step_2fa_channel = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {"two_fa_channel": "sms"},
+        )
+        assert result_step_2fa_channel["type"] is FlowResultType.FORM
+        assert result_step_2fa_channel["step_id"] == "2fa"
+
+        result_step_2fa_verify = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {"two_fa_code": "123456"},
+        )
+        await hass.async_block_till_done()
+
+    assert result_step_2fa_verify["type"] is FlowResultType.FORM
+    assert result_step_2fa_verify["errors"] == {"base": "unknown"}
+
+
+async def test_form_invalid_auth(hass: HomeAssistant, picnic_api) -> None:
+    """Test we handle invalid authentication."""
+    picnic_api.return_value.login.side_effect = PicnicAuthError
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+
+    result2 = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        {
+            "username": "test-username",
+            "password": "test-password",
+            "country_code": "NL",
+        },
+    )
 
     assert result2["type"] is FlowResultType.FORM
     assert result2["errors"] == {"base": "invalid_auth"}
 
 
-async def test_form_cannot_connect(hass: HomeAssistant) -> None:
+async def test_form_cannot_connect(hass: HomeAssistant, picnic_api) -> None:
     """Test we handle connection errors."""
+    picnic_api.return_value.login.side_effect = requests.exceptions.ConnectionError
+
     result = await hass.config_entries.flow.async_init(
         DOMAIN, context={"source": config_entries.SOURCE_USER}
     )
 
-    with patch(
-        "homeassistant.components.picnic.config_flow.PicnicHub.authenticate",
-        side_effect=requests.exceptions.ConnectionError,
-    ):
-        result2 = await hass.config_entries.flow.async_configure(
-            result["flow_id"],
-            {
-                "username": "test-username",
-                "password": "test-password",
-                "country_code": "NL",
-            },
-        )
+    result2 = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        {
+            "username": "test-username",
+            "password": "test-password",
+            "country_code": "NL",
+        },
+    )
 
     assert result2["type"] is FlowResultType.FORM
     assert result2["errors"] == {"base": "cannot_connect"}
 
 
-async def test_form_exception(hass: HomeAssistant) -> None:
+async def test_form_exception(hass: HomeAssistant, picnic_api) -> None:
     """Test we handle random exceptions."""
+    picnic_api.return_value.login.side_effect = Exception
+
     result = await hass.config_entries.flow.async_init(
         DOMAIN, context={"source": config_entries.SOURCE_USER}
     )
 
-    with patch(
-        "homeassistant.components.picnic.config_flow.PicnicHub.authenticate",
-        side_effect=Exception,
-    ):
-        result2 = await hass.config_entries.flow.async_configure(
-            result["flow_id"],
-            {
-                "username": "test-username",
-                "password": "test-password",
-                "country_code": "NL",
-            },
-        )
+    result2 = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        {
+            "username": "test-username",
+            "password": "test-password",
+            "country_code": "NL",
+        },
+    )
 
     assert result2["type"] is FlowResultType.FORM
     assert result2["errors"] == {"base": "unknown"}
@@ -203,8 +448,10 @@ async def test_step_reauth(hass: HomeAssistant, picnic_api) -> None:
     assert len(hass.config_entries.async_entries()) == 1
 
 
-async def test_step_reauth_failed(hass: HomeAssistant) -> None:
+async def test_step_reauth_failed(hass: HomeAssistant, picnic_api) -> None:
     """Test the re-auth flow when authentication fails."""
+    picnic_api.return_value.login.side_effect = PicnicAuthError
+
     # Create a mocked config entry
     user_id = "f29-2a6-o32n"
     conf = {CONF_ACCESS_TOKEN: "a3p98fsen.a39p3fap", CONF_COUNTRY_CODE: "NL"}
@@ -221,19 +468,15 @@ async def test_step_reauth_failed(hass: HomeAssistant) -> None:
     assert result_init["type"] is FlowResultType.FORM
     assert result_init["step_id"] == "user"
 
-    with patch(
-        "homeassistant.components.picnic.config_flow.PicnicHub.authenticate",
-        side_effect=PicnicAuthError,
-    ):
-        result_configure = await hass.config_entries.flow.async_configure(
-            result_init["flow_id"],
-            {
-                "username": "test-username",
-                "password": "test-password",
-                "country_code": "NL",
-            },
-        )
-        await hass.async_block_till_done()
+    result_configure = await hass.config_entries.flow.async_configure(
+        result_init["flow_id"],
+        {
+            "username": "test-username",
+            "password": "test-password",
+            "country_code": "NL",
+        },
+    )
+    await hass.async_block_till_done()
 
     # Check that the returned flow has type form with error set
     assert result_configure["type"] is FlowResultType.FORM