/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/tests/__init__.py

Merge from bzr.dev.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/__init__.py
1
 
# Copyright (C) 2005, 2006 by Canonical Ltd
 
1
# Copyright (C) 2005, 2006 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
43
43
import time
44
44
 
45
45
 
46
 
from bzrlib import memorytree
 
46
from bzrlib import (
 
47
    bzrdir,
 
48
    debug,
 
49
    errors,
 
50
    memorytree,
 
51
    osutils,
 
52
    progress,
 
53
    urlutils,
 
54
    )
47
55
import bzrlib.branch
48
 
import bzrlib.bzrdir as bzrdir
49
56
import bzrlib.commands
50
57
import bzrlib.bundle.serializer
51
 
import bzrlib.errors as errors
52
58
import bzrlib.export
53
59
import bzrlib.inventory
54
60
import bzrlib.iterablefile
61
67
from bzrlib.merge import merge_inner
62
68
import bzrlib.merge3
63
69
import bzrlib.osutils
64
 
import bzrlib.osutils as osutils
65
70
import bzrlib.plugin
66
 
import bzrlib.progress as progress
67
71
from bzrlib.revision import common_ancestor
68
72
import bzrlib.store
69
73
from bzrlib import symbol_versioning
70
74
import bzrlib.trace
71
75
from bzrlib.transport import get_transport
72
76
import bzrlib.transport
73
 
from bzrlib.transport.local import LocalRelpathServer
 
77
from bzrlib.transport.local import LocalURLServer
74
78
from bzrlib.transport.memory import MemoryServer
75
79
from bzrlib.transport.readonly import ReadonlyServer
76
 
from bzrlib.trace import mutter
 
80
from bzrlib.trace import mutter, note
77
81
from bzrlib.tests import TestUtil
78
82
from bzrlib.tests.TestUtil import (
79
83
                          TestSuite,
80
84
                          TestLoader,
81
85
                          )
82
86
from bzrlib.tests.treeshape import build_tree_contents
83
 
import bzrlib.urlutils as urlutils
84
87
from bzrlib.workingtree import WorkingTree, WorkingTreeFormat2
85
88
 
86
 
default_transport = LocalRelpathServer
 
89
default_transport = LocalURLServer
87
90
 
88
91
MODULES_TO_TEST = []
89
92
MODULES_TO_DOCTEST = [
131
134
            ]
132
135
 
133
136
 
134
 
class _MyResult(unittest._TextTestResult):
135
 
    """Custom TestResult.
 
137
class ExtendedTestResult(unittest._TextTestResult):
 
138
    """Accepts, reports and accumulates the results of running tests.
136
139
 
137
 
    Shows output in a different format, including displaying runtime for tests.
 
140
    Compared to this unittest version this class adds support for profiling,
 
141
    benchmarking, stopping as soon as a test fails,  and skipping tests.
 
142
    There are further-specialized subclasses for different types of display.
138
143
    """
 
144
 
139
145
    stop_early = False
140
146
    
141
 
    def __init__(self, stream, descriptions, verbosity, pb=None,
142
 
                 bench_history=None):
 
147
    def __init__(self, stream, descriptions, verbosity,
 
148
                 bench_history=None,
 
149
                 num_tests=None,
 
150
                 ):
143
151
        """Construct new TestResult.
144
152
 
145
153
        :param bench_history: Optionally, a writable file object to accumulate
146
154
            benchmark results.
147
155
        """
148
156
        unittest._TextTestResult.__init__(self, stream, descriptions, verbosity)
149
 
        self.pb = pb
150
157
        if bench_history is not None:
151
158
            from bzrlib.version import _get_bzr_source_tree
152
159
            src_tree = _get_bzr_source_tree()
162
169
                revision_id = ''
163
170
            bench_history.write("--date %s %s\n" % (time.time(), revision_id))
164
171
        self._bench_history = bench_history
 
172
        self.ui = bzrlib.ui.ui_factory
 
173
        self.num_tests = num_tests
 
174
        self.error_count = 0
 
175
        self.failure_count = 0
 
176
        self.skip_count = 0
 
177
        self.count = 0
 
178
        self._overall_start_time = time.time()
165
179
    
166
180
    def extractBenchmarkTime(self, testCase):
167
181
        """Add a benchmark time for the current test case."""
183
197
        """Format seconds as milliseconds with leading spaces."""
184
198
        return "%5dms" % (1000 * seconds)
185
199
 
186
 
    def _ellipsise_unimportant_words(self, a_string, final_width,
187
 
                                   keep_start=False):
188
 
        """Add ellipses (sp?) for overly long strings.
189
 
        
190
 
        :param keep_start: If true preserve the start of a_string rather
191
 
                           than the end of it.
192
 
        """
193
 
        if keep_start:
194
 
            if len(a_string) > final_width:
195
 
                result = a_string[:final_width-3] + '...'
196
 
            else:
197
 
                result = a_string
198
 
        else:
199
 
            if len(a_string) > final_width:
200
 
                result = '...' + a_string[3-final_width:]
201
 
            else:
202
 
                result = a_string
203
 
        return result.ljust(final_width)
 
200
    def _shortened_test_description(self, test):
 
201
        what = test.id()
 
202
        what = re.sub(r'^bzrlib\.(tests|benchmark)\.', '', what)
 
203
        return what
204
204
 
205
205
    def startTest(self, test):
206
206
        unittest.TestResult.startTest(self, test)
207
 
        # In a short description, the important words are in
208
 
        # the beginning, but in an id, the important words are
209
 
        # at the end
210
 
        SHOW_DESCRIPTIONS = False
211
 
 
212
 
        if not self.showAll and self.dots and self.pb is not None:
213
 
            final_width = 13
214
 
        else:
215
 
            final_width = osutils.terminal_width()
216
 
            final_width = final_width - 15 - 8
217
 
        what = None
218
 
        if SHOW_DESCRIPTIONS:
219
 
            what = test.shortDescription()
220
 
            if what:
221
 
                what = self._ellipsise_unimportant_words(what, final_width, keep_start=True)
222
 
        if what is None:
223
 
            what = test.id()
224
 
            if what.startswith('bzrlib.tests.'):
225
 
                what = what[13:]
226
 
            what = self._ellipsise_unimportant_words(what, final_width)
227
 
        if self.showAll:
228
 
            self.stream.write(what)
229
 
        elif self.dots and self.pb is not None:
230
 
            self.pb.update(what, self.testsRun - 1, None)
231
 
        self.stream.flush()
 
207
        self.report_test_start(test)
232
208
        self._recordTestStartTime()
233
209
 
234
210
    def _recordTestStartTime(self):
245
221
        if setKeepLogfile is not None:
246
222
            setKeepLogfile()
247
223
        self.extractBenchmarkTime(test)
248
 
        if self.showAll:
249
 
            self.stream.writeln("ERROR %s" % self._testTimeString())
250
 
        elif self.dots and self.pb is None:
251
 
            self.stream.write('E')
252
 
        elif self.dots:
253
 
            self.pb.update(self._ellipsise_unimportant_words('ERROR', 13), self.testsRun, None)
254
 
            self.pb.note(self._ellipsise_unimportant_words(
255
 
                            test.id() + ': ERROR',
256
 
                            osutils.terminal_width()))
257
 
        self.stream.flush()
 
224
        self.report_error(test, err)
258
225
        if self.stop_early:
259
226
            self.stop()
260
227
 
266
233
        if setKeepLogfile is not None:
267
234
            setKeepLogfile()
268
235
        self.extractBenchmarkTime(test)
269
 
        if self.showAll:
270
 
            self.stream.writeln(" FAIL %s" % self._testTimeString())
271
 
        elif self.dots and self.pb is None:
272
 
            self.stream.write('F')
273
 
        elif self.dots:
274
 
            self.pb.update(self._ellipsise_unimportant_words('FAIL', 13), self.testsRun, None)
275
 
            self.pb.note(self._ellipsise_unimportant_words(
276
 
                            test.id() + ': FAIL',
277
 
                            osutils.terminal_width()))
278
 
        self.stream.flush()
 
236
        self.report_failure(test, err)
279
237
        if self.stop_early:
280
238
            self.stop()
281
239
 
286
244
                self._bench_history.write("%s %s\n" % (
287
245
                    self._formatTime(self._benchmarkTime),
288
246
                    test.id()))
289
 
        if self.showAll:
290
 
            self.stream.writeln('   OK %s' % self._testTimeString())
291
 
            for bench_called, stats in getattr(test, '_benchcalls', []):
292
 
                self.stream.writeln('LSProf output for %s(%s, %s)' % bench_called)
293
 
                stats.pprint(file=self.stream)
294
 
        elif self.dots and self.pb is None:
295
 
            self.stream.write('~')
296
 
        elif self.dots:
297
 
            self.pb.update(self._ellipsise_unimportant_words('OK', 13), self.testsRun, None)
298
 
        self.stream.flush()
 
247
        self.report_success(test)
299
248
        unittest.TestResult.addSuccess(self, test)
300
249
 
301
250
    def addSkipped(self, test, skip_excinfo):
302
251
        self.extractBenchmarkTime(test)
303
 
        if self.showAll:
304
 
            print >>self.stream, ' SKIP %s' % self._testTimeString()
305
 
            print >>self.stream, '     %s' % skip_excinfo[1]
306
 
        elif self.dots and self.pb is None:
307
 
            self.stream.write('S')
308
 
        elif self.dots:
309
 
            self.pb.update(self._ellipsise_unimportant_words('SKIP', 13), self.testsRun, None)
310
 
        self.stream.flush()
 
252
        self.report_skip(test, skip_excinfo)
311
253
        # seems best to treat this as success from point-of-view of unittest
312
254
        # -- it actually does nothing so it barely matters :)
313
255
        try:
333
275
            self.stream.writeln(self.separator2)
334
276
            self.stream.writeln("%s" % err)
335
277
 
 
278
    def finished(self):
 
279
        pass
 
280
 
 
281
    def report_cleaning_up(self):
 
282
        pass
 
283
 
 
284
    def report_success(self, test):
 
285
        pass
 
286
 
 
287
 
 
288
class TextTestResult(ExtendedTestResult):
 
289
    """Displays progress and results of tests in text form"""
 
290
 
 
291
    def __init__(self, *args, **kw):
 
292
        ExtendedTestResult.__init__(self, *args, **kw)
 
293
        self.pb = self.ui.nested_progress_bar()
 
294
        self.pb.show_pct = False
 
295
        self.pb.show_spinner = False
 
296
        self.pb.show_eta = False, 
 
297
        self.pb.show_count = False
 
298
        self.pb.show_bar = False
 
299
 
 
300
    def report_starting(self):
 
301
        self.pb.update('[test 0/%d] starting...' % (self.num_tests))
 
302
 
 
303
    def _progress_prefix_text(self):
 
304
        a = '[%d' % self.count
 
305
        if self.num_tests is not None:
 
306
            a +='/%d' % self.num_tests
 
307
        a += ' in %ds' % (time.time() - self._overall_start_time)
 
308
        if self.error_count:
 
309
            a += ', %d errors' % self.error_count
 
310
        if self.failure_count:
 
311
            a += ', %d failed' % self.failure_count
 
312
        if self.skip_count:
 
313
            a += ', %d skipped' % self.skip_count
 
314
        a += ']'
 
315
        return a
 
316
 
 
317
    def report_test_start(self, test):
 
318
        self.count += 1
 
319
        self.pb.update(
 
320
                self._progress_prefix_text()
 
321
                + ' ' 
 
322
                + self._shortened_test_description(test))
 
323
 
 
324
    def report_error(self, test, err):
 
325
        self.error_count += 1
 
326
        self.pb.note('ERROR: %s\n    %s\n' % (
 
327
            self._shortened_test_description(test),
 
328
            err[1],
 
329
            ))
 
330
 
 
331
    def report_failure(self, test, err):
 
332
        self.failure_count += 1
 
333
        self.pb.note('FAIL: %s\n    %s\n' % (
 
334
            self._shortened_test_description(test),
 
335
            err[1],
 
336
            ))
 
337
 
 
338
    def report_skip(self, test, skip_excinfo):
 
339
        self.skip_count += 1
 
340
        if False:
 
341
            # at the moment these are mostly not things we can fix
 
342
            # and so they just produce stipple; use the verbose reporter
 
343
            # to see them.
 
344
            if False:
 
345
                # show test and reason for skip
 
346
                self.pb.note('SKIP: %s\n    %s\n' % (
 
347
                    self._shortened_test_description(test),
 
348
                    skip_excinfo[1]))
 
349
            else:
 
350
                # since the class name was left behind in the still-visible
 
351
                # progress bar...
 
352
                self.pb.note('SKIP: %s' % (skip_excinfo[1]))
 
353
 
 
354
    def report_cleaning_up(self):
 
355
        self.pb.update('cleaning up...')
 
356
 
 
357
    def finished(self):
 
358
        self.pb.finished()
 
359
 
 
360
 
 
361
class VerboseTestResult(ExtendedTestResult):
 
362
    """Produce long output, with one line per test run plus times"""
 
363
 
 
364
    def _ellipsize_to_right(self, a_string, final_width):
 
365
        """Truncate and pad a string, keeping the right hand side"""
 
366
        if len(a_string) > final_width:
 
367
            result = '...' + a_string[3-final_width:]
 
368
        else:
 
369
            result = a_string
 
370
        return result.ljust(final_width)
 
371
 
 
372
    def report_starting(self):
 
373
        self.stream.write('running %d tests...\n' % self.num_tests)
 
374
 
 
375
    def report_test_start(self, test):
 
376
        self.count += 1
 
377
        name = self._shortened_test_description(test)
 
378
        self.stream.write(self._ellipsize_to_right(name, 60))
 
379
        self.stream.flush()
 
380
 
 
381
    def report_error(self, test, err):
 
382
        self.error_count += 1
 
383
        self.stream.writeln('ERROR %s\n    %s' 
 
384
                % (self._testTimeString(), err[1]))
 
385
 
 
386
    def report_failure(self, test, err):
 
387
        self.failure_count += 1
 
388
        self.stream.writeln('FAIL %s\n    %s'
 
389
                % (self._testTimeString(), err[1]))
 
390
 
 
391
    def report_success(self, test):
 
392
        self.stream.writeln('   OK %s' % self._testTimeString())
 
393
        for bench_called, stats in getattr(test, '_benchcalls', []):
 
394
            self.stream.writeln('LSProf output for %s(%s, %s)' % bench_called)
 
395
            stats.pprint(file=self.stream)
 
396
        self.stream.flush()
 
397
 
 
398
    def report_skip(self, test, skip_excinfo):
 
399
        print >>self.stream, ' SKIP %s' % self._testTimeString()
 
400
        print >>self.stream, '     %s' % skip_excinfo[1]
 
401
 
336
402
 
337
403
class TextTestRunner(object):
338
404
    stop_on_failure = False
342
408
                 descriptions=0,
343
409
                 verbosity=1,
344
410
                 keep_output=False,
345
 
                 pb=None,
346
411
                 bench_history=None):
347
412
        self.stream = unittest._WritelnDecorator(stream)
348
413
        self.descriptions = descriptions
349
414
        self.verbosity = verbosity
350
415
        self.keep_output = keep_output
351
 
        self.pb = pb
352
416
        self._bench_history = bench_history
353
417
 
354
 
    def _makeResult(self):
355
 
        result = _MyResult(self.stream,
356
 
                           self.descriptions,
357
 
                           self.verbosity,
358
 
                           pb=self.pb,
359
 
                           bench_history=self._bench_history)
360
 
        result.stop_early = self.stop_on_failure
361
 
        return result
362
 
 
363
418
    def run(self, test):
364
419
        "Run the given test case or test suite."
365
 
        result = self._makeResult()
366
420
        startTime = time.time()
367
 
        if self.pb is not None:
368
 
            self.pb.update('Running tests', 0, test.countTestCases())
 
421
        if self.verbosity == 1:
 
422
            result_class = TextTestResult
 
423
        elif self.verbosity >= 2:
 
424
            result_class = VerboseTestResult
 
425
        result = result_class(self.stream,
 
426
                              self.descriptions,
 
427
                              self.verbosity,
 
428
                              bench_history=self._bench_history,
 
429
                              num_tests=test.countTestCases(),
 
430
                              )
 
431
        result.stop_early = self.stop_on_failure
 
432
        result.report_starting()
369
433
        test.run(result)
370
434
        stopTime = time.time()
371
435
        timeTaken = stopTime - startTime
386
450
            self.stream.writeln(")")
387
451
        else:
388
452
            self.stream.writeln("OK")
389
 
        if self.pb is not None:
390
 
            self.pb.update('Cleaning up', 0, 1)
 
453
        result.report_cleaning_up()
391
454
        # This is still a little bogus, 
392
455
        # but only a little. Folk not using our testrunner will
393
456
        # have to delete their temp directories themselves.
408
471
                        sys.getfilesystemencoding())
409
472
                osutils.rmtree(test_root)
410
473
        else:
411
 
            if self.pb is not None:
412
 
                self.pb.note("Failed tests working directories are in '%s'\n",
413
 
                             test_root)
414
 
            else:
415
 
                self.stream.writeln(
416
 
                    "Failed tests working directories are in '%s'\n" %
417
 
                    test_root)
 
474
            note("Failed tests working directories are in '%s'\n", test_root)
418
475
        TestCaseWithMemoryTransport.TEST_ROOT = None
419
 
        if self.pb is not None:
420
 
            self.pb.clear()
 
476
        result.finished()
421
477
        return result
422
478
 
423
479
 
503
559
        unittest.TestCase.setUp(self)
504
560
        self._cleanEnvironment()
505
561
        bzrlib.trace.disable_default_logging()
 
562
        self._silenceUI()
506
563
        self._startLogFile()
507
564
        self._benchcalls = []
508
565
        self._benchtime = None
509
566
 
 
567
    def _silenceUI(self):
 
568
        """Turn off UI for duration of test"""
 
569
        # by default the UI is off; tests can turn it on if they want it.
 
570
        saved = bzrlib.ui.ui_factory
 
571
        def _restore():
 
572
            bzrlib.ui.ui_factory = saved
 
573
        bzrlib.ui.ui_factory = bzrlib.ui.SilentUIFactory()
 
574
        self.addCleanup(_restore)
 
575
 
510
576
    def _ndiff_strings(self, a, b):
511
577
        """Return ndiff between two strings containing lines.
512
578
        
603
669
            a_callable(*args, **kwargs).
604
670
        """
605
671
        local_warnings = []
606
 
        def capture_warnings(msg, cls, stacklevel=None):
 
672
        def capture_warnings(msg, cls=None, stacklevel=None):
607
673
            # we've hooked into a deprecation specific callpath,
608
674
            # only deprecations should getting sent via it.
609
675
            self.assertEqual(cls, DeprecationWarning)
837
903
            os.chdir(working_dir)
838
904
 
839
905
        try:
840
 
            result = self.apply_redirected(stdin, stdout, stderr,
841
 
                                           bzrlib.commands.run_bzr_catch_errors,
842
 
                                           argv)
 
906
            saved_debug_flags = frozenset(debug.debug_flags)
 
907
            debug.debug_flags.clear()
 
908
            try:
 
909
                result = self.apply_redirected(stdin, stdout, stderr,
 
910
                                               bzrlib.commands.run_bzr_catch_errors,
 
911
                                               argv)
 
912
            finally:
 
913
                debug.debug_flags.update(saved_debug_flags)
843
914
        finally:
844
915
            logger.removeHandler(handler)
845
916
            bzrlib.ui.ui_factory = old_ui_factory
1511
1582
    def setUp(self):
1512
1583
        super(TestCaseWithTransport, self).setUp()
1513
1584
        self.__server = None
1514
 
        self.transport_server = default_transport
1515
1585
 
1516
1586
 
1517
1587
class ChrootedTestCase(TestCaseWithTransport):
1547
1617
    TestCase._gather_lsprof_in_benchmarks = lsprof_timed
1548
1618
    if verbose:
1549
1619
        verbosity = 2
1550
 
        pb = None
1551
1620
    else:
1552
1621
        verbosity = 1
1553
 
        pb = progress.ProgressBar()
1554
1622
    runner = TextTestRunner(stream=sys.stdout,
1555
1623
                            descriptions=0,
1556
1624
                            verbosity=verbosity,
1557
1625
                            keep_output=keep_output,
1558
 
                            pb=pb,
1559
1626
                            bench_history=bench_history)
1560
1627
    runner.stop_on_failure=stop_on_failure
1561
1628
    if pattern != '.*':