Buildbot BuildSlave for testcode2

testcode_buildbot.py

"""
Buildbot BuildStep that runs and parses the output of testcode2.

Copyright (c) 2013 Shawn Chin
This sofware is released under the BSD 3-Clause License
"""
import re
from itertools import chain
from collections import defaultdict
from buildbot.steps.shell import ShellCommand
from buildbot.status.results import SUCCESS, WARNINGS, FAILURE


class TestCodeStep(ShellCommand):
    """
    BuildStep for running and parsing the output of testcode2. 
    """
    name = "testcode"
    description = "running testcode"
    descriptionDone = "testcode"
    
    def __init__(self, testcode="testcode.py", test_exe=None, action="run",
                 category=None, mpi=False, num_procs=1, **kwargs):
        """
        @param testcode: testcode.py executable (path relative to workdir)
        @param test_exe: test executable (path relative to workdir)
        @param action: testcode action
        @param action: testcode category
        @param mpi: is parallel run?
        @num_procs: number of mpi procs if parallel run, number of concurrent
                    tests if serial run
        """
        self.testcode = testcode
        self.test_exe = test_exe
        self.category = category
        self.mpi = mpi

        if action == "run" and not test_exe:
            raise Exception("test_exe must be defined for 'run'")
        self.action = action
            
        try:
            self.num_procs = int(num_procs)
            if self.num_procs < 1:
                raise Exception()
        except:
            raise Exception("invalid value for num_procs")

        # insert generated command into "command" arg
        kwargs["command"] = self._get_command()
        # request that the called python script does not buffer
        kwargs.setdefault("env", {})["PYTHONUNBUFFERED"] = "x"

        # Let the baseclass do its work
        ShellCommand.__init__(self, **kwargs)

    def _get_command(self):
        """Returns command as a list"""
        cmd = [self.testcode, '--verbose']

        if self.test_exe:
            cmd.append("--executable=%s" % self.test_exe)
        if self.category:
            cmd.append("--category=%s" % self.category)
        if self.action == "run":
            if self.mpi:
                cmd.append("--processors=%d" % self.num_procs)
            else:
                cmd.append("--total-processors=%d" % self.num_procs)
                
        cmd.append(self.action)
        return cmd

    def _create_log(self, events, reports, status, label):
        if not events[status]:
            return

        header = ["The following tests resulted in %s:" % label]
        items = (" - %s (param: %s)" % x for x in events[status])
        
        blank = [""]
        summary = chain(header, items, blank)
        out = chain(summary, ["Details:", "--------"], reports[status])
        
        self.addCompleteLog("%s (%d)" % (label, len(events[status])),
                            "\n".join(out))
        
    def createSummary(self, log):
        events = defaultdict(list)
        reports = defaultdict(list)

        stopword = "All done."   # marker for end of test results
        re_result = re.compile("""
        (?P<test>[^\s]+)
        \s-\s
        (?P<param>[^\s]+)
        :\s
        (?P<status>Unknown|Passed|SKIPPED|WARNING|\*\*FAILED\*\*)\.
        """, re.VERBOSE)

        # statuses that may be followed by text we want to keep
        interesting_status = ("**FAILED**", "WARNING")
        more_expected = False   # control flags

        comparisons_skipped = 0
        
        for line in log.getText().split("\n"):
            if not line:
                continue
            if line.startswith("Skipping comparison"):
                comparisons_skipped += 1
                
            m = re_result.match(line)
            if not m and more_expected and not line.startswith(stopword):
                reports[more_expected].append(line)
            elif m:
                status = m.group("status")
                if status == "Unknown":
                    status = "WARNING"  # treat "Unknown" as warning
                events[status].append(m.groups()[:2])
                reports[status].append("\n" + line)
                more_expected = (False, status)[status in interesting_status]

        # store counts
        self.warn_count = len(events["WARNING"])
        self.fail_count = len(events["**FAILED**"])
        self.pass_count = len(events["Passed"])
        self.skip_count = len(events["SKIPPED"]) + comparisons_skipped
        
        # build log files
        self._create_log(events, reports, "**FAILED**", "failures")
        self._create_log(events, reports, "WARNING", "warnings")

    def getText(self, cmd, results):
        words = ["test"]
        words.append("Passed %d of %d" % (self.pass_count,
                        self.pass_count + self.warn_count + self.fail_count))
        if self.skip_count:
            words.append("(skipped %d)" % self.skip_count)
        return words
        
    def evaluateCommand(self, cmd):
        if (cmd.didFail() or self.fail_count):
            return FAILURE
        if self.warn_count:
            return WARNINGS
        return SUCCESS