github/ha-core
1
Fork 0
mirror of https://github.com/home-assistant/core synced 2024-07-27 18:58:57 +02:00
Code Releases Activity

Update recorder tests to avoid patching utcnow (#93489)

Browse Source
This commit is contained in:
J. Nick Koston 2023-05-24 17:47:42 -05:00 committed by GitHub
parent f8d8f5caba
commit 9dcd73b308
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 93 additions and 244 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
tests/components/recorder/common.py
View File

@ -13,6 +13,7 @@ import time
from typing import Any, Literal, cast
from unittest.mock import patch, sentinel
from freezegun import freeze_time
from sqlalchemy import create_engine
from sqlalchemy.orm.session import Session
@ -282,9 +283,7 @@ def record_states(hass):
four = three + timedelta(seconds=15 * 5)
states = {mp: [], sns1: [], sns2: [], sns3: [], sns4: []}
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=one
):
with freeze_time(one) as freezer:
states[mp].append(
set_state(mp, "idle", attributes={"media_title": str(sentinel.mt1)})
)
@ -293,25 +292,18 @@ def record_states(hass):
states[sns3].append(set_state(sns3, "10", attributes=sns3_attr))
states[sns4].append(set_state(sns4, "10", attributes=sns4_attr))
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow",
return_value=one + timedelta(microseconds=1),
):
freezer.move_to(one + timedelta(microseconds=1))
states[mp].append(
set_state(mp, "YouTube", attributes={"media_title": str(sentinel.mt2)})
)
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=two
):
freezer.move_to(two)
states[sns1].append(set_state(sns1, "15", attributes=sns1_attr))
states[sns2].append(set_state(sns2, "15", attributes=sns2_attr))
states[sns3].append(set_state(sns3, "15", attributes=sns3_attr))
states[sns4].append(set_state(sns4, "15", attributes=sns4_attr))
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=three
):
freezer.move_to(three)
states[sns1].append(set_state(sns1, "20", attributes=sns1_attr))
states[sns2].append(set_state(sns2, "20", attributes=sns2_attr))
states[sns3].append(set_state(sns3, "20", attributes=sns3_attr))

112
tests/components/recorder/test_history.py
View File

@ -9,6 +9,7 @@ from datetime import datetime, timedelta
import json
from unittest.mock import patch, sentinel
from freezegun import freeze_time
import pytest
from sqlalchemy import text
@ -223,15 +224,11 @@ def test_state_changes_during_period(
point = start + timedelta(seconds=1)
end = point + timedelta(seconds=1)
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=start
):
with freeze_time(start) as freezer:
set_state("idle")
set_state("YouTube")
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=point
):
freezer.move_to(point)
states = [
set_state("idle"),
set_state("Netflix"),
@ -239,9 +236,7 @@ def test_state_changes_during_period(
set_state("YouTube"),
]
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=end
):
freezer.move_to(end)
set_state("Netflix")
set_state("Plex")
@ -272,32 +267,23 @@ def test_state_changes_during_period_descending(
point4 = start + timedelta(seconds=1, microseconds=300)
end = point + timedelta(seconds=1, microseconds=400)
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=start
):
with freeze_time(start) as freezer:
set_state("idle")
set_state("YouTube")
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=point
):
freezer.move_to(point)
states = [set_state("idle")]
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=point2
):
freezer.move_to(point2)
states.append(set_state("Netflix"))
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=point3
):
freezer.move_to(point3)
states.append(set_state("Plex"))
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=point4
):
freezer.move_to(point4)
states.append(set_state("YouTube"))
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=end
):
freezer.move_to(end)
set_state("Netflix")
set_state("Plex")
@ -379,21 +365,15 @@ def test_get_last_state_changes(hass_recorder: Callable[..., HomeAssistant]) ->
start = dt_util.utcnow() - timedelta(minutes=2)
point = start + timedelta(minutes=1)
point2 = point + timedelta(minutes=1, seconds=1)
states = []
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=start
):
with freeze_time(start) as freezer:
set_state("1")
states = []
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=point
):
freezer.move_to(point)
states.append(set_state("2"))
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=point2
):
freezer.move_to(point2)
states.append(set_state("3"))
hist = history.get_last_state_changes(hass, 2, entity_id)
@ -415,21 +395,15 @@ def test_get_last_state_change(hass_recorder: Callable[..., HomeAssistant]) -> N
start = dt_util.utcnow() - timedelta(minutes=2)
point = start + timedelta(minutes=1)
point2 = point + timedelta(minutes=1, seconds=1)
states = []
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=start
):
with freeze_time(start) as freezer:
set_state("1")
states = []
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=point
):
freezer.move_to(point)
set_state("2")
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=point2
):
freezer.move_to(point2)
states.append(set_state("3"))
hist = history.get_last_state_changes(hass, 1, entity_id)
@ -457,14 +431,10 @@ def test_ensure_state_can_be_copied(
start = dt_util.utcnow() - timedelta(minutes=2)
point = start + timedelta(minutes=1)
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=start
):
with freeze_time(start) as freezer:
set_state("1")
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=point
):
freezer.move_to(point)
set_state("2")
hist = history.get_last_state_changes(hass, 2, entity_id)
@ -694,29 +664,18 @@ def test_get_significant_states_only(
points.append(start + timedelta(minutes=i))
states = []
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=start
):
with freeze_time(start) as freezer:
set_state("123", attributes={"attribute": 10.64})
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow",
return_value=points[0],
):
freezer.move_to(points[0])
# Attributes are different, state not
states.append(set_state("123", attributes={"attribute": 21.42}))
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow",
return_value=points[1],
):
freezer.move_to(points[1])
# state is different, attributes not
states.append(set_state("32", attributes={"attribute": 21.42}))
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow",
return_value=points[2],
):
freezer.move_to(points[2])
# everything is different
states.append(set_state("412", attributes={"attribute": 54.23}))
@ -805,9 +764,7 @@ def record_states(hass) -> tuple[datetime, datetime, dict[str, list[State]]]:
four = three + timedelta(seconds=1)
states = {therm: [], therm2: [], mp: [], mp2: [], mp3: [], script_c: []}
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=one
):
with freeze_time(one) as freezer:
states[mp].append(
set_state(mp, "idle", attributes={"media_title": str(sentinel.mt1)})
)
@ -821,17 +778,12 @@ def record_states(hass) -> tuple[datetime, datetime, dict[str, list[State]]]:
set_state(therm, 20, attributes={"current_temperature": 19.5})
)
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow",
return_value=one + timedelta(microseconds=1),
):
freezer.move_to(one + timedelta(microseconds=1))
states[mp].append(
set_state(mp, "YouTube", attributes={"media_title": str(sentinel.mt2)})
)
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=two
):
freezer.move_to(two)
# This state will be skipped only different in time
set_state(mp, "YouTube", attributes={"media_title": str(sentinel.mt3)})
# This state will be skipped because domain is excluded
@ -846,9 +798,7 @@ def record_states(hass) -> tuple[datetime, datetime, dict[str, list[State]]]:
set_state(therm2, 20, attributes={"current_temperature": 19})
)
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=three
):
freezer.move_to(three)
states[mp].append(
set_state(mp, "Netflix", attributes={"media_title": str(sentinel.mt4)})
)

99
tests/components/recorder/test_history_db_schema_30.py
View File

@ -9,6 +9,7 @@ from datetime import datetime, timedelta
import json
from unittest.mock import patch, sentinel
from freezegun import freeze_time
import pytest
from homeassistant.components import recorder
@ -129,15 +130,11 @@ def test_state_changes_during_period(
point = start + timedelta(seconds=1)
end = point + timedelta(seconds=1)
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=start
):
with freeze_time(start) as freezer:
set_state("idle")
set_state("YouTube")
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=point
):
freezer.move_to(point)
states = [
set_state("idle"),
set_state("Netflix"),
@ -145,9 +142,7 @@ def test_state_changes_during_period(
set_state("YouTube"),
]
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=end
):
freezer.move_to(end)
set_state("Netflix")
set_state("Plex")
@ -180,32 +175,24 @@ def test_state_changes_during_period_descending(
point4 = start + timedelta(seconds=1, microseconds=4)
end = point + timedelta(seconds=1)
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=start
):
with freeze_time(start) as freezer:
set_state("idle")
set_state("YouTube")
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=point
):
freezer.move_to(point)
states = [set_state("idle")]
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=point2
):
freezer.move_to(point2)
states.append(set_state("Netflix"))
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=point3
):
freezer.move_to(point3)
states.append(set_state("Plex"))
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=point4
):
freezer.move_to(point4)
states.append(set_state("YouTube"))
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=end
):
freezer.move_to(end)
set_state("Netflix")
set_state("Plex")
@ -238,21 +225,15 @@ def test_get_last_state_changes(hass_recorder: Callable[..., HomeAssistant]) ->
start = dt_util.utcnow() - timedelta(minutes=2)
point = start + timedelta(minutes=1)
point2 = point + timedelta(minutes=1, seconds=1)
states = []
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=start
):
with freeze_time(start) as freezer:
set_state("1")
states = []
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=point
):
freezer.move_to(point)
states.append(set_state("2"))
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=point2
):
freezer.move_to(point2)
states.append(set_state("3"))
hist = history.get_last_state_changes(hass, 2, entity_id)
@ -282,14 +263,10 @@ def test_ensure_state_can_be_copied(
start = dt_util.utcnow() - timedelta(minutes=2)
point = start + timedelta(minutes=1)
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=start
):
with freeze_time(start) as freezer:
set_state("1")
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=point
):
freezer.move_to(point)
set_state("2")
hist = history.get_last_state_changes(hass, 2, entity_id)
@ -546,29 +523,18 @@ def test_get_significant_states_only(
points.append(start + timedelta(minutes=i))
states = []
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=start
):
with freeze_time(start) as freezer:
set_state("123", attributes={"attribute": 10.64})
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow",
return_value=points[0],
):
freezer.move_to(points[0])
# Attributes are different, state not
states.append(set_state("123", attributes={"attribute": 21.42}))
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow",
return_value=points[1],
):
freezer.move_to(points[1])
# state is different, attributes not
states.append(set_state("32", attributes={"attribute": 21.42}))
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow",
return_value=points[2],
):
freezer.move_to(points[2])
# everything is different
states.append(set_state("412", attributes={"attribute": 54.23}))
@ -630,9 +596,7 @@ def record_states(hass) -> tuple[datetime, datetime, dict[str, list[State]]]:
four = three + timedelta(seconds=1)
states = {therm: [], therm2: [], mp: [], mp2: [], mp3: [], script_c: []}
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=one
):
with freeze_time(one) as freezer:
states[mp].append(
set_state(mp, "idle", attributes={"media_title": str(sentinel.mt1)})
)
@ -646,17 +610,12 @@ def record_states(hass) -> tuple[datetime, datetime, dict[str, list[State]]]:
set_state(therm, 20, attributes={"current_temperature": 19.5})
)
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow",
return_value=one + timedelta(microseconds=1),
):
freezer.move_to(one + timedelta(microseconds=1))
states[mp].append(
set_state(mp, "YouTube", attributes={"media_title": str(sentinel.mt2)})
)
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=two
):
freezer.move_to(two)
# This state will be skipped only different in time
set_state(mp, "YouTube", attributes={"media_title": str(sentinel.mt3)})
# This state will be skipped because domain is excluded
@ -671,9 +630,7 @@ def record_states(hass) -> tuple[datetime, datetime, dict[str, list[State]]]:
set_state(therm2, 20, attributes={"current_temperature": 19})
)
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=three
):
freezer.move_to(three)
states[mp].append(
set_state(mp, "Netflix", attributes={"media_title": str(sentinel.mt4)})
)

98
tests/components/recorder/test_history_db_schema_32.py
View File

@ -9,6 +9,7 @@ from datetime import datetime, timedelta
import json
from unittest.mock import patch, sentinel
from freezegun import freeze_time
import pytest
from homeassistant.components import recorder
@ -129,15 +130,11 @@ def test_state_changes_during_period(
point = start + timedelta(seconds=1)
end = point + timedelta(seconds=1)
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=start
):
with freeze_time(start) as freezer:
set_state("idle")
set_state("YouTube")
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=point
):
freezer.move_to(point)
states = [
set_state("idle"),
set_state("Netflix"),
@ -145,9 +142,7 @@ def test_state_changes_during_period(
set_state("YouTube"),
]
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=end
):
freezer.move_to(end)
set_state("Netflix")
set_state("Plex")
@ -180,32 +175,23 @@ def test_state_changes_during_period_descending(
point4 = start + timedelta(seconds=1, microseconds=4)
end = point + timedelta(seconds=1)
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=start
):
with freeze_time(start) as freezer:
set_state("idle")
set_state("YouTube")
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=point
):
freezer.move_to(point)
states = [set_state("idle")]
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=point2
):
freezer.move_to(point2)
states.append(set_state("Netflix"))
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=point3
):
freezer.move_to(point3)
states.append(set_state("Plex"))
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=point4
):
freezer.move_to(point4)
states.append(set_state("YouTube"))
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=end
):
freezer.move_to(end)
set_state("Netflix")
set_state("Plex")
@ -238,21 +224,15 @@ def test_get_last_state_changes(hass_recorder: Callable[..., HomeAssistant]) ->
start = dt_util.utcnow() - timedelta(minutes=2)
point = start + timedelta(minutes=1)
point2 = point + timedelta(minutes=1, seconds=1)
states = []
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=start
):
with freeze_time(start) as freezer:
set_state("1")
states = []
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=point
):
freezer.move_to(point)
states.append(set_state("2"))
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=point2
):
freezer.move_to(point2)
states.append(set_state("3"))
hist = history.get_last_state_changes(hass, 2, entity_id)
@ -282,14 +262,10 @@ def test_ensure_state_can_be_copied(
start = dt_util.utcnow() - timedelta(minutes=2)
point = start + timedelta(minutes=1)
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=start
):
with freeze_time(start) as freezer:
set_state("1")
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=point
):
freezer.move_to(point)
set_state("2")
hist = history.get_last_state_changes(hass, 2, entity_id)
@ -537,29 +513,18 @@ def test_get_significant_states_only(
points.append(start + timedelta(minutes=i))
states = []
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=start
):
with freeze_time(start) as freezer:
set_state("123", attributes={"attribute": 10.64})
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow",
return_value=points[0],
):
freezer.move_to(points[0])
# Attributes are different, state not
states.append(set_state("123", attributes={"attribute": 21.42}))
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow",
return_value=points[1],
):
freezer.move_to(points[1])
# state is different, attributes not
states.append(set_state("32", attributes={"attribute": 21.42}))
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow",
return_value=points[2],
):
freezer.move_to(points[2])
# everything is different
states.append(set_state("412", attributes={"attribute": 54.23}))
@ -621,9 +586,7 @@ def record_states(hass) -> tuple[datetime, datetime, dict[str, list[State]]]:
four = three + timedelta(seconds=1)
states = {therm: [], therm2: [], mp: [], mp2: [], mp3: [], script_c: []}
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=one
):
with freeze_time(one) as freezer:
states[mp].append(
set_state(mp, "idle", attributes={"media_title": str(sentinel.mt1)})
)
@ -637,17 +600,12 @@ def record_states(hass) -> tuple[datetime, datetime, dict[str, list[State]]]:
set_state(therm, 20, attributes={"current_temperature": 19.5})
)
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow",
return_value=one + timedelta(microseconds=1),
):
freezer.move_to(one + timedelta(microseconds=1))
states[mp].append(
set_state(mp, "YouTube", attributes={"media_title": str(sentinel.mt2)})
)
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=two
):
freezer.move_to(two)
# This state will be skipped only different in time
set_state(mp, "YouTube", attributes={"media_title": str(sentinel.mt3)})
# This state will be skipped because domain is excluded
@ -662,9 +620,7 @@ def record_states(hass) -> tuple[datetime, datetime, dict[str, list[State]]]:
set_state(therm2, 20, attributes={"current_temperature": 19})
)
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow", return_value=three
):
freezer.move_to(three)
states[mp].append(
set_state(mp, "Netflix", attributes={"media_title": str(sentinel.mt4)})
)

5
tests/components/recorder/test_purge.py
View File

@ -1438,10 +1438,7 @@ async def _add_test_states(hass: HomeAssistant):
state = f"dontpurgeme_{event_id}"
attributes = {"dontpurgeme": True, **base_attributes}
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow",
return_value=timestamp,
):
with freeze_time(timestamp):
await set_state("test.recorder2", state, attributes=attributes)

5
tests/components/recorder/test_purge_v32_schema.py
View File

@ -733,10 +733,7 @@ async def _add_test_states(hass: HomeAssistant):
state = f"dontpurgeme_{event_id}"
attributes = {"dontpurgeme": True, **base_attributes}
with patch(
"homeassistant.components.recorder.core.dt_util.utcnow",
return_value=timestamp,
):
with freeze_time(timestamp):
await set_state("test.recorder2", state, attributes=attributes)