145 lines
4.3 KiB
Python
Executable file
145 lines
4.3 KiB
Python
Executable file
#!/usr/bin/python3
|
|
#
|
|
# mass-tag.py - A utility to tag rebuilt packages.
|
|
#
|
|
# Copyright (C) 2009-2023 Red Hat, Inc.
|
|
# SPDX-License-Identifier: GPL-2.0+
|
|
#
|
|
# Authors:
|
|
# Jesse Keating <jkeating@redhat.com>
|
|
#
|
|
|
|
from __future__ import print_function
|
|
import koji
|
|
import os
|
|
import operator
|
|
import argparse
|
|
import rpm
|
|
|
|
# Set some variables
|
|
# Some of these could arguably be passed in as args.
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument('-t','--target', help='Tag to tag the builds into',required=True)
|
|
parser.add_argument('-s','--source',help='Tag holding the builds', required=True)
|
|
args = parser.parse_args()
|
|
target = args.target # tag to tag into
|
|
holdingtag = args.source # tag holding the rebuilds
|
|
newbuilds = {} # dict of packages that have a newer build in the target tag
|
|
|
|
# Permission Verification
|
|
def verify_tag_access(tag):
|
|
"""Check user has permission for target tag"""
|
|
user = kojisession.getUser()
|
|
tag_access_info = kojisession.checkTagAccess(tag, user["id"])
|
|
if tag_access_info[0] is False:
|
|
print(f"ERROR: User {user['name']} lacks {tag} access permission. Reason: {tag_access_info[2]}")
|
|
return False
|
|
else:
|
|
return True
|
|
|
|
if not verify_tag_access(target):
|
|
exit(1)
|
|
|
|
# Create a koji session
|
|
koji_module = koji.get_profile_module("fedora")
|
|
kojisession = koji_module.ClientSession(koji_module.config.server)
|
|
|
|
# Log into koji
|
|
kojisession.gssapi_login()
|
|
|
|
# Generate a list of builds to iterate over, sorted by package name
|
|
builds = sorted(kojisession.listTagged(holdingtag, latest=True),
|
|
key=operator.itemgetter('package_name'))
|
|
|
|
# Generate a list of packages in the target, reduced by not blocked.
|
|
pkgs = kojisession.listPackages(target, inherited=True)
|
|
pkgs = [pkg['package_name'] for pkg in pkgs if not pkg['blocked']]
|
|
|
|
print('Checking %s builds...' % len(builds))
|
|
|
|
# Use multicall
|
|
kojisession.multicall = True
|
|
|
|
# Loop over each build
|
|
for build in builds:
|
|
# Get the latest tagged package in the target
|
|
kojisession.listTagged(target, latest=True, package=build['package_name'])
|
|
|
|
# Get the results
|
|
results = kojisession.multiCall()
|
|
|
|
# Find builds that are newer in the target tag
|
|
kojisession.multicall = True
|
|
for build, [result] in zip(builds, results):
|
|
if not build['package_name'] in pkgs:
|
|
continue
|
|
|
|
for newbuild in result:
|
|
evr1 = (str(build['epoch'] or ''), build['version'], build['release'])
|
|
evr2 = (str(newbuild['epoch'] or ''), newbuild['version'], newbuild['release'])
|
|
if rpm.labelCompare(evr1, evr2) == -1:
|
|
newbuilds.setdefault(build['package_name'], []).append(newbuild)
|
|
|
|
requests = kojisession.multiCall()
|
|
|
|
# Loop through the results and tag if necessary
|
|
kojisession.multicall = True
|
|
all_tagged_builds = []
|
|
taglist = []
|
|
all_taglist = []
|
|
pkgcount = 0
|
|
|
|
for build in builds:
|
|
if not build['package_name'] in pkgs:
|
|
print('Skipping %s, blocked in %s' % (build['package_name'], target))
|
|
continue
|
|
|
|
if build['package_name'] in newbuilds:
|
|
print('Newer build found for %s.' % build['package_name'])
|
|
else:
|
|
print('Tagging %s into %s' % (build['nvr'], target))
|
|
taglist.append(build['nvr'])
|
|
all_tagged_builds.append(build)
|
|
pkgcount += 1
|
|
|
|
if pkgcount == 1000:
|
|
print('Tagging %s builds.' % pkgcount)
|
|
kojisession.massTag(target, taglist)
|
|
all_taglist.append(taglist)
|
|
results = kojisession.multiCall()
|
|
pkgcount = 0
|
|
taglist = []
|
|
kojisession.multicall = True
|
|
|
|
kojisession.massTag(target, taglist)
|
|
|
|
results = kojisession.multiCall()
|
|
|
|
# Final verification
|
|
print("\nFinal verification for {0} builds...".format(len(taglist)))
|
|
|
|
def verify_tagged(build_list):
|
|
"""Verify builds appear in target tag"""
|
|
missing = []
|
|
kojisession.multicall = True
|
|
for build in build_list:
|
|
kojisession.listTagged(target, package=build['package_name'], build=build['nvr'])
|
|
results = kojisession.multiCall()
|
|
|
|
for build, result in zip(build_list, results):
|
|
if not result[0]:
|
|
missing.append(build['nvr'])
|
|
|
|
return missing
|
|
|
|
missing = verify_tagged(all_tagged_builds)
|
|
|
|
if missing:
|
|
print(f"ERROR: {len(missing)} builds not in {target}:")
|
|
for nvr in missing:
|
|
print(f" - {nvr}")
|
|
exit(1)
|
|
else:
|
|
print(f"SUCCESS: All {len(all_tagged_builds)} builds verified in {target}")
|
|
print(f'Total tagged builds: {len(taglist)}')
|
|
|