Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-12-26 05:57:01 +00:00
Code Activity

Add module-level statistics to SolarEdge (#152581)

Browse Source
This commit is contained in:
tronikos
2025-10-10 04:07:39 -07:00
committed by GitHub
parent 16d4c6c95a
commit 17e997ee18
15 changed files with 1088 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
CODEOWNERS generated
View File

@@ -1479,8 +1479,8 @@ build.json @home-assistant/supervisor
/tests/components/snoo/ @Lash-L
/homeassistant/components/snooz/ @AustinBrunkhorst
/tests/components/snooz/ @AustinBrunkhorst
/homeassistant/components/solaredge/ @frenck @bdraco
/tests/components/solaredge/ @frenck @bdraco
/homeassistant/components/solaredge/ @frenck @bdraco @tronikos
/tests/components/solaredge/ @frenck @bdraco @tronikos
/homeassistant/components/solaredge_local/ @drobtravels @scheric
/homeassistant/components/solarlog/ @Ernst79 @dontinelli
/tests/components/solarlog/ @Ernst79 @dontinelli

47
homeassistant/components/solaredge/__init__.py
View File

@@ -7,12 +7,13 @@ import socket
from aiohttp import ClientError
from aiosolaredge import SolarEdge
from homeassistant.const import CONF_API_KEY, Platform
from homeassistant.const import CONF_API_KEY, CONF_USERNAME, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import CONF_SITE_ID, LOGGER
from .const import CONF_SITE_ID, DATA_API_CLIENT, DATA_MODULES_COORDINATOR, LOGGER
from .coordinator import SolarEdgeModulesCoordinator
from .types import SolarEdgeConfigEntry, SolarEdgeData
PLATFORMS = [Platform.SENSOR]
@@ -20,25 +21,37 @@ PLATFORMS = [Platform.SENSOR]
async def async_setup_entry(hass: HomeAssistant, entry: SolarEdgeConfigEntry) -> bool:
"""Set up SolarEdge from a config entry."""
session = async_get_clientsession(hass)
api = SolarEdge(entry.data[CONF_API_KEY], session)
entry.runtime_data = SolarEdgeData()
site_id = entry.data[CONF_SITE_ID]
try:
response = await api.get_details(entry.data[CONF_SITE_ID])
except (TimeoutError, ClientError, socket.gaierror) as ex:
LOGGER.error("Could not retrieve details from SolarEdge API")
raise ConfigEntryNotReady from ex
# Setup for API key (sensors)
if CONF_API_KEY in entry.data:
session = async_get_clientsession(hass)
api = SolarEdge(entry.data[CONF_API_KEY], session)
if "details" not in response:
LOGGER.error("Missing details data in SolarEdge response")
raise ConfigEntryNotReady
try:
response = await api.get_details(site_id)
except (TimeoutError, ClientError, socket.gaierror) as ex:
LOGGER.error("Could not retrieve details from SolarEdge API")
raise ConfigEntryNotReady from ex
if response["details"].get("status", "").lower() != "active":
LOGGER.error("SolarEdge site is not active")
return False
if "details" not in response:
LOGGER.error("Missing details data in SolarEdge response")
raise ConfigEntryNotReady
if response["details"].get("status", "").lower() != "active":
LOGGER.error("SolarEdge site is not active")
return False
entry.runtime_data[DATA_API_CLIENT] = api
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
# Setup for username/password (modules statistics)
if CONF_USERNAME in entry.data:
coordinator = SolarEdgeModulesCoordinator(hass, entry)
await coordinator.async_config_entry_first_refresh()
entry.runtime_data[DATA_MODULES_COORDINATOR] = coordinator
entry.runtime_data = SolarEdgeData(api_client=api)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True

109
homeassistant/components/solaredge/config_flow.py
View File

@@ -5,17 +5,25 @@ from __future__ import annotations
import socket
from typing import Any
from aiohttp import ClientError
from aiohttp import ClientError, ClientResponseError
import aiosolaredge
from solaredge_web import SolarEdgeWeb
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_API_KEY, CONF_NAME
from homeassistant.const import CONF_API_KEY, CONF_NAME, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import callback
from homeassistant.data_entry_flow import section
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.util import slugify
from .const import CONF_SITE_ID, DEFAULT_NAME, DOMAIN
from .const import (
CONF_SECTION_API_AUTH,
CONF_SECTION_WEB_AUTH,
CONF_SITE_ID,
DEFAULT_NAME,
DOMAIN,
)
class SolarEdgeConfigFlow(ConfigFlow, domain=DOMAIN):
@@ -50,13 +58,36 @@ class SolarEdgeConfigFlow(ConfigFlow, domain=DOMAIN):
self._errors[CONF_SITE_ID] = "site_not_active"
return False
except (TimeoutError, ClientError, socket.gaierror):
self._errors[CONF_SITE_ID] = "could_not_connect"
self._errors[CONF_SITE_ID] = "cannot_connect"
return False
except KeyError:
self._errors[CONF_SITE_ID] = "invalid_api_key"
return False
return True
async def _async_check_web_login(
self, site_id: str, username: str, password: str
) -> bool:
"""Validate the user input allows us to connect to the web service."""
api = SolarEdgeWeb(
username=username,
password=password,
site_id=site_id,
session=async_get_clientsession(self.hass),
)
try:
await api.async_get_equipment()
except ClientResponseError as err:
if err.status in (401, 403):
self._errors["base"] = "invalid_auth"
else:
self._errors["base"] = "cannot_connect"
return False
except (TimeoutError, ClientError):
self._errors["base"] = "cannot_connect"
return False
return True
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
@@ -64,19 +95,34 @@ class SolarEdgeConfigFlow(ConfigFlow, domain=DOMAIN):
self._errors = {}
if user_input is not None:
name = slugify(user_input.get(CONF_NAME, DEFAULT_NAME))
if self._site_in_configuration_exists(user_input[CONF_SITE_ID]):
site_id = user_input[CONF_SITE_ID]
api_auth = user_input.get(CONF_SECTION_API_AUTH, {})
web_auth = user_input.get(CONF_SECTION_WEB_AUTH, {})
api_key = api_auth.get(CONF_API_KEY)
username = web_auth.get(CONF_USERNAME)
if self._site_in_configuration_exists(site_id):
self._errors[CONF_SITE_ID] = "already_configured"
elif not api_key and not username:
self._errors["base"] = "auth_missing"
else:
site = user_input[CONF_SITE_ID]
api = user_input[CONF_API_KEY]
can_connect = await self._async_check_site(site, api)
if can_connect:
return self.async_create_entry(
title=name, data={CONF_SITE_ID: site, CONF_API_KEY: api}
api_key_ok = True
if api_key:
api_key_ok = await self._async_check_site(site_id, api_key)
web_login_ok = True
if api_key_ok and username:
web_login_ok = await self._async_check_web_login(
site_id, username, web_auth[CONF_PASSWORD]
)
if api_key_ok and web_login_ok:
data = {CONF_SITE_ID: site_id}
data.update(api_auth)
data.update(web_auth)
return self.async_create_entry(title=name, data=data)
else:
user_input = {CONF_NAME: DEFAULT_NAME, CONF_SITE_ID: "", CONF_API_KEY: ""}
user_input = {}
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
@@ -84,8 +130,43 @@ class SolarEdgeConfigFlow(ConfigFlow, domain=DOMAIN):
vol.Required(
CONF_NAME, default=user_input.get(CONF_NAME, DEFAULT_NAME)
): str,
vol.Required(CONF_SITE_ID, default=user_input[CONF_SITE_ID]): str,
vol.Required(CONF_API_KEY, default=user_input[CONF_API_KEY]): str,
vol.Required(
CONF_SITE_ID, default=user_input.get(CONF_SITE_ID, "")
): str,
vol.Optional(CONF_SECTION_API_AUTH): section(
vol.Schema(
{
vol.Optional(
CONF_API_KEY,
default=user_input.get(
CONF_SECTION_API_AUTH, {}
).get(CONF_API_KEY, ""),
): str,
}
),
options={"collapsed": False},
),
vol.Optional(CONF_SECTION_WEB_AUTH): section(
vol.Schema(
{
vol.Inclusive(
CONF_USERNAME,
"web_account",
default=user_input.get(
CONF_SECTION_WEB_AUTH, {}
).get(CONF_USERNAME, ""),
): str,
vol.Inclusive(
CONF_PASSWORD,
"web_account",
default=user_input.get(
CONF_SECTION_WEB_AUTH, {}
).get(CONF_PASSWORD, ""),
): str,
}
),
options={"collapsed": False},
),
}
),
errors=self._errors,

4
homeassistant/components/solaredge/const.py
View File

@@ -9,9 +9,12 @@ DOMAIN = "solaredge"
LOGGER = logging.getLogger(__package__)
DATA_API_CLIENT: Final = "api_client"
DATA_MODULES_COORDINATOR: Final = "modules_coordinator"
# Config for solaredge monitoring api requests.
CONF_SITE_ID = "site_id"
CONF_SECTION_API_AUTH = "api_auth"
CONF_SECTION_WEB_AUTH = "web_auth"
DEFAULT_NAME = "SolarEdge"
OVERVIEW_UPDATE_DELAY = timedelta(minutes=15)
@@ -19,5 +22,6 @@ DETAILS_UPDATE_DELAY = timedelta(hours=12)
INVENTORY_UPDATE_DELAY = timedelta(hours=12)
POWER_FLOW_UPDATE_DELAY = timedelta(minutes=15)
ENERGY_DETAILS_DELAY = timedelta(minutes=15)
MODULE_STATISTICS_UPDATE_DELAY = timedelta(hours=12)
SCAN_INTERVAL = timedelta(minutes=15)

172
homeassistant/components/solaredge/coordinator.py
View File

@@ -3,20 +3,40 @@
from __future__ import annotations
from abc import ABC, abstractmethod
from collections.abc import Iterable
from datetime import date, datetime, timedelta
from typing import TYPE_CHECKING, Any
from aiosolaredge import SolarEdge
from solaredge_web import EnergyData, SolarEdgeWeb, TimeUnit
from stringcase import snakecase
from homeassistant.components.recorder import get_instance
from homeassistant.components.recorder.models import (
StatisticData,
StatisticMeanType,
StatisticMetaData,
)
from homeassistant.components.recorder.statistics import (
async_add_external_statistics,
get_last_statistics,
statistics_during_period,
)
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME, UnitOfEnergy
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import aiohttp_client
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from homeassistant.util import dt as dt_util
from homeassistant.util.unit_conversion import EnergyConverter
from .const import (
CONF_SITE_ID,
DETAILS_UPDATE_DELAY,
DOMAIN,
ENERGY_DETAILS_DELAY,
INVENTORY_UPDATE_DELAY,
LOGGER,
MODULE_STATISTICS_UPDATE_DELAY,
OVERVIEW_UPDATE_DELAY,
POWER_FLOW_UPDATE_DELAY,
)
@@ -313,3 +333,155 @@ class SolarEdgePowerFlowDataService(SolarEdgeDataService):
self.attributes[key]["soc"] = value["chargeLevel"]
LOGGER.debug("Updated SolarEdge power flow: %s, %s", self.data, self.attributes)
class SolarEdgeModulesCoordinator(DataUpdateCoordinator[None]):
"""Handle fetching SolarEdge Modules data and inserting statistics."""
config_entry: SolarEdgeConfigEntry
def __init__(
self,
hass: HomeAssistant,
config_entry: SolarEdgeConfigEntry,
) -> None:
"""Initialize the data handler."""
super().__init__(
hass,
LOGGER,
config_entry=config_entry,
name="SolarEdge Modules",
# API refreshes every 15 minutes, but since we only have statistics
# and no sensors, refresh every 12h.
update_interval=MODULE_STATISTICS_UPDATE_DELAY,
)
self.api = SolarEdgeWeb(
username=config_entry.data[CONF_USERNAME],
password=config_entry.data[CONF_PASSWORD],
site_id=config_entry.data[CONF_SITE_ID],
session=aiohttp_client.async_get_clientsession(hass),
)
self.site_id = config_entry.data[CONF_SITE_ID]
self.title = config_entry.title
@callback
def _dummy_listener() -> None:
pass
# Force the coordinator to periodically update by registering a listener.
# Needed because there are no sensors added.
self.async_add_listener(_dummy_listener)
async def _async_update_data(self) -> None:
"""Fetch data from API endpoint and update statistics."""
equipment: dict[int, dict[str, Any]] = await self.api.async_get_equipment()
# We fetch last week's data from the API and refresh every 12h so we overwrite recent
# statistics. This is intended to allow adding any corrected/updated data from the API.
energy_data_list: list[EnergyData] = await self.api.async_get_energy_data(
TimeUnit.WEEK
)
if not energy_data_list:
LOGGER.warning(
"No data received from SolarEdge API for site: %s", self.site_id
)
return
last_sums = await self._async_get_last_sums(
equipment.keys(),
energy_data_list[0].start_time.replace(
tzinfo=dt_util.get_default_time_zone()
),
)
for equipment_id, equipment_data in equipment.items():
display_name = equipment_data.get(
"displayName", f"Equipment {equipment_id}"
)
statistic_id = self.get_statistic_id(equipment_id)
statistic_metadata = StatisticMetaData(
mean_type=StatisticMeanType.ARITHMETIC,
has_sum=True,
name=f"{self.title} {display_name}",
source=DOMAIN,
statistic_id=statistic_id,
unit_class=EnergyConverter.UNIT_CLASS,
unit_of_measurement=UnitOfEnergy.WATT_HOUR,
)
statistic_sum = last_sums[statistic_id]
statistics = []
current_hour_sum = 0.0
current_hour_count = 0
for energy_data in energy_data_list:
start_time = energy_data.start_time.replace(
tzinfo=dt_util.get_default_time_zone()
)
value = energy_data.values.get(equipment_id, 0.0)
current_hour_sum += value
current_hour_count += 1
if start_time.minute != 45:
continue
# API returns data every 15 minutes; aggregate to 1-hour statistics
# when we reach the energy_data for the last 15 minutes of the hour.
current_avg = current_hour_sum / current_hour_count
statistic_sum += current_avg
statistics.append(
StatisticData(
start=start_time - timedelta(minutes=45),
state=current_avg,
sum=statistic_sum,
)
)
current_hour_sum = 0.0
current_hour_count = 0
LOGGER.debug(
"Adding %s statistics for %s %s",
len(statistics),
statistic_id,
display_name,
)
async_add_external_statistics(self.hass, statistic_metadata, statistics)
def get_statistic_id(self, equipment_id: int) -> str:
"""Return the statistic ID for this equipment_id."""
return f"{DOMAIN}:{self.site_id}_{equipment_id}"
async def _async_get_last_sums(
self, equipment_ids: Iterable[int], start_time: datetime
) -> dict[str, float]:
"""Get the last sum from the recorder before start_time for each statistic."""
start = start_time - timedelta(hours=1)
statistic_ids = {self.get_statistic_id(eq_id) for eq_id in equipment_ids}
LOGGER.debug(
"Getting sum for %s statistic IDs at: %s", len(statistic_ids), start
)
current_stats = await get_instance(self.hass).async_add_executor_job(
statistics_during_period,
self.hass,
start,
start + timedelta(seconds=1),
statistic_ids,
"hour",
None,
{"sum"},
)
result = {}
for statistic_id in statistic_ids:
if statistic_id in current_stats:
statistic_sum = current_stats[statistic_id][0]["sum"]
else:
# If no statistics found right before start_time, try to get the last statistic
# but use it only if it's before start_time.
# This is needed if the integration hasn't run successfully for at least a week.
last_stat = await get_instance(self.hass).async_add_executor_job(
get_last_statistics, self.hass, 1, statistic_id, True, {"sum"}
)
if (
last_stat
and last_stat[statistic_id][0]["start"] < start_time.timestamp()
):
statistic_sum = last_stat[statistic_id][0]["sum"]
else:
# Expected for new installations or if the statistics were cleared,
# e.g. from the developer tools
statistic_sum = 0.0
assert isinstance(statistic_sum, float)
result[statistic_id] = statistic_sum
return result

11
homeassistant/components/solaredge/manifest.json
View File

@@ -1,8 +1,9 @@
{
"domain": "solaredge",
"name": "SolarEdge",
"codeowners": ["@frenck", "@bdraco"],
"codeowners": ["@frenck", "@bdraco", "@tronikos"],
"config_flow": true,
"dependencies": ["recorder"],
"dhcp": [
{
"hostname": "target",
@@ -12,6 +13,10 @@
"documentation": "https://www.home-assistant.io/integrations/solaredge",
"integration_type": "device",
"iot_class": "cloud_polling",
"loggers": ["aiosolaredge"],
"requirements": ["aiosolaredge==0.2.0", "stringcase==1.2.0"]
"loggers": ["aiosolaredge", "solaredge_web"],
"requirements": [
"aiosolaredge==0.2.0",
"stringcase==1.2.0",
"solaredge-web==0.0.1"
]
}

5
homeassistant/components/solaredge/sensor.py
View File

@@ -204,7 +204,10 @@ async def async_setup_entry(
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Add an solarEdge entry."""
# Add the needed sensors to hass
# Add sensor entities only if API key is configured
if DATA_API_CLIENT not in entry.runtime_data:
return
api = entry.runtime_data[DATA_API_CLIENT]
sensor_factory = SolarEdgeSensorFactory(hass, entry, entry.data[CONF_SITE_ID], api)
for service in sensor_factory.all_services:

36
homeassistant/components/solaredge/strings.json
View File

@@ -2,19 +2,47 @@
"config": {
"step": {
"user": {
"title": "Define the API parameters for this installation",
"title": "Set up your SolarEdge integration",
"description": "Provide your site ID and at least one method of authentication",
"data": {
"name": "The name of this installation",
"site_id": "The SolarEdge site ID",
"api_key": "[%key:common::config_flow::data::api_key%]"
"api_key": "[%key:common::config_flow::data::api_key%]",
"username": "[%key:common::config_flow::data::username%]",
"password": "[%key:common::config_flow::data::password%]"
},
"data_description": {
"api_key": "Optional, used for general site sensors",
"username": "Optional, used for detailed module-level production statistics",
"password": "Required if username is provided"
},
"sections": {
"api_auth": {
"name": "API key authentication",
"description": "Optionally provide your SolarEdge API key. Used for real-time detailed site sensors",
"data": {
"api_key": "[%key:common::config_flow::data::api_key%]"
}
},
"web_auth": {
"name": "Web account authentication",
"description": "Optionally provide your SolarEdge web portal credentials. Used for non real-time module-level production statistics",
"data": {
"username": "[%key:common::config_flow::data::username%]",
"password": "[%key:common::config_flow::data::password%]"
}
}
}
}
},
"error": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"invalid_api_key": "[%key:common::config_flow::error::invalid_api_key%]",
"site_not_active": "The site is not active",
"could_not_connect": "Could not connect to the SolarEdge API"
"site_not_active": "The site is not active for the provided API key",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"unknown": "[%key:common::config_flow::error::unknown%]",
"auth_missing": "You must provide either an API key or a username and password."
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"

5
homeassistant/components/solaredge/types.py
View File

@@ -8,10 +8,13 @@ from aiosolaredge import SolarEdge
from homeassistant.config_entries import ConfigEntry
from .coordinator import SolarEdgeModulesCoordinator
type SolarEdgeConfigEntry = ConfigEntry[SolarEdgeData]
class SolarEdgeData(TypedDict):
class SolarEdgeData(TypedDict, total=False):
"""Data for the solaredge integration."""
api_client: SolarEdge
modules_coordinator: SolarEdgeModulesCoordinator

3
requirements_all.txt generated
View File

@@ -2854,6 +2854,9 @@ soco==0.30.12
# homeassistant.components.solaredge_local
solaredge-local==0.2.3
# homeassistant.components.solaredge
solaredge-web==0.0.1
# homeassistant.components.solarlog
solarlog_cli==0.6.0

3
requirements_test_all.txt generated
View File

@@ -2364,6 +2364,9 @@ snapcast==2.3.6
# homeassistant.components.sonos
soco==0.30.12
# homeassistant.components.solaredge
solaredge-web==0.0.1
# homeassistant.components.solarlog
solarlog_cli==0.6.0

60
tests/components/solaredge/conftest.py Normal file
View File

@@ -0,0 +1,60 @@
"""Common fixtures for the SolarEdge tests."""
from collections.abc import Generator
from unittest.mock import AsyncMock, Mock, patch
import pytest
SITE_ID = "1a2b3c4d5e6f7g8h"
API_KEY = "a1b2c3d4e5f6g7h8"
USERNAME = "test-username"
PASSWORD = "test-password"
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Override async_setup_entry."""
with patch(
"homeassistant.components.solaredge.async_setup_entry",
return_value=True,
) as mock_setup_entry:
yield mock_setup_entry
@pytest.fixture(name="solaredge_api")
def mock_solaredge_api_fixture() -> Generator[Mock]:
"""Mock a successful SolarEdge Monitoring API."""
api = Mock()
api.get_details = AsyncMock(return_value={"details": {"status": "active"}})
with (
patch(
"homeassistant.components.solaredge.config_flow.aiosolaredge.SolarEdge",
return_value=api,
),
patch(
"homeassistant.components.solaredge.SolarEdge",
return_value=api,
),
):
yield api
@pytest.fixture(name="solaredge_web_api")
def mock_solaredge_web_api_fixture() -> Generator[AsyncMock]:
"""Mock a successful SolarEdge Web API."""
with (
patch(
"homeassistant.components.solaredge.config_flow.SolarEdgeWeb", autospec=True
) as mock_web_api_flow,
patch(
"homeassistant.components.solaredge.coordinator.SolarEdgeWeb", autospec=True
) as mock_web_api_coord,
):
# Ensure both patches use the same mock instance
api = mock_web_api_flow.return_value
mock_web_api_coord.return_value = api
api.async_get_equipment.return_value = {
1001: {"displayName": "1.1"},
1002: {"displayName": "1.2"},
}
yield api

287
tests/components/solaredge/test_config_flow.py
View File

@@ -1,48 +1,60 @@
"""Tests for the SolarEdge config flow."""
from unittest.mock import AsyncMock, Mock, patch
from unittest.mock import AsyncMock, Mock
from aiohttp import ClientError
from aiohttp import ClientError, ClientResponseError
import pytest
from homeassistant.components.solaredge.const import CONF_SITE_ID, DEFAULT_NAME, DOMAIN
from homeassistant.components.recorder import Recorder
from homeassistant.components.solaredge.const import (
CONF_SECTION_API_AUTH,
CONF_SECTION_WEB_AUTH,
CONF_SITE_ID,
DEFAULT_NAME,
DOMAIN,
)
from homeassistant.config_entries import SOURCE_IGNORE, SOURCE_USER
from homeassistant.const import CONF_API_KEY, CONF_NAME
from homeassistant.const import CONF_API_KEY, CONF_NAME, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from .conftest import API_KEY, PASSWORD, SITE_ID, USERNAME
from tests.common import MockConfigEntry
NAME = "solaredge site 1 2 3"
SITE_ID = "1a2b3c4d5e6f7g8h"
API_KEY = "a1b2c3d4e5f6g7h8"
@pytest.fixture(name="test_api")
def mock_controller():
"""Mock a successful Solaredge API."""
api = Mock()
api.get_details = AsyncMock(return_value={"details": {"status": "active"}})
with patch(
"homeassistant.components.solaredge.config_flow.aiosolaredge.SolarEdge",
return_value=api,
):
yield api
@pytest.fixture(autouse=True)
def solaredge_api_fixture(solaredge_api: Mock) -> None:
"""Mock the solaredge API."""
async def test_user(hass: HomeAssistant, test_api: Mock) -> None:
"""Test user config."""
@pytest.fixture(autouse=True)
def solaredge_web_api_fixture(solaredge_web_api: AsyncMock) -> None:
"""Mock the solaredge web API."""
async def test_user_api_key(
recorder_mock: Recorder,
hass: HomeAssistant,
solaredge_api: Mock,
mock_setup_entry: AsyncMock,
) -> None:
"""Test user config with API key."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result.get("type") is FlowResultType.FORM
assert result.get("step_id") == "user"
# test with all provided
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data={CONF_NAME: NAME, CONF_API_KEY: API_KEY, CONF_SITE_ID: SITE_ID},
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_NAME: NAME,
CONF_SITE_ID: SITE_ID,
CONF_SECTION_API_AUTH: {CONF_API_KEY: API_KEY},
},
)
assert result.get("type") is FlowResultType.CREATE_ENTRY
assert result.get("title") == "solaredge_site_1_2_3"
@@ -51,27 +63,110 @@ async def test_user(hass: HomeAssistant, test_api: Mock) -> None:
assert data
assert data[CONF_SITE_ID] == SITE_ID
assert data[CONF_API_KEY] == API_KEY
assert CONF_USERNAME not in data
assert CONF_PASSWORD not in data
assert len(mock_setup_entry.mock_calls) == 1
async def test_abort_if_already_setup(hass: HomeAssistant, test_api: str) -> None:
async def test_user_web_login(
recorder_mock: Recorder,
hass: HomeAssistant,
solaredge_web_api: AsyncMock,
mock_setup_entry: AsyncMock,
) -> None:
"""Test user config with web login."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_NAME: NAME,
CONF_SITE_ID: SITE_ID,
CONF_SECTION_WEB_AUTH: {
CONF_USERNAME: USERNAME,
CONF_PASSWORD: PASSWORD,
},
},
)
assert result.get("type") is FlowResultType.CREATE_ENTRY
assert result.get("title") == "solaredge_site_1_2_3"
data = result.get("data")
assert data
assert data[CONF_SITE_ID] == SITE_ID
assert data[CONF_USERNAME] == USERNAME
assert data[CONF_PASSWORD] == PASSWORD
assert CONF_API_KEY not in data
assert len(mock_setup_entry.mock_calls) == 1
solaredge_web_api.async_get_equipment.assert_awaited_once()
async def test_user_both_auth(
recorder_mock: Recorder,
hass: HomeAssistant,
solaredge_api: Mock,
solaredge_web_api: AsyncMock,
mock_setup_entry: AsyncMock,
) -> None:
"""Test user config with both API key and web login."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_NAME: NAME,
CONF_SITE_ID: SITE_ID,
CONF_SECTION_API_AUTH: {CONF_API_KEY: API_KEY},
CONF_SECTION_WEB_AUTH: {
CONF_USERNAME: USERNAME,
CONF_PASSWORD: PASSWORD,
},
},
)
assert result.get("type") is FlowResultType.CREATE_ENTRY
data = result.get("data")
assert data
assert data[CONF_SITE_ID] == SITE_ID
assert data[CONF_API_KEY] == API_KEY
assert data[CONF_USERNAME] == USERNAME
assert data[CONF_PASSWORD] == PASSWORD
assert len(mock_setup_entry.mock_calls) == 1
async def test_abort_if_already_setup(
recorder_mock: Recorder, hass: HomeAssistant
) -> None:
"""Test we abort if the site_id is already setup."""
MockConfigEntry(
domain="solaredge",
domain=DOMAIN,
data={CONF_NAME: DEFAULT_NAME, CONF_SITE_ID: SITE_ID, CONF_API_KEY: API_KEY},
).add_to_hass(hass)
# user: Should fail, same SITE_ID
# Should fail, same SITE_ID
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data={CONF_NAME: "test", CONF_SITE_ID: SITE_ID, CONF_API_KEY: "test"},
data={
CONF_NAME: "test",
CONF_SITE_ID: SITE_ID,
CONF_SECTION_API_AUTH: {CONF_API_KEY: "test"},
},
)
assert result.get("type") is FlowResultType.FORM
assert result.get("errors") == {CONF_SITE_ID: "already_configured"}
async def test_ignored_entry_does_not_cause_error(
hass: HomeAssistant, test_api: str
recorder_mock: Recorder, hass: HomeAssistant
) -> None:
"""Test an ignored entry does not cause and error and we can still create an new entry."""
MockConfigEntry(
@@ -80,11 +175,15 @@ async def test_ignored_entry_does_not_cause_error(
source=SOURCE_IGNORE,
).add_to_hass(hass)
# user: Should fail, same SITE_ID
# Should not fail, same SITE_ID
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data={CONF_NAME: "test", CONF_SITE_ID: SITE_ID, CONF_API_KEY: "test"},
data={
CONF_NAME: "test",
CONF_SITE_ID: SITE_ID,
CONF_SECTION_API_AUTH: {CONF_API_KEY: "test"},
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "test"
@@ -95,46 +194,116 @@ async def test_ignored_entry_does_not_cause_error(
assert data[CONF_API_KEY] == "test"
async def test_asserts(hass: HomeAssistant, test_api: Mock) -> None:
"""Test the _site_in_configuration_exists method."""
async def test_no_auth_provided(recorder_mock: Recorder, hass: HomeAssistant) -> None:
"""Test error when no authentication method is provided."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_NAME: NAME, CONF_SITE_ID: SITE_ID},
)
assert result.get("type") is FlowResultType.FORM
assert result.get("errors") == {"base": "auth_missing"}
# test with inactive site
test_api.get_details.return_value = {"details": {"status": "NOK"}}
@pytest.mark.parametrize(
("get_details_setup", "expected_error"),
[
(AsyncMock(return_value={"details": {"status": "NOK"}}), "site_not_active"),
(AsyncMock(return_value={}), "invalid_api_key"),
(AsyncMock(side_effect=TimeoutError()), "cannot_connect"),
(AsyncMock(side_effect=ClientError()), "cannot_connect"),
],
)
async def test_api_key_errors(
recorder_mock: Recorder,
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
solaredge_api: Mock,
get_details_setup: AsyncMock,
expected_error: str,
) -> None:
"""Test API key validation errors."""
solaredge_api.get_details = get_details_setup
user_input = {
CONF_NAME: NAME,
CONF_SITE_ID: SITE_ID,
CONF_SECTION_API_AUTH: {CONF_API_KEY: API_KEY},
}
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data={CONF_NAME: NAME, CONF_API_KEY: API_KEY, CONF_SITE_ID: SITE_ID},
data=user_input,
)
assert result.get("type") is FlowResultType.FORM
assert result.get("errors") == {CONF_SITE_ID: "site_not_active"}
# test with api_failure
test_api.get_details.return_value = {}
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data={CONF_NAME: NAME, CONF_API_KEY: API_KEY, CONF_SITE_ID: SITE_ID},
)
assert result.get("type") is FlowResultType.FORM
assert result.get("errors") == {CONF_SITE_ID: "invalid_api_key"}
assert result.get("errors") == {CONF_SITE_ID: expected_error}
# test with ConnectionTimeout
test_api.get_details = AsyncMock(side_effect=TimeoutError())
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data={CONF_NAME: NAME, CONF_API_KEY: API_KEY, CONF_SITE_ID: SITE_ID},
# Make sure the config flow is able to recover from above error
solaredge_api.get_details = AsyncMock(
return_value={"details": {"status": "active"}}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input
)
assert result.get("type") is FlowResultType.FORM
assert result.get("errors") == {CONF_SITE_ID: "could_not_connect"}
# test with HTTPError
test_api.get_details = AsyncMock(side_effect=ClientError())
assert result.get("type") is FlowResultType.CREATE_ENTRY
assert result.get("data") == {CONF_SITE_ID: SITE_ID, CONF_API_KEY: API_KEY}
assert len(mock_setup_entry.mock_calls) == 1
@pytest.mark.parametrize(
("api_exception", "expected_error"),
[
(ClientResponseError(None, None, status=401), "invalid_auth"),
(ClientResponseError(None, None, status=403), "invalid_auth"),
(ClientResponseError(None, None, status=400), "cannot_connect"),
(ClientResponseError(None, None, status=500), "cannot_connect"),
(TimeoutError(), "cannot_connect"),
(ClientError(), "cannot_connect"),
],
)
async def test_web_login_errors(
recorder_mock: Recorder,
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
solaredge_web_api: AsyncMock,
api_exception: Exception,
expected_error: str,
) -> None:
"""Test web login validation errors."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data={CONF_NAME: NAME, CONF_API_KEY: API_KEY, CONF_SITE_ID: SITE_ID},
DOMAIN, context={"source": SOURCE_USER}
)
solaredge_web_api.async_get_equipment.side_effect = api_exception
user_input = {
CONF_NAME: NAME,
CONF_SITE_ID: SITE_ID,
CONF_SECTION_WEB_AUTH: {
CONF_USERNAME: USERNAME,
CONF_PASSWORD: PASSWORD,
},
}
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input
)
assert result.get("type") is FlowResultType.FORM
assert result.get("errors") == {CONF_SITE_ID: "could_not_connect"}
assert result.get("errors") == {"base": expected_error}
# Make sure the config flow is able to recover from above error
solaredge_web_api.async_get_equipment.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input
)
assert result.get("type") is FlowResultType.CREATE_ENTRY
assert result.get("data") == {
CONF_SITE_ID: SITE_ID,
CONF_USERNAME: USERNAME,
CONF_PASSWORD: PASSWORD,
}
assert len(mock_setup_entry.mock_calls) == 1

317
tests/components/solaredge/test_coordinator.py
View File

@@ -1,23 +1,37 @@
"""Tests for the SolarEdge coordinator services."""
import asyncio
from datetime import datetime, timedelta
from unittest.mock import AsyncMock, patch
from freezegun.api import FrozenDateTimeFactory
import pytest
from solaredge_web import EnergyData
from homeassistant.components.recorder import Recorder
from homeassistant.components.recorder.statistics import statistics_during_period
from homeassistant.components.solaredge.const import (
CONF_SITE_ID,
DATA_MODULES_COORDINATOR,
DEFAULT_NAME,
DOMAIN,
OVERVIEW_UPDATE_DELAY,
)
from homeassistant.const import CONF_API_KEY, CONF_NAME, STATE_UNKNOWN
from homeassistant.components.solaredge.coordinator import SolarEdgeModulesCoordinator
from homeassistant.const import (
CONF_API_KEY,
CONF_NAME,
CONF_PASSWORD,
CONF_USERNAME,
STATE_UNKNOWN,
)
from homeassistant.core import HomeAssistant
from homeassistant.util import dt as dt_util
from .conftest import API_KEY, PASSWORD, SITE_ID, USERNAME
from tests.common import MockConfigEntry, async_fire_time_changed
SITE_ID = "1a2b3c4d5e6f7g8h"
API_KEY = "a1b2c3d4e5f6g7h8"
from tests.components.recorder.common import async_wait_recording_done
@pytest.fixture(autouse=True)
@@ -27,7 +41,10 @@ def enable_all_entities(entity_registry_enabled_by_default: None) -> None:
@patch("homeassistant.components.solaredge.SolarEdge")
async def test_solaredgeoverviewdataservice_energy_values_validity(
mock_solaredge, hass: HomeAssistant, freezer: FrozenDateTimeFactory
mock_solaredge,
recorder_mock: Recorder,
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test overview energy data validity."""
mock_config_entry = MockConfigEntry(
@@ -110,3 +127,293 @@ async def test_solaredgeoverviewdataservice_energy_values_validity(
state = hass.states.get("sensor.solaredge_lifetime_energy")
assert state
assert state.state == str(mock_overview_data["overview"]["lifeTimeData"]["energy"])
async def _trigger_and_wait_for_refresh(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
coordinator: SolarEdgeModulesCoordinator,
) -> None:
"""Trigger a coordinator refresh and wait for it to complete."""
# The coordinator refresh runs in the background.
# To reliably assert the result, we need to wait for the refresh to complete.
# We patch the coordinator's update method to signal completion via an asyncio.Event.
refresh_done = asyncio.Event()
original_update = coordinator._async_update_data
async def wrapped_update_data() -> None:
"""Wrap original update and set event."""
await original_update()
refresh_done.set()
with patch.object(
coordinator,
"_async_update_data",
side_effect=wrapped_update_data,
autospec=True,
):
freezer.tick(timedelta(hours=12))
async_fire_time_changed(hass)
await asyncio.wait_for(refresh_done.wait(), timeout=5)
@pytest.fixture
def mock_solar_edge_web() -> AsyncMock:
"""Mock SolarEdgeWeb."""
with patch(
"homeassistant.components.solaredge.coordinator.SolarEdgeWeb", autospec=True
) as mock_api:
api = mock_api.return_value
api.async_get_equipment.return_value = {
1001: {"displayName": "1.1"},
1002: {"displayName": "1.2"},
}
api.async_get_energy_data.return_value = [
EnergyData(
start_time=dt_util.as_utc(datetime(2025, 1, 1, 10, 0)),
values={1001: 10.0, 1002: 20.0},
),
EnergyData(
start_time=dt_util.as_utc(datetime(2025, 1, 1, 10, 15)),
values={1001: 11.0, 1002: 21.0},
),
EnergyData(
start_time=dt_util.as_utc(datetime(2025, 1, 1, 10, 30)),
values={1001: 12.0, 1002: 22.0},
),
EnergyData(
start_time=dt_util.as_utc(datetime(2025, 1, 1, 10, 45)),
values={1001: 13.0, 1002: 23.0},
),
EnergyData(
start_time=dt_util.as_utc(datetime(2025, 1, 1, 11, 0)),
values={1001: 14.0, 1002: 24.0},
),
EnergyData(
start_time=dt_util.as_utc(datetime(2025, 1, 1, 11, 15)),
values={1001: 15.0, 1002: 25.0},
),
EnergyData(
start_time=dt_util.as_utc(datetime(2025, 1, 1, 11, 30)),
values={1001: 16.0, 1002: 26.0},
),
EnergyData(
start_time=dt_util.as_utc(datetime(2025, 1, 1, 11, 45)),
values={1001: 17.0, 1002: 27.0},
),
]
yield api
async def test_modules_coordinator_first_run(
recorder_mock: Recorder,
hass: HomeAssistant,
mock_solar_edge_web: AsyncMock,
) -> None:
"""Test the modules coordinator on its first run with no existing statistics."""
entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_SITE_ID: SITE_ID, CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
await async_wait_recording_done(hass)
stats = await hass.async_add_executor_job(
statistics_during_period,
hass,
dt_util.as_utc(datetime(1970, 1, 1, 0, 0)),
None,
{f"{DOMAIN}:{SITE_ID}_1001", f"{DOMAIN}:{SITE_ID}_1002"},
"hour",
None,
{"state", "sum"},
)
assert stats == {
f"{DOMAIN}:{SITE_ID}_1001": [
{"start": 1735783200.0, "end": 1735786800.0, "state": 11.5, "sum": 11.5},
{"start": 1735786800.0, "end": 1735790400.0, "state": 15.5, "sum": 27.0},
],
f"{DOMAIN}:{SITE_ID}_1002": [
{"start": 1735783200.0, "end": 1735786800.0, "state": 21.5, "sum": 21.5},
{"start": 1735786800.0, "end": 1735790400.0, "state": 25.5, "sum": 47.0},
],
}
async def test_modules_coordinator_subsequent_run(
recorder_mock: Recorder,
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
mock_solar_edge_web: AsyncMock,
) -> None:
"""Test the coordinator correctly updates statistics on subsequent runs."""
mock_solar_edge_web.async_get_equipment.return_value = {
1001: {"displayName": "1.1"},
}
entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_SITE_ID: SITE_ID, CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
await async_wait_recording_done(hass)
mock_solar_edge_web.async_get_energy_data.return_value = [
# Updated values, different from the first run
EnergyData(
start_time=dt_util.as_utc(datetime(2025, 1, 1, 11, 0)),
values={1001: 24.0},
),
EnergyData(
start_time=dt_util.as_utc(datetime(2025, 1, 1, 11, 15)),
values={1001: 25.0},
),
EnergyData(
start_time=dt_util.as_utc(datetime(2025, 1, 1, 11, 30)),
values={1001: 26.0},
),
EnergyData(
start_time=dt_util.as_utc(datetime(2025, 1, 1, 11, 45)),
values={1001: 27.0},
),
# New values for the next hour
EnergyData(
start_time=dt_util.as_utc(datetime(2025, 1, 1, 12, 0)),
values={1001: 28.0},
),
EnergyData(
start_time=dt_util.as_utc(datetime(2025, 1, 1, 12, 15)),
values={1001: 29.0},
),
EnergyData(
start_time=dt_util.as_utc(datetime(2025, 1, 1, 12, 30)),
values={1001: 30.0},
),
EnergyData(
start_time=dt_util.as_utc(datetime(2025, 1, 1, 12, 45)),
values={1001: 31.0},
),
]
coordinator: SolarEdgeModulesCoordinator = entry.runtime_data[
DATA_MODULES_COORDINATOR
]
await _trigger_and_wait_for_refresh(hass, freezer, coordinator)
await async_wait_recording_done(hass)
stats = await hass.async_add_executor_job(
statistics_during_period,
hass,
dt_util.as_utc(datetime(1970, 1, 1, 0, 0)),
None,
{f"{DOMAIN}:{SITE_ID}_1001"},
"hour",
None,
{"state", "sum"},
)
assert stats == {
f"{DOMAIN}:{SITE_ID}_1001": [
{"start": 1735783200.0, "end": 1735786800.0, "state": 11.5, "sum": 11.5},
{"start": 1735786800.0, "end": 1735790400.0, "state": 25.5, "sum": 37.0},
{"start": 1735790400.0, "end": 1735794000.0, "state": 29.5, "sum": 66.5},
]
}
async def test_modules_coordinator_subsequent_run_with_gap(
recorder_mock: Recorder,
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
mock_solar_edge_web: AsyncMock,
) -> None:
"""Test the coordinator correctly updates statistics on subsequent runs with a gap in data."""
mock_solar_edge_web.async_get_equipment.return_value = {
1001: {"displayName": "1.1"},
}
entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_SITE_ID: SITE_ID, CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
await async_wait_recording_done(hass)
mock_solar_edge_web.async_get_energy_data.return_value = [
# New values a month later, simulating a gap in data
EnergyData(
start_time=dt_util.as_utc(datetime(2025, 2, 1, 11, 0)),
values={1001: 24.0},
),
EnergyData(
start_time=dt_util.as_utc(datetime(2025, 2, 1, 11, 15)),
values={1001: 25.0},
),
EnergyData(
start_time=dt_util.as_utc(datetime(2025, 2, 1, 11, 30)),
values={1001: 26.0},
),
EnergyData(
start_time=dt_util.as_utc(datetime(2025, 2, 1, 11, 45)),
values={1001: 27.0},
),
]
coordinator: SolarEdgeModulesCoordinator = entry.runtime_data[
DATA_MODULES_COORDINATOR
]
await _trigger_and_wait_for_refresh(hass, freezer, coordinator)
await async_wait_recording_done(hass)
stats = await hass.async_add_executor_job(
statistics_during_period,
hass,
dt_util.as_utc(datetime(1970, 1, 1, 0, 0)),
None,
{f"{DOMAIN}:{SITE_ID}_1001"},
"hour",
None,
{"state", "sum"},
)
assert stats == {
f"{DOMAIN}:{SITE_ID}_1001": [
{"start": 1735783200.0, "end": 1735786800.0, "state": 11.5, "sum": 11.5},
{"start": 1735786800.0, "end": 1735790400.0, "state": 15.5, "sum": 27.0},
{"start": 1738465200.0, "end": 1738468800.0, "state": 25.5, "sum": 52.5},
]
}
async def test_modules_coordinator_no_energy_data(
recorder_mock: Recorder,
hass: HomeAssistant,
mock_solar_edge_web: AsyncMock,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test the coordinator handles an empty energy data response from the API."""
mock_solar_edge_web.async_get_energy_data.return_value = []
entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_SITE_ID: SITE_ID, CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert "No data received from SolarEdge API" in caplog.text
await async_wait_recording_done(hass)
stats = await hass.async_add_executor_job(
statistics_during_period,
hass,
dt_util.as_utc(datetime(1970, 1, 1, 0, 0)),
None,
{f"{DOMAIN}:{SITE_ID}_1001", f"{DOMAIN}:{SITE_ID}_1002"},
"hour",
None,
{"state", "sum"},
)
assert not stats

131
tests/components/solaredge/test_init.py Normal file
View File

@@ -0,0 +1,131 @@
"""Tests for the SolarEdge integration."""
from unittest.mock import AsyncMock, Mock
from aiohttp import ClientError
from homeassistant.components.recorder import Recorder
from homeassistant.components.solaredge.const import CONF_SITE_ID, DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import CONF_API_KEY, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from .conftest import API_KEY, PASSWORD, SITE_ID, USERNAME
from tests.common import MockConfigEntry
async def test_setup_unload_api_key(
recorder_mock: Recorder, hass: HomeAssistant, solaredge_api: Mock
) -> None:
"""Test successful setup and unload of a config entry with API key."""
entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_SITE_ID: SITE_ID, CONF_API_KEY: API_KEY},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.LOADED
assert solaredge_api.get_details.await_count == 2
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.NOT_LOADED
async def test_setup_unload_web_login(
recorder_mock: Recorder, hass: HomeAssistant, solaredge_web_api: AsyncMock
) -> None:
"""Test successful setup and unload of a config entry with web login."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_SITE_ID: SITE_ID,
CONF_USERNAME: USERNAME,
CONF_PASSWORD: PASSWORD,
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.LOADED
solaredge_web_api.async_get_equipment.assert_awaited_once()
solaredge_web_api.async_get_energy_data.assert_awaited_once()
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.NOT_LOADED
async def test_setup_unload_both(
recorder_mock: Recorder,
hass: HomeAssistant,
solaredge_api: Mock,
solaredge_web_api: AsyncMock,
) -> None:
"""Test successful setup and unload of a config entry with both auth methods."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_SITE_ID: SITE_ID,
CONF_API_KEY: API_KEY,
CONF_USERNAME: USERNAME,
CONF_PASSWORD: PASSWORD,
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.LOADED
assert solaredge_api.get_details.await_count == 2
solaredge_web_api.async_get_equipment.assert_awaited_once()
solaredge_web_api.async_get_energy_data.assert_awaited_once()
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.NOT_LOADED
async def test_api_key_config_not_ready(
recorder_mock: Recorder, hass: HomeAssistant, solaredge_api: Mock
) -> None:
"""Test for setup failure with API key."""
solaredge_api.get_details.side_effect = ClientError()
entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_SITE_ID: SITE_ID, CONF_API_KEY: API_KEY},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.SETUP_RETRY
async def test_web_login_config_not_ready(
recorder_mock: Recorder, hass: HomeAssistant, solaredge_web_api: AsyncMock
) -> None:
"""Test for setup failure with web login."""
solaredge_web_api.async_get_equipment.side_effect = ClientError()
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_SITE_ID: SITE_ID,
CONF_USERNAME: USERNAME,
CONF_PASSWORD: PASSWORD,
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.SETUP_RETRY