KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use wait_for_ping method in tempest

Best Python code snippet using tempest_python

instanceha.py

Source:instanceha.py

...158            #raise exceptions.InvalidHostException()159    else:160        try:161            if self.wait_for_ping:162               self._wait_for_ping(fip["ip"])163        except exceptions.TimeoutException:164            console_logs = self._get_server_console_output(server,165                                                               None)166            LOG.debug("VM console logs:\n%s", console_logs)167            raise168def recover_instance_ha(self, image, flavor, computes,169                        volume_args=None,170                        floating_network=None,171                        use_floating_ip=True,172                        force_delete=False,173                        stop_instance=False,174                        disruptor_cmd=None,175                        iterations=1,176                        wait_for_ping=True,177                        max_log_length=None,178                        **kwargs):179    """Boot a server, trigger failover of host and verify instance.180    :param image: glance image name to use for the vm181    :param flavor: VM flavor name182    :param computes: dictionary with credentials to the compute nodes183        consisting of username, password, port, key_filename, disruptor184        command and pkey.185        Examples::186            computes: {187              username: heat-admin,188              key_filename: /path/to/ssh/id_rsa.pub189              port: 22190            }191    :param volume_args: volume args for booting server from volume192    :param floating_network: external network name, for floating ip193    :param use_floating_ip: bool, floating or fixed IP for SSH connection194    :param force_delete: whether to use force_delete for servers195    :param stop_instance: whether to stop instance before disruptor command196    :param disruptor_cmd: command to be send to hosting compute node197    :param iterations: number of compute node killing iteration198    :param wait_for_ping: whether to check connectivity on server creation199    :param **kwargs: extra arguments for booting the server200    :param max_log_length: The number of tail nova console-log lines user201                           would like to retrieve202    :returns:203    """204    self.wait_for_ping = wait_for_ping205    if volume_args:206        volume = self.cinder.create_volume(volume_args["size"], imageRef=None)207        kwargs["block_device_mapping"] = {"vdrally": "%s:::1" % volume.id}208    server, fip = self._boot_server_with_fip(209        image, flavor, use_floating_ip=use_floating_ip,210        floating_network=floating_network,211        key_name=self.context["user"]["keypair"]["name"],212        **kwargs)213    task_utils.wait_for(214        server,215        ready_statuses=["ACTIVE"],216        update_resource=task_utils.get_from_manager(),217        timeout=120,218        check_interval=2219    )220    try:221        if self.wait_for_ping:222            self._wait_for_ping(fip["ip"])223    except exceptions.TimeoutException:224        console_logs = self._get_server_console_output(server,225                                                       max_log_length)226        LOG.debug("VM console logs:\n%s", console_logs)227        raise228    if stop_instance:229        self._stop_server(server)230        task_utils.wait_for(231            server,232            ready_statuses=["SHUTOFF"],233            update_resource=task_utils.get_from_manager(),234            timeout=120,235            check_interval=2236        )...

test_state.py

Source:test_state.py

...36            â+----------+ââââââââââââââââ+---------+â37            '''38            def set_state(self, old_state, new_state):39                sequence_diagram[self.name].append(new_state.__name__)40            def wait_for_ping(self):41                '''42                    Wait for an external client to send a 'ping' event to A43                '''44                pass45            def wait_for_pong(self, b):46                b.send('ping', a)47            def __init__(self, name):48                self.name = name49                StateMachine.__init__(self, self.wait_for_ping)50                self.add_event_transitions('ping', {51                    self.wait_for_ping:  self.wait_for_pong52                })53                self.add_event_transitions('pong', {54                    self.wait_for_pong:  self.wait_for_ping55                })56        class B(StateMachine):57            '''58                     +---------+59                     |         |60                  +--+-----+   |61                  |  Wait  |   |62                  |  For   <---+63                  |  Ping  |64                  +--------+65            '''66            def set_state(self, old_state, new_state):67                sequence_diagram[self.name].append(new_state.__name__)68            def wait_for_ping(self, a):69                a.send('pong')70            def __init__(self, name):71                self.name = name72                StateMachine.__init__(self, self.wait_for_ping)73                self.add_event_transitions('ping', {74                    self.wait_for_ping:  self.wait_for_ping75                })76        a = A('a')77        b = B('b')78        with ManagedState():79            a.send('ping', b)80        test.assertEqual(sequence_diagram['a'], ['wait_for_pong', 'wait_for_ping'])81    def test_bad_initial_state(test):82        def foo():...

test_commands.py

Source:test_commands.py

...89        since = datetime(2016, 1, 4).date()90        until = datetime(2016, 1, 5).date()91        with capture_logs(logger) as stream:92            call_command('send_report')93            wait_for_ping()94        mail_mock.send_move_report.assert_called_once_with((since, until))95        assert 'INFO - Success sending report for' in stream[0]96    @patch('idv.mover.tasks.mover_mail')97    @patch('idv.mover.commands.move_credential_files')98    @freeze_time('2016-01-05')99    def test_command_send_email_for_given_dates(self, move_mock, mail_mock):100        move()101        since = datetime(2016, 1, 2).date()102        until = datetime(2016, 1, 3).date()103        call_command('send_report', since=since, until=until)104        wait_for_ping()105        mail_mock.send_move_report.assert_called_once_with((since, until))106    @patch('idv.mover.tasks.mover_mail')107    @patch('idv.mover.commands.move_credential_files')108    def test_value_error_if_no_move_checkpoint(self, move_mock, mail_mock):109        since = timezone.now().date()110        until = since + timedelta(days=1)111        call_command('send_report', since=since, until=until)112        wait_for_ping()113        assert not mail_mock.called114    @patch('idv.mover.tasks.mover_mail')115    @patch('idv.mover.commands.move_credential_files')116    def test_retries_if_checkpoint_older_than_report_end_date(117            self, move_mock, mail_mock):118        with freeze_time('2016-01-05'):119            move()120        since = datetime(2016, 1, 4).date()121        until = datetime(2016, 1, 6).date()122        call_command('send_report', since=since, until=until)123        wait_for_ping()124        assert not mail_mock.called125    @patch('idv.mover.tasks.mover_mail')126    @patch('idv.mover.commands.move_credential_files')127    def test_retries_if_move_didnt_finish_on_time(128            self, move_mock, mail_mock, settings):129        settings.SEND_REPORT_RETRY_TIME = 1130        with freeze_time('2016-01-05'):131            move()132        since = datetime(2016, 1, 4).date()133        until = datetime(2016, 1, 6).date()134        call_command('send_report', since=since, until=until)135        wait_for_ping()136        with freeze_time('2016-01-06'):137            move()138        wait_for_ping()139        mail_mock.send_move_report.assert_called_once_with((since, until))140    @patch('idv.mover.tasks.mover_mail')141    @patch('idv.mover.commands.move_credential_files')142    def test_report_sending(self, move_mock, mail_mock):143        with freeze_time('2016-01-05'):144            move()145        since = datetime(2016, 1, 4).date()146        until = datetime(2016, 1, 5).date()147        send_move_report("2016-01-04", "2016-01-05")...

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.