diff --git a/tests/components/derivative/test_sensor.py b/tests/components/derivative/test_sensor.py index 8260e5a0ad..c1541812d1 100644 --- a/tests/components/derivative/test_sensor.py +++ b/tests/components/derivative/test_sensor.py @@ -2,8 +2,7 @@ from datetime import timedelta from math import sin import random - -from freezegun import freeze_time +from unittest.mock import patch from homeassistant.const import UnitOfPower, UnitOfTime from homeassistant.core import HomeAssistant @@ -27,11 +26,12 @@ async def test_state(hass: HomeAssistant) -> None: entity_id = config["sensor"]["source"] base = dt_util.utcnow() - with freeze_time(base) as freezer: + with patch("homeassistant.util.dt.utcnow") as now: + now.return_value = base hass.states.async_set(entity_id, 1, {}) await hass.async_block_till_done() - freezer.move_to(dt_util.utcnow() + timedelta(seconds=3600)) + now.return_value += 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 freeze_time(base) as freezer: + with patch("homeassistant.util.dt.utcnow") as now: for time, value in zip(times, values): - freezer.move_to(base + timedelta(seconds=time)) + now.return_value = base + timedelta(seconds=time) hass.states.async_set(entity_id, value, {}, force_update=True) await hass.async_block_till_done() @@ -169,19 +169,18 @@ async def test_data_moving_average_for_discrete_sensor(hass: HomeAssistant) -> N ) # two minute window base = dt_util.utcnow() - with freeze_time(base) as freezer: - for time, value in zip(times, temperature_values): - now = base + timedelta(seconds=time) - freezer.move_to(now) + for time, value in zip(times, temperature_values): + now = base + timedelta(seconds=time) + with patch("homeassistant.util.dt.utcnow", return_value=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: @@ -213,19 +212,18 @@ async def test_data_moving_average_for_irregular_times(hass: HomeAssistant) -> N ) base = dt_util.utcnow() - with freeze_time(base) as freezer: - for time, value in zip(times, temperature_values): - now = base + timedelta(seconds=time) - freezer.move_to(now) + for time, value in zip(times, temperature_values): + now = base + timedelta(seconds=time) + with patch("homeassistant.util.dt.utcnow", return_value=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: @@ -255,19 +253,18 @@ async def test_double_signal_after_delay(hass: HomeAssistant) -> None: base = dt_util.utcnow() previous = 0 - with freeze_time(base) as freezer: - for time, value in zip(times, temperature_values): - now = base + timedelta(seconds=time) - freezer.move_to(now) + for time, value in zip(times, temperature_values): + now = base + timedelta(seconds=time) + with patch("homeassistant.util.dt.utcnow", return_value=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: @@ -286,7 +283,8 @@ async def test_prefix(hass: HomeAssistant) -> None: entity_id = config["sensor"]["source"] base = dt_util.utcnow() - with freeze_time(base) as freezer: + with patch("homeassistant.util.dt.utcnow") as now: + now.return_value = base hass.states.async_set( entity_id, 1000, @@ -295,7 +293,7 @@ async def test_prefix(hass: HomeAssistant) -> None: ) await hass.async_block_till_done() - freezer.move_to(dt_util.utcnow() + timedelta(seconds=3600)) + now.return_value += timedelta(seconds=3600) hass.states.async_set( entity_id, 1000, @@ -329,11 +327,12 @@ async def test_suffix(hass: HomeAssistant) -> None: entity_id = config["sensor"]["source"] base = dt_util.utcnow() - with freeze_time(base) as freezer: + with patch("homeassistant.util.dt.utcnow") as now: + now.return_value = base hass.states.async_set(entity_id, 1000, {}) await hass.async_block_till_done() - freezer.move_to(dt_util.utcnow() + timedelta(seconds=3600)) + now.return_value += timedelta(seconds=10) hass.states.async_set(entity_id, 1000, {}, force_update=True) await hass.async_block_till_done()