Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_get_yaml_data_duplicateanchor_error(
self, capsys, quiet_logger, tmp_path_factory
):
yp = get_yaml_editor()
content = """---
aliases:
- &anchor value1
- &anchor value2
"""
yaml_file = create_temp_yaml_file(tmp_path_factory, content)
assert None == get_yaml_data(yp, quiet_logger, yaml_file)
captured = capsys.readouterr()
assert -1 < captured.err.find("Duplicate YAML Anchor detected")
def test_yield_raw_children_direct(self, tmp_path_factory, quiet_logger):
from yamlpath.enums import PathSeperators, PathSearchMethods
from yamlpath.path import SearchTerms
from yamlpath.func import get_yaml_data, get_yaml_editor
from yamlpath.commands.yaml_paths import yield_children
from itertools import zip_longest
content = """some raw text value
"""
processor = get_yaml_editor()
yaml_file = create_temp_yaml_file(tmp_path_factory, content)
yaml_data = get_yaml_data(processor, quiet_logger, yaml_file)
seen_anchors = []
assertions = ["/"]
results = []
for assertion, path in zip_longest(assertions, yield_children(
quiet_logger, yaml_data,
SearchTerms(False, PathSearchMethods.STARTS_WITH, "*", "some"),
PathSeperators.FSLASH, "", seen_anchors, search_anchors=False,
include_key_aliases=False, include_value_aliases=False
)):
assert assertion == str(path)
def test_yield_seq_children_direct(self, tmp_path_factory, quiet_logger):
from yamlpath.enums import PathSeperators, PathSearchMethods
from yamlpath.path import SearchTerms
from yamlpath.func import get_yaml_data, get_yaml_editor
from yamlpath.commands.yaml_paths import yield_children
from itertools import zip_longest
content = """---
- &value Test value
- value
- *value
"""
processor = get_yaml_editor()
yaml_file = create_temp_yaml_file(tmp_path_factory, content)
yaml_data = get_yaml_data(processor, quiet_logger, yaml_file)
seen_anchors = []
assertions = ["/&value", "/[1]"]
results = []
for assertion, path in zip_longest(assertions, yield_children(
quiet_logger, yaml_data,
SearchTerms(False, PathSearchMethods.EQUALS, "*", "value"),
PathSeperators.FSLASH, "", seen_anchors, search_anchors=True,
include_aliases=False
)):
assert assertion == str(path)
def test_get_yaml_data_keyboardinterrupt_error(
self, capsys, quiet_logger, tmp_path_factory,
force_ruamel_load_keyboardinterrupt
):
yp = get_yaml_editor()
content = """---
no: ''
"""
yaml_file = create_temp_yaml_file(tmp_path_factory, content)
assert None == get_yaml_data(yp, quiet_logger, yaml_file)
captured = capsys.readouterr()
assert -1 < captured.err.find("keyboard interrupt")
def test_get_yaml_data_construction_error(
self, capsys, quiet_logger, tmp_path_factory
):
yp = get_yaml_editor()
content = """---
missing:
<<:
"""
yaml_file = create_temp_yaml_file(tmp_path_factory, content)
assert None == get_yaml_data(yp, quiet_logger, yaml_file)
captured = capsys.readouterr()
assert -1 < captured.err.find("YAML construction error")
def test_get_yaml_data_duplicatekey_error(
self, capsys, quiet_logger, tmp_path_factory
):
yp = get_yaml_editor()
content = """---
key: value1
key: value2
"""
yaml_file = create_temp_yaml_file(tmp_path_factory, content)
assert None == get_yaml_data(yp, quiet_logger, yaml_file)
captured = capsys.readouterr()
assert -1 < captured.err.find("Duplicate Hash key detected")
if args.value:
new_value = args.value
elif args.stdin:
new_value = ''.join(sys.stdin.readlines())
elif args.file:
with open(args.file, 'r') as fhnd:
new_value = fhnd.read().rstrip()
elif args.random is not None:
new_value = ''.join(
secrets.choice(
string.ascii_uppercase + string.ascii_lowercase + string.digits
) for _ in range(args.random)
)
# Prep the YAML parser
yaml = get_yaml_editor()
# Attempt to open the YAML file; check for parsing errors
yaml_data = get_yaml_data(yaml, log, args.yaml_file)
if yaml_data is None:
# An error message has already been logged
sys.exit(1)
# Load the present value at the specified YAML Path
change_node_coordinates = []
old_format = YAMLValueFormats.DEFAULT
processor = EYAMLProcessor(
log, yaml_data, binary=args.eyaml,
publickey=args.publickey, privatekey=args.privatekey)
try:
for node_coordinate in processor.get_nodes(
change_path, mustexist=(args.mustexist or args.saveto),
if args.onlykeynames:
search_values = False
search_keys = True
elif args.keynames:
search_keys = True
if args.include_aliases is IncludeAliases.INCLUDE_ALL_ALIASES:
include_key_aliases = True
include_value_aliases = True
elif args.include_aliases is IncludeAliases.INCLUDE_KEY_ALIASES:
include_key_aliases = True
elif args.include_aliases is IncludeAliases.INCLUDE_VALUE_ALIASES:
include_value_aliases = True
# Prepare the YAML processor
yaml = get_yaml_editor()
processor = EYAMLProcessor(
log, None, binary=args.eyaml,
publickey=args.publickey, privatekey=args.privatekey)
# Process the input file(s)
exit_state = 0
# pylint: disable=too-many-nested-blocks
for yaml_file in args.yaml_files:
# Try to open the file
yaml_data = get_yaml_data(yaml, log, yaml_file)
if yaml_data is None:
# An error message has already been logged
exit_state = 3
continue
def main():
"""Main code."""
# Process any command-line arguments
args = processcli()
log = ConsolePrinter(args)
validateargs(args, log)
processor = EYAMLProcessor(log, None, binary=args.eyaml)
# Prep the YAML parser
yaml = get_yaml_editor()
# Process the input file(s)
in_file_count = len(args.yaml_files)
exit_state = 0
for yaml_file in args.yaml_files:
file_changed = False
backup_file = yaml_file + ".bak"
seen_anchors = []
# Each YAML_FILE must actually be a file
if not isfile(yaml_file):
log.error("Not a file: {}".format(yaml_file))
exit_state = 2
continue
# Don't bother with the file change update when there's only one input