~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to testsweet.py

merge merge tweaks from aaron, which includes latest .dev

Show diffs side-by-side

added added

removed removed

Lines of Context:
30
30
* utilities to run external commands and check their return code
31
31
  and/or output
32
32
 
33
 
Test cases should normally subclass TestBase.  The test runner should
 
33
Test cases should normally subclass testsweet.TestCase.  The test runner should
34
34
call runsuite().
35
35
 
36
36
This is meant to become independent of bzr, though that's not quite
37
37
true yet.
38
38
"""  
39
39
 
40
 
 
41
 
from unittest import TestResult, TestCase
42
 
 
 
40
import unittest
 
41
import sys
 
42
from bzrlib.selftest import TestUtil
 
43
 
 
44
# XXX: Don't need this anymore now we depend on python2.4
43
45
def _need_subprocess():
44
46
    sys.stderr.write("sorry, this test suite requires the subprocess module\n"
45
47
                     "this is shipped with python2.4 and available separately for 2.3\n")
49
51
    pass
50
52
 
51
53
 
52
 
 
53
54
class TestSkipped(Exception):
54
55
    """Indicates that a test was intentionally skipped, rather than failing."""
55
56
    # XXX: Not used yet
56
57
 
57
 
 
58
 
class TestBase(TestCase):
59
 
    """Base class for bzr test cases.
60
 
 
61
 
    Just defines some useful helper functions; doesn't actually test
62
 
    anything.
63
 
    """
64
 
    
65
 
    # TODO: Special methods to invoke bzr, so that we can run it
66
 
    # through a specified Python intepreter
67
 
 
68
 
    OVERRIDE_PYTHON = None # to run with alternative python 'python'
69
 
    BZRPATH = 'bzr'
70
 
 
71
 
    _log_buf = ""
72
 
 
73
 
 
74
 
    def setUp(self):
75
 
        super(TestBase, self).setUp()
76
 
        self.log("%s setup" % self.id())
77
 
 
78
 
 
79
 
    def tearDown(self):
80
 
        super(TestBase, self).tearDown()
81
 
        self.log("%s teardown" % self.id())
82
 
        self.log('')
83
 
 
84
 
 
85
 
    def formcmd(self, cmd):
86
 
        if isinstance(cmd, basestring):
87
 
            cmd = cmd.split()
88
 
 
89
 
        if cmd[0] == 'bzr':
90
 
            cmd[0] = self.BZRPATH
91
 
            if self.OVERRIDE_PYTHON:
92
 
                cmd.insert(0, self.OVERRIDE_PYTHON)
93
 
 
94
 
        self.log('$ %r' % cmd)
95
 
 
96
 
        return cmd
97
 
 
98
 
 
99
 
    def runcmd(self, cmd, retcode=0):
100
 
        """Run one command and check the return code.
101
 
 
102
 
        Returns a tuple of (stdout,stderr) strings.
103
 
 
104
 
        If a single string is based, it is split into words.
105
 
        For commands that are not simple space-separated words, please
106
 
        pass a list instead."""
107
 
        try:
108
 
            import shutil
109
 
            from subprocess import call
110
 
        except ImportError, e:
111
 
            _need_subprocess()
112
 
            raise
113
 
 
114
 
 
115
 
        cmd = self.formcmd(cmd)
116
 
 
117
 
        self.log('$ ' + ' '.join(cmd))
118
 
        actual_retcode = call(cmd, stdout=self.TEST_LOG, stderr=self.TEST_LOG)
119
 
 
120
 
        if retcode != actual_retcode:
121
 
            raise CommandFailed("test failed: %r returned %d, expected %d"
122
 
                                % (cmd, actual_retcode, retcode))
123
 
 
124
 
 
125
 
    def backtick(self, cmd, retcode=0):
126
 
        """Run a command and return its output"""
127
 
        try:
128
 
            import shutil
129
 
            from subprocess import Popen, PIPE
130
 
        except ImportError, e:
131
 
            _need_subprocess()
132
 
            raise
133
 
 
134
 
        cmd = self.formcmd(cmd)
135
 
        child = Popen(cmd, stdout=PIPE, stderr=self.TEST_LOG)
136
 
        outd, errd = child.communicate()
137
 
        self.log(outd)
138
 
        actual_retcode = child.wait()
139
 
 
140
 
        outd = outd.replace('\r', '')
141
 
 
142
 
        if retcode != actual_retcode:
143
 
            raise CommandFailed("test failed: %r returned %d, expected %d"
144
 
                                % (cmd, actual_retcode, retcode))
145
 
 
146
 
        return outd
147
 
 
148
 
 
149
 
 
150
 
    def build_tree(self, shape):
151
 
        """Build a test tree according to a pattern.
152
 
 
153
 
        shape is a sequence of file specifications.  If the final
154
 
        character is '/', a directory is created.
155
 
 
156
 
        This doesn't add anything to a branch.
157
 
        """
158
 
        # XXX: It's OK to just create them using forward slashes on windows?
159
 
        import os
160
 
        for name in shape:
161
 
            assert isinstance(name, basestring)
162
 
            if name[-1] == '/':
163
 
                os.mkdir(name[:-1])
164
 
            else:
165
 
                f = file(name, 'wt')
166
 
                print >>f, "contents of", name
167
 
                f.close()
168
 
 
169
 
 
170
 
    def log(self, msg):
171
 
        """Log a message to a progress file"""
172
 
        self._log_buf = self._log_buf + str(msg) + '\n'
173
 
        print >>self.TEST_LOG, msg
174
 
 
175
 
 
176
 
    def check_inventory_shape(self, inv, shape):
177
 
        """
178
 
        Compare an inventory to a list of expected names.
179
 
 
180
 
        Fail if they are not precisely equal.
181
 
        """
182
 
        extras = []
183
 
        shape = list(shape)             # copy
184
 
        for path, ie in inv.entries():
185
 
            name = path.replace('\\', '/')
186
 
            if ie.kind == 'dir':
187
 
                name = name + '/'
188
 
            if name in shape:
189
 
                shape.remove(name)
190
 
            else:
191
 
                extras.append(name)
192
 
        if shape:
193
 
            self.fail("expected paths not found in inventory: %r" % shape)
194
 
        if extras:
195
 
            self.fail("unexpected paths found in inventory: %r" % extras)
196
 
 
197
 
 
198
 
    def check_file_contents(self, filename, expect):
199
 
        self.log("check contents of file %s" % filename)
200
 
        contents = file(filename, 'r').read()
201
 
        if contents != expect:
202
 
            self.log("expected: %r" % expect)
203
 
            self.log("actually: %r" % contents)
204
 
            self.fail("contents of %s not as expected")
205
 
            
206
 
 
207
 
 
208
 
class InTempDir(TestBase):
209
 
    """Base class for tests run in a temporary branch."""
210
 
    def setUp(self):
211
 
        import os
212
 
        self.test_dir = os.path.join(self.TEST_ROOT, self.__class__.__name__)
213
 
        os.mkdir(self.test_dir)
214
 
        os.chdir(self.test_dir)
215
 
        
216
 
    def tearDown(self):
217
 
        import os
218
 
        os.chdir(self.TEST_ROOT)
219
 
 
220
 
 
221
 
 
222
 
 
223
 
 
224
 
class _MyResult(TestResult):
 
58
class EarlyStoppingTestResultAdapter(object):
 
59
    """An adapter for TestResult to stop at the first first failure or error"""
 
60
 
 
61
    def __init__(self, result):
 
62
        self._result = result
 
63
 
 
64
    def addError(self, test, err):
 
65
        self._result.addError(test, err)
 
66
        self._result.stop()
 
67
 
 
68
    def addFailure(self, test, err):
 
69
        self._result.addFailure(test, err)
 
70
        self._result.stop()
 
71
 
 
72
    def __getattr__(self, name):
 
73
        return getattr(self._result, name)
 
74
 
 
75
    def __setattr__(self, name, value):
 
76
        if name == '_result':
 
77
            object.__setattr__(self, name, value)
 
78
        return setattr(self._result, name, value)
 
79
 
 
80
 
 
81
class _MyResult(unittest._TextTestResult):
225
82
    """
226
83
    Custom TestResult.
227
84
 
228
85
    No special behaviour for now.
229
86
    """
230
 
    def __init__(self, out):
231
 
        self.out = out
232
 
        TestResult.__init__(self)
233
87
 
234
88
    def startTest(self, test):
 
89
        unittest.TestResult.startTest(self, test)
235
90
        # TODO: Maybe show test.shortDescription somewhere?
236
 
        what = test.id()
237
 
        # python2.3 has the bad habit of just "runit" for doctests
238
 
        if what == 'runit':
239
 
            what = test.shortDescription()
240
 
        
241
 
        print >>self.out, '%-60.60s' % what,
242
 
        self.out.flush()
243
 
        TestResult.startTest(self, test)
244
 
 
245
 
    def stopTest(self, test):
246
 
        # print
247
 
        TestResult.stopTest(self, test)
248
 
 
 
91
        what = test.shortDescription() or test.id()        
 
92
        if self.showAll:
 
93
            self.stream.write('%-70.70s' % what)
 
94
        self.stream.flush()
249
95
 
250
96
    def addError(self, test, err):
251
 
        print >>self.out, 'ERROR'
252
 
        TestResult.addError(self, test, err)
253
 
        _show_test_failure('error', test, err, self.out)
 
97
        super(_MyResult, self).addError(test, err)
 
98
        self.stream.flush()
254
99
 
255
100
    def addFailure(self, test, err):
256
 
        print >>self.out, 'FAILURE'
257
 
        TestResult.addFailure(self, test, err)
258
 
        _show_test_failure('failure', test, err, self.out)
 
101
        super(_MyResult, self).addFailure(test, err)
 
102
        self.stream.flush()
259
103
 
260
104
    def addSuccess(self, test):
261
 
        print >>self.out, 'OK'
262
 
        TestResult.addSuccess(self, test)
263
 
 
264
 
 
265
 
 
266
 
def run_suite(suite, name="test"):
267
 
    import os
 
105
        if self.showAll:
 
106
            self.stream.writeln('OK')
 
107
        elif self.dots:
 
108
            self.stream.write('~')
 
109
        self.stream.flush()
 
110
        unittest.TestResult.addSuccess(self, test)
 
111
 
 
112
    def printErrorList(self, flavour, errors):
 
113
        for test, err in errors:
 
114
            self.stream.writeln(self.separator1)
 
115
            self.stream.writeln("%s: %s" % (flavour,self.getDescription(test)))
 
116
            self.stream.writeln(self.separator2)
 
117
            self.stream.writeln("%s" % err)
 
118
            if hasattr(test, '_get_log'):
 
119
                self.stream.writeln()
 
120
                self.stream.writeln('log from this test:')
 
121
                print >>self.stream, test._get_log()
 
122
 
 
123
 
 
124
class TextTestRunner(unittest.TextTestRunner):
 
125
 
 
126
    def _makeResult(self):
 
127
        result = _MyResult(self.stream, self.descriptions, self.verbosity)
 
128
        return EarlyStoppingTestResultAdapter(result)
 
129
 
 
130
 
 
131
class filteringVisitor(TestUtil.TestVisitor):
 
132
    """I accruse all the testCases I visit that pass a regexp filter on id
 
133
    into my suite
 
134
    """
 
135
 
 
136
    def __init__(self, filter):
 
137
        import re
 
138
        TestUtil.TestVisitor.__init__(self)
 
139
        self._suite=None
 
140
        self.filter=re.compile(filter)
 
141
 
 
142
    def suite(self):
 
143
        """answer the suite we are building"""
 
144
        if self._suite is None:
 
145
            self._suite=TestUtil.TestSuite()
 
146
        return self._suite
 
147
 
 
148
    def visitCase(self, aCase):
 
149
        if self.filter.match(aCase.id()):
 
150
            self.suite().addTest(aCase)
 
151
 
 
152
 
 
153
def run_suite(suite, name='test', verbose=False, pattern=".*"):
268
154
    import shutil
269
 
    import time
270
 
    import sys
271
 
    
272
 
    _setup_test_log(name)
273
 
    _setup_test_dir(name)
274
 
    print
275
 
 
276
 
    # save stdout & stderr so there's no leakage from code-under-test
277
 
    real_stdout = sys.stdout
278
 
    real_stderr = sys.stderr
279
 
    sys.stdout = sys.stderr = TestBase.TEST_LOG
280
 
    try:
281
 
        result = _MyResult(real_stdout)
282
 
        suite.run(result)
283
 
    finally:
284
 
        sys.stdout = real_stdout
285
 
        sys.stderr = real_stderr
286
 
 
287
 
    _show_results(result)
288
 
 
 
155
    from bzrlib.selftest import TestCaseInTempDir
 
156
    TestCaseInTempDir._TEST_NAME = name
 
157
    if verbose:
 
158
        verbosity = 2
 
159
    else:
 
160
        verbosity = 1
 
161
    runner = TextTestRunner(stream=sys.stdout,
 
162
                            descriptions=0,
 
163
                            verbosity=verbosity)
 
164
    visitor = filteringVisitor(pattern)
 
165
    suite.visit(visitor)
 
166
    result = runner.run(visitor.suite())
 
167
    # This is still a little bogus, 
 
168
    # but only a little. Folk not using our testrunner will
 
169
    # have to delete their temp directories themselves.
 
170
    if result.wasSuccessful():
 
171
        if TestCaseInTempDir.TEST_ROOT is not None:
 
172
            shutil.rmtree(TestCaseInTempDir.TEST_ROOT) 
 
173
    else:
 
174
        print "Failed tests working directories are in '%s'\n" % TestCaseInTempDir.TEST_ROOT
289
175
    return result.wasSuccessful()
290
 
 
291
 
 
292
 
 
293
 
def _setup_test_log(name):
294
 
    import time
295
 
    import os
296
 
    
297
 
    log_filename = os.path.abspath(name + '.log')
298
 
    TestBase.TEST_LOG = open(log_filename, 'wt', buffering=1) # line buffered
299
 
 
300
 
    print >>TestBase.TEST_LOG, "tests run at " + time.ctime()
301
 
    print '%-30s %s' % ('test log', log_filename)
302
 
 
303
 
 
304
 
def _setup_test_dir(name):
305
 
    import os
306
 
    import shutil
307
 
    
308
 
    TestBase.ORIG_DIR = os.getcwdu()
309
 
    TestBase.TEST_ROOT = os.path.abspath(name + '.tmp')
310
 
 
311
 
    print '%-30s %s' % ('running tests in', TestBase.TEST_ROOT)
312
 
 
313
 
    if os.path.exists(TestBase.TEST_ROOT):
314
 
        shutil.rmtree(TestBase.TEST_ROOT)
315
 
    os.mkdir(TestBase.TEST_ROOT)
316
 
    os.chdir(TestBase.TEST_ROOT)
317
 
 
318
 
    # make a fake bzr directory there to prevent any tests propagating
319
 
    # up onto the source directory's real branch
320
 
    os.mkdir(os.path.join(TestBase.TEST_ROOT, '.bzr'))
321
 
 
322
 
    
323
 
 
324
 
def _show_results(result):
325
 
     print
326
 
     print '%4d tests run' % result.testsRun
327
 
     print '%4d errors' % len(result.errors)
328
 
     print '%4d failures' % len(result.failures)
329
 
 
330
 
 
331
 
 
332
 
def _show_test_failure(kind, case, exc_info, out):
333
 
    from traceback import print_exception
334
 
    
335
 
    print >>out, '-' * 60
336
 
    print >>out, case
337
 
    
338
 
    desc = case.shortDescription()
339
 
    if desc:
340
 
        print >>out, '   (%s)' % desc
341
 
         
342
 
    print_exception(exc_info[0], exc_info[1], exc_info[2], None, out)
343
 
        
344
 
    if isinstance(case, TestBase):
345
 
        print >>out
346
 
        print >>out, 'log from this test:'
347
 
        print >>out, case._log_buf
348
 
         
349
 
    print >>out, '-' * 60
350
 
    
351