", loader=None,
- compress_whitespace=None, autoescape=_UNSET):
- self.name = name
- if compress_whitespace is None:
- compress_whitespace = name.endswith(".html") or \
- name.endswith(".js")
- if autoescape is not _UNSET:
- self.autoescape = autoescape
- elif loader:
- self.autoescape = loader.autoescape
- else:
- self.autoescape = _DEFAULT_AUTOESCAPE
- self.namespace = loader.namespace if loader else {}
- reader = _TemplateReader(name, escape.native_str(template_string))
- self.file = _File(self, _parse(reader, self))
- self.code = self._generate_python(loader, compress_whitespace)
- self.loader = loader
- try:
- # Under python2.5, the fake filename used here must match
- # the module name used in __name__ below.
- # The dont_inherit flag prevents template.py's future imports
- # from being applied to the generated code.
- self.compiled = compile(
- escape.to_unicode(self.code),
- "%s.generated.py" % self.name.replace('.', '_'),
- "exec", dont_inherit=True)
- except Exception:
- formatted_code = _format_code(self.code).rstrip()
- app_log.error("%s code:\n%s", self.name, formatted_code)
- raise
-
- def generate(self, **kwargs):
- """Generate this template with the given arguments."""
- namespace = {
- "escape": escape.xhtml_escape,
- "xhtml_escape": escape.xhtml_escape,
- "url_escape": escape.url_escape,
- "json_encode": escape.json_encode,
- "squeeze": escape.squeeze,
- "linkify": escape.linkify,
- "datetime": datetime,
- "_tt_utf8": escape.utf8, # for internal use
- "_tt_string_types": (unicode_type, bytes_type),
- # __name__ and __loader__ allow the traceback mechanism to find
- # the generated source code.
- "__name__": self.name.replace('.', '_'),
- "__loader__": ObjectDict(get_source=lambda name: self.code),
- }
- namespace.update(self.namespace)
- namespace.update(kwargs)
- exec_in(self.compiled, namespace)
- execute = namespace["_tt_execute"]
- # Clear the traceback module's cache of source data now that
- # we've generated a new template (mainly for this module's
- # unittests, where different tests reuse the same name).
- linecache.clearcache()
- return execute()
-
- def _generate_python(self, loader, compress_whitespace):
- buffer = StringIO()
- try:
- # named_blocks maps from names to _NamedBlock objects
- named_blocks = {}
- ancestors = self._get_ancestors(loader)
- ancestors.reverse()
- for ancestor in ancestors:
- ancestor.find_named_blocks(loader, named_blocks)
- writer = _CodeWriter(buffer, named_blocks, loader, ancestors[0].template,
- compress_whitespace)
- ancestors[0].generate(writer)
- return buffer.getvalue()
- finally:
- buffer.close()
-
- def _get_ancestors(self, loader):
- ancestors = [self.file]
- for chunk in self.file.body.chunks:
- if isinstance(chunk, _ExtendsBlock):
- if not loader:
- raise ParseError("{% extends %} block found, but no "
- "template loader")
- template = loader.load(chunk.name, self.name)
- ancestors.extend(template._get_ancestors(loader))
- return ancestors
-
-
-class BaseLoader(object):
- """Base class for template loaders.
-
- You must use a template loader to use template constructs like
- ``{% extends %}`` and ``{% include %}``. The loader caches all
- templates after they are loaded the first time.
- """
- def __init__(self, autoescape=_DEFAULT_AUTOESCAPE, namespace=None):
- """``autoescape`` must be either None or a string naming a function
- in the template namespace, such as "xhtml_escape".
- """
- self.autoescape = autoescape
- self.namespace = namespace or {}
- self.templates = {}
- # self.lock protects self.templates. It's a reentrant lock
- # because templates may load other templates via `include` or
- # `extends`. Note that thanks to the GIL this code would be safe
- # even without the lock, but could lead to wasted work as multiple
- # threads tried to compile the same template simultaneously.
- self.lock = threading.RLock()
-
- def reset(self):
- """Resets the cache of compiled templates."""
- with self.lock:
- self.templates = {}
-
- def resolve_path(self, name, parent_path=None):
- """Converts a possibly-relative path to absolute (used internally)."""
- raise NotImplementedError()
-
- def load(self, name, parent_path=None):
- """Loads a template."""
- name = self.resolve_path(name, parent_path=parent_path)
- with self.lock:
- if name not in self.templates:
- self.templates[name] = self._create_template(name)
- return self.templates[name]
-
- def _create_template(self, name):
- raise NotImplementedError()
-
-
-class Loader(BaseLoader):
- """A template loader that loads from a single root directory.
- """
- def __init__(self, root_directory, **kwargs):
- super(Loader, self).__init__(**kwargs)
- self.root = os.path.abspath(root_directory)
-
- def resolve_path(self, name, parent_path=None):
- if parent_path and not parent_path.startswith("<") and \
- not parent_path.startswith("/") and \
- not name.startswith("/"):
- current_path = os.path.join(self.root, parent_path)
- file_dir = os.path.dirname(os.path.abspath(current_path))
- relative_path = os.path.abspath(os.path.join(file_dir, name))
- if relative_path.startswith(self.root):
- name = relative_path[len(self.root) + 1:]
- return name
-
- def _create_template(self, name):
- path = os.path.join(self.root, name)
- with open(path, "rb") as f:
- template = Template(f.read(), name=name, loader=self)
- return template
-
-
-class DictLoader(BaseLoader):
- """A template loader that loads from a dictionary."""
- def __init__(self, dict, **kwargs):
- super(DictLoader, self).__init__(**kwargs)
- self.dict = dict
-
- def resolve_path(self, name, parent_path=None):
- if parent_path and not parent_path.startswith("<") and \
- not parent_path.startswith("/") and \
- not name.startswith("/"):
- file_dir = posixpath.dirname(parent_path)
- name = posixpath.normpath(posixpath.join(file_dir, name))
- return name
-
- def _create_template(self, name):
- return Template(self.dict[name], name=name, loader=self)
-
-
-class _Node(object):
- def each_child(self):
- return ()
-
- def generate(self, writer):
- raise NotImplementedError()
-
- def find_named_blocks(self, loader, named_blocks):
- for child in self.each_child():
- child.find_named_blocks(loader, named_blocks)
-
-
-class _File(_Node):
- def __init__(self, template, body):
- self.template = template
- self.body = body
- self.line = 0
-
- def generate(self, writer):
- writer.write_line("def _tt_execute():", self.line)
- with writer.indent():
- writer.write_line("_tt_buffer = []", self.line)
- writer.write_line("_tt_append = _tt_buffer.append", self.line)
- self.body.generate(writer)
- writer.write_line("return _tt_utf8('').join(_tt_buffer)", self.line)
-
- def each_child(self):
- return (self.body,)
-
-
-class _ChunkList(_Node):
- def __init__(self, chunks):
- self.chunks = chunks
-
- def generate(self, writer):
- for chunk in self.chunks:
- chunk.generate(writer)
-
- def each_child(self):
- return self.chunks
-
-
-class _NamedBlock(_Node):
- def __init__(self, name, body, template, line):
- self.name = name
- self.body = body
- self.template = template
- self.line = line
-
- def each_child(self):
- return (self.body,)
-
- def generate(self, writer):
- block = writer.named_blocks[self.name]
- with writer.include(block.template, self.line):
- block.body.generate(writer)
-
- def find_named_blocks(self, loader, named_blocks):
- named_blocks[self.name] = self
- _Node.find_named_blocks(self, loader, named_blocks)
-
-
-class _ExtendsBlock(_Node):
- def __init__(self, name):
- self.name = name
-
-
-class _IncludeBlock(_Node):
- def __init__(self, name, reader, line):
- self.name = name
- self.template_name = reader.name
- self.line = line
-
- def find_named_blocks(self, loader, named_blocks):
- included = loader.load(self.name, self.template_name)
- included.file.find_named_blocks(loader, named_blocks)
-
- def generate(self, writer):
- included = writer.loader.load(self.name, self.template_name)
- with writer.include(included, self.line):
- included.file.body.generate(writer)
-
-
-class _ApplyBlock(_Node):
- def __init__(self, method, line, body=None):
- self.method = method
- self.line = line
- self.body = body
-
- def each_child(self):
- return (self.body,)
-
- def generate(self, writer):
- method_name = "_tt_apply%d" % writer.apply_counter
- writer.apply_counter += 1
- writer.write_line("def %s():" % method_name, self.line)
- with writer.indent():
- writer.write_line("_tt_buffer = []", self.line)
- writer.write_line("_tt_append = _tt_buffer.append", self.line)
- self.body.generate(writer)
- writer.write_line("return _tt_utf8('').join(_tt_buffer)", self.line)
- writer.write_line("_tt_append(_tt_utf8(%s(%s())))" % (
- self.method, method_name), self.line)
-
-
-class _ControlBlock(_Node):
- def __init__(self, statement, line, body=None):
- self.statement = statement
- self.line = line
- self.body = body
-
- def each_child(self):
- return (self.body,)
-
- def generate(self, writer):
- writer.write_line("%s:" % self.statement, self.line)
- with writer.indent():
- self.body.generate(writer)
- # Just in case the body was empty
- writer.write_line("pass", self.line)
-
-
-class _IntermediateControlBlock(_Node):
- def __init__(self, statement, line):
- self.statement = statement
- self.line = line
-
- def generate(self, writer):
- # In case the previous block was empty
- writer.write_line("pass", self.line)
- writer.write_line("%s:" % self.statement, self.line, writer.indent_size() - 1)
-
-
-class _Statement(_Node):
- def __init__(self, statement, line):
- self.statement = statement
- self.line = line
-
- def generate(self, writer):
- writer.write_line(self.statement, self.line)
-
-
-class _Expression(_Node):
- def __init__(self, expression, line, raw=False):
- self.expression = expression
- self.line = line
- self.raw = raw
-
- def generate(self, writer):
- writer.write_line("_tt_tmp = %s" % self.expression, self.line)
- writer.write_line("if isinstance(_tt_tmp, _tt_string_types):"
- " _tt_tmp = _tt_utf8(_tt_tmp)", self.line)
- writer.write_line("else: _tt_tmp = _tt_utf8(str(_tt_tmp))", self.line)
- if not self.raw and writer.current_template.autoescape is not None:
- # In python3 functions like xhtml_escape return unicode,
- # so we have to convert to utf8 again.
- writer.write_line("_tt_tmp = _tt_utf8(%s(_tt_tmp))" %
- writer.current_template.autoescape, self.line)
- writer.write_line("_tt_append(_tt_tmp)", self.line)
-
-
-class _Module(_Expression):
- def __init__(self, expression, line):
- super(_Module, self).__init__("_tt_modules." + expression, line,
- raw=True)
-
-
-class _Text(_Node):
- def __init__(self, value, line):
- self.value = value
- self.line = line
-
- def generate(self, writer):
- value = self.value
-
- # Compress lots of white space to a single character. If the whitespace
- # breaks a line, have it continue to break a line, but just with a
- # single \n character
- if writer.compress_whitespace and "" not in value:
- value = re.sub(r"([\t ]+)", " ", value)
- value = re.sub(r"(\s*\n\s*)", "\n", value)
-
- if value:
- writer.write_line('_tt_append(%r)' % escape.utf8(value), self.line)
-
-
-class ParseError(Exception):
- """Raised for template syntax errors."""
- pass
-
-
-class _CodeWriter(object):
- def __init__(self, file, named_blocks, loader, current_template,
- compress_whitespace):
- self.file = file
- self.named_blocks = named_blocks
- self.loader = loader
- self.current_template = current_template
- self.compress_whitespace = compress_whitespace
- self.apply_counter = 0
- self.include_stack = []
- self._indent = 0
-
- def indent_size(self):
- return self._indent
-
- def indent(self):
- class Indenter(object):
- def __enter__(_):
- self._indent += 1
- return self
-
- def __exit__(_, *args):
- assert self._indent > 0
- self._indent -= 1
-
- return Indenter()
-
- def include(self, template, line):
- self.include_stack.append((self.current_template, line))
- self.current_template = template
-
- class IncludeTemplate(object):
- def __enter__(_):
- return self
-
- def __exit__(_, *args):
- self.current_template = self.include_stack.pop()[0]
-
- return IncludeTemplate()
-
- def write_line(self, line, line_number, indent=None):
- if indent is None:
- indent = self._indent
- line_comment = ' # %s:%d' % (self.current_template.name, line_number)
- if self.include_stack:
- ancestors = ["%s:%d" % (tmpl.name, lineno)
- for (tmpl, lineno) in self.include_stack]
- line_comment += ' (via %s)' % ', '.join(reversed(ancestors))
- print(" " * indent + line + line_comment, file=self.file)
-
-
-class _TemplateReader(object):
- def __init__(self, name, text):
- self.name = name
- self.text = text
- self.line = 1
- self.pos = 0
-
- def find(self, needle, start=0, end=None):
- assert start >= 0, start
- pos = self.pos
- start += pos
- if end is None:
- index = self.text.find(needle, start)
- else:
- end += pos
- assert end >= start
- index = self.text.find(needle, start, end)
- if index != -1:
- index -= pos
- return index
-
- def consume(self, count=None):
- if count is None:
- count = len(self.text) - self.pos
- newpos = self.pos + count
- self.line += self.text.count("\n", self.pos, newpos)
- s = self.text[self.pos:newpos]
- self.pos = newpos
- return s
-
- def remaining(self):
- return len(self.text) - self.pos
-
- def __len__(self):
- return self.remaining()
-
- def __getitem__(self, key):
- if type(key) is slice:
- size = len(self)
- start, stop, step = key.indices(size)
- if start is None:
- start = self.pos
- else:
- start += self.pos
- if stop is not None:
- stop += self.pos
- return self.text[slice(start, stop, step)]
- elif key < 0:
- return self.text[key]
- else:
- return self.text[self.pos + key]
-
- def __str__(self):
- return self.text[self.pos:]
-
-
-def _format_code(code):
- lines = code.splitlines()
- format = "%%%dd %%s\n" % len(repr(len(lines) + 1))
- return "".join([format % (i + 1, line) for (i, line) in enumerate(lines)])
-
-
-def _parse(reader, template, in_block=None, in_loop=None):
- body = _ChunkList([])
- while True:
- # Find next template directive
- curly = 0
- while True:
- curly = reader.find("{", curly)
- if curly == -1 or curly + 1 == reader.remaining():
- # EOF
- if in_block:
- raise ParseError("Missing {%% end %%} block for %s" %
- in_block)
- body.chunks.append(_Text(reader.consume(), reader.line))
- return body
- # If the first curly brace is not the start of a special token,
- # start searching from the character after it
- if reader[curly + 1] not in ("{", "%", "#"):
- curly += 1
- continue
- # When there are more than 2 curlies in a row, use the
- # innermost ones. This is useful when generating languages
- # like latex where curlies are also meaningful
- if (curly + 2 < reader.remaining() and
- reader[curly + 1] == '{' and reader[curly + 2] == '{'):
- curly += 1
- continue
- break
-
- # Append any text before the special token
- if curly > 0:
- cons = reader.consume(curly)
- body.chunks.append(_Text(cons, reader.line))
-
- start_brace = reader.consume(2)
- line = reader.line
-
- # Template directives may be escaped as "{{!" or "{%!".
- # In this case output the braces and consume the "!".
- # This is especially useful in conjunction with jquery templates,
- # which also use double braces.
- if reader.remaining() and reader[0] == "!":
- reader.consume(1)
- body.chunks.append(_Text(start_brace, line))
- continue
-
- # Comment
- if start_brace == "{#":
- end = reader.find("#}")
- if end == -1:
- raise ParseError("Missing end expression #} on line %d" % line)
- contents = reader.consume(end).strip()
- reader.consume(2)
- continue
-
- # Expression
- if start_brace == "{{":
- end = reader.find("}}")
- if end == -1:
- raise ParseError("Missing end expression }} on line %d" % line)
- contents = reader.consume(end).strip()
- reader.consume(2)
- if not contents:
- raise ParseError("Empty expression on line %d" % line)
- body.chunks.append(_Expression(contents, line))
- continue
-
- # Block
- assert start_brace == "{%", start_brace
- end = reader.find("%}")
- if end == -1:
- raise ParseError("Missing end block %%} on line %d" % line)
- contents = reader.consume(end).strip()
- reader.consume(2)
- if not contents:
- raise ParseError("Empty block tag ({%% %%}) on line %d" % line)
-
- operator, space, suffix = contents.partition(" ")
- suffix = suffix.strip()
-
- # Intermediate ("else", "elif", etc) blocks
- intermediate_blocks = {
- "else": set(["if", "for", "while", "try"]),
- "elif": set(["if"]),
- "except": set(["try"]),
- "finally": set(["try"]),
- }
- allowed_parents = intermediate_blocks.get(operator)
- if allowed_parents is not None:
- if not in_block:
- raise ParseError("%s outside %s block" %
- (operator, allowed_parents))
- if in_block not in allowed_parents:
- raise ParseError("%s block cannot be attached to %s block" % (operator, in_block))
- body.chunks.append(_IntermediateControlBlock(contents, line))
- continue
-
- # End tag
- elif operator == "end":
- if not in_block:
- raise ParseError("Extra {%% end %%} block on line %d" % line)
- return body
-
- elif operator in ("extends", "include", "set", "import", "from",
- "comment", "autoescape", "raw", "module"):
- if operator == "comment":
- continue
- if operator == "extends":
- suffix = suffix.strip('"').strip("'")
- if not suffix:
- raise ParseError("extends missing file path on line %d" % line)
- block = _ExtendsBlock(suffix)
- elif operator in ("import", "from"):
- if not suffix:
- raise ParseError("import missing statement on line %d" % line)
- block = _Statement(contents, line)
- elif operator == "include":
- suffix = suffix.strip('"').strip("'")
- if not suffix:
- raise ParseError("include missing file path on line %d" % line)
- block = _IncludeBlock(suffix, reader, line)
- elif operator == "set":
- if not suffix:
- raise ParseError("set missing statement on line %d" % line)
- block = _Statement(suffix, line)
- elif operator == "autoescape":
- fn = suffix.strip()
- if fn == "None":
- fn = None
- template.autoescape = fn
- continue
- elif operator == "raw":
- block = _Expression(suffix, line, raw=True)
- elif operator == "module":
- block = _Module(suffix, line)
- body.chunks.append(block)
- continue
-
- elif operator in ("apply", "block", "try", "if", "for", "while"):
- # parse inner body recursively
- if operator in ("for", "while"):
- block_body = _parse(reader, template, operator, operator)
- elif operator == "apply":
- # apply creates a nested function so syntactically it's not
- # in the loop.
- block_body = _parse(reader, template, operator, None)
- else:
- block_body = _parse(reader, template, operator, in_loop)
-
- if operator == "apply":
- if not suffix:
- raise ParseError("apply missing method name on line %d" % line)
- block = _ApplyBlock(suffix, line, block_body)
- elif operator == "block":
- if not suffix:
- raise ParseError("block missing name on line %d" % line)
- block = _NamedBlock(suffix, block_body, template, line)
- else:
- block = _ControlBlock(contents, line, block_body)
- body.chunks.append(block)
- continue
-
- elif operator in ("break", "continue"):
- if not in_loop:
- raise ParseError("%s outside %s block" % (operator, set(["for", "while"])))
- body.chunks.append(_Statement(contents, line))
- continue
-
- else:
- raise ParseError("unknown operator: %r" % operator)
diff --git a/rosbridge_server/src/tornado/test/README b/rosbridge_server/src/tornado/test/README
deleted file mode 100644
index 2d6195d80..000000000
--- a/rosbridge_server/src/tornado/test/README
+++ /dev/null
@@ -1,4 +0,0 @@
-Test coverage is almost non-existent, but it's a start. Be sure to
-set PYTHONPATH apprioriately (generally to the root directory of your
-tornado checkout) when running tests to make sure you're getting the
-version of the tornado package that you expect.
\ No newline at end of file
diff --git a/rosbridge_server/src/tornado/test/__init__.py b/rosbridge_server/src/tornado/test/__init__.py
deleted file mode 100644
index e69de29bb..000000000
diff --git a/rosbridge_server/src/tornado/test/__main__.py b/rosbridge_server/src/tornado/test/__main__.py
deleted file mode 100644
index 5953443b1..000000000
--- a/rosbridge_server/src/tornado/test/__main__.py
+++ /dev/null
@@ -1,14 +0,0 @@
-"""Shim to allow python -m tornado.test.
-
-This only works in python 2.7+.
-"""
-from __future__ import absolute_import, division, print_function, with_statement
-
-from tornado.test.runtests import all, main
-
-# tornado.testing.main autodiscovery relies on 'all' being present in
-# the main module, so import it here even though it is not used directly.
-# The following line prevents a pyflakes warning.
-all = all
-
-main()
diff --git a/rosbridge_server/src/tornado/test/auth_test.py b/rosbridge_server/src/tornado/test/auth_test.py
deleted file mode 100644
index 254e1ae13..000000000
--- a/rosbridge_server/src/tornado/test/auth_test.py
+++ /dev/null
@@ -1,451 +0,0 @@
-# These tests do not currently do much to verify the correct implementation
-# of the openid/oauth protocols, they just exercise the major code paths
-# and ensure that it doesn't blow up (e.g. with unicode/bytes issues in
-# python 3)
-
-
-from __future__ import absolute_import, division, print_function, with_statement
-from tornado.auth import OpenIdMixin, OAuthMixin, OAuth2Mixin, TwitterMixin, GoogleMixin, AuthError
-from tornado.concurrent import Future
-from tornado.escape import json_decode
-from tornado import gen
-from tornado.log import gen_log
-from tornado.testing import AsyncHTTPTestCase, ExpectLog
-from tornado.util import u
-from tornado.web import RequestHandler, Application, asynchronous, HTTPError
-
-
-class OpenIdClientLoginHandler(RequestHandler, OpenIdMixin):
- def initialize(self, test):
- self._OPENID_ENDPOINT = test.get_url('/openid/server/authenticate')
-
- @asynchronous
- def get(self):
- if self.get_argument('openid.mode', None):
- self.get_authenticated_user(
- self.on_user, http_client=self.settings['http_client'])
- return
- res = self.authenticate_redirect()
- assert isinstance(res, Future)
- assert res.done()
-
- def on_user(self, user):
- if user is None:
- raise Exception("user is None")
- self.finish(user)
-
-
-class OpenIdServerAuthenticateHandler(RequestHandler):
- def post(self):
- if self.get_argument('openid.mode') != 'check_authentication':
- raise Exception("incorrect openid.mode %r")
- self.write('is_valid:true')
-
-
-class OAuth1ClientLoginHandler(RequestHandler, OAuthMixin):
- def initialize(self, test, version):
- self._OAUTH_VERSION = version
- self._OAUTH_REQUEST_TOKEN_URL = test.get_url('/oauth1/server/request_token')
- self._OAUTH_AUTHORIZE_URL = test.get_url('/oauth1/server/authorize')
- self._OAUTH_ACCESS_TOKEN_URL = test.get_url('/oauth1/server/access_token')
-
- def _oauth_consumer_token(self):
- return dict(key='asdf', secret='qwer')
-
- @asynchronous
- def get(self):
- if self.get_argument('oauth_token', None):
- self.get_authenticated_user(
- self.on_user, http_client=self.settings['http_client'])
- return
- res = self.authorize_redirect(http_client=self.settings['http_client'])
- assert isinstance(res, Future)
-
- def on_user(self, user):
- if user is None:
- raise Exception("user is None")
- self.finish(user)
-
- def _oauth_get_user(self, access_token, callback):
- if self.get_argument('fail_in_get_user', None):
- raise Exception("failing in get_user")
- if access_token != dict(key='uiop', secret='5678'):
- raise Exception("incorrect access token %r" % access_token)
- callback(dict(email='foo@example.com'))
-
-
-class OAuth1ClientLoginCoroutineHandler(OAuth1ClientLoginHandler):
- """Replaces OAuth1ClientLoginCoroutineHandler's get() with a coroutine."""
- @gen.coroutine
- def get(self):
- if self.get_argument('oauth_token', None):
- # Ensure that any exceptions are set on the returned Future,
- # not simply thrown into the surrounding StackContext.
- try:
- yield self.get_authenticated_user()
- except Exception as e:
- self.set_status(503)
- self.write("got exception: %s" % e)
- else:
- yield self.authorize_redirect()
-
-
-class OAuth1ClientRequestParametersHandler(RequestHandler, OAuthMixin):
- def initialize(self, version):
- self._OAUTH_VERSION = version
-
- def _oauth_consumer_token(self):
- return dict(key='asdf', secret='qwer')
-
- def get(self):
- params = self._oauth_request_parameters(
- 'http://www.example.com/api/asdf',
- dict(key='uiop', secret='5678'),
- parameters=dict(foo='bar'))
- self.write(params)
-
-
-class OAuth1ServerRequestTokenHandler(RequestHandler):
- def get(self):
- self.write('oauth_token=zxcv&oauth_token_secret=1234')
-
-
-class OAuth1ServerAccessTokenHandler(RequestHandler):
- def get(self):
- self.write('oauth_token=uiop&oauth_token_secret=5678')
-
-
-class OAuth2ClientLoginHandler(RequestHandler, OAuth2Mixin):
- def initialize(self, test):
- self._OAUTH_AUTHORIZE_URL = test.get_url('/oauth2/server/authorize')
-
- def get(self):
- res = self.authorize_redirect()
- assert isinstance(res, Future)
- assert res.done()
-
-
-class TwitterClientHandler(RequestHandler, TwitterMixin):
- def initialize(self, test):
- self._OAUTH_REQUEST_TOKEN_URL = test.get_url('/oauth1/server/request_token')
- self._OAUTH_ACCESS_TOKEN_URL = test.get_url('/twitter/server/access_token')
- self._OAUTH_AUTHORIZE_URL = test.get_url('/oauth1/server/authorize')
- self._TWITTER_BASE_URL = test.get_url('/twitter/api')
-
- def get_auth_http_client(self):
- return self.settings['http_client']
-
-
-class TwitterClientLoginHandler(TwitterClientHandler):
- @asynchronous
- def get(self):
- if self.get_argument("oauth_token", None):
- self.get_authenticated_user(self.on_user)
- return
- self.authorize_redirect()
-
- def on_user(self, user):
- if user is None:
- raise Exception("user is None")
- self.finish(user)
-
-
-class TwitterClientLoginGenEngineHandler(TwitterClientHandler):
- @asynchronous
- @gen.engine
- def get(self):
- if self.get_argument("oauth_token", None):
- user = yield self.get_authenticated_user()
- self.finish(user)
- else:
- # Old style: with @gen.engine we can ignore the Future from
- # authorize_redirect.
- self.authorize_redirect()
-
-
-class TwitterClientLoginGenCoroutineHandler(TwitterClientHandler):
- @gen.coroutine
- def get(self):
- if self.get_argument("oauth_token", None):
- user = yield self.get_authenticated_user()
- self.finish(user)
- else:
- # New style: with @gen.coroutine the result must be yielded
- # or else the request will be auto-finished too soon.
- yield self.authorize_redirect()
-
-
-class TwitterClientShowUserHandler(TwitterClientHandler):
- @asynchronous
- @gen.engine
- def get(self):
- # TODO: would be nice to go through the login flow instead of
- # cheating with a hard-coded access token.
- response = yield gen.Task(self.twitter_request,
- '/users/show/%s' % self.get_argument('name'),
- access_token=dict(key='hjkl', secret='vbnm'))
- if response is None:
- self.set_status(500)
- self.finish('error from twitter request')
- else:
- self.finish(response)
-
-
-class TwitterClientShowUserFutureHandler(TwitterClientHandler):
- @asynchronous
- @gen.engine
- def get(self):
- try:
- response = yield self.twitter_request(
- '/users/show/%s' % self.get_argument('name'),
- access_token=dict(key='hjkl', secret='vbnm'))
- except AuthError as e:
- self.set_status(500)
- self.finish(str(e))
- return
- assert response is not None
- self.finish(response)
-
-
-class TwitterServerAccessTokenHandler(RequestHandler):
- def get(self):
- self.write('oauth_token=hjkl&oauth_token_secret=vbnm&screen_name=foo')
-
-
-class TwitterServerShowUserHandler(RequestHandler):
- def get(self, screen_name):
- if screen_name == 'error':
- raise HTTPError(500)
- assert 'oauth_nonce' in self.request.arguments
- assert 'oauth_timestamp' in self.request.arguments
- assert 'oauth_signature' in self.request.arguments
- assert self.get_argument('oauth_consumer_key') == 'test_twitter_consumer_key'
- assert self.get_argument('oauth_signature_method') == 'HMAC-SHA1'
- assert self.get_argument('oauth_version') == '1.0'
- assert self.get_argument('oauth_token') == 'hjkl'
- self.write(dict(screen_name=screen_name, name=screen_name.capitalize()))
-
-
-class TwitterServerVerifyCredentialsHandler(RequestHandler):
- def get(self):
- assert 'oauth_nonce' in self.request.arguments
- assert 'oauth_timestamp' in self.request.arguments
- assert 'oauth_signature' in self.request.arguments
- assert self.get_argument('oauth_consumer_key') == 'test_twitter_consumer_key'
- assert self.get_argument('oauth_signature_method') == 'HMAC-SHA1'
- assert self.get_argument('oauth_version') == '1.0'
- assert self.get_argument('oauth_token') == 'hjkl'
- self.write(dict(screen_name='foo', name='Foo'))
-
-
-class GoogleOpenIdClientLoginHandler(RequestHandler, GoogleMixin):
- def initialize(self, test):
- self._OPENID_ENDPOINT = test.get_url('/openid/server/authenticate')
-
- @asynchronous
- def get(self):
- if self.get_argument("openid.mode", None):
- self.get_authenticated_user(self.on_user)
- return
- res = self.authenticate_redirect()
- assert isinstance(res, Future)
- assert res.done()
-
- def on_user(self, user):
- if user is None:
- raise Exception("user is None")
- self.finish(user)
-
- def get_auth_http_client(self):
- return self.settings['http_client']
-
-
-class AuthTest(AsyncHTTPTestCase):
- def get_app(self):
- return Application(
- [
- # test endpoints
- ('/openid/client/login', OpenIdClientLoginHandler, dict(test=self)),
- ('/oauth10/client/login', OAuth1ClientLoginHandler,
- dict(test=self, version='1.0')),
- ('/oauth10/client/request_params',
- OAuth1ClientRequestParametersHandler,
- dict(version='1.0')),
- ('/oauth10a/client/login', OAuth1ClientLoginHandler,
- dict(test=self, version='1.0a')),
- ('/oauth10a/client/login_coroutine',
- OAuth1ClientLoginCoroutineHandler,
- dict(test=self, version='1.0a')),
- ('/oauth10a/client/request_params',
- OAuth1ClientRequestParametersHandler,
- dict(version='1.0a')),
- ('/oauth2/client/login', OAuth2ClientLoginHandler, dict(test=self)),
-
- ('/twitter/client/login', TwitterClientLoginHandler, dict(test=self)),
- ('/twitter/client/login_gen_engine', TwitterClientLoginGenEngineHandler, dict(test=self)),
- ('/twitter/client/login_gen_coroutine', TwitterClientLoginGenCoroutineHandler, dict(test=self)),
- ('/twitter/client/show_user', TwitterClientShowUserHandler, dict(test=self)),
- ('/twitter/client/show_user_future', TwitterClientShowUserFutureHandler, dict(test=self)),
- ('/google/client/openid_login', GoogleOpenIdClientLoginHandler, dict(test=self)),
-
- # simulated servers
- ('/openid/server/authenticate', OpenIdServerAuthenticateHandler),
- ('/oauth1/server/request_token', OAuth1ServerRequestTokenHandler),
- ('/oauth1/server/access_token', OAuth1ServerAccessTokenHandler),
-
- ('/twitter/server/access_token', TwitterServerAccessTokenHandler),
- (r'/twitter/api/users/show/(.*)\.json', TwitterServerShowUserHandler),
- (r'/twitter/api/account/verify_credentials\.json', TwitterServerVerifyCredentialsHandler),
- ],
- http_client=self.http_client,
- twitter_consumer_key='test_twitter_consumer_key',
- twitter_consumer_secret='test_twitter_consumer_secret')
-
- def test_openid_redirect(self):
- response = self.fetch('/openid/client/login', follow_redirects=False)
- self.assertEqual(response.code, 302)
- self.assertTrue(
- '/openid/server/authenticate?' in response.headers['Location'])
-
- def test_openid_get_user(self):
- response = self.fetch('/openid/client/login?openid.mode=blah&openid.ns.ax=http://openid.net/srv/ax/1.0&openid.ax.type.email=http://axschema.org/contact/email&openid.ax.value.email=foo@example.com')
- response.rethrow()
- parsed = json_decode(response.body)
- self.assertEqual(parsed["email"], "foo@example.com")
-
- def test_oauth10_redirect(self):
- response = self.fetch('/oauth10/client/login', follow_redirects=False)
- self.assertEqual(response.code, 302)
- self.assertTrue(response.headers['Location'].endswith(
- '/oauth1/server/authorize?oauth_token=zxcv'))
- # the cookie is base64('zxcv')|base64('1234')
- self.assertTrue(
- '_oauth_request_token="enhjdg==|MTIzNA=="' in response.headers['Set-Cookie'],
- response.headers['Set-Cookie'])
-
- def test_oauth10_get_user(self):
- response = self.fetch(
- '/oauth10/client/login?oauth_token=zxcv',
- headers={'Cookie': '_oauth_request_token=enhjdg==|MTIzNA=='})
- response.rethrow()
- parsed = json_decode(response.body)
- self.assertEqual(parsed['email'], 'foo@example.com')
- self.assertEqual(parsed['access_token'], dict(key='uiop', secret='5678'))
-
- def test_oauth10_request_parameters(self):
- response = self.fetch('/oauth10/client/request_params')
- response.rethrow()
- parsed = json_decode(response.body)
- self.assertEqual(parsed['oauth_consumer_key'], 'asdf')
- self.assertEqual(parsed['oauth_token'], 'uiop')
- self.assertTrue('oauth_nonce' in parsed)
- self.assertTrue('oauth_signature' in parsed)
-
- def test_oauth10a_redirect(self):
- response = self.fetch('/oauth10a/client/login', follow_redirects=False)
- self.assertEqual(response.code, 302)
- self.assertTrue(response.headers['Location'].endswith(
- '/oauth1/server/authorize?oauth_token=zxcv'))
- # the cookie is base64('zxcv')|base64('1234')
- self.assertTrue(
- '_oauth_request_token="enhjdg==|MTIzNA=="' in response.headers['Set-Cookie'],
- response.headers['Set-Cookie'])
-
- def test_oauth10a_get_user(self):
- response = self.fetch(
- '/oauth10a/client/login?oauth_token=zxcv',
- headers={'Cookie': '_oauth_request_token=enhjdg==|MTIzNA=='})
- response.rethrow()
- parsed = json_decode(response.body)
- self.assertEqual(parsed['email'], 'foo@example.com')
- self.assertEqual(parsed['access_token'], dict(key='uiop', secret='5678'))
-
- def test_oauth10a_request_parameters(self):
- response = self.fetch('/oauth10a/client/request_params')
- response.rethrow()
- parsed = json_decode(response.body)
- self.assertEqual(parsed['oauth_consumer_key'], 'asdf')
- self.assertEqual(parsed['oauth_token'], 'uiop')
- self.assertTrue('oauth_nonce' in parsed)
- self.assertTrue('oauth_signature' in parsed)
-
- def test_oauth10a_get_user_coroutine_exception(self):
- response = self.fetch(
- '/oauth10a/client/login_coroutine?oauth_token=zxcv&fail_in_get_user=true',
- headers={'Cookie': '_oauth_request_token=enhjdg==|MTIzNA=='})
- self.assertEqual(response.code, 503)
-
- def test_oauth2_redirect(self):
- response = self.fetch('/oauth2/client/login', follow_redirects=False)
- self.assertEqual(response.code, 302)
- self.assertTrue('/oauth2/server/authorize?' in response.headers['Location'])
-
- def base_twitter_redirect(self, url):
- # Same as test_oauth10a_redirect
- response = self.fetch(url, follow_redirects=False)
- self.assertEqual(response.code, 302)
- self.assertTrue(response.headers['Location'].endswith(
- '/oauth1/server/authorize?oauth_token=zxcv'))
- # the cookie is base64('zxcv')|base64('1234')
- self.assertTrue(
- '_oauth_request_token="enhjdg==|MTIzNA=="' in response.headers['Set-Cookie'],
- response.headers['Set-Cookie'])
-
- def test_twitter_redirect(self):
- self.base_twitter_redirect('/twitter/client/login')
-
- def test_twitter_redirect_gen_engine(self):
- self.base_twitter_redirect('/twitter/client/login_gen_engine')
-
- def test_twitter_redirect_gen_coroutine(self):
- self.base_twitter_redirect('/twitter/client/login_gen_coroutine')
-
- def test_twitter_get_user(self):
- response = self.fetch(
- '/twitter/client/login?oauth_token=zxcv',
- headers={'Cookie': '_oauth_request_token=enhjdg==|MTIzNA=='})
- response.rethrow()
- parsed = json_decode(response.body)
- self.assertEqual(parsed,
- {u('access_token'): {u('key'): u('hjkl'),
- u('screen_name'): u('foo'),
- u('secret'): u('vbnm')},
- u('name'): u('Foo'),
- u('screen_name'): u('foo'),
- u('username'): u('foo')})
-
- def test_twitter_show_user(self):
- response = self.fetch('/twitter/client/show_user?name=somebody')
- response.rethrow()
- self.assertEqual(json_decode(response.body),
- {'name': 'Somebody', 'screen_name': 'somebody'})
-
- def test_twitter_show_user_error(self):
- with ExpectLog(gen_log, 'Error response HTTP 500'):
- response = self.fetch('/twitter/client/show_user?name=error')
- self.assertEqual(response.code, 500)
- self.assertEqual(response.body, b'error from twitter request')
-
- def test_twitter_show_user_future(self):
- response = self.fetch('/twitter/client/show_user_future?name=somebody')
- response.rethrow()
- self.assertEqual(json_decode(response.body),
- {'name': 'Somebody', 'screen_name': 'somebody'})
-
- def test_twitter_show_user_future_error(self):
- response = self.fetch('/twitter/client/show_user_future?name=error')
- self.assertEqual(response.code, 500)
- self.assertIn(b'Error response HTTP 500', response.body)
-
- def test_google_redirect(self):
- # same as test_openid_redirect
- response = self.fetch('/google/client/openid_login', follow_redirects=False)
- self.assertEqual(response.code, 302)
- self.assertTrue(
- '/openid/server/authenticate?' in response.headers['Location'])
-
- def test_google_get_user(self):
- response = self.fetch('/google/client/openid_login?openid.mode=blah&openid.ns.ax=http://openid.net/srv/ax/1.0&openid.ax.type.email=http://axschema.org/contact/email&openid.ax.value.email=foo@example.com', follow_redirects=False)
- response.rethrow()
- parsed = json_decode(response.body)
- self.assertEqual(parsed["email"], "foo@example.com")
diff --git a/rosbridge_server/src/tornado/test/concurrent_test.py b/rosbridge_server/src/tornado/test/concurrent_test.py
deleted file mode 100644
index 5e93ad6a4..000000000
--- a/rosbridge_server/src/tornado/test/concurrent_test.py
+++ /dev/null
@@ -1,336 +0,0 @@
-#!/usr/bin/env python
-#
-# Copyright 2012 Facebook
-#
-# Licensed under the Apache License, Version 2.0 (the "License"); you may
-# not use this file except in compliance with the License. You may obtain
-# a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-# License for the specific language governing permissions and limitations
-# under the License.
-from __future__ import absolute_import, division, print_function, with_statement
-
-import logging
-import re
-import socket
-import sys
-import traceback
-
-from tornado.concurrent import Future, return_future, ReturnValueIgnoredError
-from tornado.escape import utf8, to_unicode
-from tornado import gen
-from tornado.iostream import IOStream
-from tornado import stack_context
-from tornado.tcpserver import TCPServer
-from tornado.testing import AsyncTestCase, LogTrapTestCase, bind_unused_port, gen_test
-
-
-try:
- from concurrent import futures
-except ImportError:
- futures = None
-
-
-class ReturnFutureTest(AsyncTestCase):
- @return_future
- def sync_future(self, callback):
- callback(42)
-
- @return_future
- def async_future(self, callback):
- self.io_loop.add_callback(callback, 42)
-
- @return_future
- def immediate_failure(self, callback):
- 1 / 0
-
- @return_future
- def delayed_failure(self, callback):
- self.io_loop.add_callback(lambda: 1 / 0)
-
- @return_future
- def return_value(self, callback):
- # Note that the result of both running the callback and returning
- # a value (or raising an exception) is unspecified; with current
- # implementations the last event prior to callback resolution wins.
- return 42
-
- @return_future
- def no_result_future(self, callback):
- callback()
-
- def test_immediate_failure(self):
- with self.assertRaises(ZeroDivisionError):
- # The caller sees the error just like a normal function.
- self.immediate_failure(callback=self.stop)
- # The callback is not run because the function failed synchronously.
- self.io_loop.add_timeout(self.io_loop.time() + 0.05, self.stop)
- result = self.wait()
- self.assertIs(result, None)
-
- def test_return_value(self):
- with self.assertRaises(ReturnValueIgnoredError):
- self.return_value(callback=self.stop)
-
- def test_callback_kw(self):
- future = self.sync_future(callback=self.stop)
- result = self.wait()
- self.assertEqual(result, 42)
- self.assertEqual(future.result(), 42)
-
- def test_callback_positional(self):
- # When the callback is passed in positionally, future_wrap shouldn't
- # add another callback in the kwargs.
- future = self.sync_future(self.stop)
- result = self.wait()
- self.assertEqual(result, 42)
- self.assertEqual(future.result(), 42)
-
- def test_no_callback(self):
- future = self.sync_future()
- self.assertEqual(future.result(), 42)
-
- def test_none_callback_kw(self):
- # explicitly pass None as callback
- future = self.sync_future(callback=None)
- self.assertEqual(future.result(), 42)
-
- def test_none_callback_pos(self):
- future = self.sync_future(None)
- self.assertEqual(future.result(), 42)
-
- def test_async_future(self):
- future = self.async_future()
- self.assertFalse(future.done())
- self.io_loop.add_future(future, self.stop)
- future2 = self.wait()
- self.assertIs(future, future2)
- self.assertEqual(future.result(), 42)
-
- @gen_test
- def test_async_future_gen(self):
- result = yield self.async_future()
- self.assertEqual(result, 42)
-
- def test_delayed_failure(self):
- future = self.delayed_failure()
- self.io_loop.add_future(future, self.stop)
- future2 = self.wait()
- self.assertIs(future, future2)
- with self.assertRaises(ZeroDivisionError):
- future.result()
-
- def test_kw_only_callback(self):
- @return_future
- def f(**kwargs):
- kwargs['callback'](42)
- future = f()
- self.assertEqual(future.result(), 42)
-
- def test_error_in_callback(self):
- self.sync_future(callback=lambda future: 1 / 0)
- # The exception gets caught by our StackContext and will be re-raised
- # when we wait.
- self.assertRaises(ZeroDivisionError, self.wait)
-
- def test_no_result_future(self):
- future = self.no_result_future(self.stop)
- result = self.wait()
- self.assertIs(result, None)
- # result of this future is undefined, but not an error
- future.result()
-
- def test_no_result_future_callback(self):
- future = self.no_result_future(callback=lambda: self.stop())
- result = self.wait()
- self.assertIs(result, None)
- future.result()
-
- @gen_test
- def test_future_traceback(self):
- @return_future
- @gen.engine
- def f(callback):
- yield gen.Task(self.io_loop.add_callback)
- try:
- 1 / 0
- except ZeroDivisionError:
- self.expected_frame = traceback.extract_tb(
- sys.exc_info()[2], limit=1)[0]
- raise
- try:
- yield f()
- self.fail("didn't get expected exception")
- except ZeroDivisionError:
- tb = traceback.extract_tb(sys.exc_info()[2])
- self.assertIn(self.expected_frame, tb)
-
-# The following series of classes demonstrate and test various styles
-# of use, with and without generators and futures.
-
-
-class CapServer(TCPServer):
- def handle_stream(self, stream, address):
- logging.info("handle_stream")
- self.stream = stream
- self.stream.read_until(b"\n", self.handle_read)
-
- def handle_read(self, data):
- logging.info("handle_read")
- data = to_unicode(data)
- if data == data.upper():
- self.stream.write(b"error\talready capitalized\n")
- else:
- # data already has \n
- self.stream.write(utf8("ok\t%s" % data.upper()))
- self.stream.close()
-
-
-class CapError(Exception):
- pass
-
-
-class BaseCapClient(object):
- def __init__(self, port, io_loop):
- self.port = port
- self.io_loop = io_loop
-
- def process_response(self, data):
- status, message = re.match('(.*)\t(.*)\n', to_unicode(data)).groups()
- if status == 'ok':
- return message
- else:
- raise CapError(message)
-
-
-class ManualCapClient(BaseCapClient):
- def capitalize(self, request_data, callback=None):
- logging.info("capitalize")
- self.request_data = request_data
- self.stream = IOStream(socket.socket(), io_loop=self.io_loop)
- self.stream.connect(('127.0.0.1', self.port),
- callback=self.handle_connect)
- self.future = Future()
- if callback is not None:
- self.future.add_done_callback(
- stack_context.wrap(lambda future: callback(future.result())))
- return self.future
-
- def handle_connect(self):
- logging.info("handle_connect")
- self.stream.write(utf8(self.request_data + "\n"))
- self.stream.read_until(b'\n', callback=self.handle_read)
-
- def handle_read(self, data):
- logging.info("handle_read")
- self.stream.close()
- try:
- self.future.set_result(self.process_response(data))
- except CapError as e:
- self.future.set_exception(e)
-
-
-class DecoratorCapClient(BaseCapClient):
- @return_future
- def capitalize(self, request_data, callback):
- logging.info("capitalize")
- self.request_data = request_data
- self.stream = IOStream(socket.socket(), io_loop=self.io_loop)
- self.stream.connect(('127.0.0.1', self.port),
- callback=self.handle_connect)
- self.callback = callback
-
- def handle_connect(self):
- logging.info("handle_connect")
- self.stream.write(utf8(self.request_data + "\n"))
- self.stream.read_until(b'\n', callback=self.handle_read)
-
- def handle_read(self, data):
- logging.info("handle_read")
- self.stream.close()
- self.callback(self.process_response(data))
-
-
-class GeneratorCapClient(BaseCapClient):
- @return_future
- @gen.engine
- def capitalize(self, request_data, callback):
- logging.info('capitalize')
- stream = IOStream(socket.socket(), io_loop=self.io_loop)
- logging.info('connecting')
- yield gen.Task(stream.connect, ('127.0.0.1', self.port))
- stream.write(utf8(request_data + '\n'))
- logging.info('reading')
- data = yield gen.Task(stream.read_until, b'\n')
- logging.info('returning')
- stream.close()
- callback(self.process_response(data))
-
-
-class ClientTestMixin(object):
- def setUp(self):
- super(ClientTestMixin, self).setUp()
- self.server = CapServer(io_loop=self.io_loop)
- sock, port = bind_unused_port()
- self.server.add_sockets([sock])
- self.client = self.client_class(io_loop=self.io_loop, port=port)
-
- def tearDown(self):
- self.server.stop()
- super(ClientTestMixin, self).tearDown()
-
- def test_callback(self):
- self.client.capitalize("hello", callback=self.stop)
- result = self.wait()
- self.assertEqual(result, "HELLO")
-
- def test_callback_error(self):
- self.client.capitalize("HELLO", callback=self.stop)
- self.assertRaisesRegexp(CapError, "already capitalized", self.wait)
-
- def test_future(self):
- future = self.client.capitalize("hello")
- self.io_loop.add_future(future, self.stop)
- self.wait()
- self.assertEqual(future.result(), "HELLO")
-
- def test_future_error(self):
- future = self.client.capitalize("HELLO")
- self.io_loop.add_future(future, self.stop)
- self.wait()
- self.assertRaisesRegexp(CapError, "already capitalized", future.result)
-
- def test_generator(self):
- @gen.engine
- def f():
- result = yield self.client.capitalize("hello")
- self.assertEqual(result, "HELLO")
- self.stop()
- f()
- self.wait()
-
- def test_generator_error(self):
- @gen.engine
- def f():
- with self.assertRaisesRegexp(CapError, "already capitalized"):
- yield self.client.capitalize("HELLO")
- self.stop()
- f()
- self.wait()
-
-
-class ManualClientTest(ClientTestMixin, AsyncTestCase, LogTrapTestCase):
- client_class = ManualCapClient
-
-
-class DecoratorClientTest(ClientTestMixin, AsyncTestCase, LogTrapTestCase):
- client_class = DecoratorCapClient
-
-
-class GeneratorClientTest(ClientTestMixin, AsyncTestCase, LogTrapTestCase):
- client_class = GeneratorCapClient
diff --git a/rosbridge_server/src/tornado/test/csv_translations/fr_FR.csv b/rosbridge_server/src/tornado/test/csv_translations/fr_FR.csv
deleted file mode 100644
index 6321b6e7c..000000000
--- a/rosbridge_server/src/tornado/test/csv_translations/fr_FR.csv
+++ /dev/null
@@ -1 +0,0 @@
-"school","école"
diff --git a/rosbridge_server/src/tornado/test/curl_httpclient_test.py b/rosbridge_server/src/tornado/test/curl_httpclient_test.py
deleted file mode 100644
index 3873cf1e3..000000000
--- a/rosbridge_server/src/tornado/test/curl_httpclient_test.py
+++ /dev/null
@@ -1,122 +0,0 @@
-from __future__ import absolute_import, division, print_function, with_statement
-
-from hashlib import md5
-
-from tornado.escape import utf8
-from tornado.httpclient import HTTPRequest
-from tornado.stack_context import ExceptionStackContext
-from tornado.testing import AsyncHTTPTestCase
-from tornado.test import httpclient_test
-from tornado.test.util import unittest
-from tornado.web import Application, RequestHandler
-
-try:
- import pycurl
-except ImportError:
- pycurl = None
-
-if pycurl is not None:
- from tornado.curl_httpclient import CurlAsyncHTTPClient
-
-
-@unittest.skipIf(pycurl is None, "pycurl module not present")
-class CurlHTTPClientCommonTestCase(httpclient_test.HTTPClientCommonTestCase):
- def get_http_client(self):
- client = CurlAsyncHTTPClient(io_loop=self.io_loop,
- defaults=dict(allow_ipv6=False))
- # make sure AsyncHTTPClient magic doesn't give us the wrong class
- self.assertTrue(isinstance(client, CurlAsyncHTTPClient))
- return client
-
-
-class DigestAuthHandler(RequestHandler):
- def get(self):
- realm = 'test'
- opaque = 'asdf'
- # Real implementations would use a random nonce.
- nonce = "1234"
- username = 'foo'
- password = 'bar'
-
- auth_header = self.request.headers.get('Authorization', None)
- if auth_header is not None:
- auth_mode, params = auth_header.split(' ', 1)
- assert auth_mode == 'Digest'
- param_dict = {}
- for pair in params.split(','):
- k, v = pair.strip().split('=', 1)
- if v[0] == '"' and v[-1] == '"':
- v = v[1:-1]
- param_dict[k] = v
- assert param_dict['realm'] == realm
- assert param_dict['opaque'] == opaque
- assert param_dict['nonce'] == nonce
- assert param_dict['username'] == username
- assert param_dict['uri'] == self.request.path
- h1 = md5(utf8('%s:%s:%s' % (username, realm, password))).hexdigest()
- h2 = md5(utf8('%s:%s' % (self.request.method,
- self.request.path))).hexdigest()
- digest = md5(utf8('%s:%s:%s' % (h1, nonce, h2))).hexdigest()
- if digest == param_dict['response']:
- self.write('ok')
- else:
- self.write('fail')
- else:
- self.set_status(401)
- self.set_header('WWW-Authenticate',
- 'Digest realm="%s", nonce="%s", opaque="%s"' %
- (realm, nonce, opaque))
-
-
-class CustomReasonHandler(RequestHandler):
- def get(self):
- self.set_status(200, "Custom reason")
-
-
-class CustomFailReasonHandler(RequestHandler):
- def get(self):
- self.set_status(400, "Custom reason")
-
-
-@unittest.skipIf(pycurl is None, "pycurl module not present")
-class CurlHTTPClientTestCase(AsyncHTTPTestCase):
- def setUp(self):
- super(CurlHTTPClientTestCase, self).setUp()
- self.http_client = CurlAsyncHTTPClient(self.io_loop,
- defaults=dict(allow_ipv6=False))
-
- def get_app(self):
- return Application([
- ('/digest', DigestAuthHandler),
- ('/custom_reason', CustomReasonHandler),
- ('/custom_fail_reason', CustomFailReasonHandler),
- ])
-
- def test_prepare_curl_callback_stack_context(self):
- exc_info = []
-
- def error_handler(typ, value, tb):
- exc_info.append((typ, value, tb))
- self.stop()
- return True
-
- with ExceptionStackContext(error_handler):
- request = HTTPRequest(self.get_url('/'),
- prepare_curl_callback=lambda curl: 1 / 0)
- self.http_client.fetch(request, callback=self.stop)
- self.wait()
- self.assertEqual(1, len(exc_info))
- self.assertIs(exc_info[0][0], ZeroDivisionError)
-
- def test_digest_auth(self):
- response = self.fetch('/digest', auth_mode='digest',
- auth_username='foo', auth_password='bar')
- self.assertEqual(response.body, b'ok')
-
- def test_custom_reason(self):
- response = self.fetch('/custom_reason')
- self.assertEqual(response.reason, "Custom reason")
-
- def test_fail_custom_reason(self):
- response = self.fetch('/custom_fail_reason')
- self.assertEqual(str(response.error), "HTTP 400: Custom reason")
diff --git a/rosbridge_server/src/tornado/test/escape_test.py b/rosbridge_server/src/tornado/test/escape_test.py
deleted file mode 100644
index 9abc74803..000000000
--- a/rosbridge_server/src/tornado/test/escape_test.py
+++ /dev/null
@@ -1,217 +0,0 @@
-#!/usr/bin/env python
-
-
-from __future__ import absolute_import, division, print_function, with_statement
-import tornado.escape
-
-from tornado.escape import utf8, xhtml_escape, xhtml_unescape, url_escape, url_unescape, to_unicode, json_decode, json_encode
-from tornado.util import u, unicode_type, bytes_type
-from tornado.test.util import unittest
-
-linkify_tests = [
- # (input, linkify_kwargs, expected_output)
-
- ("hello http://world.com/!", {},
- u('hello http://world.com/!')),
-
- ("hello http://world.com/with?param=true&stuff=yes", {},
- u('hello http://world.com/with?param=true&stuff=yes')),
-
- # an opened paren followed by many chars killed Gruber's regex
- ("http://url.com/w(aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", {},
- u('http://url.com/w(aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa')),
-
- # as did too many dots at the end
- ("http://url.com/withmany.......................................", {},
- u('http://url.com/withmany.......................................')),
-
- ("http://url.com/withmany((((((((((((((((((((((((((((((((((a)", {},
- u('http://url.com/withmany((((((((((((((((((((((((((((((((((a)')),
-
- # some examples from http://daringfireball.net/2009/11/liberal_regex_for_matching_urls
- # plus a fex extras (such as multiple parentheses).
- ("http://foo.com/blah_blah", {},
- u('http://foo.com/blah_blah')),
-
- ("http://foo.com/blah_blah/", {},
- u('http://foo.com/blah_blah/')),
-
- ("(Something like http://foo.com/blah_blah)", {},
- u('(Something like http://foo.com/blah_blah)')),
-
- ("http://foo.com/blah_blah_(wikipedia)", {},
- u('http://foo.com/blah_blah_(wikipedia)')),
-
- ("http://foo.com/blah_(blah)_(wikipedia)_blah", {},
- u('http://foo.com/blah_(blah)_(wikipedia)_blah')),
-
- ("(Something like http://foo.com/blah_blah_(wikipedia))", {},
- u('(Something like http://foo.com/blah_blah_(wikipedia))')),
-
- ("http://foo.com/blah_blah.", {},
- u('http://foo.com/blah_blah.')),
-
- ("http://foo.com/blah_blah/.", {},
- u('http://foo.com/blah_blah/.')),
-
- ("", {},
- u('<http://foo.com/blah_blah>')),
-
- ("", {},
- u('<http://foo.com/blah_blah/>')),
-
- ("http://foo.com/blah_blah,", {},
- u('http://foo.com/blah_blah,')),
-
- ("http://www.example.com/wpstyle/?p=364.", {},
- u('http://www.example.com/wpstyle/?p=364.')),
-
- ("rdar://1234",
- {"permitted_protocols": ["http", "rdar"]},
- u('rdar://1234')),
-
- ("rdar:/1234",
- {"permitted_protocols": ["rdar"]},
- u('rdar:/1234')),
-
- ("http://userid:password@example.com:8080", {},
- u('http://userid:password@example.com:8080')),
-
- ("http://userid@example.com", {},
- u('http://userid@example.com')),
-
- ("http://userid@example.com:8080", {},
- u('http://userid@example.com:8080')),
-
- ("http://userid:password@example.com", {},
- u('http://userid:password@example.com')),
-
- ("message://%3c330e7f8409726r6a4ba78dkf1fd71420c1bf6ff@mail.gmail.com%3e",
- {"permitted_protocols": ["http", "message"]},
- u('message://%3c330e7f8409726r6a4ba78dkf1fd71420c1bf6ff@mail.gmail.com%3e')),
-
- (u("http://\u27a1.ws/\u4a39"), {},
- u('http://\u27a1.ws/\u4a39')),
-
- ("http://example.com", {},
- u('<tag>http://example.com</tag>')),
-
- ("Just a www.example.com link.", {},
- u('Just a www.example.com link.')),
-
- ("Just a www.example.com link.",
- {"require_protocol": True},
- u('Just a www.example.com link.')),
-
- ("A http://reallylong.com/link/that/exceedsthelenglimit.html",
- {"require_protocol": True, "shorten": True},
- u('A http://reallylong.com/link...')),
-
- ("A http://reallylongdomainnamethatwillbetoolong.com/hi!",
- {"shorten": True},
- u('A http://reallylongdomainnametha...!')),
-
- ("A file:///passwords.txt and http://web.com link", {},
- u('A file:///passwords.txt and http://web.com link')),
-
- ("A file:///passwords.txt and http://web.com link",
- {"permitted_protocols": ["file"]},
- u('A file:///passwords.txt and http://web.com link')),
-
- ("www.external-link.com",
- {"extra_params": 'rel="nofollow" class="external"'},
- u('www.external-link.com')),
-
- ("www.external-link.com and www.internal-link.com/blogs extra",
- {"extra_params": lambda href: 'class="internal"' if href.startswith("http://www.internal-link.com") else 'rel="nofollow" class="external"'},
- u('www.external-link.com and www.internal-link.com/blogs extra')),
-
- ("www.external-link.com",
- {"extra_params": lambda href: ' rel="nofollow" class="external" '},
- u('www.external-link.com')),
-]
-
-
-class EscapeTestCase(unittest.TestCase):
- def test_linkify(self):
- for text, kwargs, html in linkify_tests:
- linked = tornado.escape.linkify(text, **kwargs)
- self.assertEqual(linked, html)
-
- def test_xhtml_escape(self):
- tests = [
- ("", "<foo>"),
- (u(""), u("<foo>")),
- (b"", b"<foo>"),
-
- ("<>&\"'", "<>&"'"),
- ("&", "&"),
-
- (u("<\u00e9>"), u("<\u00e9>")),
- (b"<\xc3\xa9>", b"<\xc3\xa9>"),
- ]
- for unescaped, escaped in tests:
- self.assertEqual(utf8(xhtml_escape(unescaped)), utf8(escaped))
- self.assertEqual(utf8(unescaped), utf8(xhtml_unescape(escaped)))
-
- def test_url_escape_unicode(self):
- tests = [
- # byte strings are passed through as-is
- (u('\u00e9').encode('utf8'), '%C3%A9'),
- (u('\u00e9').encode('latin1'), '%E9'),
-
- # unicode strings become utf8
- (u('\u00e9'), '%C3%A9'),
- ]
- for unescaped, escaped in tests:
- self.assertEqual(url_escape(unescaped), escaped)
-
- def test_url_unescape_unicode(self):
- tests = [
- ('%C3%A9', u('\u00e9'), 'utf8'),
- ('%C3%A9', u('\u00c3\u00a9'), 'latin1'),
- ('%C3%A9', utf8(u('\u00e9')), None),
- ]
- for escaped, unescaped, encoding in tests:
- # input strings to url_unescape should only contain ascii
- # characters, but make sure the function accepts both byte
- # and unicode strings.
- self.assertEqual(url_unescape(to_unicode(escaped), encoding), unescaped)
- self.assertEqual(url_unescape(utf8(escaped), encoding), unescaped)
-
- def test_url_escape_quote_plus(self):
- unescaped = '+ #%'
- plus_escaped = '%2B+%23%25'
- escaped = '%2B%20%23%25'
- self.assertEqual(url_escape(unescaped), plus_escaped)
- self.assertEqual(url_escape(unescaped, plus=False), escaped)
- self.assertEqual(url_unescape(plus_escaped), unescaped)
- self.assertEqual(url_unescape(escaped, plus=False), unescaped)
- self.assertEqual(url_unescape(plus_escaped, encoding=None),
- utf8(unescaped))
- self.assertEqual(url_unescape(escaped, encoding=None, plus=False),
- utf8(unescaped))
-
- def test_escape_return_types(self):
- # On python2 the escape methods should generally return the same
- # type as their argument
- self.assertEqual(type(xhtml_escape("foo")), str)
- self.assertEqual(type(xhtml_escape(u("foo"))), unicode_type)
-
- def test_json_decode(self):
- # json_decode accepts both bytes and unicode, but strings it returns
- # are always unicode.
- self.assertEqual(json_decode(b'"foo"'), u("foo"))
- self.assertEqual(json_decode(u('"foo"')), u("foo"))
-
- # Non-ascii bytes are interpreted as utf8
- self.assertEqual(json_decode(utf8(u('"\u00e9"'))), u("\u00e9"))
-
- def test_json_encode(self):
- # json deals with strings, not bytes. On python 2 byte strings will
- # convert automatically if they are utf8; on python 3 byte strings
- # are not allowed.
- self.assertEqual(json_decode(json_encode(u("\u00e9"))), u("\u00e9"))
- if bytes_type is str:
- self.assertEqual(json_decode(json_encode(utf8(u("\u00e9")))), u("\u00e9"))
- self.assertRaises(UnicodeDecodeError, json_encode, b"\xe9")
diff --git a/rosbridge_server/src/tornado/test/gen_test.py b/rosbridge_server/src/tornado/test/gen_test.py
deleted file mode 100644
index a15cdf73a..000000000
--- a/rosbridge_server/src/tornado/test/gen_test.py
+++ /dev/null
@@ -1,1071 +0,0 @@
-from __future__ import absolute_import, division, print_function, with_statement
-
-import contextlib
-import datetime
-import functools
-import sys
-import textwrap
-import time
-import platform
-import weakref
-
-from tornado.concurrent import return_future, Future
-from tornado.escape import url_escape
-from tornado.httpclient import AsyncHTTPClient
-from tornado.ioloop import IOLoop
-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.web import Application, RequestHandler, asynchronous, HTTPError
-
-from tornado import gen
-
-try:
- from concurrent import futures
-except ImportError:
- futures = None
-
-skipBefore33 = unittest.skipIf(sys.version_info < (3, 3), 'PEP 380 not available')
-skipNotCPython = unittest.skipIf(platform.python_implementation() != 'CPython',
- 'Not CPython implementation')
-
-
-class GenEngineTest(AsyncTestCase):
- def setUp(self):
- super(GenEngineTest, self).setUp()
- self.named_contexts = []
-
- def named_context(self, name):
- @contextlib.contextmanager
- def context():
- self.named_contexts.append(name)
- try:
- yield
- finally:
- self.assertEqual(self.named_contexts.pop(), name)
- return context
-
- def run_gen(self, f):
- f()
- return self.wait()
-
- def delay_callback(self, iterations, callback, arg):
- """Runs callback(arg) after a number of IOLoop iterations."""
- if iterations == 0:
- callback(arg)
- else:
- self.io_loop.add_callback(functools.partial(
- self.delay_callback, iterations - 1, callback, arg))
-
- @return_future
- def async_future(self, result, callback):
- self.io_loop.add_callback(callback, result)
-
- def test_no_yield(self):
- @gen.engine
- def f():
- self.stop()
- self.run_gen(f)
-
- def test_inline_cb(self):
- @gen.engine
- def f():
- (yield gen.Callback("k1"))()
- res = yield gen.Wait("k1")
- self.assertTrue(res is None)
- self.stop()
- self.run_gen(f)
-
- def test_ioloop_cb(self):
- @gen.engine
- def f():
- self.io_loop.add_callback((yield gen.Callback("k1")))
- yield gen.Wait("k1")
- self.stop()
- self.run_gen(f)
-
- def test_exception_phase1(self):
- @gen.engine
- def f():
- 1 / 0
- self.assertRaises(ZeroDivisionError, self.run_gen, f)
-
- def test_exception_phase2(self):
- @gen.engine
- def f():
- self.io_loop.add_callback((yield gen.Callback("k1")))
- yield gen.Wait("k1")
- 1 / 0
- self.assertRaises(ZeroDivisionError, self.run_gen, f)
-
- def test_exception_in_task_phase1(self):
- def fail_task(callback):
- 1 / 0
-
- @gen.engine
- def f():
- try:
- yield gen.Task(fail_task)
- raise Exception("did not get expected exception")
- except ZeroDivisionError:
- self.stop()
- self.run_gen(f)
-
- def test_exception_in_task_phase2(self):
- # This is the case that requires the use of stack_context in gen.engine
- def fail_task(callback):
- self.io_loop.add_callback(lambda: 1 / 0)
-
- @gen.engine
- def f():
- try:
- yield gen.Task(fail_task)
- raise Exception("did not get expected exception")
- except ZeroDivisionError:
- self.stop()
- self.run_gen(f)
-
- def test_with_arg(self):
- @gen.engine
- def f():
- (yield gen.Callback("k1"))(42)
- res = yield gen.Wait("k1")
- self.assertEqual(42, res)
- self.stop()
- self.run_gen(f)
-
- def test_with_arg_tuple(self):
- @gen.engine
- def f():
- (yield gen.Callback((1, 2)))((3, 4))
- res = yield gen.Wait((1, 2))
- self.assertEqual((3, 4), res)
- self.stop()
- self.run_gen(f)
-
- def test_key_reuse(self):
- @gen.engine
- def f():
- yield gen.Callback("k1")
- yield gen.Callback("k1")
- self.stop()
- self.assertRaises(gen.KeyReuseError, self.run_gen, f)
-
- def test_key_reuse_tuple(self):
- @gen.engine
- def f():
- yield gen.Callback((1, 2))
- yield gen.Callback((1, 2))
- self.stop()
- self.assertRaises(gen.KeyReuseError, self.run_gen, f)
-
- def test_key_mismatch(self):
- @gen.engine
- def f():
- yield gen.Callback("k1")
- yield gen.Wait("k2")
- self.stop()
- self.assertRaises(gen.UnknownKeyError, self.run_gen, f)
-
- def test_key_mismatch_tuple(self):
- @gen.engine
- def f():
- yield gen.Callback((1, 2))
- yield gen.Wait((2, 3))
- self.stop()
- self.assertRaises(gen.UnknownKeyError, self.run_gen, f)
-
- def test_leaked_callback(self):
- @gen.engine
- def f():
- yield gen.Callback("k1")
- self.stop()
- self.assertRaises(gen.LeakedCallbackError, self.run_gen, f)
-
- def test_leaked_callback_tuple(self):
- @gen.engine
- def f():
- yield gen.Callback((1, 2))
- self.stop()
- self.assertRaises(gen.LeakedCallbackError, self.run_gen, f)
-
- def test_parallel_callback(self):
- @gen.engine
- def f():
- for k in range(3):
- self.io_loop.add_callback((yield gen.Callback(k)))
- yield gen.Wait(1)
- self.io_loop.add_callback((yield gen.Callback(3)))
- yield gen.Wait(0)
- yield gen.Wait(3)
- yield gen.Wait(2)
- self.stop()
- self.run_gen(f)
-
- def test_bogus_yield(self):
- @gen.engine
- def f():
- yield 42
- self.assertRaises(gen.BadYieldError, self.run_gen, f)
-
- def test_bogus_yield_tuple(self):
- @gen.engine
- def f():
- yield (1, 2)
- self.assertRaises(gen.BadYieldError, self.run_gen, f)
-
- def test_reuse(self):
- @gen.engine
- def f():
- self.io_loop.add_callback((yield gen.Callback(0)))
- yield gen.Wait(0)
- self.stop()
- self.run_gen(f)
- self.run_gen(f)
-
- def test_task(self):
- @gen.engine
- def f():
- yield gen.Task(self.io_loop.add_callback)
- self.stop()
- self.run_gen(f)
-
- def test_wait_all(self):
- @gen.engine
- def f():
- (yield gen.Callback("k1"))("v1")
- (yield gen.Callback("k2"))("v2")
- results = yield gen.WaitAll(["k1", "k2"])
- self.assertEqual(results, ["v1", "v2"])
- self.stop()
- self.run_gen(f)
-
- def test_exception_in_yield(self):
- @gen.engine
- def f():
- try:
- yield gen.Wait("k1")
- raise Exception("did not get expected exception")
- except gen.UnknownKeyError:
- pass
- self.stop()
- self.run_gen(f)
-
- def test_resume_after_exception_in_yield(self):
- @gen.engine
- def f():
- try:
- yield gen.Wait("k1")
- raise Exception("did not get expected exception")
- except gen.UnknownKeyError:
- pass
- (yield gen.Callback("k2"))("v2")
- self.assertEqual((yield gen.Wait("k2")), "v2")
- self.stop()
- self.run_gen(f)
-
- def test_orphaned_callback(self):
- @gen.engine
- def f():
- self.orphaned_callback = yield gen.Callback(1)
- try:
- self.run_gen(f)
- raise Exception("did not get expected exception")
- except gen.LeakedCallbackError:
- pass
- self.orphaned_callback()
-
- def test_multi(self):
- @gen.engine
- def f():
- (yield gen.Callback("k1"))("v1")
- (yield gen.Callback("k2"))("v2")
- results = yield [gen.Wait("k1"), gen.Wait("k2")]
- self.assertEqual(results, ["v1", "v2"])
- self.stop()
- self.run_gen(f)
-
- def test_multi_dict(self):
- @gen.engine
- def f():
- (yield gen.Callback("k1"))("v1")
- (yield gen.Callback("k2"))("v2")
- results = yield dict(foo=gen.Wait("k1"), bar=gen.Wait("k2"))
- self.assertEqual(results, dict(foo="v1", bar="v2"))
- self.stop()
- self.run_gen(f)
-
- # The following tests explicitly run with both gen.Multi
- # and gen.multi_future (Task returns a Future, so it can be used
- # with either).
- def test_multi_yieldpoint_delayed(self):
- @gen.engine
- def f():
- # callbacks run at different times
- responses = yield gen.Multi([
- gen.Task(self.delay_callback, 3, arg="v1"),
- gen.Task(self.delay_callback, 1, arg="v2"),
- ])
- self.assertEqual(responses, ["v1", "v2"])
- self.stop()
- self.run_gen(f)
-
- def test_multi_yieldpoint_dict_delayed(self):
- @gen.engine
- def f():
- # callbacks run at different times
- responses = yield gen.Multi(dict(
- foo=gen.Task(self.delay_callback, 3, arg="v1"),
- bar=gen.Task(self.delay_callback, 1, arg="v2"),
- ))
- self.assertEqual(responses, dict(foo="v1", bar="v2"))
- self.stop()
- self.run_gen(f)
-
- def test_multi_future_delayed(self):
- @gen.engine
- def f():
- # callbacks run at different times
- responses = yield gen.multi_future([
- gen.Task(self.delay_callback, 3, arg="v1"),
- gen.Task(self.delay_callback, 1, arg="v2"),
- ])
- self.assertEqual(responses, ["v1", "v2"])
- self.stop()
- self.run_gen(f)
-
- def test_multi_future_dict_delayed(self):
- @gen.engine
- def f():
- # callbacks run at different times
- responses = yield gen.multi_future(dict(
- foo=gen.Task(self.delay_callback, 3, arg="v1"),
- bar=gen.Task(self.delay_callback, 1, arg="v2"),
- ))
- self.assertEqual(responses, dict(foo="v1", bar="v2"))
- self.stop()
- self.run_gen(f)
-
- @skipOnTravis
- @gen_test
- def test_multi_performance(self):
- # Yielding a list used to have quadratic performance; make
- # sure a large list stays reasonable. On my laptop a list of
- # 2000 used to take 1.8s, now it takes 0.12.
- start = time.time()
- yield [gen.Task(self.io_loop.add_callback) for i in range(2000)]
- end = time.time()
- self.assertLess(end - start, 1.0)
-
- @gen_test
- def test_multi_empty(self):
- # Empty lists or dicts should return the same type.
- x = yield []
- self.assertTrue(isinstance(x, list))
- y = yield {}
- self.assertTrue(isinstance(y, dict))
-
- @gen_test
- def test_multi_mixed_types(self):
- # A YieldPoint (Wait) and Future (Task) can be combined
- # (and use the YieldPoint codepath)
- (yield gen.Callback("k1"))("v1")
- responses = yield [gen.Wait("k1"),
- gen.Task(self.delay_callback, 3, arg="v2")]
- self.assertEqual(responses, ["v1", "v2"])
-
- @gen_test
- def test_future(self):
- result = yield self.async_future(1)
- self.assertEqual(result, 1)
-
- @gen_test
- def test_multi_future(self):
- results = yield [self.async_future(1), self.async_future(2)]
- self.assertEqual(results, [1, 2])
-
- @gen_test
- def test_multi_dict_future(self):
- results = yield dict(foo=self.async_future(1), bar=self.async_future(2))
- self.assertEqual(results, dict(foo=1, bar=2))
-
- def test_arguments(self):
- @gen.engine
- def f():
- (yield gen.Callback("noargs"))()
- self.assertEqual((yield gen.Wait("noargs")), None)
- (yield gen.Callback("1arg"))(42)
- self.assertEqual((yield gen.Wait("1arg")), 42)
-
- (yield gen.Callback("kwargs"))(value=42)
- result = yield gen.Wait("kwargs")
- self.assertTrue(isinstance(result, gen.Arguments))
- self.assertEqual(((), dict(value=42)), result)
- self.assertEqual(dict(value=42), result.kwargs)
-
- (yield gen.Callback("2args"))(42, 43)
- result = yield gen.Wait("2args")
- self.assertTrue(isinstance(result, gen.Arguments))
- self.assertEqual(((42, 43), {}), result)
- self.assertEqual((42, 43), result.args)
-
- def task_func(callback):
- callback(None, error="foo")
- result = yield gen.Task(task_func)
- self.assertTrue(isinstance(result, gen.Arguments))
- self.assertEqual(((None,), dict(error="foo")), result)
-
- self.stop()
- self.run_gen(f)
-
- def test_stack_context_leak(self):
- # regression test: repeated invocations of a gen-based
- # function should not result in accumulated stack_contexts
- def _stack_depth():
- head = stack_context._state.contexts[1]
- length = 0
-
- while head is not None:
- length += 1
- head = head.old_contexts[1]
-
- return length
-
- @gen.engine
- def inner(callback):
- yield gen.Task(self.io_loop.add_callback)
- callback()
-
- @gen.engine
- def outer():
- for i in range(10):
- yield gen.Task(inner)
-
- stack_increase = _stack_depth() - initial_stack_depth
- self.assertTrue(stack_increase <= 2)
- self.stop()
- initial_stack_depth = _stack_depth()
- self.run_gen(outer)
-
- def test_stack_context_leak_exception(self):
- # same as previous, but with a function that exits with an exception
- @gen.engine
- def inner(callback):
- yield gen.Task(self.io_loop.add_callback)
- 1 / 0
-
- @gen.engine
- def outer():
- for i in range(10):
- try:
- yield gen.Task(inner)
- except ZeroDivisionError:
- pass
- stack_increase = len(stack_context._state.contexts) - initial_stack_depth
- self.assertTrue(stack_increase <= 2)
- self.stop()
- initial_stack_depth = len(stack_context._state.contexts)
- self.run_gen(outer)
-
- def function_with_stack_context(self, callback):
- # Technically this function should stack_context.wrap its callback
- # upon entry. However, it is very common for this step to be
- # omitted.
- def step2():
- self.assertEqual(self.named_contexts, ['a'])
- self.io_loop.add_callback(callback)
-
- with stack_context.StackContext(self.named_context('a')):
- self.io_loop.add_callback(step2)
-
- @gen_test
- def test_wait_transfer_stack_context(self):
- # Wait should not pick up contexts from where callback was invoked,
- # even if that function improperly fails to wrap its callback.
- cb = yield gen.Callback('k1')
- self.function_with_stack_context(cb)
- self.assertEqual(self.named_contexts, [])
- yield gen.Wait('k1')
- self.assertEqual(self.named_contexts, [])
-
- @gen_test
- def test_task_transfer_stack_context(self):
- yield gen.Task(self.function_with_stack_context)
- self.assertEqual(self.named_contexts, [])
-
- def test_raise_after_stop(self):
- # This pattern will be used in the following tests so make sure
- # the exception propagates as expected.
- @gen.engine
- def f():
- self.stop()
- 1 / 0
-
- with self.assertRaises(ZeroDivisionError):
- self.run_gen(f)
-
- def test_sync_raise_return(self):
- # gen.Return is allowed in @gen.engine, but it may not be used
- # to return a value.
- @gen.engine
- def f():
- self.stop(42)
- raise gen.Return()
-
- result = self.run_gen(f)
- self.assertEqual(result, 42)
-
- def test_async_raise_return(self):
- @gen.engine
- def f():
- yield gen.Task(self.io_loop.add_callback)
- self.stop(42)
- raise gen.Return()
-
- result = self.run_gen(f)
- self.assertEqual(result, 42)
-
- def test_sync_raise_return_value(self):
- @gen.engine
- def f():
- raise gen.Return(42)
-
- with self.assertRaises(gen.ReturnValueIgnoredError):
- self.run_gen(f)
-
- def test_sync_raise_return_value_tuple(self):
- @gen.engine
- def f():
- raise gen.Return((1, 2))
-
- with self.assertRaises(gen.ReturnValueIgnoredError):
- self.run_gen(f)
-
- def test_async_raise_return_value(self):
- @gen.engine
- def f():
- yield gen.Task(self.io_loop.add_callback)
- raise gen.Return(42)
-
- with self.assertRaises(gen.ReturnValueIgnoredError):
- self.run_gen(f)
-
- def test_async_raise_return_value_tuple(self):
- @gen.engine
- def f():
- yield gen.Task(self.io_loop.add_callback)
- raise gen.Return((1, 2))
-
- with self.assertRaises(gen.ReturnValueIgnoredError):
- self.run_gen(f)
-
- def test_return_value(self):
- # It is an error to apply @gen.engine to a function that returns
- # a value.
- @gen.engine
- def f():
- return 42
-
- with self.assertRaises(gen.ReturnValueIgnoredError):
- self.run_gen(f)
-
- def test_return_value_tuple(self):
- # It is an error to apply @gen.engine to a function that returns
- # a value.
- @gen.engine
- def f():
- return (1, 2)
-
- with self.assertRaises(gen.ReturnValueIgnoredError):
- self.run_gen(f)
-
- @skipNotCPython
- def test_task_refcounting(self):
- # On CPython, tasks and their arguments should be released immediately
- # without waiting for garbage collection.
- @gen.engine
- def f():
- class Foo(object):
- pass
- arg = Foo()
- self.arg_ref = weakref.ref(arg)
- task = gen.Task(self.io_loop.add_callback, arg=arg)
- self.task_ref = weakref.ref(task)
- yield task
- self.stop()
-
- self.run_gen(f)
- self.assertIs(self.arg_ref(), None)
- self.assertIs(self.task_ref(), None)
-
-
-class GenCoroutineTest(AsyncTestCase):
- def setUp(self):
- # Stray StopIteration exceptions can lead to tests exiting prematurely,
- # so we need explicit checks here to make sure the tests run all
- # the way through.
- self.finished = False
- super(GenCoroutineTest, self).setUp()
-
- def tearDown(self):
- super(GenCoroutineTest, self).tearDown()
- assert self.finished
-
- @gen_test
- def test_sync_gen_return(self):
- @gen.coroutine
- def f():
- raise gen.Return(42)
- result = yield f()
- self.assertEqual(result, 42)
- self.finished = True
-
- @gen_test
- def test_async_gen_return(self):
- @gen.coroutine
- def f():
- yield gen.Task(self.io_loop.add_callback)
- raise gen.Return(42)
- result = yield f()
- self.assertEqual(result, 42)
- self.finished = True
-
- @gen_test
- def test_sync_return(self):
- @gen.coroutine
- def f():
- return 42
- result = yield f()
- self.assertEqual(result, 42)
- self.finished = True
-
- @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("""
- @gen.coroutine
- def f():
- yield gen.Task(self.io_loop.add_callback)
- return 42
- """), global_namespace, local_namespace)
- result = yield local_namespace['f']()
- self.assertEqual(result, 42)
- self.finished = True
-
- @skipBefore33
- @gen_test
- def test_async_early_return(self):
- # 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("""
- @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']()
- self.assertEqual(result, 42)
- self.finished = True
-
- @gen_test
- def test_sync_return_no_value(self):
- @gen.coroutine
- def f():
- return
- result = yield f()
- self.assertEqual(result, None)
- self.finished = True
-
- @gen_test
- def test_async_return_no_value(self):
- # Without a return value we don't need python 3.3.
- @gen.coroutine
- def f():
- yield gen.Task(self.io_loop.add_callback)
- return
- result = yield f()
- self.assertEqual(result, None)
- self.finished = True
-
- @gen_test
- def test_sync_raise(self):
- @gen.coroutine
- def f():
- 1 / 0
- # The exception is raised when the future is yielded
- # (or equivalently when its result method is called),
- # not when the function itself is called).
- future = f()
- with self.assertRaises(ZeroDivisionError):
- yield future
- self.finished = True
-
- @gen_test
- def test_async_raise(self):
- @gen.coroutine
- def f():
- yield gen.Task(self.io_loop.add_callback)
- 1 / 0
- future = f()
- with self.assertRaises(ZeroDivisionError):
- yield future
- self.finished = True
-
- @gen_test
- def test_pass_callback(self):
- @gen.coroutine
- def f():
- raise gen.Return(42)
- result = yield gen.Task(f)
- self.assertEqual(result, 42)
- self.finished = True
-
- @gen_test
- def test_replace_yieldpoint_exception(self):
- # Test exception handling: a coroutine can catch one exception
- # raised by a yield point and raise a different one.
- @gen.coroutine
- def f1():
- 1 / 0
-
- @gen.coroutine
- def f2():
- try:
- yield f1()
- except ZeroDivisionError:
- raise KeyError()
-
- future = f2()
- with self.assertRaises(KeyError):
- yield future
- self.finished = True
-
- @gen_test
- def test_swallow_yieldpoint_exception(self):
- # Test exception handling: a coroutine can catch an exception
- # raised by a yield point and not raise a different one.
- @gen.coroutine
- def f1():
- 1 / 0
-
- @gen.coroutine
- def f2():
- try:
- yield f1()
- except ZeroDivisionError:
- raise gen.Return(42)
-
- result = yield f2()
- self.assertEqual(result, 42)
- self.finished = True
-
- @gen_test
- def test_replace_context_exception(self):
- # Test exception handling: exceptions thrown into the stack context
- # can be caught and replaced.
- # Note that this test and the following are for behavior that is
- # not really supported any more: coroutines no longer create a
- # stack context automatically; but one is created after the first
- # YieldPoint (i.e. not a Future).
- @gen.coroutine
- def f2():
- (yield gen.Callback(1))()
- yield gen.Wait(1)
- self.io_loop.add_callback(lambda: 1 / 0)
- try:
- yield gen.Task(self.io_loop.add_timeout,
- self.io_loop.time() + 10)
- except ZeroDivisionError:
- raise KeyError()
-
- future = f2()
- with self.assertRaises(KeyError):
- yield future
- self.finished = True
-
- @gen_test
- def test_swallow_context_exception(self):
- # Test exception handling: exceptions thrown into the stack context
- # can be caught and ignored.
- @gen.coroutine
- def f2():
- (yield gen.Callback(1))()
- yield gen.Wait(1)
- self.io_loop.add_callback(lambda: 1 / 0)
- try:
- yield gen.Task(self.io_loop.add_timeout,
- self.io_loop.time() + 10)
- except ZeroDivisionError:
- raise gen.Return(42)
-
- result = yield f2()
- self.assertEqual(result, 42)
- self.finished = True
-
- @gen_test
- def test_moment(self):
- calls = []
- @gen.coroutine
- def f(name, yieldable):
- for i in range(5):
- calls.append(name)
- yield yieldable
- # First, confirm the behavior without moment: each coroutine
- # monopolizes the event loop until it finishes.
- immediate = Future()
- immediate.set_result(None)
- yield [f('a', immediate), f('b', immediate)]
- self.assertEqual(''.join(calls), 'aaaaabbbbb')
-
- # With moment, they take turns.
- calls = []
- yield [f('a', gen.moment), f('b', gen.moment)]
- self.assertEqual(''.join(calls), 'ababababab')
- self.finished = True
-
- calls = []
- yield [f('a', gen.moment), f('b', immediate)]
- self.assertEqual(''.join(calls), 'abbbbbaaaa')
-
-
-class GenSequenceHandler(RequestHandler):
- @asynchronous
- @gen.engine
- def get(self):
- self.io_loop = self.request.connection.stream.io_loop
- self.io_loop.add_callback((yield gen.Callback("k1")))
- yield gen.Wait("k1")
- self.write("1")
- self.io_loop.add_callback((yield gen.Callback("k2")))
- yield gen.Wait("k2")
- self.write("2")
- # reuse an old key
- self.io_loop.add_callback((yield gen.Callback("k1")))
- yield gen.Wait("k1")
- self.finish("3")
-
-
-class GenCoroutineSequenceHandler(RequestHandler):
- @gen.coroutine
- def get(self):
- self.io_loop = self.request.connection.stream.io_loop
- self.io_loop.add_callback((yield gen.Callback("k1")))
- yield gen.Wait("k1")
- self.write("1")
- self.io_loop.add_callback((yield gen.Callback("k2")))
- yield gen.Wait("k2")
- self.write("2")
- # reuse an old key
- self.io_loop.add_callback((yield gen.Callback("k1")))
- yield gen.Wait("k1")
- self.finish("3")
-
-
-class GenCoroutineUnfinishedSequenceHandler(RequestHandler):
- @asynchronous
- @gen.coroutine
- def get(self):
- self.io_loop = self.request.connection.stream.io_loop
- self.io_loop.add_callback((yield gen.Callback("k1")))
- yield gen.Wait("k1")
- self.write("1")
- self.io_loop.add_callback((yield gen.Callback("k2")))
- yield gen.Wait("k2")
- self.write("2")
- # reuse an old key
- self.io_loop.add_callback((yield gen.Callback("k1")))
- yield gen.Wait("k1")
- # just write, don't finish
- self.write("3")
-
-
-class GenTaskHandler(RequestHandler):
- @asynchronous
- @gen.engine
- def get(self):
- io_loop = self.request.connection.stream.io_loop
- client = AsyncHTTPClient(io_loop=io_loop)
- response = yield gen.Task(client.fetch, self.get_argument('url'))
- response.rethrow()
- self.finish(b"got response: " + response.body)
-
-
-class GenExceptionHandler(RequestHandler):
- @asynchronous
- @gen.engine
- def get(self):
- # This test depends on the order of the two decorators.
- io_loop = self.request.connection.stream.io_loop
- yield gen.Task(io_loop.add_callback)
- raise Exception("oops")
-
-
-class GenCoroutineExceptionHandler(RequestHandler):
- @gen.coroutine
- def get(self):
- # This test depends on the order of the two decorators.
- io_loop = self.request.connection.stream.io_loop
- yield gen.Task(io_loop.add_callback)
- raise Exception("oops")
-
-
-class GenYieldExceptionHandler(RequestHandler):
- @asynchronous
- @gen.engine
- def get(self):
- io_loop = self.request.connection.stream.io_loop
- # Test the interaction of the two stack_contexts.
-
- def fail_task(callback):
- io_loop.add_callback(lambda: 1 / 0)
- try:
- yield gen.Task(fail_task)
- raise Exception("did not get expected exception")
- except ZeroDivisionError:
- self.finish('ok')
-
-
-class UndecoratedCoroutinesHandler(RequestHandler):
- @gen.coroutine
- def prepare(self):
- self.chunks = []
- yield gen.Task(IOLoop.current().add_callback)
- self.chunks.append('1')
-
- @gen.coroutine
- def get(self):
- self.chunks.append('2')
- yield gen.Task(IOLoop.current().add_callback)
- self.chunks.append('3')
- yield gen.Task(IOLoop.current().add_callback)
- self.write(''.join(self.chunks))
-
-
-class AsyncPrepareErrorHandler(RequestHandler):
- @gen.coroutine
- def prepare(self):
- yield gen.Task(IOLoop.current().add_callback)
- raise HTTPError(403)
-
- def get(self):
- self.finish('ok')
-
-
-class GenWebTest(AsyncHTTPTestCase):
- def get_app(self):
- return Application([
- ('/sequence', GenSequenceHandler),
- ('/coroutine_sequence', GenCoroutineSequenceHandler),
- ('/coroutine_unfinished_sequence',
- GenCoroutineUnfinishedSequenceHandler),
- ('/task', GenTaskHandler),
- ('/exception', GenExceptionHandler),
- ('/coroutine_exception', GenCoroutineExceptionHandler),
- ('/yield_exception', GenYieldExceptionHandler),
- ('/undecorated_coroutine', UndecoratedCoroutinesHandler),
- ('/async_prepare_error', AsyncPrepareErrorHandler),
- ])
-
- def test_sequence_handler(self):
- response = self.fetch('/sequence')
- self.assertEqual(response.body, b"123")
-
- def test_coroutine_sequence_handler(self):
- response = self.fetch('/coroutine_sequence')
- self.assertEqual(response.body, b"123")
-
- def test_coroutine_unfinished_sequence_handler(self):
- response = self.fetch('/coroutine_unfinished_sequence')
- self.assertEqual(response.body, b"123")
-
- def test_task_handler(self):
- response = self.fetch('/task?url=%s' % url_escape(self.get_url('/sequence')))
- self.assertEqual(response.body, b"got response: 123")
-
- def test_exception_handler(self):
- # Make sure we get an error and not a timeout
- with ExpectLog(app_log, "Uncaught exception GET /exception"):
- response = self.fetch('/exception')
- self.assertEqual(500, response.code)
-
- def test_coroutine_exception_handler(self):
- # Make sure we get an error and not a timeout
- with ExpectLog(app_log, "Uncaught exception GET /coroutine_exception"):
- response = self.fetch('/coroutine_exception')
- self.assertEqual(500, response.code)
-
- def test_yield_exception_handler(self):
- response = self.fetch('/yield_exception')
- self.assertEqual(response.body, b'ok')
-
- def test_undecorated_coroutines(self):
- response = self.fetch('/undecorated_coroutine')
- self.assertEqual(response.body, b'123')
-
- def test_async_prepare_error_handler(self):
- response = self.fetch('/async_prepare_error')
- self.assertEqual(response.code, 403)
-
-
-class WithTimeoutTest(AsyncTestCase):
- @gen_test
- def test_timeout(self):
- with self.assertRaises(gen.TimeoutError):
- yield gen.with_timeout(datetime.timedelta(seconds=0.1),
- Future())
-
- @gen_test
- def test_completes_before_timeout(self):
- future = Future()
- self.io_loop.add_timeout(datetime.timedelta(seconds=0.1),
- lambda: future.set_result('asdf'))
- result = yield gen.with_timeout(datetime.timedelta(seconds=3600),
- future)
- self.assertEqual(result, 'asdf')
-
- @gen_test
- def test_fails_before_timeout(self):
- future = Future()
- self.io_loop.add_timeout(
- datetime.timedelta(seconds=0.1),
- lambda: future.set_exception(ZeroDivisionError))
- with self.assertRaises(ZeroDivisionError):
- yield gen.with_timeout(datetime.timedelta(seconds=3600), future)
-
- @gen_test
- def test_already_resolved(self):
- future = Future()
- future.set_result('asdf')
- result = yield gen.with_timeout(datetime.timedelta(seconds=3600),
- future)
- self.assertEqual(result, 'asdf')
-
- @unittest.skipIf(futures is None, 'futures module not present')
- @gen_test
- def test_timeout_concurrent_future(self):
- with futures.ThreadPoolExecutor(1) as executor:
- with self.assertRaises(gen.TimeoutError):
- yield gen.with_timeout(self.io_loop.time(),
- executor.submit(time.sleep, 0.1))
-
- @unittest.skipIf(futures is None, 'futures module not present')
- @gen_test
- def test_completed_concurrent_future(self):
- with futures.ThreadPoolExecutor(1) as executor:
- yield gen.with_timeout(datetime.timedelta(seconds=3600),
- executor.submit(lambda: None))
-
-
-if __name__ == '__main__':
- unittest.main()
diff --git a/rosbridge_server/src/tornado/test/gettext_translations/fr_FR/LC_MESSAGES/tornado_test.mo b/rosbridge_server/src/tornado/test/gettext_translations/fr_FR/LC_MESSAGES/tornado_test.mo
deleted file mode 100644
index 089f6c7ab..000000000
Binary files a/rosbridge_server/src/tornado/test/gettext_translations/fr_FR/LC_MESSAGES/tornado_test.mo and /dev/null differ
diff --git a/rosbridge_server/src/tornado/test/gettext_translations/fr_FR/LC_MESSAGES/tornado_test.po b/rosbridge_server/src/tornado/test/gettext_translations/fr_FR/LC_MESSAGES/tornado_test.po
deleted file mode 100644
index 732ee6da8..000000000
--- a/rosbridge_server/src/tornado/test/gettext_translations/fr_FR/LC_MESSAGES/tornado_test.po
+++ /dev/null
@@ -1,22 +0,0 @@
-# SOME DESCRIPTIVE TITLE.
-# Copyright (C) YEAR THE PACKAGE'S COPYRIGHT HOLDER
-# This file is distributed under the same license as the PACKAGE package.
-# FIRST AUTHOR , YEAR.
-#
-#, fuzzy
-msgid ""
-msgstr ""
-"Project-Id-Version: PACKAGE VERSION\n"
-"Report-Msgid-Bugs-To: \n"
-"POT-Creation-Date: 2012-06-14 01:10-0700\n"
-"PO-Revision-Date: YEAR-MO-DA HO:MI+ZONE\n"
-"Last-Translator: FULL NAME \n"
-"Language-Team: LANGUAGE \n"
-"Language: \n"
-"MIME-Version: 1.0\n"
-"Content-Type: text/plain; charset=utf-8\n"
-"Content-Transfer-Encoding: 8bit\n"
-
-#: extract_me.py:1
-msgid "school"
-msgstr "école"
diff --git a/rosbridge_server/src/tornado/test/httpclient_test.py b/rosbridge_server/src/tornado/test/httpclient_test.py
deleted file mode 100644
index 78daa74dc..000000000
--- a/rosbridge_server/src/tornado/test/httpclient_test.py
+++ /dev/null
@@ -1,517 +0,0 @@
-#!/usr/bin/env python
-
-from __future__ import absolute_import, division, print_function, with_statement
-
-import base64
-import binascii
-from contextlib import closing
-import functools
-import sys
-import threading
-
-from tornado.escape import utf8
-from tornado.httpclient import HTTPRequest, HTTPResponse, _RequestProxy, HTTPError, HTTPClient
-from tornado.httpserver import HTTPServer
-from tornado.ioloop import IOLoop
-from tornado.iostream import IOStream
-from tornado.log import gen_log
-from tornado import netutil
-from tornado.stack_context import ExceptionStackContext, NullContext
-from tornado.testing import AsyncHTTPTestCase, bind_unused_port, gen_test, ExpectLog
-from tornado.test.util import unittest, skipOnTravis
-from tornado.util import u, bytes_type
-from tornado.web import Application, RequestHandler, url
-
-try:
- from io import BytesIO # python 3
-except ImportError:
- from cStringIO import StringIO as BytesIO
-
-
-class HelloWorldHandler(RequestHandler):
- def get(self):
- name = self.get_argument("name", "world")
- self.set_header("Content-Type", "text/plain")
- self.finish("Hello %s!" % name)
-
-
-class PostHandler(RequestHandler):
- def post(self):
- self.finish("Post arg1: %s, arg2: %s" % (
- self.get_argument("arg1"), self.get_argument("arg2")))
-
-
-class ChunkHandler(RequestHandler):
- def get(self):
- self.write("asdf")
- self.flush()
- self.write("qwer")
-
-
-class AuthHandler(RequestHandler):
- def get(self):
- self.finish(self.request.headers["Authorization"])
-
-
-class CountdownHandler(RequestHandler):
- def get(self, count):
- count = int(count)
- if count > 0:
- self.redirect(self.reverse_url("countdown", count - 1))
- else:
- self.write("Zero")
-
-
-class EchoPostHandler(RequestHandler):
- def post(self):
- self.write(self.request.body)
-
-
-class UserAgentHandler(RequestHandler):
- def get(self):
- self.write(self.request.headers.get('User-Agent', 'User agent not set'))
-
-
-class ContentLength304Handler(RequestHandler):
- def get(self):
- self.set_status(304)
- self.set_header('Content-Length', 42)
-
- def _clear_headers_for_304(self):
- # Tornado strips content-length from 304 responses, but here we
- # want to simulate servers that include the headers anyway.
- pass
-
-
-class AllMethodsHandler(RequestHandler):
- SUPPORTED_METHODS = RequestHandler.SUPPORTED_METHODS + ('OTHER',)
-
- def method(self):
- self.write(self.request.method)
-
- get = post = put = delete = options = patch = other = method
-
-# These tests end up getting run redundantly: once here with the default
-# HTTPClient implementation, and then again in each implementation's own
-# test suite.
-
-
-class HTTPClientCommonTestCase(AsyncHTTPTestCase):
- def get_app(self):
- return Application([
- url("/hello", HelloWorldHandler),
- url("/post", PostHandler),
- url("/chunk", ChunkHandler),
- url("/auth", AuthHandler),
- url("/countdown/([0-9]+)", CountdownHandler, name="countdown"),
- url("/echopost", EchoPostHandler),
- url("/user_agent", UserAgentHandler),
- url("/304_with_content_length", ContentLength304Handler),
- url("/all_methods", AllMethodsHandler),
- ], gzip=True)
-
- @skipOnTravis
- def test_hello_world(self):
- response = self.fetch("/hello")
- self.assertEqual(response.code, 200)
- self.assertEqual(response.headers["Content-Type"], "text/plain")
- self.assertEqual(response.body, b"Hello world!")
- self.assertEqual(int(response.request_time), 0)
-
- response = self.fetch("/hello?name=Ben")
- self.assertEqual(response.body, b"Hello Ben!")
-
- def test_streaming_callback(self):
- # streaming_callback is also tested in test_chunked
- chunks = []
- response = self.fetch("/hello",
- streaming_callback=chunks.append)
- # with streaming_callback, data goes to the callback and not response.body
- self.assertEqual(chunks, [b"Hello world!"])
- self.assertFalse(response.body)
-
- def test_post(self):
- response = self.fetch("/post", method="POST",
- body="arg1=foo&arg2=bar")
- self.assertEqual(response.code, 200)
- self.assertEqual(response.body, b"Post arg1: foo, arg2: bar")
-
- def test_chunked(self):
- response = self.fetch("/chunk")
- self.assertEqual(response.body, b"asdfqwer")
-
- chunks = []
- response = self.fetch("/chunk",
- streaming_callback=chunks.append)
- self.assertEqual(chunks, [b"asdf", b"qwer"])
- self.assertFalse(response.body)
-
- def test_chunked_close(self):
- # test case in which chunks spread read-callback processing
- # over several ioloop iterations, but the connection is already closed.
- sock, port = bind_unused_port()
- with closing(sock):
- def write_response(stream, request_data):
- stream.write(b"""\
-HTTP/1.1 200 OK
-Transfer-Encoding: chunked
-
-1
-1
-1
-2
-0
-
-""".replace(b"\n", b"\r\n"), callback=stream.close)
-
- def accept_callback(conn, address):
- # fake an HTTP server using chunked encoding where the final chunks
- # and connection close all happen at once
- stream = IOStream(conn, io_loop=self.io_loop)
- stream.read_until(b"\r\n\r\n",
- functools.partial(write_response, stream))
- netutil.add_accept_handler(sock, accept_callback, self.io_loop)
- self.http_client.fetch("http://127.0.0.1:%d/" % port, self.stop)
- resp = self.wait()
- resp.rethrow()
- self.assertEqual(resp.body, b"12")
- self.io_loop.remove_handler(sock.fileno())
-
- def test_streaming_stack_context(self):
- chunks = []
- exc_info = []
-
- def error_handler(typ, value, tb):
- exc_info.append((typ, value, tb))
- return True
-
- def streaming_cb(chunk):
- chunks.append(chunk)
- if chunk == b'qwer':
- 1 / 0
-
- with ExceptionStackContext(error_handler):
- self.fetch('/chunk', streaming_callback=streaming_cb)
-
- self.assertEqual(chunks, [b'asdf', b'qwer'])
- self.assertEqual(1, len(exc_info))
- self.assertIs(exc_info[0][0], ZeroDivisionError)
-
- def test_basic_auth(self):
- self.assertEqual(self.fetch("/auth", auth_username="Aladdin",
- auth_password="open sesame").body,
- b"Basic QWxhZGRpbjpvcGVuIHNlc2FtZQ==")
-
- def test_basic_auth_explicit_mode(self):
- self.assertEqual(self.fetch("/auth", auth_username="Aladdin",
- auth_password="open sesame",
- auth_mode="basic").body,
- b"Basic QWxhZGRpbjpvcGVuIHNlc2FtZQ==")
-
- def test_unsupported_auth_mode(self):
- # curl and simple clients handle errors a bit differently; the
- # important thing is that they don't fall back to basic auth
- # on an unknown mode.
- with ExpectLog(gen_log, "uncaught exception", required=False):
- with self.assertRaises((ValueError, HTTPError)):
- response = self.fetch("/auth", auth_username="Aladdin",
- auth_password="open sesame",
- auth_mode="asdf")
- response.rethrow()
-
- def test_follow_redirect(self):
- response = self.fetch("/countdown/2", follow_redirects=False)
- self.assertEqual(302, response.code)
- self.assertTrue(response.headers["Location"].endswith("/countdown/1"))
-
- response = self.fetch("/countdown/2")
- self.assertEqual(200, response.code)
- self.assertTrue(response.effective_url.endswith("/countdown/0"))
- self.assertEqual(b"Zero", response.body)
-
- def test_credentials_in_url(self):
- url = self.get_url("/auth").replace("http://", "http://me:secret@")
- self.http_client.fetch(url, self.stop)
- response = self.wait()
- self.assertEqual(b"Basic " + base64.b64encode(b"me:secret"),
- response.body)
-
- def test_body_encoding(self):
- unicode_body = u("\xe9")
- byte_body = binascii.a2b_hex(b"e9")
-
- # unicode string in body gets converted to utf8
- response = self.fetch("/echopost", method="POST", body=unicode_body,
- headers={"Content-Type": "application/blah"})
- self.assertEqual(response.headers["Content-Length"], "2")
- self.assertEqual(response.body, utf8(unicode_body))
-
- # byte strings pass through directly
- response = self.fetch("/echopost", method="POST",
- body=byte_body,
- headers={"Content-Type": "application/blah"})
- self.assertEqual(response.headers["Content-Length"], "1")
- self.assertEqual(response.body, byte_body)
-
- # Mixing unicode in headers and byte string bodies shouldn't
- # break anything
- response = self.fetch("/echopost", method="POST", body=byte_body,
- headers={"Content-Type": "application/blah"},
- user_agent=u("foo"))
- self.assertEqual(response.headers["Content-Length"], "1")
- self.assertEqual(response.body, byte_body)
-
- def test_types(self):
- response = self.fetch("/hello")
- self.assertEqual(type(response.body), bytes_type)
- self.assertEqual(type(response.headers["Content-Type"]), str)
- self.assertEqual(type(response.code), int)
- self.assertEqual(type(response.effective_url), str)
-
- def test_header_callback(self):
- first_line = []
- headers = {}
- chunks = []
-
- def header_callback(header_line):
- if header_line.startswith('HTTP/'):
- first_line.append(header_line)
- elif header_line != '\r\n':
- k, v = header_line.split(':', 1)
- headers[k] = v.strip()
-
- def streaming_callback(chunk):
- # All header callbacks are run before any streaming callbacks,
- # so the header data is available to process the data as it
- # comes in.
- self.assertEqual(headers['Content-Type'], 'text/html; charset=UTF-8')
- chunks.append(chunk)
-
- self.fetch('/chunk', header_callback=header_callback,
- streaming_callback=streaming_callback)
- self.assertEqual(len(first_line), 1)
- self.assertRegexpMatches(first_line[0], 'HTTP/1.[01] 200 OK\r\n')
- self.assertEqual(chunks, [b'asdf', b'qwer'])
-
- def test_header_callback_stack_context(self):
- exc_info = []
-
- def error_handler(typ, value, tb):
- exc_info.append((typ, value, tb))
- return True
-
- def header_callback(header_line):
- if header_line.startswith('Content-Type:'):
- 1 / 0
-
- with ExceptionStackContext(error_handler):
- self.fetch('/chunk', header_callback=header_callback)
- self.assertEqual(len(exc_info), 1)
- self.assertIs(exc_info[0][0], ZeroDivisionError)
-
- def test_configure_defaults(self):
- defaults = dict(user_agent='TestDefaultUserAgent', allow_ipv6=False)
- # Construct a new instance of the configured client class
- client = self.http_client.__class__(self.io_loop, force_instance=True,
- defaults=defaults)
- client.fetch(self.get_url('/user_agent'), callback=self.stop)
- response = self.wait()
- self.assertEqual(response.body, b'TestDefaultUserAgent')
- client.close()
-
- def test_304_with_content_length(self):
- # According to the spec 304 responses SHOULD NOT include
- # Content-Length or other entity headers, but some servers do it
- # anyway.
- # http://www.w3.org/Protocols/rfc2616/rfc2616-sec10.html#sec10.3.5
- response = self.fetch('/304_with_content_length')
- self.assertEqual(response.code, 304)
- self.assertEqual(response.headers['Content-Length'], '42')
-
- def test_final_callback_stack_context(self):
- # The final callback should be run outside of the httpclient's
- # stack_context. We want to ensure that there is not stack_context
- # between the user's callback and the IOLoop, so monkey-patch
- # IOLoop.handle_callback_exception and disable the test harness's
- # context with a NullContext.
- # Note that this does not apply to secondary callbacks (header
- # and streaming_callback), as errors there must be seen as errors
- # by the http client so it can clean up the connection.
- exc_info = []
-
- def handle_callback_exception(callback):
- exc_info.append(sys.exc_info())
- self.stop()
- self.io_loop.handle_callback_exception = handle_callback_exception
- with NullContext():
- self.http_client.fetch(self.get_url('/hello'),
- lambda response: 1 / 0)
- self.wait()
- self.assertEqual(exc_info[0][0], ZeroDivisionError)
-
- @gen_test
- def test_future_interface(self):
- response = yield self.http_client.fetch(self.get_url('/hello'))
- self.assertEqual(response.body, b'Hello world!')
-
- @gen_test
- def test_future_http_error(self):
- with self.assertRaises(HTTPError) as context:
- yield self.http_client.fetch(self.get_url('/notfound'))
- self.assertEqual(context.exception.code, 404)
- self.assertEqual(context.exception.response.code, 404)
-
- @gen_test
- def test_reuse_request_from_response(self):
- # The response.request attribute should be an HTTPRequest, not
- # a _RequestProxy.
- # This test uses self.http_client.fetch because self.fetch calls
- # self.get_url on the input unconditionally.
- url = self.get_url('/hello')
- response = yield self.http_client.fetch(url)
- self.assertEqual(response.request.url, url)
- self.assertTrue(isinstance(response.request, HTTPRequest))
- response2 = yield self.http_client.fetch(response.request)
- self.assertEqual(response2.body, b'Hello world!')
-
- def test_all_methods(self):
- for method in ['GET', 'DELETE', 'OPTIONS']:
- response = self.fetch('/all_methods', method=method)
- self.assertEqual(response.body, utf8(method))
- for method in ['POST', 'PUT', 'PATCH']:
- response = self.fetch('/all_methods', method=method, body=b'')
- self.assertEqual(response.body, utf8(method))
- response = self.fetch('/all_methods', method='HEAD')
- self.assertEqual(response.body, b'')
- response = self.fetch('/all_methods', method='OTHER',
- allow_nonstandard_methods=True)
- self.assertEqual(response.body, b'OTHER')
-
- @gen_test
- def test_body(self):
- hello_url = self.get_url('/hello')
- with self.assertRaises(AssertionError) as context:
- yield self.http_client.fetch(hello_url, body='data')
-
- self.assertTrue('must be empty' in str(context.exception))
-
- with self.assertRaises(AssertionError) as context:
- yield self.http_client.fetch(hello_url, method='POST')
-
- self.assertTrue('must not be empty' in str(context.exception))
-
-
-class RequestProxyTest(unittest.TestCase):
- def test_request_set(self):
- proxy = _RequestProxy(HTTPRequest('http://example.com/',
- user_agent='foo'),
- dict())
- self.assertEqual(proxy.user_agent, 'foo')
-
- def test_default_set(self):
- proxy = _RequestProxy(HTTPRequest('http://example.com/'),
- dict(network_interface='foo'))
- self.assertEqual(proxy.network_interface, 'foo')
-
- def test_both_set(self):
- proxy = _RequestProxy(HTTPRequest('http://example.com/',
- proxy_host='foo'),
- dict(proxy_host='bar'))
- self.assertEqual(proxy.proxy_host, 'foo')
-
- def test_neither_set(self):
- proxy = _RequestProxy(HTTPRequest('http://example.com/'),
- dict())
- self.assertIs(proxy.auth_username, None)
-
- def test_bad_attribute(self):
- proxy = _RequestProxy(HTTPRequest('http://example.com/'),
- dict())
- with self.assertRaises(AttributeError):
- proxy.foo
-
- def test_defaults_none(self):
- proxy = _RequestProxy(HTTPRequest('http://example.com/'), None)
- self.assertIs(proxy.auth_username, None)
-
-
-class HTTPResponseTestCase(unittest.TestCase):
- def test_str(self):
- response = HTTPResponse(HTTPRequest('http://example.com'),
- 200, headers={}, buffer=BytesIO())
- s = str(response)
- self.assertTrue(s.startswith('HTTPResponse('))
- self.assertIn('code=200', s)
-
-
-class SyncHTTPClientTest(unittest.TestCase):
- def setUp(self):
- if IOLoop.configured_class().__name__ in ('TwistedIOLoop',
- 'AsyncIOMainLoop'):
- # TwistedIOLoop only supports the global reactor, so we can't have
- # separate IOLoops for client and server threads.
- # AsyncIOMainLoop doesn't work with the default policy
- # (although it could with some tweaks to this test and a
- # policy that created loops for non-main threads).
- raise unittest.SkipTest(
- 'Sync HTTPClient not compatible with TwistedIOLoop or '
- 'AsyncIOMainLoop')
- self.server_ioloop = IOLoop()
-
- sock, self.port = bind_unused_port()
- app = Application([('/', HelloWorldHandler)])
- self.server = HTTPServer(app, io_loop=self.server_ioloop)
- self.server.add_socket(sock)
-
- self.server_thread = threading.Thread(target=self.server_ioloop.start)
- self.server_thread.start()
-
- self.http_client = HTTPClient()
-
- def tearDown(self):
- def stop_server():
- self.server.stop()
- self.server_ioloop.stop()
- self.server_ioloop.add_callback(stop_server)
- self.server_thread.join()
- self.http_client.close()
- self.server_ioloop.close(all_fds=True)
-
- def get_url(self, path):
- return 'http://localhost:%d%s' % (self.port, path)
-
- def test_sync_client(self):
- response = self.http_client.fetch(self.get_url('/'))
- self.assertEqual(b'Hello world!', response.body)
-
- def test_sync_client_error(self):
- # Synchronous HTTPClient raises errors directly; no need for
- # response.rethrow()
- with self.assertRaises(HTTPError) as assertion:
- self.http_client.fetch(self.get_url('/notfound'))
- self.assertEqual(assertion.exception.code, 404)
-
-
-class HTTPRequestTestCase(unittest.TestCase):
- def test_headers(self):
- request = HTTPRequest('http://example.com', headers={'foo': 'bar'})
- self.assertEqual(request.headers, {'foo': 'bar'})
-
- def test_headers_setter(self):
- request = HTTPRequest('http://example.com')
- request.headers = {'bar': 'baz'}
- self.assertEqual(request.headers, {'bar': 'baz'})
-
- def test_null_headers_setter(self):
- request = HTTPRequest('http://example.com')
- request.headers = None
- self.assertEqual(request.headers, {})
-
- def test_body(self):
- request = HTTPRequest('http://example.com', body='foo')
- self.assertEqual(request.body, utf8('foo'))
-
- def test_body_setter(self):
- request = HTTPRequest('http://example.com')
- request.body = 'foo'
- self.assertEqual(request.body, utf8('foo'))
diff --git a/rosbridge_server/src/tornado/test/httpserver_test.py b/rosbridge_server/src/tornado/test/httpserver_test.py
deleted file mode 100644
index 63a55291b..000000000
--- a/rosbridge_server/src/tornado/test/httpserver_test.py
+++ /dev/null
@@ -1,1035 +0,0 @@
-#!/usr/bin/env python
-
-
-from __future__ import absolute_import, division, print_function, with_statement
-from tornado import netutil
-from tornado.escape import json_decode, json_encode, utf8, _unicode, recursive_unicode, native_str
-from tornado import gen
-from tornado.http1connection import HTTP1Connection
-from tornado.httpserver import HTTPServer
-from tornado.httputil import HTTPHeaders, HTTPMessageDelegate, HTTPServerConnectionDelegate, ResponseStartLine
-from tornado.iostream import IOStream
-from tornado.log import gen_log, app_log
-from tornado.netutil import ssl_options_to_context
-from tornado.simple_httpclient import SimpleAsyncHTTPClient
-from tornado.testing import AsyncHTTPTestCase, AsyncHTTPSTestCase, AsyncTestCase, ExpectLog, gen_test
-from tornado.test.util import unittest, skipOnTravis
-from tornado.util import u, bytes_type
-from tornado.web import Application, RequestHandler, asynchronous, stream_request_body
-from contextlib import closing
-import datetime
-import gzip
-import os
-import shutil
-import socket
-import ssl
-import sys
-import tempfile
-
-try:
- from io import BytesIO # python 3
-except ImportError:
- from cStringIO import StringIO as BytesIO # python 2
-
-
-def read_stream_body(stream, callback):
- """Reads an HTTP response from `stream` and runs callback with its
- headers and body."""
- chunks = []
- class Delegate(HTTPMessageDelegate):
- def headers_received(self, start_line, headers):
- self.headers = headers
-
- def data_received(self, chunk):
- chunks.append(chunk)
-
- def finish(self):
- callback((self.headers, b''.join(chunks)))
- conn = HTTP1Connection(stream, True)
- conn.read_response(Delegate())
-
-
-class HandlerBaseTestCase(AsyncHTTPTestCase):
- def get_app(self):
- return Application([('/', self.__class__.Handler)])
-
- def fetch_json(self, *args, **kwargs):
- response = self.fetch(*args, **kwargs)
- response.rethrow()
- return json_decode(response.body)
-
-
-class HelloWorldRequestHandler(RequestHandler):
- def initialize(self, protocol="http"):
- self.expected_protocol = protocol
-
- def get(self):
- if self.request.protocol != self.expected_protocol:
- raise Exception("unexpected protocol")
- self.finish("Hello world")
-
- def post(self):
- self.finish("Got %d bytes in POST" % len(self.request.body))
-
-
-# In pre-1.0 versions of openssl, SSLv23 clients always send SSLv2
-# ClientHello messages, which are rejected by SSLv3 and TLSv1
-# servers. Note that while the OPENSSL_VERSION_INFO was formally
-# introduced in python3.2, it was present but undocumented in
-# python 2.7
-skipIfOldSSL = unittest.skipIf(
- getattr(ssl, 'OPENSSL_VERSION_INFO', (0, 0)) < (1, 0),
- "old version of ssl module and/or openssl")
-
-
-class BaseSSLTest(AsyncHTTPSTestCase):
- def get_app(self):
- return Application([('/', HelloWorldRequestHandler,
- dict(protocol="https"))])
-
-
-class SSLTestMixin(object):
- def get_ssl_options(self):
- return dict(ssl_version=self.get_ssl_version(),
- **AsyncHTTPSTestCase.get_ssl_options())
-
- def get_ssl_version(self):
- raise NotImplementedError()
-
- def test_ssl(self):
- response = self.fetch('/')
- self.assertEqual(response.body, b"Hello world")
-
- def test_large_post(self):
- response = self.fetch('/',
- method='POST',
- body='A' * 5000)
- self.assertEqual(response.body, b"Got 5000 bytes in POST")
-
- def test_non_ssl_request(self):
- # Make sure the server closes the connection when it gets a non-ssl
- # connection, rather than waiting for a timeout or otherwise
- # misbehaving.
- with ExpectLog(gen_log, '(SSL Error|uncaught exception)'):
- with ExpectLog(gen_log, 'Uncaught exception', required=False):
- self.http_client.fetch(
- self.get_url("/").replace('https:', 'http:'),
- self.stop,
- request_timeout=3600,
- connect_timeout=3600)
- response = self.wait()
- self.assertEqual(response.code, 599)
-
-# Python's SSL implementation differs significantly between versions.
-# For example, SSLv3 and TLSv1 throw an exception if you try to read
-# from the socket before the handshake is complete, but the default
-# of SSLv23 allows it.
-
-
-class SSLv23Test(BaseSSLTest, SSLTestMixin):
- def get_ssl_version(self):
- return ssl.PROTOCOL_SSLv23
-
-
-@skipIfOldSSL
-class SSLv3Test(BaseSSLTest, SSLTestMixin):
- def get_ssl_version(self):
- return ssl.PROTOCOL_SSLv3
-
-
-@skipIfOldSSL
-class TLSv1Test(BaseSSLTest, SSLTestMixin):
- def get_ssl_version(self):
- return ssl.PROTOCOL_TLSv1
-
-
-@unittest.skipIf(not hasattr(ssl, 'SSLContext'), 'ssl.SSLContext not present')
-class SSLContextTest(BaseSSLTest, SSLTestMixin):
- def get_ssl_options(self):
- context = ssl_options_to_context(
- AsyncHTTPSTestCase.get_ssl_options(self))
- assert isinstance(context, ssl.SSLContext)
- return context
-
-
-class BadSSLOptionsTest(unittest.TestCase):
- def test_missing_arguments(self):
- application = Application()
- self.assertRaises(KeyError, HTTPServer, application, ssl_options={
- "keyfile": "/__missing__.crt",
- })
-
- def test_missing_key(self):
- """A missing SSL key should cause an immediate exception."""
-
- application = Application()
- module_dir = os.path.dirname(__file__)
- existing_certificate = os.path.join(module_dir, 'test.crt')
-
- self.assertRaises(ValueError, HTTPServer, application, ssl_options={
- "certfile": "/__mising__.crt",
- })
- self.assertRaises(ValueError, HTTPServer, application, ssl_options={
- "certfile": existing_certificate,
- "keyfile": "/__missing__.key"
- })
-
- # This actually works because both files exist
- HTTPServer(application, ssl_options={
- "certfile": existing_certificate,
- "keyfile": existing_certificate
- })
-
-
-class MultipartTestHandler(RequestHandler):
- def post(self):
- self.finish({"header": self.request.headers["X-Header-Encoding-Test"],
- "argument": self.get_argument("argument"),
- "filename": self.request.files["files"][0].filename,
- "filebody": _unicode(self.request.files["files"][0]["body"]),
- })
-
-
-# This test is also called from wsgi_test
-class HTTPConnectionTest(AsyncHTTPTestCase):
- def get_handlers(self):
- return [("/multipart", MultipartTestHandler),
- ("/hello", HelloWorldRequestHandler)]
-
- def get_app(self):
- return Application(self.get_handlers())
-
- def raw_fetch(self, headers, body):
- with closing(IOStream(socket.socket())) as stream:
- stream.connect(('127.0.0.1', self.get_http_port()), self.stop)
- self.wait()
- stream.write(
- b"\r\n".join(headers +
- [utf8("Content-Length: %d\r\n" % len(body))]) +
- b"\r\n" + body)
- read_stream_body(stream, self.stop)
- headers, body = self.wait()
- return body
-
- def test_multipart_form(self):
- # Encodings here are tricky: Headers are latin1, bodies can be
- # anything (we use utf8 by default).
- response = self.raw_fetch([
- b"POST /multipart HTTP/1.0",
- b"Content-Type: multipart/form-data; boundary=1234567890",
- b"X-Header-encoding-test: \xe9",
- ],
- b"\r\n".join([
- b"Content-Disposition: form-data; name=argument",
- b"",
- u("\u00e1").encode("utf-8"),
- b"--1234567890",
- u('Content-Disposition: form-data; name="files"; filename="\u00f3"').encode("utf8"),
- b"",
- u("\u00fa").encode("utf-8"),
- b"--1234567890--",
- b"",
- ]))
- data = json_decode(response)
- self.assertEqual(u("\u00e9"), data["header"])
- self.assertEqual(u("\u00e1"), data["argument"])
- self.assertEqual(u("\u00f3"), data["filename"])
- self.assertEqual(u("\u00fa"), data["filebody"])
-
- def test_100_continue(self):
- # Run through a 100-continue interaction by hand:
- # When given Expect: 100-continue, we get a 100 response after the
- # headers, and then the real response after the body.
- stream = IOStream(socket.socket(), io_loop=self.io_loop)
- stream.connect(("localhost", self.get_http_port()), callback=self.stop)
- self.wait()
- stream.write(b"\r\n".join([b"POST /hello HTTP/1.1",
- b"Content-Length: 1024",
- b"Expect: 100-continue",
- b"Connection: close",
- b"\r\n"]), callback=self.stop)
- self.wait()
- stream.read_until(b"\r\n\r\n", self.stop)
- data = self.wait()
- self.assertTrue(data.startswith(b"HTTP/1.1 100 "), data)
- stream.write(b"a" * 1024)
- stream.read_until(b"\r\n", self.stop)
- first_line = self.wait()
- self.assertTrue(first_line.startswith(b"HTTP/1.1 200"), first_line)
- stream.read_until(b"\r\n\r\n", self.stop)
- header_data = self.wait()
- headers = HTTPHeaders.parse(native_str(header_data.decode('latin1')))
- stream.read_bytes(int(headers["Content-Length"]), self.stop)
- body = self.wait()
- self.assertEqual(body, b"Got 1024 bytes in POST")
- stream.close()
-
-
-class EchoHandler(RequestHandler):
- def get(self):
- self.write(recursive_unicode(self.request.arguments))
-
- def post(self):
- self.write(recursive_unicode(self.request.arguments))
-
-
-class TypeCheckHandler(RequestHandler):
- def prepare(self):
- self.errors = {}
- fields = [
- ('method', str),
- ('uri', str),
- ('version', str),
- ('remote_ip', str),
- ('protocol', str),
- ('host', str),
- ('path', str),
- ('query', str),
- ]
- for field, expected_type in fields:
- self.check_type(field, getattr(self.request, field), expected_type)
-
- self.check_type('header_key', list(self.request.headers.keys())[0], str)
- self.check_type('header_value', list(self.request.headers.values())[0], str)
-
- self.check_type('cookie_key', list(self.request.cookies.keys())[0], str)
- self.check_type('cookie_value', list(self.request.cookies.values())[0].value, str)
- # secure cookies
-
- self.check_type('arg_key', list(self.request.arguments.keys())[0], str)
- self.check_type('arg_value', list(self.request.arguments.values())[0][0], bytes_type)
-
- def post(self):
- self.check_type('body', self.request.body, bytes_type)
- self.write(self.errors)
-
- def get(self):
- self.write(self.errors)
-
- def check_type(self, name, obj, expected_type):
- actual_type = type(obj)
- if expected_type != actual_type:
- self.errors[name] = "expected %s, got %s" % (expected_type,
- actual_type)
-
-
-class HTTPServerTest(AsyncHTTPTestCase):
- def get_app(self):
- return Application([("/echo", EchoHandler),
- ("/typecheck", TypeCheckHandler),
- ("//doubleslash", EchoHandler),
- ])
-
- def test_query_string_encoding(self):
- response = self.fetch("/echo?foo=%C3%A9")
- data = json_decode(response.body)
- self.assertEqual(data, {u("foo"): [u("\u00e9")]})
-
- def test_empty_query_string(self):
- response = self.fetch("/echo?foo=&foo=")
- data = json_decode(response.body)
- self.assertEqual(data, {u("foo"): [u(""), u("")]})
-
- def test_empty_post_parameters(self):
- response = self.fetch("/echo", method="POST", body="foo=&bar=")
- data = json_decode(response.body)
- self.assertEqual(data, {u("foo"): [u("")], u("bar"): [u("")]})
-
- def test_types(self):
- headers = {"Cookie": "foo=bar"}
- response = self.fetch("/typecheck?foo=bar", headers=headers)
- data = json_decode(response.body)
- self.assertEqual(data, {})
-
- response = self.fetch("/typecheck", method="POST", body="foo=bar", headers=headers)
- data = json_decode(response.body)
- self.assertEqual(data, {})
-
- def test_double_slash(self):
- # urlparse.urlsplit (which tornado.httpserver used to use
- # incorrectly) would parse paths beginning with "//" as
- # protocol-relative urls.
- response = self.fetch("//doubleslash")
- self.assertEqual(200, response.code)
- self.assertEqual(json_decode(response.body), {})
-
- def test_malformed_body(self):
- # parse_qs is pretty forgiving, but it will fail on python 3
- # if the data is not utf8. On python 2 parse_qs will work,
- # but then the recursive_unicode call in EchoHandler will
- # fail.
- if str is bytes_type:
- return
- with ExpectLog(gen_log, 'Invalid x-www-form-urlencoded body'):
- response = self.fetch(
- '/echo', method="POST",
- headers={'Content-Type': 'application/x-www-form-urlencoded'},
- body=b'\xe9')
- self.assertEqual(200, response.code)
- self.assertEqual(b'{}', response.body)
-
-
-class HTTPServerRawTest(AsyncHTTPTestCase):
- def get_app(self):
- return Application([
- ('/echo', EchoHandler),
- ])
-
- def setUp(self):
- super(HTTPServerRawTest, self).setUp()
- self.stream = IOStream(socket.socket())
- self.stream.connect(('localhost', self.get_http_port()), self.stop)
- self.wait()
-
- def tearDown(self):
- self.stream.close()
- super(HTTPServerRawTest, self).tearDown()
-
- def test_empty_request(self):
- self.stream.close()
- self.io_loop.add_timeout(datetime.timedelta(seconds=0.001), self.stop)
- self.wait()
-
- def test_malformed_first_line(self):
- with ExpectLog(gen_log, '.*Malformed HTTP request line'):
- self.stream.write(b'asdf\r\n\r\n')
- # TODO: need an async version of ExpectLog so we don't need
- # hard-coded timeouts here.
- self.io_loop.add_timeout(datetime.timedelta(seconds=0.01),
- self.stop)
- self.wait()
-
- def test_malformed_headers(self):
- with ExpectLog(gen_log, '.*Malformed HTTP headers'):
- self.stream.write(b'GET / HTTP/1.0\r\nasdf\r\n\r\n')
- self.io_loop.add_timeout(datetime.timedelta(seconds=0.01),
- self.stop)
- self.wait()
-
- def test_chunked_request_body(self):
- # Chunked requests are not widely supported and we don't have a way
- # to generate them in AsyncHTTPClient, but HTTPServer will read them.
- self.stream.write(b"""\
-POST /echo HTTP/1.1
-Transfer-Encoding: chunked
-Content-Type: application/x-www-form-urlencoded
-
-4
-foo=
-3
-bar
-0
-
-""".replace(b"\n", b"\r\n"))
- read_stream_body(self.stream, self.stop)
- headers, response = self.wait()
- self.assertEqual(json_decode(response), {u('foo'): [u('bar')]})
-
-
-class XHeaderTest(HandlerBaseTestCase):
- class Handler(RequestHandler):
- def get(self):
- self.write(dict(remote_ip=self.request.remote_ip,
- remote_protocol=self.request.protocol))
-
- def get_httpserver_options(self):
- return dict(xheaders=True)
-
- def test_ip_headers(self):
- self.assertEqual(self.fetch_json("/")["remote_ip"], "127.0.0.1")
-
- valid_ipv4 = {"X-Real-IP": "4.4.4.4"}
- self.assertEqual(
- self.fetch_json("/", headers=valid_ipv4)["remote_ip"],
- "4.4.4.4")
-
- valid_ipv4_list = {"X-Forwarded-For": "127.0.0.1, 4.4.4.4"}
- self.assertEqual(
- self.fetch_json("/", headers=valid_ipv4_list)["remote_ip"],
- "4.4.4.4")
-
- valid_ipv6 = {"X-Real-IP": "2620:0:1cfe:face:b00c::3"}
- self.assertEqual(
- self.fetch_json("/", headers=valid_ipv6)["remote_ip"],
- "2620:0:1cfe:face:b00c::3")
-
- valid_ipv6_list = {"X-Forwarded-For": "::1, 2620:0:1cfe:face:b00c::3"}
- self.assertEqual(
- self.fetch_json("/", headers=valid_ipv6_list)["remote_ip"],
- "2620:0:1cfe:face:b00c::3")
-
- invalid_chars = {"X-Real-IP": "4.4.4.4
-
-
-