API/CLI/tests for notification blocks
This commit is contained in:
Tomas Kopecek
Mike McLean
parent
e4c37dfffb
commit
b7d273b2fa
5 changed files with 461 additions and 60 deletions
|
|
@ -7306,7 +7306,7 @@ def handle_moshimoshi(options, session, args):
|
|||
|
||||
|
||||
def anon_handle_list_notifications(goptions, session, args):
|
||||
"[monitor] List user's notifications"
|
||||
"[monitor] List user's notifications and blocks"
|
||||
usage = _("usage: %prog list-notifications [options]")
|
||||
usage += _("\n(Specify the --help global option for a list of other help options)")
|
||||
parser = OptionParser(usage=usage)
|
||||
|
|
@ -7330,21 +7330,49 @@ def anon_handle_list_notifications(goptions, session, args):
|
|||
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)
|
||||
mask = "%(id)6s %(tag)-25s %(package)-25s %(email)-20s %(success)-12s"
|
||||
headers = {'id': 'ID', 'tag': 'Tag', 'package': 'Package', 'email': 'E-mail', 'success': 'Success-only'}
|
||||
head = mask % headers
|
||||
notifications = session.getBuildNotifications(user_id)
|
||||
if notifications:
|
||||
print('Notifications')
|
||||
print(head)
|
||||
print('-' * len(head))
|
||||
for notification in notifications:
|
||||
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)
|
||||
else:
|
||||
print('No notifications')
|
||||
|
||||
print('')
|
||||
|
||||
mask = "%(id)6s %(tag)-25s %(package)-25s"
|
||||
head = mask % headers
|
||||
blocks = session.getBuildNotificationBlocks(user_id)
|
||||
if blocks:
|
||||
print('Notification blocks')
|
||||
print(head)
|
||||
print('-' * len(head))
|
||||
for notification in blocks:
|
||||
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'] = '*'
|
||||
print(mask % notification)
|
||||
else:
|
||||
print('No notification blocks')
|
||||
|
||||
|
||||
def handle_add_notification(goptions, session, args):
|
||||
|
|
@ -7412,7 +7440,7 @@ def handle_remove_notification(goptions, session, args):
|
|||
for n_id in n_ids:
|
||||
session.deleteNotification(n_id)
|
||||
if not goptions.quiet:
|
||||
print(_("Notification %s successfully removed.") % n_id)
|
||||
print(_("Notification %d successfully removed.") % n_id)
|
||||
|
||||
|
||||
def handle_edit_notification(goptions, session, args):
|
||||
|
|
@ -7470,3 +7498,75 @@ def handle_edit_notification(goptions, session, args):
|
|||
success_only = old['success_only']
|
||||
|
||||
session.updateNotification(n_id, package_id, tag_id, success_only)
|
||||
|
||||
|
||||
def handle_block_notification(goptions, session, args):
|
||||
"[monitor] Block user's notifications"
|
||||
usage = _("usage: %prog block-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=_("Block notifications for this user (admin-only)"))
|
||||
parser.add_option("--package", help=_("Block notifications for this package"))
|
||||
parser.add_option("--tag", help=_("Block notifications for this tag"))
|
||||
parser.add_option("--all", action="store_true", help=_("Block all notification for this user"))
|
||||
(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 and not options.all:
|
||||
parser.error(_("One of --tag, --package or --all must be specified."))
|
||||
|
||||
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("Unknown tag: %s" % options.tag)
|
||||
else:
|
||||
tag_id = None
|
||||
|
||||
for block in session.getBuildNotificationBlocks(user_id):
|
||||
if (block['package_id'] == package_id and block['tag_id'] == tag_id):
|
||||
parser.error('Notification already exists.')
|
||||
|
||||
session.createNotificationBlock(user_id, package_id, tag_id)
|
||||
|
||||
|
||||
def handle_unblock_notification(goptions, session, args):
|
||||
"[monitor] Unblock user's notification"
|
||||
usage = _("usage: %prog unblock-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 block id has to be specified"))
|
||||
|
||||
try:
|
||||
n_ids = [int(x) for x in args]
|
||||
except ValueError:
|
||||
parser.error(_("All notification block ids has to be integers"))
|
||||
|
||||
for n_id in n_ids:
|
||||
session.deleteNotificationBlock(n_id)
|
||||
if not goptions.quiet:
|
||||
print(_("Notification block %d successfully removed.") % n_id)
|
||||
|
|
|
|||
115
hub/kojihub.py
115
hub/kojihub.py
|
|
@ -7666,12 +7666,20 @@ def build_notification(task_id, build_id):
|
|||
make_task('buildNotification', [recipients, build, target, web_url])
|
||||
|
||||
def get_build_notifications(user_id):
|
||||
fields = ('id', 'user_id', 'package_id', 'tag_id', 'success_only', 'email')
|
||||
query = """SELECT %s
|
||||
FROM build_notifications
|
||||
WHERE user_id = %%(user_id)i
|
||||
""" % ', '.join(fields)
|
||||
return _multiRow(query, locals(), fields)
|
||||
query = QueryProcessor(tables=['build_notifications'],
|
||||
columns=('id', 'user_id', 'package_id', 'tag_id',
|
||||
'success_only', 'email'),
|
||||
clauses=['user_id = %(user_id)i'],
|
||||
values=locals())
|
||||
return query.execute()
|
||||
|
||||
def get_build_notification_blocks(user_id):
|
||||
query = QueryProcessor(tables=['build_notifications_block'],
|
||||
columns=['id', 'user_id', 'package_id', 'tag_id'],
|
||||
clauses=['user_id = %(user_id)i'],
|
||||
values=locals())
|
||||
return query.execute()
|
||||
|
||||
|
||||
def new_group(name):
|
||||
"""Add a user group to the database"""
|
||||
|
|
@ -11409,12 +11417,37 @@ class RootExports(object):
|
|||
If there is no notification with the given ID, when strict is True,
|
||||
raise GenericError, else return None.
|
||||
"""
|
||||
fields = ('id', 'user_id', 'package_id', 'tag_id', 'success_only', 'email')
|
||||
query = """SELECT %s
|
||||
FROM build_notifications
|
||||
WHERE id = %%(id)i
|
||||
""" % ', '.join(fields)
|
||||
return _singleRow(query, locals(), fields, strict=strict)
|
||||
query = QueryProcessor(tables=['build_notifications'],
|
||||
columns = ('id', 'user_id', 'package_id', 'tag_id',
|
||||
'success_only', 'email'),
|
||||
clauses = ['id = %(id)i'],
|
||||
values = locals())
|
||||
result = query.executeOne()
|
||||
if strict and not result:
|
||||
raise koji.GenericError("No notification with ID %i found" % id)
|
||||
return result
|
||||
|
||||
def getBuildNotificationBlocks(self, userID=None):
|
||||
"""Get build notifications for the user with the given ID, name or
|
||||
Kerberos principal. If no user is specified, get the notifications for
|
||||
the currently logged-in user. If there is no currently logged-in user,
|
||||
raise a GenericError."""
|
||||
userID = get_user(userID, strict=True)['id']
|
||||
return get_build_notification_blocks(userID)
|
||||
|
||||
def getBuildNotificationBlock(self, id, strict=False):
|
||||
"""Get the build notification with the given ID.
|
||||
If there is no notification with the given ID, when strict is True,
|
||||
raise GenericError, else return None.
|
||||
"""
|
||||
query = QueryProcessor(tables=['build_notifications_block'],
|
||||
columns = ('id', 'user_id', 'package_id', 'tag_id'),
|
||||
clauses = ['id = %(id)i'],
|
||||
values = locals())
|
||||
result = query.executeOne()
|
||||
if strict and not result:
|
||||
raise koji.GenericError("No notification block with ID %i found" % id)
|
||||
return result
|
||||
|
||||
def updateNotification(self, id, package_id, tag_id, success_only):
|
||||
"""Update an existing build notification with new data. If the notification
|
||||
|
|
@ -11424,11 +11457,8 @@ class RootExports(object):
|
|||
if not currentUser:
|
||||
raise koji.GenericError('not logged-in')
|
||||
|
||||
orig_notif = self.getBuildNotification(id)
|
||||
if not orig_notif:
|
||||
raise koji.GenericError('no notification with ID: %i' % id)
|
||||
elif not (orig_notif['user_id'] == currentUser['id'] or
|
||||
self.hasPerm('admin')):
|
||||
orig_notif = self.getBuildNotification(id, strict=True)
|
||||
if not (orig_notif['user_id'] == currentUser['id'] or self.hasPerm('admin')):
|
||||
raise koji.GenericError('user %i cannot update notifications for user %i' % \
|
||||
(currentUser['id'], orig_notif['user_id']))
|
||||
|
||||
|
|
@ -11491,9 +11521,7 @@ class RootExports(object):
|
|||
def deleteNotification(self, id):
|
||||
"""Delete the notification with the given ID. If the currently logged-in
|
||||
user is not the owner of the notification or an admin, raise a GenericError."""
|
||||
notification = self.getBuildNotification(id)
|
||||
if not notification:
|
||||
raise koji.GenericError('no notification with ID: %i' % id)
|
||||
notification = self.getBuildNotification(id, strict=True)
|
||||
currentUser = self.getLoggedInUser()
|
||||
if not currentUser:
|
||||
raise koji.GenericError('not logged-in')
|
||||
|
|
@ -11505,6 +11533,53 @@ class RootExports(object):
|
|||
delete = """DELETE FROM build_notifications WHERE id = %(id)i"""
|
||||
_dml(delete, locals())
|
||||
|
||||
def createNotificationBlock(self, user_id, package_id=None, tag_id=None):
|
||||
"""Create notification block. If the user_id does not match the
|
||||
currently logged-in user and the currently logged-in user is not an
|
||||
admin, raise a GenericError."""
|
||||
currentUser = self.getLoggedInUser()
|
||||
if not currentUser:
|
||||
raise koji.GenericError('not logged in')
|
||||
|
||||
notificationUser = self.getUser(user_id)
|
||||
if not notificationUser:
|
||||
raise koji.GenericError('invalid user ID: %s' % user_id)
|
||||
|
||||
if not (notificationUser['id'] == currentUser['id'] or self.hasPerm('admin')):
|
||||
raise koji.GenericError('user %s cannot create notification blocks for user %s' % \
|
||||
(currentUser['name'], notificationUser['name']))
|
||||
|
||||
# sanitize input
|
||||
user_id = notificationUser['id']
|
||||
if package_id is not None:
|
||||
package_id = get_package_id(package_id, strict=True)
|
||||
if tag_id is not None:
|
||||
tag_id = get_tag_id(tag_id, strict=True)
|
||||
|
||||
# check existing notifications to not have same twice
|
||||
for block in get_build_notification_blocks(user_id):
|
||||
if (block['package_id'] == package_id and block['tag_id'] == tag_id):
|
||||
raise koji.GenericError('notification already exists')
|
||||
|
||||
insert = InsertProcessor('build_notifications_block')
|
||||
insert.set(user_id=user_id, package_id=package_id, tag_id=tag_id)
|
||||
insert.execute()
|
||||
|
||||
def deleteNotificationBlock(self, id):
|
||||
"""Delete the notification block with the given ID. If the currently logged-in
|
||||
user is not the owner of the notification or an admin, raise a GenericError."""
|
||||
block = self.getBuildNotificationBlock(id, strict=True)
|
||||
currentUser = self.getLoggedInUser()
|
||||
if not currentUser:
|
||||
raise koji.GenericError('not logged-in')
|
||||
|
||||
if not (block['user_id'] == currentUser['id'] or
|
||||
self.hasPerm('admin')):
|
||||
raise koji.GenericError('user %i cannot delete notification blocks for user %i' % \
|
||||
(currentUser['id'], block['user_id']))
|
||||
delete = """DELETE FROM build_notifications_block WHERE id = %(id)i"""
|
||||
_dml(delete, locals())
|
||||
|
||||
def _prepareSearchTerms(self, terms, matchType):
|
||||
r"""Process the search terms before passing them to the database.
|
||||
If matchType is "glob", "_" will be replaced with "\_" (to match literal
|
||||
|
|
|
|||
|
|
@ -124,9 +124,11 @@ miscellaneous commands:
|
|||
|
||||
monitor commands:
|
||||
add-notification Add user's notification
|
||||
block-notification Block user's notifications
|
||||
edit-notification Edit user's notification
|
||||
list-notifications List user's notifications
|
||||
list-notifications List user's notifications and blocks
|
||||
remove-notification Remove user's notifications
|
||||
unblock-notification Unblock user's notification
|
||||
wait-repo Wait for a repo to be regenerated
|
||||
watch-logs Watch logs in realtime
|
||||
watch-task Track progress of particular tasks
|
||||
|
|
|
|||
|
|
@ -25,17 +25,22 @@ class TestListNotifications(unittest.TestCase):
|
|||
{'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.getBuildNotificationBlocks.return_value = []
|
||||
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
|
||||
expected = '''\
|
||||
Notifications
|
||||
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
|
||||
1 tag package email@test.com yes
|
||||
2 * package email@test.com no
|
||||
3 tag * email@test.com yes
|
||||
|
||||
No notification blocks
|
||||
'''
|
||||
|
||||
self.maxDiff=None
|
||||
|
|
@ -54,18 +59,37 @@ class TestListNotifications(unittest.TestCase):
|
|||
{'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.getBuildNotificationBlocks.return_value = [
|
||||
{'id': 11, 'tag_id': None, 'package_id': 22},
|
||||
{'id': 12, 'tag_id': None, 'package_id': None},
|
||||
]
|
||||
self.session.getTag.side_effect = [
|
||||
{'id': 1, 'name': 'tag'},
|
||||
{'id': 3, 'name': 'tag3'},
|
||||
]
|
||||
self.session.getPackage.side_effect = [
|
||||
{'id': 11, 'name': 'package'},
|
||||
{'id': 11, 'name': 'package'},
|
||||
{'id': 22, '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
|
||||
expected = '''\
|
||||
Notifications
|
||||
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
|
||||
1 tag package email@test.com yes
|
||||
2 * package email@test.com no
|
||||
3 tag3 * email@test.com yes
|
||||
|
||||
Notification blocks
|
||||
ID Tag Package
|
||||
----------------------------------------------------------
|
||||
11 * package
|
||||
12 * *
|
||||
'''
|
||||
|
||||
self.maxDiff=None
|
||||
|
|
|
|||
|
|
@ -53,6 +53,7 @@ class TestNotifications(unittest.TestCase):
|
|||
self.exports.getUser = mock.MagicMock()
|
||||
self.exports.hasPerm = mock.MagicMock()
|
||||
self.exports.getBuildNotification = mock.MagicMock()
|
||||
self.exports.getBuildNotificationBlock = mock.MagicMock()
|
||||
|
||||
def tearDown(self):
|
||||
mock.patch.stopall()
|
||||
|
|
@ -441,34 +442,36 @@ class TestNotifications(unittest.TestCase):
|
|||
|
||||
self.exports.deleteNotification(n_id)
|
||||
|
||||
self.exports.getBuildNotification.assert_called_once_with(n_id)
|
||||
self.exports.getBuildNotification.assert_called_once_with(n_id, strict=True)
|
||||
self.exports.getLoggedInUser.assert_called_once_with()
|
||||
_dml.assert_called_once()
|
||||
|
||||
@mock.patch('kojihub._dml')
|
||||
def test_deleteNotification_missing(self, _dml):
|
||||
def test_deleteNotification_missing(self):
|
||||
user_id = 752
|
||||
n_id = 543
|
||||
self.exports.getBuildNotification.return_value = None
|
||||
self.exports.getBuildNotification.side_effect = koji.GenericError
|
||||
|
||||
with self.assertRaises(koji.GenericError):
|
||||
self.exports.deleteNotification(n_id)
|
||||
|
||||
self.exports.getBuildNotification.assert_called_once_with(n_id)
|
||||
_dml.assert_not_called()
|
||||
self.exports.getBuildNotification.assert_called_once_with(n_id, strict=True)
|
||||
|
||||
@mock.patch('kojihub._dml')
|
||||
def test_deleteNotification_not_logged(self, _dml):
|
||||
def test_deleteNotification_not_logged(self):
|
||||
user_id = 752
|
||||
n_id = 543
|
||||
self.exports.getBuildNotification.return_value = {'user_id': user_id}
|
||||
self.exports.getLoggedInUser.return_value = None
|
||||
#self.set_queries = ([
|
||||
# [{'user_id': 5, 'email': 'owner_name@%s' % self.context.opts['EmailDomain']}],
|
||||
#])
|
||||
|
||||
with self.assertRaises(koji.GenericError):
|
||||
self.exports.deleteNotification(n_id)
|
||||
|
||||
self.exports.getBuildNotification.assert_called_once_with(n_id)
|
||||
_dml.assert_not_called()
|
||||
self.exports.getBuildNotification.assert_called_once_with(n_id, strict=True)
|
||||
self.assertEqual(len(self.inserts), 0)
|
||||
self.assertEqual(len(self.updates), 0)
|
||||
self.assertEqual(len(self.queries), 0)
|
||||
|
||||
@mock.patch('kojihub._dml')
|
||||
def test_deleteNotification_no_perm(self, _dml):
|
||||
|
|
@ -481,7 +484,7 @@ class TestNotifications(unittest.TestCase):
|
|||
with self.assertRaises(koji.GenericError):
|
||||
self.exports.deleteNotification(n_id)
|
||||
|
||||
self.exports.getBuildNotification.assert_called_once_with(n_id)
|
||||
self.exports.getBuildNotification.assert_called_once_with(n_id, strict=True)
|
||||
_dml.assert_not_called()
|
||||
|
||||
|
||||
|
|
@ -549,7 +552,7 @@ class TestNotifications(unittest.TestCase):
|
|||
tag_id = 345
|
||||
success_only = True
|
||||
self.exports.getLoggedInUser.return_value = {'id': 1}
|
||||
self.exports.getBuildNotification.return_value = None
|
||||
self.exports.getBuildNotification.side_effect = koji.GenericError
|
||||
|
||||
with self.assertRaises(koji.GenericError):
|
||||
self.exports.updateNotification(n_id, package_id, tag_id, success_only)
|
||||
|
|
@ -627,3 +630,200 @@ class TestNotifications(unittest.TestCase):
|
|||
|
||||
self.assertEqual(len(self.inserts), 0)
|
||||
self.assertEqual(len(self.updates), 0)
|
||||
|
||||
###########################
|
||||
# Create notification block
|
||||
|
||||
@mock.patch('kojihub.get_build_notification_blocks')
|
||||
@mock.patch('kojihub.get_tag_id')
|
||||
@mock.patch('kojihub.get_package_id')
|
||||
def test_createNotificationBlock(self, get_package_id, get_tag_id,
|
||||
get_build_notification_blocks):
|
||||
user_id = 1
|
||||
package_id = 234
|
||||
tag_id = 345
|
||||
self.exports.getLoggedInUser.return_value = {'id': 1}
|
||||
self.exports.getUser.return_value = {'id': 2, 'name': 'username'}
|
||||
self.exports.hasPerm.return_value = True
|
||||
get_package_id.return_value = package_id
|
||||
get_tag_id.return_value = tag_id
|
||||
get_build_notification_blocks.return_value = []
|
||||
|
||||
r = self.exports.createNotificationBlock(user_id, package_id, tag_id)
|
||||
self.assertEqual(r, None)
|
||||
|
||||
self.exports.getLoggedInUser.assert_called_once()
|
||||
self.exports.getUser.asssert_called_once_with(user_id)
|
||||
self.exports.hasPerm.asssert_called_once_with('admin')
|
||||
get_package_id.assert_called_once_with(package_id, strict=True)
|
||||
get_tag_id.assert_called_once_with(tag_id, strict=True)
|
||||
get_build_notification_blocks.assert_called_once_with(2)
|
||||
self.assertEqual(len(self.inserts), 1)
|
||||
insert = self.inserts[0]
|
||||
self.assertEqual(insert.table, 'build_notifications_block')
|
||||
self.assertEqual(insert.data, {
|
||||
'package_id': package_id,
|
||||
'user_id': 2,
|
||||
'tag_id': tag_id,
|
||||
})
|
||||
self.assertEqual(insert.rawdata, {})
|
||||
|
||||
@mock.patch('kojihub.get_build_notification_blocks')
|
||||
@mock.patch('kojihub.get_tag_id')
|
||||
@mock.patch('kojihub.get_package_id')
|
||||
def test_createNotificationBlock_unauthentized(self, get_package_id, get_tag_id,
|
||||
get_build_notification_blocks):
|
||||
user_id = 1
|
||||
package_id = 234
|
||||
tag_id = 345
|
||||
self.exports.getLoggedInUser.return_value = None
|
||||
|
||||
with self.assertRaises(koji.GenericError):
|
||||
self.exports.createNotificationBlock(user_id, package_id, tag_id)
|
||||
|
||||
self.assertEqual(len(self.inserts), 0)
|
||||
|
||||
@mock.patch('kojihub.get_build_notification_blocks')
|
||||
@mock.patch('kojihub.get_tag_id')
|
||||
@mock.patch('kojihub.get_package_id')
|
||||
def test_createNotificationBlock_invalid_user(self, get_package_id, get_tag_id,
|
||||
get_build_notification_blocks):
|
||||
user_id = 2
|
||||
package_id = 234
|
||||
tag_id = 345
|
||||
self.exports.getLoggedInUser.return_value = {'id': 1}
|
||||
self.exports.getUser.return_value = None
|
||||
|
||||
with self.assertRaises(koji.GenericError):
|
||||
self.exports.createNotificationBlock(user_id, package_id, tag_id)
|
||||
|
||||
self.assertEqual(len(self.inserts), 0)
|
||||
|
||||
@mock.patch('kojihub.get_build_notification_blocks')
|
||||
@mock.patch('kojihub.get_tag_id')
|
||||
@mock.patch('kojihub.get_package_id')
|
||||
def test_createNotificationBlock_no_perm(self, get_package_id, get_tag_id,
|
||||
get_build_notification_blocks):
|
||||
user_id = 2
|
||||
package_id = 234
|
||||
tag_id = 345
|
||||
self.exports.getLoggedInUser.return_value = {'id': 1, 'name': 'a'}
|
||||
self.exports.getUser.return_value = {'id': 2, 'name': 'b'}
|
||||
self.exports.hasPerm.return_value = False
|
||||
|
||||
with self.assertRaises(koji.GenericError):
|
||||
self.exports.createNotificationBlock(user_id, package_id, tag_id)
|
||||
|
||||
self.assertEqual(len(self.inserts), 0)
|
||||
|
||||
@mock.patch('kojihub.get_build_notification_blocks')
|
||||
@mock.patch('kojihub.get_tag_id')
|
||||
@mock.patch('kojihub.get_package_id')
|
||||
def test_createNotificationBlock_invalid_pkg(self, get_package_id, get_tag_id,
|
||||
get_build_notification_blocks):
|
||||
user_id = 2
|
||||
package_id = 234
|
||||
tag_id = 345
|
||||
self.exports.getLoggedInUser.return_value = {'id': 2, 'name': 'a'}
|
||||
self.exports.getUser.return_value = {'id': 2, 'name': 'a'}
|
||||
get_package_id.side_effect = ValueError
|
||||
|
||||
with self.assertRaises(ValueError):
|
||||
self.exports.createNotificationBlock(user_id, package_id, tag_id)
|
||||
|
||||
self.assertEqual(len(self.inserts), 0)
|
||||
|
||||
@mock.patch('kojihub.get_build_notification_blocks')
|
||||
@mock.patch('kojihub.get_tag_id')
|
||||
@mock.patch('kojihub.get_package_id')
|
||||
def test_createNotificationBlock_invalid_tag(self, get_package_id, get_tag_id,
|
||||
get_build_notification_blocks):
|
||||
user_id = 2
|
||||
package_id = 234
|
||||
tag_id = 345
|
||||
self.exports.getLoggedInUser.return_value = {'id': 2, 'name': 'a'}
|
||||
self.exports.getUser.return_value = {'id': 2, 'name': 'a'}
|
||||
get_package_id.return_value = package_id
|
||||
get_tag_id.side_effect = ValueError
|
||||
|
||||
with self.assertRaises(ValueError):
|
||||
self.exports.createNotificationBlock(user_id, package_id, tag_id)
|
||||
|
||||
self.assertEqual(len(self.inserts), 0)
|
||||
|
||||
@mock.patch('kojihub.get_build_notification_blocks')
|
||||
@mock.patch('kojihub.get_tag_id')
|
||||
@mock.patch('kojihub.get_package_id')
|
||||
def test_createNotificationBlock_exists(self, get_package_id, get_tag_id,
|
||||
get_build_notification_blocks):
|
||||
user_id = 2
|
||||
package_id = 234
|
||||
tag_id = 345
|
||||
self.exports.getLoggedInUser.return_value = {'id': 2, 'name': 'a'}
|
||||
self.exports.getUser.return_value = {'id': 2, 'name': 'a'}
|
||||
get_package_id.return_value = package_id
|
||||
get_tag_id.return_value = tag_id
|
||||
get_build_notification_blocks.return_value = [{
|
||||
'package_id': package_id,
|
||||
'tag_id': tag_id,
|
||||
}]
|
||||
|
||||
with self.assertRaises(koji.GenericError):
|
||||
self.exports.createNotificationBlock(user_id, package_id, tag_id)
|
||||
|
||||
self.assertEqual(len(self.inserts), 0)
|
||||
|
||||
#####################
|
||||
# Delete notification
|
||||
@mock.patch('kojihub._dml')
|
||||
def test_deleteNotificationBlock(self, _dml):
|
||||
user_id = 752
|
||||
n_id = 543
|
||||
self.exports.getBuildNotificationBlock.return_value = {'user_id': user_id}
|
||||
|
||||
self.exports.deleteNotificationBlock(n_id)
|
||||
|
||||
self.exports.getBuildNotificationBlock.assert_called_once_with(n_id, strict=True)
|
||||
self.exports.getLoggedInUser.assert_called_once_with()
|
||||
_dml.assert_called_once()
|
||||
|
||||
def test_deleteNotification_missing(self):
|
||||
user_id = 752
|
||||
n_id = 543
|
||||
self.exports.getBuildNotificationBlock.side_effect = koji.GenericError
|
||||
|
||||
with self.assertRaises(koji.GenericError):
|
||||
self.exports.deleteNotificationBlock(n_id)
|
||||
|
||||
self.exports.getBuildNotificationBlock.assert_called_once_with(n_id, strict=True)
|
||||
|
||||
def test_deleteNotification_not_logged(self):
|
||||
user_id = 752
|
||||
n_id = 543
|
||||
self.exports.getBuildNotificationBlock.return_value = {'user_id': user_id}
|
||||
self.exports.getLoggedInUser.return_value = None
|
||||
#self.set_queries = ([
|
||||
# [{'user_id': 5, 'email': 'owner_name@%s' % self.context.opts['EmailDomain']}],
|
||||
#])
|
||||
|
||||
with self.assertRaises(koji.GenericError):
|
||||
self.exports.deleteNotificationBlock(n_id)
|
||||
|
||||
self.exports.getBuildNotificationBlock.assert_called_once_with(n_id, strict=True)
|
||||
self.assertEqual(len(self.inserts), 0)
|
||||
self.assertEqual(len(self.updates), 0)
|
||||
self.assertEqual(len(self.queries), 0)
|
||||
|
||||
@mock.patch('kojihub._dml')
|
||||
def test_deleteNotification_no_perm(self, _dml):
|
||||
user_id = 752
|
||||
n_id = 543
|
||||
self.exports.getBuildNotificationBlock.return_value = {'user_id': user_id}
|
||||
self.exports.getLoggedInUser.return_value = {'id': 1}
|
||||
self.exports.hasPerm.return_value = False
|
||||
|
||||
with self.assertRaises(koji.GenericError):
|
||||
self.exports.deleteNotificationBlock(n_id)
|
||||
|
||||
self.exports.getBuildNotificationBlock.assert_called_once_with(n_id, strict=True)
|
||||
_dml.assert_not_called()
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue