1
mirror of https://github.com/home-assistant/core synced 2024-07-27 18:58:57 +02:00

Update derivative tests to avoid patching utcnow (#93477)

This commit is contained in:
J. Nick Koston 2023-05-24 13:54:50 -05:00 committed by GitHub
parent 70c1fe8c6c
commit cff72f9653
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23

View File

@ -2,7 +2,8 @@
from datetime import timedelta
from math import sin
import random
from unittest.mock import patch
from freezegun import freeze_time
from homeassistant.const import UnitOfPower, UnitOfTime
from homeassistant.core import HomeAssistant
@ -26,12 +27,11 @@ async def test_state(hass: HomeAssistant) -> None:
entity_id = config["sensor"]["source"]
base = dt_util.utcnow()
with patch("homeassistant.util.dt.utcnow") as now:
now.return_value = base
with freeze_time(base) as freezer:
hass.states.async_set(entity_id, 1, {})
await hass.async_block_till_done()
now.return_value += timedelta(seconds=3600)
freezer.move_to(dt_util.utcnow() + timedelta(seconds=3600))
hass.states.async_set(entity_id, 1, {}, force_update=True)
await hass.async_block_till_done()
@ -68,9 +68,9 @@ async def setup_tests(hass, config, times, values, expected_state):
# Testing a energy sensor with non-monotonic intervals and values
base = dt_util.utcnow()
with patch("homeassistant.util.dt.utcnow") as now:
with freeze_time(base) as freezer:
for time, value in zip(times, values):
now.return_value = base + timedelta(seconds=time)
freezer.move_to(base + timedelta(seconds=time))
hass.states.async_set(entity_id, value, {}, force_update=True)
await hass.async_block_till_done()
@ -169,18 +169,19 @@ async def test_data_moving_average_for_discrete_sensor(hass: HomeAssistant) -> N
) # two minute window
base = dt_util.utcnow()
for time, value in zip(times, temperature_values):
now = base + timedelta(seconds=time)
with patch("homeassistant.util.dt.utcnow", return_value=now):
with freeze_time(base) as freezer:
for time, value in zip(times, temperature_values):
now = base + timedelta(seconds=time)
freezer.move_to(now)
hass.states.async_set(entity_id, value, {}, force_update=True)
await hass.async_block_till_done()
if time_window < time < times[-1] - time_window:
state = hass.states.get("sensor.power")
derivative = round(float(state.state), config["sensor"]["round"])
# Test that the error is never more than
# (time_window_in_minutes / true_derivative * 100) = 10% + ε
assert abs(1 - derivative) <= 0.1 + 1e-6
if time_window < time < times[-1] - time_window:
state = hass.states.get("sensor.power")
derivative = round(float(state.state), config["sensor"]["round"])
# Test that the error is never more than
# (time_window_in_minutes / true_derivative * 100) = 10% + ε
assert abs(1 - derivative) <= 0.1 + 1e-6
async def test_data_moving_average_for_irregular_times(hass: HomeAssistant) -> None:
@ -212,18 +213,19 @@ async def test_data_moving_average_for_irregular_times(hass: HomeAssistant) -> N
)
base = dt_util.utcnow()
for time, value in zip(times, temperature_values):
now = base + timedelta(seconds=time)
with patch("homeassistant.util.dt.utcnow", return_value=now):
with freeze_time(base) as freezer:
for time, value in zip(times, temperature_values):
now = base + timedelta(seconds=time)
freezer.move_to(now)
hass.states.async_set(entity_id, value, {}, force_update=True)
await hass.async_block_till_done()
if time_window < time and time > times[3]:
state = hass.states.get("sensor.power")
derivative = round(float(state.state), config["sensor"]["round"])
# Test that the error is never more than
# (time_window_in_minutes / true_derivative * 100) = 10% + ε
assert abs(0.1 - derivative) <= 0.01 + 1e-6
if time_window < time and time > times[3]:
state = hass.states.get("sensor.power")
derivative = round(float(state.state), config["sensor"]["round"])
# Test that the error is never more than
# (time_window_in_minutes / true_derivative * 100) = 10% + ε
assert abs(0.1 - derivative) <= 0.01 + 1e-6
async def test_double_signal_after_delay(hass: HomeAssistant) -> None:
@ -253,18 +255,19 @@ async def test_double_signal_after_delay(hass: HomeAssistant) -> None:
base = dt_util.utcnow()
previous = 0
for time, value in zip(times, temperature_values):
now = base + timedelta(seconds=time)
with patch("homeassistant.util.dt.utcnow", return_value=now):
with freeze_time(base) as freezer:
for time, value in zip(times, temperature_values):
now = base + timedelta(seconds=time)
freezer.move_to(now)
hass.states.async_set(entity_id, value, {}, force_update=True)
await hass.async_block_till_done()
state = hass.states.get("sensor.power")
derivative = round(float(state.state), config["sensor"]["round"])
if time == times[-1]:
# Test that the error is never more than
# (time_window_in_minutes / true_derivative * 100) = 10% + ε
assert abs(previous - derivative) <= 0.01 + 1e-6
previous = derivative
state = hass.states.get("sensor.power")
derivative = round(float(state.state), config["sensor"]["round"])
if time == times[-1]:
# Test that the error is never more than
# (time_window_in_minutes / true_derivative * 100) = 10% + ε
assert abs(previous - derivative) <= 0.01 + 1e-6
previous = derivative
async def test_prefix(hass: HomeAssistant) -> None:
@ -283,8 +286,7 @@ async def test_prefix(hass: HomeAssistant) -> None:
entity_id = config["sensor"]["source"]
base = dt_util.utcnow()
with patch("homeassistant.util.dt.utcnow") as now:
now.return_value = base
with freeze_time(base) as freezer:
hass.states.async_set(
entity_id,
1000,
@ -293,7 +295,7 @@ async def test_prefix(hass: HomeAssistant) -> None:
)
await hass.async_block_till_done()
now.return_value += timedelta(seconds=3600)
freezer.move_to(dt_util.utcnow() + timedelta(seconds=3600))
hass.states.async_set(
entity_id,
1000,
@ -327,12 +329,11 @@ async def test_suffix(hass: HomeAssistant) -> None:
entity_id = config["sensor"]["source"]
base = dt_util.utcnow()
with patch("homeassistant.util.dt.utcnow") as now:
now.return_value = base
with freeze_time(base) as freezer:
hass.states.async_set(entity_id, 1000, {})
await hass.async_block_till_done()
now.return_value += timedelta(seconds=10)
freezer.move_to(dt_util.utcnow() + timedelta(seconds=3600))
hass.states.async_set(entity_id, 1000, {}, force_update=True)
await hass.async_block_till_done()