starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-09-24 23:41:48 -05:00
Code Issues Packages Projects Releases Wiki Activity

Add local Powerwall control for Teslemetry energy sites (#176969)

Browse Source 
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
Brett Adams
2026-09-04 18:56:33 +02:00
committed by GitHub
co-authored by Copilot Autofix powered by AI
parent 295f19789f
commit 568136f840
11 changed files with 1867 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/teslemetry/__init__.py
+178 -6
View File
@@ -3,9 +3,11 @@
import asyncio
from collections.abc import Callable
from functools import partial
from pathlib import Path
from typing import Any, Final, cast
from aiohttp import ClientError
from aiopowerwall import PowerwallClient, PowerwallEnergySite, PowerwallError
from tesla_fleet_api.const import Scope
from tesla_fleet_api.exceptions import (
Forbidden,
@@ -14,6 +16,7 @@ from tesla_fleet_api.exceptions import (
SubscriptionRequired,
TeslaFleetError,
)
from tesla_fleet_api.tesla import EnergySiteRouter
from tesla_fleet_api.teslemetry import EnergySite, Teslemetry
from teslemetry_stream import TeslemetryStream
from teslemetry_stream.const import SseTopic
@@ -23,7 +26,7 @@ from homeassistant.components.application_credentials import (
async_import_client_credential,
)
from homeassistant.config_entries import ConfigEntry, ConfigEntryState
from homeassistant.const import CONF_ACCESS_TOKEN, Platform
from homeassistant.const import CONF_ACCESS_TOKEN, CONF_HOST, CONF_PASSWORD, Platform
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import (
ConfigEntryAuthFailed,
@@ -45,7 +48,15 @@ from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.typing import ConfigType
from homeassistant.helpers.update_coordinator import UpdateFailed
from .const import CLIENT_ID, DOMAIN, LOGGER, VEHICLE_ISSUE_LEARN_MORE
from .const import (
CLIENT_ID,
DOMAIN,
LOGGER,
POWERWALL_KEY_FILE,
RSA_PARENT_KEY,
SUBENTRY_TYPE_ENERGY_SITE,
VEHICLE_ISSUE_LEARN_MORE,
)
from .coordinator import (
TeslemetryEnergyHistoryCoordinator,
TeslemetryEnergySiteInfoCoordinator,
@@ -260,6 +271,134 @@ def _setup_vehicle_repairs(
)
def _find_energy_subentry_id(entry: TeslemetryConfigEntry, site_id: int) -> str | None:
"""Return the user-added local-control subentry id bound to site_id, if any."""
return next(
(
subentry.subentry_id
for subentry in entry.subentries.values()
if subentry.subentry_type == SUBENTRY_TYPE_ENERGY_SITE
and subentry.unique_id == str(site_id)
),
None,
)
def _remove_stale_subentries(
hass: HomeAssistant,
entry: TeslemetryConfigEntry,
subentry_type: str,
current_subentry_ids: set[str],
) -> None:
"""Remove subentries of the given type with no matching product."""
for subentry in list(entry.subentries.values()):
if (
subentry.subentry_type == subentry_type
and subentry.subentry_id not in current_subentry_ids
):
LOGGER.debug("Removing stale subentry %s", subentry.subentry_id)
hass.config_entries.async_remove_subentry(entry, subentry.subentry_id)
def _prune_energy_subentries(
hass: HomeAssistant,
entry: TeslemetryConfigEntry,
scopes: list[Scope],
products: list[dict[str, Any]],
) -> None:
"""Remove energy-site subentries whose site is no longer on the account."""
if Scope.ENERGY_DEVICE_DATA not in scopes:
return
# Prune on the raw product list; access:false can be transient, not a removal.
product_site_ids = {
str(product["energy_site_id"])
for product in products
if "energy_site_id" in product
}
_remove_stale_subentries(
hass,
entry,
SUBENTRY_TYPE_ENERGY_SITE,
{
subentry.subentry_id
for subentry in entry.subentries.values()
if subentry.subentry_type == SUBENTRY_TYPE_ENERGY_SITE
and subentry.unique_id in product_site_ids
},
)
async def _async_get_rsa_key_pem(hass: HomeAssistant) -> bytes:
"""Return the integration's RSA private key PEM, generating it if needed."""
pem: bytes | None = hass.data.get(RSA_PARENT_KEY)
if pem is None:
path = hass.config.path(POWERWALL_KEY_FILE)
await Teslemetry(
session=async_get_clientsession(hass), access_token=""
).get_rsa_private_key(path)
pem = await hass.async_add_executor_job(Path(path).read_bytes)
hass.data[RSA_PARENT_KEY] = pem
return pem
# aiopowerwall raises PowerwallError; key I/O and parsing raise OSError/ValueError.
_LOCAL_CONTROL_ERRORS: Final = (OSError, ValueError, PowerwallError)
async def _async_resolve_local_control(
hass: HomeAssistant,
entry: TeslemetryConfigEntry,
battery: bool,
site_id: int,
cloud_energy_site: EnergySite,
) -> tuple[bool, str | None, EnergySite | EnergySiteRouter]:
"""Resolve opt-in local control for an energy site."""
# Only a battery/Powerwall gateway can pair for local (TEDAPI) control.
if not battery:
return False, None, cloud_energy_site
subentry_id = _find_energy_subentry_id(entry, site_id)
if subentry_id is None:
return True, None, cloud_energy_site
# A local-gateway failure for one site must not tear down the integration.
try:
api = await _async_resolve_energy_site_api(
hass, entry, subentry_id, cloud_energy_site
)
except _LOCAL_CONTROL_ERRORS:
LOGGER.warning(
"Failed to set up local control for energy site %s; "
"falling back to cloud control",
site_id,
exc_info=True,
)
return True, subentry_id, cloud_energy_site
return True, subentry_id, api
async def _async_resolve_energy_site_api(
hass: HomeAssistant,
entry: TeslemetryConfigEntry,
subentry_id: str,
cloud_energy_site: EnergySite,
) -> EnergySite | EnergySiteRouter:
"""Return the API an energy site's platforms should call."""
data = entry.subentries[subentry_id].data
host = data.get(CONF_HOST)
password = data.get(CONF_PASSWORD)
if not host or not password:
return cloud_energy_site
key_pem = await _async_get_rsa_key_pem(hass)
powerwall_client = PowerwallClient(
host=host,
gateway_password=password,
rsa_private_key_pem=key_pem,
session=async_get_clientsession(hass),
)
local_energy_site = PowerwallEnergySite(powerwall_client)
return EnergySiteRouter(local_energy_site, cloud_energy_site)
async def async_setup_entry(hass: HomeAssistant, entry: TeslemetryConfigEntry) -> bool:
"""Set up Teslemetry config."""
@@ -406,9 +545,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: TeslemetryConfigEntry) -
):
site_id = product["energy_site_id"]
powerwall = (
product["components"]["battery"] or product["components"]["solar"]
)
battery = product["components"]["battery"]
powerwall = battery or product["components"]["solar"]
wall_connector = "wall_connectors" in product["components"]
if not powerwall and not wall_connector:
LOGGER.debug(
@@ -449,14 +587,25 @@ async def async_setup_entry(hass: HomeAssistant, entry: TeslemetryConfigEntry) -
site_id,
powerwall,
)
(
can_local_control,
subentry_id,
energy_site_api,
) = await _async_resolve_local_control(
hass, entry, bool(battery), site_id, energy_site
)
energysites.append(
TeslemetryEnergyData(
api=energy_site,
api=energy_site_api,
live_coordinator=live_coordinator,
info_coordinator=info_coordinator,
history_coordinator=history_coordinator,
id=site_id,
device=device,
can_local_control=can_local_control,
subentry_id=subentry_id,
)
)
@@ -488,6 +637,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: TeslemetryConfigEntry) -
LOGGER.debug("Removing stale device %s", device_entry.id)
device_registry.async_remove_device(device_entry.id)
_prune_energy_subentries(hass, entry, scopes, products)
entry.runtime_data = TeslemetryData(
vehicles=vehicles,
energysites=energysites,
@@ -497,6 +648,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: TeslemetryConfigEntry) -
)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
_setup_subentry_change_reload(hass, entry)
_setup_dynamic_discovery(
hass,
entry,
@@ -529,6 +682,25 @@ async def async_setup_entry(hass: HomeAssistant, entry: TeslemetryConfigEntry) -
return True
def _setup_subentry_change_reload(
hass: HomeAssistant, entry: TeslemetryConfigEntry
) -> None:
"""Reload the entry when a local-energy-site subentry is added or removed."""
known = set(entry.subentries)
async def _handle_update(
hass: HomeAssistant, updated_entry: TeslemetryConfigEntry
) -> None:
nonlocal known
current = set(updated_entry.subentries)
if known.symmetric_difference(current):
hass.config_entries.async_schedule_reload(updated_entry.entry_id)
# Refresh known so a later update does not re-fire on this same change.
known = current
entry.async_on_unload(entry.add_update_listener(_handle_update))
def create_handle_energy_stream_connection(
energysites: list[TeslemetryEnergyData],
) -> Callable[[bool], None]:
homeassistant/components/teslemetry/config_flow.py
+307 -10
View File
@@ -2,15 +2,29 @@
from collections.abc import Mapping
import logging
from typing import Any, override
from pathlib import Path
from typing import TYPE_CHECKING, Any, cast, override
from aiohttp import ClientConnectionError
from aiohttp import ClientError
from aiopowerwall import (
DEFAULT_GATEWAY_HOST,
PowerwallAuthenticationError,
PowerwallClient,
PowerwallError,
)
from tesla_fleet_api.const import (
AuthorizedClientKeyType,
AuthorizedClientState,
AuthorizedClientType,
)
from tesla_fleet_api.exceptions import (
InvalidToken,
SubscriptionRequired,
TeslaFleetError,
)
from tesla_fleet_api.teslemetry import Teslemetry
from tesla_fleet_api.teslemetry.energysite import AuthorizedClient, TeslemetryEnergySite
import voluptuous as vol
from homeassistant.components.application_credentials import (
ClientCredential,
@@ -19,12 +33,34 @@ from homeassistant.components.application_credentials import (
from homeassistant.config_entries import (
SOURCE_REAUTH,
SOURCE_RECONFIGURE,
ConfigEntry,
ConfigEntryState,
ConfigFlowResult,
ConfigSubentryFlow,
SubentryFlowResult,
)
from homeassistant.const import CONF_HOST, CONF_PASSWORD
from homeassistant.core import callback
from homeassistant.helpers import config_entry_oauth2_flow
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import CLIENT_ID, DOMAIN, LOGGER
from . import TeslemetryConfigEntry
from .const import (
CLIENT_ID,
CONF_SITE_ID,
DOMAIN,
LOGGER,
POWERWALL_KEY_FILE,
SUBENTRY_TYPE_ENERGY_SITE,
)
class PowerwallLookupError(Exception):
"""Signal that the authorized-client lookup failed for a non-retryable reason."""
class PowerwallKeyRejectedError(Exception):
"""Signal that the gateway refused a v1r-signed read with our RSA key."""
class OAuth2FlowHandler(
@@ -47,6 +83,15 @@ class OAuth2FlowHandler(
"""Return logger."""
return LOGGER
@classmethod
@callback
@override
def async_get_supported_subentry_types(
cls, config_entry: ConfigEntry
) -> dict[str, type[ConfigSubentryFlow]]:
"""Return the subentry types supported by this integration."""
return {SUBENTRY_TYPE_ENERGY_SITE: EnergySiteSubentryFlowHandler}
@override
async def async_step_user(
self, user_input: dict[str, Any] | None = None
@@ -75,14 +120,10 @@ class OAuth2FlowHandler(
await self.async_set_unique_id(self.uid)
if self.source == SOURCE_REAUTH:
self._abort_if_unique_id_mismatch(reason="reauth_account_mismatch")
return self.async_update_reload_and_abort(
self._get_reauth_entry(), data=data
)
return self._async_apply_token(self._get_reauth_entry(), data)
if self.source == SOURCE_RECONFIGURE:
self._abort_if_unique_id_mismatch(reason="reconfigure_account_mismatch")
return self.async_update_reload_and_abort(
self._get_reconfigure_entry(), data=data
)
return self._async_apply_token(self._get_reconfigure_entry(), data)
self._abort_if_unique_id_configured()
return self.async_create_entry(
@@ -90,6 +131,18 @@ class OAuth2FlowHandler(
data=data,
)
def _async_apply_token(
self, entry: ConfigEntry, data: dict[str, Any]
) -> ConfigFlowResult:
"""Store the refreshed token and reload the entry exactly once."""
if entry.state is not ConfigEntryState.LOADED:
# Unloaded entries have no update listener, so no paired-reload warning.
return self.async_update_reload_and_abort(entry, data=data)
# Reload manually: async_update_reload_and_abort would warn (listener present).
result = self.async_update_and_abort(entry, data=data)
self.hass.config_entries.async_schedule_reload(entry.entry_id)
return result
async def async_test_connection(self, token_data: dict[str, Any]) -> dict[str, str]:
"""Test the connection with OAuth token."""
access_token = token_data["token"]["access_token"]
@@ -105,7 +158,7 @@ class OAuth2FlowHandler(
return {"base": "invalid_access_token"}
except SubscriptionRequired:
return {"base": "subscription_required"}
except ClientConnectionError:
except ClientError:
return {"base": "cannot_connect"}
except TeslaFleetError as e:
LOGGER.error("Teslemetry API error: %s", e)
@@ -137,3 +190,247 @@ class OAuth2FlowHandler(
) -> ConfigFlowResult:
"""Handle reconfiguration."""
return await self.async_step_user()
class EnergySiteSubentryFlowHandler(ConfigSubentryFlow):
"""Pair a local Powerwall gateway for TEDAPI v1r command routing."""
def __init__(self) -> None:
"""Initialize the energy site subentry flow."""
self._energy_site: TeslemetryEnergySite | None = None
self._key_pem: bytes | None = None
self._public_key_der: bytes = b""
self._public_key_b64: str = ""
self._discovered_host: str = ""
self._site_id: int | None = None
self._site_name: str = ""
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> SubentryFlowResult:
"""Let the user opt an account energy site into local Powerwall control."""
entry = cast(TeslemetryConfigEntry, self._get_entry())
# runtime_data exists only while the entry is loaded; core clears it on unload.
if entry.state is not ConfigEntryState.LOADED:
return self.async_abort(reason="entry_not_loaded")
added_site_ids = {
subentry.unique_id
for subentry in entry.subentries.values()
if subentry.subentry_type == SUBENTRY_TYPE_ENERGY_SITE
}
available = {
str(energy_data.id): energy_data
for energy_data in entry.runtime_data.energysites
if energy_data.can_local_control
and str(energy_data.id) not in added_site_ids
}
if not available:
return self.async_abort(reason="no_energy_sites")
if user_input is not None:
energy_data = available[user_input[CONF_SITE_ID]]
self._site_id = energy_data.id
self._site_name = energy_data.device.get("name") or "Energy Site"
# Only unpaired sites are offered, so api is always the cloud EnergySite.
if abort := await self._prepare_energy_site(
cast(TeslemetryEnergySite, energy_data.api)
):
return abort
return await self._async_begin_pairing()
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_SITE_ID): vol.In(
{
site_id: energy_data.device.get("name") or site_id
for site_id, energy_data in available.items()
}
)
}
),
)
async def _prepare_energy_site(
self, energy_site: TeslemetryEnergySite
) -> SubentryFlowResult | None:
"""Discover the gateway address and load the integration's RSA key.
Returns an abort result if the RSA key cannot be loaded, else None.
"""
self._energy_site = energy_site
try:
self._discovered_host = await energy_site.find_gateway_address() or ""
except (ClientError, TeslaFleetError) as err:
LOGGER.debug("Gateway address discovery failed: %s", err)
self._discovered_host = ""
path = self.hass.config.path(POWERWALL_KEY_FILE)
keyholder = Teslemetry(
session=async_get_clientsession(self.hass), access_token=""
)
try:
await keyholder.get_rsa_private_key(path)
self._key_pem = await self.hass.async_add_executor_job(
Path(path).read_bytes
)
except (OSError, ValueError) as err:
LOGGER.debug("RSA key load failed: %s", err)
return self.async_abort(reason="cannot_connect")
self._public_key_der = keyholder.rsa_public_der_pkcs1
self._public_key_b64 = keyholder.rsa_public_der_pkcs1_b64
return None
async def _async_begin_pairing(self) -> SubentryFlowResult:
"""Resume or begin key pairing based on the key's state on the gateway."""
try:
client = await self._find_authorized_client()
except PowerwallLookupError:
return self.async_abort(reason="cannot_connect")
if client is not None:
# Key already registered; do not re-register a pending one (it would reset).
if client.state == AuthorizedClientState.VERIFIED:
return await self.async_step_credentials()
if client.state == AuthorizedClientState.PENDING_VERIFICATION:
return await self.async_step_pair()
if client.state != AuthorizedClientState.PENDING_VERIFICATION_TIMEOUT:
# Unrecognized state is unusable; treat it as a lookup failure.
LOGGER.debug("Unrecognized authorized-client state: %s", client.state)
return self.async_abort(reason="cannot_connect")
# Re-registering resets the expired window (no duplicate); fall through.
if TYPE_CHECKING:
assert self._energy_site is not None
try:
# Not revoked on removal: other consumers may share this key.
LOGGER.info("Powerwall key setup: id=%s", self._energy_site.energy_site_id)
await self._energy_site.add_authorized_client(
self._public_key_der,
description="Home Assistant",
key_type=AuthorizedClientKeyType.RSA,
authorized_client_type=AuthorizedClientType.CUSTOMER_MOBILE_APP,
)
except (ClientError, TeslaFleetError) as err:
LOGGER.error("Add authorized client failed: %s", err)
return self.async_abort(reason="cannot_connect")
return await self.async_step_pair()
async def async_step_pair(
self, user_input: dict[str, Any] | None = None
) -> SubentryFlowResult:
"""Check once whether the pending key has been approved on the gateway."""
if TYPE_CHECKING:
assert self._energy_site is not None
if user_input is None:
return self.async_show_form(step_id="pair")
try:
client = await self._find_authorized_client()
except PowerwallLookupError:
return self.async_show_form(
step_id="pair", errors={"base": "cannot_connect"}
)
if client is None:
return self.async_show_form(
step_id="pair", errors={"base": "key_not_registered"}
)
if client.state == AuthorizedClientState.VERIFIED:
return await self.async_step_credentials()
if client.state == AuthorizedClientState.PENDING_VERIFICATION:
return self.async_show_form(step_id="pair", errors={"base": "key_pending"})
# An unrecognized state reported as pending would trap the user forever.
LOGGER.debug("Unrecognized authorized-client state: %s", client.state)
return self.async_show_form(step_id="pair", errors={"base": "cannot_connect"})
async def _find_authorized_client(self) -> AuthorizedClient | None:
"""Return our RSA key's authorized-client entry on the gateway, or None."""
if TYPE_CHECKING:
assert self._energy_site is not None
try:
result = await self._energy_site.find_authorized_clients()
except (ClientError, TeslaFleetError) as err:
# Raise so a failed lookup is not mistaken for an unregistered key.
LOGGER.debug("find_authorized_clients failed: %s", err)
raise PowerwallLookupError from err
return next(
(
client
for client in result.clients
if client.public_key == self._public_key_b64
),
None,
)
async def _verify_local_gateway(self, host: str, password: str) -> None:
"""Prove the LAN connection and the RSA key against the gateway."""
if TYPE_CHECKING:
assert self._key_pem is not None
assert self._energy_site is not None
async with PowerwallClient(
host=host,
gateway_password=password,
rsa_private_key_pem=self._key_pem,
session=async_get_clientsession(self.hass),
) as client:
await client.connect()
try:
# connect() passed the password, so a failure here is key rejection.
await client.get_status()
except PowerwallAuthenticationError as err:
raise PowerwallKeyRejectedError from err
async def async_step_credentials(
self, user_input: dict[str, Any] | None = None
) -> SubentryFlowResult:
"""Collect the local gateway host/password and verify the LAN connection."""
errors: dict[str, str] = {}
if user_input is not None:
if TYPE_CHECKING:
assert self._energy_site is not None
host = user_input[CONF_HOST].strip()
# The gateway accepts only the last 5 characters of the Wi-Fi password.
password = user_input[CONF_PASSWORD].strip()[-5:]
try:
await self._verify_local_gateway(host, password)
except PowerwallKeyRejectedError as err:
LOGGER.debug("Powerwall rejected the signed read: %s", err.__cause__)
errors["base"] = "key_not_approved"
except PowerwallAuthenticationError:
errors["base"] = "invalid_password"
except PowerwallError as err:
LOGGER.debug("Local Powerwall verify failed: %s", err)
errors["base"] = "cannot_connect"
else:
return self._async_save_credentials(host, password)
return self.async_show_form(
step_id="credentials",
data_schema=vol.Schema(
{
vol.Required(
CONF_HOST,
default=self._discovered_host or DEFAULT_GATEWAY_HOST,
): str,
vol.Required(CONF_PASSWORD): str,
}
),
errors=errors,
)
@callback
def _async_save_credentials(self, host: str, password: str) -> SubentryFlowResult:
"""Persist the verified gateway credentials to a new subentry."""
return self.async_create_entry(
title=self._site_name,
data={
CONF_SITE_ID: self._site_id,
CONF_HOST: host,
CONF_PASSWORD: password,
},
unique_id=str(self._site_id),
)
homeassistant/components/teslemetry/const.py
+5
View File
@@ -12,6 +12,11 @@ AUTHORIZE_URL = "https://teslemetry.com/connect"
TOKEN_URL = "https://api.teslemetry.com/oauth/token"
CLIENT_ID = "homeassistant"
SUBENTRY_TYPE_ENERGY_SITE = "energy_site"
CONF_SITE_ID = "site_id"
POWERWALL_KEY_FILE = "tesla_powerwall.key"
RSA_PARENT_KEY = f"{DOMAIN}_rsa_parent"
ENERGY_HISTORY_FIELDS = [
"solar_energy_exported",
"generator_energy_exported",
homeassistant/components/teslemetry/entity.py
+4 -3
View File
@@ -4,6 +4,7 @@ from abc import abstractmethod
from typing import Any, override
from tesla_fleet_api.const import Scope
from tesla_fleet_api.tesla import EnergySiteRouter
from tesla_fleet_api.teslemetry import EnergySite, Vehicle
from homeassistant.exceptions import ServiceValidationError
@@ -137,7 +138,7 @@ class TeslemetryVehiclePollingEntity(TeslemetryPollingEntity):
class TeslemetryEnergyLiveEntity(TeslemetryPollingEntity):
"""Parent class for Teslemetry Energy Site Live entities."""
api: EnergySite
api: EnergySite | EnergySiteRouter
def __init__(
self,
@@ -158,7 +159,7 @@ class TeslemetryEnergyLiveEntity(TeslemetryPollingEntity):
class TeslemetryEnergyInfoEntity(TeslemetryPollingEntity):
"""Parent class for Teslemetry Energy Site Info Entities."""
api: EnergySite
api: EnergySite | EnergySiteRouter
def __init__(
self,
@@ -197,7 +198,7 @@ class TeslemetryWallConnectorEntity(TeslemetryPollingEntity):
"""Parent class for Teslemetry Wall Connector Entities."""
_attr_has_entity_name = True
api: EnergySite
api: EnergySite | EnergySiteRouter
def __init__(
self,
homeassistant/components/teslemetry/manifest.json
+6 -2
View File
@@ -7,7 +7,11 @@
"documentation": "https://www.home-assistant.io/integrations/teslemetry",
"integration_type": "hub",
"iot_class": "cloud_polling",
"loggers": ["tesla_fleet_api", "teslemetry_stream"],
"loggers": ["aiopowerwall", "tesla_fleet_api", "teslemetry_stream"],
"quality_scale": "platinum",
"requirements": ["tesla-fleet-api==1.12.1", "teslemetry-stream==0.10.0"]
"requirements": [
"aiopowerwall==0.2.0",
"tesla-fleet-api==1.12.1",
"teslemetry-stream==0.10.0"
]
}
homeassistant/components/teslemetry/models.py
+6 -2
View File
@@ -4,6 +4,7 @@ import asyncio
from dataclasses import dataclass, field
from tesla_fleet_api.const import Scope
from tesla_fleet_api.tesla import EnergySiteRouter
from tesla_fleet_api.teslemetry import EnergySite, Vehicle
from teslemetry_stream import TeslemetryStream, TeslemetryStreamVehicle
@@ -48,11 +49,14 @@ class TeslemetryVehicleData:
@dataclass
class TeslemetryEnergyData:
"""Data for a vehicle in the Teslemetry integration."""
"""Data for an energy site in the Teslemetry integration."""
api: EnergySite
api: EnergySite | EnergySiteRouter
live_coordinator: TeslemetryEnergySiteLiveCoordinator | None
info_coordinator: TeslemetryEnergySiteInfoCoordinator
history_coordinator: TeslemetryEnergyHistoryCoordinator | None
id: int
device: DeviceInfo
# Only sites with a battery/Powerwall can pair for local TEDAPI control.
can_local_control: bool
subentry_id: str | None
homeassistant/components/teslemetry/number.py
+2 -1
View File
@@ -7,6 +7,7 @@ from typing import Any, override
from tesla_fleet_api import firmware_at_least
from tesla_fleet_api.const import Scope
from tesla_fleet_api.tesla import EnergySiteRouter
from tesla_fleet_api.teslemetry import EnergySite, Vehicle
from teslemetry_stream import TeslemetryStreamVehicle
@@ -97,7 +98,7 @@ VEHICLE_DESCRIPTIONS: tuple[TeslemetryNumberVehicleEntityDescription, ...] = (
class TeslemetryNumberBatteryEntityDescription(NumberEntityDescription):
"""Describes Teslemetry Number entity."""
func: Callable[[EnergySite, float], Awaitable[Any]]
func: Callable[[EnergySite | EnergySiteRouter, float], Awaitable[Any]]
requires: str | None = None
scopes: list[Scope]
homeassistant/components/teslemetry/strings.json
+41
View File
@@ -46,6 +46,47 @@
}
}
},
"config_subentries": {
"energy_site": {
"abort": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"entry_not_loaded": "The Teslemetry account must be loaded before setting up local control. Try again once it has finished loading.",
"no_energy_sites": "No energy sites on your Teslemetry account are available for local control. This may be because they lack a Powerwall or have already been added."
},
"entry_type": "Energy site",
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_password": "[%key:common::config_flow::error::invalid_auth%]",
"key_not_approved": "Your Powerwall system rejected Home Assistant's key because it has not been approved. Flick the On/Off switch on your primary Powerwall off and back on to approve it, then submit again.",
"key_not_registered": "Home Assistant's key is no longer registered on your Powerwall system. Restart setup to register it again.",
"key_pending": "Home Assistant's key has not been approved yet. Flick the On/Off switch on your primary Powerwall off and back on to approve it, then submit again."
},
"initiate_flow": {
"user": "Add local energy site"
},
"step": {
"credentials": {
"data": {
"host": "[%key:common::config_flow::data::host%]",
"password": "[%key:common::config_flow::data::password%]"
},
"description": "Enter your Powerwall system's local network address and its Wi-Fi password. Only the last 5 characters of the password are needed - find them on the QR label behind the front cover of your Powerwall 3, or behind the Backup Gateway door on Powerwall 2.",
"title": "Connect to your Powerwall system"
},
"pair": {
"description": "Flick the On/Off switch on your primary Powerwall off and back on to approve Home Assistant's access, then continue. This only needs to be done once. Removing this integration later does not revoke this access on your Powerwall or Backup Gateway.",
"title": "Approve the local access key"
},
"user": {
"data": {
"site_id": "Energy site"
},
"description": "Choose the Teslemetry energy site whose Powerwall you want to control over your local network.",
"title": "Add local energy site"
}
}
}
},
"entity": {
"binary_sensor": {
"automatic_blind_spot_camera": {
requirements_all.txt
Generated
+3
View File
@@ -383,6 +383,9 @@ aiopegelonline==0.1.1
# homeassistant.components.opnsense
aiopnsense==1.0.10
# homeassistant.components.teslemetry
aiopowerwall==0.2.0
# homeassistant.components.ptdevices
aioptdevices==2026.03.2
tests/components/teslemetry/test_config_flow.py
+878 -6
View File
@@ -1,31 +1,57 @@
"""Test the Teslemetry config flow."""
from collections.abc import Generator
from copy import deepcopy
import time
from typing import Any
from unittest.mock import AsyncMock, patch
from unittest.mock import AsyncMock, MagicMock, PropertyMock, patch
from urllib.parse import parse_qs, urlparse
from aiohttp import ClientConnectionError
from aiohttp import ClientConnectionError, ClientError
from aiopowerwall import (
DEFAULT_GATEWAY_HOST,
PowerwallAuthenticationError,
PowerwallConnectionError,
PowerwallFaultError,
)
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
import pytest
from tesla_fleet_api.const import AuthorizedClientState
from tesla_fleet_api.exceptions import (
InvalidResponse,
InvalidToken,
SubscriptionRequired,
TeslaFleetError,
)
from tesla_fleet_api.teslemetry.energysite import AuthorizedClient, AuthorizedClients
from homeassistant.components.application_credentials import (
ClientCredential,
async_import_client_credential,
)
from homeassistant.components.teslemetry.const import (
AUTHORIZE_URL,
CLIENT_ID,
CONF_SITE_ID,
DOMAIN,
SUBENTRY_TYPE_ENERGY_SITE,
TOKEN_URL,
)
from homeassistant.config_entries import SOURCE_USER, ConfigEntryState
from homeassistant.config_entries import (
SOURCE_USER,
ConfigEntryState,
ConfigSubentryData,
SubentryFlowResult,
)
from homeassistant.const import CONF_HOST, CONF_PASSWORD
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers import config_entry_oauth2_flow
from homeassistant.helpers import config_entry_oauth2_flow, device_registry as dr
from homeassistant.setup import async_setup_component
from . import setup_platform
from .const import CONFIG_V1, UNIQUE_ID
from . import mock_config_entry, setup_platform
from .const import CONFIG_V1, METADATA, PRODUCTS, UNIQUE_ID
from tests.common import MockConfigEntry
from tests.test_util.aiohttp import AiohttpClientMocker
@@ -146,6 +172,89 @@ async def test_reauth(
assert result["reason"] == "reauth_successful"
async def _complete_reauth(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
entry: MockConfigEntry,
) -> MagicMock:
"""Drive a reauth flow to completion and return the schedule_reload mock."""
result = await entry.start_reauth_flow(hass)
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
state = config_entry_oauth2_flow._encode_jwt(
hass,
{
"flow_id": result["flow_id"],
"redirect_uri": REDIRECT,
},
)
client = await hass_client_no_auth()
await client.get(f"/auth/external/callback?code=abcd&state={state}")
aioclient_mock.post(
TOKEN_URL,
json={
"refresh_token": "test_refresh_token",
"access_token": "test_access_token",
"type": "Bearer",
"expires_in": 60,
},
)
with patch(
"homeassistant.config_entries.ConfigEntries.async_schedule_reload"
) as mock_schedule_reload:
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reauth_successful"
return mock_schedule_reload
@pytest.mark.usefixtures("current_request_with_host")
@pytest.mark.usefixtures("mock_setup_entry")
async def test_reauth_loaded_schedules_reload(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
) -> None:
"""A data-only reauth schedules the reload itself to apply the token.
The subentry set is unchanged, so the update listener never reloads.
"""
mock_entry = await setup_platform(hass, [])
assert mock_entry.state is ConfigEntryState.LOADED
mock_schedule_reload = await _complete_reauth(
hass, hass_client_no_auth, aioclient_mock, mock_entry
)
mock_schedule_reload.assert_called_once_with(mock_entry.entry_id)
@pytest.mark.usefixtures("current_request_with_host")
async def test_reauth_not_loaded_schedules_reload(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
) -> None:
"""An unloaded entry has no update listener, so the flow schedules the reload."""
mock_entry = mock_config_entry()
mock_entry.add_to_hass(hass)
# A failed setup imports the client credential before it aborts, so mirror that
# while leaving the entry unloaded (and therefore without an update listener).
assert await async_setup_component(hass, "application_credentials", {})
await async_import_client_credential(
hass, DOMAIN, ClientCredential(CLIENT_ID, "", name="Teslemetry")
)
assert mock_entry.state is ConfigEntryState.NOT_LOADED
mock_schedule_reload = await _complete_reauth(
hass, hass_client_no_auth, aioclient_mock, mock_entry
)
mock_schedule_reload.assert_called_once_with(mock_entry.entry_id)
@pytest.mark.usefixtures("current_request_with_host")
@pytest.mark.usefixtures("mock_setup_entry")
async def test_reauth_account_mismatch(
@@ -343,6 +452,54 @@ async def test_reconfigure(
assert "expires_at" in mock_entry.data["token"]
@pytest.mark.usefixtures("current_request_with_host")
async def test_reconfigure_not_loaded_schedules_reload(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
mock_token_response: dict[str, Any],
) -> None:
"""Reconfiguring an unloaded entry has no update listener, so the flow reloads."""
mock_entry = mock_config_entry()
mock_entry.add_to_hass(hass)
# A failed setup imports the client credential before it aborts, so mirror that
# while leaving the entry unloaded (and therefore without an update listener).
assert await async_setup_component(hass, "application_credentials", {})
await async_import_client_credential(
hass, DOMAIN, ClientCredential(CLIENT_ID, "", name="Teslemetry")
)
assert mock_entry.state is ConfigEntryState.NOT_LOADED
result = await mock_entry.start_reconfigure_flow(hass)
assert result["type"] is FlowResultType.EXTERNAL_STEP
state = config_entry_oauth2_flow._encode_jwt(
hass,
{
"flow_id": result["flow_id"],
"redirect_uri": REDIRECT,
},
)
client = await hass_client_no_auth()
await client.get(f"/auth/external/callback?code=abcd&state={state}")
new_token_response = mock_token_response | {
"refresh_token": "new_refresh_token",
"access_token": "new_access_token",
}
aioclient_mock.post(TOKEN_URL, json=new_token_response)
with patch(
"homeassistant.config_entries.ConfigEntries.async_schedule_reload"
) as mock_schedule_reload:
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reconfigure_successful"
assert mock_entry.data["token"]["refresh_token"] == "new_refresh_token"
mock_schedule_reload.assert_called_once_with(mock_entry.entry_id)
@pytest.mark.usefixtures("current_request_with_host")
@pytest.mark.usefixtures("mock_setup_entry")
async def test_reconfigure_account_mismatch(
@@ -521,3 +678,718 @@ async def test_migrate_error_from_future(
entry = hass.config_entries.async_get_entry(mock_entry.entry_id)
assert entry.state is ConfigEntryState.MIGRATION_ERROR
SITE_ID = 123456
WALL_CONNECTOR_SITE_ID = 555555
HOST = "192.168.91.1"
PASSWORD = "abcde"
# Matches the paired site's `gateway_id` in the products fixture.
GATEWAY_DIN = "ABC123"
PUBLIC_KEY_DER = b"public-key-der"
PUBLIC_KEY_B64 = "cHVibGljLWtleS1kZXI="
# aiopowerwall's PowerwallClient parses the PEM at construction time, so tests
# that build one need a real (if undersized, for speed) RSA key rather than
# arbitrary bytes.
_TEST_RSA_KEY_PEM = rsa.generate_private_key(
public_exponent=65537, key_size=1024
).private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.TraditionalOpenSSL,
encryption_algorithm=serialization.NoEncryption(),
)
def _entry_with_powerwall() -> MockConfigEntry:
"""Return a config entry whose energy site subentry is already paired."""
entry = mock_config_entry()
return MockConfigEntry(
domain=entry.domain,
version=entry.version,
minor_version=entry.minor_version,
unique_id=entry.unique_id,
data=dict(entry.data),
subentries_data=[
ConfigSubentryData(
subentry_type=SUBENTRY_TYPE_ENERGY_SITE,
unique_id=str(SITE_ID),
title="Energy Site",
data={
CONF_SITE_ID: SITE_ID,
CONF_HOST: HOST,
CONF_PASSWORD: PASSWORD,
},
)
],
)
@pytest.fixture(autouse=True)
def mock_gateway_discovery() -> Generator[AsyncMock]:
"""Default gateway-address discovery to no result."""
with patch(
"tesla_fleet_api.teslemetry.energysite.TeslemetryEnergySite.find_gateway_address",
new=AsyncMock(return_value=None),
) as mock_find:
yield mock_find
@pytest.fixture
def mock_rsa_key() -> Generator[None]:
"""Mock RSA key generation/loading, avoiding real crypto and disk I/O."""
with (
patch(
"homeassistant.components.teslemetry.config_flow.Teslemetry.get_rsa_private_key",
new=AsyncMock(),
),
patch(
"homeassistant.components.teslemetry.config_flow.Teslemetry.rsa_public_der_pkcs1",
new_callable=PropertyMock,
return_value=PUBLIC_KEY_DER,
),
patch(
"homeassistant.components.teslemetry.config_flow.Teslemetry.rsa_public_der_pkcs1_b64",
new_callable=PropertyMock,
return_value=PUBLIC_KEY_B64,
),
patch(
"homeassistant.components.teslemetry.config_flow.Path.read_bytes",
return_value=_TEST_RSA_KEY_PEM,
),
):
yield
def _mock_powerwall_client(
*,
connect_error: Exception | None = None,
din: str = GATEWAY_DIN,
status_error: Exception | None = None,
) -> MagicMock:
"""Return a mock aiopowerwall PowerwallClient async context manager."""
client = MagicMock()
client.__aenter__ = AsyncMock(return_value=client)
client.__aexit__ = AsyncMock(return_value=False)
client.connect = AsyncMock(return_value=din, side_effect=connect_error)
client.get_status = AsyncMock(side_effect=status_error)
return client
def _own_key_clients(
state: AuthorizedClientState | int | str | None,
) -> AuthorizedClients:
"""Return a typed client list carrying our key in the given state."""
return AuthorizedClients(
clients=[
AuthorizedClient(
public_key="some-other-key",
state=AuthorizedClientState.VERIFIED,
roles=None,
verification=None,
raw={},
),
AuthorizedClient(
public_key=PUBLIC_KEY_B64,
state=state,
roles=None,
verification=None,
raw={},
),
],
raw=None,
)
def _empty_clients() -> AuthorizedClients:
"""Return a typed client list that is authoritatively empty."""
return AuthorizedClients(clients=[], raw=None)
async def _start_add_flow_select_site(
hass: HomeAssistant, entry: MockConfigEntry
) -> SubentryFlowResult:
"""Start the add flow and select the battery site, returning the next step."""
result = await hass.config_entries.subentries.async_init(
(entry.entry_id, SUBENTRY_TYPE_ENERGY_SITE),
context={"source": "user"},
)
assert result["step_id"] == "user"
return await hass.config_entries.subentries.async_configure(
result["flow_id"], {CONF_SITE_ID: str(SITE_ID)}
)
async def _setup_account_no_subentry(hass: HomeAssistant) -> MockConfigEntry:
"""Set up an account entry with no local-control subentry (nothing opted in)."""
entry = mock_config_entry()
entry.add_to_hass(hass)
with patch("homeassistant.components.teslemetry.PLATFORMS", []):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
return entry
def _credentials_host_default(result: SubentryFlowResult) -> str:
"""Return the CONF_HOST field's schema default from a credentials form result."""
for key in result["data_schema"].schema:
if key == CONF_HOST:
return key.default()
raise AssertionError("CONF_HOST field not found in credentials schema")
@pytest.mark.usefixtures("mock_rsa_key")
async def test_energy_subentry_pairing_requires_key_approval(
hass: HomeAssistant,
) -> None:
"""Pairing registers the key, then advances to credentials once approved."""
entry = await _setup_account_no_subentry(hass)
client = _mock_powerwall_client()
with (
patch(
"tesla_fleet_api.teslemetry.energysite.TeslemetryEnergySite.find_authorized_clients",
new=AsyncMock(
side_effect=[
_empty_clients(),
_own_key_clients(AuthorizedClientState.PENDING_VERIFICATION),
_own_key_clients(AuthorizedClientState.VERIFIED),
]
),
),
patch(
"tesla_fleet_api.teslemetry.energysite.TeslemetryEnergySite.add_authorized_client",
new=AsyncMock(),
) as mock_add,
patch(
"homeassistant.components.teslemetry.config_flow.PowerwallClient",
return_value=client,
),
patch.object(hass.config_entries, "async_schedule_reload"),
):
result = await _start_add_flow_select_site(hass, entry)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "pair"
mock_add.assert_awaited_once()
result = await hass.config_entries.subentries.async_configure(
result["flow_id"], {}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "pair"
assert result["errors"] == {"base": "key_pending"}
result = await hass.config_entries.subentries.async_configure(
result["flow_id"], {}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "credentials"
result = await hass.config_entries.subentries.async_configure(
result["flow_id"], {CONF_HOST: HOST, CONF_PASSWORD: PASSWORD}
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
subentry = entry.get_subentries_of_type(SUBENTRY_TYPE_ENERGY_SITE)[0]
assert subentry.data[CONF_HOST] == HOST
@pytest.mark.usefixtures("mock_rsa_key")
async def test_subentry_null_body_aborts_as_lookup_failure(hass: HomeAssistant) -> None:
"""A malformed authorized-clients read aborts rather than registering."""
entry = await _setup_account_no_subentry(hass)
with (
patch(
"tesla_fleet_api.teslemetry.energysite.TeslemetryEnergySite.find_authorized_clients",
new=AsyncMock(side_effect=InvalidResponse),
),
patch(
"tesla_fleet_api.teslemetry.energysite.TeslemetryEnergySite.add_authorized_client",
new=AsyncMock(),
) as mock_add,
):
result = await _start_add_flow_select_site(hass, entry)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "cannot_connect"
mock_add.assert_not_awaited()
@pytest.mark.usefixtures("mock_rsa_key")
@pytest.mark.parametrize(
("client_kwargs", "expected_error"),
[
pytest.param(
{"connect_error": PowerwallAuthenticationError()},
"invalid_password",
id="wrong_gateway_password",
),
pytest.param(
{"connect_error": PowerwallConnectionError()},
"cannot_connect",
id="gateway_unreachable",
),
pytest.param(
{"status_error": PowerwallAuthenticationError()},
"key_not_approved",
id="signed_read_rejects_unapproved_key",
),
pytest.param(
{"status_error": PowerwallFaultError("MESSAGEFAULT_ERROR_BUSY")},
"cannot_connect",
id="signed_read_generic_gateway_fault",
),
pytest.param(
{"status_error": PowerwallConnectionError()},
"cannot_connect",
id="signed_read_unreachable",
),
],
)
async def test_subentry_credentials_errors(
hass: HomeAssistant,
client_kwargs: dict[str, Exception],
expected_error: str,
) -> None:
"""The credentials step reports each local verification failure distinctly."""
entry = await _setup_account_no_subentry(hass)
client = _mock_powerwall_client(**client_kwargs)
with (
patch(
"tesla_fleet_api.teslemetry.energysite.TeslemetryEnergySite.find_authorized_clients",
new=AsyncMock(
return_value=_own_key_clients(AuthorizedClientState.VERIFIED)
),
),
patch(
"homeassistant.components.teslemetry.config_flow.PowerwallClient",
return_value=client,
),
):
result = await _start_add_flow_select_site(hass, entry)
assert result["step_id"] == "credentials"
result = await hass.config_entries.subentries.async_configure(
result["flow_id"], {CONF_HOST: HOST, CONF_PASSWORD: PASSWORD}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "credentials"
assert result["errors"] == {"base": expected_error}
assert not entry.get_subentries_of_type(SUBENTRY_TYPE_ENERGY_SITE)
@pytest.mark.usefixtures("mock_rsa_key")
async def test_subentry_credentials_prefills_discovered_host(
hass: HomeAssistant,
) -> None:
"""A discovered gateway address pre-fills the credentials CONF_HOST default."""
entry = await _setup_account_no_subentry(hass)
discovered_host = "192.168.1.138"
with (
patch(
"tesla_fleet_api.teslemetry.energysite.TeslemetryEnergySite.find_gateway_address",
new=AsyncMock(return_value=discovered_host),
),
patch(
"tesla_fleet_api.teslemetry.energysite.TeslemetryEnergySite.find_authorized_clients",
new=AsyncMock(
return_value=_own_key_clients(AuthorizedClientState.VERIFIED)
),
),
):
result = await _start_add_flow_select_site(hass, entry)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "credentials"
assert _credentials_host_default(result) == discovered_host
@pytest.mark.usefixtures("mock_rsa_key")
async def test_add_flow_lists_only_not_added_sites(hass: HomeAssistant) -> None:
"""The add flow offers battery sites that have not already been added."""
entry = await _setup_account_no_subentry(hass)
result = await hass.config_entries.subentries.async_init(
(entry.entry_id, SUBENTRY_TYPE_ENERGY_SITE),
context={"source": "user"},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
schema = result["data_schema"].schema
site_field = next(iter(schema))
assert site_field == CONF_SITE_ID
# Only the battery-capable site is selectable; the componentless site is not.
assert set(schema[site_field].container) == {str(SITE_ID)}
async def test_add_flow_aborts_when_all_sites_added(hass: HomeAssistant) -> None:
"""The add flow aborts when every battery site is already paired."""
entry = _entry_with_powerwall()
entry.add_to_hass(hass)
with (
patch(
"homeassistant.components.teslemetry._async_get_rsa_key_pem",
return_value=_TEST_RSA_KEY_PEM,
),
patch("homeassistant.components.teslemetry.PLATFORMS", []),
):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.subentries.async_init(
(entry.entry_id, SUBENTRY_TYPE_ENERGY_SITE),
context={"source": "user"},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "no_energy_sites"
@pytest.mark.usefixtures("mock_rsa_key")
async def test_add_flow_creates_subentry_bound_to_existing_device(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
) -> None:
"""The add flow creates a subentry for the site and reuses its device."""
entry = await _setup_account_no_subentry(hass)
devices_before = dr.async_entries_for_config_entry(device_registry, entry.entry_id)
site_device = next(
device
for device in devices_before
if (DOMAIN, str(SITE_ID)) in device.identifiers
)
client = _mock_powerwall_client()
with (
patch(
"tesla_fleet_api.teslemetry.energysite.TeslemetryEnergySite.find_authorized_clients",
new=AsyncMock(
return_value=_own_key_clients(AuthorizedClientState.VERIFIED)
),
),
patch(
"homeassistant.components.teslemetry.config_flow.PowerwallClient",
return_value=client,
),
patch.object(hass.config_entries, "async_schedule_reload"),
):
result = await hass.config_entries.subentries.async_init(
(entry.entry_id, SUBENTRY_TYPE_ENERGY_SITE),
context={"source": "user"},
)
assert result["step_id"] == "user"
result = await hass.config_entries.subentries.async_configure(
result["flow_id"], {CONF_SITE_ID: str(SITE_ID)}
)
assert result["step_id"] == "credentials"
result = await hass.config_entries.subentries.async_configure(
result["flow_id"], {CONF_HOST: HOST, CONF_PASSWORD: PASSWORD}
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
subentry = entry.get_subentries_of_type(SUBENTRY_TYPE_ENERGY_SITE)[0]
assert subentry.unique_id == str(SITE_ID)
assert subentry.data[CONF_SITE_ID] == SITE_ID
assert subentry.data[CONF_HOST] == HOST
assert subentry.data[CONF_PASSWORD] == PASSWORD
# No duplicate device: the same site device is reused.
site_devices = [
device
for device in dr.async_entries_for_config_entry(device_registry, entry.entry_id)
if (DOMAIN, str(SITE_ID)) in device.identifiers
]
assert [device.id for device in site_devices] == [site_device.id]
@pytest.mark.usefixtures("mock_rsa_key")
async def test_subentry_credentials_password_truncated(hass: HomeAssistant) -> None:
"""A full Wi-Fi password is trimmed to its final five characters."""
entry = await _setup_account_no_subentry(hass)
client = _mock_powerwall_client()
with (
patch(
"tesla_fleet_api.teslemetry.energysite.TeslemetryEnergySite.find_authorized_clients",
new=AsyncMock(
return_value=_own_key_clients(AuthorizedClientState.VERIFIED)
),
),
patch(
"homeassistant.components.teslemetry.config_flow.PowerwallClient",
return_value=client,
) as mock_client,
patch.object(hass.config_entries, "async_schedule_reload"),
):
result = await _start_add_flow_select_site(hass, entry)
assert result["step_id"] == "credentials"
result = await hass.config_entries.subentries.async_configure(
result["flow_id"], {CONF_HOST: HOST, CONF_PASSWORD: "long-wifi-password"}
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
subentry = entry.get_subentries_of_type(SUBENTRY_TYPE_ENERGY_SITE)[0]
assert subentry.data[CONF_PASSWORD] == "sword"
assert mock_client.call_args.kwargs["gateway_password"] == "sword"
@pytest.mark.usefixtures("mock_rsa_key")
async def test_wall_connector_only_site_not_offered_for_local_control(
hass: HomeAssistant,
) -> None:
"""A wall-connector-only site can't do local control; only a Powerwall can."""
products = deepcopy(PRODUCTS)
products["response"].append(
{
"energy_site_id": WALL_CONNECTOR_SITE_ID,
"site_name": "Wall Connector Site",
"components": {
"battery": False,
"solar": False,
"grid": True,
"wall_connectors": [{"device_id": "wc-1", "din": "WC-DIN-1"}],
},
}
)
metadata = deepcopy(METADATA)
metadata["energy_sites"][str(WALL_CONNECTOR_SITE_ID)] = {
"access": True,
"name": "Wall Connector Site",
}
entry = mock_config_entry()
entry.add_to_hass(hass)
with (
patch("tesla_fleet_api.teslemetry.Teslemetry.products", return_value=products),
patch("tesla_fleet_api.teslemetry.Teslemetry.metadata", return_value=metadata),
patch("homeassistant.components.teslemetry.PLATFORMS", []),
):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert not entry.get_subentries_of_type(SUBENTRY_TYPE_ENERGY_SITE)
result = await hass.config_entries.subentries.async_init(
(entry.entry_id, SUBENTRY_TYPE_ENERGY_SITE),
context={"source": "user"},
)
schema = result["data_schema"].schema
site_field = next(iter(schema))
assert set(schema[site_field].container) == {str(SITE_ID)}
async def test_add_flow_aborts_when_entry_not_loaded(hass: HomeAssistant) -> None:
"""The add flow aborts when the account entry is not loaded."""
entry = mock_config_entry()
entry.add_to_hass(hass)
result = await hass.config_entries.subentries.async_init(
(entry.entry_id, SUBENTRY_TYPE_ENERGY_SITE),
context={"source": "user"},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "entry_not_loaded"
@pytest.mark.usefixtures("mock_rsa_key")
async def test_gateway_discovery_failure_proceeds_without_host(
hass: HomeAssistant,
) -> None:
"""A failed gateway-address discovery leaves the host default unset."""
entry = await _setup_account_no_subentry(hass)
with (
patch(
"tesla_fleet_api.teslemetry.energysite.TeslemetryEnergySite.find_gateway_address",
new=AsyncMock(side_effect=ClientError),
),
patch(
"tesla_fleet_api.teslemetry.energysite.TeslemetryEnergySite.find_authorized_clients",
new=AsyncMock(
return_value=_own_key_clients(AuthorizedClientState.VERIFIED)
),
),
):
result = await _start_add_flow_select_site(hass, entry)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "credentials"
assert _credentials_host_default(result) == DEFAULT_GATEWAY_HOST
@pytest.mark.usefixtures("mock_rsa_key")
async def test_pending_key_resumes_without_reregister(hass: HomeAssistant) -> None:
"""A key already pending on the gateway resumes pairing without re-adding it."""
entry = await _setup_account_no_subentry(hass)
with (
patch(
"tesla_fleet_api.teslemetry.energysite.TeslemetryEnergySite.find_authorized_clients",
new=AsyncMock(
return_value=_own_key_clients(
AuthorizedClientState.PENDING_VERIFICATION
)
),
),
patch(
"tesla_fleet_api.teslemetry.energysite.TeslemetryEnergySite.add_authorized_client",
new=AsyncMock(),
) as mock_add,
):
result = await _start_add_flow_select_site(hass, entry)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "pair"
mock_add.assert_not_awaited()
@pytest.mark.usefixtures("mock_rsa_key")
async def test_timed_out_key_reregisters_for_a_fresh_window(
hass: HomeAssistant,
) -> None:
"""A key whose approval window expired is re-registered, not left stuck."""
entry = await _setup_account_no_subentry(hass)
with (
patch(
"tesla_fleet_api.teslemetry.energysite.TeslemetryEnergySite.find_authorized_clients",
new=AsyncMock(
return_value=_own_key_clients(
AuthorizedClientState.PENDING_VERIFICATION_TIMEOUT
)
),
),
patch(
"tesla_fleet_api.teslemetry.energysite.TeslemetryEnergySite.add_authorized_client",
new=AsyncMock(),
) as mock_add,
):
result = await _start_add_flow_select_site(hass, entry)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "pair"
mock_add.assert_awaited_once()
@pytest.mark.usefixtures("mock_rsa_key")
async def test_unrecognized_state_aborts_pairing(hass: HomeAssistant) -> None:
"""An unrecognized authorized-client state aborts rather than re-registering."""
entry = await _setup_account_no_subentry(hass)
with (
patch(
"tesla_fleet_api.teslemetry.energysite.TeslemetryEnergySite.find_authorized_clients",
new=AsyncMock(return_value=_own_key_clients("gremlin")),
),
patch(
"tesla_fleet_api.teslemetry.energysite.TeslemetryEnergySite.add_authorized_client",
new=AsyncMock(),
) as mock_add,
):
result = await _start_add_flow_select_site(hass, entry)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "cannot_connect"
mock_add.assert_not_awaited()
@pytest.mark.usefixtures("mock_rsa_key")
async def test_add_authorized_client_failure_aborts(hass: HomeAssistant) -> None:
"""A failure while registering the key aborts the flow."""
entry = await _setup_account_no_subentry(hass)
with (
patch(
"tesla_fleet_api.teslemetry.energysite.TeslemetryEnergySite.find_authorized_clients",
new=AsyncMock(return_value=_empty_clients()),
),
patch(
"tesla_fleet_api.teslemetry.energysite.TeslemetryEnergySite.add_authorized_client",
new=AsyncMock(side_effect=ClientError),
),
):
result = await _start_add_flow_select_site(hass, entry)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "cannot_connect"
@pytest.mark.usefixtures("mock_rsa_key")
@pytest.mark.parametrize(
("second_lookup", "expected_error"),
[
pytest.param(InvalidResponse(), "cannot_connect", id="lookup_failure"),
pytest.param(_empty_clients(), "key_not_registered", id="key_not_registered"),
pytest.param(_own_key_clients("gremlin"), "cannot_connect", id="unknown_state"),
],
)
async def test_pair_step_second_lookup_errors(
hass: HomeAssistant,
second_lookup: Exception | AuthorizedClients,
expected_error: str,
) -> None:
"""Re-checking the pending key reports each non-approval outcome on the form."""
entry = await _setup_account_no_subentry(hass)
with (
patch(
"tesla_fleet_api.teslemetry.energysite.TeslemetryEnergySite.find_authorized_clients",
new=AsyncMock(side_effect=[_empty_clients(), second_lookup]),
),
patch(
"tesla_fleet_api.teslemetry.energysite.TeslemetryEnergySite.add_authorized_client",
new=AsyncMock(),
),
):
result = await _start_add_flow_select_site(hass, entry)
assert result["step_id"] == "pair"
result = await hass.config_entries.subentries.async_configure(
result["flow_id"], {}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "pair"
assert result["errors"] == {"base": expected_error}
@pytest.mark.usefixtures("mock_rsa_key")
@pytest.mark.parametrize(
("patch_target", "error"),
[
pytest.param(
"homeassistant.components.teslemetry.config_flow.Teslemetry.get_rsa_private_key",
OSError,
id="key_fetch_oserror",
),
pytest.param(
"homeassistant.components.teslemetry.config_flow.Path.read_bytes",
ValueError,
id="key_read_valueerror",
),
],
)
async def test_rsa_key_load_failure_aborts(
hass: HomeAssistant,
patch_target: str,
error: type[Exception],
) -> None:
"""A failure loading the integration's RSA key aborts site preparation."""
entry = await _setup_account_no_subentry(hass)
with patch(patch_target, side_effect=error):
result = await _start_add_flow_select_site(hass, entry)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "cannot_connect"
tests/components/teslemetry/test_init.py
+437 -3
View File
@@ -1,10 +1,15 @@
"""Test the Teslemetry init."""
from copy import deepcopy
import logging
import time
from types import MappingProxyType
from unittest.mock import AsyncMock, MagicMock, patch
from aiohttp import ClientResponseError
from aiopowerwall import PowerwallError
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from freezegun.api import FrozenDateTimeFactory
import pytest
from syrupy.assertion import SnapshotAssertion
@@ -18,9 +23,20 @@ from tesla_fleet_api.exceptions import (
SubscriptionRequired,
TeslaFleetError,
)
from tesla_fleet_api.tesla import EnergySiteRouter
from tesla_fleet_api.teslemetry import EnergySite
from homeassistant.components.teslemetry import STREAM_TOPICS, _get_access_token
from homeassistant.components.teslemetry.const import CLIENT_ID, DOMAIN
from homeassistant.components.teslemetry import (
STREAM_TOPICS,
_async_get_rsa_key_pem,
_get_access_token,
)
from homeassistant.components.teslemetry.const import (
CLIENT_ID,
CONF_SITE_ID,
DOMAIN,
SUBENTRY_TYPE_ENERGY_SITE,
)
# Coordinator constants
from homeassistant.components.teslemetry.coordinator import (
@@ -31,8 +47,14 @@ from homeassistant.components.teslemetry.coordinator import (
)
from homeassistant.components.teslemetry.models import TeslemetryData
from homeassistant.components.teslemetry.oauth import TeslemetryImplementation
from homeassistant.config_entries import ConfigEntryState
from homeassistant.config_entries import (
ConfigEntryState,
ConfigSubentry,
ConfigSubentryData,
)
from homeassistant.const import (
CONF_HOST,
CONF_PASSWORD,
STATE_OFF,
STATE_ON,
STATE_UNAVAILABLE,
@@ -57,6 +79,7 @@ from .const import (
LIVE_STATUS,
METADATA,
METADATA_NOSCOPE,
PRODUCTS,
PRODUCTS_MODERN,
SITE_INFO,
UNIQUE_ID,
@@ -1215,6 +1238,417 @@ async def test_get_access_token_rate_limited_after_setup_is_not_fatal(
await _get_access_token(session)
await hass.async_block_till_done()
assert not hass.config_entries.flow.async_progress()
SITE_ID = 123456
HOST = "192.168.91.1"
PASSWORD = "abcde"
# aiopowerwall's PowerwallClient parses the PEM at construction time, so tests
# that build one need a real (if undersized, for speed) RSA key rather than
# arbitrary bytes.
_TEST_RSA_KEY_PEM = rsa.generate_private_key(
public_exponent=65537, key_size=1024
).private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.TraditionalOpenSSL,
encryption_algorithm=serialization.NoEncryption(),
)
def _entry_with_powerwall() -> MockConfigEntry:
"""Return a config entry whose energy site subentry is already paired."""
entry = mock_config_entry()
return MockConfigEntry(
domain=entry.domain,
version=entry.version,
minor_version=entry.minor_version,
unique_id=entry.unique_id,
data=dict(entry.data),
subentries_data=[
ConfigSubentryData(
subentry_type=SUBENTRY_TYPE_ENERGY_SITE,
unique_id=str(SITE_ID),
title="Energy Site",
data={
CONF_SITE_ID: SITE_ID,
CONF_HOST: HOST,
CONF_PASSWORD: PASSWORD,
},
)
],
)
async def _setup_account_no_subentry(hass: HomeAssistant) -> MockConfigEntry:
"""Set up an account entry with no local-control subentry (nothing opted in)."""
entry = mock_config_entry()
entry.add_to_hass(hass)
with patch("homeassistant.components.teslemetry.PLATFORMS", []):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
return entry
async def test_energy_site_router_with_powerwall(hass: HomeAssistant) -> None:
"""A paired energy site wraps its cloud API in an EnergySiteRouter."""
entry = _entry_with_powerwall()
entry.add_to_hass(hass)
with (
patch(
"homeassistant.components.teslemetry._async_get_rsa_key_pem",
return_value=_TEST_RSA_KEY_PEM,
),
patch("homeassistant.components.teslemetry.PLATFORMS", []),
):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
energysite = entry.runtime_data.energysites[0]
assert isinstance(energysite.api, EnergySiteRouter)
async def test_energy_site_cloud_without_powerwall(hass: HomeAssistant) -> None:
"""An energy site without paired credentials keeps the plain cloud API."""
entry = mock_config_entry()
entry.add_to_hass(hass)
with patch("homeassistant.components.teslemetry.PLATFORMS", []):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
energysite = entry.runtime_data.energysites[0]
assert isinstance(energysite.api, EnergySite)
assert not isinstance(energysite.api, EnergySiteRouter)
async def test_energy_site_subentry_without_credentials_uses_cloud(
hass: HomeAssistant,
) -> None:
"""A subentry that exists but is not yet paired resolves to the cloud API.
A site whose subentry was created but has no gateway host/password stored
keeps that subentry_id (so it stays opted in) while falling back to the
plain cloud API rather than building an EnergySiteRouter.
"""
entry = mock_config_entry()
paired = MockConfigEntry(
domain=entry.domain,
version=entry.version,
minor_version=entry.minor_version,
unique_id=entry.unique_id,
data=dict(entry.data),
subentries_data=[
ConfigSubentryData(
subentry_type=SUBENTRY_TYPE_ENERGY_SITE,
unique_id=str(SITE_ID),
title="Energy Site",
data={CONF_SITE_ID: SITE_ID},
)
],
)
paired.add_to_hass(hass)
with patch("homeassistant.components.teslemetry.PLATFORMS", []):
await hass.config_entries.async_setup(paired.entry_id)
await hass.async_block_till_done()
energysite = paired.runtime_data.energysites[0]
assert isinstance(energysite.api, EnergySite)
assert not isinstance(energysite.api, EnergySiteRouter)
assert energysite.subentry_id is not None
assert energysite.can_local_control
async def test_no_subentry_created_at_setup(hass: HomeAssistant) -> None:
"""Setup never auto-creates a local-control subentry; it is opt-in."""
entry = await _setup_account_no_subentry(hass)
assert not entry.get_subentries_of_type(SUBENTRY_TYPE_ENERGY_SITE)
energysite = entry.runtime_data.energysites[0]
assert energysite.can_local_control
assert energysite.subentry_id is None
assert not isinstance(energysite.api, EnergySiteRouter)
@pytest.mark.parametrize(
"local_error",
[
pytest.param(OSError("disk gone"), id="os_error"),
pytest.param(ValueError("bad key"), id="value_error"),
pytest.param(PowerwallError("client boom"), id="powerwall_error"),
],
)
async def test_local_control_failure_falls_back_to_cloud(
hass: HomeAssistant,
local_error: Exception,
caplog: pytest.LogCaptureFixture,
) -> None:
"""A failure resolving a paired site's local gateway falls back to cloud.
Local control is opt-in per site, so one site's bad local config must leave
the entry loaded with cloud functionality intact rather than tearing the
whole integration down.
"""
entry = _entry_with_powerwall()
entry.add_to_hass(hass)
with (
patch(
"homeassistant.components.teslemetry._async_get_rsa_key_pem",
side_effect=local_error,
),
patch("homeassistant.components.teslemetry.PLATFORMS", []),
caplog.at_level(logging.WARNING),
):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.LOADED
energysite = entry.runtime_data.energysites[0]
assert isinstance(energysite.api, EnergySite)
assert not isinstance(energysite.api, EnergySiteRouter)
assert energysite.can_local_control
assert "falling back to cloud control" in caplog.text
assert any(
record.levelname == "WARNING" and str(SITE_ID) in record.message
for record in caplog.records
)
async def test_get_rsa_key_pem_generates_and_caches(hass: HomeAssistant) -> None:
"""The RSA key is generated/read once, then served from the hass.data cache."""
with (
patch(
"homeassistant.components.teslemetry.Teslemetry.get_rsa_private_key",
new=AsyncMock(),
) as mock_get_key,
patch(
"homeassistant.components.teslemetry.Path.read_bytes",
return_value=_TEST_RSA_KEY_PEM,
),
):
first = await _async_get_rsa_key_pem(hass)
second = await _async_get_rsa_key_pem(hass)
assert first == _TEST_RSA_KEY_PEM
assert second == _TEST_RSA_KEY_PEM
mock_get_key.assert_awaited_once()
@pytest.mark.parametrize(
("local_error", "expected", "cloud_awaits"),
[
pytest.param(None, {"routed": "local"}, 0, id="local_success"),
pytest.param(
PowerwallError("boom"), {"routed": "cloud"}, 1, id="cloud_fallback"
),
],
)
async def test_energy_site_router_command_routing(
hass: HomeAssistant,
local_error: Exception | None,
expected: dict[str, str],
cloud_awaits: int,
) -> None:
"""A command routes to the local Powerwall first and falls back to cloud."""
entry = _entry_with_powerwall()
entry.add_to_hass(hass)
with (
patch(
"homeassistant.components.teslemetry._async_get_rsa_key_pem",
return_value=_TEST_RSA_KEY_PEM,
),
patch("homeassistant.components.teslemetry.PLATFORMS", []),
):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
router = entry.runtime_data.energysites[0].api
assert isinstance(router, EnergySiteRouter)
local = AsyncMock(side_effect=local_error, return_value={"routed": "local"})
cloud = AsyncMock(return_value={"routed": "cloud"})
with (
patch("aiopowerwall.energysite.PowerwallEnergySite.backup", new=local),
patch(
"tesla_fleet_api.teslemetry.energysite.TeslemetryEnergySite.backup",
new=cloud,
),
):
result = await router.backup(50)
assert result == expected
local.assert_awaited_once_with(50)
assert cloud.await_count == cloud_awaits
async def test_stale_cleanup_preserves_foreign_subentry(hass: HomeAssistant) -> None:
"""Energy stale-subentry cleanup does not remove other subentry types."""
entry = mock_config_entry()
entry.add_to_hass(hass)
foreign = ConfigSubentry(
data=MappingProxyType({"vin": "VIN123"}),
subentry_type="vehicle",
title="A Vehicle",
unique_id="VIN123",
)
hass.config_entries.async_add_subentry(entry, foreign)
with patch("homeassistant.components.teslemetry.PLATFORMS", []):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert foreign.subentry_id in entry.subentries
assert entry.subentries[foreign.subentry_id].subentry_type == "vehicle"
async def test_stale_cleanup_removes_energy_subentry(hass: HomeAssistant) -> None:
"""A paired site that is gone from the account has its subentry pruned."""
entry = _entry_with_powerwall()
entry.add_to_hass(hass)
subentry_id = entry.get_subentries_of_type(SUBENTRY_TYPE_ENERGY_SITE)[0].subentry_id
products = deepcopy(PRODUCTS)
products["response"] = [
product
for product in products["response"]
if product.get("energy_site_id") != SITE_ID
]
with (
patch("tesla_fleet_api.teslemetry.Teslemetry.products", return_value=products),
patch("homeassistant.components.teslemetry.PLATFORMS", []),
):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert subentry_id not in entry.subentries
async def test_stale_cleanup_preserves_pairing_on_transient_access_loss(
hass: HomeAssistant,
) -> None:
"""A paired site that momentarily reports no access keeps its subentry."""
entry = _entry_with_powerwall()
entry.add_to_hass(hass)
subentry_id = entry.get_subentries_of_type(SUBENTRY_TYPE_ENERGY_SITE)[0].subentry_id
metadata = deepcopy(METADATA)
metadata["energy_sites"][str(SITE_ID)]["access"] = False
with (
patch("tesla_fleet_api.teslemetry.Teslemetry.metadata", return_value=metadata),
patch("homeassistant.components.teslemetry.PLATFORMS", []),
):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert subentry_id in entry.subentries
assert entry.subentries[subentry_id].data[CONF_HOST] == HOST
assert entry.subentries[subentry_id].data[CONF_PASSWORD] == PASSWORD
async def test_solar_only_site_has_no_local_control(hass: HomeAssistant) -> None:
"""A solar-only site gets no local-control subentry: there is no Powerwall."""
products = deepcopy(PRODUCTS)
site = next(
product
for product in products["response"]
if product.get("energy_site_id") == SITE_ID
)
site["components"]["battery"] = False
site["components"].pop("wall_connectors")
entry = mock_config_entry()
entry.add_to_hass(hass)
with (
patch("tesla_fleet_api.teslemetry.Teslemetry.products", return_value=products),
patch("homeassistant.components.teslemetry.PLATFORMS", []),
):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert not entry.get_subentries_of_type(SUBENTRY_TYPE_ENERGY_SITE)
energysite = entry.runtime_data.energysites[0]
assert energysite.subentry_id is None
assert not isinstance(energysite.api, EnergySiteRouter)
async def test_stale_cleanup_preserves_pairing_without_energy_scope(
hass: HomeAssistant,
) -> None:
"""Losing the energy scope must not delete a paired site's stored credentials."""
entry = _entry_with_powerwall()
entry.add_to_hass(hass)
subentry_id = entry.get_subentries_of_type(SUBENTRY_TYPE_ENERGY_SITE)[0].subentry_id
with (
patch(
"tesla_fleet_api.teslemetry.Teslemetry.metadata",
return_value=METADATA_NOSCOPE,
),
patch("homeassistant.components.teslemetry.PLATFORMS", []),
):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert not entry.runtime_data.energysites
assert subentry_id in entry.subentries
assert entry.subentries[subentry_id].data[CONF_HOST] == HOST
assert entry.subentries[subentry_id].data[CONF_PASSWORD] == PASSWORD
async def test_update_listener_ignores_token_refresh(hass: HomeAssistant) -> None:
"""An entry update that only changes token data must not reload the entry.
OAuth token refreshes call async_update_entry with new token data on every
expiry; reloading on those would needlessly drop the stream and re-fetch.
"""
entry = mock_config_entry()
entry.add_to_hass(hass)
with patch("homeassistant.components.teslemetry.PLATFORMS", []):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
with patch.object(hass.config_entries, "async_schedule_reload") as mock_reload:
new_data = dict(entry.data)
new_data["token"] = {**new_data["token"], "access_token": "refreshed_token"}
hass.config_entries.async_update_entry(entry, data=new_data)
await hass.async_block_till_done()
mock_reload.assert_not_called()
async def test_update_listener_reloads_on_subentry_change(
hass: HomeAssistant,
) -> None:
"""Adding a local-energy-site subentry reloads the entry."""
entry = mock_config_entry()
entry.add_to_hass(hass)
with patch("homeassistant.components.teslemetry.PLATFORMS", []):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
with patch.object(hass.config_entries, "async_schedule_reload") as mock_reload:
hass.config_entries.async_add_subentry(
entry,
ConfigSubentry(
data=MappingProxyType(
{CONF_SITE_ID: SITE_ID, CONF_HOST: HOST, CONF_PASSWORD: PASSWORD}
),
subentry_type=SUBENTRY_TYPE_ENERGY_SITE,
title="Energy Site",
unique_id=str(SITE_ID),
),
)
await hass.async_block_till_done()
mock_reload.assert_called_once_with(entry.entry_id)
def test_stream_topic_allowlist() -> None:
"""The stream subscribes to exactly the topics the integration consumes."""