From 1e8dc398e99818883036aeff49fbe747b47fa68c Mon Sep 17 00:00:00 2001 From: Celso Providelo Date: Wed, 4 Feb 2015 08:54:00 -0500 Subject: [PATCH] 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 | 187 +++++++++++++++++++++++++---------------- britney.py | 30 ++++++- tests/test_boottest.py | 64 ++++++++++---- 3 files changed, 189 insertions(+), 92 deletions(-) diff --git a/boottest.py b/boottest.py index 93caf60..79c212a 100644 --- a/boottest.py +++ b/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): - """Boottest - Jenkins **glue**. - - Wraps 'boottest/jenkins/boottest-britney' script for: - - * 'check' existing boottest job status ('check ') - * 'submit' new boottest jobs ('submit ') +class BootTest(object): + """Boottest criteria for Britney. + 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": 'Pass', + "SKIPPED": 'Skipped', + "FAIL": 'Regression', + "RUNNING": 'Test in progress', + } 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): - """Return the current boottest jenkins job status. + def _read(self): + """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: - status = self._run('check', name, version) - except subprocess.CalledProcessError as err: - status = self._run('submit', name, version) - return status - - -class BootTest(object): - """Boottest criteria for Britney. - - Process (update) excuses for the 'boottest' criteria. Request and monitor - boottest attempts (see `BootTestJenkinsJob`) for binaries present in the - phone image manifest (see `TouchManifest`). - """ - VALID_STATUSES = ('PASS', 'SKIPPED') + self.pkglist = defaultdict(dict) + if not os.path.exists(self._result_path): + return + with open(self._result_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 + (src, ver, status) = linebits[:3] + if not (src in self.pkglist and ver in self.pkglist[src]): + self.pkglist[src][ver] = status - EXCUSE_LABELS = { - "PASS": 'Pass', - "SKIPPED": 'Skipped', - "FAIL": 'Regression', - "RUNNING": 'Test in progress', - } + def get_status(self, name, version): + """Return test status for the given source name and version.""" + return self.pkglist[name][version] + + 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): - self.britney = britney - self.distribution = distribution - 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 submit(self): + """Submits the current boottests requests for processing.""" + self._run("submit", self._request_path) - def update(self, excuse): - """Return the boottest status for the given excuse. + def collect(self): + """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 - yet processed, otherwise the status of the corresponding job will - be returned. + def needs_test(self, name, version): + """Whether or not the given source and version should be tested. 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: - return None - + if name not in unstable_sources: + return False # Binaries are a seq of "/" 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 ] - - # 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 + return bool(phone_binaries) diff --git a/britney.py b/britney.py index 792c6f7..eef23fe 100755 --- a/britney.py +++ b/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. - status = boottest.update(excuse) + # Allows hints to skip boottest attempts + 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: diff --git a/tests/test_boottest.py b/tests/test_boottest.py index b268c1a..e9fda90 100644 --- a/tests/test_boottest.py +++ b/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(): - if args.name != 'pyqt5-src': - sys.exit(100) - if args.version == '1.1~beta': - print 'PASS' - else: - print 'FAIL' +def collect(): + with open(args.output, 'w') as f: + f.write(template) 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('version') +psubmit.add_argument('input') psubmit.set_defaults(func=submit) -pcheck = sp.add_parser('check') -pcheck.add_argument('name') -pcheck.add_argument('version') -pcheck.set_defaults(func=check) + +prequest = sp.add_parser('request') +prequest.add_argument('-O', dest='output') +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<', + '
  • boottest skipped from hints by cjwatson', + '
  • 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', '
  • 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 to .*>1.1~beta<',