httpserver_test.py 文件源码

def test_timeout(self):
        stream = IOStream(socket.socket())
        try:
            yield stream.connect(('127.0.0.1', self.get_http_port()))
            # Use a raw stream because AsyncHTTPClient won't let us read a
            # response without finishing a body.
            stream.write(b'PUT /streaming?body_timeout=0.1 HTTP/1.0\r\n'
                         b'Content-Length: 42\r\n\r\n')
            with ExpectLog(gen_log, 'Timeout reading body'):
                response = yield stream.read_until_close()
            self.assertEqual(response, b'')
        finally:
            stream.close()