543314494

Build Type

github

Committed by GitHub

Commit Message

Merge fbd45f535 into 056dc133d

Pull Request Pull Request #255: Experimental CLI interface (alpha version)

Run Details

354 of 430 branches covered (82.33%)

Branch coverage included in aggregate %.

572 of 593 new or added lines in 23 files covered. (96.46%)

1 existing line in 1 file now uncovered.

1363 of 1432 relevant lines covered (95.18%)

3.8 hits per line

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

86.14

/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 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_format = None  # type: Optional[BaseFormat]
        self.missing_dict = {}  # type: Union[JsonDict, YamlData]

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

    @staticmethod
    def _normalize_value(value: Union[JsonDict, YamlData, "BaseFormat"]) -> JsonDict:
        if isinstance(value, BaseFormat):
            dict_value = value.as_data  # type: JsonDict
        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_format = 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_format = 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.
    """

    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 = None  # type: Optional[str]
        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 = self.as_data or {}  # type: Union[JsonDict, YamlData]
        actual_copy = actual_original.copy() if isinstance(actual_original, dict) else actual_original

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

andreoliwa / nitpick / 543314494

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