26
26
# general style of bzrlib. Please continue that consistency when adding e.g.
27
27
# new assertFoo() methods.
30
31
from cStringIO import StringIO
37
from pprint import pformat
42
from subprocess import Popen, PIPE
44
62
import bzrlib.branch
45
import bzrlib.bzrdir as bzrdir
46
63
import bzrlib.commands
47
import bzrlib.errors as errors
64
import bzrlib.timestamp
48
66
import bzrlib.inventory
49
67
import bzrlib.iterablefile
50
68
import bzrlib.lockdir
72
# lsprof not available
51
74
from bzrlib.merge import merge_inner
52
75
import bzrlib.merge3
54
import bzrlib.osutils as osutils
55
76
import bzrlib.plugin
56
from bzrlib.revision import common_ancestor
57
77
import bzrlib.store
78
from bzrlib import symbol_versioning
79
from bzrlib.symbol_versioning import (
58
85
import bzrlib.trace
59
from bzrlib.transport import urlescape, get_transport
86
from bzrlib.transport import get_transport
60
87
import bzrlib.transport
61
from bzrlib.transport.local import LocalRelpathServer
88
from bzrlib.transport.local import LocalURLServer
89
from bzrlib.transport.memory import MemoryServer
62
90
from bzrlib.transport.readonly import ReadonlyServer
63
from bzrlib.trace import mutter
64
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 (
65
98
from bzrlib.tests.treeshape import build_tree_contents
99
import bzrlib.version_info_formats.format_custom
66
100
from bzrlib.workingtree import WorkingTree, WorkingTreeFormat2
68
default_transport = LocalRelpathServer
71
MODULES_TO_DOCTEST = [
83
def packages_to_test():
84
"""Return a list of packages to test.
86
The packages are not globally imported so that import failures are
87
triggered when running selftest, not when importing the command.
90
import bzrlib.tests.blackbox
91
import bzrlib.tests.branch_implementations
92
import bzrlib.tests.bzrdir_implementations
93
import bzrlib.tests.interrepository_implementations
94
import bzrlib.tests.interversionedfile_implementations
95
import bzrlib.tests.repository_implementations
96
import bzrlib.tests.revisionstore_implementations
97
import bzrlib.tests.workingtree_implementations
100
bzrlib.tests.blackbox,
101
bzrlib.tests.branch_implementations,
102
bzrlib.tests.bzrdir_implementations,
103
bzrlib.tests.interrepository_implementations,
104
bzrlib.tests.interversionedfile_implementations,
105
bzrlib.tests.repository_implementations,
106
bzrlib.tests.revisionstore_implementations,
107
bzrlib.tests.workingtree_implementations,
111
class _MyResult(unittest._TextTestResult):
112
"""Custom TestResult.
114
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.
116
126
stop_early = False
118
def _elapsedTime(self):
119
return "%5dms" % (1000 * (time.time() - self._start_time))
128
def __init__(self, stream, descriptions, verbosity,
132
"""Construct new TestResult.
134
:param bench_history: Optionally, a writable file object to accumulate
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()
164
def _extractBenchmarkTime(self, testCase):
165
"""Add a benchmark time for the current test case."""
166
return getattr(testCase, "_benchtime", None)
168
def _elapsedTestTimeString(self):
169
"""Return a time string for the overall time the current test has taken."""
170
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:
176
self._formatTime(benchmark_time),
177
self._elapsedTestTimeString())
179
return " %s" % self._elapsedTestTimeString()
181
def _formatTime(self, seconds):
182
"""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)
187
def _shortened_test_description(self, test):
189
what = re.sub(r'^bzrlib\.(tests|benchmarks)\.', '', what)
121
192
def startTest(self, test):
122
193
unittest.TestResult.startTest(self, test)
123
# In a short description, the important words are in
124
# the beginning, but in an id, the important words are
126
SHOW_DESCRIPTIONS = False
128
width = osutils.terminal_width()
129
name_width = width - 15
131
if SHOW_DESCRIPTIONS:
132
what = test.shortDescription()
134
if len(what) > name_width:
135
what = what[:name_width-3] + '...'
138
if what.startswith('bzrlib.tests.'):
140
if len(what) > name_width:
141
what = '...' + what[3-name_width:]
142
what = what.ljust(name_width)
143
self.stream.write(what)
194
self.report_test_start(test)
195
test.number = self.count
196
self._recordTestStartTime()
198
def _recordTestStartTime(self):
199
"""Record that a test has started."""
145
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:
147
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)
148
216
if isinstance(err[1], TestSkipped):
149
return self.addSkipped(test, err)
150
unittest.TestResult.addError(self, test, err)
152
self.stream.writeln("ERROR %s" % self._elapsedTime())
154
self.stream.write('E')
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)
159
228
def addFailure(self, test, err):
160
unittest.TestResult.addFailure(self, test, err)
162
self.stream.writeln(" FAIL %s" % self._elapsedTime())
164
self.stream.write('F')
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)
169
245
def addSuccess(self, test):
171
self.stream.writeln(' OK %s' % self._elapsedTime())
173
self.stream.write('~')
175
unittest.TestResult.addSuccess(self, test)
177
def addSkipped(self, test, skip_excinfo):
179
print >>self.stream, ' SKIP %s' % self._elapsedTime()
180
print >>self.stream, ' %s' % skip_excinfo[1]
182
self.stream.write('S')
184
# seems best to treat this as success from point-of-view of unittest
185
# -- it actually does nothing so it barely matters :)
186
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 = ''
188
305
def printErrorList(self, flavour, errors):
189
306
for test, err in errors:
190
307
self.stream.writeln(self.separator1)
191
self.stream.writeln("%s: %s" % (flavour, self.getDescription(test)))
308
self.stream.write("%s: " % flavour)
309
self.stream.writeln(self.getDescription(test))
192
310
if getattr(test, '_get_log', None) is not None:
194
print >>self.stream, \
195
('vvvv[log from %s]' % test.id()).ljust(78,'-')
196
print >>self.stream, test._get_log()
197
print >>self.stream, \
198
('^^^^[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')
199
320
self.stream.writeln(self.separator2)
200
321
self.stream.writeln("%s" % err)
203
class TextTestRunner(unittest.TextTestRunner):
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
class TextTestRunner(object):
204
502
stop_on_failure = False
206
def _makeResult(self):
207
result = _MyResult(self.stream, self.descriptions, self.verbosity)
511
self.stream = unittest._WritelnDecorator(stream)
512
self.descriptions = descriptions
513
self.verbosity = verbosity
514
self._bench_history = bench_history
515
self.list_only = list_only
518
"Run the given test case or test suite."
519
startTime = time.time()
520
if self.verbosity == 1:
521
result_class = TextTestResult
522
elif self.verbosity >= 2:
523
result_class = VerboseTestResult
524
result = result_class(self.stream,
527
bench_history=self._bench_history,
528
num_tests=test.countTestCases(),
208
530
result.stop_early = self.stop_on_failure
531
result.report_starting()
533
if self.verbosity >= 2:
534
self.stream.writeln("Listing tests only ...\n")
536
for t in iter_suite_tests(test):
537
self.stream.writeln("%s" % (t.id()))
539
actionTaken = "Listed"
542
run = result.testsRun
544
stopTime = time.time()
545
timeTaken = stopTime - startTime
547
self.stream.writeln(result.separator2)
548
self.stream.writeln("%s %d test%s in %.3fs" % (actionTaken,
549
run, run != 1 and "s" or "", timeTaken))
550
self.stream.writeln()
551
if not result.wasSuccessful():
552
self.stream.write("FAILED (")
553
failed, errored = map(len, (result.failures, result.errors))
555
self.stream.write("failures=%d" % failed)
557
if failed: self.stream.write(", ")
558
self.stream.write("errors=%d" % errored)
559
if result.known_failure_count:
560
if failed or errored: self.stream.write(", ")
561
self.stream.write("known_failure_count=%d" %
562
result.known_failure_count)
563
self.stream.writeln(")")
565
if result.known_failure_count:
566
self.stream.writeln("OK (known_failures=%d)" %
567
result.known_failure_count)
569
self.stream.writeln("OK")
570
if result.skip_count > 0:
571
skipped = result.skip_count
572
self.stream.writeln('%d test%s skipped' %
573
(skipped, skipped != 1 and "s" or ""))
574
if result.unsupported:
575
for feature, count in sorted(result.unsupported.items()):
576
self.stream.writeln("Missing feature '%s' skipped %d tests." %
338
976
self.assertEqual(mode, actual_mode,
339
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))
341
985
def assertIsInstance(self, obj, kls):
342
986
"""Fail if obj is not an instance of kls"""
343
987
if not isinstance(obj, kls):
344
self.fail("%r is not an instance of %s" % (obj, kls))
988
self.fail("%r is an instance of %s rather than %s" % (
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)
346
1165
def _startLogFile(self):
347
1166
"""Send bzr and test log messages to a temporary file.
424
1312
# TODO: Perhaps this should keep running cleanups even if
425
1313
# one of them fails?
426
for cleanup_fn in reversed(self._cleanups):
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()()
429
1320
def log(self, *args):
433
"""Return as a string the log for this test"""
434
if self._log_file_name:
435
return open(self._log_file_name).read()
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
437
1338
return self._log_contents
438
# TODO: Delete the log after it's been read in
440
def capture(self, cmd, retcode=0):
441
"""Shortcut that splits cmd into words, runs, and returns stdout"""
442
return self.run_bzr_captured(cmd.split(), retcode=retcode)[0]
444
def run_bzr_captured(self, argv, retcode=0):
445
"""Invoke bzr and return (stdout, stderr).
447
Useful for code that wants to check the contents of the
448
output, the way error messages are presented, etc.
450
This should be the main method for tests that want to exercise the
451
overall behavior of the bzr application (rather than a unit test
452
or a functional test of the library.)
454
Much of the old code runs bzr by forking a new copy of Python, but
455
that is slower, harder to debug, and generally not necessary.
457
This runs bzr through the interface that catches and reports
458
errors, and with logging set to something approximating the
459
default, so that error reporting can be checked.
461
argv -- arguments to invoke bzr
462
retcode -- expected return code, or None for don't-care.
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.
466
self.log('run bzr: %s', ' '.join(argv))
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
if encoding is None:
1381
encoding = bzrlib.user_encoding
1382
stdout = StringIOWrapper()
1383
stderr = StringIOWrapper()
1384
stdout.encoding = encoding
1385
stderr.encoding = encoding
1387
self.log('run bzr: %r', args)
467
1388
# FIXME: don't call into logging here
468
1389
handler = logging.StreamHandler(stderr)
469
handler.setFormatter(bzrlib.trace.QuietFormatter())
470
1390
handler.setLevel(logging.INFO)
471
1391
logger = logging.getLogger('')
472
1392
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)
474
result = self.apply_redirected(None, stdout, stderr,
475
bzrlib.commands.run_bzr_catch_errors,
1402
result = self.apply_redirected(ui.ui_factory.stdin,
1404
bzrlib.commands.run_bzr_catch_user_errors,
478
1407
logger.removeHandler(handler)
1408
ui.ui_factory = old_ui_factory
479
1412
out = stdout.getvalue()
480
1413
err = stderr.getvalue()
482
self.log('output:\n%s', out)
1415
self.log('output:\n%r', out)
484
self.log('errors:\n%s', err)
1417
self.log('errors:\n%r', err)
485
1418
if retcode is not None:
486
self.assertEquals(result, retcode)
1419
self.assertEquals(retcode, result,
1420
message='Unexpected return code')
489
def run_bzr(self, *args, **kwargs):
1423
def run_bzr(self, args, retcode=0, encoding=None, stdin=None,
1424
working_dir=None, error_regexes=[], output_encoding=None):
490
1425
"""Invoke bzr, as if it were run from the command line.
1427
The argument list should not include the bzr program name - the
1428
first argument is normally the bzr command. Arguments may be
1429
passed in three ways:
1431
1- A list of strings, eg ["commit", "a"]. This is recommended
1432
when the command contains whitespace or metacharacters, or
1433
is built up at run time.
1435
2- A single string, eg "add a". This is the most convenient
1436
for hardcoded commands.
1438
This runs bzr through the interface that catches and reports
1439
errors, and with logging set to something approximating the
1440
default, so that error reporting can be checked.
492
1442
This should be the main method for tests that want to exercise the
493
1443
overall behavior of the bzr application (rather than a unit test
494
1444
or a functional test of the library.)
496
1446
This sends the stdout/stderr results into the test's log,
497
1447
where it may be useful for debugging. See also run_captured.
499
retcode = kwargs.pop('retcode', 0)
500
return self.run_bzr_captured(args, retcode)
1449
:keyword stdin: A string to be used as stdin for the command.
1450
:keyword retcode: The status code the command should return;
1452
:keyword working_dir: The directory to run the command in
1453
:keyword error_regexes: A list of expected error messages. If
1454
specified they must be seen in the error output of the command.
1456
out, err = self._run_bzr_autosplit(
1461
working_dir=working_dir,
1463
for regex in error_regexes:
1464
self.assertContainsRe(err, regex)
1467
def run_bzr_error(self, error_regexes, *args, **kwargs):
1468
"""Run bzr, and check that stderr contains the supplied regexes
1470
:param error_regexes: Sequence of regular expressions which
1471
must each be found in the error output. The relative ordering
1473
:param args: command-line arguments for bzr
1474
:param kwargs: Keyword arguments which are interpreted by run_bzr
1475
This function changes the default value of retcode to be 3,
1476
since in most cases this is run when you expect bzr to fail.
1478
:return: (out, err) The actual output of running the command (in case
1479
you want to do more inspection)
1483
# Make sure that commit is failing because there is nothing to do
1484
self.run_bzr_error(['no changes to commit'],
1485
['commit', '-m', 'my commit comment'])
1486
# Make sure --strict is handling an unknown file, rather than
1487
# giving us the 'nothing to do' error
1488
self.build_tree(['unknown'])
1489
self.run_bzr_error(['Commit refused because there are unknown files'],
1490
['commit', --strict', '-m', 'my commit comment'])
1492
kwargs.setdefault('retcode', 3)
1493
kwargs['error_regexes'] = error_regexes
1494
out, err = self.run_bzr(*args, **kwargs)
1497
def run_bzr_subprocess(self, *args, **kwargs):
1498
"""Run bzr in a subprocess for testing.
1500
This starts a new Python interpreter and runs bzr in there.
1501
This should only be used for tests that have a justifiable need for
1502
this isolation: e.g. they are testing startup time, or signal
1503
handling, or early startup code, etc. Subprocess code can't be
1504
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.
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.
1532
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))
502
1645
def check_inventory_shape(self, inv, shape):
503
1646
"""Compare an inventory to a list of expected names.
551
1694
sys.stderr = real_stderr
552
1695
sys.stdin = real_stdin
554
def merge(self, branch_from, wt_to):
555
"""A helper for tests to do a ui-less merge.
557
This should move to the main library when someone has time to integrate
560
# minimal ui-less merge.
561
wt_to.branch.fetch(branch_from)
562
base_rev = common_ancestor(branch_from.last_revision(),
563
wt_to.branch.last_revision(),
564
wt_to.branch.repository)
565
merge_inner(wt_to.branch, branch_from.basis_tree(),
566
wt_to.branch.repository.revision_tree(base_rev),
568
wt_to.add_pending_merge(branch_from.last_revision())
571
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()
574
class TestCaseInTempDir(TestCase):
2010
class TestCaseInTempDir(TestCaseWithMemoryTransport):
575
2011
"""Derived class that runs a test within a temporary directory.
577
2013
This is useful for tests that need to create a branch, etc.
868
2230
def setUp(self):
869
2231
super(ChrootedTestCase, self).setUp()
870
if not self.transport_server == bzrlib.transport.memory.MemoryServer:
871
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)
874
2315
def filter_suite_by_re(suite, pattern):
876
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
877
2399
for test in iter_suite_tests(suite):
878
if filter_re.search(test.id()):
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))
883
2422
def run_suite(suite, name='test', verbose=False, pattern=".*",
884
stop_on_failure=False, keep_output=False,
886
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,
2430
TestCase._gather_lsprof_in_benchmarks = lsprof_timed
891
2435
runner = TextTestRunner(stream=sys.stdout,
2437
verbosity=verbosity,
2438
bench_history=bench_history,
2439
list_only=list_only,
894
2441
runner.stop_on_failure=stop_on_failure
896
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))
897
2472
result = runner.run(suite)
898
# This is still a little bogus,
899
# but only a little. Folk not using our testrunner will
900
# have to delete their temp directories themselves.
901
test_root = TestCaseInTempDir.TEST_ROOT
902
if result.wasSuccessful() or not keep_output:
903
if test_root is not None:
904
print 'Deleting test root %s...' % test_root
906
shutil.rmtree(test_root)
910
print "Failed tests working directories are in '%s'\n" % TestCaseInTempDir.TEST_ROOT
2475
return result.wasStrictlySuccessful()
911
2477
return result.wasSuccessful()
2480
# Controlled by "bzr selftest -E=..." option
2481
selftest_debug_flags = set()
914
2484
def selftest(verbose=False, pattern=".*", stop_on_failure=True,
2486
test_suite_factory=None,
2489
matching_tests_first=None,
2492
exclude_pattern=None,
917
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
918
2505
global default_transport
919
2506
if transport is None:
920
2507
transport = default_transport
921
2508
old_transport = default_transport
922
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)
2519
if test_suite_factory is None:
2520
suite = test_suite(keep_only, starting_with)
2522
suite = test_suite_factory()
925
2523
return run_suite(suite, 'testbzr', verbose=verbose, pattern=pattern,
926
stop_on_failure=stop_on_failure, keep_output=keep_output,
2524
stop_on_failure=stop_on_failure,
2525
transport=transport,
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,
929
2534
default_transport = old_transport
934
"""Build and return TestSuite for the whole program."""
935
from doctest import DocTestSuite
937
global MODULES_TO_DOCTEST
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
"""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
This function can be replaced if you need to change the default test
2649
suite on a global basis, but it is not encouraged.
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',
940
2666
'bzrlib.tests.test_ancestry',
941
2667
'bzrlib.tests.test_annotate',
942
2668
'bzrlib.tests.test_api',
2669
'bzrlib.tests.test_atomicfile',
943
2670
'bzrlib.tests.test_bad_files',
944
'bzrlib.tests.test_basis_inventory',
2671
'bzrlib.tests.test_bisect_multi',
945
2672
'bzrlib.tests.test_branch',
2673
'bzrlib.tests.test_branchbuilder',
2674
'bzrlib.tests.test_bugtracker',
2675
'bzrlib.tests.test_bundle',
946
2676
'bzrlib.tests.test_bzrdir',
947
'bzrlib.tests.test_command',
2677
'bzrlib.tests.test_cache_utf8',
2678
'bzrlib.tests.test_commands',
948
2679
'bzrlib.tests.test_commit',
949
2680
'bzrlib.tests.test_commit_merge',
950
2681
'bzrlib.tests.test_config',
951
2682
'bzrlib.tests.test_conflicts',
2683
'bzrlib.tests.test_counted_lock',
952
2684
'bzrlib.tests.test_decorators',
2685
'bzrlib.tests.test_delta',
2686
'bzrlib.tests.test_deprecated_graph',
953
2687
'bzrlib.tests.test_diff',
954
'bzrlib.tests.test_doc_generate',
2688
'bzrlib.tests.test_dirstate',
2689
'bzrlib.tests.test_directory_service',
2690
'bzrlib.tests.test_email_message',
955
2691
'bzrlib.tests.test_errors',
956
2692
'bzrlib.tests.test_escaped_store',
2693
'bzrlib.tests.test_extract',
957
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',
958
2699
'bzrlib.tests.test_gpg',
959
2700
'bzrlib.tests.test_graph',
960
2701
'bzrlib.tests.test_hashcache',
2702
'bzrlib.tests.test_help',
2703
'bzrlib.tests.test_hooks',
961
2704
'bzrlib.tests.test_http',
2705
'bzrlib.tests.test_http_implementations',
2706
'bzrlib.tests.test_http_response',
2707
'bzrlib.tests.test_https_ca_bundle',
962
2708
'bzrlib.tests.test_identitymap',
2709
'bzrlib.tests.test_ignores',
2710
'bzrlib.tests.test_index',
2711
'bzrlib.tests.test_info',
963
2712
'bzrlib.tests.test_inv',
964
2713
'bzrlib.tests.test_knit',
2714
'bzrlib.tests.test_lazy_import',
2715
'bzrlib.tests.test_lazy_regex',
965
2716
'bzrlib.tests.test_lockdir',
966
2717
'bzrlib.tests.test_lockable_files',
967
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',
968
2723
'bzrlib.tests.test_merge',
969
2724
'bzrlib.tests.test_merge3',
970
2725
'bzrlib.tests.test_merge_core',
2726
'bzrlib.tests.test_merge_directive',
971
2727
'bzrlib.tests.test_missing',
972
2728
'bzrlib.tests.test_msgeditor',
2729
'bzrlib.tests.test_multiparent',
2730
'bzrlib.tests.test_mutabletree',
973
2731
'bzrlib.tests.test_nonascii',
974
2732
'bzrlib.tests.test_options',
975
2733
'bzrlib.tests.test_osutils',
2734
'bzrlib.tests.test_osutils_encodings',
2735
'bzrlib.tests.test_pack',
2736
'bzrlib.tests.test_patch',
2737
'bzrlib.tests.test_patches',
976
2738
'bzrlib.tests.test_permissions',
977
2739
'bzrlib.tests.test_plugins',
978
2740
'bzrlib.tests.test_progress',
2741
'bzrlib.tests.test_read_bundle',
2742
'bzrlib.tests.test_reconfigure',
979
2743
'bzrlib.tests.test_reconcile',
2744
'bzrlib.tests.test_registry',
2745
'bzrlib.tests.test_remote',
980
2746
'bzrlib.tests.test_repository',
2747
'bzrlib.tests.test_revert',
981
2748
'bzrlib.tests.test_revision',
982
'bzrlib.tests.test_revisionnamespaces',
983
'bzrlib.tests.test_revprops',
2749
'bzrlib.tests.test_revisionspec',
2750
'bzrlib.tests.test_revisiontree',
984
2751
'bzrlib.tests.test_rio',
985
2752
'bzrlib.tests.test_sampler',
986
2753
'bzrlib.tests.test_selftest',
987
2754
'bzrlib.tests.test_setup',
988
2755
'bzrlib.tests.test_sftp_transport',
2756
'bzrlib.tests.test_smart',
989
2757
'bzrlib.tests.test_smart_add',
2758
'bzrlib.tests.test_smart_transport',
2759
'bzrlib.tests.test_smtp_connection',
990
2760
'bzrlib.tests.test_source',
2761
'bzrlib.tests.test_ssh_transport',
2762
'bzrlib.tests.test_status',
991
2763
'bzrlib.tests.test_store',
2764
'bzrlib.tests.test_strace',
2765
'bzrlib.tests.test_subsume',
2766
'bzrlib.tests.test_switch',
992
2767
'bzrlib.tests.test_symbol_versioning',
2768
'bzrlib.tests.test_tag',
993
2769
'bzrlib.tests.test_testament',
2770
'bzrlib.tests.test_textfile',
994
2771
'bzrlib.tests.test_textmerge',
2772
'bzrlib.tests.test_timestamp',
995
2773
'bzrlib.tests.test_trace',
996
2774
'bzrlib.tests.test_transactions',
997
2775
'bzrlib.tests.test_transform',
998
2776
'bzrlib.tests.test_transport',
2777
'bzrlib.tests.test_transport_implementations',
2778
'bzrlib.tests.test_tree',
2779
'bzrlib.tests.test_treebuilder',
999
2780
'bzrlib.tests.test_tsort',
1000
2781
'bzrlib.tests.test_tuned_gzip',
1001
2782
'bzrlib.tests.test_ui',
2783
'bzrlib.tests.test_uncommit',
1002
2784
'bzrlib.tests.test_upgrade',
2785
'bzrlib.tests.test_urlutils',
1003
2786
'bzrlib.tests.test_versionedfile',
2787
'bzrlib.tests.test_version',
2788
'bzrlib.tests.test_version_info',
1004
2789
'bzrlib.tests.test_weave',
1005
2790
'bzrlib.tests.test_whitebox',
2791
'bzrlib.tests.test_win32utils',
1006
2792
'bzrlib.tests.test_workingtree',
2793
'bzrlib.tests.test_workingtree_4',
2794
'bzrlib.tests.test_wsgi',
1007
2795
'bzrlib.tests.test_xml',
2796
'bzrlib.tests.tree_implementations',
2797
'bzrlib.tests.workingtree_implementations',
1009
test_transport_implementations = [
1010
'bzrlib.tests.test_transport_implementations']
1012
TestCase.BZRPATH = osutils.pathjoin(
1013
osutils.realpath(osutils.dirname(bzrlib.__path__[0])), 'bzr')
1014
print '%10s: %s' % ('bzr', osutils.realpath(sys.argv[0]))
1015
print '%10s: %s' % ('bzrlib', bzrlib.__path__[0])
1018
# python2.4's TestLoader.loadTestsFromNames gives very poor
1019
# errors if it fails to load a named module - no indication of what's
1020
# actually wrong, just "no such module". We should probably override that
1021
# class, but for the moment just load them ourselves. (mbp 20051202)
1022
loader = TestLoader()
1023
from bzrlib.transport import TransportTestProviderAdapter
1024
adapter = TransportTestProviderAdapter()
1025
adapt_modules(test_transport_implementations, adapter, loader, suite)
1026
for mod_name in testmod_names:
1027
mod = _load_module_by_name(mod_name)
1028
suite.addTest(loader.loadTestsFromModule(mod))
1029
for package in packages_to_test():
1030
suite.addTest(package.test_suite())
1031
for m in MODULES_TO_TEST:
1032
suite.addTest(loader.loadTestsFromModule(m))
1033
for m in (MODULES_TO_DOCTEST):
1034
suite.addTest(DocTestSuite(m))
1035
for name, plugin in bzrlib.plugin.all_plugins().items():
1036
if getattr(plugin, 'test_suite', None) is not None:
1037
suite.addTest(plugin.test_suite())
2800
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
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]
1041
2965
def adapt_modules(mods_list, adapter, loader, suite):
1042
2966
"""Adapt the modules in mods_list using adapter and add to suite."""
1043
for mod_name in mods_list:
1044
mod = _load_module_by_name(mod_name)
1045
for test in iter_suite_tests(loader.loadTestsFromModule(mod)):
1046
suite.addTests(adapter.adapt(test))
1049
def _load_module_by_name(mod_name):
1050
parts = mod_name.split('.')
1051
module = __import__(mod_name)
1053
# for historical reasons python returns the top-level module even though
1054
# it loads the submodule; we need to walk down to get the one we want.
1056
module = getattr(module, parts.pop(0))
2967
for test in iter_suite_tests(loader.loadTestsFromModuleNames(mods_list)):
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
