def get_config(parameter, file_path):
"""
Returns the value of a given parameter in file.yaml
parameter must be given in string format with dots
Example: general.openstack.image_name
"""
with open(file_path) as config_file:
file_yaml = yaml.safe_load(config_file)
config_file.close()
value = file_yaml
for element in parameter.split("."):
value = value.get(element)
if value is None:
raise ValueError("The parameter %s is not defined in"
" reporting.yaml" % parameter)
return value
python类safe_load()的实例源码
def excl_scenario():
"""Exclude scenario."""
black_tests = []
try:
with open(RallyBase.BLACKLIST_FILE, 'r') as black_list_file:
black_list_yaml = yaml.safe_load(black_list_file)
installer_type = CONST.__getattribute__('INSTALLER_TYPE')
deploy_scenario = CONST.__getattribute__('DEPLOY_SCENARIO')
if (bool(installer_type) and bool(deploy_scenario) and
'scenario' in black_list_yaml.keys()):
for item in black_list_yaml['scenario']:
scenarios = item['scenarios']
installers = item['installers']
in_it = RallyBase.in_iterable_re
if (in_it(deploy_scenario, scenarios) and
in_it(installer_type, installers)):
tests = item['tests']
black_tests.extend(tests)
except Exception:
LOGGER.debug("Scenario exclusion not applied.")
return black_tests
def excl_func(self):
"""Exclude functionalities."""
black_tests = []
func_list = []
try:
with open(RallyBase.BLACKLIST_FILE, 'r') as black_list_file:
black_list_yaml = yaml.safe_load(black_list_file)
if not self._migration_supported():
func_list.append("no_migration")
if 'functionality' in black_list_yaml.keys():
for item in black_list_yaml['functionality']:
functions = item['functions']
for func in func_list:
if func in functions:
tests = item['tests']
black_tests.extend(tests)
except Exception: # pylint: disable=broad-except
LOGGER.debug("Functionality exclusion not applied.")
return black_tests
def get_parameter_from_yaml(parameter, file):
"""
Returns the value of a given parameter in file.yaml
parameter must be given in string format with dots
Example: general.openstack.image_name
"""
with open(file) as f:
file_yaml = yaml.safe_load(f)
f.close()
value = file_yaml
for element in parameter.split("."):
value = value.get(element)
if value is None:
raise ValueError("The parameter %s is not defined in"
" %s" % (parameter, file))
return value
def create(sysctl_dict, sysctl_file):
"""Creates a sysctl.conf file from a YAML associative array
:param sysctl_dict: a YAML-formatted string of sysctl options eg "{ 'kernel.max_pid': 1337 }"
:type sysctl_dict: str
:param sysctl_file: path to the sysctl file to be saved
:type sysctl_file: str or unicode
:returns: None
"""
try:
sysctl_dict_parsed = yaml.safe_load(sysctl_dict)
except yaml.YAMLError:
log("Error parsing YAML sysctl_dict: {}".format(sysctl_dict),
level=ERROR)
return
with open(sysctl_file, "w") as fd:
for key, value in sysctl_dict_parsed.items():
fd.write("{}={}\n".format(key, value))
log("Updating sysctl_file: %s values: %s" % (sysctl_file, sysctl_dict_parsed),
level=DEBUG)
check_call(["sysctl", "-p", sysctl_file])
def _metadata_unit(unit):
"""Given the name of a unit (e.g. apache2/0), get the unit charm's
metadata.yaml. Very similar to metadata() but allows us to inspect
other units. Unit needs to be co-located, such as a subordinate or
principal/primary.
:returns: metadata.yaml as a python object.
"""
basedir = os.sep.join(charm_dir().split(os.sep)[:-2])
unitdir = 'unit-{}'.format(unit.replace(os.sep, '-'))
joineddir = os.path.join(basedir, unitdir, 'charm', 'metadata.yaml')
if not os.path.exists(joineddir):
return None
with open(joineddir) as md:
return yaml.safe_load(md)
def _get_user_provided_overrides(modules):
"""Load user-provided config overrides.
:param modules: stack modules to lookup in user overrides yaml file.
:returns: overrides dictionary.
"""
overrides = os.path.join(os.environ['JUJU_CHARM_DIR'],
'hardening.yaml')
if os.path.exists(overrides):
log("Found user-provided config overrides file '%s'" %
(overrides), level=DEBUG)
settings = yaml.safe_load(open(overrides))
if settings and settings.get(modules):
log("Applying '%s' overrides" % (modules), level=DEBUG)
return settings.get(modules)
log("No overrides found for '%s'" % (modules), level=DEBUG)
else:
log("No hardening config overrides file '%s' found in charm "
"root dir" % (overrides), level=DEBUG)
return {}
def create(sysctl_dict, sysctl_file):
"""Creates a sysctl.conf file from a YAML associative array
:param sysctl_dict: a YAML-formatted string of sysctl options eg "{ 'kernel.max_pid': 1337 }"
:type sysctl_dict: str
:param sysctl_file: path to the sysctl file to be saved
:type sysctl_file: str or unicode
:returns: None
"""
try:
sysctl_dict_parsed = yaml.safe_load(sysctl_dict)
except yaml.YAMLError:
log("Error parsing YAML sysctl_dict: {}".format(sysctl_dict),
level=ERROR)
return
with open(sysctl_file, "w") as fd:
for key, value in sysctl_dict_parsed.items():
fd.write("{}={}\n".format(key, value))
log("Updating sysctl_file: %s values: %s" % (sysctl_file, sysctl_dict_parsed),
level=DEBUG)
check_call(["sysctl", "-p", sysctl_file])
def _get_user_provided_overrides(modules):
"""Load user-provided config overrides.
:param modules: stack modules to lookup in user overrides yaml file.
:returns: overrides dictionary.
"""
overrides = os.path.join(os.environ['JUJU_CHARM_DIR'],
'hardening.yaml')
if os.path.exists(overrides):
log("Found user-provided config overrides file '%s'" %
(overrides), level=DEBUG)
settings = yaml.safe_load(open(overrides))
if settings and settings.get(modules):
log("Applying '%s' overrides" % (modules), level=DEBUG)
return settings.get(modules)
log("No overrides found for '%s'" % (modules), level=DEBUG)
else:
log("No hardening config overrides file '%s' found in charm "
"root dir" % (overrides), level=DEBUG)
return {}
def create(sysctl_dict, sysctl_file):
"""Creates a sysctl.conf file from a YAML associative array
:param sysctl_dict: a YAML-formatted string of sysctl options eg "{ 'kernel.max_pid': 1337 }"
:type sysctl_dict: str
:param sysctl_file: path to the sysctl file to be saved
:type sysctl_file: str or unicode
:returns: None
"""
try:
sysctl_dict_parsed = yaml.safe_load(sysctl_dict)
except yaml.YAMLError:
log("Error parsing YAML sysctl_dict: {}".format(sysctl_dict),
level=ERROR)
return
with open(sysctl_file, "w") as fd:
for key, value in sysctl_dict_parsed.items():
fd.write("{}={}\n".format(key, value))
log("Updating sysctl_file: %s values: %s" % (sysctl_file, sysctl_dict_parsed),
level=DEBUG)
check_call(["sysctl", "-p", sysctl_file])
def _metadata_unit(unit):
"""Given the name of a unit (e.g. apache2/0), get the unit charm's
metadata.yaml. Very similar to metadata() but allows us to inspect
other units. Unit needs to be co-located, such as a subordinate or
principal/primary.
:returns: metadata.yaml as a python object.
"""
basedir = os.sep.join(charm_dir().split(os.sep)[:-2])
unitdir = 'unit-{}'.format(unit.replace(os.sep, '-'))
joineddir = os.path.join(basedir, unitdir, 'charm', 'metadata.yaml')
if not os.path.exists(joineddir):
return None
with open(joineddir) as md:
return yaml.safe_load(md)
def load_config():
'''Walk backwords from __file__ looking for config.yaml,
load and return the 'options' section'
'''
config = None
f = __file__
while config is None:
d = os.path.dirname(f)
if os.path.isfile(os.path.join(d, 'config.yaml')):
config = os.path.join(d, 'config.yaml')
break
f = d
if not config:
logging.error('Could not find config.yaml in any parent directory '
'of %s. ' % file)
raise Exception
with open(config) as f:
return yaml.safe_load(f)['options']
def _metadata_unit(unit):
"""Given the name of a unit (e.g. apache2/0), get the unit charm's
metadata.yaml. Very similar to metadata() but allows us to inspect
other units. Unit needs to be co-located, such as a subordinate or
principal/primary.
:returns: metadata.yaml as a python object.
"""
basedir = os.sep.join(charm_dir().split(os.sep)[:-2])
unitdir = 'unit-{}'.format(unit.replace(os.sep, '-'))
joineddir = os.path.join(basedir, unitdir, 'charm', 'metadata.yaml')
if not os.path.exists(joineddir):
return None
with open(joineddir) as md:
return yaml.safe_load(md)
def create(sysctl_dict, sysctl_file):
"""Creates a sysctl.conf file from a YAML associative array
:param sysctl_dict: a YAML-formatted string of sysctl options eg "{ 'kernel.max_pid': 1337 }"
:type sysctl_dict: str
:param sysctl_file: path to the sysctl file to be saved
:type sysctl_file: str or unicode
:returns: None
"""
try:
sysctl_dict_parsed = yaml.safe_load(sysctl_dict)
except yaml.YAMLError:
log("Error parsing YAML sysctl_dict: {}".format(sysctl_dict),
level=ERROR)
return
with open(sysctl_file, "w") as fd:
for key, value in sysctl_dict_parsed.items():
fd.write("{}={}\n".format(key, value))
log("Updating sysctl_file: %s values: %s" % (sysctl_file, sysctl_dict_parsed),
level=DEBUG)
check_call(["sysctl", "-p", sysctl_file])
def _metadata_unit(unit):
"""Given the name of a unit (e.g. apache2/0), get the unit charm's
metadata.yaml. Very similar to metadata() but allows us to inspect
other units. Unit needs to be co-located, such as a subordinate or
principal/primary.
:returns: metadata.yaml as a python object.
"""
basedir = os.sep.join(charm_dir().split(os.sep)[:-2])
unitdir = 'unit-{}'.format(unit.replace(os.sep, '-'))
joineddir = os.path.join(basedir, unitdir, 'charm', 'metadata.yaml')
if not os.path.exists(joineddir):
return None
with open(joineddir) as md:
return yaml.safe_load(md)
def _get_user_provided_overrides(modules):
"""Load user-provided config overrides.
:param modules: stack modules to lookup in user overrides yaml file.
:returns: overrides dictionary.
"""
overrides = os.path.join(os.environ['JUJU_CHARM_DIR'],
'hardening.yaml')
if os.path.exists(overrides):
log("Found user-provided config overrides file '%s'" %
(overrides), level=DEBUG)
settings = yaml.safe_load(open(overrides))
if settings and settings.get(modules):
log("Applying '%s' overrides" % (modules), level=DEBUG)
return settings.get(modules)
log("No overrides found for '%s'" % (modules), level=DEBUG)
else:
log("No hardening config overrides file '%s' found in charm "
"root dir" % (overrides), level=DEBUG)
return {}
def load_config():
'''
Walk backwords from __file__ looking for config.yaml, load and return the
'options' section'
'''
config = None
f = __file__
while config is None:
d = os.path.dirname(f)
if os.path.isfile(os.path.join(d, 'config.yaml')):
config = os.path.join(d, 'config.yaml')
break
f = d
if not config:
logging.error('Could not find config.yaml in any parent directory '
'of %s. ' % f)
raise Exception
return yaml.safe_load(open(config).read())['options']
def create(sysctl_dict, sysctl_file):
"""Creates a sysctl.conf file from a YAML associative array
:param sysctl_dict: a YAML-formatted string of sysctl options eg "{ 'kernel.max_pid': 1337 }"
:type sysctl_dict: str
:param sysctl_file: path to the sysctl file to be saved
:type sysctl_file: str or unicode
:returns: None
"""
try:
sysctl_dict_parsed = yaml.safe_load(sysctl_dict)
except yaml.YAMLError:
log("Error parsing YAML sysctl_dict: {}".format(sysctl_dict),
level=ERROR)
return
with open(sysctl_file, "w") as fd:
for key, value in sysctl_dict_parsed.items():
fd.write("{}={}\n".format(key, value))
log("Updating sysctl_file: %s values: %s" % (sysctl_file, sysctl_dict_parsed),
level=DEBUG)
check_call(["sysctl", "-p", sysctl_file])
def _metadata_unit(unit):
"""Given the name of a unit (e.g. apache2/0), get the unit charm's
metadata.yaml. Very similar to metadata() but allows us to inspect
other units. Unit needs to be co-located, such as a subordinate or
principal/primary.
:returns: metadata.yaml as a python object.
"""
basedir = os.sep.join(charm_dir().split(os.sep)[:-2])
unitdir = 'unit-{}'.format(unit.replace(os.sep, '-'))
joineddir = os.path.join(basedir, unitdir, 'charm', 'metadata.yaml')
if not os.path.exists(joineddir):
return None
with open(joineddir) as md:
return yaml.safe_load(md)
