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

Add Number platform to Wallbox (#52786)

Browse Source 
Co-authored-by: jan iversen <jancasacondor@gmail.com>
This commit is contained in:
hesselonline 2021-10-27 19:53:14 +02:00 committed by GitHub
parent abb84d9756
commit 6d6cb03848
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
13 changed files with 584 additions and 150 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

96
homeassistant/components/wallbox/__init__.py
View File

@ -6,37 +6,40 @@ import logging
import requests
from wallbox import Wallbox
from homeassistant import exceptions
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import CONF_CONNECTIONS, CONF_ROUND, CONF_SENSOR_TYPES, CONF_STATION, DOMAIN
from .const import (
CONF_CONNECTIONS,
CONF_DATA_KEY,
CONF_MAX_CHARGING_CURRENT_KEY,
CONF_ROUND,
CONF_SENSOR_TYPES,
CONF_STATION,
DOMAIN,
)
_LOGGER = logging.getLogger(__name__)
PLATFORMS = ["sensor"]
PLATFORMS = ["sensor", "number"]
UPDATE_INTERVAL = 30
class WallboxHub:
"""Wallbox Hub class."""
class WallboxCoordinator(DataUpdateCoordinator):
"""Wallbox Coordinator class."""
def __init__(self, station, username, password, hass):
def __init__(self, station, wallbox, hass):
"""Initialize."""
self._station = station
self._username = username
self._password = password
self._wallbox = Wallbox(self._username, self._password)
self._hass = hass
self._coordinator = DataUpdateCoordinator(
self._wallbox = wallbox
super().__init__(
hass,
_LOGGER,
# Name of the data. For logging purposes.
name="wallbox",
update_method=self.async_get_data,
# Polling interval. Will only be polled if there are subscribers.
name=DOMAIN,
update_interval=timedelta(seconds=UPDATE_INTERVAL),
)
@ -50,11 +53,24 @@ class WallboxHub:
raise InvalidAuth from wallbox_connection_error
raise ConnectionError from wallbox_connection_error
def _validate(self):
"""Authenticate using Wallbox API."""
try:
self._wallbox.authenticate()
return True
except requests.exceptions.HTTPError as wallbox_connection_error:
if wallbox_connection_error.response.status_code == 403:
raise InvalidAuth from wallbox_connection_error
raise ConnectionError from wallbox_connection_error
def _get_data(self):
"""Get new sensor data for Wallbox component."""
try:
self._authenticate()
data = self._wallbox.getChargerStatus(self._station)
data[CONF_MAX_CHARGING_CURRENT_KEY] = data[CONF_DATA_KEY][
CONF_MAX_CHARGING_CURRENT_KEY
]
filtered_data = {k: data[k] for k in CONF_SENSOR_TYPES if k in data}
@ -69,42 +85,52 @@ class WallboxHub:
except requests.exceptions.HTTPError as wallbox_connection_error:
raise ConnectionError from wallbox_connection_error
async def async_coordinator_first_refresh(self):
"""Refresh coordinator for the first time."""
await self._coordinator.async_config_entry_first_refresh()
def _set_charging_current(self, charging_current):
"""Set maximum charging current for Wallbox."""
try:
self._authenticate()
self._wallbox.setMaxChargingCurrent(self._station, charging_current)
except requests.exceptions.HTTPError as wallbox_connection_error:
if wallbox_connection_error.response.status_code == 403:
raise InvalidAuth from wallbox_connection_error
raise ConnectionError from wallbox_connection_error
async def async_authenticate(self) -> bool:
"""Authenticate using Wallbox API."""
return await self._hass.async_add_executor_job(self._authenticate)
async def async_set_charging_current(self, charging_current):
"""Set maximum charging current for Wallbox."""
await self.hass.async_add_executor_job(
self._set_charging_current, charging_current
)
await self.async_request_refresh()
async def async_get_data(self) -> bool:
async def _async_update_data(self) -> bool:
"""Get new sensor data for Wallbox component."""
data = await self._hass.async_add_executor_job(self._get_data)
data = await self.hass.async_add_executor_job(self._get_data)
return data
@property
def coordinator(self):
"""Return the coordinator."""
return self._coordinator
async def async_validate_input(self) -> bool:
"""Get new sensor data for Wallbox component."""
data = await self.hass.async_add_executor_job(self._validate)
return data
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Wallbox from a config entry."""
wallbox = WallboxHub(
wallbox = Wallbox(entry.data[CONF_USERNAME], entry.data[CONF_PASSWORD])
wallbox_coordinator = WallboxCoordinator(
entry.data[CONF_STATION],
entry.data[CONF_USERNAME],
entry.data[CONF_PASSWORD],
wallbox,
hass,
)
await wallbox.async_authenticate()
await wallbox_coordinator.async_validate_input()
await wallbox.async_coordinator_first_refresh()
await wallbox_coordinator.async_config_entry_first_refresh()
hass.data.setdefault(DOMAIN, {CONF_CONNECTIONS: {}})
hass.data[DOMAIN][CONF_CONNECTIONS][entry.entry_id] = wallbox
hass.data[DOMAIN][CONF_CONNECTIONS][entry.entry_id] = wallbox_coordinator
for platform in PLATFORMS:
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(entry, platform)
)
@ -116,10 +142,10 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if unload_ok:
hass.data[DOMAIN]["connections"].pop(entry.entry_id)
hass.data[DOMAIN][CONF_CONNECTIONS].pop(entry.entry_id)
return unload_ok
class InvalidAuth(exceptions.HomeAssistantError):
class InvalidAuth(HomeAssistantError):
"""Error to indicate there is invalid auth."""

8
homeassistant/components/wallbox/config_flow.py
View File

@ -1,10 +1,11 @@
"""Config flow for Wallbox integration."""
import voluptuous as vol
from wallbox import Wallbox
from homeassistant import config_entries, core
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from . import InvalidAuth, WallboxHub
from . import InvalidAuth, WallboxCoordinator
from .const import CONF_STATION, DOMAIN
COMPONENT_DOMAIN = DOMAIN
@ -23,9 +24,10 @@ async def validate_input(hass: core.HomeAssistant, data):
Data has the keys from STEP_USER_DATA_SCHEMA with values provided by the user.
"""
hub = WallboxHub(data["station"], data["username"], data["password"], hass)
wallbox = Wallbox(data["username"], data["password"])
wallbox_coordinator = WallboxCoordinator(data["station"], wallbox, hass)
await hub.async_get_data()
await wallbox_coordinator.async_validate_input()
# Return info that you want to store in the config entry.
return {"title": "Wallbox Portal"}

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

@ -1,99 +1,123 @@
"""Constants for the Wallbox integration."""
from homeassistant.const import (
CONF_DEVICE_CLASS,
CONF_ICON,
CONF_NAME,
CONF_UNIT_OF_MEASUREMENT,
DEVICE_CLASS_BATTERY,
DEVICE_CLASS_CURRENT,
DEVICE_CLASS_ENERGY,
DEVICE_CLASS_POWER,
ELECTRIC_CURRENT_AMPERE,
ENERGY_KILO_WATT_HOUR,
LENGTH_KILOMETERS,
PERCENTAGE,
POWER_KILO_WATT,
STATE_UNAVAILABLE,
)
DOMAIN = "wallbox"
CONF_STATION = "station"
CONF_ADDED_ENERGY_KEY = "added_energy"
CONF_ADDED_RANGE_KEY = "added_range"
CONF_CHARGING_POWER_KEY = "charging_power"
CONF_CHARGING_SPEED_KEY = "charging_speed"
CONF_CHARGING_TIME_KEY = "charging_time"
CONF_COST_KEY = "cost"
CONF_CURRENT_MODE_KEY = "current_mode"
CONF_DATA_KEY = "config_data"
CONF_DEPOT_PRICE_KEY = "depot_price"
CONF_MAX_AVAILABLE_POWER_KEY = "max_available_power"
CONF_MAX_CHARGING_CURRENT_KEY = "max_charging_current"
CONF_STATE_OF_CHARGE_KEY = "state_of_charge"
CONF_STATUS_DESCRIPTION_KEY = "status_description"
CONF_CONNECTIONS = "connections"
CONF_ROUND = "round"
CONF_SENSOR_TYPES = {
"charging_power": {
CONF_ICON: "mdi:ev-station",
CONF_CHARGING_POWER_KEY: {
CONF_ICON: None,
CONF_NAME: "Charging Power",
CONF_ROUND: 2,
CONF_UNIT_OF_MEASUREMENT: POWER_KILO_WATT,
STATE_UNAVAILABLE: False,
CONF_DEVICE_CLASS: DEVICE_CLASS_POWER,
},
"max_available_power": {
CONF_ICON: "mdi:ev-station",
CONF_MAX_AVAILABLE_POWER_KEY: {
CONF_ICON: None,
CONF_NAME: "Max Available Power",
CONF_ROUND: 0,
CONF_UNIT_OF_MEASUREMENT: ELECTRIC_CURRENT_AMPERE,
STATE_UNAVAILABLE: False,
CONF_DEVICE_CLASS: DEVICE_CLASS_CURRENT,
},
"charging_speed": {
CONF_CHARGING_SPEED_KEY: {
CONF_ICON: "mdi:speedometer",
CONF_NAME: "Charging Speed",
CONF_ROUND: 0,
CONF_UNIT_OF_MEASUREMENT: None,
STATE_UNAVAILABLE: False,
CONF_DEVICE_CLASS: None,
},
"added_range": {
CONF_ADDED_RANGE_KEY: {
CONF_ICON: "mdi:map-marker-distance",
CONF_NAME: "Added Range",
CONF_ROUND: 0,
CONF_UNIT_OF_MEASUREMENT: LENGTH_KILOMETERS,
STATE_UNAVAILABLE: False,
CONF_DEVICE_CLASS: None,
},
"added_energy": {
CONF_ICON: "mdi:battery-positive",
CONF_ADDED_ENERGY_KEY: {
CONF_ICON: None,
CONF_NAME: "Added Energy",
CONF_ROUND: 2,
CONF_UNIT_OF_MEASUREMENT: ENERGY_KILO_WATT_HOUR,
STATE_UNAVAILABLE: False,
CONF_DEVICE_CLASS: DEVICE_CLASS_ENERGY,
},
"charging_time": {
CONF_CHARGING_TIME_KEY: {
CONF_ICON: "mdi:timer",
CONF_NAME: "Charging Time",
CONF_ROUND: None,
CONF_UNIT_OF_MEASUREMENT: None,
STATE_UNAVAILABLE: False,
CONF_DEVICE_CLASS: None,
},
"cost": {
CONF_COST_KEY: {
CONF_ICON: "mdi:ev-station",
CONF_NAME: "Cost",
CONF_ROUND: None,
CONF_UNIT_OF_MEASUREMENT: None,
STATE_UNAVAILABLE: False,
CONF_DEVICE_CLASS: None,
},
"state_of_charge": {
CONF_ICON: "mdi:battery-charging-80",
CONF_STATE_OF_CHARGE_KEY: {
CONF_ICON: None,
CONF_NAME: "State of Charge",
CONF_ROUND: None,
CONF_UNIT_OF_MEASUREMENT: PERCENTAGE,
STATE_UNAVAILABLE: False,
CONF_DEVICE_CLASS: DEVICE_CLASS_BATTERY,
},
"current_mode": {
CONF_CURRENT_MODE_KEY: {
CONF_ICON: "mdi:ev-station",
CONF_NAME: "Current Mode",
CONF_ROUND: None,
CONF_UNIT_OF_MEASUREMENT: None,
STATE_UNAVAILABLE: False,
CONF_DEVICE_CLASS: None,
},
"depot_price": {
CONF_DEPOT_PRICE_KEY: {
CONF_ICON: "mdi:ev-station",
CONF_NAME: "Depot Price",
CONF_ROUND: 2,
CONF_UNIT_OF_MEASUREMENT: None,
STATE_UNAVAILABLE: False,
CONF_DEVICE_CLASS: None,
},
"status_description": {
CONF_STATUS_DESCRIPTION_KEY: {
CONF_ICON: "mdi:ev-station",
CONF_NAME: "Status Description",
CONF_ROUND: None,
CONF_UNIT_OF_MEASUREMENT: None,
STATE_UNAVAILABLE: False,
CONF_DEVICE_CLASS: None,
},
CONF_MAX_CHARGING_CURRENT_KEY: {
CONF_ICON: None,
CONF_NAME: "Max. Charging Current",
CONF_ROUND: None,
CONF_UNIT_OF_MEASUREMENT: ELECTRIC_CURRENT_AMPERE,
CONF_DEVICE_CLASS: DEVICE_CLASS_CURRENT,
},
}

56
homeassistant/components/wallbox/number.py Normal file
View File

@ -0,0 +1,56 @@
"""Home Assistant component for accessing the Wallbox Portal API. The sensor component creates multiple sensors regarding wallbox performance."""
from homeassistant.components.number import NumberEntity
from homeassistant.const import CONF_DEVICE_CLASS
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import InvalidAuth
from .const import (
CONF_CONNECTIONS,
CONF_MAX_AVAILABLE_POWER_KEY,
CONF_MAX_CHARGING_CURRENT_KEY,
CONF_NAME,
CONF_SENSOR_TYPES,
DOMAIN,
)
async def async_setup_entry(hass, config, async_add_entities):
"""Create wallbox sensor entities in HASS."""
coordinator = hass.data[DOMAIN][CONF_CONNECTIONS][config.entry_id]
# Check if the user is authorized to change current, if so, add number component:
try:
await coordinator.async_set_charging_current(
coordinator.data[CONF_MAX_CHARGING_CURRENT_KEY]
)
except InvalidAuth:
pass
else:
async_add_entities([WallboxNumber(coordinator, config)])
class WallboxNumber(CoordinatorEntity, NumberEntity):
"""Representation of the Wallbox portal."""
def __init__(self, coordinator, config):
"""Initialize a Wallbox sensor."""
super().__init__(coordinator)
_properties = CONF_SENSOR_TYPES[CONF_MAX_CHARGING_CURRENT_KEY]
self._coordinator = coordinator
self._attr_name = f"{config.title} {_properties[CONF_NAME]}"
self._attr_min_value = 6
self._attr_device_class = _properties[CONF_DEVICE_CLASS]
@property
def max_value(self):
"""Return the maximum available current."""
return self._coordinator.data[CONF_MAX_AVAILABLE_POWER_KEY]
@property
def value(self):
"""Return the state of the sensor."""
return self._coordinator.data[CONF_MAX_CHARGING_CURRENT_KEY]
async def async_set_value(self, value: float):
"""Set the value of the entity."""
await self._coordinator.async_set_charging_current(value)

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

@ -1,6 +1,7 @@
"""Home Assistant component for accessing the Wallbox Portal API. The sensor component creates multiple sensors regarding wallbox performance."""
from homeassistant.components.sensor import SensorEntity
from homeassistant.const import CONF_DEVICE_CLASS
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import (
@ -18,9 +19,7 @@ UPDATE_INTERVAL = 30
async def async_setup_entry(hass, config, async_add_entities):
"""Create wallbox sensor entities in HASS."""
wallbox = hass.data[DOMAIN][CONF_CONNECTIONS][config.entry_id]
coordinator = wallbox.coordinator
coordinator = hass.data[DOMAIN][CONF_CONNECTIONS][config.entry_id]
async_add_entities(
WallboxSensor(coordinator, idx, ent, config)
@ -34,28 +33,15 @@ class WallboxSensor(CoordinatorEntity, SensorEntity):
def __init__(self, coordinator, idx, ent, config):
"""Initialize a Wallbox sensor."""
super().__init__(coordinator)
self._properties = CONF_SENSOR_TYPES[ent]
self._name = f"{config.title} {self._properties[CONF_NAME]}"
self._icon = self._properties[CONF_ICON]
self._unit = self._properties[CONF_UNIT_OF_MEASUREMENT]
self._attr_name = f"{config.title} {CONF_SENSOR_TYPES[ent][CONF_NAME]}"
self._attr_icon = CONF_SENSOR_TYPES[ent][CONF_ICON]
self._attr_native_unit_of_measurement = CONF_SENSOR_TYPES[ent][
CONF_UNIT_OF_MEASUREMENT
]
self._attr_device_class = CONF_SENSOR_TYPES[ent][CONF_DEVICE_CLASS]
self._ent = ent
@property
def name(self):
"""Return the name of the sensor."""
return self._name
@property
def native_value(self):
"""Return the state of the sensor."""
return self.coordinator.data[self._ent]
@property
def native_unit_of_measurement(self):
"""Return the unit of the sensor."""
return self._unit
@property
def icon(self):
"""Return the icon of the sensor."""
return self._icon

3
homeassistant/components/wallbox/translations/en.json
View File

@ -17,6 +17,5 @@
}
}
}
},
"title": "Wallbox"
}
}

10
homeassistant/components/wallbox/translations/nl.json
View File

@ -1,7 +1,8 @@
{
"config": {
"abort": {
"already_configured": "Apparaat is al geconfigureerd"
"already_configured": "Apparaat is al geconfigureerd",
"reauth_successful": "Herauthenticatie was succesvol"
},
"error": {
"cannot_connect": "Kan geen verbinding maken",
@ -15,6 +16,13 @@
"station": "Station Serienummer",
"username": "Gebruikersnaam"
}
},
"reauth_confirm": {
"data": {
"password": "Wachtwoord",
"station": "Station Serienummer",
"username": "Gebruikersnaam"
}
}
}
},

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

@ -5,35 +5,73 @@ import json
import requests_mock
from homeassistant.components.wallbox.const import CONF_STATION, DOMAIN
from homeassistant.components.wallbox.const import (
CONF_ADDED_ENERGY_KEY,
CONF_ADDED_RANGE_KEY,
CONF_CHARGING_POWER_KEY,
CONF_CHARGING_SPEED_KEY,
CONF_DATA_KEY,
CONF_MAX_AVAILABLE_POWER_KEY,
CONF_MAX_CHARGING_CURRENT_KEY,
CONF_STATION,
DOMAIN,
)
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from tests.common import MockConfigEntry
from tests.components.wallbox.const import (
CONF_ERROR,
CONF_JWT,
CONF_STATUS,
CONF_TTL,
CONF_USER_ID,
)
test_response = json.loads(
'{"charging_power": 0,"max_available_power": "xx","charging_speed": 0,"added_range": "xx","added_energy": "44.697"}'
json.dumps(
{
CONF_CHARGING_POWER_KEY: 0,
CONF_MAX_AVAILABLE_POWER_KEY: 25,
CONF_CHARGING_SPEED_KEY: 0,
CONF_ADDED_RANGE_KEY: "xx",
CONF_ADDED_ENERGY_KEY: "44.697",
CONF_DATA_KEY: {CONF_MAX_CHARGING_CURRENT_KEY: 24},
}
)
)
authorisation_response = json.loads(
json.dumps(
{
CONF_JWT: "fakekeyhere",
CONF_USER_ID: 12345,
CONF_TTL: 145656758,
CONF_ERROR: "false",
CONF_STATUS: 200,
}
)
)
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_USERNAME: "test_username",
CONF_PASSWORD: "test_password",
CONF_STATION: "12345",
},
entry_id="testEntry",
)
async def setup_integration(hass):
"""Test wallbox sensor class setup."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_USERNAME: "test_username",
CONF_PASSWORD: "test_password",
CONF_STATION: "12345",
},
entry_id="testEntry",
)
entry.add_to_hass(hass)
with requests_mock.Mocker() as mock_request:
mock_request.get(
"https://api.wall-box.com/auth/token/user",
text='{"jwt":"fakekeyhere","user_id":12345,"ttl":145656758,"error":false,"status":200}',
json=authorisation_response,
status_code=HTTPStatus.OK,
)
mock_request.get(
@ -41,5 +79,65 @@ async def setup_integration(hass):
json=test_response,
status_code=HTTPStatus.OK,
)
mock_request.put(
"https://api.wall-box.com/v2/charger/12345",
json=json.loads(json.dumps({CONF_MAX_CHARGING_CURRENT_KEY: 20})),
status_code=HTTPStatus.OK,
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
async def setup_integration_connection_error(hass):
"""Test wallbox sensor class setup with a connection error."""
with requests_mock.Mocker() as mock_request:
mock_request.get(
"https://api.wall-box.com/auth/token/user",
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,
)
mock_request.put(
"https://api.wall-box.com/v2/charger/12345",
json=json.loads(json.dumps({CONF_MAX_CHARGING_CURRENT_KEY: 20})),
status_code=HTTPStatus.FORBIDDEN,
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
async def setup_integration_read_only(hass):
"""Test wallbox sensor class setup for read only."""
with requests_mock.Mocker() as mock_request:
mock_request.get(
"https://api.wall-box.com/auth/token/user",
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,
)
mock_request.put(
"https://api.wall-box.com/v2/charger/12345",
json=test_response,
status_code=HTTPStatus.FORBIDDEN,
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()

10
tests/components/wallbox/const.py Normal file
View File

@ -0,0 +1,10 @@
"""Provides constants for Wallbox component tests."""
CONF_JWT = "jwt"
CONF_USER_ID = "user_id"
CONF_TTL = "ttl"
CONF_ERROR = "error"
CONF_STATUS = "status"
CONF_MOCK_NUMBER_ENTITY_ID = "number.mock_title_max_charging_current"
CONF_MOCK_SENSOR_CHARGING_SPEED_ID = "sensor.mock_title_charging_speed"
CONF_MOCK_SENSOR_CHARGING_POWER_ID = "sensor.mock_title_charging_power"

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

@ -6,11 +6,61 @@ import requests_mock
from homeassistant import config_entries, data_entry_flow
from homeassistant.components.wallbox import config_flow
from homeassistant.components.wallbox.const import DOMAIN
from homeassistant.components.wallbox.const import (
CONF_ADDED_ENERGY_KEY,
CONF_ADDED_RANGE_KEY,
CONF_CHARGING_POWER_KEY,
CONF_CHARGING_SPEED_KEY,
CONF_DATA_KEY,
CONF_MAX_AVAILABLE_POWER_KEY,
CONF_MAX_CHARGING_CURRENT_KEY,
DOMAIN,
)
from homeassistant.core import HomeAssistant
from tests.components.wallbox.const import (
CONF_ERROR,
CONF_JWT,
CONF_STATUS,
CONF_TTL,
CONF_USER_ID,
)
test_response = json.loads(
'{"charging_power": 0,"max_available_power": 25,"charging_speed": 0,"added_range": 372,"added_energy": 44.697}'
json.dumps(
{
CONF_CHARGING_POWER_KEY: 0,
CONF_MAX_AVAILABLE_POWER_KEY: "xx",
CONF_CHARGING_SPEED_KEY: 0,
CONF_ADDED_RANGE_KEY: "xx",
CONF_ADDED_ENERGY_KEY: "44.697",
CONF_DATA_KEY: {CONF_MAX_CHARGING_CURRENT_KEY: 24},
}
)
)
authorisation_response = json.loads(
json.dumps(
{
CONF_JWT: "fakekeyhere",
CONF_USER_ID: 12345,
CONF_TTL: 145656758,
CONF_ERROR: "false",
CONF_STATUS: 200,
}
)
)
authorisation_response_unauthorised = json.loads(
json.dumps(
{
CONF_JWT: "fakekeyhere",
CONF_USER_ID: 12345,
CONF_TTL: 145656758,
CONF_ERROR: "false",
CONF_STATUS: 404,
}
)
)
@ -33,7 +83,12 @@ async def test_form_cannot_authenticate(hass):
with requests_mock.Mocker() as mock_request:
mock_request.get(
"https://api.wall-box.com/auth/token/user",
text='{"jwt":"fakekeyhere","user_id":12345,"ttl":145656758,"error":false,"status":200}',
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,
)
result2 = await hass.config_entries.flow.async_configure(
@ -58,12 +113,12 @@ async def test_form_cannot_connect(hass):
with requests_mock.Mocker() as mock_request:
mock_request.get(
"https://api.wall-box.com/auth/token/user",
text='{"jwt":"fakekeyhere","user_id":12345,"ttl":145656758,"error":false,"status":200}',
status_code=HTTPStatus.OK,
json=authorisation_response_unauthorised,
status_code=HTTPStatus.NOT_FOUND,
)
mock_request.get(
"https://api.wall-box.com/chargers/status/12345",
text='{"Temperature": 100, "Location": "Toronto", "Datetime": "2020-07-23", "Units": "Celsius"}',
json=test_response,
status_code=HTTPStatus.NOT_FOUND,
)
result2 = await hass.config_entries.flow.async_configure(
@ -80,7 +135,7 @@ async def test_form_cannot_connect(hass):
async def test_form_validate_input(hass):
"""Test we handle cannot connect error."""
"""Test we can validate input."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
@ -88,12 +143,12 @@ async def test_form_validate_input(hass):
with requests_mock.Mocker() as mock_request:
mock_request.get(
"https://api.wall-box.com/auth/token/user",
text='{"jwt":"fakekeyhere","user_id":12345,"ttl":145656758,"error":false,"status":200}',
json=authorisation_response,
status_code=HTTPStatus.OK,
)
mock_request.get(
"https://api.wall-box.com/chargers/status/12345",
text='{"Temperature": 100, "Location": "Toronto", "Datetime": "2020-07-23", "Units": "Celsius"}',
json=test_response,
status_code=HTTPStatus.OK,
)
result2 = await hass.config_entries.flow.async_configure(

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

@ -1,35 +1,122 @@
"""Test Wallbox Init Component."""
import json
from homeassistant.components.wallbox.const import CONF_STATION, DOMAIN
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
import requests_mock
from homeassistant.components.wallbox import (
CONF_CONNECTIONS,
CONF_MAX_CHARGING_CURRENT_KEY,
)
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
from tests.components.wallbox import setup_integration
from . import test_response
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_USERNAME: "test_username",
CONF_PASSWORD: "test_password",
CONF_STATION: "12345",
},
entry_id="testEntry",
from tests.components.wallbox import (
DOMAIN,
entry,
setup_integration,
setup_integration_connection_error,
setup_integration_read_only,
)
from tests.components.wallbox.const import (
CONF_ERROR,
CONF_JWT,
CONF_STATUS,
CONF_TTL,
CONF_USER_ID,
)
test_response = json.loads(
'{"charging_power": 0,"max_available_power": 25,"charging_speed": 0,"added_range": 372,"added_energy": 44.697}'
)
test_response_rounding_error = json.loads(
'{"charging_power": "XX","max_available_power": "xx","charging_speed": 0,"added_range": "xx","added_energy": "XX"}'
authorisation_response = json.loads(
json.dumps(
{
CONF_JWT: "fakekeyhere",
CONF_USER_ID: 12345,
CONF_TTL: 145656758,
CONF_ERROR: "false",
CONF_STATUS: 200,
}
)
)
async def test_wallbox_unload_entry(hass: HomeAssistant):
async def test_wallbox_setup_unload_entry(hass: HomeAssistant):
"""Test Wallbox Unload."""
await setup_integration(hass)
assert entry.state == ConfigEntryState.LOADED
assert await hass.config_entries.async_unload(entry.entry_id)
assert entry.state == ConfigEntryState.NOT_LOADED
async def test_wallbox_unload_entry_connection_error(hass: HomeAssistant):
"""Test Wallbox Unload Connection Error."""
await setup_integration_connection_error(hass)
assert entry.state == ConfigEntryState.SETUP_ERROR
assert await hass.config_entries.async_unload(entry.entry_id)
assert entry.state == ConfigEntryState.NOT_LOADED
async def test_wallbox_refresh_failed_invalid_auth(hass: HomeAssistant):
"""Test Wallbox setup with authentication error."""
await setup_integration(hass)
assert entry.state == ConfigEntryState.LOADED
with requests_mock.Mocker() as mock_request:
mock_request.get(
"https://api.wall-box.com/auth/token/user",
json=authorisation_response,
status_code=403,
)
mock_request.put(
"https://api.wall-box.com/v2/charger/12345",
json=json.loads(json.dumps({CONF_MAX_CHARGING_CURRENT_KEY: 20})),
status_code=403,
)
wallbox = hass.data[DOMAIN][CONF_CONNECTIONS][entry.entry_id]
await wallbox.async_refresh()
assert await hass.config_entries.async_unload(entry.entry_id)
assert entry.state == ConfigEntryState.NOT_LOADED
async def test_wallbox_refresh_failed_connection_error(hass: HomeAssistant):
"""Test Wallbox setup with connection error."""
await setup_integration(hass)
assert entry.state == ConfigEntryState.LOADED
with requests_mock.Mocker() as mock_request:
mock_request.get(
"https://api.wall-box.com/auth/token/user",
json=authorisation_response,
status_code=200,
)
mock_request.get(
"https://api.wall-box.com/chargers/status/12345",
json=test_response,
status_code=403,
)
wallbox = hass.data[DOMAIN][CONF_CONNECTIONS][entry.entry_id]
await wallbox.async_refresh()
assert await hass.config_entries.async_unload(entry.entry_id)
assert entry.state == ConfigEntryState.NOT_LOADED
async def test_wallbox_refresh_failed_read_only(hass: HomeAssistant):
"""Test Wallbox setup for read-only user."""
await setup_integration_read_only(hass)
assert entry.state == ConfigEntryState.LOADED
assert await hass.config_entries.async_unload(entry.entry_id)
assert entry.state == ConfigEntryState.NOT_LOADED

91
tests/components/wallbox/test_number.py Normal file
View File

@ -0,0 +1,91 @@
"""Test Wallbox Switch component."""
import json
import pytest
import requests_mock
from homeassistant.components.input_number import ATTR_VALUE, SERVICE_SET_VALUE
from homeassistant.components.wallbox import CONF_MAX_CHARGING_CURRENT_KEY
from homeassistant.const import ATTR_ENTITY_ID
from tests.components.wallbox import entry, setup_integration
from tests.components.wallbox.const import (
CONF_ERROR,
CONF_JWT,
CONF_MOCK_NUMBER_ENTITY_ID,
CONF_STATUS,
CONF_TTL,
CONF_USER_ID,
)
authorisation_response = json.loads(
json.dumps(
{
CONF_JWT: "fakekeyhere",
CONF_USER_ID: 12345,
CONF_TTL: 145656758,
CONF_ERROR: "false",
CONF_STATUS: 200,
}
)
)
async def test_wallbox_number_class(hass):
"""Test wallbox sensor class."""
await setup_integration(hass)
with requests_mock.Mocker() as mock_request:
mock_request.get(
"https://api.wall-box.com/auth/token/user",
json=authorisation_response,
status_code=200,
)
mock_request.put(
"https://api.wall-box.com/v2/charger/12345",
json=json.loads(json.dumps({CONF_MAX_CHARGING_CURRENT_KEY: 20})),
status_code=200,
)
await hass.services.async_call(
"number",
SERVICE_SET_VALUE,
{
ATTR_ENTITY_ID: CONF_MOCK_NUMBER_ENTITY_ID,
ATTR_VALUE: 20,
},
blocking=True,
)
await hass.config_entries.async_unload(entry.entry_id)
async def test_wallbox_number_class_connection_error(hass):
"""Test wallbox sensor class."""
await setup_integration(hass)
with requests_mock.Mocker() as mock_request:
mock_request.get(
"https://api.wall-box.com/auth/token/user",
json=authorisation_response,
status_code=200,
)
mock_request.put(
"https://api.wall-box.com/v2/charger/12345",
json=json.loads(json.dumps({CONF_MAX_CHARGING_CURRENT_KEY: 20})),
status_code=404,
)
with pytest.raises(ConnectionError):
await hass.services.async_call(
"number",
SERVICE_SET_VALUE,
{
ATTR_ENTITY_ID: CONF_MOCK_NUMBER_ENTITY_ID,
ATTR_VALUE: 20,
},
blocking=True,
)
await hass.config_entries.async_unload(entry.entry_id)

30
tests/components/wallbox/test_sensor.py
View File

@ -1,19 +1,10 @@
"""Test Wallbox Switch component."""
from homeassistant.const import CONF_ICON, CONF_UNIT_OF_MEASUREMENT, POWER_KILO_WATT
from homeassistant.components.wallbox.const import CONF_STATION, DOMAIN
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from tests.common import MockConfigEntry
from tests.components.wallbox import setup_integration
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_USERNAME: "test_username",
CONF_PASSWORD: "test_password",
CONF_STATION: "12345",
},
entry_id="testEntry",
from tests.components.wallbox import entry, setup_integration
from tests.components.wallbox.const import (
CONF_MOCK_SENSOR_CHARGING_POWER_ID,
CONF_MOCK_SENSOR_CHARGING_SPEED_ID,
)
@ -22,11 +13,12 @@ async def test_wallbox_sensor_class(hass):
await setup_integration(hass)
state = hass.states.get("sensor.mock_title_charging_power")
assert state.attributes["unit_of_measurement"] == "kW"
assert state.attributes["icon"] == "mdi:ev-station"
state = hass.states.get(CONF_MOCK_SENSOR_CHARGING_POWER_ID)
assert state.attributes[CONF_UNIT_OF_MEASUREMENT] == POWER_KILO_WATT
assert state.name == "Mock Title Charging Power"
state = hass.states.get("sensor.mock_title_charging_speed")
assert state.attributes["icon"] == "mdi:speedometer"
state = hass.states.get(CONF_MOCK_SENSOR_CHARGING_SPEED_ID)
assert state.attributes[CONF_ICON] == "mdi:speedometer"
assert state.name == "Mock Title Charging Speed"
await hass.config_entries.async_unload(entry.entry_id)