mirror of
https://github.com/jpawlowski/hass.tibber_prices.git
synced 2026-03-30 05:13:40 +00:00
9ee7f81164
Fixes bug where lifecycle sensor attributes (data_completeness, tomorrow_available) didn't update after tomorrow data was successfully fetched from API. Root cause: DataTransformer had cached transformation data but no mechanism to detect when source API data changed (only checked config and midnight turnover). Changes: - coordinator/data_transformation.py: Track source_data_timestamp and invalidate cache when timestamp changes (detects new API data arrival) - coordinator/data_transformation.py: Integrate period calculation into DataTransformer (calculate_periods_fn parameter) for complete single-layer caching - coordinator/core.py: Remove duplicate transformation cache (_cached_transformed_data, _last_transformation_config), simplify _transform_data_for_*() to direct delegation - tests/test_tomorrow_data_refresh.py: Add 3 regression tests for cache invalidation (new timestamp, config change behavior, cache preservation) Impact: Lifecycle sensor attributes now update correctly when new API data arrives. Reduced code by ~40 lines in coordinator, consolidated caching to single layer. All 350 tests passing.
254 lines
8.5 KiB
Python
254 lines
8.5 KiB
Python
"""
|
|
Tests for tomorrow data arrival and cache invalidation.
|
|
|
|
Regression test for the bug where lifecycle sensor attributes (data_completeness,
|
|
tomorrow_available) didn't update after tomorrow data was successfully fetched
|
|
due to cached transformation data.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta
|
|
from unittest.mock import Mock
|
|
from zoneinfo import ZoneInfo
|
|
|
|
import pytest
|
|
|
|
from custom_components.tibber_prices.coordinator.data_transformation import (
|
|
TibberPricesDataTransformer,
|
|
)
|
|
from custom_components.tibber_prices.coordinator.time_service import (
|
|
TibberPricesTimeService,
|
|
)
|
|
|
|
|
|
def create_price_intervals(day_offset: int = 0) -> list[dict]:
|
|
"""Create 96 mock price intervals (quarter-hourly for one day)."""
|
|
base_date = datetime(2025, 11, 22, 0, 0, 0, tzinfo=ZoneInfo("Europe/Oslo"))
|
|
intervals = []
|
|
for i in range(96):
|
|
interval_time = base_date.replace(day=base_date.day + day_offset, hour=i // 4, minute=(i % 4) * 15)
|
|
intervals.append(
|
|
{
|
|
"startsAt": interval_time,
|
|
"total": 20.0 + (i % 10),
|
|
"energy": 18.0 + (i % 10),
|
|
"tax": 2.0,
|
|
"level": "NORMAL",
|
|
}
|
|
)
|
|
return intervals
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_transformation_cache_invalidation_on_new_timestamp() -> None:
|
|
"""
|
|
Test that DataTransformer cache is invalidated when source data timestamp changes.
|
|
|
|
This is the core regression test for the bug:
|
|
- Tomorrow data arrives with NEW timestamp
|
|
- Transformation cache MUST be invalidated
|
|
- Lifecycle attributes MUST be recalculated with new data
|
|
"""
|
|
config_entry = Mock()
|
|
config_entry.entry_id = "test_entry"
|
|
config_entry.data = {"home_id": "home_123"}
|
|
config_entry.options = {
|
|
"price_rating_threshold_low": 75.0,
|
|
"price_rating_threshold_high": 90.0,
|
|
}
|
|
|
|
time_service = TibberPricesTimeService()
|
|
current_time = datetime(2025, 11, 22, 13, 15, 0, tzinfo=ZoneInfo("Europe/Oslo"))
|
|
|
|
# Mock period calculator
|
|
mock_period_calc = Mock()
|
|
mock_period_calc.calculate_periods_for_price_info.return_value = {
|
|
"best_price": [],
|
|
"peak_price": [],
|
|
}
|
|
|
|
# Create transformer
|
|
transformer = TibberPricesDataTransformer(
|
|
config_entry=config_entry,
|
|
log_prefix="[Test]",
|
|
perform_turnover_fn=lambda x: x, # No-op
|
|
calculate_periods_fn=mock_period_calc.calculate_periods_for_price_info,
|
|
time=time_service,
|
|
)
|
|
|
|
# STEP 1: First transformation with only today data (timestamp T1)
|
|
# ================================================================
|
|
data_t1 = {
|
|
"timestamp": current_time,
|
|
"homes": {
|
|
"home_123": {
|
|
"price_info": {
|
|
"yesterday": [],
|
|
"today": create_price_intervals(0),
|
|
"tomorrow": [], # NO TOMORROW YET
|
|
"currency": "EUR",
|
|
}
|
|
}
|
|
},
|
|
}
|
|
|
|
result_t1 = transformer.transform_data_for_main_entry(data_t1)
|
|
assert result_t1 is not None
|
|
assert result_t1["priceInfo"]["tomorrow"] == []
|
|
|
|
# STEP 2: Second call with SAME timestamp should use cache
|
|
# =========================================================
|
|
result_t1_cached = transformer.transform_data_for_main_entry(data_t1)
|
|
assert result_t1_cached is result_t1 # SAME object (cached)
|
|
|
|
# STEP 3: Third call with DIFFERENT timestamp should NOT use cache
|
|
# =================================================================
|
|
new_time = current_time + timedelta(minutes=1)
|
|
data_t2 = {
|
|
"timestamp": new_time, # DIFFERENT timestamp
|
|
"homes": {
|
|
"home_123": {
|
|
"price_info": {
|
|
"yesterday": [],
|
|
"today": create_price_intervals(0),
|
|
"tomorrow": create_price_intervals(1), # NOW HAS TOMORROW
|
|
"currency": "EUR",
|
|
}
|
|
}
|
|
},
|
|
}
|
|
|
|
result_t2 = transformer.transform_data_for_main_entry(data_t2)
|
|
|
|
# CRITICAL ASSERTIONS: Cache must be invalidated
|
|
assert result_t2 is not result_t1 # DIFFERENT object (re-transformed)
|
|
assert len(result_t2["priceInfo"]["tomorrow"]) == 96 # New data present
|
|
assert "periods" in result_t2 # Periods recalculated
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_cache_behavior_on_config_change() -> None:
|
|
"""
|
|
Document current cache behavior when config changes.
|
|
|
|
NOTE: Currently, config changes with same timestamp DO NOT invalidate cache.
|
|
This is acceptable because:
|
|
1. Config changes trigger full coordinator reload (new instance)
|
|
2. The critical bug was about NEW API DATA not updating (timestamp change)
|
|
3. Options changes are handled at coordinator level via invalidate_config_cache()
|
|
"""
|
|
config_entry = Mock()
|
|
config_entry.entry_id = "test_entry"
|
|
config_entry.data = {"home_id": "home_123"}
|
|
config_entry.options = {
|
|
"price_rating_threshold_low": 75.0,
|
|
"price_rating_threshold_high": 90.0,
|
|
}
|
|
|
|
time_service = TibberPricesTimeService()
|
|
current_time = datetime(2025, 11, 22, 13, 15, 0, tzinfo=ZoneInfo("Europe/Oslo"))
|
|
|
|
mock_period_calc = Mock()
|
|
mock_period_calc.calculate_periods_for_price_info.return_value = {
|
|
"best_price": [],
|
|
"peak_price": [],
|
|
}
|
|
|
|
transformer = TibberPricesDataTransformer(
|
|
config_entry=config_entry,
|
|
log_prefix="[Test]",
|
|
perform_turnover_fn=lambda x: x,
|
|
calculate_periods_fn=mock_period_calc.calculate_periods_for_price_info,
|
|
time=time_service,
|
|
)
|
|
|
|
data = {
|
|
"timestamp": current_time,
|
|
"homes": {
|
|
"home_123": {
|
|
"price_info": {
|
|
"yesterday": [],
|
|
"today": create_price_intervals(0),
|
|
"tomorrow": create_price_intervals(1),
|
|
"currency": "EUR",
|
|
}
|
|
}
|
|
},
|
|
}
|
|
|
|
# First transformation
|
|
result_1 = transformer.transform_data_for_main_entry(data)
|
|
assert result_1 is not None
|
|
|
|
# Second call with SAME config and timestamp should use cache
|
|
result_1_cached = transformer.transform_data_for_main_entry(data)
|
|
assert result_1_cached is result_1 # SAME object
|
|
|
|
# Change config (note: in real system, config change triggers coordinator reload)
|
|
config_entry.options = {
|
|
"price_rating_threshold_low": 80.0, # Changed
|
|
"price_rating_threshold_high": 95.0, # Changed
|
|
}
|
|
|
|
# Call with SAME timestamp but DIFFERENT config
|
|
# Current behavior: Still uses cache (acceptable, see docstring)
|
|
result_2 = transformer.transform_data_for_main_entry(data)
|
|
assert result_2 is result_1 # SAME object (cache preserved)
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_cache_preserved_when_neither_timestamp_nor_config_changed() -> None:
|
|
"""
|
|
Test that cache is PRESERVED when both timestamp and config stay the same.
|
|
|
|
This ensures we're not invalidating cache unnecessarily.
|
|
"""
|
|
config_entry = Mock()
|
|
config_entry.entry_id = "test_entry"
|
|
config_entry.data = {"home_id": "home_123"}
|
|
config_entry.options = {
|
|
"price_rating_threshold_low": 75.0,
|
|
"price_rating_threshold_high": 90.0,
|
|
}
|
|
|
|
time_service = TibberPricesTimeService()
|
|
current_time = datetime(2025, 11, 22, 13, 15, 0, tzinfo=ZoneInfo("Europe/Oslo"))
|
|
|
|
mock_period_calc = Mock()
|
|
mock_period_calc.calculate_periods_for_price_info.return_value = {
|
|
"best_price": [],
|
|
"peak_price": [],
|
|
}
|
|
|
|
transformer = TibberPricesDataTransformer(
|
|
config_entry=config_entry,
|
|
log_prefix="[Test]",
|
|
perform_turnover_fn=lambda x: x,
|
|
calculate_periods_fn=mock_period_calc.calculate_periods_for_price_info,
|
|
time=time_service,
|
|
)
|
|
|
|
data = {
|
|
"timestamp": current_time,
|
|
"homes": {
|
|
"home_123": {
|
|
"price_info": {
|
|
"yesterday": [],
|
|
"today": create_price_intervals(0),
|
|
"tomorrow": create_price_intervals(1),
|
|
"currency": "EUR",
|
|
}
|
|
}
|
|
},
|
|
}
|
|
|
|
# Multiple calls with unchanged data/config should all use cache
|
|
result_1 = transformer.transform_data_for_main_entry(data)
|
|
result_2 = transformer.transform_data_for_main_entry(data)
|
|
result_3 = transformer.transform_data_for_main_entry(data)
|
|
|
|
assert result_1 is result_2 is result_3 # ALL same object (cached)
|
|
|
|
# Verify period calculation was only called ONCE (during first transform)
|
|
assert mock_period_calc.calculate_periods_for_price_info.call_count == 1
|