#4758

Committed 29 Sep 2023 08:26AM CUT coverage: 22.091% (-0.01%) from 22.101%

Build # #4758

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

[topcrash_highlight] Disable re-adding top crash keywords (#2229)

Run Details

716 of 3558 branches covered (0.0%)

5 of 5 new or added lines 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] = {
                "actions": actions,
            }
            return

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

mozilla / relman-auto-nag / #4758

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