# those errors quiet as well.
# https://groups.google.com/forum/?fromgroups#!topic/python-tornado/ApucKJat1_0
if (self._is_connreset(err) or
- err.args[0] in (errno.EBADF, errno.ENOTCONN)):
+ err.args[0] in (errno.EBADF, errno.ENOTCONN)):
return self.close(exc_info=True)
raise
except AttributeError:
code=self.get_argument("code"))
self.write(user)
else:
- yield self.authorize_redirect(
- redirect_uri=self.request.full_url(),
- client_id=self.settings["facebook_api_key"],
- extra_params={"scope": "read_stream,offline_access"})
+ yield self.authorize_redirect(
+ redirect_uri=self.request.full_url(),
+ client_id=self.settings["facebook_api_key"],
+ extra_params={"scope": "read_stream,offline_access"})
class FacebookServerAccessTokenHandler(RequestHandler):
assert self.get_argument('access_token') == 'fake-access-token'
# return a fake user
self.finish({
- 'name': 'Foo',
- 'email': 'foo@example.com'
- })
+ 'name': 'Foo',
+ 'email': 'foo@example.com'
+ })
class GoogleOAuth2Test(AsyncHTTPTestCase):
# Exception logging may be explicitly quieted.
with self.assertRaises(RuntimeError):
- yield gen.Multi([self.async_exception(RuntimeError("error 1")),
- self.async_exception(RuntimeError("error 2"))],
- quiet_exceptions=RuntimeError)
+ yield gen.Multi([self.async_exception(RuntimeError("error 1")),
+ self.async_exception(RuntimeError("error 2"))],
+ quiet_exceptions=RuntimeError)
@gen_test
def test_multi_future_exceptions(self):
# Exception logging may be explicitly quieted.
with self.assertRaises(RuntimeError):
- yield gen.multi_future(
- [self.async_exception(RuntimeError("error 1")),
- self.async_exception(RuntimeError("error 2"))],
- quiet_exceptions=RuntimeError)
+ yield gen.multi_future(
+ [self.async_exception(RuntimeError("error 1")),
+ self.async_exception(RuntimeError("error 2"))],
+ quiet_exceptions=RuntimeError)
def test_arguments(self):
@gen.engine
self.finished = True
+
class GenSequenceHandler(RequestHandler):
@asynchronous
@gen.engine
yield namespace['f']()
self.assertTrue(self.finished)
-
@gen_test
def test_no_ref(self):
# In this usage, there is no direct hard reference to the
self.assertRaises((ValueError, IOError),
HTTPServer, application, ssl_options={
"certfile": "/__mising__.crt",
- })
+ })
self.assertRaises((ValueError, IOError),
HTTPServer, application, ssl_options={
"certfile": existing_certificate,
"keyfile": "/__missing__.key"
- })
+ })
# This actually works because both files exist
HTTPServer(application, ssl_options={
self.assertEqual(headers['quux'], 'xyzzy')
self.assertEqual(sorted(headers.get_all()), [('Foo', 'bar'), ('Quux', 'xyzzy')])
+
class FormatTimestampTest(unittest.TestCase):
# Make sure that all the input types are supported.
TIMESTAMP = 1359312200.503611
# Now that the loop is stopped, it is no longer current.
self.assertIsNone(IOLoop.current(instance=False))
-
def test_force_current(self):
self.io_loop = IOLoop(make_current=True)
self.assertIs(self.io_loop, IOLoop.current())
# level so we have to go around it.
stream = self.request.connection.detach()
stream.write(b"HTTP/1.0 200 OK\r\n\r\n"
- b"hello")
+ b"hello")
stream.close()
else:
self.finish('HTTP/1 required')
def get_app(self):
class SmallBody(RequestHandler):
def get(self):
- self.write("a"*1024*64)
+ self.write("a" * 1024 * 64)
class LargeBody(RequestHandler):
def get(self):
- self.write("a"*1024*100)
+ self.write("a" * 1024 * 100)
return Application([('/small', SmallBody),
('/large', LargeBody)])
def get_http_client(self):
- return SimpleAsyncHTTPClient(io_loop=self.io_loop, max_body_size=1024*64)
+ return SimpleAsyncHTTPClient(io_loop=self.io_loop, max_body_size=1024 * 64)
def test_small_body(self):
response = self.fetch('/small')
response.rethrow()
- self.assertEqual(response.body, b'a'*1024*64)
+ self.assertEqual(response.body, b'a' * 1024 * 64)
def test_large_body(self):
with ExpectLog(gen_log, "Malformed HTTP message from None: Content-Length too long"):
class LargeBody(RequestHandler):
def get(self):
- self.write("a"*1024*100)
+ self.write("a" * 1024 * 100)
return Application([('/large', LargeBody)])
def get_http_client(self):
# 100KB body with 64KB buffer
- return SimpleAsyncHTTPClient(io_loop=self.io_loop, max_body_size=1024*100, max_buffer_size=1024*64)
+ return SimpleAsyncHTTPClient(io_loop=self.io_loop, max_body_size=1024 * 100, max_buffer_size=1024 * 64)
def test_large_body(self):
response = self.fetch('/large')
response.rethrow()
- self.assertEqual(response.body, b'a'*1024*100)
+ self.assertEqual(response.body, b'a' * 1024 * 100)
def handle_stream(self, stream, address):
yield gen.moment
stream.close()
- 1/0
+ 1 / 0
server = client = None
try:
def test_details(self):
loader = DictLoader({
"foo.html": "\n\n{{",
- })
+ })
with self.assertRaises(ParseError) as cm:
loader.load("foo.html")
self.assertEqual("Missing end expression }} at foo.html:3",
templates = {
"foo.html": "\t\tfoo\n\n",
"bar.txt": "\t\tbar\n\n",
- }
+ }
loader = DictLoader(templates, whitespace='all')
self.assertEqual(loader.load("foo.html").generate(), b"\t\tfoo\n\n")
self.assertEqual(loader.load("bar.txt").generate(), b"\t\tbar\n\n")
class BuggyLoggingTest(SimpleHandlerTestCase):
class Handler(RequestHandler):
def get(self):
- 1/0
+ 1 / 0
def log_exception(self, typ, value, tb):
- 1/0
+ 1 / 0
def test_buggy_log_exception(self):
# Something gets logged even though the application's
projects / thirdparty / tornado.git / commitdiff
