/brz/remove-bazaar

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

« back to all changes in this revision

Viewing changes to bzrlib/tests/__init__.py

  • Committer: Robert Collins
  • Date: 2010-05-11 08:44:59 UTC
  • mfrom: (5221 +trunk)
  • mto: This revision was merged to the branch mainline in revision 5223.
  • Revision ID: robertc@robertcollins.net-20100511084459-pb0uinna9zs3wu59
Merge trunk - resolve conflicts.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/__init__.py
1
 
# Copyright (C) 2005, 2006 by Canonical Ltd
 
1
# Copyright (C) 2005-2010 Canonical Ltd
2
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
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
 
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
16
16
 
 
17
"""Testing framework extensions"""
17
18
 
18
19
# TODO: Perhaps there should be an API to find out if bzr running under the
19
20
# test suite -- some plugins might want to avoid making intrusive changes if
26
27
# general style of bzrlib.  Please continue that consistency when adding e.g.
27
28
# new assertFoo() methods.
28
29
 
 
30
import atexit
29
31
import codecs
 
32
from copy import copy
30
33
from cStringIO import StringIO
31
34
import difflib
32
35
import doctest
33
36
import errno
34
37
import logging
 
38
import math
35
39
import os
 
40
from pprint import pformat
 
41
import random
36
42
import re
37
43
import shlex
38
44
import stat
39
 
from subprocess import Popen, PIPE
 
45
from subprocess import Popen, PIPE, STDOUT
40
46
import sys
41
47
import tempfile
 
48
import threading
 
49
import time
 
50
import traceback
42
51
import unittest
43
 
import time
44
 
 
45
 
 
 
52
import warnings
 
53
 
 
54
import testtools
 
55
# nb: check this before importing anything else from within it
 
56
_testtools_version = getattr(testtools, '__version__', ())
 
57
if _testtools_version < (0, 9, 2):
 
58
    raise ImportError("need at least testtools 0.9.2: %s is %r"
 
59
        % (testtools.__file__, _testtools_version))
 
60
from testtools import content
 
61
 
 
62
from bzrlib import (
 
63
    branchbuilder,
 
64
    bzrdir,
 
65
    chk_map,
 
66
    config,
 
67
    debug,
 
68
    errors,
 
69
    hooks,
 
70
    lock as _mod_lock,
 
71
    memorytree,
 
72
    osutils,
 
73
    progress,
 
74
    ui,
 
75
    urlutils,
 
76
    registry,
 
77
    workingtree,
 
78
    )
46
79
import bzrlib.branch
47
 
import bzrlib.bzrdir as bzrdir
48
80
import bzrlib.commands
49
 
import bzrlib.bundle.serializer
50
 
import bzrlib.errors as errors
 
81
import bzrlib.timestamp
 
82
import bzrlib.export
51
83
import bzrlib.inventory
52
84
import bzrlib.iterablefile
53
85
import bzrlib.lockdir
58
90
    pass
59
91
from bzrlib.merge import merge_inner
60
92
import bzrlib.merge3
61
 
import bzrlib.osutils
62
 
import bzrlib.osutils as osutils
63
93
import bzrlib.plugin
64
 
import bzrlib.progress as progress
65
 
from bzrlib.revision import common_ancestor
 
94
from bzrlib.smart import client, request, server
66
95
import bzrlib.store
67
96
from bzrlib import symbol_versioning
 
97
from bzrlib.symbol_versioning import (
 
98
    DEPRECATED_PARAMETER,
 
99
    deprecated_function,
 
100
    deprecated_in,
 
101
    deprecated_method,
 
102
    deprecated_passed,
 
103
    )
68
104
import bzrlib.trace
69
 
from bzrlib.transport import get_transport
 
105
from bzrlib.transport import (
 
106
    get_transport,
 
107
    memory,
 
108
    pathfilter,
 
109
    )
70
110
import bzrlib.transport
71
 
from bzrlib.transport.local import LocalRelpathServer
72
 
from bzrlib.transport.readonly import ReadonlyServer
73
 
from bzrlib.trace import mutter
74
 
from bzrlib.tests import TestUtil
 
111
from bzrlib.trace import mutter, note
 
112
from bzrlib.tests import (
 
113
    test_server,
 
114
    TestUtil,
 
115
    treeshape,
 
116
    )
 
117
from bzrlib.tests.http_server import HttpServer
75
118
from bzrlib.tests.TestUtil import (
76
119
                          TestSuite,
77
120
                          TestLoader,
78
121
                          )
79
 
from bzrlib.tests.treeshape import build_tree_contents
80
 
import bzrlib.urlutils as urlutils
 
122
from bzrlib.ui import NullProgressView
 
123
from bzrlib.ui.text import TextUIFactory
 
124
import bzrlib.version_info_formats.format_custom
81
125
from bzrlib.workingtree import WorkingTree, WorkingTreeFormat2
82
126
 
83
 
default_transport = LocalRelpathServer
84
 
 
85
 
MODULES_TO_TEST = []
86
 
MODULES_TO_DOCTEST = [
87
 
                      bzrlib.branch,
88
 
                      bzrlib.bundle.serializer,
89
 
                      bzrlib.commands,
90
 
                      bzrlib.errors,
91
 
                      bzrlib.inventory,
92
 
                      bzrlib.iterablefile,
93
 
                      bzrlib.lockdir,
94
 
                      bzrlib.merge3,
95
 
                      bzrlib.option,
96
 
                      bzrlib.osutils,
97
 
                      bzrlib.store
98
 
                      ]
99
 
 
100
 
 
101
 
def packages_to_test():
102
 
    """Return a list of packages to test.
103
 
 
104
 
    The packages are not globally imported so that import failures are
105
 
    triggered when running selftest, not when importing the command.
106
 
    """
107
 
    import bzrlib.doc
108
 
    import bzrlib.tests.blackbox
109
 
    import bzrlib.tests.branch_implementations
110
 
    import bzrlib.tests.bzrdir_implementations
111
 
    import bzrlib.tests.interrepository_implementations
112
 
    import bzrlib.tests.interversionedfile_implementations
113
 
    import bzrlib.tests.intertree_implementations
114
 
    import bzrlib.tests.repository_implementations
115
 
    import bzrlib.tests.revisionstore_implementations
116
 
    import bzrlib.tests.tree_implementations
117
 
    import bzrlib.tests.workingtree_implementations
118
 
    return [
119
 
            bzrlib.doc,
120
 
            bzrlib.tests.blackbox,
121
 
            bzrlib.tests.branch_implementations,
122
 
            bzrlib.tests.bzrdir_implementations,
123
 
            bzrlib.tests.interrepository_implementations,
124
 
            bzrlib.tests.interversionedfile_implementations,
125
 
            bzrlib.tests.intertree_implementations,
126
 
            bzrlib.tests.repository_implementations,
127
 
            bzrlib.tests.revisionstore_implementations,
128
 
            bzrlib.tests.tree_implementations,
129
 
            bzrlib.tests.workingtree_implementations,
130
 
            ]
131
 
 
132
 
 
133
 
class _MyResult(unittest._TextTestResult):
134
 
    """Custom TestResult.
135
 
 
136
 
    Shows output in a different format, including displaying runtime for tests.
137
 
    """
 
127
# Mark this python module as being part of the implementation
 
128
# of unittest: this gives us better tracebacks where the last
 
129
# shown frame is the test code, not our assertXYZ.
 
130
__unittest = 1
 
131
 
 
132
default_transport = test_server.LocalURLServer
 
133
 
 
134
 
 
135
_unitialized_attr = object()
 
136
"""A sentinel needed to act as a default value in a method signature."""
 
137
 
 
138
 
 
139
# Subunit result codes, defined here to prevent a hard dependency on subunit.
 
140
SUBUNIT_SEEK_SET = 0
 
141
SUBUNIT_SEEK_CUR = 1
 
142
 
 
143
 
 
144
class ExtendedTestResult(unittest._TextTestResult):
 
145
    """Accepts, reports and accumulates the results of running tests.
 
146
 
 
147
    Compared to the unittest version this class adds support for
 
148
    profiling, benchmarking, stopping as soon as a test fails,  and
 
149
    skipping tests.  There are further-specialized subclasses for
 
150
    different types of display.
 
151
 
 
152
    When a test finishes, in whatever way, it calls one of the addSuccess,
 
153
    addFailure or addError classes.  These in turn may redirect to a more
 
154
    specific case for the special test results supported by our extended
 
155
    tests.
 
156
 
 
157
    Note that just one of these objects is fed the results from many tests.
 
158
    """
 
159
 
138
160
    stop_early = False
139
 
    
140
 
    def __init__(self, stream, descriptions, verbosity, pb=None,
141
 
                 bench_history=None):
 
161
 
 
162
    def __init__(self, stream, descriptions, verbosity,
 
163
                 bench_history=None,
 
164
                 strict=False,
 
165
                 ):
142
166
        """Construct new TestResult.
143
167
 
144
168
        :param bench_history: Optionally, a writable file object to accumulate
145
169
            benchmark results.
146
170
        """
147
171
        unittest._TextTestResult.__init__(self, stream, descriptions, verbosity)
148
 
        self.pb = pb
149
172
        if bench_history is not None:
150
173
            from bzrlib.version import _get_bzr_source_tree
151
174
            src_tree = _get_bzr_source_tree()
161
184
                revision_id = ''
162
185
            bench_history.write("--date %s %s\n" % (time.time(), revision_id))
163
186
        self._bench_history = bench_history
164
 
    
165
 
    def extractBenchmarkTime(self, testCase):
 
187
        self.ui = ui.ui_factory
 
188
        self.num_tests = 0
 
189
        self.error_count = 0
 
190
        self.failure_count = 0
 
191
        self.known_failure_count = 0
 
192
        self.skip_count = 0
 
193
        self.not_applicable_count = 0
 
194
        self.unsupported = {}
 
195
        self.count = 0
 
196
        self._overall_start_time = time.time()
 
197
        self._strict = strict
 
198
 
 
199
    def stopTestRun(self):
 
200
        run = self.testsRun
 
201
        actionTaken = "Ran"
 
202
        stopTime = time.time()
 
203
        timeTaken = stopTime - self.startTime
 
204
        self.printErrors()
 
205
        self.stream.writeln(self.separator2)
 
206
        self.stream.writeln("%s %d test%s in %.3fs" % (actionTaken,
 
207
                            run, run != 1 and "s" or "", timeTaken))
 
208
        self.stream.writeln()
 
209
        if not self.wasSuccessful():
 
210
            self.stream.write("FAILED (")
 
211
            failed, errored = map(len, (self.failures, self.errors))
 
212
            if failed:
 
213
                self.stream.write("failures=%d" % failed)
 
214
            if errored:
 
215
                if failed: self.stream.write(", ")
 
216
                self.stream.write("errors=%d" % errored)
 
217
            if self.known_failure_count:
 
218
                if failed or errored: self.stream.write(", ")
 
219
                self.stream.write("known_failure_count=%d" %
 
220
                    self.known_failure_count)
 
221
            self.stream.writeln(")")
 
222
        else:
 
223
            if self.known_failure_count:
 
224
                self.stream.writeln("OK (known_failures=%d)" %
 
225
                    self.known_failure_count)
 
226
            else:
 
227
                self.stream.writeln("OK")
 
228
        if self.skip_count > 0:
 
229
            skipped = self.skip_count
 
230
            self.stream.writeln('%d test%s skipped' %
 
231
                                (skipped, skipped != 1 and "s" or ""))
 
232
        if self.unsupported:
 
233
            for feature, count in sorted(self.unsupported.items()):
 
234
                self.stream.writeln("Missing feature '%s' skipped %d tests." %
 
235
                    (feature, count))
 
236
        if self._strict:
 
237
            ok = self.wasStrictlySuccessful()
 
238
        else:
 
239
            ok = self.wasSuccessful()
 
240
        if TestCase._first_thread_leaker_id:
 
241
            self.stream.write(
 
242
                '%s is leaking threads among %d leaking tests.\n' % (
 
243
                TestCase._first_thread_leaker_id,
 
244
                TestCase._leaking_threads_tests))
 
245
            # We don't report the main thread as an active one.
 
246
            self.stream.write(
 
247
                '%d non-main threads were left active in the end.\n'
 
248
                % (TestCase._active_threads - 1))
 
249
 
 
250
    def getDescription(self, test):
 
251
        return test.id()
 
252
 
 
253
    def _extractBenchmarkTime(self, testCase, details=None):
166
254
        """Add a benchmark time for the current test case."""
167
 
        self._benchmarkTime = getattr(testCase, "_benchtime", None)
168
 
    
 
255
        if details and 'benchtime' in details:
 
256
            return float(''.join(details['benchtime'].iter_bytes()))
 
257
        return getattr(testCase, "_benchtime", None)
 
258
 
169
259
    def _elapsedTestTimeString(self):
170
260
        """Return a time string for the overall time the current test has taken."""
171
261
        return self._formatTime(time.time() - self._start_time)
172
262
 
173
 
    def _testTimeString(self):
174
 
        if self._benchmarkTime is not None:
175
 
            return "%s/%s" % (
176
 
                self._formatTime(self._benchmarkTime),
177
 
                self._elapsedTestTimeString())
 
263
    def _testTimeString(self, testCase):
 
264
        benchmark_time = self._extractBenchmarkTime(testCase)
 
265
        if benchmark_time is not None:
 
266
            return self._formatTime(benchmark_time) + "*"
178
267
        else:
179
 
            return "      %s" % self._elapsedTestTimeString()
 
268
            return self._elapsedTestTimeString()
180
269
 
181
270
    def _formatTime(self, seconds):
182
271
        """Format seconds as milliseconds with leading spaces."""
183
 
        return "%5dms" % (1000 * seconds)
 
272
        # some benchmarks can take thousands of seconds to run, so we need 8
 
273
        # places
 
274
        return "%8dms" % (1000 * seconds)
184
275
 
185
 
    def _ellipsise_unimportant_words(self, a_string, final_width,
186
 
                                   keep_start=False):
187
 
        """Add ellipses (sp?) for overly long strings.
188
 
        
189
 
        :param keep_start: If true preserve the start of a_string rather
190
 
                           than the end of it.
191
 
        """
192
 
        if keep_start:
193
 
            if len(a_string) > final_width:
194
 
                result = a_string[:final_width-3] + '...'
195
 
            else:
196
 
                result = a_string
197
 
        else:
198
 
            if len(a_string) > final_width:
199
 
                result = '...' + a_string[3-final_width:]
200
 
            else:
201
 
                result = a_string
202
 
        return result.ljust(final_width)
 
276
    def _shortened_test_description(self, test):
 
277
        what = test.id()
 
278
        what = re.sub(r'^bzrlib\.(tests|benchmarks)\.', '', what)
 
279
        return what
203
280
 
204
281
    def startTest(self, test):
205
282
        unittest.TestResult.startTest(self, test)
206
 
        # In a short description, the important words are in
207
 
        # the beginning, but in an id, the important words are
208
 
        # at the end
209
 
        SHOW_DESCRIPTIONS = False
210
 
 
211
 
        if not self.showAll and self.dots and self.pb is not None:
212
 
            final_width = 13
213
 
        else:
214
 
            final_width = osutils.terminal_width()
215
 
            final_width = final_width - 15 - 8
216
 
        what = None
217
 
        if SHOW_DESCRIPTIONS:
218
 
            what = test.shortDescription()
219
 
            if what:
220
 
                what = self._ellipsise_unimportant_words(what, final_width, keep_start=True)
221
 
        if what is None:
222
 
            what = test.id()
223
 
            if what.startswith('bzrlib.tests.'):
224
 
                what = what[13:]
225
 
            what = self._ellipsise_unimportant_words(what, final_width)
226
 
        if self.showAll:
227
 
            self.stream.write(what)
228
 
        elif self.dots and self.pb is not None:
229
 
            self.pb.update(what, self.testsRun - 1, None)
230
 
        self.stream.flush()
 
283
        if self.count == 0:
 
284
            self.startTests()
 
285
        self.report_test_start(test)
 
286
        test.number = self.count
231
287
        self._recordTestStartTime()
232
288
 
 
289
    def startTests(self):
 
290
        import platform
 
291
        if getattr(sys, 'frozen', None) is None:
 
292
            bzr_path = osutils.realpath(sys.argv[0])
 
293
        else:
 
294
            bzr_path = sys.executable
 
295
        self.stream.write(
 
296
            'bzr selftest: %s\n' % (bzr_path,))
 
297
        self.stream.write(
 
298
            '   %s\n' % (
 
299
                    bzrlib.__path__[0],))
 
300
        self.stream.write(
 
301
            '   bzr-%s python-%s %s\n' % (
 
302
                    bzrlib.version_string,
 
303
                    bzrlib._format_version_tuple(sys.version_info),
 
304
                    platform.platform(aliased=1),
 
305
                    ))
 
306
        self.stream.write('\n')
 
307
 
233
308
    def _recordTestStartTime(self):
234
309
        """Record that a test has started."""
235
310
        self._start_time = time.time()
236
311
 
 
312
    def _cleanupLogFile(self, test):
 
313
        # We can only do this if we have one of our TestCases, not if
 
314
        # we have a doctest.
 
315
        setKeepLogfile = getattr(test, 'setKeepLogfile', None)
 
316
        if setKeepLogfile is not None:
 
317
            setKeepLogfile()
 
318
 
237
319
    def addError(self, test, err):
238
 
        if isinstance(err[1], TestSkipped):
239
 
            return self.addSkipped(test, err)    
 
320
        """Tell result that test finished with an error.
 
321
 
 
322
        Called from the TestCase run() method when the test
 
323
        fails with an unexpected error.
 
324
        """
 
325
        self._post_mortem()
240
326
        unittest.TestResult.addError(self, test, err)
241
 
        self.extractBenchmarkTime(test)
242
 
        if self.showAll:
243
 
            self.stream.writeln("ERROR %s" % self._testTimeString())
244
 
        elif self.dots and self.pb is None:
245
 
            self.stream.write('E')
246
 
        elif self.dots:
247
 
            self.pb.update(self._ellipsise_unimportant_words('ERROR', 13), self.testsRun, None)
248
 
            self.pb.note(self._ellipsise_unimportant_words(
249
 
                            test.id() + ': ERROR',
250
 
                            osutils.terminal_width()))
251
 
        self.stream.flush()
 
327
        self.error_count += 1
 
328
        self.report_error(test, err)
252
329
        if self.stop_early:
253
330
            self.stop()
 
331
        self._cleanupLogFile(test)
254
332
 
255
333
    def addFailure(self, test, err):
 
334
        """Tell result that test failed.
 
335
 
 
336
        Called from the TestCase run() method when the test
 
337
        fails because e.g. an assert() method failed.
 
338
        """
 
339
        self._post_mortem()
256
340
        unittest.TestResult.addFailure(self, test, err)
257
 
        self.extractBenchmarkTime(test)
258
 
        if self.showAll:
259
 
            self.stream.writeln(" FAIL %s" % self._testTimeString())
260
 
        elif self.dots and self.pb is None:
261
 
            self.stream.write('F')
262
 
        elif self.dots:
263
 
            self.pb.update(self._ellipsise_unimportant_words('FAIL', 13), self.testsRun, None)
264
 
            self.pb.note(self._ellipsise_unimportant_words(
265
 
                            test.id() + ': FAIL',
266
 
                            osutils.terminal_width()))
267
 
        self.stream.flush()
 
341
        self.failure_count += 1
 
342
        self.report_failure(test, err)
268
343
        if self.stop_early:
269
344
            self.stop()
270
 
 
271
 
    def addSuccess(self, test):
272
 
        self.extractBenchmarkTime(test)
 
345
        self._cleanupLogFile(test)
 
346
 
 
347
    def addSuccess(self, test, details=None):
 
348
        """Tell result that test completed successfully.
 
349
 
 
350
        Called from the TestCase run()
 
351
        """
273
352
        if self._bench_history is not None:
274
 
            if self._benchmarkTime is not None:
 
353
            benchmark_time = self._extractBenchmarkTime(test, details)
 
354
            if benchmark_time is not None:
275
355
                self._bench_history.write("%s %s\n" % (
276
 
                    self._formatTime(self._benchmarkTime),
 
356
                    self._formatTime(benchmark_time),
277
357
                    test.id()))
278
 
        if self.showAll:
279
 
            self.stream.writeln('   OK %s' % self._testTimeString())
280
 
            for bench_called, stats in getattr(test, '_benchcalls', []):
281
 
                self.stream.writeln('LSProf output for %s(%s, %s)' % bench_called)
282
 
                stats.pprint(file=self.stream)
283
 
        elif self.dots and self.pb is None:
284
 
            self.stream.write('~')
285
 
        elif self.dots:
286
 
            self.pb.update(self._ellipsise_unimportant_words('OK', 13), self.testsRun, None)
287
 
        self.stream.flush()
 
358
        self.report_success(test)
 
359
        self._cleanupLogFile(test)
288
360
        unittest.TestResult.addSuccess(self, test)
289
 
 
290
 
    def addSkipped(self, test, skip_excinfo):
291
 
        self.extractBenchmarkTime(test)
292
 
        if self.showAll:
293
 
            print >>self.stream, ' SKIP %s' % self._testTimeString()
294
 
            print >>self.stream, '     %s' % skip_excinfo[1]
295
 
        elif self.dots and self.pb is None:
296
 
            self.stream.write('S')
297
 
        elif self.dots:
298
 
            self.pb.update(self._ellipsise_unimportant_words('SKIP', 13), self.testsRun, None)
299
 
        self.stream.flush()
300
 
        # seems best to treat this as success from point-of-view of unittest
301
 
        # -- it actually does nothing so it barely matters :)
302
 
        try:
303
 
            test.tearDown()
304
 
        except KeyboardInterrupt:
305
 
            raise
306
 
        except:
307
 
            self.addError(test, test.__exc_info())
308
 
        else:
309
 
            unittest.TestResult.addSuccess(self, test)
310
 
 
311
 
    def printErrorList(self, flavour, errors):
312
 
        for test, err in errors:
313
 
            self.stream.writeln(self.separator1)
314
 
            self.stream.writeln("%s: %s" % (flavour, self.getDescription(test)))
315
 
            if getattr(test, '_get_log', None) is not None:
316
 
                print >>self.stream
317
 
                print >>self.stream, \
318
 
                        ('vvvv[log from %s]' % test.id()).ljust(78,'-')
319
 
                print >>self.stream, test._get_log()
320
 
                print >>self.stream, \
321
 
                        ('^^^^[log from %s]' % test.id()).ljust(78,'-')
322
 
            self.stream.writeln(self.separator2)
323
 
            self.stream.writeln("%s" % err)
 
361
        test._log_contents = ''
 
362
 
 
363
    def addExpectedFailure(self, test, err):
 
364
        self.known_failure_count += 1
 
365
        self.report_known_failure(test, err)
 
366
 
 
367
    def addNotSupported(self, test, feature):
 
368
        """The test will not be run because of a missing feature.
 
369
        """
 
370
        # this can be called in two different ways: it may be that the
 
371
        # test started running, and then raised (through requireFeature)
 
372
        # UnavailableFeature.  Alternatively this method can be called
 
373
        # while probing for features before running the test code proper; in
 
374
        # that case we will see startTest and stopTest, but the test will
 
375
        # never actually run.
 
376
        self.unsupported.setdefault(str(feature), 0)
 
377
        self.unsupported[str(feature)] += 1
 
378
        self.report_unsupported(test, feature)
 
379
 
 
380
    def addSkip(self, test, reason):
 
381
        """A test has not run for 'reason'."""
 
382
        self.skip_count += 1
 
383
        self.report_skip(test, reason)
 
384
 
 
385
    def addNotApplicable(self, test, reason):
 
386
        self.not_applicable_count += 1
 
387
        self.report_not_applicable(test, reason)
 
388
 
 
389
    def _post_mortem(self):
 
390
        """Start a PDB post mortem session."""
 
391
        if os.environ.get('BZR_TEST_PDB', None):
 
392
            import pdb;pdb.post_mortem()
 
393
 
 
394
    def progress(self, offset, whence):
 
395
        """The test is adjusting the count of tests to run."""
 
396
        if whence == SUBUNIT_SEEK_SET:
 
397
            self.num_tests = offset
 
398
        elif whence == SUBUNIT_SEEK_CUR:
 
399
            self.num_tests += offset
 
400
        else:
 
401
            raise errors.BzrError("Unknown whence %r" % whence)
 
402
 
 
403
    def report_cleaning_up(self):
 
404
        pass
 
405
 
 
406
    def startTestRun(self):
 
407
        self.startTime = time.time()
 
408
 
 
409
    def report_success(self, test):
 
410
        pass
 
411
 
 
412
    def wasStrictlySuccessful(self):
 
413
        if self.unsupported or self.known_failure_count:
 
414
            return False
 
415
        return self.wasSuccessful()
 
416
 
 
417
 
 
418
class TextTestResult(ExtendedTestResult):
 
419
    """Displays progress and results of tests in text form"""
 
420
 
 
421
    def __init__(self, stream, descriptions, verbosity,
 
422
                 bench_history=None,
 
423
                 pb=None,
 
424
                 strict=None,
 
425
                 ):
 
426
        ExtendedTestResult.__init__(self, stream, descriptions, verbosity,
 
427
            bench_history, strict)
 
428
        # We no longer pass them around, but just rely on the UIFactory stack
 
429
        # for state
 
430
        if pb is not None:
 
431
            warnings.warn("Passing pb to TextTestResult is deprecated")
 
432
        self.pb = self.ui.nested_progress_bar()
 
433
        self.pb.show_pct = False
 
434
        self.pb.show_spinner = False
 
435
        self.pb.show_eta = False,
 
436
        self.pb.show_count = False
 
437
        self.pb.show_bar = False
 
438
        self.pb.update_latency = 0
 
439
        self.pb.show_transport_activity = False
 
440
 
 
441
    def stopTestRun(self):
 
442
        # called when the tests that are going to run have run
 
443
        self.pb.clear()
 
444
        self.pb.finished()
 
445
        super(TextTestResult, self).stopTestRun()
 
446
 
 
447
    def startTestRun(self):
 
448
        super(TextTestResult, self).startTestRun()
 
449
        self.pb.update('[test 0/%d] Starting' % (self.num_tests))
 
450
 
 
451
    def printErrors(self):
 
452
        # clear the pb to make room for the error listing
 
453
        self.pb.clear()
 
454
        super(TextTestResult, self).printErrors()
 
455
 
 
456
    def _progress_prefix_text(self):
 
457
        # the longer this text, the less space we have to show the test
 
458
        # name...
 
459
        a = '[%d' % self.count              # total that have been run
 
460
        # tests skipped as known not to be relevant are not important enough
 
461
        # to show here
 
462
        ## if self.skip_count:
 
463
        ##     a += ', %d skip' % self.skip_count
 
464
        ## if self.known_failure_count:
 
465
        ##     a += '+%dX' % self.known_failure_count
 
466
        if self.num_tests:
 
467
            a +='/%d' % self.num_tests
 
468
        a += ' in '
 
469
        runtime = time.time() - self._overall_start_time
 
470
        if runtime >= 60:
 
471
            a += '%dm%ds' % (runtime / 60, runtime % 60)
 
472
        else:
 
473
            a += '%ds' % runtime
 
474
        total_fail_count = self.error_count + self.failure_count
 
475
        if total_fail_count:
 
476
            a += ', %d failed' % total_fail_count
 
477
        # if self.unsupported:
 
478
        #     a += ', %d missing' % len(self.unsupported)
 
479
        a += ']'
 
480
        return a
 
481
 
 
482
    def report_test_start(self, test):
 
483
        self.count += 1
 
484
        self.pb.update(
 
485
                self._progress_prefix_text()
 
486
                + ' '
 
487
                + self._shortened_test_description(test))
 
488
 
 
489
    def _test_description(self, test):
 
490
        return self._shortened_test_description(test)
 
491
 
 
492
    def report_error(self, test, err):
 
493
        self.stream.write('ERROR: %s\n    %s\n' % (
 
494
            self._test_description(test),
 
495
            err[1],
 
496
            ))
 
497
 
 
498
    def report_failure(self, test, err):
 
499
        self.stream.write('FAIL: %s\n    %s\n' % (
 
500
            self._test_description(test),
 
501
            err[1],
 
502
            ))
 
503
 
 
504
    def report_known_failure(self, test, err):
 
505
        pass
 
506
 
 
507
    def report_skip(self, test, reason):
 
508
        pass
 
509
 
 
510
    def report_not_applicable(self, test, reason):
 
511
        pass
 
512
 
 
513
    def report_unsupported(self, test, feature):
 
514
        """test cannot be run because feature is missing."""
 
515
 
 
516
    def report_cleaning_up(self):
 
517
        self.pb.update('Cleaning up')
 
518
 
 
519
 
 
520
class VerboseTestResult(ExtendedTestResult):
 
521
    """Produce long output, with one line per test run plus times"""
 
522
 
 
523
    def _ellipsize_to_right(self, a_string, final_width):
 
524
        """Truncate and pad a string, keeping the right hand side"""
 
525
        if len(a_string) > final_width:
 
526
            result = '...' + a_string[3-final_width:]
 
527
        else:
 
528
            result = a_string
 
529
        return result.ljust(final_width)
 
530
 
 
531
    def startTestRun(self):
 
532
        super(VerboseTestResult, self).startTestRun()
 
533
        self.stream.write('running %d tests...\n' % self.num_tests)
 
534
 
 
535
    def report_test_start(self, test):
 
536
        self.count += 1
 
537
        name = self._shortened_test_description(test)
 
538
        width = osutils.terminal_width()
 
539
        if width is not None:
 
540
            # width needs space for 6 char status, plus 1 for slash, plus an
 
541
            # 11-char time string, plus a trailing blank
 
542
            # when NUMBERED_DIRS: plus 5 chars on test number, plus 1 char on
 
543
            # space
 
544
            self.stream.write(self._ellipsize_to_right(name, width-18))
 
545
        else:
 
546
            self.stream.write(name)
 
547
        self.stream.flush()
 
548
 
 
549
    def _error_summary(self, err):
 
550
        indent = ' ' * 4
 
551
        return '%s%s' % (indent, err[1])
 
552
 
 
553
    def report_error(self, test, err):
 
554
        self.stream.writeln('ERROR %s\n%s'
 
555
                % (self._testTimeString(test),
 
556
                   self._error_summary(err)))
 
557
 
 
558
    def report_failure(self, test, err):
 
559
        self.stream.writeln(' FAIL %s\n%s'
 
560
                % (self._testTimeString(test),
 
561
                   self._error_summary(err)))
 
562
 
 
563
    def report_known_failure(self, test, err):
 
564
        self.stream.writeln('XFAIL %s\n%s'
 
565
                % (self._testTimeString(test),
 
566
                   self._error_summary(err)))
 
567
 
 
568
    def report_success(self, test):
 
569
        self.stream.writeln('   OK %s' % self._testTimeString(test))
 
570
        for bench_called, stats in getattr(test, '_benchcalls', []):
 
571
            self.stream.writeln('LSProf output for %s(%s, %s)' % bench_called)
 
572
            stats.pprint(file=self.stream)
 
573
        # flush the stream so that we get smooth output. This verbose mode is
 
574
        # used to show the output in PQM.
 
575
        self.stream.flush()
 
576
 
 
577
    def report_skip(self, test, reason):
 
578
        self.stream.writeln(' SKIP %s\n%s'
 
579
                % (self._testTimeString(test), reason))
 
580
 
 
581
    def report_not_applicable(self, test, reason):
 
582
        self.stream.writeln('  N/A %s\n    %s'
 
583
                % (self._testTimeString(test), reason))
 
584
 
 
585
    def report_unsupported(self, test, feature):
 
586
        """test cannot be run because feature is missing."""
 
587
        self.stream.writeln("NODEP %s\n    The feature '%s' is not available."
 
588
                %(self._testTimeString(test), feature))
324
589
 
325
590
 
326
591
class TextTestRunner(object):
330
595
                 stream=sys.stderr,
331
596
                 descriptions=0,
332
597
                 verbosity=1,
333
 
                 keep_output=False,
334
 
                 pb=None,
335
 
                 bench_history=None):
 
598
                 bench_history=None,
 
599
                 strict=False,
 
600
                 result_decorators=None,
 
601
                 ):
 
602
        """Create a TextTestRunner.
 
603
 
 
604
        :param result_decorators: An optional list of decorators to apply
 
605
            to the result object being used by the runner. Decorators are
 
606
            applied left to right - the first element in the list is the 
 
607
            innermost decorator.
 
608
        """
 
609
        # stream may know claim to know to write unicode strings, but in older
 
610
        # pythons this goes sufficiently wrong that it is a bad idea. (
 
611
        # specifically a built in file with encoding 'UTF-8' will still try
 
612
        # to encode using ascii.
 
613
        new_encoding = osutils.get_terminal_encoding()
 
614
        codec = codecs.lookup(new_encoding)
 
615
        if type(codec) is tuple:
 
616
            # Python 2.4
 
617
            encode = codec[0]
 
618
        else:
 
619
            encode = codec.encode
 
620
        stream = osutils.UnicodeOrBytesToBytesWriter(encode, stream)
 
621
        stream.encoding = new_encoding
336
622
        self.stream = unittest._WritelnDecorator(stream)
337
623
        self.descriptions = descriptions
338
624
        self.verbosity = verbosity
339
 
        self.keep_output = keep_output
340
 
        self.pb = pb
341
625
        self._bench_history = bench_history
342
 
 
343
 
    def _makeResult(self):
344
 
        result = _MyResult(self.stream,
345
 
                           self.descriptions,
346
 
                           self.verbosity,
347
 
                           pb=self.pb,
348
 
                           bench_history=self._bench_history)
349
 
        result.stop_early = self.stop_on_failure
350
 
        return result
 
626
        self._strict = strict
 
627
        self._result_decorators = result_decorators or []
351
628
 
352
629
    def run(self, test):
353
630
        "Run the given test case or test suite."
354
 
        result = self._makeResult()
355
 
        startTime = time.time()
356
 
        if self.pb is not None:
357
 
            self.pb.update('Running tests', 0, test.countTestCases())
358
 
        test.run(result)
359
 
        stopTime = time.time()
360
 
        timeTaken = stopTime - startTime
361
 
        result.printErrors()
362
 
        self.stream.writeln(result.separator2)
363
 
        run = result.testsRun
364
 
        self.stream.writeln("Ran %d test%s in %.3fs" %
365
 
                            (run, run != 1 and "s" or "", timeTaken))
366
 
        self.stream.writeln()
367
 
        if not result.wasSuccessful():
368
 
            self.stream.write("FAILED (")
369
 
            failed, errored = map(len, (result.failures, result.errors))
370
 
            if failed:
371
 
                self.stream.write("failures=%d" % failed)
372
 
            if errored:
373
 
                if failed: self.stream.write(", ")
374
 
                self.stream.write("errors=%d" % errored)
375
 
            self.stream.writeln(")")
376
 
        else:
377
 
            self.stream.writeln("OK")
378
 
        if self.pb is not None:
379
 
            self.pb.update('Cleaning up', 0, 1)
380
 
        # This is still a little bogus, 
381
 
        # but only a little. Folk not using our testrunner will
382
 
        # have to delete their temp directories themselves.
383
 
        test_root = TestCaseInTempDir.TEST_ROOT
384
 
        if result.wasSuccessful() or not self.keep_output:
385
 
            if test_root is not None:
386
 
                # If LANG=C we probably have created some bogus paths
387
 
                # which rmtree(unicode) will fail to delete
388
 
                # so make sure we are using rmtree(str) to delete everything
389
 
                # except on win32, where rmtree(str) will fail
390
 
                # since it doesn't have the property of byte-stream paths
391
 
                # (they are either ascii or mbcs)
392
 
                if sys.platform == 'win32':
393
 
                    # make sure we are using the unicode win32 api
394
 
                    test_root = unicode(test_root)
395
 
                else:
396
 
                    test_root = test_root.encode(
397
 
                        sys.getfilesystemencoding())
398
 
                osutils.rmtree(test_root)
399
 
        else:
400
 
            if self.pb is not None:
401
 
                self.pb.note("Failed tests working directories are in '%s'\n",
402
 
                             test_root)
403
 
            else:
404
 
                self.stream.writeln(
405
 
                    "Failed tests working directories are in '%s'\n" %
406
 
                    test_root)
407
 
        TestCaseInTempDir.TEST_ROOT = None
408
 
        if self.pb is not None:
409
 
            self.pb.clear()
410
 
        return result
 
631
        if self.verbosity == 1:
 
632
            result_class = TextTestResult
 
633
        elif self.verbosity >= 2:
 
634
            result_class = VerboseTestResult
 
635
        original_result = result_class(self.stream,
 
636
                              self.descriptions,
 
637
                              self.verbosity,
 
638
                              bench_history=self._bench_history,
 
639
                              strict=self._strict,
 
640
                              )
 
641
        # Signal to result objects that look at stop early policy to stop,
 
642
        original_result.stop_early = self.stop_on_failure
 
643
        result = original_result
 
644
        for decorator in self._result_decorators:
 
645
            result = decorator(result)
 
646
            result.stop_early = self.stop_on_failure
 
647
        result.startTestRun()
 
648
        try:
 
649
            test.run(result)
 
650
        finally:
 
651
            result.stopTestRun()
 
652
        # higher level code uses our extended protocol to determine
 
653
        # what exit code to give.
 
654
        return original_result
411
655
 
412
656
 
413
657
def iter_suite_tests(suite):
414
658
    """Return all tests in a suite, recursing through nested suites"""
415
 
    for item in suite._tests:
416
 
        if isinstance(item, unittest.TestCase):
417
 
            yield item
418
 
        elif isinstance(item, unittest.TestSuite):
 
659
    if isinstance(suite, unittest.TestCase):
 
660
        yield suite
 
661
    elif isinstance(suite, unittest.TestSuite):
 
662
        for item in suite:
419
663
            for r in iter_suite_tests(item):
420
664
                yield r
421
 
        else:
422
 
            raise Exception('unknown object %r inside test suite %r'
423
 
                            % (item, suite))
424
 
 
425
 
 
426
 
class TestSkipped(Exception):
427
 
    """Indicates that a test was intentionally skipped, rather than failing."""
428
 
 
429
 
 
430
 
class CommandFailed(Exception):
431
 
    pass
 
665
    else:
 
666
        raise Exception('unknown type %r for object %r'
 
667
                        % (type(suite), suite))
 
668
 
 
669
 
 
670
TestSkipped = testtools.testcase.TestSkipped
 
671
 
 
672
 
 
673
class TestNotApplicable(TestSkipped):
 
674
    """A test is not applicable to the situation where it was run.
 
675
 
 
676
    This is only normally raised by parameterized tests, if they find that
 
677
    the instance they're constructed upon does not support one aspect
 
678
    of its interface.
 
679
    """
 
680
 
 
681
 
 
682
# traceback._some_str fails to format exceptions that have the default
 
683
# __str__ which does an implicit ascii conversion. However, repr() on those
 
684
# objects works, for all that its not quite what the doctor may have ordered.
 
685
def _clever_some_str(value):
 
686
    try:
 
687
        return str(value)
 
688
    except:
 
689
        try:
 
690
            return repr(value).replace('\\n', '\n')
 
691
        except:
 
692
            return '<unprintable %s object>' % type(value).__name__
 
693
 
 
694
traceback._some_str = _clever_some_str
 
695
 
 
696
 
 
697
# deprecated - use self.knownFailure(), or self.expectFailure.
 
698
KnownFailure = testtools.testcase._ExpectedFailure
 
699
 
 
700
 
 
701
class UnavailableFeature(Exception):
 
702
    """A feature required for this test was not available.
 
703
 
 
704
    This can be considered a specialised form of SkippedTest.
 
705
 
 
706
    The feature should be used to construct the exception.
 
707
    """
432
708
 
433
709
 
434
710
class StringIOWrapper(object):
435
711
    """A wrapper around cStringIO which just adds an encoding attribute.
436
 
    
 
712
 
437
713
    Internally we can check sys.stdout to see what the output encoding
438
714
    should be. However, cStringIO has no encoding attribute that we can
439
715
    set. So we wrap it instead.
457
733
            return setattr(self._cstring, name, val)
458
734
 
459
735
 
460
 
class TestCase(unittest.TestCase):
 
736
class TestUIFactory(TextUIFactory):
 
737
    """A UI Factory for testing.
 
738
 
 
739
    Hide the progress bar but emit note()s.
 
740
    Redirect stdin.
 
741
    Allows get_password to be tested without real tty attached.
 
742
 
 
743
    See also CannedInputUIFactory which lets you provide programmatic input in
 
744
    a structured way.
 
745
    """
 
746
    # TODO: Capture progress events at the model level and allow them to be
 
747
    # observed by tests that care.
 
748
    #
 
749
    # XXX: Should probably unify more with CannedInputUIFactory or a
 
750
    # particular configuration of TextUIFactory, or otherwise have a clearer
 
751
    # idea of how they're supposed to be different.
 
752
    # See https://bugs.edge.launchpad.net/bzr/+bug/408213
 
753
 
 
754
    def __init__(self, stdout=None, stderr=None, stdin=None):
 
755
        if stdin is not None:
 
756
            # We use a StringIOWrapper to be able to test various
 
757
            # encodings, but the user is still responsible to
 
758
            # encode the string and to set the encoding attribute
 
759
            # of StringIOWrapper.
 
760
            stdin = StringIOWrapper(stdin)
 
761
        super(TestUIFactory, self).__init__(stdin, stdout, stderr)
 
762
 
 
763
    def get_non_echoed_password(self):
 
764
        """Get password from stdin without trying to handle the echo mode"""
 
765
        password = self.stdin.readline()
 
766
        if not password:
 
767
            raise EOFError
 
768
        if password[-1] == '\n':
 
769
            password = password[:-1]
 
770
        return password
 
771
 
 
772
    def make_progress_view(self):
 
773
        return NullProgressView()
 
774
 
 
775
 
 
776
class TestCase(testtools.TestCase):
461
777
    """Base class for bzr unit tests.
462
 
    
463
 
    Tests that need access to disk resources should subclass 
 
778
 
 
779
    Tests that need access to disk resources should subclass
464
780
    TestCaseInTempDir not TestCase.
465
781
 
466
782
    Error and debug log messages are redirected from their usual
468
784
    retrieved by _get_log().  We use a real OS file, not an in-memory object,
469
785
    so that it can also capture file IO.  When the test completes this file
470
786
    is read into memory and removed from disk.
471
 
       
 
787
 
472
788
    There are also convenience functions to invoke bzr's command-line
473
789
    routine, and to build and check bzr trees.
474
 
   
 
790
 
475
791
    In addition to the usual method of overriding tearDown(), this class also
476
792
    allows subclasses to register functions into the _cleanups list, which is
477
793
    run in order as the object is torn down.  It's less likely this will be
478
794
    accidentally overlooked.
479
795
    """
480
796
 
 
797
    _active_threads = None
 
798
    _leaking_threads_tests = 0
 
799
    _first_thread_leaker_id = None
481
800
    _log_file_name = None
482
 
    _log_contents = ''
483
801
    # record lsprof data when performing benchmark calls.
484
802
    _gather_lsprof_in_benchmarks = False
485
803
 
486
804
    def __init__(self, methodName='testMethod'):
487
805
        super(TestCase, self).__init__(methodName)
488
806
        self._cleanups = []
 
807
        self._directory_isolation = True
 
808
        self.exception_handlers.insert(0,
 
809
            (UnavailableFeature, self._do_unsupported_or_skip))
 
810
        self.exception_handlers.insert(0,
 
811
            (TestNotApplicable, self._do_not_applicable))
489
812
 
490
813
    def setUp(self):
491
 
        unittest.TestCase.setUp(self)
 
814
        super(TestCase, self).setUp()
 
815
        for feature in getattr(self, '_test_needs_features', []):
 
816
            self.requireFeature(feature)
 
817
        self._log_contents = None
 
818
        self.addDetail("log", content.Content(content.ContentType("text",
 
819
            "plain", {"charset": "utf8"}),
 
820
            lambda:[self._get_log(keep_log_file=True)]))
492
821
        self._cleanEnvironment()
493
 
        bzrlib.trace.disable_default_logging()
 
822
        self._silenceUI()
494
823
        self._startLogFile()
495
824
        self._benchcalls = []
496
825
        self._benchtime = None
 
826
        self._clear_hooks()
 
827
        self._track_transports()
 
828
        self._track_locks()
 
829
        self._clear_debug_flags()
 
830
        TestCase._active_threads = threading.activeCount()
 
831
        self.addCleanup(self._check_leaked_threads)
 
832
 
 
833
    def debug(self):
 
834
        # debug a frame up.
 
835
        import pdb
 
836
        pdb.Pdb().set_trace(sys._getframe().f_back)
 
837
 
 
838
    def _check_leaked_threads(self):
 
839
        active = threading.activeCount()
 
840
        leaked_threads = active - TestCase._active_threads
 
841
        TestCase._active_threads = active
 
842
        # If some tests make the number of threads *decrease*, we'll consider
 
843
        # that they are just observing old threads dieing, not agressively kill
 
844
        # random threads. So we don't report these tests as leaking. The risk
 
845
        # is that we have false positives that way (the test see 2 threads
 
846
        # going away but leak one) but it seems less likely than the actual
 
847
        # false positives (the test see threads going away and does not leak).
 
848
        if leaked_threads > 0:
 
849
            TestCase._leaking_threads_tests += 1
 
850
            if TestCase._first_thread_leaker_id is None:
 
851
                TestCase._first_thread_leaker_id = self.id()
 
852
 
 
853
    def _clear_debug_flags(self):
 
854
        """Prevent externally set debug flags affecting tests.
 
855
 
 
856
        Tests that want to use debug flags can just set them in the
 
857
        debug_flags set during setup/teardown.
 
858
        """
 
859
        # Start with a copy of the current debug flags we can safely modify.
 
860
        self.overrideAttr(debug, 'debug_flags', set(debug.debug_flags))
 
861
        if 'allow_debug' not in selftest_debug_flags:
 
862
            debug.debug_flags.clear()
 
863
        if 'disable_lock_checks' not in selftest_debug_flags:
 
864
            debug.debug_flags.add('strict_locks')
 
865
 
 
866
    def _clear_hooks(self):
 
867
        # prevent hooks affecting tests
 
868
        self._preserved_hooks = {}
 
869
        for key, factory in hooks.known_hooks.items():
 
870
            parent, name = hooks.known_hooks_key_to_parent_and_attribute(key)
 
871
            current_hooks = hooks.known_hooks_key_to_object(key)
 
872
            self._preserved_hooks[parent] = (name, current_hooks)
 
873
        self.addCleanup(self._restoreHooks)
 
874
        for key, factory in hooks.known_hooks.items():
 
875
            parent, name = hooks.known_hooks_key_to_parent_and_attribute(key)
 
876
            setattr(parent, name, factory())
 
877
        # this hook should always be installed
 
878
        request._install_hook()
 
879
 
 
880
    def disable_directory_isolation(self):
 
881
        """Turn off directory isolation checks."""
 
882
        self._directory_isolation = False
 
883
 
 
884
    def enable_directory_isolation(self):
 
885
        """Enable directory isolation checks."""
 
886
        self._directory_isolation = True
 
887
 
 
888
    def _silenceUI(self):
 
889
        """Turn off UI for duration of test"""
 
890
        # by default the UI is off; tests can turn it on if they want it.
 
891
        self.overrideAttr(ui, 'ui_factory', ui.SilentUIFactory())
 
892
 
 
893
    def _check_locks(self):
 
894
        """Check that all lock take/release actions have been paired."""
 
895
        # We always check for mismatched locks. If a mismatch is found, we
 
896
        # fail unless -Edisable_lock_checks is supplied to selftest, in which
 
897
        # case we just print a warning.
 
898
        # unhook:
 
899
        acquired_locks = [lock for action, lock in self._lock_actions
 
900
                          if action == 'acquired']
 
901
        released_locks = [lock for action, lock in self._lock_actions
 
902
                          if action == 'released']
 
903
        broken_locks = [lock for action, lock in self._lock_actions
 
904
                        if action == 'broken']
 
905
        # trivially, given the tests for lock acquistion and release, if we
 
906
        # have as many in each list, it should be ok. Some lock tests also
 
907
        # break some locks on purpose and should be taken into account by
 
908
        # considering that breaking a lock is just a dirty way of releasing it.
 
909
        if len(acquired_locks) != (len(released_locks) + len(broken_locks)):
 
910
            message = ('Different number of acquired and '
 
911
                       'released or broken locks. (%s, %s + %s)' %
 
912
                       (acquired_locks, released_locks, broken_locks))
 
913
            if not self._lock_check_thorough:
 
914
                # Rather than fail, just warn
 
915
                print "Broken test %s: %s" % (self, message)
 
916
                return
 
917
            self.fail(message)
 
918
 
 
919
    def _track_locks(self):
 
920
        """Track lock activity during tests."""
 
921
        self._lock_actions = []
 
922
        if 'disable_lock_checks' in selftest_debug_flags:
 
923
            self._lock_check_thorough = False
 
924
        else:
 
925
            self._lock_check_thorough = True
 
926
 
 
927
        self.addCleanup(self._check_locks)
 
928
        _mod_lock.Lock.hooks.install_named_hook('lock_acquired',
 
929
                                                self._lock_acquired, None)
 
930
        _mod_lock.Lock.hooks.install_named_hook('lock_released',
 
931
                                                self._lock_released, None)
 
932
        _mod_lock.Lock.hooks.install_named_hook('lock_broken',
 
933
                                                self._lock_broken, None)
 
934
 
 
935
    def _lock_acquired(self, result):
 
936
        self._lock_actions.append(('acquired', result))
 
937
 
 
938
    def _lock_released(self, result):
 
939
        self._lock_actions.append(('released', result))
 
940
 
 
941
    def _lock_broken(self, result):
 
942
        self._lock_actions.append(('broken', result))
 
943
 
 
944
    def permit_dir(self, name):
 
945
        """Permit a directory to be used by this test. See permit_url."""
 
946
        name_transport = get_transport(name)
 
947
        self.permit_url(name)
 
948
        self.permit_url(name_transport.base)
 
949
 
 
950
    def permit_url(self, url):
 
951
        """Declare that url is an ok url to use in this test.
 
952
        
 
953
        Do this for memory transports, temporary test directory etc.
 
954
        
 
955
        Do not do this for the current working directory, /tmp, or any other
 
956
        preexisting non isolated url.
 
957
        """
 
958
        if not url.endswith('/'):
 
959
            url += '/'
 
960
        self._bzr_selftest_roots.append(url)
 
961
 
 
962
    def permit_source_tree_branch_repo(self):
 
963
        """Permit the source tree bzr is running from to be opened.
 
964
 
 
965
        Some code such as bzrlib.version attempts to read from the bzr branch
 
966
        that bzr is executing from (if any). This method permits that directory
 
967
        to be used in the test suite.
 
968
        """
 
969
        path = self.get_source_path()
 
970
        self.record_directory_isolation()
 
971
        try:
 
972
            try:
 
973
                workingtree.WorkingTree.open(path)
 
974
            except (errors.NotBranchError, errors.NoWorkingTree):
 
975
                return
 
976
        finally:
 
977
            self.enable_directory_isolation()
 
978
 
 
979
    def _preopen_isolate_transport(self, transport):
 
980
        """Check that all transport openings are done in the test work area."""
 
981
        while isinstance(transport, pathfilter.PathFilteringTransport):
 
982
            # Unwrap pathfiltered transports
 
983
            transport = transport.server.backing_transport.clone(
 
984
                transport._filter('.'))
 
985
        url = transport.base
 
986
        # ReadonlySmartTCPServer_for_testing decorates the backing transport
 
987
        # urls it is given by prepending readonly+. This is appropriate as the
 
988
        # client shouldn't know that the server is readonly (or not readonly).
 
989
        # We could register all servers twice, with readonly+ prepending, but
 
990
        # that makes for a long list; this is about the same but easier to
 
991
        # read.
 
992
        if url.startswith('readonly+'):
 
993
            url = url[len('readonly+'):]
 
994
        self._preopen_isolate_url(url)
 
995
 
 
996
    def _preopen_isolate_url(self, url):
 
997
        if not self._directory_isolation:
 
998
            return
 
999
        if self._directory_isolation == 'record':
 
1000
            self._bzr_selftest_roots.append(url)
 
1001
            return
 
1002
        # This prevents all transports, including e.g. sftp ones backed on disk
 
1003
        # from working unless they are explicitly granted permission. We then
 
1004
        # depend on the code that sets up test transports to check that they are
 
1005
        # appropriately isolated and enable their use by calling
 
1006
        # self.permit_transport()
 
1007
        if not osutils.is_inside_any(self._bzr_selftest_roots, url):
 
1008
            raise errors.BzrError("Attempt to escape test isolation: %r %r"
 
1009
                % (url, self._bzr_selftest_roots))
 
1010
 
 
1011
    def record_directory_isolation(self):
 
1012
        """Gather accessed directories to permit later access.
 
1013
        
 
1014
        This is used for tests that access the branch bzr is running from.
 
1015
        """
 
1016
        self._directory_isolation = "record"
 
1017
 
 
1018
    def start_server(self, transport_server, backing_server=None):
 
1019
        """Start transport_server for this test.
 
1020
 
 
1021
        This starts the server, registers a cleanup for it and permits the
 
1022
        server's urls to be used.
 
1023
        """
 
1024
        if backing_server is None:
 
1025
            transport_server.start_server()
 
1026
        else:
 
1027
            transport_server.start_server(backing_server)
 
1028
        self.addCleanup(transport_server.stop_server)
 
1029
        # Obtain a real transport because if the server supplies a password, it
 
1030
        # will be hidden from the base on the client side.
 
1031
        t = get_transport(transport_server.get_url())
 
1032
        # Some transport servers effectively chroot the backing transport;
 
1033
        # others like SFTPServer don't - users of the transport can walk up the
 
1034
        # transport to read the entire backing transport. This wouldn't matter
 
1035
        # except that the workdir tests are given - and that they expect the
 
1036
        # server's url to point at - is one directory under the safety net. So
 
1037
        # Branch operations into the transport will attempt to walk up one
 
1038
        # directory. Chrooting all servers would avoid this but also mean that
 
1039
        # we wouldn't be testing directly against non-root urls. Alternatively
 
1040
        # getting the test framework to start the server with a backing server
 
1041
        # at the actual safety net directory would work too, but this then
 
1042
        # means that the self.get_url/self.get_transport methods would need
 
1043
        # to transform all their results. On balance its cleaner to handle it
 
1044
        # here, and permit a higher url when we have one of these transports.
 
1045
        if t.base.endswith('/work/'):
 
1046
            # we have safety net/test root/work
 
1047
            t = t.clone('../..')
 
1048
        elif isinstance(transport_server,
 
1049
                        test_server.SmartTCPServer_for_testing):
 
1050
            # The smart server adds a path similar to work, which is traversed
 
1051
            # up from by the client. But the server is chrooted - the actual
 
1052
            # backing transport is not escaped from, and VFS requests to the
 
1053
            # root will error (because they try to escape the chroot).
 
1054
            t2 = t.clone('..')
 
1055
            while t2.base != t.base:
 
1056
                t = t2
 
1057
                t2 = t.clone('..')
 
1058
        self.permit_url(t.base)
 
1059
 
 
1060
    def _track_transports(self):
 
1061
        """Install checks for transport usage."""
 
1062
        # TestCase has no safe place it can write to.
 
1063
        self._bzr_selftest_roots = []
 
1064
        # Currently the easiest way to be sure that nothing is going on is to
 
1065
        # hook into bzr dir opening. This leaves a small window of error for
 
1066
        # transport tests, but they are well known, and we can improve on this
 
1067
        # step.
 
1068
        bzrdir.BzrDir.hooks.install_named_hook("pre_open",
 
1069
            self._preopen_isolate_transport, "Check bzr directories are safe.")
497
1070
 
498
1071
    def _ndiff_strings(self, a, b):
499
1072
        """Return ndiff between two strings containing lines.
500
 
        
 
1073
 
501
1074
        A trailing newline is added if missing to make the strings
502
1075
        print properly."""
503
1076
        if b and b[-1] != '\n':
510
1083
                                  charjunk=lambda x: False)
511
1084
        return ''.join(difflines)
512
1085
 
 
1086
    def assertEqual(self, a, b, message=''):
 
1087
        try:
 
1088
            if a == b:
 
1089
                return
 
1090
        except UnicodeError, e:
 
1091
            # If we can't compare without getting a UnicodeError, then
 
1092
            # obviously they are different
 
1093
            mutter('UnicodeError: %s', e)
 
1094
        if message:
 
1095
            message += '\n'
 
1096
        raise AssertionError("%snot equal:\na = %s\nb = %s\n"
 
1097
            % (message,
 
1098
               pformat(a), pformat(b)))
 
1099
 
 
1100
    assertEquals = assertEqual
 
1101
 
513
1102
    def assertEqualDiff(self, a, b, message=None):
514
1103
        """Assert two texts are equal, if not raise an exception.
515
 
        
516
 
        This is intended for use with multi-line strings where it can 
 
1104
 
 
1105
        This is intended for use with multi-line strings where it can
517
1106
        be hard to find the differences by eye.
518
1107
        """
519
1108
        # TODO: perhaps override assertEquals to call this for strings?
521
1110
            return
522
1111
        if message is None:
523
1112
            message = "texts not equal:\n"
524
 
        raise AssertionError(message + 
525
 
                             self._ndiff_strings(a, b))      
526
 
        
 
1113
        if a + '\n' == b:
 
1114
            message = 'first string is missing a final newline.\n'
 
1115
        if a == b + '\n':
 
1116
            message = 'second string is missing a final newline.\n'
 
1117
        raise AssertionError(message +
 
1118
                             self._ndiff_strings(a, b))
 
1119
 
527
1120
    def assertEqualMode(self, mode, mode_test):
528
1121
        self.assertEqual(mode, mode_test,
529
1122
                         'mode mismatch %o != %o' % (mode, mode_test))
530
1123
 
 
1124
    def assertEqualStat(self, expected, actual):
 
1125
        """assert that expected and actual are the same stat result.
 
1126
 
 
1127
        :param expected: A stat result.
 
1128
        :param actual: A stat result.
 
1129
        :raises AssertionError: If the expected and actual stat values differ
 
1130
            other than by atime.
 
1131
        """
 
1132
        self.assertEqual(expected.st_size, actual.st_size,
 
1133
                         'st_size did not match')
 
1134
        self.assertEqual(expected.st_mtime, actual.st_mtime,
 
1135
                         'st_mtime did not match')
 
1136
        self.assertEqual(expected.st_ctime, actual.st_ctime,
 
1137
                         'st_ctime did not match')
 
1138
        if sys.platform != 'win32':
 
1139
            # On Win32 both 'dev' and 'ino' cannot be trusted. In python2.4 it
 
1140
            # is 'dev' that varies, in python 2.5 (6?) it is st_ino that is
 
1141
            # odd. Regardless we shouldn't actually try to assert anything
 
1142
            # about their values
 
1143
            self.assertEqual(expected.st_dev, actual.st_dev,
 
1144
                             'st_dev did not match')
 
1145
            self.assertEqual(expected.st_ino, actual.st_ino,
 
1146
                             'st_ino did not match')
 
1147
        self.assertEqual(expected.st_mode, actual.st_mode,
 
1148
                         'st_mode did not match')
 
1149
 
 
1150
    def assertLength(self, length, obj_with_len):
 
1151
        """Assert that obj_with_len is of length length."""
 
1152
        if len(obj_with_len) != length:
 
1153
            self.fail("Incorrect length: wanted %d, got %d for %r" % (
 
1154
                length, len(obj_with_len), obj_with_len))
 
1155
 
 
1156
    def assertLogsError(self, exception_class, func, *args, **kwargs):
 
1157
        """Assert that func(*args, **kwargs) quietly logs a specific exception.
 
1158
        """
 
1159
        from bzrlib import trace
 
1160
        captured = []
 
1161
        orig_log_exception_quietly = trace.log_exception_quietly
 
1162
        try:
 
1163
            def capture():
 
1164
                orig_log_exception_quietly()
 
1165
                captured.append(sys.exc_info())
 
1166
            trace.log_exception_quietly = capture
 
1167
            func(*args, **kwargs)
 
1168
        finally:
 
1169
            trace.log_exception_quietly = orig_log_exception_quietly
 
1170
        self.assertLength(1, captured)
 
1171
        err = captured[0][1]
 
1172
        self.assertIsInstance(err, exception_class)
 
1173
        return err
 
1174
 
 
1175
    def assertPositive(self, val):
 
1176
        """Assert that val is greater than 0."""
 
1177
        self.assertTrue(val > 0, 'expected a positive value, but got %s' % val)
 
1178
 
 
1179
    def assertNegative(self, val):
 
1180
        """Assert that val is less than 0."""
 
1181
        self.assertTrue(val < 0, 'expected a negative value, but got %s' % val)
 
1182
 
531
1183
    def assertStartsWith(self, s, prefix):
532
1184
        if not s.startswith(prefix):
533
1185
            raise AssertionError('string %r does not start with %r' % (s, prefix))
537
1189
        if not s.endswith(suffix):
538
1190
            raise AssertionError('string %r does not end with %r' % (s, suffix))
539
1191
 
540
 
    def assertContainsRe(self, haystack, needle_re):
 
1192
    def assertContainsRe(self, haystack, needle_re, flags=0):
541
1193
        """Assert that a contains something matching a regular expression."""
542
 
        if not re.search(needle_re, haystack):
543
 
            raise AssertionError('pattern "%s" not found in "%s"'
544
 
                    % (needle_re, haystack))
 
1194
        if not re.search(needle_re, haystack, flags):
 
1195
            if '\n' in haystack or len(haystack) > 60:
 
1196
                # a long string, format it in a more readable way
 
1197
                raise AssertionError(
 
1198
                        'pattern "%s" not found in\n"""\\\n%s"""\n'
 
1199
                        % (needle_re, haystack))
 
1200
            else:
 
1201
                raise AssertionError('pattern "%s" not found in "%s"'
 
1202
                        % (needle_re, haystack))
545
1203
 
546
 
    def assertNotContainsRe(self, haystack, needle_re):
 
1204
    def assertNotContainsRe(self, haystack, needle_re, flags=0):
547
1205
        """Assert that a does not match a regular expression"""
548
 
        if re.search(needle_re, haystack):
 
1206
        if re.search(needle_re, haystack, flags):
549
1207
            raise AssertionError('pattern "%s" found in "%s"'
550
1208
                    % (needle_re, haystack))
551
1209
 
 
1210
    def assertContainsString(self, haystack, needle):
 
1211
        if haystack.find(needle) == -1:
 
1212
            self.fail("string %r not found in '''%s'''" % (needle, haystack))
 
1213
 
552
1214
    def assertSubset(self, sublist, superlist):
553
1215
        """Assert that every entry in sublist is present in superlist."""
554
 
        missing = []
555
 
        for entry in sublist:
556
 
            if entry not in superlist:
557
 
                missing.append(entry)
 
1216
        missing = set(sublist) - set(superlist)
558
1217
        if len(missing) > 0:
559
 
            raise AssertionError("value(s) %r not present in container %r" % 
 
1218
            raise AssertionError("value(s) %r not present in container %r" %
560
1219
                                 (missing, superlist))
561
1220
 
562
 
    def assertIs(self, left, right):
 
1221
    def assertListRaises(self, excClass, func, *args, **kwargs):
 
1222
        """Fail unless excClass is raised when the iterator from func is used.
 
1223
 
 
1224
        Many functions can return generators this makes sure
 
1225
        to wrap them in a list() call to make sure the whole generator
 
1226
        is run, and that the proper exception is raised.
 
1227
        """
 
1228
        try:
 
1229
            list(func(*args, **kwargs))
 
1230
        except excClass, e:
 
1231
            return e
 
1232
        else:
 
1233
            if getattr(excClass,'__name__', None) is not None:
 
1234
                excName = excClass.__name__
 
1235
            else:
 
1236
                excName = str(excClass)
 
1237
            raise self.failureException, "%s not raised" % excName
 
1238
 
 
1239
    def assertRaises(self, excClass, callableObj, *args, **kwargs):
 
1240
        """Assert that a callable raises a particular exception.
 
1241
 
 
1242
        :param excClass: As for the except statement, this may be either an
 
1243
            exception class, or a tuple of classes.
 
1244
        :param callableObj: A callable, will be passed ``*args`` and
 
1245
            ``**kwargs``.
 
1246
 
 
1247
        Returns the exception so that you can examine it.
 
1248
        """
 
1249
        try:
 
1250
            callableObj(*args, **kwargs)
 
1251
        except excClass, e:
 
1252
            return e
 
1253
        else:
 
1254
            if getattr(excClass,'__name__', None) is not None:
 
1255
                excName = excClass.__name__
 
1256
            else:
 
1257
                # probably a tuple
 
1258
                excName = str(excClass)
 
1259
            raise self.failureException, "%s not raised" % excName
 
1260
 
 
1261
    def assertIs(self, left, right, message=None):
563
1262
        if not (left is right):
564
 
            raise AssertionError("%r is not %r." % (left, right))
 
1263
            if message is not None:
 
1264
                raise AssertionError(message)
 
1265
            else:
 
1266
                raise AssertionError("%r is not %r." % (left, right))
 
1267
 
 
1268
    def assertIsNot(self, left, right, message=None):
 
1269
        if (left is right):
 
1270
            if message is not None:
 
1271
                raise AssertionError(message)
 
1272
            else:
 
1273
                raise AssertionError("%r is %r." % (left, right))
565
1274
 
566
1275
    def assertTransportMode(self, transport, path, mode):
567
 
        """Fail if a path does not have mode mode.
568
 
        
 
1276
        """Fail if a path does not have mode "mode".
 
1277
 
569
1278
        If modes are not supported on this transport, the assertion is ignored.
570
1279
        """
571
1280
        if not transport._can_roundtrip_unix_modebits():
573
1282
        path_stat = transport.stat(path)
574
1283
        actual_mode = stat.S_IMODE(path_stat.st_mode)
575
1284
        self.assertEqual(mode, actual_mode,
576
 
            'mode of %r incorrect (%o != %o)' % (path, mode, actual_mode))
577
 
 
578
 
    def assertIsInstance(self, obj, kls):
579
 
        """Fail if obj is not an instance of kls"""
 
1285
                         'mode of %r incorrect (%s != %s)'
 
1286
                         % (path, oct(mode), oct(actual_mode)))
 
1287
 
 
1288
    def assertIsSameRealPath(self, path1, path2):
 
1289
        """Fail if path1 and path2 points to different files"""
 
1290
        self.assertEqual(osutils.realpath(path1),
 
1291
                         osutils.realpath(path2),
 
1292
                         "apparent paths:\na = %s\nb = %s\n," % (path1, path2))
 
1293
 
 
1294
    def assertIsInstance(self, obj, kls, msg=None):
 
1295
        """Fail if obj is not an instance of kls
 
1296
        
 
1297
        :param msg: Supplementary message to show if the assertion fails.
 
1298
        """
580
1299
        if not isinstance(obj, kls):
581
 
            self.fail("%r is an instance of %s rather than %s" % (
582
 
                obj, obj.__class__, kls))
583
 
 
584
 
    def _capture_warnings(self, a_callable, *args, **kwargs):
 
1300
            m = "%r is an instance of %s rather than %s" % (
 
1301
                obj, obj.__class__, kls)
 
1302
            if msg:
 
1303
                m += ": " + msg
 
1304
            self.fail(m)
 
1305
 
 
1306
    def assertFileEqual(self, content, path):
 
1307
        """Fail if path does not contain 'content'."""
 
1308
        self.failUnlessExists(path)
 
1309
        f = file(path, 'rb')
 
1310
        try:
 
1311
            s = f.read()
 
1312
        finally:
 
1313
            f.close()
 
1314
        self.assertEqualDiff(content, s)
 
1315
 
 
1316
    def assertDocstring(self, expected_docstring, obj):
 
1317
        """Fail if obj does not have expected_docstring"""
 
1318
        if __doc__ is None:
 
1319
            # With -OO the docstring should be None instead
 
1320
            self.assertIs(obj.__doc__, None)
 
1321
        else:
 
1322
            self.assertEqual(expected_docstring, obj.__doc__)
 
1323
 
 
1324
    def failUnlessExists(self, path):
 
1325
        """Fail unless path or paths, which may be abs or relative, exist."""
 
1326
        if not isinstance(path, basestring):
 
1327
            for p in path:
 
1328
                self.failUnlessExists(p)
 
1329
        else:
 
1330
            self.failUnless(osutils.lexists(path),path+" does not exist")
 
1331
 
 
1332
    def failIfExists(self, path):
 
1333
        """Fail if path or paths, which may be abs or relative, exist."""
 
1334
        if not isinstance(path, basestring):
 
1335
            for p in path:
 
1336
                self.failIfExists(p)
 
1337
        else:
 
1338
            self.failIf(osutils.lexists(path),path+" exists")
 
1339
 
 
1340
    def _capture_deprecation_warnings(self, a_callable, *args, **kwargs):
585
1341
        """A helper for callDeprecated and applyDeprecated.
586
1342
 
587
1343
        :param a_callable: A callable to call.
588
1344
        :param args: The positional arguments for the callable
589
1345
        :param kwargs: The keyword arguments for the callable
590
1346
        :return: A tuple (warnings, result). result is the result of calling
591
 
            a_callable(*args, **kwargs).
 
1347
            a_callable(``*args``, ``**kwargs``).
592
1348
        """
593
1349
        local_warnings = []
594
 
        def capture_warnings(msg, cls, stacklevel=None):
 
1350
        def capture_warnings(msg, cls=None, stacklevel=None):
595
1351
            # we've hooked into a deprecation specific callpath,
596
1352
            # only deprecations should getting sent via it.
597
1353
            self.assertEqual(cls, DeprecationWarning)
607
1363
    def applyDeprecated(self, deprecation_format, a_callable, *args, **kwargs):
608
1364
        """Call a deprecated callable without warning the user.
609
1365
 
 
1366
        Note that this only captures warnings raised by symbol_versioning.warn,
 
1367
        not other callers that go direct to the warning module.
 
1368
 
 
1369
        To test that a deprecated method raises an error, do something like
 
1370
        this::
 
1371
 
 
1372
            self.assertRaises(errors.ReservedId,
 
1373
                self.applyDeprecated,
 
1374
                deprecated_in((1, 5, 0)),
 
1375
                br.append_revision,
 
1376
                'current:')
 
1377
 
610
1378
        :param deprecation_format: The deprecation format that the callable
611
 
            should have been deprecated with. This is the same type as the 
612
 
            parameter to deprecated_method/deprecated_function. If the 
 
1379
            should have been deprecated with. This is the same type as the
 
1380
            parameter to deprecated_method/deprecated_function. If the
613
1381
            callable is not deprecated with this format, an assertion error
614
1382
            will be raised.
615
1383
        :param a_callable: A callable to call. This may be a bound method or
616
 
            a regular function. It will be called with *args and **kwargs.
 
1384
            a regular function. It will be called with ``*args`` and
 
1385
            ``**kwargs``.
617
1386
        :param args: The positional arguments for the callable
618
1387
        :param kwargs: The keyword arguments for the callable
619
 
        :return: The result of a_callable(*args, **kwargs)
 
1388
        :return: The result of a_callable(``*args``, ``**kwargs``)
620
1389
        """
621
 
        call_warnings, result = self._capture_warnings(a_callable,
 
1390
        call_warnings, result = self._capture_deprecation_warnings(a_callable,
622
1391
            *args, **kwargs)
623
1392
        expected_first_warning = symbol_versioning.deprecation_string(
624
1393
            a_callable, deprecation_format)
625
1394
        if len(call_warnings) == 0:
626
 
            self.fail("No assertion generated by call to %s" %
 
1395
            self.fail("No deprecation warning generated by call to %s" %
627
1396
                a_callable)
628
1397
        self.assertEqual(expected_first_warning, call_warnings[0])
629
1398
        return result
630
1399
 
 
1400
    def callCatchWarnings(self, fn, *args, **kw):
 
1401
        """Call a callable that raises python warnings.
 
1402
 
 
1403
        The caller's responsible for examining the returned warnings.
 
1404
 
 
1405
        If the callable raises an exception, the exception is not
 
1406
        caught and propagates up to the caller.  In that case, the list
 
1407
        of warnings is not available.
 
1408
 
 
1409
        :returns: ([warning_object, ...], fn_result)
 
1410
        """
 
1411
        # XXX: This is not perfect, because it completely overrides the
 
1412
        # warnings filters, and some code may depend on suppressing particular
 
1413
        # warnings.  It's the easiest way to insulate ourselves from -Werror,
 
1414
        # though.  -- Andrew, 20071062
 
1415
        wlist = []
 
1416
        def _catcher(message, category, filename, lineno, file=None, line=None):
 
1417
            # despite the name, 'message' is normally(?) a Warning subclass
 
1418
            # instance
 
1419
            wlist.append(message)
 
1420
        saved_showwarning = warnings.showwarning
 
1421
        saved_filters = warnings.filters
 
1422
        try:
 
1423
            warnings.showwarning = _catcher
 
1424
            warnings.filters = []
 
1425
            result = fn(*args, **kw)
 
1426
        finally:
 
1427
            warnings.showwarning = saved_showwarning
 
1428
            warnings.filters = saved_filters
 
1429
        return wlist, result
 
1430
 
631
1431
    def callDeprecated(self, expected, callable, *args, **kwargs):
632
1432
        """Assert that a callable is deprecated in a particular way.
633
1433
 
634
 
        This is a very precise test for unusual requirements. The 
 
1434
        This is a very precise test for unusual requirements. The
635
1435
        applyDeprecated helper function is probably more suited for most tests
636
1436
        as it allows you to simply specify the deprecation format being used
637
1437
        and will ensure that that is issued for the function being called.
638
1438
 
 
1439
        Note that this only captures warnings raised by symbol_versioning.warn,
 
1440
        not other callers that go direct to the warning module.  To catch
 
1441
        general warnings, use callCatchWarnings.
 
1442
 
639
1443
        :param expected: a list of the deprecation warnings expected, in order
640
1444
        :param callable: The callable to call
641
1445
        :param args: The positional arguments for the callable
642
1446
        :param kwargs: The keyword arguments for the callable
643
1447
        """
644
 
        call_warnings, result = self._capture_warnings(callable,
 
1448
        call_warnings, result = self._capture_deprecation_warnings(callable,
645
1449
            *args, **kwargs)
646
1450
        self.assertEqual(expected, call_warnings)
647
1451
        return result
653
1457
        """
654
1458
        fileno, name = tempfile.mkstemp(suffix='.log', prefix='testbzr')
655
1459
        self._log_file = os.fdopen(fileno, 'w+')
656
 
        self._log_nonce = bzrlib.trace.enable_test_log(self._log_file)
 
1460
        self._log_memento = bzrlib.trace.push_log_file(self._log_file)
657
1461
        self._log_file_name = name
658
1462
        self.addCleanup(self._finishLogFile)
659
1463
 
660
1464
    def _finishLogFile(self):
661
1465
        """Finished with the log file.
662
1466
 
663
 
        Read contents into memory, close, and delete.
664
 
        """
665
 
        if self._log_file is None:
666
 
            return
667
 
        bzrlib.trace.disable_test_log(self._log_nonce)
668
 
        self._log_file.seek(0)
669
 
        self._log_contents = self._log_file.read()
670
 
        self._log_file.close()
671
 
        os.remove(self._log_file_name)
672
 
        self._log_file = self._log_file_name = None
673
 
 
674
 
    def addCleanup(self, callable):
 
1467
        Close the file and delete it, unless setKeepLogfile was called.
 
1468
        """
 
1469
        if bzrlib.trace._trace_file:
 
1470
            # flush the log file, to get all content
 
1471
            bzrlib.trace._trace_file.flush()
 
1472
        bzrlib.trace.pop_log_file(self._log_memento)
 
1473
        # Cache the log result and delete the file on disk
 
1474
        self._get_log(False)
 
1475
 
 
1476
    def thisFailsStrictLockCheck(self):
 
1477
        """It is known that this test would fail with -Dstrict_locks.
 
1478
 
 
1479
        By default, all tests are run with strict lock checking unless
 
1480
        -Edisable_lock_checks is supplied. However there are some tests which
 
1481
        we know fail strict locks at this point that have not been fixed.
 
1482
        They should call this function to disable the strict checking.
 
1483
 
 
1484
        This should be used sparingly, it is much better to fix the locking
 
1485
        issues rather than papering over the problem by calling this function.
 
1486
        """
 
1487
        debug.debug_flags.discard('strict_locks')
 
1488
 
 
1489
    def addCleanup(self, callable, *args, **kwargs):
675
1490
        """Arrange to run a callable when this case is torn down.
676
1491
 
677
 
        Callables are run in the reverse of the order they are registered, 
 
1492
        Callables are run in the reverse of the order they are registered,
678
1493
        ie last-in first-out.
679
1494
        """
680
 
        if callable in self._cleanups:
681
 
            raise ValueError("cleanup function %r already registered on %s" 
682
 
                    % (callable, self))
683
 
        self._cleanups.append(callable)
 
1495
        self._cleanups.append((callable, args, kwargs))
 
1496
 
 
1497
    def overrideAttr(self, obj, attr_name, new=_unitialized_attr):
 
1498
        """Overrides an object attribute restoring it after the test.
 
1499
 
 
1500
        :param obj: The object that will be mutated.
 
1501
 
 
1502
        :param attr_name: The attribute name we want to preserve/override in
 
1503
            the object.
 
1504
 
 
1505
        :param new: The optional value we want to set the attribute to.
 
1506
 
 
1507
        :returns: The actual attr value.
 
1508
        """
 
1509
        value = getattr(obj, attr_name)
 
1510
        # The actual value is captured by the call below
 
1511
        self.addCleanup(setattr, obj, attr_name, value)
 
1512
        if new is not _unitialized_attr:
 
1513
            setattr(obj, attr_name, new)
 
1514
        return value
684
1515
 
685
1516
    def _cleanEnvironment(self):
686
1517
        new_env = {
 
1518
            'BZR_HOME': None, # Don't inherit BZR_HOME to all the tests.
687
1519
            'HOME': os.getcwd(),
688
 
            'APPDATA': os.getcwd(),
 
1520
            # bzr now uses the Win32 API and doesn't rely on APPDATA, but the
 
1521
            # tests do check our impls match APPDATA
 
1522
            'BZR_EDITOR': None, # test_msgeditor manipulates this variable
 
1523
            'VISUAL': None,
 
1524
            'EDITOR': None,
689
1525
            'BZR_EMAIL': None,
690
1526
            'BZREMAIL': None, # may still be present in the environment
691
 
            'EMAIL': None,
 
1527
            'EMAIL': 'jrandom@example.com', # set EMAIL as bzr does not guess
692
1528
            'BZR_PROGRESS_BAR': None,
 
1529
            'BZR_LOG': None,
 
1530
            'BZR_PLUGIN_PATH': None,
 
1531
            'BZR_DISABLE_PLUGINS': None,
 
1532
            'BZR_PLUGINS_AT': None,
 
1533
            'BZR_CONCURRENCY': None,
 
1534
            # Make sure that any text ui tests are consistent regardless of
 
1535
            # the environment the test case is run in; you may want tests that
 
1536
            # test other combinations.  'dumb' is a reasonable guess for tests
 
1537
            # going to a pipe or a StringIO.
 
1538
            'TERM': 'dumb',
 
1539
            'LINES': '25',
 
1540
            'COLUMNS': '80',
 
1541
            'BZR_COLUMNS': '80',
 
1542
            # SSH Agent
 
1543
            'SSH_AUTH_SOCK': None,
 
1544
            # Proxies
 
1545
            'http_proxy': None,
 
1546
            'HTTP_PROXY': None,
 
1547
            'https_proxy': None,
 
1548
            'HTTPS_PROXY': None,
 
1549
            'no_proxy': None,
 
1550
            'NO_PROXY': None,
 
1551
            'all_proxy': None,
 
1552
            'ALL_PROXY': None,
 
1553
            # Nobody cares about ftp_proxy, FTP_PROXY AFAIK. So far at
 
1554
            # least. If you do (care), please update this comment
 
1555
            # -- vila 20080401
 
1556
            'ftp_proxy': None,
 
1557
            'FTP_PROXY': None,
 
1558
            'BZR_REMOTE_PATH': None,
 
1559
            # Generally speaking, we don't want apport reporting on crashes in
 
1560
            # the test envirnoment unless we're specifically testing apport,
 
1561
            # so that it doesn't leak into the real system environment.  We
 
1562
            # use an env var so it propagates to subprocesses.
 
1563
            'APPORT_DISABLE': '1',
693
1564
        }
694
 
        self.__old_env = {}
 
1565
        self._old_env = {}
695
1566
        self.addCleanup(self._restoreEnvironment)
696
1567
        for name, value in new_env.iteritems():
697
1568
            self._captureVar(name, value)
698
1569
 
699
1570
    def _captureVar(self, name, newvalue):
700
1571
        """Set an environment variable, and reset it when finished."""
701
 
        self.__old_env[name] = osutils.set_or_unset_env(name, newvalue)
 
1572
        self._old_env[name] = osutils.set_or_unset_env(name, newvalue)
702
1573
 
703
1574
    def _restoreEnvironment(self):
704
 
        for name, value in self.__old_env.iteritems():
 
1575
        for name, value in self._old_env.iteritems():
705
1576
            osutils.set_or_unset_env(name, value)
706
1577
 
707
 
    def tearDown(self):
708
 
        self._runCleanups()
709
 
        unittest.TestCase.tearDown(self)
 
1578
    def _restoreHooks(self):
 
1579
        for klass, (name, hooks) in self._preserved_hooks.items():
 
1580
            setattr(klass, name, hooks)
 
1581
 
 
1582
    def knownFailure(self, reason):
 
1583
        """This test has failed for some known reason."""
 
1584
        raise KnownFailure(reason)
 
1585
 
 
1586
    def _do_skip(self, result, reason):
 
1587
        addSkip = getattr(result, 'addSkip', None)
 
1588
        if not callable(addSkip):
 
1589
            result.addSuccess(result)
 
1590
        else:
 
1591
            addSkip(self, reason)
 
1592
 
 
1593
    @staticmethod
 
1594
    def _do_known_failure(self, result, e):
 
1595
        err = sys.exc_info()
 
1596
        addExpectedFailure = getattr(result, 'addExpectedFailure', None)
 
1597
        if addExpectedFailure is not None:
 
1598
            addExpectedFailure(self, err)
 
1599
        else:
 
1600
            result.addSuccess(self)
 
1601
 
 
1602
    @staticmethod
 
1603
    def _do_not_applicable(self, result, e):
 
1604
        if not e.args:
 
1605
            reason = 'No reason given'
 
1606
        else:
 
1607
            reason = e.args[0]
 
1608
        addNotApplicable = getattr(result, 'addNotApplicable', None)
 
1609
        if addNotApplicable is not None:
 
1610
            result.addNotApplicable(self, reason)
 
1611
        else:
 
1612
            self._do_skip(result, reason)
 
1613
 
 
1614
    @staticmethod
 
1615
    def _do_unsupported_or_skip(self, result, e):
 
1616
        reason = e.args[0]
 
1617
        addNotSupported = getattr(result, 'addNotSupported', None)
 
1618
        if addNotSupported is not None:
 
1619
            result.addNotSupported(self, reason)
 
1620
        else:
 
1621
            self._do_skip(result, reason)
710
1622
 
711
1623
    def time(self, callable, *args, **kwargs):
712
1624
        """Run callable and accrue the time it takes to the benchmark time.
713
 
        
 
1625
 
714
1626
        If lsprofiling is enabled (i.e. by --lsprof-time to bzr selftest) then
715
1627
        this will cause lsprofile statistics to be gathered and stored in
716
1628
        self._benchcalls.
717
1629
        """
718
1630
        if self._benchtime is None:
 
1631
            self.addDetail('benchtime', content.Content(content.ContentType(
 
1632
                "text", "plain"), lambda:[str(self._benchtime)]))
719
1633
            self._benchtime = 0
720
1634
        start = time.time()
721
1635
        try:
730
1644
        finally:
731
1645
            self._benchtime += time.time() - start
732
1646
 
733
 
    def _runCleanups(self):
734
 
        """Run registered cleanup functions. 
735
 
 
736
 
        This should only be called from TestCase.tearDown.
737
 
        """
738
 
        # TODO: Perhaps this should keep running cleanups even if 
739
 
        # one of them fails?
740
 
        for cleanup_fn in reversed(self._cleanups):
741
 
            cleanup_fn()
742
 
 
743
1647
    def log(self, *args):
744
1648
        mutter(*args)
745
1649
 
746
 
    def _get_log(self):
747
 
        """Return as a string the log for this test"""
748
 
        if self._log_file_name:
749
 
            return open(self._log_file_name).read()
750
 
        else:
 
1650
    def _get_log(self, keep_log_file=False):
 
1651
        """Internal helper to get the log from bzrlib.trace for this test.
 
1652
 
 
1653
        Please use self.getDetails, or self.get_log to access this in test case
 
1654
        code.
 
1655
 
 
1656
        :param keep_log_file: When True, if the log is still a file on disk
 
1657
            leave it as a file on disk. When False, if the log is still a file
 
1658
            on disk, the log file is deleted and the log preserved as
 
1659
            self._log_contents.
 
1660
        :return: A string containing the log.
 
1661
        """
 
1662
        if self._log_contents is not None:
 
1663
            try:
 
1664
                self._log_contents.decode('utf8')
 
1665
            except UnicodeDecodeError:
 
1666
                unicodestr = self._log_contents.decode('utf8', 'replace')
 
1667
                self._log_contents = unicodestr.encode('utf8')
751
1668
            return self._log_contents
752
 
        # TODO: Delete the log after it's been read in
753
 
 
754
 
    def capture(self, cmd, retcode=0):
755
 
        """Shortcut that splits cmd into words, runs, and returns stdout"""
756
 
        return self.run_bzr_captured(cmd.split(), retcode=retcode)[0]
757
 
 
758
 
    def run_bzr_captured(self, argv, retcode=0, encoding=None, stdin=None):
759
 
        """Invoke bzr and return (stdout, stderr).
760
 
 
761
 
        Useful for code that wants to check the contents of the
762
 
        output, the way error messages are presented, etc.
763
 
 
764
 
        This should be the main method for tests that want to exercise the
765
 
        overall behavior of the bzr application (rather than a unit test
766
 
        or a functional test of the library.)
767
 
 
768
 
        Much of the old code runs bzr by forking a new copy of Python, but
769
 
        that is slower, harder to debug, and generally not necessary.
770
 
 
771
 
        This runs bzr through the interface that catches and reports
772
 
        errors, and with logging set to something approximating the
773
 
        default, so that error reporting can be checked.
774
 
 
775
 
        :param argv: arguments to invoke bzr
776
 
        :param retcode: expected return code, or None for don't-care.
777
 
        :param encoding: encoding for sys.stdout and sys.stderr
778
 
        :param stdin: A string to be used as stdin for the command.
779
 
        """
 
1669
        import bzrlib.trace
 
1670
        if bzrlib.trace._trace_file:
 
1671
            # flush the log file, to get all content
 
1672
            bzrlib.trace._trace_file.flush()
 
1673
        if self._log_file_name is not None:
 
1674
            logfile = open(self._log_file_name)
 
1675
            try:
 
1676
                log_contents = logfile.read()
 
1677
            finally:
 
1678
                logfile.close()
 
1679
            try:
 
1680
                log_contents.decode('utf8')
 
1681
            except UnicodeDecodeError:
 
1682
                unicodestr = log_contents.decode('utf8', 'replace')
 
1683
                log_contents = unicodestr.encode('utf8')
 
1684
            if not keep_log_file:
 
1685
                close_attempts = 0
 
1686
                max_close_attempts = 100
 
1687
                first_close_error = None
 
1688
                while close_attempts < max_close_attempts:
 
1689
                    close_attempts += 1
 
1690
                    try:
 
1691
                        self._log_file.close()
 
1692
                    except IOError, ioe:
 
1693
                        if ioe.errno is None:
 
1694
                            # No errno implies 'close() called during
 
1695
                            # concurrent operation on the same file object', so
 
1696
                            # retry.  Probably a thread is trying to write to
 
1697
                            # the log file.
 
1698
                            if first_close_error is None:
 
1699
                                first_close_error = ioe
 
1700
                            continue
 
1701
                        raise
 
1702
                    else:
 
1703
                        break
 
1704
                if close_attempts > 1:
 
1705
                    sys.stderr.write(
 
1706
                        'Unable to close log file on first attempt, '
 
1707
                        'will retry: %s\n' % (first_close_error,))
 
1708
                    if close_attempts == max_close_attempts:
 
1709
                        sys.stderr.write(
 
1710
                            'Unable to close log file after %d attempts.\n'
 
1711
                            % (max_close_attempts,))
 
1712
                self._log_file = None
 
1713
                # Permit multiple calls to get_log until we clean it up in
 
1714
                # finishLogFile
 
1715
                self._log_contents = log_contents
 
1716
                try:
 
1717
                    os.remove(self._log_file_name)
 
1718
                except OSError, e:
 
1719
                    if sys.platform == 'win32' and e.errno == errno.EACCES:
 
1720
                        sys.stderr.write(('Unable to delete log file '
 
1721
                                             ' %r\n' % self._log_file_name))
 
1722
                    else:
 
1723
                        raise
 
1724
                self._log_file_name = None
 
1725
            return log_contents
 
1726
        else:
 
1727
            return "No log file content and no log file name."
 
1728
 
 
1729
    def get_log(self):
 
1730
        """Get a unicode string containing the log from bzrlib.trace.
 
1731
 
 
1732
        Undecodable characters are replaced.
 
1733
        """
 
1734
        return u"".join(self.getDetails()['log'].iter_text())
 
1735
 
 
1736
    def requireFeature(self, feature):
 
1737
        """This test requires a specific feature is available.
 
1738
 
 
1739
        :raises UnavailableFeature: When feature is not available.
 
1740
        """
 
1741
        if not feature.available():
 
1742
            raise UnavailableFeature(feature)
 
1743
 
 
1744
    def _run_bzr_autosplit(self, args, retcode, encoding, stdin,
 
1745
            working_dir):
 
1746
        """Run bazaar command line, splitting up a string command line."""
 
1747
        if isinstance(args, basestring):
 
1748
            # shlex don't understand unicode strings,
 
1749
            # so args should be plain string (bialix 20070906)
 
1750
            args = list(shlex.split(str(args)))
 
1751
        return self._run_bzr_core(args, retcode=retcode,
 
1752
                encoding=encoding, stdin=stdin, working_dir=working_dir,
 
1753
                )
 
1754
 
 
1755
    def _run_bzr_core(self, args, retcode, encoding, stdin,
 
1756
            working_dir):
 
1757
        # Clear chk_map page cache, because the contents are likely to mask
 
1758
        # locking errors.
 
1759
        chk_map.clear_cache()
780
1760
        if encoding is None:
781
 
            encoding = bzrlib.user_encoding
782
 
        if stdin is not None:
783
 
            stdin = StringIO(stdin)
 
1761
            encoding = osutils.get_user_encoding()
784
1762
        stdout = StringIOWrapper()
785
1763
        stderr = StringIOWrapper()
786
1764
        stdout.encoding = encoding
787
1765
        stderr.encoding = encoding
788
1766
 
789
 
        self.log('run bzr: %r', argv)
 
1767
        self.log('run bzr: %r', args)
790
1768
        # FIXME: don't call into logging here
791
1769
        handler = logging.StreamHandler(stderr)
792
1770
        handler.setLevel(logging.INFO)
793
1771
        logger = logging.getLogger('')
794
1772
        logger.addHandler(handler)
795
 
        old_ui_factory = bzrlib.ui.ui_factory
796
 
        bzrlib.ui.ui_factory = bzrlib.tests.blackbox.TestUIFactory(
797
 
            stdout=stdout,
798
 
            stderr=stderr)
799
 
        bzrlib.ui.ui_factory.stdin = stdin
 
1773
        old_ui_factory = ui.ui_factory
 
1774
        ui.ui_factory = TestUIFactory(stdin=stdin, stdout=stdout, stderr=stderr)
 
1775
 
 
1776
        cwd = None
 
1777
        if working_dir is not None:
 
1778
            cwd = osutils.getcwd()
 
1779
            os.chdir(working_dir)
 
1780
 
800
1781
        try:
801
 
            result = self.apply_redirected(stdin, stdout, stderr,
802
 
                                           bzrlib.commands.run_bzr_catch_errors,
803
 
                                           argv)
 
1782
            try:
 
1783
                result = self.apply_redirected(ui.ui_factory.stdin,
 
1784
                    stdout, stderr,
 
1785
                    bzrlib.commands.run_bzr_catch_user_errors,
 
1786
                    args)
 
1787
            except KeyboardInterrupt:
 
1788
                # Reraise KeyboardInterrupt with contents of redirected stdout
 
1789
                # and stderr as arguments, for tests which are interested in
 
1790
                # stdout and stderr and are expecting the exception.
 
1791
                out = stdout.getvalue()
 
1792
                err = stderr.getvalue()
 
1793
                if out:
 
1794
                    self.log('output:\n%r', out)
 
1795
                if err:
 
1796
                    self.log('errors:\n%r', err)
 
1797
                raise KeyboardInterrupt(out, err)
804
1798
        finally:
805
1799
            logger.removeHandler(handler)
806
 
            bzrlib.ui.ui_factory = old_ui_factory
 
1800
            ui.ui_factory = old_ui_factory
 
1801
            if cwd is not None:
 
1802
                os.chdir(cwd)
807
1803
 
808
1804
        out = stdout.getvalue()
809
1805
        err = stderr.getvalue()
812
1808
        if err:
813
1809
            self.log('errors:\n%r', err)
814
1810
        if retcode is not None:
815
 
            self.assertEquals(retcode, result)
816
 
        return out, err
 
1811
            self.assertEquals(retcode, result,
 
1812
                              message='Unexpected return code')
 
1813
        return result, out, err
817
1814
 
818
 
    def run_bzr(self, *args, **kwargs):
 
1815
    def run_bzr(self, args, retcode=0, encoding=None, stdin=None,
 
1816
                working_dir=None, error_regexes=[], output_encoding=None):
819
1817
        """Invoke bzr, as if it were run from the command line.
820
1818
 
 
1819
        The argument list should not include the bzr program name - the
 
1820
        first argument is normally the bzr command.  Arguments may be
 
1821
        passed in three ways:
 
1822
 
 
1823
        1- A list of strings, eg ["commit", "a"].  This is recommended
 
1824
        when the command contains whitespace or metacharacters, or
 
1825
        is built up at run time.
 
1826
 
 
1827
        2- A single string, eg "add a".  This is the most convenient
 
1828
        for hardcoded commands.
 
1829
 
 
1830
        This runs bzr through the interface that catches and reports
 
1831
        errors, and with logging set to something approximating the
 
1832
        default, so that error reporting can be checked.
 
1833
 
821
1834
        This should be the main method for tests that want to exercise the
822
1835
        overall behavior of the bzr application (rather than a unit test
823
1836
        or a functional test of the library.)
825
1838
        This sends the stdout/stderr results into the test's log,
826
1839
        where it may be useful for debugging.  See also run_captured.
827
1840
 
828
 
        :param stdin: A string to be used as stdin for the command.
 
1841
        :keyword stdin: A string to be used as stdin for the command.
 
1842
        :keyword retcode: The status code the command should return;
 
1843
            default 0.
 
1844
        :keyword working_dir: The directory to run the command in
 
1845
        :keyword error_regexes: A list of expected error messages.  If
 
1846
            specified they must be seen in the error output of the command.
829
1847
        """
830
 
        retcode = kwargs.pop('retcode', 0)
831
 
        encoding = kwargs.pop('encoding', None)
832
 
        stdin = kwargs.pop('stdin', None)
833
 
        return self.run_bzr_captured(args, retcode=retcode, encoding=encoding, stdin=stdin)
834
 
 
835
 
    def run_bzr_decode(self, *args, **kwargs):
836
 
        if 'encoding' in kwargs:
837
 
            encoding = kwargs['encoding']
838
 
        else:
839
 
            encoding = bzrlib.user_encoding
840
 
        return self.run_bzr(*args, **kwargs)[0].decode(encoding)
 
1848
        retcode, out, err = self._run_bzr_autosplit(
 
1849
            args=args,
 
1850
            retcode=retcode,
 
1851
            encoding=encoding,
 
1852
            stdin=stdin,
 
1853
            working_dir=working_dir,
 
1854
            )
 
1855
        self.assertIsInstance(error_regexes, (list, tuple))
 
1856
        for regex in error_regexes:
 
1857
            self.assertContainsRe(err, regex)
 
1858
        return out, err
841
1859
 
842
1860
    def run_bzr_error(self, error_regexes, *args, **kwargs):
843
1861
        """Run bzr, and check that stderr contains the supplied regexes
844
 
        
845
 
        :param error_regexes: Sequence of regular expressions which 
 
1862
 
 
1863
        :param error_regexes: Sequence of regular expressions which
846
1864
            must each be found in the error output. The relative ordering
847
1865
            is not enforced.
848
1866
        :param args: command-line arguments for bzr
849
1867
        :param kwargs: Keyword arguments which are interpreted by run_bzr
850
1868
            This function changes the default value of retcode to be 3,
851
1869
            since in most cases this is run when you expect bzr to fail.
852
 
        :return: (out, err) The actual output of running the command (in case you
853
 
                 want to do more inspection)
854
 
 
855
 
        Examples of use:
 
1870
 
 
1871
        :return: (out, err) The actual output of running the command (in case
 
1872
            you want to do more inspection)
 
1873
 
 
1874
        Examples of use::
 
1875
 
856
1876
            # Make sure that commit is failing because there is nothing to do
857
1877
            self.run_bzr_error(['no changes to commit'],
858
 
                               'commit', '-m', 'my commit comment')
 
1878
                               ['commit', '-m', 'my commit comment'])
859
1879
            # Make sure --strict is handling an unknown file, rather than
860
1880
            # giving us the 'nothing to do' error
861
1881
            self.build_tree(['unknown'])
862
1882
            self.run_bzr_error(['Commit refused because there are unknown files'],
863
 
                               'commit', '--strict', '-m', 'my commit comment')
 
1883
                               ['commit', --strict', '-m', 'my commit comment'])
864
1884
        """
865
1885
        kwargs.setdefault('retcode', 3)
 
1886
        kwargs['error_regexes'] = error_regexes
866
1887
        out, err = self.run_bzr(*args, **kwargs)
867
 
        for regex in error_regexes:
868
 
            self.assertContainsRe(err, regex)
869
1888
        return out, err
870
1889
 
871
1890
    def run_bzr_subprocess(self, *args, **kwargs):
872
1891
        """Run bzr in a subprocess for testing.
873
1892
 
874
 
        This starts a new Python interpreter and runs bzr in there. 
 
1893
        This starts a new Python interpreter and runs bzr in there.
875
1894
        This should only be used for tests that have a justifiable need for
876
1895
        this isolation: e.g. they are testing startup time, or signal
877
 
        handling, or early startup code, etc.  Subprocess code can't be 
 
1896
        handling, or early startup code, etc.  Subprocess code can't be
878
1897
        profiled or debugged so easily.
879
1898
 
880
 
        :param retcode: The status code that is expected.  Defaults to 0.  If
 
1899
        :keyword retcode: The status code that is expected.  Defaults to 0.  If
881
1900
            None is supplied, the status code is not checked.
 
1901
        :keyword env_changes: A dictionary which lists changes to environment
 
1902
            variables. A value of None will unset the env variable.
 
1903
            The values must be strings. The change will only occur in the
 
1904
            child, so you don't need to fix the environment after running.
 
1905
        :keyword universal_newlines: Convert CRLF => LF
 
1906
        :keyword allow_plugins: By default the subprocess is run with
 
1907
            --no-plugins to ensure test reproducibility. Also, it is possible
 
1908
            for system-wide plugins to create unexpected output on stderr,
 
1909
            which can cause unnecessary test failures.
 
1910
        """
 
1911
        env_changes = kwargs.get('env_changes', {})
 
1912
        working_dir = kwargs.get('working_dir', None)
 
1913
        allow_plugins = kwargs.get('allow_plugins', False)
 
1914
        if len(args) == 1:
 
1915
            if isinstance(args[0], list):
 
1916
                args = args[0]
 
1917
            elif isinstance(args[0], basestring):
 
1918
                args = list(shlex.split(args[0]))
 
1919
        else:
 
1920
            raise ValueError("passing varargs to run_bzr_subprocess")
 
1921
        process = self.start_bzr_subprocess(args, env_changes=env_changes,
 
1922
                                            working_dir=working_dir,
 
1923
                                            allow_plugins=allow_plugins)
 
1924
        # We distinguish between retcode=None and retcode not passed.
 
1925
        supplied_retcode = kwargs.get('retcode', 0)
 
1926
        return self.finish_bzr_subprocess(process, retcode=supplied_retcode,
 
1927
            universal_newlines=kwargs.get('universal_newlines', False),
 
1928
            process_args=args)
 
1929
 
 
1930
    def start_bzr_subprocess(self, process_args, env_changes=None,
 
1931
                             skip_if_plan_to_signal=False,
 
1932
                             working_dir=None,
 
1933
                             allow_plugins=False):
 
1934
        """Start bzr in a subprocess for testing.
 
1935
 
 
1936
        This starts a new Python interpreter and runs bzr in there.
 
1937
        This should only be used for tests that have a justifiable need for
 
1938
        this isolation: e.g. they are testing startup time, or signal
 
1939
        handling, or early startup code, etc.  Subprocess code can't be
 
1940
        profiled or debugged so easily.
 
1941
 
 
1942
        :param process_args: a list of arguments to pass to the bzr executable,
 
1943
            for example ``['--version']``.
882
1944
        :param env_changes: A dictionary which lists changes to environment
883
1945
            variables. A value of None will unset the env variable.
884
1946
            The values must be strings. The change will only occur in the
885
1947
            child, so you don't need to fix the environment after running.
886
 
        :param universal_newlines: Convert CRLF => LF
 
1948
        :param skip_if_plan_to_signal: raise TestSkipped when true and os.kill
 
1949
            is not available.
 
1950
        :param allow_plugins: If False (default) pass --no-plugins to bzr.
 
1951
 
 
1952
        :returns: Popen object for the started process.
887
1953
        """
888
 
        env_changes = kwargs.get('env_changes', {})
 
1954
        if skip_if_plan_to_signal:
 
1955
            if not getattr(os, 'kill', None):
 
1956
                raise TestSkipped("os.kill not available.")
889
1957
 
 
1958
        if env_changes is None:
 
1959
            env_changes = {}
890
1960
        old_env = {}
891
1961
 
892
1962
        def cleanup_environment():
897
1967
            for env_var, value in old_env.iteritems():
898
1968
                osutils.set_or_unset_env(env_var, value)
899
1969
 
900
 
        bzr_path = os.path.dirname(os.path.dirname(bzrlib.__file__))+'/bzr'
901
 
        args = list(args)
 
1970
        bzr_path = self.get_bzr_path()
 
1971
 
 
1972
        cwd = None
 
1973
        if working_dir is not None:
 
1974
            cwd = osutils.getcwd()
 
1975
            os.chdir(working_dir)
902
1976
 
903
1977
        try:
904
1978
            # win32 subprocess doesn't support preexec_fn
905
1979
            # so we will avoid using it on all platforms, just to
906
1980
            # make sure the code path is used, and we don't break on win32
907
1981
            cleanup_environment()
908
 
            process = Popen([sys.executable, bzr_path]+args,
909
 
                             stdout=PIPE, stderr=PIPE)
 
1982
            command = [sys.executable]
 
1983
            # frozen executables don't need the path to bzr
 
1984
            if getattr(sys, "frozen", None) is None:
 
1985
                command.append(bzr_path)
 
1986
            if not allow_plugins:
 
1987
                command.append('--no-plugins')
 
1988
            command.extend(process_args)
 
1989
            process = self._popen(command, stdin=PIPE, stdout=PIPE, stderr=PIPE)
910
1990
        finally:
911
1991
            restore_environment()
912
 
            
913
 
        out = process.stdout.read()
914
 
        err = process.stderr.read()
915
 
 
916
 
        if kwargs.get('universal_newlines', False):
 
1992
            if cwd is not None:
 
1993
                os.chdir(cwd)
 
1994
 
 
1995
        return process
 
1996
 
 
1997
    def _popen(self, *args, **kwargs):
 
1998
        """Place a call to Popen.
 
1999
 
 
2000
        Allows tests to override this method to intercept the calls made to
 
2001
        Popen for introspection.
 
2002
        """
 
2003
        return Popen(*args, **kwargs)
 
2004
 
 
2005
    def get_source_path(self):
 
2006
        """Return the path of the directory containing bzrlib."""
 
2007
        return os.path.dirname(os.path.dirname(bzrlib.__file__))
 
2008
 
 
2009
    def get_bzr_path(self):
 
2010
        """Return the path of the 'bzr' executable for this test suite."""
 
2011
        bzr_path = self.get_source_path()+'/bzr'
 
2012
        if not os.path.isfile(bzr_path):
 
2013
            # We are probably installed. Assume sys.argv is the right file
 
2014
            bzr_path = sys.argv[0]
 
2015
        return bzr_path
 
2016
 
 
2017
    def finish_bzr_subprocess(self, process, retcode=0, send_signal=None,
 
2018
                              universal_newlines=False, process_args=None):
 
2019
        """Finish the execution of process.
 
2020
 
 
2021
        :param process: the Popen object returned from start_bzr_subprocess.
 
2022
        :param retcode: The status code that is expected.  Defaults to 0.  If
 
2023
            None is supplied, the status code is not checked.
 
2024
        :param send_signal: an optional signal to send to the process.
 
2025
        :param universal_newlines: Convert CRLF => LF
 
2026
        :returns: (stdout, stderr)
 
2027
        """
 
2028
        if send_signal is not None:
 
2029
            os.kill(process.pid, send_signal)
 
2030
        out, err = process.communicate()
 
2031
 
 
2032
        if universal_newlines:
917
2033
            out = out.replace('\r\n', '\n')
918
2034
            err = err.replace('\r\n', '\n')
919
2035
 
920
 
        retcode = process.wait()
921
 
        supplied_retcode = kwargs.get('retcode', 0)
922
 
        if supplied_retcode is not None:
923
 
            assert supplied_retcode == retcode
 
2036
        if retcode is not None and retcode != process.returncode:
 
2037
            if process_args is None:
 
2038
                process_args = "(unknown args)"
 
2039
            mutter('Output of bzr %s:\n%s', process_args, out)
 
2040
            mutter('Error for bzr %s:\n%s', process_args, err)
 
2041
            self.fail('Command bzr %s failed with retcode %s != %s'
 
2042
                      % (process_args, retcode, process.returncode))
924
2043
        return [out, err]
925
2044
 
926
2045
    def check_inventory_shape(self, inv, shape):
932
2051
        shape = list(shape)             # copy
933
2052
        for path, ie in inv.entries():
934
2053
            name = path.replace('\\', '/')
935
 
            if ie.kind == 'dir':
 
2054
            if ie.kind == 'directory':
936
2055
                name = name + '/'
937
2056
            if name in shape:
938
2057
                shape.remove(name)
975
2094
            sys.stderr = real_stderr
976
2095
            sys.stdin = real_stdin
977
2096
 
978
 
    @symbol_versioning.deprecated_method(symbol_versioning.zero_eleven)
979
 
    def merge(self, branch_from, wt_to):
980
 
        """A helper for tests to do a ui-less merge.
981
 
 
982
 
        This should move to the main library when someone has time to integrate
983
 
        it in.
984
 
        """
985
 
        # minimal ui-less merge.
986
 
        wt_to.branch.fetch(branch_from)
987
 
        base_rev = common_ancestor(branch_from.last_revision(),
988
 
                                   wt_to.branch.last_revision(),
989
 
                                   wt_to.branch.repository)
990
 
        merge_inner(wt_to.branch, branch_from.basis_tree(),
991
 
                    wt_to.branch.repository.revision_tree(base_rev),
992
 
                    this_tree=wt_to)
993
 
        wt_to.add_parent_tree_id(branch_from.last_revision())
994
 
 
995
 
 
996
 
BzrTestBase = TestCase
997
 
 
998
 
     
999
 
class TestCaseInTempDir(TestCase):
 
2097
    def reduceLockdirTimeout(self):
 
2098
        """Reduce the default lock timeout for the duration of the test, so that
 
2099
        if LockContention occurs during a test, it does so quickly.
 
2100
 
 
2101
        Tests that expect to provoke LockContention errors should call this.
 
2102
        """
 
2103
        self.overrideAttr(bzrlib.lockdir, '_DEFAULT_TIMEOUT_SECONDS', 0)
 
2104
 
 
2105
    def make_utf8_encoded_stringio(self, encoding_type=None):
 
2106
        """Return a StringIOWrapper instance, that will encode Unicode
 
2107
        input to UTF-8.
 
2108
        """
 
2109
        if encoding_type is None:
 
2110
            encoding_type = 'strict'
 
2111
        sio = StringIO()
 
2112
        output_encoding = 'utf-8'
 
2113
        sio = codecs.getwriter(output_encoding)(sio, errors=encoding_type)
 
2114
        sio.encoding = output_encoding
 
2115
        return sio
 
2116
 
 
2117
    def disable_verb(self, verb):
 
2118
        """Disable a smart server verb for one test."""
 
2119
        from bzrlib.smart import request
 
2120
        request_handlers = request.request_handlers
 
2121
        orig_method = request_handlers.get(verb)
 
2122
        request_handlers.remove(verb)
 
2123
        self.addCleanup(request_handlers.register, verb, orig_method)
 
2124
 
 
2125
 
 
2126
class CapturedCall(object):
 
2127
    """A helper for capturing smart server calls for easy debug analysis."""
 
2128
 
 
2129
    def __init__(self, params, prefix_length):
 
2130
        """Capture the call with params and skip prefix_length stack frames."""
 
2131
        self.call = params
 
2132
        import traceback
 
2133
        # The last 5 frames are the __init__, the hook frame, and 3 smart
 
2134
        # client frames. Beyond this we could get more clever, but this is good
 
2135
        # enough for now.
 
2136
        stack = traceback.extract_stack()[prefix_length:-5]
 
2137
        self.stack = ''.join(traceback.format_list(stack))
 
2138
 
 
2139
    def __str__(self):
 
2140
        return self.call.method
 
2141
 
 
2142
    def __repr__(self):
 
2143
        return self.call.method
 
2144
 
 
2145
    def stack(self):
 
2146
        return self.stack
 
2147
 
 
2148
 
 
2149
class TestCaseWithMemoryTransport(TestCase):
 
2150
    """Common test class for tests that do not need disk resources.
 
2151
 
 
2152
    Tests that need disk resources should derive from TestCaseWithTransport.
 
2153
 
 
2154
    TestCaseWithMemoryTransport sets the TEST_ROOT variable for all bzr tests.
 
2155
 
 
2156
    For TestCaseWithMemoryTransport the test_home_dir is set to the name of
 
2157
    a directory which does not exist. This serves to help ensure test isolation
 
2158
    is preserved. test_dir is set to the TEST_ROOT, as is cwd, because they
 
2159
    must exist. However, TestCaseWithMemoryTransport does not offer local
 
2160
    file defaults for the transport in tests, nor does it obey the command line
 
2161
    override, so tests that accidentally write to the common directory should
 
2162
    be rare.
 
2163
 
 
2164
    :cvar TEST_ROOT: Directory containing all temporary directories, plus
 
2165
    a .bzr directory that stops us ascending higher into the filesystem.
 
2166
    """
 
2167
 
 
2168
    TEST_ROOT = None
 
2169
    _TEST_NAME = 'test'
 
2170
 
 
2171
    def __init__(self, methodName='runTest'):
 
2172
        # allow test parameterization after test construction and before test
 
2173
        # execution. Variables that the parameterizer sets need to be
 
2174
        # ones that are not set by setUp, or setUp will trash them.
 
2175
        super(TestCaseWithMemoryTransport, self).__init__(methodName)
 
2176
        self.vfs_transport_factory = default_transport
 
2177
        self.transport_server = None
 
2178
        self.transport_readonly_server = None
 
2179
        self.__vfs_server = None
 
2180
 
 
2181
    def get_transport(self, relpath=None):
 
2182
        """Return a writeable transport.
 
2183
 
 
2184
        This transport is for the test scratch space relative to
 
2185
        "self._test_root"
 
2186
 
 
2187
        :param relpath: a path relative to the base url.
 
2188
        """
 
2189
        t = get_transport(self.get_url(relpath))
 
2190
        self.assertFalse(t.is_readonly())
 
2191
        return t
 
2192
 
 
2193
    def get_readonly_transport(self, relpath=None):
 
2194
        """Return a readonly transport for the test scratch space
 
2195
 
 
2196
        This can be used to test that operations which should only need
 
2197
        readonly access in fact do not try to write.
 
2198
 
 
2199
        :param relpath: a path relative to the base url.
 
2200
        """
 
2201
        t = get_transport(self.get_readonly_url(relpath))
 
2202
        self.assertTrue(t.is_readonly())
 
2203
        return t
 
2204
 
 
2205
    def create_transport_readonly_server(self):
 
2206
        """Create a transport server from class defined at init.
 
2207
 
 
2208
        This is mostly a hook for daughter classes.
 
2209
        """
 
2210
        return self.transport_readonly_server()
 
2211
 
 
2212
    def get_readonly_server(self):
 
2213
        """Get the server instance for the readonly transport
 
2214
 
 
2215
        This is useful for some tests with specific servers to do diagnostics.
 
2216
        """
 
2217
        if self.__readonly_server is None:
 
2218
            if self.transport_readonly_server is None:
 
2219
                # readonly decorator requested
 
2220
                self.__readonly_server = test_server.ReadonlyServer()
 
2221
            else:
 
2222
                # explicit readonly transport.
 
2223
                self.__readonly_server = self.create_transport_readonly_server()
 
2224
            self.start_server(self.__readonly_server,
 
2225
                self.get_vfs_only_server())
 
2226
        return self.__readonly_server
 
2227
 
 
2228
    def get_readonly_url(self, relpath=None):
 
2229
        """Get a URL for the readonly transport.
 
2230
 
 
2231
        This will either be backed by '.' or a decorator to the transport
 
2232
        used by self.get_url()
 
2233
        relpath provides for clients to get a path relative to the base url.
 
2234
        These should only be downwards relative, not upwards.
 
2235
        """
 
2236
        base = self.get_readonly_server().get_url()
 
2237
        return self._adjust_url(base, relpath)
 
2238
 
 
2239
    def get_vfs_only_server(self):
 
2240
        """Get the vfs only read/write server instance.
 
2241
 
 
2242
        This is useful for some tests with specific servers that need
 
2243
        diagnostics.
 
2244
 
 
2245
        For TestCaseWithMemoryTransport this is always a MemoryServer, and there
 
2246
        is no means to override it.
 
2247
        """
 
2248
        if self.__vfs_server is None:
 
2249
            self.__vfs_server = memory.MemoryServer()
 
2250
            self.start_server(self.__vfs_server)
 
2251
        return self.__vfs_server
 
2252
 
 
2253
    def get_server(self):
 
2254
        """Get the read/write server instance.
 
2255
 
 
2256
        This is useful for some tests with specific servers that need
 
2257
        diagnostics.
 
2258
 
 
2259
        This is built from the self.transport_server factory. If that is None,
 
2260
        then the self.get_vfs_server is returned.
 
2261
        """
 
2262
        if self.__server is None:
 
2263
            if (self.transport_server is None or self.transport_server is
 
2264
                self.vfs_transport_factory):
 
2265
                self.__server = self.get_vfs_only_server()
 
2266
            else:
 
2267
                # bring up a decorated means of access to the vfs only server.
 
2268
                self.__server = self.transport_server()
 
2269
                self.start_server(self.__server, self.get_vfs_only_server())
 
2270
        return self.__server
 
2271
 
 
2272
    def _adjust_url(self, base, relpath):
 
2273
        """Get a URL (or maybe a path) for the readwrite transport.
 
2274
 
 
2275
        This will either be backed by '.' or to an equivalent non-file based
 
2276
        facility.
 
2277
        relpath provides for clients to get a path relative to the base url.
 
2278
        These should only be downwards relative, not upwards.
 
2279
        """
 
2280
        if relpath is not None and relpath != '.':
 
2281
            if not base.endswith('/'):
 
2282
                base = base + '/'
 
2283
            # XXX: Really base should be a url; we did after all call
 
2284
            # get_url()!  But sometimes it's just a path (from
 
2285
            # LocalAbspathServer), and it'd be wrong to append urlescaped data
 
2286
            # to a non-escaped local path.
 
2287
            if base.startswith('./') or base.startswith('/'):
 
2288
                base += relpath
 
2289
            else:
 
2290
                base += urlutils.escape(relpath)
 
2291
        return base
 
2292
 
 
2293
    def get_url(self, relpath=None):
 
2294
        """Get a URL (or maybe a path) for the readwrite transport.
 
2295
 
 
2296
        This will either be backed by '.' or to an equivalent non-file based
 
2297
        facility.
 
2298
        relpath provides for clients to get a path relative to the base url.
 
2299
        These should only be downwards relative, not upwards.
 
2300
        """
 
2301
        base = self.get_server().get_url()
 
2302
        return self._adjust_url(base, relpath)
 
2303
 
 
2304
    def get_vfs_only_url(self, relpath=None):
 
2305
        """Get a URL (or maybe a path for the plain old vfs transport.
 
2306
 
 
2307
        This will never be a smart protocol.  It always has all the
 
2308
        capabilities of the local filesystem, but it might actually be a
 
2309
        MemoryTransport or some other similar virtual filesystem.
 
2310
 
 
2311
        This is the backing transport (if any) of the server returned by
 
2312
        get_url and get_readonly_url.
 
2313
 
 
2314
        :param relpath: provides for clients to get a path relative to the base
 
2315
            url.  These should only be downwards relative, not upwards.
 
2316
        :return: A URL
 
2317
        """
 
2318
        base = self.get_vfs_only_server().get_url()
 
2319
        return self._adjust_url(base, relpath)
 
2320
 
 
2321
    def _create_safety_net(self):
 
2322
        """Make a fake bzr directory.
 
2323
 
 
2324
        This prevents any tests propagating up onto the TEST_ROOT directory's
 
2325
        real branch.
 
2326
        """
 
2327
        root = TestCaseWithMemoryTransport.TEST_ROOT
 
2328
        bzrdir.BzrDir.create_standalone_workingtree(root)
 
2329
 
 
2330
    def _check_safety_net(self):
 
2331
        """Check that the safety .bzr directory have not been touched.
 
2332
 
 
2333
        _make_test_root have created a .bzr directory to prevent tests from
 
2334
        propagating. This method ensures than a test did not leaked.
 
2335
        """
 
2336
        root = TestCaseWithMemoryTransport.TEST_ROOT
 
2337
        self.permit_url(get_transport(root).base)
 
2338
        wt = workingtree.WorkingTree.open(root)
 
2339
        last_rev = wt.last_revision()
 
2340
        if last_rev != 'null:':
 
2341
            # The current test have modified the /bzr directory, we need to
 
2342
            # recreate a new one or all the followng tests will fail.
 
2343
            # If you need to inspect its content uncomment the following line
 
2344
            # import pdb; pdb.set_trace()
 
2345
            _rmtree_temp_dir(root + '/.bzr', test_id=self.id())
 
2346
            self._create_safety_net()
 
2347
            raise AssertionError('%s/.bzr should not be modified' % root)
 
2348
 
 
2349
    def _make_test_root(self):
 
2350
        if TestCaseWithMemoryTransport.TEST_ROOT is None:
 
2351
            # Watch out for tricky test dir (on OSX /tmp -> /private/tmp)
 
2352
            root = osutils.realpath(osutils.mkdtemp(prefix='testbzr-',
 
2353
                                                    suffix='.tmp'))
 
2354
            TestCaseWithMemoryTransport.TEST_ROOT = root
 
2355
 
 
2356
            self._create_safety_net()
 
2357
 
 
2358
            # The same directory is used by all tests, and we're not
 
2359
            # specifically told when all tests are finished.  This will do.
 
2360
            atexit.register(_rmtree_temp_dir, root)
 
2361
 
 
2362
        self.permit_dir(TestCaseWithMemoryTransport.TEST_ROOT)
 
2363
        self.addCleanup(self._check_safety_net)
 
2364
 
 
2365
    def makeAndChdirToTestDir(self):
 
2366
        """Create a temporary directories for this one test.
 
2367
 
 
2368
        This must set self.test_home_dir and self.test_dir and chdir to
 
2369
        self.test_dir.
 
2370
 
 
2371
        For TestCaseWithMemoryTransport we chdir to the TEST_ROOT for this test.
 
2372
        """
 
2373
        os.chdir(TestCaseWithMemoryTransport.TEST_ROOT)
 
2374
        self.test_dir = TestCaseWithMemoryTransport.TEST_ROOT
 
2375
        self.test_home_dir = self.test_dir + "/MemoryTransportMissingHomeDir"
 
2376
        self.permit_dir(self.test_dir)
 
2377
 
 
2378
    def make_branch(self, relpath, format=None):
 
2379
        """Create a branch on the transport at relpath."""
 
2380
        repo = self.make_repository(relpath, format=format)
 
2381
        return repo.bzrdir.create_branch()
 
2382
 
 
2383
    def make_bzrdir(self, relpath, format=None):
 
2384
        try:
 
2385
            # might be a relative or absolute path
 
2386
            maybe_a_url = self.get_url(relpath)
 
2387
            segments = maybe_a_url.rsplit('/', 1)
 
2388
            t = get_transport(maybe_a_url)
 
2389
            if len(segments) > 1 and segments[-1] not in ('', '.'):
 
2390
                t.ensure_base()
 
2391
            if format is None:
 
2392
                format = 'default'
 
2393
            if isinstance(format, basestring):
 
2394
                format = bzrdir.format_registry.make_bzrdir(format)
 
2395
            return format.initialize_on_transport(t)
 
2396
        except errors.UninitializableFormat:
 
2397
            raise TestSkipped("Format %s is not initializable." % format)
 
2398
 
 
2399
    def make_repository(self, relpath, shared=False, format=None):
 
2400
        """Create a repository on our default transport at relpath.
 
2401
 
 
2402
        Note that relpath must be a relative path, not a full url.
 
2403
        """
 
2404
        # FIXME: If you create a remoterepository this returns the underlying
 
2405
        # real format, which is incorrect.  Actually we should make sure that
 
2406
        # RemoteBzrDir returns a RemoteRepository.
 
2407
        # maybe  mbp 20070410
 
2408
        made_control = self.make_bzrdir(relpath, format=format)
 
2409
        return made_control.create_repository(shared=shared)
 
2410
 
 
2411
    def make_smart_server(self, path):
 
2412
        smart_server = test_server.SmartTCPServer_for_testing()
 
2413
        self.start_server(smart_server, self.get_server())
 
2414
        remote_transport = get_transport(smart_server.get_url()).clone(path)
 
2415
        return remote_transport
 
2416
 
 
2417
    def make_branch_and_memory_tree(self, relpath, format=None):
 
2418
        """Create a branch on the default transport and a MemoryTree for it."""
 
2419
        b = self.make_branch(relpath, format=format)
 
2420
        return memorytree.MemoryTree.create_on_branch(b)
 
2421
 
 
2422
    def make_branch_builder(self, relpath, format=None):
 
2423
        branch = self.make_branch(relpath, format=format)
 
2424
        return branchbuilder.BranchBuilder(branch=branch)
 
2425
 
 
2426
    def overrideEnvironmentForTesting(self):
 
2427
        test_home_dir = self.test_home_dir
 
2428
        if isinstance(test_home_dir, unicode):
 
2429
            test_home_dir = test_home_dir.encode(sys.getfilesystemencoding())
 
2430
        os.environ['HOME'] = test_home_dir
 
2431
        os.environ['BZR_HOME'] = test_home_dir
 
2432
 
 
2433
    def setUp(self):
 
2434
        super(TestCaseWithMemoryTransport, self).setUp()
 
2435
        self._make_test_root()
 
2436
        self.addCleanup(os.chdir, os.getcwdu())
 
2437
        self.makeAndChdirToTestDir()
 
2438
        self.overrideEnvironmentForTesting()
 
2439
        self.__readonly_server = None
 
2440
        self.__server = None
 
2441
        self.reduceLockdirTimeout()
 
2442
 
 
2443
    def setup_smart_server_with_call_log(self):
 
2444
        """Sets up a smart server as the transport server with a call log."""
 
2445
        self.transport_server = test_server.SmartTCPServer_for_testing
 
2446
        self.hpss_calls = []
 
2447
        import traceback
 
2448
        # Skip the current stack down to the caller of
 
2449
        # setup_smart_server_with_call_log
 
2450
        prefix_length = len(traceback.extract_stack()) - 2
 
2451
        def capture_hpss_call(params):
 
2452
            self.hpss_calls.append(
 
2453
                CapturedCall(params, prefix_length))
 
2454
        client._SmartClient.hooks.install_named_hook(
 
2455
            'call', capture_hpss_call, None)
 
2456
 
 
2457
    def reset_smart_call_log(self):
 
2458
        self.hpss_calls = []
 
2459
 
 
2460
 
 
2461
class TestCaseInTempDir(TestCaseWithMemoryTransport):
1000
2462
    """Derived class that runs a test within a temporary directory.
1001
2463
 
1002
2464
    This is useful for tests that need to create a branch, etc.
1006
2468
    All test cases create their own directory within that.  If the
1007
2469
    tests complete successfully, the directory is removed.
1008
2470
 
1009
 
    InTempDir is an old alias for FunctionalTestCase.
 
2471
    :ivar test_base_dir: The path of the top-level directory for this
 
2472
    test, which contains a home directory and a work directory.
 
2473
 
 
2474
    :ivar test_home_dir: An initially empty directory under test_base_dir
 
2475
    which is used as $HOME for this test.
 
2476
 
 
2477
    :ivar test_dir: A directory under test_base_dir used as the current
 
2478
    directory when the test proper is run.
1010
2479
    """
1011
2480
 
1012
 
    TEST_ROOT = None
1013
 
    _TEST_NAME = 'test'
1014
2481
    OVERRIDE_PYTHON = 'python'
1015
2482
 
1016
2483
    def check_file_contents(self, filename, expect):
1021
2488
            self.log("actually: %r" % contents)
1022
2489
            self.fail("contents of %s not as expected" % filename)
1023
2490
 
1024
 
    def _make_test_root(self):
1025
 
        if TestCaseInTempDir.TEST_ROOT is not None:
1026
 
            return
1027
 
        i = 0
1028
 
        while True:
1029
 
            root = u'test%04d.tmp' % i
1030
 
            try:
1031
 
                os.mkdir(root)
1032
 
            except OSError, e:
1033
 
                if e.errno == errno.EEXIST:
1034
 
                    i += 1
1035
 
                    continue
1036
 
                else:
1037
 
                    raise
1038
 
            # successfully created
1039
 
            TestCaseInTempDir.TEST_ROOT = osutils.abspath(root)
1040
 
            break
1041
 
        # make a fake bzr directory there to prevent any tests propagating
1042
 
        # up onto the source directory's real branch
1043
 
        bzrdir.BzrDir.create_standalone_workingtree(TestCaseInTempDir.TEST_ROOT)
1044
 
 
1045
 
    def setUp(self):
1046
 
        super(TestCaseInTempDir, self).setUp()
1047
 
        self._make_test_root()
1048
 
        _currentdir = os.getcwdu()
1049
 
        # shorten the name, to avoid test failures due to path length
1050
 
        short_id = self.id().replace('bzrlib.tests.', '') \
1051
 
                   .replace('__main__.', '')[-100:]
1052
 
        # it's possible the same test class is run several times for
1053
 
        # parameterized tests, so make sure the names don't collide.  
1054
 
        i = 0
1055
 
        while True:
1056
 
            if i > 0:
1057
 
                candidate_dir = '%s/%s.%d' % (self.TEST_ROOT, short_id, i)
1058
 
            else:
1059
 
                candidate_dir = '%s/%s' % (self.TEST_ROOT, short_id)
1060
 
            if os.path.exists(candidate_dir):
1061
 
                i = i + 1
1062
 
                continue
1063
 
            else:
1064
 
                os.mkdir(candidate_dir)
1065
 
                self.test_home_dir = candidate_dir + '/home'
1066
 
                os.mkdir(self.test_home_dir)
1067
 
                self.test_dir = candidate_dir + '/work'
1068
 
                os.mkdir(self.test_dir)
1069
 
                os.chdir(self.test_dir)
 
2491
    def _getTestDirPrefix(self):
 
2492
        # create a directory within the top level test directory
 
2493
        if sys.platform in ('win32', 'cygwin'):
 
2494
            name_prefix = re.sub('[<>*=+",:;_/\\-]', '_', self.id())
 
2495
            # windows is likely to have path-length limits so use a short name
 
2496
            name_prefix = name_prefix[-30:]
 
2497
        else:
 
2498
            name_prefix = re.sub('[/]', '_', self.id())
 
2499
        return name_prefix
 
2500
 
 
2501
    def makeAndChdirToTestDir(self):
 
2502
        """See TestCaseWithMemoryTransport.makeAndChdirToTestDir().
 
2503
 
 
2504
        For TestCaseInTempDir we create a temporary directory based on the test
 
2505
        name and then create two subdirs - test and home under it.
 
2506
        """
 
2507
        name_prefix = osutils.pathjoin(TestCaseWithMemoryTransport.TEST_ROOT,
 
2508
            self._getTestDirPrefix())
 
2509
        name = name_prefix
 
2510
        for i in range(100):
 
2511
            if os.path.exists(name):
 
2512
                name = name_prefix + '_' + str(i)
 
2513
            else:
 
2514
                # now create test and home directories within this dir
 
2515
                self.test_base_dir = name
 
2516
                self.addCleanup(self.deleteTestDir)
 
2517
                os.mkdir(self.test_base_dir)
1070
2518
                break
1071
 
        os.environ['HOME'] = self.test_home_dir
1072
 
        os.environ['APPDATA'] = self.test_home_dir
1073
 
        def _leaveDirectory():
1074
 
            os.chdir(_currentdir)
1075
 
        self.addCleanup(_leaveDirectory)
1076
 
        
1077
 
    def build_tree(self, shape, line_endings='native', transport=None):
 
2519
        self.permit_dir(self.test_base_dir)
 
2520
        # 'sprouting' and 'init' of a branch both walk up the tree to find
 
2521
        # stacking policy to honour; create a bzr dir with an unshared
 
2522
        # repository (but not a branch - our code would be trying to escape
 
2523
        # then!) to stop them, and permit it to be read.
 
2524
        # control = bzrdir.BzrDir.create(self.test_base_dir)
 
2525
        # control.create_repository()
 
2526
        self.test_home_dir = self.test_base_dir + '/home'
 
2527
        os.mkdir(self.test_home_dir)
 
2528
        self.test_dir = self.test_base_dir + '/work'
 
2529
        os.mkdir(self.test_dir)
 
2530
        os.chdir(self.test_dir)
 
2531
        # put name of test inside
 
2532
        f = file(self.test_base_dir + '/name', 'w')
 
2533
        try:
 
2534
            f.write(self.id())
 
2535
        finally:
 
2536
            f.close()
 
2537
 
 
2538
    def deleteTestDir(self):
 
2539
        os.chdir(TestCaseWithMemoryTransport.TEST_ROOT)
 
2540
        _rmtree_temp_dir(self.test_base_dir, test_id=self.id())
 
2541
 
 
2542
    def build_tree(self, shape, line_endings='binary', transport=None):
1078
2543
        """Build a test tree according to a pattern.
1079
2544
 
1080
2545
        shape is a sequence of file specifications.  If the final
1083
2548
        This assumes that all the elements in the tree being built are new.
1084
2549
 
1085
2550
        This doesn't add anything to a branch.
 
2551
 
 
2552
        :type shape:    list or tuple.
1086
2553
        :param line_endings: Either 'binary' or 'native'
1087
 
                             in binary mode, exact contents are written
1088
 
                             in native mode, the line endings match the
1089
 
                             default platform endings.
1090
 
 
1091
 
        :param transport: A transport to write to, for building trees on 
1092
 
                          VFS's. If the transport is readonly or None,
1093
 
                          "." is opened automatically.
 
2554
            in binary mode, exact contents are written in native mode, the
 
2555
            line endings match the default platform endings.
 
2556
        :param transport: A transport to write to, for building trees on VFS's.
 
2557
            If the transport is readonly or None, "." is opened automatically.
 
2558
        :return: None
1094
2559
        """
 
2560
        if type(shape) not in (list, tuple):
 
2561
            raise AssertionError("Parameter 'shape' should be "
 
2562
                "a list or a tuple. Got %r instead" % (shape,))
1095
2563
        # It's OK to just create them using forward slashes on windows.
1096
2564
        if transport is None or transport.is_readonly():
1097
2565
            transport = get_transport(".")
1098
2566
        for name in shape:
1099
 
            self.assert_(isinstance(name, basestring))
 
2567
            self.assertIsInstance(name, basestring)
1100
2568
            if name[-1] == '/':
1101
2569
                transport.mkdir(urlutils.escape(name[:-1]))
1102
2570
            else:
1105
2573
                elif line_endings == 'native':
1106
2574
                    end = os.linesep
1107
2575
                else:
1108
 
                    raise errors.BzrError('Invalid line ending request %r' % (line_endings,))
 
2576
                    raise errors.BzrError(
 
2577
                        'Invalid line ending request %r' % line_endings)
1109
2578
                content = "contents of %s%s" % (name.encode('utf-8'), end)
1110
 
                # Technically 'put()' is the right command. However, put
1111
 
                # uses an AtomicFile, which requires an extra rename into place
1112
 
                # As long as the files didn't exist in the past, append() will
1113
 
                # do the same thing as put()
1114
 
                # On jam's machine, make_kernel_like_tree is:
1115
 
                #   put:    4.5-7.5s (averaging 6s)
1116
 
                #   append: 2.9-4.5s
1117
 
                #   put_non_atomic: 2.9-4.5s
1118
2579
                transport.put_bytes_non_atomic(urlutils.escape(name), content)
1119
2580
 
1120
 
    def build_tree_contents(self, shape):
1121
 
        build_tree_contents(shape)
1122
 
 
1123
 
    def failUnlessExists(self, path):
1124
 
        """Fail unless path, which may be abs or relative, exists."""
1125
 
        self.failUnless(osutils.lexists(path))
1126
 
 
1127
 
    def failIfExists(self, path):
1128
 
        """Fail if path, which may be abs or relative, exists."""
1129
 
        self.failIf(osutils.lexists(path))
1130
 
        
1131
 
    def assertFileEqual(self, content, path):
1132
 
        """Fail if path does not contain 'content'."""
1133
 
        self.failUnless(osutils.lexists(path))
1134
 
        # TODO: jam 20060427 Shouldn't this be 'rb'?
1135
 
        self.assertEqualDiff(content, open(path, 'r').read())
 
2581
    build_tree_contents = staticmethod(treeshape.build_tree_contents)
 
2582
 
 
2583
    def assertInWorkingTree(self, path, root_path='.', tree=None):
 
2584
        """Assert whether path or paths are in the WorkingTree"""
 
2585
        if tree is None:
 
2586
            tree = workingtree.WorkingTree.open(root_path)
 
2587
        if not isinstance(path, basestring):
 
2588
            for p in path:
 
2589
                self.assertInWorkingTree(p, tree=tree)
 
2590
        else:
 
2591
            self.assertIsNot(tree.path2id(path), None,
 
2592
                path+' not in working tree.')
 
2593
 
 
2594
    def assertNotInWorkingTree(self, path, root_path='.', tree=None):
 
2595
        """Assert whether path or paths are not in the WorkingTree"""
 
2596
        if tree is None:
 
2597
            tree = workingtree.WorkingTree.open(root_path)
 
2598
        if not isinstance(path, basestring):
 
2599
            for p in path:
 
2600
                self.assertNotInWorkingTree(p,tree=tree)
 
2601
        else:
 
2602
            self.assertIs(tree.path2id(path), None, path+' in working tree.')
1136
2603
 
1137
2604
 
1138
2605
class TestCaseWithTransport(TestCaseInTempDir):
1145
2612
    ReadonlyTransportDecorator is used instead which allows the use of non disk
1146
2613
    based read write transports.
1147
2614
 
1148
 
    If an explicit class is provided for readonly access, that server and the 
 
2615
    If an explicit class is provided for readonly access, that server and the
1149
2616
    readwrite one must both define get_url() as resolving to os.getcwd().
1150
2617
    """
1151
2618
 
1152
 
    def __init__(self, methodName='testMethod'):
1153
 
        super(TestCaseWithTransport, self).__init__(methodName)
1154
 
        self.__readonly_server = None
1155
 
        self.__server = None
1156
 
        self.transport_server = default_transport
1157
 
        self.transport_readonly_server = None
1158
 
 
1159
 
    def get_readonly_url(self, relpath=None):
1160
 
        """Get a URL for the readonly transport.
1161
 
 
1162
 
        This will either be backed by '.' or a decorator to the transport 
1163
 
        used by self.get_url()
1164
 
        relpath provides for clients to get a path relative to the base url.
1165
 
        These should only be downwards relative, not upwards.
1166
 
        """
1167
 
        base = self.get_readonly_server().get_url()
1168
 
        if relpath is not None:
1169
 
            if not base.endswith('/'):
1170
 
                base = base + '/'
1171
 
            base = base + relpath
1172
 
        return base
1173
 
 
1174
 
    def get_readonly_server(self):
1175
 
        """Get the server instance for the readonly transport
1176
 
 
1177
 
        This is useful for some tests with specific servers to do diagnostics.
1178
 
        """
1179
 
        if self.__readonly_server is None:
1180
 
            if self.transport_readonly_server is None:
1181
 
                # readonly decorator requested
1182
 
                # bring up the server
1183
 
                self.get_url()
1184
 
                self.__readonly_server = ReadonlyServer()
1185
 
                self.__readonly_server.setUp(self.__server)
1186
 
            else:
1187
 
                self.__readonly_server = self.transport_readonly_server()
1188
 
                self.__readonly_server.setUp()
1189
 
            self.addCleanup(self.__readonly_server.tearDown)
1190
 
        return self.__readonly_server
1191
 
 
1192
 
    def get_server(self):
1193
 
        """Get the read/write server instance.
 
2619
    def get_vfs_only_server(self):
 
2620
        """See TestCaseWithMemoryTransport.
1194
2621
 
1195
2622
        This is useful for some tests with specific servers that need
1196
2623
        diagnostics.
1197
2624
        """
1198
 
        if self.__server is None:
1199
 
            self.__server = self.transport_server()
1200
 
            self.__server.setUp()
1201
 
            self.addCleanup(self.__server.tearDown)
1202
 
        return self.__server
1203
 
 
1204
 
    def get_url(self, relpath=None):
1205
 
        """Get a URL for the readwrite transport.
1206
 
 
1207
 
        This will either be backed by '.' or to an equivalent non-file based
1208
 
        facility.
1209
 
        relpath provides for clients to get a path relative to the base url.
1210
 
        These should only be downwards relative, not upwards.
1211
 
        """
1212
 
        base = self.get_server().get_url()
1213
 
        if relpath is not None and relpath != '.':
1214
 
            if not base.endswith('/'):
1215
 
                base = base + '/'
1216
 
            base = base + urlutils.escape(relpath)
1217
 
        return base
1218
 
 
1219
 
    def get_transport(self):
1220
 
        """Return a writeable transport for the test scratch space"""
1221
 
        t = get_transport(self.get_url())
1222
 
        self.assertFalse(t.is_readonly())
1223
 
        return t
1224
 
 
1225
 
    def get_readonly_transport(self):
1226
 
        """Return a readonly transport for the test scratch space
1227
 
        
1228
 
        This can be used to test that operations which should only need
1229
 
        readonly access in fact do not try to write.
1230
 
        """
1231
 
        t = get_transport(self.get_readonly_url())
1232
 
        self.assertTrue(t.is_readonly())
1233
 
        return t
1234
 
 
1235
 
    def make_branch(self, relpath, format=None):
1236
 
        """Create a branch on the transport at relpath."""
1237
 
        repo = self.make_repository(relpath, format=format)
1238
 
        return repo.bzrdir.create_branch()
1239
 
 
1240
 
    def make_bzrdir(self, relpath, format=None):
1241
 
        try:
1242
 
            url = self.get_url(relpath)
1243
 
            mutter('relpath %r => url %r', relpath, url)
1244
 
            segments = url.split('/')
1245
 
            if segments and segments[-1] not in ('', '.'):
1246
 
                parent = '/'.join(segments[:-1])
1247
 
                t = get_transport(parent)
1248
 
                try:
1249
 
                    t.mkdir(segments[-1])
1250
 
                except errors.FileExists:
1251
 
                    pass
1252
 
            if format is None:
1253
 
                format=bzrlib.bzrdir.BzrDirFormat.get_default_format()
1254
 
            # FIXME: make this use a single transport someday. RBC 20060418
1255
 
            return format.initialize_on_transport(get_transport(relpath))
1256
 
        except errors.UninitializableFormat:
1257
 
            raise TestSkipped("Format %s is not initializable." % format)
1258
 
 
1259
 
    def make_repository(self, relpath, shared=False, format=None):
1260
 
        """Create a repository on our default transport at relpath."""
1261
 
        made_control = self.make_bzrdir(relpath, format=format)
1262
 
        return made_control.create_repository(shared=shared)
 
2625
        if self.__vfs_server is None:
 
2626
            self.__vfs_server = self.vfs_transport_factory()
 
2627
            self.start_server(self.__vfs_server)
 
2628
        return self.__vfs_server
1263
2629
 
1264
2630
    def make_branch_and_tree(self, relpath, format=None):
1265
2631
        """Create a branch on the transport and a tree locally.
1266
2632
 
1267
 
        Returns the tree.
 
2633
        If the transport is not a LocalTransport, the Tree can't be created on
 
2634
        the transport.  In that case if the vfs_transport_factory is
 
2635
        LocalURLServer the working tree is created in the local
 
2636
        directory backing the transport, and the returned tree's branch and
 
2637
        repository will also be accessed locally. Otherwise a lightweight
 
2638
        checkout is created and returned.
 
2639
 
 
2640
        We do this because we can't physically create a tree in the local
 
2641
        path, with a branch reference to the transport_factory url, and
 
2642
        a branch + repository in the vfs_transport, unless the vfs_transport
 
2643
        namespace is distinct from the local disk - the two branch objects
 
2644
        would collide. While we could construct a tree with its branch object
 
2645
        pointing at the transport_factory transport in memory, reopening it
 
2646
        would behaving unexpectedly, and has in the past caused testing bugs
 
2647
        when we tried to do it that way.
 
2648
 
 
2649
        :param format: The BzrDirFormat.
 
2650
        :returns: the WorkingTree.
1268
2651
        """
1269
2652
        # TODO: always use the local disk path for the working tree,
1270
2653
        # this obviously requires a format that supports branch references
1274
2657
        try:
1275
2658
            return b.bzrdir.create_workingtree()
1276
2659
        except errors.NotLocalUrl:
1277
 
            # new formats - catch No tree error and create
1278
 
            # a branch reference and a checkout.
1279
 
            # old formats at that point - raise TestSkipped.
1280
 
            # TODO: rbc 20060208
1281
 
            return WorkingTreeFormat2().initialize(bzrdir.BzrDir.open(relpath))
 
2660
            # We can only make working trees locally at the moment.  If the
 
2661
            # transport can't support them, then we keep the non-disk-backed
 
2662
            # branch and create a local checkout.
 
2663
            if self.vfs_transport_factory is test_server.LocalURLServer:
 
2664
                # the branch is colocated on disk, we cannot create a checkout.
 
2665
                # hopefully callers will expect this.
 
2666
                local_controldir= bzrdir.BzrDir.open(self.get_vfs_only_url(relpath))
 
2667
                wt = local_controldir.create_workingtree()
 
2668
                if wt.branch._format != b._format:
 
2669
                    wt._branch = b
 
2670
                    # Make sure that assigning to wt._branch fixes wt.branch,
 
2671
                    # in case the implementation details of workingtree objects
 
2672
                    # change.
 
2673
                    self.assertIs(b, wt.branch)
 
2674
                return wt
 
2675
            else:
 
2676
                return b.create_checkout(relpath, lightweight=True)
1282
2677
 
1283
2678
    def assertIsDirectory(self, relpath, transport):
1284
2679
        """Assert that relpath within transport is a directory.
1295
2690
            self.fail("path %s is not a directory; has mode %#o"
1296
2691
                      % (relpath, mode))
1297
2692
 
 
2693
    def assertTreesEqual(self, left, right):
 
2694
        """Check that left and right have the same content and properties."""
 
2695
        # we use a tree delta to check for equality of the content, and we
 
2696
        # manually check for equality of other things such as the parents list.
 
2697
        self.assertEqual(left.get_parent_ids(), right.get_parent_ids())
 
2698
        differences = left.changes_from(right)
 
2699
        self.assertFalse(differences.has_changed(),
 
2700
            "Trees %r and %r are different: %r" % (left, right, differences))
 
2701
 
 
2702
    def setUp(self):
 
2703
        super(TestCaseWithTransport, self).setUp()
 
2704
        self.__vfs_server = None
 
2705
 
 
2706
    def disable_missing_extensions_warning(self):
 
2707
        """Some tests expect a precise stderr content.
 
2708
 
 
2709
        There is no point in forcing them to duplicate the extension related
 
2710
        warning.
 
2711
        """
 
2712
        config.GlobalConfig().set_user_option('ignore_missing_extensions', True)
 
2713
 
1298
2714
 
1299
2715
class ChrootedTestCase(TestCaseWithTransport):
1300
2716
    """A support class that provides readonly urls outside the local namespace.
1304
2720
    for readonly urls.
1305
2721
 
1306
2722
    TODO RBC 20060127: make this an option to TestCaseWithTransport so it can
1307
 
                       be used without needed to redo it when a different 
 
2723
                       be used without needed to redo it when a different
1308
2724
                       subclass is in use ?
1309
2725
    """
1310
2726
 
1311
2727
    def setUp(self):
1312
2728
        super(ChrootedTestCase, self).setUp()
1313
 
        if not self.transport_server == bzrlib.transport.memory.MemoryServer:
1314
 
            self.transport_readonly_server = bzrlib.transport.http.HttpServer
 
2729
        if not self.vfs_transport_factory == memory.MemoryServer:
 
2730
            self.transport_readonly_server = HttpServer
 
2731
 
 
2732
 
 
2733
def condition_id_re(pattern):
 
2734
    """Create a condition filter which performs a re check on a test's id.
 
2735
 
 
2736
    :param pattern: A regular expression string.
 
2737
    :return: A callable that returns True if the re matches.
 
2738
    """
 
2739
    filter_re = osutils.re_compile_checked(pattern, 0,
 
2740
        'test filter')
 
2741
    def condition(test):
 
2742
        test_id = test.id()
 
2743
        return filter_re.search(test_id)
 
2744
    return condition
 
2745
 
 
2746
 
 
2747
def condition_isinstance(klass_or_klass_list):
 
2748
    """Create a condition filter which returns isinstance(param, klass).
 
2749
 
 
2750
    :return: A callable which when called with one parameter obj return the
 
2751
        result of isinstance(obj, klass_or_klass_list).
 
2752
    """
 
2753
    def condition(obj):
 
2754
        return isinstance(obj, klass_or_klass_list)
 
2755
    return condition
 
2756
 
 
2757
 
 
2758
def condition_id_in_list(id_list):
 
2759
    """Create a condition filter which verify that test's id in a list.
 
2760
 
 
2761
    :param id_list: A TestIdList object.
 
2762
    :return: A callable that returns True if the test's id appears in the list.
 
2763
    """
 
2764
    def condition(test):
 
2765
        return id_list.includes(test.id())
 
2766
    return condition
 
2767
 
 
2768
 
 
2769
def condition_id_startswith(starts):
 
2770
    """Create a condition filter verifying that test's id starts with a string.
 
2771
 
 
2772
    :param starts: A list of string.
 
2773
    :return: A callable that returns True if the test's id starts with one of
 
2774
        the given strings.
 
2775
    """
 
2776
    def condition(test):
 
2777
        for start in starts:
 
2778
            if test.id().startswith(start):
 
2779
                return True
 
2780
        return False
 
2781
    return condition
 
2782
 
 
2783
 
 
2784
def exclude_tests_by_condition(suite, condition):
 
2785
    """Create a test suite which excludes some tests from suite.
 
2786
 
 
2787
    :param suite: The suite to get tests from.
 
2788
    :param condition: A callable whose result evaluates True when called with a
 
2789
        test case which should be excluded from the result.
 
2790
    :return: A suite which contains the tests found in suite that fail
 
2791
        condition.
 
2792
    """
 
2793
    result = []
 
2794
    for test in iter_suite_tests(suite):
 
2795
        if not condition(test):
 
2796
            result.append(test)
 
2797
    return TestUtil.TestSuite(result)
 
2798
 
 
2799
 
 
2800
def filter_suite_by_condition(suite, condition):
 
2801
    """Create a test suite by filtering another one.
 
2802
 
 
2803
    :param suite: The source suite.
 
2804
    :param condition: A callable whose result evaluates True when called with a
 
2805
        test case which should be included in the result.
 
2806
    :return: A suite which contains the tests found in suite that pass
 
2807
        condition.
 
2808
    """
 
2809
    result = []
 
2810
    for test in iter_suite_tests(suite):
 
2811
        if condition(test):
 
2812
            result.append(test)
 
2813
    return TestUtil.TestSuite(result)
1315
2814
 
1316
2815
 
1317
2816
def filter_suite_by_re(suite, pattern):
1318
 
    result = TestUtil.TestSuite()
1319
 
    filter_re = re.compile(pattern)
 
2817
    """Create a test suite by filtering another one.
 
2818
 
 
2819
    :param suite:           the source suite
 
2820
    :param pattern:         pattern that names must match
 
2821
    :returns: the newly created suite
 
2822
    """
 
2823
    condition = condition_id_re(pattern)
 
2824
    result_suite = filter_suite_by_condition(suite, condition)
 
2825
    return result_suite
 
2826
 
 
2827
 
 
2828
def filter_suite_by_id_list(suite, test_id_list):
 
2829
    """Create a test suite by filtering another one.
 
2830
 
 
2831
    :param suite: The source suite.
 
2832
    :param test_id_list: A list of the test ids to keep as strings.
 
2833
    :returns: the newly created suite
 
2834
    """
 
2835
    condition = condition_id_in_list(test_id_list)
 
2836
    result_suite = filter_suite_by_condition(suite, condition)
 
2837
    return result_suite
 
2838
 
 
2839
 
 
2840
def filter_suite_by_id_startswith(suite, start):
 
2841
    """Create a test suite by filtering another one.
 
2842
 
 
2843
    :param suite: The source suite.
 
2844
    :param start: A list of string the test id must start with one of.
 
2845
    :returns: the newly created suite
 
2846
    """
 
2847
    condition = condition_id_startswith(start)
 
2848
    result_suite = filter_suite_by_condition(suite, condition)
 
2849
    return result_suite
 
2850
 
 
2851
 
 
2852
def exclude_tests_by_re(suite, pattern):
 
2853
    """Create a test suite which excludes some tests from suite.
 
2854
 
 
2855
    :param suite: The suite to get tests from.
 
2856
    :param pattern: A regular expression string. Test ids that match this
 
2857
        pattern will be excluded from the result.
 
2858
    :return: A TestSuite that contains all the tests from suite without the
 
2859
        tests that matched pattern. The order of tests is the same as it was in
 
2860
        suite.
 
2861
    """
 
2862
    return exclude_tests_by_condition(suite, condition_id_re(pattern))
 
2863
 
 
2864
 
 
2865
def preserve_input(something):
 
2866
    """A helper for performing test suite transformation chains.
 
2867
 
 
2868
    :param something: Anything you want to preserve.
 
2869
    :return: Something.
 
2870
    """
 
2871
    return something
 
2872
 
 
2873
 
 
2874
def randomize_suite(suite):
 
2875
    """Return a new TestSuite with suite's tests in random order.
 
2876
 
 
2877
    The tests in the input suite are flattened into a single suite in order to
 
2878
    accomplish this. Any nested TestSuites are removed to provide global
 
2879
    randomness.
 
2880
    """
 
2881
    tests = list(iter_suite_tests(suite))
 
2882
    random.shuffle(tests)
 
2883
    return TestUtil.TestSuite(tests)
 
2884
 
 
2885
 
 
2886
def split_suite_by_condition(suite, condition):
 
2887
    """Split a test suite into two by a condition.
 
2888
 
 
2889
    :param suite: The suite to split.
 
2890
    :param condition: The condition to match on. Tests that match this
 
2891
        condition are returned in the first test suite, ones that do not match
 
2892
        are in the second suite.
 
2893
    :return: A tuple of two test suites, where the first contains tests from
 
2894
        suite matching the condition, and the second contains the remainder
 
2895
        from suite. The order within each output suite is the same as it was in
 
2896
        suite.
 
2897
    """
 
2898
    matched = []
 
2899
    did_not_match = []
1320
2900
    for test in iter_suite_tests(suite):
1321
 
        if filter_re.search(test.id()):
1322
 
            result.addTest(test)
1323
 
    return result
 
2901
        if condition(test):
 
2902
            matched.append(test)
 
2903
        else:
 
2904
            did_not_match.append(test)
 
2905
    return TestUtil.TestSuite(matched), TestUtil.TestSuite(did_not_match)
 
2906
 
 
2907
 
 
2908
def split_suite_by_re(suite, pattern):
 
2909
    """Split a test suite into two by a regular expression.
 
2910
 
 
2911
    :param suite: The suite to split.
 
2912
    :param pattern: A regular expression string. Test ids that match this
 
2913
        pattern will be in the first test suite returned, and the others in the
 
2914
        second test suite returned.
 
2915
    :return: A tuple of two test suites, where the first contains tests from
 
2916
        suite matching pattern, and the second contains the remainder from
 
2917
        suite. The order within each output suite is the same as it was in
 
2918
        suite.
 
2919
    """
 
2920
    return split_suite_by_condition(suite, condition_id_re(pattern))
1324
2921
 
1325
2922
 
1326
2923
def run_suite(suite, name='test', verbose=False, pattern=".*",
1327
 
              stop_on_failure=False, keep_output=False,
1328
 
              transport=None, lsprof_timed=None, bench_history=None):
1329
 
    TestCaseInTempDir._TEST_NAME = name
 
2924
              stop_on_failure=False,
 
2925
              transport=None, lsprof_timed=None, bench_history=None,
 
2926
              matching_tests_first=None,
 
2927
              list_only=False,
 
2928
              random_seed=None,
 
2929
              exclude_pattern=None,
 
2930
              strict=False,
 
2931
              runner_class=None,
 
2932
              suite_decorators=None,
 
2933
              stream=None,
 
2934
              result_decorators=None,
 
2935
              ):
 
2936
    """Run a test suite for bzr selftest.
 
2937
 
 
2938
    :param runner_class: The class of runner to use. Must support the
 
2939
        constructor arguments passed by run_suite which are more than standard
 
2940
        python uses.
 
2941
    :return: A boolean indicating success.
 
2942
    """
1330
2943
    TestCase._gather_lsprof_in_benchmarks = lsprof_timed
1331
2944
    if verbose:
1332
2945
        verbosity = 2
1333
 
        pb = None
1334
2946
    else:
1335
2947
        verbosity = 1
1336
 
        pb = progress.ProgressBar()
1337
 
    runner = TextTestRunner(stream=sys.stdout,
 
2948
    if runner_class is None:
 
2949
        runner_class = TextTestRunner
 
2950
    if stream is None:
 
2951
        stream = sys.stdout
 
2952
    runner = runner_class(stream=stream,
1338
2953
                            descriptions=0,
1339
2954
                            verbosity=verbosity,
1340
 
                            keep_output=keep_output,
1341
 
                            pb=pb,
1342
 
                            bench_history=bench_history)
 
2955
                            bench_history=bench_history,
 
2956
                            strict=strict,
 
2957
                            result_decorators=result_decorators,
 
2958
                            )
1343
2959
    runner.stop_on_failure=stop_on_failure
1344
 
    if pattern != '.*':
1345
 
        suite = filter_suite_by_re(suite, pattern)
 
2960
    # built in decorator factories:
 
2961
    decorators = [
 
2962
        random_order(random_seed, runner),
 
2963
        exclude_tests(exclude_pattern),
 
2964
        ]
 
2965
    if matching_tests_first:
 
2966
        decorators.append(tests_first(pattern))
 
2967
    else:
 
2968
        decorators.append(filter_tests(pattern))
 
2969
    if suite_decorators:
 
2970
        decorators.extend(suite_decorators)
 
2971
    # tell the result object how many tests will be running: (except if
 
2972
    # --parallel=fork is being used. Robert said he will provide a better
 
2973
    # progress design later -- vila 20090817)
 
2974
    if fork_decorator not in decorators:
 
2975
        decorators.append(CountingDecorator)
 
2976
    for decorator in decorators:
 
2977
        suite = decorator(suite)
 
2978
    if list_only:
 
2979
        # Done after test suite decoration to allow randomisation etc
 
2980
        # to take effect, though that is of marginal benefit.
 
2981
        if verbosity >= 2:
 
2982
            stream.write("Listing tests only ...\n")
 
2983
        for t in iter_suite_tests(suite):
 
2984
            stream.write("%s\n" % (t.id()))
 
2985
        return True
1346
2986
    result = runner.run(suite)
1347
 
    return result.wasSuccessful()
 
2987
    if strict:
 
2988
        return result.wasStrictlySuccessful()
 
2989
    else:
 
2990
        return result.wasSuccessful()
 
2991
 
 
2992
 
 
2993
# A registry where get() returns a suite decorator.
 
2994
parallel_registry = registry.Registry()
 
2995
 
 
2996
 
 
2997
def fork_decorator(suite):
 
2998
    concurrency = osutils.local_concurrency()
 
2999
    if concurrency == 1:
 
3000
        return suite
 
3001
    from testtools import ConcurrentTestSuite
 
3002
    return ConcurrentTestSuite(suite, fork_for_tests)
 
3003
parallel_registry.register('fork', fork_decorator)
 
3004
 
 
3005
 
 
3006
def subprocess_decorator(suite):
 
3007
    concurrency = osutils.local_concurrency()
 
3008
    if concurrency == 1:
 
3009
        return suite
 
3010
    from testtools import ConcurrentTestSuite
 
3011
    return ConcurrentTestSuite(suite, reinvoke_for_tests)
 
3012
parallel_registry.register('subprocess', subprocess_decorator)
 
3013
 
 
3014
 
 
3015
def exclude_tests(exclude_pattern):
 
3016
    """Return a test suite decorator that excludes tests."""
 
3017
    if exclude_pattern is None:
 
3018
        return identity_decorator
 
3019
    def decorator(suite):
 
3020
        return ExcludeDecorator(suite, exclude_pattern)
 
3021
    return decorator
 
3022
 
 
3023
 
 
3024
def filter_tests(pattern):
 
3025
    if pattern == '.*':
 
3026
        return identity_decorator
 
3027
    def decorator(suite):
 
3028
        return FilterTestsDecorator(suite, pattern)
 
3029
    return decorator
 
3030
 
 
3031
 
 
3032
def random_order(random_seed, runner):
 
3033
    """Return a test suite decorator factory for randomising tests order.
 
3034
    
 
3035
    :param random_seed: now, a string which casts to a long, or a long.
 
3036
    :param runner: A test runner with a stream attribute to report on.
 
3037
    """
 
3038
    if random_seed is None:
 
3039
        return identity_decorator
 
3040
    def decorator(suite):
 
3041
        return RandomDecorator(suite, random_seed, runner.stream)
 
3042
    return decorator
 
3043
 
 
3044
 
 
3045
def tests_first(pattern):
 
3046
    if pattern == '.*':
 
3047
        return identity_decorator
 
3048
    def decorator(suite):
 
3049
        return TestFirstDecorator(suite, pattern)
 
3050
    return decorator
 
3051
 
 
3052
 
 
3053
def identity_decorator(suite):
 
3054
    """Return suite."""
 
3055
    return suite
 
3056
 
 
3057
 
 
3058
class TestDecorator(TestSuite):
 
3059
    """A decorator for TestCase/TestSuite objects.
 
3060
    
 
3061
    Usually, subclasses should override __iter__(used when flattening test
 
3062
    suites), which we do to filter, reorder, parallelise and so on, run() and
 
3063
    debug().
 
3064
    """
 
3065
 
 
3066
    def __init__(self, suite):
 
3067
        TestSuite.__init__(self)
 
3068
        self.addTest(suite)
 
3069
 
 
3070
    def countTestCases(self):
 
3071
        cases = 0
 
3072
        for test in self:
 
3073
            cases += test.countTestCases()
 
3074
        return cases
 
3075
 
 
3076
    def debug(self):
 
3077
        for test in self:
 
3078
            test.debug()
 
3079
 
 
3080
    def run(self, result):
 
3081
        # Use iteration on self, not self._tests, to allow subclasses to hook
 
3082
        # into __iter__.
 
3083
        for test in self:
 
3084
            if result.shouldStop:
 
3085
                break
 
3086
            test.run(result)
 
3087
        return result
 
3088
 
 
3089
 
 
3090
class CountingDecorator(TestDecorator):
 
3091
    """A decorator which calls result.progress(self.countTestCases)."""
 
3092
 
 
3093
    def run(self, result):
 
3094
        progress_method = getattr(result, 'progress', None)
 
3095
        if callable(progress_method):
 
3096
            progress_method(self.countTestCases(), SUBUNIT_SEEK_SET)
 
3097
        return super(CountingDecorator, self).run(result)
 
3098
 
 
3099
 
 
3100
class ExcludeDecorator(TestDecorator):
 
3101
    """A decorator which excludes test matching an exclude pattern."""
 
3102
 
 
3103
    def __init__(self, suite, exclude_pattern):
 
3104
        TestDecorator.__init__(self, suite)
 
3105
        self.exclude_pattern = exclude_pattern
 
3106
        self.excluded = False
 
3107
 
 
3108
    def __iter__(self):
 
3109
        if self.excluded:
 
3110
            return iter(self._tests)
 
3111
        self.excluded = True
 
3112
        suite = exclude_tests_by_re(self, self.exclude_pattern)
 
3113
        del self._tests[:]
 
3114
        self.addTests(suite)
 
3115
        return iter(self._tests)
 
3116
 
 
3117
 
 
3118
class FilterTestsDecorator(TestDecorator):
 
3119
    """A decorator which filters tests to those matching a pattern."""
 
3120
 
 
3121
    def __init__(self, suite, pattern):
 
3122
        TestDecorator.__init__(self, suite)
 
3123
        self.pattern = pattern
 
3124
        self.filtered = False
 
3125
 
 
3126
    def __iter__(self):
 
3127
        if self.filtered:
 
3128
            return iter(self._tests)
 
3129
        self.filtered = True
 
3130
        suite = filter_suite_by_re(self, self.pattern)
 
3131
        del self._tests[:]
 
3132
        self.addTests(suite)
 
3133
        return iter(self._tests)
 
3134
 
 
3135
 
 
3136
class RandomDecorator(TestDecorator):
 
3137
    """A decorator which randomises the order of its tests."""
 
3138
 
 
3139
    def __init__(self, suite, random_seed, stream):
 
3140
        TestDecorator.__init__(self, suite)
 
3141
        self.random_seed = random_seed
 
3142
        self.randomised = False
 
3143
        self.stream = stream
 
3144
 
 
3145
    def __iter__(self):
 
3146
        if self.randomised:
 
3147
            return iter(self._tests)
 
3148
        self.randomised = True
 
3149
        self.stream.write("Randomizing test order using seed %s\n\n" %
 
3150
            (self.actual_seed()))
 
3151
        # Initialise the random number generator.
 
3152
        random.seed(self.actual_seed())
 
3153
        suite = randomize_suite(self)
 
3154
        del self._tests[:]
 
3155
        self.addTests(suite)
 
3156
        return iter(self._tests)
 
3157
 
 
3158
    def actual_seed(self):
 
3159
        if self.random_seed == "now":
 
3160
            # We convert the seed to a long to make it reuseable across
 
3161
            # invocations (because the user can reenter it).
 
3162
            self.random_seed = long(time.time())
 
3163
        else:
 
3164
            # Convert the seed to a long if we can
 
3165
            try:
 
3166
                self.random_seed = long(self.random_seed)
 
3167
            except:
 
3168
                pass
 
3169
        return self.random_seed
 
3170
 
 
3171
 
 
3172
class TestFirstDecorator(TestDecorator):
 
3173
    """A decorator which moves named tests to the front."""
 
3174
 
 
3175
    def __init__(self, suite, pattern):
 
3176
        TestDecorator.__init__(self, suite)
 
3177
        self.pattern = pattern
 
3178
        self.filtered = False
 
3179
 
 
3180
    def __iter__(self):
 
3181
        if self.filtered:
 
3182
            return iter(self._tests)
 
3183
        self.filtered = True
 
3184
        suites = split_suite_by_re(self, self.pattern)
 
3185
        del self._tests[:]
 
3186
        self.addTests(suites)
 
3187
        return iter(self._tests)
 
3188
 
 
3189
 
 
3190
def partition_tests(suite, count):
 
3191
    """Partition suite into count lists of tests."""
 
3192
    result = []
 
3193
    tests = list(iter_suite_tests(suite))
 
3194
    tests_per_process = int(math.ceil(float(len(tests)) / count))
 
3195
    for block in range(count):
 
3196
        low_test = block * tests_per_process
 
3197
        high_test = low_test + tests_per_process
 
3198
        process_tests = tests[low_test:high_test]
 
3199
        result.append(process_tests)
 
3200
    return result
 
3201
 
 
3202
 
 
3203
def workaround_zealous_crypto_random():
 
3204
    """Crypto.Random want to help us being secure, but we don't care here.
 
3205
 
 
3206
    This workaround some test failure related to the sftp server. Once paramiko
 
3207
    stop using the controversial API in Crypto.Random, we may get rid of it.
 
3208
    """
 
3209
    try:
 
3210
        from Crypto.Random import atfork
 
3211
        atfork()
 
3212
    except ImportError:
 
3213
        pass
 
3214
 
 
3215
 
 
3216
def fork_for_tests(suite):
 
3217
    """Take suite and start up one runner per CPU by forking()
 
3218
 
 
3219
    :return: An iterable of TestCase-like objects which can each have
 
3220
        run(result) called on them to feed tests to result.
 
3221
    """
 
3222
    concurrency = osutils.local_concurrency()
 
3223
    result = []
 
3224
    from subunit import TestProtocolClient, ProtocolTestCase
 
3225
    from subunit.test_results import AutoTimingTestResultDecorator
 
3226
    class TestInOtherProcess(ProtocolTestCase):
 
3227
        # Should be in subunit, I think. RBC.
 
3228
        def __init__(self, stream, pid):
 
3229
            ProtocolTestCase.__init__(self, stream)
 
3230
            self.pid = pid
 
3231
 
 
3232
        def run(self, result):
 
3233
            try:
 
3234
                ProtocolTestCase.run(self, result)
 
3235
            finally:
 
3236
                os.waitpid(self.pid, 0)
 
3237
 
 
3238
    test_blocks = partition_tests(suite, concurrency)
 
3239
    for process_tests in test_blocks:
 
3240
        process_suite = TestSuite()
 
3241
        process_suite.addTests(process_tests)
 
3242
        c2pread, c2pwrite = os.pipe()
 
3243
        pid = os.fork()
 
3244
        if pid == 0:
 
3245
            workaround_zealous_crypto_random()
 
3246
            try:
 
3247
                os.close(c2pread)
 
3248
                # Leave stderr and stdout open so we can see test noise
 
3249
                # Close stdin so that the child goes away if it decides to
 
3250
                # read from stdin (otherwise its a roulette to see what
 
3251
                # child actually gets keystrokes for pdb etc).
 
3252
                sys.stdin.close()
 
3253
                sys.stdin = None
 
3254
                stream = os.fdopen(c2pwrite, 'wb', 1)
 
3255
                subunit_result = AutoTimingTestResultDecorator(
 
3256
                    TestProtocolClient(stream))
 
3257
                process_suite.run(subunit_result)
 
3258
            finally:
 
3259
                os._exit(0)
 
3260
        else:
 
3261
            os.close(c2pwrite)
 
3262
            stream = os.fdopen(c2pread, 'rb', 1)
 
3263
            test = TestInOtherProcess(stream, pid)
 
3264
            result.append(test)
 
3265
    return result
 
3266
 
 
3267
 
 
3268
def reinvoke_for_tests(suite):
 
3269
    """Take suite and start up one runner per CPU using subprocess().
 
3270
 
 
3271
    :return: An iterable of TestCase-like objects which can each have
 
3272
        run(result) called on them to feed tests to result.
 
3273
    """
 
3274
    concurrency = osutils.local_concurrency()
 
3275
    result = []
 
3276
    from subunit import ProtocolTestCase
 
3277
    class TestInSubprocess(ProtocolTestCase):
 
3278
        def __init__(self, process, name):
 
3279
            ProtocolTestCase.__init__(self, process.stdout)
 
3280
            self.process = process
 
3281
            self.process.stdin.close()
 
3282
            self.name = name
 
3283
 
 
3284
        def run(self, result):
 
3285
            try:
 
3286
                ProtocolTestCase.run(self, result)
 
3287
            finally:
 
3288
                self.process.wait()
 
3289
                os.unlink(self.name)
 
3290
            # print "pid %d finished" % finished_process
 
3291
    test_blocks = partition_tests(suite, concurrency)
 
3292
    for process_tests in test_blocks:
 
3293
        # ugly; currently reimplement rather than reuses TestCase methods.
 
3294
        bzr_path = os.path.dirname(os.path.dirname(bzrlib.__file__))+'/bzr'
 
3295
        if not os.path.isfile(bzr_path):
 
3296
            # We are probably installed. Assume sys.argv is the right file
 
3297
            bzr_path = sys.argv[0]
 
3298
        bzr_path = [bzr_path]
 
3299
        if sys.platform == "win32":
 
3300
            # if we're on windows, we can't execute the bzr script directly
 
3301
            bzr_path = [sys.executable] + bzr_path
 
3302
        fd, test_list_file_name = tempfile.mkstemp()
 
3303
        test_list_file = os.fdopen(fd, 'wb', 1)
 
3304
        for test in process_tests:
 
3305
            test_list_file.write(test.id() + '\n')
 
3306
        test_list_file.close()
 
3307
        try:
 
3308
            argv = bzr_path + ['selftest', '--load-list', test_list_file_name,
 
3309
                '--subunit']
 
3310
            if '--no-plugins' in sys.argv:
 
3311
                argv.append('--no-plugins')
 
3312
            # stderr=STDOUT would be ideal, but until we prevent noise on
 
3313
            # stderr it can interrupt the subunit protocol.
 
3314
            process = Popen(argv, stdin=PIPE, stdout=PIPE, stderr=PIPE,
 
3315
                bufsize=1)
 
3316
            test = TestInSubprocess(process, test_list_file_name)
 
3317
            result.append(test)
 
3318
        except:
 
3319
            os.unlink(test_list_file_name)
 
3320
            raise
 
3321
    return result
 
3322
 
 
3323
 
 
3324
class ForwardingResult(unittest.TestResult):
 
3325
 
 
3326
    def __init__(self, target):
 
3327
        unittest.TestResult.__init__(self)
 
3328
        self.result = target
 
3329
 
 
3330
    def startTest(self, test):
 
3331
        self.result.startTest(test)
 
3332
 
 
3333
    def stopTest(self, test):
 
3334
        self.result.stopTest(test)
 
3335
 
 
3336
    def startTestRun(self):
 
3337
        self.result.startTestRun()
 
3338
 
 
3339
    def stopTestRun(self):
 
3340
        self.result.stopTestRun()
 
3341
 
 
3342
    def addSkip(self, test, reason):
 
3343
        self.result.addSkip(test, reason)
 
3344
 
 
3345
    def addSuccess(self, test):
 
3346
        self.result.addSuccess(test)
 
3347
 
 
3348
    def addError(self, test, err):
 
3349
        self.result.addError(test, err)
 
3350
 
 
3351
    def addFailure(self, test, err):
 
3352
        self.result.addFailure(test, err)
 
3353
ForwardingResult = testtools.ExtendedToOriginalDecorator
 
3354
 
 
3355
 
 
3356
class ProfileResult(ForwardingResult):
 
3357
    """Generate profiling data for all activity between start and success.
 
3358
    
 
3359
    The profile data is appended to the test's _benchcalls attribute and can
 
3360
    be accessed by the forwarded-to TestResult.
 
3361
 
 
3362
    While it might be cleaner do accumulate this in stopTest, addSuccess is
 
3363
    where our existing output support for lsprof is, and this class aims to
 
3364
    fit in with that: while it could be moved it's not necessary to accomplish
 
3365
    test profiling, nor would it be dramatically cleaner.
 
3366
    """
 
3367
 
 
3368
    def startTest(self, test):
 
3369
        self.profiler = bzrlib.lsprof.BzrProfiler()
 
3370
        self.profiler.start()
 
3371
        ForwardingResult.startTest(self, test)
 
3372
 
 
3373
    def addSuccess(self, test):
 
3374
        stats = self.profiler.stop()
 
3375
        try:
 
3376
            calls = test._benchcalls
 
3377
        except AttributeError:
 
3378
            test._benchcalls = []
 
3379
            calls = test._benchcalls
 
3380
        calls.append(((test.id(), "", ""), stats))
 
3381
        ForwardingResult.addSuccess(self, test)
 
3382
 
 
3383
    def stopTest(self, test):
 
3384
        ForwardingResult.stopTest(self, test)
 
3385
        self.profiler = None
 
3386
 
 
3387
 
 
3388
# Controlled by "bzr selftest -E=..." option
 
3389
# Currently supported:
 
3390
#   -Eallow_debug           Will no longer clear debug.debug_flags() so it
 
3391
#                           preserves any flags supplied at the command line.
 
3392
#   -Edisable_lock_checks   Turns errors in mismatched locks into simple prints
 
3393
#                           rather than failing tests. And no longer raise
 
3394
#                           LockContention when fctnl locks are not being used
 
3395
#                           with proper exclusion rules.
 
3396
selftest_debug_flags = set()
1348
3397
 
1349
3398
 
1350
3399
def selftest(verbose=False, pattern=".*", stop_on_failure=True,
1351
 
             keep_output=False,
1352
3400
             transport=None,
1353
3401
             test_suite_factory=None,
1354
3402
             lsprof_timed=None,
1355
 
             bench_history=None):
 
3403
             bench_history=None,
 
3404
             matching_tests_first=None,
 
3405
             list_only=False,
 
3406
             random_seed=None,
 
3407
             exclude_pattern=None,
 
3408
             strict=False,
 
3409
             load_list=None,
 
3410
             debug_flags=None,
 
3411
             starting_with=None,
 
3412
             runner_class=None,
 
3413
             suite_decorators=None,
 
3414
             stream=None,
 
3415
             lsprof_tests=False,
 
3416
             ):
1356
3417
    """Run the whole test suite under the enhanced runner"""
1357
3418
    # XXX: Very ugly way to do this...
1358
3419
    # Disable warning about old formats because we don't want it to disturb
1365
3426
        transport = default_transport
1366
3427
    old_transport = default_transport
1367
3428
    default_transport = transport
 
3429
    global selftest_debug_flags
 
3430
    old_debug_flags = selftest_debug_flags
 
3431
    if debug_flags is not None:
 
3432
        selftest_debug_flags = set(debug_flags)
1368
3433
    try:
 
3434
        if load_list is None:
 
3435
            keep_only = None
 
3436
        else:
 
3437
            keep_only = load_test_id_list(load_list)
 
3438
        if starting_with:
 
3439
            starting_with = [test_prefix_alias_registry.resolve_alias(start)
 
3440
                             for start in starting_with]
1369
3441
        if test_suite_factory is None:
1370
 
            suite = test_suite()
 
3442
            # Reduce loading time by loading modules based on the starting_with
 
3443
            # patterns.
 
3444
            suite = test_suite(keep_only, starting_with)
1371
3445
        else:
1372
3446
            suite = test_suite_factory()
 
3447
        if starting_with:
 
3448
            # But always filter as requested.
 
3449
            suite = filter_suite_by_id_startswith(suite, starting_with)
 
3450
        result_decorators = []
 
3451
        if lsprof_tests:
 
3452
            result_decorators.append(ProfileResult)
1373
3453
        return run_suite(suite, 'testbzr', verbose=verbose, pattern=pattern,
1374
 
                     stop_on_failure=stop_on_failure, keep_output=keep_output,
 
3454
                     stop_on_failure=stop_on_failure,
1375
3455
                     transport=transport,
1376
3456
                     lsprof_timed=lsprof_timed,
1377
 
                     bench_history=bench_history)
 
3457
                     bench_history=bench_history,
 
3458
                     matching_tests_first=matching_tests_first,
 
3459
                     list_only=list_only,
 
3460
                     random_seed=random_seed,
 
3461
                     exclude_pattern=exclude_pattern,
 
3462
                     strict=strict,
 
3463
                     runner_class=runner_class,
 
3464
                     suite_decorators=suite_decorators,
 
3465
                     stream=stream,
 
3466
                     result_decorators=result_decorators,
 
3467
                     )
1378
3468
    finally:
1379
3469
        default_transport = old_transport
1380
 
 
1381
 
 
1382
 
def test_suite():
 
3470
        selftest_debug_flags = old_debug_flags
 
3471
 
 
3472
 
 
3473
def load_test_id_list(file_name):
 
3474
    """Load a test id list from a text file.
 
3475
 
 
3476
    The format is one test id by line.  No special care is taken to impose
 
3477
    strict rules, these test ids are used to filter the test suite so a test id
 
3478
    that do not match an existing test will do no harm. This allows user to add
 
3479
    comments, leave blank lines, etc.
 
3480
    """
 
3481
    test_list = []
 
3482
    try:
 
3483
        ftest = open(file_name, 'rt')
 
3484
    except IOError, e:
 
3485
        if e.errno != errno.ENOENT:
 
3486
            raise
 
3487
        else:
 
3488
            raise errors.NoSuchFile(file_name)
 
3489
 
 
3490
    for test_name in ftest.readlines():
 
3491
        test_list.append(test_name.strip())
 
3492
    ftest.close()
 
3493
    return test_list
 
3494
 
 
3495
 
 
3496
def suite_matches_id_list(test_suite, id_list):
 
3497
    """Warns about tests not appearing or appearing more than once.
 
3498
 
 
3499
    :param test_suite: A TestSuite object.
 
3500
    :param test_id_list: The list of test ids that should be found in
 
3501
         test_suite.
 
3502
 
 
3503
    :return: (absents, duplicates) absents is a list containing the test found
 
3504
        in id_list but not in test_suite, duplicates is a list containing the
 
3505
        test found multiple times in test_suite.
 
3506
 
 
3507
    When using a prefined test id list, it may occurs that some tests do not
 
3508
    exist anymore or that some tests use the same id. This function warns the
 
3509
    tester about potential problems in his workflow (test lists are volatile)
 
3510
    or in the test suite itself (using the same id for several tests does not
 
3511
    help to localize defects).
 
3512
    """
 
3513
    # Build a dict counting id occurrences
 
3514
    tests = dict()
 
3515
    for test in iter_suite_tests(test_suite):
 
3516
        id = test.id()
 
3517
        tests[id] = tests.get(id, 0) + 1
 
3518
 
 
3519
    not_found = []
 
3520
    duplicates = []
 
3521
    for id in id_list:
 
3522
        occurs = tests.get(id, 0)
 
3523
        if not occurs:
 
3524
            not_found.append(id)
 
3525
        elif occurs > 1:
 
3526
            duplicates.append(id)
 
3527
 
 
3528
    return not_found, duplicates
 
3529
 
 
3530
 
 
3531
class TestIdList(object):
 
3532
    """Test id list to filter a test suite.
 
3533
 
 
3534
    Relying on the assumption that test ids are built as:
 
3535
    <module>[.<class>.<method>][(<param>+)], <module> being in python dotted
 
3536
    notation, this class offers methods to :
 
3537
    - avoid building a test suite for modules not refered to in the test list,
 
3538
    - keep only the tests listed from the module test suite.
 
3539
    """
 
3540
 
 
3541
    def __init__(self, test_id_list):
 
3542
        # When a test suite needs to be filtered against us we compare test ids
 
3543
        # for equality, so a simple dict offers a quick and simple solution.
 
3544
        self.tests = dict().fromkeys(test_id_list, True)
 
3545
 
 
3546
        # While unittest.TestCase have ids like:
 
3547
        # <module>.<class>.<method>[(<param+)],
 
3548
        # doctest.DocTestCase can have ids like:
 
3549
        # <module>
 
3550
        # <module>.<class>
 
3551
        # <module>.<function>
 
3552
        # <module>.<class>.<method>
 
3553
 
 
3554
        # Since we can't predict a test class from its name only, we settle on
 
3555
        # a simple constraint: a test id always begins with its module name.
 
3556
 
 
3557
        modules = {}
 
3558
        for test_id in test_id_list:
 
3559
            parts = test_id.split('.')
 
3560
            mod_name = parts.pop(0)
 
3561
            modules[mod_name] = True
 
3562
            for part in parts:
 
3563
                mod_name += '.' + part
 
3564
                modules[mod_name] = True
 
3565
        self.modules = modules
 
3566
 
 
3567
    def refers_to(self, module_name):
 
3568
        """Is there tests for the module or one of its sub modules."""
 
3569
        return self.modules.has_key(module_name)
 
3570
 
 
3571
    def includes(self, test_id):
 
3572
        return self.tests.has_key(test_id)
 
3573
 
 
3574
 
 
3575
class TestPrefixAliasRegistry(registry.Registry):
 
3576
    """A registry for test prefix aliases.
 
3577
 
 
3578
    This helps implement shorcuts for the --starting-with selftest
 
3579
    option. Overriding existing prefixes is not allowed but not fatal (a
 
3580
    warning will be emitted).
 
3581
    """
 
3582
 
 
3583
    def register(self, key, obj, help=None, info=None,
 
3584
                 override_existing=False):
 
3585
        """See Registry.register.
 
3586
 
 
3587
        Trying to override an existing alias causes a warning to be emitted,
 
3588
        not a fatal execption.
 
3589
        """
 
3590
        try:
 
3591
            super(TestPrefixAliasRegistry, self).register(
 
3592
                key, obj, help=help, info=info, override_existing=False)
 
3593
        except KeyError:
 
3594
            actual = self.get(key)
 
3595
            note('Test prefix alias %s is already used for %s, ignoring %s'
 
3596
                 % (key, actual, obj))
 
3597
 
 
3598
    def resolve_alias(self, id_start):
 
3599
        """Replace the alias by the prefix in the given string.
 
3600
 
 
3601
        Using an unknown prefix is an error to help catching typos.
 
3602
        """
 
3603
        parts = id_start.split('.')
 
3604
        try:
 
3605
            parts[0] = self.get(parts[0])
 
3606
        except KeyError:
 
3607
            raise errors.BzrCommandError(
 
3608
                '%s is not a known test prefix alias' % parts[0])
 
3609
        return '.'.join(parts)
 
3610
 
 
3611
 
 
3612
test_prefix_alias_registry = TestPrefixAliasRegistry()
 
3613
"""Registry of test prefix aliases."""
 
3614
 
 
3615
 
 
3616
# This alias allows to detect typos ('bzrlin.') by making all valid test ids
 
3617
# appear prefixed ('bzrlib.' is "replaced" by 'bzrlib.').
 
3618
test_prefix_alias_registry.register('bzrlib', 'bzrlib')
 
3619
 
 
3620
# Obvious highest levels prefixes, feel free to add your own via a plugin
 
3621
test_prefix_alias_registry.register('bd', 'bzrlib.doc')
 
3622
test_prefix_alias_registry.register('bu', 'bzrlib.utils')
 
3623
test_prefix_alias_registry.register('bt', 'bzrlib.tests')
 
3624
test_prefix_alias_registry.register('bb', 'bzrlib.tests.blackbox')
 
3625
test_prefix_alias_registry.register('bp', 'bzrlib.plugins')
 
3626
 
 
3627
 
 
3628
def _test_suite_testmod_names():
 
3629
    """Return the standard list of test module names to test."""
 
3630
    return [
 
3631
        'bzrlib.doc',
 
3632
        'bzrlib.tests.blackbox',
 
3633
        'bzrlib.tests.commands',
 
3634
        'bzrlib.tests.per_branch',
 
3635
        'bzrlib.tests.per_bzrdir',
 
3636
        'bzrlib.tests.per_bzrdir_colo',
 
3637
        'bzrlib.tests.per_foreign_vcs',
 
3638
        'bzrlib.tests.per_interrepository',
 
3639
        'bzrlib.tests.per_intertree',
 
3640
        'bzrlib.tests.per_inventory',
 
3641
        'bzrlib.tests.per_interbranch',
 
3642
        'bzrlib.tests.per_lock',
 
3643
        'bzrlib.tests.per_merger',
 
3644
        'bzrlib.tests.per_transport',
 
3645
        'bzrlib.tests.per_tree',
 
3646
        'bzrlib.tests.per_pack_repository',
 
3647
        'bzrlib.tests.per_repository',
 
3648
        'bzrlib.tests.per_repository_chk',
 
3649
        'bzrlib.tests.per_repository_reference',
 
3650
        'bzrlib.tests.per_uifactory',
 
3651
        'bzrlib.tests.per_versionedfile',
 
3652
        'bzrlib.tests.per_workingtree',
 
3653
        'bzrlib.tests.test__annotator',
 
3654
        'bzrlib.tests.test__bencode',
 
3655
        'bzrlib.tests.test__chk_map',
 
3656
        'bzrlib.tests.test__dirstate_helpers',
 
3657
        'bzrlib.tests.test__groupcompress',
 
3658
        'bzrlib.tests.test__known_graph',
 
3659
        'bzrlib.tests.test__rio',
 
3660
        'bzrlib.tests.test__simple_set',
 
3661
        'bzrlib.tests.test__static_tuple',
 
3662
        'bzrlib.tests.test__walkdirs_win32',
 
3663
        'bzrlib.tests.test_ancestry',
 
3664
        'bzrlib.tests.test_annotate',
 
3665
        'bzrlib.tests.test_api',
 
3666
        'bzrlib.tests.test_atomicfile',
 
3667
        'bzrlib.tests.test_bad_files',
 
3668
        'bzrlib.tests.test_bisect_multi',
 
3669
        'bzrlib.tests.test_branch',
 
3670
        'bzrlib.tests.test_branchbuilder',
 
3671
        'bzrlib.tests.test_btree_index',
 
3672
        'bzrlib.tests.test_bugtracker',
 
3673
        'bzrlib.tests.test_bundle',
 
3674
        'bzrlib.tests.test_bzrdir',
 
3675
        'bzrlib.tests.test__chunks_to_lines',
 
3676
        'bzrlib.tests.test_cache_utf8',
 
3677
        'bzrlib.tests.test_chk_map',
 
3678
        'bzrlib.tests.test_chk_serializer',
 
3679
        'bzrlib.tests.test_chunk_writer',
 
3680
        'bzrlib.tests.test_clean_tree',
 
3681
        'bzrlib.tests.test_cleanup',
 
3682
        'bzrlib.tests.test_cmdline',
 
3683
        'bzrlib.tests.test_commands',
 
3684
        'bzrlib.tests.test_commit',
 
3685
        'bzrlib.tests.test_commit_merge',
 
3686
        'bzrlib.tests.test_config',
 
3687
        'bzrlib.tests.test_conflicts',
 
3688
        'bzrlib.tests.test_counted_lock',
 
3689
        'bzrlib.tests.test_crash',
 
3690
        'bzrlib.tests.test_decorators',
 
3691
        'bzrlib.tests.test_delta',
 
3692
        'bzrlib.tests.test_debug',
 
3693
        'bzrlib.tests.test_deprecated_graph',
 
3694
        'bzrlib.tests.test_diff',
 
3695
        'bzrlib.tests.test_directory_service',
 
3696
        'bzrlib.tests.test_dirstate',
 
3697
        'bzrlib.tests.test_email_message',
 
3698
        'bzrlib.tests.test_eol_filters',
 
3699
        'bzrlib.tests.test_errors',
 
3700
        'bzrlib.tests.test_export',
 
3701
        'bzrlib.tests.test_extract',
 
3702
        'bzrlib.tests.test_fetch',
 
3703
        'bzrlib.tests.test_fifo_cache',
 
3704
        'bzrlib.tests.test_filters',
 
3705
        'bzrlib.tests.test_ftp_transport',
 
3706
        'bzrlib.tests.test_foreign',
 
3707
        'bzrlib.tests.test_generate_docs',
 
3708
        'bzrlib.tests.test_generate_ids',
 
3709
        'bzrlib.tests.test_globbing',
 
3710
        'bzrlib.tests.test_gpg',
 
3711
        'bzrlib.tests.test_graph',
 
3712
        'bzrlib.tests.test_groupcompress',
 
3713
        'bzrlib.tests.test_hashcache',
 
3714
        'bzrlib.tests.test_help',
 
3715
        'bzrlib.tests.test_hooks',
 
3716
        'bzrlib.tests.test_http',
 
3717
        'bzrlib.tests.test_http_response',
 
3718
        'bzrlib.tests.test_https_ca_bundle',
 
3719
        'bzrlib.tests.test_identitymap',
 
3720
        'bzrlib.tests.test_ignores',
 
3721
        'bzrlib.tests.test_index',
 
3722
        'bzrlib.tests.test_import_tariff',
 
3723
        'bzrlib.tests.test_info',
 
3724
        'bzrlib.tests.test_inv',
 
3725
        'bzrlib.tests.test_inventory_delta',
 
3726
        'bzrlib.tests.test_knit',
 
3727
        'bzrlib.tests.test_lazy_import',
 
3728
        'bzrlib.tests.test_lazy_regex',
 
3729
        'bzrlib.tests.test_lock',
 
3730
        'bzrlib.tests.test_lockable_files',
 
3731
        'bzrlib.tests.test_lockdir',
 
3732
        'bzrlib.tests.test_log',
 
3733
        'bzrlib.tests.test_lru_cache',
 
3734
        'bzrlib.tests.test_lsprof',
 
3735
        'bzrlib.tests.test_mail_client',
 
3736
        'bzrlib.tests.test_matchers',
 
3737
        'bzrlib.tests.test_memorytree',
 
3738
        'bzrlib.tests.test_merge',
 
3739
        'bzrlib.tests.test_merge3',
 
3740
        'bzrlib.tests.test_merge_core',
 
3741
        'bzrlib.tests.test_merge_directive',
 
3742
        'bzrlib.tests.test_missing',
 
3743
        'bzrlib.tests.test_msgeditor',
 
3744
        'bzrlib.tests.test_multiparent',
 
3745
        'bzrlib.tests.test_mutabletree',
 
3746
        'bzrlib.tests.test_nonascii',
 
3747
        'bzrlib.tests.test_options',
 
3748
        'bzrlib.tests.test_osutils',
 
3749
        'bzrlib.tests.test_osutils_encodings',
 
3750
        'bzrlib.tests.test_pack',
 
3751
        'bzrlib.tests.test_patch',
 
3752
        'bzrlib.tests.test_patches',
 
3753
        'bzrlib.tests.test_permissions',
 
3754
        'bzrlib.tests.test_plugins',
 
3755
        'bzrlib.tests.test_progress',
 
3756
        'bzrlib.tests.test_read_bundle',
 
3757
        'bzrlib.tests.test_reconcile',
 
3758
        'bzrlib.tests.test_reconfigure',
 
3759
        'bzrlib.tests.test_registry',
 
3760
        'bzrlib.tests.test_remote',
 
3761
        'bzrlib.tests.test_rename_map',
 
3762
        'bzrlib.tests.test_repository',
 
3763
        'bzrlib.tests.test_revert',
 
3764
        'bzrlib.tests.test_revision',
 
3765
        'bzrlib.tests.test_revisionspec',
 
3766
        'bzrlib.tests.test_revisiontree',
 
3767
        'bzrlib.tests.test_rio',
 
3768
        'bzrlib.tests.test_rules',
 
3769
        'bzrlib.tests.test_sampler',
 
3770
        'bzrlib.tests.test_script',
 
3771
        'bzrlib.tests.test_selftest',
 
3772
        'bzrlib.tests.test_serializer',
 
3773
        'bzrlib.tests.test_setup',
 
3774
        'bzrlib.tests.test_sftp_transport',
 
3775
        'bzrlib.tests.test_shelf',
 
3776
        'bzrlib.tests.test_shelf_ui',
 
3777
        'bzrlib.tests.test_smart',
 
3778
        'bzrlib.tests.test_smart_add',
 
3779
        'bzrlib.tests.test_smart_request',
 
3780
        'bzrlib.tests.test_smart_transport',
 
3781
        'bzrlib.tests.test_smtp_connection',
 
3782
        'bzrlib.tests.test_source',
 
3783
        'bzrlib.tests.test_ssh_transport',
 
3784
        'bzrlib.tests.test_status',
 
3785
        'bzrlib.tests.test_store',
 
3786
        'bzrlib.tests.test_strace',
 
3787
        'bzrlib.tests.test_subsume',
 
3788
        'bzrlib.tests.test_switch',
 
3789
        'bzrlib.tests.test_symbol_versioning',
 
3790
        'bzrlib.tests.test_tag',
 
3791
        'bzrlib.tests.test_testament',
 
3792
        'bzrlib.tests.test_textfile',
 
3793
        'bzrlib.tests.test_textmerge',
 
3794
        'bzrlib.tests.test_timestamp',
 
3795
        'bzrlib.tests.test_trace',
 
3796
        'bzrlib.tests.test_transactions',
 
3797
        'bzrlib.tests.test_transform',
 
3798
        'bzrlib.tests.test_transport',
 
3799
        'bzrlib.tests.test_transport_log',
 
3800
        'bzrlib.tests.test_tree',
 
3801
        'bzrlib.tests.test_treebuilder',
 
3802
        'bzrlib.tests.test_tsort',
 
3803
        'bzrlib.tests.test_tuned_gzip',
 
3804
        'bzrlib.tests.test_ui',
 
3805
        'bzrlib.tests.test_uncommit',
 
3806
        'bzrlib.tests.test_upgrade',
 
3807
        'bzrlib.tests.test_upgrade_stacked',
 
3808
        'bzrlib.tests.test_urlutils',
 
3809
        'bzrlib.tests.test_version',
 
3810
        'bzrlib.tests.test_version_info',
 
3811
        'bzrlib.tests.test_weave',
 
3812
        'bzrlib.tests.test_whitebox',
 
3813
        'bzrlib.tests.test_win32utils',
 
3814
        'bzrlib.tests.test_workingtree',
 
3815
        'bzrlib.tests.test_workingtree_4',
 
3816
        'bzrlib.tests.test_wsgi',
 
3817
        'bzrlib.tests.test_xml',
 
3818
        ]
 
3819
 
 
3820
 
 
3821
def _test_suite_modules_to_doctest():
 
3822
    """Return the list of modules to doctest."""
 
3823
    if __doc__ is None:
 
3824
        # GZ 2009-03-31: No docstrings with -OO so there's nothing to doctest
 
3825
        return []
 
3826
    return [
 
3827
        'bzrlib',
 
3828
        'bzrlib.branchbuilder',
 
3829
        'bzrlib.decorators',
 
3830
        'bzrlib.export',
 
3831
        'bzrlib.inventory',
 
3832
        'bzrlib.iterablefile',
 
3833
        'bzrlib.lockdir',
 
3834
        'bzrlib.merge3',
 
3835
        'bzrlib.option',
 
3836
        'bzrlib.symbol_versioning',
 
3837
        'bzrlib.tests',
 
3838
        'bzrlib.timestamp',
 
3839
        'bzrlib.version_info_formats.format_custom',
 
3840
        ]
 
3841
 
 
3842
 
 
3843
def test_suite(keep_only=None, starting_with=None):
1383
3844
    """Build and return TestSuite for the whole of bzrlib.
1384
 
    
 
3845
 
 
3846
    :param keep_only: A list of test ids limiting the suite returned.
 
3847
 
 
3848
    :param starting_with: An id limiting the suite returned to the tests
 
3849
         starting with it.
 
3850
 
1385
3851
    This function can be replaced if you need to change the default test
1386
3852
    suite on a global basis, but it is not encouraged.
1387
3853
    """
1388
 
    testmod_names = [
1389
 
                   'bzrlib.tests.test_ancestry',
1390
 
                   'bzrlib.tests.test_api',
1391
 
                   'bzrlib.tests.test_atomicfile',
1392
 
                   'bzrlib.tests.test_bad_files',
1393
 
                   'bzrlib.tests.test_branch',
1394
 
                   'bzrlib.tests.test_bundle',
1395
 
                   'bzrlib.tests.test_bzrdir',
1396
 
                   'bzrlib.tests.test_cache_utf8',
1397
 
                   'bzrlib.tests.test_command',
1398
 
                   'bzrlib.tests.test_commit',
1399
 
                   'bzrlib.tests.test_commit_merge',
1400
 
                   'bzrlib.tests.test_config',
1401
 
                   'bzrlib.tests.test_conflicts',
1402
 
                   'bzrlib.tests.test_decorators',
1403
 
                   'bzrlib.tests.test_diff',
1404
 
                   'bzrlib.tests.test_doc_generate',
1405
 
                   'bzrlib.tests.test_errors',
1406
 
                   'bzrlib.tests.test_escaped_store',
1407
 
                   'bzrlib.tests.test_fetch',
1408
 
                   'bzrlib.tests.test_gpg',
1409
 
                   'bzrlib.tests.test_graph',
1410
 
                   'bzrlib.tests.test_hashcache',
1411
 
                   'bzrlib.tests.test_http',
1412
 
                   'bzrlib.tests.test_http_response',
1413
 
                   'bzrlib.tests.test_identitymap',
1414
 
                   'bzrlib.tests.test_ignores',
1415
 
                   'bzrlib.tests.test_inv',
1416
 
                   'bzrlib.tests.test_knit',
1417
 
                   'bzrlib.tests.test_lockdir',
1418
 
                   'bzrlib.tests.test_lockable_files',
1419
 
                   'bzrlib.tests.test_log',
1420
 
                   'bzrlib.tests.test_merge',
1421
 
                   'bzrlib.tests.test_merge3',
1422
 
                   'bzrlib.tests.test_merge_core',
1423
 
                   'bzrlib.tests.test_missing',
1424
 
                   'bzrlib.tests.test_msgeditor',
1425
 
                   'bzrlib.tests.test_nonascii',
1426
 
                   'bzrlib.tests.test_options',
1427
 
                   'bzrlib.tests.test_osutils',
1428
 
                   'bzrlib.tests.test_patch',
1429
 
                   'bzrlib.tests.test_patches',
1430
 
                   'bzrlib.tests.test_permissions',
1431
 
                   'bzrlib.tests.test_plugins',
1432
 
                   'bzrlib.tests.test_progress',
1433
 
                   'bzrlib.tests.test_reconcile',
1434
 
                   'bzrlib.tests.test_repository',
1435
 
                   'bzrlib.tests.test_revert',
1436
 
                   'bzrlib.tests.test_revision',
1437
 
                   'bzrlib.tests.test_revisionnamespaces',
1438
 
                   'bzrlib.tests.test_revisiontree',
1439
 
                   'bzrlib.tests.test_rio',
1440
 
                   'bzrlib.tests.test_sampler',
1441
 
                   'bzrlib.tests.test_selftest',
1442
 
                   'bzrlib.tests.test_setup',
1443
 
                   'bzrlib.tests.test_sftp_transport',
1444
 
                   'bzrlib.tests.test_ftp_transport',
1445
 
                   'bzrlib.tests.test_smart_add',
1446
 
                   'bzrlib.tests.test_source',
1447
 
                   'bzrlib.tests.test_status',
1448
 
                   'bzrlib.tests.test_store',
1449
 
                   'bzrlib.tests.test_symbol_versioning',
1450
 
                   'bzrlib.tests.test_testament',
1451
 
                   'bzrlib.tests.test_textfile',
1452
 
                   'bzrlib.tests.test_textmerge',
1453
 
                   'bzrlib.tests.test_trace',
1454
 
                   'bzrlib.tests.test_transactions',
1455
 
                   'bzrlib.tests.test_transform',
1456
 
                   'bzrlib.tests.test_transport',
1457
 
                   'bzrlib.tests.test_tree',
1458
 
                   'bzrlib.tests.test_tsort',
1459
 
                   'bzrlib.tests.test_tuned_gzip',
1460
 
                   'bzrlib.tests.test_ui',
1461
 
                   'bzrlib.tests.test_upgrade',
1462
 
                   'bzrlib.tests.test_urlutils',
1463
 
                   'bzrlib.tests.test_versionedfile',
1464
 
                   'bzrlib.tests.test_version',
1465
 
                   'bzrlib.tests.test_weave',
1466
 
                   'bzrlib.tests.test_whitebox',
1467
 
                   'bzrlib.tests.test_workingtree',
1468
 
                   'bzrlib.tests.test_xml',
1469
 
                   ]
1470
 
    test_transport_implementations = [
1471
 
        'bzrlib.tests.test_transport_implementations',
1472
 
        'bzrlib.tests.test_read_bundle',
1473
 
        ]
1474
 
    suite = TestUtil.TestSuite()
 
3854
 
1475
3855
    loader = TestUtil.TestLoader()
1476
 
    suite.addTest(loader.loadTestsFromModuleNames(testmod_names))
1477
 
    from bzrlib.transport import TransportTestProviderAdapter
1478
 
    adapter = TransportTestProviderAdapter()
1479
 
    adapt_modules(test_transport_implementations, adapter, loader, suite)
1480
 
    for package in packages_to_test():
1481
 
        suite.addTest(package.test_suite())
1482
 
    for m in MODULES_TO_TEST:
1483
 
        suite.addTest(loader.loadTestsFromModule(m))
1484
 
    for m in MODULES_TO_DOCTEST:
1485
 
        suite.addTest(doctest.DocTestSuite(m))
1486
 
    for name, plugin in bzrlib.plugin.all_plugins().items():
1487
 
        if getattr(plugin, 'test_suite', None) is not None:
1488
 
            suite.addTest(plugin.test_suite())
 
3856
 
 
3857
    if keep_only is not None:
 
3858
        id_filter = TestIdList(keep_only)
 
3859
    if starting_with:
 
3860
        # We take precedence over keep_only because *at loading time* using
 
3861
        # both options means we will load less tests for the same final result.
 
3862
        def interesting_module(name):
 
3863
            for start in starting_with:
 
3864
                if (
 
3865
                    # Either the module name starts with the specified string
 
3866
                    name.startswith(start)
 
3867
                    # or it may contain tests starting with the specified string
 
3868
                    or start.startswith(name)
 
3869
                    ):
 
3870
                    return True
 
3871
            return False
 
3872
        loader = TestUtil.FilteredByModuleTestLoader(interesting_module)
 
3873
 
 
3874
    elif keep_only is not None:
 
3875
        loader = TestUtil.FilteredByModuleTestLoader(id_filter.refers_to)
 
3876
        def interesting_module(name):
 
3877
            return id_filter.refers_to(name)
 
3878
 
 
3879
    else:
 
3880
        loader = TestUtil.TestLoader()
 
3881
        def interesting_module(name):
 
3882
            # No filtering, all modules are interesting
 
3883
            return True
 
3884
 
 
3885
    suite = loader.suiteClass()
 
3886
 
 
3887
    # modules building their suite with loadTestsFromModuleNames
 
3888
    suite.addTest(loader.loadTestsFromModuleNames(_test_suite_testmod_names()))
 
3889
 
 
3890
    for mod in _test_suite_modules_to_doctest():
 
3891
        if not interesting_module(mod):
 
3892
            # No tests to keep here, move along
 
3893
            continue
 
3894
        try:
 
3895
            # note that this really does mean "report only" -- doctest
 
3896
            # still runs the rest of the examples
 
3897
            doc_suite = doctest.DocTestSuite(mod,
 
3898
                optionflags=doctest.REPORT_ONLY_FIRST_FAILURE)
 
3899
        except ValueError, e:
 
3900
            print '**failed to get doctest for: %s\n%s' % (mod, e)
 
3901
            raise
 
3902
        if len(doc_suite._tests) == 0:
 
3903
            raise errors.BzrError("no doctests found in %s" % (mod,))
 
3904
        suite.addTest(doc_suite)
 
3905
 
 
3906
    default_encoding = sys.getdefaultencoding()
 
3907
    for name, plugin in bzrlib.plugin.plugins().items():
 
3908
        if not interesting_module(plugin.module.__name__):
 
3909
            continue
 
3910
        plugin_suite = plugin.test_suite()
 
3911
        # We used to catch ImportError here and turn it into just a warning,
 
3912
        # but really if you don't have --no-plugins this should be a failure.
 
3913
        # mbp 20080213 - see http://bugs.launchpad.net/bugs/189771
 
3914
        if plugin_suite is None:
 
3915
            plugin_suite = plugin.load_plugin_tests(loader)
 
3916
        if plugin_suite is not None:
 
3917
            suite.addTest(plugin_suite)
 
3918
        if default_encoding != sys.getdefaultencoding():
 
3919
            bzrlib.trace.warning(
 
3920
                'Plugin "%s" tried to reset default encoding to: %s', name,
 
3921
                sys.getdefaultencoding())
 
3922
            reload(sys)
 
3923
            sys.setdefaultencoding(default_encoding)
 
3924
 
 
3925
    if keep_only is not None:
 
3926
        # Now that the referred modules have loaded their tests, keep only the
 
3927
        # requested ones.
 
3928
        suite = filter_suite_by_id_list(suite, id_filter)
 
3929
        # Do some sanity checks on the id_list filtering
 
3930
        not_found, duplicates = suite_matches_id_list(suite, keep_only)
 
3931
        if starting_with:
 
3932
            # The tester has used both keep_only and starting_with, so he is
 
3933
            # already aware that some tests are excluded from the list, there
 
3934
            # is no need to tell him which.
 
3935
            pass
 
3936
        else:
 
3937
            # Some tests mentioned in the list are not in the test suite. The
 
3938
            # list may be out of date, report to the tester.
 
3939
            for id in not_found:
 
3940
                bzrlib.trace.warning('"%s" not found in the test suite', id)
 
3941
        for id in duplicates:
 
3942
            bzrlib.trace.warning('"%s" is used as an id by several tests', id)
 
3943
 
1489
3944
    return suite
1490
3945
 
1491
3946
 
1492
 
def adapt_modules(mods_list, adapter, loader, suite):
1493
 
    """Adapt the modules in mods_list using adapter and add to suite."""
1494
 
    for test in iter_suite_tests(loader.loadTestsFromModuleNames(mods_list)):
1495
 
        suite.addTests(adapter.adapt(test))
 
3947
def multiply_scenarios(scenarios_left, scenarios_right):
 
3948
    """Multiply two sets of scenarios.
 
3949
 
 
3950
    :returns: the cartesian product of the two sets of scenarios, that is
 
3951
        a scenario for every possible combination of a left scenario and a
 
3952
        right scenario.
 
3953
    """
 
3954
    return [
 
3955
        ('%s,%s' % (left_name, right_name),
 
3956
         dict(left_dict.items() + right_dict.items()))
 
3957
        for left_name, left_dict in scenarios_left
 
3958
        for right_name, right_dict in scenarios_right]
 
3959
 
 
3960
 
 
3961
def multiply_tests(tests, scenarios, result):
 
3962
    """Multiply tests_list by scenarios into result.
 
3963
 
 
3964
    This is the core workhorse for test parameterisation.
 
3965
 
 
3966
    Typically the load_tests() method for a per-implementation test suite will
 
3967
    call multiply_tests and return the result.
 
3968
 
 
3969
    :param tests: The tests to parameterise.
 
3970
    :param scenarios: The scenarios to apply: pairs of (scenario_name,
 
3971
        scenario_param_dict).
 
3972
    :param result: A TestSuite to add created tests to.
 
3973
 
 
3974
    This returns the passed in result TestSuite with the cross product of all
 
3975
    the tests repeated once for each scenario.  Each test is adapted by adding
 
3976
    the scenario name at the end of its id(), and updating the test object's
 
3977
    __dict__ with the scenario_param_dict.
 
3978
 
 
3979
    >>> import bzrlib.tests.test_sampler
 
3980
    >>> r = multiply_tests(
 
3981
    ...     bzrlib.tests.test_sampler.DemoTest('test_nothing'),
 
3982
    ...     [('one', dict(param=1)),
 
3983
    ...      ('two', dict(param=2))],
 
3984
    ...     TestSuite())
 
3985
    >>> tests = list(iter_suite_tests(r))
 
3986
    >>> len(tests)
 
3987
    2
 
3988
    >>> tests[0].id()
 
3989
    'bzrlib.tests.test_sampler.DemoTest.test_nothing(one)'
 
3990
    >>> tests[0].param
 
3991
    1
 
3992
    >>> tests[1].param
 
3993
    2
 
3994
    """
 
3995
    for test in iter_suite_tests(tests):
 
3996
        apply_scenarios(test, scenarios, result)
 
3997
    return result
 
3998
 
 
3999
 
 
4000
def apply_scenarios(test, scenarios, result):
 
4001
    """Apply the scenarios in scenarios to test and add to result.
 
4002
 
 
4003
    :param test: The test to apply scenarios to.
 
4004
    :param scenarios: An iterable of scenarios to apply to test.
 
4005
    :return: result
 
4006
    :seealso: apply_scenario
 
4007
    """
 
4008
    for scenario in scenarios:
 
4009
        result.addTest(apply_scenario(test, scenario))
 
4010
    return result
 
4011
 
 
4012
 
 
4013
def apply_scenario(test, scenario):
 
4014
    """Copy test and apply scenario to it.
 
4015
 
 
4016
    :param test: A test to adapt.
 
4017
    :param scenario: A tuple describing the scenarion.
 
4018
        The first element of the tuple is the new test id.
 
4019
        The second element is a dict containing attributes to set on the
 
4020
        test.
 
4021
    :return: The adapted test.
 
4022
    """
 
4023
    new_id = "%s(%s)" % (test.id(), scenario[0])
 
4024
    new_test = clone_test(test, new_id)
 
4025
    for name, value in scenario[1].items():
 
4026
        setattr(new_test, name, value)
 
4027
    return new_test
 
4028
 
 
4029
 
 
4030
def clone_test(test, new_id):
 
4031
    """Clone a test giving it a new id.
 
4032
 
 
4033
    :param test: The test to clone.
 
4034
    :param new_id: The id to assign to it.
 
4035
    :return: The new test.
 
4036
    """
 
4037
    new_test = copy(test)
 
4038
    new_test.id = lambda: new_id
 
4039
    return new_test
 
4040
 
 
4041
 
 
4042
def permute_tests_for_extension(standard_tests, loader, py_module_name,
 
4043
                                ext_module_name):
 
4044
    """Helper for permutating tests against an extension module.
 
4045
 
 
4046
    This is meant to be used inside a modules 'load_tests()' function. It will
 
4047
    create 2 scenarios, and cause all tests in the 'standard_tests' to be run
 
4048
    against both implementations. Setting 'test.module' to the appropriate
 
4049
    module. See bzrlib.tests.test__chk_map.load_tests as an example.
 
4050
 
 
4051
    :param standard_tests: A test suite to permute
 
4052
    :param loader: A TestLoader
 
4053
    :param py_module_name: The python path to a python module that can always
 
4054
        be loaded, and will be considered the 'python' implementation. (eg
 
4055
        'bzrlib._chk_map_py')
 
4056
    :param ext_module_name: The python path to an extension module. If the
 
4057
        module cannot be loaded, a single test will be added, which notes that
 
4058
        the module is not available. If it can be loaded, all standard_tests
 
4059
        will be run against that module.
 
4060
    :return: (suite, feature) suite is a test-suite that has all the permuted
 
4061
        tests. feature is the Feature object that can be used to determine if
 
4062
        the module is available.
 
4063
    """
 
4064
 
 
4065
    py_module = __import__(py_module_name, {}, {}, ['NO_SUCH_ATTRIB'])
 
4066
    scenarios = [
 
4067
        ('python', {'module': py_module}),
 
4068
    ]
 
4069
    suite = loader.suiteClass()
 
4070
    feature = ModuleAvailableFeature(ext_module_name)
 
4071
    if feature.available():
 
4072
        scenarios.append(('C', {'module': feature.module}))
 
4073
    else:
 
4074
        # the compiled module isn't available, so we add a failing test
 
4075
        class FailWithoutFeature(TestCase):
 
4076
            def test_fail(self):
 
4077
                self.requireFeature(feature)
 
4078
        suite.addTest(loader.loadTestsFromTestCase(FailWithoutFeature))
 
4079
    result = multiply_tests(standard_tests, scenarios, suite)
 
4080
    return result, feature
 
4081
 
 
4082
 
 
4083
def _rmtree_temp_dir(dirname, test_id=None):
 
4084
    # If LANG=C we probably have created some bogus paths
 
4085
    # which rmtree(unicode) will fail to delete
 
4086
    # so make sure we are using rmtree(str) to delete everything
 
4087
    # except on win32, where rmtree(str) will fail
 
4088
    # since it doesn't have the property of byte-stream paths
 
4089
    # (they are either ascii or mbcs)
 
4090
    if sys.platform == 'win32':
 
4091
        # make sure we are using the unicode win32 api
 
4092
        dirname = unicode(dirname)
 
4093
    else:
 
4094
        dirname = dirname.encode(sys.getfilesystemencoding())
 
4095
    try:
 
4096
        osutils.rmtree(dirname)
 
4097
    except OSError, e:
 
4098
        # We don't want to fail here because some useful display will be lost
 
4099
        # otherwise. Polluting the tmp dir is bad, but not giving all the
 
4100
        # possible info to the test runner is even worse.
 
4101
        if test_id != None:
 
4102
            ui.ui_factory.clear_term()
 
4103
            sys.stderr.write('\nWhile running: %s\n' % (test_id,))
 
4104
        sys.stderr.write('Unable to remove testing dir %s\n%s'
 
4105
                         % (os.path.basename(dirname), e))
 
4106
 
 
4107
 
 
4108
class Feature(object):
 
4109
    """An operating system Feature."""
 
4110
 
 
4111
    def __init__(self):
 
4112
        self._available = None
 
4113
 
 
4114
    def available(self):
 
4115
        """Is the feature available?
 
4116
 
 
4117
        :return: True if the feature is available.
 
4118
        """
 
4119
        if self._available is None:
 
4120
            self._available = self._probe()
 
4121
        return self._available
 
4122
 
 
4123
    def _probe(self):
 
4124
        """Implement this method in concrete features.
 
4125
 
 
4126
        :return: True if the feature is available.
 
4127
        """
 
4128
        raise NotImplementedError
 
4129
 
 
4130
    def __str__(self):
 
4131
        if getattr(self, 'feature_name', None):
 
4132
            return self.feature_name()
 
4133
        return self.__class__.__name__
 
4134
 
 
4135
 
 
4136
class _SymlinkFeature(Feature):
 
4137
 
 
4138
    def _probe(self):
 
4139
        return osutils.has_symlinks()
 
4140
 
 
4141
    def feature_name(self):
 
4142
        return 'symlinks'
 
4143
 
 
4144
SymlinkFeature = _SymlinkFeature()
 
4145
 
 
4146
 
 
4147
class _HardlinkFeature(Feature):
 
4148
 
 
4149
    def _probe(self):
 
4150
        return osutils.has_hardlinks()
 
4151
 
 
4152
    def feature_name(self):
 
4153
        return 'hardlinks'
 
4154
 
 
4155
HardlinkFeature = _HardlinkFeature()
 
4156
 
 
4157
 
 
4158
class _OsFifoFeature(Feature):
 
4159
 
 
4160
    def _probe(self):
 
4161
        return getattr(os, 'mkfifo', None)
 
4162
 
 
4163
    def feature_name(self):
 
4164
        return 'filesystem fifos'
 
4165
 
 
4166
OsFifoFeature = _OsFifoFeature()
 
4167
 
 
4168
 
 
4169
class _UnicodeFilenameFeature(Feature):
 
4170
    """Does the filesystem support Unicode filenames?"""
 
4171
 
 
4172
    def _probe(self):
 
4173
        try:
 
4174
            # Check for character combinations unlikely to be covered by any
 
4175
            # single non-unicode encoding. We use the characters
 
4176
            # - greek small letter alpha (U+03B1) and
 
4177
            # - braille pattern dots-123456 (U+283F).
 
4178
            os.stat(u'\u03b1\u283f')
 
4179
        except UnicodeEncodeError:
 
4180
            return False
 
4181
        except (IOError, OSError):
 
4182
            # The filesystem allows the Unicode filename but the file doesn't
 
4183
            # exist.
 
4184
            return True
 
4185
        else:
 
4186
            # The filesystem allows the Unicode filename and the file exists,
 
4187
            # for some reason.
 
4188
            return True
 
4189
 
 
4190
UnicodeFilenameFeature = _UnicodeFilenameFeature()
 
4191
 
 
4192
 
 
4193
class _CompatabilityThunkFeature(Feature):
 
4194
    """This feature is just a thunk to another feature.
 
4195
 
 
4196
    It issues a deprecation warning if it is accessed, to let you know that you
 
4197
    should really use a different feature.
 
4198
    """
 
4199
 
 
4200
    def __init__(self, dep_version, module, name,
 
4201
                 replacement_name, replacement_module=None):
 
4202
        super(_CompatabilityThunkFeature, self).__init__()
 
4203
        self._module = module
 
4204
        if replacement_module is None:
 
4205
            replacement_module = module
 
4206
        self._replacement_module = replacement_module
 
4207
        self._name = name
 
4208
        self._replacement_name = replacement_name
 
4209
        self._dep_version = dep_version
 
4210
        self._feature = None
 
4211
 
 
4212
    def _ensure(self):
 
4213
        if self._feature is None:
 
4214
            depr_msg = self._dep_version % ('%s.%s'
 
4215
                                            % (self._module, self._name))
 
4216
            use_msg = ' Use %s.%s instead.' % (self._replacement_module,
 
4217
                                               self._replacement_name)
 
4218
            symbol_versioning.warn(depr_msg + use_msg, DeprecationWarning)
 
4219
            # Import the new feature and use it as a replacement for the
 
4220
            # deprecated one.
 
4221
            mod = __import__(self._replacement_module, {}, {},
 
4222
                             [self._replacement_name])
 
4223
            self._feature = getattr(mod, self._replacement_name)
 
4224
 
 
4225
    def _probe(self):
 
4226
        self._ensure()
 
4227
        return self._feature._probe()
 
4228
 
 
4229
 
 
4230
class ModuleAvailableFeature(Feature):
 
4231
    """This is a feature than describes a module we want to be available.
 
4232
 
 
4233
    Declare the name of the module in __init__(), and then after probing, the
 
4234
    module will be available as 'self.module'.
 
4235
 
 
4236
    :ivar module: The module if it is available, else None.
 
4237
    """
 
4238
 
 
4239
    def __init__(self, module_name):
 
4240
        super(ModuleAvailableFeature, self).__init__()
 
4241
        self.module_name = module_name
 
4242
 
 
4243
    def _probe(self):
 
4244
        try:
 
4245
            self._module = __import__(self.module_name, {}, {}, [''])
 
4246
            return True
 
4247
        except ImportError:
 
4248
            return False
 
4249
 
 
4250
    @property
 
4251
    def module(self):
 
4252
        if self.available(): # Make sure the probe has been done
 
4253
            return self._module
 
4254
        return None
 
4255
 
 
4256
    def feature_name(self):
 
4257
        return self.module_name
 
4258
 
 
4259
 
 
4260
# This is kept here for compatibility, it is recommended to use
 
4261
# 'bzrlib.tests.feature.paramiko' instead
 
4262
ParamikoFeature = _CompatabilityThunkFeature(
 
4263
    deprecated_in((2,1,0)),
 
4264
    'bzrlib.tests.features', 'ParamikoFeature', 'paramiko')
 
4265
 
 
4266
 
 
4267
def probe_unicode_in_user_encoding():
 
4268
    """Try to encode several unicode strings to use in unicode-aware tests.
 
4269
    Return first successfull match.
 
4270
 
 
4271
    :return:  (unicode value, encoded plain string value) or (None, None)
 
4272
    """
 
4273
    possible_vals = [u'm\xb5', u'\xe1', u'\u0410']
 
4274
    for uni_val in possible_vals:
 
4275
        try:
 
4276
            str_val = uni_val.encode(osutils.get_user_encoding())
 
4277
        except UnicodeEncodeError:
 
4278
            # Try a different character
 
4279
            pass
 
4280
        else:
 
4281
            return uni_val, str_val
 
4282
    return None, None
 
4283
 
 
4284
 
 
4285
def probe_bad_non_ascii(encoding):
 
4286
    """Try to find [bad] character with code [128..255]
 
4287
    that cannot be decoded to unicode in some encoding.
 
4288
    Return None if all non-ascii characters is valid
 
4289
    for given encoding.
 
4290
    """
 
4291
    for i in xrange(128, 256):
 
4292
        char = chr(i)
 
4293
        try:
 
4294
            char.decode(encoding)
 
4295
        except UnicodeDecodeError:
 
4296
            return char
 
4297
    return None
 
4298
 
 
4299
 
 
4300
class _HTTPSServerFeature(Feature):
 
4301
    """Some tests want an https Server, check if one is available.
 
4302
 
 
4303
    Right now, the only way this is available is under python2.6 which provides
 
4304
    an ssl module.
 
4305
    """
 
4306
 
 
4307
    def _probe(self):
 
4308
        try:
 
4309
            import ssl
 
4310
            return True
 
4311
        except ImportError:
 
4312
            return False
 
4313
 
 
4314
    def feature_name(self):
 
4315
        return 'HTTPSServer'
 
4316
 
 
4317
 
 
4318
HTTPSServerFeature = _HTTPSServerFeature()
 
4319
 
 
4320
 
 
4321
class _UnicodeFilename(Feature):
 
4322
    """Does the filesystem support Unicode filenames?"""
 
4323
 
 
4324
    def _probe(self):
 
4325
        try:
 
4326
            os.stat(u'\u03b1')
 
4327
        except UnicodeEncodeError:
 
4328
            return False
 
4329
        except (IOError, OSError):
 
4330
            # The filesystem allows the Unicode filename but the file doesn't
 
4331
            # exist.
 
4332
            return True
 
4333
        else:
 
4334
            # The filesystem allows the Unicode filename and the file exists,
 
4335
            # for some reason.
 
4336
            return True
 
4337
 
 
4338
UnicodeFilename = _UnicodeFilename()
 
4339
 
 
4340
 
 
4341
class _UTF8Filesystem(Feature):
 
4342
    """Is the filesystem UTF-8?"""
 
4343
 
 
4344
    def _probe(self):
 
4345
        if osutils._fs_enc.upper() in ('UTF-8', 'UTF8'):
 
4346
            return True
 
4347
        return False
 
4348
 
 
4349
UTF8Filesystem = _UTF8Filesystem()
 
4350
 
 
4351
 
 
4352
class _BreakinFeature(Feature):
 
4353
    """Does this platform support the breakin feature?"""
 
4354
 
 
4355
    def _probe(self):
 
4356
        from bzrlib import breakin
 
4357
        if breakin.determine_signal() is None:
 
4358
            return False
 
4359
        if sys.platform == 'win32':
 
4360
            # Windows doesn't have os.kill, and we catch the SIGBREAK signal.
 
4361
            # We trigger SIGBREAK via a Console api so we need ctypes to
 
4362
            # access the function
 
4363
            try:
 
4364
                import ctypes
 
4365
            except OSError:
 
4366
                return False
 
4367
        return True
 
4368
 
 
4369
    def feature_name(self):
 
4370
        return "SIGQUIT or SIGBREAK w/ctypes on win32"
 
4371
 
 
4372
 
 
4373
BreakinFeature = _BreakinFeature()
 
4374
 
 
4375
 
 
4376
class _CaseInsCasePresFilenameFeature(Feature):
 
4377
    """Is the file-system case insensitive, but case-preserving?"""
 
4378
 
 
4379
    def _probe(self):
 
4380
        fileno, name = tempfile.mkstemp(prefix='MixedCase')
 
4381
        try:
 
4382
            # first check truly case-preserving for created files, then check
 
4383
            # case insensitive when opening existing files.
 
4384
            name = osutils.normpath(name)
 
4385
            base, rel = osutils.split(name)
 
4386
            found_rel = osutils.canonical_relpath(base, name)
 
4387
            return (found_rel == rel
 
4388
                    and os.path.isfile(name.upper())
 
4389
                    and os.path.isfile(name.lower()))
 
4390
        finally:
 
4391
            os.close(fileno)
 
4392
            os.remove(name)
 
4393
 
 
4394
    def feature_name(self):
 
4395
        return "case-insensitive case-preserving filesystem"
 
4396
 
 
4397
CaseInsCasePresFilenameFeature = _CaseInsCasePresFilenameFeature()
 
4398
 
 
4399
 
 
4400
class _CaseInsensitiveFilesystemFeature(Feature):
 
4401
    """Check if underlying filesystem is case-insensitive but *not* case
 
4402
    preserving.
 
4403
    """
 
4404
    # Note that on Windows, Cygwin, MacOS etc, the file-systems are far
 
4405
    # more likely to be case preserving, so this case is rare.
 
4406
 
 
4407
    def _probe(self):
 
4408
        if CaseInsCasePresFilenameFeature.available():
 
4409
            return False
 
4410
 
 
4411
        if TestCaseWithMemoryTransport.TEST_ROOT is None:
 
4412
            root = osutils.mkdtemp(prefix='testbzr-', suffix='.tmp')
 
4413
            TestCaseWithMemoryTransport.TEST_ROOT = root
 
4414
        else:
 
4415
            root = TestCaseWithMemoryTransport.TEST_ROOT
 
4416
        tdir = osutils.mkdtemp(prefix='case-sensitive-probe-', suffix='',
 
4417
            dir=root)
 
4418
        name_a = osutils.pathjoin(tdir, 'a')
 
4419
        name_A = osutils.pathjoin(tdir, 'A')
 
4420
        os.mkdir(name_a)
 
4421
        result = osutils.isdir(name_A)
 
4422
        _rmtree_temp_dir(tdir)
 
4423
        return result
 
4424
 
 
4425
    def feature_name(self):
 
4426
        return 'case-insensitive filesystem'
 
4427
 
 
4428
CaseInsensitiveFilesystemFeature = _CaseInsensitiveFilesystemFeature()
 
4429
 
 
4430
 
 
4431
class _CaseSensitiveFilesystemFeature(Feature):
 
4432
 
 
4433
    def _probe(self):
 
4434
        if CaseInsCasePresFilenameFeature.available():
 
4435
            return False
 
4436
        elif CaseInsensitiveFilesystemFeature.available():
 
4437
            return False
 
4438
        else:
 
4439
            return True
 
4440
 
 
4441
    def feature_name(self):
 
4442
        return 'case-sensitive filesystem'
 
4443
 
 
4444
# new coding style is for feature instances to be lowercase
 
4445
case_sensitive_filesystem_feature = _CaseSensitiveFilesystemFeature()
 
4446
 
 
4447
 
 
4448
# Kept for compatibility, use bzrlib.tests.features.subunit instead
 
4449
SubUnitFeature = _CompatabilityThunkFeature(
 
4450
    deprecated_in((2,1,0)),
 
4451
    'bzrlib.tests.features', 'SubUnitFeature', 'subunit')
 
4452
# Only define SubUnitBzrRunner if subunit is available.
 
4453
try:
 
4454
    from subunit import TestProtocolClient
 
4455
    from subunit.test_results import AutoTimingTestResultDecorator
 
4456
    class SubUnitBzrRunner(TextTestRunner):
 
4457
        def run(self, test):
 
4458
            result = AutoTimingTestResultDecorator(
 
4459
                TestProtocolClient(self.stream))
 
4460
            test.run(result)
 
4461
            return result
 
4462
except ImportError:
 
4463
    pass
 
4464
 
 
4465
class _PosixPermissionsFeature(Feature):
 
4466
 
 
4467
    def _probe(self):
 
4468
        def has_perms():
 
4469
            # create temporary file and check if specified perms are maintained.
 
4470
            import tempfile
 
4471
 
 
4472
            write_perms = stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR
 
4473
            f = tempfile.mkstemp(prefix='bzr_perms_chk_')
 
4474
            fd, name = f
 
4475
            os.close(fd)
 
4476
            os.chmod(name, write_perms)
 
4477
 
 
4478
            read_perms = os.stat(name).st_mode & 0777
 
4479
            os.unlink(name)
 
4480
            return (write_perms == read_perms)
 
4481
 
 
4482
        return (os.name == 'posix') and has_perms()
 
4483
 
 
4484
    def feature_name(self):
 
4485
        return 'POSIX permissions support'
 
4486
 
 
4487
posix_permissions_feature = _PosixPermissionsFeature()