jpawlowski/hass.tibber_prices
1
0
Fork 0
mirror of https://github.com/jpawlowski/hass.tibber_prices.git synced 2026-03-29 21:03:40 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

fix(api): handle None values in API responses to prevent AttributeError

Browse source 
Fixed issue #60 where Tibber API temporarily returning incomplete data
(None values during maintenance) caused AttributeError crashes.

Root cause: `.get(key, default)` returns None when key exists with None value,
causing chained `.get()` calls to crash (None.get() → AttributeError).

Changes:
- api/helpers.py: Use `or {}` pattern in flatten_price_info() to handle
  None values (priceInfo, priceInfoRange, today, tomorrow)
- entity.py: Use `or {}` pattern in _get_fallback_device_info() for address dict
- coordinator/data_fetching.py: Add _validate_user_data() method (67 lines)
  to reject incomplete API responses before caching
- coordinator/data_fetching.py: Modify _get_currency_for_home() to raise
  exceptions instead of silent EUR fallback
- coordinator/data_fetching.py: Add home_id parameter to constructor
- coordinator/core.py: Pass home_id to TibberPricesDataFetcher
- tests/test_user_data_validation.py: Add 12 test cases for validation logic

Architecture improvement: Instead of defensive coding with fallbacks,
implement validation to reject incomplete data upfront. This prevents
caching temporary API errors and ensures currency is always known
(critical for price calculations).

Impact: Integration now handles API maintenance periods gracefully without
crashes. No silent EUR fallbacks - raises exceptions if currency unavailable,
ensuring data integrity. Users see clear errors instead of wrong calculations.

Fixes #60
This commit is contained in:
Julian Pawlowski 2025-12-13 14:02:30 +00:00
parent 6c741e8392
commit 87f0022baa
5 changed files with 434 additions and 18 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
custom_components/tibber_prices/api/helpers.py
View file

@ -340,7 +340,8 @@ def flatten_price_info(subscription: dict) -> list[dict]:
A flat list containing all price dictionaries (startsAt, total, level).
"""
price_info_range = subscription.get("priceInfoRange", {})
# Use 'or {}' to handle None values (API may return None during maintenance)
price_info_range = subscription.get("priceInfoRange") or {}
# Transform priceInfoRange edges data (extract historical quarter-hourly prices)
# This contains 192 intervals (2 days) starting from day before yesterday midnight
@ -355,8 +356,6 @@ def flatten_price_info(subscription: dict) -> list[dict]:
historical_prices.append(edge["node"])
# Return all intervals as a single flattened array
return (
historical_prices
+ subscription.get("priceInfo", {}).get("today", [])
+ subscription.get("priceInfo", {}).get("tomorrow", [])
)
# Use 'or {}' to handle None values (API may return None during maintenance)
price_info = subscription.get("priceInfo") or {}
return historical_prices + (price_info.get("today") or []) + (price_info.get("tomorrow") or [])

1
custom_components/tibber_prices/coordinator/core.py
View file

@ -212,6 +212,7 @@ class TibberPricesDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
log_prefix=self._log_prefix,
user_update_interval=timedelta(days=1),
time=self.time,
home_id=self._home_id,
)
# Create period calculator BEFORE data transformer (transformer needs it in lambda)
self._period_calculator = TibberPricesPeriodCalculator(

128
custom_components/tibber_prices/coordinator/data_fetching.py
View file

@ -34,13 +34,14 @@ _LOGGER = logging.getLogger(__name__)
class TibberPricesDataFetcher:
"""Handles data fetching, caching, and main/subentry coordination."""
def __init__(
def __init__( # noqa: PLR0913
self,
api: TibberPricesApiClient,
store: Any,
log_prefix: str,
user_update_interval: timedelta,
time: TibberPricesTimeService,
home_id: str,
) -> None:
"""Initialize the data fetcher."""
self.api = api
@ -48,6 +49,7 @@ class TibberPricesDataFetcher:
self._log_prefix = log_prefix
self._user_update_interval = user_update_interval
self.time: TibberPricesTimeService = time
self.home_id = home_id
# Cached data
self._cached_price_data: dict[str, Any] | None = None
@ -91,10 +93,78 @@ class TibberPricesDataFetcher:
)
await cache.save_cache(self._store, cache_data, self._log_prefix)
def _validate_user_data(self, user_data: dict, home_id: str) -> bool: # noqa: PLR0911
"""
Validate user data completeness.
Rejects incomplete/invalid data from API to prevent caching temporary errors.
Currency information is critical - if missing, we cannot safely calculate prices.
Args:
user_data: User data dict from API.
home_id: Home ID to validate against.
Returns:
True if data is valid and complete, False otherwise.
"""
if not user_data:
self._log("warning", "User data validation failed: Empty data")
return False
viewer = user_data.get("viewer")
if not viewer or not isinstance(viewer, dict):
self._log("warning", "User data validation failed: Missing or invalid viewer")
return False
homes = viewer.get("homes")
if not homes or not isinstance(homes, list) or len(homes) == 0:
self._log("warning", "User data validation failed: No homes found")
return False
# Find our home and validate it has required data
home_found = False
for home in homes:
if home.get("id") == home_id:
home_found = True
# Validate home has timezone (required for cursor calculation)
if not home.get("timeZone"):
self._log("warning", "User data validation failed: Home %s missing timezone", home_id)
return False
# Currency is critical - if home has subscription, must have currency
subscription = home.get("currentSubscription")
if subscription and subscription is not None:
price_info = subscription.get("priceInfo")
if price_info and price_info is not None:
current = price_info.get("current")
if current and current is not None:
currency = current.get("currency")
if not currency:
self._log(
"warning",
"User data validation failed: Home %s has subscription but no currency",
home_id,
)
return False
break
if not home_found:
self._log("warning", "User data validation failed: Home %s not found in homes list", home_id)
return False
self._log("debug", "User data validation passed for home %s", home_id)
return True
async def update_user_data_if_needed(self, current_time: datetime) -> bool:
"""
Update user data if needed (daily check).
Only accepts complete and valid data. If API returns incomplete data
(e.g., during maintenance), keeps existing cached data and retries later.
Returns:
True if user data was updated, False otherwise
@ -103,6 +173,16 @@ class TibberPricesDataFetcher:
try:
self._log("debug", "Updating user data")
user_data = await self.api.async_get_viewer_details()
# Validate before caching
if not self._validate_user_data(user_data, self.home_id):
self._log(
"warning",
"Rejecting incomplete user data from API - keeping existing cached data",
)
return False # Keep existing data, don't update timestamp
# Data is valid, cache it
self._cached_user_data = user_data
self._last_user_update = current_time
self._log("debug", "User data updated successfully")
@ -182,6 +262,13 @@ class TibberPricesDataFetcher:
self._log("info", "User data not cached, fetching before price data")
try:
user_data = await self.api.async_get_viewer_details()
# Validate data before accepting it (especially on initial setup)
if not self._validate_user_data(user_data, self.home_id):
msg = "Received incomplete user data from API - cannot proceed with price fetching"
self._log("error", msg)
raise TibberPricesApiClientError(msg) # noqa: TRY301
self._cached_user_data = user_data
self._last_user_update = current_time
except (
@ -220,25 +307,46 @@ class TibberPricesDataFetcher:
}
def _get_currency_for_home(self, home_id: str) -> str:
"""Get currency for a specific home from cached user_data."""
"""
Get currency for a specific home from cached user_data.
Returns:
Currency code (e.g., "EUR", "NOK", "SEK").
Raises:
TibberPricesApiClientError: If currency cannot be determined.
"""
if not self._cached_user_data:
self._log("warning", "No user data cached, using EUR as default currency")
return "EUR"
msg = "No user data cached - cannot determine currency"
self._log("error", msg)
raise TibberPricesApiClientError(msg)
viewer = self._cached_user_data.get("viewer", {})
homes = viewer.get("homes", [])
for home in homes:
if home.get("id") == home_id:
# Extract currency from nested structure (with fallback to EUR)
currency = (
home.get("currentSubscription", {}).get("priceInfo", {}).get("current", {}).get("currency", "EUR")
)
# Extract currency from nested structure
# Use 'or {}' to handle None values (homes without active subscription)
subscription = home.get("currentSubscription") or {}
price_info = subscription.get("priceInfo") or {}
current = price_info.get("current") or {}
currency = current.get("currency")
if not currency:
# Home without active subscription - cannot determine currency
msg = f"Home {home_id} has no active subscription - currency unavailable"
self._log("error", msg)
raise TibberPricesApiClientError(msg)
self._log("debug", "Extracted currency %s for home %s", currency, home_id)
return currency
self._log("warning", "Home %s not found in user data, using EUR as default", home_id)
return "EUR"
# Home not found in cached data - data validation should have caught this
msg = f"Home {home_id} not found in user data - data validation failed"
self._log("error", msg)
raise TibberPricesApiClientError(msg)
def _check_home_exists(self, home_id: str) -> bool:
"""

6
custom_components/tibber_prices/entity.py
View file

@ -118,8 +118,10 @@ class TibberPricesEntity(CoordinatorEntity[TibberPricesDataUpdateCoordinator]):
return "Tibber Home", None
try:
address1 = str(self.coordinator.data.get("address", {}).get("address1", ""))
city = str(self.coordinator.data.get("address", {}).get("city", ""))
# Use 'or {}' to handle None values (API may return None during maintenance)
address = self.coordinator.data.get("address") or {}
address1 = str(address.get("address1", ""))
city = str(address.get("city", ""))
app_nickname = str(self.coordinator.data.get("appNickname", ""))
home_type = str(self.coordinator.data.get("type", ""))

306
tests/test_user_data_validation.py Normal file
View file

@ -0,0 +1,306 @@
"""
Test user data validation and currency extraction.
This test covers issue #60 where the Tibber API can temporarily return
incomplete or invalid data during maintenance or cache refresh periods.
The issue manifested when:
1. User updated integration while Tibber API was returning incomplete data
2. Integration accepted and cached the incomplete data
3. Next access crashed or used wrong currency (EUR fallback)
4. Next day at 13:02, user_data refreshed (24h interval) with correct data
5. Issue "fixed itself" because cache was updated with valid data
The fix implements data validation that:
- Rejects incomplete user data from API
- Keeps existing cached data when validation fails
- Only accepts data with complete home info (timezone, currency if subscription exists)
- Raises exception if currency cannot be determined (no silent EUR fallback)
"""
from datetime import timedelta
import pytest
from custom_components.tibber_prices.api.exceptions import TibberPricesApiClientError
from custom_components.tibber_prices.api.helpers import flatten_price_info
from custom_components.tibber_prices.coordinator.data_fetching import (
TibberPricesDataFetcher,
)
@pytest.mark.unit
def test_validate_user_data_complete(mock_api_client, mock_time_service, mock_store) -> None: # noqa: ANN001
"""Test that complete user data passes validation."""
fetcher = TibberPricesDataFetcher(
api=mock_api_client,
store=mock_store,
log_prefix="[Test]",
user_update_interval=timedelta(days=1),
time=mock_time_service,
home_id="home-123",
)
user_data = {
"viewer": {
"homes": [
{
"id": "home-123",
"timeZone": "Europe/Berlin",
"currentSubscription": {
"priceInfo": {
"current": {
"currency": "EUR",
}
}
},
}
]
}
}
assert fetcher._validate_user_data(user_data, "home-123") is True # noqa: SLF001 # noqa: SLF001
@pytest.mark.unit
def test_validate_user_data_none_subscription(mock_api_client, mock_time_service, mock_store) -> None: # noqa: ANN001
"""Test that user data without subscription (but with timezone) passes validation."""
fetcher = TibberPricesDataFetcher(
api=mock_api_client,
store=mock_store,
log_prefix="[Test]",
user_update_interval=timedelta(days=1),
time=mock_time_service,
home_id="home-123",
)
user_data = {
"viewer": {
"homes": [
{
"id": "home-123",
"timeZone": "Europe/Berlin",
"currentSubscription": None, # No active subscription
}
]
}
}
# Should pass validation - timezone is present, subscription being None is valid
assert fetcher._validate_user_data(user_data, "home-123") is True # noqa: SLF001 # noqa: SLF001
@pytest.mark.unit
def test_validate_user_data_missing_timezone(mock_api_client, mock_time_service, mock_store) -> None: # noqa: ANN001
"""Test that user data without timezone fails validation."""
fetcher = TibberPricesDataFetcher(
api=mock_api_client,
store=mock_store,
log_prefix="[Test]",
user_update_interval=timedelta(days=1),
time=mock_time_service,
home_id="home-123",
)
user_data = {
"viewer": {
"homes": [
{
"id": "home-123",
# Missing timeZone!
"currentSubscription": {
"priceInfo": {
"current": {
"currency": "EUR",
}
}
},
}
]
}
}
assert fetcher._validate_user_data(user_data, "home-123") is False # noqa: SLF001 # noqa: SLF001
@pytest.mark.unit
def test_validate_user_data_subscription_without_currency(mock_api_client, mock_time_service, mock_store) -> None: # noqa: ANN001
"""Test that user data with subscription but no currency fails validation."""
fetcher = TibberPricesDataFetcher(
api=mock_api_client,
store=mock_store,
log_prefix="[Test]",
user_update_interval=timedelta(days=1),
time=mock_time_service,
home_id="home-123",
)
user_data = {
"viewer": {
"homes": [
{
"id": "home-123",
"timeZone": "Europe/Berlin",
"currentSubscription": {
"priceInfo": {
"current": {} # Currency missing!
}
},
}
]
}
}
assert fetcher._validate_user_data(user_data, "home-123") is False # noqa: SLF001 # noqa: SLF001
@pytest.mark.unit
def test_validate_user_data_home_not_found(mock_api_client, mock_time_service, mock_store) -> None: # noqa: ANN001
"""Test that user data without the requested home fails validation."""
fetcher = TibberPricesDataFetcher(
api=mock_api_client,
store=mock_store,
log_prefix="[Test]",
user_update_interval=timedelta(days=1),
time=mock_time_service,
home_id="home-123",
)
user_data = {
"viewer": {
"homes": [
{
"id": "other-home",
"timeZone": "Europe/Berlin",
}
]
}
}
assert fetcher._validate_user_data(user_data, "home-123") is False # noqa: SLF001 # noqa: SLF001
@pytest.mark.unit
def test_get_currency_raises_on_no_cached_data(mock_api_client, mock_time_service, mock_store) -> None: # noqa: ANN001
"""Test that _get_currency_for_home raises exception when no data cached."""
fetcher = TibberPricesDataFetcher(
api=mock_api_client,
store=mock_store,
log_prefix="[Test]",
user_update_interval=timedelta(days=1),
time=mock_time_service,
home_id="home-123",
)
# No cached data
with pytest.raises(TibberPricesApiClientError, match="No user data cached"):
fetcher._get_currency_for_home("home-123") # noqa: SLF001 # noqa: SLF001
@pytest.mark.unit
def test_get_currency_raises_on_no_subscription(mock_api_client, mock_time_service, mock_store) -> None: # noqa: ANN001
"""Test that _get_currency_for_home raises exception when home has no subscription."""
fetcher = TibberPricesDataFetcher(
api=mock_api_client,
store=mock_store,
log_prefix="[Test]",
user_update_interval=timedelta(days=1),
time=mock_time_service,
home_id="home-123",
)
fetcher._cached_user_data = { # noqa: SLF001 # noqa: SLF001
"viewer": {
"homes": [
{
"id": "home-123",
"currentSubscription": None, # No subscription
}
]
}
}
with pytest.raises(TibberPricesApiClientError, match="has no active subscription"):
fetcher._get_currency_for_home("home-123") # noqa: SLF001 # noqa: SLF001
@pytest.mark.unit
def test_get_currency_extracts_valid_currency(mock_api_client, mock_time_service, mock_store) -> None: # noqa: ANN001
"""Test that _get_currency_for_home successfully extracts currency."""
fetcher = TibberPricesDataFetcher(
api=mock_api_client,
store=mock_store,
log_prefix="[Test]",
user_update_interval=timedelta(days=1),
time=mock_time_service,
home_id="home-123",
)
fetcher._cached_user_data = { # noqa: SLF001 # noqa: SLF001
"viewer": {
"homes": [
{
"id": "home-123",
"currentSubscription": {
"priceInfo": {
"current": {
"currency": "NOK",
}
}
},
}
]
}
}
assert fetcher._get_currency_for_home("home-123") == "NOK" # noqa: SLF001 # noqa: SLF001
@pytest.mark.unit
def test_flatten_price_info_with_none_priceinfo() -> None:
"""Test that flatten_price_info handles None priceInfo gracefully."""
subscription = {
"priceInfoRange": {
"edges": [
{"node": {"startsAt": "2025-12-10T00:00:00", "total": 0.25, "level": "NORMAL"}},
]
},
"priceInfo": None, # ← Key exists but value is None
}
# Should not crash, should return only historical prices
result = flatten_price_info(subscription)
assert len(result) == 1
assert result[0]["total"] == 0.25
@pytest.mark.unit
def test_flatten_price_info_with_none_today() -> None:
"""Test that flatten_price_info handles None today gracefully."""
subscription = {
"priceInfoRange": {"edges": []},
"priceInfo": {
"today": None, # ← Key exists but value is None
"tomorrow": [
{"startsAt": "2025-12-13T00:00:00", "total": 0.30, "level": "NORMAL"},
],
},
}
# Should not crash, should return only tomorrow prices
result = flatten_price_info(subscription)
assert len(result) == 1
assert result[0]["total"] == 0.30
@pytest.mark.unit
def test_flatten_price_info_with_all_none() -> None:
"""Test that flatten_price_info handles all None values gracefully."""
subscription = {
"priceInfoRange": None,
"priceInfo": None,
}
# Should not crash, should return empty list
result = flatten_price_info(subscription)
assert result == []