/brz/remove-bazaar

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

« back to all changes in this revision

Viewing changes to bzrlib/tests/__init__.py

  • Committer: John Arbash Meinel
  • Date: 2007-04-24 19:45:05 UTC
  • mto: This revision was merged to the branch mainline in revision 2452.
  • Revision ID: john@arbash-meinel.com-20070424194505-mv5dwjw0mx4tfpfs
clean up ASCII revision graph art.

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