to write the response.
"""
assert isinstance(chunk, bytes)
+ assert self.version.startswith("HTTP/1."), \
+ "deprecated interface ony supported in HTTP/1.x"
self.connection.write(chunk, callback=callback)
def finish(self):
sock, port = bind_unused_port()
with closing(sock):
def write_response(stream, request_data):
+ if b"HTTP/1." not in request_data:
+ self.skipTest("requires HTTP/1.x")
stream.write(b"""\
HTTP/1.1 200 OK
Transfer-Encoding: chunked
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.assertRegexpMatches(first_line[0], 'HTTP/[0-9]\\.[0-9] 200 OK\r\n')
self.assertEqual(chunks, [b'asdf', b'qwer'])
def test_header_callback_stack_context(self):
# delegate interface, and writes its response via request.write
# instead of request.connection.write_headers.
def handle_request(request):
+ self.http1 = request.version.startswith("HTTP/1.")
message = b"Hello world"
request.write(utf8("HTTP/1.1 200 OK\r\n"
"Content-Length: %d\r\n\r\n" % len(message)))
def test_legacy_interface(self):
response = self.fetch('/')
+ if not self.http1:
+ self.skipTest("requires HTTP/1.x")
self.assertEqual(response.body, b"Hello world")
class HTTP100ContinueTestCase(AsyncHTTPTestCase):
def respond_100(self, request):
+ self.http1 = request.version.startswith('HTTP/1.')
self.request = request
self.request.connection.stream.write(
b"HTTP/1.1 100 CONTINUE\r\n\r\n",
def test_100_continue(self):
res = self.fetch('/')
+ if not self.http1:
+ self.skipTest("requires HTTP/1.x")
self.assertEqual(res.body, b'A')
class HTTP204NoContentTestCase(AsyncHTTPTestCase):
def respond_204(self, request):
+ self.http1 = request.version.startswith('HTTP/1.')
# A 204 response never has a body, even if doesn't have a content-length
# (which would otherwise mean read-until-close). Tornado always
# sends a content-length, so we simulate here a server that sends
def test_204_no_content(self):
resp = self.fetch('/')
+ if not self.http1:
+ self.skipTest("requires HTTP/1.x")
self.assertEqual(resp.code, 204)
self.assertEqual(resp.body, b'')