Python oslo_config.cfg 模块，ConfigFilesNotFoundError() 实例源码

def load_app(conf, appname='panko+keystone'):
    global APPCONFIGS

    # Build the WSGI app
    cfg_path = conf.api_paste_config
    if not os.path.isabs(cfg_path):
        cfg_path = conf.find_file(cfg_path)

    if cfg_path is None or not os.path.exists(cfg_path):
        raise cfg.ConfigFilesNotFoundError([conf.api_paste_config])

    config = dict(conf=conf)
    configkey = str(uuid.uuid4())
    APPCONFIGS[configkey] = config

    LOG.info("Full WSGI config used: %s" % cfg_path)
    return deploy.loadapp("config:" + cfg_path, name=appname,
                          global_conf={'configkey': configkey})

def _get_policy_path(self, path):
        """Locate the policy JSON data file/path.

        :param path: It's value can be a full path or related path. When
                     full path specified, this function just returns the full
                     path. When related path specified, this function will
                     search configuration directories to find one that exists.

        :returns: The policy path

        :raises: ConfigFilesNotFoundError if the file/path couldn't
                 be located.
        """
        policy_path = self.conf.find_file(path)

        if policy_path:
            return policy_path

        raise cfg.ConfigFilesNotFoundError((path,))

def load_app():
    cfg_file = None
    cfg_path = CONF.api.api_paste_config
    paste_pipeline = CONF.api.paste_pipeline
    if not os.path.isabs(cfg_path):
        cfg_file = CONF.find_file(cfg_path)
    elif os.path.exists(cfg_path):
        cfg_file = cfg_path

    if not cfg_file:
        raise cfg.ConfigFilesNotFoundError([cfg.CONF.api.api_paste_config])
    LOG.info("Loading craton-api with pipeline %(pipeline)s and WSGI config:"
             "%(conf)s", {'conf': cfg_file, 'pipeline': paste_pipeline})
    return deploy.loadapp("config:%s" % cfg_file, name=paste_pipeline)

def main():
    """Parse options and call the appropriate class/method."""
    CONF.register_cli_opt(category_opt)
    script_name = sys.argv[0]
    if len(sys.argv) < 2:
        print(_("\nOpenStack meteos version: %(version)s\n") %
              {'version': version.version_string()})
        print(script_name + " category action [<args>]")
        print(_("Available categories:"))
        for category in CATEGORIES:
            print("\t%s" % category)
        sys.exit(2)

    try:
        log.register_options(CONF)
        CONF(sys.argv[1:], project='meteos',
             version=version.version_string())
        log.setup(CONF, "meteos")
    except cfg.ConfigFilesNotFoundError:
        cfgfile = CONF.config_file[-1] if CONF.config_file else None
        if cfgfile and not os.access(cfgfile, os.R_OK):
            st = os.stat(cfgfile)
            print(_("Could not read %s. Re-running with sudo") % cfgfile)
            try:
                os.execvp('sudo', ['sudo', '-u', '#%s' % st.st_uid] + sys.argv)
            except Exception:
                print(_('sudo failed, continuing as if nothing happened'))

        print(_('Please re-run meteos-manage as root.'))
        sys.exit(2)

    fn = CONF.category.action_fn

    fn_args = fetch_func_args(fn)
    fn(*fn_args)

def test_api_paste_file_not_exist(self):
        self.CONF.set_override('api_paste_config', 'non-existent-file')
        with mock.patch.object(self.CONF, 'find_file') as ff:
            ff.return_value = None
            self.assertRaises(cfg.ConfigFilesNotFoundError, app.load_app,
                              self.CONF)

def load_app():
    cfg_file = None
    cfg_path = CONF.api.api_paste_config
    if not os.path.isabs(cfg_path):
        cfg_file = CONF.find_file(cfg_path)
    elif os.path.exists(cfg_path):
        cfg_file = cfg_path

    if not cfg_file:
        raise cfg.ConfigFilesNotFoundError([CONF.api.api_paste_config])
    LOG.info("Full WSGI config used: %s", cfg_file)
    return deploy.loadapp("config:" + cfg_file)

def revert_snapshot(self, name):
        """Revert snapshot by name

        - Revert a libvirt snapshots for all nodes in the environment
        - Try to reload 'config' object from a file 'config_<name>.ini'
          If the file not found, then pass with defaults.
        - Set <name> as the current state of the environment after reload

        :param name: string
        """
        LOG.info("Reverting from snapshot named '{0}'".format(name))
        if self._env is not None:
            self._env.revert(name=name)
            LOG.info("Resuming environment after revert")
            self._env.resume()
        else:
            raise exceptions.EnvironmentIsNotSet()

        try:
            test_config_path = self._get_snapshot_config_name(name)
            settings_oslo.reload_snapshot_config(self.__config,
                                                 test_config_path)
        except cfg.ConfigFilesNotFoundError as conf_err:
            LOG.error("Config file(s) {0} not found!".format(
                conf_err.config_files))

        self.__config.hardware.current_snapshot = name

def revert_snapshot(self, name):
        """Revert snapshot by name

        - Revert a libvirt snapshots for all nodes in the environment
        - Try to reload 'config' object from a file 'config_<name>.ini'
          If the file not found, then pass with defaults.
        - Set <name> as the current state of the environment after reload

        :param name: string
        """
        if not settings.MAKE_SNAPSHOT_STAGES:
            LOG.info("SKIP reverting from snapshot '{0}' "
                     "because MAKE_SNAPSHOT_STAGES=false".format(name))
            return

        LOG.info("Reverting from snapshot named '{0}'".format(name))
        if self.__env is not None:
            self.__env.revert(name=name)
            LOG.info("Resuming environment after revert")
            self.__env.resume()
        else:
            raise exceptions.EnvironmentIsNotSet()

        try:
            test_config_path = self._get_snapshot_config_name(name)
            settings_oslo.reload_snapshot_config(self.__config,
                                                 test_config_path)
        except cfg.ConfigFilesNotFoundError as conf_err:
            LOG.error("Config file(s) {0} not found!".format(
                conf_err.config_files))

        self.__config.hardware.current_snapshot = name

def test_unhappy_config_io(self):
        """Test unhappy.cfg I/O and validate I/O error in config status."""
        cfg.CONF.clear()
        try:
            config = Config("unhappy.cfg")
            config_status = config.configure()
            self.validate_test("I/O error" in config_status)

        except Exception as ex:
            self.validate_test(isinstance(ex, cfg.ConfigFilesNotFoundError))

def load_app():
    # Build the WSGI app
    cfg_file = None
    cfg_path = cfg.CONF.api_paste_config
    if not os.path.isabs(cfg_path):
        cfg_file = CONF.find_file(cfg_path)
    elif os.path.exists(cfg_path):
        cfg_file = cfg_path

    if not cfg_file:
        raise cfg.ConfigFilesNotFoundError([cfg.CONF.api_paste_config])
    LOG.info("Full WSGI config used: %s" % cfg_file)
    return deploy.loadapp("config:" + cfg_file)

def test_unhappy_config_io(self):
        """Test unhappy.cfg I/O and validate I/O error in config status."""
        cfg.CONF.clear()
        try:
            config = Config("unhappy.cfg")
            config_status = config.configure()
            self.validate_test("I/O error" in config_status)

        except Exception as ex:
            self.validate_test(isinstance(ex, cfg.ConfigFilesNotFoundError))

def main():
    """Parse options and call the appropriate class/method."""
    script_name = sys.argv[0]
    if len(sys.argv) < 2:
        print(_("\nOpenStack masakari version: %(version)s\n") %
              {'version': version.version_string()})
        print(script_name + " category action [<args>]")
        print(_("Available categories:"))
        for category in CATEGORIES:
            print(_("\t%s") % category)
        sys.exit(2)

    try:
        CONF(sys.argv[1:], project='masakari',
             version=version.version_string())
        logging.setup(CONF, "masakari")
        python_logging.captureWarnings(True)
    except cfg.ConfigDirNotFoundError as details:
        print(_("Invalid directory: %s") % details)
        sys.exit(2)
    except cfg.ConfigFilesNotFoundError as e:
        cfg_files = ', '.join(e.config_files)
        print(_("Failed to read configuration file(s): %s") % cfg_files)
        sys.exit(2)

    fn = CONF.category.action_fn
    fn_args = fetch_func_args(fn)
    fn(*fn_args)

def test_cells_config_not_found(self):
        self.flags(cells_config='no_such_file_exists.conf', group='cells')
        e = self.assertRaises(cfg.ConfigFilesNotFoundError,
                              state.CellStateManager)
        self.assertEqual(['no_such_file_exists.conf'], e.config_files)

def __init__(self, cell_state_cls=None):
        cells_config = CONF.cells.cells_config
        self.cells_config_path = CONF.find_file(cells_config)
        if not self.cells_config_path:
            raise cfg.ConfigFilesNotFoundError(config_files=[cells_config])
        super(CellStateManagerFile, self).__init__(cell_state_cls)