diff --git a/requirements_test.txt b/requirements_test.txt
index 3355cc82ef17a79c5c3d99c9efdab878a714c6a0..69af0476c5ce9ea0e19db1d77e610ace6c00e855 100644
--- a/requirements_test.txt
+++ b/requirements_test.txt
@@ -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
diff --git a/tests/common.py b/tests/common.py
index 72c18822e009053222710a2f1b16071b4b6fdfc6..a9ba4ae86b4fe7a90ff6cce7cb196a8ef7e59e47 100644
--- a/tests/common.py
+++ b/tests/common.py
@@ -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):
diff --git a/tests/conftest.py b/tests/conftest.py
index 80eb75ef2bd151544cc9c1ca7a2c162f55d66e71..61dcd40e53fa3b5732baf4a820b02ffade03589f 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -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."""
diff --git a/tests/helpers/test_event.py b/tests/helpers/test_event.py
index 9d48b0c0ada4f9e12f48120cbe68d2a88c1428e5..2e8f6264f1a6c665ce3b1495012ad9fd4b431f24 100644
--- a/tests/helpers/test_event.py
+++ b/tests/helpers/test_event.py
@@ -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()
- 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)
+ unsub = async_track_time_change(
+ hass,
+ callback(lambda x: specific_runs.append(x)),
+ hour=2,
+ minute=30,
+ second=0,
)
+
+ 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()
- )
-
- 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)),
- )
+ now_local = dt_util.now()
+ assert now_local.utcoffset() != (now_local + timedelta(hours=2)).utcoffset()
- async_fire_time_changed(
- hass, datetime(yy, mm, dd, 1, 59, 59, 999999, tzinfo=timezone)
+ unsub = async_track_time_change(
+ hass,
+ callback(lambda x: specific_runs.append(x)),
+ second=list(range(59)),
)
+
+ 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