Use freezegun in DST tests (#58939)

This commit is contained in:
Erik Montnemery 2021-11-02 18:11:39 +01:00 committed by GitHub
parent 2df1ba2346
commit 30f7bc0f18
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 166 additions and 162 deletions

View file

@ -9,6 +9,7 @@
-r requirements_test_pre_commit.txt
codecov==2.1.12
coverage==6.1.1
freezegun==1.1.0
jsonpickle==1.4.1
mock-open==1.4.0
mypy==0.910
@ -18,6 +19,7 @@ pipdeptree==2.1.0
pylint-strict-informational==0.1
pytest-aiohttp==0.3.0
pytest-cov==2.12.1
pytest-freezegun==0.4.2
pytest-socket==0.4.1
pytest-test-groups==1.0.3
pytest-sugar==0.9.4

View file

@ -371,9 +371,12 @@ fire_mqtt_message = threadsafe_callback_factory(async_fire_mqtt_message)
@ha.callback
def async_fire_time_changed(
hass: HomeAssistant, datetime_: datetime, fire_all: bool = False
hass: HomeAssistant, datetime_: datetime = None, fire_all: bool = False
) -> None:
"""Fire a time changes event."""
"""Fire a time changed event."""
if datetime_ is None:
datetime_ = date_util.utcnow()
hass.bus.async_fire(EVENT_TIME_CHANGED, {"now": date_util.as_utc(datetime_)})
for task in list(hass.loop._scheduled):

View file

@ -9,6 +9,7 @@ import threading
from unittest.mock import MagicMock, patch
from aiohttp.test_utils import make_mocked_request
import freezegun
import multidict
import pytest
import pytest_socket
@ -63,15 +64,24 @@ def pytest_configure(config):
def pytest_runtest_setup():
"""Throw if tests attempt to open sockets.
"""Prepare pytest_socket and freezegun.
pytest_socket:
Throw if tests attempt to open sockets.
allow_unix_socket is set to True because it's needed by asyncio.
Important: socket_allow_hosts must be called before disable_socket, otherwise all
destinations will be allowed.
freezegun:
Modified to include https://github.com/spulec/freezegun/pull/424
"""
pytest_socket.socket_allow_hosts(["127.0.0.1"])
disable_socket(allow_unix_socket=True)
freezegun.api.datetime_to_fakedatetime = ha_datetime_to_fakedatetime
freezegun.api.FakeDatetime = HAFakeDatetime
@pytest.fixture
def socket_disabled(pytestconfig):
@ -126,6 +136,43 @@ def disable_socket(allow_unix_socket=False):
socket.socket = GuardedSocket
def ha_datetime_to_fakedatetime(datetime):
"""Convert datetime to FakeDatetime.
Modified to include https://github.com/spulec/freezegun/pull/424.
"""
return freezegun.api.FakeDatetime(
datetime.year,
datetime.month,
datetime.day,
datetime.hour,
datetime.minute,
datetime.second,
datetime.microsecond,
datetime.tzinfo,
fold=datetime.fold,
)
class HAFakeDatetime(freezegun.api.FakeDatetime):
"""Modified to include https://github.com/spulec/freezegun/pull/424."""
@classmethod
def now(cls, tz=None):
"""Return frozen now."""
now = cls._time_to_freeze() or freezegun.api.real_datetime.now()
if tz:
result = tz.fromutc(now.replace(tzinfo=tz))
else:
result = now
# Add the _tz_offset only if it's non-zero to preserve fold
if cls._tz_offset():
result += cls._tz_offset()
return ha_datetime_to_fakedatetime(result)
def check_real(func):
"""Force a function to require a keyword _test_real to be passed in."""

View file

@ -1,7 +1,7 @@
"""Test event helpers."""
# pylint: disable=protected-access
import asyncio
from datetime import datetime, timedelta
from datetime import date, datetime, timedelta
from unittest.mock import patch
from astral import LocationInfo
@ -3393,66 +3393,56 @@ async def test_periodic_task_duplicate_time(hass):
unsub()
async def test_periodic_task_entering_dst(hass):
# DST starts early morning March 28th 2021
@pytest.mark.freeze_time("2021-03-28 01:28:00+01:00")
async def test_periodic_task_entering_dst(hass, freezer):
"""Test periodic task behavior when entering dst."""
timezone = dt_util.get_time_zone("Europe/Vienna")
dt_util.set_default_time_zone(timezone)
specific_runs = []
# DST starts early morning March 27th 2022
yy = 2022
mm = 3
dd = 27
today = date.today().isoformat()
tomorrow = (date.today() + timedelta(days=1)).isoformat()
# There's no 2022-03-27 02:30, the event should not fire until 2022-03-28 02:30
time_that_will_not_match_right_away = datetime(
yy, mm, dd, 1, 28, 0, tzinfo=timezone, fold=0
)
# Make sure we enter DST during the test
assert (
time_that_will_not_match_right_away.utcoffset()
!= (time_that_will_not_match_right_away + timedelta(hours=2)).utcoffset()
now_local = dt_util.now()
assert now_local.utcoffset() != (now_local + timedelta(hours=2)).utcoffset()
unsub = async_track_time_change(
hass,
callback(lambda x: specific_runs.append(x)),
hour=2,
minute=30,
second=0,
)
with patch(
"homeassistant.util.dt.utcnow", return_value=time_that_will_not_match_right_away
):
unsub = async_track_time_change(
hass,
callback(lambda x: specific_runs.append(x)),
hour=2,
minute=30,
second=0,
)
async_fire_time_changed(
hass, datetime(yy, mm, dd, 1, 50, 0, 999999, tzinfo=timezone)
)
freezer.move_to(f"{today} 01:50:00.999999+01:00")
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(specific_runs) == 0
async_fire_time_changed(
hass, datetime(yy, mm, dd, 3, 50, 0, 999999, tzinfo=timezone)
)
# There was no 02:30 today, the event should not fire until tomorrow
freezer.move_to(f"{today} 03:50:00.999999+02:00")
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(specific_runs) == 0
async_fire_time_changed(
hass, datetime(yy, mm, dd + 1, 1, 50, 0, 999999, tzinfo=timezone)
)
freezer.move_to(f"{tomorrow} 01:50:00.999999+02:00")
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(specific_runs) == 0
async_fire_time_changed(
hass, datetime(yy, mm, dd + 1, 2, 50, 0, 999999, tzinfo=timezone)
)
freezer.move_to(f"{tomorrow} 02:50:00.999999+02:00")
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(specific_runs) == 1
unsub()
async def test_periodic_task_entering_dst_2(hass):
# DST starts early morning March 28th 2021
@pytest.mark.freeze_time("2021-03-28 01:59:59+01:00")
async def test_periodic_task_entering_dst_2(hass, freezer):
"""Test periodic task behavior when entering dst.
This tests a task firing every second in the range 0..58 (not *:*:59)
@ -3461,220 +3451,182 @@ async def test_periodic_task_entering_dst_2(hass):
dt_util.set_default_time_zone(timezone)
specific_runs = []
# DST starts early morning March 27th 2022
yy = 2022
mm = 3
dd = 27
today = date.today().isoformat()
tomorrow = (date.today() + timedelta(days=1)).isoformat()
# There's no 2022-03-27 02:00:00, the event should not fire until 2022-03-28 03:00:00
time_that_will_not_match_right_away = datetime(
yy, mm, dd, 1, 59, 59, tzinfo=timezone, fold=0
)
# Make sure we enter DST during the test
assert (
time_that_will_not_match_right_away.utcoffset()
!= (time_that_will_not_match_right_away + timedelta(hours=2)).utcoffset()
now_local = dt_util.now()
assert now_local.utcoffset() != (now_local + timedelta(hours=2)).utcoffset()
unsub = async_track_time_change(
hass,
callback(lambda x: specific_runs.append(x)),
second=list(range(59)),
)
with patch(
"homeassistant.util.dt.utcnow", return_value=time_that_will_not_match_right_away
):
unsub = async_track_time_change(
hass,
callback(lambda x: specific_runs.append(x)),
second=list(range(59)),
)
async_fire_time_changed(
hass, datetime(yy, mm, dd, 1, 59, 59, 999999, tzinfo=timezone)
)
freezer.move_to(f"{today} 01:59:59.999999+01:00")
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(specific_runs) == 0
async_fire_time_changed(
hass, datetime(yy, mm, dd, 3, 0, 0, 999999, tzinfo=timezone)
)
freezer.move_to(f"{today} 03:00:00.999999+02:00")
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(specific_runs) == 1
async_fire_time_changed(
hass, datetime(yy, mm, dd, 3, 0, 1, 999999, tzinfo=timezone)
)
freezer.move_to(f"{today} 03:00:01.999999+02:00")
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(specific_runs) == 2
async_fire_time_changed(
hass, datetime(yy, mm, dd + 1, 1, 59, 59, 999999, tzinfo=timezone)
)
freezer.move_to(f"{tomorrow} 01:59:59.999999+02:00")
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(specific_runs) == 3
async_fire_time_changed(
hass, datetime(yy, mm, dd + 1, 2, 0, 0, 999999, tzinfo=timezone)
)
freezer.move_to(f"{tomorrow} 02:00:00.999999+02:00")
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(specific_runs) == 4
unsub()
async def test_periodic_task_leaving_dst(hass):
# DST ends early morning October 31st 2021
@pytest.mark.freeze_time("2021-10-31 02:28:00+02:00")
async def test_periodic_task_leaving_dst(hass, freezer):
"""Test periodic task behavior when leaving dst."""
timezone = dt_util.get_time_zone("Europe/Vienna")
dt_util.set_default_time_zone(timezone)
specific_runs = []
# DST ends early morning Ocotber 30th 2022
yy = 2022
mm = 10
dd = 30
time_that_will_not_match_right_away = datetime(
yy, mm, dd, 2, 28, 0, tzinfo=timezone, fold=0
)
today = date.today().isoformat()
tomorrow = (date.today() + timedelta(days=1)).isoformat()
# Make sure we leave DST during the test
assert (
time_that_will_not_match_right_away.utcoffset()
!= time_that_will_not_match_right_away.replace(fold=1).utcoffset()
)
now_local = dt_util.now()
assert now_local.utcoffset() != (now_local + timedelta(hours=1)).utcoffset()
with patch(
"homeassistant.util.dt.utcnow", return_value=time_that_will_not_match_right_away
):
unsub = async_track_time_change(
hass,
callback(lambda x: specific_runs.append(x)),
hour=2,
minute=30,
second=0,
)
unsub = async_track_time_change(
hass,
callback(lambda x: specific_runs.append(x)),
hour=2,
minute=30,
second=0,
)
# The task should not fire yet
async_fire_time_changed(
hass, datetime(yy, mm, dd, 2, 28, 0, 999999, tzinfo=timezone, fold=0)
)
freezer.move_to(f"{today} 02:28:00.999999+02:00")
async_fire_time_changed(hass)
assert dt_util.now().fold == 0
await hass.async_block_till_done()
assert len(specific_runs) == 0
# The task should fire
async_fire_time_changed(
hass, datetime(yy, mm, dd, 2, 30, 0, 999999, tzinfo=timezone, fold=0)
)
freezer.move_to(f"{today} 02:30:00.999999+02:00")
async_fire_time_changed(hass)
assert dt_util.now().fold == 0
await hass.async_block_till_done()
assert len(specific_runs) == 1
# The task should not fire again
async_fire_time_changed(
hass, datetime(yy, mm, dd, 2, 55, 0, 999999, tzinfo=timezone, fold=0)
)
freezer.move_to(f"{today} 02:55:00.999999+02:00")
async_fire_time_changed(hass)
assert dt_util.now().fold == 0
await hass.async_block_till_done()
assert len(specific_runs) == 1
# DST has ended, the task should not fire yet
async_fire_time_changed(
hass,
datetime(yy, mm, dd, 2, 15, 0, 999999, tzinfo=timezone, fold=1),
)
freezer.move_to(f"{today} 02:15:00.999999+01:00")
async_fire_time_changed(hass)
assert dt_util.now().fold == 1 # DST has ended
await hass.async_block_till_done()
assert len(specific_runs) == 1
# The task should fire
async_fire_time_changed(
hass,
datetime(yy, mm, dd, 2, 45, 0, 999999, tzinfo=timezone, fold=1),
)
freezer.move_to(f"{today} 02:45:00.999999+01:00")
async_fire_time_changed(hass)
assert dt_util.now().fold == 1
await hass.async_block_till_done()
assert len(specific_runs) == 2
# The task should not fire again
async_fire_time_changed(
hass,
datetime(yy, mm, dd, 2, 55, 0, 999999, tzinfo=timezone, fold=1),
)
freezer.move_to(f"{today} 02:55:00.999999+01:00")
async_fire_time_changed(hass)
assert dt_util.now().fold == 1
await hass.async_block_till_done()
assert len(specific_runs) == 2
# The task should fire again the next day
async_fire_time_changed(
hass, datetime(yy, mm, dd + 1, 2, 55, 0, 999999, tzinfo=timezone, fold=1)
)
freezer.move_to(f"{tomorrow} 02:55:00.999999+01:00")
async_fire_time_changed(hass)
assert dt_util.now().fold == 0
await hass.async_block_till_done()
assert len(specific_runs) == 3
unsub()
async def test_periodic_task_leaving_dst_2(hass):
# DST ends early morning October 31st 2021
@pytest.mark.freeze_time("2021-10-31 02:28:00+02:00")
async def test_periodic_task_leaving_dst_2(hass, freezer):
"""Test periodic task behavior when leaving dst."""
timezone = dt_util.get_time_zone("Europe/Vienna")
dt_util.set_default_time_zone(timezone)
specific_runs = []
# DST ends early morning Ocotber 30th 2022
yy = 2022
mm = 10
dd = 30
today = date.today().isoformat()
time_that_will_not_match_right_away = datetime(
yy, mm, dd, 2, 28, 0, tzinfo=timezone, fold=0
)
# Make sure we leave DST during the test
assert (
time_that_will_not_match_right_away.utcoffset()
!= time_that_will_not_match_right_away.replace(fold=1).utcoffset()
)
now_local = dt_util.now()
assert now_local.utcoffset() != (now_local + timedelta(hours=1)).utcoffset()
with patch(
"homeassistant.util.dt.utcnow", return_value=time_that_will_not_match_right_away
):
unsub = async_track_time_change(
hass,
callback(lambda x: specific_runs.append(x)),
minute=30,
second=0,
)
unsub = async_track_time_change(
hass,
callback(lambda x: specific_runs.append(x)),
minute=30,
second=0,
)
# The task should not fire yet
async_fire_time_changed(
hass, datetime(yy, mm, dd, 2, 28, 0, 999999, tzinfo=timezone, fold=0)
)
freezer.move_to(f"{today} 02:28:00.999999+02:00")
async_fire_time_changed(hass)
assert dt_util.now().fold == 0
await hass.async_block_till_done()
assert len(specific_runs) == 0
# The task should fire
async_fire_time_changed(
hass, datetime(yy, mm, dd, 2, 55, 0, 999999, tzinfo=timezone, fold=0)
)
freezer.move_to(f"{today} 02:55:00.999999+02:00")
async_fire_time_changed(hass)
assert dt_util.now().fold == 0
await hass.async_block_till_done()
assert len(specific_runs) == 1
# DST has ended, the task should not fire yet
async_fire_time_changed(
hass, datetime(yy, mm, dd, 2, 15, 0, 999999, tzinfo=timezone, fold=1)
)
freezer.move_to(f"{today} 02:15:00.999999+01:00")
async_fire_time_changed(hass)
assert dt_util.now().fold == 1
await hass.async_block_till_done()
assert len(specific_runs) == 1
# The task should fire
async_fire_time_changed(
hass, datetime(yy, mm, dd, 2, 45, 0, 999999, tzinfo=timezone, fold=1)
)
freezer.move_to(f"{today} 02:45:00.999999+01:00")
async_fire_time_changed(hass)
assert dt_util.now().fold == 1
await hass.async_block_till_done()
assert len(specific_runs) == 2
# The task should not fire again
async_fire_time_changed(
hass,
datetime(yy, mm, dd, 2, 55, 0, 999999, tzinfo=timezone, fold=1),
)
freezer.move_to(f"{today} 02:55:00.999999+01:00")
async_fire_time_changed(hass)
assert dt_util.now().fold == 1
await hass.async_block_till_done()
assert len(specific_runs) == 2
# The task should fire again the next hour
async_fire_time_changed(
hass, datetime(yy, mm, dd, 3, 55, 0, 999999, tzinfo=timezone, fold=0)
)
freezer.move_to(f"{today} 03:55:00.999999+01:00")
async_fire_time_changed(hass)
assert dt_util.now().fold == 0
await hass.async_block_till_done()
assert len(specific_runs) == 3