Skip to content

Commit

Permalink
Improve lingering timer checks (home-assistant#99472)
Browse files Browse the repository at this point in the history
  • Loading branch information
bdraco authored Sep 2, 2023
1 parent 1048f47 commit 1ab2e90
Show file tree
Hide file tree
Showing 2 changed files with 31 additions and 8 deletions.
4 changes: 4 additions & 0 deletions homeassistant/core.py
Original file line number Diff line number Diff line change
Expand Up @@ -294,6 +294,10 @@ def __new__(cls, config_dir: str) -> HomeAssistant:
_hass.hass = hass
return hass

def __repr__(self) -> str:
"""Return the representation."""
return f"<HomeAssistant {self.state}>"

def __init__(self, config_dir: str) -> None:
"""Initialize new Home Assistant object."""
self.loop = asyncio.get_running_loop()
Expand Down
35 changes: 27 additions & 8 deletions tests/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,12 +3,13 @@

import asyncio
from collections.abc import AsyncGenerator, Callable, Coroutine, Generator
from contextlib import asynccontextmanager
from contextlib import asynccontextmanager, contextmanager
import functools
import gc
import itertools
import logging
import os
import reprlib
import sqlite3
import ssl
import threading
Expand Down Expand Up @@ -302,6 +303,21 @@ def skip_stop_scripts(
yield


@contextmanager
def long_repr_strings() -> Generator[None, None, None]:
"""Increase reprlib maxstring and maxother to 300."""
arepr = reprlib.aRepr
original_maxstring = arepr.maxstring
original_maxother = arepr.maxother
arepr.maxstring = 300
arepr.maxother = 300
try:
yield
finally:
arepr.maxstring = original_maxstring
arepr.maxother = original_maxother


@pytest.fixture(autouse=True)
def verify_cleanup(
event_loop: asyncio.AbstractEventLoop,
Expand Down Expand Up @@ -335,13 +351,16 @@ def verify_cleanup(

for handle in event_loop._scheduled: # type: ignore[attr-defined]
if not handle.cancelled():
if expected_lingering_timers:
_LOGGER.warning("Lingering timer after test %r", handle)
elif handle._args and isinstance(job := handle._args[0], HassJob):
pytest.fail(f"Lingering timer after job {repr(job)}")
else:
pytest.fail(f"Lingering timer after test {repr(handle)}")
handle.cancel()
with long_repr_strings():
if expected_lingering_timers:
_LOGGER.warning("Lingering timer after test %r", handle)
elif handle._args and isinstance(job := handle._args[-1], HassJob):
if job.cancel_on_shutdown:
continue
pytest.fail(f"Lingering timer after job {repr(job)}")
else:
pytest.fail(f"Lingering timer after test {repr(handle)}")
handle.cancel()

# Verify no threads where left behind.
threads = frozenset(threading.enumerate()) - threads_before
Expand Down

0 comments on commit 1ab2e90

Please sign in to comment.