home-assistant/core
1
0
Fork 1
mirror of https://github.com/home-assistant/core.git synced 2025-09-10 15:21:38 +02:00
Code Issues Packages Projects Releases Wiki Activity

Fix PG&E and Duquesne Light Company in Opower (#149658)

Browse Source 
Co-authored-by: Norbert Rittel <norbert@rittel.de>
This commit is contained in:
tronikos
2025-08-06 02:32:42 -07:00
committed by GitHub
parent 0db23b0da6
commit 0aeff366bd
8 changed files with 544 additions and 155 deletions
Download Patch File Download Diff File Expand all files Collapse all files

228
homeassistant/components/opower/config_flow.py
View File

@@ -9,6 +9,8 @@ from typing import Any
from opower import (
CannotConnect,
InvalidAuth,
MfaChallenge,
MfaHandlerBase,
Opower,
create_cookie_jar,
get_supported_utility_names,
@@ -16,49 +18,34 @@ from opower import (
)
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.config_entries import SOURCE_REAUTH, ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_NAME, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.aiohttp_client import async_create_clientsession
from homeassistant.helpers.typing import VolDictType
from .const import CONF_TOTP_SECRET, CONF_UTILITY, DOMAIN
from .const import CONF_LOGIN_DATA, CONF_TOTP_SECRET, CONF_UTILITY, DOMAIN
_LOGGER = logging.getLogger(__name__)
STEP_USER_DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_UTILITY): vol.In(get_supported_utility_names()),
vol.Required(CONF_USERNAME): str,
vol.Required(CONF_PASSWORD): str,
}
)
CONF_MFA_CODE = "mfa_code"
CONF_MFA_METHOD = "mfa_method"
async def _validate_login(
hass: HomeAssistant, login_data: dict[str, str]
) -> dict[str, str]:
"""Validate login data and return any errors."""
hass: HomeAssistant,
data: Mapping[str, Any],
) -> None:
"""Validate login data and raise exceptions on failure."""
api = Opower(
async_create_clientsession(hass, cookie_jar=create_cookie_jar()),
login_data[CONF_UTILITY],
login_data[CONF_USERNAME],
login_data[CONF_PASSWORD],
login_data.get(CONF_TOTP_SECRET),
data[CONF_UTILITY],
data[CONF_USERNAME],
data[CONF_PASSWORD],
data.get(CONF_TOTP_SECRET),
data.get(CONF_LOGIN_DATA),
)
errors: dict[str, str] = {}
try:
await api.async_login()
except InvalidAuth:
_LOGGER.exception(
"Invalid auth when connecting to %s", login_data[CONF_UTILITY]
)
errors["base"] = "invalid_auth"
except CannotConnect:
_LOGGER.exception("Could not connect to %s", login_data[CONF_UTILITY])
errors["base"] = "cannot_connect"
return errors
await api.async_login()
class OpowerConfigFlow(ConfigFlow, domain=DOMAIN):
@@ -68,81 +55,147 @@ class OpowerConfigFlow(ConfigFlow, domain=DOMAIN):
def __init__(self) -> None:
"""Initialize a new OpowerConfigFlow."""
self.utility_info: dict[str, Any] | None = None
self._data: dict[str, Any] = {}
self.mfa_handler: MfaHandlerBase | None = None
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the initial step."""
errors: dict[str, str] = {}
"""Handle the initial step (select utility)."""
if user_input is not None:
self._async_abort_entries_match(
{
CONF_UTILITY: user_input[CONF_UTILITY],
CONF_USERNAME: user_input[CONF_USERNAME],
}
)
if select_utility(user_input[CONF_UTILITY]).accepts_mfa():
self.utility_info = user_input
return await self.async_step_mfa()
self._data[CONF_UTILITY] = user_input[CONF_UTILITY]
return await self.async_step_credentials()
errors = await _validate_login(self.hass, user_input)
if not errors:
return self._async_create_opower_entry(user_input)
else:
user_input = {}
user_input.pop(CONF_PASSWORD, None)
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{vol.Required(CONF_UTILITY): vol.In(get_supported_utility_names())}
),
)
async def async_step_credentials(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle credentials step."""
errors: dict[str, str] = {}
utility = select_utility(self._data[CONF_UTILITY])
if user_input is not None:
self._data.update(user_input)
self._async_abort_entries_match(
{
CONF_UTILITY: self._data[CONF_UTILITY],
CONF_USERNAME: self._data[CONF_USERNAME],
}
)
try:
await _validate_login(self.hass, self._data)
except MfaChallenge as exc:
self.mfa_handler = exc.handler
return await self.async_step_mfa_options()
except InvalidAuth:
errors["base"] = "invalid_auth"
except CannotConnect:
errors["base"] = "cannot_connect"
else:
return self._async_create_opower_entry(self._data)
schema_dict: VolDictType = {
vol.Required(CONF_USERNAME): str,
vol.Required(CONF_PASSWORD): str,
}
if utility.accepts_totp_secret():
schema_dict[vol.Optional(CONF_TOTP_SECRET)] = str
return self.async_show_form(
step_id="credentials",
data_schema=self.add_suggested_values_to_schema(
STEP_USER_DATA_SCHEMA, user_input
vol.Schema(schema_dict), user_input
),
errors=errors,
)
async def async_step_mfa(
async def async_step_mfa_options(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle MFA step."""
assert self.utility_info is not None
"""Handle MFA options step."""
errors: dict[str, str] = {}
assert self.mfa_handler is not None
if user_input is not None:
method = user_input[CONF_MFA_METHOD]
try:
await self.mfa_handler.async_select_mfa_option(method)
except CannotConnect:
errors["base"] = "cannot_connect"
else:
return await self.async_step_mfa_code()
mfa_options = await self.mfa_handler.async_get_mfa_options()
if not mfa_options:
return await self.async_step_mfa_code()
return self.async_show_form(
step_id="mfa_options",
data_schema=self.add_suggested_values_to_schema(
vol.Schema({vol.Required(CONF_MFA_METHOD): vol.In(mfa_options)}),
user_input,
),
errors=errors,
)
async def async_step_mfa_code(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle MFA code submission step."""
assert self.mfa_handler is not None
errors: dict[str, str] = {}
if user_input is not None:
data = {**self.utility_info, **user_input}
errors = await _validate_login(self.hass, data)
if not errors:
return self._async_create_opower_entry(data)
if errors:
schema = {
vol.Required(
CONF_USERNAME, default=self.utility_info[CONF_USERNAME]
): str,
vol.Required(CONF_PASSWORD): str,
}
else:
schema = {}
schema[vol.Required(CONF_TOTP_SECRET)] = str
code = user_input[CONF_MFA_CODE]
try:
login_data = await self.mfa_handler.async_submit_mfa_code(code)
except InvalidAuth:
errors["base"] = "invalid_mfa_code"
except CannotConnect:
errors["base"] = "cannot_connect"
else:
self._data[CONF_LOGIN_DATA] = login_data
if self.source == SOURCE_REAUTH:
return self.async_update_reload_and_abort(
self._get_reauth_entry(), data=self._data
)
return self._async_create_opower_entry(self._data)
return self.async_show_form(
step_id="mfa",
data_schema=vol.Schema(schema),
step_id="mfa_code",
data_schema=self.add_suggested_values_to_schema(
vol.Schema({vol.Required(CONF_MFA_CODE): str}), user_input
),
errors=errors,
)
@callback
def _async_create_opower_entry(self, data: dict[str, Any]) -> ConfigFlowResult:
def _async_create_opower_entry(
self, data: dict[str, Any], **kwargs: Any
) -> ConfigFlowResult:
"""Create the config entry."""
return self.async_create_entry(
title=f"{data[CONF_UTILITY]} ({data[CONF_USERNAME]})",
data=data,
**kwargs,
)
async def async_step_reauth(
self, entry_data: Mapping[str, Any]
) -> ConfigFlowResult:
"""Handle configuration by re-auth."""
return await self.async_step_reauth_confirm()
reauth_entry = self._get_reauth_entry()
self._data = dict(reauth_entry.data)
return self.async_show_form(
step_id="reauth_confirm",
description_placeholders={CONF_NAME: reauth_entry.title},
)
async def async_step_reauth_confirm(
self, user_input: dict[str, Any] | None = None
@@ -150,21 +203,34 @@ class OpowerConfigFlow(ConfigFlow, domain=DOMAIN):
"""Dialog that informs the user that reauth is required."""
errors: dict[str, str] = {}
reauth_entry = self._get_reauth_entry()
if user_input is not None:
data = {**reauth_entry.data, **user_input}
errors = await _validate_login(self.hass, data)
if not errors:
return self.async_update_reload_and_abort(reauth_entry, data=data)
schema: VolDictType = {
vol.Required(CONF_USERNAME): reauth_entry.data[CONF_USERNAME],
if user_input is not None:
self._data.update(user_input)
try:
await _validate_login(self.hass, self._data)
except MfaChallenge as exc:
self.mfa_handler = exc.handler
return await self.async_step_mfa_options()
except InvalidAuth:
errors["base"] = "invalid_auth"
except CannotConnect:
errors["base"] = "cannot_connect"
else:
return self.async_update_reload_and_abort(reauth_entry, data=self._data)
utility = select_utility(self._data[CONF_UTILITY])
schema_dict: VolDictType = {
vol.Required(CONF_USERNAME): str,
vol.Required(CONF_PASSWORD): str,
}
if select_utility(reauth_entry.data[CONF_UTILITY]).accepts_mfa():
schema[vol.Optional(CONF_TOTP_SECRET)] = str
if utility.accepts_totp_secret():
schema_dict[vol.Optional(CONF_TOTP_SECRET)] = str
return self.async_show_form(
step_id="reauth_confirm",
data_schema=vol.Schema(schema),
data_schema=self.add_suggested_values_to_schema(
vol.Schema(schema_dict), self._data
),
errors=errors,
description_placeholders={CONF_NAME: reauth_entry.title},
)

1
homeassistant/components/opower/const.py
View File

@@ -4,3 +4,4 @@ DOMAIN = "opower"
CONF_UTILITY = "utility"
CONF_TOTP_SECRET = "totp_secret"
CONF_LOGIN_DATA = "login_data"

7
homeassistant/components/opower/coordinator.py
View File

@@ -14,7 +14,7 @@ from opower import (
ReadResolution,
create_cookie_jar,
)
from opower.exceptions import ApiException, CannotConnect, InvalidAuth
from opower.exceptions import ApiException, CannotConnect, InvalidAuth, MfaChallenge
from homeassistant.components.recorder import get_instance
from homeassistant.components.recorder.models import (
@@ -36,7 +36,7 @@ from homeassistant.helpers.aiohttp_client import async_create_clientsession
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from homeassistant.util import dt as dt_util
from .const import CONF_TOTP_SECRET, CONF_UTILITY, DOMAIN
from .const import CONF_LOGIN_DATA, CONF_TOTP_SECRET, CONF_UTILITY, DOMAIN
_LOGGER = logging.getLogger(__name__)
@@ -69,6 +69,7 @@ class OpowerCoordinator(DataUpdateCoordinator[dict[str, Forecast]]):
config_entry.data[CONF_USERNAME],
config_entry.data[CONF_PASSWORD],
config_entry.data.get(CONF_TOTP_SECRET),
config_entry.data.get(CONF_LOGIN_DATA),
)
@callback
@@ -90,7 +91,7 @@ class OpowerCoordinator(DataUpdateCoordinator[dict[str, Forecast]]):
# Given the infrequent updating (every 12h)
# assume previous session has expired and re-login.
await self.api.async_login()
except InvalidAuth as err:
except (InvalidAuth, MfaChallenge) as err:
_LOGGER.error("Error during login: %s", err)
raise ConfigEntryAuthFailed from err
except CannotConnect as err:

2
homeassistant/components/opower/manifest.json
View File

@@ -7,5 +7,5 @@
"documentation": "https://www.home-assistant.io/integrations/opower",
"iot_class": "cloud_polling",
"loggers": ["opower"],
"requirements": ["opower==0.12.4"]
"requirements": ["opower==0.15.1"]
}

49
homeassistant/components/opower/strings.json
View File

@@ -3,27 +3,43 @@
"step": {
"user": {
"data": {
"utility": "Utility name",
"username": "[%key:common::config_flow::data::username%]",
"password": "[%key:common::config_flow::data::password%]"
"utility": "Utility name"
},
"data_description": {
"utility": "The name of your utility provider",
"username": "The username for your utility account",
"password": "The password for your utility account"
"utility": "The name of your utility provider"
}
},
"mfa": {
"description": "The TOTP secret below is not one of the 6-digit time-based numeric codes. It is a string of around 16 characters containing the shared secret that enables your authenticator app to generate the correct time-based code at the appropriate time. See the documentation.",
"credentials": {
"title": "Enter Credentials",
"data": {
"username": "[%key:common::config_flow::data::username%]",
"password": "[%key:common::config_flow::data::password%]",
"totp_secret": "TOTP secret"
},
"data_description": {
"username": "[%key:component::opower::config::step::user::data_description::username%]",
"password": "[%key:component::opower::config::step::user::data_description::password%]",
"totp_secret": "The TOTP secret for your utility account, used for multi-factor authentication (MFA)."
"username": "The username for your utility account",
"password": "The password for your utility account",
"totp_secret": "This is not a 6-digit code. It is a string of around 16 characters containing the shared secret that enables your authenticator app to generate the correct time-based code at the appropriate time. See the documentation."
}
},
"mfa_options": {
"title": "Multi-factor authentication",
"description": "Your account requires multi-factor authentication (MFA). Select a method to receive your security code.",
"data": {
"mfa_method": "MFA method"
},
"data_description": {
"mfa_method": "How to receive your security code"
}
},
"mfa_code": {
"title": "Enter security code",
"description": "A security code has been sent via your selected method. Please enter it below to complete login.",
"data": {
"mfa_code": "Security code"
},
"data_description": {
"mfa_code": "Typically a 6-digit code"
}
},
"reauth_confirm": {
@@ -31,18 +47,19 @@
"data": {
"username": "[%key:common::config_flow::data::username%]",
"password": "[%key:common::config_flow::data::password%]",
"totp_secret": "[%key:component::opower::config::step::mfa::data::totp_secret%]"
"totp_secret": "[%key:component::opower::config::step::credentials::data::totp_secret%]"
},
"data_description": {
"username": "[%key:component::opower::config::step::user::data_description::username%]",
"password": "[%key:component::opower::config::step::user::data_description::password%]",
"totp_secret": "The TOTP secret for your utility account, used for multi-factor authentication (MFA)."
"username": "[%key:component::opower::config::step::credentials::data_description::username%]",
"password": "[%key:component::opower::config::step::credentials::data_description::password%]",
"totp_secret": "[%key:component::opower::config::step::credentials::data_description::totp_secret%]"
}
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]"
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"invalid_mfa_code": "The security code is incorrect. Please try again."
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_service%]",

2
requirements_all.txt generated
View File

@@ -1628,7 +1628,7 @@ openwrt-luci-rpc==1.1.17
openwrt-ubus-rpc==0.0.2
# homeassistant.components.opower
opower==0.12.4
opower==0.15.1
# homeassistant.components.oralb
oralb-ble==0.17.6

2
requirements_test_all.txt generated
View File

@@ -1384,7 +1384,7 @@ openhomedevice==2.2.0
openwebifpy==4.3.1
# homeassistant.components.opower
opower==0.12.4
opower==0.15.1
# homeassistant.components.oralb
oralb-ble==0.17.6

408
tests/components/opower/test_config_flow.py
View File

@@ -3,7 +3,7 @@
from collections.abc import Generator
from unittest.mock import AsyncMock, patch
from opower import CannotConnect, InvalidAuth
from opower import CannotConnect, InvalidAuth, MfaChallenge
import pytest
from homeassistant import config_entries
@@ -43,24 +43,32 @@ async def test_form(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert not result["errors"]
assert result["step_id"] == "user"
# Select utility
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"utility": "Pacific Gas and Electric Company (PG&E)"},
)
assert result2["type"] is FlowResultType.FORM
assert result2["step_id"] == "credentials"
# Enter credentials
with patch(
"homeassistant.components.opower.config_flow.Opower.async_login",
) as mock_login:
result2 = await hass.config_entries.flow.async_configure(
result3 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"utility": "Pacific Gas and Electric Company (PG&E)",
"username": "test-username",
"password": "test-password",
},
)
await hass.async_block_till_done()
assert result2["type"] is FlowResultType.CREATE_ENTRY
assert result2["title"] == "Pacific Gas and Electric Company (PG&E) (test-username)"
assert result2["data"] == {
assert result3["type"] is FlowResultType.CREATE_ENTRY
assert result3["title"] == "Pacific Gas and Electric Company (PG&E) (test-username)"
assert result3["data"] == {
"utility": "Pacific Gas and Electric Company (PG&E)",
"username": "test-username",
"password": "test-password",
@@ -69,33 +77,33 @@ async def test_form(
assert mock_login.call_count == 1
async def test_form_with_mfa(
async def test_form_with_totp(
recorder_mock: Recorder, hass: HomeAssistant, mock_setup_entry: AsyncMock
) -> None:
"""Test we get the form."""
"""Test we can configure a utility that accepts a TOTP secret."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert not result["errors"]
assert result["step_id"] == "user"
# Select utility
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"utility": "Consolidated Edison (ConEd)",
"username": "test-username",
"password": "test-password",
},
{"utility": "Consolidated Edison (ConEd)"},
)
assert result2["type"] is FlowResultType.FORM
assert not result2["errors"]
assert result2["step_id"] == "credentials"
# Enter credentials
with patch(
"homeassistant.components.opower.config_flow.Opower.async_login",
) as mock_login:
result3 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"username": "test-username",
"password": "test-password",
"totp_secret": "test-totp",
},
)
@@ -112,43 +120,42 @@ async def test_form_with_mfa(
assert mock_login.call_count == 1
async def test_form_with_mfa_bad_secret(
async def test_form_with_invalid_totp(
recorder_mock: Recorder, hass: HomeAssistant, mock_setup_entry: AsyncMock
) -> None:
"""Test MFA asks for password again when validation fails."""
"""Test we handle an invalid TOTP secret."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert not result["errors"]
assert result["step_id"] == "user"
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"utility": "Consolidated Edison (ConEd)",
"username": "test-username",
"password": "test-password",
},
{"utility": "Consolidated Edison (ConEd)"},
)
assert result2["type"] is FlowResultType.FORM
assert not result2["errors"]
assert result2["step_id"] == "credentials"
# Enter invalid credentials
with patch(
"homeassistant.components.opower.config_flow.Opower.async_login",
side_effect=InvalidAuth,
) as mock_login:
):
result3 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"totp_secret": "test-totp",
"username": "test-username",
"password": "test-password",
"totp_secret": "bad-totp",
},
)
assert result3["type"] is FlowResultType.FORM
assert result3["errors"] == {
"base": "invalid_auth",
}
assert result3["errors"] == {"base": "invalid_auth"}
assert result3["step_id"] == "credentials"
# Enter valid credentials
with patch(
"homeassistant.components.opower.config_flow.Opower.async_login",
) as mock_login:
@@ -157,7 +164,7 @@ async def test_form_with_mfa_bad_secret(
{
"username": "test-username",
"password": "updated-password",
"totp_secret": "updated-totp",
"totp_secret": "good-totp",
},
)
@@ -167,26 +174,195 @@ async def test_form_with_mfa_bad_secret(
"utility": "Consolidated Edison (ConEd)",
"username": "test-username",
"password": "updated-password",
"totp_secret": "updated-totp",
"totp_secret": "good-totp",
}
assert len(mock_setup_entry.mock_calls) == 1
assert mock_login.call_count == 1
async def test_form_with_mfa_challenge(
recorder_mock: Recorder, hass: HomeAssistant, mock_setup_entry: AsyncMock
) -> None:
"""Test the full interactive MFA flow, including error recovery."""
# 1. Start the flow and get to the credentials step
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
await hass.config_entries.flow.async_configure(
result["flow_id"],
{"utility": "Pacific Gas and Electric Company (PG&E)"},
)
# 2. Trigger an MfaChallenge on login
mock_mfa_handler = AsyncMock()
mock_mfa_handler.async_get_mfa_options.return_value = {
"Email": "fooxxx@mail.com",
"Phone": "xxx-123",
}
mock_mfa_handler.async_submit_mfa_code.return_value = {
"login_data_mock_key": "login_data_mock_value"
}
with patch(
"homeassistant.components.opower.config_flow.Opower.async_login",
side_effect=MfaChallenge(message="", handler=mock_mfa_handler),
) as mock_login:
result_challenge = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"username": "test-username",
"password": "test-password",
},
)
mock_login.assert_awaited_once()
# 3. Handle the MFA options step, starting with a connection error
assert result_challenge["type"] is FlowResultType.FORM
assert result_challenge["step_id"] == "mfa_options"
mock_mfa_handler.async_get_mfa_options.assert_awaited_once()
# Test CannotConnect on selecting MFA method
mock_mfa_handler.async_select_mfa_option.side_effect = CannotConnect
result_mfa_connect_fail = await hass.config_entries.flow.async_configure(
result["flow_id"], {"mfa_method": "Email"}
)
mock_mfa_handler.async_select_mfa_option.assert_awaited_once_with("Email")
assert result_mfa_connect_fail["type"] is FlowResultType.FORM
assert result_mfa_connect_fail["step_id"] == "mfa_options"
assert result_mfa_connect_fail["errors"] == {"base": "cannot_connect"}
# Retry selecting MFA method successfully
mock_mfa_handler.async_select_mfa_option.side_effect = None
result_mfa_select_ok = await hass.config_entries.flow.async_configure(
result["flow_id"], {"mfa_method": "Email"}
)
assert mock_mfa_handler.async_select_mfa_option.call_count == 2
assert result_mfa_select_ok["type"] is FlowResultType.FORM
assert result_mfa_select_ok["step_id"] == "mfa_code"
# 4. Handle the MFA code step, testing multiple failure scenarios
# Test InvalidAuth on submitting code
mock_mfa_handler.async_submit_mfa_code.side_effect = InvalidAuth
result_mfa_invalid_code = await hass.config_entries.flow.async_configure(
result["flow_id"], {"mfa_code": "bad-code"}
)
mock_mfa_handler.async_submit_mfa_code.assert_awaited_once_with("bad-code")
assert result_mfa_invalid_code["type"] is FlowResultType.FORM
assert result_mfa_invalid_code["step_id"] == "mfa_code"
assert result_mfa_invalid_code["errors"] == {"base": "invalid_mfa_code"}
# Test CannotConnect on submitting code
mock_mfa_handler.async_submit_mfa_code.side_effect = CannotConnect
result_mfa_code_connect_fail = await hass.config_entries.flow.async_configure(
result["flow_id"], {"mfa_code": "good-code"}
)
assert mock_mfa_handler.async_submit_mfa_code.call_count == 2
mock_mfa_handler.async_submit_mfa_code.assert_called_with("good-code")
assert result_mfa_code_connect_fail["type"] is FlowResultType.FORM
assert result_mfa_code_connect_fail["step_id"] == "mfa_code"
assert result_mfa_code_connect_fail["errors"] == {"base": "cannot_connect"}
# Retry submitting code successfully
mock_mfa_handler.async_submit_mfa_code.side_effect = None
result_final = await hass.config_entries.flow.async_configure(
result["flow_id"], {"mfa_code": "good-code"}
)
assert mock_mfa_handler.async_submit_mfa_code.call_count == 3
mock_mfa_handler.async_submit_mfa_code.assert_called_with("good-code")
# 5. Verify the flow completes and creates the entry
assert result_final["type"] is FlowResultType.CREATE_ENTRY
assert (
result_final["title"]
== "Pacific Gas and Electric Company (PG&E) (test-username)"
)
assert result_final["data"] == {
"utility": "Pacific Gas and Electric Company (PG&E)",
"username": "test-username",
"password": "test-password",
"login_data": {"login_data_mock_key": "login_data_mock_value"},
}
await hass.async_block_till_done()
assert len(mock_setup_entry.mock_calls) == 1
async def test_form_with_mfa_challenge_but_no_mfa_options(
recorder_mock: Recorder, hass: HomeAssistant, mock_setup_entry: AsyncMock
) -> None:
"""Test the full interactive MFA flow when there are no MFA options."""
# 1. Start the flow and get to the credentials step
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
await hass.config_entries.flow.async_configure(
result["flow_id"],
{"utility": "Pacific Gas and Electric Company (PG&E)"},
)
# 2. Trigger an MfaChallenge on login
mock_mfa_handler = AsyncMock()
mock_mfa_handler.async_get_mfa_options.return_value = {}
mock_mfa_handler.async_submit_mfa_code.return_value = {
"login_data_mock_key": "login_data_mock_value"
}
with patch(
"homeassistant.components.opower.config_flow.Opower.async_login",
side_effect=MfaChallenge(message="", handler=mock_mfa_handler),
) as mock_login:
result_challenge = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"username": "test-username",
"password": "test-password",
},
)
mock_login.assert_awaited_once()
# 3. No MFA options. Handle the MFA code step
assert result_challenge["type"] is FlowResultType.FORM
assert result_challenge["step_id"] == "mfa_code"
mock_mfa_handler.async_get_mfa_options.assert_awaited_once()
result_final = await hass.config_entries.flow.async_configure(
result["flow_id"], {"mfa_code": "good-code"}
)
mock_mfa_handler.async_submit_mfa_code.assert_called_with("good-code")
# 4. Verify the flow completes and creates the entry
assert result_final["type"] is FlowResultType.CREATE_ENTRY
assert (
result_final["title"]
== "Pacific Gas and Electric Company (PG&E) (test-username)"
)
assert result_final["data"] == {
"utility": "Pacific Gas and Electric Company (PG&E)",
"username": "test-username",
"password": "test-password",
"login_data": {"login_data_mock_key": "login_data_mock_value"},
}
await hass.async_block_till_done()
assert len(mock_setup_entry.mock_calls) == 1
@pytest.mark.parametrize(
("api_exception", "expected_error"),
[
(InvalidAuth(), "invalid_auth"),
(CannotConnect(), "cannot_connect"),
(InvalidAuth, "invalid_auth"),
(CannotConnect, "cannot_connect"),
],
)
async def test_form_exceptions(
recorder_mock: Recorder, hass: HomeAssistant, api_exception, expected_error
recorder_mock: Recorder,
hass: HomeAssistant,
api_exception: Exception,
expected_error: str,
) -> None:
"""Test we handle exceptions."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
await hass.config_entries.flow.async_configure(
result["flow_id"],
{"utility": "Pacific Gas and Electric Company (PG&E)"},
)
with patch(
"homeassistant.components.opower.config_flow.Opower.async_login",
@@ -195,7 +371,6 @@ async def test_form_exceptions(
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"utility": "Pacific Gas and Electric Company (PG&E)",
"username": "test-username",
"password": "test-password",
},
@@ -203,15 +378,10 @@ async def test_form_exceptions(
assert result2["type"] is FlowResultType.FORM
assert result2["errors"] == {"base": expected_error}
# On error, the form should have the previous user input, except password,
# as suggested values.
# On error, the form should have the previous user input as suggested values.
data_schema = result2["data_schema"].schema
assert (
get_schema_suggested_value(data_schema, "utility")
== "Pacific Gas and Electric Company (PG&E)"
)
assert get_schema_suggested_value(data_schema, "username") == "test-username"
assert get_schema_suggested_value(data_schema, "password") is None
assert get_schema_suggested_value(data_schema, "password") == "test-password"
assert mock_login.call_count == 1
@@ -224,6 +394,10 @@ async def test_form_already_configured(
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
await hass.config_entries.flow.async_configure(
result["flow_id"],
{"utility": "Pacific Gas and Electric Company (PG&E)"},
)
with patch(
"homeassistant.components.opower.config_flow.Opower.async_login",
@@ -231,7 +405,6 @@ async def test_form_already_configured(
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"utility": "Pacific Gas and Electric Company (PG&E)",
"username": "test-username",
"password": "test-password",
},
@@ -252,6 +425,10 @@ async def test_form_not_already_configured(
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
await hass.config_entries.flow.async_configure(
result["flow_id"],
{"utility": "Pacific Gas and Electric Company (PG&E)"},
)
with patch(
"homeassistant.components.opower.config_flow.Opower.async_login",
@@ -259,7 +436,6 @@ async def test_form_not_already_configured(
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"utility": "Pacific Gas and Electric Company (PG&E)",
"username": "test-username2",
"password": "test-password",
},
@@ -299,6 +475,16 @@ async def test_form_valid_reauth(
assert result["context"]["source"] == "reauth"
assert result["context"]["title_placeholders"] == {"name": mock_config_entry.title}
with patch(
"homeassistant.components.opower.config_flow.Opower.async_login",
side_effect=InvalidAuth,
):
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["data_schema"].schema.keys() == {
"username",
"password",
}
with patch(
"homeassistant.components.opower.config_flow.Opower.async_login",
) as mock_login:
@@ -321,22 +507,23 @@ async def test_form_valid_reauth(
assert mock_login.call_count == 1
async def test_form_valid_reauth_with_mfa(
async def test_form_valid_reauth_with_totp(
recorder_mock: Recorder,
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_unload_entry: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test that we can handle a valid reauth."""
hass.config_entries.async_update_entry(
mock_config_entry,
"""Test that we can handle a valid reauth for a utility with TOTP."""
mock_config_entry = MockConfigEntry(
title="Consolidated Edison (ConEd) (test-username)",
domain=DOMAIN,
data={
**mock_config_entry.data,
# Requires MFA
"utility": "Consolidated Edison (ConEd)",
"username": "test-username",
"password": "test-password",
},
)
mock_config_entry.add_to_hass(hass)
mock_config_entry.mock_state(hass, ConfigEntryState.LOADED)
hass.config.components.add(DOMAIN)
mock_config_entry.async_start_reauth(hass)
@@ -346,6 +533,17 @@ async def test_form_valid_reauth_with_mfa(
assert len(flows) == 1
result = flows[0]
with patch(
"homeassistant.components.opower.config_flow.Opower.async_login",
side_effect=InvalidAuth,
):
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["data_schema"].schema.keys() == {
"username",
"password",
"totp_secret",
}
with patch(
"homeassistant.components.opower.config_flow.Opower.async_login",
) as mock_login:
@@ -371,3 +569,109 @@ async def test_form_valid_reauth_with_mfa(
assert len(mock_unload_entry.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
assert mock_login.call_count == 1
async def test_reauth_with_mfa_challenge(
recorder_mock: Recorder,
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_unload_entry: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test the full interactive MFA flow during reauth."""
# 1. Set up the existing entry and trigger reauth
mock_config_entry.mock_state(hass, ConfigEntryState.LOADED)
hass.config.components.add(DOMAIN)
mock_config_entry.async_start_reauth(hass)
await hass.async_block_till_done()
flows = hass.config_entries.flow.async_progress()
assert len(flows) == 1
result = flows[0]
assert result["step_id"] == "reauth_confirm"
# 2. Test failure before MFA challenge (InvalidAuth)
with patch(
"homeassistant.components.opower.config_flow.Opower.async_login",
side_effect=InvalidAuth,
) as mock_login_fail_auth:
result_invalid_auth = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"username": "test-username",
"password": "bad-password",
},
)
mock_login_fail_auth.assert_awaited_once()
assert result_invalid_auth["type"] is FlowResultType.FORM
assert result_invalid_auth["step_id"] == "reauth_confirm"
assert result_invalid_auth["errors"] == {"base": "invalid_auth"}
# 3. Test failure before MFA challenge (CannotConnect)
with patch(
"homeassistant.components.opower.config_flow.Opower.async_login",
side_effect=CannotConnect,
) as mock_login_fail_connect:
result_cannot_connect = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"username": "test-username",
"password": "new-password",
},
)
mock_login_fail_connect.assert_awaited_once()
assert result_cannot_connect["type"] is FlowResultType.FORM
assert result_cannot_connect["step_id"] == "reauth_confirm"
assert result_cannot_connect["errors"] == {"base": "cannot_connect"}
# 4. Trigger the MfaChallenge on the next attempt
mock_mfa_handler = AsyncMock()
mock_mfa_handler.async_get_mfa_options.return_value = {
"Email": "fooxxx@mail.com",
"Phone": "xxx-123",
}
mock_mfa_handler.async_submit_mfa_code.return_value = {
"login_data_mock_key": "login_data_mock_value"
}
with patch(
"homeassistant.components.opower.config_flow.Opower.async_login",
side_effect=MfaChallenge(message="", handler=mock_mfa_handler),
) as mock_login_mfa:
result_mfa_challenge = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"username": "test-username",
"password": "new-password",
},
)
mock_login_mfa.assert_awaited_once()
# 5. Handle the happy path for the MFA flow
assert result_mfa_challenge["type"] is FlowResultType.FORM
assert result_mfa_challenge["step_id"] == "mfa_options"
mock_mfa_handler.async_get_mfa_options.assert_awaited_once()
result_mfa_code = await hass.config_entries.flow.async_configure(
result["flow_id"], {"mfa_method": "Phone"}
)
mock_mfa_handler.async_select_mfa_option.assert_awaited_once_with("Phone")
assert result_mfa_code["type"] is FlowResultType.FORM
assert result_mfa_code["step_id"] == "mfa_code"
result_final = await hass.config_entries.flow.async_configure(
result["flow_id"], {"mfa_code": "good-code"}
)
mock_mfa_handler.async_submit_mfa_code.assert_awaited_once_with("good-code")
# 6. Verify the reauth completes successfully
assert result_final["type"] is FlowResultType.ABORT
assert result_final["reason"] == "reauth_successful"
await hass.async_block_till_done()
# Check that data was updated and the entry was reloaded
assert mock_config_entry.data["password"] == "new-password"
assert mock_config_entry.data["login_data"] == {
"login_data_mock_key": "login_data_mock_value"
}
assert len(mock_unload_entry.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1