Merge lp:~canonical-ci-engineering/britney/queued-announce-and-collect into lp:~ubuntu-release/britney/britney2-ubuntu

[Francis Ginther]

Enable TestClient work.

[Francis Ginther (fginther)]

Two inline comments below.

Also, at some point, we need to think a little more about how the hints are used. These were originally setup for dealing with autopkgtests and the hints 'force-skiptest' and 'force-badtest' are really geared toward skipping those known tests. But we really don't want to skip other types of tests just because a package has some poorly written autopkgtest that no one wants to update. In this current MP, 'force-skiptest' is not used and I think that's the right way to go. I also think we need to remove 'force'badtest' and just use 'force' to act as a universal override (to override any type of autopkgtest, boottest, or snappy-selftest failure). This may be enough, but perhaps we'll need to define a set of hints specific to each required test type (i.e. 'force-bad-boottest', 'force-bad-snappy-selftest').

[Thomi Richards (thomir-deactivatedaccount)]

I can't say as I understand more than about 10% of this, but I left a few comments.

[Celso Providelo (cprov)]

COmments addressed in https://code.launchpad.net/~cprov/britney/integration-tests/+merge/260046

[lp:~canonical-ci-engineering/britney/queued-announce-and-collect]

436. By Celso Providelo on 2015-05-25

Merge lp:~cprov/britney/integration-tests

437. By Celso Providelo on 2015-05-26

Add integration tests for hints on testclient. Also refactoring TestBase.{overrideConfig, create_hint) so they can be reused across tests.

438. By Para Siva on 2015-05-27

Adding distribution to the testclient implementation

439. By Celso Providelo on 2015-05-29

Extending TestClient announcements implementation to include the DSC 'Binary' (and potentially other) information on the payload.

[Martin Pitt (pitti)]

Do you still want to go through with this, or is this stalled now? (I suppose the latter?)

I submitted https://code.launchpad.net/~pitti/britney/britney2-ubuntu-amqp/+merge/263679 for test requests, and I'll work on collecting test results directly from swift. This will avoid separate components like adt-britney which duplicate work from britney (like reverse dependency calculation).

Unmerged revisions

439. By Celso Providelo on 2015-05-29

Extending TestClient announcements implementation to include the DSC 'Binary' (and potentially other) information on the payload.

438. By Para Siva on 2015-05-27

Adding distribution to the testclient implementation

437. By Celso Providelo on 2015-05-26

Add integration tests for hints on testclient. Also refactoring TestBase.{overrideConfig, create_hint) so they can be reused across tests.

436. By Celso Providelo on 2015-05-25

Merge lp:~cprov/britney/integration-tests

435. By Celso Providelo on 2015-05-22

Merge lp:~cprov/britney/testclient-api/

=== modified file 'britney.conf'
--- britney.conf	2015-03-05 14:57:03 +0000
+++ britney.conf	2015-05-29 14:31:13 +0000
@@ -70,3 +70,7 @@
 BOOTTEST_DEBUG    = yes
 BOOTTEST_ARCHES   = armhf amd64
 BOOTTEST_FETCH    = yes
+
+TESTCLIENT_ENABLE = yes
+TESTCLIENT_AMQP_URIS = amqp://guest:guest@162.213.32.181:5672//
+TESTCLIENT_REQUIRED_TESTS = 

=== modified file 'britney.py'
--- britney.py	2015-02-20 19:02:00 +0000
+++ britney.py	2015-05-29 14:31:13 +0000
@@ -227,6 +227,7 @@
                    PROVIDES, RDEPENDS, RCONFLICTS, MULTIARCH, ESSENTIAL)
 from autopkgtest import AutoPackageTest, ADT_PASS, ADT_EXCUSES_LABELS
 from boottest import BootTest
+from testclient import TestClient
 
 
 __author__ = 'Fabio Tranchitella and the Debian Release Team'
@@ -1984,6 +1985,65 @@
                         upgrade_me.remove(excuse.name)
                         unconsidered.append(excuse.name)
 
+        if (getattr(self.options, "testclient_enable", "no") == "yes" and
+            self.options.series):
+
+            # Filter only new source candidates excuses.
+            testing_excuses = []
+            for excuse in self.excuses:
+                # 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
+                testing_excuses.append(excuse)
+
+            amqp_uris = getattr(
+                self.options, "testclient_amqp_uris", "").split()
+            testclient = TestClient(
+                    self.options.distribution, self.options.series, amqp_uris)
+
+            # Announce new candidates and collect new test results.
+            if not self.options.dry_run:
+                testclient.announce(testing_excuses, self.options.unstable)
+                testclient.collect()
+                testclient.cleanup(testing_excuses)
+
+            # Update excuses considering hints and required_tests (for gating).
+            required_tests = getattr(
+                self.options, "testclient_required_tests", "").split()
+            for excuse in testing_excuses:
+                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)]
+                for test in testclient.getTests(excuse.name, excuse.ver[1]):
+                    label = TestClient.EXCUSE_LABELS.get(
+                        test.get('status'), 'UNKNOWN STATUS')
+                    excuse.addhtml(
+                        "%s result: %s (<a href=\"%s\">results</a>)" % (
+                            test.get('name').capitalize(), label,
+                            test.get('url')))
+                    if forces:
+                        excuse.addhtml(
+                            "Should wait for %s %s %s test, but forced by "
+                            "%s" % (excuse.name, excuse.ver[1],
+                                    test.get('name').capitalize(), 
+                                    forces[0].user))
+                        continue
+                    if test.get('name') not in required_tests:
+                        continue
+                    if test.get('status') not in TestClient.VALID_STATUSES:
+                        excuse.addreason(test.get('name'))
+                        if excuse.is_valid:
+                            excuse.is_valid = False
+                            excuse.addhtml("Not considered")
+                            upgrade_me.remove(excuse.name)
+                            unconsidered.append(excuse.name)
+
         # invalidate impossible excuses
         for e in self.excuses:
             # parts[0] == package name

=== added file 'testclient.py'
--- testclient.py	1970-01-01 00:00:00 +0000
+++ testclient.py	2015-05-29 14:31:13 +0000
@@ -0,0 +1,216 @@
+# -*- 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
+
+import apt_pkg
+from contextlib import (
+    contextmanager,
+    nested,
+)
+import json
+import os
+
+
+import kombu
+from kombu.pools import producers
+
+
+@contextmanager
+def json_cached_info(path):
+    """Context manager for caching a JSON object on disk."""
+    info = {}
+    if os.path.exists(path):
+        with open(path) as fp:
+            try:
+                info = json.load(fp)
+            except ValueError:
+                # cache is empty or corrupted (!).
+                info = {}
+    else:
+        dirname = os.path.dirname(path)
+        if not os.path.exists(dirname):
+            os.makedirs(dirname)
+    yield info
+    with open(path, 'w') as fp:
+        json.dump(info, fp, indent=2)
+
+
+def make_cache_key(name, version):
+    """Return a json-hashable key for given source & version."""
+    return '{}_{}'.format(name, version)
+
+
+def get_repo_sources(basedir):
+    """Return 'Sources' repository index information for the given basedir.
+
+    Extract and cache repository 'Sources' index into a dictionary
+    key-ed by source-key (see `make_cache_key`) pointing to a dictionary
+    containing relevant DSC fields.
+
+    At moment, we only cache 'Binary'.
+    """
+    fields = (
+        'Binary',
+    )
+    filename = os.path.join(basedir, "Sources")
+    tag_file = apt_pkg.TagFile(open(filename))
+
+    sources = {}
+    while tag_file.step():
+        source_name = tag_file.section.get('Package')
+        source_version = tag_file.section.get('Version')
+        source_key = make_cache_key(source_name, source_version)
+        sources[source_key] = {
+            k: tag_file.section.get(k, '') for k in fields
+        }
+
+    return sources
+
+
+class TestClient(object):
+    """Generic test client implementation.
+
+    announce: announcing new source candidates to a pre-defined rabbitmq
+              exchange (testing subsystems can hook/subscribe).
+
+    collect: collect test results for the context series from a pre-defined
+             rabbitmq exchange (other promotion agents can do the same).
+
+    cleanup: sanitize internal announcement/testing registry after runs.
+    """
+
+    EXCHANGE_CANDIDATES = 'candidates.exchange.v1'
+    EXCHANGE_RESULTS = 'results.exchange.v1'
+
+    VALID_STATUSES = ('PASS', 'SKIP')
+
+    EXCUSE_LABELS = {
+        "PASS": '<span style="background:#87d96c">Pass</span>',
+        "SKIP": '<span style="background:#ffff00">Test skipped</span>',
+        "FAIL": '<span style="background:#ff6666">Regression</span>',
+        "RUNNING": '<span style="background:#99ddff">Test in progress</span>',
+    }
+
+    def __init__(self, distribution, series, amqp_uris):
+        self.distribution = distribution
+        self.series = series
+        self.amqp_uris = amqp_uris
+
+    @property
+    def cache_path(self):
+        """Series-specific test announcement/result cache."""
+        return 'testclient/{}/{}.json'.format(self.distribution, self.series)
+
+    @property
+    def results_queue(self):
+        """Series-specific queue for collecting tests results."""
+        return 'pm.results.{}.{}'.format(self.distribution,
+                                         self.series)
+
+    def announce(self, excuses, unstable_basedir,
+                 get_repo_sources=get_repo_sources):
+        """Announce new source candidates.
+
+        Post a message to the EXCHANGE_CANDATIDATES for every new given
+        excuses (cache announcements so excuses do not get re-annouced).
+        """
+        with nested(json_cached_info(self.cache_path),
+                    kombu.Connection(self.amqp_uris)) as (cache, connection):
+            # XXX cprov 20150521: nested() is deprecated, and multi-statement
+            # 'with' does not support nesting (i.e. the previous context
+            # manager is not available to the next, in this case
+            # 'connection').
+            with producers[connection].acquire(block=True) as producer:
+                publisher = connection.ensure(
+                    producer, producer.publish, max_retries=3)
+                exchange = kombu.Exchange(
+                    self.EXCHANGE_CANDIDATES, type="fanout")
+                repo = get_repo_sources(unstable_basedir)
+                for excuse in excuses:
+                    source_key = make_cache_key(excuse.name, excuse.ver[1])
+                    if source_key in cache.keys():
+                        continue
+                    repo_data = repo[source_key]
+                    payload = {
+                        'source_name': excuse.name,
+                        'source_version': excuse.ver[1],
+                        'source_binaries': [
+                            b.strip() for b in repo_data['Binary'].split(',')
+                        ],
+                        'series': self.series,
+                        'distribution': self.distribution,
+                    }
+                    publisher(payload, exchange=exchange, declare=[exchange])
+                    cache[make_cache_key(excuse.name, excuse.ver[1])] = []
+
+    def collect(self):
+        """Collect available test results.
+
+        Consume all messages from the EXCHANGE_RESULTS (routed to a series-
+        specific queue). Ignore test results for other series and update
+        test results registry.
+        """
+        with nested(json_cached_info(self.cache_path),
+                    kombu.Connection(self.amqp_uris)) as (cache, connection):
+            exchange = kombu.Exchange(
+                self.EXCHANGE_RESULTS, type="fanout")
+            queue = kombu.Queue(self.results_queue, exchange)
+            # XXX cprov 20150521: same as above about nested context managers.
+            with connection.SimpleQueue(queue) as q:
+                for i in range(len(q)):
+                    msg = q.get()
+                    payload = msg.payload
+                    if payload.get('distribution') != self.distribution:
+                        continue
+                    if payload.get('series') != self.series:
+                        continue
+                    tests = cache.setdefault(
+                        make_cache_key(
+                            payload.get('source_name'),
+                            payload.get('source_version')
+                        ), [])
+                    tests.append({
+                        'name': payload.get('test_name'),
+                        'status': payload.get('test_status'),
+                        'url': payload.get('test_url'),
+                    })
+                    msg.ack()
+
+    def cleanup(self, excuses):
+        """Remove test result entries without corresponding excuse.
+
+        If there is not excuse the test results are not relevant anymore.
+        """
+        with json_cached_info(self.cache_path) as cache:
+            current_keys = [
+                make_cache_key(e.name, e.ver[1]) for e in excuses]
+            cached_keys = list(cache.keys())
+            for k in cached_keys:
+                if k not in current_keys:
+                    del cache[k]
+
+    def getTests(self, name, version):
+        """Yields test results for a given source-version pair.
+
+        Tests results are a list of dictionaries container test-name, status
+        and url.
+        """
+        with json_cached_info(self.cache_path) as cache:
+            tests = cache.get(make_cache_key(name, version), [])
+            for test in tests:
+                yield {
+                    'name': test.get('name'),
+                    'status': test.get('status'),
+                    'url': test.get('url'),
+                }

=== modified file 'tests/__init__.py'
--- tests/__init__.py	2015-02-05 14:43:23 +0000
+++ tests/__init__.py	2015-05-29 14:31:13 +0000
@@ -32,6 +32,7 @@
         self.path = tempfile.mkdtemp(prefix='testarchive.')
         self.apt_source = 'deb file://%s /' % self.path
         self.series = 'series'
+        self.distribution = 'ubuntu'
         self.dirs = {False: os.path.join(self.path, 'data', self.series),
                      True: os.path.join(
                          self.path, 'data', '%s-proposed' % self.series)}
@@ -94,7 +95,8 @@
             src = fields.get('Source', name)
             if src not in self.added_sources[unstable]:
                 self.add_src(src, unstable, {'Version': fields['Version'],
-                                             'Section': fields['Section']})
+                                             'Section': fields['Section'],
+                                             'Binary': name})
 
     def add_src(self, name, unstable, fields={}):
         '''Add a source package to the index file.
@@ -162,3 +164,25 @@
             excuses = f.read()
 
         return (excuses, out)
+
+    def overrideConfig(self, overrides):
+        """Overrides briney configuration based on the given key-value map."""
+        with open(self.britney_conf, 'r') as fp:
+            original_config = fp.read()
+        new_config = []
+        for line in original_config.splitlines():
+            for k, v in overrides.iteritems():
+                if line.startswith(k):
+                    line = '{} = {}'.format(k, v)
+            new_config.append(line)
+        with open(self.britney_conf, 'w') as fp:
+            fp.write('\n'.join(new_config))
+        self.addCleanup(self.restore_config, original_config)
+
+    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)

=== modified file 'tests/test_autopkgtest.py'
--- tests/test_autopkgtest.py	2015-02-05 14:43:23 +0000
+++ tests/test_autopkgtest.py	2015-05-29 14:31:13 +0000
@@ -31,15 +31,12 @@
     def setUp(self):
         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)
+        self.overrideConfig({
+            'ADT_ENABLE': 'yes',
+            'BOOTTEST_ENABLE': 'no',
+            'BOOTTEST_FETCH': 'no',
+            'TESTCLIENT_ENABLE': 'no',
+        })
 
         # fake adt-britney script
         self.adt_britney = os.path.join(

=== modified file 'tests/test_boottest.py'
--- tests/test_boottest.py	2015-02-20 16:28:47 +0000
+++ tests/test_boottest.py	2015-05-29 14:31:13 +0000
@@ -123,21 +123,14 @@
     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')
-        # Enable boottest.
-        new_config = new_config.replace(
-            'BOOTTEST_ENABLE   = no', 'BOOTTEST_ENABLE   = yes')
-        # 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)
+        # Disable autopkgtests + testclient and boottest_fetch
+        # for this testing context.
+        self.overrideConfig({
+            'ADT_ENABLE': 'no',
+            'BOOTTEST_ENABLE': 'yes',
+            'BOOTTEST_FETCH': 'no',
+            'TESTCLIENT_ENABLE': 'no',
+        })
 
         self.data.add('libc6', False, {'Architecture': 'armhf'}),
 
@@ -323,14 +316,6 @@
              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
@@ -415,12 +400,9 @@
         # Boottest can run simultaneously with autopkgtest (adt).
 
         # Enable ADT in britney configuration.
-        with open(self.britney_conf, 'r') as fp:
-            original_config = fp.read()
-        new_config = original_config.replace(
-            'ADT_ENABLE        = no', 'ADT_ENABLE        = yes')
-        with open(self.britney_conf, 'w') as fp:
-            fp.write(new_config)
+        self.overrideConfig({
+            'ADT_ENABLE': 'yes',
+        })
 
         # Create a fake 'adt-britney' that reports a RUNNING job for
         # the testing source ('purple_1.1').

=== added file 'tests/test_testclient.py'
--- tests/test_testclient.py	1970-01-01 00:00:00 +0000
+++ tests/test_testclient.py	2015-05-29 14:31:13 +0000
@@ -0,0 +1,472 @@
+#!/usr/bin/python
+# (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 sys
+import tempfile
+import unittest
+
+import kombu
+from kombu.pools import producers
+
+
+PROJECT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(0, PROJECT_DIR)
+
+from excuse import Excuse
+from testclient import (
+    json_cached_info,
+    make_cache_key,
+    TestClient,
+)
+from tests import TestBase
+
+
+class TestJsonCachedInfo(unittest.TestCase):
+
+    def setUp(self):
+        super(TestJsonCachedInfo, self).setUp()
+        _, self.test_cache = tempfile.mkstemp()
+        self.addCleanup(os.unlink, self.test_cache)
+
+    def test_simple(self):
+        # `json_cached_info` context manager correctly persists a
+        # python dictionary on disk.
+        with json_cached_info(self.test_cache) as cache:
+            self.assertEqual({}, cache)
+            cache['foo'] = 'bar'
+
+        with open(self.test_cache) as fp:
+            self.assertEqual(
+                ['{\n',
+                 '  "foo": "bar"\n',
+                 '}'], fp.readlines())
+
+        with json_cached_info(self.test_cache) as cache:
+            self.assertEqual(cache['foo'], 'bar')
+
+
+def make_excuse(name, version):
+    """Return a `Excuse` for the give source name and version."""
+    e = Excuse(name)
+    e.set_vers('-', version)
+    return e
+
+
+class TestTestClient(unittest.TestCase):
+
+    def setUp(self):
+        super(TestTestClient, self).setUp()
+        self.path = tempfile.mkdtemp(prefix='testclient')
+        os.makedirs(os.path.join(self.path, 'testclient/'))
+        self.addCleanup(shutil.rmtree, self.path)
+
+        os.chdir(self.path)
+        cwd = os.getcwd()
+        self.addCleanup(os.chdir, cwd)
+
+        self.amqp_uris = ['memory://']
+
+    def test_announce(self):
+        # 'announce' post messages to the EXCHANGE_CANDIDATES exchange and
+        # updates its internal cache.
+        testclient = TestClient('ubuntu', 'vivid', self.amqp_uris)
+        test_excuses = [
+            make_excuse('foo', '1.0'),
+            make_excuse('bar', '2.0'),
+        ]
+        test_repo_data = {
+            'foo_1.0': {'Binary': 'a, b'},
+            'bar_2.0': {'Binary': 'c'},
+        }
+
+        with kombu.Connection(self.amqp_uris) as connection:
+            exchange = kombu.Exchange(
+                testclient.EXCHANGE_CANDIDATES, type="fanout")
+            queue = kombu.Queue('testing', exchange)
+            with connection.SimpleQueue(queue) as q:
+                testclient.announce(
+                    test_excuses, None, lambda x: test_repo_data)
+                self.assertEqual(
+                    [{'series': 'vivid',
+                      'distribution': 'ubuntu',
+                      'source_name': 'foo',
+                      'source_version': '1.0',
+                      'source_binaries': ['a', 'b']},
+                     {'series': 'vivid',
+                      'distribution': 'ubuntu',
+                      'source_name': 'bar',
+                      'source_version': '2.0',
+                      'source_binaries': ['c']}],
+                    [q.get().payload for i in range(len(q))])
+
+        with json_cached_info(testclient.cache_path) as cache:
+            self.assertEqual(
+                {'bar_2.0': [], 'foo_1.0': []},
+                cache)
+
+    def test_collect(self):
+        # 'collect' collects test results and aggregates them in its
+        # internal cache.
+        testclient = TestClient('ubuntu', 'vivid', self.amqp_uris)
+
+        result_payloads = [
+            {'source_name': 'foo',
+             'source_version': '1.0',
+             'series': testclient.series,
+             'distribution': testclient.distribution,
+             'test_name': 'snappy',
+             'test_status': 'RUNNING',
+             'test_url': 'http://snappy.com/foo'},
+            {'source_name': 'bar',
+             'source_version': '1.0',
+             'series': testclient.series,
+             'distribution': testclient.distribution,
+             'test_name': 'ubuntu',
+             'test_status': 'RUNNING',
+             'test_url': 'http://ubuntu.com/foo'},
+            {'source_name': 'foo',
+             'source_version': '1.0',
+             'series': testclient.series,
+             'distribution': testclient.distribution,
+             'test_name': 'bbb',
+             'test_status': 'RUNNING',
+             'test_url': 'http://bbb.com/foo'},
+            # This result will be ignored due to the series mismatch.
+            {'source_name': 'zoing',
+             'source_version': '1.0',
+             'series': 'some-other-series',
+             'test_name': 'ubuntu',
+             'test_status': 'RUNNING',
+             'test_url': 'http://ubuntu.com/foo'},
+        ]
+
+        with kombu.Connection(self.amqp_uris) as connection:
+            with producers[connection].acquire(block=True) as producer:
+                # Just for binding destination queue to the exchange.
+                testclient.collect()
+                exchange = kombu.Exchange(
+                    testclient.EXCHANGE_RESULTS, type="fanout")
+                publisher = connection.ensure(
+                    producer, producer.publish, max_retries=3)
+                for payload in result_payloads:
+                    publisher(payload, exchange=exchange, declare=[exchange])
+                testclient.collect()
+
+        with json_cached_info(testclient.cache_path) as cache:
+            self.assertEqual(
+                {'foo_1.0': [{'name': 'snappy',
+                              'status': 'RUNNING',
+                              'url': 'http://snappy.com/foo'},
+                             {'name': 'bbb',
+                              'status': 'RUNNING',
+                              'url': 'http://bbb.com/foo'}],
+                 'bar_1.0': [{'name': 'ubuntu',
+                              'status': 'RUNNING',
+                              'url': 'http://ubuntu.com/foo'}]},
+                cache)
+
+    def test_cleanup(self):
+        # `cleanup` remove cache entries that are not present in the
+        # given excuses list (i.e. not relevant for promotion anymore).
+        testclient = TestClient('ubuntu', 'vivid', self.amqp_uris)
+        test_excuses = [
+            make_excuse('foo', '1.0'),
+            make_excuse('bar', '2.0'),
+        ]
+
+        with json_cached_info(testclient.cache_path) as cache:
+            cache[make_cache_key('foo', '0.9')] = []
+            cache[make_cache_key('foo', '1.0')] = []
+            cache[make_cache_key('bar', '2.0')] = []
+
+        testclient.cleanup(test_excuses)
+
+        with json_cached_info(testclient.cache_path) as cache:
+            self.assertEqual(
+                {'bar_2.0': [], 'foo_1.0': []},
+                cache)
+
+    def test_getTests(self):
+        # `getTests` yields cached tests results for a given source name
+        # and version.
+        testclient = TestClient('ubuntu', 'vivid', self.amqp_uris)
+
+        with json_cached_info(testclient.cache_path) as cache:
+            cache[make_cache_key('foo', '1.0')] = [
+                {'name': 'snappy',
+                 'status': 'RUNNING',
+                 'url': 'http://snappy.com/foo'},
+                {'name': 'bbb',
+                 'status': 'RUNNING',
+                 'url': 'http://bbb.com/foo'}
+            ]
+
+        self.assertEqual(
+            [{'name': 'snappy',
+              'status': 'RUNNING',
+              'url': 'http://snappy.com/foo'},
+             {'name': 'bbb',
+              'status': 'RUNNING',
+              'url': 'http://bbb.com/foo'}],
+            list(testclient.getTests('foo', '1.0')))
+
+        self.assertEqual(
+            [], list(testclient.getTests('bar', '1.0')))
+
+
+def has_local_rabbitmq():
+    """Whether a local rabbitmq server is available with default creds."""
+    with kombu.Connection('amqp://guest:guest@localhost:5672//',
+                          connect_timeout=.1) as c:
+        try:
+            c.connect()
+        except:
+            return False
+        return True
+
+
[e-mail](has_local_rabbitmq(), 'No local rabbitmq')
+class TestTestClientEnd2End(TestBase):
+    """End2End tests (calling `britney`) for the TestClient usage."""
+
+    def setUp(self):
+        super(TestTestClientEnd2End, self).setUp()
+
+        # XXX cprov 20150525: unfortunately, this test requires a proper
+        # amqp transport/server layer (rabbitmq) because kombu 'memory://'
+        # cannot be shared across processes (britney & tests).
+        self.amqp_uris = ['amqp://guest:guest@localhost:5672//']
+
+        self.path = tempfile.mkdtemp(prefix='testclient')
+        os.makedirs(os.path.join(self.path, 'testclient/'))
+        self.addCleanup(shutil.rmtree, self.path)
+
+        os.chdir(self.path)
+        cwd = os.getcwd()
+        self.addCleanup(os.chdir, cwd)
+
+        # Disable autopkgtests + boottest tests and use local rabbit
+        # for this testing context.
+        self.overrideConfig({
+            'ADT_ENABLE': 'no',
+            'BOOTTEST_ENABLE': 'no',
+            'TESTCLIENT_ENABLE': 'yes',
+            'TESTCLIENT_AMQP_URIS': ' '.join(self.amqp_uris),
+        })
+        
+        # We publish a version of 'foo' source to make it 'known'.
+        self.data.add(
+            'foo-bin', False, {'Source': 'foo', 'Architecture': 'amd64'})
+
+    def publishTestResults(self, results):
+        """Publish the given list of test results."""
+        with kombu.Connection(self.amqp_uris) as connection:
+            results_exchange = kombu.Exchange(
+                TestClient.EXCHANGE_RESULTS, type="fanout")
+            with producers[connection].acquire(block=True) as producer:
+                publisher = connection.ensure(
+                    producer, producer.publish, max_retries=3)
+                for payload in results:
+                    publisher(payload, exchange=results_exchange)
+
+    def getAnnouncements(self):
+        """Yields announcements payloads.""" 
+        with kombu.Connection(self.amqp_uris) as connection:
+            candidates_exchange = kombu.Exchange(
+                TestClient.EXCHANGE_CANDIDATES, type="fanout")
+            queue = kombu.Queue('testing', candidates_exchange)
+            with connection.SimpleQueue(queue) as q:
+                for i in range(len(q)):
+                    msg = q.get()
+                    msg.ack()
+                    yield msg.payload
+
+    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)
+
+        # Creates a queue for collecting announcements from
+        # 'candidates.exchanges'.
+        with kombu.Connection(self.amqp_uris) as connection:
+            candidates_exchange = kombu.Exchange(
+                TestClient.EXCHANGE_CANDIDATES, type="fanout")
+            queue = kombu.Queue('testing', candidates_exchange)
+            with connection.SimpleQueue(queue) as q:
+                q.queue.purge()
+
+        (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_non_required_test(self):
+        # Non-required test results are collected as part of the excuse
+        # report but do not block source promotion (i.e. the excuse is
+        # a 'Valid candidate' even if the test is 'in progress').
+
+        # Publish 'in-progress' results for 'bazinga for "foo_1.1"'.
+        test_results = [{
+            'source_name': 'foo',
+            'source_version': '1.1',
+            'series': self.data.series,
+            'distribution': self.data.distribution,
+            'test_name': 'bazinga',
+            'test_status': 'RUNNING',
+            'test_url': 'http://bazinga.com/foo',
+        }]
+        self.publishTestResults(test_results)
+
+        # Run britney for 'foo_1.1' and valid candidated is recorded.
+        context = [
+            ('foo-bin', {'Source': 'foo', 'Version': '1.1',
+                         'Architecture': 'amd64'}),
+        ]
+        self.do_test(
+            context,
+            [r'\bfoo\b.*>1</a> to .*>1.1<',
+             r'<li>Bazinga result: .*>Test in progress.*'
+             r'href="http://bazinga.com/foo">results',
+             '<li>Valid candidate'])
+
+        # 'foo_1.1' source candidate was announced.
+        self.assertEqual(
+            [{'source_name': 'foo',
+              'source_version': '1.1',
+              'source_binaries': ['foo-bin'],
+              'series': self.data.series,
+              'distribution': self.data.distribution,
+            }], list(self.getAnnouncements()))
+
+    def test_required_test(self):
+        # A required-test result is collected and blocks source package
+        # promotion while it hasn't passed.
+
+        # Make 'bazinga' a required test.
+        self.overrideConfig({
+            'TESTCLIENT_REQUIRED_TESTS': 'bazinga',
+        })
+
+        # Publish 'in-progress' results for 'bazinga for "foo_1.1"'.
+        test_results = [{
+            'source_name': 'foo',
+            'source_version': '1.1',
+            'series': self.data.series,
+            'distribution': self.data.distribution,
+            'test_name': 'bazinga',
+            'test_status': 'RUNNING',
+            'test_url': 'http://bazinga.com/foo',
+        }]
+        self.publishTestResults(test_results)
+
+        # Run britney for 'foo_1.1' and an unconsidered excuse is recorded.
+        context = [
+            ('foo-bin', {'Source': 'foo', 'Version': '1.1',
+                         'Architecture': 'amd64'}),
+        ]
+        self.do_test(
+            context,
+            [r'\bfoo\b.*>1</a> to .*>1.1<',
+             r'<li>Bazinga result: .*>Test in progress.*'
+             r'href="http://bazinga.com/foo">results',
+             '<li>Not considered'])
+
+        # 'foo_1.1' source candidate was announced.
+        self.assertEqual(
+            [{'source_name': 'foo',
+              'source_version': '1.1',
+              'source_binaries': ['foo-bin'],
+              'series': self.data.series,
+              'distribution': self.data.distribution,
+            }], list(self.getAnnouncements()))
+
+    def test_promoted(self):
+        # When all required tests passed (or were skipped) the source
+        # candidate can be promoted.
+
+        # Make 'bazinga' and 'zoing' required test.
+        self.overrideConfig({
+            'TESTCLIENT_REQUIRED_TESTS': 'bazinga zoing',
+        })
+
+        # Publish 'in-progress' results for 'bazinga for "foo_1.1"'.
+        test_results = [{
+            'source_name': 'foo',
+            'source_version': '1.1',
+            'series': self.data.series,
+            'distribution': self.data.distribution,
+            'test_name': 'bazinga',
+            'test_status': 'SKIP',
+            'test_url': 'http://bazinga.com/foo',
+        }, {
+            'source_name': 'foo',
+            'source_version': '1.1',
+            'series': self.data.series,
+            'distribution': self.data.distribution,
+            'test_name': 'zoing',
+            'test_status': 'PASS',
+            'test_url': 'http://zoing.com/foo',
+        }]
+        self.publishTestResults(test_results)
+
+        context = [
+            ('foo-bin', {'Source': 'foo', 'Version': '1.1',
+                         'Architecture': 'amd64'}),
+        ]
+        self.do_test(
+            context,
+            [r'\bfoo\b.*>1</a> to .*>1.1<',
+             r'<li>Bazinga result: .*>Test skipped.*'
+             'href="http://bazinga.com/foo">results',
+             r'<li>Zoing result: .*>Pass.*href="http://zoing.com/foo">results',
+             '<li>Valid candidate'])
+
+    def test_hinted(self):
+        # 'Testclient' promotion respect 'force' and 'force-badtest' hints.
+        # 'force-skiptest' does not fit testclient approach, since it has
+        # no visibility of which tests 'will' run for a new source candidate,
+        # this decision belongs to the test-agents.
+
+        self.overrideConfig({
+            'TESTCLIENT_REQUIRED_TESTS': 'bazinga',
+        })
+        test_results = [{
+            'source_name': 'foo',
+            'source_version': '1.1',
+            'series': self.data.series,
+            'distribution': self.data.distribution,
+            'test_name': 'bazinga',
+            'test_status': 'FAIL',
+            'test_url': 'http://bazinga.com/foo',
+        }]
+        self.publishTestResults(test_results)
+
+        context = [
+            ('foo-bin', {'Source': 'foo', 'Version': '1.1',
+                         'Architecture': 'amd64'}),
+        ]
+        self.create_hint('cjwatson', 'force-badtest foo/1.1')
+        self.do_test(
+            context,
+            [r'\bfoo\b.*>1</a> to .*>1.1<',
+             r'<li>Bazinga result: .*>Regression.*'
+             'href="http://bazinga.com/foo">results',
+             r'<li>Should wait for foo 1.1 Bazinga test, but forced '
+             'by cjwatson',
+             '<li>Valid candidate'])
+
+
+if __name__ == '__main__':
+    unittest.main()