# vim: tabstop=4 shiftwidth=4 softtabstop=4 # Copyright [2010] [Anso Labs, LLC] # # 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. """ Base classes for our unit tests. Allows overriding of flags for use of fakes, and some black magic for inline callbacks. """ import logging import time import unittest from nova import vendor import mox from tornado import ioloop from twisted.internet import defer from twisted.python import failure from twisted.trial import unittest as trial_unittest import stubout from nova import datastore from nova import fakerabbit from nova import flags FLAGS = flags.FLAGS flags.DEFINE_bool('fake_tests', True, 'should we use everything for testing') def skip_if_fake(f): def _skipper(*args, **kw): if FLAGS.fake_tests: raise trial_unittest.SkipTest('Test cannot be run in fake mode') else: return f(*args, **kw) _skipper.func_name = f.func_name return _skipper class TrialTestCase(trial_unittest.TestCase): def setUp(self): super(TrialTestCase, self).setUp() # emulate some of the mox stuff, we can't use the metaclass # because it screws with our generators self.mox = mox.Mox() self.stubs = stubout.StubOutForTesting() self.flag_overrides = {} def tearDown(self): super(TrialTestCase, self).tearDown() self.reset_flags() self.mox.UnsetStubs() self.stubs.UnsetAll() self.stubs.SmartUnsetAll() self.mox.VerifyAll() if FLAGS.fake_rabbit: fakerabbit.reset_all() # attempt to wipe all keepers #keeper = datastore.Keeper() #keeper.clear_all() def flags(self, **kw): for k, v in kw.iteritems(): if k in self.flag_overrides: self.reset_flags() raise Exception( 'trying to override already overriden flag: %s' % k) self.flag_overrides[k] = getattr(FLAGS, k) setattr(FLAGS, k, v) def reset_flags(self): for k, v in self.flag_overrides.iteritems(): setattr(FLAGS, k, v) class BaseTestCase(TrialTestCase): def setUp(self): 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 self._doneWaiting = False self._timedOut = False self.set_up() def set_up(self): pass def tear_down(self): pass def tearDown(self): super(BaseTestCase, self).tearDown() for x in self.injected: x.stop() if FLAGS.fake_rabbit: fakerabbit.reset_all() self.tear_down() def _waitForTest(self, timeout=60): """ Push the ioloop along to wait for our test to complete. """ self._waiting = self.ioloop.add_timeout(time.time() + timeout, self._timeout) def _wait(): if self._timedOut: self.fail('test timed out') self._done() if self._doneWaiting: self.ioloop.stop() return # we can use add_callback here but this uses less cpu when testing self.ioloop.add_timeout(time.time() + 0.01, _wait) self.ioloop.add_callback(_wait) self.ioloop.start() def _done(self): if self._waiting: try: self.ioloop.remove_timeout(self._waiting) except Exception: pass self._waiting = None self._doneWaiting = True def _maybeInlineCallbacks(self, f): """ If we're doing async calls in our tests, wait on them. This is probably the most complicated hunk of code we have so far. First up, if the function is normal (not async) we just act normal and return. Async tests will use the "Inline Callbacks" pattern, which means you yield Deferreds at every "waiting" step of your code instead of making epic callback chains. Example (callback chain, ugly): d = self.node.terminate_instance(instance_id) # a Deferred instance def _describe(_): d_desc = self.node.describe_instances() # another Deferred instance return d_desc def _checkDescribe(rv): self.assertEqual(rv, []) d.addCallback(_describe) d.addCallback(_checkDescribe) d.addCallback(lambda x: self._done()) self._waitForTest() Example (inline callbacks! yay!): yield self.node.terminate_instance(instance_id) rv = yield self.node.describe_instances() self.assertEqual(rv, []) If the test fits the Inline Callbacks pattern we will automatically handle calling wait and done. """ # TODO(termie): this can be a wrapper function instead and # and we can make a metaclass so that we don't # have to copy all that "run" code below. g = f() if not hasattr(g, 'send'): self._done() return defer.succeed(g) inlined = defer.inlineCallbacks(f) d = inlined() return d def _catchExceptions(self, result, failure): exc = (failure.type, failure.value, failure.getTracebackObject()) if isinstance(failure.value, self.failureException): result.addFailure(self, exc) elif isinstance(failure.value, KeyboardInterrupt): raise else: result.addError(self, exc) self._done() def _timeout(self): self._waiting = False self._timedOut = True def run(self, result=None): if result is None: result = self.defaultTestResult() result.startTest(self) testMethod = getattr(self, self._testMethodName) try: try: self.setUp() except KeyboardInterrupt: raise except: result.addError(self, self._exc_info()) return ok = False try: d = self._maybeInlineCallbacks(testMethod) d.addErrback(lambda x: self._catchExceptions(result, x)) d.addBoth(lambda x: self._done() and x) self._waitForTest() ok = True except self.failureException: result.addFailure(self, self._exc_info()) except KeyboardInterrupt: raise except: result.addError(self, self._exc_info()) try: self.tearDown() except KeyboardInterrupt: raise except: result.addError(self, self._exc_info()) ok = False if ok: result.addSuccess(self) finally: result.stopTest(self)