58
74
from bzrlib.merge import merge_inner
59
75
import bzrlib.merge3
61
import bzrlib.osutils as osutils
62
76
import bzrlib.plugin
63
import bzrlib.progress as progress
64
from bzrlib.revision import common_ancestor
65
77
import bzrlib.store
78
from bzrlib import symbol_versioning
79
from bzrlib.symbol_versioning import (
66
85
import bzrlib.trace
67
86
from bzrlib.transport import get_transport
68
87
import bzrlib.transport
69
from bzrlib.transport.local import LocalRelpathServer
88
from bzrlib.transport.local import LocalURLServer
89
from bzrlib.transport.memory import MemoryServer
70
90
from bzrlib.transport.readonly import ReadonlyServer
71
from bzrlib.trace import mutter
72
from bzrlib.tests.TestUtil import TestLoader, TestSuite
91
from bzrlib.trace import mutter, note
92
from bzrlib.tests import TestUtil
93
from bzrlib.tests.http_server import HttpServer
94
from bzrlib.tests.TestUtil import (
73
98
from bzrlib.tests.treeshape import build_tree_contents
74
import bzrlib.urlutils as urlutils
99
import bzrlib.version_info_formats.format_custom
75
100
from bzrlib.workingtree import WorkingTree, WorkingTreeFormat2
77
default_transport = LocalRelpathServer
80
MODULES_TO_DOCTEST = [
82
bzrlib.bundle.serializer,
95
def packages_to_test():
96
"""Return a list of packages to test.
98
The packages are not globally imported so that import failures are
99
triggered when running selftest, not when importing the command.
102
import bzrlib.tests.blackbox
103
import bzrlib.tests.branch_implementations
104
import bzrlib.tests.bzrdir_implementations
105
import bzrlib.tests.interrepository_implementations
106
import bzrlib.tests.interversionedfile_implementations
107
import bzrlib.tests.repository_implementations
108
import bzrlib.tests.revisionstore_implementations
109
import bzrlib.tests.workingtree_implementations
112
bzrlib.tests.blackbox,
113
bzrlib.tests.branch_implementations,
114
bzrlib.tests.bzrdir_implementations,
115
bzrlib.tests.interrepository_implementations,
116
bzrlib.tests.interversionedfile_implementations,
117
bzrlib.tests.repository_implementations,
118
bzrlib.tests.revisionstore_implementations,
119
bzrlib.tests.workingtree_implementations,
123
class _MyResult(unittest._TextTestResult):
124
"""Custom TestResult.
126
Shows output in a different format, including displaying runtime for tests.
102
# Mark this python module as being part of the implementation
103
# of unittest: this gives us better tracebacks where the last
104
# shown frame is the test code, not our assertXYZ.
107
default_transport = LocalURLServer
110
class ExtendedTestResult(unittest._TextTestResult):
111
"""Accepts, reports and accumulates the results of running tests.
113
Compared to the unittest version this class adds support for
114
profiling, benchmarking, stopping as soon as a test fails, and
115
skipping tests. There are further-specialized subclasses for
116
different types of display.
118
When a test finishes, in whatever way, it calls one of the addSuccess,
119
addFailure or addError classes. These in turn may redirect to a more
120
specific case for the special test results supported by our extended
123
Note that just one of these objects is fed the results from many tests.
128
126
stop_early = False
130
def __init__(self, stream, descriptions, verbosity, pb=None):
128
def __init__(self, stream, descriptions, verbosity,
132
"""Construct new TestResult.
134
:param bench_history: Optionally, a writable file object to accumulate
131
137
unittest._TextTestResult.__init__(self, stream, descriptions, verbosity)
138
if bench_history is not None:
139
from bzrlib.version import _get_bzr_source_tree
140
src_tree = _get_bzr_source_tree()
143
revision_id = src_tree.get_parent_ids()[0]
145
# XXX: if this is a brand new tree, do the same as if there
149
# XXX: If there's no branch, what should we do?
151
bench_history.write("--date %s %s\n" % (time.time(), revision_id))
152
self._bench_history = bench_history
153
self.ui = ui.ui_factory
154
self.num_tests = num_tests
156
self.failure_count = 0
157
self.known_failure_count = 0
159
self.not_applicable_count = 0
160
self.unsupported = {}
162
self._overall_start_time = time.time()
134
def extractBenchmarkTime(self, testCase):
164
def _extractBenchmarkTime(self, testCase):
135
165
"""Add a benchmark time for the current test case."""
136
self._benchmarkTime = getattr(testCase, "_benchtime", None)
166
return getattr(testCase, "_benchtime", None)
138
168
def _elapsedTestTimeString(self):
139
169
"""Return a time string for the overall time the current test has taken."""
140
170
return self._formatTime(time.time() - self._start_time)
142
def _testTimeString(self):
143
if self._benchmarkTime is not None:
172
def _testTimeString(self, testCase):
173
benchmark_time = self._extractBenchmarkTime(testCase)
174
if benchmark_time is not None:
144
175
return "%s/%s" % (
145
self._formatTime(self._benchmarkTime),
176
self._formatTime(benchmark_time),
146
177
self._elapsedTestTimeString())
148
return " %s" % self._elapsedTestTimeString()
179
return " %s" % self._elapsedTestTimeString()
150
181
def _formatTime(self, seconds):
151
182
"""Format seconds as milliseconds with leading spaces."""
152
return "%5dms" % (1000 * seconds)
183
# some benchmarks can take thousands of seconds to run, so we need 8
185
return "%8dms" % (1000 * seconds)
154
def _ellipsise_unimportant_words(self, a_string, final_width,
156
"""Add ellipses (sp?) for overly long strings.
158
:param keep_start: If true preserve the start of a_string rather
162
if len(a_string) > final_width:
163
result = a_string[:final_width-3] + '...'
167
if len(a_string) > final_width:
168
result = '...' + a_string[3-final_width:]
171
return result.ljust(final_width)
187
def _shortened_test_description(self, test):
189
what = re.sub(r'^bzrlib\.(tests|benchmarks)\.', '', what)
173
192
def startTest(self, test):
174
193
unittest.TestResult.startTest(self, test)
175
# In a short description, the important words are in
176
# the beginning, but in an id, the important words are
178
SHOW_DESCRIPTIONS = False
180
if not self.showAll and self.dots and self.pb is not None:
183
final_width = osutils.terminal_width()
184
final_width = final_width - 15 - 8
186
if SHOW_DESCRIPTIONS:
187
what = test.shortDescription()
189
what = self._ellipsise_unimportant_words(what, final_width, keep_start=True)
192
if what.startswith('bzrlib.tests.'):
194
what = self._ellipsise_unimportant_words(what, final_width)
196
self.stream.write(what)
197
elif self.dots and self.pb is not None:
198
self.pb.update(what, self.testsRun - 1, None)
194
self.report_test_start(test)
195
test.number = self.count
200
196
self._recordTestStartTime()
202
198
def _recordTestStartTime(self):
203
199
"""Record that a test has started."""
204
200
self._start_time = time.time()
202
def _cleanupLogFile(self, test):
203
# We can only do this if we have one of our TestCases, not if
205
setKeepLogfile = getattr(test, 'setKeepLogfile', None)
206
if setKeepLogfile is not None:
206
209
def addError(self, test, err):
210
"""Tell result that test finished with an error.
212
Called from the TestCase run() method when the test
213
fails with an unexpected error.
215
self._testConcluded(test)
207
216
if isinstance(err[1], TestSkipped):
208
return self.addSkipped(test, err)
209
unittest.TestResult.addError(self, test, err)
210
self.extractBenchmarkTime(test)
212
self.stream.writeln("ERROR %s" % self._testTimeString())
213
elif self.dots and self.pb is None:
214
self.stream.write('E')
216
self.pb.update(self._ellipsise_unimportant_words('ERROR', 13), self.testsRun, None)
217
return self._addSkipped(test, err)
218
elif isinstance(err[1], UnavailableFeature):
219
return self.addNotSupported(test, err[1].args[0])
221
self._cleanupLogFile(test)
222
unittest.TestResult.addError(self, test, err)
223
self.error_count += 1
224
self.report_error(test, err)
221
228
def addFailure(self, test, err):
222
unittest.TestResult.addFailure(self, test, err)
223
self.extractBenchmarkTime(test)
225
self.stream.writeln(" FAIL %s" % self._testTimeString())
226
elif self.dots and self.pb is None:
227
self.stream.write('F')
229
self.pb.update(self._ellipsise_unimportant_words('FAIL', 13), self.testsRun, None)
229
"""Tell result that test failed.
231
Called from the TestCase run() method when the test
232
fails because e.g. an assert() method failed.
234
self._testConcluded(test)
235
if isinstance(err[1], KnownFailure):
236
return self._addKnownFailure(test, err)
238
self._cleanupLogFile(test)
239
unittest.TestResult.addFailure(self, test, err)
240
self.failure_count += 1
241
self.report_failure(test, err)
234
245
def addSuccess(self, test):
235
self.extractBenchmarkTime(test)
237
self.stream.writeln(' OK %s' % self._testTimeString())
238
for bench_called, stats in getattr(test, '_benchcalls', []):
239
self.stream.writeln('LSProf output for %s(%s, %s)' % bench_called)
240
stats.pprint(file=self.stream)
241
elif self.dots and self.pb is None:
242
self.stream.write('~')
244
self.pb.update(self._ellipsise_unimportant_words('OK', 13), self.testsRun, None)
246
unittest.TestResult.addSuccess(self, test)
248
def addSkipped(self, test, skip_excinfo):
249
self.extractBenchmarkTime(test)
251
print >>self.stream, ' SKIP %s' % self._testTimeString()
252
print >>self.stream, ' %s' % skip_excinfo[1]
253
elif self.dots and self.pb is None:
254
self.stream.write('S')
256
self.pb.update(self._ellipsise_unimportant_words('SKIP', 13), self.testsRun, None)
258
# seems best to treat this as success from point-of-view of unittest
259
# -- it actually does nothing so it barely matters :)
260
unittest.TestResult.addSuccess(self, test)
246
"""Tell result that test completed successfully.
248
Called from the TestCase run()
250
self._testConcluded(test)
251
if self._bench_history is not None:
252
benchmark_time = self._extractBenchmarkTime(test)
253
if benchmark_time is not None:
254
self._bench_history.write("%s %s\n" % (
255
self._formatTime(benchmark_time),
257
self.report_success(test)
258
self._cleanupLogFile(test)
259
unittest.TestResult.addSuccess(self, test)
260
test._log_contents = ''
262
def _testConcluded(self, test):
263
"""Common code when a test has finished.
265
Called regardless of whether it succeded, failed, etc.
269
def _addKnownFailure(self, test, err):
270
self.known_failure_count += 1
271
self.report_known_failure(test, err)
273
def addNotSupported(self, test, feature):
274
"""The test will not be run because of a missing feature.
276
# this can be called in two different ways: it may be that the
277
# test started running, and then raised (through addError)
278
# UnavailableFeature. Alternatively this method can be called
279
# while probing for features before running the tests; in that
280
# case we will see startTest and stopTest, but the test will never
282
self.unsupported.setdefault(str(feature), 0)
283
self.unsupported[str(feature)] += 1
284
self.report_unsupported(test, feature)
286
def _addSkipped(self, test, skip_excinfo):
287
if isinstance(skip_excinfo[1], TestNotApplicable):
288
self.not_applicable_count += 1
289
self.report_not_applicable(test, skip_excinfo)
292
self.report_skip(test, skip_excinfo)
295
except KeyboardInterrupt:
298
self.addError(test, test._exc_info())
300
# seems best to treat this as success from point-of-view of unittest
301
# -- it actually does nothing so it barely matters :)
302
unittest.TestResult.addSuccess(self, test)
303
test._log_contents = ''
262
305
def printErrorList(self, flavour, errors):
263
306
for test, err in errors:
264
307
self.stream.writeln(self.separator1)
265
self.stream.writeln("%s: %s" % (flavour, self.getDescription(test)))
308
self.stream.write("%s: " % flavour)
309
self.stream.writeln(self.getDescription(test))
266
310
if getattr(test, '_get_log', None) is not None:
268
print >>self.stream, \
269
('vvvv[log from %s]' % test.id()).ljust(78,'-')
270
print >>self.stream, test._get_log()
271
print >>self.stream, \
272
('^^^^[log from %s]' % test.id()).ljust(78,'-')
311
self.stream.write('\n')
313
('vvvv[log from %s]' % test.id()).ljust(78,'-'))
314
self.stream.write('\n')
315
self.stream.write(test._get_log())
316
self.stream.write('\n')
318
('^^^^[log from %s]' % test.id()).ljust(78,'-'))
319
self.stream.write('\n')
273
320
self.stream.writeln(self.separator2)
274
321
self.stream.writeln("%s" % err)
326
def report_cleaning_up(self):
329
def report_success(self, test):
332
def wasStrictlySuccessful(self):
333
if self.unsupported or self.known_failure_count:
335
return self.wasSuccessful()
338
class TextTestResult(ExtendedTestResult):
339
"""Displays progress and results of tests in text form"""
341
def __init__(self, stream, descriptions, verbosity,
346
ExtendedTestResult.__init__(self, stream, descriptions, verbosity,
347
bench_history, num_tests)
349
self.pb = self.ui.nested_progress_bar()
350
self._supplied_pb = False
353
self._supplied_pb = True
354
self.pb.show_pct = False
355
self.pb.show_spinner = False
356
self.pb.show_eta = False,
357
self.pb.show_count = False
358
self.pb.show_bar = False
360
def report_starting(self):
361
self.pb.update('[test 0/%d] starting...' % (self.num_tests))
363
def _progress_prefix_text(self):
364
# the longer this text, the less space we have to show the test
366
a = '[%d' % self.count # total that have been run
367
# tests skipped as known not to be relevant are not important enough
369
## if self.skip_count:
370
## a += ', %d skip' % self.skip_count
371
## if self.known_failure_count:
372
## a += '+%dX' % self.known_failure_count
373
if self.num_tests is not None:
374
a +='/%d' % self.num_tests
376
runtime = time.time() - self._overall_start_time
378
a += '%dm%ds' % (runtime / 60, runtime % 60)
382
a += ', %d err' % self.error_count
383
if self.failure_count:
384
a += ', %d fail' % self.failure_count
386
a += ', %d missing' % len(self.unsupported)
390
def report_test_start(self, test):
393
self._progress_prefix_text()
395
+ self._shortened_test_description(test))
397
def _test_description(self, test):
398
return self._shortened_test_description(test)
400
def report_error(self, test, err):
401
self.pb.note('ERROR: %s\n %s\n',
402
self._test_description(test),
406
def report_failure(self, test, err):
407
self.pb.note('FAIL: %s\n %s\n',
408
self._test_description(test),
412
def report_known_failure(self, test, err):
413
self.pb.note('XFAIL: %s\n%s\n',
414
self._test_description(test), err[1])
416
def report_skip(self, test, skip_excinfo):
419
def report_not_applicable(self, test, skip_excinfo):
422
def report_unsupported(self, test, feature):
423
"""test cannot be run because feature is missing."""
425
def report_cleaning_up(self):
426
self.pb.update('cleaning up...')
429
if not self._supplied_pb:
433
class VerboseTestResult(ExtendedTestResult):
434
"""Produce long output, with one line per test run plus times"""
436
def _ellipsize_to_right(self, a_string, final_width):
437
"""Truncate and pad a string, keeping the right hand side"""
438
if len(a_string) > final_width:
439
result = '...' + a_string[3-final_width:]
442
return result.ljust(final_width)
444
def report_starting(self):
445
self.stream.write('running %d tests...\n' % self.num_tests)
447
def report_test_start(self, test):
449
name = self._shortened_test_description(test)
450
# width needs space for 6 char status, plus 1 for slash, plus 2 10-char
451
# numbers, plus a trailing blank
452
# when NUMBERED_DIRS: plus 5 chars on test number, plus 1 char on space
453
self.stream.write(self._ellipsize_to_right(name,
454
osutils.terminal_width()-30))
457
def _error_summary(self, err):
459
return '%s%s' % (indent, err[1])
461
def report_error(self, test, err):
462
self.stream.writeln('ERROR %s\n%s'
463
% (self._testTimeString(test),
464
self._error_summary(err)))
466
def report_failure(self, test, err):
467
self.stream.writeln(' FAIL %s\n%s'
468
% (self._testTimeString(test),
469
self._error_summary(err)))
471
def report_known_failure(self, test, err):
472
self.stream.writeln('XFAIL %s\n%s'
473
% (self._testTimeString(test),
474
self._error_summary(err)))
476
def report_success(self, test):
477
self.stream.writeln(' OK %s' % self._testTimeString(test))
478
for bench_called, stats in getattr(test, '_benchcalls', []):
479
self.stream.writeln('LSProf output for %s(%s, %s)' % bench_called)
480
stats.pprint(file=self.stream)
481
# flush the stream so that we get smooth output. This verbose mode is
482
# used to show the output in PQM.
485
def report_skip(self, test, skip_excinfo):
486
self.stream.writeln(' SKIP %s\n%s'
487
% (self._testTimeString(test),
488
self._error_summary(skip_excinfo)))
490
def report_not_applicable(self, test, skip_excinfo):
491
self.stream.writeln(' N/A %s\n%s'
492
% (self._testTimeString(test),
493
self._error_summary(skip_excinfo)))
495
def report_unsupported(self, test, feature):
496
"""test cannot be run because feature is missing."""
497
self.stream.writeln("NODEP %s\n The feature '%s' is not available."
498
%(self._testTimeString(test), feature))
277
501
class TextTestRunner(object):
278
502
stop_on_failure = False
516
976
self.assertEqual(mode, actual_mode,
517
977
'mode of %r incorrect (%o != %o)' % (path, mode, actual_mode))
979
def assertIsSameRealPath(self, path1, path2):
980
"""Fail if path1 and path2 points to different files"""
981
self.assertEqual(osutils.realpath(path1),
982
osutils.realpath(path2),
983
"apparent paths:\na = %s\nb = %s\n," % (path1, path2))
519
985
def assertIsInstance(self, obj, kls):
520
986
"""Fail if obj is not an instance of kls"""
521
987
if not isinstance(obj, kls):
522
988
self.fail("%r is an instance of %s rather than %s" % (
523
989
obj, obj.__class__, kls))
991
def expectFailure(self, reason, assertion, *args, **kwargs):
992
"""Invoke a test, expecting it to fail for the given reason.
994
This is for assertions that ought to succeed, but currently fail.
995
(The failure is *expected* but not *wanted*.) Please be very precise
996
about the failure you're expecting. If a new bug is introduced,
997
AssertionError should be raised, not KnownFailure.
999
Frequently, expectFailure should be followed by an opposite assertion.
1002
Intended to be used with a callable that raises AssertionError as the
1003
'assertion' parameter. args and kwargs are passed to the 'assertion'.
1005
Raises KnownFailure if the test fails. Raises AssertionError if the
1010
self.expectFailure('Math is broken', self.assertNotEqual, 54,
1012
self.assertEqual(42, dynamic_val)
1014
This means that a dynamic_val of 54 will cause the test to raise
1015
a KnownFailure. Once math is fixed and the expectFailure is removed,
1016
only a dynamic_val of 42 will allow the test to pass. Anything other
1017
than 54 or 42 will cause an AssertionError.
1020
assertion(*args, **kwargs)
1021
except AssertionError:
1022
raise KnownFailure(reason)
1024
self.fail('Unexpected success. Should have failed: %s' % reason)
1026
def assertFileEqual(self, content, path):
1027
"""Fail if path does not contain 'content'."""
1028
self.failUnlessExists(path)
1029
f = file(path, 'rb')
1034
self.assertEqualDiff(content, s)
1036
def failUnlessExists(self, path):
1037
"""Fail unless path or paths, which may be abs or relative, exist."""
1038
if not isinstance(path, basestring):
1040
self.failUnlessExists(p)
1042
self.failUnless(osutils.lexists(path),path+" does not exist")
1044
def failIfExists(self, path):
1045
"""Fail if path or paths, which may be abs or relative, exist."""
1046
if not isinstance(path, basestring):
1048
self.failIfExists(p)
1050
self.failIf(osutils.lexists(path),path+" exists")
1052
def _capture_deprecation_warnings(self, a_callable, *args, **kwargs):
1053
"""A helper for callDeprecated and applyDeprecated.
1055
:param a_callable: A callable to call.
1056
:param args: The positional arguments for the callable
1057
:param kwargs: The keyword arguments for the callable
1058
:return: A tuple (warnings, result). result is the result of calling
1059
a_callable(``*args``, ``**kwargs``).
1062
def capture_warnings(msg, cls=None, stacklevel=None):
1063
# we've hooked into a deprecation specific callpath,
1064
# only deprecations should getting sent via it.
1065
self.assertEqual(cls, DeprecationWarning)
1066
local_warnings.append(msg)
1067
original_warning_method = symbol_versioning.warn
1068
symbol_versioning.set_warning_method(capture_warnings)
1070
result = a_callable(*args, **kwargs)
1072
symbol_versioning.set_warning_method(original_warning_method)
1073
return (local_warnings, result)
1075
def applyDeprecated(self, deprecation_format, a_callable, *args, **kwargs):
1076
"""Call a deprecated callable without warning the user.
1078
Note that this only captures warnings raised by symbol_versioning.warn,
1079
not other callers that go direct to the warning module.
1081
To test that a deprecated method raises an error, do something like
1084
self.assertRaises(errors.ReservedId,
1085
self.applyDeprecated,
1086
deprecated_in((1, 5, 0)),
1090
:param deprecation_format: The deprecation format that the callable
1091
should have been deprecated with. This is the same type as the
1092
parameter to deprecated_method/deprecated_function. If the
1093
callable is not deprecated with this format, an assertion error
1095
:param a_callable: A callable to call. This may be a bound method or
1096
a regular function. It will be called with ``*args`` and
1098
:param args: The positional arguments for the callable
1099
:param kwargs: The keyword arguments for the callable
1100
:return: The result of a_callable(``*args``, ``**kwargs``)
1102
call_warnings, result = self._capture_deprecation_warnings(a_callable,
1104
expected_first_warning = symbol_versioning.deprecation_string(
1105
a_callable, deprecation_format)
1106
if len(call_warnings) == 0:
1107
self.fail("No deprecation warning generated by call to %s" %
1109
self.assertEqual(expected_first_warning, call_warnings[0])
1112
def callCatchWarnings(self, fn, *args, **kw):
1113
"""Call a callable that raises python warnings.
1115
The caller's responsible for examining the returned warnings.
1117
If the callable raises an exception, the exception is not
1118
caught and propagates up to the caller. In that case, the list
1119
of warnings is not available.
1121
:returns: ([warning_object, ...], fn_result)
1123
# XXX: This is not perfect, because it completely overrides the
1124
# warnings filters, and some code may depend on suppressing particular
1125
# warnings. It's the easiest way to insulate ourselves from -Werror,
1126
# though. -- Andrew, 20071062
1128
def _catcher(message, category, filename, lineno, file=None):
1129
# despite the name, 'message' is normally(?) a Warning subclass
1131
wlist.append(message)
1132
saved_showwarning = warnings.showwarning
1133
saved_filters = warnings.filters
1135
warnings.showwarning = _catcher
1136
warnings.filters = []
1137
result = fn(*args, **kw)
1139
warnings.showwarning = saved_showwarning
1140
warnings.filters = saved_filters
1141
return wlist, result
1143
def callDeprecated(self, expected, callable, *args, **kwargs):
1144
"""Assert that a callable is deprecated in a particular way.
1146
This is a very precise test for unusual requirements. The
1147
applyDeprecated helper function is probably more suited for most tests
1148
as it allows you to simply specify the deprecation format being used
1149
and will ensure that that is issued for the function being called.
1151
Note that this only captures warnings raised by symbol_versioning.warn,
1152
not other callers that go direct to the warning module. To catch
1153
general warnings, use callCatchWarnings.
1155
:param expected: a list of the deprecation warnings expected, in order
1156
:param callable: The callable to call
1157
:param args: The positional arguments for the callable
1158
:param kwargs: The keyword arguments for the callable
1160
call_warnings, result = self._capture_deprecation_warnings(callable,
1162
self.assertEqual(expected, call_warnings)
525
1165
def _startLogFile(self):
526
1166
"""Send bzr and test log messages to a temporary file.
528
1168
The file is removed as the test is torn down.
530
1170
fileno, name = tempfile.mkstemp(suffix='.log', prefix='testbzr')
531
encoder, decoder, stream_reader, stream_writer = codecs.lookup('UTF-8')
532
self._log_file = stream_writer(os.fdopen(fileno, 'w+'))
533
self._log_nonce = bzrlib.trace.enable_test_log(self._log_file)
1171
self._log_file = os.fdopen(fileno, 'w+')
1172
self._log_memento = bzrlib.trace.push_log_file(self._log_file)
534
1173
self._log_file_name = name
535
1174
self.addCleanup(self._finishLogFile)
537
1176
def _finishLogFile(self):
538
1177
"""Finished with the log file.
540
Read contents into memory, close, and delete.
1179
Close the file and delete it, unless setKeepLogfile was called.
542
bzrlib.trace.disable_test_log(self._log_nonce)
543
self._log_file.seek(0)
544
self._log_contents = self._log_file.read()
1181
if self._log_file is None:
1183
bzrlib.trace.pop_log_file(self._log_memento)
545
1184
self._log_file.close()
546
os.remove(self._log_file_name)
547
self._log_file = self._log_file_name = None
1185
self._log_file = None
1186
if not self._keep_log_file:
1187
os.remove(self._log_file_name)
1188
self._log_file_name = None
1190
def setKeepLogfile(self):
1191
"""Make the logfile not be deleted when _finishLogFile is called."""
1192
self._keep_log_file = True
549
1194
def addCleanup(self, callable):
550
1195
"""Arrange to run a callable when this case is torn down.
803
1694
sys.stderr = real_stderr
804
1695
sys.stdin = real_stdin
806
def merge(self, branch_from, wt_to):
807
"""A helper for tests to do a ui-less merge.
809
This should move to the main library when someone has time to integrate
812
# minimal ui-less merge.
813
wt_to.branch.fetch(branch_from)
814
base_rev = common_ancestor(branch_from.last_revision(),
815
wt_to.branch.last_revision(),
816
wt_to.branch.repository)
817
merge_inner(wt_to.branch, branch_from.basis_tree(),
818
wt_to.branch.repository.revision_tree(base_rev),
820
wt_to.add_pending_merge(branch_from.last_revision())
823
BzrTestBase = TestCase
1697
def reduceLockdirTimeout(self):
1698
"""Reduce the default lock timeout for the duration of the test, so that
1699
if LockContention occurs during a test, it does so quickly.
1701
Tests that expect to provoke LockContention errors should call this.
1703
orig_timeout = bzrlib.lockdir._DEFAULT_TIMEOUT_SECONDS
1705
bzrlib.lockdir._DEFAULT_TIMEOUT_SECONDS = orig_timeout
1706
self.addCleanup(resetTimeout)
1707
bzrlib.lockdir._DEFAULT_TIMEOUT_SECONDS = 0
1709
def make_utf8_encoded_stringio(self, encoding_type=None):
1710
"""Return a StringIOWrapper instance, that will encode Unicode
1713
if encoding_type is None:
1714
encoding_type = 'strict'
1716
output_encoding = 'utf-8'
1717
sio = codecs.getwriter(output_encoding)(sio, errors=encoding_type)
1718
sio.encoding = output_encoding
1722
class TestCaseWithMemoryTransport(TestCase):
1723
"""Common test class for tests that do not need disk resources.
1725
Tests that need disk resources should derive from TestCaseWithTransport.
1727
TestCaseWithMemoryTransport sets the TEST_ROOT variable for all bzr tests.
1729
For TestCaseWithMemoryTransport the test_home_dir is set to the name of
1730
a directory which does not exist. This serves to help ensure test isolation
1731
is preserved. test_dir is set to the TEST_ROOT, as is cwd, because they
1732
must exist. However, TestCaseWithMemoryTransport does not offer local
1733
file defaults for the transport in tests, nor does it obey the command line
1734
override, so tests that accidentally write to the common directory should
1737
:cvar TEST_ROOT: Directory containing all temporary directories, plus
1738
a .bzr directory that stops us ascending higher into the filesystem.
1744
def __init__(self, methodName='runTest'):
1745
# allow test parameterization after test construction and before test
1746
# execution. Variables that the parameterizer sets need to be
1747
# ones that are not set by setUp, or setUp will trash them.
1748
super(TestCaseWithMemoryTransport, self).__init__(methodName)
1749
self.vfs_transport_factory = default_transport
1750
self.transport_server = None
1751
self.transport_readonly_server = None
1752
self.__vfs_server = None
1754
def get_transport(self, relpath=None):
1755
"""Return a writeable transport.
1757
This transport is for the test scratch space relative to
1760
:param relpath: a path relative to the base url.
1762
t = get_transport(self.get_url(relpath))
1763
self.assertFalse(t.is_readonly())
1766
def get_readonly_transport(self, relpath=None):
1767
"""Return a readonly transport for the test scratch space
1769
This can be used to test that operations which should only need
1770
readonly access in fact do not try to write.
1772
:param relpath: a path relative to the base url.
1774
t = get_transport(self.get_readonly_url(relpath))
1775
self.assertTrue(t.is_readonly())
1778
def create_transport_readonly_server(self):
1779
"""Create a transport server from class defined at init.
1781
This is mostly a hook for daughter classes.
1783
return self.transport_readonly_server()
1785
def get_readonly_server(self):
1786
"""Get the server instance for the readonly transport
1788
This is useful for some tests with specific servers to do diagnostics.
1790
if self.__readonly_server is None:
1791
if self.transport_readonly_server is None:
1792
# readonly decorator requested
1793
# bring up the server
1794
self.__readonly_server = ReadonlyServer()
1795
self.__readonly_server.setUp(self.get_vfs_only_server())
1797
self.__readonly_server = self.create_transport_readonly_server()
1798
self.__readonly_server.setUp(self.get_vfs_only_server())
1799
self.addCleanup(self.__readonly_server.tearDown)
1800
return self.__readonly_server
1802
def get_readonly_url(self, relpath=None):
1803
"""Get a URL for the readonly transport.
1805
This will either be backed by '.' or a decorator to the transport
1806
used by self.get_url()
1807
relpath provides for clients to get a path relative to the base url.
1808
These should only be downwards relative, not upwards.
1810
base = self.get_readonly_server().get_url()
1811
return self._adjust_url(base, relpath)
1813
def get_vfs_only_server(self):
1814
"""Get the vfs only read/write server instance.
1816
This is useful for some tests with specific servers that need
1819
For TestCaseWithMemoryTransport this is always a MemoryServer, and there
1820
is no means to override it.
1822
if self.__vfs_server is None:
1823
self.__vfs_server = MemoryServer()
1824
self.__vfs_server.setUp()
1825
self.addCleanup(self.__vfs_server.tearDown)
1826
return self.__vfs_server
1828
def get_server(self):
1829
"""Get the read/write server instance.
1831
This is useful for some tests with specific servers that need
1834
This is built from the self.transport_server factory. If that is None,
1835
then the self.get_vfs_server is returned.
1837
if self.__server is None:
1838
if self.transport_server is None or self.transport_server is self.vfs_transport_factory:
1839
return self.get_vfs_only_server()
1841
# bring up a decorated means of access to the vfs only server.
1842
self.__server = self.transport_server()
1844
self.__server.setUp(self.get_vfs_only_server())
1845
except TypeError, e:
1846
# This should never happen; the try:Except here is to assist
1847
# developers having to update code rather than seeing an
1848
# uninformative TypeError.
1849
raise Exception, "Old server API in use: %s, %s" % (self.__server, e)
1850
self.addCleanup(self.__server.tearDown)
1851
return self.__server
1853
def _adjust_url(self, base, relpath):
1854
"""Get a URL (or maybe a path) for the readwrite transport.
1856
This will either be backed by '.' or to an equivalent non-file based
1858
relpath provides for clients to get a path relative to the base url.
1859
These should only be downwards relative, not upwards.
1861
if relpath is not None and relpath != '.':
1862
if not base.endswith('/'):
1864
# XXX: Really base should be a url; we did after all call
1865
# get_url()! But sometimes it's just a path (from
1866
# LocalAbspathServer), and it'd be wrong to append urlescaped data
1867
# to a non-escaped local path.
1868
if base.startswith('./') or base.startswith('/'):
1871
base += urlutils.escape(relpath)
1874
def get_url(self, relpath=None):
1875
"""Get a URL (or maybe a path) for the readwrite transport.
1877
This will either be backed by '.' or to an equivalent non-file based
1879
relpath provides for clients to get a path relative to the base url.
1880
These should only be downwards relative, not upwards.
1882
base = self.get_server().get_url()
1883
return self._adjust_url(base, relpath)
1885
def get_vfs_only_url(self, relpath=None):
1886
"""Get a URL (or maybe a path for the plain old vfs transport.
1888
This will never be a smart protocol. It always has all the
1889
capabilities of the local filesystem, but it might actually be a
1890
MemoryTransport or some other similar virtual filesystem.
1892
This is the backing transport (if any) of the server returned by
1893
get_url and get_readonly_url.
1895
:param relpath: provides for clients to get a path relative to the base
1896
url. These should only be downwards relative, not upwards.
1899
base = self.get_vfs_only_server().get_url()
1900
return self._adjust_url(base, relpath)
1902
def _create_safety_net(self):
1903
"""Make a fake bzr directory.
1905
This prevents any tests propagating up onto the TEST_ROOT directory's
1908
root = TestCaseWithMemoryTransport.TEST_ROOT
1909
bzrdir.BzrDir.create_standalone_workingtree(root)
1911
def _check_safety_net(self):
1912
"""Check that the safety .bzr directory have not been touched.
1914
_make_test_root have created a .bzr directory to prevent tests from
1915
propagating. This method ensures than a test did not leaked.
1917
root = TestCaseWithMemoryTransport.TEST_ROOT
1918
wt = workingtree.WorkingTree.open(root)
1919
last_rev = wt.last_revision()
1920
if last_rev != 'null:':
1921
# The current test have modified the /bzr directory, we need to
1922
# recreate a new one or all the followng tests will fail.
1923
# If you need to inspect its content uncomment the following line
1924
# import pdb; pdb.set_trace()
1925
_rmtree_temp_dir(root + '/.bzr')
1926
self._create_safety_net()
1927
raise AssertionError('%s/.bzr should not be modified' % root)
1929
def _make_test_root(self):
1930
if TestCaseWithMemoryTransport.TEST_ROOT is None:
1931
root = osutils.mkdtemp(prefix='testbzr-', suffix='.tmp')
1932
TestCaseWithMemoryTransport.TEST_ROOT = root
1934
self._create_safety_net()
1936
# The same directory is used by all tests, and we're not
1937
# specifically told when all tests are finished. This will do.
1938
atexit.register(_rmtree_temp_dir, root)
1940
self.addCleanup(self._check_safety_net)
1942
def makeAndChdirToTestDir(self):
1943
"""Create a temporary directories for this one test.
1945
This must set self.test_home_dir and self.test_dir and chdir to
1948
For TestCaseWithMemoryTransport we chdir to the TEST_ROOT for this test.
1950
os.chdir(TestCaseWithMemoryTransport.TEST_ROOT)
1951
self.test_dir = TestCaseWithMemoryTransport.TEST_ROOT
1952
self.test_home_dir = self.test_dir + "/MemoryTransportMissingHomeDir"
1954
def make_branch(self, relpath, format=None):
1955
"""Create a branch on the transport at relpath."""
1956
repo = self.make_repository(relpath, format=format)
1957
return repo.bzrdir.create_branch()
1959
def make_bzrdir(self, relpath, format=None):
1961
# might be a relative or absolute path
1962
maybe_a_url = self.get_url(relpath)
1963
segments = maybe_a_url.rsplit('/', 1)
1964
t = get_transport(maybe_a_url)
1965
if len(segments) > 1 and segments[-1] not in ('', '.'):
1969
if isinstance(format, basestring):
1970
format = bzrdir.format_registry.make_bzrdir(format)
1971
return format.initialize_on_transport(t)
1972
except errors.UninitializableFormat:
1973
raise TestSkipped("Format %s is not initializable." % format)
1975
def make_repository(self, relpath, shared=False, format=None):
1976
"""Create a repository on our default transport at relpath.
1978
Note that relpath must be a relative path, not a full url.
1980
# FIXME: If you create a remoterepository this returns the underlying
1981
# real format, which is incorrect. Actually we should make sure that
1982
# RemoteBzrDir returns a RemoteRepository.
1983
# maybe mbp 20070410
1984
made_control = self.make_bzrdir(relpath, format=format)
1985
return made_control.create_repository(shared=shared)
1987
def make_branch_and_memory_tree(self, relpath, format=None):
1988
"""Create a branch on the default transport and a MemoryTree for it."""
1989
b = self.make_branch(relpath, format=format)
1990
return memorytree.MemoryTree.create_on_branch(b)
1992
def overrideEnvironmentForTesting(self):
1993
os.environ['HOME'] = self.test_home_dir
1994
os.environ['BZR_HOME'] = self.test_home_dir
1997
super(TestCaseWithMemoryTransport, self).setUp()
1998
self._make_test_root()
1999
_currentdir = os.getcwdu()
2000
def _leaveDirectory():
2001
os.chdir(_currentdir)
2002
self.addCleanup(_leaveDirectory)
2003
self.makeAndChdirToTestDir()
2004
self.overrideEnvironmentForTesting()
2005
self.__readonly_server = None
2006
self.__server = None
2007
self.reduceLockdirTimeout()
826
class TestCaseInTempDir(TestCase):
2010
class TestCaseInTempDir(TestCaseWithMemoryTransport):
827
2011
"""Derived class that runs a test within a temporary directory.
829
2013
This is useful for tests that need to create a branch, etc.
1125
2230
def setUp(self):
1126
2231
super(ChrootedTestCase, self).setUp()
1127
if not self.transport_server == bzrlib.transport.memory.MemoryServer:
1128
self.transport_readonly_server = bzrlib.transport.http.HttpServer
2232
if not self.vfs_transport_factory == MemoryServer:
2233
self.transport_readonly_server = HttpServer
2236
def condition_id_re(pattern):
2237
"""Create a condition filter which performs a re check on a test's id.
2239
:param pattern: A regular expression string.
2240
:return: A callable that returns True if the re matches.
2242
filter_re = re.compile(pattern)
2243
def condition(test):
2245
return filter_re.search(test_id)
2249
def condition_isinstance(klass_or_klass_list):
2250
"""Create a condition filter which returns isinstance(param, klass).
2252
:return: A callable which when called with one parameter obj return the
2253
result of isinstance(obj, klass_or_klass_list).
2256
return isinstance(obj, klass_or_klass_list)
2260
def condition_id_in_list(id_list):
2261
"""Create a condition filter which verify that test's id in a list.
2263
:param id_list: A TestIdList object.
2264
:return: A callable that returns True if the test's id appears in the list.
2266
def condition(test):
2267
return id_list.includes(test.id())
2271
def condition_id_startswith(start):
2272
"""Create a condition filter verifying that test's id starts with a string.
2274
:param start: A string.
2275
:return: A callable that returns True if the test's id starts with the
2278
def condition(test):
2279
return test.id().startswith(start)
2283
def exclude_tests_by_condition(suite, condition):
2284
"""Create a test suite which excludes some tests from suite.
2286
:param suite: The suite to get tests from.
2287
:param condition: A callable whose result evaluates True when called with a
2288
test case which should be excluded from the result.
2289
:return: A suite which contains the tests found in suite that fail
2293
for test in iter_suite_tests(suite):
2294
if not condition(test):
2296
return TestUtil.TestSuite(result)
2299
def filter_suite_by_condition(suite, condition):
2300
"""Create a test suite by filtering another one.
2302
:param suite: The source suite.
2303
:param condition: A callable whose result evaluates True when called with a
2304
test case which should be included in the result.
2305
:return: A suite which contains the tests found in suite that pass
2309
for test in iter_suite_tests(suite):
2312
return TestUtil.TestSuite(result)
1131
2315
def filter_suite_by_re(suite, pattern):
1132
result = TestSuite()
1133
filter_re = re.compile(pattern)
2316
"""Create a test suite by filtering another one.
2318
:param suite: the source suite
2319
:param pattern: pattern that names must match
2320
:returns: the newly created suite
2322
condition = condition_id_re(pattern)
2323
result_suite = filter_suite_by_condition(suite, condition)
2327
def filter_suite_by_id_list(suite, test_id_list):
2328
"""Create a test suite by filtering another one.
2330
:param suite: The source suite.
2331
:param test_id_list: A list of the test ids to keep as strings.
2332
:returns: the newly created suite
2334
condition = condition_id_in_list(test_id_list)
2335
result_suite = filter_suite_by_condition(suite, condition)
2339
def filter_suite_by_id_startswith(suite, start):
2340
"""Create a test suite by filtering another one.
2342
:param suite: The source suite.
2343
:param start: A string the test id must start with.
2344
:returns: the newly created suite
2346
condition = condition_id_startswith(start)
2347
result_suite = filter_suite_by_condition(suite, condition)
2351
def exclude_tests_by_re(suite, pattern):
2352
"""Create a test suite which excludes some tests from suite.
2354
:param suite: The suite to get tests from.
2355
:param pattern: A regular expression string. Test ids that match this
2356
pattern will be excluded from the result.
2357
:return: A TestSuite that contains all the tests from suite without the
2358
tests that matched pattern. The order of tests is the same as it was in
2361
return exclude_tests_by_condition(suite, condition_id_re(pattern))
2364
def preserve_input(something):
2365
"""A helper for performing test suite transformation chains.
2367
:param something: Anything you want to preserve.
2373
def randomize_suite(suite):
2374
"""Return a new TestSuite with suite's tests in random order.
2376
The tests in the input suite are flattened into a single suite in order to
2377
accomplish this. Any nested TestSuites are removed to provide global
2380
tests = list(iter_suite_tests(suite))
2381
random.shuffle(tests)
2382
return TestUtil.TestSuite(tests)
2385
def split_suite_by_condition(suite, condition):
2386
"""Split a test suite into two by a condition.
2388
:param suite: The suite to split.
2389
:param condition: The condition to match on. Tests that match this
2390
condition are returned in the first test suite, ones that do not match
2391
are in the second suite.
2392
:return: A tuple of two test suites, where the first contains tests from
2393
suite matching the condition, and the second contains the remainder
2394
from suite. The order within each output suite is the same as it was in
1134
2399
for test in iter_suite_tests(suite):
1135
if filter_re.search(test.id()):
1136
result.addTest(test)
2401
matched.append(test)
2403
did_not_match.append(test)
2404
return TestUtil.TestSuite(matched), TestUtil.TestSuite(did_not_match)
2407
def split_suite_by_re(suite, pattern):
2408
"""Split a test suite into two by a regular expression.
2410
:param suite: The suite to split.
2411
:param pattern: A regular expression string. Test ids that match this
2412
pattern will be in the first test suite returned, and the others in the
2413
second test suite returned.
2414
:return: A tuple of two test suites, where the first contains tests from
2415
suite matching pattern, and the second contains the remainder from
2416
suite. The order within each output suite is the same as it was in
2419
return split_suite_by_condition(suite, condition_id_re(pattern))
1140
2422
def run_suite(suite, name='test', verbose=False, pattern=".*",
1141
stop_on_failure=False, keep_output=False,
1142
transport=None, lsprof_timed=None):
1143
TestCaseInTempDir._TEST_NAME = name
2423
stop_on_failure=False,
2424
transport=None, lsprof_timed=None, bench_history=None,
2425
matching_tests_first=None,
2428
exclude_pattern=None,
1144
2430
TestCase._gather_lsprof_in_benchmarks = lsprof_timed
1150
pb = progress.ProgressBar()
1151
2435
runner = TextTestRunner(stream=sys.stdout,
1152
2436
descriptions=0,
1153
2437
verbosity=verbosity,
1154
keep_output=keep_output,
2438
bench_history=bench_history,
2439
list_only=list_only,
1156
2441
runner.stop_on_failure=stop_on_failure
1158
suite = filter_suite_by_re(suite, pattern)
2442
# Initialise the random number generator and display the seed used.
2443
# We convert the seed to a long to make it reuseable across invocations.
2444
random_order = False
2445
if random_seed is not None:
2447
if random_seed == "now":
2448
random_seed = long(time.time())
2450
# Convert the seed to a long if we can
2452
random_seed = long(random_seed)
2455
runner.stream.writeln("Randomizing test order using seed %s\n" %
2457
random.seed(random_seed)
2458
# Customise the list of tests if requested
2459
if exclude_pattern is not None:
2460
suite = exclude_tests_by_re(suite, exclude_pattern)
2462
order_changer = randomize_suite
2464
order_changer = preserve_input
2465
if pattern != '.*' or random_order:
2466
if matching_tests_first:
2467
suites = map(order_changer, split_suite_by_re(suite, pattern))
2468
suite = TestUtil.TestSuite(suites)
2470
suite = order_changer(filter_suite_by_re(suite, pattern))
1159
2472
result = runner.run(suite)
2475
return result.wasStrictlySuccessful()
1160
2477
return result.wasSuccessful()
2480
# Controlled by "bzr selftest -E=..." option
2481
selftest_debug_flags = set()
1163
2484
def selftest(verbose=False, pattern=".*", stop_on_failure=True,
1165
2485
transport=None,
1166
2486
test_suite_factory=None,
2489
matching_tests_first=None,
2492
exclude_pattern=None,
1168
2498
"""Run the whole test suite under the enhanced runner"""
2499
# XXX: Very ugly way to do this...
2500
# Disable warning about old formats because we don't want it to disturb
2501
# any blackbox tests.
2502
from bzrlib import repository
2503
repository._deprecation_warning_done = True
1169
2505
global default_transport
1170
2506
if transport is None:
1171
2507
transport = default_transport
1172
2508
old_transport = default_transport
1173
2509
default_transport = transport
2510
global selftest_debug_flags
2511
old_debug_flags = selftest_debug_flags
2512
if debug_flags is not None:
2513
selftest_debug_flags = set(debug_flags)
2515
if load_list is None:
2518
keep_only = load_test_id_list(load_list)
1175
2519
if test_suite_factory is None:
1176
suite = test_suite()
2520
suite = test_suite(keep_only, starting_with)
1178
2522
suite = test_suite_factory()
1179
2523
return run_suite(suite, 'testbzr', verbose=verbose, pattern=pattern,
1180
stop_on_failure=stop_on_failure, keep_output=keep_output,
2524
stop_on_failure=stop_on_failure,
1181
2525
transport=transport,
1182
lsprof_timed=lsprof_timed)
2526
lsprof_timed=lsprof_timed,
2527
bench_history=bench_history,
2528
matching_tests_first=matching_tests_first,
2529
list_only=list_only,
2530
random_seed=random_seed,
2531
exclude_pattern=exclude_pattern,
1184
2534
default_transport = old_transport
2535
selftest_debug_flags = old_debug_flags
2538
def load_test_id_list(file_name):
2539
"""Load a test id list from a text file.
2541
The format is one test id by line. No special care is taken to impose
2542
strict rules, these test ids are used to filter the test suite so a test id
2543
that do not match an existing test will do no harm. This allows user to add
2544
comments, leave blank lines, etc.
2548
ftest = open(file_name, 'rt')
2550
if e.errno != errno.ENOENT:
2553
raise errors.NoSuchFile(file_name)
2555
for test_name in ftest.readlines():
2556
test_list.append(test_name.strip())
2561
def suite_matches_id_list(test_suite, id_list):
2562
"""Warns about tests not appearing or appearing more than once.
2564
:param test_suite: A TestSuite object.
2565
:param test_id_list: The list of test ids that should be found in
2568
:return: (absents, duplicates) absents is a list containing the test found
2569
in id_list but not in test_suite, duplicates is a list containing the
2570
test found multiple times in test_suite.
2572
When using a prefined test id list, it may occurs that some tests do not
2573
exist anymore or that some tests use the same id. This function warns the
2574
tester about potential problems in his workflow (test lists are volatile)
2575
or in the test suite itself (using the same id for several tests does not
2576
help to localize defects).
2578
# Build a dict counting id occurrences
2580
for test in iter_suite_tests(test_suite):
2582
tests[id] = tests.get(id, 0) + 1
2587
occurs = tests.get(id, 0)
2589
not_found.append(id)
2591
duplicates.append(id)
2593
return not_found, duplicates
2596
class TestIdList(object):
2597
"""Test id list to filter a test suite.
2599
Relying on the assumption that test ids are built as:
2600
<module>[.<class>.<method>][(<param>+)], <module> being in python dotted
2601
notation, this class offers methods to :
2602
- avoid building a test suite for modules not refered to in the test list,
2603
- keep only the tests listed from the module test suite.
2606
def __init__(self, test_id_list):
2607
# When a test suite needs to be filtered against us we compare test ids
2608
# for equality, so a simple dict offers a quick and simple solution.
2609
self.tests = dict().fromkeys(test_id_list, True)
2611
# While unittest.TestCase have ids like:
2612
# <module>.<class>.<method>[(<param+)],
2613
# doctest.DocTestCase can have ids like:
2616
# <module>.<function>
2617
# <module>.<class>.<method>
2619
# Since we can't predict a test class from its name only, we settle on
2620
# a simple constraint: a test id always begins with its module name.
2623
for test_id in test_id_list:
2624
parts = test_id.split('.')
2625
mod_name = parts.pop(0)
2626
modules[mod_name] = True
2628
mod_name += '.' + part
2629
modules[mod_name] = True
2630
self.modules = modules
2632
def refers_to(self, module_name):
2633
"""Is there tests for the module or one of its sub modules."""
2634
return self.modules.has_key(module_name)
2636
def includes(self, test_id):
2637
return self.tests.has_key(test_id)
2640
def test_suite(keep_only=None, starting_with=None):
1188
2641
"""Build and return TestSuite for the whole of bzrlib.
2643
:param keep_only: A list of test ids limiting the suite returned.
2645
:param starting_with: An id limiting the suite returned to the tests
1190
2648
This function can be replaced if you need to change the default test
1191
2649
suite on a global basis, but it is not encouraged.
1193
from doctest import DocTestSuite
1195
global MODULES_TO_DOCTEST
2653
'bzrlib.util.tests.test_bencode',
2654
'bzrlib.tests.blackbox',
2655
'bzrlib.tests.branch_implementations',
2656
'bzrlib.tests.bzrdir_implementations',
2657
'bzrlib.tests.commands',
2658
'bzrlib.tests.inventory_implementations',
2659
'bzrlib.tests.interrepository_implementations',
2660
'bzrlib.tests.intertree_implementations',
2661
'bzrlib.tests.interversionedfile_implementations',
2662
'bzrlib.tests.per_lock',
2663
'bzrlib.tests.repository_implementations',
2664
'bzrlib.tests.revisionstore_implementations',
2665
'bzrlib.tests.test__dirstate_helpers',
1198
2666
'bzrlib.tests.test_ancestry',
2667
'bzrlib.tests.test_annotate',
1199
2668
'bzrlib.tests.test_api',
2669
'bzrlib.tests.test_atomicfile',
1200
2670
'bzrlib.tests.test_bad_files',
2671
'bzrlib.tests.test_bisect_multi',
1201
2672
'bzrlib.tests.test_branch',
2673
'bzrlib.tests.test_branchbuilder',
2674
'bzrlib.tests.test_bugtracker',
1202
2675
'bzrlib.tests.test_bundle',
1203
2676
'bzrlib.tests.test_bzrdir',
1204
'bzrlib.tests.test_command',
2677
'bzrlib.tests.test_cache_utf8',
2678
'bzrlib.tests.test_commands',
1205
2679
'bzrlib.tests.test_commit',
1206
2680
'bzrlib.tests.test_commit_merge',
1207
2681
'bzrlib.tests.test_config',
1208
2682
'bzrlib.tests.test_conflicts',
2683
'bzrlib.tests.test_counted_lock',
1209
2684
'bzrlib.tests.test_decorators',
2685
'bzrlib.tests.test_delta',
2686
'bzrlib.tests.test_deprecated_graph',
1210
2687
'bzrlib.tests.test_diff',
1211
'bzrlib.tests.test_doc_generate',
2688
'bzrlib.tests.test_dirstate',
2689
'bzrlib.tests.test_directory_service',
2690
'bzrlib.tests.test_email_message',
1212
2691
'bzrlib.tests.test_errors',
1213
2692
'bzrlib.tests.test_escaped_store',
2693
'bzrlib.tests.test_extract',
1214
2694
'bzrlib.tests.test_fetch',
2695
'bzrlib.tests.test_ftp_transport',
2696
'bzrlib.tests.test_generate_docs',
2697
'bzrlib.tests.test_generate_ids',
2698
'bzrlib.tests.test_globbing',
1215
2699
'bzrlib.tests.test_gpg',
1216
2700
'bzrlib.tests.test_graph',
1217
2701
'bzrlib.tests.test_hashcache',
2702
'bzrlib.tests.test_help',
2703
'bzrlib.tests.test_hooks',
1218
2704
'bzrlib.tests.test_http',
2705
'bzrlib.tests.test_http_implementations',
2706
'bzrlib.tests.test_http_response',
2707
'bzrlib.tests.test_https_ca_bundle',
1219
2708
'bzrlib.tests.test_identitymap',
2709
'bzrlib.tests.test_ignores',
2710
'bzrlib.tests.test_index',
2711
'bzrlib.tests.test_info',
1220
2712
'bzrlib.tests.test_inv',
1221
2713
'bzrlib.tests.test_knit',
2714
'bzrlib.tests.test_lazy_import',
2715
'bzrlib.tests.test_lazy_regex',
1222
2716
'bzrlib.tests.test_lockdir',
1223
2717
'bzrlib.tests.test_lockable_files',
1224
2718
'bzrlib.tests.test_log',
2719
'bzrlib.tests.test_lsprof',
2720
'bzrlib.tests.test_lru_cache',
2721
'bzrlib.tests.test_mail_client',
2722
'bzrlib.tests.test_memorytree',
1225
2723
'bzrlib.tests.test_merge',
1226
2724
'bzrlib.tests.test_merge3',
1227
2725
'bzrlib.tests.test_merge_core',
2726
'bzrlib.tests.test_merge_directive',
1228
2727
'bzrlib.tests.test_missing',
1229
2728
'bzrlib.tests.test_msgeditor',
2729
'bzrlib.tests.test_multiparent',
2730
'bzrlib.tests.test_mutabletree',
1230
2731
'bzrlib.tests.test_nonascii',
1231
2732
'bzrlib.tests.test_options',
1232
2733
'bzrlib.tests.test_osutils',
2734
'bzrlib.tests.test_osutils_encodings',
2735
'bzrlib.tests.test_pack',
1233
2736
'bzrlib.tests.test_patch',
1234
2737
'bzrlib.tests.test_patches',
1235
2738
'bzrlib.tests.test_permissions',
1236
2739
'bzrlib.tests.test_plugins',
1237
2740
'bzrlib.tests.test_progress',
2741
'bzrlib.tests.test_read_bundle',
2742
'bzrlib.tests.test_reconfigure',
1238
2743
'bzrlib.tests.test_reconcile',
2744
'bzrlib.tests.test_registry',
2745
'bzrlib.tests.test_remote',
1239
2746
'bzrlib.tests.test_repository',
2747
'bzrlib.tests.test_revert',
1240
2748
'bzrlib.tests.test_revision',
1241
'bzrlib.tests.test_revisionnamespaces',
1242
'bzrlib.tests.test_revprops',
2749
'bzrlib.tests.test_revisionspec',
2750
'bzrlib.tests.test_revisiontree',
1243
2751
'bzrlib.tests.test_rio',
1244
2752
'bzrlib.tests.test_sampler',
1245
2753
'bzrlib.tests.test_selftest',
1246
2754
'bzrlib.tests.test_setup',
1247
2755
'bzrlib.tests.test_sftp_transport',
2756
'bzrlib.tests.test_smart',
1248
2757
'bzrlib.tests.test_smart_add',
2758
'bzrlib.tests.test_smart_transport',
2759
'bzrlib.tests.test_smtp_connection',
1249
2760
'bzrlib.tests.test_source',
2761
'bzrlib.tests.test_ssh_transport',
1250
2762
'bzrlib.tests.test_status',
1251
2763
'bzrlib.tests.test_store',
2764
'bzrlib.tests.test_strace',
2765
'bzrlib.tests.test_subsume',
2766
'bzrlib.tests.test_switch',
1252
2767
'bzrlib.tests.test_symbol_versioning',
2768
'bzrlib.tests.test_tag',
1253
2769
'bzrlib.tests.test_testament',
1254
2770
'bzrlib.tests.test_textfile',
1255
2771
'bzrlib.tests.test_textmerge',
2772
'bzrlib.tests.test_timestamp',
1256
2773
'bzrlib.tests.test_trace',
1257
2774
'bzrlib.tests.test_transactions',
1258
2775
'bzrlib.tests.test_transform',
1259
2776
'bzrlib.tests.test_transport',
2777
'bzrlib.tests.test_transport_implementations',
2778
'bzrlib.tests.test_tree',
2779
'bzrlib.tests.test_treebuilder',
1260
2780
'bzrlib.tests.test_tsort',
1261
2781
'bzrlib.tests.test_tuned_gzip',
1262
2782
'bzrlib.tests.test_ui',
2783
'bzrlib.tests.test_uncommit',
1263
2784
'bzrlib.tests.test_upgrade',
1264
2785
'bzrlib.tests.test_urlutils',
1265
2786
'bzrlib.tests.test_versionedfile',
2787
'bzrlib.tests.test_version',
2788
'bzrlib.tests.test_version_info',
1266
2789
'bzrlib.tests.test_weave',
1267
2790
'bzrlib.tests.test_whitebox',
2791
'bzrlib.tests.test_win32utils',
1268
2792
'bzrlib.tests.test_workingtree',
2793
'bzrlib.tests.test_workingtree_4',
2794
'bzrlib.tests.test_wsgi',
1269
2795
'bzrlib.tests.test_xml',
2796
'bzrlib.tests.tree_implementations',
2797
'bzrlib.tests.workingtree_implementations',
1271
test_transport_implementations = [
1272
'bzrlib.tests.test_transport_implementations']
1275
2800
loader = TestUtil.TestLoader()
1276
from bzrlib.transport import TransportTestProviderAdapter
1277
adapter = TransportTestProviderAdapter()
1278
adapt_modules(test_transport_implementations, adapter, loader, suite)
2802
if starting_with is not None:
2803
# We take precedence over keep_only because *at loading time* using
2804
# both options means we will load less tests for the same final result.
2805
def interesting_module(name):
2807
# Either the module name starts with the specified string
2808
name.startswith(starting_with)
2809
# or it may contain tests starting with the specified string
2810
or starting_with.startswith(name)
2812
loader = TestUtil.FilteredByModuleTestLoader(interesting_module)
2814
elif keep_only is not None:
2815
id_filter = TestIdList(keep_only)
2816
loader = TestUtil.FilteredByModuleTestLoader(id_filter.refers_to)
2817
def interesting_module(name):
2818
return id_filter.refers_to(name)
2821
loader = TestUtil.TestLoader()
2822
def interesting_module(name):
2823
# No filtering, all modules are interesting
2826
suite = loader.suiteClass()
2828
# modules building their suite with loadTestsFromModuleNames
1279
2829
suite.addTest(loader.loadTestsFromModuleNames(testmod_names))
1280
for package in packages_to_test():
1281
suite.addTest(package.test_suite())
1282
for m in MODULES_TO_TEST:
1283
suite.addTest(loader.loadTestsFromModule(m))
1284
for m in (MODULES_TO_DOCTEST):
1285
suite.addTest(DocTestSuite(m))
1286
for name, plugin in bzrlib.plugin.all_plugins().items():
1287
if getattr(plugin, 'test_suite', None) is not None:
1288
suite.addTest(plugin.test_suite())
2831
modules_to_doctest = [
2836
'bzrlib.iterablefile',
2841
'bzrlib.symbol_versioning',
2844
'bzrlib.version_info_formats.format_custom',
2847
for mod in modules_to_doctest:
2848
if not interesting_module(mod):
2849
# No tests to keep here, move along
2852
doc_suite = doctest.DocTestSuite(mod)
2853
except ValueError, e:
2854
print '**failed to get doctest for: %s\n%s' % (mod, e)
2856
suite.addTest(doc_suite)
2858
default_encoding = sys.getdefaultencoding()
2859
for name, plugin in bzrlib.plugin.plugins().items():
2860
if not interesting_module(plugin.module.__name__):
2862
plugin_suite = plugin.test_suite()
2863
# We used to catch ImportError here and turn it into just a warning,
2864
# but really if you don't have --no-plugins this should be a failure.
2865
# mbp 20080213 - see http://bugs.launchpad.net/bugs/189771
2866
if plugin_suite is None:
2867
plugin_suite = plugin.load_plugin_tests(loader)
2868
if plugin_suite is not None:
2869
suite.addTest(plugin_suite)
2870
if default_encoding != sys.getdefaultencoding():
2871
bzrlib.trace.warning(
2872
'Plugin "%s" tried to reset default encoding to: %s', name,
2873
sys.getdefaultencoding())
2875
sys.setdefaultencoding(default_encoding)
2877
if starting_with is not None:
2878
suite = filter_suite_by_id_startswith(suite, starting_with)
2880
if keep_only is not None:
2881
# Now that the referred modules have loaded their tests, keep only the
2883
suite = filter_suite_by_id_list(suite, id_filter)
2884
# Do some sanity checks on the id_list filtering
2885
not_found, duplicates = suite_matches_id_list(suite, keep_only)
2886
if starting_with is not None:
2887
# The tester has used both keep_only and starting_with, so he is
2888
# already aware that some tests are excluded from the list, there
2889
# is no need to tell him which.
2892
# Some tests mentioned in the list are not in the test suite. The
2893
# list may be out of date, report to the tester.
2894
for id in not_found:
2895
bzrlib.trace.warning('"%s" not found in the test suite', id)
2896
for id in duplicates:
2897
bzrlib.trace.warning('"%s" is used as an id by several tests', id)
2902
def multiply_tests_from_modules(module_name_list, scenario_iter, loader=None):
2903
"""Adapt all tests in some given modules to given scenarios.
2905
This is the recommended public interface for test parameterization.
2906
Typically the test_suite() method for a per-implementation test
2907
suite will call multiply_tests_from_modules and return the
2910
:param module_name_list: List of fully-qualified names of test
2912
:param scenario_iter: Iterable of pairs of (scenario_name,
2913
scenario_param_dict).
2914
:param loader: If provided, will be used instead of a new
2915
bzrlib.tests.TestLoader() instance.
2917
This returns a new TestSuite containing the cross product of
2918
all the tests in all the modules, each repeated for each scenario.
2919
Each test is adapted by adding the scenario name at the end
2920
of its name, and updating the test object's __dict__ with the
2921
scenario_param_dict.
2923
>>> r = multiply_tests_from_modules(
2924
... ['bzrlib.tests.test_sampler'],
2925
... [('one', dict(param=1)),
2926
... ('two', dict(param=2))])
2927
>>> tests = list(iter_suite_tests(r))
2931
'bzrlib.tests.test_sampler.DemoTest.test_nothing(one)'
2937
# XXX: Isn't load_tests() a better way to provide the same functionality
2938
# without forcing a predefined TestScenarioApplier ? --vila 080215
2940
loader = TestUtil.TestLoader()
2942
suite = loader.suiteClass()
2944
adapter = TestScenarioApplier()
2945
adapter.scenarios = list(scenario_iter)
2946
adapt_modules(module_name_list, adapter, loader, suite)
2950
def multiply_scenarios(scenarios_left, scenarios_right):
2951
"""Multiply two sets of scenarios.
2953
:returns: the cartesian product of the two sets of scenarios, that is
2954
a scenario for every possible combination of a left scenario and a
2958
('%s,%s' % (left_name, right_name),
2959
dict(left_dict.items() + right_dict.items()))
2960
for left_name, left_dict in scenarios_left
2961
for right_name, right_dict in scenarios_right]
1292
2965
def adapt_modules(mods_list, adapter, loader, suite):
1293
2966
"""Adapt the modules in mods_list using adapter and add to suite."""
1294
2967
for test in iter_suite_tests(loader.loadTestsFromModuleNames(mods_list)):
1295
2968
suite.addTests(adapter.adapt(test))
2971
def adapt_tests(tests_list, adapter, loader, suite):
2972
"""Adapt the tests in tests_list using adapter and add to suite."""
2973
for test in tests_list:
2974
suite.addTests(adapter.adapt(loader.loadTestsFromName(test)))
2977
def _rmtree_temp_dir(dirname):
2978
# If LANG=C we probably have created some bogus paths
2979
# which rmtree(unicode) will fail to delete
2980
# so make sure we are using rmtree(str) to delete everything
2981
# except on win32, where rmtree(str) will fail
2982
# since it doesn't have the property of byte-stream paths
2983
# (they are either ascii or mbcs)
2984
if sys.platform == 'win32':
2985
# make sure we are using the unicode win32 api
2986
dirname = unicode(dirname)
2988
dirname = dirname.encode(sys.getfilesystemencoding())
2990
osutils.rmtree(dirname)
2992
if sys.platform == 'win32' and e.errno == errno.EACCES:
2993
sys.stderr.write(('Permission denied: '
2994
'unable to remove testing dir '
2995
'%s\n' % os.path.basename(dirname)))
3000
class Feature(object):
3001
"""An operating system Feature."""
3004
self._available = None
3006
def available(self):
3007
"""Is the feature available?
3009
:return: True if the feature is available.
3011
if self._available is None:
3012
self._available = self._probe()
3013
return self._available
3016
"""Implement this method in concrete features.
3018
:return: True if the feature is available.
3020
raise NotImplementedError
3023
if getattr(self, 'feature_name', None):
3024
return self.feature_name()
3025
return self.__class__.__name__
3028
class _SymlinkFeature(Feature):
3031
return osutils.has_symlinks()
3033
def feature_name(self):
3036
SymlinkFeature = _SymlinkFeature()
3039
class _HardlinkFeature(Feature):
3042
return osutils.has_hardlinks()
3044
def feature_name(self):
3047
HardlinkFeature = _HardlinkFeature()
3050
class _OsFifoFeature(Feature):
3053
return getattr(os, 'mkfifo', None)
3055
def feature_name(self):
3056
return 'filesystem fifos'
3058
OsFifoFeature = _OsFifoFeature()
3061
class TestScenarioApplier(object):
3062
"""A tool to apply scenarios to tests."""
3064
def adapt(self, test):
3065
"""Return a TestSuite containing a copy of test for each scenario."""
3066
result = unittest.TestSuite()
3067
for scenario in self.scenarios:
3068
result.addTest(self.adapt_test_to_scenario(test, scenario))
3071
def adapt_test_to_scenario(self, test, scenario):
3072
"""Copy test and apply scenario to it.
3074
:param test: A test to adapt.
3075
:param scenario: A tuple describing the scenarion.
3076
The first element of the tuple is the new test id.
3077
The second element is a dict containing attributes to set on the
3079
:return: The adapted test.
3081
from copy import deepcopy
3082
new_test = deepcopy(test)
3083
for name, value in scenario[1].items():
3084
setattr(new_test, name, value)
3085
new_id = "%s(%s)" % (new_test.id(), scenario[0])
3086
new_test.id = lambda: new_id
3090
def probe_unicode_in_user_encoding():
3091
"""Try to encode several unicode strings to use in unicode-aware tests.
3092
Return first successfull match.
3094
:return: (unicode value, encoded plain string value) or (None, None)
3096
possible_vals = [u'm\xb5', u'\xe1', u'\u0410']
3097
for uni_val in possible_vals:
3099
str_val = uni_val.encode(bzrlib.user_encoding)
3100
except UnicodeEncodeError:
3101
# Try a different character
3104
return uni_val, str_val
3108
def probe_bad_non_ascii(encoding):
3109
"""Try to find [bad] character with code [128..255]
3110
that cannot be decoded to unicode in some encoding.
3111
Return None if all non-ascii characters is valid
3114
for i in xrange(128, 256):
3117
char.decode(encoding)
3118
except UnicodeDecodeError:
3123
class _FTPServerFeature(Feature):
3124
"""Some tests want an FTP Server, check if one is available.
3126
Right now, the only way this is available is if 'medusa' is installed.
3127
http://www.amk.ca/python/code/medusa.html
3132
import bzrlib.tests.ftp_server
3137
def feature_name(self):
3140
FTPServerFeature = _FTPServerFeature()
3143
class _CaseInsensitiveFilesystemFeature(Feature):
3144
"""Check if underlined filesystem is case-insensitive
3145
(e.g. on Windows, Cygwin, MacOS)
3149
if TestCaseWithMemoryTransport.TEST_ROOT is None:
3150
root = osutils.mkdtemp(prefix='testbzr-', suffix='.tmp')
3151
TestCaseWithMemoryTransport.TEST_ROOT = root
3153
root = TestCaseWithMemoryTransport.TEST_ROOT
3154
tdir = osutils.mkdtemp(prefix='case-sensitive-probe-', suffix='',
3156
name_a = osutils.pathjoin(tdir, 'a')
3157
name_A = osutils.pathjoin(tdir, 'A')
3159
result = osutils.isdir(name_A)
3160
_rmtree_temp_dir(tdir)
3163
def feature_name(self):
3164
return 'case-insensitive filesystem'
3166
CaseInsensitiveFilesystemFeature = _CaseInsensitiveFilesystemFeature()
added
removed
bzrlib/tests/__init__.py
