summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorKristian Hunt <kristian.hunt@gmail.com>2015-07-28 13:41:01 +0200
committerJörgen Karlsson <jorgen.w.karlsson@ericsson.com>2015-08-20 11:07:56 +0000
commit29062d3057f116f99d21bc0f71cf9f2fb754dd22 (patch)
tree374b386ff76147eed435a25f586909e8f2f05371
parentdd444c57c2cabf35a9948af10333468c9079f3d0 (diff)
Add unit test for ssh
Unit test file for ssh.py originates from rally. It is modified to be consistent with the ssh module used in yardstick - some tests for missing functionality are removed and naming convention is changed (ssh in yardstick vs sshutils in rally). Running of unittest from run_test.sh is NOT enabled. JIRA:- Change-Id: I3ea554e3fbe6f1f906778f88ddd859424331c8df Signed-off-by: Kristian Hunt <kristian.hunt@gmail.com>
-rw-r--r--tests/unit/test_ssh.py285
1 files changed, 285 insertions, 0 deletions
diff --git a/tests/unit/test_ssh.py b/tests/unit/test_ssh.py
new file mode 100644
index 000000000..4260b39bc
--- /dev/null
+++ b/tests/unit/test_ssh.py
@@ -0,0 +1,285 @@
+# Copyright 2013: Mirantis Inc.
+# All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+# yardstick comment: this file is a modified copy of
+# rally/tests/unit/common/test_sshutils.py
+
+import os
+import unittest
+import mock
+
+from yardstick import ssh
+
+
+class FakeParamikoException(Exception):
+ pass
+
+
+class SSHTestCase(unittest.TestCase):
+ """Test all small SSH methods."""
+
+ def setUp(self):
+ super(SSHTestCase, self).setUp()
+ self.test_client = ssh.SSH("root", "example.net")
+
+ @mock.patch("yardstick.ssh.SSH._get_pkey")
+ def test_construct(self, mock_ssh__get_pkey):
+ mock_ssh__get_pkey.return_value = "pkey"
+ test_ssh = ssh.SSH("root", "example.net", port=33, pkey="key",
+ key_filename="kf", password="secret")
+ mock_ssh__get_pkey.assert_called_once_with("key")
+ self.assertEqual("root", test_ssh.user)
+ self.assertEqual("example.net", test_ssh.host)
+ self.assertEqual(33, test_ssh.port)
+ self.assertEqual("pkey", test_ssh.pkey)
+ self.assertEqual("kf", test_ssh.key_filename)
+ self.assertEqual("secret", test_ssh.password)
+
+ def test_construct_default(self):
+ self.assertEqual("root", self.test_client.user)
+ self.assertEqual("example.net", self.test_client.host)
+ self.assertEqual(22, self.test_client.port)
+ self.assertIsNone(self.test_client.pkey)
+ self.assertIsNone(self.test_client.key_filename)
+ self.assertIsNone(self.test_client.password)
+
+ @mock.patch("yardstick.ssh.paramiko")
+ def test__get_pkey_invalid(self, mock_paramiko):
+ mock_paramiko.SSHException = FakeParamikoException
+ rsa = mock_paramiko.rsakey.RSAKey
+ dss = mock_paramiko.dsskey.DSSKey
+ rsa.from_private_key.side_effect = mock_paramiko.SSHException
+ dss.from_private_key.side_effect = mock_paramiko.SSHException
+ self.assertRaises(ssh.SSHError, self.test_client._get_pkey, "key")
+
+ @mock.patch("yardstick.ssh.six.moves.StringIO")
+ @mock.patch("yardstick.ssh.paramiko")
+ def test__get_pkey_dss(self, mock_paramiko, mock_string_io):
+ mock_paramiko.SSHException = FakeParamikoException
+ mock_string_io.return_value = "string_key"
+ mock_paramiko.dsskey.DSSKey.from_private_key.return_value = "dss_key"
+ rsa = mock_paramiko.rsakey.RSAKey
+ rsa.from_private_key.side_effect = mock_paramiko.SSHException
+ key = self.test_client._get_pkey("key")
+ dss_calls = mock_paramiko.dsskey.DSSKey.from_private_key.mock_calls
+ self.assertEqual([mock.call("string_key")], dss_calls)
+ self.assertEqual(key, "dss_key")
+ mock_string_io.assert_called_once_with("key")
+
+ @mock.patch("yardstick.ssh.six.moves.StringIO")
+ @mock.patch("yardstick.ssh.paramiko")
+ def test__get_pkey_rsa(self, mock_paramiko, mock_string_io):
+ mock_paramiko.SSHException = FakeParamikoException
+ mock_string_io.return_value = "string_key"
+ mock_paramiko.rsakey.RSAKey.from_private_key.return_value = "rsa_key"
+ dss = mock_paramiko.dsskey.DSSKey
+ dss.from_private_key.side_effect = mock_paramiko.SSHException
+ key = self.test_client._get_pkey("key")
+ rsa_calls = mock_paramiko.rsakey.RSAKey.from_private_key.mock_calls
+ self.assertEqual([mock.call("string_key")], rsa_calls)
+ self.assertEqual(key, "rsa_key")
+ mock_string_io.assert_called_once_with("key")
+
+ @mock.patch("yardstick.ssh.SSH._get_pkey")
+ @mock.patch("yardstick.ssh.paramiko")
+ def test__get_client(self, mock_paramiko, mock_ssh__get_pkey):
+ mock_ssh__get_pkey.return_value = "key"
+ fake_client = mock.Mock()
+ mock_paramiko.SSHClient.return_value = fake_client
+ mock_paramiko.AutoAddPolicy.return_value = "autoadd"
+
+ test_ssh = ssh.SSH("admin", "example.net", pkey="key")
+ client = test_ssh._get_client()
+
+ self.assertEqual(fake_client, client)
+ client_calls = [
+ mock.call.set_missing_host_key_policy("autoadd"),
+ mock.call.connect("example.net", username="admin",
+ port=22, pkey="key", key_filename=None,
+ password=None, timeout=1),
+ ]
+ self.assertEqual(client_calls, client.mock_calls)
+
+ def test_close(self):
+ with mock.patch.object(self.test_client, "_client") as m_client:
+ self.test_client.close()
+ m_client.close.assert_called_once_with()
+ self.assertFalse(self.test_client._client)
+
+ @mock.patch("yardstick.ssh.six.moves.StringIO")
+ def test_execute(self, mock_string_io):
+ mock_string_io.side_effect = stdio = [mock.Mock(), mock.Mock()]
+ stdio[0].read.return_value = "stdout fake data"
+ stdio[1].read.return_value = "stderr fake data"
+ with mock.patch.object(self.test_client, "run", return_value=0)\
+ as mock_run:
+ status, stdout, stderr = self.test_client.execute(
+ "cmd",
+ stdin="fake_stdin",
+ timeout=43)
+ mock_run.assert_called_once_with(
+ "cmd", stdin="fake_stdin", stdout=stdio[0],
+ stderr=stdio[1], timeout=43, raise_on_error=False)
+ self.assertEqual(0, status)
+ self.assertEqual("stdout fake data", stdout)
+ self.assertEqual("stderr fake data", stderr)
+
+ @mock.patch("yardstick.ssh.time")
+ def test_wait_timeout(self, mock_time):
+ mock_time.time.side_effect = [1, 50, 150]
+ self.test_client.execute = mock.Mock(side_effect=[ssh.SSHError,
+ ssh.SSHError,
+ 0])
+ self.assertRaises(ssh.SSHTimeout, self.test_client.wait)
+ self.assertEqual([mock.call("uname")] * 2,
+ self.test_client.execute.mock_calls)
+
+ @mock.patch("yardstick.ssh.time")
+ def test_wait(self, mock_time):
+ mock_time.time.side_effect = [1, 50, 100]
+ self.test_client.execute = mock.Mock(side_effect=[ssh.SSHError,
+ ssh.SSHError,
+ 0])
+ self.test_client.wait()
+ self.assertEqual([mock.call("uname")] * 3,
+ self.test_client.execute.mock_calls)
+
+
+class SSHRunTestCase(unittest.TestCase):
+ """Test SSH.run method in different aspects.
+
+ Also tested method "execute".
+ """
+
+ def setUp(self):
+ super(SSHRunTestCase, self).setUp()
+
+ self.fake_client = mock.Mock()
+ self.fake_session = mock.Mock()
+ self.fake_transport = mock.Mock()
+
+ self.fake_transport.open_session.return_value = self.fake_session
+ self.fake_client.get_transport.return_value = self.fake_transport
+
+ self.fake_session.recv_ready.return_value = False
+ self.fake_session.recv_stderr_ready.return_value = False
+ self.fake_session.send_ready.return_value = False
+ self.fake_session.exit_status_ready.return_value = True
+ self.fake_session.recv_exit_status.return_value = 0
+
+ self.test_client = ssh.SSH("admin", "example.net")
+ self.test_client._get_client = mock.Mock(return_value=self.fake_client)
+
+ @mock.patch("yardstick.ssh.select")
+ def test_execute(self, mock_select):
+ mock_select.select.return_value = ([], [], [])
+ self.fake_session.recv_ready.side_effect = [1, 0, 0]
+ self.fake_session.recv_stderr_ready.side_effect = [1, 0]
+ self.fake_session.recv.return_value = "ok"
+ self.fake_session.recv_stderr.return_value = "error"
+ self.fake_session.exit_status_ready.return_value = 1
+ self.fake_session.recv_exit_status.return_value = 127
+ self.assertEqual((127, "ok", "error"), self.test_client.execute("cmd"))
+ self.fake_session.exec_command.assert_called_once_with("cmd")
+
+ @mock.patch("yardstick.ssh.select")
+ def test_execute_args(self, mock_select):
+ mock_select.select.return_value = ([], [], [])
+ self.fake_session.recv_ready.side_effect = [1, 0, 0]
+ self.fake_session.recv_stderr_ready.side_effect = [1, 0]
+ self.fake_session.recv.return_value = "ok"
+ self.fake_session.recv_stderr.return_value = "error"
+ self.fake_session.exit_status_ready.return_value = 1
+ self.fake_session.recv_exit_status.return_value = 127
+
+ result = self.test_client.execute("cmd arg1 'arg2 with space'")
+ self.assertEqual((127, "ok", "error"), result)
+ self.fake_session.exec_command.assert_called_once_with(
+ "cmd arg1 'arg2 with space'")
+
+ @mock.patch("yardstick.ssh.select")
+ def test_run(self, mock_select):
+ mock_select.select.return_value = ([], [], [])
+ self.assertEqual(0, self.test_client.run("cmd"))
+
+ @mock.patch("yardstick.ssh.select")
+ def test_run_nonzero_status(self, mock_select):
+ mock_select.select.return_value = ([], [], [])
+ self.fake_session.recv_exit_status.return_value = 1
+ self.assertRaises(ssh.SSHError, self.test_client.run, "cmd")
+ self.assertEqual(1, self.test_client.run("cmd", raise_on_error=False))
+
+ @mock.patch("yardstick.ssh.select")
+ def test_run_stdout(self, mock_select):
+ mock_select.select.return_value = ([], [], [])
+ self.fake_session.recv_ready.side_effect = [True, True, False]
+ self.fake_session.recv.side_effect = ["ok1", "ok2"]
+ stdout = mock.Mock()
+ self.test_client.run("cmd", stdout=stdout)
+ self.assertEqual([mock.call("ok1"), mock.call("ok2")],
+ stdout.write.mock_calls)
+
+ @mock.patch("yardstick.ssh.select")
+ def test_run_stderr(self, mock_select):
+ mock_select.select.return_value = ([], [], [])
+ self.fake_session.recv_stderr_ready.side_effect = [True, False]
+ self.fake_session.recv_stderr.return_value = "error"
+ stderr = mock.Mock()
+ self.test_client.run("cmd", stderr=stderr)
+ stderr.write.assert_called_once_with("error")
+
+ @mock.patch("yardstick.ssh.select")
+ def test_run_stdin(self, mock_select):
+ """Test run method with stdin.
+
+ Third send call was called with "e2" because only 3 bytes was sent
+ by second call. So remainig 2 bytes of "line2" was sent by third call.
+ """
+ mock_select.select.return_value = ([], [], [])
+ 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 = [5, 3, 2]
+ fake_stdin = mock.Mock()
+ fake_stdin.read.side_effect = ["line1", "line2", ""]
+ fake_stdin.closed = False
+
+ def close():
+ fake_stdin.closed = True
+ 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")]
+ self.assertEqual(send_calls, self.fake_session.send.mock_calls)
+
+ @mock.patch("yardstick.ssh.select")
+ def test_run_select_error(self, mock_select):
+ self.fake_session.exit_status_ready.return_value = False
+ mock_select.select.return_value = ([], [], [True])
+ self.assertRaises(ssh.SSHError, self.test_client.run, "cmd")
+
+ @mock.patch("yardstick.ssh.time")
+ @mock.patch("yardstick.ssh.select")
+ def test_run_timemout(self, mock_select, mock_time):
+ mock_time.time.side_effect = [1, 3700]
+ mock_select.select.return_value = ([], [], [])
+ self.fake_session.exit_status_ready.return_value = False
+ self.assertRaises(ssh.SSHTimeout, self.test_client.run, "cmd")
+
+
+def main():
+ unittest.main()
+
+if __name__ == '__main__':
+ main()