Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def emit(self, ctx, modules, fd):
"""Main control function.
Set up the top-level parts of the sample document, then process
recursively all nodes in all data trees, and finally emit the
sample XML document.
"""
if ctx.opts.sample_path is not None:
path = ctx.opts.sample_path.split('/')
if path[0] == '':
path = path[1:]
else:
path = []
for (epos, etag, eargs) in ctx.errors:
if error.is_error(error.err_level(etag)):
raise error.EmitError(
"sample-xml-skeleton plugin needs a valid module")
self.doctype = ctx.opts.doctype
if self.doctype not in ("config", "data"):
raise error.EmitError("Unsupported document type: %s" %
self.doctype)
self.annots = ctx.opts.sample_annots
self.defaults = ctx.opts.sample_defaults
self.node_handler = {
"container": self.container,
"leaf": self.leaf,
"anyxml": self.anyxml,
"choice": self.process_children,
"case": self.process_children,
"list": self.list,
"leaf-list": self.leaf_list
def emit(self, ctx, modules, fd):
for (epos, etag, eargs) in ctx.errors:
if ((epos.top is None or epos.top.arg in self.mods) and
error.is_error(error.err_level(etag))):
self.fatal("%s contains errors" % epos.top.arg)
if ctx.opts.uml_pages_layout is not None:
if re.match('[0-9]x[0-9]', ctx.opts.uml_pages_layout) is None:
self.fatal("Illegal page split option %s, should be [0-9]x[0-9], example 2x2" % ctx.opts.uml_pages_layout)
umldoc = uml_emitter(ctx)
umldoc.emit(modules, fd)
def emit(self, ctx, modules, fd):
# cannot do this unless everything is ok for our module
modulenames = [m.arg for m in modules]
for (epos, etag, eargs) in ctx.errors:
if ((epos.top is None or epos.top.arg in modulenames) and
error.is_error(error.err_level(etag))):
raise error.EmitError("%s contains errors" % epos.top.arg)
emit_depend(ctx, modules, fd)
def emit_dsdl(ctx, modules, fd):
for (epos, etag, eargs) in ctx.errors:
if error.is_error(error.err_level(etag)):
raise error.EmitError("DSDL translation needs a valid module")
schema = HybridDSDLSchema().from_modules(modules,
ctx.opts.dsdl_no_dublin_core,
ctx.opts.dsdl_no_documentation,
ctx.opts.dsdl_record_defs,
ctx.opts.dsdl_lax_yang_version,
debug=0)
fd.write(schema.serialize())
nbr_option_specified = 0
if ctx.opts.generate_sid_file is not None:
nbr_option_specified += 1
if ctx.opts.update_sid_file is not None:
nbr_option_specified += 1
if ctx.opts.check_sid_file is not None:
nbr_option_specified += 1
if nbr_option_specified == 0:
return
if nbr_option_specified > 1:
sys.stderr.write("Invalid option, only one process on .sid file can be requested.\n")
return
fatal_error = False
for (epos, etag, eargs) in ctx.errors:
if not error.is_warning(error.err_level(etag)):
fatal_error = True
if fatal_error or (ctx.errors != [] and ctx.opts.check_sid_file is not None):
sys.stderr.write("Invalid YANG module\n")
return
sid_file = SidFile()
if ctx.opts.sid_registration_info:
sid_file.sid_registration_info = True
if ctx.opts.generate_sid_file is not None:
sid_file.range = ctx.opts.generate_sid_file
sid_file.is_consistent = False
sid_file.sid_file_created = True
def print_pyang_output(ctx):
err = ''
out = ''
for (epos, etag, eargs) in ctx.errors:
elevel = error.err_level(etag)
if error.is_warning(elevel):
kind = "warning"
else:
kind = "error"
err += str(epos) + ': %s: ' % kind + \
error.err_to_str(etag, eargs) + '\n'
return err, out
self.ctx.validate()
def keyfun(e):
if e[0].ref == filenames[0]:
return 0
else:
return 1
self.ctx.errors.sort(key=lambda e: (e[0].ref, e[0].line))
if len(filenames) > 0:
# first print error for the first filename given
self.ctx.errors.sort(key=keyfun)
error_messages = []
for (epos, etag, eargs) in self.ctx.errors:
elevel = error.err_level(etag)
if error.is_warning(elevel):
logger.warning('%s: %s\n' %
(str(epos), error.err_to_str(etag, eargs)))
else:
err_msg = '%s: %s\n' % (str(epos), error.err_to_str(etag, eargs))
logger.error(err_msg)
error_messages.append(err_msg)
if len(error_messages) > 0:
err_msg = '\n'.join(error_messages)
raise YdkGenException(err_msg)