jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-26 22:57:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

Improve derivative sensor tests (#148941)

Browse Source
This commit is contained in:
Erik Montnemery 2025-07-17 11:44:37 +02:00 committed by GitHub
parent 9373bb287c
commit ee35fc495d
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 84 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

117
tests/components/derivative/test_sensor.py
View File

@ -27,8 +27,25 @@ from tests.common import (
mock_restore_cache_with_extra_data, mock_restore_cache_with_extra_data,
) )
A1 = {"attr": "value1"}
A2 = {"attr": "value2"}
async def test_state(hass: HomeAssistant) -> None:
@pytest.mark.parametrize("force_update", [False, True])
@pytest.mark.parametrize(
"attributes",
[
# Same attributes, fires state report
[A1, A1],
# Changing attributes, fires state change with bumped last_updated
[A1, A2],
],
)
async def test_state(
hass: HomeAssistant,
force_update: bool,
attributes: list[dict[str, Any]],
) -> None:
"""Test derivative sensor state.""" """Test derivative sensor state."""
config = { config = {
"sensor": { "sensor": {
@ -45,12 +62,13 @@ async def test_state(hass: HomeAssistant) -> None:
entity_id = config["sensor"]["source"] entity_id = config["sensor"]["source"]
base = dt_util.utcnow() base = dt_util.utcnow()
with freeze_time(base) as freezer: with freeze_time(base) as freezer:
hass.states.async_set(entity_id, 1, {}) for extra_attributes in attributes:
await hass.async_block_till_done() hass.states.async_set(
entity_id, 1, extra_attributes, force_update=force_update
)
await hass.async_block_till_done()
freezer.move_to(dt_util.utcnow() + timedelta(seconds=3600)) freezer.move_to(dt_util.utcnow() + timedelta(seconds=3600))
hass.states.async_set(entity_id, 1, {})
await hass.async_block_till_done()
state = hass.states.get("sensor.derivative") state = hass.states.get("sensor.derivative")
assert state is not None assert state is not None
@ -61,7 +79,24 @@ async def test_state(hass: HomeAssistant) -> None:
assert state.attributes.get("unit_of_measurement") == "kW" assert state.attributes.get("unit_of_measurement") == "kW"
async def test_no_change(hass: HomeAssistant) -> None: # Test unchanged states work both with and without max_sub_interval
@pytest.mark.parametrize("extra_config", [{}, {"max_sub_interval": {"minutes": 9999}}])
@pytest.mark.parametrize("force_update", [False, True])
@pytest.mark.parametrize(
"attributes",
[
# Same attributes, fires state report
[A1, A1, A1, A1],
# Changing attributes, fires state change with bumped last_updated
[A1, A2, A1, A2],
],
)
async def test_no_change(
hass: HomeAssistant,
extra_config: dict[str, Any],
force_update: bool,
attributes: list[dict[str, Any]],
) -> None:
"""Test derivative sensor state updated when source sensor doesn't change.""" """Test derivative sensor state updated when source sensor doesn't change."""
config = { config = {
"sensor": { "sensor": {
@ -71,6 +106,7 @@ async def test_no_change(hass: HomeAssistant) -> None:
"unit": "kW", "unit": "kW",
"round": 2, "round": 2,
} }
| extra_config
} }
assert await async_setup_component(hass, "sensor", config) assert await async_setup_component(hass, "sensor", config)
@ -78,20 +114,13 @@ async def test_no_change(hass: HomeAssistant) -> None:
entity_id = config["sensor"]["source"] entity_id = config["sensor"]["source"]
base = dt_util.utcnow() base = dt_util.utcnow()
with freeze_time(base) as freezer: with freeze_time(base) as freezer:
hass.states.async_set(entity_id, 0, {}) for value, extra_attributes in zip([0, 1, 1, 1], attributes, strict=True):
await hass.async_block_till_done() hass.states.async_set(
entity_id, value, extra_attributes, force_update=force_update
)
await hass.async_block_till_done()
freezer.move_to(dt_util.utcnow() + timedelta(seconds=3600)) freezer.move_to(dt_util.utcnow() + timedelta(seconds=3600))
hass.states.async_set(entity_id, 1, {})
await hass.async_block_till_done()
freezer.move_to(dt_util.utcnow() + timedelta(seconds=3600))
hass.states.async_set(entity_id, 1, {})
await hass.async_block_till_done()
freezer.move_to(dt_util.utcnow() + timedelta(seconds=3600))
hass.states.async_set(entity_id, 1, {})
await hass.async_block_till_done()
state = hass.states.get("sensor.derivative") state = hass.states.get("sensor.derivative")
assert state is not None assert state is not None
@ -138,7 +167,7 @@ async def setup_tests(
# Testing a energy sensor with non-monotonic intervals and values # Testing a energy sensor with non-monotonic intervals and values
base = dt_util.utcnow() base = dt_util.utcnow()
with freeze_time(base) as freezer: with freeze_time(base) as freezer:
for time, value in zip(times, values, strict=False): for time, value in zip(times, values, strict=True):
freezer.move_to(base + timedelta(seconds=time)) freezer.move_to(base + timedelta(seconds=time))
hass.states.async_set(entity_id, value, {}) hass.states.async_set(entity_id, value, {})
await hass.async_block_till_done() await hass.async_block_till_done()
@ -213,7 +242,24 @@ async def test_dataSet6(hass: HomeAssistant) -> None:
await setup_tests(hass, {}, times=[0, 60], values=[0, 1 / 60], expected_state=1) await setup_tests(hass, {}, times=[0, 60], values=[0, 1 / 60], expected_state=1)
async def test_data_moving_average_with_zeroes(hass: HomeAssistant) -> None: # Test unchanged states work both with and without max_sub_interval
@pytest.mark.parametrize("extra_config", [{}, {"max_sub_interval": {"minutes": 9999}}])
@pytest.mark.parametrize("force_update", [False, True])
@pytest.mark.parametrize(
"attributes",
[
# Same attributes, fires state report
[A1, A1] * 10 + [A1],
# Changing attributes, fires state change with bumped last_updated
[A1, A2] * 10 + [A1],
],
)
async def test_data_moving_average_with_zeroes(
hass: HomeAssistant,
extra_config: dict[str, Any],
force_update: bool,
attributes: list[dict[str, Any]],
) -> None:
"""Test that zeroes are properly handled within the time window.""" """Test that zeroes are properly handled within the time window."""
# We simulate the following situation: # We simulate the following situation:
# The temperature rises 1 °C per minute for 10 minutes long. Then, it # The temperature rises 1 °C per minute for 10 minutes long. Then, it
@ -235,16 +281,21 @@ async def test_data_moving_average_with_zeroes(hass: HomeAssistant) -> None:
"time_window": {"seconds": time_window}, "time_window": {"seconds": time_window},
"unit_time": UnitOfTime.MINUTES, "unit_time": UnitOfTime.MINUTES,
"round": 1, "round": 1,
}, }
| extra_config,
) )
base = dt_util.utcnow() base = dt_util.utcnow()
with freeze_time(base) as freezer: with freeze_time(base) as freezer:
last_derivative = 0 last_derivative = 0
for time, value in zip(times, temperature_values, strict=True): for time, value, extra_attributes in zip(
times, temperature_values, attributes, strict=True
):
now = base + timedelta(seconds=time) now = base + timedelta(seconds=time)
freezer.move_to(now) freezer.move_to(now)
hass.states.async_set(entity_id, value, {}) hass.states.async_set(
entity_id, value, extra_attributes, force_update=force_update
)
await hass.async_block_till_done() await hass.async_block_till_done()
state = hass.states.get("sensor.power") state = hass.states.get("sensor.power")
@ -273,7 +324,7 @@ async def test_data_moving_average_for_discrete_sensor(hass: HomeAssistant) -> N
for temperature in range(30): for temperature in range(30):
temperature_values += [temperature] * 2 # two values per minute temperature_values += [temperature] * 2 # two values per minute
time_window = 600 time_window = 600
times = list(range(0, 1800 + 30, 30)) times = list(range(0, 1800, 30))
config, entity_id = await _setup_sensor( config, entity_id = await _setup_sensor(
hass, hass,
@ -286,7 +337,7 @@ async def test_data_moving_average_for_discrete_sensor(hass: HomeAssistant) -> N
base = dt_util.utcnow() base = dt_util.utcnow()
with freeze_time(base) as freezer: with freeze_time(base) as freezer:
for time, value in zip(times, temperature_values, strict=False): for time, value in zip(times, temperature_values, strict=True):
now = base + timedelta(seconds=time) now = base + timedelta(seconds=time)
freezer.move_to(now) freezer.move_to(now)
hass.states.async_set(entity_id, value, {}) hass.states.async_set(entity_id, value, {})
@ -330,7 +381,7 @@ async def test_data_moving_average_for_irregular_times(hass: HomeAssistant) -> N
base = dt_util.utcnow() base = dt_util.utcnow()
with freeze_time(base) as freezer: with freeze_time(base) as freezer:
for time, value in zip(times, temperature_values, strict=False): for time, value in zip(times, temperature_values, strict=True):
now = base + timedelta(seconds=time) now = base + timedelta(seconds=time)
freezer.move_to(now) freezer.move_to(now)
hass.states.async_set(entity_id, value, {}) hass.states.async_set(entity_id, value, {})
@ -368,7 +419,7 @@ async def test_double_signal_after_delay(hass: HomeAssistant) -> None:
base = dt_util.utcnow() base = dt_util.utcnow()
previous = 0 previous = 0
with freeze_time(base) as freezer: with freeze_time(base) as freezer:
for time, value in zip(times, temperature_values, strict=False): for time, value in zip(times, temperature_values, strict=True):
now = base + timedelta(seconds=time) now = base + timedelta(seconds=time)
freezer.move_to(now) freezer.move_to(now)
hass.states.async_set(entity_id, value, {}) hass.states.async_set(entity_id, value, {})
@ -506,7 +557,7 @@ async def test_sub_intervals_with_time_window(hass: HomeAssistant) -> None:
base = dt_util.utcnow() base = dt_util.utcnow()
with freeze_time(base) as freezer: with freeze_time(base) as freezer:
last_state_change = None last_state_change = None
for time, value in zip(times, values, strict=False): for time, value in zip(times, values, strict=True):
now = base + timedelta(seconds=time) now = base + timedelta(seconds=time)
freezer.move_to(now) freezer.move_to(now)
hass.states.async_set(entity_id, value, {}, force_update=True) hass.states.async_set(entity_id, value, {}, force_update=True)
@ -636,7 +687,7 @@ async def test_total_increasing_reset(hass: HomeAssistant) -> None:
actual_times = [] actual_times = []
actual_values = [] actual_values = []
with freeze_time(base_time) as freezer: with freeze_time(base_time) as freezer:
for time, value in zip(times, values, strict=False): for time, value in zip(times, values, strict=True):
current_time = base_time + timedelta(seconds=time) current_time = base_time + timedelta(seconds=time)
freezer.move_to(current_time) freezer.move_to(current_time)
hass.states.async_set( hass.states.async_set(
@ -724,7 +775,7 @@ async def test_unavailable(
# Testing a energy sensor with non-monotonic intervals and values # Testing a energy sensor with non-monotonic intervals and values
base = dt_util.utcnow() base = dt_util.utcnow()
with freeze_time(base) as freezer: with freeze_time(base) as freezer:
for time, value, expect in zip(times, values, expected_state, strict=False): for time, value, expect in zip(times, values, expected_state, strict=True):
freezer.move_to(base + timedelta(seconds=time)) freezer.move_to(base + timedelta(seconds=time))
hass.states.async_set(entity_id, value, {}) hass.states.async_set(entity_id, value, {})
await hass.async_block_till_done() await hass.async_block_till_done()
@ -759,7 +810,7 @@ async def test_unavailable_2(
base = dt_util.utcnow() base = dt_util.utcnow()
with freeze_time(base) as freezer: with freeze_time(base) as freezer:
for time, value in zip(times, values, strict=False): for time, value in zip(times, values, strict=True):
freezer.move_to(base + timedelta(seconds=time)) freezer.move_to(base + timedelta(seconds=time))
hass.states.async_set(entity_id, value, {}) hass.states.async_set(entity_id, value, {})
await hass.async_block_till_done() await hass.async_block_till_done()