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

PR#3255: allow untag-build for blocked packages

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

Fixes: #3134
https://pagure.io/koji/issue/3134
Allow untagging "invisible" tagged builds
This commit is contained in:
Tomas Kopecek 2022-03-21 14:32:30 +01:00
commit 4429b88385
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):

117
tests/test_cli/test_untag_build.py
View file

@ -14,6 +14,33 @@ class TestUntagBuild(utils.CliTestCase):
self.options.debug = False
self.session = mock.MagicMock()
self.session.getAPIVersion.return_value = koji.API_VERSION
self.activate_session_mock = mock.patch('koji_cli.commands.activate_session').start()
self.tag_info = {'arches': 'x86_64',
'extra': {},
'id': 1,
'locked': False,
'maven_include_all': False,
'maven_support': False,
'name': 'test-tag',
'perm': None,
'perm_id': None}
self.tag = 'test-tag'
self.pkg_name = 'test-package'
self.tag_listing = \
{'tag_listing': [{'active': True, 'build_id': 3, 'create_event': 1357, 'epoch': 1,
'name': 'test-package', 'release': '1.f35', 'tag.name': 'test-tag',
'tag_id': 460, 'version': '1.6'},
{'active': True, 'build_id': 3, 'create_event': 1357, 'epoch': 1,
'name': 'test-package', 'release': '1.f35', 'tag.name': 'test-tag',
'tag_id': 460, 'version': '1.1'}]}
def __vm(self, result):
m = koji.VirtualCall('mcall_method', [], {})
if isinstance(result, dict) and result.get('faultCode'):
m._result = result
else:
m._result = (result,)
return m
@mock.patch('sys.stderr', new_callable=StringIO)
def test_untag_build_without_option(self, stderr):
@ -21,11 +48,15 @@ class TestUntagBuild(utils.CliTestCase):
"(Specify the --help global option for a list of other help options)\n\n" \
"%s: error: This command takes at least two arguments: " \
"a tag name/ID and one or more package " \
"n-v-r's\n" % (self.progname, self.progname)
"n-v-r's or package names\n" % (self.progname, self.progname)
with self.assertRaises(SystemExit) as ex:
handle_untag_build(self.options, self.session, [])
self.assertExitCode(ex, 2)
self.assert_console_message(stderr, expected)
self.activate_session_mock.assert_not_called()
self.session.getTag.assert_not_called()
self.session.queryHistory.assert_not_called()
self.session.untagBuild.assert_not_called()
@mock.patch('sys.stderr', new_callable=StringIO)
def test_untag_build_without_option_non_latest_force(self, stderr):
@ -36,16 +67,92 @@ class TestUntagBuild(utils.CliTestCase):
handle_untag_build(self.options, self.session, ['--non-latest', '--force'])
self.assertExitCode(ex, 2)
self.assert_console_message(stderr, expected)
self.activate_session_mock.assert_not_called()
self.session.getTag.assert_not_called()
self.session.queryHistory.assert_not_called()
self.session.untagBuild.assert_not_called()
@mock.patch('sys.stderr', new_callable=StringIO)
def test_untag_build_non_exist_tag(self, stderr):
tag = 'test-tag'
pkg_info = {'id': 9, 'name': 'test-build'}
expected = "Usage: %s untag-build [options] <tag> <pkg> [<pkg> ...]\n" \
"(Specify the --help global option for a list of other help options)\n\n" \
"%s: error: No such tag: %s\n" % (self.progname, self.progname, tag)
"%s: error: No such tag: %s\n" % (self.progname, self.progname, self.tag)
self.session.getTag.return_value = None
with self.assertRaises(SystemExit) as ex:
handle_untag_build(self.options, self.session, [tag, pkg_info['name']])
handle_untag_build(self.options, self.session, [self.tag, self.pkg_name])
self.assertExitCode(ex, 2)
self.assert_console_message(stderr, expected)
self.activate_session_mock.assert_called_once_with(self.session, self.options)
self.session.getTag.assert_called_once_with(self.tag)
self.session.queryHistory.assert_not_called()
self.session.untagBuild.assert_not_called()
@mock.patch('sys.stdout', new_callable=StringIO)
def test_untag_build_all(self, stdout):
self.session.getTag.return_value = self.tag_info
mcall = self.session.multicall.return_value.__enter__.return_value
mcall.queryHistory.return_value = self.__vm(self.tag_listing)
handle_untag_build(self.options, self.session, ['--all', self.tag, self.pkg_name])
expected = ''
self.assert_console_message(stdout, expected)
self.activate_session_mock.assert_called_once_with(self.session, self.options)
self.session.getTag.assert_called_once_with(self.tag)
self.session.multicall.has_calls()
@mock.patch('sys.stdout', new_callable=StringIO)
def test_untag_build_all_verbose(self, stdout):
self.session.getTag.return_value = self.tag_info
mcall = self.session.multicall.return_value.__enter__.return_value
mcall.queryHistory.return_value = self.__vm(self.tag_listing)
handle_untag_build(self.options, self.session,
['--all', '--verbose', self.tag, self.pkg_name])
expected = 'untagging test-package-1.1-1.f35\nuntagging test-package-1.6-1.f35\n'
self.assert_console_message(stdout, expected)
self.activate_session_mock.assert_called_once_with(self.session, self.options)
self.session.getTag.assert_called_once_with(self.tag)
self.session.multicall.has_calls()
@mock.patch('sys.stdout', new_callable=StringIO)
def test_untag_build_all_non_latest_verbose(self, stdout):
self.session.getTag.return_value = self.tag_info
mcall = self.session.multicall.return_value.__enter__.return_value
mcall.queryHistory.return_value = self.__vm(self.tag_listing)
handle_untag_build(self.options, self.session,
['--non-latest', '--verbose', self.tag, self.pkg_name])
expected = 'Leaving latest build for package test-package: test-package-1.6-1.f35\n' \
'untagging test-package-1.1-1.f35\n'
self.assert_console_message(stdout, expected)
self.activate_session_mock.assert_called_once_with(self.session, self.options)
self.session.getTag.assert_called_once_with(self.tag)
self.session.multicall.has_calls()
@mock.patch('sys.stdout', new_callable=StringIO)
def test_untag_build_all_non_latest_verbose_only_one_pkg(self, stdout):
self.session.getTag.return_value = self.tag_info
tag_listing = \
{'tag_listing': [{'active': True, 'build_id': 3, 'create_event': 1357, 'epoch': 1,
'name': 'test-package', 'release': '1.f35',
'tag.name': 'test-tag', 'tag_id': 460, 'version': '1.6'}]}
mcall = self.session.multicall.return_value.__enter__.return_value
mcall.queryHistory.return_value = self.__vm(tag_listing)
handle_untag_build(self.options, self.session,
['--non-latest', '--verbose', self.tag, self.pkg_name])
expected = 'Leaving latest build for package test-package: test-package-1.6-1.f35\n'
self.assert_console_message(stdout, expected)
self.activate_session_mock.assert_called_once_with(self.session, self.options)
self.session.getTag.assert_called_once_with(self.tag)
self.session.multicall.has_calls()
@mock.patch('sys.stdout', new_callable=StringIO)
def test_untag_build_all_non_latest_force_test(self, stdout):
self.session.getTag.return_value = self.tag_info
self.session.queryHistory.return_value = self.tag_listing
handle_untag_build(self.options, self.session,
['--non-latest', '--force', '--test', self.tag])
expected = 'would have untagged test-package-1.1-1.f35\n'
self.assert_console_message(stdout, expected)
self.activate_session_mock.assert_called_once_with(self.session, self.options)
self.session.getTag.assert_called_once_with(self.tag)
self.session.multicall.assert_called_once()
self.session.untagBuild.assert_not_called()