Merged Termie's branch and fixed rpc test cases for tesited. Nothing is testing the Eventlet version of rpc.call though yet.

7 files changed, 98 insertions, 35 deletions

diff --git a/nova/rpc.py b/nova/rpc.py
index 6363335ea..fe52ad35f 100644
--- a/nova/rpc.py
+++ b/nova/rpc.py
@@ -84,19 +84,6 @@ class Consumer(messaging.Consumer):
         self.failed_connection = False
         super(Consumer, self).__init__(*args, **kwargs)
 
-    # TODO(termie): it would be nice to give these some way of automatically
-    #               cleaning up after themselves
-    def attach_to_tornado(self, io_inst=None):
-        """Attach a callback to tornado that fires 10 times a second"""
-        from tornado import ioloop
-        if io_inst is None:
-            io_inst = ioloop.IOLoop.instance()
-
-        injected = ioloop.PeriodicCallback(
-            lambda: self.fetch(enable_callbacks=True), 100, io_loop=io_inst)
-        injected.start()
-        return injected
-
     def fetch(self, no_ack=None, auto_ack=None, enable_callbacks=False):
         """Wraps the parent fetch with some logic for failed connections"""
         # TODO(vish): the logic for failed connections and logging should be
@@ -124,6 +111,7 @@ class Consumer(messaging.Consumer):
         """Attach a callback to twisted that fires 10 times a second"""
         loop = task.LoopingCall(self.fetch, enable_callbacks=True)
         loop.start(interval=0.1)
+        return loop
 
 
 class Publisher(messaging.Publisher):
@@ -294,6 +282,37 @@ def call(topic, msg):
     return wait_msg.result
 
 
+def call_twisted(topic, msg):
+    """Sends a message on a topic and wait for a response"""
+    LOG.debug("Making asynchronous call...")
+    msg_id = uuid.uuid4().hex
+    msg.update({'_msg_id': msg_id})
+    LOG.debug("MSG_ID is %s" % (msg_id))
+
+    conn = Connection.instance()
+    d = defer.Deferred()
+    consumer = DirectConsumer(connection=conn, msg_id=msg_id)
+
+    def deferred_receive(data, message):
+        """Acks message and callbacks or errbacks"""
+        message.ack()
+        if data['failure']:
+            return d.errback(RemoteError(*data['failure']))
+        else:
+            return d.callback(data['result'])
+
+    consumer.register_callback(deferred_receive)
+    injected = consumer.attach_to_twisted()
+
+    # clean up after the injected listened and return x
+    d.addCallback(lambda x: injected.stop() and x or x)
+
+    publisher = TopicPublisher(connection=conn, topic=topic)
+    publisher.send(msg)
+    publisher.close()
+    return d
+
+
 def cast(topic, msg):
     """Sends a message on a topic without waiting for a response"""
     LOG.debug("Making asynchronous cast...")
diff --git a/nova/test.py b/nova/test.py
index c392c8a84..1f4b33272 100644
--- a/nova/test.py
+++ b/nova/test.py
@@ -33,6 +33,7 @@ from twisted.trial import unittest
 
 from nova import fakerabbit
 from nova import flags
+from nova import rpc
 
 
 FLAGS = flags.FLAGS
@@ -62,19 +63,29 @@ class TrialTestCase(unittest.TestCase):
         self.mox = mox.Mox()
         self.stubs = stubout.StubOutForTesting()
         self.flag_overrides = {}
+        self.injected = []
+        self._monkeyPatchAttach()
 
     def tearDown(self): # pylint: disable-msg=C0103
         """Runs after each test method to finalize/tear down test environment"""
-        super(TrialTestCase, self).tearDown()
         self.reset_flags()
         self.mox.UnsetStubs()
         self.stubs.UnsetAll()
         self.stubs.SmartUnsetAll()
         self.mox.VerifyAll()
+        
+        rpc.Consumer.attach_to_twisted = self.originalAttach
+        for x in self.injected:
+            try:
+                x.stop()
+            except AssertionError:
+                pass
 
         if FLAGS.fake_rabbit:
             fakerabbit.reset_all()
 
+        super(TrialTestCase, self).tearDown()
+
     def flags(self, **kw):
         """Override flag variables for a test"""
         for k, v in kw.iteritems():
@@ -90,16 +101,51 @@ class TrialTestCase(unittest.TestCase):
         for k, v in self.flag_overrides.iteritems():
             setattr(FLAGS, k, v)
 
+    def run(self, result=None):
+        test_method = getattr(self, self._testMethodName)
+        setattr(self,
+                self._testMethodName,
+                self._maybeInlineCallbacks(test_method, result))
+        rv = super(TrialTestCase, self).run(result)
+        setattr(self, self._testMethodName, test_method)
+        return rv
+
+    def _maybeInlineCallbacks(self, func, result):
+        def _wrapped():
+            g = func()
+            if isinstance(g, defer.Deferred):
+                return g
+            if not hasattr(g, 'send'):
+                return defer.succeed(g)
+
+            inlined = defer.inlineCallbacks(func)
+            d = inlined()
+            return d
+        _wrapped.func_name = func.func_name
+        return _wrapped
+
+    def _monkeyPatchAttach(self):
+        self.originalAttach = rpc.Consumer.attach_to_twisted
+        def _wrapped(innerSelf):
+            rv = self.originalAttach(innerSelf)
+            self.injected.append(rv)
+            return rv
+
+        _wrapped.func_name = self.originalAttach.func_name
+        rpc.Consumer.attach_to_twisted = _wrapped
+
 
 class BaseTestCase(TrialTestCase):
     # TODO(jaypipes): Can this be moved into the TrialTestCase class?
-    """Base test case class for all unit tests."""
+    """Base test case class for all unit tests.
+    
+    DEPRECATED: This is being removed once Tornado is gone, use TrialTestCase.
+    """
     def setUp(self): # pylint: disable-msg=C0103
         """Run before each test method to initialize test environment"""
         super(BaseTestCase, self).setUp()
         # TODO(termie): we could possibly keep a more global registry of
         #               the injected listeners... this is fine for now though
-        self.injected = []
         self.ioloop = ioloop.IOLoop.instance()
 
         self._waiting = None
@@ -109,8 +155,6 @@ class BaseTestCase(TrialTestCase):
     def tearDown(self):# pylint: disable-msg=C0103
         """Runs after each test method to finalize/tear down test environment"""
         super(BaseTestCase, self).tearDown()
-        for x in self.injected:
-            x.stop()
         if FLAGS.fake_rabbit:
             fakerabbit.reset_all()
 
diff --git a/nova/tests/access_unittest.py b/nova/tests/access_unittest.py
index c8a49d2ca..4b40ffd0a 100644
--- a/nova/tests/access_unittest.py
+++ b/nova/tests/access_unittest.py
@@ -31,7 +31,7 @@ FLAGS = flags.FLAGS
 class Context(object):
     pass
 
-class AccessTestCase(test.BaseTestCase):
+class AccessTestCase(test.TrialTestCase):
     def setUp(self):
         super(AccessTestCase, self).setUp()
         um = manager.AuthManager()
diff --git a/nova/tests/auth_unittest.py b/nova/tests/auth_unittest.py
index 3235dea39..f8be00613 100644
--- a/nova/tests/auth_unittest.py
+++ b/nova/tests/auth_unittest.py
@@ -29,7 +29,8 @@ from nova.api.ec2 import cloud
 FLAGS = flags.FLAGS
 
 
-class AuthTestCase(test.BaseTestCase):
+class AuthTestCase(test.TrialTestCase):
+    flush_db = False
     def setUp(self):
         super(AuthTestCase, self).setUp()
         self.flags(connection_type='fake')
diff --git a/nova/tests/cloud_unittest.py b/nova/tests/cloud_unittest.py
index 756ce519e..e8ff42fc5 100644
--- a/nova/tests/cloud_unittest.py
+++ b/nova/tests/cloud_unittest.py
@@ -41,7 +41,7 @@ from nova.api.ec2 import cloud
 FLAGS = flags.FLAGS
 
 
-class CloudTestCase(test.BaseTestCase):
+class CloudTestCase(test.TrialTestCase):
     def setUp(self):
         super(CloudTestCase, self).setUp()
         self.flags(connection_type='fake')
@@ -55,9 +55,9 @@ class CloudTestCase(test.BaseTestCase):
         # set up a service
         self.compute = utils.import_class(FLAGS.compute_manager)
         self.compute_consumer = rpc.AdapterConsumer(connection=self.conn,
-                                                     topic=FLAGS.compute_topic,
-                                                     proxy=self.compute)
-        self.injected.append(self.compute_consumer.attach_to_tornado(self.ioloop))
+                                                    topic=FLAGS.compute_topic,
+                                                    proxy=self.compute)
+        self.compute_consumer.attach_to_twisted()
 
         self.manager = manager.AuthManager()
         self.user = self.manager.create_user('admin', 'admin', 'admin', True)
@@ -68,7 +68,7 @@ class CloudTestCase(test.BaseTestCase):
     def tearDown(self):
         self.manager.delete_project(self.project)
         self.manager.delete_user(self.user)
-        super(CloudTestCase, self).setUp()
+        super(CloudTestCase, self).tearDown()
 
     def _create_key(self, name):
         # NOTE(vish): create depends on pool, so just call helper directly
diff --git a/nova/tests/objectstore_unittest.py b/nova/tests/objectstore_unittest.py
index dece4b5d5..b5970d405 100644
--- a/nova/tests/objectstore_unittest.py
+++ b/nova/tests/objectstore_unittest.py
@@ -53,7 +53,7 @@ os.makedirs(os.path.join(OSS_TEMPDIR, 'images'))
 os.makedirs(os.path.join(OSS_TEMPDIR, 'buckets'))
 
 
-class ObjectStoreTestCase(test.BaseTestCase):
+class ObjectStoreTestCase(test.TrialTestCase):
     """Test objectstore API directly."""
 
     def setUp(self): # pylint: disable-msg=C0103
diff --git a/nova/tests/rpc_unittest.py b/nova/tests/rpc_unittest.py
index e12a28fbc..9652841f2 100644
--- a/nova/tests/rpc_unittest.py
+++ b/nova/tests/rpc_unittest.py
@@ -30,7 +30,7 @@ from nova import test
 FLAGS = flags.FLAGS
 
 
-class RpcTestCase(test.BaseTestCase):
+class RpcTestCase(test.TrialTestCase):
     """Test cases for rpc"""
     def setUp(self):  # pylint: disable-msg=C0103
         super(RpcTestCase, self).setUp()
@@ -39,14 +39,13 @@ class RpcTestCase(test.BaseTestCase):
         self.consumer = rpc.AdapterConsumer(connection=self.conn,
                                             topic='test',
                                             proxy=self.receiver)
-
-        self.injected.append(self.consumer.attach_to_tornado(self.ioloop))
+        self.consumer.attach_to_twisted()
 
     def test_call_succeed(self):
         """Get a value through rpc call"""
         value = 42
-        result = yield rpc.call('test', {"method": "echo",
-                                         "args": {"value": value}})
+        result = yield rpc.call_twisted('test', {"method": "echo",
+                                                 "args": {"value": value}})
         self.assertEqual(value, result)
 
     def test_call_exception(self):
@@ -57,12 +56,12 @@ class RpcTestCase(test.BaseTestCase):
         to an int in the test.
         """
         value = 42
-        self.assertFailure(rpc.call('test', {"method": "fail",
-                                             "args": {"value": value}}),
+        self.assertFailure(rpc.call_twisted('test', {"method": "fail",
+                                                     "args": {"value": value}}),
                            rpc.RemoteError)
         try:
-            yield rpc.call('test', {"method": "fail",
-                                    "args": {"value": value}})
+            yield rpc.call_twisted('test', {"method": "fail",
+                                            "args": {"value": value}})
             self.fail("should have thrown rpc.RemoteError")
         except rpc.RemoteError as exc:
             self.assertEqual(int(exc.value), value)