import os
import unittest
+from cStringIO import StringIO
+
import mock
from yardstick import ssh
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)
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.
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