<s> import mock <EOL> import unittest <EOL> from cStringIO import StringIO <EOL> from sts . entities . base import SSHEntity <EOL> paramiko_installed = False <EOL> try : <EOL> import paramiko <EOL> paramiko_installed = True <EOL> except ImportError : <EOL> paramiko_installed = False <EOL> class SSHEntityTest ( unittest . TestCase ) : <EOL> @ mock . patch ( "<STR_LIT>" ) <EOL> @ unittest . skipIf ( not paramiko_installed , "<STR_LIT>" ) <EOL> def test_init ( self , ssh_client_cls ) : <EOL> host = "<STR_LIT:localhost>" <EOL> port = <NUM_LIT> <EOL> username = None <EOL> password = None <EOL> key_filename = None <EOL> ssh = SSHEntity ( host , port , username , password , key_filename ) <EOL> ssh . _ssh_cls = ssh_client_cls <EOL> client = ssh . ssh_client <EOL> self . assertEquals ( ssh . host , host ) <EOL> self . assertEquals ( ssh . port , port ) <EOL> self . assertEquals ( ssh . username , username ) <EOL> self . assertEquals ( ssh . password , password ) <EOL> self . assertEquals ( ssh . key_filename , key_filename ) <EOL> self . assertEquals ( ssh . key_filename , key_filename ) <EOL> self . assertEquals ( ssh . ssh_cls , ssh_client_cls ) <EOL> self . assertEquals ( client . set_missing_host_key_policy . call_count , <NUM_LIT:1> ) <EOL> client . connect . assert_called_once_with ( hostname = host , port = port , <EOL> username = username , password = password , <EOL> key_filename = key_filename ) <EOL> @ mock . patch ( "<STR_LIT>" ) <EOL> @ unittest . skipIf ( not paramiko_installed , "<STR_LIT>" ) <EOL> def test_get_new_session ( self , ssh_client_cls ) : <EOL> host = "<STR_LIT:localhost>" <EOL> ssh = SSHEntity ( host ) <EOL> ssh . _ssh_client = ssh_client_cls ( ) <EOL> session = ssh . get_new_session ( ) <EOL> self . assertEquals ( ssh . _ssh_client . get_transport . call_count , <NUM_LIT:1> ) <EOL> @ mock . patch ( "<STR_LIT>" ) <EOL> @ unittest . skipIf ( not paramiko_installed , "<STR_LIT>" ) <EOL> def test_execute_command ( self , ssh_client_cls ) : <EOL> host = "<STR_LIT:localhost>" <EOL> cmd = "<STR_LIT>" <EOL> ssh = SSHEntity ( host ) <EOL> ssh . _ssh_cls = ssh_client_cls <EOL> session_mock = mock . Mock ( ) <EOL> session_mock . recv_ready . return_value = True <EOL> session_mock . exit_status_ready . return_value = True <EOL> session_mock . recv . return_value = "<STR_LIT>" <EOL> stream_mock = mock . Mock ( ) <EOL> stream_mock . channel = session_mock <EOL> ssh . ssh_client . exec_command . return_value = None , stream_mock , None <EOL> def change_status ( y ) : <EOL> session_mock . recv_ready . return_value = False <EOL> return "<STR_LIT>" <EOL> session_mock . recv . side_effect = change_status <EOL> ssh . get_new_session = lambda : session_mock <EOL> ssh . execute_command ( cmd ) <EOL> ssh . ssh_client . exec_command . assert_called_once_with ( cmd ) <EOL> self . assertTrue ( session_mock . recv . call_count >= <NUM_LIT:1> ) <EOL> @ mock . patch ( "<STR_LIT>" ) <EOL> @ unittest . skipIf ( not paramiko_installed , "<STR_LIT>" ) <EOL> def test_execute_command_with_redirect ( self , ssh_client_cls ) : <EOL> host = "<STR_LIT:localhost>" <EOL> cmd = "<STR_LIT>" <EOL> ssh = SSHEntity ( host , redirect_output = True ) <EOL> ssh . _ssh_cls = ssh_client_cls <EOL> ssh . ssh_client . exec_command . return_value = ( StringIO ( ) , StringIO ( ) , <EOL> StringIO ( ) ) <EOL> ssh . execute_command ( cmd ) <EOL> ssh . ssh_client . exec_command . assert_called_once_with ( cmd ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
