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

Add Bluetooth control for Teslemetry vehicles (#176296)

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-11 15:02:19 +02:00
committed by GitHub
co-authored by Copilot Autofix powered by AI
parent 8c64cf2248
commit 8f40062f29
19 changed files with 1797 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/teslemetry/__init__.py
+109 -15
View File
@@ -8,16 +8,19 @@ from typing import Any, Final, cast
from aiohttp import ClientError
from aiopowerwall import PowerwallClient, PowerwallEnergySite, PowerwallError
from bleak.exc import BleakError
from tesla_fleet_api.const import Scope
from tesla_fleet_api.exceptions import (
Forbidden,
InvalidToken,
LoginRequired,
PrivateKeyError,
SubscriptionRequired,
TeslaFleetError,
)
from tesla_fleet_api.router import VehicleRouter
from tesla_fleet_api.tesla import EnergySiteRouter
from tesla_fleet_api.teslemetry import EnergySite, Teslemetry
from tesla_fleet_api.teslemetry import EnergySite, Teslemetry, Vehicle
from teslemetry_stream import TeslemetryStream, TeslemetryStreamAuthenticationError
from teslemetry_stream.const import SseTopic
@@ -25,8 +28,15 @@ from homeassistant.components.application_credentials import (
ClientCredential,
async_import_client_credential,
)
from homeassistant.components.bluetooth import async_ble_device_from_address
from homeassistant.config_entries import ConfigEntry, ConfigEntryState
from homeassistant.const import CONF_ACCESS_TOKEN, CONF_HOST, CONF_PASSWORD, Platform
from homeassistant.const import (
CONF_ACCESS_TOKEN,
CONF_ADDRESS,
CONF_HOST,
CONF_PASSWORD,
Platform,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import (
ConfigEntryAuthFailed,
@@ -50,11 +60,13 @@ from homeassistant.helpers.update_coordinator import UpdateFailed
from .const import (
CLIENT_ID,
CONF_VIN,
DOMAIN,
LOGGER,
POWERWALL_KEY_FILE,
RSA_PARENT_KEY,
SUBENTRY_TYPE_ENERGY_SITE,
SUBENTRY_TYPE_VEHICLE,
VEHICLE_ISSUE_LEARN_MORE,
)
from .coordinator import (
@@ -64,7 +76,7 @@ from .coordinator import (
TeslemetryMetadataCoordinator,
TeslemetryVehicleDataCoordinator,
)
from .helpers import async_update_device_sw_version, flatten
from .helpers import async_get_ble_parent, async_update_device_sw_version, flatten
from .models import TeslemetryData, TeslemetryEnergyData, TeslemetryVehicleData
from .services import async_setup_services
@@ -271,6 +283,71 @@ def _setup_vehicle_repairs(
)
def _ble_address_for_vin(entry: TeslemetryConfigEntry, vin: str) -> str | None:
"""Return the paired Bluetooth address for a vehicle, if one was added."""
for subentry in entry.subentries.values():
if (
subentry.subentry_type == SUBENTRY_TYPE_VEHICLE
and subentry.data.get(CONF_VIN) == vin
):
return subentry.data.get(CONF_ADDRESS)
return None
# Two failure shapes must be caught to fall back to cloud control: the library
# wraps existing-key failures in PrivateKeyError, the create-race path raises raw errors.
_BLE_KEY_ERRORS: Final = (
OSError,
ValueError,
AssertionError,
TypeError,
PrivateKeyError,
)
async def _async_resolve_vehicle_api(
hass: HomeAssistant,
entry: TeslemetryConfigEntry,
vin: str,
cloud_vehicle: Vehicle,
) -> Vehicle | VehicleRouter:
"""Return the API a vehicle's platforms should call."""
address = _ble_address_for_vin(entry, vin)
if not address:
return cloud_vehicle
# A bad BLE key file for one vehicle must not tear down the whole entry.
try:
parent = await async_get_ble_parent(hass)
except _BLE_KEY_ERRORS:
LOGGER.warning(
"Failed to load the Bluetooth key for vehicle %s; "
"falling back to cloud control",
vin,
exc_info=True,
)
return cloud_vehicle
# disable keep alive to allow vehicles to sleep
bluetooth_vehicle = parent.vehicles.createBluetooth(
vin,
confirmation="verify",
raise_unconfirmed=False,
keepalive_interval=None,
)
@callback
def _in_range() -> bool:
"""Report whether the vehicle is currently reachable over Bluetooth."""
device = async_ble_device_from_address(hass, address, connectable=True)
if device is None:
return False
# The library never refreshes the BLE handle, so set it here while it is known fresh.
bluetooth_vehicle.set_device(device)
return True
return VehicleRouter(bluetooth_vehicle, cloud_vehicle, health=_in_range)
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(
@@ -333,20 +410,17 @@ async def _async_get_rsa_key_pem(hass: HomeAssistant) -> bytes:
pem: bytes | None = hass.data.get(RSA_PARENT_KEY)
if pem is None:
path = hass.config.path(POWERWALL_KEY_FILE)
try:
await Teslemetry(
session=async_get_clientsession(hass), access_token=""
).get_rsa_private_key(path)
except TypeError as err:
# An encrypted PEM surfaces as TypeError from the cryptography loader.
raise ValueError("RSA private key file is encrypted") from err
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)
# Both key-load failure shapes (wrapped PrivateKeyError, raw OSError/ValueError) plus
# PowerwallError must be caught to fall back to cloud control.
_LOCAL_CONTROL_ERRORS: Final = (OSError, ValueError, PowerwallError, PrivateKeyError)
async def _async_resolve_local_control(
@@ -526,9 +600,16 @@ async def async_setup_entry(hass: HomeAssistant, entry: TeslemetryConfigEntry) -
)
stream_vehicle = stream.get_vehicle(vin)
vehicle_api = await _async_resolve_vehicle_api(
hass,
entry,
vin,
vehicle,
)
vehicles.append(
TeslemetryVehicleData(
api=vehicle,
api=vehicle_api,
config_entry=entry,
coordinator=coordinator,
poll=poll,
@@ -697,7 +778,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: TeslemetryConfigEntry) -
def _setup_subentry_change_reload(
hass: HomeAssistant, entry: TeslemetryConfigEntry
) -> None:
"""Reload the entry when a local-energy-site subentry is added or removed."""
"""Reload the entry when a subentry is added or removed."""
known = set(entry.subentries)
async def _handle_update(
@@ -818,7 +899,20 @@ async def _async_setup_energy_site(
async def async_unload_entry(hass: HomeAssistant, entry: TeslemetryConfigEntry) -> bool:
"""Unload Teslemetry Config."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
unloaded = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if unloaded:
# Release any on-demand Bluetooth link only after platforms unloaded, or the still-loaded entry's backends must keep working.
for vehicle in entry.runtime_data.vehicles:
if isinstance(vehicle.api, VehicleRouter):
try:
await vehicle.api.primary.disconnect()
except (BleakError, TeslaFleetError, TimeoutError) as err:
# Swallowed so one stuck link cannot block the unload, but
# warn: a leaked BLE connection can keep the vehicle awake.
LOGGER.warning(
"Error disconnecting Bluetooth for %s: %s", vehicle.vin, err
)
return unloaded
async def async_migrate_entry(
homeassistant/components/teslemetry/button.py
+2 -1
View File
@@ -5,6 +5,7 @@ from dataclasses import dataclass
from typing import Any, override
from tesla_fleet_api.const import Scope
from tesla_fleet_api.router import VehicleRouter
from tesla_fleet_api.teslemetry import Vehicle
from homeassistant.components.button import ButtonEntity, ButtonEntityDescription
@@ -75,7 +76,7 @@ async def async_setup_entry(
class TeslemetryButtonEntity(TeslemetryVehicleStreamEntity, ButtonEntity):
"""Base class for Teslemetry buttons."""
api: Vehicle
api: Vehicle | VehicleRouter
entity_description: TeslemetryButtonEntityDescription
def __init__(
homeassistant/components/teslemetry/climate.py
+3 -2
View File
@@ -5,6 +5,7 @@ from typing import Any, cast, override
from tesla_fleet_api import firmware_at_least
from tesla_fleet_api.const import CabinOverheatProtectionTemp, Scope
from tesla_fleet_api.router import VehicleRouter
from tesla_fleet_api.teslemetry import Vehicle
from homeassistant.components.climate import (
@@ -90,7 +91,7 @@ async def async_setup_entry(
class TeslemetryClimateEntity(TeslemetryRootEntity, ClimateEntity):
"""Vehicle Climate Control."""
api: Vehicle
api: Vehicle | VehicleRouter
_attr_precision = PRECISION_HALVES
_attr_temperature_unit = UnitOfTemperature.CELSIUS
_attr_hvac_modes = [HVACMode.HEAT_COOL, HVACMode.OFF]
@@ -385,7 +386,7 @@ COP_LEVELS = {
class TeslemetryCabinOverheatProtectionEntity(TeslemetryRootEntity, ClimateEntity):
"""Vehicle Cabin Overheat Protection."""
api: Vehicle
api: Vehicle | VehicleRouter
_attr_precision = PRECISION_WHOLE
_attr_target_temperature_step = 5
_attr_min_temp = 30
homeassistant/components/teslemetry/config_flow.py
+234 -13
View File
@@ -1,5 +1,6 @@
"""Config Flow for Teslemetry integration."""
import asyncio
from collections.abc import Mapping
import logging
from pathlib import Path
@@ -7,16 +8,23 @@ from typing import TYPE_CHECKING, Any, cast, override
from aiohttp import ClientError
from aiopowerwall import PowerwallAuthenticationError, PowerwallClient, PowerwallError
from bleak.exc import BleakError
from tesla_fleet_api.const import (
AuthorizedClientKeyType,
AuthorizedClientState,
AuthorizedClientType,
)
from tesla_fleet_api.exceptions import (
BluetoothTimeout,
BluetoothTransportError,
InvalidToken,
NotOnWhitelistFault,
PrivateKeyError,
SubscriptionRequired,
TeslaFleetError,
WhitelistOperationAttemptingToAddExistingKey,
)
from tesla_fleet_api.tesla.vehicle.bluetooth import VehicleBluetooth
from tesla_fleet_api.teslemetry import Teslemetry
from tesla_fleet_api.teslemetry.energysite import AuthorizedClient, TeslemetryEnergySite
import voluptuous as vol
@@ -25,6 +33,10 @@ from homeassistant.components.application_credentials import (
ClientCredential,
async_import_client_credential,
)
from homeassistant.components.bluetooth import (
async_discovered_service_info,
async_request_active_scan,
)
from homeassistant.config_entries import (
SOURCE_REAUTH,
SOURCE_RECONFIGURE,
@@ -34,20 +46,29 @@ from homeassistant.config_entries import (
ConfigSubentryFlow,
SubentryFlowResult,
)
from homeassistant.const import CONF_HOST, CONF_PASSWORD
from homeassistant.const import CONF_ADDRESS, 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 homeassistant.helpers.selector import (
SelectOptionDict,
SelectSelector,
SelectSelectorConfig,
SelectSelectorMode,
)
from . import TeslemetryConfigEntry
from . import _BLE_KEY_ERRORS, TeslemetryConfigEntry
from .const import (
CLIENT_ID,
CONF_SITE_ID,
CONF_VIN,
DOMAIN,
LOGGER,
POWERWALL_KEY_FILE,
SUBENTRY_TYPE_ENERGY_SITE,
SUBENTRY_TYPE_VEHICLE,
)
from .helpers import async_get_ble_parent
class PowerwallLookupError(Exception):
@@ -85,7 +106,10 @@ class OAuth2FlowHandler(
cls, config_entry: ConfigEntry
) -> dict[str, type[ConfigSubentryFlow]]:
"""Return the subentry types supported by this integration."""
return {SUBENTRY_TYPE_ENERGY_SITE: EnergySiteSubentryFlowHandler}
return {
SUBENTRY_TYPE_VEHICLE: VehicleSubentryFlowHandler,
SUBENTRY_TYPE_ENERGY_SITE: EnergySiteSubentryFlowHandler,
}
@override
async def async_step_user(
@@ -187,6 +211,210 @@ class OAuth2FlowHandler(
return await self.async_step_user()
class VehicleSubentryFlowHandler(ConfigSubentryFlow):
"""Add local Bluetooth control to one of the account's vehicles."""
def __init__(self) -> None:
"""Initialize the vehicle subentry flow."""
self._vin: str | None = None
self._title: str | None = None
self._address: str | None = None
self._vehicle: VehicleBluetooth | None = None
self._pair_task: asyncio.Task[None] | None = None
self._pair_error: dict[str, str] = {}
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> SubentryFlowResult:
"""Select an account vehicle to add over Bluetooth, then pair it."""
entry = self._get_entry()
if entry.state is not ConfigEntryState.LOADED:
return self.async_abort(reason="entry_not_loaded")
already_added = {
subentry.data[CONF_VIN]
for subentry in entry.get_subentries_of_type(SUBENTRY_TYPE_VEHICLE)
if CONF_VIN in subentry.data
}
choices = {
vehicle.vin: vehicle.device["name"] or vehicle.vin
for vehicle in entry.runtime_data.vehicles
if vehicle.vin not in already_added
}
if not choices:
return self.async_abort(reason="no_vehicles")
if user_input is not None:
self._vin = user_input[CONF_VIN]
self._title = choices[self._vin]
return await self.async_step_scan()
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_VIN): SelectSelector(
SelectSelectorConfig(
options=[
SelectOptionDict(value=vin, label=name)
for vin, name in choices.items()
],
mode=SelectSelectorMode.DROPDOWN,
)
)
}
),
)
async def async_step_scan(
self, user_input: dict[str, Any] | None = None
) -> SubentryFlowResult:
"""Find the vehicle over Bluetooth and connect to it."""
if TYPE_CHECKING:
assert self._vin is not None
errors: dict[str, str] = {}
if user_input is not None:
try:
parent = await async_get_ble_parent(self.hass)
except _BLE_KEY_ERRORS as err:
LOGGER.debug("Bluetooth key load failed: %s", err)
errors["base"] = "cannot_connect"
else:
# The advertised BLE name is a hash of the VIN; match on its prefix.
expected = parent.get_name(self._vin)[:17]
device = None
# The name is only in scan responses, so an active scan may be needed to see it.
await async_request_active_scan(self.hass)
for info in async_discovered_service_info(self.hass, connectable=True):
if info.name and info.name.startswith(expected):
device = info.device
self._address = info.address
break
if device is None:
errors["base"] = "device_not_found"
else:
# Uses default keepalive so the link survives the on-screen key-approval wait.
self._vehicle = parent.vehicles.createBluetooth(
self._vin, device=device
)
try:
await self._vehicle.connect()
except (BleakError, TeslaFleetError, TimeoutError) as err:
LOGGER.error("Failed to connect over Bluetooth: %s", err)
await self._async_disconnect()
errors["base"] = "cannot_connect"
else:
return await self.async_step_pair()
return self.async_show_form(
step_id="scan",
errors=errors,
description_placeholders={"vin": self._vin},
)
async def async_step_pair(
self, user_input: dict[str, Any] | None = None
) -> SubentryFlowResult:
"""Check whether the virtual key is already whitelisted on the vehicle."""
if TYPE_CHECKING:
assert self._vehicle is not None
try:
await self._vehicle.handshakeVehicleSecurity()
except NotOnWhitelistFault:
return await self.async_step_instructions()
except (BleakError, TeslaFleetError, TimeoutError) as err:
LOGGER.error("Bluetooth security handshake failed: %s", err)
await self._async_disconnect()
# The scan step owns the form; re-show it so a retry redoes scan and connect.
return self.async_show_form(
step_id="scan",
errors={"base": "cannot_connect"},
description_placeholders={"vin": self._vin or ""},
)
if TYPE_CHECKING:
assert self._address is not None
assert self._vin is not None
await self._async_disconnect()
return self.async_create_entry(
title=self._title or self._vin,
data={CONF_VIN: self._vin, CONF_ADDRESS: self._address},
unique_id=self._vin,
)
async def async_step_instructions(
self, user_input: dict[str, Any] | None = None
) -> SubentryFlowResult:
"""Ask the user to approve the virtual key on the vehicle touchscreen."""
if user_input is not None:
return await self.async_step_authorize()
errors = self._pair_error
self._pair_error = {}
return self.async_show_form(
step_id="instructions",
errors=errors,
description_placeholders={"vin": self._vin or ""},
)
async def async_step_authorize(
self, user_input: dict[str, Any] | None = None
) -> SubentryFlowResult:
"""Add the virtual key to the vehicle while showing pairing progress."""
if self._pair_task is None:
if TYPE_CHECKING:
assert self._vehicle is not None
# pair() can take minutes, so run it as a progress task rather than blocking the flow.
self._pair_task = self.hass.async_create_task(self._vehicle.pair())
if not self._pair_task.done():
return self.async_show_progress(
step_id="authorize",
progress_action="pair",
progress_task=self._pair_task,
description_placeholders={"vin": self._vin or ""},
)
task = self._pair_task
self._pair_task = None
try:
task.result()
except (BluetoothTransportError, BleakError) as err:
LOGGER.debug("Bluetooth transport failed during pairing: %s", err)
self._pair_error = {"base": "cannot_connect"}
return self.async_show_progress_done(next_step_id="instructions")
except (BluetoothTimeout, TimeoutError) as err:
LOGGER.debug("Bluetooth pairing timed out: %s", err)
self._pair_error = {"base": "timeout"}
return self.async_show_progress_done(next_step_id="instructions")
except WhitelistOperationAttemptingToAddExistingKey as err:
LOGGER.debug("Virtual key is already on the whitelist: %s", err)
except TeslaFleetError as err:
LOGGER.error("Bluetooth pairing was rejected: %s", err)
self._pair_error = {"base": "pair_failed"}
return self.async_show_progress_done(next_step_id="instructions")
return self.async_show_progress_done(next_step_id="pair")
async def _async_disconnect(self) -> None:
"""Disconnect the BLE link, if any, and drop the reference to it."""
vehicle = self._vehicle
if vehicle is not None:
try:
await vehicle.disconnect()
except (BleakError, TeslaFleetError, TimeoutError) as err:
LOGGER.debug("Error disconnecting Bluetooth: %s", err)
finally:
self._vehicle = None
@callback
@override
def async_remove(self) -> None:
"""Release resources if the flow is abandoned mid-pairing."""
if self._pair_task is not None and not self._pair_task.done():
self._pair_task.cancel()
if self._vehicle is not None:
self.hass.async_create_task(self._async_disconnect())
class EnergySiteSubentryFlowHandler(ConfigSubentryFlow):
"""Pair a local Powerwall gateway for TEDAPI v1r command routing."""
@@ -257,10 +485,7 @@ class EnergySiteSubentryFlowHandler(ConfigSubentryFlow):
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.
"""
"""Discover the gateway address and load the integration's RSA key."""
self._energy_site = energy_site
try:
@@ -277,15 +502,11 @@ class EnergySiteSubentryFlowHandler(ConfigSubentryFlow):
session=async_get_clientsession(self.hass), access_token=""
)
try:
try:
await keyholder.get_rsa_private_key(path)
except TypeError as err:
# An encrypted PEM surfaces as TypeError from the cryptography loader.
raise ValueError("RSA private key file is encrypted") from err
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:
except (OSError, ValueError, PrivateKeyError) 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
homeassistant/components/teslemetry/const.py
+6
View File
@@ -12,6 +12,12 @@ AUTHORIZE_URL = "https://teslemetry.com/connect"
TOKEN_URL = "https://api.teslemetry.com/oauth/token"
CLIENT_ID = "homeassistant"
SUBENTRY_TYPE_VEHICLE = "vehicle"
CONF_VIN = "vin"
VEHICLE_KEY_FILE = "tesla_vehicle.key"
BLE_PARENT_KEY = f"{DOMAIN}_ble_parent"
BLE_PARENT_LOCK_KEY = f"{DOMAIN}_ble_parent_lock"
SUBENTRY_TYPE_ENERGY_SITE = "energy_site"
CONF_SITE_ID = "site_id"
POWERWALL_KEY_FILE = "tesla_powerwall.key"
homeassistant/components/teslemetry/cover.py
+7 -6
View File
@@ -11,6 +11,7 @@ from tesla_fleet_api.const import (
Trunk,
WindowCommand,
)
from tesla_fleet_api.router import VehicleRouter
from tesla_fleet_api.teslemetry import Vehicle
from teslemetry_stream import Signal
from teslemetry_stream.const import WindowState
@@ -120,7 +121,7 @@ class CoverRestoreEntity(RestoreEntity, CoverEntity):
class TeslemetryWindowEntity(TeslemetryRootEntity, CoverEntity):
"""Base class for window cover entities."""
api: Vehicle
api: Vehicle | VehicleRouter
_attr_device_class = CoverDeviceClass.WINDOW
_attr_supported_features = CoverEntityFeature.OPEN | CoverEntityFeature.CLOSE
@@ -254,7 +255,7 @@ class TeslemetryChargePortEntity(
):
"""Base class for for charge port cover entities."""
api: Vehicle
api: Vehicle | VehicleRouter
_attr_device_class = CoverDeviceClass.DOOR
_attr_supported_features = CoverEntityFeature.OPEN | CoverEntityFeature.CLOSE
@@ -340,7 +341,7 @@ class TeslemetryStreamingChargePortEntity(
class TeslemetryFrontTrunkEntity(TeslemetryRootEntity, CoverEntity):
"""Base class for the front trunk cover entities."""
api: Vehicle
api: Vehicle | VehicleRouter
_attr_device_class = CoverDeviceClass.DOOR
_attr_supported_features = CoverEntityFeature.OPEN
@@ -407,7 +408,7 @@ class TeslemetryStreamingFrontTrunkEntity(
class TeslemetryRearTrunkEntity(TeslemetryRootEntity, CoverEntity):
"""Cover entity for the rear trunk."""
api: Vehicle
api: Vehicle | VehicleRouter
_attr_device_class = CoverDeviceClass.DOOR
_attr_supported_features = CoverEntityFeature.OPEN | CoverEntityFeature.CLOSE
@@ -482,7 +483,7 @@ class TeslemetryStreamingRearTrunkEntity(
class TeslemetrySunroofEntity(TeslemetryVehiclePollingEntity, CoverEntity):
"""Cover entity for the sunroof."""
api: Vehicle
api: Vehicle | VehicleRouter
_attr_device_class = CoverDeviceClass.WINDOW
_attr_supported_features = (
CoverEntityFeature.OPEN | CoverEntityFeature.CLOSE | CoverEntityFeature.STOP
@@ -538,7 +539,7 @@ class TeslemetrySunroofEntity(TeslemetryVehiclePollingEntity, CoverEntity):
class TeslemetryTonneauEntity(TeslemetryRootEntity, CoverEntity):
"""Base class for the Cybertruck tonneau cover entity."""
api: Vehicle
api: Vehicle | VehicleRouter
_attr_device_class = CoverDeviceClass.DOOR
_attr_supported_features = (
CoverEntityFeature.OPEN | CoverEntityFeature.CLOSE | CoverEntityFeature.STOP
homeassistant/components/teslemetry/entity.py
+3 -2
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.router import VehicleRouter
from tesla_fleet_api.tesla import EnergySiteRouter
from tesla_fleet_api.teslemetry import EnergySite, Vehicle
@@ -106,7 +107,7 @@ class TeslemetryVehiclePollingEntity(TeslemetryPollingEntity):
"""Parent class for Teslemetry Vehicle entities."""
_last_update: int = 0
api: Vehicle
api: Vehicle | VehicleRouter
vehicle: TeslemetryVehicleData
def __init__(
@@ -259,7 +260,7 @@ class TeslemetryWallConnectorEntity(TeslemetryPollingEntity):
class TeslemetryVehicleStreamEntity(TeslemetryRootEntity):
"""Parent class for Teslemetry Vehicle Stream entities."""
api: Vehicle
api: Vehicle | VehicleRouter
def __init__(self, data: TeslemetryVehicleData, key: str) -> None:
"""Initialize common aspects of a Teslemetry entity."""
homeassistant/components/teslemetry/helpers.py
+16 -1
View File
@@ -1,15 +1,30 @@
"""Teslemetry helper functions."""
import asyncio
from collections.abc import Awaitable
from typing import Any
from tesla_fleet_api.exceptions import TeslaFleetError
from tesla_fleet_api.tesla.bluetooth import TeslaBluetooth
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import device_registry as dr, entity_registry as er
from .const import DOMAIN, LOGGER
from .const import BLE_PARENT_KEY, BLE_PARENT_LOCK_KEY, DOMAIN, LOGGER, VEHICLE_KEY_FILE
async def async_get_ble_parent(hass: HomeAssistant) -> TeslaBluetooth:
"""Return a shared TeslaBluetooth parent with the private key loaded."""
lock: asyncio.Lock = hass.data.setdefault(BLE_PARENT_LOCK_KEY, asyncio.Lock())
async with lock:
existing: TeslaBluetooth | None = hass.data.get(BLE_PARENT_KEY)
if existing is not None:
return existing
parent = TeslaBluetooth()
await parent.get_private_key(hass.config.path(VEHICLE_KEY_FILE))
hass.data[BLE_PARENT_KEY] = parent
return parent
def flatten(
homeassistant/components/teslemetry/lock.py
+3 -2
View File
@@ -5,6 +5,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.router import VehicleRouter
from tesla_fleet_api.teslemetry import Vehicle
from homeassistant.components.lock import LockEntity
@@ -64,7 +65,7 @@ async def async_setup_entry(
class TeslemetryVehicleLockEntity(TeslemetryRootEntity, LockEntity):
"""Base vehicle lock entity for Teslemetry."""
api: Vehicle
api: Vehicle | VehicleRouter
@override
async def async_lock(self, **kwargs: Any) -> None:
@@ -141,7 +142,7 @@ class TeslemetryStreamingVehicleLockEntity(
class TeslemetryCableLockEntity(TeslemetryRootEntity, LockEntity):
"""Base cable Lock entity for Teslemetry."""
api: Vehicle
api: Vehicle | VehicleRouter
@override
async def async_lock(self, **kwargs: Any) -> None:
homeassistant/components/teslemetry/manifest.json
+1 -1
View File
@@ -3,7 +3,7 @@
"name": "Teslemetry",
"codeowners": ["@Bre77"],
"config_flow": true,
"dependencies": ["application_credentials"],
"dependencies": ["application_credentials", "bluetooth_adapters"],
"documentation": "https://www.home-assistant.io/integrations/teslemetry",
"integration_type": "hub",
"iot_class": "cloud_polling",
homeassistant/components/teslemetry/media_player.py
+2 -1
View File
@@ -4,6 +4,7 @@ from typing import override
from tesla_fleet_api import firmware_at_least
from tesla_fleet_api.const import Scope
from tesla_fleet_api.router import VehicleRouter
from tesla_fleet_api.teslemetry import Vehicle
from homeassistant.components.media_player import (
@@ -64,7 +65,7 @@ async def async_setup_entry(
class TeslemetryMediaEntity(TeslemetryRootEntity, MediaPlayerEntity):
"""Base vehicle media player class."""
api: Vehicle
api: Vehicle | VehicleRouter
_attr_device_class = MediaPlayerDeviceClass.SPEAKER
_attr_volume_step = VOLUME_STEP
homeassistant/components/teslemetry/models.py
+2 -1
View File
@@ -4,6 +4,7 @@ import asyncio
from dataclasses import dataclass, field
from tesla_fleet_api.const import Scope
from tesla_fleet_api.router import VehicleRouter
from tesla_fleet_api.tesla import EnergySiteRouter
from tesla_fleet_api.teslemetry import EnergySite, Vehicle
from teslemetry_stream import TeslemetryStream, TeslemetryStreamVehicle
@@ -35,7 +36,7 @@ class TeslemetryData:
class TeslemetryVehicleData:
"""Data for a vehicle in the Teslemetry integration."""
api: Vehicle
api: Vehicle | VehicleRouter
config_entry: ConfigEntry
coordinator: TeslemetryVehicleDataCoordinator
poll: bool
homeassistant/components/teslemetry/number.py
+3 -2
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.router import VehicleRouter
from tesla_fleet_api.tesla import EnergySiteRouter
from tesla_fleet_api.teslemetry import EnergySite, Vehicle
from teslemetry_stream import TeslemetryStreamVehicle
@@ -45,7 +46,7 @@ PARALLEL_UPDATES = 0
class TeslemetryNumberVehicleEntityDescription(NumberEntityDescription):
"""Describes Teslemetry Number entity."""
func: Callable[[Vehicle, int], Awaitable[Any]]
func: Callable[[Vehicle | VehicleRouter, int], Awaitable[Any]]
min_key: str | None = None
max_key: str
native_min_value: float
@@ -171,7 +172,7 @@ async def async_setup_entry(
class TeslemetryVehicleNumberEntity(TeslemetryRootEntity, NumberEntity):
"""Vehicle number entity base class."""
api: Vehicle
api: Vehicle | VehicleRouter
entity_description: TeslemetryNumberVehicleEntityDescription
@override
homeassistant/components/teslemetry/select.py
+3 -2
View File
@@ -6,6 +6,7 @@ from typing import Any, override
from tesla_fleet_api import firmware_at_least
from tesla_fleet_api.const import EnergyExportMode, EnergyOperationMode, Scope, Seat
from tesla_fleet_api.router import VehicleRouter
from tesla_fleet_api.teslemetry import Vehicle
from teslemetry_stream import TeslemetryStreamVehicle
@@ -43,7 +44,7 @@ LEVEL = {OFF: 0, LOW: 1, MEDIUM: 2, HIGH: 3}
class TeslemetrySelectEntityDescription(SelectEntityDescription):
"""Seat Heater entity description."""
select_fn: Callable[[Vehicle, int], Awaitable[Any]]
select_fn: Callable[[Vehicle | VehicleRouter, int], Awaitable[Any]]
supported_fn: Callable[[dict], bool] = lambda _: True
streaming_listener: (
Callable[
@@ -273,7 +274,7 @@ async def async_setup_entry(
class TeslemetrySelectEntity(TeslemetryRootEntity, SelectEntity):
"""Parent vehicle select entity class."""
api: Vehicle
api: Vehicle | VehicleRouter
entity_description: TeslemetrySelectEntityDescription
_climate: bool = False
homeassistant/components/teslemetry/strings.json
+41
View File
@@ -48,6 +48,7 @@
"energy_site": {
"abort": {
"all_sites_added": "Every accessible energy site with a Powerwall on your Teslemetry account has already been added for local control.",
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"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_powerwall": "Local control requires a Powerwall, and no energy site with one is currently accessible on your Teslemetry account."
@@ -84,6 +85,46 @@
"title": "Add local energy site"
}
}
},
"vehicle": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"entry_not_loaded": "The Teslemetry configuration entry is not loaded. Please ensure it is set up correctly before adding a vehicle.",
"no_vehicles": "Every vehicle in your Teslemetry account has already been added over Bluetooth."
},
"entry_type": "Bluetooth vehicle",
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"device_not_found": "No matching Tesla vehicle was found nearby over Bluetooth. Make sure the vehicle is awake and within range, then try again.",
"pair_failed": "The vehicle rejected the key. Make sure it is not in valet mode and does not already have the maximum number of keys, then try again.",
"timeout": "Timed out waiting for the vehicle to accept the key. Try again after approving the key on the vehicle's touchscreen."
},
"initiate_flow": {
"user": "Add Bluetooth vehicle"
},
"progress": {
"pair": "Approve Home Assistant's virtual key on the vehicle's touchscreen. Waiting for the vehicle to accept the key."
},
"step": {
"instructions": {
"description": "Select **Submit**, then place your key card against the center console card reader of vehicle {vin} to approve Home Assistant's virtual key. This only needs to be done once.",
"title": "Approve the virtual key"
},
"scan": {
"description": "Home Assistant will look for vehicle {vin} over Bluetooth to enable local command control. Make sure the vehicle is awake and within Bluetooth range of a Home Assistant Bluetooth adapter, then continue.",
"title": "Set up Bluetooth control"
},
"user": {
"data": {
"vin": "Vehicle"
},
"data_description": {
"vin": "The account vehicle to enable local Bluetooth control for."
},
"description": "Select which of your Teslemetry account vehicles to add over Bluetooth for local command control.",
"title": "Add Bluetooth vehicle"
}
}
}
},
"entity": {
homeassistant/components/teslemetry/switch.py
+4 -3
View File
@@ -6,6 +6,7 @@ from typing import Any, override
from tesla_fleet_api import firmware_at_least
from tesla_fleet_api.const import AutoSeat, Scope
from tesla_fleet_api.router import VehicleRouter
from tesla_fleet_api.teslemetry import Vehicle
from teslemetry_stream import TeslemetryStreamVehicle
@@ -37,8 +38,8 @@ class TeslemetrySwitchEntityDescription(SwitchEntityDescription):
"""Describes Teslemetry Switch entity."""
polling: bool = False
on_func: Callable[[Vehicle], Awaitable[dict[str, Any]]]
off_func: Callable[[Vehicle], Awaitable[dict[str, Any]]]
on_func: Callable[[Vehicle | VehicleRouter], Awaitable[dict[str, Any]]]
off_func: Callable[[Vehicle | VehicleRouter], Awaitable[dict[str, Any]]]
scopes: list[Scope]
value_func: Callable[[StateType], bool] = bool
streaming_listener: Callable[
@@ -200,7 +201,7 @@ async def async_setup_entry(
class TeslemetryVehicleSwitchEntity(TeslemetryRootEntity, SwitchEntity):
"""Base class for all Teslemetry switch entities."""
api: Vehicle
api: Vehicle | VehicleRouter
_attr_device_class = SwitchDeviceClass.SWITCH
entity_description: TeslemetrySwitchEntityDescription
homeassistant/components/teslemetry/update.py
+2 -1
View File
@@ -4,6 +4,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.router import VehicleRouter
from tesla_fleet_api.teslemetry import Vehicle
from homeassistant.components.update import (
@@ -51,7 +52,7 @@ async def async_setup_entry(
class TeslemetryUpdateEntity(TeslemetryRootEntity, UpdateEntity):
"""Teslemetry Updates entity."""
api: Vehicle
api: Vehicle | VehicleRouter
_attr_supported_features = UpdateEntityFeature.PROGRESS
@override
tests/components/teslemetry/test_config_flow.py
+736 -11
View File
@@ -1,5 +1,6 @@
"""Test the Teslemetry config flow."""
import asyncio
from collections.abc import Generator
from copy import deepcopy
import time
@@ -13,16 +14,24 @@ from aiopowerwall import (
PowerwallConnectionError,
PowerwallFaultError,
)
from bleak.exc import BleakError
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 (
BluetoothTimeout,
BluetoothTransportError,
InvalidResponse,
InvalidToken,
NotOnWhitelistFault,
PrivateKeyError,
SubscriptionRequired,
TeslaFleetError,
WhitelistOperationAttemptingToAddExistingKey,
)
from tesla_fleet_api.tesla import VehicleRouter
from tesla_fleet_api.tesla.bluetooth import TeslaBluetooth
from tesla_fleet_api.teslemetry.energysite import AuthorizedClient, AuthorizedClients
import voluptuous as vol
@@ -34,20 +43,27 @@ from homeassistant.components.teslemetry.const import (
AUTHORIZE_URL,
CLIENT_ID,
CONF_SITE_ID,
CONF_VIN,
DOMAIN,
SUBENTRY_TYPE_ENERGY_SITE,
SUBENTRY_TYPE_VEHICLE,
TOKEN_URL,
)
from homeassistant.config_entries import (
SOURCE_USER,
ConfigEntryState,
ConfigSubentry,
ConfigSubentryData,
SubentryFlowResult,
)
from homeassistant.const import CONF_HOST, CONF_PASSWORD
from homeassistant.const import CONF_ADDRESS, CONF_HOST, CONF_PASSWORD
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers import config_entry_oauth2_flow, device_registry as dr
from homeassistant.helpers import (
config_entry_oauth2_flow,
device_registry as dr,
entity_registry as er,
)
from homeassistant.setup import async_setup_component
from . import mock_config_entry, setup_platform
@@ -139,7 +155,7 @@ async def test_reauth(
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
flows = hass.config_entries.flow.async_progress()
flows = hass.config_entries.flow.async_progress_by_handler(DOMAIN)
assert len(flows) == 1
# Progress from reauth_confirm to external OAuth step
@@ -219,10 +235,7 @@ async def test_reauth_loaded_schedules_reload(
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.
"""
"""A data-only reauth schedules the reload itself to apply the token."""
mock_entry = await setup_platform(hass, [])
assert mock_entry.state is ConfigEntryState.LOADED
@@ -680,6 +693,718 @@ async def test_migrate_error_from_future(
assert entry.state is ConfigEntryState.MIGRATION_ERROR
VIN = "LRW3F7EK4NC700000"
ADDRESS = "AA:BB:CC:DD:EE:FF"
def _entry_with_ble() -> MockConfigEntry:
"""Return a config entry whose vehicle subentry is already BLE-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_VEHICLE,
unique_id=VIN,
title="Test",
data={CONF_VIN: VIN, CONF_ADDRESS: ADDRESS},
)
],
)
def _discovered_info() -> MagicMock:
"""Return a fake discovered service info matching the test VIN."""
info = MagicMock()
info.name = TeslaBluetooth().get_name(VIN)
info.address = ADDRESS
info.device = MagicMock()
return info
def _mock_vehicle(*, on_whitelist: bool = True) -> AsyncMock:
"""Return a mock VehicleBluetooth for the pairing flow."""
vehicle = AsyncMock()
if on_whitelist:
vehicle.handshakeVehicleSecurity = AsyncMock()
else:
vehicle.handshakeVehicleSecurity = AsyncMock(
side_effect=[NotOnWhitelistFault(), None]
)
return vehicle
def _mock_ble_parent(vehicle: AsyncMock) -> MagicMock:
"""Return a mock shared TeslaBluetooth parent for the pairing flow."""
parent = MagicMock()
parent.get_name.return_value = TeslaBluetooth().get_name(VIN)
parent.vehicles.createBluetooth.return_value = vehicle
return parent
async def _setup_account_entry(hass: HomeAssistant) -> MockConfigEntry:
"""Set up an account entry with no vehicle subentry."""
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 _setup_paired_entry(hass: HomeAssistant) -> MockConfigEntry:
"""Set up an entry whose only account vehicle is already BLE-paired."""
entry = _entry_with_ble()
entry.add_to_hass(hass)
with (
patch(
"homeassistant.components.teslemetry.async_ble_device_from_address",
return_value=None,
),
patch(
"homeassistant.components.teslemetry.helpers.TeslaBluetooth"
) as mock_parent,
patch("homeassistant.components.teslemetry.PLATFORMS", []),
):
mock_parent.return_value.get_private_key = AsyncMock()
mock_parent.return_value.vehicles.createBluetooth.return_value = AsyncMock()
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
return entry
async def _start_pairing_at_scan(
hass: HomeAssistant, entry: MockConfigEntry
) -> SubentryFlowResult:
"""Open the add flow and advance past VIN selection to the scan step."""
result = await hass.config_entries.subentries.async_init(
(entry.entry_id, SUBENTRY_TYPE_VEHICLE),
context={"source": "user"},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
result = await hass.config_entries.subentries.async_configure(
result["flow_id"], {CONF_VIN: VIN}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "scan"
return result
async def test_subentry_pairing_already_whitelisted(hass: HomeAssistant) -> None:
"""The add flow creates the subentry when the key is already whitelisted."""
entry = await _setup_account_entry(hass)
vehicle = _mock_vehicle(on_whitelist=True)
with (
patch(
"homeassistant.components.teslemetry.config_flow.async_discovered_service_info",
return_value=[_discovered_info()],
),
patch(
"homeassistant.components.teslemetry.config_flow.async_get_ble_parent",
return_value=_mock_ble_parent(vehicle),
),
patch.object(hass.config_entries, "async_schedule_reload"),
):
result = await _start_pairing_at_scan(hass, entry)
result = await hass.config_entries.subentries.async_configure(
result["flow_id"], {}
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
subentries = entry.get_subentries_of_type(SUBENTRY_TYPE_VEHICLE)
assert len(subentries) == 1
# The subentry is created atomically with its credentials, never identity-only.
assert subentries[0].unique_id == VIN
assert subentries[0].data == {CONF_VIN: VIN, CONF_ADDRESS: ADDRESS}
vehicle.connect.assert_awaited_once()
vehicle.disconnect.assert_awaited_once()
async def test_subentry_pairing_duplicate_vin_aborts(hass: HomeAssistant) -> None:
"""A second flow racing on the same VIN aborts with already_configured."""
entry = await _setup_account_entry(hass)
vehicle = _mock_vehicle(on_whitelist=True)
with (
patch(
"homeassistant.components.teslemetry.config_flow.async_discovered_service_info",
return_value=[_discovered_info()],
),
patch(
"homeassistant.components.teslemetry.config_flow.async_get_ble_parent",
return_value=_mock_ble_parent(vehicle),
),
patch.object(hass.config_entries, "async_schedule_reload"),
):
result = await _start_pairing_at_scan(hass, entry)
# Simulate a concurrent flow that paired the same VIN first.
hass.config_entries.async_add_subentry(
entry,
ConfigSubentry(
data={CONF_VIN: VIN, CONF_ADDRESS: ADDRESS},
subentry_type=SUBENTRY_TYPE_VEHICLE,
title="Test",
unique_id=VIN,
),
)
result = await hass.config_entries.subentries.async_configure(
result["flow_id"], {}
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
# The pre-existing subentry from the winning flow is left untouched.
assert len(entry.get_subentries_of_type(SUBENTRY_TYPE_VEHICLE)) == 1
async def test_subentry_pairing_requires_key_approval(hass: HomeAssistant) -> None:
"""Pairing walks through instructions and key install when not whitelisted."""
entry = await _setup_account_entry(hass)
vehicle = _mock_vehicle(on_whitelist=False)
release = asyncio.Event()
async def _pair() -> None:
await release.wait()
vehicle.pair = AsyncMock(side_effect=_pair)
with (
patch(
"homeassistant.components.teslemetry.config_flow.async_discovered_service_info",
return_value=[_discovered_info()],
),
patch(
"homeassistant.components.teslemetry.config_flow.async_get_ble_parent",
return_value=_mock_ble_parent(vehicle),
),
patch.object(hass.config_entries, "async_schedule_reload"),
):
result = await _start_pairing_at_scan(hass, entry)
result = await hass.config_entries.subentries.async_configure(
result["flow_id"], {}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "instructions"
result = await hass.config_entries.subentries.async_configure(
result["flow_id"], {}
)
assert result["type"] is FlowResultType.SHOW_PROGRESS
assert result["progress_action"] == "pair"
release.set()
await hass.async_block_till_done()
result = await hass.config_entries.subentries.async_configure(result["flow_id"])
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
subentries = entry.get_subentries_of_type(SUBENTRY_TYPE_VEHICLE)
assert len(subentries) == 1
assert subentries[0].data == {CONF_VIN: VIN, CONF_ADDRESS: ADDRESS}
vehicle.pair.assert_awaited_once()
async def test_subentry_scan_connect_fails(hass: HomeAssistant) -> None:
"""The scan step re-shows the form with an error when BLE connect fails."""
entry = await _setup_account_entry(hass)
vehicle = _mock_vehicle()
vehicle.connect = AsyncMock(side_effect=BleakError("nope"))
with (
patch(
"homeassistant.components.teslemetry.config_flow.async_discovered_service_info",
return_value=[_discovered_info()],
),
patch(
"homeassistant.components.teslemetry.config_flow.async_get_ble_parent",
return_value=_mock_ble_parent(vehicle),
),
):
result = await _start_pairing_at_scan(hass, entry)
result = await hass.config_entries.subentries.async_configure(
result["flow_id"], {}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "scan"
assert result["errors"] == {"base": "cannot_connect"}
# A failed pairing never creates a subentry.
assert not entry.get_subentries_of_type(SUBENTRY_TYPE_VEHICLE)
vehicle.disconnect.assert_awaited_once()
@pytest.mark.parametrize(
("error", "expected"),
[
(BluetoothTimeout, "timeout"),
(BluetoothTransportError, "cannot_connect"),
(TeslaFleetError, "pair_failed"),
],
ids=["timeout", "transport", "rejected"],
)
async def test_subentry_authorize_failure(
hass: HomeAssistant, error: type[TeslaFleetError], expected: str
) -> None:
"""Each pairing failure surfaces its own error, not a blanket timeout."""
entry = await _setup_account_entry(hass)
vehicle = _mock_vehicle(on_whitelist=False)
release = asyncio.Event()
async def _pair() -> None:
await release.wait()
raise error
vehicle.pair = AsyncMock(side_effect=_pair)
with (
patch(
"homeassistant.components.teslemetry.config_flow.async_discovered_service_info",
return_value=[_discovered_info()],
),
patch(
"homeassistant.components.teslemetry.config_flow.async_get_ble_parent",
return_value=_mock_ble_parent(vehicle),
),
):
result = await _start_pairing_at_scan(hass, entry)
result = await hass.config_entries.subentries.async_configure(
result["flow_id"], {}
)
assert result["step_id"] == "instructions"
result = await hass.config_entries.subentries.async_configure(
result["flow_id"], {}
)
assert result["type"] is FlowResultType.SHOW_PROGRESS
release.set()
await hass.async_block_till_done()
result = await hass.config_entries.subentries.async_configure(result["flow_id"])
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "instructions"
assert result["errors"] == {"base": expected}
assert not entry.get_subentries_of_type(SUBENTRY_TYPE_VEHICLE)
# pair() is a single bounded op; it is never re-sent.
vehicle.pair.assert_awaited_once()
async def test_subentry_authorize_existing_key_finishes(hass: HomeAssistant) -> None:
"""Approving the key after a timeout, then retrying, completes the pairing."""
entry = await _setup_account_entry(hass)
vehicle = _mock_vehicle(on_whitelist=False)
releases = [asyncio.Event(), asyncio.Event()]
attempts = iter(
zip(
releases,
[BluetoothTimeout(), WhitelistOperationAttemptingToAddExistingKey()],
strict=True,
)
)
async def _pair() -> None:
release, error = next(attempts)
await release.wait()
raise error
vehicle.pair = AsyncMock(side_effect=_pair)
with (
patch(
"homeassistant.components.teslemetry.config_flow.async_discovered_service_info",
return_value=[_discovered_info()],
),
patch(
"homeassistant.components.teslemetry.config_flow.async_get_ble_parent",
return_value=_mock_ble_parent(vehicle),
),
patch.object(hass.config_entries, "async_schedule_reload"),
):
result = await _start_pairing_at_scan(hass, entry)
result = await hass.config_entries.subentries.async_configure(
result["flow_id"], {}
)
assert result["step_id"] == "instructions"
result = await hass.config_entries.subentries.async_configure(
result["flow_id"], {}
)
assert result["type"] is FlowResultType.SHOW_PROGRESS
releases[0].set()
await hass.async_block_till_done()
result = await hass.config_entries.subentries.async_configure(result["flow_id"])
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "timeout"}
result = await hass.config_entries.subentries.async_configure(
result["flow_id"], {}
)
assert result["type"] is FlowResultType.SHOW_PROGRESS
releases[1].set()
await hass.async_block_till_done()
result = await hass.config_entries.subentries.async_configure(result["flow_id"])
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
subentries = entry.get_subentries_of_type(SUBENTRY_TYPE_VEHICLE)
assert len(subentries) == 1
assert subentries[0].data == {CONF_VIN: VIN, CONF_ADDRESS: ADDRESS}
assert vehicle.pair.await_count == 2
vehicle.disconnect.assert_awaited_once()
@pytest.mark.parametrize(
"handshake_error",
[
pytest.param(TeslaFleetError(), id="tesla_fleet_error"),
pytest.param(BleakError("boom"), id="bleak_error"),
pytest.param(TimeoutError(), id="timeout_error"),
],
)
async def test_subentry_handshake_error_recovers(
hass: HomeAssistant, handshake_error: Exception
) -> None:
"""A handshake failure re-shows the scan form; retrying then pairs."""
entry = await _setup_account_entry(hass)
vehicle = _mock_vehicle()
vehicle.handshakeVehicleSecurity = AsyncMock(side_effect=[handshake_error, None])
vehicle.disconnect = AsyncMock(side_effect=BleakError("boom"))
with (
patch(
"homeassistant.components.teslemetry.config_flow.async_discovered_service_info",
return_value=[_discovered_info()],
),
patch(
"homeassistant.components.teslemetry.config_flow.async_get_ble_parent",
return_value=_mock_ble_parent(vehicle),
),
patch.object(hass.config_entries, "async_schedule_reload"),
):
result = await _start_pairing_at_scan(hass, entry)
result = await hass.config_entries.subentries.async_configure(
result["flow_id"], {}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "scan"
assert result["errors"] == {"base": "cannot_connect"}
assert not entry.get_subentries_of_type(SUBENTRY_TYPE_VEHICLE)
result = await hass.config_entries.subentries.async_configure(
result["flow_id"], {}
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
subentries = entry.get_subentries_of_type(SUBENTRY_TYPE_VEHICLE)
assert len(subentries) == 1
assert subentries[0].data == {CONF_VIN: VIN, CONF_ADDRESS: ADDRESS}
# Both the failed and successful attempts disconnected; the disconnect error is swallowed.
assert vehicle.disconnect.await_count == 2
async def test_subentry_pairing_abandoned(hass: HomeAssistant) -> None:
"""Abandoning the flow mid-pairing cancels the pair task and disconnects."""
entry = await _setup_account_entry(hass)
vehicle = _mock_vehicle(on_whitelist=False)
cancelled = asyncio.Event()
async def _pair() -> None:
try:
await asyncio.Event().wait()
except asyncio.CancelledError:
cancelled.set()
raise
vehicle.pair = AsyncMock(side_effect=_pair)
with (
patch(
"homeassistant.components.teslemetry.config_flow.async_discovered_service_info",
return_value=[_discovered_info()],
),
patch(
"homeassistant.components.teslemetry.config_flow.async_get_ble_parent",
return_value=_mock_ble_parent(vehicle),
),
):
result = await _start_pairing_at_scan(hass, entry)
result = await hass.config_entries.subentries.async_configure(
result["flow_id"], {}
)
result = await hass.config_entries.subentries.async_configure(
result["flow_id"], {}
)
assert result["type"] is FlowResultType.SHOW_PROGRESS
hass.config_entries.subentries.async_abort(result["flow_id"])
await hass.async_block_till_done()
assert cancelled.is_set()
vehicle.disconnect.assert_awaited_once()
# An abandoned pairing never creates a subentry.
assert not entry.get_subentries_of_type(SUBENTRY_TYPE_VEHICLE)
async def test_subentry_scan_device_not_found(hass: HomeAssistant) -> None:
"""The scan step re-shows the form with an error when no device is found."""
entry = await _setup_account_entry(hass)
with (
patch(
"homeassistant.components.teslemetry.config_flow.async_discovered_service_info",
return_value=[],
),
patch(
"homeassistant.components.teslemetry.config_flow.async_get_ble_parent",
return_value=MagicMock(),
),
):
result = await _start_pairing_at_scan(hass, entry)
result = await hass.config_entries.subentries.async_configure(
result["flow_id"], {}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "scan"
assert result["errors"] == {"base": "device_not_found"}
assert not entry.get_subentries_of_type(SUBENTRY_TYPE_VEHICLE)
@pytest.mark.parametrize(
"key_error",
[
pytest.param(OSError("disk gone"), id="os_error"),
pytest.param(ValueError("bad key"), id="value_error"),
# PrivateKeyError is the wrapped existing-key-file shape the scan step must recover from too.
pytest.param(
PrivateKeyError("malformed", "Not a valid PEM private key"),
id="private_key_error",
),
],
)
async def test_subentry_scan_key_load_recovers(
hass: HomeAssistant, key_error: Exception
) -> None:
"""A Bluetooth key-load failure re-shows the scan form; a loadable key then pairs."""
entry = await _setup_account_entry(hass)
vehicle = _mock_vehicle(on_whitelist=True)
with (
patch(
"homeassistant.components.teslemetry.config_flow.async_discovered_service_info",
return_value=[_discovered_info()],
),
patch(
"homeassistant.components.teslemetry.config_flow.async_get_ble_parent",
side_effect=[key_error, _mock_ble_parent(vehicle)],
) as mock_ble_parent,
patch.object(hass.config_entries, "async_schedule_reload"),
):
result = await _start_pairing_at_scan(hass, entry)
result = await hass.config_entries.subentries.async_configure(
result["flow_id"], {}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "scan"
assert result["errors"] == {"base": "cannot_connect"}
assert not entry.get_subentries_of_type(SUBENTRY_TYPE_VEHICLE)
result = await hass.config_entries.subentries.async_configure(
result["flow_id"], {}
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
subentries = entry.get_subentries_of_type(SUBENTRY_TYPE_VEHICLE)
assert len(subentries) == 1
assert subentries[0].data == {CONF_VIN: VIN, CONF_ADDRESS: ADDRESS}
# Both attempts ran the key load; only the second one loaded a usable key.
assert mock_ble_parent.call_count == 2
vehicle.connect.assert_awaited_once()
async def test_subentry_scan_finds_device_after_active_scan(
hass: HomeAssistant,
) -> None:
"""An awake in-range car only in scan responses is found via active scan."""
entry = await _setup_account_entry(hass)
vehicle = _mock_vehicle()
mock_discovered = MagicMock(return_value=[])
async def _active_scan(hass: HomeAssistant) -> None:
mock_discovered.return_value = [_discovered_info()]
with (
patch(
"homeassistant.components.teslemetry.config_flow.async_discovered_service_info",
mock_discovered,
),
patch(
"homeassistant.components.teslemetry.config_flow.async_request_active_scan",
AsyncMock(side_effect=_active_scan),
) as mock_active_scan,
patch(
"homeassistant.components.teslemetry.config_flow.async_get_ble_parent",
return_value=_mock_ble_parent(vehicle),
),
patch.object(hass.config_entries, "async_schedule_reload"),
):
result = await _start_pairing_at_scan(hass, entry)
result = await hass.config_entries.subentries.async_configure(
result["flow_id"], {}
)
await hass.async_block_till_done()
mock_active_scan.assert_awaited_once()
assert result["type"] is FlowResultType.CREATE_ENTRY
subentries = entry.get_subentries_of_type(SUBENTRY_TYPE_VEHICLE)
assert len(subentries) == 1
assert subentries[0].data == {CONF_VIN: VIN, CONF_ADDRESS: ADDRESS}
vehicle.connect.assert_awaited_once()
async def test_subentry_add_flow_keeps_device_on_parent(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
) -> None:
"""The add flow pairs an account vehicle without moving its device off the parent entry."""
entry = mock_config_entry()
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
# No Bluetooth subentry exists until the user adds one.
assert not entry.get_subentries_of_type(SUBENTRY_TYPE_VEHICLE)
existing_device = device_registry.async_get_device_by_identifier(
(DOMAIN, VIN), entry.entry_id
)
assert existing_device is not None
# The device and its entities start on the parent entry, owned by no subentry.
assert existing_device.config_subentry_id is None
vehicle_entities = er.async_entries_for_device(
entity_registry, existing_device.id, include_disabled_entities=True
)
assert vehicle_entities
assert all(entity.config_subentry_id is None for entity in vehicle_entities)
vehicle = _mock_vehicle(on_whitelist=True)
result = await hass.config_entries.subentries.async_init(
(entry.entry_id, SUBENTRY_TYPE_VEHICLE),
context={"source": "user"},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
# async_schedule_reload is left unpatched so the real reload runs here with the
# committed BLE address; keep the setup-time Bluetooth mocks active so it neither
# writes the vehicle key file nor opens a real connection.
with (
patch(
"homeassistant.components.teslemetry.config_flow.async_discovered_service_info",
return_value=[_discovered_info()],
),
patch(
"homeassistant.components.teslemetry.config_flow.async_get_ble_parent",
return_value=_mock_ble_parent(vehicle),
),
patch(
"homeassistant.components.teslemetry.async_ble_device_from_address",
return_value=MagicMock(),
),
patch(
"homeassistant.components.teslemetry.helpers.TeslaBluetooth"
) as mock_parent,
):
mock_parent.return_value.get_private_key = AsyncMock()
mock_parent.return_value.vehicles.createBluetooth.return_value = MagicMock()
result = await hass.config_entries.subentries.async_configure(
result["flow_id"], {CONF_VIN: VIN}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "scan"
result = await hass.config_entries.subentries.async_configure(
result["flow_id"], {}
)
# The subentry commits after the flow step returns; its change listener
# then schedules the reload, which runs to completion here.
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
subentries = entry.get_subentries_of_type(SUBENTRY_TYPE_VEHICLE)
assert len(subentries) == 1
subentry = subentries[0]
assert subentry.unique_id == VIN
assert subentry.data == {CONF_VIN: VIN, CONF_ADDRESS: ADDRESS}
# The real reload picked up the stored address: the reloaded vehicle now
# routes over Bluetooth instead of staying cloud-only.
assert isinstance(entry.runtime_data.vehicles[0].api, VehicleRouter)
# The pairing reuses the vehicle's existing device, never a duplicate.
bound_device = device_registry.async_get_device_by_identifier(
(DOMAIN, VIN), entry.entry_id
)
assert bound_device is not None
# The same device ID is kept and it stays on the parent entry, not the
# subentry, so removing the pairing never deletes the cloud vehicle.
assert bound_device.id == existing_device.id
assert bound_device.config_subentry_id is None
# The vehicle entities keep their unique IDs and stay on the parent entry.
bound_entities = er.async_entries_for_device(
entity_registry, bound_device.id, include_disabled_entities=True
)
assert {entity.unique_id for entity in bound_entities} == {
entity.unique_id for entity in vehicle_entities
}
assert all(entity.config_subentry_id is None for entity in bound_entities)
async def test_subentry_add_flow_no_available_vehicles(hass: HomeAssistant) -> None:
"""The add flow aborts when every account vehicle is already added."""
entry = await _setup_paired_entry(hass)
result = await hass.config_entries.subentries.async_init(
(entry.entry_id, SUBENTRY_TYPE_VEHICLE),
context={"source": "user"},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "no_vehicles"
async def test_subentry_add_flow_entry_not_loaded(hass: HomeAssistant) -> None:
"""The add flow aborts when the parent entry is not loaded."""
entry = mock_config_entry()
entry.add_to_hass(hass)
assert entry.state is ConfigEntryState.NOT_LOADED
result = await hass.config_entries.subentries.async_init(
(entry.entry_id, SUBENTRY_TYPE_VEHICLE),
context={"source": "user"},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "entry_not_loaded"
SITE_ID = 123456
WALL_CONNECTOR_SITE_ID = 555555
HOST = "192.168.91.1"
@@ -1458,18 +2183,18 @@ async def test_pair_step_second_lookup_errors(
ValueError,
id="key_read_valueerror",
),
# An encrypted key PEM surfaces as TypeError from the cryptography loader.
# An encrypted existing key file surfaces as PrivateKeyError("encrypted").
pytest.param(
"homeassistant.components.teslemetry.config_flow.Teslemetry.get_rsa_private_key",
TypeError,
id="key_fetch_typeerror",
PrivateKeyError("encrypted", "Private key file is encrypted"),
id="key_fetch_private_key_error",
),
],
)
async def test_rsa_key_load_failure_aborts(
hass: HomeAssistant,
patch_target: str,
error: type[Exception],
error: type[Exception] | Exception,
) -> None:
"""A failure loading the integration's RSA key aborts site preparation."""
entry = await _setup_account_no_subentry(hass)
tests/components/teslemetry/test_init.py
+620 -47
View File
@@ -1,5 +1,8 @@
"""Test the Teslemetry init."""
import asyncio
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from copy import deepcopy
import logging
import time
@@ -8,23 +11,28 @@ from unittest.mock import AsyncMock, MagicMock, patch
from aiohttp import ClientResponseError
from aiopowerwall import PowerwallError
from bleak.exc import BleakError
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
from tesla_fleet_api.exceptions import (
BluetoothCommandFailed,
BluetoothTransportError,
BluetoothUnconfirmedCommand,
Forbidden,
InsufficientCredits,
InvalidResponse,
InvalidToken,
LoginRequired,
PrivateKeyError,
RateLimited,
SubscriptionRequired,
TeslaFleetError,
)
from tesla_fleet_api.tesla import EnergySiteRouter
from tesla_fleet_api.teslemetry import EnergySite
from tesla_fleet_api.tesla import EnergySiteRouter, VehicleRouter
from tesla_fleet_api.teslemetry import EnergySite, Vehicle
from teslemetry_stream import TeslemetryStreamAuthenticationError
from homeassistant.components.teslemetry import (
@@ -35,8 +43,10 @@ from homeassistant.components.teslemetry import (
from homeassistant.components.teslemetry.const import (
CLIENT_ID,
CONF_SITE_ID,
CONF_VIN,
DOMAIN,
SUBENTRY_TYPE_ENERGY_SITE,
SUBENTRY_TYPE_VEHICLE,
)
# Coordinator constants
@@ -46,6 +56,7 @@ from homeassistant.components.teslemetry.coordinator import (
METADATA_INTERVAL,
VEHICLE_INTERVAL,
)
from homeassistant.components.teslemetry.helpers import async_get_ble_parent
from homeassistant.components.teslemetry.models import TeslemetryData
from homeassistant.components.teslemetry.oauth import TeslemetryImplementation
from homeassistant.config_entries import (
@@ -54,6 +65,7 @@ from homeassistant.config_entries import (
ConfigSubentryData,
)
from homeassistant.const import (
CONF_ADDRESS,
CONF_HOST,
CONF_PASSWORD,
STATE_OFF,
@@ -882,11 +894,7 @@ async def test_vehicle_polling_stops_when_all_entities_disabled(
keep_one_enabled: bool,
expected_polled: bool,
) -> None:
"""Test the vehicle coordinator stops polling once every entity is disabled.
With no listeners left, core unschedules the coordinator so the charged
vehicle_data poll stops entirely; a single enabled entity keeps it running.
"""
"""Test the vehicle coordinator stops polling once every entity is disabled."""
vin = "LRW3F7EK4NC700000"
entry = await setup_platform(hass, [Platform.SENSOR])
@@ -1221,13 +1229,7 @@ def _oauth_session(hass: HomeAssistant, entry: MockConfigEntry) -> OAuth2Session
async def test_get_access_token_dead_token_during_setup_triggers_auth_failed(
hass: HomeAssistant,
) -> None:
"""A dead/revoked refresh token during setup must raise ConfigEntryAuthFailed.
OAuth servers commonly report a dead refresh token with a non-401 status
(e.g. 400 invalid_grant). Only recognizing status 401 let this fall
through to ConfigEntryNotReady, which retries setup indefinitely without
ever prompting the user to reauthenticate.
"""
"""A dead/revoked refresh token during setup must raise ConfigEntryAuthFailed."""
mock_entry = mock_config_entry()
mock_entry.add_to_hass(hass)
mock_entry.mock_state(hass, ConfigEntryState.SETUP_IN_PROGRESS)
@@ -1271,10 +1273,7 @@ async def test_get_access_token_rate_limited_during_setup_is_not_fatal(
async def test_get_access_token_dead_token_after_setup_starts_reauth(
hass: HomeAssistant,
) -> None:
"""Test a token dying after setup (re)starts reauth without tearing down.
The coordinator handles the rest once the exception is re-raised.
"""
"""Test a token dying after setup (re)starts reauth without tearing down."""
mock_entry = mock_config_entry()
mock_entry.add_to_hass(hass)
mock_entry.mock_state(hass, ConfigEntryState.LOADED)
@@ -1428,12 +1427,7 @@ async def test_energy_site_cloud_without_powerwall(hass: HomeAssistant) -> None:
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.
"""
"""A subentry that exists but is not yet paired resolves to the cloud API."""
entry = mock_config_entry()
paired = MockConfigEntry(
domain=entry.domain,
@@ -1480,6 +1474,10 @@ async def test_no_subentry_created_at_setup(hass: HomeAssistant) -> None:
pytest.param(OSError("disk gone"), id="os_error"),
pytest.param(ValueError("bad key"), id="value_error"),
pytest.param(PowerwallError("client boom"), id="powerwall_error"),
pytest.param(
PrivateKeyError("malformed", "Not a valid PEM private key"),
id="private_key_error",
),
],
)
async def test_local_control_failure_falls_back_to_cloud(
@@ -1487,12 +1485,7 @@ async def test_local_control_failure_falls_back_to_cloud(
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.
"""
"""A failure resolving a paired site's local gateway falls back to cloud."""
entry = _entry_with_powerwall()
entry.add_to_hass(hass)
@@ -1519,20 +1512,29 @@ async def test_local_control_failure_falls_back_to_cloud(
)
async def test_local_control_encrypted_key_falls_back_to_cloud(
@pytest.mark.parametrize(
"rsa_key_error",
[
# PrivateKeyError is the wrapped existing-key-file failure shape.
pytest.param(
PrivateKeyError("encrypted", "Private key file is encrypted"),
id="private_key_error",
),
],
)
async def test_local_control_key_load_failure_falls_back_to_cloud(
hass: HomeAssistant,
rsa_key_error: Exception,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Fall back to cloud control when RSA key loading reports an encrypted PEM."""
"""Fall back to cloud control when RSA key loading fails on an existing file."""
entry = _entry_with_powerwall()
entry.add_to_hass(hass)
with (
patch(
"homeassistant.components.teslemetry.Teslemetry.get_rsa_private_key",
side_effect=TypeError(
"Password was not given but private key is encrypted"
),
side_effect=rsa_key_error,
),
patch("homeassistant.components.teslemetry.PLATFORMS", []),
caplog.at_level(logging.WARNING),
@@ -1551,13 +1553,7 @@ async def test_local_control_encrypted_key_falls_back_to_cloud(
async def test_local_control_unexpected_typeerror_is_not_swallowed(
hass: HomeAssistant,
) -> None:
"""A TypeError outside the key load is a real bug and must not degrade silently.
``_LOCAL_CONTROL_ERRORS`` deliberately excludes TypeError: only the key
loader's encrypted-PEM TypeError is converted to ValueError. A TypeError
from anywhere else in the resolve path (here, client construction) must
fail setup rather than silently falling back to cloud control.
"""
"""A TypeError outside the key load is a real bug and must not degrade silently."""
entry = _entry_with_powerwall()
entry.add_to_hass(hass)
@@ -1763,11 +1759,7 @@ async def test_stale_cleanup_preserves_pairing_without_energy_scope(
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.
"""
"""An entry update that only changes token data must not reload the entry."""
entry = mock_config_entry()
entry.add_to_hass(hass)
with patch("homeassistant.components.teslemetry.PLATFORMS", []):
@@ -1921,3 +1913,584 @@ async def test_energy_stream_disconnect_marks_unavailable_and_recovers(
for flow in hass.config_entries.flow.async_progress()
if flow["handler"] == DOMAIN
]
VIN = "LRW3F7EK4NC700000"
ADDRESS = "AA:BB:CC:DD:EE:FF"
CLOUD_RESULT = {"response": {"result": True, "reason": "cloud"}}
BLE_RESULT = {"response": {"result": True, "reason": "bluetooth"}}
def _entry_with_ble() -> MockConfigEntry:
"""Return a config entry whose vehicle subentry is already BLE-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_VEHICLE,
unique_id=VIN,
title="Test",
data={CONF_VIN: VIN, CONF_ADDRESS: ADDRESS},
)
],
)
async def test_vehicle_router_with_bluetooth(hass: HomeAssistant) -> None:
"""A BLE-paired vehicle wraps its cloud API in a VehicleRouter."""
entry = _entry_with_ble()
entry.add_to_hass(hass)
with (
patch(
"homeassistant.components.teslemetry.async_ble_device_from_address",
return_value=MagicMock(),
),
patch(
"homeassistant.components.teslemetry.helpers.TeslaBluetooth"
) as mock_parent,
patch("homeassistant.components.teslemetry.PLATFORMS", []),
):
mock_parent.return_value.get_private_key = AsyncMock()
mock_parent.return_value.vehicles.createBluetooth.return_value = MagicMock()
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
vehicle = entry.runtime_data.vehicles[0]
assert isinstance(vehicle.api, VehicleRouter)
# Avoid replaying ambiguous commands or keeping the vehicle awake.
mock_parent.return_value.vehicles.createBluetooth.assert_called_once_with(
VIN,
confirmation="verify",
raise_unconfirmed=False,
keepalive_interval=None,
)
async def test_vehicle_cloud_without_bluetooth(hass: HomeAssistant) -> None:
"""A vehicle without a paired address 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()
vehicle = entry.runtime_data.vehicles[0]
assert isinstance(vehicle.api, Vehicle)
assert not isinstance(vehicle.api, VehicleRouter)
@pytest.mark.parametrize(
"key_error",
[
pytest.param(OSError("disk gone"), id="os_error"),
pytest.param(ValueError("bad key"), id="value_error"),
# A raw TypeError only escapes the key create/generation path now.
pytest.param(
TypeError("unexpected keyword argument"),
id="typeerror",
),
# PrivateKeyError is the wrapped existing-key-file shape; it must degrade too.
pytest.param(
PrivateKeyError("unreadable", "Could not read private key file"),
id="private_key_unreadable",
),
pytest.param(
PrivateKeyError("malformed", "Not a valid PEM private key"),
id="private_key_malformed",
),
pytest.param(
PrivateKeyError("encrypted", "Private key file is encrypted"),
id="private_key_encrypted",
),
pytest.param(
PrivateKeyError("wrong_type", "Not an EllipticCurvePrivateKey"),
id="private_key_wrong_type",
),
],
)
async def test_vehicle_bluetooth_key_load_falls_back_to_cloud(
hass: HomeAssistant,
key_error: Exception,
caplog: pytest.LogCaptureFixture,
) -> None:
"""A vehicle whose Bluetooth key fails to load degrades to cloud control."""
entry = _entry_with_ble()
entry.add_to_hass(hass)
with (
patch(
"homeassistant.components.teslemetry.async_ble_device_from_address",
return_value=MagicMock(),
),
patch(
"homeassistant.components.teslemetry.helpers.TeslaBluetooth"
) as mock_parent,
patch("homeassistant.components.teslemetry.PLATFORMS", []),
caplog.at_level(logging.WARNING),
):
mock_parent.return_value.get_private_key = AsyncMock(side_effect=key_error)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.LOADED
vehicle = entry.runtime_data.vehicles[0]
assert isinstance(vehicle.api, Vehicle)
assert not isinstance(vehicle.api, VehicleRouter)
# The rest of the account is unaffected: the energy site still loads.
assert len(entry.runtime_data.energysites) == 1
assert "falling back to cloud control" in caplog.text
assert any(
record.levelname == "WARNING" and VIN in record.message
for record in caplog.records
)
async def test_vehicle_bluetooth_key_load_recovers_on_reload(
hass: HomeAssistant,
caplog: pytest.LogCaptureFixture,
) -> None:
"""A vehicle degraded to cloud by a key-load failure regains BLE control on reload."""
entry = _entry_with_ble()
entry.add_to_hass(hass)
with (
patch(
"homeassistant.components.teslemetry.async_ble_device_from_address",
return_value=MagicMock(),
),
patch(
"homeassistant.components.teslemetry.helpers.TeslaBluetooth"
) as mock_parent,
patch("homeassistant.components.teslemetry.PLATFORMS", []),
caplog.at_level(logging.WARNING),
):
mock_parent.return_value.get_private_key = AsyncMock(
side_effect=OSError("disk gone")
)
mock_parent.return_value.vehicles.createBluetooth.return_value = MagicMock()
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert not isinstance(entry.runtime_data.vehicles[0].api, VehicleRouter)
assert "falling back to cloud control" in caplog.text
# The key becomes readable again; a reload must restore local Bluetooth control.
mock_parent.return_value.get_private_key = AsyncMock()
caplog.clear()
await hass.config_entries.async_reload(entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.LOADED
assert isinstance(entry.runtime_data.vehicles[0].api, VehicleRouter)
assert "falling back to cloud control" not in caplog.text
@asynccontextmanager
async def _paired_entry(
hass: HomeAssistant, ble_lookup: MagicMock
) -> AsyncIterator[tuple[VehicleRouter, AsyncMock, AsyncMock]]:
"""Set up a BLE-paired entry, yielding its router and both backends."""
entry = _entry_with_ble()
entry.add_to_hass(hass)
bluetooth_vehicle = AsyncMock()
bluetooth_vehicle.set_device = MagicMock()
with (
patch(
"homeassistant.components.teslemetry.async_ble_device_from_address",
ble_lookup,
),
patch(
"homeassistant.components.teslemetry.helpers.TeslaBluetooth"
) as mock_parent,
patch("homeassistant.components.teslemetry.PLATFORMS", []),
):
mock_parent.return_value.get_private_key = AsyncMock()
mock_parent.return_value.vehicles.createBluetooth.return_value = (
bluetooth_vehicle
)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
router = entry.runtime_data.vehicles[0].api
cloud = AsyncMock(return_value=CLOUD_RESULT)
router.secondary.flash_lights = cloud
yield router, bluetooth_vehicle, cloud
async def test_vehicle_bluetooth_out_of_range(hass: HomeAssistant) -> None:
"""A paired vehicle out of range still gets a router, and skips Bluetooth."""
async with _paired_entry(hass, MagicMock(return_value=None)) as (
router,
bluetooth_vehicle,
cloud,
):
assert isinstance(router, VehicleRouter)
assert await router.flash_lights() == CLOUD_RESULT
cloud.assert_awaited_once()
bluetooth_vehicle.flash_lights.assert_not_called()
async def test_vehicle_router_resumes_bluetooth_when_vehicle_returns(
hass: HomeAssistant,
) -> None:
"""A vehicle away at setup routes locally again once it comes home."""
ble_lookup = MagicMock(return_value=None)
async with _paired_entry(hass, ble_lookup) as (router, bluetooth_vehicle, cloud):
bluetooth_vehicle.flash_lights.return_value = BLE_RESULT
assert await router.flash_lights() == CLOUD_RESULT
bluetooth_vehicle.flash_lights.assert_not_called()
ble_lookup.return_value = MagicMock()
assert await router.flash_lights() == BLE_RESULT
bluetooth_vehicle.flash_lights.assert_awaited_once()
cloud.assert_awaited_once()
async def test_vehicle_router_falls_back_when_vehicle_leaves(
hass: HomeAssistant,
) -> None:
"""A vehicle in range at setup routes to cloud once it drives away."""
ble_lookup = MagicMock(return_value=MagicMock())
async with _paired_entry(hass, ble_lookup) as (router, bluetooth_vehicle, cloud):
bluetooth_vehicle.flash_lights.return_value = BLE_RESULT
assert await router.flash_lights() == BLE_RESULT
cloud.assert_not_called()
ble_lookup.return_value = None
assert await router.flash_lights() == CLOUD_RESULT
cloud.assert_awaited_once()
bluetooth_vehicle.flash_lights.assert_awaited_once()
async def test_vehicle_router_refreshes_device_handle(hass: HomeAssistant) -> None:
"""Each command refreshes the BLE handle from the cache before connecting."""
first_device = MagicMock()
second_device = MagicMock()
ble_lookup = MagicMock(return_value=first_device)
async with _paired_entry(hass, ble_lookup) as (router, bluetooth_vehicle, _cloud):
await router.flash_lights()
bluetooth_vehicle.set_device.assert_called_once_with(first_device)
ble_lookup.return_value = second_device
await router.flash_lights()
bluetooth_vehicle.set_device.assert_called_with(second_device)
async def test_vehicle_router_fails_over_on_stale_cache_hit(
hass: HomeAssistant,
) -> None:
"""A cache entry outliving the vehicle costs one failed attempt, not a failure."""
async with _paired_entry(hass, MagicMock(return_value=MagicMock())) as (
router,
bluetooth_vehicle,
cloud,
):
bluetooth_vehicle.flash_lights.side_effect = BluetoothTransportError()
assert await router.flash_lights() == CLOUD_RESULT
bluetooth_vehicle.flash_lights.assert_awaited_once()
cloud.assert_awaited_once()
async def test_vehicle_paired_but_never_seen(hass: HomeAssistant) -> None:
"""A paired vehicle never seen by Bluetooth is built without a device handle."""
entry = _entry_with_ble()
entry.add_to_hass(hass)
with (
patch(
"homeassistant.components.teslemetry.async_ble_device_from_address",
MagicMock(return_value=None),
),
patch(
"homeassistant.components.teslemetry.helpers.TeslaBluetooth"
) as mock_parent,
patch("homeassistant.components.teslemetry.PLATFORMS", []),
):
mock_parent.return_value.get_private_key = AsyncMock()
mock_parent.return_value.vehicles.createBluetooth.return_value = AsyncMock()
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert (
"device"
not in mock_parent.return_value.vehicles.createBluetooth.call_args.kwargs
)
@pytest.mark.parametrize(
"disconnect_error",
[None, BleakError("boom")],
ids=["clean", "error_swallowed"],
)
async def test_unload_disconnects_bluetooth(
hass: HomeAssistant, disconnect_error: Exception | None
) -> None:
"""Unloading a routed entry disconnects its Bluetooth backend, errors and all."""
entry = _entry_with_ble()
entry.add_to_hass(hass)
bluetooth_vehicle = AsyncMock()
bluetooth_vehicle.disconnect = AsyncMock(side_effect=disconnect_error)
with (
patch(
"homeassistant.components.teslemetry.async_ble_device_from_address",
return_value=MagicMock(),
),
patch(
"homeassistant.components.teslemetry.helpers.TeslaBluetooth"
) as mock_parent,
patch("homeassistant.components.teslemetry.PLATFORMS", []),
):
mock_parent.return_value.get_private_key = AsyncMock()
mock_parent.return_value.vehicles.createBluetooth.return_value = (
bluetooth_vehicle
)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert isinstance(entry.runtime_data.vehicles[0].api, VehicleRouter)
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()
bluetooth_vehicle.disconnect.assert_awaited_once()
async def test_unload_never_connected_bluetooth(hass: HomeAssistant) -> None:
"""Unloading a paired vehicle that was never in range does not raise."""
entry = _entry_with_ble()
entry.add_to_hass(hass)
bluetooth_vehicle = AsyncMock()
with (
patch(
"homeassistant.components.teslemetry.async_ble_device_from_address",
return_value=None,
),
patch(
"homeassistant.components.teslemetry.helpers.TeslaBluetooth"
) as mock_parent,
patch("homeassistant.components.teslemetry.PLATFORMS", []),
):
mock_parent.return_value.get_private_key = AsyncMock()
mock_parent.return_value.vehicles.createBluetooth.return_value = (
bluetooth_vehicle
)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()
bluetooth_vehicle.disconnect.assert_awaited_once()
async def test_ble_parent_shared_and_cached(hass: HomeAssistant) -> None:
"""The BLE parent (holding the private key) is created once and reused."""
with patch(
"homeassistant.components.teslemetry.helpers.TeslaBluetooth"
) as mock_parent:
mock_parent.return_value.get_private_key = AsyncMock()
first = await async_get_ble_parent(hass)
second = await async_get_ble_parent(hass)
assert first is second
mock_parent.assert_called_once()
mock_parent.return_value.get_private_key.assert_awaited_once()
async def test_ble_parent_concurrent_first_init(hass: HomeAssistant) -> None:
"""Concurrent first-time callers still create and load the key exactly once."""
async def _get_private_key(path: str) -> None:
await asyncio.sleep(0)
with patch(
"homeassistant.components.teslemetry.helpers.TeslaBluetooth"
) as mock_parent:
mock_parent.return_value.get_private_key = AsyncMock(
side_effect=_get_private_key
)
parents = await asyncio.gather(*(async_get_ble_parent(hass) for _ in range(5)))
assert all(parent is parents[0] for parent in parents)
mock_parent.assert_called_once()
mock_parent.return_value.get_private_key.assert_awaited_once()
async def test_router_does_not_fail_over_on_unconfirmed() -> None:
"""An unconfirmed BLE command is never replayed on the cloud backend."""
bluetooth = AsyncMock()
bluetooth.actuate_trunk = AsyncMock(side_effect=BluetoothUnconfirmedCommand())
cloud = AsyncMock()
cloud.actuate_trunk = AsyncMock(return_value={"response": {"result": True}})
router = VehicleRouter(bluetooth, cloud)
with pytest.raises(BluetoothUnconfirmedCommand):
await router.actuate_trunk()
cloud.actuate_trunk.assert_not_called()
async def test_router_fails_over_on_command_failed() -> None:
"""A command proven not to have applied over BLE fails over to the cloud."""
bluetooth = AsyncMock()
bluetooth.actuate_trunk = AsyncMock(side_effect=BluetoothCommandFailed())
cloud = AsyncMock()
cloud.actuate_trunk = AsyncMock(return_value={"response": {"result": True}})
router = VehicleRouter(bluetooth, cloud)
result = await router.actuate_trunk()
assert result == {"response": {"result": True}}
bluetooth.actuate_trunk.assert_awaited_once()
cloud.actuate_trunk.assert_awaited_once()
async def _setup_paired_entry(hass: HomeAssistant) -> MockConfigEntry:
"""Set up an entry whose only account vehicle is already BLE-paired."""
entry = _entry_with_ble()
entry.add_to_hass(hass)
with (
patch(
"homeassistant.components.teslemetry.async_ble_device_from_address",
return_value=None,
),
patch(
"homeassistant.components.teslemetry.helpers.TeslaBluetooth"
) as mock_parent,
patch("homeassistant.components.teslemetry.PLATFORMS", []),
):
mock_parent.return_value.get_private_key = AsyncMock()
mock_parent.return_value.vehicles.createBluetooth.return_value = AsyncMock()
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
return entry
async def test_subentry_removal_reloads(hass: HomeAssistant) -> None:
"""Removing a vehicle subentry reloads once; later updates do not re-schedule."""
entry = await _setup_paired_entry(hass)
subentry_id = entry.get_subentries_of_type(SUBENTRY_TYPE_VEHICLE)[0].subentry_id
with patch.object(hass.config_entries, "async_schedule_reload") as mock_reload:
assert hass.config_entries.async_remove_subentry(entry, subentry_id)
await hass.async_block_till_done()
# A later entry update before the reload runs must not re-schedule it.
hass.config_entries.async_update_entry(
entry, data={**entry.data, "marker": True}
)
await hass.async_block_till_done()
mock_reload.assert_called_once_with(entry.entry_id)
async def test_subentry_removal_keeps_vehicle_device_and_entities(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
) -> None:
"""Removing a vehicle subentry leaves the cloud vehicle device and entities intact."""
entry = _entry_with_ble()
entry.add_to_hass(hass)
with (
patch(
"homeassistant.components.teslemetry.async_ble_device_from_address",
return_value=None,
),
patch(
"homeassistant.components.teslemetry.helpers.TeslaBluetooth"
) as mock_parent,
):
mock_parent.return_value.get_private_key = AsyncMock()
mock_parent.return_value.vehicles.createBluetooth.return_value = AsyncMock()
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
subentry_id = entry.get_subentries_of_type(SUBENTRY_TYPE_VEHICLE)[0].subentry_id
device = device_registry.async_get_device_by_identifier(
(DOMAIN, VIN), entry.entry_id
)
assert device is not None
# The device and its entities belong to the parent entry, never the subentry.
assert device.config_subentry_id is None
entities_before = er.async_entries_for_device(
entity_registry, device.id, include_disabled_entities=True
)
assert entities_before
assert all(entity.config_subentry_id is None for entity in entities_before)
unique_ids_before = {entity.unique_id for entity in entities_before}
# Patch the reload so only the subentry removal itself is exercised here.
with patch.object(hass.config_entries, "async_schedule_reload"):
assert hass.config_entries.async_remove_subentry(entry, subentry_id)
await hass.async_block_till_done()
# The vehicle device and every entity on it survive the removal.
device_after = device_registry.async_get_device_by_identifier(
(DOMAIN, VIN), entry.entry_id
)
assert device_after is not None
assert device_after.id == device.id
entities_after = er.async_entries_for_device(
entity_registry, device_after.id, include_disabled_entities=True
)
assert {entity.unique_id for entity in entities_after} == unique_ids_before
async def test_no_subentry_auto_created_at_setup(hass: HomeAssistant) -> None:
"""Setup never auto-creates a Bluetooth subentry for account vehicles."""
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()
assert not entry.get_subentries_of_type(SUBENTRY_TYPE_VEHICLE)
async def test_user_subentry_persists_across_reload(hass: HomeAssistant) -> None:
"""A paired vehicle subentry survives a reload even if its vehicle leaves the account."""
entry = await _setup_paired_entry(hass)
subentry_id = entry.get_subentries_of_type(SUBENTRY_TYPE_VEHICLE)[0].subentry_id
# The vehicle drops off the account, so setup builds no vehicle for it, yet
# the user-added subentry (with its stored credentials) must not be removed.
with (
patch(
"tesla_fleet_api.teslemetry.Teslemetry.products",
return_value={"response": []},
),
patch("homeassistant.components.teslemetry.PLATFORMS", []),
):
await hass.config_entries.async_reload(entry.entry_id)
await hass.async_block_till_done()
subentries = entry.get_subentries_of_type(SUBENTRY_TYPE_VEHICLE)
assert len(subentries) == 1
assert subentries[0].subentry_id == subentry_id
assert subentries[0].data == {CONF_VIN: VIN, CONF_ADDRESS: ADDRESS}