diff --git a/cli/koji_cli/commands.py b/cli/koji_cli/commands.py index 7a67f809..419859cf 100644 --- a/cli/koji_cli/commands.py +++ b/cli/koji_cli/commands.py @@ -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) diff --git a/hub/kojihub.py b/hub/kojihub.py index 66b83be9..5d8b007b 100644 --- a/hub/kojihub.py +++ b/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 diff --git a/tests/test_cli/data/list-commands.txt b/tests/test_cli/data/list-commands.txt index a33fab26..972f42f5 100644 --- a/tests/test_cli/data/list-commands.txt +++ b/tests/test_cli/data/list-commands.txt @@ -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 diff --git a/tests/test_cli/test_list_notifications.py b/tests/test_cli/test_list_notifications.py index 53893d34..01e668ab 100644 --- a/tests/test_cli/test_list_notifications.py +++ b/tests/test_cli/test_list_notifications.py @@ -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 diff --git a/tests/test_hub/test_notifications.py b/tests/test_hub/test_notifications.py index c1c512f7..3023c32d 100644 --- a/tests/test_hub/test_notifications.py +++ b/tests/test_hub/test_notifications.py @@ -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()