~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/__init__.py

  • Committer: John Arbash Meinel
  • Author(s): Mark Hammond
  • Date: 2008-09-09 17:02:21 UTC
  • mto: This revision was merged to the branch mainline in revision 3697.
  • Revision ID: john@arbash-meinel.com-20080909170221-svim3jw2mrz0amp3
An updated transparent icon for bzr.

Show diffs side-by-side

added added

removed removed

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