diff --git a/boottest.py b/boottest.py new file mode 100644 index 0000000..74e3552 --- /dev/null +++ b/boottest.py @@ -0,0 +1,282 @@ +# -*- coding: utf-8 -*- + +# Copyright (C) 2015 Canonical Ltd. + +# 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 urllib + +import apt_pkg + +from consts import BINARIES + + +FETCH_RETRIES = 3 + + +class TouchManifest(object): + """Parses a corresponding touch image manifest. + + Based on http://cdimage.u.c/ubuntu-touch/daily-preinstalled/pending/vivid-preinstalled-touch-armhf.manifest + + Assumes the deployment is arranged in a way the manifest is available + and fresh on: + + '{britney_cwd}/boottest/images/{distribution}/{series}/manifest' + + Only binary name matters, version is ignored, so callsites can: + + >>> manifest = TouchManifest('ubuntu-touch', 'vivid') + >>> 'webbrowser-app' in manifest + True + >>> 'firefox' in manifest + False + + """ + + def __init__(self, project, series, verbose=False, fetch=True): + self.verbose = verbose + self.path = "boottest/images/{}/{}/manifest".format( + project, series) + success = False + if fetch: + retries = FETCH_RETRIES + success = self.__fetch_manifest(project, series) + + while retries > 0 and not success: + success = self.__fetch_manifest(project, series) + retries -= 1 + if not success: + print("E: [%s] - Unable to fetch manifest: %s %s" % ( + time.asctime(), project, series)) + + self._manifest = self._load() + + def __fetch_manifest(self, project, series): + url = "http://cdimage.ubuntu.com/{}/daily-preinstalled/" \ + "pending/{}-preinstalled-touch-armhf.manifest".format( + project, series + ) + success = False + if self.verbose: + print( + "I: [%s] - Fetching manifest from %s" % ( + time.asctime(), url)) + print("I: [%s] - saving it to %s" % (time.asctime(), self.path)) + try: + response = urllib.urlopen(url) + except IOError as e: + print("W: [%s] - error connecting to %s: %s" % ( + time.asctime(), self.path, e)) + return success # failure + + # Only [re]create the manifest file if one was successfully downloaded + # this allows for an existing image to be used if the download fails. + if response.code == 200: + path_dir = os.path.dirname(self.path) + if not os.path.exists(path_dir): + os.makedirs(path_dir) + with open(self.path, 'w') as fp: + fp.write(response.read()) + success = True + + return success + + def _load(self): + pkg_list = [] + + if not os.path.exists(self.path): + return pkg_list + + with open(self.path) as fd: + for line in fd.readlines(): + # skip headers and metadata + if 'DOCTYPE' in line: + continue + name, version = line.split() + name = name.split(':')[0] + if name == 'click': + continue + pkg_list.append(name) + + return sorted(pkg_list) + + def __contains__(self, key): + return key in self._manifest + + +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',) + + EXCUSE_LABELS = { + "PASS": 'Pass', + "FAIL": 'Regression', + "RUNNING": 'Test in progress', + } + + script_path = os.path.expanduser( + "~/auto-package-testing/jenkins/boottest-britney") + + 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( + 'ubuntu-touch', 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/boottest/ + 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, + "-r", self.series, + "-PU", + ] + if self.debug: + command.append("-d") + command.extend(args) + return subprocess.check_output(command).strip() + + def _read(self): + """Loads a list of results (sources tests and their status). + + Provides internal data for `get_status()`. + """ + 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 + + def get_status(self, name, version): + """Return test status for the given source name and version.""" + try: + return self.pkglist[name][version] + except KeyError: + # This error handling accounts for outdated apt caches, when + # `boottest-britney` erroneously reports results for the + # current source version, instead of the proposed. + # Returning None here will block source promotion with: + # 'UNKNOWN STATUS' excuse. If the jobs are retried and its + # results find an up-to-date cache, the problem is gone. + print("E: [%s] - Missing boottest results for %s_%s" % ( + time.asctime(), name, version)) + return None + + 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 submit(self): + """Submits the current boottests requests for processing.""" + self._run("submit", self._request_path) + + def collect(self): + """Collects boottests results and updates internal registry.""" + self._run("collect", "-O", self._result_path) + self._read() + if not self.britney.options.verbose: + return + for src in sorted(self.pkglist): + for ver in sorted(self.pkglist[src], cmp=apt_pkg.version_compare): + status = self.pkglist[src][ver] + print("I: [%s] - Collected boottest status for %s_%s: " + "%s" % (time.asctime(), src, ver, status)) + + 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 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[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) diff --git a/britney.conf b/britney.conf index 2d8d4a8..8d9b285 100644 --- a/britney.conf +++ b/britney.conf @@ -65,3 +65,8 @@ REMOVE_OBSOLETE = no ADT_ENABLE = yes ADT_DEBUG = no ADT_ARCHES = amd64 i386 + +BOOTTEST_ENABLE = yes +BOOTTEST_DEBUG = yes +BOOTTEST_ARCHES = armhf amd64 +BOOTTEST_FETCH = yes diff --git a/britney.py b/britney.py index 12516af..22ffbbd 100755 --- a/britney.py +++ b/britney.py @@ -226,6 +226,8 @@ from consts import (VERSION, SECTION, BINARIES, MAINTAINER, FAKESRC, SOURCE, SOURCEVER, ARCHITECTURE, DEPENDS, CONFLICTS, PROVIDES, RDEPENDS, RCONFLICTS, MULTIARCH, ESSENTIAL) from autopkgtest import AutoPackageTest, ADT_PASS, ADT_EXCUSES_LABELS +from boottest import BootTest + __author__ = 'Fabio Tranchitella and the Debian Release Team' __version__ = '2.0' @@ -1366,6 +1368,7 @@ class Britney(object): # the starting point is that we will update the candidate and run autopkgtests update_candidate = True run_autopkgtest = True + run_boottest = True # if the version in unstable is older, then stop here with a warning in the excuse and return False if source_t and apt_pkg.version_compare(source_u[VERSION], source_t[VERSION]) < 0: @@ -1379,6 +1382,7 @@ class Britney(object): excuse.addhtml("%s source package doesn't exist" % (src)) update_candidate = False run_autopkgtest = False + run_boottest = False # retrieve the urgency for the upload, ignoring it if this is a NEW package (not present in testing) urgency = self.urgencies.get(src, self.options.default_urgency) @@ -1397,6 +1401,7 @@ class Britney(object): excuse.addreason("remove") update_candidate = False run_autopkgtest = False + run_boottest = False # check if there is a `block' or `block-udeb' hint for this package, or a `block-all source' hint blocked = {} @@ -1476,6 +1481,7 @@ class Britney(object): else: update_candidate = False run_autopkgtest = False + run_boottest = False excuse.addreason("age") if suite in ['pu', 'tpu']: @@ -1511,6 +1517,8 @@ class Britney(object): update_candidate = False if arch in self.options.adt_arches.split(): run_autopkgtest = False + if arch in self.options.boottest_arches.split(): + run_boottest = False excuse.addreason("arch") excuse.addreason("arch-%s" % arch) excuse.addreason("build-arch") @@ -1553,6 +1561,8 @@ class Britney(object): update_candidate = False if arch in self.options.adt_arches.split(): run_autopkgtest = False + if arch in self.options.boottest_arches.split(): + run_boottest = False # if there are out-of-date packages, warn about them in the excuse and set update_candidate # to False to block the update; if the architecture where the package is out-of-date is @@ -1579,6 +1589,8 @@ class Britney(object): update_candidate = False if arch in self.options.adt_arches.split(): run_autopkgtest = False + if arch in self.options.boottest_arches.split(): + run_boottest = False excuse.addreason("arch") excuse.addreason("arch-%s" % arch) if uptodatebins: @@ -1596,11 +1608,13 @@ class Britney(object): excuse.addreason("no-binaries") update_candidate = False run_autopkgtest = False + run_boottest = False elif not built_anywhere: excuse.addhtml("%s has no up-to-date binaries on any arch" % src) excuse.addreason("no-binaries") update_candidate = False run_autopkgtest = False + run_boottest = False # if the suite is unstable, then we have to check the release-critical bug lists before # updating testing; if the unstable package has RC bugs that do not apply to the testing @@ -1633,6 +1647,7 @@ class Britney(object): ["#%s" % (urllib.quote(a), a) for a in new_bugs]))) update_candidate = False run_autopkgtest = False + run_boottest = False excuse.addreason("buggy") if len(old_bugs) > 0: @@ -1651,6 +1666,7 @@ class Britney(object): excuse.force() update_candidate = True run_autopkgtest = True + run_boottest = True # if the package can be updated, it is a valid candidate if update_candidate: @@ -1660,6 +1676,7 @@ class Britney(object): # TODO excuse.addhtml("Not considered") excuse.run_autopkgtest = run_autopkgtest + excuse.run_boottest = run_boottest self.excuses.append(excuse) return update_candidate @@ -1883,6 +1900,88 @@ class Britney(object): e.addreason("autopkgtest") e.is_valid = False + if (getattr(self.options, "boottest_enable", "no") == "yes" and + self.options.series): + # trigger 'boottest'ing for valid candidates. + boottest_debug = getattr( + self.options, "boottest_debug", "no") == "yes" + 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: + continue + # Also skip removals, binary-only candidates, proposed-updates + # and unknown versions. + if (excuse.name.startswith("-") or + "/" in excuse.name or + "_" in excuse.name or + excuse.ver[1] == "-"): + continue + # 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]): + # Silently skipping. + 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() + # Boottest Jenkins views location. + jenkins_public = "https://jenkins.qa.ubuntu.com/job" + jenkins_private = ( + "http://d-jenkins.ubuntu-ci:8080/view/%s/view/BootTest/job" % + self.options.series.title()) + # 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') + public_url = "%s/%s-boottest-%s/lastBuild" % ( + jenkins_public, self.options.series, + excuse.name.replace("+", "-")) + private_url = "%s/%s-boottest-%s/lastBuild" % ( + jenkins_private, self.options.series, + excuse.name.replace("+", "-")) + excuse.addhtml( + "Boottest result: %s (Jenkins: public" + ", private)" % ( + label, public_url, private_url)) + # Allows hints to force boottest failures/attempts + # to be ignored. + hints = self.hints.search('force', package=excuse.name) + hints.extend( + self.hints.search('force-badtest', package=excuse.name)) + forces = [x for x in hints + if same_source(excuse.ver[1], x.version)] + if forces: + excuse.addhtml( + "Should wait for %s %s boottest, but forced by " + "%s" % (excuse.name, excuse.ver[1], + forces[0].user)) + continue + # Block promotion if any boottests attempt has failed or + # still in progress. + if status not in BootTest.VALID_STATUSES: + excuse.addhtml("Not considered") + excuse.addreason("boottest") + excuse.is_valid = False + upgrade_me.remove(excuse.name) + unconsidered.append(excuse.name) + # invalidate impossible excuses for e in self.excuses: # parts[0] == package name diff --git a/excuse.py b/excuse.py index e82584e..852cc7c 100644 --- a/excuse.py +++ b/excuse.py @@ -1,6 +1,6 @@ # -*- coding: utf-8 -*- -# Copyright (C) 2001-2004 Anthony Towns +# Copyright (C) 2006, 2011-2015 Anthony Towns # Andreas Barth # Fabio Tranchitella @@ -52,6 +52,7 @@ class Excuse(object): self._dontinvalidate = False self.forced = False self.run_autopkgtest = False + self.run_boottest = False self.distribution = "ubuntu" self.invalid_deps = [] diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..81c0316 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1,164 @@ +# (C) 2015 Canonical Ltd. +# +# 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. + +import os +import shutil +import subprocess +import tempfile +import unittest + + +PROJECT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + +architectures = ['amd64', 'arm64', 'armhf', 'i386', 'powerpc', 'ppc64el'] + + +class TestData: + + def __init__(self): + '''Construct local test package indexes. + + The archive is initially empty. You can create new packages with + create_deb(). self.path contains the path of the archive, and + self.apt_source provides an apt source "deb" line. + + It is kept in a temporary directory which gets removed when the Archive + object gets deleted. + ''' + self.path = tempfile.mkdtemp(prefix='testarchive.') + self.apt_source = 'deb file://%s /' % self.path + self.series = 'series' + self.dirs = {False: os.path.join(self.path, 'data', self.series), + True: os.path.join( + self.path, 'data', '%s-proposed' % self.series)} + os.makedirs(self.dirs[False]) + os.mkdir(self.dirs[True]) + self.added_sources = {False: set(), True: set()} + self.added_binaries = {False: set(), True: set()} + + # pre-create all files for all architectures + for arch in architectures: + for dir in self.dirs.values(): + with open(os.path.join(dir, 'Packages_' + arch), 'w'): + pass + for dir in self.dirs.values(): + for fname in ['Dates', 'Blocks']: + with open(os.path.join(dir, fname), 'w'): + pass + for dname in ['Hints']: + os.mkdir(os.path.join(dir, dname)) + + os.mkdir(os.path.join(self.path, 'output')) + + # create temporary home dir for proposed-migration autopktest status + self.home = os.path.join(self.path, 'home') + os.environ['HOME'] = self.home + os.makedirs(os.path.join(self.home, 'proposed-migration', + 'autopkgtest', 'work')) + + def __del__(self): + shutil.rmtree(self.path) + + def add(self, name, unstable, fields={}, add_src=True): + '''Add a binary package to the index file. + + You need to specify at least the package name and in which list to put + it (unstable==True for unstable/proposed, or False for + testing/release). fields specifies all additional entries, e. g. + {'Depends': 'foo, bar', 'Conflicts: baz'}. There are defaults for most + fields. + + Unless add_src is set to False, this will also automatically create a + source record, based on fields['Source'] and name. + ''' + assert (name not in self.added_binaries[unstable]) + self.added_binaries[unstable].add(name) + + fields.setdefault('Architecture', architectures[0]) + fields.setdefault('Version', '1') + fields.setdefault('Priority', 'optional') + fields.setdefault('Section', 'devel') + fields.setdefault('Description', 'test pkg') + if fields['Architecture'] == 'all': + for a in architectures: + self._append(name, unstable, 'Packages_' + a, fields) + else: + self._append(name, unstable, 'Packages_' + fields['Architecture'], + fields) + + if add_src: + src = fields.get('Source', name) + if src not in self.added_sources[unstable]: + self.add_src(src, unstable, {'Version': fields['Version'], + 'Section': fields['Section']}) + + def add_src(self, name, unstable, fields={}): + '''Add a source package to the index file. + + You need to specify at least the package name and in which list to put + it (unstable==True for unstable/proposed, or False for + testing/release). fields specifies all additional entries, which can be + Version (default: 1), Section (default: devel), and Extra-Source-Only. + ''' + assert (name not in self.added_sources[unstable]) + self.added_sources[unstable].add(name) + + fields.setdefault('Version', '1') + fields.setdefault('Section', 'devel') + self._append(name, unstable, 'Sources', fields) + + def _append(self, name, unstable, file_name, fields): + with open(os.path.join(self.dirs[unstable], file_name), 'a') as f: + f.write('''Package: %s +Maintainer: Joe +''' % name) + + for k, v in fields.items(): + f.write('%s: %s\n' % (k, v)) + f.write('\n') + + +class TestBase(unittest.TestCase): + + def setUp(self): + super(TestBase, self).setUp() + self.data = TestData() + self.britney = os.path.join(PROJECT_DIR, 'britney.py') + self.britney_conf = os.path.join(PROJECT_DIR, 'britney.conf') + assert os.path.exists(self.britney) + assert os.path.exists(self.britney_conf) + + def tearDown(self): + del self.data + + def restore_config(self, content): + """Helper for restoring configuration contents on cleanup.""" + with open(self.britney_conf, 'w') as fp: + fp.write(content) + + def run_britney(self, args=[]): + '''Run britney. + + Assert that it succeeds and does not produce anything on stderr. + Return (excuses.html, britney_out). + ''' + britney = subprocess.Popen([self.britney, '-v', '-c', self.britney_conf, + '--distribution=ubuntu', + '--series=%s' % self.data.series], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + cwd=self.data.path, + universal_newlines=True) + (out, err) = britney.communicate() + self.assertEqual(britney.returncode, 0, out + err) + self.assertEqual(err, '') + + with open(os.path.join(self.data.path, 'output', self.data.series, + 'excuses.html')) as f: + excuses = f.read() + + return (excuses, out) diff --git a/tests/test_autopkgtest.py b/tests/test_autopkgtest.py index d089f29..6112098 100644 --- a/tests/test_autopkgtest.py +++ b/tests/test_autopkgtest.py @@ -6,133 +6,50 @@ # the Free Software Foundation; either version 2 of the License, or # (at your option) any later version. -import tempfile -import shutil +import apt_pkg +import operator import os import sys import subprocess import unittest -import apt_pkg -import operator -apt_pkg.init() -architectures = ['amd64', 'arm64', 'armhf', 'i386', 'powerpc', 'ppc64el'] +PROJECT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, PROJECT_DIR) -my_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +from autopkgtest import ADT_EXCUSES_LABELS +from tests import TestBase NOT_CONSIDERED = False VALID_CANDIDATE = True -sys.path.insert(0, my_dir) -from autopkgtest import ADT_EXCUSES_LABELS + +apt_pkg.init() -class TestData: - def __init__(self): - '''Construct local test package indexes. +class TestAutoPkgTest(TestBase): - The archive is initially empty. You can create new packages with - create_deb(). self.path contains the path of the archive, and - self.apt_source provides an apt source "deb" line. - - It is kept in a temporary directory which gets removed when the Archive - object gets deleted. - ''' - self.path = tempfile.mkdtemp(prefix='testarchive.') - self.apt_source = 'deb file://%s /' % self.path - self.series = 'series' - self.dirs = {False: os.path.join(self.path, 'data', self.series), - True: os.path.join(self.path, 'data', '%s-proposed' % self.series)} - os.makedirs(self.dirs[False]) - os.mkdir(self.dirs[True]) - self.added_sources = {False: set(), True: set()} - self.added_binaries = {False: set(), True: set()} - - # pre-create all files for all architectures - for arch in architectures: - for dir in self.dirs.values(): - with open(os.path.join(dir, 'Packages_' + arch), 'w'): - pass - for dir in self.dirs.values(): - for fname in ['Dates', 'Blocks']: - with open(os.path.join(dir, fname), 'w'): - pass - for dname in ['Hints']: - os.mkdir(os.path.join(dir, dname)) - - os.mkdir(os.path.join(self.path, 'output')) - - # create temporary home dir for proposed-migration autopktest status - self.home = os.path.join(self.path, 'home') - os.environ['HOME'] = self.home - os.makedirs(os.path.join(self.home, 'proposed-migration', - 'autopkgtest', 'work')) - - def __del__(self): - shutil.rmtree(self.path) - - def add(self, name, unstable, fields={}, add_src=True): - '''Add a binary package to the index file. - - You need to specify at least the package name and in which list to put - it (unstable==True for unstable/proposed, or False for - testing/release). fields specifies all additional entries, e. g. - {'Depends': 'foo, bar', 'Conflicts: baz'}. There are defaults for most - fields. - - Unless add_src is set to False, this will also automatically create a - source record, based on fields['Source'] and name. - ''' - assert (name not in self.added_binaries[unstable]) - self.added_binaries[unstable].add(name) - - fields.setdefault('Architecture', architectures[0]) - fields.setdefault('Version', '1') - fields.setdefault('Priority', 'optional') - fields.setdefault('Section', 'devel') - fields.setdefault('Description', 'test pkg') - if fields['Architecture'] == 'all': - for a in architectures: - self._append(name, unstable, 'Packages_' + a, fields) - else: - self._append(name, unstable, 'Packages_' + fields['Architecture'], - fields) - - if add_src: - src = fields.get('Source', name) - if src not in self.added_sources[unstable]: - self.add_src(src, unstable, {'Version': fields['Version'], - 'Section': fields['Section']}) - - def add_src(self, name, unstable, fields={}): - '''Add a source package to the index file. - - You need to specify at least the package name and in which list to put - it (unstable==True for unstable/proposed, or False for - testing/release). fields specifies all additional entries, which can be - Version (default: 1), Section (default: devel), and Extra-Source-Only. - ''' - assert (name not in self.added_sources[unstable]) - self.added_sources[unstable].add(name) - - fields.setdefault('Version', '1') - fields.setdefault('Section', 'devel') - self._append(name, unstable, 'Sources', fields) - - def _append(self, name, unstable, file_name, fields): - with open(os.path.join(self.dirs[unstable], file_name), 'a') as f: - f.write('''Package: %s -Maintainer: Joe -''' % name) - - for k, v in fields.items(): - f.write('%s: %s\n' % (k, v)) - f.write('\n') - - -class Test(unittest.TestCase): def setUp(self): - self.data = TestData() + super(TestAutoPkgTest, self).setUp() + + # Mofify configuration according to the test context. + with open(self.britney_conf, 'r') as fp: + original_config = fp.read() + # Disable boottests. + new_config = original_config.replace( + 'BOOTTEST_ENABLE = yes', 'BOOTTEST_ENABLE = no') + with open(self.britney_conf, 'w') as fp: + fp.write(new_config) + self.addCleanup(self.restore_config, original_config) + + # fake adt-britney script + self.adt_britney = os.path.join( + self.data.home, 'auto-package-testing', 'jenkins', 'adt-britney') + os.makedirs(os.path.dirname(self.adt_britney)) + + with open(self.adt_britney, 'w') as f: + f.write('''#!/bin/sh -e +echo "$@" >> /%s/adt-britney.log ''' % self.data.path) + os.chmod(self.adt_britney, 0o755) # add a bunch of packages to testing to avoid repetition self.data.add('libc6', False) @@ -146,24 +63,6 @@ class Test(unittest.TestCase): 'Conflicts': 'green'}) self.data.add('justdata', False, {'Architecture': 'all'}) - self.britney = os.path.join(my_dir, 'britney.py') - self.britney_conf = os.path.join(my_dir, 'britney.conf') - assert os.path.exists(self.britney) - assert os.path.exists(self.britney_conf) - - # fake adt-britney script - self.adt_britney = os.path.join(self.data.home, 'auto-package-testing', - 'jenkins', 'adt-britney') - os.makedirs(os.path.dirname(self.adt_britney)) - - with open(self.adt_britney, 'w') as f: - f.write('''#!/bin/sh -e -echo "$@" >> /%s/adt-britney.log ''' % self.data.path) - os.chmod(self.adt_britney, 0o755) - - def tearDown(self): - del self.data - def __merge_records(self, results, history=""): '''Merges a list of results with records in history. @@ -235,28 +134,29 @@ args.func() 'rq': request, 'res': self.__merge_records(request, history)}) - def run_britney(self, args=[]): - '''Run britney. + def do_test(self, unstable_add, adt_request, considered, expect=None, + no_expect=None, history=""): + for (pkg, fields) in unstable_add: + self.data.add(pkg, True, fields) - Assert that it succeeds and does not produce anything on stderr. - Return (excuses.html, britney_out). - ''' - britney = subprocess.Popen([self.britney, '-v', '-c', self.britney_conf, - '--distribution=ubuntu', - '--series=%s' % self.data.series], - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - cwd=self.data.path, - universal_newlines=True) - (out, err) = britney.communicate() - self.assertEqual(britney.returncode, 0, out + err) - self.assertEqual(err, '') + self.make_adt_britney(adt_request, history) - with open(os.path.join(self.data.path, 'output', self.data.series, - 'excuses.html')) as f: - excuses = f.read() + (excuses, out) = self.run_britney() + #print('-------\nexcuses: %s\n-----' % excuses) + #print('-------\nout: %s\n-----' % out) + #print('run:\n%s -c %s\n' % (self.britney, self.britney_conf)) + #subprocess.call(['bash', '-i'], cwd=self.data.path) + if considered: + self.assertIn('Valid candidate', excuses) + else: + self.assertIn('Not considered', excuses) - return (excuses, out) + if expect: + for re in expect: + self.assertRegexpMatches(excuses, re) + if no_expect: + for re in no_expect: + self.assertNotRegexpMatches(excuses, re) def test_no_request_for_uninstallable(self): '''Does not request a test for an uninstallable package''' @@ -557,30 +457,6 @@ args.func() history="lightgreen 1 PASS lightgreen 1" ) - def do_test(self, unstable_add, adt_request, considered, expect=None, - no_expect=None, history=""): - for (pkg, fields) in unstable_add: - self.data.add(pkg, True, fields) - - self.make_adt_britney(adt_request, history) - - (excuses, out) = self.run_britney() - #print('-------\nexcuses: %s\n-----' % excuses) - #print('-------\nout: %s\n-----' % out) - #print('run:\n%s -c %s\n' % (self.britney, self.britney_conf)) - #subprocess.call(['bash', '-i'], cwd=self.data.path) - if considered: - self.assertIn('Valid candidate', excuses) - else: - self.assertIn('Not considered', excuses) - - if expect: - for re in expect: - self.assertRegexpMatches(excuses, re) - if no_expect: - for re in no_expect: - self.assertNotRegexpMatches(excuses, re) - def shell(self): # uninstallable unstable version self.data.add('yellow', True, {'Version': '1.1~beta', @@ -593,4 +469,5 @@ args.func() subprocess.call(['bash', '-i'], cwd=self.data.path) -unittest.main() +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_boottest.py b/tests/test_boottest.py new file mode 100644 index 0000000..77be561 --- /dev/null +++ b/tests/test_boottest.py @@ -0,0 +1,411 @@ +#!/usr/bin/python +# (C) 2014 Canonical Ltd. +# +# 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. + +import mock +import os +import shutil +import sys +import tempfile +import unittest + + +PROJECT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, PROJECT_DIR) + +import boottest +from tests import TestBase + + +def create_manifest(manifest_dir, lines): + """Helper function for writing touch image manifests.""" + os.makedirs(manifest_dir) + with open(os.path.join(manifest_dir, 'manifest'), 'w') as fd: + fd.write('\n'.join(lines)) + + +class FakeResponse(object): + + def __init__(self, code=404, content=''): + self.code = code + self.content = content + + def read(self): + return self.content + + +class TestTouchManifest(unittest.TestCase): + + def setUp(self): + super(TestTouchManifest, self).setUp() + self.path = tempfile.mkdtemp(prefix='boottest') + os.chdir(self.path) + self.imagesdir = os.path.join(self.path, 'boottest/images') + os.makedirs(self.imagesdir) + self.addCleanup(shutil.rmtree, self.path) + _p = mock.patch('urllib.urlopen') + self.mocked_urlopen = _p.start() + self.mocked_urlopen.side_effect = [FakeResponse(code=404),] + self.addCleanup(_p.stop) + self.fetch_retries_orig = boottest.FETCH_RETRIES + def restore_fetch_retries(): + boottest.FETCH_RETRIES = self.fetch_retries_orig + boottest.FETCH_RETRIES = 0 + self.addCleanup(restore_fetch_retries) + + def test_missing(self): + # Missing manifest file silently results in empty contents. + manifest = boottest.TouchManifest('I-dont-exist', 'vivid') + self.assertEqual([], manifest._manifest) + self.assertNotIn('foo', manifest) + + def test_fetch(self): + # Missing manifest file is fetched dynamically + self.mocked_urlopen.side_effect = [ + FakeResponse(code=200, content='foo 1.0'), + ] + manifest = boottest.TouchManifest('ubuntu-touch', 'vivid') + self.assertNotEqual([], manifest._manifest) + + def test_fetch_disabled(self): + # Manifest auto-fetching can be disabled. + manifest = boottest.TouchManifest('ubuntu-touch', 'vivid', fetch=False) + self.mocked_urlopen.assert_not_called() + self.assertEqual([], manifest._manifest) + + def test_fetch_fails(self): + project = 'fake' + series = 'fake' + manifest_dir = os.path.join(self.imagesdir, project, series) + manifest_lines = [ + 'foo:armhf 1~beta1', + ] + create_manifest(manifest_dir, manifest_lines) + manifest = boottest.TouchManifest(project, series) + self.assertEqual(1, len(manifest._manifest)) + self.assertIn('foo', manifest) + + def test_fetch_exception(self): + self.mocked_urlopen.side_effect = [IOError("connection refused")] + manifest = boottest.TouchManifest('not-real', 'not-real') + self.assertEqual(0, len(manifest._manifest)) + + def test_simple(self): + # Existing manifest file allows callsites to properly check presence. + manifest_dir = os.path.join(self.imagesdir, 'ubuntu/vivid') + manifest_lines = [ + 'bar 1234', + 'foo:armhf 1~beta1', + 'boing1-1.2\t666', + 'click:com.ubuntu.shorts 0.2.346' + ] + create_manifest(manifest_dir, manifest_lines) + + manifest = boottest.TouchManifest('ubuntu', 'vivid') + # We can dig deeper on the manifest package names list ... + self.assertEqual( + ['bar', 'boing1-1.2', 'foo'], manifest._manifest) + # but the ' in manifest' API reads better. + self.assertIn('foo', manifest) + self.assertIn('boing1-1.2', manifest) + self.assertNotIn('baz', manifest) + # 'click' name is blacklisted due to the click package syntax. + self.assertNotIn('click', manifest) + + +class TestBoottestEnd2End(TestBase): + """End2End tests (calling `britney`) for the BootTest criteria.""" + + def setUp(self): + super(TestBoottestEnd2End, self).setUp() + + # Modify shared configuration file. + with open(self.britney_conf, 'r') as fp: + original_config = fp.read() + # Disable autopkgtests. + new_config = original_config.replace( + 'ADT_ENABLE = yes', 'ADT_ENABLE = no') + # Disable TouchManifest auto-fetching. + new_config = new_config.replace( + 'BOOTTEST_FETCH = yes', 'BOOTTEST_FETCH = no') + with open(self.britney_conf, 'w') as fp: + fp.write(new_config) + self.addCleanup(self.restore_config, original_config) + + self.data.add('libc6', False, {'Architecture': 'armhf'}), + + self.data.add( + 'libgreen1', + False, + {'Source': 'green', 'Architecture': 'armhf', + 'Depends': 'libc6 (>= 0.9)'}) + self.data.add( + 'green', + False, + {'Source': 'green', 'Architecture': 'armhf', + 'Depends': 'libc6 (>= 0.9), libgreen1'}) + self.create_manifest([ + 'green 1.0', + 'pyqt5:armhf 1.0', + 'signon 1.0' + ]) + + def create_manifest(self, lines): + """Create a manifest for this britney run context.""" + path = os.path.join( + self.data.path, + 'boottest/images/ubuntu-touch/{}'.format(self.data.series)) + create_manifest(path, lines) + + def make_boottest(self): + """Create a stub version of boottest-britney script.""" + script_path = os.path.expanduser( + "~/auto-package-testing/jenkins/boottest-britney") + os.makedirs(os.path.dirname(script_path)) + with open(script_path, 'w') as f: + f.write('''#!%(py)s +import argparse +import os +import shutil +import sys + +template = """ +green 1.1~beta RUNNING +pyqt5-src 1.1~beta PASS +pyqt5-src 1.1 FAIL +signon 1.1 PASS +""" + +def request(): + work_path = os.path.dirname(args.output) + os.makedirs(work_path) + shutil.copy(args.input, os.path.join(work_path, 'test_input')) + with open(args.output, 'w') as f: + f.write(template) + +def submit(): + pass + +def collect(): + with open(args.output, 'w') as f: + f.write(template) + +p = argparse.ArgumentParser() +p.add_argument('-r') +p.add_argument('-c') +p.add_argument('-d', default=False, action='store_true') +p.add_argument('-P', default=False, action='store_true') +p.add_argument('-U', default=False, action='store_true') + +sp = p.add_subparsers() + +psubmit = sp.add_parser('submit') +psubmit.add_argument('input') +psubmit.set_defaults(func=submit) + +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() + ''' % {'py': sys.executable}) + os.chmod(script_path, 0o755) + + def do_test(self, context, expect=None, no_expect=None): + """Process the given package context and assert britney results.""" + for (pkg, fields) in context: + self.data.add(pkg, True, fields) + self.make_boottest() + (excuses, out) = self.run_britney() + #print('-------\nexcuses: %s\n-----' % excuses) + if expect: + for re in expect: + self.assertRegexpMatches(excuses, re) + if no_expect: + for re in no_expect: + self.assertNotRegexpMatches(excuses, re) + + def test_runs(self): + # `Britney` runs and considers binary packages for boottesting + # when it is enabled in the configuration, only binaries needed + # in the phone image are considered for boottesting. + # The boottest status is presented along with its corresponding + # jenkins job urls for the public and the private servers. + # 'in progress' tests blocks package promotion. + context = [ + ('green', {'Source': 'green', 'Version': '1.1~beta', + 'Architecture': 'armhf', 'Depends': 'libc6 (>= 0.9)'}), + ('libgreen1', {'Source': 'green', 'Version': '1.1~beta', + 'Architecture': 'armhf', + 'Depends': 'libc6 (>= 0.9)'}), + ] + public_jenkins_url = ( + 'https://jenkins.qa.ubuntu.com/job/series-boottest-green/' + 'lastBuild') + private_jenkins_url = ( + 'http://d-jenkins.ubuntu-ci:8080/view/Series/view/BootTest/' + 'job/series-boottest-green/lastBuild') + self.do_test( + context, + [r'\bgreen\b.*>1 to .*>1.1~beta<', + r'
  • Boottest result: {} \(Jenkins: ' + r'public, private\)'.format( + boottest.BootTest.EXCUSE_LABELS['RUNNING'], + public_jenkins_url, private_jenkins_url), + '
  • Not considered']) + + # The `boottest-britney` input (recorded for testing purposes), + # contains a line matching the requested boottest attempt. + # ' \n' + test_input_path = os.path.join( + self.data.path, 'boottest/work/test_input') + self.assertEqual( + ['green 1.1~beta\n'], open(test_input_path).readlines()) + + def test_pass(self): + # `Britney` updates boottesting information in excuses when the + # package test pass and marks the package as a valid candidate for + # promotion. + context = [] + context.append( + ('signon', {'Version': '1.1', 'Architecture': 'armhf'})) + self.do_test( + context, + [r'\bsignon\b.*\(- to .*>1.1<', + '
  • Boottest result: {}'.format( + boottest.BootTest.EXCUSE_LABELS['PASS']), + '
  • Valid candidate']) + + def test_fail(self): + # `Britney` updates boottesting information in excuses when the + # package test fails and blocks the package promotion + # ('Not considered.') + context = [] + context.append( + ('pyqt5', {'Source': 'pyqt5-src', 'Version': '1.1', + 'Architecture': 'all'})) + self.do_test( + context, + [r'\bpyqt5-src\b.*\(- to .*>1.1<', + '
  • Boottest result: {}'.format( + boottest.BootTest.EXCUSE_LABELS['FAIL']), + '
  • Not considered']) + + def test_unknown(self): + # `Britney` does not block on missing boottest results for a + # particular source/version, in this case pyqt5-src_1.2 (not + # listed in the testing result history). Instead it renders + # excuses with 'UNKNOWN STATUS' and links to the corresponding + # jenkins jobs for further investigation. Source promotion is + # blocked, though. + context = [ + ('pyqt5', {'Source': 'pyqt5-src', 'Version': '1.2', + 'Architecture': 'armhf'})] + self.do_test( + context, + [r'\bpyqt5-src\b.*\(- to .*>1.2<', + r'
  • Boottest result: UNKNOWN STATUS \(Jenkins: .*\)', + '
  • Not considered']) + + def create_hint(self, username, content): + """Populates a hint file for the given 'username' with 'content'.""" + hints_path = os.path.join( + self.data.path, + 'data/{}-proposed/Hints/{}'.format(self.data.series, username)) + 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-badtest'. The boottest + # attempt will still be requested and its results would be considered + # for other non-forced sources. + context = [ + ('pyqt5', {'Source': 'pyqt5-src', 'Version': '1.1', + 'Architecture': 'all'}), + ] + self.create_hint('cjwatson', 'force pyqt5-src/1.1') + self.do_test( + context, + [r'\bpyqt5-src\b.*\(- to .*>1.1<', + '
  • Boottest result: {}'.format( + boottest.BootTest.EXCUSE_LABELS['FAIL']), + '
  • Should wait for pyqt5-src 1.1 boottest, ' + 'but forced by cjwatson', + '
  • Valid candidate']) + + 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-badtest green/1.1~beta') + self.do_test( + context, + [r'\bgreen\b.*>1 to .*>1.1~beta<', + '
  • Boottest result: {}'.format( + boottest.BootTest.EXCUSE_LABELS['RUNNING']), + '
  • Should wait for green 1.1~beta boottest, but forced ' + 'by cjwatson', + '
  • Valid candidate']) + + def test_skipped_not_on_phone(self): + # `Britney` updates boottesting information in excuses when the + # package was skipped and marks the package as a valid candidate for + # promotion, but no notice about 'boottest' is added to the excuse. + context = [] + context.append( + ('apache2', {'Source': 'apache2-src', 'Architecture': 'all', + 'Version': '2.4.8-1ubuntu1'})) + self.do_test( + context, + [r'\bapache2-src\b.*\(- to .*>2.4.8-1ubuntu1<', + '
  • Valid candidate'], + ['
  • Boottest result:'], + ) + + def test_skipped_architecture_not_allowed(self): + # `Britney` does not trigger boottests for source not yet built on + # the allowed architectures. + self.data.add( + 'pyqt5', False, {'Source': 'pyqt5-src', 'Architecture': 'armhf'}) + context = [ + ('pyqt5', {'Source': 'pyqt5-src', 'Version': '1.1', + 'Architecture': 'amd64'}), + ] + self.do_test( + context, + [r'\bpyqt5-src\b.*>1 to .*>1.1<', + r'
  • missing build on .*>armhf: pyqt5 \(from .*>1\)', + '
  • Not considered']) + + + +if __name__ == '__main__': + unittest.main()