Add send_location_to_vehicle

src/rivian/rivian.py

@@ -645,6 +645,27 @@ async def depart_now(
         }
         return await self.__graphql_query(headers, url, graphql_json)
 
+    async def send_location_to_vehicle(
+        self, location: str, vehicle_id: str
+    ) -> ClientResponse:
+        """Send a destination to a vehicle's navigation.
+
+        `location` is an address, a place, or coordinates as "lat,lng", and Rivian
+        parses it. A `publishResponse.result` of 0 means the request was accepted.
+        """
+        url = GRAPHQL_GATEWAY
+        headers = BASE_HEADERS | {
+            "Csrf-Token": self._csrf_token,
+            "A-Sess": self._app_session_token,
+            "U-Sess": self._user_session_token,
+        }
+        graphql_json = {
+            "operationName": "parseAndShareLocationToVehicle",
+            "query": "mutation parseAndShareLocationToVehicle($str: String!, $vehicleId: String!) { parseAndShareLocationToVehicle(str: $str, vehicleId: $vehicleId) { __typename publishResponse { __typename result } } }",
+            "variables": {"str": location, "vehicleId": vehicle_id},
+        }
+        return await self.__graphql_query(headers, url, graphql_json)
+
     async def get_vehicle_ota_update_details(self, vehicle_id: str) -> ClientResponse:
         """Get vehicle OTA update details."""
         url = GRAPHQL_GATEWAY

tests/responses.py

@@ -646,6 +646,9 @@
     "data": {"skipNextScheduledDeparture": {"success": True}}
 }
 DEPART_NOW_RESPONSE = {"data": {"departNow": {"success": True}}}
+SEND_LOCATION_TO_VEHICLE_RESPONSE = {
+    "data": {"parseAndShareLocationToVehicle": {"publishResponse": {"result": 0}}}
+}
 
 SEND_VEHICLE_OPERATION_RESPONSE = {
     "data": {

tests/test_rivian.py

@@ -38,6 +38,7 @@
     DELETE_DEPARTURE_SCHEDULE_RESPONSE,
     DEPART_NOW_RESPONSE,
     OTP_TOKEN_RESPONSE,
+    SEND_LOCATION_TO_VEHICLE_RESPONSE,
     SEND_VEHICLE_OPERATION_RESPONSE,
     SET_CHARGING_SCHEDULES_RESPONSE,
     SKIP_NEXT_SCHEDULED_DEPARTURE_RESPONSE,
@@ -418,6 +419,29 @@ async def test_skip_and_depart_now(aresponses: ResponsesMockServer) -> None:
         await rivian.close()
 
 
+async def test_send_location_to_vehicle(aresponses: ResponsesMockServer) -> None:
+    """Test sending a destination to a vehicle."""
+    requests = capture_requests(aresponses, SEND_LOCATION_TO_VEHICLE_RESPONSE)
+    async with aiohttp.ClientSession():
+        rivian = Rivian(
+            csrf_token="csrf",
+            app_session_token="token",
+            user_session_token="token",
+        )
+
+        response = await rivian.send_location_to_vehicle(
+            "123 Main St, Springfield, IL", "vehicle_id"
+        )
+        response_json = await response.json()
+        result = response_json["data"]["parseAndShareLocationToVehicle"]
+        assert result["publishResponse"]["result"] == 0
+        assert requests[0]["variables"] == {
+            "str": "123 Main St, Springfield, IL",
+            "vehicleId": "vehicle_id",
+        }
+        await rivian.close()
+
+
 async def test_subscribe_for_departure_schedules() -> None:
     """Test subscribing to departure schedules."""
     rivian = Rivian(user_session_token="token")