mirror of
https://github.com/jpawlowski/hass.tibber_prices.git
synced 2026-03-29 21:03:40 +00:00
85fe9666a7
Introduced TibberPricesMidnightHandler to prevent duplicate midnight turnover when multiple timers fire simultaneously. Problem: Timer #1 (API poll) and Timer #2 (quarter-hour refresh) both wake at midnight, each detecting day change and triggering cache clear. Race condition caused duplicate turnover operations. Solution: - Atomic flag coordination: First timer sets flag, subsequent timers skip - Persistent state survives HA restart (cache stores last_turnover_time) - Day-boundary detection: Compares current.date() vs last_check.date() - 13 comprehensive tests covering race conditions and HA restart scenarios Architecture: - coordinator/midnight_handler.py: 165 lines, atomic coordination logic - coordinator/core.py: Integrated handler in coordinator initialization - coordinator/listeners.py: Delegate midnight check to handler Impact: Eliminates duplicate cache clears at midnight. Single atomic turnover operation regardless of how many timers fire simultaneously.
322 lines
12 KiB
Python
322 lines
12 KiB
Python
"""
|
|
Unit tests for midnight turnover handler.
|
|
|
|
These tests verify the atomic coordination logic that prevents duplicate
|
|
midnight turnover between multiple timers.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from zoneinfo import ZoneInfo
|
|
|
|
import pytest
|
|
|
|
from custom_components.tibber_prices.coordinator.midnight_handler import (
|
|
TibberPricesMidnightHandler,
|
|
)
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_first_check_initializes_without_turnover() -> None:
|
|
"""Test that the first check initializes but doesn't trigger turnover."""
|
|
handler = TibberPricesMidnightHandler()
|
|
|
|
time1 = datetime(2025, 11, 22, 14, 30, 0, tzinfo=ZoneInfo("Europe/Oslo"))
|
|
|
|
# First check should return False (no turnover yet)
|
|
assert not handler.is_turnover_needed(time1)
|
|
|
|
# But update_check_time should initialize
|
|
handler.update_check_time(time1)
|
|
assert handler.last_check_time == time1
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_midnight_crossing_triggers_turnover() -> None:
|
|
"""Test that crossing midnight triggers turnover detection."""
|
|
handler = TibberPricesMidnightHandler()
|
|
|
|
# Initialize at 23:59:59 on Nov 22
|
|
time1 = datetime(2025, 11, 22, 23, 59, 59, tzinfo=ZoneInfo("Europe/Oslo"))
|
|
handler.update_check_time(time1)
|
|
|
|
# Check at 00:00:00 on Nov 23 (midnight crossed!)
|
|
time2 = datetime(2025, 11, 23, 0, 0, 0, tzinfo=ZoneInfo("Europe/Oslo"))
|
|
assert handler.is_turnover_needed(time2)
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_same_day_no_turnover() -> None:
|
|
"""Test that multiple checks on the same day don't trigger turnover."""
|
|
handler = TibberPricesMidnightHandler()
|
|
|
|
# Initialize at 10:00
|
|
time1 = datetime(2025, 11, 22, 10, 0, 0, tzinfo=ZoneInfo("Europe/Oslo"))
|
|
handler.update_check_time(time1)
|
|
|
|
# Check later same day at 14:00
|
|
time2 = datetime(2025, 11, 22, 14, 0, 0, tzinfo=ZoneInfo("Europe/Oslo"))
|
|
assert not handler.is_turnover_needed(time2)
|
|
|
|
# Check even later at 23:59
|
|
time3 = datetime(2025, 11, 22, 23, 59, 59, tzinfo=ZoneInfo("Europe/Oslo"))
|
|
assert not handler.is_turnover_needed(time3)
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_atomic_coordination_prevents_duplicate_turnover() -> None:
|
|
"""Test that marking turnover done prevents duplicate execution."""
|
|
handler = TibberPricesMidnightHandler()
|
|
|
|
# Initialize on Nov 22
|
|
time1 = datetime(2025, 11, 22, 23, 50, 0, tzinfo=ZoneInfo("Europe/Oslo"))
|
|
handler.update_check_time(time1)
|
|
|
|
# Midnight on Nov 23 - first timer detects it
|
|
midnight = datetime(2025, 11, 23, 0, 0, 0, tzinfo=ZoneInfo("Europe/Oslo"))
|
|
assert handler.is_turnover_needed(midnight)
|
|
|
|
# First timer marks it done
|
|
handler.mark_turnover_done(midnight)
|
|
|
|
# Second timer checks shortly after - should return False
|
|
time2 = datetime(2025, 11, 23, 0, 0, 10, tzinfo=ZoneInfo("Europe/Oslo"))
|
|
assert not handler.is_turnover_needed(time2)
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_mark_turnover_updates_both_timestamps() -> None:
|
|
"""Test that mark_turnover_done updates both check and turnover timestamps."""
|
|
handler = TibberPricesMidnightHandler()
|
|
|
|
time1 = datetime(2025, 11, 22, 10, 0, 0, tzinfo=ZoneInfo("Europe/Oslo"))
|
|
handler.update_check_time(time1)
|
|
|
|
midnight = datetime(2025, 11, 23, 0, 0, 0, tzinfo=ZoneInfo("Europe/Oslo"))
|
|
handler.mark_turnover_done(midnight)
|
|
|
|
# Both timestamps should be updated
|
|
assert handler.last_check_time == midnight
|
|
assert handler.last_turnover_time == midnight
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_next_day_triggers_new_turnover() -> None:
|
|
"""Test that the next day's midnight triggers turnover again."""
|
|
handler = TibberPricesMidnightHandler()
|
|
|
|
# Day 1: Initialize and mark turnover done
|
|
day1 = datetime(2025, 11, 22, 10, 0, 0, tzinfo=ZoneInfo("Europe/Oslo"))
|
|
handler.update_check_time(day1)
|
|
|
|
midnight1 = datetime(2025, 11, 23, 0, 0, 0, tzinfo=ZoneInfo("Europe/Oslo"))
|
|
assert handler.is_turnover_needed(midnight1)
|
|
handler.mark_turnover_done(midnight1)
|
|
|
|
# Day 2: Next midnight should trigger again
|
|
midnight2 = datetime(2025, 11, 24, 0, 0, 0, tzinfo=ZoneInfo("Europe/Oslo"))
|
|
assert handler.is_turnover_needed(midnight2)
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_multiple_days_skipped_still_triggers() -> None:
|
|
"""Test that skipping multiple days still triggers turnover."""
|
|
handler = TibberPricesMidnightHandler()
|
|
|
|
# Last check on Nov 22
|
|
time1 = datetime(2025, 11, 22, 10, 0, 0, tzinfo=ZoneInfo("Europe/Oslo"))
|
|
handler.update_check_time(time1)
|
|
|
|
# Check 3 days later on Nov 25 (skipped 23rd and 24th)
|
|
time2 = datetime(2025, 11, 25, 14, 0, 0, tzinfo=ZoneInfo("Europe/Oslo"))
|
|
assert handler.is_turnover_needed(time2)
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_update_check_time_without_triggering_turnover() -> None:
|
|
"""Test that update_check_time initializes without turnover side effects."""
|
|
handler = TibberPricesMidnightHandler()
|
|
|
|
time1 = datetime(2025, 11, 22, 10, 0, 0, tzinfo=ZoneInfo("Europe/Oslo"))
|
|
handler.update_check_time(time1)
|
|
|
|
# Last turnover should still be None
|
|
assert handler.last_check_time == time1
|
|
assert handler.last_turnover_time is None
|
|
|
|
# Next day should trigger turnover
|
|
time2 = datetime(2025, 11, 23, 0, 0, 0, tzinfo=ZoneInfo("Europe/Oslo"))
|
|
assert handler.is_turnover_needed(time2)
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_ha_restart_after_midnight_with_cached_turnover() -> None:
|
|
"""
|
|
Test HA restart scenario: cached turnover from yesterday, restart after midnight.
|
|
|
|
Scenario:
|
|
- Nov 21 23:50: Last turnover marked (before HA shutdown)
|
|
- Nov 22 00:30: HA restarts (handler is fresh, but turnover was cached)
|
|
- Expected: Turnover should be triggered to catch up
|
|
|
|
This simulates: mark_turnover_done() was called on Nov 21, handler state is
|
|
restored (simulated by manually setting _last_actual_turnover), then first
|
|
check after restart should detect missed midnight.
|
|
"""
|
|
handler = TibberPricesMidnightHandler()
|
|
|
|
# Simulate: Last turnover was on Nov 21 at 23:59:59 (just before midnight)
|
|
last_turnover = datetime(2025, 11, 21, 23, 59, 59, tzinfo=ZoneInfo("Europe/Oslo"))
|
|
# Manually restore handler state (simulates cache restoration)
|
|
handler._last_actual_turnover = last_turnover # noqa: SLF001 - Test setup
|
|
|
|
# HA restarts at Nov 22 00:30 (after midnight)
|
|
restart_time = datetime(2025, 11, 22, 0, 30, 0, tzinfo=ZoneInfo("Europe/Oslo"))
|
|
|
|
# First check after restart - should detect missed midnight
|
|
# _last_midnight_check is None (fresh handler), but _last_actual_turnover exists
|
|
assert handler.is_turnover_needed(restart_time) is True
|
|
|
|
# Perform turnover
|
|
handler.mark_turnover_done(restart_time)
|
|
|
|
# Second check - should not trigger again
|
|
time_2 = datetime(2025, 11, 22, 1, 0, 0, tzinfo=ZoneInfo("Europe/Oslo"))
|
|
assert handler.is_turnover_needed(time_2) is False
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_ha_restart_same_day_with_cached_turnover() -> None:
|
|
"""
|
|
Test HA restart scenario: cached turnover from today, restart same day.
|
|
|
|
Scenario:
|
|
- Nov 22 00:05: Turnover happened (after HA started)
|
|
- Nov 22 14:00: HA restarts
|
|
- Expected: No turnover needed (already done today)
|
|
|
|
This ensures we don't trigger duplicate turnover when restarting same day.
|
|
"""
|
|
handler = TibberPricesMidnightHandler()
|
|
|
|
# Simulate: Last turnover was today at 00:05
|
|
last_turnover = datetime(2025, 11, 22, 0, 5, 0, tzinfo=ZoneInfo("Europe/Oslo"))
|
|
# Manually restore handler state (simulates cache restoration)
|
|
handler._last_actual_turnover = last_turnover # noqa: SLF001 - Test setup
|
|
|
|
# HA restarts at 14:00 same day
|
|
restart_time = datetime(2025, 11, 22, 14, 0, 0, tzinfo=ZoneInfo("Europe/Oslo"))
|
|
|
|
# First check after restart - should NOT trigger (same day)
|
|
assert handler.is_turnover_needed(restart_time) is False
|
|
|
|
# Initialize check time for subsequent checks
|
|
handler.update_check_time(restart_time)
|
|
|
|
# Later check same day - still no turnover
|
|
time_2 = datetime(2025, 11, 22, 18, 0, 0, tzinfo=ZoneInfo("Europe/Oslo"))
|
|
assert handler.is_turnover_needed(time_2) is False
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_simultaneous_timer_checks_at_midnight() -> None:
|
|
"""
|
|
Test race condition: Timer #1 and Timer #2 both check at exactly 00:00:00.
|
|
|
|
This is the critical atomic coordination test - both timers detect midnight
|
|
simultaneously, but only one should perform turnover.
|
|
|
|
Scenario:
|
|
- Nov 21 23:45: Both timers initialized
|
|
- Nov 22 00:00:00: Both timers check simultaneously
|
|
- Expected: First check returns True, second returns False (atomic)
|
|
"""
|
|
handler = TibberPricesMidnightHandler()
|
|
|
|
# Initialize on Nov 21 at 23:45
|
|
init_time = datetime(2025, 11, 21, 23, 45, 0, tzinfo=ZoneInfo("Europe/Oslo"))
|
|
handler.update_check_time(init_time)
|
|
|
|
# Both timers wake up at exactly 00:00:00
|
|
midnight = datetime(2025, 11, 22, 0, 0, 0, tzinfo=ZoneInfo("Europe/Oslo"))
|
|
|
|
# Timer #1 checks first (or both check "simultaneously" in sequence)
|
|
timer1_check = handler.is_turnover_needed(midnight)
|
|
assert timer1_check is True # Midnight crossed
|
|
|
|
# Timer #1 performs turnover
|
|
handler.mark_turnover_done(midnight)
|
|
|
|
# Timer #2 checks immediately after (could be microseconds later)
|
|
timer2_check = handler.is_turnover_needed(midnight)
|
|
assert timer2_check is False # Already done by Timer #1
|
|
|
|
# Verify state: turnover happened exactly once
|
|
assert handler.last_turnover_time == midnight
|
|
assert handler.last_check_time == midnight
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_timer_check_at_00_00_01_after_turnover_at_00_00_00() -> None:
|
|
"""
|
|
Test edge case: One timer does turnover at 00:00:00, second checks at 00:00:01.
|
|
|
|
This ensures that even a 1-second delay doesn't cause duplicate turnover
|
|
when both checks happen on the same calendar day.
|
|
|
|
Scenario:
|
|
- Nov 22 00:00:00: Timer #1 does turnover
|
|
- Nov 22 00:00:01: Timer #2 checks (1 second later)
|
|
- Expected: Timer #2 should skip (same day)
|
|
"""
|
|
handler = TibberPricesMidnightHandler()
|
|
|
|
# Initialize on Nov 21
|
|
init_time = datetime(2025, 11, 21, 23, 45, 0, tzinfo=ZoneInfo("Europe/Oslo"))
|
|
handler.update_check_time(init_time)
|
|
|
|
# Timer #1 checks at exactly 00:00:00
|
|
midnight_00 = datetime(2025, 11, 22, 0, 0, 0, tzinfo=ZoneInfo("Europe/Oslo"))
|
|
assert handler.is_turnover_needed(midnight_00) is True
|
|
handler.mark_turnover_done(midnight_00)
|
|
|
|
# Timer #2 checks 1 second later
|
|
midnight_01 = datetime(2025, 11, 22, 0, 0, 1, tzinfo=ZoneInfo("Europe/Oslo"))
|
|
assert handler.is_turnover_needed(midnight_01) is False
|
|
|
|
# Both timestamps point to same day - no duplicate
|
|
assert handler.last_turnover_time.date() == midnight_01.date() # type: ignore[union-attr]
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_rapid_consecutive_checks_same_second() -> None:
|
|
"""
|
|
Test rapid consecutive checks within the same second at midnight.
|
|
|
|
Simulates worst-case race condition where both timers fire within
|
|
the same second (e.g., 00:00:00.123 and 00:00:00.456).
|
|
|
|
Expected: First check triggers, all subsequent checks skip.
|
|
"""
|
|
handler = TibberPricesMidnightHandler()
|
|
|
|
# Initialize on Nov 21
|
|
init_time = datetime(2025, 11, 21, 23, 59, 59, tzinfo=ZoneInfo("Europe/Oslo"))
|
|
handler.update_check_time(init_time)
|
|
|
|
# Simulate 3 checks at midnight within the same second
|
|
midnight_check1 = datetime(2025, 11, 22, 0, 0, 0, 123000, tzinfo=ZoneInfo("Europe/Oslo"))
|
|
midnight_check2 = datetime(2025, 11, 22, 0, 0, 0, 456000, tzinfo=ZoneInfo("Europe/Oslo"))
|
|
midnight_check3 = datetime(2025, 11, 22, 0, 0, 0, 789000, tzinfo=ZoneInfo("Europe/Oslo"))
|
|
|
|
# First check: turnover needed
|
|
assert handler.is_turnover_needed(midnight_check1) is True
|
|
handler.mark_turnover_done(midnight_check1)
|
|
|
|
# Second and third checks: already done
|
|
assert handler.is_turnover_needed(midnight_check2) is False
|
|
assert handler.is_turnover_needed(midnight_check3) is False
|
|
|
|
# Verify: turnover happened exactly once
|
|
assert handler.last_turnover_time == midnight_check1
|