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

Create DeleteProcessor class and use it

Browse source 
Fixes: https://pagure.io/koji/issue/3580
This commit is contained in:
Jana Cupova 2022-11-16 14:06:17 +01:00 • committed by Tomas Kopecek
parent fb202273b9
commit 78dee69de4
8 changed files with 219 additions and 189 deletions
Download patch file Download diff file Expand all files Collapse all files

52
tests/test_hub/test_delete_notification.py
View file

@ -4,29 +4,15 @@ import unittest
import koji
import kojihub
QP = kojihub.QueryProcessor
IP = kojihub.InsertProcessor
UP = kojihub.UpdateProcessor
DP = kojihub.DeleteProcessor
class TestDeleteNotifications(unittest.TestCase):
def getInsert(self, *args, **kwargs):
insert = IP(*args, **kwargs)
insert.execute = mock.MagicMock()
self.inserts.append(insert)
return insert
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 getDelete(self, *args, **kwargs):
delete = DP(*args, **kwargs)
delete.execute = mock.MagicMock()
self.deletes.append(delete)
return delete
def setUp(self):
self.context = mock.patch('kojihub.context').start()
@ -34,17 +20,9 @@ class TestDeleteNotifications(unittest.TestCase):
'EmailDomain': 'test.domain.com',
'NotifyOnSuccess': True,
}
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._dml = mock.patch('kojihub._dml').start()
self.DeleteProcessor = mock.patch('kojihub.DeleteProcessor',
side_effect=self.getDelete).start()
self.deletes = []
self.exports = kojihub.RootExports()
self.exports.getLoggedInUser = mock.MagicMock()
@ -60,16 +38,20 @@ class TestDeleteNotifications(unittest.TestCase):
self.exports.getBuildNotification.return_value = {'user_id': self.user_id}
self.exports.deleteNotification(self.n_id)
self.assertEqual(len(self.deletes), 1)
delete = self.deletes[0]
self.assertEqual(delete.table, 'build_notifications')
self.assertEqual(delete.clauses, ['id=%(id)i'])
self.exports.getBuildNotification.assert_called_once_with(self.n_id, strict=True)
self.exports.getLoggedInUser.assert_called_once_with()
self._dml.assert_called_once()
def test_deleteNotification_missing(self):
self.exports.getBuildNotification.side_effect = koji.GenericError
with self.assertRaises(koji.GenericError):
self.exports.deleteNotification(self.n_id)
self.assertEqual(len(self.deletes), 0)
self.exports.getBuildNotification.assert_called_once_with(self.n_id, strict=True)
@ -83,11 +65,9 @@ class TestDeleteNotifications(unittest.TestCase):
with self.assertRaises(koji.GenericError) as cm:
self.exports.deleteNotification(self.n_id)
self.assertEqual('Not logged-in', str(cm.exception))
self.assertEqual(len(self.deletes), 0)
self.exports.getBuildNotification.assert_called_once_with(self.n_id, strict=True)
self.assertEqual(len(self.inserts), 0)
self.assertEqual(len(self.updates), 0)
self.assertEqual(len(self.queries), 0)
def test_deleteNotification_no_perm(self):
self.exports.getBuildNotification.return_value = {'user_id': self.user_id}
@ -98,6 +78,6 @@ class TestDeleteNotifications(unittest.TestCase):
self.exports.deleteNotification(self.n_id)
self.assertEqual(f'user 1 cannot delete notifications for user {self.user_id}',
str(cm.exception))
self.assertEqual(len(self.deletes), 0)
self.exports.getBuildNotification.assert_called_once_with(self.n_id, strict=True)
self._dml.assert_not_called()

51
tests/test_hub/test_delete_notification_block.py
View file

@ -4,29 +4,15 @@ import unittest
import koji
import kojihub
QP = kojihub.QueryProcessor
IP = kojihub.InsertProcessor
UP = kojihub.UpdateProcessor
DP = kojihub.DeleteProcessor
class TestDeleteNotificationsBlocks(unittest.TestCase):
def getInsert(self, *args, **kwargs):
insert = IP(*args, **kwargs)
insert.execute = mock.MagicMock()
self.inserts.append(insert)
return insert
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 getDelete(self, *args, **kwargs):
delete = DP(*args, **kwargs)
delete.execute = mock.MagicMock()
self.deletes.append(delete)
return delete
def setUp(self):
self.context = mock.patch('kojihub.context').start()
@ -35,17 +21,10 @@ class TestDeleteNotificationsBlocks(unittest.TestCase):
'NotifyOnSuccess': True,
}
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.DeleteProcessor = mock.patch('kojihub.DeleteProcessor',
side_effect=self.getDelete).start()
self.deletes = []
self.get_user = mock.patch('kojihub.get_user').start()
self._dml = mock.patch('kojihub._dml').start()
self.exports = kojihub.RootExports()
self.exports.getLoggedInUser = mock.MagicMock()
@ -61,16 +40,20 @@ class TestDeleteNotificationsBlocks(unittest.TestCase):
self.exports.getBuildNotificationBlock.return_value = {'user_id': self.user_id}
self.exports.deleteNotificationBlock(self.n_id)
self.assertEqual(len(self.deletes), 1)
delete = self.deletes[0]
self.assertEqual(delete.table, 'build_notifications_block')
self.assertEqual(delete.clauses, ['id=%(id)i'])
self.exports.getBuildNotificationBlock.assert_called_once_with(self.n_id, strict=True)
self.exports.getLoggedInUser.assert_called_once_with()
self._dml.assert_called_once()
def test_deleteNotificationBlock_missing(self):
self.exports.getBuildNotificationBlock.side_effect = koji.GenericError
with self.assertRaises(koji.GenericError):
self.exports.deleteNotificationBlock(self.n_id)
self.assertEqual(len(self.deletes), 0)
self.exports.getBuildNotificationBlock.assert_called_once_with(self.n_id, strict=True)
@ -84,11 +67,9 @@ class TestDeleteNotificationsBlocks(unittest.TestCase):
with self.assertRaises(koji.GenericError) as cm:
self.exports.deleteNotificationBlock(self.n_id)
self.assertEqual('Not logged-in', str(cm.exception))
self.assertEqual(len(self.deletes), 0)
self.exports.getBuildNotificationBlock.assert_called_once_with(self.n_id, strict=True)
self.assertEqual(len(self.inserts), 0)
self.assertEqual(len(self.updates), 0)
self.assertEqual(len(self.queries), 0)
def test_deleteNotificationBlock_no_perm2(self):
self.exports.getBuildNotificationBlock.return_value = {'user_id': self.user_id}
@ -99,6 +80,6 @@ class TestDeleteNotificationsBlocks(unittest.TestCase):
self.exports.deleteNotificationBlock(self.n_id)
self.assertEqual(f'user 1 cannot delete notification blocks for user {self.user_id}',
str(cm.exception))
self.assertEqual(len(self.deletes), 0)
self.exports.getBuildNotificationBlock.assert_called_once_with(self.n_id, strict=True)
self._dml.assert_not_called()

67
tests/test_hub/test_delete_rpm_sig.py
View file

@ -5,21 +5,21 @@ import mock
import koji
import kojihub
QP = kojihub.QueryProcessor
DP = kojihub.DeleteProcessor
class TestDeleteRPMSig(unittest.TestCase):
def getQuery(self, *args, **kwargs):
query = QP(*args, **kwargs)
query.execute = mock.MagicMock()
self.queries.append(query)
return query
def getDelete(self, *args, **kwargs):
delete = DP(*args, **kwargs)
delete.execute = mock.MagicMock()
self.deletes.append(delete)
return delete
def setUp(self):
self.QueryProcessor = mock.patch('kojihub.QueryProcessor',
side_effect=self.getQuery).start()
self.queries = []
self.DeleteProcessor = mock.patch('kojihub.DeleteProcessor',
side_effect=self.getDelete).start()
self.deletes = []
self.get_rpm = mock.patch('kojihub.get_rpm').start()
self.query_rpm_sigs = mock.patch('kojihub.query_rpm_sigs').start()
self.get_build = mock.patch('kojihub.get_build').start()
@ -62,46 +62,39 @@ class TestDeleteRPMSig(unittest.TestCase):
def tearDown(self):
mock.patch.stopall()
@mock.patch('kojihub._dml')
def test_rpm_not_existing(self, dml):
def test_rpm_not_existing(self):
rpm_id = 1234
expected_msg = 'No such rpm: %s' % rpm_id
self.get_rpm.side_effect = koji.GenericError("No such rpm: %s" % rpm_id)
with self.assertRaises(koji.GenericError) as ex:
kojihub.delete_rpm_sig(rpm_id, all_sigs=True)
self.assertEqual(len(self.queries), 0)
self.assertEqual(len(self.deletes), 0)
self.assertEqual(ex.exception.args[0], expected_msg)
self.get_rpm.assert_called_once_with(rpm_id, strict=True)
self.query_rpm_sigs.assert_not_called()
dml.assert_not_called()
@mock.patch('kojihub._dml')
def test_not_all_sig_and_not_sigkey(self, dml):
def test_not_all_sig_and_not_sigkey(self):
expected_msg = 'No signature specified'
with self.assertRaises(koji.GenericError) as ex:
kojihub.delete_rpm_sig(1234)
self.assertEqual(len(self.queries), 0)
self.assertEqual(len(self.deletes), 0)
self.assertEqual(ex.exception.args[0], expected_msg)
self.get_rpm.assert_not_called()
self.query_rpm_sigs.assert_not_called()
dml.assert_not_called()
@mock.patch('kojihub._dml')
def test_external_repo(self, dml):
def test_external_repo(self):
rpminfo = 1234
rinfo = {'external_repo_id': 1, 'external_repo_name': 'INTERNAL'}
self.get_rpm.return_value = rinfo
with self.assertRaises(koji.GenericError) as ex:
kojihub.delete_rpm_sig(rpminfo, all_sigs=True)
self.assertEqual(len(self.queries), 0)
self.assertEqual(len(self.deletes), 0)
expected_msg = "Not an internal rpm: %s (from %s)" % (rpminfo, rinfo['external_repo_name'])
self.assertEqual(ex.exception.args[0], expected_msg)
self.get_rpm.assert_called_once_with(rpminfo, strict=True)
self.query_rpm_sigs.assert_not_called()
dml.assert_not_called()
@mock.patch('kojihub._dml')
def test_empty_query_sign(self, dml):
def test_empty_query_sign(self):
rpminfo = 1234
nvra = "%s-%s-%s.%s" % (self.rinfo['name'], self.rinfo['version'], self.rinfo['release'],
self.rinfo['arch'])
@ -110,33 +103,32 @@ class TestDeleteRPMSig(unittest.TestCase):
self.query_rpm_sigs.return_value = []
with self.assertRaises(koji.GenericError) as ex:
kojihub.delete_rpm_sig(rpminfo, all_sigs=True)
self.assertEqual(len(self.queries), 0)
self.assertEqual(len(self.deletes), 0)
self.assertEqual(ex.exception.args[0], expected_msg)
self.get_rpm.assert_called_once_with(rpminfo, strict=True)
self.query_rpm_sigs.assert_called_once_with(rpm_id=self.rinfo['id'], sigkey=None)
dml.assert_not_called()
@mock.patch('kojihub._dml')
@mock.patch('koji.pathinfo.build', return_value='fakebuildpath')
@mock.patch('os.remove')
def test_file_not_found_error(self, os_remove, pb, dml):
def test_file_not_found_error(self, os_remove, pb):
rpminfo = 2
os_remove.side_effect = FileNotFoundError()
self.get_rpm.return_value = self.rinfo
self.get_build.return_value = self.buildinfo
self.get_user.return_value = self.userinfo
self.query_rpm_sigs.return_value = self.queryrpmsigs
r = kojihub.delete_rpm_sig(rpminfo, all_sigs=True)
r = kojihub.delete_rpm_sig(rpminfo, sigkey='testkey')
self.assertEqual(r, None)
self.assertEqual(len(self.queries), 0)
delete = self.deletes[0]
self.assertEqual(delete.table, 'rpmsigs')
self.assertEqual(delete.clauses, ["rpm_id=%(rpm_id)i", "sigkey=%(sigkey)s"])
self.get_rpm.assert_called_once_with(rpminfo, strict=True)
self.query_rpm_sigs.assert_called_once_with(rpm_id=self.rinfo['id'], sigkey=None)
self.query_rpm_sigs.assert_called_once_with(rpm_id=self.rinfo['id'], sigkey='testkey')
self.get_build.assert_called_once_with(self.rinfo['build_id'])
@mock.patch('kojihub._dml')
@mock.patch('koji.pathinfo.build', return_value='fakebuildpath')
@mock.patch('os.remove', side_effect=OSError)
def test_not_valid(self, os_remove, pb, dml):
def test_not_valid(self, os_remove, pb):
rpminfo = 2
filepath = 'fakebuildpath/data/signed/x86_64/fs_mark-3.3-20.el8.x86_64.rpm'
self.get_rpm.return_value = self.rinfo
@ -146,21 +138,26 @@ class TestDeleteRPMSig(unittest.TestCase):
with self.assertRaises(koji.GenericError) as ex:
kojihub.delete_rpm_sig(rpminfo, all_sigs=True)
self.assertEqual(ex.exception.args[0], expected_msg)
self.assertEqual(len(self.queries), 0)
delete = self.deletes[0]
self.assertEqual(delete.table, 'rpmsigs')
self.assertEqual(delete.clauses, ["rpm_id=%(rpm_id)i"])
self.get_rpm.assert_called_once_with(rpminfo, strict=True)
self.query_rpm_sigs.assert_called_once_with(rpm_id=self.rinfo['id'], sigkey=None)
self.get_build.assert_called_once_with(self.rinfo['build_id'])
@mock.patch('kojihub._dml')
@mock.patch('koji.pathinfo.build', return_value='fakebuildpath')
@mock.patch('os.remove')
def test_valid(self, os_remove, pb, dml):
def test_valid(self, os_remove, pb):
rpminfo = 2
self.get_rpm.return_value = self.rinfo
self.get_build.return_value = self.buildinfo
self.get_user.return_value = self.userinfo
self.query_rpm_sigs.return_value = self.queryrpmsigs
kojihub.delete_rpm_sig(rpminfo, all_sigs=True)
self.assertEqual(len(self.deletes), 1)
delete = self.deletes[0]
self.assertEqual(delete.table, 'rpmsigs')
self.assertEqual(delete.clauses, ["rpm_id=%(rpm_id)i"])
self.get_rpm.assert_called_once_with(rpminfo, strict=True)
self.query_rpm_sigs.assert_called_once_with(rpm_id=self.rinfo['id'], sigkey=None)
self.get_build.assert_called_once_with(self.rinfo['build_id'])