Add lock and unlock functions

mytoyota/api.py

@@ -117,3 +117,24 @@ async def get_trip_endpoint(self, vin: str, trip_id: str) -> dict[str, Any] | No
             endpoint=f"/api/user/{self.uuid}/cms/trips/v2/{trip_id}/events/vin/{vin}",
             headers={"vin": vin},
         )
+
+    async def set_lock_unlock_vehicle_endpoint(
+        self, vin: str, action: str
+    ) -> dict[str, str] | None:
+        """Lock vehicle."""
+        return await self.controller.request(
+            method="POST",
+            base_url=BASE_URL,
+            endpoint=f"/vehicles/{vin}/lock",
+            body={"action": action},
+        )
+
+    async def get_lock_unlock_request_status(
+        self, vin: str, request_id: str
+    ) -> dict[str, Any] | None:
+        """Check lock/unlock status given a request ID"""
+        return await self.controller.request(
+            method="GET",
+            base_url=BASE_URL,
+            endpoint=f"/vehicles/{vin}/lock/{request_id}",
+        )

mytoyota/client.py

@@ -37,6 +37,10 @@
     ToyotaLocaleNotValid,
     ToyotaRegionNotSupported,
 )
+from .models.lock_unlock import (
+    VehicleLockUnlockActionResponse,
+    VehicleLockUnlockStatusResponse,
+)
 from .models.trip import DetailedTrip, Trip
 from .models.vehicle import Vehicle
 from .statistics import Statistics
@@ -457,7 +461,7 @@ async def get_driving_statistics_json(
             await self.get_driving_statistics(vin, interval, from_date), indent=3
         )
 
-    async def get_trips(self, vin: str) -> list(Trip):
+    async def get_trips(self, vin: str) -> list[Trip]:
         """Returns a list of trips.
 
         Retrieves and formats trips.
@@ -540,3 +544,62 @@ async def get_trip_json(self, vin: str, trip_id: str) -> str:
         """
         trip = await self.get_trip(vin, trip_id)
         return json.dumps(trip.raw_json, indent=3)
+
+    async def set_lock_vehicle(self, vin: str) -> VehicleLockUnlockActionResponse:
+        """Sends a lock command to the vehicle.
+
+        Args:
+            vin (str): Vehicle identification number.
+
+        Raises:
+            ToyotaLoginError: An error returned when updating token or invalid login information.
+            ToyotaActionNotSupported: The lock action is not supported on this vehicle.
+            ToyotaInternalError: An error occurred when making a request.
+            ToyotaApiError: Toyota's API returned an error.
+        """
+        _LOGGER.debug(f"Locking {censor_vin(vin)}...")
+        raw_response = await self.api.set_lock_unlock_vehicle_endpoint(vin, "lock")
+        _LOGGER.debug(f"Locking {censor_vin(vin)}... {raw_response}")
+        response = VehicleLockUnlockActionResponse(raw_response)
+        return response
+
+    async def set_unlock_vehicle(self, vin: str) -> VehicleLockUnlockActionResponse:
+        """Send an unlock command to the vehicle.
+
+        Args:
+            vin (str): Vehicle identification number.
+
+        Raises:
+            ToyotaLoginError: An error returned when updating token or invalid login information.
+            ToyotaActionNotSupported: The lock action is not supported on this vehicle.
+            ToyotaInternalError: An error occurred when making a request.
+            ToyotaApiError: Toyota's API returned an error.
+        """
+        _LOGGER.debug(f"Unlocking {censor_vin(vin)}...")
+        raw_response = await self.api.set_lock_unlock_vehicle_endpoint(vin, "unlock")
+        _LOGGER.debug(f"Unlocking {censor_vin(vin)}... {raw_response}")
+        response = VehicleLockUnlockActionResponse(raw_response)
+        return response
+
+    async def get_lock_status(
+        self, vin: str, req_id: str
+    ) -> VehicleLockUnlockStatusResponse:
+        """Get the status of a lock request.
+
+        Args:
+            vin (str): Vehicle identification number.
+            req_id (str): Lock/Unlock request id returned by
+                set_<lock/unlock>_vehicle (UUID)
+
+        Raises:
+            ToyotaLoginError: An error returned when updating token or invalid login information.
+            ToyotaInternalError: An error occurred when making a request.
+            ToyotaApiError: Toyota's API returned an error.
+        """
+        _LOGGER.debug(f"Getting lock request status for {censor_vin(vin)}...")
+        raw_response = await self.api.get_lock_unlock_request_status(vin, req_id)
+        _LOGGER.debug(
+            f"Getting lock request status for {censor_vin(vin)}... {raw_response}"
+        )
+        response = VehicleLockUnlockStatusResponse(raw_response)
+        return response

mytoyota/const.py

@@ -63,3 +63,6 @@
     "Sec-Fetch-Dest": "empty",
     "X-TME-BRAND": "TOYOTA",
 }
+
+# Timestamps
+UNLOCK_TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%S.%fZ"

mytoyota/controller.py

@@ -19,7 +19,12 @@
     TOKEN_VALID_URL,
     UUID,
 )
-from mytoyota.exceptions import ToyotaApiError, ToyotaInternalError, ToyotaLoginError
+from mytoyota.exceptions import (
+    ToyotaActionNotSupported,
+    ToyotaApiError,
+    ToyotaInternalError,
+    ToyotaLoginError,
+)
 from mytoyota.utils.logs import censor_dict
 from mytoyota.utils.token import is_valid_token
 
@@ -193,9 +198,17 @@ async def request(  # pylint: disable=too-many-branches
                 f"Body: {censor_dict(body) if body else body} - Parameters: {params}"
             )
             response = await client.request(
-                method, url, headers=headers, json=body, params=params
+                method,
+                url,
+                headers=headers,
+                json=body,
+                params=params,
+                follow_redirects=True,
             )
-            if response.status_code == HTTPStatus.OK:
+            if response.status_code in [
+                HTTPStatus.OK,
+                HTTPStatus.ACCEPTED,
+            ]:
                 result = response.json()
             elif response.status_code == HTTPStatus.NO_CONTENT:
                 # This prevents raising or logging an error
@@ -221,6 +234,10 @@ async def request(  # pylint: disable=too-many-branches
                 raise ToyotaApiError("Servers are overloaded, try again later")
             elif response.status_code == HTTPStatus.SERVICE_UNAVAILABLE:
                 raise ToyotaApiError("Servers are temporarily unavailable")
+            elif response.status_code == HTTPStatus.FORBIDDEN:
+                raise ToyotaActionNotSupported(
+                    "Action is not supported on this vehicle"
+                )
             else:
                 raise ToyotaApiError(
                     "HTTP: " + str(response.status_code) + " - " + response.text

mytoyota/exceptions.py

@@ -27,3 +27,7 @@ class ToyotaApiError(Exception):
 
 class ToyotaInternalError(Exception):
     """Raise if an internal server error occurres from Toyota."""
+
+
+class ToyotaActionNotSupported(ToyotaApiError):
+    """Raise if an action is not supported on a vehicle."""

mytoyota/models/lock_unlock.py

@@ -0,0 +1,72 @@
+""" models for vehicle lock/unlock requests and responses """
+from datetime import datetime
+
+from mytoyota.const import UNLOCK_TIMESTAMP_FORMAT
+from mytoyota.models.data import VehicleData
+
+
+class VehicleLockUnlockActionResponse(VehicleData):
+    """Model of the response to a Vehicle Lock/Unlock Action Request."""
+
+    @property
+    def status(self) -> str:
+        """Request Status."""
+        return self._data.get("status", "")
+
+    @property
+    def request_id(self) -> str:
+        """Request ID."""
+        return self._data.get("id", "")
+
+    @property
+    def type(self) -> str:
+        """Request Type."""
+        return self._data.get("type", "")
+
+
+class VehicleLockUnlockStatusResponse(VehicleData):
+    """Model of the response to a the request of the status of
+    a Vehicle Lock/Unlock action."""
+
+    @property
+    def status(self) -> str:
+        """Request Status."""
+        return self._data.get("status", "")
+
+    @property
+    def request_id(self) -> str:
+        """Request ID."""
+        return self._data.get("id", "")
+
+    @property
+    def type(self) -> str:
+        """Request Type."""
+        return self._data.get("type", "")
+
+    @property
+    def request_timestamp(self) -> datetime:
+        """Request Timestamp."""
+        raw_datetime = self._data.get("requestTimestamp")
+        return datetime.strptime(raw_datetime, UNLOCK_TIMESTAMP_FORMAT)
+
+    @property
+    def error_code(self) -> str:
+        """Request Error code"""
+        if self.status != "error":
+            return None
+        return self._data.get("errorCode", "")
+
+    @property
+    def is_success(self) -> bool:
+        """Request was successful."""
+        return self.status == "completed"
+
+    @property
+    def is_error(self) -> bool:
+        """Request failed."""
+        return self.status == "error"
+
+    @property
+    def is_in_progress(self) -> bool:
+        """Request is processing."""
+        return self.status == "inprogress"

tests/data/vehicle_JTMW1234565432109_lock_request.json

@@ -0,0 +1,5 @@
+{
+  "id": "d4f873d2-5da2-494f-a6d9-6e56d18d2ce9",
+  "status": "inprogress",
+  "type": "controlLock"
+}

tests/data/vehicle_JTMW1234565432109_lock_request_status_d4f873d2-5da2-494f-a6d9-6e56d18d2ce9.json

@@ -0,0 +1,6 @@
+{
+  "id": "d4f873d2-5da2-494f-a6d9-6e56d18d2ce9",
+  "status": "completed",
+  "requestTimestamp": "2022-10-22T08:49:20.071Z",
+  "type": "controlLock"
+}

tests/test_lock_unlock.py

@@ -0,0 +1,86 @@
+"""pytest tests for mytoyota.client.MyT sending lock/unlock requests"""
+import asyncio
+from datetime import datetime
+
+import pytest
+
+from mytoyota.exceptions import ToyotaActionNotSupported
+from mytoyota.models.lock_unlock import (
+    VehicleLockUnlockActionResponse,
+    VehicleLockUnlockStatusResponse,
+)
+from tests.test_myt import TestMyTHelper
+
+
+class TestLockUnlock(TestMyTHelper):
+    """Pytest functions to test locking and unlocking"""
+
+    _lock_request_id = "d4f873d2-5da2-494f-a6d9-6e56d18d2ce9"
+
+    def test_send_lock_request(self):
+        """Test sending the lock request"""
+        myt = self._create_offline_myt()
+        vehicle = self._lookup_vehicle(myt, 4444444)
+        result = asyncio.get_event_loop().run_until_complete(
+            myt.set_lock_vehicle(vehicle["vin"])
+        )
+        assert isinstance(result, VehicleLockUnlockActionResponse)
+        assert result.raw_json == {
+            "id": self._lock_request_id,
+            "status": "inprogress",
+            "type": "controlLock",
+        }
+        assert result.request_id == self._lock_request_id
+        assert result.status == "inprogress"
+        assert result.type == "controlLock"
+
+    def test_send_unlock_request(self):
+        """Test sending the unlock request"""
+        myt = self._create_offline_myt()
+        vehicle = self._lookup_vehicle(myt, 4444444)
+        result = asyncio.get_event_loop().run_until_complete(
+            myt.set_unlock_vehicle(vehicle["vin"])
+        )
+        assert isinstance(result, VehicleLockUnlockActionResponse)
+        assert result.raw_json == {
+            "id": self._lock_request_id,
+            "status": "inprogress",
+            "type": "controlLock",
+        }
+
+    def test_get_lock_status(self):
+        """Test getting the lock status"""
+        myt = self._create_offline_myt()
+        vehicle = self._lookup_vehicle(myt, 4444444)
+        result = asyncio.get_event_loop().run_until_complete(
+            myt.get_lock_status(vehicle["vin"], self._lock_request_id)
+        )
+        assert isinstance(result, VehicleLockUnlockStatusResponse)
+        assert result.raw_json == {
+            "id": self._lock_request_id,
+            "status": "completed",
+            "requestTimestamp": "2022-10-22T08:49:20.071Z",
+            "type": "controlLock",
+        }
+        assert result.request_id == self._lock_request_id
+        assert result.status == "completed"
+        assert result.type == "controlLock"
+        assert result.request_timestamp == datetime(2022, 10, 22, 8, 49, 20, 71000)
+
+    def test_set_lock_vehicle_unsupported(self):
+        """Test sending the lock request to a vehicle for which it is not supported"""
+        myt = self._create_offline_myt()
+        vehicle = self._lookup_vehicle(myt, 1111111)
+        with pytest.raises(ToyotaActionNotSupported):
+            asyncio.get_event_loop().run_until_complete(
+                myt.set_lock_vehicle(vehicle["vin"])
+            )
+
+    def test_set_unlock_vehicle_unsupported(self):
+        """Test sending the lock request to a vehicle for which it is not supported"""
+        myt = self._create_offline_myt()
+        vehicle = self._lookup_vehicle(myt, 1111111)
+        with pytest.raises(ToyotaActionNotSupported):
+            asyncio.get_event_loop().run_until_complete(
+                myt.set_unlock_vehicle(vehicle["vin"])
+            )

tests/test_myt.py

@@ -12,6 +12,7 @@
 
 from mytoyota.client import MyT
 from mytoyota.exceptions import (
+    ToyotaActionNotSupported,
     ToyotaInternalError,
     ToyotaInvalidUsername,
     ToyotaLocaleNotValid,
@@ -54,7 +55,8 @@ def _load_from_file(self, filename: str):
         with open(filename, encoding="UTF-8") as json_file:
             return json.load(json_file)
 
-    async def request(
+    # Disables pylint warning about too many statements and branches when matching API paths
+    async def request(  # pylint: disable=too-many-statements, too-many-branches, too-many-locals
         self,
         method: str,
         endpoint: str,
@@ -144,6 +146,25 @@ async def request(
                 os.path.join(data_files, f"vehicle_{vin}_trip_{trip_id}.json")
             )
 
+        match = re.match(r".*/vehicles/([^?]+)/lock", endpoint)
+        if match:
+            vin = match.group(1)
+            try:
+                response = self._load_from_file(
+                    os.path.join(data_files, f"vehicle_{vin}_lock_request.json")
+                )
+            except FileNotFoundError as exc:
+                raise ToyotaActionNotSupported("Action is not supported") from exc
+
+        match = re.match(r".*/vehicles/([^?]+)/lock/([^?]+)", endpoint)
+        if match:
+            vin = match.group(1)
+            request_id = match.group(2)
+            response = self._load_from_file(
+                os.path.join(
+                    data_files, f"vehicle_{vin}_lock_request_status_{request_id}.json"
+                )
+            )
         return response