Merge lp:~vila/uci-engine/1302471-fake-rabbit into lp:uci-engine

[Vincent Ladeuil]

Fake rabbit queue and tests against a real rabbit server.

This MP is a first (rough) shot at starting tests for a real rabbit
server and providing a fake queue for test writers.

I've added a *few* tests against a deployed (real) rabbit server with
required explicit setup for credentials so we can run the tests from a
dev host (instead of from a deployed instance with the rabbitmq-worker
charm providing the credentials automagically).

For the isolated tests, I've added a first, narrowed, implementation
of rabbit queue without a real server. For the API there, I aligned
with the terms used in rabbit documentation. A test (as a producer)
can publish messages in a queue and (as a consumer) subscribe to a
queue providing a callback used for each message.

There is no error checking, no nothing, only the happy path: push a
list of messages, subscribe and execute the callback for each message
(no error, no nothing), look at the state of the queue. More data can
be provided by the callback.

[PS Jenkins bot (ps-jenkins)]

FAILED: Continuous integration, rev:452
http://s-jenkins.ubuntu-ci:8080/job/uci-engine-ci/623/
Executed test runs:

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/uci-engine-ci/623/rebuild

review: Needs Fixing (continuous-integration)

[Andy Doan (doanac)]

You have a merge conflict in this, and it also seems like you may be combining 3 different MPs, so I'm wondering if this was intentional? I think there are 3-4 change-sets:

1) re-ordering of parameters in functions
I'd separate this so this diff will be small and easy to ack.

2) FakeTaskQueue
I think this plus the change to the test-runner

3) tests/test_rabbit.py

4) amqp_utils split up of send into send/declare_queue

I might be reading too fast, but I think all 4 of these changes are independent of one another (with the expception or #4).

review: Needs Information

[Vincent Ladeuil (vila)]

> You have a merge conflict in this,

Yeah, that's weird, I resolved it locally but haven't pushed the result, sorry about that.

> and it also seems like you may be combining
> 3 different MPs, so I'm wondering if this was intentional?

Well, yes, all the changes are related to each other so splitting them means the reviewer has to jump around MPs to understand *why* they are done.

> I think there are
> 3-4 change-sets:
>
> 1) re-ordering of parameters in functions
> I'd separate this so this diff will be small and easy to ack.

I can do that but if the related MPs are landed at different times the resulting history will lose the reasoning leading to that change.

>
> 2) FakeTaskQueue
> I think this plus the change to the test-runner
>
> 3) tests/test_rabbit.py
>
> 4) amqp_utils split up of send into send/declare_queue
>
> I might be reading too fast, but I think all 4 of these changes are
> independent of one another (with the expception or #4).

[lp:~vila/uci-engine/1302471-fake-rabbit]

453. By Vincent Ladeuil on 2014-05-13

Merge trunk fixing conflicts

454. By Vincent Ladeuil on 2014-05-13

Fix imports.

[PS Jenkins bot (ps-jenkins)]

PASSED: Continuous integration, rev:454
http://s-jenkins.ubuntu-ci:8080/job/uci-engine-ci/627/
Executed test runs:

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/uci-engine-ci/627/rebuild

review: Approve (continuous-integration)

[Andy Doan (doanac)]

looks good. at some point we probably should figure out what of:

 http://bazaar.launchpad.net/~canonical-ci-engineering/uci-engine/trunk/view/head:/ci-utils/ci_utils/tests/test_amqp_worker.py

could fit into this new FakeTaskQueue Fixture so that we can really do a thorough testing of each of our runworker handlers.

[Andy Doan (doanac)]

review: Approve

[Vincent Ladeuil (vila)]

> looks good. at some point we probably should figure out what of:
>
> http://bazaar.launchpad.net/~canonical-ci-engineering/uci-
> engine/trunk/view/head:/ci-utils/ci_utils/tests/test_amqp_worker.py
>
> could fit into this new FakeTaskQueue Fixture so that we can really do a
> thorough testing of each of our runworker handlers.

You got that perfectly right ;)

But I'm focused on isolated tests for the test runner right now and didn't attempt to do more than strictly necessary.

[Andy Doan (doanac)]

Just realized we did this all wrong. See in-line comment.

[Andy Doan (doanac)]

I missed something else really important in this. I *thought* I added an inline comment last night, but I noticed this morning its not there. I'll just re-explain here:

369 + def test_handle_request(self):
370 + worker = run_worker.TestRunnerWorker()
371 + # FIXME: We need a way to get an image that is available to the nova
372 + # setup we use. -- vila 20140512
373 + params = dict(ticket_id='t1', progress_trigger='progress',
374 + image_id=('ubuntu-released/ubuntu-precise-12.04'
375 + '-amd64-server-20140428-disk1.img'),
376 + package_list=['libpng'])
377 + self.queue.publish(params)
378 + self.queue.subscribe(worker.handle_request, self.logger)

If you look at AMQPWorker[1], you'll see that it isolates classes that extend it from dealing with rabbit already. The real method that handles amqp is _on_message. This pops the message off the queue, does some basic sanity checks and then calls _handle_request with the message converted into a dictionary.

So this test method really doesn't need to try and fake rabbit. You should just call _handle_request directoy with the "params" value you set up on line 373.

The real place to use the FakeTaskQueue would be in test_amqp_worker.py[2]. As you'll see there, I do quite a bit of testing against what rabbit will send the "_on_message" method already. However, I'm sure we could find some ways to incorporate that logic into the FakeTaskQueue more cleanly and add some new tests.

Sorry for missing this after its already been merged. It hit me while walking the dog.

1: http://bazaar.launchpad.net/~canonical-ci-engineering/uci-engine/trunk/view/head:/ci-utils/ci_utils/amqp_worker.py

2: http://bazaar.launchpad.net/~canonical-ci-engineering/uci-engine/trunk/view/head:/ci-utils/ci_utils/tests/test_amqp_worker.py

review: Needs Fixing

=== modified file 'branch-source-builder/bsbuilder/run_worker.py'
--- branch-source-builder/bsbuilder/run_worker.py	2014-04-28 14:45:39 +0000
+++ branch-source-builder/bsbuilder/run_worker.py	2014-05-14 13:40:29 +0000
@@ -72,7 +72,7 @@
     def __init__(self):
         super(BSBuilderWorker, self).__init__('bsbuilder')
 
-    def handle_request(self, log, params):
+    def handle_request(self, params, log):
         subtickets = params['subtickets']
         series = params['series']
         ppa = params['ppa']

=== modified file 'ci-utils/ci_utils/amqp_utils.py'
--- ci-utils/ci_utils/amqp_utils.py	2014-04-28 14:45:39 +0000
+++ ci-utils/ci_utils/amqp_utils.py	2014-05-14 13:40:29 +0000
@@ -28,12 +28,15 @@
 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:
         # try and find the config file, should be in the root of our bzr branch
-        f = os.path.join(os.path.dirname(__file__), '../../amqp_config.py')
+        f = os.path.join(HERE, '../../amqp_config.py')
         if os.path.exists(f):
             import imp
             config = imp.load_source('amqp_config', f)
@@ -53,7 +56,25 @@
     )
 
 
-def send(queue, msg, raise_errors=False):
+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
@@ -63,14 +84,12 @@
     config = get_config()
     if not config:
         return 'rabbitmq settings not available.'
-    con = channel = None
+    conn = channel = None
     try:
-        con = connection(config)
-        channel = con.channel()
-        channel.queue_declare(queue=queue, durable=True, auto_delete=False)
+        conn, channel = declare_queue(queue_name, config)
         body = amqp.Message(msg)
         body.properties['delivery_mode'] = 2  # Persistent
-        channel.basic_publish(body, routing_key=queue)
+        channel.basic_publish(exchange='', routing_key=queue_name, body=body)
     except Exception as e:
         if raise_errors:
             raise
@@ -79,12 +98,13 @@
     finally:
         if channel:
             channel.close()
-        if con:
-            con.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()
@@ -107,19 +127,16 @@
             channel.basic_cancel(tag)
 
 
-def process_queue(config, queue, callback, delete=False):
+def process_queue(config, queue_name, callback, delete=False):
     conn = channel = None
     try:
-        conn = connection(config)
-        channel = conn.channel()
-        channel.queue_declare(queue=queue, durable=True, auto_delete=False)
+        conn, channel = declare_queue(queue_name, config)
         channel.basic_qos(0, 1, False)
-        logging.info('Waiting for messages. ^C to exit.')
-        _run_forever(channel, queue, callback)
+        _run_forever(channel, queue_name, callback)
     finally:
         if channel:
             if delete:
-                channel.queue_delete(queue)
+                channel.queue_delete(queue_name)
             channel.close()
         if conn:
             conn.close()

=== modified file 'ci-utils/ci_utils/amqp_worker.py'
--- ci-utils/ci_utils/amqp_worker.py	2014-04-29 12:44:36 +0000
+++ ci-utils/ci_utils/amqp_worker.py	2014-05-14 13:40:29 +0000
@@ -24,8 +24,14 @@
 import urllib2
 import yaml
 
-from ci_utils import amqp_utils, data_store, dump_stack
-
+from ci_utils import (
+    amqp_utils,
+    data_store,
+    dump_stack,
+)
+
+
+HERE = os.path.abspath(os.path.dirname(__file__))
 logging.basicConfig(level=logging.INFO)
 log = logging.getLogger(__name__)
 
@@ -75,7 +81,7 @@
     # how often to check if the current task should be cancelled
     cancel_interval = 120
 
-    def handle_request(self, logger, params):
+    def handle_request(self, params, logger):
         '''To be implemented by the subclass to do their work
 
         amqp_cb should either be one of:
@@ -110,7 +116,7 @@
 
     def _create_data_store(self, ticket):
         # Note that this may be called multiple times
-        fname = os.path.join(os.path.dirname(__file__), '../../unit_config')
+        fname = os.path.join(HERE, '../../unit_config')
         with open(fname) as f:
             config = yaml.safe_load(f)
             return data_store.create_for_ticket(ticket, config)
@@ -189,7 +195,7 @@
             cancel_thread = self._handle_cancel(params)
 
             logger = self._create_worker_logger()
-            amqp_cb, ret = self.handle_request(logger, 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:

=== modified file 'ci-utils/ci_utils/gpg.py'
--- ci-utils/ci_utils/gpg.py	2014-05-08 01:58:36 +0000
+++ ci-utils/ci_utils/gpg.py	2014-05-14 13:40:29 +0000
@@ -1,5 +1,5 @@
 # Ubuntu Continuous Integration Engine
-# Copyright 2013, 2014 Canonical Ltd.
+# 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

=== modified file 'ci-utils/ci_utils/sourcecode.py'
--- ci-utils/ci_utils/sourcecode.py	2014-04-30 16:15:05 +0000
+++ ci-utils/ci_utils/sourcecode.py	2014-05-14 13:40:29 +0000
@@ -1,5 +1,5 @@
 #!/usr/bin/python
-# Copyright 2009 Canonical Ltd.  This software is licensed under the
+# Copyright 2014 Canonical Ltd.  This software is licensed under the
 # GNU Affero General Public License version 3 (see the file LICENSE).
 
 """Tools for maintaining the Launchpad source code."""

=== modified file 'ci-utils/ci_utils/testing/fixtures.py'
--- ci-utils/ci_utils/testing/fixtures.py	2014-05-06 14:16:05 +0000
+++ ci-utils/ci_utils/testing/fixtures.py	2014-05-14 13:40:29 +0000
@@ -77,3 +77,22 @@
 
     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):
+        while not self.is_empty():
+            func(self.msgs.pop(0), *args, **kwargs)
+
+    def is_empty(self):
+        return not self.msgs

=== modified file 'ci-utils/ci_utils/tests/test_amqp_worker.py'
--- ci-utils/ci_utils/tests/test_amqp_worker.py	2014-03-12 18:20:22 +0000
+++ ci-utils/ci_utils/tests/test_amqp_worker.py	2014-05-14 13:40:29 +0000
@@ -38,7 +38,7 @@
     def _create_data_store(self, ticket):
         return self.store
 
-    def handle_request(self, log, params):
+    def handle_request(self, params, log):
         log.info('called')
         self.params = params
         if self.exception:
@@ -146,7 +146,7 @@
             cmd = ['/bin/foo', 'bar']
             output = 'i can\'t believe its not butter'
 
-            def handle_request(self, log, params):
+            def handle_request(self, params, log):
                 raise subprocess.CalledProcessError(
                     self.rc, self.cmd, self.output)
 

=== modified file 'image-builder/imagebuilder/run_worker.py'
--- image-builder/imagebuilder/run_worker.py	2014-05-01 08:34:49 +0000
+++ image-builder/imagebuilder/run_worker.py	2014-05-14 13:40:29 +0000
@@ -23,7 +23,7 @@
     def __init__(self):
         super(ImageBuildWorker, self).__init__('image_build')
 
-    def handle_request(self, log, params):
+    def handle_request(self, params, log):
         image = params['image']
         repos = params['ppa_list']
         series = params['series']

=== modified file 'lander/bin/json_status_cgi.py'
--- lander/bin/json_status_cgi.py	2014-04-25 16:54:18 +0000
+++ lander/bin/json_status_cgi.py	2014-05-14 13:40:29 +0000
@@ -1,6 +1,6 @@
 #!/usr/bin/env python
 # Ubuntu CI Engine
-# Copyright 2013, 2014 Canonical Ltd.
+# 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

=== modified file 'lander/bin/lander_process_ticket.py'
--- lander/bin/lander_process_ticket.py	2014-05-07 16:25:13 +0000
+++ lander/bin/lander_process_ticket.py	2014-05-14 13:40:29 +0000
@@ -1,6 +1,6 @@
 #!/usr/bin/env python
 # Ubuntu CI Engine
-# Copyright 2013, 2014 Canonical Ltd.
+# 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

=== modified file 'test_runner/tstrun/run_worker.py'
--- test_runner/tstrun/run_worker.py	2014-04-16 18:36:03 +0000
+++ test_runner/tstrun/run_worker.py	2014-05-14 13:40:29 +0000
@@ -33,7 +33,7 @@
         """Save an artifact catching and reporting exceptions.
 
         This should only be used to upload artifacts after the pass/fail status
-        has been established to it's safe to fail the upload.
+        has been established so it's safe to fail the upload.
 
         :param logger: To report execution.
 
@@ -81,7 +81,7 @@
             summary_log_path, test_bed.get_remote_content(summary_log_path),
             'adt-run summary for {}'.format(package))
 
-    def handle_request(self, logger, params):
+    def handle_request(self, params, logger):
         ticket_id = params['ticket_id']
         progress_queue = params['progress_trigger']
         image_id = params['image_id']
@@ -97,7 +97,7 @@
         try:
             status_cb('Setting up the testbed for ticket {}'.format(ticket_id))
             flavors = tstrun.get_auth_config()['tr_flavors']
-            test_bed = testbed.TestBed('testbed-{}'.format(progress_queue),
+            test_bed = testbed.TestBed('testbed-{}'.format(ticket_id),
                                        flavors, image_id, status_cb)
         except:
             logger.exception(

=== modified file 'test_runner/tstrun/testbed.py'
--- test_runner/tstrun/testbed.py	2014-04-16 18:36:03 +0000
+++ test_runner/tstrun/testbed.py	2014-05-14 13:40:29 +0000
@@ -184,6 +184,7 @@
 
     def setup(self):
         flavor = self.find_flavor()
+        # FIXME: Better check for NotFound on wrong images -- vila 2014-05-12
         image = self.nova.images.find(name=self.image)
         self.create_user_data()
         self.instance = self.nova.servers.create(
@@ -332,10 +333,10 @@
     # whether canonistack or hpcloud is used.
     test_images = dict(
         cs_precise=('ubuntu-released/ubuntu-precise-12.04'
-                    '-amd64-server-20131205-disk1.img'),
+                    '-amd64-server-20140428-disk1.img'),
         cs_saucy=('ubuntu-released/ubuntu-saucy-13.10'
                   '-amd64-server-20140212-disk1.img'),
-        hp_precise=('Ubuntu Server 12.04.4 LTS (amd64 20140227)'
+        hp_precise=('Ubuntu Server 12.04.4 LTS (amd64 20140306)'
                     ' - CI Engineering'),
         hp_saucy=('Ubuntu Server 13.10 (amd64 20131030) - Partner Image'),
     )

=== added file 'test_runner/tstrun/tests/test_worker.py'
--- test_runner/tstrun/tests/test_worker.py	1970-01-01 00:00:00 +0000
+++ test_runner/tstrun/tests/test_worker.py	2014-05-14 13:40:29 +0000
@@ -0,0 +1,47 @@
+# 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 ci_utils.testing import (
+    features,
+    fixtures,
+)
+from tstrun import run_worker
+
+
[e-mail](features.nova_compute)
+class TestWorker(unittest.TestCase):
+
+    def setUp(self):
+        super(TestWorker, self).setUp()
+        self.logger = logging.getLogger()
+        self.queue = fixtures.FakeTaskQueue('test')
+
+    def test_handle_request(self):
+        worker = run_worker.TestRunnerWorker()
+        # FIXME: We need a way to get an image that is available to the nova
+        # setup we use. -- vila 20140512
+        params = dict(ticket_id='t1', progress_trigger='progress',
+                      image_id=('ubuntu-released/ubuntu-precise-12.04'
+                                '-amd64-server-20140428-disk1.img'),
+                      package_list=['libpng'])
+        self.queue.publish(params)
+        self.queue.subscribe(worker.handle_request, self.logger)
+        self.assertTrue(self.queue.is_empty())
+        # FIXME: More assertions could be added from the data store content, at
+        # least on artifacts present there. Revisit once we use the fake data
+        # store. -- vila 20140512

=== modified file 'tests/deployers.py'
--- tests/deployers.py	2014-05-12 14:17:51 +0000
+++ tests/deployers.py	2014-05-14 13:40:29 +0000
@@ -29,45 +29,33 @@
 # We need access to ci_utils
 sys.path.append(os.path.join(HERE, '..', 'ci-utils'))
 
-from ci_utils.testing import features, TestCaseWithGnupg
+from ci_utils import testing
+from ci_utils.testing import features
 
 
 class AmuletDeployment(object):
     '''Fixture for a deployed set of juju services from a deployer config'''
 
-    def __init__(self, deployer_cfg, juju_env=None, sentries=False):
+    def __init__(self, juju_env=None, sentries=False):
         self.deployment = amulet.Deployment(juju_env, sentries=sentries)
-        with open(deployer_cfg) as f:
-            jd_script = yaml.safe_load(f.read())
-        self.fixup_includes(os.path.dirname(deployer_cfg), jd_script)
+
+    def setUp(self, timeout):
+        self.deployment.setup(timeout)
+
+    def load(self, script):
         # Use juju_env given or defaulted above rather than taking the name
         # from the deployer config and assuming an env of that name exists
-        script = {self.deployment.juju_env: jd_script.values()[0]}
-        self.deployment.load(script)
-
-    def setUp(self, timeout):
-        self.deployment.setup(timeout)
-
-    def fixup_includes(self, path, jd_script):
-        # amulet doesn't deal with "include-base64" options, we have to
-        # patch the config ourselves
-        for deployment in jd_script:
-            for service in jd_script[deployment]['services'].values():
-                for key, val in service.get('options', {}).iteritems():
-                    include_token = 'include-base64://'
-                    if type(val) is str and val.startswith(include_token):
-                        val = val.replace(include_token, '')
-                        p = os.path.join(path, val)
-                        with open(p) as f:
-                            val = base64.b64encode(f.read())
-                        service['options'][key] = val
+        self.deployment.load({self.deployment.juju_env: script.values()[0]})
+
+    def status(self):
+        return amulet.waiter.status(self.deployment.juju_env)
 
     def tearDown(self):
         self.deployment.cleanup()
 
 
 @features.requires(features.bootstrapped_juju)
-class DeployerTest(TestCaseWithGnupg):
+class DeployerTest(testing.TestCaseWithGnupg):
     '''Base class for building juju deployer based tests.'''
 
     deployer_cfg = None
@@ -75,10 +63,29 @@
 
     def setUp(self):
         super(DeployerTest, self).setUp()
-        self.deployer = AmuletDeployment(
-            os.path.join(os.environ['JUJU_DEPLOYER_DIR'], self.deployer_cfg))
+        self.deployer = AmuletDeployment()
+        deployer_dir = os.environ['JUJU_DEPLOYER_DIR']
+        with open(os.path.join(deployer_dir, self.deployer_cfg)) as f:
+            script = yaml.safe_load(f.read())
+        self.fixup_includes(script, deployer_dir)
+        self.deployer.load(script)
         self.addCleanup(self.deployer.tearDown)
         self.deployer.setUp(self.timeout)
+        self.status =  self.deployer.status()
+
+    def fixup_includes(self, script, base_dir):
+        # amulet can only process "include-base64" option if the file is under
+        # a relative 'tests' dir, so we process the option ourselves.
+        for deployment in script:
+            for service in script[deployment]['services'].values():
+                for key, val in service.get('options', {}).iteritems():
+                    include_token = 'include-base64://'
+                    if type(val) is str and val.startswith(include_token):
+                        val = val.replace(include_token, '')
+                        include_path = os.path.join(base_dir, val)
+                        with open(include_path) as f:
+                            val = base64.b64encode(f.read())
+                        service['options'][key] = val
 
     def get_ip_and_port(self, service, unit=0):
         tries = 0

=== added file 'tests/test_rabbit.py'
--- tests/test_rabbit.py	1970-01-01 00:00:00 +0000
+++ tests/test_rabbit.py	2014-05-14 13:40:29 +0000
@@ -0,0 +1,149 @@
+# 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 subprocess
+import unittest
+import yaml
+
+
+import deployers
+from ucitests import fixtures
+
+
+from ci_utils import amqp_utils
+from ci_utils.testing import features
+
+
+def juju(args):
+    try:
+        p = subprocess.Popen(['juju'] + args,
+                             stdout=subprocess.PIPE,
+                             stderr=subprocess.PIPE)
+    except OSError as e:
+        if e.errno != errno.ENOENT:
+            raise
+        raise OSError("juju not found, do you have Juju installed?")
+    out, err = p.communicate()
+    if p.returncode:
+        raise IOError('juju command failed {!r}:\n{}'.format(args, err))
+    return out
+
+
+class RabbitServer(object):
+
+    timeout = 600
+
+    def __init__(self):
+        self.deployer = None
+
+    def setUp(self):
+        just_rabbit ='''
+ci-airline-staging:
+    series: precise
+    services:
+        rabbit:
+            to: 0
+            branch: lp:~canonical-ci-engineering/charms/precise/uci-engine/rabbitmq-server@46
+            charm: rabbitmq
+            options:
+                management_plugin: true
+    relations: []
+'''
+        self.deployer = deployers.AmuletDeployment()
+        self.deployer.load(yaml.safe_load(just_rabbit))
+        self.deployer.setUp(self.timeout)
+        status = self.deployer.status()
+        units = status['services']['rabbit']['units']
+        self.unit = 'rabbit/0'
+        self.public_ip = units[self.unit]['public-address']
+        # So caller can handle tearDown/cleanup
+        return self.deployer.tearDown
+
+    def run_ctl_cmd(self, args):
+        cmd = ' '.join(args)
+        juju(['run', '--unit', self.unit, 'sudo rabbitmqctl {}'.format(cmd)])
+
+    def create_user(self, user_name, password=None):
+        if password is None:
+            password = user_name
+        self.run_ctl_cmd(['add_user', user_name, password])
+        self.run_ctl_cmd(['set_permissions', '-p', '/',
+                          user_name, '".*" ".*" ".*"'])
+
+    def delete_user(self, user_name):
+        self.run_ctl_cmd(['delete_user', user_name])
+
+
+
[e-mail](features.bootstrapped_juju)
+class TestRabbit(unittest.TestCase):
+
+    timeout = 600
+
+    def setUp(self):
+        super(TestRabbit, self).setUp()
+        self.rabbit = RabbitServer()
+        tearDown = self.rabbit.setUp()
+        self.addCleanup(tearDown)
+        self.rabbit.create_user('tester', 's3cr3t')
+        self.addCleanup(self.rabbit.delete_user, 'tester')
+
+        class AmqpConfig(object):
+
+            def __init__(inner):
+                inner.AMQP_HOST = self.rabbit.public_ip
+                inner.AMQP_VHOST = '/'
+                inner.AMQP_USER = 'tester'
+                inner.AMQP_PASSWORD = 's3cr3t'
+        fixtures.patch(self, amqp_utils, 'get_config', AmqpConfig)
+
+    def test_acked_message(self):
+        amqp_utils.send('myqueue', 'hello')
+        conn, channel = amqp_utils.declare_queue('myqueue')
+        self.addCleanup(conn.close)
+        self.addCleanup(channel.close)
+        self.addCleanup(channel.queue_delete, 'myqueue')
+        channel.basic_qos(prefetch_size=0, prefetch_count=1, a_global=False)
+        def ack(msg):
+            self.assertEqual('hello', msg.body)
+            msg.channel.basic_ack(msg.delivery_tag)
+
+        channel.basic_consume(callback=ack, queue='myqueue')
+
+    def test_nacked_message(self):
+        amqp_utils.send('myqueue', 'hello')
+        conn, channel = amqp_utils.declare_queue('myqueue')
+        self.addCleanup(conn.close)
+        self.addCleanup(channel.close)
+        self.addCleanup(channel.queue_delete, 'myqueue')
+        channel.basic_qos(prefetch_size=0, prefetch_count=1, a_global=False)
+        def nack(msg):
+            self.assertEqual('hello', msg.body)
+            msg.channel.basic_reject(msg.delivery_tag, requeue=True)
+
+        channel.basic_consume(callback=nack, queue='myqueue')
+
+        # Use a different channel to get the message again
+        conn, channel = amqp_utils.declare_queue('myqueue')
+        self.addCleanup(conn.close)
+        self.addCleanup(channel.close)
+        # We don't need to delete the queue during cleanup, it's already done
+        # above.
+        channel.basic_qos(prefetch_size=0, prefetch_count=1, a_global=False)
+        def ack(msg):
+            self.assertEqual('hello', msg.body)
+            msg.channel.basic_ack(msg.delivery_tag)
+        channel.basic_consume(callback=ack, queue='myqueue')