<s> from __future__ import absolute_import , division , print_function , with_statement <EOL> import collections <EOL> from contextlib import closing <EOL> import errno <EOL> import gzip <EOL> import logging <EOL> import os <EOL> import re <EOL> import socket <EOL> import sys <EOL> from tornado import gen <EOL> from tornado . httpclient import AsyncHTTPClient <EOL> from tornado . httputil import HTTPHeaders <EOL> from tornado . ioloop import IOLoop <EOL> from tornado . log import gen_log , app_log <EOL> from tornado . netutil import Resolver , bind_sockets <EOL> from tornado . simple_httpclient import SimpleAsyncHTTPClient , _default_ca_certs <EOL> from tornado . test . httpclient_test import ChunkHandler , CountdownHandler , HelloWorldHandler <EOL> from tornado . test import httpclient_test <EOL> from tornado . testing import AsyncHTTPTestCase , AsyncHTTPSTestCase , AsyncTestCase , bind_unused_port , ExpectLog <EOL> from tornado . test . util import skipOnTravis , skipIfNoIPv6 <EOL> from tornado . web import RequestHandler , Application , asynchronous , url , stream_request_body <EOL> class SimpleHTTPClientCommonTestCase ( httpclient_test . HTTPClientCommonTestCase ) : <EOL> def get_http_client ( self ) : <EOL> client = SimpleAsyncHTTPClient ( io_loop = self . io_loop , <EOL> force_instance = True ) <EOL> self . assertTrue ( isinstance ( client , SimpleAsyncHTTPClient ) ) <EOL> return client <EOL> class TriggerHandler ( RequestHandler ) : <EOL> def initialize ( self , queue , wake_callback ) : <EOL> self . queue = queue <EOL> self . wake_callback = wake_callback <EOL> @ asynchronous <EOL> def get ( self ) : <EOL> logging . debug ( "<STR_LIT>" ) <EOL> self . queue . append ( self . finish ) <EOL> if self . get_argument ( "<STR_LIT>" , "<STR_LIT:true>" ) == "<STR_LIT:true>" : <EOL> self . wake_callback ( ) <EOL> class HangHandler ( RequestHandler ) : <EOL> @ asynchronous <EOL> def get ( self ) : <EOL> pass <EOL> class ContentLengthHandler ( RequestHandler ) : <EOL> def get ( self ) : <EOL> self . set_header ( "<STR_LIT>" , self . get_argument ( "<STR_LIT:value>" ) ) <EOL> self . write ( "<STR_LIT>" ) <EOL> class HeadHandler ( RequestHandler ) : <EOL> def head ( self ) : <EOL> self . set_header ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> class OptionsHandler ( RequestHandler ) : <EOL> def options ( self ) : <EOL> self . set_header ( "<STR_LIT>" , "<STR_LIT:*>" ) <EOL> self . write ( "<STR_LIT>" ) <EOL> class NoContentHandler ( RequestHandler ) : <EOL> def get ( self ) : <EOL> if self . get_argument ( "<STR_LIT:error>" , None ) : <EOL> self . set_header ( "<STR_LIT>" , "<STR_LIT:5>" ) <EOL> self . write ( "<STR_LIT:hello>" ) <EOL> self . set_status ( <NUM_LIT> ) <EOL> class SeeOtherPostHandler ( RequestHandler ) : <EOL> def post ( self ) : <EOL> redirect_code = int ( self . request . body ) <EOL> assert redirect_code in ( <NUM_LIT> , <NUM_LIT> ) , "<STR_LIT>" % self . request . body <EOL> self . set_header ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . set_status ( redirect_code ) <EOL> class SeeOtherGetHandler ( RequestHandler ) : <EOL> def get ( self ) : <EOL> if self . request . body : <EOL> raise Exception ( "<STR_LIT>" % self . request . body ) <EOL> self . write ( "<STR_LIT>" ) <EOL> class HostEchoHandler ( RequestHandler ) : <EOL> def get ( self ) : <EOL> self . write ( self . request . headers [ "<STR_LIT>" ] ) <EOL> class NoContentLengthHandler ( RequestHandler ) : <EOL> @ gen . coroutine <EOL> def get ( self ) : <EOL> stream = self . request . connection . stream <EOL> yield stream . write ( b"<STR_LIT>" <EOL> b"<STR_LIT:hello>" ) <EOL> stream . close ( ) <EOL> class EchoPostHandler ( RequestHandler ) : <EOL> def post ( self ) : <EOL> self . write ( self . request . body ) <EOL> @ stream_request_body <EOL> class RespondInPrepareHandler ( RequestHandler ) : <EOL> def prepare ( self ) : <EOL> self . set_status ( <NUM_LIT> ) <EOL> self . finish ( "<STR_LIT>" ) <EOL> class SimpleHTTPClientTestMixin ( object ) : <EOL> def get_app ( self ) : <EOL> self . triggers = collections . deque ( ) <EOL> return Application ( [ <EOL> url ( "<STR_LIT>" , TriggerHandler , dict ( queue = self . triggers , <EOL> wake_callback = self . stop ) ) , <EOL> url ( "<STR_LIT>" , ChunkHandler ) , <EOL> url ( "<STR_LIT>" , CountdownHandler , name = "<STR_LIT>" ) , <EOL> url ( "<STR_LIT>" , HangHandler ) , <EOL> url ( "<STR_LIT>" , HelloWorldHandler ) , <EOL> url ( "<STR_LIT>" , ContentLengthHandler ) , <EOL> url ( "<STR_LIT>" , HeadHandler ) , <EOL> url ( "<STR_LIT>" , OptionsHandler ) , <EOL> url ( "<STR_LIT>" , NoContentHandler ) , <EOL> url ( "<STR_LIT>" , SeeOtherPostHandler ) , <EOL> url ( "<STR_LIT>" , SeeOtherGetHandler ) , <EOL> url ( "<STR_LIT>" , HostEchoHandler ) , <EOL> url ( "<STR_LIT>" , NoContentLengthHandler ) , <EOL> url ( "<STR_LIT>" , EchoPostHandler ) , <EOL> url ( "<STR_LIT>" , RespondInPrepareHandler ) , <EOL> ] , gzip = True ) <EOL> def test_singleton ( self ) : <EOL> self . assertTrue ( SimpleAsyncHTTPClient ( self . io_loop ) is <EOL> SimpleAsyncHTTPClient ( self . io_loop ) ) <EOL> self . assertTrue ( SimpleAsyncHTTPClient ( self . io_loop ) is not <EOL> SimpleAsyncHTTPClient ( self . io_loop , <EOL> force_instance = True ) ) <EOL> with closing ( IOLoop ( ) ) as io_loop2 : <EOL> self . assertTrue ( SimpleAsyncHTTPClient ( self . io_loop ) is not <EOL> SimpleAsyncHTTPClient ( io_loop2 ) ) <EOL> def test_connection_limit ( self ) : <EOL> with closing ( self . create_client ( max_clients = <NUM_LIT:2> ) ) as client : <EOL> self . assertEqual ( client . max_clients , <NUM_LIT:2> ) <EOL> seen = [ ] <EOL> for i in range ( <NUM_LIT:4> ) : <EOL> client . fetch ( self . get_url ( "<STR_LIT>" ) , <EOL> lambda response , i = i : ( seen . append ( i ) , self . stop ( ) ) ) <EOL> self . wait ( condition = lambda : len ( self . triggers ) == <NUM_LIT:2> ) <EOL> self . assertEqual ( len ( client . queue ) , <NUM_LIT:2> ) <EOL> self . triggers . popleft ( ) ( ) <EOL> self . triggers . popleft ( ) ( ) <EOL> self . wait ( condition = lambda : ( len ( self . triggers ) == <NUM_LIT:2> and <EOL> len ( seen ) == <NUM_LIT:2> ) ) <EOL> self . assertEqual ( set ( seen ) , set ( [ <NUM_LIT:0> , <NUM_LIT:1> ] ) ) <EOL> self . assertEqual ( len ( client . queue ) , <NUM_LIT:0> ) <EOL> self . triggers . popleft ( ) ( ) <EOL> self . triggers . popleft ( ) ( ) <EOL> self . wait ( condition = lambda : len ( seen ) == <NUM_LIT:4> ) <EOL> self . assertEqual ( set ( seen ) , set ( [ <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] ) ) <EOL> self . assertEqual ( len ( self . triggers ) , <NUM_LIT:0> ) <EOL> def test_redirect_connection_limit ( self ) : <EOL> with closing ( self . create_client ( max_clients = <NUM_LIT:1> ) ) as client : <EOL> client . fetch ( self . get_url ( '<STR_LIT>' ) , self . stop , <EOL> max_redirects = <NUM_LIT:3> ) <EOL> response = self . wait ( ) <EOL> response . rethrow ( ) <EOL> def test_default_certificates_exist ( self ) : <EOL> open ( _default_ca_certs ( ) ) . close ( ) <EOL> def test_gzip ( self ) : <EOL> response = self . fetch ( "<STR_LIT>" , use_gzip = False , <EOL> headers = { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> self . assertEqual ( response . headers [ "<STR_LIT>" ] , "<STR_LIT>" ) <EOL> self . assertNotEqual ( response . body , b"<STR_LIT>" ) <EOL> self . assertEqual ( len ( response . body ) , <NUM_LIT> ) <EOL> f = gzip . GzipFile ( mode = "<STR_LIT:r>" , fileobj = response . buffer ) <EOL> self . assertEqual ( f . read ( ) , b"<STR_LIT>" ) <EOL> def test_max_redirects ( self ) : <EOL> response = self . fetch ( "<STR_LIT>" , max_redirects = <NUM_LIT:3> ) <EOL> self . assertEqual ( <NUM_LIT> , response . code ) <EOL> self . assertTrue ( response . request . url . endswith ( "<STR_LIT>" ) ) <EOL> self . assertTrue ( response . effective_url . endswith ( "<STR_LIT>" ) ) <EOL> self . assertTrue ( response . headers [ "<STR_LIT>" ] . endswith ( "<STR_LIT>" ) ) <EOL> def test_header_reuse ( self ) : <EOL> headers = HTTPHeaders ( { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . fetch ( "<STR_LIT>" , headers = headers ) <EOL> self . assertEqual ( list ( headers . get_all ( ) ) , [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> def test_see_other_redirect ( self ) : <EOL> for code in ( <NUM_LIT> , <NUM_LIT> ) : <EOL> response = self . fetch ( "<STR_LIT>" , method = "<STR_LIT:POST>" , body = "<STR_LIT>" % code ) <EOL> self . assertEqual ( <NUM_LIT:200> , response . code ) <EOL> self . assertTrue ( response . request . url . endswith ( "<STR_LIT>" ) ) <EOL> self . assertTrue ( response . effective_url . endswith ( "<STR_LIT>" ) ) <EOL> self . assertEqual ( "<STR_LIT:POST>" , response . request . method ) <EOL> @ skipOnTravis <EOL> def test_request_timeout ( self ) : <EOL> response = self . fetch ( '<STR_LIT>' , request_timeout = <NUM_LIT:0.1> ) <EOL> self . assertEqual ( response . code , <NUM_LIT> ) <EOL> self . assertTrue ( <NUM_LIT> < response . request_time < <NUM_LIT> , response . request_time ) <EOL> self . assertEqual ( str ( response . error ) , "<STR_LIT>" ) <EOL> self . triggers . popleft ( ) ( ) <EOL> @ skipIfNoIPv6 <EOL> def test_ipv6 ( self ) : <EOL> try : <EOL> [ sock ] = bind_sockets ( None , '<STR_LIT>' , family = socket . AF_INET6 ) <EOL> port = sock . getsockname ( ) [ <NUM_LIT:1> ] <EOL> self . http_server . add_socket ( sock ) <EOL> except socket . gaierror as e : <EOL> if e . args [ <NUM_LIT:0> ] == socket . EAI_ADDRFAMILY : <EOL> return <EOL> raise <EOL> url = '<STR_LIT>' % ( self . get_protocol ( ) , port ) <EOL> self . http_client . fetch ( url , self . stop , allow_ipv6 = False ) <EOL> response = self . wait ( ) <EOL> self . assertEqual ( response . code , <NUM_LIT> ) <EOL> self . http_client . fetch ( url , self . stop ) <EOL> response = self . wait ( ) <EOL> self . assertEqual ( response . body , b"<STR_LIT>" ) <EOL> def xtest_multiple_content_length_accepted ( self ) : <EOL> response = self . fetch ( "<STR_LIT>" ) <EOL> self . assertEqual ( response . body , b"<STR_LIT>" ) <EOL> response = self . fetch ( "<STR_LIT>" ) <EOL> self . assertEqual ( response . body , b"<STR_LIT>" ) <EOL> response = self . fetch ( "<STR_LIT>" ) <EOL> self . assertEqual ( response . code , <NUM_LIT> ) <EOL> response = self . fetch ( "<STR_LIT>" ) <EOL> self . assertEqual ( response . code , <NUM_LIT> ) <EOL> def test_head_request ( self ) : <EOL> response = self . fetch ( "<STR_LIT>" , method = "<STR_LIT>" ) <EOL> self . assertEqual ( response . code , <NUM_LIT:200> ) <EOL> self . assertEqual ( response . headers [ "<STR_LIT>" ] , "<STR_LIT>" ) <EOL> self . assertFalse ( response . body ) <EOL> def test_options_request ( self ) : <EOL> response = self . fetch ( "<STR_LIT>" , method = "<STR_LIT>" ) <EOL> self . assertEqual ( response . code , <NUM_LIT:200> ) <EOL> self . assertEqual ( response . headers [ "<STR_LIT>" ] , "<STR_LIT:2>" ) <EOL> self . assertEqual ( response . headers [ "<STR_LIT>" ] , "<STR_LIT:*>" ) <EOL> self . assertEqual ( response . body , b"<STR_LIT>" ) <EOL> def test_no_content ( self ) : <EOL> response = self . fetch ( "<STR_LIT>" ) <EOL> self . assertEqual ( response . code , <NUM_LIT> ) <EOL> self . assertEqual ( response . headers [ "<STR_LIT>" ] , "<STR_LIT:0>" ) <EOL> with ExpectLog ( gen_log , "<STR_LIT>" ) : <EOL> response = self . fetch ( "<STR_LIT>" ) <EOL> self . assertEqual ( response . code , <NUM_LIT> ) <EOL> def test_host_header ( self ) : <EOL> host_re = re . compile ( b"<STR_LIT>" ) <EOL> response = self . fetch ( "<STR_LIT>" ) <EOL> self . assertTrue ( host_re . match ( response . body ) ) <EOL> url = self . get_url ( "<STR_LIT>" ) . replace ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . http_client . fetch ( url , self . stop ) <EOL> response = self . wait ( ) <EOL> self . assertTrue ( host_re . match ( response . body ) , response . body ) <EOL> def test_connection_refused ( self ) : <EOL> server_socket , port = bind_unused_port ( ) <EOL> server_socket . close ( ) <EOL> with ExpectLog ( gen_log , "<STR_LIT>" , required = False ) : <EOL> self . http_client . fetch ( "<STR_LIT>" % port , self . stop ) <EOL> response = self . wait ( ) <EOL> self . assertEqual ( <NUM_LIT> , response . code ) <EOL> if sys . platform != '<STR_LIT>' : <EOL> contains_errno = str ( errno . ECONNREFUSED ) in str ( response . error ) <EOL> if not contains_errno and hasattr ( errno , "<STR_LIT>" ) : <EOL> contains_errno = str ( errno . WSAECONNREFUSED ) in str ( response . error ) <EOL> self . assertTrue ( contains_errno , response . error ) <EOL> expected_message = os . strerror ( errno . ECONNREFUSED ) <EOL> self . assertTrue ( expected_message in str ( response . error ) , <EOL> response . error ) <EOL> def test_queue_timeout ( self ) : <EOL> with closing ( self . create_client ( max_clients = <NUM_LIT:1> ) ) as client : <EOL> client . fetch ( self . get_url ( '<STR_LIT>' ) , self . stop , <EOL> request_timeout = <NUM_LIT:10> ) <EOL> self . wait ( ) <EOL> client . fetch ( self . get_url ( '<STR_LIT>' ) , self . stop , <EOL> connect_timeout = <NUM_LIT:0.1> ) <EOL> response = self . wait ( ) <EOL> self . assertEqual ( response . code , <NUM_LIT> ) <EOL> self . assertTrue ( response . request_time < <NUM_LIT:1> , response . request_time ) <EOL> self . assertEqual ( str ( response . error ) , "<STR_LIT>" ) <EOL> self . triggers . popleft ( ) ( ) <EOL> self . wait ( ) <EOL> def test_no_content_length ( self ) : <EOL> response = self . fetch ( "<STR_LIT>" ) <EOL> self . assertEquals ( b"<STR_LIT:hello>" , response . body ) <EOL> def sync_body_producer ( self , write ) : <EOL> write ( b'<STR_LIT>' ) <EOL> write ( b'<STR_LIT>' ) <EOL> @ gen . coroutine <EOL> def async_body_producer ( self , write ) : <EOL> yield write ( b'<STR_LIT>' ) <EOL> yield gen . Task ( IOLoop . current ( ) . add_callback ) <EOL> yield write ( b'<STR_LIT>' ) <EOL> def test_sync_body_producer_chunked ( self ) : <EOL> response = self . fetch ( "<STR_LIT>" , method = "<STR_LIT:POST>" , <EOL> body_producer = self . sync_body_producer ) <EOL> response . rethrow ( ) <EOL> self . assertEqual ( response . body , b"<STR_LIT>" ) <EOL> def test_sync_body_producer_content_length ( self ) : <EOL> response = self . fetch ( "<STR_LIT>" , method = "<STR_LIT:POST>" , <EOL> body_producer = self . sync_body_producer , <EOL> headers = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> response . rethrow ( ) <EOL> self . assertEqual ( response . body , b"<STR_LIT>" ) <EOL> def test_async_body_producer_chunked ( self ) : <EOL> response = self . fetch ( "<STR_LIT>" , method = "<STR_LIT:POST>" , <EOL> body_producer = self . async_body_producer ) <EOL> response . rethrow ( ) <EOL> self . assertEqual ( response . body , b"<STR_LIT>" ) <EOL> def test_async_body_producer_content_length ( self ) : <EOL> response = self . fetch ( "<STR_LIT>" , method = "<STR_LIT:POST>" , <EOL> body_producer = self . async_body_producer , <EOL> headers = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> response . rethrow ( ) <EOL> self . assertEqual ( response . body , b"<STR_LIT>" ) <EOL> def test_100_continue ( self ) : <EOL> response = self . fetch ( "<STR_LIT>" , method = "<STR_LIT:POST>" , <EOL> body = b"<STR_LIT>" , <EOL> expect_100_continue = True ) <EOL> self . assertEqual ( response . body , b"<STR_LIT>" ) <EOL> def test_100_continue_early_response ( self ) : <EOL> def body_producer ( write ) : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> response = self . fetch ( "<STR_LIT>" , method = "<STR_LIT:POST>" , <EOL> body_producer = body_producer , <EOL> expect_100_continue = True ) <EOL> self . assertEqual ( response . code , <NUM_LIT> ) <EOL> class SimpleHTTPClientTestCase ( SimpleHTTPClientTestMixin , AsyncHTTPTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( SimpleHTTPClientTestCase , self ) . setUp ( ) <EOL> self . http_client = self . create_client ( ) <EOL> def create_client ( self , ** kwargs ) : <EOL> return SimpleAsyncHTTPClient ( self . io_loop , force_instance = True , <EOL> ** kwargs ) <EOL> class SimpleHTTPSClientTestCase ( SimpleHTTPClientTestMixin , AsyncHTTPSTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( SimpleHTTPSClientTestCase , self ) . setUp ( ) <EOL> self . http_client = self . create_client ( ) <EOL> def create_client ( self , ** kwargs ) : <EOL> return SimpleAsyncHTTPClient ( self . io_loop , force_instance = True , <EOL> defaults = dict ( validate_cert = False ) , <EOL> ** kwargs ) <EOL> class CreateAsyncHTTPClientTestCase ( AsyncTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( CreateAsyncHTTPClientTestCase , self ) . setUp ( ) <EOL> self . saved = AsyncHTTPClient . _save_configuration ( ) <EOL> def tearDown ( self ) : <EOL> AsyncHTTPClient . _restore_configuration ( self . saved ) <EOL> super ( CreateAsyncHTTPClientTestCase , self ) . tearDown ( ) <EOL> def test_max_clients ( self ) : <EOL> AsyncHTTPClient . configure ( SimpleAsyncHTTPClient ) <EOL> with closing ( AsyncHTTPClient ( <EOL> self . io_loop , force_instance = True ) ) as client : <EOL> self . assertEqual ( client . max_clients , <NUM_LIT:10> ) <EOL> with closing ( AsyncHTTPClient ( <EOL> self . io_loop , max_clients = <NUM_LIT:11> , force_instance = True ) ) as client : <EOL> self . assertEqual ( client . max_clients , <NUM_LIT:11> ) <EOL> AsyncHTTPClient . configure ( SimpleAsyncHTTPClient , max_clients = <NUM_LIT:12> ) <EOL> with closing ( AsyncHTTPClient ( <EOL> self . io_loop , force_instance = True ) ) as client : <EOL> self . assertEqual ( client . max_clients , <NUM_LIT:12> ) <EOL> with closing ( AsyncHTTPClient ( <EOL> self . io_loop , max_clients = <NUM_LIT> , force_instance = True ) ) as client : <EOL> self . assertEqual ( client . max_clients , <NUM_LIT> ) <EOL> with closing ( AsyncHTTPClient ( <EOL> self . io_loop , max_clients = <NUM_LIT> , force_instance = True ) ) as client : <EOL> self . assertEqual ( client . max_clients , <NUM_LIT> ) <EOL> class HTTP100ContinueTestCase ( AsyncHTTPTestCase ) : <EOL> def respond_100 ( self , request ) : <EOL> self . request = request <EOL> self . request . connection . stream . write ( <EOL> b"<STR_LIT>" , <EOL> self . respond_200 ) <EOL> def respond_200 ( self ) : <EOL> self . request . connection . stream . write ( <EOL> b"<STR_LIT>" , <EOL> self . request . connection . stream . close ) <EOL> def get_app ( self ) : <EOL> return self . respond_100 <EOL> def test_100_continue ( self ) : <EOL> res = self . fetch ( '<STR_LIT:/>' ) <EOL> self . assertEqual ( res . body , b'<STR_LIT:A>' ) <EOL> class HTTP204NoContentTestCase ( AsyncHTTPTestCase ) : <EOL> def respond_204 ( self , request ) : <EOL> request . connection . stream . write ( <EOL> b"<STR_LIT>" ) <EOL> def get_app ( self ) : <EOL> return self . respond_204 <EOL> def test_204_no_content ( self ) : <EOL> resp = self . fetch ( '<STR_LIT:/>' ) <EOL> self . assertEqual ( resp . code , <NUM_LIT> ) <EOL> self . assertEqual ( resp . body , b'<STR_LIT>' ) <EOL> class HostnameMappingTestCase ( AsyncHTTPTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( HostnameMappingTestCase , self ) . setUp ( ) <EOL> self . http_client = SimpleAsyncHTTPClient ( <EOL> self . io_loop , <EOL> hostname_mapping = { <EOL> '<STR_LIT>' : '<STR_LIT:127.0.0.1>' , <EOL> ( '<STR_LIT>' , <NUM_LIT> ) : ( '<STR_LIT:127.0.0.1>' , self . get_http_port ( ) ) , <EOL> } ) <EOL> def get_app ( self ) : <EOL> return Application ( [ url ( "<STR_LIT>" , HelloWorldHandler ) , ] ) <EOL> def test_hostname_mapping ( self ) : <EOL> self . http_client . fetch ( <EOL> '<STR_LIT>' % self . get_http_port ( ) , self . stop ) <EOL> response = self . wait ( ) <EOL> response . rethrow ( ) <EOL> self . assertEqual ( response . body , b'<STR_LIT>' ) <EOL> def test_port_mapping ( self ) : <EOL> self . http_client . fetch ( '<STR_LIT>' , self . stop ) <EOL> response = self . wait ( ) <EOL> response . rethrow ( ) <EOL> self . assertEqual ( response . body , b'<STR_LIT>' ) <EOL> class ResolveTimeoutTestCase ( AsyncHTTPTestCase ) : <EOL> def setUp ( self ) : <EOL> class BadResolver ( Resolver ) : <EOL> def resolve ( self , * args , ** kwargs ) : <EOL> pass <EOL> super ( ResolveTimeoutTestCase , self ) . setUp ( ) <EOL> self . http_client = SimpleAsyncHTTPClient ( <EOL> self . io_loop , <EOL> resolver = BadResolver ( ) ) <EOL> def get_app ( self ) : <EOL> return Application ( [ url ( "<STR_LIT>" , HelloWorldHandler ) , ] ) <EOL> def test_resolve_timeout ( self ) : <EOL> response = self . fetch ( '<STR_LIT>' , connect_timeout = <NUM_LIT:0.1> ) <EOL> self . assertEqual ( response . code , <NUM_LIT> ) <EOL> class MaxHeaderSizeTest ( AsyncHTTPTestCase ) : <EOL> def get_app ( self ) : <EOL> class SmallHeaders ( RequestHandler ) : <EOL> def get ( self ) : <EOL> self . set_header ( "<STR_LIT>" , "<STR_LIT:a>" * <NUM_LIT:100> ) <EOL> self . write ( "<STR_LIT>" ) <EOL> class LargeHeaders ( RequestHandler ) : <EOL> def get ( self ) : <EOL> self . set_header ( "<STR_LIT>" , "<STR_LIT:a>" * <NUM_LIT:1000> ) <EOL> self . write ( "<STR_LIT>" ) <EOL> return Application ( [ ( '<STR_LIT>' , SmallHeaders ) , <EOL> ( '<STR_LIT>' , LargeHeaders ) ] ) <EOL> def get_http_client ( self ) : <EOL> return SimpleAsyncHTTPClient ( io_loop = self . io_loop , max_header_size = <NUM_LIT> ) <EOL> def test_small_headers ( self ) : <EOL> response = self . fetch ( '<STR_LIT>' ) <EOL> response . rethrow ( ) <EOL> self . assertEqual ( response . body , b'<STR_LIT>' ) <EOL> def test_large_headers ( self ) : <EOL> with ExpectLog ( gen_log , "<STR_LIT>" ) : <EOL> response = self . fetch ( '<STR_LIT>' ) <EOL> self . assertEqual ( response . code , <NUM_LIT> ) </s>
