from __future__ import annotations
import functools
-import inspect
import sys
from collections.abc import Awaitable, Generator
from contextlib import AbstractAsyncContextManager, contextmanager
from starlette.types import Scope
if sys.version_info >= (3, 13): # pragma: no cover
+ from inspect import iscoroutinefunction
from typing import TypeIs
else: # pragma: no cover
+ from asyncio import iscoroutinefunction
+
from typing_extensions import TypeIs
has_exceptiongroups = True
while isinstance(obj, functools.partial):
obj = obj.func
- return inspect.iscoroutinefunction(obj) or (callable(obj) and inspect.iscoroutinefunction(obj.__call__))
+ return iscoroutinefunction(obj) or (callable(obj) and iscoroutinefunction(obj.__call__))
T_co = TypeVar("T_co", covariant=True)
import functools
from typing import Any
+from unittest.mock import create_autospec
import pytest
assert is_async_callable(nested_partial)
+def test_async_mocked_async_function() -> None:
+ async def async_func() -> None: ... # pragma: no cover
+
+ mock = create_autospec(async_func)
+ assert is_async_callable(mock)
+
+
@pytest.mark.parametrize(
"scope, expected_result",
[