X-Git-Url: https://gerrit.opnfv.org/gerrit/gitweb?a=blobdiff_plain;f=tests%2Funit%2Ftest_ssh.py;h=1c63c00a372a8ac3026bbaf5a43a69a4ed0df013;hb=4b86cc0678bcbd2ad2aac571bb40eb97a7ea65f7;hp=1e021a051879129dc2b3bfc15184e7db70541a95;hpb=fad4811d1a722f2577f158840b7c8e61489e2e90;p=yardstick.git diff --git a/tests/unit/test_ssh.py b/tests/unit/test_ssh.py index 1e021a051..1c63c00a3 100644 --- a/tests/unit/test_ssh.py +++ b/tests/unit/test_ssh.py @@ -16,11 +16,14 @@ # yardstick comment: this file is a modified copy of # rally/tests/unit/common/test_sshutils.py +from __future__ import absolute_import import os +import socket import unittest -from cStringIO import StringIO +from io import StringIO import mock +from oslo_utils import encodeutils from yardstick import ssh @@ -164,10 +167,10 @@ class SSHTestCase(unittest.TestCase): def test_send_command(self, mock_paramiko): paramiko_sshclient = self.test_client._get_client() with mock.patch.object(paramiko_sshclient, "exec_command") \ - as mock_paramiko_exec_command: + as mock_paramiko_exec_command: self.test_client.send_command('cmd') mock_paramiko_exec_command.assert_called_once_with('cmd', - get_pty=True) + get_pty=True) class SSHRunTestCase(unittest.TestCase): @@ -273,7 +276,9 @@ class SSHRunTestCase(unittest.TestCase): fake_stdin.close = mock.Mock(side_effect=close) self.test_client.run("cmd", stdin=fake_stdin) call = mock.call - send_calls = [call("line1"), call("line2"), call("e2")] + send_calls = [call(encodeutils.safe_encode("line1", "utf-8")), + call(encodeutils.safe_encode("line2", "utf-8")), + call(encodeutils.safe_encode("e2", "utf-8"))] self.assertEqual(send_calls, self.fake_session.send.mock_calls) @mock.patch("yardstick.ssh.select") @@ -287,10 +292,10 @@ class SSHRunTestCase(unittest.TestCase): self.fake_session.exit_status_ready.side_effect = [0, 0, 0, True] self.fake_session.send_ready.return_value = True self.fake_session.send.side_effect = len - fake_stdin = StringIO("line1\nline2\n") + fake_stdin = StringIO(u"line1\nline2\n") self.test_client.run("cmd", stdin=fake_stdin, keep_stdin_open=True) call = mock.call - send_calls = [call("line1\nline2\n")] + send_calls = [call(encodeutils.safe_encode("line1\nline2\n", "utf-8"))] self.assertEqual(send_calls, self.fake_session.send.mock_calls) @mock.patch("yardstick.ssh.select") @@ -307,9 +312,91 @@ class SSHRunTestCase(unittest.TestCase): self.fake_session.exit_status_ready.return_value = False self.assertRaises(ssh.SSHTimeout, self.test_client.run, "cmd") + @mock.patch("yardstick.ssh.open", create=True) + def test__put_file_shell(self, mock_open): + with mock.patch.object(self.test_client, "run") as run_mock: + self.test_client._put_file_shell("localfile", "remotefile", 0o42) + run_mock.assert_called_once_with( + 'cat > "remotefile"&& chmod -- 042 "remotefile"', + stdin=mock_open.return_value.__enter__.return_value) + + @mock.patch("yardstick.ssh.open", create=True) + def test__put_file_shell_space(self, mock_open): + with mock.patch.object(self.test_client, "run") as run_mock: + self.test_client._put_file_shell("localfile", + "filename with space", 0o42) + run_mock.assert_called_once_with( + 'cat > "filename with space"&& chmod -- 042 "filename with ' + 'space"', + stdin=mock_open.return_value.__enter__.return_value) + + @mock.patch("yardstick.ssh.open", create=True) + def test__put_file_shell_tilde(self, mock_open): + with mock.patch.object(self.test_client, "run") as run_mock: + self.test_client._put_file_shell("localfile", "~/remotefile", 0o42) + run_mock.assert_called_once_with( + 'cat > ~/"remotefile"&& chmod -- 042 ~/"remotefile"', + stdin=mock_open.return_value.__enter__.return_value) + + @mock.patch("yardstick.ssh.open", create=True) + def test__put_file_shell_tilde_spaces(self, mock_open): + with mock.patch.object(self.test_client, "run") as run_mock: + self.test_client._put_file_shell("localfile", "~/file with space", + 0o42) + run_mock.assert_called_once_with( + 'cat > ~/"file with space"&& chmod -- 042 ~/"file with space"', + stdin=mock_open.return_value.__enter__.return_value) + + @mock.patch("yardstick.ssh.os.stat") + def test__put_file_sftp(self, mock_stat): + sftp = self.fake_client.open_sftp.return_value = mock.MagicMock() + sftp.__enter__.return_value = sftp + + mock_stat.return_value = os.stat_result([0o753] + [0] * 9) + + self.test_client._put_file_sftp("localfile", "remotefile") + + sftp.put.assert_called_once_with("localfile", "remotefile") + mock_stat.assert_called_once_with("localfile") + sftp.chmod.assert_called_once_with("remotefile", 0o753) + sftp.__exit__.assert_called_once_with(None, None, None) + + def test__put_file_sftp_mode(self): + sftp = self.fake_client.open_sftp.return_value = mock.MagicMock() + sftp.__enter__.return_value = sftp + + self.test_client._put_file_sftp("localfile", "remotefile", mode=0o753) + + sftp.put.assert_called_once_with("localfile", "remotefile") + sftp.chmod.assert_called_once_with("remotefile", 0o753) + sftp.__exit__.assert_called_once_with(None, None, None) + + def test_put_file_SSHException(self): + exc = ssh.paramiko.SSHException + self.test_client._put_file_sftp = mock.Mock(side_effect=exc()) + self.test_client._put_file_shell = mock.Mock() + + self.test_client.put_file("foo", "bar", 42) + self.test_client._put_file_sftp.assert_called_once_with("foo", "bar", + mode=42) + self.test_client._put_file_shell.assert_called_once_with("foo", "bar", + mode=42) + + def test_put_file_socket_error(self): + exc = socket.error + self.test_client._put_file_sftp = mock.Mock(side_effect=exc()) + self.test_client._put_file_shell = mock.Mock() + + self.test_client.put_file("foo", "bar", 42) + self.test_client._put_file_sftp.assert_called_once_with("foo", "bar", + mode=42) + self.test_client._put_file_shell.assert_called_once_with("foo", "bar", + mode=42) + def main(): unittest.main() + if __name__ == '__main__': main()