Add lock platform to Tesla Fleet (#126412)

* Add lock platform

* Add lock platform tests

* Fix json

homeassistant/components/tesla_fleet/__init__.py

@@ -44,6 +44,7 @@ PLATFORMS: Final = [
     Platform.CLIMATE,
     Platform.COVER,
     Platform.DEVICE_TRACKER,
+    Platform.LOCK,
     Platform.MEDIA_PLAYER,
     Platform.SELECT,
     Platform.SENSOR,

homeassistant/components/tesla_fleet/icons.json

@@ -65,6 +65,17 @@
         "default": "mdi:routes"
       }
     },
+    "lock": {
+      "charge_state_charge_port_latch": {
+        "default": "mdi:ev-plug-tesla"
+      },
+      "vehicle_state_locked": {
+        "state": {
+          "locked": "mdi:car-door-lock",
+          "unlocked": "mdi:car-door-lock-open"
+        }
+      }
+    },
     "select": {
       "climate_state_seat_heater_left": {
         "default": "mdi:car-seat-heater",

homeassistant/components/tesla_fleet/lock.py

@@ -0,0 +1,103 @@
+"""Lock platform for Tesla Fleet integration."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from tesla_fleet_api.const import Scope
+
+from homeassistant.components.lock import LockEntity
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ServiceValidationError
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+
+from . import TeslaFleetConfigEntry
+from .const import DOMAIN
+from .entity import TeslaFleetVehicleEntity
+from .helpers import handle_vehicle_command
+from .models import TeslaFleetVehicleData
+
+ENGAGED = "Engaged"
+
+PARALLEL_UPDATES = 0
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: TeslaFleetConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up the TeslaFleet lock platform from a config entry."""
+
+    async_add_entities(
+        klass(vehicle, Scope.VEHICLE_CMDS in entry.runtime_data.scopes)
+        for klass in (
+            TeslaFleetVehicleLockEntity,
+            TeslaFleetCableLockEntity,
+        )
+        for vehicle in entry.runtime_data.vehicles
+    )
+
+
+class TeslaFleetVehicleLockEntity(TeslaFleetVehicleEntity, LockEntity):
+    """Lock entity for TeslaFleet."""
+
+    def __init__(self, data: TeslaFleetVehicleData, scoped: bool) -> None:
+        """Initialize the lock."""
+        super().__init__(data, "vehicle_state_locked")
+        self.scoped = scoped
+
+    def _async_update_attrs(self) -> None:
+        """Update entity attributes."""
+        self._attr_is_locked = self._value
+
+    async def async_lock(self, **kwargs: Any) -> None:
+        """Lock the doors."""
+        self.raise_for_read_only(Scope.VEHICLE_CMDS)
+        await self.wake_up_if_asleep()
+        await handle_vehicle_command(self.api.door_lock())
+        self._attr_is_locked = True
+        self.async_write_ha_state()
+
+    async def async_unlock(self, **kwargs: Any) -> None:
+        """Unlock the doors."""
+        self.raise_for_read_only(Scope.VEHICLE_CMDS)
+        await self.wake_up_if_asleep()
+        await handle_vehicle_command(self.api.door_unlock())
+        self._attr_is_locked = False
+        self.async_write_ha_state()
+
+
+class TeslaFleetCableLockEntity(TeslaFleetVehicleEntity, LockEntity):
+    """Cable Lock entity for TeslaFleet."""
+
+    def __init__(
+        self,
+        data: TeslaFleetVehicleData,
+        scoped: bool,
+    ) -> None:
+        """Initialize the lock."""
+        super().__init__(data, "charge_state_charge_port_latch")
+        self.scoped = scoped
+
+    def _async_update_attrs(self) -> None:
+        """Update entity attributes."""
+        if self._value is None:
+            self._attr_is_locked = None
+        self._attr_is_locked = self._value == ENGAGED
+
+    async def async_lock(self, **kwargs: Any) -> None:
+        """Charge cable Lock cannot be manually locked."""
+        raise ServiceValidationError(
+            "Insert cable to lock",
+            translation_domain=DOMAIN,
+            translation_key="no_cable",
+        )
+
+    async def async_unlock(self, **kwargs: Any) -> None:
+        """Unlock charge cable lock."""
+        self.raise_for_read_only(Scope.VEHICLE_CMDS)
+        await self.wake_up_if_asleep()
+        await handle_vehicle_command(self.api.charge_port_door_open())
+        self._attr_is_locked = False
+        self.async_write_ha_state()

homeassistant/components/tesla_fleet/strings.json

@@ -150,6 +150,14 @@
         "name": "Route"
       }
     },
+    "lock": {
+      "charge_state_charge_port_latch": {
+        "name": "Charge cable lock"
+      },
+      "vehicle_state_locked": {
+        "name": "[%key:component::lock::title%]"
+      }
+    },
     "media_player": {
       "media": {
         "name": "[%key:component::media_player::title%]"
@@ -443,6 +451,9 @@
     }
   },
   "exceptions": {
+    "no_cable": {
+      "message": "Charge cable will lock automatically when connected"
+    },
     "update_failed": {
       "message": "{endpoint} data request failed: {message}"
     },

tests/components/tesla_fleet/snapshots/test_lock.ambr

@@ -0,0 +1,95 @@
+# serializer version: 1
+# name: test_lock[lock.test_charge_cable_lock-entry]
+  EntityRegistryEntrySnapshot({
+    'aliases': set({
+    }),
+    'area_id': None,
+    'capabilities': None,
+    'config_entry_id': <ANY>,
+    'device_class': None,
+    'device_id': <ANY>,
+    'disabled_by': None,
+    'domain': 'lock',
+    'entity_category': None,
+    'entity_id': 'lock.test_charge_cable_lock',
+    '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': 'Charge cable lock',
+    'platform': 'tesla_fleet',
+    'previous_unique_id': None,
+    'supported_features': 0,
+    'translation_key': 'charge_state_charge_port_latch',
+    'unique_id': 'LRWXF7EK4KC700000-charge_state_charge_port_latch',
+    'unit_of_measurement': None,
+  })
+# ---
+# name: test_lock[lock.test_charge_cable_lock-state]
+  StateSnapshot({
+    'attributes': ReadOnlyDict({
+      'friendly_name': 'Test Charge cable lock',
+      'supported_features': <LockEntityFeature: 0>,
+    }),
+    'context': <ANY>,
+    'entity_id': 'lock.test_charge_cable_lock',
+    'last_changed': <ANY>,
+    'last_reported': <ANY>,
+    'last_updated': <ANY>,
+    'state': 'locked',
+  })
+# ---
+# name: test_lock[lock.test_lock-entry]
+  EntityRegistryEntrySnapshot({
+    'aliases': set({
+    }),
+    'area_id': None,
+    'capabilities': None,
+    'config_entry_id': <ANY>,
+    'device_class': None,
+    'device_id': <ANY>,
+    'disabled_by': None,
+    'domain': 'lock',
+    'entity_category': None,
+    'entity_id': 'lock.test_lock',
+    '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': 'Lock',
+    'platform': 'tesla_fleet',
+    'previous_unique_id': None,
+    'supported_features': 0,
+    'translation_key': 'vehicle_state_locked',
+    'unique_id': 'LRWXF7EK4KC700000-vehicle_state_locked',
+    'unit_of_measurement': None,
+  })
+# ---
+# name: test_lock[lock.test_lock-state]
+  StateSnapshot({
+    'attributes': ReadOnlyDict({
+      'friendly_name': 'Test Lock',
+      'supported_features': <LockEntityFeature: 0>,
+    }),
+    'context': <ANY>,
+    'entity_id': 'lock.test_lock',
+    'last_changed': <ANY>,
+    'last_reported': <ANY>,
+    'last_updated': <ANY>,
+    'state': 'unlocked',
+  })
+# ---

tests/components/tesla_fleet/test_lock.py

@@ -0,0 +1,116 @@
+"""Test the Tesla Fleet lock platform."""
+
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from syrupy import SnapshotAssertion
+from tesla_fleet_api.exceptions import VehicleOffline
+
+from homeassistant.components.lock import (
+    DOMAIN as LOCK_DOMAIN,
+    SERVICE_LOCK,
+    SERVICE_UNLOCK,
+)
+from homeassistant.const import (
+    ATTR_ENTITY_ID,
+    STATE_LOCKED,
+    STATE_UNKNOWN,
+    STATE_UNLOCKED,
+    Platform,
+)
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ServiceValidationError
+from homeassistant.helpers import entity_registry as er
+
+from . import assert_entities, setup_platform
+from .const import COMMAND_OK
+
+from tests.common import MockConfigEntry
+
+
+async def test_lock(
+    hass: HomeAssistant,
+    snapshot: SnapshotAssertion,
+    entity_registry: er.EntityRegistry,
+    normal_config_entry: MockConfigEntry,
+) -> None:
+    """Tests that the lock entities are correct."""
+
+    await setup_platform(hass, normal_config_entry, [Platform.LOCK])
+    assert_entities(hass, normal_config_entry.entry_id, entity_registry, snapshot)
+
+
+async def test_lock_offline(
+    hass: HomeAssistant,
+    mock_vehicle_data: AsyncMock,
+    normal_config_entry: MockConfigEntry,
+) -> None:
+    """Tests that the lock entities are correct when offline."""
+
+    mock_vehicle_data.side_effect = VehicleOffline
+    await setup_platform(hass, normal_config_entry, [Platform.LOCK])
+    state = hass.states.get("lock.test_lock")
+    assert state.state == STATE_UNKNOWN
+
+
+async def test_lock_services(
+    hass: HomeAssistant,
+    normal_config_entry: MockConfigEntry,
+) -> None:
+    """Tests that the lock services work."""
+
+    await setup_platform(hass, normal_config_entry, [Platform.LOCK])
+
+    entity_id = "lock.test_lock"
+
+    with patch(
+        "homeassistant.components.tesla_fleet.VehicleSpecific.door_lock",
+        return_value=COMMAND_OK,
+    ) as call:
+        await hass.services.async_call(
+            LOCK_DOMAIN,
+            SERVICE_LOCK,
+            {ATTR_ENTITY_ID: entity_id},
+            blocking=True,
+        )
+        state = hass.states.get(entity_id)
+        assert state.state == STATE_LOCKED
+        call.assert_called_once()
+
+    with patch(
+        "homeassistant.components.tesla_fleet.VehicleSpecific.door_unlock",
+        return_value=COMMAND_OK,
+    ) as call:
+        await hass.services.async_call(
+            LOCK_DOMAIN,
+            SERVICE_UNLOCK,
+            {ATTR_ENTITY_ID: entity_id},
+            blocking=True,
+        )
+        state = hass.states.get(entity_id)
+        assert state.state == STATE_UNLOCKED
+        call.assert_called_once()
+
+    entity_id = "lock.test_charge_cable_lock"
+
+    with pytest.raises(ServiceValidationError):
+        await hass.services.async_call(
+            LOCK_DOMAIN,
+            SERVICE_LOCK,
+            {ATTR_ENTITY_ID: entity_id},
+            blocking=True,
+        )
+
+    with patch(
+        "homeassistant.components.tesla_fleet.VehicleSpecific.charge_port_door_open",
+        return_value=COMMAND_OK,
+    ) as call:
+        await hass.services.async_call(
+            LOCK_DOMAIN,
+            SERVICE_UNLOCK,
+            {ATTR_ENTITY_ID: entity_id},
+            blocking=True,
+        )
+        state = hass.states.get(entity_id)
+        assert state.state == STATE_UNLOCKED
+        call.assert_called_once()