<s> """<STR_LIT>""" <EOL> import os <EOL> import unittest <EOL> import time <EOL> import tempfile <EOL> import pycurl <EOL> import itertools <EOL> import threading <EOL> from cStringIO import StringIO <EOL> from ganeti import http <EOL> from ganeti import compat <EOL> from ganeti . rapi . auth import users_file <EOL> import ganeti . http . server <EOL> import ganeti . http . client <EOL> import ganeti . http . auth <EOL> import testutils <EOL> class TestStartLines ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def testClientToServerStartLine ( self ) : <EOL> """<STR_LIT>""" <EOL> start_line = http . HttpClientToServerStartLine ( "<STR_LIT:GET>" , "<STR_LIT:/>" , "<STR_LIT>" ) <EOL> self . assertEqual ( str ( start_line ) , "<STR_LIT>" ) <EOL> def testServerToClientStartLine ( self ) : <EOL> """<STR_LIT>""" <EOL> start_line = http . HttpServerToClientStartLine ( "<STR_LIT>" , <NUM_LIT:200> , "<STR_LIT:OK>" ) <EOL> self . assertEqual ( str ( start_line ) , "<STR_LIT>" ) <EOL> class TestMisc ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def _TestDateTimeHeader ( self , gmnow , expected ) : <EOL> self . assertEqual ( http . server . _DateTimeHeader ( gmnow = gmnow ) , expected ) <EOL> def testDateTimeHeader ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _TestDateTimeHeader ( ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> , <NUM_LIT:5> , <NUM_LIT:3> , <NUM_LIT:0> , <NUM_LIT:0> ) , <EOL> "<STR_LIT>" ) <EOL> self . _TestDateTimeHeader ( ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) , <EOL> "<STR_LIT>" ) <EOL> self . _TestDateTimeHeader ( ( <NUM_LIT> , <NUM_LIT:12> , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) , <EOL> "<STR_LIT>" ) <EOL> self . _TestDateTimeHeader ( ( <NUM_LIT> , <NUM_LIT:12> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) , <EOL> "<STR_LIT>" ) <EOL> self . _TestDateTimeHeader ( ( <NUM_LIT> , <NUM_LIT:12> , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:6> , <NUM_LIT:0> , <NUM_LIT:0> ) , <EOL> "<STR_LIT>" ) <EOL> def testHttpServerRequest ( self ) : <EOL> """<STR_LIT>""" <EOL> server_request = http . server . _HttpServerRequest ( "<STR_LIT:GET>" , "<STR_LIT:/>" , None , None ) <EOL> self . assert_ ( hasattr ( server_request , "<STR_LIT>" ) ) <EOL> self . assert_ ( hasattr ( server_request , "<STR_LIT>" ) ) <EOL> self . assert_ ( hasattr ( server_request , "<STR_LIT>" ) ) <EOL> self . assert_ ( hasattr ( server_request , "<STR_LIT>" ) ) <EOL> self . assert_ ( isinstance ( server_request . resp_headers , dict ) ) <EOL> self . assert_ ( hasattr ( server_request , "<STR_LIT>" ) ) <EOL> def testServerSizeLimits ( self ) : <EOL> """<STR_LIT>""" <EOL> message_reader_class = http . server . _HttpClientToServerMessageReader <EOL> self . assert_ ( message_reader_class . START_LINE_LENGTH_MAX > <NUM_LIT:0> ) <EOL> self . assert_ ( message_reader_class . HEADER_LENGTH_MAX > <NUM_LIT:0> ) <EOL> def testFormatAuthHeader ( self ) : <EOL> self . assertEqual ( http . auth . _FormatAuthHeader ( "<STR_LIT>" , { } ) , <EOL> "<STR_LIT>" ) <EOL> self . assertEqual ( http . auth . _FormatAuthHeader ( "<STR_LIT>" , { "<STR_LIT:foo>" : "<STR_LIT:bar>" , } ) , <EOL> "<STR_LIT>" ) <EOL> self . assertEqual ( http . auth . _FormatAuthHeader ( "<STR_LIT>" , { "<STR_LIT:foo>" : "<STR_LIT>" , } ) , <EOL> "<STR_LIT>" ) <EOL> self . assertEqual ( http . auth . _FormatAuthHeader ( "<STR_LIT>" , { "<STR_LIT:foo>" : "<STR_LIT>" , } ) , <EOL> "<STR_LIT>" ) <EOL> params = { <EOL> "<STR_LIT:foo>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } <EOL> self . assert_ ( http . auth . _FormatAuthHeader ( "<STR_LIT>" , params ) in <EOL> ( "<STR_LIT>" , <EOL> "<STR_LIT>" ) ) <EOL> class _FakeRequestAuth ( http . auth . HttpServerRequestAuthentication ) : <EOL> def __init__ ( self , realm , authreq , authenticator ) : <EOL> http . auth . HttpServerRequestAuthentication . __init__ ( self ) <EOL> self . realm = realm <EOL> self . authreq = authreq <EOL> self . authenticator = authenticator <EOL> def AuthenticationRequired ( self , req ) : <EOL> return self . authreq <EOL> def GetAuthRealm ( self , req ) : <EOL> return self . realm <EOL> def Authenticate ( self , req ) : <EOL> handler_access = [ ] <EOL> if self . authenticator : <EOL> return self . authenticator . ValidateRequest ( <EOL> req , handler_access , self . GetAuthRealm ( req ) ) <EOL> raise NotImplementedError ( ) <EOL> class TestAuth ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> hsra = http . auth . HttpServerRequestAuthentication <EOL> def testConstants ( self ) : <EOL> for scheme in [ self . hsra . _CLEARTEXT_SCHEME , self . hsra . _HA1_SCHEME ] : <EOL> self . assertEqual ( scheme , scheme . upper ( ) ) <EOL> self . assert_ ( scheme . startswith ( "<STR_LIT:{>" ) ) <EOL> self . assert_ ( scheme . endswith ( "<STR_LIT:}>" ) ) <EOL> def _testVerifyBasicAuthPassword ( self , realm , user , password , expected ) : <EOL> ra = _FakeRequestAuth ( realm , False , None ) <EOL> return ra . VerifyBasicAuthPassword ( user , password , expected , realm ) <EOL> def testVerifyBasicAuthPassword ( self ) : <EOL> tvbap = self . _testVerifyBasicAuthPassword <EOL> good_pws = [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT:0>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] <EOL> for pw in good_pws : <EOL> self . assert_ ( tvbap ( "<STR_LIT:abc>" , "<STR_LIT:user>" , pw , pw ) ) <EOL> self . assert_ ( tvbap ( "<STR_LIT:abc>" , "<STR_LIT:user>" , pw , "<STR_LIT>" + pw ) ) <EOL> self . assert_ ( tvbap ( "<STR_LIT:abc>" , "<STR_LIT:user>" , pw , "<STR_LIT>" + pw ) ) <EOL> self . assert_ ( tvbap ( "<STR_LIT:abc>" , "<STR_LIT:user>" , pw , "<STR_LIT>" + pw ) ) <EOL> self . failIf ( tvbap ( "<STR_LIT:abc>" , "<STR_LIT:user>" , pw , "<STR_LIT>" ) ) <EOL> self . failIf ( tvbap ( "<STR_LIT:abc>" , "<STR_LIT:user>" , pw , "<STR_LIT>" + pw ) ) <EOL> self . failIf ( tvbap ( "<STR_LIT:abc>" , "<STR_LIT:user>" , pw , "<STR_LIT>" + pw ) ) <EOL> self . failIf ( tvbap ( "<STR_LIT:abc>" , "<STR_LIT:user>" , pw , "<STR_LIT>" + pw ) ) <EOL> self . failIf ( tvbap ( "<STR_LIT:abc>" , "<STR_LIT:user>" , pw , "<STR_LIT>" + pw ) ) <EOL> self . failIf ( tvbap ( "<STR_LIT:abc>" , "<STR_LIT:user>" , "<STR_LIT>" , "<STR_LIT>" ) ) <EOL> self . assert_ ( tvbap ( "<STR_LIT>" , "<STR_LIT:user>" , "<STR_LIT>" , <EOL> "<STR_LIT>" ) ) <EOL> self . assert_ ( tvbap ( "<STR_LIT>" , "<STR_LIT:user>" , "<STR_LIT>" , <EOL> "<STR_LIT>" ) ) <EOL> self . failUnlessRaises ( AssertionError , tvbap , None , "<STR_LIT:user>" , "<STR_LIT>" , <EOL> "<STR_LIT>" ) <EOL> self . failIf ( tvbap ( "<STR_LIT>" , "<STR_LIT:user>" , "<STR_LIT>" , <EOL> "<STR_LIT>" ) ) <EOL> self . failIf ( tvbap ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" ) ) <EOL> self . failIf ( tvbap ( "<STR_LIT>" , "<STR_LIT:user>" , "<STR_LIT>" , <EOL> "<STR_LIT>" ) ) <EOL> class _SimpleAuthenticator : <EOL> def __init__ ( self , user , password ) : <EOL> self . user = user <EOL> self . password = password <EOL> self . called = False <EOL> def ValidateRequest ( self , req , handler_access , realm ) : <EOL> self . called = True <EOL> username , password = http . auth . HttpServerRequestAuthentication . ExtractUserPassword ( req ) <EOL> if username is None or password is None : <EOL> return False <EOL> return ( self . user == username and <EOL> http . auth . HttpServerRequestAuthentication . VerifyBasicAuthPassword ( <EOL> username , password , self . password , realm ) ) <EOL> class TestHttpServerRequestAuthentication ( unittest . TestCase ) : <EOL> def testNoAuth ( self ) : <EOL> req = http . server . _HttpServerRequest ( "<STR_LIT:GET>" , "<STR_LIT:/>" , None , None ) <EOL> _FakeRequestAuth ( "<STR_LIT>" , False , None ) . PreHandleRequest ( req ) <EOL> def testNoRealm ( self ) : <EOL> headers = { http . HTTP_AUTHORIZATION : "<STR_LIT>" , } <EOL> req = http . server . _HttpServerRequest ( "<STR_LIT:GET>" , "<STR_LIT:/>" , headers , None ) <EOL> ac = _SimpleAuthenticator ( "<STR_LIT:foo>" , "<STR_LIT:bar>" ) <EOL> ra = _FakeRequestAuth ( None , False , ac ) <EOL> self . assertRaises ( AssertionError , ra . PreHandleRequest , req ) <EOL> def testNoScheme ( self ) : <EOL> headers = { http . HTTP_AUTHORIZATION : "<STR_LIT>" , } <EOL> req = http . server . _HttpServerRequest ( "<STR_LIT:GET>" , "<STR_LIT:/>" , headers , None ) <EOL> ac = _SimpleAuthenticator ( "<STR_LIT:foo>" , "<STR_LIT:bar>" ) <EOL> ra = _FakeRequestAuth ( "<STR_LIT>" , False , ac ) <EOL> self . assertRaises ( http . HttpUnauthorized , ra . PreHandleRequest , req ) <EOL> def testUnknownScheme ( self ) : <EOL> headers = { http . HTTP_AUTHORIZATION : "<STR_LIT>" , } <EOL> req = http . server . _HttpServerRequest ( "<STR_LIT:GET>" , "<STR_LIT:/>" , headers , None ) <EOL> ac = _SimpleAuthenticator ( "<STR_LIT:foo>" , "<STR_LIT:bar>" ) <EOL> ra = _FakeRequestAuth ( "<STR_LIT>" , False , ac ) <EOL> self . assertRaises ( http . HttpUnauthorized , ra . PreHandleRequest , req ) <EOL> def testInvalidBase64 ( self ) : <EOL> headers = { http . HTTP_AUTHORIZATION : "<STR_LIT>" , } <EOL> req = http . server . _HttpServerRequest ( "<STR_LIT:GET>" , "<STR_LIT:/>" , headers , None ) <EOL> ac = _SimpleAuthenticator ( "<STR_LIT:foo>" , "<STR_LIT:bar>" ) <EOL> ra = _FakeRequestAuth ( "<STR_LIT>" , False , ac ) <EOL> self . assertRaises ( http . HttpBadRequest , ra . PreHandleRequest , req ) <EOL> def testAuthForPublicResource ( self ) : <EOL> headers = { <EOL> http . HTTP_AUTHORIZATION : "<STR_LIT>" % ( "<STR_LIT:foo>" . encode ( "<STR_LIT>" ) . strip ( ) , ) , <EOL> } <EOL> req = http . server . _HttpServerRequest ( "<STR_LIT:GET>" , "<STR_LIT:/>" , headers , None ) <EOL> ra = _FakeRequestAuth ( "<STR_LIT>" , False , None ) <EOL> self . assertRaises ( http . HttpUnauthorized , ra . PreHandleRequest , req ) <EOL> def testAuthForPublicResource ( self ) : <EOL> headers = { <EOL> http . HTTP_AUTHORIZATION : <EOL> "<STR_LIT>" % ( "<STR_LIT>" . encode ( "<STR_LIT>" ) . strip ( ) , ) , <EOL> } <EOL> req = http . server . _HttpServerRequest ( "<STR_LIT:GET>" , "<STR_LIT:/>" , headers , None ) <EOL> ac = _SimpleAuthenticator ( "<STR_LIT:foo>" , "<STR_LIT:bar>" ) <EOL> ra = _FakeRequestAuth ( "<STR_LIT>" , False , ac ) <EOL> ra . PreHandleRequest ( req ) <EOL> req = http . server . _HttpServerRequest ( "<STR_LIT:GET>" , "<STR_LIT:/>" , headers , None ) <EOL> ac = _SimpleAuthenticator ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> ra = _FakeRequestAuth ( "<STR_LIT>" , False , ac ) <EOL> self . assertRaises ( http . HttpUnauthorized , ra . PreHandleRequest , req ) <EOL> def testInvalidRequestHeader ( self ) : <EOL> checks = { <EOL> http . HttpUnauthorized : [ "<STR_LIT>" , "<STR_LIT:\t>" , "<STR_LIT:->" , "<STR_LIT:.>" , "<STR_LIT:@>" , "<STR_LIT:<>" , "<STR_LIT:>>" , "<STR_LIT>" , <EOL> "<STR_LIT>" % "<STR_LIT>" . encode ( "<STR_LIT>" ) . strip ( ) ] , <EOL> http . HttpBadRequest : [ "<STR_LIT>" ] , <EOL> } <EOL> ac = _SimpleAuthenticator ( "<STR_LIT:foo>" , "<STR_LIT:bar>" ) <EOL> for exc , headers in checks . items ( ) : <EOL> for i in headers : <EOL> headers = { http . HTTP_AUTHORIZATION : i , } <EOL> req = http . server . _HttpServerRequest ( "<STR_LIT:GET>" , "<STR_LIT:/>" , headers , None ) <EOL> ra = _FakeRequestAuth ( "<STR_LIT>" , False , ac ) <EOL> self . assertRaises ( exc , ra . PreHandleRequest , req ) <EOL> def testBasicAuth ( self ) : <EOL> for user in [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] : <EOL> for pw in [ "<STR_LIT>" , "<STR_LIT:->" , "<STR_LIT::>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" ] : <EOL> for wrong_pw in [ True , False ] : <EOL> basic_auth = "<STR_LIT>" % ( user , pw ) <EOL> if wrong_pw : <EOL> basic_auth += "<STR_LIT>" <EOL> headers = { <EOL> http . HTTP_AUTHORIZATION : <EOL> "<STR_LIT>" % ( basic_auth . encode ( "<STR_LIT>" ) . strip ( ) , ) , <EOL> } <EOL> req = http . server . _HttpServerRequest ( "<STR_LIT:GET>" , "<STR_LIT:/>" , headers , None ) <EOL> ac = _SimpleAuthenticator ( user , pw ) <EOL> self . assertFalse ( ac . called ) <EOL> ra = _FakeRequestAuth ( "<STR_LIT>" , True , ac ) <EOL> if wrong_pw : <EOL> try : <EOL> ra . PreHandleRequest ( req ) <EOL> except http . HttpUnauthorized , err : <EOL> www_auth = err . headers [ http . HTTP_WWW_AUTHENTICATE ] <EOL> self . assert_ ( www_auth . startswith ( http . auth . HTTP_BASIC_AUTH ) ) <EOL> else : <EOL> self . fail ( "<STR_LIT>" ) <EOL> else : <EOL> ra . PreHandleRequest ( req ) <EOL> self . assert_ ( ac . called ) <EOL> class TestReadPasswordFile ( unittest . TestCase ) : <EOL> def testSimple ( self ) : <EOL> users = users_file . ParsePasswordFile ( "<STR_LIT>" ) <EOL> self . assertEqual ( len ( users ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( users [ "<STR_LIT>" ] . password , "<STR_LIT:password>" ) <EOL> self . assertEqual ( len ( users [ "<STR_LIT>" ] . options ) , <NUM_LIT:0> ) <EOL> def testOptions ( self ) : <EOL> buf = StringIO ( ) <EOL> buf . write ( "<STR_LIT>" ) <EOL> buf . write ( "<STR_LIT>" ) <EOL> buf . write ( "<STR_LIT:\n>" ) <EOL> buf . write ( "<STR_LIT>" ) <EOL> buf . write ( "<STR_LIT>" ) <EOL> buf . write ( "<STR_LIT>" ) <EOL> buf . write ( "<STR_LIT>" ) <EOL> users = users_file . ParsePasswordFile ( buf . getvalue ( ) ) <EOL> self . assertEqual ( len ( users ) , <NUM_LIT:2> ) <EOL> self . assertEqual ( users [ "<STR_LIT>" ] . password , "<STR_LIT:password>" ) <EOL> self . assertEqual ( len ( users [ "<STR_LIT>" ] . options ) , <NUM_LIT:0> ) <EOL> self . assertEqual ( users [ "<STR_LIT>" ] . password , "<STR_LIT>" ) <EOL> self . assertEqual ( users [ "<STR_LIT>" ] . options , [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> class TestClientRequest ( unittest . TestCase ) : <EOL> def testRepr ( self ) : <EOL> cr = http . client . HttpClientRequest ( "<STR_LIT:localhost>" , <NUM_LIT> , "<STR_LIT:GET>" , "<STR_LIT>" , <EOL> headers = [ ] , post_data = "<STR_LIT>" ) <EOL> self . assert_ ( repr ( cr ) . startswith ( "<STR_LIT:<>" ) ) <EOL> def testNoHeaders ( self ) : <EOL> cr = http . client . HttpClientRequest ( "<STR_LIT:localhost>" , <NUM_LIT> , "<STR_LIT:GET>" , "<STR_LIT>" , <EOL> headers = None ) <EOL> self . assert_ ( isinstance ( cr . headers , list ) ) <EOL> self . assertEqual ( cr . headers , [ ] ) <EOL> self . assertEqual ( cr . url , "<STR_LIT>" ) <EOL> def testPlainAddressIPv4 ( self ) : <EOL> cr = http . client . HttpClientRequest ( "<STR_LIT>" , <NUM_LIT> , "<STR_LIT:GET>" , "<STR_LIT>" ) <EOL> self . assertEqual ( cr . url , "<STR_LIT>" ) <EOL> def testPlainAddressIPv6 ( self ) : <EOL> cr = http . client . HttpClientRequest ( "<STR_LIT>" , <NUM_LIT> , "<STR_LIT:GET>" , "<STR_LIT>" ) <EOL> self . assertEqual ( cr . url , "<STR_LIT>" ) <EOL> def testOldStyleHeaders ( self ) : <EOL> headers = { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } <EOL> cr = http . client . HttpClientRequest ( "<STR_LIT:localhost>" , <NUM_LIT> , "<STR_LIT:GET>" , "<STR_LIT>" , <EOL> headers = headers ) <EOL> self . assert_ ( isinstance ( cr . headers , list ) ) <EOL> self . assertEqual ( sorted ( cr . headers ) , [ <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> ] ) <EOL> self . assertEqual ( cr . url , "<STR_LIT>" ) <EOL> def testNewStyleHeaders ( self ) : <EOL> headers = [ <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> ] <EOL> cr = http . client . HttpClientRequest ( "<STR_LIT:localhost>" , <NUM_LIT> , "<STR_LIT:GET>" , "<STR_LIT>" , <EOL> headers = headers ) <EOL> self . assert_ ( isinstance ( cr . headers , list ) ) <EOL> self . assertEqual ( sorted ( cr . headers ) , sorted ( headers ) ) <EOL> self . assertEqual ( cr . url , "<STR_LIT>" ) <EOL> def testPostData ( self ) : <EOL> cr = http . client . HttpClientRequest ( "<STR_LIT:localhost>" , <NUM_LIT> , "<STR_LIT:GET>" , "<STR_LIT>" , <EOL> post_data = "<STR_LIT>" ) <EOL> self . assertEqual ( cr . post_data , "<STR_LIT>" ) <EOL> def testNoPostData ( self ) : <EOL> cr = http . client . HttpClientRequest ( "<STR_LIT:localhost>" , <NUM_LIT> , "<STR_LIT:GET>" , "<STR_LIT>" ) <EOL> self . assertEqual ( cr . post_data , "<STR_LIT>" ) <EOL> def testCompletionCallback ( self ) : <EOL> for argname in [ "<STR_LIT>" , "<STR_LIT>" ] : <EOL> kwargs = { <EOL> argname : NotImplementedError , <EOL> } <EOL> cr = http . client . HttpClientRequest ( "<STR_LIT:localhost>" , <NUM_LIT> , "<STR_LIT:GET>" , "<STR_LIT>" , <EOL> ** kwargs ) <EOL> self . assertEqual ( getattr ( cr , argname ) , NotImplementedError ) <EOL> for fn in [ NotImplemented , { } , <NUM_LIT:1> ] : <EOL> kwargs = { <EOL> argname : fn , <EOL> } <EOL> self . assertRaises ( AssertionError , http . client . HttpClientRequest , <EOL> "<STR_LIT:localhost>" , <NUM_LIT> , "<STR_LIT:GET>" , "<STR_LIT>" , ** kwargs ) <EOL> class _FakeCurl : <EOL> def __init__ ( self ) : <EOL> self . opts = { } <EOL> self . info = NotImplemented <EOL> def setopt ( self , opt , value ) : <EOL> assert opt not in self . opts , "<STR_LIT>" <EOL> self . opts [ opt ] = value <EOL> def getinfo ( self , info ) : <EOL> return self . info . pop ( info ) <EOL> class TestClientStartRequest ( unittest . TestCase ) : <EOL> @ staticmethod <EOL> def _TestCurlConfig ( curl ) : <EOL> curl . setopt ( pycurl . SSLKEYTYPE , "<STR_LIT>" ) <EOL> def test ( self ) : <EOL> for method in [ http . HTTP_GET , http . HTTP_PUT , "<STR_LIT>" ] : <EOL> for port in [ <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] : <EOL> for curl_config_fn in [ None , self . _TestCurlConfig ] : <EOL> for read_timeout in [ None , <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT> , <NUM_LIT> ] : <EOL> self . _TestInner ( method , port , curl_config_fn , read_timeout ) <EOL> def _TestInner ( self , method , port , curl_config_fn , read_timeout ) : <EOL> for response_code in [ http . HTTP_OK , http . HttpNotFound . code , <EOL> http . HTTP_NOT_MODIFIED ] : <EOL> for response_body in [ None , "<STR_LIT>" , <EOL> "<STR_LIT>" * <NUM_LIT> ] : <EOL> for errmsg in [ None , "<STR_LIT:error>" ] : <EOL> req = http . client . HttpClientRequest ( "<STR_LIT:localhost>" , port , method , <EOL> "<STR_LIT>" , <EOL> curl_config_fn = curl_config_fn , <EOL> read_timeout = read_timeout ) <EOL> curl = _FakeCurl ( ) <EOL> pending = http . client . _StartRequest ( curl , req ) <EOL> self . assertEqual ( pending . GetCurlHandle ( ) , curl ) <EOL> self . assertEqual ( pending . GetCurrentRequest ( ) , req ) <EOL> opts = curl . opts <EOL> self . assertEqual ( opts . pop ( pycurl . CUSTOMREQUEST ) , method ) <EOL> self . assertEqual ( opts . pop ( pycurl . URL ) , <EOL> "<STR_LIT>" % port ) <EOL> if read_timeout is None : <EOL> self . assertEqual ( opts . pop ( pycurl . TIMEOUT ) , <NUM_LIT:0> ) <EOL> else : <EOL> self . assertEqual ( opts . pop ( pycurl . TIMEOUT ) , read_timeout ) <EOL> self . assertFalse ( opts . pop ( pycurl . VERBOSE ) ) <EOL> self . assertTrue ( opts . pop ( pycurl . NOSIGNAL ) ) <EOL> self . assertEqual ( opts . pop ( pycurl . USERAGENT ) , <EOL> http . HTTP_GANETI_VERSION ) <EOL> self . assertEqual ( opts . pop ( pycurl . PROXY ) , "<STR_LIT>" ) <EOL> self . assertFalse ( opts . pop ( pycurl . POSTFIELDS ) ) <EOL> self . assertFalse ( opts . pop ( pycurl . HTTPHEADER ) ) <EOL> write_fn = opts . pop ( pycurl . WRITEFUNCTION ) <EOL> self . assertTrue ( callable ( write_fn ) ) <EOL> if hasattr ( pycurl , "<STR_LIT>" ) : <EOL> self . assertFalse ( opts . pop ( pycurl . SSL_SESSIONID_CACHE ) ) <EOL> if curl_config_fn : <EOL> self . assertEqual ( opts . pop ( pycurl . SSLKEYTYPE ) , "<STR_LIT>" ) <EOL> else : <EOL> self . assertFalse ( pycurl . SSLKEYTYPE in opts ) <EOL> self . assertFalse ( opts ) <EOL> if response_body is not None : <EOL> offset = <NUM_LIT:0> <EOL> while offset < len ( response_body ) : <EOL> piece = response_body [ offset : offset + <NUM_LIT:10> ] <EOL> write_fn ( piece ) <EOL> offset += len ( piece ) <EOL> curl . info = { <EOL> pycurl . RESPONSE_CODE : response_code , <EOL> } <EOL> if hasattr ( pycurl , '<STR_LIT>' ) : <EOL> curl . info [ pycurl . LOCAL_IP ] = '<STR_LIT:127.0.0.1>' <EOL> if hasattr ( pycurl , '<STR_LIT>' ) : <EOL> curl . info [ pycurl . LOCAL_PORT ] = port <EOL> pending . Done ( errmsg ) <EOL> self . assertFalse ( curl . info ) <EOL> self . assertRaises ( AssertionError , pending . Done , True ) <EOL> if errmsg : <EOL> self . assertFalse ( req . success ) <EOL> else : <EOL> self . assertTrue ( req . success ) <EOL> self . assertEqual ( req . error , errmsg ) <EOL> self . assertEqual ( req . resp_status_code , response_code ) <EOL> if response_body is None : <EOL> self . assertEqual ( req . resp_body , "<STR_LIT>" ) <EOL> else : <EOL> self . assertEqual ( req . resp_body , response_body ) <EOL> assert not hasattr ( curl , "<STR_LIT>" ) <EOL> opts = curl . opts <EOL> self . assertFalse ( opts . pop ( pycurl . POSTFIELDS ) ) <EOL> self . assertTrue ( callable ( opts . pop ( pycurl . WRITEFUNCTION ) ) ) <EOL> self . assertFalse ( opts ) <EOL> self . assertFalse ( curl . opts , <EOL> msg = "<STR_LIT>" ) <EOL> assert id ( opts ) == id ( curl . opts ) <EOL> def _TestWrongTypes ( self , * args , ** kwargs ) : <EOL> req = http . client . HttpClientRequest ( * args , ** kwargs ) <EOL> self . assertRaises ( AssertionError , http . client . _StartRequest , <EOL> _FakeCurl ( ) , req ) <EOL> def testWrongHostType ( self ) : <EOL> self . _TestWrongTypes ( unicode ( "<STR_LIT:localhost>" ) , <NUM_LIT> , "<STR_LIT:GET>" , "<STR_LIT>" ) <EOL> def testWrongUrlType ( self ) : <EOL> self . _TestWrongTypes ( "<STR_LIT:localhost>" , <NUM_LIT> , "<STR_LIT:GET>" , unicode ( "<STR_LIT>" ) ) <EOL> def testWrongMethodType ( self ) : <EOL> self . _TestWrongTypes ( "<STR_LIT:localhost>" , <NUM_LIT> , unicode ( "<STR_LIT:GET>" ) , "<STR_LIT>" ) <EOL> def testWrongHeaderType ( self ) : <EOL> self . _TestWrongTypes ( "<STR_LIT:localhost>" , <NUM_LIT> , "<STR_LIT:GET>" , "<STR_LIT>" , <EOL> headers = { <EOL> unicode ( "<STR_LIT:foo>" ) : "<STR_LIT:bar>" , <EOL> } ) <EOL> def testWrongPostDataType ( self ) : <EOL> self . _TestWrongTypes ( "<STR_LIT:localhost>" , <NUM_LIT> , "<STR_LIT:GET>" , "<STR_LIT>" , <EOL> post_data = unicode ( "<STR_LIT>" * <NUM_LIT:100> ) ) <EOL> class _EmptyCurlMulti : <EOL> def perform ( self ) : <EOL> return ( pycurl . E_MULTI_OK , <NUM_LIT:0> ) <EOL> def info_read ( self ) : <EOL> return ( <NUM_LIT:0> , [ ] , [ ] ) <EOL> class TestClientProcessRequests ( unittest . TestCase ) : <EOL> def testEmpty ( self ) : <EOL> requests = [ ] <EOL> http . client . ProcessRequests ( requests , _curl = NotImplemented , <EOL> _curl_multi = _EmptyCurlMulti ) <EOL> self . assertEqual ( requests , [ ] ) <EOL> class TestProcessCurlRequests ( unittest . TestCase ) : <EOL> class _FakeCurlMulti : <EOL> def __init__ ( self ) : <EOL> self . handles = [ ] <EOL> self . will_fail = [ ] <EOL> self . _expect = [ "<STR_LIT>" ] <EOL> self . _counter = itertools . count ( ) <EOL> def add_handle ( self , curl ) : <EOL> assert curl not in self . handles <EOL> self . handles . append ( curl ) <EOL> if self . _counter . next ( ) % <NUM_LIT:3> == <NUM_LIT:0> : <EOL> self . will_fail . append ( curl ) <EOL> def remove_handle ( self , curl ) : <EOL> self . handles . remove ( curl ) <EOL> def perform ( self ) : <EOL> assert self . _expect . pop ( <NUM_LIT:0> ) == "<STR_LIT>" <EOL> if self . _counter . next ( ) % <NUM_LIT:2> == <NUM_LIT:0> : <EOL> self . _expect . append ( "<STR_LIT>" ) <EOL> return ( pycurl . E_CALL_MULTI_PERFORM , None ) <EOL> self . _expect . append ( "<STR_LIT>" ) <EOL> return ( pycurl . E_MULTI_OK , len ( self . handles ) ) <EOL> def info_read ( self ) : <EOL> assert self . _expect . pop ( <NUM_LIT:0> ) == "<STR_LIT>" <EOL> successful = [ ] <EOL> failed = [ ] <EOL> if self . handles : <EOL> if self . _counter . next ( ) % <NUM_LIT> == <NUM_LIT:0> : <EOL> curl = self . handles [ <NUM_LIT:0> ] <EOL> if curl in self . will_fail : <EOL> failed . append ( ( curl , - <NUM_LIT:1> , "<STR_LIT>" ) ) <EOL> else : <EOL> successful . append ( curl ) <EOL> remaining_messages = len ( self . handles ) % <NUM_LIT:3> <EOL> if remaining_messages > <NUM_LIT:0> : <EOL> self . _expect . append ( "<STR_LIT>" ) <EOL> else : <EOL> self . _expect . append ( "<STR_LIT>" ) <EOL> else : <EOL> remaining_messages = <NUM_LIT:0> <EOL> self . _expect . append ( "<STR_LIT>" ) <EOL> return ( remaining_messages , successful , failed ) <EOL> def select ( self , timeout ) : <EOL> assert timeout >= <NUM_LIT> and timeout <= <NUM_LIT> <EOL> assert self . _expect . pop ( <NUM_LIT:0> ) == "<STR_LIT>" <EOL> self . _expect . append ( "<STR_LIT>" ) <EOL> def test ( self ) : <EOL> requests = [ _FakeCurl ( ) for _ in range ( <NUM_LIT:10> ) ] <EOL> multi = self . _FakeCurlMulti ( ) <EOL> for ( curl , errmsg ) in http . client . _ProcessCurlRequests ( multi , requests ) : <EOL> self . assertTrue ( curl not in multi . handles ) <EOL> if curl in multi . will_fail : <EOL> self . assertTrue ( "<STR_LIT>" in errmsg ) <EOL> else : <EOL> self . assertTrue ( errmsg is None ) <EOL> self . assertFalse ( multi . handles ) <EOL> self . assertEqual ( multi . _expect , [ "<STR_LIT>" ] ) <EOL> class TestProcessRequests ( unittest . TestCase ) : <EOL> class _DummyCurlMulti : <EOL> pass <EOL> def testNoMonitor ( self ) : <EOL> self . _Test ( False ) <EOL> def testWithMonitor ( self ) : <EOL> self . _Test ( True ) <EOL> class _MonitorChecker : <EOL> def __init__ ( self ) : <EOL> self . _monitor = None <EOL> def GetMonitor ( self ) : <EOL> return self . _monitor <EOL> def __call__ ( self , monitor ) : <EOL> assert callable ( monitor . GetLockInfo ) <EOL> self . _monitor = monitor <EOL> def _Test ( self , use_monitor ) : <EOL> def cfg_fn ( port , curl ) : <EOL> curl . opts [ "<STR_LIT>" ] = port <EOL> def _LockCheckReset ( monitor , req ) : <EOL> self . assertTrue ( monitor . _lock . is_owned ( shared = <NUM_LIT:0> ) , <EOL> msg = "<STR_LIT>" ) <EOL> assert not hasattr ( req , "<STR_LIT>" ) <EOL> setattr ( req , "<STR_LIT>" , True ) <EOL> def _BuildNiceName ( port , default = None ) : <EOL> if port % <NUM_LIT:5> == <NUM_LIT:0> : <EOL> return "<STR_LIT>" % port <EOL> else : <EOL> return default <EOL> requests = [ http . client . HttpClientRequest ( "<STR_LIT:localhost>" , i , "<STR_LIT:POST>" , "<STR_LIT>" % i , <EOL> curl_config_fn = compat . partial ( cfg_fn , i ) , <EOL> completion_cb = NotImplementedError , <EOL> nicename = _BuildNiceName ( i ) ) <EOL> for i in range ( <NUM_LIT> , <NUM_LIT> ) ] <EOL> requests_count = len ( requests ) <EOL> if use_monitor : <EOL> lock_monitor_cb = self . _MonitorChecker ( ) <EOL> else : <EOL> lock_monitor_cb = None <EOL> def _ProcessRequests ( multi , handles ) : <EOL> self . assertTrue ( isinstance ( multi , self . _DummyCurlMulti ) ) <EOL> self . assertEqual ( len ( requests ) , len ( handles ) ) <EOL> self . assertTrue ( compat . all ( isinstance ( curl , _FakeCurl ) <EOL> for curl in handles ) ) <EOL> for req in requests : <EOL> assert req . completion_cb is NotImplementedError <EOL> if use_monitor : <EOL> req . completion_cb = compat . partial ( _LockCheckReset , lock_monitor_cb . GetMonitor ( ) ) <EOL> for idx , curl in enumerate ( handles ) : <EOL> try : <EOL> port = curl . opts [ "<STR_LIT>" ] <EOL> except KeyError : <EOL> self . fail ( "<STR_LIT>" ) <EOL> if use_monitor : <EOL> lock_info = lock_monitor_cb . GetMonitor ( ) . GetLockInfo ( None ) <EOL> expected = [ ( "<STR_LIT>" % ( _BuildNiceName ( handle . opts [ "<STR_LIT>" ] , <EOL> default = ( "<STR_LIT>" % <EOL> handle . opts [ "<STR_LIT>" ] ) ) ) , <EOL> None , <EOL> [ threading . currentThread ( ) . getName ( ) ] , None ) <EOL> for handle in handles [ idx : ] ] <EOL> self . assertEqual ( sorted ( lock_info ) , sorted ( expected ) ) <EOL> if port % <NUM_LIT:3> == <NUM_LIT:0> : <EOL> response_code = http . HTTP_OK <EOL> msg = None <EOL> else : <EOL> response_code = http . HttpNotFound . code <EOL> msg = "<STR_LIT>" <EOL> curl . info = { <EOL> pycurl . RESPONSE_CODE : response_code , <EOL> } <EOL> if hasattr ( pycurl , '<STR_LIT>' ) : <EOL> curl . info [ pycurl . LOCAL_IP ] = '<STR_LIT:127.0.0.1>' <EOL> if hasattr ( pycurl , '<STR_LIT>' ) : <EOL> curl . info [ pycurl . LOCAL_PORT ] = port <EOL> self . assertFalse ( curl . opts . pop ( pycurl . POSTFIELDS ) ) <EOL> self . assertTrue ( callable ( curl . opts . pop ( pycurl . WRITEFUNCTION ) ) ) <EOL> yield ( curl , msg ) <EOL> if use_monitor : <EOL> self . assertTrue ( compat . all ( req . lockcheck__ for req in requests ) ) <EOL> if use_monitor : <EOL> self . assertEqual ( lock_monitor_cb . GetMonitor ( ) , None ) <EOL> http . client . ProcessRequests ( requests , lock_monitor_cb = lock_monitor_cb , <EOL> _curl = _FakeCurl , <EOL> _curl_multi = self . _DummyCurlMulti , <EOL> _curl_process = _ProcessRequests ) <EOL> for req in requests : <EOL> if req . port % <NUM_LIT:3> == <NUM_LIT:0> : <EOL> self . assertTrue ( req . success ) <EOL> self . assertEqual ( req . error , None ) <EOL> else : <EOL> self . assertFalse ( req . success ) <EOL> self . assertTrue ( "<STR_LIT>" in req . error ) <EOL> if use_monitor : <EOL> monitor = lock_monitor_cb . GetMonitor ( ) <EOL> self . assertEqual ( monitor . _pending_fn , None ) <EOL> self . assertEqual ( monitor . GetLockInfo ( None ) , [ ] ) <EOL> else : <EOL> self . assertEqual ( lock_monitor_cb , None ) <EOL> self . assertEqual ( len ( requests ) , requests_count ) <EOL> def testBadRequest ( self ) : <EOL> bad_request = http . client . HttpClientRequest ( "<STR_LIT:localhost>" , <NUM_LIT> , <EOL> "<STR_LIT:POST>" , "<STR_LIT>" ) <EOL> bad_request . success = False <EOL> self . assertRaises ( AssertionError , http . client . ProcessRequests , <EOL> [ bad_request ] , _curl = NotImplemented , <EOL> _curl_multi = NotImplemented , _curl_process = NotImplemented ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> testutils . GanetiTestProgram ( ) </s>
