connection: update test coverage for SSH connection plugin (#82916)

* connection: update test coverage for SSH connection plugin

Signed-off-by: Abhijeet Kasurde <akasurde@redhat.com>

changelogs/fragments/ssh_connection_test.yml

@@ -0,0 +1,3 @@
+---
+bugfixes:
+  - ssh - add tests for the SSH connection plugin.

lib/ansible/plugins/connection/ssh.py

@@ -1224,14 +1224,11 @@ class Connection(ConnectionBase):
 
         # Use the transfer_method option if set
         ssh_transfer_method = self.get_option('ssh_transfer_method')
-        if ssh_transfer_method is None:
-            ssh_transfer_method = 'smart'
 
-        if ssh_transfer_method is not None:
+        if ssh_transfer_method == 'smart':
-            if ssh_transfer_method == 'smart':
+            methods = smart_methods
-                methods = smart_methods
+        else:
-            else:
+            methods = [ssh_transfer_method]
-                methods = [ssh_transfer_method]
 
         for method in methods:
             returncode = stdout = stderr = None

test/units/plugins/connection/test_ssh.py

@@ -26,7 +26,7 @@ import pytest
 from ansible.errors import AnsibleAuthenticationFailure
 import unittest
 from unittest.mock import patch, MagicMock, PropertyMock
-from ansible.errors import AnsibleConnectionFailure
+from ansible.errors import AnsibleError, AnsibleConnectionFailure, AnsibleFileNotFound
 import shlex
 from ansible.module_utils.common.text.converters import to_bytes
 from ansible.playbook.play_context import PlayContext
@@ -241,19 +241,44 @@ class TestConnectionBaseClass(unittest.TestCase):
         conn.host = "some_host"
 
         conn.set_option('reconnection_retries', 9)
-        conn.set_option('ssh_transfer_method', None)  # unless set to None
+        conn.set_option('ssh_transfer_method', None)  # default is smart
 
         # Test when SFTP works
         expected_in_data = b' '.join((b'put', to_bytes(shlex.quote('/path/to/in/file')), to_bytes(shlex.quote('/path/to/dest/file')))) + b'\n'
         conn.put_file('/path/to/in/file', '/path/to/dest/file')
         conn._bare_run.assert_called_with('some command to run', expected_in_data, checkrc=False)
 
+        # Test filenames with unicode
+        expected_in_data = b' '.join((b'put',
+                                      to_bytes(shlex.quote('/path/to/in/file/with/unicode-fö〩')),
+                                      to_bytes(shlex.quote('/path/to/dest/file/with/unicode-fö〩')))) + b'\n'
+        conn.put_file(u'/path/to/in/file/with/unicode-fö〩', u'/path/to/dest/file/with/unicode-fö〩')
+        conn._bare_run.assert_called_with('some command to run', expected_in_data, checkrc=False)
+
         # Test when SFTP doesn't work but SCP does
         conn._bare_run.side_effect = [(1, 'stdout', 'some errors'), (0, '', '')]
         conn.put_file('/path/to/in/file', '/path/to/dest/file')
         conn._bare_run.assert_called_with('some command to run', None, checkrc=False)
         conn._bare_run.side_effect = None
 
+        # Test that a non-zero rc raises an error
+        conn.set_option('ssh_transfer_method', 'sftp')
+        conn._bare_run.return_value = (1, 'stdout', 'some errors')
+        self.assertRaises(AnsibleError, conn.put_file, '/path/to/bad/file', '/remote/path/to/file')
+
+        # Test that rc=255 raises an error
+        conn._bare_run.return_value = (255, 'stdout', 'some errors')
+        self.assertRaises(AnsibleConnectionFailure, conn.put_file, '/path/to/bad/file', '/remote/path/to/file')
+
+        # Test that rc=256 raises an error
+        conn._bare_run.return_value = (256, 'stdout', 'some errors')
+        self.assertRaises(AnsibleError, conn.put_file, '/path/to/bad/file', '/remote/path/to/file')
+
+        # Test that a not-found path raises an error
+        mock_ospe.return_value = False
+        conn._bare_run.return_value = (0, 'stdout', '')
+        self.assertRaises(AnsibleFileNotFound, conn.put_file, '/path/to/bad/file', '/remote/path/to/file')
+
     @patch('time.sleep')
     def test_plugins_connection_ssh_fetch_file(self, mock_sleep):
         pc = PlayContext()
@@ -268,7 +293,7 @@ class TestConnectionBaseClass(unittest.TestCase):
         conn.host = "some_host"
 
         conn.set_option('reconnection_retries', 9)
-        conn.set_option('ssh_transfer_method', None)  # unless set to None
+        conn.set_option('ssh_transfer_method', None)  # default is smart
 
         # Test when SFTP works
         expected_in_data = b' '.join((b'get', to_bytes(shlex.quote('/path/to/in/file')), to_bytes(shlex.quote('/path/to/dest/file')))) + b'\n'
@@ -280,6 +305,28 @@ class TestConnectionBaseClass(unittest.TestCase):
         conn._bare_run.side_effect = [(1, 'stdout', 'some errors'), (0, '', '')]
         conn.fetch_file('/path/to/in/file', '/path/to/dest/file')
         conn._bare_run.assert_called_with('some command to run', None, checkrc=False)
+        conn._bare_run.side_effect = None
+
+        # Test when filename is unicode
+        expected_in_data = b' '.join((b'get',
+                                      to_bytes(shlex.quote('/path/to/in/file/with/unicode-fö〩')),
+                                      to_bytes(shlex.quote('/path/to/dest/file/with/unicode-fö〩')))) + b'\n'
+        conn.fetch_file(u'/path/to/in/file/with/unicode-fö〩', u'/path/to/dest/file/with/unicode-fö〩')
+        conn._bare_run.assert_called_with('some command to run', expected_in_data, checkrc=False)
+        conn._bare_run.side_effect = None
+
+        # Test that a non-zero rc raises an error
+        conn.set_option('ssh_transfer_method', 'sftp')
+        conn._bare_run.return_value = (1, 'stdout', 'some errors')
+        self.assertRaises(AnsibleError, conn.fetch_file, '/path/to/bad/file', '/remote/path/to/file')
+
+        # Test that rc=255 raises an error
+        conn._bare_run.return_value = (255, 'stdout', 'some errors')
+        self.assertRaises(AnsibleConnectionFailure, conn.fetch_file, '/path/to/bad/file', '/remote/path/to/file')
+
+        # Test that rc=256 raises an error
+        conn._bare_run.return_value = (256, 'stdout', 'some errors')
+        self.assertRaises(AnsibleError, conn.fetch_file, '/path/to/bad/file', '/remote/path/to/file')
 
 
 class MockSelector(object):