evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Better handling of source sensor unavailability in Riemman Integration (#93137)

Browse source 
* refactor and increase coverage

* fix log order
This commit is contained in:
Diogo Gomes 2023-05-23 10:19:29 +01:00 • committed by GitHub
parent b754f03eb1
commit e100bcfaea
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 126 additions and 35 deletions
Download patch file Download diff file Expand all files Collapse all files

91
tests/components/integration/test_sensor.py
View file

@ -2,6 +2,8 @@
from datetime import timedelta
from unittest.mock import patch
import pytest
from homeassistant.components.sensor import SensorDeviceClass, SensorStateClass
from homeassistant.const import (
ATTR_UNIT_OF_MEASUREMENT,
@ -20,7 +22,8 @@ import homeassistant.util.dt as dt_util
from tests.common import mock_restore_cache
async def test_state(hass: HomeAssistant) -> None:
@pytest.mark.parametrize("method", ["trapezoidal", "left", "right"])
async def test_state(hass: HomeAssistant, method) -> None:
"""Test integration sensor state."""
config = {
"sensor": {
@ -28,6 +31,7 @@ async def test_state(hass: HomeAssistant) -> None:
"name": "integration",
"source": "sensor.power",
"round": 2,
"method": method,
}
}
@ -46,8 +50,8 @@ async def test_state(hass: HomeAssistant) -> None:
assert state.attributes.get("state_class") is SensorStateClass.TOTAL
assert "device_class" not in state.attributes
future_now = dt_util.utcnow() + timedelta(seconds=3600)
with patch("homeassistant.util.dt.utcnow", return_value=future_now):
now += timedelta(seconds=3600)
with patch("homeassistant.util.dt.utcnow", return_value=now):
hass.states.async_set(
entity_id,
1,
@ -69,6 +73,62 @@ async def test_state(hass: HomeAssistant) -> None:
assert state.attributes.get("device_class") == SensorDeviceClass.ENERGY
assert state.attributes.get("state_class") is SensorStateClass.TOTAL
# 1 hour after last update, power sensor is unavailable
now += timedelta(seconds=3600)
with patch("homeassistant.util.dt.utcnow", return_value=now):
hass.states.async_set(
entity_id,
STATE_UNAVAILABLE,
{
"device_class": SensorDeviceClass.POWER,
ATTR_UNIT_OF_MEASUREMENT: UnitOfPower.KILO_WATT,
},
force_update=True,
)
await hass.async_block_till_done()
state = hass.states.get("sensor.integration")
assert state.state == STATE_UNAVAILABLE
# 1 hour after last update, power sensor is back to normal at 2 KiloWatts and stays for 1 hour += 2kWh
now += timedelta(seconds=3600)
with patch("homeassistant.util.dt.utcnow", return_value=now):
hass.states.async_set(
entity_id,
2,
{
"device_class": SensorDeviceClass.POWER,
ATTR_UNIT_OF_MEASUREMENT: UnitOfPower.KILO_WATT,
},
force_update=True,
)
await hass.async_block_till_done()
state = hass.states.get("sensor.integration")
assert (
round(float(state.state), config["sensor"]["round"]) == 3.0
if method == "right"
else 1.0
)
now += timedelta(seconds=3600)
with patch("homeassistant.util.dt.utcnow", return_value=now):
hass.states.async_set(
entity_id,
2,
{
"device_class": SensorDeviceClass.POWER,
ATTR_UNIT_OF_MEASUREMENT: UnitOfPower.KILO_WATT,
},
force_update=True,
)
await hass.async_block_till_done()
state = hass.states.get("sensor.integration")
assert (
round(float(state.state), config["sensor"]["round"]) == 5.0
if method == "right"
else 3.0
)
async def test_restore_state(hass: HomeAssistant) -> None:
"""Test integration sensor state is restored correctly."""
@ -416,13 +476,15 @@ async def test_units(hass: HomeAssistant) -> None:
assert new_state.state == STATE_UNAVAILABLE
async def test_device_class(hass: HomeAssistant) -> None:
@pytest.mark.parametrize("method", ["trapezoidal", "left", "right"])
async def test_device_class(hass: HomeAssistant, method) -> None:
"""Test integration sensor units using a power source."""
config = {
"sensor": {
"platform": "integration",
"name": "integration",
"source": "sensor.power",
"method": method,
}
}
@ -465,13 +527,15 @@ async def test_device_class(hass: HomeAssistant) -> None:
assert state.attributes.get("device_class") == SensorDeviceClass.ENERGY
async def test_calc_errors(hass: HomeAssistant) -> None:
@pytest.mark.parametrize("method", ["trapezoidal", "left", "right"])
async def test_calc_errors(hass: HomeAssistant, method) -> None:
"""Test integration sensor units using a power source."""
config = {
"sensor": {
"platform": "integration",
"name": "integration",
"source": "sensor.power",
"method": method,
}
}
@ -479,6 +543,7 @@ async def test_calc_errors(hass: HomeAssistant) -> None:
entity_id = config["sensor"]["source"]
now = dt_util.utcnow()
hass.states.async_set(entity_id, None, {})
await hass.async_block_till_done()
@ -489,19 +554,25 @@ async def test_calc_errors(hass: HomeAssistant) -> None:
assert state.state == STATE_UNKNOWN
# Moving from an unknown state to a value is a calc error and should
# not change the value of the Reimann sensor.
hass.states.async_set(entity_id, 0, {"device_class": None})
# not change the value of the Reimann sensor, unless the method used is "right".
now += timedelta(seconds=3600)
with patch("homeassistant.util.dt.utcnow", return_value=now):
hass.states.async_set(entity_id, 0, {"device_class": None})
await hass.async_block_till_done()
await hass.async_block_till_done()
state = hass.states.get("sensor.integration")
assert state is not None
assert state.state == STATE_UNKNOWN
assert state.state == STATE_UNKNOWN if method != "right" else "0.000"
# With the source sensor updated successfully, the Reimann sensor
# should have a zero (known) value.
hass.states.async_set(entity_id, 1, {"device_class": None})
now += timedelta(seconds=3600)
with patch("homeassistant.util.dt.utcnow", return_value=now):
hass.states.async_set(entity_id, 1, {"device_class": None})
await hass.async_block_till_done()
await hass.async_block_till_done()
state = hass.states.get("sensor.integration")
assert state is not None
assert round(float(state.state)) == 0
assert round(float(state.state)) == 0 if method != "right" else 1