def handle_inactive_requestee()

in bugbot/rules/inactive_ni_pending.py [0:0]

• 64 lines of code
• 20 McCabe index (conditional complexity)

    def handle_inactive_requestee(self, bugs):
        """
        Detect inactive users and filter bugs to keep only the ones with needinfo pending on
        inactive users.
        """
        requestee_bugs = defaultdict(list)
        for bugid, bug in bugs.items():
            for flag in bug["needinfo_flags"]:
                if "requestee" not in flag:
                    flag["requestee"] = ""

                requestee_bugs[flag["requestee"]].append(bugid)

        user_activity = UserActivity(include_fields=["groups", "creation_time"])
        needinfo_requestees = set(requestee_bugs.keys())
        triage_owners = {bug["triage_owner"] for bug in bugs.values()}
        inactive_users = user_activity.check_users(
            needinfo_requestees | triage_owners, ignore_bots=True
        )

        inactive_requestee_bugs = {
            bugid
            for requestee, bugids in requestee_bugs.items()
            if requestee in inactive_users
            for bugid in bugids
        }

        def has_canconfirm_group(user_email):
            for group in inactive_users[user_email].get("groups", []):
                if group["name"] == "canconfirm":
                    return True
            return False

        def get_inactive_ni(bug):
            return [
                {
                    "id": flag["id"],
                    "setter": flag["setter"],
                    "requestee": flag["requestee"],
                    "requestee_status": user_activity.get_string_status(
                        inactive_users[flag["requestee"]]["status"],
                    ),
                    "requestee_canconfirm": has_canconfirm_group(flag["requestee"]),
                }
                for flag in bug["needinfo_flags"]
                if flag["requestee"] in inactive_users
                and (
                    # Exclude recent needinfos to allow some time for external
                    # users to respond.
                    flag["modification_date"] < RECENT_NEEDINFO_LIMIT
                    or inactive_users[flag["requestee"]]["status"]
                    in [UserStatus.DISABLED, UserStatus.UNDEFINED]
                )
            ]

        res = {}
        skiplist = self.get_auto_ni_skiplist()
        for bugid, bug in bugs.items():
            if (
                bugid not in inactive_requestee_bugs
                or bug["triage_owner"] in inactive_users
                or bug["triage_owner"] in skiplist
            ):
                continue

            inactive_ni = get_inactive_ni(bug)
            if len(inactive_ni) == 0:
                continue

            bug = {
                **bug,
                "inactive_ni": inactive_ni,
                "inactive_ni_count": len(inactive_ni),
                "action": self.get_action_type(bug, inactive_ni),
            }
            res[bugid] = bug
            self.add_action(bug)

        return res