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

How to use rescue_server method in tempest

Best Python code snippet using tempest_python

test_server_rescue.py

Source:test_server_rescue.py

...62        cls.servers_client.wait_for_server_status(cls.server_id, 'ACTIVE')63        # Server for negative tests64        cls.rescue_id = resc_server['id']65        cls.rescue_password = resc_server['adminPass']66        cls.servers_client.rescue_server(67            cls.rescue_id, cls.rescue_password)68        cls.servers_client.wait_for_server_status(cls.rescue_id, 'RESCUE')69    def setUp(self):70        super(ServerRescueTestJSON, self).setUp()71    @classmethod72    def tearDownClass(cls):73        # Deleting the floating IP which is created in this method74        cls.floating_ips_client.delete_floating_ip(cls.floating_ip_id)75        client = cls.volumes_extensions_client76        client.delete_volume(str(cls.volume_to_attach['id']).strip())77        client.delete_volume(str(cls.volume_to_detach['id']).strip())78        resp, cls.sg = cls.security_groups_client.delete_security_group(79            cls.sg_id)80        super(ServerRescueTestJSON, cls).tearDownClass()81    def tearDown(self):82        super(ServerRescueTestJSON, self).tearDown()83    def _detach(self, server_id, volume_id):84        self.servers_client.detach_volume(server_id, volume_id)85        self.volumes_extensions_client.wait_for_volume_status(volume_id,86                                                              'available')87    def _delete(self, volume_id):88        self.volumes_extensions_client.delete_volume(volume_id)89    def _unrescue(self, server_id):90        resp, body = self.servers_client.unrescue_server(server_id)91        self.assertEqual(202, resp.status)92        self.servers_client.wait_for_server_status(server_id, 'ACTIVE')93    def _unpause(self, server_id):94        resp, body = self.servers_client.unpause_server(server_id)95        self.assertEqual(202, resp.status)96        self.servers_client.wait_for_server_status(server_id, 'ACTIVE')97    @attr(type='smoke')98    def test_rescue_unrescue_instance(self):99        resp, body = self.servers_client.rescue_server(100            self.server_id, self.password)101        self.assertEqual(200, resp.status)102        self.servers_client.wait_for_server_status(self.server_id, 'RESCUE')103        resp, body = self.servers_client.unrescue_server(self.server_id)104        self.assertEqual(202, resp.status)105        self.servers_client.wait_for_server_status(self.server_id, 'ACTIVE')106    @attr(type=['negative', 'gate'])107    def test_rescue_paused_instance(self):108        # Rescue a paused server109        resp, body = self.servers_client.pause_server(110            self.server_id)111        self.addCleanup(self._unpause, self.server_id)112        self.assertEqual(202, resp.status)113        self.servers_client.wait_for_server_status(self.server_id, 'PAUSED')114        self.assertRaises(exceptions.Duplicate,115                          self.servers_client.rescue_server,116                          self.server_id)117    @attr(type=['negative', 'gate'])118    def test_rescued_vm_reboot(self):119        self.assertRaises(exceptions.Duplicate, self.servers_client.reboot,120                          self.rescue_id, 'HARD')121    @attr(type=['negative', 'gate'])122    def test_rescue_non_existent_server(self):123        # Rescue a non-existing server124        self.assertRaises(exceptions.NotFound,125                          self.servers_client.rescue_server,126                          '999erra43')127    @attr(type=['negative', 'gate'])128    def test_rescued_vm_rebuild(self):129        self.assertRaises(exceptions.Duplicate,130                          self.servers_client.rebuild,131                          self.rescue_id,132                          self.image_ref_alt)133    @attr(type=['negative', 'gate'])134    def test_rescued_vm_attach_volume(self):135        # Rescue the server136        self.servers_client.rescue_server(self.server_id, self.password)137        self.servers_client.wait_for_server_status(self.server_id, 'RESCUE')138        self.addCleanup(self._unrescue, self.server_id)139        # Attach the volume to the server140        self.assertRaises(exceptions.Duplicate,141                          self.servers_client.attach_volume,142                          self.server_id,143                          self.volume_to_attach['id'],144                          device='/dev/%s' % self.device)145    @attr(type=['negative', 'gate'])146    def test_rescued_vm_detach_volume(self):147        # Attach the volume to the server148        self.servers_client.attach_volume(self.server_id,149                                          self.volume_to_detach['id'],150                                          device='/dev/%s' % self.device)151        self.volumes_extensions_client.wait_for_volume_status(152            self.volume_to_detach['id'], 'in-use')153        # Rescue the server154        self.servers_client.rescue_server(self.server_id, self.password)155        self.servers_client.wait_for_server_status(self.server_id, 'RESCUE')156        # addCleanup is a LIFO queue157        self.addCleanup(self._detach, self.server_id,158                        self.volume_to_detach['id'])159        self.addCleanup(self._unrescue, self.server_id)160        # Detach the volume from the server expecting failure161        self.assertRaises(exceptions.Duplicate,162                          self.servers_client.detach_volume,163                          self.server_id,164                          self.volume_to_detach['id'])165    @attr(type='gate')166    def test_rescued_vm_associate_dissociate_floating_ip(self):167        # Rescue the server168        self.servers_client.rescue_server(169            self.server_id, self.password)170        self.servers_client.wait_for_server_status(self.server_id, 'RESCUE')171        self.addCleanup(self._unrescue, self.server_id)172        # Association of floating IP to a rescued vm173        client = self.floating_ips_client174        resp, body = client.associate_floating_ip_to_server(self.floating_ip,175                                                            self.server_id)176        self.assertEqual(202, resp.status)177        # Disassociation of floating IP that was associated in this method178        resp, body = \179            client.disassociate_floating_ip_from_server(self.floating_ip,180                                                        self.server_id)181        self.assertEqual(202, resp.status)182    @attr(type='gate')183    def test_rescued_vm_add_remove_security_group(self):184        # Rescue the server185        self.servers_client.rescue_server(186            self.server_id, self.password)187        self.servers_client.wait_for_server_status(self.server_id, 'RESCUE')188        # Add Security group189        resp, body = self.servers_client.add_security_group(self.server_id,190                                                            self.sg_name)191        self.assertEqual(202, resp.status)192        # Delete Security group193        resp, body = self.servers_client.remove_security_group(self.server_id,194                                                               self.sg_name)195        self.assertEqual(202, resp.status)196        # Unrescue the server197        resp, body = self.servers_client.unrescue_server(self.server_id)198        self.assertEqual(202, resp.status)199        self.servers_client.wait_for_server_status(self.server_id, 'ACTIVE')200class ServerRescueTestXML(ServerRescueTestJSON):...

test_server_rescue_negative.py

Source:test_server_rescue_negative.py

...43        resc_server = cls.create_test_server(adminPass=rescue_password,44                                             wait_until='ACTIVE')45        cls.server_id = server['id']46        cls.rescue_id = resc_server['id']47        cls.servers_client.rescue_server(48            cls.rescue_id, adminPass=rescue_password)49        waiters.wait_for_server_status(cls.servers_client,50                                       cls.rescue_id, 'RESCUE')51        waiters.wait_for_server_status(cls.servers_client,52                                       cls.server_id, 'ACTIVE')53    def _create_volume(self):54        volume = self.volumes_extensions_client.create_volume(55            size=CONF.volume.volume_size, display_name=data_utils.rand_name(56                self.__class__.__name__ + '_volume'))['volume']57        self.addCleanup(self.delete_volume, volume['id'])58        waiters.wait_for_volume_status(self.volumes_extensions_client,59                                       volume['id'], 'available')60        return volume61    def _detach(self, server_id, volume_id):62        self.servers_client.detach_volume(server_id, volume_id)63        waiters.wait_for_volume_status(self.volumes_extensions_client,64                                       volume_id, 'available')65    def _unrescue(self, server_id):66        self.servers_client.unrescue_server(server_id)67        waiters.wait_for_server_status(self.servers_client,68                                       server_id, 'ACTIVE')69    def _unpause(self, server_id):70        self.servers_client.unpause_server(server_id)71        waiters.wait_for_server_status(self.servers_client,72                                       server_id, 'ACTIVE')73    @test.idempotent_id('cc3a883f-43c0-4fb6-a9bb-5579d64984ed')74    @testtools.skipUnless(CONF.compute_feature_enabled.pause,75                          'Pause is not available.')76    @test.attr(type=['negative'])77    def test_rescue_paused_instance(self):78        # Rescue a paused server79        self.servers_client.pause_server(self.server_id)80        self.addCleanup(self._unpause, self.server_id)81        waiters.wait_for_server_status(self.servers_client,82                                       self.server_id, 'PAUSED')83        self.assertRaises(lib_exc.Conflict,84                          self.servers_client.rescue_server,85                          self.server_id)86    @test.attr(type=['negative'])87    @test.idempotent_id('db22b618-f157-4566-a317-1b6d467a8094')88    def test_rescued_vm_reboot(self):89        self.assertRaises(lib_exc.Conflict, self.servers_client.reboot_server,90                          self.rescue_id, type='HARD')91    @test.attr(type=['negative'])92    @test.idempotent_id('6dfc0a55-3a77-4564-a144-1587b7971dde')93    def test_rescue_non_existent_server(self):94        # Rescue a non-existing server95        non_existent_server = data_utils.rand_uuid()96        self.assertRaises(lib_exc.NotFound,97                          self.servers_client.rescue_server,98                          non_existent_server)99    @test.attr(type=['negative'])100    @test.idempotent_id('70cdb8a1-89f8-437d-9448-8844fd82bf46')101    def test_rescued_vm_rebuild(self):102        self.assertRaises(lib_exc.Conflict,103                          self.servers_client.rebuild_server,104                          self.rescue_id,105                          self.image_ref_alt)106    @test.idempotent_id('d0ccac79-0091-4cf4-a1ce-26162d0cc55f')107    @test.services('volume')108    @test.attr(type=['negative'])109    def test_rescued_vm_attach_volume(self):110        volume = self._create_volume()111        # Rescue the server112        self.servers_client.rescue_server(self.server_id,113                                          adminPass=self.password)114        waiters.wait_for_server_status(self.servers_client,115                                       self.server_id, 'RESCUE')116        self.addCleanup(self._unrescue, self.server_id)117        # Attach the volume to the server118        self.assertRaises(lib_exc.Conflict,119                          self.servers_client.attach_volume,120                          self.server_id,121                          volumeId=volume['id'],122                          device='/dev/%s' % self.device)123    @test.idempotent_id('f56e465b-fe10-48bf-b75d-646cda3a8bc9')124    @test.services('volume')125    @test.attr(type=['negative'])126    def test_rescued_vm_detach_volume(self):127        volume = self._create_volume()128        # Attach the volume to the server129        self.servers_client.attach_volume(self.server_id,130                                          volumeId=volume['id'],131                                          device='/dev/%s' % self.device)132        waiters.wait_for_volume_status(self.volumes_extensions_client,133                                       volume['id'], 'in-use')134        # Rescue the server135        self.servers_client.rescue_server(self.server_id,136                                          adminPass=self.password)137        waiters.wait_for_server_status(self.servers_client,138                                       self.server_id, 'RESCUE')139        # addCleanup is a LIFO queue140        self.addCleanup(self._detach, self.server_id, volume['id'])141        self.addCleanup(self._unrescue, self.server_id)142        # Detach the volume from the server expecting failure143        self.assertRaises(lib_exc.Conflict,144                          self.servers_client.detach_volume,145                          self.server_id,...

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.