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 check_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

@@ -13,7 +13,7 @@
 import asyncio
 import json
 import logging
-from typing import Any
+from typing import Any, List
 
 import arrow
 
@@ -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
@@ -127,7 +131,7 @@ def uuid(self) -> str | None:
         """
         return self.api.uuid
 
-    async def set_alias(self, vehicle_id: int, new_alias: str) -> dict[str, Any]:
+    async def set_alias(self, vehicle_id: int, new_alias: str) -> dict[str, Any] | None:
         """Set a new alias for your vehicle.
 
         Sets a new alias for a vehicle specified by its vehicle id.
@@ -153,7 +157,7 @@ async def set_alias(self, vehicle_id: int, new_alias: str) -> dict[str, Any]:
             vehicle_id=vehicle_id, new_alias=new_alias
         )
 
-    async def get_vehicles(self) -> list[dict[str, Any]]:
+    async def get_vehicles(self) -> List[dict[str, Any]] | None:
         """Returns a list of vehicles.
 
         Retrieves list of vehicles associated with the account. The list contains static
@@ -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,58 @@ 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 request_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.
+            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 request_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.
+            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, "unlock")
+        _LOGGER.debug(f"Locking {censor_vin(vin)}... {raw_response}")
+        response = VehicleLockUnlockActionResponse(raw_response)
+        return response
+
+    async def get_lock_request_status(
+        self, vin: str, req_id: str
+    ) -> VehicleLockUnlockStatusResponse:
+        """Get the status of a lock request.
+
+        Args:
+            vin (str): Vehicle identification number.
+
+        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.check_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,7 @@
     "Sec-Fetch-Dest": "empty",
     "X-TME-BRAND": "TOYOTA",
 }
+
+# Timestamps
+TRIP_TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%SZ"
+UNLOCK_TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%S.%fZ"

mytoyota/controller.py

@@ -193,9 +193,18 @@ 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,
+                HTTPStatus.FOUND,
+            ]:
                 result = response.json()
             elif response.status_code == HTTPStatus.NO_CONTENT:
                 # This prevents raising or logging an error

mytoyota/models/lock_unlock.py

@@ -0,0 +1,50 @@
+""" 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)

mytoyota/models/trip.py

@@ -2,7 +2,9 @@
 from __future__ import annotations
 
 from datetime import datetime
+from typing import List
 
+from mytoyota.const import TRIP_TIMESTAMP_FORMAT
 from mytoyota.models.data import VehicleData
 
 
@@ -44,14 +46,14 @@ class DetailedTrip(VehicleData):
     """Detailed Trip model."""
 
     @property
-    def trip_events(self) -> list(TripEvent):
+    def trip_events(self) -> List[TripEvent]:
         """Trip events."""
         if not self._data.get("tripEvents"):
             return []
         return [TripEvent(event) for event in self._data.get("tripEvents", [])]
 
     @property
-    def trip_events_type(self) -> list(dict):
+    def trip_events_type(self) -> List[dict]:
         """Trip events type."""
         return self._data.get("tripEventsType", [])
 
@@ -75,20 +77,20 @@ def start_address(self) -> str:
         return self._data.get("startAddress", "")
 
     @property
-    def start_time_gmt(self) -> datetime.datetime | None:
+    def start_time_gmt(self) -> datetime | None:
         """Trip Start time GMT."""
         start_time_str = self._data.get("startTimeGmt", None)
         if not start_time_str:
             return None
-        return datetime.strptime(start_time_str, "%Y-%m-%dT%H:%M:%SZ")
+        return datetime.strptime(start_time_str, TRIP_TIMESTAMP_FORMAT)
 
     @property
-    def end_time_gmt(self) -> datetime.datetime | None:
+    def end_time_gmt(self) -> datetime | None:
         """Trip End time GMT."""
         end_time_str = self._data.get("endTimeGmt", None)
         if not end_time_str:
             return None
-        return datetime.strptime(end_time_str, "%Y-%m-%dT%H:%M:%SZ")
+        return datetime.strptime(end_time_str, TRIP_TIMESTAMP_FORMAT)
 
     @property
     def end_address(self) -> str:

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,63 @@
+"""pytest tests for mytoyota.client.MyT sending lock/unlock requests"""
+import asyncio
+
+from mytoyota.models.lock_unlock import VehicleLockUnlockActionResponse
+from tests.test_myt import TestMyTHelper
+
+
+class TestLockUnlock(TestMyTHelper):
+    """Pytest functions to test locking and unlocking"""
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.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.request_lock_vehicle(vehicle["vin"])
+        )
+        assert isinstance(result, VehicleLockUnlockActionResponse)
+        assert result == {
+            "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.request_unlock_vehicle(vehicle["vin"])
+        )
+        assert isinstance(result, VehicleLockUnlockActionResponse)
+        assert result == {
+            "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_request_status(vehicle["vin"], self.lock_request_id)
+        )
+        assert isinstance(result, VehicleLockUnlockActionResponse)
+        assert result == {
+            "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 == "2022-10-22T08:49:20.071Z"

tests/test_myt.py

@@ -54,7 +54,7 @@ def _load_from_file(self, filename: str):
         with open(filename, encoding="UTF-8") as json_file:
             return json.load(json_file)
 
-    async def request(
+    async def request(  # pylint: disable=R0915; # noqa
         self,
         method: str,
         endpoint: str,
@@ -144,6 +144,22 @@ 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)
+            response = self._load_from_file(
+                os.path.join(data_files, f"vehicle_{vin}_lock_request.json")
+            )
+
+        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