#4154

pending completion

Build # #4154

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

Bump tenacity from 8.2.0 to 8.2.1 (#1884)

Bumps [tenacity](https://github.com/jd/tenacity) from 8.2.0 to 8.2.1.
- [Release notes](https://github.com/jd/tenacity/releases)
- [Commits](https://github.com/jd/tenacity/compare/8.2.0...8.2.1)

---
updated-dependencies:
- dependency-name: tenacity
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

Run Details

549 of 3075 branches covered (17.85%)

1775 of 7952 relevant lines covered (22.32%)

0.22 hits per line

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

30.52

/auto_nag/scripts/duplicate_copy_metadata.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/.

from typing import Any, Dict, List, Set

from libmozdata.bugzilla import Bugzilla

from auto_nag import utils
from auto_nag.bzcleaner import BzCleaner
from auto_nag.history import History

FIELD_NAME_TO_LABEL = {
    "keywords": "Keywords",
    "severity": "Severity",
    "whiteboard": "Whiteboard",
    "cf_performance_impact": "Performance Impact",
    "regressed_by": "Regressed by",
    "status": "Status",
}

FIELD_LABEL_TO_NAME = {label: name for name, label in FIELD_NAME_TO_LABEL.items()}


class DuplicateCopyMetadata(BzCleaner):
    def description(self):
        return "Copied fields from duplicate bugs"

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

        return bug

    def get_bugs(self, date="today", bug_ids=[], chunk_size=None):
        dup_bugs = super().get_bugs(date, bug_ids, chunk_size)

        original_bug_ids = {bug["dupe_of"] for bug in dup_bugs.values()}
        original_bugs = {}

        Bugzilla(
            original_bug_ids,
            include_fields=[
                "id",
                "summary",
                "whiteboard",
                "keywords",
                "duplicates",
                "cf_performance_impact",
                "comments",
                "history",
                "status",
                "regressed_by",
                "is_open",
            ],
            bughandler=self.handle_bug,
            bugdata=original_bugs,
        ).wait()

        results = {}
        for bug_id, bug in original_bugs.items():
            if not bug["is_open"]:
                continue

            copied_fields = {}
            for dup_bug_id in bug["duplicates"]:
                dup_bug_id = str(dup_bug_id)
                dup_bug = dup_bugs.get(dup_bug_id)
                if not dup_bug:
                    continue

                # TODO: Since the logic for copied fields is getting bigger,
                # consider refactoring it in a separate method.

                # Performance Impact: copy the assessment result from duplicates
                if bug.get("cf_performance_impact") == "---" and dup_bug.get(
                    "cf_performance_impact"
                ) not in ("---", "?", None):
                    if "cf_performance_impact" not in copied_fields:
                        copied_fields["cf_performance_impact"] = {
                            "from": [dup_bug["id"]],
                            "value": dup_bug["cf_performance_impact"],
                        }
                    else:
                        copied_fields["cf_performance_impact"]["from"].append(
                            dup_bug["id"]
                        )

                # Keywords: copy the `access` keyword from duplicates
                if "access" not in bug["keywords"] and "access" in dup_bug["keywords"]:
                    if "keywords" not in copied_fields:
                        copied_fields["keywords"] = {
                            "from": [dup_bug["id"]],
                            "value": "access",
                        }
                    else:
                        copied_fields["keywords"]["from"].append(dup_bug["id"])

                # Whiteboard: copy the `access-s*` whiteboard rating from duplicates
                if (
                    "access-s" not in bug["whiteboard"]
                    and "access-s" in dup_bug["whiteboard"]
                ):
                    new_access_tag = utils.get_whiteboard_access_rating(
                        dup_bug["whiteboard"]
                    )

                    if (
                        "whiteboard" not in copied_fields
                        or new_access_tag < copied_fields["whiteboard"]["value"]
                    ):
                        copied_fields["whiteboard"] = {
                            "from": [dup_bug["id"]],
                            "value": new_access_tag,
                        }
                    elif new_access_tag == copied_fields["whiteboard"]["value"]:
                        copied_fields["whiteboard"]["from"].append(dup_bug["id"])
                # Status: confirm the bug if the duplicate was confirmed
                if bug["status"] == "UNCONFIRMED" and self.was_confirmed(dup_bug):
                    if "status" not in copied_fields:
                        copied_fields["status"] = {
                            "from": [dup_bug["id"]],
                            "value": "NEW",
                        }
                    else:
                        copied_fields["status"]["from"].append(dup_bug["id"])

                # Regressed by: move the regressed_by field to the duplicate of
                if dup_bug["regressed_by"]:
                    added_regressed_by = self.get_previously_added_regressors(bug)
                    new_regressed_by = {
                        regression_bug_id
                        for regression_bug_id in dup_bug["regressed_by"]
                        if regression_bug_id not in added_regressed_by
                        and regression_bug_id < int(bug_id)
                    }
                    if new_regressed_by:
                        if "regressed_by" not in copied_fields:
                            copied_fields["regressed_by"] = {
                                "from": [dup_bug["id"]],
                                "value": new_regressed_by,
                            }
                        else:
                            copied_fields["regressed_by"]["from"].append(dup_bug["id"])
                            copied_fields["regressed_by"]["value"] |= new_regressed_by

            previously_copied_fields = self.get_previously_copied_fields(bug)
            # We do not need to ignore the `regressed_by` field because we
            # already check the history to avoid overwriting the engineers.
            previously_copied_fields.discard("regressed_by")
            copied_fields = sorted(
                (
                    field,
                    change["value"],
                    change["from"],
                )
                for field, change in copied_fields.items()
                if field not in previously_copied_fields
            )

            if copied_fields:
                results[bug_id] = {
                    "id": bug_id,
                    "summary": bug["summary"],
                    "copied_fields": copied_fields,
                }

                self.set_autofix(bug, copied_fields)

        return results

    def set_autofix(self, bug: dict, copied_fields: List[tuple]) -> None:
        """Set the autofix for a bug

        Args:
            bug: The bug to set the autofix for.
            copied_fields: The list of copied fields with their values and the
                bugs they were copied from (field, value, source).
        """
        bug_id = str(bug["id"])
        autofix: Dict[str, Any] = {}

        duplicates = {id for _, _, source in copied_fields for id in source}

        # NOTE: modifying the following comment template should also be
        # reflected in the `get_previously_copied_fields` method.
        comment = (
            f"The following {utils.plural('field has', copied_fields, 'fields have')} been copied "
            f"from {utils.plural('a duplicate bug', duplicates, 'duplicate bugs')}:\n\n"
            "| Field | Value | Source |\n"
            "| ----- | ----- | ------ |\n"
        )

        for field, value, source in copied_fields:
            if field == "keywords":
                autofix["keywords"] = {"add": [value]}
            elif field == "whiteboard":
                autofix["whiteboard"] = bug["whiteboard"] + value
            elif field == "cf_performance_impact":
                autofix["cf_performance_impact"] = value
            elif field == "status":
                autofix["status"] = value
            elif field == "regressed_by":
                autofix["regressed_by"] = {"add": list(value)}
                value = utils.english_list(sorted(f"bug {id}" for id in value))
            else:
                raise ValueError(f"Unsupported field: {field}")

            field_label = FIELD_NAME_TO_LABEL[field]
            source = utils.english_list(sorted(f"bug {id}" for id in source))
            comment += f"| {field_label} | {value} | {source} |\n"

        comment += "\n\n" + self.get_documentation()
        autofix["comment"] = {"body": comment}
        # The following is to reduce noise by having the bot to comme later to
        # add the `regression` keyword.
        if "regressed_by" in autofix and "regression" not in bug["keywords"]:
            if "keywords" not in autofix:
                autofix["keywords"] = {"add": ["regression"]}
            else:
                autofix["keywords"]["add"].append("regression")

        self.autofix_changes[bug_id] = autofix

    def get_previously_copied_fields(self, bug: dict) -> Set[str]:
        """Get the fields that have been copied from a bug's duplicates in the past.

        Args:
            bug: The bug to get the previously copied fields for.

        Returns:
            A set of previously copied fields.
        """
        previously_copied_fields = set()

        for comment in bug["comments"]:
            if comment["author"] != History.BOT or not comment["text"].startswith(
                "The following field"
            ):
                continue

            lines = comment["text"].splitlines()
            try:
                table_first_line = lines.index("| Field | Value | Source |")
            except ValueError:
                continue

            for line in lines[table_first_line + 2 :]:
                if not line.startswith("|"):
                    break
                field_label = line.split("|")[1].strip()
                field_name = FIELD_LABEL_TO_NAME[field_label]
                previously_copied_fields.add(field_name)

        return previously_copied_fields

    def get_previously_added_regressors(self, bug: dict) -> Set[int]:
        """Get the bug ids for regressors that have been added to a bug in the
        past.

        Args:
            bug: The bug to get the previously added regressors for.

        Returns:
            A set of ids for previously added regressors.
        """
        added_regressors = {
            int(bug_id)
            for entry in bug["history"]
            for change in entry["changes"]
            if change["field_name"] == "regressed_by"
            for bug_id in change["removed"].split(",")
            if bug_id
        }
        added_regressors.update(bug["regressed_by"])

        return added_regressors

    def was_confirmed(self, bug: dict) -> bool:
        """Check if the bug was confirmed."""

        for entry in reversed(bug["history"]):
            for change in entry["changes"]:
                if change["field_name"] != "status":
                    continue

                if change["removed"] in (
                    "REOPENED",
                    "CLOSED",
                    "RESOLVED",
                ):
                    break

                return change["removed"] != "UNCONFIRMED"

        return False

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

    def get_bz_params(self, date):
        fields = [
            "history",
            "whiteboard",
            "keywords",
            "cf_performance_impact",
            "dupe_of",
            "regressed_by",
        ]

        params = {
            "include_fields": fields,
            "resolution": "DUPLICATE",
            "chfieldfrom": "-7d",
            "chfield": [
                "resolution",
                "keywords",
                "status_whiteboard",
                "cf_performance_impact",
                "regressed_by",
            ],
        }

        return params


if __name__ == "__main__":
    DuplicateCopyMetadata().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	from typing import Any, Dict, List, Set	1✔
6
7	from libmozdata.bugzilla import Bugzilla	1✔
8
9	from auto_nag import utils	1✔
10	from auto_nag.bzcleaner import BzCleaner	1✔
11	from auto_nag.history import History	1✔
12
13	FIELD_NAME_TO_LABEL = {	1✔
14	"keywords": "Keywords",
15	"severity": "Severity",
16	"whiteboard": "Whiteboard",
17	"cf_performance_impact": "Performance Impact",
18	"regressed_by": "Regressed by",
19	"status": "Status",
20	}
21
22	FIELD_LABEL_TO_NAME = {label: name for name, label in FIELD_NAME_TO_LABEL.items()}	1✔
23
24
25	class DuplicateCopyMetadata(BzCleaner):	1✔
26	def description(self):	1✔
27	return "Copied fields from duplicate bugs"	×
28
29	def handle_bug(self, bug, data):	1✔
30	bugid = str(bug["id"])	×
31	data[bugid] = bug	×
32
33	return bug	×
34
35	def get_bugs(self, date="today", bug_ids=[], chunk_size=None):	1✔
36	dup_bugs = super().get_bugs(date, bug_ids, chunk_size)	×
37
38	original_bug_ids = {bug["dupe_of"] for bug in dup_bugs.values()}	×
39	original_bugs = {}	×
40
41	Bugzilla(	×
42	original_bug_ids,
43	include_fields=[
44	"id",
45	"summary",
46	"whiteboard",
47	"keywords",
48	"duplicates",
49	"cf_performance_impact",
50	"comments",
51	"history",
52	"status",
53	"regressed_by",
54	"is_open",
55	],
56	bughandler=self.handle_bug,
57	bugdata=original_bugs,
58	).wait()
59
60	results = {}	×
61	for bug_id, bug in original_bugs.items():	×
62	if not bug["is_open"]:	×
63	continue	×
64
65	copied_fields = {}	×
66	for dup_bug_id in bug["duplicates"]:	×
67	dup_bug_id = str(dup_bug_id)	×
68	dup_bug = dup_bugs.get(dup_bug_id)	×
69	if not dup_bug:	×
70	continue	×
71
72	# TODO: Since the logic for copied fields is getting bigger,
73	# consider refactoring it in a separate method.
74
75	# Performance Impact: copy the assessment result from duplicates
76	if bug.get("cf_performance_impact") == "---" and dup_bug.get(	×
77	"cf_performance_impact"
78	) not in ("---", "?", None):
79	if "cf_performance_impact" not in copied_fields:	×
80	copied_fields["cf_performance_impact"] = {	×
81	"from": [dup_bug["id"]],
82	"value": dup_bug["cf_performance_impact"],
83	}
84	else:
85	copied_fields["cf_performance_impact"]["from"].append(	×
86	dup_bug["id"]
87	)
88
89	# Keywords: copy the `access` keyword from duplicates
90	if "access" not in bug["keywords"] and "access" in dup_bug["keywords"]:	×
91	if "keywords" not in copied_fields:	×
92	copied_fields["keywords"] = {	×
93	"from": [dup_bug["id"]],
94	"value": "access",
95	}
96	else:
97	copied_fields["keywords"]["from"].append(dup_bug["id"])	×
98
99	# Whiteboard: copy the `access-s*` whiteboard rating from duplicates
100	if (	×
101	"access-s" not in bug["whiteboard"]
102	and "access-s" in dup_bug["whiteboard"]
103	):
104	new_access_tag = utils.get_whiteboard_access_rating(	×
105	dup_bug["whiteboard"]
106	)
107
108	if (	×
109	"whiteboard" not in copied_fields
110	or new_access_tag < copied_fields["whiteboard"]["value"]
111	):
112	copied_fields["whiteboard"] = {	×
113	"from": [dup_bug["id"]],
114	"value": new_access_tag,
115	}
116	elif new_access_tag == copied_fields["whiteboard"]["value"]:	×
117	copied_fields["whiteboard"]["from"].append(dup_bug["id"])	×
118	# Status: confirm the bug if the duplicate was confirmed
119	if bug["status"] == "UNCONFIRMED" and self.was_confirmed(dup_bug):	×
120	if "status" not in copied_fields:	×
121	copied_fields["status"] = {	×
122	"from": [dup_bug["id"]],
123	"value": "NEW",
124	}
125	else:
126	copied_fields["status"]["from"].append(dup_bug["id"])	×
127
128	# Regressed by: move the regressed_by field to the duplicate of
129	if dup_bug["regressed_by"]:	×
130	added_regressed_by = self.get_previously_added_regressors(bug)	×
131	new_regressed_by = {	×
132	regression_bug_id
133	for regression_bug_id in dup_bug["regressed_by"]
134	if regression_bug_id not in added_regressed_by
135	and regression_bug_id < int(bug_id)
136	}
137	if new_regressed_by:	×
138	if "regressed_by" not in copied_fields:	×
139	copied_fields["regressed_by"] = {	×
140	"from": [dup_bug["id"]],
141	"value": new_regressed_by,
142	}
143	else:
144	copied_fields["regressed_by"]["from"].append(dup_bug["id"])	×
145	copied_fields["regressed_by"]["value"] \|= new_regressed_by	×
146
147	previously_copied_fields = self.get_previously_copied_fields(bug)	×
148	# We do not need to ignore the `regressed_by` field because we
149	# already check the history to avoid overwriting the engineers.
150	previously_copied_fields.discard("regressed_by")	×
151	copied_fields = sorted(	×
152	(
153	field,
154	change["value"],
155	change["from"],
156	)
157	for field, change in copied_fields.items()
158	if field not in previously_copied_fields
159	)
160
161	if copied_fields:	×
162	results[bug_id] = {	×
163	"id": bug_id,
164	"summary": bug["summary"],
165	"copied_fields": copied_fields,
166	}
167
168	self.set_autofix(bug, copied_fields)	×
169
170	return results	×
171
172	def set_autofix(self, bug: dict, copied_fields: List[tuple]) -> None:	1✔
173	"""Set the autofix for a bug
174
175	Args:
176	bug: The bug to set the autofix for.
177	copied_fields: The list of copied fields with their values and the
178	bugs they were copied from (field, value, source).
179	"""
180	bug_id = str(bug["id"])	1✔
181	autofix: Dict[str, Any] = {}	1✔
182
183	duplicates = {id for _, _, source in copied_fields for id in source}	1✔
184
185	# NOTE: modifying the following comment template should also be
186	# reflected in the `get_previously_copied_fields` method.
187	comment = (	1✔
188	f"The following {utils.plural('field has', copied_fields, 'fields have')} been copied "
189	f"from {utils.plural('a duplicate bug', duplicates, 'duplicate bugs')}:\n\n"
190	"\| Field \| Value \| Source \|\n"
191	"\| ----- \| ----- \| ------ \|\n"
192	)
193
194	for field, value, source in copied_fields:	1✔
195	if field == "keywords":	1!
196	autofix["keywords"] = {"add": [value]}	×
197	elif field == "whiteboard":	1!
198	autofix["whiteboard"] = bug["whiteboard"] + value	1✔
199	elif field == "cf_performance_impact":	×
200	autofix["cf_performance_impact"] = value	×
201	elif field == "status":	×
202	autofix["status"] = value	×
203	elif field == "regressed_by":	×
204	autofix["regressed_by"] = {"add": list(value)}	×
205	value = utils.english_list(sorted(f"bug {id}" for id in value))	×
206	else:
207	raise ValueError(f"Unsupported field: {field}")	×
208
209	field_label = FIELD_NAME_TO_LABEL[field]	1✔
210	source = utils.english_list(sorted(f"bug {id}" for id in source))	1✔
211	comment += f"\| {field_label} \| {value} \| {source} \|\n"	1✔
212
213	comment += "\n\n" + self.get_documentation()	1✔
214	autofix["comment"] = {"body": comment}	1✔
215	# The following is to reduce noise by having the bot to comme later to
216	# add the `regression` keyword.
217	if "regressed_by" in autofix and "regression" not in bug["keywords"]:	1!
218	if "keywords" not in autofix:	×
219	autofix["keywords"] = {"add": ["regression"]}	×
220	else:
221	autofix["keywords"]["add"].append("regression")	×
222
223	self.autofix_changes[bug_id] = autofix	1✔
224
225	def get_previously_copied_fields(self, bug: dict) -> Set[str]:	1✔
226	"""Get the fields that have been copied from a bug's duplicates in the past.
227
228	Args:
229	bug: The bug to get the previously copied fields for.
230
231	Returns:
232	A set of previously copied fields.
233	"""
234	previously_copied_fields = set()	1✔
235
236	for comment in bug["comments"]:	1✔
237	if comment["author"] != History.BOT or not comment["text"].startswith(	1!
238	"The following field"
239	):
240	continue	×
241
242	lines = comment["text"].splitlines()	1✔
243	try:	1✔
244	table_first_line = lines.index("\| Field \| Value \| Source \|")	1✔
245	except ValueError:	×
246	continue	×
247
248	for line in lines[table_first_line + 2 :]:	1!
249	if not line.startswith("\|"):	1✔
250	break	1✔
251	field_label = line.split("\|")[1].strip()	1✔
252	field_name = FIELD_LABEL_TO_NAME[field_label]	1✔
253	previously_copied_fields.add(field_name)	1✔
254
255	return previously_copied_fields	1✔
256
257	def get_previously_added_regressors(self, bug: dict) -> Set[int]:	1✔
258	"""Get the bug ids for regressors that have been added to a bug in the
259	past.
260
261	Args:
262	bug: The bug to get the previously added regressors for.
263
264	Returns:
265	A set of ids for previously added regressors.
266	"""
267	added_regressors = {	×
268	int(bug_id)
269	for entry in bug["history"]
270	for change in entry["changes"]
271	if change["field_name"] == "regressed_by"
272	for bug_id in change["removed"].split(",")
273	if bug_id
274	}
275	added_regressors.update(bug["regressed_by"])	×
276
277	return added_regressors	×
278
279	def was_confirmed(self, bug: dict) -> bool:	1✔
280	"""Check if the bug was confirmed."""
281
282	for entry in reversed(bug["history"]):	×
283	for change in entry["changes"]:	×
284	if change["field_name"] != "status":	×
285	continue	×
286
287	if change["removed"] in (	×
288	"REOPENED",
289	"CLOSED",
290	"RESOLVED",
291	):
292	break	×
293
294	return change["removed"] != "UNCONFIRMED"	×
295
296	return False	×
297
298	def columns(self):	1✔
299	return ["id", "summary", "copied_fields"]	×
300
301	def get_bz_params(self, date):	1✔
302	fields = [	×
303	"history",
304	"whiteboard",
305	"keywords",
306	"cf_performance_impact",
307	"dupe_of",
308	"regressed_by",
309	]
310
311	params = {	×
312	"include_fields": fields,
313	"resolution": "DUPLICATE",
314	"chfieldfrom": "-7d",
315	"chfield": [
316	"resolution",
317	"keywords",
318	"status_whiteboard",
319	"cf_performance_impact",
320	"regressed_by",
321	],
322	}
323
324	return params	×
325
326
327	if __name__ == "__main__":	1!
328	DuplicateCopyMetadata().run()	×

mozilla / relman-auto-nag / #4154

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