+ python-fedora

+ python-bugzilla>=2.4.0

+ from unittest.mock import patch, Mock

+ 

+ import pytest

+ 

+ import toddlers.plugins.packager_bugzilla_sync

+ 

+ 

+ class TestPackagerBugzillaSyncToddler:

+     def test_accepts_topic_invalid(self):

+         assert (

+             toddlers.plugins.packager_bugzilla_sync.PackagerBugzillaSync.accepts_topic(

+                 "foo.bar"

+             )

+             is False

+         )

+ 

+     @pytest.mark.parametrize(

+         "topic",

+         [

+             "org.fedoraproject.*.toddlers.trigger.packager_bugzilla_sync",

+             "org.fedoraproject.prod.toddlers.trigger.packager_bugzilla_sync",

+             "org.fedoraproject.stg.toddlers.trigger.packager_bugzilla_sync",

+         ],

+     )

+     def test_accepts_topic_valid(self, topic):

+         assert toddlers.plugins.packager_bugzilla_sync.PackagerBugzillaSync.accepts_topic(

+             topic

+         )

+ 

+     def test_process_no_email_override(self, capsys):

+         with pytest.raises(KeyError, match=r"'email_overrides_file'"):

+             toddlers.plugins.packager_bugzilla_sync.PackagerBugzillaSync.process(

+                 config={}, message=None, username=False, dry_run=True

+             )

+ 

+         out, err = capsys.readouterr()

+         assert out == "Failed to load the file containing the email-overrides\n"

+         assert err == ""

+ 

+     def test_process_no_email_override_file(self, capsys):

+         with pytest.raises(

+             FileNotFoundError, match=r"No such file or directory: 'test'"

+         ):

+             toddlers.plugins.packager_bugzilla_sync.PackagerBugzillaSync.process(

+                 config={"email_overrides_file": "test"},

+                 message=None,

+                 username=False,

+                 dry_run=True,

+             )

+ 

+         out, err = capsys.readouterr()

+         assert out == "Failed to load the file containing the email-overrides\n"

+         assert err == ""

+ 

+     @patch("toddlers.utils.fedora_account.set_fas", new=Mock(return_value=True))

+     @patch("toddlers.utils.bugzilla_system.set_bz", new=Mock(return_value=True))

+     @patch("toddlers.utils.fedora_account.get_group_member")

+     @patch("toddlers.utils.fedora_account.get_bz_email_user")

+     @patch("toddlers.utils.bugzilla_system.get_group_member")

+     @patch("toddlers.utils.bugzilla_system.add_user_to_group")

+     @patch("toml.load")

+     def test_process(

+         self, toml_load, bz_user_grp, get_bz_grp_mbr, get_bz_email, get_fas_grp_mbr

+     ):

+         toml_load.return_value = {}

+         get_fas_grp_mbr.return_value = ["pingou", "nils"]

+         get_bz_email.side_effect = ["pingou@fp.o", "nils@fp.o"]

+         get_bz_grp_mbr.return_value = ["pingou@fp.o", "nphilipp@fp.o"]

+ 

+         toddlers.plugins.packager_bugzilla_sync.PackagerBugzillaSync.process(

+             config={"email_overrides_file": "test", "bugzilla_group": "fedora_contrib"},

+             message=None,

+             username=False,

+             dry_run=False,

+         )

+ 

+         toml_load.assert_called_with("test")

+         get_fas_grp_mbr.assert_called_with("packager")

+         get_bz_email.assert_called_with("pingou", {})

+         get_bz_grp_mbr.assert_called_with("fedora_contrib")

+         bz_user_grp.assert_called_with(

+             user_email="nils@fp.o",

+             bz_group="fedora_contrib",

+             no_bz_account=[],

+             dry_run=False,

+         )

+ 

+     @patch("toddlers.utils.fedora_account.set_fas", new=Mock(return_value=True))

+     @patch("toddlers.utils.bugzilla_system.set_bz", new=Mock(return_value=True))

+     @patch("toddlers.utils.fedora_account.get_bz_email_user")

+     @patch("toddlers.utils.bugzilla_system.get_group_member")

+     @patch("toddlers.utils.bugzilla_system.add_user_to_group")

+     @patch("toml.load")

+     def test_process_username(

+         self, toml_load, bz_user_grp, get_bz_grp_mbr, get_bz_email

+     ):

+         toml_load.return_value = {}

+         get_bz_email.side_effect = ["nils@fp.o"]

+         get_bz_grp_mbr.return_value = ["pingou@fp.o"]

+ 

+         toddlers.plugins.packager_bugzilla_sync.PackagerBugzillaSync.process(

+             config={"email_overrides_file": "test", "bugzilla_group": "fedora_contrib"},

+             message=None,

+             username="nils",

+             dry_run=False,

+         )

+ 

+         toml_load.assert_called_with("test")

+         get_bz_email.assert_called_with("nils", {})

+         get_bz_grp_mbr.assert_called_with("fedora_contrib")

+         bz_user_grp.assert_called_with(

+             user_email="nils@fp.o",

+             bz_group="fedora_contrib",

+             no_bz_account=[],

+             dry_run=False,

+         )

+ 

+     def test_main_no_args(self, capsys):

+         with pytest.raises(SystemExit):

+             toddlers.plugins.packager_bugzilla_sync.main([])

+         out, err = capsys.readouterr()

+         exp = """usage: pytest [-h] [--dry-run] [-q | --debug] conf [username]

+ pytest: error: the following arguments are required: conf

+ """

+         assert out == ""

+         assert err == exp

+ 

+     @patch("toml.load", new=Mock(return_value={}))

+     def test_main_debug(self, capsys):

+         with pytest.raises(KeyError, match=r"'email_overrides_file'"):

+             toddlers.plugins.packager_bugzilla_sync.main(["test.cfg", "--debug"])

+         out, err = capsys.readouterr()

+         assert out == "Failed to load the file containing the email-overrides\n"

+         assert err == ""

+ 

+     @patch("toml.load", new=Mock(return_value={}))

+     def test_main(self, capsys):

+         with pytest.raises(KeyError, match=r"'email_overrides_file'"):

+             toddlers.plugins.packager_bugzilla_sync.main(["test.cfg"])

+         out, err = capsys.readouterr()

+         assert out == "Failed to load the file containing the email-overrides\n"

+         assert err == ""

+         "packager_bugzilla_sync",

+             "packager_bugzilla_sync",

+                 "packager_bugzilla_sync",

+         assert sorted([t.name for t in runner.toddlers]) == [

+             "flag_commit_build",

+             "packager_bugzilla_sync",

+         ]

+ import logging

+ import xmlrpc.client

+ from unittest.mock import Mock, patch

+ 

+ import pytest

+ 

+ import toddlers.utils.bugzilla_system

+ 

+ 

+ class TestBugzillaSystem:

+     def test_set_bz_no_bugzilla_url(self):

+         with pytest.raises(

+             ValueError, match=r"No bugzilla_url found in the configuration file"

+         ):

+             toddlers.utils.bugzilla_system.set_bz({})

+ 

+     def test_set_bz_no_bugzilla_username(self):

+         with pytest.raises(

+             ValueError, match=r"No bugzilla_username found in the configuration file"

+         ):

+             config = {

+                 "bugzilla_url": "https:bz.example.com",

+             }

+             toddlers.utils.bugzilla_system.set_bz(config)

+ 

+     def test_set_bz_no_bugzilla_password(self):

+         with pytest.raises(

+             ValueError, match=r"No bugzilla_password found in the configuration file"

+         ):

+             config = {

+                 "bugzilla_url": "https:bz.example.com",

+                 "bugzilla_username": "bz_username",

+             }

+             toddlers.utils.bugzilla_system.set_bz(config)

+ 

+     @patch("toddlers.utils.bugzilla_system.Bugzilla")

+     def test_set_bz(self, mock_bz):

+         mock_bz.return_value = "bugzilla_object"

+         config = {

+             "bugzilla_url": "https:bz.example.com",

+             "bugzilla_username": "bz_username",

+             "bugzilla_password": "bz_password",

+         }

+         output = toddlers.utils.bugzilla_system.set_bz(config)

+         mock_bz.assert_called_with(

+             url="https:bz.example.com/xmlrpc.cgi",

+             user="bz_username",

+             password="bz_password",

+             cookiefile=None,

+             tokenfile=None,

+         )

+         assert output == "bugzilla_object"

+ 

+     @patch("toddlers.utils.bugzilla_system._BUGZILLA", new=None)

+     def test_get_bz_not_set(self):

+         with pytest.raises(

+             ValueError, match=r"No bugzilla connection set, call set_bz first"

+         ):

+             toddlers.utils.bugzilla_system.get_bz()

+ 

+     def test_get_bz(self):

+         output = toddlers.utils.bugzilla_system.get_bz()

+         assert output == "bugzilla_object"

+ 

+     @patch("toddlers.utils.bugzilla_system.get_bz")

+     def test_get_group_member(self, mock_bz):

+         groups = Mock()

+         groups.member_emails = ["foo@bar.com", "foo@baz.com"]

+         server = Mock()

+         server.getgroup.return_value = groups

+         mock_bz.return_value = server

+ 

+         output = toddlers.utils.bugzilla_system.get_group_member("test")

+         mock_bz.assert_called_with()

+         server.getgroup.assert_called_with("test", membership=True)

+         assert output == ["foo@bar.com", "foo@baz.com"]

+ 

+     @patch("toddlers.utils.bugzilla_system.get_bz")

+     def test_remove_user_from_group(self, mock_bz):

+         server = Mock()

+         server.updateperms.return_value = True

+         mock_bz.return_value = server

+ 

+         toddlers.utils.bugzilla_system.remove_user_from_group(

+             "test@foo.com", "groupname"

+         )

+         mock_bz.assert_called_with()

+         server.updateperms.assert_called_with("test@foo.com", "rem", "groupname")

+ 

+     @patch("toddlers.utils.bugzilla_system.get_bz")

+     def test_remove_user_from_group_failed(self, mock_bz):

+         server = Mock()

+         server.updateperms.side_effect = xmlrpc.client.Fault(55, "error string")

+         mock_bz.return_value = server

+ 

+         with pytest.raises(xmlrpc.client.Fault, match=r"<Fault 55: 'error string'>"):

+             toddlers.utils.bugzilla_system.remove_user_from_group(

+                 "test@foo.com", "groupname"

+             )

+         mock_bz.assert_called_with()

+         server.updateperms.assert_called_with("test@foo.com", "rem", "groupname")

+ 

+     @patch("toddlers.utils.bugzilla_system.get_bz")

+     def test_remove_user_from_group_no_user(self, mock_bz):

+         server = Mock()

+         server.updateperms.side_effect = xmlrpc.client.Fault(51, "No such user")

+         mock_bz.return_value = server

+ 

+         toddlers.utils.bugzilla_system.remove_user_from_group(

+             "test@foo.com", "groupname"

+         )

+         mock_bz.assert_called_with()

+         server.updateperms.assert_called_with("test@foo.com", "rem", "groupname")

+ 

+     @patch("toddlers.utils.bugzilla_system.get_bz")

+     def test_user_exists(self, mock_bz):

+         server = Mock()

+         server.getuser.return_value = True

+         mock_bz.return_value = server

+ 

+         output = toddlers.utils.bugzilla_system.user_exists("test@foo.com")

+         mock_bz.assert_called_with()

+         server.getuser.assert_called_with("test@foo.com")

+         assert output

+ 

+     @patch("toddlers.utils.bugzilla_system.get_bz")

+     def test_user_exists_no_user(self, mock_bz):

+         server = Mock()

+         server.getuser.side_effect = xmlrpc.client.Fault(51, "No such user")

+         mock_bz.return_value = server

+ 

+         output = toddlers.utils.bugzilla_system.user_exists("test@foo.com")

+         mock_bz.assert_called_with()

+         server.getuser.assert_called_with("test@foo.com")

+         assert output is False

+ 

+     @patch("toddlers.utils.bugzilla_system.get_bz")

+     def test_user_exists_failed(self, mock_bz):

+         server = Mock()

+         server.getuser.side_effect = xmlrpc.client.Fault(55, "error string")

+         mock_bz.return_value = server

+ 

+         with pytest.raises(xmlrpc.client.Fault, match=r"<Fault 55: 'error string'>"):

+             toddlers.utils.bugzilla_system.user_exists("test@foo.com")

+         mock_bz.assert_called_with()

+         server.getuser.assert_called_with("test@foo.com")

+ 

+     @patch("toddlers.utils.bugzilla_system.user_exists", new=Mock(return_value=False))

+     @patch("toddlers.utils.bugzilla_system.get_bz")

+     def test_add_user_to_group_no_user(self, mock_bz):

+         server = Mock()

+         server.updateperms.return_value = True

+         mock_bz.return_value = server

+ 

+         output = toddlers.utils.bugzilla_system.add_user_to_group(

+             user_email="test@foo.com",

+             bz_group="groupname",

+             no_bz_account=[],

+             dry_run=False,

+         )

+         mock_bz.assert_called_with()

+         server.updateperms.assert_not_called()

+         assert output == ["test@foo.com"]

+ 

+     @patch("toddlers.utils.bugzilla_system.user_exists", new=Mock(return_value=True))

+     @patch("toddlers.utils.bugzilla_system.get_bz")

+     def test_add_user_to_group_dry_run(self, mock_bz, caplog):

+         caplog.set_level(logging.INFO)

+         server = Mock()

+         server.updateperms.return_value = True

+         mock_bz.return_value = server

+ 

+         output = toddlers.utils.bugzilla_system.add_user_to_group(

+             user_email="test@foo.com",

+             bz_group="groupname",

+             no_bz_account=[],

+             dry_run=True,

+         )

+         mock_bz.assert_called_with()

+         server.updateperms.assert_not_called()

+         assert output == []

+         assert (

+             caplog.records[-1].message

+             == "   Would add test@foo.com to the group groupname"

+         )

+ 

+     @patch("toddlers.utils.bugzilla_system.user_exists", new=Mock(return_value=True))

+     @patch("toddlers.utils.bugzilla_system.get_bz")

+     def test_add_user_to_group(self, mock_bz, caplog):

+         caplog.set_level(logging.INFO)

+         server = Mock()

+         server.updateperms.return_value = True

+         mock_bz.return_value = server

+ 

+         output = toddlers.utils.bugzilla_system.add_user_to_group(

+             user_email="test@foo.com",

+             bz_group="groupname",

+             no_bz_account=[],

+             dry_run=False,

+         )

+         mock_bz.assert_called_with()

+         server.updateperms.assert_called_with("test@foo.com", "add", "groupname")

+         assert output == []

+ from unittest.mock import Mock, patch

+ 

+ import pytest

+ 

+ import toddlers.utils.fedora_account

+ 

+ 

+ class TestFedoraAccount:

+     def test_set_fas_no_fas_url(self):

+         with pytest.raises(

+             ValueError, match=r"No fas_url found in the configuration file"

+         ):

+             toddlers.utils.fedora_account.set_fas({})

+ 

+     def test_set_fas_no_fas_username(self):

+         with pytest.raises(

+             ValueError, match=r"No fas_username found in the configuration file"

+         ):

+             config = {

+                 "fas_url": "https:fas.example.com",

+             }

+             toddlers.utils.fedora_account.set_fas(config)

+ 

+     def test_set_fas_no_fas_password(self):

+         with pytest.raises(

+             ValueError, match=r"No fas_password found in the configuration file"

+         ):

+             config = {

+                 "fas_url": "https:fas.example.com",

+                 "fas_username": "fas_user",

+             }

+             toddlers.utils.fedora_account.set_fas(config)

+ 

+     @patch("toddlers.utils.fedora_account.AccountSystem")

+     def test_set_fas(self, mock_fas):

+         mock_fas.return_value = "fas_object"

+         config = {

+             "fas_url": "https:fas.example.com",

+             "fas_username": "fas_user",

+             "fas_password": "fas_password",

+         }

+         output = toddlers.utils.fedora_account.set_fas(config)

+         mock_fas.assert_called_with(

+             "https:fas.example.com",

+             username="fas_user",

+             password="fas_password",

+             cache_session=False,

+         )

+         assert output == "fas_object"

+ 

+     @patch("toddlers.utils.fedora_account._FAS", new=None)

+     def test_get_fas_not_set(self):

+         with pytest.raises(

+             ValueError, match=r"No FAS connection set, call set_fas first"

+         ):

+             toddlers.utils.fedora_account.get_fas()

+ 

+     def test_get_fas(self):

+         output = toddlers.utils.fedora_account.get_fas()

+         assert output == "fas_object"

+ 

+     @patch("toddlers.utils.fedora_account.get_fas")

+     def test_get_group_member(self, mock_fas):

+         members = []

+         for name in ["pingou", "ralph", "kevin", "nils"]:

+             member = Mock()

+             member.role_type = "administrator"

+             member.username = name

+             members.append(member)

+         server = Mock()

+         server.group_members.return_value = members

+         mock_fas.return_value = server

+ 

+         output = toddlers.utils.fedora_account.get_group_member("sysadmin")

+         assert output == {"kevin", "nils", "pingou", "ralph"}

+ 

+     @patch("toddlers.utils.fedora_account.get_fas")

+     def test_get_group_member_empty(self, mock_fas):

+         members = []

+         server = Mock()

+         server.group_members.return_value = members

+         mock_fas.return_value = server

+ 

+         output = toddlers.utils.fedora_account.get_group_member("sysadmin")

+         assert output == set()

+ 

+     @patch("toddlers.utils.fedora_account.get_fas")

+     def test_get_bz_email_user_no_bugzilla_email(self, mock_fas):

+         server = Mock()

+         server.person_by_username.return_value = {}

+         mock_fas.return_value = server

+ 

+         output = toddlers.utils.fedora_account.get_bz_email_user("pingou", {})

+         assert output is None

+ 

+     @patch("toddlers.utils.fedora_account.get_fas")

+     def test_get_bz_email_user(self, mock_fas):

+         server = Mock()

+         server.person_by_username.return_value = {"bugzilla_email": "foo@bar.com"}

+         mock_fas.return_value = server

+ 

+         output = toddlers.utils.fedora_account.get_bz_email_user("pingou", {})

+         assert output == "foo@bar.com"

+ 

+     @patch("toddlers.utils.fedora_account.get_fas")

+     def test_get_bz_email_user_overriden(self, mock_fas):

+         server = Mock()

+         server.person_by_username.return_value = {"bugzilla_email": "foo@bar.com"}

+         mock_fas.return_value = server

+ 

+         output = toddlers.utils.fedora_account.get_bz_email_user(

+             "pingou", {"foo@bar.com": "foo@baz.com"}

+         )

+         assert output == "foo@baz.com"

+ 

+     @patch("toddlers.utils.fedora_account.get_fas")

+     def test_get_bz_email_group_no_bugzilla_email(self, mock_fas):

+         server = Mock()

+         server.group_by_name.return_value = {}

+         mock_fas.return_value = server

+ 

+         output = toddlers.utils.fedora_account.get_bz_email_group("toddlers-sig", {})

+         assert output is None

+ 

+     @patch("toddlers.utils.fedora_account.get_fas")

+     def test_get_bz_email_group(self, mock_fas):

+         server = Mock()

+         server.group_by_name.return_value = {"mailing_list": "foo@lists.bar.com"}

+         mock_fas.return_value = server

+ 

+         output = toddlers.utils.fedora_account.get_bz_email_group("toddlers-sig", {})

+         assert output == "foo@lists.bar.com"

+ 

+     @patch("toddlers.utils.fedora_account.get_fas")

+     def test_get_bz_email_group_overriden(self, mock_fas):

+         server = Mock()

+         server.group_by_name.return_value = {"mailing_list": "foo@lists.bar.com"}

+         mock_fas.return_value = server

+ 

+         output = toddlers.utils.fedora_account.get_bz_email_group(

+             "toddlers-sig", {"foo@lists.bar.com": "foo@baz.com"}

+         )

+         assert output == "foo@baz.com"

+ from unittest.mock import Mock, patch

+ 

+ import toddlers.utils.notify

+ 

+ 

+ class TestNotify:

+     @patch("toddlers.utils.notify.smtplib")

+     def test_notify_packager(self, mock_smtp):

+         smtp_server = Mock()

+         mock_smtp.SMTP.return_value = smtp_server

+ 

+         msg = """To: pingou@email

+ From: admin@server

+ Subject: Fedora Account System and Bugzilla Mismatch

+ 

+ Hello pingou,

+ 

+ We have identified you[1] as either a Fedora packager or someone who has asked to

+ be included in the CC list of tickets created for one or more component on

+ bugzilla. Fedora packagers are granted special permissions on the Fedora bugs in

+ bugzilla.

+ However, to enable these functionalities (granting you these permissions or

+ including you to the CC list of your packages of interest), we need to have your

+ bugzilla email address stored in the Fedora Account System[2].

+ At the moment you have:

+ 

+ pingou@email

+ 

+ which bugzilla is telling us is not an account in bugzilla.  If you could

+ please set up an account in bugzilla with this address or change your email

+ address on your Fedora Account to match an existing bugzilla account this would

+ let us go forward.

+ 

+ Note: this message is being generated by an automated script.  You'll continue

+ getting this message until the problem is resolved.  Sorry for the

+ inconvenience.

+ 

+ Thank you,

+ The Fedora Account System

+ admin@server

+ 

+ 

+ [1] the source of this information is the following JSON file:

+     https://src.fedoraproject.org/extras/pagure_bz.json

+     We are happy to tell you exactly which packages are linked to your account

+     if you wish.

+ [2] https://admin.fedoraproject.org/accounts

+ """

+ 

+         toddlers.utils.notify.notify_packager(

+             mail_server="server.mail",

+             admin_email="admin@server",

+             username="pingou",

+             email="pingou@email",

+         )

+         mock_smtp.SMTP.assert_called_with("server.mail")

+         smtp_server.sendmail.assert_called_with("admin@server", ["pingou@email"], msg)

+ 

+     @patch("toddlers.utils.notify.smtplib")

+     def test_notify_admins(self, mock_smtp):

+         users = []

+         for name, email in [

+             ("pingou", "pingou@mail"),

+             ("nils", "nils@mail"),

+             ("trasher", "trasher@mail"),

+         ]:

+             user = Mock()

+             if name == "trasher":

+                 user.person.status = "Inactive"

+             else:

+                 user.person.status = "Active"

+             user.person.username = name

+             user.person.human_name = name

+             user.email = email

+             users.append(user)

+ 

+         smtp_server = Mock()

+         mock_smtp.SMTP.return_value = smtp_server

+ 

+         msg = """To: info_admin@server

+ From: admin@server

+ Subject: Fedora Account System and Bugzilla Mismatch

+ 

+ 

+ The following people are in the packager group but do not have email addresses

+ that are valid in bugzilla:

+   pingou  --  pingou  --  pingou@mail

+   nils  --  nils  --  nils@mail

+ 

+ """

+ 

+         toddlers.utils.notify.notify_admins(

+             mail_server="server.mail",

+             admin_email="admin@server",

+             recipients=["info_admin@server"],

+             users=users,

+         )

+         mock_smtp.SMTP.assert_called_with("server.mail")

+         smtp_server.sendmail.assert_called_with(

+             "admin@server", ["info_admin@server"], msg

+         )

+ """

+ This script takes as input the fedmsg messages published under the topic

+ ``toddlers.trigger.packager_bugzilla_sync`` and runs a sync of all packagers

+ found in FAS to bugzilla so that they can do thins like editing flags on Fedora

+ tickets.

+ 

+ Authors:    Pierre-Yves Chibon <pingou@pingoured.fr>

+ 

+ """

+ 

+ import argparse

+ import logging

+ import sys

+ 

+ import requests

+ import toml

+ from requests.packages.urllib3.util import retry

+ 

+ try:

+     import tqdm

+ except ImportError:

+     tqdm = None

+ 

+ from toddlers.base import ToddlerBase

+ import toddlers.utils.fedora_account

+ import toddlers.utils.bugzilla_system

+ 

+ _log = logging.getLogger(__name__)

+ 

+ timeout = (30, 30)

+ retries = 3

+ requests_session = requests.Session()

+ retry_conf = retry.Retry(total=retries, connect=retries, read=retries, backoff_factor=1)

+ retry_conf.BACKOFF_MAX = 5

+ requests_session.mount("http://", requests.adapters.HTTPAdapter(max_retries=retry_conf))

+ requests_session.mount(

+     "https://", requests.adapters.HTTPAdapter(max_retries=retry_conf)

+ )

+ 

+ 

+ class PackagerBugzillaSync(ToddlerBase):

+     """ Listens to messages sent by playtime (which lives in toddlers) to sync

+     all the packagers found in FAS to bugzilla.

+     """

+ 

+     name = "packager_bugzilla_sync"

+ 

+     amqp_topics = [

+         "org.fedoraproject.*.toddlers.trigger.packager_bugzilla_sync",

+     ]

+ 

+     def accepts_topic(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.packager_bugzilla_sync"

+         )

+ 

+     def process(config, message, username=False, dry_run=False):

+         """Process a given message."""

+ 

+         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")

+         toddlers.utils.fedora_account.set_fas(config)

+ 

+         if not username:

+             # Retrieve all the packagers in FAS:

+             _log.info("Retrieving the list of packagers in FAS")

+             fas_packagers = sorted(

+                 toddlers.utils.fedora_account.get_group_member("packager")

+             )

+         else:

+             fas_packagers = [username]

+ 

+         n_packagers = len(fas_packagers)

+         _log.info(f"{n_packagers} packagers found in FAS")

+         fas_packagers_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(fas_packagers):

+             _log.debug(

+                 f"   Retrieving bz email of user {username}: {idx}/{n_packagers}"

+             )

+             bz_email = toddlers.utils.fedora_account.get_bz_email_user(

+                 username, email_overrides

+             )

+             fas_packagers_info[bz_email] = username

+ 

+         _log.info("Setting up connection to bugzilla")

+         toddlers.utils.bugzilla_system.set_bz(config)

+ 

+         # Retrieve all the packagers in bugzilla

+         _log.info("Retrieving the list of packagers in bugzilla")

+         bz_packagers = toddlers.utils.bugzilla_system.get_group_member(

+             config["bugzilla_group"]

+         )

+         n_bz_packagers = len(bz_packagers)

+         _log.info(

+             f"{n_bz_packagers} members of {config['bugzilla_group']} found in bugzilla"

+         )

+ 

+         fas_set = set(fas_packagers_info)

+         bz_set = set(bz_packagers)

+ 

+         overlap = len(fas_set.intersection(bz_set))

+         fas_only = fas_set - bz_set

+         bz_only = bz_set - fas_set

+ 

+         _log.info(f"{overlap} packagers found in both places")

+         _log.info(f"{len(fas_only)} packagers found only in FAS (to be added)")

+         _log.info(f"{len(bz_only)} packagers found only in BZ (to be removed)")

+ 

+         # Store a list of user with no bugzilla account

+         no_bz_account = []

+         # Add the packagers found only in FAS to the bugzilla group

+         _log.info(f"Adding to {config['bugzilla_group']} the packagers found in FAS")

+         for user_email in sorted(fas_only):

+             no_bz_account = toddlers.utils.bugzilla_system.add_user_to_group(

+                 user_email=user_email,

+                 bz_group=config["bugzilla_group"],

+                 no_bz_account=no_bz_account,

+                 dry_run=dry_run,

+             )

+ 

+         _log.info(f"{len(no_bz_account)} emails had no corresponding bugzilla account")

+ 

+ 

+ # 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 get_arguments(args):

+     """ Load and parse the CLI arguments."""

+     parser = argparse.ArgumentParser(

+         description="Sync packagers access from FAS to bugzilla"

+     )

+     parser.add_argument(

+         "conf", help="Configuration file",

+     )

+     parser.add_argument(

+         "--dry-run",

+         action="store_true",

+         dest="dry_run",

+         default=False,

+         help="Do not change anything on 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 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 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)

+     PackagerBugzillaSync.process(

+         config=config.get("consumer_config", {}).get("packager_bugzilla_sync", {}),

+         message={},

+         username=args.username,

+         dry_run=args.dry_run,

+     )

+ 

+ 

+ if __name__ == "__main__":  # pragma: no cover

+     try: