path: root/nova/test.py

# 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)