1 # Copyright 2013: Mirantis Inc.
4 # Licensed under the Apache License, Version 2.0 (the "License"); you may
5 # not use this file except in compliance with the License. You may obtain
6 # a copy of the License at
8 # http://www.apache.org/licenses/LICENSE-2.0
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13 # License for the specific language governing permissions and limitations
16 # yardstick comment: this file is a modified copy of
17 # rally/tests/unit/common/test_sshutils.py
21 from cStringIO import StringIO
25 from yardstick import ssh
28 class FakeParamikoException(Exception):
32 class SSHTestCase(unittest.TestCase):
33 """Test all small SSH methods."""
36 super(SSHTestCase, self).setUp()
37 self.test_client = ssh.SSH("root", "example.net")
39 @mock.patch("yardstick.ssh.SSH._get_pkey")
40 def test_construct(self, mock_ssh__get_pkey):
41 mock_ssh__get_pkey.return_value = "pkey"
42 test_ssh = ssh.SSH("root", "example.net", port=33, pkey="key",
43 key_filename="kf", password="secret")
44 mock_ssh__get_pkey.assert_called_once_with("key")
45 self.assertEqual("root", test_ssh.user)
46 self.assertEqual("example.net", test_ssh.host)
47 self.assertEqual(33, test_ssh.port)
48 self.assertEqual("pkey", test_ssh.pkey)
49 self.assertEqual("kf", test_ssh.key_filename)
50 self.assertEqual("secret", test_ssh.password)
52 def test_construct_default(self):
53 self.assertEqual("root", self.test_client.user)
54 self.assertEqual("example.net", self.test_client.host)
55 self.assertEqual(22, self.test_client.port)
56 self.assertIsNone(self.test_client.pkey)
57 self.assertIsNone(self.test_client.key_filename)
58 self.assertIsNone(self.test_client.password)
60 @mock.patch("yardstick.ssh.paramiko")
61 def test__get_pkey_invalid(self, mock_paramiko):
62 mock_paramiko.SSHException = FakeParamikoException
63 rsa = mock_paramiko.rsakey.RSAKey
64 dss = mock_paramiko.dsskey.DSSKey
65 rsa.from_private_key.side_effect = mock_paramiko.SSHException
66 dss.from_private_key.side_effect = mock_paramiko.SSHException
67 self.assertRaises(ssh.SSHError, self.test_client._get_pkey, "key")
69 @mock.patch("yardstick.ssh.six.moves.StringIO")
70 @mock.patch("yardstick.ssh.paramiko")
71 def test__get_pkey_dss(self, mock_paramiko, mock_string_io):
72 mock_paramiko.SSHException = FakeParamikoException
73 mock_string_io.return_value = "string_key"
74 mock_paramiko.dsskey.DSSKey.from_private_key.return_value = "dss_key"
75 rsa = mock_paramiko.rsakey.RSAKey
76 rsa.from_private_key.side_effect = mock_paramiko.SSHException
77 key = self.test_client._get_pkey("key")
78 dss_calls = mock_paramiko.dsskey.DSSKey.from_private_key.mock_calls
79 self.assertEqual([mock.call("string_key")], dss_calls)
80 self.assertEqual(key, "dss_key")
81 mock_string_io.assert_called_once_with("key")
83 @mock.patch("yardstick.ssh.six.moves.StringIO")
84 @mock.patch("yardstick.ssh.paramiko")
85 def test__get_pkey_rsa(self, mock_paramiko, mock_string_io):
86 mock_paramiko.SSHException = FakeParamikoException
87 mock_string_io.return_value = "string_key"
88 mock_paramiko.rsakey.RSAKey.from_private_key.return_value = "rsa_key"
89 dss = mock_paramiko.dsskey.DSSKey
90 dss.from_private_key.side_effect = mock_paramiko.SSHException
91 key = self.test_client._get_pkey("key")
92 rsa_calls = mock_paramiko.rsakey.RSAKey.from_private_key.mock_calls
93 self.assertEqual([mock.call("string_key")], rsa_calls)
94 self.assertEqual(key, "rsa_key")
95 mock_string_io.assert_called_once_with("key")
97 @mock.patch("yardstick.ssh.SSH._get_pkey")
98 @mock.patch("yardstick.ssh.paramiko")
99 def test__get_client(self, mock_paramiko, mock_ssh__get_pkey):
100 mock_ssh__get_pkey.return_value = "key"
101 fake_client = mock.Mock()
102 mock_paramiko.SSHClient.return_value = fake_client
103 mock_paramiko.AutoAddPolicy.return_value = "autoadd"
105 test_ssh = ssh.SSH("admin", "example.net", pkey="key")
106 client = test_ssh._get_client()
108 self.assertEqual(fake_client, client)
110 mock.call.set_missing_host_key_policy("autoadd"),
111 mock.call.connect("example.net", username="admin",
112 port=22, pkey="key", key_filename=None,
114 allow_agent=False, look_for_keys=False,
117 self.assertEqual(client_calls, client.mock_calls)
119 def test_close(self):
120 with mock.patch.object(self.test_client, "_client") as m_client:
121 self.test_client.close()
122 m_client.close.assert_called_once_with()
123 self.assertFalse(self.test_client._client)
125 @mock.patch("yardstick.ssh.six.moves.StringIO")
126 def test_execute(self, mock_string_io):
127 mock_string_io.side_effect = stdio = [mock.Mock(), mock.Mock()]
128 stdio[0].read.return_value = "stdout fake data"
129 stdio[1].read.return_value = "stderr fake data"
130 with mock.patch.object(self.test_client, "run", return_value=0)\
132 status, stdout, stderr = self.test_client.execute(
136 mock_run.assert_called_once_with(
137 "cmd", stdin="fake_stdin", stdout=stdio[0],
138 stderr=stdio[1], timeout=43, raise_on_error=False)
139 self.assertEqual(0, status)
140 self.assertEqual("stdout fake data", stdout)
141 self.assertEqual("stderr fake data", stderr)
143 @mock.patch("yardstick.ssh.time")
144 def test_wait_timeout(self, mock_time):
145 mock_time.time.side_effect = [1, 50, 150]
146 self.test_client.execute = mock.Mock(side_effect=[ssh.SSHError,
149 self.assertRaises(ssh.SSHTimeout, self.test_client.wait)
150 self.assertEqual([mock.call("uname")] * 2,
151 self.test_client.execute.mock_calls)
153 @mock.patch("yardstick.ssh.time")
154 def test_wait(self, mock_time):
155 mock_time.time.side_effect = [1, 50, 100]
156 self.test_client.execute = mock.Mock(side_effect=[ssh.SSHError,
159 self.test_client.wait()
160 self.assertEqual([mock.call("uname")] * 3,
161 self.test_client.execute.mock_calls)
163 @mock.patch("yardstick.ssh.paramiko")
164 def test_send_command(self, mock_paramiko):
165 paramiko_sshclient = self.test_client._get_client()
166 with mock.patch.object(paramiko_sshclient, "exec_command") \
167 as mock_paramiko_exec_command:
168 self.test_client.send_command('cmd')
169 mock_paramiko_exec_command.assert_called_once_with('cmd',
173 class SSHRunTestCase(unittest.TestCase):
174 """Test SSH.run method in different aspects.
176 Also tested method "execute".
180 super(SSHRunTestCase, self).setUp()
182 self.fake_client = mock.Mock()
183 self.fake_session = mock.Mock()
184 self.fake_transport = mock.Mock()
186 self.fake_transport.open_session.return_value = self.fake_session
187 self.fake_client.get_transport.return_value = self.fake_transport
189 self.fake_session.recv_ready.return_value = False
190 self.fake_session.recv_stderr_ready.return_value = False
191 self.fake_session.send_ready.return_value = False
192 self.fake_session.exit_status_ready.return_value = True
193 self.fake_session.recv_exit_status.return_value = 0
195 self.test_client = ssh.SSH("admin", "example.net")
196 self.test_client._get_client = mock.Mock(return_value=self.fake_client)
198 @mock.patch("yardstick.ssh.select")
199 def test_execute(self, mock_select):
200 mock_select.select.return_value = ([], [], [])
201 self.fake_session.recv_ready.side_effect = [1, 0, 0]
202 self.fake_session.recv_stderr_ready.side_effect = [1, 0]
203 self.fake_session.recv.return_value = "ok"
204 self.fake_session.recv_stderr.return_value = "error"
205 self.fake_session.exit_status_ready.return_value = 1
206 self.fake_session.recv_exit_status.return_value = 127
207 self.assertEqual((127, "ok", "error"), self.test_client.execute("cmd"))
208 self.fake_session.exec_command.assert_called_once_with("cmd")
210 @mock.patch("yardstick.ssh.select")
211 def test_execute_args(self, mock_select):
212 mock_select.select.return_value = ([], [], [])
213 self.fake_session.recv_ready.side_effect = [1, 0, 0]
214 self.fake_session.recv_stderr_ready.side_effect = [1, 0]
215 self.fake_session.recv.return_value = "ok"
216 self.fake_session.recv_stderr.return_value = "error"
217 self.fake_session.exit_status_ready.return_value = 1
218 self.fake_session.recv_exit_status.return_value = 127
220 result = self.test_client.execute("cmd arg1 'arg2 with space'")
221 self.assertEqual((127, "ok", "error"), result)
222 self.fake_session.exec_command.assert_called_once_with(
223 "cmd arg1 'arg2 with space'")
225 @mock.patch("yardstick.ssh.select")
226 def test_run(self, mock_select):
227 mock_select.select.return_value = ([], [], [])
228 self.assertEqual(0, self.test_client.run("cmd"))
230 @mock.patch("yardstick.ssh.select")
231 def test_run_nonzero_status(self, mock_select):
232 mock_select.select.return_value = ([], [], [])
233 self.fake_session.recv_exit_status.return_value = 1
234 self.assertRaises(ssh.SSHError, self.test_client.run, "cmd")
235 self.assertEqual(1, self.test_client.run("cmd", raise_on_error=False))
237 @mock.patch("yardstick.ssh.select")
238 def test_run_stdout(self, mock_select):
239 mock_select.select.return_value = ([], [], [])
240 self.fake_session.recv_ready.side_effect = [True, True, False]
241 self.fake_session.recv.side_effect = ["ok1", "ok2"]
243 self.test_client.run("cmd", stdout=stdout)
244 self.assertEqual([mock.call("ok1"), mock.call("ok2")],
245 stdout.write.mock_calls)
247 @mock.patch("yardstick.ssh.select")
248 def test_run_stderr(self, mock_select):
249 mock_select.select.return_value = ([], [], [])
250 self.fake_session.recv_stderr_ready.side_effect = [True, False]
251 self.fake_session.recv_stderr.return_value = "error"
253 self.test_client.run("cmd", stderr=stderr)
254 stderr.write.assert_called_once_with("error")
256 @mock.patch("yardstick.ssh.select")
257 def test_run_stdin(self, mock_select):
258 """Test run method with stdin.
260 Third send call was called with "e2" because only 3 bytes was sent
261 by second call. So remainig 2 bytes of "line2" was sent by third call.
263 mock_select.select.return_value = ([], [], [])
264 self.fake_session.exit_status_ready.side_effect = [0, 0, 0, True]
265 self.fake_session.send_ready.return_value = True
266 self.fake_session.send.side_effect = [5, 3, 2]
267 fake_stdin = mock.Mock()
268 fake_stdin.read.side_effect = ["line1", "line2", ""]
269 fake_stdin.closed = False
272 fake_stdin.closed = True
273 fake_stdin.close = mock.Mock(side_effect=close)
274 self.test_client.run("cmd", stdin=fake_stdin)
276 send_calls = [call("line1"), call("line2"), call("e2")]
277 self.assertEqual(send_calls, self.fake_session.send.mock_calls)
279 @mock.patch("yardstick.ssh.select")
280 def test_run_stdin_keep_open(self, mock_select):
281 """Test run method with stdin.
283 Third send call was called with "e2" because only 3 bytes was sent
284 by second call. So remainig 2 bytes of "line2" was sent by third call.
286 mock_select.select.return_value = ([], [], [])
287 self.fake_session.exit_status_ready.side_effect = [0, 0, 0, True]
288 self.fake_session.send_ready.return_value = True
289 self.fake_session.send.side_effect = len
290 fake_stdin = StringIO("line1\nline2\n")
291 self.test_client.run("cmd", stdin=fake_stdin, keep_stdin_open=True)
293 send_calls = [call("line1\nline2\n")]
294 self.assertEqual(send_calls, self.fake_session.send.mock_calls)
296 @mock.patch("yardstick.ssh.select")
297 def test_run_select_error(self, mock_select):
298 self.fake_session.exit_status_ready.return_value = False
299 mock_select.select.return_value = ([], [], [True])
300 self.assertRaises(ssh.SSHError, self.test_client.run, "cmd")
302 @mock.patch("yardstick.ssh.time")
303 @mock.patch("yardstick.ssh.select")
304 def test_run_timemout(self, mock_select, mock_time):
305 mock_time.time.side_effect = [1, 3700]
306 mock_select.select.return_value = ([], [], [])
307 self.fake_session.exit_status_ready.return_value = False
308 self.assertRaises(ssh.SSHTimeout, self.test_client.run, "cmd")
314 if __name__ == '__main__':