PR#773: create/edit notification checks for duplicity
Merges #773 https://pagure.io/koji/pull-request/773 Fixes: #687 https://pagure.io/koji/issue/687 createNotification doesn't check existing ones
This commit is contained in:
Mike McLean
commit
96a52c28c2
2 changed files with 413 additions and 16 deletions
|
|
@ -35,7 +35,6 @@ import errno
|
|||
import logging
|
||||
import fcntl
|
||||
import fnmatch
|
||||
import hashlib
|
||||
from koji.util import md5_constructor
|
||||
from koji.util import sha1_constructor
|
||||
from koji.util import dslice
|
||||
|
|
@ -53,7 +52,6 @@ import tarfile
|
|||
import tempfile
|
||||
import traceback
|
||||
import time
|
||||
import types
|
||||
import xmlrpclib
|
||||
import zipfile
|
||||
|
||||
|
|
@ -10825,14 +10823,24 @@ class RootExports(object):
|
|||
raise koji.GenericError('user %i cannot update notifications for user %i' % \
|
||||
(currentUser['id'], orig_notif['user_id']))
|
||||
|
||||
update = """UPDATE build_notifications
|
||||
SET package_id = %(package_id)s,
|
||||
tag_id = %(tag_id)s,
|
||||
success_only = %(success_only)s
|
||||
WHERE id = %(id)i
|
||||
"""
|
||||
# sanitize input
|
||||
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)
|
||||
success_only = bool(success_only)
|
||||
|
||||
_dml(update, locals())
|
||||
# check existing notifications to not have same twice
|
||||
for notification in get_build_notifications(orig_notif['user_id']):
|
||||
if (notification['package_id'] == package_id and
|
||||
notification['tag_id'] == tag_id and
|
||||
notification['success_only'] == success_only):
|
||||
raise koji.GenericError('notification already exists')
|
||||
|
||||
update = UpdateProcessor('build_notifications',
|
||||
clauses=['id = %(id)i'], values=locals())
|
||||
update.set(package_id=package_id, tag_id=tag_id, success_only=success_only)
|
||||
update.execute()
|
||||
|
||||
def createNotification(self, user_id, package_id, tag_id, success_only):
|
||||
"""Create a new notification. If the user_id does not match the currently logged-in user
|
||||
|
|
@ -10849,13 +10857,27 @@ class RootExports(object):
|
|||
raise koji.GenericError('user %s cannot create notifications 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)
|
||||
success_only = bool(success_only)
|
||||
|
||||
email = '%s@%s' % (notificationUser['name'], context.opts['EmailDomain'])
|
||||
insert = """INSERT INTO build_notifications
|
||||
(user_id, package_id, tag_id, success_only, email)
|
||||
VALUES
|
||||
(%(user_id)i, %(package_id)s, %(tag_id)s, %(success_only)s, %(email)s)
|
||||
"""
|
||||
_dml(insert, locals())
|
||||
|
||||
# check existing notifications to not have same twice
|
||||
for notification in get_build_notifications(user_id):
|
||||
if (notification['package_id'] == package_id and
|
||||
notification['tag_id'] == tag_id and
|
||||
notification['success_only'] == success_only):
|
||||
raise koji.GenericError('notification already exists')
|
||||
|
||||
insert = InsertProcessor('build_notifications')
|
||||
insert.set(user_id=user_id, package_id=package_id, tag_id=tag_id,
|
||||
success_only=success_only, email=email)
|
||||
insert.execute()
|
||||
|
||||
def deleteNotification(self, id):
|
||||
"""Delete the notification with the given ID. If the currently logged-in
|
||||
|
|
|
|||
|
|
@ -5,14 +5,28 @@ import koji
|
|||
import kojihub
|
||||
|
||||
QP = kojihub.QueryProcessor
|
||||
IP = kojihub.InsertProcessor
|
||||
UP = kojihub.UpdateProcessor
|
||||
|
||||
class TestNotifications(unittest.TestCase):
|
||||
def getInsert(self, *args, **kwargs):
|
||||
insert = IP(*args, **kwargs)
|
||||
insert.execute = mock.MagicMock()
|
||||
self.inserts.append(insert)
|
||||
return insert
|
||||
|
||||
class TestGetNotificationRecipients(unittest.TestCase):
|
||||
def getQuery(self, *args, **kwargs):
|
||||
query = QP(*args, **kwargs)
|
||||
query.execute = mock.MagicMock()
|
||||
self.queries.append(query)
|
||||
return query
|
||||
|
||||
def getUpdate(self, *args, **kwargs):
|
||||
update = UP(*args, **kwargs)
|
||||
update.execute = mock.MagicMock()
|
||||
self.updates.append(update)
|
||||
return update
|
||||
|
||||
def setUp(self):
|
||||
self.context = mock.patch('kojihub.context').start()
|
||||
self.context.opts = {
|
||||
|
|
@ -23,6 +37,18 @@ class TestGetNotificationRecipients(unittest.TestCase):
|
|||
self.QueryProcessor = mock.patch('kojihub.QueryProcessor',
|
||||
side_effect=self.getQuery).start()
|
||||
self.queries = []
|
||||
self.InsertProcessor = mock.patch('kojihub.InsertProcessor',
|
||||
side_effect=self.getInsert).start()
|
||||
self.inserts = []
|
||||
self.UpdateProcessor = mock.patch('kojihub.UpdateProcessor',
|
||||
side_effect=self.getUpdate).start()
|
||||
self.updates = []
|
||||
|
||||
self.exports = kojihub.RootExports()
|
||||
self.exports.getLoggedInUser = mock.MagicMock()
|
||||
self.exports.getUser = mock.MagicMock()
|
||||
self.exports.hasPerm = mock.MagicMock()
|
||||
self.exports.getBuildNotification = mock.MagicMock()
|
||||
|
||||
def tearDown(self):
|
||||
mock.patch.stopall()
|
||||
|
|
@ -144,3 +170,352 @@ class TestGetNotificationRecipients(unittest.TestCase):
|
|||
}
|
||||
emails = kojihub.get_notification_recipients(build, tag_id, state)
|
||||
self.assertEqual(emails, ['owner_name@test.domain.com'])
|
||||
|
||||
#####################
|
||||
# Create notification
|
||||
|
||||
@mock.patch('kojihub.get_build_notifications')
|
||||
@mock.patch('kojihub.get_tag_id')
|
||||
@mock.patch('kojihub.get_package_id')
|
||||
def test_createNotification(self, get_package_id, get_tag_id,
|
||||
get_build_notifications):
|
||||
user_id = 1
|
||||
package_id = 234
|
||||
tag_id = 345
|
||||
success_only = True
|
||||
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_notifications.return_value = []
|
||||
|
||||
r = self.exports.createNotification(user_id, package_id, tag_id, success_only)
|
||||
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_notifications.assert_called_once_with(2)
|
||||
self.assertEqual(len(self.inserts), 1)
|
||||
insert = self.inserts[0]
|
||||
self.assertEqual(insert.table, 'build_notifications')
|
||||
self.assertEqual(insert.data, {
|
||||
'package_id': package_id,
|
||||
'user_id': 2,
|
||||
'tag_id': tag_id,
|
||||
'success_only': success_only,
|
||||
'email': 'username@test.domain.com',
|
||||
})
|
||||
self.assertEqual(insert.rawdata, {})
|
||||
|
||||
@mock.patch('kojihub.get_build_notifications')
|
||||
@mock.patch('kojihub.get_tag_id')
|
||||
@mock.patch('kojihub.get_package_id')
|
||||
def test_createNotification_unauthentized(self, get_package_id, get_tag_id,
|
||||
get_build_notifications):
|
||||
user_id = 1
|
||||
package_id = 234
|
||||
tag_id = 345
|
||||
success_only = True
|
||||
self.exports.getLoggedInUser.return_value = None
|
||||
|
||||
with self.assertRaises(koji.GenericError):
|
||||
self.exports.createNotification(user_id, package_id, tag_id, success_only)
|
||||
|
||||
self.assertEqual(len(self.inserts), 0)
|
||||
|
||||
@mock.patch('kojihub.get_build_notifications')
|
||||
@mock.patch('kojihub.get_tag_id')
|
||||
@mock.patch('kojihub.get_package_id')
|
||||
def test_createNotification_invalid_user(self, get_package_id, get_tag_id,
|
||||
get_build_notifications):
|
||||
user_id = 2
|
||||
package_id = 234
|
||||
tag_id = 345
|
||||
success_only = True
|
||||
self.exports.getLoggedInUser.return_value = {'id': 1}
|
||||
self.exports.getUser.return_value = None
|
||||
|
||||
with self.assertRaises(koji.GenericError):
|
||||
self.exports.createNotification(user_id, package_id, tag_id, success_only)
|
||||
|
||||
self.assertEqual(len(self.inserts), 0)
|
||||
|
||||
@mock.patch('kojihub.get_build_notifications')
|
||||
@mock.patch('kojihub.get_tag_id')
|
||||
@mock.patch('kojihub.get_package_id')
|
||||
def test_createNotification_no_perm(self, get_package_id, get_tag_id,
|
||||
get_build_notifications):
|
||||
user_id = 2
|
||||
package_id = 234
|
||||
tag_id = 345
|
||||
success_only = True
|
||||
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.createNotification(user_id, package_id, tag_id, success_only)
|
||||
|
||||
self.assertEqual(len(self.inserts), 0)
|
||||
|
||||
@mock.patch('kojihub.get_build_notifications')
|
||||
@mock.patch('kojihub.get_tag_id')
|
||||
@mock.patch('kojihub.get_package_id')
|
||||
def test_createNotification_invalid_pkg(self, get_package_id, get_tag_id,
|
||||
get_build_notifications):
|
||||
user_id = 2
|
||||
package_id = 234
|
||||
tag_id = 345
|
||||
success_only = True
|
||||
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.createNotification(user_id, package_id, tag_id, success_only)
|
||||
|
||||
self.assertEqual(len(self.inserts), 0)
|
||||
|
||||
@mock.patch('kojihub.get_build_notifications')
|
||||
@mock.patch('kojihub.get_tag_id')
|
||||
@mock.patch('kojihub.get_package_id')
|
||||
def test_createNotification_invalid_tag(self, get_package_id, get_tag_id,
|
||||
get_build_notifications):
|
||||
user_id = 2
|
||||
package_id = 234
|
||||
tag_id = 345
|
||||
success_only = True
|
||||
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.createNotification(user_id, package_id, tag_id, success_only)
|
||||
|
||||
self.assertEqual(len(self.inserts), 0)
|
||||
|
||||
@mock.patch('kojihub.get_build_notifications')
|
||||
@mock.patch('kojihub.get_tag_id')
|
||||
@mock.patch('kojihub.get_package_id')
|
||||
def test_createNotification_exists(self, get_package_id, get_tag_id,
|
||||
get_build_notifications):
|
||||
user_id = 2
|
||||
package_id = 234
|
||||
tag_id = 345
|
||||
success_only = True
|
||||
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_notifications.return_value = [{
|
||||
'package_id': package_id,
|
||||
'tag_id': tag_id,
|
||||
'success_only': success_only,
|
||||
}]
|
||||
|
||||
with self.assertRaises(koji.GenericError):
|
||||
self.exports.createNotification(user_id, package_id, tag_id, success_only)
|
||||
|
||||
self.assertEqual(len(self.inserts), 0)
|
||||
|
||||
#####################
|
||||
# Delete notification
|
||||
@mock.patch('kojihub._dml')
|
||||
def test_deleteNotification(self, _dml):
|
||||
user_id = 752
|
||||
n_id = 543
|
||||
self.exports.getBuildNotification.return_value = {'user_id': user_id}
|
||||
|
||||
self.exports.deleteNotification(n_id)
|
||||
|
||||
self.exports.getBuildNotification.assert_called_once_with(n_id)
|
||||
self.exports.getLoggedInUser.assert_called_once_with()
|
||||
_dml.assert_called_once()
|
||||
|
||||
@mock.patch('kojihub._dml')
|
||||
def test_deleteNotification_missing(self, _dml):
|
||||
user_id = 752
|
||||
n_id = 543
|
||||
self.exports.getBuildNotification.return_value = None
|
||||
|
||||
with self.assertRaises(koji.GenericError):
|
||||
self.exports.deleteNotification(n_id)
|
||||
|
||||
self.exports.getBuildNotification.assert_called_once_with(n_id)
|
||||
_dml.assert_not_called()
|
||||
|
||||
@mock.patch('kojihub._dml')
|
||||
def test_deleteNotification_not_logged(self, _dml):
|
||||
user_id = 752
|
||||
n_id = 543
|
||||
self.exports.getBuildNotification.return_value = {'user_id': user_id}
|
||||
self.exports.getLoggedInUser.return_value = None
|
||||
|
||||
with self.assertRaises(koji.GenericError):
|
||||
self.exports.deleteNotification(n_id)
|
||||
|
||||
self.exports.getBuildNotification.assert_called_once_with(n_id)
|
||||
_dml.assert_not_called()
|
||||
|
||||
@mock.patch('kojihub._dml')
|
||||
def test_deleteNotification_no_perm(self, _dml):
|
||||
user_id = 752
|
||||
n_id = 543
|
||||
self.exports.getBuildNotification.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.deleteNotification(n_id)
|
||||
|
||||
self.exports.getBuildNotification.assert_called_once_with(n_id)
|
||||
_dml.assert_not_called()
|
||||
|
||||
|
||||
#####################
|
||||
# Update notification
|
||||
@mock.patch('kojihub.get_build_notifications')
|
||||
@mock.patch('kojihub.get_tag_id')
|
||||
@mock.patch('kojihub.get_package_id')
|
||||
def test_updateNotification(self, get_package_id, get_tag_id,
|
||||
get_build_notifications):
|
||||
n_id = 5432
|
||||
user_id = 1
|
||||
package_id = 234
|
||||
tag_id = 345
|
||||
success_only = True
|
||||
self.exports.getLoggedInUser.return_value = {'id': 1}
|
||||
self.exports.hasPerm.return_value = True
|
||||
get_package_id.return_value = package_id
|
||||
get_tag_id.return_value = tag_id
|
||||
get_build_notifications.return_value = [{
|
||||
'tag_id': tag_id,
|
||||
'user_id': user_id,
|
||||
'package_id': package_id,
|
||||
'success_only': not success_only,
|
||||
}]
|
||||
self.exports.getBuildNotification.return_value = {'user_id': user_id}
|
||||
|
||||
r = self.exports.updateNotification(n_id, package_id, tag_id, success_only)
|
||||
self.assertEqual(r, None)
|
||||
|
||||
self.exports.getLoggedInUser.assert_called_once()
|
||||
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_notifications.assert_called_once_with(user_id)
|
||||
self.assertEqual(len(self.inserts), 0)
|
||||
self.assertEqual(len(self.updates), 1)
|
||||
|
||||
@mock.patch('kojihub.get_build_notifications')
|
||||
@mock.patch('kojihub.get_tag_id')
|
||||
@mock.patch('kojihub.get_package_id')
|
||||
def test_updateNotification_not_logged(self, get_package_id, get_tag_id,
|
||||
get_build_notifications):
|
||||
n_id = 5432
|
||||
user_id = 1
|
||||
package_id = 234
|
||||
tag_id = 345
|
||||
success_only = True
|
||||
self.exports.getLoggedInUser.return_value = None
|
||||
|
||||
with self.assertRaises(koji.GenericError):
|
||||
self.exports.updateNotification(n_id, package_id, tag_id, success_only)
|
||||
|
||||
self.assertEqual(len(self.inserts), 0)
|
||||
self.assertEqual(len(self.updates), 0)
|
||||
|
||||
@mock.patch('kojihub.get_build_notifications')
|
||||
@mock.patch('kojihub.get_tag_id')
|
||||
@mock.patch('kojihub.get_package_id')
|
||||
def test_updateNotification_missing(self, get_package_id, get_tag_id,
|
||||
get_build_notifications):
|
||||
n_id = 5432
|
||||
user_id = 1
|
||||
package_id = 234
|
||||
tag_id = 345
|
||||
success_only = True
|
||||
self.exports.getLoggedInUser.return_value = {'id': 1}
|
||||
self.exports.getBuildNotification.return_value = None
|
||||
|
||||
with self.assertRaises(koji.GenericError):
|
||||
self.exports.updateNotification(n_id, package_id, tag_id, success_only)
|
||||
|
||||
self.assertEqual(len(self.inserts), 0)
|
||||
self.assertEqual(len(self.updates), 0)
|
||||
|
||||
@mock.patch('kojihub.get_build_notifications')
|
||||
@mock.patch('kojihub.get_tag_id')
|
||||
@mock.patch('kojihub.get_package_id')
|
||||
def test_updateNotification_no_perm(self, get_package_id, get_tag_id,
|
||||
get_build_notifications):
|
||||
n_id = 5432
|
||||
user_id = 1
|
||||
package_id = 234
|
||||
tag_id = 345
|
||||
success_only = True
|
||||
self.exports.getLoggedInUser.return_value = {'id': 132}
|
||||
self.exports.getBuildNotification.return_value = {'user_id': user_id}
|
||||
self.exports.hasPerm.return_value = False
|
||||
|
||||
with self.assertRaises(koji.GenericError):
|
||||
self.exports.updateNotification(n_id, package_id, tag_id, success_only)
|
||||
|
||||
self.assertEqual(len(self.inserts), 0)
|
||||
self.assertEqual(len(self.updates), 0)
|
||||
|
||||
@mock.patch('kojihub.get_build_notifications')
|
||||
@mock.patch('kojihub.get_tag_id')
|
||||
@mock.patch('kojihub.get_package_id')
|
||||
def test_updateNotification_exists(self, get_package_id, get_tag_id,
|
||||
get_build_notifications):
|
||||
n_id = 5432
|
||||
user_id = 1
|
||||
package_id = 234
|
||||
tag_id = 345
|
||||
success_only = True
|
||||
self.exports.getLoggedInUser.return_value = {'id': 1}
|
||||
self.exports.hasPerm.return_value = True
|
||||
get_package_id.return_value = package_id
|
||||
get_tag_id.return_value = tag_id
|
||||
get_build_notifications.return_value = [{
|
||||
'tag_id': tag_id,
|
||||
'user_id': user_id,
|
||||
'package_id': package_id,
|
||||
'success_only': success_only,
|
||||
}]
|
||||
self.exports.getBuildNotification.return_value = {'user_id': user_id}
|
||||
|
||||
with self.assertRaises(koji.GenericError):
|
||||
self.exports.updateNotification(n_id, package_id, tag_id, success_only)
|
||||
|
||||
self.exports.getLoggedInUser.assert_called_once()
|
||||
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_notifications.assert_called_once_with(user_id)
|
||||
self.assertEqual(len(self.inserts), 0)
|
||||
self.assertEqual(len(self.updates), 0)
|
||||
|
||||
@mock.patch('kojihub.get_build_notifications')
|
||||
@mock.patch('kojihub.get_tag_id')
|
||||
@mock.patch('kojihub.get_package_id')
|
||||
def test_updateNotification_not_logged(self, get_package_id, get_tag_id,
|
||||
get_build_notifications):
|
||||
n_id = 5432
|
||||
user_id = 1
|
||||
package_id = 234
|
||||
tag_id = 345
|
||||
success_only = True
|
||||
self.exports.getLoggedInUser.return_value = None
|
||||
|
||||
with self.assertRaises(koji.GenericError):
|
||||
self.exports.updateNotification(n_id, package_id, tag_id, success_only)
|
||||
|
||||
self.assertEqual(len(self.inserts), 0)
|
||||
self.assertEqual(len(self.updates), 0)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue