/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/selftest/__init__.py

  • Committer: Robert Collins
  • Date: 2005-10-17 20:38:12 UTC
  • Revision ID: robertc@robertcollins.net-20051017203812-7f01cc90974d01ca
Typo in config.py (Thanks Fabbione)

Show diffs side-by-side

added added

removed removed

Lines of Context:
 
1
# Copyright (C) 2005 by Canonical Ltd
 
2
 
 
3
# This program is free software; you can redistribute it and/or modify
 
4
# it under the terms of the GNU General Public License as published by
 
5
# the Free Software Foundation; either version 2 of the License, or
 
6
# (at your option) any later version.
 
7
 
 
8
# This program is distributed in the hope that it will be useful,
 
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
 
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
11
# GNU General Public License for more details.
 
12
 
 
13
# You should have received a copy of the GNU General Public License
 
14
# along with this program; if not, write to the Free Software
 
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
16
 
 
17
 
 
18
from cStringIO import StringIO
 
19
import difflib
 
20
import errno
 
21
import logging
 
22
import os
 
23
import re
 
24
import shutil
 
25
import sys
 
26
import tempfile
 
27
import unittest
 
28
import time
 
29
 
 
30
import bzrlib.commands
 
31
import bzrlib.trace
 
32
import bzrlib.fetch
 
33
import bzrlib.osutils as osutils
 
34
from bzrlib.selftest import TestUtil
 
35
from bzrlib.selftest.TestUtil import TestLoader, TestSuite
 
36
from bzrlib.selftest.treeshape import build_tree_contents
 
37
 
 
38
MODULES_TO_TEST = []
 
39
MODULES_TO_DOCTEST = []
 
40
 
 
41
from logging import debug, warning, error
 
42
 
 
43
 
 
44
 
 
45
class EarlyStoppingTestResultAdapter(object):
 
46
    """An adapter for TestResult to stop at the first first failure or error"""
 
47
 
 
48
    def __init__(self, result):
 
49
        self._result = result
 
50
 
 
51
    def addError(self, test, err):
 
52
        self._result.addError(test, err)
 
53
        self._result.stop()
 
54
 
 
55
    def addFailure(self, test, err):
 
56
        self._result.addFailure(test, err)
 
57
        self._result.stop()
 
58
 
 
59
    def __getattr__(self, name):
 
60
        return getattr(self._result, name)
 
61
 
 
62
    def __setattr__(self, name, value):
 
63
        if name == '_result':
 
64
            object.__setattr__(self, name, value)
 
65
        return setattr(self._result, name, value)
 
66
 
 
67
 
 
68
class _MyResult(unittest._TextTestResult):
 
69
    """
 
70
    Custom TestResult.
 
71
 
 
72
    No special behaviour for now.
 
73
    """
 
74
 
 
75
    def _elapsedTime(self):
 
76
        return "(Took %.3fs)" % (time.time() - self._start_time)
 
77
 
 
78
    def startTest(self, test):
 
79
        unittest.TestResult.startTest(self, test)
 
80
        # TODO: Maybe show test.shortDescription somewhere?
 
81
        what = test.shortDescription() or test.id()        
 
82
        if self.showAll:
 
83
            self.stream.write('%-70.70s' % what)
 
84
        self.stream.flush()
 
85
        self._start_time = time.time()
 
86
 
 
87
    def addError(self, test, err):
 
88
        unittest.TestResult.addError(self, test, err)
 
89
        if self.showAll:
 
90
            self.stream.writeln("ERROR %s" % self._elapsedTime())
 
91
        elif self.dots:
 
92
            self.stream.write('E')
 
93
        self.stream.flush()
 
94
 
 
95
    def addFailure(self, test, err):
 
96
        unittest.TestResult.addFailure(self, test, err)
 
97
        if self.showAll:
 
98
            self.stream.writeln("FAIL %s" % self._elapsedTime())
 
99
        elif self.dots:
 
100
            self.stream.write('F')
 
101
        self.stream.flush()
 
102
 
 
103
    def addSuccess(self, test):
 
104
        if self.showAll:
 
105
            self.stream.writeln('OK %s' % self._elapsedTime())
 
106
        elif self.dots:
 
107
            self.stream.write('~')
 
108
        self.stream.flush()
 
109
        unittest.TestResult.addSuccess(self, test)
 
110
 
 
111
    def printErrorList(self, flavour, errors):
 
112
        for test, err in errors:
 
113
            self.stream.writeln(self.separator1)
 
114
            self.stream.writeln("%s: %s" % (flavour,self.getDescription(test)))
 
115
            if hasattr(test, '_get_log'):
 
116
                self.stream.writeln()
 
117
                self.stream.writeln('log from this test:')
 
118
                print >>self.stream, test._get_log()
 
119
            self.stream.writeln(self.separator2)
 
120
            self.stream.writeln("%s" % err)
 
121
 
 
122
 
 
123
class TextTestRunner(unittest.TextTestRunner):
 
124
    stop_on_failure = False
 
125
 
 
126
    def _makeResult(self):
 
127
        result = _MyResult(self.stream, self.descriptions, self.verbosity)
 
128
        if self.stop_on_failure:
 
129
            result = EarlyStoppingTestResultAdapter(result)
 
130
        return result
 
131
 
 
132
 
 
133
def iter_suite_tests(suite):
 
134
    """Return all tests in a suite, recursing through nested suites"""
 
135
    for item in suite._tests:
 
136
        if isinstance(item, unittest.TestCase):
 
137
            yield item
 
138
        elif isinstance(item, unittest.TestSuite):
 
139
            for r in iter_suite_tests(item):
 
140
                yield r
 
141
        else:
 
142
            raise Exception('unknown object %r inside test suite %r'
 
143
                            % (item, suite))
 
144
 
 
145
 
 
146
class TestSkipped(Exception):
 
147
    """Indicates that a test was intentionally skipped, rather than failing."""
 
148
    # XXX: Not used yet
 
149
 
 
150
 
 
151
class CommandFailed(Exception):
 
152
    pass
 
153
 
 
154
class TestCase(unittest.TestCase):
 
155
    """Base class for bzr unit tests.
 
156
    
 
157
    Tests that need access to disk resources should subclass 
 
158
    TestCaseInTempDir not TestCase.
 
159
 
 
160
    Error and debug log messages are redirected from their usual
 
161
    location into a temporary file, the contents of which can be
 
162
    retrieved by _get_log().
 
163
       
 
164
    There are also convenience functions to invoke bzr's command-line
 
165
    routine, and to build and check bzr trees."""
 
166
 
 
167
    BZRPATH = 'bzr'
 
168
    _log_file_name = None
 
169
 
 
170
    def setUp(self):
 
171
        unittest.TestCase.setUp(self)
 
172
        self.oldenv = os.environ.get('HOME', None)
 
173
        os.environ['HOME'] = os.getcwd()
 
174
        self.bzr_email = os.environ.get('BZREMAIL')
 
175
        if self.bzr_email is not None:
 
176
            del os.environ['BZREMAIL']
 
177
        self.email = os.environ.get('EMAIL')
 
178
        if self.email is not None:
 
179
            del os.environ['EMAIL']
 
180
        bzrlib.trace.disable_default_logging()
 
181
        self._enable_file_logging()
 
182
 
 
183
    def _ndiff_strings(self, a, b):
 
184
        """Return ndiff between two strings containing lines."""
 
185
        difflines = difflib.ndiff(a.splitlines(True),
 
186
                                  b.splitlines(True),
 
187
                                  linejunk=lambda x: False,
 
188
                                  charjunk=lambda x: False)
 
189
        return ''.join(difflines)
 
190
 
 
191
    def assertEqualDiff(self, a, b):
 
192
        """Assert two texts are equal, if not raise an exception.
 
193
        
 
194
        This is intended for use with multi-line strings where it can 
 
195
        be hard to find the differences by eye.
 
196
        """
 
197
        # TODO: perhaps override assertEquals to call this for strings?
 
198
        if a == b:
 
199
            return
 
200
        raise AssertionError("texts not equal:\n" + 
 
201
                             self._ndiff_strings(a, b))      
 
202
 
 
203
    def assertContainsRe(self, haystack, needle_re):
 
204
        """Assert that a contains something matching a regular expression."""
 
205
        if not re.search(needle_re, haystack):
 
206
            raise AssertionError('pattern "%s" not found in "%s"'
 
207
                    % (needle_re, haystack))
 
208
        
 
209
    def _enable_file_logging(self):
 
210
        fileno, name = tempfile.mkstemp(suffix='.log', prefix='testbzr')
 
211
 
 
212
        self._log_file = os.fdopen(fileno, 'w+')
 
213
 
 
214
        hdlr = logging.StreamHandler(self._log_file)
 
215
        hdlr.setLevel(logging.DEBUG)
 
216
        hdlr.setFormatter(logging.Formatter('%(levelname)8s  %(message)s'))
 
217
        logging.getLogger('').addHandler(hdlr)
 
218
        logging.getLogger('').setLevel(logging.DEBUG)
 
219
        self._log_hdlr = hdlr
 
220
        debug('opened log file %s', name)
 
221
        
 
222
        self._log_file_name = name
 
223
 
 
224
    def tearDown(self):
 
225
        os.environ['HOME'] = self.oldenv
 
226
        if os.environ.get('BZREMAIL') is not None:
 
227
            del os.environ['BZREMAIL']
 
228
        if self.bzr_email is not None:
 
229
            os.environ['BZREMAIL'] = self.bzr_email
 
230
        if os.environ.get('EMAIL') is not None:
 
231
            del os.environ['EMAIL']
 
232
        if self.email is not None:
 
233
            os.environ['EMAIL'] = self.email
 
234
        logging.getLogger('').removeHandler(self._log_hdlr)
 
235
        bzrlib.trace.enable_default_logging()
 
236
        logging.debug('%s teardown', self.id())
 
237
        self._log_file.close()
 
238
        unittest.TestCase.tearDown(self)
 
239
 
 
240
    def log(self, *args):
 
241
        logging.debug(*args)
 
242
 
 
243
    def _get_log(self):
 
244
        """Return as a string the log for this test"""
 
245
        if self._log_file_name:
 
246
            return open(self._log_file_name).read()
 
247
        else:
 
248
            return ''
 
249
 
 
250
    def capture(self, cmd):
 
251
        """Shortcut that splits cmd into words, runs, and returns stdout"""
 
252
        return self.run_bzr_captured(cmd.split())[0]
 
253
 
 
254
    def run_bzr_captured(self, argv, retcode=0):
 
255
        """Invoke bzr and return (result, stdout, stderr).
 
256
 
 
257
        Useful for code that wants to check the contents of the
 
258
        output, the way error messages are presented, etc.
 
259
 
 
260
        This should be the main method for tests that want to exercise the
 
261
        overall behavior of the bzr application (rather than a unit test
 
262
        or a functional test of the library.)
 
263
 
 
264
        Much of the old code runs bzr by forking a new copy of Python, but
 
265
        that is slower, harder to debug, and generally not necessary.
 
266
 
 
267
        This runs bzr through the interface that catches and reports
 
268
        errors, and with logging set to something approximating the
 
269
        default, so that error reporting can be checked.
 
270
 
 
271
        argv -- arguments to invoke bzr
 
272
        retcode -- expected return code, or None for don't-care.
 
273
        """
 
274
        stdout = StringIO()
 
275
        stderr = StringIO()
 
276
        self.log('run bzr: %s', ' '.join(argv))
 
277
        handler = logging.StreamHandler(stderr)
 
278
        handler.setFormatter(bzrlib.trace.QuietFormatter())
 
279
        handler.setLevel(logging.INFO)
 
280
        logger = logging.getLogger('')
 
281
        logger.addHandler(handler)
 
282
        try:
 
283
            result = self.apply_redirected(None, stdout, stderr,
 
284
                                           bzrlib.commands.run_bzr_catch_errors,
 
285
                                           argv)
 
286
        finally:
 
287
            logger.removeHandler(handler)
 
288
        out = stdout.getvalue()
 
289
        err = stderr.getvalue()
 
290
        if out:
 
291
            self.log('output:\n%s', out)
 
292
        if err:
 
293
            self.log('errors:\n%s', err)
 
294
        if retcode is not None:
 
295
            self.assertEquals(result, retcode)
 
296
        return out, err
 
297
 
 
298
    def run_bzr(self, *args, **kwargs):
 
299
        """Invoke bzr, as if it were run from the command line.
 
300
 
 
301
        This should be the main method for tests that want to exercise the
 
302
        overall behavior of the bzr application (rather than a unit test
 
303
        or a functional test of the library.)
 
304
 
 
305
        This sends the stdout/stderr results into the test's log,
 
306
        where it may be useful for debugging.  See also run_captured.
 
307
        """
 
308
        retcode = kwargs.pop('retcode', 0)
 
309
        return self.run_bzr_captured(args, retcode)
 
310
 
 
311
    def check_inventory_shape(self, inv, shape):
 
312
        """Compare an inventory to a list of expected names.
 
313
 
 
314
        Fail if they are not precisely equal.
 
315
        """
 
316
        extras = []
 
317
        shape = list(shape)             # copy
 
318
        for path, ie in inv.entries():
 
319
            name = path.replace('\\', '/')
 
320
            if ie.kind == 'dir':
 
321
                name = name + '/'
 
322
            if name in shape:
 
323
                shape.remove(name)
 
324
            else:
 
325
                extras.append(name)
 
326
        if shape:
 
327
            self.fail("expected paths not found in inventory: %r" % shape)
 
328
        if extras:
 
329
            self.fail("unexpected paths found in inventory: %r" % extras)
 
330
 
 
331
    def apply_redirected(self, stdin=None, stdout=None, stderr=None,
 
332
                         a_callable=None, *args, **kwargs):
 
333
        """Call callable with redirected std io pipes.
 
334
 
 
335
        Returns the return code."""
 
336
        if not callable(a_callable):
 
337
            raise ValueError("a_callable must be callable.")
 
338
        if stdin is None:
 
339
            stdin = StringIO("")
 
340
        if stdout is None:
 
341
            if hasattr(self, "_log_file"):
 
342
                stdout = self._log_file
 
343
            else:
 
344
                stdout = StringIO()
 
345
        if stderr is None:
 
346
            if hasattr(self, "_log_file"):
 
347
                stderr = self._log_file
 
348
            else:
 
349
                stderr = StringIO()
 
350
        real_stdin = sys.stdin
 
351
        real_stdout = sys.stdout
 
352
        real_stderr = sys.stderr
 
353
        try:
 
354
            sys.stdout = stdout
 
355
            sys.stderr = stderr
 
356
            sys.stdin = stdin
 
357
            return a_callable(*args, **kwargs)
 
358
        finally:
 
359
            sys.stdout = real_stdout
 
360
            sys.stderr = real_stderr
 
361
            sys.stdin = real_stdin
 
362
 
 
363
 
 
364
BzrTestBase = TestCase
 
365
 
 
366
     
 
367
class TestCaseInTempDir(TestCase):
 
368
    """Derived class that runs a test within a temporary directory.
 
369
 
 
370
    This is useful for tests that need to create a branch, etc.
 
371
 
 
372
    The directory is created in a slightly complex way: for each
 
373
    Python invocation, a new temporary top-level directory is created.
 
374
    All test cases create their own directory within that.  If the
 
375
    tests complete successfully, the directory is removed.
 
376
 
 
377
    InTempDir is an old alias for FunctionalTestCase.
 
378
    """
 
379
 
 
380
    TEST_ROOT = None
 
381
    _TEST_NAME = 'test'
 
382
    OVERRIDE_PYTHON = 'python'
 
383
 
 
384
    def check_file_contents(self, filename, expect):
 
385
        self.log("check contents of file %s" % filename)
 
386
        contents = file(filename, 'r').read()
 
387
        if contents != expect:
 
388
            self.log("expected: %r" % expect)
 
389
            self.log("actually: %r" % contents)
 
390
            self.fail("contents of %s not as expected" % filename)
 
391
 
 
392
    def _make_test_root(self):
 
393
        if TestCaseInTempDir.TEST_ROOT is not None:
 
394
            return
 
395
        i = 0
 
396
        while True:
 
397
            root = 'test%04d.tmp' % i
 
398
            try:
 
399
                os.mkdir(root)
 
400
            except OSError, e:
 
401
                if e.errno == errno.EEXIST:
 
402
                    i += 1
 
403
                    continue
 
404
                else:
 
405
                    raise
 
406
            # successfully created
 
407
            TestCaseInTempDir.TEST_ROOT = os.path.abspath(root)
 
408
            break
 
409
        # make a fake bzr directory there to prevent any tests propagating
 
410
        # up onto the source directory's real branch
 
411
        os.mkdir(os.path.join(TestCaseInTempDir.TEST_ROOT, '.bzr'))
 
412
 
 
413
    def setUp(self):
 
414
        self._make_test_root()
 
415
        self._currentdir = os.getcwdu()
 
416
        short_id = self.id().replace('bzrlib.selftest.', '') \
 
417
                   .replace('__main__.', '')
 
418
        self.test_dir = os.path.join(self.TEST_ROOT, short_id)
 
419
        os.mkdir(self.test_dir)
 
420
        os.chdir(self.test_dir)
 
421
        super(TestCaseInTempDir, self).setUp()
 
422
        
 
423
    def tearDown(self):
 
424
        os.chdir(self._currentdir)
 
425
        super(TestCaseInTempDir, self).tearDown()
 
426
 
 
427
    def build_tree(self, shape):
 
428
        """Build a test tree according to a pattern.
 
429
 
 
430
        shape is a sequence of file specifications.  If the final
 
431
        character is '/', a directory is created.
 
432
 
 
433
        This doesn't add anything to a branch.
 
434
        """
 
435
        # XXX: It's OK to just create them using forward slashes on windows?
 
436
        for name in shape:
 
437
            assert isinstance(name, basestring)
 
438
            if name[-1] == '/':
 
439
                os.mkdir(name[:-1])
 
440
            else:
 
441
                f = file(name, 'wt')
 
442
                print >>f, "contents of", name
 
443
                f.close()
 
444
 
 
445
    def build_tree_contents(self, shape):
 
446
        bzrlib.selftest.build_tree_contents(shape)
 
447
 
 
448
    def failUnlessExists(self, path):
 
449
        """Fail unless path, which may be abs or relative, exists."""
 
450
        self.failUnless(osutils.lexists(path))
 
451
        
 
452
 
 
453
class MetaTestLog(TestCase):
 
454
    def test_logging(self):
 
455
        """Test logs are captured when a test fails."""
 
456
        logging.info('an info message')
 
457
        warning('something looks dodgy...')
 
458
        logging.debug('hello, test is running')
 
459
        ## assert 0
 
460
 
 
461
 
 
462
def filter_suite_by_re(suite, pattern):
 
463
    result = TestUtil.TestSuite()
 
464
    filter_re = re.compile(pattern)
 
465
    for test in iter_suite_tests(suite):
 
466
        if filter_re.search(test.id()):
 
467
            result.addTest(test)
 
468
    return result
 
469
 
 
470
 
 
471
def run_suite(suite, name='test', verbose=False, pattern=".*",
 
472
              stop_on_failure=False):
 
473
    TestCaseInTempDir._TEST_NAME = name
 
474
    if verbose:
 
475
        verbosity = 2
 
476
    else:
 
477
        verbosity = 1
 
478
    runner = TextTestRunner(stream=sys.stdout,
 
479
                            descriptions=0,
 
480
                            verbosity=verbosity)
 
481
    runner.stop_on_failure=stop_on_failure
 
482
    if pattern != '.*':
 
483
        suite = filter_suite_by_re(suite, pattern)
 
484
    result = runner.run(suite)
 
485
    # This is still a little bogus, 
 
486
    # but only a little. Folk not using our testrunner will
 
487
    # have to delete their temp directories themselves.
 
488
    if result.wasSuccessful():
 
489
        if TestCaseInTempDir.TEST_ROOT is not None:
 
490
            shutil.rmtree(TestCaseInTempDir.TEST_ROOT) 
 
491
    else:
 
492
        print "Failed tests working directories are in '%s'\n" % TestCaseInTempDir.TEST_ROOT
 
493
    return result.wasSuccessful()
 
494
 
 
495
 
 
496
def selftest(verbose=False, pattern=".*", stop_on_failure=True):
 
497
    """Run the whole test suite under the enhanced runner"""
 
498
    return run_suite(test_suite(), 'testbzr', verbose=verbose, pattern=pattern,
 
499
                     stop_on_failure=stop_on_failure)
 
500
 
 
501
 
 
502
def test_suite():
 
503
    """Build and return TestSuite for the whole program."""
 
504
    import bzrlib.store, bzrlib.inventory, bzrlib.branch
 
505
    import bzrlib.osutils, bzrlib.merge3, bzrlib.plugin
 
506
    from doctest import DocTestSuite
 
507
 
 
508
    global MODULES_TO_TEST, MODULES_TO_DOCTEST
 
509
 
 
510
    testmod_names = \
 
511
                  ['bzrlib.selftest.MetaTestLog',
 
512
                   'bzrlib.selftest.testgpg',
 
513
                   'bzrlib.selftest.testidentitymap',
 
514
                   'bzrlib.selftest.testinv',
 
515
                   'bzrlib.selftest.test_ancestry',
 
516
                   'bzrlib.selftest.test_commit',
 
517
                   'bzrlib.selftest.test_commit_merge',
 
518
                   'bzrlib.selftest.testconfig',
 
519
                   'bzrlib.selftest.versioning',
 
520
                   'bzrlib.selftest.testmerge3',
 
521
                   'bzrlib.selftest.testmerge',
 
522
                   'bzrlib.selftest.testhashcache',
 
523
                   'bzrlib.selftest.teststatus',
 
524
                   'bzrlib.selftest.testlog',
 
525
                   'bzrlib.selftest.testrevisionnamespaces',
 
526
                   'bzrlib.selftest.testbranch',
 
527
                   'bzrlib.selftest.testrevision',
 
528
                   'bzrlib.selftest.test_revision_info',
 
529
                   'bzrlib.selftest.test_merge_core',
 
530
                   'bzrlib.selftest.test_smart_add',
 
531
                   'bzrlib.selftest.test_bad_files',
 
532
                   'bzrlib.selftest.testdiff',
 
533
                   'bzrlib.selftest.test_parent',
 
534
                   'bzrlib.selftest.test_xml',
 
535
                   'bzrlib.selftest.test_weave',
 
536
                   'bzrlib.selftest.testfetch',
 
537
                   'bzrlib.selftest.whitebox',
 
538
                   'bzrlib.selftest.teststore',
 
539
                   'bzrlib.selftest.blackbox',
 
540
                   'bzrlib.selftest.testsampler',
 
541
                   'bzrlib.selftest.testtransactions',
 
542
                   'bzrlib.selftest.testtransport',
 
543
                   'bzrlib.selftest.testgraph',
 
544
                   'bzrlib.selftest.testworkingtree',
 
545
                   'bzrlib.selftest.test_upgrade',
 
546
                   'bzrlib.selftest.test_conflicts',
 
547
                   'bzrlib.selftest.testtestament',
 
548
                   'bzrlib.selftest.testannotate',
 
549
                   'bzrlib.selftest.testrevprops',
 
550
                   'bzrlib.selftest.testoptions',
 
551
                   ]
 
552
 
 
553
    for m in (bzrlib.store, bzrlib.inventory, bzrlib.branch,
 
554
              bzrlib.osutils, bzrlib.commands, bzrlib.merge3):
 
555
        if m not in MODULES_TO_DOCTEST:
 
556
            MODULES_TO_DOCTEST.append(m)
 
557
 
 
558
    TestCase.BZRPATH = os.path.join(os.path.realpath(os.path.dirname(bzrlib.__path__[0])), 'bzr')
 
559
    print '%-30s %s' % ('bzr binary', TestCase.BZRPATH)
 
560
    print
 
561
    suite = TestSuite()
 
562
    suite.addTest(TestLoader().loadTestsFromNames(testmod_names))
 
563
    for m in MODULES_TO_TEST:
 
564
         suite.addTest(TestLoader().loadTestsFromModule(m))
 
565
    for m in (MODULES_TO_DOCTEST):
 
566
        suite.addTest(DocTestSuite(m))
 
567
    for p in bzrlib.plugin.all_plugins:
 
568
        if hasattr(p, 'test_suite'):
 
569
            suite.addTest(p.test_suite())
 
570
    return suite
 
571