Refactor the primer command to be able to make the comment better later on

.coveragerc

@@ -24,5 +24,5 @@ exclude_lines =
     if TYPE_CHECKING:
     @overload
 
-    # Abstract methods are not exectued during pytest runs
+    # Abstract methods are not executed during pytest runs
     raise NotImplementedError()

.pyenchant_pylint_custom_dict.txt

@@ -154,6 +154,7 @@ iterables
 iteritems
 jn
 jpg
+json
 jx
 jython
 # class is a reserved word

pylint/testutils/_primer/primer.py

@@ -6,25 +6,13 @@
 
 import argparse
 import json
-import sys
-import warnings
-from io import StringIO
-from itertools import chain
 from pathlib import Path
-from typing import Dict, List, Union
 
-import git
-
-from pylint.lint import Run
-from pylint.reporters import JSONReporter
 from pylint.testutils._primer import PackageToLint
-
-MAX_GITHUB_COMMENT_LENGTH = 65536
-
-PackageMessages = Dict[str, List[Dict[str, Union[str, int]]]]
-
-GITHUB_CRASH_TEMPLATE_LOCATION = "/home/runner/.cache"
-CRASH_TEMPLATE_INTRO = "There is a pre-filled template"
+from pylint.testutils._primer.primer_command import PrimerCommand
+from pylint.testutils._primer.primer_compare_command import CompareCommand
+from pylint.testutils._primer.primer_prepare_command import PrepareCommand
+from pylint.testutils._primer.primer_run_command import RunCommand
 
 
 class Primer:
@@ -90,212 +78,16 @@ def __init__(self, primer_directory: Path, json_path: Path) -> None:
         self.packages = self._get_packages_to_lint_from_json(json_path)
         """All packages to prime."""
 
-    def run(self) -> None:
         if self.config.command == "prepare":
-            self._handle_prepare_command()
+            command_class: type[PrimerCommand] = PrepareCommand
         if self.config.command == "run":
-            self._handle_run_command()
+            command_class = RunCommand
         if self.config.command == "compare":
-            self._handle_compare_command()
-
-    def _handle_prepare_command(self) -> None:
-        commit_string = ""
-        if self.config.clone:
-            for package, data in self.packages.items():
-                local_commit = data.lazy_clone()
-                print(f"Cloned '{package}' at commit '{local_commit}'.")
-                commit_string += local_commit + "_"
-        elif self.config.check:
-            for package, data in self.packages.items():
-                local_commit = git.Repo(data.clone_directory).head.object.hexsha
-                print(f"Found '{package}' at commit '{local_commit}'.")
-                commit_string += local_commit + "_"
-        elif self.config.make_commit_string:
-            for package, data in self.packages.items():
-                remote_sha1_commit = (
-                    git.cmd.Git().ls_remote(data.url, data.branch).split("\t")[0]
-                )
-                print(f"'{package}' remote is at commit '{remote_sha1_commit}'.")
-                commit_string += remote_sha1_commit + "_"
-        elif self.config.read_commit_string:
-            with open(
-                self.primer_directory / "commit_string.txt", encoding="utf-8"
-            ) as f:
-                print(f.read())
-
-        if commit_string:
-            with open(
-                self.primer_directory / "commit_string.txt", "w", encoding="utf-8"
-            ) as f:
-                f.write(commit_string)
-
-    def _handle_run_command(self) -> None:
-        packages: PackageMessages = {}
-
-        for package, data in self.packages.items():
-            output = self._lint_package(data)
-            packages[package] = output
-            print(f"Successfully primed {package}.")
-
-        astroid_errors = []
-        other_fatal_msgs = []
-        for msg in chain.from_iterable(packages.values()):
-            if msg["type"] == "fatal":
-                # Remove the crash template location if we're running on GitHub.
-                # We were falsely getting "new" errors when the timestamp changed.
-                assert isinstance(msg["message"], str)
-                if GITHUB_CRASH_TEMPLATE_LOCATION in msg["message"]:
-                    msg["message"] = msg["message"].rsplit(CRASH_TEMPLATE_INTRO)[0]
-                if msg["symbol"] == "astroid-error":
-                    astroid_errors.append(msg)
-                else:
-                    other_fatal_msgs.append(msg)
-
-        with open(
-            self.primer_directory
-            / f"output_{'.'.join(str(i) for i in sys.version_info[:3])}_{self.config.type}.txt",
-            "w",
-            encoding="utf-8",
-        ) as f:
-            json.dump(packages, f)
-
-        # Fail loudly (and fail CI pipelines) if any fatal errors are found,
-        # unless they are astroid-errors, in which case just warn.
-        # This is to avoid introducing a dependency on bleeding-edge astroid
-        # for pylint CI pipelines generally, even though we want to use astroid main
-        # for the purpose of diffing emitted messages and generating PR comments.
-        if astroid_errors:
-            warnings.warn(f"Fatal errors traced to astroid:  {astroid_errors}")
-        assert not other_fatal_msgs, other_fatal_msgs
-
-    def _handle_compare_command(self) -> None:
-        with open(self.config.base_file, encoding="utf-8") as f:
-            main_dict: PackageMessages = json.load(f)
-        with open(self.config.new_file, encoding="utf-8") as f:
-            new_dict: PackageMessages = json.load(f)
+            command_class = CompareCommand
+        self.command = command_class(self.primer_directory, self.packages, self.config)
 
-        final_main_dict: PackageMessages = {}
-        for package, messages in main_dict.items():
-            final_main_dict[package] = []
-            for message in messages:
-                try:
-                    new_dict[package].remove(message)
-                except ValueError:
-                    final_main_dict[package].append(message)
-
-        self._create_comment(final_main_dict, new_dict)
-
-    def _create_comment(
-        self, all_missing_messages: PackageMessages, all_new_messages: PackageMessages
-    ) -> None:
-        comment = ""
-        for package, missing_messages in all_missing_messages.items():
-            if len(comment) >= MAX_GITHUB_COMMENT_LENGTH:
-                break
-
-            new_messages = all_new_messages[package]
-            package_data = self.packages[package]
-
-            if not missing_messages and not new_messages:
-                continue
-
-            comment += f"\n\n**Effect on [{package}]({self.packages[package].url}):**\n"
-
-            # Create comment for new messages
-            count = 1
-            astroid_errors = 0
-            new_non_astroid_messages = ""
-            if new_messages:
-                print("Now emitted:")
-            for message in new_messages:
-                filepath = str(message["path"]).replace(
-                    str(package_data.clone_directory), ""
-                )
-                # Existing astroid errors may still show up as "new" because the timestamp
-                # in the message is slightly different.
-                if message["symbol"] == "astroid-error":
-                    astroid_errors += 1
-                else:
-                    new_non_astroid_messages += (
-                        f"{count}) {message['symbol']}:\n*{message['message']}*\n"
-                        f"{package_data.url}/blob/{package_data.branch}{filepath}#L{message['line']}\n"
-                    )
-                    print(message)
-                    count += 1
-
-            if astroid_errors:
-                comment += (
-                    f"{astroid_errors} error(s) were found stemming from the `astroid` library. "
-                    "This is unlikely to have been caused by your changes. "
-                    "A GitHub Actions warning links directly to the crash report template. "
-                    "Please open an issue against `astroid` if one does not exist already. \n\n"
-                )
-            if new_non_astroid_messages:
-                comment += (
-                    "The following messages are now emitted:\n\n<details>\n\n"
-                    + new_non_astroid_messages
-                    + "\n</details>\n\n"
-                )
-
-            # Create comment for missing messages
-            count = 1
-            if missing_messages:
-                comment += (
-                    "The following messages are no longer emitted:\n\n<details>\n\n"
-                )
-                print("No longer emitted:")
-            for message in missing_messages:
-                comment += f"{count}) {message['symbol']}:\n*{message['message']}*\n"
-                filepath = str(message["path"]).replace(
-                    str(package_data.clone_directory), ""
-                )
-                assert not package_data.url.endswith(
-                    ".git"
-                ), "You don't need the .git at the end of the github url."
-                comment += f"{package_data.url}/blob/{package_data.branch}{filepath}#L{message['line']}\n"
-                count += 1
-                print(message)
-            if missing_messages:
-                comment += "\n</details>\n\n"
-
-        if comment == "":
-            comment = (
-                "🤖 According to the primer, this change has **no effect** on the"
-                " checked open source code. 🤖🎉\n\n"
-            )
-        else:
-            comment = (
-                f"🤖 **Effect of this PR on checked open source code:** 🤖\n\n{comment}"
-            )
-        hash_information = (
-            f"*This comment was generated for commit {self.config.commit}*"
-        )
-        if len(comment) + len(hash_information) >= MAX_GITHUB_COMMENT_LENGTH:
-            truncation_information = (
-                f"*This comment was truncated because GitHub allows only"
-                f" {MAX_GITHUB_COMMENT_LENGTH} characters in a comment.*"
-            )
-            max_len = (
-                MAX_GITHUB_COMMENT_LENGTH
-                - len(hash_information)
-                - len(truncation_information)
-            )
-            comment = f"{comment[:max_len - 10]}...\n\n{truncation_information}\n\n"
-        comment += hash_information
-        with open(self.primer_directory / "comment.txt", "w", encoding="utf-8") as f:
-            f.write(comment)
-
-    def _lint_package(self, data: PackageToLint) -> list[dict[str, str | int]]:
-        # We want to test all the code we can
-        enables = ["--enable-all-extensions", "--enable=all"]
-        # Duplicate code takes too long and is relatively safe
-        # TODO: Find a way to allow cyclic-import and compare output correctly
-        disables = ["--disable=duplicate-code,cyclic-import"]
-        arguments = data.pylint_args + enables + disables
-        output = StringIO()
-        reporter = JSONReporter(output)
-        Run(arguments, reporter=reporter, exit=False)
-        return json.loads(output.getvalue())
+    def run(self) -> None:
+        self.command.run()
 
     @staticmethod
     def _get_packages_to_lint_from_json(json_path: Path) -> dict[str, PackageToLint]:

pylint/testutils/_primer/primer_command.py

@@ -0,0 +1,32 @@
+# Licensed under the GPL: https://www.gnu.org/licenses/old-licenses/gpl-2.0.html
+# For details: https://github.com/PyCQA/pylint/blob/main/LICENSE
+# Copyright (c) https://github.com/PyCQA/pylint/blob/main/CONTRIBUTORS.txt
+
+from __future__ import annotations
+
+import abc
+import argparse
+from pathlib import Path
+from typing import Dict, List, Union
+
+from pylint.testutils._primer import PackageToLint
+
+PackageMessages = Dict[str, List[Dict[str, Union[str, int]]]]
+
+
+class PrimerCommand:
+    """Generic primer action with required arguments."""
+
+    def __init__(
+        self,
+        primer_directory: Path,
+        packages: dict[str, PackageToLint],
+        config: argparse.Namespace,
+    ) -> None:
+        self.primer_directory = primer_directory
+        self.packages = packages
+        self.config = config
+
+    @abc.abstractmethod
+    def run(self) -> None:
+        pass