~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/__init__.py

  • Committer: mbp at sourcefrog
  • Date: 2005-03-22 00:37:17 UTC
  • Revision ID: mbp@sourcefrog.net-20050322003717-7b1d4a370c237846
lift out tracefile creation code

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2005, 2006, 2007 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
 
# TODO: Perhaps there should be an API to find out if bzr running under the
19
 
# test suite -- some plugins might want to avoid making intrusive changes if
20
 
# this is the case.  However, we want behaviour under to test to diverge as
21
 
# little as possible, so this should be used rarely if it's added at all.
22
 
# (Suggestion from j-a-meinel, 2005-11-24)
23
 
 
24
 
# NOTE: Some classes in here use camelCaseNaming() rather than
25
 
# underscore_naming().  That's for consistency with unittest; it's not the
26
 
# general style of bzrlib.  Please continue that consistency when adding e.g.
27
 
# new assertFoo() methods.
28
 
 
29
 
import codecs
30
 
from cStringIO import StringIO
31
 
import difflib
32
 
import doctest
33
 
import errno
34
 
import logging
35
 
import os
36
 
from pprint import pformat
37
 
import re
38
 
import shlex
39
 
import stat
40
 
from subprocess import Popen, PIPE
41
 
import sys
42
 
import tempfile
43
 
import unittest
44
 
import time
45
 
 
46
 
 
47
 
from bzrlib import (
48
 
    bzrdir,
49
 
    debug,
50
 
    errors,
51
 
    memorytree,
52
 
    osutils,
53
 
    progress,
54
 
    ui,
55
 
    urlutils,
56
 
    )
57
 
import bzrlib.branch
58
 
import bzrlib.commands
59
 
import bzrlib.timestamp
60
 
import bzrlib.export
61
 
import bzrlib.inventory
62
 
import bzrlib.iterablefile
63
 
import bzrlib.lockdir
64
 
try:
65
 
    import bzrlib.lsprof
66
 
except ImportError:
67
 
    # lsprof not available
68
 
    pass
69
 
from bzrlib.merge import merge_inner
70
 
import bzrlib.merge3
71
 
import bzrlib.osutils
72
 
import bzrlib.plugin
73
 
from bzrlib.revision import common_ancestor
74
 
import bzrlib.store
75
 
from bzrlib import symbol_versioning
76
 
import bzrlib.trace
77
 
from bzrlib.transport import get_transport
78
 
import bzrlib.transport
79
 
from bzrlib.transport.local import LocalURLServer
80
 
from bzrlib.transport.memory import MemoryServer
81
 
from bzrlib.transport.readonly import ReadonlyServer
82
 
from bzrlib.trace import mutter, note
83
 
from bzrlib.tests import TestUtil
84
 
from bzrlib.tests.HttpServer import HttpServer
85
 
from bzrlib.tests.TestUtil import (
86
 
                          TestSuite,
87
 
                          TestLoader,
88
 
                          )
89
 
from bzrlib.tests.treeshape import build_tree_contents
90
 
from bzrlib.workingtree import WorkingTree, WorkingTreeFormat2
91
 
 
92
 
default_transport = LocalURLServer
93
 
 
94
 
MODULES_TO_TEST = []
95
 
MODULES_TO_DOCTEST = [
96
 
                      bzrlib.timestamp,
97
 
                      bzrlib.errors,
98
 
                      bzrlib.export,
99
 
                      bzrlib.inventory,
100
 
                      bzrlib.iterablefile,
101
 
                      bzrlib.lockdir,
102
 
                      bzrlib.merge3,
103
 
                      bzrlib.option,
104
 
                      bzrlib.store,
105
 
                      ]
106
 
 
107
 
NUMBERED_DIRS = False   # dirs kind for TestCaseInTempDir (numbered or named)
108
 
 
109
 
 
110
 
def packages_to_test():
111
 
    """Return a list of packages to test.
112
 
 
113
 
    The packages are not globally imported so that import failures are
114
 
    triggered when running selftest, not when importing the command.
115
 
    """
116
 
    import bzrlib.doc
117
 
    import bzrlib.tests.blackbox
118
 
    import bzrlib.tests.branch_implementations
119
 
    import bzrlib.tests.bzrdir_implementations
120
 
    import bzrlib.tests.interrepository_implementations
121
 
    import bzrlib.tests.interversionedfile_implementations
122
 
    import bzrlib.tests.intertree_implementations
123
 
    import bzrlib.tests.per_lock
124
 
    import bzrlib.tests.repository_implementations
125
 
    import bzrlib.tests.revisionstore_implementations
126
 
    import bzrlib.tests.tree_implementations
127
 
    import bzrlib.tests.workingtree_implementations
128
 
    return [
129
 
            bzrlib.doc,
130
 
            bzrlib.tests.blackbox,
131
 
            bzrlib.tests.branch_implementations,
132
 
            bzrlib.tests.bzrdir_implementations,
133
 
            bzrlib.tests.interrepository_implementations,
134
 
            bzrlib.tests.interversionedfile_implementations,
135
 
            bzrlib.tests.intertree_implementations,
136
 
            bzrlib.tests.per_lock,
137
 
            bzrlib.tests.repository_implementations,
138
 
            bzrlib.tests.revisionstore_implementations,
139
 
            bzrlib.tests.tree_implementations,
140
 
            bzrlib.tests.workingtree_implementations,
141
 
            ]
142
 
 
143
 
 
144
 
class ExtendedTestResult(unittest._TextTestResult):
145
 
    """Accepts, reports and accumulates the results of running tests.
146
 
 
147
 
    Compared to this unittest version this class adds support for profiling,
148
 
    benchmarking, stopping as soon as a test fails,  and skipping tests.
149
 
    There are further-specialized subclasses for different types of display.
150
 
    """
151
 
 
152
 
    stop_early = False
153
 
    
154
 
    def __init__(self, stream, descriptions, verbosity,
155
 
                 bench_history=None,
156
 
                 num_tests=None,
157
 
                 ):
158
 
        """Construct new TestResult.
159
 
 
160
 
        :param bench_history: Optionally, a writable file object to accumulate
161
 
            benchmark results.
162
 
        """
163
 
        unittest._TextTestResult.__init__(self, stream, descriptions, verbosity)
164
 
        if bench_history is not None:
165
 
            from bzrlib.version import _get_bzr_source_tree
166
 
            src_tree = _get_bzr_source_tree()
167
 
            if src_tree:
168
 
                try:
169
 
                    revision_id = src_tree.get_parent_ids()[0]
170
 
                except IndexError:
171
 
                    # XXX: if this is a brand new tree, do the same as if there
172
 
                    # is no branch.
173
 
                    revision_id = ''
174
 
            else:
175
 
                # XXX: If there's no branch, what should we do?
176
 
                revision_id = ''
177
 
            bench_history.write("--date %s %s\n" % (time.time(), revision_id))
178
 
        self._bench_history = bench_history
179
 
        self.ui = ui.ui_factory
180
 
        self.num_tests = num_tests
181
 
        self.error_count = 0
182
 
        self.failure_count = 0
183
 
        self.known_failure_count = 0
184
 
        self.skip_count = 0
185
 
        self.unsupported = {}
186
 
        self.count = 0
187
 
        self._overall_start_time = time.time()
188
 
    
189
 
    def extractBenchmarkTime(self, testCase):
190
 
        """Add a benchmark time for the current test case."""
191
 
        self._benchmarkTime = getattr(testCase, "_benchtime", None)
192
 
    
193
 
    def _elapsedTestTimeString(self):
194
 
        """Return a time string for the overall time the current test has taken."""
195
 
        return self._formatTime(time.time() - self._start_time)
196
 
 
197
 
    def _testTimeString(self):
198
 
        if self._benchmarkTime is not None:
199
 
            return "%s/%s" % (
200
 
                self._formatTime(self._benchmarkTime),
201
 
                self._elapsedTestTimeString())
202
 
        else:
203
 
            return "           %s" % self._elapsedTestTimeString()
204
 
 
205
 
    def _formatTime(self, seconds):
206
 
        """Format seconds as milliseconds with leading spaces."""
207
 
        # some benchmarks can take thousands of seconds to run, so we need 8
208
 
        # places
209
 
        return "%8dms" % (1000 * seconds)
210
 
 
211
 
    def _shortened_test_description(self, test):
212
 
        what = test.id()
213
 
        what = re.sub(r'^bzrlib\.(tests|benchmarks)\.', '', what)
214
 
        return what
215
 
 
216
 
    def startTest(self, test):
217
 
        unittest.TestResult.startTest(self, test)
218
 
        self.report_test_start(test)
219
 
        test.number = self.count
220
 
        self._recordTestStartTime()
221
 
 
222
 
    def _recordTestStartTime(self):
223
 
        """Record that a test has started."""
224
 
        self._start_time = time.time()
225
 
 
226
 
    def _cleanupLogFile(self, test):
227
 
        # We can only do this if we have one of our TestCases, not if
228
 
        # we have a doctest.
229
 
        setKeepLogfile = getattr(test, 'setKeepLogfile', None)
230
 
        if setKeepLogfile is not None:
231
 
            setKeepLogfile()
232
 
 
233
 
    def addError(self, test, err):
234
 
        self.extractBenchmarkTime(test)
235
 
        self._cleanupLogFile(test)
236
 
        if isinstance(err[1], TestSkipped):
237
 
            return self.addSkipped(test, err)
238
 
        elif isinstance(err[1], UnavailableFeature):
239
 
            return self.addNotSupported(test, err[1].args[0])
240
 
        unittest.TestResult.addError(self, test, err)
241
 
        self.error_count += 1
242
 
        self.report_error(test, err)
243
 
        if self.stop_early:
244
 
            self.stop()
245
 
 
246
 
    def addFailure(self, test, err):
247
 
        self._cleanupLogFile(test)
248
 
        self.extractBenchmarkTime(test)
249
 
        if isinstance(err[1], KnownFailure):
250
 
            return self.addKnownFailure(test, err)
251
 
        unittest.TestResult.addFailure(self, test, err)
252
 
        self.failure_count += 1
253
 
        self.report_failure(test, err)
254
 
        if self.stop_early:
255
 
            self.stop()
256
 
 
257
 
    def addKnownFailure(self, test, err):
258
 
        self.known_failure_count += 1
259
 
        self.report_known_failure(test, err)
260
 
 
261
 
    def addNotSupported(self, test, feature):
262
 
        self.unsupported.setdefault(str(feature), 0)
263
 
        self.unsupported[str(feature)] += 1
264
 
        self.report_unsupported(test, feature)
265
 
 
266
 
    def addSuccess(self, test):
267
 
        self.extractBenchmarkTime(test)
268
 
        if self._bench_history is not None:
269
 
            if self._benchmarkTime is not None:
270
 
                self._bench_history.write("%s %s\n" % (
271
 
                    self._formatTime(self._benchmarkTime),
272
 
                    test.id()))
273
 
        self.report_success(test)
274
 
        unittest.TestResult.addSuccess(self, test)
275
 
 
276
 
    def addSkipped(self, test, skip_excinfo):
277
 
        self.report_skip(test, skip_excinfo)
278
 
        # seems best to treat this as success from point-of-view of unittest
279
 
        # -- it actually does nothing so it barely matters :)
280
 
        try:
281
 
            test.tearDown()
282
 
        except KeyboardInterrupt:
283
 
            raise
284
 
        except:
285
 
            self.addError(test, test.__exc_info())
286
 
        else:
287
 
            unittest.TestResult.addSuccess(self, test)
288
 
 
289
 
    def printErrorList(self, flavour, errors):
290
 
        for test, err in errors:
291
 
            self.stream.writeln(self.separator1)
292
 
            self.stream.write("%s: " % flavour)
293
 
            if NUMBERED_DIRS:
294
 
                self.stream.write('#%d ' % test.number)
295
 
            self.stream.writeln(self.getDescription(test))
296
 
            if getattr(test, '_get_log', None) is not None:
297
 
                print >>self.stream
298
 
                print >>self.stream, \
299
 
                        ('vvvv[log from %s]' % test.id()).ljust(78,'-')
300
 
                print >>self.stream, test._get_log()
301
 
                print >>self.stream, \
302
 
                        ('^^^^[log from %s]' % test.id()).ljust(78,'-')
303
 
            self.stream.writeln(self.separator2)
304
 
            self.stream.writeln("%s" % err)
305
 
 
306
 
    def finished(self):
307
 
        pass
308
 
 
309
 
    def report_cleaning_up(self):
310
 
        pass
311
 
 
312
 
    def report_success(self, test):
313
 
        pass
314
 
 
315
 
 
316
 
class TextTestResult(ExtendedTestResult):
317
 
    """Displays progress and results of tests in text form"""
318
 
 
319
 
    def __init__(self, stream, descriptions, verbosity,
320
 
                 bench_history=None,
321
 
                 num_tests=None,
322
 
                 pb=None,
323
 
                 ):
324
 
        ExtendedTestResult.__init__(self, stream, descriptions, verbosity,
325
 
            bench_history, num_tests)
326
 
        if pb is None:
327
 
            self.pb = self.ui.nested_progress_bar()
328
 
            self._supplied_pb = False
329
 
        else:
330
 
            self.pb = pb
331
 
            self._supplied_pb = True
332
 
        self.pb.show_pct = False
333
 
        self.pb.show_spinner = False
334
 
        self.pb.show_eta = False,
335
 
        self.pb.show_count = False
336
 
        self.pb.show_bar = False
337
 
 
338
 
    def report_starting(self):
339
 
        self.pb.update('[test 0/%d] starting...' % (self.num_tests))
340
 
 
341
 
    def _progress_prefix_text(self):
342
 
        a = '[%d' % self.count
343
 
        if self.num_tests is not None:
344
 
            a +='/%d' % self.num_tests
345
 
        a += ' in %ds' % (time.time() - self._overall_start_time)
346
 
        if self.error_count:
347
 
            a += ', %d errors' % self.error_count
348
 
        if self.failure_count:
349
 
            a += ', %d failed' % self.failure_count
350
 
        if self.known_failure_count:
351
 
            a += ', %d known failures' % self.known_failure_count
352
 
        if self.skip_count:
353
 
            a += ', %d skipped' % self.skip_count
354
 
        if self.unsupported:
355
 
            a += ', %d missing features' % len(self.unsupported)
356
 
        a += ']'
357
 
        return a
358
 
 
359
 
    def report_test_start(self, test):
360
 
        self.count += 1
361
 
        self.pb.update(
362
 
                self._progress_prefix_text()
363
 
                + ' ' 
364
 
                + self._shortened_test_description(test))
365
 
 
366
 
    def _test_description(self, test):
367
 
        if NUMBERED_DIRS:
368
 
            return '#%d %s' % (self.count,
369
 
                               self._shortened_test_description(test))
370
 
        else:
371
 
            return self._shortened_test_description(test)
372
 
 
373
 
    def report_error(self, test, err):
374
 
        self.pb.note('ERROR: %s\n    %s\n', 
375
 
            self._test_description(test),
376
 
            err[1],
377
 
            )
378
 
 
379
 
    def report_failure(self, test, err):
380
 
        self.pb.note('FAIL: %s\n    %s\n', 
381
 
            self._test_description(test),
382
 
            err[1],
383
 
            )
384
 
 
385
 
    def report_known_failure(self, test, err):
386
 
        self.pb.note('XFAIL: %s\n%s\n',
387
 
            self._test_description(test), err[1])
388
 
 
389
 
    def report_skip(self, test, skip_excinfo):
390
 
        self.skip_count += 1
391
 
        if False:
392
 
            # at the moment these are mostly not things we can fix
393
 
            # and so they just produce stipple; use the verbose reporter
394
 
            # to see them.
395
 
            if False:
396
 
                # show test and reason for skip
397
 
                self.pb.note('SKIP: %s\n    %s\n', 
398
 
                    self._shortened_test_description(test),
399
 
                    skip_excinfo[1])
400
 
            else:
401
 
                # since the class name was left behind in the still-visible
402
 
                # progress bar...
403
 
                self.pb.note('SKIP: %s', skip_excinfo[1])
404
 
 
405
 
    def report_unsupported(self, test, feature):
406
 
        """test cannot be run because feature is missing."""
407
 
                  
408
 
    def report_cleaning_up(self):
409
 
        self.pb.update('cleaning up...')
410
 
 
411
 
    def finished(self):
412
 
        if not self._supplied_pb:
413
 
            self.pb.finished()
414
 
 
415
 
 
416
 
class VerboseTestResult(ExtendedTestResult):
417
 
    """Produce long output, with one line per test run plus times"""
418
 
 
419
 
    def _ellipsize_to_right(self, a_string, final_width):
420
 
        """Truncate and pad a string, keeping the right hand side"""
421
 
        if len(a_string) > final_width:
422
 
            result = '...' + a_string[3-final_width:]
423
 
        else:
424
 
            result = a_string
425
 
        return result.ljust(final_width)
426
 
 
427
 
    def report_starting(self):
428
 
        self.stream.write('running %d tests...\n' % self.num_tests)
429
 
 
430
 
    def report_test_start(self, test):
431
 
        self.count += 1
432
 
        name = self._shortened_test_description(test)
433
 
        # width needs space for 6 char status, plus 1 for slash, plus 2 10-char
434
 
        # numbers, plus a trailing blank
435
 
        # when NUMBERED_DIRS: plus 5 chars on test number, plus 1 char on space
436
 
        if NUMBERED_DIRS:
437
 
            self.stream.write('%5d ' % self.count)
438
 
            self.stream.write(self._ellipsize_to_right(name,
439
 
                                osutils.terminal_width()-36))
440
 
        else:
441
 
            self.stream.write(self._ellipsize_to_right(name,
442
 
                                osutils.terminal_width()-30))
443
 
        self.stream.flush()
444
 
 
445
 
    def _error_summary(self, err):
446
 
        indent = ' ' * 4
447
 
        if NUMBERED_DIRS:
448
 
            indent += ' ' * 6
449
 
        return '%s%s' % (indent, err[1])
450
 
 
451
 
    def report_error(self, test, err):
452
 
        self.stream.writeln('ERROR %s\n%s'
453
 
                % (self._testTimeString(),
454
 
                   self._error_summary(err)))
455
 
 
456
 
    def report_failure(self, test, err):
457
 
        self.stream.writeln(' FAIL %s\n%s'
458
 
                % (self._testTimeString(),
459
 
                   self._error_summary(err)))
460
 
 
461
 
    def report_known_failure(self, test, err):
462
 
        self.stream.writeln('XFAIL %s\n%s'
463
 
                % (self._testTimeString(),
464
 
                   self._error_summary(err)))
465
 
 
466
 
    def report_success(self, test):
467
 
        self.stream.writeln('   OK %s' % self._testTimeString())
468
 
        for bench_called, stats in getattr(test, '_benchcalls', []):
469
 
            self.stream.writeln('LSProf output for %s(%s, %s)' % bench_called)
470
 
            stats.pprint(file=self.stream)
471
 
        # flush the stream so that we get smooth output. This verbose mode is
472
 
        # used to show the output in PQM.
473
 
        self.stream.flush()
474
 
 
475
 
    def report_skip(self, test, skip_excinfo):
476
 
        self.skip_count += 1
477
 
        self.stream.writeln(' SKIP %s\n%s'
478
 
                % (self._testTimeString(),
479
 
                   self._error_summary(skip_excinfo)))
480
 
 
481
 
    def report_unsupported(self, test, feature):
482
 
        """test cannot be run because feature is missing."""
483
 
        self.stream.writeln("NODEP %s\n    The feature '%s' is not available."
484
 
                %(self._testTimeString(), feature))
485
 
                  
486
 
 
487
 
 
488
 
class TextTestRunner(object):
489
 
    stop_on_failure = False
490
 
 
491
 
    def __init__(self,
492
 
                 stream=sys.stderr,
493
 
                 descriptions=0,
494
 
                 verbosity=1,
495
 
                 keep_output=False,
496
 
                 bench_history=None):
497
 
        self.stream = unittest._WritelnDecorator(stream)
498
 
        self.descriptions = descriptions
499
 
        self.verbosity = verbosity
500
 
        self.keep_output = keep_output
501
 
        self._bench_history = bench_history
502
 
 
503
 
    def run(self, test):
504
 
        "Run the given test case or test suite."
505
 
        startTime = time.time()
506
 
        if self.verbosity == 1:
507
 
            result_class = TextTestResult
508
 
        elif self.verbosity >= 2:
509
 
            result_class = VerboseTestResult
510
 
        result = result_class(self.stream,
511
 
                              self.descriptions,
512
 
                              self.verbosity,
513
 
                              bench_history=self._bench_history,
514
 
                              num_tests=test.countTestCases(),
515
 
                              )
516
 
        result.stop_early = self.stop_on_failure
517
 
        result.report_starting()
518
 
        test.run(result)
519
 
        stopTime = time.time()
520
 
        timeTaken = stopTime - startTime
521
 
        result.printErrors()
522
 
        self.stream.writeln(result.separator2)
523
 
        run = result.testsRun
524
 
        self.stream.writeln("Ran %d test%s in %.3fs" %
525
 
                            (run, run != 1 and "s" or "", timeTaken))
526
 
        self.stream.writeln()
527
 
        if not result.wasSuccessful():
528
 
            self.stream.write("FAILED (")
529
 
            failed, errored = map(len, (result.failures, result.errors))
530
 
            if failed:
531
 
                self.stream.write("failures=%d" % failed)
532
 
            if errored:
533
 
                if failed: self.stream.write(", ")
534
 
                self.stream.write("errors=%d" % errored)
535
 
            if result.known_failure_count:
536
 
                if failed or errored: self.stream.write(", ")
537
 
                self.stream.write("known_failure_count=%d" %
538
 
                    result.known_failure_count)
539
 
            self.stream.writeln(")")
540
 
        else:
541
 
            if result.known_failure_count:
542
 
                self.stream.writeln("OK (known_failures=%d)" %
543
 
                    result.known_failure_count)
544
 
            else:
545
 
                self.stream.writeln("OK")
546
 
        if result.skip_count > 0:
547
 
            skipped = result.skip_count
548
 
            self.stream.writeln('%d test%s skipped' %
549
 
                                (skipped, skipped != 1 and "s" or ""))
550
 
        if result.unsupported:
551
 
            for feature, count in sorted(result.unsupported.items()):
552
 
                self.stream.writeln("Missing feature '%s' skipped %d tests." %
553
 
                    (feature, count))
554
 
        result.report_cleaning_up()
555
 
        # This is still a little bogus, 
556
 
        # but only a little. Folk not using our testrunner will
557
 
        # have to delete their temp directories themselves.
558
 
        test_root = TestCaseWithMemoryTransport.TEST_ROOT
559
 
        if result.wasSuccessful() or not self.keep_output:
560
 
            if test_root is not None:
561
 
                # If LANG=C we probably have created some bogus paths
562
 
                # which rmtree(unicode) will fail to delete
563
 
                # so make sure we are using rmtree(str) to delete everything
564
 
                # except on win32, where rmtree(str) will fail
565
 
                # since it doesn't have the property of byte-stream paths
566
 
                # (they are either ascii or mbcs)
567
 
                if sys.platform == 'win32':
568
 
                    # make sure we are using the unicode win32 api
569
 
                    test_root = unicode(test_root)
570
 
                else:
571
 
                    test_root = test_root.encode(
572
 
                        sys.getfilesystemencoding())
573
 
                try:
574
 
                    osutils.rmtree(test_root)
575
 
                except OSError, e:
576
 
                    if sys.platform == 'win32' and e.errno == errno.EACCES:
577
 
                        print >>sys.stderr, ('Permission denied: '
578
 
                                             'unable to remove testing dir '
579
 
                                             '%s' % os.path.basename(test_root))
580
 
                    else:
581
 
                        raise
582
 
        else:
583
 
            note("Failed tests working directories are in '%s'\n", test_root)
584
 
        TestCaseWithMemoryTransport.TEST_ROOT = None
585
 
        result.finished()
586
 
        return result
587
 
 
588
 
 
589
 
def iter_suite_tests(suite):
590
 
    """Return all tests in a suite, recursing through nested suites"""
591
 
    for item in suite._tests:
592
 
        if isinstance(item, unittest.TestCase):
593
 
            yield item
594
 
        elif isinstance(item, unittest.TestSuite):
595
 
            for r in iter_suite_tests(item):
596
 
                yield r
597
 
        else:
598
 
            raise Exception('unknown object %r inside test suite %r'
599
 
                            % (item, suite))
600
 
 
601
 
 
602
 
class TestSkipped(Exception):
603
 
    """Indicates that a test was intentionally skipped, rather than failing."""
604
 
 
605
 
 
606
 
class KnownFailure(AssertionError):
607
 
    """Indicates that a test failed in a precisely expected manner.
608
 
 
609
 
    Such failures dont block the whole test suite from passing because they are
610
 
    indicators of partially completed code or of future work. We have an
611
 
    explicit error for them so that we can ensure that they are always visible:
612
 
    KnownFailures are always shown in the output of bzr selftest.
613
 
    """
614
 
 
615
 
 
616
 
class UnavailableFeature(Exception):
617
 
    """A feature required for this test was not available.
618
 
 
619
 
    The feature should be used to construct the exception.
620
 
    """
621
 
 
622
 
 
623
 
class CommandFailed(Exception):
624
 
    pass
625
 
 
626
 
 
627
 
class StringIOWrapper(object):
628
 
    """A wrapper around cStringIO which just adds an encoding attribute.
629
 
    
630
 
    Internally we can check sys.stdout to see what the output encoding
631
 
    should be. However, cStringIO has no encoding attribute that we can
632
 
    set. So we wrap it instead.
633
 
    """
634
 
    encoding='ascii'
635
 
    _cstring = None
636
 
 
637
 
    def __init__(self, s=None):
638
 
        if s is not None:
639
 
            self.__dict__['_cstring'] = StringIO(s)
640
 
        else:
641
 
            self.__dict__['_cstring'] = StringIO()
642
 
 
643
 
    def __getattr__(self, name, getattr=getattr):
644
 
        return getattr(self.__dict__['_cstring'], name)
645
 
 
646
 
    def __setattr__(self, name, val):
647
 
        if name == 'encoding':
648
 
            self.__dict__['encoding'] = val
649
 
        else:
650
 
            return setattr(self._cstring, name, val)
651
 
 
652
 
 
653
 
class TestUIFactory(ui.CLIUIFactory):
654
 
    """A UI Factory for testing.
655
 
 
656
 
    Hide the progress bar but emit note()s.
657
 
    Redirect stdin.
658
 
    Allows get_password to be tested without real tty attached.
659
 
    """
660
 
 
661
 
    def __init__(self,
662
 
                 stdout=None,
663
 
                 stderr=None,
664
 
                 stdin=None):
665
 
        super(TestUIFactory, self).__init__()
666
 
        if stdin is not None:
667
 
            # We use a StringIOWrapper to be able to test various
668
 
            # encodings, but the user is still responsible to
669
 
            # encode the string and to set the encoding attribute
670
 
            # of StringIOWrapper.
671
 
            self.stdin = StringIOWrapper(stdin)
672
 
        if stdout is None:
673
 
            self.stdout = sys.stdout
674
 
        else:
675
 
            self.stdout = stdout
676
 
        if stderr is None:
677
 
            self.stderr = sys.stderr
678
 
        else:
679
 
            self.stderr = stderr
680
 
 
681
 
    def clear(self):
682
 
        """See progress.ProgressBar.clear()."""
683
 
 
684
 
    def clear_term(self):
685
 
        """See progress.ProgressBar.clear_term()."""
686
 
 
687
 
    def clear_term(self):
688
 
        """See progress.ProgressBar.clear_term()."""
689
 
 
690
 
    def finished(self):
691
 
        """See progress.ProgressBar.finished()."""
692
 
 
693
 
    def note(self, fmt_string, *args, **kwargs):
694
 
        """See progress.ProgressBar.note()."""
695
 
        self.stdout.write((fmt_string + "\n") % args)
696
 
 
697
 
    def progress_bar(self):
698
 
        return self
699
 
 
700
 
    def nested_progress_bar(self):
701
 
        return self
702
 
 
703
 
    def update(self, message, count=None, total=None):
704
 
        """See progress.ProgressBar.update()."""
705
 
 
706
 
    def get_non_echoed_password(self, prompt):
707
 
        """Get password from stdin without trying to handle the echo mode"""
708
 
        if prompt:
709
 
            self.stdout.write(prompt)
710
 
        password = self.stdin.readline()
711
 
        if not password:
712
 
            raise EOFError
713
 
        if password[-1] == '\n':
714
 
            password = password[:-1]
715
 
        return password
716
 
 
717
 
 
718
 
class TestCase(unittest.TestCase):
719
 
    """Base class for bzr unit tests.
720
 
    
721
 
    Tests that need access to disk resources should subclass 
722
 
    TestCaseInTempDir not TestCase.
723
 
 
724
 
    Error and debug log messages are redirected from their usual
725
 
    location into a temporary file, the contents of which can be
726
 
    retrieved by _get_log().  We use a real OS file, not an in-memory object,
727
 
    so that it can also capture file IO.  When the test completes this file
728
 
    is read into memory and removed from disk.
729
 
       
730
 
    There are also convenience functions to invoke bzr's command-line
731
 
    routine, and to build and check bzr trees.
732
 
   
733
 
    In addition to the usual method of overriding tearDown(), this class also
734
 
    allows subclasses to register functions into the _cleanups list, which is
735
 
    run in order as the object is torn down.  It's less likely this will be
736
 
    accidentally overlooked.
737
 
    """
738
 
 
739
 
    _log_file_name = None
740
 
    _log_contents = ''
741
 
    _keep_log_file = False
742
 
    # record lsprof data when performing benchmark calls.
743
 
    _gather_lsprof_in_benchmarks = False
744
 
 
745
 
    def __init__(self, methodName='testMethod'):
746
 
        super(TestCase, self).__init__(methodName)
747
 
        self._cleanups = []
748
 
 
749
 
    def setUp(self):
750
 
        unittest.TestCase.setUp(self)
751
 
        self._cleanEnvironment()
752
 
        bzrlib.trace.disable_default_logging()
753
 
        self._silenceUI()
754
 
        self._startLogFile()
755
 
        self._benchcalls = []
756
 
        self._benchtime = None
757
 
        # prevent hooks affecting tests
758
 
        self._preserved_hooks = bzrlib.branch.Branch.hooks
759
 
        self.addCleanup(self._restoreHooks)
760
 
        # this list of hooks must be kept in sync with the defaults
761
 
        # in branch.py
762
 
        bzrlib.branch.Branch.hooks = bzrlib.branch.BranchHooks()
763
 
 
764
 
    def _silenceUI(self):
765
 
        """Turn off UI for duration of test"""
766
 
        # by default the UI is off; tests can turn it on if they want it.
767
 
        saved = ui.ui_factory
768
 
        def _restore():
769
 
            ui.ui_factory = saved
770
 
        ui.ui_factory = ui.SilentUIFactory()
771
 
        self.addCleanup(_restore)
772
 
 
773
 
    def _ndiff_strings(self, a, b):
774
 
        """Return ndiff between two strings containing lines.
775
 
        
776
 
        A trailing newline is added if missing to make the strings
777
 
        print properly."""
778
 
        if b and b[-1] != '\n':
779
 
            b += '\n'
780
 
        if a and a[-1] != '\n':
781
 
            a += '\n'
782
 
        difflines = difflib.ndiff(a.splitlines(True),
783
 
                                  b.splitlines(True),
784
 
                                  linejunk=lambda x: False,
785
 
                                  charjunk=lambda x: False)
786
 
        return ''.join(difflines)
787
 
 
788
 
    def assertEqual(self, a, b, message=''):
789
 
        try:
790
 
            if a == b:
791
 
                return
792
 
        except UnicodeError, e:
793
 
            # If we can't compare without getting a UnicodeError, then
794
 
            # obviously they are different
795
 
            mutter('UnicodeError: %s', e)
796
 
        if message:
797
 
            message += '\n'
798
 
        raise AssertionError("%snot equal:\na = %s\nb = %s\n"
799
 
            % (message,
800
 
               pformat(a, indent=4), pformat(b, indent=4)))
801
 
 
802
 
    assertEquals = assertEqual
803
 
 
804
 
    def assertEqualDiff(self, a, b, message=None):
805
 
        """Assert two texts are equal, if not raise an exception.
806
 
        
807
 
        This is intended for use with multi-line strings where it can 
808
 
        be hard to find the differences by eye.
809
 
        """
810
 
        # TODO: perhaps override assertEquals to call this for strings?
811
 
        if a == b:
812
 
            return
813
 
        if message is None:
814
 
            message = "texts not equal:\n"
815
 
        raise AssertionError(message + 
816
 
                             self._ndiff_strings(a, b))      
817
 
        
818
 
    def assertEqualMode(self, mode, mode_test):
819
 
        self.assertEqual(mode, mode_test,
820
 
                         'mode mismatch %o != %o' % (mode, mode_test))
821
 
 
822
 
    def assertStartsWith(self, s, prefix):
823
 
        if not s.startswith(prefix):
824
 
            raise AssertionError('string %r does not start with %r' % (s, prefix))
825
 
 
826
 
    def assertEndsWith(self, s, suffix):
827
 
        """Asserts that s ends with suffix."""
828
 
        if not s.endswith(suffix):
829
 
            raise AssertionError('string %r does not end with %r' % (s, suffix))
830
 
 
831
 
    def assertContainsRe(self, haystack, needle_re):
832
 
        """Assert that a contains something matching a regular expression."""
833
 
        if not re.search(needle_re, haystack):
834
 
            raise AssertionError('pattern "%r" not found in "%r"'
835
 
                    % (needle_re, haystack))
836
 
 
837
 
    def assertNotContainsRe(self, haystack, needle_re):
838
 
        """Assert that a does not match a regular expression"""
839
 
        if re.search(needle_re, haystack):
840
 
            raise AssertionError('pattern "%s" found in "%s"'
841
 
                    % (needle_re, haystack))
842
 
 
843
 
    def assertSubset(self, sublist, superlist):
844
 
        """Assert that every entry in sublist is present in superlist."""
845
 
        missing = []
846
 
        for entry in sublist:
847
 
            if entry not in superlist:
848
 
                missing.append(entry)
849
 
        if len(missing) > 0:
850
 
            raise AssertionError("value(s) %r not present in container %r" % 
851
 
                                 (missing, superlist))
852
 
 
853
 
    def assertListRaises(self, excClass, func, *args, **kwargs):
854
 
        """Fail unless excClass is raised when the iterator from func is used.
855
 
        
856
 
        Many functions can return generators this makes sure
857
 
        to wrap them in a list() call to make sure the whole generator
858
 
        is run, and that the proper exception is raised.
859
 
        """
860
 
        try:
861
 
            list(func(*args, **kwargs))
862
 
        except excClass:
863
 
            return
864
 
        else:
865
 
            if getattr(excClass,'__name__', None) is not None:
866
 
                excName = excClass.__name__
867
 
            else:
868
 
                excName = str(excClass)
869
 
            raise self.failureException, "%s not raised" % excName
870
 
 
871
 
    def assertRaises(self, excClass, func, *args, **kwargs):
872
 
        """Assert that a callable raises a particular exception.
873
 
 
874
 
        :param excClass: As for the except statement, this may be either an
875
 
        exception class, or a tuple of classes.
876
 
 
877
 
        Returns the exception so that you can examine it.
878
 
        """
879
 
        try:
880
 
            func(*args, **kwargs)
881
 
        except excClass, e:
882
 
            return e
883
 
        else:
884
 
            if getattr(excClass,'__name__', None) is not None:
885
 
                excName = excClass.__name__
886
 
            else:
887
 
                # probably a tuple
888
 
                excName = str(excClass)
889
 
            raise self.failureException, "%s not raised" % excName
890
 
 
891
 
    def assertIs(self, left, right, message=None):
892
 
        if not (left is right):
893
 
            if message is not None:
894
 
                raise AssertionError(message)
895
 
            else:
896
 
                raise AssertionError("%r is not %r." % (left, right))
897
 
 
898
 
    def assertIsNot(self, left, right, message=None):
899
 
        if (left is right):
900
 
            if message is not None:
901
 
                raise AssertionError(message)
902
 
            else:
903
 
                raise AssertionError("%r is %r." % (left, right))
904
 
 
905
 
    def assertTransportMode(self, transport, path, mode):
906
 
        """Fail if a path does not have mode mode.
907
 
        
908
 
        If modes are not supported on this transport, the assertion is ignored.
909
 
        """
910
 
        if not transport._can_roundtrip_unix_modebits():
911
 
            return
912
 
        path_stat = transport.stat(path)
913
 
        actual_mode = stat.S_IMODE(path_stat.st_mode)
914
 
        self.assertEqual(mode, actual_mode,
915
 
            'mode of %r incorrect (%o != %o)' % (path, mode, actual_mode))
916
 
 
917
 
    def assertIsInstance(self, obj, kls):
918
 
        """Fail if obj is not an instance of kls"""
919
 
        if not isinstance(obj, kls):
920
 
            self.fail("%r is an instance of %s rather than %s" % (
921
 
                obj, obj.__class__, kls))
922
 
 
923
 
    def _capture_warnings(self, a_callable, *args, **kwargs):
924
 
        """A helper for callDeprecated and applyDeprecated.
925
 
 
926
 
        :param a_callable: A callable to call.
927
 
        :param args: The positional arguments for the callable
928
 
        :param kwargs: The keyword arguments for the callable
929
 
        :return: A tuple (warnings, result). result is the result of calling
930
 
            a_callable(*args, **kwargs).
931
 
        """
932
 
        local_warnings = []
933
 
        def capture_warnings(msg, cls=None, stacklevel=None):
934
 
            # we've hooked into a deprecation specific callpath,
935
 
            # only deprecations should getting sent via it.
936
 
            self.assertEqual(cls, DeprecationWarning)
937
 
            local_warnings.append(msg)
938
 
        original_warning_method = symbol_versioning.warn
939
 
        symbol_versioning.set_warning_method(capture_warnings)
940
 
        try:
941
 
            result = a_callable(*args, **kwargs)
942
 
        finally:
943
 
            symbol_versioning.set_warning_method(original_warning_method)
944
 
        return (local_warnings, result)
945
 
 
946
 
    def applyDeprecated(self, deprecation_format, a_callable, *args, **kwargs):
947
 
        """Call a deprecated callable without warning the user.
948
 
 
949
 
        :param deprecation_format: The deprecation format that the callable
950
 
            should have been deprecated with. This is the same type as the 
951
 
            parameter to deprecated_method/deprecated_function. If the 
952
 
            callable is not deprecated with this format, an assertion error
953
 
            will be raised.
954
 
        :param a_callable: A callable to call. This may be a bound method or
955
 
            a regular function. It will be called with *args and **kwargs.
956
 
        :param args: The positional arguments for the callable
957
 
        :param kwargs: The keyword arguments for the callable
958
 
        :return: The result of a_callable(*args, **kwargs)
959
 
        """
960
 
        call_warnings, result = self._capture_warnings(a_callable,
961
 
            *args, **kwargs)
962
 
        expected_first_warning = symbol_versioning.deprecation_string(
963
 
            a_callable, deprecation_format)
964
 
        if len(call_warnings) == 0:
965
 
            self.fail("No deprecation warning generated by call to %s" %
966
 
                a_callable)
967
 
        self.assertEqual(expected_first_warning, call_warnings[0])
968
 
        return result
969
 
 
970
 
    def callDeprecated(self, expected, callable, *args, **kwargs):
971
 
        """Assert that a callable is deprecated in a particular way.
972
 
 
973
 
        This is a very precise test for unusual requirements. The 
974
 
        applyDeprecated helper function is probably more suited for most tests
975
 
        as it allows you to simply specify the deprecation format being used
976
 
        and will ensure that that is issued for the function being called.
977
 
 
978
 
        :param expected: a list of the deprecation warnings expected, in order
979
 
        :param callable: The callable to call
980
 
        :param args: The positional arguments for the callable
981
 
        :param kwargs: The keyword arguments for the callable
982
 
        """
983
 
        call_warnings, result = self._capture_warnings(callable,
984
 
            *args, **kwargs)
985
 
        self.assertEqual(expected, call_warnings)
986
 
        return result
987
 
 
988
 
    def _startLogFile(self):
989
 
        """Send bzr and test log messages to a temporary file.
990
 
 
991
 
        The file is removed as the test is torn down.
992
 
        """
993
 
        fileno, name = tempfile.mkstemp(suffix='.log', prefix='testbzr')
994
 
        self._log_file = os.fdopen(fileno, 'w+')
995
 
        self._log_nonce = bzrlib.trace.enable_test_log(self._log_file)
996
 
        self._log_file_name = name
997
 
        self.addCleanup(self._finishLogFile)
998
 
 
999
 
    def _finishLogFile(self):
1000
 
        """Finished with the log file.
1001
 
 
1002
 
        Close the file and delete it, unless setKeepLogfile was called.
1003
 
        """
1004
 
        if self._log_file is None:
1005
 
            return
1006
 
        bzrlib.trace.disable_test_log(self._log_nonce)
1007
 
        self._log_file.close()
1008
 
        self._log_file = None
1009
 
        if not self._keep_log_file:
1010
 
            os.remove(self._log_file_name)
1011
 
            self._log_file_name = None
1012
 
 
1013
 
    def setKeepLogfile(self):
1014
 
        """Make the logfile not be deleted when _finishLogFile is called."""
1015
 
        self._keep_log_file = True
1016
 
 
1017
 
    def addCleanup(self, callable):
1018
 
        """Arrange to run a callable when this case is torn down.
1019
 
 
1020
 
        Callables are run in the reverse of the order they are registered, 
1021
 
        ie last-in first-out.
1022
 
        """
1023
 
        if callable in self._cleanups:
1024
 
            raise ValueError("cleanup function %r already registered on %s" 
1025
 
                    % (callable, self))
1026
 
        self._cleanups.append(callable)
1027
 
 
1028
 
    def _cleanEnvironment(self):
1029
 
        new_env = {
1030
 
            'BZR_HOME': None, # Don't inherit BZR_HOME to all the tests.
1031
 
            'HOME': os.getcwd(),
1032
 
            'APPDATA': None,  # bzr now use Win32 API and don't rely on APPDATA
1033
 
            'BZR_EMAIL': None,
1034
 
            'BZREMAIL': None, # may still be present in the environment
1035
 
            'EMAIL': None,
1036
 
            'BZR_PROGRESS_BAR': None,
1037
 
            # Proxies
1038
 
            'http_proxy': None,
1039
 
            'HTTP_PROXY': None,
1040
 
            'https_proxy': None,
1041
 
            'HTTPS_PROXY': None,
1042
 
            'no_proxy': None,
1043
 
            'NO_PROXY': None,
1044
 
            'all_proxy': None,
1045
 
            'ALL_PROXY': None,
1046
 
            # Nobody cares about these ones AFAIK. So far at
1047
 
            # least. If you do (care), please update this comment
1048
 
            # -- vila 20061212
1049
 
            'ftp_proxy': None,
1050
 
            'FTP_PROXY': None,
1051
 
        }
1052
 
        self.__old_env = {}
1053
 
        self.addCleanup(self._restoreEnvironment)
1054
 
        for name, value in new_env.iteritems():
1055
 
            self._captureVar(name, value)
1056
 
 
1057
 
    def _captureVar(self, name, newvalue):
1058
 
        """Set an environment variable, and reset it when finished."""
1059
 
        self.__old_env[name] = osutils.set_or_unset_env(name, newvalue)
1060
 
 
1061
 
    def _restoreEnvironment(self):
1062
 
        for name, value in self.__old_env.iteritems():
1063
 
            osutils.set_or_unset_env(name, value)
1064
 
 
1065
 
    def _restoreHooks(self):
1066
 
        bzrlib.branch.Branch.hooks = self._preserved_hooks
1067
 
 
1068
 
    def knownFailure(self, reason):
1069
 
        """This test has failed for some known reason."""
1070
 
        raise KnownFailure(reason)
1071
 
 
1072
 
    def run(self, result=None):
1073
 
        if result is None: result = self.defaultTestResult()
1074
 
        for feature in getattr(self, '_test_needs_features', []):
1075
 
            if not feature.available():
1076
 
                result.startTest(self)
1077
 
                if getattr(result, 'addNotSupported', None):
1078
 
                    result.addNotSupported(self, feature)
1079
 
                else:
1080
 
                    result.addSuccess(self)
1081
 
                result.stopTest(self)
1082
 
                return
1083
 
        return unittest.TestCase.run(self, result)
1084
 
 
1085
 
    def tearDown(self):
1086
 
        self._runCleanups()
1087
 
        unittest.TestCase.tearDown(self)
1088
 
 
1089
 
    def time(self, callable, *args, **kwargs):
1090
 
        """Run callable and accrue the time it takes to the benchmark time.
1091
 
        
1092
 
        If lsprofiling is enabled (i.e. by --lsprof-time to bzr selftest) then
1093
 
        this will cause lsprofile statistics to be gathered and stored in
1094
 
        self._benchcalls.
1095
 
        """
1096
 
        if self._benchtime is None:
1097
 
            self._benchtime = 0
1098
 
        start = time.time()
1099
 
        try:
1100
 
            if not self._gather_lsprof_in_benchmarks:
1101
 
                return callable(*args, **kwargs)
1102
 
            else:
1103
 
                # record this benchmark
1104
 
                ret, stats = bzrlib.lsprof.profile(callable, *args, **kwargs)
1105
 
                stats.sort()
1106
 
                self._benchcalls.append(((callable, args, kwargs), stats))
1107
 
                return ret
1108
 
        finally:
1109
 
            self._benchtime += time.time() - start
1110
 
 
1111
 
    def _runCleanups(self):
1112
 
        """Run registered cleanup functions. 
1113
 
 
1114
 
        This should only be called from TestCase.tearDown.
1115
 
        """
1116
 
        # TODO: Perhaps this should keep running cleanups even if 
1117
 
        # one of them fails?
1118
 
 
1119
 
        # Actually pop the cleanups from the list so tearDown running
1120
 
        # twice is safe (this happens for skipped tests).
1121
 
        while self._cleanups:
1122
 
            self._cleanups.pop()()
1123
 
 
1124
 
    def log(self, *args):
1125
 
        mutter(*args)
1126
 
 
1127
 
    def _get_log(self, keep_log_file=False):
1128
 
        """Return as a string the log for this test. If the file is still
1129
 
        on disk and keep_log_file=False, delete the log file and store the
1130
 
        content in self._log_contents."""
1131
 
        # flush the log file, to get all content
1132
 
        import bzrlib.trace
1133
 
        bzrlib.trace._trace_file.flush()
1134
 
        if self._log_contents:
1135
 
            return self._log_contents
1136
 
        if self._log_file_name is not None:
1137
 
            logfile = open(self._log_file_name)
1138
 
            try:
1139
 
                log_contents = logfile.read()
1140
 
            finally:
1141
 
                logfile.close()
1142
 
            if not keep_log_file:
1143
 
                self._log_contents = log_contents
1144
 
                try:
1145
 
                    os.remove(self._log_file_name)
1146
 
                except OSError, e:
1147
 
                    if sys.platform == 'win32' and e.errno == errno.EACCES:
1148
 
                        print >>sys.stderr, ('Unable to delete log file '
1149
 
                                             ' %r' % self._log_file_name)
1150
 
                    else:
1151
 
                        raise
1152
 
            return log_contents
1153
 
        else:
1154
 
            return "DELETED log file to reduce memory footprint"
1155
 
 
1156
 
    def capture(self, cmd, retcode=0):
1157
 
        """Shortcut that splits cmd into words, runs, and returns stdout"""
1158
 
        return self.run_bzr_captured(cmd.split(), retcode=retcode)[0]
1159
 
 
1160
 
    def requireFeature(self, feature):
1161
 
        """This test requires a specific feature is available.
1162
 
 
1163
 
        :raises UnavailableFeature: When feature is not available.
1164
 
        """
1165
 
        if not feature.available():
1166
 
            raise UnavailableFeature(feature)
1167
 
 
1168
 
    def run_bzr_captured(self, argv, retcode=0, encoding=None, stdin=None,
1169
 
                         working_dir=None):
1170
 
        """Invoke bzr and return (stdout, stderr).
1171
 
 
1172
 
        Useful for code that wants to check the contents of the
1173
 
        output, the way error messages are presented, etc.
1174
 
 
1175
 
        This should be the main method for tests that want to exercise the
1176
 
        overall behavior of the bzr application (rather than a unit test
1177
 
        or a functional test of the library.)
1178
 
 
1179
 
        Much of the old code runs bzr by forking a new copy of Python, but
1180
 
        that is slower, harder to debug, and generally not necessary.
1181
 
 
1182
 
        This runs bzr through the interface that catches and reports
1183
 
        errors, and with logging set to something approximating the
1184
 
        default, so that error reporting can be checked.
1185
 
 
1186
 
        :param argv: arguments to invoke bzr
1187
 
        :param retcode: expected return code, or None for don't-care.
1188
 
        :param encoding: encoding for sys.stdout and sys.stderr
1189
 
        :param stdin: A string to be used as stdin for the command.
1190
 
        :param working_dir: Change to this directory before running
1191
 
        """
1192
 
        if encoding is None:
1193
 
            encoding = bzrlib.user_encoding
1194
 
        stdout = StringIOWrapper()
1195
 
        stderr = StringIOWrapper()
1196
 
        stdout.encoding = encoding
1197
 
        stderr.encoding = encoding
1198
 
 
1199
 
        self.log('run bzr: %r', argv)
1200
 
        # FIXME: don't call into logging here
1201
 
        handler = logging.StreamHandler(stderr)
1202
 
        handler.setLevel(logging.INFO)
1203
 
        logger = logging.getLogger('')
1204
 
        logger.addHandler(handler)
1205
 
        old_ui_factory = ui.ui_factory
1206
 
        ui.ui_factory = TestUIFactory(stdin=stdin, stdout=stdout, stderr=stderr)
1207
 
 
1208
 
        cwd = None
1209
 
        if working_dir is not None:
1210
 
            cwd = osutils.getcwd()
1211
 
            os.chdir(working_dir)
1212
 
 
1213
 
        try:
1214
 
            saved_debug_flags = frozenset(debug.debug_flags)
1215
 
            debug.debug_flags.clear()
1216
 
            try:
1217
 
                result = self.apply_redirected(ui.ui_factory.stdin,
1218
 
                                               stdout, stderr,
1219
 
                                               bzrlib.commands.run_bzr_catch_errors,
1220
 
                                               argv)
1221
 
            finally:
1222
 
                debug.debug_flags.update(saved_debug_flags)
1223
 
        finally:
1224
 
            logger.removeHandler(handler)
1225
 
            ui.ui_factory = old_ui_factory
1226
 
            if cwd is not None:
1227
 
                os.chdir(cwd)
1228
 
 
1229
 
        out = stdout.getvalue()
1230
 
        err = stderr.getvalue()
1231
 
        if out:
1232
 
            self.log('output:\n%r', out)
1233
 
        if err:
1234
 
            self.log('errors:\n%r', err)
1235
 
        if retcode is not None:
1236
 
            self.assertEquals(retcode, result)
1237
 
        return out, err
1238
 
 
1239
 
    def run_bzr(self, *args, **kwargs):
1240
 
        """Invoke bzr, as if it were run from the command line.
1241
 
 
1242
 
        This should be the main method for tests that want to exercise the
1243
 
        overall behavior of the bzr application (rather than a unit test
1244
 
        or a functional test of the library.)
1245
 
 
1246
 
        This sends the stdout/stderr results into the test's log,
1247
 
        where it may be useful for debugging.  See also run_captured.
1248
 
 
1249
 
        :param stdin: A string to be used as stdin for the command.
1250
 
        :param retcode: The status code the command should return
1251
 
        :param working_dir: The directory to run the command in
1252
 
        """
1253
 
        retcode = kwargs.pop('retcode', 0)
1254
 
        encoding = kwargs.pop('encoding', None)
1255
 
        stdin = kwargs.pop('stdin', None)
1256
 
        working_dir = kwargs.pop('working_dir', None)
1257
 
        return self.run_bzr_captured(args, retcode=retcode, encoding=encoding,
1258
 
                                     stdin=stdin, working_dir=working_dir)
1259
 
 
1260
 
    def run_bzr_decode(self, *args, **kwargs):
1261
 
        if 'encoding' in kwargs:
1262
 
            encoding = kwargs['encoding']
1263
 
        else:
1264
 
            encoding = bzrlib.user_encoding
1265
 
        return self.run_bzr(*args, **kwargs)[0].decode(encoding)
1266
 
 
1267
 
    def run_bzr_error(self, error_regexes, *args, **kwargs):
1268
 
        """Run bzr, and check that stderr contains the supplied regexes
1269
 
        
1270
 
        :param error_regexes: Sequence of regular expressions which 
1271
 
            must each be found in the error output. The relative ordering
1272
 
            is not enforced.
1273
 
        :param args: command-line arguments for bzr
1274
 
        :param kwargs: Keyword arguments which are interpreted by run_bzr
1275
 
            This function changes the default value of retcode to be 3,
1276
 
            since in most cases this is run when you expect bzr to fail.
1277
 
        :return: (out, err) The actual output of running the command (in case you
1278
 
                 want to do more inspection)
1279
 
 
1280
 
        Examples of use:
1281
 
            # Make sure that commit is failing because there is nothing to do
1282
 
            self.run_bzr_error(['no changes to commit'],
1283
 
                               'commit', '-m', 'my commit comment')
1284
 
            # Make sure --strict is handling an unknown file, rather than
1285
 
            # giving us the 'nothing to do' error
1286
 
            self.build_tree(['unknown'])
1287
 
            self.run_bzr_error(['Commit refused because there are unknown files'],
1288
 
                               'commit', '--strict', '-m', 'my commit comment')
1289
 
        """
1290
 
        kwargs.setdefault('retcode', 3)
1291
 
        out, err = self.run_bzr(*args, **kwargs)
1292
 
        for regex in error_regexes:
1293
 
            self.assertContainsRe(err, regex)
1294
 
        return out, err
1295
 
 
1296
 
    def run_bzr_subprocess(self, *args, **kwargs):
1297
 
        """Run bzr in a subprocess for testing.
1298
 
 
1299
 
        This starts a new Python interpreter and runs bzr in there. 
1300
 
        This should only be used for tests that have a justifiable need for
1301
 
        this isolation: e.g. they are testing startup time, or signal
1302
 
        handling, or early startup code, etc.  Subprocess code can't be 
1303
 
        profiled or debugged so easily.
1304
 
 
1305
 
        :param retcode: The status code that is expected.  Defaults to 0.  If
1306
 
            None is supplied, the status code is not checked.
1307
 
        :param env_changes: A dictionary which lists changes to environment
1308
 
            variables. A value of None will unset the env variable.
1309
 
            The values must be strings. The change will only occur in the
1310
 
            child, so you don't need to fix the environment after running.
1311
 
        :param universal_newlines: Convert CRLF => LF
1312
 
        :param allow_plugins: By default the subprocess is run with
1313
 
            --no-plugins to ensure test reproducibility. Also, it is possible
1314
 
            for system-wide plugins to create unexpected output on stderr,
1315
 
            which can cause unnecessary test failures.
1316
 
        """
1317
 
        env_changes = kwargs.get('env_changes', {})
1318
 
        working_dir = kwargs.get('working_dir', None)
1319
 
        allow_plugins = kwargs.get('allow_plugins', False)
1320
 
        process = self.start_bzr_subprocess(args, env_changes=env_changes,
1321
 
                                            working_dir=working_dir,
1322
 
                                            allow_plugins=allow_plugins)
1323
 
        # We distinguish between retcode=None and retcode not passed.
1324
 
        supplied_retcode = kwargs.get('retcode', 0)
1325
 
        return self.finish_bzr_subprocess(process, retcode=supplied_retcode,
1326
 
            universal_newlines=kwargs.get('universal_newlines', False),
1327
 
            process_args=args)
1328
 
 
1329
 
    def start_bzr_subprocess(self, process_args, env_changes=None,
1330
 
                             skip_if_plan_to_signal=False,
1331
 
                             working_dir=None,
1332
 
                             allow_plugins=False):
1333
 
        """Start bzr in a subprocess for testing.
1334
 
 
1335
 
        This starts a new Python interpreter and runs bzr in there.
1336
 
        This should only be used for tests that have a justifiable need for
1337
 
        this isolation: e.g. they are testing startup time, or signal
1338
 
        handling, or early startup code, etc.  Subprocess code can't be
1339
 
        profiled or debugged so easily.
1340
 
 
1341
 
        :param process_args: a list of arguments to pass to the bzr executable,
1342
 
            for example `['--version']`.
1343
 
        :param env_changes: A dictionary which lists changes to environment
1344
 
            variables. A value of None will unset the env variable.
1345
 
            The values must be strings. The change will only occur in the
1346
 
            child, so you don't need to fix the environment after running.
1347
 
        :param skip_if_plan_to_signal: raise TestSkipped when true and os.kill
1348
 
            is not available.
1349
 
        :param allow_plugins: If False (default) pass --no-plugins to bzr.
1350
 
 
1351
 
        :returns: Popen object for the started process.
1352
 
        """
1353
 
        if skip_if_plan_to_signal:
1354
 
            if not getattr(os, 'kill', None):
1355
 
                raise TestSkipped("os.kill not available.")
1356
 
 
1357
 
        if env_changes is None:
1358
 
            env_changes = {}
1359
 
        old_env = {}
1360
 
 
1361
 
        def cleanup_environment():
1362
 
            for env_var, value in env_changes.iteritems():
1363
 
                old_env[env_var] = osutils.set_or_unset_env(env_var, value)
1364
 
 
1365
 
        def restore_environment():
1366
 
            for env_var, value in old_env.iteritems():
1367
 
                osutils.set_or_unset_env(env_var, value)
1368
 
 
1369
 
        bzr_path = self.get_bzr_path()
1370
 
 
1371
 
        cwd = None
1372
 
        if working_dir is not None:
1373
 
            cwd = osutils.getcwd()
1374
 
            os.chdir(working_dir)
1375
 
 
1376
 
        try:
1377
 
            # win32 subprocess doesn't support preexec_fn
1378
 
            # so we will avoid using it on all platforms, just to
1379
 
            # make sure the code path is used, and we don't break on win32
1380
 
            cleanup_environment()
1381
 
            command = [sys.executable, bzr_path]
1382
 
            if not allow_plugins:
1383
 
                command.append('--no-plugins')
1384
 
            command.extend(process_args)
1385
 
            process = self._popen(command, stdin=PIPE, stdout=PIPE, stderr=PIPE)
1386
 
        finally:
1387
 
            restore_environment()
1388
 
            if cwd is not None:
1389
 
                os.chdir(cwd)
1390
 
 
1391
 
        return process
1392
 
 
1393
 
    def _popen(self, *args, **kwargs):
1394
 
        """Place a call to Popen.
1395
 
 
1396
 
        Allows tests to override this method to intercept the calls made to
1397
 
        Popen for introspection.
1398
 
        """
1399
 
        return Popen(*args, **kwargs)
1400
 
 
1401
 
    def get_bzr_path(self):
1402
 
        """Return the path of the 'bzr' executable for this test suite."""
1403
 
        bzr_path = os.path.dirname(os.path.dirname(bzrlib.__file__))+'/bzr'
1404
 
        if not os.path.isfile(bzr_path):
1405
 
            # We are probably installed. Assume sys.argv is the right file
1406
 
            bzr_path = sys.argv[0]
1407
 
        return bzr_path
1408
 
 
1409
 
    def finish_bzr_subprocess(self, process, retcode=0, send_signal=None,
1410
 
                              universal_newlines=False, process_args=None):
1411
 
        """Finish the execution of process.
1412
 
 
1413
 
        :param process: the Popen object returned from start_bzr_subprocess.
1414
 
        :param retcode: The status code that is expected.  Defaults to 0.  If
1415
 
            None is supplied, the status code is not checked.
1416
 
        :param send_signal: an optional signal to send to the process.
1417
 
        :param universal_newlines: Convert CRLF => LF
1418
 
        :returns: (stdout, stderr)
1419
 
        """
1420
 
        if send_signal is not None:
1421
 
            os.kill(process.pid, send_signal)
1422
 
        out, err = process.communicate()
1423
 
 
1424
 
        if universal_newlines:
1425
 
            out = out.replace('\r\n', '\n')
1426
 
            err = err.replace('\r\n', '\n')
1427
 
 
1428
 
        if retcode is not None and retcode != process.returncode:
1429
 
            if process_args is None:
1430
 
                process_args = "(unknown args)"
1431
 
            mutter('Output of bzr %s:\n%s', process_args, out)
1432
 
            mutter('Error for bzr %s:\n%s', process_args, err)
1433
 
            self.fail('Command bzr %s failed with retcode %s != %s'
1434
 
                      % (process_args, retcode, process.returncode))
1435
 
        return [out, err]
1436
 
 
1437
 
    def check_inventory_shape(self, inv, shape):
1438
 
        """Compare an inventory to a list of expected names.
1439
 
 
1440
 
        Fail if they are not precisely equal.
1441
 
        """
1442
 
        extras = []
1443
 
        shape = list(shape)             # copy
1444
 
        for path, ie in inv.entries():
1445
 
            name = path.replace('\\', '/')
1446
 
            if ie.kind == 'dir':
1447
 
                name = name + '/'
1448
 
            if name in shape:
1449
 
                shape.remove(name)
1450
 
            else:
1451
 
                extras.append(name)
1452
 
        if shape:
1453
 
            self.fail("expected paths not found in inventory: %r" % shape)
1454
 
        if extras:
1455
 
            self.fail("unexpected paths found in inventory: %r" % extras)
1456
 
 
1457
 
    def apply_redirected(self, stdin=None, stdout=None, stderr=None,
1458
 
                         a_callable=None, *args, **kwargs):
1459
 
        """Call callable with redirected std io pipes.
1460
 
 
1461
 
        Returns the return code."""
1462
 
        if not callable(a_callable):
1463
 
            raise ValueError("a_callable must be callable.")
1464
 
        if stdin is None:
1465
 
            stdin = StringIO("")
1466
 
        if stdout is None:
1467
 
            if getattr(self, "_log_file", None) is not None:
1468
 
                stdout = self._log_file
1469
 
            else:
1470
 
                stdout = StringIO()
1471
 
        if stderr is None:
1472
 
            if getattr(self, "_log_file", None is not None):
1473
 
                stderr = self._log_file
1474
 
            else:
1475
 
                stderr = StringIO()
1476
 
        real_stdin = sys.stdin
1477
 
        real_stdout = sys.stdout
1478
 
        real_stderr = sys.stderr
1479
 
        try:
1480
 
            sys.stdout = stdout
1481
 
            sys.stderr = stderr
1482
 
            sys.stdin = stdin
1483
 
            return a_callable(*args, **kwargs)
1484
 
        finally:
1485
 
            sys.stdout = real_stdout
1486
 
            sys.stderr = real_stderr
1487
 
            sys.stdin = real_stdin
1488
 
 
1489
 
    @symbol_versioning.deprecated_method(symbol_versioning.zero_eleven)
1490
 
    def merge(self, branch_from, wt_to):
1491
 
        """A helper for tests to do a ui-less merge.
1492
 
 
1493
 
        This should move to the main library when someone has time to integrate
1494
 
        it in.
1495
 
        """
1496
 
        # minimal ui-less merge.
1497
 
        wt_to.branch.fetch(branch_from)
1498
 
        base_rev = common_ancestor(branch_from.last_revision(),
1499
 
                                   wt_to.branch.last_revision(),
1500
 
                                   wt_to.branch.repository)
1501
 
        merge_inner(wt_to.branch, branch_from.basis_tree(),
1502
 
                    wt_to.branch.repository.revision_tree(base_rev),
1503
 
                    this_tree=wt_to)
1504
 
        wt_to.add_parent_tree_id(branch_from.last_revision())
1505
 
 
1506
 
    def reduceLockdirTimeout(self):
1507
 
        """Reduce the default lock timeout for the duration of the test, so that
1508
 
        if LockContention occurs during a test, it does so quickly.
1509
 
 
1510
 
        Tests that expect to provoke LockContention errors should call this.
1511
 
        """
1512
 
        orig_timeout = bzrlib.lockdir._DEFAULT_TIMEOUT_SECONDS
1513
 
        def resetTimeout():
1514
 
            bzrlib.lockdir._DEFAULT_TIMEOUT_SECONDS = orig_timeout
1515
 
        self.addCleanup(resetTimeout)
1516
 
        bzrlib.lockdir._DEFAULT_TIMEOUT_SECONDS = 0
1517
 
 
1518
 
BzrTestBase = TestCase
1519
 
 
1520
 
 
1521
 
class TestCaseWithMemoryTransport(TestCase):
1522
 
    """Common test class for tests that do not need disk resources.
1523
 
 
1524
 
    Tests that need disk resources should derive from TestCaseWithTransport.
1525
 
 
1526
 
    TestCaseWithMemoryTransport sets the TEST_ROOT variable for all bzr tests.
1527
 
 
1528
 
    For TestCaseWithMemoryTransport the test_home_dir is set to the name of
1529
 
    a directory which does not exist. This serves to help ensure test isolation
1530
 
    is preserved. test_dir is set to the TEST_ROOT, as is cwd, because they
1531
 
    must exist. However, TestCaseWithMemoryTransport does not offer local
1532
 
    file defaults for the transport in tests, nor does it obey the command line
1533
 
    override, so tests that accidentally write to the common directory should
1534
 
    be rare.
1535
 
    """
1536
 
 
1537
 
    TEST_ROOT = None
1538
 
    _TEST_NAME = 'test'
1539
 
 
1540
 
 
1541
 
    def __init__(self, methodName='runTest'):
1542
 
        # allow test parameterisation after test construction and before test
1543
 
        # execution. Variables that the parameteriser sets need to be 
1544
 
        # ones that are not set by setUp, or setUp will trash them.
1545
 
        super(TestCaseWithMemoryTransport, self).__init__(methodName)
1546
 
        self.vfs_transport_factory = default_transport
1547
 
        self.transport_server = None
1548
 
        self.transport_readonly_server = None
1549
 
        self.__vfs_server = None
1550
 
 
1551
 
    def get_transport(self):
1552
 
        """Return a writeable transport for the test scratch space"""
1553
 
        t = get_transport(self.get_url())
1554
 
        self.assertFalse(t.is_readonly())
1555
 
        return t
1556
 
 
1557
 
    def get_readonly_transport(self):
1558
 
        """Return a readonly transport for the test scratch space
1559
 
        
1560
 
        This can be used to test that operations which should only need
1561
 
        readonly access in fact do not try to write.
1562
 
        """
1563
 
        t = get_transport(self.get_readonly_url())
1564
 
        self.assertTrue(t.is_readonly())
1565
 
        return t
1566
 
 
1567
 
    def create_transport_readonly_server(self):
1568
 
        """Create a transport server from class defined at init.
1569
 
 
1570
 
        This is mostly a hook for daughter classes.
1571
 
        """
1572
 
        return self.transport_readonly_server()
1573
 
 
1574
 
    def get_readonly_server(self):
1575
 
        """Get the server instance for the readonly transport
1576
 
 
1577
 
        This is useful for some tests with specific servers to do diagnostics.
1578
 
        """
1579
 
        if self.__readonly_server is None:
1580
 
            if self.transport_readonly_server is None:
1581
 
                # readonly decorator requested
1582
 
                # bring up the server
1583
 
                self.__readonly_server = ReadonlyServer()
1584
 
                self.__readonly_server.setUp(self.get_vfs_only_server())
1585
 
            else:
1586
 
                self.__readonly_server = self.create_transport_readonly_server()
1587
 
                self.__readonly_server.setUp(self.get_vfs_only_server())
1588
 
            self.addCleanup(self.__readonly_server.tearDown)
1589
 
        return self.__readonly_server
1590
 
 
1591
 
    def get_readonly_url(self, relpath=None):
1592
 
        """Get a URL for the readonly transport.
1593
 
 
1594
 
        This will either be backed by '.' or a decorator to the transport 
1595
 
        used by self.get_url()
1596
 
        relpath provides for clients to get a path relative to the base url.
1597
 
        These should only be downwards relative, not upwards.
1598
 
        """
1599
 
        base = self.get_readonly_server().get_url()
1600
 
        if relpath is not None:
1601
 
            if not base.endswith('/'):
1602
 
                base = base + '/'
1603
 
            base = base + relpath
1604
 
        return base
1605
 
 
1606
 
    def get_vfs_only_server(self):
1607
 
        """Get the vfs only read/write server instance.
1608
 
 
1609
 
        This is useful for some tests with specific servers that need
1610
 
        diagnostics.
1611
 
 
1612
 
        For TestCaseWithMemoryTransport this is always a MemoryServer, and there
1613
 
        is no means to override it.
1614
 
        """
1615
 
        if self.__vfs_server is None:
1616
 
            self.__vfs_server = MemoryServer()
1617
 
            self.__vfs_server.setUp()
1618
 
            self.addCleanup(self.__vfs_server.tearDown)
1619
 
        return self.__vfs_server
1620
 
 
1621
 
    def get_server(self):
1622
 
        """Get the read/write server instance.
1623
 
 
1624
 
        This is useful for some tests with specific servers that need
1625
 
        diagnostics.
1626
 
 
1627
 
        This is built from the self.transport_server factory. If that is None,
1628
 
        then the self.get_vfs_server is returned.
1629
 
        """
1630
 
        if self.__server is None:
1631
 
            if self.transport_server is None or self.transport_server is self.vfs_transport_factory:
1632
 
                return self.get_vfs_only_server()
1633
 
            else:
1634
 
                # bring up a decorated means of access to the vfs only server.
1635
 
                self.__server = self.transport_server()
1636
 
                try:
1637
 
                    self.__server.setUp(self.get_vfs_only_server())
1638
 
                except TypeError, e:
1639
 
                    # This should never happen; the try:Except here is to assist
1640
 
                    # developers having to update code rather than seeing an
1641
 
                    # uninformative TypeError.
1642
 
                    raise Exception, "Old server API in use: %s, %s" % (self.__server, e)
1643
 
            self.addCleanup(self.__server.tearDown)
1644
 
        return self.__server
1645
 
 
1646
 
    def _adjust_url(self, base, relpath):
1647
 
        """Get a URL (or maybe a path) for the readwrite transport.
1648
 
 
1649
 
        This will either be backed by '.' or to an equivalent non-file based
1650
 
        facility.
1651
 
        relpath provides for clients to get a path relative to the base url.
1652
 
        These should only be downwards relative, not upwards.
1653
 
        """
1654
 
        if relpath is not None and relpath != '.':
1655
 
            if not base.endswith('/'):
1656
 
                base = base + '/'
1657
 
            # XXX: Really base should be a url; we did after all call
1658
 
            # get_url()!  But sometimes it's just a path (from
1659
 
            # LocalAbspathServer), and it'd be wrong to append urlescaped data
1660
 
            # to a non-escaped local path.
1661
 
            if base.startswith('./') or base.startswith('/'):
1662
 
                base += relpath
1663
 
            else:
1664
 
                base += urlutils.escape(relpath)
1665
 
        return base
1666
 
 
1667
 
    def get_url(self, relpath=None):
1668
 
        """Get a URL (or maybe a path) for the readwrite transport.
1669
 
 
1670
 
        This will either be backed by '.' or to an equivalent non-file based
1671
 
        facility.
1672
 
        relpath provides for clients to get a path relative to the base url.
1673
 
        These should only be downwards relative, not upwards.
1674
 
        """
1675
 
        base = self.get_server().get_url()
1676
 
        return self._adjust_url(base, relpath)
1677
 
 
1678
 
    def get_vfs_only_url(self, relpath=None):
1679
 
        """Get a URL (or maybe a path for the plain old vfs transport.
1680
 
 
1681
 
        This will never be a smart protocol.
1682
 
        :param relpath: provides for clients to get a path relative to the base
1683
 
            url.  These should only be downwards relative, not upwards.
1684
 
        """
1685
 
        base = self.get_vfs_only_server().get_url()
1686
 
        return self._adjust_url(base, relpath)
1687
 
 
1688
 
    def _make_test_root(self):
1689
 
        if TestCaseWithMemoryTransport.TEST_ROOT is not None:
1690
 
            return
1691
 
        i = 0
1692
 
        while True:
1693
 
            root = u'test%04d.tmp' % i
1694
 
            try:
1695
 
                os.mkdir(root)
1696
 
            except OSError, e:
1697
 
                if e.errno == errno.EEXIST:
1698
 
                    i += 1
1699
 
                    continue
1700
 
                else:
1701
 
                    raise
1702
 
            # successfully created
1703
 
            TestCaseWithMemoryTransport.TEST_ROOT = osutils.abspath(root)
1704
 
            break
1705
 
        # make a fake bzr directory there to prevent any tests propagating
1706
 
        # up onto the source directory's real branch
1707
 
        bzrdir.BzrDir.create_standalone_workingtree(
1708
 
            TestCaseWithMemoryTransport.TEST_ROOT)
1709
 
 
1710
 
    def makeAndChdirToTestDir(self):
1711
 
        """Create a temporary directories for this one test.
1712
 
        
1713
 
        This must set self.test_home_dir and self.test_dir and chdir to
1714
 
        self.test_dir.
1715
 
        
1716
 
        For TestCaseWithMemoryTransport we chdir to the TEST_ROOT for this test.
1717
 
        """
1718
 
        os.chdir(TestCaseWithMemoryTransport.TEST_ROOT)
1719
 
        self.test_dir = TestCaseWithMemoryTransport.TEST_ROOT
1720
 
        self.test_home_dir = self.test_dir + "/MemoryTransportMissingHomeDir"
1721
 
        
1722
 
    def make_branch(self, relpath, format=None):
1723
 
        """Create a branch on the transport at relpath."""
1724
 
        repo = self.make_repository(relpath, format=format)
1725
 
        return repo.bzrdir.create_branch()
1726
 
 
1727
 
    def make_bzrdir(self, relpath, format=None):
1728
 
        try:
1729
 
            # might be a relative or absolute path
1730
 
            maybe_a_url = self.get_url(relpath)
1731
 
            segments = maybe_a_url.rsplit('/', 1)
1732
 
            t = get_transport(maybe_a_url)
1733
 
            if len(segments) > 1 and segments[-1] not in ('', '.'):
1734
 
                try:
1735
 
                    t.mkdir('.')
1736
 
                except errors.FileExists:
1737
 
                    pass
1738
 
            if format is None:
1739
 
                format = 'default'
1740
 
            if isinstance(format, basestring):
1741
 
                format = bzrdir.format_registry.make_bzrdir(format)
1742
 
            return format.initialize_on_transport(t)
1743
 
        except errors.UninitializableFormat:
1744
 
            raise TestSkipped("Format %s is not initializable." % format)
1745
 
 
1746
 
    def make_repository(self, relpath, shared=False, format=None):
1747
 
        """Create a repository on our default transport at relpath."""
1748
 
        made_control = self.make_bzrdir(relpath, format=format)
1749
 
        return made_control.create_repository(shared=shared)
1750
 
 
1751
 
    def make_branch_and_memory_tree(self, relpath, format=None):
1752
 
        """Create a branch on the default transport and a MemoryTree for it."""
1753
 
        b = self.make_branch(relpath, format=format)
1754
 
        return memorytree.MemoryTree.create_on_branch(b)
1755
 
 
1756
 
    def overrideEnvironmentForTesting(self):
1757
 
        os.environ['HOME'] = self.test_home_dir
1758
 
        os.environ['BZR_HOME'] = self.test_home_dir
1759
 
        
1760
 
    def setUp(self):
1761
 
        super(TestCaseWithMemoryTransport, self).setUp()
1762
 
        self._make_test_root()
1763
 
        _currentdir = os.getcwdu()
1764
 
        def _leaveDirectory():
1765
 
            os.chdir(_currentdir)
1766
 
        self.addCleanup(_leaveDirectory)
1767
 
        self.makeAndChdirToTestDir()
1768
 
        self.overrideEnvironmentForTesting()
1769
 
        self.__readonly_server = None
1770
 
        self.__server = None
1771
 
        self.reduceLockdirTimeout()
1772
 
 
1773
 
     
1774
 
class TestCaseInTempDir(TestCaseWithMemoryTransport):
1775
 
    """Derived class that runs a test within a temporary directory.
1776
 
 
1777
 
    This is useful for tests that need to create a branch, etc.
1778
 
 
1779
 
    The directory is created in a slightly complex way: for each
1780
 
    Python invocation, a new temporary top-level directory is created.
1781
 
    All test cases create their own directory within that.  If the
1782
 
    tests complete successfully, the directory is removed.
1783
 
 
1784
 
    InTempDir is an old alias for FunctionalTestCase.
1785
 
    """
1786
 
 
1787
 
    OVERRIDE_PYTHON = 'python'
1788
 
 
1789
 
    def check_file_contents(self, filename, expect):
1790
 
        self.log("check contents of file %s" % filename)
1791
 
        contents = file(filename, 'r').read()
1792
 
        if contents != expect:
1793
 
            self.log("expected: %r" % expect)
1794
 
            self.log("actually: %r" % contents)
1795
 
            self.fail("contents of %s not as expected" % filename)
1796
 
 
1797
 
    def makeAndChdirToTestDir(self):
1798
 
        """See TestCaseWithMemoryTransport.makeAndChdirToTestDir().
1799
 
        
1800
 
        For TestCaseInTempDir we create a temporary directory based on the test
1801
 
        name and then create two subdirs - test and home under it.
1802
 
        """
1803
 
        if NUMBERED_DIRS:       # strongly recommended on Windows
1804
 
                                # due the path length limitation (260 chars)
1805
 
            candidate_dir = '%s/%dK/%05d' % (self.TEST_ROOT,
1806
 
                                             int(self.number/1000),
1807
 
                                             self.number)
1808
 
            os.makedirs(candidate_dir)
1809
 
            self.test_home_dir = candidate_dir + '/home'
1810
 
            os.mkdir(self.test_home_dir)
1811
 
            self.test_dir = candidate_dir + '/work'
1812
 
            os.mkdir(self.test_dir)
1813
 
            os.chdir(self.test_dir)
1814
 
            # put name of test inside
1815
 
            f = file(candidate_dir + '/name', 'w')
1816
 
            f.write(self.id())
1817
 
            f.close()
1818
 
            return
1819
 
        # Else NAMED DIRS
1820
 
        # shorten the name, to avoid test failures due to path length
1821
 
        short_id = self.id().replace('bzrlib.tests.', '') \
1822
 
                   .replace('__main__.', '')[-100:]
1823
 
        # it's possible the same test class is run several times for
1824
 
        # parameterized tests, so make sure the names don't collide.  
1825
 
        i = 0
1826
 
        while True:
1827
 
            if i > 0:
1828
 
                candidate_dir = '%s/%s.%d' % (self.TEST_ROOT, short_id, i)
1829
 
            else:
1830
 
                candidate_dir = '%s/%s' % (self.TEST_ROOT, short_id)
1831
 
            if os.path.exists(candidate_dir):
1832
 
                i = i + 1
1833
 
                continue
1834
 
            else:
1835
 
                os.mkdir(candidate_dir)
1836
 
                self.test_home_dir = candidate_dir + '/home'
1837
 
                os.mkdir(self.test_home_dir)
1838
 
                self.test_dir = candidate_dir + '/work'
1839
 
                os.mkdir(self.test_dir)
1840
 
                os.chdir(self.test_dir)
1841
 
                break
1842
 
 
1843
 
    def build_tree(self, shape, line_endings='binary', transport=None):
1844
 
        """Build a test tree according to a pattern.
1845
 
 
1846
 
        shape is a sequence of file specifications.  If the final
1847
 
        character is '/', a directory is created.
1848
 
 
1849
 
        This assumes that all the elements in the tree being built are new.
1850
 
 
1851
 
        This doesn't add anything to a branch.
1852
 
        :param line_endings: Either 'binary' or 'native'
1853
 
                             in binary mode, exact contents are written
1854
 
                             in native mode, the line endings match the
1855
 
                             default platform endings.
1856
 
 
1857
 
        :param transport: A transport to write to, for building trees on 
1858
 
                          VFS's. If the transport is readonly or None,
1859
 
                          "." is opened automatically.
1860
 
        """
1861
 
        # It's OK to just create them using forward slashes on windows.
1862
 
        if transport is None or transport.is_readonly():
1863
 
            transport = get_transport(".")
1864
 
        for name in shape:
1865
 
            self.assert_(isinstance(name, basestring))
1866
 
            if name[-1] == '/':
1867
 
                transport.mkdir(urlutils.escape(name[:-1]))
1868
 
            else:
1869
 
                if line_endings == 'binary':
1870
 
                    end = '\n'
1871
 
                elif line_endings == 'native':
1872
 
                    end = os.linesep
1873
 
                else:
1874
 
                    raise errors.BzrError(
1875
 
                        'Invalid line ending request %r' % line_endings)
1876
 
                content = "contents of %s%s" % (name.encode('utf-8'), end)
1877
 
                transport.put_bytes_non_atomic(urlutils.escape(name), content)
1878
 
 
1879
 
    def build_tree_contents(self, shape):
1880
 
        build_tree_contents(shape)
1881
 
 
1882
 
    def assertFileEqual(self, content, path):
1883
 
        """Fail if path does not contain 'content'."""
1884
 
        self.failUnlessExists(path)
1885
 
        # TODO: jam 20060427 Shouldn't this be 'rb'?
1886
 
        f = file(path, 'r')
1887
 
        try:
1888
 
            s = f.read()
1889
 
        finally:
1890
 
            f.close()
1891
 
        self.assertEqualDiff(content, s)
1892
 
 
1893
 
    def failUnlessExists(self, path):
1894
 
        """Fail unless path, which may be abs or relative, exists."""
1895
 
        self.failUnless(osutils.lexists(path),path+" does not exist")
1896
 
 
1897
 
    def failIfExists(self, path):
1898
 
        """Fail if path, which may be abs or relative, exists."""
1899
 
        self.failIf(osutils.lexists(path),path+" exists")
1900
 
 
1901
 
 
1902
 
class TestCaseWithTransport(TestCaseInTempDir):
1903
 
    """A test case that provides get_url and get_readonly_url facilities.
1904
 
 
1905
 
    These back onto two transport servers, one for readonly access and one for
1906
 
    read write access.
1907
 
 
1908
 
    If no explicit class is provided for readonly access, a
1909
 
    ReadonlyTransportDecorator is used instead which allows the use of non disk
1910
 
    based read write transports.
1911
 
 
1912
 
    If an explicit class is provided for readonly access, that server and the 
1913
 
    readwrite one must both define get_url() as resolving to os.getcwd().
1914
 
    """
1915
 
 
1916
 
    def get_vfs_only_server(self):
1917
 
        """See TestCaseWithMemoryTransport.
1918
 
 
1919
 
        This is useful for some tests with specific servers that need
1920
 
        diagnostics.
1921
 
        """
1922
 
        if self.__vfs_server is None:
1923
 
            self.__vfs_server = self.vfs_transport_factory()
1924
 
            self.__vfs_server.setUp()
1925
 
            self.addCleanup(self.__vfs_server.tearDown)
1926
 
        return self.__vfs_server
1927
 
 
1928
 
    def make_branch_and_tree(self, relpath, format=None):
1929
 
        """Create a branch on the transport and a tree locally.
1930
 
 
1931
 
        If the transport is not a LocalTransport, the Tree can't be created on
1932
 
        the transport.  In that case if the vfs_transport_factory is
1933
 
        LocalURLServer the working tree is created in the local
1934
 
        directory backing the transport, and the returned tree's branch and
1935
 
        repository will also be accessed locally. Otherwise a lightweight
1936
 
        checkout is created and returned.
1937
 
 
1938
 
        :param format: The BzrDirFormat.
1939
 
        :returns: the WorkingTree.
1940
 
        """
1941
 
        # TODO: always use the local disk path for the working tree,
1942
 
        # this obviously requires a format that supports branch references
1943
 
        # so check for that by checking bzrdir.BzrDirFormat.get_default_format()
1944
 
        # RBC 20060208
1945
 
        b = self.make_branch(relpath, format=format)
1946
 
        try:
1947
 
            return b.bzrdir.create_workingtree()
1948
 
        except errors.NotLocalUrl:
1949
 
            # We can only make working trees locally at the moment.  If the
1950
 
            # transport can't support them, then we keep the non-disk-backed
1951
 
            # branch and create a local checkout.
1952
 
            if self.vfs_transport_factory is LocalURLServer:
1953
 
                # the branch is colocated on disk, we cannot create a checkout.
1954
 
                # hopefully callers will expect this.
1955
 
                local_controldir= bzrdir.BzrDir.open(self.get_vfs_only_url(relpath))
1956
 
                return local_controldir.create_workingtree()
1957
 
            else:
1958
 
                return b.create_checkout(relpath, lightweight=True)
1959
 
 
1960
 
    def assertIsDirectory(self, relpath, transport):
1961
 
        """Assert that relpath within transport is a directory.
1962
 
 
1963
 
        This may not be possible on all transports; in that case it propagates
1964
 
        a TransportNotPossible.
1965
 
        """
1966
 
        try:
1967
 
            mode = transport.stat(relpath).st_mode
1968
 
        except errors.NoSuchFile:
1969
 
            self.fail("path %s is not a directory; no such file"
1970
 
                      % (relpath))
1971
 
        if not stat.S_ISDIR(mode):
1972
 
            self.fail("path %s is not a directory; has mode %#o"
1973
 
                      % (relpath, mode))
1974
 
 
1975
 
    def assertTreesEqual(self, left, right):
1976
 
        """Check that left and right have the same content and properties."""
1977
 
        # we use a tree delta to check for equality of the content, and we
1978
 
        # manually check for equality of other things such as the parents list.
1979
 
        self.assertEqual(left.get_parent_ids(), right.get_parent_ids())
1980
 
        differences = left.changes_from(right)
1981
 
        self.assertFalse(differences.has_changed(),
1982
 
            "Trees %r and %r are different: %r" % (left, right, differences))
1983
 
 
1984
 
    def setUp(self):
1985
 
        super(TestCaseWithTransport, self).setUp()
1986
 
        self.__vfs_server = None
1987
 
 
1988
 
 
1989
 
class ChrootedTestCase(TestCaseWithTransport):
1990
 
    """A support class that provides readonly urls outside the local namespace.
1991
 
 
1992
 
    This is done by checking if self.transport_server is a MemoryServer. if it
1993
 
    is then we are chrooted already, if it is not then an HttpServer is used
1994
 
    for readonly urls.
1995
 
 
1996
 
    TODO RBC 20060127: make this an option to TestCaseWithTransport so it can
1997
 
                       be used without needed to redo it when a different 
1998
 
                       subclass is in use ?
1999
 
    """
2000
 
 
2001
 
    def setUp(self):
2002
 
        super(ChrootedTestCase, self).setUp()
2003
 
        if not self.vfs_transport_factory == MemoryServer:
2004
 
            self.transport_readonly_server = HttpServer
2005
 
 
2006
 
 
2007
 
def filter_suite_by_re(suite, pattern):
2008
 
    result = TestUtil.TestSuite()
2009
 
    filter_re = re.compile(pattern)
2010
 
    for test in iter_suite_tests(suite):
2011
 
        if filter_re.search(test.id()):
2012
 
            result.addTest(test)
2013
 
    return result
2014
 
 
2015
 
 
2016
 
def sort_suite_by_re(suite, pattern):
2017
 
    first = []
2018
 
    second = []
2019
 
    filter_re = re.compile(pattern)
2020
 
    for test in iter_suite_tests(suite):
2021
 
        if filter_re.search(test.id()):
2022
 
            first.append(test)
2023
 
        else:
2024
 
            second.append(test)
2025
 
    return TestUtil.TestSuite(first + second)
2026
 
 
2027
 
 
2028
 
def run_suite(suite, name='test', verbose=False, pattern=".*",
2029
 
              stop_on_failure=False, keep_output=False,
2030
 
              transport=None, lsprof_timed=None, bench_history=None,
2031
 
              matching_tests_first=None,
2032
 
              numbered_dirs=None):
2033
 
    global NUMBERED_DIRS
2034
 
    if numbered_dirs is not None:
2035
 
        NUMBERED_DIRS = bool(numbered_dirs)
2036
 
 
2037
 
    TestCase._gather_lsprof_in_benchmarks = lsprof_timed
2038
 
    if verbose:
2039
 
        verbosity = 2
2040
 
    else:
2041
 
        verbosity = 1
2042
 
    runner = TextTestRunner(stream=sys.stdout,
2043
 
                            descriptions=0,
2044
 
                            verbosity=verbosity,
2045
 
                            keep_output=keep_output,
2046
 
                            bench_history=bench_history)
2047
 
    runner.stop_on_failure=stop_on_failure
2048
 
    if pattern != '.*':
2049
 
        if matching_tests_first:
2050
 
            suite = sort_suite_by_re(suite, pattern)
2051
 
        else:
2052
 
            suite = filter_suite_by_re(suite, pattern)
2053
 
    result = runner.run(suite)
2054
 
    return result.wasSuccessful()
2055
 
 
2056
 
 
2057
 
def selftest(verbose=False, pattern=".*", stop_on_failure=True,
2058
 
             keep_output=False,
2059
 
             transport=None,
2060
 
             test_suite_factory=None,
2061
 
             lsprof_timed=None,
2062
 
             bench_history=None,
2063
 
             matching_tests_first=None,
2064
 
             numbered_dirs=None):
2065
 
    """Run the whole test suite under the enhanced runner"""
2066
 
    # XXX: Very ugly way to do this...
2067
 
    # Disable warning about old formats because we don't want it to disturb
2068
 
    # any blackbox tests.
2069
 
    from bzrlib import repository
2070
 
    repository._deprecation_warning_done = True
2071
 
 
2072
 
    global default_transport
2073
 
    if transport is None:
2074
 
        transport = default_transport
2075
 
    old_transport = default_transport
2076
 
    default_transport = transport
2077
 
    try:
2078
 
        if test_suite_factory is None:
2079
 
            suite = test_suite()
2080
 
        else:
2081
 
            suite = test_suite_factory()
2082
 
        return run_suite(suite, 'testbzr', verbose=verbose, pattern=pattern,
2083
 
                     stop_on_failure=stop_on_failure, keep_output=keep_output,
2084
 
                     transport=transport,
2085
 
                     lsprof_timed=lsprof_timed,
2086
 
                     bench_history=bench_history,
2087
 
                     matching_tests_first=matching_tests_first,
2088
 
                     numbered_dirs=numbered_dirs)
2089
 
    finally:
2090
 
        default_transport = old_transport
2091
 
 
2092
 
 
2093
 
def test_suite():
2094
 
    """Build and return TestSuite for the whole of bzrlib.
2095
 
    
2096
 
    This function can be replaced if you need to change the default test
2097
 
    suite on a global basis, but it is not encouraged.
2098
 
    """
2099
 
    testmod_names = [
2100
 
                   'bzrlib.tests.test_ancestry',
2101
 
                   'bzrlib.tests.test_annotate',
2102
 
                   'bzrlib.tests.test_api',
2103
 
                   'bzrlib.tests.test_atomicfile',
2104
 
                   'bzrlib.tests.test_bad_files',
2105
 
                   'bzrlib.tests.test_branch',
2106
 
                   'bzrlib.tests.test_bundle',
2107
 
                   'bzrlib.tests.test_bzrdir',
2108
 
                   'bzrlib.tests.test_cache_utf8',
2109
 
                   'bzrlib.tests.test_commands',
2110
 
                   'bzrlib.tests.test_commit',
2111
 
                   'bzrlib.tests.test_commit_merge',
2112
 
                   'bzrlib.tests.test_config',
2113
 
                   'bzrlib.tests.test_conflicts',
2114
 
                   'bzrlib.tests.test_decorators',
2115
 
                   'bzrlib.tests.test_delta',
2116
 
                   'bzrlib.tests.test_diff',
2117
 
                   'bzrlib.tests.test_dirstate',
2118
 
                   'bzrlib.tests.test_doc_generate',
2119
 
                   'bzrlib.tests.test_errors',
2120
 
                   'bzrlib.tests.test_escaped_store',
2121
 
                   'bzrlib.tests.test_extract',
2122
 
                   'bzrlib.tests.test_fetch',
2123
 
                   'bzrlib.tests.test_ftp_transport',
2124
 
                   'bzrlib.tests.test_generate_docs',
2125
 
                   'bzrlib.tests.test_generate_ids',
2126
 
                   'bzrlib.tests.test_globbing',
2127
 
                   'bzrlib.tests.test_gpg',
2128
 
                   'bzrlib.tests.test_graph',
2129
 
                   'bzrlib.tests.test_hashcache',
2130
 
                   'bzrlib.tests.test_http',
2131
 
                   'bzrlib.tests.test_http_response',
2132
 
                   'bzrlib.tests.test_https_ca_bundle',
2133
 
                   'bzrlib.tests.test_identitymap',
2134
 
                   'bzrlib.tests.test_ignores',
2135
 
                   'bzrlib.tests.test_inv',
2136
 
                   'bzrlib.tests.test_knit',
2137
 
                   'bzrlib.tests.test_lazy_import',
2138
 
                   'bzrlib.tests.test_lazy_regex',
2139
 
                   'bzrlib.tests.test_lockdir',
2140
 
                   'bzrlib.tests.test_lockable_files',
2141
 
                   'bzrlib.tests.test_log',
2142
 
                   'bzrlib.tests.test_memorytree',
2143
 
                   'bzrlib.tests.test_merge',
2144
 
                   'bzrlib.tests.test_merge3',
2145
 
                   'bzrlib.tests.test_merge_core',
2146
 
                   'bzrlib.tests.test_merge_directive',
2147
 
                   'bzrlib.tests.test_missing',
2148
 
                   'bzrlib.tests.test_msgeditor',
2149
 
                   'bzrlib.tests.test_nonascii',
2150
 
                   'bzrlib.tests.test_options',
2151
 
                   'bzrlib.tests.test_osutils',
2152
 
                   'bzrlib.tests.test_osutils_encodings',
2153
 
                   'bzrlib.tests.test_patch',
2154
 
                   'bzrlib.tests.test_patches',
2155
 
                   'bzrlib.tests.test_permissions',
2156
 
                   'bzrlib.tests.test_plugins',
2157
 
                   'bzrlib.tests.test_progress',
2158
 
                   'bzrlib.tests.test_reconcile',
2159
 
                   'bzrlib.tests.test_registry',
2160
 
                   'bzrlib.tests.test_repository',
2161
 
                   'bzrlib.tests.test_revert',
2162
 
                   'bzrlib.tests.test_revision',
2163
 
                   'bzrlib.tests.test_revisionnamespaces',
2164
 
                   'bzrlib.tests.test_revisiontree',
2165
 
                   'bzrlib.tests.test_rio',
2166
 
                   'bzrlib.tests.test_sampler',
2167
 
                   'bzrlib.tests.test_selftest',
2168
 
                   'bzrlib.tests.test_setup',
2169
 
                   'bzrlib.tests.test_sftp_transport',
2170
 
                   'bzrlib.tests.test_smart_add',
2171
 
                   'bzrlib.tests.test_smart_transport',
2172
 
                   'bzrlib.tests.test_source',
2173
 
                   'bzrlib.tests.test_ssh_transport',
2174
 
                   'bzrlib.tests.test_status',
2175
 
                   'bzrlib.tests.test_store',
2176
 
                   'bzrlib.tests.test_strace',
2177
 
                   'bzrlib.tests.test_subsume',
2178
 
                   'bzrlib.tests.test_symbol_versioning',
2179
 
                   'bzrlib.tests.test_tag',
2180
 
                   'bzrlib.tests.test_testament',
2181
 
                   'bzrlib.tests.test_textfile',
2182
 
                   'bzrlib.tests.test_textmerge',
2183
 
                   'bzrlib.tests.test_timestamp',
2184
 
                   'bzrlib.tests.test_trace',
2185
 
                   'bzrlib.tests.test_transactions',
2186
 
                   'bzrlib.tests.test_transform',
2187
 
                   'bzrlib.tests.test_transport',
2188
 
                   'bzrlib.tests.test_tree',
2189
 
                   'bzrlib.tests.test_treebuilder',
2190
 
                   'bzrlib.tests.test_tsort',
2191
 
                   'bzrlib.tests.test_tuned_gzip',
2192
 
                   'bzrlib.tests.test_ui',
2193
 
                   'bzrlib.tests.test_upgrade',
2194
 
                   'bzrlib.tests.test_urlutils',
2195
 
                   'bzrlib.tests.test_versionedfile',
2196
 
                   'bzrlib.tests.test_version',
2197
 
                   'bzrlib.tests.test_version_info',
2198
 
                   'bzrlib.tests.test_weave',
2199
 
                   'bzrlib.tests.test_whitebox',
2200
 
                   'bzrlib.tests.test_workingtree',
2201
 
                   'bzrlib.tests.test_workingtree_4',
2202
 
                   'bzrlib.tests.test_wsgi',
2203
 
                   'bzrlib.tests.test_xml',
2204
 
                   ]
2205
 
    test_transport_implementations = [
2206
 
        'bzrlib.tests.test_transport_implementations',
2207
 
        'bzrlib.tests.test_read_bundle',
2208
 
        ]
2209
 
    suite = TestUtil.TestSuite()
2210
 
    loader = TestUtil.TestLoader()
2211
 
    suite.addTest(loader.loadTestsFromModuleNames(testmod_names))
2212
 
    from bzrlib.transport import TransportTestProviderAdapter
2213
 
    adapter = TransportTestProviderAdapter()
2214
 
    adapt_modules(test_transport_implementations, adapter, loader, suite)
2215
 
    for package in packages_to_test():
2216
 
        suite.addTest(package.test_suite())
2217
 
    for m in MODULES_TO_TEST:
2218
 
        suite.addTest(loader.loadTestsFromModule(m))
2219
 
    for m in MODULES_TO_DOCTEST:
2220
 
        try:
2221
 
            suite.addTest(doctest.DocTestSuite(m))
2222
 
        except ValueError, e:
2223
 
            print '**failed to get doctest for: %s\n%s' %(m,e)
2224
 
            raise
2225
 
    for name, plugin in bzrlib.plugin.all_plugins().items():
2226
 
        if getattr(plugin, 'test_suite', None) is not None:
2227
 
            default_encoding = sys.getdefaultencoding()
2228
 
            try:
2229
 
                plugin_suite = plugin.test_suite()
2230
 
            except ImportError, e:
2231
 
                bzrlib.trace.warning(
2232
 
                    'Unable to test plugin "%s": %s', name, e)
2233
 
            else:
2234
 
                suite.addTest(plugin_suite)
2235
 
            if default_encoding != sys.getdefaultencoding():
2236
 
                bzrlib.trace.warning(
2237
 
                    'Plugin "%s" tried to reset default encoding to: %s', name,
2238
 
                    sys.getdefaultencoding())
2239
 
                reload(sys)
2240
 
                sys.setdefaultencoding(default_encoding)
2241
 
    return suite
2242
 
 
2243
 
 
2244
 
def adapt_modules(mods_list, adapter, loader, suite):
2245
 
    """Adapt the modules in mods_list using adapter and add to suite."""
2246
 
    for test in iter_suite_tests(loader.loadTestsFromModuleNames(mods_list)):
2247
 
        suite.addTests(adapter.adapt(test))
2248
 
 
2249
 
 
2250
 
def clean_selftest_output(root=None, quiet=False):
2251
 
    """Remove all selftest output directories from root directory.
2252
 
 
2253
 
    :param  root:   root directory for clean
2254
 
                    (if ommitted or None then clean current directory).
2255
 
    :param  quiet:  suppress report about deleting directories
2256
 
    """
2257
 
    import re
2258
 
    import shutil
2259
 
 
2260
 
    re_dir = re.compile(r'''test\d\d\d\d\.tmp''')
2261
 
    if root is None:
2262
 
        root = u'.'
2263
 
    for i in os.listdir(root):
2264
 
        if os.path.isdir(i) and re_dir.match(i):
2265
 
            if not quiet:
2266
 
                print 'delete directory:', i
2267
 
            shutil.rmtree(i)
2268
 
 
2269
 
 
2270
 
class Feature(object):
2271
 
    """An operating system Feature."""
2272
 
 
2273
 
    def __init__(self):
2274
 
        self._available = None
2275
 
 
2276
 
    def available(self):
2277
 
        """Is the feature available?
2278
 
 
2279
 
        :return: True if the feature is available.
2280
 
        """
2281
 
        if self._available is None:
2282
 
            self._available = self._probe()
2283
 
        return self._available
2284
 
 
2285
 
    def _probe(self):
2286
 
        """Implement this method in concrete features.
2287
 
 
2288
 
        :return: True if the feature is available.
2289
 
        """
2290
 
        raise NotImplementedError
2291
 
 
2292
 
    def __str__(self):
2293
 
        if getattr(self, 'feature_name', None):
2294
 
            return self.feature_name()
2295
 
        return self.__class__.__name__