import sys
import textwrap
import time
-import platform
import weakref
from tornado.concurrent import return_future, Future
from tornado.log import app_log
from tornado import stack_context
from tornado.testing import AsyncHTTPTestCase, AsyncTestCase, ExpectLog, gen_test
-from tornado.test.util import unittest, skipOnTravis
+from tornado.test.util import unittest, skipOnTravis, skipBefore33, skipBefore35, skipNotCPython, exec_test
from tornado.web import Application, RequestHandler, asynchronous, HTTPError
from tornado import gen
except ImportError:
futures = None
-skipBefore33 = unittest.skipIf(sys.version_info < (3, 3), 'PEP 380 (yield from) not available')
-skipBefore35 = unittest.skipIf(sys.version_info < (3, 5), 'PEP 492 (async/await) not available')
-skipNotCPython = unittest.skipIf(platform.python_implementation() != 'CPython',
- 'Not CPython implementation')
-
class GenEngineTest(AsyncTestCase):
def setUp(self):
@skipBefore33
@gen_test
def test_async_return(self):
- # It is a compile-time error to return a value in a generator
- # before Python 3.3, so we must test this with exec.
- # Flatten the real global and local namespace into our fake globals:
- # it's all global from the perspective of f().
- global_namespace = dict(globals(), **locals())
- local_namespace = {}
- exec(textwrap.dedent("""
+ namespace = exec_test(globals(), locals(), """
@gen.coroutine
def f():
yield gen.Task(self.io_loop.add_callback)
return 42
- """), global_namespace, local_namespace)
- result = yield local_namespace['f']()
+ """)
+ result = yield namespace['f']()
self.assertEqual(result, 42)
self.finished = True
# A yield statement exists but is not executed, which means
# this function "returns" via an exception. This exception
# doesn't happen before the exception handling is set up.
- global_namespace = dict(globals(), **locals())
- local_namespace = {}
- exec(textwrap.dedent("""
+ namespace = exec_test(globals(), locals(), """
@gen.coroutine
def f():
if True:
return 42
yield gen.Task(self.io_loop.add_callback)
- """), global_namespace, local_namespace)
- result = yield local_namespace['f']()
+ """)
+ result = yield namespace['f']()
self.assertEqual(result, 42)
self.finished = True
# This test verifies that an async function can await a
# yield-based gen.coroutine, and that a gen.coroutine
# (the test method itself) can yield an async function.
- global_namespace = dict(globals(), **locals())
- local_namespace = {}
- exec(textwrap.dedent("""
+ namespace = exec_test(globals(), locals(), """
async def f():
await gen.Task(self.io_loop.add_callback)
return 42
- """), global_namespace, local_namespace)
- result = yield local_namespace['f']()
+ """)
+ result = yield namespace['f']()
self.assertEqual(result, 42)
self.finished = True
@skipBefore35
@gen_test
def test_async_await_mixed_multi(self):
- global_namespace = dict(globals(), **locals())
- local_namespace = {}
- exec(textwrap.dedent("""
+ namespace = exec_test(globals(), locals(), """
async def f1():
await gen.Task(self.io_loop.add_callback)
return 42
- """), global_namespace, local_namespace)
+ """)
@gen.coroutine
def f2():
yield gen.Task(self.io_loop.add_callback)
raise gen.Return(43)
- results = yield [local_namespace['f1'](), f2()]
+ results = yield [namespace['f1'](), f2()]
self.assertEqual(results, [42, 43])
self.finished = True
# under the License.
from datetime import timedelta
-import sys
-import textwrap
from tornado import gen, locks
from tornado.gen import TimeoutError
from tornado.testing import gen_test, AsyncTestCase
-from tornado.test.util import unittest
-
-skipBefore35 = unittest.skipIf(sys.version_info < (3, 5), 'PEP 492 (async/await) not available')
+from tornado.test.util import unittest, skipBefore35, exec_test
class ConditionTest(AsyncTestCase):
# Repeat the above test using 'async with'.
sem = locks.Semaphore()
- global_namespace = dict(globals(), **locals())
- local_namespace = {}
- exec(textwrap.dedent("""
+ namespace = exec_test(globals(), locals(), """
async def f():
async with sem as yielded:
self.assertTrue(yielded is None)
- """), global_namespace, local_namespace)
- yield local_namespace['f']()
+ """)
+ yield namespace['f']()
# Semaphore was released and can be acquired again.
self.assertTrue(sem.acquire().done())
N = 5
history = []
- global_namespace = dict(globals(), **locals())
- local_namespace = {}
- exec(textwrap.dedent("""
+ namespace = exec_test(globals(), locals(), """
async def f(idx):
async with lock:
history.append(idx)
- """), global_namespace, local_namespace)
- futures = [local_namespace['f'](i) for i in range(N)]
+ """)
+ futures = [namespace['f'](i) for i in range(N)]
lock.release()
yield futures
self.assertEqual(list(range(N)), history)
from __future__ import absolute_import, division, print_function, with_statement
import os
+import platform
import socket
import sys
+import textwrap
from tornado.testing import bind_unused_port
skipIfNoIPv6 = unittest.skipIf(not socket.has_ipv6, 'ipv6 support not present')
+skipBefore33 = unittest.skipIf(sys.version_info < (3, 3), 'PEP 380 (yield from) not available')
+skipBefore35 = unittest.skipIf(sys.version_info < (3, 5), 'PEP 492 (async/await) not available')
+skipNotCPython = unittest.skipIf(platform.python_implementation() != 'CPython',
+ 'Not CPython implementation')
+
+
def refusing_port():
"""Returns a local port number that will refuse all connections.
conn.close()
server_socket.close()
return (client_socket.close, client_addr[1])
+
+
+def exec_test(caller_globals, caller_locals, s):
+ """Execute ``s`` in a given context and return the result namespace.
+
+ Used to define functions for tests in particular python
+ versions that would be syntax errors in older versions.
+ """
+ # Flatten the real global and local namespace into our fake
+ # globals: it's all global from the perspective of code defined
+ # in s.
+ global_namespace = dict(caller_globals, **caller_locals)
+ local_namespace = {}
+ exec(textwrap.dedent(s), global_namespace, local_namespace)
+ return local_namespace
projects / thirdparty / tornado.git / commitdiff
