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

allow untag-build for blocked packages

Browse source 
https://pagure.io/koji/issue/3134
This commit is contained in:
Jana Cupova 2022-02-07 14:58:10 +01:00 • committed by Tomas Kopecek
parent 2a6f186210
commit d561ded4fa
2 changed files with 149 additions and 26 deletions
Download patch file Download diff file Expand all files Collapse all files

58
cli/koji_cli/commands.py
View file

@ -6943,9 +6943,10 @@ def handle_untag_build(goptions, session, args):
usage = "usage: %prog untag-build [options] <tag> <pkg> [<pkg> ...]"
parser = OptionParser(usage=get_usage_str(usage))
parser.add_option("--all", action="store_true",
help="untag all versions of the package in this tag")
help="untag all versions of the package in this tag, pkg is package name")
parser.add_option("--non-latest", action="store_true",
help="untag all versions of the package in this tag except the latest")
help="untag all versions of the package in this tag except the latest, "
"pkg is package name")
parser.add_option("-n", "--test", action="store_true", help="test mode")
parser.add_option("-v", "--verbose", action="store_true", help="print details")
parser.add_option("--force", action="store_true", help="force operation")
@ -6956,43 +6957,56 @@ def handle_untag_build(goptions, session, args):
elif len(args) < 2:
parser.error(
"This command takes at least two arguments: a tag name/ID and one or more package "
"n-v-r's")
"n-v-r's or package names")
activate_session(session, goptions)
tag = session.getTag(args[0])
if not tag:
parser.error("No such tag: %s" % args[0])
if options.all:
result = []
with session.multicall() as m:
result.extend([m.queryHistory(tag=args[0], package=pkg, active=True)
for pkg in args[1:]])
builds = []
for pkg in args[1:]:
builds.extend(session.listTagged(args[0], package=pkg))
for r in result:
builds.extend(r.result['tag_listing'])
elif options.non_latest:
if options.force and len(args) == 1:
tagged = session.listTagged(args[0])
tagged = session.queryHistory(tag=args[0], active=True)['tag_listing']
tagged = sorted(tagged, key=lambda k: (k['create_event']), reverse=True)
else:
result = []
with session.multicall() as m:
result.extend([m.queryHistory(tag=args[0], package=pkg, active=True)
for pkg in args[1:]])
tagged = []
for pkg in args[1:]:
tagged.extend(session.listTagged(args[0], package=pkg))
for r in result:
tagged.extend(r.result['tag_listing'])
tagged = sorted(tagged, key=lambda k: (k['create_event']), reverse=True)
# listTagged orders entries latest first
seen_pkg = {}
builds = []
for binfo in tagged:
if binfo['name'] not in seen_pkg:
# latest for this package
nvr = '%s-%s-%s' % (binfo['name'], binfo['version'], binfo['release'])
if options.verbose:
print("Leaving latest build for package %(name)s: %(nvr)s" % binfo)
print("Leaving latest build for package %s: %s" % (binfo['name'], nvr))
else:
builds.append(binfo)
seen_pkg[binfo['name']] = 1
else:
# find all pkg's builds in tag
pkgs = set([koji.parse_NVR(nvr)['name'] for nvr in args[1:]])
tagged = []
result = []
with session.multicall() as m:
for pkg in pkgs:
tagged.append(m.listTagged(args[0], package=pkg))
result.extend([m.queryHistory(tag=args[0], pkg=pkg, active=True) for pkg in pkgs])
tagged = []
for r in result:
tagged.append(r.result['tag_listing'])
# flatten
tagged = list(itertools.chain(*[t.result for t in tagged]))
idx = dict([(b['nvr'], b) for b in tagged])
tagged = list(itertools.chain(*[t for t in tagged]))
idx = dict([('%s-%s-%s' % (b['name'], b['version'], b['release']), b) for b in tagged])
# check exact builds
builds = []
@ -7010,13 +7024,15 @@ def handle_untag_build(goptions, session, args):
if not options.force:
error()
builds.reverse()
for binfo in builds:
if options.test:
print("would have untagged %(nvr)s" % binfo)
else:
if options.verbose:
print("untagging %(nvr)s" % binfo)
session.untagBuild(tag['name'], binfo['nvr'], force=options.force)
with session.multicall(strict=True) as m:
for binfo in builds:
build_nvr = '%s-%s-%s' % (binfo['name'], binfo['version'], binfo['release'])
if options.test:
print("would have untagged %s" % build_nvr)
else:
if options.verbose:
print("untagging %s" % build_nvr)
m.untagBuild(tag['name'], build_nvr, force=options.force)
def handle_unblock_pkg(goptions, session, args):