ssh.py: add flag to keep stdin open
[yardstick.git] / tests / unit / test_ssh.py
index 4260b39..1e021a0 100644 (file)
@@ -18,6 +18,8 @@
 
 import os
 import unittest
+from cStringIO import StringIO
+
 import mock
 
 from yardstick import ssh
@@ -108,7 +110,9 @@ class SSHTestCase(unittest.TestCase):
             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),
+                              password=None,
+                              allow_agent=False, look_for_keys=False,
+                              timeout=1),
         ]
         self.assertEqual(client_calls, client.mock_calls)
 
@@ -156,6 +160,15 @@ class SSHTestCase(unittest.TestCase):
         self.assertEqual([mock.call("uname")] * 3,
                          self.test_client.execute.mock_calls)
 
+    @mock.patch("yardstick.ssh.paramiko")
+    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:
+            self.test_client.send_command('cmd')
+        mock_paramiko_exec_command.assert_called_once_with('cmd',
+                                                            get_pty=True)
+
 
 class SSHRunTestCase(unittest.TestCase):
     """Test SSH.run method in different aspects.
@@ -263,6 +276,23 @@ class SSHRunTestCase(unittest.TestCase):
         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_stdin_keep_open(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 = len
+        fake_stdin = StringIO("line1\nline2\n")
+        self.test_client.run("cmd", stdin=fake_stdin, keep_stdin_open=True)
+        call = mock.call
+        send_calls = [call("line1\nline2\n")]
+        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