@ -102,6 +102,7 @@ class TestConnectionBaseClass(unittest.TestCase):
def test_plugins_connection_ssh__examine_output ( self ) :
pc = PlayContext ( )
new_stdin = StringIO ( )
become_success_token = b ' BECOME-SUCCESS-abcdefghijklmnopqrstuvxyz '
conn = connection_loader . get ( ' ssh ' , pc , new_stdin )
conn . set_become_plugin ( become_loader . get ( ' sudo ' ) )
@ -112,24 +113,16 @@ class TestConnectionBaseClass(unittest.TestCase):
conn . become . check_missing_password = MagicMock ( )
def _check_password_prompt ( line ) :
if b ' foo ' in line :
return True
return False
return b ' foo ' in line
def _check_become_success ( line ) :
if b ' BECOME-SUCCESS-abcdefghijklmnopqrstuvxyz ' in line :
return True
return False
return become_success_token in line
def _check_incorrect_password ( line ) :
if b ' incorrect password ' in line :
return True
return False
return b ' incorrect password ' in line
def _check_missing_password ( line ) :
if b ' bad password ' in line :
return True
return False
return b ' bad password ' in line
# test examining output for prompt
conn . _flags = dict (
@ -172,9 +165,9 @@ class TestConnectionBaseClass(unittest.TestCase):
pc . prompt = False
conn . become . prompt = False
pc . success_key = u ' BECOME-SUCCESS-abcdefghijklmnopqrstuvxyz '
conn . become . success = u ' BECOME-SUCCESS-abcdefghijklmnopqrstuvxyz '
output , unprocessed = conn . _examine_output ( u ' source ' , u ' state ' , b ' line 1 \n line 2 \n BECOME-SUCCESS-abcdefghijklmnopqrstuvxyz \n line 3 \n ' , False )
pc . success_key = str ( become_success_token )
conn . become . success = str ( become_success_token )
output , unprocessed = conn . _examine_output ( u ' source ' , u ' state ' , b ' line 1 \n line 2 \n %s \n line 3 \n ' % become_success_token , False )
self . assertEqual ( output , b ' line 1 \n line 2 \n line 3 \n ' )
self . assertEqual ( unprocessed , b ' ' )
self . assertFalse ( conn . _flags [ ' become_prompt ' ] )
@ -182,6 +175,23 @@ class TestConnectionBaseClass(unittest.TestCase):
self . assertFalse ( conn . _flags [ ' become_error ' ] )
self . assertFalse ( conn . _flags [ ' become_nopasswd_error ' ] )
# test we dont detect become success from ssh debug: lines
conn . _flags = dict (
become_prompt = False ,
become_success = False ,
become_error = False ,
become_nopasswd_error = False ,
)
pc . prompt = False
conn . become . prompt = True
pc . success_key = str ( become_success_token )
conn . become . success = str ( become_success_token )
output , unprocessed = conn . _examine_output ( u ' source ' , u ' state ' , b ' line 1 \n line 2 \n debug1: %s \n line 3 \n ' % become_success_token , False )
self . assertEqual ( output , b ' line 1 \n line 2 \n debug1: %s \n line 3 \n ' % become_success_token )
self . assertEqual ( unprocessed , b ' ' )
self . assertFalse ( conn . _flags [ ' become_success ' ] )
# test examining output for become failure
conn . _flags = dict (
become_prompt = False ,