Merge the boottest-jenkins script wrapper into BootTest class for simplicity and extend the API to match what we already for auto-package-testing (ADT).

boottest.py

@@ -13,11 +13,17 @@
 # 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 urllib
 
+import apt_pkg
+
 from consts import BINARIES
 
 
@@ -41,12 +47,23 @@ class TouchManifest(object):
 
     """
 
+    def __init__(self, distribution, series, verbose=False, fetch=True):
+        self.verbose = verbose
+        self.path = "boottest/images/{}/{}/manifest".format(
+            distribution, series)
+        if fetch:
+            self.__fetch_manifest(distribution, series)
+        self._manifest = self._load()
+
     def __fetch_manifest(self, distribution, series):
         url = "http://cdimage.ubuntu.com/{}/daily-preinstalled/" \
               "pending/{}-preinstalled-touch-armhf.manifest".format(
                   distribution, series
         )
-        print("I: [%s] - Fetching manifest from %s" % (time.asctime(), url))
+        if self.verbose:
+            print(
+                "I: [%s] - Fetching manifest from %s" % (
+                    time.asctime(), url))
         response = urllib.urlopen(url)
         # Only [re]create the manifest file if one was successfully downloaded
         # this allows for an existing image to be used if the download fails.
@@ -55,15 +72,6 @@ class TouchManifest(object):
             with open(self.path, 'w') as fp:
                 fp.write(response.read())
 
-    def __init__(self, distribution, series, fetch=True):
-        self.path = "boottest/images/{}/{}/manifest".format(
-            distribution, series)
-
-        if fetch:
-            self.__fetch_manifest(distribution, series)
-
-        self._manifest = self._load()
-
     def _load(self):
         pkg_list = []
 
@@ -87,104 +95,141 @@ class TouchManifest(object):
         return key in self._manifest
 
 
-class BootTestJenkinsJob(object):
+class BootTest(object):
-    """Boottest - Jenkins **glue**.
+    """Boottest criteria for Britney.
-
-    Wraps 'boottest/jenkins/boottest-britney' script for:
-
-    * 'check' existing boottest job status ('check <source> <version>')
-    * 'submit' new boottest jobs ('submit <source> <version>')
 
+    This class provides an API for handling the boottest-jenkins
+    integration layer (mostly derived from auto-package-testing/adt):
     """
+    VALID_STATUSES = ('PASS', 'SKIPPED')
+
+    EXCUSE_LABELS = {
+        "PASS": '<span style="background:#87d96c">Pass</span>',
+        "SKIPPED": '<span style="background:#e5c545">Skipped</span>',
+        "FAIL": '<span style="background:#ff6666">Regression</span>',
+        "RUNNING": '<span style="background:#99ddff">Test in progress</span>',
+    }
 
     script_path = "boottest/jenkins/boottest-britney"
 
-    def __init__(self, distribution, series):
+    def __init__(self, britney, distribution, series, debug=False):
+        self.britney = britney
         self.distribution = distribution
         self.series = series
+        self.debug = debug
+        self.rc_path = None
+        self._read()
+        manifest_fetch = getattr(
+            self.britney.options, "boottest_fetch", "no") == "yes"
+        self.phone_manifest = TouchManifest(
+            self.distribution, self.series, fetch=manifest_fetch,
+            verbose=self.britney.options.verbose)
+
+    @property
+    def _request_path(self):
+        return "boottest/work/adt.request.%s" % self.series
+
+    @property
+    def _result_path(self):
+        return "boottest/work/adt.result.%s" % self.series
+
+    def _ensure_rc_file(self):
+        if self.rc_path:
+            return
+        self.rc_path = os.path.abspath("boottest/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-armhf/
+                apturi: file:%s/mirror/%s
+                components: main restricted universe multiverse
+                rsync_host: rsync://tachash.ubuntu-ci/adt/
+                datadir: ~/proposed-migration/boottest/data""" %
+                         (self.series, self.series, home, self.distribution)),
+                         file=rc_file)
 
     def _run(self, *args):
+        self._ensure_rc_file()
         if not os.path.exists(self.script_path):
             print("E: [%s] - Boottest/Jenking glue script missing: %s" % (
                 time.asctime(), self.script_path))
             return '-'
         command = [
             self.script_path,
+            "-c", self.rc_path,
             "-d", self.distribution, "-s", self.series,
             ]
         command.extend(args)
         return subprocess.check_output(command).strip()
 
-    def get_status(self, name, version):
+    def _read(self):
-        """Return the current boottest jenkins job status.
+        """Loads a list of results (sources tests and their status).
 
-        Request a boottest attempt if it's new.
+        Provides internal data for `get_status()`.
         """
-        try:
+        self.pkglist = defaultdict(dict)
-            status = self._run('check', name, version)
+        if not os.path.exists(self._result_path):
-        except subprocess.CalledProcessError as err:
+            return
-            status = self._run('submit', name, version)
+        with open(self._result_path) as f:
-        return status
+            for line in f:
-
+                line = line.strip()
-
+                if line.startswith("Suite:") or line.startswith("Date:"):
-class BootTest(object):
+                    continue
-    """Boottest criteria for Britney.
+                linebits = line.split()
-
+                if len(linebits) < 2:
-    Process (update) excuses for the 'boottest' criteria. Request and monitor
+                    print("W: Invalid line format: '%s', skipped" % line)
-    boottest attempts (see `BootTestJenkinsJob`) for binaries present in the
+                    continue
-    phone image manifest (see `TouchManifest`).
+                (src, ver, status) = linebits[:3]
-    """
+                if not (src in self.pkglist and ver in self.pkglist[src]):
-    VALID_STATUSES = ('PASS', 'SKIPPED')
+                    self.pkglist[src][ver] = status
 
-    EXCUSE_LABELS = {
+    def get_status(self, name, version):
-        "PASS": '<span style="background:#87d96c">Pass</span>',
+        """Return test status for the given source name and version."""
-        "SKIPPED": '<span style="background:#e5c545">Skipped</span>',
+        return self.pkglist[name][version]
-        "FAIL": '<span style="background:#ff6666">Regression</span>',
+
-        "RUNNING": '<span style="background:#99ddff">Test in progress</span>',
+    def request(self, packages):
-    }
+        """Requests boottests for the given sources list ([(src, ver),])."""
+        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)
+                # Update 'pkglist' so even if submit/collect is not called
+                # (dry-run), britney has some results.
+                self.pkglist[src][ver] = 'RUNNING'
+            request_file.flush()
+            self._run("request", "-O", request_path, request_file.name)
 
-    def __init__(self, britney, distribution, series, debug=False):
+    def submit(self):
-        self.britney = britney
+        """Submits the current boottests requests for processing."""
-        self.distribution = distribution
+        self._run("submit", self._request_path)
-        self.series = series
-        self.debug = debug
-        manifest_fetch = getattr(
-            self.britney.options, "boottest_fetch", "no") == "yes"
-        self.phone_manifest = TouchManifest(
-            self.distribution, self.series, fetch=manifest_fetch)
-        self.dispatcher = BootTestJenkinsJob(self.distribution, self.series)
 
-    def update(self, excuse):
+    def collect(self):
-        """Return the boottest status for the given excuse.
+        """Collects boottests results and updates internal registry."""
+        self._run("collect", "-O", self._result_path)
+        self._read()
 
-        A new boottest job will be requested if the the source was not
+    def needs_test(self, name, version):
-        yet processed, otherwise the status of the corresponding job will
+        """Whether or not the given source and version should be tested.
-        be returned.
 
         Sources are only considered for boottesting if they produce binaries
         that are part of the phone image manifest. See `TouchManifest`.
         """
         # Discover all binaries for the 'excused' source.
         unstable_sources = self.britney.sources['unstable']
-
         # Dismiss if source is not yet recognized (??).
-        if excuse.name not in unstable_sources:
+        if name not in unstable_sources:
-            return None
+            return False
-
         # Binaries are a seq of "<binname>/<arch>" and, practically, boottest
         # is only concerned about armhf binaries mentioned in the phone
         # manifest. Anything else should be skipped.
         phone_binaries = [
-            b for b in unstable_sources[excuse.name][BINARIES]
+            b for b in unstable_sources[name][BINARIES]
             if b.split('/')[1] in self.britney.options.boottest_arches.split()
             and b.split('/')[0] in self.phone_manifest
         ]
-
+        return bool(phone_binaries)
-        # Process (request or update) a boottest attempt for the source
-        # if one or more of its binaries are part of the phone image.
-        if phone_binaries:
-            status = self.dispatcher.get_status(excuse.name, excuse.ver[1])
-        else:
-            status = 'SKIPPED'
-
-        return status

britney.py

@@ -1908,6 +1908,7 @@ class Britney(object):
             boottest = BootTest(
                 self, self.options.distribution, self.options.series,
                 debug=boottest_debug)
+            boottest_excuses = []
             for excuse in self.excuses:
                 # Skip already invalid excuses.
                 if not excuse.run_boottest:
@@ -1919,15 +1920,38 @@ class Britney(object):
                     "_" in excuse.name or
                     excuse.ver[1] == "-"):
                     continue
-                # Update valid excuses from the boottest context.
+                # Allows hints to skip boottest attempts
-                status = boottest.update(excuse)
+                hints = self.hints.search(
+                    'force-skiptest', package=excuse.name)
+                forces = [x for x in hints
+                          if same_source(excuse.ver[1], x.version)]
+                if forces:
+                    excuse.addhtml(
+                        "boottest skipped from hints by %s" % forces[0].user)
+                    continue
+                # Only sources whitelisted in the boottest context should
+                # be tested (currently only sources building phone binaries).
+                if not boottest.needs_test(excuse.name, excuse.ver[1]):
+                    label = BootTest.EXCUSE_LABELS.get('SKIPPED')
+                    excuse.addhtml("Boottest result: %s" % (label))
+                    continue
+                # Okay, aggregate required boottests requests.
+                boottest_excuses.append(excuse)
+            boottest.request([(e.name, e.ver[1]) for e in boottest_excuses])
+            # Dry-run avoids data exchange with external systems.
+            if not self.options.dry_run:
+                boottest.submit()
+                boottest.collect()
+            # Update excuses from the boottest context.
+            for excuse in boottest_excuses:
+                status = boottest.get_status(excuse.name, excuse.ver[1])
                 label = BootTest.EXCUSE_LABELS.get(status, 'UNKNOWN STATUS')
                 excuse.addhtml("Boottest result: %s" % (label))
                 # Allows hints to force boottest failures/attempts
                 # to be ignored.
                 hints = self.hints.search('force', package=excuse.name)
                 hints.extend(
-                    self.hints.search('force-skiptest', package=excuse.name))
+                    self.hints.search('force-badtest', package=excuse.name))
                 forces = [x for x in hints
                           if same_source(excuse.ver[1], x.version)]
                 if forces:

tests/test_boottest.py

@@ -20,7 +20,6 @@ sys.path.insert(0, PROJECT_DIR)
 from tests import TestBase
 from boottest import (
     BootTest,
-    BootTestJenkinsJob,
     TouchManifest,
 )
 
@@ -166,31 +165,45 @@ class TestBoottestEnd2End(TestBase):
         with open(script_path, 'w') as f:
             f.write('''#!%(py)s
 import argparse
+import os
 import sys
 
+template = """
+green 1.1~beta RUNNING green 1.1~beta
+pyqt5-src 1.1~beta PASS pyqt5-src 1.1~beta
+pyqt5-src 1.1 FAIL pyqt5-src 1.1
+"""
+
+def request():
+    os.makedirs(os.path.dirname(args.output))
+    with open(args.output, 'w') as f:
+        f.write(template)
+
 def submit():
-    print 'RUNNING'
+    with open(args.input, 'w') as f:
+        f.write(template)
 
-def check():
+def collect():
-    if args.name != 'pyqt5-src':
+    with open(args.output, 'w') as f:
-       sys.exit(100)
+        f.write(template)
-    if args.version == '1.1~beta':
-       print 'PASS'
-    else:
-       print 'FAIL'
 
 p = argparse.ArgumentParser()
 p.add_argument('-d')
 p.add_argument('-s')
+p.add_argument('-c')
 sp = p.add_subparsers()
 psubmit = sp.add_parser('submit')
-psubmit.add_argument('name')
+psubmit.add_argument('input')
-psubmit.add_argument('version')
 psubmit.set_defaults(func=submit)
-pcheck = sp.add_parser('check')
+
-pcheck.add_argument('name')
+prequest = sp.add_parser('request')
-pcheck.add_argument('version')
+prequest.add_argument('-O', dest='output')
-pcheck.set_defaults(func=check)
+prequest.add_argument('input')
+prequest.set_defaults(func=request)
+
+pcollect = sp.add_parser('collect')
+pcollect.add_argument('-O', dest='output')
+pcollect.set_defaults(func=collect)
 
 args = p.parse_args()
 args.func()
@@ -268,9 +281,24 @@ args.func()
         with open(hints_path, 'w') as fd:
             fd.write(content)
 
+    def test_skipped_by_hints(self):
+        # `Britney` allows boottests to be skipped by hinting the
+        # corresponding source with 'force-skiptest'. The boottest
+        # attempt will not be requested.
+        context = [
+            ('pyqt5', {'Source': 'pyqt5-src', 'Version': '1.1',
+                       'Architecture': 'all'}),
+        ]
+        self.create_hint('cjwatson', 'force-skiptest pyqt5-src/1.1')
+        self.do_test(
+            context,
+            [r'\bpyqt5-src\b.*\(- to .*>1.1<',
+             '<li>boottest skipped from hints by cjwatson',
+             '<li>Valid candidate'])
+
     def test_fail_but_forced_by_hints(self):
         # `Britney` allows boottests results to be ignored by hinting the
-        # corresponding source with 'force' or 'force-skiptest'. The boottest
+        # corresponding source with 'force' or 'force-badtest'. The boottest
         # attempt will still be requested and its results would be considered
         # for other non-forced sources.
         context = [
@@ -287,13 +315,13 @@ args.func()
              'but forced by cjwatson',
              '<li>Valid candidate'])
 
-    def test_fail_but_skipped_by_hints(self):
+    def test_fail_but_ignored_by_hints(self):
         # See `test_fail_but_forced_by_hints`.
         context = [
             ('green', {'Source': 'green', 'Version': '1.1~beta',
                        'Architecture': 'armhf', 'Depends': 'libc6 (>= 0.9)'}),
         ]
-        self.create_hint('cjwatson', 'force-skiptest green/1.1~beta')
+        self.create_hint('cjwatson', 'force-badtest green/1.1~beta')
         self.do_test(
             context,
             [r'\bgreen\b.*>1</a> to .*>1.1~beta<',