Detect blocking module imports in the event loop (#114488)

homeassistant/block_async_io.py

@@ -1,9 +1,36 @@
 """Block blocking calls being done in asyncio."""
 
+from contextlib import suppress
 from http.client import HTTPConnection
+import importlib
+import sys
 import time
+from typing import Any
 
-from .util.async_ import protect_loop
+from .helpers.frame import get_current_frame
+from .util.loop import protect_loop
+
+_IN_TESTS = "unittest" in sys.modules
+
+
+def _check_import_call_allowed(mapped_args: dict[str, Any]) -> bool:
+    # If the module is already imported, we can ignore it.
+    return bool((args := mapped_args.get("args")) and args[0] in sys.modules)
+
+
+def _check_sleep_call_allowed(mapped_args: dict[str, Any]) -> bool:
+    #
+    # Avoid extracting the stack unless we need to since it
+    # will have to access the linecache which can do blocking
+    # I/O and we are trying to avoid blocking calls.
+    #
+    # frame[0] is us
+    # frame[1] is check_loop
+    # frame[2] is protected_loop_func
+    # frame[3] is the offender
+    with suppress(ValueError):
+        return get_current_frame(4).f_code.co_filename.endswith("pydevd.py")
+    return False
 
 
 def enable() -> None:
@@ -14,8 +41,20 @@ def enable() -> None:
     )
 
     # Prevent sleeping in event loop. Non-strict since 2022.02
-    time.sleep = protect_loop(time.sleep, strict=False)
+    time.sleep = protect_loop(
+        time.sleep, strict=False, check_allowed=_check_sleep_call_allowed
+    )
 
     # Currently disabled. pytz doing I/O when getting timezone.
     # Prevent files being opened inside the event loop
     # builtins.open = protect_loop(builtins.open)
+
+    if not _IN_TESTS:
+        # unittest uses `importlib.import_module` to do mocking
+        # so we cannot protect it if we are running tests
+        importlib.import_module = protect_loop(
+            importlib.import_module,
+            strict_core=False,
+            strict=False,
+            check_allowed=_check_import_call_allowed,
+        )

homeassistant/bootstrap.py

@@ -23,7 +23,14 @@ import cryptography.hazmat.backends.openssl.backend  # noqa: F401
 import voluptuous as vol
 import yarl
 
-from . import config as conf_util, config_entries, core, loader, requirements
+from . import (
+    block_async_io,
+    config as conf_util,
+    config_entries,
+    core,
+    loader,
+    requirements,
+)
 
 # Pre-import frontend deps which have no requirements here to avoid
 # loading them at run time and blocking the event loop. We do this ahead
@@ -260,6 +267,8 @@ async def async_setup_hass(
     _LOGGER.info("Config directory: %s", runtime_config.config_dir)
 
     loader.async_setup(hass)
+    block_async_io.enable()
+
     config_dict = None
     basic_setup_success = False
 

homeassistant/components/recorder/pool.py

@@ -14,7 +14,7 @@ from sqlalchemy.pool import (
 )
 
 from homeassistant.helpers.frame import report
-from homeassistant.util.async_ import check_loop
+from homeassistant.util.loop import check_loop
 
 from .const import DB_WORKER_PREFIX
 

homeassistant/core.py

@@ -50,7 +50,7 @@ from typing_extensions import TypeVar
 import voluptuous as vol
 import yarl
 
-from . import block_async_io, util
+from . import util
 from .const import (
     ATTR_DOMAIN,
     ATTR_FRIENDLY_NAME,
@@ -130,7 +130,6 @@ STOP_STAGE_SHUTDOWN_TIMEOUT = 100
 FINAL_WRITE_STAGE_SHUTDOWN_TIMEOUT = 60
 CLOSE_STAGE_SHUTDOWN_TIMEOUT = 30
 
-block_async_io.enable()
 
 _T = TypeVar("_T")
 _R = TypeVar("_R")

homeassistant/util/async_.py

@@ -5,21 +5,15 @@ from __future__ import annotations
 from asyncio import AbstractEventLoop, Future, Semaphore, Task, gather, get_running_loop
 from collections.abc import Awaitable, Callable, Coroutine
 import concurrent.futures
-from contextlib import suppress
-import functools
 import logging
 import threading
-from typing import Any, ParamSpec, TypeVar, TypeVarTuple
+from typing import Any, TypeVar, TypeVarTuple
-
-from homeassistant.exceptions import HomeAssistantError
 
 _LOGGER = logging.getLogger(__name__)
 
 _SHUTDOWN_RUN_CALLBACK_THREADSAFE = "_shutdown_run_callback_threadsafe"
 
 _T = TypeVar("_T")
-_R = TypeVar("_R")
-_P = ParamSpec("_P")
 _Ts = TypeVarTuple("_Ts")
 
 
@@ -92,105 +86,6 @@ def run_callback_threadsafe(
     return future
 
 
-def check_loop(
-    func: Callable[..., Any], strict: bool = True, advise_msg: str | None = None
-) -> None:
-    """Warn if called inside the event loop. Raise if `strict` is True.
-
-    The default advisory message is 'Use `await hass.async_add_executor_job()'
-    Set `advise_msg` to an alternate message if the solution differs.
-    """
-    try:
-        get_running_loop()
-        in_loop = True
-    except RuntimeError:
-        in_loop = False
-
-    if not in_loop:
-        return
-
-    # Import only after we know we are running in the event loop
-    # so threads do not have to pay the late import cost.
-    # pylint: disable=import-outside-toplevel
-    from homeassistant.core import HomeAssistant, async_get_hass
-    from homeassistant.helpers.frame import (
-        MissingIntegrationFrame,
-        get_current_frame,
-        get_integration_frame,
-    )
-    from homeassistant.loader import async_suggest_report_issue
-
-    found_frame = None
-
-    if func.__name__ == "sleep":
-        #
-        # Avoid extracting the stack unless we need to since it
-        # will have to access the linecache which can do blocking
-        # I/O and we are trying to avoid blocking calls.
-        #
-        # frame[1] is us
-        # frame[2] is protected_loop_func
-        # frame[3] is the offender
-        with suppress(ValueError):
-            offender_frame = get_current_frame(3)
-            if offender_frame.f_code.co_filename.endswith("pydevd.py"):
-                return
-
-    try:
-        integration_frame = get_integration_frame()
-    except MissingIntegrationFrame:
-        # Did not source from integration? Hard error.
-        if found_frame is None:
-            raise RuntimeError(  # noqa: TRY200
-                f"Detected blocking call to {func.__name__} inside the event loop. "
-                f"{advise_msg or 'Use `await hass.async_add_executor_job()`'}; "
-                "This is causing stability issues. Please create a bug report at "
-                f"https://github.com/home-assistant/core/issues?q=is%3Aopen+is%3Aissue"
-            )
-
-    hass: HomeAssistant | None = None
-    with suppress(HomeAssistantError):
-        hass = async_get_hass()
-    report_issue = async_suggest_report_issue(
-        hass,
-        integration_domain=integration_frame.integration,
-        module=integration_frame.module,
-    )
-
-    _LOGGER.warning(
-        (
-            "Detected blocking call to %s inside the event loop by %sintegration '%s' "
-            "at %s, line %s: %s, please %s"
-        ),
-        func.__name__,
-        "custom " if integration_frame.custom_integration else "",
-        integration_frame.integration,
-        integration_frame.relative_filename,
-        integration_frame.line_number,
-        integration_frame.line,
-        report_issue,
-    )
-
-    if strict:
-        raise RuntimeError(
-            "Blocking calls must be done in the executor or a separate thread;"
-            f" {advise_msg or 'Use `await hass.async_add_executor_job()`'}; at"
-            f" {integration_frame.relative_filename}, line {integration_frame.line_number}:"
-            f" {integration_frame.line}"
-        )
-
-
-def protect_loop(func: Callable[_P, _R], strict: bool = True) -> Callable[_P, _R]:
-    """Protect function from running in event loop."""
-
-    @functools.wraps(func)
-    def protected_loop_func(*args: _P.args, **kwargs: _P.kwargs) -> _R:
-        check_loop(func, strict=strict)
-        return func(*args, **kwargs)
-
-    return protected_loop_func
-
-
 async def gather_with_limited_concurrency(
     limit: int, *tasks: Any, return_exceptions: bool = False
 ) -> Any:

homeassistant/util/loop.py

@@ -0,0 +1,146 @@
+"""asyncio loop utilities."""
+
+from __future__ import annotations
+
+from asyncio import get_running_loop
+from collections.abc import Callable
+from contextlib import suppress
+import functools
+import linecache
+import logging
+from typing import Any, ParamSpec, TypeVar
+
+from homeassistant.core import HomeAssistant, async_get_hass
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers.frame import (
+    MissingIntegrationFrame,
+    get_current_frame,
+    get_integration_frame,
+)
+from homeassistant.loader import async_suggest_report_issue
+
+_LOGGER = logging.getLogger(__name__)
+
+
+_R = TypeVar("_R")
+_P = ParamSpec("_P")
+
+
+def _get_line_from_cache(filename: str, lineno: int) -> str:
+    """Get line from cache or read from file."""
+    return (linecache.getline(filename, lineno) or "?").strip()
+
+
+def check_loop(
+    func: Callable[..., Any],
+    check_allowed: Callable[[dict[str, Any]], bool] | None = None,
+    strict: bool = True,
+    strict_core: bool = True,
+    advise_msg: str | None = None,
+    **mapped_args: Any,
+) -> None:
+    """Warn if called inside the event loop. Raise if `strict` is True.
+
+    The default advisory message is 'Use `await hass.async_add_executor_job()'
+    Set `advise_msg` to an alternate message if the solution differs.
+    """
+    try:
+        get_running_loop()
+        in_loop = True
+    except RuntimeError:
+        in_loop = False
+
+    if not in_loop:
+        return
+
+    if check_allowed is not None and check_allowed(mapped_args):
+        return
+
+    found_frame = None
+    offender_frame = get_current_frame(2)
+    offender_filename = offender_frame.f_code.co_filename
+    offender_lineno = offender_frame.f_lineno
+    offender_line = _get_line_from_cache(offender_filename, offender_lineno)
+
+    try:
+        integration_frame = get_integration_frame()
+    except MissingIntegrationFrame:
+        # Did not source from integration? Hard error.
+        if not strict_core:
+            _LOGGER.warning(
+                "Detected blocking call to %s with args %s in %s, "
+                "line %s: %s inside the event loop",
+                func.__name__,
+                mapped_args.get("args"),
+                offender_filename,
+                offender_lineno,
+                offender_line,
+            )
+            return
+
+        if found_frame is None:
+            raise RuntimeError(  # noqa: TRY200
+                f"Detected blocking call to {func.__name__} inside the event loop "
+                f"in {offender_filename}, line {offender_lineno}: {offender_line}. "
+                f"{advise_msg or 'Use `await hass.async_add_executor_job()`'}; "
+                "This is causing stability issues. Please create a bug report at "
+                f"https://github.com/home-assistant/core/issues?q=is%3Aopen+is%3Aissue"
+            )
+
+    hass: HomeAssistant | None = None
+    with suppress(HomeAssistantError):
+        hass = async_get_hass()
+    report_issue = async_suggest_report_issue(
+        hass,
+        integration_domain=integration_frame.integration,
+        module=integration_frame.module,
+    )
+
+    _LOGGER.warning(
+        (
+            "Detected blocking call to %s inside the event loop by %sintegration '%s' "
+            "at %s, line %s: %s (offender: %s, line %s: %s), please %s"
+        ),
+        func.__name__,
+        "custom " if integration_frame.custom_integration else "",
+        integration_frame.integration,
+        integration_frame.relative_filename,
+        integration_frame.line_number,
+        integration_frame.line,
+        offender_filename,
+        offender_lineno,
+        offender_line,
+        report_issue,
+    )
+
+    if strict:
+        raise RuntimeError(
+            "Blocking calls must be done in the executor or a separate thread;"
+            f" {advise_msg or 'Use `await hass.async_add_executor_job()`'}; at"
+            f" {integration_frame.relative_filename}, line {integration_frame.line_number}:"
+            f" {integration_frame.line} "
+            f"(offender: {offender_filename}, line {offender_lineno}: {offender_line})"
+        )
+
+
+def protect_loop(
+    func: Callable[_P, _R],
+    strict: bool = True,
+    strict_core: bool = True,
+    check_allowed: Callable[[dict[str, Any]], bool] | None = None,
+) -> Callable[_P, _R]:
+    """Protect function from running in event loop."""
+
+    @functools.wraps(func)
+    def protected_loop_func(*args: _P.args, **kwargs: _P.kwargs) -> _R:
+        check_loop(
+            func,
+            strict=strict,
+            strict_core=strict_core,
+            check_allowed=check_allowed,
+            args=args,
+            kwargs=kwargs,
+        )
+        return func(*args, **kwargs)
+
+    return protected_loop_func

tests/test_block_async_io.py

@@ -0,0 +1,200 @@
+"""Tests for async util methods from Python source."""
+
+import importlib
+import time
+from unittest.mock import Mock, patch
+
+import pytest
+
+from homeassistant import block_async_io
+
+from tests.common import extract_stack_to_frame
+
+
+async def test_protect_loop_debugger_sleep(caplog: pytest.LogCaptureFixture) -> None:
+    """Test time.sleep injected by the debugger is not reported."""
+    block_async_io.enable()
+    frames = extract_stack_to_frame(
+        [
+            Mock(
+                filename="/home/paulus/homeassistant/.venv/blah/pydevd.py",
+                lineno="23",
+                line="do_something()",
+            ),
+        ]
+    )
+    with (
+        patch(
+            "homeassistant.block_async_io.get_current_frame",
+            return_value=frames,
+        ),
+        patch(
+            "homeassistant.helpers.frame.get_current_frame",
+            return_value=frames,
+        ),
+    ):
+        time.sleep(0)
+    assert "Detected blocking call inside the event loop" not in caplog.text
+
+
+async def test_protect_loop_sleep(caplog: pytest.LogCaptureFixture) -> None:
+    """Test time.sleep not injected by the debugger raises."""
+    block_async_io.enable()
+    frames = extract_stack_to_frame(
+        [
+            Mock(
+                filename="/home/paulus/homeassistant/no_dev.py",
+                lineno="23",
+                line="do_something()",
+            ),
+        ]
+    )
+    with (
+        pytest.raises(
+            RuntimeError, match="Detected blocking call to sleep inside the event loop"
+        ),
+        patch(
+            "homeassistant.block_async_io.get_current_frame",
+            return_value=frames,
+        ),
+        patch(
+            "homeassistant.helpers.frame.get_current_frame",
+            return_value=frames,
+        ),
+    ):
+        time.sleep(0)
+
+
+async def test_protect_loop_sleep_get_current_frame_raises(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Test time.sleep when get_current_frame raises ValueError."""
+    block_async_io.enable()
+    frames = extract_stack_to_frame(
+        [
+            Mock(
+                filename="/home/paulus/homeassistant/no_dev.py",
+                lineno="23",
+                line="do_something()",
+            ),
+        ]
+    )
+    with (
+        pytest.raises(
+            RuntimeError, match="Detected blocking call to sleep inside the event loop"
+        ),
+        patch(
+            "homeassistant.block_async_io.get_current_frame",
+            side_effect=ValueError,
+        ),
+        patch(
+            "homeassistant.helpers.frame.get_current_frame",
+            return_value=frames,
+        ),
+    ):
+        time.sleep(0)
+
+
+async def test_protect_loop_importlib_import_module_non_integration(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Test import_module in the loop for non-loaded module."""
+    frames = extract_stack_to_frame(
+        [
+            Mock(
+                filename="/home/paulus/homeassistant/no_dev.py",
+                lineno="23",
+                line="do_something()",
+            ),
+        ]
+    )
+    with (
+        pytest.raises(ImportError),
+        patch.object(block_async_io, "_IN_TESTS", False),
+        patch(
+            "homeassistant.block_async_io.get_current_frame",
+            return_value=frames,
+        ),
+        patch(
+            "homeassistant.helpers.frame.get_current_frame",
+            return_value=frames,
+        ),
+    ):
+        block_async_io.enable()
+        importlib.import_module("not_loaded_module")
+
+    assert "Detected blocking call to import_module" in caplog.text
+
+
+async def test_protect_loop_importlib_import_loaded_module_non_integration(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Test import_module in the loop for a loaded module."""
+    frames = extract_stack_to_frame(
+        [
+            Mock(
+                filename="/home/paulus/homeassistant/no_dev.py",
+                lineno="23",
+                line="do_something()",
+            ),
+        ]
+    )
+    with (
+        patch.object(block_async_io, "_IN_TESTS", False),
+        patch(
+            "homeassistant.block_async_io.get_current_frame",
+            return_value=frames,
+        ),
+        patch(
+            "homeassistant.helpers.frame.get_current_frame",
+            return_value=frames,
+        ),
+    ):
+        block_async_io.enable()
+        importlib.import_module("sys")
+
+    assert "Detected blocking call to import_module" not in caplog.text
+
+
+async def test_protect_loop_importlib_import_module_in_integration(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Test import_module in the loop for non-loaded module in an integration."""
+    frames = extract_stack_to_frame(
+        [
+            Mock(
+                filename="/home/paulus/homeassistant/core.py",
+                lineno="23",
+                line="do_something()",
+            ),
+            Mock(
+                filename="/home/paulus/homeassistant/components/hue/light.py",
+                lineno="23",
+                line="self.light.is_on",
+            ),
+            Mock(
+                filename="/home/paulus/aiohue/lights.py",
+                lineno="2",
+                line="something()",
+            ),
+        ]
+    )
+    with (
+        pytest.raises(ImportError),
+        patch.object(block_async_io, "_IN_TESTS", False),
+        patch(
+            "homeassistant.block_async_io.get_current_frame",
+            return_value=frames,
+        ),
+        patch(
+            "homeassistant.helpers.frame.get_current_frame",
+            return_value=frames,
+        ),
+    ):
+        block_async_io.enable()
+        importlib.import_module("not_loaded_module")
+
+    assert (
+        "Detected blocking call to import_module inside the event loop by "
+        "integration 'hue' at homeassistant/components/hue/light.py, line 23"
+    ) in caplog.text

tests/util/test_async.py

@@ -6,12 +6,9 @@ from unittest.mock import MagicMock, Mock, patch
 
 import pytest
 
-from homeassistant import block_async_io
 from homeassistant.core import HomeAssistant
 from homeassistant.util import async_ as hasync
 
-from tests.common import extract_stack_to_frame
-
 
 @patch("concurrent.futures.Future")
 @patch("threading.get_ident")
@@ -38,172 +35,6 @@ def test_run_callback_threadsafe_from_inside_event_loop(mock_ident, _) -> None:
     assert len(loop.call_soon_threadsafe.mock_calls) == 2
 
 
-def banned_function():
-    """Mock banned function."""
-
-
-async def test_check_loop_async() -> None:
-    """Test check_loop detects when called from event loop without integration context."""
-    with pytest.raises(RuntimeError):
-        hasync.check_loop(banned_function)
-
-
-async def test_check_loop_async_integration(caplog: pytest.LogCaptureFixture) -> None:
-    """Test check_loop detects and raises when called from event loop from integration context."""
-    with (
-        pytest.raises(RuntimeError),
-        patch(
-            "homeassistant.helpers.frame.linecache.getline",
-            return_value="self.light.is_on",
-        ),
-        patch(
-            "homeassistant.helpers.frame.get_current_frame",
-            return_value=extract_stack_to_frame(
-                [
-                    Mock(
-                        filename="/home/paulus/homeassistant/core.py",
-                        lineno="23",
-                        line="do_something()",
-                    ),
-                    Mock(
-                        filename="/home/paulus/homeassistant/components/hue/light.py",
-                        lineno="23",
-                        line="self.light.is_on",
-                    ),
-                    Mock(
-                        filename="/home/paulus/aiohue/lights.py",
-                        lineno="2",
-                        line="something()",
-                    ),
-                ]
-            ),
-        ),
-    ):
-        hasync.check_loop(banned_function)
-    assert (
-        "Detected blocking call to banned_function inside the event loop by integration"
-        " 'hue' at homeassistant/components/hue/light.py, line 23: self.light.is_on, "
-        "please create a bug report at https://github.com/home-assistant/core/issues?"
-        "q=is%3Aopen+is%3Aissue+label%3A%22integration%3A+hue%22" in caplog.text
-    )
-
-
-async def test_check_loop_async_integration_non_strict(
-    caplog: pytest.LogCaptureFixture,
-) -> None:
-    """Test check_loop detects when called from event loop from integration context."""
-    with (
-        patch(
-            "homeassistant.helpers.frame.linecache.getline",
-            return_value="self.light.is_on",
-        ),
-        patch(
-            "homeassistant.helpers.frame.get_current_frame",
-            return_value=extract_stack_to_frame(
-                [
-                    Mock(
-                        filename="/home/paulus/homeassistant/core.py",
-                        lineno="23",
-                        line="do_something()",
-                    ),
-                    Mock(
-                        filename="/home/paulus/homeassistant/components/hue/light.py",
-                        lineno="23",
-                        line="self.light.is_on",
-                    ),
-                    Mock(
-                        filename="/home/paulus/aiohue/lights.py",
-                        lineno="2",
-                        line="something()",
-                    ),
-                ]
-            ),
-        ),
-    ):
-        hasync.check_loop(banned_function, strict=False)
-    assert (
-        "Detected blocking call to banned_function inside the event loop by integration"
-        " 'hue' at homeassistant/components/hue/light.py, line 23: self.light.is_on, "
-        "please create a bug report at https://github.com/home-assistant/core/issues?"
-        "q=is%3Aopen+is%3Aissue+label%3A%22integration%3A+hue%22" in caplog.text
-    )
-
-
-async def test_check_loop_async_custom(caplog: pytest.LogCaptureFixture) -> None:
-    """Test check_loop detects when called from event loop with custom component context."""
-    with (
-        pytest.raises(RuntimeError),
-        patch(
-            "homeassistant.helpers.frame.linecache.getline",
-            return_value="self.light.is_on",
-        ),
-        patch(
-            "homeassistant.helpers.frame.get_current_frame",
-            return_value=extract_stack_to_frame(
-                [
-                    Mock(
-                        filename="/home/paulus/homeassistant/core.py",
-                        lineno="23",
-                        line="do_something()",
-                    ),
-                    Mock(
-                        filename="/home/paulus/config/custom_components/hue/light.py",
-                        lineno="23",
-                        line="self.light.is_on",
-                    ),
-                    Mock(
-                        filename="/home/paulus/aiohue/lights.py",
-                        lineno="2",
-                        line="something()",
-                    ),
-                ]
-            ),
-        ),
-    ):
-        hasync.check_loop(banned_function)
-    assert (
-        "Detected blocking call to banned_function inside the event loop by custom "
-        "integration 'hue' at custom_components/hue/light.py, line 23: self.light.is_on"
-        ", please create a bug report at https://github.com/home-assistant/core/issues?"
-        "q=is%3Aopen+is%3Aissue+label%3A%22integration%3A+hue%22"
-    ) in caplog.text
-
-
-def test_check_loop_sync(caplog: pytest.LogCaptureFixture) -> None:
-    """Test check_loop does nothing when called from thread."""
-    hasync.check_loop(banned_function)
-    assert "Detected blocking call inside the event loop" not in caplog.text
-
-
-def test_protect_loop_sync() -> None:
-    """Test protect_loop calls check_loop."""
-    func = Mock()
-    with patch("homeassistant.util.async_.check_loop") as mock_check_loop:
-        hasync.protect_loop(func)(1, test=2)
-    mock_check_loop.assert_called_once_with(func, strict=True)
-    func.assert_called_once_with(1, test=2)
-
-
-async def test_protect_loop_debugger_sleep(caplog: pytest.LogCaptureFixture) -> None:
-    """Test time.sleep injected by the debugger is not reported."""
-    block_async_io.enable()
-
-    with patch(
-        "homeassistant.helpers.frame.get_current_frame",
-        return_value=extract_stack_to_frame(
-            [
-                Mock(
-                    filename="/home/paulus/homeassistant/.venv/blah/pydevd.py",
-                    lineno="23",
-                    line="do_something()",
-                ),
-            ]
-        ),
-    ):
-        time.sleep(0)
-    assert "Detected blocking call inside the event loop" not in caplog.text
-
-
 async def test_gather_with_limited_concurrency() -> None:
     """Test gather_with_limited_concurrency limits the number of running tasks."""
 

tests/util/test_loop.py

@@ -0,0 +1,200 @@
+"""Tests for async util methods from Python source."""
+
+from unittest.mock import Mock, patch
+
+import pytest
+
+from homeassistant.util import loop as haloop
+
+from tests.common import extract_stack_to_frame
+
+
+def banned_function():
+    """Mock banned function."""
+
+
+async def test_check_loop_async() -> None:
+    """Test check_loop detects when called from event loop without integration context."""
+    with pytest.raises(RuntimeError):
+        haloop.check_loop(banned_function)
+
+
+async def test_check_loop_async_non_strict_core(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Test non_strict_core check_loop detects from event loop without integration context."""
+    haloop.check_loop(banned_function, strict_core=False)
+    assert "Detected blocking call to banned_function" in caplog.text
+
+
+async def test_check_loop_async_integration(caplog: pytest.LogCaptureFixture) -> None:
+    """Test check_loop detects and raises when called from event loop from integration context."""
+    frames = extract_stack_to_frame(
+        [
+            Mock(
+                filename="/home/paulus/homeassistant/core.py",
+                lineno="23",
+                line="do_something()",
+            ),
+            Mock(
+                filename="/home/paulus/homeassistant/components/hue/light.py",
+                lineno="23",
+                line="self.light.is_on",
+            ),
+            Mock(
+                filename="/home/paulus/aiohue/lights.py",
+                lineno="2",
+                line="something()",
+            ),
+        ]
+    )
+    with (
+        pytest.raises(RuntimeError),
+        patch(
+            "homeassistant.helpers.frame.linecache.getline",
+            return_value="self.light.is_on",
+        ),
+        patch(
+            "homeassistant.util.loop._get_line_from_cache",
+            return_value="mock_line",
+        ),
+        patch(
+            "homeassistant.util.loop.get_current_frame",
+            return_value=frames,
+        ),
+        patch(
+            "homeassistant.helpers.frame.get_current_frame",
+            return_value=frames,
+        ),
+    ):
+        haloop.check_loop(banned_function)
+    assert (
+        "Detected blocking call to banned_function inside the event loop by integration"
+        " 'hue' at homeassistant/components/hue/light.py, line 23: self.light.is_on "
+        "(offender: /home/paulus/aiohue/lights.py, line 2: mock_line), please create "
+        "a bug report at https://github.com/home-assistant/core/issues?"
+        "q=is%3Aopen+is%3Aissue+label%3A%22integration%3A+hue%22" in caplog.text
+    )
+
+
+async def test_check_loop_async_integration_non_strict(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Test check_loop detects when called from event loop from integration context."""
+    frames = extract_stack_to_frame(
+        [
+            Mock(
+                filename="/home/paulus/homeassistant/core.py",
+                lineno="23",
+                line="do_something()",
+            ),
+            Mock(
+                filename="/home/paulus/homeassistant/components/hue/light.py",
+                lineno="23",
+                line="self.light.is_on",
+            ),
+            Mock(
+                filename="/home/paulus/aiohue/lights.py",
+                lineno="2",
+                line="something()",
+            ),
+        ]
+    )
+    with (
+        patch(
+            "homeassistant.helpers.frame.linecache.getline",
+            return_value="self.light.is_on",
+        ),
+        patch(
+            "homeassistant.util.loop._get_line_from_cache",
+            return_value="mock_line",
+        ),
+        patch(
+            "homeassistant.util.loop.get_current_frame",
+            return_value=frames,
+        ),
+        patch(
+            "homeassistant.helpers.frame.get_current_frame",
+            return_value=frames,
+        ),
+    ):
+        haloop.check_loop(banned_function, strict=False)
+    assert (
+        "Detected blocking call to banned_function inside the event loop by integration"
+        " 'hue' at homeassistant/components/hue/light.py, line 23: self.light.is_on "
+        "(offender: /home/paulus/aiohue/lights.py, line 2: mock_line), "
+        "please create a bug report at https://github.com/home-assistant/core/issues?"
+        "q=is%3Aopen+is%3Aissue+label%3A%22integration%3A+hue%22" in caplog.text
+    )
+
+
+async def test_check_loop_async_custom(caplog: pytest.LogCaptureFixture) -> None:
+    """Test check_loop detects when called from event loop with custom component context."""
+    frames = extract_stack_to_frame(
+        [
+            Mock(
+                filename="/home/paulus/homeassistant/core.py",
+                lineno="23",
+                line="do_something()",
+            ),
+            Mock(
+                filename="/home/paulus/config/custom_components/hue/light.py",
+                lineno="23",
+                line="self.light.is_on",
+            ),
+            Mock(
+                filename="/home/paulus/aiohue/lights.py",
+                lineno="2",
+                line="something()",
+            ),
+        ]
+    )
+    with (
+        pytest.raises(RuntimeError),
+        patch(
+            "homeassistant.helpers.frame.linecache.getline",
+            return_value="self.light.is_on",
+        ),
+        patch(
+            "homeassistant.util.loop._get_line_from_cache",
+            return_value="mock_line",
+        ),
+        patch(
+            "homeassistant.util.loop.get_current_frame",
+            return_value=frames,
+        ),
+        patch(
+            "homeassistant.helpers.frame.get_current_frame",
+            return_value=frames,
+        ),
+    ):
+        haloop.check_loop(banned_function)
+    assert (
+        "Detected blocking call to banned_function inside the event loop by custom "
+        "integration 'hue' at custom_components/hue/light.py, line 23: self.light.is_on"
+        " (offender: /home/paulus/aiohue/lights.py, line 2: mock_line), "
+        "please create a bug report at https://github.com/home-assistant/core/issues?"
+        "q=is%3Aopen+is%3Aissue+label%3A%22integration%3A+hue%22"
+    ) in caplog.text
+
+
+def test_check_loop_sync(caplog: pytest.LogCaptureFixture) -> None:
+    """Test check_loop does nothing when called from thread."""
+    haloop.check_loop(banned_function)
+    assert "Detected blocking call inside the event loop" not in caplog.text
+
+
+def test_protect_loop_sync() -> None:
+    """Test protect_loop calls check_loop."""
+    func = Mock()
+    with patch("homeassistant.util.loop.check_loop") as mock_check_loop:
+        haloop.protect_loop(func)(1, test=2)
+    mock_check_loop.assert_called_once_with(
+        func,
+        strict=True,
+        args=(1,),
+        check_allowed=None,
+        kwargs={"test": 2},
+        strict_core=True,
+    )
+    func.assert_called_once_with(1, test=2)