| 12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061 |
- import socket
- import typing # noqa(F401)
- from tornado.http1connection import HTTP1Connection
- from tornado.httputil import HTTPMessageDelegate
- from tornado.iostream import IOStream
- from tornado.locks import Event
- from tornado.netutil import add_accept_handler
- from tornado.testing import AsyncTestCase, bind_unused_port, gen_test
- class HTTP1ConnectionTest(AsyncTestCase):
- code = None # type: typing.Optional[int]
- def setUp(self):
- super().setUp()
- self.asyncSetUp()
- @gen_test
- def asyncSetUp(self):
- listener, port = bind_unused_port()
- event = Event()
- def accept_callback(conn, addr):
- self.server_stream = IOStream(conn)
- self.addCleanup(self.server_stream.close)
- event.set()
- add_accept_handler(listener, accept_callback)
- self.client_stream = IOStream(socket.socket())
- self.addCleanup(self.client_stream.close)
- yield [self.client_stream.connect(("127.0.0.1", port)), event.wait()]
- self.io_loop.remove_handler(listener)
- listener.close()
- @gen_test
- def test_http10_no_content_length(self):
- # Regression test for a bug in which can_keep_alive would crash
- # for an HTTP/1.0 (not 1.1) response with no content-length.
- conn = HTTP1Connection(self.client_stream, True)
- self.server_stream.write(b"HTTP/1.0 200 Not Modified\r\n\r\nhello")
- self.server_stream.close()
- event = Event()
- test = self
- body = []
- class Delegate(HTTPMessageDelegate):
- def headers_received(self, start_line, headers):
- test.code = start_line.code
- def data_received(self, data):
- body.append(data)
- def finish(self):
- event.set()
- yield conn.read_response(Delegate())
- yield event.wait()
- self.assertEqual(self.code, 200)
- self.assertEqual(b"".join(body), b"hello")
|
