~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/selftest/__init__.py

  • Committer: Robert Collins
  • Date: 2005-10-11 08:31:29 UTC
  • Revision ID: robertc@lifelesslap.robertcollins.net-20051011083129-fa720bc6cd6c039f
inline and simplify branch.find_branch_root, it should just try to create a branch at each step, which is simpler than probing for a specific dir and has less round trips.

Show diffs side-by-side

added added

removed removed

Lines of Context:
15
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
16
 
17
17
 
 
18
from cStringIO import StringIO
18
19
import logging
19
20
import unittest
20
21
import tempfile
21
22
import os
 
23
import errno
 
24
import re
 
25
import shutil
 
26
import subprocess
22
27
import sys
23
 
import subprocess
 
28
import time
24
29
 
25
 
from testsweet import run_suite
26
30
import bzrlib.commands
27
 
 
28
31
import bzrlib.trace
29
32
import bzrlib.fetch
 
33
from bzrlib.selftest import TestUtil
 
34
from bzrlib.selftest.TestUtil import TestLoader, TestSuite
30
35
 
31
36
 
32
37
MODULES_TO_TEST = []
34
39
 
35
40
from logging import debug, warning, error
36
41
 
 
42
 
 
43
 
 
44
class EarlyStoppingTestResultAdapter(object):
 
45
    """An adapter for TestResult to stop at the first first failure or error"""
 
46
 
 
47
    def __init__(self, result):
 
48
        self._result = result
 
49
 
 
50
    def addError(self, test, err):
 
51
        self._result.addError(test, err)
 
52
        self._result.stop()
 
53
 
 
54
    def addFailure(self, test, err):
 
55
        self._result.addFailure(test, err)
 
56
        self._result.stop()
 
57
 
 
58
    def __getattr__(self, name):
 
59
        return getattr(self._result, name)
 
60
 
 
61
    def __setattr__(self, name, value):
 
62
        if name == '_result':
 
63
            object.__setattr__(self, name, value)
 
64
        return setattr(self._result, name, value)
 
65
 
 
66
 
 
67
class _MyResult(unittest._TextTestResult):
 
68
    """
 
69
    Custom TestResult.
 
70
 
 
71
    No special behaviour for now.
 
72
    """
 
73
 
 
74
    def _elapsedTime(self):
 
75
        return "(Took %.3fs)" % (time.time() - self._start_time)
 
76
 
 
77
    def startTest(self, test):
 
78
        unittest.TestResult.startTest(self, test)
 
79
        # TODO: Maybe show test.shortDescription somewhere?
 
80
        what = test.shortDescription() or test.id()        
 
81
        if self.showAll:
 
82
            self.stream.write('%-70.70s' % what)
 
83
        self.stream.flush()
 
84
        self._start_time = time.time()
 
85
 
 
86
    def addError(self, test, err):
 
87
        unittest.TestResult.addError(self, test, err)
 
88
        if self.showAll:
 
89
            self.stream.writeln("ERROR %s" % self._elapsedTime())
 
90
        elif self.dots:
 
91
            self.stream.write('E')
 
92
        self.stream.flush()
 
93
 
 
94
    def addFailure(self, test, err):
 
95
        unittest.TestResult.addFailure(self, test, err)
 
96
        if self.showAll:
 
97
            self.stream.writeln("FAIL %s" % self._elapsedTime())
 
98
        elif self.dots:
 
99
            self.stream.write('F')
 
100
        self.stream.flush()
 
101
 
 
102
    def addSuccess(self, test):
 
103
        if self.showAll:
 
104
            self.stream.writeln('OK %s' % self._elapsedTime())
 
105
        elif self.dots:
 
106
            self.stream.write('~')
 
107
        self.stream.flush()
 
108
        unittest.TestResult.addSuccess(self, test)
 
109
 
 
110
    def printErrorList(self, flavour, errors):
 
111
        for test, err in errors:
 
112
            self.stream.writeln(self.separator1)
 
113
            self.stream.writeln("%s: %s" % (flavour,self.getDescription(test)))
 
114
            if hasattr(test, '_get_log'):
 
115
                self.stream.writeln()
 
116
                self.stream.writeln('log from this test:')
 
117
                print >>self.stream, test._get_log()
 
118
            self.stream.writeln(self.separator2)
 
119
            self.stream.writeln("%s" % err)
 
120
 
 
121
 
 
122
class TextTestRunner(unittest.TextTestRunner):
 
123
 
 
124
    def _makeResult(self):
 
125
        result = _MyResult(self.stream, self.descriptions, self.verbosity)
 
126
        return EarlyStoppingTestResultAdapter(result)
 
127
 
 
128
 
 
129
def iter_suite_tests(suite):
 
130
    """Return all tests in a suite, recursing through nested suites"""
 
131
    for item in suite._tests:
 
132
        if isinstance(item, unittest.TestCase):
 
133
            yield item
 
134
        elif isinstance(item, unittest.TestSuite):
 
135
            for r in iter_suite_tests(item):
 
136
                yield r
 
137
        else:
 
138
            raise Exception('unknown object %r inside test suite %r'
 
139
                            % (item, suite))
 
140
 
 
141
 
 
142
class TestSkipped(Exception):
 
143
    """Indicates that a test was intentionally skipped, rather than failing."""
 
144
    # XXX: Not used yet
 
145
 
 
146
 
37
147
class CommandFailed(Exception):
38
148
    pass
39
149
 
53
163
    BZRPATH = 'bzr'
54
164
 
55
165
    def setUp(self):
56
 
        # this replaces the default testsweet.TestCase; we don't want logging changed
57
166
        unittest.TestCase.setUp(self)
58
167
        bzrlib.trace.disable_default_logging()
59
168
        self._enable_file_logging()
74
183
        
75
184
        self._log_file_name = name
76
185
 
77
 
        
78
186
    def tearDown(self):
79
187
        logging.getLogger('').removeHandler(self._log_hdlr)
80
188
        bzrlib.trace.enable_default_logging()
82
190
        self._log_file.close()
83
191
        unittest.TestCase.tearDown(self)
84
192
 
85
 
 
86
193
    def log(self, *args):
87
194
        logging.debug(*args)
88
195
 
90
197
        """Return as a string the log for this test"""
91
198
        return open(self._log_file_name).read()
92
199
 
 
200
 
 
201
    def capture(self, cmd):
 
202
        """Shortcut that splits cmd into words, runs, and returns stdout"""
 
203
        return self.run_bzr_captured(cmd.split())[0]
 
204
 
 
205
    def run_bzr_captured(self, argv, retcode=0):
 
206
        """Invoke bzr and return (result, stdout, stderr).
 
207
 
 
208
        Useful for code that wants to check the contents of the
 
209
        output, the way error messages are presented, etc.
 
210
 
 
211
        This should be the main method for tests that want to exercise the
 
212
        overall behavior of the bzr application (rather than a unit test
 
213
        or a functional test of the library.)
 
214
 
 
215
        Much of the old code runs bzr by forking a new copy of Python, but
 
216
        that is slower, harder to debug, and generally not necessary.
 
217
 
 
218
        This runs bzr through the interface that catches and reports
 
219
        errors, and with logging set to something approximating the
 
220
        default, so that error reporting can be checked.
 
221
 
 
222
        argv -- arguments to invoke bzr
 
223
        retcode -- expected return code, or None for don't-care.
 
224
        """
 
225
        stdout = StringIO()
 
226
        stderr = StringIO()
 
227
        self.log('run bzr: %s', ' '.join(argv))
 
228
        handler = logging.StreamHandler(stderr)
 
229
        handler.setFormatter(bzrlib.trace.QuietFormatter())
 
230
        handler.setLevel(logging.INFO)
 
231
        logger = logging.getLogger('')
 
232
        logger.addHandler(handler)
 
233
        try:
 
234
            result = self.apply_redirected(None, stdout, stderr,
 
235
                                           bzrlib.commands.run_bzr_catch_errors,
 
236
                                           argv)
 
237
        finally:
 
238
            logger.removeHandler(handler)
 
239
        out = stdout.getvalue()
 
240
        err = stderr.getvalue()
 
241
        if out:
 
242
            self.log('output:\n%s', out)
 
243
        if err:
 
244
            self.log('errors:\n%s', err)
 
245
        if retcode is not None:
 
246
            self.assertEquals(result, retcode)
 
247
        return out, err
 
248
 
93
249
    def run_bzr(self, *args, **kwargs):
94
250
        """Invoke bzr, as if it were run from the command line.
95
251
 
97
253
        overall behavior of the bzr application (rather than a unit test
98
254
        or a functional test of the library.)
99
255
 
100
 
        Much of the old code runs bzr by forking a new copy of Python, but
101
 
        that is slower, harder to debug, and generally not necessary.
 
256
        This sends the stdout/stderr results into the test's log,
 
257
        where it may be useful for debugging.  See also run_captured.
102
258
        """
103
 
        retcode = kwargs.get('retcode', 0)
104
 
        result = self.apply_redirected(None, None, None,
105
 
                                       bzrlib.commands.run_bzr, args)
106
 
        self.assertEquals(result, retcode)
107
 
        
108
 
        
 
259
        retcode = kwargs.pop('retcode', 0)
 
260
        return self.run_bzr_captured(args, retcode)
 
261
 
109
262
    def check_inventory_shape(self, inv, shape):
110
 
        """
111
 
        Compare an inventory to a list of expected names.
 
263
        """Compare an inventory to a list of expected names.
112
264
 
113
265
        Fail if they are not precisely equal.
114
266
        """
132
284
        """Call callable with redirected std io pipes.
133
285
 
134
286
        Returns the return code."""
135
 
        from StringIO import StringIO
136
287
        if not callable(a_callable):
137
288
            raise ValueError("a_callable must be callable.")
138
289
        if stdin is None:
139
290
            stdin = StringIO("")
140
291
        if stdout is None:
141
 
            stdout = self._log_file
 
292
            if hasattr(self, "_log_file"):
 
293
                stdout = self._log_file
 
294
            else:
 
295
                stdout = StringIO()
142
296
        if stderr is None:
143
 
            stderr = self._log_file
 
297
            if hasattr(self, "_log_file"):
 
298
                stderr = self._log_file
 
299
            else:
 
300
                stderr = StringIO()
144
301
        real_stdin = sys.stdin
145
302
        real_stdout = sys.stdout
146
303
        real_stderr = sys.stderr
181
338
        if contents != expect:
182
339
            self.log("expected: %r" % expect)
183
340
            self.log("actually: %r" % contents)
184
 
            self.fail("contents of %s not as expected")
 
341
            self.fail("contents of %s not as expected" % filename)
185
342
 
186
343
    def _make_test_root(self):
187
 
        import os
188
 
        import shutil
189
 
        import tempfile
190
 
        
191
344
        if TestCaseInTempDir.TEST_ROOT is not None:
192
345
            return
193
 
        TestCaseInTempDir.TEST_ROOT = os.path.abspath(
194
 
                                 tempfile.mkdtemp(suffix='.tmp',
195
 
                                                  prefix=self._TEST_NAME + '-',
196
 
                                                  dir=os.curdir))
197
 
    
 
346
        i = 0
 
347
        while True:
 
348
            root = 'test%04d.tmp' % i
 
349
            try:
 
350
                os.mkdir(root)
 
351
            except OSError, e:
 
352
                if e.errno == errno.EEXIST:
 
353
                    i += 1
 
354
                    continue
 
355
                else:
 
356
                    raise
 
357
            # successfully created
 
358
            TestCaseInTempDir.TEST_ROOT = os.path.abspath(root)
 
359
            break
198
360
        # make a fake bzr directory there to prevent any tests propagating
199
361
        # up onto the source directory's real branch
200
362
        os.mkdir(os.path.join(TestCaseInTempDir.TEST_ROOT, '.bzr'))
201
363
 
202
364
    def setUp(self):
203
365
        super(TestCaseInTempDir, self).setUp()
204
 
        import os
205
366
        self._make_test_root()
206
367
        self._currentdir = os.getcwdu()
207
 
        short_id = self.id().replace('bzrlib.selftest.', '')
 
368
        short_id = self.id().replace('bzrlib.selftest.', '') \
 
369
                   .replace('__main__.', '')
208
370
        self.test_dir = os.path.join(self.TEST_ROOT, short_id)
209
371
        os.mkdir(self.test_dir)
210
372
        os.chdir(self.test_dir)
211
373
        
212
374
    def tearDown(self):
213
 
        import os
214
375
        os.chdir(self._currentdir)
215
376
        super(TestCaseInTempDir, self).tearDown()
216
377
 
217
 
    def _formcmd(self, cmd):
218
 
        if isinstance(cmd, basestring):
219
 
            cmd = cmd.split()
220
 
        if cmd[0] == 'bzr':
221
 
            cmd[0] = self.BZRPATH
222
 
            if self.OVERRIDE_PYTHON:
223
 
                cmd.insert(0, self.OVERRIDE_PYTHON)
224
 
        self.log('$ %r' % cmd)
225
 
        return cmd
226
 
 
227
 
    def runcmd(self, cmd, retcode=0):
228
 
        """Run one command and check the return code.
229
 
 
230
 
        Returns a tuple of (stdout,stderr) strings.
231
 
 
232
 
        If a single string is based, it is split into words.
233
 
        For commands that are not simple space-separated words, please
234
 
        pass a list instead."""
235
 
        cmd = self._formcmd(cmd)
236
 
        self.log('$ ' + ' '.join(cmd))
237
 
        actual_retcode = subprocess.call(cmd, stdout=self._log_file,
238
 
                                         stderr=self._log_file)
239
 
        if retcode != actual_retcode:
240
 
            raise CommandFailed("test failed: %r returned %d, expected %d"
241
 
                                % (cmd, actual_retcode, retcode))
242
 
 
243
 
    def backtick(self, cmd, retcode=0):
244
 
        """Run a command and return its output"""
245
 
        cmd = self._formcmd(cmd)
246
 
        child = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=self._log_file)
247
 
        outd, errd = child.communicate()
248
 
        self.log(outd)
249
 
        actual_retcode = child.wait()
250
 
 
251
 
        outd = outd.replace('\r', '')
252
 
 
253
 
        if retcode != actual_retcode:
254
 
            raise CommandFailed("test failed: %r returned %d, expected %d"
255
 
                                % (cmd, actual_retcode, retcode))
256
 
 
257
 
        return outd
258
 
 
259
 
 
260
 
 
261
378
    def build_tree(self, shape):
262
379
        """Build a test tree according to a pattern.
263
380
 
267
384
        This doesn't add anything to a branch.
268
385
        """
269
386
        # XXX: It's OK to just create them using forward slashes on windows?
270
 
        import os
271
387
        for name in shape:
272
388
            assert isinstance(name, basestring)
273
389
            if name[-1] == '/':
276
392
                f = file(name, 'wt')
277
393
                print >>f, "contents of", name
278
394
                f.close()
279
 
                
280
395
 
 
396
    def failUnlessExists(self, path):
 
397
        """Fail unless path, which may be abs or relative, exists."""
 
398
        self.failUnless(os.path.exists(path))
 
399
        
281
400
 
282
401
class MetaTestLog(TestCase):
283
402
    def test_logging(self):
288
407
        ##assert 0
289
408
 
290
409
 
 
410
def filter_suite_by_re(suite, pattern):
 
411
    result = TestUtil.TestSuite()
 
412
    filter_re = re.compile(pattern)
 
413
    for test in iter_suite_tests(suite):
 
414
        if filter_re.search(test.id()):
 
415
            result.addTest(test)
 
416
    return result
 
417
 
 
418
 
 
419
def run_suite(suite, name='test', verbose=False, pattern=".*"):
 
420
    TestCaseInTempDir._TEST_NAME = name
 
421
    if verbose:
 
422
        verbosity = 2
 
423
    else:
 
424
        verbosity = 1
 
425
    runner = TextTestRunner(stream=sys.stdout,
 
426
                            descriptions=0,
 
427
                            verbosity=verbosity)
 
428
    if pattern != '.*':
 
429
        suite = filter_suite_by_re(suite, pattern)
 
430
    result = runner.run(suite)
 
431
    # This is still a little bogus, 
 
432
    # but only a little. Folk not using our testrunner will
 
433
    # have to delete their temp directories themselves.
 
434
    if result.wasSuccessful():
 
435
        if TestCaseInTempDir.TEST_ROOT is not None:
 
436
            shutil.rmtree(TestCaseInTempDir.TEST_ROOT) 
 
437
    else:
 
438
        print "Failed tests working directories are in '%s'\n" % TestCaseInTempDir.TEST_ROOT
 
439
    return result.wasSuccessful()
 
440
 
 
441
 
291
442
def selftest(verbose=False, pattern=".*"):
292
443
    """Run the whole test suite under the enhanced runner"""
293
444
    return run_suite(test_suite(), 'testbzr', verbose=verbose, pattern=pattern)
295
446
 
296
447
def test_suite():
297
448
    """Build and return TestSuite for the whole program."""
298
 
    from bzrlib.selftest.TestUtil import TestLoader, TestSuite
299
 
    import bzrlib, bzrlib.store, bzrlib.inventory, bzrlib.branch
300
 
    import bzrlib.osutils, bzrlib.commands, bzrlib.merge3, bzrlib.plugin
 
449
    import bzrlib.store, bzrlib.inventory, bzrlib.branch
 
450
    import bzrlib.osutils, bzrlib.merge3, bzrlib.plugin
301
451
    from doctest import DocTestSuite
302
 
    import os
303
 
    import shutil
304
 
    import time
305
 
    import sys
306
452
 
307
453
    global MODULES_TO_TEST, MODULES_TO_DOCTEST
308
454
 
309
455
    testmod_names = \
310
456
                  ['bzrlib.selftest.MetaTestLog',
 
457
                   'bzrlib.selftest.testidentitymap',
311
458
                   'bzrlib.selftest.testinv',
 
459
                   'bzrlib.selftest.test_ancestry',
 
460
                   'bzrlib.selftest.test_commit',
 
461
                   'bzrlib.selftest.test_commit_merge',
 
462
                   'bzrlib.selftest.testconfig',
312
463
                   'bzrlib.selftest.versioning',
313
464
                   'bzrlib.selftest.testmerge3',
 
465
                   'bzrlib.selftest.testmerge',
314
466
                   'bzrlib.selftest.testhashcache',
315
467
                   'bzrlib.selftest.teststatus',
316
468
                   'bzrlib.selftest.testlog',
317
469
                   'bzrlib.selftest.testrevisionnamespaces',
318
470
                   'bzrlib.selftest.testbranch',
319
471
                   'bzrlib.selftest.testrevision',
 
472
                   'bzrlib.selftest.test_revision_info',
320
473
                   'bzrlib.selftest.test_merge_core',
321
474
                   'bzrlib.selftest.test_smart_add',
 
475
                   'bzrlib.selftest.test_bad_files',
322
476
                   'bzrlib.selftest.testdiff',
323
477
                   'bzrlib.selftest.test_parent',
324
478
                   'bzrlib.selftest.test_xml',
 
479
                   'bzrlib.selftest.test_weave',
325
480
                   'bzrlib.selftest.testfetch',
326
481
                   'bzrlib.selftest.whitebox',
327
482
                   'bzrlib.selftest.teststore',
328
483
                   'bzrlib.selftest.blackbox',
 
484
                   'bzrlib.selftest.testsampler',
 
485
                   'bzrlib.selftest.testtransactions',
 
486
                   'bzrlib.selftest.testtransport',
 
487
                   'bzrlib.selftest.testgraph',
 
488
                   'bzrlib.selftest.testworkingtree',
 
489
                   'bzrlib.selftest.test_upgrade',
 
490
                   'bzrlib.selftest.test_conflicts',
329
491
                   ]
330
492
 
331
493
    for m in (bzrlib.store, bzrlib.inventory, bzrlib.branch,