KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use wait_for_server_floating_ip method in tempest
Best Python code snippet using tempest_python
waiters.py
Source:waiters.py 
...447 message = ('Interface %s failed to detach from server %s within '448 'the required time (%s s)' % (port_id, server_id,449 client.build_timeout))450 raise lib_exc.TimeoutException(message)451def wait_for_server_floating_ip(servers_client, server, floating_ip,452 wait_for_disassociate=False):453 """Wait for floating IP association or disassociation.454 :param servers_client: The servers client to use when querying the server's455 floating IPs.456 :param server: The server JSON dict on which to wait.457 :param floating_ip: The floating IP JSON dict on which to wait.458 :param wait_for_disassociate: Boolean indiating whether to wait for459 disassociation instead of association.460 """461 def _get_floating_ip_in_server_addresses(floating_ip, server):462 for addresses in server['addresses'].values():463 for address in addresses:464 if (465 address['OS-EXT-IPS:type'] == 'floating' and...
test_waiters.py
Source:test_waiters.py
1# Copyright 2014 IBM Corp.2#3# Licensed under the Apache License, Version 2.0 (the "License"); you may4# not use this file except in compliance with the License. You may obtain5# a copy of the License at6#7# http://www.apache.org/licenses/LICENSE-2.08#9# Unless required by applicable law or agreed to in writing, software10# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT11# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the12# License for the specific language governing permissions and limitations13# under the License.14import time15from unittest import mock16from oslo_utils.fixture import uuidsentinel as uuids17from tempest.common import waiters18from tempest import exceptions19from tempest.lib import exceptions as lib_exc20from tempest.lib.services.compute import servers_client21from tempest.lib.services.volume.v2 import volumes_client22from tempest.tests import base23import tempest.tests.utils as utils24class TestImageWaiters(base.TestCase):25 def setUp(self):26 super(TestImageWaiters, self).setUp()27 self.client = mock.MagicMock()28 self.client.build_timeout = 129 self.client.build_interval = 130 def test_wait_for_image_status(self):31 self.client.show_image.return_value = ({'status': 'active'})32 start_time = int(time.time())33 waiters.wait_for_image_status(self.client, 'fake_image_id', 'active')34 end_time = int(time.time())35 # Ensure waiter returns before build_timeout36 self.assertLess((end_time - start_time), 10)37 def test_wait_for_image_status_timeout(self):38 time_mock = self.patch('time.time')39 time_mock.side_effect = utils.generate_timeout_series(1)40 self.client.show_image.return_value = ({'status': 'saving'})41 self.assertRaises(lib_exc.TimeoutException,42 waiters.wait_for_image_status,43 self.client, 'fake_image_id', 'active')44 def test_wait_for_image_status_error_on_image_create(self):45 self.client.show_image.return_value = ({'status': 'ERROR'})46 self.assertRaises(exceptions.AddImageException,47 waiters.wait_for_image_status,48 self.client, 'fake_image_id', 'active')49 def test_wait_for_image_imported_to_stores(self):50 self.client.show_image.return_value = ({'status': 'active',51 'stores': 'fake_store'})52 start_time = int(time.time())53 waiters.wait_for_image_imported_to_stores(54 self.client, 'fake_image_id', 'fake_store')55 end_time = int(time.time())56 # Ensure waiter returns before build_timeout57 self.assertLess((end_time - start_time), 10)58 def test_wait_for_image_imported_to_stores_failure(self):59 time_mock = self.patch('time.time')60 client = mock.MagicMock()61 client.build_timeout = 262 self.patch('time.time', side_effect=[0., 1., 2.])63 time_mock.side_effect = utils.generate_timeout_series(1)64 client.show_image.return_value = ({65 'status': 'saving',66 'stores': 'fake_store',67 'os_glance_failed_import': 'fake_os_glance_failed_import'})68 self.assertRaises(lib_exc.OtherRestClientException,69 waiters.wait_for_image_imported_to_stores,70 client, 'fake_image_id', 'fake_store')71 def test_wait_for_image_imported_to_stores_timeout(self):72 time_mock = self.patch('time.time')73 client = mock.MagicMock()74 client.build_timeout = 275 self.patch('time.time', side_effect=[0., 1., 2.])76 time_mock.side_effect = utils.generate_timeout_series(1)77 client.show_image.return_value = ({78 'status': 'saving',79 'stores': 'fake_store'})80 self.assertRaises(lib_exc.TimeoutException,81 waiters.wait_for_image_imported_to_stores,82 client, 'fake_image_id', 'fake_store')83 def test_wait_for_image_copied_to_stores(self):84 self.client.show_image.return_value = ({85 'status': 'active',86 'os_glance_importing_to_stores': '',87 'os_glance_failed_import': 'fake_os_glance_failed_import'})88 start_time = int(time.time())89 waiters.wait_for_image_copied_to_stores(90 self.client, 'fake_image_id')91 end_time = int(time.time())92 # Ensure waiter returns before build_timeout93 self.assertLess((end_time - start_time), 10)94 def test_wait_for_image_copied_to_stores_timeout(self):95 time_mock = self.patch('time.time')96 self.patch('time.time', side_effect=[0., 1.])97 time_mock.side_effect = utils.generate_timeout_series(1)98 self.client.show_image.return_value = ({99 'status': 'active',100 'os_glance_importing_to_stores': 'processing',101 'os_glance_failed_import': 'fake_os_glance_failed_import'})102 self.assertRaises(lib_exc.TimeoutException,103 waiters.wait_for_image_copied_to_stores,104 self.client, 'fake_image_id')105 def test_wait_for_image_tasks_status(self):106 self.client.show_image_tasks.return_value = ({107 'tasks': [{'status': 'success'}]})108 start_time = int(time.time())109 waiters.wait_for_image_tasks_status(110 self.client, 'fake_image_id', 'success')111 end_time = int(time.time())112 # Ensure waiter returns before build_timeout113 self.assertLess((end_time - start_time), 10)114 def test_wait_for_image_tasks_status_timeout(self):115 time_mock = self.patch('time.time')116 self.patch('time.time', side_effect=[0., 1.])117 time_mock.side_effect = utils.generate_timeout_series(1)118 self.client.show_image_tasks.return_value = ({119 'tasks': [120 {'status': 'success'},121 {'status': 'processing'}]})122 self.assertRaises(lib_exc.TimeoutException,123 waiters.wait_for_image_tasks_status,124 self.client, 'fake_image_id', 'success')125class TestInterfaceWaiters(base.TestCase):126 build_timeout = 1.127 build_interval = 1128 port_down = {'interfaceAttachment': {'port_state': 'DOWN'}}129 port_active = {'interfaceAttachment': {'port_state': 'ACTIVE'}}130 def mock_client(self, **kwargs):131 return mock.MagicMock(132 build_timeout=self.build_timeout,133 build_interval=self.build_interval,134 **kwargs)135 def test_wait_for_interface_status(self):136 show_interface = mock.Mock(137 side_effect=[self.port_down, self.port_active])138 client = self.mock_client(show_interface=show_interface)139 self.patch('time.time', return_value=0.)140 sleep = self.patch('time.sleep')141 result = waiters.wait_for_interface_status(142 client, 'server_id', 'port_id', 'ACTIVE')143 self.assertIs(self.port_active['interfaceAttachment'], result)144 show_interface.assert_has_calls([mock.call('server_id', 'port_id'),145 mock.call('server_id', 'port_id')])146 sleep.assert_called_once_with(client.build_interval)147 def test_wait_for_interface_status_timeout(self):148 show_interface = mock.MagicMock(return_value=self.port_down)149 client = self.mock_client(show_interface=show_interface)150 self.patch('time.time', side_effect=[0., client.build_timeout + 1.])151 sleep = self.patch('time.sleep')152 self.assertRaises(lib_exc.TimeoutException,153 waiters.wait_for_interface_status,154 client, 'server_id', 'port_id', 'ACTIVE')155 show_interface.assert_has_calls([mock.call('server_id', 'port_id'),156 mock.call('server_id', 'port_id')])157 sleep.assert_called_once_with(client.build_interval)158 def test_wait_for_interface_detach(self):159 no_event = {160 'instanceAction': {161 'events': []162 }163 }164 one_event_without_result = {165 'instanceAction': {166 'events': [167 {168 'event': 'compute_detach_interface',169 'result': None170 }171 ]172 }173 }174 one_event_successful = {175 'instanceAction': {176 'events': [177 {178 'event': 'compute_detach_interface',179 'result': 'Success'180 }181 ]182 }183 }184 show_instance_action = mock.MagicMock(185 # there is an extra call to return the result from the waiter186 side_effect=[187 no_event,188 one_event_without_result,189 one_event_successful,190 one_event_successful,191 ]192 )193 client = self.mock_client(show_instance_action=show_instance_action)194 self.patch('time.time', return_value=0.)195 sleep = self.patch('time.sleep')196 result = waiters.wait_for_interface_detach(197 client, mock.sentinel.server_id, mock.sentinel.port_id,198 mock.sentinel.detach_request_id199 )200 self.assertIs(one_event_successful['instanceAction'], result)201 show_instance_action.assert_has_calls(202 # there is an extra call to return the result from the waiter203 [204 mock.call(205 mock.sentinel.server_id, mock.sentinel.detach_request_id)206 ] * 4207 )208 sleep.assert_has_calls([mock.call(client.build_interval)] * 2)209 def test_wait_for_interface_detach_timeout(self):210 one_event_without_result = {211 'instanceAction': {212 'events': [213 {214 'event': 'compute_detach_interface',215 'result': None216 }217 ]218 }219 }220 show_instance_action = mock.MagicMock(221 return_value=one_event_without_result)222 client = self.mock_client(show_instance_action=show_instance_action)223 self.patch('time.time', side_effect=[0., client.build_timeout + 1.])224 sleep = self.patch('time.sleep')225 self.assertRaises(226 lib_exc.TimeoutException,227 waiters.wait_for_interface_detach,228 client, mock.sentinel.server_id, mock.sentinel.port_id,229 mock.sentinel.detach_request_id230 )231 show_instance_action.assert_has_calls(232 [233 mock.call(234 mock.sentinel.server_id, mock.sentinel.detach_request_id)235 ] * 2236 )237 sleep.assert_called_once_with(client.build_interval)238class TestVolumeWaiters(base.TestCase):239 vol_migrating_src_host = {240 'volume': {'migration_status': 'migrating',241 'os-vol-host-attr:host': 'src_host@backend#type'}}242 vol_migrating_dst_host = {243 'volume': {'migration_status': 'migrating',244 'os-vol-host-attr:host': 'dst_host@backend#type'}}245 vol_migration_success = {246 'volume': {'migration_status': 'success',247 'os-vol-host-attr:host': 'dst_host@backend#type'}}248 vol_migration_error = {249 'volume': {'migration_status': 'error',250 'os-vol-host-attr:host': 'src_host@backend#type'}}251 def test_wait_for_volume_migration_timeout(self):252 show_volume = mock.MagicMock(return_value=self.vol_migrating_src_host)253 client = mock.Mock(spec=volumes_client.VolumesClient,254 resource_type="volume",255 build_interval=1,256 build_timeout=1,257 show_volume=show_volume)258 self.patch('time.time', side_effect=[0., client.build_timeout + 1.])259 self.patch('time.sleep')260 self.assertRaises(lib_exc.TimeoutException,261 waiters.wait_for_volume_migration,262 client, mock.sentinel.volume_id, 'dst_host')263 def test_wait_for_volume_migration_error(self):264 show_volume = mock.MagicMock(side_effect=[265 self.vol_migrating_src_host,266 self.vol_migrating_src_host,267 self.vol_migration_error])268 client = mock.Mock(spec=volumes_client.VolumesClient,269 resource_type="volume",270 build_interval=1,271 build_timeout=1,272 show_volume=show_volume)273 self.patch('time.time', return_value=0.)274 self.patch('time.sleep')275 self.assertRaises(lib_exc.TempestException,276 waiters.wait_for_volume_migration,277 client, mock.sentinel.volume_id, 'dst_host')278 def test_wait_for_volume_migration_success_and_dst(self):279 show_volume = mock.MagicMock(side_effect=[280 self.vol_migrating_src_host,281 self.vol_migrating_dst_host,282 self.vol_migration_success])283 client = mock.Mock(spec=volumes_client.VolumesClient,284 resource_type="volume",285 build_interval=1,286 build_timeout=1,287 show_volume=show_volume)288 self.patch('time.time', return_value=0.)289 self.patch('time.sleep')290 waiters.wait_for_volume_migration(291 client, mock.sentinel.volume_id, 'dst_host')292 # Assert that we wait until migration_status is success and dst_host is293 # part of the returned os-vol-host-attr:host.294 show_volume.assert_has_calls([mock.call(mock.sentinel.volume_id),295 mock.call(mock.sentinel.volume_id),296 mock.call(mock.sentinel.volume_id)])297 @mock.patch.object(time, 'sleep')298 def test_wait_for_volume_status_error_restoring(self, mock_sleep):299 # Tests that the wait method raises VolumeRestoreErrorException if300 # the volume status is 'error_restoring'.301 client = mock.Mock(spec=volumes_client.VolumesClient,302 resource_type="volume",303 build_interval=1)304 volume1 = {'volume': {'status': 'restoring-backup'}}305 volume2 = {'volume': {'status': 'error_restoring'}}306 mock_show = mock.Mock(side_effect=(volume1, volume2))307 client.show_volume = mock_show308 volume_id = '7532b91e-aa0a-4e06-b3e5-20c0c5ee1caa'309 self.assertRaises(exceptions.VolumeRestoreErrorException,310 waiters.wait_for_volume_resource_status,311 client, volume_id, 'available')312 mock_show.assert_has_calls([mock.call(volume_id),313 mock.call(volume_id)])314 mock_sleep.assert_called_once_with(1)315 @mock.patch.object(time, 'sleep')316 def test_wait_for_volume_status_error_extending(self, mock_sleep):317 # Tests that the wait method raises VolumeExtendErrorException if318 # the volume status is 'error_extending'.319 client = mock.Mock(spec=volumes_client.VolumesClient,320 resource_type="volume",321 build_interval=1)322 volume1 = {'volume': {'status': 'extending'}}323 volume2 = {'volume': {'status': 'error_extending'}}324 mock_show = mock.Mock(side_effect=(volume1, volume2))325 client.show_volume = mock_show326 volume_id = '7532b91e-aa0a-4e06-b3e5-20c0c5ee1caa'327 self.assertRaises(exceptions.VolumeExtendErrorException,328 waiters.wait_for_volume_resource_status,329 client, volume_id, 'available')330 mock_show.assert_has_calls([mock.call(volume_id),331 mock.call(volume_id)])332 mock_sleep.assert_called_once_with(1)333 def test_wait_for_volume_attachment_create(self):334 vol_detached = {'volume': {'attachments': []}}335 vol_attached = {'volume': {'attachments': [336 {'id': uuids.volume_id,337 'attachment_id': uuids.attachment_id,338 'server_id': uuids.server_id,339 'volume_id': uuids.volume_id}]}}340 show_volume = mock.MagicMock(side_effect=[341 vol_detached, vol_detached, vol_attached])342 client = mock.Mock(spec=volumes_client.VolumesClient,343 build_interval=1,344 build_timeout=5,345 show_volume=show_volume)346 self.patch('time.time')347 self.patch('time.sleep')348 att = waiters.wait_for_volume_attachment_create(349 client, uuids.volume_id, uuids.server_id)350 assert att == vol_attached['volume']['attachments'][0]351 # Assert that show volume is called until the attachment is removed.352 show_volume.assert_has_calls([mock.call(uuids.volume_id),353 mock.call(uuids.volume_id),354 mock.call(uuids.volume_id)])355 def test_wait_for_volume_attachment(self):356 vol_detached = {'volume': {'attachments': []}}357 vol_attached = {'volume': {'attachments': [358 {'attachment_id': uuids.attachment_id}]}}359 show_volume = mock.MagicMock(side_effect=[360 vol_attached, vol_attached, vol_detached])361 client = mock.Mock(spec=volumes_client.VolumesClient,362 build_interval=1,363 build_timeout=5,364 show_volume=show_volume)365 self.patch('time.time')366 self.patch('time.sleep')367 waiters.wait_for_volume_attachment_remove(client, uuids.volume_id,368 uuids.attachment_id)369 # Assert that show volume is called until the attachment is removed.370 show_volume.assert_has_calls([mock.call(uuids.volume_id),371 mock.call(uuids.volume_id),372 mock.call(uuids.volume_id)])373 def test_wait_for_volume_attachment_timeout(self):374 show_volume = mock.MagicMock(return_value={375 'volume': {'attachments': [376 {'attachment_id': uuids.attachment_id}]}})377 client = mock.Mock(spec=volumes_client.VolumesClient,378 build_interval=1,379 build_timeout=1,380 show_volume=show_volume)381 self.patch('time.time', side_effect=[0., client.build_timeout + 1.])382 self.patch('time.sleep')383 # Assert that a timeout is raised if the attachment remains.384 self.assertRaises(lib_exc.TimeoutException,385 waiters.wait_for_volume_attachment_remove,386 client, uuids.volume_id, uuids.attachment_id)387 def test_wait_for_volume_attachment_not_present(self):388 show_volume = mock.MagicMock(return_value={389 'volume': {'attachments': []}})390 client = mock.Mock(spec=volumes_client.VolumesClient,391 build_interval=1,392 build_timeout=1,393 show_volume=show_volume)394 self.patch('time.time', side_effect=[0., client.build_timeout + 1.])395 self.patch('time.sleep')396 waiters.wait_for_volume_attachment_remove(client, uuids.volume_id,397 uuids.attachment_id)398 # Assert that show volume is only called once before we return399 show_volume.assert_called_once_with(uuids.volume_id)400 def test_wait_for_volume_attachment_remove_from_server(self):401 volume_attached = {402 "volumeAttachments": [{"volumeId": uuids.volume_id}]}403 volume_not_attached = {"volumeAttachments": []}404 mock_list_volume_attachments = mock.Mock(405 side_effect=[volume_attached, volume_not_attached])406 mock_client = mock.Mock(407 spec=servers_client.ServersClient,408 build_interval=1,409 build_timeout=1,410 list_volume_attachments=mock_list_volume_attachments)411 self.patch(412 'time.time',413 side_effect=[0., 0.5, mock_client.build_timeout + 1.])414 self.patch('time.sleep')415 waiters.wait_for_volume_attachment_remove_from_server(416 mock_client, uuids.server_id, uuids.volume_id)417 # Assert that list_volume_attachments is called until the attachment is418 # removed.419 mock_list_volume_attachments.assert_has_calls([420 mock.call(uuids.server_id),421 mock.call(uuids.server_id)])422 def test_wait_for_volume_attachment_remove_from_server_timeout(self):423 volume_attached = {424 "volumeAttachments": [{"volumeId": uuids.volume_id}]}425 mock_list_volume_attachments = mock.Mock(426 side_effect=[volume_attached, volume_attached])427 mock_get_console_output = mock.Mock(428 return_value={'output': 'output'})429 mock_client = mock.Mock(430 spec=servers_client.ServersClient,431 build_interval=1,432 build_timeout=1,433 list_volume_attachments=mock_list_volume_attachments,434 get_console_output=mock_get_console_output)435 self.patch(436 'time.time',437 side_effect=[0., 0.5, mock_client.build_timeout + 1.])438 self.patch('time.sleep')439 self.assertRaises(440 lib_exc.TimeoutException,441 waiters.wait_for_volume_attachment_remove_from_server,442 mock_client, uuids.server_id, uuids.volume_id)443 # Assert that list_volume_attachments is called until the attachment is444 # removed.445 mock_list_volume_attachments.assert_has_calls([446 mock.call(uuids.server_id),447 mock.call(uuids.server_id)])448 # Assert that we fetch console output449 mock_get_console_output.assert_called_once_with(uuids.server_id)450 def test_wait_for_volume_attachment_remove_from_server_not_found(self):451 mock_list_volume_attachments = mock.Mock(452 side_effect=lib_exc.NotFound)453 mock_client = mock.Mock(454 spec=servers_client.ServersClient,455 list_volume_attachments=mock_list_volume_attachments)456 # Assert that nothing is raised when lib_exc_NotFound is raised457 # by the client call to list_volume_attachments458 waiters.wait_for_volume_attachment_remove_from_server(459 mock_client, mock.sentinel.server_id, mock.sentinel.volume_id)460 # Assert that list_volume_attachments was actually called461 mock_list_volume_attachments.assert_called_once_with(462 mock.sentinel.server_id)463 @mock.patch('os.system')464 def test_wait_for_ping_host_alive(self, mock_ping):465 mock_ping.return_value = 0466 # Assert that nothing is raised as the host is alive467 waiters.wait_for_ping('127.0.0.1', 10, 1)468 @mock.patch('os.system')469 def test_wait_for_ping_host_eventually_alive(self, mock_ping):470 mock_ping.side_effect = [1, 1, 0]471 # Assert that nothing is raised when the host is eventually alive472 waiters.wait_for_ping('127.0.0.1', 10, 1)473 @mock.patch('os.system')474 def test_wait_for_ping_timeout(self, mock_ping):475 mock_ping.return_value = 1476 # Assert that TimeoutException is raised when the host is dead477 self.assertRaises(478 lib_exc.TimeoutException,479 waiters.wait_for_ping,480 '127.0.0.1',481 .1,482 .1483 )484 def test_wait_for_ssh(self):485 mock_ssh_client = mock.Mock()486 mock_ssh_client.validate_authentication.return_value = True487 # Assert that nothing is raised when validate_authentication returns488 waiters.wait_for_ssh(mock_ssh_client, .1)489 mock_ssh_client.validate_authentication.assert_called_once()490 def test_wait_for_ssh_eventually_up(self):491 mock_ssh_client = mock.Mock()492 timeout = lib_exc.SSHTimeout(493 host='foo',494 username='bar',495 password='fizz'496 )497 mock_ssh_client.validate_authentication.side_effect = [498 timeout,499 timeout,500 True501 ]502 # Assert that nothing is raised if validate_authentication passes503 # before the timeout504 waiters.wait_for_ssh(mock_ssh_client, 10)505 def test_wait_for_ssh_timeout(self):506 mock_ssh_client = mock.Mock()507 timeout = lib_exc.SSHTimeout(508 host='foo',509 username='bar',510 password='fizz'511 )512 mock_ssh_client.validate_authentication.side_effect = timeout513 # Assert that TimeoutException is raised when validate_authentication514 # doesn't pass in time.515 self.assertRaises(516 lib_exc.TimeoutException,517 waiters.wait_for_ssh,518 mock_ssh_client,519 .1520 )521class TestServerFloatingIPWaiters(base.TestCase):522 def test_wait_for_server_floating_ip_associate_timeout(self):523 mock_server = {'server': {'id': 'fake_uuid', 'addresses': {}}}524 mock_client = mock.Mock(525 spec=servers_client.ServersClient,526 build_timeout=1, build_interval=1,527 show_server=lambda id: mock_server)528 fake_server = {'id': 'fake-uuid'}529 fake_fip = {'floating_ip_address': 'fake_address'}530 self.assertRaises(531 lib_exc.TimeoutException,532 waiters.wait_for_server_floating_ip, mock_client, fake_server,533 fake_fip)534 def test_wait_for_server_floating_ip_disassociate_timeout(self):535 mock_addresses = {'shared': [{'OS-EXT-IPS:type': 'floating',536 'addr': 'fake_address'}]}537 mock_server = {'server': {'id': 'fake_uuid',538 'addresses': mock_addresses}}539 mock_client = mock.Mock(540 spec=servers_client.ServersClient,541 build_timeout=1, build_interval=1,542 show_server=lambda id: mock_server)543 fake_server = {'id': 'fake-uuid'}544 fake_fip = {'floating_ip_address': 'fake_address'}545 self.assertRaises(546 lib_exc.TimeoutException,547 waiters.wait_for_server_floating_ip, mock_client, fake_server,...
test_minimum_basic.py
Source:test_minimum_basic.py
...106 CONF.network.floating_network_name):107 fip = self.create_floating_ip(server)108 floating_ip = self.associate_floating_ip(109 fip, server)110 waiters.wait_for_server_floating_ip(self.servers_client,111 server, floating_ip)112 ssh_ip = floating_ip['floating_ip_address']113 else:114 ssh_ip = self.get_server_ip(server)115 self.create_and_add_security_group_to_server(server)116 # check that we can SSH to the server before reboot117 self.linux_client = self.get_remote_client(118 ssh_ip, private_key=keypair['private_key'],119 server=server)120 self.nova_reboot(server)121 # check that we can SSH to the server after reboot122 # (both connections are part of the scenario)123 self.linux_client = self.get_remote_client(124 ssh_ip, private_key=keypair['private_key'],125 server=server)126 self.check_disks()127 if floating_ip:128 # delete the floating IP, this should refresh the server addresses129 self.disassociate_floating_ip(floating_ip)130 waiters.wait_for_server_floating_ip(131 self.servers_client, server, floating_ip,132 wait_for_disassociate=True)133 @decorators.idempotent_id('a8fd48ec-1d01-4895-b932-02321661ec1e')134 @testtools.skipUnless(CONF.volume_feature_enabled.snapshot,135 "Cinder volume snapshots are disabled")136 @utils.services('compute', 'volume', 'image', 'network')137 def test_minimum_basic_instance_hard_reboot_after_vol_snap_deletion(self):138 """Test compute hard reboot after volume snapshot deleted139 Steps:140 1. Create image141 2. Create keypair142 3. Boot instance with keypair and get list of instances143 4. Create volume and show list of volumes144 5. Attach volume to instance and getlist of volumes145 6. Create a snapshot from volume146 7. Add IP to instance147 8. Create and add security group to instance148 9. Check SSH connection to instance149 10. Write data timestamp to the attached volume150 11. Delete volume snapshot before reboot instance151 12. Reboot instance (HARD)152 13. Check SSH connection to instance after reboot153 14. Verify attached disk data timestamp post instance reboot154 """155 image = self.image_create()156 keypair = self.create_keypair()157 server = self.create_server(image_id=image, key_name=keypair['name'])158 servers = self.servers_client.list_servers()['servers']159 self.assertIn(server['id'], [x['id'] for x in servers])160 self.nova_show(server)161 volume = self.create_volume()162 volumes = self.volumes_client.list_volumes()['volumes']163 self.assertIn(volume['id'], [x['id'] for x in volumes])164 self.cinder_show(volume)165 volume = self.nova_volume_attach(server, volume)166 self.addCleanup(self.nova_volume_detach, server, volume)167 snapshot = self.create_volume_snapshot(volume['id'], force=True)168 self.cinder_show(volume)169 floating_ip = None170 server = self.servers_client.show_server(server['id'])['server']171 if (CONF.network_feature_enabled.floating_ips and172 CONF.network.floating_network_name):173 fip = self.create_floating_ip(server)174 floating_ip = self.associate_floating_ip(175 fip, server)176 waiters.wait_for_server_floating_ip(self.servers_client, server,177 floating_ip)178 ssh_ip = floating_ip['floating_ip_address']179 else:180 ssh_ip = self.get_server_ip(server)181 self.create_and_add_security_group_to_server(server)182 # check that we can SSH to the server before reboot183 self.linux_client = self.get_remote_client(184 ssh_ip, private_key=keypair['private_key'],185 server=server)186 # write data to the volume before reboot instance187 timestamp_before = self.create_timestamp(188 ssh_ip, private_key=keypair['private_key'], server=server)189 # delete the snapshot before rebooting the instance190 self.snapshots_client.delete_snapshot(snapshot['id'])191 self.snapshots_client.wait_for_resource_deletion(snapshot['id'])192 self.nova_reboot(server, hard=True)193 # check that we can SSH to the server after reboot194 # (both connections are part of the scenario)195 self.linux_client = self.get_remote_client(196 ssh_ip, private_key=keypair['private_key'],197 server=server)198 self.check_disks()199 timestamp_after = self.get_timestamp(200 ssh_ip, private_key=keypair['private_key'], server=server)201 self.assertEqual(timestamp_before, timestamp_after)202 if floating_ip:203 # delete the floating IP, this should refresh the server addresses204 self.disassociate_floating_ip(floating_ip)205 waiters.wait_for_server_floating_ip(206 self.servers_client, server, floating_ip,...
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:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
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!!
Was this article helpful?
