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})
python类ConfigFilesNotFoundError()的实例源码
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)
test_cells_state_manager.py 文件源码
项目:Trusted-Platform-Module-nova
作者: BU-NU-CLOUD-SP16
项目源码
文件源码
阅读 23
收藏 0
点赞 0
评论 0
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)