httpclient_test.py
Go to the documentation of this file.
00001 #!/usr/bin/env python
00002 
00003 from __future__ import absolute_import, division, with_statement
00004 
00005 import base64
00006 import binascii
00007 from contextlib import closing
00008 import functools
00009 
00010 from tornado.escape import utf8
00011 from tornado.httpclient import AsyncHTTPClient
00012 from tornado.iostream import IOStream
00013 from tornado import netutil
00014 from tornado.testing import AsyncHTTPTestCase, LogTrapTestCase, get_unused_port
00015 from tornado.util import b, bytes_type
00016 from tornado.web import Application, RequestHandler, url
00017 
00018 
00019 class HelloWorldHandler(RequestHandler):
00020     def get(self):
00021         name = self.get_argument("name", "world")
00022         self.set_header("Content-Type", "text/plain")
00023         self.finish("Hello %s!" % name)
00024 
00025 
00026 class PostHandler(RequestHandler):
00027     def post(self):
00028         self.finish("Post arg1: %s, arg2: %s" % (
00029             self.get_argument("arg1"), self.get_argument("arg2")))
00030 
00031 
00032 class ChunkHandler(RequestHandler):
00033     def get(self):
00034         self.write("asdf")
00035         self.flush()
00036         self.write("qwer")
00037 
00038 
00039 class AuthHandler(RequestHandler):
00040     def get(self):
00041         self.finish(self.request.headers["Authorization"])
00042 
00043 
00044 class CountdownHandler(RequestHandler):
00045     def get(self, count):
00046         count = int(count)
00047         if count > 0:
00048             self.redirect(self.reverse_url("countdown", count - 1))
00049         else:
00050             self.write("Zero")
00051 
00052 
00053 class EchoPostHandler(RequestHandler):
00054     def post(self):
00055         self.write(self.request.body)
00056 
00057 # These tests end up getting run redundantly: once here with the default
00058 # HTTPClient implementation, and then again in each implementation's own
00059 # test suite.
00060 
00061 
00062 class HTTPClientCommonTestCase(AsyncHTTPTestCase, LogTrapTestCase):
00063     def get_http_client(self):
00064         """Returns AsyncHTTPClient instance.  May be overridden in subclass."""
00065         return AsyncHTTPClient(io_loop=self.io_loop)
00066 
00067     def get_app(self):
00068         return Application([
00069             url("/hello", HelloWorldHandler),
00070             url("/post", PostHandler),
00071             url("/chunk", ChunkHandler),
00072             url("/auth", AuthHandler),
00073             url("/countdown/([0-9]+)", CountdownHandler, name="countdown"),
00074             url("/echopost", EchoPostHandler),
00075             ], gzip=True)
00076 
00077     def setUp(self):
00078         super(HTTPClientCommonTestCase, self).setUp()
00079         # replace the client defined in the parent class
00080         self.http_client = self.get_http_client()
00081 
00082     def test_hello_world(self):
00083         response = self.fetch("/hello")
00084         self.assertEqual(response.code, 200)
00085         self.assertEqual(response.headers["Content-Type"], "text/plain")
00086         self.assertEqual(response.body, b("Hello world!"))
00087         self.assertEqual(int(response.request_time), 0)
00088 
00089         response = self.fetch("/hello?name=Ben")
00090         self.assertEqual(response.body, b("Hello Ben!"))
00091 
00092     def test_streaming_callback(self):
00093         # streaming_callback is also tested in test_chunked
00094         chunks = []
00095         response = self.fetch("/hello",
00096                               streaming_callback=chunks.append)
00097         # with streaming_callback, data goes to the callback and not response.body
00098         self.assertEqual(chunks, [b("Hello world!")])
00099         self.assertFalse(response.body)
00100 
00101     def test_post(self):
00102         response = self.fetch("/post", method="POST",
00103                               body="arg1=foo&arg2=bar")
00104         self.assertEqual(response.code, 200)
00105         self.assertEqual(response.body, b("Post arg1: foo, arg2: bar"))
00106 
00107     def test_chunked(self):
00108         response = self.fetch("/chunk")
00109         self.assertEqual(response.body, b("asdfqwer"))
00110 
00111         chunks = []
00112         response = self.fetch("/chunk",
00113                               streaming_callback=chunks.append)
00114         self.assertEqual(chunks, [b("asdf"), b("qwer")])
00115         self.assertFalse(response.body)
00116 
00117     def test_chunked_close(self):
00118         # test case in which chunks spread read-callback processing
00119         # over several ioloop iterations, but the connection is already closed.
00120         port = get_unused_port()
00121         (sock,) = netutil.bind_sockets(port, address="127.0.0.1")
00122         with closing(sock):
00123             def write_response(stream, request_data):
00124                 stream.write(b("""\
00125 HTTP/1.1 200 OK
00126 Transfer-Encoding: chunked
00127 
00128 1
00129 1
00130 1
00131 2
00132 0
00133 
00134 """).replace(b("\n"), b("\r\n")), callback=stream.close)
00135 
00136             def accept_callback(conn, address):
00137                 # fake an HTTP server using chunked encoding where the final chunks
00138                 # and connection close all happen at once
00139                 stream = IOStream(conn, io_loop=self.io_loop)
00140                 stream.read_until(b("\r\n\r\n"),
00141                                   functools.partial(write_response, stream))
00142             netutil.add_accept_handler(sock, accept_callback, self.io_loop)
00143             self.http_client.fetch("http://127.0.0.1:%d/" % port, self.stop)
00144             resp = self.wait()
00145             resp.rethrow()
00146             self.assertEqual(resp.body, b("12"))
00147 
00148     def test_basic_auth(self):
00149         self.assertEqual(self.fetch("/auth", auth_username="Aladdin",
00150                                     auth_password="open sesame").body,
00151                          b("Basic QWxhZGRpbjpvcGVuIHNlc2FtZQ=="))
00152 
00153     def test_follow_redirect(self):
00154         response = self.fetch("/countdown/2", follow_redirects=False)
00155         self.assertEqual(302, response.code)
00156         self.assertTrue(response.headers["Location"].endswith("/countdown/1"))
00157 
00158         response = self.fetch("/countdown/2")
00159         self.assertEqual(200, response.code)
00160         self.assertTrue(response.effective_url.endswith("/countdown/0"))
00161         self.assertEqual(b("Zero"), response.body)
00162 
00163     def test_credentials_in_url(self):
00164         url = self.get_url("/auth").replace("http://", "http://me:secret@")
00165         self.http_client.fetch(url, self.stop)
00166         response = self.wait()
00167         self.assertEqual(b("Basic ") + base64.b64encode(b("me:secret")),
00168                          response.body)
00169 
00170     def test_body_encoding(self):
00171         unicode_body = u"\xe9"
00172         byte_body = binascii.a2b_hex(b("e9"))
00173 
00174         # unicode string in body gets converted to utf8
00175         response = self.fetch("/echopost", method="POST", body=unicode_body,
00176                               headers={"Content-Type": "application/blah"})
00177         self.assertEqual(response.headers["Content-Length"], "2")
00178         self.assertEqual(response.body, utf8(unicode_body))
00179 
00180         # byte strings pass through directly
00181         response = self.fetch("/echopost", method="POST",
00182                               body=byte_body,
00183                               headers={"Content-Type": "application/blah"})
00184         self.assertEqual(response.headers["Content-Length"], "1")
00185         self.assertEqual(response.body, byte_body)
00186 
00187         # Mixing unicode in headers and byte string bodies shouldn't
00188         # break anything
00189         response = self.fetch("/echopost", method="POST", body=byte_body,
00190                               headers={"Content-Type": "application/blah"},
00191                               user_agent=u"foo")
00192         self.assertEqual(response.headers["Content-Length"], "1")
00193         self.assertEqual(response.body, byte_body)
00194 
00195     def test_types(self):
00196         response = self.fetch("/hello")
00197         self.assertEqual(type(response.body), bytes_type)
00198         self.assertEqual(type(response.headers["Content-Type"]), str)
00199         self.assertEqual(type(response.code), int)
00200         self.assertEqual(type(response.effective_url), str)


rosbridge_server
Author(s): Jonathan Mace
autogenerated on Thu Jan 2 2014 11:53:55