Skip to content

Commit

Permalink
Moved to decorator; extra tests.
Browse files Browse the repository at this point in the history
  • Loading branch information
carltongibson committed Nov 17, 2022
1 parent 513d358 commit 397a975
Show file tree
Hide file tree
Showing 2 changed files with 30 additions and 21 deletions.
12 changes: 11 additions & 1 deletion Lib/inspect.py
Original file line number Diff line number Diff line change
Expand Up @@ -125,6 +125,7 @@
"ismodule",
"isroutine",
"istraceback",
"markcoroutinefunction",
"signature",
"stack",
"trace",
Expand Down Expand Up @@ -391,13 +392,22 @@ def isgeneratorfunction(obj):
See help(isfunction) for a list of attributes."""
return _has_code_flag(obj, CO_GENERATOR)

def markcoroutinefunction(func):
"""
Decorator to ensure callable is recognised as a coroutine function.
"""
func.__code__ = func.__code__.replace(
co_flags=func.__code__.co_flags | CO_COROUTINE
)
return func

def iscoroutinefunction(obj):
"""Return true if the object is a coroutine function.
Coroutine functions are defined with "async def" syntax.
"""
return _has_code_flag(obj, CO_COROUTINE) or (
callable(obj) and _has_code_flag(obj.__call__, CO_COROUTINE)
not isclass(obj) and callable(obj) and _has_code_flag(obj.__call__, CO_COROUTINE)
)

def isasyncgenfunction(obj):
Expand Down
39 changes: 19 additions & 20 deletions Lib/test/test_inspect.py
Original file line number Diff line number Diff line change
Expand Up @@ -202,30 +202,29 @@ def test_iscoroutine(self):
gen_coroutine_function_example))))
self.assertTrue(inspect.isgenerator(gen_coro))

# Use subtest initially to see both failures.
with self.subTest("Wrapper not recognised."):
# First case: sync function returning an awaitable.
async def _fn3():
pass
async def _fn3():
pass

def fn3():
return _fn3()
@inspect.markcoroutinefunction
def fn3():
return _fn3()

# TODO: Move this to decorator function.
fn3.__code__ = fn3.__code__.replace(
co_flags=fn3.__code__.co_flags | inspect.CO_COROUTINE
)
self.assertTrue(inspect.iscoroutinefunction(fn3))

self.assertTrue(inspect.iscoroutinefunction(fn3))
class Cl:
async def __call__(self):
pass

self.assertFalse(inspect.iscoroutinefunction(Cl))
self.assertTrue(inspect.iscoroutinefunction(Cl()))

class Cl2:
@inspect.markcoroutinefunction
def __call__(self):
pass

with self.subTest("Awaitable instance not recongnised."):
# Second case: a class with an async def __call__.
# - instance is awaitable.
class Cl:
async def __call__(self):
pass
cl = Cl()
self.assertTrue(inspect.iscoroutinefunction(cl))
self.assertFalse(inspect.iscoroutinefunction(Cl2))
self.assertTrue(inspect.iscoroutinefunction(Cl2()))

self.assertFalse(
inspect.iscoroutinefunction(unittest.mock.Mock()))
Expand Down

0 comments on commit 397a975

Please sign in to comment.