1
0
mirror of https://github.com/home-assistant/core.git synced 2026-04-18 07:56:03 +01:00

growatt_server: implement reconfiguration flow (Gold) (#165961)

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
johanzander
2026-04-01 19:56:53 +02:00
committed by GitHub
parent 6470cbeada
commit 7daaf3de6a
5 changed files with 346 additions and 230 deletions

View File

@@ -8,7 +8,7 @@ import growattServer
import requests import requests
import voluptuous as vol import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult from homeassistant.config_entries import ConfigEntry, ConfigFlow, ConfigFlowResult
from homeassistant.const import ( from homeassistant.const import (
CONF_NAME, CONF_NAME,
CONF_PASSWORD, CONF_PASSWORD,
@@ -64,6 +64,16 @@ class GrowattServerConfigFlow(ConfigFlow, domain=DOMAIN):
menu_options=["password_auth", "token_auth"], menu_options=["password_auth", "token_auth"],
) )
async def async_step_reconfigure(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle reconfiguration."""
return await self._async_step_credentials(
step_id="reconfigure",
entry=self._get_reconfigure_entry(),
user_input=user_input,
)
async def async_step_reauth(self, _: Mapping[str, Any]) -> ConfigFlowResult: async def async_step_reauth(self, _: Mapping[str, Any]) -> ConfigFlowResult:
"""Handle reauth.""" """Handle reauth."""
return await self.async_step_reauth_confirm() return await self.async_step_reauth_confirm()
@@ -72,11 +82,23 @@ class GrowattServerConfigFlow(ConfigFlow, domain=DOMAIN):
self, user_input: dict[str, Any] | None = None self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult: ) -> ConfigFlowResult:
"""Handle reauth confirmation.""" """Handle reauth confirmation."""
return await self._async_step_credentials(
step_id="reauth_confirm",
entry=self._get_reauth_entry(),
user_input=user_input,
)
async def _async_step_credentials(
self,
step_id: str,
entry: ConfigEntry,
user_input: dict[str, Any] | None,
) -> ConfigFlowResult:
"""Handle credential update for both reauth and reconfigure."""
errors: dict[str, str] = {} errors: dict[str, str] = {}
reauth_entry = self._get_reauth_entry()
if user_input is not None: if user_input is not None:
auth_type = reauth_entry.data.get(CONF_AUTH_TYPE) auth_type = entry.data.get(CONF_AUTH_TYPE)
if auth_type == AUTH_PASSWORD: if auth_type == AUTH_PASSWORD:
server_url = SERVER_URLS_NAMES[user_input[CONF_REGION]] server_url = SERVER_URLS_NAMES[user_input[CONF_REGION]]
@@ -91,17 +113,19 @@ class GrowattServerConfigFlow(ConfigFlow, domain=DOMAIN):
api.login, user_input[CONF_USERNAME], user_input[CONF_PASSWORD] api.login, user_input[CONF_USERNAME], user_input[CONF_PASSWORD]
) )
except requests.exceptions.RequestException as ex: except requests.exceptions.RequestException as ex:
_LOGGER.debug("Network error during reauth login: %s", ex) _LOGGER.debug("Network error during credential update: %s", ex)
errors["base"] = ERROR_CANNOT_CONNECT errors["base"] = ERROR_CANNOT_CONNECT
except (ValueError, KeyError, TypeError, AttributeError) as ex: except (ValueError, KeyError, TypeError, AttributeError) as ex:
_LOGGER.debug("Invalid response format during reauth login: %s", ex) _LOGGER.debug(
"Invalid response format during credential update: %s", ex
)
errors["base"] = ERROR_CANNOT_CONNECT errors["base"] = ERROR_CANNOT_CONNECT
else: else:
if not isinstance(login_response, dict): if not isinstance(login_response, dict):
errors["base"] = ERROR_CANNOT_CONNECT errors["base"] = ERROR_CANNOT_CONNECT
elif login_response.get("success"): elif login_response.get("success"):
return self.async_update_reload_and_abort( return self.async_update_reload_and_abort(
reauth_entry, entry,
data_updates={ data_updates={
CONF_USERNAME: user_input[CONF_USERNAME], CONF_USERNAME: user_input[CONF_USERNAME],
CONF_PASSWORD: user_input[CONF_PASSWORD], CONF_PASSWORD: user_input[CONF_PASSWORD],
@@ -121,28 +145,26 @@ class GrowattServerConfigFlow(ConfigFlow, domain=DOMAIN):
try: try:
await self.hass.async_add_executor_job(api.plant_list) await self.hass.async_add_executor_job(api.plant_list)
except requests.exceptions.RequestException as ex: except requests.exceptions.RequestException as ex:
_LOGGER.debug( _LOGGER.debug("Network error during credential update: %s", ex)
"Network error during reauth token validation: %s", ex
)
errors["base"] = ERROR_CANNOT_CONNECT errors["base"] = ERROR_CANNOT_CONNECT
except growattServer.GrowattV1ApiError as err: except growattServer.GrowattV1ApiError as err:
if err.error_code == V1_API_ERROR_NO_PRIVILEGE: if err.error_code == V1_API_ERROR_NO_PRIVILEGE:
errors["base"] = ERROR_INVALID_AUTH errors["base"] = ERROR_INVALID_AUTH
else: else:
_LOGGER.debug( _LOGGER.debug(
"Growatt V1 API error during reauth: %s (Code: %s)", "Growatt V1 API error during credential update: %s (Code: %s)",
err.error_msg or str(err), err.error_msg or str(err),
err.error_code, err.error_code,
) )
errors["base"] = ERROR_CANNOT_CONNECT errors["base"] = ERROR_CANNOT_CONNECT
except (ValueError, KeyError, TypeError, AttributeError) as ex: except (ValueError, KeyError, TypeError, AttributeError) as ex:
_LOGGER.debug( _LOGGER.debug(
"Invalid response format during reauth token validation: %s", ex "Invalid response format during credential update: %s", ex
) )
errors["base"] = ERROR_CANNOT_CONNECT errors["base"] = ERROR_CANNOT_CONNECT
else: else:
return self.async_update_reload_and_abort( return self.async_update_reload_and_abort(
reauth_entry, entry,
data_updates={ data_updates={
CONF_TOKEN: user_input[CONF_TOKEN], CONF_TOKEN: user_input[CONF_TOKEN],
CONF_URL: server_url, CONF_URL: server_url,
@@ -151,19 +173,19 @@ class GrowattServerConfigFlow(ConfigFlow, domain=DOMAIN):
# Determine the current region key from the stored config value. # Determine the current region key from the stored config value.
# Legacy entries may store the region key directly; newer entries store the URL. # Legacy entries may store the region key directly; newer entries store the URL.
stored_url = reauth_entry.data.get(CONF_URL, "") stored_url = entry.data.get(CONF_URL, "")
if stored_url in SERVER_URLS_NAMES: if stored_url in SERVER_URLS_NAMES:
current_region = stored_url current_region = stored_url
else: else:
current_region = _URL_TO_REGION.get(stored_url, DEFAULT_URL) current_region = _URL_TO_REGION.get(stored_url, DEFAULT_URL)
auth_type = reauth_entry.data.get(CONF_AUTH_TYPE) auth_type = entry.data.get(CONF_AUTH_TYPE)
if auth_type == AUTH_PASSWORD: if auth_type == AUTH_PASSWORD:
data_schema = vol.Schema( data_schema = vol.Schema(
{ {
vol.Required( vol.Required(
CONF_USERNAME, CONF_USERNAME,
default=reauth_entry.data.get(CONF_USERNAME), default=entry.data.get(CONF_USERNAME),
): str, ): str,
vol.Required(CONF_PASSWORD): str, vol.Required(CONF_PASSWORD): str,
vol.Required(CONF_REGION, default=current_region): SelectSelector( vol.Required(CONF_REGION, default=current_region): SelectSelector(
@@ -189,8 +211,18 @@ class GrowattServerConfigFlow(ConfigFlow, domain=DOMAIN):
else: else:
return self.async_abort(reason=ERROR_CANNOT_CONNECT) return self.async_abort(reason=ERROR_CANNOT_CONNECT)
if user_input is not None:
data_schema = self.add_suggested_values_to_schema(
data_schema,
{
key: value
for key, value in user_input.items()
if key not in (CONF_PASSWORD, CONF_TOKEN)
},
)
return self.async_show_form( return self.async_show_form(
step_id="reauth_confirm", step_id=step_id,
data_schema=data_schema, data_schema=data_schema,
errors=errors, errors=errors,
) )

View File

@@ -50,7 +50,7 @@ rules:
entity-translations: done entity-translations: done
exception-translations: done exception-translations: done
icon-translations: done icon-translations: done
reconfiguration-flow: todo reconfiguration-flow: done
repair-issues: repair-issues:
status: exempt status: exempt
comment: Integration does not raise repairable issues. comment: Integration does not raise repairable issues.

View File

@@ -4,7 +4,8 @@
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]", "already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]", "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"no_plants": "No plants have been found on this account", "no_plants": "No plants have been found on this account",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]" "reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]",
"reconfigure_successful": "[%key:common::config_flow::abort::reconfigure_successful%]"
}, },
"error": { "error": {
"cannot_connect": "Cannot connect to Growatt servers. Please check your internet connection and try again.", "cannot_connect": "Cannot connect to Growatt servers. Please check your internet connection and try again.",
@@ -49,6 +50,22 @@
"description": "Re-enter your credentials to continue using this integration.", "description": "Re-enter your credentials to continue using this integration.",
"title": "Re-authenticate with Growatt" "title": "Re-authenticate with Growatt"
}, },
"reconfigure": {
"data": {
"password": "[%key:common::config_flow::data::password%]",
"region": "[%key:component::growatt_server::config::step::password_auth::data::region%]",
"token": "[%key:component::growatt_server::config::step::token_auth::data::token%]",
"username": "[%key:common::config_flow::data::username%]"
},
"data_description": {
"password": "[%key:component::growatt_server::config::step::password_auth::data_description::password%]",
"region": "[%key:component::growatt_server::config::step::password_auth::data_description::region%]",
"token": "[%key:component::growatt_server::config::step::token_auth::data_description::token%]",
"username": "[%key:component::growatt_server::config::step::password_auth::data_description::username%]"
},
"description": "Update your credentials to continue using this integration.",
"title": "Reconfigure Growatt"
},
"token_auth": { "token_auth": {
"data": { "data": {
"region": "[%key:component::growatt_server::config::step::password_auth::data::region%]", "region": "[%key:component::growatt_server::config::step::password_auth::data::region%]",

View File

@@ -1,170 +0,0 @@
# serializer version: 1
# name: test_reauth_password_error_then_recovery[None-login_return_value0]
FlowResultSnapshot({
'description_placeholders': dict({
'name': 'Mock Title',
}),
'errors': dict({
'base': 'invalid_auth',
}),
'flow_id': <ANY>,
'handler': 'growatt_server',
'last_step': None,
'preview': None,
'step_id': 'reauth_confirm',
'type': <FlowResultType.FORM: 'form'>,
})
# ---
# name: test_reauth_password_error_then_recovery[login_side_effect1-None]
FlowResultSnapshot({
'description_placeholders': dict({
'name': 'Mock Title',
}),
'errors': dict({
'base': 'cannot_connect',
}),
'flow_id': <ANY>,
'handler': 'growatt_server',
'last_step': None,
'preview': None,
'step_id': 'reauth_confirm',
'type': <FlowResultType.FORM: 'form'>,
})
# ---
# name: test_reauth_password_exception
dict({
'auth_type': 'password',
'name': 'Test Plant',
'password': 'password',
'plant_id': '123456',
'url': 'https://openapi.growatt.com/',
'username': 'username',
})
# ---
# name: test_reauth_password_non_auth_login_failure
dict({
'auth_type': 'password',
'name': 'Test Plant',
'password': 'password',
'plant_id': '123456',
'url': 'https://openapi.growatt.com/',
'username': 'username',
})
# ---
# name: test_reauth_password_success[https://openapi-us.growatt.com/-user_input1-north_america]
FlowResultSnapshot({
'description_placeholders': dict({
'name': 'Mock Title',
}),
'errors': dict({
}),
'flow_id': <ANY>,
'handler': 'growatt_server',
'last_step': None,
'preview': None,
'step_id': 'reauth_confirm',
'type': <FlowResultType.FORM: 'form'>,
})
# ---
# name: test_reauth_password_success[https://openapi-us.growatt.com/-user_input1-north_america].1
dict({
'auth_type': 'password',
'name': 'Test Plant',
'password': 'password',
'plant_id': '123456',
'url': 'https://openapi-us.growatt.com/',
'username': 'username',
})
# ---
# name: test_reauth_password_success[https://openapi.growatt.com/-user_input0-other_regions]
FlowResultSnapshot({
'description_placeholders': dict({
'name': 'Mock Title',
}),
'errors': dict({
}),
'flow_id': <ANY>,
'handler': 'growatt_server',
'last_step': None,
'preview': None,
'step_id': 'reauth_confirm',
'type': <FlowResultType.FORM: 'form'>,
})
# ---
# name: test_reauth_password_success[https://openapi.growatt.com/-user_input0-other_regions].1
dict({
'auth_type': 'password',
'name': 'Test Plant',
'password': 'password',
'plant_id': '123456',
'url': 'https://openapi.growatt.com/',
'username': 'username',
})
# ---
# name: test_reauth_token_error_then_recovery[plant_list_side_effect0]
FlowResultSnapshot({
'description_placeholders': dict({
'name': 'Mock Title',
}),
'errors': dict({
'base': 'invalid_auth',
}),
'flow_id': <ANY>,
'handler': 'growatt_server',
'last_step': None,
'preview': None,
'step_id': 'reauth_confirm',
'type': <FlowResultType.FORM: 'form'>,
})
# ---
# name: test_reauth_token_error_then_recovery[plant_list_side_effect1]
FlowResultSnapshot({
'description_placeholders': dict({
'name': 'Mock Title',
}),
'errors': dict({
'base': 'cannot_connect',
}),
'flow_id': <ANY>,
'handler': 'growatt_server',
'last_step': None,
'preview': None,
'step_id': 'reauth_confirm',
'type': <FlowResultType.FORM: 'form'>,
})
# ---
# name: test_reauth_token_exception
dict({
'auth_type': 'api_token',
'name': 'Test Plant',
'plant_id': '123456',
'token': 'test_api_token_12345',
'url': 'https://openapi.growatt.com/',
'user_id': '12345',
})
# ---
# name: test_reauth_token_success
FlowResultSnapshot({
'description_placeholders': dict({
'name': 'Mock Title',
}),
'errors': dict({
}),
'flow_id': <ANY>,
'handler': 'growatt_server',
'last_step': None,
'preview': None,
'step_id': 'reauth_confirm',
'type': <FlowResultType.FORM: 'form'>,
})
# ---
# name: test_reauth_token_success.1
dict({
'auth_type': 'api_token',
'name': 'Test Plant',
'plant_id': '123456',
'token': 'test_api_token_12345',
'url': 'https://openapi.growatt.com/',
'user_id': '12345',
})
# ---

View File

@@ -1,12 +1,13 @@
"""Tests for the Growatt server config flow.""" """Tests for the Growatt server config flow."""
from collections.abc import Callable
from copy import deepcopy from copy import deepcopy
from typing import Any
from unittest.mock import MagicMock
import growattServer import growattServer
import pytest import pytest
import requests import requests
from syrupy.assertion import SnapshotAssertion
from syrupy.filters import props
import voluptuous as vol import voluptuous as vol
from homeassistant import config_entries from homeassistant import config_entries
@@ -718,10 +719,9 @@ async def test_password_auth_plant_list_invalid_format(
) )
async def test_reauth_password_success( async def test_reauth_password_success(
hass: HomeAssistant, hass: HomeAssistant,
mock_growatt_classic_api, mock_growatt_classic_api: MagicMock,
snapshot: SnapshotAssertion,
stored_url: str, stored_url: str,
user_input: dict, user_input: dict[str, str],
expected_region: str, expected_region: str,
) -> None: ) -> None:
"""Test successful reauthentication with password auth for default and non-default regions.""" """Test successful reauthentication with password auth for default and non-default regions."""
@@ -733,7 +733,7 @@ async def test_reauth_password_success(
CONF_PASSWORD: "test_password", CONF_PASSWORD: "test_password",
CONF_URL: stored_url, CONF_URL: stored_url,
CONF_PLANT_ID: "123456", CONF_PLANT_ID: "123456",
"name": "Test Plant", CONF_NAME: "Test Plant",
}, },
unique_id="123456", unique_id="123456",
) )
@@ -743,7 +743,6 @@ async def test_reauth_password_success(
assert result["type"] is FlowResultType.FORM assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reauth_confirm" assert result["step_id"] == "reauth_confirm"
assert result == snapshot(exclude=props("data_schema"))
region_key = next( region_key = next(
k k
for k in result["data_schema"].schema for k in result["data_schema"].schema
@@ -758,29 +757,35 @@ async def test_reauth_password_success(
assert result["type"] is FlowResultType.ABORT assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reauth_successful" assert result["reason"] == "reauth_successful"
assert entry.data == snapshot assert entry.data == {
CONF_AUTH_TYPE: AUTH_PASSWORD,
CONF_NAME: "Test Plant",
CONF_PASSWORD: user_input[CONF_PASSWORD],
CONF_PLANT_ID: "123456",
CONF_URL: SERVER_URLS_NAMES[user_input[CONF_REGION]],
CONF_USERNAME: user_input[CONF_USERNAME],
}
@pytest.mark.parametrize( @pytest.mark.parametrize(
("login_side_effect", "login_return_value"), ("login_side_effect", "expected_error"),
[ [
( (
None, lambda *args, **kwargs: {"msg": LOGIN_INVALID_AUTH_CODE, "success": False},
{"msg": LOGIN_INVALID_AUTH_CODE, "success": False}, ERROR_INVALID_AUTH,
), ),
( (
requests.exceptions.ConnectionError("Connection failed"), requests.exceptions.ConnectionError("Connection failed"),
None, ERROR_CANNOT_CONNECT,
), ),
], ],
) )
async def test_reauth_password_error_then_recovery( async def test_reauth_password_error_then_recovery(
hass: HomeAssistant, hass: HomeAssistant,
mock_growatt_classic_api, mock_growatt_classic_api: MagicMock,
mock_config_entry_classic: MockConfigEntry, mock_config_entry_classic: MockConfigEntry,
snapshot: SnapshotAssertion, login_side_effect: Callable[..., Any] | Exception,
login_side_effect: Exception | None, expected_error: str,
login_return_value: dict | None,
) -> None: ) -> None:
"""Test password reauth shows error then allows recovery.""" """Test password reauth shows error then allows recovery."""
mock_config_entry_classic.add_to_hass(hass) mock_config_entry_classic.add_to_hass(hass)
@@ -791,15 +796,13 @@ async def test_reauth_password_error_then_recovery(
assert result["step_id"] == "reauth_confirm" assert result["step_id"] == "reauth_confirm"
mock_growatt_classic_api.login.side_effect = login_side_effect mock_growatt_classic_api.login.side_effect = login_side_effect
if login_return_value is not None:
mock_growatt_classic_api.login.return_value = login_return_value
result = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
result["flow_id"], FIXTURE_USER_INPUT_PASSWORD result["flow_id"], FIXTURE_USER_INPUT_PASSWORD
) )
assert result["type"] is FlowResultType.FORM assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reauth_confirm" assert result["step_id"] == "reauth_confirm"
assert result == snapshot(exclude=props("data_schema")) assert result["errors"] == {"base": expected_error}
# Recover with correct credentials # Recover with correct credentials
mock_growatt_classic_api.login.side_effect = None mock_growatt_classic_api.login.side_effect = None
@@ -814,9 +817,8 @@ async def test_reauth_password_error_then_recovery(
async def test_reauth_token_success( async def test_reauth_token_success(
hass: HomeAssistant, hass: HomeAssistant,
mock_growatt_v1_api, mock_growatt_v1_api: MagicMock,
mock_config_entry: MockConfigEntry, mock_config_entry: MockConfigEntry,
snapshot: SnapshotAssertion,
) -> None: ) -> None:
"""Test successful reauthentication with token auth.""" """Test successful reauthentication with token auth."""
mock_config_entry.add_to_hass(hass) mock_config_entry.add_to_hass(hass)
@@ -825,7 +827,6 @@ async def test_reauth_token_success(
assert result["type"] is FlowResultType.FORM assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reauth_confirm" assert result["step_id"] == "reauth_confirm"
assert result == snapshot(exclude=props("data_schema"))
mock_growatt_v1_api.plant_list.return_value = GROWATT_V1_PLANT_LIST_RESPONSE mock_growatt_v1_api.plant_list.return_value = GROWATT_V1_PLANT_LIST_RESPONSE
result = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
@@ -834,7 +835,14 @@ async def test_reauth_token_success(
assert result["type"] is FlowResultType.ABORT assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reauth_successful" assert result["reason"] == "reauth_successful"
assert mock_config_entry.data == snapshot assert mock_config_entry.data == {
CONF_AUTH_TYPE: AUTH_API_TOKEN,
CONF_NAME: "Test Plant",
CONF_PLANT_ID: "123456",
CONF_TOKEN: FIXTURE_USER_INPUT_TOKEN[CONF_TOKEN],
CONF_URL: SERVER_URLS_NAMES[FIXTURE_USER_INPUT_TOKEN[CONF_REGION]],
"user_id": "12345",
}
def _make_no_privilege_error() -> growattServer.GrowattV1ApiError: def _make_no_privilege_error() -> growattServer.GrowattV1ApiError:
@@ -844,18 +852,18 @@ def _make_no_privilege_error() -> growattServer.GrowattV1ApiError:
@pytest.mark.parametrize( @pytest.mark.parametrize(
"plant_list_side_effect", ("plant_list_side_effect", "expected_error"),
[ [
_make_no_privilege_error(), (_make_no_privilege_error(), ERROR_INVALID_AUTH),
requests.exceptions.ConnectionError("Network error"), (requests.exceptions.ConnectionError("Network error"), ERROR_CANNOT_CONNECT),
], ],
) )
async def test_reauth_token_error_then_recovery( async def test_reauth_token_error_then_recovery(
hass: HomeAssistant, hass: HomeAssistant,
mock_growatt_v1_api, mock_growatt_v1_api: MagicMock,
mock_config_entry: MockConfigEntry, mock_config_entry: MockConfigEntry,
snapshot: SnapshotAssertion,
plant_list_side_effect: Exception, plant_list_side_effect: Exception,
expected_error: str,
) -> None: ) -> None:
"""Test token reauth shows error then allows recovery.""" """Test token reauth shows error then allows recovery."""
mock_config_entry.add_to_hass(hass) mock_config_entry.add_to_hass(hass)
@@ -872,7 +880,7 @@ async def test_reauth_token_error_then_recovery(
assert result["type"] is FlowResultType.FORM assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reauth_confirm" assert result["step_id"] == "reauth_confirm"
assert result == snapshot(exclude=props("data_schema")) assert result["errors"] == {"base": expected_error}
# Recover with a valid token # Recover with a valid token
mock_growatt_v1_api.plant_list.side_effect = None mock_growatt_v1_api.plant_list.side_effect = None
@@ -887,7 +895,7 @@ async def test_reauth_token_error_then_recovery(
async def test_reauth_token_non_auth_api_error( async def test_reauth_token_non_auth_api_error(
hass: HomeAssistant, hass: HomeAssistant,
mock_growatt_v1_api, mock_growatt_v1_api: MagicMock,
mock_config_entry: MockConfigEntry, mock_config_entry: MockConfigEntry,
) -> None: ) -> None:
"""Test reauth token with non-auth V1 API error (e.g. rate limit) shows cannot_connect.""" """Test reauth token with non-auth V1 API error (e.g. rate limit) shows cannot_connect."""
@@ -912,7 +920,7 @@ async def test_reauth_token_non_auth_api_error(
async def test_reauth_password_invalid_response( async def test_reauth_password_invalid_response(
hass: HomeAssistant, hass: HomeAssistant,
mock_growatt_classic_api, mock_growatt_classic_api: MagicMock,
mock_config_entry_classic: MockConfigEntry, mock_config_entry_classic: MockConfigEntry,
) -> None: ) -> None:
"""Test reauth password flow with non-dict login response, then recovery.""" """Test reauth password flow with non-dict login response, then recovery."""
@@ -940,9 +948,8 @@ async def test_reauth_password_invalid_response(
async def test_reauth_password_non_auth_login_failure( async def test_reauth_password_non_auth_login_failure(
hass: HomeAssistant, hass: HomeAssistant,
mock_growatt_classic_api, mock_growatt_classic_api: MagicMock,
mock_config_entry_classic: MockConfigEntry, mock_config_entry_classic: MockConfigEntry,
snapshot: SnapshotAssertion,
) -> None: ) -> None:
"""Test reauth password flow when login fails with a non-auth error.""" """Test reauth password flow when login fails with a non-auth error."""
mock_config_entry_classic.add_to_hass(hass) mock_config_entry_classic.add_to_hass(hass)
@@ -968,14 +975,20 @@ async def test_reauth_password_non_auth_login_failure(
assert result["type"] is FlowResultType.ABORT assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reauth_successful" assert result["reason"] == "reauth_successful"
assert mock_config_entry_classic.data == snapshot assert mock_config_entry_classic.data == {
CONF_AUTH_TYPE: AUTH_PASSWORD,
CONF_NAME: "Test Plant",
CONF_PASSWORD: FIXTURE_USER_INPUT_PASSWORD[CONF_PASSWORD],
CONF_PLANT_ID: "123456",
CONF_URL: SERVER_URLS_NAMES[FIXTURE_USER_INPUT_PASSWORD[CONF_REGION]],
CONF_USERNAME: FIXTURE_USER_INPUT_PASSWORD[CONF_USERNAME],
}
async def test_reauth_password_exception( async def test_reauth_password_exception(
hass: HomeAssistant, hass: HomeAssistant,
mock_growatt_classic_api, mock_growatt_classic_api: MagicMock,
mock_config_entry_classic: MockConfigEntry, mock_config_entry_classic: MockConfigEntry,
snapshot: SnapshotAssertion,
) -> None: ) -> None:
"""Test reauth password flow with unexpected exception from login, then recovery.""" """Test reauth password flow with unexpected exception from login, then recovery."""
mock_config_entry_classic.add_to_hass(hass) mock_config_entry_classic.add_to_hass(hass)
@@ -999,14 +1012,20 @@ async def test_reauth_password_exception(
assert result["type"] is FlowResultType.ABORT assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reauth_successful" assert result["reason"] == "reauth_successful"
assert mock_config_entry_classic.data == snapshot assert mock_config_entry_classic.data == {
CONF_AUTH_TYPE: AUTH_PASSWORD,
CONF_NAME: "Test Plant",
CONF_PASSWORD: FIXTURE_USER_INPUT_PASSWORD[CONF_PASSWORD],
CONF_PLANT_ID: "123456",
CONF_URL: SERVER_URLS_NAMES[FIXTURE_USER_INPUT_PASSWORD[CONF_REGION]],
CONF_USERNAME: FIXTURE_USER_INPUT_PASSWORD[CONF_USERNAME],
}
async def test_reauth_token_exception( async def test_reauth_token_exception(
hass: HomeAssistant, hass: HomeAssistant,
mock_growatt_v1_api, mock_growatt_v1_api: MagicMock,
mock_config_entry: MockConfigEntry, mock_config_entry: MockConfigEntry,
snapshot: SnapshotAssertion,
) -> None: ) -> None:
"""Test reauth token flow with unexpected exception from plant_list, then recovery.""" """Test reauth token flow with unexpected exception from plant_list, then recovery."""
mock_config_entry.add_to_hass(hass) mock_config_entry.add_to_hass(hass)
@@ -1030,7 +1049,14 @@ async def test_reauth_token_exception(
assert result["type"] is FlowResultType.ABORT assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reauth_successful" assert result["reason"] == "reauth_successful"
assert mock_config_entry.data == snapshot assert mock_config_entry.data == {
CONF_AUTH_TYPE: AUTH_API_TOKEN,
CONF_NAME: "Test Plant",
CONF_PLANT_ID: "123456",
CONF_TOKEN: FIXTURE_USER_INPUT_TOKEN[CONF_TOKEN],
CONF_URL: SERVER_URLS_NAMES[FIXTURE_USER_INPUT_TOKEN[CONF_REGION]],
"user_id": "12345",
}
async def test_reauth_unknown_auth_type(hass: HomeAssistant) -> None: async def test_reauth_unknown_auth_type(hass: HomeAssistant) -> None:
@@ -1039,8 +1065,8 @@ async def test_reauth_unknown_auth_type(hass: HomeAssistant) -> None:
domain=DOMAIN, domain=DOMAIN,
data={ data={
CONF_AUTH_TYPE: "unknown_type", CONF_AUTH_TYPE: "unknown_type",
"plant_id": "123456", CONF_PLANT_ID: "123456",
"name": "Test Plant", CONF_NAME: "Test Plant",
}, },
unique_id="123456", unique_id="123456",
) )
@@ -1051,3 +1077,214 @@ async def test_reauth_unknown_auth_type(hass: HomeAssistant) -> None:
assert result["type"] is FlowResultType.ABORT assert result["type"] is FlowResultType.ABORT
assert result["reason"] == ERROR_CANNOT_CONNECT assert result["reason"] == ERROR_CANNOT_CONNECT
# Reconfiguration flow tests
@pytest.mark.parametrize(
("stored_url", "user_input", "expected_region"),
[
(
SERVER_URLS_NAMES["other_regions"],
FIXTURE_USER_INPUT_PASSWORD,
"other_regions",
),
(
SERVER_URLS_NAMES["north_america"],
{
CONF_USERNAME: "username",
CONF_PASSWORD: "password",
CONF_REGION: "north_america",
},
"north_america",
),
],
)
async def test_reconfigure_password_success(
hass: HomeAssistant,
mock_growatt_classic_api: MagicMock,
stored_url: str,
user_input: dict[str, str],
expected_region: str,
) -> None:
"""Test successful reconfiguration with password auth for default and non-default regions."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_AUTH_TYPE: AUTH_PASSWORD,
CONF_USERNAME: "test_user",
CONF_PASSWORD: "test_password",
CONF_URL: stored_url,
CONF_PLANT_ID: "123456",
CONF_NAME: "Test Plant",
},
unique_id="123456",
)
entry.add_to_hass(hass)
result = await entry.start_reconfigure_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reconfigure"
region_key = next(
k
for k in result["data_schema"].schema
if isinstance(k, vol.Required) and k.schema == CONF_REGION
)
assert region_key.default() == expected_region
mock_growatt_classic_api.login.return_value = GROWATT_LOGIN_RESPONSE
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reconfigure_successful"
assert entry.data == {
CONF_AUTH_TYPE: AUTH_PASSWORD,
CONF_NAME: "Test Plant",
CONF_PASSWORD: user_input[CONF_PASSWORD],
CONF_PLANT_ID: "123456",
CONF_URL: SERVER_URLS_NAMES[user_input[CONF_REGION]],
CONF_USERNAME: user_input[CONF_USERNAME],
}
@pytest.mark.parametrize(
("login_side_effect", "expected_error"),
[
(
lambda *args, **kwargs: {"msg": LOGIN_INVALID_AUTH_CODE, "success": False},
ERROR_INVALID_AUTH,
),
(
requests.exceptions.ConnectionError("Connection failed"),
ERROR_CANNOT_CONNECT,
),
],
)
async def test_reconfigure_password_error_then_recovery(
hass: HomeAssistant,
mock_growatt_classic_api: MagicMock,
mock_config_entry_classic: MockConfigEntry,
login_side_effect: Callable[..., Any] | Exception,
expected_error: str,
) -> None:
"""Test password reconfigure shows error then allows recovery."""
mock_config_entry_classic.add_to_hass(hass)
result = await mock_config_entry_classic.start_reconfigure_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reconfigure"
mock_growatt_classic_api.login.side_effect = login_side_effect
result = await hass.config_entries.flow.async_configure(
result["flow_id"], FIXTURE_USER_INPUT_PASSWORD
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reconfigure"
assert result["errors"] == {"base": expected_error}
# Recover with correct credentials
mock_growatt_classic_api.login.side_effect = None
mock_growatt_classic_api.login.return_value = GROWATT_LOGIN_RESPONSE
result = await hass.config_entries.flow.async_configure(
result["flow_id"], FIXTURE_USER_INPUT_PASSWORD
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reconfigure_successful"
async def test_reconfigure_token_success(
hass: HomeAssistant,
mock_growatt_v1_api: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test successful reconfiguration with token auth."""
mock_config_entry.add_to_hass(hass)
result = await mock_config_entry.start_reconfigure_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reconfigure"
mock_growatt_v1_api.plant_list.return_value = GROWATT_V1_PLANT_LIST_RESPONSE
result = await hass.config_entries.flow.async_configure(
result["flow_id"], FIXTURE_USER_INPUT_TOKEN
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reconfigure_successful"
assert mock_config_entry.data == {
CONF_AUTH_TYPE: AUTH_API_TOKEN,
CONF_NAME: "Test Plant",
CONF_PLANT_ID: "123456",
CONF_TOKEN: FIXTURE_USER_INPUT_TOKEN[CONF_TOKEN],
CONF_URL: SERVER_URLS_NAMES[FIXTURE_USER_INPUT_TOKEN[CONF_REGION]],
"user_id": "12345",
}
@pytest.mark.parametrize(
("plant_list_side_effect", "expected_error"),
[
(_make_no_privilege_error(), ERROR_INVALID_AUTH),
(requests.exceptions.ConnectionError("Network error"), ERROR_CANNOT_CONNECT),
],
)
async def test_reconfigure_token_error_then_recovery(
hass: HomeAssistant,
mock_growatt_v1_api: MagicMock,
mock_config_entry: MockConfigEntry,
plant_list_side_effect: Exception,
expected_error: str,
) -> None:
"""Test token reconfigure shows error then allows recovery."""
mock_config_entry.add_to_hass(hass)
result = await mock_config_entry.start_reconfigure_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reconfigure"
mock_growatt_v1_api.plant_list.side_effect = plant_list_side_effect
result = await hass.config_entries.flow.async_configure(
result["flow_id"], FIXTURE_USER_INPUT_TOKEN
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reconfigure"
assert result["errors"] == {"base": expected_error}
# Recover with a valid token
mock_growatt_v1_api.plant_list.side_effect = None
mock_growatt_v1_api.plant_list.return_value = GROWATT_V1_PLANT_LIST_RESPONSE
result = await hass.config_entries.flow.async_configure(
result["flow_id"], FIXTURE_USER_INPUT_TOKEN
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reconfigure_successful"
async def test_reconfigure_unknown_auth_type(hass: HomeAssistant) -> None:
"""Test reconfigure aborts immediately when the config entry has an unknown auth type."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_AUTH_TYPE: "unknown_type",
CONF_PLANT_ID: "123456",
CONF_NAME: "Test Plant",
},
unique_id="123456",
)
entry.add_to_hass(hass)
result = await entry.start_reconfigure_flow(hass)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == ERROR_CANNOT_CONNECT