evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Update powerwall for tesla_powerwall 0.5.0 which is async (#107164)

Browse source 
Co-authored-by: J. Nick Koston <nick@koston.org>
This commit is contained in:
bubonicbob 2024-01-10 13:21:53 -08:00 • committed by GitHub
parent bdba6f41c9
commit c74bef265a
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
20 changed files with 401 additions and 209 deletions
Download patch file Download diff file Expand all files Collapse all files

142
homeassistant/components/powerwall/__init__.py
View file

@ -1,17 +1,18 @@
"""The Tesla Powerwall integration.""" """The Tesla Powerwall integration."""
from __future__ import annotations from __future__ import annotations
import contextlib import asyncio
from contextlib import AsyncExitStack
from datetime import timedelta from datetime import timedelta
import logging import logging
from typing import Optional
import requests from aiohttp import CookieJar
from tesla_powerwall import ( from tesla_powerwall import (
AccessDeniedError, AccessDeniedError,
APIError, ApiError,
MissingAttributeError, MissingAttributeError,
Powerwall, Powerwall,
PowerwallError,
PowerwallUnreachableError, PowerwallUnreachableError,
) )
@ -20,17 +21,12 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_IP_ADDRESS, CONF_PASSWORD, Platform from homeassistant.const import CONF_IP_ADDRESS, CONF_PASSWORD, Platform
from homeassistant.core import HomeAssistant, callback from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers.aiohttp_client import async_create_clientsession
import homeassistant.helpers.config_validation as cv import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from homeassistant.util.network import is_ip_address from homeassistant.util.network import is_ip_address
from .const import ( from .const import DOMAIN, POWERWALL_API_CHANGED, POWERWALL_COORDINATOR, UPDATE_INTERVAL
DOMAIN,
POWERWALL_API_CHANGED,
POWERWALL_COORDINATOR,
POWERWALL_HTTP_SESSION,
UPDATE_INTERVAL,
)
from .models import PowerwallBaseInfo, PowerwallData, PowerwallRuntimeData from .models import PowerwallBaseInfo, PowerwallData, PowerwallRuntimeData
CONFIG_SCHEMA = cv.removed(DOMAIN, raise_if_present=False) CONFIG_SCHEMA = cv.removed(DOMAIN, raise_if_present=False)
@ -70,11 +66,11 @@ class PowerwallDataManager:
"""Return true if the api has changed out from under us.""" """Return true if the api has changed out from under us."""
return self.runtime_data[POWERWALL_API_CHANGED] return self.runtime_data[POWERWALL_API_CHANGED]
def _recreate_powerwall_login(self) -> None: async def _recreate_powerwall_login(self) -> None:
"""Recreate the login on auth failure.""" """Recreate the login on auth failure."""
if self.power_wall.is_authenticated(): if self.power_wall.is_authenticated():
self.power_wall.logout() await self.power_wall.logout()
self.power_wall.login(self.password or "") await self.power_wall.login(self.password or "")
async def async_update_data(self) -> PowerwallData: async def async_update_data(self) -> PowerwallData:
"""Fetch data from API endpoint.""" """Fetch data from API endpoint."""
@ -82,17 +78,17 @@ class PowerwallDataManager:
_LOGGER.debug("Checking if update failed") _LOGGER.debug("Checking if update failed")
if self.api_changed: if self.api_changed:
raise UpdateFailed("The powerwall api has changed") raise UpdateFailed("The powerwall api has changed")
return await self.hass.async_add_executor_job(self._update_data) return await self._update_data()
def _update_data(self) -> PowerwallData: async def _update_data(self) -> PowerwallData:
"""Fetch data from API endpoint.""" """Fetch data from API endpoint."""
_LOGGER.debug("Updating data") _LOGGER.debug("Updating data")
for attempt in range(2): for attempt in range(2):
try: try:
if attempt == 1: if attempt == 1:
self._recreate_powerwall_login() await self._recreate_powerwall_login()
data = _fetch_powerwall_data(self.power_wall) data = await _fetch_powerwall_data(self.power_wall)
except PowerwallUnreachableError as err: except (asyncio.TimeoutError, PowerwallUnreachableError) as err:
raise UpdateFailed("Unable to fetch data from powerwall") from err raise UpdateFailed("Unable to fetch data from powerwall") from err
except MissingAttributeError as err: except MissingAttributeError as err:
_LOGGER.error("The powerwall api has changed: %s", str(err)) _LOGGER.error("The powerwall api has changed: %s", str(err))
@ -112,7 +108,7 @@ class PowerwallDataManager:
_LOGGER.debug("Access denied, trying to reauthenticate") _LOGGER.debug("Access denied, trying to reauthenticate")
# there is still an attempt left to authenticate, # there is still an attempt left to authenticate,
# so we continue in the loop # so we continue in the loop
except APIError as err: except ApiError as err:
raise UpdateFailed(f"Updated failed due to {err}, will retry") from err raise UpdateFailed(f"Updated failed due to {err}, will retry") from err
else: else:
return data return data
@ -121,20 +117,27 @@ class PowerwallDataManager:
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Tesla Powerwall from a config entry.""" """Set up Tesla Powerwall from a config entry."""
http_session = requests.Session()
ip_address: str = entry.data[CONF_IP_ADDRESS] ip_address: str = entry.data[CONF_IP_ADDRESS]
password: str | None = entry.data.get(CONF_PASSWORD) password: str | None = entry.data.get(CONF_PASSWORD)
power_wall = Powerwall(ip_address, http_session=http_session) http_session = async_create_clientsession(
try: hass, verify_ssl=False, cookie_jar=CookieJar(unsafe=True)
base_info = await hass.async_add_executor_job(
_login_and_fetch_base_info, power_wall, ip_address, password
) )
except PowerwallUnreachableError as err:
http_session.close() async with AsyncExitStack() as stack:
power_wall = Powerwall(ip_address, http_session=http_session, verify_ssl=False)
stack.push_async_callback(power_wall.close)
try:
base_info = await _login_and_fetch_base_info(
power_wall, ip_address, password
)
# Cancel closing power_wall on success
stack.pop_all()
except (asyncio.TimeoutError, PowerwallUnreachableError) as err:
raise ConfigEntryNotReady from err raise ConfigEntryNotReady from err
except MissingAttributeError as err: except MissingAttributeError as err:
http_session.close()
# The error might include some important information about what exactly changed. # The error might include some important information about what exactly changed.
_LOGGER.error("The powerwall api has changed: %s", str(err)) _LOGGER.error("The powerwall api has changed: %s", str(err))
persistent_notification.async_create( persistent_notification.async_create(
@ -143,10 +146,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return False return False
except AccessDeniedError as err: except AccessDeniedError as err:
_LOGGER.debug("Authentication failed", exc_info=err) _LOGGER.debug("Authentication failed", exc_info=err)
http_session.close()
raise ConfigEntryAuthFailed from err raise ConfigEntryAuthFailed from err
except APIError as err: except ApiError as err:
http_session.close()
raise ConfigEntryNotReady from err raise ConfigEntryNotReady from err
gateway_din = base_info.gateway_din gateway_din = base_info.gateway_din
@ -156,7 +157,6 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
runtime_data = PowerwallRuntimeData( runtime_data = PowerwallRuntimeData(
api_changed=False, api_changed=False,
base_info=base_info, base_info=base_info,
http_session=http_session,
coordinator=None, coordinator=None,
api_instance=power_wall, api_instance=power_wall,
) )
@ -183,44 +183,76 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return True return True
def _login_and_fetch_base_info( async def _login_and_fetch_base_info(
power_wall: Powerwall, host: str, password: str | None power_wall: Powerwall, host: str, password: str | None
) -> PowerwallBaseInfo: ) -> PowerwallBaseInfo:
"""Login to the powerwall and fetch the base info.""" """Login to the powerwall and fetch the base info."""
if password is not None: if password is not None:
power_wall.login(password) await power_wall.login(password)
return call_base_info(power_wall, host) return await _call_base_info(power_wall, host)
def call_base_info(power_wall: Powerwall, host: str) -> PowerwallBaseInfo: async def _call_base_info(power_wall: Powerwall, host: str) -> PowerwallBaseInfo:
"""Return PowerwallBaseInfo for the device.""" """Return PowerwallBaseInfo for the device."""
# Make sure the serial numbers always have the same order
gateway_din = None (
with contextlib.suppress(AssertionError, PowerwallError): gateway_din,
gateway_din = power_wall.get_gateway_din().upper() site_info,
status,
device_type,
serial_numbers,
) = await asyncio.gather(
power_wall.get_gateway_din(),
power_wall.get_site_info(),
power_wall.get_status(),
power_wall.get_device_type(),
power_wall.get_serial_numbers(),
)
# Serial numbers MUST be sorted to ensure the unique_id is always the same
# for backwards compatibility.
return PowerwallBaseInfo( return PowerwallBaseInfo(
gateway_din=gateway_din, gateway_din=gateway_din.upper(),
site_info=power_wall.get_site_info(), site_info=site_info,
status=power_wall.get_status(), status=status,
device_type=power_wall.get_device_type(), device_type=device_type,
serial_numbers=sorted(power_wall.get_serial_numbers()), serial_numbers=sorted(serial_numbers),
url=f"https://{host}", url=f"https://{host}",
) )
def _fetch_powerwall_data(power_wall: Powerwall) -> PowerwallData: async def get_backup_reserve_percentage(power_wall: Powerwall) -> Optional[float]:
"""Process and update powerwall data.""" """Return the backup reserve percentage."""
try: try:
backup_reserve = power_wall.get_backup_reserve_percentage() return await power_wall.get_backup_reserve_percentage()
except MissingAttributeError: except MissingAttributeError:
backup_reserve = None return None
async def _fetch_powerwall_data(power_wall: Powerwall) -> PowerwallData:
"""Process and update powerwall data."""
(
backup_reserve,
charge,
site_master,
meters,
grid_services_active,
grid_status,
) = await asyncio.gather(
get_backup_reserve_percentage(power_wall),
power_wall.get_charge(),
power_wall.get_sitemaster(),
power_wall.get_meters(),
power_wall.is_grid_services_active(),
power_wall.get_grid_status(),
)
return PowerwallData( return PowerwallData(
charge=power_wall.get_charge(), charge=charge,
site_master=power_wall.get_sitemaster(), site_master=site_master,
meters=power_wall.get_meters(), meters=meters,
grid_services_active=power_wall.is_grid_services_active(), grid_services_active=grid_services_active,
grid_status=power_wall.get_grid_status(), grid_status=grid_status,
backup_reserve=backup_reserve, backup_reserve=backup_reserve,
) )
@ -240,8 +272,6 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry.""" """Unload a config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS) unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
hass.data[DOMAIN][entry.entry_id][POWERWALL_HTTP_SESSION].close()
if unload_ok: if unload_ok:
hass.data[DOMAIN].pop(entry.entry_id) hass.data[DOMAIN].pop(entry.entry_id)

8
homeassistant/components/powerwall/binary_sensor.py
View file

@ -1,5 +1,7 @@
"""Support for powerwall binary sensors.""" """Support for powerwall binary sensors."""
from typing import TYPE_CHECKING
from tesla_powerwall import GridStatus, MeterType from tesla_powerwall import GridStatus, MeterType
from homeassistant.components.binary_sensor import ( from homeassistant.components.binary_sensor import (
@ -131,5 +133,9 @@ class PowerWallChargingStatusSensor(PowerWallEntity, BinarySensorEntity):
@property @property
def is_on(self) -> bool: def is_on(self) -> bool:
"""Powerwall is charging.""" """Powerwall is charging."""
meter = self.data.meters.get_meter(MeterType.BATTERY)
# Meter cannot be None because of the available property
if TYPE_CHECKING:
assert meter is not None
# is_sending_to returns true for values greater than 100 watts # is_sending_to returns true for values greater than 100 watts
return self.data.meters.get_meter(MeterType.BATTERY).is_sending_to() return meter.is_sending_to()

64
homeassistant/components/powerwall/config_flow.py
View file

@ -1,16 +1,18 @@
"""Config flow for Tesla Powerwall integration.""" """Config flow for Tesla Powerwall integration."""
from __future__ import annotations from __future__ import annotations
import asyncio
from collections.abc import Mapping from collections.abc import Mapping
import logging import logging
from typing import Any from typing import Any
from aiohttp import CookieJar
from tesla_powerwall import ( from tesla_powerwall import (
AccessDeniedError, AccessDeniedError,
MissingAttributeError, MissingAttributeError,
Powerwall, Powerwall,
PowerwallUnreachableError, PowerwallUnreachableError,
SiteInfo, SiteInfoResponse,
) )
import voluptuous as vol import voluptuous as vol
@ -18,6 +20,7 @@ from homeassistant import config_entries, core, exceptions
from homeassistant.components import dhcp from homeassistant.components import dhcp
from homeassistant.const import CONF_IP_ADDRESS, CONF_PASSWORD from homeassistant.const import CONF_IP_ADDRESS, CONF_PASSWORD
from homeassistant.data_entry_flow import FlowResult from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers.aiohttp_client import async_create_clientsession
from homeassistant.util.network import is_ip_address from homeassistant.util.network import is_ip_address
from . import async_last_update_was_successful from . import async_last_update_was_successful
@ -32,19 +35,23 @@ ENTRY_FAILURE_STATES = {
} }
def _login_and_fetch_site_info( async def _login_and_fetch_site_info(
power_wall: Powerwall, password: str power_wall: Powerwall, password: str
) -> tuple[SiteInfo, str]: ) -> tuple[SiteInfoResponse, str]:
"""Login to the powerwall and fetch the base info.""" """Login to the powerwall and fetch the base info."""
if password is not None: if password is not None:
power_wall.login(password) await power_wall.login(password)
return power_wall.get_site_info(), power_wall.get_gateway_din()
return await asyncio.gather(
power_wall.get_site_info(), power_wall.get_gateway_din()
)
def _powerwall_is_reachable(ip_address: str, password: str) -> bool: async def _powerwall_is_reachable(ip_address: str, password: str) -> bool:
"""Check if the powerwall is reachable.""" """Check if the powerwall is reachable."""
try: try:
Powerwall(ip_address).login(password) async with Powerwall(ip_address) as power_wall:
await power_wall.login(password)
except AccessDeniedError: except AccessDeniedError:
return True return True
except PowerwallUnreachableError: except PowerwallUnreachableError:
@ -59,13 +66,15 @@ async def validate_input(
Data has the keys from schema with values provided by the user. Data has the keys from schema with values provided by the user.
""" """
session = async_create_clientsession(
power_wall = Powerwall(data[CONF_IP_ADDRESS]) hass, verify_ssl=False, cookie_jar=CookieJar(unsafe=True)
)
async with Powerwall(data[CONF_IP_ADDRESS], http_session=session) as power_wall:
password = data[CONF_PASSWORD] password = data[CONF_PASSWORD]
try: try:
site_info, gateway_din = await hass.async_add_executor_job( site_info, gateway_din = await _login_and_fetch_site_info(
_login_and_fetch_site_info, power_wall, password power_wall, password
) )
except MissingAttributeError as err: except MissingAttributeError as err:
# Only log the exception without the traceback # Only log the exception without the traceback
@ -102,9 +111,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return bool( return bool(
entry.state in ENTRY_FAILURE_STATES entry.state in ENTRY_FAILURE_STATES
or not async_last_update_was_successful(self.hass, entry) or not async_last_update_was_successful(self.hass, entry)
) and not await self.hass.async_add_executor_job( ) and not await _powerwall_is_reachable(ip_address, password)
_powerwall_is_reachable, ip_address, password
)
async def async_step_dhcp(self, discovery_info: dhcp.DhcpServiceInfo) -> FlowResult: async def async_step_dhcp(self, discovery_info: dhcp.DhcpServiceInfo) -> FlowResult:
"""Handle dhcp discovery.""" """Handle dhcp discovery."""
@ -137,7 +144,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"name": gateway_din, "name": gateway_din,
"ip_address": self.ip_address, "ip_address": self.ip_address,
} }
errors, info = await self._async_try_connect( errors, info, _ = await self._async_try_connect(
{CONF_IP_ADDRESS: self.ip_address, CONF_PASSWORD: gateway_din[-5:]} {CONF_IP_ADDRESS: self.ip_address, CONF_PASSWORD: gateway_din[-5:]}
) )
if errors: if errors:
@ -152,23 +159,28 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
async def _async_try_connect( async def _async_try_connect(
self, user_input: dict[str, Any] self, user_input: dict[str, Any]
) -> tuple[dict[str, Any] | None, dict[str, str] | None]: ) -> tuple[dict[str, Any] | None, dict[str, str] | None, dict[str, str]]:
"""Try to connect to the powerwall.""" """Try to connect to the powerwall."""
info = None info = None
errors: dict[str, str] = {} errors: dict[str, str] = {}
description_placeholders: dict[str, str] = {}
try: try:
info = await validate_input(self.hass, user_input) info = await validate_input(self.hass, user_input)
except PowerwallUnreachableError: except PowerwallUnreachableError as ex:
errors[CONF_IP_ADDRESS] = "cannot_connect" errors[CONF_IP_ADDRESS] = "cannot_connect"
except WrongVersion: description_placeholders = {"error": str(ex)}
except WrongVersion as ex:
errors["base"] = "wrong_version" errors["base"] = "wrong_version"
except AccessDeniedError: description_placeholders = {"error": str(ex)}
except AccessDeniedError as ex:
errors[CONF_PASSWORD] = "invalid_auth" errors[CONF_PASSWORD] = "invalid_auth"
except Exception: # pylint: disable=broad-except description_placeholders = {"error": str(ex)}
except Exception as ex: # pylint: disable=broad-except
_LOGGER.exception("Unexpected exception") _LOGGER.exception("Unexpected exception")
errors["base"] = "unknown" errors["base"] = "unknown"
description_placeholders = {"error": str(ex)}
return errors, info return errors, info, description_placeholders
async def async_step_confirm_discovery( async def async_step_confirm_discovery(
self, user_input: dict[str, Any] | None = None self, user_input: dict[str, Any] | None = None
@ -204,8 +216,11 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
) -> FlowResult: ) -> FlowResult:
"""Handle the initial step.""" """Handle the initial step."""
errors: dict[str, str] | None = {} errors: dict[str, str] | None = {}
description_placeholders: dict[str, str] = {}
if user_input is not None: if user_input is not None:
errors, info = await self._async_try_connect(user_input) errors, info, description_placeholders = await self._async_try_connect(
user_input
)
if not errors: if not errors:
assert info is not None assert info is not None
if info["unique_id"]: if info["unique_id"]:
@ -227,6 +242,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
} }
), ),
errors=errors, errors=errors,
description_placeholders=description_placeholders,
) )
async def async_step_reauth_confirm( async def async_step_reauth_confirm(
@ -235,9 +251,10 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle reauth confirmation.""" """Handle reauth confirmation."""
assert self.reauth_entry is not None assert self.reauth_entry is not None
errors: dict[str, str] | None = {} errors: dict[str, str] | None = {}
description_placeholders: dict[str, str] = {}
if user_input is not None: if user_input is not None:
entry_data = self.reauth_entry.data entry_data = self.reauth_entry.data
errors, _ = await self._async_try_connect( errors, _, description_placeholders = await self._async_try_connect(
{CONF_IP_ADDRESS: entry_data[CONF_IP_ADDRESS], **user_input} {CONF_IP_ADDRESS: entry_data[CONF_IP_ADDRESS], **user_input}
) )
if not errors: if not errors:
@ -251,6 +268,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
step_id="reauth_confirm", step_id="reauth_confirm",
data_schema=vol.Schema({vol.Optional(CONF_PASSWORD): str}), data_schema=vol.Schema({vol.Optional(CONF_PASSWORD): str}),
errors=errors, errors=errors,
description_placeholders=description_placeholders,
) )
async def async_step_reauth(self, entry_data: Mapping[str, Any]) -> FlowResult: async def async_step_reauth(self, entry_data: Mapping[str, Any]) -> FlowResult:

1
homeassistant/components/powerwall/const.py
View file

@ -7,7 +7,6 @@ POWERWALL_BASE_INFO: Final = "base_info"
POWERWALL_COORDINATOR: Final = "coordinator" POWERWALL_COORDINATOR: Final = "coordinator"
POWERWALL_API: Final = "api_instance" POWERWALL_API: Final = "api_instance"
POWERWALL_API_CHANGED: Final = "api_changed" POWERWALL_API_CHANGED: Final = "api_changed"
POWERWALL_HTTP_SESSION: Final = "http_session"
UPDATE_INTERVAL = 30 UPDATE_INTERVAL = 30

2
homeassistant/components/powerwall/manifest.json
View file

@ -14,5 +14,5 @@
"documentation": "https://www.home-assistant.io/integrations/powerwall", "documentation": "https://www.home-assistant.io/integrations/powerwall",
"iot_class": "local_polling", "iot_class": "local_polling",
"loggers": ["tesla_powerwall"], "loggers": ["tesla_powerwall"],
"requirements": ["tesla-powerwall==0.3.19"] "requirements": ["tesla-powerwall==0.5.0"]
} }

18
homeassistant/components/powerwall/models.py
View file

@ -4,15 +4,14 @@ from __future__ import annotations
from dataclasses import dataclass from dataclasses import dataclass
from typing import TypedDict from typing import TypedDict
from requests import Session
from tesla_powerwall import ( from tesla_powerwall import (
DeviceType, DeviceType,
GridStatus, GridStatus,
MetersAggregates, MetersAggregatesResponse,
Powerwall, Powerwall,
PowerwallStatus, PowerwallStatusResponse,
SiteInfo, SiteInfoResponse,
SiteMaster, SiteMasterResponse,
) )
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
@ -23,8 +22,8 @@ class PowerwallBaseInfo:
"""Base information for the powerwall integration.""" """Base information for the powerwall integration."""
gateway_din: None | str gateway_din: None | str
site_info: SiteInfo site_info: SiteInfoResponse
status: PowerwallStatus status: PowerwallStatusResponse
device_type: DeviceType device_type: DeviceType
serial_numbers: list[str] serial_numbers: list[str]
url: str url: str
@ -35,8 +34,8 @@ class PowerwallData:
"""Point in time data for the powerwall integration.""" """Point in time data for the powerwall integration."""
charge: float charge: float
site_master: SiteMaster site_master: SiteMasterResponse
meters: MetersAggregates meters: MetersAggregatesResponse
grid_services_active: bool grid_services_active: bool
grid_status: GridStatus grid_status: GridStatus
backup_reserve: float | None backup_reserve: float | None
@ -49,4 +48,3 @@ class PowerwallRuntimeData(TypedDict):
api_instance: Powerwall api_instance: Powerwall
base_info: PowerwallBaseInfo base_info: PowerwallBaseInfo
api_changed: bool api_changed: bool
http_session: Session

41
homeassistant/components/powerwall/sensor.py
View file

@ -3,8 +3,9 @@ from __future__ import annotations
from collections.abc import Callable from collections.abc import Callable
from dataclasses import dataclass from dataclasses import dataclass
from typing import TYPE_CHECKING
from tesla_powerwall import Meter, MeterType from tesla_powerwall import MeterResponse, MeterType
from homeassistant.components.sensor import ( from homeassistant.components.sensor import (
SensorDeviceClass, SensorDeviceClass,
@ -36,7 +37,7 @@ _METER_DIRECTION_IMPORT = "import"
class PowerwallRequiredKeysMixin: class PowerwallRequiredKeysMixin:
"""Mixin for required keys.""" """Mixin for required keys."""
value_fn: Callable[[Meter], float] value_fn: Callable[[MeterResponse], float]
@dataclass(frozen=True) @dataclass(frozen=True)
@ -46,24 +47,24 @@ class PowerwallSensorEntityDescription(
"""Describes Powerwall entity.""" """Describes Powerwall entity."""
def _get_meter_power(meter: Meter) -> float: def _get_meter_power(meter: MeterResponse) -> float:
"""Get the current value in kW.""" """Get the current value in kW."""
return meter.get_power(precision=3) return meter.get_power(precision=3)
def _get_meter_frequency(meter: Meter) -> float: def _get_meter_frequency(meter: MeterResponse) -> float:
"""Get the current value in Hz.""" """Get the current value in Hz."""
return round(meter.frequency, 1) return round(meter.frequency, 1)
def _get_meter_total_current(meter: Meter) -> float: def _get_meter_total_current(meter: MeterResponse) -> float:
"""Get the current value in A.""" """Get the current value in A."""
return meter.get_instant_total_current() return meter.get_instant_total_current()
def _get_meter_average_voltage(meter: Meter) -> float: def _get_meter_average_voltage(meter: MeterResponse) -> float:
"""Get the current value in V.""" """Get the current value in V."""
return round(meter.average_voltage, 1) return round(meter.instant_average_voltage, 1)
POWERWALL_INSTANT_SENSORS = ( POWERWALL_INSTANT_SENSORS = (
@ -171,9 +172,13 @@ class PowerWallEnergySensor(PowerWallEntity, SensorEntity):
self._attr_unique_id = f"{self.base_unique_id}_{meter.value}_{description.key}" self._attr_unique_id = f"{self.base_unique_id}_{meter.value}_{description.key}"
@property @property
def native_value(self) -> float: def native_value(self) -> float | None:
"""Get the current value.""" """Get the current value."""
return self.entity_description.value_fn(self.data.meters.get_meter(self._meter)) meter = self.data.meters.get_meter(self._meter)
if meter is not None:
return self.entity_description.value_fn(meter)
return None
class PowerWallBackupReserveSensor(PowerWallEntity, SensorEntity): class PowerWallBackupReserveSensor(PowerWallEntity, SensorEntity):
@ -224,10 +229,10 @@ class PowerWallEnergyDirectionSensor(PowerWallEntity, SensorEntity):
we do not want to include in statistics and its a we do not want to include in statistics and its a
transient data error. transient data error.
""" """
return super().available and self.native_value != 0 return super().available and self.meter is not None
@property @property
def meter(self) -> Meter: def meter(self) -> MeterResponse | None:
"""Get the meter for the sensor.""" """Get the meter for the sensor."""
return self.data.meters.get_meter(self._meter) return self.data.meters.get_meter(self._meter)
@ -244,9 +249,12 @@ class PowerWallExportSensor(PowerWallEnergyDirectionSensor):
super().__init__(powerwall_data, meter, _METER_DIRECTION_EXPORT) super().__init__(powerwall_data, meter, _METER_DIRECTION_EXPORT)
@property @property
def native_value(self) -> float: def native_value(self) -> float | None:
"""Get the current value in kWh.""" """Get the current value in kWh."""
return self.meter.get_energy_exported() meter = self.meter
if TYPE_CHECKING:
assert meter is not None
return meter.get_energy_exported()
class PowerWallImportSensor(PowerWallEnergyDirectionSensor): class PowerWallImportSensor(PowerWallEnergyDirectionSensor):
@ -261,6 +269,9 @@ class PowerWallImportSensor(PowerWallEnergyDirectionSensor):
super().__init__(powerwall_data, meter, _METER_DIRECTION_IMPORT) super().__init__(powerwall_data, meter, _METER_DIRECTION_IMPORT)
@property @property
def native_value(self) -> float: def native_value(self) -> float | None:
"""Get the current value in kWh.""" """Get the current value in kWh."""
return self.meter.get_energy_imported() meter = self.meter
if TYPE_CHECKING:
assert meter is not None
return meter.get_energy_imported()

8
homeassistant/components/powerwall/strings.json
View file

@ -23,10 +23,10 @@
} }
}, },
"error": { "error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]", "cannot_connect": "A connection error occurred while connecting to the Powerwall: {error}",
"wrong_version": "Your Powerwall uses a software version that is not supported. Please consider upgrading or reporting this issue so it can be resolved.", "wrong_version": "Your Powerwall uses a software version that is not supported. Please consider upgrading or reporting this issue so it can be resolved: {error}",
"unknown": "[%key:common::config_flow::error::unknown%]", "unknown": "An unknown error occurred: {error}",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]" "invalid_auth": "Authentication failed with error: {error}"
}, },
"abort": { "abort": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]", "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",

4
homeassistant/components/powerwall/switch.py
View file

@ -59,9 +59,7 @@ class PowerwallOffGridEnabledEntity(PowerWallEntity, SwitchEntity):
async def _async_set_island_mode(self, island_mode: IslandMode) -> None: async def _async_set_island_mode(self, island_mode: IslandMode) -> None:
"""Toggles off-grid mode using the island_mode argument.""" """Toggles off-grid mode using the island_mode argument."""
try: try:
await self.hass.async_add_executor_job( await self.power_wall.set_island_mode(island_mode)
self.power_wall.set_island_mode, island_mode
)
except PowerwallError as ex: except PowerwallError as ex:
raise HomeAssistantError( raise HomeAssistantError(
f"Setting off-grid operation to {island_mode} failed: {ex}" f"Setting off-grid operation to {island_mode} failed: {ex}"

2
requirements_all.txt
View file

@ -2635,7 +2635,7 @@ temperusb==1.6.1
# tensorflow==2.5.0 # tensorflow==2.5.0
# homeassistant.components.powerwall # homeassistant.components.powerwall
tesla-powerwall==0.3.19 tesla-powerwall==0.5.0
# homeassistant.components.tesla_wall_connector # homeassistant.components.tesla_wall_connector
tesla-wall-connector==1.0.2 tesla-wall-connector==1.0.2

2
requirements_test_all.txt
View file

@ -1991,7 +1991,7 @@ temescal==0.5
temperusb==1.6.1 temperusb==1.6.1
# homeassistant.components.powerwall # homeassistant.components.powerwall
tesla-powerwall==0.3.19 tesla-powerwall==0.5.0
# homeassistant.components.tesla_wall_connector # homeassistant.components.tesla_wall_connector
tesla-wall-connector==1.0.2 tesla-wall-connector==1.0.2

1
tests/components/powerwall/fixtures/meters_empty.json Normal file
View file

@ -0,0 +1 @@
{}

7
tests/components/powerwall/fixtures/sitemaster.json
View file

@ -1 +1,6 @@
{ "connected_to_tesla": true, "running": true, "status": "StatusUp" } {
"connected_to_tesla": true,
"power_supply_mode": false,
"running": true,
"status": "StatusUp"
}

11
tests/components/powerwall/fixtures/status.json
View file

@ -1,7 +1,10 @@
{ {
"start_time": "2020-03-10 11:57:25 +0800", "commission_count": 0,
"up_time_seconds": "217h40m57.470801079s", "device_type": "hec",
"git_hash": "d0e69bde519634961cca04a616d2d4dae80b9f61",
"is_new": false, "is_new": false,
"version": "1.45.1", "start_time": "2020-10-28 20:14:11 +0800",
"git_hash": "13bf684a633175f884079ec79f42997080d90310" "sync_type": "v1",
"up_time_seconds": "17h11m31.214751424s",
"version": "1.50.1 c58c2df3"
} }

89
tests/components/powerwall/mocks.py
View file

@ -1,17 +1,18 @@
"""Mocks for powerwall.""" """Mocks for powerwall."""
import asyncio
import json import json
import os import os
from unittest.mock import MagicMock, Mock from unittest.mock import MagicMock
from tesla_powerwall import ( from tesla_powerwall import (
DeviceType, DeviceType,
GridStatus, GridStatus,
MetersAggregates, MetersAggregatesResponse,
Powerwall, Powerwall,
PowerwallStatus, PowerwallStatusResponse,
SiteInfo, SiteInfoResponse,
SiteMaster, SiteMasterResponse,
) )
from tests.common import load_fixture from tests.common import load_fixture
@ -19,29 +20,31 @@ from tests.common import load_fixture
MOCK_GATEWAY_DIN = "111-0----2-000000000FFA" MOCK_GATEWAY_DIN = "111-0----2-000000000FFA"
async def _mock_powerwall_with_fixtures(hass): async def _mock_powerwall_with_fixtures(hass, empty_meters: bool = False) -> MagicMock:
"""Mock data used to build powerwall state.""" """Mock data used to build powerwall state."""
meters = await _async_load_json_fixture(hass, "meters.json") async with asyncio.TaskGroup() as tg:
sitemaster = await _async_load_json_fixture(hass, "sitemaster.json") meters_file = "meters_empty.json" if empty_meters else "meters.json"
site_info = await _async_load_json_fixture(hass, "site_info.json") meters = tg.create_task(_async_load_json_fixture(hass, meters_file))
status = await _async_load_json_fixture(hass, "status.json") sitemaster = tg.create_task(_async_load_json_fixture(hass, "sitemaster.json"))
device_type = await _async_load_json_fixture(hass, "device_type.json") site_info = tg.create_task(_async_load_json_fixture(hass, "site_info.json"))
status = tg.create_task(_async_load_json_fixture(hass, "status.json"))
device_type = tg.create_task(_async_load_json_fixture(hass, "device_type.json"))
return _mock_powerwall_return_value( return await _mock_powerwall_return_value(
site_info=SiteInfo(site_info), site_info=SiteInfoResponse.from_dict(site_info.result()),
charge=47.34587394586, charge=47.34587394586,
sitemaster=SiteMaster(sitemaster), sitemaster=SiteMasterResponse.from_dict(sitemaster.result()),
meters=MetersAggregates(meters), meters=MetersAggregatesResponse.from_dict(meters.result()),
grid_services_active=True, grid_services_active=True,
grid_status=GridStatus.CONNECTED, grid_status=GridStatus.CONNECTED,
status=PowerwallStatus(status), status=PowerwallStatusResponse.from_dict(status.result()),
device_type=DeviceType(device_type["device_type"]), device_type=DeviceType(device_type.result()["device_type"]),
serial_numbers=["TG0123456789AB", "TG9876543210BA"], serial_numbers=["TG0123456789AB", "TG9876543210BA"],
backup_reserve_percentage=15.0, backup_reserve_percentage=15.0,
) )
def _mock_powerwall_return_value( async def _mock_powerwall_return_value(
site_info=None, site_info=None,
charge=None, charge=None,
sitemaster=None, sitemaster=None,
@ -53,38 +56,46 @@ def _mock_powerwall_return_value(
serial_numbers=None, serial_numbers=None,
backup_reserve_percentage=None, backup_reserve_percentage=None,
): ):
powerwall_mock = MagicMock(Powerwall("1.2.3.4")) powerwall_mock = MagicMock(Powerwall)
powerwall_mock.get_site_info = Mock(return_value=site_info) powerwall_mock.__aenter__.return_value = powerwall_mock
powerwall_mock.get_charge = Mock(return_value=charge)
powerwall_mock.get_sitemaster = Mock(return_value=sitemaster) powerwall_mock.get_site_info.return_value = site_info
powerwall_mock.get_meters = Mock(return_value=meters) powerwall_mock.get_charge.return_value = charge
powerwall_mock.get_grid_status = Mock(return_value=grid_status) powerwall_mock.get_sitemaster.return_value = sitemaster
powerwall_mock.get_status = Mock(return_value=status) powerwall_mock.get_meters.return_value = meters
powerwall_mock.get_device_type = Mock(return_value=device_type) powerwall_mock.get_grid_status.return_value = grid_status
powerwall_mock.get_serial_numbers = Mock(return_value=serial_numbers) powerwall_mock.get_status.return_value = status
powerwall_mock.get_backup_reserve_percentage = Mock( powerwall_mock.get_device_type.return_value = device_type
return_value=backup_reserve_percentage powerwall_mock.get_serial_numbers.return_value = serial_numbers
powerwall_mock.get_backup_reserve_percentage.return_value = (
backup_reserve_percentage
) )
powerwall_mock.is_grid_services_active = Mock(return_value=grid_services_active) powerwall_mock.is_grid_services_active.return_value = grid_services_active
powerwall_mock.get_gateway_din.return_value = MOCK_GATEWAY_DIN
return powerwall_mock return powerwall_mock
async def _mock_powerwall_site_name(hass, site_name): async def _mock_powerwall_site_name(hass, site_name):
powerwall_mock = MagicMock(Powerwall("1.2.3.4")) powerwall_mock = MagicMock(Powerwall)
powerwall_mock.__aenter__.return_value = powerwall_mock
site_info_resp = SiteInfo(await _async_load_json_fixture(hass, "site_info.json")) site_info_resp = SiteInfoResponse.from_dict(
# Sets site_info_resp.site_name to return site_name await _async_load_json_fixture(hass, "site_info.json")
site_info_resp.response["site_name"] = site_name )
powerwall_mock.get_site_info = Mock(return_value=site_info_resp) site_info_resp._raw["site_name"] = site_name
powerwall_mock.get_gateway_din = Mock(return_value=MOCK_GATEWAY_DIN) site_info_resp.site_name = site_name
powerwall_mock.get_site_info.return_value = site_info_resp
powerwall_mock.get_gateway_din.return_value = MOCK_GATEWAY_DIN
return powerwall_mock return powerwall_mock
def _mock_powerwall_side_effect(site_info=None): async def _mock_powerwall_side_effect(site_info=None):
powerwall_mock = MagicMock(Powerwall("1.2.3.4")) powerwall_mock = MagicMock(Powerwall)
powerwall_mock.get_site_info = Mock(side_effect=site_info) powerwall_mock.__aenter__.return_value = powerwall_mock
powerwall_mock.get_site_info.side_effect = site_info
return powerwall_mock return powerwall_mock

22
tests/components/powerwall/test_binary_sensor.py
View file

@ -2,7 +2,7 @@
from unittest.mock import patch from unittest.mock import patch
from homeassistant.components.powerwall.const import DOMAIN from homeassistant.components.powerwall.const import DOMAIN
from homeassistant.const import CONF_IP_ADDRESS, STATE_ON from homeassistant.const import CONF_IP_ADDRESS, STATE_ON, STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from .mocks import _mock_powerwall_with_fixtures from .mocks import _mock_powerwall_with_fixtures
@ -75,3 +75,23 @@ async def test_sensors(hass: HomeAssistant) -> None:
# Only test for a subset of attributes in case # Only test for a subset of attributes in case
# HA changes the implementation and a new one appears # HA changes the implementation and a new one appears
assert all(item in state.attributes.items() for item in expected_attributes.items()) assert all(item in state.attributes.items() for item in expected_attributes.items())
async def test_sensors_with_empty_meters(hass: HomeAssistant) -> None:
"""Test creation of the binary sensors with empty meters."""
mock_powerwall = await _mock_powerwall_with_fixtures(hass, empty_meters=True)
config_entry = MockConfigEntry(domain=DOMAIN, data={CONF_IP_ADDRESS: "1.2.3.4"})
config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.powerwall.config_flow.Powerwall",
return_value=mock_powerwall,
), patch(
"homeassistant.components.powerwall.Powerwall", return_value=mock_powerwall
):
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
state = hass.states.get("binary_sensor.mysite_charging")
assert state.state == STATE_UNAVAILABLE

72
tests/components/powerwall/test_config_flow.py
View file

@ -1,5 +1,7 @@
"""Test the Powerwall config flow.""" """Test the Powerwall config flow."""
import asyncio
from datetime import timedelta
from unittest.mock import MagicMock, patch from unittest.mock import MagicMock, patch
from tesla_powerwall import ( from tesla_powerwall import (
@ -14,6 +16,7 @@ from homeassistant.components.powerwall.const import DOMAIN
from homeassistant.const import CONF_IP_ADDRESS, CONF_PASSWORD from homeassistant.const import CONF_IP_ADDRESS, CONF_PASSWORD
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType from homeassistant.data_entry_flow import FlowResultType
import homeassistant.util.dt as dt_util
from .mocks import ( from .mocks import (
MOCK_GATEWAY_DIN, MOCK_GATEWAY_DIN,
@ -22,7 +25,7 @@ from .mocks import (
_mock_powerwall_with_fixtures, _mock_powerwall_with_fixtures,
) )
from tests.common import MockConfigEntry from tests.common import MockConfigEntry, async_fire_time_changed
VALID_CONFIG = {CONF_IP_ADDRESS: "1.2.3.4", CONF_PASSWORD: "00GGX"} VALID_CONFIG = {CONF_IP_ADDRESS: "1.2.3.4", CONF_PASSWORD: "00GGX"}
@ -36,7 +39,7 @@ async def test_form_source_user(hass: HomeAssistant) -> None:
assert result["type"] == "form" assert result["type"] == "form"
assert result["errors"] == {} assert result["errors"] == {}
mock_powerwall = await _mock_powerwall_site_name(hass, "My site") mock_powerwall = await _mock_powerwall_site_name(hass, "MySite")
with patch( with patch(
"homeassistant.components.powerwall.config_flow.Powerwall", "homeassistant.components.powerwall.config_flow.Powerwall",
@ -52,7 +55,7 @@ async def test_form_source_user(hass: HomeAssistant) -> None:
await hass.async_block_till_done() await hass.async_block_till_done()
assert result2["type"] == "create_entry" assert result2["type"] == "create_entry"
assert result2["title"] == "My site" assert result2["title"] == "MySite"
assert result2["data"] == VALID_CONFIG assert result2["data"] == VALID_CONFIG
assert len(mock_setup_entry.mock_calls) == 1 assert len(mock_setup_entry.mock_calls) == 1
@ -63,7 +66,9 @@ async def test_form_cannot_connect(hass: HomeAssistant) -> None:
DOMAIN, context={"source": config_entries.SOURCE_USER} DOMAIN, context={"source": config_entries.SOURCE_USER}
) )
mock_powerwall = _mock_powerwall_side_effect(site_info=PowerwallUnreachableError) mock_powerwall = await _mock_powerwall_side_effect(
site_info=PowerwallUnreachableError
)
with patch( with patch(
"homeassistant.components.powerwall.config_flow.Powerwall", "homeassistant.components.powerwall.config_flow.Powerwall",
@ -84,7 +89,9 @@ async def test_invalid_auth(hass: HomeAssistant) -> None:
DOMAIN, context={"source": config_entries.SOURCE_USER} DOMAIN, context={"source": config_entries.SOURCE_USER}
) )
mock_powerwall = _mock_powerwall_side_effect(site_info=AccessDeniedError("any")) mock_powerwall = await _mock_powerwall_side_effect(
site_info=AccessDeniedError("any")
)
with patch( with patch(
"homeassistant.components.powerwall.config_flow.Powerwall", "homeassistant.components.powerwall.config_flow.Powerwall",
@ -105,7 +112,7 @@ async def test_form_unknown_exeption(hass: HomeAssistant) -> None:
DOMAIN, context={"source": config_entries.SOURCE_USER} DOMAIN, context={"source": config_entries.SOURCE_USER}
) )
mock_powerwall = _mock_powerwall_side_effect(site_info=ValueError) mock_powerwall = await _mock_powerwall_side_effect(site_info=ValueError)
with patch( with patch(
"homeassistant.components.powerwall.config_flow.Powerwall", "homeassistant.components.powerwall.config_flow.Powerwall",
@ -125,7 +132,7 @@ async def test_form_wrong_version(hass: HomeAssistant) -> None:
DOMAIN, context={"source": config_entries.SOURCE_USER} DOMAIN, context={"source": config_entries.SOURCE_USER}
) )
mock_powerwall = _mock_powerwall_side_effect( mock_powerwall = await _mock_powerwall_side_effect(
site_info=MissingAttributeError({}, "") site_info=MissingAttributeError({}, "")
) )
@ -286,7 +293,9 @@ async def test_dhcp_discovery_auto_configure(hass: HomeAssistant) -> None:
async def test_dhcp_discovery_cannot_connect(hass: HomeAssistant) -> None: async def test_dhcp_discovery_cannot_connect(hass: HomeAssistant) -> None:
"""Test we can process the discovery from dhcp and we cannot connect.""" """Test we can process the discovery from dhcp and we cannot connect."""
mock_powerwall = _mock_powerwall_side_effect(site_info=PowerwallUnreachableError) mock_powerwall = await _mock_powerwall_side_effect(
site_info=PowerwallUnreachableError
)
with patch( with patch(
"homeassistant.components.powerwall.config_flow.Powerwall", "homeassistant.components.powerwall.config_flow.Powerwall",
@ -354,6 +363,7 @@ async def test_dhcp_discovery_update_ip_address(hass: HomeAssistant) -> None:
) )
entry.add_to_hass(hass) entry.add_to_hass(hass)
mock_powerwall = MagicMock(login=MagicMock(side_effect=PowerwallUnreachableError)) mock_powerwall = MagicMock(login=MagicMock(side_effect=PowerwallUnreachableError))
mock_powerwall.__aenter__.return_value = mock_powerwall
with patch( with patch(
"homeassistant.components.powerwall.config_flow.Powerwall", "homeassistant.components.powerwall.config_flow.Powerwall",
@ -547,3 +557,49 @@ async def test_discovered_wifi_does_not_update_ip_if_is_still_online(
assert result["type"] == FlowResultType.ABORT assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "already_configured" assert result["reason"] == "already_configured"
assert entry.data[CONF_IP_ADDRESS] == "1.2.3.4" assert entry.data[CONF_IP_ADDRESS] == "1.2.3.4"
async def test_discovered_wifi_does_not_update_ip_online_but_access_denied(
hass: HomeAssistant,
) -> None:
"""Test a discovery does not update the ip unless the powerwall at the old ip is offline."""
entry = MockConfigEntry(
domain=DOMAIN,
data=VALID_CONFIG,
unique_id=MOCK_GATEWAY_DIN,
)
entry.add_to_hass(hass)
mock_powerwall = await _mock_powerwall_with_fixtures(hass)
mock_powerwall_no_access = await _mock_powerwall_with_fixtures(hass)
mock_powerwall_no_access.login.side_effect = AccessDeniedError("any")
with patch(
"homeassistant.components.powerwall.config_flow.Powerwall",
return_value=mock_powerwall_no_access,
), patch(
"homeassistant.components.powerwall.Powerwall", return_value=mock_powerwall
):
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
# Now mock the powerwall to be offline to force
# the discovery flow to probe to see if its online
# which will result in an access denied error, which
# means its still online and we should not update the ip
mock_powerwall.get_meters.side_effect = asyncio.TimeoutError
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=60))
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_DHCP},
data=dhcp.DhcpServiceInfo(
ip="1.2.3.5",
macaddress="AA:BB:CC:DD:EE:FF",
hostname=MOCK_GATEWAY_DIN.lower(),
),
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "already_configured"
assert entry.data[CONF_IP_ADDRESS] == "1.2.3.4"

17
tests/components/powerwall/test_init.py
View file

@ -1,7 +1,7 @@
"""Tests for the PowerwallDataManager.""" """Tests for the PowerwallDataManager."""
import datetime import datetime
from unittest.mock import MagicMock, patch from unittest.mock import patch
from tesla_powerwall import AccessDeniedError, LoginResponse from tesla_powerwall import AccessDeniedError, LoginResponse
@ -24,12 +24,17 @@ async def test_update_data_reauthenticate_on_access_denied(hass: HomeAssistant)
# 1. login success on entry setup # 1. login success on entry setup
# 2. login success after reauthentication # 2. login success after reauthentication
# 3. login failure after reauthentication # 3. login failure after reauthentication
mock_powerwall.login = MagicMock(name="login", return_value=LoginResponse({})) mock_powerwall.login.return_value = LoginResponse.from_dict(
mock_powerwall.get_charge = MagicMock(name="get_charge", return_value=90.0) {
mock_powerwall.is_authenticated = MagicMock( "firstname": "firstname",
name="is_authenticated", return_value=True "lastname": "lastname",
"token": "token",
"roles": [],
"loginTime": "loginTime",
}
) )
mock_powerwall.logout = MagicMock(name="logout") mock_powerwall.get_charge.return_value = 90.0
mock_powerwall.is_authenticated.return_value = True
config_entry = MockConfigEntry( config_entry = MockConfigEntry(
domain=DOMAIN, data={CONF_IP_ADDRESS: "1.2.3.4", CONF_PASSWORD: "password"} domain=DOMAIN, data={CONF_IP_ADDRESS: "1.2.3.4", CONF_PASSWORD: "password"}

41
tests/components/powerwall/test_sensor.py
View file

@ -1,6 +1,8 @@
"""The sensor tests for the powerwall platform.""" """The sensor tests for the powerwall platform."""
from datetime import timedelta
from unittest.mock import Mock, patch from unittest.mock import Mock, patch
from tesla_powerwall import MetersAggregatesResponse
from tesla_powerwall.error import MissingAttributeError from tesla_powerwall.error import MissingAttributeError
from homeassistant.components.powerwall.const import DOMAIN from homeassistant.components.powerwall.const import DOMAIN
@ -11,13 +13,15 @@ from homeassistant.const import (
ATTR_UNIT_OF_MEASUREMENT, ATTR_UNIT_OF_MEASUREMENT,
CONF_IP_ADDRESS, CONF_IP_ADDRESS,
PERCENTAGE, PERCENTAGE,
STATE_UNKNOWN,
) )
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr from homeassistant.helpers import device_registry as dr
import homeassistant.util.dt as dt_util
from .mocks import _mock_powerwall_with_fixtures from .mocks import _mock_powerwall_with_fixtures
from tests.common import MockConfigEntry from tests.common import MockConfigEntry, async_fire_time_changed
async def test_sensors( async def test_sensors(
@ -43,7 +47,7 @@ async def test_sensors(
identifiers={("powerwall", "TG0123456789AB_TG9876543210BA")}, identifiers={("powerwall", "TG0123456789AB_TG9876543210BA")},
) )
assert reg_device.model == "PowerWall 2 (GW1)" assert reg_device.model == "PowerWall 2 (GW1)"
assert reg_device.sw_version == "1.45.1" assert reg_device.sw_version == "1.50.1 c58c2df3"
assert reg_device.manufacturer == "Tesla" assert reg_device.manufacturer == "Tesla"
assert reg_device.name == "MySite" assert reg_device.name == "MySite"
@ -118,13 +122,23 @@ async def test_sensors(
for key, value in expected_attributes.items(): for key, value in expected_attributes.items():
assert state.attributes[key] == value assert state.attributes[key] == value
mock_powerwall.get_meters.return_value = MetersAggregatesResponse.from_dict({})
mock_powerwall.get_backup_reserve_percentage.return_value = None
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=60))
await hass.async_block_till_done()
assert hass.states.get("sensor.mysite_load_power").state == STATE_UNKNOWN
assert hass.states.get("sensor.mysite_load_frequency").state == STATE_UNKNOWN
assert hass.states.get("sensor.mysite_backup_reserve").state == STATE_UNKNOWN
async def test_sensor_backup_reserve_unavailable(hass: HomeAssistant) -> None: async def test_sensor_backup_reserve_unavailable(hass: HomeAssistant) -> None:
"""Confirm that backup reserve sensor is not added if data is unavailable from the device.""" """Confirm that backup reserve sensor is not added if data is unavailable from the device."""
mock_powerwall = await _mock_powerwall_with_fixtures(hass) mock_powerwall = await _mock_powerwall_with_fixtures(hass)
mock_powerwall.get_backup_reserve_percentage = Mock( mock_powerwall.get_backup_reserve_percentage.side_effect = MissingAttributeError(
side_effect=MissingAttributeError(Mock(), "backup_reserve_percent", "operation") Mock(), "backup_reserve_percent", "operation"
) )
config_entry = MockConfigEntry(domain=DOMAIN, data={CONF_IP_ADDRESS: "1.2.3.4"}) config_entry = MockConfigEntry(domain=DOMAIN, data={CONF_IP_ADDRESS: "1.2.3.4"})
@ -140,3 +154,22 @@ async def test_sensor_backup_reserve_unavailable(hass: HomeAssistant) -> None:
state = hass.states.get("sensor.powerwall_backup_reserve") state = hass.states.get("sensor.powerwall_backup_reserve")
assert state is None assert state is None
async def test_sensors_with_empty_meters(hass: HomeAssistant) -> None:
"""Test creation of the sensors with empty meters."""
mock_powerwall = await _mock_powerwall_with_fixtures(hass, empty_meters=True)
config_entry = MockConfigEntry(domain=DOMAIN, data={CONF_IP_ADDRESS: "1.2.3.4"})
config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.powerwall.config_flow.Powerwall",
return_value=mock_powerwall,
), patch(
"homeassistant.components.powerwall.Powerwall", return_value=mock_powerwall
):
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert hass.states.get("sensor.mysite_solar_power") is None

14
tests/components/powerwall/test_switch.py
View file

@ -1,5 +1,5 @@
"""Test for Powerwall off-grid switch.""" """Test for Powerwall off-grid switch."""
from unittest.mock import Mock, patch from unittest.mock import patch
import pytest import pytest
from tesla_powerwall import GridStatus, PowerwallError from tesla_powerwall import GridStatus, PowerwallError
@ -43,7 +43,7 @@ async def test_entity_registry(
) -> None: ) -> None:
"""Test powerwall off-grid switch device.""" """Test powerwall off-grid switch device."""
mock_powerwall.get_grid_status = Mock(return_value=GridStatus.CONNECTED) mock_powerwall.get_grid_status.return_value = GridStatus.CONNECTED
assert ENTITY_ID in entity_registry.entities assert ENTITY_ID in entity_registry.entities
@ -51,7 +51,7 @@ async def test_entity_registry(
async def test_initial(hass: HomeAssistant, mock_powerwall) -> None: async def test_initial(hass: HomeAssistant, mock_powerwall) -> None:
"""Test initial grid status without off grid switch selected.""" """Test initial grid status without off grid switch selected."""
mock_powerwall.get_grid_status = Mock(return_value=GridStatus.CONNECTED) mock_powerwall.get_grid_status.return_value = GridStatus.CONNECTED
state = hass.states.get(ENTITY_ID) state = hass.states.get(ENTITY_ID)
assert state.state == STATE_OFF assert state.state == STATE_OFF
@ -60,7 +60,7 @@ async def test_initial(hass: HomeAssistant, mock_powerwall) -> None:
async def test_on(hass: HomeAssistant, mock_powerwall) -> None: async def test_on(hass: HomeAssistant, mock_powerwall) -> None:
"""Test state once offgrid switch has been turned on.""" """Test state once offgrid switch has been turned on."""
mock_powerwall.get_grid_status = Mock(return_value=GridStatus.ISLANDED) mock_powerwall.get_grid_status.return_value = GridStatus.ISLANDED
await hass.services.async_call( await hass.services.async_call(
SWITCH_DOMAIN, SWITCH_DOMAIN,
@ -76,7 +76,7 @@ async def test_on(hass: HomeAssistant, mock_powerwall) -> None:
async def test_off(hass: HomeAssistant, mock_powerwall) -> None: async def test_off(hass: HomeAssistant, mock_powerwall) -> None:
"""Test state once offgrid switch has been turned off.""" """Test state once offgrid switch has been turned off."""
mock_powerwall.get_grid_status = Mock(return_value=GridStatus.CONNECTED) mock_powerwall.get_grid_status.return_value = GridStatus.CONNECTED
await hass.services.async_call( await hass.services.async_call(
SWITCH_DOMAIN, SWITCH_DOMAIN,
@ -95,9 +95,7 @@ async def test_exception_on_powerwall_error(
"""Ensure that an exception in the tesla_powerwall library causes a HomeAssistantError.""" """Ensure that an exception in the tesla_powerwall library causes a HomeAssistantError."""
with pytest.raises(HomeAssistantError, match="Setting off-grid operation to"): with pytest.raises(HomeAssistantError, match="Setting off-grid operation to"):
mock_powerwall.set_island_mode = Mock( mock_powerwall.set_island_mode.side_effect = PowerwallError("Mock exception")
side_effect=PowerwallError("Mock exception")
)
await hass.services.async_call( await hass.services.async_call(
SWITCH_DOMAIN, SWITCH_DOMAIN,