# Copyright 2014 OpenStack Foundation # # 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. from io import StringIO import socket from unittest import mock import testtools from tempest.lib.common import ssh from tempest.lib import exceptions from tempest.tests import base import tempest.tests.utils as utils class TestSshClient(base.TestCase): SELECT_POLLIN = 1 @mock.patch('paramiko.RSAKey.from_private_key') @mock.patch('io.StringIO') def test_pkey_calls_paramiko_RSAKey(self, cs_mock, rsa_mock): cs_mock.return_value = mock.sentinel.csio pkey = 'mykey' ssh.Client('localhost', 'root', pkey=pkey) rsa_mock.assert_called_once_with(mock.sentinel.csio) cs_mock.assert_called_once_with('mykey') rsa_mock.reset_mock() cs_mock.reset_mock() pkey = mock.sentinel.pkey # Shouldn't call out to load a file from RSAKey, since # a sentinel isn't a basestring... ssh.Client('localhost', 'root', pkey=pkey) self.assertEqual(0, rsa_mock.call_count) self.assertEqual(0, cs_mock.call_count) def _set_ssh_connection_mocks(self): client_mock = mock.MagicMock() client_mock.connect.return_value = True return (self.patch('paramiko.SSHClient'), self.patch('paramiko.AutoAddPolicy'), client_mock) def test_get_ssh_connection(self): c_mock, aa_mock, client_mock = self._set_ssh_connection_mocks() s_mock = self.patch('time.sleep') c_mock.return_value = client_mock aa_mock.return_value = mock.sentinel.aa # Test normal case for successful connection on first try client = ssh.Client('localhost', 'root', timeout=2) client._get_ssh_connection(sleep=1) aa_mock.assert_called_once_with() client_mock.set_missing_host_key_policy.assert_called_once_with( mock.sentinel.aa) expected_connect = [mock.call( 'localhost', port=22, username='root', pkey=None, key_filename=None, look_for_keys=False, timeout=10.0, password=None, sock=None, allow_agent=True )] self.assertEqual(expected_connect, client_mock.connect.mock_calls) self.assertEqual(0, s_mock.call_count) def test_get_ssh_connection_over_ssh(self): c_mock, aa_mock, client_mock = self._set_ssh_connection_mocks() proxy_client_mock = mock.MagicMock() proxy_client_mock.connect.return_value = True s_mock = self.patch('time.sleep') c_mock.side_effect = [client_mock, proxy_client_mock] aa_mock.return_value = mock.sentinel.aa proxy_client = ssh.Client('proxy-host', 'proxy-user', timeout=2) client = ssh.Client('localhost', 'root', timeout=2, proxy_client=proxy_client, ssh_allow_agent=False) client._get_ssh_connection(sleep=1) aa_mock.assert_has_calls([mock.call(), mock.call()]) proxy_client_mock.set_missing_host_key_policy.assert_called_once_with( mock.sentinel.aa) proxy_expected_connect = [mock.call( 'proxy-host', port=22, username='proxy-user', pkey=None, key_filename=None, look_for_keys=False, timeout=10.0, password=None, sock=None, allow_agent=True )] self.assertEqual(proxy_expected_connect, proxy_client_mock.connect.mock_calls) client_mock.set_missing_host_key_policy.assert_called_once_with( mock.sentinel.aa) expected_connect = [mock.call( 'localhost', port=22, username='root', pkey=None, key_filename=None, look_for_keys=False, timeout=10.0, password=None, sock=proxy_client_mock.get_transport().open_session(), allow_agent=False )] self.assertEqual(expected_connect, client_mock.connect.mock_calls) self.assertEqual(0, s_mock.call_count) @mock.patch('time.sleep') def test_get_ssh_connection_two_attemps(self, sleep_mock): c_mock, aa_mock, client_mock = self._set_ssh_connection_mocks() c_mock.return_value = client_mock client_mock.connect.side_effect = [ socket.error, mock.MagicMock() ] client = ssh.Client('localhost', 'root', timeout=1) client._get_ssh_connection(sleep=1) # We slept 2 seconds: because sleep is "1" and backoff is "1" too sleep_mock.assert_called_once_with(2) self.assertEqual(2, client_mock.connect.call_count) def test_get_ssh_connection_timeout(self): c_mock, aa_mock, client_mock = self._set_ssh_connection_mocks() timeout = 2 time_mock = self.patch('time.time') time_mock.side_effect = utils.generate_timeout_series(timeout + 1) c_mock.return_value = client_mock client_mock.connect.side_effect = [ socket.error, socket.error, socket.error, ] client = ssh.Client('localhost', 'root', timeout=timeout) # We need to mock LOG here because LOG.info() calls time.time() # in order to preprend a timestamp. with mock.patch.object(ssh, 'LOG'): self.assertRaises(exceptions.SSHTimeout, client._get_ssh_connection) # time.time() should be called twice, first to start the timer # and then to compute the timedelta self.assertEqual(2, time_mock.call_count) @mock.patch('select.POLLIN', SELECT_POLLIN, create=True) def test_timeout_in_exec_command(self): chan_mock, poll_mock, _, _ = ( self._set_mocks_for_select([0, 0, 0], True)) # Test for a timeout condition immediately raised client = ssh.Client('localhost', 'root', timeout=2) with testtools.ExpectedException(exceptions.TimeoutException): client.exec_command("test") chan_mock.fileno.assert_called_once_with() chan_mock.exec_command.assert_called_once_with("test") chan_mock.shutdown_write.assert_called_once_with() poll_mock.register.assert_called_once_with( chan_mock, self.SELECT_POLLIN) poll_mock.poll.assert_called_once_with(10) @mock.patch('select.POLLIN', SELECT_POLLIN, create=True) def test_exec_command(self): chan_mock, poll_mock, select_mock, client_mock = ( self._set_mocks_for_select([[1, 0, 0]], True)) chan_mock.recv_exit_status.return_value = 0 chan_mock.recv.return_value = b'' chan_mock.recv_stderr.return_value = b'' client = ssh.Client('localhost', 'root', timeout=2) client.exec_command("test") chan_mock.fileno.assert_called_once_with() chan_mock.exec_command.assert_called_once_with("test") chan_mock.shutdown_write.assert_called_once_with() select_mock.assert_called_once_with() poll_mock.register.assert_called_once_with( chan_mock, self.SELECT_POLLIN) poll_mock.poll.assert_called_once_with(10) chan_mock.recv_ready.assert_called_once_with() chan_mock.recv.assert_called_once_with(1024) chan_mock.recv_stderr_ready.assert_called_once_with() chan_mock.recv_stderr.assert_called_once_with(1024) chan_mock.recv_exit_status.assert_called_once_with() client_mock.close.assert_called_once_with() def _set_mocks_for_select(self, poll_data, ito_value=False): gsc_mock = self.patch('tempest.lib.common.ssh.Client.' '_get_ssh_connection') ito_mock = self.patch('tempest.lib.common.ssh.Client._is_timed_out') csp_mock = self.patch( 'tempest.lib.common.ssh.Client._can_system_poll') csp_mock.return_value = True select_mock = self.patch('select.poll', create=True) client_mock = mock.MagicMock() tran_mock = mock.MagicMock() chan_mock = mock.MagicMock() poll_mock = mock.MagicMock() select_mock.return_value = poll_mock gsc_mock.return_value = client_mock ito_mock.return_value = ito_value client_mock.get_transport.return_value = tran_mock tran_mock.open_session().__enter__.return_value = chan_mock if isinstance(poll_data[0], list): poll_mock.poll.side_effect = poll_data else: poll_mock.poll.return_value = poll_data return chan_mock, poll_mock, select_mock, client_mock _utf8_string = chr(1071) _utf8_bytes = _utf8_string.encode("utf-8") @mock.patch('select.POLLIN', SELECT_POLLIN, create=True) def test_exec_good_command_output(self): chan_mock, poll_mock, _, _ = ( self._set_mocks_for_select([1, 0, 0])) closed_prop = mock.PropertyMock(return_value=True) type(chan_mock).closed = closed_prop chan_mock.recv_exit_status.return_value = 0 chan_mock.recv.side_effect = [self._utf8_bytes[0:1], self._utf8_bytes[1:], b'R', b''] chan_mock.recv_stderr.return_value = b'' client = ssh.Client('localhost', 'root', timeout=2) out_data = client.exec_command("test") self.assertEqual(self._utf8_string + 'R', out_data) @mock.patch('select.POLLIN', SELECT_POLLIN, create=True) def test_exec_bad_command_output(self): chan_mock, poll_mock, _, _ = ( self._set_mocks_for_select([1, 0, 0])) closed_prop = mock.PropertyMock(return_value=True) type(chan_mock).closed = closed_prop chan_mock.recv_exit_status.return_value = 1 chan_mock.recv.return_value = b'' chan_mock.recv_stderr.side_effect = [b'R', self._utf8_bytes[0:1], self._utf8_bytes[1:], b''] client = ssh.Client('localhost', 'root', timeout=2) exc = self.assertRaises(exceptions.SSHExecCommandFailed, client.exec_command, "test") self.assertIn('R' + self._utf8_string, str(exc)) def test_exec_command_no_select(self): gsc_mock = self.patch('tempest.lib.common.ssh.Client.' '_get_ssh_connection') csp_mock = self.patch( 'tempest.lib.common.ssh.Client._can_system_poll') csp_mock.return_value = False select_mock = self.patch('select.poll', create=True) client_mock = mock.MagicMock() tran_mock = mock.MagicMock() chan_mock = mock.MagicMock() # Test for proper reading of STDOUT and STDERROR gsc_mock.return_value = client_mock client_mock.get_transport.return_value = tran_mock tran_mock.open_session().__enter__.return_value = chan_mock chan_mock.recv_exit_status.return_value = 0 std_out_mock = mock.MagicMock(StringIO) std_err_mock = mock.MagicMock(StringIO) chan_mock.makefile.return_value = std_out_mock chan_mock.makefile_stderr.return_value = std_err_mock client = ssh.Client('localhost', 'root', timeout=2) client.exec_command("test") chan_mock.makefile.assert_called_once_with('rb', 1024) chan_mock.makefile_stderr.assert_called_once_with('rb', 1024) std_out_mock.read.assert_called_once_with() std_err_mock.read.assert_called_once_with() self.assertFalse(select_mock.called)