if access_token:
all_args["access_token"] = access_token
all_args.update(args)
-
+
if all_args:
url += "?" + urllib.urlencode(all_args)
callback = self.async_callback(self._on_facebook_request, callback)
if request.client_cert is not None:
curl.setopt(pycurl.SSLCERT, request.client_cert)
-
+
if request.client_key is not None:
curl.setopt(pycurl.SSLKEY, request.client_key)
# which has limitations on third-party modules)
MySQLdb = None
+
class Connection(object):
"""A lightweight wrapper around MySQLdb DB-API connections.
for field_type in field_types:
CONVERSIONS[field_type] = [(FLAG.BINARY, str)] + CONVERSIONS[field_type]
-
# Alias some common MySQL exceptions
IntegrityError = MySQLdb.IntegrityError
OperationalError = MySQLdb.OperationalError
if (len(args) == 1 and len(kwargs) == 0 and
isinstance(args[0], HTTPHeaders)):
# Copy constructor
- for k,v in args[0].get_all():
- self.add(k,v)
+ for k, v in args[0].get_all():
+ self.add(k, v)
else:
# Dict-style initialization
self.update(*args, **kwargs)
WRITE_BUFFER_CHUNK_SIZE = 128 * 1024
if len(data) > WRITE_BUFFER_CHUNK_SIZE:
for i in range(0, len(data), WRITE_BUFFER_CHUNK_SIZE):
- self._write_buffer.append(data[i:i+WRITE_BUFFER_CHUNK_SIZE])
+ self._write_buffer.append(data[i:i + WRITE_BUFFER_CHUNK_SIZE])
else:
self._write_buffer.append(data)
self._write_callback = stack_context.wrap(callback)
else:
self._maybe_run_close_callback()
-
def _set_read_callback(self, callback):
assert not self._read_callback, "Already reading"
self._read_callback = callback
return None
return chunk
+
def _double_prefix(deque):
"""Grow by doubling, but don't split the second chunk just because the
first one is small.
return _unicode(value)
-
options = _Options()
"""Global options dictionary.
from tornado.platform import interface
from tornado.util import b
+
class Waker(interface.Waker):
"""Create an OS independent asynchronous pipe.
success = SetHandleInformation(fd, HANDLE_FLAG_INHERIT, 0)
if not success:
raise ctypes.GetLastError()
-
-
from tornado.util import raise_exc_info
+
class _State(threading.local):
def __init__(self):
self.contexts = ()
# regression test: repeated invocations of a gen-based
# function should not result in accumulated stack_contexts
from tornado import stack_context
+
@gen.engine
def inner(callback):
yield gen.Task(self.io_loop.add_callback)
callback()
+
@gen.engine
def outer():
for i in xrange(10):
from tornado.options import _Options
+
class OptionsTest(unittest.TestCase):
def setUp(self):
self.options = _Options()
assert not self.request.body
self.write("ok")
+
class HostEchoHandler(RequestHandler):
def get(self):
self.write(self.request.headers["Host"])
def test_deactivate(self):
deactivate_callbacks = []
+
def f1():
with StackContext(functools.partial(self.context, 'c1')) as c1:
deactivate_callbacks.append(c1)
self.io_loop.add_callback(f2)
+
def f2():
with StackContext(functools.partial(self.context, 'c2')) as c2:
deactivate_callbacks.append(c2)
self.io_loop.add_callback(f3)
+
def f3():
with StackContext(functools.partial(self.context, 'c3')) as c3:
deactivate_callbacks.append(c3)
self.io_loop.add_callback(f4)
+
def f4():
self.assertEqual(self.active_contexts, ['c1', 'c2', 'c3'])
deactivate_callbacks[1]()
# but it will be missing from the next iteration
self.assertEqual(self.active_contexts, ['c1', 'c2', 'c3'])
self.io_loop.add_callback(f5)
+
def f5():
self.assertEqual(self.active_contexts, ['c1', 'c3'])
self.stop()
self.assertEqual(render("foo.py", ["not a string"]),
b("""s = "['not a string']"\n"""))
+
class TemplateLoaderTest(LogTrapTestCase):
def setUp(self):
self.loader = Loader(os.path.join(os.path.dirname(__file__), "templates"))
clears the first timeout.
"""
self.io_loop.add_timeout(time.time() + 0.01, self.stop)
- self.wait(timeout = 0.02)
+ self.wait(timeout=0.02)
self.io_loop.add_timeout(time.time() + 0.03, self.stop)
- self.wait(timeout = 0.1)
+ self.wait(timeout=0.1)
+
class SetUpTearDownTest(unittest.TestCase):
def test_set_up_tear_down(self):
from tornado.util import import_object
from tornado.web import RequestHandler, Application
+
def save_signal_handlers():
saved = {}
for sig in [signal.SIGINT, signal.SIGTERM, signal.SIGCHLD]:
assert "twisted" not in repr(saved), repr(saved)
return saved
+
def restore_signal_handlers(saved):
for sig, handler in saved.iteritems():
signal.signal(sig, handler)
+
class ReactorTestCase(unittest.TestCase):
def setUp(self):
self._saved_signals = save_signal_handlers()
from tornado.util import raise_exc_info
+
class RaiseExcInfoTest(unittest.TestCase):
def test_two_arg_exception(self):
# This test would fail on python 3 if raise_exc_info were simply
class SetCookieOverwriteHandler(RequestHandler):
def get(self):
self.set_cookie("a", "b", domain="example.com")
- self.set_cookie("c", "d" ,domain="example.com")
+ self.set_cookie("c", "d", domain="example.com")
# A second call with the same name clobbers the first.
# Attributes from the first call are not carried over.
self.set_cookie("a", "e")
else:
raise Exception("didn't get permanent or status arguments")
+
class EmptyFlushCallbackHandler(RequestHandler):
@gen.engine
@asynchronous
class WebTest(AsyncHTTPTestCase, LogTrapTestCase):
COOKIE_SECRET = "WebTest.COOKIE_SECRET"
+
def get_app(self):
loader = DictLoader({
"linkify.html": "{% module linkify(message) %}",
response = self.fetch("/static_url/foo.txt")
self.assertEqual(response.body, b("/static/foo.42.txt"))
+
class NamedURLSpecGroupsTest(AsyncHTTPTestCase, LogTrapTestCase):
def get_app(self):
class EchoHandler(RequestHandler):
response = self.fetch("/unicode/bar")
self.assertEqual(response.body, b("bar"))
+
class ClearHeaderTest(SimpleHandlerTestCase):
class Handler(RequestHandler):
def get(self):
response = self.fetch("/")
self.assertTrue("h1" not in response.headers)
self.assertEqual(response.headers["h2"], "bar")
-
+
class Header304Test(SimpleHandlerTestCase):
class Handler(RequestHandler):
def get(self):
self.set_header("Content-Language", "en_US")
self.write("hello")
-
+
def test_304_headers(self):
response1 = self.fetch('/')
self.assertEqual(response1.headers["Content-Length"], "5")
self.assertEqual(response1.headers["Content-Language"], "en_US")
-
+
response2 = self.fetch('/', headers={
'If-None-Match': response1.headers["Etag"]})
self.assertEqual(response2.code, 304)
self.__failure = None
raise_exc_info(failure)
-
def run(self, result=None):
with StackContext(self._stack_context):
super(AsyncTestCase, self).run(result)
if uri: # don't try to redirect '/' to ''
if self.request.query:
uri += "?" + self.request.query
- self.redirect(uri, permanent = True)
+ self.redirect(uri, permanent=True)
return
else:
raise HTTPError(404)
uri = self.request.path + "/"
if self.request.query:
uri += "?" + self.request.query
- self.redirect(uri, permanent = True)
+ self.redirect(uri, permanent=True)
return
raise HTTPError(404)
return method(self, *args, **kwargs)
projects / thirdparty / tornado.git / commitdiff
