#688 CLI commands for notifications
Merged 4 years ago by mikem. Opened 4 years ago by tkopecek.
tkopecek/koji issue680  into  master

file modified
+167
@@ -7050,3 +7050,170 @@ 

          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)

@@ -123,6 +123,10 @@ 

          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

@@ -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()

@@ -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()

@@ -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()

@@ -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()