/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

Use ROOT_ID when the repository supports old clients (Bug #107168)

Show diffs side-by-side

added added

removed removed

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