Add admin check when priority has negative value in wrapperRPM

Fixes: https://pagure.io/koji/issue/3320
This commit is contained in:
Jana Cupova 2022-04-12 07:31:08 +02:00 committed by Tomas Kopecek
parent 6caca6f49f
commit 56d4ec8e21
2 changed files with 46 additions and 1 deletions

View file

@ -10705,6 +10705,9 @@ class RootExports(object):
taskOpts = {}
if priority:
if priority < 0:
if not context.session.hasPerm('admin'):
raise koji.ActionNotAllowed('only admins may create high-priority tasks')
taskOpts['priority'] = koji.PRIO_DEFAULT + priority
convert_value(channel, cast=str, check_only=True)
taskOpts['channel'] = channel

View file

@ -54,6 +54,48 @@ class TestWrapperRPM(unittest.TestCase):
self.make_task.return_value = 123
self.get_channel.return_value = {'comment': None, 'description': None, 'enabled': True,
'id': 2, 'name': 'maven'}
with self.assertRaises(koji.ParameterError):
with self.assertRaises(koji.ParameterError) as cm:
self.exports.wrapperRPM(self.build, self.url, self.target,
priority=priority, channel=2)
self.assertEqual(f"Invalid type for value '2': {type(2)}", str(cm.exception))
def test_maven_not_supported(self):
self.context.opts.get.return_value = False
with self.assertRaises(koji.GenericError) as cm:
self.exports.wrapperRPM(self.build, self.url, self.target)
self.assertEqual("Maven support not enabled", str(cm.exception))
def test_wrapper_rpms_exists(self):
self.context.opts.get.return_value = True
self.exports.getBuild.return_value = self.buildinfo
self.list_rpms.return_value = [{'id': 1, 'buildroot_id': None}]
with self.assertRaises(koji.GenericError) as cm:
self.exports.wrapperRPM(self.build, self.url, self.target)
self.assertEqual("wrapper rpms for %s have already been built" % self.build,
str(cm.exception))
def test_no_repo_for_tag(self):
self.context.opts.get.return_value = True
self.exports.getBuild.return_value = self.buildinfo
self.list_rpms.return_value = []
self.exports.getBuildTarget.return_value = self.targetinfo
self.exports.getTag.return_value = self.taginfo
self.exports.getRepo.return_value = None
with self.assertRaises(koji.PreBuildError) as cm:
self.exports.wrapperRPM(self.build, self.url, self.target)
self.assertEqual("no repo for tag: %s" % self.taginfo['name'], str(cm.exception))
def test_priority_without_admin(self):
priority = -10
self.context.opts.get.return_value = True
self.exports.getBuild.return_value = self.buildinfo
self.exports.getBuildTarget.return_value = self.targetinfo
self.list_rpms.return_value = []
self.exports.getTag.return_value = self.taginfo
self.exports.getRepo.return_value = self.repoinfo
self.context.session.hasPerm.return_value = False
with self.assertRaises(koji.GenericError) as cm:
self.exports.wrapperRPM(self.build, self.url, self.target, priority=priority)
self.assertEqual("only admins may create high-priority tasks", str(cm.exception))