Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 40 additions & 8 deletions packages/modules/vehicles/skoda/libskoda.py
Original file line number Diff line number Diff line change
Expand Up @@ -214,20 +214,21 @@ async def refresh_tokens(self):
return True

async def get_status(self):
status_url = f"{API_BASE}/v2/vehicle-status/{self.vin}/driving-range"
response = await self.session.get(status_url, headers=self.headers)
vehicle_status_url = f"{API_BASE}/v2/vehicle-status/{self.vin}/driving-range"
charging_url = f"{API_BASE}/v1/charging/{self.vin}"
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(status_url, headers=self.headers)
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(status_url, headers=self.headers)
response = await self.session.get(vehicle_status_url, headers=self.headers)
else:
self.log.error("Reconnect failed")
return {}
Expand All @@ -237,15 +238,46 @@ async def get_status(self):
return {}

status_data = await response.json()
self.log.debug(f"Status data from Skoda API: {status_data}")
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 = {}
if 'primaryEngineRange' in status_data and status_data['primaryEngineRange']['engineType'] == "electric":
electric_engine_range = status_data['primaryEngineRange']
elif 'secondaryEngineRange' in status_data and status_data['secondaryEngineRange']['engineType'] == "electric":
electric_engine_range = status_data['secondaryEngineRange']

required_keys = ['currentSoCInPercent', 'remainingRangeInKm']
if not all(k in electric_engine_range for k in required_keys) or 'carCapturedTimestamp' not in status_data:
self.log.info("vehicle-status did not contain all values, trying charging_url")
response = await self.session.get(charging_url, headers=self.headers)

if response.status >= 400:
self.log.error("Get status from charging_url failed")
return {}

status_data = await response.json()
self.log.debug(f"Status data from Skoda API (charging): {status_data}")

soc = status_data['status']['battery']['stateOfChargeInPercent']
range_km = (
status_data['status']['battery'].get('remainingCruisingRangeInMeters', 1000) / 1000
)
else:
soc = electric_engine_range['currentSoCInPercent']
range_km = electric_engine_range['remainingRangeInKm']

timestamp = status_data['carCapturedTimestamp'].split('.')[0]
if not timestamp.endswith('Z'):
timestamp += 'Z'

return {
'charging': {
'batteryStatus': {
'value': {
'currentSOC_pct': status_data['primaryEngineRange']['currentSoCInPercent'],
'cruisingRangeElectric_km': status_data['primaryEngineRange']['remainingRangeInKm'],
'carCapturedTimestamp': status_data['carCapturedTimestamp'].split('.')[0] + 'Z',
'currentSOC_pct': soc,
'cruisingRangeElectric_km': range_km,
'carCapturedTimestamp': timestamp,
}
}
}
Expand Down
209 changes: 209 additions & 0 deletions packages/modules/vehicles/skoda/soc_test.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,209 @@
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.skoda import api
from modules.vehicles.skoda.soc import create_vehicle
from modules.vehicles.skoda.config import Skoda, SkodaConfiguration
from modules.vehicles.skoda.libskoda import skoda as SkodaApi


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_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 = Skoda(configuration=SkodaConfiguration(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 == 50
assert call_args.range == 250
assert call_args.soc_timestamp == 1760822400.0

def test_update_passes_errors_to_context(self):
# setup
dummy_error = Exception("API Error")
self.mock_fetch_soc.side_effect = dummy_error
config = Skoda(configuration=SkodaConfiguration(user_id="test_user", password="test_password", vin="test_vin"))

# execution
create_vehicle(config, 1).update(VehicleUpdateData())

# evaluation
self.assert_context_manager_called_with(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


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 TestSkodaGetStatus:
@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 skoda_instance(self, mock_session):
instance = SkodaApi(mock_session)
instance.set_vin("test_vin")
instance.headers = {"Authorization": "Bearer test_token"}
return instance

def test_get_status_success_primary_url(self, skoda_instance, mock_session):
# setup
response_data = {
"carType": "electric",
"totalRangeInKm": 291,
"primaryEngineRange": {
"engineType": "electric",
"currentSoCInPercent": 66,
"remainingRangeInKm": 291
},
"carCapturedTimestamp": "2025-10-17T15:40:35.679Z"
}
mock_session.get.return_value = MockAiohttpResponse(response_data, 200)

# execution
status = asyncio.run(skoda_instance.get_status())

# evaluation
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(
"https://mysmob.api.connect.skoda-auto.cz/api/v2/vehicle-status/test_vin/driving-range",
headers=skoda_instance.headers
)

def test_get_status_fallback_to_charging_url(self, skoda_instance, mock_session):
# setup
vehicle_status_response_data = {
"carType": "electric",
"primaryEngineRange": {
"engineType": "electric",
"currentSoCInPercent": 42
},
"carCapturedTimestamp": "2025-09-26T11:00:25.848Z"
}
charging_url_response_data = {
"isVehicleInSavedLocation": False,
"status": {
"chargingRateInKilometersPerHour": 0.0,
"chargePowerInKw": 0.0,
"remainingTimeToFullyChargedInMinutes": 0,
"battery": {
"remainingCruisingRangeInMeters": 291000,
"stateOfChargeInPercent": 66
}
},
"settings": {
"targetStateOfChargeInPercent": 80,
"preferredChargeMode": "MANUAL",
"availableChargeModes": [
"MANUAL"
],
"chargingCareMode": "ACTIVATED",
"autoUnlockPlugWhenCharged": "OFF",
"maxChargeCurrentAc": "MAXIMUM"
},
"carCapturedTimestamp": "2025-10-17T15:38:55Z",
"errors": [
{
"type": "STATUS_OF_CHARGING_NOT_AVAILABLE",
"description": "Status of charging is not available."
}
]
}
responses = [
MockAiohttpResponse(vehicle_status_response_data, 200),
MockAiohttpResponse(charging_url_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'] == 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
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
)

def test_get_status_timestamp_without_milliseconds(self, skoda_instance, mock_session):
# setup
response_data = {
"carType": "hybrid",
"totalRangeInKm": 647,
"primaryEngineRange": {
"engineType": "gasoline",
"currentSoCInPercent": 100,
"currentFuelLevelInPercent": 100,
"remainingRangeInKm": 600
},
"secondaryEngineRange": {
"engineType": "electric",
"currentSoCInPercent": 42,
"currentFuelLevelInPercent": 88,
"remainingRangeInKm": 47
},
"carCapturedTimestamp": "2025-10-06T10:12:44Z"
}
mock_session.get.return_value = MockAiohttpResponse(response_data, 200)

# execution
status = asyncio.run(skoda_instance.get_status())

# evaluation
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"