#4153

pending completion

Build # #4153

Build Type

push

coveralls-python

Committed by

suhaibmujahid

Commit Message

[duplicate_copy_metadata] Copy WebCompat Priority flag

Run Details

553 of 3089 branches covered (17.9%)

34 of 34 new or added lines in 2 files covered. (100.0%)

1797 of 7986 relevant lines covered (22.5%)

0.23 hits per line

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

28.21

/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
from auto_nag.webcompat_priority import WebcompatPriority

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

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",
                "cf_webcompat_priority",
            ],
            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"])

                # Webcompat Priority: copy the `cf_webcompat_priority` from duplicates
                if (
                    bug.get("cf_webcompat_priority") == "---"
                    and dup_bug.get("cf_webcompat_priority")
                    in WebcompatPriority.NOT_EMPTY_VALUES
                ):
                    new_priority = dup_bug["cf_webcompat_priority"]

                    # Since the bug do not have a priority, it does not make
                    # sense to set it to `revisit`. Instead, we set it to `?` to
                    # request triage.
                    if new_priority == "revisit":
                        new_priority = "?"

                    if (
                        "cf_webcompat_priority" not in copied_fields
                        or WebcompatPriority(new_priority)
                        > WebcompatPriority(
                            copied_fields["cf_webcompat_priority"]["value"]
                        )
                    ):
                        copied_fields["cf_webcompat_priority"] = {
                            "from": [dup_bug["id"]],
                            "value": new_priority,
                        }
                    elif (
                        new_priority == copied_fields["cf_webcompat_priority"]["value"]
                    ):
                        copied_fields["cf_webcompat_priority"]["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 == "cf_webcompat_priority":
                autofix["cf_webcompat_priority"] = 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",
            "cf_webcompat_priority",
        ]

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

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

mozilla / relman-auto-nag / #4153

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