| # |
| # Copyright (C) 2018 Codethink Limited |
| # |
| # This program is free software; you can redistribute it and/or |
| # modify it under the terms of the GNU Lesser General Public |
| # License as published by the Free Software Foundation; either |
| # version 2 of the License, or (at your option) any later version. |
| # |
| # This library is distributed in the hope that it will be useful, |
| # but WITHOUT ANY WARRANTY; without even the implied warranty of |
| # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU |
| # Lesser General Public License for more details. |
| # |
| # You should have received a copy of the GNU Lesser General Public |
| # License along with this library. If not, see <http://www.gnu.org/licenses/>. |
| # |
| # Authors: |
| # Tristan Van Berkom <tristan.vanberkom@codethink.co.uk> |
| |
| import sys |
| import collections |
| import string |
| from copy import deepcopy |
| from contextlib import ExitStack |
| from pathlib import Path |
| |
| from ruamel import yaml |
| from ruamel.yaml.representer import SafeRepresenter, RoundTripRepresenter |
| from ruamel.yaml.constructor import RoundTripConstructor |
| from ._exceptions import LoadError, LoadErrorReason |
| |
| # This overrides the ruamel constructor to treat everything as a string |
| RoundTripConstructor.add_constructor(u'tag:yaml.org,2002:int', RoundTripConstructor.construct_yaml_str) |
| RoundTripConstructor.add_constructor(u'tag:yaml.org,2002:float', RoundTripConstructor.construct_yaml_str) |
| |
| # We store information in the loaded yaml on a DictProvenance |
| # stored in all dictionaries under this key |
| PROVENANCE_KEY = '__bst_provenance_info' |
| |
| |
| # Provides information about file for provenance |
| # |
| # Args: |
| # name (str): Full path to the file |
| # shortname (str): Relative path to the file |
| # project (Project): Project where the shortname is relative from |
| class ProvenanceFile(): |
| def __init__(self, name, shortname, project): |
| self.name = name |
| self.shortname = shortname |
| self.project = project |
| |
| |
| # Provenance tracks the origin of a given node in the parsed dictionary. |
| # |
| # Args: |
| # node (dict, list, value): A binding to the originally parsed value |
| # filename (string): The filename the node was loaded from |
| # toplevel (dict): The toplevel of the loaded file, suitable for later dumps |
| # line (int): The line number where node was parsed |
| # col (int): The column number where node was parsed |
| # |
| class Provenance(): |
| def __init__(self, filename, node, toplevel, line=0, col=0): |
| self.filename = filename |
| self.node = node |
| self.toplevel = toplevel |
| self.line = line |
| self.col = col |
| |
| # Convert a Provenance to a string for error reporting |
| def __str__(self): |
| filename = self.filename.shortname |
| if self.filename.project and self.filename.project.junction: |
| filename = "{}:{}".format(self.filename.project.junction.name, self.filename.shortname) |
| |
| return "{} [line {:d} column {:d}]".format(filename, self.line, self.col) |
| |
| # Abstract method |
| def clone(self): |
| pass # pragma: nocover |
| |
| |
| # A Provenance for dictionaries, these are stored in the copy of the |
| # loaded YAML tree and track the provenance of all members |
| # |
| class DictProvenance(Provenance): |
| def __init__(self, filename, node, toplevel, line=None, col=None): |
| |
| if line is None or col is None: |
| # Special case for loading an empty dict |
| if hasattr(node, 'lc'): |
| line = node.lc.line + 1 |
| col = node.lc.col |
| else: |
| line = 1 |
| col = 0 |
| |
| super(DictProvenance, self).__init__(filename, node, toplevel, line=line, col=col) |
| |
| self.members = {} |
| |
| def clone(self): |
| provenance = DictProvenance(self.filename, self.node, self.toplevel, |
| line=self.line, col=self.col) |
| |
| provenance.members = { |
| member_name: member.clone() |
| for member_name, member in self.members.items() |
| } |
| return provenance |
| |
| |
| # A Provenance for dict members |
| # |
| class MemberProvenance(Provenance): |
| def __init__(self, filename, parent_dict, member_name, toplevel, |
| node=None, line=None, col=None): |
| |
| if parent_dict is not None: |
| node = parent_dict[member_name] |
| line, col = parent_dict.lc.value(member_name) |
| line += 1 |
| |
| super(MemberProvenance, self).__init__( |
| filename, node, toplevel, line=line, col=col) |
| |
| # Only used if member is a list |
| self.elements = [] |
| |
| def clone(self): |
| provenance = MemberProvenance(self.filename, None, None, self.toplevel, |
| node=self.node, line=self.line, col=self.col) |
| provenance.elements = [e.clone() for e in self.elements] |
| return provenance |
| |
| |
| # A Provenance for list elements |
| # |
| class ElementProvenance(Provenance): |
| def __init__(self, filename, parent_list, index, toplevel, |
| node=None, line=None, col=None): |
| |
| if parent_list is not None: |
| node = parent_list[index] |
| line, col = parent_list.lc.item(index) |
| line += 1 |
| |
| super(ElementProvenance, self).__init__( |
| filename, node, toplevel, line=line, col=col) |
| |
| # Only used if element is a list |
| self.elements = [] |
| |
| def clone(self): |
| provenance = ElementProvenance(self.filename, None, None, self.toplevel, |
| node=self.node, line=self.line, col=self.col) |
| |
| provenance.elements = [e.clone for e in self.elements] |
| return provenance |
| |
| |
| # These exceptions are intended to be caught entirely within |
| # the BuildStream framework, hence they do not reside in the |
| # public exceptions.py |
| class CompositeError(Exception): |
| def __init__(self, path, message): |
| super(CompositeError, self).__init__(message) |
| self.path = path |
| |
| |
| class CompositeTypeError(CompositeError): |
| def __init__(self, path, expected_type, actual_type): |
| super(CompositeTypeError, self).__init__( |
| path, |
| "Error compositing dictionary key '{}', expected source type '{}' " |
| "but received type '{}'" |
| .format(path, expected_type.__name__, actual_type.__name__)) |
| self.expected_type = expected_type |
| self.actual_type = actual_type |
| |
| |
| # Loads a dictionary from some YAML |
| # |
| # Args: |
| # filename (str): The YAML file to load |
| # shortname (str): The filename in shorthand for error reporting (or None) |
| # copy_tree (bool): Whether to make a copy, preserving the original toplevels |
| # for later serialization |
| # |
| # Returns (dict): A loaded copy of the YAML file with provenance information |
| # |
| # Raises: LoadError |
| # |
| def load(filename, shortname=None, copy_tree=False, *, project=None): |
| if not shortname: |
| shortname = filename |
| |
| file = ProvenanceFile(filename, shortname, project) |
| |
| try: |
| with open(filename) as f: |
| return load_data(f, file, copy_tree=copy_tree) |
| except FileNotFoundError as e: |
| raise LoadError(LoadErrorReason.MISSING_FILE, |
| "Could not find file at {}".format(filename)) from e |
| except IsADirectoryError as e: |
| raise LoadError(LoadErrorReason.LOADING_DIRECTORY, |
| "{} is a directory. bst command expects a .bst file." |
| .format(filename)) from e |
| |
| |
| # Like load(), but doesnt require the data to be in a file |
| # |
| def load_data(data, file=None, copy_tree=False): |
| |
| try: |
| contents = yaml.load(data, yaml.loader.RoundTripLoader, preserve_quotes=True) |
| except (yaml.scanner.ScannerError, yaml.composer.ComposerError, yaml.parser.ParserError) as e: |
| raise LoadError(LoadErrorReason.INVALID_YAML, |
| "Malformed YAML:\n\n{}\n\n{}\n".format(e.problem, e.problem_mark)) from e |
| |
| if not isinstance(contents, dict): |
| # Special case allowance for None, when the loaded file has only comments in it. |
| if contents is None: |
| contents = {} |
| else: |
| raise LoadError(LoadErrorReason.INVALID_YAML, |
| "YAML file has content of type '{}' instead of expected type 'dict': {}" |
| .format(type(contents).__name__, file.name)) |
| |
| return node_decorated_copy(file, contents, copy_tree=copy_tree) |
| |
| |
| # Dumps a previously loaded YAML node to a file |
| # |
| # Args: |
| # node (dict): A node previously loaded with _yaml.load() above |
| # filename (str): The YAML file to load |
| # |
| def dump(node, filename=None): |
| with ExitStack() as stack: |
| if filename: |
| from . import utils |
| f = stack.enter_context(utils.save_file_atomic(filename, 'w')) |
| else: |
| f = sys.stdout |
| yaml.round_trip_dump(node, f) |
| |
| |
| # node_decorated_copy() |
| # |
| # Create a copy of a loaded dict tree decorated with Provenance |
| # information, used directly after loading yaml |
| # |
| # Args: |
| # filename (str): The filename |
| # toplevel (node): The toplevel dictionary node |
| # copy_tree (bool): Whether to load a copy and preserve the original |
| # |
| # Returns: A copy of the toplevel decorated with Provinance |
| # |
| def node_decorated_copy(filename, toplevel, copy_tree=False): |
| if copy_tree: |
| result = deepcopy(toplevel) |
| else: |
| result = toplevel |
| |
| node_decorate_dict(filename, result, toplevel, toplevel) |
| |
| return result |
| |
| |
| def node_decorate_dict(filename, target, source, toplevel): |
| provenance = DictProvenance(filename, source, toplevel) |
| target[PROVENANCE_KEY] = provenance |
| |
| for key, value in node_items(source): |
| member = MemberProvenance(filename, source, key, toplevel) |
| provenance.members[key] = member |
| |
| target_value = target.get(key) |
| if isinstance(value, collections.abc.Mapping): |
| node_decorate_dict(filename, target_value, value, toplevel) |
| elif isinstance(value, list): |
| member.elements = node_decorate_list(filename, target_value, value, toplevel) |
| |
| |
| def node_decorate_list(filename, target, source, toplevel): |
| |
| elements = [] |
| |
| for item in source: |
| idx = source.index(item) |
| target_item = target[idx] |
| element = ElementProvenance(filename, source, idx, toplevel) |
| |
| if isinstance(item, collections.abc.Mapping): |
| node_decorate_dict(filename, target_item, item, toplevel) |
| elif isinstance(item, list): |
| element.elements = node_decorate_list(filename, target_item, item, toplevel) |
| |
| elements.append(element) |
| |
| return elements |
| |
| |
| # node_get_provenance() |
| # |
| # Gets the provenance for a node |
| # |
| # Args: |
| # node (dict): a dictionary |
| # key (str): key in the dictionary |
| # indices (list of indexes): Index path, in the case of list values |
| # |
| # Returns: The Provenance of the dict, member or list element |
| # |
| def node_get_provenance(node, key=None, indices=None): |
| |
| provenance = node.get(PROVENANCE_KEY) |
| if provenance and key: |
| provenance = provenance.members.get(key) |
| if provenance and indices is not None: |
| for index in indices: |
| provenance = provenance.elements[index] |
| |
| return provenance |
| |
| |
| # Helper to use utils.sentinel without unconditional utils import, |
| # which causes issues for completion. |
| # |
| # Local private, but defined here because sphinx appears to break if |
| # it's not defined before any functions calling it in default kwarg |
| # values. |
| # |
| def _get_sentinel(): |
| from .utils import _sentinel |
| return _sentinel |
| |
| |
| # node_get() |
| # |
| # Fetches a value from a dictionary node and checks it for |
| # an expected value. Use default_value when parsing a value |
| # which is only optionally supplied. |
| # |
| # Args: |
| # node (dict): The dictionary node |
| # expected_type (type): The expected type for the value being searched |
| # key (str): The key to get a value for in node |
| # indices (list of ints): Optionally decend into lists of lists |
| # |
| # Returns: |
| # The value if found in node, otherwise default_value is returned |
| # |
| # Raises: |
| # LoadError, when the value found is not of the expected type |
| # |
| # Note: |
| # Returned strings are stripped of leading and trailing whitespace |
| # |
| def node_get(node, expected_type, key, indices=None, default_value=_get_sentinel()): |
| value = node.get(key, default_value) |
| provenance = node_get_provenance(node) |
| if value is _get_sentinel(): |
| raise LoadError(LoadErrorReason.INVALID_DATA, |
| "{}: Dictionary did not contain expected key '{}'".format(provenance, key)) |
| |
| path = key |
| if indices is not None: |
| # Implied type check of the element itself |
| value = node_get(node, list, key) |
| for index in indices: |
| value = value[index] |
| path += '[{:d}]'.format(index) |
| |
| # We want to allow None as a valid value for any type |
| if value is None: |
| return None |
| |
| if not isinstance(value, expected_type): |
| # Attempt basic conversions if possible, typically we want to |
| # be able to specify numeric values and convert them to strings, |
| # but we dont want to try converting dicts/lists |
| try: |
| if (expected_type == bool and isinstance(value, str)): |
| # Dont coerce booleans to string, this makes "False" strings evaluate to True |
| if value == 'true' or value == 'True': |
| value = True |
| elif value == 'false' or value == 'False': |
| value = False |
| else: |
| raise ValueError() |
| elif not (expected_type == list or |
| expected_type == dict or |
| isinstance(value, (list, dict))): |
| value = expected_type(value) |
| else: |
| raise ValueError() |
| except (ValueError, TypeError): |
| provenance = node_get_provenance(node, key=key, indices=indices) |
| raise LoadError(LoadErrorReason.INVALID_DATA, |
| "{}: Value of '{}' is not of the expected type '{}'" |
| .format(provenance, path, expected_type.__name__)) |
| |
| # Trim it at the bud, let all loaded strings from yaml be stripped of whitespace |
| if isinstance(value, str): |
| value = value.strip() |
| |
| return value |
| |
| |
| # node_get_project_path() |
| # |
| # Fetches a project path from a dictionary node and validates it |
| # |
| # Paths are asserted to never lead to a directory outside of the project |
| # directory. In addition, paths can not point to symbolic links, fifos, |
| # sockets and block/character devices. |
| # |
| # The `check_is_file` and `check_is_dir` parameters can be used to |
| # perform additional validations on the path. Note that an exception |
| # will always be raised if both parameters are set to ``True``. |
| # |
| # Args: |
| # node (dict): A dictionary loaded from YAML |
| # key (str): The key whose value contains a path to validate |
| # project_dir (str): The project directory |
| # check_is_file (bool): If ``True`` an error will also be raised |
| # if path does not point to a regular file. |
| # Defaults to ``False`` |
| # check_is_dir (bool): If ``True`` an error will be also raised |
| # if path does not point to a directory. |
| # Defaults to ``False`` |
| # Returns: |
| # (str): The project path |
| # |
| # Raises: |
| # (LoadError): In case that the project path is not valid or does not |
| # exist |
| # |
| def node_get_project_path(node, key, project_dir, *, |
| check_is_file=False, check_is_dir=False): |
| path_str = node_get(node, str, key) |
| path = Path(path_str) |
| project_dir_path = Path(project_dir) |
| |
| provenance = node_get_provenance(node, key=key) |
| |
| if (project_dir_path / path).is_symlink(): |
| raise LoadError(LoadErrorReason.PROJ_PATH_INVALID_KIND, |
| "{}: Specified path '{}' must not point to " |
| "symbolic links " |
| .format(provenance, path_str)) |
| |
| if path.parts and path.parts[0] == '..': |
| raise LoadError(LoadErrorReason.PROJ_PATH_INVALID, |
| "{}: Specified path '{}' first component must " |
| "not be '..'" |
| .format(provenance, path_str)) |
| |
| try: |
| if sys.version_info[0] == 3 and sys.version_info[1] < 6: |
| full_resolved_path = (project_dir_path / path).resolve() |
| else: |
| full_resolved_path = (project_dir_path / path).resolve(strict=True) |
| except FileNotFoundError: |
| raise LoadError(LoadErrorReason.MISSING_FILE, |
| "{}: Specified path '{}' does not exist" |
| .format(provenance, path_str)) |
| |
| is_inside = project_dir_path.resolve() in full_resolved_path.parents or ( |
| full_resolved_path == project_dir_path) |
| |
| if path.is_absolute() or not is_inside: |
| raise LoadError(LoadErrorReason.PROJ_PATH_INVALID, |
| "{}: Specified path '{}' must not lead outside of the " |
| "project directory" |
| .format(provenance, path_str)) |
| |
| if full_resolved_path.is_socket() or ( |
| full_resolved_path.is_fifo() or |
| full_resolved_path.is_block_device()): |
| raise LoadError(LoadErrorReason.PROJ_PATH_INVALID_KIND, |
| "{}: Specified path '{}' points to an unsupported " |
| "file kind" |
| .format(provenance, path_str)) |
| |
| if check_is_file and not full_resolved_path.is_file(): |
| raise LoadError(LoadErrorReason.PROJ_PATH_INVALID_KIND, |
| "{}: Specified path '{}' is not a regular file" |
| .format(provenance, path_str)) |
| |
| if check_is_dir and not full_resolved_path.is_dir(): |
| raise LoadError(LoadErrorReason.PROJ_PATH_INVALID_KIND, |
| "{}: Specified path '{}' is not a directory" |
| .format(provenance, path_str)) |
| |
| return path_str |
| |
| |
| # node_items() |
| # |
| # A convenience generator for iterating over loaded key/value |
| # tuples in a dictionary loaded from project YAML. |
| # |
| # Args: |
| # node (dict): The dictionary node |
| # |
| # Yields: |
| # (str): The key name |
| # (anything): The value for the key |
| # |
| def node_items(node): |
| for key, value in node.items(): |
| if key == PROVENANCE_KEY: |
| continue |
| yield (key, value) |
| |
| |
| # Gives a node a dummy provenance, in case of compositing dictionaries |
| # where the target is an empty {} |
| def ensure_provenance(node): |
| provenance = node.get(PROVENANCE_KEY) |
| if not provenance: |
| provenance = DictProvenance(ProvenanceFile('', '', None), node, node) |
| node[PROVENANCE_KEY] = provenance |
| |
| return provenance |
| |
| |
| # is_ruamel_str(): |
| # |
| # Args: |
| # value: A value loaded from ruamel |
| # |
| # This returns if the value is "stringish", since ruamel |
| # has some complex types to represent strings, this is needed |
| # to avoid compositing exceptions in order to allow various |
| # string types to be interchangable and acceptable |
| # |
| def is_ruamel_str(value): |
| |
| if isinstance(value, str): |
| return True |
| elif isinstance(value, yaml.scalarstring.ScalarString): |
| return True |
| |
| return False |
| |
| |
| # is_composite_list |
| # |
| # Checks if the given node is a Mapping with array composition |
| # directives. |
| # |
| # Args: |
| # node (value): Any node |
| # |
| # Returns: |
| # (bool): True if node was a Mapping containing only |
| # list composition directives |
| # |
| # Raises: |
| # (LoadError): If node was a mapping and contained a mix of |
| # list composition directives and other keys |
| # |
| def is_composite_list(node): |
| |
| if isinstance(node, collections.abc.Mapping): |
| has_directives = False |
| has_keys = False |
| |
| for key, _ in node_items(node): |
| if key in ['(>)', '(<)', '(=)']: # pylint: disable=simplifiable-if-statement |
| has_directives = True |
| else: |
| has_keys = True |
| |
| if has_keys and has_directives: |
| provenance = node_get_provenance(node) |
| raise LoadError(LoadErrorReason.INVALID_DATA, |
| "{}: Dictionary contains array composition directives and arbitrary keys" |
| .format(provenance)) |
| return has_directives |
| |
| return False |
| |
| |
| # composite_list_prepend |
| # |
| # Internal helper for list composition |
| # |
| # Args: |
| # target_node (dict): A simple dictionary |
| # target_key (dict): The key indicating a literal array to prepend to |
| # source_node (dict): Another simple dictionary |
| # source_key (str): The key indicating an array to prepend to the target |
| # |
| # Returns: |
| # (bool): True if a source list was found and compositing occurred |
| # |
| def composite_list_prepend(target_node, target_key, source_node, source_key): |
| |
| source_list = node_get(source_node, list, source_key, default_value=[]) |
| if not source_list: |
| return False |
| |
| target_provenance = node_get_provenance(target_node) |
| source_provenance = node_get_provenance(source_node) |
| |
| if target_node.get(target_key) is None: |
| target_node[target_key] = [] |
| |
| source_list = list_chain_copy(source_list) |
| target_list = target_node[target_key] |
| |
| for element in reversed(source_list): |
| target_list.insert(0, element) |
| |
| if not target_provenance.members.get(target_key): |
| target_provenance.members[target_key] = source_provenance.members[source_key].clone() |
| else: |
| for p in reversed(source_provenance.members[source_key].elements): |
| target_provenance.members[target_key].elements.insert(0, p.clone()) |
| |
| return True |
| |
| |
| # composite_list_append |
| # |
| # Internal helper for list composition |
| # |
| # Args: |
| # target_node (dict): A simple dictionary |
| # target_key (dict): The key indicating a literal array to append to |
| # source_node (dict): Another simple dictionary |
| # source_key (str): The key indicating an array to append to the target |
| # |
| # Returns: |
| # (bool): True if a source list was found and compositing occurred |
| # |
| def composite_list_append(target_node, target_key, source_node, source_key): |
| |
| source_list = node_get(source_node, list, source_key, default_value=[]) |
| if not source_list: |
| return False |
| |
| target_provenance = node_get_provenance(target_node) |
| source_provenance = node_get_provenance(source_node) |
| |
| if target_node.get(target_key) is None: |
| target_node[target_key] = [] |
| |
| source_list = list_chain_copy(source_list) |
| target_list = target_node[target_key] |
| |
| target_list.extend(source_list) |
| |
| if not target_provenance.members.get(target_key): |
| target_provenance.members[target_key] = source_provenance.members[source_key].clone() |
| else: |
| target_provenance.members[target_key].elements.extend([ |
| p.clone() for p in source_provenance.members[source_key].elements |
| ]) |
| |
| return True |
| |
| |
| # composite_list_overwrite |
| # |
| # Internal helper for list composition |
| # |
| # Args: |
| # target_node (dict): A simple dictionary |
| # target_key (dict): The key indicating a literal array to overwrite |
| # source_node (dict): Another simple dictionary |
| # source_key (str): The key indicating an array to overwrite the target with |
| # |
| # Returns: |
| # (bool): True if a source list was found and compositing occurred |
| # |
| def composite_list_overwrite(target_node, target_key, source_node, source_key): |
| |
| # We need to handle the legitimate case of overwriting a list with an empty |
| # list, hence the slightly odd default_value of [None] rather than []. |
| source_list = node_get(source_node, list, source_key, default_value=[None]) |
| if source_list == [None]: |
| return False |
| |
| target_provenance = node_get_provenance(target_node) |
| source_provenance = node_get_provenance(source_node) |
| |
| target_node[target_key] = list_chain_copy(source_list) |
| target_provenance.members[target_key] = source_provenance.members[source_key].clone() |
| |
| return True |
| |
| |
| # composite_list(): |
| # |
| # Composite the source value onto the target value, if either |
| # sides are lists, or dictionaries containing list compositing directives |
| # |
| # Args: |
| # target_node (dict): A simple dictionary |
| # source_node (dict): Another simple dictionary |
| # key (str): The key to compose on |
| # |
| # Returns: |
| # (bool): True if both sides were logical lists |
| # |
| # Raises: |
| # (LoadError): If one side was a logical list and the other was not |
| # |
| def composite_list(target_node, source_node, key): |
| target_value = target_node.get(key) |
| source_value = source_node[key] |
| |
| target_key_provenance = node_get_provenance(target_node, key) |
| source_key_provenance = node_get_provenance(source_node, key) |
| |
| # Whenever a literal list is encountered in the source, it |
| # overwrites the target values and provenance completely. |
| # |
| if isinstance(source_value, list): |
| |
| source_provenance = node_get_provenance(source_node) |
| target_provenance = node_get_provenance(target_node) |
| |
| # Assert target type |
| if not (target_value is None or |
| isinstance(target_value, list) or |
| is_composite_list(target_value)): |
| raise LoadError(LoadErrorReason.INVALID_DATA, |
| "{}: List cannot overwrite value at: {}" |
| .format(source_key_provenance, target_key_provenance)) |
| |
| composite_list_overwrite(target_node, key, source_node, key) |
| return True |
| |
| # When a composite list is encountered in the source, then |
| # multiple outcomes can occur... |
| # |
| elif is_composite_list(source_value): |
| |
| # If there is nothing there, then the composite list |
| # is copied in it's entirety as is, and preserved |
| # for later composition |
| # |
| if target_value is None: |
| source_provenance = node_get_provenance(source_node) |
| target_provenance = node_get_provenance(target_node) |
| |
| target_node[key] = node_chain_copy(source_value) |
| target_provenance.members[key] = source_provenance.members[key].clone() |
| |
| # If the target is a literal list, then composition |
| # occurs directly onto that target, leaving the target |
| # as a literal list to overwrite anything in later composition |
| # |
| elif isinstance(target_value, list): |
| composite_list_overwrite(target_node, key, source_value, '(=)') |
| composite_list_prepend(target_node, key, source_value, '(<)') |
| composite_list_append(target_node, key, source_value, '(>)') |
| |
| # If the target is a composite list, then composition |
| # occurs in the target composite list, and the composite |
| # target list is preserved in dictionary form for further |
| # composition. |
| # |
| elif is_composite_list(target_value): |
| |
| if composite_list_overwrite(target_value, '(=)', source_value, '(=)'): |
| |
| # When overwriting a target with composition directives, remove any |
| # existing prepend/append directives in the target before adding our own |
| target_provenance = node_get_provenance(target_value) |
| |
| for directive in ['(<)', '(>)']: |
| try: |
| del target_value[directive] |
| del target_provenance.members[directive] |
| except KeyError: |
| # Ignore errors from deletion of non-existing keys |
| pass |
| |
| # Prepend to the target prepend array, and append to the append array |
| composite_list_prepend(target_value, '(<)', source_value, '(<)') |
| composite_list_append(target_value, '(>)', source_value, '(>)') |
| |
| else: |
| raise LoadError(LoadErrorReason.INVALID_DATA, |
| "{}: List cannot overwrite value at: {}" |
| .format(source_key_provenance, target_key_provenance)) |
| |
| # We handled list composition in some way |
| return True |
| |
| # Source value was not a logical list |
| return False |
| |
| |
| # composite_dict(): |
| # |
| # Composites values in target with values from source |
| # |
| # Args: |
| # target (dict): A simple dictionary |
| # source (dict): Another simple dictionary |
| # |
| # Raises: CompositeError |
| # |
| # Unlike the dictionary update() method, nested values in source |
| # will not obsolete entire subdictionaries in target, instead both |
| # dictionaries will be recursed and a composition of both will result |
| # |
| # This is useful for overriding configuration files and element |
| # configurations. |
| # |
| def composite_dict(target, source, path=None): |
| target_provenance = ensure_provenance(target) |
| source_provenance = ensure_provenance(source) |
| |
| for key, source_value in node_items(source): |
| |
| # Track the full path of keys, only for raising CompositeError |
| if path: |
| thispath = path + '.' + key |
| else: |
| thispath = key |
| |
| # Handle list composition separately |
| if composite_list(target, source, key): |
| continue |
| |
| target_value = target.get(key) |
| |
| if isinstance(source_value, collections.abc.Mapping): |
| |
| # Handle creating new dicts on target side |
| if target_value is None: |
| target_value = {} |
| target[key] = target_value |
| |
| # Give the new dict provenance |
| value_provenance = source_value.get(PROVENANCE_KEY) |
| if value_provenance: |
| target_value[PROVENANCE_KEY] = value_provenance.clone() |
| |
| # Add a new provenance member element to the containing dict |
| target_provenance.members[key] = source_provenance.members[key] |
| |
| if not isinstance(target_value, collections.abc.Mapping): |
| raise CompositeTypeError(thispath, type(target_value), type(source_value)) |
| |
| # Recurse into matching dictionary |
| composite_dict(target_value, source_value, path=thispath) |
| |
| else: |
| |
| if target_value is not None: |
| |
| # Exception here: depending on how strings were declared ruamel may |
| # use a different type, but for our purposes, any stringish type will do. |
| if not (is_ruamel_str(source_value) and is_ruamel_str(target_value)) \ |
| and not isinstance(source_value, type(target_value)): |
| raise CompositeTypeError(thispath, type(target_value), type(source_value)) |
| |
| # Overwrite simple values, lists and mappings have already been handled |
| target_provenance.members[key] = source_provenance.members[key].clone() |
| target[key] = source_value |
| |
| |
| # Like composite_dict(), but raises an all purpose LoadError for convenience |
| # |
| def composite(target, source): |
| assert hasattr(source, 'get') |
| |
| source_provenance = node_get_provenance(source) |
| try: |
| composite_dict(target, source) |
| except CompositeTypeError as e: |
| error_prefix = "" |
| if source_provenance: |
| error_prefix = "{}: ".format(source_provenance) |
| raise LoadError(LoadErrorReason.ILLEGAL_COMPOSITE, |
| "{}Expected '{}' type for configuration '{}', instead received '{}'" |
| .format(error_prefix, |
| e.expected_type.__name__, |
| e.path, |
| e.actual_type.__name__)) from e |
| |
| |
| # SanitizedDict is an OrderedDict that is dumped as unordered mapping. |
| # This provides deterministic output for unordered mappings. |
| # |
| class SanitizedDict(collections.OrderedDict): |
| pass |
| |
| |
| RoundTripRepresenter.add_representer(SanitizedDict, |
| SafeRepresenter.represent_dict) |
| |
| |
| # node_sanitize() |
| # |
| # Returnes an alphabetically ordered recursive copy |
| # of the source node with internal provenance information stripped. |
| # |
| # Only dicts are ordered, list elements are left in order. |
| # |
| def node_sanitize(node): |
| |
| if isinstance(node, collections.abc.Mapping): |
| |
| result = SanitizedDict() |
| |
| key_list = [key for key, _ in node_items(node)] |
| for key in sorted(key_list): |
| result[key] = node_sanitize(node[key]) |
| |
| return result |
| |
| elif isinstance(node, list): |
| return [node_sanitize(elt) for elt in node] |
| |
| return node |
| |
| |
| # node_validate() |
| # |
| # Validate the node so as to ensure the user has not specified |
| # any keys which are unrecognized by buildstream (usually this |
| # means a typo which would otherwise not trigger an error). |
| # |
| # Args: |
| # node (dict): A dictionary loaded from YAML |
| # valid_keys (list): A list of valid keys for the specified node |
| # |
| # Raises: |
| # LoadError: In the case that the specified node contained |
| # one or more invalid keys |
| # |
| def node_validate(node, valid_keys): |
| |
| # Probably the fastest way to do this: https://stackoverflow.com/a/23062482 |
| valid_keys = set(valid_keys) |
| valid_keys.add(PROVENANCE_KEY) |
| invalid = next((key for key in node if key not in valid_keys), None) |
| |
| if invalid: |
| provenance = node_get_provenance(node, key=invalid) |
| raise LoadError(LoadErrorReason.INVALID_DATA, |
| "{}: Unexpected key: {}".format(provenance, invalid)) |
| |
| |
| # ChainMap |
| # |
| # This is a derivative of collections.ChainMap(), but supports |
| # explicit deletions of keys. |
| # |
| # The purpose of this is to create a virtual copy-on-write |
| # copy of a dictionary, so that mutating it in any way does |
| # not effect the underlying dictionaries. |
| # |
| # collections.ChainMap covers this already mostly, but fails |
| # to record internal state so as to hide keys which have been |
| # explicitly deleted. |
| # |
| class ChainMap(collections.ChainMap): |
| |
| def __init__(self, *maps): |
| super().__init__(*maps) |
| self.__deletions = set() |
| |
| def __getitem__(self, key): |
| |
| # Honor deletion state of 'key' |
| if key in self.__deletions: |
| return self.__missing__(key) |
| |
| return super().__getitem__(key) |
| |
| def __len__(self): |
| return len(set().union(*self.maps) - self.__deletions) |
| |
| def __iter__(self): |
| return iter(set().union(*self.maps) - self.__deletions) |
| |
| def __contains__(self, key): |
| if key in self.__deletions: |
| return False |
| return any(key in m for m in self.maps) |
| |
| def __bool__(self): |
| # Attempt to preserve 'any' optimization |
| any_keys = any(self.maps) |
| |
| # Something existed, try again with deletions subtracted |
| if any_keys: |
| return any(set().union(*self.maps) - self.__deletions) |
| |
| return False |
| |
| def __setitem__(self, key, value): |
| self.__deletions.discard(key) |
| super().__setitem__(key, value) |
| |
| def __delitem__(self, key): |
| if key in self.__deletions: |
| raise KeyError('Key was already deleted from this mapping: {!r}'.format(key)) |
| |
| # Ignore KeyError if it's not in the first map, just save the deletion state |
| try: |
| super().__delitem__(key) |
| except KeyError: |
| pass |
| |
| # Store deleted state |
| self.__deletions.add(key) |
| |
| def popitem(self): |
| poppable = set().union(*self.maps) - self.__deletions |
| for key in poppable: |
| return self.pop(key) |
| |
| raise KeyError('No keys found.') |
| |
| __marker = object() |
| |
| def pop(self, key, default=__marker): |
| # Reimplement MutableMapping's behavior here |
| try: |
| value = self[key] |
| except KeyError: |
| if default is self.__marker: |
| raise |
| return default |
| else: |
| del self[key] |
| return value |
| |
| def clear(self): |
| clearable = set().union(*self.maps) - self.__deletions |
| for key in clearable: |
| del self[key] |
| |
| |
| def node_chain_copy(source): |
| copy = ChainMap({}, source) |
| for key, value in source.items(): |
| if isinstance(value, collections.abc.Mapping): |
| copy[key] = node_chain_copy(value) |
| elif isinstance(value, list): |
| copy[key] = list_chain_copy(value) |
| elif isinstance(value, Provenance): |
| copy[key] = value.clone() |
| |
| return copy |
| |
| |
| def list_chain_copy(source): |
| copy = [] |
| for item in source: |
| if isinstance(item, collections.abc.Mapping): |
| copy.append(node_chain_copy(item)) |
| elif isinstance(item, list): |
| copy.append(list_chain_copy(item)) |
| elif isinstance(item, Provenance): |
| copy.append(item.clone()) |
| else: |
| copy.append(item) |
| |
| return copy |
| |
| |
| def node_copy(source): |
| copy = {} |
| for key, value in source.items(): |
| if isinstance(value, collections.abc.Mapping): |
| copy[key] = node_copy(value) |
| elif isinstance(value, list): |
| copy[key] = list_copy(value) |
| elif isinstance(value, Provenance): |
| copy[key] = value.clone() |
| else: |
| copy[key] = value |
| |
| ensure_provenance(copy) |
| |
| return copy |
| |
| |
| def list_copy(source): |
| copy = [] |
| for item in source: |
| if isinstance(item, collections.abc.Mapping): |
| copy.append(node_copy(item)) |
| elif isinstance(item, list): |
| copy.append(list_copy(item)) |
| elif isinstance(item, Provenance): |
| copy.append(item.clone()) |
| else: |
| copy.append(item) |
| |
| return copy |
| |
| |
| # node_final_assertions() |
| # |
| # This must be called on a fully loaded and composited node, |
| # after all composition has completed. |
| # |
| # Args: |
| # node (Mapping): The final composited node |
| # |
| # Raises: |
| # (LoadError): If any assertions fail |
| # |
| def node_final_assertions(node): |
| for key, value in node_items(node): |
| |
| # Assert that list composition directives dont remain, this |
| # indicates that the user intended to override a list which |
| # never existed in the underlying data |
| # |
| if key in ['(>)', '(<)', '(=)']: |
| provenance = node_get_provenance(node, key) |
| raise LoadError(LoadErrorReason.TRAILING_LIST_DIRECTIVE, |
| "{}: Attempt to override non-existing list".format(provenance)) |
| |
| if isinstance(value, collections.abc.Mapping): |
| node_final_assertions(value) |
| elif isinstance(value, list): |
| list_final_assertions(value) |
| |
| |
| def list_final_assertions(values): |
| for value in values: |
| if isinstance(value, collections.abc.Mapping): |
| node_final_assertions(value) |
| elif isinstance(value, list): |
| list_final_assertions(value) |
| |
| |
| # assert_symbol_name() |
| # |
| # A helper function to check if a loaded string is a valid symbol |
| # name and to raise a consistent LoadError if not. For strings which |
| # are required to be symbols. |
| # |
| # Args: |
| # provenance (Provenance): The provenance of the loaded symbol, or None |
| # symbol_name (str): The loaded symbol name |
| # purpose (str): The purpose of the string, for an error message |
| # allow_dashes (bool): Whether dashes are allowed for this symbol |
| # |
| # Raises: |
| # LoadError: If the symbol_name is invalid |
| # |
| # Note that dashes are generally preferred for variable names and |
| # usage in YAML, but things such as option names which will be |
| # evaluated with jinja2 cannot use dashes. |
| # |
| def assert_symbol_name(provenance, symbol_name, purpose, *, allow_dashes=True): |
| valid_chars = string.digits + string.ascii_letters + '_' |
| if allow_dashes: |
| valid_chars += '-' |
| |
| valid = True |
| if not symbol_name: |
| valid = False |
| elif any(x not in valid_chars for x in symbol_name): |
| valid = False |
| elif symbol_name[0] in string.digits: |
| valid = False |
| |
| if not valid: |
| detail = "Symbol names must contain only alphanumeric characters, " + \ |
| "may not start with a digit, and may contain underscores" |
| if allow_dashes: |
| detail += " or dashes" |
| |
| message = "Invalid symbol name for {}: '{}'".format(purpose, symbol_name) |
| if provenance is not None: |
| message = "{}: {}".format(provenance, message) |
| |
| raise LoadError(LoadErrorReason.INVALID_SYMBOL_NAME, |
| message, detail=detail) |
