def generate_config(args):
setup_yaml()
if args.existing:
existing = next(yaml.safe_load_all(open(args.existing, 'r', encoding='utf-8')))
else:
existing = None
cfg, failed = chores.generate_chore_config(args.db, args.bookmark, existing)
with open(args.output, 'w', encoding='utf-8') as f:
yaml.dump_all((cfg, failed), f, default_flow_style=False)
logging.info('Done.')
python类safe_load_all()的实例源码
def invoke(self):
if not self.ctx.obj.get('api', False):
with open(self.filename) as f:
armada = Armada(
list(yaml.safe_load_all(f.read())),
self.disable_update_pre,
self.disable_update_post,
self.enable_chart_cleanup,
self.dry_run,
self.set,
self.wait,
self.timeout,
self.tiller_host,
self.tiller_port,
self.values)
resp = armada.sync()
self.output(resp)
else:
query = {
'disable_update_post': self.disable_update_post,
'disable_update_pre': self.disable_update_pre,
'dry_run': self.dry_run,
'enable_chart_cleanup': self.enable_chart_cleanup,
'tiller_host': self.tiller_host,
'tiller_port': self.tiller_port,
'timeout': self.timeout,
'wait': self.wait
}
client = self.ctx.obj.get('CLIENT')
with open(self.filename, 'r') as f:
resp = client.post_apply(
manifest=f.read(), values=self.values, set=self.set,
query=query)
self.output(resp.get('message'))
def test_lint_armada_yaml_pass(self):
template = '{}/templates/valid_armada_document.yaml'.format(
self.basepath)
document = yaml.safe_load_all(open(template).read())
resp = lint.validate_armada_documents(document)
self.assertTrue(resp)
def test_lint_armada_manifest_no_groups(self):
template_manifest = """
schema: armada/Manifest/v1
metadata:
schema: metadata/Document/v1
name: example-manifest
data:
release_prefix: example
"""
document = yaml.safe_load_all(template_manifest)
with self.assertRaises(Exception):
lint.validate_armada_documents(document)
def test_lint_validate_manifest_pass(self):
template_manifest = """
schema: armada/Manifest/v1
metadata:
schema: metadata/Document/v1
name: example-manifest
data:
release_prefix: example
chart_groups:
- example-group
"""
document = yaml.safe_load_all(template_manifest)
self.assertTrue(lint.validate_manifest_document(document))
def test_lint_validate_group_pass(self):
template_manifest = """
schema: armada/ChartGroup/v1
metadata:
schema: metadata/Document/v1
name: example-manifest
data:
description: this is sample
chart_group:
- example-group
"""
document = yaml.safe_load_all(template_manifest)
self.assertTrue(lint.validate_chart_group_document(document))
def test_lint_validate_group_no_chart_group(self):
template_manifest = """
schema: armada/ChartGroup/v1
metadata:
schema: metadata/Document/v1
name: example-manifest
data:
description: this is sample
"""
document = yaml.safe_load_all(template_manifest)
with self.assertRaises(Exception):
lint.validate_chart_group_document(document)
def test_lint_validate_chart_pass(self):
template_manifest = """
schema: armada/Chart/v1
metadata:
schema: metadata/Document/v1
name: example-chart
data:
name: keystone
release: keystone
namespace: undercloud
timeout: 100
install:
no_hooks: false
upgrade:
no_hooks: false
values: {}
source:
type: git
location: git://github.com/example/example
subpath: example-chart
reference: master
dependencies:
- dep-chart
"""
document = yaml.safe_load_all(template_manifest)
self.assertTrue(lint.validate_chart_document(document))
def test_lint_validate_chart_no_release(self):
template_manifest = """
schema: armada/Chart/v1
metadata:
schema: metadata/Document/v1
name: example-chart
data:
name: keystone
namespace: undercloud
timeout: 100
install:
no_hooks: false
upgrade:
no_hooks: false
values: {}
source:
type: git
location: git://github.com/example/example
subpath: example-chart
reference: master
dependencies:
- dep-chart
"""
document = yaml.safe_load_all(template_manifest)
with self.assertRaises(Exception):
lint.validate_chart_document(document)
def test_pre_flight_ops(self, mock_tiller, mock_lint, mock_git):
'''Test pre-flight checks and operations'''
armada = Armada('')
armada.tiller = mock_tiller
armada.documents = yaml.safe_load_all(self.test_yaml)
armada.config = Manifest(armada.documents).get_manifest()
CHART_SOURCES = [('git://github.com/dummy/armada', 'chart_1'),
('/tmp/dummy/armada', 'chart_2')]
# mock methods called by pre_flight_ops()
mock_tiller.tiller_status.return_value = True
mock_lint.valid_manifest.return_value = True
mock_git.git_clone.return_value = CHART_SOURCES[0][0]
armada.pre_flight_ops()
mock_git.git_clone.assert_called_once_with(CHART_SOURCES[0][0],
'master')
for group in armada.config.get('armada').get('charts'):
for counter, chart in enumerate(group.get('chart_group')):
self.assertEqual(
chart.get('chart').get('source_dir')[0],
CHART_SOURCES[counter][0])
self.assertEqual(
chart.get('chart').get('source_dir')[1],
CHART_SOURCES[counter][1])
def test_find_document_type_valid(self):
with open(self.base_manifest) as f:
doc_obj = list(yaml.safe_load_all(f.read()))
ovr = Override(doc_obj)
test_group = ovr.find_document_type('chart_group')
self.assertEqual(test_group, const.DOCUMENT_GROUP)
test_chart = ovr.find_document_type('chart')
self.assertEqual(test_chart, const.DOCUMENT_CHART)
test_manifest = ovr.find_document_type('manifest')
self.assertEqual(test_manifest, const.DOCUMENT_MANIFEST)
def test_find_document_type_invalid(self):
with self.assertRaises(Exception):
with open(self.base_manifest) as f:
doc_obj = list(yaml.safe_load_all(f.read()))
ovr = Override(doc_obj)
ovr.find_document_type('charts')
def test_find_manifest_document_valid(self):
expected = "{}/templates/override-{}-expected.yaml".format(
self.basepath, '02')
with open(self.base_manifest) as f, open(expected) as e:
doc_path = ['chart', 'blog-1']
doc_obj = list(yaml.safe_load_all(f.read()))
ovr = Override(doc_obj).find_manifest_document(doc_path)
expected_doc = list(yaml.safe_load_all(e.read()))[0]
self.assertEqual(ovr, expected_doc)
def _load_yaml_file(self, doc):
'''
Retrieve yaml file as a dictionary.
'''
try:
with open(doc) as f:
return list(yaml.safe_load_all(f.read()))
except IOError:
raise override_exceptions.InvalidOverrideFileException(doc)
def post_apply(self, manifest=None, values=None, set=None, query=None):
if values or set:
document = list(yaml.safe_load_all(manifest))
override = Override(
document, overrides=set, values=values).update_manifests()
manifest = yaml.dump(override)
endpoint = self._set_endpoint('1.0', 'apply')
resp = self.session.post(endpoint, body=manifest, query=query)
self._check_response(resp)
return resp.json()
def __init__(self, filepath):
with open(filepath) as f:
RequirementsParser.Inner._rbac_map = \
list(yaml.safe_load_all(f))
def _load_schemas():
'''
Fills the cache of known schemas
'''
schema_dir = _get_schema_dir()
for schema_file in os.listdir(schema_dir):
with open(os.path.join(schema_dir, schema_file)) as f:
for schema in yaml.safe_load_all(f):
name = schema['metadata']['name']
if name in SCHEMAS:
raise RuntimeError(
'Duplicate schema specified for: %s' % name)
SCHEMAS[name] = schema['data']
def from_streams(cls, *, streams, **kwargs):
documents = []
for stream in streams:
stream_name = getattr(stream, 'name')
if stream_name is not None:
LOG.info('Loading documents from %s', stream_name)
stream_documents = list(yaml.safe_load_all(stream))
validation.check_schemas(stream_documents)
if stream_name is not None:
LOG.info('Successfully validated documents from %s',
stream_name)
documents.extend(stream_documents)
return cls(documents=documents, **kwargs)
def from_design_ref(cls, design_ref):
response = requests.get(design_ref)
response.raise_for_status()
documents = list(yaml.safe_load_all(response.text))
validation.check_schemas(documents)
return cls(documents=documents)
def load_yaml_config(database, subreddit, config_file):
"""Parse the given file and return a list of Browsers."""
with config_file.open() as file:
config = yaml.safe_load_all(file)
return parse_subreddit_config(database, subreddit, config)
