ssg.yaml._open_yaml() - Code Metrics - Inspection of "Merge pull request #4429 from cipherboy/oh-no-docu..." - ComplianceAsCode/content - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 2e0e1e...e323ea )

by Marek

created 2019-06-20 14:58 UTC

ssg.yaml._open_yaml() C

↳ Parent: ssg.yaml

Complexity

Conditions

Size

Total Lines	36
Code Lines	25

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	5
CRAP Score	49.1945

Importance

Changes

Metric	Value
cc	9
eloc	25
nop	3
dl	0
loc	36
ccs	5
cts	24
cp	0.2083
crap	49.1945
rs	6.6666
c	0
b	0
f	0

from __future__ import absolute_import
from __future__ import print_function

import codecs
import yaml
import sys
import re

from collections import OrderedDict

from .jinja import load_macros, process_file
from .constants import (PKG_MANAGER_TO_SYSTEM,
                        PKG_MANAGER_TO_CONFIG_FILE)
from .constants import DEFAULT_UID_MIN

try:
    from yaml import CSafeLoader as yaml_SafeLoader
except ImportError:
    from yaml import SafeLoader as yaml_SafeLoader


def _bool_constructor(self, node):
    return self.construct_scalar(node)


# Don't follow python bool case
yaml_SafeLoader.add_constructor(u'tag:yaml.org,2002:bool', _bool_constructor)


class DocumentationNotComplete(Exception):
    pass


def _save_rename(result, stem, prefix):
    result["{0}_{1}".format(prefix, stem)] = stem


def _open_yaml(stream, original_file=None, substitutions_dict={}):
    """
    Open given file-like object and parse it as YAML.

    Optionally, pass the path to the original_file for better error handling
    when the file contents are passed.

    Return None if it contains "documentation_complete" key set to "false".
    """
    try:
        yaml_contents = yaml.load(stream, Loader=yaml_SafeLoader)

        if yaml_contents.pop("documentation_complete", "true") == "false" and \
                substitutions_dict.get("cmake_build_type") != "Debug":
            raise DocumentationNotComplete("documentation not complete and not a debug build")

        return yaml_contents
    except DocumentationNotComplete as e:
        raise e
    except Exception as e:
        count = 0
        _file = original_file
        if not _file:
            _file = stream
        with open(_file, "r") as e_file:
            lines = e_file.readlines()
            for line in lines:
                count = count + 1
                if re.match(r"^\s*\t+\s*", line):
                    print("Exception while handling file: %s" % _file, file=sys.stderr)
                    print("TabIndentationError: Line %s contains tabs instead of spaces:" % (count), file=sys.stderr)
                    print("%s\n\n" % repr(line.strip("\n")), file=sys.stderr)
                    sys.exit(1)

        print("Exception while handling file: %s" % _file, file=sys.stderr)
        raise e


def _get_implied_properties(existing_properties):
    result = dict()
    if "pkg_manager" in existing_properties:
        pkg_manager = existing_properties["pkg_manager"]
        if "pkg_system" not in existing_properties:
            result["pkg_system"] = PKG_MANAGER_TO_SYSTEM[pkg_manager]
        if "pkg_manager_config_file" not in existing_properties:
            if pkg_manager in PKG_MANAGER_TO_CONFIG_FILE:
                result["pkg_manager_config_file"] = PKG_MANAGER_TO_CONFIG_FILE[pkg_manager]

    if "uid_min" not in existing_properties:
        result["uid_min"] = DEFAULT_UID_MIN

    if "auid" not in existing_properties:
        result["auid"] = existing_properties.get("uid_min", DEFAULT_UID_MIN)

    return result


def open_and_expand(yaml_file, substitutions_dict=None):
    """
    Process the file as a template, using substitutions_dict to perform
    expansion. Then, process the expansion result as a YAML content.

    See also: _open_yaml
    """
    if substitutions_dict is None:
        substitutions_dict = dict()

    expanded_template = process_file(yaml_file, substitutions_dict)
    yaml_contents = _open_yaml(expanded_template, yaml_file, substitutions_dict)
    return yaml_contents


def open_and_macro_expand(yaml_file, substitutions_dict=None):
    """
    Do the same as open_and_expand, but load definitions of macros
    so they can be expanded in the template.
    """
    substitutions_dict = load_macros(substitutions_dict)
    return open_and_expand(yaml_file, substitutions_dict)


def open_raw(yaml_file):
    """
    Open given file-like object and parse it as YAML
    without performing any kind of template processing

    See also: _open_yaml
    """
    with codecs.open(yaml_file, "r", "utf8") as stream:
        yaml_contents = _open_yaml(stream, original_file=yaml_file)
    return yaml_contents


def open_environment(build_config_yaml, product_yaml):
    contents = open_raw(build_config_yaml)
    contents.update(open_raw(product_yaml))
    contents.update(_get_implied_properties(contents))
    return contents


def ordered_load(stream, Loader=yaml.Loader, object_pairs_hook=OrderedDict):
    """
    Drop-in replacement for yaml.load(), but preserves order of dictionaries
    """
    class OrderedLoader(Loader):
        pass

    def construct_mapping(loader, node):
        loader.flatten_mapping(node)
        return object_pairs_hook(loader.construct_pairs(node))
    OrderedLoader.add_constructor(
        yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG,
        construct_mapping)
    return yaml.load(stream, OrderedLoader)


def ordered_dump(data, stream=None, Dumper=yaml.Dumper, **kwds):
    """
    Drop-in replacement for yaml.dump(), but preserves order of dictionaries
    """
    class OrderedDumper(Dumper):
        # fix tag indentations
        def increase_indent(self, flow=False, indentless=False):
            return super(OrderedDumper, self).increase_indent(flow, False)

    def _dict_representer(dumper, data):
        return dumper.represent_mapping(
            yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG,
            data.items())

    OrderedDumper.add_representer(OrderedDict, _dict_representer)

    # Fix formatting by adding a space in between tasks
    unformatted_yaml = yaml.dump(data, None, OrderedDumper, **kwds)
    formatted_yaml = re.sub(r"[\n]+([\s]*)- name", r"\n\n\1- name", unformatted_yaml)

    if stream is not None:
        return stream.write(formatted_yaml)
    else:
        return formatted_yaml
    return yaml.dump(data, stream, OrderedDumper, **kwds)


def _strings_to_list(one_or_more_strings):
    """
    Output a list, that either contains one string, or a list of strings.
    In Python, strings can be cast to lists without error, but with unexpected result.
    """
    if isinstance(one_or_more_strings, str):
        return [one_or_more_strings]
    else:
        return list(one_or_more_strings)


def update_yaml_list_or_string(current_contents, new_contents):

    result = []
    if current_contents:
        result += _strings_to_list(current_contents)
    if new_contents:
        result += _strings_to_list(new_contents)
    if not result:
        result = ""
    if len(result) == 1:
        result = result[0]
    return result


1	2		from __future__ import absolute_import
2	2		from __future__ import print_function
3
4	2		import codecs
5	2		import yaml
6	2		import sys
7	2		import re
8
9	2		from collections import OrderedDict
10
11	2		from .jinja import load_macros, process_file
12	2		from .constants import (PKG_MANAGER_TO_SYSTEM,
13			PKG_MANAGER_TO_CONFIG_FILE)
14	2		from .constants import DEFAULT_UID_MIN
15
16	2		try:
17	2		from yaml import CSafeLoader as yaml_SafeLoader
18	2		except ImportError:
19	2		from yaml import SafeLoader as yaml_SafeLoader
20
21
22	2		def _bool_constructor(self, node):
23	2		return self.construct_scalar(node)
24
25
26			# Don't follow python bool case
27	2		yaml_SafeLoader.add_constructor(u'tag:yaml.org,2002:bool', _bool_constructor)
28
29
30	2		class DocumentationNotComplete(Exception):
31	2		pass
32
33
34	2		def _save_rename(result, stem, prefix):
35			result["{0}_{1}".format(prefix, stem)] = stem
36
37
38	2		def _open_yaml(stream, original_file=None, substitutions_dict={}):
39			"""
40			Open given file-like object and parse it as YAML.
41
42			Optionally, pass the path to the original_file for better error handling
43			when the file contents are passed.
44
45			Return None if it contains "documentation_complete" key set to "false".
46			"""
47	2		try:
48	2		yaml_contents = yaml.load(stream, Loader=yaml_SafeLoader)
49
50	2		if yaml_contents.pop("documentation_complete", "true") == "false" and \
51			substitutions_dict.get("cmake_build_type") != "Debug":
52			raise DocumentationNotComplete("documentation not complete and not a debug build")
53
54	2		return yaml_contents
55			except DocumentationNotComplete as e:
56			raise e
57			except Exception as e:
58			count = 0
59			_file = original_file
60			if not _file:
61			_file = stream
62			with open(_file, "r") as e_file:
63			lines = e_file.readlines()
64			for line in lines:
65			count = count + 1
66			if re.match(r"^\s\t+\s", line):
67			print("Exception while handling file: %s" % _file, file=sys.stderr)
68			print("TabIndentationError: Line %s contains tabs instead of spaces:" % (count), file=sys.stderr)
69			print("%s\n\n" % repr(line.strip("\n")), file=sys.stderr)
70			sys.exit(1)
71
72			print("Exception while handling file: %s" % _file, file=sys.stderr)
73			raise e
74
75
76	2		def _get_implied_properties(existing_properties):
77			result = dict()
78			if "pkg_manager" in existing_properties:
79			pkg_manager = existing_properties["pkg_manager"]
80			if "pkg_system" not in existing_properties:
81			result["pkg_system"] = PKG_MANAGER_TO_SYSTEM[pkg_manager]
82			if "pkg_manager_config_file" not in existing_properties:
83			if pkg_manager in PKG_MANAGER_TO_CONFIG_FILE:
84			result["pkg_manager_config_file"] = PKG_MANAGER_TO_CONFIG_FILE[pkg_manager]
85
86			if "uid_min" not in existing_properties:
87			result["uid_min"] = DEFAULT_UID_MIN
88
89			if "auid" not in existing_properties:
90			result["auid"] = existing_properties.get("uid_min", DEFAULT_UID_MIN)
91
92			return result
93
94
95	2		def open_and_expand(yaml_file, substitutions_dict=None):
96			"""
97			Process the file as a template, using substitutions_dict to perform
98			expansion. Then, process the expansion result as a YAML content.
99
100			See also: _open_yaml
101			"""
102	2		if substitutions_dict is None:
103	2		substitutions_dict = dict()
104
105	2		expanded_template = process_file(yaml_file, substitutions_dict)
106	2		yaml_contents = _open_yaml(expanded_template, yaml_file, substitutions_dict)
107	2		return yaml_contents
108
109
110	2		def open_and_macro_expand(yaml_file, substitutions_dict=None):
111			"""
112			Do the same as open_and_expand, but load definitions of macros
113			so they can be expanded in the template.
114			"""
115	2		substitutions_dict = load_macros(substitutions_dict)
116	2		return open_and_expand(yaml_file, substitutions_dict)
117
118
119	2		def open_raw(yaml_file):
120			"""
121			Open given file-like object and parse it as YAML
122			without performing any kind of template processing
123
124			See also: _open_yaml
125			"""
126	2		with codecs.open(yaml_file, "r", "utf8") as stream:
127	2		yaml_contents = _open_yaml(stream, original_file=yaml_file)
128	2		return yaml_contents
129
130
131	2		def open_environment(build_config_yaml, product_yaml):
132			contents = open_raw(build_config_yaml)
133			contents.update(open_raw(product_yaml))
134			contents.update(_get_implied_properties(contents))
135			return contents
136
137
138	2		def ordered_load(stream, Loader=yaml.Loader, object_pairs_hook=OrderedDict):
139			"""
140			Drop-in replacement for yaml.load(), but preserves order of dictionaries
141			"""
142	2		class OrderedLoader(Loader):
143	2		pass
144
145	2		def construct_mapping(loader, node):
146	2		loader.flatten_mapping(node)
147	2		return object_pairs_hook(loader.construct_pairs(node))
148	2		OrderedLoader.add_constructor(
149			yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG,
150			construct_mapping)
151	2		return yaml.load(stream, OrderedLoader)
152
153
154	2		def ordered_dump(data, stream=None, Dumper=yaml.Dumper, **kwds):
155			"""
156			Drop-in replacement for yaml.dump(), but preserves order of dictionaries
157			"""
158	2		class OrderedDumper(Dumper):
159			# fix tag indentations
160	2		def increase_indent(self, flow=False, indentless=False):
161	2		return super(OrderedDumper, self).increase_indent(flow, False)
162
163	2		def _dict_representer(dumper, data):
164	2		return dumper.represent_mapping(
165			yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG,
166			data.items())
167
168	2		OrderedDumper.add_representer(OrderedDict, _dict_representer)
169
170			# Fix formatting by adding a space in between tasks
171	2		unformatted_yaml = yaml.dump(data, None, OrderedDumper, **kwds)
172	2		formatted_yaml = re.sub(r"[\n]+([\s]*)- name", r"\n\n\1- name", unformatted_yaml)
173
174	2		if stream is not None:
175	2		return stream.write(formatted_yaml)
176			else:
177	2		return formatted_yaml
178			return yaml.dump(data, stream, OrderedDumper, **kwds)
179
180
181	2		def _strings_to_list(one_or_more_strings):
182			"""
183			Output a list, that either contains one string, or a list of strings.
184			In Python, strings can be cast to lists without error, but with unexpected result.
185			"""
186	2		if isinstance(one_or_more_strings, str):
187	2		return [one_or_more_strings]
188			else:
189	2		return list(one_or_more_strings)
190
191
192	2	View Code Duplication	def update_yaml_list_or_string(current_contents, new_contents):
			0 ignored issues – show Duplication introduced 2019-03-26 10:32 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
193	2		result = []
194	2		if current_contents:
195	2		result += _strings_to_list(current_contents)
196	2		if new_contents:
197	2		result += _strings_to_list(new_contents)
198	2		if not result:
199	2		result = ""
200	2		if len(result) == 1:
201	2		result = result[0]
202			return result
203

ComplianceAsCode / content

Push — master ( 2e0e1e...e323ea )

ssg.yaml._open_yaml() C

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like