[PATCH 3/5] Improve code of multihook tests

Kamil Páral kparal at redhat.com
Fri Sep 10 13:52:29 UTC 2010


This patch fixes some issues created by adding multihook capabilities to
rpmlint, rpmguard and initscripts tests with patch
f16b2646fa397b0cd55e3e4bf9918d21541e8840. It re-enables opt-in emails.
It fixes problem where rpmlint's rpm dir cache wasn't cleared between
successive runs. It also cleans up the code a lot. A lot of recent
changes made the code almost unmaintainable and very hard to read
(especially in printing/output appending/log appending tasks). This
patch reworked all of that quite a lot, it should be much more readable
and simpler now. In short, it tries again to have these scripts ready to
be served as examples for other people.

This patch also removes autotest exception throwing and uses assertions.
---
 tests/initscripts/initscripts.py |  144 ++++++++--------
 tests/rpmguard/rpmguard.py       |  349 +++++++++++++++++++------------------
 tests/rpmlint/rpmlint.py         |  145 ++++++++++------
 3 files changed, 348 insertions(+), 290 deletions(-)

diff --git a/tests/initscripts/initscripts.py b/tests/initscripts/initscripts.py
index 1728cef..c7b322a 100644
--- a/tests/initscripts/initscripts.py
+++ b/tests/initscripts/initscripts.py
@@ -26,7 +26,6 @@ import autoqa.koji_utils
 import autoqa.util
 import koji
 from autotest_lib.client.bin import utils
-from autotest_lib.client.common_lib import error
 from autoqa.repoinfo import repoinfo
 from autoqa.test import AutoQATest
 from autoqa.decorators import ExceptionCatcher
@@ -45,6 +44,12 @@ class initscripts(AutoQATest):
             f.close()
         utils.system('yum -y install beakerlib') 
 
+    @ExceptionCatcher("self.initialize_failed")
+    def initialize(self, config, **kwargs):
+        super(initscripts, self).initialize(config)
+        self.rpmdir = os.path.join(self.tmpdir, 'rpms')
+        os.makedirs(self.rpmdir)
+
     def run_beakerlib_test(self, name, path_to_runtest):
         #install required packages
         """
@@ -55,6 +60,8 @@ class initscripts(AutoQATest):
 
         We also want to skip the "Requires" line with the package name, since it was installed
          from koji. Makefile has "RunFor" line, which specifies this package name.
+
+        Returns: (result, outputs)
         """
         makefile = os.path.join(path_to_runtest, "Makefile")
         if (os.path.isfile(makefile)):
@@ -73,8 +80,9 @@ class initscripts(AutoQATest):
 
         #run the test
         cmd = os.path.join(path_to_runtest, "runtest.sh")
-        result = utils.run(cmd, ignore_status = True, stdout_tee = utils.TEE_TO_LOGS)
-        self.results += result.stdout
+        cmd_result = utils.run(cmd, ignore_status = True, stdout_tee = utils.TEE_TO_LOGS)
+        result = 'PASSED' if cmd_result.exit_status == 0 else 'FAILED'
+        outputs = cmd_result.stdout
 
         #store the log journal created by beakerlib
         """
@@ -83,73 +91,61 @@ class initscripts(AutoQATest):
         :: [15:22:06] ::  JOURNAL TXT: /tmp/beakerlib-CoEg7JM/journal.txt
         extracts paths to these files, ands stores the files in self.resultsdir
         """
-        journal_lines = filter(lambda x: x.find("JOURNAL") > -1, self.results.splitlines())
+        journal_lines = filter(lambda x: x.find("JOURNAL") > -1, outputs.splitlines())
         journal_files = map(lambda x: x.split(':')[-1].strip(), journal_lines)
         for fname in journal_files:
             shutil.copy(fname, os.path.join(self.resultsdir, "journal_%s.%s" % (name, fname.split('.')[-1])))
 
-        return result.exit_status
+        return (result, outputs)
 
     @ExceptionCatcher("self.run_once_failed")
-    def run_once(self, name, kojitag, **kwargs):
-        """
-        name - name of the package for initscript test
-        """
+    def run_once(self, kojitag, **kwargs):
         if kwargs['hook'] == 'post-koji-build':
             envrs = [kwargs['envr']]
+            update_id = kwargs['envr']
         elif kwargs['hook'] == 'post-bodhi-update':
             envrs = kwargs['envrs']
+            update_id = kwargs['name'] or kwargs['id']
+
+        self.result = 'PASSED'
+        self.results_count = {'PASSED': 0, 'FAILED': 0} # count how many of which results do we have
+        self.tests_passed = []
+        self.tests_failed = []
+        self.outputs = []
+        self.highlights = []
+        self.log = open(os.path.join(self.resultsdir,'initscripts.log'),'wb') # where to log output
 
-        tests_exit_status = []
-        tests_exit_status_zero = []
-        tests_exit_status_nonzero = []
-        failed_packages = []
-        self.result = "PASSED"
-        self.results = ''
-        self.outputs = ''
         for envr in envrs:
-            msg = 40*'=' + '\n' + envr + '\n' + 40*'='
-            self.outputs += msg
-            self.outputs += "\n"
+            # add header
+            msg = '%s\n%s\n%s' % ('='*40, envr, '='*40)
             print msg
+            self.outputs.append(msg)
             #find all tests for package $name
             nvrea = rpmUtils.miscutils.splitFilename(envr + '.noarch')
             name = nvrea[0]
             testdir = os.path.join(self.bindir, "./tests/%s" % name)
+            assert os.path.isdir(testdir), "No initscript checker found for package %s" % name
             tests = []
-            if (os.path.isdir(testdir)):
-                """
-                walks through ./tests/$NAME and all it's subdirs. Every time it finds
-                file runtest.sh, it stores a tuple(test_name, path) to tests[],
-                where test_name is 'openssh' for ./tests/openssh, openssh_subdir for ./tests/openssh/subdir, etc.
-                """
-                for (path, dirs, files) in os.walk(testdir):
-                    if "runtest.sh" in files:
-                        test_name = path.replace(os.path.join(self.bindir, "./tests/"), "").replace(os.path.sep, "_")
-                        tests.append((test_name, path))
-            else:
-                msg = "No initscript checker found for package %s" % name
-                if kwargs['hook'] == 'post-bodhi-update':
-                    self.outputs += msg
-                    self.outputs += "\n"
-                self.summary = msg
-                failed_packages.append(envr)
-                self.result = "FAILED"
-                continue
-                #raise error.TestWarn(self.summary)
-
-            #install packages from koji (stolen from rpmlint test)
+            """
+            walks through ./tests/$NAME and all it's subdirs. Every time it finds
+            file runtest.sh, it stores a tuple(test_name, path) to tests[],
+            where test_name is 'openssh' for ./tests/openssh, openssh_subdir for ./tests/openssh/subdir, etc.
+            """
+            for (path, dirs, files) in os.walk(testdir):
+                if "runtest.sh" in files:
+                    test_name = path.replace(os.path.join(self.bindir, "./tests/"), "").replace(os.path.sep, "_")
+                    tests.append((test_name, path))
+
+            #install packages from koji
             koji = autoqa.koji_utils.SimpleKojiClientSession()
             pkgurls = koji.nvr_to_urls(envr, arches = os.uname()[-1])
-            rpmdir = os.path.join(self.tmpdir, 'rpms')
-            os.makedirs(rpmdir)
             rpms = []
-            print "Saving RPMs to %s" % rpmdir
+            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(rpmdir, os.path.basename(p))
+                localfile = os.path.join(self.rpmdir, os.path.basename(p))
                 autoqa.util.grabber.urlgrab(p, localfile)
                 rpms.append(localfile)
             #and install them
@@ -158,38 +154,48 @@ class initscripts(AutoQATest):
 
             #sort tests alphabetically by test_name
             tests.sort(key = lambda x: x[0])
-            for test in tests:
-                msg = "RUNNING: %s\n%s\n" % (test[0], "=" * len("RUNNING: %s" % test[0]))
-                self.outputs += msg
-                tests_exit_status.append((test[0], self.run_beakerlib_test(test[0], test[1])))
-                self.outputs += "\n"
-                self.outputs += self.results
-                self.outputs += "\n"
 
-            tests_exit_status_nonzero = filter(lambda x: x[1] != 0, tests_exit_status)
-            tests_exit_status_zero = filter(lambda x: x[1] == 0, tests_exit_status)
+            # run tests
+            for test in tests:
+                msg = "RUNNING: %s" % test[0]
+                msg += '\n%s' % ("="*len(msg),)
+                print msg
+                self.outputs.append(msg)
+
+                (result, outputs) = self.run_beakerlib_test(test[0], test[1])
+                self.results_count[result] += 1
+                self.outputs.append(outputs)
+
+                if result == 'PASSED':
+                    self.tests_passed.append(test[0])
+                else:
+                    self.tests_failed.append(test[0])
+
+            # add one empty line
+            msg = ''
+            print msg
+            self.outputs.append(msg)
 
             # email results to mailing list and to pkg owner if they optin
             repo = repoinfo.getrepo_by_tag(kojitag)
-            #if repo is not None and autoqa.util.check_opt_in(name, repo['collection_name']):
+            if repo is not None and autoqa.util.check_opt_in(name, repo['collection_name']):
                 #FIXME - hardcoded partial email address here - obviously sub-par
-                #self.mail_to.append('%s-owner at fedoraproject.org' % name)
+                self.mail_to.append('%s-owner at fedoraproject.org' % name)
 
-            if len(tests_exit_status_nonzero) != 0:
-                self.result = "FAILED"
 
-            self.highlights += "Following tests returned zero exit code: %s\n" % repr(tests_exit_status_zero)
-            self.highlights += "Following tests returned NON-zero exit code: %s\n" % repr(tests_exit_status_nonzero)
+        # set result variables
+        if len(self.tests_failed) > 0:
+            self.result = 'FAILED'
 
-        self.summary = "%d PASSED, %d FAILED for package %s" % (len(tests_exit_status_zero), len(tests_exit_status_nonzero), envr)
+        self.highlights.append("Following tests passed: %s" % ', '.join(self.tests_passed))
+        self.highlights.append("Following tests failed: %s" % ', '.join(self.tests_failed))
 
-        if kwargs['hook'] == 'post-bodhi-update':
-            self.summary = '%d OK, %d FAILED' % (len(envrs) - len(failed_packages), len(failed_packages))
-            self.outputs += '---\n' + self.summary + '\n---'
-            if self.result == 'FAILED':
-                raise error.TestFail
+        self.summary = "%d PASSED, %d FAILED for %s" % (self.results_count['PASSED'], self.results_count['FAILED'], update_id)
 
-        if len(tests_exit_status_nonzero) > 0:
-             msg = "Following tests returned non-zero exit code: %s" % repr(tests_exit_status_nonzero)
-             raise error.TestError(msg)
+        # reformat result variables
+        self.outputs = '\n'.join(self.outputs)
+        self.highlights = '\n'.join(self.highlights)
 
+        # log outputs
+        self.log.write(self.outputs)
+        self.log.close()
diff --git a/tests/rpmguard/rpmguard.py b/tests/rpmguard/rpmguard.py
index 54e30a6..41032df 100644
--- a/tests/rpmguard/rpmguard.py
+++ b/tests/rpmguard/rpmguard.py
@@ -19,7 +19,6 @@
 # Author: Kamil Paral <kparal at redhat.com>
 
 from autotest_lib.client.bin import utils
-from autotest_lib.client.common_lib import error
 import autoqa.koji_utils
 import autoqa.util
 from autoqa.repoinfo import repoinfo
@@ -32,196 +31,206 @@ class rpmguard(AutoQATest):
     version = 1 # increment if setup() changes
 
     def setup(self):
-        # Download latest rawhide rpmlint because rpmlint in F12 (ver 0.91)
-        # is too old
-        rpmlint = utils.system_output("koji latest-pkg dist-f13 --quiet \
-rpmlint | cut -d ' ' -f 1", retain_output=True)
-        utils.system('cd %s; koji download-build --arch noarch \
---latestfrom dist-f13 rpmlint' % self.tmpdir)
-        utils.system('yum -y localinstall --nogpgcheck %s/%s.noarch.rpm' \
-            % (self.tmpdir, rpmlint))
+        utils.system('yum -y install rpmlint')
 
     @ExceptionCatcher("self.initialize_failed")
     def initialize(self, config, **kwargs):
         super(rpmguard, self).initialize(config)
-        # we have several results (comparing different architectures), let's
-        # store it separately
-        self.results = []
-        # temporary results for each architecture
-        self.arch_results = {}
         self.rpmguard = os.path.join(self.bindir, 'rpmguard')
         self.rpmdir = os.path.join(self.tmpdir, 'rpms')
         os.makedirs(self.rpmdir)
 
     @ExceptionCatcher("self.run_once_failed")
     def run_once(self, kojitag, **kwargs):
-        test_result = 'PASSED'
-        output = []
-        warning_count = 0
-        log = ''
-        failed_packages = set()
-        # where to log output
-        out = open(os.path.join(self.resultsdir,'rpmguard.log'),'wb')
-        koji = autoqa.koji_utils.SimpleKojiClientSession()
-
         if kwargs['hook'] == 'post-koji-build':
             envrs = [kwargs['envr']]
+            update_id = kwargs['envr']
         elif kwargs['hook'] == 'post-bodhi-update':
             envrs = kwargs['envrs']
+            update_id = kwargs['name'] or kwargs['id']
+
+        self.result = 'PASSED'
+        # order for evaluation of final result; higher index means preference
+        self.result_order = ('PASSED','INFO','FAILED','ABORTED')
+        self.envr_results = {} # results for invidual packages
+        self.outputs = []
+        self.highlights = []
+        self.log = open(os.path.join(self.resultsdir,'rpmguard.log'),'wb') # where to log output
 
         for envr in envrs:
-            # get the most recent release available
-            # add .noarch to parse filename correctly
-            msg = 40*'=' + '\n' + envr + '\n' + 40*'='
-            output.append(msg)
+            # add header
+            msg = '%s\n%s\n%s' % ('='*40, envr, '='*40)
             print msg
-            nvrea = rpmUtils.miscutils.splitFilename(envr + '.noarch')
-            name = nvrea[0]
-            lastBuild = koji.list_previous_release(name, kojitag,
-                max_evr=(nvrea[3], nvrea[1], nvrea[2]))
-            # if there is no such build, we don't have anything to compare
-            if not lastBuild:
-                warn = "N: There is no previous build of %s in parents of \
-    %s tag." % (envr, kojitag)
-                output.append(warn)
-                print warn
-                self.result = "PASSED"
-                self.summary = warn
-                out.write(warn + '\n')
-                continue
-
-            # now we need list of RPMs available for each build
-            new_rpms = koji.nvr_to_rpms(envr, src=False)
-            old_rpms = koji.nvr_to_rpms(lastBuild['nvr'], src=False)
-            # and match the RPMs according to build name and architecture as
-            # (old one, new one)
-            rpm_to_match = []
-
-            # traverse all RPMs in the newer build
-            for new_rpm in new_rpms:
-                # find the matching older RPM, there should be exactly one
-                old_rpm = [r for r in old_rpms if r['arch'] == new_rpm['arch']
-                            and r['name'] == new_rpm['name']]
-                # we should mark oldere these packages as already matched
-                for o in old_rpm:
-                    o['seen'] = True
-                # older RPM may not exist
-                if not old_rpm:
-                    err = 'N: There is no previous RPM for recent NVRA: %s.%s' % \
-                            (new_rpm['nvr'], new_rpm['arch'])
-                    output.append(err)
-                    print err
-                    out.write(err + '\n')
-                    self.highlights += "%s\n" % err
-                    continue
-                # there certainly should be more than 1 matching older RPM, but
-                # let's check as a precaution
-                if len(old_rpm) > 1:
-                    err = 'N: There are %d existing older RPMs for NVRA: %s.%s' % \
-                            (len(old_rpm), new_rpm['nvr'], new_rpm['arch'])
-                    print err
-                    output.append(err)
-                    out.write(err + '\n')
-                    self.highlights += "%s\n" % err
-                    continue
-
-                # now there is certainly one older RPM matching the newer one
-                old_rpm = old_rpm[0]
-                rpm_to_match.append((old_rpm, new_rpm))
-
-            # there may be some older RPMs not longer available in the newer build
-            for old_rpm in old_rpms:
-                if not old_rpm.has_key('seen'):
-                    err = 'N: There is no new RPM to match older NVRA: \
-    %s.%s' % (old_rpm['nvr'], old_rpm['arch'])
-                    print err
-                    self.highlights += "%s\n" % err
-                    out.write(err + '\n')
-
-            # let's remember if any of invoked command failed
-            #test_result = 'PASSED'
-
-            # let's compare the packages finally
-            for old_rpm, new_rpm in rpm_to_match:
-                # fetch old and new rpms to rpmdir
-                url = old_rpm['url']
-                msg = "Grabbing %s" % url
-                print msg
-                old_file = os.path.join(self.rpmdir, os.path.basename(url))
-                autoqa.util.grabber.urlgrab(url, old_file)
-                url = new_rpm['url']
-                msg = "Grabbing %s" % url
-                print msg
-                new_file = os.path.join(self.rpmdir, os.path.basename(url))
-                autoqa.util.grabber.urlgrab(url, new_file)
-
-                # run rpmguard
-                info = 'N: Comparing %s.%s and %s.%s ...' % \
-                (old_rpm['nvr'], old_rpm['arch'], new_rpm['nvr'], new_rpm['arch'])
-                print info
-                output.append(info)
-                cmd = '%s %s %s' % (self.rpmguard, old_file, new_file)
-                result = utils.run(cmd+" 2>&1", ignore_status=True,
-                    stdout_tee=utils.TEE_TO_LOGS, stderr_tee=utils.TEE_TO_LOGS)
-                # if no warnings or some warnings
-                if result.exit_status == 0 or result.exit_status == 10:
-                    # store the results in temporary arch_results
-                    key = (old_rpm['nvr'],new_rpm['nvr'])
-                    if not self.arch_results.has_key(key):
-                        self.arch_results[key] = {}
-                    archs = self.arch_results[key]
-                    archs[old_rpm['arch']] = result.stdout
-                # if command failed
-                else:
-                    test_result = 'FAILED'
-                    failed_packages.add(envr)
-                    self.results.append(info + '\n' + result.stdout + '\n' + \
-                        result.stderr + '\nN: ----')
-                msg = 'N: ----'
-                print msg
-                output.append(msg)
-
-            # Copy arch_results to results and eliminate duplicates
-            for (old_nvr, new_nvr) in self.arch_results.keys():
-                res_arch = {}
-                for arch, result in \
-                self.arch_results[(old_nvr, new_nvr)].iteritems():
-                    if not res_arch.has_key(result):
-                        res_arch[result] = [arch]
-                    else:
-                        res_arch[result].append(arch)
-                for result, archs in res_arch.iteritems():
-                    archlist = ''
-                    for arch in archs:
-                        archlist += arch + ', '
-                    archlist = archlist[:-2]
-                    info = 'N: Comparing %s and %s (archs: %s) ...\n' % \
-                        (old_nvr, new_nvr, archlist)
-                    outro = 'N: ----'
-                    self.results.append(info + result + outro)
-
-            # Write a copy of the output into the log
-            log += '\n'.join(self.results) + '\n'
-            out.write(log)
-
-            warning_count = len([line for line in log.splitlines() if line.startswith('W: ')])
+            self.outputs.append(msg)
+            # run the test for this envr
+            (result, highlights, outputs, warn_count) = self.test_envr(envr, kojitag)
+            # collect output
+            self.envr_results[envr] = result
+            if self.result_order.index(result) > self.result_order.index(self.result):
+                self.result = result
+            if highlights:
+                self.highlights.append(highlights)
+            self.outputs.append(outputs)
+            # add footer
+            msg = 'N: ----'
+            print msg
+            if (self.outputs and self.outputs[-1].splitlines() and
+                self.outputs[-1].splitlines()[-1] != msg):
+                self.outputs.append(msg)
+            # add one empty line
+            msg = ''
+            print msg
+            self.outputs.append(msg)
 
             # email results to mailing list and to pkg owner if they optin
             repo = repoinfo.getrepo_by_tag(kojitag)
-            #if repo is not None and autoqa.util.check_opt_in(name, repo['collection_name']):
+            pkg_name = rpmUtils.miscutils.splitFilename(envr + '.noarch')[0]
+            if repo is not None and autoqa.util.check_opt_in(pkg_name, repo['collection_name']):
                 #FIXME - hardcoded partial email address here - obviously sub-par
-                #self.mail_to.append('%s-owner at fedoraproject.org' % name)
+                self.mail_to.append('%s-owner at fedoraproject.org' % pkg_name)
+
+        # create self.summary
+        if kwargs['hook'] == 'post-koji-build':
+            if result != 'ABORTED':
+                self.summary = '%d warnings for %s' % (warn_count, update_id)
+            else:
+                self.summary = update_id
+        elif kwargs['hook'] == 'post-bodhi-update':
+            # create result line like "1 PASSED, 2 FAILED, 3 INFO"
+            result_count = []
+            for res in self.result_order:
+                if res in self.envr_results.values():
+                    count = len([k for k in self.envr_results.keys() if self.envr_results[k] == res])
+                    result_count.append('%d %s' % (count, res))
+            result_count = ', '.join(result_count)
+            self.summary = '%s for %s' % (result_count, update_id)
+
+        # reformat result variables
+        self.outputs = '\n'.join(self.outputs)
+        self.highlights = '\n'.join(self.highlights)
+
+        # log outputs
+        self.log.write(self.outputs)
+        self.log.close()
+
+    def test_envr(self, envr, kojitag):
+        '''
+        Test a single ENVR.
+        Returns (result, highlights, outputs, warn_count).
+        '''
+        result = 'PASSED'
+        highlights = []
+        outputs = []
+        warn_count = 0
+
+        def get_result():
+            '''Format the variables as to be returned'''
+            warn_count = len([line for line in outputs if line.startswith('W: ')])
+            return (result, '\n'.join(highlights), '\n'.join(outputs), warn_count)
 
-        self.summary = '%u warnings for package %s' % (warning_count, envr)
-        if kwargs['hook']  == 'post-bodhi-update':
-            self.summary = '%d OK, %d FAILED' % (len(envrs) - len(failed_packages), len(failed_packages))
+        # temporary results for each architecture
+        arch_results = {}
+        koji = autoqa.koji_utils.SimpleKojiClientSession()
 
-        for i in output:
-            self.outputs += i + '\n'
-        self.outputs += log + '\n' + '---' + '\n' + self.summary + '\n' + '---'
-        self.result = test_result
+        # get the most recent release available
+        # add .noarch to parse filename correctly
+        nvrea = rpmUtils.miscutils.splitFilename(envr + '.noarch')
+        name = nvrea[0]
+        lastBuild = koji.list_previous_release(name, kojitag,
+            max_evr=(nvrea[3], nvrea[1], nvrea[2]))
 
-        out.close()
-        # if some command failed, make this test fail
-        if test_result == 'FAILED':
-            raise error.TestFail
+        # if there is no such build, we don't have anything to compare
+        if not lastBuild:
+            msg = "N: There is no previous build of %s in %s tag (or its parents)." % (envr, kojitag)
+            print msg
+            outputs.append(msg)
+            return get_result()
+
+        # now we need list of RPMs available for each build
+        new_rpms = koji.nvr_to_rpms(envr, src=False)
+        old_rpms = koji.nvr_to_rpms(lastBuild['nvr'], src=False)
+        # and match the RPMs according to build name and architecture as
+        # (old one, new one)
+        rpm_to_match = []
+
+        # traverse all RPMs in the newer build
+        for new_rpm in new_rpms:
+            # find the matching older RPM, there should be exactly one
+            old_rpm = [r for r in old_rpms if r['arch'] == new_rpm['arch']
+                        and r['name'] == new_rpm['name']]
+            # we should mark all these packages as already matched
+            for o in old_rpm:
+                assert not o.has_key('seen'), 'The foreign object already has the same attribute as we use'
+                o['seen'] = True
+            # older RPM may not exist
+            if not old_rpm:
+                msg = 'N: There is no previous RPM for %s.%s' % (new_rpm['nvr'], new_rpm['arch'])
+                print msg
+                outputs.append(msg)
+                continue
+            # there certainly shouldn't be more than 1 matching older RPM, if there is,
+            # we have a bug somewhere
+            assert len(old_rpm) <= 1, 'There are %d existing older RPMs for %s.%s: %s' % \
+                (len(old_rpm), new_rpm['nvr'], new_rpm['arch'], old_rpm)
+
+            # now there is certainly one older RPM matching the newer one
+            old_rpm = old_rpm[0]
+            rpm_to_match.append((old_rpm, new_rpm))
+
+        # there may be some older RPMs not longer available in the newer build
+        for old_rpm in old_rpms:
+            if not old_rpm.has_key('seen'):
+                msg = 'N: There is no new RPM to match %s.%s' % (old_rpm['nvr'], old_rpm['arch'])
+                print msg
+                outputs.append(msg)
+
+        # let's compare the packages finally
+        for old_rpm, new_rpm in rpm_to_match:
+            # fetch old and new rpms to rpmdir
+            url = old_rpm['url']
+            print "Grabbing %s" % url
+            old_file = os.path.join(self.rpmdir, os.path.basename(url))
+            autoqa.util.grabber.urlgrab(url, old_file)
+            url = new_rpm['url']
+            print "Grabbing %s" % url
+            new_file = os.path.join(self.rpmdir, os.path.basename(url))
+            autoqa.util.grabber.urlgrab(url, new_file)
+
+            # run rpmguard
+            cmd = '%s %s %s' % (self.rpmguard, old_file, new_file)
+            cmd_result = utils.run(cmd+" 2>&1", ignore_status=True,
+                stdout_tee=utils.TEE_TO_LOGS, stderr_tee=utils.TEE_TO_LOGS)
+            # store the results in temporary arch_results
+            key = (old_rpm['nvr'],new_rpm['nvr'])
+            if not arch_results.has_key(key):
+                arch_results[key] = {}
+            archs = arch_results[key]
+            archs[old_rpm['arch']] = cmd_result.stdout
+            # check command return code
+            if cmd_result.exit_status == 0:
+                result = 'PASSED'
+            elif cmd_result.exit_status == 10:
+                result = 'INFO'
+            else:
+                result = 'ABORTED'
+
+        # Copy arch_results to results and eliminate duplicates
+        for (old_nvr, new_nvr) in arch_results.keys():
+            res_arch = {}
+            for arch, res in arch_results[(old_nvr, new_nvr)].iteritems():
+                if not res_arch.has_key(res):
+                    res_arch[res] = [arch]
+                else:
+                    res_arch[res].append(arch)
+            for res, archs in res_arch.iteritems():
+                archlist = ''
+                for arch in archs:
+                    archlist += arch + ', '
+                archlist = archlist[:-2]
+                outputs.append('N: Comparing %s and %s (archs: %s) ...' % \
+                    (old_nvr, new_nvr, archlist))
+                if res:
+                    outputs.append(res.rstrip()) #strip newline at the end
+                outputs.append('N: ----')
+
+        return get_result()
diff --git a/tests/rpmlint/rpmlint.py b/tests/rpmlint/rpmlint.py
index b3a53e6..9b36e5c 100644
--- a/tests/rpmlint/rpmlint.py
+++ b/tests/rpmlint/rpmlint.py
@@ -18,9 +18,11 @@
 # Author: Will Woods <wwoods at redhat.com>
 
 from autotest_lib.client.bin import utils
-from autotest_lib.client.common_lib import error
 import autoqa.koji_utils, autoqa.util
 import os
+import re
+import fnmatch
+import rpmUtils.miscutils
 from autoqa.repoinfo import repoinfo
 from autoqa.test import AutoQATest
 from autoqa.decorators import ExceptionCatcher
@@ -31,76 +33,117 @@ class rpmlint(AutoQATest):
     def setup(self):
         utils.system('yum -y install rpmlint')
 
-    @ExceptionCatcher("self.run_once_failed")
-    def run_once(self, name, kojitag, **kwargs):
-        failed_packages = set()
-        koji = autoqa.koji_utils.SimpleKojiClientSession()
-        rpmdir = os.path.join(self.tmpdir, 'rpms')
-        os.makedirs(rpmdir)
-        fin_result = 'PASS'
-        self.result = 'PASSED'
-        log = []
+    @ExceptionCatcher("self.initialize_failed")
+    def initialize(self, config, **kwargs):
+        super(rpmlint, self).initialize(config)
+        self.rpmdir = os.path.join(self.tmpdir, 'rpms')
+        os.makedirs(self.rpmdir)
 
+    @ExceptionCatcher("self.run_once_failed")
+    def run_once(self, kojitag, **kwargs):
         if kwargs['hook'] == 'post-koji-build':
             envrs = [kwargs['envr']]
+            update_id = kwargs['envr']
         elif kwargs['hook'] == 'post-bodhi-update':
             envrs = kwargs['envrs']
+            update_id = kwargs['name'] or kwargs['id']
+
+        self.result = 'PASSED'
+        # order for evaluation of final result; higher index means preference
+        self.result_order = ('PASSED','INFO','FAILED','ABORTED')
+        self.envr_results = {} # results for invidual packages
+        self.outputs = []
+        self.highlights = []
+        self.log = open(os.path.join(self.resultsdir,'rpmlint.log'),'wb') # where to log output
+
+        koji = autoqa.koji_utils.SimpleKojiClientSession()
 
         for envr in envrs:
-            msg = 40*'=' + '\n' + envr + '\n' + 40*'='
-            log.append(msg)
+            # add header
+            msg = '%s\n%s\n%s' % ('='*40, envr, '='*40)
             print msg
+            self.outputs.append(msg)
+
+            # erase old rpm packages from self.rpmdir, needed to cleanup from last run
+            rpms = fnmatch.filter(os.listdir(self.rpmdir), '*.rpm')
+            print 'Removing all RPMs from %s' % self.rpmdir
+            for rpm in rpms:
+                os.remove(os.path.join(self.rpmdir, rpm))
+
+            # download packages
             pkgurls = koji.nvr_to_urls(envr)
-            print "Saving RPMs to %s" % rpmdir
+            print "Saving RPMs to %s" % self.rpmdir
             for p in pkgurls:
-                # fetch package to rpmdir
                 print "Grabbing %s" % p
-                localfile = os.path.join(rpmdir, os.path.basename(p))
+                localfile = os.path.join(self.rpmdir, os.path.basename(p))
                 autoqa.util.grabber.urlgrab(p, localfile)
-            cmd = 'rpmlint %s' % rpmdir
 
-            result = 'PASS'
-            try:
-                log.append(utils.system_output(cmd+" 2>&1", retain_output=True))
-            except error.CmdError, e:
-                result = 'FAIL'
-                log.append(e.result_obj.stdout)
+            # run rpmlint
+            cmd = 'rpmlint %s' % self.rpmdir
+            cmd_result = utils.run(cmd+" 2>&1", ignore_status=True,
+                stdout_tee=utils.TEE_TO_LOGS, stderr_tee=utils.TEE_TO_LOGS)
+            outputs = cmd_result.stdout
+
+            # check command return code
+            if cmd_result.exit_status == 0:
+                result = 'PASSED'
+            elif cmd_result.exit_status == 64 or cmd_result.exit_status == 66:
+                result = 'FAILED'
+            else:
+                result = 'ABORTED'
 
             # TODO filter known/waived errors/warnings
             # - Need to store the filters somewhere authoritative, get signoff, etc
 
-            if result == 'FAIL':
-                failed_packages.add(envr)
-                self.result = "FAILED"
-                fin_result = 'FAIL'
+            # extract rpmlint summary line, e.g.:
+            # 3 packages and 0 specfiles checked; 9 errors, 80 warnings.
+            if result != 'ABORTED':
+                match = re.match(r'.*?; ((\d+) errors, (\d+) warnings).', outputs.splitlines()[-1])
+                rpmlint_summary = match.group(1)
+                warn_count = match.group(3)
+                # mark as INFO if some warnings present
+                if result == 'PASSED' and int(warn_count) > 0:
+                    result = 'INFO'
+
+            # collect output
+            self.envr_results[envr] = result
+            if self.result_order.index(result) > self.result_order.index(self.result):
+                self.result = result
+            self.outputs.append(outputs)
+
+            # add one empty line
+            msg = ''
+            print msg
+            self.outputs.append(msg)
 
             # email results to mailing list and to pkg owner if they optin
             repo = repoinfo.getrepo_by_tag(kojitag)
-            #if repo is not None and autoqa.util.check_opt_in(name, repo['collection_name']):
+            pkg_name = rpmUtils.miscutils.splitFilename(envr + '.noarch')[0]
+            if repo is not None and autoqa.util.check_opt_in(pkg_name, repo['collection_name']):
                 #FIXME - hardcoded partial email address here - obviously sub-par
-                #self.mail_to.append('%s-owner at fedoraproject.org' % name)
-
-        for i in log:
-            self.outputs += i + '\n'
-        summary = ((self.outputs.splitlines()[-1]).split(';')[-1])[1:-1]
-
-        # Write a copy of the output into the resultsdir
-        out = open(os.path.join(self.resultsdir,'rpmlint.log'),'wb')
-        out.write(self.outputs)
-        out.close()
-
-        failed = len(failed_packages)
-        passed = len(envrs) - failed
-        msg = 'SUMMARY: %d OK, %d FAILED' % (passed, failed)
-
-        if kwargs['hook'] == 'post-bodhi-update':
-            self.summary = msg
-            self.outputs += '---\n' + msg +'\n---'
-            print msg
-        elif kwargs['hook'] == 'post-koji-build':
-            self.summary = '%s for package %s' % (summary, envr)
+                self.mail_to.append('%s-owner at fedoraproject.org' % pkg_name)
 
+        # create self.summary
+        if kwargs['hook'] == 'post-koji-build':
+            if result != 'ABORTED':
+                self.summary = '%s for %s' % (rpmlint_summary, update_id)
+            else:
+                self.summary = update_id
+        elif kwargs['hook'] == 'post-bodhi-update':
+            # create result line like "1 PASSED, 2 FAILED, 3 INFO"
+            result_count = []
+            for res in self.result_order:
+                if res in self.envr_results.values():
+                    count = len([k for k in self.envr_results.keys() if self.envr_results[k] == res])
+                    result_count.append('%d %s' % (count, res))
+            result_count = ', '.join(result_count)
+            self.summary = '%s for %s' % (result_count, update_id)
+
+        # reformat result variables
+        self.outputs = '\n'.join(self.outputs)
+        self.highlights = '\n'.join(self.highlights)
+
+        # log outputs
+        self.log.write(self.outputs)
+        self.log.close()
 
-        # if some command failed, make this test fail
-        if fin_result == 'FAIL':
-            raise error.TestFail
-- 
1.7.2.2



More information about the autoqa-devel mailing list