Merge lp:~canonical-ci-engineering/britney/boottesting-support

This commit is contained in:
Steve Langasek 2015-02-19 17:11:39 -08:00
commit 9afd502e7f
7 changed files with 1015 additions and 176 deletions

282
boottest.py Normal file
View File

@ -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": '<span style="background:#87d96c">Pass</span>',
"FAIL": '<span style="background:#ff6666">Regression</span>',
"RUNNING": '<span style="background:#99ddff">Test in progress</span>',
}
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 "<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[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)

View File

@ -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

View File

@ -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):
["<a href=\"http://bugs.debian.org/%s\">#%s</a>" % (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: <a href=\"%s\">public</a>"
", <a href=\"%s\">private</a>)" % (
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

View File

@ -1,6 +1,6 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2001-2004 Anthony Towns <ajt@debian.org>
# Copyright (C) 2006, 2011-2015 Anthony Towns <ajt@debian.org>
# Andreas Barth <aba@debian.org>
# Fabio Tranchitella <kobold@debian.org>
@ -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 = []

164
tests/__init__.py Normal file
View File

@ -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 <joe@example.com>
''' % 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)

View File

@ -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 <joe@example.com>
''' % 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()

411
tests/test_boottest.py Normal file
View File

@ -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 '<name> 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</a> to .*>1.1~beta<',
r'<li>Boottest result: {} \(Jenkins: '
r'<a href="{}">public</a>, <a href="{}">private</a>\)'.format(
boottest.BootTest.EXCUSE_LABELS['RUNNING'],
public_jenkins_url, private_jenkins_url),
'<li>Not considered'])
# The `boottest-britney` input (recorded for testing purposes),
# contains a line matching the requested boottest attempt.
# '<source> <version>\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<',
'<li>Boottest result: {}'.format(
boottest.BootTest.EXCUSE_LABELS['PASS']),
'<li>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<',
'<li>Boottest result: {}'.format(
boottest.BootTest.EXCUSE_LABELS['FAIL']),
'<li>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'<li>Boottest result: UNKNOWN STATUS \(Jenkins: .*\)',
'<li>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<',
'<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-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<',
'<li>Boottest result: {}'.format(
boottest.BootTest.EXCUSE_LABELS['FAIL']),
'<li>Should wait for pyqt5-src 1.1 boottest, '
'but forced by cjwatson',
'<li>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</a> to .*>1.1~beta<',
'<li>Boottest result: {}'.format(
boottest.BootTest.EXCUSE_LABELS['RUNNING']),
'<li>Should wait for green 1.1~beta boottest, but forced '
'by cjwatson',
'<li>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<',
'<li>Valid candidate'],
['<li>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</a> to .*>1.1<',
r'<li>missing build on .*>armhf</a>: pyqt5 \(from .*>1</a>\)',
'<li>Not considered'])
if __name__ == '__main__':
unittest.main()