1669453638

Build Type

github

Committed by GitHub

Commit Message

Merge 6c228d75f into 82625d2fb

Pull Request Pull Request #434: feat(yaml): autofix .pre-commit-config.yaml (note: style changed!)

Run Details

538 of 588 branches covered (91.5%)

Branch coverage included in aggregate %.

190 of 200 new or added lines in 7 files covered. (95.0%)

15 existing lines in 2 files now uncovered.

1992 of 2060 relevant lines covered (96.7%)

4.81 hits per line

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

84.72

/src/nitpick/style/core.py

"""Style files."""
import os
from collections import OrderedDict
from dataclasses import dataclass
from functools import lru_cache
from pathlib import Path
from typing import Dict, Iterator, Optional, Set, Tuple, Type
from urllib.parse import urljoin, urlsplit, urlunsplit

from identify import identify
from loguru import logger
from more_itertools import always_iterable
from slugify import slugify
from toml import TomlDecodeError

from nitpick import __version__, fields
from nitpick.constants import (
    CACHE_DIR_NAME,
    DOT_SLASH,
    MERGED_STYLE_TOML,
    NITPICK_STYLE_TOML,
    NITPICK_STYLES_INCLUDE_JMEX,
    PROJECT_NAME,
    PROJECT_OWNER,
    PYPROJECT_TOML,
    TOML_EXTENSION,
)
from nitpick.documents import TomlDoc
from nitpick.exceptions import QuitComplainingError, pretty_exception
from nitpick.generic import DictBlender, is_url, search_dict
from nitpick.plugins.base import NitpickPlugin
from nitpick.plugins.info import FileInfo
from nitpick.project import Project, glob_files
from nitpick.schemas import BaseStyleSchema, flatten_marshmallow_errors
from nitpick.style.config import ConfigValidator
from nitpick.style.fetchers import StyleFetcherManager
from nitpick.style.fetchers.github import GitHubURL
from nitpick.typedefs import JsonDict, StrOrIterable, StrOrList, mypy_property
from nitpick.violations import Fuss, Reporter, StyleViolations

Plugins = Set[Type[NitpickPlugin]]


@dataclass(repr=True)
class Style:  # pylint: disable=too-many-instance-attributes
    """Include styles recursively from one another."""

    project: Project
    offline: bool
    cache_option: str

    def __post_init__(self) -> None:
        """Initialize dependant fields."""
        self._blender: DictBlender = DictBlender()
        self._already_included: Set[str] = set()
        self._first_full_path: str = ""
        self._dynamic_schema_class: type = BaseStyleSchema
        self._style_fetcher_manager = StyleFetcherManager(self.offline, self.cache_dir, self.cache_option)
        self._config_validator = ConfigValidator(self.project)
        self.rebuild_dynamic_schema()

    def __hash__(self):
        """Calculate hash on hashable items so lru_cache knows how to cache data from this class."""
        return hash((self.project, self.offline, self.cache_option))

    @mypy_property
    @lru_cache()
    def cache_dir(self) -> Path:
        """Clear the cache directory (on the project root or on the current directory)."""
        path: Path = self.project.root / CACHE_DIR_NAME / PROJECT_NAME

        # TODO: check if the merged style file is still needed; if not, this line can be removed
        path.mkdir(parents=True, exist_ok=True)

        return path

    @staticmethod
    def get_default_style_url():
        """Return the URL of the default style for the current version."""
        return GitHubURL(PROJECT_OWNER, PROJECT_NAME, f"v{__version__}", NITPICK_STYLE_TOML, None).long_protocol_url

    def find_initial_styles(self, configured_styles: StrOrIterable) -> Iterator[Fuss]:
        """Find the initial style(s) and include them."""
        if configured_styles:
            chosen_styles: StrOrIterable = list(configured_styles)
            log_message = f"Using styles configured in {PYPROJECT_TOML}"
        else:
            paths = glob_files(self.project.root, [NITPICK_STYLE_TOML])
            if paths:
                chosen_styles = str(sorted(paths)[0])
                log_message = "Using local style found climbing the directory tree"
            else:
                chosen_styles = self.get_default_style_url()
                log_message = "Using default remote Nitpick style"
        logger.info(f"{log_message}: {chosen_styles}")

        yield from self.include_multiple_styles(chosen_styles)

    def include_multiple_styles(self, chosen_styles: StrOrIterable) -> Iterator[Fuss]:
        """Include a list of styles (or just one) into this style tree."""
        for style_uri in always_iterable(chosen_styles):
            yield from self._include_style(style_uri)

    def _include_style(self, style_uri):
        style_uri = self._normalize_style_uri(style_uri)
        style_path, file_contents = self.get_style_path(style_uri)
        if not style_path:
            return

        resolved_path = str(style_path.resolve())
        if resolved_path in self._already_included:
            return
        self._already_included.add(resolved_path)

        read_toml_dict = self._read_toml(file_contents, style_path)

        try:
            display_name = str(style_path.relative_to(self.project.root))
        except ValueError:
            display_name = style_uri

        toml_dict, validation_errors = self._config_validator.validate(read_toml_dict)

        if validation_errors:
            yield Reporter(FileInfo(self.project, display_name)).make_fuss(
                StyleViolations.INVALID_CONFIG, flatten_marshmallow_errors(validation_errors)
            )

        self._blender.add(toml_dict)

        sub_styles: StrOrList = search_dict(NITPICK_STYLES_INCLUDE_JMEX, toml_dict, [])
        if sub_styles:
            yield from self.include_multiple_styles(sub_styles)

    def _read_toml(self, file_contents, style_path):
        toml = TomlDoc(string=file_contents)
        try:
            read_toml_dict = toml.as_object
        # TODO: replace by this error when using tomlkit only in the future:
        #  except TOMLKitError as err:
        except TomlDecodeError as err:
            # If the TOML itself could not be parsed, we can't go on
            raise QuitComplainingError(
                Reporter(FileInfo(self.project, style_path.name)).make_fuss(
                    StyleViolations.INVALID_TOML, exception=pretty_exception(err)
                )
            ) from err
        return read_toml_dict

    def _normalize_style_uri(self, uri):
        is_current_uri_url = is_url(uri)
        if is_current_uri_url:
            self._first_full_path = uri
            return self._append_toml_extension_url(uri)

        uri = self._append_toml_extension(uri)

        if not self._first_full_path:
            self._first_full_path = str(Path(uri).resolve().parent) + "/"
            return uri

        if self._first_full_path and not uri.startswith(DOT_SLASH):
            if os.path.isabs(uri):
                return uri

            return self._join_uri(uri)

        return uri

    def _join_uri(self, uri):
        if is_url(self._first_full_path):
            return urljoin(self._first_full_path, uri)

        return str(Path(self._first_full_path).joinpath(uri))

    def _append_toml_extension_url(self, url):
        scheme, netloc, path, query, fragment = urlsplit(url)
        path = self._append_toml_extension(path)
        return urlunsplit((scheme, netloc, path, query, fragment))

    @staticmethod
    def _append_toml_extension(path):
        if path.endswith(TOML_EXTENSION):
            return path
        return f"{path}{TOML_EXTENSION}"

    def get_style_path(self, style_uri: str) -> Tuple[Optional[Path], str]:
        """Get the style path from the URI. Add the .toml extension if it's missing."""
        clean_style_uri = style_uri.strip()
        return self._style_fetcher_manager.fetch(clean_style_uri)

    def merge_toml_dict(self) -> JsonDict:
        """Merge all included styles into a TOML (actually JSON) dictionary."""
        merged_dict = self._blender.mix()
        # TODO: check if the merged style file is still needed
        merged_style_path: Path = self.cache_dir / MERGED_STYLE_TOML
        toml = TomlDoc(obj=merged_dict)

        attempt = 1
        while attempt < 5:
            try:
                merged_style_path.write_text(toml.reformatted)
                break
            except OSError:
                attempt += 1

        return merged_dict

    @staticmethod
    def file_field_pair(filename: str, base_file_class: Type[NitpickPlugin]) -> Dict[str, fields.Field]:
        """Return a schema field with info from a config file class."""
        unique_filename_with_underscore = slugify(filename, separator="_")

        kwargs = {"data_key": filename}
        if base_file_class.validation_schema:
            file_field = fields.Nested(base_file_class.validation_schema, **kwargs)
        else:
            # For some files (e.g.: TOML/ INI files), there is no strict schema;
            # it can be anything they allow.
            # It's out of Nitpick's scope to validate those files.
            file_field = fields.Dict(fields.String, **kwargs)
        return {unique_filename_with_underscore: file_field}

    @lru_cache()
    def load_fixed_name_plugins(self) -> Plugins:
        """Separate classes with fixed file names from classes with dynamic files names."""
        fixed_name_classes: Plugins = set()
        for plugin_class in self.project.plugin_manager.hook.plugin_class():  # pylint: disable=no-member
            if plugin_class.filename:
                fixed_name_classes.add(plugin_class)
        return fixed_name_classes

    def rebuild_dynamic_schema(self) -> None:
        """Rebuild the dynamic Marshmallow schema when needed, adding new fields that were found on the style."""
        new_files_found: Dict[str, fields.Field] = OrderedDict()

        fixed_name_classes = self.load_fixed_name_plugins()

        for subclass in fixed_name_classes:
            new_files_found.update(self.file_field_pair(subclass.filename, subclass))

        # Only recreate the schema if new fields were found.
        if new_files_found:
            self._dynamic_schema_class = type("DynamicStyleSchema", (self._dynamic_schema_class,), new_files_found)

    def _find_subclasses(self, data, handled_tags, new_files_found):
        for possible_file in data.keys():
            found_subclasses = []
            for file_tag in identify.tags_from_filename(possible_file):
                handler_subclass = handled_tags.get(file_tag)
                if handler_subclass:
                    found_subclasses.append(handler_subclass)

            for found_subclass in found_subclasses:
                new_files_found.update(self.file_field_pair(possible_file, found_subclass))

1	"""Style files."""
2	import os	5✔
3	from collections import OrderedDict	5✔
4	from dataclasses import dataclass	5✔
5	from functools import lru_cache	5✔
6	from pathlib import Path	5✔
7	from typing import Dict, Iterator, Optional, Set, Tuple, Type	5✔
8	from urllib.parse import urljoin, urlsplit, urlunsplit	5✔
9
10	from identify import identify	5✔
11	from loguru import logger	5✔
12	from more_itertools import always_iterable	5✔
13	from slugify import slugify	5✔
14	from toml import TomlDecodeError	5✔
15
16	from nitpick import __version__, fields	5✔
17	from nitpick.constants import (	5✔
18	CACHE_DIR_NAME,
19	DOT_SLASH,
20	MERGED_STYLE_TOML,
21	NITPICK_STYLE_TOML,
22	NITPICK_STYLES_INCLUDE_JMEX,
23	PROJECT_NAME,
24	PROJECT_OWNER,
25	PYPROJECT_TOML,
26	TOML_EXTENSION,
27	)
28	from nitpick.documents import TomlDoc	5✔
29	from nitpick.exceptions import QuitComplainingError, pretty_exception	5✔
30	from nitpick.generic import DictBlender, is_url, search_dict	5✔
31	from nitpick.plugins.base import NitpickPlugin	5✔
32	from nitpick.plugins.info import FileInfo	5✔
33	from nitpick.project import Project, glob_files	5✔
34	from nitpick.schemas import BaseStyleSchema, flatten_marshmallow_errors	5✔
35	from nitpick.style.config import ConfigValidator	5✔
36	from nitpick.style.fetchers import StyleFetcherManager	5✔
37	from nitpick.style.fetchers.github import GitHubURL	5✔
38	from nitpick.typedefs import JsonDict, StrOrIterable, StrOrList, mypy_property	5✔
39	from nitpick.violations import Fuss, Reporter, StyleViolations	5✔
40
41	Plugins = Set[Type[NitpickPlugin]]	5✔
42
43
44	@dataclass(repr=True)	5✔
45	class Style: # pylint: disable=too-many-instance-attributes	3✔
46	"""Include styles recursively from one another."""
47
48	project: Project	5✔
49	offline: bool	5✔
50	cache_option: str	5✔
51
52	def __post_init__(self) -> None:	5✔
53	"""Initialize dependant fields."""
54	self._blender: DictBlender = DictBlender()	5✔
55	self._already_included: Set[str] = set()	5✔
56	self._first_full_path: str = ""	5✔
57	self._dynamic_schema_class: type = BaseStyleSchema	5✔
58	self._style_fetcher_manager = StyleFetcherManager(self.offline, self.cache_dir, self.cache_option)	5✔
59	self._config_validator = ConfigValidator(self.project)	5✔
60	self.rebuild_dynamic_schema()	5✔
61
62	def __hash__(self):	5✔
63	"""Calculate hash on hashable items so lru_cache knows how to cache data from this class."""
64	return hash((self.project, self.offline, self.cache_option))	5✔
65
66	@mypy_property	5✔
67	@lru_cache()	5✔
68	def cache_dir(self) -> Path:	5✔
69	"""Clear the cache directory (on the project root or on the current directory)."""
70	path: Path = self.project.root / CACHE_DIR_NAME / PROJECT_NAME	5✔
71
72	# TODO: check if the merged style file is still needed; if not, this line can be removed
73	path.mkdir(parents=True, exist_ok=True)	5✔
74
75	return path	5✔
76
77	@staticmethod	5✔
78	def get_default_style_url():	3✔
79	"""Return the URL of the default style for the current version."""
80	return GitHubURL(PROJECT_OWNER, PROJECT_NAME, f"v{__version__}", NITPICK_STYLE_TOML, None).long_protocol_url	5✔
81
82	def find_initial_styles(self, configured_styles: StrOrIterable) -> Iterator[Fuss]:	5✔
83	"""Find the initial style(s) and include them."""
84	if configured_styles:	5✔
85	chosen_styles: StrOrIterable = list(configured_styles)	5✔
86	log_message = f"Using styles configured in {PYPROJECT_TOML}"	5✔
87	else:
88	paths = glob_files(self.project.root, [NITPICK_STYLE_TOML])	5✔
89	if paths:	5✔
90	chosen_styles = str(sorted(paths)[0])	5✔
91	log_message = "Using local style found climbing the directory tree"	5✔
92	else:
93	chosen_styles = self.get_default_style_url()	5✔
94	log_message = "Using default remote Nitpick style"	5✔
95	logger.info(f"{log_message}: {chosen_styles}")	5✔
96
97	yield from self.include_multiple_styles(chosen_styles)	5✔
98
99	def include_multiple_styles(self, chosen_styles: StrOrIterable) -> Iterator[Fuss]:	5✔
100	"""Include a list of styles (or just one) into this style tree."""
101	for style_uri in always_iterable(chosen_styles):	5✔
102	yield from self._include_style(style_uri)	5✔
103
104	def _include_style(self, style_uri):	5✔
105	style_uri = self._normalize_style_uri(style_uri)	5✔
106	style_path, file_contents = self.get_style_path(style_uri)	5✔
107	if not style_path:	5✔
108	return	5✔
109
110	resolved_path = str(style_path.resolve())	5✔
111	if resolved_path in self._already_included:	5✔
112	return	5✔
113	self._already_included.add(resolved_path)	5✔
114
115	read_toml_dict = self._read_toml(file_contents, style_path)	5✔
116
117	try:	5✔
118	display_name = str(style_path.relative_to(self.project.root))	5✔
119	except ValueError:	5✔
120	display_name = style_uri	5✔
121
122	toml_dict, validation_errors = self._config_validator.validate(read_toml_dict)	5✔
123
124	if validation_errors:	5✔
125	yield Reporter(FileInfo(self.project, display_name)).make_fuss(	5✔
126	StyleViolations.INVALID_CONFIG, flatten_marshmallow_errors(validation_errors)
127	)
128
129	self._blender.add(toml_dict)	5✔
130
131	sub_styles: StrOrList = search_dict(NITPICK_STYLES_INCLUDE_JMEX, toml_dict, [])	5✔
132	if sub_styles:	5✔
133	yield from self.include_multiple_styles(sub_styles)	5✔
134
135	def _read_toml(self, file_contents, style_path):	5✔
136	toml = TomlDoc(string=file_contents)	5✔
137	try:	5✔
138	read_toml_dict = toml.as_object	5✔
139	# TODO: replace by this error when using tomlkit only in the future:
140	# except TOMLKitError as err:
141	except TomlDecodeError as err:	5✔
142	# If the TOML itself could not be parsed, we can't go on
143	raise QuitComplainingError(	5✔
144	Reporter(FileInfo(self.project, style_path.name)).make_fuss(
145	StyleViolations.INVALID_TOML, exception=pretty_exception(err)
146	)
147	) from err
148	return read_toml_dict	5✔
149
150	def _normalize_style_uri(self, uri):	5✔
151	is_current_uri_url = is_url(uri)	5✔
152	if is_current_uri_url:	5✔
153	self._first_full_path = uri	5✔
154	return self._append_toml_extension_url(uri)	5✔
155
156	uri = self._append_toml_extension(uri)	5✔
157
158	if not self._first_full_path:	5✔
159	self._first_full_path = str(Path(uri).resolve().parent) + "/"	5✔
160	return uri	5✔
161
162	if self._first_full_path and not uri.startswith(DOT_SLASH):	5✔
163	if os.path.isabs(uri):	5✔
164	return uri	5✔
165
166	return self._join_uri(uri)	5✔
167
168	return uri	5✔
169
170	def _join_uri(self, uri):	5✔
171	if is_url(self._first_full_path):	5✔
172	return urljoin(self._first_full_path, uri)	5✔
173
174	return str(Path(self._first_full_path).joinpath(uri))	5✔
175
176	def _append_toml_extension_url(self, url):	5✔
177	scheme, netloc, path, query, fragment = urlsplit(url)	5✔
178	path = self._append_toml_extension(path)	5✔
179	return urlunsplit((scheme, netloc, path, query, fragment))	5✔
180
181	@staticmethod	5✔
182	def _append_toml_extension(path):	3✔
183	if path.endswith(TOML_EXTENSION):	5✔
184	return path	5✔
185	return f"{path}{TOML_EXTENSION}"	5✔
186
187	def get_style_path(self, style_uri: str) -> Tuple[Optional[Path], str]:	5✔
188	"""Get the style path from the URI. Add the .toml extension if it's missing."""
189	clean_style_uri = style_uri.strip()	5✔
190	return self._style_fetcher_manager.fetch(clean_style_uri)	5✔
191
192	def merge_toml_dict(self) -> JsonDict:	5✔
193	"""Merge all included styles into a TOML (actually JSON) dictionary."""
194	merged_dict = self._blender.mix()	5✔
195	# TODO: check if the merged style file is still needed
196	merged_style_path: Path = self.cache_dir / MERGED_STYLE_TOML	5✔
197	toml = TomlDoc(obj=merged_dict)	5✔
198
199	attempt = 1	5✔
200	while attempt < 5:	5!
201	try:	5✔
202	merged_style_path.write_text(toml.reformatted)	5✔
203	break	5✔
204	except OSError:	×
205	attempt += 1	×
206
207	return merged_dict	5✔
208
209	@staticmethod	5✔
210	def file_field_pair(filename: str, base_file_class: Type[NitpickPlugin]) -> Dict[str, fields.Field]:	5✔
211	"""Return a schema field with info from a config file class."""
UNCOV 212	unique_filename_with_underscore = slugify(filename, separator="_")	×
213
UNCOV 214	kwargs = {"data_key": filename}	×
UNCOV 215	if base_file_class.validation_schema:	×
216	file_field = fields.Nested(base_file_class.validation_schema, **kwargs)	×
217	else:
218	# For some files (e.g.: TOML/ INI files), there is no strict schema;
219	# it can be anything they allow.
220	# It's out of Nitpick's scope to validate those files.
UNCOV 221	file_field = fields.Dict(fields.String, **kwargs)	×
UNCOV 222	return {unique_filename_with_underscore: file_field}	×
223
224	@lru_cache()	5✔
225	def load_fixed_name_plugins(self) -> Plugins:	5✔
226	"""Separate classes with fixed file names from classes with dynamic files names."""
227	fixed_name_classes: Plugins = set()	5✔
228	for plugin_class in self.project.plugin_manager.hook.plugin_class(): # pylint: disable=no-member	5✔
229	if plugin_class.filename:	5!
UNCOV 230	fixed_name_classes.add(plugin_class)	×
231	return fixed_name_classes	5✔
232
233	def rebuild_dynamic_schema(self) -> None:	5✔
234	"""Rebuild the dynamic Marshmallow schema when needed, adding new fields that were found on the style."""
235	new_files_found: Dict[str, fields.Field] = OrderedDict()	5✔
236
237	fixed_name_classes = self.load_fixed_name_plugins()	5✔
238
239	for subclass in fixed_name_classes:	5!
UNCOV 240	new_files_found.update(self.file_field_pair(subclass.filename, subclass))	×
241
242	# Only recreate the schema if new fields were found.
243	if new_files_found:	5!
UNCOV 244	self._dynamic_schema_class = type("DynamicStyleSchema", (self._dynamic_schema_class,), new_files_found)	×
245
246	def _find_subclasses(self, data, handled_tags, new_files_found):	5✔
247	for possible_file in data.keys():	×
248	found_subclasses = []	×
249	for file_tag in identify.tags_from_filename(possible_file):	×
250	handler_subclass = handled_tags.get(file_tag)	×
251	if handler_subclass:	×
252	found_subclasses.append(handler_subclass)	×
253
254	for found_subclass in found_subclasses:	×
255	new_files_found.update(self.file_field_pair(possible_file, found_subclass))	×

andreoliwa / nitpick / 1669453638

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