#5134

Committed 02 Jul 2024 04:26PM CUT coverage: 21.657%. Remained the same

Build # #5134

Build Type

push

coveralls-python

Committed by

benjaminmah

Commit Message

Replaced `Exception` with `KeyError`

Run Details

716 of 3630 branches covered (19.72%)

0 of 1 new or added line in 1 file covered. (0.0%)

1934 of 8930 relevant lines covered (21.66%)

0.22 hits per line

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

53.6

/bugbot/user_activity.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 logging
from datetime import timedelta
from enum import Enum, auto
from typing import Iterable, List, Optional

from libmozdata import utils as lmdutils
from libmozdata.bugzilla import BugzillaUser
from libmozdata.connection import Connection
from libmozdata.phabricator import PhabricatorAPI
from tenacity import retry, stop_after_attempt, wait_exponential

from bugbot import utils
from bugbot.people import People

logging.basicConfig(level=logging.DEBUG)

# The chunk size here should not be more than 100; which is the maximum number of
# items that Phabricator could return in one response.
PHAB_CHUNK_SIZE = 100


class UserStatus(Enum):
    ACTIVE = auto()
    UNDEFINED = auto()
    DISABLED = auto()
    INACTIVE = auto()
    ABSENT = auto()
    UNAVAILABLE = auto()


class UserActivity:
    """Check the user activity on Bugzilla and Phabricator"""

    def __init__(
        self,
        activity_weeks_count: int = 26,
        absent_weeks_count: int = 26,
        unavailable_max_days: int = 7,
        include_fields: list | None = None,
        phab: PhabricatorAPI | None = None,
        people: People | None = None,
        reference_date: str = "today",
    ) -> None:
        """
        Constructor

        Args:
            activity_weeks_count: the number of weeks since last made a change
                to a bug before a user being considered as inactive.
            absent_weeks_count: the number of weeks since last loaded any page
                from Bugzilla before a user being considered as inactive.
            unavailable_max_days: a user will be considered inactive if they
                have more days left to be available than `unavailable_max_days`.
            include_fields: the list of fields to include with the the Bugzilla
                user object.
            phab: if an instance of PhabricatorAPI is not provided, it will be
                created when it is needed.
            people: if an instance of People is not provided, the global
                instance will be used.
            reference_date: the reference date to use for checking user
                activity. This is needed for testing because the dates in the
                mock data are fixed.
        """
        self.activity_weeks_count = activity_weeks_count
        self.absent_weeks_count = absent_weeks_count
        self.include_fields = include_fields or []
        self.people = people if people is not None else People.get_instance()
        self.phab = phab
        self.availability_limit = (
            lmdutils.get_date_ymd(reference_date) + timedelta(unavailable_max_days)
        ).timestamp()

        self.activity_limit = lmdutils.get_date(
            reference_date, self.activity_weeks_count * 7
        )
        self.activity_limit_ts = lmdutils.get_date_ymd(self.activity_limit).timestamp()
        self.seen_limit = lmdutils.get_date(reference_date, self.absent_weeks_count * 7)

    def _get_phab(self):
        if not self.phab:
            self.phab = PhabricatorAPI(utils.get_login_info()["phab_api_key"])

        return self.phab

    def check_users(
        self,
        user_emails: Iterable[str],
        keep_active: bool = False,
        ignore_bots: bool = False,
        fetch_employee_info: bool = False,
    ) -> dict:
        """Check user activity using their emails

        Args:
            user_emails: the email addresses of the users.
            keep_active: whether the returned results should include the active
                users.
            ignore_bots: whether the returned results should include bot and
                component-watching accounts.
            fetch_employee_info: whether to fetch the employee info from
                Bugzilla. Only fields specified in `include_fields` will be
                guaranteed to be fetched.

        Returns:
            A dictionary where the key is the user email and the value is the
                user info with the status.
        """

        user_statuses = {
            user_email: {
                "status": (
                    UserStatus.UNDEFINED
                    if utils.is_no_assignee(user_email)
                    else UserStatus.ACTIVE
                ),
                "is_employee": self.people.is_mozilla(user_email),
            }
            for user_email in user_emails
            if not ignore_bots or not utils.is_bot_email(user_email)
        }

        # Employees will always be considered active
        user_emails = [
            user_email
            for user_email, info in user_statuses.items()
            if not info["is_employee"] and info["status"] == UserStatus.ACTIVE
        ]

        if not keep_active:
            user_statuses = {
                user_email: info
                for user_email, info in user_statuses.items()
                if info["status"] != UserStatus.ACTIVE
            }

        if fetch_employee_info:
            employee_emails = [
                user_email
                for user_email, info in user_statuses.items()
                if info["is_employee"]
            ]
            if employee_emails:
                BugzillaUser(
                    user_names=employee_emails,
                    user_data=user_statuses,
                    user_handler=lambda user, data: data[user["name"]].update(user),
                    include_fields=self.include_fields + ["name"],
                ).wait()

        if user_emails:
            user_statuses.update(
                self.get_bz_users_with_status(user_emails, keep_active)
            )

        return user_statuses

    def get_status_from_bz_user(self, user: dict) -> UserStatus:
        """Get the user status from a Bugzilla user object."""

        if not user["can_login"]:
            return UserStatus.DISABLED

        if user["creation_time"] > self.seen_limit:
            return UserStatus.ACTIVE

        if user["last_seen_date"] is None or user["last_seen_date"] < self.seen_limit:
            return UserStatus.ABSENT

        if (
            user["last_activity_time"] is None
            or user["last_activity_time"] < self.activity_limit
        ):
            return UserStatus.INACTIVE

        return UserStatus.ACTIVE

    def get_bz_users_with_status(
        self, id_or_name: list, keep_active: bool = True
    ) -> dict:
        """Get Bugzilla users with their activity statuses.

        Args:
            id_or_name: An integer user ID or login name of the user on
                bugzilla.
            keep_active: whether the returned results should include the active
                users.

        Returns:
            A dictionary where the key is the user login name and the value is
            the user info with the status.
        """

        def handler(user, data):
            status = self.get_status_from_bz_user(user)
            if keep_active or status != UserStatus.ACTIVE:
                user["status"] = status
                data[user["name"]] = user

        users: dict = {}
        BugzillaUser(
            user_data=users,
            user_names=id_or_name,
            user_handler=handler,
            include_fields=[
                "name",
                "can_login",
                "last_activity_time",
                "last_seen_date",
                "creation_time",
            ]
            + self.include_fields,
        ).wait()

        return users

    def _get_status_from_phab_user(self, user: dict) -> Optional[UserStatus]:
        if "disabled" in user["fields"]["roles"]:
            return UserStatus.DISABLED

        availability = user["attachments"]["availability"]
        if availability["value"] != "available":
            # We do not need to consider the user inactive they will be
            # available again soon.
            if (
                not availability["until"]
                or availability["until"] > self.availability_limit
            ):
                return UserStatus.UNAVAILABLE

        return None

    def get_phab_users_with_status(
        self, user_phids: List[str], keep_active: bool = False
    ) -> dict:
        """Get Phabricator users with their activity statuses.

        Args:
            user_phids: A list of user PHIDs.
            keep_active: whether the returned results should include the active
                users.

        Returns:
            A dictionary where the key is the user PHID and the value is
            the user info with the status.
        """

        bzid_to_phid = {
            int(user["id"]): user["phid"]
            for _user_phids in Connection.chunks(user_phids, PHAB_CHUNK_SIZE)
            for user in self._fetch_bz_user_ids(user_phids=_user_phids)
        }
        if not bzid_to_phid:
            return {}

        if "id" not in self.include_fields:
            self.include_fields.append("id")

        user_bz_ids = list(bzid_to_phid.keys())
        users = self.get_bz_users_with_status(user_bz_ids, keep_active=True)
        users = {bzid_to_phid[user["id"]]: user for user in users.values()}

        # To cover cases where a person is temporary off (e.g., long PTO), we
        # will rely on the calendar from phab.
        for _user_phids in Connection.chunks(user_phids, PHAB_CHUNK_SIZE):
            for phab_user in self._fetch_phab_users(_user_phids):
                try:
                    user = users[phab_user["phid"]]
                    phab_status = self._get_status_from_phab_user(phab_user)
                    if phab_status:
                        user["status"] = phab_status

                    elif user["status"] in (
                        UserStatus.ABSENT,
                        UserStatus.INACTIVE,
                    ) and self.is_active_on_phab(phab_user["phid"]):
                        user["status"] = UserStatus.ACTIVE

                    if not keep_active and user["status"] == UserStatus.ACTIVE:
                        del users[phab_user["phid"]]
                        continue

                    user["phab_username"] = phab_user["fields"]["username"]
                    user["unavailable_until"] = phab_user["attachments"][
                        "availability"
                    ]["until"]
                except KeyError as e:
                    logging.error(
                        f"Error fetching inactive patch authors: '{phab_user['phid']}' - {str(e)}"
                    )
                    continue

        return users

    def is_active_on_phab(self, user_phid: str) -> bool:
        """Check if the user has recent activities on Phabricator.

        Args:
            user_phid: The user PHID.

        Returns:
            True if the user is active on Phabricator, False otherwise.
        """

        feed = self._get_phab().request(
            "feed.query",
            filterPHIDs=[user_phid],
            limit=1,
        )
        for story in feed.values():
            if story["epoch"] >= self.activity_limit_ts:
                return True

        return False

    def get_string_status(self, status: UserStatus):
        """Get a string representation of the user status."""

        if status == UserStatus.UNDEFINED:
            return "Not specified"
        if status == UserStatus.DISABLED:
            return "Account disabled"
        if status == UserStatus.INACTIVE:
            return f"Inactive on Bugzilla in last {self.activity_weeks_count} weeks"
        if status == UserStatus.ABSENT:
            return f"Not seen on Bugzilla in last {self.absent_weeks_count} weeks"

        return status.name

    @retry(
        wait=wait_exponential(min=4),
        stop=stop_after_attempt(5),
    )
    def _fetch_phab_users(self, phids: list):
        if len(phids) == 0:
            return []

        return self._get_phab().search_users(
            constraints={"phids": phids},
            attachments={"availability": True},
        )

    @retry(
        wait=wait_exponential(min=4),
        stop=stop_after_attempt(5),
    )
    def _fetch_bz_user_ids(self, *args, **kwargs):
        return self._get_phab().load_bz_account(*args, **kwargs)

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 logging	1✔
6	from datetime import timedelta	1✔
7	from enum import Enum, auto	1✔
8	from typing import Iterable, List, Optional	1✔
9
10	from libmozdata import utils as lmdutils	1✔
11	from libmozdata.bugzilla import BugzillaUser	1✔
12	from libmozdata.connection import Connection	1✔
13	from libmozdata.phabricator import PhabricatorAPI	1✔
14	from tenacity import retry, stop_after_attempt, wait_exponential	1✔
15
16	from bugbot import utils	1✔
17	from bugbot.people import People	1✔
18
19	logging.basicConfig(level=logging.DEBUG)	1✔
20
21	# The chunk size here should not be more than 100; which is the maximum number of
22	# items that Phabricator could return in one response.
23	PHAB_CHUNK_SIZE = 100	1✔
24
25
26	class UserStatus(Enum):	1✔
27	ACTIVE = auto()	1✔
28	UNDEFINED = auto()	1✔
29	DISABLED = auto()	1✔
30	INACTIVE = auto()	1✔
31	ABSENT = auto()	1✔
32	UNAVAILABLE = auto()	1✔
33
34
35	class UserActivity:	1✔
36	"""Check the user activity on Bugzilla and Phabricator"""
37
38	def __init__(	1✔
39	self,
40	activity_weeks_count: int = 26,
41	absent_weeks_count: int = 26,
42	unavailable_max_days: int = 7,
43	include_fields: list \| None = None,
44	phab: PhabricatorAPI \| None = None,
45	people: People \| None = None,
46	reference_date: str = "today",
47	) -> None:
48	"""
49	Constructor
50
51	Args:
52	activity_weeks_count: the number of weeks since last made a change
53	to a bug before a user being considered as inactive.
54	absent_weeks_count: the number of weeks since last loaded any page
55	from Bugzilla before a user being considered as inactive.
56	unavailable_max_days: a user will be considered inactive if they
57	have more days left to be available than `unavailable_max_days`.
58	include_fields: the list of fields to include with the the Bugzilla
59	user object.
60	phab: if an instance of PhabricatorAPI is not provided, it will be
61	created when it is needed.
62	people: if an instance of People is not provided, the global
63	instance will be used.
64	reference_date: the reference date to use for checking user
65	activity. This is needed for testing because the dates in the
66	mock data are fixed.
67	"""
68	self.activity_weeks_count = activity_weeks_count	1✔
69	self.absent_weeks_count = absent_weeks_count	1✔
70	self.include_fields = include_fields or []	1✔
71	self.people = people if people is not None else People.get_instance()	1✔
72	self.phab = phab	1✔
73	self.availability_limit = (	1✔
74	lmdutils.get_date_ymd(reference_date) + timedelta(unavailable_max_days)
75	).timestamp()
76
77	self.activity_limit = lmdutils.get_date(	1✔
78	reference_date, self.activity_weeks_count * 7
79	)
80	self.activity_limit_ts = lmdutils.get_date_ymd(self.activity_limit).timestamp()	1✔
81	self.seen_limit = lmdutils.get_date(reference_date, self.absent_weeks_count * 7)	1✔
82
83	def _get_phab(self):	1✔
84	if not self.phab:	×
85	self.phab = PhabricatorAPI(utils.get_login_info()["phab_api_key"])	×
86
87	return self.phab	×
88
89	def check_users(	1✔
90	self,
91	user_emails: Iterable[str],
92	keep_active: bool = False,
93	ignore_bots: bool = False,
94	fetch_employee_info: bool = False,
95	) -> dict:
96	"""Check user activity using their emails
97
98	Args:
99	user_emails: the email addresses of the users.
100	keep_active: whether the returned results should include the active
101	users.
102	ignore_bots: whether the returned results should include bot and
103	component-watching accounts.
104	fetch_employee_info: whether to fetch the employee info from
105	Bugzilla. Only fields specified in `include_fields` will be
106	guaranteed to be fetched.
107
108	Returns:
109	A dictionary where the key is the user email and the value is the
110	user info with the status.
111	"""
112
113	user_statuses = {	1✔
114	user_email: {
115	"status": (
116	UserStatus.UNDEFINED
117	if utils.is_no_assignee(user_email)
118	else UserStatus.ACTIVE
119	),
120	"is_employee": self.people.is_mozilla(user_email),
121	}
122	for user_email in user_emails
123	if not ignore_bots or not utils.is_bot_email(user_email)
124	}
125
126	# Employees will always be considered active
127	user_emails = [	1✔
128	user_email
129	for user_email, info in user_statuses.items()
130	if not info["is_employee"] and info["status"] == UserStatus.ACTIVE
131	]
132
133	if not keep_active:	1✔
134	user_statuses = {	1✔
135	user_email: info
136	for user_email, info in user_statuses.items()
137	if info["status"] != UserStatus.ACTIVE
138	}
139
140	if fetch_employee_info:	1!
141	employee_emails = [	×
142	user_email
143	for user_email, info in user_statuses.items()
144	if info["is_employee"]
145	]
146	if employee_emails:	×
147	BugzillaUser(	×
148	user_names=employee_emails,
149	user_data=user_statuses,
150	user_handler=lambda user, data: data[user["name"]].update(user),
151	include_fields=self.include_fields + ["name"],
152	).wait()
153
154	if user_emails:	1!
155	user_statuses.update(	1✔
156	self.get_bz_users_with_status(user_emails, keep_active)
157	)
158
159	return user_statuses	1✔
160
161	def get_status_from_bz_user(self, user: dict) -> UserStatus:	1✔
162	"""Get the user status from a Bugzilla user object."""
163
164	if not user["can_login"]:	1✔
165	return UserStatus.DISABLED	1✔
166
167	if user["creation_time"] > self.seen_limit:	1✔
168	return UserStatus.ACTIVE	1✔
169
170	if user["last_seen_date"] is None or user["last_seen_date"] < self.seen_limit:	1✔
171	return UserStatus.ABSENT	1✔
172
173	if (	1!
174	user["last_activity_time"] is None
175	or user["last_activity_time"] < self.activity_limit
176	):
177	return UserStatus.INACTIVE	×
178
179	return UserStatus.ACTIVE	1✔
180
181	def get_bz_users_with_status(	1✔
182	self, id_or_name: list, keep_active: bool = True
183	) -> dict:
184	"""Get Bugzilla users with their activity statuses.
185
186	Args:
187	id_or_name: An integer user ID or login name of the user on
188	bugzilla.
189	keep_active: whether the returned results should include the active
190	users.
191
192	Returns:
193	A dictionary where the key is the user login name and the value is
194	the user info with the status.
195	"""
196
197	def handler(user, data):	1✔
198	status = self.get_status_from_bz_user(user)	1✔
199	if keep_active or status != UserStatus.ACTIVE:	1✔
200	user["status"] = status	1✔
201	data[user["name"]] = user	1✔
202
203	users: dict = {}	1✔
204	BugzillaUser(	1✔
205	user_data=users,
206	user_names=id_or_name,
207	user_handler=handler,
208	include_fields=[
209	"name",
210	"can_login",
211	"last_activity_time",
212	"last_seen_date",
213	"creation_time",
214	]
215	+ self.include_fields,
216	).wait()
217
218	return users	1✔
219
220	def _get_status_from_phab_user(self, user: dict) -> Optional[UserStatus]:	1✔
221	if "disabled" in user["fields"]["roles"]:	×
222	return UserStatus.DISABLED	×
223
224	availability = user["attachments"]["availability"]	×
225	if availability["value"] != "available":	×
226	# We do not need to consider the user inactive they will be
227	# available again soon.
228	if (	×
229	not availability["until"]
230	or availability["until"] > self.availability_limit
231	):
232	return UserStatus.UNAVAILABLE	×
233
234	return None	×
235
236	def get_phab_users_with_status(	1✔
237	self, user_phids: List[str], keep_active: bool = False
238	) -> dict:
239	"""Get Phabricator users with their activity statuses.
240
241	Args:
242	user_phids: A list of user PHIDs.
243	keep_active: whether the returned results should include the active
244	users.
245
246	Returns:
247	A dictionary where the key is the user PHID and the value is
248	the user info with the status.
249	"""
250
251	bzid_to_phid = {	×
252	int(user["id"]): user["phid"]
253	for _user_phids in Connection.chunks(user_phids, PHAB_CHUNK_SIZE)
254	for user in self._fetch_bz_user_ids(user_phids=_user_phids)
255	}
256	if not bzid_to_phid:	×
257	return {}	×
258
259	if "id" not in self.include_fields:	×
260	self.include_fields.append("id")	×
261
262	user_bz_ids = list(bzid_to_phid.keys())	×
263	users = self.get_bz_users_with_status(user_bz_ids, keep_active=True)	×
264	users = {bzid_to_phid[user["id"]]: user for user in users.values()}	×
265
266	# To cover cases where a person is temporary off (e.g., long PTO), we
267	# will rely on the calendar from phab.
268	for _user_phids in Connection.chunks(user_phids, PHAB_CHUNK_SIZE):	×
269	for phab_user in self._fetch_phab_users(_user_phids):	×
270	try:	×
271	user = users[phab_user["phid"]]	×
272	phab_status = self._get_status_from_phab_user(phab_user)	×
273	if phab_status:	×
274	user["status"] = phab_status	×
275
276	elif user["status"] in (	×
277	UserStatus.ABSENT,
278	UserStatus.INACTIVE,
279	) and self.is_active_on_phab(phab_user["phid"]):
280	user["status"] = UserStatus.ACTIVE	×
281
282	if not keep_active and user["status"] == UserStatus.ACTIVE:	×
283	del users[phab_user["phid"]]	×
284	continue	×
285
286	user["phab_username"] = phab_user["fields"]["username"]	×
287	user["unavailable_until"] = phab_user["attachments"][	×
288	"availability"
289	]["until"]
NEW 290	except KeyError as e:	×
291	logging.error(	×
292	f"Error fetching inactive patch authors: '{phab_user['phid']}' - {str(e)}"
293	)
294	continue	×
295
296	return users	×
297
298	def is_active_on_phab(self, user_phid: str) -> bool:	1✔
299	"""Check if the user has recent activities on Phabricator.
300
301	Args:
302	user_phid: The user PHID.
303
304	Returns:
305	True if the user is active on Phabricator, False otherwise.
306	"""
307
308	feed = self._get_phab().request(	×
309	"feed.query",
310	filterPHIDs=[user_phid],
311	limit=1,
312	)
313	for story in feed.values():	×
314	if story["epoch"] >= self.activity_limit_ts:	×
315	return True	×
316
317	return False	×
318
319	def get_string_status(self, status: UserStatus):	1✔
320	"""Get a string representation of the user status."""
321
322	if status == UserStatus.UNDEFINED:	×
323	return "Not specified"	×
324	if status == UserStatus.DISABLED:	×
325	return "Account disabled"	×
326	if status == UserStatus.INACTIVE:	×
327	return f"Inactive on Bugzilla in last {self.activity_weeks_count} weeks"	×
328	if status == UserStatus.ABSENT:	×
329	return f"Not seen on Bugzilla in last {self.absent_weeks_count} weeks"	×
330
331	return status.name	×
332
333	@retry(	1✔
334	wait=wait_exponential(min=4),
335	stop=stop_after_attempt(5),
336	)
337	def _fetch_phab_users(self, phids: list):	1✔
338	if len(phids) == 0:	×
339	return []	×
340
341	return self._get_phab().search_users(	×
342	constraints={"phids": phids},
343	attachments={"availability": True},
344	)
345
346	@retry(	1✔
347	wait=wait_exponential(min=4),
348	stop=stop_after_attempt(5),
349	)
350	def _fetch_bz_user_ids(self, args, *kwargs):	1✔
351	return self._get_phab().load_bz_account(args, *kwargs)	×

mozilla / relman-auto-nag / #5134

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