<s> import unittest <EOL> import logging <EOL> from twisted . web import resource <EOL> from twisted . internet import defer , reactor , task <EOL> from twisted . application import internet <EOL> from twisted . trial . unittest import TestCase <EOL> from twisted . cred . portal import IRealm <EOL> from twisted . cred . checkers import InMemoryUsernamePasswordDatabaseDontUse <EOL> from twisted . cred . portal import Portal <EOL> from twisted . test import proto_helpers <EOL> from zope . interface import implements <EOL> from jasmin . vendor . smpp . twisted . config import SMPPServerConfig , SMPPClientConfig <EOL> from jasmin . vendor . smpp . twisted . server import SMPPServerFactory , SMPPBindManager <EOL> from jasmin . vendor . smpp . twisted . protocol import SMPPSessionStates , DataHandlerResponse <EOL> from jasmin . vendor . smpp . twisted . client import SMPPClientTransceiver <EOL> from jasmin . vendor . smpp . pdu import pdu_types , operations , pdu_encoding <EOL> import mock <EOL> logging . basicConfig ( level = logging . DEBUG ) <EOL> def _makeMockServerConnection ( key , bind_type ) : <EOL> mk_server_cnxn = mock . Mock ( '<STR_LIT>' ) <EOL> mk_server_cnxn . system_id = key <EOL> mk_server_cnxn . bind_type = bind_type <EOL> return mk_server_cnxn <EOL> class SMPPServerFactoryTests ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . config = mock . Mock ( '<STR_LIT>' ) <EOL> self . config . systems = { '<STR_LIT>' : { '<STR_LIT>' : <NUM_LIT:2> } } <EOL> def tearDown ( self ) : <EOL> pass <EOL> @ unittest . skip ( '''<STR_LIT>''' ) <EOL> def test_canOpenNewConnection_transceiver ( self ) : <EOL> server = SMPPServerFactory ( self . config , None ) <EOL> can_bind = server . canOpenNewConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_transceiver ) <EOL> self . assertTrue ( can_bind , '<STR_LIT>' ) <EOL> can_bind = server . canOpenNewConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_transmitter ) <EOL> self . assertTrue ( can_bind , '<STR_LIT>' ) <EOL> can_bind = server . canOpenNewConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_receiver ) <EOL> self . assertTrue ( can_bind , '<STR_LIT>' ) <EOL> mk_server_cnxn = _makeMockServerConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_transceiver ) <EOL> server . addBoundConnection ( mk_server_cnxn ) <EOL> can_bind = server . canOpenNewConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_transceiver ) <EOL> self . assertTrue ( can_bind , '<STR_LIT>' ) <EOL> can_bind = server . canOpenNewConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_transmitter ) <EOL> self . assertTrue ( can_bind , '<STR_LIT>' ) <EOL> can_bind = server . canOpenNewConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_receiver ) <EOL> self . assertTrue ( can_bind , '<STR_LIT>' ) <EOL> mk_server_cnxn = _makeMockServerConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_transceiver ) <EOL> server . addBoundConnection ( mk_server_cnxn ) <EOL> can_bind = server . canOpenNewConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_transceiver ) <EOL> self . assertFalse ( can_bind , '<STR_LIT>' ) <EOL> can_bind = server . canOpenNewConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_transmitter ) <EOL> self . assertFalse ( can_bind , '<STR_LIT>' ) <EOL> can_bind = server . canOpenNewConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_receiver ) <EOL> self . assertFalse ( can_bind , '<STR_LIT>' ) <EOL> @ unittest . skip ( '''<STR_LIT>''' ) <EOL> def test_canOpenNewConnection_transmitter ( self ) : <EOL> server = SMPPServerFactory ( self . config , None ) <EOL> can_bind = server . canOpenNewConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_transceiver ) <EOL> self . assertTrue ( can_bind , '<STR_LIT>' ) <EOL> can_bind = server . canOpenNewConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_transmitter ) <EOL> self . assertTrue ( can_bind , '<STR_LIT>' ) <EOL> can_bind = server . canOpenNewConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_receiver ) <EOL> self . assertTrue ( can_bind , '<STR_LIT>' ) <EOL> mk_server_cnxn = _makeMockServerConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_transmitter ) <EOL> server . addBoundConnection ( mk_server_cnxn ) <EOL> can_bind = server . canOpenNewConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_transceiver ) <EOL> self . assertTrue ( can_bind , '<STR_LIT>' ) <EOL> can_bind = server . canOpenNewConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_transmitter ) <EOL> self . assertTrue ( can_bind , '<STR_LIT>' ) <EOL> can_bind = server . canOpenNewConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_receiver ) <EOL> self . assertTrue ( can_bind , '<STR_LIT>' ) <EOL> mk_server_cnxn = _makeMockServerConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_transmitter ) <EOL> server . addBoundConnection ( mk_server_cnxn ) <EOL> can_bind = server . canOpenNewConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_transceiver ) <EOL> self . assertFalse ( can_bind , '<STR_LIT>' ) <EOL> can_bind = server . canOpenNewConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_transmitter ) <EOL> self . assertFalse ( can_bind , '<STR_LIT>' ) <EOL> can_bind = server . canOpenNewConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_receiver ) <EOL> self . assertTrue ( can_bind , '<STR_LIT>' ) <EOL> @ unittest . skip ( '''<STR_LIT>''' ) <EOL> def test_canOpenNewConnection_receiver ( self ) : <EOL> server = SMPPServerFactory ( self . config , None ) <EOL> can_bind = server . canOpenNewConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_transceiver ) <EOL> self . assertTrue ( can_bind , '<STR_LIT>' ) <EOL> can_bind = server . canOpenNewConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_transmitter ) <EOL> self . assertTrue ( can_bind , '<STR_LIT>' ) <EOL> can_bind = server . canOpenNewConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_receiver ) <EOL> self . assertTrue ( can_bind , '<STR_LIT>' ) <EOL> mk_server_cnxn = _makeMockServerConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_receiver ) <EOL> server . addBoundConnection ( mk_server_cnxn ) <EOL> can_bind = server . canOpenNewConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_transceiver ) <EOL> self . assertTrue ( can_bind , '<STR_LIT>' ) <EOL> can_bind = server . canOpenNewConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_transmitter ) <EOL> self . assertTrue ( can_bind , '<STR_LIT>' ) <EOL> can_bind = server . canOpenNewConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_receiver ) <EOL> self . assertTrue ( can_bind , '<STR_LIT>' ) <EOL> mk_server_cnxn = _makeMockServerConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_receiver ) <EOL> server . addBoundConnection ( mk_server_cnxn ) <EOL> can_bind = server . canOpenNewConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_transceiver ) <EOL> self . assertFalse ( can_bind , '<STR_LIT>' ) <EOL> can_bind = server . canOpenNewConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_transmitter ) <EOL> self . assertTrue ( can_bind , '<STR_LIT>' ) <EOL> can_bind = server . canOpenNewConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_receiver ) <EOL> self . assertFalse ( can_bind , '<STR_LIT>' ) <EOL> @ unittest . skip ( '''<STR_LIT>''' ) <EOL> def test_canOpenNewConnection_multitypes ( self ) : <EOL> server = SMPPServerFactory ( self . config , None ) <EOL> can_bind = server . canOpenNewConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_transceiver ) <EOL> self . assertTrue ( can_bind , '<STR_LIT>' ) <EOL> can_bind = server . canOpenNewConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_transmitter ) <EOL> self . assertTrue ( can_bind , '<STR_LIT>' ) <EOL> can_bind = server . canOpenNewConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_receiver ) <EOL> self . assertTrue ( can_bind , '<STR_LIT>' ) <EOL> mk_server_cnxn = _makeMockServerConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_transceiver ) <EOL> server . addBoundConnection ( mk_server_cnxn ) <EOL> can_bind = server . canOpenNewConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_transceiver ) <EOL> self . assertTrue ( can_bind , '<STR_LIT>' ) <EOL> can_bind = server . canOpenNewConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_transmitter ) <EOL> self . assertTrue ( can_bind , '<STR_LIT>' ) <EOL> can_bind = server . canOpenNewConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_receiver ) <EOL> self . assertTrue ( can_bind , '<STR_LIT>' ) <EOL> mk_server_cnxn = _makeMockServerConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_transmitter ) <EOL> server . addBoundConnection ( mk_server_cnxn ) <EOL> can_bind = server . canOpenNewConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_transceiver ) <EOL> self . assertFalse ( can_bind , '<STR_LIT>' ) <EOL> can_bind = server . canOpenNewConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_transmitter ) <EOL> self . assertFalse ( can_bind , '<STR_LIT>' ) <EOL> can_bind = server . canOpenNewConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_receiver ) <EOL> self . assertTrue ( can_bind , '<STR_LIT>' ) <EOL> mk_server_cnxn = _makeMockServerConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_receiver ) <EOL> server . addBoundConnection ( mk_server_cnxn ) <EOL> can_bind = server . canOpenNewConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_transceiver ) <EOL> self . assertFalse ( can_bind , '<STR_LIT>' ) <EOL> can_bind = server . canOpenNewConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_transmitter ) <EOL> self . assertFalse ( can_bind , '<STR_LIT>' ) <EOL> can_bind = server . canOpenNewConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_receiver ) <EOL> self . assertFalse ( can_bind , '<STR_LIT>' ) <EOL> class SMPPBindManagerTests ( unittest . TestCase ) : <EOL> @ unittest . skip ( '''<STR_LIT>''' ) <EOL> def test_getNextBindingForDelivery ( self ) : <EOL> bm = SMPPBindManager ( '<STR_LIT>' ) <EOL> mk_server_cnxn_rx1 = _makeMockServerConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_receiver ) <EOL> bm . addBinding ( mk_server_cnxn_rx1 ) <EOL> deliverer = bm . getNextBindingForDelivery ( ) <EOL> self . assertEqual ( mk_server_cnxn_rx1 , deliverer ) <EOL> deliverer = bm . getNextBindingForDelivery ( ) <EOL> self . assertEqual ( mk_server_cnxn_rx1 , deliverer ) <EOL> mk_server_cnxn_rx2 = _makeMockServerConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_receiver ) <EOL> bm . addBinding ( mk_server_cnxn_rx2 ) <EOL> deliverer = bm . getNextBindingForDelivery ( ) <EOL> self . assertEqual ( mk_server_cnxn_rx2 , deliverer ) <EOL> deliverer = bm . getNextBindingForDelivery ( ) <EOL> self . assertEqual ( mk_server_cnxn_rx1 , deliverer ) <EOL> mk_server_cnxn_tx1 = _makeMockServerConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_transmitter ) <EOL> bm . addBinding ( mk_server_cnxn_tx1 ) <EOL> deliverer = bm . getNextBindingForDelivery ( ) <EOL> self . assertEqual ( mk_server_cnxn_rx2 , deliverer ) <EOL> mk_server_cnxn_trx1 = _makeMockServerConnection ( '<STR_LIT>' , pdu_types . CommandId . bind_transceiver ) <EOL> bm . addBinding ( mk_server_cnxn_trx1 ) <EOL> deliverer = bm . getNextBindingForDelivery ( ) <EOL> self . assertEqual ( mk_server_cnxn_trx1 , deliverer ) <EOL> bm . removeBinding ( mk_server_cnxn_rx1 ) <EOL> deliverer = bm . getNextBindingForDelivery ( ) <EOL> self . assertEqual ( mk_server_cnxn_rx2 , deliverer ) <EOL> deliverer = bm . getNextBindingForDelivery ( ) <EOL> self . assertEqual ( mk_server_cnxn_trx1 , deliverer ) <EOL> deliverer = bm . getNextBindingForDelivery ( ) <EOL> self . assertEqual ( mk_server_cnxn_rx2 , deliverer ) <EOL> class SMPPServerBaseTest ( TestCase ) : <EOL> def _serviceHandler ( self , system_id , smpp , pdu ) : <EOL> self . service_calls . append ( ( system_id , smpp , pdu ) ) <EOL> return pdu_types . CommandStatus . ESME_ROK <EOL> class SmppRealm ( object ) : <EOL> implements ( IRealm ) <EOL> def requestAvatar ( self , avatarId , mind , * interfaces ) : <EOL> return ( '<STR_LIT>' , avatarId , lambda : None ) <EOL> def _bind ( self ) : <EOL> self . proto . bind_type = pdu_types . CommandId . bind_transceiver <EOL> self . proto . sessionState = SMPPSessionStates . BOUND_TRX <EOL> self . proto . system_id = '<STR_LIT>' <EOL> self . factory . addBoundConnection ( self . proto ) <EOL> class SMPPServerTestCase ( SMPPServerBaseTest ) : <EOL> def setUp ( self ) : <EOL> self . service_calls = [ ] <EOL> self . encoder = pdu_encoding . PDUEncoder ( ) <EOL> self . smpp_config = SMPPServerConfig ( msgHandler = self . _serviceHandler , <EOL> systems = { '<STR_LIT>' : { "<STR_LIT>" : <NUM_LIT:2> } } <EOL> ) <EOL> portal = Portal ( self . SmppRealm ( ) ) <EOL> credential_checker = InMemoryUsernamePasswordDatabaseDontUse ( ) <EOL> credential_checker . addUser ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> portal . registerChecker ( credential_checker ) <EOL> self . factory = SMPPServerFactory ( self . smpp_config , auth_portal = portal ) <EOL> self . proto = self . factory . buildProtocol ( ( '<STR_LIT:127.0.0.1>' , <NUM_LIT:0> ) ) <EOL> self . tr = proto_helpers . StringTransport ( ) <EOL> self . proto . makeConnection ( self . tr ) <EOL> def tearDown ( self ) : <EOL> self . proto . connectionLost ( '<STR_LIT>' ) <EOL> @ unittest . skip ( '''<STR_LIT>''' ) <EOL> def testTRXBindRequest ( self ) : <EOL> pdu = operations . BindTransceiver ( <EOL> system_id = '<STR_LIT>' , <EOL> password = '<STR_LIT>' , <EOL> seqNum = <NUM_LIT:1> <EOL> ) <EOL> self . proto . dataReceived ( self . encoder . encode ( pdu ) ) <EOL> expected_pdu = operations . BindTransceiverResp ( system_id = '<STR_LIT>' , seqNum = <NUM_LIT:1> ) <EOL> self . assertEqual ( self . tr . value ( ) , self . encoder . encode ( expected_pdu ) ) <EOL> connection = self . factory . getBoundConnections ( '<STR_LIT>' ) <EOL> self . assertEqual ( connection . system_id , '<STR_LIT>' ) <EOL> self . assertEqual ( connection . _binds [ pdu_types . CommandId . bind_transceiver ] [ <NUM_LIT:0> ] , self . proto ) <EOL> @ unittest . skip ( '''<STR_LIT>''' ) <EOL> def testTRXBindRequestInvalidSysId ( self ) : <EOL> pdu = operations . BindTransceiver ( <EOL> system_id = '<STR_LIT>' , <EOL> password = '<STR_LIT>' , <EOL> seqNum = <NUM_LIT:1> <EOL> ) <EOL> self . proto . dataReceived ( self . encoder . encode ( pdu ) ) <EOL> expected_pdu = operations . BindTransceiverResp ( system_id = '<STR_LIT>' , seqNum = <NUM_LIT:1> , status = pdu_types . CommandStatus . ESME_RINVSYSID ) <EOL> self . assertEqual ( self . tr . value ( ) , self . encoder . encode ( expected_pdu ) ) <EOL> connection = self . factory . getBoundConnections ( '<STR_LIT>' ) <EOL> self . assertEqual ( connection , None ) <EOL> connection = self . factory . getBoundConnections ( '<STR_LIT>' ) <EOL> self . assertEqual ( connection , None ) <EOL> @ unittest . skip ( '''<STR_LIT>''' ) <EOL> def testTRXBindRequestInvalidPassword ( self ) : <EOL> pdu = operations . BindTransceiver ( <EOL> system_id = '<STR_LIT>' , <EOL> password = '<STR_LIT>' , <EOL> seqNum = <NUM_LIT:1> <EOL> ) <EOL> self . proto . dataReceived ( self . encoder . encode ( pdu ) ) <EOL> expected_pdu = operations . BindTransceiverResp ( system_id = '<STR_LIT>' , seqNum = <NUM_LIT:1> , status = pdu_types . CommandStatus . ESME_RINVPASWD ) <EOL> self . assertEqual ( self . tr . value ( ) , self . encoder . encode ( expected_pdu ) ) <EOL> connection = self . factory . getBoundConnections ( '<STR_LIT>' ) <EOL> self . assertEqual ( connection , None ) <EOL> @ unittest . skip ( '''<STR_LIT>''' ) <EOL> def testTransmitterBindRequest ( self ) : <EOL> system_id = '<STR_LIT>' <EOL> pdu = operations . BindTransmitter ( <EOL> system_id = system_id , <EOL> password = '<STR_LIT>' , <EOL> seqNum = <NUM_LIT:1> <EOL> ) <EOL> self . proto . dataReceived ( self . encoder . encode ( pdu ) ) <EOL> expected_pdu = operations . BindTransmitterResp ( system_id = '<STR_LIT>' , seqNum = <NUM_LIT:1> ) <EOL> self . assertEqual ( self . tr . value ( ) , self . encoder . encode ( expected_pdu ) ) <EOL> self . tr . clear ( ) <EOL> connection = self . factory . getBoundConnections ( system_id ) <EOL> self . assertEqual ( connection . system_id , system_id ) <EOL> self . assertEqual ( connection . _binds [ pdu_types . CommandId . bind_transmitter ] [ <NUM_LIT:0> ] , self . proto ) <EOL> bind_manager = self . factory . getBoundConnections ( system_id ) <EOL> delivery_binding = bind_manager . getNextBindingForDelivery ( ) <EOL> self . assertTrue ( delivery_binding is None ) <EOL> @ unittest . skip ( '''<STR_LIT>''' ) <EOL> def testReceiverBindRequest ( self ) : <EOL> system_id = '<STR_LIT>' <EOL> pdu = operations . BindReceiver ( <EOL> system_id = system_id , <EOL> password = '<STR_LIT>' , <EOL> seqNum = <NUM_LIT:1> <EOL> ) <EOL> self . proto . dataReceived ( self . encoder . encode ( pdu ) ) <EOL> expected_pdu = operations . BindReceiverResp ( system_id = '<STR_LIT>' , seqNum = <NUM_LIT:1> ) <EOL> self . assertEqual ( self . tr . value ( ) , self . encoder . encode ( expected_pdu ) ) <EOL> self . tr . clear ( ) <EOL> connection = self . factory . getBoundConnections ( system_id ) <EOL> self . assertEqual ( connection . system_id , system_id ) <EOL> self . assertEqual ( connection . _binds [ pdu_types . CommandId . bind_receiver ] [ <NUM_LIT:0> ] , self . proto ) <EOL> bind_manager = self . factory . getBoundConnections ( system_id ) <EOL> delivery_binding = bind_manager . getNextBindingForDelivery ( ) <EOL> self . assertTrue ( delivery_binding is not None ) <EOL> pdu = operations . SubmitSM ( source_addr = '<STR_LIT>' , destination_addr = '<STR_LIT>' , short_message = '<STR_LIT>' , seqNum = <NUM_LIT:6> ) <EOL> self . proto . dataReceived ( self . encoder . encode ( pdu ) ) <EOL> expected_pdu = operations . SubmitSMResp ( seqNum = <NUM_LIT:6> ) <EOL> self . assertEqual ( self . tr . value ( ) , self . encoder . encode ( expected_pdu ) ) <EOL> @ unittest . skip ( '''<STR_LIT>''' ) <EOL> def testUnboundSubmitRequest ( self ) : <EOL> pdu = operations . SubmitSM ( source_addr = '<STR_LIT>' , destination_addr = '<STR_LIT>' , short_message = '<STR_LIT>' , seqNum = <NUM_LIT:1> ) <EOL> self . proto . dataReceived ( self . encoder . encode ( pdu ) ) <EOL> expected_pdu = operations . SubmitSMResp ( status = pdu_types . CommandStatus . ESME_RINVBNDSTS , seqNum = <NUM_LIT:1> ) <EOL> self . assertEqual ( self . tr . value ( ) , self . encoder . encode ( expected_pdu ) ) <EOL> @ unittest . skip ( '''<STR_LIT>''' ) <EOL> def testUnboundSubmitRequest ( self ) : <EOL> pdu = operations . EnquireLink ( seqNum = <NUM_LIT> ) <EOL> self . proto . dataReceived ( self . encoder . encode ( pdu ) ) <EOL> expected_pdu = operations . EnquireLinkResp ( status = pdu_types . CommandStatus . ESME_RINVBNDSTS , seqNum = <NUM_LIT> ) <EOL> self . assertEqual ( self . tr . value ( ) , self . encoder . encode ( expected_pdu ) ) <EOL> @ unittest . skip ( '''<STR_LIT>''' ) <EOL> def testTRXUnbindRequest ( self ) : <EOL> self . _bind ( ) <EOL> pdu = operations . Unbind ( seqNum = <NUM_LIT> ) <EOL> self . proto . dataReceived ( self . encoder . encode ( pdu ) ) <EOL> expected_pdu = operations . UnbindResp ( seqNum = <NUM_LIT> ) <EOL> self . assertEqual ( self . tr . value ( ) , self . encoder . encode ( expected_pdu ) ) <EOL> connection = self . factory . getBoundConnections ( '<STR_LIT>' ) <EOL> self . assertEqual ( connection . system_id , '<STR_LIT>' ) <EOL> self . assertEqual ( connection . _binds [ pdu_types . CommandId . bind_transceiver ] [ <NUM_LIT:0> ] . sessionState , SMPPSessionStates . UNBOUND ) <EOL> @ unittest . skip ( '''<STR_LIT>''' ) <EOL> def testTRXSubmitSM ( self ) : <EOL> self . _bind ( ) <EOL> pdu = operations . SubmitSM ( source_addr = '<STR_LIT>' , destination_addr = '<STR_LIT>' , short_message = '<STR_LIT>' , seqNum = <NUM_LIT:6> ) <EOL> self . proto . dataReceived ( self . encoder . encode ( pdu ) ) <EOL> expected_pdu = operations . SubmitSMResp ( seqNum = <NUM_LIT:6> ) <EOL> self . assertEqual ( self . tr . value ( ) , self . encoder . encode ( expected_pdu ) ) <EOL> system_id , smpp , pdu_notified = self . service_calls . pop ( ) <EOL> self . assertEqual ( system_id , self . proto . system_id ) <EOL> self . assertEqual ( pdu . params [ '<STR_LIT>' ] , pdu_notified . params [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( pdu . params [ '<STR_LIT>' ] , pdu_notified . params [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( pdu . params [ '<STR_LIT>' ] , pdu_notified . params [ '<STR_LIT>' ] ) <EOL> @ unittest . skip ( '''<STR_LIT>''' ) <EOL> def testTRXDataSM ( self ) : <EOL> self . _bind ( ) <EOL> pdu = operations . DataSM ( source_addr = '<STR_LIT>' , destination_addr = '<STR_LIT>' , short_message = '<STR_LIT>' , seqNum = <NUM_LIT:6> ) <EOL> self . proto . dataReceived ( self . encoder . encode ( pdu ) ) <EOL> expected_pdu = operations . DataSMResp ( seqNum = <NUM_LIT:6> ) <EOL> self . assertEqual ( self . tr . value ( ) , self . encoder . encode ( expected_pdu ) ) <EOL> system_id , smpp , pdu_notified = self . service_calls . pop ( ) <EOL> self . assertEqual ( system_id , self . proto . system_id ) <EOL> @ unittest . skip ( '''<STR_LIT>''' ) <EOL> def testTRXQuerySM ( self ) : <EOL> def _serviceHandler ( system_id , smpp , pdu ) : <EOL> self . service_calls . append ( ( system_id , smpp , pdu ) ) <EOL> return DataHandlerResponse ( status = pdu_types . CommandStatus . ESME_ROK , <EOL> message_id = '<STR_LIT>' , <EOL> final_date = None , <EOL> message_state = pdu_types . MessageState . ACCEPTED , <EOL> error_code = <NUM_LIT:0> ) <EOL> self . proto . dataRequestHandler = lambda * args , ** kwargs : _serviceHandler ( self . proto . system_id , * args , ** kwargs ) <EOL> self . _bind ( ) <EOL> pdu = operations . QuerySM ( message_id = '<STR_LIT>' , source_addr = '<STR_LIT>' , seqNum = <NUM_LIT> ) <EOL> self . proto . dataReceived ( self . encoder . encode ( pdu ) ) <EOL> expected_pdu = operations . QuerySMResp ( message_id = '<STR_LIT>' , error_code = <NUM_LIT:0> , final_date = None , message_state = pdu_types . MessageState . ACCEPTED , seqNum = <NUM_LIT> ) <EOL> print self . tr . value ( ) <EOL> self . assertEqual ( self . tr . value ( ) , self . encoder . encode ( expected_pdu ) ) <EOL> system_id , smpp , pdu_notified = self . service_calls . pop ( ) <EOL> self . assertEqual ( system_id , self . proto . system_id ) <EOL> @ unittest . skip ( '''<STR_LIT>''' ) <EOL> def testRecievePduWhileUnbindPending ( self ) : <EOL> self . _bind ( ) <EOL> self . proto . unbind ( ) <EOL> expected_pdu = operations . Unbind ( seqNum = <NUM_LIT:1> ) <EOL> self . assertEqual ( self . tr . value ( ) , self . encoder . encode ( expected_pdu ) ) <EOL> self . tr . clear ( ) <EOL> pdu = operations . SubmitSM ( source_addr = '<STR_LIT>' , destination_addr = '<STR_LIT>' , short_message = '<STR_LIT>' , seqNum = <NUM_LIT:6> ) <EOL> self . proto . dataReceived ( self . encoder . encode ( pdu ) ) <EOL> expected_pdu = operations . SubmitSMResp ( seqNum = <NUM_LIT:6> ) <EOL> self . assertEqual ( self . tr . value ( ) , '<STR_LIT>' ) <EOL> pdu = operations . UnbindResp ( seqNum = <NUM_LIT:1> ) <EOL> self . proto . dataReceived ( self . encoder . encode ( pdu ) ) <EOL> @ unittest . skip ( '''<STR_LIT>''' ) <EOL> def testTRXClientUnbindRequestAfterSubmit ( self ) : <EOL> d = defer . Deferred ( ) <EOL> def _serviceHandler ( system_id , smpp , pdu ) : <EOL> logging . debug ( "<STR_LIT>" , system_id , smpp , pdu ) <EOL> return d <EOL> self . proto . dataRequestHandler = lambda * args , ** kwargs : _serviceHandler ( self . proto . system_id , * args , ** kwargs ) <EOL> self . _bind ( ) <EOL> pdu = operations . SubmitSM ( source_addr = '<STR_LIT>' , destination_addr = '<STR_LIT>' , short_message = '<STR_LIT>' , seqNum = <NUM_LIT:1> ) <EOL> self . proto . dataReceived ( self . encoder . encode ( pdu ) ) <EOL> pdu = operations . Unbind ( seqNum = <NUM_LIT> ) <EOL> self . proto . dataReceived ( self . encoder . encode ( pdu ) ) <EOL> pdu = operations . SubmitSM ( source_addr = '<STR_LIT>' , destination_addr = '<STR_LIT>' , short_message = '<STR_LIT>' , seqNum = <NUM_LIT:5> ) <EOL> self . proto . dataReceived ( self . encoder . encode ( pdu ) ) <EOL> unbind_resp_pdu = operations . UnbindResp ( seqNum = <NUM_LIT> ) <EOL> submit_fail_pdu = operations . SubmitSMResp ( status = pdu_types . CommandStatus . ESME_RINVBNDSTS , seqNum = <NUM_LIT:5> ) <EOL> self . assertEqual ( self . tr . value ( ) , self . encoder . encode ( submit_fail_pdu ) ) <EOL> self . tr . clear ( ) <EOL> d . callback ( pdu_types . CommandStatus . ESME_ROK ) <EOL> expected_pdu = operations . SubmitSMResp ( seqNum = <NUM_LIT:1> ) <EOL> self . assertEqual ( self . tr . value ( ) , '<STR_LIT>' % ( self . encoder . encode ( expected_pdu ) , self . encoder . encode ( unbind_resp_pdu ) ) ) <EOL> @ unittest . skip ( '''<STR_LIT>''' ) <EOL> def testTRXServerUnbindRequestAfterSubmit ( self ) : <EOL> deferreds = [ ] <EOL> def _serviceHandler ( system_id , smpp , pdu ) : <EOL> d = defer . Deferred ( ) <EOL> deferreds . append ( d ) <EOL> logging . debug ( "<STR_LIT>" , system_id , smpp , pdu ) <EOL> return d <EOL> self . proto . dataRequestHandler = lambda * args , ** kwargs : _serviceHandler ( self . proto . system_id , * args , ** kwargs ) <EOL> self . _bind ( ) <EOL> pdu = operations . SubmitSM ( source_addr = '<STR_LIT>' , destination_addr = '<STR_LIT>' , short_message = '<STR_LIT>' , seqNum = <NUM_LIT:1> ) <EOL> self . proto . dataReceived ( self . encoder . encode ( pdu ) ) <EOL> unbind_d = self . proto . unbind ( ) <EOL> print self . tr . value ( ) <EOL> pdu2 = operations . SubmitSM ( source_addr = '<STR_LIT>' , destination_addr = '<STR_LIT>' , short_message = '<STR_LIT>' , seqNum = <NUM_LIT:2> ) <EOL> self . proto . dataReceived ( self . encoder . encode ( pdu ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( deferreds ) ) <EOL> self . assertEqual ( self . tr . value ( ) , '<STR_LIT>' ) <EOL> self . tr . clear ( ) <EOL> deferreds [ - <NUM_LIT:1> ] . callback ( pdu_types . CommandStatus . ESME_ROK ) <EOL> deferreds = deferreds [ : - <NUM_LIT:1> ] <EOL> submit_resp_pdu = operations . SubmitSMResp ( seqNum = <NUM_LIT:1> ) <EOL> unbind_pdu = operations . Unbind ( seqNum = <NUM_LIT:1> ) <EOL> self . assertEqual ( self . tr . value ( ) , '<STR_LIT>' % ( self . encoder . encode ( submit_resp_pdu ) , self . encoder . encode ( unbind_pdu ) ) ) <EOL> self . tr . clear ( ) <EOL> pdu = operations . UnbindResp ( seqNum = <NUM_LIT:1> ) <EOL> self . proto . dataReceived ( self . encoder . encode ( pdu ) ) <EOL> class SMPPServerTimeoutTestCase ( SMPPServerBaseTest ) : <EOL> def setUp ( self ) : <EOL> self . service_calls = [ ] <EOL> self . clock = task . Clock ( ) <EOL> self . encoder = pdu_encoding . PDUEncoder ( ) <EOL> self . smpp_config = SMPPServerConfig ( msgHandler = self . _serviceHandler , <EOL> systems = { '<STR_LIT>' : { "<STR_LIT>" : <NUM_LIT:2> } } , <EOL> enquireLinkTimerSecs = <NUM_LIT:0.1> , <EOL> responseTimerSecs = <NUM_LIT:0.1> <EOL> ) <EOL> portal = Portal ( self . SmppRealm ( ) ) <EOL> credential_checker = InMemoryUsernamePasswordDatabaseDontUse ( ) <EOL> credential_checker . addUser ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> portal . registerChecker ( credential_checker ) <EOL> self . factory = SMPPServerFactory ( self . smpp_config , auth_portal = portal ) <EOL> self . proto = self . factory . buildProtocol ( ( '<STR_LIT:127.0.0.1>' , <NUM_LIT:0> ) ) <EOL> self . proto . callLater = self . clock . callLater <EOL> self . tr = proto_helpers . StringTransport ( ) <EOL> self . proto . makeConnection ( self . tr ) <EOL> @ unittest . skip ( '''<STR_LIT>''' ) <EOL> def testEnquireTimeout ( self ) : <EOL> self . _bind ( ) <EOL> pdu = operations . SubmitSM ( source_addr = '<STR_LIT>' , destination_addr = '<STR_LIT>' , short_message = '<STR_LIT>' , seqNum = <NUM_LIT:6> ) <EOL> self . proto . dataReceived ( self . encoder . encode ( pdu ) ) <EOL> expected_pdu = operations . SubmitSMResp ( seqNum = <NUM_LIT:6> ) <EOL> self . assertEqual ( self . tr . value ( ) , self . encoder . encode ( expected_pdu ) ) <EOL> self . service_calls . pop ( ) <EOL> self . tr . clear ( ) <EOL> self . clock . advance ( <NUM_LIT:0.1> ) <EOL> self . clock . advance ( <NUM_LIT:0.1> ) <EOL> self . assertEqual ( self . proto . sessionState , SMPPSessionStates . UNBIND_PENDING ) </s>
