python类import_object()的实例源码

def initialize_driver(self, conf):
        self.conf = conf or cfg.CONF.BGP
        try:
            self.dr_driver_cls = (
                    importutils.import_object(self.conf.bgp_speaker_driver,
                                              self.conf))
        except ImportError:
            LOG.exception(_LE("Error while importing BGP speaker driver %s"),
                          self.conf.bgp_speaker_driver)
            raise SystemExit(1)

def __init__(self, topic, host=None):
        super(SchedulerManager, self).__init__()
        self.host = host or CONF.host
        self.topic = topic
        scheduler_driver = CONF.scheduler.scheduler_driver
        self.driver = importutils.import_object(scheduler_driver)
        self._startup_delay = True

def __init__(self):
        LOG.info(_LI("Init huawei l3 driver."))
        self.setup_rpc()
        self.router_scheduler = importutils.import_object(
            cfg.CONF.router_scheduler_driver)
        self.start_periodic_l3_agent_status_check()
        super(HuaweiACL3RouterPlugin, self).__init__()
        if 'dvr' in self.supported_extension_aliases:
            l3_dvrscheduler_db.subscribe()
        l3_db.subscribe()
        LOG.info(_LI("Initialization finished successfully"
                 " for huawei l3 driver."))

def setUp(self):
        super(EngineManagerUnitTestCase, self).setUp()
        self.engine = importutils.import_object(CONF.engine_manager)
        self.context = context.RequestContext()

def create(self, label=None, cidr=None, num_networks=None,
               network_size=None, multi_host=None, vlan=None,
               vlan_start=None, vpn_start=None, cidr_v6=None, gateway=None,
               gateway_v6=None, bridge=None, bridge_interface=None,
               dns1=None, dns2=None, project_id=None, priority=None,
               uuid=None, fixed_cidr=None):
        """Creates fixed IPs for host by range."""
        kwargs = {k: v for k, v in six.iteritems(locals())
                  if v and k != "self"}
        if multi_host is not None:
            kwargs['multi_host'] = multi_host == 'T'
        net_manager = importutils.import_object(CONF.network_manager)
        net_manager.create_networks(context.get_admin_context(), **kwargs)

def init_leases(network_id):
    """Get the list of hosts for a network."""
    ctxt = context.get_admin_context()
    network = objects.Network.get_by_id(ctxt, network_id)
    network_manager = importutils.import_object(CONF.network_manager)
    return network_manager.get_dhcp_leases(ctxt, network)

def __init__(self, scheduler_driver=None, *args, **kwargs):
        if not scheduler_driver:
            scheduler_driver = CONF.scheduler_driver
        try:
            self.driver = driver.DriverManager(
                    "nova.scheduler.driver",
                    scheduler_driver,
                    invoke_on_load=True).driver
        # TODO(Yingxin): Change to catch stevedore.exceptions.NoMatches after
        # stevedore v1.9.0
        except RuntimeError:
            # NOTE(Yingxin): Loading full class path is deprecated and should
            # be removed in the N release.
            try:
                self.driver = importutils.import_object(scheduler_driver)
                LOG.warning(_LW("DEPRECATED: scheduler_driver uses "
                                "classloader to load %(path)s. This legacy "
                                "loading style will be removed in the "
                                "N release."),
                            {'path': scheduler_driver})
            except (ImportError, ValueError):
                raise RuntimeError(
                        _("Cannot load scheduler driver from configuration "
                          "%(conf)s."),
                        {'conf': scheduler_driver})
        super(SchedulerManager, self).__init__(service_name='scheduler',
                                               *args, **kwargs)

def _get_openstack_security_group_driver(skip_policy_check=False):
    if is_neutron_security_groups():
        return importutils.import_object(NEUTRON_DRIVER,
                                         skip_policy_check=skip_policy_check)
    elif CONF.security_group_api.lower() == 'nova':
        return importutils.import_object(NOVA_DRIVER,
                                         skip_policy_check=skip_policy_check)
    else:
        return importutils.import_object(CONF.security_group_api,
                                         skip_policy_check=skip_policy_check)

def _get_interface_driver():
    global interface_driver
    if not interface_driver:
        interface_driver = importutils.import_object(
                CONF.linuxnet_interface_driver)
    return interface_driver

def get_volume_encryptor(connection_info, **kwargs):
    """Creates a VolumeEncryptor used to encrypt the specified volume.

    :param: the connection information used to attach the volume
    :returns VolumeEncryptor: the VolumeEncryptor for the volume
    """
    encryptor = nop.NoOpEncryptor(connection_info, **kwargs)

    location = kwargs.get('control_location', None)
    if location and location.lower() == 'front-end':  # case insensitive
        provider = kwargs.get('provider')

        if provider == 'LuksEncryptor':
            provider = 'nova.volume.encryptors.luks.' + provider
        elif provider == 'CryptsetupEncryptor':
            provider = 'nova.volume.encryptors.cryptsetup.' + provider
        elif provider == 'NoOpEncryptor':
            provider = 'nova.volume.encryptors.nop.' + provider
        try:
            encryptor = importutils.import_object(provider, connection_info,
                                                  **kwargs)
        except Exception as e:
            LOG.error(_LE("Error instantiating %(provider)s: %(exception)s"),
                      {'provider': provider, 'exception': e})
            raise

    msg = ("Using volume encryptor '%(encryptor)s' for connection: "
           "%(connection_info)s" %
           {'encryptor': encryptor, 'connection_info': connection_info})
    LOG.debug(strutils.mask_password(msg))

    return encryptor

def __init__(self):
        transport = CONF.libvirt.remote_filesystem_transport
        cls_name = '.'.join([__name__, transport.capitalize()])
        cls_name += 'Driver'
        self.driver = importutils.import_object(cls_name)

def instance_for_format(image, mountdir, partition):
        """Get a Mount instance for the image type

        :param image: instance of nova.virt.image.model.Image
        :param mountdir: path to mount the image at
        :param partition: partition number to mount
        """
        LOG.debug("Instance for format image=%(image)s "
                  "mountdir=%(mountdir)s partition=%(partition)s",
                  {'image': image, 'mountdir': mountdir,
                   'partition': partition})

        if isinstance(image, imgmodel.LocalFileImage):
            if image.format == imgmodel.FORMAT_RAW:
                LOG.debug("Using LoopMount")
                return importutils.import_object(
                    "nova.virt.disk.mount.loop.LoopMount",
                    image, mountdir, partition)
            else:
                LOG.debug("Using NbdMount")
                return importutils.import_object(
                    "nova.virt.disk.mount.nbd.NbdMount",
                    image, mountdir, partition)
        elif isinstance(image, imgmodel.LocalBlockImage):
            LOG.debug("Using BlockMount")
            return importutils.import_object(
                    "nova.virt.disk.mount.block.BlockMount",
                    image, mountdir, partition)
        else:
            # TODO(berrange) We could mount RBDImage directly
            # using kernel RBD block dev support.
            #
            # This is left as an enhancement for future
            # motivated developers todo, since raising
            # an exception is on par with what this
            # code did historically
            raise exception.UnsupportedImageModel(
                image.__class__.__name__)

def instance_for_device(image, mountdir, partition, device):
        """Get a Mount instance for the device type

        :param image: instance of nova.virt.image.model.Image
        :param mountdir: path to mount the image at
        :param partition: partition number to mount
        :param device: mounted device path
        """

        LOG.debug("Instance for device image=%(image)s "
                  "mountdir=%(mountdir)s partition=%(partition)s "
                  "device=%(device)s",
                  {'image': image, 'mountdir': mountdir,
                   'partition': partition, 'device': device})

        if "loop" in device:
            LOG.debug("Using LoopMount")
            return importutils.import_object(
                "nova.virt.disk.mount.loop.LoopMount",
                image, mountdir, partition, device)
        elif "nbd" in device:
            LOG.debug("Using NbdMount")
            return importutils.import_object(
                "nova.virt.disk.mount.nbd.NbdMount",
                image, mountdir, partition, device)
        else:
            LOG.debug("Using BlockMount")
            return importutils.import_object(
                "nova.virt.disk.mount.block.BlockMount",
                image, mountdir, partition, device)

def instance_for_image(image, partition):
        """Get a VFS instance for the image

        :param image: instance of nova.virt.image.model.Image
        :param partition: the partition number to access
        """

        LOG.debug("Instance for image image=%(image)s "
                  "partition=%(partition)s",
                  {'image': image, 'partition': partition})

        vfs = None
        try:
            LOG.debug("Using primary VFSGuestFS")
            vfs = importutils.import_object(
                "nova.virt.disk.vfs.guestfs.VFSGuestFS",
                image, partition)
            if not VFS.guestfs_ready:
                # Inspect for capabilities and keep
                # track of the result only if succeeded.
                vfs.inspect_capabilities()
                VFS.guestfs_ready = True
            return vfs
        except exception.NovaException:
            if vfs is not None:
                # We are able to load libguestfs but
                # something wrong happens when trying to
                # check for capabilities.
                raise
            else:
                LOG.info(_LI("Unable to import guestfs, "
                             "falling back to VFSLocalFS"))

        return importutils.import_object(
            "nova.virt.disk.vfs.localfs.VFSLocalFS",
            image, partition)

def _default_download_handler():
    # TODO(sirp):  This should be configurable like upload_handler
    return importutils.import_object(
            'nova.virt.xenapi.image.glance.GlanceStore')

def test_live_migration_uses_migrateToURI_without_migratable_flag(self):
        self.compute = importutils.import_object(CONF.compute_manager)
        instance_dict = dict(self.test_instance)
        instance_dict.update({'host': 'fake',
                              'power_state': power_state.RUNNING,
                              'vm_state': vm_states.ACTIVE})
        instance_ref = objects.Instance(**instance_dict)

        drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), False)

        # Preparing mocks
        vdmock = self.mox.CreateMock(fakelibvirt.virDomain)
        self.mox.StubOutWithMock(vdmock, "migrateToURI")
        _bandwidth = CONF.libvirt.live_migration_bandwidth
        vdmock.migrateToURI(drvr._live_migration_uri('dest'),
                            mox.IgnoreArg(),
                            None,
                            _bandwidth).AndRaise(
                                fakelibvirt.libvirtError("ERR"))

        # start test
        migrate_data = objects.LibvirtLiveMigrateData(
            graphics_listen_addr_vnc='0.0.0.0',
            graphics_listen_addr_spice='0.0.0.0',
            serial_listen_addr='127.0.0.1',
            target_connect_addr=None,
            bdms=[],
            block_migration=False)
        self.mox.ReplayAll()
        self.assertRaises(fakelibvirt.libvirtError,
                          drvr._live_migration_operation,
                          self.context, instance_ref, 'dest',
                          False, migrate_data, vdmock, [])

def test_live_migration_uses_migrateToURI_without_dest_listen_addrs(self):
        self.compute = importutils.import_object(CONF.compute_manager)
        instance_dict = dict(self.test_instance)
        instance_dict.update({'host': 'fake',
                              'power_state': power_state.RUNNING,
                              'vm_state': vm_states.ACTIVE})
        instance_ref = objects.Instance(**instance_dict)

        drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), False)

        # Preparing mocks
        vdmock = self.mox.CreateMock(fakelibvirt.virDomain)
        self.mox.StubOutWithMock(vdmock, "migrateToURI")
        _bandwidth = CONF.libvirt.live_migration_bandwidth
        vdmock.migrateToURI(drvr._live_migration_uri('dest'),
                            mox.IgnoreArg(),
                            None,
                            _bandwidth).AndRaise(
                                fakelibvirt.libvirtError("ERR"))

        # start test
        migrate_data = objects.LibvirtLiveMigrateData(
            serial_listen_addr='',
            target_connect_addr=None,
            bdms=[],
            block_migration=False)
        self.mox.ReplayAll()
        self.assertRaises(fakelibvirt.libvirtError,
                          drvr._live_migration_operation,
                          self.context, instance_ref, 'dest',
                          False, migrate_data, vdmock, [])

def test_live_migration_fails_without_migratable_flag_or_0_addr(self):
        self.flags(enabled=True, vncserver_listen='1.2.3.4', group='vnc')
        self.compute = importutils.import_object(CONF.compute_manager)
        instance_dict = dict(self.test_instance)
        instance_dict.update({'host': 'fake',
                              'power_state': power_state.RUNNING,
                              'vm_state': vm_states.ACTIVE})
        instance_ref = objects.Instance(**instance_dict)

        # Preparing mocks
        vdmock = self.mox.CreateMock(fakelibvirt.virDomain)
        self.mox.StubOutWithMock(vdmock, "migrateToURI")

        # start test
        migrate_data = objects.LibvirtLiveMigrateData(
            graphics_listen_addr_vnc='1.2.3.4',
            graphics_listen_addr_spice='1.2.3.4',
            serial_listen_addr='127.0.0.1',
            target_connect_addr=None,
            block_migration=False)
        self.mox.ReplayAll()
        drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), False)
        self.assertRaises(exception.MigrationError,
                          drvr._live_migration_operation,
                          self.context, instance_ref, 'dest',
                          False, migrate_data, vdmock, [])

def test_run_image_cache_manager_pass(self):
        was = {'called': False}

        def fake_get_all_by_filters(context, *args, **kwargs):
            was['called'] = True
            instances = []
            for x in range(2):
                instances.append(fake_instance.fake_db_instance(
                                                        image_ref='1',
                                                        uuid=x,
                                                        name=x,
                                                        vm_state='',
                                                        task_state=''))
            return instances

        with utils.tempdir() as tmpdir:
            self.flags(instances_path=tmpdir)

            self.stub_out('nova.db.instance_get_all_by_filters',
                          fake_get_all_by_filters)
            compute = importutils.import_object(CONF.compute_manager)
            self.flags(use_local=True, group='conductor')
            compute.conductor_api = conductor.API()
            ctxt = context.get_admin_context()
            compute._run_image_cache_manager_pass(ctxt)
            self.assertTrue(was['called'])

def setUp(self):
        super(XenAPIAggregateTestCase, self).setUp()
        self.flags(connection_url='http://test_url',
                   connection_username='test_user',
                   connection_password='test_pass',
                   group='xenserver')
        self.flags(instance_name_template='%d',
                   firewall_driver='nova.virt.xenapi.firewall.'
                                   'Dom0IptablesFirewallDriver',
                   host='host',
                   compute_driver='xenapi.XenAPIDriver',
                   default_availability_zone='avail_zone1')
        self.flags(use_local=True, group='conductor')
        host_ref = xenapi_fake.get_all('host')[0]
        stubs.stubout_session(self.stubs, stubs.FakeSessionForVMTests)
        self.context = context.get_admin_context()
        self.conn = xenapi_conn.XenAPIDriver(fake.FakeVirtAPI(), False)
        self.compute = importutils.import_object(CONF.compute_manager)
        self.api = compute_api.AggregateAPI()
        values = {'name': 'test_aggr',
                  'metadata': {'availability_zone': 'test_zone',
                  pool_states.POOL_FLAG: 'XenAPI'}}
        self.aggr = objects.Aggregate(context=self.context, id=1,
                                      **values)
        self.fake_metadata = {pool_states.POOL_FLAG: 'XenAPI',
                              'master_compute': 'host',
                              'availability_zone': 'fake_zone',
                              pool_states.KEY: pool_states.ACTIVE,
                              'host': xenapi_fake.get_record('host',
                                                             host_ref)['uuid']}