From 1820bc12dde8b91bfd1a9eb6f7d0aa4eb690baad Mon Sep 17 00:00:00 2001 From: Mike McLean Date: Feb 15 2022 21:57:56 +0000 Subject: [PATCH 1/2] add strict option to getRPMHeaders related: https://pagure.io/koji/issue/3178 --- diff --git a/hub/kojihub.py b/hub/kojihub.py index 81cdbc6..f769fa1 100644 --- a/hub/kojihub.py +++ b/hub/kojihub.py @@ -12235,23 +12235,50 @@ class RootExports(object): "No file: %s found in RPM: %s" % (filename, rpmID)) return {} - def getRPMHeaders(self, rpmID=None, taskID=None, filepath=None, headers=None): + def getRPMHeaders(self, rpmID=None, taskID=None, filepath=None, headers=None, strict=False): """ - Get the requested headers from the rpm. Header names are case-insensitive. - If a header is requested that does not exist an exception will be raised. - Returns a map of header names to values. If the specified ID - is not valid or the rpm does not exist on the file system, an empty map - will be returned. + Get the requested headers from the rpm, specified either by rpmID or taskID + filepath + + If the specified ID is not valid or the rpm does not exist on the file system an empty + map will be returned, unless strict=True is given. + + Header names are case-insensitive. If a header is requested that does not exist an + exception will be raised (regardless of strict option). + + :param int|str rpmID: query the specified rpm + :param int taskID: query a file from the specified task (filepath must also be passed) + :param str filepath: the rpm path relative to the task directory + :param list headers: a list of rpm header names (as strings) + :param bool strict: raise an exception for invalid or missing rpms/paths + :returns dict: a map of header names to values """ if rpmID: - rpm_info = get_rpm(rpmID) - if not rpm_info or not rpm_info['build_id']: + rpm_info = get_rpm(rpmID, strict=strict) + if not rpm_info: + # can only happen if not strict return {} - build_info = get_build(rpm_info['build_id']) + if rpm_info['external_repo_id'] != 0: + if strict: + raise koji.GenericError('External rpm: %(id)s' % rpm_info) + else: + return {} + # get_build should be strict regardless since this is an internal rpm + build_info = get_build(rpm_info['build_id'], strict=True) + build_state = koji.BUILD_STATES[build_info['state']] + if build_state == 'DELETED': + if strict: + raise koji.GenericError('Build %(nvr)s is deleted' % build_info) + else: + return {} rpm_path = joinpath(koji.pathinfo.build(build_info), koji.pathinfo.rpm(rpm_info)) if not os.path.exists(rpm_path): - return {} + if strict: + raise koji.GenericError('Missing rpm file: %s' % rpm_path) + else: + # strict or not, this is still unexpected + logger.error('Missing rpm file: %s' % rpm_path) + return {} elif taskID: if not filepath: raise koji.GenericError('filepath must be specified with taskID') @@ -12260,6 +12287,11 @@ class RootExports(object): rpm_path = joinpath(koji.pathinfo.work(), koji.pathinfo.taskrelpath(taskID), filepath) + if not os.path.exists(rpm_path): + if strict: + raise koji.GenericError('Missing rpm file: %s' % rpm_path) + else: + return {} else: raise koji.GenericError('either rpmID or taskID and filepath must be specified') From e3fd853a69a01a3e3668072dd5ecabd9ad49bf6b Mon Sep 17 00:00:00 2001 From: Mike McLean Date: Feb 16 2022 00:32:45 +0000 Subject: [PATCH 2/2] add unit tests --- diff --git a/tests/test_hub/test_getRPM.py b/tests/test_hub/test_getRPM.py index 7b10fc6..7d25c55 100644 --- a/tests/test_hub/test_getRPM.py +++ b/tests/test_hub/test_getRPM.py @@ -1,3 +1,6 @@ +import os.path +import shutil +import tempfile import unittest import mock @@ -21,6 +24,16 @@ class TestGetRPMHeaders(unittest.TestCase): self.exports.getLoggedInUser = mock.MagicMock() self.context = mock.patch('kojihub.context').start() self.cursor = mock.MagicMock() + self.get_rpm = mock.patch('kojihub.get_rpm').start() + self.get_build = mock.patch('kojihub.get_build').start() + self.get_header_fields = mock.patch('koji.get_header_fields').start() + self.tempdir = tempfile.mkdtemp() + self.pathinfo = koji.PathInfo(self.tempdir) + mock.patch('koji.pathinfo', new=self.pathinfo).start() + + def tearDown(self): + mock.patch.stopall() + shutil.rmtree(self.tempdir) def test_taskid_invalid_path(self): self.cursor.fetchone.return_value = None @@ -29,6 +42,9 @@ class TestGetRPMHeaders(unittest.TestCase): with self.assertRaises(koji.GenericError) as cm: self.exports.getRPMHeaders(taskID=99, filepath=filepath) self.assertEqual("Invalid filepath: %s" % filepath, str(cm.exception)) + self.get_rpm.assert_not_called() + self.get_build.assert_not_called() + self.get_header_fields.assert_not_called() def test_taskid_without_filepath(self): self.cursor.fetchone.return_value = None @@ -36,3 +52,135 @@ class TestGetRPMHeaders(unittest.TestCase): with self.assertRaises(koji.GenericError) as cm: self.exports.getRPMHeaders(taskID=99) self.assertEqual("filepath must be specified with taskID", str(cm.exception)) + self.get_rpm.assert_not_called() + self.get_build.assert_not_called() + self.get_header_fields.assert_not_called() + + def test_insufficient_args(self): + with self.assertRaises(koji.GenericError): + result = self.exports.getRPMHeaders(strict=False) + with self.assertRaises(koji.GenericError): + result = self.exports.getRPMHeaders(filepath='something') + # we already test taskid without filepath above + + def test_unknown_rpm(self): + self.get_rpm.return_value = None + result = self.exports.getRPMHeaders(rpmID='FOO-1-1.noarch', strict=False) + self.assertEqual(result, {}) + self.get_rpm.assert_called_with('FOO-1-1.noarch', strict=False) + + # again with strict mode + self.get_rpm.side_effect = koji.GenericError('NO SUCH RPM') + with self.assertRaises(koji.GenericError): + result = self.exports.getRPMHeaders(rpmID='FOO-1-1.noarch', strict=True) + self.get_rpm.assert_called_with('FOO-1-1.noarch', strict=True) + self.get_build.assert_not_called() + self.get_header_fields.assert_not_called() + + def test_external_rpm(self): + self.get_rpm.return_value = {'external_repo_id': 1, 'id': 'RPMID'} + result = self.exports.getRPMHeaders(rpmID='FOO-1-1.noarch', strict=False) + self.assertEqual(result, {}) + + # again with strict mode + with self.assertRaises(koji.GenericError): + result = self.exports.getRPMHeaders(rpmID='FOO-1-1.noarch', strict=True) + self.get_build.assert_not_called() + self.get_header_fields.assert_not_called() + + def test_deleted_build(self): + self.get_rpm.return_value = {'build_id': 'BUILDID', 'external_repo_id': 0} + self.get_build.return_value = {'nvr': 'NVR', 'state': koji.BUILD_STATES['DELETED']} + result = self.exports.getRPMHeaders(rpmID='FOO-1-1.noarch', strict=False) + self.assertEqual(result, {}) + self.get_build.assert_called_with('BUILDID', strict=True) + + # again with strict mode + with self.assertRaises(koji.GenericError): + result = self.exports.getRPMHeaders(rpmID='FOO-1-1.noarch', strict=True) + self.get_build.assert_called_with('BUILDID', strict=True) + self.get_header_fields.assert_not_called() + + def test_missing_rpm(self): + self.get_rpm.return_value = { + 'build_id': 'BUILDID', + 'external_repo_id': 0, + 'name': 'pkg', + 'version': '1', + 'release': '2', + 'arch': 'noarch'} + self.get_build.return_value = { + 'name': 'pkg', + 'version': '1', + 'release': '2', + 'nvr': 'pkg-1-2', + 'state': koji.BUILD_STATES['COMPLETE']} + rpmpath = '%s/packages/pkg/1/2/noarch/pkg-1-2.noarch.rpm' % self.tempdir + + # rpm does not exist + result = self.exports.getRPMHeaders(rpmID='pkg-1-2.noarch', strict=False) + self.assertEqual(result, {}) + self.get_build.assert_called_with('BUILDID', strict=True) + + # again with strict mode + with self.assertRaises(koji.GenericError) as e: + result = self.exports.getRPMHeaders(rpmID='FOO-1-1.noarch', strict=True) + self.get_build.assert_called_with('BUILDID', strict=True) + self.get_header_fields.assert_not_called() + + def test_rpm_exists(self): + self.get_rpm.return_value = { + 'build_id': 'BUILDID', + 'external_repo_id': 0, + 'name': 'pkg', + 'version': '1', + 'release': '2', + 'arch': 'noarch'} + self.get_build.return_value = { + 'name': 'pkg', + 'version': '1', + 'release': '2', + 'nvr': 'pkg-1-2', + 'state': koji.BUILD_STATES['COMPLETE']} + rpmpath = '%s/packages/pkg/1/2/noarch/pkg-1-2.noarch.rpm' % self.tempdir + koji.ensuredir(os.path.dirname(rpmpath)) + with open(rpmpath, 'w') as fo: + fo.write('hello world') + fakeheaders = {'HEADER': 'SOMETHING'} + self.get_header_fields.return_value = fakeheaders + + result = self.exports.getRPMHeaders(rpmID='pkg-1-2.noarch', strict=True) + self.assertEqual(result, fakeheaders) + self.get_build.assert_called_with('BUILDID', strict=True) + self.get_header_fields.assert_called_with(rpmpath, None) + + def test_task_rpm_exists(self): + taskid = 137 + filepath = 'pkg-1-2.noarch.rpm' + rpmpath = '%s/work/tasks/137/137/pkg-1-2.noarch.rpm' % self.tempdir + koji.ensuredir(os.path.dirname(rpmpath)) + with open(rpmpath, 'w') as fo: + fo.write('hello world') + fakeheaders = {'HEADER': 'SOMETHING'} + self.get_header_fields.return_value = fakeheaders + + result = self.exports.getRPMHeaders(taskID=taskid, filepath=filepath, strict=False) + self.assertEqual(result, fakeheaders) + self.get_rpm.assert_not_called() + self.get_build.assert_not_called() + self.get_header_fields.assert_called_with(rpmpath, None) + + def test_task_rpm_missing(self): + taskid = 137 + filepath = 'pkg-1-2.noarch.rpm' + rpmpath = '%s/work/tasks/137/137/pkg-1-2.noarch.rpm' % self.tempdir + + result = self.exports.getRPMHeaders(taskID=taskid, filepath=filepath, strict=False) + self.assertEqual(result, {}) + + with self.assertRaises(koji.GenericError): + result = self.exports.getRPMHeaders(taskID=taskid, filepath=filepath, strict=True) + + self.get_rpm.assert_not_called() + self.get_build.assert_not_called() + self.get_header_fields.assert_not_called()