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

Improve statistics tests (#148937)

Browse Source
This commit is contained in:
Erik Montnemery 2025-07-17 11:42:25 +02:00 committed by GitHub
parent 0e6a1e3242
commit d72fb021c1
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 65 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

99
tests/components/statistics/test_sensor.py
View File

@ -54,6 +54,9 @@ VALUES_BINARY = ["on", "off", "on", "off", "on", "off", "on", "off", "on"]
VALUES_NUMERIC = [17, 20, 15.2, 5, 3.8, 9.2, 6.7, 14, 6]
VALUES_NUMERIC_LINEAR = [1, 2, 3, 4, 5, 6, 7, 8, 9]
A1 = {"attr": "value1"}
A2 = {"attr": "value2"}
async def test_unique_id(
hass: HomeAssistant, entity_registry: er.EntityRegistry
@ -249,7 +252,22 @@ async def test_sensor_defaults_binary(hass: HomeAssistant) -> None:
assert "age_coverage_ratio" not in state.attributes
async def test_sensor_state_reported(hass: HomeAssistant) -> None:
@pytest.mark.parametrize("force_update", [True, False])
@pytest.mark.parametrize(
("values", "attributes"),
[
# Fires last reported events
([18, 1, 1, 1, 1, 1, 1, 1, 9], [A1, A1, A1, A1, A1, A1, A1, A1, A1]),
# Fires state change events
([18, 1, 1, 1, 1, 1, 1, 1, 9], [A1, A2, A1, A2, A1, A2, A1, A2, A1]),
],
)
async def test_sensor_state_updated_reported(
hass: HomeAssistant,
values: list[float],
attributes: list[dict[str, Any]],
force_update: bool,
) -> None:
"""Test the behavior of the sensor with a sequence of identical values.
Forced updates no longer make a difference, since the statistics are now reacting not
@ -258,7 +276,6 @@ async def test_sensor_state_reported(hass: HomeAssistant) -> None:
This fixes problems with time based averages and some other functions that behave
differently when repeating values are reported.
"""
repeating_values = [18, 0, 0, 0, 0, 0, 0, 0, 9]
assert await async_setup_component(
hass,
"sensor",
@ -267,14 +284,7 @@ async def test_sensor_state_reported(hass: HomeAssistant) -> None:
{
"platform": "statistics",
"name": "test_normal",
"entity_id": "sensor.test_monitored_normal",
"state_characteristic": "mean",
"sampling_size": 20,
},
{
"platform": "statistics",
"name": "test_force",
"entity_id": "sensor.test_monitored_force",
"entity_id": "sensor.test_monitored",
"state_characteristic": "mean",
"sampling_size": 20,
},
@ -283,27 +293,19 @@ async def test_sensor_state_reported(hass: HomeAssistant) -> None:
)
await hass.async_block_till_done()
for value in repeating_values:
for value, attribute in zip(values, attributes, strict=True):
hass.states.async_set(
"sensor.test_monitored_normal",
"sensor.test_monitored",
str(value),
{ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.CELSIUS},
)
hass.states.async_set(
"sensor.test_monitored_force",
str(value),
{ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.CELSIUS},
force_update=True,
{ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.CELSIUS} | attribute,
force_update=force_update,
)
await hass.async_block_till_done()
state_normal = hass.states.get("sensor.test_normal")
state_force = hass.states.get("sensor.test_force")
assert state_normal and state_force
assert state_normal.state == str(round(sum(repeating_values) / 9, 2))
assert state_force.state == str(round(sum(repeating_values) / 9, 2))
assert state_normal.attributes.get("buffer_usage_ratio") == round(9 / 20, 2)
assert state_force.attributes.get("buffer_usage_ratio") == round(9 / 20, 2)
state = hass.states.get("sensor.test_normal")
assert state
assert state.state == str(round(sum(values) / 9, 2))
assert state.attributes.get("buffer_usage_ratio") == round(9 / 20, 2)
async def test_sampling_boundaries_given(hass: HomeAssistant) -> None:
@ -1785,12 +1787,40 @@ async def test_update_before_load(recorder_mock: Recorder, hass: HomeAssistant)
assert float(hass.states.get("sensor.test").state) == pytest.approx(4.5)
async def test_average_linear_unevenly_timed(hass: HomeAssistant) -> None:
@pytest.mark.parametrize("force_update", [True, False])
@pytest.mark.parametrize(
("values_attributes_and_times", "expected_state"),
[
(
# Fires last reported events
[(5.0, A1, 2), (10.0, A1, 1), (10.0, A1, 1), (10.0, A1, 2), (5.0, A1, 1)],
"8.33",
),
( # Fires state change events
[(5.0, A1, 2), (10.0, A2, 1), (10.0, A1, 1), (10.0, A2, 2), (5.0, A1, 1)],
"8.33",
),
(
# Fires last reported events
[(10.0, A1, 2), (10.0, A1, 1), (10.0, A1, 1), (10.0, A1, 2), (10.0, A1, 1)],
"10.0",
),
( # Fires state change events
[(10.0, A1, 2), (10.0, A2, 1), (10.0, A1, 1), (10.0, A2, 2), (10.0, A1, 1)],
"10.0",
),
],
)
async def test_average_linear_unevenly_timed(
hass: HomeAssistant,
force_update: bool,
values_attributes_and_times: list[tuple[float, dict[str, Any], float]],
expected_state: str,
) -> None:
"""Test the average_linear state characteristic with unevenly distributed values.
This also implicitly tests the correct timing of repeating values.
"""
values_and_times = [[5.0, 2], [10.0, 1], [10.0, 1], [10.0, 2], [5.0, 1]]
current_time = dt_util.utcnow()
@ -1814,22 +1844,23 @@ async def test_average_linear_unevenly_timed(hass: HomeAssistant) -> None:
)
await hass.async_block_till_done()
for value_and_time in values_and_times:
for value, extra_attributes, time in values_attributes_and_times:
hass.states.async_set(
"sensor.test_monitored",
str(value_and_time[0]),
{ATTR_UNIT_OF_MEASUREMENT: DEGREE},
str(value),
{ATTR_UNIT_OF_MEASUREMENT: DEGREE} | extra_attributes,
force_update=force_update,
)
current_time += timedelta(seconds=value_and_time[1])
current_time += timedelta(seconds=time)
freezer.move_to(current_time)
await hass.async_block_till_done()
state = hass.states.get("sensor.test_sensor_average_linear")
assert state is not None
assert state.state == "8.33", (
assert state.state == expected_state, (
"value mismatch for characteristic 'sensor/average_linear' - "
f"assert {state.state} == 8.33"
f"assert {state.state} == {expected_state}"
)