/brz/remove-bazaar

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

« back to all changes in this revision

Viewing changes to bzrlib/tests/__init__.py

  • Committer: Alexander Belchenko
  • Date: 2007-09-04 11:20:50 UTC
  • mto: This revision was merged to the branch mainline in revision 2800.
  • Revision ID: bialix@ukr.net-20070904112050-muj2o81u0xuaheoz
show path to python dll instead of bzr.exe for standalone application

Show diffs side-by-side

added added

removed removed

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