self.exc_info = sys.exc_info()
else:
self.exc_info = (BadYieldError(
- "yielded unknown object %r" % (yielded,)),)
+ "yielded unknown object %r" % (yielded,)),)
finally:
self.running = False
self.body = utf8(body)
self.auth_username = auth_username
self.auth_password = auth_password
- self.auth_mode = auth_mode
+ self.auth_mode = auth_mode
self.connect_timeout = connect_timeout
self.request_timeout = request_timeout
self.follow_redirects = follow_redirects
if proto in ("http", "https"):
self.protocol = proto
-
self.host = host or self.headers.get("Host") or "127.0.0.1"
self.files = files or {}
self.connection = connection
except ImportError:
from urllib.parse import urlencode # py3
+
class _NormalizedHeaderCache(dict):
"""Dynamic cached mapping of header names to Http-Header-Case.
_normalized_headers = _NormalizedHeaderCache(1000)
+
class HTTPHeaders(dict):
"""A dictionary that maintains ``Http-Header-Case`` for all keys.
"""
pass
+
def _parse_request_range(range_header):
"""Parses a Range header.
end += 1
return (start, end)
+
def _get_content_range(start, end, total):
"""Returns a suitable Content-Range header:
end = (end or total) - 1
return "%s-%s/%s" % (start, end, total)
+
def _int_or_none(val):
val = val.strip()
if val == "":
self.error = exc_info[1]
if self._read_until_close:
if (self._streaming_callback is not None and
- self._read_buffer_size):
+ self._read_buffer_size):
self._run_callback(self._streaming_callback,
self._consume(self._read_buffer_size))
callback = self._read_callback
def set_nodelay(self, value):
if (self.socket is not None and
- self.socket.family in (socket.AF_INET, socket.AF_INET6)):
+ self.socket.family in (socket.AF_INET, socket.AF_INET6)):
try:
self.socket.setsockopt(socket.IPPROTO_TCP,
socket.TCP_NODELAY, 1 if value else 0)
'Digest realm="%s", nonce="%s", opaque="%s"' %
(realm, nonce, opaque))
+
@unittest.skipIf(pycurl is None, "pycurl module not present")
class CurlHTTPClientTestCase(AsyncHTTPTestCase):
def setUp(self):
skipBefore33 = unittest.skipIf(sys.version_info < (3, 3), 'PEP 380 not available')
skipNotCPython = unittest.skipIf(platform.python_implementation() != 'CPython',
- 'Not CPython implementation')
+ 'Not CPython implementation')
class GenEngineTest(AsyncTestCase):
# without waiting for garbage collection.
@gen.engine
def f():
- class Foo(object): pass
+ class Foo(object):
+ pass
arg = Foo()
self.arg_ref = weakref.ref(arg)
task = gen.Task(self.io_loop.add_callback, arg=arg)
# want to simulate servers that include the headers anyway.
pass
+
class AllMethodsHandler(RequestHandler):
SUPPORTED_METHODS = RequestHandler.SUPPORTED_METHODS + ('OTHER',)
def post(self):
self.write(recursive_unicode(self.request.arguments))
+
class TypeCheckHandler(RequestHandler):
def prepare(self):
self.errors = {}
class HTTPServerRawTest(AsyncHTTPTestCase):
def get_app(self):
return Application([
- ('/echo', EchoHandler),
- ])
+ ('/echo', EchoHandler),
+ ])
def setUp(self):
super(HTTPServerRawTest, self).setUp()
class Handler(RequestHandler):
def get(self):
self.write(dict(remote_ip=self.request.remote_ip,
- remote_protocol=self.request.protocol))
+ remote_protocol=self.request.protocol))
def get_httpserver_options(self):
return dict(xheaders=True)
self.fail()
server.read_until_close(callback=closed_callback,
streaming_callback=self.stop)
- #self.io_loop.add_timeout(self.io_loop.time() + 0.01, self.stop)
+ # self.io_loop.add_timeout(self.io_loop.time() + 0.01, self.stop)
data = self.wait()
self.assertEqual(data, b"efgh")
server.close()
help="A comma-separated list of gc module debug constants, "
"e.g. DEBUG_STATS or DEBUG_COLLECTABLE,DEBUG_OBJECTS",
callback=lambda values: gc.set_debug(
- reduce(operator.or_, (getattr(gc, v) for v in values))))
+ reduce(operator.or_, (getattr(gc, v) for v in values))))
define('locale', type=str, default=None,
callback=lambda x: locale.setlocale(locale.LC_ALL, x))
deactivate_callbacks[2]()
self.assertEqual(func(), ['c0', 'c1'])
-
def test_isolation_nonempty(self):
# f2 and f3 are a chain of operations started in context c1.
# f2 is incidentally run under context c2, but that context should
relpath = lambda *a: os.path.join(os.path.dirname(__file__), *a)
+
def wsgi_safe(cls):
wsgi_safe_tests.append(cls)
return cls
data = [('foo=a=b', 'a=b'),
('foo="a=b"', 'a=b'),
('foo="a;b"', 'a;b'),
- #('foo=a\\073b', 'a;b'), # even encoded, ";" is a delimiter
+ # ('foo=a\\073b', 'a;b'), # even encoded, ";" is a delimiter
('foo="a\\073b"', 'a;b'),
('foo="a\\"b"', 'a"b'),
]
# negative values, and at least one client (processing.js) seems
# to use if-modified-since 1/1/1960 as a cache-busting technique.
response = self.fetch("/static/robots.txt", headers={
- 'If-Modified-Since': 'Fri, 01 Jan 1960 00:00:00 GMT'})
+ 'If-Modified-Since': 'Fri, 01 Jan 1960 00:00:00 GMT'})
self.assertEqual(response.code, 200)
def test_static_if_modified_since_time_zone(self):
stat = os.stat(relpath('static/robots.txt'))
response = self.fetch('/static/robots.txt', headers={
- 'If-Modified-Since': format_timestamp(stat.st_mtime - 1)})
+ 'If-Modified-Since': format_timestamp(stat.st_mtime - 1)})
self.assertEqual(response.code, 200)
response = self.fetch('/static/robots.txt', headers={
- 'If-Modified-Since': format_timestamp(stat.st_mtime + 1)})
+ 'If-Modified-Since': format_timestamp(stat.st_mtime + 1)})
self.assertEqual(response.code, 304)
def test_static_etag(self):
def test_static_with_range(self):
response = self.fetch('/static/robots.txt', headers={
- 'Range': 'bytes=0-9'})
+ 'Range': 'bytes=0-9'})
self.assertEqual(response.code, 206)
self.assertEqual(response.body, b"User-agent")
self.assertEqual(utf8(response.headers.get("Etag")),
def test_static_with_range_full_file(self):
response = self.fetch('/static/robots.txt', headers={
- 'Range': 'bytes=0-'})
+ 'Range': 'bytes=0-'})
# Note: Chrome refuses to play audio if it gets an HTTP 206 in response
# to ``Range: bytes=0-`` :(
self.assertEqual(response.code, 200)
def test_static_with_range_end_edge(self):
response = self.fetch('/static/robots.txt', headers={
- 'Range': 'bytes=22-'})
+ 'Range': 'bytes=22-'})
self.assertEqual(response.body, b": /\n")
self.assertEqual(response.headers.get("Content-Length"), "4")
self.assertEqual(response.headers.get("Content-Range"),
def test_static_with_range_neg_end(self):
response = self.fetch('/static/robots.txt', headers={
- 'Range': 'bytes=-4'})
+ 'Range': 'bytes=-4'})
self.assertEqual(response.body, b": /\n")
self.assertEqual(response.headers.get("Content-Length"), "4")
self.assertEqual(response.headers.get("Content-Range"),
def test_static_invalid_range(self):
response = self.fetch('/static/robots.txt', headers={
- 'Range': 'asdf'})
+ 'Range': 'asdf'})
self.assertEqual(response.code, 416)
def test_static_head(self):
x, self.handler.value())
loader = DictLoader({
- 'foo.html': '{{ my_ui_method(42) }} {% module MyModule(123) %}',
- })
+ 'foo.html': '{{ my_ui_method(42) }} {% module MyModule(123) %}',
+ })
return dict(template_loader=loader,
ui_methods={'my_ui_method': my_ui_method},
ui_modules={'MyModule': MyModule})
response = self.fetch('/', method=method, body=b'')
self.assertEqual(response.code, 405)
+
class UnimplementedNonStandardMethodsTest(SimpleHandlerTestCase):
# wsgiref.validate complains about unknown methods in a way that makes
# this test not wsgi_safe.
allow_nonstandard_methods=True)
self.assertEqual(response.code, 405)
+
@wsgi_safe
class AllHTTPMethodsTest(SimpleHandlerTestCase):
class Handler(RequestHandler):
response = self.fetch('/', method=method, body=b'')
self.assertEqual(response.body, utf8(method))
+
class PatchMethodTest(SimpleHandlerTestCase):
class Handler(RequestHandler):
SUPPORTED_METHODS = RequestHandler.SUPPORTED_METHODS + ('OTHER',)
def on_message(self, message):
self.write_message(message, isinstance(message, bytes))
+
class NonWebSocketHandler(RequestHandler):
def get(self):
self.write('ok')
self.set_default_headers()
if (not self.request.supports_http_1_1() and
getattr(self.request, 'connection', None) and
- not self.request.connection.no_keep_alive):
+ not self.request.connection.no_keep_alive):
conn_header = self.request.headers.get("Connection")
if conn_header and (conn_header.lower() == "keep-alive"):
self.set_header("Connection", "Keep-Alive")
# additional headers or split the request. Also cap length to
# prevent obviously erroneous values.
if (len(value) > 4000 or
- RequestHandler._INVALID_HEADER_CHAR_RE.search(value)):
+ RequestHandler._INVALID_HEADER_CHAR_RE.search(value)):
raise ValueError("Unsafe header value %r", value)
return value
version_hash = self._get_cached_version(self.absolute_path)
if not version_hash:
return None
- return '"%s"' %(version_hash, )
+ return '"%s"' % (version_hash, )
def set_headers(self):
"""Sets the content and caching headers on the response."""
raise HTTPError(403, "%s is not in root static directory",
self.path)
if (os.path.isdir(absolute_path) and
- self.default_filename is not None):
+ self.default_filename is not None):
# need to look at the request.path here for when path is empty
# but there is some prefix to the path that was already
# trimmed by the routing
self.connect_future.set_exception(response.error)
else:
self.connect_future.set_exception(WebSocketError(
- "Non-websocket response"))
+ "Non-websocket response"))
def _handle_1xx(self, code):
assert code == 101
"REQUEST_METHOD": request.method,
"SCRIPT_NAME": "",
"PATH_INFO": to_wsgi_str(escape.url_unescape(
- request.path, encoding=None, plus=False)),
+ request.path, encoding=None, plus=False)),
"QUERY_STRING": request.query,
"REMOTE_ADDR": request.remote_ip,
"SERVER_NAME": host,
projects / thirdparty / tornado.git / commitdiff
