Add ability to run raw HTTP requests from RenaultVehicle

src/renault_api/renault_vehicle.py

@@ -76,24 +76,39 @@ def vin(self) -> str:
         """Get vin."""
         return self._vin
 
-    async def _get_vehicle_response(self, endpoint: str) -> models.KamereonResponse:
-        """GET to /v{endpoint_version}/cars/{vin}/{endpoint}."""
+    def _convert_variables(self, endpoint: str) -> str:
+        """Replace account_id / vin"""
+        return endpoint.replace("{account_id}", self.account_id).replace(
+            "{vin}", self.vin
+        )
+
+    async def http_get(self, endpoint: str) -> models.KamereonResponse:
+        """Run HTTP GET to endpoint."""
+        endpoint = self._convert_variables(endpoint)
+        return await self.session.http_request("GET", endpoint)
+
+    async def http_post(
+        self, endpoint: str, json: Optional[dict[str, Any]] = None
+    ) -> models.KamereonResponse:
+        """Run HTTP POST to endpoint."""
+        endpoint = self._convert_variables(endpoint)
+        return await self.session.http_request("POST", endpoint, json)
+
+    async def get_full_endpoint(self, endpoint: str) -> str:
+        """From VEHICLE_ENDPOINTS / DEFAULT_ENDPOINT."""
         details = await self.get_details()
         full_endpoint = details.get_endpoint(endpoint)
         if full_endpoint is None:
             raise EndpointNotAvailableError(endpoint, details.get_model_code())
 
-        full_endpoint = ACCOUNT_ENDPOINT_ROOT.replace(
-            "{account_id}", self.account_id
-        ) + full_endpoint.replace("{vin}", self.vin)
-
-        return await self.session.http_request("GET", full_endpoint)
+        return ACCOUNT_ENDPOINT_ROOT + full_endpoint
 
     async def _get_vehicle_data(
         self, endpoint: str
     ) -> models.KamereonVehicleDataResponse:
         """GET to /v{endpoint_version}/cars/{vin}/{endpoint}."""
-        response = await self._get_vehicle_response(endpoint)
+        full_endpoint = await self.get_full_endpoint(endpoint)
+        response = await self.http_get(full_endpoint)
         return cast(
             models.KamereonVehicleDataResponse,
             schemas.KamereonVehicleDataResponseSchema.load(response.raw_data),
@@ -226,7 +241,8 @@ async def get_charging_settings(self) -> models.KamereonVehicleChargingSettingsD
 
     async def get_charge_schedule(self) -> dict[str, Any]:
         """Get vehicle charging schedule."""
-        response = await self._get_vehicle_response("charge-schedule")
+        full_endpoint = await self.get_full_endpoint("charge-schedule")
+        response = await self.http_get(full_endpoint)
         if "data" in response.raw_data and "attributes" in response.raw_data["data"]:
             return response.raw_data["data"]["attributes"]  # type:ignore[no-any-return]
         return response.raw_data

tests/__snapshots__/test_renault_vehicle.ambr

@@ -227,6 +227,22 @@
     'soc-levels': '/kcm/v1/vehicles/{vin}/ev/soc-levels',
   })
 # ---
+# name: test_http_get
+  KamereonResponse(raw_data={'data': {'type': 'Car', 'id': 'VF1AAAAA555777999', 'attributes': {'calendar': {'monday': [{'startTime': '2300', 'duration': 480, 'activationState': False}], 'tuesday': [{'startTime': '2300', 'duration': 480, 'activationState': False}], 'wednesday': [{'startTime': '2300', 'duration': 480, 'activationState': False}], 'thursday': [{'startTime': '2300', 'duration': 480, 'activationState': False}], 'friday': [{'startTime': '2300', 'duration': 480, 'activationState': False}], 'saturday': [{'startTime': '2300', 'duration': 480, 'activationState': False}], 'sunday': [{'startTime': '2300', 'duration': 480, 'activationState': False}]}}}}, errors=None)
+# ---
+# name: test_http_post
+  KamereonResponse(raw_data={'data': {'type': 'ChargingStart', 'id': 'guid', 'attributes': {'action': 'start'}}}, errors=None)
+# ---
+# name: test_http_post.1
+  dict({
+    'data': dict({
+      'attributes': dict({
+        'action': 'start',
+      }),
+      'type': 'ChargingStart',
+    }),
+  })
+# ---
 # name: test_set_ac_start
   dict({
     'data': dict({

tests/test_renault_vehicle.py

@@ -19,6 +19,7 @@
 from tests.test_credential_store import get_logged_in_credential_store
 from tests.test_renault_session import get_logged_in_session
 
+from renault_api.exceptions import EndpointNotAvailableError
 from renault_api.kamereon.helpers import DAYS_OF_WEEK
 from renault_api.kamereon.models import ChargeSchedule
 from renault_api.kamereon.models import HvacSchedule
@@ -370,6 +371,46 @@ async def test_set_hvac_schedules(
     assert request.kwargs["json"] == snapshot
 
 
+@pytest.mark.asyncio
+async def test_http_get(
+    vehicle: RenaultVehicle, mocked_responses: aioresponses, snapshot: SnapshotAssertion
+) -> None:
+    """Test http_get."""
+    fixtures.inject_get_vehicle_details(mocked_responses, "zoe_40.1.json")
+    endpoint = await vehicle.get_full_endpoint("charge-schedule")
+    url = fixtures.inject_get_charge_schedule(mocked_responses, "single")
+
+    assert await vehicle.http_get(endpoint) == snapshot
+    request: RequestCall = mocked_responses.requests[("GET", URL(url))][0]
+
+    assert request.kwargs["json"] is None
+
+
+@pytest.mark.asyncio
+async def test_http_post(
+    vehicle: RenaultVehicle, mocked_responses: aioresponses, snapshot: SnapshotAssertion
+) -> None:
+    """Test http_post."""
+    endpoint = (
+        "/commerce/v1/accounts/{account_id}"
+        "/kamereon/kca/car-adapter/v1/cars/{vin}/actions/charging-start"
+    )
+    json = {
+        "data": {
+            "attributes": {
+                "action": "start",
+            },
+            "type": "ChargingStart",
+        },
+    }
+    url = fixtures.inject_set_charging_start(mocked_responses, "start")
+
+    assert await vehicle.http_post(endpoint, json) == snapshot
+    request: RequestCall = mocked_responses.requests[("POST", URL(url))][0]
+
+    assert request.kwargs["json"] == snapshot
+
+
 @pytest.mark.parametrize(
     "filename", fixtures.get_json_files(f"{fixtures.KAMEREON_FIXTURE_PATH}/vehicles")
 )
@@ -387,3 +428,14 @@ async def test_get_endpoints(
     endpoints = details.get_endpoints()
 
     assert endpoints == snapshot
+
+
+@pytest.mark.asyncio
+async def test_get_full_endpoint_unknown(
+    vehicle: RenaultVehicle, mocked_responses: aioresponses
+) -> None:
+    """Test http_get."""
+    # Unkown endpoint
+    fixtures.inject_get_vehicle_details(mocked_responses, "zoe_40.1.json")
+    with pytest.raises(EndpointNotAvailableError):
+        await vehicle.get_full_endpoint("random")