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

How to use unrescue_server method in tempest

Best Python code snippet using tempest_python

openstackbmc.py

Source:openstackbmc.py

...110          current = metadata.metadata["bootorder"]111        if current != bootdevice:112          if bootdevice in ("hd"):113            if self._vm.status == "RESCUE":114              self._client.compute.unrescue_server(self._vm)115              self._server_unrescued = True116          self._boot_order_changed = bootdevice117        return True118    def set_kg(self, kg):119        """Desactivated IPMI call."""120        raise NotImplementedError121    def power_reset(self):122        """Reset a VM."""123        # Shmulik wrote "Currently, limited to: "chassis power on/off/status"124        raise NotImplementedError125    # Implement power state BMC features126    def get_power_state(self):127        """Get the power state of a OpenStack VM."""128        try:129            self._vm = self.get_vm(self._vm_name)130            if self._vm.status in ['STARTED','ACTIVE','BUILDING', 'RESCUE']:131                if self._vm.status == "ACTIVE" and self._server_unrescued:132                  self._client.compute.stop_server(self._vm)133                  self._server_unrescued = False134                  return "off"135                logging.info('returning power state ON for vm ' + self._vm_name)136                return "on"137            else:138                if self._boot_order_changed:139                  logging.info('boot order changed for vm ' + self._vm_name + ' to ' + self._boot_order_changed)140                  self._client.compute.set_server_metadata(self._vm,bootorder=self._boot_order_changed)141                  metadata = self._client.compute.get_server_metadata(self._vm)142                  if self._boot_order_changed == "network":143                    logging.info('autostart (rescue) ' + self._vm_name + ' using' + self._boot_order_changed)144                    self._client.compute.rescue_server(self._vm,image_ref=self._pxe_image)145                  else:146                    logging.info('autostart (unrescue) ' + self._vm_name + ' using ' + self._boot_order_changed)147                    try:148                        self._client.compute.unrescue_server(self._vm)149                    except:150                        pass151                    self._client.compute.start_server( self._vm)152                  self._boot_order_changed = False153                logging.info('returning power state OFF for vm ' + self._vm_name)154                return "off"155        except Exception as e:156            logging.error(self._vm_name + ' get_power_state:' + str(e))157            if self._vm.status in ['STARTED','ACTIVE','BUILDING', 'RESCUE']:158                return "on"159            return "off"160        return "off"161    def power_off(self):162        """Cut the power without waiting for clean shutdown."""163        self._vm = self.get_vm( self._vm_name)164        logging.info("Power OFF called for VM " + self._vm_name + " with state: " + self._vm.status)165        if self._vm.status in ['STARTED','ACTIVE']:166          try:167              self._client.compute.stop_server( self._vm)168          except Exception as e:169              logging.error(self._vm_name + ' power_off:' + str(e))170              return 0xc0171        elif self._vm.status in ['RESCUE']:172          try:173              self._client.compute.unrescue_server(self._vm)174              self._server_unrescued = True175              #self._client.compute.reset_server_state(self._vm,"active")176              #self._client.compute.stop_server( self._vm)177          except Exception as e:178              logging.error(self._vm_name + ' power_off:' + str(e))179              return 0xc0180        elif self._vm.status in ['SHUTOFF']:181          return False182        else:183          return 0xc0184    def power_on(self):185        """Start a vm."""186        self._vm = self.get_vm( self._vm_name)187        metadata = self._client.compute.get_server_metadata(self._vm)...

test_server_rescue.py

Source:test_server_rescue.py

...87                                                              'available')88    def _delete(self, volume_id):89        self.volumes_extensions_client.delete_volume(volume_id)90    def _unrescue(self, server_id):91        resp, body = self.servers_client.unrescue_server(server_id)92        self.assertEqual(202, resp.status)93        self.servers_client.wait_for_server_status(server_id, 'ACTIVE')94    @attr(type='smoke')95    def test_rescue_unrescue_instance(self):96        resp, body = self.servers_client.rescue_server(97            self.server_id, self.password)98        self.assertEqual(200, resp.status)99        self.servers_client.wait_for_server_status(self.server_id, 'RESCUE')100        resp, body = self.servers_client.unrescue_server(self.server_id)101        self.assertEqual(202, resp.status)102        self.servers_client.wait_for_server_status(self.server_id, 'ACTIVE')103    @attr(type='negative')104    def test_rescued_vm_reboot(self):105        self.assertRaises(exceptions.Duplicate, self.servers_client.reboot,106                          self.rescue_id, 'HARD')107    @attr(type='negative')108    def test_rescued_vm_rebuild(self):109        self.assertRaises(exceptions.Duplicate,110                          self.servers_client.rebuild,111                          self.rescue_id,112                          self.image_ref_alt)113    @attr(type='negative')114    def test_rescued_vm_attach_volume(self):115        # Rescue the server116        self.servers_client.rescue_server(self.server_id, self.password)117        self.servers_client.wait_for_server_status(self.server_id, 'RESCUE')118        self.addCleanup(self._unrescue, self.server_id)119        # Attach the volume to the server120        self.assertRaises(exceptions.Duplicate,121                          self.servers_client.attach_volume,122                          self.server_id,123                          self.volume_to_attach['id'],124                          device='/dev/%s' % self.device)125    @attr(type='negative')126    def test_rescued_vm_detach_volume(self):127        # Attach the volume to the server128        self.servers_client.attach_volume(self.server_id,129                                          self.volume_to_detach['id'],130                                          device='/dev/%s' % self.device)131        self.volumes_extensions_client.wait_for_volume_status132        (self.volume_to_detach['id'], 'in-use')133        # Rescue the server134        self.servers_client.rescue_server(self.server_id, self.password)135        self.servers_client.wait_for_server_status(self.server_id, 'RESCUE')136        self.addCleanup(self._unrescue, self.server_id)137        # Detach the volume from the server expecting failure138        self.assertRaises(exceptions.Duplicate,139                          self.servers_client.detach_volume,140                          self.server_id,141                          self.volume_to_detach['id'])142    @attr(type='positive')143    def test_rescued_vm_associate_dissociate_floating_ip(self):144        # Rescue the server145        self.servers_client.rescue_server(146            self.server_id, self.password)147        self.servers_client.wait_for_server_status(self.server_id, 'RESCUE')148        #Association of floating IP to a rescued vm149        client = self.floating_ips_client150        resp, body =\151        client.associate_floating_ip_to_server(self.floating_ip,152                                               self.server_id)153        self.assertEqual(202, resp.status)154        #Disassociation of floating IP that was associated in this method155        resp, body = \156            client.disassociate_floating_ip_from_server(self.floating_ip,157                                                        self.server_id)158        self.assertEqual(202, resp.status)159        # Unrescue the server160        resp, body = self.servers_client.unrescue_server(self.server_id)161        self.assertEqual(202, resp.status)162        self.servers_client.wait_for_server_status(self.server_id, 'ACTIVE')163    @attr(type='positive')164    @testtools.skip("Skipped until Bug #1126257 is resolved")165    def test_rescued_vm_add_remove_security_group(self):166        #Add Security group167        resp, body = self.servers_client.add_security_group(self.server_id,168                                                            self.sg_name)169        self.assertEqual(202, resp.status)170        #Delete Security group171        resp, body = self.servers_client.remove_security_group(self.server_id,172                                                               self.sg_id)173        self.assertEqual(202, resp.status)174        # Unrescue the server175        resp, body = self.servers_client.unrescue_server(self.server_id)176        self.assertEqual(202, resp.status)177        self.servers_client.wait_for_server_status(self.server_id, 'ACTIVE')178class ServerRescueTestXML(ServerRescueTestJSON):...

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.