#NEWLINE# @ patch ( 'ncclient.transport.Session.send' ) #NEWLINE# @ patch ( patch_str ) #NEWLINE# def test_rpc_timeout_error( self , mock_thread , mock_send ) : #NEWLINE# #INDENT# device_handler = manager . make_device_handler ( { #NEWLINE# #INDENT# 'name' : 'junos' , #NEWLINE# #UNINDENT# } ) #NEWLINE# capabilities = Capabilities ( device_handler . get_capabilities ( ) ) #NEWLINE# session = ncclient . transport . Session ( capabilities ) #NEWLINE# obj = RPC ( session , self , raise_mode = RaiseMode . ALL , timeout = 0 ) #NEWLINE# reply = RPCReply ( xml1 ) #NEWLINE# obj . deliver_reply ( reply ) #NEWLINE# node = new_ele ( 'commit' ) #NEWLINE# sub_ele ( node , 'confirmed' ) #NEWLINE# mock_thread . return_value = False #NEWLINE# with self . assertRaises ( TimeoutExpiredError ) : #NEWLINE# #INDENT# obj . _request ( node )