Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
after_repo = Repository(repo.path)
nodes_metadata_after = {}
for node_name in nodes_metadata_before:
if QUIT_EVENT.is_set():
exit(1)
nodes_metadata_after[node_name] = \
after_repo.get_node(node_name).metadata_hash()
node_hashes_before = sorted(
["{}\t{}".format(i, h) for i, h in nodes_metadata_before.items()]
)
node_hashes_after = sorted(
["{}\t{}".format(i, h) for i, h in nodes_metadata_after.items()]
)
io.stdout("\n".join(
filter(
lambda line: line.startswith("+") or line.startswith("-"),
unified_diff(
node_hashes_before,
node_hashes_after,
fromfile=_("before"),
tofile=_("after"),
lineterm='',
n=0,
),
),
))
for epilogue in epilogues:
epilogue()
item_b_dict = item_b.cdict()
if (
item.startswith("file:")
and item_a.attributes['content_type'] not in ('base64', 'binary')
and item_b.attributes['content_type'] not in ('base64', 'binary')
and len(item_a.content) < DIFF_MAX_FILE_SIZE
and len(item_b.content) < DIFF_MAX_FILE_SIZE
):
del item_a_dict['content_hash']
del item_b_dict['content_hash']
item_a_dict['content'] = item_a.content
item_b_dict['content'] = item_b.content
relevant_keys = diff_keys(item_a_dict, item_b_dict)
io.stdout(item_a.ask(item_b_dict, item_a_dict, relevant_keys))
def handle_result(task_id, return_value, duration):
io.progress_advance()
io.stdout(_("{x} {node} locked with ID {id} (expires in {exp})").format(
x=green("✓"),
node=bold(task_id.ljust(max_node_name_length)),
id=return_value,
exp=args['expiry'],
))
if args['dict']:
if args['group_membership']:
if target_type in ('node', 'repo'):
for group in target.groups:
io.stdout(group.name)
else:
for node in target.nodes:
io.stdout(node.name)
elif args['metadata']:
for node in target.nodes:
io.stdout("{}\t{}".format(node.name, node.metadata_hash()))
else:
cdict = target.cached_cdict if args['item'] else target.cdict
if cdict is None:
io.stdout("REMOVE")
else:
for key, value in sorted(cdict.items()):
io.stdout("{}\t{}".format(key, value) if args['item'] else "{} {}".format(value, key))
else:
if args['group_membership']:
io.stdout(target.group_membership_hash())
elif args['metadata']:
io.stdout(target.metadata_hash())
else:
io.stdout(target.hash())
def diff_metadata(node_a, node_b):
node_a_metadata = metadata_to_json(node_a.metadata).splitlines()
node_b_metadata = metadata_to_json(node_b.metadata).splitlines()
io.stdout("\n".join(unified_diff(
node_a_metadata,
node_b_metadata,
fromfile=node_a.name,
tofile=node_b.name,
lineterm='',
)))
def remove_dummy_nodes(targets):
_targets = []
for node in targets:
if node.dummy:
io.stdout(_("{x} {node} is a dummy node").format(node=bold(node.name), x=yellow("»")))
else:
_targets.append(node)
return _targets
statedict = item.sdict()
else:
statedict = item.cdict()
if statedict is None:
io.stdout("REMOVE")
else:
if args['attr']:
io.stdout(repr(statedict[args['attr']]))
else:
io.stdout(statedict_to_json(statedict, pretty=True))
else:
for item in sorted(node.items):
if args['show_repr']:
io.stdout(repr(item))
else:
io.stdout(item.id)
def run_on_node(node, command, skip_list):
if node.dummy:
io.stdout(_("{x} {node} is a dummy node").format(node=bold(node.name), x=yellow("»")))
return None
if node.name in skip_list:
io.stdout(_("{x} {node} skipped by --resume-file").format(node=bold(node.name), x=yellow("»")))
return None
try:
node.repo.hooks.node_run_start(
node.repo,
node,
command,
)
except SkipNode as exc:
io.stdout(_("{x} {node} skipped by hook ({reason})").format(
node=bold(node.name),
reason=str(exc) or _("no reason given"),
x=yellow("»"),
))
return None
def bw_zen(repo, args):
io.stdout(ZEN)
"{0:.1f}%".format(totals['health']),
format_duration(total_duration),
])
alignments = {
1: 'right',
2: 'right',
3: 'right',
4: 'right',
5: 'right',
6: 'right',
7: 'right',
}
for line in render_table(rows, alignments=alignments):
io.stdout("{x} {line}".format(x=blue("i"), line=line))