cpython: fe842efbe1ed (original) (raw)
--- a/Lib/test/test_typing.py +++ b/Lib/test/test_typing.py @@ -517,6 +517,9 @@ class GenericTests(BaseTestCase): Y[str, str] def test_generic_errors(self):
T = TypeVar('T')[](#l1.7)
with self.assertRaises(TypeError):[](#l1.8)
Generic[T]()[](#l1.9) with self.assertRaises(TypeError):[](#l1.10) isinstance([], List[int])[](#l1.11) with self.assertRaises(TypeError):[](#l1.12)
@@ -1255,7 +1258,7 @@ ASYNCIO = sys.version_info[:2] >= (3, 5) ASYNCIO_TESTS = """ import asyncio -T_a = TypeVar('T') +T_a = TypeVar('T_a') class AwaitableWrapper(typing.Awaitable[T_a]): @@ -1404,6 +1407,24 @@ class CollectionsAbcTests(BaseTestCase): g.send(None) # Run foo() till completion, to avoid warning. @skipUnless(ASYNCIO, 'Python 3.5 and multithreading required')
- def test_coroutine(self):
ns = {}[](#l1.27)
exec([](#l1.28)
"async def foo():\n"[](#l1.29)
" return\n",[](#l1.30)
globals(), ns)[](#l1.31)
foo = ns['foo'][](#l1.32)
g = foo()[](#l1.33)
self.assertIsInstance(g, typing.Coroutine)[](#l1.34)
with self.assertRaises(TypeError):[](#l1.35)
isinstance(g, typing.Coroutine[int])[](#l1.36)
self.assertNotIsInstance(foo, typing.Coroutine)[](#l1.37)
try:[](#l1.38)
g.send(None)[](#l1.39)
except StopIteration:[](#l1.40)
pass[](#l1.41)
- @skipUnless(ASYNCIO, 'Python 3.5 and multithreading required') def test_async_iterable(self): base_it = range(10) # type: Iterator[int] it = AsyncIteratorWrapper(base_it)
--- a/Lib/typing.py +++ b/Lib/typing.py @@ -29,9 +29,6 @@ if sys.version_info[:2] >= (3, 3): # ABCs (from collections.abc). 'AbstractSet', # collections.abc.Set.
- 'Awaitable',
- 'AsyncIterator',
- 'AsyncIterable', 'ByteString', 'Container', 'Hashable', @@ -47,6 +44,14 @@ if sys.version_info[:2] >= (3, 3): 'Sequence', 'Sized', 'ValuesView',
The following are added depending on presence
of their non-generic counterparts in stdlib:
Awaitable,
AsyncIterator,
AsyncIterable,
Coroutine,
Collection,
ContextManager
# Structural checks, a.k.a. protocols. 'Reversible', @@ -1104,6 +1109,9 @@ class Generic(metaclass=GenericMeta): slots = () def new(cls, *args, **kwds):
if _geqv(cls, Generic):[](#l2.32)
raise TypeError("Type Generic cannot be instantiated; "[](#l2.33)
"it can be used only as a base class")[](#l2.34) return _generic_new(cls.__next_in_mro__, cls, *args, **kwds)[](#l2.35)
@@ -1639,8 +1647,16 @@ Hashable = collections_abc.Hashable # N if hasattr(collections_abc, 'Awaitable'): class Awaitable(Generic[T_co], extra=collections_abc.Awaitable): slots = () -else:
+ + +if hasattr(collections_abc, 'Coroutine'):
- class Coroutine(Awaitable[V_co], Generic[T_co, T_contra, V_co],
extra=collections_abc.Coroutine):[](#l2.50)
__slots__ = ()[](#l2.51)
if hasattr(collections_abc, 'AsyncIterable'): @@ -1652,9 +1668,8 @@ if hasattr(collections_abc, 'AsyncIterab extra=collections_abc.AsyncIterator): slots = () -else:
class Iterable(Generic[T_co], extra=collections_abc.Iterable):