/brz/remove-bazaar

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

« back to all changes in this revision

Viewing changes to bzrlib/tests/__init__.py

  • Committer: James Westby
  • Date: 2009-02-20 20:10:12 UTC
  • mto: This revision was merged to the branch mainline in revision 4049.
  • Revision ID: jw+debian@jameswestby.net-20090220201012-y9wonjwu4ekt7iqe
Handle files that are not present in the tree when exporting (#174539)

When export is acting on a working tree (unusual) it may encounter files
that are missing. It should not try and export these.

Show diffs side-by-side

added added

removed removed

Lines of Context:
 
1
# Copyright (C) 2005, 2006, 2007, 2008 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 atexit
 
30
import codecs
 
31
from cStringIO import StringIO
 
32
import difflib
 
33
import doctest
 
34
import errno
 
35
import logging
 
36
import os
 
37
from pprint import pformat
 
38
import random
 
39
import re
 
40
import shlex
 
41
import stat
 
42
from subprocess import Popen, PIPE
 
43
import sys
 
44
import tempfile
 
45
import threading
 
46
import time
 
47
import unittest
 
48
import warnings
 
49
 
 
50
 
 
51
from bzrlib import (
 
52
    branchbuilder,
 
53
    bzrdir,
 
54
    debug,
 
55
    errors,
 
56
    memorytree,
 
57
    osutils,
 
58
    progress,
 
59
    ui,
 
60
    urlutils,
 
61
    registry,
 
62
    workingtree,
 
63
    )
 
64
import bzrlib.branch
 
65
import bzrlib.commands
 
66
import bzrlib.timestamp
 
67
import bzrlib.export
 
68
import bzrlib.inventory
 
69
import bzrlib.iterablefile
 
70
import bzrlib.lockdir
 
71
try:
 
72
    import bzrlib.lsprof
 
73
except ImportError:
 
74
    # lsprof not available
 
75
    pass
 
76
from bzrlib.merge import merge_inner
 
77
import bzrlib.merge3
 
78
import bzrlib.plugin
 
79
import bzrlib.store
 
80
from bzrlib import symbol_versioning
 
81
from bzrlib.symbol_versioning import (
 
82
    DEPRECATED_PARAMETER,
 
83
    deprecated_function,
 
84
    deprecated_method,
 
85
    deprecated_passed,
 
86
    )
 
87
import bzrlib.trace
 
88
from bzrlib.transport import get_transport
 
89
import bzrlib.transport
 
90
from bzrlib.transport.local import LocalURLServer
 
91
from bzrlib.transport.memory import MemoryServer
 
92
from bzrlib.transport.readonly import ReadonlyServer
 
93
from bzrlib.trace import mutter, note
 
94
from bzrlib.tests import TestUtil
 
95
from bzrlib.tests.http_server import HttpServer
 
96
from bzrlib.tests.TestUtil import (
 
97
                          TestSuite,
 
98
                          TestLoader,
 
99
                          )
 
100
from bzrlib.tests.treeshape import build_tree_contents
 
101
import bzrlib.version_info_formats.format_custom
 
102
from bzrlib.workingtree import WorkingTree, WorkingTreeFormat2
 
103
 
 
104
# Mark this python module as being part of the implementation
 
105
# of unittest: this gives us better tracebacks where the last
 
106
# shown frame is the test code, not our assertXYZ.
 
107
__unittest = 1
 
108
 
 
109
default_transport = LocalURLServer
 
110
 
 
111
 
 
112
class ExtendedTestResult(unittest._TextTestResult):
 
113
    """Accepts, reports and accumulates the results of running tests.
 
114
 
 
115
    Compared to the unittest version this class adds support for
 
116
    profiling, benchmarking, stopping as soon as a test fails,  and
 
117
    skipping tests.  There are further-specialized subclasses for
 
118
    different types of display.
 
119
 
 
120
    When a test finishes, in whatever way, it calls one of the addSuccess,
 
121
    addFailure or addError classes.  These in turn may redirect to a more
 
122
    specific case for the special test results supported by our extended
 
123
    tests.
 
124
 
 
125
    Note that just one of these objects is fed the results from many tests.
 
126
    """
 
127
 
 
128
    stop_early = False
 
129
    
 
130
    def __init__(self, stream, descriptions, verbosity,
 
131
                 bench_history=None,
 
132
                 num_tests=None,
 
133
                 ):
 
134
        """Construct new TestResult.
 
135
 
 
136
        :param bench_history: Optionally, a writable file object to accumulate
 
137
            benchmark results.
 
138
        """
 
139
        unittest._TextTestResult.__init__(self, stream, descriptions, verbosity)
 
140
        if bench_history is not None:
 
141
            from bzrlib.version import _get_bzr_source_tree
 
142
            src_tree = _get_bzr_source_tree()
 
143
            if src_tree:
 
144
                try:
 
145
                    revision_id = src_tree.get_parent_ids()[0]
 
146
                except IndexError:
 
147
                    # XXX: if this is a brand new tree, do the same as if there
 
148
                    # is no branch.
 
149
                    revision_id = ''
 
150
            else:
 
151
                # XXX: If there's no branch, what should we do?
 
152
                revision_id = ''
 
153
            bench_history.write("--date %s %s\n" % (time.time(), revision_id))
 
154
        self._bench_history = bench_history
 
155
        self.ui = ui.ui_factory
 
156
        self.num_tests = num_tests
 
157
        self.error_count = 0
 
158
        self.failure_count = 0
 
159
        self.known_failure_count = 0
 
160
        self.skip_count = 0
 
161
        self.not_applicable_count = 0
 
162
        self.unsupported = {}
 
163
        self.count = 0
 
164
        self._overall_start_time = time.time()
 
165
    
 
166
    def _extractBenchmarkTime(self, testCase):
 
167
        """Add a benchmark time for the current test case."""
 
168
        return getattr(testCase, "_benchtime", None)
 
169
    
 
170
    def _elapsedTestTimeString(self):
 
171
        """Return a time string for the overall time the current test has taken."""
 
172
        return self._formatTime(time.time() - self._start_time)
 
173
 
 
174
    def _testTimeString(self, testCase):
 
175
        benchmark_time = self._extractBenchmarkTime(testCase)
 
176
        if benchmark_time is not None:
 
177
            return "%s/%s" % (
 
178
                self._formatTime(benchmark_time),
 
179
                self._elapsedTestTimeString())
 
180
        else:
 
181
            return "           %s" % self._elapsedTestTimeString()
 
182
 
 
183
    def _formatTime(self, seconds):
 
184
        """Format seconds as milliseconds with leading spaces."""
 
185
        # some benchmarks can take thousands of seconds to run, so we need 8
 
186
        # places
 
187
        return "%8dms" % (1000 * seconds)
 
188
 
 
189
    def _shortened_test_description(self, test):
 
190
        what = test.id()
 
191
        what = re.sub(r'^bzrlib\.(tests|benchmarks)\.', '', what)
 
192
        return what
 
193
 
 
194
    def startTest(self, test):
 
195
        unittest.TestResult.startTest(self, test)
 
196
        self.report_test_start(test)
 
197
        test.number = self.count
 
198
        self._recordTestStartTime()
 
199
 
 
200
    def _recordTestStartTime(self):
 
201
        """Record that a test has started."""
 
202
        self._start_time = time.time()
 
203
 
 
204
    def _cleanupLogFile(self, test):
 
205
        # We can only do this if we have one of our TestCases, not if
 
206
        # we have a doctest.
 
207
        setKeepLogfile = getattr(test, 'setKeepLogfile', None)
 
208
        if setKeepLogfile is not None:
 
209
            setKeepLogfile()
 
210
 
 
211
    def addError(self, test, err):
 
212
        """Tell result that test finished with an error.
 
213
 
 
214
        Called from the TestCase run() method when the test
 
215
        fails with an unexpected error.
 
216
        """
 
217
        self._testConcluded(test)
 
218
        if isinstance(err[1], TestSkipped):
 
219
            return self._addSkipped(test, err)
 
220
        elif isinstance(err[1], UnavailableFeature):
 
221
            return self.addNotSupported(test, err[1].args[0])
 
222
        else:
 
223
            unittest.TestResult.addError(self, test, err)
 
224
            self.error_count += 1
 
225
            self.report_error(test, err)
 
226
            if self.stop_early:
 
227
                self.stop()
 
228
            self._cleanupLogFile(test)
 
229
 
 
230
    def addFailure(self, test, err):
 
231
        """Tell result that test failed.
 
232
 
 
233
        Called from the TestCase run() method when the test
 
234
        fails because e.g. an assert() method failed.
 
235
        """
 
236
        self._testConcluded(test)
 
237
        if isinstance(err[1], KnownFailure):
 
238
            return self._addKnownFailure(test, err)
 
239
        else:
 
240
            unittest.TestResult.addFailure(self, test, err)
 
241
            self.failure_count += 1
 
242
            self.report_failure(test, err)
 
243
            if self.stop_early:
 
244
                self.stop()
 
245
            self._cleanupLogFile(test)
 
246
 
 
247
    def addSuccess(self, test):
 
248
        """Tell result that test completed successfully.
 
249
 
 
250
        Called from the TestCase run()
 
251
        """
 
252
        self._testConcluded(test)
 
253
        if self._bench_history is not None:
 
254
            benchmark_time = self._extractBenchmarkTime(test)
 
255
            if benchmark_time is not None:
 
256
                self._bench_history.write("%s %s\n" % (
 
257
                    self._formatTime(benchmark_time),
 
258
                    test.id()))
 
259
        self.report_success(test)
 
260
        self._cleanupLogFile(test)
 
261
        unittest.TestResult.addSuccess(self, test)
 
262
        test._log_contents = ''
 
263
 
 
264
    def _testConcluded(self, test):
 
265
        """Common code when a test has finished.
 
266
 
 
267
        Called regardless of whether it succeded, failed, etc.
 
268
        """
 
269
        pass
 
270
 
 
271
    def _addKnownFailure(self, test, err):
 
272
        self.known_failure_count += 1
 
273
        self.report_known_failure(test, err)
 
274
 
 
275
    def addNotSupported(self, test, feature):
 
276
        """The test will not be run because of a missing feature.
 
277
        """
 
278
        # this can be called in two different ways: it may be that the
 
279
        # test started running, and then raised (through addError) 
 
280
        # UnavailableFeature.  Alternatively this method can be called
 
281
        # while probing for features before running the tests; in that
 
282
        # case we will see startTest and stopTest, but the test will never
 
283
        # actually run.
 
284
        self.unsupported.setdefault(str(feature), 0)
 
285
        self.unsupported[str(feature)] += 1
 
286
        self.report_unsupported(test, feature)
 
287
 
 
288
    def _addSkipped(self, test, skip_excinfo):
 
289
        if isinstance(skip_excinfo[1], TestNotApplicable):
 
290
            self.not_applicable_count += 1
 
291
            self.report_not_applicable(test, skip_excinfo)
 
292
        else:
 
293
            self.skip_count += 1
 
294
            self.report_skip(test, skip_excinfo)
 
295
        try:
 
296
            test.tearDown()
 
297
        except KeyboardInterrupt:
 
298
            raise
 
299
        except:
 
300
            self.addError(test, test._exc_info())
 
301
        else:
 
302
            # seems best to treat this as success from point-of-view of unittest
 
303
            # -- it actually does nothing so it barely matters :)
 
304
            unittest.TestResult.addSuccess(self, test)
 
305
            test._log_contents = ''
 
306
 
 
307
    def printErrorList(self, flavour, errors):
 
308
        for test, err in errors:
 
309
            self.stream.writeln(self.separator1)
 
310
            self.stream.write("%s: " % flavour)
 
311
            self.stream.writeln(self.getDescription(test))
 
312
            if getattr(test, '_get_log', None) is not None:
 
313
                self.stream.write('\n')
 
314
                self.stream.write(
 
315
                        ('vvvv[log from %s]' % test.id()).ljust(78,'-'))
 
316
                self.stream.write('\n')
 
317
                self.stream.write(test._get_log())
 
318
                self.stream.write('\n')
 
319
                self.stream.write(
 
320
                        ('^^^^[log from %s]' % test.id()).ljust(78,'-'))
 
321
                self.stream.write('\n')
 
322
            self.stream.writeln(self.separator2)
 
323
            self.stream.writeln("%s" % err)
 
324
 
 
325
    def finished(self):
 
326
        pass
 
327
 
 
328
    def report_cleaning_up(self):
 
329
        pass
 
330
 
 
331
    def report_success(self, test):
 
332
        pass
 
333
 
 
334
    def wasStrictlySuccessful(self):
 
335
        if self.unsupported or self.known_failure_count:
 
336
            return False
 
337
        return self.wasSuccessful()
 
338
 
 
339
 
 
340
class TextTestResult(ExtendedTestResult):
 
341
    """Displays progress and results of tests in text form"""
 
342
 
 
343
    def __init__(self, stream, descriptions, verbosity,
 
344
                 bench_history=None,
 
345
                 num_tests=None,
 
346
                 pb=None,
 
347
                 ):
 
348
        ExtendedTestResult.__init__(self, stream, descriptions, verbosity,
 
349
            bench_history, num_tests)
 
350
        if pb is None:
 
351
            self.pb = self.ui.nested_progress_bar()
 
352
            self._supplied_pb = False
 
353
        else:
 
354
            self.pb = pb
 
355
            self._supplied_pb = True
 
356
        self.pb.show_pct = False
 
357
        self.pb.show_spinner = False
 
358
        self.pb.show_eta = False,
 
359
        self.pb.show_count = False
 
360
        self.pb.show_bar = False
 
361
 
 
362
    def report_starting(self):
 
363
        self.pb.update('[test 0/%d] starting...' % (self.num_tests))
 
364
 
 
365
    def _progress_prefix_text(self):
 
366
        # the longer this text, the less space we have to show the test
 
367
        # name...
 
368
        a = '[%d' % self.count              # total that have been run
 
369
        # tests skipped as known not to be relevant are not important enough
 
370
        # to show here
 
371
        ## if self.skip_count:
 
372
        ##     a += ', %d skip' % self.skip_count
 
373
        ## if self.known_failure_count:
 
374
        ##     a += '+%dX' % self.known_failure_count
 
375
        if self.num_tests is not None:
 
376
            a +='/%d' % self.num_tests
 
377
        a += ' in '
 
378
        runtime = time.time() - self._overall_start_time
 
379
        if runtime >= 60:
 
380
            a += '%dm%ds' % (runtime / 60, runtime % 60)
 
381
        else:
 
382
            a += '%ds' % runtime
 
383
        if self.error_count:
 
384
            a += ', %d err' % self.error_count
 
385
        if self.failure_count:
 
386
            a += ', %d fail' % self.failure_count
 
387
        if self.unsupported:
 
388
            a += ', %d missing' % len(self.unsupported)
 
389
        a += ']'
 
390
        return a
 
391
 
 
392
    def report_test_start(self, test):
 
393
        self.count += 1
 
394
        self.pb.update(
 
395
                self._progress_prefix_text()
 
396
                + ' ' 
 
397
                + self._shortened_test_description(test))
 
398
 
 
399
    def _test_description(self, test):
 
400
        return self._shortened_test_description(test)
 
401
 
 
402
    def report_error(self, test, err):
 
403
        self.pb.note('ERROR: %s\n    %s\n', 
 
404
            self._test_description(test),
 
405
            err[1],
 
406
            )
 
407
 
 
408
    def report_failure(self, test, err):
 
409
        self.pb.note('FAIL: %s\n    %s\n', 
 
410
            self._test_description(test),
 
411
            err[1],
 
412
            )
 
413
 
 
414
    def report_known_failure(self, test, err):
 
415
        self.pb.note('XFAIL: %s\n%s\n',
 
416
            self._test_description(test), err[1])
 
417
 
 
418
    def report_skip(self, test, skip_excinfo):
 
419
        pass
 
420
 
 
421
    def report_not_applicable(self, test, skip_excinfo):
 
422
        pass
 
423
 
 
424
    def report_unsupported(self, test, feature):
 
425
        """test cannot be run because feature is missing."""
 
426
                  
 
427
    def report_cleaning_up(self):
 
428
        self.pb.update('cleaning up...')
 
429
 
 
430
    def finished(self):
 
431
        if not self._supplied_pb:
 
432
            self.pb.finished()
 
433
 
 
434
 
 
435
class VerboseTestResult(ExtendedTestResult):
 
436
    """Produce long output, with one line per test run plus times"""
 
437
 
 
438
    def _ellipsize_to_right(self, a_string, final_width):
 
439
        """Truncate and pad a string, keeping the right hand side"""
 
440
        if len(a_string) > final_width:
 
441
            result = '...' + a_string[3-final_width:]
 
442
        else:
 
443
            result = a_string
 
444
        return result.ljust(final_width)
 
445
 
 
446
    def report_starting(self):
 
447
        self.stream.write('running %d tests...\n' % self.num_tests)
 
448
 
 
449
    def report_test_start(self, test):
 
450
        self.count += 1
 
451
        name = self._shortened_test_description(test)
 
452
        # width needs space for 6 char status, plus 1 for slash, plus 2 10-char
 
453
        # numbers, plus a trailing blank
 
454
        # when NUMBERED_DIRS: plus 5 chars on test number, plus 1 char on space
 
455
        self.stream.write(self._ellipsize_to_right(name,
 
456
                          osutils.terminal_width()-30))
 
457
        self.stream.flush()
 
458
 
 
459
    def _error_summary(self, err):
 
460
        indent = ' ' * 4
 
461
        return '%s%s' % (indent, err[1])
 
462
 
 
463
    def report_error(self, test, err):
 
464
        self.stream.writeln('ERROR %s\n%s'
 
465
                % (self._testTimeString(test),
 
466
                   self._error_summary(err)))
 
467
 
 
468
    def report_failure(self, test, err):
 
469
        self.stream.writeln(' FAIL %s\n%s'
 
470
                % (self._testTimeString(test),
 
471
                   self._error_summary(err)))
 
472
 
 
473
    def report_known_failure(self, test, err):
 
474
        self.stream.writeln('XFAIL %s\n%s'
 
475
                % (self._testTimeString(test),
 
476
                   self._error_summary(err)))
 
477
 
 
478
    def report_success(self, test):
 
479
        self.stream.writeln('   OK %s' % self._testTimeString(test))
 
480
        for bench_called, stats in getattr(test, '_benchcalls', []):
 
481
            self.stream.writeln('LSProf output for %s(%s, %s)' % bench_called)
 
482
            stats.pprint(file=self.stream)
 
483
        # flush the stream so that we get smooth output. This verbose mode is
 
484
        # used to show the output in PQM.
 
485
        self.stream.flush()
 
486
 
 
487
    def report_skip(self, test, skip_excinfo):
 
488
        self.stream.writeln(' SKIP %s\n%s'
 
489
                % (self._testTimeString(test),
 
490
                   self._error_summary(skip_excinfo)))
 
491
 
 
492
    def report_not_applicable(self, test, skip_excinfo):
 
493
        self.stream.writeln('  N/A %s\n%s'
 
494
                % (self._testTimeString(test),
 
495
                   self._error_summary(skip_excinfo)))
 
496
 
 
497
    def report_unsupported(self, test, feature):
 
498
        """test cannot be run because feature is missing."""
 
499
        self.stream.writeln("NODEP %s\n    The feature '%s' is not available."
 
500
                %(self._testTimeString(test), feature))
 
501
 
 
502
 
 
503
class TextTestRunner(object):
 
504
    stop_on_failure = False
 
505
 
 
506
    def __init__(self,
 
507
                 stream=sys.stderr,
 
508
                 descriptions=0,
 
509
                 verbosity=1,
 
510
                 bench_history=None,
 
511
                 list_only=False
 
512
                 ):
 
513
        self.stream = unittest._WritelnDecorator(stream)
 
514
        self.descriptions = descriptions
 
515
        self.verbosity = verbosity
 
516
        self._bench_history = bench_history
 
517
        self.list_only = list_only
 
518
 
 
519
    def run(self, test):
 
520
        "Run the given test case or test suite."
 
521
        startTime = time.time()
 
522
        if self.verbosity == 1:
 
523
            result_class = TextTestResult
 
524
        elif self.verbosity >= 2:
 
525
            result_class = VerboseTestResult
 
526
        result = result_class(self.stream,
 
527
                              self.descriptions,
 
528
                              self.verbosity,
 
529
                              bench_history=self._bench_history,
 
530
                              num_tests=test.countTestCases(),
 
531
                              )
 
532
        result.stop_early = self.stop_on_failure
 
533
        result.report_starting()
 
534
        if self.list_only:
 
535
            if self.verbosity >= 2:
 
536
                self.stream.writeln("Listing tests only ...\n")
 
537
            run = 0
 
538
            for t in iter_suite_tests(test):
 
539
                self.stream.writeln("%s" % (t.id()))
 
540
                run += 1
 
541
            actionTaken = "Listed"
 
542
        else: 
 
543
            test.run(result)
 
544
            run = result.testsRun
 
545
            actionTaken = "Ran"
 
546
        stopTime = time.time()
 
547
        timeTaken = stopTime - startTime
 
548
        result.printErrors()
 
549
        self.stream.writeln(result.separator2)
 
550
        self.stream.writeln("%s %d test%s in %.3fs" % (actionTaken,
 
551
                            run, run != 1 and "s" or "", timeTaken))
 
552
        self.stream.writeln()
 
553
        if not result.wasSuccessful():
 
554
            self.stream.write("FAILED (")
 
555
            failed, errored = map(len, (result.failures, result.errors))
 
556
            if failed:
 
557
                self.stream.write("failures=%d" % failed)
 
558
            if errored:
 
559
                if failed: self.stream.write(", ")
 
560
                self.stream.write("errors=%d" % errored)
 
561
            if result.known_failure_count:
 
562
                if failed or errored: self.stream.write(", ")
 
563
                self.stream.write("known_failure_count=%d" %
 
564
                    result.known_failure_count)
 
565
            self.stream.writeln(")")
 
566
        else:
 
567
            if result.known_failure_count:
 
568
                self.stream.writeln("OK (known_failures=%d)" %
 
569
                    result.known_failure_count)
 
570
            else:
 
571
                self.stream.writeln("OK")
 
572
        if result.skip_count > 0:
 
573
            skipped = result.skip_count
 
574
            self.stream.writeln('%d test%s skipped' %
 
575
                                (skipped, skipped != 1 and "s" or ""))
 
576
        if result.unsupported:
 
577
            for feature, count in sorted(result.unsupported.items()):
 
578
                self.stream.writeln("Missing feature '%s' skipped %d tests." %
 
579
                    (feature, count))
 
580
        result.finished()
 
581
        return result
 
582
 
 
583
 
 
584
def iter_suite_tests(suite):
 
585
    """Return all tests in a suite, recursing through nested suites"""
 
586
    for item in suite._tests:
 
587
        if isinstance(item, unittest.TestCase):
 
588
            yield item
 
589
        elif isinstance(item, unittest.TestSuite):
 
590
            for r in iter_suite_tests(item):
 
591
                yield r
 
592
        else:
 
593
            raise Exception('unknown object %r inside test suite %r'
 
594
                            % (item, suite))
 
595
 
 
596
 
 
597
class TestSkipped(Exception):
 
598
    """Indicates that a test was intentionally skipped, rather than failing."""
 
599
 
 
600
 
 
601
class TestNotApplicable(TestSkipped):
 
602
    """A test is not applicable to the situation where it was run.
 
603
 
 
604
    This is only normally raised by parameterized tests, if they find that 
 
605
    the instance they're constructed upon does not support one aspect 
 
606
    of its interface.
 
607
    """
 
608
 
 
609
 
 
610
class KnownFailure(AssertionError):
 
611
    """Indicates that a test failed in a precisely expected manner.
 
612
 
 
613
    Such failures dont block the whole test suite from passing because they are
 
614
    indicators of partially completed code or of future work. We have an
 
615
    explicit error for them so that we can ensure that they are always visible:
 
616
    KnownFailures are always shown in the output of bzr selftest.
 
617
    """
 
618
 
 
619
 
 
620
class UnavailableFeature(Exception):
 
621
    """A feature required for this test was not available.
 
622
 
 
623
    The feature should be used to construct the exception.
 
624
    """
 
625
 
 
626
 
 
627
class CommandFailed(Exception):
 
628
    pass
 
629
 
 
630
 
 
631
class StringIOWrapper(object):
 
632
    """A wrapper around cStringIO which just adds an encoding attribute.
 
633
    
 
634
    Internally we can check sys.stdout to see what the output encoding
 
635
    should be. However, cStringIO has no encoding attribute that we can
 
636
    set. So we wrap it instead.
 
637
    """
 
638
    encoding='ascii'
 
639
    _cstring = None
 
640
 
 
641
    def __init__(self, s=None):
 
642
        if s is not None:
 
643
            self.__dict__['_cstring'] = StringIO(s)
 
644
        else:
 
645
            self.__dict__['_cstring'] = StringIO()
 
646
 
 
647
    def __getattr__(self, name, getattr=getattr):
 
648
        return getattr(self.__dict__['_cstring'], name)
 
649
 
 
650
    def __setattr__(self, name, val):
 
651
        if name == 'encoding':
 
652
            self.__dict__['encoding'] = val
 
653
        else:
 
654
            return setattr(self._cstring, name, val)
 
655
 
 
656
 
 
657
class TestUIFactory(ui.CLIUIFactory):
 
658
    """A UI Factory for testing.
 
659
 
 
660
    Hide the progress bar but emit note()s.
 
661
    Redirect stdin.
 
662
    Allows get_password to be tested without real tty attached.
 
663
    """
 
664
 
 
665
    def __init__(self,
 
666
                 stdout=None,
 
667
                 stderr=None,
 
668
                 stdin=None):
 
669
        super(TestUIFactory, self).__init__()
 
670
        if stdin is not None:
 
671
            # We use a StringIOWrapper to be able to test various
 
672
            # encodings, but the user is still responsible to
 
673
            # encode the string and to set the encoding attribute
 
674
            # of StringIOWrapper.
 
675
            self.stdin = StringIOWrapper(stdin)
 
676
        if stdout is None:
 
677
            self.stdout = sys.stdout
 
678
        else:
 
679
            self.stdout = stdout
 
680
        if stderr is None:
 
681
            self.stderr = sys.stderr
 
682
        else:
 
683
            self.stderr = stderr
 
684
 
 
685
    def clear(self):
 
686
        """See progress.ProgressBar.clear()."""
 
687
 
 
688
    def clear_term(self):
 
689
        """See progress.ProgressBar.clear_term()."""
 
690
 
 
691
    def clear_term(self):
 
692
        """See progress.ProgressBar.clear_term()."""
 
693
 
 
694
    def finished(self):
 
695
        """See progress.ProgressBar.finished()."""
 
696
 
 
697
    def note(self, fmt_string, *args, **kwargs):
 
698
        """See progress.ProgressBar.note()."""
 
699
        self.stdout.write((fmt_string + "\n") % args)
 
700
 
 
701
    def progress_bar(self):
 
702
        return self
 
703
 
 
704
    def nested_progress_bar(self):
 
705
        return self
 
706
 
 
707
    def update(self, message, count=None, total=None):
 
708
        """See progress.ProgressBar.update()."""
 
709
 
 
710
    def get_non_echoed_password(self, prompt):
 
711
        """Get password from stdin without trying to handle the echo mode"""
 
712
        if prompt:
 
713
            self.stdout.write(prompt.encode(self.stdout.encoding, 'replace'))
 
714
        password = self.stdin.readline()
 
715
        if not password:
 
716
            raise EOFError
 
717
        if password[-1] == '\n':
 
718
            password = password[:-1]
 
719
        return password
 
720
 
 
721
 
 
722
def _report_leaked_threads():
 
723
    bzrlib.trace.warning('%s is leaking threads among %d leaking tests',
 
724
                         TestCase._first_thread_leaker_id,
 
725
                         TestCase._leaking_threads_tests)
 
726
 
 
727
 
 
728
class TestCase(unittest.TestCase):
 
729
    """Base class for bzr unit tests.
 
730
    
 
731
    Tests that need access to disk resources should subclass 
 
732
    TestCaseInTempDir not TestCase.
 
733
 
 
734
    Error and debug log messages are redirected from their usual
 
735
    location into a temporary file, the contents of which can be
 
736
    retrieved by _get_log().  We use a real OS file, not an in-memory object,
 
737
    so that it can also capture file IO.  When the test completes this file
 
738
    is read into memory and removed from disk.
 
739
       
 
740
    There are also convenience functions to invoke bzr's command-line
 
741
    routine, and to build and check bzr trees.
 
742
   
 
743
    In addition to the usual method of overriding tearDown(), this class also
 
744
    allows subclasses to register functions into the _cleanups list, which is
 
745
    run in order as the object is torn down.  It's less likely this will be
 
746
    accidentally overlooked.
 
747
    """
 
748
 
 
749
    _active_threads = None
 
750
    _leaking_threads_tests = 0
 
751
    _first_thread_leaker_id = None
 
752
    _log_file_name = None
 
753
    _log_contents = ''
 
754
    _keep_log_file = False
 
755
    # record lsprof data when performing benchmark calls.
 
756
    _gather_lsprof_in_benchmarks = False
 
757
    attrs_to_keep = ('id', '_testMethodName', '_testMethodDoc',
 
758
                     '_log_contents', '_log_file_name', '_benchtime',
 
759
                     '_TestCase__testMethodName')
 
760
 
 
761
    def __init__(self, methodName='testMethod'):
 
762
        super(TestCase, self).__init__(methodName)
 
763
        self._cleanups = []
 
764
 
 
765
    def setUp(self):
 
766
        unittest.TestCase.setUp(self)
 
767
        self._cleanEnvironment()
 
768
        self._silenceUI()
 
769
        self._startLogFile()
 
770
        self._benchcalls = []
 
771
        self._benchtime = None
 
772
        self._clear_hooks()
 
773
        self._clear_debug_flags()
 
774
        TestCase._active_threads = threading.activeCount()
 
775
        self.addCleanup(self._check_leaked_threads)
 
776
 
 
777
    def _check_leaked_threads(self):
 
778
        active = threading.activeCount()
 
779
        leaked_threads = active - TestCase._active_threads
 
780
        TestCase._active_threads = active
 
781
        if leaked_threads:
 
782
            TestCase._leaking_threads_tests += 1
 
783
            if TestCase._first_thread_leaker_id is None:
 
784
                TestCase._first_thread_leaker_id = self.id()
 
785
                # we're not specifically told when all tests are finished.
 
786
                # This will do. We use a function to avoid keeping a reference
 
787
                # to a TestCase object.
 
788
                atexit.register(_report_leaked_threads)
 
789
 
 
790
    def _clear_debug_flags(self):
 
791
        """Prevent externally set debug flags affecting tests.
 
792
 
 
793
        Tests that want to use debug flags can just set them in the
 
794
        debug_flags set during setup/teardown.
 
795
        """
 
796
        self._preserved_debug_flags = set(debug.debug_flags)
 
797
        if 'allow_debug' not in selftest_debug_flags:
 
798
            debug.debug_flags.clear()
 
799
        self.addCleanup(self._restore_debug_flags)
 
800
 
 
801
    def _clear_hooks(self):
 
802
        # prevent hooks affecting tests
 
803
        import bzrlib.branch
 
804
        import bzrlib.smart.client
 
805
        import bzrlib.smart.server
 
806
        self._preserved_hooks = {
 
807
            bzrlib.branch.Branch: bzrlib.branch.Branch.hooks,
 
808
            bzrlib.mutabletree.MutableTree: bzrlib.mutabletree.MutableTree.hooks,
 
809
            bzrlib.smart.client._SmartClient: bzrlib.smart.client._SmartClient.hooks,
 
810
            bzrlib.smart.server.SmartTCPServer: bzrlib.smart.server.SmartTCPServer.hooks,
 
811
            bzrlib.commands.Command: bzrlib.commands.Command.hooks,
 
812
            }
 
813
        self.addCleanup(self._restoreHooks)
 
814
        # reset all hooks to an empty instance of the appropriate type
 
815
        bzrlib.branch.Branch.hooks = bzrlib.branch.BranchHooks()
 
816
        bzrlib.smart.client._SmartClient.hooks = bzrlib.smart.client.SmartClientHooks()
 
817
        bzrlib.smart.server.SmartTCPServer.hooks = bzrlib.smart.server.SmartServerHooks()
 
818
        bzrlib.commands.Command.hooks = bzrlib.commands.CommandHooks()
 
819
 
 
820
    def _silenceUI(self):
 
821
        """Turn off UI for duration of test"""
 
822
        # by default the UI is off; tests can turn it on if they want it.
 
823
        saved = ui.ui_factory
 
824
        def _restore():
 
825
            ui.ui_factory = saved
 
826
        ui.ui_factory = ui.SilentUIFactory()
 
827
        self.addCleanup(_restore)
 
828
 
 
829
    def _ndiff_strings(self, a, b):
 
830
        """Return ndiff between two strings containing lines.
 
831
        
 
832
        A trailing newline is added if missing to make the strings
 
833
        print properly."""
 
834
        if b and b[-1] != '\n':
 
835
            b += '\n'
 
836
        if a and a[-1] != '\n':
 
837
            a += '\n'
 
838
        difflines = difflib.ndiff(a.splitlines(True),
 
839
                                  b.splitlines(True),
 
840
                                  linejunk=lambda x: False,
 
841
                                  charjunk=lambda x: False)
 
842
        return ''.join(difflines)
 
843
 
 
844
    def assertEqual(self, a, b, message=''):
 
845
        try:
 
846
            if a == b:
 
847
                return
 
848
        except UnicodeError, e:
 
849
            # If we can't compare without getting a UnicodeError, then
 
850
            # obviously they are different
 
851
            mutter('UnicodeError: %s', e)
 
852
        if message:
 
853
            message += '\n'
 
854
        raise AssertionError("%snot equal:\na = %s\nb = %s\n"
 
855
            % (message,
 
856
               pformat(a), pformat(b)))
 
857
 
 
858
    assertEquals = assertEqual
 
859
 
 
860
    def assertEqualDiff(self, a, b, message=None):
 
861
        """Assert two texts are equal, if not raise an exception.
 
862
        
 
863
        This is intended for use with multi-line strings where it can 
 
864
        be hard to find the differences by eye.
 
865
        """
 
866
        # TODO: perhaps override assertEquals to call this for strings?
 
867
        if a == b:
 
868
            return
 
869
        if message is None:
 
870
            message = "texts not equal:\n"
 
871
        if a == b + '\n':
 
872
            message = 'first string is missing a final newline.\n'
 
873
        if a + '\n' == b:
 
874
            message = 'second string is missing a final newline.\n'
 
875
        raise AssertionError(message +
 
876
                             self._ndiff_strings(a, b))
 
877
        
 
878
    def assertEqualMode(self, mode, mode_test):
 
879
        self.assertEqual(mode, mode_test,
 
880
                         'mode mismatch %o != %o' % (mode, mode_test))
 
881
 
 
882
    def assertEqualStat(self, expected, actual):
 
883
        """assert that expected and actual are the same stat result.
 
884
 
 
885
        :param expected: A stat result.
 
886
        :param actual: A stat result.
 
887
        :raises AssertionError: If the expected and actual stat values differ
 
888
            other than by atime.
 
889
        """
 
890
        self.assertEqual(expected.st_size, actual.st_size)
 
891
        self.assertEqual(expected.st_mtime, actual.st_mtime)
 
892
        self.assertEqual(expected.st_ctime, actual.st_ctime)
 
893
        self.assertEqual(expected.st_dev, actual.st_dev)
 
894
        self.assertEqual(expected.st_ino, actual.st_ino)
 
895
        self.assertEqual(expected.st_mode, actual.st_mode)
 
896
 
 
897
    def assertPositive(self, val):
 
898
        """Assert that val is greater than 0."""
 
899
        self.assertTrue(val > 0, 'expected a positive value, but got %s' % val)
 
900
 
 
901
    def assertNegative(self, val):
 
902
        """Assert that val is less than 0."""
 
903
        self.assertTrue(val < 0, 'expected a negative value, but got %s' % val)
 
904
 
 
905
    def assertStartsWith(self, s, prefix):
 
906
        if not s.startswith(prefix):
 
907
            raise AssertionError('string %r does not start with %r' % (s, prefix))
 
908
 
 
909
    def assertEndsWith(self, s, suffix):
 
910
        """Asserts that s ends with suffix."""
 
911
        if not s.endswith(suffix):
 
912
            raise AssertionError('string %r does not end with %r' % (s, suffix))
 
913
 
 
914
    def assertContainsRe(self, haystack, needle_re, flags=0):
 
915
        """Assert that a contains something matching a regular expression."""
 
916
        if not re.search(needle_re, haystack, flags):
 
917
            if '\n' in haystack or len(haystack) > 60:
 
918
                # a long string, format it in a more readable way
 
919
                raise AssertionError(
 
920
                        'pattern "%s" not found in\n"""\\\n%s"""\n'
 
921
                        % (needle_re, haystack))
 
922
            else:
 
923
                raise AssertionError('pattern "%s" not found in "%s"'
 
924
                        % (needle_re, haystack))
 
925
 
 
926
    def assertNotContainsRe(self, haystack, needle_re, flags=0):
 
927
        """Assert that a does not match a regular expression"""
 
928
        if re.search(needle_re, haystack, flags):
 
929
            raise AssertionError('pattern "%s" found in "%s"'
 
930
                    % (needle_re, haystack))
 
931
 
 
932
    def assertSubset(self, sublist, superlist):
 
933
        """Assert that every entry in sublist is present in superlist."""
 
934
        missing = set(sublist) - set(superlist)
 
935
        if len(missing) > 0:
 
936
            raise AssertionError("value(s) %r not present in container %r" %
 
937
                                 (missing, superlist))
 
938
 
 
939
    def assertListRaises(self, excClass, func, *args, **kwargs):
 
940
        """Fail unless excClass is raised when the iterator from func is used.
 
941
        
 
942
        Many functions can return generators this makes sure
 
943
        to wrap them in a list() call to make sure the whole generator
 
944
        is run, and that the proper exception is raised.
 
945
        """
 
946
        try:
 
947
            list(func(*args, **kwargs))
 
948
        except excClass, e:
 
949
            return e
 
950
        else:
 
951
            if getattr(excClass,'__name__', None) is not None:
 
952
                excName = excClass.__name__
 
953
            else:
 
954
                excName = str(excClass)
 
955
            raise self.failureException, "%s not raised" % excName
 
956
 
 
957
    def assertRaises(self, excClass, callableObj, *args, **kwargs):
 
958
        """Assert that a callable raises a particular exception.
 
959
 
 
960
        :param excClass: As for the except statement, this may be either an
 
961
            exception class, or a tuple of classes.
 
962
        :param callableObj: A callable, will be passed ``*args`` and
 
963
            ``**kwargs``.
 
964
 
 
965
        Returns the exception so that you can examine it.
 
966
        """
 
967
        try:
 
968
            callableObj(*args, **kwargs)
 
969
        except excClass, e:
 
970
            return e
 
971
        else:
 
972
            if getattr(excClass,'__name__', None) is not None:
 
973
                excName = excClass.__name__
 
974
            else:
 
975
                # probably a tuple
 
976
                excName = str(excClass)
 
977
            raise self.failureException, "%s not raised" % excName
 
978
 
 
979
    def assertIs(self, left, right, message=None):
 
980
        if not (left is right):
 
981
            if message is not None:
 
982
                raise AssertionError(message)
 
983
            else:
 
984
                raise AssertionError("%r is not %r." % (left, right))
 
985
 
 
986
    def assertIsNot(self, left, right, message=None):
 
987
        if (left is right):
 
988
            if message is not None:
 
989
                raise AssertionError(message)
 
990
            else:
 
991
                raise AssertionError("%r is %r." % (left, right))
 
992
 
 
993
    def assertTransportMode(self, transport, path, mode):
 
994
        """Fail if a path does not have mode mode.
 
995
        
 
996
        If modes are not supported on this transport, the assertion is ignored.
 
997
        """
 
998
        if not transport._can_roundtrip_unix_modebits():
 
999
            return
 
1000
        path_stat = transport.stat(path)
 
1001
        actual_mode = stat.S_IMODE(path_stat.st_mode)
 
1002
        self.assertEqual(mode, actual_mode,
 
1003
            'mode of %r incorrect (%o != %o)' % (path, mode, actual_mode))
 
1004
 
 
1005
    def assertIsSameRealPath(self, path1, path2):
 
1006
        """Fail if path1 and path2 points to different files"""
 
1007
        self.assertEqual(osutils.realpath(path1),
 
1008
                         osutils.realpath(path2),
 
1009
                         "apparent paths:\na = %s\nb = %s\n," % (path1, path2))
 
1010
 
 
1011
    def assertIsInstance(self, obj, kls):
 
1012
        """Fail if obj is not an instance of kls"""
 
1013
        if not isinstance(obj, kls):
 
1014
            self.fail("%r is an instance of %s rather than %s" % (
 
1015
                obj, obj.__class__, kls))
 
1016
 
 
1017
    def expectFailure(self, reason, assertion, *args, **kwargs):
 
1018
        """Invoke a test, expecting it to fail for the given reason.
 
1019
 
 
1020
        This is for assertions that ought to succeed, but currently fail.
 
1021
        (The failure is *expected* but not *wanted*.)  Please be very precise
 
1022
        about the failure you're expecting.  If a new bug is introduced,
 
1023
        AssertionError should be raised, not KnownFailure.
 
1024
 
 
1025
        Frequently, expectFailure should be followed by an opposite assertion.
 
1026
        See example below.
 
1027
 
 
1028
        Intended to be used with a callable that raises AssertionError as the
 
1029
        'assertion' parameter.  args and kwargs are passed to the 'assertion'.
 
1030
 
 
1031
        Raises KnownFailure if the test fails.  Raises AssertionError if the
 
1032
        test succeeds.
 
1033
 
 
1034
        example usage::
 
1035
 
 
1036
          self.expectFailure('Math is broken', self.assertNotEqual, 54,
 
1037
                             dynamic_val)
 
1038
          self.assertEqual(42, dynamic_val)
 
1039
 
 
1040
          This means that a dynamic_val of 54 will cause the test to raise
 
1041
          a KnownFailure.  Once math is fixed and the expectFailure is removed,
 
1042
          only a dynamic_val of 42 will allow the test to pass.  Anything other
 
1043
          than 54 or 42 will cause an AssertionError.
 
1044
        """
 
1045
        try:
 
1046
            assertion(*args, **kwargs)
 
1047
        except AssertionError:
 
1048
            raise KnownFailure(reason)
 
1049
        else:
 
1050
            self.fail('Unexpected success.  Should have failed: %s' % reason)
 
1051
 
 
1052
    def assertFileEqual(self, content, path):
 
1053
        """Fail if path does not contain 'content'."""
 
1054
        self.failUnlessExists(path)
 
1055
        f = file(path, 'rb')
 
1056
        try:
 
1057
            s = f.read()
 
1058
        finally:
 
1059
            f.close()
 
1060
        self.assertEqualDiff(content, s)
 
1061
 
 
1062
    def failUnlessExists(self, path):
 
1063
        """Fail unless path or paths, which may be abs or relative, exist."""
 
1064
        if not isinstance(path, basestring):
 
1065
            for p in path:
 
1066
                self.failUnlessExists(p)
 
1067
        else:
 
1068
            self.failUnless(osutils.lexists(path),path+" does not exist")
 
1069
 
 
1070
    def failIfExists(self, path):
 
1071
        """Fail if path or paths, which may be abs or relative, exist."""
 
1072
        if not isinstance(path, basestring):
 
1073
            for p in path:
 
1074
                self.failIfExists(p)
 
1075
        else:
 
1076
            self.failIf(osutils.lexists(path),path+" exists")
 
1077
 
 
1078
    def _capture_deprecation_warnings(self, a_callable, *args, **kwargs):
 
1079
        """A helper for callDeprecated and applyDeprecated.
 
1080
 
 
1081
        :param a_callable: A callable to call.
 
1082
        :param args: The positional arguments for the callable
 
1083
        :param kwargs: The keyword arguments for the callable
 
1084
        :return: A tuple (warnings, result). result is the result of calling
 
1085
            a_callable(``*args``, ``**kwargs``).
 
1086
        """
 
1087
        local_warnings = []
 
1088
        def capture_warnings(msg, cls=None, stacklevel=None):
 
1089
            # we've hooked into a deprecation specific callpath,
 
1090
            # only deprecations should getting sent via it.
 
1091
            self.assertEqual(cls, DeprecationWarning)
 
1092
            local_warnings.append(msg)
 
1093
        original_warning_method = symbol_versioning.warn
 
1094
        symbol_versioning.set_warning_method(capture_warnings)
 
1095
        try:
 
1096
            result = a_callable(*args, **kwargs)
 
1097
        finally:
 
1098
            symbol_versioning.set_warning_method(original_warning_method)
 
1099
        return (local_warnings, result)
 
1100
 
 
1101
    def applyDeprecated(self, deprecation_format, a_callable, *args, **kwargs):
 
1102
        """Call a deprecated callable without warning the user.
 
1103
 
 
1104
        Note that this only captures warnings raised by symbol_versioning.warn,
 
1105
        not other callers that go direct to the warning module.
 
1106
 
 
1107
        To test that a deprecated method raises an error, do something like
 
1108
        this::
 
1109
 
 
1110
            self.assertRaises(errors.ReservedId,
 
1111
                self.applyDeprecated,
 
1112
                deprecated_in((1, 5, 0)),
 
1113
                br.append_revision,
 
1114
                'current:')
 
1115
 
 
1116
        :param deprecation_format: The deprecation format that the callable
 
1117
            should have been deprecated with. This is the same type as the
 
1118
            parameter to deprecated_method/deprecated_function. If the
 
1119
            callable is not deprecated with this format, an assertion error
 
1120
            will be raised.
 
1121
        :param a_callable: A callable to call. This may be a bound method or
 
1122
            a regular function. It will be called with ``*args`` and
 
1123
            ``**kwargs``.
 
1124
        :param args: The positional arguments for the callable
 
1125
        :param kwargs: The keyword arguments for the callable
 
1126
        :return: The result of a_callable(``*args``, ``**kwargs``)
 
1127
        """
 
1128
        call_warnings, result = self._capture_deprecation_warnings(a_callable,
 
1129
            *args, **kwargs)
 
1130
        expected_first_warning = symbol_versioning.deprecation_string(
 
1131
            a_callable, deprecation_format)
 
1132
        if len(call_warnings) == 0:
 
1133
            self.fail("No deprecation warning generated by call to %s" %
 
1134
                a_callable)
 
1135
        self.assertEqual(expected_first_warning, call_warnings[0])
 
1136
        return result
 
1137
 
 
1138
    def callCatchWarnings(self, fn, *args, **kw):
 
1139
        """Call a callable that raises python warnings.
 
1140
 
 
1141
        The caller's responsible for examining the returned warnings.
 
1142
 
 
1143
        If the callable raises an exception, the exception is not
 
1144
        caught and propagates up to the caller.  In that case, the list
 
1145
        of warnings is not available.
 
1146
 
 
1147
        :returns: ([warning_object, ...], fn_result)
 
1148
        """
 
1149
        # XXX: This is not perfect, because it completely overrides the
 
1150
        # warnings filters, and some code may depend on suppressing particular
 
1151
        # warnings.  It's the easiest way to insulate ourselves from -Werror,
 
1152
        # though.  -- Andrew, 20071062
 
1153
        wlist = []
 
1154
        def _catcher(message, category, filename, lineno, file=None, line=None):
 
1155
            # despite the name, 'message' is normally(?) a Warning subclass
 
1156
            # instance
 
1157
            wlist.append(message)
 
1158
        saved_showwarning = warnings.showwarning
 
1159
        saved_filters = warnings.filters
 
1160
        try:
 
1161
            warnings.showwarning = _catcher
 
1162
            warnings.filters = []
 
1163
            result = fn(*args, **kw)
 
1164
        finally:
 
1165
            warnings.showwarning = saved_showwarning
 
1166
            warnings.filters = saved_filters
 
1167
        return wlist, result
 
1168
 
 
1169
    def callDeprecated(self, expected, callable, *args, **kwargs):
 
1170
        """Assert that a callable is deprecated in a particular way.
 
1171
 
 
1172
        This is a very precise test for unusual requirements. The 
 
1173
        applyDeprecated helper function is probably more suited for most tests
 
1174
        as it allows you to simply specify the deprecation format being used
 
1175
        and will ensure that that is issued for the function being called.
 
1176
 
 
1177
        Note that this only captures warnings raised by symbol_versioning.warn,
 
1178
        not other callers that go direct to the warning module.  To catch
 
1179
        general warnings, use callCatchWarnings.
 
1180
 
 
1181
        :param expected: a list of the deprecation warnings expected, in order
 
1182
        :param callable: The callable to call
 
1183
        :param args: The positional arguments for the callable
 
1184
        :param kwargs: The keyword arguments for the callable
 
1185
        """
 
1186
        call_warnings, result = self._capture_deprecation_warnings(callable,
 
1187
            *args, **kwargs)
 
1188
        self.assertEqual(expected, call_warnings)
 
1189
        return result
 
1190
 
 
1191
    def _startLogFile(self):
 
1192
        """Send bzr and test log messages to a temporary file.
 
1193
 
 
1194
        The file is removed as the test is torn down.
 
1195
        """
 
1196
        fileno, name = tempfile.mkstemp(suffix='.log', prefix='testbzr')
 
1197
        self._log_file = os.fdopen(fileno, 'w+')
 
1198
        self._log_memento = bzrlib.trace.push_log_file(self._log_file)
 
1199
        self._log_file_name = name
 
1200
        self.addCleanup(self._finishLogFile)
 
1201
 
 
1202
    def _finishLogFile(self):
 
1203
        """Finished with the log file.
 
1204
 
 
1205
        Close the file and delete it, unless setKeepLogfile was called.
 
1206
        """
 
1207
        if self._log_file is None:
 
1208
            return
 
1209
        bzrlib.trace.pop_log_file(self._log_memento)
 
1210
        self._log_file.close()
 
1211
        self._log_file = None
 
1212
        if not self._keep_log_file:
 
1213
            os.remove(self._log_file_name)
 
1214
            self._log_file_name = None
 
1215
 
 
1216
    def setKeepLogfile(self):
 
1217
        """Make the logfile not be deleted when _finishLogFile is called."""
 
1218
        self._keep_log_file = True
 
1219
 
 
1220
    def addCleanup(self, callable, *args, **kwargs):
 
1221
        """Arrange to run a callable when this case is torn down.
 
1222
 
 
1223
        Callables are run in the reverse of the order they are registered, 
 
1224
        ie last-in first-out.
 
1225
        """
 
1226
        self._cleanups.append((callable, args, kwargs))
 
1227
 
 
1228
    def _cleanEnvironment(self):
 
1229
        new_env = {
 
1230
            'BZR_HOME': None, # Don't inherit BZR_HOME to all the tests.
 
1231
            'HOME': os.getcwd(),
 
1232
            # bzr now uses the Win32 API and doesn't rely on APPDATA, but the
 
1233
            # tests do check our impls match APPDATA
 
1234
            'BZR_EDITOR': None, # test_msgeditor manipulates this variable
 
1235
            'BZR_EMAIL': None,
 
1236
            'BZREMAIL': None, # may still be present in the environment
 
1237
            'EMAIL': None,
 
1238
            'BZR_PROGRESS_BAR': None,
 
1239
            'BZR_LOG': None,
 
1240
            'BZR_PLUGIN_PATH': None,
 
1241
            # SSH Agent
 
1242
            'SSH_AUTH_SOCK': None,
 
1243
            # Proxies
 
1244
            'http_proxy': None,
 
1245
            'HTTP_PROXY': None,
 
1246
            'https_proxy': None,
 
1247
            'HTTPS_PROXY': None,
 
1248
            'no_proxy': None,
 
1249
            'NO_PROXY': None,
 
1250
            'all_proxy': None,
 
1251
            'ALL_PROXY': None,
 
1252
            # Nobody cares about these ones AFAIK. So far at
 
1253
            # least. If you do (care), please update this comment
 
1254
            # -- vila 20061212
 
1255
            'ftp_proxy': None,
 
1256
            'FTP_PROXY': None,
 
1257
            'BZR_REMOTE_PATH': None,
 
1258
        }
 
1259
        self.__old_env = {}
 
1260
        self.addCleanup(self._restoreEnvironment)
 
1261
        for name, value in new_env.iteritems():
 
1262
            self._captureVar(name, value)
 
1263
 
 
1264
    def _captureVar(self, name, newvalue):
 
1265
        """Set an environment variable, and reset it when finished."""
 
1266
        self.__old_env[name] = osutils.set_or_unset_env(name, newvalue)
 
1267
 
 
1268
    def _restore_debug_flags(self):
 
1269
        debug.debug_flags.clear()
 
1270
        debug.debug_flags.update(self._preserved_debug_flags)
 
1271
 
 
1272
    def _restoreEnvironment(self):
 
1273
        for name, value in self.__old_env.iteritems():
 
1274
            osutils.set_or_unset_env(name, value)
 
1275
 
 
1276
    def _restoreHooks(self):
 
1277
        for klass, hooks in self._preserved_hooks.items():
 
1278
            setattr(klass, 'hooks', hooks)
 
1279
 
 
1280
    def knownFailure(self, reason):
 
1281
        """This test has failed for some known reason."""
 
1282
        raise KnownFailure(reason)
 
1283
 
 
1284
    def run(self, result=None):
 
1285
        if result is None: result = self.defaultTestResult()
 
1286
        for feature in getattr(self, '_test_needs_features', []):
 
1287
            if not feature.available():
 
1288
                result.startTest(self)
 
1289
                if getattr(result, 'addNotSupported', None):
 
1290
                    result.addNotSupported(self, feature)
 
1291
                else:
 
1292
                    result.addSuccess(self)
 
1293
                result.stopTest(self)
 
1294
                return
 
1295
        try:
 
1296
            return unittest.TestCase.run(self, result)
 
1297
        finally:
 
1298
            saved_attrs = {}
 
1299
            absent_attr = object()
 
1300
            for attr_name in self.attrs_to_keep:
 
1301
                attr = getattr(self, attr_name, absent_attr)
 
1302
                if attr is not absent_attr:
 
1303
                    saved_attrs[attr_name] = attr
 
1304
            self.__dict__ = saved_attrs
 
1305
 
 
1306
    def tearDown(self):
 
1307
        self._runCleanups()
 
1308
        unittest.TestCase.tearDown(self)
 
1309
 
 
1310
    def time(self, callable, *args, **kwargs):
 
1311
        """Run callable and accrue the time it takes to the benchmark time.
 
1312
        
 
1313
        If lsprofiling is enabled (i.e. by --lsprof-time to bzr selftest) then
 
1314
        this will cause lsprofile statistics to be gathered and stored in
 
1315
        self._benchcalls.
 
1316
        """
 
1317
        if self._benchtime is None:
 
1318
            self._benchtime = 0
 
1319
        start = time.time()
 
1320
        try:
 
1321
            if not self._gather_lsprof_in_benchmarks:
 
1322
                return callable(*args, **kwargs)
 
1323
            else:
 
1324
                # record this benchmark
 
1325
                ret, stats = bzrlib.lsprof.profile(callable, *args, **kwargs)
 
1326
                stats.sort()
 
1327
                self._benchcalls.append(((callable, args, kwargs), stats))
 
1328
                return ret
 
1329
        finally:
 
1330
            self._benchtime += time.time() - start
 
1331
 
 
1332
    def _runCleanups(self):
 
1333
        """Run registered cleanup functions. 
 
1334
 
 
1335
        This should only be called from TestCase.tearDown.
 
1336
        """
 
1337
        # TODO: Perhaps this should keep running cleanups even if 
 
1338
        # one of them fails?
 
1339
 
 
1340
        # Actually pop the cleanups from the list so tearDown running
 
1341
        # twice is safe (this happens for skipped tests).
 
1342
        while self._cleanups:
 
1343
            cleanup, args, kwargs = self._cleanups.pop()
 
1344
            cleanup(*args, **kwargs)
 
1345
 
 
1346
    def log(self, *args):
 
1347
        mutter(*args)
 
1348
 
 
1349
    def _get_log(self, keep_log_file=False):
 
1350
        """Get the log from bzrlib.trace calls from this test.
 
1351
 
 
1352
        :param keep_log_file: When True, if the log is still a file on disk
 
1353
            leave it as a file on disk. When False, if the log is still a file
 
1354
            on disk, the log file is deleted and the log preserved as
 
1355
            self._log_contents.
 
1356
        :return: A string containing the log.
 
1357
        """
 
1358
        # flush the log file, to get all content
 
1359
        import bzrlib.trace
 
1360
        if bzrlib.trace._trace_file:
 
1361
            bzrlib.trace._trace_file.flush()
 
1362
        if self._log_contents:
 
1363
            # XXX: this can hardly contain the content flushed above --vila
 
1364
            # 20080128
 
1365
            return self._log_contents
 
1366
        if self._log_file_name is not None:
 
1367
            logfile = open(self._log_file_name)
 
1368
            try:
 
1369
                log_contents = logfile.read()
 
1370
            finally:
 
1371
                logfile.close()
 
1372
            if not keep_log_file:
 
1373
                self._log_contents = log_contents
 
1374
                try:
 
1375
                    os.remove(self._log_file_name)
 
1376
                except OSError, e:
 
1377
                    if sys.platform == 'win32' and e.errno == errno.EACCES:
 
1378
                        sys.stderr.write(('Unable to delete log file '
 
1379
                                             ' %r\n' % self._log_file_name))
 
1380
                    else:
 
1381
                        raise
 
1382
            return log_contents
 
1383
        else:
 
1384
            return "DELETED log file to reduce memory footprint"
 
1385
 
 
1386
    def requireFeature(self, feature):
 
1387
        """This test requires a specific feature is available.
 
1388
 
 
1389
        :raises UnavailableFeature: When feature is not available.
 
1390
        """
 
1391
        if not feature.available():
 
1392
            raise UnavailableFeature(feature)
 
1393
 
 
1394
    def _run_bzr_autosplit(self, args, retcode, encoding, stdin,
 
1395
            working_dir):
 
1396
        """Run bazaar command line, splitting up a string command line."""
 
1397
        if isinstance(args, basestring):
 
1398
            # shlex don't understand unicode strings,
 
1399
            # so args should be plain string (bialix 20070906)
 
1400
            args = list(shlex.split(str(args)))
 
1401
        return self._run_bzr_core(args, retcode=retcode,
 
1402
                encoding=encoding, stdin=stdin, working_dir=working_dir,
 
1403
                )
 
1404
 
 
1405
    def _run_bzr_core(self, args, retcode, encoding, stdin,
 
1406
            working_dir):
 
1407
        if encoding is None:
 
1408
            encoding = osutils.get_user_encoding()
 
1409
        stdout = StringIOWrapper()
 
1410
        stderr = StringIOWrapper()
 
1411
        stdout.encoding = encoding
 
1412
        stderr.encoding = encoding
 
1413
 
 
1414
        self.log('run bzr: %r', args)
 
1415
        # FIXME: don't call into logging here
 
1416
        handler = logging.StreamHandler(stderr)
 
1417
        handler.setLevel(logging.INFO)
 
1418
        logger = logging.getLogger('')
 
1419
        logger.addHandler(handler)
 
1420
        old_ui_factory = ui.ui_factory
 
1421
        ui.ui_factory = TestUIFactory(stdin=stdin, stdout=stdout, stderr=stderr)
 
1422
 
 
1423
        cwd = None
 
1424
        if working_dir is not None:
 
1425
            cwd = osutils.getcwd()
 
1426
            os.chdir(working_dir)
 
1427
 
 
1428
        try:
 
1429
            result = self.apply_redirected(ui.ui_factory.stdin,
 
1430
                stdout, stderr,
 
1431
                bzrlib.commands.run_bzr_catch_user_errors,
 
1432
                args)
 
1433
        finally:
 
1434
            logger.removeHandler(handler)
 
1435
            ui.ui_factory = old_ui_factory
 
1436
            if cwd is not None:
 
1437
                os.chdir(cwd)
 
1438
 
 
1439
        out = stdout.getvalue()
 
1440
        err = stderr.getvalue()
 
1441
        if out:
 
1442
            self.log('output:\n%r', out)
 
1443
        if err:
 
1444
            self.log('errors:\n%r', err)
 
1445
        if retcode is not None:
 
1446
            self.assertEquals(retcode, result,
 
1447
                              message='Unexpected return code')
 
1448
        return out, err
 
1449
 
 
1450
    def run_bzr(self, args, retcode=0, encoding=None, stdin=None,
 
1451
                working_dir=None, error_regexes=[], output_encoding=None):
 
1452
        """Invoke bzr, as if it were run from the command line.
 
1453
 
 
1454
        The argument list should not include the bzr program name - the
 
1455
        first argument is normally the bzr command.  Arguments may be
 
1456
        passed in three ways:
 
1457
 
 
1458
        1- A list of strings, eg ["commit", "a"].  This is recommended
 
1459
        when the command contains whitespace or metacharacters, or 
 
1460
        is built up at run time.
 
1461
 
 
1462
        2- A single string, eg "add a".  This is the most convenient 
 
1463
        for hardcoded commands.
 
1464
 
 
1465
        This runs bzr through the interface that catches and reports
 
1466
        errors, and with logging set to something approximating the
 
1467
        default, so that error reporting can be checked.
 
1468
 
 
1469
        This should be the main method for tests that want to exercise the
 
1470
        overall behavior of the bzr application (rather than a unit test
 
1471
        or a functional test of the library.)
 
1472
 
 
1473
        This sends the stdout/stderr results into the test's log,
 
1474
        where it may be useful for debugging.  See also run_captured.
 
1475
 
 
1476
        :keyword stdin: A string to be used as stdin for the command.
 
1477
        :keyword retcode: The status code the command should return;
 
1478
            default 0.
 
1479
        :keyword working_dir: The directory to run the command in
 
1480
        :keyword error_regexes: A list of expected error messages.  If
 
1481
            specified they must be seen in the error output of the command.
 
1482
        """
 
1483
        out, err = self._run_bzr_autosplit(
 
1484
            args=args,
 
1485
            retcode=retcode,
 
1486
            encoding=encoding,
 
1487
            stdin=stdin,
 
1488
            working_dir=working_dir,
 
1489
            )
 
1490
        for regex in error_regexes:
 
1491
            self.assertContainsRe(err, regex)
 
1492
        return out, err
 
1493
 
 
1494
    def run_bzr_error(self, error_regexes, *args, **kwargs):
 
1495
        """Run bzr, and check that stderr contains the supplied regexes
 
1496
 
 
1497
        :param error_regexes: Sequence of regular expressions which
 
1498
            must each be found in the error output. The relative ordering
 
1499
            is not enforced.
 
1500
        :param args: command-line arguments for bzr
 
1501
        :param kwargs: Keyword arguments which are interpreted by run_bzr
 
1502
            This function changes the default value of retcode to be 3,
 
1503
            since in most cases this is run when you expect bzr to fail.
 
1504
 
 
1505
        :return: (out, err) The actual output of running the command (in case
 
1506
            you want to do more inspection)
 
1507
 
 
1508
        Examples of use::
 
1509
 
 
1510
            # Make sure that commit is failing because there is nothing to do
 
1511
            self.run_bzr_error(['no changes to commit'],
 
1512
                               ['commit', '-m', 'my commit comment'])
 
1513
            # Make sure --strict is handling an unknown file, rather than
 
1514
            # giving us the 'nothing to do' error
 
1515
            self.build_tree(['unknown'])
 
1516
            self.run_bzr_error(['Commit refused because there are unknown files'],
 
1517
                               ['commit', --strict', '-m', 'my commit comment'])
 
1518
        """
 
1519
        kwargs.setdefault('retcode', 3)
 
1520
        kwargs['error_regexes'] = error_regexes
 
1521
        out, err = self.run_bzr(*args, **kwargs)
 
1522
        return out, err
 
1523
 
 
1524
    def run_bzr_subprocess(self, *args, **kwargs):
 
1525
        """Run bzr in a subprocess for testing.
 
1526
 
 
1527
        This starts a new Python interpreter and runs bzr in there. 
 
1528
        This should only be used for tests that have a justifiable need for
 
1529
        this isolation: e.g. they are testing startup time, or signal
 
1530
        handling, or early startup code, etc.  Subprocess code can't be 
 
1531
        profiled or debugged so easily.
 
1532
 
 
1533
        :keyword retcode: The status code that is expected.  Defaults to 0.  If
 
1534
            None is supplied, the status code is not checked.
 
1535
        :keyword env_changes: A dictionary which lists changes to environment
 
1536
            variables. A value of None will unset the env variable.
 
1537
            The values must be strings. The change will only occur in the
 
1538
            child, so you don't need to fix the environment after running.
 
1539
        :keyword universal_newlines: Convert CRLF => LF
 
1540
        :keyword allow_plugins: By default the subprocess is run with
 
1541
            --no-plugins to ensure test reproducibility. Also, it is possible
 
1542
            for system-wide plugins to create unexpected output on stderr,
 
1543
            which can cause unnecessary test failures.
 
1544
        """
 
1545
        env_changes = kwargs.get('env_changes', {})
 
1546
        working_dir = kwargs.get('working_dir', None)
 
1547
        allow_plugins = kwargs.get('allow_plugins', False)
 
1548
        if len(args) == 1:
 
1549
            if isinstance(args[0], list):
 
1550
                args = args[0]
 
1551
            elif isinstance(args[0], basestring):
 
1552
                args = list(shlex.split(args[0]))
 
1553
        else:
 
1554
            raise ValueError("passing varargs to run_bzr_subprocess")
 
1555
        process = self.start_bzr_subprocess(args, env_changes=env_changes,
 
1556
                                            working_dir=working_dir,
 
1557
                                            allow_plugins=allow_plugins)
 
1558
        # We distinguish between retcode=None and retcode not passed.
 
1559
        supplied_retcode = kwargs.get('retcode', 0)
 
1560
        return self.finish_bzr_subprocess(process, retcode=supplied_retcode,
 
1561
            universal_newlines=kwargs.get('universal_newlines', False),
 
1562
            process_args=args)
 
1563
 
 
1564
    def start_bzr_subprocess(self, process_args, env_changes=None,
 
1565
                             skip_if_plan_to_signal=False,
 
1566
                             working_dir=None,
 
1567
                             allow_plugins=False):
 
1568
        """Start bzr in a subprocess for testing.
 
1569
 
 
1570
        This starts a new Python interpreter and runs bzr in there.
 
1571
        This should only be used for tests that have a justifiable need for
 
1572
        this isolation: e.g. they are testing startup time, or signal
 
1573
        handling, or early startup code, etc.  Subprocess code can't be
 
1574
        profiled or debugged so easily.
 
1575
 
 
1576
        :param process_args: a list of arguments to pass to the bzr executable,
 
1577
            for example ``['--version']``.
 
1578
        :param env_changes: A dictionary which lists changes to environment
 
1579
            variables. A value of None will unset the env variable.
 
1580
            The values must be strings. The change will only occur in the
 
1581
            child, so you don't need to fix the environment after running.
 
1582
        :param skip_if_plan_to_signal: raise TestSkipped when true and os.kill
 
1583
            is not available.
 
1584
        :param allow_plugins: If False (default) pass --no-plugins to bzr.
 
1585
 
 
1586
        :returns: Popen object for the started process.
 
1587
        """
 
1588
        if skip_if_plan_to_signal:
 
1589
            if not getattr(os, 'kill', None):
 
1590
                raise TestSkipped("os.kill not available.")
 
1591
 
 
1592
        if env_changes is None:
 
1593
            env_changes = {}
 
1594
        old_env = {}
 
1595
 
 
1596
        def cleanup_environment():
 
1597
            for env_var, value in env_changes.iteritems():
 
1598
                old_env[env_var] = osutils.set_or_unset_env(env_var, value)
 
1599
 
 
1600
        def restore_environment():
 
1601
            for env_var, value in old_env.iteritems():
 
1602
                osutils.set_or_unset_env(env_var, value)
 
1603
 
 
1604
        bzr_path = self.get_bzr_path()
 
1605
 
 
1606
        cwd = None
 
1607
        if working_dir is not None:
 
1608
            cwd = osutils.getcwd()
 
1609
            os.chdir(working_dir)
 
1610
 
 
1611
        try:
 
1612
            # win32 subprocess doesn't support preexec_fn
 
1613
            # so we will avoid using it on all platforms, just to
 
1614
            # make sure the code path is used, and we don't break on win32
 
1615
            cleanup_environment()
 
1616
            command = [sys.executable]
 
1617
            # frozen executables don't need the path to bzr
 
1618
            if getattr(sys, "frozen", None) is None:
 
1619
                command.append(bzr_path)
 
1620
            if not allow_plugins:
 
1621
                command.append('--no-plugins')
 
1622
            command.extend(process_args)
 
1623
            process = self._popen(command, stdin=PIPE, stdout=PIPE, stderr=PIPE)
 
1624
        finally:
 
1625
            restore_environment()
 
1626
            if cwd is not None:
 
1627
                os.chdir(cwd)
 
1628
 
 
1629
        return process
 
1630
 
 
1631
    def _popen(self, *args, **kwargs):
 
1632
        """Place a call to Popen.
 
1633
 
 
1634
        Allows tests to override this method to intercept the calls made to
 
1635
        Popen for introspection.
 
1636
        """
 
1637
        return Popen(*args, **kwargs)
 
1638
 
 
1639
    def get_bzr_path(self):
 
1640
        """Return the path of the 'bzr' executable for this test suite."""
 
1641
        bzr_path = os.path.dirname(os.path.dirname(bzrlib.__file__))+'/bzr'
 
1642
        if not os.path.isfile(bzr_path):
 
1643
            # We are probably installed. Assume sys.argv is the right file
 
1644
            bzr_path = sys.argv[0]
 
1645
        return bzr_path
 
1646
 
 
1647
    def finish_bzr_subprocess(self, process, retcode=0, send_signal=None,
 
1648
                              universal_newlines=False, process_args=None):
 
1649
        """Finish the execution of process.
 
1650
 
 
1651
        :param process: the Popen object returned from start_bzr_subprocess.
 
1652
        :param retcode: The status code that is expected.  Defaults to 0.  If
 
1653
            None is supplied, the status code is not checked.
 
1654
        :param send_signal: an optional signal to send to the process.
 
1655
        :param universal_newlines: Convert CRLF => LF
 
1656
        :returns: (stdout, stderr)
 
1657
        """
 
1658
        if send_signal is not None:
 
1659
            os.kill(process.pid, send_signal)
 
1660
        out, err = process.communicate()
 
1661
 
 
1662
        if universal_newlines:
 
1663
            out = out.replace('\r\n', '\n')
 
1664
            err = err.replace('\r\n', '\n')
 
1665
 
 
1666
        if retcode is not None and retcode != process.returncode:
 
1667
            if process_args is None:
 
1668
                process_args = "(unknown args)"
 
1669
            mutter('Output of bzr %s:\n%s', process_args, out)
 
1670
            mutter('Error for bzr %s:\n%s', process_args, err)
 
1671
            self.fail('Command bzr %s failed with retcode %s != %s'
 
1672
                      % (process_args, retcode, process.returncode))
 
1673
        return [out, err]
 
1674
 
 
1675
    def check_inventory_shape(self, inv, shape):
 
1676
        """Compare an inventory to a list of expected names.
 
1677
 
 
1678
        Fail if they are not precisely equal.
 
1679
        """
 
1680
        extras = []
 
1681
        shape = list(shape)             # copy
 
1682
        for path, ie in inv.entries():
 
1683
            name = path.replace('\\', '/')
 
1684
            if ie.kind == 'directory':
 
1685
                name = name + '/'
 
1686
            if name in shape:
 
1687
                shape.remove(name)
 
1688
            else:
 
1689
                extras.append(name)
 
1690
        if shape:
 
1691
            self.fail("expected paths not found in inventory: %r" % shape)
 
1692
        if extras:
 
1693
            self.fail("unexpected paths found in inventory: %r" % extras)
 
1694
 
 
1695
    def apply_redirected(self, stdin=None, stdout=None, stderr=None,
 
1696
                         a_callable=None, *args, **kwargs):
 
1697
        """Call callable with redirected std io pipes.
 
1698
 
 
1699
        Returns the return code."""
 
1700
        if not callable(a_callable):
 
1701
            raise ValueError("a_callable must be callable.")
 
1702
        if stdin is None:
 
1703
            stdin = StringIO("")
 
1704
        if stdout is None:
 
1705
            if getattr(self, "_log_file", None) is not None:
 
1706
                stdout = self._log_file
 
1707
            else:
 
1708
                stdout = StringIO()
 
1709
        if stderr is None:
 
1710
            if getattr(self, "_log_file", None is not None):
 
1711
                stderr = self._log_file
 
1712
            else:
 
1713
                stderr = StringIO()
 
1714
        real_stdin = sys.stdin
 
1715
        real_stdout = sys.stdout
 
1716
        real_stderr = sys.stderr
 
1717
        try:
 
1718
            sys.stdout = stdout
 
1719
            sys.stderr = stderr
 
1720
            sys.stdin = stdin
 
1721
            return a_callable(*args, **kwargs)
 
1722
        finally:
 
1723
            sys.stdout = real_stdout
 
1724
            sys.stderr = real_stderr
 
1725
            sys.stdin = real_stdin
 
1726
 
 
1727
    def reduceLockdirTimeout(self):
 
1728
        """Reduce the default lock timeout for the duration of the test, so that
 
1729
        if LockContention occurs during a test, it does so quickly.
 
1730
 
 
1731
        Tests that expect to provoke LockContention errors should call this.
 
1732
        """
 
1733
        orig_timeout = bzrlib.lockdir._DEFAULT_TIMEOUT_SECONDS
 
1734
        def resetTimeout():
 
1735
            bzrlib.lockdir._DEFAULT_TIMEOUT_SECONDS = orig_timeout
 
1736
        self.addCleanup(resetTimeout)
 
1737
        bzrlib.lockdir._DEFAULT_TIMEOUT_SECONDS = 0
 
1738
 
 
1739
    def make_utf8_encoded_stringio(self, encoding_type=None):
 
1740
        """Return a StringIOWrapper instance, that will encode Unicode
 
1741
        input to UTF-8.
 
1742
        """
 
1743
        if encoding_type is None:
 
1744
            encoding_type = 'strict'
 
1745
        sio = StringIO()
 
1746
        output_encoding = 'utf-8'
 
1747
        sio = codecs.getwriter(output_encoding)(sio, errors=encoding_type)
 
1748
        sio.encoding = output_encoding
 
1749
        return sio
 
1750
 
 
1751
 
 
1752
class TestCaseWithMemoryTransport(TestCase):
 
1753
    """Common test class for tests that do not need disk resources.
 
1754
 
 
1755
    Tests that need disk resources should derive from TestCaseWithTransport.
 
1756
 
 
1757
    TestCaseWithMemoryTransport sets the TEST_ROOT variable for all bzr tests.
 
1758
 
 
1759
    For TestCaseWithMemoryTransport the test_home_dir is set to the name of
 
1760
    a directory which does not exist. This serves to help ensure test isolation
 
1761
    is preserved. test_dir is set to the TEST_ROOT, as is cwd, because they
 
1762
    must exist. However, TestCaseWithMemoryTransport does not offer local
 
1763
    file defaults for the transport in tests, nor does it obey the command line
 
1764
    override, so tests that accidentally write to the common directory should
 
1765
    be rare.
 
1766
 
 
1767
    :cvar TEST_ROOT: Directory containing all temporary directories, plus
 
1768
    a .bzr directory that stops us ascending higher into the filesystem.
 
1769
    """
 
1770
 
 
1771
    TEST_ROOT = None
 
1772
    _TEST_NAME = 'test'
 
1773
 
 
1774
    def __init__(self, methodName='runTest'):
 
1775
        # allow test parameterization after test construction and before test
 
1776
        # execution. Variables that the parameterizer sets need to be 
 
1777
        # ones that are not set by setUp, or setUp will trash them.
 
1778
        super(TestCaseWithMemoryTransport, self).__init__(methodName)
 
1779
        self.vfs_transport_factory = default_transport
 
1780
        self.transport_server = None
 
1781
        self.transport_readonly_server = None
 
1782
        self.__vfs_server = None
 
1783
 
 
1784
    def get_transport(self, relpath=None):
 
1785
        """Return a writeable transport.
 
1786
 
 
1787
        This transport is for the test scratch space relative to
 
1788
        "self._test_root"
 
1789
        
 
1790
        :param relpath: a path relative to the base url.
 
1791
        """
 
1792
        t = get_transport(self.get_url(relpath))
 
1793
        self.assertFalse(t.is_readonly())
 
1794
        return t
 
1795
 
 
1796
    def get_readonly_transport(self, relpath=None):
 
1797
        """Return a readonly transport for the test scratch space
 
1798
        
 
1799
        This can be used to test that operations which should only need
 
1800
        readonly access in fact do not try to write.
 
1801
 
 
1802
        :param relpath: a path relative to the base url.
 
1803
        """
 
1804
        t = get_transport(self.get_readonly_url(relpath))
 
1805
        self.assertTrue(t.is_readonly())
 
1806
        return t
 
1807
 
 
1808
    def create_transport_readonly_server(self):
 
1809
        """Create a transport server from class defined at init.
 
1810
 
 
1811
        This is mostly a hook for daughter classes.
 
1812
        """
 
1813
        return self.transport_readonly_server()
 
1814
 
 
1815
    def get_readonly_server(self):
 
1816
        """Get the server instance for the readonly transport
 
1817
 
 
1818
        This is useful for some tests with specific servers to do diagnostics.
 
1819
        """
 
1820
        if self.__readonly_server is None:
 
1821
            if self.transport_readonly_server is None:
 
1822
                # readonly decorator requested
 
1823
                # bring up the server
 
1824
                self.__readonly_server = ReadonlyServer()
 
1825
                self.__readonly_server.setUp(self.get_vfs_only_server())
 
1826
            else:
 
1827
                self.__readonly_server = self.create_transport_readonly_server()
 
1828
                self.__readonly_server.setUp(self.get_vfs_only_server())
 
1829
            self.addCleanup(self.__readonly_server.tearDown)
 
1830
        return self.__readonly_server
 
1831
 
 
1832
    def get_readonly_url(self, relpath=None):
 
1833
        """Get a URL for the readonly transport.
 
1834
 
 
1835
        This will either be backed by '.' or a decorator to the transport 
 
1836
        used by self.get_url()
 
1837
        relpath provides for clients to get a path relative to the base url.
 
1838
        These should only be downwards relative, not upwards.
 
1839
        """
 
1840
        base = self.get_readonly_server().get_url()
 
1841
        return self._adjust_url(base, relpath)
 
1842
 
 
1843
    def get_vfs_only_server(self):
 
1844
        """Get the vfs only read/write server instance.
 
1845
 
 
1846
        This is useful for some tests with specific servers that need
 
1847
        diagnostics.
 
1848
 
 
1849
        For TestCaseWithMemoryTransport this is always a MemoryServer, and there
 
1850
        is no means to override it.
 
1851
        """
 
1852
        if self.__vfs_server is None:
 
1853
            self.__vfs_server = MemoryServer()
 
1854
            self.__vfs_server.setUp()
 
1855
            self.addCleanup(self.__vfs_server.tearDown)
 
1856
        return self.__vfs_server
 
1857
 
 
1858
    def get_server(self):
 
1859
        """Get the read/write server instance.
 
1860
 
 
1861
        This is useful for some tests with specific servers that need
 
1862
        diagnostics.
 
1863
 
 
1864
        This is built from the self.transport_server factory. If that is None,
 
1865
        then the self.get_vfs_server is returned.
 
1866
        """
 
1867
        if self.__server is None:
 
1868
            if self.transport_server is None or self.transport_server is self.vfs_transport_factory:
 
1869
                return self.get_vfs_only_server()
 
1870
            else:
 
1871
                # bring up a decorated means of access to the vfs only server.
 
1872
                self.__server = self.transport_server()
 
1873
                try:
 
1874
                    self.__server.setUp(self.get_vfs_only_server())
 
1875
                except TypeError, e:
 
1876
                    # This should never happen; the try:Except here is to assist
 
1877
                    # developers having to update code rather than seeing an
 
1878
                    # uninformative TypeError.
 
1879
                    raise Exception, "Old server API in use: %s, %s" % (self.__server, e)
 
1880
            self.addCleanup(self.__server.tearDown)
 
1881
        return self.__server
 
1882
 
 
1883
    def _adjust_url(self, base, relpath):
 
1884
        """Get a URL (or maybe a path) for the readwrite transport.
 
1885
 
 
1886
        This will either be backed by '.' or to an equivalent non-file based
 
1887
        facility.
 
1888
        relpath provides for clients to get a path relative to the base url.
 
1889
        These should only be downwards relative, not upwards.
 
1890
        """
 
1891
        if relpath is not None and relpath != '.':
 
1892
            if not base.endswith('/'):
 
1893
                base = base + '/'
 
1894
            # XXX: Really base should be a url; we did after all call
 
1895
            # get_url()!  But sometimes it's just a path (from
 
1896
            # LocalAbspathServer), and it'd be wrong to append urlescaped data
 
1897
            # to a non-escaped local path.
 
1898
            if base.startswith('./') or base.startswith('/'):
 
1899
                base += relpath
 
1900
            else:
 
1901
                base += urlutils.escape(relpath)
 
1902
        return base
 
1903
 
 
1904
    def get_url(self, relpath=None):
 
1905
        """Get a URL (or maybe a path) for the readwrite transport.
 
1906
 
 
1907
        This will either be backed by '.' or to an equivalent non-file based
 
1908
        facility.
 
1909
        relpath provides for clients to get a path relative to the base url.
 
1910
        These should only be downwards relative, not upwards.
 
1911
        """
 
1912
        base = self.get_server().get_url()
 
1913
        return self._adjust_url(base, relpath)
 
1914
 
 
1915
    def get_vfs_only_url(self, relpath=None):
 
1916
        """Get a URL (or maybe a path for the plain old vfs transport.
 
1917
 
 
1918
        This will never be a smart protocol.  It always has all the
 
1919
        capabilities of the local filesystem, but it might actually be a
 
1920
        MemoryTransport or some other similar virtual filesystem.
 
1921
 
 
1922
        This is the backing transport (if any) of the server returned by
 
1923
        get_url and get_readonly_url.
 
1924
 
 
1925
        :param relpath: provides for clients to get a path relative to the base
 
1926
            url.  These should only be downwards relative, not upwards.
 
1927
        :return: A URL
 
1928
        """
 
1929
        base = self.get_vfs_only_server().get_url()
 
1930
        return self._adjust_url(base, relpath)
 
1931
 
 
1932
    def _create_safety_net(self):
 
1933
        """Make a fake bzr directory.
 
1934
 
 
1935
        This prevents any tests propagating up onto the TEST_ROOT directory's
 
1936
        real branch.
 
1937
        """
 
1938
        root = TestCaseWithMemoryTransport.TEST_ROOT
 
1939
        bzrdir.BzrDir.create_standalone_workingtree(root)
 
1940
 
 
1941
    def _check_safety_net(self):
 
1942
        """Check that the safety .bzr directory have not been touched.
 
1943
 
 
1944
        _make_test_root have created a .bzr directory to prevent tests from
 
1945
        propagating. This method ensures than a test did not leaked.
 
1946
        """
 
1947
        root = TestCaseWithMemoryTransport.TEST_ROOT
 
1948
        wt = workingtree.WorkingTree.open(root)
 
1949
        last_rev = wt.last_revision()
 
1950
        if last_rev != 'null:':
 
1951
            # The current test have modified the /bzr directory, we need to
 
1952
            # recreate a new one or all the followng tests will fail.
 
1953
            # If you need to inspect its content uncomment the following line
 
1954
            # import pdb; pdb.set_trace()
 
1955
            _rmtree_temp_dir(root + '/.bzr')
 
1956
            self._create_safety_net()
 
1957
            raise AssertionError('%s/.bzr should not be modified' % root)
 
1958
 
 
1959
    def _make_test_root(self):
 
1960
        if TestCaseWithMemoryTransport.TEST_ROOT is None:
 
1961
            root = osutils.mkdtemp(prefix='testbzr-', suffix='.tmp')
 
1962
            TestCaseWithMemoryTransport.TEST_ROOT = root
 
1963
 
 
1964
            self._create_safety_net()
 
1965
 
 
1966
            # The same directory is used by all tests, and we're not
 
1967
            # specifically told when all tests are finished.  This will do.
 
1968
            atexit.register(_rmtree_temp_dir, root)
 
1969
 
 
1970
        self.addCleanup(self._check_safety_net)
 
1971
 
 
1972
    def makeAndChdirToTestDir(self):
 
1973
        """Create a temporary directories for this one test.
 
1974
        
 
1975
        This must set self.test_home_dir and self.test_dir and chdir to
 
1976
        self.test_dir.
 
1977
        
 
1978
        For TestCaseWithMemoryTransport we chdir to the TEST_ROOT for this test.
 
1979
        """
 
1980
        os.chdir(TestCaseWithMemoryTransport.TEST_ROOT)
 
1981
        self.test_dir = TestCaseWithMemoryTransport.TEST_ROOT
 
1982
        self.test_home_dir = self.test_dir + "/MemoryTransportMissingHomeDir"
 
1983
        
 
1984
    def make_branch(self, relpath, format=None):
 
1985
        """Create a branch on the transport at relpath."""
 
1986
        repo = self.make_repository(relpath, format=format)
 
1987
        return repo.bzrdir.create_branch()
 
1988
 
 
1989
    def make_bzrdir(self, relpath, format=None):
 
1990
        try:
 
1991
            # might be a relative or absolute path
 
1992
            maybe_a_url = self.get_url(relpath)
 
1993
            segments = maybe_a_url.rsplit('/', 1)
 
1994
            t = get_transport(maybe_a_url)
 
1995
            if len(segments) > 1 and segments[-1] not in ('', '.'):
 
1996
                t.ensure_base()
 
1997
            if format is None:
 
1998
                format = 'default'
 
1999
            if isinstance(format, basestring):
 
2000
                format = bzrdir.format_registry.make_bzrdir(format)
 
2001
            return format.initialize_on_transport(t)
 
2002
        except errors.UninitializableFormat:
 
2003
            raise TestSkipped("Format %s is not initializable." % format)
 
2004
 
 
2005
    def make_repository(self, relpath, shared=False, format=None):
 
2006
        """Create a repository on our default transport at relpath.
 
2007
        
 
2008
        Note that relpath must be a relative path, not a full url.
 
2009
        """
 
2010
        # FIXME: If you create a remoterepository this returns the underlying
 
2011
        # real format, which is incorrect.  Actually we should make sure that 
 
2012
        # RemoteBzrDir returns a RemoteRepository.
 
2013
        # maybe  mbp 20070410
 
2014
        made_control = self.make_bzrdir(relpath, format=format)
 
2015
        return made_control.create_repository(shared=shared)
 
2016
 
 
2017
    def make_branch_and_memory_tree(self, relpath, format=None):
 
2018
        """Create a branch on the default transport and a MemoryTree for it."""
 
2019
        b = self.make_branch(relpath, format=format)
 
2020
        return memorytree.MemoryTree.create_on_branch(b)
 
2021
 
 
2022
    def make_branch_builder(self, relpath, format=None):
 
2023
        url = self.get_url(relpath)
 
2024
        tran = get_transport(url)
 
2025
        return branchbuilder.BranchBuilder(get_transport(url), format=format)
 
2026
 
 
2027
    def overrideEnvironmentForTesting(self):
 
2028
        os.environ['HOME'] = self.test_home_dir
 
2029
        os.environ['BZR_HOME'] = self.test_home_dir
 
2030
        
 
2031
    def setUp(self):
 
2032
        super(TestCaseWithMemoryTransport, self).setUp()
 
2033
        self._make_test_root()
 
2034
        _currentdir = os.getcwdu()
 
2035
        def _leaveDirectory():
 
2036
            os.chdir(_currentdir)
 
2037
        self.addCleanup(_leaveDirectory)
 
2038
        self.makeAndChdirToTestDir()
 
2039
        self.overrideEnvironmentForTesting()
 
2040
        self.__readonly_server = None
 
2041
        self.__server = None
 
2042
        self.reduceLockdirTimeout()
 
2043
 
 
2044
     
 
2045
class TestCaseInTempDir(TestCaseWithMemoryTransport):
 
2046
    """Derived class that runs a test within a temporary directory.
 
2047
 
 
2048
    This is useful for tests that need to create a branch, etc.
 
2049
 
 
2050
    The directory is created in a slightly complex way: for each
 
2051
    Python invocation, a new temporary top-level directory is created.
 
2052
    All test cases create their own directory within that.  If the
 
2053
    tests complete successfully, the directory is removed.
 
2054
 
 
2055
    :ivar test_base_dir: The path of the top-level directory for this 
 
2056
    test, which contains a home directory and a work directory.
 
2057
 
 
2058
    :ivar test_home_dir: An initially empty directory under test_base_dir
 
2059
    which is used as $HOME for this test.
 
2060
 
 
2061
    :ivar test_dir: A directory under test_base_dir used as the current
 
2062
    directory when the test proper is run.
 
2063
    """
 
2064
 
 
2065
    OVERRIDE_PYTHON = 'python'
 
2066
 
 
2067
    def check_file_contents(self, filename, expect):
 
2068
        self.log("check contents of file %s" % filename)
 
2069
        contents = file(filename, 'r').read()
 
2070
        if contents != expect:
 
2071
            self.log("expected: %r" % expect)
 
2072
            self.log("actually: %r" % contents)
 
2073
            self.fail("contents of %s not as expected" % filename)
 
2074
 
 
2075
    def _getTestDirPrefix(self):
 
2076
        # create a directory within the top level test directory
 
2077
        if sys.platform == 'win32':
 
2078
            name_prefix = re.sub('[<>*=+",:;_/\\-]', '_', self.id())
 
2079
            # windows is likely to have path-length limits so use a short name
 
2080
            name_prefix = name_prefix[-30:]
 
2081
        else:
 
2082
            name_prefix = re.sub('[/]', '_', self.id())
 
2083
        return name_prefix
 
2084
 
 
2085
    def makeAndChdirToTestDir(self):
 
2086
        """See TestCaseWithMemoryTransport.makeAndChdirToTestDir().
 
2087
        
 
2088
        For TestCaseInTempDir we create a temporary directory based on the test
 
2089
        name and then create two subdirs - test and home under it.
 
2090
        """
 
2091
        name_prefix = osutils.pathjoin(self.TEST_ROOT, self._getTestDirPrefix())
 
2092
        name = name_prefix
 
2093
        for i in range(100):
 
2094
            if os.path.exists(name):
 
2095
                name = name_prefix + '_' + str(i)
 
2096
            else:
 
2097
                os.mkdir(name)
 
2098
                break
 
2099
        # now create test and home directories within this dir
 
2100
        self.test_base_dir = name
 
2101
        self.test_home_dir = self.test_base_dir + '/home'
 
2102
        os.mkdir(self.test_home_dir)
 
2103
        self.test_dir = self.test_base_dir + '/work'
 
2104
        os.mkdir(self.test_dir)
 
2105
        os.chdir(self.test_dir)
 
2106
        # put name of test inside
 
2107
        f = file(self.test_base_dir + '/name', 'w')
 
2108
        try:
 
2109
            f.write(self.id())
 
2110
        finally:
 
2111
            f.close()
 
2112
        self.addCleanup(self.deleteTestDir)
 
2113
 
 
2114
    def deleteTestDir(self):
 
2115
        os.chdir(self.TEST_ROOT)
 
2116
        _rmtree_temp_dir(self.test_base_dir)
 
2117
 
 
2118
    def build_tree(self, shape, line_endings='binary', transport=None):
 
2119
        """Build a test tree according to a pattern.
 
2120
 
 
2121
        shape is a sequence of file specifications.  If the final
 
2122
        character is '/', a directory is created.
 
2123
 
 
2124
        This assumes that all the elements in the tree being built are new.
 
2125
 
 
2126
        This doesn't add anything to a branch.
 
2127
 
 
2128
        :type shape:    list or tuple.
 
2129
        :param line_endings: Either 'binary' or 'native'
 
2130
            in binary mode, exact contents are written in native mode, the
 
2131
            line endings match the default platform endings.
 
2132
        :param transport: A transport to write to, for building trees on VFS's.
 
2133
            If the transport is readonly or None, "." is opened automatically.
 
2134
        :return: None
 
2135
        """
 
2136
        if type(shape) not in (list, tuple):
 
2137
            raise AssertionError("Parameter 'shape' should be "
 
2138
                "a list or a tuple. Got %r instead" % (shape,))
 
2139
        # It's OK to just create them using forward slashes on windows.
 
2140
        if transport is None or transport.is_readonly():
 
2141
            transport = get_transport(".")
 
2142
        for name in shape:
 
2143
            self.assertIsInstance(name, basestring)
 
2144
            if name[-1] == '/':
 
2145
                transport.mkdir(urlutils.escape(name[:-1]))
 
2146
            else:
 
2147
                if line_endings == 'binary':
 
2148
                    end = '\n'
 
2149
                elif line_endings == 'native':
 
2150
                    end = os.linesep
 
2151
                else:
 
2152
                    raise errors.BzrError(
 
2153
                        'Invalid line ending request %r' % line_endings)
 
2154
                content = "contents of %s%s" % (name.encode('utf-8'), end)
 
2155
                transport.put_bytes_non_atomic(urlutils.escape(name), content)
 
2156
 
 
2157
    def build_tree_contents(self, shape):
 
2158
        build_tree_contents(shape)
 
2159
 
 
2160
    def assertInWorkingTree(self, path, root_path='.', tree=None):
 
2161
        """Assert whether path or paths are in the WorkingTree"""
 
2162
        if tree is None:
 
2163
            tree = workingtree.WorkingTree.open(root_path)
 
2164
        if not isinstance(path, basestring):
 
2165
            for p in path:
 
2166
                self.assertInWorkingTree(p, tree=tree)
 
2167
        else:
 
2168
            self.assertIsNot(tree.path2id(path), None,
 
2169
                path+' not in working tree.')
 
2170
 
 
2171
    def assertNotInWorkingTree(self, path, root_path='.', tree=None):
 
2172
        """Assert whether path or paths are not in the WorkingTree"""
 
2173
        if tree is None:
 
2174
            tree = workingtree.WorkingTree.open(root_path)
 
2175
        if not isinstance(path, basestring):
 
2176
            for p in path:
 
2177
                self.assertNotInWorkingTree(p,tree=tree)
 
2178
        else:
 
2179
            self.assertIs(tree.path2id(path), None, path+' in working tree.')
 
2180
 
 
2181
 
 
2182
class TestCaseWithTransport(TestCaseInTempDir):
 
2183
    """A test case that provides get_url and get_readonly_url facilities.
 
2184
 
 
2185
    These back onto two transport servers, one for readonly access and one for
 
2186
    read write access.
 
2187
 
 
2188
    If no explicit class is provided for readonly access, a
 
2189
    ReadonlyTransportDecorator is used instead which allows the use of non disk
 
2190
    based read write transports.
 
2191
 
 
2192
    If an explicit class is provided for readonly access, that server and the 
 
2193
    readwrite one must both define get_url() as resolving to os.getcwd().
 
2194
    """
 
2195
 
 
2196
    def get_vfs_only_server(self):
 
2197
        """See TestCaseWithMemoryTransport.
 
2198
 
 
2199
        This is useful for some tests with specific servers that need
 
2200
        diagnostics.
 
2201
        """
 
2202
        if self.__vfs_server is None:
 
2203
            self.__vfs_server = self.vfs_transport_factory()
 
2204
            self.__vfs_server.setUp()
 
2205
            self.addCleanup(self.__vfs_server.tearDown)
 
2206
        return self.__vfs_server
 
2207
 
 
2208
    def make_branch_and_tree(self, relpath, format=None):
 
2209
        """Create a branch on the transport and a tree locally.
 
2210
 
 
2211
        If the transport is not a LocalTransport, the Tree can't be created on
 
2212
        the transport.  In that case if the vfs_transport_factory is
 
2213
        LocalURLServer the working tree is created in the local
 
2214
        directory backing the transport, and the returned tree's branch and
 
2215
        repository will also be accessed locally. Otherwise a lightweight
 
2216
        checkout is created and returned.
 
2217
 
 
2218
        :param format: The BzrDirFormat.
 
2219
        :returns: the WorkingTree.
 
2220
        """
 
2221
        # TODO: always use the local disk path for the working tree,
 
2222
        # this obviously requires a format that supports branch references
 
2223
        # so check for that by checking bzrdir.BzrDirFormat.get_default_format()
 
2224
        # RBC 20060208
 
2225
        b = self.make_branch(relpath, format=format)
 
2226
        try:
 
2227
            return b.bzrdir.create_workingtree()
 
2228
        except errors.NotLocalUrl:
 
2229
            # We can only make working trees locally at the moment.  If the
 
2230
            # transport can't support them, then we keep the non-disk-backed
 
2231
            # branch and create a local checkout.
 
2232
            if self.vfs_transport_factory is LocalURLServer:
 
2233
                # the branch is colocated on disk, we cannot create a checkout.
 
2234
                # hopefully callers will expect this.
 
2235
                local_controldir= bzrdir.BzrDir.open(self.get_vfs_only_url(relpath))
 
2236
                wt = local_controldir.create_workingtree()
 
2237
                if wt.branch._format != b._format:
 
2238
                    wt._branch = b
 
2239
                    # Make sure that assigning to wt._branch fixes wt.branch,
 
2240
                    # in case the implementation details of workingtree objects
 
2241
                    # change.
 
2242
                    self.assertIs(b, wt.branch)
 
2243
                return wt
 
2244
            else:
 
2245
                return b.create_checkout(relpath, lightweight=True)
 
2246
 
 
2247
    def assertIsDirectory(self, relpath, transport):
 
2248
        """Assert that relpath within transport is a directory.
 
2249
 
 
2250
        This may not be possible on all transports; in that case it propagates
 
2251
        a TransportNotPossible.
 
2252
        """
 
2253
        try:
 
2254
            mode = transport.stat(relpath).st_mode
 
2255
        except errors.NoSuchFile:
 
2256
            self.fail("path %s is not a directory; no such file"
 
2257
                      % (relpath))
 
2258
        if not stat.S_ISDIR(mode):
 
2259
            self.fail("path %s is not a directory; has mode %#o"
 
2260
                      % (relpath, mode))
 
2261
 
 
2262
    def assertTreesEqual(self, left, right):
 
2263
        """Check that left and right have the same content and properties."""
 
2264
        # we use a tree delta to check for equality of the content, and we
 
2265
        # manually check for equality of other things such as the parents list.
 
2266
        self.assertEqual(left.get_parent_ids(), right.get_parent_ids())
 
2267
        differences = left.changes_from(right)
 
2268
        self.assertFalse(differences.has_changed(),
 
2269
            "Trees %r and %r are different: %r" % (left, right, differences))
 
2270
 
 
2271
    def setUp(self):
 
2272
        super(TestCaseWithTransport, self).setUp()
 
2273
        self.__vfs_server = None
 
2274
 
 
2275
 
 
2276
class ChrootedTestCase(TestCaseWithTransport):
 
2277
    """A support class that provides readonly urls outside the local namespace.
 
2278
 
 
2279
    This is done by checking if self.transport_server is a MemoryServer. if it
 
2280
    is then we are chrooted already, if it is not then an HttpServer is used
 
2281
    for readonly urls.
 
2282
 
 
2283
    TODO RBC 20060127: make this an option to TestCaseWithTransport so it can
 
2284
                       be used without needed to redo it when a different 
 
2285
                       subclass is in use ?
 
2286
    """
 
2287
 
 
2288
    def setUp(self):
 
2289
        super(ChrootedTestCase, self).setUp()
 
2290
        if not self.vfs_transport_factory == MemoryServer:
 
2291
            self.transport_readonly_server = HttpServer
 
2292
 
 
2293
 
 
2294
def condition_id_re(pattern):
 
2295
    """Create a condition filter which performs a re check on a test's id.
 
2296
    
 
2297
    :param pattern: A regular expression string.
 
2298
    :return: A callable that returns True if the re matches.
 
2299
    """
 
2300
    filter_re = re.compile(pattern)
 
2301
    def condition(test):
 
2302
        test_id = test.id()
 
2303
        return filter_re.search(test_id)
 
2304
    return condition
 
2305
 
 
2306
 
 
2307
def condition_isinstance(klass_or_klass_list):
 
2308
    """Create a condition filter which returns isinstance(param, klass).
 
2309
    
 
2310
    :return: A callable which when called with one parameter obj return the
 
2311
        result of isinstance(obj, klass_or_klass_list).
 
2312
    """
 
2313
    def condition(obj):
 
2314
        return isinstance(obj, klass_or_klass_list)
 
2315
    return condition
 
2316
 
 
2317
 
 
2318
def condition_id_in_list(id_list):
 
2319
    """Create a condition filter which verify that test's id in a list.
 
2320
    
 
2321
    :param id_list: A TestIdList object.
 
2322
    :return: A callable that returns True if the test's id appears in the list.
 
2323
    """
 
2324
    def condition(test):
 
2325
        return id_list.includes(test.id())
 
2326
    return condition
 
2327
 
 
2328
 
 
2329
def condition_id_startswith(starts):
 
2330
    """Create a condition filter verifying that test's id starts with a string.
 
2331
    
 
2332
    :param starts: A list of string.
 
2333
    :return: A callable that returns True if the test's id starts with one of 
 
2334
        the given strings.
 
2335
    """
 
2336
    def condition(test):
 
2337
        for start in starts:
 
2338
            if test.id().startswith(start):
 
2339
                return True
 
2340
        return False
 
2341
    return condition
 
2342
 
 
2343
 
 
2344
def exclude_tests_by_condition(suite, condition):
 
2345
    """Create a test suite which excludes some tests from suite.
 
2346
 
 
2347
    :param suite: The suite to get tests from.
 
2348
    :param condition: A callable whose result evaluates True when called with a
 
2349
        test case which should be excluded from the result.
 
2350
    :return: A suite which contains the tests found in suite that fail
 
2351
        condition.
 
2352
    """
 
2353
    result = []
 
2354
    for test in iter_suite_tests(suite):
 
2355
        if not condition(test):
 
2356
            result.append(test)
 
2357
    return TestUtil.TestSuite(result)
 
2358
 
 
2359
 
 
2360
def filter_suite_by_condition(suite, condition):
 
2361
    """Create a test suite by filtering another one.
 
2362
    
 
2363
    :param suite: The source suite.
 
2364
    :param condition: A callable whose result evaluates True when called with a
 
2365
        test case which should be included in the result.
 
2366
    :return: A suite which contains the tests found in suite that pass
 
2367
        condition.
 
2368
    """
 
2369
    result = []
 
2370
    for test in iter_suite_tests(suite):
 
2371
        if condition(test):
 
2372
            result.append(test)
 
2373
    return TestUtil.TestSuite(result)
 
2374
 
 
2375
 
 
2376
def filter_suite_by_re(suite, pattern):
 
2377
    """Create a test suite by filtering another one.
 
2378
    
 
2379
    :param suite:           the source suite
 
2380
    :param pattern:         pattern that names must match
 
2381
    :returns: the newly created suite
 
2382
    """
 
2383
    condition = condition_id_re(pattern)
 
2384
    result_suite = filter_suite_by_condition(suite, condition)
 
2385
    return result_suite
 
2386
 
 
2387
 
 
2388
def filter_suite_by_id_list(suite, test_id_list):
 
2389
    """Create a test suite by filtering another one.
 
2390
 
 
2391
    :param suite: The source suite.
 
2392
    :param test_id_list: A list of the test ids to keep as strings.
 
2393
    :returns: the newly created suite
 
2394
    """
 
2395
    condition = condition_id_in_list(test_id_list)
 
2396
    result_suite = filter_suite_by_condition(suite, condition)
 
2397
    return result_suite
 
2398
 
 
2399
 
 
2400
def filter_suite_by_id_startswith(suite, start):
 
2401
    """Create a test suite by filtering another one.
 
2402
 
 
2403
    :param suite: The source suite.
 
2404
    :param start: A list of string the test id must start with one of.
 
2405
    :returns: the newly created suite
 
2406
    """
 
2407
    condition = condition_id_startswith(start)
 
2408
    result_suite = filter_suite_by_condition(suite, condition)
 
2409
    return result_suite
 
2410
 
 
2411
 
 
2412
def exclude_tests_by_re(suite, pattern):
 
2413
    """Create a test suite which excludes some tests from suite.
 
2414
 
 
2415
    :param suite: The suite to get tests from.
 
2416
    :param pattern: A regular expression string. Test ids that match this
 
2417
        pattern will be excluded from the result.
 
2418
    :return: A TestSuite that contains all the tests from suite without the
 
2419
        tests that matched pattern. The order of tests is the same as it was in
 
2420
        suite.
 
2421
    """
 
2422
    return exclude_tests_by_condition(suite, condition_id_re(pattern))
 
2423
 
 
2424
 
 
2425
def preserve_input(something):
 
2426
    """A helper for performing test suite transformation chains.
 
2427
 
 
2428
    :param something: Anything you want to preserve.
 
2429
    :return: Something.
 
2430
    """
 
2431
    return something
 
2432
 
 
2433
 
 
2434
def randomize_suite(suite):
 
2435
    """Return a new TestSuite with suite's tests in random order.
 
2436
    
 
2437
    The tests in the input suite are flattened into a single suite in order to
 
2438
    accomplish this. Any nested TestSuites are removed to provide global
 
2439
    randomness.
 
2440
    """
 
2441
    tests = list(iter_suite_tests(suite))
 
2442
    random.shuffle(tests)
 
2443
    return TestUtil.TestSuite(tests)
 
2444
 
 
2445
 
 
2446
def split_suite_by_condition(suite, condition):
 
2447
    """Split a test suite into two by a condition.
 
2448
    
 
2449
    :param suite: The suite to split.
 
2450
    :param condition: The condition to match on. Tests that match this
 
2451
        condition are returned in the first test suite, ones that do not match
 
2452
        are in the second suite.
 
2453
    :return: A tuple of two test suites, where the first contains tests from
 
2454
        suite matching the condition, and the second contains the remainder
 
2455
        from suite. The order within each output suite is the same as it was in
 
2456
        suite.
 
2457
    """
 
2458
    matched = []
 
2459
    did_not_match = []
 
2460
    for test in iter_suite_tests(suite):
 
2461
        if condition(test):
 
2462
            matched.append(test)
 
2463
        else:
 
2464
            did_not_match.append(test)
 
2465
    return TestUtil.TestSuite(matched), TestUtil.TestSuite(did_not_match)
 
2466
 
 
2467
 
 
2468
def split_suite_by_re(suite, pattern):
 
2469
    """Split a test suite into two by a regular expression.
 
2470
    
 
2471
    :param suite: The suite to split.
 
2472
    :param pattern: A regular expression string. Test ids that match this
 
2473
        pattern will be in the first test suite returned, and the others in the
 
2474
        second test suite returned.
 
2475
    :return: A tuple of two test suites, where the first contains tests from
 
2476
        suite matching pattern, and the second contains the remainder from
 
2477
        suite. The order within each output suite is the same as it was in
 
2478
        suite.
 
2479
    """
 
2480
    return split_suite_by_condition(suite, condition_id_re(pattern))
 
2481
 
 
2482
 
 
2483
def run_suite(suite, name='test', verbose=False, pattern=".*",
 
2484
              stop_on_failure=False,
 
2485
              transport=None, lsprof_timed=None, bench_history=None,
 
2486
              matching_tests_first=None,
 
2487
              list_only=False,
 
2488
              random_seed=None,
 
2489
              exclude_pattern=None,
 
2490
              strict=False,
 
2491
              runner_class=None):
 
2492
    """Run a test suite for bzr selftest.
 
2493
 
 
2494
    :param runner_class: The class of runner to use. Must support the
 
2495
        constructor arguments passed by run_suite which are more than standard
 
2496
        python uses.
 
2497
    :return: A boolean indicating success.
 
2498
    """
 
2499
    TestCase._gather_lsprof_in_benchmarks = lsprof_timed
 
2500
    if verbose:
 
2501
        verbosity = 2
 
2502
    else:
 
2503
        verbosity = 1
 
2504
    if runner_class is None:
 
2505
        runner_class = TextTestRunner
 
2506
    runner = runner_class(stream=sys.stdout,
 
2507
                            descriptions=0,
 
2508
                            verbosity=verbosity,
 
2509
                            bench_history=bench_history,
 
2510
                            list_only=list_only,
 
2511
                            )
 
2512
    runner.stop_on_failure=stop_on_failure
 
2513
    # Initialise the random number generator and display the seed used.
 
2514
    # We convert the seed to a long to make it reuseable across invocations.
 
2515
    random_order = False
 
2516
    if random_seed is not None:
 
2517
        random_order = True
 
2518
        if random_seed == "now":
 
2519
            random_seed = long(time.time())
 
2520
        else:
 
2521
            # Convert the seed to a long if we can
 
2522
            try:
 
2523
                random_seed = long(random_seed)
 
2524
            except:
 
2525
                pass
 
2526
        runner.stream.writeln("Randomizing test order using seed %s\n" %
 
2527
            (random_seed))
 
2528
        random.seed(random_seed)
 
2529
    # Customise the list of tests if requested
 
2530
    if exclude_pattern is not None:
 
2531
        suite = exclude_tests_by_re(suite, exclude_pattern)
 
2532
    if random_order:
 
2533
        order_changer = randomize_suite
 
2534
    else:
 
2535
        order_changer = preserve_input
 
2536
    if pattern != '.*' or random_order:
 
2537
        if matching_tests_first:
 
2538
            suites = map(order_changer, split_suite_by_re(suite, pattern))
 
2539
            suite = TestUtil.TestSuite(suites)
 
2540
        else:
 
2541
            suite = order_changer(filter_suite_by_re(suite, pattern))
 
2542
 
 
2543
    result = runner.run(suite)
 
2544
 
 
2545
    if strict:
 
2546
        return result.wasStrictlySuccessful()
 
2547
 
 
2548
    return result.wasSuccessful()
 
2549
 
 
2550
 
 
2551
# Controlled by "bzr selftest -E=..." option
 
2552
selftest_debug_flags = set()
 
2553
 
 
2554
 
 
2555
def selftest(verbose=False, pattern=".*", stop_on_failure=True,
 
2556
             transport=None,
 
2557
             test_suite_factory=None,
 
2558
             lsprof_timed=None,
 
2559
             bench_history=None,
 
2560
             matching_tests_first=None,
 
2561
             list_only=False,
 
2562
             random_seed=None,
 
2563
             exclude_pattern=None,
 
2564
             strict=False,
 
2565
             load_list=None,
 
2566
             debug_flags=None,
 
2567
             starting_with=None,
 
2568
             runner_class=None,
 
2569
             ):
 
2570
    """Run the whole test suite under the enhanced runner"""
 
2571
    # XXX: Very ugly way to do this...
 
2572
    # Disable warning about old formats because we don't want it to disturb
 
2573
    # any blackbox tests.
 
2574
    from bzrlib import repository
 
2575
    repository._deprecation_warning_done = True
 
2576
 
 
2577
    global default_transport
 
2578
    if transport is None:
 
2579
        transport = default_transport
 
2580
    old_transport = default_transport
 
2581
    default_transport = transport
 
2582
    global selftest_debug_flags
 
2583
    old_debug_flags = selftest_debug_flags
 
2584
    if debug_flags is not None:
 
2585
        selftest_debug_flags = set(debug_flags)
 
2586
    try:
 
2587
        if load_list is None:
 
2588
            keep_only = None
 
2589
        else:
 
2590
            keep_only = load_test_id_list(load_list)
 
2591
        if test_suite_factory is None:
 
2592
            suite = test_suite(keep_only, starting_with)
 
2593
        else:
 
2594
            suite = test_suite_factory()
 
2595
        return run_suite(suite, 'testbzr', verbose=verbose, pattern=pattern,
 
2596
                     stop_on_failure=stop_on_failure,
 
2597
                     transport=transport,
 
2598
                     lsprof_timed=lsprof_timed,
 
2599
                     bench_history=bench_history,
 
2600
                     matching_tests_first=matching_tests_first,
 
2601
                     list_only=list_only,
 
2602
                     random_seed=random_seed,
 
2603
                     exclude_pattern=exclude_pattern,
 
2604
                     strict=strict,
 
2605
                     runner_class=runner_class,
 
2606
                     )
 
2607
    finally:
 
2608
        default_transport = old_transport
 
2609
        selftest_debug_flags = old_debug_flags
 
2610
 
 
2611
 
 
2612
def load_test_id_list(file_name):
 
2613
    """Load a test id list from a text file.
 
2614
 
 
2615
    The format is one test id by line.  No special care is taken to impose
 
2616
    strict rules, these test ids are used to filter the test suite so a test id
 
2617
    that do not match an existing test will do no harm. This allows user to add
 
2618
    comments, leave blank lines, etc.
 
2619
    """
 
2620
    test_list = []
 
2621
    try:
 
2622
        ftest = open(file_name, 'rt')
 
2623
    except IOError, e:
 
2624
        if e.errno != errno.ENOENT:
 
2625
            raise
 
2626
        else:
 
2627
            raise errors.NoSuchFile(file_name)
 
2628
 
 
2629
    for test_name in ftest.readlines():
 
2630
        test_list.append(test_name.strip())
 
2631
    ftest.close()
 
2632
    return test_list
 
2633
 
 
2634
 
 
2635
def suite_matches_id_list(test_suite, id_list):
 
2636
    """Warns about tests not appearing or appearing more than once.
 
2637
 
 
2638
    :param test_suite: A TestSuite object.
 
2639
    :param test_id_list: The list of test ids that should be found in 
 
2640
         test_suite.
 
2641
 
 
2642
    :return: (absents, duplicates) absents is a list containing the test found
 
2643
        in id_list but not in test_suite, duplicates is a list containing the
 
2644
        test found multiple times in test_suite.
 
2645
 
 
2646
    When using a prefined test id list, it may occurs that some tests do not
 
2647
    exist anymore or that some tests use the same id. This function warns the
 
2648
    tester about potential problems in his workflow (test lists are volatile)
 
2649
    or in the test suite itself (using the same id for several tests does not
 
2650
    help to localize defects).
 
2651
    """
 
2652
    # Build a dict counting id occurrences
 
2653
    tests = dict()
 
2654
    for test in iter_suite_tests(test_suite):
 
2655
        id = test.id()
 
2656
        tests[id] = tests.get(id, 0) + 1
 
2657
 
 
2658
    not_found = []
 
2659
    duplicates = []
 
2660
    for id in id_list:
 
2661
        occurs = tests.get(id, 0)
 
2662
        if not occurs:
 
2663
            not_found.append(id)
 
2664
        elif occurs > 1:
 
2665
            duplicates.append(id)
 
2666
 
 
2667
    return not_found, duplicates
 
2668
 
 
2669
 
 
2670
class TestIdList(object):
 
2671
    """Test id list to filter a test suite.
 
2672
 
 
2673
    Relying on the assumption that test ids are built as:
 
2674
    <module>[.<class>.<method>][(<param>+)], <module> being in python dotted
 
2675
    notation, this class offers methods to :
 
2676
    - avoid building a test suite for modules not refered to in the test list,
 
2677
    - keep only the tests listed from the module test suite.
 
2678
    """
 
2679
 
 
2680
    def __init__(self, test_id_list):
 
2681
        # When a test suite needs to be filtered against us we compare test ids
 
2682
        # for equality, so a simple dict offers a quick and simple solution.
 
2683
        self.tests = dict().fromkeys(test_id_list, True)
 
2684
 
 
2685
        # While unittest.TestCase have ids like:
 
2686
        # <module>.<class>.<method>[(<param+)],
 
2687
        # doctest.DocTestCase can have ids like:
 
2688
        # <module>
 
2689
        # <module>.<class>
 
2690
        # <module>.<function>
 
2691
        # <module>.<class>.<method>
 
2692
 
 
2693
        # Since we can't predict a test class from its name only, we settle on
 
2694
        # a simple constraint: a test id always begins with its module name.
 
2695
 
 
2696
        modules = {}
 
2697
        for test_id in test_id_list:
 
2698
            parts = test_id.split('.')
 
2699
            mod_name = parts.pop(0)
 
2700
            modules[mod_name] = True
 
2701
            for part in parts:
 
2702
                mod_name += '.' + part
 
2703
                modules[mod_name] = True
 
2704
        self.modules = modules
 
2705
 
 
2706
    def refers_to(self, module_name):
 
2707
        """Is there tests for the module or one of its sub modules."""
 
2708
        return self.modules.has_key(module_name)
 
2709
 
 
2710
    def includes(self, test_id):
 
2711
        return self.tests.has_key(test_id)
 
2712
 
 
2713
 
 
2714
class TestPrefixAliasRegistry(registry.Registry):
 
2715
    """A registry for test prefix aliases.
 
2716
 
 
2717
    This helps implement shorcuts for the --starting-with selftest
 
2718
    option. Overriding existing prefixes is not allowed but not fatal (a
 
2719
    warning will be emitted).
 
2720
    """
 
2721
 
 
2722
    def register(self, key, obj, help=None, info=None,
 
2723
                 override_existing=False):
 
2724
        """See Registry.register.
 
2725
 
 
2726
        Trying to override an existing alias causes a warning to be emitted,
 
2727
        not a fatal execption.
 
2728
        """
 
2729
        try:
 
2730
            super(TestPrefixAliasRegistry, self).register(
 
2731
                key, obj, help=help, info=info, override_existing=False)
 
2732
        except KeyError:
 
2733
            actual = self.get(key)
 
2734
            note('Test prefix alias %s is already used for %s, ignoring %s'
 
2735
                 % (key, actual, obj))
 
2736
 
 
2737
    def resolve_alias(self, id_start):
 
2738
        """Replace the alias by the prefix in the given string.
 
2739
 
 
2740
        Using an unknown prefix is an error to help catching typos.
 
2741
        """
 
2742
        parts = id_start.split('.')
 
2743
        try:
 
2744
            parts[0] = self.get(parts[0])
 
2745
        except KeyError:
 
2746
            raise errors.BzrCommandError(
 
2747
                '%s is not a known test prefix alias' % parts[0])
 
2748
        return '.'.join(parts)
 
2749
 
 
2750
 
 
2751
test_prefix_alias_registry = TestPrefixAliasRegistry()
 
2752
"""Registry of test prefix aliases."""
 
2753
 
 
2754
 
 
2755
# This alias allows to detect typos ('bzrlin.') by making all valid test ids
 
2756
# appear prefixed ('bzrlib.' is "replaced" by 'bzrlib.').
 
2757
test_prefix_alias_registry.register('bzrlib', 'bzrlib')
 
2758
 
 
2759
# Obvious higest levels prefixes, feel free to add your own via a plugin
 
2760
test_prefix_alias_registry.register('bd', 'bzrlib.doc')
 
2761
test_prefix_alias_registry.register('bu', 'bzrlib.utils')
 
2762
test_prefix_alias_registry.register('bt', 'bzrlib.tests')
 
2763
test_prefix_alias_registry.register('bb', 'bzrlib.tests.blackbox')
 
2764
test_prefix_alias_registry.register('bp', 'bzrlib.plugins')
 
2765
 
 
2766
 
 
2767
def test_suite(keep_only=None, starting_with=None):
 
2768
    """Build and return TestSuite for the whole of bzrlib.
 
2769
 
 
2770
    :param keep_only: A list of test ids limiting the suite returned.
 
2771
 
 
2772
    :param starting_with: An id limiting the suite returned to the tests
 
2773
         starting with it.
 
2774
 
 
2775
    This function can be replaced if you need to change the default test
 
2776
    suite on a global basis, but it is not encouraged.
 
2777
    """
 
2778
    testmod_names = [
 
2779
                   'bzrlib.doc',
 
2780
                   'bzrlib.tests.blackbox',
 
2781
                   'bzrlib.tests.branch_implementations',
 
2782
                   'bzrlib.tests.bzrdir_implementations',
 
2783
                   'bzrlib.tests.commands',
 
2784
                   'bzrlib.tests.interrepository_implementations',
 
2785
                   'bzrlib.tests.intertree_implementations',
 
2786
                   'bzrlib.tests.inventory_implementations',
 
2787
                   'bzrlib.tests.per_lock',
 
2788
                   'bzrlib.tests.per_repository',
 
2789
                   'bzrlib.tests.per_repository_reference',
 
2790
                   'bzrlib.tests.test__dirstate_helpers',
 
2791
                   'bzrlib.tests.test__walkdirs_win32',
 
2792
                   'bzrlib.tests.test_ancestry',
 
2793
                   'bzrlib.tests.test_annotate',
 
2794
                   'bzrlib.tests.test_api',
 
2795
                   'bzrlib.tests.test_atomicfile',
 
2796
                   'bzrlib.tests.test_bad_files',
 
2797
                   'bzrlib.tests.test_bisect_multi',
 
2798
                   'bzrlib.tests.test_branch',
 
2799
                   'bzrlib.tests.test_branchbuilder',
 
2800
                   'bzrlib.tests.test_btree_index',
 
2801
                   'bzrlib.tests.test_bugtracker',
 
2802
                   'bzrlib.tests.test_bundle',
 
2803
                   'bzrlib.tests.test_bzrdir',
 
2804
                   'bzrlib.tests.test_cache_utf8',
 
2805
                   'bzrlib.tests.test_chunk_writer',
 
2806
                   'bzrlib.tests.test__chunks_to_lines',
 
2807
                   'bzrlib.tests.test_commands',
 
2808
                   'bzrlib.tests.test_commit',
 
2809
                   'bzrlib.tests.test_commit_merge',
 
2810
                   'bzrlib.tests.test_config',
 
2811
                   'bzrlib.tests.test_conflicts',
 
2812
                   'bzrlib.tests.test_counted_lock',
 
2813
                   'bzrlib.tests.test_decorators',
 
2814
                   'bzrlib.tests.test_delta',
 
2815
                   'bzrlib.tests.test_deprecated_graph',
 
2816
                   'bzrlib.tests.test_diff',
 
2817
                   'bzrlib.tests.test_directory_service',
 
2818
                   'bzrlib.tests.test_dirstate',
 
2819
                   'bzrlib.tests.test_email_message',
 
2820
                   'bzrlib.tests.test_errors',
 
2821
                   'bzrlib.tests.test_export',
 
2822
                   'bzrlib.tests.test_extract',
 
2823
                   'bzrlib.tests.test_fetch',
 
2824
                   'bzrlib.tests.test_fifo_cache',
 
2825
                   'bzrlib.tests.test_ftp_transport',
 
2826
                   'bzrlib.tests.test_foreign',
 
2827
                   'bzrlib.tests.test_generate_docs',
 
2828
                   'bzrlib.tests.test_generate_ids',
 
2829
                   'bzrlib.tests.test_globbing',
 
2830
                   'bzrlib.tests.test_gpg',
 
2831
                   'bzrlib.tests.test_graph',
 
2832
                   'bzrlib.tests.test_hashcache',
 
2833
                   'bzrlib.tests.test_help',
 
2834
                   'bzrlib.tests.test_hooks',
 
2835
                   'bzrlib.tests.test_http',
 
2836
                   'bzrlib.tests.test_http_implementations',
 
2837
                   'bzrlib.tests.test_http_response',
 
2838
                   'bzrlib.tests.test_https_ca_bundle',
 
2839
                   'bzrlib.tests.test_identitymap',
 
2840
                   'bzrlib.tests.test_ignores',
 
2841
                   'bzrlib.tests.test_index',
 
2842
                   'bzrlib.tests.test_info',
 
2843
                   'bzrlib.tests.test_inv',
 
2844
                   'bzrlib.tests.test_knit',
 
2845
                   'bzrlib.tests.test_lazy_import',
 
2846
                   'bzrlib.tests.test_lazy_regex',
 
2847
                   'bzrlib.tests.test_lockable_files',
 
2848
                   'bzrlib.tests.test_lockdir',
 
2849
                   'bzrlib.tests.test_log',
 
2850
                   'bzrlib.tests.test_lru_cache',
 
2851
                   'bzrlib.tests.test_lsprof',
 
2852
                   'bzrlib.tests.test_mail_client',
 
2853
                   'bzrlib.tests.test_memorytree',
 
2854
                   'bzrlib.tests.test_merge',
 
2855
                   'bzrlib.tests.test_merge3',
 
2856
                   'bzrlib.tests.test_merge_core',
 
2857
                   'bzrlib.tests.test_merge_directive',
 
2858
                   'bzrlib.tests.test_missing',
 
2859
                   'bzrlib.tests.test_msgeditor',
 
2860
                   'bzrlib.tests.test_multiparent',
 
2861
                   'bzrlib.tests.test_mutabletree',
 
2862
                   'bzrlib.tests.test_nonascii',
 
2863
                   'bzrlib.tests.test_options',
 
2864
                   'bzrlib.tests.test_osutils',
 
2865
                   'bzrlib.tests.test_osutils_encodings',
 
2866
                   'bzrlib.tests.test_pack',
 
2867
                   'bzrlib.tests.test_pack_repository',
 
2868
                   'bzrlib.tests.test_patch',
 
2869
                   'bzrlib.tests.test_patches',
 
2870
                   'bzrlib.tests.test_permissions',
 
2871
                   'bzrlib.tests.test_plugins',
 
2872
                   'bzrlib.tests.test_progress',
 
2873
                   'bzrlib.tests.test_read_bundle',
 
2874
                   'bzrlib.tests.test_reconcile',
 
2875
                   'bzrlib.tests.test_reconfigure',
 
2876
                   'bzrlib.tests.test_registry',
 
2877
                   'bzrlib.tests.test_remote',
 
2878
                   'bzrlib.tests.test_repository',
 
2879
                   'bzrlib.tests.test_revert',
 
2880
                   'bzrlib.tests.test_revision',
 
2881
                   'bzrlib.tests.test_revisionspec',
 
2882
                   'bzrlib.tests.test_revisiontree',
 
2883
                   'bzrlib.tests.test_rio',
 
2884
                   'bzrlib.tests.test_rules',
 
2885
                   'bzrlib.tests.test_sampler',
 
2886
                   'bzrlib.tests.test_selftest',
 
2887
                   'bzrlib.tests.test_setup',
 
2888
                   'bzrlib.tests.test_sftp_transport',
 
2889
                   'bzrlib.tests.test_shelf',
 
2890
                   'bzrlib.tests.test_shelf_ui',
 
2891
                   'bzrlib.tests.test_smart',
 
2892
                   'bzrlib.tests.test_smart_add',
 
2893
                   'bzrlib.tests.test_smart_request',
 
2894
                   'bzrlib.tests.test_smart_transport',
 
2895
                   'bzrlib.tests.test_smtp_connection',
 
2896
                   'bzrlib.tests.test_source',
 
2897
                   'bzrlib.tests.test_ssh_transport',
 
2898
                   'bzrlib.tests.test_status',
 
2899
                   'bzrlib.tests.test_store',
 
2900
                   'bzrlib.tests.test_strace',
 
2901
                   'bzrlib.tests.test_subsume',
 
2902
                   'bzrlib.tests.test_switch',
 
2903
                   'bzrlib.tests.test_symbol_versioning',
 
2904
                   'bzrlib.tests.test_tag',
 
2905
                   'bzrlib.tests.test_testament',
 
2906
                   'bzrlib.tests.test_textfile',
 
2907
                   'bzrlib.tests.test_textmerge',
 
2908
                   'bzrlib.tests.test_timestamp',
 
2909
                   'bzrlib.tests.test_trace',
 
2910
                   'bzrlib.tests.test_transactions',
 
2911
                   'bzrlib.tests.test_transform',
 
2912
                   'bzrlib.tests.test_transport',
 
2913
                   'bzrlib.tests.test_transport_implementations',
 
2914
                   'bzrlib.tests.test_transport_log',
 
2915
                   'bzrlib.tests.test_tree',
 
2916
                   'bzrlib.tests.test_treebuilder',
 
2917
                   'bzrlib.tests.test_tsort',
 
2918
                   'bzrlib.tests.test_tuned_gzip',
 
2919
                   'bzrlib.tests.test_ui',
 
2920
                   'bzrlib.tests.test_uncommit',
 
2921
                   'bzrlib.tests.test_upgrade',
 
2922
                   'bzrlib.tests.test_upgrade_stacked',
 
2923
                   'bzrlib.tests.test_urlutils',
 
2924
                   'bzrlib.tests.test_version',
 
2925
                   'bzrlib.tests.test_version_info',
 
2926
                   'bzrlib.tests.test_versionedfile',
 
2927
                   'bzrlib.tests.test_weave',
 
2928
                   'bzrlib.tests.test_whitebox',
 
2929
                   'bzrlib.tests.test_win32utils',
 
2930
                   'bzrlib.tests.test_workingtree',
 
2931
                   'bzrlib.tests.test_workingtree_4',
 
2932
                   'bzrlib.tests.test_wsgi',
 
2933
                   'bzrlib.tests.test_xml',
 
2934
                   'bzrlib.tests.tree_implementations',
 
2935
                   'bzrlib.tests.workingtree_implementations',
 
2936
                   'bzrlib.util.tests.test_bencode',
 
2937
                   ]
 
2938
 
 
2939
    loader = TestUtil.TestLoader()
 
2940
 
 
2941
    if starting_with:
 
2942
        starting_with = [test_prefix_alias_registry.resolve_alias(start)
 
2943
                         for start in starting_with]
 
2944
        # We take precedence over keep_only because *at loading time* using
 
2945
        # both options means we will load less tests for the same final result.
 
2946
        def interesting_module(name):
 
2947
            for start in starting_with:
 
2948
                if (
 
2949
                    # Either the module name starts with the specified string
 
2950
                    name.startswith(start)
 
2951
                    # or it may contain tests starting with the specified string
 
2952
                    or start.startswith(name)
 
2953
                    ):
 
2954
                    return True
 
2955
            return False
 
2956
        loader = TestUtil.FilteredByModuleTestLoader(interesting_module)
 
2957
 
 
2958
    elif keep_only is not None:
 
2959
        id_filter = TestIdList(keep_only)
 
2960
        loader = TestUtil.FilteredByModuleTestLoader(id_filter.refers_to)
 
2961
        def interesting_module(name):
 
2962
            return id_filter.refers_to(name)
 
2963
 
 
2964
    else:
 
2965
        loader = TestUtil.TestLoader()
 
2966
        def interesting_module(name):
 
2967
            # No filtering, all modules are interesting
 
2968
            return True
 
2969
 
 
2970
    suite = loader.suiteClass()
 
2971
 
 
2972
    # modules building their suite with loadTestsFromModuleNames
 
2973
    suite.addTest(loader.loadTestsFromModuleNames(testmod_names))
 
2974
 
 
2975
    modules_to_doctest = [
 
2976
        'bzrlib',
 
2977
        'bzrlib.branchbuilder',
 
2978
        'bzrlib.export',
 
2979
        'bzrlib.inventory',
 
2980
        'bzrlib.iterablefile',
 
2981
        'bzrlib.lockdir',
 
2982
        'bzrlib.merge3',
 
2983
        'bzrlib.option',
 
2984
        'bzrlib.symbol_versioning',
 
2985
        'bzrlib.tests',
 
2986
        'bzrlib.timestamp',
 
2987
        'bzrlib.version_info_formats.format_custom',
 
2988
        ]
 
2989
 
 
2990
    for mod in modules_to_doctest:
 
2991
        if not interesting_module(mod):
 
2992
            # No tests to keep here, move along
 
2993
            continue
 
2994
        try:
 
2995
            # note that this really does mean "report only" -- doctest 
 
2996
            # still runs the rest of the examples
 
2997
            doc_suite = doctest.DocTestSuite(mod,
 
2998
                optionflags=doctest.REPORT_ONLY_FIRST_FAILURE)
 
2999
        except ValueError, e:
 
3000
            print '**failed to get doctest for: %s\n%s' % (mod, e)
 
3001
            raise
 
3002
        if len(doc_suite._tests) == 0:
 
3003
            raise errors.BzrError("no doctests found in %s" % (mod,))
 
3004
        suite.addTest(doc_suite)
 
3005
 
 
3006
    default_encoding = sys.getdefaultencoding()
 
3007
    for name, plugin in bzrlib.plugin.plugins().items():
 
3008
        if not interesting_module(plugin.module.__name__):
 
3009
            continue
 
3010
        plugin_suite = plugin.test_suite()
 
3011
        # We used to catch ImportError here and turn it into just a warning,
 
3012
        # but really if you don't have --no-plugins this should be a failure.
 
3013
        # mbp 20080213 - see http://bugs.launchpad.net/bugs/189771
 
3014
        if plugin_suite is None:
 
3015
            plugin_suite = plugin.load_plugin_tests(loader)
 
3016
        if plugin_suite is not None:
 
3017
            suite.addTest(plugin_suite)
 
3018
        if default_encoding != sys.getdefaultencoding():
 
3019
            bzrlib.trace.warning(
 
3020
                'Plugin "%s" tried to reset default encoding to: %s', name,
 
3021
                sys.getdefaultencoding())
 
3022
            reload(sys)
 
3023
            sys.setdefaultencoding(default_encoding)
 
3024
 
 
3025
    if starting_with:
 
3026
        suite = filter_suite_by_id_startswith(suite, starting_with)
 
3027
 
 
3028
    if keep_only is not None:
 
3029
        # Now that the referred modules have loaded their tests, keep only the
 
3030
        # requested ones.
 
3031
        suite = filter_suite_by_id_list(suite, id_filter)
 
3032
        # Do some sanity checks on the id_list filtering
 
3033
        not_found, duplicates = suite_matches_id_list(suite, keep_only)
 
3034
        if starting_with:
 
3035
            # The tester has used both keep_only and starting_with, so he is
 
3036
            # already aware that some tests are excluded from the list, there
 
3037
            # is no need to tell him which.
 
3038
            pass
 
3039
        else:
 
3040
            # Some tests mentioned in the list are not in the test suite. The
 
3041
            # list may be out of date, report to the tester.
 
3042
            for id in not_found:
 
3043
                bzrlib.trace.warning('"%s" not found in the test suite', id)
 
3044
        for id in duplicates:
 
3045
            bzrlib.trace.warning('"%s" is used as an id by several tests', id)
 
3046
 
 
3047
    return suite
 
3048
 
 
3049
 
 
3050
def multiply_tests_from_modules(module_name_list, scenario_iter, loader=None):
 
3051
    """Adapt all tests in some given modules to given scenarios.
 
3052
 
 
3053
    This is the recommended public interface for test parameterization.
 
3054
    Typically the test_suite() method for a per-implementation test
 
3055
    suite will call multiply_tests_from_modules and return the 
 
3056
    result.
 
3057
 
 
3058
    :param module_name_list: List of fully-qualified names of test
 
3059
        modules.
 
3060
    :param scenario_iter: Iterable of pairs of (scenario_name, 
 
3061
        scenario_param_dict).
 
3062
    :param loader: If provided, will be used instead of a new 
 
3063
        bzrlib.tests.TestLoader() instance.
 
3064
 
 
3065
    This returns a new TestSuite containing the cross product of
 
3066
    all the tests in all the modules, each repeated for each scenario.
 
3067
    Each test is adapted by adding the scenario name at the end 
 
3068
    of its name, and updating the test object's __dict__ with the
 
3069
    scenario_param_dict.
 
3070
 
 
3071
    >>> r = multiply_tests_from_modules(
 
3072
    ...     ['bzrlib.tests.test_sampler'],
 
3073
    ...     [('one', dict(param=1)), 
 
3074
    ...      ('two', dict(param=2))])
 
3075
    >>> tests = list(iter_suite_tests(r))
 
3076
    >>> len(tests)
 
3077
    2
 
3078
    >>> tests[0].id()
 
3079
    'bzrlib.tests.test_sampler.DemoTest.test_nothing(one)'
 
3080
    >>> tests[0].param
 
3081
    1
 
3082
    >>> tests[1].param
 
3083
    2
 
3084
    """
 
3085
    # XXX: Isn't load_tests() a better way to provide the same functionality
 
3086
    # without forcing a predefined TestScenarioApplier ? --vila 080215
 
3087
    if loader is None:
 
3088
        loader = TestUtil.TestLoader()
 
3089
 
 
3090
    suite = loader.suiteClass()
 
3091
 
 
3092
    adapter = TestScenarioApplier()
 
3093
    adapter.scenarios = list(scenario_iter)
 
3094
    adapt_modules(module_name_list, adapter, loader, suite)
 
3095
    return suite
 
3096
 
 
3097
 
 
3098
def multiply_scenarios(scenarios_left, scenarios_right):
 
3099
    """Multiply two sets of scenarios.
 
3100
 
 
3101
    :returns: the cartesian product of the two sets of scenarios, that is
 
3102
        a scenario for every possible combination of a left scenario and a
 
3103
        right scenario.
 
3104
    """
 
3105
    return [
 
3106
        ('%s,%s' % (left_name, right_name),
 
3107
         dict(left_dict.items() + right_dict.items()))
 
3108
        for left_name, left_dict in scenarios_left
 
3109
        for right_name, right_dict in scenarios_right]
 
3110
 
 
3111
 
 
3112
 
 
3113
def adapt_modules(mods_list, adapter, loader, suite):
 
3114
    """Adapt the modules in mods_list using adapter and add to suite."""
 
3115
    tests = loader.loadTestsFromModuleNames(mods_list)
 
3116
    adapt_tests(tests, adapter, suite)
 
3117
 
 
3118
 
 
3119
def adapt_tests(tests_list, adapter, suite):
 
3120
    """Adapt the tests in tests_list using adapter and add to suite."""
 
3121
    for test in iter_suite_tests(tests_list):
 
3122
        suite.addTests(adapter.adapt(test))
 
3123
 
 
3124
 
 
3125
def _rmtree_temp_dir(dirname):
 
3126
    # If LANG=C we probably have created some bogus paths
 
3127
    # which rmtree(unicode) will fail to delete
 
3128
    # so make sure we are using rmtree(str) to delete everything
 
3129
    # except on win32, where rmtree(str) will fail
 
3130
    # since it doesn't have the property of byte-stream paths
 
3131
    # (they are either ascii or mbcs)
 
3132
    if sys.platform == 'win32':
 
3133
        # make sure we are using the unicode win32 api
 
3134
        dirname = unicode(dirname)
 
3135
    else:
 
3136
        dirname = dirname.encode(sys.getfilesystemencoding())
 
3137
    try:
 
3138
        osutils.rmtree(dirname)
 
3139
    except OSError, e:
 
3140
        if sys.platform == 'win32' and e.errno == errno.EACCES:
 
3141
            sys.stderr.write(('Permission denied: '
 
3142
                                 'unable to remove testing dir '
 
3143
                                 '%s\n' % os.path.basename(dirname)))
 
3144
        else:
 
3145
            raise
 
3146
 
 
3147
 
 
3148
class Feature(object):
 
3149
    """An operating system Feature."""
 
3150
 
 
3151
    def __init__(self):
 
3152
        self._available = None
 
3153
 
 
3154
    def available(self):
 
3155
        """Is the feature available?
 
3156
 
 
3157
        :return: True if the feature is available.
 
3158
        """
 
3159
        if self._available is None:
 
3160
            self._available = self._probe()
 
3161
        return self._available
 
3162
 
 
3163
    def _probe(self):
 
3164
        """Implement this method in concrete features.
 
3165
 
 
3166
        :return: True if the feature is available.
 
3167
        """
 
3168
        raise NotImplementedError
 
3169
 
 
3170
    def __str__(self):
 
3171
        if getattr(self, 'feature_name', None):
 
3172
            return self.feature_name()
 
3173
        return self.__class__.__name__
 
3174
 
 
3175
 
 
3176
class _SymlinkFeature(Feature):
 
3177
 
 
3178
    def _probe(self):
 
3179
        return osutils.has_symlinks()
 
3180
 
 
3181
    def feature_name(self):
 
3182
        return 'symlinks'
 
3183
 
 
3184
SymlinkFeature = _SymlinkFeature()
 
3185
 
 
3186
 
 
3187
class _HardlinkFeature(Feature):
 
3188
 
 
3189
    def _probe(self):
 
3190
        return osutils.has_hardlinks()
 
3191
 
 
3192
    def feature_name(self):
 
3193
        return 'hardlinks'
 
3194
 
 
3195
HardlinkFeature = _HardlinkFeature()
 
3196
 
 
3197
 
 
3198
class _OsFifoFeature(Feature):
 
3199
 
 
3200
    def _probe(self):
 
3201
        return getattr(os, 'mkfifo', None)
 
3202
 
 
3203
    def feature_name(self):
 
3204
        return 'filesystem fifos'
 
3205
 
 
3206
OsFifoFeature = _OsFifoFeature()
 
3207
 
 
3208
 
 
3209
class _UnicodeFilenameFeature(Feature):
 
3210
    """Does the filesystem support Unicode filenames?"""
 
3211
 
 
3212
    def _probe(self):
 
3213
        try:
 
3214
            # Check for character combinations unlikely to be covered by any
 
3215
            # single non-unicode encoding. We use the characters
 
3216
            # - greek small letter alpha (U+03B1) and
 
3217
            # - braille pattern dots-123456 (U+283F).
 
3218
            os.stat(u'\u03b1\u283f')
 
3219
        except UnicodeEncodeError:
 
3220
            return False
 
3221
        except (IOError, OSError):
 
3222
            # The filesystem allows the Unicode filename but the file doesn't
 
3223
            # exist.
 
3224
            return True
 
3225
        else:
 
3226
            # The filesystem allows the Unicode filename and the file exists,
 
3227
            # for some reason.
 
3228
            return True
 
3229
 
 
3230
UnicodeFilenameFeature = _UnicodeFilenameFeature()
 
3231
 
 
3232
 
 
3233
class TestScenarioApplier(object):
 
3234
    """A tool to apply scenarios to tests."""
 
3235
 
 
3236
    def adapt(self, test):
 
3237
        """Return a TestSuite containing a copy of test for each scenario."""
 
3238
        result = unittest.TestSuite()
 
3239
        for scenario in self.scenarios:
 
3240
            result.addTest(self.adapt_test_to_scenario(test, scenario))
 
3241
        return result
 
3242
 
 
3243
    def adapt_test_to_scenario(self, test, scenario):
 
3244
        """Copy test and apply scenario to it.
 
3245
 
 
3246
        :param test: A test to adapt.
 
3247
        :param scenario: A tuple describing the scenarion.
 
3248
            The first element of the tuple is the new test id.
 
3249
            The second element is a dict containing attributes to set on the
 
3250
            test.
 
3251
        :return: The adapted test.
 
3252
        """
 
3253
        from copy import deepcopy
 
3254
        new_test = deepcopy(test)
 
3255
        for name, value in scenario[1].items():
 
3256
            setattr(new_test, name, value)
 
3257
        new_id = "%s(%s)" % (new_test.id(), scenario[0])
 
3258
        new_test.id = lambda: new_id
 
3259
        return new_test
 
3260
 
 
3261
 
 
3262
def probe_unicode_in_user_encoding():
 
3263
    """Try to encode several unicode strings to use in unicode-aware tests.
 
3264
    Return first successfull match.
 
3265
 
 
3266
    :return:  (unicode value, encoded plain string value) or (None, None)
 
3267
    """
 
3268
    possible_vals = [u'm\xb5', u'\xe1', u'\u0410']
 
3269
    for uni_val in possible_vals:
 
3270
        try:
 
3271
            str_val = uni_val.encode(osutils.get_user_encoding())
 
3272
        except UnicodeEncodeError:
 
3273
            # Try a different character
 
3274
            pass
 
3275
        else:
 
3276
            return uni_val, str_val
 
3277
    return None, None
 
3278
 
 
3279
 
 
3280
def probe_bad_non_ascii(encoding):
 
3281
    """Try to find [bad] character with code [128..255]
 
3282
    that cannot be decoded to unicode in some encoding.
 
3283
    Return None if all non-ascii characters is valid
 
3284
    for given encoding.
 
3285
    """
 
3286
    for i in xrange(128, 256):
 
3287
        char = chr(i)
 
3288
        try:
 
3289
            char.decode(encoding)
 
3290
        except UnicodeDecodeError:
 
3291
            return char
 
3292
    return None
 
3293
 
 
3294
 
 
3295
class _FTPServerFeature(Feature):
 
3296
    """Some tests want an FTP Server, check if one is available.
 
3297
 
 
3298
    Right now, the only way this is available is if 'medusa' is installed.
 
3299
    http://www.amk.ca/python/code/medusa.html
 
3300
    """
 
3301
 
 
3302
    def _probe(self):
 
3303
        try:
 
3304
            import bzrlib.tests.ftp_server
 
3305
            return True
 
3306
        except ImportError:
 
3307
            return False
 
3308
 
 
3309
    def feature_name(self):
 
3310
        return 'FTPServer'
 
3311
 
 
3312
 
 
3313
FTPServerFeature = _FTPServerFeature()
 
3314
 
 
3315
 
 
3316
class _HTTPSServerFeature(Feature):
 
3317
    """Some tests want an https Server, check if one is available.
 
3318
 
 
3319
    Right now, the only way this is available is under python2.6 which provides
 
3320
    an ssl module.
 
3321
    """
 
3322
 
 
3323
    def _probe(self):
 
3324
        try:
 
3325
            import ssl
 
3326
            return True
 
3327
        except ImportError:
 
3328
            return False
 
3329
 
 
3330
    def feature_name(self):
 
3331
        return 'HTTPSServer'
 
3332
 
 
3333
 
 
3334
HTTPSServerFeature = _HTTPSServerFeature()
 
3335
 
 
3336
 
 
3337
class _UnicodeFilename(Feature):
 
3338
    """Does the filesystem support Unicode filenames?"""
 
3339
 
 
3340
    def _probe(self):
 
3341
        try:
 
3342
            os.stat(u'\u03b1')
 
3343
        except UnicodeEncodeError:
 
3344
            return False
 
3345
        except (IOError, OSError):
 
3346
            # The filesystem allows the Unicode filename but the file doesn't
 
3347
            # exist.
 
3348
            return True
 
3349
        else:
 
3350
            # The filesystem allows the Unicode filename and the file exists,
 
3351
            # for some reason.
 
3352
            return True
 
3353
 
 
3354
UnicodeFilename = _UnicodeFilename()
 
3355
 
 
3356
 
 
3357
class _UTF8Filesystem(Feature):
 
3358
    """Is the filesystem UTF-8?"""
 
3359
 
 
3360
    def _probe(self):
 
3361
        if osutils._fs_enc.upper() in ('UTF-8', 'UTF8'):
 
3362
            return True
 
3363
        return False
 
3364
 
 
3365
UTF8Filesystem = _UTF8Filesystem()
 
3366
 
 
3367
 
 
3368
class _CaseInsCasePresFilenameFeature(Feature):
 
3369
    """Is the file-system case insensitive, but case-preserving?"""
 
3370
 
 
3371
    def _probe(self):
 
3372
        fileno, name = tempfile.mkstemp(prefix='MixedCase')
 
3373
        try:
 
3374
            # first check truly case-preserving for created files, then check
 
3375
            # case insensitive when opening existing files.
 
3376
            name = osutils.normpath(name)
 
3377
            base, rel = osutils.split(name)
 
3378
            found_rel = osutils.canonical_relpath(base, name)
 
3379
            return (found_rel == rel
 
3380
                    and os.path.isfile(name.upper())
 
3381
                    and os.path.isfile(name.lower()))
 
3382
        finally:
 
3383
            os.close(fileno)
 
3384
            os.remove(name)
 
3385
 
 
3386
    def feature_name(self):
 
3387
        return "case-insensitive case-preserving filesystem"
 
3388
 
 
3389
CaseInsCasePresFilenameFeature = _CaseInsCasePresFilenameFeature()
 
3390
 
 
3391
 
 
3392
class _CaseInsensitiveFilesystemFeature(Feature):
 
3393
    """Check if underlying filesystem is case-insensitive but *not* case
 
3394
    preserving.
 
3395
    """
 
3396
    # Note that on Windows, Cygwin, MacOS etc, the file-systems are far
 
3397
    # more likely to be case preserving, so this case is rare.
 
3398
 
 
3399
    def _probe(self):
 
3400
        if CaseInsCasePresFilenameFeature.available():
 
3401
            return False
 
3402
 
 
3403
        if TestCaseWithMemoryTransport.TEST_ROOT is None:
 
3404
            root = osutils.mkdtemp(prefix='testbzr-', suffix='.tmp')
 
3405
            TestCaseWithMemoryTransport.TEST_ROOT = root
 
3406
        else:
 
3407
            root = TestCaseWithMemoryTransport.TEST_ROOT
 
3408
        tdir = osutils.mkdtemp(prefix='case-sensitive-probe-', suffix='',
 
3409
            dir=root)
 
3410
        name_a = osutils.pathjoin(tdir, 'a')
 
3411
        name_A = osutils.pathjoin(tdir, 'A')
 
3412
        os.mkdir(name_a)
 
3413
        result = osutils.isdir(name_A)
 
3414
        _rmtree_temp_dir(tdir)
 
3415
        return result
 
3416
 
 
3417
    def feature_name(self):
 
3418
        return 'case-insensitive filesystem'
 
3419
 
 
3420
CaseInsensitiveFilesystemFeature = _CaseInsensitiveFilesystemFeature()