define("ioloop", type=str, default=None)
+
class RootHandler(RequestHandler):
def get(self):
self.write("Hello, world")
def _log(self):
pass
+
def handle_sigchld(sig, frame):
IOLoop.current().add_callback_from_signal(IOLoop.current().stop)
+
def main():
parse_command_line()
if options.ioloop:
for i in xrange(options.num_runs):
run()
+
def run():
io_loop = IOLoop(make_current=True)
app = Application([("/", RootHandler)])
io_loop.close()
io_loop.clear_current()
+
if __name__ == '__main__':
main()
# This removes noise from the results, but it's easy to change things
# in a way that completely invalidates the results.
+
@gen.engine
def e2(callback):
callback()
+
@gen.engine
def e1():
for i in range(10):
yield gen.Task(e2)
+
@gen.coroutine
def c2():
pass
+
@gen.coroutine
def c1():
for i in range(10):
yield c2()
+
def main():
parse_command_line()
t = Timer(e1)
results = t.timeit(options.num) / options.num
print('coroutine: %0.3f ms per iteration' % (results * 1000))
+
if __name__ == '__main__':
main()
from tornado import stack_context
+
class Benchmark(object):
def enter_exit(self, count):
"""Measures the overhead of the nested "with" statements
queue.append(stack_context.wrap(
functools.partial(self.call_wrapped_inner, queue, count - 1)))
+
class StackBenchmark(Benchmark):
def make_context(self):
return stack_context.StackContext(self.__context)
def __context(self):
yield
+
class ExceptionBenchmark(Benchmark):
def make_context(self):
return stack_context.ExceptionStackContext(self.__handle_exception)
def __handle_exception(self, typ, value, tb):
pass
+
def main():
base_cmd = [
sys.executable, '-m', 'timeit', '-s',
print(cmd)
subprocess.check_call(base_cmd + [cmd])
+
if __name__ == '__main__':
main()
</html>\
""")
+
def render():
tmpl.generate(**context)
+
def main():
parse_command_line()
if options.dump:
results = t.timeit(options.num) / options.num
print('%0.3f ms per iteration' % (results * 1000))
+
if __name__ == '__main__':
main()
from tornado import ioloop
from tornado import web
+
def start(port, root_directory="/tmp/s3", bucket_depth=0):
"""Starts the mock S3 server on the given port at the given path."""
application = S3Application(root_directory, bucket_depth)
default='__TODO:_GENERATE_YOUR_OWN_RANDOM_VALUE__',
help="signing key for secure cookies")
+
class BaseHandler(RequestHandler):
COOKIE_NAME = 'twitterdemo_user'
return None
return json_decode(user_json)
+
class MainHandler(BaseHandler, TwitterMixin):
@authenticated
@gen.coroutine
access_token=self.current_user['access_token'])
self.render('home.html', timeline=timeline)
+
class LoginHandler(BaseHandler, TwitterMixin):
@gen.coroutine
def get(self):
else:
yield self.authorize_redirect(callback_uri=self.request.full_url())
+
class LogoutHandler(BaseHandler):
def get(self):
self.clear_cookie(self.COOKIE_NAME)
+
def main():
parse_command_line(final=False)
parse_config_file(options.config_file)
logging.info('Listening on http://localhost:%d' % options.port)
IOLoop.current().start()
+
if __name__ == '__main__':
main()
def get(self):
self.render("index.html", messages=ChatSocketHandler.cache)
+
class ChatSocketHandler(tornado.websocket.WebSocketHandler):
waiters = set()
cache = []
from lib2to3.pgen2 import token
from lib2to3.fixer_util import FromImport, Name, Comma, Newline
+
# copied from fix_tuple_params.py
def is_docstring(stmt):
return isinstance(stmt, pytree.Node) and \
stmt.children[0].type == token.STRING
+
class FixFutureImports(fixer_base.BaseFix):
BM_compatible = True
define('family', default='unspec',
help='Address family to query: unspec, inet, or inet6')
+
@gen.coroutine
def main():
args = parse_command_line()
pprint.pformat(addrinfo)))
print()
+
if __name__ == '__main__':
IOLoop.instance().run_sync(main)
#'tornado.test.wsgi_test',
]
+
def all():
return unittest.defaultTestLoader.loadTestsFromNames(TEST_MODULES)
+
def main():
print "Content-Type: text/plain\r\n\r\n",
else:
raise
+
if __name__ == '__main__':
main()
from tornado import gen
+
@gen.coroutine
def hello():
yield gen.sleep(0.001)
from tornado.web import RequestHandler, Application, asynchronous
import unittest
+
class HelloHandler(RequestHandler):
def get(self):
self.write("Hello world")
+
class RedirectHandler(RequestHandler):
def get(self, path):
self.redirect(path, status=int(self.get_argument('status', '302')))
+
class PostHandler(RequestHandler):
def post(self):
assert self.get_argument('foo') == 'bar'
self.redirect('/hello', status=303)
+
class ChunkedHandler(RequestHandler):
@asynchronous
@gen.engine
yield gen.Task(self.flush)
self.finish()
+
class CacheHandler(RequestHandler):
def get(self, computed_etag):
self.write(computed_etag)
def compute_etag(self):
return self._write_buffer[0]
+
class TestMixin(object):
def get_handlers(self):
return [
headers=[('If-None-Match', etags)],
expected_status=200)
+
class DefaultHTTPTest(AsyncHTTPTestCase, TestMixin):
def get_app(self):
return Application(self.get_handlers(), **self.get_app_kwargs())
+
class GzipHTTPTest(AsyncHTTPTestCase, TestMixin):
def get_app(self):
return Application(self.get_handlers(), gzip=True, **self.get_app_kwargs())
rs.VARY_ETAG_DOESNT_CHANGE,
]
+
if __name__ == '__main__':
parse_command_line()
unittest.main()
define('url', default='ws://localhost:9001')
define('name', default='Tornado')
+
@gen.engine
def run_tests():
url = options.url + '/getCaseCount'
assert msg is None
IOLoop.instance().stop()
+
def main():
parse_command_line()
IOLoop.instance().start()
+
if __name__ == '__main__':
main()
define('port', default=9000)
+
class EchoHandler(WebSocketHandler):
def on_message(self, message):
self.write_message(message, binary=isinstance(message, bytes))
def get_compression_options(self):
return {}
+
if __name__ == '__main__':
parse_command_line()
app = Application([
PY_PACKAGES = ['tox', 'virtualenv', 'pip']
+
def download_to_cache(url, local_name=None):
if local_name is None:
local_name = url.split('/')[-1]
f.write(data)
return filename
+
def main():
if not os.path.exists(TMPDIR):
os.mkdir(TMPDIR)
sys.pypy_version_info < (5, 9),
'pypy3 5.8 has buggy ssl module')
+
def _detect_ipv6():
if not socket.has_ipv6:
# socket.has_ipv6 check reports whether ipv6 was present at compile
sock.close()
return True
+
skipIfNoIPv6 = unittest.skipIf(not _detect_ipv6(), 'ipv6 support not present')
projects / thirdparty / tornado.git / commitdiff
