python类libvirtError()的实例源码

def test_set_boot_device_error(self, mock_libvirt_domain,
                                   mock_libvirt_open):
        mock_libvirt_domain.side_effect = libvirt.libvirtError('boom')
        ret = self.vbmc.set_boot_device('network')
        self.assertEqual(0xd5, ret)
        self._assert_libvirt_calls(mock_libvirt_domain, mock_libvirt_open)

def test_get_power_state_error(self, mock_libvirt_domain,
                                   mock_libvirt_open):
        mock_libvirt_domain.side_effect = libvirt.libvirtError('boom')
        ret = self.vbmc.get_power_state()
        self.assertEqual(0xd5, ret)
        self._assert_libvirt_calls(mock_libvirt_domain, mock_libvirt_open,
                                   readonly=True)

def test_pulse_diag_error(self, mock_libvirt_domain, mock_libvirt_open):
        mock_libvirt_domain.side_effect = libvirt.libvirtError('boom')
        ret = self.vbmc.pulse_diag()
        self.assertEqual(0xd5, ret)
        mock_libvirt_domain.return_value.injectNMI.assert_not_called()
        self._assert_libvirt_calls(mock_libvirt_domain, mock_libvirt_open)

def test_power_off_error(self, mock_libvirt_domain, mock_libvirt_open):
        mock_libvirt_domain.side_effect = libvirt.libvirtError('boom')
        ret = self.vbmc.power_off()
        self.assertEqual(0xd5, ret)
        mock_libvirt_domain.return_value.destroy.assert_not_called()
        self._assert_libvirt_calls(mock_libvirt_domain, mock_libvirt_open)

def test_power_shutdown_error(self, mock_libvirt_domain,
                                  mock_libvirt_open):
        mock_libvirt_domain.side_effect = libvirt.libvirtError('boom')
        ret = self.vbmc.power_shutdown()
        self.assertEqual(0xd5, ret)
        mock_libvirt_domain.return_value.shutdown.assert_not_called()
        self._assert_libvirt_calls(mock_libvirt_domain, mock_libvirt_open)

def test_get_libvirt_domain_not_found(self):
        self.fake_connection.lookupByName.side_effect = libvirt.libvirtError(
            'boom')
        self.assertRaises(exception.DomainNotFound, utils.get_libvirt_domain,
                          self.fake_connection, 'Fred')
        self.fake_connection.lookupByName.assert_called_once_with('Fred')

def test_libvirt_open_error(self, mock_open):
        mock_open.side_effect = libvirt.libvirtError('boom')
        self.assertRaises(exception.LibvirtConnectionOpenError,
                          self._test_libvirt_open, mock_open)
        mock_open.assert_called_once_with(self.uri)

def set_boot_device(self, bootdevice):
        LOG.debug('Set boot device called for %(domain)s with boot '
                  'device "%(bootdev)s"', {'domain': self.domain_name,
                                           'bootdev': bootdevice})
        device = SET_BOOT_DEVICES_MAP.get(bootdevice)
        if device is None:
            # Invalid data field in request
            return 0xcc

        try:
            with utils.libvirt_open(**self._conn_args) as conn:
                domain = utils.get_libvirt_domain(conn, self.domain_name)
                tree = ET.fromstring(domain.XMLDesc())

                for os_element in tree.findall('os'):
                    # Remove all "boot" elements
                    for boot_element in os_element.findall('boot'):
                        os_element.remove(boot_element)

                    # Add a new boot element with the request boot device
                    boot_element = ET.SubElement(os_element, 'boot')
                    boot_element.set('dev', device)

                conn.defineXML(ET.tostring(tree))
        except libvirt.libvirtError:
            LOG.error('Failed setting the boot device  %(bootdev)s for '
                      'domain %(domain)s', {'bootdev': device,
                                            'domain': self.domain_name})
            # Command not supported in present state
            return 0xd5

def get_power_state(self):
        LOG.debug('Get power state called for domain %s', self.domain_name)
        try:
            with utils.libvirt_open(readonly=True, **self._conn_args) as conn:
                domain = utils.get_libvirt_domain(conn, self.domain_name)
                if domain.isActive():
                    return POWERON
        except libvirt.libvirtError as e:
            LOG.error('Error getting the power state of domain %(domain)s. '
                      'Error: %(error)s', {'domain': self.domain_name,
                                           'error': e})
            # Command not supported in present state
            return 0xd5

        return POWEROFF

def pulse_diag(self):
        LOG.debug('Power diag called for domain %s', self.domain_name)
        try:
            with utils.libvirt_open(**self._conn_args) as conn:
                domain = utils.get_libvirt_domain(conn, self.domain_name)
                if domain.isActive():
                    domain.injectNMI()
        except libvirt.libvirtError as e:
            LOG.error('Error powering diag the domain %(domain)s. '
                      'Error: %(error)s' % {'domain': self.domain_name,
                                            'error': e})
            # Command not supported in present state
            return 0xd5

def power_on(self):
        LOG.debug('Power on called for domain %s', self.domain_name)
        try:
            with utils.libvirt_open(**self._conn_args) as conn:
                domain = utils.get_libvirt_domain(conn, self.domain_name)
                if not domain.isActive():
                    domain.create()
        except libvirt.libvirtError as e:
            LOG.error('Error powering on the domain %(domain)s. '
                      'Error: %(error)s' % {'domain': self.domain_name,
                                            'error': e})
            # Command not supported in present state
            return 0xd5

def power_shutdown(self):
        LOG.debug('Soft power off called for domain %s', self.domain_name)
        try:
            with utils.libvirt_open(**self._conn_args) as conn:
                domain = utils.get_libvirt_domain(conn, self.domain_name)
                if domain.isActive():
                    domain.shutdown()
        except libvirt.libvirtError as e:
            LOG.error('Error soft powering off the domain %(domain)s. '
                      'Error: %(error)s' % {'domain': self.domain_name,
                                            'error': e})
            # Command not supported in present state
            return 0xd5

def get_libvirt_domain(conn, domain):
    try:
        return conn.lookupByName(domain)
    except libvirt.libvirtError:
        raise exception.DomainNotFound(domain=domain)

def __enter__(self):
        try:
            self._conn = (libvirt.openReadOnly(self.uri)
                          if self.readonly else
                          libvirt.open(self.uri))
            return self._conn
        except libvirt.libvirtError as e:
            print('Error when connecting to the libvirt URI "%(uri)s": '
                  '%(error)s' % {'uri': self.uri, 'error': e})
            flask.abort(500)

def get_libvirt_domain(connection, domain):
    try:
        return connection.lookupByName(domain)
    except libvirt.libvirtError:
        flask.abort(404)

def system_reset_action(identity):
    reset_type = flask.request.json.get('ResetType')
    with libvirt_open(LIBVIRT_URI) as conn:
        domain = get_libvirt_domain(conn, identity)
        try:
            if reset_type in ('On', 'ForceOn'):
                if not domain.isActive():
                    domain.create()
            elif reset_type == 'ForceOff':
                if domain.isActive():
                    domain.destroy()
            elif reset_type == 'GracefulShutdown':
                if domain.isActive():
                    domain.shutdown()
            elif reset_type == 'GracefulRestart':
                if domain.isActive():
                    domain.reboot()
            elif reset_type == 'ForceRestart':
                if domain.isActive():
                    domain.reset()
            elif reset_type == 'Nmi':
                if domain.isActive():
                    domain.injectNMI()
        except libvirt.libvirtError:
            flask.abort(500)

    return '', 204

def __connect_tcp(self):
        flags = [libvirt.VIR_CRED_AUTHNAME, libvirt.VIR_CRED_PASSPHRASE]
        auth = [flags, self.__libvirt_auth_credentials_callback, None]
        uri = 'qemu+tcp://%s/system' % self.host

        try:
            return libvirt.openAuth(uri, auth, 0)
        except libvirtError as e:
            self.last_error = 'Connection Failed: ' + str(e)
            self.connection = None

def __connect_socket(self):
        uri = 'qemu:///system'

        try:
            return libvirt.open(uri)
        except libvirtError as e:
            self.last_error = 'Connection Failed: ' + str(e)
            self.connection = None