particle-os/debian-koji
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

CLI commands for notifications

Browse source 
Fixes: https://pagure.io/koji/issue/680
This commit is contained in:
Tomas Kopecek 2017-11-08 16:53:36 +01:00 • committed by Mike McLean
parent c228806e9e
commit 252c9e1c20
5 changed files with 427 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

167
cli/koji_cli/commands.py
View file

@ -7061,3 +7061,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)

4
tests/test_cli/data/list-commands.txt
View file

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

116
tests/test_cli/test_add_notification.py Normal file
View file

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

97
tests/test_cli/test_list_notifications.py Normal file
View file

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

43
tests/test_cli/test_remove_notification.py Normal file
View file

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