<s> from twisted . internet import reactor , defer <EOL> from twisted . web . client import getPage <EOL> from test_router import HappySMSCTestCase , id_generator <EOL> from jasmin . routing . jasminApi import * <EOL> from jasmin . routing . proxies import RouterPBProxy <EOL> from jasmin . protocols . smpp . configs import SMPPClientConfig <EOL> from jasmin . routing . Routes import DefaultRoute <EOL> from jasmin . routing . Interceptors import DefaultInterceptor <EOL> from jasmin . interceptor . interceptor import InterceptorPB <EOL> from jasmin . interceptor . configs import InterceptorPBConfig , InterceptorPBClientConfig <EOL> from jasmin . interceptor . proxies import InterceptorPBProxy <EOL> from twisted . cred import portal <EOL> from twisted . cred . checkers import AllowAnonymousAccess , InMemoryUsernamePasswordDatabaseDontUse <EOL> from jasmin . tools . cred . portal import JasminPBRealm <EOL> from jasmin . tools . spread . pb import JasminPBPortalRoot <EOL> from twisted . spread import pb <EOL> from jasmin . protocols . http . stats import HttpAPIStatsCollector <EOL> @ defer . inlineCallbacks <EOL> def waitFor ( seconds ) : <EOL> waitDeferred = defer . Deferred ( ) <EOL> reactor . callLater ( seconds , waitDeferred . callback , None ) <EOL> yield waitDeferred <EOL> class ProvisionWithoutInterceptorPB ( object ) : <EOL> script = '<STR_LIT>' <EOL> @ defer . inlineCallbacks <EOL> def setUp ( self ) : <EOL> if hasattr ( self , '<STR_LIT>' ) : <EOL> yield HappySMSCTestCase . setUp ( self , interceptorpb_client = self . ipb_client ) <EOL> else : <EOL> yield HappySMSCTestCase . setUp ( self ) <EOL> yield self . connect ( '<STR_LIT:127.0.0.1>' , self . pbPort ) <EOL> self . g1 = Group ( <NUM_LIT:1> ) <EOL> yield self . group_add ( self . g1 ) <EOL> self . c1 = SmppClientConnector ( id_generator ( ) ) <EOL> self . mt_interceptor = MTInterceptorScript ( self . script ) <EOL> self . u1_password = '<STR_LIT:password>' <EOL> self . u1 = User ( <NUM_LIT:1> , self . g1 , '<STR_LIT:username>' , self . u1_password ) <EOL> self . u2_password = '<STR_LIT:password>' <EOL> self . u2 = User ( <NUM_LIT:1> , self . g1 , '<STR_LIT>' , self . u2_password ) <EOL> yield self . user_add ( self . u1 ) <EOL> yield self . mtroute_add ( DefaultRoute ( self . c1 ) , <NUM_LIT:0> ) <EOL> yield self . mtinterceptor_add ( DefaultInterceptor ( self . mt_interceptor ) , <NUM_LIT:0> ) <EOL> yield self . SMPPClientManagerPBProxy . connect ( '<STR_LIT:127.0.0.1>' , self . CManagerPort ) <EOL> c1Config = SMPPClientConfig ( id = self . c1 . cid , port = self . SMSCPort . getHost ( ) . port ) <EOL> yield self . SMPPClientManagerPBProxy . add ( c1Config ) <EOL> yield self . SMPPClientManagerPBProxy . start ( self . c1 . cid ) <EOL> self . stats_http = HttpAPIStatsCollector ( ) . get ( ) <EOL> @ defer . inlineCallbacks <EOL> def tearDown ( self ) : <EOL> yield self . SMPPClientManagerPBProxy . stop ( self . c1 . cid ) <EOL> while True : <EOL> ssRet = yield self . SMPPClientManagerPBProxy . session_state ( self . c1 . cid ) <EOL> if ssRet == '<STR_LIT>' or ssRet == '<STR_LIT>' : <EOL> break ; <EOL> else : <EOL> yield waitFor ( <NUM_LIT> ) <EOL> yield HappySMSCTestCase . tearDown ( self ) <EOL> class ProvisionInterceptorPB ( ProvisionWithoutInterceptorPB ) : <EOL> @ defer . inlineCallbacks <EOL> def setUp ( self , authentication = False ) : <EOL> "<STR_LIT>" <EOL> self . ipb_client = InterceptorPBProxy ( ) <EOL> yield ProvisionWithoutInterceptorPB . setUp ( self ) <EOL> InterceptorPBConfigInstance = InterceptorPBConfig ( ) <EOL> pbInterceptor_factory = InterceptorPB ( ) <EOL> pbInterceptor_factory . setConfig ( InterceptorPBConfigInstance ) <EOL> p = portal . Portal ( JasminPBRealm ( pbInterceptor_factory ) ) <EOL> if not authentication : <EOL> p . registerChecker ( AllowAnonymousAccess ( ) ) <EOL> else : <EOL> c = InMemoryUsernamePasswordDatabaseDontUse ( ) <EOL> c . addUser ( '<STR_LIT>' , md5 ( '<STR_LIT>' ) . digest ( ) ) <EOL> p . registerChecker ( c ) <EOL> jPBPortalRoot = JasminPBPortalRoot ( p ) <EOL> self . pbInterceptor_server = reactor . listenTCP ( <NUM_LIT:0> , pb . PBServerFactory ( jPBPortalRoot ) ) <EOL> self . pbInterceptor_port = self . pbInterceptor_server . getHost ( ) . port <EOL> @ defer . inlineCallbacks <EOL> def ipb_connect ( self , config = None ) : <EOL> if config is None : <EOL> config = InterceptorPBClientConfig ( ) <EOL> config . username = None <EOL> config . port = self . pbInterceptor_port <EOL> if config . username is not None : <EOL> yield self . ipb_client . connect ( <EOL> config . host , <EOL> config . port , <EOL> config . username , <EOL> config . password <EOL> ) <EOL> else : <EOL> yield self . ipb_client . connect ( <EOL> config . host , <EOL> config . port <EOL> ) <EOL> @ defer . inlineCallbacks <EOL> def tearDown ( self ) : <EOL> yield ProvisionWithoutInterceptorPB . tearDown ( self ) <EOL> if self . ipb_client . isConnected : <EOL> self . ipb_client . disconnect ( ) <EOL> yield self . pbInterceptor_server . stopListening ( ) <EOL> class HttpAPISubmitSmNoInterceptorPBTestCases ( ProvisionWithoutInterceptorPB , RouterPBProxy , HappySMSCTestCase ) : <EOL> @ defer . inlineCallbacks <EOL> def test_send_interceptorpb_not_set ( self ) : <EOL> _ic = self . stats_http . get ( '<STR_LIT>' ) <EOL> _iec = self . stats_http . get ( '<STR_LIT>' ) <EOL> url = '<STR_LIT>' % ( <EOL> self . u1 . username , self . u1_password ) <EOL> lastErrorStatus = None <EOL> lastResponse = None <EOL> try : <EOL> yield getPage ( url ) <EOL> except Exception , e : <EOL> lastErrorStatus = e . status <EOL> lastResponse = e . response <EOL> self . assertEqual ( lastErrorStatus , '<STR_LIT>' ) <EOL> self . assertEqual ( lastResponse , '<STR_LIT>' ) <EOL> self . assertEqual ( _ic , self . stats_http . get ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( _iec + <NUM_LIT:1> , self . stats_http . get ( '<STR_LIT>' ) ) <EOL> @ defer . inlineCallbacks <EOL> def test_rate_interceptorpb_not_set ( self ) : <EOL> _ic = self . stats_http . get ( '<STR_LIT>' ) <EOL> _iec = self . stats_http . get ( '<STR_LIT>' ) <EOL> url = '<STR_LIT>' % ( <EOL> self . u1 . username , self . u1_password ) <EOL> lastErrorStatus = None <EOL> lastResponse = None <EOL> try : <EOL> yield getPage ( url ) <EOL> except Exception , e : <EOL> lastErrorStatus = e . status <EOL> lastResponse = e . response <EOL> self . assertEqual ( lastErrorStatus , '<STR_LIT>' ) <EOL> self . assertEqual ( lastResponse , '<STR_LIT>' ) <EOL> self . assertEqual ( _ic , self . stats_http . get ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( _iec + <NUM_LIT:1> , self . stats_http . get ( '<STR_LIT>' ) ) <EOL> class HttpAPISubmitSmInterceptionTestCases ( ProvisionInterceptorPB , RouterPBProxy , HappySMSCTestCase ) : <EOL> update_message_sript = "<STR_LIT>" <EOL> raise_any_exception = "<STR_LIT>" <EOL> return_ESME_RINVESMCLASS = "<STR_LIT>" <EOL> return_HTTP_300 = "<STR_LIT>" <EOL> @ defer . inlineCallbacks <EOL> def test_send_interceptorpb_not_connected ( self ) : <EOL> _ic = self . stats_http . get ( '<STR_LIT>' ) <EOL> _iec = self . stats_http . get ( '<STR_LIT>' ) <EOL> url = '<STR_LIT>' % ( <EOL> self . u1 . username , self . u1_password ) <EOL> lastErrorStatus = None <EOL> lastResponse = None <EOL> try : <EOL> yield getPage ( url ) <EOL> except Exception , e : <EOL> lastErrorStatus = e . status <EOL> lastResponse = e . response <EOL> self . assertEqual ( lastErrorStatus , '<STR_LIT>' ) <EOL> self . assertEqual ( lastResponse , '<STR_LIT>' ) <EOL> self . assertEqual ( _ic , self . stats_http . get ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( _iec + <NUM_LIT:1> , self . stats_http . get ( '<STR_LIT>' ) ) <EOL> @ defer . inlineCallbacks <EOL> def test_send_syntax_error ( self ) : <EOL> _ic = self . stats_http . get ( '<STR_LIT>' ) <EOL> _iec = self . stats_http . get ( '<STR_LIT>' ) <EOL> yield self . ipb_connect ( ) <EOL> url = '<STR_LIT>' % ( <EOL> self . u1 . username , self . u1_password ) <EOL> lastErrorStatus = None <EOL> lastResponse = None <EOL> try : <EOL> yield getPage ( url ) <EOL> except Exception , e : <EOL> lastErrorStatus = e . status <EOL> lastResponse = e . response <EOL> self . assertEqual ( lastErrorStatus , '<STR_LIT>' ) <EOL> self . assertEqual ( lastResponse , '<STR_LIT>' ) <EOL> self . assertEqual ( _ic , self . stats_http . get ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( _iec + <NUM_LIT:1> , self . stats_http . get ( '<STR_LIT>' ) ) <EOL> @ defer . inlineCallbacks <EOL> def test_send_success ( self ) : <EOL> _ic = self . stats_http . get ( '<STR_LIT>' ) <EOL> _iec = self . stats_http . get ( '<STR_LIT>' ) <EOL> mt_interceptor = MTInterceptorScript ( self . update_message_sript ) <EOL> yield self . mtinterceptor_add ( DefaultInterceptor ( mt_interceptor ) , <NUM_LIT:0> ) <EOL> yield self . ipb_connect ( ) <EOL> url = '<STR_LIT>' % ( <EOL> self . u1 . username , self . u1_password ) <EOL> lastErrorStatus = None <EOL> lastResponse = None <EOL> try : <EOL> yield getPage ( url ) <EOL> except Exception , e : <EOL> lastErrorStatus = e . status <EOL> lastResponse = e . response <EOL> yield waitFor ( <NUM_LIT:2> ) <EOL> self . assertEqual ( lastErrorStatus , None ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( self . SMSCPort . factory . lastClient . submitRecords ) ) <EOL> self . assertEqual ( '<STR_LIT>' , self . SMSCPort . factory . lastClient . submitRecords [ <NUM_LIT:0> ] . params [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( _ic + <NUM_LIT:1> , self . stats_http . get ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( _iec , self . stats_http . get ( '<STR_LIT>' ) ) <EOL> @ defer . inlineCallbacks <EOL> def test_send_any_exception_from_script ( self ) : <EOL> _ic = self . stats_http . get ( '<STR_LIT>' ) <EOL> _iec = self . stats_http . get ( '<STR_LIT>' ) <EOL> mt_interceptor = MTInterceptorScript ( self . raise_any_exception ) <EOL> yield self . mtinterceptor_add ( DefaultInterceptor ( mt_interceptor ) , <NUM_LIT:0> ) <EOL> yield self . ipb_connect ( ) <EOL> url = '<STR_LIT>' % ( <EOL> self . u1 . username , self . u1_password ) <EOL> lastErrorStatus = None <EOL> lastResponse = None <EOL> try : <EOL> yield getPage ( url ) <EOL> except Exception , e : <EOL> lastErrorStatus = e . status <EOL> lastResponse = e . response <EOL> yield waitFor ( <NUM_LIT:2> ) <EOL> self . assertEqual ( lastErrorStatus , '<STR_LIT>' ) <EOL> self . assertEqual ( lastResponse , '<STR_LIT>' ) <EOL> self . assertEqual ( _ic , self . stats_http . get ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( _iec + <NUM_LIT:1> , self . stats_http . get ( '<STR_LIT>' ) ) <EOL> @ defer . inlineCallbacks <EOL> def test_send_HTTP_300_from_script ( self ) : <EOL> _ic = self . stats_http . get ( '<STR_LIT>' ) <EOL> _iec = self . stats_http . get ( '<STR_LIT>' ) <EOL> mt_interceptor = MTInterceptorScript ( self . return_HTTP_300 ) <EOL> yield self . mtinterceptor_add ( DefaultInterceptor ( mt_interceptor ) , <NUM_LIT:0> ) <EOL> yield self . ipb_connect ( ) <EOL> url = '<STR_LIT>' % ( <EOL> self . u1 . username , self . u1_password ) <EOL> lastErrorStatus = None <EOL> lastResponse = None <EOL> try : <EOL> yield getPage ( url ) <EOL> except Exception , e : <EOL> lastErrorStatus = e . status <EOL> lastResponse = e . response <EOL> yield waitFor ( <NUM_LIT:2> ) <EOL> self . assertEqual ( lastErrorStatus , '<STR_LIT>' ) <EOL> self . assertEqual ( lastResponse , '<STR_LIT>' ) <EOL> self . assertEqual ( _ic , self . stats_http . get ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( _iec + <NUM_LIT:1> , self . stats_http . get ( '<STR_LIT>' ) ) <EOL> @ defer . inlineCallbacks <EOL> def test_send_ESME_RINVESMCLASS_from_script ( self ) : <EOL> "<STR_LIT>" <EOL> _ic = self . stats_http . get ( '<STR_LIT>' ) <EOL> _iec = self . stats_http . get ( '<STR_LIT>' ) <EOL> mt_interceptor = MTInterceptorScript ( self . return_ESME_RINVESMCLASS ) <EOL> yield self . mtinterceptor_add ( DefaultInterceptor ( mt_interceptor ) , <NUM_LIT:0> ) <EOL> yield self . ipb_connect ( ) <EOL> url = '<STR_LIT>' % ( <EOL> self . u1 . username , self . u1_password ) <EOL> lastErrorStatus = None <EOL> lastResponse = None <EOL> try : <EOL> yield getPage ( url ) <EOL> except Exception , e : <EOL> lastErrorStatus = e . status <EOL> lastResponse = e . response <EOL> yield waitFor ( <NUM_LIT:2> ) <EOL> self . assertEqual ( lastErrorStatus , '<STR_LIT>' ) <EOL> self . assertEqual ( lastResponse , '<STR_LIT>' ) <EOL> self . assertEqual ( _ic , self . stats_http . get ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( _iec + <NUM_LIT:1> , self . stats_http . get ( '<STR_LIT>' ) ) <EOL> @ defer . inlineCallbacks <EOL> def test_rate_interceptorpb_not_connected ( self ) : <EOL> _ic = self . stats_http . get ( '<STR_LIT>' ) <EOL> _iec = self . stats_http . get ( '<STR_LIT>' ) <EOL> url = '<STR_LIT>' % ( <EOL> self . u1 . username , self . u1_password ) <EOL> lastErrorStatus = None <EOL> lastResponse = None <EOL> try : <EOL> yield getPage ( url ) <EOL> except Exception , e : <EOL> lastErrorStatus = e . status <EOL> lastResponse = e . response <EOL> self . assertEqual ( lastErrorStatus , '<STR_LIT>' ) <EOL> self . assertEqual ( lastResponse , '<STR_LIT>' ) <EOL> self . assertEqual ( _ic , self . stats_http . get ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( _iec + <NUM_LIT:1> , self . stats_http . get ( '<STR_LIT>' ) ) <EOL> @ defer . inlineCallbacks <EOL> def test_rate_syntax_error ( self ) : <EOL> _ic = self . stats_http . get ( '<STR_LIT>' ) <EOL> _iec = self . stats_http . get ( '<STR_LIT>' ) <EOL> yield self . ipb_connect ( ) <EOL> url = '<STR_LIT>' % ( <EOL> self . u1 . username , self . u1_password ) <EOL> lastErrorStatus = None <EOL> lastResponse = None <EOL> try : <EOL> yield getPage ( url ) <EOL> except Exception , e : <EOL> lastErrorStatus = e . status <EOL> lastResponse = e . response <EOL> self . assertEqual ( lastErrorStatus , '<STR_LIT>' ) <EOL> self . assertEqual ( lastResponse , '<STR_LIT>' ) <EOL> self . assertEqual ( _ic , self . stats_http . get ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( _iec + <NUM_LIT:1> , self . stats_http . get ( '<STR_LIT>' ) ) <EOL> @ defer . inlineCallbacks <EOL> def test_rate_success ( self ) : <EOL> _ic = self . stats_http . get ( '<STR_LIT>' ) <EOL> _iec = self . stats_http . get ( '<STR_LIT>' ) <EOL> mt_interceptor = MTInterceptorScript ( self . update_message_sript ) <EOL> yield self . mtinterceptor_add ( DefaultInterceptor ( mt_interceptor ) , <NUM_LIT:0> ) <EOL> yield self . ipb_connect ( ) <EOL> url = '<STR_LIT>' % ( <EOL> self . u1 . username , self . u1_password ) <EOL> lastErrorStatus = None <EOL> lastResponse = None <EOL> try : <EOL> yield getPage ( url ) <EOL> except Exception , e : <EOL> lastErrorStatus = e . status <EOL> lastResponse = e . response <EOL> self . assertEqual ( lastErrorStatus , None ) <EOL> self . assertEqual ( _ic + <NUM_LIT:1> , self . stats_http . get ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( _iec , self . stats_http . get ( '<STR_LIT>' ) ) <EOL> @ defer . inlineCallbacks <EOL> def test_rate_any_exception_from_script ( self ) : <EOL> _ic = self . stats_http . get ( '<STR_LIT>' ) <EOL> _iec = self . stats_http . get ( '<STR_LIT>' ) <EOL> mt_interceptor = MTInterceptorScript ( self . raise_any_exception ) <EOL> yield self . mtinterceptor_add ( DefaultInterceptor ( mt_interceptor ) , <NUM_LIT:0> ) <EOL> yield self . ipb_connect ( ) <EOL> url = '<STR_LIT>' % ( <EOL> self . u1 . username , self . u1_password ) <EOL> lastErrorStatus = None <EOL> lastResponse = None <EOL> try : <EOL> yield getPage ( url ) <EOL> except Exception , e : <EOL> lastErrorStatus = e . status <EOL> lastResponse = e . response <EOL> yield waitFor ( <NUM_LIT:2> ) <EOL> self . assertEqual ( lastErrorStatus , '<STR_LIT>' ) <EOL> self . assertEqual ( lastResponse , '<STR_LIT>' ) <EOL> self . assertEqual ( _ic , self . stats_http . get ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( _iec + <NUM_LIT:1> , self . stats_http . get ( '<STR_LIT>' ) ) <EOL> @ defer . inlineCallbacks <EOL> def test_rate_HTTP_300_from_script ( self ) : <EOL> _ic = self . stats_http . get ( '<STR_LIT>' ) <EOL> _iec = self . stats_http . get ( '<STR_LIT>' ) <EOL> mt_interceptor = MTInterceptorScript ( self . return_HTTP_300 ) <EOL> yield self . mtinterceptor_add ( DefaultInterceptor ( mt_interceptor ) , <NUM_LIT:0> ) <EOL> yield self . ipb_connect ( ) <EOL> url = '<STR_LIT>' % ( <EOL> self . u1 . username , self . u1_password ) <EOL> lastErrorStatus = None <EOL> lastResponse = None <EOL> try : <EOL> yield getPage ( url ) <EOL> except Exception , e : <EOL> lastErrorStatus = e . status <EOL> lastResponse = e . response <EOL> yield waitFor ( <NUM_LIT:2> ) <EOL> self . assertEqual ( lastErrorStatus , '<STR_LIT>' ) <EOL> self . assertEqual ( lastResponse , '<STR_LIT>' ) <EOL> self . assertEqual ( _ic , self . stats_http . get ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( _iec + <NUM_LIT:1> , self . stats_http . get ( '<STR_LIT>' ) ) <EOL> @ defer . inlineCallbacks <EOL> def test_rate_ESME_RINVESMCLASS_from_script ( self ) : <EOL> "<STR_LIT>" <EOL> _ic = self . stats_http . get ( '<STR_LIT>' ) <EOL> _iec = self . stats_http . get ( '<STR_LIT>' ) <EOL> mt_interceptor = MTInterceptorScript ( self . return_ESME_RINVESMCLASS ) <EOL> yield self . mtinterceptor_add ( DefaultInterceptor ( mt_interceptor ) , <NUM_LIT:0> ) <EOL> yield self . ipb_connect ( ) <EOL> url = '<STR_LIT>' % ( <EOL> self . u1 . username , self . u1_password ) <EOL> lastErrorStatus = None <EOL> lastResponse = None <EOL> try : <EOL> yield getPage ( url ) <EOL> except Exception , e : <EOL> lastErrorStatus = e . status <EOL> lastResponse = e . response <EOL> yield waitFor ( <NUM_LIT:2> ) <EOL> self . assertEqual ( lastErrorStatus , '<STR_LIT>' ) <EOL> self . assertEqual ( lastResponse , '<STR_LIT>' ) <EOL> self . assertEqual ( _ic , self . stats_http . get ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( _iec + <NUM_LIT:1> , self . stats_http . get ( '<STR_LIT>' ) ) </s>
