britney2-ubuntu/autopkgtest.py

# -*- coding: utf-8 -*-

# Copyright (C) 2013 Canonical Ltd.
# Author: Colin Watson <cjwatson@ubuntu.com>
# Partly based on code in auto-package-testing by
# Jean-Baptiste Lallement <jean-baptiste.lallement@canonical.com>

# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.

# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.

from __future__ import print_function

from collections import defaultdict
from contextlib import closing
import os
import subprocess
import tempfile
from textwrap import dedent
import time
import apt_pkg


adt_britney = os.path.expanduser("~/auto-package-testing/jenkins/adt-britney")

ADT_PASS = ["PASS", "ALWAYSFAIL"]
ADT_EXCUSES_LABELS = {
    "PASS": '<span style="background:#87d96c">Pass</span>',
    "ALWAYSFAIL": '<span style="background:#e5c545">Always failed</span>',
    "REGRESSION": '<span style="background:#ff6666">Regression</span>',
    "RUNNING": '<span style="background:#99ddff">Test in progress</span>',
}


class AutoPackageTest(object):
    """autopkgtest integration

    Look for autopkgtest jobs to run for each update that is otherwise a
    valid candidate, and collect the results.  If an update causes any
    autopkgtest jobs to be run, then they must all pass before the update is
    accepted.
    """

    def __init__(self, britney, distribution, series, debug=False):
        self.britney = britney
        self.distribution = distribution
        self.series = series
        self.debug = debug
        self.read()
        self.rc_path = None

    def _ensure_rc_file(self):
        if self.rc_path:
            return
        self.rc_path = os.path.expanduser(
            "~/proposed-migration/autopkgtest/rc.%s" % self.series)
        with open(self.rc_path, "w") as rc_file:
            home = os.path.expanduser("~")
            print(dedent("""\
                release: %s
                aptroot: ~/.chdist/%s-proposed-amd64/
                apturi: file:%s/mirror/%s
                components: main restricted universe multiverse
                rsync_host: rsync://tachash.ubuntu-ci/adt/
                datadir: ~/proposed-migration/autopkgtest/data""" %
                         (self.series, self.series, home, self.distribution)),
                         file=rc_file)

    @property
    def _request_path(self):
        return os.path.expanduser(
            "~/proposed-migration/autopkgtest/work/adt.request.%s" %
            self.series)

    @property
    def _result_path(self):
        return os.path.expanduser(
            "~/proposed-migration/autopkgtest/work/adt.result.%s" %
            self.series)

    def _parse(self, path):
        if os.path.exists(path):
            with open(path) as f:
                for line in f:
                    line = line.strip()
                    if line.startswith("Suite:") or line.startswith("Date:"):
                        continue
                    linebits = line.split()
                    if len(linebits) < 2:
                        print("W: Invalid line format: '%s', skipped" % line)
                        continue
                    yield linebits

    def read(self):
        '''Loads a list of results

        This function loads a list of results returned by __parse() and builds
        2 lists:
            - a list of source package/version with all the causes that
            triggered a test and the result of the test for this trigger.
            - a list of packages/version that triggered a test with the source
            package/version and result triggered by this package.
        These lists will be used in result() called from britney.py to generate
        excuses and now which uploads passed, caused regression or which tests
        have always been failing
        '''
        self.pkglist = defaultdict(dict)
        self.pkgcauses = defaultdict(lambda: defaultdict(list))
        for linebits in self._parse(self._result_path):
            (src, ver, status) = linebits[:3]

            if not (src in self.pkglist and ver in self.pkglist[src]):
                self.pkglist[src][ver] = {
                    "status": status,
                    "causes": {}
                }

            i = iter(linebits[3:])
            for trigsrc, trigver in zip(i, i):
                self.pkglist[src][ver]['causes'].setdefault(
                    trigsrc, []).append((trigver, status))
                self.pkgcauses[trigsrc][trigver].append((status, src, ver))

    def _adt_britney(self, *args):
        command = [
            adt_britney,
            "-c", self.rc_path, "-r", self.series, "-PU",
            ]
        if self.debug:
            command.append("-d")
        command.extend(args)
        subprocess.check_call(command)

    def request(self, packages, excludes=None):
        if excludes is None:
            excludes = []

        self._ensure_rc_file()
        request_path = self._request_path
        if os.path.exists(request_path):
            os.unlink(request_path)
        with closing(tempfile.NamedTemporaryFile(mode="w")) as request_file:
            for src, ver in packages:
                if src in self.pkglist and ver in self.pkglist[src]:
                    continue
                print("%s %s" % (src, ver), file=request_file)
            request_file.flush()
            self._adt_britney("request", "-O", request_path, request_file.name)

        # Remove packages that have been identified as invalid candidates for
        # testing from the request file i.e run_autopkgtest = False
        with open(request_path, 'r') as request_file:
            lines = request_file.readlines()
        with open(request_path, 'w') as request_file:
            for line in lines:
                src = line.split()[0]
                if src not in excludes:
                    request_file.write(line)
                else:
                    if self.britney.options.verbose:
                        print("I: [%s] - Requested autopkgtest for %s but "
                              "run_autopkgtest set to False" %
                              (time.asctime(), src))

        for linebits in self._parse(request_path):
            # Make sure that there's an entry in pkgcauses for each new
            # request, so that results() gives useful information without
            # relying on the submit/collect cycle.  This improves behaviour
            # in dry-run mode.
            src = linebits.pop(0)
            ver = linebits.pop(0)
            if self.britney.options.verbose:
                print("I: [%s] - Requested autopkgtest for %s_%s (%s)" %
                      (time.asctime(), src, ver, " ".join(linebits)))
            try:
                status = linebits.pop(0).upper()
                while True:
                    trigsrc = linebits.pop(0)
                    trigver = linebits.pop(0)
                    for status, csrc, cver in self.pkgcauses[trigsrc][trigver]:
                        if csrc == trigsrc and cver == trigver:
                            break
                    else:
                        self.pkgcauses[trigsrc][trigver].append(
                            (status, src, ver))
            except IndexError:
                # End of the list
                pass

    def submit(self):
        self._ensure_rc_file()
        request_path = self._request_path
        if os.path.exists(request_path):
            self._adt_britney("submit", request_path)

    def collect(self):
        self._ensure_rc_file()
        result_path = self._result_path
        self._adt_britney("collect", "-O", result_path)
        self.read()
        if self.britney.options.verbose:
            for src in sorted(self.pkglist):
                for ver in sorted(self.pkglist[src],
                                  cmp=apt_pkg.version_compare):
                    for trigsrc in sorted(self.pkglist[src][ver]['causes']):
                        for trigver, status \
                                in self.pkglist[src][ver]['causes'][trigsrc]:
                            print("I: [%s] - Collected autopkgtest status "
                                  "for %s_%s/%s_%s: " "%s" % (
                                      time.asctime(), src, ver, trigsrc,
                                      trigver, status))

    def results(self, trigsrc, trigver):
        for status, src, ver in self.pkgcauses[trigsrc][trigver]:
            # Check for regresssion
            if status == 'FAIL':
                passed_once = False
                for ver in self.pkglist[src]:
                    for trigsrc in self.pkglist[src][ver]['causes']:
                        for trigver, status \
                                in self.pkglist[src][ver]['causes'][trigsrc]:
                            if status == 'PASS':
                                passed_once = True
                if not passed_once:
                    status = 'ALWAYSFAIL'
                else:
                    status = 'REGRESSION'
            yield status, src, ver