🪲 Verify that async debug is available before eval

async_eval/async_eval.py

@@ -15,15 +15,25 @@ def save_locals(frame: types.FrameType) -> None:
         ctypes.pythonapi.PyFrame_LocalsToFast(ctypes.py_object(frame), ctypes.c_int(1))
 
 
+def _noop(*_: Any, **__: Any) -> Any:  # pragma: no cover
+    return None
+
+
+try:
+    _ = verify_async_debug_available  # noqa
+except NameError:  # pragma: no cover
+    try:
+        from async_eval.asyncio_patch import verify_async_debug_available
+    except ImportError:
+        verify_async_debug_available = _noop
+
 try:
     _ = apply  # noqa
 except NameError:  # pragma: no cover
     try:
         from nest_asyncio import apply
     except ImportError:
-
-        def apply(_: Any = None) -> None:
-            pass
+        apply = _noop
 
 
 _ASYNC_EVAL_CODE_TEMPLATE = textwrap.dedent(
@@ -189,6 +199,7 @@ def async_eval(
     *,
     filename: str = "<eval>",
 ) -> Any:
+    verify_async_debug_available()
     apply()  # double check that loop is patched
 
     caller: types.FrameType = inspect.currentframe().f_back  # type: ignore

async_eval/asyncio_patch.py

@@ -12,12 +12,22 @@
         from nest_asyncio import _patch_loop, apply
     except ImportError:  # pragma: no cover
 
-        def _noop(*args: Any, **kwargs: Any) -> None:
+        def _noop(*_: Any, **__: Any) -> None:
             pass
 
         _patch_loop = apply = _noop
 
 
+try:
+    from trio._core._run import GLOBAL_RUN_CONTEXT
+except ImportError:  # pragma: no cover
+    GLOBAL_RUN_CONTEXT = object()
+
+
+def is_trio_not_running() -> bool:
+    return not hasattr(GLOBAL_RUN_CONTEXT, "runner")
+
+
 def get_current_loop() -> Optional[Any]:  # pragma: no cover
     try:
         return asyncio.get_running_loop()  # type: ignore
@@ -34,6 +44,22 @@ def is_async_debug_available(loop: Any = None) -> bool:
     return bool(loop.__class__.__module__.lstrip("_").startswith("asyncio"))
 
 
+def verify_async_debug_available() -> None:
+    if not is_trio_not_running():
+        raise RuntimeError(
+            "Can not evaluate async code with trio event loop. "
+            "Only native asyncio event loop can be used for async code evaluating."
+        )
+
+    if not is_async_debug_available():
+        cls = get_current_loop().__class__
+
+        raise RuntimeError(
+            f"Can not evaluate async code with event loop {cls.__module__}.{cls.__qualname__}. "
+            "Only native asyncio event loop can be used for async code evaluating."
+        )
+
+
 def patch_asyncio() -> None:
     if hasattr(sys, "__async_eval_patched__"):  # pragma: no cover
         return
@@ -82,4 +108,5 @@ def set_loop_wrapper(loop: AbstractEventLoop) -> None:
     "patch_asyncio",
     "get_current_loop",
     "is_async_debug_available",
+    "verify_async_debug_available",
 ]

async_eval/ext/pydevd/code.py

@@ -1,53 +1,21 @@
-from functools import partial
 from typing import Any
 
 
-def _noop(*_: Any, __return__: Any = True, **__: Any) -> Any:  # pragma: no cover
-    return True
+def _noop(*_: Any, **__: Any) -> Any:  # pragma: no cover
+    return False
 
 
 try:  # pragma: no cover
     # only for testing purposes
-    _ = is_async_debug_available  # noqa
-    _ = is_async_code  # type: ignore  # noqa
-    _ = get_current_loop  # type: ignore  # noqa
+    _ = is_async_code  # noqa
+    _ = verify_async_debug_available  # type: ignore  # noqa
 except NameError:  # pragma: no cover
     try:
         from async_eval.async_eval import is_async_code
-        from async_eval.asyncio_patch import (
-            get_current_loop,
-            is_async_debug_available,
-        )
+        from async_eval.asyncio_patch import verify_async_debug_available
     except ImportError:
-
         is_async_code = _noop
-        is_async_debug_available = _noop
-        get_current_loop = partial(_noop, __return__=None)
-
-try:
-    from trio._core._run import GLOBAL_RUN_CONTEXT
-
-    def is_trio_not_running() -> bool:
-        return not hasattr(GLOBAL_RUN_CONTEXT, "runner")
-
-except ImportError:  # pragma: no cover
-    is_trio_not_running = _noop
-
-
-def verify_async_debug_available() -> None:
-    if not is_trio_not_running():
-        raise RuntimeError(
-            "Can not evaluate async code with trio event loop. "
-            "Only native asyncio event loop can be used for async code evaluating."
-        )
-
-    if not is_async_debug_available():
-        cls = get_current_loop().__class__
-
-        raise RuntimeError(
-            f"Can not evaluate async code with event loop {cls.__module__}.{cls.__qualname__}. "
-            "Only native asyncio event loop can be used for async code evaluating."
-        )
+        verify_async_debug_available = _noop
 
 
 def make_code_async(code: str) -> str:

tests/test_asyncio_patch.py

@@ -2,7 +2,7 @@
 from asyncio import AbstractEventLoop, BaseEventLoop, get_event_loop_policy
 from platform import system
 
-from pytest import mark
+from pytest import mark, raises
 
 
 def _is_patched(loop: AbstractEventLoop) -> bool:
@@ -64,3 +64,34 @@ def _test_windows_asyncio_policy():
 )
 def test_windows_asyncio_policy(run_in_process):
     run_in_process(_test_windows_asyncio_policy)
+
+
+def test_async_evaluate_is_not_available_for_eventloop(mocker):
+    mocker.patch("async_eval.asyncio_patch.is_async_debug_available", return_value=False)
+
+    from async_eval.ext.pydevd.code import evaluate_expression
+
+    with raises(
+        RuntimeError,
+        match=r"^Can not evaluate async code with event loop .*\. "
+        r"Only native asyncio event loop can be used for async code evaluating.$",
+    ):
+        evaluate_expression(
+            object(),
+            object(),
+            "await regular()",
+            True,
+        )
+
+
+def test_async_evaluate_is_not_available_for_trio(mocker):
+    mocker.patch("async_eval.asyncio_patch.is_trio_not_running", return_value=False)
+
+    from async_eval.asyncio_patch import verify_async_debug_available
+
+    with raises(
+        RuntimeError,
+        match=r"^Can not evaluate async code with trio event loop. "
+        r"Only native asyncio event loop can be used for async code evaluating.$",
+    ):
+        verify_async_debug_available()

tests/test_pydevd_patch.py

@@ -1,7 +1,7 @@
 import sys
 from unittest.mock import MagicMock
 
-from pytest import fixture, mark, raises
+from pytest import fixture, mark
 
 from .utils import ctxmanager, regular  # noqa
 
@@ -120,42 +120,6 @@ def _with_locals():
     assert g.gi_frame.f_locals["f"] == 10
 
 
-def test_async_evaluate_is_not_available_for_eventloop(mocker):
-    mocker.patch("async_eval.ext.pydevd.code.is_async_debug_available", return_value=False)
-
-    from async_eval.ext.pydevd.code import evaluate_expression
-
-    with raises(
-        RuntimeError,
-        match=r"^Can not evaluate async code with event loop .*\. "
-        r"Only native asyncio event loop can be used for async code evaluating.$",
-    ):
-        evaluate_expression(
-            object(),
-            object(),
-            "await regular()",
-            True,
-        )
-
-
-def test_async_evaluate_is_not_available_for_trio(mocker):
-    mocker.patch("async_eval.ext.pydevd.code.is_trio_not_running", return_value=False)
-
-    from async_eval.ext.pydevd.code import evaluate_expression
-
-    with raises(
-        RuntimeError,
-        match=r"^Can not evaluate async code with trio event loop. "
-        r"Only native asyncio event loop can be used for async code evaluating.$",
-    ):
-        evaluate_expression(
-            object(),
-            object(),
-            "await regular()",
-            True,
-        )
-
-
 def test_pydevd_integration():
     from async_eval.ext.pydevd import generate_main_script