Create function to download RPMs from koji for given NVR Let's try this again, without the wrapping. I'll push to a remote branch if this doesn't turn out.
This patch for master provides a fix for #265 without the extra py.test
stuff from my pytest branch. It also changes 2 tests to use the new
function. I have tested this on my local setup and as far as I can tell,
it works with no regressions.
There were some concerns about the refactoring that I did in
koji_utils.py. I can remove those changes and re-submit if desired.
Tim
diff --git a/lib/python/koji_utils.py b/lib/python/koji_utils.py
index 1eb2c44..71d6d8a 100644
--- a/lib/python/koji_utils.py
+++ b/lib/python/koji_utils.py
@@ -25,26 +25,40 @@ from repoinfo import repoinfo
import rpmUtils.miscutils
import sys
from autoqa.config import SingleConfigParser, autoqa_conf
+import urlgrabber.grabber
+import os.path
-kojiurl = autoqa_conf.get('resources', 'koji_url')
-pkgurl = autoqa_conf.get('resources', 'pkg_url')
+class SimpleKojiClientSession():
-class SimpleKojiClientSession(koji.ClientSession):
- '''Convenience wrapper class for interacting with koji'''
- def __init__(self, server=kojiurl, opts=None):
- return koji.ClientSession.__init__(self, server, opts)
+ '''Convenience wrapper class for interacting with kojiclient'''
+ def __init__(self, server=None, opts=None, kojiclient=None,
url_grabber=None):
+ self.kojiurl = autoqa_conf.get('resources', 'koji_url')
+ self.pkgurl = autoqa_conf.get('resources', 'pkg_url')
+
+ if server == None:
+ server = self.kojiurl
+
+ if kojiclient == None:
+ self.kojiclient = koji.ClientSession(server, opts)
+ else:
+ self.kojiclient = kojiclient
+
+ if url_grabber == None:
+ self.url_grabber = urlgrabber.grabber.URLGrabber()
+ else:
+ self.url_grabber = url_grabber
def list_builds_since(self, timestamp):
'''Return a list of new builds since the given timestamp'''
- return self.listBuilds(completeAfter=timestamp)
+ return self.kojiclient.listBuilds(completeAfter=timestamp)
def list_tags(self, nvr):
'''Return a list of tags applied to the package with the given nvr'''
- return [t['name'] for t in self.listTags(nvr)]
+ return [t['name'] for t in self.kojiclient.listTags(nvr)]
def tag_history(self, nvr):
'''Returns a list of every tag ever applied to the given nvr'''
- return [t['tag_name'] for t in self.tagHistory(build=nvr)]
+ return [t['tag_name'] for t in self.kojiclient.tagHistory(build=nvr)]
def latest_by_tag(self, tag, pkgname, max_evr=None):
'''Get the latest package for the given name in the given tag. If you
@@ -56,7 +70,7 @@ class SimpleKojiClientSession(koji.ClientSession):
max_evr = (0, max_evr[1], max_evr[2])
if max_evr:
- pkgs = self.listTagged(tag, package=pkgname)
+ pkgs = self.kojiclient.listTagged(tag, package=pkgname)
# sort the packages from highest EVR to lowest
pkgs.sort(cmp=lambda x, y: compareEVR(x, y), reverse=True)
for pkg in pkgs[:]:
@@ -68,19 +82,19 @@ class SimpleKojiClientSession(koji.ClientSession):
# since packages are sorted, we can stop here
break
else:
- pkgs = self.listTagged(tag, package=pkgname, latest=True)
+ pkgs = self.kojiclient.listTagged(tag, package=pkgname,
latest=True)
if pkgs:
# the winner is the first package in list
return pkgs[0]
def build_failed(self, nvr):
- return (self.getBuild(nvr)['state'] > 1)
+ return (self.kojiclient.getBuild(nvr)['state'] > 1)
- # Idea taken from the koji CLI
+ # Idea taken from the kojiclient CLI
def ensure_connection(self):
'''Check the connection to koji.'''
try:
- ret = self.getAPIVersion()
+ ret = self.kojiclient.getAPIVersion()
except xmlrpclib.ProtocolError:
print >> sys.stderr, "Unable to connect to server"
sys.exit(2)
@@ -93,7 +107,7 @@ class SimpleKojiClientSession(koji.ClientSession):
timestamp = time.time()
while not builds and (starttime - timestamp < 10000):
timestamp -= 3000
- builds = self.list_builds_since(timestamp)
+ builds = self.kojiclient.list_builds_since(timestamp)
print "Most recent builds: %.0f min ago" % ((starttime-timestamp)/60.0)
return builds
@@ -146,11 +160,11 @@ class SimpleKojiClientSession(koji.ClientSession):
return prev_rlss[0]
def nvr_to_urls(self, nvr, arches=None, debuginfo=False):
- info = self.getBuild(nvr)
- baseurl = '/'.join((pkgurl, info['package_name'],
+ info = self.kojiclient.getBuild(nvr)
+ baseurl = '/'.join((self.pkgurl, info['package_name'],
info['version'], info['release']))
- rpms = self.listRPMs(buildID=info['id'], arches=arches)
+ rpms = self.kojiclient.listRPMs(buildID=info['id'], arches=arches)
if not debuginfo:
rpms = filter(lambda r: not r['name'].endswith('-debuginfo'), rpms)
# XXX option to skip srpms too?
@@ -161,11 +175,11 @@ class SimpleKojiClientSession(koji.ClientSession):
''' Get a list of RPMs existing for specified NVR. All RPMs have 'url'
key added with URL where to download the RPM as the value.'''
- info = self.getBuild(nvr)
- baseurl = '/'.join((pkgurl, info['package_name'],
+ info = self.kojiclient.getBuild(nvr)
+ baseurl = '/'.join((self.pkgurl, info['package_name'],
info['version'], info['release']))
- rpms = self.listRPMs(buildID=info['id'])
+ rpms = self.kojiclient.listRPMs(buildID=info['id'])
if not debuginfo:
rpms = filter(lambda r: not r['name'].endswith('-debuginfo'), rpms)
if not src:
@@ -176,6 +190,19 @@ class SimpleKojiClientSession(koji.ClientSession):
return rpms
+ def get_nvr_rpms(self, nvr, rpm_dir, arches=None, debuginfo=None):
+ ''' Retrieve the rpms associated with an nvr into the specified
+ directory. Return the local filenames of the grabbed rpms
+ '''
+
+ rpm_urls = self.nvr_to_urls(nvr, arches=arches, debuginfo=debuginfo)
+ rpm_files = []
+ for url in rpm_urls:
+ local_file = os.path.join(rpm_dir, os.path.basename(url))
+ rpm_file = self.url_grabber.urlgrab(url, local_file)
+ rpm_files.append(rpm_file)
+ return rpm_files
+
def compareEVR(nvr1, nvr2):
''' Compare EVR from two NVR objects. '''
diff --git a/tests/initscripts/initscripts.py b/tests/initscripts/initscripts.py
index 2261ceb..69a49ba 100644
--- a/tests/initscripts/initscripts.py
+++ b/tests/initscripts/initscripts.py
@@ -140,16 +140,9 @@ class initscripts(AutoQATest):
#install packages from koji
koji = autoqa.koji_utils.SimpleKojiClientSession()
- pkgurls = koji.nvr_to_urls(nvr, arches = os.uname()[-1])
- rpms = []
print "Saving RPMs to %s" % self.rpmdir
- #download packages
- for p in pkgurls:
- # fetch package to rpmdir
- print "Grabbing %s" % p
- localfile = os.path.join(self.rpmdir, os.path.basename(p))
- autoqa.util.grabber.urlgrab(p, localfile)
- rpms.append(localfile)
+ rpms = koji.get_nvr_rpms(nvr, self.rpmdir, arches = os.uname()[-1])
+
#and install them
cmd = "yum -y --nogpgcheck localinstall %s" % " ".join(rpms)
utils.system_output(cmd)
diff --git a/tests/rpmlint/rpmlint.py b/tests/rpmlint/rpmlint.py
index 73d798f..3997485 100644
--- a/tests/rpmlint/rpmlint.py
+++ b/tests/rpmlint/rpmlint.py
@@ -74,12 +74,8 @@ class rpmlint(AutoQATest):
os.remove(os.path.join(self.rpmdir, rpm))
# download packages
- pkgurls = koji.nvr_to_urls(nvr)
print "Saving RPMs to %s" % self.rpmdir
- for p in pkgurls:
- print "Grabbing %s" % p
- localfile = os.path.join(self.rpmdir, os.path.basename(p))
- autoqa.util.grabber.urlgrab(p, localfile)
+ koji.get_nvr_rpms(nvr, self.rpmdir)
# run rpmlint
cmd = 'rpmlint %s' % self.rpmdir
signature.asc
Description: OpenPGP digital signature
_______________________________________________ autoqa-devel mailing list [email protected] https://fedorahosted.org/mailman/listinfo/autoqa-devel
