#4777

Committed 18 Oct 2023 07:48PM CUT coverage: 22.091%. Remained the same

Build # #4777

Build Type

push

coveralls-python

Committed by

suhaibmujahid

Commit Message

[topcrash_highlight] Supply bug metadata even when not applying changes

Run Details

716 of 3558 branches covered (0.0%)

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

1925 of 8714 relevant lines covered (22.09%)

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:
            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)

        if (
            keywords_to_add
            and is_keywords_removed
            and self._is_matching_restrictive_criteria(topcrash_signatures)
        ):
            # FIXME: This is a workaround to monitor the cases where the bot
            # supposed re-add topcrash keywords.
            # More context in: https://github.com/mozilla/bugbot/issues/2100
            actions.extend(
                f"Add {keyword} keyword (not applied to avoid noise)"
                for keyword in keywords_to_add
            )
            data[bugid] = {
                "severity": bug["severity"],
                "actions": actions,
            }
            return bug

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

mozilla / relman-auto-nag / #4777

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