<s> import mock <EOL> from datetime import datetime <EOL> from twisted . internet import reactor , defer <EOL> from jasmin . routing . test . test_router_smpps import SMPPClientTestCases <EOL> from jasmin . routing . test . test_router import SubmitSmTestCaseTools <EOL> from jasmin . routing . proxies import RouterPBProxy <EOL> from jasmin . vendor . smpp . pdu import pdu_types <EOL> from jasmin . routing . Interceptors import DefaultInterceptor <EOL> from jasmin . routing . jasminApi import * <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 . smpp . stats import SMPPServerStatsCollector <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 SMPPClientTestCases . setUp ( self , interceptorpb_client = self . ipb_client ) <EOL> else : <EOL> yield SMPPClientTestCases . setUp ( self ) <EOL> yield self . connect ( '<STR_LIT:127.0.0.1>' , self . pbPort ) <EOL> self . mt_interceptor = MTInterceptorScript ( self . script ) <EOL> yield self . mtinterceptor_add ( DefaultInterceptor ( self . mt_interceptor ) , <NUM_LIT:0> ) <EOL> self . disconnect ( ) <EOL> self . stats_smpps = SMPPServerStatsCollector ( ) . get ( cid = self . smpps_config . id ) <EOL> @ defer . inlineCallbacks <EOL> def tearDown ( self ) : <EOL> yield SMPPClientTestCases . 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 SmppsSubmitSmNoInterceptorPBTestCases ( ProvisionWithoutInterceptorPB , RouterPBProxy , SMPPClientTestCases , SubmitSmTestCaseTools ) : <EOL> @ defer . inlineCallbacks <EOL> def test_interceptorpb_not_set ( self ) : <EOL> _ic = self . stats_smpps . get ( '<STR_LIT>' ) <EOL> _iec = self . stats_smpps . get ( '<STR_LIT>' ) <EOL> yield self . connect ( '<STR_LIT:127.0.0.1>' , self . pbPort ) <EOL> yield self . prepareRoutingsAndStartConnector ( ) <EOL> yield self . smppc_factory . connectAndBind ( ) <EOL> self . smpps_factory . lastProto . sendPDU = mock . Mock ( wraps = self . smpps_factory . lastProto . sendPDU ) <EOL> yield self . smppc_factory . lastProto . sendDataRequest ( self . SubmitSmPDU ) <EOL> yield waitFor ( <NUM_LIT:3> ) <EOL> yield self . smppc_factory . smpp . unbindAndDisconnect ( ) <EOL> yield self . stopSmppClientConnectors ( ) <EOL> self . assertEqual ( <NUM_LIT:0> , len ( self . SMSCPort . factory . lastClient . submitRecords ) ) <EOL> self . assertEqual ( self . smpps_factory . lastProto . sendPDU . call_count , <NUM_LIT:2> ) <EOL> response_pdu = self . smpps_factory . lastProto . sendPDU . call_args_list [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> self . assertEqual ( response_pdu . id , pdu_types . CommandId . submit_sm_resp ) <EOL> self . assertEqual ( response_pdu . seqNum , <NUM_LIT:2> ) <EOL> self . assertEqual ( response_pdu . status , pdu_types . CommandStatus . ESME_RSYSERR ) <EOL> self . assertTrue ( '<STR_LIT>' not in response_pdu . params ) <EOL> self . assertEqual ( _ic , self . stats_smpps . get ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( _iec + <NUM_LIT:1> , self . stats_smpps . get ( '<STR_LIT>' ) ) <EOL> class SmppsSubmitSmInterceptionTestCases ( ProvisionInterceptorPB , RouterPBProxy , SMPPClientTestCases , SubmitSmTestCaseTools ) : <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_interceptorpb_not_connected ( self ) : <EOL> _ic = self . stats_smpps . get ( '<STR_LIT>' ) <EOL> _iec = self . stats_smpps . get ( '<STR_LIT>' ) <EOL> yield self . connect ( '<STR_LIT:127.0.0.1>' , self . pbPort ) <EOL> yield self . prepareRoutingsAndStartConnector ( ) <EOL> yield self . smppc_factory . connectAndBind ( ) <EOL> self . smpps_factory . lastProto . sendPDU = mock . Mock ( wraps = self . smpps_factory . lastProto . sendPDU ) <EOL> yield self . smppc_factory . lastProto . sendDataRequest ( self . SubmitSmPDU ) <EOL> yield waitFor ( <NUM_LIT:3> ) <EOL> yield self . smppc_factory . smpp . unbindAndDisconnect ( ) <EOL> yield self . stopSmppClientConnectors ( ) <EOL> self . assertEqual ( <NUM_LIT:0> , len ( self . SMSCPort . factory . lastClient . submitRecords ) ) <EOL> self . assertEqual ( self . smpps_factory . lastProto . sendPDU . call_count , <NUM_LIT:2> ) <EOL> response_pdu = self . smpps_factory . lastProto . sendPDU . call_args_list [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> self . assertEqual ( response_pdu . id , pdu_types . CommandId . submit_sm_resp ) <EOL> self . assertEqual ( response_pdu . seqNum , <NUM_LIT:2> ) <EOL> self . assertEqual ( response_pdu . status , pdu_types . CommandStatus . ESME_RSYSERR ) <EOL> self . assertTrue ( '<STR_LIT>' not in response_pdu . params ) <EOL> self . assertEqual ( _ic , self . stats_smpps . get ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( _iec + <NUM_LIT:1> , self . stats_smpps . get ( '<STR_LIT>' ) ) <EOL> @ defer . inlineCallbacks <EOL> def test_syntax_error ( self ) : <EOL> _ic = self . stats_smpps . get ( '<STR_LIT>' ) <EOL> _iec = self . stats_smpps . get ( '<STR_LIT>' ) <EOL> yield self . ipb_connect ( ) <EOL> yield self . connect ( '<STR_LIT:127.0.0.1>' , self . pbPort ) <EOL> yield self . prepareRoutingsAndStartConnector ( ) <EOL> yield self . smppc_factory . connectAndBind ( ) <EOL> self . smpps_factory . lastProto . sendPDU = mock . Mock ( wraps = self . smpps_factory . lastProto . sendPDU ) <EOL> yield self . smppc_factory . lastProto . sendDataRequest ( self . SubmitSmPDU ) <EOL> yield waitFor ( <NUM_LIT:3> ) <EOL> yield self . smppc_factory . smpp . unbindAndDisconnect ( ) <EOL> yield self . stopSmppClientConnectors ( ) <EOL> self . assertEqual ( <NUM_LIT:0> , len ( self . SMSCPort . factory . lastClient . submitRecords ) ) <EOL> self . assertEqual ( self . smpps_factory . lastProto . sendPDU . call_count , <NUM_LIT:2> ) <EOL> response_pdu = self . smpps_factory . lastProto . sendPDU . call_args_list [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> self . assertEqual ( response_pdu . id , pdu_types . CommandId . submit_sm_resp ) <EOL> self . assertEqual ( response_pdu . seqNum , <NUM_LIT:2> ) <EOL> self . assertEqual ( response_pdu . status , pdu_types . CommandStatus . ESME_RSYSERR ) <EOL> self . assertTrue ( '<STR_LIT>' not in response_pdu . params ) <EOL> self . assertEqual ( _ic , self . stats_smpps . get ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( _iec + <NUM_LIT:1> , self . stats_smpps . get ( '<STR_LIT>' ) ) <EOL> @ defer . inlineCallbacks <EOL> def test_success ( self ) : <EOL> _ic = self . stats_smpps . get ( '<STR_LIT>' ) <EOL> _iec = self . stats_smpps . get ( '<STR_LIT>' ) <EOL> yield self . connect ( '<STR_LIT:127.0.0.1>' , self . pbPort ) <EOL> mt_interceptor = MTInterceptorScript ( self . update_message_sript ) <EOL> yield self . mtinterceptor_add ( DefaultInterceptor ( mt_interceptor ) , <NUM_LIT:0> ) <EOL> self . disconnect ( ) <EOL> yield self . ipb_connect ( ) <EOL> yield self . connect ( '<STR_LIT:127.0.0.1>' , self . pbPort ) <EOL> yield self . prepareRoutingsAndStartConnector ( ) <EOL> yield self . smppc_factory . connectAndBind ( ) <EOL> self . smpps_factory . lastProto . sendPDU = mock . Mock ( wraps = self . smpps_factory . lastProto . sendPDU ) <EOL> yield self . smppc_factory . lastProto . sendDataRequest ( self . SubmitSmPDU ) <EOL> yield waitFor ( <NUM_LIT:3> ) <EOL> yield self . smppc_factory . smpp . unbindAndDisconnect ( ) <EOL> yield self . stopSmppClientConnectors ( ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( self . SMSCPort . factory . lastClient . submitRecords ) ) <EOL> self . assertEqual ( self . smpps_factory . lastProto . sendPDU . call_count , <NUM_LIT:2> ) <EOL> response_pdu = self . smpps_factory . lastProto . sendPDU . call_args_list [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> self . assertEqual ( response_pdu . id , pdu_types . CommandId . submit_sm_resp ) <EOL> self . assertEqual ( response_pdu . seqNum , <NUM_LIT:2> ) <EOL> self . assertEqual ( response_pdu . status , pdu_types . CommandStatus . ESME_ROK ) <EOL> self . assertNotEqual ( None , response_pdu . params [ '<STR_LIT>' ] ) <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_smpps . get ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( _iec , self . stats_smpps . get ( '<STR_LIT>' ) ) <EOL> @ defer . inlineCallbacks <EOL> def test_any_exception_from_script ( self ) : <EOL> _ic = self . stats_smpps . get ( '<STR_LIT>' ) <EOL> _iec = self . stats_smpps . get ( '<STR_LIT>' ) <EOL> yield self . connect ( '<STR_LIT:127.0.0.1>' , self . pbPort ) <EOL> mt_interceptor = MTInterceptorScript ( self . raise_any_exception ) <EOL> yield self . mtinterceptor_add ( DefaultInterceptor ( mt_interceptor ) , <NUM_LIT:0> ) <EOL> self . disconnect ( ) <EOL> yield self . ipb_connect ( ) <EOL> yield self . connect ( '<STR_LIT:127.0.0.1>' , self . pbPort ) <EOL> yield self . prepareRoutingsAndStartConnector ( ) <EOL> yield self . smppc_factory . connectAndBind ( ) <EOL> self . smpps_factory . lastProto . sendPDU = mock . Mock ( wraps = self . smpps_factory . lastProto . sendPDU ) <EOL> yield self . smppc_factory . lastProto . sendDataRequest ( self . SubmitSmPDU ) <EOL> yield waitFor ( <NUM_LIT:3> ) <EOL> yield self . smppc_factory . smpp . unbindAndDisconnect ( ) <EOL> yield self . stopSmppClientConnectors ( ) <EOL> self . assertEqual ( <NUM_LIT:0> , len ( self . SMSCPort . factory . lastClient . submitRecords ) ) <EOL> self . assertEqual ( self . smpps_factory . lastProto . sendPDU . call_count , <NUM_LIT:2> ) <EOL> response_pdu = self . smpps_factory . lastProto . sendPDU . call_args_list [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> self . assertEqual ( response_pdu . id , pdu_types . CommandId . submit_sm_resp ) <EOL> self . assertEqual ( response_pdu . seqNum , <NUM_LIT:2> ) <EOL> self . assertEqual ( response_pdu . status , pdu_types . CommandStatus . ESME_RSYSERR ) <EOL> self . assertTrue ( '<STR_LIT>' not in response_pdu . params ) <EOL> self . assertEqual ( _ic , self . stats_smpps . get ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( _iec + <NUM_LIT:1> , self . stats_smpps . get ( '<STR_LIT>' ) ) <EOL> @ defer . inlineCallbacks <EOL> def test_ESME_RINVESMCLASS_from_script ( self ) : <EOL> _ic = self . stats_smpps . get ( '<STR_LIT>' ) <EOL> _iec = self . stats_smpps . get ( '<STR_LIT>' ) <EOL> yield self . connect ( '<STR_LIT:127.0.0.1>' , self . pbPort ) <EOL> mt_interceptor = MTInterceptorScript ( self . return_ESME_RINVESMCLASS ) <EOL> yield self . mtinterceptor_add ( DefaultInterceptor ( mt_interceptor ) , <NUM_LIT:0> ) <EOL> self . disconnect ( ) <EOL> yield self . ipb_connect ( ) <EOL> yield self . connect ( '<STR_LIT:127.0.0.1>' , self . pbPort ) <EOL> yield self . prepareRoutingsAndStartConnector ( ) <EOL> yield self . smppc_factory . connectAndBind ( ) <EOL> self . smpps_factory . lastProto . sendPDU = mock . Mock ( wraps = self . smpps_factory . lastProto . sendPDU ) <EOL> yield self . smppc_factory . lastProto . sendDataRequest ( self . SubmitSmPDU ) <EOL> yield waitFor ( <NUM_LIT:3> ) <EOL> yield self . smppc_factory . smpp . unbindAndDisconnect ( ) <EOL> yield self . stopSmppClientConnectors ( ) <EOL> self . assertEqual ( <NUM_LIT:0> , len ( self . SMSCPort . factory . lastClient . submitRecords ) ) <EOL> self . assertEqual ( self . smpps_factory . lastProto . sendPDU . call_count , <NUM_LIT:2> ) <EOL> response_pdu = self . smpps_factory . lastProto . sendPDU . call_args_list [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> self . assertEqual ( response_pdu . id , pdu_types . CommandId . submit_sm_resp ) <EOL> self . assertEqual ( response_pdu . seqNum , <NUM_LIT:2> ) <EOL> self . assertEqual ( response_pdu . status , pdu_types . CommandStatus . ESME_RINVESMCLASS ) <EOL> self . assertTrue ( '<STR_LIT>' not in response_pdu . params ) <EOL> self . assertEqual ( _ic , self . stats_smpps . get ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( _iec + <NUM_LIT:1> , self . stats_smpps . get ( '<STR_LIT>' ) ) <EOL> @ defer . inlineCallbacks <EOL> def test_HTTP_300_from_script ( self ) : <EOL> "<STR_LIT>" <EOL> _ic = self . stats_smpps . get ( '<STR_LIT>' ) <EOL> _iec = self . stats_smpps . get ( '<STR_LIT>' ) <EOL> yield self . connect ( '<STR_LIT:127.0.0.1>' , self . pbPort ) <EOL> mt_interceptor = MTInterceptorScript ( self . return_HTTP_300 ) <EOL> yield self . mtinterceptor_add ( DefaultInterceptor ( mt_interceptor ) , <NUM_LIT:0> ) <EOL> self . disconnect ( ) <EOL> yield self . ipb_connect ( ) <EOL> yield self . connect ( '<STR_LIT:127.0.0.1>' , self . pbPort ) <EOL> yield self . prepareRoutingsAndStartConnector ( ) <EOL> yield self . smppc_factory . connectAndBind ( ) <EOL> self . smpps_factory . lastProto . sendPDU = mock . Mock ( wraps = self . smpps_factory . lastProto . sendPDU ) <EOL> yield self . smppc_factory . lastProto . sendDataRequest ( self . SubmitSmPDU ) <EOL> yield waitFor ( <NUM_LIT:3> ) <EOL> yield self . smppc_factory . smpp . unbindAndDisconnect ( ) <EOL> yield self . stopSmppClientConnectors ( ) <EOL> self . assertEqual ( <NUM_LIT:0> , len ( self . SMSCPort . factory . lastClient . submitRecords ) ) <EOL> self . assertEqual ( self . smpps_factory . lastProto . sendPDU . call_count , <NUM_LIT:2> ) <EOL> response_pdu = self . smpps_factory . lastProto . sendPDU . call_args_list [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> self . assertEqual ( response_pdu . id , pdu_types . CommandId . submit_sm_resp ) <EOL> self . assertEqual ( response_pdu . seqNum , <NUM_LIT:2> ) <EOL> self . assertEqual ( response_pdu . status , pdu_types . CommandStatus . ESME_RUNKNOWNERR ) <EOL> self . assertTrue ( '<STR_LIT>' not in response_pdu . params ) <EOL> self . assertEqual ( _ic , self . stats_smpps . get ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( _iec + <NUM_LIT:1> , self . stats_smpps . get ( '<STR_LIT>' ) ) </s>
