#4381

pending completion

Build # #4381

Build Type

push

coveralls-python

Committed by

suhaibmujahid

Commit Message

[topcrash_add_keyword] Rename `top_crash_signatures` to `topcrash_signatures`

Run Details

563 of 3085 branches covered (18.25%)

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

1798 of 7981 relevant lines covered (22.53%)

0.23 hits per line

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

0.0

/auto_nag/scripts/topcrash_add_keyword.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 auto_nag import utils
from auto_nag.bzcleaner import BzCleaner
from auto_nag.constants import LOW_SEVERITY
from auto_nag.history import History
from auto_nag.topcrash import TOP_CRASH_IDENTIFICATION_CRITERIA, Topcrash

MAX_SIGNATURES_PER_QUERY = 30


class TopcrashAddKeyword(BzCleaner):
    def __init__(self):
        super().__init__()
        self.topcrashes = None
        self.topcrashes_restrictive = None

    def description(self):
        return "Bugs with missing topcrash keywords"

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

    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_autonag(bug, keywords_to_add)

        autofix = {
            "comment": {
                "body": "",
            },
        }

        if keywords_to_add and (
            not is_keywords_removed
            or self._is_matching_restrictive_criteria(topcrash_signatures)
        ):
            autofix["keywords"] = {
                "add": sorted(keywords_to_add),
            }
            autofix["comment"]["body"] += self.get_matching_criteria_comment(
                topcrash_signatures, is_keywords_removed
            )

        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?"
            )

        if not autofix["comment"]["body"]:
            # No comment, no action
            return

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

        data[bugid] = {
            "severity": bug["severity"],
            "added_keywords": (
                utils.english_list(autofix["keywords"]["add"])
                if "keywords" in autofix
                else "-"
            ),
        }

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

mozilla / relman-auto-nag / #4381

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