#4719

Committed 29 Aug 2023 01:05PM CUT coverage: 22.018% (-0.005%) from 22.023%

Build # #4719

Build Type

push

coveralls-python

Committed by

suhaibmujahid

Commit Message

[topcrash_highlight] Don't run the rule when the versions are not consistent

Run Details

716 of 3576 branches covered (0.0%)

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

1925 of 8743 relevant lines covered (22.02%)

0.22 hits per line

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

0.0

/bugbot/rules/topcrash_highlight.py

# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this file,
# You can obtain one at http://mozilla.org/MPL/2.0/.

import itertools
import re
from collections import defaultdict
from typing import Dict, Iterable

from libmozdata.bugzilla import Bugzilla
from libmozdata.connection import Connection

from bugbot import utils
from bugbot.bzcleaner import BzCleaner
from bugbot.constants import LOW_SEVERITY
from bugbot.history import History
from bugbot.topcrash import TOP_CRASH_IDENTIFICATION_CRITERIA, Topcrash

MAX_SIGNATURES_PER_QUERY = 30


class TopcrashHighlight(BzCleaner):
    def __init__(self):
        super().__init__()
        if not self.init_versions():
            return

        self.topcrashes = None
        self.topcrashes_restrictive = None

    def description(self):
        return "Highlighted topcrash bugs"

    def columns(self):
        return ["id", "summary", "severity", "actions"]

    def handle_bug(self, bug, data):
        bugid = str(bug["id"])
        if bugid in data:
            return

        topcrash_signatures = self._get_topcrash_signatures(bug)
        keywords_to_add = self._get_keywords_to_be_added(bug, topcrash_signatures)
        is_keywords_removed = utils.is_keywords_removed_by_bugbot(bug, keywords_to_add)

        actions = []
        autofix = {
            "comment": {
                "body": "",
            },
        }

        if keywords_to_add and (
            not is_keywords_removed
            or self._is_matching_restrictive_criteria(topcrash_signatures)
        ):
            autofix["keywords"] = {"add": keywords_to_add}
            autofix["comment"]["body"] += self.get_matching_criteria_comment(
                topcrash_signatures, is_keywords_removed
            )
            actions.extend(f"Add {keyword} keyword" for keyword in keywords_to_add)

        ni_person = utils.get_mail_to_ni(bug)
        if (
            ni_person
            and bug["severity"] in LOW_SEVERITY
            and "meta" not in bug["keywords"]
            and not self._has_severity_increase_comment(bug)
            and not self._was_severity_set_after_topcrash(bug)
        ):
            autofix["flags"] = [
                {
                    "name": "needinfo",
                    "requestee": ni_person["mail"],
                    "status": "?",
                    "new": "true",
                }
            ]
            autofix["comment"]["body"] += (
                f'\n:{ ni_person["nickname"] }, '
                "could you consider increasing the severity of this top-crash bug?"
            )
            actions.append("Suggest increasing the severity")

        if not actions:
            return

        autofix["comment"]["body"] += f"\n\n{ self.get_documentation() }\n"
        self.autofix_changes[bugid] = autofix

        data[bugid] = {
            "severity": bug["severity"],
            "actions": actions,
        }

        return bug

    def get_matching_criteria_comment(
        self,
        signatures: list,
        is_keywords_removed: bool,
    ) -> str:
        """Generate a comment with the matching criteria for the given signatures.

        Args:
            signatures: The list of signatures to generate the comment for.
            is_keywords_removed: Whether the topcrash keywords was removed earlier.

        Returns:
            The comment for the matching criteria.
        """
        matching_criteria: Dict[str, bool] = defaultdict(bool)
        for signature in signatures:
            for criterion in self.topcrashes[signature]:
                matching_criteria[criterion["criterion_name"]] |= criterion[
                    "is_startup"
                ]

        introduction = (
            (
                "Sorry for removing the keyword earlier but there is a recent "
                "change in the ranking, so the bug is again linked to "
                if is_keywords_removed
                else "The bug is linked to "
            ),
            (
                "a topcrash signature, which matches "
                if len(signatures) == 1
                else "topcrash signatures, which match "
            ),
            "the following [",
            "criterion" if len(matching_criteria) == 1 else "criteria",
            "](https://wiki.mozilla.org/CrashKill/Topcrash):\n",
        )

        criteria = (
            " ".join(("-", criterion_name, "(startup)\n" if is_startup else "\n"))
            for criterion_name, is_startup in matching_criteria.items()
        )

        return "".join(itertools.chain(introduction, criteria))

    def _has_severity_increase_comment(self, bug):
        return any(
            "could you consider increasing the severity of this top-crash bug?"
            in comment["raw_text"]
            for comment in reversed(bug["comments"])
            if comment["creator"] == History.BOT
        )

    def _was_severity_set_after_topcrash(self, bug: dict) -> bool:
        """Check if the severity was set after the topcrash keyword was added.

        Note: this will not catch cases where the topcrash keyword was added
        added when the bug was created and the severity was set later.

        Args:
            bug: The bug to check.

        Returns:
            True if the severity was set after the topcrash keyword was added.
        """
        has_topcrash_added = False
        for history in bug["history"]:
            for change in history["changes"]:
                if not has_topcrash_added and change["field_name"] == "keywords":
                    has_topcrash_added = "topcrash" in change["added"]

                elif has_topcrash_added and change["field_name"] == "severity":
                    return True

        return False

    def _get_topcrash_signatures(self, bug: dict) -> list:
        topcrash_signatures = [
            signature
            for signature in utils.get_signatures(bug["cf_crash_signature"])
            if signature in self.topcrashes
        ]

        if not topcrash_signatures:
            raise Exception("The bug should have a topcrash signature.")

        return topcrash_signatures

    def _get_keywords_to_be_added(self, bug: dict, topcrash_signatures: list) -> list:
        existing_keywords = {
            keyword
            for keyword in ("topcrash", "topcrash-startup")
            if keyword in bug["keywords"]
        }

        if any(
            # Is it a startup topcrash bug?
            any(criterion["is_startup"] for criterion in self.topcrashes[signature])
            for signature in topcrash_signatures
        ):
            keywords_to_add = {"topcrash", "topcrash-startup"}

        else:
            keywords_to_add = {"topcrash"}

        return list(keywords_to_add - existing_keywords)

    def get_bugs(self, date="today", bug_ids=[], chunk_size=None):
        self.query_url = None
        timeout = self.get_config("bz_query_timeout")
        bugs = self.get_data()
        params_list = self.get_bz_params_list(date)

        searches = [
            Bugzilla(
                params,
                bughandler=self.bughandler,
                bugdata=bugs,
                timeout=timeout,
            ).get_data()
            for params in params_list
        ]

        for search in searches:
            search.wait()

        return bugs

    def _is_matching_restrictive_criteria(self, signatures: Iterable) -> bool:
        topcrashes = self._get_restrictive_topcrash_signatures()
        return any(signature in topcrashes for signature in signatures)

    def _get_restrictive_topcrash_signatures(self) -> dict:
        if self.topcrashes_restrictive is None:
            restrictive_criteria = []
            for criterion in TOP_CRASH_IDENTIFICATION_CRITERIA:
                restrictive_criterion = {
                    **criterion,
                    "tc_limit": criterion["tc_limit"] // 2,
                }

                if "tc_limit_startup" in criterion:
                    restrictive_criterion["tc_limit_startup"] //= 2

                restrictive_criteria.append(restrictive_criterion)

            self.topcrashes_restrictive = Topcrash(
                criteria=restrictive_criteria
            ).get_signatures()

        return self.topcrashes_restrictive

    def get_bz_params_list(self, date):
        self.topcrashes = Topcrash(date).get_signatures()

        fields = [
            "triage_owner",
            "assigned_to",
            "severity",
            "keywords",
            "cf_crash_signature",
            "history",
            "comments.creator",
            "comments.raw_text",
        ]
        params_base = {
            "include_fields": fields,
            "resolution": "---",
        }
        self.amend_bzparams(params_base, [])

        params_list = []
        for signatures in Connection.chunks(
            list(self.topcrashes.keys()),
            MAX_SIGNATURES_PER_QUERY,
        ):
            params = params_base.copy()
            n = int(utils.get_last_field_num(params))
            params[f"f{n}"] = "OP"
            params[f"j{n}"] = "OR"
            for signature in signatures:
                n += 1
                params[f"f{n}"] = "cf_crash_signature"
                params[f"o{n}"] = "regexp"
                # Using `(@ |@)` instead of `@ ?` and ( \]|\]) instead of ` ?]`
                # is a workaround. Strangely `?` stays with the encoded form (%3F)
                # in Bugzilla query.
                # params[f"v{n}"] = f"\[@ ?{re.escape(signature)} ?\]"
                params[f"v{n}"] = rf"\[(@ |@){re.escape(signature)}( \]|\])"
            params[f"f{n+1}"] = "CP"
            params_list.append(params)

        return params_list


if __name__ == "__main__":
    TopcrashHighlight().run()

1	# This Source Code Form is subject to the terms of the Mozilla Public
2	# License, v. 2.0. If a copy of the MPL was not distributed with this file,
3	# You can obtain one at http://mozilla.org/MPL/2.0/.
4
5	import itertools	×
6	import re	×
7	from collections import defaultdict	×
8	from typing import Dict, Iterable	×
9
10	from libmozdata.bugzilla import Bugzilla	×
11	from libmozdata.connection import Connection	×
12
13	from bugbot import utils	×
14	from bugbot.bzcleaner import BzCleaner	×
15	from bugbot.constants import LOW_SEVERITY	×
16	from bugbot.history import History	×
17	from bugbot.topcrash import TOP_CRASH_IDENTIFICATION_CRITERIA, Topcrash	×
18
19	MAX_SIGNATURES_PER_QUERY = 30	×
20
21
22	class TopcrashHighlight(BzCleaner):	×
23	def __init__(self):	×
24	super().__init__()	×
25	if not self.init_versions():	×
26	return	×
27
28	self.topcrashes = None	×
29	self.topcrashes_restrictive = None	×
30
31	def description(self):	×
32	return "Highlighted topcrash bugs"	×
33
34	def columns(self):	×
35	return ["id", "summary", "severity", "actions"]	×
36
37	def handle_bug(self, bug, data):	×
38	bugid = str(bug["id"])	×
39	if bugid in data:	×
40	return	×
41
42	topcrash_signatures = self._get_topcrash_signatures(bug)	×
43	keywords_to_add = self._get_keywords_to_be_added(bug, topcrash_signatures)	×
44	is_keywords_removed = utils.is_keywords_removed_by_bugbot(bug, keywords_to_add)	×
45
46	actions = []	×
47	autofix = {	×
48	"comment": {
49	"body": "",
50	},
51	}
52
53	if keywords_to_add and (	×
54	not is_keywords_removed
55	or self._is_matching_restrictive_criteria(topcrash_signatures)
56	):
57	autofix["keywords"] = {"add": keywords_to_add}	×
58	autofix["comment"]["body"] += self.get_matching_criteria_comment(	×
59	topcrash_signatures, is_keywords_removed
60	)
61	actions.extend(f"Add {keyword} keyword" for keyword in keywords_to_add)	×
62
63	ni_person = utils.get_mail_to_ni(bug)	×
64	if (	×
65	ni_person
66	and bug["severity"] in LOW_SEVERITY
67	and "meta" not in bug["keywords"]
68	and not self._has_severity_increase_comment(bug)
69	and not self._was_severity_set_after_topcrash(bug)
70	):
71	autofix["flags"] = [	×
72	{
73	"name": "needinfo",
74	"requestee": ni_person["mail"],
75	"status": "?",
76	"new": "true",
77	}
78	]
79	autofix["comment"]["body"] += (	×
80	f'\n:{ ni_person["nickname"] }, '
81	"could you consider increasing the severity of this top-crash bug?"
82	)
83	actions.append("Suggest increasing the severity")	×
84
85	if not actions:	×
86	return	×
87
88	autofix["comment"]["body"] += f"\n\n{ self.get_documentation() }\n"	×
89	self.autofix_changes[bugid] = autofix	×
90
91	data[bugid] = {	×
92	"severity": bug["severity"],
93	"actions": actions,
94	}
95
96	return bug	×
97
98	def get_matching_criteria_comment(	×
99	self,
100	signatures: list,
101	is_keywords_removed: bool,
102	) -> str:
103	"""Generate a comment with the matching criteria for the given signatures.
104
105	Args:
106	signatures: The list of signatures to generate the comment for.
107	is_keywords_removed: Whether the topcrash keywords was removed earlier.
108
109	Returns:
110	The comment for the matching criteria.
111	"""
112	matching_criteria: Dict[str, bool] = defaultdict(bool)	×
113	for signature in signatures:	×
114	for criterion in self.topcrashes[signature]:	×
115	matching_criteria[criterion["criterion_name"]] \|= criterion[	×
116	"is_startup"
117	]
118
119	introduction = (	×
120	(
121	"Sorry for removing the keyword earlier but there is a recent "
122	"change in the ranking, so the bug is again linked to "
123	if is_keywords_removed
124	else "The bug is linked to "
125	),
126	(
127	"a topcrash signature, which matches "
128	if len(signatures) == 1
129	else "topcrash signatures, which match "
130	),
131	"the following [",
132	"criterion" if len(matching_criteria) == 1 else "criteria",
133	"](https://wiki.mozilla.org/CrashKill/Topcrash):\n",
134	)
135
136	criteria = (	×
137	" ".join(("-", criterion_name, "(startup)\n" if is_startup else "\n"))
138	for criterion_name, is_startup in matching_criteria.items()
139	)
140
141	return "".join(itertools.chain(introduction, criteria))	×
142
143	def _has_severity_increase_comment(self, bug):	×
144	return any(	×
145	"could you consider increasing the severity of this top-crash bug?"
146	in comment["raw_text"]
147	for comment in reversed(bug["comments"])
148	if comment["creator"] == History.BOT
149	)
150
151	def _was_severity_set_after_topcrash(self, bug: dict) -> bool:	×
152	"""Check if the severity was set after the topcrash keyword was added.
153
154	Note: this will not catch cases where the topcrash keyword was added
155	added when the bug was created and the severity was set later.
156
157	Args:
158	bug: The bug to check.
159
160	Returns:
161	True if the severity was set after the topcrash keyword was added.
162	"""
163	has_topcrash_added = False	×
164	for history in bug["history"]:	×
165	for change in history["changes"]:	×
166	if not has_topcrash_added and change["field_name"] == "keywords":	×
167	has_topcrash_added = "topcrash" in change["added"]	×
168
169	elif has_topcrash_added and change["field_name"] == "severity":	×
170	return True	×
171
172	return False	×
173
174	def _get_topcrash_signatures(self, bug: dict) -> list:	×
175	topcrash_signatures = [	×
176	signature
177	for signature in utils.get_signatures(bug["cf_crash_signature"])
178	if signature in self.topcrashes
179	]
180
181	if not topcrash_signatures:	×
182	raise Exception("The bug should have a topcrash signature.")	×
183
184	return topcrash_signatures	×
185
186	def _get_keywords_to_be_added(self, bug: dict, topcrash_signatures: list) -> list:	×
187	existing_keywords = {	×
188	keyword
189	for keyword in ("topcrash", "topcrash-startup")
190	if keyword in bug["keywords"]
191	}
192
193	if any(	×
194	# Is it a startup topcrash bug?
195	any(criterion["is_startup"] for criterion in self.topcrashes[signature])
196	for signature in topcrash_signatures
197	):
198	keywords_to_add = {"topcrash", "topcrash-startup"}	×
199
200	else:
201	keywords_to_add = {"topcrash"}	×
202
203	return list(keywords_to_add - existing_keywords)	×
204
205	def get_bugs(self, date="today", bug_ids=[], chunk_size=None):	×
206	self.query_url = None	×
207	timeout = self.get_config("bz_query_timeout")	×
208	bugs = self.get_data()	×
209	params_list = self.get_bz_params_list(date)	×
210
211	searches = [	×
212	Bugzilla(
213	params,
214	bughandler=self.bughandler,
215	bugdata=bugs,
216	timeout=timeout,
217	).get_data()
218	for params in params_list
219	]
220
221	for search in searches:	×
222	search.wait()	×
223
224	return bugs	×
225
226	def _is_matching_restrictive_criteria(self, signatures: Iterable) -> bool:	×
227	topcrashes = self._get_restrictive_topcrash_signatures()	×
228	return any(signature in topcrashes for signature in signatures)	×
229
230	def _get_restrictive_topcrash_signatures(self) -> dict:	×
231	if self.topcrashes_restrictive is None:	×
232	restrictive_criteria = []	×
233	for criterion in TOP_CRASH_IDENTIFICATION_CRITERIA:	×
234	restrictive_criterion = {	×
235	**criterion,
236	"tc_limit": criterion["tc_limit"] // 2,
237	}
238
239	if "tc_limit_startup" in criterion:	×
240	restrictive_criterion["tc_limit_startup"] //= 2	×
241
242	restrictive_criteria.append(restrictive_criterion)	×
243
244	self.topcrashes_restrictive = Topcrash(	×
245	criteria=restrictive_criteria
246	).get_signatures()
247
248	return self.topcrashes_restrictive	×
249
250	def get_bz_params_list(self, date):	×
251	self.topcrashes = Topcrash(date).get_signatures()	×
252
253	fields = [	×
254	"triage_owner",
255	"assigned_to",
256	"severity",
257	"keywords",
258	"cf_crash_signature",
259	"history",
260	"comments.creator",
261	"comments.raw_text",
262	]
263	params_base = {	×
264	"include_fields": fields,
265	"resolution": "---",
266	}
267	self.amend_bzparams(params_base, [])	×
268
269	params_list = []	×
270	for signatures in Connection.chunks(	×
271	list(self.topcrashes.keys()),
272	MAX_SIGNATURES_PER_QUERY,
273	):
274	params = params_base.copy()	×
275	n = int(utils.get_last_field_num(params))	×
276	params[f"f{n}"] = "OP"	×
277	params[f"j{n}"] = "OR"	×
278	for signature in signatures:	×
279	n += 1	×
280	params[f"f{n}"] = "cf_crash_signature"	×
281	params[f"o{n}"] = "regexp"	×
282	# Using `(@ \|@)` instead of `@ ?` and ( \]\|\]) instead of ` ?]`
283	# is a workaround. Strangely `?` stays with the encoded form (%3F)
284	# in Bugzilla query.
285	# params[f"v{n}"] = f"\[@ ?{re.escape(signature)} ?\]"
286	params[f"v{n}"] = rf"\[(@ \|@){re.escape(signature)}( \]\|\])"	×
287	params[f"f{n+1}"] = "CP"	×
288	params_list.append(params)	×
289
290	return params_list	×
291
292
293	if __name__ == "__main__":	×
294	TopcrashHighlight().run()	×

mozilla / relman-auto-nag / #4719

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