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

Add extra param to listTagged*

Browse source 
Fixes: https://pagure.io/koji/issue/3110
This commit is contained in:
Jana Cupova 2022-03-14 10:07:58 +01:00 • committed by Tomas Kopecek
parent c9bca5b910
commit 75d545e740
8 changed files with 733 additions and 151 deletions
Download patch file Download diff file Expand all files Collapse all files

43
tests/test_hub/test_list_tagged.py
View file

@ -3,6 +3,7 @@ import unittest
import koji
import kojihub
import mock
import copy
QP = kojihub.QueryProcessor
@ -16,6 +17,15 @@ class TestListTagged(unittest.TestCase):
self.exports = kojihub.RootExports()
self.context = mock.patch('kojihub.context').start()
self.cursor = mock.MagicMock()
self.readTaggedBuilds = mock.patch('kojihub.readTaggedBuilds').start()
self.tag_name = 'test-tag'
self.taginfo = {'id': 1, 'name': 'tag'}
self.tagged_build = [
{'build_id': 1, 'create_event': 1172, 'creation_event_id': 1171, 'epoch': None,
'id': 1, 'name': 'test-pkg', 'nvr': 'test-pkg-2.52-1.fc35', 'owner_id': 1,
'owner_name': 'kojiuser', 'package_id': 1, 'package_name': 'test-pkg',
'release': '1.fc35', 'state': 1, 'tag_id': 1, 'tag_name': 'test-tag',
'task_id': None, 'version': '2.52', 'volume_id': 0, 'volume_name': 'DEFAULT'}]
def getQuery(self, *args, **kwargs):
query = QP(*args, **kwargs)
@ -26,13 +36,38 @@ class TestListTagged(unittest.TestCase):
def test_non_exist_tag_without_strict(self):
self.get_tag.return_value = None
self.exports.listTagged('non-existing-tag', strict=False)
self.exports.listTagged(self.tag_name, strict=False)
self.assertEqual(len(self.queries), 0)
def test_non_exist_tag_with_strict(self):
tag_name = 'non-existing-tag'
error_message = "No such tagInfo: '%s'" % tag_name
error_message = "No such tagInfo: '%s'" % self.tag_name
self.get_tag.side_effect = koji.GenericError(error_message)
with self.assertRaises(koji.GenericError) as cm:
self.exports.listTagged(tag_name)
self.exports.listTagged(self.tag_name)
self.assertEqual(error_message, str(cm.exception))
def test_list_tagged(self):
self.get_tag.return_value = self.taginfo
self.readTaggedBuilds.return_value = self.tagged_build
rv = self.exports.listTagged(self.tag_name)
self.assertEqual(rv, self.tagged_build)
def test_list_tagged_with_extra(self):
self.get_tag.return_value = self.taginfo
tagged_build = copy.deepcopy(self.tagged_build)
tagged_build[0]['extra'] = 'extra_value'
self.readTaggedBuilds.return_value = tagged_build
rv = self.exports.listTagged(self.tag_name, extra=True)
self.assertEqual(rv, tagged_build)
def test_list_tagged_with_prefix(self):
self.get_tag.return_value = self.taginfo
self.readTaggedBuilds.return_value = self.tagged_build
rv = self.exports.listTagged(self.tag_name, prefix='test')
self.assertEqual(rv, self.tagged_build)
def test_list_tagged_with_prefix_empty_result(self):
self.get_tag.return_value = self.taginfo
self.readTaggedBuilds.return_value = self.tagged_build
rv = self.exports.listTagged(self.tag_name, prefix='pkg')
self.assertEqual(rv, [])

52
tests/test_hub/test_list_tagged_archives.py
View file

@ -3,6 +3,7 @@ import unittest
import koji
import kojihub
import mock
import copy
QP = kojihub.QueryProcessor
@ -16,6 +17,36 @@ class TestListTaggedArchives(unittest.TestCase):
self.exports = kojihub.RootExports()
self.context = mock.patch('kojihub.context').start()
self.cursor = mock.MagicMock()
self.readTaggedArchives = mock.patch('kojihub.readTaggedArchives').start()
self.tag_name = 'test-tag'
self.taginfo = {'id': 1, 'name': 'tag'}
self.tagged_archives = [
[{'btype': 'maven',
'btype_id': 2,
'build_id': 1,
'checksum': '75a8f47a6626fa1934b3ca5b2c51eb4d2c3bf7c88c621d5327af0a64e08797f6',
'checksum_type': 2,
'filename': 'test-1.4.7.pom',
'id': 1,
'type_id': 3}],
[{'build_id': 1,
'epoch': 7,
'id': 1,
'name': 'kojitest-rpm',
'nvr': 'kojitest-rpm-1.1-11',
'owner_id': 1,
'owner_name': 'kojiadmin',
'package_id': 1,
'package_name': 'kojitest-rpm',
'release': '11',
'state': 1,
'tag_id': self.taginfo['id'],
'tag_name': self.tag_name,
'task_id': 3,
'version': '1.1',
'volume_id': 0,
'volume_name': 'DEFAULT'}]
]
def getQuery(self, *args, **kwargs):
query = QP(*args, **kwargs)
@ -26,13 +57,26 @@ class TestListTaggedArchives(unittest.TestCase):
def test_non_exist_tag_without_strict(self):
self.get_tag.return_value = None
self.exports.listTagged('non-existing-tag', strict=False)
self.exports.listTaggedArchives(self.tag_name, strict=False)
self.assertEqual(len(self.queries), 0)
def test_non_exist_tag_with_strict(self):
tag_name = 'non-existing-tag'
error_message = "No such tagInfo: '%s'" % tag_name
error_message = "No such tagInfo: '%s'" % self.tag_name
self.get_tag.side_effect = koji.GenericError(error_message)
with self.assertRaises(koji.GenericError) as cm:
self.exports.listTagged(tag_name)
self.exports.listTaggedArchives(self.tag_name)
self.assertEqual(error_message, str(cm.exception))
def test_list_tagged_archives_default(self):
self.get_tag.return_value = self.taginfo
tagged_archives = copy.deepcopy(self.tagged_archives)
tagged_archives[0][0]['extra'] = 'extra_value'
self.readTaggedArchives.return_value = tagged_archives
rv = self.exports.listTaggedArchives(self.tag_name)
self.assertEqual(rv, tagged_archives)
def test_list_tagged_archives_without_extra(self):
self.get_tag.return_value = self.taginfo
self.readTaggedArchives.return_value = self.tagged_archives
rv = self.exports.listTaggedArchives(self.tag_name, extra=False)
self.assertEqual(rv, self.tagged_archives)

51
tests/test_hub/test_list_tagged_rpms.py
View file

@ -3,6 +3,7 @@ import unittest
import koji
import kojihub
import mock
import copy
QP = kojihub.QueryProcessor
@ -16,6 +17,35 @@ class TestListTaggedRPMS(unittest.TestCase):
self.exports = kojihub.RootExports()
self.context = mock.patch('kojihub.context').start()
self.cursor = mock.MagicMock()
self.readTaggedRPMS = mock.patch('kojihub.readTaggedRPMS').start()
self.tag_name = 'test-tag'
self.taginfo = {'id': 1, 'name': 'tag'}
self.tagged_rpms = [
[{'arch': 'src',
'build_id': 1,
'epoch': 7,
'id': 158,
'name': 'kojitest-rpm',
'release': '11',
'version': '1.1'}],
[{'build_id': 1,
'epoch': 7,
'id': 1,
'name': 'kojitest-rpm',
'nvr': 'kojitest-rpm-1.1-11',
'owner_id': 1,
'owner_name': 'kojiadmin',
'package_id': 1,
'package_name': 'kojitest-rpm',
'release': '11',
'state': 1,
'tag_id': self.taginfo['id'],
'tag_name': self.tag_name,
'task_id': 3,
'version': '1.1',
'volume_id': 0,
'volume_name': 'DEFAULT'}]
]
def getQuery(self, *args, **kwargs):
query = QP(*args, **kwargs)
@ -26,13 +56,26 @@ class TestListTaggedRPMS(unittest.TestCase):
def test_non_exist_tag_without_strict(self):
self.get_tag.return_value = None
self.exports.listTaggedRPMS('non-existing-tag', strict=False)
self.exports.listTaggedRPMS(self.tag_name, strict=False)
self.assertEqual(len(self.queries), 0)
def test_non_exist_tag_with_strict(self):
tag_name = 'non-existing-tag'
error_message = "No such tagInfo: '%s'" % tag_name
error_message = "No such tagInfo: '%s'" % self.tag_name
self.get_tag.side_effect = koji.GenericError(error_message)
with self.assertRaises(koji.GenericError) as cm:
self.exports.listTaggedRPMS(tag_name)
self.exports.listTaggedRPMS(self.tag_name)
self.assertEqual(error_message, str(cm.exception))
def test_list_tagged_archives_default(self):
self.get_tag.return_value = self.taginfo
tagged_rpms = copy.deepcopy(self.tagged_rpms)
tagged_rpms[0][0]['extra'] = 'extra_value'
self.readTaggedRPMS.return_value = tagged_rpms
rv = self.exports.listTaggedRPMS(self.tag_name)
self.assertEqual(rv, tagged_rpms)
def test_list_tagged_archives_without_extra(self):
self.get_tag.return_value = self.taginfo
self.readTaggedRPMS.return_value = self.tagged_rpms
rv = self.exports.listTaggedRPMS(self.tag_name, extra=False)
self.assertEqual(rv, self.tagged_rpms)

127
tests/test_hub/test_read_tagged_archives.py Normal file
View file

@ -0,0 +1,127 @@
import unittest
import mock
import koji
import kojihub
import copy
QP = kojihub.QueryProcessor
class TestReadTaggedArchives(unittest.TestCase):
def getQuery(self, *args, **kwargs):
self.maxDiff = None
query = QP(*args, **kwargs)
query.execute = mock.MagicMock()
query.executeOne = mock.MagicMock()
self.queries.append(query)
return query
def setUp(self):
self.QueryProcessor = mock.patch('kojihub.QueryProcessor',
side_effect=self.getQuery).start()
self.queries = []
self.context = mock.patch('kojihub.context').start()
# It seems MagicMock will not automatically handle attributes that
# start with "assert"
self.exports = kojihub.RootExports()
self.readTaggedBuilds = mock.patch('kojihub.readTaggedBuilds').start()
self.tag_name = 'test-tag'
self.columns = ['archiveinfo.id', 'archiveinfo.type_id', 'archiveinfo.btype_id',
'btype.name', 'archiveinfo.build_id', 'archiveinfo.buildroot_id',
'archiveinfo.filename', 'archiveinfo.size', 'archiveinfo.checksum',
'archiveinfo.checksum_type', 'archiveinfo.metadata_only']
self.joins = ['tag_listing ON archiveinfo.build_id = tag_listing.build_id',
'btype ON archiveinfo.btype_id = btype.id']
self.aliases = ['id', 'type_id', 'btype_id', 'btype', 'build_id', 'buildroot_id',
'filename', 'size', 'checksum', 'checksum_type', 'metadata_only']
self.clauses = ['(active = TRUE)',
'tag_listing.tag_id = %(tagid)i']
self.tables = ['archiveinfo']
self.pkg_name = 'test_pkg'
self.build_list = [
{'build_id': 1, 'create_event': 1172, 'creation_event_id': 1171, 'epoch': None,
'id': 1, 'name': 'test-pkg', 'nvr': 'test-pkg-2.52-1.fc35', 'owner_id': 1,
'owner_name': 'kojiuser', 'package_id': 1, 'package_name': 'test-pkg',
'release': '1.fc35', 'state': 1, 'tag_id': 1, 'tag_name': 'test-tag',
'task_id': None, 'version': '2.52', 'volume_id': 0, 'volume_name': 'DEFAULT'}]
def tearDown(self):
mock.patch.stopall()
def test_get_tagged_archives_default(self):
self.readTaggedBuilds.return_value = self.build_list
kojihub.readTaggedArchives(self.tag_name)
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
columns = copy.deepcopy(self.columns)
columns.append('archiveinfo.extra')
aliases = copy.deepcopy(self.aliases)
aliases.append('extra')
values = {'package': None, 'tagid': self.tag_name}
self.assertEqual(query.tables, self.tables)
self.assertEqual(query.joins, self.joins)
self.assertEqual(set(query.columns), set(columns))
self.assertEqual(set(query.aliases), set(aliases))
self.assertEqual(set(query.clauses), set(self.clauses))
self.assertEqual(query.values, values)
def test_get_tagged_archives_package_type_maven_without_extra(self):
self.readTaggedBuilds.return_value = self.build_list
kojihub.readTaggedArchives(self.tag_name, package=self.pkg_name, type='maven', extra=False)
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
columns = copy.deepcopy(self.columns)
columns.extend(['maven_archives.group_id', 'maven_archives.artifact_id',
'maven_archives.version'])
aliases = copy.deepcopy(self.aliases)
aliases.extend(['maven_group_id', 'maven_artifact_id', 'maven_version'])
clauses = copy.deepcopy(self.clauses)
clauses.extend(['package.name = %(package)s'])
joins = copy.deepcopy(self.joins)
joins.extend(['build ON archiveinfo.build_id = build.id',
'package ON build.pkg_id = package.id',
'maven_archives ON archiveinfo.id = maven_archives.archive_id'])
values = {'package': self.pkg_name, 'tagid': self.tag_name}
self.assertEqual(query.tables, self.tables)
self.assertEqual(query.joins, joins)
self.assertEqual(set(query.columns), set(columns))
self.assertEqual(set(query.aliases), set(aliases))
self.assertEqual(set(query.clauses), set(clauses))
self.assertEqual(query.values, values)
def test_get_tagged_archives_package_type_win_without_extra(self):
self.readTaggedBuilds.return_value = self.build_list
kojihub.readTaggedArchives(self.tag_name, type='win', extra=False)
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
columns = copy.deepcopy(self.columns)
columns.extend(['win_archives.relpath', 'win_archives.platforms', 'win_archives.flags'])
aliases = copy.deepcopy(self.aliases)
aliases.extend(['relpath', 'platforms', 'flags'])
joins = copy.deepcopy(self.joins)
joins.append('win_archives ON archiveinfo.id = win_archives.archive_id')
values = {'package': None, 'tagid': self.tag_name}
self.assertEqual(query.tables, self.tables)
self.assertEqual(query.joins, joins)
self.assertEqual(set(query.columns), set(columns))
self.assertEqual(set(query.aliases), set(aliases))
self.assertEqual(set(query.clauses), set(self.clauses))
self.assertEqual(query.values, values)
def test_get_tagged_archives_type_non_exist(self):
self.readTaggedBuilds.return_value = self.build_list
error_message = "unsupported archive type: non-exist-type"
with self.assertRaises(koji.GenericError) as cm:
kojihub.readTaggedArchives(self.tag_name, type='non-exist-type')
self.assertEqual(error_message, str(cm.exception))

221
tests/test_hub/test_read_tagged_builds.py Normal file
View file

@ -0,0 +1,221 @@
import unittest
import mock
import koji
import kojihub
import copy
QP = kojihub.QueryProcessor
class TestReadTaggedBuilds(unittest.TestCase):
def getQuery(self, *args, **kwargs):
self.maxDiff = None
query = QP(*args, **kwargs)
query.execute = mock.MagicMock()
query.executeOne = mock.MagicMock()
self.queries.append(query)
return query
def setUp(self):
self.QueryProcessor = mock.patch('kojihub.QueryProcessor',
side_effect=self.getQuery).start()
self.queries = []
self.context = mock.patch('kojihub.context').start()
# It seems MagicMock will not automatically handle attributes that
# start with "assert"
self.exports = kojihub.RootExports()
self.readPackageList = mock.patch('kojihub.readPackageList').start()
self.lookup_name = mock.patch('kojihub.lookup_name').start()
self.tag_name = 'test-tag'
self.columns = ['tag.id', 'tag.name', 'build.id', 'build.version', 'build.release',
'build.epoch', 'build.state', 'build.completion_time', 'build.start_time',
'build.task_id', 'users.id', 'users.name', 'events.id', 'events.time',
'volume.id', 'volume.name', 'package.id', 'package.name',
'package.name || \'-\' || build.version || \'-\' || build.release',
'tag_listing.create_event']
self.fields = [('tag.id', 'tag_id'), ('tag.name', 'tag_name'), ('build.id', 'id'),
('build.id', 'build_id'), ('build.version', 'version'),
('build.release', 'release'), ('build.epoch', 'epoch'),
('build.state', 'state'), ('build.completion_time', 'completion_time'),
('build.start_time', 'start_time'), ('build.task_id', 'task_id'),
('users.id', 'owner_id'), ('users.name', 'owner_name'),
('events.id', 'creation_event_id'), ('events.time', 'creation_time'),
('volume.id', 'volume_id'), ('volume.name', 'volume_name'),
('package.id', 'package_id'), ('package.name', 'package_name'),
('package.name', 'name'),
("package.name || '-' || build.version || '-' || build.release", 'nvr'),
('tag_listing.create_event', 'create_event')]
self.joins = ['tag ON tag.id = tag_listing.tag_id',
'build ON build.id = tag_listing.build_id',
'events ON events.id = build.create_event',
'package ON package.id = build.pkg_id',
'volume ON volume.id = build.volume_id',
'users ON users.id = build.owner', ]
self.aliases = ['tag_id', 'tag_name', 'id', 'build_id', 'version', 'release', 'epoch',
'state', 'completion_time', 'start_time', 'task_id', 'owner_id',
'owner_name', 'creation_event_id', 'creation_time', 'volume_id',
'volume_name', 'package_id', 'package_name', 'name', 'nvr', 'create_event']
self.clauses = ['(tag_listing.active = TRUE)',
'tag_id = %(tagid)s',
'build.state = %(st_complete)i']
self.tables = ['tag_listing']
self.pkg_name = 'test_pkg'
self.username = 'testuser'
self.package_list = {1: {'blocked': False, 'extra_arches': '', 'package_id': 1,
'package_name': self.pkg_name, 'tag_id': 4,
'tag_name': self.tag_name},
'owner_name': self.username}
def tearDown(self):
mock.patch.stopall()
def test_get_tagged_builds_default(self):
self.readPackageList.return_value = self.package_list
kojihub.readTaggedBuilds(self.tag_name)
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
values = {'clauses': self.clauses, 'event': None, 'extra': False, 'fields': self.fields,
'inherit': False, 'joins': self.joins, 'latest': False, 'owner': None,
'package': None, 'packages': self.package_list,
'queryOpts': {'order': '-create_event'}, 'st_complete': 1, 'tables': self.tables,
'tag': self.tag_name, 'tagid': self.tag_name, 'taglist': [self.tag_name],
'type': None
}
self.assertEqual(query.tables, self.tables)
self.assertEqual(query.joins, self.joins)
self.assertEqual(set(query.columns), set(self.columns))
self.assertEqual(set(query.aliases), set(self.aliases))
self.assertEqual(set(query.clauses), set(self.clauses))
self.assertEqual(query.values, values)
def test_get_tagged_builds_package_owner_type_maven_extra(self):
self.readPackageList.return_value = self.package_list
kojihub.readTaggedBuilds(self.tag_name, package=self.pkg_name, owner=self.username,
type='maven', extra=True)
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
columns = copy.deepcopy(self.columns)
columns.extend(['maven_builds.group_id', 'maven_builds.artifact_id',
'maven_builds.version', 'build.extra'])
aliases = copy.deepcopy(self.aliases)
aliases.extend(['maven_group_id', 'maven_artifact_id', 'maven_version', 'extra'])
fields = copy.deepcopy(self.fields)
fields.extend([('maven_builds.group_id', 'maven_group_id'),
('maven_builds.artifact_id', 'maven_artifact_id'),
('maven_builds.version', 'maven_version'), ('build.extra', 'extra')])
clauses = copy.deepcopy(self.clauses)
clauses.extend(['package.name = %(package)s', 'users.name = %(owner)s'])
joins = copy.deepcopy(self.joins)
joins.append('maven_builds ON maven_builds.build_id = tag_listing.build_id')
values = {'clauses': clauses, 'event': None, 'extra': True, 'fields': fields,
'inherit': False, 'joins': joins, 'latest': False, 'owner': self.username,
'package': self.pkg_name, 'packages': self.package_list,
'queryOpts': {'order': '-create_event'}, 'st_complete': 1, 'tables': self.tables,
'tag': self.tag_name, 'tagid': self.tag_name, 'taglist': [self.tag_name],
'type': 'maven'}
self.assertEqual(query.tables, self.tables)
self.assertEqual(query.joins, joins)
self.assertEqual(set(query.columns), set(columns))
self.assertEqual(set(query.aliases), set(aliases))
self.assertEqual(set(query.clauses), set(clauses))
self.assertEqual(query.values, values)
def test_get_tagged_builds_type_win_latest(self):
self.readPackageList.return_value = self.package_list
kojihub.readTaggedBuilds(self.tag_name, type='win', latest=True)
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
columns = copy.deepcopy(self.columns)
columns.append('win_builds.platform')
aliases = copy.deepcopy(self.aliases)
aliases.append('platform')
fields = copy.deepcopy(self.fields)
fields.append(('win_builds.platform', 'platform'))
joins = copy.deepcopy(self.joins)
joins.append('win_builds ON win_builds.build_id = tag_listing.build_id')
values = {'clauses': self.clauses, 'event': None, 'extra': False, 'fields': fields,
'inherit': False, 'joins': joins, 'latest': True, 'owner': None,
'package': None, 'packages': self.package_list,
'queryOpts': {'order': '-create_event'}, 'st_complete': 1, 'tables': self.tables,
'tag': self.tag_name, 'tagid': self.tag_name, 'taglist': [self.tag_name],
'type': 'win'}
self.assertEqual(query.tables, self.tables)
self.assertEqual(query.joins, joins)
self.assertEqual(set(query.columns), set(columns))
self.assertEqual(set(query.aliases), set(aliases))
self.assertEqual(set(query.clauses), set(self.clauses))
self.assertEqual(query.values, values)
def test_get_tagged_builds_type_image(self):
self.readPackageList.return_value = self.package_list
kojihub.readTaggedBuilds(self.tag_name, type='image')
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
columns = copy.deepcopy(self.columns)
columns.append('image_builds.build_id')
aliases = copy.deepcopy(self.aliases)
aliases.append('build_id')
fields = copy.deepcopy(self.fields)
fields.append(('image_builds.build_id', 'build_id'))
joins = copy.deepcopy(self.joins)
joins.append('image_builds ON image_builds.build_id = tag_listing.build_id')
values = {'clauses': self.clauses, 'event': None, 'extra': False, 'fields': fields,
'inherit': False, 'joins': joins, 'latest': False, 'owner': None,
'package': None, 'packages': self.package_list,
'queryOpts': {'order': '-create_event'}, 'st_complete': 1, 'tables': self.tables,
'tag': self.tag_name, 'tagid': self.tag_name, 'taglist': [self.tag_name],
'type': 'image'}
self.assertEqual(query.tables, self.tables)
self.assertEqual(query.joins, joins)
self.assertEqual(set(query.columns), set(columns))
self.assertEqual(set(query.aliases), set(aliases))
self.assertEqual(set(query.clauses), set(self.clauses))
self.assertEqual(query.values, values)
def test_get_tagged_builds_type_non_exist(self):
self.readPackageList.return_value = self.package_list
self.lookup_name.return_value = None
error_message = "unsupported build type: non-exist-type"
with self.assertRaises(koji.GenericError) as cm:
kojihub.readTaggedBuilds(self.tag_name, type='non-exist-type')
self.assertEqual(error_message, str(cm.exception))
def test_get_tagged_builds_type_exists(self):
self.readPackageList.return_value = self.package_list
type = 'exist-type'
typeinfo = {'id': 11, 'name': type}
self.lookup_name.return_value = typeinfo
kojihub.readTaggedBuilds(self.tag_name, type=type)
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
joins = copy.deepcopy(self.joins)
joins.append('build_types ON build.id = build_types.build_id AND btype_id = %(btype_id)s')
values = {'btype': typeinfo, 'btype_id': typeinfo['id'], 'clauses': self.clauses,
'event': None, 'extra': False, 'fields': self.fields,
'inherit': False, 'joins': joins, 'latest': False, 'owner': None,
'package': None, 'packages': self.package_list,
'queryOpts': {'order': '-create_event'}, 'st_complete': 1, 'tables': self.tables,
'tag': self.tag_name, 'tagid': self.tag_name, 'taglist': [self.tag_name],
'type': type}
self.assertEqual(query.tables, self.tables)
self.assertEqual(query.joins, joins)
self.assertEqual(set(query.columns), set(self.columns))
self.assertEqual(set(query.aliases), set(self.aliases))
self.assertEqual(set(query.clauses), set(self.clauses))
self.assertEqual(query.values, values)

71
tests/test_hub/test_read_tagged_rpms.py
View file

@ -4,17 +4,41 @@ import mock
import koji
import kojihub
import copy
QP = kojihub.QueryProcessor
class TestReadTaggedRPMS(unittest.TestCase):
def getQuery(self, *args, **kwargs):
self.maxDiff = None
query = QP(*args, **kwargs)
query.execute = mock.MagicMock()
query.executeOne = mock.MagicMock()
self.queries.append(query)
return query
def setUp(self):
self.QueryProcessor = mock.patch('kojihub.QueryProcessor',
side_effect=self.getQuery).start()
self.queries = []
self.context = mock.patch('kojihub.context').start()
# It seems MagicMock will not automatically handle attributes that
# start with "assert"
self.exports = kojihub.RootExports()
self.readTaggedBuilds = mock.patch('kojihub.readTaggedBuilds').start()
self.tag_name = 'test-tag'
self.columns = ['rpminfo.name', 'rpminfo.version', 'rpminfo.release', 'rpminfo.arch',
'rpminfo.id', 'rpminfo.epoch', 'rpminfo.payloadhash', 'rpminfo.size',
'rpminfo.buildtime', 'rpminfo.buildroot_id', 'rpminfo.build_id',
'rpminfo.metadata_only']
self.joins = ['tag_listing ON rpminfo.build_id = tag_listing.build_id']
self.aliases = ['name', 'version', 'release', 'arch', 'id', 'epoch', 'payloadhash',
'size', 'buildtime', 'buildroot_id', 'build_id', 'metadata_only']
self.clauses = ['(tag_listing.active = TRUE)',
'tag_id=%(tagid)s']
self.tables = ['rpminfo']
self.pkg_name = 'test_pkg'
self.build_list = [
{'build_id': 1, 'create_event': 1172, 'creation_event_id': 1171, 'epoch': None,
'id': 1, 'name': 'test-pkg', 'nvr': 'test-pkg-2.52-1.fc35', 'owner_id': 1,
@ -31,3 +55,50 @@ class TestReadTaggedRPMS(unittest.TestCase):
with self.assertRaises(koji.GenericError) as cm:
kojihub.readTaggedRPMS(self.tag_name, arch=1245)
self.assertEqual(error_message, str(cm.exception))
def test_get_tagged_rpms_package_arch_list_without_extra(self):
self.readTaggedBuilds.return_value = self.build_list
kojihub.readTaggedRPMS(self.tag_name, package=self.pkg_name, arch=['x86_64', 'ppc'],
extra=False)
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
joins = copy.deepcopy(self.joins)
joins.extend(['build ON rpminfo.build_id = build.id',
'package ON package.id = build.pkg_id'])
clauses = copy.deepcopy(self.clauses)
clauses.extend(['package.name = %(package)s', 'rpminfo.arch IN %(arch)s'])
values = {'arch': ['x86_64', 'ppc'], 'package': 'test_pkg'}
self.assertEqual(query.tables, self.tables)
self.assertEqual(set(query.columns), set(self.columns))
self.assertEqual(query.joins, joins)
self.assertEqual(set(query.aliases), set(self.aliases))
self.assertEqual(set(query.clauses), set(clauses))
self.assertEqual(query.values, values)
def test_get_tagged_rpms_rpmsigs_arch_extra(self):
self.readTaggedBuilds.return_value = self.build_list
kojihub.readTaggedRPMS(self.tag_name, rpmsigs='FD431D51', arch='x86_64')
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
joins = copy.deepcopy(self.joins)
joins.extend(['LEFT OUTER JOIN rpmsigs on rpminfo.id = rpmsigs.rpm_id'])
clauses = copy.deepcopy(self.clauses)
clauses.append('rpminfo.arch = %(arch)s')
columns = copy.deepcopy(self.columns)
columns.extend(['rpmsigs.sigkey', 'rpminfo.extra'])
aliases = copy.deepcopy(self.aliases)
aliases.extend(['sigkey', 'extra'])
values = {'arch': 'x86_64'}
self.assertEqual(query.tables, self.tables)
self.assertEqual(set(query.columns), set(columns))
self.assertEqual(query.joins, joins)
self.assertEqual(set(query.aliases), set(aliases))
self.assertEqual(set(query.clauses), set(clauses))
self.assertEqual(query.values, values)