1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
|
# 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 json
from pathlib import Path, PurePosixPath
from pylint.reporters.json_reporter import OldJsonExport
from pylint.testutils._primer.primer_command import (
PackageData,
PackageMessages,
PrimerCommand,
)
MAX_GITHUB_COMMENT_LENGTH = 65536
class CompareCommand(PrimerCommand):
def run(self) -> None:
main_data = self._load_json(self.config.base_file)
pr_data = self._load_json(self.config.new_file)
missing_messages_data, new_messages_data = self._cross_reference(
main_data, pr_data
)
comment = self._create_comment(missing_messages_data, new_messages_data)
with open(self.primer_directory / "comment.txt", "w", encoding="utf-8") as f:
f.write(comment)
@staticmethod
def _cross_reference(
main_data: PackageMessages, pr_data: PackageMessages
) -> tuple[PackageMessages, PackageMessages]:
missing_messages_data: PackageMessages = {}
for package, data in main_data.items():
package_missing_messages: list[OldJsonExport] = []
for message in data["messages"]:
try:
pr_data[package]["messages"].remove(message)
except ValueError:
package_missing_messages.append(message)
missing_messages_data[package] = PackageData(
commit=pr_data[package]["commit"], messages=package_missing_messages
)
return missing_messages_data, pr_data
@staticmethod
def _load_json(file_path: Path | str) -> PackageMessages:
with open(file_path, encoding="utf-8") as f:
result: PackageMessages = json.load(f)
return result
def _create_comment(
self, all_missing_messages: PackageMessages, all_new_messages: PackageMessages
) -> str:
comment = ""
for package, missing_messages in all_missing_messages.items():
if len(comment) >= MAX_GITHUB_COMMENT_LENGTH:
break
new_messages = all_new_messages[package]
if not missing_messages["messages"] and not new_messages["messages"]:
continue
comment += self._create_comment_for_package(
package, new_messages, missing_messages
)
comment = (
f"🤖 **Effect of this PR on checked open source code:** 🤖\n\n{comment}"
if comment
else (
"🤖 According to the primer, this change has **no effect** on the"
" checked open source code. 🤖🎉\n\n"
)
)
return self._truncate_comment(comment)
def _create_comment_for_package(
self, package: str, new_messages: PackageData, missing_messages: PackageData
) -> str:
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["messages"]:
print("Now emitted:")
for message in new_messages["messages"]:
filepath = str(
PurePosixPath(message["path"]).relative_to(
self.packages[package].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"{self.packages[package].url}/blob/{new_messages['commit']}/{filepath}#L{message['line']}\n"
)
print(message)
count += 1
if astroid_errors:
comment += (
f'{astroid_errors} "astroid error(s)" were found. '
"Please open the GitHub Actions log to see what failed or crashed.\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["messages"]:
comment += "The following messages are no longer emitted:\n\n<details>\n\n"
print("No longer emitted:")
for message in missing_messages["messages"]:
comment += f"{count}) {message['symbol']}:\n*{message['message']}*\n"
filepath = str(
PurePosixPath(message["path"]).relative_to(
self.packages[package].clone_directory
)
)
assert not self.packages[package].url.endswith(
".git"
), "You don't need the .git at the end of the github url."
comment += (
f"{self.packages[package].url}"
f"/blob/{new_messages['commit']}/{filepath}#L{message['line']}\n"
)
count += 1
print(message)
if missing_messages:
comment += "\n</details>\n\n"
return comment
def _truncate_comment(self, comment: str) -> str:
"""GitHub allows only a set number of characters in a 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
return comment
|
