Plugins¶

def predefined_special_config(self) -> SpecialConfig:
    """Create a predefined special configuration for this plugin.

    Each plugin can override this method.
    """
    return SpecialConfig()

@MypyProperty
def nitpick_file_dict(self) -> JsonDict:
    """Nitpick configuration for this file as a TOML dict, taken from the style file."""
    return search_json(self.info.project.nitpick_section, f'files."{self.filename}"', {})

def entry_point(self) -> Iterator[Fuss]:
    """Entry point of the Nitpick plugin."""
    self.post_init()

    should_exist: bool = bool(self.info.project.nitpick_files_section.get(self.filename, True))
    if self.file_path.exists() and not should_exist:
        logger.info(f"{self}: File {self.filename} exists when it should not")
        # Only display this message if the style is valid.
        yield self.reporter.make_fuss(SharedViolations.DELETE_FILE)
        return

    has_config_dict = bool(self.expected_config or self.nitpick_file_dict)
    if not has_config_dict:
        return

    yield from self._enforce_file_configuration()

def post_init(self):  # noqa: B027
    """Hook for plugin initialization after the instance was created.

    The name mimics `__post_init__()` on dataclasses, without the magic double underscores.
    See the [dataclasses documentation](https://docs.python.org/3/library/dataclasses.html#post-init-processing) for more details.
    """

def write_file(
    self,
    file_exists: bool,  # pylint: disable=unused-argument # noqa: ARG002
) -> Fuss | None:
    """Hook to write the new file when autofix mode is on.

    Should be used by inherited classes.
    """
    return None

@abc.abstractmethod
def enforce_rules(self) -> Iterator[Fuss]:
    """Enforce rules for this file. It must be overridden by inherited classes if needed."""

def write_initial_contents(self, doc_class: type[BaseDoc], expected_dict: dict | None = None) -> str:
    """Helper to write initial contents based on a format."""
    if not expected_dict:
        expected_dict = self.expected_config

    formatted_str = doc_class(obj=expected_dict).reformatted
    if self.autofix:
        self.file_path.parent.mkdir(exist_ok=True, parents=True)
        self.file_path.write_text(formatted_str)
    return formatted_str

def post_init(self):
    """Post initialization after the instance was created."""
    self.updater = ConfigUpdater()
    self.comma_separated_values = set(self.nitpick_file_dict.get(COMMA_SEPARATED_VALUES, []))

    if not self.needs_top_section:
        return
    if all(isinstance(v, dict) for v in self.expected_config.values()):
        return

    new_config = {TOP_SECTION: {}}
    for key, value in self.expected_config.items():
        if isinstance(value, dict):
            new_config[key] = value
            continue
        new_config[TOP_SECTION][key] = value
    self.expected_config = new_config

def write_file(self, file_exists: bool) -> Fuss | None:
    """Write the new file."""
    try:
        if self.needs_top_section:
            self.file_path.write_text(self.contents_without_top_section(str(self.updater)))
            return None

        if file_exists:
            self.updater.update_file()
        else:
            self.updater.write(self.file_path.open("w"))
    except ParsingError as err:
        return self.reporter.make_fuss(Violations.PARSING_ERROR, cls=err.__class__.__name__, msg=err)
    return None

@staticmethod
def contents_without_top_section(multiline_text: str) -> str:
    """Remove the temporary top section from multiline text, and keep the newline at the end of the file."""
    return "\n".join(line for line in multiline_text.splitlines() if TOP_SECTION not in line) + "\n"

def get_missing_output(self) -> str:
    """Get a missing output string example from the missing sections in an INI file."""
    missing = self.missing_sections
    if not missing:
        return ""

    parser = ConfigParser()
    for section in sorted(missing, key=lambda s: "0" if s == TOP_SECTION else f"1{s}"):
        expected_config: dict = self.expected_config[section]
        if self.autofix:
            if self.updater.last_block:
                self.updater.last_block.add_after.space(1)
            self.updater.add_section(section)
            self.updater[section].update(expected_config)
            self.dirty = True
        parser[section] = expected_config
    return self.contents_without_top_section(self.get_example_cfg(parser))

def enforce_rules(self) -> Iterator[Fuss]:
    """Enforce rules on missing sections and missing key/value pairs in an INI file."""
    try:
        yield from self._read_file()
    except Error:
        return

    yield from self.enforce_missing_sections()

    csv_sections = {v.split(SECTION_SEPARATOR)[0] for v in self.comma_separated_values}
    missing_csv = csv_sections.difference(self.current_sections)
    if missing_csv:
        yield self.reporter.make_fuss(
            Violations.INVALID_COMMA_SEPARATED_VALUES_SECTION, ", ".join(sorted(missing_csv))
        )
        # Don't continue if the comma-separated values are invalid
        return

    for section in self.expected_sections.intersection(self.current_sections) - self.missing_sections:
        yield from self.enforce_section(section)

def enforce_missing_sections(self) -> Iterator[Fuss]:
    """Enforce missing sections."""
    missing = self.get_missing_output()
    if missing:
        yield self.reporter.make_fuss(Violations.MISSING_SECTIONS, missing, self.autofix)

def enforce_section(self, section: str) -> Iterator[Fuss]:
    """Enforce rules for a section."""
    expected_dict = self.expected_config[section]
    actual_dict = {k: v.value for k, v in self.updater[section].items()}
    # TODO: refactor: add a class Ini(BaseDoc) and move this dictdiffer code there
    for diff_type, key, values in dictdiffer.diff(actual_dict, expected_dict):
        if diff_type == dictdiffer.CHANGE:
            if f"{section}.{key}" in self.comma_separated_values:
                yield from self.enforce_comma_separated_values(section, key, values[0], values[1])
            else:
                yield from self.compare_different_keys(section, key, values[0], values[1])
        elif diff_type == dictdiffer.ADD:
            yield from self.show_missing_keys(section, values)

def enforce_comma_separated_values(self, section, key, raw_actual: Any, raw_expected: Any) -> Iterator[Fuss]:
    """Enforce sections and keys with comma-separated values.

    The values might contain spaces.
    """
    actual_set = {s.strip() for s in raw_actual.split(",")}
    expected_set = {s.strip() for s in raw_expected.split(",")}
    missing = expected_set - actual_set
    if not missing:
        return

    joined_values = ",".join(sorted(missing))
    value_to_append = f",{joined_values}"
    if self.autofix:
        self.updater[section][key].value += value_to_append
        self.dirty = True
    section_header = "" if section == TOP_SECTION else f"[{section}]\n"
    # TODO: test: top section with separated values in https://github.com/andreoliwa/nitpick/issues/271
    yield self.reporter.make_fuss(
        Violations.MISSING_VALUES_IN_LIST,
        f"{section_header}{key} = (...){value_to_append}",
        key=key,
        fixed=self.autofix,
    )

def compare_different_keys(self, section, key, raw_actual: Any, raw_expected: Any) -> Iterator[Fuss]:
    """Compare different keys, with special treatment when they are lists or numeric."""
    if isinstance(raw_actual, (int, float, bool)) or isinstance(raw_expected, (int, float, bool)):
        # A boolean "True" or "true" has the same effect on ConfigParser files.
        actual = str(raw_actual).lower()
        expected = str(raw_expected).lower()
    else:
        actual = raw_actual
        expected = raw_expected
    if actual == expected:
        return

    if self.autofix:
        self.updater[section][key].value = expected
        self.dirty = True
    if section == TOP_SECTION:
        yield self.reporter.make_fuss(
            Violations.TOP_SECTION_HAS_DIFFERENT_VALUE,
            f"{key} = {raw_expected}",
            key=key,
            actual=raw_actual,
            fixed=self.autofix,
        )
    else:
        yield self.reporter.make_fuss(
            Violations.OPTION_HAS_DIFFERENT_VALUE,
            f"[{section}]\n{key} = {raw_expected}",
            section=section,
            key=key,
            actual=raw_actual,
            fixed=self.autofix,
        )

def show_missing_keys(self, section: str, values: list[tuple[str, Any]]) -> Iterator[Fuss]:
    """Show the keys that are not present in a section."""
    parser = ConfigParser()
    missing_dict = dict(values)
    parser[section] = missing_dict
    output = self.get_example_cfg(parser)
    self.add_options_before_space(section, missing_dict)

    if section == TOP_SECTION:
        yield self.reporter.make_fuss(
            Violations.TOP_SECTION_MISSING_OPTION, self.contents_without_top_section(output), self.autofix
        )
    else:
        yield self.reporter.make_fuss(Violations.MISSING_OPTION, output, self.autofix, section=section)

def add_options_before_space(self, section: str, options: dict) -> None:
    """Add new options before a blank line in the end of the section."""
    if not self.autofix:
        return

    section_obj = self.updater[section]

    # Collect all trailing Space blocks
    # We need to collect them first before detaching to avoid NotAttachedError
    # when there are multiple consecutive spaces
    trailing_spaces = []
    for block in reversed(list(section_obj.iter_blocks())):
        if isinstance(block, Space):
            trailing_spaces.append(block)
        else:
            break  # Stop at the first non-Space block

    # Detach all trailing spaces
    space_removed = len(trailing_spaces) > 0
    for space in trailing_spaces:
        space.detach()

    section_obj.update(options)
    self.dirty = True

    # Add back a single space if we removed any
    if space_removed:
        section_obj.last_block.add_after.space(1)

@staticmethod
def get_example_cfg(parser: ConfigParser) -> str:
    """Print an example of a config parser in a string instead of a file."""
    string_stream = StringIO()
    parser.write(string_stream)
    return string_stream.getvalue().strip()

def enforce_rules(self) -> Iterator[Fuss]:
    """Enforce rules for missing keys and JSON content."""
    json_doc = JsonDoc(path=self.file_path)
    blender: JsonDict = json_doc.as_object.copy() if self.autofix else {}

    comparison = Comparison(json_doc, self.expected_dict_from_contains_keys(), self.special_config)()
    if comparison.missing:
        yield from self.report(SharedViolations.MISSING_VALUES, blender, comparison.missing)

    comparison = Comparison(json_doc, self.expected_dict_from_contains_json(), self.special_config)()
    if comparison.has_changes:
        yield from chain(
            self.report(SharedViolations.DIFFERENT_VALUES, blender, comparison.diff),
            self.report(SharedViolations.MISSING_VALUES, blender, comparison.missing),
        )

    if self.autofix and self.dirty and blender:
        self.file_path.write_text(JsonDoc(obj=unflatten_quotes(blender)).reformatted)

def expected_dict_from_contains_keys(self):
    """Expected dict created from "contains_keys" values."""
    return unflatten_quotes(
        dict.fromkeys(set(self.expected_config.get(KEY_CONTAINS_KEYS) or []), VALUE_PLACEHOLDER)
    )

def expected_dict_from_contains_json(self):
    """Expected dict created from "contains_json" values."""
    expected_config = {}
    # TODO: feat: accept key as a jmespath expression, value is valid JSON
    for key, json_string in (self.expected_config.get(KEY_CONTAINS_JSON) or {}).items():
        try:
            expected_config[key] = json.loads(json_string)
        except json.JSONDecodeError as err:  # noqa: PERF203
            # This should not happen, because the style was already validated before.
            # Maybe the NIP??? code was disabled by the user?
            logger.error(f"{err} on {KEY_CONTAINS_JSON} while checking {self.file_path}")
            continue
    return expected_config

def report(self, violation: ViolationEnum, blender: JsonDict, change: BaseDoc | None):
    """Report a violation while optionally modifying the JSON dict."""
    if not change:
        return
    if blender:
        blender.update(flatten_quotes(change.as_object))
        self.dirty = True
    yield self.reporter.make_fuss(violation, change.reformatted, prefix="", fixed=self.autofix)

def enforce_rules(self) -> Iterator[Fuss]:
    """Enforce rules for missing key/value pairs in the TOML file."""
    toml_doc = TomlDoc(path=self.file_path)
    comparison = Comparison(toml_doc, self.expected_config, self.special_config)()
    if not comparison.has_changes:
        return

    document = parse(toml_doc.as_string) if self.autofix else None
    yield from chain(
        self.report(SharedViolations.DIFFERENT_VALUES, document, cast("TomlDoc", comparison.diff)),
        self.report(
            SharedViolations.MISSING_VALUES,
            document,
            cast("TomlDoc", comparison.missing),
            cast("TomlDoc", comparison.replace),
        ),
    )
    if self.autofix and self.dirty:
        self.file_path.write_text(dumps(document))

def report(
    self,
    violation: ViolationEnum,
    document: TOMLDocument | None,
    change: TomlDoc | None,
    replacement: TomlDoc | None = None,
):
    """Report a violation while optionally modifying the TOML document."""
    if not (change or replacement):
        return
    if self.autofix:
        real_change = cast("TomlDoc", replacement or change)
        traverse_toml_tree(document, real_change.as_object)
        self.dirty = True

    to_display = cast("TomlDoc", change or replacement)
    yield self.reporter.make_fuss(violation, to_display.reformatted.strip(), prefix="", fixed=self.autofix)

def predefined_special_config(self) -> SpecialConfig:
    """Predefined special config, with list keys for .pre-commit-config.yaml and GitHub Workflow files."""
    spc = SpecialConfig()
    # pylint: disable=assigning-non-slot
    if self.filename == PRE_COMMIT_CONFIG_YAML:
        spc.list_keys.from_plugin = {"repos": "hooks.id"}
    elif self.filename.startswith(".github/workflows"):
        spc.list_keys.from_plugin = {"jobs.*.steps": "name"}
    return spc

def enforce_rules(self) -> Iterator[Fuss]:
    """Enforce rules for missing data in the YAML file."""
    if KEY_CONTAINS in self.expected_config:
        # If the expected configuration has this key, it means that this config is being handled by TextPlugin.
        # TODO: fix: allow a YAML file with a "contains" key on its root (how?)
        return

    yaml_doc = YamlDoc(path=self.file_path)
    comparison = Comparison(yaml_doc, self._remove_yaml_subkey(self.expected_config), self.special_config)()
    if not comparison.has_changes:
        return

    yield from chain(
        self.report(SharedViolations.DIFFERENT_VALUES, yaml_doc.as_object, cast("YamlDoc", comparison.diff)),
        self.report(
            SharedViolations.MISSING_VALUES,
            yaml_doc.as_object,
            cast("YamlDoc", comparison.missing),
            cast("YamlDoc", comparison.replace),
        ),
    )
    if self.autofix and self.dirty:
        yaml_doc.updater.dump(yaml_doc.as_object, self.file_path)

def report(
    self,
    violation: ViolationEnum,
    yaml_object: YamlObject,
    change: YamlDoc | None,
    replacement: YamlDoc | None = None,
):
    """Report a violation while optionally modifying the YAML document."""
    if not (change or replacement):
        return
    if self.autofix:
        real_change = cast("YamlDoc", replacement or change)
        traverse_yaml_tree(yaml_object, real_change.as_object)
        self.dirty = True

    to_display = cast("YamlDoc", change or replacement)
    yield self.reporter.make_fuss(violation, to_display.reformatted.strip(), prefix="", fixed=self.autofix)

def enforce_rules(self) -> Iterator[Fuss]:
    """Enforce rules for missing lines."""
    expected = OrderedSet(self._expected_lines())
    actual = OrderedSet(self.file_path.read_text().split("\n"))
    missing = expected - actual
    if missing:
        yield self.reporter.make_fuss(Violations.MISSING_LINES, "\n".join(sorted(missing)))