From 74694c58d1e705f6647bab2cf8e6efd105fbf60c Mon Sep 17 00:00:00 2001 From: Tomas Kopecek Date: Dec 21 2017 11:06:53 +0000 Subject: [PATCH 1/2] CLI commands for notifications Fixes: https://pagure.io/koji/issue/680 --- diff --git a/cli/koji_cli/commands.py b/cli/koji_cli/commands.py index c3a66d7..cc55ae0 100644 --- a/cli/koji_cli/commands.py +++ b/cli/koji_cli/commands.py @@ -7050,3 +7050,170 @@ def handle_moshimoshi(options, session, args): print("Authenticated via Kerberos principal %s" % u["krb_principal"]) elif authtype == koji.AUTHTYPE_SSL: print("Authenticated via client certificate %s" % options.cert) + + +def anon_handle_list_notifications(goptions, session, args): + "[monitor] List user's notifications" + usage = _("usage: %prog list-notifications [options]") + usage += _("\n(Specify the --help global option for a list of other help options)") + parser = OptionParser(usage=usage) + parser.add_option("--mine", action="store_true", help=_("Just print your notifications")) + parser.add_option("--user", help=_("Only notifications for this user")) + (options, args) = parser.parse_args(args) + + if len(args) != 0: + parser.error(_("This command takes no arguments")) + if not options.mine and not options.user: + parser.error(_("Use --user or --mine.")) + + activate_session(session, goptions) + + if options.user: + user = session.getUser(options.user) + if not user: + print("User %s does not exist" % options.user) + return 1 + user_id = user['id'] + else: + user_id = None + + mask = "%(id)6s %(tag)-25s %(package)-25s %(email)-20s %(success)s" + head = mask % {'id': 'ID', 'tag': 'Tag', 'package': 'Package', 'email': 'E-mail', 'success': 'Success-only'} + print(head) + print('-' * len(head)) + for notification in session.getBuildNotifications(user_id): + if notification['tag_id']: + notification['tag'] = session.getTag(notification['tag_id'])['name'] + else: + notification['tag'] = '*' + if notification['package_id']: + notification['package'] = session.getPackage(notification['package_id'])['name'] + else: + notification['package'] = '*' + notification['success'] = ['no', 'yes'][notification['success_only']] + print(mask % notification) + + +def handle_add_notification(goptions, session, args): + "[monitor] Add user's notification" + usage = _("usage: %prog add-notification [options]") + usage += _("\n(Specify the --help global option for a list of other help options)") + parser = OptionParser(usage=usage) + parser.add_option("--user", help=_("Add notifications for this user (admin-only)")) + parser.add_option("--package", help=_("Add notifications for this package")) + parser.add_option("--tag", help=_("Add notifications for this tag")) + parser.add_option("--success-only", action="store_true", default=False, help=_("")) + (options, args) = parser.parse_args(args) + + if len(args) != 0: + parser.error(_("This command takes no arguments")) + + if not options.package and not options.tag: + parser.error(_("Command need at least one from --tag or --package options.")) + + activate_session(session, goptions) + + if options.user and not session.hasPerm('admin'): + parser.error("--user requires admin permission") + + if options.user: + user_id = session.getUser(options.user)['id'] + else: + user_id = session.getLoggedInUser()['id'] + + if options.package: + package_id = session.getPackageID(options.package) + if package_id is None: + parser.error("Unknown package: %s" % options.package) + else: + package_id = None + + if options.tag: + try: + tag_id = session.getTagID(options.tag, strict=True) + except koji.GenericError: + parser.error("Uknown tag: %s" % options.tag) + else: + tag_id = None + + session.createNotification(user_id, package_id, tag_id, options.success_only) + + +def handle_remove_notification(goptions, session, args): + "[monitor] Remove user's notifications" + usage = _("usage: %prog remove-notification [options] ID [ID2, ...]") + usage += _("\n(Specify the --help global option for a list of other help options)") + parser = OptionParser(usage=usage) + (options, args) = parser.parse_args(args) + + activate_session(session, goptions) + + if len(args) < 1: + parser.error(_("At least one notification id has to be specified")) + + try: + n_ids = [int(x) for x in args] + except ValueError as e: + parser.error(_("All notification ids has to be integers")) + + for n_id in n_ids: + session.deleteNotification(n_id) + if not goptions.quiet: + print(_("Notification %s successfully removed.") % n_id) + + +def handle_edit_notification(goptions, session, args): + "[monitor] Edit user's notification" + usage = _("usage: %prog edit-notification [options] ID") + usage += _("\n(Specify the --help global option for a list of other help options)") + parser = OptionParser(usage=usage) + parser.add_option("--package", + help=_("Notifications for this package, '*' for all")) + parser.add_option("--tag", + help=_("Notifications for this tag, '*' for all")) + parser.add_option("--success-only", action="store_true", default=None, + dest='success_only', help=_("Notify only on successful events")) + parser.add_option("--no-success-only", action="store_false", + default=None, dest='success_only', help=_("Notify on all events")) + (options, args) = parser.parse_args(args) + + if len(args) != 1: + parser.error(_("Only argument is notification ID")) + + try: + n_id = int(args[0]) + except ValueError as e: + parser.error(_("Notification ID has to be numeric")) + + if not options.package and not options.tag and options.success_only is None: + parser.error(_("Command need at least one option")) + + activate_session(session, goptions) + + old = session.getBuildNotification(n_id) + + if options.package == '*': + package_id = None + elif options.package: + package_id = session.getPackageID(options.package) + if package_id is None: + parser.error("Unknown package: %s" % options.package) + else: + package_id = old['package_id'] + + if options.tag == '*': + tag_id = None + elif options.tag: + try: + tag_id = session.getTagID(options.tag, strict=True) + except koji.GenericError: + parser.error("Uknown tag: %s" % options.tag) + else: + tag_id = old['tag_id'] + + if options.success_only is not None: + success_only = options.success_only + else: + success_only = old['success_only'] + + session.updateNotification(n_id, package_id, tag_id, success_only) diff --git a/tests/test_cli/data/list-commands.txt b/tests/test_cli/data/list-commands.txt index 76e04d1..98a901b 100644 --- a/tests/test_cli/data/list-commands.txt +++ b/tests/test_cli/data/list-commands.txt @@ -123,6 +123,10 @@ miscellaneous commands: moshimoshi Introduce yourself monitor commands: + add-notification Add user's notification + edit-notification Edit user's notification + list-notifications List user's notifications + remove-notification Remove user's notifications wait-repo Wait for a repo to be regenerated watch-logs Watch logs in realtime watch-task Track progress of particular tasks diff --git a/tests/test_cli/test_add_notification.py b/tests/test_cli/test_add_notification.py new file mode 100644 index 0000000..51da4a2 --- /dev/null +++ b/tests/test_cli/test_add_notification.py @@ -0,0 +1,116 @@ +from __future__ import absolute_import +import koji +import mock +import unittest +from six.moves import StringIO + +from koji_cli.commands import handle_add_notification + +class TestAddNotification(unittest.TestCase): + def setUp(self): + self.options = mock.MagicMock() + self.options.quiet = True + self.options.debug = False + self.session = mock.MagicMock() + self.session.getAPIVersion.return_value = koji.API_VERSION + + + @mock.patch('koji_cli.commands.activate_session') + def test_handle_add_notification(self, activate_session_mock): + self.session.getPackageID.return_value = 1234 + self.session.getTagID.return_value = 4321 + self.session.getLoggedInUser.return_value = {'id': 678} + + handle_add_notification(self.options, self.session, ['--package', 'pkg_a', '--tag', 'tag_a', '--success-only']) + + self.session.getPackageID.assert_called_once_with('pkg_a') + self.session.getTagID.assert_called_once_with('tag_a', strict=True) + self.session.getLoggedInUser.assert_called_once_with() + self.session.getUser.assert_not_called() + self.session.createNotification.assert_called_once_with(678, 1234, 4321, True) + + + @mock.patch('koji_cli.commands.activate_session') + def test_handle_add_notification_no_pkg(self, activate_session_mock): + self.session.getTagID.return_value = 4321 + self.session.getLoggedInUser.return_value = {'id': 678} + + handle_add_notification(self.options, self.session, ['--tag', 'tag_a', '--success-only']) + + self.session.getPackageID.assert_not_called() + self.session.getTagID.assert_called_once_with('tag_a', strict=True) + self.session.getLoggedInUser.assert_called_once_with() + self.session.getUser.assert_not_called() + self.session.createNotification.assert_called_once_with(678, None, 4321, True) + + + @mock.patch('koji_cli.commands.activate_session') + def test_handle_add_notification_no_tag(self, activate_session_mock): + self.session.getPackageID.return_value = 1234 + self.session.getLoggedInUser.return_value = {'id': 678} + + handle_add_notification(self.options, self.session, ['--package', 'pkg_a']) + + self.session.getPackageID.assert_called_once_with('pkg_a') + self.session.getTagID.assert_not_called() + self.session.getLoggedInUser.assert_called_once_with() + self.session.getUser.assert_not_called() + self.session.createNotification.assert_called_once_with(678, 1234, None, False) + + + @mock.patch('sys.exit') + @mock.patch('sys.stderr', new_callable=StringIO) + def test_handle_add_notification_no_pkg_no_tag(self, sys_stderr, sys_exit): + sys_exit.side_effect = SystemExit() + + with self.assertRaises(SystemExit): + handle_add_notification(self.options, self.session, ['--success-only']) + + self.session.getPackageID.assert_not_called() + self.session.getTagID.assert_not_called() + self.session.getLoggedInUser.assert_not_called() + self.session.getUser.assert_not_called() + self.session.createNotification.assert_not_called() + + + @mock.patch('sys.exit') + @mock.patch('sys.stderr', new_callable=StringIO) + def test_handle_add_notification_user_no_admin(self, sys_stderr, sys_exit): + sys_exit.side_effect = SystemExit() + self.session.hasPerm.return_value = False + + with self.assertRaises(SystemExit): + handle_add_notification(self.options, self.session, ['--user', 'username', '--tag', 'tag_a']) + + self.session.getPackageID.assert_not_called() + self.session.getTagID.assert_not_called() + self.session.getLoggedInUser.assert_not_called() + self.session.getUser.assert_not_called() + self.session.createNotification.assert_not_called() + + + @mock.patch('sys.exit') + @mock.patch('sys.stderr', new_callable=StringIO) + def test_handle_add_notification_user_admin(self, sys_stderr, sys_exit): + self.session.hasPerm.return_value = True + self.session.getPackageID.return_value = 1234 + self.session.getUser.return_value = {'id': 789} + + handle_add_notification(self.options, self.session, ['--package', 'pkg_a', '--user', 'username']) + + self.session.getPackageID.assert_called_once_with('pkg_a') + self.session.getTagID.assert_not_called() + self.session.getLoggedInUser.assert_not_called() + self.session.getUser.assert_called_once_with('username') + self.session.createNotification.assert_called_once_with(789, 1234, None, False) + + + @mock.patch('sys.exit') + @mock.patch('sys.stderr', new_callable=StringIO) + def test_handle_add_notification_args(self, sys_stderr, sys_exit): + sys_exit.side_effect = SystemExit() + + with self.assertRaises(SystemExit): + handle_add_notification(self.options, self.session, ['bogus']) + + self.session.createNotification.assert_not_called() diff --git a/tests/test_cli/test_list_notifications.py b/tests/test_cli/test_list_notifications.py new file mode 100644 index 0000000..89e1960 --- /dev/null +++ b/tests/test_cli/test_list_notifications.py @@ -0,0 +1,97 @@ +import mock +import unittest +from six.moves import StringIO + +import koji + +from koji_cli.commands import anon_handle_list_notifications + +class TestListNotifications(unittest.TestCase): + def setUp(self): + self.options = mock.MagicMock() + self.options.debug = False + self.session = mock.MagicMock() + self.session.getAPIVersion.return_value = koji.API_VERSION + + @mock.patch('sys.stdout', new_callable=StringIO) + @mock.patch('koji_cli.commands.activate_session') + def test_list_notifications(self, activate_session_mock, stdout): + self.session.getBuildNotifications.return_value = [ + {'id': 1, 'tag_id': 1, 'package_id': 11, 'email': 'email@test.com', 'success_only': True}, + {'id': 2, 'tag_id': None, 'package_id': 11, 'email': 'email@test.com', 'success_only': False}, + {'id': 3, 'tag_id': 1, 'package_id': None, 'email': 'email@test.com', 'success_only': True}, + ] + self.session.getTag.return_value = {'id': 1, 'name': 'tag'} + self.session.getPackage.return_value = {'id': 11, 'name': 'package'} + + anon_handle_list_notifications(self.options, self.session, ['--mine']) + + actual = stdout.getvalue() + expected = ''' ID Tag Package E-mail Success-only +-------------------------------------------------------------------------------------------- + 1 tag package email@test.com yes + 2 * package email@test.com no + 3 tag * email@test.com yes +''' + + self.maxDiff=None + self.assertMultiLineEqual(actual, expected) + activate_session_mock.assert_called_once_with(self.session, self.options) + self.session.getTag.assert_has_calls((mock.call(1), mock.call(1))) + self.session.getPackage.assert_has_calls((mock.call(11), mock.call(11))) + self.session.getUser.assert_not_called() + self.session.getBuildNotifications.assert_called_once_with(None) + + @mock.patch('sys.stdout', new_callable=StringIO) + @mock.patch('koji_cli.commands.activate_session') + def test_list_notifications_user(self, activate_session_mock, stdout): + self.session.getBuildNotifications.return_value = [ + {'id': 1, 'tag_id': 1, 'package_id': 11, 'email': 'email@test.com', 'success_only': True}, + {'id': 2, 'tag_id': None, 'package_id': 11, 'email': 'email@test.com', 'success_only': False}, + {'id': 3, 'tag_id': 1, 'package_id': None, 'email': 'email@test.com', 'success_only': True}, + ] + self.session.getTag.return_value = {'id': 1, 'name': 'tag'} + self.session.getPackage.return_value = {'id': 11, 'name': 'package'} + self.session.getUser.return_value = {'id': 321} + + anon_handle_list_notifications(self.options, self.session, ['--user', 'random_name']) + + actual = stdout.getvalue() + expected = ''' ID Tag Package E-mail Success-only +-------------------------------------------------------------------------------------------- + 1 tag package email@test.com yes + 2 * package email@test.com no + 3 tag * email@test.com yes +''' + + self.maxDiff=None + self.assertMultiLineEqual(actual, expected) + activate_session_mock.assert_called_once_with(self.session, self.options) + self.session.getTag.assert_has_calls((mock.call(1), mock.call(1))) + self.session.getPackage.assert_has_calls((mock.call(11), mock.call(11))) + self.session.getUser.assert_called_once_with('random_name') + self.session.getBuildNotifications.assert_called_once_with(321) + + @mock.patch('sys.exit') + @mock.patch('sys.stderr', new_callable=StringIO) + def test_list_notifications_missing_params(self, sys_stderr, sys_exit): + sys_exit.side_effect = SystemExit() + + with self.assertRaises(SystemExit): + anon_handle_list_notifications(self.options, self.session, []) + + self.session.getUser.assert_not_called() + self.session.getBuildNotifications.assert_not_called() + self.session.getTag.assert_not_called() + self.session.getPackage.assert_not_called() + + + @mock.patch('sys.exit') + @mock.patch('sys.stderr', new_callable=StringIO) + def test_handle_list_notifications_no_args(self, sys_stderr, sys_exit): + sys_exit.side_effect = SystemExit() + + with self.assertRaises(SystemExit): + anon_handle_list_notifications(self.options, self.session, []) + + self.session.getBuildNotifications.assert_not_called() diff --git a/tests/test_cli/test_remove_notification.py b/tests/test_cli/test_remove_notification.py new file mode 100644 index 0000000..2d08730 --- /dev/null +++ b/tests/test_cli/test_remove_notification.py @@ -0,0 +1,43 @@ +from __future__ import absolute_import +import koji +import mock +import unittest +from six.moves import StringIO + +from koji_cli.commands import handle_remove_notification + +class TestAddHost(unittest.TestCase): + def setUp(self): + self.options = mock.MagicMock() + self.options.debug = False + self.session = mock.MagicMock() + self.session.getAPIVersion.return_value = koji.API_VERSION + + + @mock.patch('koji_cli.commands.activate_session') + def test_handle_remove_notification(self, activate_session_mock): + handle_remove_notification(self.options, self.session, ['1', '3', '5']) + + self.session.deleteNotification.assert_has_calls([mock.call(1), mock.call(3), mock.call(5)]) + + + @mock.patch('sys.exit') + @mock.patch('sys.stderr', new_callable=StringIO) + def test_handle_remove_notification_bogus(self, sys_stderr, sys_exit): + sys_exit.side_effect = SystemExit() + + with self.assertRaises(SystemExit): + handle_remove_notification(self.options, self.session, ['bogus']) + + self.session.deleteNotification.assert_not_called() + + + @mock.patch('sys.exit') + @mock.patch('sys.stderr', new_callable=StringIO) + def test_handle_remove_notifications_no_args(self, sys_stderr, sys_exit): + sys_exit.side_effect = SystemExit() + + with self.assertRaises(SystemExit): + handle_remove_notification(self.options, self.session, []) + + self.session.deleteNotification.assert_not_called() From 81f0f7863b032b7f0cc993cea243a94d169aaab4 Mon Sep 17 00:00:00 2001 From: Tomas Kopecek Date: Dec 21 2017 11:06:53 +0000 Subject: [PATCH 2/2] added test --- diff --git a/tests/test_cli/test_edit_notification.py b/tests/test_cli/test_edit_notification.py new file mode 100644 index 0000000..8204709 --- /dev/null +++ b/tests/test_cli/test_edit_notification.py @@ -0,0 +1,89 @@ +from __future__ import absolute_import +import koji +import mock +import unittest +from six.moves import StringIO + +from koji_cli.commands import handle_edit_notification + +class TestEditNotification(unittest.TestCase): + def setUp(self): + self.options = mock.MagicMock() + self.options.debug = False + self.session = mock.MagicMock() + self.session.getAPIVersion.return_value = koji.API_VERSION + + + @mock.patch('koji_cli.commands.activate_session') + def test_handle_edit_notification(self, activate_session_mock): + self.session.getPackageID.return_value = 1234 + self.session.getTagID.return_value = 4321 + self.session.getBuildNotification.return_value = {'id': 2345} + + handle_edit_notification(self.options, self.session, + ['--package', 'pkg_a', '--tag', 'tag_a', '--success-only', '2345']) + + self.session.getPackageID.assert_called_once_with('pkg_a') + self.session.getTagID.assert_called_once_with('tag_a', strict=True) + self.session.updateNotification.assert_called_once_with(2345, 1234, 4321, True) + + + @mock.patch('koji_cli.commands.activate_session') + def test_handle_edit_notification_no_pkg(self, activate_session_mock): + self.session.getBuildNotification.return_value = \ + {'id': 2345, 'package_id': 135, 'success_only': False} + + handle_edit_notification(self.options, self.session, + ['--tag', '*', '2345']) + + self.session.getPackageID.assert_not_called() + self.session.getTagID.assert_not_called() + self.session.updateNotification.assert_called_once_with(2345, 135, None, False) + self.session.getBuildNotification.assert_called_once_with(2345) + + + @mock.patch('koji_cli.commands.activate_session') + def test_handle_edit_notification_no_tag(self, activate_session_mock): + self.session.getBuildNotification.return_value = \ + {'id': 2345, 'tag_id': 135, 'success_only': True} + + handle_edit_notification(self.options, self.session, + ['--package', '*', '--no-success-only', '2345']) + + self.session.getPackageID.assert_not_called() + self.session.getTagID.assert_not_called() + self.session.updateNotification.assert_called_once_with(2345, None, 135, False) + self.session.getBuildNotification.assert_called_once_with(2345) + + + @mock.patch('sys.exit') + @mock.patch('sys.stderr', new_callable=StringIO) + def test_handle_edit_notification_bogus(self, sys_stderr, sys_exit): + sys_exit.side_effect = SystemExit() + + with self.assertRaises(SystemExit): + handle_edit_notification(self.options, self.session, ['bogus']) + + self.session.updateNotification.assert_not_called() + + + @mock.patch('sys.exit') + @mock.patch('sys.stderr', new_callable=StringIO) + def test_handle_edit_notification_no_id(self, sys_stderr, sys_exit): + sys_exit.side_effect = SystemExit() + + with self.assertRaises(SystemExit): + handle_edit_notification(self.options, self.session, []) + + self.session.updateNotification.assert_not_called() + + + @mock.patch('sys.exit') + @mock.patch('sys.stderr', new_callable=StringIO) + def test_handle_edit_notification_no_opts(self, sys_stderr, sys_exit): + sys_exit.side_effect = SystemExit() + + with self.assertRaises(SystemExit): + handle_edit_notification(self.options, self.session, ['123']) + + self.session.updateNotification.assert_not_called()