<s> import unittest <EOL> import json <EOL> from pants . http import HTTPClient <EOL> from pants . engine import Engine <EOL> from pants . test . _pants_util import * <EOL> try : <EOL> HTTPClient ( verify_ssl = True ) <EOL> VERIFY_SSL = True <EOL> except RuntimeError : <EOL> VERIFY_SSL = False <EOL> class HTTPTestCase ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . engine = Engine . instance ( ) <EOL> self . client = HTTPClient ( self . on_response , self . on_headers , self . on_progress , self . on_ssl_error , self . on_error , engine = self . engine ) <EOL> def start ( self , timeout = <NUM_LIT> ) : <EOL> self . _timeout = self . engine . defer ( timeout , self . timeout ) <EOL> self . engine . start ( ) <EOL> self . assertTrue ( self . got_response ) <EOL> self . assertTrue ( self . response_valid ) <EOL> def timeout ( self ) : <EOL> self . stop ( ) <EOL> raise AssertionError ( "<STR_LIT>" ) <EOL> def stop ( self ) : <EOL> self . _timeout ( ) <EOL> del self . _timeout <EOL> self . engine . stop ( ) <EOL> def tearDown ( self ) : <EOL> if self . client . _stream : <EOL> self . client . _want_close = True <EOL> self . client . _no_process = True <EOL> self . client . _stream . close ( ) <EOL> self . client . _stream = None <EOL> del self . client <EOL> del self . engine <EOL> got_response = False <EOL> response_valid = True <EOL> def on_response ( self , response ) : <EOL> self . stop ( ) <EOL> self . got_response = True <EOL> def on_headers ( self , response ) : <EOL> pass <EOL> def on_progress ( self , response , received , total ) : <EOL> pass <EOL> def on_ssl_error ( self , response , cert , error ) : <EOL> pass <EOL> def on_error ( self , response , error ) : <EOL> pass <EOL> class GetTest ( HTTPTestCase ) : <EOL> def on_response ( self , response ) : <EOL> self . got_response = True <EOL> self . stop ( ) <EOL> def test_get ( self ) : <EOL> self . client . get ( "<STR_LIT>" , { "<STR_LIT:foo>" : "<STR_LIT:bar>" } ) <EOL> self . start ( ) <EOL> class PostTest ( HTTPTestCase ) : <EOL> def on_response ( self , response ) : <EOL> self . stop ( ) <EOL> self . got_response = True <EOL> data = response . json <EOL> if not data [ "<STR_LIT>" ] [ "<STR_LIT:foo>" ] == "<STR_LIT:bar>" : <EOL> self . response_valid = False <EOL> def test_post ( self ) : <EOL> self . client . post ( "<STR_LIT>" , { "<STR_LIT:foo>" : "<STR_LIT:bar>" } ) <EOL> self . start ( ) <EOL> def test_multipart ( self ) : <EOL> self . got_response = False <EOL> self . response_valid = True <EOL> self . client . post ( "<STR_LIT>" , { "<STR_LIT:foo>" : "<STR_LIT:bar>" } , <EOL> { "<STR_LIT:file>" : ( "<STR_LIT>" , "<STR_LIT>" ) } ) <EOL> self . start ( ) <EOL> class HostChangeTest ( HTTPTestCase ) : <EOL> resp_count = <NUM_LIT:0> <EOL> def on_response ( self , response ) : <EOL> self . resp_count += <NUM_LIT:1> <EOL> if self . resp_count < <NUM_LIT:2> : <EOL> return <EOL> self . stop ( ) <EOL> self . got_response = True <EOL> def test_cookie ( self ) : <EOL> self . client . get ( "<STR_LIT>" ) <EOL> self . client . get ( "<STR_LIT>" ) <EOL> self . start ( ) <EOL> class TimeoutTest ( HTTPTestCase ) : <EOL> def on_response ( self , response ) : <EOL> self . stop ( ) <EOL> self . response_valid = False <EOL> def on_error ( self , response , error ) : <EOL> self . stop ( ) <EOL> self . got_response = True <EOL> def test_timeout ( self ) : <EOL> with self . client . session ( timeout = <NUM_LIT:1> ) as ses : <EOL> ses . get ( "<STR_LIT>" ) <EOL> self . start ( ) <EOL> class BadHostTest ( HTTPTestCase ) : <EOL> def on_response ( self , response ) : <EOL> self . stop ( ) <EOL> def on_error ( self , response , error ) : <EOL> self . got_response = True <EOL> self . stop ( ) <EOL> def test_bad_host ( self ) : <EOL> self . client . get ( "<STR_LIT>" ) <EOL> self . start ( ) <EOL> class BadPortTest ( HTTPTestCase ) : <EOL> def on_response ( self , response ) : <EOL> self . stop ( ) <EOL> def on_error ( self , response , error ) : <EOL> self . got_response = True <EOL> self . stop ( ) <EOL> def test_bad_port ( self ) : <EOL> self . client . get ( "<STR_LIT>" ) <EOL> self . start ( ) <EOL> class GzippedTest ( HTTPTestCase ) : <EOL> def on_response ( self , response ) : <EOL> self . stop ( ) <EOL> self . got_response = True <EOL> data = response . json <EOL> if not data [ "<STR_LIT>" ] and response . headers [ '<STR_LIT>' ] == '<STR_LIT>' : <EOL> self . response_valid = False <EOL> def test_gzipped ( self ) : <EOL> self . client . get ( "<STR_LIT>" ) <EOL> self . start ( ) <EOL> class TeapotTest ( HTTPTestCase ) : <EOL> def on_response ( self , response ) : <EOL> self . stop ( ) <EOL> self . got_response = True <EOL> if not response . status_code == <NUM_LIT> : <EOL> self . response_valid = False <EOL> def test_teapot ( self ) : <EOL> self . client . get ( "<STR_LIT>" ) <EOL> self . start ( ) <EOL> class RedirectTest ( HTTPTestCase ) : <EOL> def on_response ( self , response ) : <EOL> self . stop ( ) <EOL> self . got_response = True <EOL> if not len ( response . history ) == <NUM_LIT:3> or not response . path == "<STR_LIT>" : <EOL> self . response_valid = False <EOL> def test_redirect ( self ) : <EOL> self . client . get ( "<STR_LIT>" ) <EOL> self . start ( ) <EOL> class RedirectLimitTest ( HTTPTestCase ) : <EOL> def on_response ( self , response ) : <EOL> self . stop ( ) <EOL> self . got_response = True <EOL> if not ( len ( response . history ) == <NUM_LIT:10> and response . status_code == <NUM_LIT> ) : <EOL> self . response_valid = False <EOL> def test_limit ( self ) : <EOL> self . client . get ( "<STR_LIT>" ) <EOL> self . start ( <NUM_LIT:10> ) <EOL> class RedirectRelativeTest ( HTTPTestCase ) : <EOL> def on_response ( self , response ) : <EOL> self . stop ( ) <EOL> self . got_response = True <EOL> if not response . path == "<STR_LIT>" : <EOL> self . response_valid = False <EOL> def test_relative ( self ) : <EOL> self . client . get ( "<STR_LIT>" ) <EOL> self . start ( ) <EOL> class CookieTest ( HTTPTestCase ) : <EOL> resp_count = <NUM_LIT:0> <EOL> def on_response ( self , response ) : <EOL> self . resp_count += <NUM_LIT:1> <EOL> if self . resp_count < <NUM_LIT:2> : <EOL> return <EOL> self . stop ( ) <EOL> self . got_response = True <EOL> if not response . cookies [ "<STR_LIT:foo>" ] . value == "<STR_LIT:bar>" : <EOL> self . response_valid = False <EOL> def test_cookie ( self ) : <EOL> self . client . get ( "<STR_LIT>" ) <EOL> self . client . get ( "<STR_LIT>" ) <EOL> self . start ( ) <EOL> class AuthTest ( HTTPTestCase ) : <EOL> def on_response ( self , response ) : <EOL> self . stop ( ) <EOL> self . got_response = True <EOL> if not response . status_code == <NUM_LIT> : <EOL> self . response_valid = False <EOL> def test_auth ( self ) : <EOL> self . client . get ( "<STR_LIT>" ) <EOL> self . start ( ) <EOL> class DoAuthTest ( HTTPTestCase ) : <EOL> def on_response ( self , response ) : <EOL> self . stop ( ) <EOL> self . got_response = True <EOL> if not response . status_code == <NUM_LIT:200> : <EOL> self . response_valid = False <EOL> def test_do_auth ( self ) : <EOL> with self . client . session ( auth = ( "<STR_LIT:user>" , "<STR_LIT>" ) ) as ses : <EOL> ses . get ( "<STR_LIT>" ) <EOL> self . start ( ) <EOL> class HTTPSTest ( HTTPTestCase ) : <EOL> def on_response ( self , response ) : <EOL> self . stop ( ) <EOL> self . got_response = True <EOL> def test_https ( self ) : <EOL> self . client . get ( "<STR_LIT>" ) <EOL> self . start ( ) <EOL> @ unittest . skipIf ( not VERIFY_SSL , "<STR_LIT>" ) <EOL> class BadCertTest ( HTTPTestCase ) : <EOL> def on_response ( self , response ) : <EOL> self . stop ( ) <EOL> def on_ssl_error ( self , response , cert , error ) : <EOL> self . stop ( ) <EOL> self . got_response = True <EOL> def test_bad ( self ) : <EOL> with self . client . session ( verify_ssl = True ) as ses : <EOL> ses . get ( "<STR_LIT>" ) <EOL> self . start ( ) <EOL> @ unittest . skipIf ( not VERIFY_SSL , "<STR_LIT>" ) <EOL> class SSLOverrideTest ( HTTPTestCase ) : <EOL> def on_response ( self , response ) : <EOL> self . stop ( ) <EOL> self . got_response = True <EOL> def on_ssl_error ( self , response , cert , error ) : <EOL> return True <EOL> def test_override ( self ) : <EOL> with self . client . session ( verify_ssl = True ) as ses : <EOL> ses . get ( "<STR_LIT>" ) <EOL> self . start ( ) <EOL> class ProgressTest ( HTTPTestCase ) : <EOL> def on_response ( self , response ) : <EOL> self . stop ( ) <EOL> def on_progress ( self , response , received , total ) : <EOL> self . got_response = True <EOL> def test_progress ( self ) : <EOL> self . client . get ( "<STR_LIT>" ) <EOL> self . start ( ) <EOL> class ForceURLEncodedTest ( HTTPTestCase ) : <EOL> def on_response ( self , response ) : <EOL> self . stop ( ) <EOL> self . got_response = True <EOL> def test_force_urlencoded ( self ) : <EOL> self . client . post ( "<STR_LIT>" , <EOL> headers = { "<STR_LIT:Content-Type>" : "<STR_LIT>" } ) <EOL> self . start ( ) </s>
