Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _set_mocks_for_select method in tempest

Best Python code snippet using tempest_python

test_ssh.py

Source:test_ssh.py

...95        self.assertLess((end_time - start_time), 5)96        self.assertGreaterEqual((end_time - start_time), 2)97    @mock.patch('select.POLLIN', SELECT_POLLIN, create=True)98    def test_timeout_in_exec_command(self):99        chan_mock, poll_mock, _ = self._set_mocks_for_select([0, 0, 0], True)100        # Test for a timeout condition immediately raised101        client = ssh.Client('localhost', 'root', timeout=2)102        with testtools.ExpectedException(exceptions.TimeoutException):103            client.exec_command("test")104        chan_mock.fileno.assert_called_once_with()105        chan_mock.exec_command.assert_called_once_with("test")106        chan_mock.shutdown_write.assert_called_once_with()107        poll_mock.register.assert_called_once_with(108            chan_mock, self.SELECT_POLLIN)109        poll_mock.poll.assert_called_once_with(10)110    @mock.patch('select.POLLIN', SELECT_POLLIN, create=True)111    def test_exec_command(self):112        chan_mock, poll_mock, select_mock = (113            self._set_mocks_for_select([[1, 0, 0]], True))114        closed_prop = mock.PropertyMock(return_value=True)115        type(chan_mock).closed = closed_prop116        chan_mock.recv_exit_status.return_value = 0117        chan_mock.recv.return_value = b''118        chan_mock.recv_stderr.return_value = b''119        client = ssh.Client('localhost', 'root', timeout=2)120        client.exec_command("test")121        chan_mock.fileno.assert_called_once_with()122        chan_mock.exec_command.assert_called_once_with("test")123        chan_mock.shutdown_write.assert_called_once_with()124        select_mock.assert_called_once_with()125        poll_mock.register.assert_called_once_with(126            chan_mock, self.SELECT_POLLIN)127        poll_mock.poll.assert_called_once_with(10)128        chan_mock.recv_ready.assert_called_once_with()129        chan_mock.recv.assert_called_once_with(1024)130        chan_mock.recv_stderr_ready.assert_called_once_with()131        chan_mock.recv_stderr.assert_called_once_with(1024)132        chan_mock.recv_exit_status.assert_called_once_with()133        closed_prop.assert_called_once_with()134    def _set_mocks_for_select(self, poll_data, ito_value=False):135        gsc_mock = self.patch('tempest_lib.common.ssh.Client.'136                              '_get_ssh_connection')137        ito_mock = self.patch('tempest_lib.common.ssh.Client._is_timed_out')138        csp_mock = self.patch(139            'tempest_lib.common.ssh.Client._can_system_poll')140        csp_mock.return_value = True141        select_mock = self.patch('select.poll', create=True)142        client_mock = mock.MagicMock()143        tran_mock = mock.MagicMock()144        chan_mock = mock.MagicMock()145        poll_mock = mock.MagicMock()146        select_mock.return_value = poll_mock147        gsc_mock.return_value = client_mock148        ito_mock.return_value = ito_value149        client_mock.get_transport.return_value = tran_mock150        tran_mock.open_session.return_value = chan_mock151        if isinstance(poll_data[0], list):152            poll_mock.poll.side_effect = poll_data153        else:154            poll_mock.poll.return_value = poll_data155        return chan_mock, poll_mock, select_mock156    _utf8_string = six.unichr(1071)157    _utf8_bytes = _utf8_string.encode("utf-8")158    @mock.patch('select.POLLIN', SELECT_POLLIN, create=True)159    def test_exec_good_command_output(self):160        chan_mock, poll_mock, _ = self._set_mocks_for_select([1, 0, 0])161        closed_prop = mock.PropertyMock(return_value=True)162        type(chan_mock).closed = closed_prop163        chan_mock.recv_exit_status.return_value = 0164        chan_mock.recv.side_effect = [self._utf8_bytes[0:1],165                                      self._utf8_bytes[1:], b'R', b'']166        chan_mock.recv_stderr.return_value = b''167        client = ssh.Client('localhost', 'root', timeout=2)168        out_data = client.exec_command("test")169        self.assertEqual(self._utf8_string + 'R', out_data)170    @mock.patch('select.POLLIN', SELECT_POLLIN, create=True)171    def test_exec_bad_command_output(self):172        chan_mock, poll_mock, _ = self._set_mocks_for_select([1, 0, 0])173        closed_prop = mock.PropertyMock(return_value=True)174        type(chan_mock).closed = closed_prop175        chan_mock.recv_exit_status.return_value = 1176        chan_mock.recv.return_value = b''177        chan_mock.recv_stderr.side_effect = [b'R', self._utf8_bytes[0:1],178                                             self._utf8_bytes[1:], b'']179        client = ssh.Client('localhost', 'root', timeout=2)180        exc = self.assertRaises(exceptions.SSHExecCommandFailed,181                                client.exec_command, "test")182        self.assertIn('R' + self._utf8_string, six.text_type(exc))183    def test_exec_command_no_select(self):184        gsc_mock = self.patch('tempest_lib.common.ssh.Client.'185                              '_get_ssh_connection')186        csp_mock = self.patch(...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.