1700668254

Build Type

push

github

Committed by W. Augusto Andreoli

Commit Message

fix: pin attrs to >=20.1.0, change imports

Run Details

552 of 586 branches covered (94.2%)

Branch coverage included in aggregate %.

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

15 existing lines in 2 files now uncovered.

1987 of 2031 relevant lines covered (97.83%)

3.9 hits per line

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

84.79

/src/nitpick/style/core.py

"""Style files."""
import os
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

import dpath.util
from flatten_dict import flatten, unflatten
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.blender import SEPARATOR_FLATTEN, TomlDoc, custom_reducer, custom_splitter, search_json
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.exceptions import QuitComplainingError, pretty_exception
from nitpick.generic import is_url
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._merged_styles: JsonDict = {}
        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)
            )

        dpath.util.merge(self._merged_styles, flatten(toml_dict, custom_reducer(SEPARATOR_FLATTEN)))

        sub_styles: StrOrList = search_json(toml_dict, NITPICK_STYLES_INCLUDE_JMEX, [])
        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 = unflatten(self._merged_styles, custom_splitter(SEPARATOR_FLATTEN))
        # 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] = {}

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

andreoliwa / nitpick / 1700668254

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