#4767

Committed 13 Oct 2023 01:27AM CUT coverage: 22.091%. Remained the same

Build # #4767

Build Type

push

coveralls-python

Committed by

suhaibmujahid

Commit Message

Format the .pre-commit-config.yaml file

Run Details

716 of 3558 branches covered (0.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

67.44

/bugbot/round_robin.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 Dict, Set

import gspread
from dateutil.relativedelta import relativedelta
from libmozdata import utils as lmdutils
from libmozdata.bugzilla import BugzillaUser

from bugbot import logger, utils
from bugbot.components import Components
from bugbot.people import People
from bugbot.round_robin_calendar import BadFallback, Calendar, InvalidCalendar


class RoundRobin(object):
    _instances: dict = {}

    def __init__(self, rotation_definitions=None, people=None, teams=None):
        self.people = People.get_instance() if people is None else people
        self.components_by_triager: Dict[str, list] = {}
        self.rotation_definitions = (
            RotationDefinitions()
            if rotation_definitions is None
            else rotation_definitions
        )
        self.feed(None if teams is None else set(teams))
        self.nicks = {}
        self.erroneous_bzmail = {}
        utils.init_random()

    @staticmethod
    def get_instance(teams=None):
        if teams is None:
            if None not in RoundRobin._instances:
                RoundRobin._instances[None] = RoundRobin()
            return RoundRobin._instances[None]

        teams = tuple(sorted(teams))
        if teams not in RoundRobin._instances:
            RoundRobin._instances[teams] = RoundRobin(teams=teams)
        return RoundRobin._instances[teams]

    def feed(self, teams: Set[str] | None = None) -> None:
        """Fetch the rotations calendars.

        Args:
            teams: if provided, only calendars for the specified teams will be
                fetched.
        """

        self.data = {}
        cache = {}

        team_calendars = self.rotation_definitions.fetch_by_teams()
        for team_name, components in team_calendars.items():
            if teams is not None and team_name not in teams:
                continue
            try:
                for component_name, calendar_info in components.items():
                    url = calendar_info["url"]
                    if url not in cache:
                        calendar = cache[url] = Calendar.get(
                            url,
                            calendar_info["fallback"],
                            team_name,
                            people=self.people,
                        )
                    else:
                        calendar = cache[url]
                        if calendar.get_fallback() != calendar_info["fallback"]:
                            raise BadFallback(
                                "Cannot have different fallback triagers for the same calendar"
                            )

                    self.data[component_name] = calendar

            except (BadFallback, InvalidCalendar) as err:
                logger.error(err)
                # If one the team's calendars failed, it is better to fail loud,
                # and disable all team's calendars.
                for component_name in components:
                    if component_name in self.data:
                        del self.data[component_name]

    def get_components(self):
        return list(self.data.keys())

    def get_components_for_triager(self, triager):
        return self.components_by_triager[triager]

    def add_component_for_triager(self, component, triagers):
        if not isinstance(triagers, list):
            triagers = [triagers]
        for triager in triagers:
            if triager in self.components_by_triager:
                self.components_by_triager[triager].add(component)
            else:
                self.components_by_triager[triager] = {component}

    def get_fallback(self, bug):
        pc = bug["product"] + "::" + bug["component"]
        if pc not in self.data:
            mail = bug.get("triage_owner")
        else:
            cal = self.data[pc]
            mail = cal.get_fallback_bzmail()

        return self.people.get_moz_mail(mail)

    def get_erroneous_bzmail(self):
        return self.erroneous_bzmail

    def add_erroneous_bzmail(self, bzmail, prod_comp, cal):
        logger.error(f"No nick for {bzmail} for {prod_comp}")
        fb = cal.get_fallback_mozmail()
        if fb not in self.erroneous_bzmail:
            self.erroneous_bzmail[fb] = {bzmail}
        else:
            self.erroneous_bzmail[fb].add(bzmail)

    def get_nick(self, bzmail, prod_comp, cal):
        if bzmail not in self.nicks:

            def handler(user):
                self.nicks[bzmail] = user["nick"]

            BugzillaUser(user_names=[bzmail], user_handler=handler).wait()

        if bzmail not in self.nicks:
            self.add_erroneous_bzmail(bzmail, prod_comp, cal)
            return None

        return self.nicks[bzmail]

    def get(self, bug, date, only_one=True, has_nick=True):
        pc = bug["product"] + "::" + bug["component"]
        if pc not in self.data:
            mail = bug.get("triage_owner")
            nick = bug.get("triage_owner_detail", {}).get("nick")
            if utils.is_no_assignee(mail):
                mail, nick = None, None

            if mail is None:
                logger.error("No triage owner for {}".format(pc))

            self.add_component_for_triager(pc, mail)

            if has_nick:
                return mail, nick if only_one else [(mail, nick)]
            return mail if only_one else [mail]

        cal = self.data[pc]
        persons = cal.get_persons(date)
        fb = cal.get_fallback_bzmail()
        if not persons or all(p is None for _, p in persons):
            # the fallback is the triage owner
            self.add_component_for_triager(pc, [fb])
            return (fb, self.get_nick(fb, pc, cal)) if has_nick else fb

        bzmails = []
        for _, p in persons:
            bzmails.append(fb if p is None else p)

        self.add_component_for_triager(pc, bzmails)

        if only_one:
            bzmail = bzmails[0]
            if has_nick:
                nick = self.get_nick(bzmail, pc, cal)
                return bzmail, nick
            return bzmail

        if has_nick:
            return [(bzmail, self.get_nick(bzmail, pc, cal)) for bzmail in bzmails]
        return bzmails

    def get_who_to_nag(self, date):
        fallbacks = {}
        date = lmdutils.get_date_ymd(date)
        days = utils.get_config("round-robin", "days_to_nag", 7)
        next_date = date + relativedelta(days=days)
        for cal in set(self.data.values()):
            persons = cal.get_persons(next_date)
            if persons and all(p is not None for _, p in persons):
                continue

            name = cal.get_team_name()
            fb = cal.get_fallback_mozmail()
            if fb not in fallbacks:
                fallbacks[fb] = {}
            if name not in fallbacks[fb]:
                fallbacks[fb][name] = {"nobody": False, "persons": []}
            info = fallbacks[fb][name]

            if not persons:
                info["nobody"] = True
            else:
                people_names = [n for n, p in persons if p is None]
                if people_names:
                    info["persons"] += people_names
        return fallbacks


CalendarDef = Dict[str, str]
ComponentCalendarDefs = Dict[str, CalendarDef]
TeamCalendarDefs = Dict[str, ComponentCalendarDefs]


class RotationDefinitions:
    """Definitions for triage owner rotations"""

    def __init__(self) -> None:
        self.definitions_url = utils.get_private()["round_robin_sheet"]
        self.components = Components.get_instance()

        gc = gspread.service_account_from_dict(utils.get_gcp_service_account_info())
        # The spreadsheet key should match the one listed in the Firefox Source Docs:
        # https://firefox-source-docs.mozilla.org/bug-mgmt/policies/triage-bugzilla.html#rotating-triage
        doc = gc.open_by_key("1EK6iCtdD8KP4UflIHscuZo6W5er2vy_TX7vsmaaBVd4")
        self.sheet = doc.worksheet("definitions")

    def fetch_by_teams(self) -> TeamCalendarDefs:
        """Fetch the triage owner rotation definitions and group them by team

        Returns:
            A dictionary that maps each component to its calendar and fallback
            person. The components are grouped by their teams. The following is
            the shape of the returned dictionary:
            {
                team_name: {
                    component_name:{
                            "fallback": "the name of the fallback person",
                            "calendar": "the URL for the rotation calendar"
                    }
                    ...
                }
                ...
            }
        """

        teams: TeamCalendarDefs = {}
        seen = set()
        for row in self.get_definitions_records():
            team_name = row["Team Name"]
            scope = row["Calendar Scope"]
            fallback_triager = row["Fallback Triager"]
            calendar_url = row["Calendar URL"]

            if (team_name, scope) in seen:
                logger.error(
                    "The triage owner rotation definitions show more than one "
                    "entry for the %s team with the component scope '%s'",
                    team_name,
                    scope,
                )
            else:
                seen.add((team_name, scope))

            if team_name in teams:
                component_calendar = teams[team_name]
            else:
                component_calendar = teams[team_name] = {}

            if scope == "All Team's Components":
                team_components = self.components.get_team_components(team_name)
                components_to_add = [
                    str(component_name)
                    for component_name in team_components
                    if str(component_name) not in component_calendar
                ]
            else:
                components_to_add = [scope]

            for component_name in components_to_add:
                component_calendar[component_name] = {
                    "fallback": fallback_triager,
                    "url": calendar_url,
                }

        return teams

    def get_definitions_records(self) -> list:
        """Fetch the triage owner rotation definitions."""
        return self.sheet.get_all_records()

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 Dict, Set	1✔
6
7	import gspread	1✔
8	from dateutil.relativedelta import relativedelta	1✔
9	from libmozdata import utils as lmdutils	1✔
10	from libmozdata.bugzilla import BugzillaUser	1✔
11
12	from bugbot import logger, utils	1✔
13	from bugbot.components import Components	1✔
14	from bugbot.people import People	1✔
15	from bugbot.round_robin_calendar import BadFallback, Calendar, InvalidCalendar	1✔
16
17
18	class RoundRobin(object):	1✔
19	_instances: dict = {}	1✔
20
21	def __init__(self, rotation_definitions=None, people=None, teams=None):	1✔
22	self.people = People.get_instance() if people is None else people	1✔
23	self.components_by_triager: Dict[str, list] = {}	1✔
24	self.rotation_definitions = (	1✔
25	RotationDefinitions()
26	if rotation_definitions is None
27	else rotation_definitions
28	)
29	self.feed(None if teams is None else set(teams))	1✔
30	self.nicks = {}	1✔
31	self.erroneous_bzmail = {}	1✔
32	utils.init_random()	1✔
33
34	@staticmethod	1✔
35	def get_instance(teams=None):	1✔
36	if teams is None:	×
37	if None not in RoundRobin._instances:	×
38	RoundRobin._instances[None] = RoundRobin()	×
39	return RoundRobin._instances[None]	×
40
41	teams = tuple(sorted(teams))	×
42	if teams not in RoundRobin._instances:	×
43	RoundRobin._instances[teams] = RoundRobin(teams=teams)	×
44	return RoundRobin._instances[teams]	×
45
46	def feed(self, teams: Set[str] \| None = None) -> None:	1✔
47	"""Fetch the rotations calendars.
48
49	Args:
50	teams: if provided, only calendars for the specified teams will be
51	fetched.
52	"""
53
54	self.data = {}	1✔
55	cache = {}	1✔
56
57	team_calendars = self.rotation_definitions.fetch_by_teams()	1✔
58	for team_name, components in team_calendars.items():	1✔
59	if teams is not None and team_name not in teams:	1!
60	continue	×
61	try:	1✔
62	for component_name, calendar_info in components.items():	1✔
63	url = calendar_info["url"]	1✔
64	if url not in cache:	1✔
65	calendar = cache[url] = Calendar.get(	1✔
66	url,
67	calendar_info["fallback"],
68	team_name,
69	people=self.people,
70	)
71	else:
72	calendar = cache[url]	1✔
73	if calendar.get_fallback() != calendar_info["fallback"]:	1!
74	raise BadFallback(	×
75	"Cannot have different fallback triagers for the same calendar"
76	)
77
78	self.data[component_name] = calendar	1✔
79
80	except (BadFallback, InvalidCalendar) as err:	×
81	logger.error(err)	×
82	# If one the team's calendars failed, it is better to fail loud,
83	# and disable all team's calendars.
84	for component_name in components:	×
85	if component_name in self.data:	×
86	del self.data[component_name]	×
87
88	def get_components(self):	1✔
89	return list(self.data.keys())	×
90
91	def get_components_for_triager(self, triager):	1✔
92	return self.components_by_triager[triager]	×
93
94	def add_component_for_triager(self, component, triagers):	1✔
95	if not isinstance(triagers, list):	1✔
96	triagers = [triagers]	1✔
97	for triager in triagers:	1✔
98	if triager in self.components_by_triager:	1✔
99	self.components_by_triager[triager].add(component)	1✔
100	else:
101	self.components_by_triager[triager] = {component}	1✔
102
103	def get_fallback(self, bug):	1✔
104	pc = bug["product"] + "::" + bug["component"]	×
105	if pc not in self.data:	×
106	mail = bug.get("triage_owner")	×
107	else:
108	cal = self.data[pc]	×
109	mail = cal.get_fallback_bzmail()	×
110
111	return self.people.get_moz_mail(mail)	×
112
113	def get_erroneous_bzmail(self):	1✔
114	return self.erroneous_bzmail	×
115
116	def add_erroneous_bzmail(self, bzmail, prod_comp, cal):	1✔
117	logger.error(f"No nick for {bzmail} for {prod_comp}")	×
118	fb = cal.get_fallback_mozmail()	×
119	if fb not in self.erroneous_bzmail:	×
120	self.erroneous_bzmail[fb] = {bzmail}	×
121	else:
122	self.erroneous_bzmail[fb].add(bzmail)	×
123
124	def get_nick(self, bzmail, prod_comp, cal):	1✔
125	if bzmail not in self.nicks:	×
126
127	def handler(user):	×
128	self.nicks[bzmail] = user["nick"]	×
129
130	BugzillaUser(user_names=[bzmail], user_handler=handler).wait()	×
131
132	if bzmail not in self.nicks:	×
133	self.add_erroneous_bzmail(bzmail, prod_comp, cal)	×
134	return None	×
135
136	return self.nicks[bzmail]	×
137
138	def get(self, bug, date, only_one=True, has_nick=True):	1✔
139	pc = bug["product"] + "::" + bug["component"]	1✔
140	if pc not in self.data:	1✔
141	mail = bug.get("triage_owner")	1✔
142	nick = bug.get("triage_owner_detail", {}).get("nick")	1✔
143	if utils.is_no_assignee(mail):	1!
144	mail, nick = None, None	×
145
146	if mail is None:	1!
147	logger.error("No triage owner for {}".format(pc))	×
148
149	self.add_component_for_triager(pc, mail)	1✔
150
151	if has_nick:	1!
152	return mail, nick if only_one else [(mail, nick)]	1✔
153	return mail if only_one else [mail]	×
154
155	cal = self.data[pc]	1✔
156	persons = cal.get_persons(date)	1✔
157	fb = cal.get_fallback_bzmail()	1✔
158	if not persons or all(p is None for _, p in persons):	1!
159	# the fallback is the triage owner
160	self.add_component_for_triager(pc, [fb])	1✔
161	return (fb, self.get_nick(fb, pc, cal)) if has_nick else fb	1✔
162
163	bzmails = []	1✔
164	for _, p in persons:	1✔
165	bzmails.append(fb if p is None else p)	1✔
166
167	self.add_component_for_triager(pc, bzmails)	1✔
168
169	if only_one:	1!
170	bzmail = bzmails[0]	1✔
171	if has_nick:	1!
172	nick = self.get_nick(bzmail, pc, cal)	1✔
173	return bzmail, nick	1✔
174	return bzmail	×
175
176	if has_nick:	×
177	return [(bzmail, self.get_nick(bzmail, pc, cal)) for bzmail in bzmails]	×
178	return bzmails	×
179
180	def get_who_to_nag(self, date):	1✔
181	fallbacks = {}	1✔
182	date = lmdutils.get_date_ymd(date)	1✔
183	days = utils.get_config("round-robin", "days_to_nag", 7)	1✔
184	next_date = date + relativedelta(days=days)	1✔
185	for cal in set(self.data.values()):	1✔
186	persons = cal.get_persons(next_date)	1✔
187	if persons and all(p is not None for _, p in persons):	1✔
188	continue	1✔
189
190	name = cal.get_team_name()	1✔
191	fb = cal.get_fallback_mozmail()	1✔
192	if fb not in fallbacks:	1✔
193	fallbacks[fb] = {}	1✔
194	if name not in fallbacks[fb]:	1✔
195	fallbacks[fb][name] = {"nobody": False, "persons": []}	1✔
196	info = fallbacks[fb][name]	1✔
197
198	if not persons:	1!
199	info["nobody"] = True	1✔
200	else:
201	people_names = [n for n, p in persons if p is None]	×
202	if people_names:	×
203	info["persons"] += people_names	×
204	return fallbacks	1✔
205
206
207	CalendarDef = Dict[str, str]	1✔
208	ComponentCalendarDefs = Dict[str, CalendarDef]	1✔
209	TeamCalendarDefs = Dict[str, ComponentCalendarDefs]	1✔
210
211
212	class RotationDefinitions:	1✔
213	"""Definitions for triage owner rotations"""
214
215	def __init__(self) -> None:	1✔
216	self.definitions_url = utils.get_private()["round_robin_sheet"]	×
217	self.components = Components.get_instance()	×
218
219	gc = gspread.service_account_from_dict(utils.get_gcp_service_account_info())	×
220	# The spreadsheet key should match the one listed in the Firefox Source Docs:
221	# https://firefox-source-docs.mozilla.org/bug-mgmt/policies/triage-bugzilla.html#rotating-triage
222	doc = gc.open_by_key("1EK6iCtdD8KP4UflIHscuZo6W5er2vy_TX7vsmaaBVd4")	×
223	self.sheet = doc.worksheet("definitions")	×
224
225	def fetch_by_teams(self) -> TeamCalendarDefs:	1✔
226	"""Fetch the triage owner rotation definitions and group them by team
227
228	Returns:
229	A dictionary that maps each component to its calendar and fallback
230	person. The components are grouped by their teams. The following is
231	the shape of the returned dictionary:
232	{
233	team_name: {
234	component_name:{
235	"fallback": "the name of the fallback person",
236	"calendar": "the URL for the rotation calendar"
237	}
238	...
239	}
240	...
241	}
242	"""
243
244	teams: TeamCalendarDefs = {}	1✔
245	seen = set()	1✔
246	for row in self.get_definitions_records():	1✔
247	team_name = row["Team Name"]	1✔
248	scope = row["Calendar Scope"]	1✔
249	fallback_triager = row["Fallback Triager"]	1✔
250	calendar_url = row["Calendar URL"]	1✔
251
252	if (team_name, scope) in seen:	1!
253	logger.error(	×
254	"The triage owner rotation definitions show more than one "
255	"entry for the %s team with the component scope '%s'",
256	team_name,
257	scope,
258	)
259	else:
260	seen.add((team_name, scope))	1✔
261
262	if team_name in teams:	1✔
263	component_calendar = teams[team_name]	1✔
264	else:
265	component_calendar = teams[team_name] = {}	1✔
266
267	if scope == "All Team's Components":	1!
268	team_components = self.components.get_team_components(team_name)	×
269	components_to_add = [	×
270	str(component_name)
271	for component_name in team_components
272	if str(component_name) not in component_calendar
273	]
274	else:
275	components_to_add = [scope]	1✔
276
277	for component_name in components_to_add:	1✔
278	component_calendar[component_name] = {	1✔
279	"fallback": fallback_triager,
280	"url": calendar_url,
281	}
282
283	return teams	1✔
284
285	def get_definitions_records(self) -> list:	1✔
286	"""Fetch the triage owner rotation definitions."""
287	return self.sheet.get_all_records()	×

mozilla / relman-auto-nag / #4767

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