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

refactor get_rpm

Browse source
This commit is contained in:
Mike McLean 2023-12-13 18:22:45 -05:00 • committed by Yu Ming Zhu
parent 62e010d44e
commit af36318ead
4 changed files with 78 additions and 124 deletions
Download patch file Download diff file Expand all files Collapse all files

14
cli/koji_cli/commands.py
View file

@ -3551,28 +3551,18 @@ def anon_handle_list_builds(goptions, session, args):
def anon_handle_rpminfo(goptions, session, args):
"[info] Print basic information about an RPM"
usage = "usage: %prog rpminfo [options] <n-v-r.a> [<n-v-r.a> ...]"
usage = "usage: %prog rpminfo [options] <n-v-r.a|id> [<n-v-r.a|id> ...]"
parser = OptionParser(usage=get_usage_str(usage))
parser.add_option("--buildroots", action="store_true",
help="show buildroots the rpm was used in")
parser.add_option("--build", metavar="NVR|ID",
help="show the rpm(s) in the build")
(options, args) = parser.parse_args(args)
if len(args) < 1:
parser.error("Please specify an RPM")
opts = {}
build = options.build
if options.build:
try:
build = int(build)
except ValueError:
pass
opts['build'] = build
ensure_connection(session, goptions)
error_hit = False
for rpm in args:
info = session.getRPM(rpm, **opts)
info = session.getRPM(rpm)
if info is None:
warn("No such rpm: %s\n" % rpm)
error_hit = True

117
kojihub/kojihub.py
View file

@ -4609,11 +4609,12 @@ def _fix_rpm_row(row):
_fix_archive_row = _fix_rpm_row
def get_rpm(rpminfo, strict=False, multi=False, build=None):
def get_rpm(rpminfo, strict=False, multi=False):
"""Get information about the specified RPM
rpminfo ma4666y be any one of the following:
- a int ID
rpminfo may be any one of the following:
- the rpm id as an int
- the rpm id as a string
- a string N-V-R.A
- a string N-V-R.A@location
- a map containing 'name', 'version', 'release', and 'arch'
@ -4621,10 +4622,6 @@ def get_rpm(rpminfo, strict=False, multi=False, build=None):
If specified, location should match the name of an external repo
If build is specfied, the rpm is limited to the build's
build and location(not INTERNAL) is conflict because a rpm in
A map will be returned, with the following keys:
- id
- name
@ -4646,10 +4643,70 @@ def get_rpm(rpminfo, strict=False, multi=False, build=None):
If there is no RPM with the given ID, None is returned, unless strict
is True in which case an exception is raised
If more than one RPM matches, and multi is True, then a list of results is
returned. If multi is False, a single match is returned (an internal one if
possible).
This function is normally expected to return a single rpm. However, there
are cases where the given rpminfo could refer to multiple rpms. This is
because of nvra overlap involving:
* draft rpms
* external rpms
If more than one RPM matches, then in the default case (multi=False), this function
will choose the best option in order of preference:
1. internal non-draft rpms (nvras are unique within this subset)
2. internal draft rpms (highest rpm id)
3. external rpms (highest rpm id)
OTOH if multi is True, then all matching results are returned as a list
"""
# we can look up by id or NVRA
data = None
if isinstance(rpminfo, int):
data = {'id': rpminfo}
elif isinstance(rpminfo, str):
# either nvra or id as a string
try:
data = {'id': int(rpminfo)}
except ValueError:
data = koji.parse_NVRA(rpminfo)
elif isinstance(rpminfo, dict):
data = rpminfo.copy()
else:
raise koji.GenericError("Invalid type for rpminfo: %r" % type(rpminfo))
rpms = _get_rpms(data)
if multi:
return rpms
# otherwise make sure we have a single rpm
if not rpms:
if strict:
raise koji.GenericError("No such rpm: %r" % data)
return None
elif len(rpms) == 1:
return rpms[0]
else:
# pick our preferred, as described above
nondraft = None
draft = None
external = None
for rinfo in rpms:
if rinfo['external_repo_id']:
if external is None or rinfo['id'] > external['id']:
external = rinfo
elif rinfo['draft']:
if draft is None or rinfo['id'] > draft['id']:
draft = rinfo
else:
# rinfo is internal and nondraft
if nondraft:
# should not happen
# none of our selection options should result in more than one nondraft build
raise koji.GenericError("Multiple nondraft rpm matches for: %r" % data)
else:
nondraft = rinfo
return nondraft or draft or external
def _get_rpms(data):
"""Helper function for get_rpm"""
fields = (
('rpminfo.id', 'id'),
('build_id', 'build_id'),
@ -4668,57 +4725,21 @@ def get_rpm(rpminfo, strict=False, multi=False, build=None):
('metadata_only', 'metadata_only'),
('extra', 'extra'),
)
# we can look up by id or NVRA
data = None
if isinstance(rpminfo, int):
data = {'id': rpminfo}
elif isinstance(rpminfo, str):
data = koji.parse_NVRA(rpminfo)
elif isinstance(rpminfo, dict):
data = rpminfo.copy()
else:
raise koji.GenericError("Invalid type for rpminfo: %r" % type(rpminfo))
clauses = []
if 'id' in data:
clauses.append("rpminfo.id=%(id)s")
else:
clauses.append("rpminfo.name=%(name)s AND version=%(version)s "
"AND release=%(release)s AND arch=%(arch)s")
# build and non-INTERNAL location (and multi=True) conflict in theory,
# but we just do the query and return None.
if build:
# strict=True as we treate build not found as an input error
data['build_id'] = find_build_id(build, strict=True)
clauses.append("rpminfo.build_id = %(build_id)s")
retry = False
if 'location' in data:
data['external_repo_id'] = get_external_repo_id(data['location'], strict=True)
clauses.append("""external_repo_id = %(external_repo_id)i""")
elif not multi:
# try to match internal first, otherwise first matching external
retry = True # if no internal match
orig_clauses = list(clauses) # copy
clauses.append("""external_repo_id = 0""")
clauses.append("""external_repo_id = %(external_repo_id)s""")
joins = ['external_repo ON rpminfo.external_repo_id = external_repo.id']
query = QueryProcessor(columns=[f[0] for f in fields], aliases=[f[1] for f in fields],
tables=['rpminfo'], joins=joins, clauses=clauses,
values=data, transform=_fix_rpm_row)
if multi:
return query.execute()
ret = query.executeOne()
if ret:
return ret
if retry:
# at this point we have just an NVRA with no internal match. Open it up to externals
query.clauses = orig_clauses
ret = query.executeOne()
if not ret:
if strict:
raise koji.GenericError("No such rpm: %r" % data)
return None
return ret
return query.execute()
def list_rpms(buildID=None, buildrootID=None, imageID=None, componentBuildrootID=None, hostID=None,

45
tests/test_cli/test_rpminfo.py
View file

@ -64,7 +64,7 @@ class TestRpminfo(utils.CliTestCase):
'version': '1.1',
'payloadhash': 'b2b95550390e5f213fc25f33822425f7',
'size': 7030}]
self.error_format = """Usage: %s rpminfo [options] <n-v-r.a> [<n-v-r.a> ...]
self.error_format = """Usage: %s rpminfo [options] <n-v-r.a|id> [<n-v-r.a|id> ...]
(Specify the --help global option for a list of other help options)
%s: error: {message}
@ -171,42 +171,6 @@ Used in 1 buildroots:
arches='src')
@mock.patch('sys.stdout', new_callable=StringIO)
def test_handle_rpminfo_with_build(self, stdout):
rpm_nvra = 'test-rpm-1.1-11.noarch'
self.session.getBuildroot.return_value = self.buildroot_info
self.session.listBuildroots.return_value = [self.buildroot_info]
self.session.getBuild.return_value = self.buildinfo
self.session.getRPM.return_value = self.getrpminfo
self.session.listRPMs.return_value = self.listrpminfos
expected_output = """RPM: 7:test-rpm-1.1-11.noarch [294]
Build: test-rpm-1.1-11 [1]
RPM Path: /mnt/koji/packages/test-rpm/1.1/11/noarch/test-rpm-1.1-11.noarch.rpm
SRPM: 7:test-rpm-1.1-11 [290]
SRPM Path: /mnt/koji/packages/test-rpm/1.1/11/src/test-rpm-1.1-11.src.rpm
Built: Tue, 16 Mar 2021 06:56:49 UTC
SIGMD5: b2b95550390e5f213fc25f33822425f7
Size: 7030
Build ID: 1
Buildroot: 3 (tag test-tag, arch x86_64, repo 2)
Build Host: kojibuilder
Build Task: 10
Used in 1 buildroots:
id build tag arch build host
-------- ---------------------------- -------- -----------------------------
3 test-tag x86_64 kojibuilder
"""
anon_handle_rpminfo(self.options, self.session, ['--buildroot', '--build', 'any', rpm_nvra])
self.assert_console_message(stdout, expected_output)
self.session.getBuildroot.assert_called_once_with(self.getrpminfo['buildroot_id'])
self.session.listBuildroots.assert_called_once_with(queryOpts={'order': 'buildroot.id'},
rpmID=self.getrpminfo['id'])
self.session.getBuild.assert_called_once_with(self.getrpminfo['build_id'])
self.session.getRPM.assert_called_once_with(rpm_nvra, build='any')
self.session.listRPMs.assert_called_once_with(buildID=self.getrpminfo['build_id'],
arches='src')
def test_rpminfo_without_option(self):
arguments = []
self.assert_system_exit(
@ -223,11 +187,10 @@ Used in 1 buildroots:
def test_rpminfo_help(self):
self.assert_help(
anon_handle_rpminfo,
"""Usage: %s rpminfo [options] <n-v-r.a> [<n-v-r.a> ...]
"""Usage: %s rpminfo [options] <n-v-r.a|id> [<n-v-r.a|id> ...]
(Specify the --help global option for a list of other help options)
Options:
-h, --help show this help message and exit
--buildroots show buildroots the rpm was used in
--build=NVR|ID show the rpm(s) in the build
-h, --help show this help message and exit
--buildroots show buildroots the rpm was used in
""" % self.progname)

26
tests/test_hub/test_getRPM.py
View file

@ -29,7 +29,7 @@ class TestGetRPM(DBQueryTestCase):
def test_rpm_info_int(self):
rpminfo = 123
self.qp_execute_one_return_value = {'rpminfo.id': 123}
self.qp_execute_return_value = [{'rpminfo.id': 123}]
result = kojihub.get_rpm(rpminfo)
self.assertEqual(result, {'rpminfo.id': 123})
@ -40,7 +40,7 @@ class TestGetRPM(DBQueryTestCase):
'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 = 0', "rpminfo.id=%(id)s"])
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})
@ -116,27 +116,7 @@ class TestGetRPM(DBQueryTestCase):
'payloadhash', 'size', 'buildtime', 'metadata_only', 'extra']
self.assertEqual(set(query.columns), set(columns))
self.assertEqual(query.clauses,
["external_repo_id = %(external_repo_id)i", "rpminfo.id=%(id)s"])
self.assertEqual(query.joins,
['external_repo ON rpminfo.external_repo_id = external_repo.id'])
self.assertEqual(query.values, rpminfo_data)
def test_rpm_info_with_build(self):
rpminfo = {'id': 123, 'name': 'testrpm-1.23-4.x86_64.rpm', 'build_id': 101}
self.find_build_id.return_value = 101
rpminfo_data = rpminfo.copy()
kojihub.get_rpm(rpminfo, multi=True, build='any')
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.build_id = %(build_id)s", "rpminfo.id=%(id)s"])
["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)