Fix broken test test_auto_purge in recorder (#44687)

* Fix failing test due to edge-of-2021 bug

* Rewrite test_auto_purge to make the intent more clear
This commit is contained in:
Allen Porter 2020-12-31 14:04:12 -08:00 committed by GitHub
parent fe9a254017
commit f1dff973dc
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23

View File

@ -23,7 +23,7 @@ from homeassistant.util import dt as dt_util
from .common import wait_recording_done
from tests.async_mock import patch
from tests.common import async_fire_time_changed, get_test_home_assistant
from tests.common import fire_time_changed, get_test_home_assistant
def test_saving_state(hass, hass_recorder):
@ -351,8 +351,15 @@ async def test_defaults_set(hass):
assert recorder_config["purge_keep_days"] == 10
def run_tasks_at_time(hass, test_time):
"""Advance the clock and wait for any callbacks to finish."""
fire_time_changed(hass, test_time)
hass.block_till_done()
hass.data[DATA_INSTANCE].block_till_done()
def test_auto_purge(hass_recorder):
"""Test saving and restoring a state."""
"""Test periodic purge alarm scheduling."""
hass = hass_recorder()
original_tz = dt_util.DEFAULT_TIME_ZONE
@ -360,18 +367,40 @@ def test_auto_purge(hass_recorder):
tz = dt_util.get_time_zone("Europe/Copenhagen")
dt_util.set_default_time_zone(tz)
# Purging is schedule to happen at 4:12am every day. Exercise this behavior
# by firing alarms and advancing the clock around this time. Pick an arbitrary
# year in the future to avoid boundary conditions relative to the current date.
#
# The clock is started at 4:15am then advanced forward below
now = dt_util.utcnow()
test_time = tz.localize(datetime(now.year + 1, 1, 1, 4, 12, 0))
async_fire_time_changed(hass, test_time)
test_time = tz.localize(datetime(now.year + 2, 1, 1, 4, 15, 0))
run_tasks_at_time(hass, test_time)
with patch(
"homeassistant.components.recorder.purge.purge_old_data", return_value=True
) as purge_old_data:
for delta in (-1, 0, 1):
async_fire_time_changed(hass, test_time + timedelta(seconds=delta))
hass.block_till_done()
hass.data[DATA_INSTANCE].block_till_done()
# Advance one day, and the purge task should run
test_time = test_time + timedelta(days=1)
run_tasks_at_time(hass, test_time)
assert len(purge_old_data.mock_calls) == 1
purge_old_data.reset_mock()
# Advance one day, and the purge task should run again
test_time = test_time + timedelta(days=1)
run_tasks_at_time(hass, test_time)
assert len(purge_old_data.mock_calls) == 1
purge_old_data.reset_mock()
# Advance less than one full day. The alarm should not yet fire.
test_time = test_time + timedelta(hours=23)
run_tasks_at_time(hass, test_time)
assert len(purge_old_data.mock_calls) == 0
# Advance to the next day and fire the alarm again
test_time = test_time + timedelta(hours=1)
run_tasks_at_time(hass, test_time)
assert len(purge_old_data.mock_calls) == 1
dt_util.set_default_time_zone(original_tz)