← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/__init__.py

  • Committer: John Arbash Meinel
  • Date: 2009-06-02 19:56:24 UTC
  • mto: This revision was merged to the branch mainline in revision 4469.
  • Revision ID: john@arbash-meinel.com-20090602195624-utljsyz0qgmq63lg
Add a chunks_to_gzip function.
This allows the _record_to_data code to build up a list of chunks,
rather than requiring a single string.
It should be ~ the same performance when using a single string, since
we are only adding a for() loop over the chunks and an if check.
We could possibly just remove the if check and not worry about adding
some empty strings in there.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/__init__.py
1
 
# Copyright (C) 2005-2010 Canonical Ltd
 
1
# Copyright (C) 2005, 2006, 2007, 2008, 2009 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
14
14
# along with this program; if not, write to the Free Software
15
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 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
 
import copy
33
31
from cStringIO import StringIO
34
32
import difflib
35
33
import doctest
36
34
import errno
37
 
import itertools
38
35
import logging
39
36
import math
40
37
import os
41
 
import pprint
 
38
from pprint import pformat
42
39
import random
43
40
import re
44
41
import shlex
45
42
import stat
46
 
import subprocess
 
43
from subprocess import Popen, PIPE, STDOUT
47
44
import sys
48
45
import tempfile
49
46
import threading
50
47
import time
51
 
import traceback
52
48
import unittest
53
49
import warnings
54
50
 
55
 
import testtools
56
 
# nb: check this before importing anything else from within it
57
 
_testtools_version = getattr(testtools, '__version__', ())
58
 
if _testtools_version < (0, 9, 2):
59
 
    raise ImportError("need at least testtools 0.9.2: %s is %r"
60
 
        % (testtools.__file__, _testtools_version))
61
 
from testtools import content
62
51
 
63
52
from bzrlib import (
64
53
    branchbuilder,
65
54
    bzrdir,
66
 
    chk_map,
67
 
    config,
68
55
    debug,
69
56
    errors,
70
57
    hooks,
75
62
    ui,
76
63
    urlutils,
77
64
    registry,
78
 
    transport as _mod_transport,
79
65
    workingtree,
80
66
    )
81
67
import bzrlib.branch
99
85
from bzrlib.symbol_versioning import (
100
86
    DEPRECATED_PARAMETER,
101
87
    deprecated_function,
102
 
    deprecated_in,
103
88
    deprecated_method,
104
89
    deprecated_passed,
105
90
    )
106
91
import bzrlib.trace
107
 
from bzrlib.transport import (
108
 
    memory,
109
 
    pathfilter,
110
 
    )
 
92
from bzrlib.transport import get_transport
 
93
import bzrlib.transport
 
94
from bzrlib.transport.local import LocalURLServer
 
95
from bzrlib.transport.memory import MemoryServer
 
96
from bzrlib.transport.readonly import ReadonlyServer
111
97
from bzrlib.trace import mutter, note
112
 
from bzrlib.tests import (
113
 
    test_server,
114
 
    TestUtil,
115
 
    treeshape,
116
 
    )
117
 
from bzrlib.ui import NullProgressView
118
 
from bzrlib.ui.text import TextUIFactory
 
98
from bzrlib.tests import TestUtil
 
99
from bzrlib.tests.http_server import HttpServer
 
100
from bzrlib.tests.TestUtil import (
 
101
                          TestSuite,
 
102
                          TestLoader,
 
103
                          )
 
104
from bzrlib.tests.treeshape import build_tree_contents
119
105
import bzrlib.version_info_formats.format_custom
120
106
from bzrlib.workingtree import WorkingTree, WorkingTreeFormat2
121
107
 
124
110
# shown frame is the test code, not our assertXYZ.
125
111
__unittest = 1
126
112
 
127
 
default_transport = test_server.LocalURLServer
128
 
 
129
 
 
130
 
_unitialized_attr = object()
131
 
"""A sentinel needed to act as a default value in a method signature."""
132
 
 
133
 
 
134
 
# Subunit result codes, defined here to prevent a hard dependency on subunit.
135
 
SUBUNIT_SEEK_SET = 0
136
 
SUBUNIT_SEEK_CUR = 1
137
 
 
138
 
 
139
 
class ExtendedTestResult(testtools.TextTestResult):
 
113
default_transport = LocalURLServer
 
114
 
 
115
 
 
116
class ExtendedTestResult(unittest._TextTestResult):
140
117
    """Accepts, reports and accumulates the results of running tests.
141
118
 
142
119
    Compared to the unittest version this class adds support for
156
133
 
157
134
    def __init__(self, stream, descriptions, verbosity,
158
135
                 bench_history=None,
 
136
                 num_tests=None,
159
137
                 strict=False,
160
138
                 ):
161
139
        """Construct new TestResult.
163
141
        :param bench_history: Optionally, a writable file object to accumulate
164
142
            benchmark results.
165
143
        """
166
 
        testtools.TextTestResult.__init__(self, stream)
 
144
        unittest._TextTestResult.__init__(self, stream, descriptions, verbosity)
167
145
        if bench_history is not None:
168
146
            from bzrlib.version import _get_bzr_source_tree
169
147
            src_tree = _get_bzr_source_tree()
180
158
            bench_history.write("--date %s %s\n" % (time.time(), revision_id))
181
159
        self._bench_history = bench_history
182
160
        self.ui = ui.ui_factory
183
 
        self.num_tests = 0
 
161
        self.num_tests = num_tests
184
162
        self.error_count = 0
185
163
        self.failure_count = 0
186
164
        self.known_failure_count = 0
191
169
        self._overall_start_time = time.time()
192
170
        self._strict = strict
193
171
 
194
 
    def stopTestRun(self):
195
 
        run = self.testsRun
196
 
        actionTaken = "Ran"
197
 
        stopTime = time.time()
198
 
        timeTaken = stopTime - self.startTime
199
 
        # GZ 2010-07-19: Seems testtools has no printErrors method, and though
200
 
        #                the parent class method is similar have to duplicate
201
 
        self._show_list('ERROR', self.errors)
202
 
        self._show_list('FAIL', self.failures)
203
 
        self.stream.write(self.sep2)
204
 
        self.stream.write("%s %d test%s in %.3fs\n\n" % (actionTaken,
205
 
                            run, run != 1 and "s" or "", timeTaken))
206
 
        if not self.wasSuccessful():
207
 
            self.stream.write("FAILED (")
208
 
            failed, errored = map(len, (self.failures, self.errors))
209
 
            if failed:
210
 
                self.stream.write("failures=%d" % failed)
211
 
            if errored:
212
 
                if failed: self.stream.write(", ")
213
 
                self.stream.write("errors=%d" % errored)
214
 
            if self.known_failure_count:
215
 
                if failed or errored: self.stream.write(", ")
216
 
                self.stream.write("known_failure_count=%d" %
217
 
                    self.known_failure_count)
218
 
            self.stream.write(")\n")
219
 
        else:
220
 
            if self.known_failure_count:
221
 
                self.stream.write("OK (known_failures=%d)\n" %
222
 
                    self.known_failure_count)
223
 
            else:
224
 
                self.stream.write("OK\n")
225
 
        if self.skip_count > 0:
226
 
            skipped = self.skip_count
227
 
            self.stream.write('%d test%s skipped\n' %
228
 
                                (skipped, skipped != 1 and "s" or ""))
229
 
        if self.unsupported:
230
 
            for feature, count in sorted(self.unsupported.items()):
231
 
                self.stream.write("Missing feature '%s' skipped %d tests.\n" %
232
 
                    (feature, count))
 
172
    def done(self):
233
173
        if self._strict:
234
174
            ok = self.wasStrictlySuccessful()
235
175
        else:
236
176
            ok = self.wasSuccessful()
 
177
        if ok:
 
178
            self.stream.write('tests passed\n')
 
179
        else:
 
180
            self.stream.write('tests failed\n')
237
181
        if TestCase._first_thread_leaker_id:
238
182
            self.stream.write(
239
183
                '%s is leaking threads among %d leaking tests.\n' % (
240
184
                TestCase._first_thread_leaker_id,
241
185
                TestCase._leaking_threads_tests))
242
 
            # We don't report the main thread as an active one.
243
 
            self.stream.write(
244
 
                '%d non-main threads were left active in the end.\n'
245
 
                % (TestCase._active_threads - 1))
246
 
 
247
 
    def getDescription(self, test):
248
 
        return test.id()
249
 
 
250
 
    def _extractBenchmarkTime(self, testCase, details=None):
 
186
 
 
187
    def _extractBenchmarkTime(self, testCase):
251
188
        """Add a benchmark time for the current test case."""
252
 
        if details and 'benchtime' in details:
253
 
            return float(''.join(details['benchtime'].iter_bytes()))
254
189
        return getattr(testCase, "_benchtime", None)
255
190
 
256
191
    def _elapsedTestTimeString(self):
260
195
    def _testTimeString(self, testCase):
261
196
        benchmark_time = self._extractBenchmarkTime(testCase)
262
197
        if benchmark_time is not None:
263
 
            return self._formatTime(benchmark_time) + "*"
 
198
            return "%s/%s" % (
 
199
                self._formatTime(benchmark_time),
 
200
                self._elapsedTestTimeString())
264
201
        else:
265
 
            return self._elapsedTestTimeString()
 
202
            return "           %s" % self._elapsedTestTimeString()
266
203
 
267
204
    def _formatTime(self, seconds):
268
205
        """Format seconds as milliseconds with leading spaces."""
272
209
 
273
210
    def _shortened_test_description(self, test):
274
211
        what = test.id()
275
 
        what = re.sub(r'^bzrlib\.tests\.', '', what)
 
212
        what = re.sub(r'^bzrlib\.(tests|benchmarks)\.', '', what)
276
213
        return what
277
214
 
278
215
    def startTest(self, test):
279
 
        super(ExtendedTestResult, self).startTest(test)
 
216
        unittest.TestResult.startTest(self, test)
280
217
        if self.count == 0:
281
218
            self.startTests()
282
219
        self.report_test_start(test)
284
221
        self._recordTestStartTime()
285
222
 
286
223
    def startTests(self):
287
 
        import platform
288
 
        if getattr(sys, 'frozen', None) is None:
289
 
            bzr_path = osutils.realpath(sys.argv[0])
290
 
        else:
291
 
            bzr_path = sys.executable
292
 
        self.stream.write(
293
 
            'bzr selftest: %s\n' % (bzr_path,))
294
 
        self.stream.write(
295
 
            '   %s\n' % (
296
 
                    bzrlib.__path__[0],))
297
 
        self.stream.write(
298
 
            '   bzr-%s python-%s %s\n' % (
 
224
        self.stream.write(
 
225
            'testing: %s\n' % (osutils.realpath(sys.argv[0]),))
 
226
        self.stream.write(
 
227
            '   %s (%s python%s)\n' % (
 
228
                    bzrlib.__path__[0],
299
229
                    bzrlib.version_string,
300
230
                    bzrlib._format_version_tuple(sys.version_info),
301
 
                    platform.platform(aliased=1),
302
231
                    ))
303
232
        self.stream.write('\n')
304
233
 
319
248
        Called from the TestCase run() method when the test
320
249
        fails with an unexpected error.
321
250
        """
322
 
        self._post_mortem()
323
 
        super(ExtendedTestResult, self).addError(test, err)
324
 
        self.error_count += 1
325
 
        self.report_error(test, err)
326
 
        if self.stop_early:
327
 
            self.stop()
328
 
        self._cleanupLogFile(test)
 
251
        self._testConcluded(test)
 
252
        if isinstance(err[1], TestNotApplicable):
 
253
            return self._addNotApplicable(test, err)
 
254
        elif isinstance(err[1], UnavailableFeature):
 
255
            return self.addNotSupported(test, err[1].args[0])
 
256
        else:
 
257
            unittest.TestResult.addError(self, test, err)
 
258
            self.error_count += 1
 
259
            self.report_error(test, err)
 
260
            if self.stop_early:
 
261
                self.stop()
 
262
            self._cleanupLogFile(test)
329
263
 
330
264
    def addFailure(self, test, err):
331
265
        """Tell result that test failed.
333
267
        Called from the TestCase run() method when the test
334
268
        fails because e.g. an assert() method failed.
335
269
        """
336
 
        self._post_mortem()
337
 
        super(ExtendedTestResult, self).addFailure(test, err)
338
 
        self.failure_count += 1
339
 
        self.report_failure(test, err)
340
 
        if self.stop_early:
341
 
            self.stop()
342
 
        self._cleanupLogFile(test)
 
270
        self._testConcluded(test)
 
271
        if isinstance(err[1], KnownFailure):
 
272
            return self._addKnownFailure(test, err)
 
273
        else:
 
274
            unittest.TestResult.addFailure(self, test, err)
 
275
            self.failure_count += 1
 
276
            self.report_failure(test, err)
 
277
            if self.stop_early:
 
278
                self.stop()
 
279
            self._cleanupLogFile(test)
343
280
 
344
 
    def addSuccess(self, test, details=None):
 
281
    def addSuccess(self, test):
345
282
        """Tell result that test completed successfully.
346
283
 
347
284
        Called from the TestCase run()
348
285
        """
 
286
        self._testConcluded(test)
349
287
        if self._bench_history is not None:
350
 
            benchmark_time = self._extractBenchmarkTime(test, details)
 
288
            benchmark_time = self._extractBenchmarkTime(test)
351
289
            if benchmark_time is not None:
352
290
                self._bench_history.write("%s %s\n" % (
353
291
                    self._formatTime(benchmark_time),
354
292
                    test.id()))
355
293
        self.report_success(test)
356
294
        self._cleanupLogFile(test)
357
 
        super(ExtendedTestResult, self).addSuccess(test)
 
295
        unittest.TestResult.addSuccess(self, test)
358
296
        test._log_contents = ''
359
297
 
360
 
    def addExpectedFailure(self, test, err):
 
298
    def _testConcluded(self, test):
 
299
        """Common code when a test has finished.
 
300
 
 
301
        Called regardless of whether it succeded, failed, etc.
 
302
        """
 
303
        pass
 
304
 
 
305
    def _addKnownFailure(self, test, err):
361
306
        self.known_failure_count += 1
362
307
        self.report_known_failure(test, err)
363
308
 
365
310
        """The test will not be run because of a missing feature.
366
311
        """
367
312
        # this can be called in two different ways: it may be that the
368
 
        # test started running, and then raised (through requireFeature)
 
313
        # test started running, and then raised (through addError)
369
314
        # UnavailableFeature.  Alternatively this method can be called
370
 
        # while probing for features before running the test code proper; in
371
 
        # that case we will see startTest and stopTest, but the test will
372
 
        # never actually run.
 
315
        # while probing for features before running the tests; in that
 
316
        # case we will see startTest and stopTest, but the test will never
 
317
        # actually run.
373
318
        self.unsupported.setdefault(str(feature), 0)
374
319
        self.unsupported[str(feature)] += 1
375
320
        self.report_unsupported(test, feature)
379
324
        self.skip_count += 1
380
325
        self.report_skip(test, reason)
381
326
 
382
 
    def addNotApplicable(self, test, reason):
383
 
        self.not_applicable_count += 1
384
 
        self.report_not_applicable(test, reason)
385
 
 
386
 
    def _post_mortem(self):
387
 
        """Start a PDB post mortem session."""
388
 
        if os.environ.get('BZR_TEST_PDB', None):
389
 
            import pdb;pdb.post_mortem()
390
 
 
391
 
    def progress(self, offset, whence):
392
 
        """The test is adjusting the count of tests to run."""
393
 
        if whence == SUBUNIT_SEEK_SET:
394
 
            self.num_tests = offset
395
 
        elif whence == SUBUNIT_SEEK_CUR:
396
 
            self.num_tests += offset
 
327
    def _addNotApplicable(self, test, skip_excinfo):
 
328
        if isinstance(skip_excinfo[1], TestNotApplicable):
 
329
            self.not_applicable_count += 1
 
330
            self.report_not_applicable(test, skip_excinfo)
 
331
        try:
 
332
            test.tearDown()
 
333
        except KeyboardInterrupt:
 
334
            raise
 
335
        except:
 
336
            self.addError(test, test.exc_info())
397
337
        else:
398
 
            raise errors.BzrError("Unknown whence %r" % whence)
 
338
            # seems best to treat this as success from point-of-view of unittest
 
339
            # -- it actually does nothing so it barely matters :)
 
340
            unittest.TestResult.addSuccess(self, test)
 
341
            test._log_contents = ''
 
342
 
 
343
    def printErrorList(self, flavour, errors):
 
344
        for test, err in errors:
 
345
            self.stream.writeln(self.separator1)
 
346
            self.stream.write("%s: " % flavour)
 
347
            self.stream.writeln(self.getDescription(test))
 
348
            if getattr(test, '_get_log', None) is not None:
 
349
                self.stream.write('\n')
 
350
                self.stream.write(
 
351
                        ('vvvv[log from %s]' % test.id()).ljust(78,'-'))
 
352
                self.stream.write('\n')
 
353
                self.stream.write(test._get_log())
 
354
                self.stream.write('\n')
 
355
                self.stream.write(
 
356
                        ('^^^^[log from %s]' % test.id()).ljust(78,'-'))
 
357
                self.stream.write('\n')
 
358
            self.stream.writeln(self.separator2)
 
359
            self.stream.writeln("%s" % err)
 
360
 
 
361
    def finished(self):
 
362
        pass
399
363
 
400
364
    def report_cleaning_up(self):
401
365
        pass
402
366
 
403
 
    def startTestRun(self):
404
 
        self.startTime = time.time()
405
 
 
406
367
    def report_success(self, test):
407
368
        pass
408
369
 
417
378
 
418
379
    def __init__(self, stream, descriptions, verbosity,
419
380
                 bench_history=None,
 
381
                 num_tests=None,
420
382
                 pb=None,
421
383
                 strict=None,
422
384
                 ):
423
385
        ExtendedTestResult.__init__(self, stream, descriptions, verbosity,
424
 
            bench_history, strict)
425
 
        # We no longer pass them around, but just rely on the UIFactory stack
426
 
        # for state
427
 
        if pb is not None:
428
 
            warnings.warn("Passing pb to TextTestResult is deprecated")
429
 
        self.pb = self.ui.nested_progress_bar()
 
386
            bench_history, num_tests, strict)
 
387
        if pb is None:
 
388
            self.pb = self.ui.nested_progress_bar()
 
389
            self._supplied_pb = False
 
390
        else:
 
391
            self.pb = pb
 
392
            self._supplied_pb = True
430
393
        self.pb.show_pct = False
431
394
        self.pb.show_spinner = False
432
395
        self.pb.show_eta = False,
433
396
        self.pb.show_count = False
434
397
        self.pb.show_bar = False
435
 
        self.pb.update_latency = 0
436
 
        self.pb.show_transport_activity = False
437
 
 
438
 
    def stopTestRun(self):
439
 
        # called when the tests that are going to run have run
440
 
        self.pb.clear()
441
 
        self.pb.finished()
442
 
        super(TextTestResult, self).stopTestRun()
443
 
 
444
 
    def startTestRun(self):
445
 
        super(TextTestResult, self).startTestRun()
 
398
 
 
399
    def report_starting(self):
446
400
        self.pb.update('[test 0/%d] Starting' % (self.num_tests))
447
401
 
448
 
    def printErrors(self):
449
 
        # clear the pb to make room for the error listing
450
 
        self.pb.clear()
451
 
        super(TextTestResult, self).printErrors()
452
 
 
453
402
    def _progress_prefix_text(self):
454
403
        # the longer this text, the less space we have to show the test
455
404
        # name...
460
409
        ##     a += ', %d skip' % self.skip_count
461
410
        ## if self.known_failure_count:
462
411
        ##     a += '+%dX' % self.known_failure_count
463
 
        if self.num_tests:
 
412
        if self.num_tests is not None:
464
413
            a +='/%d' % self.num_tests
465
414
        a += ' in '
466
415
        runtime = time.time() - self._overall_start_time
468
417
            a += '%dm%ds' % (runtime / 60, runtime % 60)
469
418
        else:
470
419
            a += '%ds' % runtime
471
 
        total_fail_count = self.error_count + self.failure_count
472
 
        if total_fail_count:
473
 
            a += ', %d failed' % total_fail_count
474
 
        # if self.unsupported:
475
 
        #     a += ', %d missing' % len(self.unsupported)
 
420
        if self.error_count:
 
421
            a += ', %d err' % self.error_count
 
422
        if self.failure_count:
 
423
            a += ', %d fail' % self.failure_count
 
424
        if self.unsupported:
 
425
            a += ', %d missing' % len(self.unsupported)
476
426
        a += ']'
477
427
        return a
478
428
 
487
437
        return self._shortened_test_description(test)
488
438
 
489
439
    def report_error(self, test, err):
490
 
        self.stream.write('ERROR: %s\n    %s\n' % (
 
440
        self.pb.note('ERROR: %s\n    %s\n',
491
441
            self._test_description(test),
492
442
            err[1],
493
 
            ))
 
443
            )
494
444
 
495
445
    def report_failure(self, test, err):
496
 
        self.stream.write('FAIL: %s\n    %s\n' % (
 
446
        self.pb.note('FAIL: %s\n    %s\n',
497
447
            self._test_description(test),
498
448
            err[1],
499
 
            ))
 
449
            )
500
450
 
501
451
    def report_known_failure(self, test, err):
502
 
        pass
 
452
        self.pb.note('XFAIL: %s\n%s\n',
 
453
            self._test_description(test), err[1])
503
454
 
504
455
    def report_skip(self, test, reason):
505
456
        pass
506
457
 
507
 
    def report_not_applicable(self, test, reason):
 
458
    def report_not_applicable(self, test, skip_excinfo):
508
459
        pass
509
460
 
510
461
    def report_unsupported(self, test, feature):
513
464
    def report_cleaning_up(self):
514
465
        self.pb.update('Cleaning up')
515
466
 
 
467
    def finished(self):
 
468
        if not self._supplied_pb:
 
469
            self.pb.finished()
 
470
 
516
471
 
517
472
class VerboseTestResult(ExtendedTestResult):
518
473
    """Produce long output, with one line per test run plus times"""
525
480
            result = a_string
526
481
        return result.ljust(final_width)
527
482
 
528
 
    def startTestRun(self):
529
 
        super(VerboseTestResult, self).startTestRun()
 
483
    def report_starting(self):
530
484
        self.stream.write('running %d tests...\n' % self.num_tests)
531
485
 
532
486
    def report_test_start(self, test):
533
487
        self.count += 1
534
488
        name = self._shortened_test_description(test)
535
 
        width = osutils.terminal_width()
536
 
        if width is not None:
537
 
            # width needs space for 6 char status, plus 1 for slash, plus an
538
 
            # 11-char time string, plus a trailing blank
539
 
            # when NUMBERED_DIRS: plus 5 chars on test number, plus 1 char on
540
 
            # space
541
 
            self.stream.write(self._ellipsize_to_right(name, width-18))
542
 
        else:
543
 
            self.stream.write(name)
 
489
        # width needs space for 6 char status, plus 1 for slash, plus 2 10-char
 
490
        # numbers, plus a trailing blank
 
491
        # when NUMBERED_DIRS: plus 5 chars on test number, plus 1 char on space
 
492
        self.stream.write(self._ellipsize_to_right(name,
 
493
                          osutils.terminal_width()-30))
544
494
        self.stream.flush()
545
495
 
546
496
    def _error_summary(self, err):
548
498
        return '%s%s' % (indent, err[1])
549
499
 
550
500
    def report_error(self, test, err):
551
 
        self.stream.write('ERROR %s\n%s\n'
 
501
        self.stream.writeln('ERROR %s\n%s'
552
502
                % (self._testTimeString(test),
553
503
                   self._error_summary(err)))
554
504
 
555
505
    def report_failure(self, test, err):
556
 
        self.stream.write(' FAIL %s\n%s\n'
 
506
        self.stream.writeln(' FAIL %s\n%s'
557
507
                % (self._testTimeString(test),
558
508
                   self._error_summary(err)))
559
509
 
560
510
    def report_known_failure(self, test, err):
561
 
        self.stream.write('XFAIL %s\n%s\n'
 
511
        self.stream.writeln('XFAIL %s\n%s'
562
512
                % (self._testTimeString(test),
563
513
                   self._error_summary(err)))
564
514
 
565
515
    def report_success(self, test):
566
 
        self.stream.write('   OK %s\n' % self._testTimeString(test))
 
516
        self.stream.writeln('   OK %s' % self._testTimeString(test))
567
517
        for bench_called, stats in getattr(test, '_benchcalls', []):
568
 
            self.stream.write('LSProf output for %s(%s, %s)\n' % bench_called)
 
518
            self.stream.writeln('LSProf output for %s(%s, %s)' % bench_called)
569
519
            stats.pprint(file=self.stream)
570
520
        # flush the stream so that we get smooth output. This verbose mode is
571
521
        # used to show the output in PQM.
572
522
        self.stream.flush()
573
523
 
574
524
    def report_skip(self, test, reason):
575
 
        self.stream.write(' SKIP %s\n%s\n'
 
525
        self.stream.writeln(' SKIP %s\n%s'
576
526
                % (self._testTimeString(test), reason))
577
527
 
578
 
    def report_not_applicable(self, test, reason):
579
 
        self.stream.write('  N/A %s\n    %s\n'
580
 
                % (self._testTimeString(test), reason))
 
528
    def report_not_applicable(self, test, skip_excinfo):
 
529
        self.stream.writeln('  N/A %s\n%s'
 
530
                % (self._testTimeString(test),
 
531
                   self._error_summary(skip_excinfo)))
581
532
 
582
533
    def report_unsupported(self, test, feature):
583
534
        """test cannot be run because feature is missing."""
584
 
        self.stream.write("NODEP %s\n    The feature '%s' is not available.\n"
 
535
        self.stream.writeln("NODEP %s\n    The feature '%s' is not available."
585
536
                %(self._testTimeString(test), feature))
586
537
 
587
538
 
593
544
                 descriptions=0,
594
545
                 verbosity=1,
595
546
                 bench_history=None,
 
547
                 list_only=False,
596
548
                 strict=False,
597
 
                 result_decorators=None,
598
549
                 ):
599
 
        """Create a TextTestRunner.
600
 
 
601
 
        :param result_decorators: An optional list of decorators to apply
602
 
            to the result object being used by the runner. Decorators are
603
 
            applied left to right - the first element in the list is the 
604
 
            innermost decorator.
605
 
        """
606
 
        # stream may know claim to know to write unicode strings, but in older
607
 
        # pythons this goes sufficiently wrong that it is a bad idea. (
608
 
        # specifically a built in file with encoding 'UTF-8' will still try
609
 
        # to encode using ascii.
610
 
        new_encoding = osutils.get_terminal_encoding()
611
 
        codec = codecs.lookup(new_encoding)
612
 
        if type(codec) is tuple:
613
 
            # Python 2.4
614
 
            encode = codec[0]
615
 
        else:
616
 
            encode = codec.encode
617
 
        stream = osutils.UnicodeOrBytesToBytesWriter(encode, stream)
618
 
        stream.encoding = new_encoding
619
 
        self.stream = stream
 
550
        self.stream = unittest._WritelnDecorator(stream)
620
551
        self.descriptions = descriptions
621
552
        self.verbosity = verbosity
622
553
        self._bench_history = bench_history
 
554
        self.list_only = list_only
623
555
        self._strict = strict
624
 
        self._result_decorators = result_decorators or []
625
556
 
626
557
    def run(self, test):
627
558
        "Run the given test case or test suite."
 
559
        startTime = time.time()
628
560
        if self.verbosity == 1:
629
561
            result_class = TextTestResult
630
562
        elif self.verbosity >= 2:
631
563
            result_class = VerboseTestResult
632
 
        original_result = result_class(self.stream,
 
564
        result = result_class(self.stream,
633
565
                              self.descriptions,
634
566
                              self.verbosity,
635
567
                              bench_history=self._bench_history,
 
568
                              num_tests=test.countTestCases(),
636
569
                              strict=self._strict,
637
570
                              )
638
 
        # Signal to result objects that look at stop early policy to stop,
639
 
        original_result.stop_early = self.stop_on_failure
640
 
        result = original_result
641
 
        for decorator in self._result_decorators:
642
 
            result = decorator(result)
643
 
            result.stop_early = self.stop_on_failure
644
 
        result.startTestRun()
645
 
        try:
646
 
            test.run(result)
647
 
        finally:
648
 
            result.stopTestRun()
649
 
        # higher level code uses our extended protocol to determine
650
 
        # what exit code to give.
651
 
        return original_result
 
571
        result.stop_early = self.stop_on_failure
 
572
        result.report_starting()
 
573
        if self.list_only:
 
574
            if self.verbosity >= 2:
 
575
                self.stream.writeln("Listing tests only ...\n")
 
576
            run = 0
 
577
            for t in iter_suite_tests(test):
 
578
                self.stream.writeln("%s" % (t.id()))
 
579
                run += 1
 
580
            return None
 
581
        else:
 
582
            try:
 
583
                import testtools
 
584
            except ImportError:
 
585
                test.run(result)
 
586
            else:
 
587
                if isinstance(test, testtools.ConcurrentTestSuite):
 
588
                    # We need to catch bzr specific behaviors
 
589
                    test.run(BZRTransformingResult(result))
 
590
                else:
 
591
                    test.run(result)
 
592
            run = result.testsRun
 
593
            actionTaken = "Ran"
 
594
        stopTime = time.time()
 
595
        timeTaken = stopTime - startTime
 
596
        result.printErrors()
 
597
        self.stream.writeln(result.separator2)
 
598
        self.stream.writeln("%s %d test%s in %.3fs" % (actionTaken,
 
599
                            run, run != 1 and "s" or "", timeTaken))
 
600
        self.stream.writeln()
 
601
        if not result.wasSuccessful():
 
602
            self.stream.write("FAILED (")
 
603
            failed, errored = map(len, (result.failures, result.errors))
 
604
            if failed:
 
605
                self.stream.write("failures=%d" % failed)
 
606
            if errored:
 
607
                if failed: self.stream.write(", ")
 
608
                self.stream.write("errors=%d" % errored)
 
609
            if result.known_failure_count:
 
610
                if failed or errored: self.stream.write(", ")
 
611
                self.stream.write("known_failure_count=%d" %
 
612
                    result.known_failure_count)
 
613
            self.stream.writeln(")")
 
614
        else:
 
615
            if result.known_failure_count:
 
616
                self.stream.writeln("OK (known_failures=%d)" %
 
617
                    result.known_failure_count)
 
618
            else:
 
619
                self.stream.writeln("OK")
 
620
        if result.skip_count > 0:
 
621
            skipped = result.skip_count
 
622
            self.stream.writeln('%d test%s skipped' %
 
623
                                (skipped, skipped != 1 and "s" or ""))
 
624
        if result.unsupported:
 
625
            for feature, count in sorted(result.unsupported.items()):
 
626
                self.stream.writeln("Missing feature '%s' skipped %d tests." %
 
627
                    (feature, count))
 
628
        result.finished()
 
629
        return result
652
630
 
653
631
 
654
632
def iter_suite_tests(suite):
664
642
                        % (type(suite), suite))
665
643
 
666
644
 
667
 
TestSkipped = testtools.testcase.TestSkipped
 
645
class TestSkipped(Exception):
 
646
    """Indicates that a test was intentionally skipped, rather than failing."""
668
647
 
669
648
 
670
649
class TestNotApplicable(TestSkipped):
676
655
    """
677
656
 
678
657
 
679
 
# traceback._some_str fails to format exceptions that have the default
680
 
# __str__ which does an implicit ascii conversion. However, repr() on those
681
 
# objects works, for all that its not quite what the doctor may have ordered.
682
 
def _clever_some_str(value):
683
 
    try:
684
 
        return str(value)
685
 
    except:
686
 
        try:
687
 
            return repr(value).replace('\\n', '\n')
688
 
        except:
689
 
            return '<unprintable %s object>' % type(value).__name__
690
 
 
691
 
traceback._some_str = _clever_some_str
692
 
 
693
 
 
694
 
# deprecated - use self.knownFailure(), or self.expectFailure.
695
 
KnownFailure = testtools.testcase._ExpectedFailure
 
658
class KnownFailure(AssertionError):
 
659
    """Indicates that a test failed in a precisely expected manner.
 
660
 
 
661
    Such failures dont block the whole test suite from passing because they are
 
662
    indicators of partially completed code or of future work. We have an
 
663
    explicit error for them so that we can ensure that they are always visible:
 
664
    KnownFailures are always shown in the output of bzr selftest.
 
665
    """
696
666
 
697
667
 
698
668
class UnavailableFeature(Exception):
699
669
    """A feature required for this test was not available.
700
670
 
701
 
    This can be considered a specialised form of SkippedTest.
702
 
 
703
671
    The feature should be used to construct the exception.
704
672
    """
705
673
 
706
674
 
 
675
class CommandFailed(Exception):
 
676
    pass
 
677
 
 
678
 
707
679
class StringIOWrapper(object):
708
680
    """A wrapper around cStringIO which just adds an encoding attribute.
709
681
 
730
702
            return setattr(self._cstring, name, val)
731
703
 
732
704
 
733
 
class TestUIFactory(TextUIFactory):
 
705
class TestUIFactory(ui.CLIUIFactory):
734
706
    """A UI Factory for testing.
735
707
 
736
708
    Hide the progress bar but emit note()s.
737
709
    Redirect stdin.
738
710
    Allows get_password to be tested without real tty attached.
739
 
 
740
 
    See also CannedInputUIFactory which lets you provide programmatic input in
741
 
    a structured way.
742
711
    """
743
 
    # TODO: Capture progress events at the model level and allow them to be
744
 
    # observed by tests that care.
745
 
    #
746
 
    # XXX: Should probably unify more with CannedInputUIFactory or a
747
 
    # particular configuration of TextUIFactory, or otherwise have a clearer
748
 
    # idea of how they're supposed to be different.
749
 
    # See https://bugs.launchpad.net/bzr/+bug/408213
750
712
 
751
713
    def __init__(self, stdout=None, stderr=None, stdin=None):
752
714
        if stdin is not None:
757
719
            stdin = StringIOWrapper(stdin)
758
720
        super(TestUIFactory, self).__init__(stdin, stdout, stderr)
759
721
 
 
722
    def clear(self):
 
723
        """See progress.ProgressBar.clear()."""
 
724
 
 
725
    def clear_term(self):
 
726
        """See progress.ProgressBar.clear_term()."""
 
727
 
 
728
    def finished(self):
 
729
        """See progress.ProgressBar.finished()."""
 
730
 
 
731
    def note(self, fmt_string, *args, **kwargs):
 
732
        """See progress.ProgressBar.note()."""
 
733
        self.stdout.write((fmt_string + "\n") % args)
 
734
 
 
735
    def progress_bar(self):
 
736
        return self
 
737
 
 
738
    def nested_progress_bar(self):
 
739
        return self
 
740
 
 
741
    def update(self, message, count=None, total=None):
 
742
        """See progress.ProgressBar.update()."""
 
743
 
760
744
    def get_non_echoed_password(self):
761
745
        """Get password from stdin without trying to handle the echo mode"""
762
746
        password = self.stdin.readline()
766
750
            password = password[:-1]
767
751
        return password
768
752
 
769
 
    def make_progress_view(self):
770
 
        return NullProgressView()
771
 
 
772
 
 
773
 
class TestCase(testtools.TestCase):
 
753
 
 
754
class TestCase(unittest.TestCase):
774
755
    """Base class for bzr unit tests.
775
756
 
776
757
    Tests that need access to disk resources should subclass
795
776
    _leaking_threads_tests = 0
796
777
    _first_thread_leaker_id = None
797
778
    _log_file_name = None
 
779
    _log_contents = ''
 
780
    _keep_log_file = False
798
781
    # record lsprof data when performing benchmark calls.
799
782
    _gather_lsprof_in_benchmarks = False
 
783
    attrs_to_keep = ('id', '_testMethodName', '_testMethodDoc',
 
784
                     '_log_contents', '_log_file_name', '_benchtime',
 
785
                     '_TestCase__testMethodName', '_TestCase__testMethodDoc',)
800
786
 
801
787
    def __init__(self, methodName='testMethod'):
802
788
        super(TestCase, self).__init__(methodName)
803
789
        self._cleanups = []
804
 
        self._directory_isolation = True
805
 
        self.exception_handlers.insert(0,
806
 
            (UnavailableFeature, self._do_unsupported_or_skip))
807
 
        self.exception_handlers.insert(0,
808
 
            (TestNotApplicable, self._do_not_applicable))
 
790
        self._bzr_test_setUp_run = False
 
791
        self._bzr_test_tearDown_run = False
809
792
 
810
793
    def setUp(self):
811
 
        super(TestCase, self).setUp()
812
 
        for feature in getattr(self, '_test_needs_features', []):
813
 
            self.requireFeature(feature)
814
 
        self._log_contents = None
815
 
        self.addDetail("log", content.Content(content.ContentType("text",
816
 
            "plain", {"charset": "utf8"}),
817
 
            lambda:[self._get_log(keep_log_file=True)]))
 
794
        unittest.TestCase.setUp(self)
 
795
        self._bzr_test_setUp_run = True
818
796
        self._cleanEnvironment()
819
797
        self._silenceUI()
820
798
        self._startLogFile()
821
799
        self._benchcalls = []
822
800
        self._benchtime = None
823
801
        self._clear_hooks()
824
 
        self._track_transports()
 
802
        # Track locks - needs to be called before _clear_debug_flags.
825
803
        self._track_locks()
826
804
        self._clear_debug_flags()
827
805
        TestCase._active_threads = threading.activeCount()
836
814
        active = threading.activeCount()
837
815
        leaked_threads = active - TestCase._active_threads
838
816
        TestCase._active_threads = active
839
 
        # If some tests make the number of threads *decrease*, we'll consider
840
 
        # that they are just observing old threads dieing, not agressively kill
841
 
        # random threads. So we don't report these tests as leaking. The risk
842
 
        # is that we have false positives that way (the test see 2 threads
843
 
        # going away but leak one) but it seems less likely than the actual
844
 
        # false positives (the test see threads going away and does not leak).
845
 
        if leaked_threads > 0:
846
 
            if 'threads' in selftest_debug_flags:
847
 
                print '%s is leaking, active is now %d' % (self.id(), active)
 
817
        if leaked_threads:
848
818
            TestCase._leaking_threads_tests += 1
849
819
            if TestCase._first_thread_leaker_id is None:
850
820
                TestCase._first_thread_leaker_id = self.id()
855
825
        Tests that want to use debug flags can just set them in the
856
826
        debug_flags set during setup/teardown.
857
827
        """
858
 
        # Start with a copy of the current debug flags we can safely modify.
859
 
        self.overrideAttr(debug, 'debug_flags', set(debug.debug_flags))
 
828
        self._preserved_debug_flags = set(debug.debug_flags)
860
829
        if 'allow_debug' not in selftest_debug_flags:
861
830
            debug.debug_flags.clear()
862
 
        if 'disable_lock_checks' not in selftest_debug_flags:
863
 
            debug.debug_flags.add('strict_locks')
 
831
        self.addCleanup(self._restore_debug_flags)
864
832
 
865
833
    def _clear_hooks(self):
866
834
        # prevent hooks affecting tests
876
844
        # this hook should always be installed
877
845
        request._install_hook()
878
846
 
879
 
    def disable_directory_isolation(self):
880
 
        """Turn off directory isolation checks."""
881
 
        self._directory_isolation = False
882
 
 
883
 
    def enable_directory_isolation(self):
884
 
        """Enable directory isolation checks."""
885
 
        self._directory_isolation = True
886
 
 
887
847
    def _silenceUI(self):
888
848
        """Turn off UI for duration of test"""
889
849
        # by default the UI is off; tests can turn it on if they want it.
890
 
        self.overrideAttr(ui, 'ui_factory', ui.SilentUIFactory())
 
850
        saved = ui.ui_factory
 
851
        def _restore():
 
852
            ui.ui_factory = saved
 
853
        ui.ui_factory = ui.SilentUIFactory()
 
854
        self.addCleanup(_restore)
891
855
 
892
856
    def _check_locks(self):
893
857
        """Check that all lock take/release actions have been paired."""
894
 
        # We always check for mismatched locks. If a mismatch is found, we
895
 
        # fail unless -Edisable_lock_checks is supplied to selftest, in which
896
 
        # case we just print a warning.
 
858
        # once we have fixed all the current lock problems, we can change the
 
859
        # following code to always check for mismatched locks, but only do
 
860
        # traceback showing with -Dlock (self._lock_check_thorough is True).
 
861
        # For now, because the test suite will fail, we only assert that lock
 
862
        # matching has occured with -Dlock.
897
863
        # unhook:
898
864
        acquired_locks = [lock for action, lock in self._lock_actions
899
865
                          if action == 'acquired']
918
884
    def _track_locks(self):
919
885
        """Track lock activity during tests."""
920
886
        self._lock_actions = []
921
 
        if 'disable_lock_checks' in selftest_debug_flags:
922
 
            self._lock_check_thorough = False
923
 
        else:
924
 
            self._lock_check_thorough = True
925
 
 
 
887
        self._lock_check_thorough = 'lock' not in debug.debug_flags
926
888
        self.addCleanup(self._check_locks)
927
889
        _mod_lock.Lock.hooks.install_named_hook('lock_acquired',
928
890
                                                self._lock_acquired, None)
940
902
    def _lock_broken(self, result):
941
903
        self._lock_actions.append(('broken', result))
942
904
 
943
 
    def permit_dir(self, name):
944
 
        """Permit a directory to be used by this test. See permit_url."""
945
 
        name_transport = _mod_transport.get_transport(name)
946
 
        self.permit_url(name)
947
 
        self.permit_url(name_transport.base)
948
 
 
949
 
    def permit_url(self, url):
950
 
        """Declare that url is an ok url to use in this test.
951
 
        
952
 
        Do this for memory transports, temporary test directory etc.
953
 
        
954
 
        Do not do this for the current working directory, /tmp, or any other
955
 
        preexisting non isolated url.
956
 
        """
957
 
        if not url.endswith('/'):
958
 
            url += '/'
959
 
        self._bzr_selftest_roots.append(url)
960
 
 
961
 
    def permit_source_tree_branch_repo(self):
962
 
        """Permit the source tree bzr is running from to be opened.
963
 
 
964
 
        Some code such as bzrlib.version attempts to read from the bzr branch
965
 
        that bzr is executing from (if any). This method permits that directory
966
 
        to be used in the test suite.
967
 
        """
968
 
        path = self.get_source_path()
969
 
        self.record_directory_isolation()
970
 
        try:
971
 
            try:
972
 
                workingtree.WorkingTree.open(path)
973
 
            except (errors.NotBranchError, errors.NoWorkingTree):
974
 
                return
975
 
        finally:
976
 
            self.enable_directory_isolation()
977
 
 
978
 
    def _preopen_isolate_transport(self, transport):
979
 
        """Check that all transport openings are done in the test work area."""
980
 
        while isinstance(transport, pathfilter.PathFilteringTransport):
981
 
            # Unwrap pathfiltered transports
982
 
            transport = transport.server.backing_transport.clone(
983
 
                transport._filter('.'))
984
 
        url = transport.base
985
 
        # ReadonlySmartTCPServer_for_testing decorates the backing transport
986
 
        # urls it is given by prepending readonly+. This is appropriate as the
987
 
        # client shouldn't know that the server is readonly (or not readonly).
988
 
        # We could register all servers twice, with readonly+ prepending, but
989
 
        # that makes for a long list; this is about the same but easier to
990
 
        # read.
991
 
        if url.startswith('readonly+'):
992
 
            url = url[len('readonly+'):]
993
 
        self._preopen_isolate_url(url)
994
 
 
995
 
    def _preopen_isolate_url(self, url):
996
 
        if not self._directory_isolation:
997
 
            return
998
 
        if self._directory_isolation == 'record':
999
 
            self._bzr_selftest_roots.append(url)
1000
 
            return
1001
 
        # This prevents all transports, including e.g. sftp ones backed on disk
1002
 
        # from working unless they are explicitly granted permission. We then
1003
 
        # depend on the code that sets up test transports to check that they are
1004
 
        # appropriately isolated and enable their use by calling
1005
 
        # self.permit_transport()
1006
 
        if not osutils.is_inside_any(self._bzr_selftest_roots, url):
1007
 
            raise errors.BzrError("Attempt to escape test isolation: %r %r"
1008
 
                % (url, self._bzr_selftest_roots))
1009
 
 
1010
 
    def record_directory_isolation(self):
1011
 
        """Gather accessed directories to permit later access.
1012
 
        
1013
 
        This is used for tests that access the branch bzr is running from.
1014
 
        """
1015
 
        self._directory_isolation = "record"
1016
 
 
1017
 
    def start_server(self, transport_server, backing_server=None):
1018
 
        """Start transport_server for this test.
1019
 
 
1020
 
        This starts the server, registers a cleanup for it and permits the
1021
 
        server's urls to be used.
1022
 
        """
1023
 
        if backing_server is None:
1024
 
            transport_server.start_server()
1025
 
        else:
1026
 
            transport_server.start_server(backing_server)
1027
 
        self.addCleanup(transport_server.stop_server)
1028
 
        # Obtain a real transport because if the server supplies a password, it
1029
 
        # will be hidden from the base on the client side.
1030
 
        t = _mod_transport.get_transport(transport_server.get_url())
1031
 
        # Some transport servers effectively chroot the backing transport;
1032
 
        # others like SFTPServer don't - users of the transport can walk up the
1033
 
        # transport to read the entire backing transport. This wouldn't matter
1034
 
        # except that the workdir tests are given - and that they expect the
1035
 
        # server's url to point at - is one directory under the safety net. So
1036
 
        # Branch operations into the transport will attempt to walk up one
1037
 
        # directory. Chrooting all servers would avoid this but also mean that
1038
 
        # we wouldn't be testing directly against non-root urls. Alternatively
1039
 
        # getting the test framework to start the server with a backing server
1040
 
        # at the actual safety net directory would work too, but this then
1041
 
        # means that the self.get_url/self.get_transport methods would need
1042
 
        # to transform all their results. On balance its cleaner to handle it
1043
 
        # here, and permit a higher url when we have one of these transports.
1044
 
        if t.base.endswith('/work/'):
1045
 
            # we have safety net/test root/work
1046
 
            t = t.clone('../..')
1047
 
        elif isinstance(transport_server,
1048
 
                        test_server.SmartTCPServer_for_testing):
1049
 
            # The smart server adds a path similar to work, which is traversed
1050
 
            # up from by the client. But the server is chrooted - the actual
1051
 
            # backing transport is not escaped from, and VFS requests to the
1052
 
            # root will error (because they try to escape the chroot).
1053
 
            t2 = t.clone('..')
1054
 
            while t2.base != t.base:
1055
 
                t = t2
1056
 
                t2 = t.clone('..')
1057
 
        self.permit_url(t.base)
1058
 
 
1059
 
    def _track_transports(self):
1060
 
        """Install checks for transport usage."""
1061
 
        # TestCase has no safe place it can write to.
1062
 
        self._bzr_selftest_roots = []
1063
 
        # Currently the easiest way to be sure that nothing is going on is to
1064
 
        # hook into bzr dir opening. This leaves a small window of error for
1065
 
        # transport tests, but they are well known, and we can improve on this
1066
 
        # step.
1067
 
        bzrdir.BzrDir.hooks.install_named_hook("pre_open",
1068
 
            self._preopen_isolate_transport, "Check bzr directories are safe.")
1069
 
 
1070
905
    def _ndiff_strings(self, a, b):
1071
906
        """Return ndiff between two strings containing lines.
1072
907
 
1094
929
            message += '\n'
1095
930
        raise AssertionError("%snot equal:\na = %s\nb = %s\n"
1096
931
            % (message,
1097
 
               pprint.pformat(a), pprint.pformat(b)))
 
932
               pformat(a), pformat(b)))
1098
933
 
1099
934
    assertEquals = assertEqual
1100
935
 
1109
944
            return
1110
945
        if message is None:
1111
946
            message = "texts not equal:\n"
 
947
        if a == b + '\n':
 
948
            message = 'first string is missing a final newline.\n'
1112
949
        if a + '\n' == b:
1113
 
            message = 'first string is missing a final newline.\n'
1114
 
        if a == b + '\n':
1115
950
            message = 'second string is missing a final newline.\n'
1116
951
        raise AssertionError(message +
1117
952
                             self._ndiff_strings(a, b))
1128
963
        :raises AssertionError: If the expected and actual stat values differ
1129
964
            other than by atime.
1130
965
        """
1131
 
        self.assertEqual(expected.st_size, actual.st_size,
1132
 
                         'st_size did not match')
1133
 
        self.assertEqual(expected.st_mtime, actual.st_mtime,
1134
 
                         'st_mtime did not match')
1135
 
        self.assertEqual(expected.st_ctime, actual.st_ctime,
1136
 
                         'st_ctime did not match')
1137
 
        if sys.platform != 'win32':
1138
 
            # On Win32 both 'dev' and 'ino' cannot be trusted. In python2.4 it
1139
 
            # is 'dev' that varies, in python 2.5 (6?) it is st_ino that is
1140
 
            # odd. Regardless we shouldn't actually try to assert anything
1141
 
            # about their values
1142
 
            self.assertEqual(expected.st_dev, actual.st_dev,
1143
 
                             'st_dev did not match')
1144
 
            self.assertEqual(expected.st_ino, actual.st_ino,
1145
 
                             'st_ino did not match')
1146
 
        self.assertEqual(expected.st_mode, actual.st_mode,
1147
 
                         'st_mode did not match')
 
966
        self.assertEqual(expected.st_size, actual.st_size)
 
967
        self.assertEqual(expected.st_mtime, actual.st_mtime)
 
968
        self.assertEqual(expected.st_ctime, actual.st_ctime)
 
969
        self.assertEqual(expected.st_dev, actual.st_dev)
 
970
        self.assertEqual(expected.st_ino, actual.st_ino)
 
971
        self.assertEqual(expected.st_mode, actual.st_mode)
1148
972
 
1149
973
    def assertLength(self, length, obj_with_len):
1150
974
        """Assert that obj_with_len is of length length."""
1152
976
            self.fail("Incorrect length: wanted %d, got %d for %r" % (
1153
977
                length, len(obj_with_len), obj_with_len))
1154
978
 
1155
 
    def assertLogsError(self, exception_class, func, *args, **kwargs):
1156
 
        """Assert that func(*args, **kwargs) quietly logs a specific exception.
1157
 
        """
1158
 
        from bzrlib import trace
1159
 
        captured = []
1160
 
        orig_log_exception_quietly = trace.log_exception_quietly
1161
 
        try:
1162
 
            def capture():
1163
 
                orig_log_exception_quietly()
1164
 
                captured.append(sys.exc_info())
1165
 
            trace.log_exception_quietly = capture
1166
 
            func(*args, **kwargs)
1167
 
        finally:
1168
 
            trace.log_exception_quietly = orig_log_exception_quietly
1169
 
        self.assertLength(1, captured)
1170
 
        err = captured[0][1]
1171
 
        self.assertIsInstance(err, exception_class)
1172
 
        return err
1173
 
 
1174
979
    def assertPositive(self, val):
1175
980
        """Assert that val is greater than 0."""
1176
981
        self.assertTrue(val > 0, 'expected a positive value, but got %s' % val)
1206
1011
            raise AssertionError('pattern "%s" found in "%s"'
1207
1012
                    % (needle_re, haystack))
1208
1013
 
1209
 
    def assertContainsString(self, haystack, needle):
1210
 
        if haystack.find(needle) == -1:
1211
 
            self.fail("string %r not found in '''%s'''" % (needle, haystack))
1212
 
 
1213
1014
    def assertSubset(self, sublist, superlist):
1214
1015
        """Assert that every entry in sublist is present in superlist."""
1215
1016
        missing = set(sublist) - set(superlist)
1290
1091
                         osutils.realpath(path2),
1291
1092
                         "apparent paths:\na = %s\nb = %s\n," % (path1, path2))
1292
1093
 
1293
 
    def assertIsInstance(self, obj, kls, msg=None):
1294
 
        """Fail if obj is not an instance of kls
1295
 
        
1296
 
        :param msg: Supplementary message to show if the assertion fails.
1297
 
        """
 
1094
    def assertIsInstance(self, obj, kls):
 
1095
        """Fail if obj is not an instance of kls"""
1298
1096
        if not isinstance(obj, kls):
1299
 
            m = "%r is an instance of %s rather than %s" % (
1300
 
                obj, obj.__class__, kls)
1301
 
            if msg:
1302
 
                m += ": " + msg
1303
 
            self.fail(m)
 
1097
            self.fail("%r is an instance of %s rather than %s" % (
 
1098
                obj, obj.__class__, kls))
 
1099
 
 
1100
    def expectFailure(self, reason, assertion, *args, **kwargs):
 
1101
        """Invoke a test, expecting it to fail for the given reason.
 
1102
 
 
1103
        This is for assertions that ought to succeed, but currently fail.
 
1104
        (The failure is *expected* but not *wanted*.)  Please be very precise
 
1105
        about the failure you're expecting.  If a new bug is introduced,
 
1106
        AssertionError should be raised, not KnownFailure.
 
1107
 
 
1108
        Frequently, expectFailure should be followed by an opposite assertion.
 
1109
        See example below.
 
1110
 
 
1111
        Intended to be used with a callable that raises AssertionError as the
 
1112
        'assertion' parameter.  args and kwargs are passed to the 'assertion'.
 
1113
 
 
1114
        Raises KnownFailure if the test fails.  Raises AssertionError if the
 
1115
        test succeeds.
 
1116
 
 
1117
        example usage::
 
1118
 
 
1119
          self.expectFailure('Math is broken', self.assertNotEqual, 54,
 
1120
                             dynamic_val)
 
1121
          self.assertEqual(42, dynamic_val)
 
1122
 
 
1123
          This means that a dynamic_val of 54 will cause the test to raise
 
1124
          a KnownFailure.  Once math is fixed and the expectFailure is removed,
 
1125
          only a dynamic_val of 42 will allow the test to pass.  Anything other
 
1126
          than 54 or 42 will cause an AssertionError.
 
1127
        """
 
1128
        try:
 
1129
            assertion(*args, **kwargs)
 
1130
        except AssertionError:
 
1131
            raise KnownFailure(reason)
 
1132
        else:
 
1133
            self.fail('Unexpected success.  Should have failed: %s' % reason)
1304
1134
 
1305
1135
    def assertFileEqual(self, content, path):
1306
1136
        """Fail if path does not contain 'content'."""
1312
1142
            f.close()
1313
1143
        self.assertEqualDiff(content, s)
1314
1144
 
1315
 
    def assertDocstring(self, expected_docstring, obj):
1316
 
        """Fail if obj does not have expected_docstring"""
1317
 
        if __doc__ is None:
1318
 
            # With -OO the docstring should be None instead
1319
 
            self.assertIs(obj.__doc__, None)
1320
 
        else:
1321
 
            self.assertEqual(expected_docstring, obj.__doc__)
1322
 
 
1323
1145
    def failUnlessExists(self, path):
1324
1146
        """Fail unless path or paths, which may be abs or relative, exist."""
1325
1147
        if not isinstance(path, basestring):
1465
1287
 
1466
1288
        Close the file and delete it, unless setKeepLogfile was called.
1467
1289
        """
1468
 
        if bzrlib.trace._trace_file:
1469
 
            # flush the log file, to get all content
1470
 
            bzrlib.trace._trace_file.flush()
 
1290
        if self._log_file is None:
 
1291
            return
1471
1292
        bzrlib.trace.pop_log_file(self._log_memento)
1472
 
        # Cache the log result and delete the file on disk
1473
 
        self._get_log(False)
1474
 
 
1475
 
    def thisFailsStrictLockCheck(self):
1476
 
        """It is known that this test would fail with -Dstrict_locks.
1477
 
 
1478
 
        By default, all tests are run with strict lock checking unless
1479
 
        -Edisable_lock_checks is supplied. However there are some tests which
1480
 
        we know fail strict locks at this point that have not been fixed.
1481
 
        They should call this function to disable the strict checking.
1482
 
 
1483
 
        This should be used sparingly, it is much better to fix the locking
1484
 
        issues rather than papering over the problem by calling this function.
1485
 
        """
1486
 
        debug.debug_flags.discard('strict_locks')
 
1293
        self._log_file.close()
 
1294
        self._log_file = None
 
1295
        if not self._keep_log_file:
 
1296
            os.remove(self._log_file_name)
 
1297
            self._log_file_name = None
 
1298
 
 
1299
    def setKeepLogfile(self):
 
1300
        """Make the logfile not be deleted when _finishLogFile is called."""
 
1301
        self._keep_log_file = True
1487
1302
 
1488
1303
    def addCleanup(self, callable, *args, **kwargs):
1489
1304
        """Arrange to run a callable when this case is torn down.
1493
1308
        """
1494
1309
        self._cleanups.append((callable, args, kwargs))
1495
1310
 
1496
 
    def overrideAttr(self, obj, attr_name, new=_unitialized_attr):
1497
 
        """Overrides an object attribute restoring it after the test.
1498
 
 
1499
 
        :param obj: The object that will be mutated.
1500
 
 
1501
 
        :param attr_name: The attribute name we want to preserve/override in
1502
 
            the object.
1503
 
 
1504
 
        :param new: The optional value we want to set the attribute to.
1505
 
 
1506
 
        :returns: The actual attr value.
1507
 
        """
1508
 
        value = getattr(obj, attr_name)
1509
 
        # The actual value is captured by the call below
1510
 
        self.addCleanup(setattr, obj, attr_name, value)
1511
 
        if new is not _unitialized_attr:
1512
 
            setattr(obj, attr_name, new)
1513
 
        return value
1514
 
 
1515
1311
    def _cleanEnvironment(self):
1516
1312
        new_env = {
1517
1313
            'BZR_HOME': None, # Don't inherit BZR_HOME to all the tests.
1523
1319
            'EDITOR': None,
1524
1320
            'BZR_EMAIL': None,
1525
1321
            'BZREMAIL': None, # may still be present in the environment
1526
 
            'EMAIL': 'jrandom@example.com', # set EMAIL as bzr does not guess
 
1322
            'EMAIL': None,
1527
1323
            'BZR_PROGRESS_BAR': None,
1528
1324
            'BZR_LOG': None,
1529
1325
            'BZR_PLUGIN_PATH': None,
1530
 
            'BZR_DISABLE_PLUGINS': None,
1531
 
            'BZR_PLUGINS_AT': None,
1532
 
            'BZR_CONCURRENCY': None,
1533
 
            # Make sure that any text ui tests are consistent regardless of
1534
 
            # the environment the test case is run in; you may want tests that
1535
 
            # test other combinations.  'dumb' is a reasonable guess for tests
1536
 
            # going to a pipe or a StringIO.
1537
 
            'TERM': 'dumb',
1538
 
            'LINES': '25',
1539
 
            'COLUMNS': '80',
1540
 
            'BZR_COLUMNS': '80',
1541
1326
            # SSH Agent
1542
1327
            'SSH_AUTH_SOCK': None,
1543
1328
            # Proxies
1555
1340
            'ftp_proxy': None,
1556
1341
            'FTP_PROXY': None,
1557
1342
            'BZR_REMOTE_PATH': None,
1558
 
            # Generally speaking, we don't want apport reporting on crashes in
1559
 
            # the test envirnoment unless we're specifically testing apport,
1560
 
            # so that it doesn't leak into the real system environment.  We
1561
 
            # use an env var so it propagates to subprocesses.
1562
 
            'APPORT_DISABLE': '1',
1563
1343
        }
1564
 
        self._old_env = {}
 
1344
        self.__old_env = {}
1565
1345
        self.addCleanup(self._restoreEnvironment)
1566
1346
        for name, value in new_env.iteritems():
1567
1347
            self._captureVar(name, value)
1568
1348
 
1569
1349
    def _captureVar(self, name, newvalue):
1570
1350
        """Set an environment variable, and reset it when finished."""
1571
 
        self._old_env[name] = osutils.set_or_unset_env(name, newvalue)
 
1351
        self.__old_env[name] = osutils.set_or_unset_env(name, newvalue)
 
1352
 
 
1353
    def _restore_debug_flags(self):
 
1354
        debug.debug_flags.clear()
 
1355
        debug.debug_flags.update(self._preserved_debug_flags)
1572
1356
 
1573
1357
    def _restoreEnvironment(self):
1574
 
        for name, value in self._old_env.iteritems():
 
1358
        for name, value in self.__old_env.iteritems():
1575
1359
            osutils.set_or_unset_env(name, value)
1576
1360
 
1577
1361
    def _restoreHooks(self):
1585
1369
    def _do_skip(self, result, reason):
1586
1370
        addSkip = getattr(result, 'addSkip', None)
1587
1371
        if not callable(addSkip):
1588
 
            result.addSuccess(result)
 
1372
            result.addError(self, sys.exc_info())
1589
1373
        else:
1590
1374
            addSkip(self, reason)
1591
1375
 
1592
 
    @staticmethod
1593
 
    def _do_known_failure(self, result, e):
1594
 
        err = sys.exc_info()
1595
 
        addExpectedFailure = getattr(result, 'addExpectedFailure', None)
1596
 
        if addExpectedFailure is not None:
1597
 
            addExpectedFailure(self, err)
1598
 
        else:
1599
 
            result.addSuccess(self)
1600
 
 
1601
 
    @staticmethod
1602
 
    def _do_not_applicable(self, result, e):
1603
 
        if not e.args:
1604
 
            reason = 'No reason given'
1605
 
        else:
1606
 
            reason = e.args[0]
1607
 
        addNotApplicable = getattr(result, 'addNotApplicable', None)
1608
 
        if addNotApplicable is not None:
1609
 
            result.addNotApplicable(self, reason)
1610
 
        else:
1611
 
            self._do_skip(result, reason)
1612
 
 
1613
 
    @staticmethod
1614
 
    def _do_unsupported_or_skip(self, result, e):
1615
 
        reason = e.args[0]
1616
 
        addNotSupported = getattr(result, 'addNotSupported', None)
1617
 
        if addNotSupported is not None:
1618
 
            result.addNotSupported(self, reason)
1619
 
        else:
1620
 
            self._do_skip(result, reason)
 
1376
    def run(self, result=None):
 
1377
        if result is None: result = self.defaultTestResult()
 
1378
        for feature in getattr(self, '_test_needs_features', []):
 
1379
            if not feature.available():
 
1380
                result.startTest(self)
 
1381
                if getattr(result, 'addNotSupported', None):
 
1382
                    result.addNotSupported(self, feature)
 
1383
                else:
 
1384
                    result.addSuccess(self)
 
1385
                result.stopTest(self)
 
1386
                return result
 
1387
        try:
 
1388
            try:
 
1389
                result.startTest(self)
 
1390
                absent_attr = object()
 
1391
                # Python 2.5
 
1392
                method_name = getattr(self, '_testMethodName', absent_attr)
 
1393
                if method_name is absent_attr:
 
1394
                    # Python 2.4
 
1395
                    method_name = getattr(self, '_TestCase__testMethodName')
 
1396
                testMethod = getattr(self, method_name)
 
1397
                try:
 
1398
                    try:
 
1399
                        self.setUp()
 
1400
                        if not self._bzr_test_setUp_run:
 
1401
                            self.fail(
 
1402
                                "test setUp did not invoke "
 
1403
                                "bzrlib.tests.TestCase's setUp")
 
1404
                    except KeyboardInterrupt:
 
1405
                        self._runCleanups()
 
1406
                        raise
 
1407
                    except TestSkipped, e:
 
1408
                        self._do_skip(result, e.args[0])
 
1409
                        self.tearDown()
 
1410
                        return result
 
1411
                    except:
 
1412
                        result.addError(self, sys.exc_info())
 
1413
                        self._runCleanups()
 
1414
                        return result
 
1415
 
 
1416
                    ok = False
 
1417
                    try:
 
1418
                        testMethod()
 
1419
                        ok = True
 
1420
                    except self.failureException:
 
1421
                        result.addFailure(self, sys.exc_info())
 
1422
                    except TestSkipped, e:
 
1423
                        if not e.args:
 
1424
                            reason = "No reason given."
 
1425
                        else:
 
1426
                            reason = e.args[0]
 
1427
                        self._do_skip(result, reason)
 
1428
                    except KeyboardInterrupt:
 
1429
                        self._runCleanups()
 
1430
                        raise
 
1431
                    except:
 
1432
                        result.addError(self, sys.exc_info())
 
1433
 
 
1434
                    try:
 
1435
                        self.tearDown()
 
1436
                        if not self._bzr_test_tearDown_run:
 
1437
                            self.fail(
 
1438
                                "test tearDown did not invoke "
 
1439
                                "bzrlib.tests.TestCase's tearDown")
 
1440
                    except KeyboardInterrupt:
 
1441
                        self._runCleanups()
 
1442
                        raise
 
1443
                    except:
 
1444
                        result.addError(self, sys.exc_info())
 
1445
                        self._runCleanups()
 
1446
                        ok = False
 
1447
                    if ok: result.addSuccess(self)
 
1448
                finally:
 
1449
                    result.stopTest(self)
 
1450
                return result
 
1451
            except TestNotApplicable:
 
1452
                # Not moved from the result [yet].
 
1453
                self._runCleanups()
 
1454
                raise
 
1455
            except KeyboardInterrupt:
 
1456
                self._runCleanups()
 
1457
                raise
 
1458
        finally:
 
1459
            saved_attrs = {}
 
1460
            for attr_name in self.attrs_to_keep:
 
1461
                if attr_name in self.__dict__:
 
1462
                    saved_attrs[attr_name] = self.__dict__[attr_name]
 
1463
            self.__dict__ = saved_attrs
 
1464
 
 
1465
    def tearDown(self):
 
1466
        self._runCleanups()
 
1467
        self._log_contents = ''
 
1468
        self._bzr_test_tearDown_run = True
 
1469
        unittest.TestCase.tearDown(self)
1621
1470
 
1622
1471
    def time(self, callable, *args, **kwargs):
1623
1472
        """Run callable and accrue the time it takes to the benchmark time.
1627
1476
        self._benchcalls.
1628
1477
        """
1629
1478
        if self._benchtime is None:
1630
 
            self.addDetail('benchtime', content.Content(content.ContentType(
1631
 
                "text", "plain"), lambda:[str(self._benchtime)]))
1632
1479
            self._benchtime = 0
1633
1480
        start = time.time()
1634
1481
        try:
1643
1490
        finally:
1644
1491
            self._benchtime += time.time() - start
1645
1492
 
 
1493
    def _runCleanups(self):
 
1494
        """Run registered cleanup functions.
 
1495
 
 
1496
        This should only be called from TestCase.tearDown.
 
1497
        """
 
1498
        # TODO: Perhaps this should keep running cleanups even if
 
1499
        # one of them fails?
 
1500
 
 
1501
        # Actually pop the cleanups from the list so tearDown running
 
1502
        # twice is safe (this happens for skipped tests).
 
1503
        while self._cleanups:
 
1504
            cleanup, args, kwargs = self._cleanups.pop()
 
1505
            cleanup(*args, **kwargs)
 
1506
 
1646
1507
    def log(self, *args):
1647
1508
        mutter(*args)
1648
1509
 
1649
1510
    def _get_log(self, keep_log_file=False):
1650
 
        """Internal helper to get the log from bzrlib.trace for this test.
1651
 
 
1652
 
        Please use self.getDetails, or self.get_log to access this in test case
1653
 
        code.
 
1511
        """Get the log from bzrlib.trace calls from this test.
1654
1512
 
1655
1513
        :param keep_log_file: When True, if the log is still a file on disk
1656
1514
            leave it as a file on disk. When False, if the log is still a file
1658
1516
            self._log_contents.
1659
1517
        :return: A string containing the log.
1660
1518
        """
1661
 
        if self._log_contents is not None:
1662
 
            try:
1663
 
                self._log_contents.decode('utf8')
1664
 
            except UnicodeDecodeError:
1665
 
                unicodestr = self._log_contents.decode('utf8', 'replace')
1666
 
                self._log_contents = unicodestr.encode('utf8')
 
1519
        # flush the log file, to get all content
 
1520
        import bzrlib.trace
 
1521
        if bzrlib.trace._trace_file:
 
1522
            bzrlib.trace._trace_file.flush()
 
1523
        if self._log_contents:
 
1524
            # XXX: this can hardly contain the content flushed above --vila
 
1525
            # 20080128
1667
1526
            return self._log_contents
1668
 
        import bzrlib.trace
1669
 
        if bzrlib.trace._trace_file:
1670
 
            # flush the log file, to get all content
1671
 
            bzrlib.trace._trace_file.flush()
1672
1527
        if self._log_file_name is not None:
1673
1528
            logfile = open(self._log_file_name)
1674
1529
            try:
1675
1530
                log_contents = logfile.read()
1676
1531
            finally:
1677
1532
                logfile.close()
1678
 
            try:
1679
 
                log_contents.decode('utf8')
1680
 
            except UnicodeDecodeError:
1681
 
                unicodestr = log_contents.decode('utf8', 'replace')
1682
 
                log_contents = unicodestr.encode('utf8')
1683
1533
            if not keep_log_file:
1684
 
                close_attempts = 0
1685
 
                max_close_attempts = 100
1686
 
                first_close_error = None
1687
 
                while close_attempts < max_close_attempts:
1688
 
                    close_attempts += 1
1689
 
                    try:
1690
 
                        self._log_file.close()
1691
 
                    except IOError, ioe:
1692
 
                        if ioe.errno is None:
1693
 
                            # No errno implies 'close() called during
1694
 
                            # concurrent operation on the same file object', so
1695
 
                            # retry.  Probably a thread is trying to write to
1696
 
                            # the log file.
1697
 
                            if first_close_error is None:
1698
 
                                first_close_error = ioe
1699
 
                            continue
1700
 
                        raise
1701
 
                    else:
1702
 
                        break
1703
 
                if close_attempts > 1:
1704
 
                    sys.stderr.write(
1705
 
                        'Unable to close log file on first attempt, '
1706
 
                        'will retry: %s\n' % (first_close_error,))
1707
 
                    if close_attempts == max_close_attempts:
1708
 
                        sys.stderr.write(
1709
 
                            'Unable to close log file after %d attempts.\n'
1710
 
                            % (max_close_attempts,))
1711
 
                self._log_file = None
1712
 
                # Permit multiple calls to get_log until we clean it up in
1713
 
                # finishLogFile
1714
1534
                self._log_contents = log_contents
1715
1535
                try:
1716
1536
                    os.remove(self._log_file_name)
1720
1540
                                             ' %r\n' % self._log_file_name))
1721
1541
                    else:
1722
1542
                        raise
1723
 
                self._log_file_name = None
1724
1543
            return log_contents
1725
1544
        else:
1726
 
            return "No log file content and no log file name."
1727
 
 
1728
 
    def get_log(self):
1729
 
        """Get a unicode string containing the log from bzrlib.trace.
1730
 
 
1731
 
        Undecodable characters are replaced.
1732
 
        """
1733
 
        return u"".join(self.getDetails()['log'].iter_text())
 
1545
            return "DELETED log file to reduce memory footprint"
1734
1546
 
1735
1547
    def requireFeature(self, feature):
1736
1548
        """This test requires a specific feature is available.
1753
1565
 
1754
1566
    def _run_bzr_core(self, args, retcode, encoding, stdin,
1755
1567
            working_dir):
1756
 
        # Clear chk_map page cache, because the contents are likely to mask
1757
 
        # locking errors.
1758
 
        chk_map.clear_cache()
1759
1568
        if encoding is None:
1760
1569
            encoding = osutils.get_user_encoding()
1761
1570
        stdout = StringIOWrapper()
1778
1587
            os.chdir(working_dir)
1779
1588
 
1780
1589
        try:
1781
 
            try:
1782
 
                result = self.apply_redirected(ui.ui_factory.stdin,
1783
 
                    stdout, stderr,
1784
 
                    bzrlib.commands.run_bzr_catch_user_errors,
1785
 
                    args)
1786
 
            except KeyboardInterrupt:
1787
 
                # Reraise KeyboardInterrupt with contents of redirected stdout
1788
 
                # and stderr as arguments, for tests which are interested in
1789
 
                # stdout and stderr and are expecting the exception.
1790
 
                out = stdout.getvalue()
1791
 
                err = stderr.getvalue()
1792
 
                if out:
1793
 
                    self.log('output:\n%r', out)
1794
 
                if err:
1795
 
                    self.log('errors:\n%r', err)
1796
 
                raise KeyboardInterrupt(out, err)
 
1590
            result = self.apply_redirected(ui.ui_factory.stdin,
 
1591
                stdout, stderr,
 
1592
                bzrlib.commands.run_bzr_catch_user_errors,
 
1593
                args)
1797
1594
        finally:
1798
1595
            logger.removeHandler(handler)
1799
1596
            ui.ui_factory = old_ui_factory
1809
1606
        if retcode is not None:
1810
1607
            self.assertEquals(retcode, result,
1811
1608
                              message='Unexpected return code')
1812
 
        return result, out, err
 
1609
        return out, err
1813
1610
 
1814
1611
    def run_bzr(self, args, retcode=0, encoding=None, stdin=None,
1815
1612
                working_dir=None, error_regexes=[], output_encoding=None):
1844
1641
        :keyword error_regexes: A list of expected error messages.  If
1845
1642
            specified they must be seen in the error output of the command.
1846
1643
        """
1847
 
        retcode, out, err = self._run_bzr_autosplit(
 
1644
        out, err = self._run_bzr_autosplit(
1848
1645
            args=args,
1849
1646
            retcode=retcode,
1850
1647
            encoding=encoding,
1985
1782
            if not allow_plugins:
1986
1783
                command.append('--no-plugins')
1987
1784
            command.extend(process_args)
1988
 
            process = self._popen(command, stdin=subprocess.PIPE,
1989
 
                                  stdout=subprocess.PIPE,
1990
 
                                  stderr=subprocess.PIPE)
 
1785
            process = self._popen(command, stdin=PIPE, stdout=PIPE, stderr=PIPE)
1991
1786
        finally:
1992
1787
            restore_environment()
1993
1788
            if cwd is not None:
2001
1796
        Allows tests to override this method to intercept the calls made to
2002
1797
        Popen for introspection.
2003
1798
        """
2004
 
        return subprocess.Popen(*args, **kwargs)
2005
 
 
2006
 
    def get_source_path(self):
2007
 
        """Return the path of the directory containing bzrlib."""
2008
 
        return os.path.dirname(os.path.dirname(bzrlib.__file__))
 
1799
        return Popen(*args, **kwargs)
2009
1800
 
2010
1801
    def get_bzr_path(self):
2011
1802
        """Return the path of the 'bzr' executable for this test suite."""
2012
 
        bzr_path = os.path.join(self.get_source_path(), "bzr")
 
1803
        bzr_path = os.path.dirname(os.path.dirname(bzrlib.__file__))+'/bzr'
2013
1804
        if not os.path.isfile(bzr_path):
2014
1805
            # We are probably installed. Assume sys.argv is the right file
2015
1806
            bzr_path = sys.argv[0]
2101
1892
 
2102
1893
        Tests that expect to provoke LockContention errors should call this.
2103
1894
        """
2104
 
        self.overrideAttr(bzrlib.lockdir, '_DEFAULT_TIMEOUT_SECONDS', 0)
 
1895
        orig_timeout = bzrlib.lockdir._DEFAULT_TIMEOUT_SECONDS
 
1896
        def resetTimeout():
 
1897
            bzrlib.lockdir._DEFAULT_TIMEOUT_SECONDS = orig_timeout
 
1898
        self.addCleanup(resetTimeout)
 
1899
        bzrlib.lockdir._DEFAULT_TIMEOUT_SECONDS = 0
2105
1900
 
2106
1901
    def make_utf8_encoded_stringio(self, encoding_type=None):
2107
1902
        """Return a StringIOWrapper instance, that will encode Unicode
2115
1910
        sio.encoding = output_encoding
2116
1911
        return sio
2117
1912
 
2118
 
    def disable_verb(self, verb):
2119
 
        """Disable a smart server verb for one test."""
2120
 
        from bzrlib.smart import request
2121
 
        request_handlers = request.request_handlers
2122
 
        orig_method = request_handlers.get(verb)
2123
 
        request_handlers.remove(verb)
2124
 
        self.addCleanup(request_handlers.register, verb, orig_method)
2125
 
 
2126
1913
 
2127
1914
class CapturedCall(object):
2128
1915
    """A helper for capturing smart server calls for easy debug analysis."""
2187
1974
 
2188
1975
        :param relpath: a path relative to the base url.
2189
1976
        """
2190
 
        t = _mod_transport.get_transport(self.get_url(relpath))
 
1977
        t = get_transport(self.get_url(relpath))
2191
1978
        self.assertFalse(t.is_readonly())
2192
1979
        return t
2193
1980
 
2199
1986
 
2200
1987
        :param relpath: a path relative to the base url.
2201
1988
        """
2202
 
        t = _mod_transport.get_transport(self.get_readonly_url(relpath))
 
1989
        t = get_transport(self.get_readonly_url(relpath))
2203
1990
        self.assertTrue(t.is_readonly())
2204
1991
        return t
2205
1992
 
2218
2005
        if self.__readonly_server is None:
2219
2006
            if self.transport_readonly_server is None:
2220
2007
                # readonly decorator requested
2221
 
                self.__readonly_server = test_server.ReadonlyServer()
 
2008
                # bring up the server
 
2009
                self.__readonly_server = ReadonlyServer()
 
2010
                self.__readonly_server.setUp(self.get_vfs_only_server())
2222
2011
            else:
2223
 
                # explicit readonly transport.
2224
2012
                self.__readonly_server = self.create_transport_readonly_server()
2225
 
            self.start_server(self.__readonly_server,
2226
 
                self.get_vfs_only_server())
 
2013
                self.__readonly_server.setUp(self.get_vfs_only_server())
 
2014
            self.addCleanup(self.__readonly_server.tearDown)
2227
2015
        return self.__readonly_server
2228
2016
 
2229
2017
    def get_readonly_url(self, relpath=None):
2247
2035
        is no means to override it.
2248
2036
        """
2249
2037
        if self.__vfs_server is None:
2250
 
            self.__vfs_server = memory.MemoryServer()
2251
 
            self.start_server(self.__vfs_server)
 
2038
            self.__vfs_server = MemoryServer()
 
2039
            self.__vfs_server.setUp()
 
2040
            self.addCleanup(self.__vfs_server.tearDown)
2252
2041
        return self.__vfs_server
2253
2042
 
2254
2043
    def get_server(self):
2261
2050
        then the self.get_vfs_server is returned.
2262
2051
        """
2263
2052
        if self.__server is None:
2264
 
            if (self.transport_server is None or self.transport_server is
2265
 
                self.vfs_transport_factory):
2266
 
                self.__server = self.get_vfs_only_server()
 
2053
            if self.transport_server is None or self.transport_server is self.vfs_transport_factory:
 
2054
                return self.get_vfs_only_server()
2267
2055
            else:
2268
2056
                # bring up a decorated means of access to the vfs only server.
2269
2057
                self.__server = self.transport_server()
2270
 
                self.start_server(self.__server, self.get_vfs_only_server())
 
2058
                try:
 
2059
                    self.__server.setUp(self.get_vfs_only_server())
 
2060
                except TypeError, e:
 
2061
                    # This should never happen; the try:Except here is to assist
 
2062
                    # developers having to update code rather than seeing an
 
2063
                    # uninformative TypeError.
 
2064
                    raise Exception, "Old server API in use: %s, %s" % (self.__server, e)
 
2065
            self.addCleanup(self.__server.tearDown)
2271
2066
        return self.__server
2272
2067
 
2273
2068
    def _adjust_url(self, base, relpath):
2335
2130
        propagating. This method ensures than a test did not leaked.
2336
2131
        """
2337
2132
        root = TestCaseWithMemoryTransport.TEST_ROOT
2338
 
        self.permit_url(_mod_transport.get_transport(root).base)
2339
2133
        wt = workingtree.WorkingTree.open(root)
2340
2134
        last_rev = wt.last_revision()
2341
2135
        if last_rev != 'null:':
2343
2137
            # recreate a new one or all the followng tests will fail.
2344
2138
            # If you need to inspect its content uncomment the following line
2345
2139
            # import pdb; pdb.set_trace()
2346
 
            _rmtree_temp_dir(root + '/.bzr', test_id=self.id())
 
2140
            _rmtree_temp_dir(root + '/.bzr')
2347
2141
            self._create_safety_net()
2348
2142
            raise AssertionError('%s/.bzr should not be modified' % root)
2349
2143
 
2350
2144
    def _make_test_root(self):
2351
2145
        if TestCaseWithMemoryTransport.TEST_ROOT is None:
2352
 
            # Watch out for tricky test dir (on OSX /tmp -> /private/tmp)
2353
 
            root = osutils.realpath(osutils.mkdtemp(prefix='testbzr-',
2354
 
                                                    suffix='.tmp'))
 
2146
            root = osutils.mkdtemp(prefix='testbzr-', suffix='.tmp')
2355
2147
            TestCaseWithMemoryTransport.TEST_ROOT = root
2356
2148
 
2357
2149
            self._create_safety_net()
2360
2152
            # specifically told when all tests are finished.  This will do.
2361
2153
            atexit.register(_rmtree_temp_dir, root)
2362
2154
 
2363
 
        self.permit_dir(TestCaseWithMemoryTransport.TEST_ROOT)
2364
2155
        self.addCleanup(self._check_safety_net)
2365
2156
 
2366
2157
    def makeAndChdirToTestDir(self):
2374
2165
        os.chdir(TestCaseWithMemoryTransport.TEST_ROOT)
2375
2166
        self.test_dir = TestCaseWithMemoryTransport.TEST_ROOT
2376
2167
        self.test_home_dir = self.test_dir + "/MemoryTransportMissingHomeDir"
2377
 
        self.permit_dir(self.test_dir)
2378
2168
 
2379
2169
    def make_branch(self, relpath, format=None):
2380
2170
        """Create a branch on the transport at relpath."""
2386
2176
            # might be a relative or absolute path
2387
2177
            maybe_a_url = self.get_url(relpath)
2388
2178
            segments = maybe_a_url.rsplit('/', 1)
2389
 
            t = _mod_transport.get_transport(maybe_a_url)
 
2179
            t = get_transport(maybe_a_url)
2390
2180
            if len(segments) > 1 and segments[-1] not in ('', '.'):
2391
2181
                t.ensure_base()
2392
2182
            if format is None:
2409
2199
        made_control = self.make_bzrdir(relpath, format=format)
2410
2200
        return made_control.create_repository(shared=shared)
2411
2201
 
2412
 
    def make_smart_server(self, path, backing_server=None):
2413
 
        if backing_server is None:
2414
 
            backing_server = self.get_server()
2415
 
        smart_server = test_server.SmartTCPServer_for_testing()
2416
 
        self.start_server(smart_server, backing_server)
2417
 
        remote_transport = _mod_transport.get_transport(smart_server.get_url()
2418
 
                                                   ).clone(path)
 
2202
    def make_smart_server(self, path):
 
2203
        smart_server = server.SmartTCPServer_for_testing()
 
2204
        smart_server.setUp(self.get_server())
 
2205
        remote_transport = get_transport(smart_server.get_url()).clone(path)
 
2206
        self.addCleanup(smart_server.tearDown)
2419
2207
        return remote_transport
2420
2208
 
2421
2209
    def make_branch_and_memory_tree(self, relpath, format=None):
2428
2216
        return branchbuilder.BranchBuilder(branch=branch)
2429
2217
 
2430
2218
    def overrideEnvironmentForTesting(self):
2431
 
        test_home_dir = self.test_home_dir
2432
 
        if isinstance(test_home_dir, unicode):
2433
 
            test_home_dir = test_home_dir.encode(sys.getfilesystemencoding())
2434
 
        os.environ['HOME'] = test_home_dir
2435
 
        os.environ['BZR_HOME'] = test_home_dir
 
2219
        os.environ['HOME'] = self.test_home_dir
 
2220
        os.environ['BZR_HOME'] = self.test_home_dir
2436
2221
 
2437
2222
    def setUp(self):
2438
2223
        super(TestCaseWithMemoryTransport, self).setUp()
2439
 
        # Ensure that ConnectedTransport doesn't leak sockets
2440
 
        def get_transport_with_cleanup(*args, **kwargs):
2441
 
            t = orig_get_transport(*args, **kwargs)
2442
 
            if isinstance(t, _mod_transport.ConnectedTransport):
2443
 
                self.addCleanup(t.disconnect)
2444
 
            return t
2445
 
 
2446
 
        orig_get_transport = self.overrideAttr(_mod_transport, 'get_transport',
2447
 
                                               get_transport_with_cleanup)
2448
2224
        self._make_test_root()
2449
 
        self.addCleanup(os.chdir, os.getcwdu())
 
2225
        _currentdir = os.getcwdu()
 
2226
        def _leaveDirectory():
 
2227
            os.chdir(_currentdir)
 
2228
        self.addCleanup(_leaveDirectory)
2450
2229
        self.makeAndChdirToTestDir()
2451
2230
        self.overrideEnvironmentForTesting()
2452
2231
        self.__readonly_server = None
2455
2234
 
2456
2235
    def setup_smart_server_with_call_log(self):
2457
2236
        """Sets up a smart server as the transport server with a call log."""
2458
 
        self.transport_server = test_server.SmartTCPServer_for_testing
 
2237
        self.transport_server = server.SmartTCPServer_for_testing
2459
2238
        self.hpss_calls = []
2460
2239
        import traceback
2461
2240
        # Skip the current stack down to the caller of
2495
2274
 
2496
2275
    def check_file_contents(self, filename, expect):
2497
2276
        self.log("check contents of file %s" % filename)
2498
 
        f = file(filename)
2499
 
        try:
2500
 
            contents = f.read()
2501
 
        finally:
2502
 
            f.close()
 
2277
        contents = file(filename, 'r').read()
2503
2278
        if contents != expect:
2504
2279
            self.log("expected: %r" % expect)
2505
2280
            self.log("actually: %r" % contents)
2507
2282
 
2508
2283
    def _getTestDirPrefix(self):
2509
2284
        # create a directory within the top level test directory
2510
 
        if sys.platform in ('win32', 'cygwin'):
 
2285
        if sys.platform == 'win32':
2511
2286
            name_prefix = re.sub('[<>*=+",:;_/\\-]', '_', self.id())
2512
2287
            # windows is likely to have path-length limits so use a short name
2513
2288
            name_prefix = name_prefix[-30:]
2528
2303
            if os.path.exists(name):
2529
2304
                name = name_prefix + '_' + str(i)
2530
2305
            else:
2531
 
                # now create test and home directories within this dir
2532
 
                self.test_base_dir = name
2533
 
                self.addCleanup(self.deleteTestDir)
2534
 
                os.mkdir(self.test_base_dir)
 
2306
                os.mkdir(name)
2535
2307
                break
2536
 
        self.permit_dir(self.test_base_dir)
2537
 
        # 'sprouting' and 'init' of a branch both walk up the tree to find
2538
 
        # stacking policy to honour; create a bzr dir with an unshared
2539
 
        # repository (but not a branch - our code would be trying to escape
2540
 
        # then!) to stop them, and permit it to be read.
2541
 
        # control = bzrdir.BzrDir.create(self.test_base_dir)
2542
 
        # control.create_repository()
 
2308
        # now create test and home directories within this dir
 
2309
        self.test_base_dir = name
2543
2310
        self.test_home_dir = self.test_base_dir + '/home'
2544
2311
        os.mkdir(self.test_home_dir)
2545
2312
        self.test_dir = self.test_base_dir + '/work'
2551
2318
            f.write(self.id())
2552
2319
        finally:
2553
2320
            f.close()
 
2321
        self.addCleanup(self.deleteTestDir)
2554
2322
 
2555
2323
    def deleteTestDir(self):
2556
2324
        os.chdir(TestCaseWithMemoryTransport.TEST_ROOT)
2557
 
        _rmtree_temp_dir(self.test_base_dir, test_id=self.id())
 
2325
        _rmtree_temp_dir(self.test_base_dir)
2558
2326
 
2559
2327
    def build_tree(self, shape, line_endings='binary', transport=None):
2560
2328
        """Build a test tree according to a pattern.
2579
2347
                "a list or a tuple. Got %r instead" % (shape,))
2580
2348
        # It's OK to just create them using forward slashes on windows.
2581
2349
        if transport is None or transport.is_readonly():
2582
 
            transport = _mod_transport.get_transport(".")
 
2350
            transport = get_transport(".")
2583
2351
        for name in shape:
2584
2352
            self.assertIsInstance(name, basestring)
2585
2353
            if name[-1] == '/':
2595
2363
                content = "contents of %s%s" % (name.encode('utf-8'), end)
2596
2364
                transport.put_bytes_non_atomic(urlutils.escape(name), content)
2597
2365
 
2598
 
    build_tree_contents = staticmethod(treeshape.build_tree_contents)
 
2366
    def build_tree_contents(self, shape):
 
2367
        build_tree_contents(shape)
2599
2368
 
2600
2369
    def assertInWorkingTree(self, path, root_path='.', tree=None):
2601
2370
        """Assert whether path or paths are in the WorkingTree"""
2641
2410
        """
2642
2411
        if self.__vfs_server is None:
2643
2412
            self.__vfs_server = self.vfs_transport_factory()
2644
 
            self.start_server(self.__vfs_server)
 
2413
            self.__vfs_server.setUp()
 
2414
            self.addCleanup(self.__vfs_server.tearDown)
2645
2415
        return self.__vfs_server
2646
2416
 
2647
2417
    def make_branch_and_tree(self, relpath, format=None):
2654
2424
        repository will also be accessed locally. Otherwise a lightweight
2655
2425
        checkout is created and returned.
2656
2426
 
2657
 
        We do this because we can't physically create a tree in the local
2658
 
        path, with a branch reference to the transport_factory url, and
2659
 
        a branch + repository in the vfs_transport, unless the vfs_transport
2660
 
        namespace is distinct from the local disk - the two branch objects
2661
 
        would collide. While we could construct a tree with its branch object
2662
 
        pointing at the transport_factory transport in memory, reopening it
2663
 
        would behaving unexpectedly, and has in the past caused testing bugs
2664
 
        when we tried to do it that way.
2665
 
 
2666
2427
        :param format: The BzrDirFormat.
2667
2428
        :returns: the WorkingTree.
2668
2429
        """
2677
2438
            # We can only make working trees locally at the moment.  If the
2678
2439
            # transport can't support them, then we keep the non-disk-backed
2679
2440
            # branch and create a local checkout.
2680
 
            if self.vfs_transport_factory is test_server.LocalURLServer:
 
2441
            if self.vfs_transport_factory is LocalURLServer:
2681
2442
                # the branch is colocated on disk, we cannot create a checkout.
2682
2443
                # hopefully callers will expect this.
2683
2444
                local_controldir= bzrdir.BzrDir.open(self.get_vfs_only_url(relpath))
2720
2481
        super(TestCaseWithTransport, self).setUp()
2721
2482
        self.__vfs_server = None
2722
2483
 
2723
 
    def disable_missing_extensions_warning(self):
2724
 
        """Some tests expect a precise stderr content.
2725
 
 
2726
 
        There is no point in forcing them to duplicate the extension related
2727
 
        warning.
2728
 
        """
2729
 
        config.GlobalConfig().set_user_option('ignore_missing_extensions', True)
2730
 
 
2731
2484
 
2732
2485
class ChrootedTestCase(TestCaseWithTransport):
2733
2486
    """A support class that provides readonly urls outside the local namespace.
2742
2495
    """
2743
2496
 
2744
2497
    def setUp(self):
2745
 
        from bzrlib.tests import http_server
2746
2498
        super(ChrootedTestCase, self).setUp()
2747
 
        if not self.vfs_transport_factory == memory.MemoryServer:
2748
 
            self.transport_readonly_server = http_server.HttpServer
 
2499
        if not self.vfs_transport_factory == MemoryServer:
 
2500
            self.transport_readonly_server = HttpServer
2749
2501
 
2750
2502
 
2751
2503
def condition_id_re(pattern):
2754
2506
    :param pattern: A regular expression string.
2755
2507
    :return: A callable that returns True if the re matches.
2756
2508
    """
2757
 
    filter_re = re.compile(pattern, 0)
 
2509
    filter_re = osutils.re_compile_checked(pattern, 0,
 
2510
        'test filter')
2758
2511
    def condition(test):
2759
2512
        test_id = test.id()
2760
2513
        return filter_re.search(test_id)
2947
2700
              strict=False,
2948
2701
              runner_class=None,
2949
2702
              suite_decorators=None,
2950
 
              stream=None,
2951
 
              result_decorators=None,
2952
 
              ):
 
2703
              stream=None):
2953
2704
    """Run a test suite for bzr selftest.
2954
2705
 
2955
2706
    :param runner_class: The class of runner to use. Must support the
2970
2721
                            descriptions=0,
2971
2722
                            verbosity=verbosity,
2972
2723
                            bench_history=bench_history,
 
2724
                            list_only=list_only,
2973
2725
                            strict=strict,
2974
 
                            result_decorators=result_decorators,
2975
2726
                            )
2976
2727
    runner.stop_on_failure=stop_on_failure
2977
2728
    # built in decorator factories:
2985
2736
        decorators.append(filter_tests(pattern))
2986
2737
    if suite_decorators:
2987
2738
        decorators.extend(suite_decorators)
2988
 
    # tell the result object how many tests will be running: (except if
2989
 
    # --parallel=fork is being used. Robert said he will provide a better
2990
 
    # progress design later -- vila 20090817)
2991
 
    if fork_decorator not in decorators:
2992
 
        decorators.append(CountingDecorator)
2993
2739
    for decorator in decorators:
2994
2740
        suite = decorator(suite)
 
2741
    result = runner.run(suite)
2995
2742
    if list_only:
2996
 
        # Done after test suite decoration to allow randomisation etc
2997
 
        # to take effect, though that is of marginal benefit.
2998
 
        if verbosity >= 2:
2999
 
            stream.write("Listing tests only ...\n")
3000
 
        for t in iter_suite_tests(suite):
3001
 
            stream.write("%s\n" % (t.id()))
3002
2743
        return True
3003
 
    result = runner.run(suite)
 
2744
    result.done()
3004
2745
    if strict:
3005
2746
        return result.wasStrictlySuccessful()
3006
2747
    else:
3012
2753
 
3013
2754
 
3014
2755
def fork_decorator(suite):
3015
 
    concurrency = osutils.local_concurrency()
 
2756
    concurrency = local_concurrency()
3016
2757
    if concurrency == 1:
3017
2758
        return suite
3018
2759
    from testtools import ConcurrentTestSuite
3021
2762
 
3022
2763
 
3023
2764
def subprocess_decorator(suite):
3024
 
    concurrency = osutils.local_concurrency()
 
2765
    concurrency = local_concurrency()
3025
2766
    if concurrency == 1:
3026
2767
        return suite
3027
2768
    from testtools import ConcurrentTestSuite
3072
2813
    return suite
3073
2814
 
3074
2815
 
3075
 
class TestDecorator(TestUtil.TestSuite):
 
2816
class TestDecorator(TestSuite):
3076
2817
    """A decorator for TestCase/TestSuite objects.
3077
2818
    
3078
2819
    Usually, subclasses should override __iter__(used when flattening test
3081
2822
    """
3082
2823
 
3083
2824
    def __init__(self, suite):
3084
 
        TestUtil.TestSuite.__init__(self)
 
2825
        TestSuite.__init__(self)
3085
2826
        self.addTest(suite)
3086
2827
 
3087
2828
    def countTestCases(self):
3104
2845
        return result
3105
2846
 
3106
2847
 
3107
 
class CountingDecorator(TestDecorator):
3108
 
    """A decorator which calls result.progress(self.countTestCases)."""
3109
 
 
3110
 
    def run(self, result):
3111
 
        progress_method = getattr(result, 'progress', None)
3112
 
        if callable(progress_method):
3113
 
            progress_method(self.countTestCases(), SUBUNIT_SEEK_SET)
3114
 
        return super(CountingDecorator, self).run(result)
3115
 
 
3116
 
 
3117
2848
class ExcludeDecorator(TestDecorator):
3118
2849
    """A decorator which excludes test matching an exclude pattern."""
3119
2850
 
3163
2894
        if self.randomised:
3164
2895
            return iter(self._tests)
3165
2896
        self.randomised = True
3166
 
        self.stream.write("Randomizing test order using seed %s\n\n" %
 
2897
        self.stream.writeln("Randomizing test order using seed %s\n" %
3167
2898
            (self.actual_seed()))
3168
2899
        # Initialise the random number generator.
3169
2900
        random.seed(self.actual_seed())
3206
2937
 
3207
2938
def partition_tests(suite, count):
3208
2939
    """Partition suite into count lists of tests."""
3209
 
    # This just assigns tests in a round-robin fashion.  On one hand this
3210
 
    # splits up blocks of related tests that might run faster if they shared
3211
 
    # resources, but on the other it avoids assigning blocks of slow tests to
3212
 
    # just one partition.  So the slowest partition shouldn't be much slower
3213
 
    # than the fastest.
3214
 
    partitions = [list() for i in range(count)]
3215
 
    tests = iter_suite_tests(suite)
3216
 
    for partition, test in itertools.izip(itertools.cycle(partitions), tests):
3217
 
        partition.append(test)
3218
 
    return partitions
3219
 
 
3220
 
 
3221
 
def workaround_zealous_crypto_random():
3222
 
    """Crypto.Random want to help us being secure, but we don't care here.
3223
 
 
3224
 
    This workaround some test failure related to the sftp server. Once paramiko
3225
 
    stop using the controversial API in Crypto.Random, we may get rid of it.
3226
 
    """
3227
 
    try:
3228
 
        from Crypto.Random import atfork
3229
 
        atfork()
3230
 
    except ImportError:
3231
 
        pass
 
2940
    result = []
 
2941
    tests = list(iter_suite_tests(suite))
 
2942
    tests_per_process = int(math.ceil(float(len(tests)) / count))
 
2943
    for block in range(count):
 
2944
        low_test = block * tests_per_process
 
2945
        high_test = low_test + tests_per_process
 
2946
        process_tests = tests[low_test:high_test]
 
2947
        result.append(process_tests)
 
2948
    return result
3232
2949
 
3233
2950
 
3234
2951
def fork_for_tests(suite):
3237
2954
    :return: An iterable of TestCase-like objects which can each have
3238
2955
        run(result) called on them to feed tests to result.
3239
2956
    """
3240
 
    concurrency = osutils.local_concurrency()
 
2957
    concurrency = local_concurrency()
3241
2958
    result = []
3242
2959
    from subunit import TestProtocolClient, ProtocolTestCase
3243
 
    from subunit.test_results import AutoTimingTestResultDecorator
3244
2960
    class TestInOtherProcess(ProtocolTestCase):
3245
2961
        # Should be in subunit, I think. RBC.
3246
2962
        def __init__(self, stream, pid):
3251
2967
            try:
3252
2968
                ProtocolTestCase.run(self, result)
3253
2969
            finally:
3254
 
                os.waitpid(self.pid, 0)
 
2970
                os.waitpid(self.pid, os.WNOHANG)
3255
2971
 
3256
2972
    test_blocks = partition_tests(suite, concurrency)
3257
2973
    for process_tests in test_blocks:
3258
 
        process_suite = TestUtil.TestSuite()
 
2974
        process_suite = TestSuite()
3259
2975
        process_suite.addTests(process_tests)
3260
2976
        c2pread, c2pwrite = os.pipe()
3261
2977
        pid = os.fork()
3262
2978
        if pid == 0:
3263
 
            workaround_zealous_crypto_random()
3264
2979
            try:
3265
2980
                os.close(c2pread)
3266
2981
                # Leave stderr and stdout open so we can see test noise
3270
2985
                sys.stdin.close()
3271
2986
                sys.stdin = None
3272
2987
                stream = os.fdopen(c2pwrite, 'wb', 1)
3273
 
                subunit_result = AutoTimingTestResultDecorator(
3274
 
                    TestProtocolClient(stream))
 
2988
                subunit_result = TestProtocolClient(stream)
3275
2989
                process_suite.run(subunit_result)
3276
2990
            finally:
3277
2991
                os._exit(0)
3289
3003
    :return: An iterable of TestCase-like objects which can each have
3290
3004
        run(result) called on them to feed tests to result.
3291
3005
    """
3292
 
    concurrency = osutils.local_concurrency()
 
3006
    concurrency = local_concurrency()
3293
3007
    result = []
3294
 
    from subunit import ProtocolTestCase
 
3008
    from subunit import TestProtocolClient, ProtocolTestCase
3295
3009
    class TestInSubprocess(ProtocolTestCase):
3296
3010
        def __init__(self, process, name):
3297
3011
            ProtocolTestCase.__init__(self, process.stdout)
3313
3027
        if not os.path.isfile(bzr_path):
3314
3028
            # We are probably installed. Assume sys.argv is the right file
3315
3029
            bzr_path = sys.argv[0]
3316
 
        bzr_path = [bzr_path]
3317
 
        if sys.platform == "win32":
3318
 
            # if we're on windows, we can't execute the bzr script directly
3319
 
            bzr_path = [sys.executable] + bzr_path
3320
3030
        fd, test_list_file_name = tempfile.mkstemp()
3321
3031
        test_list_file = os.fdopen(fd, 'wb', 1)
3322
3032
        for test in process_tests:
3323
3033
            test_list_file.write(test.id() + '\n')
3324
3034
        test_list_file.close()
3325
3035
        try:
3326
 
            argv = bzr_path + ['selftest', '--load-list', test_list_file_name,
 
3036
            argv = [bzr_path, 'selftest', '--load-list', test_list_file_name,
3327
3037
                '--subunit']
3328
3038
            if '--no-plugins' in sys.argv:
3329
3039
                argv.append('--no-plugins')
3330
 
            # stderr=subprocess.STDOUT would be ideal, but until we prevent
3331
 
            # noise on stderr it can interrupt the subunit protocol.
3332
 
            process = subprocess.Popen(argv, stdin=subprocess.PIPE,
3333
 
                                      stdout=subprocess.PIPE,
3334
 
                                      stderr=subprocess.PIPE,
3335
 
                                      bufsize=1)
 
3040
            # stderr=STDOUT would be ideal, but until we prevent noise on
 
3041
            # stderr it can interrupt the subunit protocol.
 
3042
            process = Popen(argv, stdin=PIPE, stdout=PIPE, stderr=PIPE,
 
3043
                bufsize=1)
3336
3044
            test = TestInSubprocess(process, test_list_file_name)
3337
3045
            result.append(test)
3338
3046
        except:
3341
3049
    return result
3342
3050
 
3343
3051
 
3344
 
class ForwardingResult(unittest.TestResult):
 
3052
def cpucount(content):
 
3053
    lines = content.splitlines()
 
3054
    prefix = 'processor'
 
3055
    for line in lines:
 
3056
        if line.startswith(prefix):
 
3057
            concurrency = int(line[line.find(':')+1:]) + 1
 
3058
    return concurrency
 
3059
 
 
3060
 
 
3061
def local_concurrency():
 
3062
    try:
 
3063
        content = file('/proc/cpuinfo', 'rb').read()
 
3064
        concurrency = cpucount(content)
 
3065
    except Exception, e:
 
3066
        concurrency = 1
 
3067
    return concurrency
 
3068
 
 
3069
 
 
3070
class BZRTransformingResult(unittest.TestResult):
3345
3071
 
3346
3072
    def __init__(self, target):
3347
3073
        unittest.TestResult.__init__(self)
3353
3079
    def stopTest(self, test):
3354
3080
        self.result.stopTest(test)
3355
3081
 
3356
 
    def startTestRun(self):
3357
 
        self.result.startTestRun()
 
3082
    def addError(self, test, err):
 
3083
        feature = self._error_looks_like('UnavailableFeature: ', err)
 
3084
        if feature is not None:
 
3085
            self.result.addNotSupported(test, feature)
 
3086
        else:
 
3087
            self.result.addError(test, err)
3358
3088
 
3359
 
    def stopTestRun(self):
3360
 
        self.result.stopTestRun()
 
3089
    def addFailure(self, test, err):
 
3090
        known = self._error_looks_like('KnownFailure: ', err)
 
3091
        if known is not None:
 
3092
            self.result._addKnownFailure(test, [KnownFailure,
 
3093
                                                KnownFailure(known), None])
 
3094
        else:
 
3095
            self.result.addFailure(test, err)
3361
3096
 
3362
3097
    def addSkip(self, test, reason):
3363
3098
        self.result.addSkip(test, reason)
3365
3100
    def addSuccess(self, test):
3366
3101
        self.result.addSuccess(test)
3367
3102
 
3368
 
    def addError(self, test, err):
3369
 
        self.result.addError(test, err)
3370
 
 
3371
 
    def addFailure(self, test, err):
3372
 
        self.result.addFailure(test, err)
3373
 
ForwardingResult = testtools.ExtendedToOriginalDecorator
3374
 
 
3375
 
 
3376
 
class ProfileResult(ForwardingResult):
3377
 
    """Generate profiling data for all activity between start and success.
3378
 
    
3379
 
    The profile data is appended to the test's _benchcalls attribute and can
3380
 
    be accessed by the forwarded-to TestResult.
3381
 
 
3382
 
    While it might be cleaner do accumulate this in stopTest, addSuccess is
3383
 
    where our existing output support for lsprof is, and this class aims to
3384
 
    fit in with that: while it could be moved it's not necessary to accomplish
3385
 
    test profiling, nor would it be dramatically cleaner.
3386
 
    """
3387
 
 
3388
 
    def startTest(self, test):
3389
 
        self.profiler = bzrlib.lsprof.BzrProfiler()
3390
 
        # Prevent deadlocks in tests that use lsprof: those tests will
3391
 
        # unavoidably fail.
3392
 
        bzrlib.lsprof.BzrProfiler.profiler_block = 0
3393
 
        self.profiler.start()
3394
 
        ForwardingResult.startTest(self, test)
3395
 
 
3396
 
    def addSuccess(self, test):
3397
 
        stats = self.profiler.stop()
3398
 
        try:
3399
 
            calls = test._benchcalls
3400
 
        except AttributeError:
3401
 
            test._benchcalls = []
3402
 
            calls = test._benchcalls
3403
 
        calls.append(((test.id(), "", ""), stats))
3404
 
        ForwardingResult.addSuccess(self, test)
3405
 
 
3406
 
    def stopTest(self, test):
3407
 
        ForwardingResult.stopTest(self, test)
3408
 
        self.profiler = None
 
3103
    def _error_looks_like(self, prefix, err):
 
3104
        """Deserialize exception and returns the stringify value."""
 
3105
        import subunit
 
3106
        value = None
 
3107
        typ, exc, _ = err
 
3108
        if isinstance(exc, subunit.RemoteException):
 
3109
            # stringify the exception gives access to the remote traceback
 
3110
            # We search the last line for 'prefix'
 
3111
            lines = str(exc).split('\n')
 
3112
            while lines and not lines[-1]:
 
3113
                lines.pop(-1)
 
3114
            if lines:
 
3115
                if lines[-1].startswith(prefix):
 
3116
                    value = lines[-1][len(prefix):]
 
3117
        return value
3409
3118
 
3410
3119
 
3411
3120
# Controlled by "bzr selftest -E=..." option
3412
 
# Currently supported:
3413
 
#   -Eallow_debug           Will no longer clear debug.debug_flags() so it
3414
 
#                           preserves any flags supplied at the command line.
3415
 
#   -Edisable_lock_checks   Turns errors in mismatched locks into simple prints
3416
 
#                           rather than failing tests. And no longer raise
3417
 
#                           LockContention when fctnl locks are not being used
3418
 
#                           with proper exclusion rules.
3419
 
#   -Ethreads               Will display thread ident at creation/join time to
3420
 
#                           help track thread leaks
3421
3121
selftest_debug_flags = set()
3422
3122
 
3423
3123
 
3436
3136
             starting_with=None,
3437
3137
             runner_class=None,
3438
3138
             suite_decorators=None,
3439
 
             stream=None,
3440
 
             lsprof_tests=False,
3441
3139
             ):
3442
3140
    """Run the whole test suite under the enhanced runner"""
3443
3141
    # XXX: Very ugly way to do this...
3460
3158
            keep_only = None
3461
3159
        else:
3462
3160
            keep_only = load_test_id_list(load_list)
3463
 
        if starting_with:
3464
 
            starting_with = [test_prefix_alias_registry.resolve_alias(start)
3465
 
                             for start in starting_with]
3466
3161
        if test_suite_factory is None:
3467
 
            # Reduce loading time by loading modules based on the starting_with
3468
 
            # patterns.
3469
3162
            suite = test_suite(keep_only, starting_with)
3470
3163
        else:
3471
3164
            suite = test_suite_factory()
3472
 
        if starting_with:
3473
 
            # But always filter as requested.
3474
 
            suite = filter_suite_by_id_startswith(suite, starting_with)
3475
 
        result_decorators = []
3476
 
        if lsprof_tests:
3477
 
            result_decorators.append(ProfileResult)
3478
3165
        return run_suite(suite, 'testbzr', verbose=verbose, pattern=pattern,
3479
3166
                     stop_on_failure=stop_on_failure,
3480
3167
                     transport=transport,
3487
3174
                     strict=strict,
3488
3175
                     runner_class=runner_class,
3489
3176
                     suite_decorators=suite_decorators,
3490
 
                     stream=stream,
3491
 
                     result_decorators=result_decorators,
3492
3177
                     )
3493
3178
    finally:
3494
3179
        default_transport = old_transport
3642
3327
# appear prefixed ('bzrlib.' is "replaced" by 'bzrlib.').
3643
3328
test_prefix_alias_registry.register('bzrlib', 'bzrlib')
3644
3329
 
3645
 
# Obvious highest levels prefixes, feel free to add your own via a plugin
 
3330
# Obvious higest levels prefixes, feel free to add your own via a plugin
3646
3331
test_prefix_alias_registry.register('bd', 'bzrlib.doc')
3647
3332
test_prefix_alias_registry.register('bu', 'bzrlib.utils')
3648
3333
test_prefix_alias_registry.register('bt', 'bzrlib.tests')
3650
3335
test_prefix_alias_registry.register('bp', 'bzrlib.plugins')
3651
3336
 
3652
3337
 
3653
 
def _test_suite_testmod_names():
3654
 
    """Return the standard list of test module names to test."""
3655
 
    return [
3656
 
        'bzrlib.doc',
3657
 
        'bzrlib.tests.blackbox',
3658
 
        'bzrlib.tests.commands',
3659
 
        'bzrlib.tests.doc_generate',
3660
 
        'bzrlib.tests.per_branch',
3661
 
        'bzrlib.tests.per_controldir',
3662
 
        'bzrlib.tests.per_controldir_colo',
3663
 
        'bzrlib.tests.per_foreign_vcs',
3664
 
        'bzrlib.tests.per_interrepository',
3665
 
        'bzrlib.tests.per_intertree',
3666
 
        'bzrlib.tests.per_inventory',
3667
 
        'bzrlib.tests.per_interbranch',
3668
 
        'bzrlib.tests.per_lock',
3669
 
        'bzrlib.tests.per_merger',
3670
 
        'bzrlib.tests.per_transport',
3671
 
        'bzrlib.tests.per_tree',
3672
 
        'bzrlib.tests.per_pack_repository',
3673
 
        'bzrlib.tests.per_repository',
3674
 
        'bzrlib.tests.per_repository_chk',
3675
 
        'bzrlib.tests.per_repository_reference',
3676
 
        'bzrlib.tests.per_uifactory',
3677
 
        'bzrlib.tests.per_versionedfile',
3678
 
        'bzrlib.tests.per_workingtree',
3679
 
        'bzrlib.tests.test__annotator',
3680
 
        'bzrlib.tests.test__bencode',
3681
 
        'bzrlib.tests.test__btree_serializer',
3682
 
        'bzrlib.tests.test__chk_map',
3683
 
        'bzrlib.tests.test__dirstate_helpers',
3684
 
        'bzrlib.tests.test__groupcompress',
3685
 
        'bzrlib.tests.test__known_graph',
3686
 
        'bzrlib.tests.test__rio',
3687
 
        'bzrlib.tests.test__simple_set',
3688
 
        'bzrlib.tests.test__static_tuple',
3689
 
        'bzrlib.tests.test__walkdirs_win32',
3690
 
        'bzrlib.tests.test_ancestry',
3691
 
        'bzrlib.tests.test_annotate',
3692
 
        'bzrlib.tests.test_api',
3693
 
        'bzrlib.tests.test_atomicfile',
3694
 
        'bzrlib.tests.test_bad_files',
3695
 
        'bzrlib.tests.test_bisect_multi',
3696
 
        'bzrlib.tests.test_branch',
3697
 
        'bzrlib.tests.test_branchbuilder',
3698
 
        'bzrlib.tests.test_btree_index',
3699
 
        'bzrlib.tests.test_bugtracker',
3700
 
        'bzrlib.tests.test_bundle',
3701
 
        'bzrlib.tests.test_bzrdir',
3702
 
        'bzrlib.tests.test__chunks_to_lines',
3703
 
        'bzrlib.tests.test_cache_utf8',
3704
 
        'bzrlib.tests.test_chk_map',
3705
 
        'bzrlib.tests.test_chk_serializer',
3706
 
        'bzrlib.tests.test_chunk_writer',
3707
 
        'bzrlib.tests.test_clean_tree',
3708
 
        'bzrlib.tests.test_cleanup',
3709
 
        'bzrlib.tests.test_cmdline',
3710
 
        'bzrlib.tests.test_commands',
3711
 
        'bzrlib.tests.test_commit',
3712
 
        'bzrlib.tests.test_commit_merge',
3713
 
        'bzrlib.tests.test_config',
3714
 
        'bzrlib.tests.test_conflicts',
3715
 
        'bzrlib.tests.test_counted_lock',
3716
 
        'bzrlib.tests.test_crash',
3717
 
        'bzrlib.tests.test_decorators',
3718
 
        'bzrlib.tests.test_delta',
3719
 
        'bzrlib.tests.test_debug',
3720
 
        'bzrlib.tests.test_deprecated_graph',
3721
 
        'bzrlib.tests.test_diff',
3722
 
        'bzrlib.tests.test_directory_service',
3723
 
        'bzrlib.tests.test_dirstate',
3724
 
        'bzrlib.tests.test_email_message',
3725
 
        'bzrlib.tests.test_eol_filters',
3726
 
        'bzrlib.tests.test_errors',
3727
 
        'bzrlib.tests.test_export',
3728
 
        'bzrlib.tests.test_extract',
3729
 
        'bzrlib.tests.test_fetch',
3730
 
        'bzrlib.tests.test_fixtures',
3731
 
        'bzrlib.tests.test_fifo_cache',
3732
 
        'bzrlib.tests.test_filters',
3733
 
        'bzrlib.tests.test_ftp_transport',
3734
 
        'bzrlib.tests.test_foreign',
3735
 
        'bzrlib.tests.test_generate_docs',
3736
 
        'bzrlib.tests.test_generate_ids',
3737
 
        'bzrlib.tests.test_globbing',
3738
 
        'bzrlib.tests.test_gpg',
3739
 
        'bzrlib.tests.test_graph',
3740
 
        'bzrlib.tests.test_groupcompress',
3741
 
        'bzrlib.tests.test_hashcache',
3742
 
        'bzrlib.tests.test_help',
3743
 
        'bzrlib.tests.test_hooks',
3744
 
        'bzrlib.tests.test_http',
3745
 
        'bzrlib.tests.test_http_response',
3746
 
        'bzrlib.tests.test_https_ca_bundle',
3747
 
        'bzrlib.tests.test_identitymap',
3748
 
        'bzrlib.tests.test_ignores',
3749
 
        'bzrlib.tests.test_index',
3750
 
        'bzrlib.tests.test_import_tariff',
3751
 
        'bzrlib.tests.test_info',
3752
 
        'bzrlib.tests.test_inv',
3753
 
        'bzrlib.tests.test_inventory_delta',
3754
 
        'bzrlib.tests.test_knit',
3755
 
        'bzrlib.tests.test_lazy_import',
3756
 
        'bzrlib.tests.test_lazy_regex',
3757
 
        'bzrlib.tests.test_library_state',
3758
 
        'bzrlib.tests.test_lock',
3759
 
        'bzrlib.tests.test_lockable_files',
3760
 
        'bzrlib.tests.test_lockdir',
3761
 
        'bzrlib.tests.test_log',
3762
 
        'bzrlib.tests.test_lru_cache',
3763
 
        'bzrlib.tests.test_lsprof',
3764
 
        'bzrlib.tests.test_mail_client',
3765
 
        'bzrlib.tests.test_matchers',
3766
 
        'bzrlib.tests.test_memorytree',
3767
 
        'bzrlib.tests.test_merge',
3768
 
        'bzrlib.tests.test_merge3',
3769
 
        'bzrlib.tests.test_merge_core',
3770
 
        'bzrlib.tests.test_merge_directive',
3771
 
        'bzrlib.tests.test_missing',
3772
 
        'bzrlib.tests.test_msgeditor',
3773
 
        'bzrlib.tests.test_multiparent',
3774
 
        'bzrlib.tests.test_mutabletree',
3775
 
        'bzrlib.tests.test_nonascii',
3776
 
        'bzrlib.tests.test_options',
3777
 
        'bzrlib.tests.test_osutils',
3778
 
        'bzrlib.tests.test_osutils_encodings',
3779
 
        'bzrlib.tests.test_pack',
3780
 
        'bzrlib.tests.test_patch',
3781
 
        'bzrlib.tests.test_patches',
3782
 
        'bzrlib.tests.test_permissions',
3783
 
        'bzrlib.tests.test_plugins',
3784
 
        'bzrlib.tests.test_progress',
3785
 
        'bzrlib.tests.test_read_bundle',
3786
 
        'bzrlib.tests.test_reconcile',
3787
 
        'bzrlib.tests.test_reconfigure',
3788
 
        'bzrlib.tests.test_registry',
3789
 
        'bzrlib.tests.test_remote',
3790
 
        'bzrlib.tests.test_rename_map',
3791
 
        'bzrlib.tests.test_repository',
3792
 
        'bzrlib.tests.test_revert',
3793
 
        'bzrlib.tests.test_revision',
3794
 
        'bzrlib.tests.test_revisionspec',
3795
 
        'bzrlib.tests.test_revisiontree',
3796
 
        'bzrlib.tests.test_rio',
3797
 
        'bzrlib.tests.test_rules',
3798
 
        'bzrlib.tests.test_sampler',
3799
 
        'bzrlib.tests.test_script',
3800
 
        'bzrlib.tests.test_selftest',
3801
 
        'bzrlib.tests.test_serializer',
3802
 
        'bzrlib.tests.test_setup',
3803
 
        'bzrlib.tests.test_sftp_transport',
3804
 
        'bzrlib.tests.test_shelf',
3805
 
        'bzrlib.tests.test_shelf_ui',
3806
 
        'bzrlib.tests.test_smart',
3807
 
        'bzrlib.tests.test_smart_add',
3808
 
        'bzrlib.tests.test_smart_request',
3809
 
        'bzrlib.tests.test_smart_transport',
3810
 
        'bzrlib.tests.test_smtp_connection',
3811
 
        'bzrlib.tests.test_source',
3812
 
        'bzrlib.tests.test_ssh_transport',
3813
 
        'bzrlib.tests.test_status',
3814
 
        'bzrlib.tests.test_store',
3815
 
        'bzrlib.tests.test_strace',
3816
 
        'bzrlib.tests.test_subsume',
3817
 
        'bzrlib.tests.test_switch',
3818
 
        'bzrlib.tests.test_symbol_versioning',
3819
 
        'bzrlib.tests.test_tag',
3820
 
        'bzrlib.tests.test_test_server',
3821
 
        'bzrlib.tests.test_testament',
3822
 
        'bzrlib.tests.test_textfile',
3823
 
        'bzrlib.tests.test_textmerge',
3824
 
        'bzrlib.tests.test_timestamp',
3825
 
        'bzrlib.tests.test_trace',
3826
 
        'bzrlib.tests.test_transactions',
3827
 
        'bzrlib.tests.test_transform',
3828
 
        'bzrlib.tests.test_transport',
3829
 
        'bzrlib.tests.test_transport_log',
3830
 
        'bzrlib.tests.test_tree',
3831
 
        'bzrlib.tests.test_treebuilder',
3832
 
        'bzrlib.tests.test_treeshape',
3833
 
        'bzrlib.tests.test_tsort',
3834
 
        'bzrlib.tests.test_tuned_gzip',
3835
 
        'bzrlib.tests.test_ui',
3836
 
        'bzrlib.tests.test_uncommit',
3837
 
        'bzrlib.tests.test_upgrade',
3838
 
        'bzrlib.tests.test_upgrade_stacked',
3839
 
        'bzrlib.tests.test_urlutils',
3840
 
        'bzrlib.tests.test_version',
3841
 
        'bzrlib.tests.test_version_info',
3842
 
        'bzrlib.tests.test_versionedfile',
3843
 
        'bzrlib.tests.test_weave',
3844
 
        'bzrlib.tests.test_whitebox',
3845
 
        'bzrlib.tests.test_win32utils',
3846
 
        'bzrlib.tests.test_workingtree',
3847
 
        'bzrlib.tests.test_workingtree_4',
3848
 
        'bzrlib.tests.test_wsgi',
3849
 
        'bzrlib.tests.test_xml',
3850
 
        ]
3851
 
 
3852
 
 
3853
 
def _test_suite_modules_to_doctest():
3854
 
    """Return the list of modules to doctest."""
3855
 
    if __doc__ is None:
3856
 
        # GZ 2009-03-31: No docstrings with -OO so there's nothing to doctest
3857
 
        return []
3858
 
    return [
3859
 
        'bzrlib',
3860
 
        'bzrlib.branchbuilder',
3861
 
        'bzrlib.decorators',
3862
 
        'bzrlib.export',
3863
 
        'bzrlib.inventory',
3864
 
        'bzrlib.iterablefile',
3865
 
        'bzrlib.lockdir',
3866
 
        'bzrlib.merge3',
3867
 
        'bzrlib.option',
3868
 
        'bzrlib.symbol_versioning',
3869
 
        'bzrlib.tests',
3870
 
        'bzrlib.tests.fixtures',
3871
 
        'bzrlib.timestamp',
3872
 
        'bzrlib.version_info_formats.format_custom',
3873
 
        ]
3874
 
 
3875
 
 
3876
3338
def test_suite(keep_only=None, starting_with=None):
3877
3339
    """Build and return TestSuite for the whole of bzrlib.
3878
3340
 
3884
3346
    This function can be replaced if you need to change the default test
3885
3347
    suite on a global basis, but it is not encouraged.
3886
3348
    """
 
3349
    testmod_names = [
 
3350
                   'bzrlib.doc',
 
3351
                   'bzrlib.tests.blackbox',
 
3352
                   'bzrlib.tests.branch_implementations',
 
3353
                   'bzrlib.tests.bzrdir_implementations',
 
3354
                   'bzrlib.tests.commands',
 
3355
                   'bzrlib.tests.interrepository_implementations',
 
3356
                   'bzrlib.tests.intertree_implementations',
 
3357
                   'bzrlib.tests.inventory_implementations',
 
3358
                   'bzrlib.tests.per_interbranch',
 
3359
                   'bzrlib.tests.per_lock',
 
3360
                   'bzrlib.tests.per_repository',
 
3361
                   'bzrlib.tests.per_repository_chk',
 
3362
                   'bzrlib.tests.per_repository_reference',
 
3363
                   'bzrlib.tests.test__chk_map',
 
3364
                   'bzrlib.tests.test__dirstate_helpers',
 
3365
                   'bzrlib.tests.test__groupcompress',
 
3366
                   'bzrlib.tests.test__rio',
 
3367
                   'bzrlib.tests.test__walkdirs_win32',
 
3368
                   'bzrlib.tests.test_ancestry',
 
3369
                   'bzrlib.tests.test_annotate',
 
3370
                   'bzrlib.tests.test_api',
 
3371
                   'bzrlib.tests.test_atomicfile',
 
3372
                   'bzrlib.tests.test_bad_files',
 
3373
                   'bzrlib.tests.test_bisect_multi',
 
3374
                   'bzrlib.tests.test_branch',
 
3375
                   'bzrlib.tests.test_branchbuilder',
 
3376
                   'bzrlib.tests.test_btree_index',
 
3377
                   'bzrlib.tests.test_bugtracker',
 
3378
                   'bzrlib.tests.test_bundle',
 
3379
                   'bzrlib.tests.test_bzrdir',
 
3380
                   'bzrlib.tests.test__chunks_to_lines',
 
3381
                   'bzrlib.tests.test_cache_utf8',
 
3382
                   'bzrlib.tests.test_chk_map',
 
3383
                   'bzrlib.tests.test_chunk_writer',
 
3384
                   'bzrlib.tests.test_clean_tree',
 
3385
                   'bzrlib.tests.test_commands',
 
3386
                   'bzrlib.tests.test_commit',
 
3387
                   'bzrlib.tests.test_commit_merge',
 
3388
                   'bzrlib.tests.test_config',
 
3389
                   'bzrlib.tests.test_conflicts',
 
3390
                   'bzrlib.tests.test_counted_lock',
 
3391
                   'bzrlib.tests.test_decorators',
 
3392
                   'bzrlib.tests.test_delta',
 
3393
                   'bzrlib.tests.test_debug',
 
3394
                   'bzrlib.tests.test_deprecated_graph',
 
3395
                   'bzrlib.tests.test_diff',
 
3396
                   'bzrlib.tests.test_directory_service',
 
3397
                   'bzrlib.tests.test_dirstate',
 
3398
                   'bzrlib.tests.test_email_message',
 
3399
                   'bzrlib.tests.test_eol_filters',
 
3400
                   'bzrlib.tests.test_errors',
 
3401
                   'bzrlib.tests.test_export',
 
3402
                   'bzrlib.tests.test_extract',
 
3403
                   'bzrlib.tests.test_fetch',
 
3404
                   'bzrlib.tests.test_fifo_cache',
 
3405
                   'bzrlib.tests.test_filters',
 
3406
                   'bzrlib.tests.test_ftp_transport',
 
3407
                   'bzrlib.tests.test_foreign',
 
3408
                   'bzrlib.tests.test_generate_docs',
 
3409
                   'bzrlib.tests.test_generate_ids',
 
3410
                   'bzrlib.tests.test_globbing',
 
3411
                   'bzrlib.tests.test_gpg',
 
3412
                   'bzrlib.tests.test_graph',
 
3413
                   'bzrlib.tests.test_groupcompress',
 
3414
                   'bzrlib.tests.test_hashcache',
 
3415
                   'bzrlib.tests.test_help',
 
3416
                   'bzrlib.tests.test_hooks',
 
3417
                   'bzrlib.tests.test_http',
 
3418
                   'bzrlib.tests.test_http_response',
 
3419
                   'bzrlib.tests.test_https_ca_bundle',
 
3420
                   'bzrlib.tests.test_identitymap',
 
3421
                   'bzrlib.tests.test_ignores',
 
3422
                   'bzrlib.tests.test_index',
 
3423
                   'bzrlib.tests.test_info',
 
3424
                   'bzrlib.tests.test_inv',
 
3425
                   'bzrlib.tests.test_inventory_delta',
 
3426
                   'bzrlib.tests.test_knit',
 
3427
                   'bzrlib.tests.test_lazy_import',
 
3428
                   'bzrlib.tests.test_lazy_regex',
 
3429
                   'bzrlib.tests.test_lockable_files',
 
3430
                   'bzrlib.tests.test_lockdir',
 
3431
                   'bzrlib.tests.test_log',
 
3432
                   'bzrlib.tests.test_lru_cache',
 
3433
                   'bzrlib.tests.test_lsprof',
 
3434
                   'bzrlib.tests.test_mail_client',
 
3435
                   'bzrlib.tests.test_memorytree',
 
3436
                   'bzrlib.tests.test_merge',
 
3437
                   'bzrlib.tests.test_merge3',
 
3438
                   'bzrlib.tests.test_merge_core',
 
3439
                   'bzrlib.tests.test_merge_directive',
 
3440
                   'bzrlib.tests.test_missing',
 
3441
                   'bzrlib.tests.test_msgeditor',
 
3442
                   'bzrlib.tests.test_multiparent',
 
3443
                   'bzrlib.tests.test_mutabletree',
 
3444
                   'bzrlib.tests.test_nonascii',
 
3445
                   'bzrlib.tests.test_options',
 
3446
                   'bzrlib.tests.test_osutils',
 
3447
                   'bzrlib.tests.test_osutils_encodings',
 
3448
                   'bzrlib.tests.test_pack',
 
3449
                   'bzrlib.tests.test_pack_repository',
 
3450
                   'bzrlib.tests.test_patch',
 
3451
                   'bzrlib.tests.test_patches',
 
3452
                   'bzrlib.tests.test_permissions',
 
3453
                   'bzrlib.tests.test_plugins',
 
3454
                   'bzrlib.tests.test_progress',
 
3455
                   'bzrlib.tests.test_read_bundle',
 
3456
                   'bzrlib.tests.test_reconcile',
 
3457
                   'bzrlib.tests.test_reconfigure',
 
3458
                   'bzrlib.tests.test_registry',
 
3459
                   'bzrlib.tests.test_remote',
 
3460
                   'bzrlib.tests.test_rename_map',
 
3461
                   'bzrlib.tests.test_repository',
 
3462
                   'bzrlib.tests.test_revert',
 
3463
                   'bzrlib.tests.test_revision',
 
3464
                   'bzrlib.tests.test_revisionspec',
 
3465
                   'bzrlib.tests.test_revisiontree',
 
3466
                   'bzrlib.tests.test_rio',
 
3467
                   'bzrlib.tests.test_rules',
 
3468
                   'bzrlib.tests.test_sampler',
 
3469
                   'bzrlib.tests.test_selftest',
 
3470
                   'bzrlib.tests.test_serializer',
 
3471
                   'bzrlib.tests.test_setup',
 
3472
                   'bzrlib.tests.test_sftp_transport',
 
3473
                   'bzrlib.tests.test_shelf',
 
3474
                   'bzrlib.tests.test_shelf_ui',
 
3475
                   'bzrlib.tests.test_smart',
 
3476
                   'bzrlib.tests.test_smart_add',
 
3477
                   'bzrlib.tests.test_smart_request',
 
3478
                   'bzrlib.tests.test_smart_transport',
 
3479
                   'bzrlib.tests.test_smtp_connection',
 
3480
                   'bzrlib.tests.test_source',
 
3481
                   'bzrlib.tests.test_ssh_transport',
 
3482
                   'bzrlib.tests.test_status',
 
3483
                   'bzrlib.tests.test_store',
 
3484
                   'bzrlib.tests.test_strace',
 
3485
                   'bzrlib.tests.test_subsume',
 
3486
                   'bzrlib.tests.test_switch',
 
3487
                   'bzrlib.tests.test_symbol_versioning',
 
3488
                   'bzrlib.tests.test_tag',
 
3489
                   'bzrlib.tests.test_testament',
 
3490
                   'bzrlib.tests.test_textfile',
 
3491
                   'bzrlib.tests.test_textmerge',
 
3492
                   'bzrlib.tests.test_timestamp',
 
3493
                   'bzrlib.tests.test_trace',
 
3494
                   'bzrlib.tests.test_transactions',
 
3495
                   'bzrlib.tests.test_transform',
 
3496
                   'bzrlib.tests.test_transport',
 
3497
                   'bzrlib.tests.test_transport_implementations',
 
3498
                   'bzrlib.tests.test_transport_log',
 
3499
                   'bzrlib.tests.test_tree',
 
3500
                   'bzrlib.tests.test_treebuilder',
 
3501
                   'bzrlib.tests.test_tsort',
 
3502
                   'bzrlib.tests.test_tuned_gzip',
 
3503
                   'bzrlib.tests.test_ui',
 
3504
                   'bzrlib.tests.test_uncommit',
 
3505
                   'bzrlib.tests.test_upgrade',
 
3506
                   'bzrlib.tests.test_upgrade_stacked',
 
3507
                   'bzrlib.tests.test_urlutils',
 
3508
                   'bzrlib.tests.test_version',
 
3509
                   'bzrlib.tests.test_version_info',
 
3510
                   'bzrlib.tests.test_versionedfile',
 
3511
                   'bzrlib.tests.test_weave',
 
3512
                   'bzrlib.tests.test_whitebox',
 
3513
                   'bzrlib.tests.test_win32utils',
 
3514
                   'bzrlib.tests.test_workingtree',
 
3515
                   'bzrlib.tests.test_workingtree_4',
 
3516
                   'bzrlib.tests.test_wsgi',
 
3517
                   'bzrlib.tests.test_xml',
 
3518
                   'bzrlib.tests.tree_implementations',
 
3519
                   'bzrlib.tests.workingtree_implementations',
 
3520
                   'bzrlib.util.tests.test_bencode',
 
3521
                   ]
3887
3522
 
3888
3523
    loader = TestUtil.TestLoader()
3889
3524
 
3890
 
    if keep_only is not None:
3891
 
        id_filter = TestIdList(keep_only)
3892
3525
    if starting_with:
 
3526
        starting_with = [test_prefix_alias_registry.resolve_alias(start)
 
3527
                         for start in starting_with]
3893
3528
        # We take precedence over keep_only because *at loading time* using
3894
3529
        # both options means we will load less tests for the same final result.
3895
3530
        def interesting_module(name):
3905
3540
        loader = TestUtil.FilteredByModuleTestLoader(interesting_module)
3906
3541
 
3907
3542
    elif keep_only is not None:
 
3543
        id_filter = TestIdList(keep_only)
3908
3544
        loader = TestUtil.FilteredByModuleTestLoader(id_filter.refers_to)
3909
3545
        def interesting_module(name):
3910
3546
            return id_filter.refers_to(name)
3918
3554
    suite = loader.suiteClass()
3919
3555
 
3920
3556
    # modules building their suite with loadTestsFromModuleNames
3921
 
    suite.addTest(loader.loadTestsFromModuleNames(_test_suite_testmod_names()))
3922
 
 
3923
 
    for mod in _test_suite_modules_to_doctest():
 
3557
    suite.addTest(loader.loadTestsFromModuleNames(testmod_names))
 
3558
 
 
3559
    modules_to_doctest = [
 
3560
        'bzrlib',
 
3561
        'bzrlib.branchbuilder',
 
3562
        'bzrlib.export',
 
3563
        'bzrlib.inventory',
 
3564
        'bzrlib.iterablefile',
 
3565
        'bzrlib.lockdir',
 
3566
        'bzrlib.merge3',
 
3567
        'bzrlib.option',
 
3568
        'bzrlib.symbol_versioning',
 
3569
        'bzrlib.tests',
 
3570
        'bzrlib.timestamp',
 
3571
        'bzrlib.version_info_formats.format_custom',
 
3572
        ]
 
3573
 
 
3574
    for mod in modules_to_doctest:
3924
3575
        if not interesting_module(mod):
3925
3576
            # No tests to keep here, move along
3926
3577
            continue
3955
3606
            reload(sys)
3956
3607
            sys.setdefaultencoding(default_encoding)
3957
3608
 
 
3609
    if starting_with:
 
3610
        suite = filter_suite_by_id_startswith(suite, starting_with)
 
3611
 
3958
3612
    if keep_only is not None:
3959
3613
        # Now that the referred modules have loaded their tests, keep only the
3960
3614
        # requested ones.
4014
3668
    ...     bzrlib.tests.test_sampler.DemoTest('test_nothing'),
4015
3669
    ...     [('one', dict(param=1)),
4016
3670
    ...      ('two', dict(param=2))],
4017
 
    ...     TestUtil.TestSuite())
 
3671
    ...     TestSuite())
4018
3672
    >>> tests = list(iter_suite_tests(r))
4019
3673
    >>> len(tests)
4020
3674
    2
4067
3721
    :param new_id: The id to assign to it.
4068
3722
    :return: The new test.
4069
3723
    """
4070
 
    new_test = copy.copy(test)
 
3724
    from copy import deepcopy
 
3725
    new_test = deepcopy(test)
4071
3726
    new_test.id = lambda: new_id
4072
3727
    return new_test
4073
3728
 
4074
3729
 
4075
 
def permute_tests_for_extension(standard_tests, loader, py_module_name,
4076
 
                                ext_module_name):
4077
 
    """Helper for permutating tests against an extension module.
4078
 
 
4079
 
    This is meant to be used inside a modules 'load_tests()' function. It will
4080
 
    create 2 scenarios, and cause all tests in the 'standard_tests' to be run
4081
 
    against both implementations. Setting 'test.module' to the appropriate
4082
 
    module. See bzrlib.tests.test__chk_map.load_tests as an example.
4083
 
 
4084
 
    :param standard_tests: A test suite to permute
4085
 
    :param loader: A TestLoader
4086
 
    :param py_module_name: The python path to a python module that can always
4087
 
        be loaded, and will be considered the 'python' implementation. (eg
4088
 
        'bzrlib._chk_map_py')
4089
 
    :param ext_module_name: The python path to an extension module. If the
4090
 
        module cannot be loaded, a single test will be added, which notes that
4091
 
        the module is not available. If it can be loaded, all standard_tests
4092
 
        will be run against that module.
4093
 
    :return: (suite, feature) suite is a test-suite that has all the permuted
4094
 
        tests. feature is the Feature object that can be used to determine if
4095
 
        the module is available.
4096
 
    """
4097
 
 
4098
 
    py_module = __import__(py_module_name, {}, {}, ['NO_SUCH_ATTRIB'])
4099
 
    scenarios = [
4100
 
        ('python', {'module': py_module}),
4101
 
    ]
4102
 
    suite = loader.suiteClass()
4103
 
    feature = ModuleAvailableFeature(ext_module_name)
4104
 
    if feature.available():
4105
 
        scenarios.append(('C', {'module': feature.module}))
4106
 
    else:
4107
 
        # the compiled module isn't available, so we add a failing test
4108
 
        class FailWithoutFeature(TestCase):
4109
 
            def test_fail(self):
4110
 
                self.requireFeature(feature)
4111
 
        suite.addTest(loader.loadTestsFromTestCase(FailWithoutFeature))
4112
 
    result = multiply_tests(standard_tests, scenarios, suite)
4113
 
    return result, feature
4114
 
 
4115
 
 
4116
 
def _rmtree_temp_dir(dirname, test_id=None):
 
3730
def _rmtree_temp_dir(dirname):
4117
3731
    # If LANG=C we probably have created some bogus paths
4118
3732
    # which rmtree(unicode) will fail to delete
4119
3733
    # so make sure we are using rmtree(str) to delete everything
4128
3742
    try:
4129
3743
        osutils.rmtree(dirname)
4130
3744
    except OSError, e:
4131
 
        # We don't want to fail here because some useful display will be lost
4132
 
        # otherwise. Polluting the tmp dir is bad, but not giving all the
4133
 
        # possible info to the test runner is even worse.
4134
 
        if test_id != None:
4135
 
            ui.ui_factory.clear_term()
4136
 
            sys.stderr.write('\nWhile running: %s\n' % (test_id,))
4137
 
        # Ugly, but the last thing we want here is fail, so bear with it.
4138
 
        printable_e = str(e).decode(osutils.get_user_encoding(), 'replace'
4139
 
                                    ).encode('ascii', 'replace')
4140
 
        sys.stderr.write('Unable to remove testing dir %s\n%s'
4141
 
                         % (os.path.basename(dirname), printable_e))
 
3745
        if sys.platform == 'win32' and e.errno == errno.EACCES:
 
3746
            sys.stderr.write('Permission denied: '
 
3747
                             'unable to remove testing dir '
 
3748
                             '%s\n%s'
 
3749
                             % (os.path.basename(dirname), e))
 
3750
        else:
 
3751
            raise
4142
3752
 
4143
3753
 
4144
3754
class Feature(object):
4226
3836
UnicodeFilenameFeature = _UnicodeFilenameFeature()
4227
3837
 
4228
3838
 
4229
 
class _CompatabilityThunkFeature(Feature):
4230
 
    """This feature is just a thunk to another feature.
4231
 
 
4232
 
    It issues a deprecation warning if it is accessed, to let you know that you
4233
 
    should really use a different feature.
4234
 
    """
4235
 
 
4236
 
    def __init__(self, dep_version, module, name,
4237
 
                 replacement_name, replacement_module=None):
4238
 
        super(_CompatabilityThunkFeature, self).__init__()
4239
 
        self._module = module
4240
 
        if replacement_module is None:
4241
 
            replacement_module = module
4242
 
        self._replacement_module = replacement_module
4243
 
        self._name = name
4244
 
        self._replacement_name = replacement_name
4245
 
        self._dep_version = dep_version
4246
 
        self._feature = None
4247
 
 
4248
 
    def _ensure(self):
4249
 
        if self._feature is None:
4250
 
            depr_msg = self._dep_version % ('%s.%s'
4251
 
                                            % (self._module, self._name))
4252
 
            use_msg = ' Use %s.%s instead.' % (self._replacement_module,
4253
 
                                               self._replacement_name)
4254
 
            symbol_versioning.warn(depr_msg + use_msg, DeprecationWarning)
4255
 
            # Import the new feature and use it as a replacement for the
4256
 
            # deprecated one.
4257
 
            mod = __import__(self._replacement_module, {}, {},
4258
 
                             [self._replacement_name])
4259
 
            self._feature = getattr(mod, self._replacement_name)
4260
 
 
4261
 
    def _probe(self):
4262
 
        self._ensure()
4263
 
        return self._feature._probe()
4264
 
 
4265
 
 
4266
 
class ModuleAvailableFeature(Feature):
4267
 
    """This is a feature than describes a module we want to be available.
4268
 
 
4269
 
    Declare the name of the module in __init__(), and then after probing, the
4270
 
    module will be available as 'self.module'.
4271
 
 
4272
 
    :ivar module: The module if it is available, else None.
4273
 
    """
4274
 
 
4275
 
    def __init__(self, module_name):
4276
 
        super(ModuleAvailableFeature, self).__init__()
4277
 
        self.module_name = module_name
4278
 
 
4279
 
    def _probe(self):
4280
 
        try:
4281
 
            self._module = __import__(self.module_name, {}, {}, [''])
4282
 
            return True
4283
 
        except ImportError:
4284
 
            return False
4285
 
 
4286
 
    @property
4287
 
    def module(self):
4288
 
        if self.available(): # Make sure the probe has been done
4289
 
            return self._module
4290
 
        return None
4291
 
 
4292
 
    def feature_name(self):
4293
 
        return self.module_name
4294
 
 
4295
 
 
4296
 
# This is kept here for compatibility, it is recommended to use
4297
 
# 'bzrlib.tests.feature.paramiko' instead
4298
 
ParamikoFeature = _CompatabilityThunkFeature(
4299
 
    deprecated_in((2,1,0)),
4300
 
    'bzrlib.tests.features', 'ParamikoFeature', 'paramiko')
4301
 
 
4302
 
 
4303
3839
def probe_unicode_in_user_encoding():
4304
3840
    """Try to encode several unicode strings to use in unicode-aware tests.
4305
3841
    Return first successfull match.
4374
3910
UnicodeFilename = _UnicodeFilename()
4375
3911
 
4376
3912
 
4377
 
class _ByteStringNamedFilesystem(Feature):
4378
 
    """Is the filesystem based on bytes?"""
4379
 
 
4380
 
    def _probe(self):
4381
 
        if os.name == "posix":
4382
 
            return True
4383
 
        return False
4384
 
 
4385
 
ByteStringNamedFilesystem = _ByteStringNamedFilesystem()
4386
 
 
4387
 
 
4388
3913
class _UTF8Filesystem(Feature):
4389
3914
    """Is the filesystem UTF-8?"""
4390
3915
 
4396
3921
UTF8Filesystem = _UTF8Filesystem()
4397
3922
 
4398
3923
 
4399
 
class _BreakinFeature(Feature):
4400
 
    """Does this platform support the breakin feature?"""
4401
 
 
4402
 
    def _probe(self):
4403
 
        from bzrlib import breakin
4404
 
        if breakin.determine_signal() is None:
4405
 
            return False
4406
 
        if sys.platform == 'win32':
4407
 
            # Windows doesn't have os.kill, and we catch the SIGBREAK signal.
4408
 
            # We trigger SIGBREAK via a Console api so we need ctypes to
4409
 
            # access the function
4410
 
            try:
4411
 
                import ctypes
4412
 
            except OSError:
4413
 
                return False
4414
 
        return True
4415
 
 
4416
 
    def feature_name(self):
4417
 
        return "SIGQUIT or SIGBREAK w/ctypes on win32"
4418
 
 
4419
 
 
4420
 
BreakinFeature = _BreakinFeature()
4421
 
 
4422
 
 
4423
3924
class _CaseInsCasePresFilenameFeature(Feature):
4424
3925
    """Is the file-system case insensitive, but case-preserving?"""
4425
3926
 
4475
3976
CaseInsensitiveFilesystemFeature = _CaseInsensitiveFilesystemFeature()
4476
3977
 
4477
3978
 
4478
 
class _CaseSensitiveFilesystemFeature(Feature):
 
3979
class _SubUnitFeature(Feature):
 
3980
    """Check if subunit is available."""
4479
3981
 
4480
3982
    def _probe(self):
4481
 
        if CaseInsCasePresFilenameFeature.available():
4482
 
            return False
4483
 
        elif CaseInsensitiveFilesystemFeature.available():
4484
 
            return False
4485
 
        else:
 
3983
        try:
 
3984
            import subunit
4486
3985
            return True
 
3986
        except ImportError:
 
3987
            return False
4487
3988
 
4488
3989
    def feature_name(self):
4489
 
        return 'case-sensitive filesystem'
4490
 
 
4491
 
# new coding style is for feature instances to be lowercase
4492
 
case_sensitive_filesystem_feature = _CaseSensitiveFilesystemFeature()
4493
 
 
4494
 
 
4495
 
# Kept for compatibility, use bzrlib.tests.features.subunit instead
4496
 
SubUnitFeature = _CompatabilityThunkFeature(
4497
 
    deprecated_in((2,1,0)),
4498
 
    'bzrlib.tests.features', 'SubUnitFeature', 'subunit')
 
3990
        return 'subunit'
 
3991
 
 
3992
SubUnitFeature = _SubUnitFeature()
4499
3993
# Only define SubUnitBzrRunner if subunit is available.
4500
3994
try:
4501
3995
    from subunit import TestProtocolClient
4502
 
    from subunit.test_results import AutoTimingTestResultDecorator
4503
3996
    class SubUnitBzrRunner(TextTestRunner):
4504
3997
        def run(self, test):
4505
 
            result = AutoTimingTestResultDecorator(
4506
 
                TestProtocolClient(self.stream))
 
3998
            result = TestProtocolClient(self.stream)
4507
3999
            test.run(result)
4508
4000
            return result
4509
4001
except ImportError:
4510
4002
    pass
4511
 
 
4512
 
class _PosixPermissionsFeature(Feature):
4513
 
 
4514
 
    def _probe(self):
4515
 
        def has_perms():
4516
 
            # create temporary file and check if specified perms are maintained.
4517
 
            import tempfile
4518
 
 
4519
 
            write_perms = stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR
4520
 
            f = tempfile.mkstemp(prefix='bzr_perms_chk_')
4521
 
            fd, name = f
4522
 
            os.close(fd)
4523
 
            os.chmod(name, write_perms)
4524
 
 
4525
 
            read_perms = os.stat(name).st_mode & 0777
4526
 
            os.unlink(name)
4527
 
            return (write_perms == read_perms)
4528
 
 
4529
 
        return (os.name == 'posix') and has_perms()
4530
 
 
4531
 
    def feature_name(self):
4532
 
        return 'POSIX permissions support'
4533
 
 
4534
 
posix_permissions_feature = _PosixPermissionsFeature()