Jelle Lauwers
d90562017f
Minute-rounding stopped identical values from re-logging, but finish_time still updates on nearly every poll because now() + remaining is a continuously-drifting value between the device's own remaining-time revisions, and washers/dryers/dishwashers commonly revise that estimate by a minute or two mid-cycle anyway -- both are real, small changes that individually don't matter but each cost a recorder/logbook entry. Add a per-device Options Flow setting (finish_time_debounce_minutes, default 3) and a SensorDesc(debounce=True) opt-in. LocalThingsSensor now caches the last value it actually reported and only adopts a new one once it differs by at least the configured threshold, a cycle starts (no prior cache), or a cycle ends (new value is None) -- 0 disables it entirely, restoring today's behavior. Also pass config_entry explicitly into DataUpdateCoordinator's super().__init__() -- self.config_entry previously relied on an undocumented ContextVar fallback that upstream has flagged as removed in HA 2026.8, which the new debounce lookup needed to not be built on top of.
101 lines
3.4 KiB
Python
101 lines
3.4 KiB
Python
"""Unit tests for LocalThingsSensor's debounce gate (finish_time churn)."""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from custom_components.localthings.const import CONF_FINISH_TIME_DEBOUNCE_MINUTES
|
|
from custom_components.localthings.registry.capabilities.operational import OPERATIONAL_STATE
|
|
from custom_components.localthings.registry.discovery import BoundEntity
|
|
from custom_components.localthings.sensor import LocalThingsSensor
|
|
|
|
_FINISH_TIME_DESC = next(e for e in OPERATIONAL_STATE.entities if e.key == 'finish_time')
|
|
|
|
|
|
class _FakeConfigEntry:
|
|
def __init__(self, options):
|
|
self.options = options
|
|
|
|
|
|
class _FakeCoordinator:
|
|
"""Just enough surface for LocalThingsEntity/LocalThingsSensor."""
|
|
|
|
def __init__(self, threshold_minutes):
|
|
self.device_serial = 'TEST-SERIAL'
|
|
self.config_entry = _FakeConfigEntry({
|
|
CONF_FINISH_TIME_DEBOUNCE_MINUTES: threshold_minutes,
|
|
})
|
|
self.data: dict = {}
|
|
|
|
|
|
def _sensor(threshold_minutes=3):
|
|
coordinator = _FakeCoordinator(threshold_minutes)
|
|
bound = BoundEntity(
|
|
href=OPERATIONAL_STATE.href, capability=OPERATIONAL_STATE, desc=_FINISH_TIME_DESC,
|
|
)
|
|
sensor = LocalThingsSensor(coordinator, bound)
|
|
return sensor, coordinator
|
|
|
|
|
|
def test_small_change_is_suppressed():
|
|
sensor, coordinator = _sensor(threshold_minutes=3)
|
|
base = datetime(2026, 7, 31, 17, 0, tzinfo=timezone.utc)
|
|
|
|
coordinator.data = {'finish_time': base}
|
|
assert sensor.native_value == base
|
|
|
|
coordinator.data = {'finish_time': base + timedelta(minutes=1)}
|
|
assert sensor.native_value == base, "a 1-minute wobble should be held back"
|
|
|
|
|
|
def test_change_past_threshold_is_reported():
|
|
sensor, coordinator = _sensor(threshold_minutes=3)
|
|
base = datetime(2026, 7, 31, 17, 0, tzinfo=timezone.utc)
|
|
|
|
coordinator.data = {'finish_time': base}
|
|
assert sensor.native_value == base
|
|
|
|
new = base + timedelta(minutes=5)
|
|
coordinator.data = {'finish_time': new}
|
|
assert sensor.native_value == new
|
|
|
|
|
|
def test_zero_threshold_disables_debounce():
|
|
sensor, coordinator = _sensor(threshold_minutes=0)
|
|
base = datetime(2026, 7, 31, 17, 0, tzinfo=timezone.utc)
|
|
|
|
coordinator.data = {'finish_time': base}
|
|
assert sensor.native_value == base
|
|
|
|
new = base + timedelta(seconds=1)
|
|
coordinator.data = {'finish_time': new}
|
|
assert sensor.native_value == new
|
|
|
|
|
|
def test_cycle_end_passes_through_immediately():
|
|
"""A cycle ending (finish_time -> None) must never be held back."""
|
|
sensor, coordinator = _sensor(threshold_minutes=3)
|
|
base = datetime(2026, 7, 31, 17, 0, tzinfo=timezone.utc)
|
|
|
|
coordinator.data = {'finish_time': base}
|
|
assert sensor.native_value == base
|
|
|
|
coordinator.data = {'finish_time': None}
|
|
assert sensor.native_value is None
|
|
|
|
|
|
def test_non_debounced_sensor_is_unaffected():
|
|
"""A SensorDesc without debounce=True reads straight through, unchanged."""
|
|
machine_state_desc = next(
|
|
e for e in OPERATIONAL_STATE.entities if e.key == 'machine_state'
|
|
)
|
|
coordinator = _FakeCoordinator(threshold_minutes=3)
|
|
bound = BoundEntity(
|
|
href=OPERATIONAL_STATE.href, capability=OPERATIONAL_STATE, desc=machine_state_desc,
|
|
)
|
|
sensor = LocalThingsSensor(coordinator, bound)
|
|
|
|
coordinator.data = {'machine_state': 'active'}
|
|
assert sensor.native_value == 'active'
|
|
coordinator.data = {'machine_state': 'idle'}
|
|
assert sensor.native_value == 'idle'
|