Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def mock_nvr_downloads(build_info=None, archives=None, overrides=None):
if not build_info:
build_info = DEFAULT_KOJI_BUILD
if not archives:
archives = DEFAULT_ARCHIVES
if not overrides:
overrides = {}
pi = koji.PathInfo(topdir=KOJI_ROOT)
for archive in archives:
url = pi.mavenbuild(build_info) + '/' + pi.mavenfile(archive)
# Use any overrides for this archive ID
archive_overrides = overrides.get(archive['id'], {})
status = archive_overrides.get('status', 200)
body = archive_overrides.get('body', archive['filename'] + archive['group_id'])
responses.add(responses.GET, url, body=body, status=status)
expect_error = not root_url
if expect_error:
with pytest.raises(Exception):
read_yaml(config_yaml, 'schemas/config.json')
return
parsed_config = read_yaml(config_yaml, 'schemas/config.json')
fallback_map = {}
if fallback:
fallback_map = deepcopy(config['koji'])
else:
workflow.plugin_workspace[ReactorConfigPlugin.key][WORKSPACE_CONF_KEY] = \
ReactorConfig(parsed_config)
(flexmock(koji.PathInfo)
.should_receive('__init__')
.with_args(topdir=expected_root_url)
.once())
get_koji_path_info(workflow, fallback_map)
def fetch_koji_build(build):
"""
build ==> buildID or NVR
"""
if build.isdigit():
build = int(build)
urls = [] # output
pathinfo = koji.PathInfo(topdir=topurl)
session = koji.ClientSession(server)
info = session.getBuild(build)
# print session.listArchives(build)
# rpms = session.listRPMs(buildID=info['id'])
# if not rpms:
# print ":-("
# for rpm in rpms:
# fname = pathinfo.rpm(rpm)
# url = pathinfo.build(info) + '/' + fname
# print url
if not info:
return
task_id = info["task_id"]
nvr = info.get("nvr", str(task_id))
Args:
session (koji.ClientSession): Active Koji session instance.
build_id (str): Koji build ID.
destination (str): Path where to download files to.
arches (list): List of architectures to be downloaded.
Returns:
tuple: List of downloaded RPMs and list of downloaded logs.
Raises:
DownloadError: If download failed.
"""
build = session.getBuild(build_id)
pathinfo = koji.PathInfo(topdir=session.opts['topurl'])
rpms: List[str] = []
logs: List[str] = []
os.makedirs(destination, exist_ok=True)
for pkg in session.listBuildRPMs(build_id):
if pkg['arch'] not in arches:
continue
rpmpath = pathinfo.rpm(pkg)
local_path = os.path.join(destination, os.path.basename(rpmpath))
if local_path not in rpms:
url = pathinfo.build(build) + '/' + rpmpath
DownloadHelper.download_file(url, local_path)
rpms.append(local_path)
for logfile in session.getBuildLogs(build_id):
if logfile['dir'] not in arches:
continue
local_path = os.path.join(destination, logfile['name'])
def get_koji_archives(tag, package):
"""
Get the list of archives to download from koji
given a release tag and a package name.
"""
client = koji.ClientSession("https://koji.fedoraproject.org/kojihub")
builds = client.listTagged(tag, latest=True, package=package, type="image")
images = client.listArchives(buildID=builds[0]["id"], type="image")
pi = koji.PathInfo(topdir=DOWNLOAD_URL)
urls = []
for image in images:
if ".tar.xz" in image["filename"]:
urls.append(f"{pi.imagebuild(builds[0])}/{image['filename']}")
return urls
def get_srpm_url(srpm):
rel_pathinfo = koji.PathInfo(topdir=primary_koji_config['srpm_relative_path_root'])
return rel_pathinfo.build(srpm) + '/' + rel_pathinfo.rpm(srpm)
def buildinfo(environ, buildID):
values = _initValues(environ, 'Build Info', 'builds')
server = _getServer(environ)
topurl = environ['koji.options']['KojiFilesURL']
pathinfo = koji.PathInfo(topdir=topurl)
buildID = int(buildID)
build = server.getBuild(buildID)
values['title'] = koji.buildLabel(build) + ' | Build Info'
tags = server.listTags(build['id'])
tags.sort(_sortbyname)
rpms = server.listBuildRPMs(build['id'])
rpms.sort(_sortbyname)
typeinfo = server.getBuildType(buildID)
archiveIndex = {}
for btype in typeinfo:
archives = server.listArchives(build['id'], type=btype, queryOpts={'order': 'filename'})
idx = archiveIndex.setdefault(btype, {})
:param koji_session: Koji session used for queries
:param tag: Koji build tag name
:param name: Package name
:param relative: Whether the URL should be relative to Koji's work dir. Used for
submitting scratch-builds from SRPMS existing in the same Koji.
:param topdir: Alternative Koji topdir, defaults to one supplied in configuration.
:return: a tuple of (srpm_info, srpm_url) or None.
srpm_info is Koji's rpm info dictionary, contains 'epoch', 'version',
'release' fields (and more)
srpm_url is the URL pointing to the SRPM. May be relative if `relative` is
specified
"""
if not topdir:
topdir = koji_session.config[
'srpm_relative_path_root' if relative else 'topurl']
rel_pathinfo = koji.PathInfo(topdir=topdir)
info = koji_session.listTagged(tag, latest=True,
package=name, inherit=True)
if info:
srpms = koji_session.listRPMs(buildID=info[0]['build_id'],
arches='src')
if srpms:
return (srpms[0],
rel_pathinfo.build(info[0]) + '/' +
rel_pathinfo.rpm(srpms[0]))