#4461

pending completion

Build # #4461

Build Type

push

coveralls-python

Committed by

suhaibmujahid

Commit Message

[needinfo_regression_author] Fetch the info for employees from Bugzilla

Run Details

640 of 3213 branches covered (19.92%)

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

1817 of 8008 relevant lines covered (22.69%)

0.23 hits per line

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

45.21

/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/.

from datetime import timedelta
from enum import Enum, auto
from typing import 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

# 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,
        phab: PhabricatorAPI = None,
        people: People = None,
    ) -> 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.
        """
        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("today") + timedelta(unavailable_max_days)
        ).timestamp()

        self.activity_limit = lmdutils.get_date("today", self.activity_weeks_count * 7)
        self.activity_limit_ts = lmdutils.get_date_ymd(self.activity_limit).timestamp()
        self.seen_limit = lmdutils.get_date("today", 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: List[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.

        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):
                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"
                ]

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

mozilla / relman-auto-nag / #4461

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