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

PR#3282: Add extra of builds to listTagged call result

Browse source 
Merges #3282
https://pagure.io/koji/pull-request/3282

Fixes: #3110
https://pagure.io/koji/issue/3110
RFE: expose build.extra by listTagged API directly
This commit is contained in:
Tomas Kopecek 2022-05-09 14:04:36 +02:00
commit 87aa8eb17b
8 changed files with 733 additions and 151 deletions
Download patch file Download diff file Expand all files Collapse all files

81
hub/kojihub.py
View file

@ -1354,7 +1354,7 @@ def list_tags(build=None, package=None, perms=True, queryOpts=None, pattern=None
def readTaggedBuilds(tag, event=None, inherit=False, latest=False, package=None, owner=None,
type=None):
type=None, extra=False):
"""Returns a list of builds for specified tag
:param int tag: tag ID
@ -1365,6 +1365,7 @@ def readTaggedBuilds(tag, event=None, inherit=False, latest=False, package=None,
:param str owner: filter on user name
:param str type: restrict the list to builds of the given type. Currently the supported
types are 'maven', 'win', 'image', or any custom content generator btypes.
:param bool extra: Set to "True" to get the build extra info
:returns [dict]: list of buildinfo dicts
"""
# build - id pkg_id version release epoch
@ -1437,9 +1438,15 @@ def readTaggedBuilds(tag, event=None, inherit=False, latest=False, package=None,
if owner:
clauses.append('users.name = %(owner)s')
queryOpts = {'order': '-create_event'} # latest first
query = QueryProcessor(columns=[x[0] for x in fields], aliases=[x[1] for x in fields],
tables=tables, joins=joins, clauses=clauses, values=locals(),
opts=queryOpts)
if extra:
fields.append(('build.extra', 'extra'))
query = QueryProcessor(columns=[x[0] for x in fields], aliases=[x[1] for x in fields],
tables=tables, joins=joins, clauses=clauses, values=locals(),
transform=_fix_extra_field, opts=queryOpts)
else:
query = QueryProcessor(columns=[x[0] for x in fields], aliases=[x[1] for x in fields],
tables=tables, joins=joins, clauses=clauses, values=locals(),
opts=queryOpts)
builds = []
seen = {} # used to enforce the 'latest' option
@ -1467,7 +1474,7 @@ def readTaggedBuilds(tag, event=None, inherit=False, latest=False, package=None,
def readTaggedRPMS(tag, package=None, arch=None, event=None, inherit=False, latest=True,
rpmsigs=False, owner=None, type=None):
rpmsigs=False, owner=None, type=None, extra=True):
"""Returns a list of rpms and builds for specified tag
:param int|str tag: The tag name or ID to search
@ -1486,6 +1493,7 @@ def readTaggedRPMS(tag, package=None, arch=None, event=None, inherit=False, late
:param str owner: Filter by build owner name
:param str type: Filter by build type. Supported types are 'maven',
'win', and 'image'.
:param bool extra: Set to "False" to skip the rpm extra info
:returns: a two-element list. The first element is the list of RPMs, and
the second element is the list of builds.
"""
@ -1513,7 +1521,6 @@ def readTaggedRPMS(tag, package=None, arch=None, event=None, inherit=False, late
('rpminfo.buildroot_id', 'buildroot_id'),
('rpminfo.build_id', 'build_id'),
('rpminfo.metadata_only', 'metadata_only'),
('rpminfo.extra', 'extra'),
]
tables = ['rpminfo']
joins = ['tag_listing ON rpminfo.build_id = tag_listing.build_id']
@ -1536,9 +1543,17 @@ def readTaggedRPMS(tag, package=None, arch=None, event=None, inherit=False, late
else:
raise koji.GenericError('Invalid type for arch option: %s' % builtins.type(arch))
fields, aliases = zip(*fields)
query = QueryProcessor(tables=tables, joins=joins, clauses=clauses,
columns=fields, aliases=aliases, values=data, transform=_fix_rpm_row)
if extra:
fields.append(('rpminfo.extra', 'extra'))
query = QueryProcessor(tables=tables, joins=joins, clauses=clauses,
columns=[pair[0] for pair in fields],
aliases=[pair[1] for pair in fields],
values=data, transform=_fix_rpm_row)
else:
query = QueryProcessor(tables=tables, joins=joins, clauses=clauses,
columns=[pair[0] for pair in fields],
aliases=[pair[1] for pair in fields],
values=data)
# unique constraints ensure that each of these queries will not report
# duplicate rpminfo entries, BUT since we make the query multiple times,
@ -1573,13 +1588,20 @@ def readTaggedRPMS(tag, package=None, arch=None, event=None, inherit=False, late
return [_iter_rpms(), builds]
def readTaggedArchives(tag, package=None, event=None, inherit=False, latest=True, type=None):
def readTaggedArchives(tag, package=None, event=None, inherit=False, latest=True, type=None,
extra=True):
"""Returns a list of archives for specified tag
set inherit=True to follow inheritance
set event to query at a time in the past
set latest=False to get all tagged archives (not just from the latest builds)
set latest=N to get only the N latest tagged RPMs
:param int|str tag: The tag name or ID to search
:param str package: Filter on a package name.
:param int event: set event to query at a time in the past
:param bool inherit: set to "True" to follow inheritance
:param bool|int latest: set to "False" to get all tagged archives (not just from the latest
builds), set to "N" to get only the "N" latest tagged RPMs
:param str type: Filter by build type. Supported types are 'maven' and 'win'.
:param bool extra: Set to "False" to skip the archives extra info
:returns: a two-element list. The first element is the list of archives, and
the second element is the list of builds.
If type is not None, restrict the listing to archives of the given type. Currently
the supported types are 'maven' and 'win'.
@ -1608,7 +1630,6 @@ def readTaggedArchives(tag, package=None, event=None, inherit=False, latest=True
('archiveinfo.checksum', 'checksum'),
('archiveinfo.checksum_type', 'checksum_type'),
('archiveinfo.metadata_only', 'metadata_only'),
('archiveinfo.extra', 'extra'),
]
tables = ['archiveinfo']
joins = ['tag_listing ON archiveinfo.build_id = tag_listing.build_id',
@ -1633,10 +1654,16 @@ def readTaggedArchives(tag, package=None, event=None, inherit=False, latest=True
else:
raise koji.GenericError('unsupported archive type: %s' % type)
query = QueryProcessor(tables=tables, joins=joins, clauses=clauses,
transform=_fix_archive_row,
columns=[pair[0] for pair in fields],
aliases=[pair[1] for pair in fields])
if extra:
fields.append(('archiveinfo.extra', 'extra'))
query = QueryProcessor(tables=tables, joins=joins, clauses=clauses,
transform=_fix_archive_row,
columns=[pair[0] for pair in fields],
aliases=[pair[1] for pair in fields])
else:
query = QueryProcessor(tables=tables, joins=joins, clauses=clauses,
columns=[pair[0] for pair in fields],
aliases=[pair[1] for pair in fields])
# unique constraints ensure that each of these queries will not report
# duplicate archiveinfo entries, BUT since we make the query multiple times,
@ -11770,7 +11797,7 @@ class RootExports(object):
task.setPriority(priority, recurse=recurse)
def listTagged(self, tag, event=None, inherit=False, prefix=None, latest=False, package=None,
owner=None, type=None, strict=True):
owner=None, type=None, strict=True, extra=False):
"""List builds tagged with tag.
:param int|str tag: tag name or ID number
@ -11785,13 +11812,14 @@ class RootExports(object):
:param str type: only builds of the given btype (such as maven or image)
:param bool strict: If tag doesn't exist, an exception is raised,
unless strict is False in which case returns an empty list.
:param bool extra: Set to "True" to get the build extra info
"""
# lookup tag id
tag = get_tag(tag, strict=strict, event=event)
if not tag:
return []
results = readTaggedBuilds(tag['id'], event, inherit=inherit, latest=latest,
package=package, owner=owner, type=type)
package=package, owner=owner, type=type, extra=extra)
if prefix:
prefix = prefix.lower()
results = [build for build in results
@ -11799,7 +11827,7 @@ class RootExports(object):
return results
def listTaggedRPMS(self, tag, event=None, inherit=False, latest=False, package=None, arch=None,
rpmsigs=False, owner=None, type=None, strict=True):
rpmsigs=False, owner=None, type=None, strict=True, extra=True):
"""List rpms and builds within tag.
:param int|str tag: tag name or ID number
@ -11817,16 +11845,18 @@ class RootExports(object):
:param str type: only rpms of the given btype (such as maven or image)
:param bool strict: If tag doesn't exist, an exception is raised,
unless strict is False in which case returns an empty list.
:param bool extra: Set to "False" to skip the rpms extra info
"""
# lookup tag id
tag = get_tag(tag, strict=strict, event=event)
if not tag:
return []
return readTaggedRPMS(tag['id'], event=event, inherit=inherit, latest=latest,
package=package, arch=arch, rpmsigs=rpmsigs, owner=owner, type=type)
package=package, arch=arch, rpmsigs=rpmsigs, owner=owner,
type=type, extra=extra)
def listTaggedArchives(self, tag, event=None, inherit=False, latest=False, package=None,
type=None, strict=True):
type=None, strict=True, extra=True):
"""List archives and builds within a tag.
:param int|str tag: tag name or ID number
@ -11839,13 +11869,14 @@ class RootExports(object):
:param str type: only archives of the given btype (such as maven or image)
:param bool strict: If tag doesn't exist, an exception is raised,
unless strict is False in which case returns an empty list.
:param bool extra: Set to "False" to skip the archive extra info
"""
# lookup tag id
tag = get_tag(tag, strict=strict, event=event)
if not tag:
return []
return readTaggedArchives(tag['id'], event=event, inherit=inherit, latest=latest,
package=package, type=type)
package=package, type=type, extra=extra)
def listBuilds(self, packageID=None, userID=None, taskID=None, prefix=None, state=None,
volumeID=None, source=None, createdBefore=None, createdAfter=None,

238
tests/test_cli/test_list_tagged.py
View file

@ -9,10 +9,9 @@ from . import utils
class TestCliListTagged(utils.CliTestCase):
# Show long diffs in error output...
maxDiff = None
def setUp(self):
self.maxDiff = None
self.original_timezone = os.environ.get('TZ')
os.environ['TZ'] = 'US/Eastern'
time.tzset()
@ -23,6 +22,10 @@ class TestCliListTagged(utils.CliTestCase):
""" % (self.progname, self.progname)
self.session = mock.MagicMock()
self.options = mock.MagicMock(quiet=False)
self.tag = 'tag'
self.pkg = 'pkg'
self.event_id = 1000
self.type = 'maven'
self.session.getTag.return_value = {'id': 1}
self.session.listTaggedRPMS.return_value = [[{'id': 100,
'build_id': 1,
@ -30,28 +33,33 @@ class TestCliListTagged(utils.CliTestCase):
'version': '0.0.1',
'release': '1.el6',
'arch': 'noarch',
'sigkey': 'sigkey'},
'sigkey': 'sigkey',
'extra': 'extra-value'},
{'id': 101,
'build_id': 1,
'name': 'rpmA',
'version': '0.0.1',
'release': '1.el6',
'arch': 'x86_64',
'sigkey': 'sigkey'}
'sigkey': 'sigkey',
'extra': None}
], [{'id': 1,
'name': 'packagename',
'version': 'version',
'release': '1.el6',
'nvr': 'n-v-r',
'tag_name': 'tag',
'owner_name': 'owner'}]]
'owner_name': 'owner',
'extra': 'extra-value-2'}]]
self.session.listTagged.return_value = [{'id': 1,
'name': 'packagename',
'version': 'version',
'release': '1.el6',
'nvr': 'n-v-r',
'tag_name': 'tag',
'owner_name': 'owner'}]
'owner_name': 'owner',
'extra': 'extra-value-2'}]
self.ensure_connection_mock = mock.patch('koji_cli.commands.ensure_connection').start()
def tearDown(self):
if self.original_timezone is None:
@ -63,92 +71,96 @@ class TestCliListTagged(utils.CliTestCase):
@mock.patch('sys.stdout', new_callable=six.StringIO)
@mock.patch('koji.util.eventFromOpts', return_value={'id': 1000,
'ts': 1000000.11})
@mock.patch('koji_cli.commands.ensure_connection')
def test_list_tagged_builds(self, ensure_connection_mock,
event_from_opts_mock, stdout):
args = ['tag', 'pkg', '--latest', '--inherit', '--event=1000']
def test_list_tagged_builds(self, event_from_opts_mock, stdout):
expected = """Querying at event 1000 (Mon Jan 12 08:46:40 1970)
Build Tag Built by
---------------------------------------- -------------------- ----------------
n-v-r tag owner
"""
args = [self.tag, self.pkg, '--latest', '--inherit', '--event', str(self.event_id)]
anon_handle_list_tagged(self.options, self.session, args)
ensure_connection_mock.assert_called_once_with(self.session, self.options)
self.session.getTag.assert_called_once_with('tag', event=1000)
self.session.listTagged.assert_called_once_with('tag',
event=1000,
inherit=True,
latest=True,
package='pkg')
self.ensure_connection_mock.assert_called_once_with(self.session, self.options)
self.session.getTag.assert_called_once_with(self.tag, event=self.event_id)
self.session.listTagged.assert_called_once_with(
self.tag, event=self.event_id, inherit=True, latest=True, package=self.pkg)
self.session.listTaggedRPMS.assert_not_called()
self.assert_console_message(stdout,
'Querying at event 1000 (Mon Jan 12 08:46:40 1970)\n'
'Build Tag Built by\n'
'---------------------------------------- -------------------- ----------------\n'
'n-v-r tag owner\n')
self.assert_console_message(stdout, expected)
@mock.patch('sys.stdout', new_callable=six.StringIO)
@mock.patch('koji.util.eventFromOpts', return_value=None)
@mock.patch('koji_cli.commands.ensure_connection')
def test_list_tagged_builds_paths(self, ensure_connection_mock,
event_from_opts_mock, stdout):
args = ['tag', 'pkg', '--latest', '--inherit', '--paths']
def test_list_tagged_builds_paths(self, event_from_opts_mock, stdout):
expected = """Build Tag Built by
---------------------------------------- -------------------- ----------------
/mnt/koji/packages/packagename/version/1.el6 tag owner
"""
args = [self.tag, self.pkg, '--latest', '--inherit', '--paths']
anon_handle_list_tagged(self.options, self.session, args)
self.assert_console_message(stdout,
'Build Tag Built by\n'
'---------------------------------------- -------------------- ----------------\n'
'/mnt/koji/packages/packagename/version/1.el6 tag owner\n')
self.assert_console_message(stdout, expected)
self.ensure_connection_mock.assert_called_once_with(self.session, self.options)
self.session.getTag.assert_called_once_with(self.tag, event=None)
self.session.listTagged.assert_called_once_with(
self.tag, inherit=True, latest=True, package=self.pkg)
self.session.listTaggedRPMS.assert_not_called()
@mock.patch('sys.stdout', new_callable=six.StringIO)
@mock.patch('koji.util.eventFromOpts', return_value=None)
@mock.patch('koji_cli.commands.ensure_connection')
def test_list_tagged_rpms(self, ensure_connection_mock,
event_from_opts_mock, stdout):
args = ['tag', 'pkg', '--latest-n=3', '--rpms', '--sigs',
def test_list_tagged_rpms(self, event_from_opts_mock, stdout):
expected = """sigkey rpmA-0.0.1-1.el6.noarch
sigkey rpmA-0.0.1-1.el6.x86_64
"""
args = [self.tag, self.pkg, '--latest-n=3', '--rpms', '--sigs',
'--arch=x86_64', '--arch=noarch']
anon_handle_list_tagged(self.options, self.session, args)
ensure_connection_mock.assert_called_once_with(self.session, self.options)
self.session.getTag.assert_called_once_with('tag', event=None)
self.session.listTaggedRPMS.assert_called_once_with('tag',
package='pkg',
inherit=None,
latest=3,
rpmsigs=True,
arch=['x86_64',
'noarch'])
self.ensure_connection_mock.assert_called_once_with(self.session, self.options)
self.session.getTag.assert_called_once_with(self.tag, event=None)
self.session.listTaggedRPMS.assert_called_once_with(
self.tag, package=self.pkg, inherit=None, latest=3, rpmsigs=True,
arch=['x86_64', 'noarch'])
self.session.listTagged.assert_not_called()
self.assert_console_message(stdout,
'sigkey rpmA-0.0.1-1.el6.noarch\n'
'sigkey rpmA-0.0.1-1.el6.x86_64\n')
self.assert_console_message(stdout, expected)
@mock.patch('sys.stdout', new_callable=six.StringIO)
@mock.patch('koji.util.eventFromOpts', return_value=None)
@mock.patch('koji_cli.commands.ensure_connection')
def test_list_tagged_rpms_paths(self, ensure_connection_mock,
event_from_opts_mock, stdout):
args = ['tag', 'pkg', '--latest-n=3', '--rpms',
def test_list_tagged_rpms_paths(self, event_from_opts_mock, stdout):
expected = """/mnt/koji/packages/packagename/version/1.el6/noarch/rpmA-0.0.1-1.el6.noarch.rpm
/mnt/koji/packages/packagename/version/1.el6/x86_64/rpmA-0.0.1-1.el6.x86_64.rpm
"""
args = [self.tag, self.pkg, '--latest-n=3', '--rpms', '--arch=x86_64', '--paths']
anon_handle_list_tagged(self.options, self.session, args)
self.assert_console_message(stdout, expected)
self.ensure_connection_mock.assert_called_once_with(self.session, self.options)
self.session.getTag.assert_called_once_with(self.tag, event=None)
self.session.listTaggedRPMS.assert_called_once_with(
self.tag, package=self.pkg, inherit=None, latest=3, arch=['x86_64'])
self.session.listTagged.assert_not_called()
@mock.patch('sys.stdout', new_callable=six.StringIO)
@mock.patch('koji.util.eventFromOpts', return_value=None)
def test_list_tagged_sigs_paths(self, event_from_opts_mock, stdout):
expected = ""
args = [self.tag, self.pkg, '--latest-n=3', '--rpms', '--sigs',
'--arch=x86_64', '--paths']
anon_handle_list_tagged(self.options, self.session, args)
self.assert_console_message(stdout,
'/mnt/koji/packages/packagename/version/1.el6/noarch/rpmA-0.0.1-1.el6.noarch.rpm\n'
'/mnt/koji/packages/packagename/version/1.el6/x86_64/rpmA-0.0.1-1.el6.x86_64.rpm\n')
self.assert_console_message(stdout, expected)
self.ensure_connection_mock.assert_called_once_with(self.session, self.options)
self.session.getTag.assert_called_once_with(self.tag, event=None)
self.session.listTaggedRPMS.assert_called_once_with(
self.tag, package=self.pkg, inherit=None, latest=3, rpmsigs=True, arch=['x86_64'])
self.session.listTagged.assert_not_called()
@mock.patch('sys.stdout', new_callable=six.StringIO)
@mock.patch('koji.util.eventFromOpts', return_value=None)
@mock.patch('koji_cli.commands.ensure_connection')
def test_list_tagged_sigs_paths(self, ensure_connection_mock,
event_from_opts_mock, stdout):
args = ['tag', 'pkg', '--latest-n=3', '--rpms', '--sigs',
'--arch=x86_64', '--paths']
anon_handle_list_tagged(self.options, self.session, args)
self.assert_console_message(stdout, '')
@mock.patch('sys.stdout', new_callable=six.StringIO)
@mock.patch('koji.util.eventFromOpts', return_value=None)
@mock.patch('koji_cli.commands.ensure_connection')
def test_list_tagged_type(self, ensure_connection_mock,
event_from_opts_mock, stdout):
args = ['tag', 'pkg', '--latest-n=3', '--type=maven']
def test_list_tagged_type(self, event_from_opts_mock, stdout):
expected = """Build Tag Group Id Artifact Id Built by
---------------------------------------- -------------------- -------------------- -------------------- ----------------
n-v-r tag group artifact owner
"""
args = [self.tag, self.pkg, '--latest-n=3', '--type', self.type]
self.session.listTagged.return_value = [{'id': 1,
'name': 'packagename',
'version': 'version',
@ -160,24 +172,21 @@ class TestCliListTagged(utils.CliTestCase):
'maven_artifact_id': 'artifact'}]
anon_handle_list_tagged(self.options, self.session, args)
ensure_connection_mock.assert_called_once_with(self.session, self.options)
self.session.getTag.assert_called_once_with('tag', event=None)
self.session.listTagged.assert_called_once_with('tag',
package='pkg',
inherit=None,
latest=3,
type='maven')
self.ensure_connection_mock.assert_called_once_with(self.session, self.options)
self.session.getTag.assert_called_once_with(self.tag, event=None)
self.session.listTagged.assert_called_once_with(
self.tag, package=self.pkg, inherit=None, latest=3, type=self.type)
self.session.listTaggedRPMS.assert_not_called()
self.assert_console_message(stdout,
'Build Tag Group Id Artifact Id Built by\n'
'---------------------------------------- -------------------- -------------------- -------------------- ----------------\n'
'n-v-r tag group artifact owner\n')
self.assert_console_message(stdout, expected)
@mock.patch('sys.stdout', new_callable=six.StringIO)
@mock.patch('koji.util.eventFromOpts', return_value=None)
@mock.patch('koji_cli.commands.ensure_connection')
def test_list_tagged_type_paths(self, ensure_connection_mock, event_from_opts_mock, stdout):
args = ['tag', 'pkg', '--latest-n=3', '--type=maven', '--paths']
def test_list_tagged_type_paths(self, event_from_opts_mock, stdout):
expected = """Build Tag Group Id Artifact Id Built by
---------------------------------------- -------------------- -------------------- -------------------- ----------------
/mnt/koji/packages/packagename/version/1.el6/maven tag group artifact owner
"""
args = [self.tag, self.pkg, '--latest-n=3', '--type', self.type, '--paths']
self.session.listTagged.return_value = [{'id': 1,
'name': 'packagename',
'version': 'version',
@ -189,48 +198,49 @@ class TestCliListTagged(utils.CliTestCase):
'maven_artifact_id': 'artifact'}]
anon_handle_list_tagged(self.options, self.session, args)
self.assert_console_message(stdout,
'Build Tag Group Id Artifact Id Built by\n'
'---------------------------------------- -------------------- -------------------- -------------------- ----------------\n'
'/mnt/koji/packages/packagename/version/1.el6/maven tag group artifact owner\n')
self.assert_console_message(stdout, expected)
self.ensure_connection_mock.assert_called_once_with(self.session, self.options)
self.session.getTag.assert_called_once_with(self.tag, event=None)
self.session.listTaggedRPMS.assert_not_called()
self.session.listTagged.assert_called_once_with(
self.tag, inherit=None, latest=3, package=self.pkg, type=self.type)
@mock.patch('koji_cli.commands.ensure_connection')
@mock.patch('koji.util.eventFromOpts', return_value={'id': 1000,
'ts': 1000000.11})
def test_list_tagged_args(self, event_from_opts_mock, ensure_connection_mock):
# Case 1, no argument
expected = self.format_error_message(
"A tag name must be specified")
def test_list_tagged_without_args(self):
self.assert_system_exit(
anon_handle_list_tagged,
self.options,
self.session,
[],
stderr=expected,
activate_session=None)
self.options, self.session, [],
stderr=self.format_error_message("A tag name must be specified"),
activate_session=None,
exit_code=2)
self.ensure_connection_mock.assert_not_called()
self.session.getTag.assert_not_called()
self.session.listTaggedRPMS.assert_not_called()
self.session.listTagged.assert_not_called()
# Case 2, arguments > 2
expected = self.format_error_message(
"Only one package name may be specified")
def test_list_tagged_more_args(self):
self.assert_system_exit(
anon_handle_list_tagged,
self.options,
self.session,
['tag', 'pkg1', 'pkg2'],
stderr=expected,
activate_session=None)
self.options, self.session, ['tag', 'pkg1', 'pkg2'],
stderr=self.format_error_message("Only one package name may be specified"),
activate_session=None,
exit_code=2)
self.ensure_connection_mock.assert_not_called()
self.session.getTag.assert_not_called()
self.session.listTaggedRPMS.assert_not_called()
self.session.listTagged.assert_not_called()
# Case 3, no tag found
expected = self.format_error_message(
"No such tag: tag")
def test_list_tagged_tag_not_found(self):
self.session.getTag.return_value = None
self.assert_system_exit(
anon_handle_list_tagged,
self.options,
self.session,
['tag', 'pkg1'],
stderr=expected,
activate_session=None)
self.options, self.session, ['tag', 'pkg1'],
stderr=self.format_error_message("No such tag: tag"),
activate_session=None,
exit_code=2)
self.ensure_connection_mock.assert_called_once_with(self.session, self.options)
self.session.getTag.assert_called_once_with(self.tag, event=None)
self.session.listTaggedRPMS.assert_not_called()
self.session.listTagged.assert_not_called()
def test_handle_list_tagged_help(self):
self.assert_help(

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)