Add services to Teslemetry (#119119)

* Add custom services

* Fixes

* wip

* Test coverage

* Update homeassistant/components/teslemetry/__init__.py

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

* Add error translations

* Translate command error

* Fix test

* Expand on comment as requested

* Remove impossible cases

---------

Co-authored-by: G Johansson <goran.johansson@shiftit.se>
This commit is contained in:
Brett Adams 2024-06-25 20:44:06 +10:00 committed by GitHub
parent 3a5acd6a57
commit de8bccb650
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 869 additions and 0 deletions

View File

@ -15,8 +15,11 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_ACCESS_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
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.typing import ConfigType
from .const import DOMAIN, LOGGER, MODELS
from .coordinator import (
@ -25,6 +28,7 @@ from .coordinator import (
TeslemetryVehicleDataCoordinator,
)
from .models import TeslemetryData, TeslemetryEnergyData, TeslemetryVehicleData
from .services import async_register_services
PLATFORMS: Final = [
Platform.BINARY_SENSOR,
@ -43,6 +47,14 @@ PLATFORMS: Final = [
type TeslemetryConfigEntry = ConfigEntry[TeslemetryData]
CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the Telemetry integration."""
async_register_services(hass)
return True
async def async_setup_entry(hass: HomeAssistant, entry: TeslemetryConfigEntry) -> bool:
"""Set up Teslemetry config."""
@ -65,6 +77,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: TeslemetryConfigEntry) -
except TeslaFleetError as e:
raise ConfigEntryNotReady from e
device_registry = dr.async_get(hass)
# Create array of classes
vehicles: list[TeslemetryVehicleData] = []
energysites: list[TeslemetryEnergyData] = []
@ -143,6 +157,12 @@ async def async_setup_entry(hass: HomeAssistant, entry: TeslemetryConfigEntry) -
if models:
energysite.device["model"] = ", ".join(sorted(models))
# 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, **energysite.device
)
# Setup Platforms
entry.runtime_data = TeslemetryData(vehicles, energysites, scopes)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)

View File

@ -257,5 +257,13 @@
"default": "mdi:speedometer-slow"
}
}
},
"services": {
"navigation_gps_request": "mdi:crosshairs-gps",
"set_scheduled_charging": "mdi:timeline-clock-outline",
"set_scheduled_departure": "mdi:home-clock",
"speed_limit": "mdi:car-speed-limiter",
"valet_mode": "mdi:speedometer-slow",
"time_of_use": "mdi:clock-time-eight-outline"
}
}

View File

@ -0,0 +1,321 @@
"""Service calls for the Teslemetry integration."""
import logging
import voluptuous as vol
from voluptuous import All, Range
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_DEVICE_ID, CONF_LATITUDE, CONF_LONGITUDE
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.helpers import config_validation as cv, device_registry as dr
from .const import DOMAIN
from .helpers import handle_command, handle_vehicle_command, wake_up_vehicle
from .models import TeslemetryEnergyData, TeslemetryVehicleData
_LOGGER = logging.getLogger(__name__)
# Attributes
ATTR_ID = "id"
ATTR_GPS = "gps"
ATTR_TYPE = "type"
ATTR_VALUE = "value"
ATTR_LOCALE = "locale"
ATTR_ORDER = "order"
ATTR_TIMESTAMP = "timestamp"
ATTR_FIELDS = "fields"
ATTR_ENABLE = "enable"
ATTR_TIME = "time"
ATTR_PIN = "pin"
ATTR_TOU_SETTINGS = "tou_settings"
ATTR_PRECONDITIONING_ENABLED = "preconditioning_enabled"
ATTR_PRECONDITIONING_WEEKDAYS = "preconditioning_weekdays_only"
ATTR_DEPARTURE_TIME = "departure_time"
ATTR_OFF_PEAK_CHARGING_ENABLED = "off_peak_charging_enabled"
ATTR_OFF_PEAK_CHARGING_WEEKDAYS = "off_peak_charging_weekdays_only"
ATTR_END_OFF_PEAK_TIME = "end_off_peak_time"
# Services
SERVICE_NAVIGATE_ATTR_GPS_REQUEST = "navigation_gps_request"
SERVICE_SET_SCHEDULED_CHARGING = "set_scheduled_charging"
SERVICE_SET_SCHEDULED_DEPARTURE = "set_scheduled_departure"
SERVICE_VALET_MODE = "valet_mode"
SERVICE_SPEED_LIMIT = "speed_limit"
SERVICE_TIME_OF_USE = "time_of_use"
def async_get_device_for_service_call(
hass: HomeAssistant, call: ServiceCall
) -> dr.DeviceEntry:
"""Get the device entry related to a service call."""
device_id = call.data[CONF_DEVICE_ID]
device_registry = dr.async_get(hass)
if (device_entry := device_registry.async_get(device_id)) is None:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="invalid_device",
translation_placeholders={"device_id": device_id},
)
return device_entry
def async_get_config_for_device(
hass: HomeAssistant, device_entry: dr.DeviceEntry
) -> ConfigEntry:
"""Get the config entry related to a device entry."""
config_entry: ConfigEntry
for entry_id in device_entry.config_entries:
if entry := hass.config_entries.async_get_entry(entry_id):
if entry.domain == DOMAIN:
config_entry = entry
return config_entry
def async_get_vehicle_for_entry(
hass: HomeAssistant, device: dr.DeviceEntry, config: ConfigEntry
) -> TeslemetryVehicleData:
"""Get the vehicle data for a config entry."""
vehicle_data: TeslemetryVehicleData
assert device.serial_number is not None
for vehicle in config.runtime_data.vehicles:
if vehicle.vin == device.serial_number:
vehicle_data = vehicle
return vehicle_data
def async_get_energy_site_for_entry(
hass: HomeAssistant, device: dr.DeviceEntry, config: ConfigEntry
) -> TeslemetryEnergyData:
"""Get the energy site data for a config entry."""
energy_data: TeslemetryEnergyData
assert device.serial_number is not None
for energysite in config.runtime_data.energysites:
if str(energysite.id) == device.serial_number:
energy_data = energysite
return energy_data
def async_register_services(hass: HomeAssistant) -> None: # noqa: C901
"""Set up the Teslemetry services."""
async def navigate_gps_request(call: ServiceCall) -> None:
"""Send lat,lon,order with a vehicle."""
device = async_get_device_for_service_call(hass, call)
config = async_get_config_for_device(hass, device)
vehicle = async_get_vehicle_for_entry(hass, device, config)
await wake_up_vehicle(vehicle)
await handle_vehicle_command(
vehicle.api.navigation_gps_request(
lat=call.data[ATTR_GPS][CONF_LATITUDE],
lon=call.data[ATTR_GPS][CONF_LONGITUDE],
order=call.data.get(ATTR_ORDER),
)
)
hass.services.async_register(
DOMAIN,
SERVICE_NAVIGATE_ATTR_GPS_REQUEST,
navigate_gps_request,
schema=vol.Schema(
{
vol.Required(CONF_DEVICE_ID): cv.string,
vol.Required(ATTR_GPS): {
vol.Required(CONF_LATITUDE): cv.latitude,
vol.Required(CONF_LONGITUDE): cv.longitude,
},
vol.Optional(ATTR_ORDER): cv.positive_int,
}
),
)
async def set_scheduled_charging(call: ServiceCall) -> None:
"""Configure fleet telemetry."""
device = async_get_device_for_service_call(hass, call)
config = async_get_config_for_device(hass, device)
vehicle = async_get_vehicle_for_entry(hass, device, config)
time: int | None = None
# Convert time to minutes since minute
if "time" in call.data:
(hours, minutes, *seconds) = call.data["time"].split(":")
time = int(hours) * 60 + int(minutes)
elif call.data["enable"]:
raise ServiceValidationError(
translation_domain=DOMAIN, translation_key="set_scheduled_charging_time"
)
await wake_up_vehicle(vehicle)
await handle_vehicle_command(
vehicle.api.set_scheduled_charging(enable=call.data["enable"], time=time)
)
hass.services.async_register(
DOMAIN,
SERVICE_SET_SCHEDULED_CHARGING,
set_scheduled_charging,
schema=vol.Schema(
{
vol.Required(CONF_DEVICE_ID): cv.string,
vol.Required(ATTR_ENABLE): bool,
vol.Optional(ATTR_TIME): str,
}
),
)
async def set_scheduled_departure(call: ServiceCall) -> None:
"""Configure fleet telemetry."""
device = async_get_device_for_service_call(hass, call)
config = async_get_config_for_device(hass, device)
vehicle = async_get_vehicle_for_entry(hass, device, config)
enable = call.data.get("enable", True)
# Preconditioning
preconditioning_enabled = call.data.get(ATTR_PRECONDITIONING_ENABLED, False)
preconditioning_weekdays_only = call.data.get(
ATTR_PRECONDITIONING_WEEKDAYS, False
)
departure_time: int | None = None
if ATTR_DEPARTURE_TIME in call.data:
(hours, minutes, *seconds) = call.data[ATTR_DEPARTURE_TIME].split(":")
departure_time = int(hours) * 60 + int(minutes)
elif preconditioning_enabled:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="set_scheduled_departure_preconditioning",
)
# Off peak charging
off_peak_charging_enabled = call.data.get(ATTR_OFF_PEAK_CHARGING_ENABLED, False)
off_peak_charging_weekdays_only = call.data.get(
ATTR_OFF_PEAK_CHARGING_WEEKDAYS, False
)
end_off_peak_time: int | None = None
if ATTR_END_OFF_PEAK_TIME in call.data:
(hours, minutes, *seconds) = call.data[ATTR_END_OFF_PEAK_TIME].split(":")
end_off_peak_time = int(hours) * 60 + int(minutes)
elif off_peak_charging_enabled:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="set_scheduled_departure_off_peak",
)
await wake_up_vehicle(vehicle)
await handle_vehicle_command(
vehicle.api.set_scheduled_departure(
enable,
preconditioning_enabled,
preconditioning_weekdays_only,
departure_time,
off_peak_charging_enabled,
off_peak_charging_weekdays_only,
end_off_peak_time,
)
)
hass.services.async_register(
DOMAIN,
SERVICE_SET_SCHEDULED_DEPARTURE,
set_scheduled_departure,
schema=vol.Schema(
{
vol.Required(CONF_DEVICE_ID): cv.string,
vol.Optional(ATTR_ENABLE): bool,
vol.Optional(ATTR_PRECONDITIONING_ENABLED): bool,
vol.Optional(ATTR_PRECONDITIONING_WEEKDAYS): bool,
vol.Optional(ATTR_DEPARTURE_TIME): str,
vol.Optional(ATTR_OFF_PEAK_CHARGING_ENABLED): bool,
vol.Optional(ATTR_OFF_PEAK_CHARGING_WEEKDAYS): bool,
vol.Optional(ATTR_END_OFF_PEAK_TIME): str,
}
),
)
async def valet_mode(call: ServiceCall) -> None:
"""Configure fleet telemetry."""
device = async_get_device_for_service_call(hass, call)
config = async_get_config_for_device(hass, device)
vehicle = async_get_vehicle_for_entry(hass, device, config)
await wake_up_vehicle(vehicle)
await handle_vehicle_command(
vehicle.api.set_valet_mode(
call.data.get("enable"), call.data.get("pin", "")
)
)
hass.services.async_register(
DOMAIN,
SERVICE_VALET_MODE,
valet_mode,
schema=vol.Schema(
{
vol.Required(CONF_DEVICE_ID): cv.string,
vol.Required(ATTR_ENABLE): cv.boolean,
vol.Required(ATTR_PIN): All(cv.positive_int, Range(min=1000, max=9999)),
}
),
)
async def speed_limit(call: ServiceCall) -> None:
"""Configure fleet telemetry."""
device = async_get_device_for_service_call(hass, call)
config = async_get_config_for_device(hass, device)
vehicle = async_get_vehicle_for_entry(hass, device, config)
await wake_up_vehicle(vehicle)
enable = call.data.get("enable")
if enable is True:
await handle_vehicle_command(
vehicle.api.speed_limit_activate(call.data.get("pin"))
)
elif enable is False:
await handle_vehicle_command(
vehicle.api.speed_limit_deactivate(call.data.get("pin"))
)
hass.services.async_register(
DOMAIN,
SERVICE_SPEED_LIMIT,
speed_limit,
schema=vol.Schema(
{
vol.Required(CONF_DEVICE_ID): cv.string,
vol.Required(ATTR_ENABLE): cv.boolean,
vol.Required(ATTR_PIN): All(cv.positive_int, Range(min=1000, max=9999)),
}
),
)
async def time_of_use(call: ServiceCall) -> None:
"""Configure time of use settings."""
device = async_get_device_for_service_call(hass, call)
config = async_get_config_for_device(hass, device)
site = async_get_energy_site_for_entry(hass, device, config)
resp = await handle_command(
site.api.time_of_use_settings(call.data.get(ATTR_TOU_SETTINGS))
)
if "error" in resp:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="command_error",
translation_placeholders={"error": resp["error"]},
)
hass.services.async_register(
DOMAIN,
SERVICE_TIME_OF_USE,
time_of_use,
schema=vol.Schema(
{
vol.Required(CONF_DEVICE_ID): cv.string,
vol.Required(ATTR_TOU_SETTINGS): dict,
}
),
)

View File

@ -0,0 +1,132 @@
navigation_gps_request:
fields:
device_id:
required: true
selector:
device:
filter:
- integration: teslemetry
gps:
required: true
example: '{"latitude": -27.9699373, "longitude": 153.4081865}'
selector:
location:
radius: false
order:
required: false
default: 1
selector:
number:
time_of_use:
fields:
device_id:
required: true
selector:
device:
filter:
- integration: teslemetry
tou_settings:
required: true
selector:
object:
set_scheduled_charging:
fields:
device_id:
required: true
selector:
device:
filter:
integration: teslemetry
enable:
required: true
default: true
selector:
boolean:
time:
required: false
selector:
time:
set_scheduled_departure:
fields:
device_id:
required: true
selector:
device:
filter:
integration: teslemetry
enable:
required: false
default: true
selector:
boolean:
preconditioning_enabled:
required: false
default: false
selector:
boolean:
preconditioning_weekdays_only:
required: false
default: false
selector:
boolean:
departure_time:
required: false
selector:
time:
off_peak_charging_enabled:
required: false
default: false
selector:
boolean:
off_peak_charging_weekdays_only:
required: false
default: false
selector:
boolean:
end_off_peak_time:
required: false
selector:
time:
valet_mode:
fields:
device_id:
required: true
selector:
device:
filter:
integration: teslemetry
enable:
required: true
selector:
boolean:
pin:
required: true
selector:
number:
min: 1000
max: 9999
mode: box
speed_limit:
fields:
device_id:
required: true
selector:
device:
filter:
integration: teslemetry
enable:
required: true
selector:
boolean:
pin:
required: true
selector:
number:
min: 1000
max: 9999
mode: box

View File

@ -473,6 +473,156 @@
},
"invalid_cop_temp": {
"message": "Cabin overheat protection does not support that temperature"
},
"set_scheduled_charging_time": {
"message": "Time required to complete the operation"
},
"set_scheduled_departure_preconditioning": {
"message": "Departure time required to enable preconditioning"
},
"set_scheduled_departure_off_peak": {
"message": "To enable scheduled departure, end off peak time is required."
},
"invalid_device": {
"message": "Invalid device ID: {device_id}"
},
"no_config_entry_for_device": {
"message": "No config entry for device ID: {device_id}"
},
"no_vehicle_data_for_device": {
"message": "No vehicle data for device ID: {device_id}"
},
"no_energy_site_data_for_device": {
"message": "No energy site data for device ID: {device_id}"
},
"command_error": {
"message": "Command returned error: {error}"
}
},
"services": {
"navigation_gps_request": {
"description": "Set vehicle navigation to the provided latitude/longitude coordinates.",
"fields": {
"device_id": {
"description": "Vehicle to share to.",
"name": "Vehicle"
},
"gps": {
"description": "Location to navigate to.",
"name": "Location"
},
"order": {
"description": "Order for this destination if specifying multiple destinations.",
"name": "Order"
}
},
"name": "Navigate to coordinates"
},
"set_scheduled_charging": {
"description": "Sets a time at which charging should be completed.",
"fields": {
"device_id": {
"description": "Vehicle to schedule.",
"name": "Vehicle"
},
"enable": {
"description": "Enable or disable scheduled charging.",
"name": "Enable"
},
"time": {
"description": "Time to start charging.",
"name": "Time"
}
},
"name": "Set scheduled charging"
},
"set_scheduled_departure": {
"description": "Sets a time at which departure should be completed.",
"fields": {
"departure_time": {
"description": "Time to be preconditioned by.",
"name": "Departure time"
},
"device_id": {
"description": "Vehicle to schedule.",
"name": "Vehicle"
},
"enable": {
"description": "Enable or disable scheduled departure.",
"name": "Enable"
},
"end_off_peak_time": {
"description": "Time to complete charging by.",
"name": "End off peak time"
},
"off_peak_charging_enabled": {
"description": "Enable off peak charging.",
"name": "Off peak charging enabled"
},
"off_peak_charging_weekdays_only": {
"description": "Enable off peak charging on weekdays only.",
"name": "Off peak charging weekdays only"
},
"preconditioning_enabled": {
"description": "Enable preconditioning.",
"name": "Preconditioning enabled"
},
"preconditioning_weekdays_only": {
"description": "Enable preconditioning on weekdays only.",
"name": "Preconditioning weekdays only"
}
},
"name": "Set scheduled departure"
},
"speed_limit": {
"description": "Activate the speed limit of the vehicle.",
"fields": {
"device_id": {
"description": "Vehicle to limit.",
"name": "Vehicle"
},
"enable": {
"description": "Enable or disable speed limit.",
"name": "Enable"
},
"pin": {
"description": "4 digit PIN.",
"name": "PIN"
}
},
"name": "Set speed limit"
},
"time_of_use": {
"description": "Update the time of use settings for the energy site.",
"fields": {
"device_id": {
"description": "Energy Site to configure.",
"name": "Energy Site"
},
"tou_settings": {
"description": "See https://developer.tesla.com/docs/fleet-api#time_of_use_settings for details.",
"name": "Settings"
}
},
"name": "Time of use settings"
},
"valet_mode": {
"description": "Activate the valet mode of the vehicle.",
"fields": {
"device_id": {
"description": "Vehicle to limit.",
"name": "Vehicle"
},
"enable": {
"description": "Enable or disable valet mode.",
"name": "Enable"
},
"pin": {
"description": "4 digit PIN.",
"name": "PIN"
}
},
"name": "Set valet mode"
}
}
}

View File

@ -0,0 +1,238 @@
"""Test the Teslemetry services."""
from unittest.mock import patch
import pytest
from homeassistant.components.teslemetry.const import DOMAIN
from homeassistant.components.teslemetry.services import (
ATTR_DEPARTURE_TIME,
ATTR_ENABLE,
ATTR_END_OFF_PEAK_TIME,
ATTR_GPS,
ATTR_OFF_PEAK_CHARGING_ENABLED,
ATTR_OFF_PEAK_CHARGING_WEEKDAYS,
ATTR_PIN,
ATTR_PRECONDITIONING_ENABLED,
ATTR_PRECONDITIONING_WEEKDAYS,
ATTR_TIME,
ATTR_TOU_SETTINGS,
SERVICE_NAVIGATE_ATTR_GPS_REQUEST,
SERVICE_SET_SCHEDULED_CHARGING,
SERVICE_SET_SCHEDULED_DEPARTURE,
SERVICE_SPEED_LIMIT,
SERVICE_TIME_OF_USE,
SERVICE_VALET_MODE,
)
from homeassistant.const import CONF_DEVICE_ID, CONF_LATITUDE, CONF_LONGITUDE
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.helpers import entity_registry as er
from . import setup_platform
from .const import COMMAND_ERROR, COMMAND_OK
lat = -27.9699373
lon = 153.3726526
async def test_services(
hass: HomeAssistant,
) -> None:
"""Tests that the custom services are correct."""
await setup_platform(hass)
entity_registry = er.async_get(hass)
# Get a vehicle device ID
vehicle_device = entity_registry.async_get("sensor.test_charging").device_id
energy_device = entity_registry.async_get(
"sensor.energy_site_battery_power"
).device_id
with patch(
"homeassistant.components.teslemetry.VehicleSpecific.navigation_gps_request",
return_value=COMMAND_OK,
) as navigation_gps_request:
await hass.services.async_call(
DOMAIN,
SERVICE_NAVIGATE_ATTR_GPS_REQUEST,
{
CONF_DEVICE_ID: vehicle_device,
ATTR_GPS: {CONF_LATITUDE: lat, CONF_LONGITUDE: lon},
},
blocking=True,
)
navigation_gps_request.assert_called_once()
with patch(
"homeassistant.components.teslemetry.VehicleSpecific.set_scheduled_charging",
return_value=COMMAND_OK,
) as set_scheduled_charging:
await hass.services.async_call(
DOMAIN,
SERVICE_SET_SCHEDULED_CHARGING,
{
CONF_DEVICE_ID: vehicle_device,
ATTR_ENABLE: True,
ATTR_TIME: "6:00",
},
blocking=True,
)
set_scheduled_charging.assert_called_once()
with pytest.raises(ServiceValidationError):
await hass.services.async_call(
DOMAIN,
SERVICE_SET_SCHEDULED_CHARGING,
{
CONF_DEVICE_ID: vehicle_device,
ATTR_ENABLE: True,
},
blocking=True,
)
with patch(
"homeassistant.components.teslemetry.VehicleSpecific.set_scheduled_departure",
return_value=COMMAND_OK,
) as set_scheduled_departure:
await hass.services.async_call(
DOMAIN,
SERVICE_SET_SCHEDULED_DEPARTURE,
{
CONF_DEVICE_ID: vehicle_device,
ATTR_ENABLE: True,
ATTR_PRECONDITIONING_ENABLED: True,
ATTR_PRECONDITIONING_WEEKDAYS: False,
ATTR_DEPARTURE_TIME: "6:00",
ATTR_OFF_PEAK_CHARGING_ENABLED: True,
ATTR_OFF_PEAK_CHARGING_WEEKDAYS: False,
ATTR_END_OFF_PEAK_TIME: "5:00",
},
blocking=True,
)
set_scheduled_departure.assert_called_once()
with pytest.raises(ServiceValidationError):
await hass.services.async_call(
DOMAIN,
SERVICE_SET_SCHEDULED_DEPARTURE,
{
CONF_DEVICE_ID: vehicle_device,
ATTR_ENABLE: True,
ATTR_PRECONDITIONING_ENABLED: True,
},
blocking=True,
)
with pytest.raises(ServiceValidationError):
await hass.services.async_call(
DOMAIN,
SERVICE_SET_SCHEDULED_DEPARTURE,
{
CONF_DEVICE_ID: vehicle_device,
ATTR_ENABLE: True,
ATTR_OFF_PEAK_CHARGING_ENABLED: True,
},
blocking=True,
)
with patch(
"homeassistant.components.teslemetry.VehicleSpecific.set_valet_mode",
return_value=COMMAND_OK,
) as set_valet_mode:
await hass.services.async_call(
DOMAIN,
SERVICE_VALET_MODE,
{
CONF_DEVICE_ID: vehicle_device,
ATTR_ENABLE: True,
ATTR_PIN: 1234,
},
blocking=True,
)
set_valet_mode.assert_called_once()
with patch(
"homeassistant.components.teslemetry.VehicleSpecific.speed_limit_activate",
return_value=COMMAND_OK,
) as speed_limit_activate:
await hass.services.async_call(
DOMAIN,
SERVICE_SPEED_LIMIT,
{
CONF_DEVICE_ID: vehicle_device,
ATTR_ENABLE: True,
ATTR_PIN: 1234,
},
blocking=True,
)
speed_limit_activate.assert_called_once()
with patch(
"homeassistant.components.teslemetry.VehicleSpecific.speed_limit_deactivate",
return_value=COMMAND_OK,
) as speed_limit_deactivate:
await hass.services.async_call(
DOMAIN,
SERVICE_SPEED_LIMIT,
{
CONF_DEVICE_ID: vehicle_device,
ATTR_ENABLE: False,
ATTR_PIN: 1234,
},
blocking=True,
)
speed_limit_deactivate.assert_called_once()
with patch(
"homeassistant.components.teslemetry.EnergySpecific.time_of_use_settings",
return_value=COMMAND_OK,
) as set_time_of_use:
await hass.services.async_call(
DOMAIN,
SERVICE_TIME_OF_USE,
{
CONF_DEVICE_ID: energy_device,
ATTR_TOU_SETTINGS: {},
},
blocking=True,
)
set_time_of_use.assert_called_once()
with (
patch(
"homeassistant.components.teslemetry.EnergySpecific.time_of_use_settings",
return_value=COMMAND_ERROR,
) as set_time_of_use,
pytest.raises(HomeAssistantError),
):
await hass.services.async_call(
DOMAIN,
SERVICE_TIME_OF_USE,
{
CONF_DEVICE_ID: energy_device,
ATTR_TOU_SETTINGS: {},
},
blocking=True,
)
async def test_service_validation_errors(
hass: HomeAssistant,
) -> None:
"""Tests that the custom services handle bad data."""
await setup_platform(hass)
# Bad device ID
with pytest.raises(ServiceValidationError):
await hass.services.async_call(
DOMAIN,
SERVICE_NAVIGATE_ATTR_GPS_REQUEST,
{
CONF_DEVICE_ID: "nope",
ATTR_GPS: {CONF_LATITUDE: lat, CONF_LONGITUDE: lon},
},
blocking=True,
)