Add climate platform to Tesla Fleet (#123169)

* Add climate

* docstring

* Add tests

* Fix limited scope situation

* Add another test

* Add icons

* Type vehicle data

* Replace inline temperatures

* Fix handle_vehicle_command type

* Fix preset turning HVAC off

* Fix cop_mode check

* Use constants

* Reference docs in command signing error

* Move to a read-only check

* Remove raise_for

* Fixes

* Tests

* Remove raise_for_signing

* Remove unused strings

* Fix async_set_temperature

* Correct tests

* Remove HVAC modes at startup in read-only mode

* Fix order of init actions to set hvac_modes correctly

* Fix no temp test

* Add handle command type

* Docstrings

* fix matches and fix a bug

* Split tests

* Fix issues from rebase
This commit is contained in:
Brett Adams 2024-09-03 22:38:47 +10:00 committed by GitHub
parent c321bd70e1
commit 6ecc5c19a2
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
11 changed files with 1407 additions and 14 deletions

View File

@ -39,7 +39,12 @@ from .coordinator import (
from .models import TeslaFleetData, TeslaFleetEnergyData, TeslaFleetVehicleData
from .oauth import TeslaSystemImplementation
PLATFORMS: Final = [Platform.BINARY_SENSOR, Platform.DEVICE_TRACKER, Platform.SENSOR]
PLATFORMS: Final = [
Platform.BINARY_SENSOR,
Platform.CLIMATE,
Platform.DEVICE_TRACKER,
Platform.SENSOR,
]
type TeslaFleetConfigEntry = ConfigEntry[TeslaFleetData]
@ -53,8 +58,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: TeslaFleetConfigEntry) -
session = async_get_clientsession(hass)
token = jwt.decode(access_token, options={"verify_signature": False})
scopes = token["scp"]
region = token["ou_code"].lower()
scopes: list[Scope] = [Scope(s) for s in token["scp"]]
region: str = token["ou_code"].lower()
OAuth2FlowHandler.async_register_implementation(
hass,
@ -133,6 +138,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: TeslaFleetConfigEntry) -
coordinator=coordinator,
vin=vin,
device=device,
signing=product["command_signing"] == "required",
)
)
elif "energy_site_id" in product and hasattr(tesla, "energy"):

View File

@ -0,0 +1,330 @@
"""Climate platform for Tesla Fleet integration."""
from __future__ import annotations
from itertools import chain
from typing import Any, cast
from tesla_fleet_api.const import CabinOverheatProtectionTemp, Scope
from homeassistant.components.climate import (
ATTR_HVAC_MODE,
ClimateEntity,
ClimateEntityFeature,
HVACMode,
)
from homeassistant.const import (
ATTR_TEMPERATURE,
PRECISION_HALVES,
PRECISION_WHOLE,
UnitOfTemperature,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ServiceValidationError
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import TeslaFleetConfigEntry
from .const import DOMAIN, TeslaFleetClimateSide
from .entity import TeslaFleetVehicleEntity
from .helpers import handle_vehicle_command
from .models import TeslaFleetVehicleData
DEFAULT_MIN_TEMP = 15
DEFAULT_MAX_TEMP = 28
PARALLEL_UPDATES = 0
async def async_setup_entry(
hass: HomeAssistant,
entry: TeslaFleetConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the Tesla Fleet Climate platform from a config entry."""
async_add_entities(
chain(
(
TeslaFleetClimateEntity(
vehicle, TeslaFleetClimateSide.DRIVER, entry.runtime_data.scopes
)
for vehicle in entry.runtime_data.vehicles
),
(
TeslaFleetCabinOverheatProtectionEntity(
vehicle, entry.runtime_data.scopes
)
for vehicle in entry.runtime_data.vehicles
),
)
)
class TeslaFleetClimateEntity(TeslaFleetVehicleEntity, ClimateEntity):
"""Tesla Fleet vehicle climate entity."""
_attr_precision = PRECISION_HALVES
_attr_temperature_unit = UnitOfTemperature.CELSIUS
_attr_hvac_modes = [HVACMode.HEAT_COOL, HVACMode.OFF]
_attr_supported_features = (
ClimateEntityFeature.TURN_ON
| ClimateEntityFeature.TURN_OFF
| ClimateEntityFeature.TARGET_TEMPERATURE
| ClimateEntityFeature.PRESET_MODE
)
_attr_preset_modes = ["off", "keep", "dog", "camp"]
_enable_turn_on_off_backwards_compatibility = False
def __init__(
self,
data: TeslaFleetVehicleData,
side: TeslaFleetClimateSide,
scopes: Scope,
) -> None:
"""Initialize the climate."""
self.read_only = Scope.VEHICLE_CMDS not in scopes or data.signing
if self.read_only:
self._attr_supported_features = ClimateEntityFeature(0)
self._attr_hvac_modes = []
super().__init__(
data,
side,
)
def _async_update_attrs(self) -> None:
"""Update the attributes of the entity."""
value = self.get("climate_state_is_climate_on")
if value is None:
self._attr_hvac_mode = None
elif value:
self._attr_hvac_mode = HVACMode.HEAT_COOL
else:
self._attr_hvac_mode = HVACMode.OFF
# If not scoped, prevent the user from changing the HVAC mode by making it the only option
if self._attr_hvac_mode and self.read_only:
self._attr_hvac_modes = [self._attr_hvac_mode]
self._attr_current_temperature = self.get("climate_state_inside_temp")
self._attr_target_temperature = self.get(f"climate_state_{self.key}_setting")
self._attr_preset_mode = self.get("climate_state_climate_keeper_mode")
self._attr_min_temp = cast(
float, self.get("climate_state_min_avail_temp", DEFAULT_MIN_TEMP)
)
self._attr_max_temp = cast(
float, self.get("climate_state_max_avail_temp", DEFAULT_MAX_TEMP)
)
async def async_turn_on(self) -> None:
"""Set the climate state to on."""
await self.wake_up_if_asleep()
await handle_vehicle_command(self.api.auto_conditioning_start())
self._attr_hvac_mode = HVACMode.HEAT_COOL
self.async_write_ha_state()
async def async_turn_off(self) -> None:
"""Set the climate state to off."""
await self.wake_up_if_asleep()
await handle_vehicle_command(self.api.auto_conditioning_stop())
self._attr_hvac_mode = HVACMode.OFF
self._attr_preset_mode = self._attr_preset_modes[0]
self.async_write_ha_state()
async def async_set_temperature(self, **kwargs: Any) -> None:
"""Set the climate temperature."""
if ATTR_TEMPERATURE not in kwargs:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="missing_temperature",
)
temp = kwargs[ATTR_TEMPERATURE]
await self.wake_up_if_asleep()
await handle_vehicle_command(
self.api.set_temps(
driver_temp=temp,
passenger_temp=temp,
)
)
self._attr_target_temperature = temp
if mode := kwargs.get(ATTR_HVAC_MODE):
# Set HVAC mode will call write_ha_state
await self.async_set_hvac_mode(mode)
else:
self.async_write_ha_state()
async def async_set_hvac_mode(self, hvac_mode: HVACMode) -> None:
"""Set the climate mode and state."""
if hvac_mode not in self.hvac_modes:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="invalid_hvac_mode",
translation_placeholders={"hvac_mode": hvac_mode},
)
if hvac_mode == HVACMode.OFF:
await self.async_turn_off()
else:
await self.async_turn_on()
async def async_set_preset_mode(self, preset_mode: str) -> None:
"""Set the climate preset mode."""
await self.wake_up_if_asleep()
await handle_vehicle_command(
self.api.set_climate_keeper_mode(
climate_keeper_mode=self._attr_preset_modes.index(preset_mode)
)
)
self._attr_preset_mode = preset_mode
if preset_mode != self._attr_preset_modes[0]:
self._attr_hvac_mode = HVACMode.HEAT_COOL
self.async_write_ha_state()
COP_MODES = {
"Off": HVACMode.OFF,
"On": HVACMode.COOL,
"FanOnly": HVACMode.FAN_ONLY,
}
# String to celsius
COP_LEVELS = {
"Low": 30,
"Medium": 35,
"High": 40,
}
# Celsius to IntEnum
TEMP_LEVELS = {
30: CabinOverheatProtectionTemp.LOW,
35: CabinOverheatProtectionTemp.MEDIUM,
40: CabinOverheatProtectionTemp.HIGH,
}
class TeslaFleetCabinOverheatProtectionEntity(TeslaFleetVehicleEntity, ClimateEntity):
"""Tesla Fleet vehicle cabin overheat protection entity."""
_attr_precision = PRECISION_WHOLE
_attr_target_temperature_step = 5
_attr_min_temp = COP_LEVELS["Low"]
_attr_max_temp = COP_LEVELS["High"]
_attr_temperature_unit = UnitOfTemperature.CELSIUS
_attr_hvac_modes = list(COP_MODES.values())
_enable_turn_on_off_backwards_compatibility = False
_attr_entity_registry_enabled_default = False
def __init__(
self,
data: TeslaFleetVehicleData,
scopes: Scope,
) -> None:
"""Initialize the cabin overheat climate entity."""
# Scopes
self.read_only = Scope.VEHICLE_CMDS not in scopes or data.signing
# Supported Features
if self.read_only:
self._attr_supported_features = ClimateEntityFeature(0)
self._attr_hvac_modes = []
else:
self._attr_supported_features = (
ClimateEntityFeature.TURN_ON | ClimateEntityFeature.TURN_OFF
)
super().__init__(data, "climate_state_cabin_overheat_protection")
def _async_update_attrs(self) -> None:
"""Update the attributes of the entity."""
if (state := self.get("climate_state_cabin_overheat_protection")) is None:
self._attr_hvac_mode = None
else:
self._attr_hvac_mode = COP_MODES.get(state)
# If not scoped, prevent the user from changing the HVAC mode by making it the only option
if self._attr_hvac_mode and self.read_only:
self._attr_hvac_modes = [self._attr_hvac_mode]
if (level := self.get("climate_state_cop_activation_temperature")) is None:
self._attr_target_temperature = None
else:
self._attr_target_temperature = COP_LEVELS.get(level)
self._attr_current_temperature = self.get("climate_state_inside_temp")
@property
def supported_features(self) -> ClimateEntityFeature:
"""Return the list of supported features."""
if not self.read_only and self.get(
"vehicle_config_cop_user_set_temp_supported"
):
return (
self._attr_supported_features | ClimateEntityFeature.TARGET_TEMPERATURE
)
return self._attr_supported_features
async def async_turn_on(self) -> None:
"""Set the climate state to on."""
await self.async_set_hvac_mode(HVACMode.COOL)
async def async_turn_off(self) -> None:
"""Set the climate state to off."""
await self.async_set_hvac_mode(HVACMode.OFF)
async def async_set_temperature(self, **kwargs: Any) -> None:
"""Set the climate temperature."""
if ATTR_TEMPERATURE not in kwargs:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="missing_temperature",
)
temp = kwargs[ATTR_TEMPERATURE]
if (cop_mode := TEMP_LEVELS.get(temp)) is None:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="invalid_cop_temp",
)
await self.wake_up_if_asleep()
await handle_vehicle_command(self.api.set_cop_temp(cop_mode))
self._attr_target_temperature = temp
if mode := kwargs.get(ATTR_HVAC_MODE):
await self._async_set_cop(mode)
self.async_write_ha_state()
async def async_set_hvac_mode(self, hvac_mode: HVACMode) -> None:
"""Set the climate mode and state."""
await self.wake_up_if_asleep()
await self._async_set_cop(hvac_mode)
self.async_write_ha_state()
async def _async_set_cop(self, hvac_mode: HVACMode) -> None:
if hvac_mode == HVACMode.OFF:
await handle_vehicle_command(
self.api.set_cabin_overheat_protection(on=False, fan_only=False)
)
elif hvac_mode == HVACMode.COOL:
await handle_vehicle_command(
self.api.set_cabin_overheat_protection(on=True, fan_only=False)
)
elif hvac_mode == HVACMode.FAN_ONLY:
await handle_vehicle_command(
self.api.set_cabin_overheat_protection(on=True, fan_only=True)
)
self._attr_hvac_mode = hvac_mode

View File

@ -41,3 +41,10 @@ class TeslaFleetState(StrEnum):
ONLINE = "online"
ASLEEP = "asleep"
OFFLINE = "offline"
class TeslaFleetClimateSide(StrEnum):
"""Tesla Fleet Climate Keeper Modes."""
DRIVER = "driver_temp"
PASSENGER = "passenger_temp"

View File

@ -14,6 +14,7 @@ from .coordinator import (
TeslaFleetEnergySiteLiveCoordinator,
TeslaFleetVehicleDataCoordinator,
)
from .helpers import wake_up_vehicle
from .models import TeslaFleetEnergyData, TeslaFleetVehicleData
@ -27,6 +28,7 @@ class TeslaFleetEntity(
"""Parent class for all TeslaFleet entities."""
_attr_has_entity_name = True
read_only: bool
def __init__(
self,
@ -100,6 +102,10 @@ class TeslaFleetVehicleEntity(TeslaFleetEntity):
"""Return a specific value from coordinator data."""
return self.coordinator.data.get(self.key)
async def wake_up_if_asleep(self) -> None:
"""Wake up the vehicle if its asleep."""
await wake_up_vehicle(self.vehicle)
class TeslaFleetEnergyLiveEntity(TeslaFleetEntity):
"""Parent class for TeslaFleet Energy Site Live entities."""

View File

@ -0,0 +1,80 @@
"""Tesla Fleet helper functions."""
import asyncio
from collections.abc import Awaitable
from typing import Any
from tesla_fleet_api.exceptions import TeslaFleetError
from homeassistant.exceptions import HomeAssistantError
from .const import DOMAIN, LOGGER, TeslaFleetState
from .models import TeslaFleetVehicleData
async def wake_up_vehicle(vehicle: TeslaFleetVehicleData) -> None:
"""Wake up a vehicle."""
async with vehicle.wakelock:
times = 0
while vehicle.coordinator.data["state"] != TeslaFleetState.ONLINE:
try:
if times == 0:
cmd = await vehicle.api.wake_up()
else:
cmd = await vehicle.api.vehicle()
state = cmd["response"]["state"]
except TeslaFleetError as e:
raise HomeAssistantError(str(e)) from e
vehicle.coordinator.data["state"] = state
if state != TeslaFleetState.ONLINE:
times += 1
if times >= 4: # Give up after 30 seconds total
raise HomeAssistantError("Could not wake up vehicle")
await asyncio.sleep(times * 5)
async def handle_command(command: Awaitable) -> dict[str, Any]:
"""Handle a command."""
try:
result = await command
except TeslaFleetError as e:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="command_failed",
translation_placeholders={"message": e.message},
) from e
LOGGER.debug("Command result: %s", result)
return result
async def handle_vehicle_command(command: Awaitable) -> bool:
"""Handle a vehicle command."""
result = await handle_command(command)
if (response := result.get("response")) is None:
if error := result.get("error"):
# No response with error
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="command_error",
translation_placeholders={"error": error},
)
# No response without error (unexpected)
raise HomeAssistantError(f"Unknown response: {response}")
if (result := response.get("result")) is not True:
if reason := response.get("reason"):
if reason in ("already_set", "not_charging", "requested"):
# Reason is acceptable
return result
# Result of false with reason
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="command_reason",
translation_placeholders={"reason": reason},
)
# Result of false without reason (unexpected)
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="command_no_reason",
)
# Response with result of true
return result

View File

@ -38,6 +38,20 @@
}
}
},
"climate": {
"driver_temp": {
"state_attributes": {
"preset_mode": {
"state": {
"off": "mdi:power",
"keep": "mdi:fan",
"dog": "mdi:dog",
"camp": "mdi:tent"
}
}
}
}
},
"device_tracker": {
"location": {
"default": "mdi:map-marker"

View File

@ -2,6 +2,7 @@
from __future__ import annotations
import asyncio
from dataclasses import dataclass
from tesla_fleet_api import EnergySpecific, VehicleSpecific
@ -33,6 +34,8 @@ class TeslaFleetVehicleData:
coordinator: TeslaFleetVehicleDataCoordinator
vin: str
device: DeviceInfo
signing: bool
wakelock = asyncio.Lock()
@dataclass

View File

@ -107,6 +107,24 @@
"name": "Tire pressure warning rear right"
}
},
"climate": {
"climate_state_cabin_overheat_protection": {
"name": "Cabin overheat protection"
},
"driver_temp": {
"name": "[%key:component::climate::title%]",
"state_attributes": {
"preset_mode": {
"state": {
"off": "Normal",
"keep": "Keep mode",
"dog": "Dog mode",
"camp": "Camp mode"
}
}
}
}
},
"device_tracker": {
"location": {
"name": "Location"
@ -272,7 +290,28 @@
},
"exceptions": {
"update_failed": {
"message": "{endpoint} data request failed. {message}"
"message": "{endpoint} data request failed: {message}"
},
"command_failed": {
"message": "Command failed: {message}"
},
"command_error": {
"message": "Command returned an error: {error}"
},
"command_reason": {
"message": "Command was unsuccessful: {reason}"
},
"command_no_reason": {
"message": "Command was unsuccessful but did not return a reason why."
},
"invalid_cop_temp": {
"message": "Cabin overheat protection does not support that temperature."
},
"invalid_hvac_mode": {
"message": "Climate mode {hvac_mode} is not supported."
},
"missing_temperature": {
"message": "Temperature is required for this action."
}
}
}

View File

@ -9,10 +9,18 @@ from unittest.mock import AsyncMock, patch
import jwt
import pytest
from tesla_fleet_api.const import Scope
from homeassistant.components.tesla_fleet.const import DOMAIN, SCOPES
from .const import LIVE_STATUS, PRODUCTS, SITE_INFO, VEHICLE_DATA, VEHICLE_ONLINE
from .const import (
COMMAND_OK,
LIVE_STATUS,
PRODUCTS,
SITE_INFO,
VEHICLE_DATA,
VEHICLE_ONLINE,
)
from tests.common import MockConfigEntry
@ -25,16 +33,8 @@ def mock_expires_at() -> int:
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:
def create_config_entry(expires_at: int, scopes: list[Scope]) -> MockConfigEntry:
"""Create Tesla Fleet entry in Home Assistant."""
access_token = jwt.encode(
{
"sub": UID,
@ -64,6 +64,32 @@ def normal_config_entry(expires_at: int, scopes: list[str]) -> MockConfigEntry:
)
@pytest.fixture
def normal_config_entry(expires_at: int) -> MockConfigEntry:
"""Create Tesla Fleet entry in Home Assistant."""
return create_config_entry(expires_at, SCOPES)
@pytest.fixture
def noscope_config_entry(expires_at: int) -> MockConfigEntry:
"""Create Tesla Fleet entry in Home Assistant without scopes."""
return create_config_entry(expires_at, [Scope.OPENID, Scope.OFFLINE_ACCESS])
@pytest.fixture
def readonly_config_entry(expires_at: int) -> MockConfigEntry:
"""Create Tesla Fleet entry in Home Assistant without scopes."""
return create_config_entry(
expires_at,
[
Scope.OPENID,
Scope.OFFLINE_ACCESS,
Scope.VEHICLE_DEVICE_DATA,
Scope.ENERGY_DEVICE_DATA,
],
)
@pytest.fixture(autouse=True)
def mock_products() -> Generator[AsyncMock]:
"""Mock Tesla Fleet Api products method."""
@ -131,3 +157,13 @@ def mock_find_server() -> Generator[AsyncMock]:
"homeassistant.components.tesla_fleet.TeslaFleetApi.find_server",
) as mock_find_server:
yield mock_find_server
@pytest.fixture
def mock_request():
"""Mock all Tesla Fleet API requests."""
with patch(
"homeassistant.components.tesla_fleet.TeslaFleetApi._request",
return_value=COMMAND_OK,
) as mock_request:
yield mock_request

View File

@ -0,0 +1,422 @@
# serializer version: 1
# name: test_climate[climate.test_cabin_overheat_protection-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'hvac_modes': list([
<HVACMode.OFF: 'off'>,
<HVACMode.COOL: 'cool'>,
<HVACMode.FAN_ONLY: 'fan_only'>,
]),
'max_temp': 40,
'min_temp': 30,
'target_temp_step': 5,
}),
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'climate',
'entity_category': None,
'entity_id': 'climate.test_cabin_overheat_protection',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Cabin overheat protection',
'platform': 'tesla_fleet',
'previous_unique_id': None,
'supported_features': <ClimateEntityFeature: 385>,
'translation_key': 'climate_state_cabin_overheat_protection',
'unique_id': 'LRWXF7EK4KC700000-climate_state_cabin_overheat_protection',
'unit_of_measurement': None,
})
# ---
# name: test_climate[climate.test_cabin_overheat_protection-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'current_temperature': 30,
'friendly_name': 'Test Cabin overheat protection',
'hvac_modes': list([
<HVACMode.OFF: 'off'>,
<HVACMode.COOL: 'cool'>,
<HVACMode.FAN_ONLY: 'fan_only'>,
]),
'max_temp': 40,
'min_temp': 30,
'supported_features': <ClimateEntityFeature: 385>,
'target_temp_step': 5,
'temperature': 40,
}),
'context': <ANY>,
'entity_id': 'climate.test_cabin_overheat_protection',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'cool',
})
# ---
# name: test_climate[climate.test_climate-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'hvac_modes': list([
<HVACMode.HEAT_COOL: 'heat_cool'>,
<HVACMode.OFF: 'off'>,
]),
'max_temp': 28.0,
'min_temp': 15.0,
'preset_modes': list([
'off',
'keep',
'dog',
'camp',
]),
}),
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'climate',
'entity_category': None,
'entity_id': 'climate.test_climate',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Climate',
'platform': 'tesla_fleet',
'previous_unique_id': None,
'supported_features': <ClimateEntityFeature: 401>,
'translation_key': <TeslaFleetClimateSide.DRIVER: 'driver_temp'>,
'unique_id': 'LRWXF7EK4KC700000-driver_temp',
'unit_of_measurement': None,
})
# ---
# name: test_climate[climate.test_climate-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'current_temperature': 30.0,
'friendly_name': 'Test Climate',
'hvac_modes': list([
<HVACMode.HEAT_COOL: 'heat_cool'>,
<HVACMode.OFF: 'off'>,
]),
'max_temp': 28.0,
'min_temp': 15.0,
'preset_mode': 'keep',
'preset_modes': list([
'off',
'keep',
'dog',
'camp',
]),
'supported_features': <ClimateEntityFeature: 401>,
'temperature': 22.0,
}),
'context': <ANY>,
'entity_id': 'climate.test_climate',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'heat_cool',
})
# ---
# name: test_climate_alt[climate.test_cabin_overheat_protection-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'hvac_modes': list([
<HVACMode.OFF: 'off'>,
<HVACMode.COOL: 'cool'>,
<HVACMode.FAN_ONLY: 'fan_only'>,
]),
'max_temp': 40,
'min_temp': 30,
'target_temp_step': 5,
}),
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'climate',
'entity_category': None,
'entity_id': 'climate.test_cabin_overheat_protection',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Cabin overheat protection',
'platform': 'tesla_fleet',
'previous_unique_id': None,
'supported_features': <ClimateEntityFeature: 384>,
'translation_key': 'climate_state_cabin_overheat_protection',
'unique_id': 'LRWXF7EK4KC700000-climate_state_cabin_overheat_protection',
'unit_of_measurement': None,
})
# ---
# name: test_climate_alt[climate.test_cabin_overheat_protection-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'current_temperature': 30,
'friendly_name': 'Test Cabin overheat protection',
'hvac_modes': list([
<HVACMode.OFF: 'off'>,
<HVACMode.COOL: 'cool'>,
<HVACMode.FAN_ONLY: 'fan_only'>,
]),
'max_temp': 40,
'min_temp': 30,
'supported_features': <ClimateEntityFeature: 384>,
'target_temp_step': 5,
}),
'context': <ANY>,
'entity_id': 'climate.test_cabin_overheat_protection',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_climate_alt[climate.test_climate-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'hvac_modes': list([
<HVACMode.HEAT_COOL: 'heat_cool'>,
<HVACMode.OFF: 'off'>,
]),
'max_temp': 28.0,
'min_temp': 15.0,
'preset_modes': list([
'off',
'keep',
'dog',
'camp',
]),
}),
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'climate',
'entity_category': None,
'entity_id': 'climate.test_climate',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Climate',
'platform': 'tesla_fleet',
'previous_unique_id': None,
'supported_features': <ClimateEntityFeature: 401>,
'translation_key': <TeslaFleetClimateSide.DRIVER: 'driver_temp'>,
'unique_id': 'LRWXF7EK4KC700000-driver_temp',
'unit_of_measurement': None,
})
# ---
# name: test_climate_alt[climate.test_climate-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'current_temperature': 30.0,
'friendly_name': 'Test Climate',
'hvac_modes': list([
<HVACMode.HEAT_COOL: 'heat_cool'>,
<HVACMode.OFF: 'off'>,
]),
'max_temp': 28.0,
'min_temp': 15.0,
'preset_mode': 'off',
'preset_modes': list([
'off',
'keep',
'dog',
'camp',
]),
'supported_features': <ClimateEntityFeature: 401>,
'temperature': 22.0,
}),
'context': <ANY>,
'entity_id': 'climate.test_climate',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_climate_offline[climate.test_cabin_overheat_protection-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'hvac_modes': list([
<HVACMode.OFF: 'off'>,
<HVACMode.COOL: 'cool'>,
<HVACMode.FAN_ONLY: 'fan_only'>,
]),
'max_temp': 40,
'min_temp': 30,
'target_temp_step': 5,
}),
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'climate',
'entity_category': None,
'entity_id': 'climate.test_cabin_overheat_protection',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Cabin overheat protection',
'platform': 'tesla_fleet',
'previous_unique_id': None,
'supported_features': <ClimateEntityFeature: 384>,
'translation_key': 'climate_state_cabin_overheat_protection',
'unique_id': 'LRWXF7EK4KC700000-climate_state_cabin_overheat_protection',
'unit_of_measurement': None,
})
# ---
# name: test_climate_offline[climate.test_cabin_overheat_protection-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'current_temperature': None,
'friendly_name': 'Test Cabin overheat protection',
'hvac_modes': list([
<HVACMode.OFF: 'off'>,
<HVACMode.COOL: 'cool'>,
<HVACMode.FAN_ONLY: 'fan_only'>,
]),
'max_temp': 40,
'min_temp': 30,
'supported_features': <ClimateEntityFeature: 384>,
'target_temp_step': 5,
}),
'context': <ANY>,
'entity_id': 'climate.test_cabin_overheat_protection',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_climate_offline[climate.test_climate-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'hvac_modes': list([
<HVACMode.HEAT_COOL: 'heat_cool'>,
<HVACMode.OFF: 'off'>,
]),
'max_temp': 28.0,
'min_temp': 15.0,
'preset_modes': list([
'off',
'keep',
'dog',
'camp',
]),
}),
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'climate',
'entity_category': None,
'entity_id': 'climate.test_climate',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Climate',
'platform': 'tesla_fleet',
'previous_unique_id': None,
'supported_features': <ClimateEntityFeature: 401>,
'translation_key': <TeslaFleetClimateSide.DRIVER: 'driver_temp'>,
'unique_id': 'LRWXF7EK4KC700000-driver_temp',
'unit_of_measurement': None,
})
# ---
# name: test_climate_offline[climate.test_climate-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'current_temperature': None,
'friendly_name': 'Test Climate',
'hvac_modes': list([
<HVACMode.HEAT_COOL: 'heat_cool'>,
<HVACMode.OFF: 'off'>,
]),
'max_temp': 28.0,
'min_temp': 15.0,
'preset_mode': None,
'preset_modes': list([
'off',
'keep',
'dog',
'camp',
]),
'supported_features': <ClimateEntityFeature: 401>,
'temperature': None,
}),
'context': <ANY>,
'entity_id': 'climate.test_climate',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---

View File

@ -0,0 +1,450 @@
"""Test the Tesla Fleet climate platform."""
from unittest.mock import AsyncMock, patch
from freezegun.api import FrozenDateTimeFactory
import pytest
from syrupy.assertion import SnapshotAssertion
from tesla_fleet_api.exceptions import InvalidCommand, VehicleOffline
from homeassistant.components.climate import (
ATTR_HVAC_MODE,
ATTR_PRESET_MODE,
ATTR_TARGET_TEMP_HIGH,
ATTR_TARGET_TEMP_LOW,
ATTR_TEMPERATURE,
DOMAIN as CLIMATE_DOMAIN,
SERVICE_SET_HVAC_MODE,
SERVICE_SET_PRESET_MODE,
SERVICE_SET_TEMPERATURE,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
HVACMode,
)
from homeassistant.components.tesla_fleet.coordinator import VEHICLE_INTERVAL
from homeassistant.const import ATTR_ENTITY_ID, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.helpers import entity_registry as er
from . import assert_entities, setup_platform
from .const import (
COMMAND_ERRORS,
COMMAND_IGNORED_REASON,
VEHICLE_ASLEEP,
VEHICLE_DATA_ALT,
VEHICLE_ONLINE,
)
from tests.common import MockConfigEntry, async_fire_time_changed
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_climate(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
entity_registry: er.EntityRegistry,
normal_config_entry: MockConfigEntry,
) -> None:
"""Tests that the climate entities are correct."""
await setup_platform(hass, normal_config_entry, [Platform.CLIMATE])
assert_entities(hass, normal_config_entry.entry_id, entity_registry, snapshot)
async def test_climate_services(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
entity_registry: er.EntityRegistry,
normal_config_entry: MockConfigEntry,
mock_request: AsyncMock,
) -> None:
"""Tests that the climate services work."""
await setup_platform(hass, normal_config_entry, [Platform.CLIMATE])
entity_id = "climate.test_climate"
# Turn On and Set Temp
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_SET_TEMPERATURE,
{
ATTR_ENTITY_ID: [entity_id],
ATTR_TEMPERATURE: 20,
ATTR_HVAC_MODE: HVACMode.HEAT_COOL,
},
blocking=True,
)
state = hass.states.get(entity_id)
assert state.attributes[ATTR_TEMPERATURE] == 20
assert state.state == HVACMode.HEAT_COOL
# Set Temp
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_SET_TEMPERATURE,
{
ATTR_ENTITY_ID: [entity_id],
ATTR_TEMPERATURE: 21,
},
blocking=True,
)
state = hass.states.get(entity_id)
assert state.attributes[ATTR_TEMPERATURE] == 21
# Set Preset
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_SET_PRESET_MODE,
{ATTR_ENTITY_ID: [entity_id], ATTR_PRESET_MODE: "keep"},
blocking=True,
)
state = hass.states.get(entity_id)
assert state.attributes[ATTR_PRESET_MODE] == "keep"
# Set Preset
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_SET_PRESET_MODE,
{ATTR_ENTITY_ID: [entity_id], ATTR_PRESET_MODE: "off"},
blocking=True,
)
state = hass.states.get(entity_id)
assert state.attributes[ATTR_PRESET_MODE] == "off"
# Turn Off
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_SET_HVAC_MODE,
{ATTR_ENTITY_ID: [entity_id], ATTR_HVAC_MODE: HVACMode.OFF},
blocking=True,
)
state = hass.states.get(entity_id)
assert state.state == HVACMode.OFF
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_climate_overheat_protection_services(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
entity_registry: er.EntityRegistry,
normal_config_entry: MockConfigEntry,
mock_request: AsyncMock,
) -> None:
"""Tests that the climate overheat protection services work."""
await setup_platform(hass, normal_config_entry, [Platform.CLIMATE])
entity_id = "climate.test_cabin_overheat_protection"
# Turn On and Set Low
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_SET_TEMPERATURE,
{
ATTR_ENTITY_ID: [entity_id],
ATTR_TEMPERATURE: 30,
ATTR_HVAC_MODE: HVACMode.FAN_ONLY,
},
blocking=True,
)
state = hass.states.get(entity_id)
assert state.attributes[ATTR_TEMPERATURE] == 30
assert state.state == HVACMode.FAN_ONLY
# Set Temp Medium
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_SET_TEMPERATURE,
{
ATTR_ENTITY_ID: [entity_id],
ATTR_TEMPERATURE: 35,
},
blocking=True,
)
state = hass.states.get(entity_id)
assert state.attributes[ATTR_TEMPERATURE] == 35
# Set Temp High
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_SET_TEMPERATURE,
{
ATTR_ENTITY_ID: [entity_id],
ATTR_TEMPERATURE: 40,
},
blocking=True,
)
state = hass.states.get(entity_id)
assert state.attributes[ATTR_TEMPERATURE] == 40
# Turn Off
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: [entity_id]},
blocking=True,
)
state = hass.states.get(entity_id)
assert state.state == HVACMode.OFF
# Turn On
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: [entity_id]},
blocking=True,
)
state = hass.states.get(entity_id)
assert state.state == HVACMode.COOL
# Call set temp with invalid temperature
with pytest.raises(
ServiceValidationError,
match="Cabin overheat protection does not support that temperature",
):
# Invalid Temp
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_SET_TEMPERATURE,
{ATTR_ENTITY_ID: [entity_id], ATTR_TEMPERATURE: 34},
blocking=True,
)
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_climate_alt(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
entity_registry: er.EntityRegistry,
mock_vehicle_data: AsyncMock,
normal_config_entry: MockConfigEntry,
) -> None:
"""Tests that the climate entity is correct."""
mock_vehicle_data.return_value = VEHICLE_DATA_ALT
await setup_platform(hass, normal_config_entry, [Platform.CLIMATE])
assert_entities(hass, normal_config_entry.entry_id, entity_registry, snapshot)
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_climate_offline(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
entity_registry: er.EntityRegistry,
mock_vehicle_data: AsyncMock,
normal_config_entry: MockConfigEntry,
) -> None:
"""Tests that the climate entity is correct."""
mock_vehicle_data.side_effect = VehicleOffline
await setup_platform(hass, normal_config_entry, [Platform.CLIMATE])
assert_entities(hass, normal_config_entry.entry_id, entity_registry, snapshot)
async def test_invalid_error(
hass: HomeAssistant,
normal_config_entry: MockConfigEntry,
) -> None:
"""Tests service error is handled."""
await setup_platform(hass, normal_config_entry, platforms=[Platform.CLIMATE])
entity_id = "climate.test_climate"
with (
patch(
"homeassistant.components.tesla_fleet.VehicleSpecific.auto_conditioning_start",
side_effect=InvalidCommand,
) as mock_on,
pytest.raises(
HomeAssistantError,
match="Command failed: The data request or command is unknown.",
),
):
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: [entity_id]},
blocking=True,
)
mock_on.assert_called_once()
@pytest.mark.parametrize("response", COMMAND_ERRORS)
async def test_errors(
hass: HomeAssistant, response: str, normal_config_entry: MockConfigEntry
) -> None:
"""Tests service reason is handled."""
await setup_platform(hass, normal_config_entry, [Platform.CLIMATE])
entity_id = "climate.test_climate"
with (
patch(
"homeassistant.components.tesla_fleet.VehicleSpecific.auto_conditioning_start",
return_value=response,
) as mock_on,
pytest.raises(HomeAssistantError),
):
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: [entity_id]},
blocking=True,
)
mock_on.assert_called_once()
async def test_ignored_error(
hass: HomeAssistant,
normal_config_entry: MockConfigEntry,
) -> None:
"""Tests ignored error is handled."""
await setup_platform(hass, normal_config_entry, [Platform.CLIMATE])
entity_id = "climate.test_climate"
with patch(
"homeassistant.components.tesla_fleet.VehicleSpecific.auto_conditioning_start",
return_value=COMMAND_IGNORED_REASON,
) as mock_on:
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: [entity_id]},
blocking=True,
)
mock_on.assert_called_once()
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_asleep_or_offline(
hass: HomeAssistant,
mock_vehicle_data: AsyncMock,
mock_wake_up: AsyncMock,
mock_vehicle_state: AsyncMock,
freezer: FrozenDateTimeFactory,
normal_config_entry: MockConfigEntry,
mock_request: AsyncMock,
) -> None:
"""Tests asleep is handled."""
await setup_platform(hass, normal_config_entry, [Platform.CLIMATE])
entity_id = "climate.test_climate"
mock_vehicle_data.assert_called_once()
# Put the vehicle alseep
mock_vehicle_data.reset_mock()
mock_vehicle_data.side_effect = VehicleOffline
freezer.tick(VEHICLE_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
mock_vehicle_data.assert_called_once()
mock_wake_up.reset_mock()
# Run a command but fail trying to wake up the vehicle
mock_wake_up.side_effect = InvalidCommand
with pytest.raises(
HomeAssistantError, match="The data request or command is unknown."
):
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: [entity_id]},
blocking=True,
)
mock_wake_up.assert_called_once()
mock_wake_up.side_effect = None
mock_wake_up.reset_mock()
# Run a command but timeout trying to wake up the vehicle
mock_wake_up.return_value = VEHICLE_ASLEEP
mock_vehicle_state.return_value = VEHICLE_ASLEEP
with (
patch("homeassistant.components.tesla_fleet.helpers.asyncio.sleep"),
pytest.raises(HomeAssistantError, match="Could not wake up vehicle"),
):
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: [entity_id]},
blocking=True,
)
mock_wake_up.assert_called_once()
mock_vehicle_state.assert_called()
mock_wake_up.reset_mock()
mock_vehicle_state.reset_mock()
mock_wake_up.return_value = VEHICLE_ONLINE
mock_vehicle_state.return_value = VEHICLE_ONLINE
# Run a command and wake up the vehicle immediately
await hass.services.async_call(
CLIMATE_DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: [entity_id]}, blocking=True
)
await hass.async_block_till_done()
mock_wake_up.assert_called_once()
async def test_climate_noscope(
hass: HomeAssistant,
readonly_config_entry: MockConfigEntry,
snapshot: SnapshotAssertion,
) -> None:
"""Tests with no command scopes."""
await setup_platform(hass, readonly_config_entry, [Platform.CLIMATE])
entity_id = "climate.test_climate"
with pytest.raises(
ServiceValidationError, match="Climate mode off is not supported"
):
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_SET_HVAC_MODE,
{ATTR_ENTITY_ID: [entity_id], ATTR_HVAC_MODE: HVACMode.OFF},
blocking=True,
)
with pytest.raises(
HomeAssistantError,
match="Entity climate.test_climate does not support this service.",
):
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_SET_TEMPERATURE,
{ATTR_ENTITY_ID: [entity_id], ATTR_TEMPERATURE: 20},
blocking=True,
)
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
@pytest.mark.parametrize(
("entity_id", "high", "low"),
[
("climate.test_climate", 16, 28),
("climate.test_cabin_overheat_protection", 30, 40),
],
)
async def test_climate_notemp(
hass: HomeAssistant,
normal_config_entry: MockConfigEntry,
entity_id: str,
high: int,
low: int,
) -> None:
"""Tests that set temp fails without a temp attribute."""
await setup_platform(hass, normal_config_entry, [Platform.CLIMATE])
with pytest.raises(
ServiceValidationError, match="Temperature is required for this action"
):
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_SET_TEMPERATURE,
{
ATTR_ENTITY_ID: [entity_id],
ATTR_TARGET_TEMP_HIGH: high,
ATTR_TARGET_TEMP_LOW: low,
},
blocking=True,
)