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

Modernize and cleanup trend tests (#105010)

Browse Source 
Co-authored-by: Em <emmanuel.sciara@gmail.com>
This commit is contained in:
Jan-Philipp Benecke 2023-12-06 18:46:35 +01:00 committed by GitHub
parent 32febcda5a
commit 05e122e22b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 223 additions and 392 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

487
tests/components/trend/test_binary_sensor.py
View File

@ -1,398 +1,247 @@
"""The test for the Trend sensor platform.""" """The test for the Trend sensor platform."""
from datetime import timedelta from datetime import timedelta
import logging import logging
from typing import Any
from unittest.mock import patch from unittest.mock import patch
from freezegun.api import FrozenDateTimeFactory
import pytest import pytest
from homeassistant import config as hass_config, setup from homeassistant import config as hass_config, setup
from homeassistant.components.trend.const import DOMAIN from homeassistant.components.trend.const import DOMAIN
from homeassistant.const import SERVICE_RELOAD, STATE_UNKNOWN from homeassistant.const import SERVICE_RELOAD, STATE_OFF, STATE_ON, STATE_UNKNOWN
from homeassistant.core import HomeAssistant, State from homeassistant.core import HomeAssistant, State
import homeassistant.util.dt as dt_util from homeassistant.setup import async_setup_component
from tests.common import ( from tests.common import assert_setup_component, get_fixture_path, mock_restore_cache
assert_setup_component,
get_fixture_path,
get_test_home_assistant,
mock_restore_cache,
)
class TestTrendBinarySensor: async def _setup_component(hass: HomeAssistant, params: dict[str, Any]) -> None:
"""Test the Trend sensor.""" """Set up the trend component."""
assert await async_setup_component(
hass = None hass,
def setup_method(self, method):
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
def teardown_method(self, method):
"""Stop everything that was started."""
self.hass.stop()
def test_up(self):
"""Test up trend."""
assert setup.setup_component(
self.hass,
"binary_sensor", "binary_sensor",
{ {
"binary_sensor": { "binary_sensor": {
"platform": "trend", "platform": "trend",
"sensors": { "sensors": {
"test_trend_sensor": {"entity_id": "sensor.test_state"} "test_trend_sensor": params,
}, },
} }
}, },
) )
self.hass.block_till_done() await hass.async_block_till_done()
self.hass.states.set("sensor.test_state", "1")
self.hass.block_till_done()
self.hass.states.set("sensor.test_state", "2")
self.hass.block_till_done()
state = self.hass.states.get("binary_sensor.test_trend_sensor")
assert state.state == "on"
def test_up_using_trendline(self): @pytest.mark.parametrize(
"""Test up trend using multiple samples and trendline calculation.""" ("states", "inverted", "expected_state"),
assert setup.setup_component( [
self.hass, (["1", "2"], False, STATE_ON),
"binary_sensor", (["2", "1"], False, STATE_OFF),
(["1", "2"], True, STATE_OFF),
(["2", "1"], True, STATE_ON),
],
ids=["up", "down", "up inverted", "down inverted"],
)
async def test_basic_trend(
hass: HomeAssistant,
states: list[str],
inverted: bool,
expected_state: str,
):
"""Test trend with a basic setup."""
await _setup_component(
hass,
{
"entity_id": "sensor.test_state",
"invert": inverted,
},
)
for state in states:
hass.states.async_set("sensor.test_state", state)
await hass.async_block_till_done()
assert (sensor_state := hass.states.get("binary_sensor.test_trend_sensor"))
assert sensor_state.state == expected_state
@pytest.mark.parametrize(
("state_series", "inverted", "expected_states"),
[
(
[[10, 0, 20, 30], [100], [0, 30, 1, 0]],
False,
[STATE_UNKNOWN, STATE_ON, STATE_OFF],
),
(
[[10, 0, 20, 30], [100], [0, 30, 1, 0]],
True,
[STATE_UNKNOWN, STATE_OFF, STATE_ON],
),
(
[[30, 20, 30, 10], [5], [30, 0, 45, 60]],
True,
[STATE_UNKNOWN, STATE_ON, STATE_OFF],
),
],
ids=["up", "up inverted", "down"],
)
async def test_using_trendline(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
state_series: list[list[str]],
inverted: bool,
expected_states: list[str],
):
"""Test uptrend using multiple samples and trendline calculation."""
await _setup_component(
hass,
{ {
"binary_sensor": {
"platform": "trend",
"sensors": {
"test_trend_sensor": {
"entity_id": "sensor.test_state", "entity_id": "sensor.test_state",
"sample_duration": 10000, "sample_duration": 10000,
"min_gradient": 1, "min_gradient": 1,
"max_samples": 25, "max_samples": 25,
"min_samples": 5, "min_samples": 5,
} "invert": inverted,
},
}
}, },
) )
self.hass.block_till_done()
now = dt_util.utcnow() for idx, states in enumerate(state_series):
for state in states:
freezer.tick(timedelta(seconds=2))
hass.states.async_set("sensor.test_state", state)
await hass.async_block_till_done()
# add not enough states to trigger calculation assert (sensor_state := hass.states.get("binary_sensor.test_trend_sensor"))
for val in [10, 0, 20, 30]: assert sensor_state.state == expected_states[idx]
with patch("homeassistant.util.dt.utcnow", return_value=now):
self.hass.states.set("sensor.test_state", val)
self.hass.block_till_done()
now += timedelta(seconds=2)
assert (
self.hass.states.get("binary_sensor.test_trend_sensor").state == "unknown" @pytest.mark.parametrize(
("attr_values", "expected_state"),
[
(["1", "2"], STATE_ON),
(["2", "1"], STATE_OFF),
],
ids=["up", "down"],
) )
async def test_attribute_trend(
# add one more state to trigger gradient calculation hass: HomeAssistant,
for val in [100]: attr_values: list[str],
with patch("homeassistant.util.dt.utcnow", return_value=now): expected_state: str,
self.hass.states.set("sensor.test_state", val) ):
self.hass.block_till_done()
now += timedelta(seconds=2)
assert self.hass.states.get("binary_sensor.test_trend_sensor").state == "on"
# add more states to trigger a downtrend
for val in [0, 30, 1, 0]:
with patch("homeassistant.util.dt.utcnow", return_value=now):
self.hass.states.set("sensor.test_state", val)
self.hass.block_till_done()
now += timedelta(seconds=2)
assert self.hass.states.get("binary_sensor.test_trend_sensor").state == "off"
def test_down_using_trendline(self):
"""Test down trend using multiple samples and trendline calculation."""
assert setup.setup_component(
self.hass,
"binary_sensor",
{
"binary_sensor": {
"platform": "trend",
"sensors": {
"test_trend_sensor": {
"entity_id": "sensor.test_state",
"sample_duration": 10000,
"min_gradient": 1,
"max_samples": 25,
"invert": "Yes",
}
},
}
},
)
self.hass.block_till_done()
now = dt_util.utcnow()
for val in [30, 20, 30, 10]:
with patch("homeassistant.util.dt.utcnow", return_value=now):
self.hass.states.set("sensor.test_state", val)
self.hass.block_till_done()
now += timedelta(seconds=2)
state = self.hass.states.get("binary_sensor.test_trend_sensor")
assert state.state == "on"
for val in [30, 0, 45, 50]:
with patch("homeassistant.util.dt.utcnow", return_value=now):
self.hass.states.set("sensor.test_state", val)
self.hass.block_till_done()
now += timedelta(seconds=2)
state = self.hass.states.get("binary_sensor.test_trend_sensor")
assert state.state == "off"
def test_down(self):
"""Test down trend."""
assert setup.setup_component(
self.hass,
"binary_sensor",
{
"binary_sensor": {
"platform": "trend",
"sensors": {
"test_trend_sensor": {"entity_id": "sensor.test_state"}
},
}
},
)
self.hass.block_till_done()
self.hass.states.set("sensor.test_state", "2")
self.hass.block_till_done()
self.hass.states.set("sensor.test_state", "1")
self.hass.block_till_done()
state = self.hass.states.get("binary_sensor.test_trend_sensor")
assert state.state == "off"
def test_invert_up(self):
"""Test up trend with custom message."""
assert setup.setup_component(
self.hass,
"binary_sensor",
{
"binary_sensor": {
"platform": "trend",
"sensors": {
"test_trend_sensor": {
"entity_id": "sensor.test_state",
"invert": "Yes",
}
},
}
},
)
self.hass.block_till_done()
self.hass.states.set("sensor.test_state", "1")
self.hass.block_till_done()
self.hass.states.set("sensor.test_state", "2")
self.hass.block_till_done()
state = self.hass.states.get("binary_sensor.test_trend_sensor")
assert state.state == "off"
def test_invert_down(self):
"""Test down trend with custom message."""
assert setup.setup_component(
self.hass,
"binary_sensor",
{
"binary_sensor": {
"platform": "trend",
"sensors": {
"test_trend_sensor": {
"entity_id": "sensor.test_state",
"invert": "Yes",
}
},
}
},
)
self.hass.block_till_done()
self.hass.states.set("sensor.test_state", "2")
self.hass.block_till_done()
self.hass.states.set("sensor.test_state", "1")
self.hass.block_till_done()
state = self.hass.states.get("binary_sensor.test_trend_sensor")
assert state.state == "on"
def test_attribute_up(self):
"""Test attribute uptrend.""" """Test attribute uptrend."""
assert setup.setup_component( await _setup_component(
self.hass, hass,
"binary_sensor",
{ {
"binary_sensor": {
"platform": "trend",
"sensors": {
"test_trend_sensor": {
"entity_id": "sensor.test_state", "entity_id": "sensor.test_state",
"attribute": "attr", "attribute": "attr",
}
},
}
}, },
) )
self.hass.block_till_done()
self.hass.states.set("sensor.test_state", "State", {"attr": "1"})
self.hass.block_till_done()
self.hass.states.set("sensor.test_state", "State", {"attr": "2"})
self.hass.block_till_done()
state = self.hass.states.get("binary_sensor.test_trend_sensor")
assert state.state == "on"
def test_attribute_down(self): for attr in attr_values:
"""Test attribute down trend.""" hass.states.async_set("sensor.test_state", "State", {"attr": attr})
assert setup.setup_component( await hass.async_block_till_done()
self.hass,
"binary_sensor",
{
"binary_sensor": {
"platform": "trend",
"sensors": {
"test_trend_sensor": {
"entity_id": "sensor.test_state",
"attribute": "attr",
}
},
}
},
)
self.hass.block_till_done()
self.hass.states.set("sensor.test_state", "State", {"attr": "2"}) assert (sensor_state := hass.states.get("binary_sensor.test_trend_sensor"))
self.hass.block_till_done() assert sensor_state.state == expected_state
self.hass.states.set("sensor.test_state", "State", {"attr": "1"})
self.hass.block_till_done()
state = self.hass.states.get("binary_sensor.test_trend_sensor")
assert state.state == "off"
def test_max_samples(self):
async def test_max_samples(hass: HomeAssistant):
"""Test that sample count is limited correctly.""" """Test that sample count is limited correctly."""
assert setup.setup_component( await _setup_component(
self.hass, hass,
"binary_sensor",
{ {
"binary_sensor": {
"platform": "trend",
"sensors": {
"test_trend_sensor": {
"entity_id": "sensor.test_state", "entity_id": "sensor.test_state",
"max_samples": 3, "max_samples": 3,
"min_gradient": -1, "min_gradient": -1,
}
},
}
}, },
) )
self.hass.block_till_done()
for val in [0, 1, 2, 3, 2, 1]: for val in [0, 1, 2, 3, 2, 1]:
self.hass.states.set("sensor.test_state", val) hass.states.async_set("sensor.test_state", val)
self.hass.block_till_done() await hass.async_block_till_done()
state = self.hass.states.get("binary_sensor.test_trend_sensor") assert (state := hass.states.get("binary_sensor.test_trend_sensor"))
assert state.state == "on" assert state.state == "on"
assert state.attributes["sample_count"] == 3 assert state.attributes["sample_count"] == 3
def test_non_numeric(self):
"""Test up trend."""
assert setup.setup_component(
self.hass,
"binary_sensor",
{
"binary_sensor": {
"platform": "trend",
"sensors": {
"test_trend_sensor": {"entity_id": "sensor.test_state"}
},
}
},
)
self.hass.block_till_done()
self.hass.states.set("sensor.test_state", "Non") async def test_non_numeric(hass: HomeAssistant):
self.hass.block_till_done() """Test for non-numeric sensor."""
self.hass.states.set("sensor.test_state", "Numeric") await _setup_component(hass, {"entity_id": "sensor.test_state"})
self.hass.block_till_done()
state = self.hass.states.get("binary_sensor.test_trend_sensor") hass.states.async_set("sensor.test_state", "Non")
await hass.async_block_till_done()
hass.states.async_set("sensor.test_state", "Numeric")
await hass.async_block_till_done()
assert (state := hass.states.get("binary_sensor.test_trend_sensor"))
assert state.state == STATE_UNKNOWN assert state.state == STATE_UNKNOWN
def test_missing_attribute(self):
"""Test attribute down trend.""" async def test_missing_attribute(hass: HomeAssistant):
assert setup.setup_component( """Test for missing attribute."""
self.hass, await _setup_component(
"binary_sensor", hass,
{ {
"binary_sensor": {
"platform": "trend",
"sensors": {
"test_trend_sensor": {
"entity_id": "sensor.test_state", "entity_id": "sensor.test_state",
"attribute": "missing", "attribute": "missing",
}
},
}
}, },
) )
self.hass.block_till_done()
self.hass.states.set("sensor.test_state", "State", {"attr": "2"}) hass.states.async_set("sensor.test_state", "State", {"attr": "2"})
self.hass.block_till_done() await hass.async_block_till_done()
self.hass.states.set("sensor.test_state", "State", {"attr": "1"}) hass.states.async_set("sensor.test_state", "State", {"attr": "1"})
self.hass.block_till_done() await hass.async_block_till_done()
state = self.hass.states.get("binary_sensor.test_trend_sensor")
assert (state := hass.states.get("binary_sensor.test_trend_sensor"))
assert state.state == STATE_UNKNOWN assert state.state == STATE_UNKNOWN
def test_invalid_name_does_not_create(self):
"""Test invalid name.""" async def test_invalid_name_does_not_create(hass: HomeAssistant):
"""Test for invalid name."""
with assert_setup_component(0): with assert_setup_component(0):
assert setup.setup_component( assert await setup.async_setup_component(
self.hass, hass,
"binary_sensor", "binary_sensor",
{ {
"binary_sensor": { "binary_sensor": {
"platform": "template", "platform": "trend",
"sensors": { "sensors": {
"test INVALID sensor": {"entity_id": "sensor.test_state"} "test INVALID sensor": {"entity_id": "sensor.test_state"}
}, },
} }
}, },
) )
assert self.hass.states.all("binary_sensor") == [] assert hass.states.async_all("binary_sensor") == []
def test_invalid_sensor_does_not_create(self):
async def test_invalid_sensor_does_not_create(hass: HomeAssistant):
"""Test invalid sensor.""" """Test invalid sensor."""
with assert_setup_component(0): with assert_setup_component(0):
assert setup.setup_component( assert await setup.async_setup_component(
self.hass, hass,
"binary_sensor", "binary_sensor",
{ {
"binary_sensor": { "binary_sensor": {
"platform": "template", "platform": "trend",
"sensors": { "sensors": {
"test_trend_sensor": {"not_entity_id": "sensor.test_state"} "test_trend_sensor": {"not_entity_id": "sensor.test_state"}
}, },
} }
}, },
) )
assert self.hass.states.all("binary_sensor") == [] assert hass.states.async_all("binary_sensor") == []
def test_no_sensors_does_not_create(self):
async def test_no_sensors_does_not_create(hass: HomeAssistant):
"""Test no sensors.""" """Test no sensors."""
with assert_setup_component(0): with assert_setup_component(0):
assert setup.setup_component( assert await setup.async_setup_component(
self.hass, "binary_sensor", {"binary_sensor": {"platform": "trend"}} hass, "binary_sensor", {"binary_sensor": {"platform": "trend"}}
) )
assert self.hass.states.all("binary_sensor") == [] assert hass.states.async_all("binary_sensor") == []
async def test_reload(hass: HomeAssistant) -> None: async def test_reload(hass: HomeAssistant) -> None:
@ -436,79 +285,61 @@ async def test_reload(hass: HomeAssistant) -> None:
[("on", "on"), ("off", "off"), ("unknown", "unknown")], [("on", "on"), ("off", "off"), ("unknown", "unknown")],
) )
async def test_restore_state( async def test_restore_state(
hass: HomeAssistant, saved_state: str, restored_state: str hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
saved_state: str,
restored_state: str,
) -> None: ) -> None:
"""Test we restore the trend state.""" """Test we restore the trend state."""
mock_restore_cache(hass, (State("binary_sensor.test_trend_sensor", saved_state),)) mock_restore_cache(hass, (State("binary_sensor.test_trend_sensor", saved_state),))
assert await setup.async_setup_component( await _setup_component(
hass, hass,
"binary_sensor",
{ {
"binary_sensor": {
"platform": "trend",
"sensors": {
"test_trend_sensor": {
"entity_id": "sensor.test_state", "entity_id": "sensor.test_state",
"sample_duration": 10000, "sample_duration": 10000,
"min_gradient": 1, "min_gradient": 1,
"max_samples": 25, "max_samples": 25,
"min_samples": 5, "min_samples": 5,
}
},
}
}, },
) )
await hass.async_block_till_done()
# restored sensor should match saved one # restored sensor should match saved one
assert hass.states.get("binary_sensor.test_trend_sensor").state == restored_state assert hass.states.get("binary_sensor.test_trend_sensor").state == restored_state
now = dt_util.utcnow()
# add not enough samples to trigger calculation # add not enough samples to trigger calculation
for val in [10, 20, 30, 40]: for val in [10, 20, 30, 40]:
with patch("homeassistant.util.dt.utcnow", return_value=now): freezer.tick(timedelta(seconds=2))
hass.states.async_set("sensor.test_state", val) hass.states.async_set("sensor.test_state", val)
await hass.async_block_till_done() await hass.async_block_till_done()
now += timedelta(seconds=2)
# state should match restored state as no calculation happened # state should match restored state as no calculation happened
assert hass.states.get("binary_sensor.test_trend_sensor").state == restored_state assert hass.states.get("binary_sensor.test_trend_sensor").state == restored_state
# add more samples to trigger calculation # add more samples to trigger calculation
for val in [50, 60, 70, 80]: for val in [50, 60, 70, 80]:
with patch("homeassistant.util.dt.utcnow", return_value=now): freezer.tick(timedelta(seconds=2))
hass.states.async_set("sensor.test_state", val) hass.states.async_set("sensor.test_state", val)
await hass.async_block_till_done() await hass.async_block_till_done()
now += timedelta(seconds=2)
# sensor should detect an upwards trend and turn on # sensor should detect an upwards trend and turn on
assert hass.states.get("binary_sensor.test_trend_sensor").state == "on" assert hass.states.get("binary_sensor.test_trend_sensor").state == "on"
async def test_invalid_min_sample( async def test_invalid_min_sample(
hass: HomeAssistant, caplog: pytest.LogCaptureFixture hass: HomeAssistant,
caplog: pytest.LogCaptureFixture,
) -> None: ) -> None:
"""Test if error is logged when min_sample is larger than max_samples.""" """Test if error is logged when min_sample is larger than max_samples."""
with caplog.at_level(logging.ERROR): with caplog.at_level(logging.ERROR):
assert await setup.async_setup_component( await _setup_component(
hass, hass,
"binary_sensor",
{ {
"binary_sensor": {
"platform": "trend",
"sensors": {
"test_trend_sensor": {
"entity_id": "sensor.test_state", "entity_id": "sensor.test_state",
"max_samples": 25, "max_samples": 25,
"min_samples": 30, "min_samples": 30,
}
},
}
}, },
) )
await hass.async_block_till_done()
record = caplog.records[0] record = caplog.records[0]
assert record.levelname == "ERROR" assert record.levelname == "ERROR"