#4579

pending completion

Build # #4579

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

[user_activity] Use a fixed reference date in the tests (#2115)

Run Details

641 of 3217 branches covered (19.93%)

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

1823 of 8002 relevant lines covered (22.78%)

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

mozilla / relman-auto-nag / #4579

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