mirror of https://github.com/python/cpython.git
asyncio: Add a test for asyncio.iscoroutine().
Test that asyncio.iscoroutine() supports 'async def' coroutines and collections.abc.Coroutine types.
This commit is contained in:
parent
af928b65fc
commit
6c792bd65a
|
@ -1,6 +1,10 @@
|
||||||
"""Tests support for new syntax introduced by PEP 492."""
|
"""Tests support for new syntax introduced by PEP 492."""
|
||||||
|
|
||||||
|
import collections.abc
|
||||||
|
import gc
|
||||||
import unittest
|
import unittest
|
||||||
|
|
||||||
|
from test import support
|
||||||
from unittest import mock
|
from unittest import mock
|
||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
|
@ -83,5 +87,30 @@ async def reader():
|
||||||
self.assertEqual(data, [b'line1\n', b'line2\n', b'line3'])
|
self.assertEqual(data, [b'line1\n', b'line2\n', b'line3'])
|
||||||
|
|
||||||
|
|
||||||
|
class CoroutineTests(BaseTest):
|
||||||
|
|
||||||
|
def test_iscoroutine(self):
|
||||||
|
async def foo(): pass
|
||||||
|
|
||||||
|
f = foo()
|
||||||
|
try:
|
||||||
|
self.assertTrue(asyncio.iscoroutine(f))
|
||||||
|
finally:
|
||||||
|
f.close() # silence warning
|
||||||
|
|
||||||
|
class FakeCoro(collections.abc.Coroutine):
|
||||||
|
def send(self, value): pass
|
||||||
|
def throw(self, typ, val=None, tb=None): pass
|
||||||
|
|
||||||
|
fc = FakeCoro()
|
||||||
|
try:
|
||||||
|
self.assertTrue(asyncio.iscoroutine(fc))
|
||||||
|
finally:
|
||||||
|
# To make sure that ABCMeta caches are freed
|
||||||
|
# from FakeCoro ASAP.
|
||||||
|
fc = FakeCoro = None
|
||||||
|
support.gc_collect()
|
||||||
|
|
||||||
|
|
||||||
if __name__ == '__main__':
|
if __name__ == '__main__':
|
||||||
unittest.main()
|
unittest.main()
|
||||||
|
|
Loading…
Reference in New Issue