340 lines
15 KiB
Python
340 lines
15 KiB
Python
import os.path
|
|
import shutil
|
|
import tempfile
|
|
from unittest import mock
|
|
import unittest
|
|
|
|
import koji
|
|
import kojihub
|
|
from .utils import DBQueryTestCase
|
|
|
|
|
|
class TestGetRPM(DBQueryTestCase):
|
|
|
|
def setUp(self):
|
|
super(TestGetRPM, self).setUp()
|
|
self.exports = kojihub.RootExports()
|
|
self.context = mock.patch('kojihub.kojihub.context').start()
|
|
self.get_external_repo_id = mock.patch('kojihub.kojihub.get_external_repo_id').start()
|
|
self.find_build_id = mock.patch('kojihub.kojihub.find_build_id').start()
|
|
|
|
def tearDown(self):
|
|
mock.patch.stopall()
|
|
|
|
def test_wrong_type_rpminfo(self):
|
|
rpminfo = ['test-user']
|
|
with self.assertRaises(koji.GenericError) as cm:
|
|
kojihub.get_rpm(rpminfo)
|
|
self.assertEqual(f"Invalid type for rpminfo: {type(rpminfo)}", str(cm.exception))
|
|
|
|
def test_rpm_info_int(self):
|
|
rpminfo = 123
|
|
self.qp_execute_return_value = [{'rpminfo.id': 123}]
|
|
result = kojihub.get_rpm(rpminfo)
|
|
self.assertEqual(result, {'rpminfo.id': 123})
|
|
|
|
self.assertEqual(len(self.queries), 1)
|
|
query = self.queries[0]
|
|
self.assertEqual(query.tables, ['rpminfo'])
|
|
columns = ['rpminfo.id', 'build_id', 'buildroot_id', 'rpminfo.name', 'version', 'release',
|
|
'epoch', 'arch', 'draft', 'external_repo_id', 'external_repo.name',
|
|
'payloadhash', 'size', 'buildtime', 'metadata_only', 'extra']
|
|
self.assertEqual(set(query.columns), set(columns))
|
|
self.assertEqual(query.clauses, ["rpminfo.id=%(id)s"])
|
|
self.assertEqual(query.joins,
|
|
['external_repo ON rpminfo.external_repo_id = external_repo.id'])
|
|
self.assertEqual(query.values, {'id': rpminfo})
|
|
|
|
def test_rpm_info_empty_query_result_without_strict(self):
|
|
rpminfo = 123
|
|
self.qp_execute_one_return_value = {}
|
|
result = kojihub.get_rpm(rpminfo)
|
|
self.assertEqual(result, None)
|
|
|
|
self.assertEqual(len(self.queries), 1)
|
|
query = self.queries[0]
|
|
self.assertEqual(query.tables, ['rpminfo'])
|
|
columns = ['rpminfo.id', 'build_id', 'buildroot_id', 'rpminfo.name', 'version', 'release',
|
|
'epoch', 'arch', 'draft', 'external_repo_id', 'external_repo.name',
|
|
'payloadhash', 'size', 'buildtime', 'metadata_only', 'extra']
|
|
self.assertEqual(set(query.columns), set(columns))
|
|
self.assertEqual(query.clauses, ["rpminfo.id=%(id)s"])
|
|
self.assertEqual(query.joins,
|
|
['external_repo ON rpminfo.external_repo_id = external_repo.id'])
|
|
self.assertEqual(query.values, {'id': rpminfo})
|
|
|
|
def test_rpm_info_empty_query_result_with_strict(self):
|
|
rpminfo = 123
|
|
self.qp_execute_one_return_value = {}
|
|
with self.assertRaises(koji.GenericError) as ex:
|
|
kojihub.get_rpm(rpminfo, strict=True)
|
|
self.assertEqual("No such rpm: {'id': 123}", str(ex.exception))
|
|
|
|
self.assertEqual(len(self.queries), 1)
|
|
query = self.queries[0]
|
|
self.assertEqual(query.tables, ['rpminfo'])
|
|
columns = ['rpminfo.id', 'build_id', 'buildroot_id', 'rpminfo.name', 'version', 'release',
|
|
'epoch', 'arch', 'draft', 'external_repo_id', 'external_repo.name',
|
|
'payloadhash', 'size', 'buildtime', 'metadata_only', 'extra']
|
|
self.assertEqual(set(query.columns), set(columns))
|
|
self.assertEqual(query.clauses, ["rpminfo.id=%(id)s"])
|
|
self.assertEqual(query.joins,
|
|
['external_repo ON rpminfo.external_repo_id = external_repo.id'])
|
|
self.assertEqual(query.values, {'id': rpminfo})
|
|
|
|
def test_rpm_info_str(self):
|
|
rpminfo = 'testrpm-1.23-4.x86_64.rpm'
|
|
kojihub.get_rpm(rpminfo, multi=True)
|
|
|
|
self.assertEqual(len(self.queries), 1)
|
|
query = self.queries[0]
|
|
self.assertEqual(query.tables, ['rpminfo'])
|
|
columns = ['rpminfo.id', 'build_id', 'buildroot_id', 'rpminfo.name', 'version', 'release',
|
|
'epoch', 'arch', 'draft', 'external_repo_id', 'external_repo.name',
|
|
'payloadhash', 'size', 'buildtime', 'metadata_only', 'extra']
|
|
self.assertEqual(set(query.columns), set(columns))
|
|
self.assertEqual(query.clauses, ["rpminfo.name=%(name)s AND version=%(version)s "
|
|
"AND release=%(release)s AND arch=%(arch)s"])
|
|
self.assertEqual(query.joins,
|
|
['external_repo ON rpminfo.external_repo_id = external_repo.id'])
|
|
self.assertEqual(query.values, {'arch': 'x86_64', 'epoch': '', 'name': 'testrpm',
|
|
'release': '4', 'src': False, 'version': '1.23'})
|
|
|
|
def test_rpm_info_dict_location(self):
|
|
rpminfo = {'id': 123, 'name': 'testrpm-1.23-4.x86_64.rpm', 'location': 'test-location'}
|
|
self.get_external_repo_id.return_value = 125
|
|
rpminfo_data = rpminfo.copy()
|
|
rpminfo_data['external_repo_id'] = 125
|
|
|
|
kojihub.get_rpm(rpminfo, multi=True)
|
|
|
|
self.assertEqual(len(self.queries), 1)
|
|
query = self.queries[0]
|
|
self.assertEqual(query.tables, ['rpminfo'])
|
|
columns = ['rpminfo.id', 'build_id', 'buildroot_id', 'rpminfo.name', 'version', 'release',
|
|
'epoch', 'arch', 'draft', 'external_repo_id', 'external_repo.name',
|
|
'payloadhash', 'size', 'buildtime', 'metadata_only', 'extra']
|
|
self.assertEqual(set(query.columns), set(columns))
|
|
self.assertEqual(query.clauses,
|
|
["external_repo_id = %(external_repo_id)s", "rpminfo.id=%(id)s"])
|
|
self.assertEqual(query.joins,
|
|
['external_repo ON rpminfo.external_repo_id = external_repo.id'])
|
|
self.assertEqual(query.values, rpminfo_data)
|
|
|
|
@mock.patch('kojihub.kojihub._get_rpms')
|
|
def test_rpm_info_preferred(self, _get_rpms):
|
|
rpminfo = 'testrpm-1.23-4.x86_64.rpm'
|
|
rpm_ext = {'id': 1, 'external_repo_id': 2, 'draft': False}
|
|
rpm_ext2 = {'id': 2, 'external_repo_id': 2, 'draft': False}
|
|
rpm_draft = {'id': 3, 'external_repo_id': 0, 'draft': True}
|
|
rpm_draft2 = {'id': 4, 'external_repo_id': 0, 'draft': True}
|
|
rpm_nondraft = {'id': 5, 'external_repo_id': 0, 'draft': False}
|
|
|
|
# this test checks that the expected preferences are followed for multiple matches
|
|
|
|
_get_rpms.return_value = [rpm_ext, rpm_ext2, rpm_draft, rpm_draft2, rpm_nondraft]
|
|
ret = kojihub.get_rpm(rpminfo, multi=False)
|
|
self.assertEqual(ret, rpm_nondraft)
|
|
|
|
_get_rpms.return_value = [rpm_ext, rpm_ext2, rpm_draft, rpm_draft2]
|
|
ret = kojihub.get_rpm(rpminfo, multi=False)
|
|
self.assertEqual(ret, rpm_draft2)
|
|
|
|
_get_rpms.return_value = [rpm_ext, rpm_ext2, rpm_draft]
|
|
ret = kojihub.get_rpm(rpminfo, multi=False)
|
|
self.assertEqual(ret, rpm_draft)
|
|
|
|
_get_rpms.return_value = [rpm_ext, rpm_ext2]
|
|
ret = kojihub.get_rpm(rpminfo, multi=False)
|
|
self.assertEqual(ret, rpm_ext2)
|
|
|
|
_get_rpms.return_value = [rpm_ext]
|
|
ret = kojihub.get_rpm(rpminfo, multi=False)
|
|
self.assertEqual(ret, rpm_ext)
|
|
|
|
# multiple nondraft matches should error
|
|
rpm_bad = {'id': 6, 'external_repo_id': 0, 'draft': False}
|
|
_get_rpms.return_value = [rpm_nondraft, rpm_bad]
|
|
with self.assertRaises(koji.GenericError):
|
|
ret = kojihub.get_rpm(rpminfo, multi=False)
|
|
|
|
self.assertEqual(len(self.queries), 0) # _get_rpm is mocked
|
|
|
|
|
|
class TestGetRPMHeaders(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.exports = kojihub.RootExports()
|
|
self.exports.getLoggedInUser = mock.MagicMock()
|
|
self.context = mock.patch('kojihub.kojihub.context').start()
|
|
self.cursor = mock.MagicMock()
|
|
self.get_rpm = mock.patch('kojihub.kojihub.get_rpm').start()
|
|
self.get_build = mock.patch('kojihub.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
|
|
self.context.cnx.cursor.return_value = self.cursor
|
|
filepath = '../test/path'
|
|
with self.assertRaises(koji.GenericError) as cm:
|
|
self.exports.getRPMHeaders(taskID=99, filepath=filepath)
|
|
self.assertEqual(f"Invalid filepath: {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
|
|
self.context.cnx.cursor.return_value = self.cursor
|
|
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) as cm:
|
|
self.exports.getRPMHeaders(strict=False)
|
|
self.assertEqual("either rpmID or taskID and filepath must be specified",
|
|
str(cm.exception))
|
|
with self.assertRaises(koji.GenericError) as cm:
|
|
self.exports.getRPMHeaders(filepath='something')
|
|
self.assertEqual("either rpmID or taskID and filepath must be specified",
|
|
str(cm.exception))
|
|
# 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) as cm:
|
|
self.exports.getRPMHeaders(rpmID='FOO-1-1.noarch', strict=True)
|
|
self.assertEqual("NO SUCH RPM", str(cm.exception))
|
|
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):
|
|
rpm_info = {'external_repo_id': 1, 'id': 'RPMID'}
|
|
self.get_rpm.return_value = rpm_info
|
|
result = self.exports.getRPMHeaders(rpmID='FOO-1-1.noarch', strict=False)
|
|
self.assertEqual(result, {})
|
|
|
|
# again with strict mode
|
|
with self.assertRaises(koji.GenericError) as cm:
|
|
self.exports.getRPMHeaders(rpmID='FOO-1-1.noarch', strict=True)
|
|
self.assertEqual(f"External rpm: {rpm_info['id']}", str(cm.exception))
|
|
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}
|
|
build_info = {'nvr': 'NVR', 'state': koji.BUILD_STATES['DELETED']}
|
|
self.get_build.return_value = build_info
|
|
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) as cm:
|
|
self.exports.getRPMHeaders(rpmID='FOO-1-1.noarch', strict=True)
|
|
self.assertEqual(f"Build {build_info['nvr']} is deleted", str(cm.exception))
|
|
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:
|
|
self.exports.getRPMHeaders(rpmID='FOO-1-1.noarch', strict=True)
|
|
self.assertEqual(f"Missing rpm file: {rpmpath}", str(e.exception))
|
|
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) as cm:
|
|
self.exports.getRPMHeaders(taskID=taskid, filepath=filepath, strict=True)
|
|
self.assertEqual(f"Missing rpm file: {rpmpath}", str(cm.exception))
|
|
self.get_rpm.assert_not_called()
|
|
self.get_build.assert_not_called()
|
|
self.get_header_fields.assert_not_called()
|