jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-23 08:47:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add login and rewrite access to extended data for solarlog (#126024)

Browse Source 
* Initial commit

* Add/update tests

* Minor adjustment

* Update data_schema

* Adjust get password

* Set const for has_password, remove deletion of extended_data

* Update diagnostics snapshot

* Correct typo

* Add test for migration from mv 2 to 3

* Adjust migration test
This commit is contained in:
dontinelli 2024-09-20 14:19:45 +02:00 committed by GitHub
parent 184580257d
commit 41ffa8d6db
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
9 changed files with 433 additions and 76 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
homeassistant/components/solarlog/__init__.py
View File

@ -7,6 +7,7 @@ from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from .const import CONF_HAS_PWD
from .coordinator import SolarLogCoordinator
_LOGGER = logging.getLogger(__name__)
@ -57,12 +58,13 @@ async def async_migrate_entry(
entity.entity_id, new_unique_id=new_uid
)
if config_entry.minor_version < 3:
# migrate config_entry
new = {**config_entry.data}
new["extended_data"] = False
new[CONF_HAS_PWD] = False
hass.config_entries.async_update_entry(
config_entry, data=new, minor_version=2, version=1
config_entry, data=new, minor_version=3, version=1
)
_LOGGER.debug(

128
homeassistant/components/solarlog/config_flow.py
View File

@ -1,18 +1,24 @@
"""Config flow for solarlog integration."""
from collections.abc import Mapping
import logging
from typing import TYPE_CHECKING, Any
from urllib.parse import ParseResult, urlparse
from solarlog_cli.solarlog_connector import SolarLogConnector
from solarlog_cli.solarlog_exceptions import SolarLogConnectionError, SolarLogError
from solarlog_cli.solarlog_exceptions import (
SolarLogAuthenticationError,
SolarLogConnectionError,
SolarLogError,
)
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_HOST, CONF_NAME
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PASSWORD
from homeassistant.util import slugify
from .const import DEFAULT_HOST, DEFAULT_NAME, DOMAIN
from . import SolarlogConfigEntry
from .const import CONF_HAS_PWD, DEFAULT_HOST, DEFAULT_NAME, DOMAIN
_LOGGER = logging.getLogger(__name__)
@ -20,12 +26,14 @@ _LOGGER = logging.getLogger(__name__)
class SolarLogConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for solarlog."""
_entry: SolarlogConfigEntry | None = None
VERSION = 1
MINOR_VERSION = 2
MINOR_VERSION = 3
def __init__(self) -> None:
"""Initialize the config flow."""
self._errors: dict = {}
self._user_input: dict = {}
def _parse_url(self, host: str) -> str:
"""Return parsed host url."""
@ -51,6 +59,23 @@ class SolarLogConfigFlow(ConfigFlow, domain=DOMAIN):
return True
async def _test_extended_data(self, host: str, pwd: str = "") -> bool:
"""Check if we get extended data from Solar-Log device."""
response: bool = False
solarlog = SolarLogConnector(host, password=pwd)
try:
response = await solarlog.test_extended_data_available()
except SolarLogAuthenticationError:
self._errors = {CONF_HOST: "password_error"}
response = False
except SolarLogError:
self._errors = {CONF_HOST: "unknown"}
response = False
finally:
await solarlog.client.close()
return response
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
@ -64,6 +89,10 @@ class SolarLogConfigFlow(ConfigFlow, domain=DOMAIN):
user_input[CONF_NAME] = slugify(user_input[CONF_NAME])
if await self._test_connection(user_input[CONF_HOST]):
if user_input[CONF_HAS_PWD]:
self._user_input = user_input
return await self.async_step_password()
return self.async_create_entry(
title=user_input[CONF_NAME], data=user_input
)
@ -76,7 +105,33 @@ class SolarLogConfigFlow(ConfigFlow, domain=DOMAIN):
{
vol.Required(CONF_NAME, default=user_input[CONF_NAME]): str,
vol.Required(CONF_HOST, default=user_input[CONF_HOST]): str,
vol.Required("extended_data", default=False): bool,
vol.Required(CONF_HAS_PWD, default=False): bool,
}
),
errors=self._errors,
)
async def async_step_password(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Step when user sets password ."""
self._errors = {}
if user_input is not None:
if await self._test_extended_data(
self._user_input[CONF_HOST], user_input[CONF_PASSWORD]
):
self._user_input |= user_input
return self.async_create_entry(
title=self._user_input[CONF_NAME], data=self._user_input
)
else:
user_input = {CONF_PASSWORD: ""}
return self.async_show_form(
step_id="password",
data_schema=vol.Schema(
{
vol.Required(CONF_PASSWORD): str,
}
),
errors=self._errors,
@ -93,19 +148,66 @@ class SolarLogConfigFlow(ConfigFlow, domain=DOMAIN):
assert entry is not None
if user_input is not None:
return self.async_update_reload_and_abort(
entry,
reason="reconfigure_successful",
data={**entry.data, **user_input},
)
if not user_input[CONF_HAS_PWD] or user_input.get(CONF_PASSWORD, "") == "":
user_input[CONF_PASSWORD] = ""
user_input[CONF_HAS_PWD] = False
return self.async_update_reload_and_abort(
entry,
reason="reconfigure_successful",
data={**entry.data, **user_input},
)
if await self._test_extended_data(
entry.data[CONF_HOST], user_input.get(CONF_PASSWORD, "")
):
# if password has been provided, only save if extended data is available
return self.async_update_reload_and_abort(
entry,
reason="reconfigure_successful",
data={**entry.data, **user_input},
)
return self.async_show_form(
step_id="reconfigure",
data_schema=vol.Schema(
{
vol.Required(
"extended_data", default=entry.data["extended_data"]
): bool,
vol.Optional(CONF_HAS_PWD, default=entry.data[CONF_HAS_PWD]): bool,
vol.Optional(CONF_PASSWORD): str,
}
),
)
async def async_step_reauth(
self, entry_data: Mapping[str, Any]
) -> ConfigFlowResult:
"""Handle flow upon an API authentication error."""
self._entry = self.hass.config_entries.async_get_entry(self.context["entry_id"])
return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle reauthorization flow."""
assert self._entry is not None
if user_input and await self._test_extended_data(
self._entry.data[CONF_HOST], user_input.get(CONF_PASSWORD, "")
):
return self.async_update_reload_and_abort(
self._entry, data={**self._entry.data, **user_input}
)
data_schema = vol.Schema(
{
vol.Optional(
CONF_HAS_PWD, default=self._entry.data[CONF_HAS_PWD]
): bool,
vol.Optional(CONF_PASSWORD): str,
}
)
return self.async_show_form(
step_id="reauth_confirm",
data_schema=data_schema,
errors=self._errors,
)

2
homeassistant/components/solarlog/const.py
View File

@ -7,3 +7,5 @@ DOMAIN = "solarlog"
# Default config for solarlog.
DEFAULT_HOST = "http://solar-log"
DEFAULT_NAME = "solarlog"
CONF_HAS_PWD = "has_password"

42
homeassistant/components/solarlog/coordinator.py
View File

@ -9,6 +9,7 @@ from urllib.parse import ParseResult, urlparse
from solarlog_cli.solarlog_connector import SolarLogConnector
from solarlog_cli.solarlog_exceptions import (
SolarLogAuthenticationError,
SolarLogConnectionError,
SolarLogUpdateError,
)
@ -16,7 +17,7 @@ from solarlog_cli.solarlog_models import SolarlogData
from homeassistant.const import CONF_HOST
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
_LOGGER = logging.getLogger(__name__)
@ -35,6 +36,7 @@ class SolarLogCoordinator(DataUpdateCoordinator[SolarlogData]):
)
host_entry = entry.data[CONF_HOST]
password = entry.data.get("password", "")
url = urlparse(host_entry, "http")
netloc = url.netloc or url.path
@ -45,12 +47,18 @@ class SolarLogCoordinator(DataUpdateCoordinator[SolarlogData]):
self.host = url.geturl()
self.solarlog = SolarLogConnector(
self.host, entry.data["extended_data"], hass.config.time_zone
self.host,
tz=hass.config.time_zone,
password=password,
)
async def _async_setup(self) -> None:
"""Do initialization logic."""
if self.solarlog.extended_data:
_LOGGER.debug("Start async_setup")
logged_in = False
if self.solarlog.password != "":
logged_in = await self.renew_authentication()
if logged_in or await self.solarlog.test_extended_data_available():
device_list = await self.solarlog.update_device_list()
self.solarlog.set_enabled_devices({key: True for key in device_list})
@ -63,11 +71,31 @@ class SolarLogCoordinator(DataUpdateCoordinator[SolarlogData]):
if self.solarlog.extended_data:
await self.solarlog.update_device_list()
data.inverter_data = await self.solarlog.update_inverter_data()
except SolarLogConnectionError as err:
raise ConfigEntryNotReady(err) from err
except SolarLogUpdateError as err:
raise UpdateFailed(err) from err
except SolarLogConnectionError as ex:
raise ConfigEntryNotReady(ex) from ex
except SolarLogAuthenticationError as ex:
if await self.renew_authentication():
# login was successful, update availability of extended data, retry data update
await self.solarlog.test_extended_data_available()
raise ConfigEntryNotReady from ex
raise ConfigEntryAuthFailed from ex
except SolarLogUpdateError as ex:
raise UpdateFailed(ex) from ex
_LOGGER.debug("Data successfully updated")
return data
async def renew_authentication(self) -> bool:
"""Renew access token for SolarLog API."""
logged_in = False
try:
logged_in = await self.solarlog.login()
except SolarLogAuthenticationError as ex:
raise ConfigEntryAuthFailed from ex
except (SolarLogConnectionError, SolarLogUpdateError) as ex:
raise ConfigEntryNotReady from ex
_LOGGER.debug("Credentials successfully updated? %s", logged_in)
return logged_in

25
homeassistant/components/solarlog/strings.json
View File

@ -6,26 +6,45 @@
"data": {
"host": "[%key:common::config_flow::data::host%]",
"name": "The prefix to be used for your Solar-Log sensors",
"extended_data": "Get additional data from Solar-Log. Extended data is only accessible, if no password is set for the Solar-Log. Use at your own risk!"
"has_password": "I have the password for the Solar-Log user account."
},
"data_description": {
"host": "The hostname or IP address of your Solar-Log device."
"host": "The hostname or IP address of your Solar-Log device.",
"has_password": "The password is required, if the open JSON-API is deactivated or if you would like to access additional data provided by your Solar-Log device."
}
},
"password": {
"title": "Define your Solar-Log connection",
"data": {
"password": "[%key:common::config_flow::data::password%]"
},
"data_description": {
"password": "The password for the general user of your Solar-Log device."
}
},
"reauth_confirm": {
"description": "Update your credentials for Solar-Log device",
"data": {
"has_password": "[%key:component::solarlog::config::step::user::data::has_password%]",
"password": "[%key:common::config_flow::data::password%]"
}
},
"reconfigure": {
"title": "Configure SolarLog",
"data": {
"extended_data": "[%key:component::solarlog::config::step::user::data::extended_data%]"
"has_password": "[%key:component::solarlog::config::step::user::data::has_password%]"
}
}
},
"error": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"password_error": "[%key:common::config_flow::error::invalid_auth%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]",
"reconfigure_successful": "[%key:common::config_flow::abort::reconfigure_successful%]"
}
},

17
tests/components/solarlog/conftest.py
View File

@ -6,8 +6,11 @@ from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from solarlog_cli.solarlog_models import InverterData, SolarlogData
from homeassistant.components.solarlog.const import DOMAIN as SOLARLOG_DOMAIN
from homeassistant.const import CONF_HOST, CONF_NAME
from homeassistant.components.solarlog.const import (
CONF_HAS_PWD,
DOMAIN as SOLARLOG_DOMAIN,
)
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PASSWORD
from .const import HOST, NAME
@ -36,9 +39,10 @@ def mock_config_entry() -> MockConfigEntry:
data={
CONF_HOST: HOST,
CONF_NAME: NAME,
"extended_data": True,
CONF_HAS_PWD: True,
CONF_PASSWORD: "pwd",
},
minor_version=2,
minor_version=3,
entry_id="ce5f5431554d101905d31797e1232da8",
)
@ -55,11 +59,14 @@ def mock_solarlog_connector():
mock_solarlog_api = AsyncMock()
mock_solarlog_api.set_enabled_devices = MagicMock()
mock_solarlog_api.test_connection.return_value = True
mock_solarlog_api.test_extended_data_available.return_value = True
mock_solarlog_api.extended_data.return_value = True
mock_solarlog_api.update_data.return_value = data
mock_solarlog_api.update_device_list.return_value = INVERTER_DATA
mock_solarlog_api.update_device_list.return_value = DEVICE_LIST
mock_solarlog_api.update_inverter_data.return_value = INVERTER_DATA
mock_solarlog_api.device_name = {0: "Inverter 1", 1: "Inverter 2"}.get
mock_solarlog_api.device_enabled = {0: True, 1: False}.get
mock_solarlog_api.password.return_value = "pwd"
with (
patch(

5
tests/components/solarlog/snapshots/test_diagnostics.ambr
View File

@ -3,14 +3,15 @@
dict({
'config_entry': dict({
'data': dict({
'extended_data': True,
'has_password': True,
'host': '**REDACTED**',
'name': 'Solarlog test 1 2 3',
'password': 'pwd',
}),
'disabled_by': None,
'domain': 'solarlog',
'entry_id': 'ce5f5431554d101905d31797e1232da8',
'minor_version': 2,
'minor_version': 3,
'options': dict({
}),
'pref_disable_new_entities': False,

175
tests/components/solarlog/test_config_flow.py
View File

@ -1,14 +1,18 @@
"""Test the solarlog config flow."""
from unittest.mock import AsyncMock, patch
from unittest.mock import AsyncMock
import pytest
from solarlog_cli.solarlog_exceptions import SolarLogConnectionError, SolarLogError
from solarlog_cli.solarlog_exceptions import (
SolarLogAuthenticationError,
SolarLogConnectionError,
SolarLogError,
)
from homeassistant.components.solarlog import config_flow
from homeassistant.components.solarlog.const import DOMAIN
from homeassistant.components.solarlog.const import CONF_HAS_PWD, DOMAIN
from homeassistant.config_entries import SOURCE_RECONFIGURE, SOURCE_USER
from homeassistant.const import CONF_HOST, CONF_NAME
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PASSWORD
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
@ -17,6 +21,7 @@ from .const import HOST, NAME
from tests.common import MockConfigEntry
@pytest.mark.usefixtures("test_connect")
async def test_form(hass: HomeAssistant, mock_setup_entry: AsyncMock) -> None:
"""Test we get the form."""
@ -26,22 +31,16 @@ async def test_form(hass: HomeAssistant, mock_setup_entry: AsyncMock) -> None:
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {}
with (
patch(
"homeassistant.components.solarlog.config_flow.SolarLogConfigFlow._test_connection",
return_value=True,
),
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_HOST: HOST, CONF_NAME: NAME, "extended_data": False},
)
await hass.async_block_till_done()
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_HOST: HOST, CONF_NAME: NAME, CONF_HAS_PWD: False},
)
await hass.async_block_till_done()
assert result2["type"] is FlowResultType.CREATE_ENTRY
assert result2["title"] == "solarlog_test_1_2_3"
assert result2["data"][CONF_HOST] == "http://1.1.1.1"
assert result2["data"]["extended_data"] is False
assert result2["data"][CONF_HAS_PWD] is False
assert len(mock_setup_entry.mock_calls) == 1
@ -67,7 +66,7 @@ async def test_user(
# tests with all provided
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_HOST: HOST, CONF_NAME: NAME, "extended_data": True}
result["flow_id"], {CONF_HOST: HOST, CONF_NAME: NAME, CONF_HAS_PWD: False}
)
await hass.async_block_till_done()
@ -78,16 +77,23 @@ async def test_user(
@pytest.mark.parametrize(
("exception", "error"),
("exception1", "error1", "exception2", "error2"),
[
(SolarLogConnectionError, {CONF_HOST: "cannot_connect"}),
(SolarLogError, {CONF_HOST: "unknown"}),
(
SolarLogConnectionError,
{CONF_HOST: "cannot_connect"},
SolarLogAuthenticationError,
{CONF_HOST: "password_error"},
),
(SolarLogError, {CONF_HOST: "unknown"}, SolarLogError, {CONF_HOST: "unknown"}),
],
)
async def test_form_exceptions(
hass: HomeAssistant,
exception: Exception,
error: dict[str, str],
exception1: Exception,
error1: dict[str, str],
exception2: Exception,
error2: dict[str, str],
mock_solarlog_connector: AsyncMock,
) -> None:
"""Test we can handle Form exceptions."""
@ -97,30 +103,57 @@ async def test_form_exceptions(
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
mock_solarlog_connector.test_connection.side_effect = exception
mock_solarlog_connector.test_connection.side_effect = exception1
# tests with connection error
result = await flow.async_step_user(
{CONF_NAME: NAME, CONF_HOST: HOST, "extended_data": False}
{CONF_NAME: NAME, CONF_HOST: HOST, CONF_HAS_PWD: False}
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == error
assert result["errors"] == error1
# tests with password error
mock_solarlog_connector.test_connection.side_effect = None
mock_solarlog_connector.test_extended_data_available.side_effect = exception2
# tests with all provided
result = await flow.async_step_user(
{CONF_NAME: NAME, CONF_HOST: HOST, "extended_data": False}
{CONF_NAME: NAME, CONF_HOST: HOST, CONF_HAS_PWD: True}
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "password"
result = await flow.async_step_password({CONF_PASSWORD: "pwd"})
await hass.async_block_till_done()
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "password"
assert result["errors"] == error2
mock_solarlog_connector.test_extended_data_available.side_effect = None
# tests with all provided (no password)
result = await flow.async_step_user(
{CONF_NAME: NAME, CONF_HOST: HOST, CONF_HAS_PWD: False}
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "solarlog_test_1_2_3"
assert result["data"][CONF_HOST] == HOST
assert result["data"]["extended_data"] is False
assert result["data"][CONF_HAS_PWD] is False
# tests with all provided (password)
result = await flow.async_step_password({CONF_PASSWORD: "pwd"})
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "solarlog_test_1_2_3"
assert result["data"][CONF_PASSWORD] == "pwd"
async def test_abort_if_already_setup(hass: HomeAssistant, test_connect: None) -> None:
@ -140,14 +173,25 @@ async def test_abort_if_already_setup(hass: HomeAssistant, test_connect: None) -
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_HOST: HOST, CONF_NAME: "solarlog_test_7_8_9", "extended_data": False},
{CONF_HOST: HOST, CONF_NAME: "solarlog_test_7_8_9", CONF_HAS_PWD: False},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
@pytest.mark.parametrize(
("has_password", "password"),
[
(True, "pwd"),
(False, ""),
],
)
async def test_reconfigure_flow(
hass: HomeAssistant, mock_setup_entry: AsyncMock
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_solarlog_connector: AsyncMock,
has_password: bool,
password: str,
) -> None:
"""Test config flow options."""
entry = MockConfigEntry(
@ -155,8 +199,9 @@ async def test_reconfigure_flow(
title="solarlog_test_1_2_3",
data={
CONF_HOST: HOST,
"extended_data": False,
CONF_HAS_PWD: False,
},
minor_version=3,
)
entry.add_to_hass(hass)
@ -170,11 +215,77 @@ async def test_reconfigure_flow(
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reconfigure"
# test with all data provided
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"extended_data": True}
result["flow_id"], {CONF_HAS_PWD: True, CONF_PASSWORD: password}
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reconfigure_successful"
assert len(mock_setup_entry.mock_calls) == 1
entry = hass.config_entries.async_get_entry(entry.entry_id)
assert entry
assert entry.title == "solarlog_test_1_2_3"
assert entry.data[CONF_HAS_PWD] == has_password
assert entry.data[CONF_PASSWORD] == password
@pytest.mark.parametrize(
("exception", "error"),
[
(SolarLogAuthenticationError, {CONF_HOST: "password_error"}),
(SolarLogError, {CONF_HOST: "unknown"}),
],
)
async def test_reauth(
hass: HomeAssistant,
exception: Exception,
error: dict[str, str],
mock_solarlog_connector: AsyncMock,
mock_setup_entry: AsyncMock,
) -> None:
"""Test reauth-flow works."""
entry = MockConfigEntry(
domain=DOMAIN,
title="solarlog_test_1_2_3",
data={
CONF_HOST: HOST,
CONF_HAS_PWD: True,
CONF_PASSWORD: "pwd",
},
minor_version=3,
)
entry.add_to_hass(hass)
result = await entry.start_reauth_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
mock_solarlog_connector.test_extended_data_available.side_effect = exception
# tests with connection error
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_PASSWORD: "other_pwd"},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
assert result["errors"] == error
mock_solarlog_connector.test_extended_data_available.side_effect = None
# tests with all information provided
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_PASSWORD: "other_pwd"},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reauth_successful"
assert entry.data[CONF_PASSWORD] == "other_pwd"

109
tests/components/solarlog/test_init.py
View File

@ -2,12 +2,19 @@
from unittest.mock import AsyncMock
from solarlog_cli.solarlog_exceptions import SolarLogConnectionError
import pytest
from solarlog_cli.solarlog_exceptions import (
SolarLogAuthenticationError,
SolarLogConnectionError,
SolarLogError,
SolarLogUpdateError,
)
from homeassistant.components.solarlog.const import DOMAIN
from homeassistant.components.solarlog.const import CONF_HAS_PWD, DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import CONF_HOST, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers.device_registry import DeviceRegistry
from homeassistant.helpers.entity_registry import EntityRegistry
@ -32,27 +39,103 @@ async def test_load_unload(
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
async def test_raise_config_entry_not_ready_when_offline(
@pytest.mark.parametrize(
("exception", "error"),
[
(SolarLogAuthenticationError, ConfigEntryState.SETUP_ERROR),
(SolarLogUpdateError, ConfigEntryState.SETUP_RETRY),
],
)
async def test_setup_error(
hass: HomeAssistant,
exception: SolarLogError,
error: str,
mock_config_entry: MockConfigEntry,
mock_solarlog_connector: AsyncMock,
) -> None:
"""Config entry state is SETUP_RETRY when Solarlog is offline."""
"""Test errors in setting up coordinator (i.e. login error)."""
mock_solarlog_connector.update_data.side_effect = SolarLogConnectionError
mock_solarlog_connector.login.side_effect = exception
await setup_platform(hass, mock_config_entry, [Platform.SENSOR])
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
assert mock_config_entry.state == error
if error == ConfigEntryState.SETUP_RETRY:
assert len(hass.config_entries.flow.async_progress()) == 0
@pytest.mark.parametrize(
("login_side_effect", "login_return_value", "entry_state"),
[
(SolarLogAuthenticationError, False, ConfigEntryState.SETUP_ERROR),
(ConfigEntryNotReady, False, ConfigEntryState.SETUP_RETRY),
(None, False, ConfigEntryState.SETUP_ERROR),
(None, True, ConfigEntryState.SETUP_RETRY),
],
)
async def test_auth_error_during_first_refresh(
hass: HomeAssistant,
login_side_effect: Exception | None,
login_return_value: bool,
entry_state: str,
mock_config_entry: MockConfigEntry,
mock_solarlog_connector: AsyncMock,
) -> None:
"""Test the correct exceptions are thrown for auth error during first refresh."""
mock_solarlog_connector.password.return_value = ""
mock_solarlog_connector.update_data.side_effect = SolarLogAuthenticationError
mock_solarlog_connector.login.return_value = login_return_value
mock_solarlog_connector.login.side_effect = login_side_effect
await setup_platform(hass, mock_config_entry, [Platform.SENSOR])
await hass.async_block_till_done()
assert mock_config_entry.state == entry_state
@pytest.mark.parametrize(
("exception"),
[
(SolarLogConnectionError),
(SolarLogUpdateError),
],
)
async def test_other_exceptions_during_first_refresh(
hass: HomeAssistant,
exception: SolarLogError,
mock_config_entry: MockConfigEntry,
mock_solarlog_connector: AsyncMock,
) -> None:
"""Test the correct exceptions are thrown during first refresh."""
mock_solarlog_connector.update_data.side_effect = exception
await setup_platform(hass, mock_config_entry, [Platform.SENSOR])
await hass.async_block_till_done()
assert mock_config_entry.state == ConfigEntryState.SETUP_RETRY
assert len(hass.config_entries.flow.async_progress()) == 0
@pytest.mark.parametrize(
("minor_version", "suffix"),
[
(1, "time"),
(2, "last_updated"),
],
)
async def test_migrate_config_entry(
hass: HomeAssistant,
minor_version: int,
suffix: str,
device_registry: DeviceRegistry,
entity_registry: EntityRegistry,
mock_solarlog_connector: AsyncMock,
) -> None:
"""Test successful migration of entry data."""
entry = MockConfigEntry(
@ -62,7 +145,7 @@ async def test_migrate_config_entry(
CONF_HOST: HOST,
},
version=1,
minor_version=1,
minor_version=minor_version,
)
entry.add_to_hass(hass)
@ -72,17 +155,19 @@ async def test_migrate_config_entry(
manufacturer="Solar-Log",
name="solarlog",
)
uid = f"{entry.entry_id}_{suffix}"
sensor_entity = entity_registry.async_get_or_create(
config_entry=entry,
platform=DOMAIN,
domain=Platform.SENSOR,
unique_id=f"{entry.entry_id}_time",
unique_id=uid,
device_id=device.id,
)
assert entry.version == 1
assert entry.minor_version == 1
assert sensor_entity.unique_id == f"{entry.entry_id}_time"
assert entry.minor_version == minor_version
assert sensor_entity.unique_id == f"{entry.entry_id}_{suffix}"
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
@ -92,6 +177,6 @@ async def test_migrate_config_entry(
assert entity_migrated.unique_id == f"{entry.entry_id}_last_updated"
assert entry.version == 1
assert entry.minor_version == 2
assert entry.minor_version == 3
assert entry.data[CONF_HOST] == HOST
assert entry.data["extended_data"] is False
assert entry.data[CONF_HAS_PWD] is False