13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
17
"""Testing framework extensions"""
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
18
# TODO: Perhaps there should be an API to find out if bzr running under the
19
# test suite -- some plugins might want to avoid making intrusive changes if
20
# this is the case. However, we want behaviour under to test to diverge as
21
# little as possible, so this should be used rarely if it's added at all.
22
# (Suggestion from j-a-meinel, 2005-11-24)
19
24
# NOTE: Some classes in here use camelCaseNaming() rather than
20
25
# underscore_naming(). That's for consistency with unittest; it's not the
21
26
# general style of bzrlib. Please continue that consistency when adding e.g.
22
27
# new assertFoo() methods.
27
30
from cStringIO import StringIO
39
from subprocess import Popen, PIPE
50
# nb: check this before importing anything else from within it
51
_testtools_version = getattr(testtools, '__version__', ())
52
if _testtools_version < (0, 9, 5):
53
raise ImportError("need at least testtools 0.9.5: %s is %r"
54
% (testtools.__file__, _testtools_version))
55
from testtools import content
62
commands as _mod_commands,
71
plugin as _mod_plugin,
78
transport as _mod_transport,
47
import bzrlib.bzrdir as bzrdir
48
import bzrlib.commands
49
import bzrlib.bundle.serializer
50
import bzrlib.errors as errors
51
import bzrlib.inventory
52
import bzrlib.iterablefile
82
55
import bzrlib.lsprof
83
56
except ImportError:
84
57
# lsprof not available
86
from bzrlib.smart import client, request
87
from bzrlib.transport import (
91
from bzrlib.tests import (
96
from bzrlib.ui import NullProgressView
97
from bzrlib.ui.text import TextUIFactory
99
# Mark this python module as being part of the implementation
100
# of unittest: this gives us better tracebacks where the last
101
# shown frame is the test code, not our assertXYZ.
104
default_transport = test_server.LocalURLServer
107
_unitialized_attr = object()
108
"""A sentinel needed to act as a default value in a method signature."""
111
# Subunit result codes, defined here to prevent a hard dependency on subunit.
115
# These are intentionally brought into this namespace. That way plugins, etc
116
# can just "from bzrlib.tests import TestCase, TestLoader, etc"
117
TestSuite = TestUtil.TestSuite
118
TestLoader = TestUtil.TestLoader
120
# Tests should run in a clean and clearly defined environment. The goal is to
121
# keep them isolated from the running environment as mush as possible. The test
122
# framework ensures the variables defined below are set (or deleted if the
123
# value is None) before a test is run and reset to their original value after
124
# the test is run. Generally if some code depends on an environment variable,
125
# the tests should start without this variable in the environment. There are a
126
# few exceptions but you shouldn't violate this rule lightly.
130
# bzr now uses the Win32 API and doesn't rely on APPDATA, but the
131
# tests do check our impls match APPDATA
132
'BZR_EDITOR': None, # test_msgeditor manipulates this variable
136
'BZREMAIL': None, # may still be present in the environment
137
'EMAIL': 'jrandom@example.com', # set EMAIL as bzr does not guess
138
'BZR_PROGRESS_BAR': None,
139
# This should trap leaks to ~/.bzr.log. This occurs when tests use TestCase
140
# as a base class instead of TestCaseInTempDir. Tests inheriting from
141
# TestCase should not use disk resources, BZR_LOG is one.
142
'BZR_LOG': '/you-should-use-TestCaseInTempDir-if-you-need-a-log-file',
143
'BZR_PLUGIN_PATH': None,
144
'BZR_DISABLE_PLUGINS': None,
145
'BZR_PLUGINS_AT': None,
146
'BZR_CONCURRENCY': None,
147
# Make sure that any text ui tests are consistent regardless of
148
# the environment the test case is run in; you may want tests that
149
# test other combinations. 'dumb' is a reasonable guess for tests
150
# going to a pipe or a StringIO.
156
'SSH_AUTH_SOCK': None,
166
# Nobody cares about ftp_proxy, FTP_PROXY AFAIK. So far at
167
# least. If you do (care), please update this comment
171
'BZR_REMOTE_PATH': None,
172
# Generally speaking, we don't want apport reporting on crashes in
173
# the test envirnoment unless we're specifically testing apport,
174
# so that it doesn't leak into the real system environment. We
175
# use an env var so it propagates to subprocesses.
176
'APPORT_DISABLE': '1',
180
def override_os_environ(test, env=None):
181
"""Modify os.environ keeping a copy.
183
:param test: A test instance
185
:param env: A dict containing variable definitions to be installed
188
env = isolated_environ
189
test._original_os_environ = dict([(var, value)
190
for var, value in os.environ.iteritems()])
191
for var, value in env.iteritems():
192
osutils.set_or_unset_env(var, value)
193
if var not in test._original_os_environ:
194
# The var is new, add it with a value of None, so
195
# restore_os_environ will delete it
196
test._original_os_environ[var] = None
199
def restore_os_environ(test):
200
"""Restore os.environ to its original state.
202
:param test: A test instance previously passed to override_os_environ.
204
for var, value in test._original_os_environ.iteritems():
205
# Restore the original value (or delete it if the value has been set to
206
# None in override_os_environ).
207
osutils.set_or_unset_env(var, value)
210
class ExtendedTestResult(testtools.TextTestResult):
211
"""Accepts, reports and accumulates the results of running tests.
213
Compared to the unittest version this class adds support for
214
profiling, benchmarking, stopping as soon as a test fails, and
215
skipping tests. There are further-specialized subclasses for
216
different types of display.
218
When a test finishes, in whatever way, it calls one of the addSuccess,
219
addFailure or addError classes. These in turn may redirect to a more
220
specific case for the special test results supported by our extended
223
Note that just one of these objects is fed the results from many tests.
59
from bzrlib.merge import merge_inner
62
import bzrlib.osutils as osutils
64
import bzrlib.progress as progress
65
from bzrlib.revision import common_ancestor
68
from bzrlib.transport import get_transport
69
import bzrlib.transport
70
from bzrlib.transport.local import LocalRelpathServer
71
from bzrlib.transport.readonly import ReadonlyServer
72
from bzrlib.trace import mutter
73
from bzrlib.tests import TestUtil
74
from bzrlib.tests.TestUtil import (
78
from bzrlib.tests.treeshape import build_tree_contents
79
import bzrlib.urlutils as urlutils
80
from bzrlib.workingtree import WorkingTree, WorkingTreeFormat2
82
default_transport = LocalRelpathServer
85
MODULES_TO_DOCTEST = [
87
bzrlib.bundle.serializer,
100
def packages_to_test():
101
"""Return a list of packages to test.
103
The packages are not globally imported so that import failures are
104
triggered when running selftest, not when importing the command.
107
import bzrlib.tests.blackbox
108
import bzrlib.tests.branch_implementations
109
import bzrlib.tests.bzrdir_implementations
110
import bzrlib.tests.interrepository_implementations
111
import bzrlib.tests.interversionedfile_implementations
112
import bzrlib.tests.intertree_implementations
113
import bzrlib.tests.repository_implementations
114
import bzrlib.tests.revisionstore_implementations
115
import bzrlib.tests.tree_implementations
116
import bzrlib.tests.workingtree_implementations
119
bzrlib.tests.blackbox,
120
bzrlib.tests.branch_implementations,
121
bzrlib.tests.bzrdir_implementations,
122
bzrlib.tests.interrepository_implementations,
123
bzrlib.tests.interversionedfile_implementations,
124
bzrlib.tests.intertree_implementations,
125
bzrlib.tests.repository_implementations,
126
bzrlib.tests.revisionstore_implementations,
127
bzrlib.tests.tree_implementations,
128
bzrlib.tests.workingtree_implementations,
132
class _MyResult(unittest._TextTestResult):
133
"""Custom TestResult.
135
Shows output in a different format, including displaying runtime for tests.
226
137
stop_early = False
228
def __init__(self, stream, descriptions, verbosity,
232
"""Construct new TestResult.
234
:param bench_history: Optionally, a writable file object to accumulate
237
testtools.TextTestResult.__init__(self, stream)
238
if bench_history is not None:
239
from bzrlib.version import _get_bzr_source_tree
240
src_tree = _get_bzr_source_tree()
243
revision_id = src_tree.get_parent_ids()[0]
245
# XXX: if this is a brand new tree, do the same as if there
249
# XXX: If there's no branch, what should we do?
251
bench_history.write("--date %s %s\n" % (time.time(), revision_id))
252
self._bench_history = bench_history
253
self.ui = ui.ui_factory
256
self.failure_count = 0
257
self.known_failure_count = 0
259
self.not_applicable_count = 0
260
self.unsupported = {}
262
self._overall_start_time = time.time()
263
self._strict = strict
264
self._first_thread_leaker_id = None
265
self._tests_leaking_threads_count = 0
266
self._traceback_from_test = None
268
def stopTestRun(self):
271
stopTime = time.time()
272
timeTaken = stopTime - self.startTime
273
# GZ 2010-07-19: Seems testtools has no printErrors method, and though
274
# the parent class method is similar have to duplicate
275
self._show_list('ERROR', self.errors)
276
self._show_list('FAIL', self.failures)
277
self.stream.write(self.sep2)
278
self.stream.write("%s %d test%s in %.3fs\n\n" % (actionTaken,
279
run, run != 1 and "s" or "", timeTaken))
280
if not self.wasSuccessful():
281
self.stream.write("FAILED (")
282
failed, errored = map(len, (self.failures, self.errors))
284
self.stream.write("failures=%d" % failed)
286
if failed: self.stream.write(", ")
287
self.stream.write("errors=%d" % errored)
288
if self.known_failure_count:
289
if failed or errored: self.stream.write(", ")
290
self.stream.write("known_failure_count=%d" %
291
self.known_failure_count)
292
self.stream.write(")\n")
294
if self.known_failure_count:
295
self.stream.write("OK (known_failures=%d)\n" %
296
self.known_failure_count)
298
self.stream.write("OK\n")
299
if self.skip_count > 0:
300
skipped = self.skip_count
301
self.stream.write('%d test%s skipped\n' %
302
(skipped, skipped != 1 and "s" or ""))
304
for feature, count in sorted(self.unsupported.items()):
305
self.stream.write("Missing feature '%s' skipped %d tests.\n" %
308
ok = self.wasStrictlySuccessful()
310
ok = self.wasSuccessful()
311
if self._first_thread_leaker_id:
313
'%s is leaking threads among %d leaking tests.\n' % (
314
self._first_thread_leaker_id,
315
self._tests_leaking_threads_count))
316
# We don't report the main thread as an active one.
318
'%d non-main threads were left active in the end.\n'
319
% (len(self._active_threads) - 1))
321
def getDescription(self, test):
324
def _extractBenchmarkTime(self, testCase, details=None):
139
def __init__(self, stream, descriptions, verbosity, pb=None):
140
unittest._TextTestResult.__init__(self, stream, descriptions, verbosity)
143
def extractBenchmarkTime(self, testCase):
325
144
"""Add a benchmark time for the current test case."""
326
if details and 'benchtime' in details:
327
return float(''.join(details['benchtime'].iter_bytes()))
328
return getattr(testCase, "_benchtime", None)
145
self._benchmarkTime = getattr(testCase, "_benchtime", None)
330
147
def _elapsedTestTimeString(self):
331
148
"""Return a time string for the overall time the current test has taken."""
332
return self._formatTime(self._delta_to_float(
333
self._now() - self._start_datetime))
149
return self._formatTime(time.time() - self._start_time)
335
def _testTimeString(self, testCase):
336
benchmark_time = self._extractBenchmarkTime(testCase)
337
if benchmark_time is not None:
338
return self._formatTime(benchmark_time) + "*"
151
def _testTimeString(self):
152
if self._benchmarkTime is not None:
154
self._formatTime(self._benchmarkTime),
155
self._elapsedTestTimeString())
340
return self._elapsedTestTimeString()
157
return " %s" % self._elapsedTestTimeString()
342
159
def _formatTime(self, seconds):
343
160
"""Format seconds as milliseconds with leading spaces."""
344
# some benchmarks can take thousands of seconds to run, so we need 8
346
return "%8dms" % (1000 * seconds)
348
def _shortened_test_description(self, test):
350
what = re.sub(r'^bzrlib\.tests\.', '', what)
353
# GZ 2010-10-04: Cloned tests may end up harmlessly calling this method
354
# multiple times in a row, because the handler is added for
355
# each test but the container list is shared between cases.
356
# See lp:498869 lp:625574 and lp:637725 for background.
357
def _record_traceback_from_test(self, exc_info):
358
"""Store the traceback from passed exc_info tuple till"""
359
self._traceback_from_test = exc_info[2]
161
return "%5dms" % (1000 * seconds)
163
def _ellipsise_unimportant_words(self, a_string, final_width,
165
"""Add ellipses (sp?) for overly long strings.
167
:param keep_start: If true preserve the start of a_string rather
171
if len(a_string) > final_width:
172
result = a_string[:final_width-3] + '...'
176
if len(a_string) > final_width:
177
result = '...' + a_string[3-final_width:]
180
return result.ljust(final_width)
361
182
def startTest(self, test):
362
super(ExtendedTestResult, self).startTest(test)
366
self.report_test_start(test)
367
test.number = self.count
183
unittest.TestResult.startTest(self, test)
184
# In a short description, the important words are in
185
# the beginning, but in an id, the important words are
187
SHOW_DESCRIPTIONS = False
189
if not self.showAll and self.dots and self.pb is not None:
192
final_width = osutils.terminal_width()
193
final_width = final_width - 15 - 8
195
if SHOW_DESCRIPTIONS:
196
what = test.shortDescription()
198
what = self._ellipsise_unimportant_words(what, final_width, keep_start=True)
201
if what.startswith('bzrlib.tests.'):
203
what = self._ellipsise_unimportant_words(what, final_width)
205
self.stream.write(what)
206
elif self.dots and self.pb is not None:
207
self.pb.update(what, self.testsRun - 1, None)
368
209
self._recordTestStartTime()
369
# Make testtools cases give us the real traceback on failure
370
addOnException = getattr(test, "addOnException", None)
371
if addOnException is not None:
372
addOnException(self._record_traceback_from_test)
373
# Only check for thread leaks on bzrlib derived test cases
374
if isinstance(test, TestCase):
375
test.addCleanup(self._check_leaked_threads, test)
377
def stopTest(self, test):
378
super(ExtendedTestResult, self).stopTest(test)
379
# Manually break cycles, means touching various private things but hey
380
getDetails = getattr(test, "getDetails", None)
381
if getDetails is not None:
383
type_equality_funcs = getattr(test, "_type_equality_funcs", None)
384
if type_equality_funcs is not None:
385
type_equality_funcs.clear()
386
self._traceback_from_test = None
388
def startTests(self):
389
self.report_tests_starting()
390
self._active_threads = threading.enumerate()
392
def _check_leaked_threads(self, test):
393
"""See if any threads have leaked since last call
395
A sample of live threads is stored in the _active_threads attribute,
396
when this method runs it compares the current live threads and any not
397
in the previous sample are treated as having leaked.
399
now_active_threads = set(threading.enumerate())
400
threads_leaked = now_active_threads.difference(self._active_threads)
402
self._report_thread_leak(test, threads_leaked, now_active_threads)
403
self._tests_leaking_threads_count += 1
404
if self._first_thread_leaker_id is None:
405
self._first_thread_leaker_id = test.id()
406
self._active_threads = now_active_threads
408
211
def _recordTestStartTime(self):
409
212
"""Record that a test has started."""
410
self._start_datetime = self._now()
213
self._start_time = time.time()
412
215
def addError(self, test, err):
413
"""Tell result that test finished with an error.
415
Called from the TestCase run() method when the test
416
fails with an unexpected error.
418
self._post_mortem(self._traceback_from_test)
419
super(ExtendedTestResult, self).addError(test, err)
420
self.error_count += 1
421
self.report_error(test, err)
216
if isinstance(err[1], TestSkipped):
217
return self.addSkipped(test, err)
218
unittest.TestResult.addError(self, test, err)
219
self.extractBenchmarkTime(test)
221
self.stream.writeln("ERROR %s" % self._testTimeString())
222
elif self.dots and self.pb is None:
223
self.stream.write('E')
225
self.pb.update(self._ellipsise_unimportant_words('ERROR', 13), self.testsRun, None)
226
self.pb.note(self._ellipsise_unimportant_words(
227
test.id() + ': ERROR',
228
osutils.terminal_width()))
422
230
if self.stop_early:
425
233
def addFailure(self, test, err):
426
"""Tell result that test failed.
428
Called from the TestCase run() method when the test
429
fails because e.g. an assert() method failed.
431
self._post_mortem(self._traceback_from_test)
432
super(ExtendedTestResult, self).addFailure(test, err)
433
self.failure_count += 1
434
self.report_failure(test, err)
438
def addSuccess(self, test, details=None):
439
"""Tell result that test completed successfully.
441
Called from the TestCase run()
443
if self._bench_history is not None:
444
benchmark_time = self._extractBenchmarkTime(test, details)
445
if benchmark_time is not None:
446
self._bench_history.write("%s %s\n" % (
447
self._formatTime(benchmark_time),
449
self.report_success(test)
450
super(ExtendedTestResult, self).addSuccess(test)
451
test._log_contents = ''
453
def addExpectedFailure(self, test, err):
454
self.known_failure_count += 1
455
self.report_known_failure(test, err)
457
def addUnexpectedSuccess(self, test, details=None):
458
"""Tell result the test unexpectedly passed, counting as a failure
460
When the minimum version of testtools required becomes 0.9.8 this
461
can be updated to use the new handling there.
463
super(ExtendedTestResult, self).addFailure(test, details=details)
464
self.failure_count += 1
465
self.report_unexpected_success(test,
466
"".join(details["reason"].iter_text()))
470
def addNotSupported(self, test, feature):
471
"""The test will not be run because of a missing feature.
473
# this can be called in two different ways: it may be that the
474
# test started running, and then raised (through requireFeature)
475
# UnavailableFeature. Alternatively this method can be called
476
# while probing for features before running the test code proper; in
477
# that case we will see startTest and stopTest, but the test will
478
# never actually run.
479
self.unsupported.setdefault(str(feature), 0)
480
self.unsupported[str(feature)] += 1
481
self.report_unsupported(test, feature)
483
def addSkip(self, test, reason):
484
"""A test has not run for 'reason'."""
486
self.report_skip(test, reason)
488
def addNotApplicable(self, test, reason):
489
self.not_applicable_count += 1
490
self.report_not_applicable(test, reason)
492
def _post_mortem(self, tb=None):
493
"""Start a PDB post mortem session."""
494
if os.environ.get('BZR_TEST_PDB', None):
498
def progress(self, offset, whence):
499
"""The test is adjusting the count of tests to run."""
500
if whence == SUBUNIT_SEEK_SET:
501
self.num_tests = offset
502
elif whence == SUBUNIT_SEEK_CUR:
503
self.num_tests += offset
505
raise errors.BzrError("Unknown whence %r" % whence)
507
def report_tests_starting(self):
508
"""Display information before the test run begins"""
509
if getattr(sys, 'frozen', None) is None:
510
bzr_path = osutils.realpath(sys.argv[0])
512
bzr_path = sys.executable
514
'bzr selftest: %s\n' % (bzr_path,))
517
bzrlib.__path__[0],))
519
' bzr-%s python-%s %s\n' % (
520
bzrlib.version_string,
521
bzrlib._format_version_tuple(sys.version_info),
522
platform.platform(aliased=1),
524
self.stream.write('\n')
526
def report_test_start(self, test):
527
"""Display information on the test just about to be run"""
529
def _report_thread_leak(self, test, leaked_threads, active_threads):
530
"""Display information on a test that leaked one or more threads"""
531
# GZ 2010-09-09: A leak summary reported separately from the general
532
# thread debugging would be nice. Tests under subunit
533
# need something not using stream, perhaps adding a
534
# testtools details object would be fitting.
535
if 'threads' in selftest_debug_flags:
536
self.stream.write('%s is leaking, active is now %d\n' %
537
(test.id(), len(active_threads)))
539
def startTestRun(self):
540
self.startTime = time.time()
542
def report_success(self, test):
545
def wasStrictlySuccessful(self):
546
if self.unsupported or self.known_failure_count:
548
return self.wasSuccessful()
551
class TextTestResult(ExtendedTestResult):
552
"""Displays progress and results of tests in text form"""
554
def __init__(self, stream, descriptions, verbosity,
559
ExtendedTestResult.__init__(self, stream, descriptions, verbosity,
560
bench_history, strict)
561
# We no longer pass them around, but just rely on the UIFactory stack
564
warnings.warn("Passing pb to TextTestResult is deprecated")
565
self.pb = self.ui.nested_progress_bar()
566
self.pb.show_pct = False
567
self.pb.show_spinner = False
568
self.pb.show_eta = False,
569
self.pb.show_count = False
570
self.pb.show_bar = False
571
self.pb.update_latency = 0
572
self.pb.show_transport_activity = False
574
def stopTestRun(self):
575
# called when the tests that are going to run have run
578
super(TextTestResult, self).stopTestRun()
580
def report_tests_starting(self):
581
super(TextTestResult, self).report_tests_starting()
582
self.pb.update('[test 0/%d] Starting' % (self.num_tests))
584
def _progress_prefix_text(self):
585
# the longer this text, the less space we have to show the test
587
a = '[%d' % self.count # total that have been run
588
# tests skipped as known not to be relevant are not important enough
590
## if self.skip_count:
591
## a += ', %d skip' % self.skip_count
592
## if self.known_failure_count:
593
## a += '+%dX' % self.known_failure_count
595
a +='/%d' % self.num_tests
597
runtime = time.time() - self._overall_start_time
599
a += '%dm%ds' % (runtime / 60, runtime % 60)
602
total_fail_count = self.error_count + self.failure_count
604
a += ', %d failed' % total_fail_count
605
# if self.unsupported:
606
# a += ', %d missing' % len(self.unsupported)
610
def report_test_start(self, test):
612
self._progress_prefix_text()
614
+ self._shortened_test_description(test))
616
def _test_description(self, test):
617
return self._shortened_test_description(test)
619
def report_error(self, test, err):
620
self.stream.write('ERROR: %s\n %s\n' % (
621
self._test_description(test),
625
def report_failure(self, test, err):
626
self.stream.write('FAIL: %s\n %s\n' % (
627
self._test_description(test),
631
def report_known_failure(self, test, err):
634
def report_unexpected_success(self, test, reason):
635
self.stream.write('FAIL: %s\n %s: %s\n' % (
636
self._test_description(test),
637
"Unexpected success. Should have failed",
641
def report_skip(self, test, reason):
644
def report_not_applicable(self, test, reason):
647
def report_unsupported(self, test, feature):
648
"""test cannot be run because feature is missing."""
651
class VerboseTestResult(ExtendedTestResult):
652
"""Produce long output, with one line per test run plus times"""
654
def _ellipsize_to_right(self, a_string, final_width):
655
"""Truncate and pad a string, keeping the right hand side"""
656
if len(a_string) > final_width:
657
result = '...' + a_string[3-final_width:]
660
return result.ljust(final_width)
662
def report_tests_starting(self):
663
self.stream.write('running %d tests...\n' % self.num_tests)
664
super(VerboseTestResult, self).report_tests_starting()
666
def report_test_start(self, test):
667
name = self._shortened_test_description(test)
668
width = osutils.terminal_width()
669
if width is not None:
670
# width needs space for 6 char status, plus 1 for slash, plus an
671
# 11-char time string, plus a trailing blank
672
# when NUMBERED_DIRS: plus 5 chars on test number, plus 1 char on
674
self.stream.write(self._ellipsize_to_right(name, width-18))
676
self.stream.write(name)
679
def _error_summary(self, err):
681
return '%s%s' % (indent, err[1])
683
def report_error(self, test, err):
684
self.stream.write('ERROR %s\n%s\n'
685
% (self._testTimeString(test),
686
self._error_summary(err)))
688
def report_failure(self, test, err):
689
self.stream.write(' FAIL %s\n%s\n'
690
% (self._testTimeString(test),
691
self._error_summary(err)))
693
def report_known_failure(self, test, err):
694
self.stream.write('XFAIL %s\n%s\n'
695
% (self._testTimeString(test),
696
self._error_summary(err)))
698
def report_unexpected_success(self, test, reason):
699
self.stream.write(' FAIL %s\n%s: %s\n'
700
% (self._testTimeString(test),
701
"Unexpected success. Should have failed",
704
def report_success(self, test):
705
self.stream.write(' OK %s\n' % self._testTimeString(test))
706
for bench_called, stats in getattr(test, '_benchcalls', []):
707
self.stream.write('LSProf output for %s(%s, %s)\n' % bench_called)
708
stats.pprint(file=self.stream)
709
# flush the stream so that we get smooth output. This verbose mode is
710
# used to show the output in PQM.
713
def report_skip(self, test, reason):
714
self.stream.write(' SKIP %s\n%s\n'
715
% (self._testTimeString(test), reason))
717
def report_not_applicable(self, test, reason):
718
self.stream.write(' N/A %s\n %s\n'
719
% (self._testTimeString(test), reason))
721
def report_unsupported(self, test, feature):
722
"""test cannot be run because feature is missing."""
723
self.stream.write("NODEP %s\n The feature '%s' is not available.\n"
724
%(self._testTimeString(test), feature))
234
unittest.TestResult.addFailure(self, test, err)
235
self.extractBenchmarkTime(test)
237
self.stream.writeln(" FAIL %s" % self._testTimeString())
238
elif self.dots and self.pb is None:
239
self.stream.write('F')
241
self.pb.update(self._ellipsise_unimportant_words('FAIL', 13), self.testsRun, None)
242
self.pb.note(self._ellipsise_unimportant_words(
243
test.id() + ': FAIL',
244
osutils.terminal_width()))
249
def addSuccess(self, test):
250
self.extractBenchmarkTime(test)
252
self.stream.writeln(' OK %s' % self._testTimeString())
253
for bench_called, stats in getattr(test, '_benchcalls', []):
254
self.stream.writeln('LSProf output for %s(%s, %s)' % bench_called)
255
stats.pprint(file=self.stream)
256
elif self.dots and self.pb is None:
257
self.stream.write('~')
259
self.pb.update(self._ellipsise_unimportant_words('OK', 13), self.testsRun, None)
261
unittest.TestResult.addSuccess(self, test)
263
def addSkipped(self, test, skip_excinfo):
264
self.extractBenchmarkTime(test)
266
print >>self.stream, ' SKIP %s' % self._testTimeString()
267
print >>self.stream, ' %s' % skip_excinfo[1]
268
elif self.dots and self.pb is None:
269
self.stream.write('S')
271
self.pb.update(self._ellipsise_unimportant_words('SKIP', 13), self.testsRun, None)
273
# seems best to treat this as success from point-of-view of unittest
274
# -- it actually does nothing so it barely matters :)
277
except KeyboardInterrupt:
280
self.addError(test, test.__exc_info())
282
unittest.TestResult.addSuccess(self, test)
284
def printErrorList(self, flavour, errors):
285
for test, err in errors:
286
self.stream.writeln(self.separator1)
287
self.stream.writeln("%s: %s" % (flavour, self.getDescription(test)))
288
if getattr(test, '_get_log', None) is not None:
290
print >>self.stream, \
291
('vvvv[log from %s]' % test.id()).ljust(78,'-')
292
print >>self.stream, test._get_log()
293
print >>self.stream, \
294
('^^^^[log from %s]' % test.id()).ljust(78,'-')
295
self.stream.writeln(self.separator2)
296
self.stream.writeln("%s" % err)
727
299
class TextTestRunner(object):
942
439
retrieved by _get_log(). We use a real OS file, not an in-memory object,
943
440
so that it can also capture file IO. When the test completes this file
944
441
is read into memory and removed from disk.
946
443
There are also convenience functions to invoke bzr's command-line
947
444
routine, and to build and check bzr trees.
949
446
In addition to the usual method of overriding tearDown(), this class also
950
allows subclasses to register cleanup functions via addCleanup, which are
447
allows subclasses to register functions into the _cleanups list, which is
951
448
run in order as the object is torn down. It's less likely this will be
952
449
accidentally overlooked.
452
_log_file_name = None
956
454
# record lsprof data when performing benchmark calls.
957
455
_gather_lsprof_in_benchmarks = False
959
457
def __init__(self, methodName='testMethod'):
960
458
super(TestCase, self).__init__(methodName)
961
self._directory_isolation = True
962
self.exception_handlers.insert(0,
963
(UnavailableFeature, self._do_unsupported_or_skip))
964
self.exception_handlers.insert(0,
965
(TestNotApplicable, self._do_not_applicable))
968
super(TestCase, self).setUp()
969
for feature in getattr(self, '_test_needs_features', []):
970
self.requireFeature(feature)
462
unittest.TestCase.setUp(self)
971
463
self._cleanEnvironment()
464
bzrlib.trace.disable_default_logging()
973
465
self._startLogFile()
974
466
self._benchcalls = []
975
467
self._benchtime = None
977
self._track_transports()
979
self._clear_debug_flags()
980
# Isolate global verbosity level, to make sure it's reproducible
981
# between tests. We should get rid of this altogether: bug 656694. --
983
self.overrideAttr(bzrlib.trace, '_verbosity_level', 0)
984
# Isolate config option expansion until its default value for bzrlib is
985
# settled on or a the FIXME associated with _get_expand_default_value
986
# is addressed -- vila 20110219
987
self.overrideAttr(config, '_expand_default_value', None)
988
self._log_files = set()
989
# Each key in the ``_counters`` dict holds a value for a different
990
# counter. When the test ends, addDetail() should be used to output the
991
# counter values. This happens in install_counter_hook().
993
if 'config_stats' in selftest_debug_flags:
994
self._install_config_stats_hooks()
999
pdb.Pdb().set_trace(sys._getframe().f_back)
1001
def discardDetail(self, name):
1002
"""Extend the addDetail, getDetails api so we can remove a detail.
1004
eg. bzr always adds the 'log' detail at startup, but we don't want to
1005
include it for skipped, xfail, etc tests.
1007
It is safe to call this for a detail that doesn't exist, in case this
1008
gets called multiple times.
1010
# We cheat. details is stored in __details which means we shouldn't
1011
# touch it. but getDetails() returns the dict directly, so we can
1013
details = self.getDetails()
1017
def install_counter_hook(self, hooks, name, counter_name=None):
1018
"""Install a counting hook.
1020
Any hook can be counted as long as it doesn't need to return a value.
1022
:param hooks: Where the hook should be installed.
1024
:param name: The hook name that will be counted.
1026
:param counter_name: The counter identifier in ``_counters``, defaults
1029
_counters = self._counters # Avoid closing over self
1030
if counter_name is None:
1032
if _counters.has_key(counter_name):
1033
raise AssertionError('%s is already used as a counter name'
1035
_counters[counter_name] = 0
1036
self.addDetail(counter_name, content.Content(content.UTF8_TEXT,
1037
lambda: ['%d' % (_counters[counter_name],)]))
1038
def increment_counter(*args, **kwargs):
1039
_counters[counter_name] += 1
1040
label = 'count %s calls' % (counter_name,)
1041
hooks.install_named_hook(name, increment_counter, label)
1042
self.addCleanup(hooks.uninstall_named_hook, name, label)
1044
def _install_config_stats_hooks(self):
1045
"""Install config hooks to count hook calls.
1048
for hook_name in ('get', 'set', 'remove', 'load', 'save'):
1049
self.install_counter_hook(config.ConfigHooks, hook_name,
1050
'config.%s' % (hook_name,))
1052
# The OldConfigHooks are private and need special handling to protect
1053
# against recursive tests (tests that run other tests), so we just do
1054
# manually what registering them into _builtin_known_hooks will provide
1056
self.overrideAttr(config, 'OldConfigHooks', config._OldConfigHooks())
1057
for hook_name in ('get', 'set', 'remove', 'load', 'save'):
1058
self.install_counter_hook(config.OldConfigHooks, hook_name,
1059
'old_config.%s' % (hook_name,))
1061
def _clear_debug_flags(self):
1062
"""Prevent externally set debug flags affecting tests.
1064
Tests that want to use debug flags can just set them in the
1065
debug_flags set during setup/teardown.
1067
# Start with a copy of the current debug flags we can safely modify.
1068
self.overrideAttr(debug, 'debug_flags', set(debug.debug_flags))
1069
if 'allow_debug' not in selftest_debug_flags:
1070
debug.debug_flags.clear()
1071
if 'disable_lock_checks' not in selftest_debug_flags:
1072
debug.debug_flags.add('strict_locks')
1074
def _clear_hooks(self):
1075
# prevent hooks affecting tests
1076
known_hooks = hooks.known_hooks
1077
self._preserved_hooks = {}
1078
for key, (parent, name) in known_hooks.iter_parent_objects():
1079
current_hooks = getattr(parent, name)
1080
self._preserved_hooks[parent] = (name, current_hooks)
1081
self._preserved_lazy_hooks = hooks._lazy_hooks
1082
hooks._lazy_hooks = {}
1083
self.addCleanup(self._restoreHooks)
1084
for key, (parent, name) in known_hooks.iter_parent_objects():
1085
factory = known_hooks.get(key)
1086
setattr(parent, name, factory())
1087
# this hook should always be installed
1088
request._install_hook()
1090
def disable_directory_isolation(self):
1091
"""Turn off directory isolation checks."""
1092
self._directory_isolation = False
1094
def enable_directory_isolation(self):
1095
"""Enable directory isolation checks."""
1096
self._directory_isolation = True
1098
def _silenceUI(self):
1099
"""Turn off UI for duration of test"""
1100
# by default the UI is off; tests can turn it on if they want it.
1101
self.overrideAttr(ui, 'ui_factory', ui.SilentUIFactory())
1103
def _check_locks(self):
1104
"""Check that all lock take/release actions have been paired."""
1105
# We always check for mismatched locks. If a mismatch is found, we
1106
# fail unless -Edisable_lock_checks is supplied to selftest, in which
1107
# case we just print a warning.
1109
acquired_locks = [lock for action, lock in self._lock_actions
1110
if action == 'acquired']
1111
released_locks = [lock for action, lock in self._lock_actions
1112
if action == 'released']
1113
broken_locks = [lock for action, lock in self._lock_actions
1114
if action == 'broken']
1115
# trivially, given the tests for lock acquistion and release, if we
1116
# have as many in each list, it should be ok. Some lock tests also
1117
# break some locks on purpose and should be taken into account by
1118
# considering that breaking a lock is just a dirty way of releasing it.
1119
if len(acquired_locks) != (len(released_locks) + len(broken_locks)):
1121
'Different number of acquired and '
1122
'released or broken locks.\n'
1126
(acquired_locks, released_locks, broken_locks))
1127
if not self._lock_check_thorough:
1128
# Rather than fail, just warn
1129
print "Broken test %s: %s" % (self, message)
1133
def _track_locks(self):
1134
"""Track lock activity during tests."""
1135
self._lock_actions = []
1136
if 'disable_lock_checks' in selftest_debug_flags:
1137
self._lock_check_thorough = False
1139
self._lock_check_thorough = True
1141
self.addCleanup(self._check_locks)
1142
_mod_lock.Lock.hooks.install_named_hook('lock_acquired',
1143
self._lock_acquired, None)
1144
_mod_lock.Lock.hooks.install_named_hook('lock_released',
1145
self._lock_released, None)
1146
_mod_lock.Lock.hooks.install_named_hook('lock_broken',
1147
self._lock_broken, None)
1149
def _lock_acquired(self, result):
1150
self._lock_actions.append(('acquired', result))
1152
def _lock_released(self, result):
1153
self._lock_actions.append(('released', result))
1155
def _lock_broken(self, result):
1156
self._lock_actions.append(('broken', result))
1158
def permit_dir(self, name):
1159
"""Permit a directory to be used by this test. See permit_url."""
1160
name_transport = _mod_transport.get_transport(name)
1161
self.permit_url(name)
1162
self.permit_url(name_transport.base)
1164
def permit_url(self, url):
1165
"""Declare that url is an ok url to use in this test.
1167
Do this for memory transports, temporary test directory etc.
1169
Do not do this for the current working directory, /tmp, or any other
1170
preexisting non isolated url.
1172
if not url.endswith('/'):
1174
self._bzr_selftest_roots.append(url)
1176
def permit_source_tree_branch_repo(self):
1177
"""Permit the source tree bzr is running from to be opened.
1179
Some code such as bzrlib.version attempts to read from the bzr branch
1180
that bzr is executing from (if any). This method permits that directory
1181
to be used in the test suite.
1183
path = self.get_source_path()
1184
self.record_directory_isolation()
1187
workingtree.WorkingTree.open(path)
1188
except (errors.NotBranchError, errors.NoWorkingTree):
1189
raise TestSkipped('Needs a working tree of bzr sources')
1191
self.enable_directory_isolation()
1193
def _preopen_isolate_transport(self, transport):
1194
"""Check that all transport openings are done in the test work area."""
1195
while isinstance(transport, pathfilter.PathFilteringTransport):
1196
# Unwrap pathfiltered transports
1197
transport = transport.server.backing_transport.clone(
1198
transport._filter('.'))
1199
url = transport.base
1200
# ReadonlySmartTCPServer_for_testing decorates the backing transport
1201
# urls it is given by prepending readonly+. This is appropriate as the
1202
# client shouldn't know that the server is readonly (or not readonly).
1203
# We could register all servers twice, with readonly+ prepending, but
1204
# that makes for a long list; this is about the same but easier to
1206
if url.startswith('readonly+'):
1207
url = url[len('readonly+'):]
1208
self._preopen_isolate_url(url)
1210
def _preopen_isolate_url(self, url):
1211
if not self._directory_isolation:
1213
if self._directory_isolation == 'record':
1214
self._bzr_selftest_roots.append(url)
1216
# This prevents all transports, including e.g. sftp ones backed on disk
1217
# from working unless they are explicitly granted permission. We then
1218
# depend on the code that sets up test transports to check that they are
1219
# appropriately isolated and enable their use by calling
1220
# self.permit_transport()
1221
if not osutils.is_inside_any(self._bzr_selftest_roots, url):
1222
raise errors.BzrError("Attempt to escape test isolation: %r %r"
1223
% (url, self._bzr_selftest_roots))
1225
def record_directory_isolation(self):
1226
"""Gather accessed directories to permit later access.
1228
This is used for tests that access the branch bzr is running from.
1230
self._directory_isolation = "record"
1232
def start_server(self, transport_server, backing_server=None):
1233
"""Start transport_server for this test.
1235
This starts the server, registers a cleanup for it and permits the
1236
server's urls to be used.
1238
if backing_server is None:
1239
transport_server.start_server()
1241
transport_server.start_server(backing_server)
1242
self.addCleanup(transport_server.stop_server)
1243
# Obtain a real transport because if the server supplies a password, it
1244
# will be hidden from the base on the client side.
1245
t = _mod_transport.get_transport(transport_server.get_url())
1246
# Some transport servers effectively chroot the backing transport;
1247
# others like SFTPServer don't - users of the transport can walk up the
1248
# transport to read the entire backing transport. This wouldn't matter
1249
# except that the workdir tests are given - and that they expect the
1250
# server's url to point at - is one directory under the safety net. So
1251
# Branch operations into the transport will attempt to walk up one
1252
# directory. Chrooting all servers would avoid this but also mean that
1253
# we wouldn't be testing directly against non-root urls. Alternatively
1254
# getting the test framework to start the server with a backing server
1255
# at the actual safety net directory would work too, but this then
1256
# means that the self.get_url/self.get_transport methods would need
1257
# to transform all their results. On balance its cleaner to handle it
1258
# here, and permit a higher url when we have one of these transports.
1259
if t.base.endswith('/work/'):
1260
# we have safety net/test root/work
1261
t = t.clone('../..')
1262
elif isinstance(transport_server,
1263
test_server.SmartTCPServer_for_testing):
1264
# The smart server adds a path similar to work, which is traversed
1265
# up from by the client. But the server is chrooted - the actual
1266
# backing transport is not escaped from, and VFS requests to the
1267
# root will error (because they try to escape the chroot).
1269
while t2.base != t.base:
1272
self.permit_url(t.base)
1274
def _track_transports(self):
1275
"""Install checks for transport usage."""
1276
# TestCase has no safe place it can write to.
1277
self._bzr_selftest_roots = []
1278
# Currently the easiest way to be sure that nothing is going on is to
1279
# hook into bzr dir opening. This leaves a small window of error for
1280
# transport tests, but they are well known, and we can improve on this
1282
bzrdir.BzrDir.hooks.install_named_hook("pre_open",
1283
self._preopen_isolate_transport, "Check bzr directories are safe.")
1285
469
def _ndiff_strings(self, a, b):
1286
470
"""Return ndiff between two strings containing lines.
1288
472
A trailing newline is added if missing to make the strings
1289
473
print properly."""
1290
474
if b and b[-1] != '\n':
1503
544
path_stat = transport.stat(path)
1504
545
actual_mode = stat.S_IMODE(path_stat.st_mode)
1505
546
self.assertEqual(mode, actual_mode,
1506
'mode of %r incorrect (%s != %s)'
1507
% (path, oct(mode), oct(actual_mode)))
1509
def assertIsSameRealPath(self, path1, path2):
1510
"""Fail if path1 and path2 points to different files"""
1511
self.assertEqual(osutils.realpath(path1),
1512
osutils.realpath(path2),
1513
"apparent paths:\na = %s\nb = %s\n," % (path1, path2))
1515
def assertIsInstance(self, obj, kls, msg=None):
1516
"""Fail if obj is not an instance of kls
1518
:param msg: Supplementary message to show if the assertion fails.
547
'mode of %r incorrect (%o != %o)' % (path, mode, actual_mode))
549
def assertIsInstance(self, obj, kls):
550
"""Fail if obj is not an instance of kls"""
1520
551
if not isinstance(obj, kls):
1521
m = "%r is an instance of %s rather than %s" % (
1522
obj, obj.__class__, kls)
1527
def assertFileEqual(self, content, path):
1528
"""Fail if path does not contain 'content'."""
1529
self.assertPathExists(path)
1530
f = file(path, 'rb')
1535
self.assertEqualDiff(content, s)
1537
def assertDocstring(self, expected_docstring, obj):
1538
"""Fail if obj does not have expected_docstring"""
1540
# With -OO the docstring should be None instead
1541
self.assertIs(obj.__doc__, None)
1543
self.assertEqual(expected_docstring, obj.__doc__)
1545
@symbol_versioning.deprecated_method(symbol_versioning.deprecated_in((2, 4)))
1546
def failUnlessExists(self, path):
1547
return self.assertPathExists(path)
1549
def assertPathExists(self, path):
1550
"""Fail unless path or paths, which may be abs or relative, exist."""
1551
if not isinstance(path, basestring):
1553
self.assertPathExists(p)
1555
self.assertTrue(osutils.lexists(path),
1556
path + " does not exist")
1558
@symbol_versioning.deprecated_method(symbol_versioning.deprecated_in((2, 4)))
1559
def failIfExists(self, path):
1560
return self.assertPathDoesNotExist(path)
1562
def assertPathDoesNotExist(self, path):
1563
"""Fail if path or paths, which may be abs or relative, exist."""
1564
if not isinstance(path, basestring):
1566
self.assertPathDoesNotExist(p)
1568
self.assertFalse(osutils.lexists(path),
1571
def _capture_deprecation_warnings(self, a_callable, *args, **kwargs):
1572
"""A helper for callDeprecated and applyDeprecated.
1574
:param a_callable: A callable to call.
1575
:param args: The positional arguments for the callable
1576
:param kwargs: The keyword arguments for the callable
1577
:return: A tuple (warnings, result). result is the result of calling
1578
a_callable(``*args``, ``**kwargs``).
1581
def capture_warnings(msg, cls=None, stacklevel=None):
1582
# we've hooked into a deprecation specific callpath,
1583
# only deprecations should getting sent via it.
1584
self.assertEqual(cls, DeprecationWarning)
1585
local_warnings.append(msg)
1586
original_warning_method = symbol_versioning.warn
1587
symbol_versioning.set_warning_method(capture_warnings)
1589
result = a_callable(*args, **kwargs)
1591
symbol_versioning.set_warning_method(original_warning_method)
1592
return (local_warnings, result)
1594
def applyDeprecated(self, deprecation_format, a_callable, *args, **kwargs):
1595
"""Call a deprecated callable without warning the user.
1597
Note that this only captures warnings raised by symbol_versioning.warn,
1598
not other callers that go direct to the warning module.
1600
To test that a deprecated method raises an error, do something like
1601
this (remember that both assertRaises and applyDeprecated delays *args
1602
and **kwargs passing)::
1604
self.assertRaises(errors.ReservedId,
1605
self.applyDeprecated,
1606
deprecated_in((1, 5, 0)),
1610
:param deprecation_format: The deprecation format that the callable
1611
should have been deprecated with. This is the same type as the
1612
parameter to deprecated_method/deprecated_function. If the
1613
callable is not deprecated with this format, an assertion error
1615
:param a_callable: A callable to call. This may be a bound method or
1616
a regular function. It will be called with ``*args`` and
1618
:param args: The positional arguments for the callable
1619
:param kwargs: The keyword arguments for the callable
1620
:return: The result of a_callable(``*args``, ``**kwargs``)
1622
call_warnings, result = self._capture_deprecation_warnings(a_callable,
1624
expected_first_warning = symbol_versioning.deprecation_string(
1625
a_callable, deprecation_format)
1626
if len(call_warnings) == 0:
1627
self.fail("No deprecation warning generated by call to %s" %
1629
self.assertEqual(expected_first_warning, call_warnings[0])
1632
def callCatchWarnings(self, fn, *args, **kw):
1633
"""Call a callable that raises python warnings.
1635
The caller's responsible for examining the returned warnings.
1637
If the callable raises an exception, the exception is not
1638
caught and propagates up to the caller. In that case, the list
1639
of warnings is not available.
1641
:returns: ([warning_object, ...], fn_result)
1643
# XXX: This is not perfect, because it completely overrides the
1644
# warnings filters, and some code may depend on suppressing particular
1645
# warnings. It's the easiest way to insulate ourselves from -Werror,
1646
# though. -- Andrew, 20071062
1648
def _catcher(message, category, filename, lineno, file=None, line=None):
1649
# despite the name, 'message' is normally(?) a Warning subclass
1651
wlist.append(message)
1652
saved_showwarning = warnings.showwarning
1653
saved_filters = warnings.filters
1655
warnings.showwarning = _catcher
1656
warnings.filters = []
1657
result = fn(*args, **kw)
1659
warnings.showwarning = saved_showwarning
1660
warnings.filters = saved_filters
1661
return wlist, result
1663
def callDeprecated(self, expected, callable, *args, **kwargs):
1664
"""Assert that a callable is deprecated in a particular way.
1666
This is a very precise test for unusual requirements. The
1667
applyDeprecated helper function is probably more suited for most tests
1668
as it allows you to simply specify the deprecation format being used
1669
and will ensure that that is issued for the function being called.
1671
Note that this only captures warnings raised by symbol_versioning.warn,
1672
not other callers that go direct to the warning module. To catch
1673
general warnings, use callCatchWarnings.
1675
:param expected: a list of the deprecation warnings expected, in order
1676
:param callable: The callable to call
1677
:param args: The positional arguments for the callable
1678
:param kwargs: The keyword arguments for the callable
1680
call_warnings, result = self._capture_deprecation_warnings(callable,
1682
self.assertEqual(expected, call_warnings)
552
self.fail("%r is an instance of %s rather than %s" % (
553
obj, obj.__class__, kls))
1685
555
def _startLogFile(self):
1686
556
"""Send bzr and test log messages to a temporary file.
1688
558
The file is removed as the test is torn down.
1690
pseudo_log_file = StringIO()
1691
def _get_log_contents_for_weird_testtools_api():
1692
return [pseudo_log_file.getvalue().decode(
1693
"utf-8", "replace").encode("utf-8")]
1694
self.addDetail("log", content.Content(content.ContentType("text",
1695
"plain", {"charset": "utf8"}),
1696
_get_log_contents_for_weird_testtools_api))
1697
self._log_file = pseudo_log_file
1698
self._log_memento = trace.push_log_file(self._log_file)
560
fileno, name = tempfile.mkstemp(suffix='.log', prefix='testbzr')
561
encoder, decoder, stream_reader, stream_writer = codecs.lookup('UTF-8')
562
self._log_file = stream_writer(os.fdopen(fileno, 'w+'))
563
self._log_nonce = bzrlib.trace.enable_test_log(self._log_file)
564
self._log_file_name = name
1699
565
self.addCleanup(self._finishLogFile)
1701
567
def _finishLogFile(self):
1702
568
"""Finished with the log file.
1704
Close the file and delete it.
1706
if trace._trace_file:
1707
# flush the log file, to get all content
1708
trace._trace_file.flush()
1709
trace.pop_log_file(self._log_memento)
1711
def thisFailsStrictLockCheck(self):
1712
"""It is known that this test would fail with -Dstrict_locks.
1714
By default, all tests are run with strict lock checking unless
1715
-Edisable_lock_checks is supplied. However there are some tests which
1716
we know fail strict locks at this point that have not been fixed.
1717
They should call this function to disable the strict checking.
1719
This should be used sparingly, it is much better to fix the locking
1720
issues rather than papering over the problem by calling this function.
1722
debug.debug_flags.discard('strict_locks')
1724
def overrideAttr(self, obj, attr_name, new=_unitialized_attr):
1725
"""Overrides an object attribute restoring it after the test.
1727
:param obj: The object that will be mutated.
1729
:param attr_name: The attribute name we want to preserve/override in
1732
:param new: The optional value we want to set the attribute to.
1734
:returns: The actual attr value.
1736
value = getattr(obj, attr_name)
1737
# The actual value is captured by the call below
1738
self.addCleanup(setattr, obj, attr_name, value)
1739
if new is not _unitialized_attr:
1740
setattr(obj, attr_name, new)
1743
def overrideEnv(self, name, new):
1744
"""Set an environment variable, and reset it after the test.
1746
:param name: The environment variable name.
1748
:param new: The value to set the variable to. If None, the
1749
variable is deleted from the environment.
1751
:returns: The actual variable value.
1753
value = osutils.set_or_unset_env(name, new)
1754
self.addCleanup(osutils.set_or_unset_env, name, value)
570
Read contents into memory, close, and delete.
572
if self._log_file is None:
574
bzrlib.trace.disable_test_log(self._log_nonce)
575
self._log_file.seek(0)
576
self._log_contents = self._log_file.read()
577
self._log_file.close()
578
os.remove(self._log_file_name)
579
self._log_file = self._log_file_name = None
581
def addCleanup(self, callable):
582
"""Arrange to run a callable when this case is torn down.
584
Callables are run in the reverse of the order they are registered,
585
ie last-in first-out.
587
if callable in self._cleanups:
588
raise ValueError("cleanup function %r already registered on %s"
590
self._cleanups.append(callable)
1757
592
def _cleanEnvironment(self):
1758
for name, value in isolated_environ.iteritems():
1759
self.overrideEnv(name, value)
1761
def _restoreHooks(self):
1762
for klass, (name, hooks) in self._preserved_hooks.items():
1763
setattr(klass, name, hooks)
1764
self._preserved_hooks.clear()
1765
bzrlib.hooks._lazy_hooks = self._preserved_lazy_hooks
1766
self._preserved_lazy_hooks.clear()
1768
def knownFailure(self, reason):
1769
"""This test has failed for some known reason."""
1770
raise KnownFailure(reason)
1772
def _suppress_log(self):
1773
"""Remove the log info from details."""
1774
self.discardDetail('log')
1776
def _do_skip(self, result, reason):
1777
self._suppress_log()
1778
addSkip = getattr(result, 'addSkip', None)
1779
if not callable(addSkip):
1780
result.addSuccess(result)
1782
addSkip(self, reason)
1785
def _do_known_failure(self, result, e):
1786
self._suppress_log()
1787
err = sys.exc_info()
1788
addExpectedFailure = getattr(result, 'addExpectedFailure', None)
1789
if addExpectedFailure is not None:
1790
addExpectedFailure(self, err)
1792
result.addSuccess(self)
1795
def _do_not_applicable(self, result, e):
1797
reason = 'No reason given'
1800
self._suppress_log ()
1801
addNotApplicable = getattr(result, 'addNotApplicable', None)
1802
if addNotApplicable is not None:
1803
result.addNotApplicable(self, reason)
1805
self._do_skip(result, reason)
1808
def _report_skip(self, result, err):
1809
"""Override the default _report_skip.
1811
We want to strip the 'log' detail. If we waint until _do_skip, it has
1812
already been formatted into the 'reason' string, and we can't pull it
1815
self._suppress_log()
1816
super(TestCase, self)._report_skip(self, result, err)
1819
def _report_expected_failure(self, result, err):
1822
See _report_skip for motivation.
1824
self._suppress_log()
1825
super(TestCase, self)._report_expected_failure(self, result, err)
1828
def _do_unsupported_or_skip(self, result, e):
1830
self._suppress_log()
1831
addNotSupported = getattr(result, 'addNotSupported', None)
1832
if addNotSupported is not None:
1833
result.addNotSupported(self, reason)
1835
self._do_skip(result, reason)
595
'APPDATA': os.getcwd(),
600
self.addCleanup(self._restoreEnvironment)
601
for name, value in new_env.iteritems():
602
self._captureVar(name, value)
605
def _captureVar(self, name, newvalue):
606
"""Set an environment variable, preparing it to be reset when finished."""
607
self.__old_env[name] = os.environ.get(name, None)
609
if name in os.environ:
612
os.environ[name] = newvalue
615
def _restoreVar(name, value):
617
if name in os.environ:
620
os.environ[name] = value
622
def _restoreEnvironment(self):
623
for name, value in self.__old_env.iteritems():
624
self._restoreVar(name, value)
628
unittest.TestCase.tearDown(self)
1837
630
def time(self, callable, *args, **kwargs):
1838
631
"""Run callable and accrue the time it takes to the benchmark time.
1840
633
If lsprofiling is enabled (i.e. by --lsprof-time to bzr selftest) then
1841
634
this will cause lsprofile statistics to be gathered and stored in
1842
635
self._benchcalls.
1844
637
if self._benchtime is None:
1845
self.addDetail('benchtime', content.Content(content.ContentType(
1846
"text", "plain"), lambda:[str(self._benchtime)]))
1847
638
self._benchtime = 0
1848
639
start = time.time()
2267
863
sys.stderr = real_stderr
2268
864
sys.stdin = real_stdin
2270
def reduceLockdirTimeout(self):
2271
"""Reduce the default lock timeout for the duration of the test, so that
2272
if LockContention occurs during a test, it does so quickly.
2274
Tests that expect to provoke LockContention errors should call this.
2276
self.overrideAttr(lockdir, '_DEFAULT_TIMEOUT_SECONDS', 0)
2278
def make_utf8_encoded_stringio(self, encoding_type=None):
2279
"""Return a StringIOWrapper instance, that will encode Unicode
2282
if encoding_type is None:
2283
encoding_type = 'strict'
2285
output_encoding = 'utf-8'
2286
sio = codecs.getwriter(output_encoding)(sio, errors=encoding_type)
2287
sio.encoding = output_encoding
2290
def disable_verb(self, verb):
2291
"""Disable a smart server verb for one test."""
2292
from bzrlib.smart import request
2293
request_handlers = request.request_handlers
2294
orig_method = request_handlers.get(verb)
2295
request_handlers.remove(verb)
2296
self.addCleanup(request_handlers.register, verb, orig_method)
2299
class CapturedCall(object):
2300
"""A helper for capturing smart server calls for easy debug analysis."""
2302
def __init__(self, params, prefix_length):
2303
"""Capture the call with params and skip prefix_length stack frames."""
2306
# The last 5 frames are the __init__, the hook frame, and 3 smart
2307
# client frames. Beyond this we could get more clever, but this is good
2309
stack = traceback.extract_stack()[prefix_length:-5]
2310
self.stack = ''.join(traceback.format_list(stack))
2313
return self.call.method
2316
return self.call.method
2322
class TestCaseWithMemoryTransport(TestCase):
2323
"""Common test class for tests that do not need disk resources.
2325
Tests that need disk resources should derive from TestCaseInTempDir
2326
orTestCaseWithTransport.
2328
TestCaseWithMemoryTransport sets the TEST_ROOT variable for all bzr tests.
2330
For TestCaseWithMemoryTransport the ``test_home_dir`` is set to the name of
2331
a directory which does not exist. This serves to help ensure test isolation
2332
is preserved. ``test_dir`` is set to the TEST_ROOT, as is cwd, because they
2333
must exist. However, TestCaseWithMemoryTransport does not offer local file
2334
defaults for the transport in tests, nor does it obey the command line
2335
override, so tests that accidentally write to the common directory should
2338
:cvar TEST_ROOT: Directory containing all temporary directories, plus a
2339
``.bzr`` directory that stops us ascending higher into the filesystem.
2345
def __init__(self, methodName='runTest'):
2346
# allow test parameterization after test construction and before test
2347
# execution. Variables that the parameterizer sets need to be
2348
# ones that are not set by setUp, or setUp will trash them.
2349
super(TestCaseWithMemoryTransport, self).__init__(methodName)
2350
self.vfs_transport_factory = default_transport
2351
self.transport_server = None
2352
self.transport_readonly_server = None
2353
self.__vfs_server = None
2355
def get_transport(self, relpath=None):
2356
"""Return a writeable transport.
2358
This transport is for the test scratch space relative to
2361
:param relpath: a path relative to the base url.
2363
t = _mod_transport.get_transport(self.get_url(relpath))
2364
self.assertFalse(t.is_readonly())
2367
def get_readonly_transport(self, relpath=None):
2368
"""Return a readonly transport for the test scratch space
2370
This can be used to test that operations which should only need
2371
readonly access in fact do not try to write.
2373
:param relpath: a path relative to the base url.
2375
t = _mod_transport.get_transport(self.get_readonly_url(relpath))
2376
self.assertTrue(t.is_readonly())
2379
def create_transport_readonly_server(self):
2380
"""Create a transport server from class defined at init.
2382
This is mostly a hook for daughter classes.
2384
return self.transport_readonly_server()
2386
def get_readonly_server(self):
2387
"""Get the server instance for the readonly transport
2389
This is useful for some tests with specific servers to do diagnostics.
2391
if self.__readonly_server is None:
2392
if self.transport_readonly_server is None:
2393
# readonly decorator requested
2394
self.__readonly_server = test_server.ReadonlyServer()
2396
# explicit readonly transport.
2397
self.__readonly_server = self.create_transport_readonly_server()
2398
self.start_server(self.__readonly_server,
2399
self.get_vfs_only_server())
2400
return self.__readonly_server
2402
def get_readonly_url(self, relpath=None):
2403
"""Get a URL for the readonly transport.
2405
This will either be backed by '.' or a decorator to the transport
2406
used by self.get_url()
2407
relpath provides for clients to get a path relative to the base url.
2408
These should only be downwards relative, not upwards.
2410
base = self.get_readonly_server().get_url()
2411
return self._adjust_url(base, relpath)
2413
def get_vfs_only_server(self):
2414
"""Get the vfs only read/write server instance.
2416
This is useful for some tests with specific servers that need
2419
For TestCaseWithMemoryTransport this is always a MemoryServer, and there
2420
is no means to override it.
2422
if self.__vfs_server is None:
2423
self.__vfs_server = memory.MemoryServer()
2424
self.start_server(self.__vfs_server)
2425
return self.__vfs_server
2427
def get_server(self):
2428
"""Get the read/write server instance.
2430
This is useful for some tests with specific servers that need
2433
This is built from the self.transport_server factory. If that is None,
2434
then the self.get_vfs_server is returned.
2436
if self.__server is None:
2437
if (self.transport_server is None or self.transport_server is
2438
self.vfs_transport_factory):
2439
self.__server = self.get_vfs_only_server()
2441
# bring up a decorated means of access to the vfs only server.
2442
self.__server = self.transport_server()
2443
self.start_server(self.__server, self.get_vfs_only_server())
2444
return self.__server
2446
def _adjust_url(self, base, relpath):
2447
"""Get a URL (or maybe a path) for the readwrite transport.
2449
This will either be backed by '.' or to an equivalent non-file based
2451
relpath provides for clients to get a path relative to the base url.
2452
These should only be downwards relative, not upwards.
2454
if relpath is not None and relpath != '.':
2455
if not base.endswith('/'):
2457
# XXX: Really base should be a url; we did after all call
2458
# get_url()! But sometimes it's just a path (from
2459
# LocalAbspathServer), and it'd be wrong to append urlescaped data
2460
# to a non-escaped local path.
2461
if base.startswith('./') or base.startswith('/'):
2464
base += urlutils.escape(relpath)
2467
def get_url(self, relpath=None):
2468
"""Get a URL (or maybe a path) for the readwrite transport.
2470
This will either be backed by '.' or to an equivalent non-file based
2472
relpath provides for clients to get a path relative to the base url.
2473
These should only be downwards relative, not upwards.
2475
base = self.get_server().get_url()
2476
return self._adjust_url(base, relpath)
2478
def get_vfs_only_url(self, relpath=None):
2479
"""Get a URL (or maybe a path for the plain old vfs transport.
2481
This will never be a smart protocol. It always has all the
2482
capabilities of the local filesystem, but it might actually be a
2483
MemoryTransport or some other similar virtual filesystem.
2485
This is the backing transport (if any) of the server returned by
2486
get_url and get_readonly_url.
2488
:param relpath: provides for clients to get a path relative to the base
2489
url. These should only be downwards relative, not upwards.
2492
base = self.get_vfs_only_server().get_url()
2493
return self._adjust_url(base, relpath)
2495
def _create_safety_net(self):
2496
"""Make a fake bzr directory.
2498
This prevents any tests propagating up onto the TEST_ROOT directory's
2501
root = TestCaseWithMemoryTransport.TEST_ROOT
2502
wt = bzrdir.BzrDir.create_standalone_workingtree(root)
2503
# Hack for speed: remember the raw bytes of the dirstate file so that
2504
# we don't need to re-open the wt to check it hasn't changed.
2505
TestCaseWithMemoryTransport._SAFETY_NET_PRISTINE_DIRSTATE = (
2506
wt.control_transport.get_bytes('dirstate'))
2508
def _check_safety_net(self):
2509
"""Check that the safety .bzr directory have not been touched.
2511
_make_test_root have created a .bzr directory to prevent tests from
2512
propagating. This method ensures than a test did not leaked.
2514
root = TestCaseWithMemoryTransport.TEST_ROOT
2515
t = _mod_transport.get_transport(root)
2516
self.permit_url(t.base)
2517
if (t.get_bytes('.bzr/checkout/dirstate') !=
2518
TestCaseWithMemoryTransport._SAFETY_NET_PRISTINE_DIRSTATE):
2519
# The current test have modified the /bzr directory, we need to
2520
# recreate a new one or all the followng tests will fail.
2521
# If you need to inspect its content uncomment the following line
2522
# import pdb; pdb.set_trace()
2523
_rmtree_temp_dir(root + '/.bzr', test_id=self.id())
2524
self._create_safety_net()
2525
raise AssertionError('%s/.bzr should not be modified' % root)
2527
def _make_test_root(self):
2528
if TestCaseWithMemoryTransport.TEST_ROOT is None:
2529
# Watch out for tricky test dir (on OSX /tmp -> /private/tmp)
2530
root = osutils.realpath(osutils.mkdtemp(prefix='testbzr-',
2532
TestCaseWithMemoryTransport.TEST_ROOT = root
2534
self._create_safety_net()
2536
# The same directory is used by all tests, and we're not
2537
# specifically told when all tests are finished. This will do.
2538
atexit.register(_rmtree_temp_dir, root)
2540
self.permit_dir(TestCaseWithMemoryTransport.TEST_ROOT)
2541
self.addCleanup(self._check_safety_net)
2543
def makeAndChdirToTestDir(self):
2544
"""Create a temporary directories for this one test.
2546
This must set self.test_home_dir and self.test_dir and chdir to
2549
For TestCaseWithMemoryTransport we chdir to the TEST_ROOT for this test.
2551
os.chdir(TestCaseWithMemoryTransport.TEST_ROOT)
2552
self.test_dir = TestCaseWithMemoryTransport.TEST_ROOT
2553
self.test_home_dir = self.test_dir + "/MemoryTransportMissingHomeDir"
2554
self.permit_dir(self.test_dir)
2556
def make_branch(self, relpath, format=None):
2557
"""Create a branch on the transport at relpath."""
2558
repo = self.make_repository(relpath, format=format)
2559
return repo.bzrdir.create_branch()
2561
def make_bzrdir(self, relpath, format=None):
2563
# might be a relative or absolute path
2564
maybe_a_url = self.get_url(relpath)
2565
segments = maybe_a_url.rsplit('/', 1)
2566
t = _mod_transport.get_transport(maybe_a_url)
2567
if len(segments) > 1 and segments[-1] not in ('', '.'):
2571
if isinstance(format, basestring):
2572
format = bzrdir.format_registry.make_bzrdir(format)
2573
return format.initialize_on_transport(t)
2574
except errors.UninitializableFormat:
2575
raise TestSkipped("Format %s is not initializable." % format)
2577
def make_repository(self, relpath, shared=False, format=None):
2578
"""Create a repository on our default transport at relpath.
2580
Note that relpath must be a relative path, not a full url.
2582
# FIXME: If you create a remoterepository this returns the underlying
2583
# real format, which is incorrect. Actually we should make sure that
2584
# RemoteBzrDir returns a RemoteRepository.
2585
# maybe mbp 20070410
2586
made_control = self.make_bzrdir(relpath, format=format)
2587
return made_control.create_repository(shared=shared)
2589
def make_smart_server(self, path, backing_server=None):
2590
if backing_server is None:
2591
backing_server = self.get_server()
2592
smart_server = test_server.SmartTCPServer_for_testing()
2593
self.start_server(smart_server, backing_server)
2594
remote_transport = _mod_transport.get_transport(smart_server.get_url()
2596
return remote_transport
2598
def make_branch_and_memory_tree(self, relpath, format=None):
2599
"""Create a branch on the default transport and a MemoryTree for it."""
2600
b = self.make_branch(relpath, format=format)
2601
return memorytree.MemoryTree.create_on_branch(b)
2603
def make_branch_builder(self, relpath, format=None):
2604
branch = self.make_branch(relpath, format=format)
2605
return branchbuilder.BranchBuilder(branch=branch)
2607
def overrideEnvironmentForTesting(self):
2608
test_home_dir = self.test_home_dir
2609
if isinstance(test_home_dir, unicode):
2610
test_home_dir = test_home_dir.encode(sys.getfilesystemencoding())
2611
self.overrideEnv('HOME', test_home_dir)
2612
self.overrideEnv('BZR_HOME', test_home_dir)
2615
super(TestCaseWithMemoryTransport, self).setUp()
2616
# Ensure that ConnectedTransport doesn't leak sockets
2617
def get_transport_with_cleanup(*args, **kwargs):
2618
t = orig_get_transport(*args, **kwargs)
2619
if isinstance(t, _mod_transport.ConnectedTransport):
2620
self.addCleanup(t.disconnect)
2623
orig_get_transport = self.overrideAttr(_mod_transport, 'get_transport',
2624
get_transport_with_cleanup)
2625
self._make_test_root()
2626
self.addCleanup(os.chdir, os.getcwdu())
2627
self.makeAndChdirToTestDir()
2628
self.overrideEnvironmentForTesting()
2629
self.__readonly_server = None
2630
self.__server = None
2631
self.reduceLockdirTimeout()
2633
def setup_smart_server_with_call_log(self):
2634
"""Sets up a smart server as the transport server with a call log."""
2635
self.transport_server = test_server.SmartTCPServer_for_testing
2636
self.hpss_calls = []
2638
# Skip the current stack down to the caller of
2639
# setup_smart_server_with_call_log
2640
prefix_length = len(traceback.extract_stack()) - 2
2641
def capture_hpss_call(params):
2642
self.hpss_calls.append(
2643
CapturedCall(params, prefix_length))
2644
client._SmartClient.hooks.install_named_hook(
2645
'call', capture_hpss_call, None)
2647
def reset_smart_call_log(self):
2648
self.hpss_calls = []
2651
class TestCaseInTempDir(TestCaseWithMemoryTransport):
866
def merge(self, branch_from, wt_to):
867
"""A helper for tests to do a ui-less merge.
869
This should move to the main library when someone has time to integrate
872
# minimal ui-less merge.
873
wt_to.branch.fetch(branch_from)
874
base_rev = common_ancestor(branch_from.last_revision(),
875
wt_to.branch.last_revision(),
876
wt_to.branch.repository)
877
merge_inner(wt_to.branch, branch_from.basis_tree(),
878
wt_to.branch.repository.revision_tree(base_rev),
880
wt_to.add_pending_merge(branch_from.last_revision())
883
BzrTestBase = TestCase
886
class TestCaseInTempDir(TestCase):
2652
887
"""Derived class that runs a test within a temporary directory.
2654
889
This is useful for tests that need to create a branch, etc.
2920
1178
for readonly urls.
2922
1180
TODO RBC 20060127: make this an option to TestCaseWithTransport so it can
2923
be used without needed to redo it when a different
1181
be used without needed to redo it when a different
2924
1182
subclass is in use ?
2927
1185
def setUp(self):
2928
from bzrlib.tests import http_server
2929
1186
super(ChrootedTestCase, self).setUp()
2930
if not self.vfs_transport_factory == memory.MemoryServer:
2931
self.transport_readonly_server = http_server.HttpServer
2934
def condition_id_re(pattern):
2935
"""Create a condition filter which performs a re check on a test's id.
2937
:param pattern: A regular expression string.
2938
:return: A callable that returns True if the re matches.
2940
filter_re = re.compile(pattern, 0)
2941
def condition(test):
2943
return filter_re.search(test_id)
2947
def condition_isinstance(klass_or_klass_list):
2948
"""Create a condition filter which returns isinstance(param, klass).
2950
:return: A callable which when called with one parameter obj return the
2951
result of isinstance(obj, klass_or_klass_list).
2954
return isinstance(obj, klass_or_klass_list)
2958
def condition_id_in_list(id_list):
2959
"""Create a condition filter which verify that test's id in a list.
2961
:param id_list: A TestIdList object.
2962
:return: A callable that returns True if the test's id appears in the list.
2964
def condition(test):
2965
return id_list.includes(test.id())
2969
def condition_id_startswith(starts):
2970
"""Create a condition filter verifying that test's id starts with a string.
2972
:param starts: A list of string.
2973
:return: A callable that returns True if the test's id starts with one of
2976
def condition(test):
2977
for start in starts:
2978
if test.id().startswith(start):
2984
def exclude_tests_by_condition(suite, condition):
2985
"""Create a test suite which excludes some tests from suite.
2987
:param suite: The suite to get tests from.
2988
:param condition: A callable whose result evaluates True when called with a
2989
test case which should be excluded from the result.
2990
:return: A suite which contains the tests found in suite that fail
2994
for test in iter_suite_tests(suite):
2995
if not condition(test):
2997
return TestUtil.TestSuite(result)
3000
def filter_suite_by_condition(suite, condition):
3001
"""Create a test suite by filtering another one.
3003
:param suite: The source suite.
3004
:param condition: A callable whose result evaluates True when called with a
3005
test case which should be included in the result.
3006
:return: A suite which contains the tests found in suite that pass
3010
for test in iter_suite_tests(suite):
3013
return TestUtil.TestSuite(result)
1187
if not self.transport_server == bzrlib.transport.memory.MemoryServer:
1188
self.transport_readonly_server = bzrlib.transport.http.HttpServer
3016
1191
def filter_suite_by_re(suite, pattern):
3017
"""Create a test suite by filtering another one.
3019
:param suite: the source suite
3020
:param pattern: pattern that names must match
3021
:returns: the newly created suite
3023
condition = condition_id_re(pattern)
3024
result_suite = filter_suite_by_condition(suite, condition)
3028
def filter_suite_by_id_list(suite, test_id_list):
3029
"""Create a test suite by filtering another one.
3031
:param suite: The source suite.
3032
:param test_id_list: A list of the test ids to keep as strings.
3033
:returns: the newly created suite
3035
condition = condition_id_in_list(test_id_list)
3036
result_suite = filter_suite_by_condition(suite, condition)
3040
def filter_suite_by_id_startswith(suite, start):
3041
"""Create a test suite by filtering another one.
3043
:param suite: The source suite.
3044
:param start: A list of string the test id must start with one of.
3045
:returns: the newly created suite
3047
condition = condition_id_startswith(start)
3048
result_suite = filter_suite_by_condition(suite, condition)
3052
def exclude_tests_by_re(suite, pattern):
3053
"""Create a test suite which excludes some tests from suite.
3055
:param suite: The suite to get tests from.
3056
:param pattern: A regular expression string. Test ids that match this
3057
pattern will be excluded from the result.
3058
:return: A TestSuite that contains all the tests from suite without the
3059
tests that matched pattern. The order of tests is the same as it was in
3062
return exclude_tests_by_condition(suite, condition_id_re(pattern))
3065
def preserve_input(something):
3066
"""A helper for performing test suite transformation chains.
3068
:param something: Anything you want to preserve.
3074
def randomize_suite(suite):
3075
"""Return a new TestSuite with suite's tests in random order.
3077
The tests in the input suite are flattened into a single suite in order to
3078
accomplish this. Any nested TestSuites are removed to provide global
3081
tests = list(iter_suite_tests(suite))
3082
random.shuffle(tests)
3083
return TestUtil.TestSuite(tests)
3086
def split_suite_by_condition(suite, condition):
3087
"""Split a test suite into two by a condition.
3089
:param suite: The suite to split.
3090
:param condition: The condition to match on. Tests that match this
3091
condition are returned in the first test suite, ones that do not match
3092
are in the second suite.
3093
:return: A tuple of two test suites, where the first contains tests from
3094
suite matching the condition, and the second contains the remainder
3095
from suite. The order within each output suite is the same as it was in
1192
result = TestUtil.TestSuite()
1193
filter_re = re.compile(pattern)
3100
1194
for test in iter_suite_tests(suite):
3102
matched.append(test)
3104
did_not_match.append(test)
3105
return TestUtil.TestSuite(matched), TestUtil.TestSuite(did_not_match)
3108
def split_suite_by_re(suite, pattern):
3109
"""Split a test suite into two by a regular expression.
3111
:param suite: The suite to split.
3112
:param pattern: A regular expression string. Test ids that match this
3113
pattern will be in the first test suite returned, and the others in the
3114
second test suite returned.
3115
:return: A tuple of two test suites, where the first contains tests from
3116
suite matching pattern, and the second contains the remainder from
3117
suite. The order within each output suite is the same as it was in
3120
return split_suite_by_condition(suite, condition_id_re(pattern))
1195
if filter_re.search(test.id()):
1196
result.addTest(test)
3123
1200
def run_suite(suite, name='test', verbose=False, pattern=".*",
3124
stop_on_failure=False,
3125
transport=None, lsprof_timed=None, bench_history=None,
3126
matching_tests_first=None,
3129
exclude_pattern=None,
3132
suite_decorators=None,
3134
result_decorators=None,
3136
"""Run a test suite for bzr selftest.
3138
:param runner_class: The class of runner to use. Must support the
3139
constructor arguments passed by run_suite which are more than standard
3141
:return: A boolean indicating success.
1201
stop_on_failure=False, keep_output=False,
1202
transport=None, lsprof_timed=None):
1203
TestCaseInTempDir._TEST_NAME = name
3143
1204
TestCase._gather_lsprof_in_benchmarks = lsprof_timed
3148
if runner_class is None:
3149
runner_class = TextTestRunner
3152
runner = runner_class(stream=stream,
1210
pb = progress.ProgressBar()
1211
runner = TextTestRunner(stream=sys.stdout,
3153
1212
descriptions=0,
3154
1213
verbosity=verbosity,
3155
bench_history=bench_history,
3157
result_decorators=result_decorators,
1214
keep_output=keep_output,
3159
1216
runner.stop_on_failure=stop_on_failure
3160
# built in decorator factories:
3162
random_order(random_seed, runner),
3163
exclude_tests(exclude_pattern),
3165
if matching_tests_first:
3166
decorators.append(tests_first(pattern))
3168
decorators.append(filter_tests(pattern))
3169
if suite_decorators:
3170
decorators.extend(suite_decorators)
3171
# tell the result object how many tests will be running: (except if
3172
# --parallel=fork is being used. Robert said he will provide a better
3173
# progress design later -- vila 20090817)
3174
if fork_decorator not in decorators:
3175
decorators.append(CountingDecorator)
3176
for decorator in decorators:
3177
suite = decorator(suite)
3179
# Done after test suite decoration to allow randomisation etc
3180
# to take effect, though that is of marginal benefit.
3182
stream.write("Listing tests only ...\n")
3183
for t in iter_suite_tests(suite):
3184
stream.write("%s\n" % (t.id()))
1218
suite = filter_suite_by_re(suite, pattern)
3186
1219
result = runner.run(suite)
3188
return result.wasStrictlySuccessful()
3190
return result.wasSuccessful()
3193
# A registry where get() returns a suite decorator.
3194
parallel_registry = registry.Registry()
3197
def fork_decorator(suite):
3198
if getattr(os, "fork", None) is None:
3199
raise errors.BzrCommandError("platform does not support fork,"
3200
" try --parallel=subprocess instead.")
3201
concurrency = osutils.local_concurrency()
3202
if concurrency == 1:
3204
from testtools import ConcurrentTestSuite
3205
return ConcurrentTestSuite(suite, fork_for_tests)
3206
parallel_registry.register('fork', fork_decorator)
3209
def subprocess_decorator(suite):
3210
concurrency = osutils.local_concurrency()
3211
if concurrency == 1:
3213
from testtools import ConcurrentTestSuite
3214
return ConcurrentTestSuite(suite, reinvoke_for_tests)
3215
parallel_registry.register('subprocess', subprocess_decorator)
3218
def exclude_tests(exclude_pattern):
3219
"""Return a test suite decorator that excludes tests."""
3220
if exclude_pattern is None:
3221
return identity_decorator
3222
def decorator(suite):
3223
return ExcludeDecorator(suite, exclude_pattern)
3227
def filter_tests(pattern):
3229
return identity_decorator
3230
def decorator(suite):
3231
return FilterTestsDecorator(suite, pattern)
3235
def random_order(random_seed, runner):
3236
"""Return a test suite decorator factory for randomising tests order.
3238
:param random_seed: now, a string which casts to a long, or a long.
3239
:param runner: A test runner with a stream attribute to report on.
3241
if random_seed is None:
3242
return identity_decorator
3243
def decorator(suite):
3244
return RandomDecorator(suite, random_seed, runner.stream)
3248
def tests_first(pattern):
3250
return identity_decorator
3251
def decorator(suite):
3252
return TestFirstDecorator(suite, pattern)
3256
def identity_decorator(suite):
3261
class TestDecorator(TestUtil.TestSuite):
3262
"""A decorator for TestCase/TestSuite objects.
3264
Usually, subclasses should override __iter__(used when flattening test
3265
suites), which we do to filter, reorder, parallelise and so on, run() and
3269
def __init__(self, suite):
3270
TestUtil.TestSuite.__init__(self)
3273
def countTestCases(self):
3276
cases += test.countTestCases()
3283
def run(self, result):
3284
# Use iteration on self, not self._tests, to allow subclasses to hook
3287
if result.shouldStop:
3293
class CountingDecorator(TestDecorator):
3294
"""A decorator which calls result.progress(self.countTestCases)."""
3296
def run(self, result):
3297
progress_method = getattr(result, 'progress', None)
3298
if callable(progress_method):
3299
progress_method(self.countTestCases(), SUBUNIT_SEEK_SET)
3300
return super(CountingDecorator, self).run(result)
3303
class ExcludeDecorator(TestDecorator):
3304
"""A decorator which excludes test matching an exclude pattern."""
3306
def __init__(self, suite, exclude_pattern):
3307
TestDecorator.__init__(self, suite)
3308
self.exclude_pattern = exclude_pattern
3309
self.excluded = False
3313
return iter(self._tests)
3314
self.excluded = True
3315
suite = exclude_tests_by_re(self, self.exclude_pattern)
3317
self.addTests(suite)
3318
return iter(self._tests)
3321
class FilterTestsDecorator(TestDecorator):
3322
"""A decorator which filters tests to those matching a pattern."""
3324
def __init__(self, suite, pattern):
3325
TestDecorator.__init__(self, suite)
3326
self.pattern = pattern
3327
self.filtered = False
3331
return iter(self._tests)
3332
self.filtered = True
3333
suite = filter_suite_by_re(self, self.pattern)
3335
self.addTests(suite)
3336
return iter(self._tests)
3339
class RandomDecorator(TestDecorator):
3340
"""A decorator which randomises the order of its tests."""
3342
def __init__(self, suite, random_seed, stream):
3343
TestDecorator.__init__(self, suite)
3344
self.random_seed = random_seed
3345
self.randomised = False
3346
self.stream = stream
3350
return iter(self._tests)
3351
self.randomised = True
3352
self.stream.write("Randomizing test order using seed %s\n\n" %
3353
(self.actual_seed()))
3354
# Initialise the random number generator.
3355
random.seed(self.actual_seed())
3356
suite = randomize_suite(self)
3358
self.addTests(suite)
3359
return iter(self._tests)
3361
def actual_seed(self):
3362
if self.random_seed == "now":
3363
# We convert the seed to a long to make it reuseable across
3364
# invocations (because the user can reenter it).
3365
self.random_seed = long(time.time())
3367
# Convert the seed to a long if we can
3369
self.random_seed = long(self.random_seed)
3372
return self.random_seed
3375
class TestFirstDecorator(TestDecorator):
3376
"""A decorator which moves named tests to the front."""
3378
def __init__(self, suite, pattern):
3379
TestDecorator.__init__(self, suite)
3380
self.pattern = pattern
3381
self.filtered = False
3385
return iter(self._tests)
3386
self.filtered = True
3387
suites = split_suite_by_re(self, self.pattern)
3389
self.addTests(suites)
3390
return iter(self._tests)
3393
def partition_tests(suite, count):
3394
"""Partition suite into count lists of tests."""
3395
# This just assigns tests in a round-robin fashion. On one hand this
3396
# splits up blocks of related tests that might run faster if they shared
3397
# resources, but on the other it avoids assigning blocks of slow tests to
3398
# just one partition. So the slowest partition shouldn't be much slower
3400
partitions = [list() for i in range(count)]
3401
tests = iter_suite_tests(suite)
3402
for partition, test in itertools.izip(itertools.cycle(partitions), tests):
3403
partition.append(test)
3407
def workaround_zealous_crypto_random():
3408
"""Crypto.Random want to help us being secure, but we don't care here.
3410
This workaround some test failure related to the sftp server. Once paramiko
3411
stop using the controversial API in Crypto.Random, we may get rid of it.
3414
from Crypto.Random import atfork
3420
def fork_for_tests(suite):
3421
"""Take suite and start up one runner per CPU by forking()
3423
:return: An iterable of TestCase-like objects which can each have
3424
run(result) called on them to feed tests to result.
3426
concurrency = osutils.local_concurrency()
3428
from subunit import TestProtocolClient, ProtocolTestCase
3429
from subunit.test_results import AutoTimingTestResultDecorator
3430
class TestInOtherProcess(ProtocolTestCase):
3431
# Should be in subunit, I think. RBC.
3432
def __init__(self, stream, pid):
3433
ProtocolTestCase.__init__(self, stream)
3436
def run(self, result):
3438
ProtocolTestCase.run(self, result)
3440
os.waitpid(self.pid, 0)
3442
test_blocks = partition_tests(suite, concurrency)
3443
for process_tests in test_blocks:
3444
process_suite = TestUtil.TestSuite()
3445
process_suite.addTests(process_tests)
3446
c2pread, c2pwrite = os.pipe()
3449
workaround_zealous_crypto_random()
3452
# Leave stderr and stdout open so we can see test noise
3453
# Close stdin so that the child goes away if it decides to
3454
# read from stdin (otherwise its a roulette to see what
3455
# child actually gets keystrokes for pdb etc).
3458
stream = os.fdopen(c2pwrite, 'wb', 1)
3459
subunit_result = AutoTimingTestResultDecorator(
3460
TestProtocolClient(stream))
3461
process_suite.run(subunit_result)
3466
stream = os.fdopen(c2pread, 'rb', 1)
3467
test = TestInOtherProcess(stream, pid)
3472
def reinvoke_for_tests(suite):
3473
"""Take suite and start up one runner per CPU using subprocess().
3475
:return: An iterable of TestCase-like objects which can each have
3476
run(result) called on them to feed tests to result.
3478
concurrency = osutils.local_concurrency()
3480
from subunit import ProtocolTestCase
3481
class TestInSubprocess(ProtocolTestCase):
3482
def __init__(self, process, name):
3483
ProtocolTestCase.__init__(self, process.stdout)
3484
self.process = process
3485
self.process.stdin.close()
3488
def run(self, result):
3490
ProtocolTestCase.run(self, result)
3493
os.unlink(self.name)
3494
# print "pid %d finished" % finished_process
3495
test_blocks = partition_tests(suite, concurrency)
3496
for process_tests in test_blocks:
3497
# ugly; currently reimplement rather than reuses TestCase methods.
3498
bzr_path = os.path.dirname(os.path.dirname(bzrlib.__file__))+'/bzr'
3499
if not os.path.isfile(bzr_path):
3500
# We are probably installed. Assume sys.argv is the right file
3501
bzr_path = sys.argv[0]
3502
bzr_path = [bzr_path]
3503
if sys.platform == "win32":
3504
# if we're on windows, we can't execute the bzr script directly
3505
bzr_path = [sys.executable] + bzr_path
3506
fd, test_list_file_name = tempfile.mkstemp()
3507
test_list_file = os.fdopen(fd, 'wb', 1)
3508
for test in process_tests:
3509
test_list_file.write(test.id() + '\n')
3510
test_list_file.close()
3512
argv = bzr_path + ['selftest', '--load-list', test_list_file_name,
3514
if '--no-plugins' in sys.argv:
3515
argv.append('--no-plugins')
3516
# stderr=subprocess.STDOUT would be ideal, but until we prevent
3517
# noise on stderr it can interrupt the subunit protocol.
3518
process = subprocess.Popen(argv, stdin=subprocess.PIPE,
3519
stdout=subprocess.PIPE,
3520
stderr=subprocess.PIPE,
3522
test = TestInSubprocess(process, test_list_file_name)
3525
os.unlink(test_list_file_name)
3530
class ProfileResult(testtools.ExtendedToOriginalDecorator):
3531
"""Generate profiling data for all activity between start and success.
3533
The profile data is appended to the test's _benchcalls attribute and can
3534
be accessed by the forwarded-to TestResult.
3536
While it might be cleaner do accumulate this in stopTest, addSuccess is
3537
where our existing output support for lsprof is, and this class aims to
3538
fit in with that: while it could be moved it's not necessary to accomplish
3539
test profiling, nor would it be dramatically cleaner.
3542
def startTest(self, test):
3543
self.profiler = bzrlib.lsprof.BzrProfiler()
3544
# Prevent deadlocks in tests that use lsprof: those tests will
3546
bzrlib.lsprof.BzrProfiler.profiler_block = 0
3547
self.profiler.start()
3548
testtools.ExtendedToOriginalDecorator.startTest(self, test)
3550
def addSuccess(self, test):
3551
stats = self.profiler.stop()
3553
calls = test._benchcalls
3554
except AttributeError:
3555
test._benchcalls = []
3556
calls = test._benchcalls
3557
calls.append(((test.id(), "", ""), stats))
3558
testtools.ExtendedToOriginalDecorator.addSuccess(self, test)
3560
def stopTest(self, test):
3561
testtools.ExtendedToOriginalDecorator.stopTest(self, test)
3562
self.profiler = None
3565
# Controlled by "bzr selftest -E=..." option
3566
# Currently supported:
3567
# -Eallow_debug Will no longer clear debug.debug_flags() so it
3568
# preserves any flags supplied at the command line.
3569
# -Edisable_lock_checks Turns errors in mismatched locks into simple prints
3570
# rather than failing tests. And no longer raise
3571
# LockContention when fctnl locks are not being used
3572
# with proper exclusion rules.
3573
# -Ethreads Will display thread ident at creation/join time to
3574
# help track thread leaks
3576
# -Econfig_stats Will collect statistics using addDetail
3577
selftest_debug_flags = set()
1220
return result.wasSuccessful()
3580
1223
def selftest(verbose=False, pattern=".*", stop_on_failure=True,
3581
1225
transport=None,
3582
1226
test_suite_factory=None,
3585
matching_tests_first=None,
3588
exclude_pattern=None,
3594
suite_decorators=None,
3598
1228
"""Run the whole test suite under the enhanced runner"""
3599
# XXX: Very ugly way to do this...
3600
# Disable warning about old formats because we don't want it to disturb
3601
# any blackbox tests.
3602
from bzrlib import repository
3603
repository._deprecation_warning_done = True
3605
1229
global default_transport
3606
1230
if transport is None:
3607
1231
transport = default_transport
3608
1232
old_transport = default_transport
3609
1233
default_transport = transport
3610
global selftest_debug_flags
3611
old_debug_flags = selftest_debug_flags
3612
if debug_flags is not None:
3613
selftest_debug_flags = set(debug_flags)
3615
if load_list is None:
3618
keep_only = load_test_id_list(load_list)
3620
starting_with = [test_prefix_alias_registry.resolve_alias(start)
3621
for start in starting_with]
3622
1235
if test_suite_factory is None:
3623
# Reduce loading time by loading modules based on the starting_with
3625
suite = test_suite(keep_only, starting_with)
1236
suite = test_suite()
3627
1238
suite = test_suite_factory()
3629
# But always filter as requested.
3630
suite = filter_suite_by_id_startswith(suite, starting_with)
3631
result_decorators = []
3633
result_decorators.append(ProfileResult)
3634
1239
return run_suite(suite, 'testbzr', verbose=verbose, pattern=pattern,
3635
stop_on_failure=stop_on_failure,
1240
stop_on_failure=stop_on_failure, keep_output=keep_output,
3636
1241
transport=transport,
3637
lsprof_timed=lsprof_timed,
3638
bench_history=bench_history,
3639
matching_tests_first=matching_tests_first,
3640
list_only=list_only,
3641
random_seed=random_seed,
3642
exclude_pattern=exclude_pattern,
3644
runner_class=runner_class,
3645
suite_decorators=suite_decorators,
3647
result_decorators=result_decorators,
1242
lsprof_timed=lsprof_timed)
3650
1244
default_transport = old_transport
3651
selftest_debug_flags = old_debug_flags
3654
def load_test_id_list(file_name):
3655
"""Load a test id list from a text file.
3657
The format is one test id by line. No special care is taken to impose
3658
strict rules, these test ids are used to filter the test suite so a test id
3659
that do not match an existing test will do no harm. This allows user to add
3660
comments, leave blank lines, etc.
3664
ftest = open(file_name, 'rt')
3666
if e.errno != errno.ENOENT:
3669
raise errors.NoSuchFile(file_name)
3671
for test_name in ftest.readlines():
3672
test_list.append(test_name.strip())
3677
def suite_matches_id_list(test_suite, id_list):
3678
"""Warns about tests not appearing or appearing more than once.
3680
:param test_suite: A TestSuite object.
3681
:param test_id_list: The list of test ids that should be found in
3684
:return: (absents, duplicates) absents is a list containing the test found
3685
in id_list but not in test_suite, duplicates is a list containing the
3686
test found multiple times in test_suite.
3688
When using a prefined test id list, it may occurs that some tests do not
3689
exist anymore or that some tests use the same id. This function warns the
3690
tester about potential problems in his workflow (test lists are volatile)
3691
or in the test suite itself (using the same id for several tests does not
3692
help to localize defects).
3694
# Build a dict counting id occurrences
3696
for test in iter_suite_tests(test_suite):
3698
tests[id] = tests.get(id, 0) + 1
3703
occurs = tests.get(id, 0)
3705
not_found.append(id)
3707
duplicates.append(id)
3709
return not_found, duplicates
3712
class TestIdList(object):
3713
"""Test id list to filter a test suite.
3715
Relying on the assumption that test ids are built as:
3716
<module>[.<class>.<method>][(<param>+)], <module> being in python dotted
3717
notation, this class offers methods to :
3718
- avoid building a test suite for modules not refered to in the test list,
3719
- keep only the tests listed from the module test suite.
3722
def __init__(self, test_id_list):
3723
# When a test suite needs to be filtered against us we compare test ids
3724
# for equality, so a simple dict offers a quick and simple solution.
3725
self.tests = dict().fromkeys(test_id_list, True)
3727
# While unittest.TestCase have ids like:
3728
# <module>.<class>.<method>[(<param+)],
3729
# doctest.DocTestCase can have ids like:
3732
# <module>.<function>
3733
# <module>.<class>.<method>
3735
# Since we can't predict a test class from its name only, we settle on
3736
# a simple constraint: a test id always begins with its module name.
3739
for test_id in test_id_list:
3740
parts = test_id.split('.')
3741
mod_name = parts.pop(0)
3742
modules[mod_name] = True
3744
mod_name += '.' + part
3745
modules[mod_name] = True
3746
self.modules = modules
3748
def refers_to(self, module_name):
3749
"""Is there tests for the module or one of its sub modules."""
3750
return self.modules.has_key(module_name)
3752
def includes(self, test_id):
3753
return self.tests.has_key(test_id)
3756
class TestPrefixAliasRegistry(registry.Registry):
3757
"""A registry for test prefix aliases.
3759
This helps implement shorcuts for the --starting-with selftest
3760
option. Overriding existing prefixes is not allowed but not fatal (a
3761
warning will be emitted).
3764
def register(self, key, obj, help=None, info=None,
3765
override_existing=False):
3766
"""See Registry.register.
3768
Trying to override an existing alias causes a warning to be emitted,
3769
not a fatal execption.
3772
super(TestPrefixAliasRegistry, self).register(
3773
key, obj, help=help, info=info, override_existing=False)
3775
actual = self.get(key)
3777
'Test prefix alias %s is already used for %s, ignoring %s'
3778
% (key, actual, obj))
3780
def resolve_alias(self, id_start):
3781
"""Replace the alias by the prefix in the given string.
3783
Using an unknown prefix is an error to help catching typos.
3785
parts = id_start.split('.')
3787
parts[0] = self.get(parts[0])
3789
raise errors.BzrCommandError(
3790
'%s is not a known test prefix alias' % parts[0])
3791
return '.'.join(parts)
3794
test_prefix_alias_registry = TestPrefixAliasRegistry()
3795
"""Registry of test prefix aliases."""
3798
# This alias allows to detect typos ('bzrlin.') by making all valid test ids
3799
# appear prefixed ('bzrlib.' is "replaced" by 'bzrlib.').
3800
test_prefix_alias_registry.register('bzrlib', 'bzrlib')
3802
# Obvious highest levels prefixes, feel free to add your own via a plugin
3803
test_prefix_alias_registry.register('bd', 'bzrlib.doc')
3804
test_prefix_alias_registry.register('bu', 'bzrlib.utils')
3805
test_prefix_alias_registry.register('bt', 'bzrlib.tests')
3806
test_prefix_alias_registry.register('bb', 'bzrlib.tests.blackbox')
3807
test_prefix_alias_registry.register('bp', 'bzrlib.plugins')
3810
def _test_suite_testmod_names():
3811
"""Return the standard list of test module names to test."""
3814
'bzrlib.tests.blackbox',
3815
'bzrlib.tests.commands',
3816
'bzrlib.tests.doc_generate',
3817
'bzrlib.tests.per_branch',
3818
'bzrlib.tests.per_bzrdir',
3819
'bzrlib.tests.per_controldir',
3820
'bzrlib.tests.per_controldir_colo',
3821
'bzrlib.tests.per_foreign_vcs',
3822
'bzrlib.tests.per_interrepository',
3823
'bzrlib.tests.per_intertree',
3824
'bzrlib.tests.per_inventory',
3825
'bzrlib.tests.per_interbranch',
3826
'bzrlib.tests.per_lock',
3827
'bzrlib.tests.per_merger',
3828
'bzrlib.tests.per_transport',
3829
'bzrlib.tests.per_tree',
3830
'bzrlib.tests.per_pack_repository',
3831
'bzrlib.tests.per_repository',
3832
'bzrlib.tests.per_repository_chk',
3833
'bzrlib.tests.per_repository_reference',
3834
'bzrlib.tests.per_repository_vf',
3835
'bzrlib.tests.per_uifactory',
3836
'bzrlib.tests.per_versionedfile',
3837
'bzrlib.tests.per_workingtree',
3838
'bzrlib.tests.test__annotator',
3839
'bzrlib.tests.test__bencode',
3840
'bzrlib.tests.test__btree_serializer',
3841
'bzrlib.tests.test__chk_map',
3842
'bzrlib.tests.test__dirstate_helpers',
3843
'bzrlib.tests.test__groupcompress',
3844
'bzrlib.tests.test__known_graph',
3845
'bzrlib.tests.test__rio',
3846
'bzrlib.tests.test__simple_set',
3847
'bzrlib.tests.test__static_tuple',
3848
'bzrlib.tests.test__walkdirs_win32',
3849
'bzrlib.tests.test_ancestry',
3850
'bzrlib.tests.test_annotate',
3851
'bzrlib.tests.test_api',
3852
'bzrlib.tests.test_atomicfile',
3853
'bzrlib.tests.test_bad_files',
3854
'bzrlib.tests.test_bisect_multi',
3855
'bzrlib.tests.test_branch',
3856
'bzrlib.tests.test_branchbuilder',
3857
'bzrlib.tests.test_btree_index',
3858
'bzrlib.tests.test_bugtracker',
3859
'bzrlib.tests.test_bundle',
3860
'bzrlib.tests.test_bzrdir',
3861
'bzrlib.tests.test__chunks_to_lines',
3862
'bzrlib.tests.test_cache_utf8',
3863
'bzrlib.tests.test_chk_map',
3864
'bzrlib.tests.test_chk_serializer',
3865
'bzrlib.tests.test_chunk_writer',
3866
'bzrlib.tests.test_clean_tree',
3867
'bzrlib.tests.test_cleanup',
3868
'bzrlib.tests.test_cmdline',
3869
'bzrlib.tests.test_commands',
3870
'bzrlib.tests.test_commit',
3871
'bzrlib.tests.test_commit_merge',
3872
'bzrlib.tests.test_config',
3873
'bzrlib.tests.test_conflicts',
3874
'bzrlib.tests.test_controldir',
3875
'bzrlib.tests.test_counted_lock',
3876
'bzrlib.tests.test_crash',
3877
'bzrlib.tests.test_decorators',
3878
'bzrlib.tests.test_delta',
3879
'bzrlib.tests.test_debug',
3880
'bzrlib.tests.test_diff',
3881
'bzrlib.tests.test_directory_service',
3882
'bzrlib.tests.test_dirstate',
3883
'bzrlib.tests.test_email_message',
3884
'bzrlib.tests.test_eol_filters',
3885
'bzrlib.tests.test_errors',
3886
'bzrlib.tests.test_export',
3887
'bzrlib.tests.test_export_pot',
3888
'bzrlib.tests.test_extract',
3889
'bzrlib.tests.test_fetch',
3890
'bzrlib.tests.test_fixtures',
3891
'bzrlib.tests.test_fifo_cache',
3892
'bzrlib.tests.test_filters',
3893
'bzrlib.tests.test_filter_tree',
3894
'bzrlib.tests.test_ftp_transport',
3895
'bzrlib.tests.test_foreign',
3896
'bzrlib.tests.test_generate_docs',
3897
'bzrlib.tests.test_generate_ids',
3898
'bzrlib.tests.test_globbing',
3899
'bzrlib.tests.test_gpg',
3900
'bzrlib.tests.test_graph',
3901
'bzrlib.tests.test_groupcompress',
3902
'bzrlib.tests.test_hashcache',
3903
'bzrlib.tests.test_help',
3904
'bzrlib.tests.test_hooks',
3905
'bzrlib.tests.test_http',
3906
'bzrlib.tests.test_http_response',
3907
'bzrlib.tests.test_https_ca_bundle',
3908
'bzrlib.tests.test_i18n',
3909
'bzrlib.tests.test_identitymap',
3910
'bzrlib.tests.test_ignores',
3911
'bzrlib.tests.test_index',
3912
'bzrlib.tests.test_import_tariff',
3913
'bzrlib.tests.test_info',
3914
'bzrlib.tests.test_inv',
3915
'bzrlib.tests.test_inventory_delta',
3916
'bzrlib.tests.test_knit',
3917
'bzrlib.tests.test_lazy_import',
3918
'bzrlib.tests.test_lazy_regex',
3919
'bzrlib.tests.test_library_state',
3920
'bzrlib.tests.test_lock',
3921
'bzrlib.tests.test_lockable_files',
3922
'bzrlib.tests.test_lockdir',
3923
'bzrlib.tests.test_log',
3924
'bzrlib.tests.test_lru_cache',
3925
'bzrlib.tests.test_lsprof',
3926
'bzrlib.tests.test_mail_client',
3927
'bzrlib.tests.test_matchers',
3928
'bzrlib.tests.test_memorytree',
3929
'bzrlib.tests.test_merge',
3930
'bzrlib.tests.test_merge3',
3931
'bzrlib.tests.test_merge_core',
3932
'bzrlib.tests.test_merge_directive',
3933
'bzrlib.tests.test_mergetools',
3934
'bzrlib.tests.test_missing',
3935
'bzrlib.tests.test_msgeditor',
3936
'bzrlib.tests.test_multiparent',
3937
'bzrlib.tests.test_mutabletree',
3938
'bzrlib.tests.test_nonascii',
3939
'bzrlib.tests.test_options',
3940
'bzrlib.tests.test_osutils',
3941
'bzrlib.tests.test_osutils_encodings',
3942
'bzrlib.tests.test_pack',
3943
'bzrlib.tests.test_patch',
3944
'bzrlib.tests.test_patches',
3945
'bzrlib.tests.test_permissions',
3946
'bzrlib.tests.test_plugins',
3947
'bzrlib.tests.test_progress',
3948
'bzrlib.tests.test_pyutils',
3949
'bzrlib.tests.test_read_bundle',
3950
'bzrlib.tests.test_reconcile',
3951
'bzrlib.tests.test_reconfigure',
3952
'bzrlib.tests.test_registry',
3953
'bzrlib.tests.test_remote',
3954
'bzrlib.tests.test_rename_map',
3955
'bzrlib.tests.test_repository',
3956
'bzrlib.tests.test_revert',
3957
'bzrlib.tests.test_revision',
3958
'bzrlib.tests.test_revisionspec',
3959
'bzrlib.tests.test_revisiontree',
3960
'bzrlib.tests.test_rio',
3961
'bzrlib.tests.test_rules',
3962
'bzrlib.tests.test_sampler',
3963
'bzrlib.tests.test_scenarios',
3964
'bzrlib.tests.test_script',
3965
'bzrlib.tests.test_selftest',
3966
'bzrlib.tests.test_serializer',
3967
'bzrlib.tests.test_setup',
3968
'bzrlib.tests.test_sftp_transport',
3969
'bzrlib.tests.test_shelf',
3970
'bzrlib.tests.test_shelf_ui',
3971
'bzrlib.tests.test_smart',
3972
'bzrlib.tests.test_smart_add',
3973
'bzrlib.tests.test_smart_request',
3974
'bzrlib.tests.test_smart_transport',
3975
'bzrlib.tests.test_smtp_connection',
3976
'bzrlib.tests.test_source',
3977
'bzrlib.tests.test_ssh_transport',
3978
'bzrlib.tests.test_status',
3979
'bzrlib.tests.test_store',
3980
'bzrlib.tests.test_strace',
3981
'bzrlib.tests.test_subsume',
3982
'bzrlib.tests.test_switch',
3983
'bzrlib.tests.test_symbol_versioning',
3984
'bzrlib.tests.test_tag',
3985
'bzrlib.tests.test_test_server',
3986
'bzrlib.tests.test_testament',
3987
'bzrlib.tests.test_textfile',
3988
'bzrlib.tests.test_textmerge',
3989
'bzrlib.tests.test_cethread',
3990
'bzrlib.tests.test_timestamp',
3991
'bzrlib.tests.test_trace',
3992
'bzrlib.tests.test_transactions',
3993
'bzrlib.tests.test_transform',
3994
'bzrlib.tests.test_transport',
3995
'bzrlib.tests.test_transport_log',
3996
'bzrlib.tests.test_tree',
3997
'bzrlib.tests.test_treebuilder',
3998
'bzrlib.tests.test_treeshape',
3999
'bzrlib.tests.test_tsort',
4000
'bzrlib.tests.test_tuned_gzip',
4001
'bzrlib.tests.test_ui',
4002
'bzrlib.tests.test_uncommit',
4003
'bzrlib.tests.test_upgrade',
4004
'bzrlib.tests.test_upgrade_stacked',
4005
'bzrlib.tests.test_urlutils',
4006
'bzrlib.tests.test_utextwrap',
4007
'bzrlib.tests.test_version',
4008
'bzrlib.tests.test_version_info',
4009
'bzrlib.tests.test_versionedfile',
4010
'bzrlib.tests.test_weave',
4011
'bzrlib.tests.test_whitebox',
4012
'bzrlib.tests.test_win32utils',
4013
'bzrlib.tests.test_workingtree',
4014
'bzrlib.tests.test_workingtree_4',
4015
'bzrlib.tests.test_wsgi',
4016
'bzrlib.tests.test_xml',
4020
def _test_suite_modules_to_doctest():
4021
"""Return the list of modules to doctest."""
4023
# GZ 2009-03-31: No docstrings with -OO so there's nothing to doctest
4027
'bzrlib.branchbuilder',
4028
'bzrlib.decorators',
4030
'bzrlib.iterablefile',
4035
'bzrlib.symbol_versioning',
4037
'bzrlib.tests.fixtures',
4039
'bzrlib.transport.http',
4040
'bzrlib.version_info_formats.format_custom',
4044
def test_suite(keep_only=None, starting_with=None):
4045
1248
"""Build and return TestSuite for the whole of bzrlib.
4047
:param keep_only: A list of test ids limiting the suite returned.
4049
:param starting_with: An id limiting the suite returned to the tests
4052
1250
This function can be replaced if you need to change the default test
4053
1251
suite on a global basis, but it is not encouraged.
1254
'bzrlib.tests.test_ancestry',
1255
'bzrlib.tests.test_api',
1256
'bzrlib.tests.test_bad_files',
1257
'bzrlib.tests.test_branch',
1258
'bzrlib.tests.test_bundle',
1259
'bzrlib.tests.test_bzrdir',
1260
'bzrlib.tests.test_command',
1261
'bzrlib.tests.test_commit',
1262
'bzrlib.tests.test_commit_merge',
1263
'bzrlib.tests.test_config',
1264
'bzrlib.tests.test_conflicts',
1265
'bzrlib.tests.test_decorators',
1266
'bzrlib.tests.test_diff',
1267
'bzrlib.tests.test_doc_generate',
1268
'bzrlib.tests.test_errors',
1269
'bzrlib.tests.test_escaped_store',
1270
'bzrlib.tests.test_fetch',
1271
'bzrlib.tests.test_gpg',
1272
'bzrlib.tests.test_graph',
1273
'bzrlib.tests.test_hashcache',
1274
'bzrlib.tests.test_http',
1275
'bzrlib.tests.test_http_response',
1276
'bzrlib.tests.test_identitymap',
1277
'bzrlib.tests.test_ignores',
1278
'bzrlib.tests.test_inv',
1279
'bzrlib.tests.test_knit',
1280
'bzrlib.tests.test_lockdir',
1281
'bzrlib.tests.test_lockable_files',
1282
'bzrlib.tests.test_log',
1283
'bzrlib.tests.test_merge',
1284
'bzrlib.tests.test_merge3',
1285
'bzrlib.tests.test_merge_core',
1286
'bzrlib.tests.test_missing',
1287
'bzrlib.tests.test_msgeditor',
1288
'bzrlib.tests.test_nonascii',
1289
'bzrlib.tests.test_options',
1290
'bzrlib.tests.test_osutils',
1291
'bzrlib.tests.test_patch',
1292
'bzrlib.tests.test_patches',
1293
'bzrlib.tests.test_permissions',
1294
'bzrlib.tests.test_plugins',
1295
'bzrlib.tests.test_progress',
1296
'bzrlib.tests.test_reconcile',
1297
'bzrlib.tests.test_repository',
1298
'bzrlib.tests.test_revision',
1299
'bzrlib.tests.test_revisionnamespaces',
1300
'bzrlib.tests.test_revprops',
1301
'bzrlib.tests.test_revisiontree',
1302
'bzrlib.tests.test_rio',
1303
'bzrlib.tests.test_sampler',
1304
'bzrlib.tests.test_selftest',
1305
'bzrlib.tests.test_setup',
1306
'bzrlib.tests.test_sftp_transport',
1307
'bzrlib.tests.test_smart_add',
1308
'bzrlib.tests.test_source',
1309
'bzrlib.tests.test_status',
1310
'bzrlib.tests.test_store',
1311
'bzrlib.tests.test_symbol_versioning',
1312
'bzrlib.tests.test_testament',
1313
'bzrlib.tests.test_textfile',
1314
'bzrlib.tests.test_textmerge',
1315
'bzrlib.tests.test_trace',
1316
'bzrlib.tests.test_transactions',
1317
'bzrlib.tests.test_transform',
1318
'bzrlib.tests.test_transport',
1319
'bzrlib.tests.test_tree',
1320
'bzrlib.tests.test_tsort',
1321
'bzrlib.tests.test_tuned_gzip',
1322
'bzrlib.tests.test_ui',
1323
'bzrlib.tests.test_upgrade',
1324
'bzrlib.tests.test_urlutils',
1325
'bzrlib.tests.test_versionedfile',
1326
'bzrlib.tests.test_weave',
1327
'bzrlib.tests.test_whitebox',
1328
'bzrlib.tests.test_workingtree',
1329
'bzrlib.tests.test_xml',
1331
test_transport_implementations = [
1332
'bzrlib.tests.test_transport_implementations',
1333
'bzrlib.tests.test_read_bundle',
1335
suite = TestUtil.TestSuite()
4056
1336
loader = TestUtil.TestLoader()
4058
if keep_only is not None:
4059
id_filter = TestIdList(keep_only)
4061
# We take precedence over keep_only because *at loading time* using
4062
# both options means we will load less tests for the same final result.
4063
def interesting_module(name):
4064
for start in starting_with:
4066
# Either the module name starts with the specified string
4067
name.startswith(start)
4068
# or it may contain tests starting with the specified string
4069
or start.startswith(name)
4073
loader = TestUtil.FilteredByModuleTestLoader(interesting_module)
4075
elif keep_only is not None:
4076
loader = TestUtil.FilteredByModuleTestLoader(id_filter.refers_to)
4077
def interesting_module(name):
4078
return id_filter.refers_to(name)
4081
loader = TestUtil.TestLoader()
4082
def interesting_module(name):
4083
# No filtering, all modules are interesting
4086
suite = loader.suiteClass()
4088
# modules building their suite with loadTestsFromModuleNames
4089
suite.addTest(loader.loadTestsFromModuleNames(_test_suite_testmod_names()))
4091
for mod in _test_suite_modules_to_doctest():
4092
if not interesting_module(mod):
4093
# No tests to keep here, move along
4096
# note that this really does mean "report only" -- doctest
4097
# still runs the rest of the examples
4098
doc_suite = IsolatedDocTestSuite(
4099
mod, optionflags=doctest.REPORT_ONLY_FIRST_FAILURE)
4100
except ValueError, e:
4101
print '**failed to get doctest for: %s\n%s' % (mod, e)
4103
if len(doc_suite._tests) == 0:
4104
raise errors.BzrError("no doctests found in %s" % (mod,))
4105
suite.addTest(doc_suite)
4107
default_encoding = sys.getdefaultencoding()
4108
for name, plugin in _mod_plugin.plugins().items():
4109
if not interesting_module(plugin.module.__name__):
4111
plugin_suite = plugin.test_suite()
4112
# We used to catch ImportError here and turn it into just a warning,
4113
# but really if you don't have --no-plugins this should be a failure.
4114
# mbp 20080213 - see http://bugs.launchpad.net/bugs/189771
4115
if plugin_suite is None:
4116
plugin_suite = plugin.load_plugin_tests(loader)
4117
if plugin_suite is not None:
4118
suite.addTest(plugin_suite)
4119
if default_encoding != sys.getdefaultencoding():
4121
'Plugin "%s" tried to reset default encoding to: %s', name,
4122
sys.getdefaultencoding())
4124
sys.setdefaultencoding(default_encoding)
4126
if keep_only is not None:
4127
# Now that the referred modules have loaded their tests, keep only the
4129
suite = filter_suite_by_id_list(suite, id_filter)
4130
# Do some sanity checks on the id_list filtering
4131
not_found, duplicates = suite_matches_id_list(suite, keep_only)
4133
# The tester has used both keep_only and starting_with, so he is
4134
# already aware that some tests are excluded from the list, there
4135
# is no need to tell him which.
4138
# Some tests mentioned in the list are not in the test suite. The
4139
# list may be out of date, report to the tester.
4140
for id in not_found:
4141
trace.warning('"%s" not found in the test suite', id)
4142
for id in duplicates:
4143
trace.warning('"%s" is used as an id by several tests', id)
1337
suite.addTest(loader.loadTestsFromModuleNames(testmod_names))
1338
from bzrlib.transport import TransportTestProviderAdapter
1339
adapter = TransportTestProviderAdapter()
1340
adapt_modules(test_transport_implementations, adapter, loader, suite)
1341
for package in packages_to_test():
1342
suite.addTest(package.test_suite())
1343
for m in MODULES_TO_TEST:
1344
suite.addTest(loader.loadTestsFromModule(m))
1345
for m in MODULES_TO_DOCTEST:
1346
suite.addTest(doctest.DocTestSuite(m))
1347
for name, plugin in bzrlib.plugin.all_plugins().items():
1348
if getattr(plugin, 'test_suite', None) is not None:
1349
suite.addTest(plugin.test_suite())
4148
def multiply_scenarios(*scenarios):
4149
"""Multiply two or more iterables of scenarios.
4151
It is safe to pass scenario generators or iterators.
4153
:returns: A list of compound scenarios: the cross-product of all
4154
scenarios, with the names concatenated and the parameters
4157
return reduce(_multiply_two_scenarios, map(list, scenarios))
4160
def _multiply_two_scenarios(scenarios_left, scenarios_right):
4161
"""Multiply two sets of scenarios.
4163
:returns: the cartesian product of the two sets of scenarios, that is
4164
a scenario for every possible combination of a left scenario and a
4168
('%s,%s' % (left_name, right_name),
4169
dict(left_dict.items() + right_dict.items()))
4170
for left_name, left_dict in scenarios_left
4171
for right_name, right_dict in scenarios_right]
4174
def multiply_tests(tests, scenarios, result):
4175
"""Multiply tests_list by scenarios into result.
4177
This is the core workhorse for test parameterisation.
4179
Typically the load_tests() method for a per-implementation test suite will
4180
call multiply_tests and return the result.
4182
:param tests: The tests to parameterise.
4183
:param scenarios: The scenarios to apply: pairs of (scenario_name,
4184
scenario_param_dict).
4185
:param result: A TestSuite to add created tests to.
4187
This returns the passed in result TestSuite with the cross product of all
4188
the tests repeated once for each scenario. Each test is adapted by adding
4189
the scenario name at the end of its id(), and updating the test object's
4190
__dict__ with the scenario_param_dict.
4192
>>> import bzrlib.tests.test_sampler
4193
>>> r = multiply_tests(
4194
... bzrlib.tests.test_sampler.DemoTest('test_nothing'),
4195
... [('one', dict(param=1)),
4196
... ('two', dict(param=2))],
4197
... TestUtil.TestSuite())
4198
>>> tests = list(iter_suite_tests(r))
4202
'bzrlib.tests.test_sampler.DemoTest.test_nothing(one)'
4208
for test in iter_suite_tests(tests):
4209
apply_scenarios(test, scenarios, result)
4213
def apply_scenarios(test, scenarios, result):
4214
"""Apply the scenarios in scenarios to test and add to result.
4216
:param test: The test to apply scenarios to.
4217
:param scenarios: An iterable of scenarios to apply to test.
4219
:seealso: apply_scenario
4221
for scenario in scenarios:
4222
result.addTest(apply_scenario(test, scenario))
4226
def apply_scenario(test, scenario):
4227
"""Copy test and apply scenario to it.
4229
:param test: A test to adapt.
4230
:param scenario: A tuple describing the scenarion.
4231
The first element of the tuple is the new test id.
4232
The second element is a dict containing attributes to set on the
4234
:return: The adapted test.
4236
new_id = "%s(%s)" % (test.id(), scenario[0])
4237
new_test = clone_test(test, new_id)
4238
for name, value in scenario[1].items():
4239
setattr(new_test, name, value)
4243
def clone_test(test, new_id):
4244
"""Clone a test giving it a new id.
4246
:param test: The test to clone.
4247
:param new_id: The id to assign to it.
4248
:return: The new test.
4250
new_test = copy.copy(test)
4251
new_test.id = lambda: new_id
4252
# XXX: Workaround <https://bugs.launchpad.net/testtools/+bug/637725>, which
4253
# causes cloned tests to share the 'details' dict. This makes it hard to
4254
# read the test output for parameterized tests, because tracebacks will be
4255
# associated with irrelevant tests.
4257
details = new_test._TestCase__details
4258
except AttributeError:
4259
# must be a different version of testtools than expected. Do nothing.
4262
# Reset the '__details' dict.
4263
new_test._TestCase__details = {}
4267
def permute_tests_for_extension(standard_tests, loader, py_module_name,
4269
"""Helper for permutating tests against an extension module.
4271
This is meant to be used inside a modules 'load_tests()' function. It will
4272
create 2 scenarios, and cause all tests in the 'standard_tests' to be run
4273
against both implementations. Setting 'test.module' to the appropriate
4274
module. See bzrlib.tests.test__chk_map.load_tests as an example.
4276
:param standard_tests: A test suite to permute
4277
:param loader: A TestLoader
4278
:param py_module_name: The python path to a python module that can always
4279
be loaded, and will be considered the 'python' implementation. (eg
4280
'bzrlib._chk_map_py')
4281
:param ext_module_name: The python path to an extension module. If the
4282
module cannot be loaded, a single test will be added, which notes that
4283
the module is not available. If it can be loaded, all standard_tests
4284
will be run against that module.
4285
:return: (suite, feature) suite is a test-suite that has all the permuted
4286
tests. feature is the Feature object that can be used to determine if
4287
the module is available.
4290
py_module = pyutils.get_named_object(py_module_name)
4292
('python', {'module': py_module}),
4294
suite = loader.suiteClass()
4295
feature = ModuleAvailableFeature(ext_module_name)
4296
if feature.available():
4297
scenarios.append(('C', {'module': feature.module}))
4299
# the compiled module isn't available, so we add a failing test
4300
class FailWithoutFeature(TestCase):
4301
def test_fail(self):
4302
self.requireFeature(feature)
4303
suite.addTest(loader.loadTestsFromTestCase(FailWithoutFeature))
4304
result = multiply_tests(standard_tests, scenarios, suite)
4305
return result, feature
4308
def _rmtree_temp_dir(dirname, test_id=None):
4309
# If LANG=C we probably have created some bogus paths
4310
# which rmtree(unicode) will fail to delete
4311
# so make sure we are using rmtree(str) to delete everything
4312
# except on win32, where rmtree(str) will fail
4313
# since it doesn't have the property of byte-stream paths
4314
# (they are either ascii or mbcs)
4315
if sys.platform == 'win32':
4316
# make sure we are using the unicode win32 api
4317
dirname = unicode(dirname)
4319
dirname = dirname.encode(sys.getfilesystemencoding())
4321
osutils.rmtree(dirname)
4323
# We don't want to fail here because some useful display will be lost
4324
# otherwise. Polluting the tmp dir is bad, but not giving all the
4325
# possible info to the test runner is even worse.
4327
ui.ui_factory.clear_term()
4328
sys.stderr.write('\nWhile running: %s\n' % (test_id,))
4329
# Ugly, but the last thing we want here is fail, so bear with it.
4330
printable_e = str(e).decode(osutils.get_user_encoding(), 'replace'
4331
).encode('ascii', 'replace')
4332
sys.stderr.write('Unable to remove testing dir %s\n%s'
4333
% (os.path.basename(dirname), printable_e))
4336
class Feature(object):
4337
"""An operating system Feature."""
4340
self._available = None
4342
def available(self):
4343
"""Is the feature available?
4345
:return: True if the feature is available.
4347
if self._available is None:
4348
self._available = self._probe()
4349
return self._available
4352
"""Implement this method in concrete features.
4354
:return: True if the feature is available.
4356
raise NotImplementedError
4359
if getattr(self, 'feature_name', None):
4360
return self.feature_name()
4361
return self.__class__.__name__
4364
class _SymlinkFeature(Feature):
4367
return osutils.has_symlinks()
4369
def feature_name(self):
4372
SymlinkFeature = _SymlinkFeature()
4375
class _HardlinkFeature(Feature):
4378
return osutils.has_hardlinks()
4380
def feature_name(self):
4383
HardlinkFeature = _HardlinkFeature()
4386
class _OsFifoFeature(Feature):
4389
return getattr(os, 'mkfifo', None)
4391
def feature_name(self):
4392
return 'filesystem fifos'
4394
OsFifoFeature = _OsFifoFeature()
4397
class _UnicodeFilenameFeature(Feature):
4398
"""Does the filesystem support Unicode filenames?"""
4402
# Check for character combinations unlikely to be covered by any
4403
# single non-unicode encoding. We use the characters
4404
# - greek small letter alpha (U+03B1) and
4405
# - braille pattern dots-123456 (U+283F).
4406
os.stat(u'\u03b1\u283f')
4407
except UnicodeEncodeError:
4409
except (IOError, OSError):
4410
# The filesystem allows the Unicode filename but the file doesn't
4414
# The filesystem allows the Unicode filename and the file exists,
4418
UnicodeFilenameFeature = _UnicodeFilenameFeature()
4421
class _CompatabilityThunkFeature(Feature):
4422
"""This feature is just a thunk to another feature.
4424
It issues a deprecation warning if it is accessed, to let you know that you
4425
should really use a different feature.
4428
def __init__(self, dep_version, module, name,
4429
replacement_name, replacement_module=None):
4430
super(_CompatabilityThunkFeature, self).__init__()
4431
self._module = module
4432
if replacement_module is None:
4433
replacement_module = module
4434
self._replacement_module = replacement_module
4436
self._replacement_name = replacement_name
4437
self._dep_version = dep_version
4438
self._feature = None
4441
if self._feature is None:
4442
depr_msg = self._dep_version % ('%s.%s'
4443
% (self._module, self._name))
4444
use_msg = ' Use %s.%s instead.' % (self._replacement_module,
4445
self._replacement_name)
4446
symbol_versioning.warn(depr_msg + use_msg, DeprecationWarning)
4447
# Import the new feature and use it as a replacement for the
4449
self._feature = pyutils.get_named_object(
4450
self._replacement_module, self._replacement_name)
4454
return self._feature._probe()
4457
class ModuleAvailableFeature(Feature):
4458
"""This is a feature than describes a module we want to be available.
4460
Declare the name of the module in __init__(), and then after probing, the
4461
module will be available as 'self.module'.
4463
:ivar module: The module if it is available, else None.
4466
def __init__(self, module_name):
4467
super(ModuleAvailableFeature, self).__init__()
4468
self.module_name = module_name
4472
self._module = __import__(self.module_name, {}, {}, [''])
4479
if self.available(): # Make sure the probe has been done
4483
def feature_name(self):
4484
return self.module_name
4487
def probe_unicode_in_user_encoding():
4488
"""Try to encode several unicode strings to use in unicode-aware tests.
4489
Return first successfull match.
4491
:return: (unicode value, encoded plain string value) or (None, None)
4493
possible_vals = [u'm\xb5', u'\xe1', u'\u0410']
4494
for uni_val in possible_vals:
4496
str_val = uni_val.encode(osutils.get_user_encoding())
4497
except UnicodeEncodeError:
4498
# Try a different character
4501
return uni_val, str_val
4505
def probe_bad_non_ascii(encoding):
4506
"""Try to find [bad] character with code [128..255]
4507
that cannot be decoded to unicode in some encoding.
4508
Return None if all non-ascii characters is valid
4511
for i in xrange(128, 256):
4514
char.decode(encoding)
4515
except UnicodeDecodeError:
4520
class _HTTPSServerFeature(Feature):
4521
"""Some tests want an https Server, check if one is available.
4523
Right now, the only way this is available is under python2.6 which provides
4534
def feature_name(self):
4535
return 'HTTPSServer'
4538
HTTPSServerFeature = _HTTPSServerFeature()
4541
class _UnicodeFilename(Feature):
4542
"""Does the filesystem support Unicode filenames?"""
4547
except UnicodeEncodeError:
4549
except (IOError, OSError):
4550
# The filesystem allows the Unicode filename but the file doesn't
4554
# The filesystem allows the Unicode filename and the file exists,
4558
UnicodeFilename = _UnicodeFilename()
4561
class _ByteStringNamedFilesystem(Feature):
4562
"""Is the filesystem based on bytes?"""
4565
if os.name == "posix":
4569
ByteStringNamedFilesystem = _ByteStringNamedFilesystem()
4572
class _UTF8Filesystem(Feature):
4573
"""Is the filesystem UTF-8?"""
4576
if osutils._fs_enc.upper() in ('UTF-8', 'UTF8'):
4580
UTF8Filesystem = _UTF8Filesystem()
4583
class _BreakinFeature(Feature):
4584
"""Does this platform support the breakin feature?"""
4587
from bzrlib import breakin
4588
if breakin.determine_signal() is None:
4590
if sys.platform == 'win32':
4591
# Windows doesn't have os.kill, and we catch the SIGBREAK signal.
4592
# We trigger SIGBREAK via a Console api so we need ctypes to
4593
# access the function
4600
def feature_name(self):
4601
return "SIGQUIT or SIGBREAK w/ctypes on win32"
4604
BreakinFeature = _BreakinFeature()
4607
class _CaseInsCasePresFilenameFeature(Feature):
4608
"""Is the file-system case insensitive, but case-preserving?"""
4611
fileno, name = tempfile.mkstemp(prefix='MixedCase')
4613
# first check truly case-preserving for created files, then check
4614
# case insensitive when opening existing files.
4615
name = osutils.normpath(name)
4616
base, rel = osutils.split(name)
4617
found_rel = osutils.canonical_relpath(base, name)
4618
return (found_rel == rel
4619
and os.path.isfile(name.upper())
4620
and os.path.isfile(name.lower()))
4625
def feature_name(self):
4626
return "case-insensitive case-preserving filesystem"
4628
CaseInsCasePresFilenameFeature = _CaseInsCasePresFilenameFeature()
4631
class _CaseInsensitiveFilesystemFeature(Feature):
4632
"""Check if underlying filesystem is case-insensitive but *not* case
4635
# Note that on Windows, Cygwin, MacOS etc, the file-systems are far
4636
# more likely to be case preserving, so this case is rare.
4639
if CaseInsCasePresFilenameFeature.available():
4642
if TestCaseWithMemoryTransport.TEST_ROOT is None:
4643
root = osutils.mkdtemp(prefix='testbzr-', suffix='.tmp')
4644
TestCaseWithMemoryTransport.TEST_ROOT = root
4646
root = TestCaseWithMemoryTransport.TEST_ROOT
4647
tdir = osutils.mkdtemp(prefix='case-sensitive-probe-', suffix='',
4649
name_a = osutils.pathjoin(tdir, 'a')
4650
name_A = osutils.pathjoin(tdir, 'A')
4652
result = osutils.isdir(name_A)
4653
_rmtree_temp_dir(tdir)
4656
def feature_name(self):
4657
return 'case-insensitive filesystem'
4659
CaseInsensitiveFilesystemFeature = _CaseInsensitiveFilesystemFeature()
4662
class _CaseSensitiveFilesystemFeature(Feature):
4665
if CaseInsCasePresFilenameFeature.available():
4667
elif CaseInsensitiveFilesystemFeature.available():
4672
def feature_name(self):
4673
return 'case-sensitive filesystem'
4675
# new coding style is for feature instances to be lowercase
4676
case_sensitive_filesystem_feature = _CaseSensitiveFilesystemFeature()
4679
# Only define SubUnitBzrRunner if subunit is available.
4681
from subunit import TestProtocolClient
4682
from subunit.test_results import AutoTimingTestResultDecorator
4683
class SubUnitBzrProtocolClient(TestProtocolClient):
4685
def addSuccess(self, test, details=None):
4686
# The subunit client always includes the details in the subunit
4687
# stream, but we don't want to include it in ours.
4688
if details is not None and 'log' in details:
4690
return super(SubUnitBzrProtocolClient, self).addSuccess(
4693
class SubUnitBzrRunner(TextTestRunner):
4694
def run(self, test):
4695
result = AutoTimingTestResultDecorator(
4696
SubUnitBzrProtocolClient(self.stream))
4702
class _PosixPermissionsFeature(Feature):
4706
# create temporary file and check if specified perms are maintained.
4709
write_perms = stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR
4710
f = tempfile.mkstemp(prefix='bzr_perms_chk_')
4713
os.chmod(name, write_perms)
4715
read_perms = os.stat(name).st_mode & 0777
4717
return (write_perms == read_perms)
4719
return (os.name == 'posix') and has_perms()
4721
def feature_name(self):
4722
return 'POSIX permissions support'
4724
posix_permissions_feature = _PosixPermissionsFeature()
1353
def adapt_modules(mods_list, adapter, loader, suite):
1354
"""Adapt the modules in mods_list using adapter and add to suite."""
1355
for test in iter_suite_tests(loader.loadTestsFromModuleNames(mods_list)):
1356
suite.addTests(adapter.adapt(test))
added
removed
bzrlib/tests/__init__.py
