From cff72f96534a833c0e06e7020a212518611a32b4 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Wed, 24 May 2023 13:54:50 -0500 Subject: [PATCH] Update derivative tests to avoid patching utcnow (#93477) --- tests/components/derivative/test_sensor.py | 81 +++++++++++----------- 1 file changed, 41 insertions(+), 40 deletions(-) diff --git a/tests/components/derivative/test_sensor.py b/tests/components/derivative/test_sensor.py index c1541812d1b..8260e5a0ada 100644 --- a/tests/components/derivative/test_sensor.py +++ b/tests/components/derivative/test_sensor.py @@ -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()