74
59
from bzrlib.merge import merge_inner
75
60
import bzrlib.merge3
62
import bzrlib.osutils as osutils
76
63
import bzrlib.plugin
64
import bzrlib.progress as progress
65
from bzrlib.revision import common_ancestor
66
from bzrlib.revisionspec import RevisionSpec
77
67
import bzrlib.store
78
68
from bzrlib import symbol_versioning
79
from bzrlib.symbol_versioning import (
85
69
import bzrlib.trace
86
70
from bzrlib.transport import get_transport
87
71
import bzrlib.transport
88
from bzrlib.transport.local import LocalURLServer
89
from bzrlib.transport.memory import MemoryServer
72
from bzrlib.transport.local import LocalRelpathServer
90
73
from bzrlib.transport.readonly import ReadonlyServer
91
from bzrlib.trace import mutter, note
74
from bzrlib.trace import mutter
92
75
from bzrlib.tests import TestUtil
93
from bzrlib.tests.http_server import HttpServer
94
76
from bzrlib.tests.TestUtil import (
98
80
from bzrlib.tests.treeshape import build_tree_contents
99
import bzrlib.version_info_formats.format_custom
81
import bzrlib.urlutils as urlutils
100
82
from bzrlib.workingtree import WorkingTree, WorkingTreeFormat2
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.
84
default_transport = LocalRelpathServer
87
MODULES_TO_DOCTEST = [
89
bzrlib.bundle.serializer,
102
def packages_to_test():
103
"""Return a list of packages to test.
105
The packages are not globally imported so that import failures are
106
triggered when running selftest, not when importing the command.
109
import bzrlib.tests.blackbox
110
import bzrlib.tests.branch_implementations
111
import bzrlib.tests.bzrdir_implementations
112
import bzrlib.tests.interrepository_implementations
113
import bzrlib.tests.interversionedfile_implementations
114
import bzrlib.tests.intertree_implementations
115
import bzrlib.tests.repository_implementations
116
import bzrlib.tests.revisionstore_implementations
117
import bzrlib.tests.tree_implementations
118
import bzrlib.tests.workingtree_implementations
121
bzrlib.tests.blackbox,
122
bzrlib.tests.branch_implementations,
123
bzrlib.tests.bzrdir_implementations,
124
bzrlib.tests.interrepository_implementations,
125
bzrlib.tests.interversionedfile_implementations,
126
bzrlib.tests.intertree_implementations,
127
bzrlib.tests.repository_implementations,
128
bzrlib.tests.revisionstore_implementations,
129
bzrlib.tests.tree_implementations,
130
bzrlib.tests.workingtree_implementations,
134
class _MyResult(unittest._TextTestResult):
135
"""Custom TestResult.
137
Shows output in a different format, including displaying runtime for tests.
126
139
stop_early = False
128
def __init__(self, stream, descriptions, verbosity,
141
def __init__(self, stream, descriptions, verbosity, pb=None,
132
143
"""Construct new TestResult.
134
145
:param bench_history: Optionally, a writable file object to accumulate
135
146
benchmark results.
137
148
unittest._TextTestResult.__init__(self, stream, descriptions, verbosity)
138
150
if bench_history is not None:
139
151
from bzrlib.version import _get_bzr_source_tree
140
152
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
154
revision_id = src_tree.last_revision()
149
156
# XXX: If there's no branch, what should we do?
151
158
bench_history.write("--date %s %s\n" % (time.time(), revision_id))
152
159
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()
164
def _extractBenchmarkTime(self, testCase):
161
def extractBenchmarkTime(self, testCase):
165
162
"""Add a benchmark time for the current test case."""
166
return getattr(testCase, "_benchtime", None)
163
self._benchmarkTime = getattr(testCase, "_benchtime", None)
168
165
def _elapsedTestTimeString(self):
169
166
"""Return a time string for the overall time the current test has taken."""
170
167
return self._formatTime(time.time() - self._start_time)
172
def _testTimeString(self, testCase):
173
benchmark_time = self._extractBenchmarkTime(testCase)
174
if benchmark_time is not None:
169
def _testTimeString(self):
170
if self._benchmarkTime is not None:
175
171
return "%s/%s" % (
176
self._formatTime(benchmark_time),
172
self._formatTime(self._benchmarkTime),
177
173
self._elapsedTestTimeString())
179
return " %s" % self._elapsedTestTimeString()
175
return " %s" % self._elapsedTestTimeString()
181
177
def _formatTime(self, seconds):
182
178
"""Format seconds as milliseconds with leading spaces."""
183
# some benchmarks can take thousands of seconds to run, so we need 8
185
return "%8dms" % (1000 * seconds)
179
return "%5dms" % (1000 * seconds)
187
def _shortened_test_description(self, test):
189
what = re.sub(r'^bzrlib\.(tests|benchmarks)\.', '', what)
181
def _ellipsise_unimportant_words(self, a_string, final_width,
183
"""Add ellipses (sp?) for overly long strings.
185
:param keep_start: If true preserve the start of a_string rather
189
if len(a_string) > final_width:
190
result = a_string[:final_width-3] + '...'
194
if len(a_string) > final_width:
195
result = '...' + a_string[3-final_width:]
198
return result.ljust(final_width)
192
200
def startTest(self, test):
193
201
unittest.TestResult.startTest(self, test)
194
self.report_test_start(test)
195
test.number = self.count
202
# In a short description, the important words are in
203
# the beginning, but in an id, the important words are
205
SHOW_DESCRIPTIONS = False
207
if not self.showAll and self.dots and self.pb is not None:
210
final_width = osutils.terminal_width()
211
final_width = final_width - 15 - 8
213
if SHOW_DESCRIPTIONS:
214
what = test.shortDescription()
216
what = self._ellipsise_unimportant_words(what, final_width, keep_start=True)
219
if what.startswith('bzrlib.tests.'):
221
what = self._ellipsise_unimportant_words(what, final_width)
223
self.stream.write(what)
224
elif self.dots and self.pb is not None:
225
self.pb.update(what, self.testsRun - 1, None)
196
227
self._recordTestStartTime()
198
229
def _recordTestStartTime(self):
199
230
"""Record that a test has started."""
200
231
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:
209
233
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)
216
234
if isinstance(err[1], TestSkipped):
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)
235
return self.addSkipped(test, err)
236
unittest.TestResult.addError(self, test, err)
237
self.extractBenchmarkTime(test)
239
self.stream.writeln("ERROR %s" % self._testTimeString())
240
elif self.dots and self.pb is None:
241
self.stream.write('E')
243
self.pb.update(self._ellipsise_unimportant_words('ERROR', 13), self.testsRun, None)
244
self.pb.note(self._ellipsise_unimportant_words(
245
test.id() + ': ERROR',
246
osutils.terminal_width()))
228
251
def addFailure(self, test, err):
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)
252
unittest.TestResult.addFailure(self, test, err)
253
self.extractBenchmarkTime(test)
255
self.stream.writeln(" FAIL %s" % self._testTimeString())
256
elif self.dots and self.pb is None:
257
self.stream.write('F')
259
self.pb.update(self._ellipsise_unimportant_words('FAIL', 13), self.testsRun, None)
260
self.pb.note(self._ellipsise_unimportant_words(
261
test.id() + ': FAIL',
262
osutils.terminal_width()))
245
267
def addSuccess(self, test):
246
"""Tell result that test completed successfully.
248
Called from the TestCase run()
250
self._testConcluded(test)
268
self.extractBenchmarkTime(test)
251
269
if self._bench_history is not None:
252
benchmark_time = self._extractBenchmarkTime(test)
253
if benchmark_time is not None:
270
if self._benchmarkTime is not None:
254
271
self._bench_history.write("%s %s\n" % (
255
self._formatTime(benchmark_time),
272
self._formatTime(self._benchmarkTime),
257
self.report_success(test)
258
self._cleanupLogFile(test)
275
self.stream.writeln(' OK %s' % self._testTimeString())
276
for bench_called, stats in getattr(test, '_benchcalls', []):
277
self.stream.writeln('LSProf output for %s(%s, %s)' % bench_called)
278
stats.pprint(file=self.stream)
279
elif self.dots and self.pb is None:
280
self.stream.write('~')
282
self.pb.update(self._ellipsise_unimportant_words('OK', 13), self.testsRun, None)
259
284
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)
286
def addSkipped(self, test, skip_excinfo):
287
self.extractBenchmarkTime(test)
289
print >>self.stream, ' SKIP %s' % self._testTimeString()
290
print >>self.stream, ' %s' % skip_excinfo[1]
291
elif self.dots and self.pb is None:
292
self.stream.write('S')
294
self.pb.update(self._ellipsise_unimportant_words('SKIP', 13), self.testsRun, None)
296
# seems best to treat this as success from point-of-view of unittest
297
# -- it actually does nothing so it barely matters :)
295
300
except KeyboardInterrupt:
298
self.addError(test, test._exc_info())
303
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
305
unittest.TestResult.addSuccess(self, test)
303
test._log_contents = ''
305
307
def printErrorList(self, flavour, errors):
306
308
for test, err in errors:
307
309
self.stream.writeln(self.separator1)
308
self.stream.write("%s: " % flavour)
309
self.stream.writeln(self.getDescription(test))
310
self.stream.writeln("%s: %s" % (flavour, self.getDescription(test)))
310
311
if getattr(test, '_get_log', None) is not None:
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')
313
print >>self.stream, \
314
('vvvv[log from %s]' % test.id()).ljust(78,'-')
315
print >>self.stream, test._get_log()
316
print >>self.stream, \
317
('^^^^[log from %s]' % test.id()).ljust(78,'-')
320
318
self.stream.writeln(self.separator2)
321
319
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))
501
322
class TextTestRunner(object):
502
323
stop_on_failure = False
976
571
self.assertEqual(mode, actual_mode,
977
572
'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))
985
574
def assertIsInstance(self, obj, kls):
986
575
"""Fail if obj is not an instance of kls"""
987
576
if not isinstance(obj, kls):
988
577
self.fail("%r is an instance of %s rather than %s" % (
989
578
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.
580
def callDeprecated(self, expected, callable, *args, **kwargs):
581
"""Assert that a callable is deprecated in a particular way.
583
:param expected: a list of the deprecation warnings expected, in order
584
:param callable: The callable to call
1056
585
:param args: The positional arguments for the callable
1057
586
: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``).
1061
588
local_warnings = []
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.
589
def capture_warnings(msg, cls, stacklevel=None):
1065
590
self.assertEqual(cls, DeprecationWarning)
1066
591
local_warnings.append(msg)
1067
original_warning_method = symbol_versioning.warn
592
method = symbol_versioning.warn
1068
593
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)
595
result = callable(*args, **kwargs)
597
symbol_versioning.set_warning_method(method)
598
self.assertEqual(expected, local_warnings)
1165
601
def _startLogFile(self):
1312
704
# TODO: Perhaps this should keep running cleanups even if
1313
705
# one of them fails?
1315
# Actually pop the cleanups from the list so tearDown running
1316
# twice is safe (this happens for skipped tests).
1317
while self._cleanups:
1318
self._cleanups.pop()()
706
for cleanup_fn in reversed(self._cleanups):
1320
709
def log(self, *args):
1323
def _get_log(self, keep_log_file=False):
1324
"""Get the log from bzrlib.trace calls from this test.
1326
:param keep_log_file: When True, if the log is still a file on disk
1327
leave it as a file on disk. When False, if the log is still a file
1328
on disk, the log file is deleted and the log preserved as
1330
:return: A string containing the log.
1332
# flush the log file, to get all content
1334
bzrlib.trace._trace_file.flush()
1335
if self._log_contents:
1336
# XXX: this can hardly contain the content flushed above --vila
713
"""Return as a string the log for this test"""
714
if self._log_file_name:
715
return open(self._log_file_name).read()
1338
717
return self._log_contents
1339
if self._log_file_name is not None:
1340
logfile = open(self._log_file_name)
1342
log_contents = logfile.read()
1345
if not keep_log_file:
1346
self._log_contents = log_contents
1348
os.remove(self._log_file_name)
1350
if sys.platform == 'win32' and e.errno == errno.EACCES:
1351
sys.stderr.write(('Unable to delete log file '
1352
' %r\n' % self._log_file_name))
1357
return "DELETED log file to reduce memory footprint"
1359
def requireFeature(self, feature):
1360
"""This test requires a specific feature is available.
1362
:raises UnavailableFeature: When feature is not available.
718
# TODO: Delete the log after it's been read in
720
def capture(self, cmd, retcode=0):
721
"""Shortcut that splits cmd into words, runs, and returns stdout"""
722
return self.run_bzr_captured(cmd.split(), retcode=retcode)[0]
724
def run_bzr_captured(self, argv, retcode=0, encoding=None, stdin=None):
725
"""Invoke bzr and return (stdout, stderr).
727
Useful for code that wants to check the contents of the
728
output, the way error messages are presented, etc.
730
This should be the main method for tests that want to exercise the
731
overall behavior of the bzr application (rather than a unit test
732
or a functional test of the library.)
734
Much of the old code runs bzr by forking a new copy of Python, but
735
that is slower, harder to debug, and generally not necessary.
737
This runs bzr through the interface that catches and reports
738
errors, and with logging set to something approximating the
739
default, so that error reporting can be checked.
741
:param argv: arguments to invoke bzr
742
:param retcode: expected return code, or None for don't-care.
743
:param encoding: encoding for sys.stdout and sys.stderr
744
:param stdin: A string to be used as stdin for the command.
1364
if not feature.available():
1365
raise UnavailableFeature(feature)
1367
def _run_bzr_autosplit(self, args, retcode, encoding, stdin,
1369
"""Run bazaar command line, splitting up a string command line."""
1370
if isinstance(args, basestring):
1371
# shlex don't understand unicode strings,
1372
# so args should be plain string (bialix 20070906)
1373
args = list(shlex.split(str(args)))
1374
return self._run_bzr_core(args, retcode=retcode,
1375
encoding=encoding, stdin=stdin, working_dir=working_dir,
1378
def _run_bzr_core(self, args, retcode, encoding, stdin,
1380
746
if encoding is None:
1381
747
encoding = bzrlib.user_encoding
748
if stdin is not None:
749
stdin = StringIO(stdin)
1382
750
stdout = StringIOWrapper()
1383
751
stderr = StringIOWrapper()
1384
752
stdout.encoding = encoding
1385
753
stderr.encoding = encoding
1387
self.log('run bzr: %r', args)
755
self.log('run bzr: %r', argv)
1388
756
# FIXME: don't call into logging here
1389
757
handler = logging.StreamHandler(stderr)
1390
758
handler.setLevel(logging.INFO)
1391
759
logger = logging.getLogger('')
1392
760
logger.addHandler(handler)
1393
old_ui_factory = ui.ui_factory
1394
ui.ui_factory = TestUIFactory(stdin=stdin, stdout=stdout, stderr=stderr)
1397
if working_dir is not None:
1398
cwd = osutils.getcwd()
1399
os.chdir(working_dir)
761
old_ui_factory = bzrlib.ui.ui_factory
762
bzrlib.ui.ui_factory = bzrlib.tests.blackbox.TestUIFactory(
765
bzrlib.ui.ui_factory.stdin = stdin
1402
result = self.apply_redirected(ui.ui_factory.stdin,
1404
bzrlib.commands.run_bzr_catch_user_errors,
767
result = self.apply_redirected(stdin, stdout, stderr,
768
bzrlib.commands.run_bzr_catch_errors,
1407
771
logger.removeHandler(handler)
1408
ui.ui_factory = old_ui_factory
772
bzrlib.ui.ui_factory = old_ui_factory
1412
774
out = stdout.getvalue()
1413
775
err = stderr.getvalue()
1503
843
handling, or early startup code, etc. Subprocess code can't be
1504
844
profiled or debugged so easily.
1506
:keyword retcode: The status code that is expected. Defaults to 0. If
1507
None is supplied, the status code is not checked.
1508
:keyword env_changes: A dictionary which lists changes to environment
1509
variables. A value of None will unset the env variable.
1510
The values must be strings. The change will only occur in the
1511
child, so you don't need to fix the environment after running.
1512
:keyword universal_newlines: Convert CRLF => LF
1513
:keyword allow_plugins: By default the subprocess is run with
1514
--no-plugins to ensure test reproducibility. Also, it is possible
1515
for system-wide plugins to create unexpected output on stderr,
1516
which can cause unnecessary test failures.
846
:param retcode: The status code that is expected. Defaults to 0. If
847
None is supplied, the status code is not checked.
1518
env_changes = kwargs.get('env_changes', {})
1519
working_dir = kwargs.get('working_dir', None)
1520
allow_plugins = kwargs.get('allow_plugins', False)
1522
if isinstance(args[0], list):
1524
elif isinstance(args[0], basestring):
1525
args = list(shlex.split(args[0]))
1527
raise ValueError("passing varargs to run_bzr_subprocess")
1528
process = self.start_bzr_subprocess(args, env_changes=env_changes,
1529
working_dir=working_dir,
1530
allow_plugins=allow_plugins)
1531
# We distinguish between retcode=None and retcode not passed.
849
bzr_path = os.path.dirname(os.path.dirname(bzrlib.__file__))+'/bzr'
851
process = Popen([sys.executable, bzr_path]+args, stdout=PIPE,
853
out = process.stdout.read()
854
err = process.stderr.read()
855
retcode = process.wait()
1532
856
supplied_retcode = kwargs.get('retcode', 0)
1533
return self.finish_bzr_subprocess(process, retcode=supplied_retcode,
1534
universal_newlines=kwargs.get('universal_newlines', False),
1537
def start_bzr_subprocess(self, process_args, env_changes=None,
1538
skip_if_plan_to_signal=False,
1540
allow_plugins=False):
1541
"""Start bzr in a subprocess for testing.
1543
This starts a new Python interpreter and runs bzr in there.
1544
This should only be used for tests that have a justifiable need for
1545
this isolation: e.g. they are testing startup time, or signal
1546
handling, or early startup code, etc. Subprocess code can't be
1547
profiled or debugged so easily.
1549
:param process_args: a list of arguments to pass to the bzr executable,
1550
for example ``['--version']``.
1551
:param env_changes: A dictionary which lists changes to environment
1552
variables. A value of None will unset the env variable.
1553
The values must be strings. The change will only occur in the
1554
child, so you don't need to fix the environment after running.
1555
:param skip_if_plan_to_signal: raise TestSkipped when true and os.kill
1557
:param allow_plugins: If False (default) pass --no-plugins to bzr.
1559
:returns: Popen object for the started process.
1561
if skip_if_plan_to_signal:
1562
if not getattr(os, 'kill', None):
1563
raise TestSkipped("os.kill not available.")
1565
if env_changes is None:
1569
def cleanup_environment():
1570
for env_var, value in env_changes.iteritems():
1571
old_env[env_var] = osutils.set_or_unset_env(env_var, value)
1573
def restore_environment():
1574
for env_var, value in old_env.iteritems():
1575
osutils.set_or_unset_env(env_var, value)
1577
bzr_path = self.get_bzr_path()
1580
if working_dir is not None:
1581
cwd = osutils.getcwd()
1582
os.chdir(working_dir)
1585
# win32 subprocess doesn't support preexec_fn
1586
# so we will avoid using it on all platforms, just to
1587
# make sure the code path is used, and we don't break on win32
1588
cleanup_environment()
1589
command = [sys.executable, bzr_path]
1590
if not allow_plugins:
1591
command.append('--no-plugins')
1592
command.extend(process_args)
1593
process = self._popen(command, stdin=PIPE, stdout=PIPE, stderr=PIPE)
1595
restore_environment()
1601
def _popen(self, *args, **kwargs):
1602
"""Place a call to Popen.
1604
Allows tests to override this method to intercept the calls made to
1605
Popen for introspection.
1607
return Popen(*args, **kwargs)
1609
def get_bzr_path(self):
1610
"""Return the path of the 'bzr' executable for this test suite."""
1611
bzr_path = os.path.dirname(os.path.dirname(bzrlib.__file__))+'/bzr'
1612
if not os.path.isfile(bzr_path):
1613
# We are probably installed. Assume sys.argv is the right file
1614
bzr_path = sys.argv[0]
1617
def finish_bzr_subprocess(self, process, retcode=0, send_signal=None,
1618
universal_newlines=False, process_args=None):
1619
"""Finish the execution of process.
1621
:param process: the Popen object returned from start_bzr_subprocess.
1622
:param retcode: The status code that is expected. Defaults to 0. If
1623
None is supplied, the status code is not checked.
1624
:param send_signal: an optional signal to send to the process.
1625
:param universal_newlines: Convert CRLF => LF
1626
:returns: (stdout, stderr)
1628
if send_signal is not None:
1629
os.kill(process.pid, send_signal)
1630
out, err = process.communicate()
1632
if universal_newlines:
1633
out = out.replace('\r\n', '\n')
1634
err = err.replace('\r\n', '\n')
1636
if retcode is not None and retcode != process.returncode:
1637
if process_args is None:
1638
process_args = "(unknown args)"
1639
mutter('Output of bzr %s:\n%s', process_args, out)
1640
mutter('Error for bzr %s:\n%s', process_args, err)
1641
self.fail('Command bzr %s failed with retcode %s != %s'
1642
% (process_args, retcode, process.returncode))
857
if supplied_retcode is not None:
858
assert supplied_retcode == retcode
1643
859
return [out, err]
1645
861
def check_inventory_shape(self, inv, shape):
1694
910
sys.stderr = real_stderr
1695
911
sys.stdin = real_stdin
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()
913
def merge(self, branch_from, wt_to):
914
"""A helper for tests to do a ui-less merge.
916
This should move to the main library when someone has time to integrate
919
# minimal ui-less merge.
920
wt_to.branch.fetch(branch_from)
921
base_rev = common_ancestor(branch_from.last_revision(),
922
wt_to.branch.last_revision(),
923
wt_to.branch.repository)
924
merge_inner(wt_to.branch, branch_from.basis_tree(),
925
wt_to.branch.repository.revision_tree(base_rev),
927
wt_to.add_pending_merge(branch_from.last_revision())
930
BzrTestBase = TestCase
2010
class TestCaseInTempDir(TestCaseWithMemoryTransport):
933
class TestCaseInTempDir(TestCase):
2011
934
"""Derived class that runs a test within a temporary directory.
2013
936
This is useful for tests that need to create a branch, etc.
2230
1241
def setUp(self):
2231
1242
super(ChrootedTestCase, self).setUp()
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.
1243
if not self.transport_server == bzrlib.transport.memory.MemoryServer:
1244
self.transport_readonly_server = bzrlib.transport.http.HttpServer
1247
def filter_suite_by_re(suite, pattern):
1248
result = TestUtil.TestSuite()
2242
1249
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)
2315
def filter_suite_by_re(suite, 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
2399
for test in iter_suite_tests(suite):
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))
1250
for test in iter_suite_tests(suite):
1251
if filter_re.search(test.id()):
1252
result.addTest(test)
2422
1256
def run_suite(suite, name='test', verbose=False, pattern=".*",
2423
stop_on_failure=False,
2424
transport=None, lsprof_timed=None, bench_history=None,
2425
matching_tests_first=None,
2428
exclude_pattern=None,
1257
stop_on_failure=False, keep_output=False,
1258
transport=None, lsprof_timed=None, bench_history=None):
1259
TestCaseInTempDir._TEST_NAME = name
2430
1260
TestCase._gather_lsprof_in_benchmarks = lsprof_timed
1266
pb = progress.ProgressBar()
2435
1267
runner = TextTestRunner(stream=sys.stdout,
2436
1268
descriptions=0,
2437
1269
verbosity=verbosity,
2438
bench_history=bench_history,
2439
list_only=list_only,
1270
keep_output=keep_output,
1272
bench_history=bench_history)
2441
1273
runner.stop_on_failure=stop_on_failure
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))
1275
suite = filter_suite_by_re(suite, pattern)
2472
1276
result = runner.run(suite)
2475
return result.wasStrictlySuccessful()
2477
1277
return result.wasSuccessful()
2480
# Controlled by "bzr selftest -E=..." option
2481
selftest_debug_flags = set()
2484
1280
def selftest(verbose=False, pattern=".*", stop_on_failure=True,
2485
1282
transport=None,
2486
1283
test_suite_factory=None,
2487
1284
lsprof_timed=None,
2489
matching_tests_first=None,
2492
exclude_pattern=None,
1285
bench_history=None):
2498
1286
"""Run the whole test suite under the enhanced runner"""
2499
1287
# XXX: Very ugly way to do this...
2500
1288
# Disable warning about old formats because we don't want it to disturb
2507
1295
transport = default_transport
2508
1296
old_transport = default_transport
2509
1297
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)
2519
1299
if test_suite_factory is None:
2520
suite = test_suite(keep_only, starting_with)
1300
suite = test_suite()
2522
1302
suite = test_suite_factory()
2523
1303
return run_suite(suite, 'testbzr', verbose=verbose, pattern=pattern,
2524
stop_on_failure=stop_on_failure,
1304
stop_on_failure=stop_on_failure, keep_output=keep_output,
2525
1305
transport=transport,
2526
1306
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,
1307
bench_history=bench_history)
2534
1309
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):
2641
1313
"""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
2648
1315
This function can be replaced if you need to change the default test
2649
1316
suite on a global basis, but it is not encouraged.
2651
1318
testmod_names = [
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',
2666
1319
'bzrlib.tests.test_ancestry',
2667
'bzrlib.tests.test_annotate',
2668
1320
'bzrlib.tests.test_api',
2669
1321
'bzrlib.tests.test_atomicfile',
2670
1322
'bzrlib.tests.test_bad_files',
2671
'bzrlib.tests.test_bisect_multi',
2672
1323
'bzrlib.tests.test_branch',
2673
'bzrlib.tests.test_branchbuilder',
2674
'bzrlib.tests.test_bugtracker',
2675
1324
'bzrlib.tests.test_bundle',
2676
1325
'bzrlib.tests.test_bzrdir',
2677
1326
'bzrlib.tests.test_cache_utf8',
2678
'bzrlib.tests.test_commands',
1327
'bzrlib.tests.test_command',
2679
1328
'bzrlib.tests.test_commit',
2680
1329
'bzrlib.tests.test_commit_merge',
2681
1330
'bzrlib.tests.test_config',
2682
1331
'bzrlib.tests.test_conflicts',
2683
'bzrlib.tests.test_counted_lock',
2684
1332
'bzrlib.tests.test_decorators',
2685
'bzrlib.tests.test_delta',
2686
'bzrlib.tests.test_deprecated_graph',
2687
1333
'bzrlib.tests.test_diff',
2688
'bzrlib.tests.test_dirstate',
2689
'bzrlib.tests.test_directory_service',
2690
'bzrlib.tests.test_email_message',
1334
'bzrlib.tests.test_doc_generate',
2691
1335
'bzrlib.tests.test_errors',
2692
1336
'bzrlib.tests.test_escaped_store',
2693
'bzrlib.tests.test_extract',
2694
1337
'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',
2699
1338
'bzrlib.tests.test_gpg',
2700
1339
'bzrlib.tests.test_graph',
2701
1340
'bzrlib.tests.test_hashcache',
2702
'bzrlib.tests.test_help',
2703
'bzrlib.tests.test_hooks',
2704
1341
'bzrlib.tests.test_http',
2705
'bzrlib.tests.test_http_implementations',
2706
1342
'bzrlib.tests.test_http_response',
2707
'bzrlib.tests.test_https_ca_bundle',
2708
1343
'bzrlib.tests.test_identitymap',
2709
1344
'bzrlib.tests.test_ignores',
2710
'bzrlib.tests.test_index',
2711
'bzrlib.tests.test_info',
2712
1345
'bzrlib.tests.test_inv',
2713
1346
'bzrlib.tests.test_knit',
2714
'bzrlib.tests.test_lazy_import',
2715
'bzrlib.tests.test_lazy_regex',
2716
1347
'bzrlib.tests.test_lockdir',
2717
1348
'bzrlib.tests.test_lockable_files',
2718
1349
'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',
2723
1350
'bzrlib.tests.test_merge',
2724
1351
'bzrlib.tests.test_merge3',
2725
1352
'bzrlib.tests.test_merge_core',
2726
'bzrlib.tests.test_merge_directive',
2727
1353
'bzrlib.tests.test_missing',
2728
1354
'bzrlib.tests.test_msgeditor',
2729
'bzrlib.tests.test_multiparent',
2730
'bzrlib.tests.test_mutabletree',
2731
1355
'bzrlib.tests.test_nonascii',
2732
1356
'bzrlib.tests.test_options',
2733
1357
'bzrlib.tests.test_osutils',
2734
'bzrlib.tests.test_osutils_encodings',
2735
'bzrlib.tests.test_pack',
2736
1358
'bzrlib.tests.test_patch',
2737
1359
'bzrlib.tests.test_patches',
2738
1360
'bzrlib.tests.test_permissions',
2739
1361
'bzrlib.tests.test_plugins',
2740
1362
'bzrlib.tests.test_progress',
2741
'bzrlib.tests.test_read_bundle',
2742
'bzrlib.tests.test_reconfigure',
2743
1363
'bzrlib.tests.test_reconcile',
2744
'bzrlib.tests.test_registry',
2745
'bzrlib.tests.test_remote',
2746
1364
'bzrlib.tests.test_repository',
2747
'bzrlib.tests.test_revert',
2748
1365
'bzrlib.tests.test_revision',
2749
'bzrlib.tests.test_revisionspec',
1366
'bzrlib.tests.test_revisionnamespaces',
2750
1367
'bzrlib.tests.test_revisiontree',
2751
1368
'bzrlib.tests.test_rio',
2752
1369
'bzrlib.tests.test_sampler',
2753
1370
'bzrlib.tests.test_selftest',
2754
1371
'bzrlib.tests.test_setup',
2755
1372
'bzrlib.tests.test_sftp_transport',
2756
'bzrlib.tests.test_smart',
1373
'bzrlib.tests.test_ftp_transport',
2757
1374
'bzrlib.tests.test_smart_add',
2758
'bzrlib.tests.test_smart_transport',
2759
'bzrlib.tests.test_smtp_connection',
2760
1375
'bzrlib.tests.test_source',
2761
'bzrlib.tests.test_ssh_transport',
2762
1376
'bzrlib.tests.test_status',
2763
1377
'bzrlib.tests.test_store',
2764
'bzrlib.tests.test_strace',
2765
'bzrlib.tests.test_subsume',
2766
'bzrlib.tests.test_switch',
2767
1378
'bzrlib.tests.test_symbol_versioning',
2768
'bzrlib.tests.test_tag',
2769
1379
'bzrlib.tests.test_testament',
2770
1380
'bzrlib.tests.test_textfile',
2771
1381
'bzrlib.tests.test_textmerge',
2772
'bzrlib.tests.test_timestamp',
2773
1382
'bzrlib.tests.test_trace',
2774
1383
'bzrlib.tests.test_transactions',
2775
1384
'bzrlib.tests.test_transform',
2776
1385
'bzrlib.tests.test_transport',
2777
'bzrlib.tests.test_transport_implementations',
2778
1386
'bzrlib.tests.test_tree',
2779
'bzrlib.tests.test_treebuilder',
2780
1387
'bzrlib.tests.test_tsort',
2781
1388
'bzrlib.tests.test_tuned_gzip',
2782
1389
'bzrlib.tests.test_ui',
2783
'bzrlib.tests.test_uncommit',
2784
1390
'bzrlib.tests.test_upgrade',
2785
1391
'bzrlib.tests.test_urlutils',
2786
1392
'bzrlib.tests.test_versionedfile',
2787
1393
'bzrlib.tests.test_version',
2788
'bzrlib.tests.test_version_info',
2789
1394
'bzrlib.tests.test_weave',
2790
1395
'bzrlib.tests.test_whitebox',
2791
'bzrlib.tests.test_win32utils',
2792
1396
'bzrlib.tests.test_workingtree',
2793
'bzrlib.tests.test_workingtree_4',
2794
'bzrlib.tests.test_wsgi',
2795
1397
'bzrlib.tests.test_xml',
2796
'bzrlib.tests.tree_implementations',
2797
'bzrlib.tests.workingtree_implementations',
1399
test_transport_implementations = [
1400
'bzrlib.tests.test_transport_implementations',
1401
'bzrlib.tests.test_read_bundle',
1403
suite = TestUtil.TestSuite()
2800
1404
loader = TestUtil.TestLoader()
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
2829
1405
suite.addTest(loader.loadTestsFromModuleNames(testmod_names))
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]
1406
from bzrlib.transport import TransportTestProviderAdapter
1407
adapter = TransportTestProviderAdapter()
1408
adapt_modules(test_transport_implementations, adapter, loader, suite)
1409
for package in packages_to_test():
1410
suite.addTest(package.test_suite())
1411
for m in MODULES_TO_TEST:
1412
suite.addTest(loader.loadTestsFromModule(m))
1413
for m in MODULES_TO_DOCTEST:
1414
suite.addTest(doctest.DocTestSuite(m))
1415
for name, plugin in bzrlib.plugin.all_plugins().items():
1416
if getattr(plugin, 'test_suite', None) is not None:
1417
suite.addTest(plugin.test_suite())
2965
1421
def adapt_modules(mods_list, adapter, loader, suite):
2966
1422
"""Adapt the modules in mods_list using adapter and add to suite."""
2967
1423
for test in iter_suite_tests(loader.loadTestsFromModuleNames(mods_list)):
2968
1424
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
