jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-23 16:57:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Remove EVENT_TIME_CHANGED and EVENT_TIMER_OUT_OF_SYNC (#69643)

Browse Source 
Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
J. Nick Koston 2022-04-09 09:05:54 -10:00 committed by GitHub
parent 689b347904
commit fe6a4bfb1d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
44 changed files with 397 additions and 839 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
homeassistant/components/api/__init__.py
View File

@ -14,7 +14,6 @@ from homeassistant.bootstrap import DATA_LOGGING
from homeassistant.components.http import HomeAssistantView
from homeassistant.const import (
EVENT_HOMEASSISTANT_STOP,
EVENT_TIME_CHANGED,
MATCH_ALL,
URL_API,
URL_API_COMPONENTS,
@ -102,9 +101,6 @@ class APIEventStream(HomeAssistantView):
async def forward_events(event):
"""Forward events to the open request."""
if event.event_type == EVENT_TIME_CHANGED:
return
if restrict and event.event_type not in restrict:
return

21
homeassistant/components/configurator/__init__.py
View File

@ -7,21 +7,14 @@ A callback has to be provided to `request_config` which will be called when
the user has submitted configuration information.
"""
from contextlib import suppress
from datetime import datetime
import functools as ft
from typing import Any
from homeassistant.const import (
ATTR_ENTITY_PICTURE,
ATTR_FRIENDLY_NAME,
EVENT_TIME_CHANGED,
)
from homeassistant.core import (
Event,
HomeAssistant,
ServiceCall,
callback as async_callback,
)
from homeassistant.const import ATTR_ENTITY_PICTURE, ATTR_FRIENDLY_NAME
from homeassistant.core import HomeAssistant, ServiceCall, callback as async_callback
from homeassistant.helpers.entity import async_generate_entity_id
from homeassistant.helpers.event import async_call_later
from homeassistant.helpers.typing import ConfigType
from homeassistant.loader import bind_hass
from homeassistant.util.async_ import run_callback_threadsafe
@ -213,11 +206,11 @@ class Configurator:
self.hass.states.async_set(entity_id, STATE_CONFIGURED)
@async_callback
def deferred_remove(event: Event):
def deferred_remove(now: datetime):
"""Remove the request state."""
self.hass.states.async_remove(entity_id, context=event.context)
self.hass.states.async_remove(entity_id)
self.hass.bus.async_listen_once(EVENT_TIME_CHANGED, deferred_remove)
async_call_later(self.hass, 1, deferred_remove)
async def async_handle_service_call(self, call: ServiceCall) -> None:
"""Handle a configure service call."""

2
homeassistant/components/mqtt_eventstream/__init__.py
View File

@ -14,7 +14,6 @@ from homeassistant.const import (
EVENT_HOMEASSISTANT_STARTED,
EVENT_HOMEASSISTANT_STOP,
EVENT_STATE_CHANGED,
EVENT_TIME_CHANGED,
MATCH_ALL,
)
from homeassistant.core import EventOrigin, HomeAssistant, State, callback
@ -59,7 +58,6 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
pub_topic = conf.get(CONF_PUBLISH_TOPIC)
sub_topic = conf.get(CONF_SUBSCRIBE_TOPIC)
ignore_event = conf.get(CONF_IGNORE_EVENT)
ignore_event.append(EVENT_TIME_CHANGED)
async def _event_publisher(event):
"""Handle events by publishing them on the MQTT queue."""

4
homeassistant/components/recorder/__init__.py
View File

@ -30,7 +30,6 @@ from homeassistant.const import (
EVENT_HOMEASSISTANT_STARTED,
EVENT_HOMEASSISTANT_STOP,
EVENT_STATE_CHANGED,
EVENT_TIME_CHANGED,
MATCH_ALL,
)
from homeassistant.core import (
@ -775,9 +774,6 @@ class Recorder(threading.Thread):
@callback
def _async_event_filter(self, event: Event) -> bool:
"""Filter events."""
if event.event_type == EVENT_TIME_CHANGED:
return False
if event.event_type in self.exclude_t:
return False

4
homeassistant/components/websocket_api/commands.py
View File

@ -12,7 +12,6 @@ import voluptuous as vol
from homeassistant.auth.permissions.const import CAT_ENTITIES, POLICY_READ
from homeassistant.const import (
EVENT_STATE_CHANGED,
EVENT_TIME_CHANGED,
MATCH_ALL,
SIGNAL_BOOTSTRAP_INTEGRATONS,
)
@ -113,9 +112,6 @@ def handle_subscribe_events(
@callback
def forward_events(event: Event) -> None:
"""Forward events to websocket."""
if event.event_type == EVENT_TIME_CHANGED:
return
connection.send_message(messages.cached_event_message(msg["id"], event))
connection.subscriptions[msg["id"]] = hass.bus.async_listen(

3
homeassistant/const.py
View File

@ -269,9 +269,6 @@ EVENT_SERVICE_REGISTERED: Final = "service_registered"
EVENT_SERVICE_REMOVED: Final = "service_removed"
EVENT_STATE_CHANGED: Final = "state_changed"
EVENT_THEMES_UPDATED: Final = "themes_updated"
EVENT_TIMER_OUT_OF_SYNC: Final = "timer_out_of_sync"
EVENT_TIME_CHANGED: Final = "time_changed"
# #### DEVICE CLASSES ####
# DEVICE_CLASS_* below are deprecated as of 2021.12

53
homeassistant/core.py
View File

@ -45,8 +45,6 @@ from .backports.enum import StrEnum
from .const import (
ATTR_DOMAIN,
ATTR_FRIENDLY_NAME,
ATTR_NOW,
ATTR_SECONDS,
ATTR_SERVICE,
ATTR_SERVICE_DATA,
CONF_UNIT_SYSTEM_IMPERIAL,
@ -60,8 +58,6 @@ from .const import (
EVENT_SERVICE_REGISTERED,
EVENT_SERVICE_REMOVED,
EVENT_STATE_CHANGED,
EVENT_TIME_CHANGED,
EVENT_TIMER_OUT_OF_SYNC,
LENGTH_METERS,
MATCH_ALL,
MAX_LENGTH_EVENT_EVENT_TYPE,
@ -348,7 +344,6 @@ class HomeAssistant:
self.state = CoreState.running
self.bus.async_fire(EVENT_CORE_CONFIG_UPDATE)
self.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
_async_create_timer(self)
def add_job(
self, target: Callable[..., Any] | Coroutine[Any, Any, Any], *args: Any
@ -843,8 +838,7 @@ class EventBus:
event = Event(event_type, event_data, origin, time_fired, context)
if event_type != EVENT_TIME_CHANGED:
_LOGGER.debug("Bus:Handling %s", event)
_LOGGER.debug("Bus:Handling %s", event)
if not listeners:
return
@ -1911,48 +1905,3 @@ class Config:
CORE_STORAGE_VERSION, CORE_STORAGE_KEY, private=True, atomic_writes=True
)
await store.async_save(data)
def _async_create_timer(hass: HomeAssistant) -> None:
"""Create a timer that will start on HOMEASSISTANT_START."""
handle = None
timer_context = Context()
def schedule_tick(now: datetime.datetime) -> None:
"""Schedule a timer tick when the next second rolls around."""
nonlocal handle
slp_seconds = 1 - (now.microsecond / 10**6)
target = monotonic() + slp_seconds
handle = hass.loop.call_later(slp_seconds, fire_time_event, target)
@callback
def fire_time_event(target: float) -> None:
"""Fire next time event."""
now = dt_util.utcnow()
hass.bus.async_fire(
EVENT_TIME_CHANGED, {ATTR_NOW: now}, time_fired=now, context=timer_context
)
# If we are more than a second late, a tick was missed
if (late := monotonic() - target) > 1:
hass.bus.async_fire(
EVENT_TIMER_OUT_OF_SYNC,
{ATTR_SECONDS: late},
time_fired=now,
context=timer_context,
)
schedule_tick(now)
@callback
def stop_timer(_: Event) -> None:
"""Stop the timer."""
if handle is not None:
handle.cancel()
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, stop_timer)
_LOGGER.info("Timer:starting")
schedule_tick(dt_util.utcnow())

17
homeassistant/helpers/event.py
View File

@ -16,10 +16,8 @@ from typing_extensions import Concatenate, ParamSpec
from homeassistant.const import (
ATTR_ENTITY_ID,
ATTR_NOW,
EVENT_CORE_CONFIG_UPDATE,
EVENT_STATE_CHANGED,
EVENT_TIME_CHANGED,
MATCH_ALL,
SUN_EVENT_SUNRISE,
SUN_EVENT_SUNSET,
@ -1463,18 +1461,17 @@ def async_track_utc_time_change(
local: bool = False,
) -> CALLBACK_TYPE:
"""Add a listener that will fire if time matches a pattern."""
job = HassJob(action)
# We do not have to wrap the function with time pattern matching logic
# if no pattern given
if all(val is None for val in (hour, minute, second)):
# Previously this relied on EVENT_TIME_FIRED
# which meant it would not fire right away because
# the caller would always be misaligned with the call
# time vs the fire time by < 1s. To preserve this
# misalignment we use async_track_time_interval here
return async_track_time_interval(hass, action, timedelta(seconds=1))
@callback
def time_change_listener(event: Event) -> None:
"""Fire every time event that comes in."""
hass.async_run_hass_job(job, cast(datetime, event.data[ATTR_NOW]))
return hass.bus.async_listen(EVENT_TIME_CHANGED, time_change_listener)
job = HassJob(action)
matching_seconds = dt_util.parse_time_expression(second, 0, 59)
matching_minutes = dt_util.parse_time_expression(minute, 0, 59)
matching_hours = dt_util.parse_time_expression(hour, 0, 23)

31
homeassistant/scripts/benchmark/__init__.py
View File

@ -6,7 +6,6 @@ import asyncio
import collections
from collections.abc import Callable
from contextlib import suppress
from datetime import datetime
import json
import logging
from timeit import default_timer as timer
@ -14,7 +13,7 @@ from typing import TypeVar
from homeassistant import core
from homeassistant.components.websocket_api.const import JSON_DUMP
from homeassistant.const import ATTR_NOW, EVENT_STATE_CHANGED, EVENT_TIME_CHANGED
from homeassistant.const import EVENT_STATE_CHANGED
from homeassistant.helpers.entityfilter import convert_include_exclude_filter
from homeassistant.helpers.json import JSONEncoder
from homeassistant.util import dt as dt_util
@ -119,34 +118,6 @@ async def fire_events_with_filter(hass):
return timer() - start
@benchmark
async def time_changed_helper(hass):
"""Run a million events through time changed helper."""
count = 0
event = asyncio.Event()
@core.callback
def listener(_):
"""Handle event."""
nonlocal count
count += 1
if count == 10**6:
event.set()
hass.helpers.event.async_track_time_change(listener, minute=0, second=0)
event_data = {ATTR_NOW: datetime(2017, 10, 10, 15, 0, 0, tzinfo=dt_util.UTC)}
for _ in range(10**6):
hass.bus.async_fire(EVENT_TIME_CHANGED, event_data)
start = timer()
await event.wait()
return timer() - start
@benchmark
async def state_changed_helper(hass):
"""Run a million events through state changed helper with 1000 entities."""

7
tests/common.py
View File

@ -40,7 +40,6 @@ from homeassistant.const import (
DEVICE_DEFAULT_NAME,
EVENT_HOMEASSISTANT_CLOSE,
EVENT_STATE_CHANGED,
EVENT_TIME_CHANGED,
STATE_OFF,
STATE_ON,
)
@ -314,9 +313,7 @@ async def async_test_home_assistant(loop, load_registries=True):
async def mock_async_start():
"""Start the mocking."""
# We only mock time during tests and we want to track tasks
with patch("homeassistant.core._async_create_timer"), patch.object(
hass, "async_stop_track_tasks"
):
with patch.object(hass, "async_stop_track_tasks"):
await orig_start()
hass.async_start = mock_async_start
@ -386,8 +383,6 @@ def async_fire_time_changed(
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):
if not isinstance(task, asyncio.TimerHandle):
continue

76
tests/components/cloud/test_google_config.py
View File

@ -2,6 +2,7 @@
from http import HTTPStatus
from unittest.mock import Mock, patch
from freezegun import freeze_time
import pytest
from homeassistant.components.cloud import GACTIONS_SCHEMA
@ -82,49 +83,48 @@ async def test_sync_entities(mock_conf, hass, cloud_prefs):
assert len(mock_request_sync.mock_calls) == 1
async def test_google_update_expose_trigger_sync(
hass, legacy_patchable_time, cloud_prefs
):
async def test_google_update_expose_trigger_sync(hass, cloud_prefs):
"""Test Google config responds to updating exposed entities."""
config = CloudGoogleConfig(
hass,
GACTIONS_SCHEMA({}),
"mock-user-id",
cloud_prefs,
Mock(claims={"cognito:username": "abcdefghjkl"}),
)
await config.async_initialize()
await config.async_connect_agent_user("mock-user-id")
with patch.object(config, "async_sync_entities") as mock_sync, patch.object(
ga_helpers, "SYNC_DELAY", 0
):
await cloud_prefs.async_update_google_entity_config(
entity_id="light.kitchen", should_expose=True
with freeze_time(utcnow()):
config = CloudGoogleConfig(
hass,
GACTIONS_SCHEMA({}),
"mock-user-id",
cloud_prefs,
Mock(claims={"cognito:username": "abcdefghjkl"}),
)
await hass.async_block_till_done()
async_fire_time_changed(hass, utcnow())
await hass.async_block_till_done()
await config.async_initialize()
await config.async_connect_agent_user("mock-user-id")
assert len(mock_sync.mock_calls) == 1
with patch.object(config, "async_sync_entities") as mock_sync, patch.object(
ga_helpers, "SYNC_DELAY", 0
):
await cloud_prefs.async_update_google_entity_config(
entity_id="light.kitchen", should_expose=True
)
await hass.async_block_till_done()
async_fire_time_changed(hass, utcnow())
await hass.async_block_till_done()
with patch.object(config, "async_sync_entities") as mock_sync, patch.object(
ga_helpers, "SYNC_DELAY", 0
):
await cloud_prefs.async_update_google_entity_config(
entity_id="light.kitchen", should_expose=False
)
await cloud_prefs.async_update_google_entity_config(
entity_id="binary_sensor.door", should_expose=True
)
await cloud_prefs.async_update_google_entity_config(
entity_id="sensor.temp", should_expose=True
)
await hass.async_block_till_done()
async_fire_time_changed(hass, utcnow())
await hass.async_block_till_done()
assert len(mock_sync.mock_calls) == 1
assert len(mock_sync.mock_calls) == 1
with patch.object(config, "async_sync_entities") as mock_sync, patch.object(
ga_helpers, "SYNC_DELAY", 0
):
await cloud_prefs.async_update_google_entity_config(
entity_id="light.kitchen", should_expose=False
)
await cloud_prefs.async_update_google_entity_config(
entity_id="binary_sensor.door", should_expose=True
)
await cloud_prefs.async_update_google_entity_config(
entity_id="sensor.temp", should_expose=True
)
await hass.async_block_till_done()
async_fire_time_changed(hass, utcnow())
await hass.async_block_till_done()
assert len(mock_sync.mock_calls) == 1
async def test_google_entity_registry_sync(hass, mock_cloud_login, cloud_prefs):

10
tests/components/configurator/test_init.py
View File

@ -1,7 +1,12 @@
"""The tests for the Configurator component."""
from datetime import timedelta
import homeassistant.components.configurator as configurator
from homeassistant.const import ATTR_FRIENDLY_NAME, EVENT_TIME_CHANGED
from homeassistant.const import ATTR_FRIENDLY_NAME
import homeassistant.util.dt as dt_util
from tests.common import async_fire_time_changed
async def test_request_least_info(hass):
@ -95,8 +100,7 @@ async def test_request_done_works(hass):
request_id = configurator.async_request_config(hass, "Test Request", lambda _: None)
configurator.async_request_done(hass, request_id)
assert len(hass.states.async_all()) == 1
hass.bus.async_fire(EVENT_TIME_CHANGED)
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=1))
await hass.async_block_till_done()
assert len(hass.states.async_all()) == 0

12
tests/components/demo/test_geo_location.py
View File

@ -1,8 +1,6 @@
"""The tests for the demo platform."""
from unittest.mock import patch
import pytest
from freezegun import freeze_time
from homeassistant.components import geo_location
from homeassistant.components.demo.geo_location import (
@ -23,17 +21,11 @@ from tests.common import assert_setup_component, async_fire_time_changed
CONFIG = {geo_location.DOMAIN: [{"platform": "demo"}]}
@pytest.fixture(autouse=True)
def mock_legacy_time(legacy_patchable_time):
"""Make time patchable for all the tests."""
yield
async def test_setup_platform(hass):
"""Test setup of demo platform via configuration."""
utcnow = dt_util.utcnow()
# Patching 'utcnow' to gain more control over the timed update.
with patch("homeassistant.util.dt.utcnow", return_value=utcnow):
with freeze_time(utcnow):
with assert_setup_component(1, geo_location.DOMAIN):
assert await async_setup_component(hass, geo_location.DOMAIN, CONFIG)
await hass.async_block_till_done()

125
tests/components/flux/test_switch.py
View File

@ -1,6 +1,7 @@
"""The tests for the Flux switch platform."""
from unittest.mock import patch
from freezegun import freeze_time
import pytest
from homeassistant.components import light, switch
@ -133,9 +134,7 @@ async def test_invalid_config_no_lights(hass):
await hass.async_block_till_done()
async def test_flux_when_switch_is_off(
hass, legacy_patchable_time, enable_custom_integrations
):
async def test_flux_when_switch_is_off(hass, enable_custom_integrations):
"""Test the flux switch when it is off."""
platform = getattr(hass.components, "test.light")
platform.init()
@ -161,9 +160,7 @@ async def test_flux_when_switch_is_off(
return sunrise_time
return sunset_time
with patch(
"homeassistant.components.flux.switch.dt_utcnow", return_value=test_time
), patch(
with freeze_time(test_time), patch(
"homeassistant.components.flux.switch.get_astral_event_date",
side_effect=event_date,
):
@ -186,9 +183,7 @@ async def test_flux_when_switch_is_off(
assert not turn_on_calls
async def test_flux_before_sunrise(
hass, legacy_patchable_time, enable_custom_integrations
):
async def test_flux_before_sunrise(hass, enable_custom_integrations):
"""Test the flux switch before sunrise."""
platform = getattr(hass.components, "test.light")
platform.init()
@ -215,9 +210,7 @@ async def test_flux_before_sunrise(
return sunset_time
await hass.async_block_till_done()
with patch(
"homeassistant.components.flux.switch.dt_utcnow", return_value=test_time
), patch(
with freeze_time(test_time), patch(
"homeassistant.components.flux.switch.get_astral_event_date",
side_effect=event_date,
):
@ -247,9 +240,7 @@ async def test_flux_before_sunrise(
assert call.data[light.ATTR_XY_COLOR] == [0.606, 0.379]
async def test_flux_before_sunrise_known_location(
hass, legacy_patchable_time, enable_custom_integrations
):
async def test_flux_before_sunrise_known_location(hass, enable_custom_integrations):
"""Test the flux switch before sunrise."""
platform = getattr(hass.components, "test.light")
platform.init()
@ -274,9 +265,7 @@ async def test_flux_before_sunrise_known_location(
)
await hass.async_block_till_done()
with patch(
"homeassistant.components.flux.switch.dt_utcnow", return_value=test_time
):
with freeze_time(test_time):
assert await async_setup_component(
hass,
switch.DOMAIN,
@ -308,9 +297,7 @@ async def test_flux_before_sunrise_known_location(
# pylint: disable=invalid-name
async def test_flux_after_sunrise_before_sunset(
hass, legacy_patchable_time, enable_custom_integrations
):
async def test_flux_after_sunrise_before_sunset(hass, enable_custom_integrations):
"""Test the flux switch after sunrise and before sunset."""
platform = getattr(hass.components, "test.light")
platform.init()
@ -336,9 +323,7 @@ async def test_flux_after_sunrise_before_sunset(
return sunrise_time
return sunset_time
with patch(
"homeassistant.components.flux.switch.dt_utcnow", return_value=test_time
), patch(
with freeze_time(test_time), patch(
"homeassistant.components.flux.switch.get_astral_event_date",
side_effect=event_date,
):
@ -369,9 +354,7 @@ async def test_flux_after_sunrise_before_sunset(
# pylint: disable=invalid-name
async def test_flux_after_sunset_before_stop(
hass, legacy_patchable_time, enable_custom_integrations
):
async def test_flux_after_sunset_before_stop(hass, enable_custom_integrations):
"""Test the flux switch after sunset and before stop."""
platform = getattr(hass.components, "test.light")
platform.init()
@ -397,9 +380,7 @@ async def test_flux_after_sunset_before_stop(
return sunrise_time
return sunset_time
with patch(
"homeassistant.components.flux.switch.dt_utcnow", return_value=test_time
), patch(
with freeze_time(test_time), patch(
"homeassistant.components.flux.switch.get_astral_event_date",
side_effect=event_date,
):
@ -431,9 +412,7 @@ async def test_flux_after_sunset_before_stop(
# pylint: disable=invalid-name
async def test_flux_after_stop_before_sunrise(
hass, legacy_patchable_time, enable_custom_integrations
):
async def test_flux_after_stop_before_sunrise(hass, enable_custom_integrations):
"""Test the flux switch after stop and before sunrise."""
platform = getattr(hass.components, "test.light")
platform.init()
@ -459,9 +438,7 @@ async def test_flux_after_stop_before_sunrise(
return sunrise_time
return sunset_time
with patch(
"homeassistant.components.flux.switch.dt_utcnow", return_value=test_time
), patch(
with freeze_time(test_time), patch(
"homeassistant.components.flux.switch.get_astral_event_date",
side_effect=event_date,
):
@ -492,9 +469,7 @@ async def test_flux_after_stop_before_sunrise(
# pylint: disable=invalid-name
async def test_flux_with_custom_start_stop_times(
hass, legacy_patchable_time, enable_custom_integrations
):
async def test_flux_with_custom_start_stop_times(hass, enable_custom_integrations):
"""Test the flux with custom start and stop times."""
platform = getattr(hass.components, "test.light")
platform.init()
@ -520,9 +495,7 @@ async def test_flux_with_custom_start_stop_times(
return sunrise_time
return sunset_time
with patch(
"homeassistant.components.flux.switch.dt_utcnow", return_value=test_time
), patch(
with freeze_time(test_time), patch(
"homeassistant.components.flux.switch.get_astral_event_date",
side_effect=event_date,
):
@ -554,9 +527,7 @@ async def test_flux_with_custom_start_stop_times(
assert call.data[light.ATTR_XY_COLOR] == [0.504, 0.385]
async def test_flux_before_sunrise_stop_next_day(
hass, legacy_patchable_time, enable_custom_integrations
):
async def test_flux_before_sunrise_stop_next_day(hass, enable_custom_integrations):
"""Test the flux switch before sunrise.
This test has the stop_time on the next day (after midnight).
@ -585,9 +556,7 @@ async def test_flux_before_sunrise_stop_next_day(
return sunrise_time
return sunset_time
with patch(
"homeassistant.components.flux.switch.dt_utcnow", return_value=test_time
), patch(
with freeze_time(test_time), patch(
"homeassistant.components.flux.switch.get_astral_event_date",
side_effect=event_date,
):
@ -620,7 +589,7 @@ async def test_flux_before_sunrise_stop_next_day(
# pylint: disable=invalid-name
async def test_flux_after_sunrise_before_sunset_stop_next_day(
hass, legacy_patchable_time, enable_custom_integrations
hass, enable_custom_integrations
):
"""
Test the flux switch after sunrise and before sunset.
@ -651,9 +620,7 @@ async def test_flux_after_sunrise_before_sunset_stop_next_day(
return sunrise_time
return sunset_time
with patch(
"homeassistant.components.flux.switch.dt_utcnow", return_value=test_time
), patch(
with freeze_time(test_time), patch(
"homeassistant.components.flux.switch.get_astral_event_date",
side_effect=event_date,
):
@ -687,7 +654,7 @@ async def test_flux_after_sunrise_before_sunset_stop_next_day(
# pylint: disable=invalid-name
@pytest.mark.parametrize("x", [0, 1])
async def test_flux_after_sunset_before_midnight_stop_next_day(
hass, legacy_patchable_time, x, enable_custom_integrations
hass, x, enable_custom_integrations
):
"""Test the flux switch after sunset and before stop.
@ -717,9 +684,7 @@ async def test_flux_after_sunset_before_midnight_stop_next_day(
return sunrise_time
return sunset_time
with patch(
"homeassistant.components.flux.switch.dt_utcnow", return_value=test_time
), patch(
with freeze_time(test_time), patch(
"homeassistant.components.flux.switch.get_astral_event_date",
side_effect=event_date,
):
@ -752,7 +717,7 @@ async def test_flux_after_sunset_before_midnight_stop_next_day(
# pylint: disable=invalid-name
async def test_flux_after_sunset_after_midnight_stop_next_day(
hass, legacy_patchable_time, enable_custom_integrations
hass, enable_custom_integrations
):
"""Test the flux switch after sunset and before stop.
@ -782,9 +747,7 @@ async def test_flux_after_sunset_after_midnight_stop_next_day(
return sunrise_time
return sunset_time
with patch(
"homeassistant.components.flux.switch.dt_utcnow", return_value=test_time
), patch(
with freeze_time(test_time), patch(
"homeassistant.components.flux.switch.get_astral_event_date",
side_effect=event_date,
):
@ -817,7 +780,7 @@ async def test_flux_after_sunset_after_midnight_stop_next_day(
# pylint: disable=invalid-name
async def test_flux_after_stop_before_sunrise_stop_next_day(
hass, legacy_patchable_time, enable_custom_integrations
hass, enable_custom_integrations
):
"""Test the flux switch after stop and before sunrise.
@ -847,9 +810,7 @@ async def test_flux_after_stop_before_sunrise_stop_next_day(
return sunrise_time
return sunset_time
with patch(
"homeassistant.components.flux.switch.dt_utcnow", return_value=test_time
), patch(
with freeze_time(test_time), patch(
"homeassistant.components.flux.switch.get_astral_event_date",
side_effect=event_date,
):
@ -881,9 +842,7 @@ async def test_flux_after_stop_before_sunrise_stop_next_day(
# pylint: disable=invalid-name
async def test_flux_with_custom_colortemps(
hass, legacy_patchable_time, enable_custom_integrations
):
async def test_flux_with_custom_colortemps(hass, enable_custom_integrations):
"""Test the flux with custom start and stop colortemps."""
platform = getattr(hass.components, "test.light")
platform.init()
@ -909,9 +868,7 @@ async def test_flux_with_custom_colortemps(
return sunrise_time
return sunset_time
with patch(
"homeassistant.components.flux.switch.dt_utcnow", return_value=test_time
), patch(
with freeze_time(test_time), patch(
"homeassistant.components.flux.switch.get_astral_event_date",
side_effect=event_date,
):
@ -945,9 +902,7 @@ async def test_flux_with_custom_colortemps(
# pylint: disable=invalid-name
async def test_flux_with_custom_brightness(
hass, legacy_patchable_time, enable_custom_integrations
):
async def test_flux_with_custom_brightness(hass, enable_custom_integrations):
"""Test the flux with custom start and stop colortemps."""
platform = getattr(hass.components, "test.light")
platform.init()
@ -973,9 +928,7 @@ async def test_flux_with_custom_brightness(
return sunrise_time
return sunset_time
with patch(
"homeassistant.components.flux.switch.dt_utcnow", return_value=test_time
), patch(
with freeze_time(test_time), patch(
"homeassistant.components.flux.switch.get_astral_event_date",
side_effect=event_date,
):
@ -1007,9 +960,7 @@ async def test_flux_with_custom_brightness(
assert call.data[light.ATTR_XY_COLOR] == [0.506, 0.385]
async def test_flux_with_multiple_lights(
hass, legacy_patchable_time, enable_custom_integrations
):
async def test_flux_with_multiple_lights(hass, enable_custom_integrations):
"""Test the flux switch with multiple light entities."""
platform = getattr(hass.components, "test.light")
platform.init()
@ -1052,9 +1003,7 @@ async def test_flux_with_multiple_lights(
return sunrise_time
return sunset_time
with patch(
"homeassistant.components.flux.switch.dt_utcnow", return_value=test_time
), patch(
with freeze_time(test_time), patch(
"homeassistant.components.flux.switch.get_astral_event_date",
side_effect=event_date,
):
@ -1090,7 +1039,7 @@ async def test_flux_with_multiple_lights(
assert call.data[light.ATTR_XY_COLOR] == [0.46, 0.376]
async def test_flux_with_mired(hass, legacy_patchable_time, enable_custom_integrations):
async def test_flux_with_mired(hass, enable_custom_integrations):
"""Test the flux switch´s mode mired."""
platform = getattr(hass.components, "test.light")
platform.init()
@ -1115,9 +1064,7 @@ async def test_flux_with_mired(hass, legacy_patchable_time, enable_custom_integr
return sunrise_time
return sunset_time
with patch(
"homeassistant.components.flux.switch.dt_utcnow", return_value=test_time
), patch(
with freeze_time(test_time), patch(
"homeassistant.components.flux.switch.get_astral_event_date",
side_effect=event_date,
):
@ -1147,7 +1094,7 @@ async def test_flux_with_mired(hass, legacy_patchable_time, enable_custom_integr
assert call.data[light.ATTR_COLOR_TEMP] == 269
async def test_flux_with_rgb(hass, legacy_patchable_time, enable_custom_integrations):
async def test_flux_with_rgb(hass, enable_custom_integrations):
"""Test the flux switch´s mode rgb."""
platform = getattr(hass.components, "test.light")
platform.init()
@ -1172,9 +1119,7 @@ async def test_flux_with_rgb(hass, legacy_patchable_time, enable_custom_integrat
return sunrise_time
return sunset_time
with patch(
"homeassistant.components.flux.switch.dt_utcnow", return_value=test_time
), patch(
with freeze_time(test_time), patch(
"homeassistant.components.flux.switch.get_astral_event_date",
side_effect=event_date,
):

10
tests/components/gdacs/test_geo_location.py
View File

@ -2,6 +2,8 @@
import datetime
from unittest.mock import patch
from freezegun import freeze_time
from homeassistant.components import gdacs
from homeassistant.components.gdacs import DEFAULT_SCAN_INTERVAL, DOMAIN, FEED
from homeassistant.components.gdacs.geo_location import (
@ -40,7 +42,7 @@ from tests.components.gdacs import _generate_mock_feed_entry
CONFIG = {gdacs.DOMAIN: {CONF_RADIUS: 200}}
async def test_setup(hass, legacy_patchable_time):
async def test_setup(hass):
"""Test the general setup of the integration."""
# Set up some mock feed entries for this test.
mock_entry_1 = _generate_mock_feed_entry(
@ -86,7 +88,7 @@ async def test_setup(hass, legacy_patchable_time):
# Patching 'utcnow' to gain more control over the timed update.
utcnow = dt_util.utcnow()
with patch("homeassistant.util.dt.utcnow", return_value=utcnow), patch(
with freeze_time(utcnow), patch(
"aio_georss_client.feed.GeoRssFeed.update"
) as mock_feed_update:
mock_feed_update.return_value = "OK", [mock_entry_1, mock_entry_2, mock_entry_3]
@ -203,7 +205,7 @@ async def test_setup(hass, legacy_patchable_time):
assert len(entity_registry.entities) == 1
async def test_setup_imperial(hass, legacy_patchable_time):
async def test_setup_imperial(hass):
"""Test the setup of the integration using imperial unit system."""
hass.config.units = IMPERIAL_SYSTEM
# Set up some mock feed entries for this test.
@ -219,7 +221,7 @@ async def test_setup_imperial(hass, legacy_patchable_time):
# Patching 'utcnow' to gain more control over the timed update.
utcnow = dt_util.utcnow()
with patch("homeassistant.util.dt.utcnow", return_value=utcnow), patch(
with freeze_time(utcnow), patch(
"aio_georss_client.feed.GeoRssFeed.update"
) as mock_feed_update, patch(
"aio_georss_client.feed.GeoRssFeed.last_timestamp", create=True

6
tests/components/gdacs/test_sensor.py
View File

@ -1,6 +1,8 @@
"""The tests for the GDACS Feed integration."""
from unittest.mock import patch
from freezegun import freeze_time
from homeassistant.components import gdacs
from homeassistant.components.gdacs import DEFAULT_SCAN_INTERVAL
from homeassistant.components.gdacs.sensor import (
@ -26,7 +28,7 @@ from tests.components.gdacs import _generate_mock_feed_entry
CONFIG = {gdacs.DOMAIN: {CONF_RADIUS: 200}}
async def test_setup(hass, legacy_patchable_time):
async def test_setup(hass):
"""Test the general setup of the integration."""
# Set up some mock feed entries for this test.
mock_entry_1 = _generate_mock_feed_entry(
@ -52,7 +54,7 @@ async def test_setup(hass, legacy_patchable_time):
# Patching 'utcnow' to gain more control over the timed update.
utcnow = dt_util.utcnow()
with patch("homeassistant.util.dt.utcnow", return_value=utcnow), patch(
with freeze_time(utcnow), patch(
"aio_georss_client.feed.GeoRssFeed.update"
) as mock_feed_update:
mock_feed_update.return_value = "OK", [mock_entry_1, mock_entry_2, mock_entry_3]

9
tests/components/geo_json_events/test_geo_location.py
View File

@ -2,6 +2,7 @@
from unittest.mock import ANY, MagicMock, call, patch
from aio_geojson_generic_client import GenericFeed
from freezegun import freeze_time
from homeassistant.components import geo_location
from homeassistant.components.geo_json_events.geo_location import (
@ -57,7 +58,7 @@ def _generate_mock_feed_entry(external_id, title, distance_to_home, coordinates)
return feed_entry
async def test_setup(hass, legacy_patchable_time):
async def test_setup(hass):
"""Test the general setup of the platform."""
# Set up some mock feed entries for this test.
mock_entry_1 = _generate_mock_feed_entry("1234", "Title 1", 15.5, (-31.0, 150.0))
@ -67,7 +68,7 @@ async def test_setup(hass, legacy_patchable_time):
# Patching 'utcnow' to gain more control over the timed update.
utcnow = dt_util.utcnow()
with patch("homeassistant.util.dt.utcnow", return_value=utcnow), patch(
with freeze_time(utcnow), patch(
"aio_geojson_client.feed.GeoJsonFeed.update"
) as mock_feed_update:
mock_feed_update.return_value = (
@ -186,7 +187,7 @@ async def test_setup_with_custom_location(hass):
)
async def test_setup_race_condition(hass, legacy_patchable_time):
async def test_setup_race_condition(hass):
"""Test a particular race condition experienced."""
# 1. Feed returns 1 entry -> Feed manager creates 1 entity.
# 2. Feed returns error -> Feed manager removes 1 entity.
@ -205,7 +206,7 @@ async def test_setup_race_condition(hass, legacy_patchable_time):
# Patching 'utcnow' to gain more control over the timed update.
utcnow = dt_util.utcnow()
with patch("homeassistant.util.dt.utcnow", return_value=utcnow), patch(
with freeze_time(utcnow), patch(
"aio_geojson_client.feed.GeoJsonFeed.update"
) as mock_feed_update, assert_setup_component(1, geo_location.DOMAIN):
assert await async_setup_component(hass, geo_location.DOMAIN, CONFIG)

6
tests/components/geonetnz_quakes/test_sensor.py
View File

@ -2,6 +2,8 @@
import datetime
from unittest.mock import patch
from freezegun import freeze_time
from homeassistant.components import geonetnz_quakes
from homeassistant.components.geonetnz_quakes import DEFAULT_SCAN_INTERVAL
from homeassistant.components.geonetnz_quakes.sensor import (
@ -27,7 +29,7 @@ from tests.components.geonetnz_quakes import _generate_mock_feed_entry
CONFIG = {geonetnz_quakes.DOMAIN: {CONF_RADIUS: 200}}
async def test_setup(hass, legacy_patchable_time):
async def test_setup(hass):
"""Test the general setup of the integration."""
# Set up some mock feed entries for this test.
mock_entry_1 = _generate_mock_feed_entry(
@ -53,7 +55,7 @@ async def test_setup(hass, legacy_patchable_time):
# Patching 'utcnow' to gain more control over the timed update.
utcnow = dt_util.utcnow()
with patch("homeassistant.util.dt.utcnow", return_value=utcnow), patch(
with freeze_time(utcnow), patch(
"aio_geojson_client.feed.GeoJsonFeed.update"
) as mock_feed_update:
mock_feed_update.return_value = "OK", [mock_entry_1, mock_entry_2, mock_entry_3]

6
tests/components/geonetnz_volcano/test_sensor.py
View File

@ -1,6 +1,8 @@
"""The tests for the GeoNet NZ Volcano Feed integration."""
from unittest.mock import AsyncMock, patch
from freezegun import freeze_time
from homeassistant.components import geonetnz_volcano
from homeassistant.components.geo_location import ATTR_DISTANCE
from homeassistant.components.geonetnz_volcano import DEFAULT_SCAN_INTERVAL
@ -29,7 +31,7 @@ from tests.components.geonetnz_volcano import _generate_mock_feed_entry
CONFIG = {geonetnz_volcano.DOMAIN: {CONF_RADIUS: 200}}
async def test_setup(hass, legacy_patchable_time):
async def test_setup(hass):
"""Test the general setup of the integration."""
# Set up some mock feed entries for this test.
mock_entry_1 = _generate_mock_feed_entry(
@ -48,7 +50,7 @@ async def test_setup(hass, legacy_patchable_time):
# Patching 'utcnow' to gain more control over the timed update.
utcnow = dt_util.utcnow()
with patch("homeassistant.util.dt.utcnow", return_value=utcnow), patch(
with freeze_time(utcnow), patch(
"aio_geojson_client.feed.GeoJsonFeed.update", new_callable=AsyncMock
) as mock_feed_update:
mock_feed_update.return_value = "OK", [mock_entry_1, mock_entry_2, mock_entry_3]

2
tests/components/google_assistant/test_report_state.py
View File

@ -11,7 +11,7 @@ from . import BASIC_CONFIG
from tests.common import async_fire_time_changed
async def test_report_state(hass, caplog, legacy_patchable_time):
async def test_report_state(hass, caplog):
"""Test report state works."""
assert await async_setup_component(hass, "switch", {})
hass.states.async_set("light.ceiling", "off")

4
tests/components/islamic_prayer_times/__init__.py
View File

@ -2,6 +2,8 @@
from datetime import datetime
import homeassistant.util.dt as dt_util
PRAYER_TIMES = {
"Fajr": "06:10",
"Sunrise": "07:25",
@ -42,4 +44,4 @@ NEW_PRAYER_TIMES_TIMESTAMPS = {
"Midnight": datetime(2020, 1, 1, 00, 43, 0),
}
NOW = datetime(2020, 1, 1, 00, 00, 0).astimezone()
NOW = datetime(2020, 1, 1, 00, 00, 0, tzinfo=dt_util.UTC)

24
tests/components/islamic_prayer_times/test_init.py
View File

@ -3,7 +3,9 @@
from datetime import timedelta
from unittest.mock import patch
from freezegun import freeze_time
from prayer_times_calculator.exceptions import InvalidResponseError
import pytest
from homeassistant import config_entries
from homeassistant.components import islamic_prayer_times
@ -20,7 +22,13 @@ from . import (
from tests.common import MockConfigEntry, async_fire_time_changed
async def test_setup_with_config(hass, legacy_patchable_time):
@pytest.fixture(autouse=True)
def set_utc(hass):
"""Set timezone to UTC."""
hass.config.set_time_zone("UTC")
async def test_setup_with_config(hass):
"""Test that we import the config and setup the client."""
config = {
islamic_prayer_times.DOMAIN: {islamic_prayer_times.CONF_CALC_METHOD: "isna"}
@ -36,7 +44,7 @@ async def test_setup_with_config(hass, legacy_patchable_time):
await hass.async_block_till_done()
async def test_successful_config_entry(hass, legacy_patchable_time):
async def test_successful_config_entry(hass):
"""Test that Islamic Prayer Times is configured successfully."""
entry = MockConfigEntry(
@ -58,7 +66,7 @@ async def test_successful_config_entry(hass, legacy_patchable_time):
}
async def test_setup_failed(hass, legacy_patchable_time):
async def test_setup_failed(hass):
"""Test Islamic Prayer Times failed due to an error."""
entry = MockConfigEntry(
@ -77,7 +85,7 @@ async def test_setup_failed(hass, legacy_patchable_time):
assert entry.state is config_entries.ConfigEntryState.SETUP_RETRY
async def test_unload_entry(hass, legacy_patchable_time):
async def test_unload_entry(hass):
"""Test removing Islamic Prayer Times."""
entry = MockConfigEntry(
domain=islamic_prayer_times.DOMAIN,
@ -97,7 +105,7 @@ async def test_unload_entry(hass, legacy_patchable_time):
assert islamic_prayer_times.DOMAIN not in hass.data
async def test_islamic_prayer_times_timestamp_format(hass, legacy_patchable_time):
async def test_islamic_prayer_times_timestamp_format(hass):
"""Test Islamic prayer times timestamp format."""
entry = MockConfigEntry(domain=islamic_prayer_times.DOMAIN, data={})
entry.add_to_hass(hass)
@ -105,7 +113,7 @@ async def test_islamic_prayer_times_timestamp_format(hass, legacy_patchable_time
with patch(
"prayer_times_calculator.PrayerTimesCalculator.fetch_prayer_times",
return_value=PRAYER_TIMES,
), patch("homeassistant.util.dt.now", return_value=NOW):
), freeze_time(NOW):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
@ -115,14 +123,14 @@ async def test_islamic_prayer_times_timestamp_format(hass, legacy_patchable_time
)
async def test_update(hass, legacy_patchable_time):
async def test_update(hass):
"""Test sensors are updated with new prayer times."""
entry = MockConfigEntry(domain=islamic_prayer_times.DOMAIN, data={})
entry.add_to_hass(hass)
with patch(
"prayer_times_calculator.PrayerTimesCalculator.fetch_prayer_times"
) as FetchPrayerTimes, patch("homeassistant.util.dt.now", return_value=NOW):
) as FetchPrayerTimes, freeze_time(NOW):
FetchPrayerTimes.side_effect = [
PRAYER_TIMES,
PRAYER_TIMES,

13
tests/components/islamic_prayer_times/test_sensor.py
View File

@ -1,6 +1,9 @@
"""The tests for the Islamic prayer times sensor platform."""
from unittest.mock import patch
from freezegun import freeze_time
import pytest
from homeassistant.components import islamic_prayer_times
import homeassistant.util.dt as dt_util
@ -9,7 +12,13 @@ from . import NOW, PRAYER_TIMES, PRAYER_TIMES_TIMESTAMPS
from tests.common import MockConfigEntry
async def test_islamic_prayer_times_sensors(hass, legacy_patchable_time):
@pytest.fixture(autouse=True)
def set_utc(hass):
"""Set timezone to UTC."""
hass.config.set_time_zone("UTC")
async def test_islamic_prayer_times_sensors(hass):
"""Test minimum Islamic prayer times configuration."""
entry = MockConfigEntry(domain=islamic_prayer_times.DOMAIN, data={})
entry.add_to_hass(hass)
@ -17,7 +26,7 @@ async def test_islamic_prayer_times_sensors(hass, legacy_patchable_time):
with patch(
"prayer_times_calculator.PrayerTimesCalculator.fetch_prayer_times",
return_value=PRAYER_TIMES,
), patch("homeassistant.util.dt.now", return_value=NOW):
), freeze_time(NOW):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()

15
tests/components/jewish_calendar/__init__.py
View File

@ -1,8 +1,8 @@
"""Tests for the jewish_calendar component."""
from collections import namedtuple
from contextlib import contextmanager
from datetime import datetime
from unittest.mock import patch
from freezegun import freeze_time as alter_time # noqa: F401
from homeassistant.components import jewish_calendar
import homeassistant.util.dt as dt_util
@ -56,14 +56,3 @@ def make_jerusalem_test_params(dtime, results, havdalah_offset=0):
JERUSALEM_LATLNG.lng,
results,
)
@contextmanager
def alter_time(local_time):
"""Manage multiple time mocks."""
utc_time = dt_util.as_utc(local_time)
patch1 = patch("homeassistant.util.dt.utcnow", return_value=utc_time)
patch2 = patch("homeassistant.util.dt.now", return_value=local_time)
with patch1, patch2:
yield

2
tests/components/jewish_calendar/test_binary_sensor.py
View File

@ -167,7 +167,6 @@ MELACHA_TEST_IDS = [
)
async def test_issur_melacha_sensor(
hass,
legacy_patchable_time,
now,
candle_lighting,
havdalah,
@ -258,7 +257,6 @@ async def test_issur_melacha_sensor(
)
async def test_issur_melacha_sensor_update(
hass,
legacy_patchable_time,
now,
candle_lighting,
havdalah,

6
tests/components/jewish_calendar/test_sensor.py
View File

@ -151,7 +151,6 @@ TEST_IDS = [
)
async def test_jewish_calendar_sensor(
hass,
legacy_patchable_time,
now,
tzname,
latitude,
@ -495,7 +494,6 @@ SHABBAT_TEST_IDS = [
)
async def test_shabbat_times_sensor(
hass,
legacy_patchable_time,
language,
now,
candle_lighting,
@ -592,7 +590,7 @@ OMER_TEST_IDS = [
@pytest.mark.parametrize(["test_time", "result"], OMER_PARAMS, ids=OMER_TEST_IDS)
async def test_omer_sensor(hass, legacy_patchable_time, test_time, result):
async def test_omer_sensor(hass, test_time, result):
"""Test Omer Count sensor output."""
test_time = test_time.replace(tzinfo=dt_util.get_time_zone(hass.config.time_zone))
@ -626,7 +624,7 @@ DAFYOMI_TEST_IDS = [
@pytest.mark.parametrize(["test_time", "result"], DAFYOMI_PARAMS, ids=DAFYOMI_TEST_IDS)
async def test_dafyomi_sensor(hass, legacy_patchable_time, test_time, result):
async def test_dafyomi_sensor(hass, test_time, result):
"""Test Daf Yomi sensor output."""
test_time = test_time.replace(tzinfo=dt_util.get_time_zone(hass.config.time_zone))

13
tests/components/manual/test_alarm_control_panel.py
View File

@ -2,6 +2,7 @@
from datetime import timedelta
from unittest.mock import MagicMock, patch
from freezegun import freeze_time
import pytest
from homeassistant.components import alarm_control_panel
@ -1099,7 +1100,7 @@ async def test_disarm_with_template_code(hass):
assert state.state == STATE_ALARM_DISARMED
async def test_arm_away_after_disabled_disarmed(hass, legacy_patchable_time):
async def test_arm_away_after_disabled_disarmed(hass):
"""Test pending state with and without zero trigger time."""
assert await async_setup_component(
hass,
@ -1138,10 +1139,7 @@ async def test_arm_away_after_disabled_disarmed(hass, legacy_patchable_time):
assert state.attributes["next_state"] == STATE_ALARM_ARMED_AWAY
future = dt_util.utcnow() + timedelta(seconds=1)
with patch(
("homeassistant.components.manual.alarm_control_panel.dt_util.utcnow"),
return_value=future,
):
with freeze_time(future):
async_fire_time_changed(hass, future)
await hass.async_block_till_done()
@ -1156,10 +1154,7 @@ async def test_arm_away_after_disabled_disarmed(hass, legacy_patchable_time):
assert state.attributes["next_state"] == STATE_ALARM_TRIGGERED
future += timedelta(seconds=1)
with patch(
("homeassistant.components.manual.alarm_control_panel.dt_util.utcnow"),
return_value=future,
):
with freeze_time(future):
async_fire_time_changed(hass, future)
await hass.async_block_till_done()

14
tests/components/manual_mqtt/test_alarm_control_panel.py
View File

@ -2,6 +2,8 @@
from datetime import timedelta
from unittest.mock import patch
from freezegun import freeze_time
from homeassistant.components import alarm_control_panel
from homeassistant.const import (
STATE_ALARM_ARMED_AWAY,
@ -1348,7 +1350,7 @@ async def test_trigger_with_specific_pending(hass, mqtt_mock):
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
async def test_arm_away_after_disabled_disarmed(hass, legacy_patchable_time, mqtt_mock):
async def test_arm_away_after_disabled_disarmed(hass, mqtt_mock):
"""Test pending state with and without zero trigger time."""
assert await async_setup_component(
hass,
@ -1391,10 +1393,7 @@ async def test_arm_away_after_disabled_disarmed(hass, legacy_patchable_time, mqt
assert state.attributes["post_pending_state"] == STATE_ALARM_ARMED_AWAY
future = dt_util.utcnow() + timedelta(seconds=1)
with patch(
("homeassistant.components.manual_mqtt.alarm_control_panel.dt_util.utcnow"),
return_value=future,
):
with freeze_time(future):
async_fire_time_changed(hass, future)
await hass.async_block_till_done()
@ -1410,10 +1409,7 @@ async def test_arm_away_after_disabled_disarmed(hass, legacy_patchable_time, mqt
assert state.attributes["post_pending_state"] == STATE_ALARM_TRIGGERED
future += timedelta(seconds=1)
with patch(
("homeassistant.components.manual_mqtt.alarm_control_panel.dt_util.utcnow"),
return_value=future,
):
with freeze_time(future):
async_fire_time_changed(hass, future)
await hass.async_block_till_done()

11
tests/components/metoffice/__init__.py
View File

@ -1,12 +1 @@
"""Tests for the metoffice component."""
import datetime
class NewDateTime(datetime.datetime):
"""Patch time to a specific point."""
@classmethod
def now(cls, *args, **kwargs): # pylint: disable=signature-differs
"""Overload datetime.datetime.now."""
return cls(2020, 4, 25, 12, tzinfo=datetime.timezone.utc)

19
tests/components/metoffice/test_sensor.py
View File

@ -1,11 +1,12 @@
"""The tests for the Met Office sensor component."""
import datetime
import json
from unittest.mock import patch
from freezegun import freeze_time
from homeassistant.components.metoffice.const import ATTRIBUTION, DOMAIN
from homeassistant.helpers.device_registry import async_get as get_dev_reg
from . import NewDateTime
from .const import (
DEVICE_KEY_KINGSLYNN,
DEVICE_KEY_WAVERTREE,
@ -21,11 +22,8 @@ from .const import (
from tests.common import MockConfigEntry, load_fixture
@patch(
"datapoint.Forecast.datetime.datetime",
NewDateTime,
)
async def test_one_sensor_site_running(hass, requests_mock, legacy_patchable_time):
@freeze_time(datetime.datetime(2020, 4, 25, 12, tzinfo=datetime.timezone.utc))
async def test_one_sensor_site_running(hass, requests_mock):
"""Test the Met Office sensor platform."""
# all metoffice test data encapsulated in here
mock_json = json.loads(load_fixture("metoffice.json"))
@ -70,11 +68,8 @@ async def test_one_sensor_site_running(hass, requests_mock, legacy_patchable_tim
assert sensor.attributes.get("attribution") == ATTRIBUTION
@patch(
"datapoint.Forecast.datetime.datetime",
NewDateTime,
)
async def test_two_sensor_sites_running(hass, requests_mock, legacy_patchable_time):
@freeze_time(datetime.datetime(2020, 4, 25, 12, tzinfo=datetime.timezone.utc))
async def test_two_sensor_sites_running(hass, requests_mock):
"""Test we handle two sets of sensors running for two different sites."""
# all metoffice test data encapsulated in here

33
tests/components/metoffice/test_weather.py
View File

@ -1,14 +1,15 @@
"""The tests for the Met Office sensor component."""
import datetime
from datetime import timedelta
import json
from unittest.mock import patch
from freezegun import freeze_time
from homeassistant.components.metoffice.const import DOMAIN
from homeassistant.const import STATE_UNAVAILABLE
from homeassistant.helpers.device_registry import async_get as get_dev_reg
from homeassistant.util import utcnow
from . import NewDateTime
from .const import (
DEVICE_KEY_KINGSLYNN,
DEVICE_KEY_WAVERTREE,
@ -20,11 +21,8 @@ from .const import (
from tests.common import MockConfigEntry, async_fire_time_changed, load_fixture
@patch(
"datapoint.Forecast.datetime.datetime",
NewDateTime,
)
async def test_site_cannot_connect(hass, requests_mock, legacy_patchable_time):
@freeze_time(datetime.datetime(2020, 4, 25, 12, tzinfo=datetime.timezone.utc))
async def test_site_cannot_connect(hass, requests_mock):
"""Test we handle cannot connect error."""
requests_mock.get("/public/data/val/wxfcs/all/json/sitelist/", text="")
@ -50,11 +48,8 @@ async def test_site_cannot_connect(hass, requests_mock, legacy_patchable_time):
assert sensor is None
@patch(
"datapoint.Forecast.datetime.datetime",
NewDateTime,
)
async def test_site_cannot_update(hass, requests_mock, legacy_patchable_time):
@freeze_time(datetime.datetime(2020, 4, 25, 12, tzinfo=datetime.timezone.utc))
async def test_site_cannot_update(hass, requests_mock):
"""Test we handle cannot connect error."""
# all metoffice test data encapsulated in here
@ -99,11 +94,8 @@ async def test_site_cannot_update(hass, requests_mock, legacy_patchable_time):
assert weather.state == STATE_UNAVAILABLE
@patch(
"datapoint.Forecast.datetime.datetime",
NewDateTime,
)
async def test_one_weather_site_running(hass, requests_mock, legacy_patchable_time):
@freeze_time(datetime.datetime(2020, 4, 25, 12, tzinfo=datetime.timezone.utc))
async def test_one_weather_site_running(hass, requests_mock):
"""Test the Met Office weather platform."""
# all metoffice test data encapsulated in here
@ -183,11 +175,8 @@ async def test_one_weather_site_running(hass, requests_mock, legacy_patchable_ti
assert weather.attributes.get("forecast")[7]["wind_bearing"] == "SE"
@patch(
"datapoint.Forecast.datetime.datetime",
NewDateTime,
)
async def test_two_weather_sites_running(hass, requests_mock, legacy_patchable_time):
@freeze_time(datetime.datetime(2020, 4, 25, 12, tzinfo=datetime.timezone.utc))
async def test_two_weather_sites_running(hass, requests_mock):
"""Test we handle two different weather sites both running."""
# all metoffice test data encapsulated in here

4
tests/components/mikrotik/test_device_tracker.py
View File

@ -60,9 +60,7 @@ async def test_platform_manually_configured(hass):
assert mikrotik.DOMAIN not in hass.data
async def test_device_trackers(
hass, legacy_patchable_time, mock_device_registry_devices
):
async def test_device_trackers(hass, mock_device_registry_devices):
"""Test device_trackers created by mikrotik."""
# test devices are added from wireless list only

12
tests/components/pvpc_hourly_pricing/test_config_flow.py
View File

@ -1,6 +1,7 @@
"""Tests for the pvpc_hourly_pricing config_flow."""
from datetime import datetime
from unittest.mock import patch
from freezegun import freeze_time
from homeassistant import config_entries, data_entry_flow
from homeassistant.components.pvpc_hourly_pricing import (
@ -19,9 +20,7 @@ from tests.common import date_util
from tests.test_util.aiohttp import AiohttpClientMocker
async def test_config_flow(
hass, legacy_patchable_time, pvpc_aioclient_mock: AiohttpClientMocker
):
async def test_config_flow(hass, pvpc_aioclient_mock: AiohttpClientMocker):
"""
Test config flow for pvpc_hourly_pricing.
@ -40,10 +39,7 @@ async def test_config_flow(
}
mock_data = {"return_time": datetime(2021, 6, 1, 12, 0, tzinfo=date_util.UTC)}
def mock_now():
return mock_data["return_time"]
with patch("homeassistant.util.dt.utcnow", new=mock_now):
with freeze_time(mock_data["return_time"]):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)

18
tests/components/pvpc_hourly_pricing/test_sensor.py
View File

@ -1,7 +1,8 @@
"""Tests for the pvpc_hourly_pricing sensor component."""
from datetime import datetime, timedelta
import logging
from unittest.mock import patch
from freezegun import freeze_time
from homeassistant.components.pvpc_hourly_pricing import (
ATTR_POWER,
@ -24,7 +25,7 @@ from tests.test_util.aiohttp import AiohttpClientMocker
async def test_multi_sensor_migration(
hass, caplog, legacy_patchable_time, pvpc_aioclient_mock: AiohttpClientMocker
hass, caplog, pvpc_aioclient_mock: AiohttpClientMocker
):
"""Test tariff migration when there are >1 old sensors."""
entity_reg = mock_registry(hass)
@ -59,14 +60,15 @@ async def test_multi_sensor_migration(
mock_data = {"return_time": datetime(2021, 6, 1, 21, tzinfo=date_util.UTC)}
def mock_now():
return mock_data["return_time"]
caplog.clear()
with caplog.at_level(logging.WARNING):
with patch("homeassistant.util.dt.utcnow", new=mock_now):
with freeze_time(mock_data["return_time"]):
assert await hass.config_entries.async_setup(config_entry_1.entry_id)
assert len(caplog.messages) == 2
assert any("Migrating PVPC" in message for message in caplog.messages)
assert any(
"Old PVPC Sensor sensor.test_pvpc_2 is removed" in message
for message in caplog.messages
)
# check migration with removal of extra sensors
assert len(entity_reg.entities) == 1
@ -87,7 +89,7 @@ async def test_multi_sensor_migration(
state = hass.states.get("sensor.test_pvpc_1")
check_valid_state(state, tariff=TARIFFS[0], value=0.1565)
mock_data["return_time"] += timedelta(minutes=60)
with freeze_time(mock_data["return_time"] + timedelta(minutes=60)):
async_fire_time_changed(hass, mock_data["return_time"])
await list(hass.data[DOMAIN].values())[0].async_refresh()
await hass.async_block_till_done()

13
tests/components/rflink/test_binary_sensor.py
View File

@ -5,7 +5,8 @@ Test setup of rflink sensor component/platform. Verify manual and
automatic sensor creation.
"""
from datetime import timedelta
from unittest.mock import patch
from freezegun import freeze_time
from homeassistant.components.rflink import CONF_RECONNECT_INTERVAL
from homeassistant.const import (
@ -123,7 +124,7 @@ async def test_entity_availability(hass, monkeypatch):
assert hass.states.get("binary_sensor.test").state == STATE_ON
async def test_off_delay(hass, legacy_patchable_time, monkeypatch):
async def test_off_delay(hass, monkeypatch):
"""Test off_delay option."""
# setup mocking rflink module
event_callback, create, _, _ = await mock_rflink(hass, CONFIG, DOMAIN, monkeypatch)
@ -145,7 +146,7 @@ async def test_off_delay(hass, legacy_patchable_time, monkeypatch):
now = dt_util.utcnow()
# fake time and turn on sensor
future = now + timedelta(seconds=0)
with patch(("homeassistant.helpers.event.dt_util.utcnow"), return_value=future):
with freeze_time(future):
async_fire_time_changed(hass, future)
event_callback(on_event)
await hass.async_block_till_done()
@ -156,7 +157,7 @@ async def test_off_delay(hass, legacy_patchable_time, monkeypatch):
# fake time and turn on sensor again
future = now + timedelta(seconds=15)
with patch(("homeassistant.helpers.event.dt_util.utcnow"), return_value=future):
with freeze_time(future):
async_fire_time_changed(hass, future)
event_callback(on_event)
await hass.async_block_till_done()
@ -167,7 +168,7 @@ async def test_off_delay(hass, legacy_patchable_time, monkeypatch):
# fake time and verify sensor still on (de-bounce)
future = now + timedelta(seconds=35)
with patch(("homeassistant.helpers.event.dt_util.utcnow"), return_value=future):
with freeze_time(future):
async_fire_time_changed(hass, future)
await hass.async_block_till_done()
await hass.async_block_till_done()
@ -177,7 +178,7 @@ async def test_off_delay(hass, legacy_patchable_time, monkeypatch):
# fake time and verify sensor is off
future = now + timedelta(seconds=45)
with patch(("homeassistant.helpers.event.dt_util.utcnow"), return_value=future):
with freeze_time(future):
async_fire_time_changed(hass, future)
await hass.async_block_till_done()
await hass.async_block_till_done()

39
tests/components/sun/test_init.py
View File

@ -2,6 +2,7 @@
from datetime import datetime, timedelta
from unittest.mock import patch
from freezegun import freeze_time
from pytest import mark
import homeassistant.components.sun as sun
@ -10,13 +11,13 @@ import homeassistant.core as ha
from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util
from tests.common import MockConfigEntry
from tests.common import MockConfigEntry, async_fire_time_changed
async def test_setting_rising(hass, legacy_patchable_time):
async def test_setting_rising(hass):
"""Test retrieving sun setting and rising."""
utc_now = datetime(2016, 11, 1, 8, 0, 0, tzinfo=dt_util.UTC)
with patch("homeassistant.helpers.condition.dt_util.utcnow", return_value=utc_now):
with freeze_time(utc_now):
await async_setup_component(
hass, sun.DOMAIN, {sun.DOMAIN: {sun.CONF_ELEVATION: 0}}
)
@ -107,10 +108,10 @@ async def test_setting_rising(hass, legacy_patchable_time):
)
async def test_state_change(hass, legacy_patchable_time, caplog):
async def test_state_change(hass, caplog):
"""Test if the state changes at next setting/rising."""
now = datetime(2016, 6, 1, 8, 0, 0, tzinfo=dt_util.UTC)
with patch("homeassistant.helpers.condition.dt_util.utcnow", return_value=now):
with freeze_time(now):
await async_setup_component(
hass, sun.DOMAIN, {sun.DOMAIN: {sun.CONF_ELEVATION: 0}}
)
@ -125,10 +126,8 @@ async def test_state_change(hass, legacy_patchable_time, caplog):
assert sun.STATE_BELOW_HORIZON == hass.states.get(sun.ENTITY_ID).state
patched_time = test_time + timedelta(seconds=5)
with patch(
"homeassistant.helpers.condition.dt_util.utcnow", return_value=patched_time
):
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: patched_time})
with freeze_time(patched_time):
async_fire_time_changed(hass, patched_time)
await hass.async_block_till_done()
assert sun.STATE_ABOVE_HORIZON == hass.states.get(sun.ENTITY_ID).state
@ -148,10 +147,8 @@ async def test_state_change(hass, legacy_patchable_time, caplog):
patched_time = test_time + timedelta(seconds=5)
caplog.clear()
with patch(
"homeassistant.helpers.condition.dt_util.utcnow", return_value=patched_time
):
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: patched_time})
with freeze_time(patched_time):
async_fire_time_changed(hass, patched_time)
await hass.async_block_till_done()
await hass.async_block_till_done()
@ -197,7 +194,7 @@ async def test_state_change_count(hass):
now = datetime(2016, 6, 1, tzinfo=dt_util.UTC)
with patch("homeassistant.helpers.condition.dt_util.utcnow", return_value=now):
with freeze_time(now):
assert await async_setup_component(
hass, sun.DOMAIN, {sun.DOMAIN: {sun.CONF_ELEVATION: 0}}
)
@ -214,21 +211,19 @@ async def test_state_change_count(hass):
for _ in range(24 * 60 * 60):
now += timedelta(seconds=1)
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: now})
async_fire_time_changed(hass, now)
await hass.async_block_till_done()
assert len(events) < 721
async def test_setup_and_remove_config_entry(
hass: ha.HomeAssistant, legacy_patchable_time
) -> None:
async def test_setup_and_remove_config_entry(hass: ha.HomeAssistant) -> None:
"""Test setting up and removing a config entry."""
# Setup the config entry
config_entry = MockConfigEntry(domain=sun.DOMAIN)
config_entry.add_to_hass(hass)
now = datetime(2016, 6, 1, 8, 0, 0, tzinfo=dt_util.UTC)
with patch("homeassistant.helpers.condition.dt_util.utcnow", return_value=now):
with freeze_time(now):
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
@ -250,10 +245,8 @@ async def test_setup_and_remove_config_entry(
assert hass.states.get("sun.sun") is None
patched_time = test_time + timedelta(seconds=5)
with patch(
"homeassistant.helpers.condition.dt_util.utcnow", return_value=patched_time
):
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: patched_time})
with freeze_time(patched_time):
async_fire_time_changed(hass, patched_time)
await hass.async_block_till_done()
assert hass.states.get("sun.sun") is None

66
tests/components/sun/test_trigger.py
View File

@ -1,7 +1,7 @@
"""The tests for the sun automation."""
from datetime import datetime
from unittest.mock import patch
from freezegun import freeze_time
import pytest
from homeassistant.components import sun
@ -36,12 +36,12 @@ def setup_comp(hass):
)
async def test_sunset_trigger(hass, calls, legacy_patchable_time):
async def test_sunset_trigger(hass, calls):
"""Test the sunset trigger."""
now = datetime(2015, 9, 15, 23, tzinfo=dt_util.UTC)
trigger_time = datetime(2015, 9, 16, 2, tzinfo=dt_util.UTC)
with patch("homeassistant.util.dt.utcnow", return_value=now):
with freeze_time(now):
await async_setup_component(
hass,
automation.DOMAIN,
@ -56,18 +56,18 @@ async def test_sunset_trigger(hass, calls, legacy_patchable_time):
},
)
await hass.services.async_call(
automation.DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: ENTITY_MATCH_ALL},
blocking=True,
)
await hass.services.async_call(
automation.DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: ENTITY_MATCH_ALL},
blocking=True,
)
async_fire_time_changed(hass, trigger_time)
await hass.async_block_till_done()
assert len(calls) == 0
async_fire_time_changed(hass, trigger_time)
await hass.async_block_till_done()
assert len(calls) == 0
with patch("homeassistant.util.dt.utcnow", return_value=now):
with freeze_time(now):
await hass.services.async_call(
automation.DOMAIN,
SERVICE_TURN_ON,
@ -75,18 +75,18 @@ async def test_sunset_trigger(hass, calls, legacy_patchable_time):
blocking=True,
)
async_fire_time_changed(hass, trigger_time)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["id"] == 0
async_fire_time_changed(hass, trigger_time)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["id"] == 0
async def test_sunrise_trigger(hass, calls, legacy_patchable_time):
async def test_sunrise_trigger(hass, calls):
"""Test the sunrise trigger."""
now = datetime(2015, 9, 13, 23, tzinfo=dt_util.UTC)
trigger_time = datetime(2015, 9, 16, 14, tzinfo=dt_util.UTC)
with patch("homeassistant.util.dt.utcnow", return_value=now):
with freeze_time(now):
await async_setup_component(
hass,
automation.DOMAIN,
@ -98,17 +98,17 @@ async def test_sunrise_trigger(hass, calls, legacy_patchable_time):
},
)
async_fire_time_changed(hass, trigger_time)
await hass.async_block_till_done()
assert len(calls) == 1
async_fire_time_changed(hass, trigger_time)
await hass.async_block_till_done()
assert len(calls) == 1
async def test_sunset_trigger_with_offset(hass, calls, legacy_patchable_time):
async def test_sunset_trigger_with_offset(hass, calls):
"""Test the sunset trigger with offset."""
now = datetime(2015, 9, 15, 23, tzinfo=dt_util.UTC)
trigger_time = datetime(2015, 9, 16, 2, 30, tzinfo=dt_util.UTC)
with patch("homeassistant.util.dt.utcnow", return_value=now):
with freeze_time(now):
await async_setup_component(
hass,
automation.DOMAIN,
@ -130,18 +130,18 @@ async def test_sunset_trigger_with_offset(hass, calls, legacy_patchable_time):
},
)
async_fire_time_changed(hass, trigger_time)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["some"] == "sun - sunset - 0:30:00"
async_fire_time_changed(hass, trigger_time)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["some"] == "sun - sunset - 0:30:00"
async def test_sunrise_trigger_with_offset(hass, calls, legacy_patchable_time):
async def test_sunrise_trigger_with_offset(hass, calls):
"""Test the sunrise trigger with offset."""
now = datetime(2015, 9, 13, 23, tzinfo=dt_util.UTC)
trigger_time = datetime(2015, 9, 16, 13, 30, tzinfo=dt_util.UTC)
with patch("homeassistant.util.dt.utcnow", return_value=now):
with freeze_time(now):
await async_setup_component(
hass,
automation.DOMAIN,
@ -157,6 +157,6 @@ async def test_sunrise_trigger_with_offset(hass, calls, legacy_patchable_time):
},
)
async_fire_time_changed(hass, trigger_time)
await hass.async_block_till_done()
assert len(calls) == 1
async_fire_time_changed(hass, trigger_time)
await hass.async_block_till_done()
assert len(calls) == 1

82
tests/components/tod/test_binary_sensor.py
View File

@ -5,18 +5,11 @@ from freezegun import freeze_time
import pytest
from homeassistant.const import STATE_OFF, STATE_ON
import homeassistant.core as ha
from homeassistant.helpers.sun import get_astral_event_date, get_astral_event_next
from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util
from tests.common import assert_setup_component
@pytest.fixture(autouse=True)
def mock_legacy_time(legacy_patchable_time):
"""Make time patchable for all the tests."""
yield
from tests.common import assert_setup_component, async_fire_time_changed
@pytest.fixture
@ -126,7 +119,7 @@ async def test_midnight_turnover_after_midnight_inside_period(
await hass.async_block_till_done()
freezer.move_to(test_time + timedelta(hours=1))
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: dt_util.utcnow()})
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done()
state = hass.states.get("binary_sensor.night")
@ -184,15 +177,14 @@ async def test_midnight_turnover_after_midnight_outside_period(
switchover_time = datetime(2019, 1, 11, 4, 59, 0, tzinfo=hass_tz_info)
freezer.move_to(switchover_time)
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: dt_util.utcnow()})
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done()
state = hass.states.get("binary_sensor.night")
assert state.state == STATE_ON
freezer.move_to(switchover_time + timedelta(minutes=1, seconds=1))
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: dt_util.utcnow()})
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done()
state = hass.states.get("binary_sensor.night")
assert state.state == STATE_OFF
@ -225,31 +217,31 @@ async def test_from_sunrise_to_sunset(hass, freezer, hass_tz_info):
assert state.state == STATE_OFF
freezer.move_to(sunrise)
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: dt_util.utcnow()})
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_ON
freezer.move_to(sunrise + timedelta(seconds=1))
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: dt_util.utcnow()})
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_ON
freezer.move_to(sunset + timedelta(seconds=-1))
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: dt_util.utcnow()})
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_ON
freezer.move_to(sunset)
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: dt_util.utcnow()})
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_OFF
freezer.move_to(sunset + timedelta(seconds=1))
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: dt_util.utcnow()})
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_OFF
@ -279,31 +271,31 @@ async def test_from_sunset_to_sunrise(hass, freezer, hass_tz_info):
assert state.state == STATE_OFF
freezer.move_to(sunset)
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: dt_util.utcnow()})
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_ON
freezer.move_to(sunset + timedelta(minutes=1))
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: dt_util.utcnow()})
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_ON
freezer.move_to(sunrise + timedelta(minutes=-1))
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: dt_util.utcnow()})
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_ON
freezer.move_to(sunrise)
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: dt_util.utcnow()})
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_OFF
freezer.move_to(sunrise + timedelta(minutes=1))
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: dt_util.utcnow()})
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_OFF
@ -339,25 +331,25 @@ async def test_offset(hass, freezer, hass_tz_info):
assert state.state == STATE_OFF
freezer.move_to(after)
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: dt_util.utcnow()})
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_ON
freezer.move_to(before + timedelta(seconds=-1))
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: dt_util.utcnow()})
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_ON
freezer.move_to(before)
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: dt_util.utcnow()})
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_OFF
freezer.move_to(before + timedelta(seconds=1))
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: dt_util.utcnow()})
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_OFF
@ -388,7 +380,7 @@ async def test_offset_overnight(hass, freezer, hass_tz_info):
assert state.state == STATE_OFF
freezer.move_to(after)
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: dt_util.utcnow()})
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_ON
@ -424,37 +416,37 @@ async def test_norwegian_case_winter(hass, freezer, hass_tz_info):
assert state.state == STATE_OFF
freezer.move_to(sunrise + timedelta(seconds=-1))
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: dt_util.utcnow()})
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_OFF
freezer.move_to(sunrise)
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: dt_util.utcnow()})
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_ON
freezer.move_to(sunrise + timedelta(seconds=1))
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: dt_util.utcnow()})
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_ON
freezer.move_to(sunset + timedelta(seconds=-1))
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: dt_util.utcnow()})
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_ON
freezer.move_to(sunset)
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: dt_util.utcnow()})
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_OFF
freezer.move_to(sunset + timedelta(seconds=1))
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: dt_util.utcnow()})
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_OFF
@ -492,37 +484,37 @@ async def test_norwegian_case_summer(hass, freezer, hass_tz_info):
assert state.state == STATE_OFF
freezer.move_to(sunrise + timedelta(seconds=-1))
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: dt_util.utcnow()})
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_OFF
freezer.move_to(sunrise)
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: dt_util.utcnow()})
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_ON
freezer.move_to(sunrise + timedelta(seconds=1))
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: dt_util.utcnow()})
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_ON
freezer.move_to(sunset + timedelta(seconds=-1))
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: dt_util.utcnow()})
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_ON
freezer.move_to(sunset)
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: dt_util.utcnow()})
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_OFF
freezer.move_to(sunset + timedelta(seconds=1))
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: dt_util.utcnow()})
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_OFF
@ -559,19 +551,19 @@ async def test_sun_offset(hass, freezer, hass_tz_info):
assert state.state == STATE_OFF
freezer.move_to(sunrise)
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: dt_util.utcnow()})
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_ON
freezer.move_to(sunrise + timedelta(seconds=1))
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: dt_util.utcnow()})
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_ON
freezer.move_to(sunset + timedelta(seconds=-1))
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: dt_util.utcnow()})
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_ON
@ -579,13 +571,13 @@ async def test_sun_offset(hass, freezer, hass_tz_info):
await hass.async_block_till_done()
freezer.move_to(sunset)
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: dt_util.utcnow()})
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_OFF
freezer.move_to(sunset + timedelta(seconds=1))
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: dt_util.utcnow()})
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_OFF
@ -596,7 +588,7 @@ async def test_sun_offset(hass, freezer, hass_tz_info):
+ timedelta(hours=-1, minutes=-30)
)
freezer.move_to(sunrise)
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: dt_util.utcnow()})
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_ON

10
tests/components/utility_meter/test_init.py
View File

@ -314,7 +314,7 @@ async def test_services_config_entry(hass):
assert state.state == "4"
async def test_cron(hass, legacy_patchable_time):
async def test_cron(hass):
"""Test cron pattern."""
config = {
@ -329,7 +329,7 @@ async def test_cron(hass, legacy_patchable_time):
assert await async_setup_component(hass, DOMAIN, config)
async def test_cron_and_meter(hass, legacy_patchable_time):
async def test_cron_and_meter(hass):
"""Test cron pattern and meter type fails."""
config = {
"utility_meter": {
@ -344,7 +344,7 @@ async def test_cron_and_meter(hass, legacy_patchable_time):
assert not await async_setup_component(hass, DOMAIN, config)
async def test_both_cron_and_meter(hass, legacy_patchable_time):
async def test_both_cron_and_meter(hass):
"""Test cron pattern and meter type passes in different meter."""
config = {
"utility_meter": {
@ -362,7 +362,7 @@ async def test_both_cron_and_meter(hass, legacy_patchable_time):
assert await async_setup_component(hass, DOMAIN, config)
async def test_cron_and_offset(hass, legacy_patchable_time):
async def test_cron_and_offset(hass):
"""Test cron pattern and offset fails."""
config = {
@ -378,7 +378,7 @@ async def test_cron_and_offset(hass, legacy_patchable_time):
assert not await async_setup_component(hass, DOMAIN, config)
async def test_bad_cron(hass, legacy_patchable_time):
async def test_bad_cron(hass):
"""Test bad cron pattern."""
config = {

64
tests/components/utility_meter/test_sensor.py
View File

@ -1,8 +1,8 @@
"""The tests for the utility_meter sensor platform."""
from contextlib import contextmanager
from datetime import timedelta
from unittest.mock import patch
from freezegun import freeze_time
import pytest
from homeassistant.components.select.const import (
@ -44,14 +44,10 @@ import homeassistant.util.dt as dt_util
from tests.common import MockConfigEntry, async_fire_time_changed, mock_restore_cache
@contextmanager
def alter_time(retval):
"""Manage multiple time mocks."""
patch1 = patch("homeassistant.util.dt.utcnow", return_value=retval)
patch2 = patch("homeassistant.util.dt.now", return_value=retval)
with patch1, patch2:
yield
@pytest.fixture(autouse=True)
def set_utc(hass):
"""Set timezone to UTC."""
hass.config.set_time_zone("UTC")
@pytest.mark.parametrize(
@ -696,7 +692,7 @@ async def test_delta_values(hass, yaml_config, config_entry_config, caplog):
hass.state = CoreState.not_running
now = dt_util.utcnow()
with alter_time(now):
with freeze_time(now):
if yaml_config:
assert await async_setup_component(hass, DOMAIN, yaml_config)
await hass.async_block_till_done()
@ -725,7 +721,7 @@ async def test_delta_values(hass, yaml_config, config_entry_config, caplog):
assert state.attributes.get("status") == PAUSED
now += timedelta(seconds=30)
with alter_time(now):
with freeze_time(now):
async_fire_time_changed(hass, now)
hass.states.async_set(
entity_id,
@ -737,7 +733,7 @@ async def test_delta_values(hass, yaml_config, config_entry_config, caplog):
assert "Invalid adjustment of None" in caplog.text
now += timedelta(seconds=30)
with alter_time(now):
with freeze_time(now):
async_fire_time_changed(hass, now)
hass.states.async_set(
entity_id,
@ -751,7 +747,7 @@ async def test_delta_values(hass, yaml_config, config_entry_config, caplog):
assert state.attributes.get("status") == COLLECTING
now += timedelta(seconds=30)
with alter_time(now):
with freeze_time(now):
async_fire_time_changed(hass, now)
await hass.async_block_till_done()
hass.states.async_set(
@ -785,7 +781,7 @@ def gen_config(cycle, offset=None):
async def _test_self_reset(hass, config, start_time, expect_reset=True):
"""Test energy sensor self reset."""
now = dt_util.parse_datetime(start_time)
with alter_time(now):
with freeze_time(now):
assert await async_setup_component(hass, DOMAIN, config)
await hass.async_block_till_done()
@ -799,7 +795,7 @@ async def _test_self_reset(hass, config, start_time, expect_reset=True):
await hass.async_block_till_done()
now += timedelta(seconds=30)
with alter_time(now):
with freeze_time(now):
async_fire_time_changed(hass, now)
hass.states.async_set(
entity_id,
@ -810,7 +806,7 @@ async def _test_self_reset(hass, config, start_time, expect_reset=True):
await hass.async_block_till_done()
now += timedelta(seconds=30)
with alter_time(now):
with freeze_time(now):
async_fire_time_changed(hass, now)
await hass.async_block_till_done()
hass.states.async_set(
@ -841,7 +837,7 @@ async def _test_self_reset(hass, config, start_time, expect_reset=True):
now += timedelta(minutes=5)
else:
now += timedelta(days=5)
with alter_time(now):
with freeze_time(now):
async_fire_time_changed(hass, now)
await hass.async_block_till_done()
hass.states.async_set(
@ -860,7 +856,7 @@ async def _test_self_reset(hass, config, start_time, expect_reset=True):
assert state.state == "9"
async def test_self_reset_cron_pattern(hass, legacy_patchable_time):
async def test_self_reset_cron_pattern(hass):
"""Test cron pattern reset of meter."""
config = {
"utility_meter": {
@ -871,70 +867,70 @@ async def test_self_reset_cron_pattern(hass, legacy_patchable_time):
await _test_self_reset(hass, config, "2017-01-31T23:59:00.000000+00:00")
async def test_self_reset_quarter_hourly(hass, legacy_patchable_time):
async def test_self_reset_quarter_hourly(hass):
"""Test quarter-hourly reset of meter."""
await _test_self_reset(
hass, gen_config("quarter-hourly"), "2017-12-31T23:59:00.000000+00:00"
)
async def test_self_reset_quarter_hourly_first_quarter(hass, legacy_patchable_time):
async def test_self_reset_quarter_hourly_first_quarter(hass):
"""Test quarter-hourly reset of meter."""
await _test_self_reset(
hass, gen_config("quarter-hourly"), "2017-12-31T23:14:00.000000+00:00"
)
async def test_self_reset_quarter_hourly_second_quarter(hass, legacy_patchable_time):
async def test_self_reset_quarter_hourly_second_quarter(hass):
"""Test quarter-hourly reset of meter."""
await _test_self_reset(
hass, gen_config("quarter-hourly"), "2017-12-31T23:29:00.000000+00:00"
)
async def test_self_reset_quarter_hourly_third_quarter(hass, legacy_patchable_time):
async def test_self_reset_quarter_hourly_third_quarter(hass):
"""Test quarter-hourly reset of meter."""
await _test_self_reset(
hass, gen_config("quarter-hourly"), "2017-12-31T23:44:00.000000+00:00"
)
async def test_self_reset_hourly(hass, legacy_patchable_time):
async def test_self_reset_hourly(hass):
"""Test hourly reset of meter."""
await _test_self_reset(
hass, gen_config("hourly"), "2017-12-31T23:59:00.000000+00:00"
)
async def test_self_reset_daily(hass, legacy_patchable_time):
async def test_self_reset_daily(hass):
"""Test daily reset of meter."""
await _test_self_reset(
hass, gen_config("daily"), "2017-12-31T23:59:00.000000+00:00"
)
async def test_self_reset_weekly(hass, legacy_patchable_time):
async def test_self_reset_weekly(hass):
"""Test weekly reset of meter."""
await _test_self_reset(
hass, gen_config("weekly"), "2017-12-31T23:59:00.000000+00:00"
)
async def test_self_reset_monthly(hass, legacy_patchable_time):
async def test_self_reset_monthly(hass):
"""Test monthly reset of meter."""
await _test_self_reset(
hass, gen_config("monthly"), "2017-12-31T23:59:00.000000+00:00"
)
async def test_self_reset_bimonthly(hass, legacy_patchable_time):
async def test_self_reset_bimonthly(hass):
"""Test bimonthly reset of meter occurs on even months."""
await _test_self_reset(
hass, gen_config("bimonthly"), "2017-12-31T23:59:00.000000+00:00"
)
async def test_self_no_reset_bimonthly(hass, legacy_patchable_time):
async def test_self_no_reset_bimonthly(hass):
"""Test bimonthly reset of meter does not occur on odd months."""
await _test_self_reset(
hass,
@ -944,21 +940,21 @@ async def test_self_no_reset_bimonthly(hass, legacy_patchable_time):
)
async def test_self_reset_quarterly(hass, legacy_patchable_time):
async def test_self_reset_quarterly(hass):
"""Test quarterly reset of meter."""
await _test_self_reset(
hass, gen_config("quarterly"), "2017-03-31T23:59:00.000000+00:00"
)
async def test_self_reset_yearly(hass, legacy_patchable_time):
async def test_self_reset_yearly(hass):
"""Test yearly reset of meter."""
await _test_self_reset(
hass, gen_config("yearly"), "2017-12-31T23:59:00.000000+00:00"
)
async def test_self_no_reset_yearly(hass, legacy_patchable_time):
async def test_self_no_reset_yearly(hass):
"""Test yearly reset of meter does not occur after 1st January."""
await _test_self_reset(
hass,
@ -968,7 +964,7 @@ async def test_self_no_reset_yearly(hass, legacy_patchable_time):
)
async def test_reset_yearly_offset(hass, legacy_patchable_time):
async def test_reset_yearly_offset(hass):
"""Test yearly reset of meter."""
await _test_self_reset(
hass,
@ -977,7 +973,7 @@ async def test_reset_yearly_offset(hass, legacy_patchable_time):
)
async def test_no_reset_yearly_offset(hass, legacy_patchable_time):
async def test_no_reset_yearly_offset(hass):
"""Test yearly reset of meter."""
await _test_self_reset(
hass,
@ -987,7 +983,7 @@ async def test_no_reset_yearly_offset(hass, legacy_patchable_time):
)
async def test_bad_offset(hass, legacy_patchable_time):
async def test_bad_offset(hass):
"""Test bad offset of meter."""
assert not await async_setup_component(
hass, DOMAIN, gen_config("monthly", timedelta(days=31))

114
tests/conftest.py
View File

@ -1,6 +1,5 @@
"""Set up some common test helper things."""
import asyncio
import datetime
import functools
import logging
import socket
@ -26,8 +25,8 @@ from homeassistant.components.websocket_api.auth import (
TYPE_AUTH_REQUIRED,
)
from homeassistant.components.websocket_api.http import URL
from homeassistant.const import ATTR_NOW, EVENT_TIME_CHANGED, HASSIO_USER_NAME
from homeassistant.helpers import config_entry_oauth2_flow, event
from homeassistant.const import HASSIO_USER_NAME
from homeassistant.helpers import config_entry_oauth2_flow
from homeassistant.setup import async_setup_component
from homeassistant.util import dt as dt_util, location
@ -661,115 +660,6 @@ def mock_async_zeroconf(mock_zeroconf):
yield zc
@pytest.fixture
def legacy_patchable_time():
"""Allow time to be patchable by using event listeners instead of asyncio loop."""
@ha.callback
@loader.bind_hass
def async_track_point_in_utc_time(hass, action, point_in_time):
"""Add a listener that fires once after a specific point in UTC time."""
# Ensure point_in_time is UTC
point_in_time = event.dt_util.as_utc(point_in_time)
# Since this is called once, we accept a HassJob so we can avoid
# having to figure out how to call the action every time its called.
job = action if isinstance(action, ha.HassJob) else ha.HassJob(action)
@ha.callback
def point_in_time_listener(event):
"""Listen for matching time_changed events."""
now = event.data[ATTR_NOW]
if now < point_in_time or hasattr(point_in_time_listener, "run"):
return
# Set variable so that we will never run twice.
# Because the event bus might have to wait till a thread comes
# available to execute this listener it might occur that the
# listener gets lined up twice to be executed. This will make
# sure the second time it does nothing.
setattr(point_in_time_listener, "run", True)
async_unsub()
hass.async_run_hass_job(job, now)
async_unsub = hass.bus.async_listen(EVENT_TIME_CHANGED, point_in_time_listener)
return async_unsub
@ha.callback
@loader.bind_hass
def async_track_utc_time_change(
hass, action, hour=None, minute=None, second=None, local=False
):
"""Add a listener that will fire if time matches a pattern."""
job = ha.HassJob(action)
# We do not have to wrap the function with time pattern matching logic
# if no pattern given
if all(val is None for val in (hour, minute, second)):
@ha.callback
def time_change_listener(ev) -> None:
"""Fire every time event that comes in."""
hass.async_run_hass_job(job, ev.data[ATTR_NOW])
return hass.bus.async_listen(EVENT_TIME_CHANGED, time_change_listener)
matching_seconds = event.dt_util.parse_time_expression(second, 0, 59)
matching_minutes = event.dt_util.parse_time_expression(minute, 0, 59)
matching_hours = event.dt_util.parse_time_expression(hour, 0, 23)
next_time = None
def calculate_next(now) -> None:
"""Calculate and set the next time the trigger should fire."""
nonlocal next_time
localized_now = event.dt_util.as_local(now) if local else now
next_time = event.dt_util.find_next_time_expression_time(
localized_now, matching_seconds, matching_minutes, matching_hours
)
# Make sure rolling back the clock doesn't prevent the timer from
# triggering.
last_now = None
@ha.callback
def pattern_time_change_listener(ev) -> None:
"""Listen for matching time_changed events."""
nonlocal next_time, last_now
now = ev.data[ATTR_NOW]
if last_now is None or now < last_now:
# Time rolled back or next time not yet calculated
calculate_next(now)
last_now = now
if next_time <= now:
hass.async_run_hass_job(
job, event.dt_util.as_local(now) if local else now
)
calculate_next(now + datetime.timedelta(seconds=1))
# We can't use async_track_point_in_utc_time here because it would
# break in the case that the system time abruptly jumps backwards.
# Our custom last_now logic takes care of resolving that scenario.
return hass.bus.async_listen(EVENT_TIME_CHANGED, pattern_time_change_listener)
with patch(
"homeassistant.helpers.event.async_track_point_in_utc_time",
async_track_point_in_utc_time,
), patch(
"homeassistant.helpers.event.async_track_utc_time_change",
async_track_utc_time_change,
):
yield
@pytest.fixture
def enable_custom_integrations(hass):
"""Enable custom integrations defined in the test dir."""

5
tests/helpers/test_entity_component.py
View File

@ -5,6 +5,7 @@ from datetime import timedelta
import logging
from unittest.mock import AsyncMock, Mock, patch
from freezegun import freeze_time
import pytest
import voluptuous as vol
@ -177,7 +178,7 @@ async def test_extract_from_service_available_device(hass):
)
async def test_platform_not_ready(hass, legacy_patchable_time):
async def test_platform_not_ready(hass):
"""Test that we retry when platform not ready."""
platform1_setup = Mock(side_effect=[PlatformNotReady, PlatformNotReady, None])
mock_integration(hass, MockModule("mod1"))
@ -192,7 +193,7 @@ async def test_platform_not_ready(hass, legacy_patchable_time):
utcnow = dt_util.utcnow()
with patch("homeassistant.util.dt.utcnow", return_value=utcnow):
with freeze_time(utcnow):
# Should not trigger attempt 2
async_fire_time_changed(hass, utcnow + timedelta(seconds=29))
await hass.async_block_till_done()

116
tests/helpers/test_event.py
View File

@ -6,6 +6,7 @@ from unittest.mock import patch
from astral import LocationInfo
import astral.sun
from freezegun import freeze_time
import jinja2
import pytest
@ -3260,7 +3261,7 @@ async def test_track_time_interval(hass):
assert len(specific_runs) == 2
async def test_track_sunrise(hass, legacy_patchable_time):
async def test_track_sunrise(hass):
"""Test track the sunrise."""
latitude = 32.87336
longitude = 117.22743
@ -3291,42 +3292,46 @@ async def test_track_sunrise(hass, legacy_patchable_time):
# Track sunrise
runs = []
with patch("homeassistant.util.dt.utcnow", return_value=utc_now):
with freeze_time(utc_now):
unsub = async_track_sunrise(hass, callback(lambda: runs.append(1)))
offset_runs = []
offset = timedelta(minutes=30)
with patch("homeassistant.util.dt.utcnow", return_value=utc_now):
with freeze_time(utc_now):
unsub2 = async_track_sunrise(
hass, callback(lambda: offset_runs.append(1)), offset
)
# run tests
async_fire_time_changed(hass, next_rising - offset)
await hass.async_block_till_done()
assert len(runs) == 0
assert len(offset_runs) == 0
with freeze_time(next_rising - offset):
async_fire_time_changed(hass, next_rising - offset)
await hass.async_block_till_done()
assert len(runs) == 0
assert len(offset_runs) == 0
async_fire_time_changed(hass, next_rising)
await hass.async_block_till_done()
assert len(runs) == 1
assert len(offset_runs) == 0
with freeze_time(next_rising):
async_fire_time_changed(hass, next_rising)
await hass.async_block_till_done()
assert len(runs) == 1
assert len(offset_runs) == 0
async_fire_time_changed(hass, next_rising + offset)
await hass.async_block_till_done()
assert len(runs) == 1
assert len(offset_runs) == 1
with freeze_time(next_rising + offset):
async_fire_time_changed(hass, next_rising + offset)
await hass.async_block_till_done()
assert len(runs) == 1
assert len(offset_runs) == 1
unsub()
unsub2()
async_fire_time_changed(hass, next_rising + offset)
await hass.async_block_till_done()
assert len(runs) == 1
assert len(offset_runs) == 1
with freeze_time(next_rising + offset):
async_fire_time_changed(hass, next_rising + offset)
await hass.async_block_till_done()
assert len(runs) == 1
assert len(offset_runs) == 1
async def test_track_sunrise_update_location(hass, legacy_patchable_time):
async def test_track_sunrise_update_location(hass):
"""Test track the sunrise."""
# Setup sun component
hass.config.latitude = 32.87336
@ -3354,16 +3359,17 @@ async def test_track_sunrise_update_location(hass, legacy_patchable_time):
# Track sunrise
runs = []
with patch("homeassistant.util.dt.utcnow", return_value=utc_now):
with freeze_time(utc_now):
async_track_sunrise(hass, callback(lambda: runs.append(1)))
# Mimic sunrise
async_fire_time_changed(hass, next_rising)
await hass.async_block_till_done()
assert len(runs) == 1
with freeze_time(next_rising):
async_fire_time_changed(hass, next_rising)
await hass.async_block_till_done()
assert len(runs) == 1
# Move!
with patch("homeassistant.util.dt.utcnow", return_value=utc_now):
with freeze_time(utc_now):
await hass.config.async_update(latitude=40.755931, longitude=-73.984606)
await hass.async_block_till_done()
@ -3373,10 +3379,11 @@ async def test_track_sunrise_update_location(hass, legacy_patchable_time):
)
# Mimic sunrise
async_fire_time_changed(hass, next_rising)
await hass.async_block_till_done()
# Did not increase
assert len(runs) == 1
with freeze_time(next_rising):
async_fire_time_changed(hass, next_rising)
await hass.async_block_till_done()
# Did not increase
assert len(runs) == 1
# Get next sunrise
mod = -1
@ -3388,13 +3395,14 @@ async def test_track_sunrise_update_location(hass, legacy_patchable_time):
break
mod += 1
# Mimic sunrise at new location
async_fire_time_changed(hass, next_rising)
await hass.async_block_till_done()
assert len(runs) == 2
with freeze_time(next_rising):
# Mimic sunrise at new location
async_fire_time_changed(hass, next_rising)
await hass.async_block_till_done()
assert len(runs) == 2
async def test_track_sunset(hass, legacy_patchable_time):
async def test_track_sunset(hass):
"""Test track the sunset."""
latitude = 32.87336
longitude = 117.22743
@ -3423,39 +3431,43 @@ async def test_track_sunset(hass, legacy_patchable_time):
# Track sunset
runs = []
with patch("homeassistant.util.dt.utcnow", return_value=utc_now):
with freeze_time(utc_now):
unsub = async_track_sunset(hass, callback(lambda: runs.append(1)))
offset_runs = []
offset = timedelta(minutes=30)
with patch("homeassistant.util.dt.utcnow", return_value=utc_now):
with freeze_time(utc_now):
unsub2 = async_track_sunset(
hass, callback(lambda: offset_runs.append(1)), offset
)
# Run tests
async_fire_time_changed(hass, next_setting - offset)
await hass.async_block_till_done()
assert len(runs) == 0
assert len(offset_runs) == 0
with freeze_time(next_setting - offset):
async_fire_time_changed(hass, next_setting - offset)
await hass.async_block_till_done()
assert len(runs) == 0
assert len(offset_runs) == 0
async_fire_time_changed(hass, next_setting)
await hass.async_block_till_done()
assert len(runs) == 1
assert len(offset_runs) == 0
with freeze_time(next_setting):
async_fire_time_changed(hass, next_setting)
await hass.async_block_till_done()
assert len(runs) == 1
assert len(offset_runs) == 0
async_fire_time_changed(hass, next_setting + offset)
await hass.async_block_till_done()
assert len(runs) == 1
assert len(offset_runs) == 1
with freeze_time(next_setting + offset):
async_fire_time_changed(hass, next_setting + offset)
await hass.async_block_till_done()
assert len(runs) == 1
assert len(offset_runs) == 1
unsub()
unsub2()
async_fire_time_changed(hass, next_setting + offset)
await hass.async_block_till_done()
assert len(runs) == 1
assert len(offset_runs) == 1
with freeze_time(next_setting + offset):
async_fire_time_changed(hass, next_setting + offset)
await hass.async_block_till_done()
assert len(runs) == 1
assert len(offset_runs) == 1
async def test_async_track_time_change(hass):

131
tests/test_core.py
View File

@ -13,8 +13,6 @@ import voluptuous as vol
from homeassistant.const import (
ATTR_FRIENDLY_NAME,
ATTR_NOW,
ATTR_SECONDS,
CONF_UNIT_SYSTEM,
EVENT_CALL_SERVICE,
EVENT_CORE_CONFIG_UPDATE,
@ -26,8 +24,6 @@ from homeassistant.const import (
EVENT_SERVICE_REGISTERED,
EVENT_SERVICE_REMOVED,
EVENT_STATE_CHANGED,
EVENT_TIME_CHANGED,
EVENT_TIMER_OUT_OF_SYNC,
MATCH_ALL,
__version__,
)
@ -1061,129 +1057,6 @@ async def test_bad_timezone_raises_value_error(hass):
await hass.config.async_update(time_zone="not_a_timezone")
@patch("homeassistant.core.monotonic")
def test_create_timer(mock_monotonic, loop):
"""Test create timer."""
hass = MagicMock()
funcs = []
orig_callback = ha.callback
def mock_callback(func):
funcs.append(func)
return orig_callback(func)
mock_monotonic.side_effect = 10.2, 10.8, 11.3
with patch.object(ha, "callback", mock_callback), patch(
"homeassistant.core.dt_util.utcnow",
return_value=datetime(2018, 12, 31, 3, 4, 5, 333333),
):
ha._async_create_timer(hass)
assert len(funcs) == 2
fire_time_event, stop_timer = funcs
assert len(hass.loop.call_later.mock_calls) == 1
delay, callback, target = hass.loop.call_later.mock_calls[0][1]
assert abs(delay - 0.666667) < 0.001
assert callback is fire_time_event
assert abs(target - 10.866667) < 0.001
with patch(
"homeassistant.core.dt_util.utcnow",
return_value=datetime(2018, 12, 31, 3, 4, 6, 100000),
):
callback(target)
assert len(hass.bus.async_listen_once.mock_calls) == 1
assert len(hass.bus.async_fire.mock_calls) == 1
assert len(hass.loop.call_later.mock_calls) == 2
event_type, callback = hass.bus.async_listen_once.mock_calls[0][1]
assert event_type == EVENT_HOMEASSISTANT_STOP
assert callback is stop_timer
delay, callback, target = hass.loop.call_later.mock_calls[1][1]
assert abs(delay - 0.9) < 0.001
assert callback is fire_time_event
assert abs(target - 12.2) < 0.001
event_type, event_data = hass.bus.async_fire.mock_calls[0][1]
assert event_type == EVENT_TIME_CHANGED
assert event_data[ATTR_NOW] == datetime(2018, 12, 31, 3, 4, 6, 100000)
@patch("homeassistant.core.monotonic")
def test_timer_out_of_sync(mock_monotonic, loop):
"""Test create timer."""
hass = MagicMock()
funcs = []
orig_callback = ha.callback
def mock_callback(func):
funcs.append(func)
return orig_callback(func)
mock_monotonic.side_effect = 10.2, 13.3, 13.4
with patch.object(ha, "callback", mock_callback), patch(
"homeassistant.core.dt_util.utcnow",
return_value=datetime(2018, 12, 31, 3, 4, 5, 333333),
):
ha._async_create_timer(hass)
delay, callback, target = hass.loop.call_later.mock_calls[0][1]
with patch(
"homeassistant.core.dt_util.utcnow",
return_value=datetime(2018, 12, 31, 3, 4, 8, 200000),
):
callback(target)
_, event_0_args, event_0_kwargs = hass.bus.async_fire.mock_calls[0]
event_context_0 = event_0_kwargs["context"]
event_type_0, _ = event_0_args
assert event_type_0 == EVENT_TIME_CHANGED
_, event_1_args, event_1_kwargs = hass.bus.async_fire.mock_calls[1]
event_type_1, event_data_1 = event_1_args
event_context_1 = event_1_kwargs["context"]
assert event_type_1 == EVENT_TIMER_OUT_OF_SYNC
assert abs(event_data_1[ATTR_SECONDS] - 2.433333) < 0.001
assert event_context_0 == event_context_1
assert len(funcs) == 2
fire_time_event, _ = funcs
assert len(hass.loop.call_later.mock_calls) == 2
delay, callback, target = hass.loop.call_later.mock_calls[1][1]
assert abs(delay - 0.8) < 0.001
assert callback is fire_time_event
assert abs(target - 14.2) < 0.001
async def test_hass_start_starts_the_timer(loop):
"""Test when hass starts, it starts the timer."""
hass = ha.HomeAssistant()
try:
with patch("homeassistant.core._async_create_timer") as mock_timer:
await hass.async_start()
assert hass.state == ha.CoreState.running
assert not hass._track_task
assert len(mock_timer.mock_calls) == 1
assert mock_timer.mock_calls[0][1][0] is hass
finally:
await hass.async_stop()
assert hass.state == ha.CoreState.stopped
async def test_start_taking_too_long(loop, caplog):
"""Test when async_start takes too long."""
hass = ha.HomeAssistant()
@ -1192,12 +1065,10 @@ async def test_start_taking_too_long(loop, caplog):
try:
with patch.object(
hass, "async_block_till_done", side_effect=asyncio.TimeoutError
), patch("homeassistant.core._async_create_timer") as mock_timer:
):
await hass.async_start()
assert hass.state == ha.CoreState.running
assert len(mock_timer.mock_calls) == 1
assert mock_timer.mock_calls[0][1][0] is hass
assert "Something is blocking Home Assistant" in caplog.text
finally: