toddlers/toddlers/plugins/packagers_without_bugzilla.py
Michal Konečný d9c03e08ce Fix formatting for for black > 23
Signed-off-by: Michal Konečný <mkonecny@redhat.com>
2023-02-03 14:33:39 +01:00

307 lines
9.8 KiB
Python

"""
This script is triggered by fedora-messaging messages published under the topic
``toddlers.trigger.packager_without_bugzilla`` and checks if all packagers
currently maintaining packages have a corresponding bugzilla account.
Authors: Pierre-Yves Chibon <pingou@pingoured.fr>
"""
import argparse
import logging
import sys
import time
import toml
try:
import tqdm
except ImportError:
tqdm = None
from ..base import ToddlerBase
from ..utils import bugzilla_system
from ..utils import fedora_account
from ..utils import notify
from ..utils.requests import make_session
_log = logging.getLogger(__name__)
def get_bugzilla_user_with_retries(user_email, cnt):
try:
return bugzilla_system.get_user(user_email=user_email)
except Exception:
if cnt == 5:
raise # pragma no cover
else:
time.sleep(2)
cnt = cnt + 1
return get_bugzilla_user_with_retries(user_email, cnt)
class PackagersWithoutBugzilla(ToddlerBase):
"""Listens to messages sent by playtime (which lives in toddlers) to checks
that all packagers have a valid bugzilla account.
"""
name = "packagers_without_bugzilla"
amqp_topics = [
"org.fedoraproject.*.toddlers.trigger.packagers_without_bugzilla",
]
def __init__(self):
self.requests_session = make_session()
self.logs = []
def accepts_topic(self, topic):
"""Returns a boolean whether this toddler is interested in messages
from this specific topic.
"""
return topic.startswith("org.fedoraproject.") and topic.endswith(
"toddlers.trigger.packagers_without_bugzilla"
)
def get_user_and_groups_dist_git(self, dist_git_url, ignorable_namespaces):
"""Returns a tuple of list containing in the first one all the users and
in the second all the groups found in dist-git in the JSON file meant to be
synced to bugzilla.
"""
dist_git_url = dist_git_url.rstrip("/")
url = f"{dist_git_url}/extras/pagure_bz.json"
req = self.requests_session.get(url)
data = req.json()
users = set()
groups = set()
for namespace in data:
if namespace in ignorable_namespaces:
continue
for package in data[namespace]:
for name in data[namespace][package]:
if name.startswith("@"):
groups.add(name[1:])
else:
users.add(name)
return (users, groups)
def process(self, config, message, send_email=True, username=None):
"""Looks for packagers/groups without bugzilla email."""
self.logs = []
try:
email_overrides = toml.load(config["email_overrides_file"])
except Exception:
print("Failed to load the file containing the email-overrides")
raise
_log.info("Setting up connection to FAS")
fedora_account.set_fasjson(config)
if not username:
# Retrieve all the packagers and groups in dist-git
_log.info("Retrieving the list of packagers and group in dist-git")
fas_packagers, fas_groups = self.get_user_and_groups_dist_git(
config["dist_git_url"],
config.get("ignorable_namespaces") or [],
)
else:
if username.startswith("@"):
fas_groups = [username[1:]]
fas_packagers = []
else:
fas_groups = []
fas_packagers = [username]
n_packagers = len(fas_packagers)
n_groups = len(fas_groups)
_log.info("%s packagers found on dist-git", n_packagers)
_log.info("%s groups found on dist-git", n_groups)
fas_packagers_info = {}
fas_groups_info = {}
_log.info("Retrieving the bugzilla email for each packager")
# If the import fails, no progress bar
# At DEBUG or below, we're showing things at each iteration so the progress
# bar doesn't look good.
# At WARNING or above, we do not want to show anything.
if (
tqdm is not None and _log.getEffectiveLevel() == logging.INFO
): # pragma no cover
fas_packagers = tqdm.tqdm(fas_packagers)
for idx, username in enumerate(sorted(fas_packagers)):
_log.debug(
" Retrieving bz email of user %s: %s/%s", username, idx, n_packagers
)
bz_email = fedora_account.get_bz_email_user(username, email_overrides)
_log.debug("%s has email: %s", username, bz_email)
if bz_email:
fas_packagers_info[bz_email] = username
else:
_log.debug(
" -> Could not find a bugzilla email associated with: %s",
username,
)
for idx, groupname in enumerate(sorted(fas_groups)):
_log.debug(
" Retrieving bz email of group %s: %s/%s", groupname, idx, n_groups
)
bz_email = fedora_account.get_bz_email_group(groupname, email_overrides)
if bz_email:
fas_groups_info[bz_email] = "@%s" % groupname
else:
_log.debug(
" -> Could not find a bugzilla email associated with: @%s",
groupname,
)
_log.info("Setting up connection to bugzilla")
bugzilla_system.set_bz(config)
# Retrieve all the packagers in bugzilla
_log.info("Retrieving the list of packagers in bugzilla")
bz_packagers = bugzilla_system.get_group_member(config["bugzilla_group"])
n_bz_packagers = len(bz_packagers)
_log.info(
"%s members of %s found in bugzilla",
n_bz_packagers,
config["bugzilla_group"],
)
fas_set = set(fas_packagers_info) | set(fas_groups_info)
bz_set = set(bz_packagers)
overlap = len(fas_set.intersection(bz_set))
fas_only = fas_set - bz_set
_log.info("%s packagers found in both places", overlap)
_log.info("%s packagers found only in FAS (to be checked)", len(fas_only))
# Store a list of user with no bugzilla account
no_bz_account = []
for user_email in sorted(fas_only):
if not get_bugzilla_user_with_retries(user_email, 0):
name = fas_packagers_info.get(user_email)
if not name:
name = fas_groups_info.get(user_email)
if name in (config.get("ignorable_accounts") or []):
continue
info = f"{name} (email: {user_email}) has no corresponding bugzilla account"
self.logs.append(info)
_log.info(info)
if send_email:
_log.info(f" Sending email to {user_email}")
notify.notify_packager(
config["mail_server"],
config["admin_email"],
username=name,
email=user_email,
)
no_bz_account.append(user_email)
_log.info("%s emails had no corresponding bugzilla account", len(no_bz_account))
if self.logs:
logs_text = "\n- ".join(self.logs)
notify.notify_admins_on_packagers_without_bugzilla_accounts(
to_addresses=[config["admin_email"]],
from_address=config["admin_email"],
mail_server=config["mail_server"],
logs_text=logs_text,
)
# We have had the situation in the past where we've had to check a specific
# account, so the following code allows to run this script stand-alone if
# needed.
def setup_logging(log_level: int):
handlers = []
_log.setLevel(log_level)
# We want all messages logged at level INFO or lower to be printed to stdout
info_handler = logging.StreamHandler(stream=sys.stdout)
handlers.append(info_handler)
if log_level == logging.INFO:
# In normal operation, don't decorate messages
for handler in handlers:
handler.setFormatter(logging.Formatter("%(message)s"))
logging.basicConfig(level=log_level, handlers=handlers)
def get_arguments(args):
"""Load and parse the CLI arguments."""
parser = argparse.ArgumentParser(
description="Checks that packagers have a valid bugzilla account"
)
parser.add_argument(
"conf",
help="Configuration file",
)
parser.add_argument(
"--send-email",
action="store_true",
dest="send_email",
default=False,
help="Notify the packager(s) about their lack of account in bugzilla.",
)
parser.add_argument(
"username",
default=None,
nargs="?",
help="Process a specific user instead of all the packagers",
)
log_level_group = parser.add_mutually_exclusive_group()
log_level_group.add_argument(
"-q",
"--quiet",
action="store_const",
dest="log_level",
const=logging.WARNING,
default=logging.INFO,
help="Be less talkative",
)
log_level_group.add_argument(
"--debug",
action="store_const",
dest="log_level",
const=logging.DEBUG,
help="Enable debugging output",
)
return parser.parse_args(args)
def main(args):
"""Schedule the first test and run the scheduler."""
args = get_arguments(args)
setup_logging(log_level=args.log_level)
config = toml.load(args.conf)
PackagersWithoutBugzilla().process(
config=config.get("consumer_config", {}).get("packagers_without_bugzilla", {}),
message={},
username=args.username,
)
if __name__ == "__main__": # pragma: no cover
try:
main(sys.argv[1:])
except KeyboardInterrupt:
pass