1635541107

Build Type

github

Committed by GitHub

Commit Message

Merge b7edd0924 into c871a313b

Pull Request Pull Request #429: feat(json): autofix JSON files

Run Details

519 of 563 branches covered (92.18%)

Branch coverage included in aggregate %.

58 of 58 new or added lines in 7 files covered. (100.0%)

4 existing lines in 1 file now uncovered.

1942 of 1994 relevant lines covered (97.39%)

4.85 hits per line

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

85.29

/src/nitpick/formats.py

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

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

from nitpick.generic import flatten, unflatten
from nitpick.typedefs import JsonDict, PathOrStr, YamlData


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

    def __init__(
        self,
        actual: Union[JsonDict, YamlData, "BaseFormat"],
        expected: Union[JsonDict, YamlData, "BaseFormat"],
        format_class: Type["BaseFormat"],
    ) -> None:
        self.flat_actual = self._normalize_value(actual)
        self.flat_expected = self._normalize_value(expected)

        self.format_class = format_class

        self.missing: Optional[BaseFormat] = None
        self.missing_dict: Union[JsonDict, YamlData] = {}

        self.diff: Optional[BaseFormat] = None
        self.diff_dict: Union[JsonDict, YamlData] = {}

    @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: Union[JsonDict, YamlData, "BaseFormat"]) -> JsonDict:
        if isinstance(value, BaseFormat):
            dict_value: JsonDict = value.as_data
        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(data=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(data=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 BaseFormat(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 data: Config data in Python format (dict, YAMLFormat, TOMLFormat 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,
        data: Union[JsonDict, YamlData, "BaseFormat"] = None,
        ignore_keys: List[str] = None
    ) -> None:
        self.path = path
        self._string = string
        self._data = data
        if path is None and string is None and data is None:
            raise RuntimeError("Inform at least one argument: path, string or data")

        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_data(self) -> Union[JsonDict, YamlData]:
        """String content converted to a Python data structure (a dict, YAML data, etc.)."""
        if self._data is None:
            self.load()
        return self._data 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.: YAMLFormat accepts 'True' or 'true'."""
        return list(*args)

    def _create_comparison(self, expected: Union[JsonDict, YamlData, "BaseFormat"]):
        if not self._ignore_keys:
            return Comparison(self.as_data or {}, expected or {}, self.__class__)

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

        expected_original: Union[JsonDict, YamlData, "BaseFormat"] = expected or {}
        if isinstance(expected_original, dict):
            expected_copy = expected_original.copy()
        elif isinstance(expected_original, BaseFormat):
            expected_copy = expected_original.as_data.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, "BaseFormat"] = 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})
        )
        comparison.set_diff(
            unflatten(
                {
                    k: v
                    for k, v in comparison.flat_expected.items()
                    if k in comparison.flat_actual and v != comparison.flat_actual[k]
                }
            )
        )
        return comparison

    def compare_with_dictdiffer(
        self, expected: Union[JsonDict, "BaseFormat"] = 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 TOMLFormat(BaseFormat):
    """TOML configuration format."""

    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).
            # self._data = OrderedDict(tomlkit.loads(self._string))
            self._data = toml.loads(self._string, _dict=OrderedDict)
        if self._data is not None:
            if isinstance(self._data, BaseFormat):
                self._reformatted = self._data.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.
                # self._reformatted = tomlkit.dumps(OrderedDict(self._data), sort_keys=True)
                self._reformatted = toml.dumps(self._data)
        self._loaded = True
        return True


class YAMLFormat(BaseFormat):
    """YAML configuration format."""

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

        yaml = YAML()
        yaml.map_indent = 2
        yaml.sequence_indent = 4
        yaml.sequence_dash_offset = 2

        if self.path is not None:
            self._string = Path(self.path).read_text(encoding="UTF-8")
        if self._string is not None:
            self._data = yaml.load(io.StringIO(self._string))
        if self._data is not None:
            if isinstance(self._data, BaseFormat):
                self._reformatted = self._data.reformatted
            else:
                output = io.StringIO()
                yaml.dump(self._data, output)
                self._reformatted = output.getvalue()

        self._loaded = True
        return True

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

    @property
    def as_list(self) -> CommentedSeq:
        """A list of dicts. On YAML, ``as_data`` might contain a ``list``. This property is just a proxy for typing."""
        return self.as_data

    @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]


RoundTripRepresenter.add_representer(SortedDict, RoundTripRepresenter.represent_dict)


class JSONFormat(BaseFormat):
    """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._data = json.loads(self._string, object_pairs_hook=OrderedDict)
        if self._data is not None:
            if isinstance(self._data, BaseFormat):
                self._reformatted = self._data.reformatted
            else:
                # Every file should end with a blank line
                self._reformatted = json.dumps(self._data, 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.: YAMLFormat accepts 'True' or 'true'."""
        return list(args)

1	"""Configuration file formats."""
2	import abc	5✔
3	import io	5✔
4	import json	5✔
5	from collections import OrderedDict	5✔
6	from pathlib import Path	5✔
7	from typing import Any, Callable, List, Optional, Type, Union	5✔
8
9	import dictdiffer	5✔
10	import toml	5✔
11	from autorepr import autorepr	5✔
12	from loguru import logger	5✔
13	from ruamel.yaml import YAML, RoundTripRepresenter	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, YamlData	5✔
19
20
21	class Comparison:	5✔
22	"""A comparison between two dictionaries, computing missing items and differences."""
23
24	def __init__(	5✔
25	self,
26	actual: Union[JsonDict, YamlData, "BaseFormat"],
27	expected: Union[JsonDict, YamlData, "BaseFormat"],
28	format_class: Type["BaseFormat"],
29	) -> None:
30	self.flat_actual = self._normalize_value(actual)	5✔
31	self.flat_expected = self._normalize_value(expected)	5✔
32
33	self.format_class = format_class	5✔
34
35	self.missing: Optional[BaseFormat] = None	5✔
36	self.missing_dict: Union[JsonDict, YamlData] = {}	5✔
37
38	self.diff: Optional[BaseFormat] = None	5✔
39	self.diff_dict: Union[JsonDict, YamlData] = {}	5✔
40
41	@property	5✔
42	def has_changes(self) -> bool:	5✔
43	"""Return True is there is a difference or something missing."""
44	return bool(self.missing or self.diff)	5✔
45
46	@staticmethod	5✔
47	def _normalize_value(value: Union[JsonDict, YamlData, "BaseFormat"]) -> JsonDict:	5✔
48	if isinstance(value, BaseFormat):	5!
49	dict_value: JsonDict = value.as_data	×
50	else:
51	dict_value = value	5✔
52	return flatten(dict_value)	5✔
53
54	def set_missing(self, missing_dict):	5✔
55	"""Set the missing dict and corresponding format."""
56	if not missing_dict:	5✔
57	return	5✔
58	self.missing_dict = missing_dict	5✔
59	if self.format_class:	5!
60	self.missing = self.format_class(data=missing_dict)	5✔
61
62	def set_diff(self, diff_dict):	5✔
63	"""Set the diff dict and corresponding format."""
64	if not diff_dict:	5✔
65	return	5✔
66	self.diff_dict = diff_dict	5✔
67	if self.format_class:	5!
68	self.diff = self.format_class(data=diff_dict)	5✔
69
70	def update_pair(self, key, raw_expected):	5✔
71	"""Update a key on one of the comparison dicts, with its raw expected value."""
72	if isinstance(key, str):	5✔
73	self.diff_dict.update({key: raw_expected})	5✔
74	return	5✔
75
76	if isinstance(key, list):	5!
77	if len(key) == 4 and isinstance(key[3], int):	5!
78	_, _, new_key, index = key	5✔
79	if new_key not in self.diff_dict:	5✔
80	self.diff_dict[new_key] = []	5✔
81	self.diff_dict[new_key].insert(index, raw_expected)	5✔
82	return	5✔
UNCOV 83	if len(key) == 1:	×
UNCOV 84	self.diff_dict.update(unflatten({key[0]: raw_expected}))	×
UNCOV 85	return	×
86
87	logger.warning("Err... this is unexpected, please open an issue: key={} raw_expected={}", key, raw_expected)	×
88
89
90	class BaseFormat(metaclass=abc.ABCMeta):	5✔
91	"""Base class for configuration file formats.
92
93	:param path: Path of the config file to be loaded.
94	:param string: Config in string format.
95	:param data: Config data in Python format (dict, YAMLFormat, TOMLFormat instances).
96	:param ignore_keys: List of keys to ignore when using the comparison methods.
97	"""
98
99	__repr__ = autorepr(["path"])	5✔
100
101	def __init__(	5✔
102	self,
103	*,
104	path: PathOrStr = None,
105	string: str = None,
106	data: Union[JsonDict, YamlData, "BaseFormat"] = None,
107	ignore_keys: List[str] = None
108	) -> None:
109	self.path = path	5✔
110	self._string = string	5✔
111	self._data = data	5✔
112	if path is None and string is None and data is None:	5!
113	raise RuntimeError("Inform at least one argument: path, string or data")	×
114
115	self._ignore_keys = ignore_keys or []	5✔
116	self._reformatted: Optional[str] = None	5✔
117	self._loaded = False	5✔
118
119	@abc.abstractmethod	5✔
120	def load(self):	3✔
121	"""Load the configuration from a file, a string or a dict."""
122
123	@property	5✔
124	def as_string(self) -> str:	5✔
125	"""Contents of the file or the original string provided when the instance was created."""
126	if self._string is None:	5!
127	self.load()	×
128	return self._string or ""	5✔
129
130	@property	5✔
131	def as_data(self) -> Union[JsonDict, YamlData]:	5✔
132	"""String content converted to a Python data structure (a dict, YAML data, etc.)."""
133	if self._data is None:	5✔
134	self.load()	5✔
135	return self._data or {}	5✔
136
137	@property	5✔
138	def reformatted(self) -> str:	5✔
139	"""Reformat the configuration dict as a new string (it might not match the original string/file contents)."""
140	if self._reformatted is None:	5!
141	self.load()	5✔
142	return self._reformatted or ""	5✔
143
144	@classmethod	5✔
145	def cleanup(cls, *args: List[Any]) -> List[Any]:	5✔
146	"""Cleanup similar values according to the specific format. E.g.: YAMLFormat accepts 'True' or 'true'."""
147	return list(*args)	×
148
149	def _create_comparison(self, expected: Union[JsonDict, YamlData, "BaseFormat"]):	5✔
150	if not self._ignore_keys:	5✔
151	return Comparison(self.as_data or {}, expected or {}, self.__class__)	5✔
152
153	actual_original: Union[JsonDict, YamlData] = self.as_data or {}	5✔
154	actual_copy = actual_original.copy() if isinstance(actual_original, dict) else actual_original	5✔
155
156	expected_original: Union[JsonDict, YamlData, "BaseFormat"] = expected or {}	5✔
157	if isinstance(expected_original, dict):	5!
158	expected_copy = expected_original.copy()	5✔
159	elif isinstance(expected_original, BaseFormat):	×
160	expected_copy = expected_original.as_data.copy()	×
161	else:
162	expected_copy = expected_original	×
163	for key in self._ignore_keys:	5✔
164	actual_copy.pop(key, None)	5✔
165	expected_copy.pop(key, None)	5✔
166	return Comparison(actual_copy, expected_copy, self.__class__)	5✔
167
168	def compare_with_flatten(self, expected: Union[JsonDict, "BaseFormat"] = None) -> Comparison:	5✔
169	"""Compare two flattened dictionaries and compute missing and different items."""
170	comparison = self._create_comparison(expected)	5✔
171	if comparison.flat_expected.items() <= comparison.flat_actual.items():	5✔
172	return comparison	5✔
173
174	comparison.set_missing(	5✔
175	unflatten({k: v for k, v in comparison.flat_expected.items() if k not in comparison.flat_actual})
176	)
177	comparison.set_diff(	5✔
178	unflatten(
179	{
180	k: v
181	for k, v in comparison.flat_expected.items()
182	if k in comparison.flat_actual and v != comparison.flat_actual[k]
183	}
184	)
185	)
186	return comparison	5✔
187
188	def compare_with_dictdiffer(	5✔
189	self, expected: Union[JsonDict, "BaseFormat"] = None, transform_function: Callable = None
190	) -> Comparison:
191	"""Compare two structures and compute missing and different items using ``dictdiffer``."""
192	comparison = self._create_comparison(expected)	5✔
193
194	comparison.missing_dict = SortedDict()	5✔
195	comparison.diff_dict = SortedDict()	5✔
196	for diff_type, key, values in dictdiffer.diff(comparison.flat_actual, comparison.flat_expected):	5✔
197	if diff_type == dictdiffer.ADD:	5✔
198	comparison.missing_dict.update(dict(values))	5✔
199	elif diff_type == dictdiffer.CHANGE:	5✔
200	raw_actual, raw_expected = values	5✔
201	actual_value, expected_value = comparison.format_class.cleanup(raw_actual, raw_expected)	5✔
202	if actual_value != expected_value:	5!
203	comparison.update_pair(key, raw_expected)	5✔
204
205	missing = transform_function(comparison.missing_dict) if transform_function else comparison.missing_dict	5✔
206	comparison.set_missing(missing)	5✔
207	comparison.set_diff(comparison.diff_dict)	5✔
208	return comparison	5✔
209
210
211	class TOMLFormat(BaseFormat):	5✔
212	"""TOML configuration format."""
213
214	def load(self) -> bool:	5✔
215	"""Load a TOML file by its path, a string or a dict."""
216	if self._loaded:	5!
217	return False	×
218	if self.path is not None:	5✔
219	self._string = Path(self.path).read_text(encoding="UTF-8")	5✔
220	if self._string is not None:	5✔
221	# TODO: I tried to replace toml by tomlkit, but lots of tests break.
222	# The conversion to OrderedDict is not being done recursively (although I'm not sure this is a problem).
223	# self._data = OrderedDict(tomlkit.loads(self._string))
224	self._data = toml.loads(self._string, _dict=OrderedDict)	5✔
225	if self._data is not None:	5!
226	if isinstance(self._data, BaseFormat):	5!
227	self._reformatted = self._data.reformatted	×
228	else:
229	# TODO: tomlkit.dumps() renders comments and I didn't find a way to turn this off,
230	# but comments are being lost when the TOML plugin does dict comparisons.
231	# self._reformatted = tomlkit.dumps(OrderedDict(self._data), sort_keys=True)
232	self._reformatted = toml.dumps(self._data)	5✔
233	self._loaded = True	5✔
234	return True	5✔
235
236
237	class YAMLFormat(BaseFormat):	5✔
238	"""YAML configuration format."""
239
240	def load(self) -> bool:	5✔
241	"""Load a YAML file by its path, a string or a dict."""
242	if self._loaded:	5!
243	return False	×
244
245	yaml = YAML()	5✔
246	yaml.map_indent = 2	5✔
247	yaml.sequence_indent = 4	5✔
248	yaml.sequence_dash_offset = 2	5✔
249
250	if self.path is not None:	5✔
251	self._string = Path(self.path).read_text(encoding="UTF-8")	5✔
252	if self._string is not None:	5✔
253	self._data = yaml.load(io.StringIO(self._string))	5✔
254	if self._data is not None:	5✔
255	if isinstance(self._data, BaseFormat):	5!
256	self._reformatted = self._data.reformatted	×
257	else:
258	output = io.StringIO()	5✔
259	yaml.dump(self._data, output)	5✔
260	self._reformatted = output.getvalue()	5✔
261
262	self._loaded = True	5✔
263	return True	5✔
264
265	@property	5✔
266	def as_data(self) -> CommentedMap:	5✔
267	"""On YAML, this dict is a special object with comments and ordered keys."""
268	return super().as_data	5✔
269
270	@property	5✔
271	def as_list(self) -> CommentedSeq:	5✔
272	"""A list of dicts. On YAML, ``as_data`` might contain a ``list``. This property is just a proxy for typing."""
273	return self.as_data	5✔
274
275	@classmethod	5✔
276	def cleanup(cls, *args: List[Any]) -> List[Any]:	5✔
277	"""A boolean "True" or "true" might have the same effect on YAML."""
278	return [str(value).lower() if isinstance(value, (int, float, bool)) else value for value in args]	5✔
279
280
281	RoundTripRepresenter.add_representer(SortedDict, RoundTripRepresenter.represent_dict)	5✔
282
283
284	class JSONFormat(BaseFormat):	5✔
285	"""JSON configuration format."""
286
287	def load(self) -> bool:	5✔
288	"""Load a JSON file by its path, a string or a dict."""
289	if self._loaded:	5!
290	return False	×
291	if self.path is not None:	5✔
292	self._string = Path(self.path).read_text(encoding="UTF-8")	5✔
293	if self._string is not None:	5✔
294	self._data = json.loads(self._string, object_pairs_hook=OrderedDict)	5✔
295	if self._data is not None:	5!
296	if isinstance(self._data, BaseFormat):	5!
297	self._reformatted = self._data.reformatted	×
298	else:
299	# Every file should end with a blank line
300	self._reformatted = json.dumps(self._data, sort_keys=True, indent=2) + "\n"	5✔
301	self._loaded = True	5✔
302	return True	5✔
303
304	@classmethod	5✔
305	def cleanup(cls, *args: List[Any]) -> List[Any]:	5✔
306	"""Cleanup similar values according to the specific format. E.g.: YAMLFormat accepts 'True' or 'true'."""
UNCOV 307	return list(args)	×

andreoliwa / nitpick / 1635541107

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