1644037831

Build Type

push

github

Committed by W. Augusto Andreoli

Commit Message

style: add GitHub Workflow for Python

Run Details

564 of 613 branches covered (92.01%)

Branch coverage included in aggregate %.

32 of 34 new or added lines in 1 file covered. (94.12%)

18 existing lines in 1 file now uncovered.

2058 of 2112 relevant lines covered (97.44%)

4.85 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

87.95

/src/nitpick/documents.py

"""Configuration file formats."""
import abc
import json
from collections import OrderedDict
from pathlib import Path
from typing import Any, Callable, List, Optional, Type, Union

import dictdiffer
import toml
import tomlkit
from autorepr import autorepr
from loguru import logger
from ruamel.yaml import YAML, RoundTripRepresenter, StringIO
from ruamel.yaml.comments import CommentedMap, CommentedSeq
from sortedcontainers import SortedDict

from nitpick.generic import flatten, unflatten
from nitpick.typedefs import JsonDict, PathOrStr, YamlObject, YamlValue

DictOrYamlObject = Union[JsonDict, YamlObject, "BaseDoc"]
DICT_CLASSES = (dict, SortedDict, OrderedDict)
LIST_CLASSES = (list, tuple)


def compare_lists_with_dictdiffer(actual: List[Any], expected: List[Any]) -> List:
    """Compare two lists using dictdiffer."""
    additions_and_changes = [change for change in dictdiffer.diff(actual, expected) if change[0] != "remove"]
    if additions_and_changes:
        try:
            changed_dict = dictdiffer.patch(additions_and_changes, {})
            return list(changed_dict.values())
        except KeyError:
            return expected
    return []


class Comparison:
    """A comparison between two dictionaries, computing missing items and differences."""

    def __init__(
        self,
        actual: DictOrYamlObject,
        expected: DictOrYamlObject,
        format_class: Type["BaseDoc"],
    ) -> None:
        self.flat_actual = self._normalize_value(actual)
        self.flat_expected = self._normalize_value(expected)

        self.format_class = format_class

        self.missing: Optional[BaseDoc] = None
        self.missing_dict: Union[JsonDict, YamlObject] = {}

        self.diff: Optional[BaseDoc] = None
        self.diff_dict: Union[JsonDict, YamlObject] = {}

    @property
    def has_changes(self) -> bool:
        """Return True is there is a difference or something missing."""
        return bool(self.missing or self.diff)

    @staticmethod
    def _normalize_value(value: DictOrYamlObject) -> JsonDict:
        if isinstance(value, BaseDoc):
            dict_value: JsonDict = value.as_object
        else:
            dict_value = value
        return flatten(dict_value)

    def set_missing(self, missing_dict):
        """Set the missing dict and corresponding format."""
        if not missing_dict:
            return
        self.missing_dict = missing_dict
        if self.format_class:
            self.missing = self.format_class(obj=missing_dict)

    def set_diff(self, diff_dict):
        """Set the diff dict and corresponding format."""
        if not diff_dict:
            return
        self.diff_dict = diff_dict
        if self.format_class:
            self.diff = self.format_class(obj=diff_dict)

    def update_pair(self, key, raw_expected):
        """Update a key on one of the comparison dicts, with its raw expected value."""
        if isinstance(key, str):
            self.diff_dict.update({key: raw_expected})
            return

        if isinstance(key, list):
            if len(key) == 4 and isinstance(key[3], int):
                _, _, new_key, index = key
                if new_key not in self.diff_dict:
                    self.diff_dict[new_key] = []
                self.diff_dict[new_key].insert(index, raw_expected)
                return
            if len(key) == 1:
                self.diff_dict.update(unflatten({key[0]: raw_expected}))
                return

        logger.warning("Err... this is unexpected, please open an issue: key={} raw_expected={}", key, raw_expected)


class BaseDoc(metaclass=abc.ABCMeta):
    """Base class for configuration file formats.

    :param path: Path of the config file to be loaded.
    :param string: Config in string format.
    :param obj: Config object (Python dict, YamlDoc, TomlDoc instances).
    :param ignore_keys: List of keys to ignore when using the comparison methods.
    """

    __repr__ = autorepr(["path"])

    def __init__(
        self, *, path: PathOrStr = None, string: str = None, obj: DictOrYamlObject = None, ignore_keys: List[str] = None
    ) -> None:
        self.path = path
        self._string = string
        self._object = obj
        if path is None and string is None and obj is None:
            raise RuntimeError("Inform at least one argument: path, string or object")

        self._ignore_keys = ignore_keys or []
        self._reformatted: Optional[str] = None
        self._loaded = False

    @abc.abstractmethod
    def load(self):
        """Load the configuration from a file, a string or a dict."""

    @property
    def as_string(self) -> str:
        """Contents of the file or the original string provided when the instance was created."""
        if self._string is None:
            self.load()
        return self._string or ""

    @property
    def as_object(self) -> Union[JsonDict, YamlObject]:
        """String content converted to a Python object (dict, YAML object instance, etc.)."""
        if self._object is None:
            self.load()
        return self._object or {}

    @property
    def reformatted(self) -> str:
        """Reformat the configuration dict as a new string (it might not match the original string/file contents)."""
        if self._reformatted is None:
            self.load()
        return self._reformatted or ""

    @classmethod
    def cleanup(cls, *args: List[Any]) -> List[Any]:
        """Cleanup similar values according to the specific format. E.g.: YamlDoc accepts 'True' or 'true'."""
        return list(*args)

    def _create_comparison(self, expected: DictOrYamlObject):
        if not self._ignore_keys:
            return Comparison(self.as_object or {}, expected or {}, self.__class__)

        actual_original: Union[JsonDict, YamlObject] = self.as_object or {}
        actual_copy = actual_original.copy() if isinstance(actual_original, dict) else actual_original

        expected_original: DictOrYamlObject = expected or {}
        if isinstance(expected_original, dict):
            expected_copy = expected_original.copy()
        elif isinstance(expected_original, BaseDoc):
            expected_copy = expected_original.as_object.copy()
        else:
            expected_copy = expected_original
        for key in self._ignore_keys:
            actual_copy.pop(key, None)
            expected_copy.pop(key, None)
        return Comparison(actual_copy, expected_copy, self.__class__)

    def compare_with_flatten(self, expected: Union[JsonDict, "BaseDoc"] = None) -> Comparison:
        """Compare two flattened dictionaries and compute missing and different items."""
        comparison = self._create_comparison(expected)
        if comparison.flat_expected.items() <= comparison.flat_actual.items():
            return comparison

        comparison.set_missing(
            unflatten({k: v for k, v in comparison.flat_expected.items() if k not in comparison.flat_actual})
        )
        diff = {}
        for key, value in comparison.flat_expected.items():
            if key not in comparison.flat_actual:
                continue

            if isinstance(value, list):
                difference = compare_lists_with_dictdiffer(comparison.flat_actual[key], value)
            elif value != comparison.flat_actual[key]:
                difference = value
            else:
                difference = None

            if difference:
                diff.update({key: difference})

        comparison.set_diff(unflatten(diff))
        return comparison

    def compare_with_dictdiffer(
        self, expected: Union[JsonDict, "BaseDoc"] = None, transform_function: Callable = None
    ) -> Comparison:
        """Compare two structures and compute missing and different items using ``dictdiffer``."""
        comparison = self._create_comparison(expected)

        comparison.missing_dict = SortedDict()
        comparison.diff_dict = SortedDict()
        for diff_type, key, values in dictdiffer.diff(comparison.flat_actual, comparison.flat_expected):
            if diff_type == dictdiffer.ADD:
                comparison.missing_dict.update(dict(values))
            elif diff_type == dictdiffer.CHANGE:
                raw_actual, raw_expected = values
                actual_value, expected_value = comparison.format_class.cleanup(raw_actual, raw_expected)
                if actual_value != expected_value:
                    comparison.update_pair(key, raw_expected)

        missing = transform_function(comparison.missing_dict) if transform_function else comparison.missing_dict
        comparison.set_missing(missing)
        comparison.set_diff(comparison.diff_dict)
        return comparison


class InlineTableTomlDecoder(toml.TomlDecoder):  # type: ignore[name-defined]
    """A hacky decoder to work around some bug (or unfinished work) in the Python TOML package.

    https://github.com/uiri/toml/issues/362.
    """

    def get_empty_inline_table(self):
        """Hackity hack for a crappy unmaintained package.

        Total lack of respect, the guy doesn't even reply: https://github.com/uiri/toml/issues/361
        """
        return self.get_empty_table()


class TomlDoc(BaseDoc):
    """TOML configuration format."""

    def __init__(
        self,
        *,
        path: PathOrStr = None,
        string: str = None,
        obj: DictOrYamlObject = None,
        ignore_keys: List[str] = None,
        use_tomlkit=False
    ) -> None:
        super().__init__(path=path, string=string, obj=obj, ignore_keys=ignore_keys)
        self.use_tomlkit = use_tomlkit

    def load(self) -> bool:
        """Load a TOML file by its path, a string or a dict."""
        if self._loaded:
            return False
        if self.path is not None:
            self._string = Path(self.path).read_text(encoding="UTF-8")
        if self._string is not None:
            # TODO: I tried to replace toml by tomlkit, but lots of tests break.
            #  The conversion to OrderedDict is not being done recursively (although I'm not sure this is a problem).
            if self.use_tomlkit:
                self._object = OrderedDict(tomlkit.loads(self._string))
            else:
                self._object = toml.loads(self._string, decoder=InlineTableTomlDecoder(OrderedDict))  # type: ignore[call-arg]
        if self._object is not None:
            if isinstance(self._object, BaseDoc):
                self._reformatted = self._object.reformatted
            else:
                # TODO: tomlkit.dumps() renders comments and I didn't find a way to turn this off,
                #  but comments are being lost when the TOML plugin does dict comparisons.
                if self.use_tomlkit:
                    self._reformatted = tomlkit.dumps(OrderedDict(self._object), sort_keys=True)
                else:
                    self._reformatted = toml.dumps(self._object)
        self._loaded = True
        return True


def traverse_toml_tree(document: tomlkit.TOMLDocument, dictionary):
    """Traverse a TOML document recursively and change values, keeping its formatting and comments."""
    for key, value in dictionary.items():
        if isinstance(value, (dict, OrderedDict)):
            if key in document:
                traverse_toml_tree(document[key], value)
            else:
                document.add(key, value)
        else:
            document[key] = value


class SensibleYAML(YAML):
    """YAML with sensible defaults but an inefficient dump to string.

    `Output of dump() as a string
    <https://yaml.readthedocs.io/en/latest/example.html#output-of-dump-as-a-string>`_
    """

    def __init__(self, *, typ=None, pure=False, output=None, plug_ins=None):
        super().__init__(typ=typ, pure=pure, output=output, plug_ins=plug_ins)
        self.map_indent = 2
        self.sequence_indent = 4
        self.sequence_dash_offset = 2
        self.preserve_quotes = True

    def loads(self, string: str):
        """Load YAML from a string... that unusual use case in a world of files only."""
        return self.load(StringIO(string))

    def dumps(self, data) -> str:
        """Dump to a string... who would want such a thing? One can dump to a file or stdout."""
        output = StringIO()
        self.dump(data, output, transform=None)
        return output.getvalue()


class YamlDoc(BaseDoc):
    """YAML configuration format."""

    updater: SensibleYAML

    def load(self) -> bool:
        """Load a YAML file by its path, a string or a dict."""
        if self._loaded:
            return False

        self.updater = SensibleYAML()

        if self.path is not None:
            self._string = Path(self.path).read_text(encoding="UTF-8")
        if self._string is not None:
            self._object = self.updater.loads(self._string)
        if self._object is not None:
            if isinstance(self._object, BaseDoc):
                self._reformatted = self._object.reformatted
            else:
                self._reformatted = self.updater.dumps(self._object)

        self._loaded = True
        return True

    @property
    def as_object(self) -> CommentedMap:
        """On YAML, this dict is a special object with comments and ordered keys."""
        return super().as_object

    @property
    def as_list(self) -> CommentedSeq:
        """A list of dicts, for typing purposes. On YAML, ``as_object`` might contain a ``list``."""
        return self.as_object

    @classmethod
    def cleanup(cls, *args: List[Any]) -> List[Any]:
        """A boolean "True" or "true" might have the same effect on YAML."""
        return [str(value).lower() if isinstance(value, (int, float, bool)) else value for value in args]


for dict_class in (SortedDict, OrderedDict):
    RoundTripRepresenter.add_representer(dict_class, RoundTripRepresenter.represent_dict)


def is_scalar(value: YamlValue) -> bool:
    """Return True if the value is NOT a dict or a list."""
    return not isinstance(value, LIST_CLASSES + DICT_CLASSES)


def traverse_yaml_tree(yaml_obj: YamlObject, change: Union[JsonDict, OrderedDict]):
    """Traverse a YAML document recursively and change values, keeping its formatting and comments."""
    for key, value in change.items():
        if key not in yaml_obj:
            # Key doesn't exist: we can insert the whole nested OrderedDict at once, no regrets
            last_pos = len(yaml_obj.keys()) + 1
            yaml_obj.insert(last_pos, key, value)
            continue

        if is_scalar(value):
            yaml_obj[key] = value
        elif isinstance(value, OrderedDict):
            traverse_yaml_tree(yaml_obj[key], value)
        elif isinstance(value, list):
            _traverse_yaml_list(yaml_obj, key, value)


def _traverse_yaml_list(yaml_obj: YamlObject, key: str, value: List[YamlValue]):
    for index, element in enumerate(value):
        insert: bool = index >= len(yaml_obj[key])

        if not insert and is_scalar(yaml_obj[key][index]):
            # If the original object is scalar, replace it with whatever element;
            # without traversing, even if it's a dict
            yaml_obj[key][index] = element
            continue

        if is_scalar(element):
            if insert:
                yaml_obj[key].append(element)
            else:
                yaml_obj[key][index] = element
            continue

        traverse_yaml_tree(yaml_obj[key][index], element)  # type: ignore # mypy kept complaining about the Union


class JsonDoc(BaseDoc):
    """JSON configuration format."""

    def load(self) -> bool:
        """Load a JSON file by its path, a string or a dict."""
        if self._loaded:
            return False
        if self.path is not None:
            self._string = Path(self.path).read_text(encoding="UTF-8")
        if self._string is not None:
            self._object = json.loads(self._string, object_pairs_hook=OrderedDict)
        if self._object is not None:
            if isinstance(self._object, BaseDoc):
                self._reformatted = self._object.reformatted
            else:
                # Every file should end with a blank line
                self._reformatted = json.dumps(self._object, sort_keys=True, indent=2) + "\n"
        self._loaded = True
        return True

    @classmethod
    def cleanup(cls, *args: List[Any]) -> List[Any]:
        """Cleanup similar values according to the specific format. E.g.: YamlDoc accepts 'True' or 'true'."""
        return list(args)

1	"""Configuration file formats."""
2	import abc	5✔
3	import json	5✔
4	from collections import OrderedDict	5✔
5	from pathlib import Path	5✔
6	from typing import Any, Callable, List, Optional, Type, Union	5✔
7
8	import dictdiffer	5✔
9	import toml	5✔
10	import tomlkit	5✔
11	from autorepr import autorepr	5✔
12	from loguru import logger	5✔
13	from ruamel.yaml import YAML, RoundTripRepresenter, StringIO	5✔
14	from ruamel.yaml.comments import CommentedMap, CommentedSeq	5✔
15	from sortedcontainers import SortedDict	5✔
16
17	from nitpick.generic import flatten, unflatten	5✔
18	from nitpick.typedefs import JsonDict, PathOrStr, YamlObject, YamlValue	5✔
19
20	DictOrYamlObject = Union[JsonDict, YamlObject, "BaseDoc"]	5✔
21	DICT_CLASSES = (dict, SortedDict, OrderedDict)	5✔
22	LIST_CLASSES = (list, tuple)	5✔
23
24
25	def compare_lists_with_dictdiffer(actual: List[Any], expected: List[Any]) -> List:	5✔
26	"""Compare two lists using dictdiffer."""
27	additions_and_changes = [change for change in dictdiffer.diff(actual, expected) if change[0] != "remove"]	5✔
28	if additions_and_changes:	5✔
29	try:	5✔
30	changed_dict = dictdiffer.patch(additions_and_changes, {})	5✔
31	return list(changed_dict.values())	5✔
32	except KeyError:	5✔
33	return expected	5✔
34	return []	5✔
35
36
37	class Comparison:	5✔
38	"""A comparison between two dictionaries, computing missing items and differences."""
39
40	def __init__(	5✔
41	self,
42	actual: DictOrYamlObject,
43	expected: DictOrYamlObject,
44	format_class: Type["BaseDoc"],
45	) -> None:
46	self.flat_actual = self._normalize_value(actual)	5✔
47	self.flat_expected = self._normalize_value(expected)	5✔
48
49	self.format_class = format_class	5✔
50
51	self.missing: Optional[BaseDoc] = None	5✔
52	self.missing_dict: Union[JsonDict, YamlObject] = {}	5✔
53
54	self.diff: Optional[BaseDoc] = None	5✔
55	self.diff_dict: Union[JsonDict, YamlObject] = {}	5✔
56
57	@property	5✔
58	def has_changes(self) -> bool:	5✔
59	"""Return True is there is a difference or something missing."""
60	return bool(self.missing or self.diff)	5✔
61
62	@staticmethod	5✔
63	def _normalize_value(value: DictOrYamlObject) -> JsonDict:	5✔
64	if isinstance(value, BaseDoc):	5!
UNCOV 65	dict_value: JsonDict = value.as_object	×
66	else:
67	dict_value = value	5✔
68	return flatten(dict_value)	5✔
69
70	def set_missing(self, missing_dict):	5✔
71	"""Set the missing dict and corresponding format."""
72	if not missing_dict:	5✔
73	return	5✔
74	self.missing_dict = missing_dict	5✔
75	if self.format_class:	5!
76	self.missing = self.format_class(obj=missing_dict)	5✔
77
78	def set_diff(self, diff_dict):	5✔
79	"""Set the diff dict and corresponding format."""
80	if not diff_dict:	5✔
81	return	5✔
82	self.diff_dict = diff_dict	5✔
83	if self.format_class:	5!
84	self.diff = self.format_class(obj=diff_dict)	5✔
85
86	def update_pair(self, key, raw_expected):	5✔
87	"""Update a key on one of the comparison dicts, with its raw expected value."""
88	if isinstance(key, str):	5✔
89	self.diff_dict.update({key: raw_expected})	5✔
90	return	5✔
91
92	if isinstance(key, list):	5!
93	if len(key) == 4 and isinstance(key[3], int):	5!
94	_, _, new_key, index = key	5✔
95	if new_key not in self.diff_dict:	5✔
96	self.diff_dict[new_key] = []	5✔
97	self.diff_dict[new_key].insert(index, raw_expected)	5✔
98	return	5✔
UNCOV 99	if len(key) == 1:	×
UNCOV 100	self.diff_dict.update(unflatten({key[0]: raw_expected}))	×
UNCOV 101	return	×
102
UNCOV 103	logger.warning("Err... this is unexpected, please open an issue: key={} raw_expected={}", key, raw_expected)	×
104
105
106	class BaseDoc(metaclass=abc.ABCMeta):	5✔
107	"""Base class for configuration file formats.
108
109	:param path: Path of the config file to be loaded.
110	:param string: Config in string format.
111	:param obj: Config object (Python dict, YamlDoc, TomlDoc instances).
112	:param ignore_keys: List of keys to ignore when using the comparison methods.
113	"""
114
115	__repr__ = autorepr(["path"])	5✔
116
117	def __init__(	5✔
118	self, *, path: PathOrStr = None, string: str = None, obj: DictOrYamlObject = None, ignore_keys: List[str] = None
119	) -> None:
120	self.path = path	5✔
121	self._string = string	5✔
122	self._object = obj	5✔
123	if path is None and string is None and obj is None:	5!
UNCOV 124	raise RuntimeError("Inform at least one argument: path, string or object")	×
125
126	self._ignore_keys = ignore_keys or []	5✔
127	self._reformatted: Optional[str] = None	5✔
128	self._loaded = False	5✔
129
130	@abc.abstractmethod	5✔
131	def load(self):	3✔
132	"""Load the configuration from a file, a string or a dict."""
133
134	@property	5✔
135	def as_string(self) -> str:	5✔
136	"""Contents of the file or the original string provided when the instance was created."""
137	if self._string is None:	5!
138	self.load()	×
139	return self._string or ""	5✔
140
141	@property	5✔
142	def as_object(self) -> Union[JsonDict, YamlObject]:	5✔
143	"""String content converted to a Python object (dict, YAML object instance, etc.)."""
144	if self._object is None:	5✔
145	self.load()	5✔
146	return self._object or {}	5✔
147
148	@property	5✔
149	def reformatted(self) -> str:	5✔
150	"""Reformat the configuration dict as a new string (it might not match the original string/file contents)."""
151	if self._reformatted is None:	5!
152	self.load()	5✔
153	return self._reformatted or ""	5✔
154
155	@classmethod	5✔
156	def cleanup(cls, *args: List[Any]) -> List[Any]:	5✔
157	"""Cleanup similar values according to the specific format. E.g.: YamlDoc accepts 'True' or 'true'."""
UNCOV 158	return list(*args)	×
159
160	def _create_comparison(self, expected: DictOrYamlObject):	5✔
161	if not self._ignore_keys:	5✔
162	return Comparison(self.as_object or {}, expected or {}, self.__class__)	5✔
163
164	actual_original: Union[JsonDict, YamlObject] = self.as_object or {}	5✔
165	actual_copy = actual_original.copy() if isinstance(actual_original, dict) else actual_original	5✔
166
167	expected_original: DictOrYamlObject = expected or {}	5✔
168	if isinstance(expected_original, dict):	5!
169	expected_copy = expected_original.copy()	5✔
UNCOV 170	elif isinstance(expected_original, BaseDoc):	×
UNCOV 171	expected_copy = expected_original.as_object.copy()	×
172	else:
UNCOV 173	expected_copy = expected_original	×
174	for key in self._ignore_keys:	5✔
175	actual_copy.pop(key, None)	5✔
176	expected_copy.pop(key, None)	5✔
177	return Comparison(actual_copy, expected_copy, self.__class__)	5✔
178
179	def compare_with_flatten(self, expected: Union[JsonDict, "BaseDoc"] = None) -> Comparison:	5✔
180	"""Compare two flattened dictionaries and compute missing and different items."""
181	comparison = self._create_comparison(expected)	5✔
182	if comparison.flat_expected.items() <= comparison.flat_actual.items():	5✔
183	return comparison	5✔
184
185	comparison.set_missing(	5✔
186	unflatten({k: v for k, v in comparison.flat_expected.items() if k not in comparison.flat_actual})
187	)
188	diff = {}	5✔
189	for key, value in comparison.flat_expected.items():	5✔
190	if key not in comparison.flat_actual:	5✔
191	continue	5✔
192
193	if isinstance(value, list):	5✔
194	difference = compare_lists_with_dictdiffer(comparison.flat_actual[key], value)	5✔
195	elif value != comparison.flat_actual[key]:	5✔
196	difference = value	5✔
197	else:
198	difference = None	5✔
199
200	if difference:	5✔
201	diff.update({key: difference})	5✔
202
203	comparison.set_diff(unflatten(diff))	5✔
204	return comparison	5✔
205
206	def compare_with_dictdiffer(	5✔
207	self, expected: Union[JsonDict, "BaseDoc"] = None, transform_function: Callable = None
208	) -> Comparison:
209	"""Compare two structures and compute missing and different items using ``dictdiffer``."""
210	comparison = self._create_comparison(expected)	5✔
211
212	comparison.missing_dict = SortedDict()	5✔
213	comparison.diff_dict = SortedDict()	5✔
214	for diff_type, key, values in dictdiffer.diff(comparison.flat_actual, comparison.flat_expected):	5✔
215	if diff_type == dictdiffer.ADD:	5✔
216	comparison.missing_dict.update(dict(values))	5✔
217	elif diff_type == dictdiffer.CHANGE:	5✔
218	raw_actual, raw_expected = values	5✔
219	actual_value, expected_value = comparison.format_class.cleanup(raw_actual, raw_expected)	5✔
220	if actual_value != expected_value:	5!
221	comparison.update_pair(key, raw_expected)	5✔
222
223	missing = transform_function(comparison.missing_dict) if transform_function else comparison.missing_dict	5✔
224	comparison.set_missing(missing)	5✔
225	comparison.set_diff(comparison.diff_dict)	5✔
226	return comparison	5✔
227
228
229	class InlineTableTomlDecoder(toml.TomlDecoder): # type: ignore[name-defined]	5✔
230	"""A hacky decoder to work around some bug (or unfinished work) in the Python TOML package.
231
232	https://github.com/uiri/toml/issues/362.
233	"""
234
235	def get_empty_inline_table(self):	5✔
236	"""Hackity hack for a crappy unmaintained package.
237
238	Total lack of respect, the guy doesn't even reply: https://github.com/uiri/toml/issues/361
239	"""
240	return self.get_empty_table()	5✔
241
242
243	class TomlDoc(BaseDoc):	5✔
244	"""TOML configuration format."""
245
246	def __init__(	5✔
247	self,
248	*,
249	path: PathOrStr = None,
250	string: str = None,
251	obj: DictOrYamlObject = None,
252	ignore_keys: List[str] = None,
253	use_tomlkit=False
254	) -> None:
255	super().__init__(path=path, string=string, obj=obj, ignore_keys=ignore_keys)	5✔
256	self.use_tomlkit = use_tomlkit	5✔
257
258	def load(self) -> bool:	5✔
259	"""Load a TOML file by its path, a string or a dict."""
260	if self._loaded:	5!
UNCOV 261	return False	×
262	if self.path is not None:	5✔
263	self._string = Path(self.path).read_text(encoding="UTF-8")	5✔
264	if self._string is not None:	5✔
265	# TODO: I tried to replace toml by tomlkit, but lots of tests break.
266	# The conversion to OrderedDict is not being done recursively (although I'm not sure this is a problem).
267	if self.use_tomlkit:	5!
NEW 268	self._object = OrderedDict(tomlkit.loads(self._string))	×
269	else:
270	self._object = toml.loads(self._string, decoder=InlineTableTomlDecoder(OrderedDict)) # type: ignore[call-arg]	5✔
271	if self._object is not None:	5!
272	if isinstance(self._object, BaseDoc):	5!
UNCOV 273	self._reformatted = self._object.reformatted	×
274	else:
275	# TODO: tomlkit.dumps() renders comments and I didn't find a way to turn this off,
276	# but comments are being lost when the TOML plugin does dict comparisons.
277	if self.use_tomlkit:	5!
NEW 278	self._reformatted = tomlkit.dumps(OrderedDict(self._object), sort_keys=True)	×
279	else:
280	self._reformatted = toml.dumps(self._object)	5✔
281	self._loaded = True	5✔
282	return True	5✔
283
284
285	def traverse_toml_tree(document: tomlkit.TOMLDocument, dictionary):	5✔
286	"""Traverse a TOML document recursively and change values, keeping its formatting and comments."""
287	for key, value in dictionary.items():	5✔
288	if isinstance(value, (dict, OrderedDict)):	5✔
289	if key in document:	5✔
290	traverse_toml_tree(document[key], value)	5✔
291	else:
292	document.add(key, value)	5✔
293	else:
294	document[key] = value	5✔
295
296
297	class SensibleYAML(YAML):	5✔
298	"""YAML with sensible defaults but an inefficient dump to string.
299
300	`Output of dump() as a string
301	<https://yaml.readthedocs.io/en/latest/example.html#output-of-dump-as-a-string>`_
302	"""
303
304	def __init__(self, *, typ=None, pure=False, output=None, plug_ins=None):	5✔
305	super().__init__(typ=typ, pure=pure, output=output, plug_ins=plug_ins)	5✔
306	self.map_indent = 2	5✔
307	self.sequence_indent = 4	5✔
308	self.sequence_dash_offset = 2	5✔
309	self.preserve_quotes = True	5✔
310
311	def loads(self, string: str):	5✔
312	"""Load YAML from a string... that unusual use case in a world of files only."""
313	return self.load(StringIO(string))	5✔
314
315	def dumps(self, data) -> str:	5✔
316	"""Dump to a string... who would want such a thing? One can dump to a file or stdout."""
317	output = StringIO()	5✔
318	self.dump(data, output, transform=None)	5✔
319	return output.getvalue()	5✔
320
321
322	class YamlDoc(BaseDoc):	5✔
323	"""YAML configuration format."""
324
325	updater: SensibleYAML	5✔
326
327	def load(self) -> bool:	5✔
328	"""Load a YAML file by its path, a string or a dict."""
329	if self._loaded:	5!
UNCOV 330	return False	×
331
332	self.updater = SensibleYAML()	5✔
333
334	if self.path is not None:	5✔
335	self._string = Path(self.path).read_text(encoding="UTF-8")	5✔
336	if self._string is not None:	5✔
337	self._object = self.updater.loads(self._string)	5✔
338	if self._object is not None:	5✔
339	if isinstance(self._object, BaseDoc):	5!
UNCOV 340	self._reformatted = self._object.reformatted	×
341	else:
342	self._reformatted = self.updater.dumps(self._object)	5✔
343
344	self._loaded = True	5✔
345	return True	5✔
346
347	@property	5✔
348	def as_object(self) -> CommentedMap:	5✔
349	"""On YAML, this dict is a special object with comments and ordered keys."""
350	return super().as_object	5✔
351
352	@property	5✔
353	def as_list(self) -> CommentedSeq:	5✔
354	"""A list of dicts, for typing purposes. On YAML, ``as_object`` might contain a ``list``."""
355	return self.as_object	5✔
356
357	@classmethod	5✔
358	def cleanup(cls, *args: List[Any]) -> List[Any]:	5✔
359	"""A boolean "True" or "true" might have the same effect on YAML."""
360	return [str(value).lower() if isinstance(value, (int, float, bool)) else value for value in args]	5✔
361
362
363	for dict_class in (SortedDict, OrderedDict):	5✔
364	RoundTripRepresenter.add_representer(dict_class, RoundTripRepresenter.represent_dict)	5✔
365
366
367	def is_scalar(value: YamlValue) -> bool:	5✔
368	"""Return True if the value is NOT a dict or a list."""
369	return not isinstance(value, LIST_CLASSES + DICT_CLASSES)	5✔
370
371
372	def traverse_yaml_tree(yaml_obj: YamlObject, change: Union[JsonDict, OrderedDict]):	5✔
373	"""Traverse a YAML document recursively and change values, keeping its formatting and comments."""
374	for key, value in change.items():	5✔
375	if key not in yaml_obj:	5✔
376	# Key doesn't exist: we can insert the whole nested OrderedDict at once, no regrets
377	last_pos = len(yaml_obj.keys()) + 1	5✔
378	yaml_obj.insert(last_pos, key, value)	5✔
379	continue	5✔
380
381	if is_scalar(value):	5✔
382	yaml_obj[key] = value	5✔
383	elif isinstance(value, OrderedDict):	5✔
384	traverse_yaml_tree(yaml_obj[key], value)	5✔
385	elif isinstance(value, list):	5!
386	_traverse_yaml_list(yaml_obj, key, value)	5✔
387
388
389	def _traverse_yaml_list(yaml_obj: YamlObject, key: str, value: List[YamlValue]):	5✔
390	for index, element in enumerate(value):	5✔
391	insert: bool = index >= len(yaml_obj[key])	5✔
392
393	if not insert and is_scalar(yaml_obj[key][index]):	5✔
394	# If the original object is scalar, replace it with whatever element;
395	# without traversing, even if it's a dict
396	yaml_obj[key][index] = element	5✔
397	continue	5✔
398
399	if is_scalar(element):	5✔
400	if insert:	5!
401	yaml_obj[key].append(element)	5✔
402	else:
UNCOV 403	yaml_obj[key][index] = element	×
404	continue	1✔
405
406	traverse_yaml_tree(yaml_obj[key][index], element) # type: ignore # mypy kept complaining about the Union	5✔
407
408
409	class JsonDoc(BaseDoc):	5✔
410	"""JSON configuration format."""
411
412	def load(self) -> bool:	5✔
413	"""Load a JSON file by its path, a string or a dict."""
414	if self._loaded:	5!
UNCOV 415	return False	×
416	if self.path is not None:	5✔
417	self._string = Path(self.path).read_text(encoding="UTF-8")	5✔
418	if self._string is not None:	5✔
419	self._object = json.loads(self._string, object_pairs_hook=OrderedDict)	5✔
420	if self._object is not None:	5!
421	if isinstance(self._object, BaseDoc):	5!
UNCOV 422	self._reformatted = self._object.reformatted	×
423	else:
424	# Every file should end with a blank line
425	self._reformatted = json.dumps(self._object, sort_keys=True, indent=2) + "\n"	5✔
426	self._loaded = True	5✔
427	return True	5✔
428
429	@classmethod	5✔
430	def cleanup(cls, *args: List[Any]) -> List[Any]:	5✔
431	"""Cleanup similar values according to the specific format. E.g.: YamlDoc accepts 'True' or 'true'."""
UNCOV 432	return list(args)	×

andreoliwa / nitpick / 1644037831

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous