jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-13 16:27:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix litejet tests (#82324)

Browse Source 
Co-authored-by: epenet <6771947+epenet@users.noreply.github.com>
This commit is contained in:
J. Nick Koston 2022-11-18 08:19:38 -06:00 committed by GitHub
parent 077bd594eb
commit 1c6b4967cf
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 37 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
tests/common.py
View File

@ -380,10 +380,35 @@ fire_mqtt_message = threadsafe_callback_factory(async_fire_mqtt_message)
@ha.callback @ha.callback
def async_fire_time_changed( def async_fire_time_changed_exact(
hass: HomeAssistant, datetime_: datetime = None, fire_all: bool = False hass: HomeAssistant, datetime_: datetime | None = None, fire_all: bool = False
) -> None: ) -> None:
"""Fire a time changed event.""" """Fire a time changed event at an exact microsecond.
Consider that its not possible to actually achieve an exact microsecond
in production as the event loop is not precise enough. If your code
relies on this level of precision, consider a different approach
as this is only for testing.
"""
if datetime_ is None:
utc_datetime = date_util.utcnow()
else:
utc_datetime = date_util.as_utc(datetime_)
_async_fire_time_changed(hass, utc_datetime, fire_all)
@ha.callback
def async_fire_time_changed(
hass: HomeAssistant, datetime_: datetime | None = None, fire_all: bool = False
) -> None:
"""Fire a time changed event.
This function will ensure microseconds at at least 500000
to account for the synchronization repeating listeners.
If you need to fire an exact microsecond, use async_fire_time_changed_exact.
"""
if datetime_ is None: if datetime_ is None:
utc_datetime = date_util.utcnow() utc_datetime = date_util.utcnow()
else: else:
@ -396,8 +421,14 @@ def async_fire_time_changed(
# staggering to avoid thundering herd. # staggering to avoid thundering herd.
utc_datetime = utc_datetime.replace(microsecond=500000) utc_datetime = utc_datetime.replace(microsecond=500000)
timestamp = date_util.utc_to_timestamp(utc_datetime) _async_fire_time_changed(hass, utc_datetime, fire_all)
@ha.callback
def _async_fire_time_changed(
hass: HomeAssistant, utc_datetime: datetime | None, fire_all: bool
) -> None:
timestamp = date_util.utc_to_timestamp(utc_datetime)
for task in list(hass.loop._scheduled): for task in list(hass.loop._scheduled):
if not isinstance(task, asyncio.TimerHandle): if not isinstance(task, asyncio.TimerHandle):
continue continue

4
tests/components/litejet/test_trigger.py
View File

@ -12,7 +12,7 @@ import homeassistant.util.dt as dt_util
from . import async_init_integration from . import async_init_integration
from tests.common import async_fire_time_changed, async_mock_service from tests.common import async_fire_time_changed_exact, async_mock_service
from tests.components.blueprint.conftest import stub_blueprint_populate # noqa: F401 from tests.components.blueprint.conftest import stub_blueprint_populate # noqa: F401
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@ -66,7 +66,7 @@ async def simulate_time(hass, mock_litejet, delta):
return_value=mock_litejet.start_time + delta, return_value=mock_litejet.start_time + delta,
): ):
_LOGGER.info("now=%s", dt_util.utcnow()) _LOGGER.info("now=%s", dt_util.utcnow())
async_fire_time_changed(hass, mock_litejet.start_time + delta) async_fire_time_changed_exact(hass, mock_litejet.start_time + delta)
await hass.async_block_till_done() await hass.async_block_till_done()
_LOGGER.info("done with now=%s", dt_util.utcnow()) _LOGGER.info("done with now=%s", dt_util.utcnow())