Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-04-02 00:20:30 +01:00
Code Activity

Hive: Fix bug in config flow for authentication and device registration (#165061)

Browse Source
This commit is contained in:
Khole
2026-03-16 20:07:34 +00:00
committed by GitHub
parent c25a664365
commit 41c497c49e
2 changed files with 108 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
homeassistant/components/hive/config_flow.py
View File

@@ -3,6 +3,7 @@
from __future__ import annotations
from collections.abc import Mapping
import logging
from typing import Any
from apyhiveapi import Auth
@@ -26,6 +27,8 @@ from homeassistant.core import callback
from . import HiveConfigEntry
from .const import CONF_CODE, CONF_DEVICE_NAME, CONFIG_ENTRY_VERSION, DOMAIN
_LOGGER = logging.getLogger(__name__)
class HiveFlowHandler(ConfigFlow, domain=DOMAIN):
"""Handle a Hive config flow."""
@@ -36,7 +39,7 @@ class HiveFlowHandler(ConfigFlow, domain=DOMAIN):
def __init__(self) -> None:
"""Initialize the config flow."""
self.data: dict[str, Any] = {}
self.tokens: dict[str, str] = {}
self.tokens: dict[str, Any] = {}
self.device_registration: bool = False
self.device_name = "Home Assistant"
@@ -67,11 +70,22 @@ class HiveFlowHandler(ConfigFlow, domain=DOMAIN):
except HiveApiError:
errors["base"] = "no_internet_available"
if (
auth_result := self.tokens.get("AuthenticationResult", {})
) and auth_result.get("NewDeviceMetadata"):
_LOGGER.debug("Login successful, New device detected")
self.device_registration = True
return await self.async_step_configuration()
if self.tokens.get("ChallengeName") == "SMS_MFA":
_LOGGER.debug("Login successful, SMS 2FA required")
# Complete SMS 2FA.
return await self.async_step_2fa()
if not errors:
_LOGGER.debug(
"Login successful, no new device detected, no 2FA required"
)
# Complete the entry.
try:
return await self.async_setup_hive_entry()
@@ -103,6 +117,7 @@ class HiveFlowHandler(ConfigFlow, domain=DOMAIN):
errors["base"] = "no_internet_available"
if not errors:
_LOGGER.debug("2FA successful")
if self.source == SOURCE_REAUTH:
return await self.async_setup_hive_entry()
self.device_registration = True
@@ -119,10 +134,11 @@ class HiveFlowHandler(ConfigFlow, domain=DOMAIN):
if user_input:
if self.device_registration:
_LOGGER.debug("Attempting to register device")
self.device_name = user_input["device_name"]
await self.hive_auth.device_registration(user_input["device_name"])
self.data["device_data"] = await self.hive_auth.get_device_data()
_LOGGER.debug("Device registration successful")
try:
return await self.async_setup_hive_entry()
except UnknownHiveError:
@@ -142,6 +158,7 @@ class HiveFlowHandler(ConfigFlow, domain=DOMAIN):
raise UnknownHiveError
# Setup the config entry
_LOGGER.debug("Setting up Hive entry")
self.data["tokens"] = self.tokens
if self.source == SOURCE_REAUTH:
return self.async_update_reload_and_abort(
@@ -160,6 +177,7 @@ class HiveFlowHandler(ConfigFlow, domain=DOMAIN):
CONF_USERNAME: entry_data[CONF_USERNAME],
CONF_PASSWORD: entry_data[CONF_PASSWORD],
}
_LOGGER.debug("Reauthenticating user")
return await self.async_step_user(data)
@staticmethod

88
tests/components/hive/test_config_flow.py
View File

@@ -74,6 +74,94 @@ async def test_user_flow(hass: HomeAssistant) -> None:
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
async def test_user_flow_with_no_2fa(hass: HomeAssistant) -> None:
"""Test user flow with no 2FA required and device registration."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {}
with patch(
"homeassistant.components.hive.config_flow.Auth.login",
return_value={
"ChallengeName": "SUCCESS",
"AuthenticationResult": {
"RefreshToken": "mock-refresh-token",
"AccessToken": "mock-access-token",
"NewDeviceMetadata": {
"DeviceGroupKey": "mock-device-group-key",
"DeviceKey": "mock-device-key",
},
},
},
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: USERNAME,
CONF_PASSWORD: PASSWORD,
},
)
assert result2["type"] is FlowResultType.FORM
assert result2["step_id"] == "configuration"
assert result2["errors"] == {}
with (
patch(
"homeassistant.components.hive.config_flow.Auth.device_registration",
return_value=True,
),
patch(
"homeassistant.components.hive.config_flow.Auth.get_device_data",
return_value=[
"mock-device-group-key",
"mock-device-key",
"mock-device-password",
],
),
patch(
"homeassistant.components.hive.async_setup_entry",
return_value=True,
) as mock_setup_entry,
):
result3 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_DEVICE_NAME: DEVICE_NAME,
},
)
await hass.async_block_till_done()
assert result3["type"] is FlowResultType.CREATE_ENTRY
assert result3["title"] == USERNAME
assert result3["data"] == {
CONF_USERNAME: USERNAME,
CONF_PASSWORD: PASSWORD,
"tokens": {
"AuthenticationResult": {
"AccessToken": "mock-access-token",
"RefreshToken": "mock-refresh-token",
"NewDeviceMetadata": {
"DeviceGroupKey": "mock-device-group-key",
"DeviceKey": "mock-device-key",
},
},
"ChallengeName": "SUCCESS",
},
"device_data": [
"mock-device-group-key",
"mock-device-key",
"mock-device-password",
],
}
assert len(mock_setup_entry.mock_calls) == 1
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
async def test_user_flow_2fa(hass: HomeAssistant) -> None:
"""Test user flow with 2FA."""
result = await hass.config_entries.flow.async_init(