@ -25,7 +25,7 @@ from io import StringIO
import pytest
import pytest
from ansible . compat . tests import unittest
from ansible . compat . tests import unittest
from ansible . compat . tests . mock import patch , MagicMock
from ansible . compat . tests . mock import patch , MagicMock , PropertyMock
from ansible import constants as C
from ansible import constants as C
from ansible . compat . selectors import SelectorKey , EVENT_READ
from ansible . compat . selectors import SelectorKey , EVENT_READ
@ -72,7 +72,7 @@ class TestConnectionBaseClass(unittest.TestCase):
conn = ssh . Connection ( pc , new_stdin )
conn = ssh . Connection ( pc , new_stdin )
conn . _build_command ( ' ssh ' )
conn . _build_command ( ' ssh ' )
def test_plugins_connection_ssh_ _ exec_command( self ) :
def test_plugins_connection_ssh_ exec_command( self ) :
pc = PlayContext ( )
pc = PlayContext ( )
new_stdin = StringIO ( )
new_stdin = StringIO ( )
conn = ssh . Connection ( pc , new_stdin )
conn = ssh . Connection ( pc , new_stdin )
@ -82,8 +82,8 @@ class TestConnectionBaseClass(unittest.TestCase):
conn . _run = MagicMock ( )
conn . _run = MagicMock ( )
conn . _run . return_value = ( 0 , ' stdout ' , ' stderr ' )
conn . _run . return_value = ( 0 , ' stdout ' , ' stderr ' )
res , stdout , stderr = conn . _ exec_command( ' ssh ' )
res , stdout , stderr = conn . exec_command( ' ssh ' )
res , stdout , stderr = conn . _ exec_command( ' ssh ' , ' this is some data ' )
res , stdout , stderr = conn . exec_command( ' ssh ' , ' this is some data ' )
def test_plugins_connection_ssh__examine_output ( self ) :
def test_plugins_connection_ssh__examine_output ( self ) :
pc = PlayContext ( )
pc = PlayContext ( )
@ -193,36 +193,8 @@ class TestConnectionBaseClass(unittest.TestCase):
self . assertTrue ( conn . _flags [ ' become_nopasswd_error ' ] )
self . assertTrue ( conn . _flags [ ' become_nopasswd_error ' ] )
@patch ( ' time.sleep ' )
@patch ( ' time.sleep ' )
def test_plugins_connection_ssh_exec_command ( self , mock_sleep ) :
pc = PlayContext ( )
new_stdin = StringIO ( )
conn = ssh . Connection ( pc , new_stdin )
conn . _build_command = MagicMock ( )
conn . _exec_command = MagicMock ( )
C . ANSIBLE_SSH_RETRIES = 9
# test a regular, successful execution
conn . _exec_command . return_value = ( 0 , b ' stdout ' , b ' ' )
res = conn . exec_command ( ' ssh ' , ' some data ' )
self . assertEquals ( res , ( 0 , b ' stdout ' , b ' ' ) , msg = ' exec_command did not return what the _exec_command helper returned ' )
# test a retry, followed by success
conn . _exec_command . return_value = None
conn . _exec_command . side_effect = [ ( 255 , ' ' , ' ' ) , ( 0 , b ' stdout ' , b ' ' ) ]
res = conn . exec_command ( ' ssh ' , ' some data ' )
self . assertEquals ( res , ( 0 , b ' stdout ' , b ' ' ) , msg = ' exec_command did not return what the _exec_command helper returned ' )
# test multiple failures
conn . _exec_command . side_effect = [ ( 255 , b ' ' , b ' ' ) ] * 10
self . assertRaises ( AnsibleConnectionFailure , conn . exec_command , ' ssh ' , ' some data ' )
# test other failure from exec_command
conn . _exec_command . side_effect = [ Exception ( ' bad ' ) ] * 10
self . assertRaises ( Exception , conn . exec_command , ' ssh ' , ' some data ' )
@patch ( ' os.path.exists ' )
@patch ( ' os.path.exists ' )
def test_plugins_connection_ssh_put_file ( self , mock_ospe ):
def test_plugins_connection_ssh_put_file ( self , mock_ospe , mock_sleep ) :
pc = PlayContext ( )
pc = PlayContext ( )
new_stdin = StringIO ( )
new_stdin = StringIO ( )
conn = ssh . Connection ( pc , new_stdin )
conn = ssh . Connection ( pc , new_stdin )
@ -234,6 +206,8 @@ class TestConnectionBaseClass(unittest.TestCase):
conn . _run . return_value = ( 0 , ' ' , ' ' )
conn . _run . return_value = ( 0 , ' ' , ' ' )
conn . host = " some_host "
conn . host = " some_host "
C . ANSIBLE_SSH_RETRIES = 9
# Test with C.DEFAULT_SCP_IF_SSH set to smart
# Test with C.DEFAULT_SCP_IF_SSH set to smart
# Test when SFTP works
# Test when SFTP works
C . DEFAULT_SCP_IF_SSH = ' smart '
C . DEFAULT_SCP_IF_SSH = ' smart '
@ -276,7 +250,8 @@ class TestConnectionBaseClass(unittest.TestCase):
conn . _run . return_value = ( 0 , ' stdout ' , ' ' )
conn . _run . return_value = ( 0 , ' stdout ' , ' ' )
self . assertRaises ( AnsibleFileNotFound , conn . put_file , ' /path/to/bad/file ' , ' /remote/path/to/file ' )
self . assertRaises ( AnsibleFileNotFound , conn . put_file , ' /path/to/bad/file ' , ' /remote/path/to/file ' )
def test_plugins_connection_ssh_fetch_file ( self ) :
@patch ( ' time.sleep ' )
def test_plugins_connection_ssh_fetch_file ( self , mock_sleep ) :
pc = PlayContext ( )
pc = PlayContext ( )
new_stdin = StringIO ( )
new_stdin = StringIO ( )
conn = ssh . Connection ( pc , new_stdin )
conn = ssh . Connection ( pc , new_stdin )
@ -287,6 +262,8 @@ class TestConnectionBaseClass(unittest.TestCase):
conn . _run . return_value = ( 0 , ' ' , ' ' )
conn . _run . return_value = ( 0 , ' ' , ' ' )
conn . host = " some_host "
conn . host = " some_host "
C . ANSIBLE_SSH_RETRIES = 9
# Test with C.DEFAULT_SCP_IF_SSH set to smart
# Test with C.DEFAULT_SCP_IF_SSH set to smart
# Test when SFTP works
# Test when SFTP works
C . DEFAULT_SCP_IF_SSH = ' smart '
C . DEFAULT_SCP_IF_SSH = ' smart '
@ -535,3 +512,120 @@ class TestSSHConnectionRun(object):
assert self . mock_selector . register . called is True
assert self . mock_selector . register . called is True
assert self . mock_selector . register . call_count == 2
assert self . mock_selector . register . call_count == 2
assert self . conn . _send_initial_data . called is False
assert self . conn . _send_initial_data . called is False
@pytest.mark.usefixtures ( ' mock_run_env ' )
class TestSSHConnectionRetries ( object ) :
def test_retry_then_success ( self ) :
self . mock_popen_res . stdout . read . side_effect = [ b " " , b " my_stdout \n " , b " second_line " ]
self . mock_popen_res . stderr . read . side_effect = [ b " " , b " my_stderr " ]
type ( self . mock_popen_res ) . returncode = PropertyMock ( side_effect = [ 255 ] * 3 + [ 0 ] * 4 )
self . mock_selector . select . side_effect = [
[ ( SelectorKey ( self . mock_popen_res . stdout , 1001 , [ EVENT_READ ] , None ) , EVENT_READ ) ] ,
[ ( SelectorKey ( self . mock_popen_res . stderr , 1002 , [ EVENT_READ ] , None ) , EVENT_READ ) ] ,
[ ] ,
[ ( SelectorKey ( self . mock_popen_res . stdout , 1001 , [ EVENT_READ ] , None ) , EVENT_READ ) ] ,
[ ( SelectorKey ( self . mock_popen_res . stdout , 1001 , [ EVENT_READ ] , None ) , EVENT_READ ) ] ,
[ ( SelectorKey ( self . mock_popen_res . stderr , 1002 , [ EVENT_READ ] , None ) , EVENT_READ ) ] ,
[ ]
]
self . mock_selector . get_map . side_effect = lambda : True
self . conn . _build_command = MagicMock ( )
self . conn . _build_command . return_value = ' ssh '
return_code , b_stdout , b_stderr = self . conn . exec_command ( ' ssh ' , ' some data ' )
assert return_code == 0
assert b_stdout == b ' my_stdout \n second_line '
assert b_stderr == b ' my_stderr '
@patch ( ' time.sleep ' )
def test_multiple_failures ( self , mock_sleep ) :
C . ANSIBLE_SSH_RETRIES = 9
self . mock_popen_res . stdout . read . side_effect = [ b " " ] * 11
self . mock_popen_res . stderr . read . side_effect = [ b " " ] * 11
type ( self . mock_popen_res ) . returncode = PropertyMock ( side_effect = [ 255 ] * 30 )
self . mock_selector . select . side_effect = [
[ ( SelectorKey ( self . mock_popen_res . stdout , 1001 , [ EVENT_READ ] , None ) , EVENT_READ ) ] ,
[ ( SelectorKey ( self . mock_popen_res . stderr , 1002 , [ EVENT_READ ] , None ) , EVENT_READ ) ] ,
[ ] ,
] * 10
self . mock_selector . get_map . side_effect = lambda : True
self . conn . _build_command = MagicMock ( )
self . conn . _build_command . return_value = ' ssh '
pytest . raises ( AnsibleConnectionFailure , self . conn . exec_command , ' ssh ' , ' some data ' )
assert self . mock_popen . call_count == 10
@patch ( ' time.sleep ' )
def test_abitrary_exceptions ( self , mock_sleep ) :
C . ANSIBLE_SSH_RETRIES = 9
self . conn . _build_command = MagicMock ( )
self . conn . _build_command . return_value = ' ssh '
self . mock_popen . side_effect = [ Exception ( ' bad ' ) ] * 10
pytest . raises ( Exception , self . conn . exec_command , ' ssh ' , ' some data ' )
assert self . mock_popen . call_count == 10
@patch ( ' time.sleep ' )
@patch ( ' ansible.plugins.connection.ssh.os ' )
def test_put_file_retries ( self , os_mock , time_mock ) :
os_mock . path . exists . return_value = True
self . mock_popen_res . stdout . read . side_effect = [ b " " , b " my_stdout \n " , b " second_line " ]
self . mock_popen_res . stderr . read . side_effect = [ b " " , b " my_stderr " ]
type ( self . mock_popen_res ) . returncode = PropertyMock ( side_effect = [ 255 ] * 3 + [ 0 ] * 4 )
self . mock_selector . select . side_effect = [
[ ( SelectorKey ( self . mock_popen_res . stdout , 1001 , [ EVENT_READ ] , None ) , EVENT_READ ) ] ,
[ ( SelectorKey ( self . mock_popen_res . stderr , 1002 , [ EVENT_READ ] , None ) , EVENT_READ ) ] ,
[ ] ,
[ ( SelectorKey ( self . mock_popen_res . stdout , 1001 , [ EVENT_READ ] , None ) , EVENT_READ ) ] ,
[ ( SelectorKey ( self . mock_popen_res . stdout , 1001 , [ EVENT_READ ] , None ) , EVENT_READ ) ] ,
[ ( SelectorKey ( self . mock_popen_res . stderr , 1002 , [ EVENT_READ ] , None ) , EVENT_READ ) ] ,
[ ]
]
self . mock_selector . get_map . side_effect = lambda : True
self . conn . _build_command = MagicMock ( )
self . conn . _build_command . return_value = ' ssh '
return_code , b_stdout , b_stderr = self . conn . put_file ( ' /path/to/in/file ' , ' /path/to/dest/file ' )
assert return_code == 0
assert b_stdout == b " my_stdout \n second_line "
assert b_stderr == b " my_stderr "
assert self . mock_popen . call_count == 2
@patch ( ' time.sleep ' )
@patch ( ' ansible.plugins.connection.ssh.os ' )
def test_fetch_file_retries ( self , os_mock , time_mock ) :
os_mock . path . exists . return_value = True
self . mock_popen_res . stdout . read . side_effect = [ b " " , b " my_stdout \n " , b " second_line " ]
self . mock_popen_res . stderr . read . side_effect = [ b " " , b " my_stderr " ]
type ( self . mock_popen_res ) . returncode = PropertyMock ( side_effect = [ 255 ] * 3 + [ 0 ] * 4 )
self . mock_selector . select . side_effect = [
[ ( SelectorKey ( self . mock_popen_res . stdout , 1001 , [ EVENT_READ ] , None ) , EVENT_READ ) ] ,
[ ( SelectorKey ( self . mock_popen_res . stderr , 1002 , [ EVENT_READ ] , None ) , EVENT_READ ) ] ,
[ ] ,
[ ( SelectorKey ( self . mock_popen_res . stdout , 1001 , [ EVENT_READ ] , None ) , EVENT_READ ) ] ,
[ ( SelectorKey ( self . mock_popen_res . stdout , 1001 , [ EVENT_READ ] , None ) , EVENT_READ ) ] ,
[ ( SelectorKey ( self . mock_popen_res . stderr , 1002 , [ EVENT_READ ] , None ) , EVENT_READ ) ] ,
[ ]
]
self . mock_selector . get_map . side_effect = lambda : True
self . conn . _build_command = MagicMock ( )
self . conn . _build_command . return_value = ' ssh '
return_code , b_stdout , b_stderr = self . conn . fetch_file ( ' /path/to/in/file ' , ' /path/to/dest/file ' )
assert return_code == 0
assert b_stdout == b " my_stdout \n second_line "
assert b_stderr == b " my_stderr "
assert self . mock_popen . call_count == 2