From 716bb11cd7e8bfc8e514604a179df9ba105c6ccf Mon Sep 17 00:00:00 2001 From: vuffiraa72 Date: Sun, 29 Mar 2026 11:05:06 +0200 Subject: [PATCH 1/7] Add odometer data retrieval to Skoda API integration --- packages/modules/vehicles/skoda/libskoda.py | 11 +++++ packages/modules/vehicles/skoda/soc.py | 6 +-- packages/modules/vehicles/skoda/soc_test.py | 52 ++++++++++++++++++--- 3 files changed, 60 insertions(+), 9 deletions(-) diff --git a/packages/modules/vehicles/skoda/libskoda.py b/packages/modules/vehicles/skoda/libskoda.py index 49bd5e5464..3c65aa79a6 100755 --- a/packages/modules/vehicles/skoda/libskoda.py +++ b/packages/modules/vehicles/skoda/libskoda.py @@ -238,6 +238,7 @@ async def refresh_tokens(self): async def get_status(self): vehicle_status_url = f"{API_BASE}/v2/vehicle-status/{self.vin}/driving-range" charging_url = f"{API_BASE}/v1/charging/{self.vin}" + maintenance_report_url = f"{API_BASE}/v3/vehicle-maintenance/vehicles/{self.vin}/report" response = await self.session.get(vehicle_status_url, headers=self.headers) # If first attempt fails, try to refresh tokens @@ -293,6 +294,15 @@ async def get_status(self): if not timestamp.endswith('Z'): timestamp += 'Z' + response = await self.session.get(maintenance_report_url, headers=self.headers) + if response.status >= 400: + self.log.error("Get status from maintenance_report_url failed") + odometer = None + else: + maintenance_data = await response.json() + self.log.debug(f"Status data from Skoda API (maintenance report): {maintenance_data}") + odometer = maintenance_data.get('mileageInKm', None) + return { 'charging': { 'batteryStatus': { @@ -300,6 +310,7 @@ async def get_status(self): 'currentSOC_pct': soc, 'cruisingRangeElectric_km': range_km, 'carCapturedTimestamp': timestamp, + 'odometer': odometer, } } } diff --git a/packages/modules/vehicles/skoda/soc.py b/packages/modules/vehicles/skoda/soc.py index c2de41bb2d..2032f89c24 100755 --- a/packages/modules/vehicles/skoda/soc.py +++ b/packages/modules/vehicles/skoda/soc.py @@ -16,9 +16,9 @@ def fetch(vehicle_update_data: VehicleUpdateData, config: Skoda, vehicle: int) -> CarState: - soc, range, soc_ts, soc_tsX = api.fetch_soc(config, vehicle) - log.info("Result: soc=" + str(soc)+", range=" + str(range) + "@" + soc_ts) - return CarState(soc=soc, range=range, soc_timestamp=soc_tsX) + soc, range, soc_ts, soc_tsX, odometer = api.fetch_soc(config, vehicle) + log.info("Result: soc=" + str(soc)+", range=" + str(range) + "@" + soc_ts + ", odometer=" + str(odometer)) + return CarState(soc=soc, range=range, soc_timestamp=soc_tsX, odometer=odometer) def create_vehicle(vehicle_config: Skoda, vehicle: int): diff --git a/packages/modules/vehicles/skoda/soc_test.py b/packages/modules/vehicles/skoda/soc_test.py index bd57070bf7..4a89368da2 100644 --- a/packages/modules/vehicles/skoda/soc_test.py +++ b/packages/modules/vehicles/skoda/soc_test.py @@ -14,7 +14,10 @@ class TestSkoda: @pytest.fixture(autouse=True) def set_up(self, monkeypatch): self.mock_context_exit = Mock(return_value=True) - self.mock_fetch_soc = Mock(name="fetch_soc", return_value=(50, 250, "2025-10-18T10:00:00Z", 1760822400.0)) + self.mock_fetch_soc = Mock( + name="fetch_soc", + return_value=(50, 250, "2025-10-18T10:00:00Z", 1760822400.0, 12345), + ) self.mock_value_store = Mock(name="value_store") monkeypatch.setattr(api, "fetch_soc", self.mock_fetch_soc) monkeypatch.setattr(store, "get_car_value_store", Mock(return_value=self.mock_value_store)) @@ -35,6 +38,7 @@ def test_update_updates_value_store(self): assert call_args.soc == 50 assert call_args.range == 250 assert call_args.soc_timestamp == 1760822400.0 + assert call_args.odometer == 12345 def test_update_passes_errors_to_context(self): # setup @@ -105,7 +109,17 @@ def test_get_status_success_primary_url(self, skoda_instance, mock_session): }, "carCapturedTimestamp": "2025-10-17T15:40:35.679Z" } - mock_session.get.return_value = MockAiohttpResponse(response_data, 200) + maintenance_response_data = { + "mileageInKm": 54321 + } + responses = [ + MockAiohttpResponse(response_data, 200), + MockAiohttpResponse(maintenance_response_data, 200) + ] + + async def side_effect_func(*args, **kwargs): + return responses.pop(0) + mock_session.get.side_effect = side_effect_func # execution status = asyncio.run(skoda_instance.get_status()) @@ -114,10 +128,16 @@ def test_get_status_success_primary_url(self, skoda_instance, mock_session): assert status['charging']['batteryStatus']['value']['currentSOC_pct'] == 66 assert status['charging']['batteryStatus']['value']['cruisingRangeElectric_km'] == 291 assert status['charging']['batteryStatus']['value']['carCapturedTimestamp'] == "2025-10-17T15:40:35Z" - mock_session.get.assert_called_once_with( + assert status['charging']['batteryStatus']['value']['odometer'] == 54321 + assert mock_session.get.call_count == 2 + mock_session.get.assert_any_call( "https://mysmob.api.connect.skoda-auto.cz/api/v2/vehicle-status/test_vin/driving-range", headers=skoda_instance.headers ) + mock_session.get.assert_any_call( + "https://mysmob.api.connect.skoda-auto.cz/api/v3/vehicle-maintenance/vehicles/test_vin/report", + headers=skoda_instance.headers + ) def test_get_status_fallback_to_charging_url(self, skoda_instance, mock_session): # setup @@ -158,9 +178,13 @@ def test_get_status_fallback_to_charging_url(self, skoda_instance, mock_session) } ] } + maintenance_response_data = { + "mileageInKm": 123456 + } responses = [ MockAiohttpResponse(vehicle_status_response_data, 200), - MockAiohttpResponse(charging_url_response_data, 200) + MockAiohttpResponse(charging_url_response_data, 200), + MockAiohttpResponse(maintenance_response_data, 200) ] async def side_effect_func(*args, **kwargs): @@ -174,7 +198,8 @@ async def side_effect_func(*args, **kwargs): assert status['charging']['batteryStatus']['value']['currentSOC_pct'] == 66 assert status['charging']['batteryStatus']['value']['cruisingRangeElectric_km'] == 291.0 assert status['charging']['batteryStatus']['value']['carCapturedTimestamp'] == "2025-10-17T15:38:55Z" - assert mock_session.get.call_count == 2 + assert status['charging']['batteryStatus']['value']['odometer'] == 123456 + assert mock_session.get.call_count == 3 mock_session.get.assert_any_call( "https://mysmob.api.connect.skoda-auto.cz/api/v2/vehicle-status/test_vin/driving-range", headers=skoda_instance.headers @@ -183,6 +208,10 @@ async def side_effect_func(*args, **kwargs): "https://mysmob.api.connect.skoda-auto.cz/api/v1/charging/test_vin", headers=skoda_instance.headers ) + mock_session.get.assert_any_call( + "https://mysmob.api.connect.skoda-auto.cz/api/v3/vehicle-maintenance/vehicles/test_vin/report", + headers=skoda_instance.headers + ) def test_get_status_timestamp_without_milliseconds(self, skoda_instance, mock_session): # setup @@ -203,7 +232,17 @@ def test_get_status_timestamp_without_milliseconds(self, skoda_instance, mock_se }, "carCapturedTimestamp": "2025-10-06T10:12:44Z" } - mock_session.get.return_value = MockAiohttpResponse(response_data, 200) + maintenance_response_data = { + "mileageInKm": 98765 + } + responses = [ + MockAiohttpResponse(response_data, 200), + MockAiohttpResponse(maintenance_response_data, 200) + ] + + async def side_effect_func(*args, **kwargs): + return responses.pop(0) + mock_session.get.side_effect = side_effect_func # execution status = asyncio.run(skoda_instance.get_status()) @@ -212,3 +251,4 @@ def test_get_status_timestamp_without_milliseconds(self, skoda_instance, mock_se assert status['charging']['batteryStatus']['value']['currentSOC_pct'] == 42 assert status['charging']['batteryStatus']['value']['cruisingRangeElectric_km'] == 47 assert status['charging']['batteryStatus']['value']['carCapturedTimestamp'] == "2025-10-06T10:12:44Z" + assert status['charging']['batteryStatus']['value']['odometer'] == 98765 From d02d21b9d2c70e2d26fdc1cc7a5cbfa6357506b2 Mon Sep 17 00:00:00 2001 From: vuffiraa72 Date: Sun, 29 Mar 2026 11:26:17 +0200 Subject: [PATCH 2/7] Improve error handling in Skoda API vehicle status retrieval and add test for fallback to charging URL on 403 response --- packages/modules/vehicles/skoda/libskoda.py | 46 ++++++++++++-------- packages/modules/vehicles/skoda/soc_test.py | 48 +++++++++++++++++++++ 2 files changed, 75 insertions(+), 19 deletions(-) diff --git a/packages/modules/vehicles/skoda/libskoda.py b/packages/modules/vehicles/skoda/libskoda.py index 3c65aa79a6..0144b6ed4e 100755 --- a/packages/modules/vehicles/skoda/libskoda.py +++ b/packages/modules/vehicles/skoda/libskoda.py @@ -241,27 +241,35 @@ async def get_status(self): maintenance_report_url = f"{API_BASE}/v3/vehicle-maintenance/vehicles/{self.vin}/report" response = await self.session.get(vehicle_status_url, headers=self.headers) - # If first attempt fails, try to refresh tokens - if response.status >= 400: - self.log.debug("Refreshing tokens") - if await self.refresh_tokens(): - response = await self.session.get(vehicle_status_url, headers=self.headers) - - # If refreshing tokens failed, try a full reconnect - if response.status >= 400: - self.log.info("Reconnecting") - if await self.reconnect(): - response = await self.session.get(vehicle_status_url, headers=self.headers) - else: - self.log.error("Reconnect failed") - return {} + if response.status == 403: + self.log.info("vehicle-status returned 403, trying charging_url") + status_data = {} + else: + # If first attempt fails, try to refresh tokens + if response.status >= 400: + self.log.debug("Refreshing tokens") + if await self.refresh_tokens(): + response = await self.session.get(vehicle_status_url, headers=self.headers) - if response.status >= 400: - self.log.error("Get status failed") - return {} + # If refreshing tokens failed, try a full reconnect + if response.status >= 400: + self.log.info("Reconnecting") + if await self.reconnect(): + response = await self.session.get(vehicle_status_url, headers=self.headers) + else: + self.log.error("Reconnect failed") + return {} - status_data = await response.json() - self.log.debug(f"Status data from Skoda API (vehicle-status): {status_data}") + if response.status >= 400: + if response.status == 403: + self.log.info("vehicle-status returned 403, trying charging_url") + status_data = {} + else: + self.log.error("Get status failed") + return {} + else: + status_data = await response.json() + self.log.debug(f"Status data from Skoda API (vehicle-status): {status_data}") # check if all values are valid, otherwise use charging_url electric_engine_range = {} diff --git a/packages/modules/vehicles/skoda/soc_test.py b/packages/modules/vehicles/skoda/soc_test.py index 4a89368da2..b1a9d56577 100644 --- a/packages/modules/vehicles/skoda/soc_test.py +++ b/packages/modules/vehicles/skoda/soc_test.py @@ -252,3 +252,51 @@ async def side_effect_func(*args, **kwargs): assert status['charging']['batteryStatus']['value']['cruisingRangeElectric_km'] == 47 assert status['charging']['batteryStatus']['value']['carCapturedTimestamp'] == "2025-10-06T10:12:44Z" assert status['charging']['batteryStatus']['value']['odometer'] == 98765 + + def test_get_status_fallback_to_charging_url_on_vehicle_status_403(self, skoda_instance, mock_session): + # setup + vehicle_status_error = {"error": "forbidden"} + charging_url_response_data = { + "status": { + "battery": { + "remainingCruisingRangeInMeters": 302000, + "stateOfChargeInPercent": 71 + } + }, + "carCapturedTimestamp": "2026-03-29T09:15:00Z" + } + maintenance_response_data = { + "mileageInKm": 65432 + } + + responses = [ + MockAiohttpResponse(vehicle_status_error, 403), + MockAiohttpResponse(charging_url_response_data, 200), + MockAiohttpResponse(maintenance_response_data, 200) + ] + + async def side_effect_func(*args, **kwargs): + return responses.pop(0) + mock_session.get.side_effect = side_effect_func + + # execution + status = asyncio.run(skoda_instance.get_status()) + + # evaluation + assert status['charging']['batteryStatus']['value']['currentSOC_pct'] == 71 + assert status['charging']['batteryStatus']['value']['cruisingRangeElectric_km'] == 302.0 + assert status['charging']['batteryStatus']['value']['carCapturedTimestamp'] == "2026-03-29T09:15:00Z" + assert status['charging']['batteryStatus']['value']['odometer'] == 65432 + assert mock_session.get.call_count == 3 + mock_session.get.assert_any_call( + "https://mysmob.api.connect.skoda-auto.cz/api/v2/vehicle-status/test_vin/driving-range", + headers=skoda_instance.headers + ) + mock_session.get.assert_any_call( + "https://mysmob.api.connect.skoda-auto.cz/api/v1/charging/test_vin", + headers=skoda_instance.headers + ) + mock_session.get.assert_any_call( + "https://mysmob.api.connect.skoda-auto.cz/api/v3/vehicle-maintenance/vehicles/test_vin/report", + headers=skoda_instance.headers + ) From eace5077bcf0f7e052f31641bc9c31e7ae9e2c26 Mon Sep 17 00:00:00 2001 From: vuffiraa72 Date: Sun, 29 Mar 2026 11:35:33 +0200 Subject: [PATCH 3/7] Update fetch_soc to include odometer data retrieval from server --- packages/modules/vehicles/skoda/api.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/packages/modules/vehicles/skoda/api.py b/packages/modules/vehicles/skoda/api.py index 33cfb805b8..d9ef882627 100755 --- a/packages/modules/vehicles/skoda/api.py +++ b/packages/modules/vehicles/skoda/api.py @@ -26,8 +26,8 @@ def fetch_soc(conf: Skoda, vehicle: int) -> Union[int, float, str]: loop = new_event_loop() set_event_loop(loop) - # get soc, range from server + # get soc, range, timestamp, and odometer from server a = api(conf, vehicle) - soc, range, soc_ts, soc_tsX = loop.run_until_complete(a._fetch_soc()) + soc, range, soc_ts, soc_tsX, odometer = loop.run_until_complete(a._fetch_soc()) - return soc, range, soc_ts, soc_tsX + return soc, range, soc_ts, soc_tsX, odometer From f6370a49fb06d9b30390a367b2ffb759839da2ca Mon Sep 17 00:00:00 2001 From: vuffiraa72 Date: Sun, 29 Mar 2026 13:36:00 +0200 Subject: [PATCH 4/7] Add odometer data retrieval to Cupra API integration --- packages/modules/vehicles/cupra/api.py | 4 ++-- packages/modules/vehicles/cupra/libcupra.py | 20 +++++++++++++++++++- packages/modules/vehicles/cupra/soc.py | 6 +++--- 3 files changed, 24 insertions(+), 6 deletions(-) diff --git a/packages/modules/vehicles/cupra/api.py b/packages/modules/vehicles/cupra/api.py index cb877b8e2b..9163a99b8d 100755 --- a/packages/modules/vehicles/cupra/api.py +++ b/packages/modules/vehicles/cupra/api.py @@ -28,6 +28,6 @@ def fetch_soc(conf: Cupra, vehicle: int) -> Union[int, float, str]: # get soc, range from server a = api(conf, vehicle) - soc, range, soc_ts, soc_tsX = loop.run_until_complete(a._fetch_soc()) + soc, range, soc_ts, soc_tsX, odometer = loop.run_until_complete(a._fetch_soc()) - return soc, range, soc_ts, soc_tsX + return soc, range, soc_ts, soc_tsX, odometer diff --git a/packages/modules/vehicles/cupra/libcupra.py b/packages/modules/vehicles/cupra/libcupra.py index 317df0ee8c..553302b442 100755 --- a/packages/modules/vehicles/cupra/libcupra.py +++ b/packages/modules/vehicles/cupra/libcupra.py @@ -253,6 +253,7 @@ async def refresh_tokens(self): async def get_status(self): status_url = f"{API_BASE}/vehicles/{self.vin}/charging/status" + milage_url = f"{API_BASE}/v1/vehicles/{self.vin}/mileage" response = await self.session.get(status_url, headers=self.headers) # If first attempt fails, try to refresh tokens @@ -277,10 +278,27 @@ async def get_status(self): status_data = await response.json() self.log.debug(f"Status data from Cupra API: {status_data}") + # Fetch mileage data + response = await self.session.get(milage_url, headers=self.headers) + if response.status >= 400: + self.log.error("Get mileage failed") + odometer = None + else: + mileage_data = await response.json() + self.log.debug(f"Mileage data from Cupra API: {mileage_data}") + odometer = mileage_data.get('mileageKm', None) + + battery_value = { + 'currentSOC_pct': status_data['status']['battery']['currentSOC_pct'], + 'cruisingRangeElectric_km': status_data['status']['battery']['cruisingRangeElectric_km'], + 'carCapturedTimestamp': status_data['status']['battery']['carCapturedTimestamp'], + 'odometer': odometer, + } + return { 'charging': { 'batteryStatus': { - 'value': status_data['status']['battery'] + 'value': battery_value, } } } diff --git a/packages/modules/vehicles/cupra/soc.py b/packages/modules/vehicles/cupra/soc.py index b5326c192b..cc6cefb0c2 100755 --- a/packages/modules/vehicles/cupra/soc.py +++ b/packages/modules/vehicles/cupra/soc.py @@ -16,9 +16,9 @@ def fetch(vehicle_update_data: VehicleUpdateData, config: Cupra, vehicle: int) -> CarState: - soc, range, soc_ts, soc_tsX = api.fetch_soc(config, vehicle) - log.info("Result: soc=" + str(soc)+", range=" + str(range) + "@" + soc_ts) - return CarState(soc=soc, range=range, soc_timestamp=soc_tsX) + soc, range, soc_ts, soc_tsX, odometer = api.fetch_soc(config, vehicle) + log.info("Result: soc=" + str(soc)+", range=" + str(range) + "@" + soc_ts + ", odometer=" + str(odometer)) + return CarState(soc=soc, range=range, soc_timestamp=soc_tsX, odometer=odometer) def create_vehicle(vehicle_config: Cupra, vehicle: int): From f71c907fb864dad6d7d9bc8a448bd37a8717eaef Mon Sep 17 00:00:00 2001 From: vuffiraa72 Date: Sun, 29 Mar 2026 13:38:16 +0200 Subject: [PATCH 5/7] Add unit tests for Cupra API odometer data retrieval and error handling --- packages/modules/vehicles/cupra/soc_test.py | 170 ++++++++++++++++++++ 1 file changed, 170 insertions(+) create mode 100644 packages/modules/vehicles/cupra/soc_test.py diff --git a/packages/modules/vehicles/cupra/soc_test.py b/packages/modules/vehicles/cupra/soc_test.py new file mode 100644 index 0000000000..ade6ce013c --- /dev/null +++ b/packages/modules/vehicles/cupra/soc_test.py @@ -0,0 +1,170 @@ +from unittest.mock import Mock, MagicMock +import pytest +import asyncio +from modules.common import store +from modules.common.abstract_vehicle import VehicleUpdateData +from modules.common.component_context import SingleComponentUpdateContext +from modules.vehicles.cupra import api +from modules.vehicles.cupra.soc import create_vehicle +from modules.vehicles.cupra.config import Cupra, CupraConfiguration +from modules.vehicles.cupra.libcupra import cupra as CupraApi + + +class TestCupra: + @pytest.fixture(autouse=True) + def set_up(self, monkeypatch): + self.mock_context_exit = Mock(return_value=True) + self.mock_fetch_soc = Mock( + name="fetch_soc", + return_value=(60, 320, "2026-03-29T11:00:00Z", 1774782000.0, 45678), + ) + self.mock_value_store = Mock(name="value_store") + monkeypatch.setattr(api, "fetch_soc", self.mock_fetch_soc) + monkeypatch.setattr(store, "get_car_value_store", Mock(return_value=self.mock_value_store)) + monkeypatch.setattr(SingleComponentUpdateContext, '__exit__', self.mock_context_exit) + + def test_update_updates_value_store(self): + # setup + config = Cupra(configuration=CupraConfiguration(user_id="test_user", password="test_password", vin="test_vin")) + + # execution + create_vehicle(config, 1).update(VehicleUpdateData()) + + # evaluation + self.assert_context_manager_called_with(None) + self.mock_fetch_soc.assert_called_once_with(config, 1) + assert self.mock_value_store.set.call_count == 1 + call_args = self.mock_value_store.set.call_args[0][0] + assert call_args.soc == 60 + assert call_args.range == 320 + assert call_args.soc_timestamp == 1774782000.0 + assert call_args.odometer == 45678 + + def test_update_passes_errors_to_context(self): + # setup + dummy_error = Exception("Der SoC kann nicht ausgelesen werden") + self.mock_fetch_soc.side_effect = dummy_error + config = Cupra(configuration=CupraConfiguration(user_id="test_user", password="test_password", vin="test_vin")) + + # execution + create_vehicle(config, 1).update(VehicleUpdateData()) + + # evaluation + self.assert_context_manager_called_with_substr(dummy_error) + + def assert_context_manager_called_with(self, error): + assert self.mock_context_exit.call_count == 1 + assert self.mock_context_exit.call_args[0][1] is error + + def assert_context_manager_called_with_substr(self, error): + assert self.mock_context_exit.call_count == 1 + assert str(error) in str(self.mock_context_exit.call_args[0][1]) + + +class MockAiohttpResponse: + def __init__(self, json_data, status_code): + self._json_data = json_data + self.status = status_code + + async def json(self): + return self._json_data + + def release(self): + pass + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc_val, exc_tb): + pass + + +class TestCupraGetStatus: + @pytest.fixture + def mock_session(self): + session = MagicMock() + + async def get_side_effect(*args, **kwargs): + return session.get.return_value + session.get = MagicMock(side_effect=get_side_effect) + return session + + @pytest.fixture + def cupra_instance(self, mock_session): + instance = CupraApi(mock_session) + instance.set_vin("test_vin") + instance.headers = {"Authorization": "Bearer test_token"} + return instance + + def test_get_status_success(self, cupra_instance, mock_session): + # setup + status_response_data = { + "status": { + "battery": { + "currentSOC_pct": 67, + "cruisingRangeElectric_km": 305, + "carCapturedTimestamp": "2026-03-29T11:20:00Z" + } + } + } + mileage_response_data = { + "mileageKm": 77889 + } + responses = [ + MockAiohttpResponse(status_response_data, 200), + MockAiohttpResponse(mileage_response_data, 200) + ] + + async def side_effect_func(*args, **kwargs): + return responses.pop(0) + mock_session.get.side_effect = side_effect_func + + # execution + status = asyncio.run(cupra_instance.get_status()) + + # evaluation + assert status['charging']['batteryStatus']['value']['currentSOC_pct'] == 67 + assert status['charging']['batteryStatus']['value']['cruisingRangeElectric_km'] == 305 + assert status['charging']['batteryStatus']['value']['carCapturedTimestamp'] == "2026-03-29T11:20:00Z" + assert status['charging']['batteryStatus']['value']['odometer'] == 77889 + assert mock_session.get.call_count == 2 + mock_session.get.assert_any_call( + "https://ola.prod.code.seat.cloud.vwgroup.com/vehicles/test_vin/charging/status", + headers=cupra_instance.headers + ) + mock_session.get.assert_any_call( + "https://ola.prod.code.seat.cloud.vwgroup.com/v1/vehicles/test_vin/mileage", + headers=cupra_instance.headers + ) + + def test_get_status_mileage_error_sets_odometer_none(self, cupra_instance, mock_session): + # setup + status_response_data = { + "status": { + "battery": { + "currentSOC_pct": 67, + "cruisingRangeElectric_km": 305, + "carCapturedTimestamp": "2026-03-29T11:20:00Z" + } + } + } + mileage_error_response = { + "error": "service_unavailable" + } + responses = [ + MockAiohttpResponse(status_response_data, 200), + MockAiohttpResponse(mileage_error_response, 503) + ] + + async def side_effect_func(*args, **kwargs): + return responses.pop(0) + mock_session.get.side_effect = side_effect_func + + # execution + status = asyncio.run(cupra_instance.get_status()) + + # evaluation + assert status['charging']['batteryStatus']['value']['currentSOC_pct'] == 67 + assert status['charging']['batteryStatus']['value']['cruisingRangeElectric_km'] == 305 + assert status['charging']['batteryStatus']['value']['carCapturedTimestamp'] == "2026-03-29T11:20:00Z" + assert status['charging']['batteryStatus']['value']['odometer'] is None From 8f0ef192f0f87133f3dc51e36650c5caab3e43dc Mon Sep 17 00:00:00 2001 From: vuffiraa72 Date: Sun, 29 Mar 2026 13:41:21 +0200 Subject: [PATCH 6/7] Update fetch_soc comment to clarify data retrieval including odometer --- packages/modules/vehicles/cupra/api.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/modules/vehicles/cupra/api.py b/packages/modules/vehicles/cupra/api.py index 9163a99b8d..dc06d088c2 100755 --- a/packages/modules/vehicles/cupra/api.py +++ b/packages/modules/vehicles/cupra/api.py @@ -26,7 +26,7 @@ def fetch_soc(conf: Cupra, vehicle: int) -> Union[int, float, str]: loop = new_event_loop() set_event_loop(loop) - # get soc, range from server + # get soc, range, timestamp, and odometer from server a = api(conf, vehicle) soc, range, soc_ts, soc_tsX, odometer = loop.run_until_complete(a._fetch_soc()) From ba892ed031418a78f2450a9655767928fc7707e2 Mon Sep 17 00:00:00 2001 From: vuffiraa72 Date: Mon, 30 Mar 2026 17:37:26 +0200 Subject: [PATCH 7/7] Refactor Cupra and Skoda API to return structured SoCResult type and fix mileage URL variable name --- packages/modules/vehicles/cupra/api.py | 9 ++++++--- packages/modules/vehicles/cupra/libcupra.py | 4 ++-- packages/modules/vehicles/skoda/api.py | 9 ++++++--- 3 files changed, 14 insertions(+), 8 deletions(-) diff --git a/packages/modules/vehicles/cupra/api.py b/packages/modules/vehicles/cupra/api.py index dc06d088c2..7533cecc1d 100755 --- a/packages/modules/vehicles/cupra/api.py +++ b/packages/modules/vehicles/cupra/api.py @@ -2,25 +2,28 @@ import aiohttp from asyncio import new_event_loop, set_event_loop -from typing import Union +from typing import Optional, Tuple from modules.vehicles.cupra import libcupra from modules.vehicles.cupra.config import Cupra from modules.vehicles.vwgroup.vwgroup import VwGroup +SoCResult = Tuple[int, float, str, float, Optional[float]] + + class api(VwGroup): def __init__(self, conf: Cupra, vehicle: int): super().__init__(conf, vehicle) # async method, called from sync fetch_soc, required because libvwid/libskoda/libcupra expect async environment - async def _fetch_soc(self) -> Union[int, float, str]: + async def _fetch_soc(self) -> SoCResult: async with aiohttp.ClientSession() as self.session: cupra = libcupra.cupra(self.session) return await super().request_data(cupra) -def fetch_soc(conf: Cupra, vehicle: int) -> Union[int, float, str]: +def fetch_soc(conf: Cupra, vehicle: int) -> SoCResult: # prepare and call async method loop = new_event_loop() diff --git a/packages/modules/vehicles/cupra/libcupra.py b/packages/modules/vehicles/cupra/libcupra.py index 553302b442..9332ff2acd 100755 --- a/packages/modules/vehicles/cupra/libcupra.py +++ b/packages/modules/vehicles/cupra/libcupra.py @@ -253,7 +253,7 @@ async def refresh_tokens(self): async def get_status(self): status_url = f"{API_BASE}/vehicles/{self.vin}/charging/status" - milage_url = f"{API_BASE}/v1/vehicles/{self.vin}/mileage" + mileage_url = f"{API_BASE}/v1/vehicles/{self.vin}/mileage" response = await self.session.get(status_url, headers=self.headers) # If first attempt fails, try to refresh tokens @@ -279,7 +279,7 @@ async def get_status(self): self.log.debug(f"Status data from Cupra API: {status_data}") # Fetch mileage data - response = await self.session.get(milage_url, headers=self.headers) + response = await self.session.get(mileage_url, headers=self.headers) if response.status >= 400: self.log.error("Get mileage failed") odometer = None diff --git a/packages/modules/vehicles/skoda/api.py b/packages/modules/vehicles/skoda/api.py index d9ef882627..7cb2fee565 100755 --- a/packages/modules/vehicles/skoda/api.py +++ b/packages/modules/vehicles/skoda/api.py @@ -2,25 +2,28 @@ import aiohttp from asyncio import new_event_loop, set_event_loop -from typing import Union +from typing import Optional, Tuple from modules.vehicles.skoda import libskoda from modules.vehicles.skoda.config import Skoda from modules.vehicles.vwgroup.vwgroup import VwGroup +SoCResult = Tuple[int, float, str, float, Optional[float]] + + class api(VwGroup): def __init__(self, conf: Skoda, vehicle: int): super().__init__(conf, vehicle) # async method, called from sync fetch_soc, required because libvwid/libskoda expect async environment - async def _fetch_soc(self) -> Union[int, float, str]: + async def _fetch_soc(self) -> SoCResult: async with aiohttp.ClientSession() as self.session: skoda = libskoda.skoda(self.session) return await super().request_data(skoda) -def fetch_soc(conf: Skoda, vehicle: int) -> Union[int, float, str]: +def fetch_soc(conf: Skoda, vehicle: int) -> SoCResult: # prepare and call async method loop = new_event_loop()