Files
heating-mpc-ha/tests/test_data_quality.py
T
buzeraandClaude Opus 5.5 4daba5028b Show each sensor's longest report gap; stop recording volatile attributes
Choosing room and weather freshness limits required knowing how long
each sensor normally goes without reporting, which HA does not show.
Add max_age_seconds_24h to every source in input_quality: the largest
report age seen at polls over the last 24 hours (hourly buckets, reset
on reload, underestimates by up to one poll interval).

input_quality and weather_comparison change every poll, so the Recorder
stored a new attribute row every five minutes. Mark them unrecorded;
they remain visible on the Status entity. Model coefficients and other
attributes are still recorded.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-26 21:51:35 +02:00

53 lines
2.6 KiB
Python

"""Timestamp boundaries and UI translation coverage without HA dependencies."""
import importlib.util
import json
import unittest
from datetime import datetime, timedelta, timezone
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1] / 'custom_components' / 'heating_mpc'
spec = importlib.util.spec_from_file_location('quality', ROOT / 'data_quality.py')
quality = importlib.util.module_from_spec(spec)
spec.loader.exec_module(quality)
class FreshnessTests(unittest.TestCase):
def test_age_boundaries(self):
now = datetime(2026, 10, 26, tzinfo=timezone.utc)
for seconds, expected in ((0, 'valid'), (1800, 'valid'), (1801, 'stale'), (-1, 'future_timestamp')):
with self.subTest(seconds=seconds):
result = quality.freshness(now - timedelta(seconds=seconds), now, 1800)
self.assertEqual(result['status'], expected)
self.assertEqual(result['age_seconds'], seconds)
def test_missing_naive_and_malformed_times(self):
now = datetime.now(timezone.utc)
for timestamp in (None, 'bad', datetime.now()):
self.assertEqual(quality.freshness(timestamp, now, 1800)['status'], 'invalid_timestamp')
def test_invalid_limits_rejected(self):
now = datetime.now(timezone.utc)
for limit in (0, -1, float('inf'), float('nan')):
with self.assertRaises(ValueError):
quality.freshness(now, now, limit)
def test_report_age_history_keeps_24_hour_maximum(self):
history = quality.ReportAgeHistory()
start = datetime(2026, 10, 26, tzinfo=timezone.utc)
self.assertEqual(history.observe('sensor.a', start, 2400), 2400)
self.assertEqual(history.observe('sensor.b', start, 60), 60)
for minutes in range(5, 24 * 60, 5):
result = history.observe('sensor.a', start + timedelta(minutes=minutes), 300)
self.assertEqual(result, 2400) # still inside the 24-hour window
later = start + timedelta(hours=24, minutes=5)
self.assertEqual(history.observe('sensor.a', later, 300), 300)
self.assertEqual(history.observe('sensor.b', later, 30), 30)
def test_freshness_settings_translated(self):
for name in ('strings.json', 'translations/en.json', 'translations/ro.json'):
data = json.loads((ROOT / name).read_text())
for section, step in (('config', 'user'), ('options', 'init')):
labels = data[section]['step'][step]['data']
for key in ('room_max_age_minutes', 'weather_max_age_minutes'):
self.assertTrue(labels[key])