← Back to branch summary

~hudson-openstack/nova/trunk

« back to all changes in this revision

Viewing changes to nova/test.py

  • Committer: Jesse Andrews
  • Date: 2010-05-28 06:05:26 UTC
  • Revision ID: git-v1:bf6e6e718cdc7488e2da87b21e258ccc065fe499
initial commit

Show diffs side-by-side

added added

removed removed

Lines of Context:
nova/test.py
 
1
# vim: tabstop=4 shiftwidth=4 softtabstop=4
 
2
# Copyright [2010] [Anso Labs, LLC]
 
3
# 
 
4
#    Licensed under the Apache License, Version 2.0 (the "License");
 
5
#    you may not use this file except in compliance with the License.
 
6
#    You may obtain a copy of the License at
 
7
# 
 
8
#        http://www.apache.org/licenses/LICENSE-2.0
 
9
# 
 
10
#    Unless required by applicable law or agreed to in writing, software
 
11
#    distributed under the License is distributed on an "AS IS" BASIS,
 
12
#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 
13
#    See the License for the specific language governing permissions and
 
14
#    limitations under the License.
 
15
 
 
16
"""
 
17
Base classes for our unit tests.
 
18
Allows overriding of flags for use of fakes, 
 
19
and some black magic for inline callbacks.
 
20
"""
 
21
 
 
22
import logging
 
23
import time
 
24
import unittest
 
25
 
 
26
from nova import vendor
 
27
import mox
 
28
from tornado import ioloop
 
29
from twisted.internet import defer
 
30
from twisted.python import failure
 
31
from twisted.trial import unittest as trial_unittest
 
32
import stubout
 
33
 
 
34
from nova import datastore
 
35
from nova import fakerabbit
 
36
from nova import flags
 
37
 
 
38
 
 
39
FLAGS = flags.FLAGS
 
40
flags.DEFINE_bool('fake_tests', True,
 
41
                  'should we use everything for testing')
 
42
 
 
43
 
 
44
def skip_if_fake(f):
 
45
    def _skipper(*args, **kw):
 
46
        if FLAGS.fake_tests:
 
47
            raise trial_unittest.SkipTest('Test cannot be run in fake mode')
 
48
        else:
 
49
            return f(*args, **kw)
 
50
    
 
51
    _skipper.func_name = f.func_name
 
52
    return _skipper
 
53
 
 
54
 
 
55
class TrialTestCase(trial_unittest.TestCase):
 
56
    def setUp(self):
 
57
        super(TrialTestCase, self).setUp()
 
58
 
 
59
        # emulate some of the mox stuff, we can't use the metaclass
 
60
        # because it screws with our generators
 
61
        self.mox = mox.Mox()
 
62
        self.stubs = stubout.StubOutForTesting()
 
63
        self.flag_overrides = {}
 
64
 
 
65
    def tearDown(self):
 
66
        super(TrialTestCase, self).tearDown()
 
67
        self.reset_flags()
 
68
        self.mox.UnsetStubs()
 
69
        self.stubs.UnsetAll()
 
70
        self.stubs.SmartUnsetAll()
 
71
        self.mox.VerifyAll()
 
72
 
 
73
        if FLAGS.fake_rabbit:
 
74
            fakerabbit.reset_all()
 
75
        
 
76
        # attempt to wipe all keepers
 
77
        #keeper = datastore.Keeper()
 
78
        #keeper.clear_all()
 
79
 
 
80
    def flags(self, **kw):
 
81
        for k, v in kw.iteritems():
 
82
            if k in self.flag_overrides:
 
83
                self.reset_flags()
 
84
                raise Exception(
 
85
                        'trying to override already overriden flag: %s' % k)
 
86
            self.flag_overrides[k] = getattr(FLAGS, k)
 
87
            setattr(FLAGS, k, v)
 
88
 
 
89
    def reset_flags(self):
 
90
        for k, v in self.flag_overrides.iteritems():
 
91
            setattr(FLAGS, k, v)
 
92
 
 
93
    
 
94
 
 
95
class BaseTestCase(TrialTestCase):
 
96
    def setUp(self):
 
97
        super(BaseTestCase, self).setUp()
 
98
        # TODO(termie): we could possibly keep a more global registry of
 
99
        #               the injected listeners... this is fine for now though
 
100
        self.injected = []
 
101
        self.ioloop = ioloop.IOLoop.instance()
 
102
  
 
103
        self._waiting = None
 
104
        self._doneWaiting = False
 
105
        self._timedOut = False
 
106
        self.set_up()
 
107
 
 
108
    def set_up(self):
 
109
        pass
 
110
 
 
111
    def tear_down(self):
 
112
        pass
 
113
 
 
114
    def tearDown(self):
 
115
        super(BaseTestCase, self).tearDown()
 
116
        for x in self.injected:
 
117
            x.stop()
 
118
        if FLAGS.fake_rabbit:
 
119
            fakerabbit.reset_all()
 
120
        self.tear_down()
 
121
 
 
122
    def _waitForTest(self, timeout=60):
 
123
        """ Push the ioloop along to wait for our test to complete. """
 
124
        self._waiting = self.ioloop.add_timeout(time.time() + timeout,
 
125
                                                self._timeout)
 
126
        def _wait():
 
127
            if self._timedOut:
 
128
                self.fail('test timed out')
 
129
                self._done()
 
130
            if self._doneWaiting:
 
131
                self.ioloop.stop()
 
132
                return
 
133
            # we can use add_callback here but this uses less cpu when testing
 
134
            self.ioloop.add_timeout(time.time() + 0.01, _wait)
 
135
 
 
136
        self.ioloop.add_callback(_wait)
 
137
        self.ioloop.start()
 
138
 
 
139
    def _done(self):
 
140
        if self._waiting:
 
141
            try:
 
142
                self.ioloop.remove_timeout(self._waiting)
 
143
            except Exception:
 
144
                pass
 
145
            self._waiting = None
 
146
        self._doneWaiting = True
 
147
    
 
148
    def _maybeInlineCallbacks(self, f):
 
149
        """ If we're doing async calls in our tests, wait on them.
 
150
        
 
151
        This is probably the most complicated hunk of code we have so far.
 
152
 
 
153
        First up, if the function is normal (not async) we just act normal
 
154
        and return.
 
155
 
 
156
        Async tests will use the "Inline Callbacks" pattern, which means
 
157
        you yield Deferreds at every "waiting" step of your code instead
 
158
        of making epic callback chains.
 
159
 
 
160
        Example (callback chain, ugly):
 
161
    
 
162
        d = self.node.terminate_instance(instance_id) # a Deferred instance
 
163
        def _describe(_):
 
164
            d_desc = self.node.describe_instances() # another Deferred instance
 
165
            return d_desc
 
166
        def _checkDescribe(rv):
 
167
            self.assertEqual(rv, [])
 
168
        d.addCallback(_describe)
 
169
        d.addCallback(_checkDescribe)
 
170
        d.addCallback(lambda x: self._done())
 
171
        self._waitForTest()
 
172
        
 
173
        Example (inline callbacks! yay!):
 
174
 
 
175
        yield self.node.terminate_instance(instance_id)
 
176
        rv = yield self.node.describe_instances()
 
177
        self.assertEqual(rv, [])
 
178
 
 
179
        If the test fits the Inline Callbacks pattern we will automatically
 
180
        handle calling wait and done.
 
181
        """
 
182
        # TODO(termie): this can be a wrapper function instead and
 
183
        #               and we can make a metaclass so that we don't
 
184
        #               have to copy all that "run" code below.
 
185
        g = f()
 
186
        if not hasattr(g, 'send'):
 
187
            self._done()
 
188
            return defer.succeed(g)
 
189
        
 
190
        inlined = defer.inlineCallbacks(f)
 
191
        d = inlined()
 
192
        return d
 
193
    
 
194
    def _catchExceptions(self, result, failure):
 
195
        exc = (failure.type, failure.value, failure.getTracebackObject())
 
196
        if isinstance(failure.value, self.failureException):
 
197
            result.addFailure(self, exc)
 
198
        elif isinstance(failure.value, KeyboardInterrupt):
 
199
            raise
 
200
        else:
 
201
            result.addError(self, exc)
 
202
 
 
203
        self._done()
 
204
 
 
205
    def _timeout(self):
 
206
        self._waiting = False
 
207
        self._timedOut = True
 
208
 
 
209
    def run(self, result=None):
 
210
        if result is None: result = self.defaultTestResult()
 
211
 
 
212
        result.startTest(self)
 
213
        testMethod = getattr(self, self._testMethodName)
 
214
        try:
 
215
            try:
 
216
                self.setUp()
 
217
            except KeyboardInterrupt:
 
218
                raise
 
219
            except:
 
220
                result.addError(self, self._exc_info())
 
221
                return
 
222
 
 
223
            ok = False
 
224
            try:
 
225
                d = self._maybeInlineCallbacks(testMethod)
 
226
                d.addErrback(lambda x: self._catchExceptions(result, x))
 
227
                d.addBoth(lambda x: self._done() and x)
 
228
                self._waitForTest()
 
229
                ok = True
 
230
            except self.failureException:
 
231
                result.addFailure(self, self._exc_info())
 
232
            except KeyboardInterrupt:
 
233
                raise
 
234
            except:
 
235
                result.addError(self, self._exc_info())
 
236
 
 
237
            try:
 
238
                self.tearDown()
 
239
            except KeyboardInterrupt:
 
240
                raise
 
241
            except:
 
242
                result.addError(self, self._exc_info())
 
243
                ok = False
 
244
            if ok: result.addSuccess(self)
 
245
        finally:
 
246
            result.stopTest(self)