Merge lp:~coreygoldberg/uci-engine/subunit-results into lp:uci-engine

[Corey Goldberg]

WIP

[lp:~coreygoldberg/uci-engine/subunit-results]

668. By Corey Goldberg on 2014-07-21

reshuffled files layout and updated readme and docs

669. By Corey Goldberg on 2014-07-21

added tox and fixed setup

670. By Corey Goldberg on 2014-07-24

more updates

671. By Corey Goldberg on 2014-07-24

more unit tests

672. By Corey Goldberg on 2014-07-24

more unit tests

673. By Corey Goldberg on 2014-07-24

unit tests for fetch_subunit

674. By Corey Goldberg on 2014-07-24

removed comment

675. By Corey Goldberg on 2014-07-24

removed test dep

676. By Corey Goldberg on 2014-07-24

removed dep from uci_tests

677. By Corey Goldberg on 2014-07-27

subunit parser initial checkin with test file

678. By Corey Goldberg on 2014-07-27

subunit parser initial checkin with test file

679. By Corey Goldberg on 2014-07-27

unit tests passing

[Thomi Richards (thomir-deactivatedaccount)]

Hi Corey,

I'll preface this by saying that I'm sick, and my brain isn't working as well as it usually does, so if any of this doesn't make sense, there's a very real chance it's my fault - please feel free to ask any clarifying questions if that's the case.

The code is mostly good, but:

There are a few places that are missing implementation.
I've made a few suggestions throughout the code.
I haven't yet run the coverage metrics - at this stage you should be hitting 100% coverage, or very close to it.

I suggest you implement the missing pieces, so this can be deployed with the rest of the CI engine, and get reviews from the rest of the CI team.

Cheers,

[lp:~coreygoldberg/uci-engine/subunit-results]

680. By Corey Goldberg on 2014-07-27

first py3 port of ci_utils subset

681. By Corey Goldberg on 2014-07-27

merged trunk

682. By Corey Goldberg on 2014-07-28

removed unused modules from ci_utils after fork

683. By Corey Goldberg on 2014-07-28

fixed failiong json unit tests

684. By Corey Goldberg on 2014-07-28

removed unused module

685. By Corey Goldberg on 2014-07-28

test fixes

686. By Corey Goldberg on 2014-07-28

updated tox

687. By Corey Goldberg on 2014-07-29

updated tox using coverage

688. By Corey Goldberg on 2014-07-30

renaming and more unit tests

689. By Corey Goldberg on 2014-07-31

updated FakeDataStore and tests to only accept bytes

690. By Corey Goldberg on 2014-08-04

results summary implementation

691. By Corey Goldberg on 2014-08-06

updated test

692. By Corey Goldberg on 2014-08-06

updated README

693. By Corey Goldberg on 2014-08-06

changed file permissions

694. By Corey Goldberg on 2014-08-07

create_unique_name added

695. By Corey Goldberg on 2014-08-07

create_unique_name fixed

696. By Corey Goldberg on 2014-08-09

fixed test_details and summary

697. By Corey Goldberg on 2014-08-13

many fixes to implementation

698. By Corey Goldberg on 2014-08-13

pep8 fixes

699. By Corey Goldberg on 2014-08-13

re-merged trunk

700. By Corey Goldberg on 2014-08-18

queue qorker

701. By Corey Goldberg on 2014-08-21

updates to swift storage tests and functions

Unmerged revisions

701. By Corey Goldberg on 2014-08-21

updates to swift storage tests and functions

700. By Corey Goldberg on 2014-08-18

queue qorker

699. By Corey Goldberg on 2014-08-13

re-merged trunk

698. By Corey Goldberg on 2014-08-13

pep8 fixes

697. By Corey Goldberg on 2014-08-13

many fixes to implementation

696. By Corey Goldberg on 2014-08-09

fixed test_details and summary

695. By Corey Goldberg on 2014-08-07

create_unique_name fixed

694. By Corey Goldberg on 2014-08-07

create_unique_name added

693. By Corey Goldberg on 2014-08-06

changed file permissions

692. By Corey Goldberg on 2014-08-06

updated README

=== modified file '.bzrignore'
--- .bzrignore	2014-06-04 09:28:06 +0000
+++ .bzrignore	2014-08-21 14:14:59 +0000
@@ -19,3 +19,7 @@
 charms/precise/*/hooks/charmhelpers
 # The webui copies what it needs here to be deployed with the charm.
 charms/precise/webui/files/webroot
+# ignore virtualenvs created by tox
+.tox/
+# ignore coverage
+.coverage

=== added file 'docs/subunit-results.rst'
--- docs/subunit-results.rst	1970-01-01 00:00:00 +0000
+++ docs/subunit-results.rst	2014-08-21 14:14:59 +0000
@@ -0,0 +1,31 @@
+++++++++++++++++++++++
+SubUnit Results System
+++++++++++++++++++++++
+
+`subunitresults` is a Python3 application that extracts SubUnit test results
+and attachments, and makes them available via RESTful interface.
+
+SubUnit is a streaming binary protocol for test results (testcase status events
+and file attachments).  Individual status events are aggregated and used to
+create a test suite summary.  File attachments are included to provide rich
+detail about the nature of a failure.  File attachments can also be used to
+encapsulate stdout and stderr both during and outside tests.
+
+----
+
+The Subunit Results System has 2 basic parts:
+
+1. **Subunit Extractor Service** : A Python3/RabbitMQ Worker that listens for
+   test result messages on a queue. The message payload contains a URL to the
+   Swift location containing a SubUnit result file.  The SubUnit file is
+   fetched, and the service decodes the binary stream into:
+
+  * a JSON test suite summary
+  * arbitrary test attachments (files of supplied mime-type)
+    summary + attachments are sent to Swift data store.
+
+  The files it generates are stored in Swift as they are extracted.
+
+
+2. **Test Result Details API** : A Python3/Pyramid app that exposes a RESTful
+   interface to the test summary and test attachments stored in Swift.

=== added file 'juju-deployer/subunit-results.yaml.tmpl'
--- juju-deployer/subunit-results.yaml.tmpl	1970-01-01 00:00:00 +0000
+++ juju-deployer/subunit-results.yaml.tmpl	2014-08-21 14:14:59 +0000
@@ -0,0 +1,1 @@
+

=== added directory 'subunit-results'
=== added file 'subunit-results/README'
--- subunit-results/README	1970-01-01 00:00:00 +0000
+++ subunit-results/README	2014-08-21 14:14:59 +0000
@@ -0,0 +1,52 @@
+======================
+SubUnit Results System
+======================
+
+`subunitresults` is a Python3 package, and is a component of the Ubuntu CI
+Engine (lp:uci-engine).
+
+----------
+More Info:
+----------
+
+documentation in reStructuredText:
+
+ * /docs/subunit-results.rst
+
+published uci-engine documentation:
+
+ * http://uci.readthedocs.org/en/latest/subunit-results.html
+
+---------------
+Python Version:
+---------------
+
+ * Python 3.4
+
+--------------------
+Use the source Luke:
+--------------------
+
+  $ bzr branch lp:uci-engine
+  $ cd uci-engine/subunit-results
+
+--------------
+Run the Tests:
+--------------
+
+To run the tests directly from a development branch, you can use Tox.  Tox is
+a generic virtualenv management and test command line tool.  See `tox.ini` for
+current configuration.  Tox info: https://testrun.org/tox
+
+ * install Tox:
+
+  $ sudo apt-get install python-tox
+
+ * grab a branch of `uci-engine` from Launchpad:
+
+  $ bzr branch lp:uci-engine mybranch
+  $ cd mybranch/subunit-results
+
+ * run the tests:
+
+  $ tox

=== added file 'subunit-results/setup.py'
--- subunit-results/setup.py	1970-01-01 00:00:00 +0000
+++ subunit-results/setup.py	2014-08-21 14:14:59 +0000
@@ -0,0 +1,38 @@
+#!/usr/bin/env python
+# Ubuntu CI Engine
+# Copyright 2014 Canonical Ltd.
+
+# This program is free software: you can redistribute it and/or modify it
+# under the terms of the GNU Affero General Public License version 3, as
+# published by the Free Software Foundation.
+
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+
+from setuptools import setup, find_packages
+
+
+requires = [
+    'amqplib==1.0.2',
+    'pyramid==1.5.1',
+    'python-swiftclient==2.2.0',
+    'PyYAML==3.10',
+    'python-subunit==0.0.18',
+    'testtools==0.9.35',
+    "ucitests==0.1.4",
+]
+
+
+setup(
+    name='subunitresults',
+    version='0.1',
+    packages=find_packages(),
+    description='SubUnit Results System',
+    install_requires=requires,
+)

=== added directory 'subunit-results/subunitresults'
=== added file 'subunit-results/subunitresults/__init__.py'
--- subunit-results/subunitresults/__init__.py	1970-01-01 00:00:00 +0000
+++ subunit-results/subunitresults/__init__.py	2014-08-21 14:14:59 +0000
@@ -0,0 +1,41 @@
+#!/usr/bin/python3
+# Ubuntu CI Engine
+# Copyright 2014 Canonical Ltd.
+
+# This program is free software: you can redistribute it and/or modify it
+# under the terms of the GNU Affero General Public License version 3, as
+# published by the Free Software Foundation.
+
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+
+from wsgiref.simple_server import make_server
+from pyramid.config import Configurator
+
+from subunitresults.api import v1 as api_v1
+
+
+def make_wsgi_app():
+    config = Configurator()
+    # Instead of configuring all the routes here, we defer to the individual
+    # API version. This allows us to easily add 'v2' in the future. The actual
+    # routes are added in the 'configure_routes' callable.
+    #
+    # The root app is served at '/api/v1/' - note the trailing '/'.
+    config.include(api_v1.configure_routes, route_prefix='/api/v1')
+    config.scan()
+    return config.make_wsgi_app()
+
+
+app = make_wsgi_app()
+
+
+if __name__ == '__main__':
+    server = make_server('0.0.0.0', 8000, app)
+    server.serve_forever()

=== added file 'subunit-results/subunitresults/__main__.py'
--- subunit-results/subunitresults/__main__.py	1970-01-01 00:00:00 +0000
+++ subunit-results/subunitresults/__main__.py	2014-08-21 14:14:59 +0000
@@ -0,0 +1,26 @@
+#!/usr/bin/env python3
+# Ubuntu CI Engine
+# Copyright 2014 Canonical Ltd.
+#
+# This program is free software: you can redistribute it and/or modify it
+# under the terms of the GNU Affero General Public License version 3, as
+# published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+
+from subunitresults.queue_worker import launch_amqp_listener
+
+
+def main():
+    launch_amqp_listener()
+
+
+if __name__ == '__main__':
+    main()

=== added directory 'subunit-results/subunitresults/ci_utils'
=== added file 'subunit-results/subunitresults/ci_utils/__init__.py'
--- subunit-results/subunitresults/ci_utils/__init__.py	1970-01-01 00:00:00 +0000
+++ subunit-results/subunitresults/ci_utils/__init__.py	2014-08-21 14:14:59 +0000
@@ -0,0 +1,51 @@
+# Ubuntu CI Engine
+# Copyright 2013, 2014 Canonical Ltd.
+
+# This program is free software: you can redistribute it and/or modify it
+# under the terms of the GNU Affero General Public License version 3, as
+# published by the Free Software Foundation.
+
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import shutil
+import tempfile
+
+
+SUPPORTED_SERIES = [
+    'precise',
+    'trusty',
+    'utopic',
+]
+
+
+SUPPORTED_POCKETS = {
+    '': 'Release',
+    # '-updates': 'Updates',
+}
+
+
+def valid_suites():
+    suites = []
+    for pocket in SUPPORTED_POCKETS.keys():
+        for series in SUPPORTED_SERIES:
+            suites.append('{}{}'.format(series, pocket))
+    return suites
+
+
+class TmpDir(object):
+    '''Create a temporary directory with a context manager'''
+
+    def __init__(self, suffix='', prefix='tmp', dir=None):
+        self.path = tempfile.mkdtemp(suffix, prefix, dir)
+
+    def __enter__(self):
+        return self.path
+
+    def __exit__(self, exc, value, tb):
+        shutil.rmtree(self.path)

=== added file 'subunit-results/subunitresults/ci_utils/amqp_utils.py'
--- subunit-results/subunitresults/ci_utils/amqp_utils.py	1970-01-01 00:00:00 +0000
+++ subunit-results/subunitresults/ci_utils/amqp_utils.py	2014-08-21 14:14:59 +0000
@@ -0,0 +1,203 @@
+# Ubuntu CI Engine
+# Copyright 2013, 2014 Canonical Ltd.
+
+# This program is free software: you can redistribute it and/or modify it
+# under the terms of the GNU Affero General Public License version 3, as
+# published by the Free Software Foundation.
+
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import json
+import logging
+import os
+import socket
+import time
+
+from amqplib import client_0_8 as amqp
+
+BSBUILDER_QUEUE = 'bsbuilder'
+IMAGE_BUILDER_QUEUE = 'imagebuilder'
+TEST_RUNNER_QUEUE = 'test_runner'
+PUBLISHER_QUEUE = 'publisher'
+LANDER_QUEUE = 'lander'
+
+MONITORED_QUEUES = [
+    BSBUILDER_QUEUE,
+    IMAGE_BUILDER_QUEUE,
+    TEST_RUNNER_QUEUE,
+    PUBLISHER_QUEUE,
+    LANDER_QUEUE,
+]
+
+log = logging.getLogger(__name__)
+
+
+HERE = os.path.abspath(os.path.dirname(__file__))
+
+
+def get_config():
+    '''Load the rabbit config created by the charm'''
+    config = None
+    try:
+        f = os.path.join(HERE, '../../../../amqp_config.py')
+        # TODO remove once all charms have converted to new code layout
+        if not os.path.exists(f):
+            f = os.path.join(HERE, '../../amqp_config.py')
+        if os.path.exists(f):
+            import imp
+            config = imp.load_source('amqp_config', f)
+        else:
+            log.warn('No amqp_config found at: %s' % os.path.abspath(f))
+    except:
+        log.exception('ERROR detecting rabbit args')
+    return config
+
+
+def connection(rabbit_config):
+    return amqp.Connection(
+        userid=rabbit_config.AMQP_USER,
+        virtual_host=rabbit_config.AMQP_VHOST,
+        host=rabbit_config.AMQP_HOST,
+        password=rabbit_config.AMQP_PASSWORD
+    )
+
+
+def declare_queue(queue_name, config=None):
+    if config is None:
+        config = get_config()
+    conn = channel = None
+    try:
+        conn = connection(config)
+        channel = conn.channel()
+        channel.queue_declare(queue=queue_name, durable=True,
+                              auto_delete=False)
+    except Exception as e:
+        if channel:
+            channel.close()
+        if conn:
+            conn.close()
+        raise e
+    return conn, channel
+
+
+def send(queue_name, msg, raise_errors=False):
+    '''send a message to the given queue.
+
+    :param raise_errors: There are two users of this function. Some users are
+    sending non-critical messages in which errors can safely be ignored. Other
+    users need to catch the exception when this function fails.
+    '''
+    config = get_config()
+    if not config:
+        return 'rabbitmq settings not available.'
+    conn = channel = None
+    try:
+        conn, channel = declare_queue(queue_name, config)
+        body = amqp.Message(msg)
+        body.properties['delivery_mode'] = 2  # Persistent
+        channel.basic_publish(body, exchange='', routing_key=queue_name)
+    except Exception as e:
+        if raise_errors:
+            raise
+        logging.exception('unable to queue up message: %s', e)
+        return str(e)
+    finally:
+        if channel:
+            channel.close()
+        if conn:
+            conn.close()
+    return None
+
+
+def _run_forever(channel, queue, callback, retry_period=120):
+    logging.info('Waiting for messages. ^C to exit.')
+    tag = channel.basic_consume(callback=callback, queue=queue)
+    try:
+        timeout = time.time()
+        while channel.callbacks and time.time() < timeout + retry_period:
+            try:
+                channel.wait()
+                timeout = time.time()
+            except (amqp.AMQPConnectionException, socket.error):
+                logging.error('lost connection to Rabbit')
+                # TODO metrics.meter('lost_rabbit_connection')
+                # Don't probe immediately, give the network/process
+                # time to come back.
+                time.sleep(0.1)
+        if channel.callbacks:
+            logging.error('Rabbit did not reappear quickly enough.')
+    except KeyboardInterrupt:
+        pass
+    finally:
+        if channel and channel.callbacks and channel.is_open:
+            channel.basic_cancel(tag)
+
+
+def process_queue(config, queue_name, callback, delete=False):
+    conn = channel = None
+    try:
+        conn, channel = declare_queue(queue_name, config)
+        channel.basic_qos(0, 1, False)
+        _run_forever(channel, queue_name, callback)
+    finally:
+        if channel:
+            if delete:
+                channel.queue_delete(queue_name)
+            channel.close()
+        if conn:
+            conn.close()
+
+
+def progress_waiting(progress_trigger, data=None):
+    if data is None:
+        data = {}
+    data['state'] = 'WAITING'
+    return send(progress_trigger, json.dumps(data))
+
+
+_ratelimit = {'trigger': None, 'timestamp': 0, 'delay': 5}
+
+
+def progress_update(progress_trigger, data=None, ratelimit=False):
+    if ratelimit:
+        now = time.time()
+        if _ratelimit['trigger'] == progress_trigger:
+            if now < _ratelimit['timestamp'] + _ratelimit['delay']:
+                # too soon
+                return
+            else:
+                # we'll allow the message to be logged, but we should back
+                # off a little more before showing the same thing:
+                if _ratelimit['delay'] < 20 * 60:  # max throttle = 20 min
+                    _ratelimit['delay'] *= 2
+        _ratelimit['trigger'] = progress_trigger
+        _ratelimit['timestamp'] = now
+
+    if data is None:
+        data = {}
+    data['state'] = 'STATUS'
+    return send(progress_trigger, json.dumps(data))
+
+
+def progress_completed(progress_trigger, data=None):
+    if data is None:
+        data = {}
+    data['state'] = 'COMPLETED'
+    data['result'] = 'PASSED'
+    data['exit'] = True
+    return send(progress_trigger, json.dumps(data))
+
+
+def progress_failed(progress_trigger, data=None):
+    if data is None:
+        data = {}
+    data['state'] = 'COMPLETED'
+    data['result'] = 'FAILED'
+    data['exit'] = True
+    return send(progress_trigger, json.dumps(data))

=== added file 'subunit-results/subunitresults/ci_utils/amqp_worker.py'
--- subunit-results/subunitresults/ci_utils/amqp_worker.py	1970-01-01 00:00:00 +0000
+++ subunit-results/subunitresults/ci_utils/amqp_worker.py	2014-08-21 14:14:59 +0000
@@ -0,0 +1,257 @@
+# Ubuntu CI Engine
+# Copyright 2014 Canonical Ltd.
+
+# This program is free software: you can redistribute it and/or modify it
+# under the terms of the GNU Affero General Public License version 3, as
+# published by the Free Software Foundation.
+
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import fcntl
+import json
+import logging
+import os
+import io
+import subprocess
+import sys
+import threading
+import time
+import traceback
+import urllib.error
+import urllib.parse
+import urllib.request
+
+from ci_utils import (
+    amqp_utils,
+    data_store,
+    dump_stack,
+    json_status,
+    unit_config,
+)
+
+
+HERE = os.path.abspath(os.path.dirname(__file__))
+logging.basicConfig(level=logging.INFO)
+log = logging.getLogger(__name__)
+
+
+def _get(url):
+    try:
+        resp = urllib.request.urlopen(url)
+        resp = resp.read()
+        return json.loads(resp)
+    except:
+        logging.exception('error checking url')
+        return {}
+
+
+class _timer(threading.Thread):
+    '''similar to threading.Timer, but repeats until cancel is called'''
+    def __init__(self, interval, cb):
+        super(_timer, self).__init__()
+        self.interval = interval
+        self.cb = cb
+        self.finished = threading.Event()
+        self.exit = False
+
+    def cancel(self):
+        """Stop the timer if it hasn't finished yet"""
+        self.finished.set()
+        self.exit = True
+
+    def run(self):
+        while not self.exit:
+            self.finished.wait(self.interval)
+            if not self.finished.is_set():
+                self.cb()
+        self.finished.set()
+
+
+class AMQPWorker(object):
+    '''Base class that handles the more complex issues of a rabbit worker.
+
+    This allows a user to subclass this and provide a custom "handle_request"
+    method.
+    '''
+
+    def __init__(self, logger_name):
+        self.logger_name = logger_name
+        self.last_check_file = '/tmp/worker-data-' + logger_name + '.json'
+
+    # how often to check if the current task should be cancelled
+    cancel_interval = 120
+
+    def handle_request(self, params, logger):
+        '''To be implemented by the subclass to do their work
+
+        amqp_cb should either be one of:
+          * amqp_utils.progress_failed
+          * amqp_utils.progress_completed
+        :return: tuple(ampq_cb, return_value_dictionary)
+        '''
+        raise NotImplementedError
+
+    def main(self, queue):
+        dump_stack.install_stack_dump_signal()
+        config = amqp_utils.get_config()
+        if not config:
+            sys.exit(1)  # the get_config code prints an error
+
+        json_status.run_server(self.engine_health)
+        amqp_utils.process_queue(config, queue, self._on_message)
+
+    def engine_health(self):
+        status = json_status.JSONStatus()
+        status.add_deployed_revision()
+        status.add_data_store_health()
+        return status
+
+    def get_last_check(self):
+        check = None
+        # use a flock to ensure concurrent reader/writers are safe
+        with open(self.last_check_file, 'a+') as f:
+            try:
+                fcntl.lockf(f, fcntl.LOCK_EX)
+                f.seek(0, os.SEEK_END)
+                # If the file exists but has zero length,
+                # the check hasn't been run.
+                if f.tell() > 0:
+                    f.seek(0)
+                    check = json.load(f)
+            finally:
+                fcntl.lockf(f, fcntl.LOCK_UN)
+        return check
+
+    def save_last_check(self, results):
+        '''used to store results from our "check" scripts
+
+        This file can also be analyzed by our `engine_health` api to show
+        if the last check on this component worked.
+        '''
+        # use a flock to ensure concurrent reader/writers are safe
+        with open(self.last_check_file, 'a+') as f:
+            try:
+                fcntl.lockf(f, fcntl.LOCK_EX)
+                f.seek(0)
+                f.truncate()
+                # we can't depend on the "mtime" attribute of the file since
+                # we have to open in it read-write mode for flocking, so we
+                # store a timestamp in the file itself
+                if 'last_run' not in results:
+                    results['last_run'] = time.time()
+                json.dump(results, f)
+            finally:
+                fcntl.lockf(f, fcntl.LOCK_UN)
+
+    def _create_worker_logger(self):
+        '''Create a logger that captures output into StringIO.
+
+        This provides an easy way to let workers get free logging that will
+        get added to the ticket.
+        '''
+        log = logging.getLogger(self.logger_name)
+        log.buffer = io.StringIO()
+        logstream = logging.StreamHandler(log.buffer)
+        formatter = logging.Formatter(
+            '[%(asctime)s] %(name)s:%(levelname)s:%(message)s')
+        logstream.setFormatter(formatter)
+        logstream.setLevel(logging.INFO)
+        log.addHandler(logstream)
+        return log
+
+    def _create_data_store(self, ticket):
+        # Note that this may be called multiple times
+        return data_store.create_for_ticket(
+            ticket, unit_config.get_auth_config())
+
+    def _store_worker_log(self, store, logger, retval):
+        '''An exception safe mechanism to upload log files.
+
+        This needs to catch any exception so that progress complete/fail
+        call can be made on the queue.
+        '''
+        if not store or not logger:
+            return
+        content = logger.buffer.getvalue()
+        if not content:
+            log.info('no logging content from action, skipping upload')
+        try:
+            name = '{}.output.log'.format(logger.name)
+            url = store.put_file(name, content, 'text/plain')
+            self._create_artifact(name, url, retval)
+        except:
+            log.exception('unable to store worker log')
+
+    def _create_artifact(self, name, url, retval, type='LOGS'):
+        retval.setdefault('artifacts', []).append({
+            'name': name,
+            'reference': url,
+            'type': type,
+        })
+
+    def _handle_cancel(self, params):
+        '''provide an advisory mechansim to inform the worker to stop
+
+        This checks a URL to determine if the job has been cancelled. If so,
+        it updates the "params" object with a params['cancelled'] = True. Thus
+        this is an advisory mechansim that the worker can choose how/if to
+        deal with.
+        '''
+        url = params.get('cancel_url')
+        if not url:
+            return None
+
+        def check_url():
+            data = _get(url)
+            if not data.get('building', True):
+                params['cancelled'] = True
+        t = _timer(self.cancel_interval, check_url)
+        t.start()
+        return t
+
+    def _handle_exception(self, e, store, trigger, logger, ret):
+        try:
+            err = 'Unexpected exception occurred'
+            if isinstance(e, KeyboardInterrupt):
+                err = 'Worker terminated'
+            elif isinstance(e, subprocess.CalledProcessError) and logger:
+                err = 'error running {}: {}'.format(e.cmd, e.output)
+            ret['error_message'] = err
+            ret['traceback'] = traceback.format_exc()
+            if logger:
+                logger.exception(err)
+            self._store_worker_log(store, logger, ret)
+        finally:
+            if trigger:
+                amqp_utils.progress_failed(trigger, ret)
+
+    def _on_message(self, msg):
+        log.info('on_message: %s', msg.body)
+        ret = {}
+        store = logger = trigger = cancel_thread = None
+        try:
+            logger = self._create_worker_logger()
+            params = json.loads(msg.body)
+            trigger = params['progress_trigger']
+            amqp_utils.progress_update(trigger, params)
+            # Create the ticket specific data store
+            store = self._create_data_store(params['ticket_id'])
+            cancel_thread = self._handle_cancel(params)
+
+            amqp_cb, ret = self.handle_request(params, logger)
+            self._store_worker_log(store, logger, ret)
+            amqp_cb(trigger, ret)
+        except (KeyboardInterrupt, Exception) as e:
+            self._handle_exception(e, store, trigger, logger, ret)
+            if isinstance(e, KeyboardInterrupt):
+                raise  # re-raise so amqp_utils.process_queue can exit
+        finally:
+            msg.channel.basic_ack(msg.delivery_tag)
+            if cancel_thread:
+                cancel_thread.cancel()

=== added file 'subunit-results/subunitresults/ci_utils/data_store.py'
--- subunit-results/subunitresults/ci_utils/data_store.py	1970-01-01 00:00:00 +0000
+++ subunit-results/subunitresults/ci_utils/data_store.py	2014-08-21 14:14:59 +0000
@@ -0,0 +1,168 @@
+# Ubuntu CI Engine
+# Copyright 2013, 2014 Canonical Ltd.
+
+# This program is free software: you can redistribute it and/or modify it
+# under the terms of the GNU Affero General Public License version 3, as
+# published by the Free Software Foundation.
+
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import logging
+import os
+
+from swiftclient import client
+from swiftclient.client import ClientException
+
+
+class DataStoreException(Exception):
+    pass
+
+
+def create_for_ticket(ticket_id, auth_config, public=True):
+    return DataStore('ticket-{}'.format(ticket_id), auth_config, public=public)
+
+
+def _get_file_name(filename):
+    return os.path.basename(filename)
+
+
+class DataStore(object):
+
+    def __init__(self, component, auth_config, public=False):
+        self.container_id = component
+        self.auth_config = auth_config
+        self.public = public
+        self.client = None
+
+    @staticmethod
+    def validate_auth_config(auth_config):
+        required_fields = ['auth_user', 'auth_password', 'auth_tenant_name',
+                           'auth_url', 'auth_region']
+
+        missing_fields = []
+
+        for field in required_fields:
+            if field not in list(auth_config.keys()):
+                missing_fields.append(field)
+
+        if len(missing_fields) > 0:
+            raise DataStoreException(
+                "missing fields: {}".format(', '.join(missing_fields))
+            )
+
+    def ensure_swift_client(self):
+        if self.client is not None:
+            return self.client
+
+        self.validate_auth_config(self.auth_config)
+        config = self.auth_config
+        self.client = client.Connection(
+            authurl=config.get('auth_url'),
+            user=config.get('auth_user'),
+            key=config.get('auth_password'),
+            os_options={'tenant_name': config.get('auth_tenant_name'),
+                        'region_name': config.get('auth_region')},
+            auth_version='2.0')
+        try:
+            self._create_container(self.container_id)
+        except client.ClientException:
+            logging.exception('Unable to connect to swift')
+            raise DataStoreException("Missing or invalid authentication info.")
+        self.change_visibility(public=self.public)
+        self.container_url = "{}/{}".format(self.client.url, self.container_id)
+
+    def list_files(self):
+        self.ensure_swift_client()
+        _, objects = self.client.get_container(self.container_id)
+        return [x['name'] for x in objects]
+
+    def put_file(self, filename, contents, content_type=None):
+        self.ensure_swift_client()
+        name = _get_file_name(filename)
+        try:
+            self.client.put_object(self.container_id, obj=name,
+                                   contents=contents,
+                                   content_type=content_type)
+        except ClientException as e:
+            raise DataStoreException(
+                "Failed to upload file: {}, Error: {}".format(filename, e)
+            )
+
+        return self.file_path(filename)
+
+    def get_file(self, filename):
+        self.ensure_swift_client()
+        name = _get_file_name(filename)
+
+        contents = None
+        try:
+            hdr, contents = self.client.get_object(self.container_id, obj=name)
+        except ClientException as e:
+            raise DataStoreException(
+                "Failed to get file: {}, Error: {}".format(filename, e)
+            )
+        return contents
+
+    def change_visibility(self, public=False):
+        self.ensure_swift_client()
+        if public:
+            read_acl = '.r:*'
+        else:
+            read_acl = ''
+        self.client.post_container(self.container_id,
+                                   {'X-Container-Read': read_acl})
+        self.public = public
+
+    def delete_file(self, filename):
+        self.ensure_swift_client()
+        name = _get_file_name(filename)
+        try:
+            self.client.delete_object(self.container_id, obj=name)
+        except ClientException as e:
+            raise DataStoreException(
+                "Failed to delete file: {}, Error: {}".format(filename, e)
+            )
+
+    def clear(self):
+        self.ensure_swift_client()
+        try:
+            files = self.client.get_container(self.container_id)[1]
+
+            for f in files:
+                try:
+                    self.client.delete_object(self.container_id,
+                                              obj=f['name'])
+                except ClientException as e:
+                    raise DataStoreException(
+                        "Failed to delete file: {}".format(e)
+                    )
+        except ClientException as e:
+            raise DataStoreException(
+                "Failed to get container: {}".format(e)
+            )
+
+    def delete(self, recursive=False):
+        self.ensure_swift_client()
+        try:
+            if recursive:
+                self.clear()
+
+            self.client.delete_container(self.container_id)
+        except ClientException as e:
+            raise DataStoreException(
+                "Failed to delete container: {}".format(e)
+            )
+
+    def file_path(self, filename):
+        name = _get_file_name(filename)
+
+        return "{}/{}".format(self.container_url, name)
+
+    def _create_container(self, container):
+        self.client.put_container(self.container_id)

=== added file 'subunit-results/subunitresults/ci_utils/dump_stack.py'
--- subunit-results/subunitresults/ci_utils/dump_stack.py	1970-01-01 00:00:00 +0000
+++ subunit-results/subunitresults/ci_utils/dump_stack.py	2014-08-21 14:14:59 +0000
@@ -0,0 +1,41 @@
+# Ubuntu CI Engine
+# Copyright 2014 Canonical Ltd.
+
+# This program is free software: you can redistribute it and/or modify it
+# under the terms of the GNU Affero General Public License version 3, as
+# published by the Free Software Foundation.
+
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import threading
+import sys
+import traceback
+import signal
+
+
+def dumpstacks(signal, frame):
+    sys.stderr.writeline("")
+    sys.stderr.writeline("User-requested stack trace:")
+    sys.stderr.writeline("---------------------------")
+    id2name = dict([(th.ident, th.name) for th in threading.enumerate()])
+    code = []
+    for threadId, stack in list(sys._current_frames().items()):
+        tloc = "\n# Thread: %s(%d)" % (id2name.get(threadId, ""), threadId)
+        code.append(tloc)
+        for filename, lineno, name, line in traceback.extract_stack(stack):
+            if not filename.endswith('dump_stack.py'):
+                loc = 'File: "%s", line %d, in %s' % (filename, lineno, name)
+                code.append(loc)
+                if line:
+                    code.append("  %s" % (line.strip()))
+    sys.stderr.write("\n".join(code))
+
+
+def install_stack_dump_signal():
+    signal.signal(signal.SIGQUIT, dumpstacks)

=== added file 'subunit-results/subunitresults/ci_utils/json_status.py'
--- subunit-results/subunitresults/ci_utils/json_status.py	1970-01-01 00:00:00 +0000
+++ subunit-results/subunitresults/ci_utils/json_status.py	2014-08-21 14:14:59 +0000
@@ -0,0 +1,162 @@
+# Ubuntu Continuous Integration Engine
+# Copyright 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 version 3, as published by the
+# Free Software Foundation.
+
+# This program is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranties of MERCHANTABILITY,
+# SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+# General Public License for more details.
+
+# You should have received a copy of the GNU General Public License along with
+# this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import http.server
+import json
+import os
+import threading
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+
+
+HERE = os.path.abspath(os.path.dirname(__file__))
+
+# We have a few checks that take a while. The implementations for them use
+# caching logic. This gives us a consistent place to manage this value.
+CACHE_TIMEOUT = 60
+
+
+class JSONStatus(object):
+    def __init__(self):
+        self.results = []
+
+    def _add_result(self, name, value, status):
+        self.results.append({
+            'label': name,
+            'value': value,
+            'status': status,
+        })
+
+    def add_okay(self, name, value):
+        self._add_result(name, value, 'okay')
+
+    def add_fail(self, name, value):
+        self._add_result(name, value, 'fail')
+
+    def add_true_false(self, name, value, condition):
+        '''adds an 'okay' result if condition is true, 'fail' otherwise'''
+        if condition:
+            self.add_okay(name, value)
+        else:
+            self.add_fail(name, value)
+
+    def add_url_test(self, name, url):
+        try:
+            urllib.request.urlopen(url).read()
+            self.add_okay(name, 'connected')
+        except Exception as e:
+            self.add_fail(name, str(e))
+
+    def add_rabbit_configured(self):
+        '''adds a result status for status of rabbit-configuration'''
+        # we don't want to import this for all users of the API. Some
+        # may not install rabbit (eg the ppa-assigner)
+        from ci_utils import amqp_utils
+        if amqp_utils.get_config() is not None:
+            self.add_okay('rabbit configured', True)
+        else:
+            self.add_fail('rabbit configured', False)
+
+    def add_rabbit_worker_health(self, queue):
+        '''checks if there are any workers attached to the queue.'''
+        from ci_utils import amqp_utils
+        config = amqp_utils.get_config()
+        if config is None:
+            # we already report rabbit isn't configured, so no sense adding
+            # another failure for this
+            return
+
+        mgr = urllib.request.HTTPPasswordMgrWithDefaultRealm()
+        base = 'http://{}:55672'.format(config.AMQP_HOST)
+        # NOTE: juju leaves this guest/guest and doesn't seem to expose
+        # anything in the relation to override it
+        mgr.add_password(None, base, 'guest', 'guest')
+        handler = urllib.request.HTTPBasicAuthHandler(mgr)
+
+        try:
+            resp = urllib.request.build_opener(handler).open(
+                base + '/api/queues'
+            )
+            content = resp.read()
+            data = json.loads(content)
+            consumers = None
+            for q in data:
+                if q['name'] == queue:
+                    consumers = q['consumers']
+                    break
+            if consumers is None:
+                self.add_fail('workers-online', 'no queue defined')
+            else:
+                self.add_true_false('workers-online', consumers, consumers > 0)
+        except:
+            self.add_fail('workers-online', 'unable to check')
+
+    def add_data_store_health(self, ticket=1):
+        from ci_utils import unit_config, data_store
+        try:
+            ds = data_store.create_for_ticket(
+                ticket, unit_config.get_auth_config())
+            ds.ensure_swift_client()
+            self.add_okay('data-store', 'connected')
+            return ds
+        except Exception as e:
+            self.add_fail('data-store', str(e))
+
+
+class StatusHandler(http.server.BaseHTTPRequestHandler):
+    # must be static because this object is create for each request
+    _cache = (None, 0)
+
+    def __init__(self, *args, **kwargs):
+        http.server.BaseHTTPRequestHandler.__init__(self, *args, **kwargs)
+
+    def _get_cached_response(self):
+        query = urllib.parse.urlparse(self.path).query
+        query = urllib.parse.parse_qs(query)
+
+        resp, timeout = StatusHandler._cache
+        if query.get('refresh', ['0'])[0] == '1' or time.time() > timeout:
+            resp = json.dumps(self.health_cb().results)
+            StatusHandler._cache = (resp, time.time() + CACHE_TIMEOUT)
+
+        if 'jsonp' in query.get('format', []):
+            cb = query.get('callback', ['cb'])[0]
+            resp = cb + '(' + resp + ')'
+
+        return resp.encode('utf-8')
+
+    def do_GET(self):
+        self.send_response(200)
+        self.send_header('Content-type', 'text/json')
+        self.end_headers()
+        resp = self._get_cached_response()
+        self.wfile.write(resp)
+
+    def log_message(self, format, *args):
+        return
+
+
+def run_server(health_cb, port=8080, threaded=True):
+    httpd = http.server.HTTPServer(('0.0.0.0', port), StatusHandler)
+    httpd.RequestHandlerClass.health_cb = health_cb
+    if threaded:
+        t = threading.Thread(target=httpd.serve_forever)
+        t.daemon = True
+        t.start()
+        return httpd
+    else:
+        httpd.serve_forever()

=== added directory 'subunit-results/subunitresults/ci_utils/testing'
=== added file 'subunit-results/subunitresults/ci_utils/testing/__init__.py'
--- subunit-results/subunitresults/ci_utils/testing/__init__.py	1970-01-01 00:00:00 +0000
+++ subunit-results/subunitresults/ci_utils/testing/__init__.py	2014-08-21 14:14:59 +0000
@@ -0,0 +1,15 @@
+# Ubuntu Continuous Integration Engine
+# Copyright 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 version 3, as published by the
+# Free Software Foundation.
+
+# This program is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranties of MERCHANTABILITY,
+# SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+# General Public License for more details.
+
+# You should have received a copy of the GNU General Public License along with
+# this program.  If not, see <http://www.gnu.org/licenses/>.

=== added file 'subunit-results/subunitresults/ci_utils/testing/fixtures.py'
--- subunit-results/subunitresults/ci_utils/testing/fixtures.py	1970-01-01 00:00:00 +0000
+++ subunit-results/subunitresults/ci_utils/testing/fixtures.py	2014-08-21 14:14:59 +0000
@@ -0,0 +1,120 @@
+# Ubuntu CI Engine
+# Copyright 2014 Canonical Ltd.
+
+# This program is free software: you can redistribute it and/or modify it
+# under the terms of the GNU Affero General Public License version 3, as
+# published by the Free Software Foundation.
+
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+
+import os
+
+from ucitests import fixtures
+
+from ci_utils import data_store
+
+
+# A data store defines a namespace for objects. The namespace is rooted at the
+# container id and objects are identified by a path inside the container.
+# It defines an API to store and retrieve objects
+
+
+class FakeDataStore(data_store.DataStore):
+    """A fake data store to avoid requiring a swift server.
+
+    This stores data on disk and as such should be
+    isolated. ucitests.fixtures.set_uniq_cwd() provides such isolation and
+    cleanup. It's the test responsibility to properly isolate itself.
+    """
+
+    def __init__(self, component, auth_config=None, public=False):
+        super(FakeDataStore, self).__init__(
+            # We use the container id as our backend directory and just store
+            # files there.
+            os.path.abspath(component),
+            # We don't use the config so we inject a fake but valid one
+            {
+                'auth_url': 'http://example.com',
+                'auth_user': 'user',
+                'auth_password': 'pass',
+                'auth_tenant_name': 'tenant',
+                'auth_region': 'region',
+            },
+            public=public
+        )
+        # The following call is unguarded to fail if one attempts to create the
+        # same data store twice.
+        os.mkdir(self.container_id)
+
+    def list_files(self):
+        return sorted(os.listdir(self.container_id))
+
+    def put_file(self, filename, contents, content_type=None):
+        full_path = os.path.join(self.container_id, filename)
+        with open(full_path, 'wb') as f:
+            f.write(contents)
+        return self.file_path(filename)
+
+    def get_file(self, filename):
+        full_path = os.path.join(self.container_id, filename)
+        with open(full_path, 'rb') as f:
+            return f.read()
+
+    def delete_file(self, filename):
+        full_path = os.path.join(self.container_id, filename)
+        os.remove(full_path)
+
+    def clear(self):
+        for f in self.list_files():
+            self.delete_file(f)
+
+    def delete(self, recursive=False):
+        self.clear()
+        os.rmdir(self.container_id)
+
+    def file_path(self, filename):
+        return '{}{}/{}'.format('file://', self.container_id, filename)
+
+
+class FakeTaskQueue(object):
+
+    def __init__(self, queue_name, config=None):
+        self.msgs = []
+        self.queue_name = queue_name
+        # We don't use the config which contains the credentials to access a
+        # real rabbit server.
+
+    def publish(self, msg):
+        self.msgs.append(msg)
+
+    def subscribe(self, func, *args, **kwargs):
+        returns = []
+        while not self.is_empty():
+            ret = func(self.msgs.pop(0), *args, **kwargs)
+            returns.append(ret)
+        return returns
+
+    def is_empty(self):
+        return not self.msgs
+
+
+def isolate_from_juju_env(test):
+    """Isolate a test from juju environment definitions.
+
+    This is usually called in setUp for tests that want to modify the juju
+    environement definitions and restore them after the test is run.
+
+    :param test: A test instance.
+    """
+    # Juju environments are defined in ${JUJU_HOME}/environments.yaml and can
+    # be specified via ${JUJU_ENV}
+    fixtures.set_uniq_cwd(test)
+    fixtures.isolate_from_env(
+        test, dict(HOME=test.uniq_dir, JUJU_HOME='.', JUJU_ENV=None))

=== added directory 'subunit-results/subunitresults/ci_utils/tests'
=== added file 'subunit-results/subunitresults/ci_utils/tests/__init__.py'
=== added file 'subunit-results/subunitresults/ci_utils/tests/test_amqp.py'
--- subunit-results/subunitresults/ci_utils/tests/test_amqp.py	1970-01-01 00:00:00 +0000
+++ subunit-results/subunitresults/ci_utils/tests/test_amqp.py	2014-08-21 14:14:59 +0000
@@ -0,0 +1,101 @@
+# Ubuntu CI Engine
+# Copyright 2013, 2014 Canonical Ltd.
+
+# This program is free software: you can redistribute it and/or modify it
+# under the terms of the GNU Affero General Public License version 3, as
+# published by the Free Software Foundation.
+
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import json
+import socket
+import time
+import unittest
+from unittest import mock
+
+from ci_utils import amqp_utils
+
+
+class TestAMQP(unittest.TestCase):
+    @mock.patch('ci_utils.amqp_utils.connection')
+    @mock.patch('ci_utils.amqp_utils.get_config')
+    def testConnectFailed(self, get_config, connect):
+        '''Ensure a failed queue connection returns an HTTP 503 error'''
+        get_config.return_value = mock.Mock()
+        error = 'mocked test exception'
+        connect.side_effect = RuntimeError(error)
+        r = amqp_utils.send('fake_queue', 'fake_message')
+        self.assertIsNotNone(r)
+        self.assertEqual(error, r)
+
+    @mock.patch('ci_utils.amqp_utils.connection')
+    @mock.patch('ci_utils.amqp_utils.get_config')
+    def testSent(self, get_config, connect):
+        '''Test a successful send returns nothing
+
+        There's not much you can test in isolation here, but this gives
+        a bit of a sanity check.
+        '''
+        get_config.return_value = mock.Mock()
+        connect.return_value = mock.Mock()
+        r = amqp_utils.send('fake_queue', 'fake_message')
+        self.assertIsNone(r)
+
+        connect.side_effect = RuntimeError('testSent')
+        r = amqp_utils.send('fake_queue', 'fake message')
+        self.assertIsNotNone(r)
+
+        with self.assertRaises(RuntimeError):
+            amqp_utils.send('fake_queue', 'fake message', True)
+
+    @mock.patch('ci_utils.amqp_utils._run_forever')
+    @mock.patch('ci_utils.amqp_utils.connection')
+    def testProcessQueue(self, connection, run_forever):
+        '''Ensure we close the connection if something fails'''
+        conn = mock.Mock()
+        connection.return_value = conn
+        run_forever.side_effect = RuntimeError
+        with self.assertRaises(RuntimeError):
+            amqp_utils.process_queue(None, None, None)
+        self.assertTrue(conn.close.called)
+
+    def testRunForever(self):
+        '''Ensure this times out after the right amount of time'''
+        channel = mock.Mock()
+        callback = mock.Mock()
+        channel.wait.side_effect = socket.error()
+        start = time.time()
+        retry_period = 0.1
+        amqp_utils._run_forever(channel, 'foo', callback, retry_period)
+        # see if it was within 1/100 of a second of the retry_period
+        self.assertAlmostEqual(start + retry_period, time.time(), places=2)
+
+
+class TestProgressTrigger(unittest.TestCase):
+    @mock.patch('ci_utils.amqp_utils.send')
+    def testProgress(self, send):
+        trigger = 'foo_trigger'
+        amqp_utils.progress_waiting(trigger)
+        send.assert_called_with(trigger, json.dumps({'state': 'WAITING'}))
+
+        send.reset_mock()
+        params = {'foo': 'bar'}
+        amqp_utils.progress_completed(trigger, params)
+        params['state'] = 'COMPLETED'
+        params['result'] = 'PASSED'
+        params['exit'] = True
+        send.assert_called_with(trigger, json.dumps(params))
+
+        send.reset_mock()
+        params = {'foo': 'bar'}
+        amqp_utils.progress_failed(trigger, params)
+        params['state'] = 'COMPLETED'
+        params['result'] = 'FAILED'
+        params['exit'] = True
+        send.assert_called_with(trigger, json.dumps(params))

=== added file 'subunit-results/subunitresults/ci_utils/tests/test_amqp_worker.py'
--- subunit-results/subunitresults/ci_utils/tests/test_amqp_worker.py	1970-01-01 00:00:00 +0000
+++ subunit-results/subunitresults/ci_utils/tests/test_amqp_worker.py	2014-08-21 14:14:59 +0000
@@ -0,0 +1,267 @@
+# Ubuntu CI Engine
+# Copyright 2014 Canonical Ltd.
+
+# This program is free software: you can redistribute it and/or modify it
+# under the terms of the GNU Affero General Public License version 3, as
+# published by the Free Software Foundation.
+
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import json
+import os
+import subprocess
+import time
+import unittest
+from unittest import mock
+
+from ci_utils import amqp_worker
+
+
+class _worker(amqp_worker.AMQPWorker):
+    '''Simple class to help mock the datastore api used in the worker.'''
+
+    def __init__(self, amqp_utils, retval, exception=None, error_msg=None):
+        super(_worker, self).__init__('queue-name')
+        self.store = mock.Mock()
+        if amqp_utils:
+            self.amqp_utils = amqp_utils
+        self.retval = retval
+        self.exception = exception
+        self.error_msg = error_msg
+
+    def _create_data_store(self, ticket):
+        # Always use our own store so the tests can assert on it
+        return self.store
+
+    def handle_request(self, params, log):
+        log.info('called')
+        self.params = params
+        if self.exception:
+            raise self.exception(self.error_msg)
+        if self.error_msg:
+            return self.amqp_utils.progress_failed, self.retval
+        if params.get('cancel_url'):
+            time.sleep(0.1)
+            self.retval['cancelled'] = params['cancelled']
+        return self.amqp_utils.progress_completed, self.retval
+
+
+class TestAMQPWorker(unittest.TestCase):
+    @mock.patch('ci_utils.amqp_worker.amqp_utils')
+    def testOnMessageSimple(self, amqp_utils):
+        '''Test simple on message logic'''
+        msg = mock.Mock()
+        msg.delivery_tag = 'foo'
+        msg.body = json.dumps({
+            'ticket_id': 1,
+            'progress_trigger': 'queue-name',
+            'param1': 'foo',
+            'param2': 42,
+        })
+        retval = {'msg': 'this is the worker return value'}
+        worker = _worker(amqp_utils, retval)
+        worker._on_message(msg)
+        self.assertEqual(worker.params['param1'], 'foo')
+        self.assertEqual(worker.params['param2'], 42)
+        self.assertEqual(1, amqp_utils.progress_update.call_count)
+        msg.channel.basic_ack.assert_called_once_with('foo')
+        self.assertEqual(0, amqp_utils.progress_failed.call_count)
+        self.assertEqual(1, worker.store.put_file.call_count)
+        amqp_utils.progress_completed.assert_called_once_with(
+            'queue-name', retval)
+        # assert_called_once_with works, but doesn't check the values
+        # added to retval
+        retval = amqp_utils.progress_completed.call_args[0][1]
+        self.assertEqual(1, len(retval['artifacts']))
+        self.assertEqual('queue-name.output.log',
+                         retval['artifacts'][0]['name'])
+
+    @mock.patch('ci_utils.amqp_worker.amqp_utils')
+    def testOnMessageFail(self, amqp_utils):
+        '''Test on message logic for a failure handled by the worker'''
+        msg = mock.Mock()
+        msg.delivery_tag = 'foo'
+        msg.body = json.dumps({
+            'ticket_id': 1,
+            'progress_trigger': 'queue-name',
+            'param1': 'foo',
+            'param2': 42,
+        })
+        retval = {'msg': 'this is the worker return value'}
+        worker = _worker(amqp_utils, retval, None, True)
+        worker._on_message(msg)
+        self.assertEqual(worker.params['param1'], 'foo')
+        self.assertEqual(worker.params['param2'], 42)
+        self.assertEqual(1, amqp_utils.progress_update.call_count)
+        msg.channel.basic_ack.assert_called_once_with('foo')
+        amqp_utils.progress_failed.assert_called_once_with(
+            'queue-name', retval)
+        self.assertEqual(0, amqp_utils.progress_completed.call_count)
+        # assert_called_once_with works, but doesn't check the values
+        # added to retval
+        retval = amqp_utils.progress_failed.call_args[0][1]
+        self.assertEqual(1, len(retval['artifacts']))
+        self.assertEqual('queue-name.output.log',
+                         retval['artifacts'][0]['name'])
+
+    @mock.patch('ci_utils.amqp_worker.amqp_utils')
+    def testOnMessageUnexpected(self, amqp_utils):
+        '''Test on message logic for an unexpected failure'''
+        msg = mock.Mock()
+        msg.delivery_tag = 'foo'
+        msg.body = json.dumps({
+            'ticket_id': 1,
+            'progress_trigger': 'queue-name',
+        })
+        err = 'this is the worker error'
+        worker = _worker(None, None, RuntimeError, err)
+        worker._on_message(msg)
+        self.assertEqual(1, amqp_utils.progress_update.call_count)
+        msg.channel.basic_ack.assert_called_once_with('foo')
+        self.assertEqual(1, amqp_utils.progress_failed.call_count)
+        self.assertEqual(
+            'queue-name', amqp_utils.progress_failed.call_args[0][0])
+        self.assertIn(
+            err, amqp_utils.progress_failed.call_args[0][1]['traceback'])
+        self.assertEqual(1, worker.store.put_file.call_count)
+
+    @mock.patch('ci_utils.amqp_worker.amqp_utils')
+    def testOnMessageCalledProcessError(self, amqp_utils):
+        '''Test on message logic nicely describes failed subprocess calls'''
+        msg = mock.Mock()
+        msg.delivery_tag = 'foo'
+        msg.body = json.dumps({
+            'ticket_id': 1,
+            'progress_trigger': 'queue-name',
+        })
+        err = 'this is the worker error'
+
+        class _calledprocess_worker(_worker):
+            rc = 1234
+            cmd = ['/bin/foo', 'bar']
+            output = 'i can\'t believe its not butter'
+
+            def handle_request(self, params, log):
+                raise subprocess.CalledProcessError(
+                    self.rc, self.cmd, self.output)
+
+        worker = _calledprocess_worker(None, None, None, None)
+        worker._on_message(msg)
+
+        self.assertEqual(1, amqp_utils.progress_update.call_count)
+        msg.channel.basic_ack.assert_called_once_with('foo')
+        self.assertEqual(1, amqp_utils.progress_failed.call_count)
+        self.assertEqual(
+            'queue-name', amqp_utils.progress_failed.call_args[0][0])
+        self.assertIn('CalledProcessError',
+                      amqp_utils.progress_failed.call_args[0][1]['traceback'])
+        err = amqp_utils.progress_failed.call_args[0][1]['error_message']
+        self.assertIn(str(worker.cmd), err)
+        self.assertIn(str(worker.output), err)
+        self.assertEqual(1, worker.store.put_file.call_count)
+
+    @mock.patch('ci_utils.amqp_worker.amqp_utils')
+    def testOnMessageKilled(self, amqp_utils):
+        '''Test on message logic for handling ctrl-c (upstart stopping)'''
+        msg = mock.Mock()
+        msg.delivery_tag = 'foo'
+        msg.body = json.dumps({
+            'ticket_id': 1,
+            'progress_trigger': 'queue-name',
+        })
+        worker = _worker(None, None, KeyboardInterrupt)
+        with self.assertRaises(KeyboardInterrupt):
+            worker._on_message(msg)
+        self.assertEqual(1, amqp_utils.progress_update.call_count)
+        msg.channel.basic_ack.assert_called_once_with('foo')
+        self.assertEqual(1, amqp_utils.progress_failed.call_count)
+        self.assertEqual(
+            'queue-name', amqp_utils.progress_failed.call_args[0][0])
+        self.assertIn('KeyboardInterrupt',
+                      amqp_utils.progress_failed.call_args[0][1]['traceback'])
+        self.assertEqual(1, worker.store.put_file.call_count)
+
+    @mock.patch('ci_utils.amqp_worker.amqp_utils')
+    def testNoTicket(self, amqp_utils):
+        '''Ensure we can gracefully deal with a bad message in the queue'''
+        msg = mock.Mock()
+        msg.delivery_tag = 'foo'
+        msg.body = json.dumps({
+            'progress_trigger': 'queue-name',
+        })
+        worker = _worker(amqp_utils, 1)
+        worker._on_message(msg)
+        msg.channel.basic_ack.assert_called_once_with('foo')
+        self.assertEqual(1, amqp_utils.progress_failed.call_count)
+        self.assertEqual(
+            'queue-name', amqp_utils.progress_failed.call_args[0][0])
+
+    @mock.patch('ci_utils.amqp_worker.amqp_utils')
+    def testNoQueue(self, amqp_utils):
+        '''Ensure we can gracefully deal with a bad message in the queue'''
+        msg = mock.Mock()
+        msg.delivery_tag = 'foo'
+        msg.body = json.dumps({
+        })
+        worker = _worker(amqp_utils, 1)
+        worker._on_message(msg)
+        # all we can check is that the message got acked
+        msg.channel.basic_ack.assert_called_once_with('foo')
+
+    @mock.patch('ci_utils.amqp_worker.amqp_utils')
+    @mock.patch('ci_utils.amqp_worker._get')
+    def testCancel(self, get, amqp_utils):
+        '''Ensure workers get the "cancel" message'''
+        msg = mock.Mock()
+        msg.delivery_tag = 'foo'
+        msg.body = json.dumps({
+            'ticket_id': 1,
+            'progress_trigger': 'queue-name',
+            'param1': 'foo',
+            'param2': 42,
+            'cancel_url': 'foo'
+        })
+        get.return_value = {'building': False}
+        retval = {'msg': 'this is the worker return value'}
+        worker = _worker(amqp_utils, retval)
+        worker.cancel_interval = 0.01
+        worker._on_message(msg)
+        msg.channel.basic_ack.assert_called_once_with('foo')
+        amqp_utils.progress_completed.assert_called_once_with(
+            'queue-name', retval)
+        retval = amqp_utils.progress_completed.call_args[0][1]
+        self.assertTrue(retval['cancelled'])
+
+    def testSaveLastRun(self):
+        results = {'foo': 'bar'}
+        worker = _worker(None, None)
+        worker.save_last_check(results)
+        self.addCleanup(os.unlink, worker.last_check_file)
+        check = worker.get_last_check()
+        self.assertIn('last_run', check)
+        self.assertEqual(results['foo'], check['foo'])
+
+
+class TestTimer(unittest.TestCase):
+    def testCanCancel(self):
+        def cb():
+            raise RuntimeError('cancel failed')
+        t = amqp_worker._timer(1, cb)
+        t.start()
+        t.cancel()
+
+    def testCBRuns(self):
+        def cb():
+            self.run = True
+        self.run = False
+        t = amqp_worker._timer(0.01, cb)
+        t.start()
+        time.sleep(.02)
+        t.cancel()
+        self.assertTrue(self.run)

=== added file 'subunit-results/subunitresults/ci_utils/tests/test_data_store.py'
--- subunit-results/subunitresults/ci_utils/tests/test_data_store.py	1970-01-01 00:00:00 +0000
+++ subunit-results/subunitresults/ci_utils/tests/test_data_store.py	2014-08-21 14:14:59 +0000
@@ -0,0 +1,60 @@
+# Ubuntu CI Engine
+# Copyright 2014 Canonical Ltd.
+
+# This program is free software: you can redistribute it and/or modify it
+# under the terms of the GNU Affero General Public License version 3, as
+# published by the Free Software Foundation.
+
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import unittest
+
+
+from ci_utils import data_store
+
+
+filename = 'myproject_1.0-1_source.changes'
+
+
+class TestDataStoreConfig(unittest.TestCase):
+
+    def test_valid_auth_config(self):
+        data_store.DataStore.validate_auth_config({
+            'auth_url': 'http://example.com',
+            'auth_user': 'user',
+            'auth_password': 'pass',
+            'auth_tenant_name': 'tenant',
+            'auth_region': 'region',
+        })
+
+    def test_invalid_auth_config(self):
+        with self.assertRaises(data_store.DataStoreException):
+            data_store.DataStore.validate_auth_config({})
+
+
+class TestDataStoreFileName(unittest.TestCase):
+
+    def test_get_file_name(self):
+        path = '/home/ubuntu/myproject/{}'.format(filename)
+        resp = data_store._get_file_name(path)
+        self.assertEqual(resp, filename)
+
+    def test_get_relativepath_file_name(self):
+        path = '../../myproject/{}'.format(filename)
+        resp = data_store._get_file_name(path)
+        self.assertEqual(resp, filename)
+
+    def test_get_file_name_same_directory(self):
+        path = '{}'.format(filename)
+        resp = data_store._get_file_name(path)
+        self.assertEqual(resp, filename)
+
+
+if __name__ == "__main__":
+    unittest.main()

=== added file 'subunit-results/subunitresults/ci_utils/tests/test_fixtures.py'
--- subunit-results/subunitresults/ci_utils/tests/test_fixtures.py	1970-01-01 00:00:00 +0000
+++ subunit-results/subunitresults/ci_utils/tests/test_fixtures.py	2014-08-21 14:14:59 +0000
@@ -0,0 +1,87 @@
+# Ubuntu CI Engine
+# Copyright 2014 Canonical Ltd.
+
+# This program is free software: you can redistribute it and/or modify it
+# under the terms of the GNU Affero General Public License version 3, as
+# published by the Free Software Foundation.
+
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import errno
+import os
+import unittest
+
+from ucitests import fixtures as uci_fixtures
+
+from ci_utils.testing import fixtures
+
+
+class TestFakeDataStore(unittest.TestCase):
+    """Test the FakeDataStore implementation.
+
+    Since the target is tests themselves focusing on happy paths, no failures
+    are tested here.
+    """
+
+    def setUp(self):
+        super().setUp()
+        uci_fixtures.set_uniq_cwd(self)
+        self.ds_path = os.path.join(self.uniq_dir, 'data_store')
+        self.ds = fixtures.FakeDataStore(self.ds_path, None)
+
+    def test_put_get_file(self):
+        self.assertEqual([], self.ds.list_files())
+        url = self.ds.put_file('foo', b'foo content\n')
+        self.assertTrue(url.startswith('file:///'))
+        self.assertTrue(url.endswith(os.path.join(self.ds_path, 'foo')))
+        self.assertEqual(['foo'], self.ds.list_files())
+        self.assertEqual(b'foo content\n', self.ds.get_file('foo'))
+
+    def test_list_files_empty(self):
+        self.assertEqual([], self.ds.list_files())
+
+    def test_list_files_some_files(self):
+        self.ds.put_file('foo', b'foo content\n')
+        self.ds.put_file('bar', b'bar content\n')
+        self.assertEqual(sorted(['foo', 'bar']), self.ds.list_files())
+
+    def test_delete_file(self):
+        self.ds.put_file('foo', b'foo content\n')
+        self.assertEqual(['foo'], self.ds.list_files())
+        self.ds.delete_file('foo')
+        self.assertEqual([], self.ds.list_files())
+
+    def test_clear(self):
+        self.ds.put_file('foo', b'foo content\n')
+        self.ds.put_file('bar', b'bar content\n')
+        self.assertEqual(sorted(['foo', 'bar']), self.ds.list_files())
+        self.ds.clear()
+        self.assertEqual([], self.ds.list_files())
+
+    def test_delete(self):
+        self.assertTrue(os.path.exists(self.ds_path))
+        self.ds.put_file('foo', b'foo content\n')
+        self.ds.delete()
+        self.assertFalse(os.path.exists(self.ds_path))
+
+
+class TestUsingFakeDataStore(unittest.TestCase):
+
+    def setUp(self):
+        super().setUp()
+        uci_fixtures.set_uniq_cwd(self)
+        self.ds_path = os.path.join(self.uniq_dir, 'data_store')
+        self.ds = fixtures.FakeDataStore(self.ds_path, None)
+
+    def test_create_same_data_store(self):
+        # Trying to create the same data store twice fails
+        with self.assertRaises(OSError) as cm:
+            fixtures.FakeDataStore(self.ds_path, None)
+        self.assertEqual(errno.EEXIST, cm.exception.errno)
+        self.assertEqual(self.ds_path, cm.exception.filename)

=== added file 'subunit-results/subunitresults/ci_utils/tests/test_json_status.py'
--- subunit-results/subunitresults/ci_utils/tests/test_json_status.py	1970-01-01 00:00:00 +0000
+++ subunit-results/subunitresults/ci_utils/tests/test_json_status.py	2014-08-21 14:14:59 +0000
@@ -0,0 +1,144 @@
+# Ubuntu CI Engine
+# Copyright 2014 Canonical Ltd.
+
+# This program is free software: you can redistribute it and/or modify it
+# under the terms of the GNU Affero General Public License version 3, as
+# published by the Free Software Foundation.
+
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import json
+import unittest
+from unittest import mock
+import urllib.error
+import urllib.parse
+import urllib.request
+
+
+from ci_utils.json_status import JSONStatus, run_server
+
+
+class TestRabbitStatus(unittest.TestCase):
+    @mock.patch('ci_utils.amqp_utils.get_config')
+    def testNoConfig(self, get_config):
+        get_config.return_value = None
+        status = JSONStatus()
+        status.add_rabbit_worker_health('foo')
+        self.assertEqual([], status.results)
+
+    @mock.patch('ci_utils.amqp_utils.get_config')
+    @mock.patch('urllib.request.build_opener')
+    def testCantReach(self, opener, get_config):
+        opener.side_effect = RuntimeError('foo')
+        status = JSONStatus()
+        status.add_rabbit_worker_health('foo')
+        expected = {
+            'status': 'fail',
+            'value': 'unable to check',
+            'label': 'workers-online'
+        }
+        self.assertEqual([expected], status.results)
+
+    @mock.patch('ci_utils.amqp_utils.get_config')
+    @mock.patch('urllib.request.build_opener')
+    def _testRabbitHealth(self, queue, data, opener, get_config):
+        fdfake = mock.Mock()
+        fdfake.read.return_value = json.dumps(data)
+        con = mock.Mock()
+        con.open.return_value = fdfake
+        opener.return_value = con
+        status = JSONStatus()
+        status.add_rabbit_worker_health(queue)
+        return status.results
+
+    def testWorkers(self):
+        data = [
+            {'name': 'bla', 'consumers': 2},
+            {'name': 'foo', 'consumers': 1},
+            {'name': 'bar', 'consumers': 1},
+        ]
+        expected = {
+            'status': 'okay',
+            'value': 1,
+            'label': 'workers-online'
+        }
+        results = self._testRabbitHealth('foo', data)
+        self.assertEqual([expected], results)
+
+    def testNoWorkers(self):
+        data = [
+            {'name': 'bla', 'consumers': 2},
+            {'name': 'foo', 'consumers': 0},
+            {'name': 'bar', 'consumers': 1},
+        ]
+        expected = {
+            'status': 'fail',
+            'value': 0,
+            'label': 'workers-online'
+        }
+        results = self._testRabbitHealth('foo', data)
+        self.assertEqual([expected], results)
+
+    def testNoQueue(self):
+        data = [
+            {'name': 'bla', 'consumers': 2},
+            {'name': 'bar', 'consumers': 1},
+        ]
+        expected = {
+            'status': 'fail',
+            'value': 'no queue defined',
+            'label': 'workers-online'
+        }
+        results = self._testRabbitHealth('foo', data)
+        self.assertEqual([expected], results)
+
+
+class TestHealthServer(unittest.TestCase):
+    def get_resp(self, path='/'):
+        server = run_server(self.cb, 0)
+        try:
+            url = 'http://localhost:%d%s' % (server.server_port, path)
+            return urllib.request.urlopen(url)
+        finally:
+            server.shutdown()
+
+    def cb(self):
+        self.cb_called = True
+        status = JSONStatus()
+        status.add_okay('name', 'value')
+        return status
+
+    def test_plain(self):
+        resp = self.get_resp()
+        self.assertEqual(200, resp.code)
+        self.assertEqual(
+            self.cb().results,
+            json.loads(resp.read().decode('utf-8'))
+        )
+
+    def test_jsonp(self):
+        resp = self.get_resp('/?format=jsonp&callback=foo')
+        self.assertEqual(200, resp.code)
+        expected = 'foo(' + json.dumps(self.cb().results) + ')'
+        self.assertEqual(expected, resp.read().decode('utf-8'))
+
+    def test_cache(self):
+        self.cb_called = False
+        resp = self.get_resp()
+        self.assertEqual(200, resp.code)
+        self.assertTrue(self.cb_called)
+
+        self.cb_called = False
+        resp = self.get_resp()
+        self.assertEqual(200, resp.code)
+        self.assertFalse(self.cb_called)
+
+        resp = self.get_resp('/?refresh=1')
+        self.assertEqual(200, resp.code)
+        self.assertTrue(self.cb_called)

=== added file 'subunit-results/subunitresults/ci_utils/tests/test_tmpdir.py'
--- subunit-results/subunitresults/ci_utils/tests/test_tmpdir.py	1970-01-01 00:00:00 +0000
+++ subunit-results/subunitresults/ci_utils/tests/test_tmpdir.py	2014-08-21 14:14:59 +0000
@@ -0,0 +1,27 @@
+# Ubuntu CI Engine
+# Copyright 2014 Canonical Ltd.
+
+# This program is free software: you can redistribute it and/or modify it
+# under the terms of the GNU Affero General Public License version 3, as
+# published by the Free Software Foundation.
+
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import os
+import unittest
+
+from ci_utils import TmpDir
+
+
+class TestTmpDir(unittest.TestCase):
+    def test_tmpdir(self):
+        with TmpDir() as tmpdir:
+            dir_created = tmpdir
+            self.assertTrue(os.path.exists(dir_created))
+        self.assertFalse(os.path.exists(dir_created))

=== added file 'subunit-results/subunitresults/ci_utils/unit_config.py'
--- subunit-results/subunitresults/ci_utils/unit_config.py	1970-01-01 00:00:00 +0000
+++ subunit-results/subunitresults/ci_utils/unit_config.py	2014-08-21 14:14:59 +0000
@@ -0,0 +1,60 @@
+# Ubuntu CI Engine
+# Copyright 2014 Canonical Ltd.
+
+# This program is free software: you can redistribute it and/or modify it
+# under the terms of the GNU Affero General Public License version 3, as
+# published by the Free Software Foundation.
+
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import os
+
+import yaml
+
+
+HERE = os.path.abspath(os.path.dirname(__file__))
+
+
+_unit_config = None
+
+
+def _load():
+    path = os.path.join(HERE, '../../../../unit_config')
+    # TODO remove once all charms have converted to new code layout
+    # Fallback to old location configs (inside code tree) for local branches
+    # testing and charms transition.
+    if not os.path.exists(path):
+        path = os.path.abspath(os.path.join(HERE, '../../unit_config'))
+    with open(path) as f:
+        global _unit_config
+        _unit_config = yaml.safe_load(f)
+
+
+def get(key):
+    if not _unit_config:
+        _load()
+    return _unit_config[key]
+
+
+def get_auth_config():
+    if not _unit_config:
+        _load()
+    return _unit_config
+
+
+def is_hpcloud(string=None):
+    """Are we running against/on HP cloud.
+
+    :param string: The authentication URL to check (defaults to OS_AUTH_URL).
+    """
+    if string is None:
+        string = os.environ.get('OS_AUTH_URL', None)
+    if string and 'hpcloud' in string:
+        return True
+    return False

=== added file 'subunit-results/subunitresults/queue_worker.py'
--- subunit-results/subunitresults/queue_worker.py	1970-01-01 00:00:00 +0000
+++ subunit-results/subunitresults/queue_worker.py	2014-08-21 14:14:59 +0000
@@ -0,0 +1,91 @@
+#!/usr/bin/env python3
+# Ubuntu CI Engine
+# Copyright 2014 Canonical Ltd.
+#
+# This program is free software: you can redistribute it and/or modify it
+# under the terms of the GNU Affero General Public License version 3, as
+# published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+
+import logging
+import json
+from queue import Queue
+
+# TODO:
+# use real data store and remove this import
+from subunitresults.ci_utils.testing import fixtures
+from subunitresults.ci_utils import amqp_utils, amqp_worker
+
+from subunitresults.utils import retry
+from subunitresults.subunit_parser import run_result
+
+
+logging.basicConfig(level='INFO')
+logger = logging.getLogger(__name__)
+
+
+def get_item(queue):
+    return queue.get(block=True, timeout=None)
+
+
+def get_subunit_path(json_blob):
+    d = json.loads(json_blob)
+    try:
+        subunit_path = d['subunit_path']
+    except KeyError:
+        raise Exception("subunit_path not found in json message")
+    return subunit_path
+
+
+@retry(Exception, tries=5, delay=1, backoff=2, logger=None)
+def fetch_subunit(data_store, url):
+    contents = data_store.get_file(url)
+    return contents
+
+
+class Worker:
+    def __init__(self, data_store, queue):
+        self.queue = queue
+        self.data_store = data_store
+
+    def do_one_job(self):
+        json_blob = get_item(self.queue)
+        path = get_subunit_path(json_blob)
+        subunit_file = fetch_subunit(self.data_store, path)
+        run_result(self.data_store, subunit_file)
+
+
+class RabbitWorker(amqp_worker.AMQPWorker):
+    def __init__(self, logger_name, data_store, queue):
+        super().__init__(logger_name)
+        self.logger_name = logger_name
+        self.data_store = data_store
+        self.queue = queue
+        self.worker = Worker(self.data_store, self.queue)
+
+    def handle_request(self, params, logger):
+        try:
+            # TODO:
+            # need to get message from rabbit queue, put on python queue
+            self.worker.do_one_job()
+            return (amqp_utils.progress_completed, {})
+        except Exception:
+            return (amqp_utils.progress_failed, {})
+
+
+def launch_amqp_listener():
+    # TODO:
+    # need to create RabbitWorker instance with
+    # real data store
+    logger_name = 'subunit-results'
+    data_store = fixtures.FakeDataStore('./data_store', None)
+    queue = Queue()
+    RabbitWorker(logger_name, data_store, queue)

=== added file 'subunit-results/subunitresults/subunit_parser.py'
--- subunit-results/subunitresults/subunit_parser.py	1970-01-01 00:00:00 +0000
+++ subunit-results/subunitresults/subunit_parser.py	2014-08-21 14:14:59 +0000
@@ -0,0 +1,151 @@
+#!/usr/bin/env python3
+# Ubuntu CI Engine
+# Copyright 2014 Canonical Ltd.
+#
+# This program is free software: you can redistribute it and/or modify it
+# under the terms of the GNU Affero General Public License version 3, as
+# published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+from datetime import datetime
+import logging
+import re
+
+from subunit import ByteStreamToStreamResult
+from testtools import StreamToExtendedDecorator, TestResult
+
+
+logging.basicConfig(level='INFO')
+logger = logging.getLogger(__name__)
+
+
+class FileSaveResult(TestResult):
+
+    """A TestResult that saves a test summary, details, and attachments."""
+
+    def __init__(self, data_store):
+        super().__init__()
+        self.data_store = data_store
+
+        self.testcase_summaries = {}
+        self.summary = dict(
+            tests_run=0,
+            errors=0,
+            expected_failures=0,
+            failures=0,
+            num_attachments=0,
+            run_time=0,
+            skipped=0,
+            unexpected_successes=0,
+            tests=self.testcase_summaries
+        )
+
+        self.content_attachments = []
+        self.testcase_details = dict(
+            run_time=2.0,
+            content_attachments=self.content_attachments
+        )
+
+    def addError(self, test, err=None, details=None):
+        super().addError(test, err=err, details=details)
+        self.summary['num_attachments'] += len(details)
+        test.status = 'error'
+        self.save_file_attachments(test, details)
+
+    def addExpectedFailure(self, test, err=None, details=None):
+        super().addExpectedFailure(test, err=err, details=details)
+        self.summary['num_attachments'] += len(details)
+        test.status = 'xfail'
+        self.save_file_attachments(test, details)
+
+    def addFailure(self, test, err=None, details=None):
+        super().addFailure(test, err=err, details=details)
+        self.summary['num_attachments'] += len(details)
+        test.status = 'fail'
+        self.save_file_attachments(test, details)
+
+    def addSkip(self, test, reason=None, details=None):
+        super().addSkip(test, reason=reason, details=details)
+        self.summary['num_attachments'] += len(details)
+        test.status = 'skip'
+        self.save_file_attachments(test, details)
+
+    def addSuccess(self, test, details=None):
+        super().addSuccess(test, details=details)
+        self.summary['num_attachments'] += len(details)
+        test.status = 'pass'
+        self.save_file_attachments(test, details)
+
+    def addUnexpectedSuccess(self, test, details=None):
+        super().addUnexpectedSuccess(test, details=details)
+        self.summary['num_attachments'] += len(details)
+        test.status = 'uxsuccess'
+        self.save_file_attachments(test, details)
+
+    def stopTest(self, test):
+        super().stopTest(test)
+        elapsed = test._timestamps[1] - test._timestamps[0]
+        test_time = elapsed.total_seconds()
+        self.summary['run_time'] += test_time
+        self.testcase_summaries[test.id()] = [test.status, test_time]
+        self.testcase_details['run_time'] = test_time
+        self.testcase_details['status'] = test.status
+
+    def stopTestRun(self):
+        super().stopTestRun()
+        self.summary['errors'] = len(self.errors)
+        self.summary['expected_failures'] = len(self.expectedFailures)
+        self.summary['failures'] = len(self.failures)
+        self.summary['skipped'] = len(self.skip_reasons)
+        self.summary['tests_run'] = self.testsRun
+        self.summary['unexpected_successes'] = len(self.unexpectedSuccesses)
+
+    def save_file_attachments(self, test, details):
+        testcase_details = dict(
+            content_attachments=[]
+        )
+
+        if details is not None:
+            for name, detail in details.items():
+                if detail.iter_bytes() != [b'']:
+                    saved_name = create_unique_name(test.id(), name)
+                    content = b''.join(chunk for chunk in detail.iter_bytes())
+                    mime_type = str(detail.content_type)
+                    self.data_store.put_file(saved_name, content, mime_type)
+                    content_attachment = dict(
+                        name=saved_name,
+                        mime_type=mime_type,
+                        size=len(content),
+                        location=self.data_store.file_path(saved_name)
+                    )
+                    testcase_details['content_attachments'].append(
+                        content_attachment
+                    )
+        self.testcase_details = testcase_details
+        return testcase_details
+
+
+def create_unique_name(test_id, file_name, timestamp=None):
+    if timestamp is None:
+        timestamp = datetime.timestamp(datetime.utcnow())
+    slug = re.sub(r'(?u)[^-\w.]', '', file_name)
+    return '%s+%s+%s' % (timestamp, test_id, slug)
+
+
+def run_result(data_store, subunit_stream):
+    suite = ByteStreamToStreamResult(subunit_stream)
+    fs_result = FileSaveResult(data_store)
+    result = StreamToExtendedDecorator(fs_result)
+    result.startTestRun()
+    try:
+        suite.run(result)
+    finally:
+        result.stopTestRun()
+    return True

=== added directory 'subunit-results/subunitresults/tests'
=== added file 'subunit-results/subunitresults/tests/__init__.py'
--- subunit-results/subunitresults/tests/__init__.py	1970-01-01 00:00:00 +0000
+++ subunit-results/subunitresults/tests/__init__.py	2014-08-21 14:14:59 +0000
@@ -0,0 +1,14 @@
+# Ubuntu CI Engine
+# Copyright 2014 Canonical Ltd.
+#
+# This program is free software: you can redistribute it and/or modify it
+# under the terms of the GNU Affero General Public License version 3, as
+# published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.

=== added file 'subunit-results/subunitresults/tests/results_for_tests.subunit'
Binary files subunit-results/subunitresults/tests/results_for_tests.subunit	1970-01-01 00:00:00 +0000 and subunit-results/subunitresults/tests/results_for_tests.subunit	2014-08-21 14:14:59 +0000 differ
=== added file 'subunit-results/subunitresults/tests/test_queue_worker.py'
--- subunit-results/subunitresults/tests/test_queue_worker.py	1970-01-01 00:00:00 +0000
+++ subunit-results/subunitresults/tests/test_queue_worker.py	2014-08-21 14:14:59 +0000
@@ -0,0 +1,62 @@
+#!/usr/bin/env python3
+# Ubuntu CI Engine
+# Copyright 2014 Canonical Ltd.
+#
+# This program is free software: you can redistribute it and/or modify it
+# under the terms of the GNU Affero General Public License version 3, as
+# published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import json
+import os
+from queue import Queue
+import unittest
+
+from testtools.content_type import ContentType
+from ucitests import fixtures as uci_fixtures
+
+from ci_utils.testing import fixtures
+from subunitresults import queue_worker
+
+
+class TestQueueWorker(unittest.TestCase):
+
+    def test_get_item(self):
+        test_msg = 'hello'
+        q = Queue()
+        q.put(test_msg)
+        msg = queue_worker.get_item(q)
+        self.assertEqual(msg, test_msg)
+
+    def test_get_subunit_path(self):
+        test_path = 'http://foo/12345/results.sub'
+        json_blob = json.dumps({
+            'subunit_path': '%s' % test_path,
+            'ticket_id': 12345
+        })
+        path = queue_worker.get_subunit_path(json_blob)
+        self.assertEqual(path, test_path)
+
+    def test_fetch_subunit(self):
+        uci_fixtures.set_uniq_cwd(self)
+        ds_path = os.path.join(self.uniq_dir, 'data_store')
+        ds = fixtures.FakeDataStore(ds_path, None)
+        self.addCleanup(ds.delete, recursive=True)
+        filename = 'test_file'
+        test_path = 'http://foo/12345/results.sub'
+        json_blob = json.dumps({
+            'subunit_path': '%s' % test_path,
+            'ticket_id': 12345
+        }).encode('utf-8')
+        content_type = ContentType('application', 'json')
+        ds.put_file(filename, json_blob, content_type)
+        self.addCleanup(ds.delete_file, filename)
+        fetched = queue_worker.fetch_subunit(ds, filename)
+        self.assertEqual(json_blob, fetched)

=== added file 'subunit-results/subunitresults/tests/test_subunit_parser.py'
--- subunit-results/subunitresults/tests/test_subunit_parser.py	1970-01-01 00:00:00 +0000
+++ subunit-results/subunitresults/tests/test_subunit_parser.py	2014-08-21 14:14:59 +0000
@@ -0,0 +1,161 @@
+#!/usr/bin/env python3
+# Ubuntu CI Engine
+# Copyright 2014 Canonical Ltd.
+#
+# This program is free software: you can redistribute it and/or modify it
+# under the terms of the GNU Affero General Public License version 3, as
+# published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+
+from datetime import datetime
+import unittest
+import os
+
+from testtools import TestCase, TestResult
+from testtools.content import Content, ContentType
+from testtools.content_type import UTF8_TEXT
+from ucitests import fixtures as uci_fixtures
+
+from subunitresults import subunit_parser
+from subunitresults.ci_utils.testing import fixtures
+
+
+def get_case(attachment_name, data):
+    # Define the class in a funCtion so test loading won't try to load it as a
+    # regular test class.
+    class TestyTest(TestCase):
+        def test_add_utf8_detail(self):
+            self._timestamps = (
+                datetime(2007, 12, 6, 15, 29, 43, 79060),
+                datetime(2007, 12, 6, 15, 29, 43, 79060)
+            )
+            self.addDetail(
+                attachment_name,
+                Content(UTF8_TEXT, lambda: [data])
+            )
+    return TestyTest('test_add_utf8_detail')
+
+
+class TestSubUnit(unittest.TestCase):
+
+    def setUp(self):
+        uci_fixtures.set_uniq_cwd(self)
+        ds_path = os.path.join(self.uniq_dir, 'data_store')
+        self.ds = fixtures.FakeDataStore(ds_path, None)
+        self.addCleanup(self.ds.delete, recursive=True)
+
+    def run_test_with_utf8_attachment(self, attachment_name, data):
+        test = get_case(attachment_name, data)
+        fs_result = subunit_parser.FileSaveResult(self.ds)
+        test.run(fs_result)
+        return fs_result
+
+    def test_file_save_result(self):
+        fs_result = subunit_parser.FileSaveResult(self.ds)
+        self.assertIsInstance(fs_result, TestResult)
+
+    def test_default_summary(self):
+        fs_result = subunit_parser.FileSaveResult(self.ds)
+        default_summary = dict(
+            tests_run=0,
+            errors=0,
+            expected_failures=0,
+            failures=0,
+            num_attachments=0,
+            run_time=0,
+            skipped=0,
+            unexpected_successes=0,
+            tests=dict()
+        )
+        self.assertDictEqual(fs_result.summary, default_summary)
+
+    def test_testsuite_summary(self):
+        attachment_name = 'foo'
+        attachment_data = b'bar'
+        fs_result = self.run_test_with_utf8_attachment(
+            attachment_name,
+            attachment_data
+        )
+        expected_summary = {
+            'failures': 0,
+            'skipped': 0,
+            'num_attachments': 1,
+            'run_time': 0.0,
+            'tests_run': 0,
+            'errors': 0,
+            'unexpected_successes': 0,
+            'expected_failures': 0,
+            'tests': {
+                'tests.test_subunit_parser.TestyTest.test_add_utf8_detail':
+                    ['pass', 0.0]
+            }
+        }
+        self.assertDictEqual(fs_result.summary, expected_summary)
+
+    def test_testcase_details(self):
+        attachment_name = 'foo'
+        attachment_data = b'bar'
+        fs_result = self.run_test_with_utf8_attachment(
+            attachment_name,
+            attachment_data
+        )
+        self.assertEqual(fs_result.testcase_details['status'], 'pass')
+        self.assertEqual(fs_result.testcase_details['run_time'], 0.0)
+        attachment = fs_result.testcase_details['content_attachments'][0]
+        self.assertEqual(attachment['size'], 3)
+
+    def test_save_attachment(self):
+        attachment_name = 'foo'
+        attachment_data = b'bar'
+        fs_result = self.run_test_with_utf8_attachment(
+            attachment_name,
+            attachment_data
+        )
+        attachments = fs_result.testcase_details['content_attachments']
+        saved_name = attachments[0]['name']
+        self.addCleanup(self.ds.delete_file, saved_name)
+        self.assertEqual([saved_name], self.ds.list_files())
+        self.assertEqual(attachment_data, self.ds.get_file(saved_name))
+
+    def test_put_attachment(self):
+        # uci_fixtures.set_uniq_cwd(self)
+        # ds_path = os.path.join(self.uniq_dir, 'data_store')
+        # ds = fixtures.FakeDataStore(ds_path, None)
+        # self.addCleanup(ds.delete, recursive=True)
+        filename = 'test_file'
+        content = b'{"subunit_path": "http://example/foo"}'
+        mime_type = ContentType('application', 'json')
+        # queue_worker.put_attachment(self.ds, filename, contents, content_type)
+        self.ds.put_file(filename, content, mime_type)
+        self.addCleanup(self.ds.delete_file, filename)
+        self.assertEqual([filename], self.ds.list_files())
+
+    def test_run_result(self):
+        test_dir = os.path.dirname(os.path.realpath(__file__))
+        subunit_file = os.path.join(test_dir, 'results_for_tests.subunit')
+        with open(subunit_file, 'rb') as f:
+            success = subunit_parser.run_result(self.ds, f)
+        self.assertTrue(success)
+
+
+class TestSubUnitName(unittest.TestCase):
+
+    def test_create_unique_name(self):
+        id = 'a.b.c'
+        name = 'name'
+        fake_epoch = '1407454141.539095'
+        uniq_name = subunit_parser.create_unique_name(
+            id,
+            name,
+            timestamp=fake_epoch
+        )
+        expected_name = '%s+%s+%s' % (fake_epoch, id, name)
+        self.assertEqual(uniq_name, expected_name)

=== added file 'subunit-results/subunitresults/tests/test_utils.py'
--- subunit-results/subunitresults/tests/test_utils.py	1970-01-01 00:00:00 +0000
+++ subunit-results/subunitresults/tests/test_utils.py	2014-08-21 14:14:59 +0000
@@ -0,0 +1,121 @@
+#!/usr/bin/python3
+# Ubuntu CI Engine
+# Copyright 2014 Canonical Ltd.
+
+# This program is free software: you can redistribute it and/or modify it
+# under the terms of the GNU Affero General Public License version 3, as
+# published by the Free Software Foundation.
+
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+
+import logging
+import unittest
+
+from subunitresults.utils import retry
+
+
+class RetryableError(Exception):
+    pass
+
+
+class AnotherRetryableError(Exception):
+    pass
+
+
+class UnexpectedError(Exception):
+    pass
+
+
+class RetryTestCase(unittest.TestCase):
+
+    def test_no_retry_required(self):
+        self.counter = 0
+
+        @retry(RetryableError, tries=4, delay=0.1)
+        def succeeds():
+            self.counter += 1
+            return 'success'
+
+        r = succeeds()
+
+        self.assertEqual(r, 'success')
+        self.assertEqual(self.counter, 1)
+
+    def test_retries_once(self):
+        self.counter = 0
+
+        @retry(RetryableError, tries=4, delay=0.1)
+        def fails_once():
+            self.counter += 1
+            if self.counter < 2:
+                raise RetryableError('failed')
+            else:
+                return 'success'
+
+        r = fails_once()
+        self.assertEqual(r, 'success')
+        self.assertEqual(self.counter, 2)
+
+    def test_limit_is_reached(self):
+        self.counter = 0
+
+        @retry(RetryableError, tries=4, delay=0.1)
+        def always_fails():
+            self.counter += 1
+            raise RetryableError('failed')
+
+        with self.assertRaises(RetryableError):
+            always_fails()
+        self.assertEqual(self.counter, 4)
+
+    def test_multiple_exception_types(self):
+        self.counter = 0
+
+        @retry((RetryableError, AnotherRetryableError), tries=4, delay=0.1)
+        def raise_multiple_exceptions():
+            self.counter += 1
+            if self.counter == 1:
+                raise RetryableError('a retryable error')
+            elif self.counter == 2:
+                raise AnotherRetryableError('another retryable error')
+            else:
+                return 'success'
+
+        r = raise_multiple_exceptions()
+        self.assertEqual(r, 'success')
+        self.assertEqual(self.counter, 3)
+
+    def test_unexpected_exception_does_not_retry(self):
+
+        @retry(RetryableError, tries=4, delay=0.1)
+        def raise_unexpected_error():
+            raise UnexpectedError('unexpected error')
+
+        with self.assertRaises(UnexpectedError):
+            raise_unexpected_error()
+
+    def test_using_a_logger(self):
+        self.counter = 0
+
+        sh = logging.StreamHandler()
+        logger = logging.getLogger(__name__)
+        logger.addHandler(sh)
+
+        @retry(RetryableError, tries=4, delay=0.1, logger=logger)
+        def fails_once():
+            self.counter += 1
+            if self.counter < 2:
+                raise RetryableError('failed')
+            else:
+                return 'success'
+
+
+if __name__ == '__main__':
+    unittest.main()

=== added file 'subunit-results/subunitresults/utils.py'
--- subunit-results/subunitresults/utils.py	1970-01-01 00:00:00 +0000
+++ subunit-results/subunitresults/utils.py	2014-08-21 14:14:59 +0000
@@ -0,0 +1,59 @@
+# Ubuntu CI Engine
+# Copyright 2014 Canonical Ltd.
+
+# This program is free software: you can redistribute it and/or modify it
+# under the terms of the GNU Affero General Public License version 3, as
+# published by the Free Software Foundation.
+
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+
+import time
+from functools import wraps
+
+
+def retry(ExceptionToCheck, tries=4, delay=3, backoff=2, logger=None):
+    """Retry calling the decorated function using an exponential backoff.
+
+    http://www.saltycrane.com/blog/2009/11/trying-out-retry-decorator-python/
+    original from: http://wiki.python.org/moin/PythonDecoratorLibrary#Retry
+
+    :param ExceptionToCheck: the exception to check. may be a tuple of
+        exceptions to check
+    :type ExceptionToCheck: Exception or tuple
+    :param tries: number of times to try (not retry) before giving up
+    :type tries: int
+    :param delay: initial delay between retries in seconds
+    :type delay: int
+    :param backoff: backoff multiplier e.g. value of 2 will double the delay
+        each retry
+    :type backoff: int
+    :param logger: logger to use.
+    :type logger: logging.Logger instance
+    """
+    def deco_retry(f):
+
+        @wraps(f)
+        def f_retry(*args, **kwargs):
+            mtries, mdelay = tries, delay
+            while mtries > 1:
+                try:
+                    return f(*args, **kwargs)
+                except ExceptionToCheck as e:
+                    msg = "%s, Retrying in %d seconds..." % (str(e), mdelay)
+                    if logger:
+                        logger.warning(msg)
+                    time.sleep(mdelay)
+                    mtries -= 1
+                    mdelay *= backoff
+            return f(*args, **kwargs)
+
+        return f_retry  # true decorator
+
+    return deco_retry

=== added file 'subunit-results/tox.ini'
--- subunit-results/tox.ini	1970-01-01 00:00:00 +0000
+++ subunit-results/tox.ini	2014-08-21 14:14:59 +0000
@@ -0,0 +1,21 @@
+[tox]
+envlist = py34, flake8
+
+[testenv]
+deps =
+    ucitests
+    coverage
+commands =
+    {envpython} setup.py develop
+    {envpython} -m unittest discover -s subunitresults -v
+    # coverage erase
+    # coverage run --omit=.tox/* -m unittest discover -s subunitresults
+    # coverage html
+    # coverage report -m
+    # coverage html
+
+[testenv:flake8]
+deps =
+    flake8
+commands =
+    flake8 subunitresults