pull-up from trunk

22 files changed, 331 insertions, 372 deletions

diff --git a/Authors b/Authors
index 1a07946bd..0e5931462 100644
--- a/Authors
+++ b/Authors
@@ -67,6 +67,7 @@ Lvov Maxim <usrleon@gmail.com>
 Mark Washenberger <mark.washenberger@rackspace.com>
 Masanori Itoh <itoumsn@nttdata.co.jp>
 Matt Dietz <matt.dietz@rackspace.com>
+Matthew Hooker <matt@cloudscaling.com>
 Michael Gundlach <michael.gundlach@rackspace.com>
 Mike Scherbakov <mihgen@gmail.com>
 Mohammed Naser <mnaser@vexxhost.com>
@@ -105,3 +106,4 @@ Yoshiaki Tamura <yoshi@midokura.jp>
 Youcef Laribi <Youcef.Laribi@eu.citrix.com>
 Yuriy Taraday <yorik.sar@gmail.com>
 Zhixue Wu <Zhixue.Wu@citrix.com>
+Zed Shaw <zedshaw@zedshaw.com>
diff --git a/nova/api/openstack/servers.py b/nova/api/openstack/servers.py
index f6841318d..30169d450 100644
--- a/nova/api/openstack/servers.py
+++ b/nova/api/openstack/servers.py
@@ -17,11 +17,10 @@ import base64
 import traceback
 
 from webob import exc
-import webob
 from xml.dom import minidom
+import webob
 
 from nova import compute
-from nova import db
 from nova import exception
 from nova import flags
 from nova import log as logging
@@ -29,13 +28,14 @@ from nova import utils
 from nova.api.openstack import common
 from nova.api.openstack import create_instance_helper as helper
 from nova.api.openstack import ips
+from nova.api.openstack import wsgi
+from nova.compute import instance_types
+from nova.scheduler import api as scheduler_api
+import nova.api.openstack
 import nova.api.openstack.views.addresses
 import nova.api.openstack.views.flavors
 import nova.api.openstack.views.images
 import nova.api.openstack.views.servers
-from nova.api.openstack import wsgi
-import nova.api.openstack
-from nova.scheduler import api as scheduler_api
 
 
 LOG = logging.getLogger('nova.api.openstack.servers')
@@ -438,13 +438,21 @@ class ControllerV10(Controller):
 
     def _action_resize(self, input_dict, req, id):
         """ Resizes a given instance to the flavor size requested """
-        if 'resize' in input_dict and 'flavorId' in input_dict['resize']:
-            flavor_id = input_dict['resize']['flavorId']
-            self.compute_api.resize(req.environ['nova.context'], id,
-                    flavor_id)
-        else:
-            LOG.exception(_("Missing 'flavorId' argument for resize"))
-            raise exc.HTTPUnprocessableEntity()
+        try:
+            flavor_id = input_dict["resize"]["flavorId"]
+        except (KeyError, TypeError):
+            msg = _("Resize requests require 'flavorId' attribute.")
+            raise exc.HTTPBadRequest(explanation=msg)
+
+        try:
+            i_type = instance_types.get_instance_type_by_flavor_id(flavor_id)
+        except exception.FlavorNotFound:
+            msg = _("Unable to locate requested flavor.")
+            raise exc.HTTPBadRequest(explanation=msg)
+
+        context = req.environ["nova.context"]
+        self.compute_api.resize(context, id, i_type["id"])
+
         return webob.Response(status_int=202)
 
     def _action_rebuild(self, info, request, instance_id):
@@ -555,17 +563,20 @@ class ControllerV11(Controller):
     def _action_resize(self, input_dict, req, id):
         """ Resizes a given instance to the flavor size requested """
         try:
-            if 'resize' in input_dict and 'flavorRef' in input_dict['resize']:
-                flavor_ref = input_dict['resize']['flavorRef']
-                flavor_id = common.get_id_from_href(flavor_ref)
-                self.compute_api.resize(req.environ['nova.context'], id,
-                        flavor_id)
-            else:
-                LOG.exception(_("Missing 'flavorRef' argument for resize"))
-                raise exc.HTTPUnprocessableEntity()
-        except Exception, e:
-            LOG.exception(_("Error in resize %s"), e)
-            raise exc.HTTPBadRequest()
+            flavor_ref = input_dict["resize"]["flavorRef"]
+        except (KeyError, TypeError):
+            msg = _("Resize requests require 'flavorRef' attribute.")
+            raise exc.HTTPBadRequest(explanation=msg)
+
+        try:
+            i_type = instance_types.get_instance_type_by_flavor_id(flavor_ref)
+        except exception.FlavorNotFound:
+            msg = _("Unable to locate requested flavor.")
+            raise exc.HTTPBadRequest(explanation=msg)
+
+        context = req.environ["nova.context"]
+        self.compute_api.resize(context, id, i_type["id"])
+
         return webob.Response(status_int=202)
 
     def _action_rebuild(self, info, request, instance_id):
diff --git a/nova/compute/instance_types.py b/nova/compute/instance_types.py
index c13a629a9..824416514 100644
--- a/nova/compute/instance_types.py
+++ b/nova/compute/instance_types.py
@@ -132,11 +132,8 @@ def get_instance_type_by_name(name):
 #               flavors.
 def get_instance_type_by_flavor_id(flavor_id):
     """Retrieve instance type by flavor_id."""
-    if flavor_id is None:
-        return get_default_instance_type()
+    ctxt = context.get_admin_context()
     try:
-        ctxt = context.get_admin_context()
         return db.instance_type_get_by_flavor_id(ctxt, flavor_id)
-    except exception.DBError, e:
-        LOG.exception(_('DB error: %s') % e)
-        raise exception.ApiError(_("Unknown flavor: %s") % flavor_id)
+    except ValueError:
+        raise exception.FlavorNotFound(flavor_id=flavor_id)
diff --git a/nova/db/sqlalchemy/api.py b/nova/db/sqlalchemy/api.py
index d7810098a..0c3c5af6a 100644
--- a/nova/db/sqlalchemy/api.py
+++ b/nova/db/sqlalchemy/api.py
@@ -1312,7 +1312,7 @@ def instance_get_fixed_addresses_v6(context, instance_id):
         # combine prefixes, macs, and project_id into (prefix,mac,p_id) tuples
         prefix_mac_tuples = zip(prefixes, macs, [project_id for m in macs])
         # return list containing ipv6 address for each tuple
-        return [ipv6.to_global_ipv6(*t) for t in prefix_mac_tuples]
+        return [ipv6.to_global(*t) for t in prefix_mac_tuples]
 
 
 @require_context
@@ -3247,8 +3247,8 @@ def agent_build_destroy(context, agent_build_id):
     with session.begin():
         session.query(models.AgentBuild).\
                 filter_by(id=agent_build_id).\
-                update({'deleted': 1,
-                        'deleted_at': datetime.datetime.utcnow(),
+                update({'deleted': True,
+                        'deleted_at': utils.utcnow(),
                         'updated_at': literal_column('updated_at')})
 
 
@@ -3297,7 +3297,7 @@ def instance_type_extra_specs_delete(context, instance_type_id, key):
 def instance_type_extra_specs_get_item(context, instance_type_id, key):
     session = get_session()
 
-    sppec_result = session.query(models.InstanceTypeExtraSpecs).\
+    spec_result = session.query(models.InstanceTypeExtraSpecs).\
                     filter_by(instance_type_id=instance_type_id).\
                     filter_by(key=key).\
                     filter_by(deleted=False).\
diff --git a/nova/rpc/__init__.py b/nova/rpc/__init__.py
new file mode 100644
index 000000000..bdf7f705b
--- /dev/null
+++ b/nova/rpc/__init__.py
@@ -0,0 +1,66 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2010 United States Government as represented by the
+# Administrator of the National Aeronautics and Space Administration.
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+
+from nova.utils import import_object
+from nova.rpc.common import RemoteError, LOG
+from nova import flags
+
+FLAGS = flags.FLAGS
+flags.DEFINE_string('rpc_backend',
+                    'nova.rpc.amqp',
+                    "The messaging module to use, defaults to AMQP.")
+
+RPCIMPL = import_object(FLAGS.rpc_backend)
+
+
+def create_connection(new=True):
+    return RPCIMPL.Connection.instance(new=True)
+
+
+def create_consumer(conn, topic, proxy, fanout=False):
+    if fanout:
+        return RPCIMPL.FanoutAdapterConsumer(
+                connection=conn,
+                topic=topic,
+                proxy=proxy)
+    else:
+        return RPCIMPL.TopicAdapterConsumer(
+                connection=conn,
+                topic=topic,
+                proxy=proxy)
+
+
+def create_consumer_set(conn, consumers):
+    return RPCIMPL.ConsumerSet(connection=conn, consumer_list=consumers)
+
+
+def call(context, topic, msg):
+    return RPCIMPL.call(context, topic, msg)
+
+
+def cast(context, topic, msg):
+    return RPCIMPL.cast(context, topic, msg)
+
+
+def fanout_cast(context, topic, msg):
+    return RPCIMPL.fanout_cast(context, topic, msg)
+
+
+def multicall(context, topic, msg):
+    return RPCIMPL.multicall(context, topic, msg)
diff --git a/nova/rpc.py b/nova/rpc/amqp.py
index e2771ca88..61555795a 100644
--- a/nova/rpc.py
+++ b/nova/rpc/amqp.py
@@ -44,9 +44,7 @@ from nova import fakerabbit
 from nova import flags
 from nova import log as logging
 from nova import utils
-
-
-LOG = logging.getLogger('nova.rpc')
+from nova.rpc.common import RemoteError, LOG
 
 
 FLAGS = flags.FLAGS
@@ -418,25 +416,6 @@ def msg_reply(msg_id, reply=None, failure=None):
         publisher.close()
 
 
-class RemoteError(exception.Error):
-    """Signifies that a remote class has raised an exception.
-
-    Containes a string representation of the type of the original exception,
-    the value of the original exception, and the traceback.  These are
-    sent to the parent as a joined string so printing the exception
-    contains all of the relevent info.
-
-    """
-
-    def __init__(self, exc_type, value, traceback):
-        self.exc_type = exc_type
-        self.value = value
-        self.traceback = traceback
-        super(RemoteError, self).__init__('%s %s\n%s' % (exc_type,
-                                                         value,
-                                                         traceback))
-
-
 def _unpack_context(msg):
     """Unpack context from msg."""
     context_dict = {}
diff --git a/nova/rpc/common.py b/nova/rpc/common.py
new file mode 100644
index 000000000..1d3065a83
--- /dev/null
+++ b/nova/rpc/common.py
@@ -0,0 +1,23 @@
+from nova import exception
+from nova import log as logging
+
+LOG = logging.getLogger('nova.rpc')
+
+
+class RemoteError(exception.Error):
+    """Signifies that a remote class has raised an exception.
+
+    Containes a string representation of the type of the original exception,
+    the value of the original exception, and the traceback.  These are
+    sent to the parent as a joined string so printing the exception
+    contains all of the relevent info.
+
+    """
+
+    def __init__(self, exc_type, value, traceback):
+        self.exc_type = exc_type
+        self.value = value
+        self.traceback = traceback
+        super(RemoteError, self).__init__('%s %s\n%s' % (exc_type,
+                                                         value,
+                                                         traceback))
diff --git a/nova/scheduler/least_cost.py b/nova/scheduler/least_cost.py
index 6f5eb66fd..8c400d476 100644
--- a/nova/scheduler/least_cost.py
+++ b/nova/scheduler/least_cost.py
@@ -28,6 +28,7 @@ from nova import flags
 from nova import log as logging
 from nova.scheduler import zone_aware_scheduler
 from nova import utils
+from nova import exception
 
 LOG = logging.getLogger('nova.scheduler.least_cost')
 
diff --git a/nova/scheduler/zone_aware_scheduler.py b/nova/scheduler/zone_aware_scheduler.py
index c429fdfcc..d99d7214c 100644
--- a/nova/scheduler/zone_aware_scheduler.py
+++ b/nova/scheduler/zone_aware_scheduler.py
@@ -81,7 +81,7 @@ class ZoneAwareScheduler(driver.Scheduler):
         decryptor = crypto.decryptor(FLAGS.build_plan_encryption_key)
         try:
             json_entry = decryptor(blob)
-            return json.dumps(entry)
+            return json.dumps(json_entry)
         except M2Crypto.EVP.EVPError:
             pass
         return None
diff --git a/nova/service.py b/nova/service.py
index 00e4f61e5..6e9eddc5a 100644
--- a/nova/service.py
+++ b/nova/service.py
@@ -149,26 +149,22 @@ class Service(object):
         if 'nova-compute' == self.binary:
             self.manager.update_available_resource(ctxt)
 
-        self.conn = rpc.Connection.instance(new=True)
+        self.conn = rpc.create_connection(new=True)
         logging.debug("Creating Consumer connection for Service %s" %
                       self.topic)
 
         # Share this same connection for these Consumers
-        consumer_all = rpc.TopicAdapterConsumer(
-                connection=self.conn,
-                topic=self.topic,
-                proxy=self)
-        consumer_node = rpc.TopicAdapterConsumer(
-                connection=self.conn,
-                topic='%s.%s' % (self.topic, self.host),
-                proxy=self)
-        fanout = rpc.FanoutAdapterConsumer(
-                connection=self.conn,
-                topic=self.topic,
-                proxy=self)
-        consumer_set = rpc.ConsumerSet(
-                connection=self.conn,
-                consumer_list=[consumer_all, consumer_node, fanout])
+        consumer_all = rpc.create_consumer(self.conn, self.topic, self,
+                                           fanout=False)
+
+        node_topic = '%s.%s' % (self.topic, self.host)
+        consumer_node = rpc.create_consumer(self.conn, node_topic, self,
+                                            fanout=False)
+
+        fanout = rpc.create_consumer(self.conn, self.topic, self, fanout=True)
+
+        consumers = [consumer_all, consumer_node, fanout]
+        consumer_set = rpc.create_consumer_set(self.conn, consumers)
 
         # Wait forever, processing these consumers
         def _wait():
diff --git a/nova/test.py b/nova/test.py
index 9790b0aa1..549aa6fcf 100644
--- a/nova/test.py
+++ b/nova/test.py
@@ -99,9 +99,7 @@ class TestCase(unittest.TestCase):
         self.flag_overrides = {}
         self.injected = []
         self._services = []
-        self._monkey_patch_attach()
         self._original_flags = FLAGS.FlagValuesDict()
-        rpc.ConnectionPool = rpc.Pool(max_size=FLAGS.rpc_conn_pool_size)
 
     def tearDown(self):
         """Runs after each test method to tear down test environment."""
@@ -126,9 +124,6 @@ class TestCase(unittest.TestCase):
             # Reset any overriden flags
             self.reset_flags()
 
-            # Reset our monkey-patches
-            rpc.Consumer.attach_to_eventlet = self.original_attach
-
             # Stop any timers
             for x in self.injected:
                 try:
@@ -172,17 +167,6 @@ class TestCase(unittest.TestCase):
         self._services.append(svc)
         return svc
 
-    def _monkey_patch_attach(self):
-        self.original_attach = rpc.Consumer.attach_to_eventlet
-
-        def _wrapped(inner_self):
-            rv = self.original_attach(inner_self)
-            self.injected.append(rv)
-            return rv
-
-        _wrapped.func_name = self.original_attach.func_name
-        rpc.Consumer.attach_to_eventlet = _wrapped
-
     # Useful assertions
     def assertDictMatch(self, d1, d2, approx_equal=False, tolerance=0.001):
         """Assert two dicts are equivalent.
diff --git a/nova/tests/api/openstack/test_servers.py b/nova/tests/api/openstack/test_servers.py
index 4027ef829..221c1f4f7 100644
--- a/nova/tests/api/openstack/test_servers.py
+++ b/nova/tests/api/openstack/test_servers.py
@@ -2167,11 +2167,14 @@ class ServersTest(test.TestCase):
         self.assertEqual(self.resize_called, True)
 
     def test_resize_server_v11(self):
-
         req = webob.Request.blank('/v1.1/servers/1/action')
         req.content_type = 'application/json'
         req.method = 'POST'
-        body_dict = dict(resize=dict(flavorRef="http://localhost/3"))
+        body_dict = {
+            "resize": {
+                "flavorRef": 3,
+            },
+        }
         req.body = json.dumps(body_dict)
 
         self.resize_called = False
@@ -2185,8 +2188,8 @@ class ServersTest(test.TestCase):
         self.assertEqual(res.status_int, 202)
         self.assertEqual(self.resize_called, True)
 
-    def test_resize_bad_flavor_fails(self):
-        req = self.webreq('/1/action', 'POST', dict(resize=dict(derp=3)))
+    def test_resize_bad_flavor_data(self):
+        req = self.webreq('/1/action', 'POST', {"resize": "bad_data"})
 
         self.resize_called = False
 
@@ -2196,14 +2199,54 @@ class ServersTest(test.TestCase):
         self.stubs.Set(nova.compute.api.API, 'resize', resize_mock)
 
         res = req.get_response(fakes.wsgi_app())
-        self.assertEqual(res.status_int, 422)
+        self.assertEqual(res.status_int, 400)
         self.assertEqual(self.resize_called, False)
 
+    def test_resize_invalid_flavorid(self):
+        req = self.webreq('/1/action', 'POST', {"resize": {"flavorId": 300}})
+        res = req.get_response(fakes.wsgi_app())
+        self.assertEqual(res.status_int, 400)
+
+    def test_resize_nonint_flavorid(self):
+        req = self.webreq('/1/action', 'POST', {"resize": {"flavorId": "a"}})
+        res = req.get_response(fakes.wsgi_app())
+        self.assertEqual(res.status_int, 400)
+
+    def test_resize_invalid_flavorid_v1_1(self):
+        req = webob.Request.blank('/v1.1/servers/1/action')
+        req.content_type = 'application/json'
+        req.method = 'POST'
+        resize_body = {
+            "resize": {
+                "image": {
+                    "id": 300,
+                },
+            },
+        }
+        req.body = json.dumps(resize_body)
+        res = req.get_response(fakes.wsgi_app())
+        self.assertEqual(res.status_int, 400)
+
+    def test_resize_nonint_flavorid_v1_1(self):
+        req = webob.Request.blank('/v1.1/servers/1/action')
+        req.content_type = 'application/json'
+        req.method = 'POST'
+        resize_body = {
+            "resize": {
+                "image": {
+                    "id": "a",
+                },
+            },
+        }
+        req.body = json.dumps(resize_body)
+        res = req.get_response(fakes.wsgi_app())
+        self.assertEqual(res.status_int, 400)
+
     def test_resize_raises_fails(self):
         req = self.webreq('/1/action', 'POST', dict(resize=dict(flavorId=3)))
 
         def resize_mock(*args):
-            raise Exception('hurr durr')
+            raise Exception("An error occurred.")
 
         self.stubs.Set(nova.compute.api.API, 'resize', resize_mock)
 
@@ -2241,7 +2284,7 @@ class ServersTest(test.TestCase):
         req = self.webreq('/1/action', 'POST', dict(confirmResize=None))
 
         def confirm_resize_mock(*args):
-            raise Exception('hurr durr')
+            raise Exception("An error occurred.")
 
         self.stubs.Set(nova.compute.api.API, 'confirm_resize',
                 confirm_resize_mock)
@@ -2268,7 +2311,7 @@ class ServersTest(test.TestCase):
         req = self.webreq('/1/action', 'POST', dict(revertResize=None))
 
         def revert_resize_mock(*args):
-            raise Exception('hurr durr')
+            raise Exception("An error occurred.")
 
         self.stubs.Set(nova.compute.api.API, 'revert_resize',
                 revert_resize_mock)
diff --git a/nova/tests/scheduler/test_scheduler.py b/nova/tests/scheduler/test_scheduler.py
index daea826fd..18fbef5ca 100644
--- a/nova/tests/scheduler/test_scheduler.py
+++ b/nova/tests/scheduler/test_scheduler.py
@@ -485,11 +485,6 @@ class SimpleDriverTestCase(test.TestCase):
         self.assertEqual(host, 'host2')
         volume1.delete_volume(self.context, volume_id1)
         db.volume_destroy(self.context, volume_id2)
-        dic = {'service_id': s_ref['id'],
-               'vcpus': 16, 'memory_mb': 32, 'local_gb': 100,
-               'vcpus_used': 16, 'memory_mb_used': 12, 'local_gb_used': 10,
-               'hypervisor_type': 'qemu', 'hypervisor_version': 12003,
-               'cpu_info': ''}
 
     def test_doesnt_report_disabled_hosts_as_up(self):
         """Ensures driver doesn't find hosts before they are enabled"""
diff --git a/nova/tests/scheduler/test_zone_aware_scheduler.py b/nova/tests/scheduler/test_zone_aware_scheduler.py
index d74b71fb6..7833028c3 100644
--- a/nova/tests/scheduler/test_zone_aware_scheduler.py
+++ b/nova/tests/scheduler/test_zone_aware_scheduler.py
@@ -16,6 +16,8 @@
 Tests For Zone Aware Scheduler.
 """
 
+import json
+
 import nova.db
 
 from nova import exception
@@ -327,3 +329,19 @@ class ZoneAwareSchedulerTestCase(test.TestCase):
         sched._provision_resource_from_blob(None, request_spec, 1,
                                             request_spec, {})
         self.assertTrue(was_called)
+
+    def test_decrypt_blob(self):
+        """Test that the decrypt method works."""
+
+        fixture = FakeZoneAwareScheduler()
+        test_data = {"foo": "bar"}
+
+        class StubDecryptor(object):
+            def decryptor(self, key):
+                return lambda blob: blob
+
+        self.stubs.Set(zone_aware_scheduler, 'crypto',
+                       StubDecryptor())
+
+        self.assertEqual(fixture._decrypt_blob(test_data),
+                         json.dumps(test_data))
diff --git a/nova/tests/test_adminapi.py b/nova/tests/test_adminapi.py
index 877cf4ea1..6bbe15f53 100644
--- a/nova/tests/test_adminapi.py
+++ b/nova/tests/test_adminapi.py
@@ -39,7 +39,7 @@ class AdminApiTestCase(test.TestCase):
         super(AdminApiTestCase, self).setUp()
         self.flags(connection_type='fake')
 
-        self.conn = rpc.Connection.instance()
+        self.conn = rpc.create_connection()
 
         # set up our cloud
         self.api = admin.AdminController()
diff --git a/nova/tests/test_cloud.py b/nova/tests/test_cloud.py
index e419e7a50..50df344d6 100644
--- a/nova/tests/test_cloud.py
+++ b/nova/tests/test_cloud.py
@@ -50,7 +50,7 @@ class CloudTestCase(test.TestCase):
         self.flags(connection_type='fake',
                    stub_network=True)
 
-        self.conn = rpc.Connection.instance()
+        self.conn = rpc.create_connection()
 
         # set up our cloud
         self.cloud = cloud.CloudController()
@@ -326,22 +326,15 @@ class CloudTestCase(test.TestCase):
         revoke = self.cloud.revoke_security_group_ingress
         self.assertTrue(revoke(self.context, group_name=sec['name'], **kwargs))
 
-    def test_revoke_security_group_ingress_by_id(self):
-        kwargs = {'project_id': self.context.project_id, 'name': 'test'}
-        sec = db.security_group_create(self.context, kwargs)
-        authz = self.cloud.authorize_security_group_ingress
-        kwargs = {'to_port': '999', 'from_port': '999', 'ip_protocol': 'tcp'}
-        authz(self.context, group_id=sec['id'], **kwargs)
-        revoke = self.cloud.revoke_security_group_ingress
-        self.assertTrue(revoke(self.context, group_id=sec['id'], **kwargs))
-
-    def test_authorize_security_group_ingress_by_id(self):
+    def test_authorize_revoke_security_group_ingress_by_id(self):
         sec = db.security_group_create(self.context,
                                        {'project_id': self.context.project_id,
                                         'name': 'test'})
         authz = self.cloud.authorize_security_group_ingress
         kwargs = {'to_port': '999', 'from_port': '999', 'ip_protocol': 'tcp'}
-        self.assertTrue(authz(self.context, group_id=sec['id'], **kwargs))
+        authz(self.context, group_id=sec['id'], **kwargs)
+        revoke = self.cloud.revoke_security_group_ingress
+        self.assertTrue(revoke(self.context, group_id=sec['id'], **kwargs))
 
     def test_authorize_security_group_ingress_missing_protocol_params(self):
         sec = db.security_group_create(self.context,
@@ -961,21 +954,6 @@ class CloudTestCase(test.TestCase):
         self._wait_for_running(ec2_instance_id)
         return ec2_instance_id
 
-    def test_rescue_unrescue_instance(self):
-        instance_id = self._run_instance(
-            image_id='ami-1',
-            instance_type=FLAGS.default_instance_type,
-            max_count=1)
-        self.cloud.rescue_instance(context=self.context,
-                                   instance_id=instance_id)
-        # NOTE(vish): This currently does no validation, it simply makes sure
-        #             that the code path doesn't throw an exception.
-        self.cloud.unrescue_instance(context=self.context,
-                                   instance_id=instance_id)
-        # TODO(soren): We need this until we can stop polling in the rpc code
-        #              for unit tests.
-        self.cloud.terminate_instances(self.context, [instance_id])
-
     def test_console_output(self):
         instance_id = self._run_instance(
             image_id='ami-1',
diff --git a/nova/tests/test_rpc.py b/nova/tests/test_rpc.py
index ffd748efe..2d2436175 100644
--- a/nova/tests/test_rpc.py
+++ b/nova/tests/test_rpc.py
@@ -33,11 +33,12 @@ LOG = logging.getLogger('nova.tests.rpc')
 class RpcTestCase(test.TestCase):
     def setUp(self):
         super(RpcTestCase, self).setUp()
-        self.conn = rpc.Connection.instance(True)
+        self.conn = rpc.create_connection(True)
         self.receiver = TestReceiver()
-        self.consumer = rpc.TopicAdapterConsumer(connection=self.conn,
-                                            topic='test',
-                                            proxy=self.receiver)
+        self.consumer = rpc.create_consumer(self.conn,
+                                            'test',
+                                            self.receiver,
+                                            False)
         self.consumer.attach_to_eventlet()
         self.context = context.get_admin_context()
 
@@ -129,6 +130,8 @@ class RpcTestCase(test.TestCase):
                 """Calls echo in the passed queue"""
                 LOG.debug(_("Nested received %(queue)s, %(value)s")
                         % locals())
+                # TODO: so, it will replay the context and use the same REQID?
+                # that's bizarre.
                 ret = rpc.call(context,
                                queue,
                                {"method": "echo",
@@ -137,10 +140,11 @@ class RpcTestCase(test.TestCase):
                 return value
 
         nested = Nested()
-        conn = rpc.Connection.instance(True)
-        consumer = rpc.TopicAdapterConsumer(connection=conn,
-                                       topic='nested',
-                                       proxy=nested)
+        conn = rpc.create_connection(True)
+        consumer = rpc.create_consumer(conn,
+                                       'nested',
+                                       nested,
+                                       False)
         consumer.attach_to_eventlet()
         value = 42
         result = rpc.call(self.context,
@@ -149,47 +153,6 @@ class RpcTestCase(test.TestCase):
                                               "value": value}})
         self.assertEqual(value, result)
 
-    def test_connectionpool_single(self):
-        """Test that ConnectionPool recycles a single connection."""
-        conn1 = rpc.ConnectionPool.get()
-        rpc.ConnectionPool.put(conn1)
-        conn2 = rpc.ConnectionPool.get()
-        rpc.ConnectionPool.put(conn2)
-        self.assertEqual(conn1, conn2)
-
-    def test_connectionpool_double(self):
-        """Test that ConnectionPool returns and reuses separate connections.
-
-        When called consecutively we should get separate connections and upon
-        returning them those connections should be reused for future calls
-        before generating a new connection.
-
-        """
-        conn1 = rpc.ConnectionPool.get()
-        conn2 = rpc.ConnectionPool.get()
-
-        self.assertNotEqual(conn1, conn2)
-        rpc.ConnectionPool.put(conn1)
-        rpc.ConnectionPool.put(conn2)
-
-        conn3 = rpc.ConnectionPool.get()
-        conn4 = rpc.ConnectionPool.get()
-        self.assertEqual(conn1, conn3)
-        self.assertEqual(conn2, conn4)
-
-    def test_connectionpool_limit(self):
-        """Test connection pool limit and connection uniqueness."""
-        max_size = FLAGS.rpc_conn_pool_size
-        conns = []
-
-        for i in xrange(max_size):
-            conns.append(rpc.ConnectionPool.get())
-
-        self.assertFalse(rpc.ConnectionPool.free_items)
-        self.assertEqual(rpc.ConnectionPool.current_size,
-                rpc.ConnectionPool.max_size)
-        self.assertEqual(len(set(conns)), max_size)
-
 
 class TestReceiver(object):
     """Simple Proxy class so the consumer has methods to call.
diff --git a/nova/tests/test_rpc_amqp.py b/nova/tests/test_rpc_amqp.py
new file mode 100644
index 000000000..d29f7ae32
--- /dev/null
+++ b/nova/tests/test_rpc_amqp.py
@@ -0,0 +1,68 @@
+from nova import context
+from nova import flags
+from nova import log as logging
+from nova import rpc
+from nova.rpc import amqp
+from nova import test
+
+
+FLAGS = flags.FLAGS
+LOG = logging.getLogger('nova.tests.rpc')
+
+
+class RpcAMQPTestCase(test.TestCase):
+    def setUp(self):
+        super(RpcAMQPTestCase, self).setUp()
+        self.conn = rpc.create_connection(True)
+        self.receiver = TestReceiver()
+        self.consumer = rpc.create_consumer(self.conn,
+                                            'test',
+                                            self.receiver,
+                                            False)
+        self.consumer.attach_to_eventlet()
+        self.context = context.get_admin_context()
+
+    def test_connectionpool_single(self):
+        """Test that ConnectionPool recycles a single connection."""
+        conn1 = amqp.ConnectionPool.get()
+        amqp.ConnectionPool.put(conn1)
+        conn2 = amqp.ConnectionPool.get()
+        amqp.ConnectionPool.put(conn2)
+        self.assertEqual(conn1, conn2)
+
+
+class TestReceiver(object):
+    """Simple Proxy class so the consumer has methods to call.
+
+    Uses static methods because we aren't actually storing any state.
+
+    """
+
+    @staticmethod
+    def echo(context, value):
+        """Simply returns whatever value is sent in."""
+        LOG.debug(_("Received %s"), value)
+        return value
+
+    @staticmethod
+    def context(context, value):
+        """Returns dictionary version of context."""
+        LOG.debug(_("Received %s"), context)
+        return context.to_dict()
+
+    @staticmethod
+    def echo_three_times(context, value):
+        context.reply(value)
+        context.reply(value + 1)
+        context.reply(value + 2)
+
+    @staticmethod
+    def echo_three_times_yield(context, value):
+        yield value
+        yield value + 1
+        yield value + 2
+
+    @staticmethod
+    def fail(context, value):
+        """Raises an exception with the value sent in."""
+        raise Exception(value)
diff --git a/nova/tests/test_service.py b/nova/tests/test_service.py
index f45f76b73..bbf47b50f 100644
--- a/nova/tests/test_service.py
+++ b/nova/tests/test_service.py
@@ -109,103 +109,8 @@ class ServiceTestCase(test.TestCase):
         #             the looping calls are created in StartService.
         app = service.Service.create(host=host, binary=binary, topic=topic)
 
-        self.mox.StubOutWithMock(service.rpc.Connection, 'instance')
-        service.rpc.Connection.instance(new=mox.IgnoreArg())
-
-        self.mox.StubOutWithMock(rpc,
-                                 'TopicAdapterConsumer',
-                                 use_mock_anything=True)
-        self.mox.StubOutWithMock(rpc,
-                                 'FanoutAdapterConsumer',
-                                 use_mock_anything=True)
-
-        self.mox.StubOutWithMock(rpc,
-                                 'ConsumerSet',
-                                 use_mock_anything=True)
-
-        rpc.TopicAdapterConsumer(connection=mox.IgnoreArg(),
-                            topic=topic,
-                            proxy=mox.IsA(service.Service)).AndReturn(
-                                    rpc.TopicAdapterConsumer)
-
-        rpc.TopicAdapterConsumer(connection=mox.IgnoreArg(),
-                            topic='%s.%s' % (topic, host),
-                            proxy=mox.IsA(service.Service)).AndReturn(
-                                    rpc.TopicAdapterConsumer)
-
-        rpc.FanoutAdapterConsumer(connection=mox.IgnoreArg(),
-                            topic=topic,
-                            proxy=mox.IsA(service.Service)).AndReturn(
-                                    rpc.FanoutAdapterConsumer)
-
-        def wait_func(self, limit=None):
-            return None
-
-        mock_cset = self.mox.CreateMock(rpc.ConsumerSet,
-                {'wait': wait_func})
-        rpc.ConsumerSet(connection=mox.IgnoreArg(),
-                        consumer_list=mox.IsA(list)).AndReturn(mock_cset)
-        wait_func(mox.IgnoreArg())
-
-        service_create = {'host': host,
-                          'binary': binary,
-                          'topic': topic,
-                          'report_count': 0,
-                          'availability_zone': 'nova'}
-        service_ref = {'host': host,
-                       'binary': binary,
-                       'report_count': 0,
-                       'id': 1}
-
-        service.db.service_get_by_args(mox.IgnoreArg(),
-                                       host,
-                                       binary).AndRaise(exception.NotFound())
-        service.db.service_create(mox.IgnoreArg(),
-                                  service_create).AndReturn(service_ref)
-        self.mox.ReplayAll()
-
-        app.start()
-        app.stop()
         self.assert_(app)
 
-    # We're testing sort of weird behavior in how report_state decides
-    # whether it is disconnected, it looks for a variable on itself called
-    # 'model_disconnected' and report_state doesn't really do much so this
-    # these are mostly just for coverage
-    def test_report_state_no_service(self):
-        host = 'foo'
-        binary = 'bar'
-        topic = 'test'
-        service_create = {'host': host,
-                          'binary': binary,
-                          'topic': topic,
-                          'report_count': 0,
-                          'availability_zone': 'nova'}
-        service_ref = {'host': host,
-                          'binary': binary,
-                          'topic': topic,
-                          'report_count': 0,
-                          'availability_zone': 'nova',
-                          'id': 1}
-
-        service.db.service_get_by_args(mox.IgnoreArg(),
-                                      host,
-                                      binary).AndRaise(exception.NotFound())
-        service.db.service_create(mox.IgnoreArg(),
-                                  service_create).AndReturn(service_ref)
-        service.db.service_get(mox.IgnoreArg(),
-                               service_ref['id']).AndReturn(service_ref)
-        service.db.service_update(mox.IgnoreArg(), service_ref['id'],
-                                  mox.ContainsKeyValue('report_count', 1))
-
-        self.mox.ReplayAll()
-        serv = service.Service(host,
-                               binary,
-                               topic,
-                               'nova.tests.test_service.FakeManager')
-        serv.start()
-        serv.report_state()
-
     def test_report_state_newly_disconnected(self):
         host = 'foo'
         binary = 'bar'
@@ -276,81 +181,6 @@ class ServiceTestCase(test.TestCase):
 
         self.assert_(not serv.model_disconnected)
 
-    def test_compute_can_update_available_resource(self):
-        """Confirm compute updates their record of compute-service table."""
-        host = 'foo'
-        binary = 'nova-compute'
-        topic = 'compute'
-
-        # Any mocks are not working without UnsetStubs() here.
-        self.mox.UnsetStubs()
-        ctxt = context.get_admin_context()
-        service_ref = db.service_create(ctxt, {'host': host,
-                                               'binary': binary,
-                                               'topic': topic})
-        serv = service.Service(host,
-                               binary,
-                               topic,
-                               'nova.compute.manager.ComputeManager')
-
-        # This testcase want to test calling update_available_resource.
-        # No need to call periodic call, then below variable must be set 0.
-        serv.report_interval = 0
-        serv.periodic_interval = 0
-
-        # Creating mocks
-        self.mox.StubOutWithMock(service.rpc.Connection, 'instance')
-        service.rpc.Connection.instance(new=mox.IgnoreArg())
-
-        self.mox.StubOutWithMock(rpc,
-                                 'TopicAdapterConsumer',
-                                 use_mock_anything=True)
-        self.mox.StubOutWithMock(rpc,
-                                 'FanoutAdapterConsumer',
-                                 use_mock_anything=True)
-
-        self.mox.StubOutWithMock(rpc,
-                                 'ConsumerSet',
-                                 use_mock_anything=True)
-
-        rpc.TopicAdapterConsumer(connection=mox.IgnoreArg(),
-                            topic=topic,
-                            proxy=mox.IsA(service.Service)).AndReturn(
-                                    rpc.TopicAdapterConsumer)
-
-        rpc.TopicAdapterConsumer(connection=mox.IgnoreArg(),
-                            topic='%s.%s' % (topic, host),
-                            proxy=mox.IsA(service.Service)).AndReturn(
-                                    rpc.TopicAdapterConsumer)
-
-        rpc.FanoutAdapterConsumer(connection=mox.IgnoreArg(),
-                            topic=topic,
-                            proxy=mox.IsA(service.Service)).AndReturn(
-                                    rpc.FanoutAdapterConsumer)
-
-        def wait_func(self, limit=None):
-            return None
-
-        mock_cset = self.mox.CreateMock(rpc.ConsumerSet,
-                {'wait': wait_func})
-        rpc.ConsumerSet(connection=mox.IgnoreArg(),
-                        consumer_list=mox.IsA(list)).AndReturn(mock_cset)
-        wait_func(mox.IgnoreArg())
-
-        self.mox.StubOutWithMock(serv.manager.driver,
-                                 'update_available_resource')
-        serv.manager.driver.update_available_resource(mox.IgnoreArg(), host)
-
-        # Just doing start()-stop(), not confirm new db record is created,
-        # because update_available_resource() works only in
-        # libvirt environment. This testcase confirms
-        # update_available_resource() is called. Otherwise, mox complains.
-        self.mox.ReplayAll()
-        serv.start()
-        serv.stop()
-
-        db.service_destroy(ctxt, service_ref['id'])
-
 
 class TestWSGIService(test.TestCase):
 
diff --git a/nova/tests/test_test.py b/nova/tests/test_test.py
index 35c838065..64f11fa45 100644
--- a/nova/tests/test_test.py
+++ b/nova/tests/test_test.py
@@ -33,8 +33,13 @@ class IsolationTestCase(test.TestCase):
         self.start_service('compute')
 
     def test_rpc_consumer_isolation(self):
-        connection = rpc.Connection.instance(new=True)
-        consumer = rpc.TopicAdapterConsumer(connection, topic='compute')
-        consumer.register_callback(
-                lambda x, y: self.fail('I should never be called'))
+        class NeverCalled(object):
+
+            def __getattribute__(*args):
+                assert False, "I should never get called."
+
+        connection = rpc.create_connection(new=True)
+        proxy = NeverCalled()
+        consumer = rpc.create_consumer(connection, 'compute',
+                                       proxy, fanout=False)
         consumer.attach_to_eventlet()
diff --git a/nova/tests/test_xenapi.py b/nova/tests/test_xenapi.py
index 1a8ddadc4..c2b97a5e3 100644
--- a/nova/tests/test_xenapi.py
+++ b/nova/tests/test_xenapi.py
@@ -881,6 +881,22 @@ class XenAPIMigrateInstance(test.TestCase):
                               network_info, resize_instance=False)
 
 
+class XenAPIImageTypeTestCase(test.TestCase):
+    """Test ImageType class."""
+
+    def test_to_string(self):
+        """Can convert from type id to type string."""
+        self.assertEquals(
+            vm_utils.ImageType.to_string(vm_utils.ImageType.KERNEL),
+            vm_utils.ImageType.KERNEL_STR)
+
+    def test_from_string(self):
+        """Can convert from string to type id."""
+        self.assertEquals(
+            vm_utils.ImageType.from_string(vm_utils.ImageType.KERNEL_STR),
+            vm_utils.ImageType.KERNEL)
+
+
 class XenAPIDetermineDiskImageTestCase(test.TestCase):
     """Unit tests for code that detects the ImageType."""
     def setUp(self):
diff --git a/nova/virt/xenapi/vm_utils.py b/nova/virt/xenapi/vm_utils.py
index 14465d8b4..088160075 100644
--- a/nova/virt/xenapi/vm_utils.py
+++ b/nova/virt/xenapi/vm_utils.py
@@ -85,38 +85,22 @@ class ImageType:
     DISK = 2
     DISK_RAW = 3
     DISK_VHD = 4
+    _ids = (KERNEL, RAMDISK, DISK, DISK_RAW, DISK_VHD)
 
     KERNEL_STR = "kernel"
     RAMDISK_STR = "ramdisk"
     DISK_STR = "os"
     DISK_RAW_STR = "os_raw"
     DISK_VHD_STR = "vhd"
+    _strs = (KERNEL_STR, RAMDISK_STR, DISK_STR, DISK_RAW_STR, DISK_VHD_STR)
 
     @classmethod
     def to_string(cls, image_type):
-        if image_type == ImageType.KERNEL:
-            return ImageType.KERNEL_STR
-        elif image_type == ImageType.RAMDISK:
-            return ImageType.RAMDISK_STR
-        elif image_type == ImageType.DISK:
-            return ImageType.DISK_STR
-        elif image_type == ImageType.DISK_RAW:
-            return ImageType.DISK_RAW_STR
-        elif image_type == ImageType.DISK_VHD:
-            return ImageType.VHD_STR
+        return dict(zip(ImageType._ids, ImageType._strs)).get(image_type)
 
     @classmethod
     def from_string(cls, image_type_str):
-        if image_type_str == ImageType.KERNEL_STR:
-            return ImageType.KERNEL
-        elif image_type == ImageType.RAMDISK_STR:
-            return ImageType.RAMDISK
-        elif image_type == ImageType.DISK_STR:
-            return ImageType.DISK
-        elif image_type == ImageType.DISK_RAW_STR:
-            return ImageType.DISK_RAW
-        elif image_type == ImageType.DISK_VHD_STR:
-            return ImageType.VHD
+        return dict(zip(ImageType._strs, ImageType._ids)).get(image_type_str)
 
 
 class VMHelper(HelperBase):