• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

mozilla / relman-auto-nag / #5288

24 Oct 2024 04:16PM CUT coverage: 21.688% (+0.02%) from 21.668%
#5288

push

coveralls-python

web-flow
[user_activity] Removed `user_creation_time` parameter in `get_string_status()` (#2518)

426 of 2872 branches covered (14.83%)

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

1 existing line in 1 file now uncovered.

1943 of 8959 relevant lines covered (21.69%)

0.22 hits per line

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

56.39
/bugbot/user_activity.py
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 Iterable, 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
    INACTIVE_NEW = auto()
1✔
31
    ABSENT_NEW = auto()
1✔
32

33

34
class UserActivity:
1✔
35
    """Check the user activity on Bugzilla and Phabricator"""
36

37
    def __init__(
1✔
38
        self,
39
        activity_weeks_count: int = 26,
40
        absent_weeks_count: int = 26,
41
        new_user_weeks_count: int = 4,
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
            new_user_weeks_count: the number of weeks since last made a change
57
                to a bug before a new user being considered as inactive.
58
            unavailable_max_days: a user will be considered inactive if they
59
                have more days left to be available than `unavailable_max_days`.
60
            include_fields: the list of fields to include with the the Bugzilla
61
                user object.
62
            phab: if an instance of PhabricatorAPI is not provided, it will be
63
                created when it is needed.
64
            people: if an instance of People is not provided, the global
65
                instance will be used.
66
            reference_date: the reference date to use for checking user
67
                activity. This is needed for testing because the dates in the
68
                mock data are fixed.
69
        """
70
        self.activity_weeks_count = activity_weeks_count
1✔
71
        self.absent_weeks_count = absent_weeks_count
1✔
72
        self.new_user_weeks_count = new_user_weeks_count
1✔
73
        self.include_fields = include_fields or []
1✔
74
        self.people = people if people is not None else People.get_instance()
1✔
75
        self.phab = phab
1✔
76
        self.availability_limit = (
1✔
77
            lmdutils.get_date_ymd(reference_date) + timedelta(unavailable_max_days)
78
        ).timestamp()
79

80
        self.activity_limit = lmdutils.get_date(
1✔
81
            reference_date, self.activity_weeks_count * 7
82
        )
83
        self.activity_limit_ts = lmdutils.get_date_ymd(self.activity_limit).timestamp()
1✔
84
        self.seen_limit = lmdutils.get_date(reference_date, self.absent_weeks_count * 7)
1✔
85

86
        self.new_user_activity_limit = lmdutils.get_date(
1✔
87
            reference_date, self.new_user_weeks_count * 7
88
        )
89
        self.new_user_activity_limit_ts = lmdutils.get_date_ymd(
1✔
90
            self.new_user_activity_limit
91
        ).timestamp()
92
        self.new_user_seen_limit = lmdutils.get_date(
1✔
93
            reference_date, self.new_user_weeks_count * 7
94
        )
95

96
        # Bugzilla accounts younger than 61 days are considered new users
97
        self.new_user_limit = lmdutils.get_date(reference_date, 61)
1✔
98

99
    def _get_phab(self):
1✔
100
        if not self.phab:
×
101
            self.phab = PhabricatorAPI(utils.get_login_info()["phab_api_key"])
×
102

103
        return self.phab
×
104

105
    def check_users(
1✔
106
        self,
107
        user_emails: Iterable[str],
108
        keep_active: bool = False,
109
        ignore_bots: bool = False,
110
        fetch_employee_info: bool = False,
111
    ) -> dict:
112
        """Check user activity using their emails
113

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

124
        Returns:
125
            A dictionary where the key is the user email and the value is the
126
                user info with the status.
127
        """
128

129
        user_statuses = {
1✔
130
            user_email: {
131
                "status": (
132
                    UserStatus.UNDEFINED
133
                    if utils.is_no_assignee(user_email)
134
                    else UserStatus.ACTIVE
135
                ),
136
                "is_employee": self.people.is_mozilla(user_email),
137
            }
138
            for user_email in user_emails
139
            if not ignore_bots or not utils.is_bot_email(user_email)
140
        }
141

142
        # Employees will always be considered active
143
        user_emails = [
1✔
144
            user_email
145
            for user_email, info in user_statuses.items()
146
            if not info["is_employee"] and info["status"] == UserStatus.ACTIVE
147
        ]
148

149
        if not keep_active:
1✔
150
            user_statuses = {
1✔
151
                user_email: info
152
                for user_email, info in user_statuses.items()
153
                if info["status"] != UserStatus.ACTIVE
154
            }
155

156
        if fetch_employee_info:
1!
157
            employee_emails = [
×
158
                user_email
159
                for user_email, info in user_statuses.items()
160
                if info["is_employee"]
161
            ]
162
            if employee_emails:
×
163
                BugzillaUser(
×
164
                    user_names=employee_emails,
165
                    user_data=user_statuses,
166
                    user_handler=lambda user, data: data[user["name"]].update(user),
167
                    include_fields=self.include_fields + ["name"],
168
                ).wait()
169

170
        if user_emails:
1!
171
            user_statuses.update(
1✔
172
                self.get_bz_users_with_status(user_emails, keep_active)
173
            )
174

175
        return user_statuses
1✔
176

177
    def get_status_from_bz_user(self, user: dict) -> UserStatus:
1✔
178
        """Get the user status from a Bugzilla user object."""
179
        is_new_user = user["creation_time"] > self.new_user_limit
1✔
180

181
        seen_limit = self.seen_limit if not is_new_user else self.new_user_seen_limit
1✔
182
        activity_limit = (
1✔
183
            self.activity_limit if not is_new_user else self.new_user_activity_limit
184
        )
185

186
        if not user["can_login"]:
1✔
187
            return UserStatus.DISABLED
1✔
188

189
        if user["creation_time"] > seen_limit:
1✔
190
            return UserStatus.ACTIVE
1✔
191

192
        if user["last_seen_date"] is None or user["last_seen_date"] < seen_limit:
1✔
193
            return UserStatus.ABSENT_NEW if is_new_user else UserStatus.ABSENT
1✔
194

195
        if (
1!
196
            user["last_activity_time"] is None
197
            or user["last_activity_time"] < activity_limit
198
        ):
NEW
199
            return UserStatus.INACTIVE_NEW if is_new_user else UserStatus.INACTIVE
×
200

201
        return UserStatus.ACTIVE
1✔
202

203
    def get_bz_users_with_status(
1✔
204
        self, id_or_name: list, keep_active: bool = True
205
    ) -> dict:
206
        """Get Bugzilla users with their activity statuses.
207

208
        Args:
209
            id_or_name: An integer user ID or login name of the user on
210
                bugzilla.
211
            keep_active: whether the returned results should include the active
212
                users.
213

214
        Returns:
215
            A dictionary where the key is the user login name and the value is
216
            the user info with the status.
217
        """
218

219
        def handler(user, data):
1✔
220
            status = self.get_status_from_bz_user(user)
1✔
221
            if keep_active or status != UserStatus.ACTIVE:
1✔
222
                user["status"] = status
1✔
223
                data[user["name"]] = user
1✔
224

225
        users: dict = {}
1✔
226
        BugzillaUser(
1✔
227
            user_data=users,
228
            user_names=id_or_name,
229
            user_handler=handler,
230
            include_fields=[
231
                "name",
232
                "can_login",
233
                "last_activity_time",
234
                "last_seen_date",
235
                "creation_time",
236
            ]
237
            + self.include_fields,
238
        ).wait()
239

240
        return users
1✔
241

242
    def _get_status_from_phab_user(self, user: dict) -> Optional[UserStatus]:
1✔
243
        if "disabled" in user["fields"]["roles"]:
×
244
            return UserStatus.DISABLED
×
245

246
        availability = user["attachments"]["availability"]
×
247
        if availability["value"] != "available":
×
248
            # We do not need to consider the user inactive they will be
249
            # available again soon.
250
            if (
×
251
                not availability["until"]
252
                or availability["until"] > self.availability_limit
253
            ):
254
                return UserStatus.UNAVAILABLE
×
255

256
        return None
×
257

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

263
        Args:
264
            user_phids: A list of user PHIDs.
265
            keep_active: whether the returned results should include the active
266
                users.
267

268
        Returns:
269
            A dictionary where the key is the user PHID and the value is
270
            the user info with the status.
271
        """
272

273
        bzid_to_phid = {
×
274
            int(user["id"]): user["phid"]
275
            for _user_phids in Connection.chunks(user_phids, PHAB_CHUNK_SIZE)
276
            for user in self._fetch_bz_user_ids(user_phids=_user_phids)
277
        }
278
        if not bzid_to_phid:
×
279
            return {}
×
280

281
        if "id" not in self.include_fields:
×
282
            self.include_fields.append("id")
×
283

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

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

297
                elif user["status"] in (
×
298
                    UserStatus.ABSENT,
299
                    UserStatus.INACTIVE,
300
                ) and self.is_active_on_phab(phab_user["phid"]):
301
                    user["status"] = UserStatus.ACTIVE
×
302

303
                if not keep_active and user["status"] == UserStatus.ACTIVE:
×
304
                    del users[phab_user["phid"]]
×
305
                    continue
×
306

307
                user["phab_username"] = phab_user["fields"]["username"]
×
308
                user["unavailable_until"] = phab_user["attachments"]["availability"][
×
309
                    "until"
310
                ]
311

312
        return users
×
313

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

317
        Args:
318
            user_phid: The user PHID.
319

320
        Returns:
321
            True if the user is active on Phabricator, False otherwise.
322
        """
323

324
        feed = self._get_phab().request(
×
325
            "feed.query",
326
            filterPHIDs=[user_phid],
327
            limit=1,
328
        )
329
        for story in feed.values():
×
330
            if story["epoch"] >= self.activity_limit_ts:
×
331
                return True
×
332

333
        return False
×
334

335
    def get_string_status(self, status: UserStatus):
1✔
336
        """Get a string representation of the user status."""
UNCOV
337
        if status == UserStatus.UNDEFINED:
×
338
            return "Not specified"
×
339
        if status == UserStatus.DISABLED:
×
340
            return "Account disabled"
×
341
        if status == UserStatus.INACTIVE:
×
342
            return f"Inactive on Bugzilla in last {self.activity_weeks_count} weeks"
×
343
        if status == UserStatus.ABSENT:
×
344
            return f"Not seen on Bugzilla in last {self.absent_weeks_count} weeks"
×
NEW
345
        if status == UserStatus.INACTIVE_NEW:
×
NEW
346
            return f"Inactive on Bugzilla in last {self.new_user_weeks_count} weeks (new user)"
×
NEW
347
        if status == UserStatus.ABSENT_NEW:
×
NEW
348
            return f"Not seen on Bugzilla in last {self.new_user_weeks_count} weeks (new user)"
×
349

350
        return status.name
×
351

352
    @retry(
1✔
353
        wait=wait_exponential(min=4),
354
        stop=stop_after_attempt(5),
355
    )
356
    def _fetch_phab_users(self, phids: list):
1✔
357
        if len(phids) == 0:
×
358
            return []
×
359

360
        return self._get_phab().search_users(
×
361
            constraints={"phids": phids},
362
            attachments={"availability": True},
363
        )
364

365
    @retry(
1✔
366
        wait=wait_exponential(min=4),
367
        stop=stop_after_attempt(5),
368
    )
369
    def _fetch_bz_user_ids(self, *args, **kwargs):
1✔
370
        return self._get_phab().load_bz_account(*args, **kwargs)
×
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2025 Coveralls, Inc