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

Add Tesla Fleet integration (#122019)

Browse source 
* Add Tesla Fleet

* Remove debug

* Improvements

* Fix refresh and stage tests

* Working oauth flow test

* Config Flow tests

* Fixes

* fixes

* Remove comment

Co-authored-by: G Johansson <goran.johansson@shiftit.se>

* Remove TYPE_CHECKING

* Add more tests

* More tests

* More tests

* revert teslemetry change

---------

Co-authored-by: G Johansson <goran.johansson@shiftit.se>
This commit is contained in:
Brett Adams 2024-07-19 17:05:27 +10:00 • committed by GitHub
parent 474e8b7a43
commit a2c2488c8b
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
31 changed files with 6887 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
CODEOWNERS
View file

@ -1432,6 +1432,8 @@ build.json @home-assistant/supervisor
/tests/components/tellduslive/ @fredrike
/homeassistant/components/template/ @PhracturedBlue @tetienne @home-assistant/core
/tests/components/template/ @PhracturedBlue @tetienne @home-assistant/core
/homeassistant/components/tesla_fleet/ @Bre77
/tests/components/tesla_fleet/ @Bre77
/homeassistant/components/tesla_wall_connector/ @einarhauks
/tests/components/tesla_wall_connector/ @einarhauks
/homeassistant/components/teslemetry/ @Bre77

2
homeassistant/brands/tesla.json
View file

@ -1,5 +1,5 @@
{
"domain": "tesla",
"name": "Tesla",
"integrations": ["powerwall", "tesla_wall_connector"]
"integrations": ["powerwall", "tesla_wall_connector", "tesla_fleet"]
}

169
homeassistant/components/tesla_fleet/__init__.py Normal file
View file

@ -0,0 +1,169 @@
"""Tesla Fleet integration."""
import asyncio
from typing import Final
import jwt
from tesla_fleet_api import EnergySpecific, TeslaFleetApi, VehicleSpecific
from tesla_fleet_api.const import Scope
from tesla_fleet_api.exceptions import (
InvalidToken,
LoginRequired,
OAuthExpired,
TeslaFleetError,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_ACCESS_TOKEN, CONF_TOKEN, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.config_entry_oauth2_flow import (
OAuth2Session,
async_get_config_entry_implementation,
)
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.device_registry import DeviceInfo
from .const import DOMAIN, MODELS
from .coordinator import (
TeslaFleetEnergySiteInfoCoordinator,
TeslaFleetEnergySiteLiveCoordinator,
TeslaFleetVehicleDataCoordinator,
)
from .models import TeslaFleetData, TeslaFleetEnergyData, TeslaFleetVehicleData
PLATFORMS: Final = [Platform.SENSOR]
type TeslaFleetConfigEntry = ConfigEntry[TeslaFleetData]
CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
async def async_setup_entry(hass: HomeAssistant, entry: TeslaFleetConfigEntry) -> bool:
"""Set up TeslaFleet config."""
access_token = entry.data[CONF_TOKEN][CONF_ACCESS_TOKEN]
session = async_get_clientsession(hass)
token = jwt.decode(access_token, options={"verify_signature": False})
scopes = token["scp"]
region = token["ou_code"].lower()
implementation = await async_get_config_entry_implementation(hass, entry)
oauth_session = OAuth2Session(hass, entry, implementation)
refresh_lock = asyncio.Lock()
async def _refresh_token() -> str:
async with refresh_lock:
await oauth_session.async_ensure_token_valid()
token: str = oauth_session.token[CONF_ACCESS_TOKEN]
return token
# Create API connection
tesla = TeslaFleetApi(
session=session,
access_token=access_token,
region=region,
charging_scope=False,
partner_scope=False,
user_scope=False,
energy_scope=Scope.ENERGY_DEVICE_DATA in scopes,
vehicle_scope=Scope.VEHICLE_DEVICE_DATA in scopes,
refresh_hook=_refresh_token,
)
try:
products = (await tesla.products())["response"]
except (InvalidToken, OAuthExpired, LoginRequired) as e:
raise ConfigEntryAuthFailed from e
except TeslaFleetError as e:
raise ConfigEntryNotReady from e
device_registry = dr.async_get(hass)
# Create array of classes
vehicles: list[TeslaFleetVehicleData] = []
energysites: list[TeslaFleetEnergyData] = []
for product in products:
if "vin" in product and tesla.vehicle:
# Remove the protobuff 'cached_data' that we do not use to save memory
product.pop("cached_data", None)
vin = product["vin"]
api = VehicleSpecific(tesla.vehicle, vin)
coordinator = TeslaFleetVehicleDataCoordinator(hass, api, product)
await coordinator.async_config_entry_first_refresh()
device = DeviceInfo(
identifiers={(DOMAIN, vin)},
manufacturer="Tesla",
name=product["display_name"],
model=MODELS.get(vin[3]),
serial_number=vin,
)
vehicles.append(
TeslaFleetVehicleData(
api=api,
coordinator=coordinator,
vin=vin,
device=device,
)
)
elif "energy_site_id" in product and tesla.energy:
site_id = product["energy_site_id"]
api = EnergySpecific(tesla.energy, site_id)
live_coordinator = TeslaFleetEnergySiteLiveCoordinator(hass, api)
info_coordinator = TeslaFleetEnergySiteInfoCoordinator(hass, api, product)
await live_coordinator.async_config_entry_first_refresh()
await info_coordinator.async_config_entry_first_refresh()
# Create energy site model
model = None
models = set()
for gateway in info_coordinator.data.get("components_gateways", []):
if gateway.get("part_name"):
models.add(gateway["part_name"])
for battery in info_coordinator.data.get("components_batteries", []):
if battery.get("part_name"):
models.add(battery["part_name"])
if models:
model = ", ".join(sorted(models))
device = DeviceInfo(
identifiers={(DOMAIN, str(site_id))},
manufacturer="Tesla",
name=product.get("site_name", "Energy Site"),
model=model,
serial_number=str(site_id),
)
# Create the energy site device regardless of it having entities
# This is so users with a Wall Connector but without a Powerwall can still make service calls
device_registry.async_get_or_create(
config_entry_id=entry.entry_id, **device
)
energysites.append(
TeslaFleetEnergyData(
api=api,
live_coordinator=live_coordinator,
info_coordinator=info_coordinator,
id=site_id,
device=device,
)
)
# Setup Platforms
entry.runtime_data = TeslaFleetData(vehicles, energysites, scopes)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: TeslaFleetConfigEntry) -> bool:
"""Unload TeslaFleet Config."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

71
homeassistant/components/tesla_fleet/application_credentials.py Normal file
View file

@ -0,0 +1,71 @@
"""Application Credentials platform the Tesla Fleet integration."""
import base64
import hashlib
import secrets
from typing import Any
from homeassistant.components.application_credentials import ClientCredential
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_entry_oauth2_flow
from .const import DOMAIN, SCOPES
CLIENT_ID = "71b813eb-4a2e-483a-b831-4dec5cb9bf0d"
AUTHORIZE_URL = "https://auth.tesla.com/oauth2/v3/authorize"
TOKEN_URL = "https://auth.tesla.com/oauth2/v3/token"
async def async_get_auth_implementation(
hass: HomeAssistant, auth_domain: str, credential: ClientCredential
) -> config_entry_oauth2_flow.AbstractOAuth2Implementation:
"""Return auth implementation."""
return TeslaOAuth2Implementation(
hass,
DOMAIN,
)
class TeslaOAuth2Implementation(config_entry_oauth2_flow.LocalOAuth2Implementation):
"""Tesla Fleet API Open Source Oauth2 implementation."""
_name = "Tesla Fleet API"
def __init__(self, hass: HomeAssistant, domain: str) -> None:
"""Initialize local auth implementation."""
self.hass = hass
self._domain = domain
# Setup PKCE
self.code_verifier = secrets.token_urlsafe(32)
hashed_verifier = hashlib.sha256(self.code_verifier.encode()).digest()
self.code_challenge = (
base64.urlsafe_b64encode(hashed_verifier).decode().replace("=", "")
)
super().__init__(
hass,
domain,
CLIENT_ID,
"", # Implementation has no client secret
AUTHORIZE_URL,
TOKEN_URL,
)
@property
def extra_authorize_data(self) -> dict[str, Any]:
"""Extra data that needs to be appended to the authorize url."""
return {
"scope": " ".join(SCOPES),
"code_challenge": self.code_challenge, # PKCE
}
async def async_resolve_external_data(self, external_data: Any) -> dict:
"""Resolve the authorization code to tokens."""
return await self._token_request(
{
"grant_type": "authorization_code",
"code": external_data["code"],
"redirect_uri": external_data["state"]["redirect_uri"],
"code_verifier": self.code_verifier, # PKCE
}
)

75
homeassistant/components/tesla_fleet/config_flow.py Normal file
View file

@ -0,0 +1,75 @@
"""Config Flow for Tesla Fleet integration."""
from __future__ import annotations
from collections.abc import Mapping
import logging
from typing import Any
import jwt
from homeassistant.config_entries import ConfigEntry, ConfigFlowResult
from homeassistant.helpers import config_entry_oauth2_flow
from .const import DOMAIN, LOGGER
class OAuth2FlowHandler(
config_entry_oauth2_flow.AbstractOAuth2FlowHandler, domain=DOMAIN
):
"""Config flow to handle Tesla Fleet API OAuth2 authentication."""
DOMAIN = DOMAIN
reauth_entry: ConfigEntry | None = None
@property
def logger(self) -> logging.Logger:
"""Return logger."""
return LOGGER
async def async_oauth_create_entry(
self,
data: dict[str, Any],
) -> ConfigFlowResult:
"""Handle the initial step."""
token = jwt.decode(
data["token"]["access_token"], options={"verify_signature": False}
)
uid = token["sub"]
if not self.reauth_entry:
await self.async_set_unique_id(uid)
self._abort_if_unique_id_configured()
return self.async_create_entry(title=uid, data=data)
if self.reauth_entry.unique_id == uid:
self.hass.config_entries.async_update_entry(
self.reauth_entry,
data=data,
)
await self.hass.config_entries.async_reload(self.reauth_entry.entry_id)
return self.async_abort(reason="reauth_successful")
return self.async_abort(
reason="reauth_account_mismatch",
description_placeholders={"title": self.reauth_entry.title},
)
async def async_step_reauth(
self, entry_data: Mapping[str, Any]
) -> ConfigFlowResult:
"""Perform reauth upon an API authentication error."""
self.reauth_entry = self.hass.config_entries.async_get_entry(
self.context["entry_id"]
)
return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Confirm reauth dialog."""
if user_input is None:
return self.async_show_form(step_id="reauth_confirm")
return await self.async_step_user()

39
homeassistant/components/tesla_fleet/const.py Normal file
View file

@ -0,0 +1,39 @@
"""Constants used by Tesla Fleet integration."""
from __future__ import annotations
from enum import StrEnum
import logging
from tesla_fleet_api.const import Scope
DOMAIN = "tesla_fleet"
CONF_REFRESH_TOKEN = "refresh_token"
LOGGER = logging.getLogger(__package__)
SCOPES = [
Scope.OPENID,
Scope.OFFLINE_ACCESS,
Scope.VEHICLE_DEVICE_DATA,
Scope.VEHICLE_CMDS,
Scope.VEHICLE_CHARGING_CMDS,
Scope.ENERGY_DEVICE_DATA,
Scope.ENERGY_CMDS,
]
MODELS = {
"S": "Model S",
"3": "Model 3",
"X": "Model X",
"Y": "Model Y",
}
class TeslaFleetState(StrEnum):
"""Teslemetry Vehicle States."""
ONLINE = "online"
ASLEEP = "asleep"
OFFLINE = "offline"

215
homeassistant/components/tesla_fleet/coordinator.py Normal file
View file

@ -0,0 +1,215 @@
"""Tesla Fleet Data Coordinator."""
from datetime import datetime, timedelta
from typing import Any
from tesla_fleet_api import EnergySpecific, VehicleSpecific
from tesla_fleet_api.const import VehicleDataEndpoint
from tesla_fleet_api.exceptions import (
InvalidToken,
LoginRequired,
OAuthExpired,
RateLimited,
TeslaFleetError,
VehicleOffline,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import LOGGER, TeslaFleetState
VEHICLE_INTERVAL_SECONDS = 120
VEHICLE_INTERVAL = timedelta(seconds=VEHICLE_INTERVAL_SECONDS)
VEHICLE_WAIT = timedelta(minutes=15)
ENERGY_INTERVAL_SECONDS = 60
ENERGY_INTERVAL = timedelta(seconds=ENERGY_INTERVAL_SECONDS)
ENDPOINTS = [
VehicleDataEndpoint.CHARGE_STATE,
VehicleDataEndpoint.CLIMATE_STATE,
VehicleDataEndpoint.DRIVE_STATE,
VehicleDataEndpoint.LOCATION_DATA,
VehicleDataEndpoint.VEHICLE_STATE,
VehicleDataEndpoint.VEHICLE_CONFIG,
]
def flatten(data: dict[str, Any], parent: str | None = None) -> dict[str, Any]:
"""Flatten the data structure."""
result = {}
for key, value in data.items():
if parent:
key = f"{parent}_{key}"
if isinstance(value, dict):
result.update(flatten(value, key))
else:
result[key] = value
return result
class TeslaFleetVehicleDataCoordinator(DataUpdateCoordinator[dict[str, Any]]):
"""Class to manage fetching data from the TeslaFleet API."""
updated_once: bool
pre2021: bool
last_active: datetime
def __init__(
self, hass: HomeAssistant, api: VehicleSpecific, product: dict
) -> None:
"""Initialize TeslaFleet Vehicle Update Coordinator."""
super().__init__(
hass,
LOGGER,
name="Tesla Fleet Vehicle",
update_interval=timedelta(seconds=5),
)
self.api = api
self.data = flatten(product)
self.updated_once = False
self.last_active = datetime.now()
async def _async_update_data(self) -> dict[str, Any]:
"""Update vehicle data using TeslaFleet API."""
self.update_interval = VEHICLE_INTERVAL
try:
# Check if the vehicle is awake using a non-rate limited API call
state = (await self.api.vehicle())["response"]
if state and state["state"] != TeslaFleetState.ONLINE:
self.data["state"] = state["state"]
return self.data
# This is a rated limited API call
data = (await self.api.vehicle_data(endpoints=ENDPOINTS))["response"]
except VehicleOffline:
self.data["state"] = TeslaFleetState.ASLEEP
return self.data
except RateLimited as e:
LOGGER.warning(
"%s rate limited, will retry in %s seconds",
self.name,
e.data.get("after"),
)
if "after" in e.data:
self.update_interval = timedelta(seconds=int(e.data["after"]))
return self.data
except (InvalidToken, OAuthExpired, LoginRequired) as e:
raise ConfigEntryAuthFailed from e
except TeslaFleetError as e:
raise UpdateFailed(e.message) from e
self.updated_once = True
if self.api.pre2021 and data["state"] == TeslaFleetState.ONLINE:
# Handle pre-2021 vehicles which cannot sleep by themselves
if (
data["charge_state"].get("charging_state") == "Charging"
or data["vehicle_state"].get("is_user_present")
or data["vehicle_state"].get("sentry_mode")
):
# Vehicle is active, reset timer
self.last_active = datetime.now()
else:
elapsed = datetime.now() - self.last_active
if elapsed > timedelta(minutes=20):
# Vehicle didn't sleep, try again in 15 minutes
self.last_active = datetime.now()
elif elapsed > timedelta(minutes=15):
# Let vehicle go to sleep now
self.update_interval = VEHICLE_WAIT
return flatten(data)
class TeslaFleetEnergySiteLiveCoordinator(DataUpdateCoordinator[dict[str, Any]]):
"""Class to manage fetching energy site live status from the TeslaFleet API."""
updated_once: bool
def __init__(self, hass: HomeAssistant, api: EnergySpecific) -> None:
"""Initialize TeslaFleet Energy Site Live coordinator."""
super().__init__(
hass,
LOGGER,
name="Tesla Fleet Energy Site Live",
update_interval=timedelta(seconds=10),
)
self.api = api
self.data = {}
self.updated_once = False
async def _async_update_data(self) -> dict[str, Any]:
"""Update energy site data using TeslaFleet API."""
self.update_interval = ENERGY_INTERVAL
try:
data = (await self.api.live_status())["response"]
except RateLimited as e:
LOGGER.warning(
"%s rate limited, will retry in %s seconds",
self.name,
e.data.get("after"),
)
if "after" in e.data:
self.update_interval = timedelta(seconds=int(e.data["after"]))
return self.data
except (InvalidToken, OAuthExpired, LoginRequired) as e:
raise ConfigEntryAuthFailed from e
except TeslaFleetError as e:
raise UpdateFailed(e.message) from e
# Convert Wall Connectors from array to dict
data["wall_connectors"] = {
wc["din"]: wc for wc in (data.get("wall_connectors") or [])
}
self.updated_once = True
return data
class TeslaFleetEnergySiteInfoCoordinator(DataUpdateCoordinator[dict[str, Any]]):
"""Class to manage fetching energy site info from the TeslaFleet API."""
updated_once: bool
def __init__(self, hass: HomeAssistant, api: EnergySpecific, product: dict) -> None:
"""Initialize TeslaFleet Energy Info coordinator."""
super().__init__(
hass,
LOGGER,
name="Tesla Fleet Energy Site Info",
update_interval=timedelta(seconds=15),
)
self.api = api
self.data = flatten(product)
self.updated_once = False
async def _async_update_data(self) -> dict[str, Any]:
"""Update energy site data using TeslaFleet API."""
self.update_interval = ENERGY_INTERVAL
try:
data = (await self.api.site_info())["response"]
except RateLimited as e:
LOGGER.warning(
"%s rate limited, will retry in %s seconds",
self.name,
e.data.get("after"),
)
if "after" in e.data:
self.update_interval = timedelta(seconds=int(e.data["after"]))
return self.data
except (InvalidToken, OAuthExpired, LoginRequired) as e:
raise ConfigEntryAuthFailed from e
except TeslaFleetError as e:
raise UpdateFailed(e.message) from e
self.updated_once = True
return flatten(data)

176
homeassistant/components/tesla_fleet/entity.py Normal file
View file

@ -0,0 +1,176 @@
"""Tesla Fleet parent entity class."""
from abc import abstractmethod
from typing import Any
from tesla_fleet_api import EnergySpecific, VehicleSpecific
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from .coordinator import (
TeslaFleetEnergySiteInfoCoordinator,
TeslaFleetEnergySiteLiveCoordinator,
TeslaFleetVehicleDataCoordinator,
)
from .models import TeslaFleetEnergyData, TeslaFleetVehicleData
class TeslaFleetEntity(
CoordinatorEntity[
TeslaFleetVehicleDataCoordinator
| TeslaFleetEnergySiteLiveCoordinator
| TeslaFleetEnergySiteInfoCoordinator
]
):
"""Parent class for all TeslaFleet entities."""
_attr_has_entity_name = True
def __init__(
self,
coordinator: TeslaFleetVehicleDataCoordinator
| TeslaFleetEnergySiteLiveCoordinator
| TeslaFleetEnergySiteInfoCoordinator,
api: VehicleSpecific | EnergySpecific,
key: str,
) -> None:
"""Initialize common aspects of a TeslaFleet entity."""
super().__init__(coordinator)
self.api = api
self.key = key
self._attr_translation_key = self.key
self._async_update_attrs()
@property
def available(self) -> bool:
"""Return if sensor is available."""
return self.coordinator.last_update_success and self._attr_available
@property
def _value(self) -> Any | None:
"""Return a specific value from coordinator data."""
return self.coordinator.data.get(self.key)
def get(self, key: str, default: Any | None = None) -> Any | None:
"""Return a specific value from coordinator data."""
return self.coordinator.data.get(key, default)
@property
def is_none(self) -> bool:
"""Return if the value is a literal None."""
return self.get(self.key, False) is None
@property
def has(self) -> bool:
"""Return True if a specific value is in coordinator data."""
return self.key in self.coordinator.data
def _handle_coordinator_update(self) -> None:
"""Handle updated data from the coordinator."""
self._async_update_attrs()
self.async_write_ha_state()
@abstractmethod
def _async_update_attrs(self) -> None:
"""Update the attributes of the entity."""
class TeslaFleetVehicleEntity(TeslaFleetEntity):
"""Parent class for TeslaFleet Vehicle entities."""
_last_update: int = 0
def __init__(
self,
data: TeslaFleetVehicleData,
key: str,
) -> None:
"""Initialize common aspects of a Tesla Fleet entity."""
self._attr_unique_id = f"{data.vin}-{key}"
self.vehicle = data
self._attr_device_info = data.device
super().__init__(data.coordinator, data.api, key)
@property
def _value(self) -> Any | None:
"""Return a specific value from coordinator data."""
return self.coordinator.data.get(self.key)
class TeslaFleetEnergyLiveEntity(TeslaFleetEntity):
"""Parent class for TeslaFleet Energy Site Live entities."""
def __init__(
self,
data: TeslaFleetEnergyData,
key: str,
) -> None:
"""Initialize common aspects of a Tesla Fleet Energy Site Live entity."""
self._attr_unique_id = f"{data.id}-{key}"
self._attr_device_info = data.device
super().__init__(data.live_coordinator, data.api, key)
class TeslaFleetEnergyInfoEntity(TeslaFleetEntity):
"""Parent class for TeslaFleet Energy Site Info entities."""
def __init__(
self,
data: TeslaFleetEnergyData,
key: str,
) -> None:
"""Initialize common aspects of a Tesla Fleet Energy Site Info entity."""
self._attr_unique_id = f"{data.id}-{key}"
self._attr_device_info = data.device
super().__init__(data.info_coordinator, data.api, key)
class TeslaFleetWallConnectorEntity(
TeslaFleetEntity, CoordinatorEntity[TeslaFleetEnergySiteLiveCoordinator]
):
"""Parent class for Tesla Fleet Wall Connector entities."""
_attr_has_entity_name = True
def __init__(
self,
data: TeslaFleetEnergyData,
din: str,
key: str,
) -> None:
"""Initialize common aspects of a Tesla Fleet entity."""
self.din = din
self._attr_unique_id = f"{data.id}-{din}-{key}"
# Find the model from the info coordinator
model: str | None = None
for wc in data.info_coordinator.data.get("components_wall_connectors", []):
if wc["din"] == din:
model = wc.get("part_name")
break
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, din)},
manufacturer="Tesla",
name="Wall Connector",
via_device=(DOMAIN, str(data.id)),
serial_number=din.split("-")[-1],
model=model,
)
super().__init__(data.live_coordinator, data.api, key)
@property
def _value(self) -> int:
"""Return a specific wall connector value from coordinator data."""
return (
self.coordinator.data.get("wall_connectors", {})
.get(self.din, {})
.get(self.key)
)

66
homeassistant/components/tesla_fleet/icons.json Normal file
View file

@ -0,0 +1,66 @@
{
"entity": {
"sensor": {
"battery_power": {
"default": "mdi:home-battery"
},
"charge_state_charging_state": {
"default": "mdi:ev-station",
"state": {
"disconnected": "mdi:connection",
"no_power": "mdi:power-plug-off-outline",
"starting": "mdi:play-circle",
"stopped": "mdi:stop-circle"
}
},
"drive_state_active_route_destination": {
"default": "mdi:routes"
},
"drive_state_active_route_minutes_to_arrival": {
"default": "mdi:routes-clock"
},
"drive_state_shift_state": {
"default": "mdi:car-shift-pattern",
"state": {
"d": "mdi:alpha-d",
"n": "mdi:alpha-n",
"p": "mdi:alpha-p",
"r": "mdi:alpha-r"
}
},
"energy_left": {
"default": "mdi:battery"
},
"generator_power": {
"default": "mdi:generator-stationary"
},
"grid_power": {
"default": "mdi:transmission-tower"
},
"grid_services_power": {
"default": "mdi:transmission-tower"
},
"load_power": {
"default": "mdi:power-plug"
},
"solar_power": {
"default": "mdi:solar-power"
},
"total_pack_energy": {
"default": "mdi:battery-high"
},
"vin": {
"default": "mdi:car-electric"
},
"wall_connector_fault_state": {
"default": "mdi:ev-station"
},
"wall_connector_power": {
"default": "mdi:ev-station"
},
"wall_connector_state": {
"default": "mdi:ev-station"
}
}
}
}

11
homeassistant/components/tesla_fleet/manifest.json Normal file
View file

@ -0,0 +1,11 @@
{
"domain": "tesla_fleet",
"name": "Tesla Fleet",
"codeowners": ["@Bre77"],
"config_flow": true,
"dependencies": ["application_credentials", "http"],
"documentation": "https://www.home-assistant.io/integrations/tesla_fleet",
"iot_class": "cloud_polling",
"loggers": ["tesla-fleet-api"],
"requirements": ["tesla-fleet-api==0.7.2"]
}

46
homeassistant/components/tesla_fleet/models.py Normal file
View file

@ -0,0 +1,46 @@
"""The Tesla Fleet integration models."""
from __future__ import annotations
from dataclasses import dataclass
from tesla_fleet_api import EnergySpecific, VehicleSpecific
from tesla_fleet_api.const import Scope
from homeassistant.helpers.device_registry import DeviceInfo
from .coordinator import (
TeslaFleetEnergySiteInfoCoordinator,
TeslaFleetEnergySiteLiveCoordinator,
TeslaFleetVehicleDataCoordinator,
)
@dataclass
class TeslaFleetData:
"""Data for the TeslaFleet integration."""
vehicles: list[TeslaFleetVehicleData]
energysites: list[TeslaFleetEnergyData]
scopes: list[Scope]
@dataclass
class TeslaFleetVehicleData:
"""Data for a vehicle in the TeslaFleet integration."""
api: VehicleSpecific
coordinator: TeslaFleetVehicleDataCoordinator
vin: str
device: DeviceInfo
@dataclass
class TeslaFleetEnergyData:
"""Data for a vehicle in the TeslaFleet integration."""
api: EnergySpecific
live_coordinator: TeslaFleetEnergySiteLiveCoordinator
info_coordinator: TeslaFleetEnergySiteInfoCoordinator
id: int
device: DeviceInfo

599
homeassistant/components/tesla_fleet/sensor.py Normal file
View file

@ -0,0 +1,599 @@
"""Sensor platform for Tesla Fleet integration."""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from datetime import timedelta
from itertools import chain
from typing import cast
from homeassistant.components.sensor import (
RestoreSensor,
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.const import (
PERCENTAGE,
EntityCategory,
UnitOfElectricCurrent,
UnitOfElectricPotential,
UnitOfEnergy,
UnitOfLength,
UnitOfPower,
UnitOfPressure,
UnitOfSpeed,
UnitOfTemperature,
UnitOfTime,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import StateType
from homeassistant.util import dt as dt_util
from homeassistant.util.variance import ignore_variance
from . import TeslaFleetConfigEntry
from .const import TeslaFleetState
from .entity import (
TeslaFleetEnergyInfoEntity,
TeslaFleetEnergyLiveEntity,
TeslaFleetVehicleEntity,
TeslaFleetWallConnectorEntity,
)
from .models import TeslaFleetEnergyData, TeslaFleetVehicleData
PARALLEL_UPDATES = 0
CHARGE_STATES = {
"Starting": "starting",
"Charging": "charging",
"Stopped": "stopped",
"Complete": "complete",
"Disconnected": "disconnected",
"NoPower": "no_power",
}
SHIFT_STATES = {"P": "p", "D": "d", "R": "r", "N": "n"}
@dataclass(frozen=True, kw_only=True)
class TeslaFleetSensorEntityDescription(SensorEntityDescription):
"""Describes Tesla Fleet Sensor entity."""
value_fn: Callable[[StateType], StateType] = lambda x: x
VEHICLE_DESCRIPTIONS: tuple[TeslaFleetSensorEntityDescription, ...] = (
TeslaFleetSensorEntityDescription(
key="charge_state_charging_state",
options=list(CHARGE_STATES.values()),
device_class=SensorDeviceClass.ENUM,
value_fn=lambda value: CHARGE_STATES.get(cast(str, value)),
),
TeslaFleetSensorEntityDescription(
key="charge_state_battery_level",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PERCENTAGE,
device_class=SensorDeviceClass.BATTERY,
),
TeslaFleetSensorEntityDescription(
key="charge_state_usable_battery_level",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PERCENTAGE,
device_class=SensorDeviceClass.BATTERY,
entity_registry_enabled_default=False,
),
TeslaFleetSensorEntityDescription(
key="charge_state_charge_energy_added",
state_class=SensorStateClass.TOTAL_INCREASING,
native_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
device_class=SensorDeviceClass.ENERGY,
suggested_display_precision=1,
),
TeslaFleetSensorEntityDescription(
key="charge_state_charger_power",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfPower.KILO_WATT,
device_class=SensorDeviceClass.POWER,
),
TeslaFleetSensorEntityDescription(
key="charge_state_charger_voltage",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
device_class=SensorDeviceClass.VOLTAGE,
entity_category=EntityCategory.DIAGNOSTIC,
),
TeslaFleetSensorEntityDescription(
key="charge_state_charger_actual_current",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
device_class=SensorDeviceClass.CURRENT,
entity_category=EntityCategory.DIAGNOSTIC,
),
TeslaFleetSensorEntityDescription(
key="charge_state_charge_rate",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfSpeed.MILES_PER_HOUR,
device_class=SensorDeviceClass.SPEED,
entity_category=EntityCategory.DIAGNOSTIC,
),
TeslaFleetSensorEntityDescription(
key="charge_state_conn_charge_cable",
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
),
TeslaFleetSensorEntityDescription(
key="charge_state_fast_charger_type",
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
),
TeslaFleetSensorEntityDescription(
key="charge_state_battery_range",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfLength.MILES,
device_class=SensorDeviceClass.DISTANCE,
suggested_display_precision=1,
),
TeslaFleetSensorEntityDescription(
key="charge_state_est_battery_range",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfLength.MILES,
device_class=SensorDeviceClass.DISTANCE,
suggested_display_precision=1,
entity_registry_enabled_default=False,
),
TeslaFleetSensorEntityDescription(
key="charge_state_ideal_battery_range",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfLength.MILES,
device_class=SensorDeviceClass.DISTANCE,
suggested_display_precision=1,
entity_registry_enabled_default=False,
),
TeslaFleetSensorEntityDescription(
key="drive_state_speed",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfSpeed.MILES_PER_HOUR,
device_class=SensorDeviceClass.SPEED,
entity_registry_enabled_default=False,
value_fn=lambda value: value or 0,
),
TeslaFleetSensorEntityDescription(
key="drive_state_power",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfPower.KILO_WATT,
device_class=SensorDeviceClass.POWER,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
value_fn=lambda value: value or 0,
),
TeslaFleetSensorEntityDescription(
key="drive_state_shift_state",
options=list(SHIFT_STATES.values()),
device_class=SensorDeviceClass.ENUM,
value_fn=lambda x: SHIFT_STATES.get(str(x), "p"),
entity_registry_enabled_default=False,
),
TeslaFleetSensorEntityDescription(
key="vehicle_state_odometer",
state_class=SensorStateClass.TOTAL_INCREASING,
native_unit_of_measurement=UnitOfLength.MILES,
device_class=SensorDeviceClass.DISTANCE,
suggested_display_precision=0,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
),
TeslaFleetSensorEntityDescription(
key="vehicle_state_tpms_pressure_fl",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfPressure.BAR,
suggested_unit_of_measurement=UnitOfPressure.PSI,
device_class=SensorDeviceClass.PRESSURE,
suggested_display_precision=1,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
),
TeslaFleetSensorEntityDescription(
key="vehicle_state_tpms_pressure_fr",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfPressure.BAR,
suggested_unit_of_measurement=UnitOfPressure.PSI,
device_class=SensorDeviceClass.PRESSURE,
suggested_display_precision=1,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
),
TeslaFleetSensorEntityDescription(
key="vehicle_state_tpms_pressure_rl",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfPressure.BAR,
suggested_unit_of_measurement=UnitOfPressure.PSI,
device_class=SensorDeviceClass.PRESSURE,
suggested_display_precision=1,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
),
TeslaFleetSensorEntityDescription(
key="vehicle_state_tpms_pressure_rr",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfPressure.BAR,
suggested_unit_of_measurement=UnitOfPressure.PSI,
device_class=SensorDeviceClass.PRESSURE,
suggested_display_precision=1,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
),
TeslaFleetSensorEntityDescription(
key="climate_state_inside_temp",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
device_class=SensorDeviceClass.TEMPERATURE,
suggested_display_precision=1,
),
TeslaFleetSensorEntityDescription(
key="climate_state_outside_temp",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
device_class=SensorDeviceClass.TEMPERATURE,
suggested_display_precision=1,
),
TeslaFleetSensorEntityDescription(
key="climate_state_driver_temp_setting",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
device_class=SensorDeviceClass.TEMPERATURE,
suggested_display_precision=1,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
),
TeslaFleetSensorEntityDescription(
key="climate_state_passenger_temp_setting",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
device_class=SensorDeviceClass.TEMPERATURE,
suggested_display_precision=1,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
),
TeslaFleetSensorEntityDescription(
key="drive_state_active_route_traffic_minutes_delay",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfTime.MINUTES,
device_class=SensorDeviceClass.DURATION,
entity_registry_enabled_default=False,
),
TeslaFleetSensorEntityDescription(
key="drive_state_active_route_energy_at_arrival",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PERCENTAGE,
device_class=SensorDeviceClass.BATTERY,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
),
TeslaFleetSensorEntityDescription(
key="drive_state_active_route_miles_to_arrival",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfLength.MILES,
device_class=SensorDeviceClass.DISTANCE,
),
)
@dataclass(frozen=True, kw_only=True)
class TeslaFleetTimeEntityDescription(SensorEntityDescription):
"""Describes Tesla Fleet Sensor entity."""
variance: int
VEHICLE_TIME_DESCRIPTIONS: tuple[TeslaFleetTimeEntityDescription, ...] = (
TeslaFleetTimeEntityDescription(
key="charge_state_minutes_to_full_charge",
device_class=SensorDeviceClass.TIMESTAMP,
entity_category=EntityCategory.DIAGNOSTIC,
variance=4,
),
TeslaFleetTimeEntityDescription(
key="drive_state_active_route_minutes_to_arrival",
device_class=SensorDeviceClass.TIMESTAMP,
variance=1,
),
)
ENERGY_LIVE_DESCRIPTIONS: tuple[SensorEntityDescription, ...] = (
SensorEntityDescription(
key="solar_power",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfPower.WATT,
suggested_unit_of_measurement=UnitOfPower.KILO_WATT,
suggested_display_precision=2,
device_class=SensorDeviceClass.POWER,
),
SensorEntityDescription(
key="energy_left",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfEnergy.WATT_HOUR,
suggested_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
suggested_display_precision=2,
device_class=SensorDeviceClass.ENERGY_STORAGE,
entity_category=EntityCategory.DIAGNOSTIC,
),
SensorEntityDescription(
key="total_pack_energy",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfEnergy.WATT_HOUR,
suggested_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
suggested_display_precision=2,
device_class=SensorDeviceClass.ENERGY_STORAGE,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
),
SensorEntityDescription(
key="percentage_charged",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PERCENTAGE,
device_class=SensorDeviceClass.BATTERY,
suggested_display_precision=2,
),
SensorEntityDescription(
key="battery_power",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfPower.WATT,
suggested_unit_of_measurement=UnitOfPower.KILO_WATT,
suggested_display_precision=2,
device_class=SensorDeviceClass.POWER,
),
SensorEntityDescription(
key="load_power",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfPower.WATT,
suggested_unit_of_measurement=UnitOfPower.KILO_WATT,
suggested_display_precision=2,
device_class=SensorDeviceClass.POWER,
),
SensorEntityDescription(
key="grid_power",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfPower.WATT,
suggested_unit_of_measurement=UnitOfPower.KILO_WATT,
suggested_display_precision=2,
device_class=SensorDeviceClass.POWER,
),
SensorEntityDescription(
key="grid_services_power",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfPower.WATT,
suggested_unit_of_measurement=UnitOfPower.KILO_WATT,
suggested_display_precision=2,
device_class=SensorDeviceClass.POWER,
),
SensorEntityDescription(
key="generator_power",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfPower.WATT,
suggested_unit_of_measurement=UnitOfPower.KILO_WATT,
suggested_display_precision=2,
device_class=SensorDeviceClass.POWER,
entity_registry_enabled_default=False,
),
)
WALL_CONNECTOR_DESCRIPTIONS: tuple[SensorEntityDescription, ...] = (
SensorEntityDescription(
key="wall_connector_state",
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
),
SensorEntityDescription(
key="wall_connector_fault_state",
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
),
SensorEntityDescription(
key="wall_connector_power",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfPower.WATT,
suggested_unit_of_measurement=UnitOfPower.KILO_WATT,
suggested_display_precision=2,
device_class=SensorDeviceClass.POWER,
),
SensorEntityDescription(
key="vin",
),
)
ENERGY_INFO_DESCRIPTIONS: tuple[SensorEntityDescription, ...] = (
SensorEntityDescription(
key="vpp_backup_reserve_percent",
entity_category=EntityCategory.DIAGNOSTIC,
device_class=SensorDeviceClass.BATTERY,
native_unit_of_measurement=PERCENTAGE,
),
SensorEntityDescription(key="version"),
)
async def async_setup_entry(
hass: HomeAssistant,
entry: TeslaFleetConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the Tesla Fleet sensor platform from a config entry."""
async_add_entities(
chain(
( # Add vehicles
TeslaFleetVehicleSensorEntity(vehicle, description)
for vehicle in entry.runtime_data.vehicles
for description in VEHICLE_DESCRIPTIONS
),
( # Add vehicles time sensors
TeslaFleetVehicleTimeSensorEntity(vehicle, description)
for vehicle in entry.runtime_data.vehicles
for description in VEHICLE_TIME_DESCRIPTIONS
),
( # Add energy site live
TeslaFleetEnergyLiveSensorEntity(energysite, description)
for energysite in entry.runtime_data.energysites
for description in ENERGY_LIVE_DESCRIPTIONS
if description.key in energysite.live_coordinator.data
),
( # Add wall connectors
TeslaFleetWallConnectorSensorEntity(energysite, wc["din"], description)
for energysite in entry.runtime_data.energysites
for wc in energysite.info_coordinator.data.get(
"components_wall_connectors", []
)
if "din" in wc
for description in WALL_CONNECTOR_DESCRIPTIONS
),
( # Add energy site info
TeslaFleetEnergyInfoSensorEntity(energysite, description)
for energysite in entry.runtime_data.energysites
for description in ENERGY_INFO_DESCRIPTIONS
if description.key in energysite.info_coordinator.data
),
)
)
class TeslaFleetVehicleSensorEntity(TeslaFleetVehicleEntity, RestoreSensor):
"""Base class for Tesla Fleet vehicle metric sensors."""
entity_description: TeslaFleetSensorEntityDescription
def __init__(
self,
data: TeslaFleetVehicleData,
description: TeslaFleetSensorEntityDescription,
) -> None:
"""Initialize the sensor."""
self.entity_description = description
super().__init__(data, description.key)
async def async_added_to_hass(self) -> None:
"""Handle entity which will be added."""
await super().async_added_to_hass()
if self.coordinator.data.get("state") == TeslaFleetState.OFFLINE:
if (sensor_data := await self.async_get_last_sensor_data()) is not None:
self._attr_native_value = sensor_data.native_value
def _async_update_attrs(self) -> None:
"""Update the attributes of the sensor."""
if self.has:
self._attr_native_value = self.entity_description.value_fn(self._value)
else:
self._attr_native_value = None
class TeslaFleetVehicleTimeSensorEntity(TeslaFleetVehicleEntity, SensorEntity):
"""Base class for Tesla Fleet vehicle time sensors."""
entity_description: TeslaFleetTimeEntityDescription
def __init__(
self,
data: TeslaFleetVehicleData,
description: TeslaFleetTimeEntityDescription,
) -> None:
"""Initialize the sensor."""
self.entity_description = description
self._get_timestamp = ignore_variance(
func=lambda value: dt_util.now() + timedelta(minutes=value),
ignored_variance=timedelta(minutes=description.variance),
)
super().__init__(data, description.key)
def _async_update_attrs(self) -> None:
"""Update the attributes of the sensor."""
self._attr_available = isinstance(self._value, int | float) and self._value > 0
if self._attr_available:
self._attr_native_value = self._get_timestamp(self._value)
class TeslaFleetEnergyLiveSensorEntity(TeslaFleetEnergyLiveEntity, RestoreSensor):
"""Base class for Tesla Fleet energy site metric sensors."""
entity_description: SensorEntityDescription
def __init__(
self,
data: TeslaFleetEnergyData,
description: SensorEntityDescription,
) -> None:
"""Initialize the sensor."""
self.entity_description = description
super().__init__(data, description.key)
async def async_added_to_hass(self) -> None:
"""Handle entity which will be added."""
await super().async_added_to_hass()
if not self.coordinator.updated_once:
if (sensor_data := await self.async_get_last_sensor_data()) is not None:
self._attr_native_value = sensor_data.native_value
def _async_update_attrs(self) -> None:
"""Update the attributes of the sensor."""
self._attr_available = not self.is_none
self._attr_native_value = self._value
class TeslaFleetWallConnectorSensorEntity(TeslaFleetWallConnectorEntity, RestoreSensor):
"""Base class for Tesla Fleet energy site metric sensors."""
entity_description: SensorEntityDescription
def __init__(
self,
data: TeslaFleetEnergyData,
din: str,
description: SensorEntityDescription,
) -> None:
"""Initialize the sensor."""
self.entity_description = description
super().__init__(
data,
din,
description.key,
)
async def async_added_to_hass(self) -> None:
"""Handle entity which will be added."""
await super().async_added_to_hass()
if not self.coordinator.updated_once:
if (sensor_data := await self.async_get_last_sensor_data()) is not None:
self._attr_native_value = sensor_data.native_value
def _async_update_attrs(self) -> None:
"""Update the attributes of the sensor."""
self._attr_available = not self.is_none
self._attr_native_value = self._value
class TeslaFleetEnergyInfoSensorEntity(TeslaFleetEnergyInfoEntity, RestoreSensor):
"""Base class for Tesla Fleet energy site metric sensors."""
entity_description: SensorEntityDescription
def __init__(
self,
data: TeslaFleetEnergyData,
description: SensorEntityDescription,
) -> None:
"""Initialize the sensor."""
self.entity_description = description
super().__init__(data, description.key)
async def async_added_to_hass(self) -> None:
"""Handle entity which will be added."""
await super().async_added_to_hass()
if not self.coordinator.updated_once:
if (sensor_data := await self.async_get_last_sensor_data()) is not None:
self._attr_native_value = sensor_data.native_value
def _async_update_attrs(self) -> None:
"""Update the attributes of the sensor."""
self._attr_available = not self.is_none
self._attr_native_value = self._value

190
homeassistant/components/tesla_fleet/strings.json Normal file
View file

@ -0,0 +1,190 @@
{
"config": {
"abort": {
"authorize_url_timeout": "[%key:common::config_flow::abort::oauth2_authorize_url_timeout%]",
"missing_configuration": "[%key:common::config_flow::abort::oauth2_missing_configuration%]",
"already_configured": "Configuration updated for profile.",
"no_url_available": "[%key:common::config_flow::abort::oauth2_no_url_available%]",
"oauth_error": "[%key:common::config_flow::abort::oauth2_error%]",
"oauth_timeout": "[%key:common::config_flow::abort::oauth2_timeout%]",
"oauth_unauthorized": "[%key:common::config_flow::abort::oauth2_unauthorized%]",
"oauth_failed": "[%key:common::config_flow::abort::oauth2_failed%]"
},
"error": {
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]"
},
"step": {
"pick_implementation": {
"title": "[%key:common::config_flow::title::oauth2_pick_implementation%]"
},
"reauth_confirm": {
"title": "[%key:common::config_flow::title::reauth%]",
"description": "The Withings integration needs to re-authenticate your account"
}
},
"create_entry": {
"default": "Successfully authenticated with Tesla."
}
},
"entity": {
"sensor": {
"battery_power": {
"name": "Battery power"
},
"charge_state_battery_range": {
"name": "Battery range"
},
"charge_state_est_battery_range": {
"name": "Estimate battery range"
},
"charge_state_ideal_battery_range": {
"name": "Ideal battery range"
},
"charge_state_charge_energy_added": {
"name": "Charge energy added"
},
"charge_state_charge_rate": {
"name": "Charge rate"
},
"charge_state_charger_actual_current": {
"name": "Charger current"
},
"charge_state_charger_power": {
"name": "Charger power"
},
"charge_state_charger_voltage": {
"name": "Charger voltage"
},
"charge_state_conn_charge_cable": {
"name": "Charge cable"
},
"charge_state_fast_charger_type": {
"name": "Fast charger type"
},
"charge_state_charging_state": {
"name": "Charging",
"state": {
"starting": "Starting",
"charging": "Charging",
"disconnected": "Disconnected",
"stopped": "Stopped",
"complete": "Complete",
"no_power": "No power"
}
},
"charge_state_minutes_to_full_charge": {
"name": "Time to full charge"
},
"charge_state_battery_level": {
"name": "Battery level"
},
"charge_state_usable_battery_level": {
"name": "Usable battery level"
},
"climate_state_driver_temp_setting": {
"name": "Driver temperature setting"
},
"climate_state_inside_temp": {
"name": "Inside temperature"
},
"climate_state_outside_temp": {
"name": "Outside temperature"
},
"climate_state_passenger_temp_setting": {
"name": "Passenger temperature setting"
},
"drive_state_active_route_destination": {
"name": "Destination"
},
"drive_state_active_route_energy_at_arrival": {
"name": "State of charge at arrival"
},
"drive_state_active_route_miles_to_arrival": {
"name": "Distance to arrival"
},
"drive_state_active_route_minutes_to_arrival": {
"name": "Time to arrival"
},
"drive_state_active_route_traffic_minutes_delay": {
"name": "Traffic delay"
},
"drive_state_power": {
"name": "Power"
},
"drive_state_shift_state": {
"name": "Shift state",
"state": {
"d": "Drive",
"n": "Neutral",
"p": "Park",
"r": "Reverse"
}
},
"drive_state_speed": {
"name": "Speed"
},
"energy_left": {
"name": "Energy left"
},
"generator_power": {
"name": "Generator power"
},
"grid_power": {
"name": "Grid power"
},
"grid_services_power": {
"name": "Grid services power"
},
"load_power": {
"name": "Load power"
},
"percentage_charged": {
"name": "Percentage charged"
},
"solar_power": {
"name": "Solar power"
},
"total_pack_energy": {
"name": "Total pack energy"
},
"vehicle_state_odometer": {
"name": "Odometer"
},
"vehicle_state_tpms_pressure_fl": {
"name": "Tire pressure front left"
},
"vehicle_state_tpms_pressure_fr": {
"name": "Tire pressure front right"
},
"vehicle_state_tpms_pressure_rl": {
"name": "Tire pressure rear left"
},
"vehicle_state_tpms_pressure_rr": {
"name": "Tire pressure rear right"
},
"version": {
"name": "version"
},
"vin": {
"name": "Vehicle"
},
"vpp_backup_reserve_percent": {
"name": "VPP backup reserve"
},
"wall_connector_fault_state": {
"name": "Fault state code"
},
"wall_connector_power": {
"name": "Power"
},
"wall_connector_state": {
"name": "State code"
}
}
},
"exceptions": {
"update_failed": {
"message": "{endpoint} data request failed. {message}"
}
}
}

1
homeassistant/generated/application_credentials.py
View file

@ -24,6 +24,7 @@ APPLICATION_CREDENTIALS = [
"netatmo",
"senz",
"spotify",
"tesla_fleet",
"twitch",
"withings",
"xbox",

1
homeassistant/generated/config_flows.py
View file

@ -565,6 +565,7 @@ FLOWS = {
"technove",
"tedee",
"tellduslive",
"tesla_fleet",
"tesla_wall_connector",
"teslemetry",
"tessie",

6
homeassistant/generated/integrations.json
View file

@ -6121,6 +6121,12 @@
"config_flow": true,
"iot_class": "local_polling",
"name": "Tesla Wall Connector"
},
"tesla_fleet": {
"integration_type": "hub",
"config_flow": true,
"iot_class": "cloud_polling",
"name": "Tesla Fleet"
}
}
},

1
requirements_all.txt
View file

@ -2711,6 +2711,7 @@ temperusb==1.6.1
# homeassistant.components.tensorflow
# tensorflow==2.5.0
# homeassistant.components.tesla_fleet
# homeassistant.components.teslemetry
# homeassistant.components.tessie
tesla-fleet-api==0.7.2

1
requirements_test_all.txt
View file

@ -2121,6 +2121,7 @@ temescal==0.5
# homeassistant.components.temper
temperusb==1.6.1
# homeassistant.components.tesla_fleet
# homeassistant.components.teslemetry
# homeassistant.components.tessie
tesla-fleet-api==0.7.2

60
tests/components/tesla_fleet/__init__.py Normal file
View file

@ -0,0 +1,60 @@
"""Tests for the Tesla Fleet integration."""
from unittest.mock import patch
from syrupy import SnapshotAssertion
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from tests.common import MockConfigEntry
async def setup_platform(
hass: HomeAssistant,
config_entry: MockConfigEntry,
platforms: list[Platform] | None = None,
) -> None:
"""Set up the Tesla Fleet platform."""
config_entry.add_to_hass(hass)
if platforms is None:
await hass.config_entries.async_setup(config_entry.entry_id)
else:
with patch("homeassistant.components.tesla_fleet.PLATFORMS", platforms):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
def assert_entities(
hass: HomeAssistant,
entry_id: str,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test that all entities match their snapshot."""
entity_entries = er.async_entries_for_config_entry(entity_registry, entry_id)
assert entity_entries
for entity_entry in entity_entries:
assert entity_entry == snapshot(name=f"{entity_entry.entity_id}-entry")
assert (state := hass.states.get(entity_entry.entity_id))
assert state == snapshot(name=f"{entity_entry.entity_id}-state")
def assert_entities_alt(
hass: HomeAssistant,
entry_id: str,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test that all entities match their alt snapshot."""
entity_entries = er.async_entries_for_config_entry(entity_registry, entry_id)
assert entity_entries
for entity_entry in entity_entries:
assert (state := hass.states.get(entity_entry.entity_id))
assert state == snapshot(name=f"{entity_entry.entity_id}-statealt")

142
tests/components/tesla_fleet/conftest.py Normal file
View file

@ -0,0 +1,142 @@
"""Fixtures for Tessie."""
from __future__ import annotations
from copy import deepcopy
import time
from unittest.mock import patch
import jwt
import pytest
from homeassistant.components.application_credentials import (
ClientCredential,
async_import_client_credential,
)
from homeassistant.components.tesla_fleet.application_credentials import CLIENT_ID
from homeassistant.components.tesla_fleet.const import DOMAIN, SCOPES
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from .const import LIVE_STATUS, PRODUCTS, SITE_INFO, VEHICLE_DATA, VEHICLE_ONLINE
from tests.common import MockConfigEntry
UID = "abc-123"
@pytest.fixture(name="expires_at")
def mock_expires_at() -> int:
"""Fixture to set the oauth token expiration time."""
return time.time() + 3600
@pytest.fixture(name="scopes")
def mock_scopes() -> list[str]:
"""Fixture to set the scopes present in the OAuth token."""
return SCOPES
@pytest.fixture
def normal_config_entry(expires_at: int, scopes: list[str]) -> MockConfigEntry:
"""Create Tesla Fleet entry in Home Assistant."""
access_token = jwt.encode(
{
"sub": UID,
"aud": [],
"scp": scopes,
"ou_code": "NA",
},
key="",
algorithm="none",
)
return MockConfigEntry(
domain=DOMAIN,
title=UID,
unique_id=UID,
data={
"auth_implementation": DOMAIN,
"token": {
"status": 0,
"userid": UID,
"access_token": access_token,
"refresh_token": "mock-refresh-token",
"expires_at": expires_at,
"scope": ",".join(scopes),
},
},
)
@pytest.fixture(autouse=True)
async def setup_credentials(hass: HomeAssistant) -> None:
"""Fixture to setup credentials."""
assert await async_setup_component(hass, "application_credentials", {})
await async_import_client_credential(
hass,
DOMAIN,
ClientCredential(CLIENT_ID, ""),
DOMAIN,
)
@pytest.fixture(autouse=True)
def mock_products():
"""Mock Tesla Fleet Api products method."""
with patch(
"homeassistant.components.tesla_fleet.TeslaFleetApi.products",
return_value=PRODUCTS,
) as mock_products:
yield mock_products
@pytest.fixture(autouse=True)
def mock_vehicle_state():
"""Mock Tesla Fleet API Vehicle Specific vehicle method."""
with patch(
"homeassistant.components.tesla_fleet.VehicleSpecific.vehicle",
return_value=VEHICLE_ONLINE,
) as mock_vehicle:
yield mock_vehicle
@pytest.fixture(autouse=True)
def mock_vehicle_data():
"""Mock Tesla Fleet API Vehicle Specific vehicle_data method."""
with patch(
"homeassistant.components.tesla_fleet.VehicleSpecific.vehicle_data",
return_value=VEHICLE_DATA,
) as mock_vehicle_data:
yield mock_vehicle_data
@pytest.fixture(autouse=True)
def mock_wake_up():
"""Mock Tesla Fleet API Vehicle Specific wake_up method."""
with patch(
"homeassistant.components.tesla_fleet.VehicleSpecific.wake_up",
return_value=VEHICLE_ONLINE,
) as mock_wake_up:
yield mock_wake_up
@pytest.fixture(autouse=True)
def mock_live_status():
"""Mock Teslemetry Energy Specific live_status method."""
with patch(
"homeassistant.components.tesla_fleet.EnergySpecific.live_status",
side_effect=lambda: deepcopy(LIVE_STATUS),
) as mock_live_status:
yield mock_live_status
@pytest.fixture(autouse=True)
def mock_site_info():
"""Mock Teslemetry Energy Specific site_info method."""
with patch(
"homeassistant.components.tesla_fleet.EnergySpecific.site_info",
side_effect=lambda: deepcopy(SITE_INFO),
) as mock_live_status:
yield mock_live_status

28
tests/components/tesla_fleet/const.py Normal file
View file

@ -0,0 +1,28 @@
"""Constants for the Tesla Fleet tests."""
from homeassistant.components.tesla_fleet.const import DOMAIN, TeslaFleetState
from tests.common import load_json_object_fixture
VEHICLE_ONLINE = {"response": {"state": TeslaFleetState.ONLINE}, "error": None}
VEHICLE_ASLEEP = {"response": {"state": TeslaFleetState.ASLEEP}, "error": None}
PRODUCTS = load_json_object_fixture("products.json", DOMAIN)
VEHICLE_DATA = load_json_object_fixture("vehicle_data.json", DOMAIN)
VEHICLE_DATA_ALT = load_json_object_fixture("vehicle_data_alt.json", DOMAIN)
LIVE_STATUS = load_json_object_fixture("live_status.json", DOMAIN)
SITE_INFO = load_json_object_fixture("site_info.json", DOMAIN)
COMMAND_OK = {"response": {"result": True, "reason": ""}}
COMMAND_REASON = {"response": {"result": False, "reason": "already closed"}}
COMMAND_IGNORED_REASON = {"response": {"result": False, "reason": "already_set"}}
COMMAND_NOREASON = {"response": {"result": False}} # Unexpected
COMMAND_ERROR = {
"response": None,
"error": "vehicle unavailable: vehicle is offline or asleep",
"error_description": "",
}
COMMAND_NOERROR = {"answer": 42}
COMMAND_ERRORS = (COMMAND_REASON, COMMAND_NOREASON, COMMAND_ERROR, COMMAND_NOERROR)
RESPONSE_OK = {"response": {}, "error": None}

33
tests/components/tesla_fleet/fixtures/live_status.json Normal file
View file

@ -0,0 +1,33 @@
{
"response": {
"solar_power": 1185,
"energy_left": 38896.47368421053,
"total_pack_energy": 40727,
"percentage_charged": 95.50537403739663,
"backup_capable": true,
"battery_power": 5060,
"load_power": 6245,
"grid_status": "Active",
"grid_services_active": false,
"grid_power": 0,
"grid_services_power": 0,
"generator_power": 0,
"island_status": "on_grid",
"storm_mode_active": false,
"timestamp": "2024-01-01T00:00:00+00:00",
"wall_connectors": [
{
"din": "abd-123",
"wall_connector_state": 2,
"wall_connector_fault_state": 2,
"wall_connector_power": 0
},
{
"din": "bcd-234",
"wall_connector_state": 2,
"wall_connector_fault_state": 2,
"wall_connector_power": 0
}
]
}
}

121
tests/components/tesla_fleet/fixtures/products.json Normal file
View file

@ -0,0 +1,121 @@
{
"response": [
{
"id": 1234,
"user_id": 1234,
"vehicle_id": 1234,
"vin": "LRWXF7EK4KC700000",
"color": null,
"access_type": "OWNER",
"display_name": "Test",
"option_codes": null,
"cached_data": null,
"granular_access": { "hide_private": false },
"tokens": ["abc", "def"],
"state": "asleep",
"in_service": false,
"id_s": "1234",
"calendar_enabled": true,
"api_version": 71,
"backseat_token": null,
"backseat_token_updated_at": null,
"ble_autopair_enrolled": false,
"vehicle_config": {
"aux_park_lamps": "Eu",
"badge_version": 1,
"can_accept_navigation_requests": true,
"can_actuate_trunks": true,
"car_special_type": "base",
"car_type": "model3",
"charge_port_type": "CCS",
"cop_user_set_temp_supported": false,
"dashcam_clip_save_supported": true,
"default_charge_to_max": false,
"driver_assist": "TeslaAP3",
"ece_restrictions": false,
"efficiency_package": "M32021",
"eu_vehicle": true,
"exterior_color": "DeepBlue",
"exterior_trim": "Black",
"exterior_trim_override": "",
"has_air_suspension": false,
"has_ludicrous_mode": false,
"has_seat_cooling": false,
"headlamp_type": "Global",
"interior_trim_type": "White2",
"key_version": 2,
"motorized_charge_port": true,
"paint_color_override": "0,9,25,0.7,0.04",
"performance_package": "Base",
"plg": true,
"pws": true,
"rear_drive_unit": "PM216MOSFET",
"rear_seat_heaters": 1,
"rear_seat_type": 0,
"rhd": true,
"roof_color": "RoofColorGlass",
"seat_type": null,
"spoiler_type": "None",
"sun_roof_installed": null,
"supports_qr_pairing": false,
"third_row_seats": "None",
"timestamp": 1705701487912,
"trim_badging": "74d",
"use_range_badging": true,
"utc_offset": 36000,
"webcam_selfie_supported": true,
"webcam_supported": true,
"wheel_type": "Pinwheel18CapKit"
},
"command_signing": "allowed",
"release_notes_supported": true
},
{
"energy_site_id": 123456,
"resource_type": "battery",
"site_name": "Energy Site",
"id": "ABC123",
"gateway_id": "ABC123",
"asset_site_id": "c0ffee",
"warp_site_number": "GA123456",
"energy_left": 23286.105263157893,
"total_pack_energy": 40804,
"percentage_charged": 57.068192488868476,
"battery_type": "ac_powerwall",
"backup_capable": true,
"battery_power": 14990,
"go_off_grid_test_banner_enabled": null,
"storm_mode_enabled": true,
"powerwall_onboarding_settings_set": true,
"powerwall_tesla_electric_interested_in": null,
"vpp_tour_enabled": null,
"sync_grid_alert_enabled": true,
"breaker_alert_enabled": true,
"components": {
"battery": true,
"battery_type": "ac_powerwall",
"solar": true,
"solar_type": "pv_panel",
"grid": true,
"load_meter": true,
"market_type": "residential",
"wall_connectors": [
{
"device_id": "abc-123",
"din": "123-abc",
"is_active": true
},
{
"device_id": "bcd-234",
"din": "234-bcd",
"is_active": true
}
]
},
"features": {
"rate_plan_manager_no_pricing_constraint": true
}
}
],
"count": 2
}

127
tests/components/tesla_fleet/fixtures/site_info.json Normal file
View file

@ -0,0 +1,127 @@
{
"response": {
"id": "1233-abcd",
"site_name": "Site",
"backup_reserve_percent": 0,
"default_real_mode": "self_consumption",
"installation_date": "2022-01-01T00:00:00+00:00",
"user_settings": {
"go_off_grid_test_banner_enabled": false,
"storm_mode_enabled": true,
"powerwall_onboarding_settings_set": true,
"powerwall_tesla_electric_interested_in": false,
"vpp_tour_enabled": true,
"sync_grid_alert_enabled": true,
"breaker_alert_enabled": false
},
"components": {
"solar": true,
"solar_type": "pv_panel",
"battery": true,
"grid": true,
"backup": true,
"gateway": "teg",
"load_meter": true,
"tou_capable": true,
"storm_mode_capable": true,
"flex_energy_request_capable": false,
"car_charging_data_supported": false,
"off_grid_vehicle_charging_reserve_supported": true,
"vehicle_charging_performance_view_enabled": false,
"vehicle_charging_solar_offset_view_enabled": false,
"battery_solar_offset_view_enabled": true,
"solar_value_enabled": true,
"energy_value_header": "Energy Value",
"energy_value_subheader": "Estimated Value",
"energy_service_self_scheduling_enabled": true,
"show_grid_import_battery_source_cards": true,
"set_islanding_mode_enabled": true,
"wifi_commissioning_enabled": true,
"backup_time_remaining_enabled": true,
"battery_type": "ac_powerwall",
"configurable": true,
"grid_services_enabled": false,
"gateways": [
{
"device_id": "gateway-id",
"din": "gateway-din",
"serial_number": "CN00000000J50D",
"part_number": "1152100-14-J",
"part_type": 10,
"part_name": "Tesla Backup Gateway 2",
"is_active": true,
"site_id": "1234-abcd",
"firmware_version": "24.4.0 0fe780c9",
"updated_datetime": "2024-05-14T00:00:00.000Z"
}
],
"batteries": [
{
"device_id": "battery-1-id",
"din": "battery-1-din",
"serial_number": "TG000000001DA5",
"part_number": "3012170-10-B",
"part_type": 2,
"part_name": "Powerwall 2",
"nameplate_max_charge_power": 5000,
"nameplate_max_discharge_power": 5000,
"nameplate_energy": 13500
},
{
"device_id": "battery-2-id",
"din": "battery-2-din",
"serial_number": "TG000000002DA5",
"part_number": "3012170-05-C",
"part_type": 2,
"part_name": "Powerwall 2",
"nameplate_max_charge_power": 5000,
"nameplate_max_discharge_power": 5000,
"nameplate_energy": 13500
}
],
"wall_connectors": [
{
"device_id": "123abc",
"din": "abd-123",
"part_name": "Gen 3 Wall Connector",
"is_active": true
},
{
"device_id": "234bcd",
"din": "bcd-234",
"part_name": "Gen 3 Wall Connector",
"is_active": true
}
],
"disallow_charge_from_grid_with_solar_installed": true,
"customer_preferred_export_rule": "pv_only",
"net_meter_mode": "battery_ok",
"system_alerts_enabled": true
},
"version": "23.44.0 eb113390",
"battery_count": 2,
"tou_settings": {
"optimization_strategy": "economics",
"schedule": [
{
"target": "off_peak",
"week_days": [1, 0],
"start_seconds": 0,
"end_seconds": 3600
},
{
"target": "peak",
"week_days": [1, 0],
"start_seconds": 3600,
"end_seconds": 0
}
]
},
"nameplate_power": 15000,
"nameplate_energy": 40500,
"installation_time_zone": "",
"max_site_meter_power_ac": 1000000000,
"min_site_meter_power_ac": -1000000000,
"vpp_backup_reserve_percent": 0
}
}

282
tests/components/tesla_fleet/fixtures/vehicle_data.json Normal file
View file

@ -0,0 +1,282 @@
{
"response": {
"id": 1234,
"user_id": 1234,
"vehicle_id": 1234,
"vin": "LRWXF7EK4KC700000",
"color": null,
"access_type": "OWNER",
"granular_access": {
"hide_private": false
},
"tokens": ["abc", "def"],
"state": "online",
"in_service": false,
"id_s": "1234",
"calendar_enabled": true,
"api_version": 71,
"backseat_token": null,
"backseat_token_updated_at": null,
"ble_autopair_enrolled": false,
"charge_state": {
"battery_heater_on": false,
"battery_level": 77,
"battery_range": 266.87,
"charge_amps": 16,
"charge_current_request": 16,
"charge_current_request_max": 16,
"charge_enable_request": true,
"charge_energy_added": 0,
"charge_limit_soc": 80,
"charge_limit_soc_max": 100,
"charge_limit_soc_min": 50,
"charge_limit_soc_std": 80,
"charge_miles_added_ideal": 0,
"charge_miles_added_rated": 0,
"charge_port_cold_weather_mode": false,
"charge_port_color": "<invalid>",
"charge_port_door_open": true,
"charge_port_latch": "Engaged",
"charge_rate": 0,
"charger_actual_current": 0,
"charger_phases": null,
"charger_pilot_current": 16,
"charger_power": 0,
"charger_voltage": 2,
"charging_state": "Stopped",
"conn_charge_cable": "IEC",
"est_battery_range": 275.04,
"fast_charger_brand": "<invalid>",
"fast_charger_present": false,
"fast_charger_type": "ACSingleWireCAN",
"ideal_battery_range": 266.87,
"max_range_charge_counter": 0,
"minutes_to_full_charge": 0,
"not_enough_power_to_heat": null,
"off_peak_charging_enabled": false,
"off_peak_charging_times": "all_week",
"off_peak_hours_end_time": 900,
"preconditioning_enabled": false,
"preconditioning_times": "all_week",
"scheduled_charging_mode": "Off",
"scheduled_charging_pending": false,
"scheduled_charging_start_time": null,
"scheduled_charging_start_time_app": 600,
"scheduled_departure_time": 1704837600,
"scheduled_departure_time_minutes": 480,
"supercharger_session_trip_planner": false,
"time_to_full_charge": 0,
"timestamp": 1705707520649,
"trip_charging": false,
"usable_battery_level": 77,
"user_charge_enable_request": null
},
"climate_state": {
"allow_cabin_overheat_protection": true,
"auto_seat_climate_left": true,
"auto_seat_climate_right": true,
"auto_steering_wheel_heat": false,
"battery_heater": false,
"battery_heater_no_power": null,
"cabin_overheat_protection": "On",
"cabin_overheat_protection_actively_cooling": false,
"climate_keeper_mode": "keep",
"cop_activation_temperature": "High",
"defrost_mode": 0,
"driver_temp_setting": 22,
"fan_status": 0,
"hvac_auto_request": "On",
"inside_temp": 29.8,
"is_auto_conditioning_on": false,
"is_climate_on": true,
"is_front_defroster_on": false,
"is_preconditioning": false,
"is_rear_defroster_on": false,
"left_temp_direction": 251,
"max_avail_temp": 28,
"min_avail_temp": 15,
"outside_temp": 30,
"passenger_temp_setting": 22,
"remote_heater_control_enabled": false,
"right_temp_direction": 251,
"seat_heater_left": 0,
"seat_heater_rear_center": 0,
"seat_heater_rear_left": 0,
"seat_heater_rear_right": 0,
"seat_heater_right": 0,
"side_mirror_heaters": false,
"steering_wheel_heat_level": 0,
"steering_wheel_heater": false,
"supports_fan_only_cabin_overheat_protection": true,
"timestamp": 1705707520649,
"wiper_blade_heater": false
},
"drive_state": {
"active_route_latitude": 30.2226265,
"active_route_longitude": -97.6236871,
"active_route_miles_to_arrival": 0.039491,
"active_route_minutes_to_arrival": 0.103577,
"active_route_traffic_minutes_delay": 0,
"gps_as_of": 1701129612,
"heading": 185,
"latitude": -30.222626,
"longitude": -97.6236871,
"native_latitude": -30.222626,
"native_location_supported": 1,
"native_longitude": -97.6236871,
"native_type": "wgs",
"power": -7,
"shift_state": null,
"speed": null,
"timestamp": 1705707520649
},
"gui_settings": {
"gui_24_hour_time": false,
"gui_charge_rate_units": "kW",
"gui_distance_units": "km/hr",
"gui_range_display": "Rated",
"gui_temperature_units": "C",
"gui_tirepressure_units": "Psi",
"show_range_units": false,
"timestamp": 1705707520649
},
"vehicle_config": {
"aux_park_lamps": "Eu",
"badge_version": 1,
"can_accept_navigation_requests": true,
"can_actuate_trunks": true,
"car_special_type": "base",
"car_type": "model3",
"charge_port_type": "CCS",
"cop_user_set_temp_supported": true,
"dashcam_clip_save_supported": true,
"default_charge_to_max": false,
"driver_assist": "TeslaAP3",
"ece_restrictions": false,
"efficiency_package": "M32021",
"eu_vehicle": true,
"exterior_color": "DeepBlue",
"exterior_trim": "Black",
"exterior_trim_override": "",
"has_air_suspension": false,
"has_ludicrous_mode": false,
"has_seat_cooling": false,
"headlamp_type": "Global",
"interior_trim_type": "White2",
"key_version": 2,
"motorized_charge_port": true,
"paint_color_override": "0,9,25,0.7,0.04",
"performance_package": "Base",
"plg": true,
"pws": true,
"rear_drive_unit": "PM216MOSFET",
"rear_seat_heaters": 1,
"rear_seat_type": 0,
"rhd": true,
"roof_color": "RoofColorGlass",
"seat_type": null,
"spoiler_type": "None",
"sun_roof_installed": true,
"supports_qr_pairing": false,
"third_row_seats": "None",
"timestamp": 1705707520649,
"trim_badging": "74d",
"use_range_badging": true,
"utc_offset": 36000,
"webcam_selfie_supported": true,
"webcam_supported": true,
"wheel_type": "Pinwheel18CapKit"
},
"vehicle_state": {
"api_version": 71,
"autopark_state_v2": "unavailable",
"calendar_supported": true,
"car_version": "2023.44.30.8 06f534d46010",
"center_display_state": 0,
"dashcam_clip_save_available": true,
"dashcam_state": "Recording",
"df": 0,
"dr": 0,
"fd_window": 0,
"feature_bitmask": "fbdffbff,187f",
"fp_window": 0,
"ft": 0,
"is_user_present": false,
"locked": false,
"media_info": {
"a2dp_source_name": "Pixel 8 Pro",
"audio_volume": 1.6667,
"audio_volume_increment": 0.333333,
"audio_volume_max": 10.333333,
"media_playback_status": "Playing",
"now_playing_album": "Elon Musk",
"now_playing_artist": "Walter Isaacson",
"now_playing_duration": 651000,
"now_playing_elapsed": 1000,
"now_playing_source": "Audible",
"now_playing_station": "Elon Musk",
"now_playing_title": "Chapter 51: Cybertruck: Tesla, 2018–2019"
},
"media_state": {
"remote_control_enabled": true
},
"notifications_supported": true,
"odometer": 6481.019282,
"parsed_calendar_supported": true,
"pf": 0,
"pr": 0,
"rd_window": 0,
"remote_start": false,
"remote_start_enabled": true,
"remote_start_supported": true,
"rp_window": 0,
"rt": 0,
"santa_mode": 0,
"sentry_mode": false,
"sentry_mode_available": true,
"service_mode": false,
"service_mode_plus": false,
"software_update": {
"download_perc": 100,
"expected_duration_sec": 2700,
"install_perc": 1,
"status": "available",
"version": "2024.12.0.0"
},
"speed_limit_mode": {
"active": false,
"current_limit_mph": 69,
"max_limit_mph": 120,
"min_limit_mph": 50,
"pin_code_set": true
},
"sun_roof_state": "open",
"vehicle_state_sun_roof_percent_open": 20,
"timestamp": 1705707520649,
"tpms_hard_warning_fl": false,
"tpms_hard_warning_fr": false,
"tpms_hard_warning_rl": false,
"tpms_hard_warning_rr": false,
"tpms_last_seen_pressure_time_fl": 1705700812,
"tpms_last_seen_pressure_time_fr": 1705700793,
"tpms_last_seen_pressure_time_rl": 1705700794,
"tpms_last_seen_pressure_time_rr": 1705700823,
"tpms_pressure_fl": 2.775,
"tpms_pressure_fr": 2.8,
"tpms_pressure_rl": 2.775,
"tpms_pressure_rr": 2.775,
"tpms_rcp_front_value": 2.9,
"tpms_rcp_rear_value": 2.9,
"tpms_soft_warning_fl": false,
"tpms_soft_warning_fr": false,
"tpms_soft_warning_rl": false,
"tpms_soft_warning_rr": false,
"valet_mode": false,
"valet_pin_needed": false,
"vehicle_name": "Test",
"vehicle_self_test_progress": 0,
"vehicle_self_test_requested": false,
"webcam_available": true
}
}
}

279
tests/components/tesla_fleet/fixtures/vehicle_data_alt.json Normal file
View file

@ -0,0 +1,279 @@
{
"response": {
"id": 1234,
"user_id": 1234,
"vehicle_id": 1234,
"vin": "LRWXF7EK4KC700000",
"color": null,
"access_type": "OWNER",
"granular_access": {
"hide_private": false
},
"tokens": ["abc", "def"],
"state": "online",
"in_service": false,
"id_s": "1234",
"calendar_enabled": true,
"api_version": 71,
"backseat_token": null,
"backseat_token_updated_at": null,
"ble_autopair_enrolled": false,
"charge_state": {
"battery_heater_on": true,
"battery_level": 77,
"battery_range": 266.87,
"charge_amps": 16,
"charge_current_request": 16,
"charge_current_request_max": 16,
"charge_enable_request": true,
"charge_energy_added": 0,
"charge_limit_soc": 80,
"charge_limit_soc_max": 100,
"charge_limit_soc_min": 50,
"charge_limit_soc_std": 80,
"charge_miles_added_ideal": 0,
"charge_miles_added_rated": 0,
"charge_port_cold_weather_mode": false,
"charge_port_color": "<invalid>",
"charge_port_door_open": true,
"charge_port_latch": "Engaged",
"charge_rate": 0,
"charger_actual_current": 0,
"charger_phases": null,
"charger_pilot_current": 16,
"charger_power": 0,
"charger_voltage": 2,
"charging_state": "Stopped",
"conn_charge_cable": "IEC",
"est_battery_range": 275.04,
"fast_charger_brand": "<invalid>",
"fast_charger_present": false,
"fast_charger_type": "ACSingleWireCAN",
"ideal_battery_range": 266.87,
"max_range_charge_counter": 0,
"minutes_to_full_charge": "bad value",
"not_enough_power_to_heat": null,
"off_peak_charging_enabled": false,
"off_peak_charging_times": "all_week",
"off_peak_hours_end_time": 900,
"preconditioning_enabled": false,
"preconditioning_times": "all_week",
"scheduled_charging_mode": "Off",
"scheduled_charging_pending": false,
"scheduled_charging_start_time": null,
"scheduled_charging_start_time_app": 600,
"scheduled_departure_time": 1704837600,
"scheduled_departure_time_minutes": 480,
"supercharger_session_trip_planner": false,
"time_to_full_charge": null,
"timestamp": null,
"trip_charging": false,
"usable_battery_level": 77,
"user_charge_enable_request": true
},
"climate_state": {
"allow_cabin_overheat_protection": true,
"auto_seat_climate_left": false,
"auto_seat_climate_right": false,
"auto_steering_wheel_heat": false,
"battery_heater": true,
"battery_heater_no_power": null,
"cabin_overheat_protection": "Off",
"cabin_overheat_protection_actively_cooling": false,
"climate_keeper_mode": "off",
"cop_activation_temperature": "Low",
"defrost_mode": 0,
"driver_temp_setting": 22,
"fan_status": 0,
"hvac_auto_request": "On",
"inside_temp": 29.8,
"is_auto_conditioning_on": false,
"is_climate_on": false,
"is_front_defroster_on": false,
"is_preconditioning": false,
"is_rear_defroster_on": false,
"left_temp_direction": 251,
"max_avail_temp": 28,
"min_avail_temp": 15,
"outside_temp": 30,
"passenger_temp_setting": 22,
"remote_heater_control_enabled": false,
"right_temp_direction": 251,
"seat_heater_left": 0,
"seat_heater_rear_center": 0,
"seat_heater_rear_left": 0,
"seat_heater_rear_right": 0,
"seat_heater_right": 0,
"side_mirror_heaters": false,
"steering_wheel_heat_level": 0,
"steering_wheel_heater": false,
"supports_fan_only_cabin_overheat_protection": true,
"timestamp": 1705707520649,
"wiper_blade_heater": false
},
"drive_state": {
"active_route_latitude": 30.2226265,
"active_route_longitude": -97.6236871,
"active_route_miles_to_arrival": 0,
"active_route_minutes_to_arrival": 0,
"active_route_traffic_minutes_delay": 0,
"gps_as_of": 1701129612,
"heading": 185,
"latitude": -30.222626,
"longitude": -97.6236871,
"native_latitude": -30.222626,
"native_location_supported": 1,
"native_longitude": -97.6236871,
"native_type": "wgs",
"power": -7,
"shift_state": null,
"speed": null,
"timestamp": 1705707520649
},
"gui_settings": {
"gui_24_hour_time": false,
"gui_charge_rate_units": "kW",
"gui_distance_units": "km/hr",
"gui_range_display": "Rated",
"gui_temperature_units": "C",
"gui_tirepressure_units": "Psi",
"show_range_units": false,
"timestamp": 1705707520649
},
"vehicle_config": {
"aux_park_lamps": "Eu",
"badge_version": 1,
"can_accept_navigation_requests": true,
"can_actuate_trunks": true,
"car_special_type": "base",
"car_type": "model3",
"charge_port_type": "CCS",
"cop_user_set_temp_supported": false,
"dashcam_clip_save_supported": true,
"default_charge_to_max": false,
"driver_assist": "TeslaAP3",
"ece_restrictions": false,
"efficiency_package": "M32021",
"eu_vehicle": true,
"exterior_color": "DeepBlue",
"exterior_trim": "Black",
"exterior_trim_override": "",
"has_air_suspension": false,
"has_ludicrous_mode": false,
"has_seat_cooling": false,
"headlamp_type": "Global",
"interior_trim_type": "White2",
"key_version": 2,
"motorized_charge_port": true,
"paint_color_override": "0,9,25,0.7,0.04",
"performance_package": "Base",
"plg": true,
"pws": true,
"rear_drive_unit": "PM216MOSFET",
"rear_seat_heaters": 1,
"rear_seat_type": 0,
"rhd": true,
"roof_color": "RoofColorGlass",
"seat_type": null,
"spoiler_type": "None",
"sun_roof_installed": null,
"supports_qr_pairing": false,
"third_row_seats": "None",
"timestamp": 1705707520649,
"trim_badging": "74d",
"use_range_badging": true,
"utc_offset": 36000,
"webcam_selfie_supported": true,
"webcam_supported": true,
"wheel_type": "Pinwheel18CapKit"
},
"vehicle_state": {
"api_version": 71,
"autopark_state_v2": "unavailable",
"calendar_supported": true,
"car_version": "2023.44.30.8 06f534d46010",
"center_display_state": 0,
"dashcam_clip_save_available": true,
"dashcam_state": "Recording",
"df": 0,
"dr": 0,
"fd_window": 1,
"feature_bitmask": "fbdffbff,187f",
"fp_window": 1,
"ft": 1,
"is_user_present": true,
"locked": false,
"media_info": {
"audio_volume": 2.6667,
"audio_volume_increment": 0.333333,
"audio_volume_max": 10.333333,
"media_playback_status": "Stopped",
"now_playing_album": "",
"now_playing_artist": "",
"now_playing_duration": 0,
"now_playing_elapsed": 0,
"now_playing_source": "Spotify",
"now_playing_station": "",
"now_playing_title": ""
},
"media_state": {
"remote_control_enabled": true
},
"notifications_supported": true,
"odometer": 6481.019282,
"parsed_calendar_supported": true,
"pf": 0,
"pr": 0,
"rd_window": 1,
"remote_start": false,
"remote_start_enabled": true,
"remote_start_supported": true,
"rp_window": 1,
"rt": 1,
"santa_mode": 0,
"sentry_mode": false,
"sentry_mode_available": true,
"service_mode": false,
"service_mode_plus": false,
"software_update": {
"download_perc": 0,
"expected_duration_sec": 2700,
"install_perc": 1,
"status": "",
"version": " "
},
"speed_limit_mode": {
"active": false,
"current_limit_mph": 69,
"max_limit_mph": 120,
"min_limit_mph": 50,
"pin_code_set": true
},
"timestamp": 1705707520649,
"tpms_hard_warning_fl": false,
"tpms_hard_warning_fr": false,
"tpms_hard_warning_rl": false,
"tpms_hard_warning_rr": false,
"tpms_last_seen_pressure_time_fl": 1705700812,
"tpms_last_seen_pressure_time_fr": 1705700793,
"tpms_last_seen_pressure_time_rl": 1705700794,
"tpms_last_seen_pressure_time_rr": 1705700823,
"tpms_pressure_fl": 2.775,
"tpms_pressure_fr": 2.8,
"tpms_pressure_rl": 2.775,
"tpms_pressure_rr": 2.775,
"tpms_rcp_front_value": 2.9,
"tpms_rcp_rear_value": 2.9,
"tpms_soft_warning_fl": false,
"tpms_soft_warning_fr": false,
"tpms_soft_warning_rl": false,
"tpms_soft_warning_rr": false,
"valet_mode": false,
"valet_pin_needed": false,
"vehicle_name": "Test",
"vehicle_self_test_progress": 0,
"vehicle_self_test_requested": false,
"webcam_available": true
}
}
}

129
tests/components/tesla_fleet/snapshots/test_init.ambr Normal file
View file

@ -0,0 +1,129 @@
# serializer version: 1
# name: test_devices[{('tesla_fleet', '123456')}]
DeviceRegistryEntrySnapshot({
'area_id': None,
'config_entries': <ANY>,
'configuration_url': None,
'connections': set({
}),
'disabled_by': None,
'entry_type': None,
'hw_version': None,
'id': <ANY>,
'identifiers': set({
tuple(
'tesla_fleet',
'123456',
),
}),
'is_new': False,
'labels': set({
}),
'manufacturer': 'Tesla',
'model': 'Powerwall 2, Tesla Backup Gateway 2',
'model_id': None,
'name': 'Energy Site',
'name_by_user': None,
'primary_config_entry': <ANY>,
'serial_number': '123456',
'suggested_area': None,
'sw_version': None,
'via_device_id': None,
})
# ---
# name: test_devices[{('tesla_fleet', 'LRWXF7EK4KC700000')}]
DeviceRegistryEntrySnapshot({
'area_id': None,
'config_entries': <ANY>,
'configuration_url': None,
'connections': set({
}),
'disabled_by': None,
'entry_type': None,
'hw_version': None,
'id': <ANY>,
'identifiers': set({
tuple(
'tesla_fleet',
'LRWXF7EK4KC700000',
),
}),
'is_new': False,
'labels': set({
}),
'manufacturer': 'Tesla',
'model': 'Model X',
'model_id': None,
'name': 'Test',
'name_by_user': None,
'primary_config_entry': <ANY>,
'serial_number': 'LRWXF7EK4KC700000',
'suggested_area': None,
'sw_version': None,
'via_device_id': None,
})
# ---
# name: test_devices[{('tesla_fleet', 'abd-123')}]
DeviceRegistryEntrySnapshot({
'area_id': None,
'config_entries': <ANY>,
'configuration_url': None,
'connections': set({
}),
'disabled_by': None,
'entry_type': None,
'hw_version': None,
'id': <ANY>,
'identifiers': set({
tuple(
'tesla_fleet',
'abd-123',
),
}),
'is_new': False,
'labels': set({
}),
'manufacturer': 'Tesla',
'model': 'Gen 3 Wall Connector',
'model_id': None,
'name': 'Wall Connector',
'name_by_user': None,
'primary_config_entry': <ANY>,
'serial_number': '123',
'suggested_area': None,
'sw_version': None,
'via_device_id': <ANY>,
})
# ---
# name: test_devices[{('tesla_fleet', 'bcd-234')}]
DeviceRegistryEntrySnapshot({
'area_id': None,
'config_entries': <ANY>,
'configuration_url': None,
'connections': set({
}),
'disabled_by': None,
'entry_type': None,
'hw_version': None,
'id': <ANY>,
'identifiers': set({
tuple(
'tesla_fleet',
'bcd-234',
),
}),
'is_new': False,
'labels': set({
}),
'manufacturer': 'Tesla',
'model': 'Gen 3 Wall Connector',
'model_id': None,
'name': 'Wall Connector',
'name_by_user': None,
'primary_config_entry': <ANY>,
'serial_number': '234',
'suggested_area': None,
'sw_version': None,
'via_device_id': <ANY>,
})
# ---

3424
tests/components/tesla_fleet/snapshots/test_sensor.ambr Normal file
View file

File diff suppressed because it is too large Load diff

224
tests/components/tesla_fleet/test_config_flow.py Normal file
View file

@ -0,0 +1,224 @@
"""Test the Tesla Fleet config flow."""
from unittest.mock import patch
from urllib.parse import parse_qs, urlparse
import pytest
from homeassistant.components.tesla_fleet.application_credentials import (
AUTHORIZE_URL,
CLIENT_ID,
TOKEN_URL,
)
from homeassistant.components.tesla_fleet.const import DOMAIN, SCOPES
from homeassistant.config_entries import SOURCE_REAUTH, SOURCE_USER
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers import config_entry_oauth2_flow
from tests.common import MockConfigEntry
from tests.test_util.aiohttp import AiohttpClientMocker
from tests.typing import ClientSessionGenerator
REDIRECT = "https://example.com/auth/external/callback"
UNIQUE_ID = "uid"
@pytest.fixture
async def access_token(hass: HomeAssistant) -> dict[str, str | list[str]]:
"""Return a valid access token."""
return config_entry_oauth2_flow._encode_jwt(
hass,
{
"sub": UNIQUE_ID,
"aud": [],
"scp": [
"vehicle_device_data",
"vehicle_cmds",
"vehicle_charging_cmds",
"energy_device_data",
"energy_cmds",
"offline_access",
"openid",
],
"ou_code": "NA",
},
)
@pytest.mark.usefixtures("current_request_with_host")
async def test_full_flow(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
access_token,
) -> None:
"""Check full flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
state = config_entry_oauth2_flow._encode_jwt(
hass,
{
"flow_id": result["flow_id"],
"redirect_uri": REDIRECT,
},
)
assert result["type"] is FlowResultType.EXTERNAL_STEP
assert result["url"].startswith(AUTHORIZE_URL)
parsed_url = urlparse(result["url"])
parsed_query = parse_qs(parsed_url.query)
assert parsed_query["response_type"][0] == "code"
assert parsed_query["client_id"][0] == CLIENT_ID
assert parsed_query["redirect_uri"][0] == REDIRECT
assert parsed_query["state"][0] == state
assert parsed_query["scope"][0] == " ".join(SCOPES)
assert parsed_query["code_challenge"][0] is not None
client = await hass_client_no_auth()
resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
assert resp.status == 200
assert resp.headers["content-type"] == "text/html; charset=utf-8"
aioclient_mock.clear_requests()
aioclient_mock.post(
TOKEN_URL,
json={
"refresh_token": "mock-refresh-token",
"access_token": access_token,
"type": "Bearer",
"expires_in": 60,
},
)
with patch(
"homeassistant.components.tesla_fleet.async_setup_entry", return_value=True
) as mock_setup:
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
assert len(mock_setup.mock_calls) == 1
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == UNIQUE_ID
assert "result" in result
assert result["result"].unique_id == UNIQUE_ID
assert "token" in result["result"].data
assert result["result"].data["token"]["access_token"] == access_token
assert result["result"].data["token"]["refresh_token"] == "mock-refresh-token"
@pytest.mark.usefixtures("current_request_with_host")
async def test_reauthentication(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
access_token,
) -> None:
"""Test Tesla Fleet reauthentication."""
old_entry = MockConfigEntry(
domain=DOMAIN,
unique_id=UNIQUE_ID,
version=1,
data={},
)
old_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={
"source": SOURCE_REAUTH,
"unique_id": old_entry.unique_id,
"entry_id": old_entry.entry_id,
},
data=old_entry.data,
)
flows = hass.config_entries.flow.async_progress()
assert len(flows) == 1
result = await hass.config_entries.flow.async_configure(flows[0]["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": "mock-refresh-token",
"access_token": access_token,
"type": "Bearer",
"expires_in": 60,
},
)
with patch(
"homeassistant.components.tesla_fleet.async_setup_entry", return_value=True
):
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reauth_successful"
@pytest.mark.usefixtures("current_request_with_host")
async def test_reauth_account_mismatch(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
access_token,
) -> None:
"""Test Tesla Fleet reauthentication with different account."""
old_entry = MockConfigEntry(domain=DOMAIN, unique_id="baduid", version=1, data={})
old_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={
"source": SOURCE_REAUTH,
"unique_id": old_entry.unique_id,
"entry_id": old_entry.entry_id,
},
data=old_entry.data,
)
flows = hass.config_entries.flow.async_progress()
result = await hass.config_entries.flow.async_configure(flows[0]["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": "mock-refresh-token",
"access_token": access_token,
"type": "Bearer",
"expires_in": 60,
},
)
with patch(
"homeassistant.components.tesla_fleet.async_setup_entry", return_value=True
):
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reauth_account_mismatch"

327
tests/components/tesla_fleet/test_init.py Normal file
View file

@ -0,0 +1,327 @@
"""Test the Tesla Fleet init."""
from freezegun.api import FrozenDateTimeFactory
import pytest
from syrupy import SnapshotAssertion
from tesla_fleet_api.exceptions import (
InvalidToken,
LoginRequired,
OAuthExpired,
RateLimited,
TeslaFleetError,
VehicleOffline,
)
from homeassistant.components.tesla_fleet.coordinator import (
ENERGY_INTERVAL,
ENERGY_INTERVAL_SECONDS,
VEHICLE_INTERVAL,
VEHICLE_INTERVAL_SECONDS,
VEHICLE_WAIT,
)
from homeassistant.components.tesla_fleet.models import TeslaFleetData
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from . import setup_platform
from .const import VEHICLE_ASLEEP, VEHICLE_DATA_ALT
from tests.common import MockConfigEntry, async_fire_time_changed
ERRORS = [
(InvalidToken, ConfigEntryState.SETUP_ERROR),
(OAuthExpired, ConfigEntryState.SETUP_ERROR),
(LoginRequired, ConfigEntryState.SETUP_ERROR),
(TeslaFleetError, ConfigEntryState.SETUP_RETRY),
]
async def test_load_unload(
hass: HomeAssistant,
normal_config_entry: MockConfigEntry,
) -> None:
"""Test load and unload."""
await setup_platform(hass, normal_config_entry)
assert normal_config_entry.state is ConfigEntryState.LOADED
assert isinstance(normal_config_entry.runtime_data, TeslaFleetData)
assert await hass.config_entries.async_unload(normal_config_entry.entry_id)
await hass.async_block_till_done()
assert normal_config_entry.state is ConfigEntryState.NOT_LOADED
assert not hasattr(normal_config_entry, "runtime_data")
@pytest.mark.parametrize(("side_effect", "state"), ERRORS)
async def test_init_error(
hass: HomeAssistant,
normal_config_entry: MockConfigEntry,
mock_products,
side_effect,
state,
) -> None:
"""Test init with errors."""
mock_products.side_effect = side_effect
await setup_platform(hass, normal_config_entry)
assert normal_config_entry.state is state
# Test devices
async def test_devices(
hass: HomeAssistant,
normal_config_entry: MockConfigEntry,
device_registry: dr.DeviceRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test device registry."""
await setup_platform(hass, normal_config_entry)
devices = dr.async_entries_for_config_entry(
device_registry, normal_config_entry.entry_id
)
for device in devices:
assert device == snapshot(name=f"{device.identifiers}")
# Vehicle Coordinator
async def test_vehicle_refresh_offline(
hass: HomeAssistant,
normal_config_entry: MockConfigEntry,
mock_vehicle_state,
mock_vehicle_data,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test coordinator refresh with an error."""
await setup_platform(hass, normal_config_entry)
assert normal_config_entry.state is ConfigEntryState.LOADED
mock_vehicle_state.assert_called_once()
mock_vehicle_data.assert_called_once()
mock_vehicle_state.reset_mock()
mock_vehicle_data.reset_mock()
# Test the unlikely condition that a vehicle state is online but actually offline
mock_vehicle_data.side_effect = VehicleOffline
freezer.tick(VEHICLE_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
mock_vehicle_state.assert_called_once()
mock_vehicle_data.assert_called_once()
mock_vehicle_state.reset_mock()
mock_vehicle_data.reset_mock()
# Test the normal condition that a vehcile state is offline
mock_vehicle_state.return_value = VEHICLE_ASLEEP
freezer.tick(VEHICLE_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
mock_vehicle_state.assert_called_once()
mock_vehicle_data.assert_not_called()
@pytest.mark.parametrize(("side_effect"), ERRORS)
async def test_vehicle_refresh_error(
hass: HomeAssistant,
normal_config_entry: MockConfigEntry,
mock_vehicle_state,
side_effect,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test coordinator refresh makes entity unavailable."""
await setup_platform(hass, normal_config_entry)
mock_vehicle_state.side_effect = side_effect
freezer.tick(VEHICLE_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert (state := hass.states.get("sensor.test_battery_level"))
assert state.state == "unavailable"
async def test_vehicle_refresh_ratelimited(
hass: HomeAssistant,
normal_config_entry: MockConfigEntry,
mock_vehicle_data,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test coordinator refresh handles 429."""
mock_vehicle_data.side_effect = RateLimited(
{"after": VEHICLE_INTERVAL_SECONDS + 10}
)
await setup_platform(hass, normal_config_entry)
assert (state := hass.states.get("sensor.test_battery_level"))
assert state.state == "unknown"
assert mock_vehicle_data.call_count == 1
freezer.tick(VEHICLE_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
# Should not call for another 10 seconds
assert mock_vehicle_data.call_count == 1
freezer.tick(VEHICLE_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert mock_vehicle_data.call_count == 2
async def test_vehicle_sleep(
hass: HomeAssistant,
normal_config_entry: MockConfigEntry,
mock_vehicle_data,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test coordinator refresh with an error."""
await setup_platform(hass, normal_config_entry)
assert mock_vehicle_data.call_count == 1
freezer.tick(VEHICLE_WAIT + VEHICLE_INTERVAL)
async_fire_time_changed(hass)
# Let vehicle sleep, no updates for 15 minutes
await hass.async_block_till_done()
assert mock_vehicle_data.call_count == 2
freezer.tick(VEHICLE_INTERVAL)
async_fire_time_changed(hass)
# No polling, call_count should not increase
await hass.async_block_till_done()
assert mock_vehicle_data.call_count == 2
freezer.tick(VEHICLE_INTERVAL)
async_fire_time_changed(hass)
# No polling, call_count should not increase
await hass.async_block_till_done()
assert mock_vehicle_data.call_count == 2
freezer.tick(VEHICLE_WAIT)
async_fire_time_changed(hass)
# Vehicle didn't sleep, go back to normal
await hass.async_block_till_done()
assert mock_vehicle_data.call_count == 3
freezer.tick(VEHICLE_INTERVAL)
async_fire_time_changed(hass)
# Regular polling
await hass.async_block_till_done()
assert mock_vehicle_data.call_count == 4
mock_vehicle_data.return_value = VEHICLE_DATA_ALT
freezer.tick(VEHICLE_INTERVAL)
async_fire_time_changed(hass)
# Vehicle active
await hass.async_block_till_done()
assert mock_vehicle_data.call_count == 5
freezer.tick(VEHICLE_WAIT)
async_fire_time_changed(hass)
# Dont let sleep when active
await hass.async_block_till_done()
assert mock_vehicle_data.call_count == 6
freezer.tick(VEHICLE_WAIT)
async_fire_time_changed(hass)
# Dont let sleep when active
await hass.async_block_till_done()
assert mock_vehicle_data.call_count == 7
# Test Energy Live Coordinator
@pytest.mark.parametrize(("side_effect", "state"), ERRORS)
async def test_energy_live_refresh_error(
hass: HomeAssistant,
normal_config_entry: MockConfigEntry,
mock_live_status,
side_effect,
state,
) -> None:
"""Test coordinator refresh with an error."""
mock_live_status.side_effect = side_effect
await setup_platform(hass, normal_config_entry)
assert normal_config_entry.state is state
# Test Energy Site Coordinator
@pytest.mark.parametrize(("side_effect", "state"), ERRORS)
async def test_energy_site_refresh_error(
hass: HomeAssistant,
normal_config_entry: MockConfigEntry,
mock_site_info,
side_effect,
state,
) -> None:
"""Test coordinator refresh with an error."""
mock_site_info.side_effect = side_effect
await setup_platform(hass, normal_config_entry)
assert normal_config_entry.state is state
async def test_energy_live_refresh_ratelimited(
hass: HomeAssistant,
normal_config_entry: MockConfigEntry,
mock_live_status,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test coordinator refresh handles 429."""
await setup_platform(hass, normal_config_entry)
mock_live_status.side_effect = RateLimited({"after": ENERGY_INTERVAL_SECONDS + 10})
freezer.tick(ENERGY_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert mock_live_status.call_count == 2
freezer.tick(ENERGY_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
# Should not call for another 10 seconds
assert mock_live_status.call_count == 2
freezer.tick(ENERGY_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert mock_live_status.call_count == 3
async def test_energy_info_refresh_ratelimited(
hass: HomeAssistant,
normal_config_entry: MockConfigEntry,
mock_site_info,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test coordinator refresh handles 429."""
await setup_platform(hass, normal_config_entry)
mock_site_info.side_effect = RateLimited({"after": ENERGY_INTERVAL_SECONDS + 10})
freezer.tick(ENERGY_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert mock_site_info.call_count == 2
freezer.tick(ENERGY_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
# Should not call for another 10 seconds
assert mock_site_info.call_count == 2
freezer.tick(ENERGY_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert mock_site_info.call_count == 3

41
tests/components/tesla_fleet/test_sensor.py Normal file
View file

@ -0,0 +1,41 @@
"""Test the Tesla Fleet sensor platform."""
from freezegun.api import FrozenDateTimeFactory
import pytest
from syrupy import SnapshotAssertion
from homeassistant.components.tesla_fleet.coordinator import VEHICLE_INTERVAL
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from . import assert_entities, assert_entities_alt, setup_platform
from .const import VEHICLE_DATA_ALT
from tests.common import MockConfigEntry, async_fire_time_changed
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_sensors(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
normal_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
freezer: FrozenDateTimeFactory,
mock_vehicle_data,
) -> None:
"""Tests that the sensor entities are correct."""
freezer.move_to("2024-01-01 00:00:00+00:00")
await setup_platform(hass, normal_config_entry, [Platform.SENSOR])
assert_entities(hass, normal_config_entry.entry_id, entity_registry, snapshot)
# Coordinator refresh
mock_vehicle_data.return_value = VEHICLE_DATA_ALT
freezer.tick(VEHICLE_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert_entities_alt(hass, normal_config_entry.entry_id, entity_registry, snapshot)