| |
@@ -0,0 +1,409 @@
|
| |
+ """
|
| |
+ Unit tests for `toddlers.plugins.scm_request_processor`
|
| |
+ """
|
| |
+
|
| |
+ import logging
|
| |
+ from unittest.mock import Mock, patch
|
| |
+
|
| |
+ from fedora_messaging.api import Message
|
| |
+ from noggin_messages import MemberSponsorV1
|
| |
+ import pytest
|
| |
+
|
| |
+ from toddlers.exceptions import PagureError
|
| |
+ from toddlers.plugins import pagure_fas_groups_sync
|
| |
+
|
| |
+
|
| |
+ class TestAcceptsTopic:
|
| |
+ """
|
| |
+ Test class for `toddlers.plugins.pagure_fas_groups_sync.PagureFASGroupsSync.accepts_topic`
|
| |
+ method.
|
| |
+ """
|
| |
+
|
| |
+ toddler_cls = pagure_fas_groups_sync.PagureFASGroupsSync
|
| |
+
|
| |
+ def test_accepts_topic_invalid(self, toddler):
|
| |
+ """
|
| |
+ Assert that invalid topic is not accepted.
|
| |
+ """
|
| |
+ assert toddler.accepts_topic("foo.bar") is False
|
| |
+
|
| |
+ @pytest.mark.parametrize(
|
| |
+ "topic",
|
| |
+ [
|
| |
+ "org.fedoraproject.*.fas.group.member.sponsor",
|
| |
+ "org.fedoraproject.*.toddlers.trigger.pagure_fas_groups_sync",
|
| |
+ "org.fedoraproject.stg.fas.group.member.sponsor",
|
| |
+ "org.fedoraproject.stg.toddlers.trigger.pagure_fas_groups_sync",
|
| |
+ "org.fedoraproject.prod.fas.group.member.sponsor",
|
| |
+ "org.fedoraproject.prod.toddlers.trigger.pagure_fas_groups_sync",
|
| |
+ ],
|
| |
+ )
|
| |
+ def test_accepts_topic_valid(self, topic, toddler):
|
| |
+ """
|
| |
+ Assert that valid topics are accepted.
|
| |
+ """
|
| |
+ assert toddler.accepts_topic(topic)
|
| |
+
|
| |
+
|
| |
+ class TestProcess:
|
| |
+ """
|
| |
+ Test class for `toddlers.plugins.pagure_fas_groups_sync.PagureFASGroupsSync.process` method.
|
| |
+ """
|
| |
+
|
| |
+ toddler_cls = pagure_fas_groups_sync.PagureFASGroupsSync
|
| |
+
|
| |
+ @patch("toddlers.utils.pagure.set_pagure")
|
| |
+ @patch("toddlers.utils.fedora_account.set_fasjson")
|
| |
+ def test_process_trigger_message(self, mock_fas, mock_pagure, toddler):
|
| |
+ """
|
| |
+ Assert that trigger message is processed correctly.
|
| |
+ """
|
| |
+ # Preparation
|
| |
+ config = {"group_map": {"fas_group": "pagure_group"}}
|
| |
+ msg = Message()
|
| |
+ msg.topic = "org.fedoraproject.prod.toddlers.trigger.pagure_fas_groups_sync"
|
| |
+
|
| |
+ # Test
|
| |
+ with patch(
|
| |
+ "toddlers.plugins.pagure_fas_groups_sync.PagureFASGroupsSync.sync_group"
|
| |
+ ) as mock_sync_group:
|
| |
+ toddler.process(config, msg)
|
| |
+
|
| |
+ # Assertions
|
| |
+ mock_sync_group.assert_called_with("fas_group", "pagure_group")
|
| |
+
|
| |
+ mock_fas.assert_called_with(config)
|
| |
+ mock_pagure.assert_called_with(config)
|
| |
+
|
| |
+ @patch("toddlers.utils.pagure.set_pagure")
|
| |
+ @patch("toddlers.utils.fedora_account.set_fasjson")
|
| |
+ def test_process_sponsor_message(self, mock_fas, mock_pagure, toddler):
|
| |
+ """
|
| |
+ Assert that sponsor message is processed correctly.
|
| |
+ """
|
| |
+ # Preparation
|
| |
+ mock_pagure_obj = Mock()
|
| |
+ mock_pagure.return_value = mock_pagure_obj
|
| |
+ config = {"group_map": {"fas_group": "pagure_group"}}
|
| |
+ msg = MemberSponsorV1(
|
| |
+ {"msg": {"agent": "agent", "user": "user", "group": "fas_group"}}
|
| |
+ )
|
| |
+
|
| |
+ # Test
|
| |
+ toddler.process(config, msg)
|
| |
+
|
| |
+ # Assertions
|
| |
+ mock_pagure_obj.add_member_to_group.assert_called_with("user", "pagure_group")
|
| |
+
|
| |
+ mock_fas.assert_called_with(config)
|
| |
+ mock_pagure.assert_called_with(config)
|
| |
+
|
| |
+ @patch("toddlers.utils.pagure.set_pagure")
|
| |
+ @patch("toddlers.utils.fedora_account.set_fasjson")
|
| |
+ def test_process_sponsor_message_user_not_in_pagure(
|
| |
+ self, mock_fas, mock_pagure, toddler
|
| |
+ ):
|
| |
+ """
|
| |
+ Assert that sponsor message is processed correctly when user
|
| |
+ doesn't exist in pagure.
|
| |
+ """
|
| |
+ # Preparation
|
| |
+ mock_pagure_obj = Mock()
|
| |
+ mock_pagure_obj.user_exists.return_value = False
|
| |
+ mock_pagure.return_value = mock_pagure_obj
|
| |
+ config = {"group_map": {"fas_group": "pagure_group"}}
|
| |
+ msg = MemberSponsorV1(
|
| |
+ {"msg": {"agent": "agent", "user": "user", "group": "fas_group"}}
|
| |
+ )
|
| |
+
|
| |
+ # Test
|
| |
+ toddler.process(config, msg)
|
| |
+
|
| |
+ # Assertions
|
| |
+ mock_pagure_obj.add_member_to_group.assert_not_called()
|
| |
+
|
| |
+ mock_fas.assert_called_with(config)
|
| |
+ mock_pagure.assert_called_with(config)
|
| |
+
|
| |
+ @patch("toddlers.utils.pagure.set_pagure")
|
| |
+ @patch("toddlers.utils.fedora_account.set_fasjson")
|
| |
+ def test_process_sponsor_message_failure(
|
| |
+ self, mock_fas, mock_pagure, caplog, toddler
|
| |
+ ):
|
| |
+ """
|
| |
+ Assert that failure during processing sponsor message is handled correctly.
|
| |
+ """
|
| |
+ # Preparation
|
| |
+ caplog.set_level(logging.ERROR)
|
| |
+ mock_pagure_obj = Mock()
|
| |
+ mock_pagure_obj.add_member_to_group.side_effect = PagureError("PagureError")
|
| |
+ mock_pagure.return_value = mock_pagure_obj
|
| |
+ config = {"group_map": {"fas_group": "pagure_group"}}
|
| |
+ msg = MemberSponsorV1(
|
| |
+ {"msg": {"agent": "agent", "user": "user", "group": "fas_group"}}
|
| |
+ )
|
| |
+
|
| |
+ # Test
|
| |
+ toddler.process(config, msg)
|
| |
+
|
| |
+ # Assertions
|
| |
+ mock_pagure_obj.add_member_to_group.assert_called_with("user", "pagure_group")
|
| |
+
|
| |
+ mock_fas.assert_called_with(config)
|
| |
+ mock_pagure.assert_called_with(config)
|
| |
+
|
| |
+ assert caplog.records[-1].message == "PagureError"
|
| |
+
|
| |
+ @patch("toddlers.utils.pagure.set_pagure")
|
| |
+ @patch("toddlers.utils.fedora_account.set_fasjson")
|
| |
+ def test_process_sponsor_message_skipping(
|
| |
+ self, mock_fas, mock_pagure, caplog, toddler
|
| |
+ ):
|
| |
+ """
|
| |
+ Assert that group is skipped when we don't care about it.
|
| |
+ """
|
| |
+ # Preparation
|
| |
+ caplog.set_level(logging.INFO)
|
| |
+ mock_pagure_obj = Mock()
|
| |
+ mock_pagure.return_value = mock_pagure_obj
|
| |
+ config = {"group_map": {"fas_group": "pagure_group"}}
|
| |
+ msg = MemberSponsorV1(
|
| |
+ {"msg": {"agent": "agent", "user": "user", "group": "some_group"}}
|
| |
+ )
|
| |
+
|
| |
+ # Test
|
| |
+ toddler.process(config, msg)
|
| |
+
|
| |
+ # Assertions
|
| |
+ mock_pagure_obj.add_member_to_group.assert_not_called()
|
| |
+
|
| |
+ mock_fas.assert_called_with(config)
|
| |
+ mock_pagure.assert_called_with(config)
|
| |
+
|
| |
+ assert caplog.records[-1].message.startswith(
|
| |
+ "User 'user' was added to group 'some_group'"
|
| |
+ )
|
| |
+
|
| |
+
|
| |
+ class TestSyncGroup:
|
| |
+ """
|
| |
+ Test class for `toddlers.plugins.pagure_fas_groups_sync.sync_group` function.
|
| |
+ """
|
| |
+
|
| |
+ def setup_method(self):
|
| |
+ """Prepare the toddler object."""
|
| |
+ self.sync_object = pagure_fas_groups_sync.PagureFASGroupsSync()
|
| |
+ self.sync_object.pagure = Mock()
|
| |
+
|
| |
+ @patch("toddlers.plugins.pagure_fas_groups_sync.fedora_account")
|
| |
+ def test_sync_group_add_member(self, mock_fas):
|
| |
+ """Assert that adding member to group works as intended."""
|
| |
+ # Preparation
|
| |
+ fas_group = "fas_group"
|
| |
+ pagure_group = "pagure_group"
|
| |
+ self.sync_object.pagure.get_group_members.return_value = ["user1"]
|
| |
+ mock_fas.get_group_member.return_value = ["user1", "user2"]
|
| |
+
|
| |
+ # Test
|
| |
+ self.sync_object.sync_group(fas_group, pagure_group)
|
| |
+
|
| |
+ # Assertions
|
| |
+ mock_fas.get_group_member.assert_called_with(fas_group)
|
| |
+ self.sync_object.pagure.get_group_members.assert_called_with(pagure_group)
|
| |
+
|
| |
+ self.sync_object.pagure.add_member_to_group.assert_called_with(
|
| |
+ "user2", pagure_group
|
| |
+ )
|
| |
+ self.sync_object.pagure.remove_member_from_group.assert_not_called()
|
| |
+
|
| |
+ @patch("toddlers.plugins.pagure_fas_groups_sync.fedora_account")
|
| |
+ def test_sync_group_add_member_user_not_in_pagure(self, mock_fas):
|
| |
+ """
|
| |
+ Assert that adding member to group works as intended
|
| |
+ when the user doesn't exist in pagure.
|
| |
+ """
|
| |
+ # Preparation
|
| |
+ fas_group = "fas_group"
|
| |
+ pagure_group = "pagure_group"
|
| |
+ self.sync_object.pagure.get_group_members.return_value = ["user1"]
|
| |
+ self.sync_object.pagure.user_exists.return_value = False
|
| |
+ mock_fas.get_group_member.return_value = ["user1", "user2"]
|
| |
+
|
| |
+ # Test
|
| |
+ self.sync_object.sync_group(fas_group, pagure_group)
|
| |
+
|
| |
+ # Assertions
|
| |
+ mock_fas.get_group_member.assert_called_with(fas_group)
|
| |
+ self.sync_object.pagure.get_group_members.assert_called_with(pagure_group)
|
| |
+
|
| |
+ self.sync_object.pagure.add_member_to_group.assert_not_called()
|
| |
+ self.sync_object.pagure.remove_member_from_group.assert_not_called()
|
| |
+
|
| |
+ @patch("toddlers.plugins.pagure_fas_groups_sync.fedora_account")
|
| |
+ def test_sync_group_add_member_failure(self, mock_fas, caplog):
|
| |
+ """Assert that failing to adding member to group works as intended."""
|
| |
+ # Preparation
|
| |
+ caplog.set_level(logging.ERROR)
|
| |
+ fas_group = "fas_group"
|
| |
+ pagure_group = "pagure_group"
|
| |
+ self.sync_object.pagure.get_group_members.return_value = ["user1"]
|
| |
+ mock_fas.get_group_member.return_value = ["user1", "user2"]
|
| |
+ self.sync_object.pagure.add_member_to_group.side_effect = PagureError(
|
| |
+ "PagureError"
|
| |
+ )
|
| |
+
|
| |
+ # Test
|
| |
+ self.sync_object.sync_group(fas_group, pagure_group)
|
| |
+
|
| |
+ # Assertions
|
| |
+ mock_fas.get_group_member.assert_called_with(fas_group)
|
| |
+ self.sync_object.pagure.get_group_members.assert_called_with(pagure_group)
|
| |
+
|
| |
+ self.sync_object.pagure.add_member_to_group.assert_called_with(
|
| |
+ "user2", pagure_group
|
| |
+ )
|
| |
+ self.sync_object.pagure.remove_member_from_group.assert_not_called()
|
| |
+ assert caplog.records[-1].message == "PagureError"
|
| |
+
|
| |
+ @patch("toddlers.plugins.pagure_fas_groups_sync.fedora_account")
|
| |
+ def test_sync_group_remove_member(self, mock_fas):
|
| |
+ """Assert that removing member from group works as intended."""
|
| |
+ # Preparation
|
| |
+ fas_group = "fas_group"
|
| |
+ pagure_group = "pagure_group"
|
| |
+ self.sync_object.pagure.get_group_members.return_value = ["user1", "user2"]
|
| |
+ mock_fas.get_group_member.return_value = ["user1"]
|
| |
+
|
| |
+ # Test
|
| |
+ self.sync_object.sync_group(fas_group, pagure_group)
|
| |
+
|
| |
+ # Assertions
|
| |
+ mock_fas.get_group_member.assert_called_with(fas_group)
|
| |
+ self.sync_object.pagure.get_group_members.assert_called_with(pagure_group)
|
| |
+
|
| |
+ self.sync_object.pagure.remove_member_from_group.assert_called_with(
|
| |
+ "user2", pagure_group
|
| |
+ )
|
| |
+ self.sync_object.pagure.add_member_to_group.assert_not_called()
|
| |
+
|
| |
+ @patch("toddlers.plugins.pagure_fas_groups_sync.fedora_account")
|
| |
+ def test_sync_group_remove_member_user_not_in_pagure(self, mock_fas):
|
| |
+ """
|
| |
+ Assert that removing member from group works
|
| |
+ as intended when user doesn't exist in pagure.
|
| |
+ """
|
| |
+ # Preparation
|
| |
+ fas_group = "fas_group"
|
| |
+ pagure_group = "pagure_group"
|
| |
+ self.sync_object.pagure.get_group_members.return_value = ["user1", "user2"]
|
| |
+ self.sync_object.pagure.user_exists.return_value = False
|
| |
+ mock_fas.get_group_member.return_value = ["user1"]
|
| |
+
|
| |
+ # Test
|
| |
+ self.sync_object.sync_group(fas_group, pagure_group)
|
| |
+
|
| |
+ # Assertions
|
| |
+ mock_fas.get_group_member.assert_called_with(fas_group)
|
| |
+ self.sync_object.pagure.get_group_members.assert_called_with(pagure_group)
|
| |
+
|
| |
+ self.sync_object.pagure.remove_member_from_group.assert_not_called()
|
| |
+ self.sync_object.pagure.add_member_to_group.assert_not_called()
|
| |
+
|
| |
+ @patch("toddlers.plugins.pagure_fas_groups_sync.fedora_account")
|
| |
+ def test_sync_group_remove_member_failure(self, mock_fas, caplog):
|
| |
+ """Assert that removing member from group works as intended."""
|
| |
+ # Preparation
|
| |
+ caplog.set_level(logging.ERROR)
|
| |
+ fas_group = "fas_group"
|
| |
+ pagure_group = "pagure_group"
|
| |
+ self.sync_object.pagure.get_group_members.return_value = ["user1", "user2"]
|
| |
+ mock_fas.get_group_member.return_value = ["user1"]
|
| |
+ self.sync_object.pagure.remove_member_from_group.side_effect = PagureError(
|
| |
+ "PagureError"
|
| |
+ )
|
| |
+
|
| |
+ # Test
|
| |
+ self.sync_object.sync_group(fas_group, pagure_group)
|
| |
+
|
| |
+ # Assertions
|
| |
+ mock_fas.get_group_member.assert_called_with(fas_group)
|
| |
+ self.sync_object.pagure.get_group_members.assert_called_with(pagure_group)
|
| |
+
|
| |
+ self.sync_object.pagure.remove_member_from_group.assert_called_with(
|
| |
+ "user2", pagure_group
|
| |
+ )
|
| |
+ self.sync_object.pagure.add_member_to_group.assert_not_called()
|
| |
+ assert caplog.records[-1].message == "PagureError"
|
| |
+
|
| |
+ @patch("toddlers.plugins.pagure_fas_groups_sync.fedora_account")
|
| |
+ @pytest.mark.parametrize(
|
| |
+ "mock_fas_group, mock_pagure_group", [([], ["user"]), (["user"], [])]
|
| |
+ )
|
| |
+ def test_sync_group_group_empty(self, mock_fas, mock_fas_group, mock_pagure_group):
|
| |
+ """
|
| |
+ Assert that nothing is done if any of the retrieved groups is empty.
|
| |
+ That should not happen, there always needs to be at least one member
|
| |
+ in the group.
|
| |
+ """
|
| |
+ # Preparation
|
| |
+ fas_group = "fas_group"
|
| |
+ pagure_group = "pagure_group"
|
| |
+ self.sync_object.pagure.get_group_members.return_value = mock_pagure_group
|
| |
+ mock_fas.get_group_member.return_value = mock_fas_group
|
| |
+
|
| |
+ # Test
|
| |
+ self.sync_object.sync_group(fas_group, pagure_group)
|
| |
+
|
| |
+ # Assertions
|
| |
+ mock_fas.get_group_member.assert_called_with(fas_group)
|
| |
+ self.sync_object.pagure.get_group_members.assert_called_with(pagure_group)
|
| |
+
|
| |
+ self.sync_object.pagure.add_member_to_group.assert_not_called()
|
| |
+ self.sync_object.pagure.remove_member_from_group.assert_not_called()
|
| |
+
|
| |
+
|
| |
+ class TestMain:
|
| |
+ """
|
| |
+ Test class for `toddlers.plugins.pagure_fas_groups_sync.main` function.
|
| |
+ """
|
| |
+
|
| |
+ def test_main_no_args(self, capsys):
|
| |
+ """Assert that help is printed if no arg is provided."""
|
| |
+ # Test
|
| |
+ with pytest.raises(SystemExit):
|
| |
+ pagure_fas_groups_sync.main([])
|
| |
+
|
| |
+ # Assertions
|
| |
+ out, err = capsys.readouterr()
|
| |
+ assert out == ""
|
| |
+ # Expecting something along these lines, but don't make the test too tight:
|
| |
+ #
|
| |
+ # usage: pytest [-h] [--dry-run] [-q | --debug] conf [username]
|
| |
+ # pytest: error: the following arguments are required: conf
|
| |
+ assert err.startswith("usage:")
|
| |
+ assert "error: the following arguments are required:" in err
|
| |
+
|
| |
+ @patch("toddlers.utils.pagure.set_pagure")
|
| |
+ @patch("toddlers.utils.fedora_account.set_fasjson")
|
| |
+ @patch("toddlers.plugins.pagure_fas_groups_sync.PagureFASGroupsSync.sync_group")
|
| |
+ def test_main(self, mock_sync_group, mock_fas, mock_pagure):
|
| |
+ """Assert that main is processed correctly."""
|
| |
+ # Test
|
| |
+ pagure_fas_groups_sync.main(
|
| |
+ [
|
| |
+ "--api-key",
|
| |
+ "api_key",
|
| |
+ "--group",
|
| |
+ "fas_group",
|
| |
+ "--target-group",
|
| |
+ "pagure_group",
|
| |
+ "--debug",
|
| |
+ ]
|
| |
+ )
|
| |
+
|
| |
+ # Assertions
|
| |
+ mock_pagure.assert_called_with(
|
| |
+ {"pagure_url": "https://pagure.io", "pagure_api_key": "api_key"}
|
| |
+ )
|
| |
+ mock_fas.assert_called_with({"fas_url": "https://fasjson.fedoraproject.org"})
|
| |
+ mock_sync_group.assert_called_with("fas_group", "pagure_group")
|
| |
https://pagure.io/fedora-infra/toddlers/issue/177
This PR will stay open till the work on https://pagure.io/pagure/issue/5333 is finished.