#4463

pending completion

Build # #4463

Build Type

push

coveralls-python

Committed by

suhaibmujahid

Commit Message

Update the method docstrings to include the new parameter

Run Details

640 of 3213 branches covered (19.92%)

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.
            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):
                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	fetch_employee_info: whether to fetch the employee info from
96	Bugzilla. Only fields specified in `include_fields` will be
97	guaranteed to be fetched.
98
99	Returns:
100	A dictionary where the key is the user email and the value is the
101	user info with the status.
102	"""
103
104	user_statuses = {	1✔
105	user_email: {
106	"status": (
107	UserStatus.UNDEFINED
108	if utils.is_no_assignee(user_email)
109	else UserStatus.ACTIVE
110	),
111	"is_employee": self.people.is_mozilla(user_email),
112	}
113	for user_email in user_emails
114	if not ignore_bots or not utils.is_bot_email(user_email)
115	}
116
117	# Employees will always be considered active
118	user_emails = [	1✔
119	user_email
120	for user_email, info in user_statuses.items()
121	if not info["is_employee"] and info["status"] == UserStatus.ACTIVE
122	]
123
124	if not keep_active:	1✔
125	user_statuses = {	1✔
126	user_email: info
127	for user_email, info in user_statuses.items()
128	if info["status"] != UserStatus.ACTIVE
129	}
130
131	if fetch_employee_info:	1!
132	employee_emails = [	×
133	user_email
134	for user_email, info in user_statuses.items()
135	if info["is_employee"]
136	]
137	if employee_emails:	×
138	BugzillaUser(	×
139	user_names=employee_emails,
140	user_data=user_statuses,
141	user_handler=lambda user, data: data[user["name"]].update(user),
142	include_fields=self.include_fields + ["name"],
143	).wait()
144
145	if user_emails:	1!
146	user_statuses.update(	1✔
147	self.get_bz_users_with_status(user_emails, keep_active)
148	)
149
150	return user_statuses	1✔
151
152	def get_status_from_bz_user(self, user: dict) -> UserStatus:	1✔
153	"""Get the user status from a Bugzilla user object."""
154
155	if not user["can_login"]:	1✔
156	return UserStatus.DISABLED	1✔
157
158	if user["creation_time"] > self.seen_limit:	1✔
159	return UserStatus.ACTIVE	1✔
160
161	if user["last_seen_date"] is None or user["last_seen_date"] < self.seen_limit:	1✔
162	return UserStatus.ABSENT	1✔
163
164	if (	1!
165	user["last_activity_time"] is None
166	or user["last_activity_time"] < self.activity_limit
167	):
168	return UserStatus.INACTIVE	×
169
170	return UserStatus.ACTIVE	1✔
171
172	def get_bz_users_with_status(	1✔
173	self, id_or_name: list, keep_active: bool = True
174	) -> dict:
175	"""Get Bugzilla users with their activity statuses.
176
177	Args:
178	id_or_name: An integer user ID or login name of the user on
179	bugzilla.
180	keep_active: whether the returned results should include the active
181	users.
182
183	Returns:
184	A dictionary where the key is the user login name and the value is
185	the user info with the status.
186	"""
187
188	def handler(user, data):	1✔
189	status = self.get_status_from_bz_user(user)	1✔
190	if keep_active or status != UserStatus.ACTIVE:	1✔
191	user["status"] = status	1✔
192	data[user["name"]] = user	1✔
193
194	users: dict = {}	1✔
195	BugzillaUser(	1✔
196	user_data=users,
197	user_names=id_or_name,
198	user_handler=handler,
199	include_fields=[
200	"name",
201	"can_login",
202	"last_activity_time",
203	"last_seen_date",
204	"creation_time",
205	]
206	+ self.include_fields,
207	).wait()
208
209	return users	1✔
210
211	def _get_status_from_phab_user(self, user: dict) -> Optional[UserStatus]:	1✔
212	if "disabled" in user["fields"]["roles"]:	×
213	return UserStatus.DISABLED	×
214
215	availability = user["attachments"]["availability"]	×
216	if availability["value"] != "available":	×
217	# We do not need to consider the user inactive they will be
218	# available again soon.
219	if (	×
220	not availability["until"]
221	or availability["until"] > self.availability_limit
222	):
223	return UserStatus.UNAVAILABLE	×
224
225	return None	×
226
227	def get_phab_users_with_status(	1✔
228	self, user_phids: List[str], keep_active: bool = False
229	) -> dict:
230	"""Get Phabricator users with their activity statuses.
231
232	Args:
233	user_phids: A list of user PHIDs.
234	keep_active: whether the returned results should include the active
235	users.
236
237	Returns:
238	A dictionary where the key is the user PHID and the value is
239	the user info with the status.
240	"""
241
242	bzid_to_phid = {	×
243	int(user["id"]): user["phid"]
244	for _user_phids in Connection.chunks(user_phids, PHAB_CHUNK_SIZE)
245	for user in self._fetch_bz_user_ids(user_phids=_user_phids)
246	}
247	if not bzid_to_phid:	×
248	return {}	×
249
250	if "id" not in self.include_fields:	×
251	self.include_fields.append("id")	×
252
253	user_bz_ids = list(bzid_to_phid.keys())	×
254	users = self.get_bz_users_with_status(user_bz_ids, keep_active=True)	×
255	users = {bzid_to_phid[user["id"]]: user for user in users.values()}	×
256
257	# To cover cases where a person is temporary off (e.g., long PTO), we
258	# will rely on the calendar from phab.
259	for _user_phids in Connection.chunks(user_phids, PHAB_CHUNK_SIZE):	×
260	for phab_user in self._fetch_phab_users(_user_phids):	×
261	user = users[phab_user["phid"]]	×
262	phab_status = self._get_status_from_phab_user(phab_user)	×
263	if phab_status:	×
264	user["status"] = phab_status	×
265
266	elif user["status"] in (	×
267	UserStatus.ABSENT,
268	UserStatus.INACTIVE,
269	) and self.is_active_on_phab(phab_user["phid"]):
270	user["status"] = UserStatus.ACTIVE	×
271
272	if not keep_active and user["status"] == UserStatus.ACTIVE:	×
273	del users[phab_user["phid"]]	×
274	continue	×
275
276	user["phab_username"] = phab_user["fields"]["username"]	×
277	user["unavailable_until"] = phab_user["attachments"]["availability"][	×
278	"until"
279	]
280
281	return users	×
282
283	def is_active_on_phab(self, user_phid: str) -> bool:	1✔
284	"""Check if the user has recent activities on Phabricator.
285
286	Args:
287	user_phid: The user PHID.
288
289	Returns:
290	True if the user is active on Phabricator, False otherwise.
291	"""
292
293	feed = self._get_phab().request(	×
294	"feed.query",
295	filterPHIDs=[user_phid],
296	limit=1,
297	)
298	for story in feed.values():	×
299	if story["epoch"] >= self.activity_limit_ts:	×
300	return True	×
301
302	return False	×
303
304	def get_string_status(self, status: UserStatus):	1✔
305	"""Get a string representation of the user status."""
306
307	if status == UserStatus.UNDEFINED:	×
308	return "Not specified"	×
309	if status == UserStatus.DISABLED:	×
310	return "Account disabled"	×
311	if status == UserStatus.INACTIVE:	×
312	return f"Inactive on Bugzilla in last {self.activity_weeks_count} weeks"	×
313	if status == UserStatus.ABSENT:	×
314	return f"Not seen on Bugzilla in last {self.absent_weeks_count} weeks"	×
315
316	return status.name	×
317
318	@retry(	1✔
319	wait=wait_exponential(min=4),
320	stop=stop_after_attempt(5),
321	)
322	def _fetch_phab_users(self, phids: list):	1✔
323	if len(phids) == 0:	×
324	return []	×
325
326	return self._get_phab().search_users(	×
327	constraints={"phids": phids},
328	attachments={"availability": True},
329	)
330
331	@retry(	1✔
332	wait=wait_exponential(min=4),
333	stop=stop_after_attempt(5),
334	)
335	def _fetch_bz_user_ids(self, args, *kwargs):	1✔
336	return self._get_phab().load_bz_account(args, *kwargs)	×

mozilla / relman-auto-nag / #4463

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