How to use wait_for_ping method in tempest

Best Python code snippet using tempest_python

instanceha.py

Source:instanceha.py Github

copy

Full Screen

...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 )...

Full Screen

Full Screen

test_state.py

Source:test_state.py Github

copy

Full Screen

...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():...

Full Screen

Full Screen

test_commands.py

Source:test_commands.py Github

copy

Full Screen

...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")...

Full Screen

Full Screen

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.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run tempest automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful