mirror of
https://github.com/home-assistant/core.git
synced 2025-07-28 15:47:12 +00:00
Support new local token generation method in Overkiz (#143181)
* Initial implementation of new token method for Local API * Improve translations * Update text * Bugfix * Bugfix * Bugfixes * Fixes * Bugfix * Bugfix * Fix * small fix * Fix tests * Refactor token usage in Overkiz config flow tests * Refactor local API configuration flow tests for clarity and update reauthentication logic * Improve comments * Update tests * Update homeassistant/components/overkiz/strings.json Co-authored-by: Josef Zweck <josef@zweck.dev> --------- Co-authored-by: Josef Zweck <josef@zweck.dev>
This commit is contained in:
Mick Vleeshouwer
GitHub
parent
cbb4ff2fd9
commit
6b09fe2377
@ -12,6 +12,7 @@ from pyoverkiz.enums import APIType, OverkizState, UIClass, UIWidget
|
||||
from pyoverkiz.exceptions import (
|
||||
BadCredentialsException,
|
||||
MaintenanceException,
|
||||
NotAuthenticatedException,
|
||||
NotSuchTokenException,
|
||||
TooManyRequestsException,
|
||||
)
|
||||
@ -92,7 +93,11 @@ async def async_setup_entry(hass: HomeAssistant, entry: OverkizDataConfigEntry)
|
||||
scenarios = await client.get_scenarios()
|
||||
else:
|
||||
scenarios = []
|
||||
except (BadCredentialsException, NotSuchTokenException) as exception:
|
||||
except (
|
||||
BadCredentialsException,
|
||||
NotSuchTokenException,
|
||||
NotAuthenticatedException,
|
||||
) as exception:
|
||||
raise ConfigEntryAuthFailed("Invalid authentication") from exception
|
||||
except TooManyRequestsException as exception:
|
||||
raise ConfigEntryNotReady("Too many requests, try again later") from exception
|
||||
|
||||
@ -13,12 +13,12 @@ from pyoverkiz.exceptions import (
|
||||
BadCredentialsException,
|
||||
CozyTouchBadCredentialsException,
|
||||
MaintenanceException,
|
||||
NotAuthenticatedException,
|
||||
NotSuchTokenException,
|
||||
TooManyAttemptsBannedException,
|
||||
TooManyRequestsException,
|
||||
UnknownUserException,
|
||||
)
|
||||
from pyoverkiz.models import OverkizServer
|
||||
from pyoverkiz.obfuscate import obfuscate_id
|
||||
from pyoverkiz.utils import generate_local_server, is_overkiz_gateway
|
||||
import voluptuous as vol
|
||||
@ -31,7 +31,6 @@ from homeassistant.const import (
|
||||
CONF_USERNAME,
|
||||
CONF_VERIFY_SSL,
|
||||
)
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
from homeassistant.helpers.aiohttp_client import async_create_clientsession
|
||||
from homeassistant.helpers.service_info.dhcp import DhcpServiceInfo
|
||||
from homeassistant.helpers.service_info.zeroconf import ZeroconfServiceInfo
|
||||
@ -39,15 +38,12 @@ from homeassistant.helpers.service_info.zeroconf import ZeroconfServiceInfo
|
||||
from .const import CONF_API_TYPE, CONF_HUB, DEFAULT_SERVER, DOMAIN, LOGGER
|
||||
|
||||
|
||||
class DeveloperModeDisabled(HomeAssistantError):
|
||||
"""Error to indicate Somfy Developer Mode is disabled."""
|
||||
|
||||
|
||||
class OverkizConfigFlow(ConfigFlow, domain=DOMAIN):
|
||||
"""Handle a config flow for Overkiz (by Somfy)."""
|
||||
|
||||
VERSION = 1
|
||||
|
||||
_verify_ssl: bool = True
|
||||
_api_type: APIType = APIType.CLOUD
|
||||
_user: str | None = None
|
||||
_server: str = DEFAULT_SERVER
|
||||
@ -57,27 +53,36 @@ class OverkizConfigFlow(ConfigFlow, domain=DOMAIN):
|
||||
"""Validate user credentials."""
|
||||
user_input[CONF_API_TYPE] = self._api_type
|
||||
|
||||
client = self._create_cloud_client(
|
||||
if self._api_type == APIType.LOCAL:
|
||||
user_input[CONF_VERIFY_SSL] = self._verify_ssl
|
||||
session = async_create_clientsession(
|
||||
self.hass, verify_ssl=user_input[CONF_VERIFY_SSL]
|
||||
)
|
||||
client = OverkizClient(
|
||||
username="",
|
||||
password="",
|
||||
token=user_input[CONF_TOKEN],
|
||||
session=session,
|
||||
server=generate_local_server(host=user_input[CONF_HOST]),
|
||||
verify_ssl=user_input[CONF_VERIFY_SSL],
|
||||
)
|
||||
else: # APIType.CLOUD
|
||||
session = async_create_clientsession(self.hass)
|
||||
client = OverkizClient(
|
||||
username=user_input[CONF_USERNAME],
|
||||
password=user_input[CONF_PASSWORD],
|
||||
server=SUPPORTED_SERVERS[user_input[CONF_HUB]],
|
||||
session=session,
|
||||
)
|
||||
await client.login(register_event_listener=False)
|
||||
|
||||
# For Local API, we create and activate a local token
|
||||
if self._api_type == APIType.LOCAL:
|
||||
user_input[CONF_TOKEN] = await self._create_local_api_token(
|
||||
cloud_client=client,
|
||||
host=user_input[CONF_HOST],
|
||||
verify_ssl=user_input[CONF_VERIFY_SSL],
|
||||
)
|
||||
await client.login(register_event_listener=False)
|
||||
|
||||
# Set main gateway id as unique id
|
||||
if gateways := await client.get_gateways():
|
||||
for gateway in gateways:
|
||||
if is_overkiz_gateway(gateway.id):
|
||||
gateway_id = gateway.id
|
||||
await self.async_set_unique_id(gateway_id, raise_on_progress=False)
|
||||
await self.async_set_unique_id(gateway.id, raise_on_progress=False)
|
||||
break
|
||||
|
||||
return user_input
|
||||
|
||||
@ -141,15 +146,13 @@ class OverkizConfigFlow(ConfigFlow, domain=DOMAIN):
|
||||
|
||||
if user_input:
|
||||
self._user = user_input[CONF_USERNAME]
|
||||
|
||||
# inherit the server from previous step
|
||||
user_input[CONF_HUB] = self._server
|
||||
|
||||
try:
|
||||
await self.async_validate_input(user_input)
|
||||
except TooManyRequestsException:
|
||||
errors["base"] = "too_many_requests"
|
||||
except BadCredentialsException as exception:
|
||||
except (BadCredentialsException, NotAuthenticatedException) as exception:
|
||||
# If authentication with CozyTouch auth server is valid, but token is invalid
|
||||
# for Overkiz API server, the hardware is not supported.
|
||||
if user_input[CONF_HUB] in {
|
||||
@ -211,16 +214,18 @@ class OverkizConfigFlow(ConfigFlow, domain=DOMAIN):
|
||||
|
||||
if user_input:
|
||||
self._host = user_input[CONF_HOST]
|
||||
self._user = user_input[CONF_USERNAME]
|
||||
|
||||
# inherit the server from previous step
|
||||
self._verify_ssl = user_input[CONF_VERIFY_SSL]
|
||||
user_input[CONF_HUB] = self._server
|
||||
|
||||
try:
|
||||
user_input = await self.async_validate_input(user_input)
|
||||
except TooManyRequestsException:
|
||||
errors["base"] = "too_many_requests"
|
||||
except BadCredentialsException:
|
||||
except (
|
||||
BadCredentialsException,
|
||||
NotSuchTokenException,
|
||||
NotAuthenticatedException,
|
||||
):
|
||||
errors["base"] = "invalid_auth"
|
||||
except ClientConnectorCertificateError as exception:
|
||||
errors["base"] = "certificate_verify_failed"
|
||||
@ -232,10 +237,6 @@ class OverkizConfigFlow(ConfigFlow, domain=DOMAIN):
|
||||
errors["base"] = "server_in_maintenance"
|
||||
except TooManyAttemptsBannedException:
|
||||
errors["base"] = "too_many_attempts"
|
||||
except NotSuchTokenException:
|
||||
errors["base"] = "no_such_token"
|
||||
except DeveloperModeDisabled:
|
||||
errors["base"] = "developer_mode_disabled"
|
||||
except UnknownUserException:
|
||||
# Somfy Protect accounts are not supported since they don't use
|
||||
# the Overkiz API server. Login will return unknown user.
|
||||
@ -264,9 +265,8 @@ class OverkizConfigFlow(ConfigFlow, domain=DOMAIN):
|
||||
data_schema=vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_HOST, default=self._host): str,
|
||||
vol.Required(CONF_USERNAME, default=self._user): str,
|
||||
vol.Required(CONF_PASSWORD): str,
|
||||
vol.Required(CONF_VERIFY_SSL, default=True): bool,
|
||||
vol.Required(CONF_TOKEN): str,
|
||||
vol.Required(CONF_VERIFY_SSL, default=self._verify_ssl): bool,
|
||||
}
|
||||
),
|
||||
description_placeholders=description_placeholders,
|
||||
@ -320,64 +320,15 @@ class OverkizConfigFlow(ConfigFlow, domain=DOMAIN):
|
||||
self, entry_data: Mapping[str, Any]
|
||||
) -> ConfigFlowResult:
|
||||
"""Handle reauth."""
|
||||
# overkiz entries always have unique IDs
|
||||
# Overkiz entries always have unique IDs
|
||||
self.context["title_placeholders"] = {"gateway_id": cast(str, self.unique_id)}
|
||||
|
||||
self._user = entry_data[CONF_USERNAME]
|
||||
self._server = entry_data[CONF_HUB]
|
||||
self._api_type = entry_data.get(CONF_API_TYPE, APIType.CLOUD)
|
||||
self._server = entry_data[CONF_HUB]
|
||||
|
||||
if self._api_type == APIType.LOCAL:
|
||||
self._host = entry_data[CONF_HOST]
|
||||
self._verify_ssl = entry_data[CONF_VERIFY_SSL]
|
||||
else:
|
||||
self._user = entry_data[CONF_USERNAME]
|
||||
|
||||
return await self.async_step_user(dict(entry_data))
|
||||
|
||||
def _create_cloud_client(
|
||||
self, username: str, password: str, server: OverkizServer
|
||||
) -> OverkizClient:
|
||||
session = async_create_clientsession(self.hass)
|
||||
return OverkizClient(
|
||||
username=username, password=password, server=server, session=session
|
||||
)
|
||||
|
||||
async def _create_local_api_token(
|
||||
self, cloud_client: OverkizClient, host: str, verify_ssl: bool
|
||||
) -> str:
|
||||
"""Create local API token."""
|
||||
# Create session on Somfy cloud server to generate an access token for local API
|
||||
gateways = await cloud_client.get_gateways()
|
||||
|
||||
gateway_id = ""
|
||||
for gateway in gateways:
|
||||
# Overkiz can return multiple gateways, but we only can generate a token
|
||||
# for the main gateway.
|
||||
if is_overkiz_gateway(gateway.id):
|
||||
gateway_id = gateway.id
|
||||
|
||||
developer_mode = await cloud_client.get_setup_option(
|
||||
f"developerMode-{gateway_id}"
|
||||
)
|
||||
|
||||
if developer_mode is None:
|
||||
raise DeveloperModeDisabled
|
||||
|
||||
token = await cloud_client.generate_local_token(gateway_id)
|
||||
await cloud_client.activate_local_token(
|
||||
gateway_id=gateway_id, token=token, label="Home Assistant/local"
|
||||
)
|
||||
|
||||
session = async_create_clientsession(self.hass, verify_ssl=verify_ssl)
|
||||
|
||||
# Local API
|
||||
local_client = OverkizClient(
|
||||
username="",
|
||||
password="",
|
||||
token=token,
|
||||
session=session,
|
||||
server=generate_local_server(host=host),
|
||||
verify_ssl=verify_ssl,
|
||||
)
|
||||
|
||||
await local_client.login()
|
||||
|
||||
return token
|
||||
|
||||
@ -79,7 +79,7 @@ class OverkizDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Device]]):
|
||||
"""Fetch Overkiz data via event listener."""
|
||||
try:
|
||||
events = await self.client.fetch_events()
|
||||
except BadCredentialsException as exception:
|
||||
except (BadCredentialsException, NotAuthenticatedException) as exception:
|
||||
raise ConfigEntryAuthFailed("Invalid authentication.") from exception
|
||||
except TooManyConcurrentRequestsException as exception:
|
||||
raise UpdateFailed("Too many concurrent requests.") from exception
|
||||
@ -98,7 +98,7 @@ class OverkizDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Device]]):
|
||||
try:
|
||||
await self.client.login()
|
||||
self.devices = await self._get_devices()
|
||||
except BadCredentialsException as exception:
|
||||
except (BadCredentialsException, NotAuthenticatedException) as exception:
|
||||
raise ConfigEntryAuthFailed("Invalid authentication.") from exception
|
||||
except TooManyRequestsException as exception:
|
||||
raise UpdateFailed("Too many requests, try again later.") from exception
|
||||
|
||||
@ -32,17 +32,15 @@
|
||||
}
|
||||
},
|
||||
"local": {
|
||||
"description": "By activating the [Developer Mode of your TaHoma box](https://github.com/Somfy-Developer/Somfy-TaHoma-Developer-Mode#getting-started), you can authorize third-party software (like Home Assistant) to connect to it via your local network.\n\nAfter activation, enter your application credentials and change the host to include your Gateway PIN or enter the IP address of your gateway.",
|
||||
"description": "By activating the [Developer Mode of your TaHoma box](https://github.com/Somfy-Developer/Somfy-TaHoma-Developer-Mode#getting-started), you can authorize third-party software (like Home Assistant) to connect to it via your local network.\n\n1. Open the TaHoma By Somfy application on your device.\n2. Navigate to the Help & advanced features -> Advanced features menu in the application.\n3. Activate Developer Mode by tapping 7 times on the version number of your gateway (like 2025.1.4-11).\n4. Generate a token from the Developer Mode menu to authenticate your API calls.\n\n5. Enter the generated token below and update the host to include your Gateway PIN or the IP address of your gateway.",
|
||||
"data": {
|
||||
"host": "[%key:common::config_flow::data::host%]",
|
||||
"username": "[%key:common::config_flow::data::username%]",
|
||||
"password": "[%key:common::config_flow::data::password%]",
|
||||
"token": "[%key:common::config_flow::data::api_token%]",
|
||||
"verify_ssl": "[%key:common::config_flow::data::verify_ssl%]"
|
||||
},
|
||||
"data_description": {
|
||||
"host": "The hostname or IP address of your Overkiz hub.",
|
||||
"username": "The username of your cloud account (app).",
|
||||
"password": "The password of your cloud account (app).",
|
||||
"token": "Token generated by the app used to control your device.",
|
||||
"verify_ssl": "Verify the SSL certificate. Select this only if you are connecting via the hostname."
|
||||
}
|
||||
}
|
||||
|
||||
@ -40,6 +40,7 @@ TEST_GATEWAY_ID3 = "SOMFY_PROTECT-v0NT53occUBPyuJRzx59kalW1hFfzimN"
|
||||
|
||||
TEST_HOST = "gateway-1234-5678-9123.local:8443"
|
||||
TEST_HOST2 = "192.168.11.104:8443"
|
||||
TEST_TOKEN = "1234123412341234"
|
||||
|
||||
MOCK_GATEWAY_RESPONSE = [Mock(id=TEST_GATEWAY_ID)]
|
||||
MOCK_GATEWAY2_RESPONSE = [Mock(id=TEST_GATEWAY_ID3), Mock(id=TEST_GATEWAY_ID2)]
|
||||
@ -152,7 +153,7 @@ async def test_form_only_cloud_supported(
|
||||
async def test_form_local_happy_flow(
|
||||
hass: HomeAssistant, mock_setup_entry: AsyncMock
|
||||
) -> None:
|
||||
"""Test we get the form."""
|
||||
"""Test local API configuration flow."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
@ -179,21 +180,27 @@ async def test_form_local_happy_flow(
|
||||
"pyoverkiz.client.OverkizClient",
|
||||
login=AsyncMock(return_value=True),
|
||||
get_gateways=AsyncMock(return_value=MOCK_GATEWAY_RESPONSE),
|
||||
get_setup_option=AsyncMock(return_value=True),
|
||||
generate_local_token=AsyncMock(return_value="1234123412341234"),
|
||||
activate_local_token=AsyncMock(return_value=True),
|
||||
):
|
||||
await hass.config_entries.flow.async_configure(
|
||||
result4 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
"username": TEST_EMAIL,
|
||||
"password": TEST_PASSWORD,
|
||||
"host": "gateway-1234-5678-1234.local:8443",
|
||||
"token": TEST_TOKEN,
|
||||
"verify_ssl": True,
|
||||
},
|
||||
)
|
||||
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result4["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert result4["title"] == "gateway-1234-5678-1234.local:8443"
|
||||
assert result4["data"] == {
|
||||
"host": "gateway-1234-5678-1234.local:8443",
|
||||
"token": TEST_TOKEN,
|
||||
"verify_ssl": True,
|
||||
"hub": TEST_SERVER,
|
||||
"api_type": "local",
|
||||
}
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
|
||||
|
||||
@ -262,7 +269,7 @@ async def test_form_invalid_auth_cloud(
|
||||
(MaintenanceException, "server_in_maintenance"),
|
||||
(TooManyAttemptsBannedException, "too_many_attempts"),
|
||||
(UnknownUserException, "unsupported_hardware"),
|
||||
(NotSuchTokenException, "no_such_token"),
|
||||
(NotSuchTokenException, "invalid_auth"),
|
||||
(Exception, "unknown"),
|
||||
],
|
||||
)
|
||||
@ -297,8 +304,7 @@ async def test_form_invalid_auth_local(
|
||||
result["flow_id"],
|
||||
{
|
||||
"host": TEST_HOST,
|
||||
"username": TEST_EMAIL,
|
||||
"password": TEST_PASSWORD,
|
||||
"token": TEST_TOKEN,
|
||||
"verify_ssl": True,
|
||||
},
|
||||
)
|
||||
@ -309,52 +315,6 @@ async def test_form_invalid_auth_local(
|
||||
assert result4["errors"] == {"base": error}
|
||||
|
||||
|
||||
async def test_form_local_developer_mode_disabled(
|
||||
hass: HomeAssistant, mock_setup_entry: AsyncMock
|
||||
) -> None:
|
||||
"""Test we get the form."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{"hub": TEST_SERVER},
|
||||
)
|
||||
|
||||
assert result2["type"] is FlowResultType.FORM
|
||||
assert result2["step_id"] == "local_or_cloud"
|
||||
|
||||
result3 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{"api_type": "local"},
|
||||
)
|
||||
|
||||
assert result3["type"] is FlowResultType.FORM
|
||||
assert result3["step_id"] == "local"
|
||||
|
||||
with patch.multiple(
|
||||
"pyoverkiz.client.OverkizClient",
|
||||
login=AsyncMock(return_value=True),
|
||||
get_gateways=AsyncMock(return_value=MOCK_GATEWAY_RESPONSE),
|
||||
get_setup_option=AsyncMock(return_value=None),
|
||||
):
|
||||
result4 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
"username": TEST_EMAIL,
|
||||
"password": TEST_PASSWORD,
|
||||
"host": "gateway-1234-5678-1234.local:8443",
|
||||
"verify_ssl": True,
|
||||
},
|
||||
)
|
||||
|
||||
assert result4["type"] is FlowResultType.FORM
|
||||
assert result4["errors"] == {"base": "developer_mode_disabled"}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("side_effect", "error"),
|
||||
[
|
||||
@ -444,16 +404,18 @@ async def test_cloud_abort_on_duplicate_entry(
|
||||
async def test_local_abort_on_duplicate_entry(
|
||||
hass: HomeAssistant, mock_setup_entry: AsyncMock
|
||||
) -> None:
|
||||
"""Test we get the form."""
|
||||
"""Test local API configuration is aborted if gateway already exists."""
|
||||
|
||||
MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
unique_id=TEST_GATEWAY_ID,
|
||||
version=2,
|
||||
data={
|
||||
"host": TEST_HOST,
|
||||
"username": TEST_EMAIL,
|
||||
"password": TEST_PASSWORD,
|
||||
"token": TEST_TOKEN,
|
||||
"verify_ssl": True,
|
||||
"hub": TEST_SERVER,
|
||||
"api_type": "local",
|
||||
},
|
||||
).add_to_hass(hass)
|
||||
|
||||
@ -484,15 +446,12 @@ async def test_local_abort_on_duplicate_entry(
|
||||
login=AsyncMock(return_value=True),
|
||||
get_gateways=AsyncMock(return_value=MOCK_GATEWAY_RESPONSE),
|
||||
get_setup_option=AsyncMock(return_value=True),
|
||||
generate_local_token=AsyncMock(return_value="1234123412341234"),
|
||||
activate_local_token=AsyncMock(return_value=True),
|
||||
):
|
||||
result4 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
"host": TEST_HOST,
|
||||
"username": TEST_EMAIL,
|
||||
"password": TEST_PASSWORD,
|
||||
"token": TEST_TOKEN,
|
||||
"verify_ssl": True,
|
||||
},
|
||||
)
|
||||
@ -639,18 +598,18 @@ async def test_cloud_reauth_wrong_account(hass: HomeAssistant) -> None:
|
||||
assert result2["reason"] == "reauth_wrong_account"
|
||||
|
||||
|
||||
async def test_local_reauth_success(hass: HomeAssistant) -> None:
|
||||
"""Test reauthentication flow."""
|
||||
|
||||
async def test_local_reauth_legacy(hass: HomeAssistant) -> None:
|
||||
"""Test legacy reauthentication flow with username/password."""
|
||||
mock_entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
unique_id=TEST_GATEWAY_ID,
|
||||
version=2,
|
||||
data={
|
||||
"host": TEST_HOST,
|
||||
"username": TEST_EMAIL,
|
||||
"password": TEST_PASSWORD,
|
||||
"verify_ssl": True,
|
||||
"hub": TEST_SERVER,
|
||||
"host": TEST_HOST,
|
||||
"api_type": "local",
|
||||
},
|
||||
)
|
||||
@ -672,36 +631,85 @@ async def test_local_reauth_success(hass: HomeAssistant) -> None:
|
||||
"pyoverkiz.client.OverkizClient",
|
||||
login=AsyncMock(return_value=True),
|
||||
get_gateways=AsyncMock(return_value=MOCK_GATEWAY_RESPONSE),
|
||||
get_setup_option=AsyncMock(return_value=True),
|
||||
generate_local_token=AsyncMock(return_value="1234123412341234"),
|
||||
activate_local_token=AsyncMock(return_value=True),
|
||||
):
|
||||
result3 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
"username": TEST_EMAIL,
|
||||
"password": TEST_PASSWORD2,
|
||||
{
|
||||
"host": TEST_HOST,
|
||||
"token": "new_token",
|
||||
"verify_ssl": True,
|
||||
},
|
||||
)
|
||||
|
||||
assert result3["type"] is FlowResultType.ABORT
|
||||
assert result3["reason"] == "reauth_successful"
|
||||
assert mock_entry.data["username"] == TEST_EMAIL
|
||||
assert mock_entry.data["password"] == TEST_PASSWORD2
|
||||
assert mock_entry.data["host"] == TEST_HOST
|
||||
assert mock_entry.data["token"] == "new_token"
|
||||
assert mock_entry.data["verify_ssl"] is True
|
||||
|
||||
|
||||
async def test_local_reauth_success(hass: HomeAssistant) -> None:
|
||||
"""Test modern local reauth flow."""
|
||||
mock_entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
unique_id=TEST_GATEWAY_ID,
|
||||
version=2,
|
||||
data={
|
||||
"host": TEST_HOST,
|
||||
"token": "old_token",
|
||||
"verify_ssl": True,
|
||||
"hub": TEST_SERVER,
|
||||
"api_type": "local",
|
||||
},
|
||||
)
|
||||
mock_entry.add_to_hass(hass)
|
||||
|
||||
result = await mock_entry.start_reauth_flow(hass)
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["step_id"] == "local_or_cloud"
|
||||
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{"api_type": "local"},
|
||||
)
|
||||
|
||||
assert result2["step_id"] == "local"
|
||||
|
||||
with patch.multiple(
|
||||
"pyoverkiz.client.OverkizClient",
|
||||
login=AsyncMock(return_value=True),
|
||||
get_gateways=AsyncMock(return_value=MOCK_GATEWAY_RESPONSE),
|
||||
):
|
||||
result3 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
"host": TEST_HOST,
|
||||
"token": "new_token",
|
||||
"verify_ssl": True,
|
||||
},
|
||||
)
|
||||
|
||||
assert result3["type"] is FlowResultType.ABORT
|
||||
assert result3["reason"] == "reauth_successful"
|
||||
assert mock_entry.data["host"] == TEST_HOST
|
||||
assert mock_entry.data["token"] == "new_token"
|
||||
assert mock_entry.data["verify_ssl"] is True
|
||||
assert "username" not in mock_entry.data
|
||||
assert "password" not in mock_entry.data
|
||||
|
||||
|
||||
async def test_local_reauth_wrong_account(hass: HomeAssistant) -> None:
|
||||
"""Test reauthentication flow."""
|
||||
"""Test local reauth flow with wrong gateway account."""
|
||||
|
||||
mock_entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
unique_id=TEST_GATEWAY_ID2,
|
||||
version=2,
|
||||
data={
|
||||
"username": TEST_EMAIL,
|
||||
"password": TEST_PASSWORD,
|
||||
"hub": TEST_SERVER,
|
||||
"host": TEST_HOST,
|
||||
"token": "old_token",
|
||||
"verify_ssl": True,
|
||||
"hub": TEST_SERVER,
|
||||
"api_type": "local",
|
||||
},
|
||||
)
|
||||
@ -722,15 +730,13 @@ async def test_local_reauth_wrong_account(hass: HomeAssistant) -> None:
|
||||
"pyoverkiz.client.OverkizClient",
|
||||
login=AsyncMock(return_value=True),
|
||||
get_gateways=AsyncMock(return_value=MOCK_GATEWAY_RESPONSE),
|
||||
get_setup_option=AsyncMock(return_value=True),
|
||||
generate_local_token=AsyncMock(return_value="1234123412341234"),
|
||||
activate_local_token=AsyncMock(return_value=True),
|
||||
):
|
||||
result3 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
"username": TEST_EMAIL,
|
||||
"password": TEST_PASSWORD2,
|
||||
{
|
||||
"host": TEST_HOST,
|
||||
"token": "new_token",
|
||||
"verify_ssl": True,
|
||||
},
|
||||
)
|
||||
|
||||
@ -897,27 +903,27 @@ async def test_local_zeroconf_flow(
|
||||
"pyoverkiz.client.OverkizClient",
|
||||
login=AsyncMock(return_value=True),
|
||||
get_gateways=AsyncMock(return_value=MOCK_GATEWAY_RESPONSE),
|
||||
get_setup_option=AsyncMock(return_value=True),
|
||||
generate_local_token=AsyncMock(return_value="1234123412341234"),
|
||||
activate_local_token=AsyncMock(return_value=True),
|
||||
):
|
||||
result4 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{"username": TEST_EMAIL, "password": TEST_PASSWORD, "verify_ssl": False},
|
||||
{
|
||||
"host": "gateway-1234-5678-9123.local:8443",
|
||||
"token": TEST_TOKEN,
|
||||
"verify_ssl": False,
|
||||
},
|
||||
)
|
||||
|
||||
assert result4["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert result4["title"] == "gateway-1234-5678-9123.local:8443"
|
||||
assert result4["data"] == {
|
||||
"username": TEST_EMAIL,
|
||||
"password": TEST_PASSWORD,
|
||||
"hub": TEST_SERVER,
|
||||
"host": "gateway-1234-5678-9123.local:8443",
|
||||
"api_type": "local",
|
||||
"token": "1234123412341234",
|
||||
"verify_ssl": False,
|
||||
}
|
||||
|
||||
# Verify no username/password in data
|
||||
assert result4["data"] == {
|
||||
"host": "gateway-1234-5678-9123.local:8443",
|
||||
"token": TEST_TOKEN,
|
||||
"verify_ssl": False,
|
||||
"hub": TEST_SERVER,
|
||||
"api_type": "local",
|
||||
}
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user