jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-21 20:27:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

Wallbox fix too many requests by API (#147197)

Browse Source
This commit is contained in:
Hessel 2025-06-23 14:10:50 +02:00 committed by GitHub
parent b2520394f4
commit 756b858840
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
13 changed files with 692 additions and 454 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -3,7 +3,7 @@
from enum import StrEnum
DOMAIN = "wallbox"
UPDATE_INTERVAL = 30
UPDATE_INTERVAL = 60
BIDIRECTIONAL_MODEL_PREFIXES = ["QS"]

69
homeassistant/components/wallbox/coordinator.py
View File

@ -90,7 +90,9 @@ def _require_authentication[_WallboxCoordinatorT: WallboxCoordinator, **_P](
except requests.exceptions.HTTPError as wallbox_connection_error:
if wallbox_connection_error.response.status_code == HTTPStatus.FORBIDDEN:
raise ConfigEntryAuthFailed from wallbox_connection_error
raise ConnectionError from wallbox_connection_error
raise HomeAssistantError(
translation_domain=DOMAIN, translation_key="api_failed"
) from wallbox_connection_error
return require_authentication
@ -137,6 +139,7 @@ class WallboxCoordinator(DataUpdateCoordinator[dict[str, Any]]):
@_require_authentication
def _get_data(self) -> dict[str, Any]:
"""Get new sensor data for Wallbox component."""
try:
data: dict[str, Any] = self._wallbox.getChargerStatus(self._station)
data[CHARGER_MAX_CHARGING_CURRENT_KEY] = data[CHARGER_DATA_KEY][
CHARGER_MAX_CHARGING_CURRENT_KEY
@ -186,7 +189,15 @@ class WallboxCoordinator(DataUpdateCoordinator[dict[str, Any]]):
elif eco_smart_mode == 1:
data[CHARGER_ECO_SMART_KEY] = EcoSmartMode.FULL_SOLAR
return data
return data # noqa: TRY300
except requests.exceptions.HTTPError as wallbox_connection_error:
if wallbox_connection_error.response.status_code == 429:
raise HomeAssistantError(
translation_domain=DOMAIN, translation_key="too_many_requests"
) from wallbox_connection_error
raise HomeAssistantError(
translation_domain=DOMAIN, translation_key="api_failed"
) from wallbox_connection_error
async def _async_update_data(self) -> dict[str, Any]:
"""Get new sensor data for Wallbox component."""
@ -200,7 +211,13 @@ class WallboxCoordinator(DataUpdateCoordinator[dict[str, Any]]):
except requests.exceptions.HTTPError as wallbox_connection_error:
if wallbox_connection_error.response.status_code == 403:
raise InvalidAuth from wallbox_connection_error
raise
if wallbox_connection_error.response.status_code == 429:
raise HomeAssistantError(
translation_domain=DOMAIN, translation_key="too_many_requests"
) from wallbox_connection_error
raise HomeAssistantError(
translation_domain=DOMAIN, translation_key="api_failed"
) from wallbox_connection_error
async def async_set_charging_current(self, charging_current: float) -> None:
"""Set maximum charging current for Wallbox."""
@ -217,7 +234,13 @@ class WallboxCoordinator(DataUpdateCoordinator[dict[str, Any]]):
except requests.exceptions.HTTPError as wallbox_connection_error:
if wallbox_connection_error.response.status_code == 403:
raise InvalidAuth from wallbox_connection_error
raise
if wallbox_connection_error.response.status_code == 429:
raise HomeAssistantError(
translation_domain=DOMAIN, translation_key="too_many_requests"
) from wallbox_connection_error
raise HomeAssistantError(
translation_domain=DOMAIN, translation_key="api_failed"
) from wallbox_connection_error
async def async_set_icp_current(self, icp_current: float) -> None:
"""Set maximum icp current for Wallbox."""
@ -227,8 +250,16 @@ class WallboxCoordinator(DataUpdateCoordinator[dict[str, Any]]):
@_require_authentication
def _set_energy_cost(self, energy_cost: float) -> None:
"""Set energy cost for Wallbox."""
try:
self._wallbox.setEnergyCost(self._station, energy_cost)
except requests.exceptions.HTTPError as wallbox_connection_error:
if wallbox_connection_error.response.status_code == 429:
raise HomeAssistantError(
translation_domain=DOMAIN, translation_key="too_many_requests"
) from wallbox_connection_error
raise HomeAssistantError(
translation_domain=DOMAIN, translation_key="api_failed"
) from wallbox_connection_error
async def async_set_energy_cost(self, energy_cost: float) -> None:
"""Set energy cost for Wallbox."""
@ -246,7 +277,13 @@ class WallboxCoordinator(DataUpdateCoordinator[dict[str, Any]]):
except requests.exceptions.HTTPError as wallbox_connection_error:
if wallbox_connection_error.response.status_code == 403:
raise InvalidAuth from wallbox_connection_error
raise
if wallbox_connection_error.response.status_code == 429:
raise HomeAssistantError(
translation_domain=DOMAIN, translation_key="too_many_requests"
) from wallbox_connection_error
raise HomeAssistantError(
translation_domain=DOMAIN, translation_key="api_failed"
) from wallbox_connection_error
async def async_set_lock_unlock(self, lock: bool) -> None:
"""Set wallbox to locked or unlocked."""
@ -256,11 +293,19 @@ class WallboxCoordinator(DataUpdateCoordinator[dict[str, Any]]):
@_require_authentication
def _pause_charger(self, pause: bool) -> None:
"""Set wallbox to pause or resume."""
try:
if pause:
self._wallbox.pauseChargingSession(self._station)
else:
self._wallbox.resumeChargingSession(self._station)
except requests.exceptions.HTTPError as wallbox_connection_error:
if wallbox_connection_error.response.status_code == 429:
raise HomeAssistantError(
translation_domain=DOMAIN, translation_key="too_many_requests"
) from wallbox_connection_error
raise HomeAssistantError(
translation_domain=DOMAIN, translation_key="api_failed"
) from wallbox_connection_error
async def async_pause_charger(self, pause: bool) -> None:
"""Set wallbox to pause or resume."""
@ -270,13 +315,21 @@ class WallboxCoordinator(DataUpdateCoordinator[dict[str, Any]]):
@_require_authentication
def _set_eco_smart(self, option: str) -> None:
"""Set wallbox solar charging mode."""
try:
if option == EcoSmartMode.ECO_MODE:
self._wallbox.enableEcoSmart(self._station, 0)
elif option == EcoSmartMode.FULL_SOLAR:
self._wallbox.enableEcoSmart(self._station, 1)
else:
self._wallbox.disableEcoSmart(self._station)
except requests.exceptions.HTTPError as wallbox_connection_error:
if wallbox_connection_error.response.status_code == 429:
raise HomeAssistantError(
translation_domain=DOMAIN, translation_key="too_many_requests"
) from wallbox_connection_error
raise HomeAssistantError(
translation_domain=DOMAIN, translation_key="api_failed"
) from wallbox_connection_error
async def async_set_eco_smart(self, option: str) -> None:
"""Set wallbox solar charging mode."""

4
homeassistant/components/wallbox/lock.py
View File

@ -7,7 +7,7 @@ from typing import Any
from homeassistant.components.lock import LockEntity, LockEntityDescription
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import PlatformNotReady
from homeassistant.exceptions import HomeAssistantError, PlatformNotReady
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import (
@ -41,7 +41,7 @@ async def async_setup_entry(
)
except InvalidAuth:
return
except ConnectionError as exc:
except HomeAssistantError as exc:
raise PlatformNotReady from exc
async_add_entities(

4
homeassistant/components/wallbox/number.py
View File

@ -12,7 +12,7 @@ from typing import cast
from homeassistant.components.number import NumberEntity, NumberEntityDescription
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import PlatformNotReady
from homeassistant.exceptions import HomeAssistantError, PlatformNotReady
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import (
@ -93,7 +93,7 @@ async def async_setup_entry(
)
except InvalidAuth:
return
except ConnectionError as exc:
except HomeAssistantError as exc:
raise PlatformNotReady from exc
async_add_entities(

6
homeassistant/components/wallbox/sensor.py
View File

@ -3,7 +3,6 @@
from __future__ import annotations
from dataclasses import dataclass
import logging
from typing import cast
from homeassistant.components.sensor import (
@ -49,11 +48,6 @@ from .const import (
from .coordinator import WallboxCoordinator
from .entity import WallboxEntity
CHARGER_STATION = "station"
UPDATE_INTERVAL = 30
_LOGGER = logging.getLogger(__name__)
@dataclass(frozen=True)
class WallboxSensorEntityDescription(SensorEntityDescription):

3
homeassistant/components/wallbox/strings.json
View File

@ -112,6 +112,9 @@
"exceptions": {
"api_failed": {
"message": "Error communicating with Wallbox API"
},
"too_many_requests": {
"message": "Error communicating with Wallbox API, too many requests"
}
}
}

21
tests/components/wallbox/__init__.py
View File

@ -162,6 +162,9 @@ test_response_no_power_boost = {
http_404_error = requests.exceptions.HTTPError()
http_404_error.response = requests.Response()
http_404_error.response.status_code = HTTPStatus.NOT_FOUND
http_429_error = requests.exceptions.HTTPError()
http_429_error.response = requests.Response()
http_429_error.response.status_code = HTTPStatus.TOO_MANY_REQUESTS
authorisation_response = {
"data": {
@ -192,6 +195,24 @@ authorisation_response_unauthorised = {
}
}
invalid_reauth_response = {
"jwt": "fakekeyhere",
"refresh_token": "refresh_fakekeyhere",
"user_id": 12345,
"ttl": 145656758,
"refresh_token_ttl": 145756758,
"error": False,
"status": 200,
}
http_403_error = requests.exceptions.HTTPError()
http_403_error.response = requests.Response()
http_403_error.response.status_code = HTTPStatus.FORBIDDEN
http_404_error = requests.exceptions.HTTPError()
http_404_error.response = requests.Response()
http_404_error.response.status_code = HTTPStatus.NOT_FOUND
async def setup_integration(hass: HomeAssistant, entry: MockConfigEntry) -> None:
"""Test wallbox sensor class setup."""

128
tests/components/wallbox/test_config_flow.py
View File

@ -1,9 +1,6 @@
"""Test the Wallbox config flow."""
from http import HTTPStatus
import json
import requests_mock
from unittest.mock import Mock, patch
from homeassistant import config_entries
from homeassistant.components.wallbox import config_flow
@ -24,14 +21,14 @@ from homeassistant.data_entry_flow import FlowResultType
from . import (
authorisation_response,
authorisation_response_unauthorised,
http_403_error,
http_404_error,
setup_integration,
)
from tests.common import MockConfigEntry
test_response = json.loads(
json.dumps(
{
test_response = {
CHARGER_CHARGING_POWER_KEY: 0,
CHARGER_MAX_AVAILABLE_POWER_KEY: "xx",
CHARGER_CHARGING_SPEED_KEY: 0,
@ -39,8 +36,6 @@ test_response = json.loads(
CHARGER_ADDED_ENERGY_KEY: "44.697",
CHARGER_DATA_KEY: {CHARGER_MAX_CHARGING_CURRENT_KEY: 24},
}
)
)
async def test_show_set_form(hass: HomeAssistant) -> None:
@ -59,17 +54,16 @@ async def test_form_cannot_authenticate(hass: HomeAssistant) -> None:
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with requests_mock.Mocker() as mock_request:
mock_request.get(
"https://user-api.wall-box.com/users/signin",
json=authorisation_response,
status_code=HTTPStatus.FORBIDDEN,
)
mock_request.get(
"https://api.wall-box.com/chargers/status/12345",
json=test_response,
status_code=HTTPStatus.FORBIDDEN,
)
with (
patch(
"homeassistant.components.wallbox.Wallbox.authenticate",
new=Mock(side_effect=http_403_error),
),
patch(
"homeassistant.components.wallbox.Wallbox.pauseChargingSession",
new=Mock(side_effect=http_403_error),
),
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
@ -89,17 +83,16 @@ async def test_form_cannot_connect(hass: HomeAssistant) -> None:
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with requests_mock.Mocker() as mock_request:
mock_request.get(
"https://user-api.wall-box.com/users/signin",
json=authorisation_response_unauthorised,
status_code=HTTPStatus.NOT_FOUND,
)
mock_request.get(
"https://api.wall-box.com/chargers/status/12345",
json=test_response,
status_code=HTTPStatus.NOT_FOUND,
)
with (
patch(
"homeassistant.components.wallbox.Wallbox.authenticate",
new=Mock(side_effect=http_404_error),
),
patch(
"homeassistant.components.wallbox.Wallbox.pauseChargingSession",
new=Mock(side_effect=http_404_error),
),
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
@ -119,17 +112,16 @@ async def test_form_validate_input(hass: HomeAssistant) -> None:
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with requests_mock.Mocker() as mock_request:
mock_request.get(
"https://user-api.wall-box.com/users/signin",
json=authorisation_response,
status_code=HTTPStatus.OK,
)
mock_request.get(
"https://api.wall-box.com/chargers/status/12345",
json=test_response,
status_code=HTTPStatus.OK,
)
with (
patch(
"homeassistant.components.wallbox.Wallbox.authenticate",
new=Mock(return_value=authorisation_response),
),
patch(
"homeassistant.components.wallbox.Wallbox.getChargerStatus",
new=Mock(return_value=test_response),
),
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
@ -148,18 +140,16 @@ async def test_form_reauth(hass: HomeAssistant, entry: MockConfigEntry) -> None:
await setup_integration(hass, entry)
assert entry.state is ConfigEntryState.LOADED
with requests_mock.Mocker() as mock_request:
mock_request.get(
"https://user-api.wall-box.com/users/signin",
json=authorisation_response,
status_code=200,
)
mock_request.get(
"https://api.wall-box.com/chargers/status/12345",
json=test_response,
status_code=200,
)
with (
patch(
"homeassistant.components.wallbox.Wallbox.authenticate",
new=Mock(return_value=authorisation_response_unauthorised),
),
patch(
"homeassistant.components.wallbox.Wallbox.getChargerStatus",
new=Mock(return_value=test_response),
),
):
result = await entry.start_reauth_flow(hass)
result2 = await hass.config_entries.flow.async_configure(
@ -183,26 +173,16 @@ async def test_form_reauth_invalid(hass: HomeAssistant, entry: MockConfigEntry)
await setup_integration(hass, entry)
assert entry.state is ConfigEntryState.LOADED
with requests_mock.Mocker() as mock_request:
mock_request.get(
"https://user-api.wall-box.com/users/signin",
json={
"jwt": "fakekeyhere",
"refresh_token": "refresh_fakekeyhere",
"user_id": 12345,
"ttl": 145656758,
"refresh_token_ttl": 145756758,
"error": False,
"status": 200,
},
status_code=200,
)
mock_request.get(
"https://api.wall-box.com/chargers/status/12345",
json=test_response,
status_code=200,
)
with (
patch(
"homeassistant.components.wallbox.Wallbox.authenticate",
new=Mock(return_value=authorisation_response_unauthorised),
),
patch(
"homeassistant.components.wallbox.Wallbox.getChargerStatus",
new=Mock(return_value=test_response),
),
):
result = await entry.start_reauth_flow(hass)
result2 = await hass.config_entries.flow.async_configure(

125
tests/components/wallbox/test_init.py
View File

@ -1,16 +1,15 @@
"""Test Wallbox Init Component."""
import requests_mock
from unittest.mock import Mock, patch
from homeassistant.components.wallbox.const import (
CHARGER_MAX_CHARGING_CURRENT_KEY,
DOMAIN,
)
from homeassistant.components.wallbox.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from . import (
authorisation_response,
http_403_error,
http_429_error,
setup_integration,
setup_integration_connection_error,
setup_integration_no_eco_mode,
@ -53,18 +52,16 @@ async def test_wallbox_refresh_failed_connection_error_auth(
await setup_integration(hass, entry)
assert entry.state is ConfigEntryState.LOADED
with requests_mock.Mocker() as mock_request:
mock_request.get(
"https://user-api.wall-box.com/users/signin",
json=authorisation_response,
status_code=404,
)
mock_request.get(
"https://api.wall-box.com/chargers/status/12345",
json=test_response,
status_code=200,
)
with (
patch(
"homeassistant.components.wallbox.Wallbox.authenticate",
new=Mock(side_effect=http_429_error),
),
patch(
"homeassistant.components.wallbox.Wallbox.pauseChargingSession",
new=Mock(return_value=test_response),
),
):
wallbox = hass.data[DOMAIN][entry.entry_id]
await wallbox.async_refresh()
@ -81,18 +78,68 @@ async def test_wallbox_refresh_failed_invalid_auth(
await setup_integration(hass, entry)
assert entry.state is ConfigEntryState.LOADED
with requests_mock.Mocker() as mock_request:
mock_request.get(
"https://user-api.wall-box.com/users/signin",
json=authorisation_response,
status_code=403,
)
mock_request.put(
"https://api.wall-box.com/v2/charger/12345",
json={CHARGER_MAX_CHARGING_CURRENT_KEY: 20},
status_code=403,
)
with (
patch(
"homeassistant.components.wallbox.Wallbox.authenticate",
new=Mock(side_effect=http_403_error),
),
patch(
"homeassistant.components.wallbox.Wallbox.pauseChargingSession",
new=Mock(side_effect=http_403_error),
),
):
wallbox = hass.data[DOMAIN][entry.entry_id]
await wallbox.async_refresh()
assert await hass.config_entries.async_unload(entry.entry_id)
assert entry.state is ConfigEntryState.NOT_LOADED
async def test_wallbox_refresh_failed_http_error(
hass: HomeAssistant, entry: MockConfigEntry
) -> None:
"""Test Wallbox setup with authentication error."""
await setup_integration(hass, entry)
assert entry.state is ConfigEntryState.LOADED
with (
patch(
"homeassistant.components.wallbox.Wallbox.authenticate",
new=Mock(return_value=authorisation_response),
),
patch(
"homeassistant.components.wallbox.Wallbox.getChargerStatus",
new=Mock(side_effect=http_403_error),
),
):
wallbox = hass.data[DOMAIN][entry.entry_id]
await wallbox.async_refresh()
assert await hass.config_entries.async_unload(entry.entry_id)
assert entry.state is ConfigEntryState.NOT_LOADED
async def test_wallbox_refresh_failed_too_many_requests(
hass: HomeAssistant, entry: MockConfigEntry
) -> None:
"""Test Wallbox setup with authentication error."""
await setup_integration(hass, entry)
assert entry.state is ConfigEntryState.LOADED
with (
patch(
"homeassistant.components.wallbox.Wallbox.authenticate",
new=Mock(return_value=authorisation_response),
),
patch(
"homeassistant.components.wallbox.Wallbox.getChargerStatus",
new=Mock(side_effect=http_429_error),
),
):
wallbox = hass.data[DOMAIN][entry.entry_id]
await wallbox.async_refresh()
@ -109,18 +156,16 @@ async def test_wallbox_refresh_failed_connection_error(
await setup_integration(hass, entry)
assert entry.state is ConfigEntryState.LOADED
with requests_mock.Mocker() as mock_request:
mock_request.get(
"https://user-api.wall-box.com/users/signin",
json=authorisation_response,
status_code=200,
)
mock_request.get(
"https://api.wall-box.com/chargers/status/12345",
json=test_response,
status_code=403,
)
with (
patch(
"homeassistant.components.wallbox.Wallbox.authenticate",
new=Mock(return_value=authorisation_response),
),
patch(
"homeassistant.components.wallbox.Wallbox.pauseChargingSession",
new=Mock(side_effect=http_403_error),
),
):
wallbox = hass.data[DOMAIN][entry.entry_id]
await wallbox.async_refresh()

96
tests/components/wallbox/test_lock.py
View File

@ -1,15 +1,18 @@
"""Test Wallbox Lock component."""
from unittest.mock import Mock, patch
import pytest
import requests_mock
from homeassistant.components.lock import SERVICE_LOCK, SERVICE_UNLOCK
from homeassistant.components.wallbox.const import CHARGER_LOCKED_UNLOCKED_KEY
from homeassistant.const import ATTR_ENTITY_ID
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from . import (
authorisation_response,
http_429_error,
setup_integration,
setup_integration_platform_not_ready,
setup_integration_read_only,
@ -28,18 +31,20 @@ async def test_wallbox_lock_class(hass: HomeAssistant, entry: MockConfigEntry) -
assert state
assert state.state == "unlocked"
with requests_mock.Mocker() as mock_request:
mock_request.get(
"https://user-api.wall-box.com/users/signin",
json=authorisation_response,
status_code=200,
)
mock_request.put(
"https://api.wall-box.com/v2/charger/12345",
json={CHARGER_LOCKED_UNLOCKED_KEY: False},
status_code=200,
)
with (
patch(
"homeassistant.components.wallbox.Wallbox.authenticate",
new=Mock(return_value=authorisation_response),
),
patch(
"homeassistant.components.wallbox.Wallbox.lockCharger",
new=Mock(return_value={CHARGER_LOCKED_UNLOCKED_KEY: False}),
),
patch(
"homeassistant.components.wallbox.Wallbox.unlockCharger",
new=Mock(return_value={CHARGER_LOCKED_UNLOCKED_KEY: False}),
),
):
await hass.services.async_call(
"lock",
SERVICE_LOCK,
@ -66,19 +71,17 @@ async def test_wallbox_lock_class_connection_error(
await setup_integration(hass, entry)
with requests_mock.Mocker() as mock_request:
mock_request.get(
"https://user-api.wall-box.com/users/signin",
json=authorisation_response,
status_code=200,
)
mock_request.put(
"https://api.wall-box.com/v2/charger/12345",
json={CHARGER_LOCKED_UNLOCKED_KEY: False},
status_code=404,
)
with pytest.raises(ConnectionError):
with (
patch(
"homeassistant.components.wallbox.Wallbox.authenticate",
new=Mock(return_value=authorisation_response),
),
patch(
"homeassistant.components.wallbox.Wallbox.lockCharger",
new=Mock(side_effect=ConnectionError),
),
pytest.raises(ConnectionError),
):
await hass.services.async_call(
"lock",
SERVICE_LOCK,
@ -87,7 +90,46 @@ async def test_wallbox_lock_class_connection_error(
},
blocking=True,
)
with pytest.raises(ConnectionError):
with (
patch(
"homeassistant.components.wallbox.Wallbox.authenticate",
new=Mock(return_value=authorisation_response),
),
patch(
"homeassistant.components.wallbox.Wallbox.lockCharger",
new=Mock(side_effect=ConnectionError),
),
patch(
"homeassistant.components.wallbox.Wallbox.unlockCharger",
new=Mock(side_effect=ConnectionError),
),
pytest.raises(ConnectionError),
):
await hass.services.async_call(
"lock",
SERVICE_UNLOCK,
{
ATTR_ENTITY_ID: MOCK_LOCK_ENTITY_ID,
},
blocking=True,
)
with (
patch(
"homeassistant.components.wallbox.Wallbox.authenticate",
new=Mock(return_value=authorisation_response),
),
patch(
"homeassistant.components.wallbox.Wallbox.lockCharger",
new=Mock(side_effect=http_429_error),
),
patch(
"homeassistant.components.wallbox.Wallbox.unlockCharger",
new=Mock(side_effect=http_429_error),
),
pytest.raises(HomeAssistantError),
):
await hass.services.async_call(
"lock",
SERVICE_UNLOCK,

290
tests/components/wallbox/test_number.py
View File

@ -1,22 +1,26 @@
"""Test Wallbox Switch component."""
from unittest.mock import Mock, patch
import pytest
import requests_mock
from homeassistant.components.input_number import ATTR_VALUE, SERVICE_SET_VALUE
from homeassistant.components.number import DOMAIN as NUMBER_DOMAIN
from homeassistant.components.wallbox import InvalidAuth
from homeassistant.components.wallbox.const import (
CHARGER_ENERGY_PRICE_KEY,
CHARGER_MAX_CHARGING_CURRENT_KEY,
CHARGER_MAX_ICP_CURRENT_KEY,
)
from homeassistant.components.wallbox.coordinator import InvalidAuth
from homeassistant.const import ATTR_ENTITY_ID
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.exceptions import HomeAssistantError
from . import (
authorisation_response,
http_403_error,
http_404_error,
http_429_error,
setup_integration,
setup_integration_bidir,
setup_integration_platform_not_ready,
@ -29,6 +33,14 @@ from .const import (
from tests.common import MockConfigEntry
mock_wallbox = Mock()
mock_wallbox.authenticate = Mock(return_value=authorisation_response)
mock_wallbox.setEnergyCost = Mock(return_value={CHARGER_ENERGY_PRICE_KEY: 1.1})
mock_wallbox.setMaxChargingCurrent = Mock(
return_value={CHARGER_MAX_CHARGING_CURRENT_KEY: 20}
)
mock_wallbox.setIcpMaxCurrent = Mock(return_value={CHARGER_MAX_ICP_CURRENT_KEY: 10})
async def test_wallbox_number_class(
hass: HomeAssistant, entry: MockConfigEntry
@ -37,17 +49,16 @@ async def test_wallbox_number_class(
await setup_integration(hass, entry)
with requests_mock.Mocker() as mock_request:
mock_request.get(
"https://user-api.wall-box.com/users/signin",
json=authorisation_response,
status_code=200,
)
mock_request.put(
"https://api.wall-box.com/v2/charger/12345",
json={CHARGER_MAX_CHARGING_CURRENT_KEY: 20},
status_code=200,
)
with (
patch(
"homeassistant.components.wallbox.Wallbox.authenticate",
new=Mock(return_value=authorisation_response),
),
patch(
"homeassistant.components.wallbox.Wallbox.setMaxChargingCurrent",
new=Mock(return_value={CHARGER_MAX_CHARGING_CURRENT_KEY: 20}),
),
):
state = hass.states.get(MOCK_NUMBER_ENTITY_ID)
assert state.attributes["min"] == 6
assert state.attributes["max"] == 25
@ -82,19 +93,16 @@ async def test_wallbox_number_energy_class(
await setup_integration(hass, entry)
with requests_mock.Mocker() as mock_request:
mock_request.get(
"https://user-api.wall-box.com/users/signin",
json=authorisation_response,
status_code=200,
)
mock_request.post(
"https://api.wall-box.com/chargers/config/12345",
json={CHARGER_ENERGY_PRICE_KEY: 1.1},
status_code=200,
)
with (
patch(
"homeassistant.components.wallbox.Wallbox.authenticate",
new=Mock(return_value=authorisation_response),
),
patch(
"homeassistant.components.wallbox.Wallbox.setEnergyCost",
new=Mock(return_value={CHARGER_ENERGY_PRICE_KEY: 1.1}),
),
):
await hass.services.async_call(
"number",
SERVICE_SET_VALUE,
@ -113,19 +121,17 @@ async def test_wallbox_number_class_connection_error(
await setup_integration(hass, entry)
with requests_mock.Mocker() as mock_request:
mock_request.get(
"https://user-api.wall-box.com/users/signin",
json=authorisation_response,
status_code=200,
)
mock_request.put(
"https://api.wall-box.com/v2/charger/12345",
json={CHARGER_MAX_CHARGING_CURRENT_KEY: 20},
status_code=404,
)
with pytest.raises(ConnectionError):
with (
patch(
"homeassistant.components.wallbox.Wallbox.authenticate",
new=Mock(return_value=authorisation_response),
),
patch(
"homeassistant.components.wallbox.Wallbox.setMaxChargingCurrent",
new=Mock(side_effect=http_404_error),
),
pytest.raises(HomeAssistantError),
):
await hass.services.async_call(
"number",
SERVICE_SET_VALUE,
@ -137,26 +143,82 @@ async def test_wallbox_number_class_connection_error(
)
async def test_wallbox_number_class_energy_price_connection_error(
async def test_wallbox_number_class_too_many_requests(
hass: HomeAssistant, entry: MockConfigEntry
) -> None:
"""Test wallbox sensor class."""
await setup_integration(hass, entry)
with requests_mock.Mocker() as mock_request:
mock_request.get(
"https://user-api.wall-box.com/users/signin",
json=authorisation_response,
status_code=200,
)
mock_request.post(
"https://api.wall-box.com/chargers/config/12345",
json={CHARGER_ENERGY_PRICE_KEY: 1.1},
status_code=404,
with (
patch(
"homeassistant.components.wallbox.Wallbox.authenticate",
new=Mock(return_value=authorisation_response),
),
patch(
"homeassistant.components.wallbox.Wallbox.setMaxChargingCurrent",
new=Mock(side_effect=http_429_error),
),
pytest.raises(HomeAssistantError),
):
await hass.services.async_call(
"number",
SERVICE_SET_VALUE,
{
ATTR_ENTITY_ID: MOCK_NUMBER_ENTITY_ID,
ATTR_VALUE: 20,
},
blocking=True,
)
with pytest.raises(ConnectionError):
async def test_wallbox_number_class_energy_price_update_failed(
hass: HomeAssistant, entry: MockConfigEntry
) -> None:
"""Test wallbox sensor class."""
await setup_integration(hass, entry)
with (
patch(
"homeassistant.components.wallbox.Wallbox.authenticate",
new=Mock(return_value=authorisation_response),
),
patch(
"homeassistant.components.wallbox.Wallbox.setEnergyCost",
new=Mock(side_effect=http_429_error),
),
pytest.raises(HomeAssistantError),
):
await hass.services.async_call(
"number",
SERVICE_SET_VALUE,
{
ATTR_ENTITY_ID: MOCK_NUMBER_ENTITY_ENERGY_PRICE_ID,
ATTR_VALUE: 1.1,
},
blocking=True,
)
async def test_wallbox_number_class_energy_price_update_connection_error(
hass: HomeAssistant, entry: MockConfigEntry
) -> None:
"""Test wallbox sensor class."""
await setup_integration(hass, entry)
with (
patch(
"homeassistant.components.wallbox.Wallbox.authenticate",
new=Mock(return_value=authorisation_response),
),
patch(
"homeassistant.components.wallbox.Wallbox.setEnergyCost",
new=Mock(side_effect=http_404_error),
),
pytest.raises(HomeAssistantError),
):
await hass.services.async_call(
"number",
SERVICE_SET_VALUE,
@ -175,19 +237,17 @@ async def test_wallbox_number_class_energy_price_auth_error(
await setup_integration(hass, entry)
with requests_mock.Mocker() as mock_request:
mock_request.get(
"https://user-api.wall-box.com/users/signin",
json=authorisation_response,
status_code=200,
)
mock_request.post(
"https://api.wall-box.com/chargers/config/12345",
json={CHARGER_ENERGY_PRICE_KEY: 1.1},
status_code=403,
)
with pytest.raises(ConfigEntryAuthFailed):
with (
patch(
"homeassistant.components.wallbox.Wallbox.authenticate",
new=Mock(return_value=authorisation_response),
),
patch(
"homeassistant.components.wallbox.Wallbox.setEnergyCost",
new=Mock(side_effect=http_429_error),
),
pytest.raises(HomeAssistantError),
):
await hass.services.async_call(
"number",
SERVICE_SET_VALUE,
@ -218,19 +278,16 @@ async def test_wallbox_number_class_icp_energy(
await setup_integration(hass, entry)
with requests_mock.Mocker() as mock_request:
mock_request.get(
"https://user-api.wall-box.com/users/signin",
json=authorisation_response,
status_code=200,
)
mock_request.post(
"https://api.wall-box.com/chargers/config/12345",
json={CHARGER_MAX_ICP_CURRENT_KEY: 10},
status_code=200,
)
with (
patch(
"homeassistant.components.wallbox.Wallbox.authenticate",
new=Mock(return_value=authorisation_response),
),
patch(
"homeassistant.components.wallbox.Wallbox.setIcpMaxCurrent",
new=Mock(return_value={CHARGER_MAX_ICP_CURRENT_KEY: 10}),
),
):
await hass.services.async_call(
NUMBER_DOMAIN,
SERVICE_SET_VALUE,
@ -249,19 +306,17 @@ async def test_wallbox_number_class_icp_energy_auth_error(
await setup_integration(hass, entry)
with requests_mock.Mocker() as mock_request:
mock_request.get(
"https://user-api.wall-box.com/users/signin",
json=authorisation_response,
status_code=200,
)
mock_request.post(
"https://api.wall-box.com/chargers/config/12345",
json={CHARGER_MAX_ICP_CURRENT_KEY: 10},
status_code=403,
)
with pytest.raises(InvalidAuth):
with (
patch(
"homeassistant.components.wallbox.Wallbox.authenticate",
new=Mock(return_value=authorisation_response),
),
patch(
"homeassistant.components.wallbox.Wallbox.setIcpMaxCurrent",
new=Mock(side_effect=http_403_error),
),
pytest.raises(InvalidAuth),
):
await hass.services.async_call(
NUMBER_DOMAIN,
SERVICE_SET_VALUE,
@ -280,19 +335,46 @@ async def test_wallbox_number_class_icp_energy_connection_error(
await setup_integration(hass, entry)
with requests_mock.Mocker() as mock_request:
mock_request.get(
"https://user-api.wall-box.com/users/signin",
json=authorisation_response,
status_code=200,
)
mock_request.post(
"https://api.wall-box.com/chargers/config/12345",
json={CHARGER_MAX_ICP_CURRENT_KEY: 10},
status_code=404,
)
with pytest.raises(ConnectionError):
with (
patch(
"homeassistant.components.wallbox.Wallbox.authenticate",
new=Mock(return_value=authorisation_response),
),
patch(
"homeassistant.components.wallbox.Wallbox.setIcpMaxCurrent",
new=Mock(side_effect=http_404_error),
),
pytest.raises(HomeAssistantError),
):
await hass.services.async_call(
NUMBER_DOMAIN,
SERVICE_SET_VALUE,
{
ATTR_ENTITY_ID: MOCK_NUMBER_ENTITY_ICP_CURRENT_ID,
ATTR_VALUE: 10,
},
blocking=True,
)
async def test_wallbox_number_class_icp_energy_too_many_request(
hass: HomeAssistant, entry: MockConfigEntry
) -> None:
"""Test wallbox sensor class."""
await setup_integration(hass, entry)
with (
patch(
"homeassistant.components.wallbox.Wallbox.authenticate",
new=Mock(return_value=authorisation_response),
),
patch(
"homeassistant.components.wallbox.Wallbox.setIcpMaxCurrent",
new=Mock(side_effect=http_429_error),
),
pytest.raises(HomeAssistantError),
):
await hass.services.async_call(
NUMBER_DOMAIN,
SERVICE_SET_VALUE,

37
tests/components/wallbox/test_select.py
View File

@ -16,6 +16,7 @@ from homeassistant.core import HomeAssistant, HomeAssistantError
from . import (
authorisation_response,
http_404_error,
http_429_error,
setup_integration_select,
test_response,
test_response_eco_mode,
@ -109,7 +110,41 @@ async def test_wallbox_select_class_error(
"homeassistant.components.wallbox.Wallbox.enableEcoSmart",
new=Mock(side_effect=error),
),
pytest.raises(HomeAssistantError, match="Error communicating with Wallbox API"),
pytest.raises(HomeAssistantError),
):
await hass.services.async_call(
SELECT_DOMAIN,
SERVICE_SELECT_OPTION,
{
ATTR_ENTITY_ID: MOCK_SELECT_ENTITY_ID,
ATTR_OPTION: mode,
},
blocking=True,
)
@pytest.mark.parametrize(("mode", "response"), TEST_OPTIONS)
async def test_wallbox_select_too_many_requests_error(
hass: HomeAssistant,
entry: MockConfigEntry,
mode,
response,
mock_authenticate,
) -> None:
"""Test wallbox select class connection error."""
await setup_integration_select(hass, entry, response)
with (
patch(
"homeassistant.components.wallbox.Wallbox.disableEcoSmart",
new=Mock(side_effect=http_429_error),
),
patch(
"homeassistant.components.wallbox.Wallbox.enableEcoSmart",
new=Mock(side_effect=http_429_error),
),
pytest.raises(HomeAssistantError),
):
await hass.services.async_call(
SELECT_DOMAIN,

103
tests/components/wallbox/test_switch.py
View File

@ -1,15 +1,16 @@
"""Test Wallbox Lock component."""
from unittest.mock import Mock, patch
import pytest
import requests_mock
from homeassistant.components.switch import SERVICE_TURN_OFF, SERVICE_TURN_ON
from homeassistant.components.wallbox.const import CHARGER_STATUS_ID_KEY
from homeassistant.const import ATTR_ENTITY_ID
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.exceptions import HomeAssistantError
from . import authorisation_response, setup_integration
from . import authorisation_response, http_404_error, http_429_error, setup_integration
from .const import MOCK_SWITCH_ENTITY_ID
from tests.common import MockConfigEntry
@ -26,18 +27,20 @@ async def test_wallbox_switch_class(
assert state
assert state.state == "on"
with requests_mock.Mocker() as mock_request:
mock_request.get(
"https://user-api.wall-box.com/users/signin",
json=authorisation_response,
status_code=200,
)
mock_request.post(
"https://api.wall-box.com/v3/chargers/12345/remote-action",
json={CHARGER_STATUS_ID_KEY: 193},
status_code=200,
)
with (
patch(
"homeassistant.components.wallbox.Wallbox.authenticate",
new=Mock(return_value=authorisation_response),
),
patch(
"homeassistant.components.wallbox.Wallbox.pauseChargingSession",
new=Mock(return_value={CHARGER_STATUS_ID_KEY: 193}),
),
patch(
"homeassistant.components.wallbox.Wallbox.resumeChargingSession",
new=Mock(return_value={CHARGER_STATUS_ID_KEY: 193}),
),
):
await hass.services.async_call(
"switch",
SERVICE_TURN_ON,
@ -64,19 +67,18 @@ async def test_wallbox_switch_class_connection_error(
await setup_integration(hass, entry)
with requests_mock.Mocker() as mock_request:
mock_request.get(
"https://user-api.wall-box.com/users/signin",
json=authorisation_response,
status_code=200,
)
mock_request.post(
"https://api.wall-box.com/v3/chargers/12345/remote-action",
json={CHARGER_STATUS_ID_KEY: 193},
status_code=404,
)
with pytest.raises(ConnectionError):
with (
patch(
"homeassistant.components.wallbox.Wallbox.authenticate",
new=Mock(return_value=authorisation_response),
),
patch(
"homeassistant.components.wallbox.Wallbox.resumeChargingSession",
new=Mock(side_effect=http_404_error),
),
pytest.raises(HomeAssistantError),
):
# Test behavior when a connection error occurs
await hass.services.async_call(
"switch",
SERVICE_TURN_ON,
@ -85,37 +87,27 @@ async def test_wallbox_switch_class_connection_error(
},
blocking=True,
)
with pytest.raises(ConnectionError):
await hass.services.async_call(
"switch",
SERVICE_TURN_OFF,
{
ATTR_ENTITY_ID: MOCK_SWITCH_ENTITY_ID,
},
blocking=True,
)
async def test_wallbox_switch_class_authentication_error(
async def test_wallbox_switch_class_too_many_requests(
hass: HomeAssistant, entry: MockConfigEntry
) -> None:
"""Test wallbox switch class connection error."""
await setup_integration(hass, entry)
with requests_mock.Mocker() as mock_request:
mock_request.get(
"https://user-api.wall-box.com/users/signin",
json=authorisation_response,
status_code=200,
)
mock_request.post(
"https://api.wall-box.com/v3/chargers/12345/remote-action",
json={CHARGER_STATUS_ID_KEY: 193},
status_code=403,
)
with pytest.raises(ConfigEntryAuthFailed):
with (
patch(
"homeassistant.components.wallbox.Wallbox.authenticate",
new=Mock(return_value=authorisation_response),
),
patch(
"homeassistant.components.wallbox.Wallbox.resumeChargingSession",
new=Mock(side_effect=http_429_error),
),
pytest.raises(HomeAssistantError),
):
# Test behavior when a connection error occurs
await hass.services.async_call(
"switch",
SERVICE_TURN_ON,
@ -124,12 +116,3 @@ async def test_wallbox_switch_class_authentication_error(
},
blocking=True,
)
with pytest.raises(ConfigEntryAuthFailed):
await hass.services.async_call(
"switch",
SERVICE_TURN_OFF,
{
ATTR_ENTITY_ID: MOCK_SWITCH_ENTITY_ID,
},
blocking=True,
)