← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/__init__.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2011-05-04 12:10:51 UTC
  • mfrom: (5819.1.4 777007-developer-doc)
  • Revision ID: pqm@pqm.ubuntu.com-20110504121051-aovlsmqiivjmc4fc
(jelmer) Small fixes to developer documentation. (Jonathan Riddell)

Show diffs side-by-side

added added

removed removed

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