def update_bugzilla_emails()

in bugbot/iam.py [0:0]

• 40 lines of code
• 14 McCabe index (conditional complexity)

def update_bugzilla_emails(data: Dict[str, dict]) -> None:
    """Update the bugzilla emails based on the Bugzilla ID and the Mozilla LDAP email.

    Args:
        data: The data to update.
    """

    users_by_bugzilla_id = {
        int(person["bugzillaID"]): person
        for person in data.values()
        if person["bugzillaID"]
    }

    # Currently employees can have permissions if they use their Mozilla email
    # without the need to link their Bugzilla accounts to PMO. Thus we check here
    # if employees already have a Bugzilla account using their Mozilla emails.
    #
    # Once BMO and PMO are fully integrated (plan in progress), this will be
    # changed and employees will not have permissions unless they link their
    # Bugzilla account to PMO.
    users_to_check = [*data, *users_by_bugzilla_id]

    def handler(bz_user, data):
        if bz_user["id"] in users_by_bugzilla_id:
            person = users_by_bugzilla_id[bz_user["id"]]
        elif bz_user["name"].lower() in data:
            person = data[bz_user["name"].lower()]
        else:
            raise Exception(f"Can't find {bz_user['name']} in the data")

        if (
            person.get("found_on_bugzilla")
            and str(bz_user["id"]) != person["bugzillaID"]
        ):
            # If the linked Bugzilla account is still active, we should not
            # overwrite it with the other account.
            return

        person["found_on_bugzilla"] = True
        if person["bugzillaEmail"] != bz_user["name"]:
            logger.info(
                "Update bugzilla email for %s from '%s' to '%s'",
                person["cn"],
                person["bugzillaEmail"],
                bz_user["name"],
            )
            person["bugzillaEmail"] = bz_user["name"]

    def fault_user_handler(bz_user, data):
        logger.debug("Can't find %s on bugzilla", bz_user["name"])

    BugzillaUser(
        users_to_check,
        include_fields=["id", "name"],
        user_handler=handler,
        user_data=data,
        fault_user_handler=fault_user_handler,
    ).wait()

    for person in data.values():
        if "found_on_bugzilla" not in person:
            person["found_on_bugzilla"] = False