80
59
from bzrlib.merge import merge_inner
81
60
import bzrlib.merge3
62
import bzrlib.osutils as osutils
82
63
import bzrlib.plugin
83
from bzrlib.smart import client, request, server
64
import bzrlib.progress as progress
65
from bzrlib.revision import common_ancestor
84
66
import bzrlib.store
85
from bzrlib import symbol_versioning
86
from bzrlib.symbol_versioning import (
92
67
import bzrlib.trace
93
68
from bzrlib.transport import get_transport
94
69
import bzrlib.transport
95
from bzrlib.transport.local import LocalURLServer
96
from bzrlib.transport.memory import MemoryServer
70
from bzrlib.transport.local import LocalRelpathServer
97
71
from bzrlib.transport.readonly import ReadonlyServer
98
from bzrlib.trace import mutter, note
72
from bzrlib.trace import mutter
99
73
from bzrlib.tests import TestUtil
100
from bzrlib.tests.http_server import HttpServer
101
74
from bzrlib.tests.TestUtil import (
105
78
from bzrlib.tests.treeshape import build_tree_contents
106
from bzrlib.ui import NullProgressView
107
from bzrlib.ui.text import TextUIFactory
108
import bzrlib.version_info_formats.format_custom
79
import bzrlib.urlutils as urlutils
109
80
from bzrlib.workingtree import WorkingTree, WorkingTreeFormat2
111
# Mark this python module as being part of the implementation
112
# of unittest: this gives us better tracebacks where the last
113
# shown frame is the test code, not our assertXYZ.
116
default_transport = LocalURLServer
118
# Subunit result codes, defined here to prevent a hard dependency on subunit.
123
class ExtendedTestResult(unittest._TextTestResult):
124
"""Accepts, reports and accumulates the results of running tests.
126
Compared to the unittest version this class adds support for
127
profiling, benchmarking, stopping as soon as a test fails, and
128
skipping tests. There are further-specialized subclasses for
129
different types of display.
131
When a test finishes, in whatever way, it calls one of the addSuccess,
132
addFailure or addError classes. These in turn may redirect to a more
133
specific case for the special test results supported by our extended
136
Note that just one of these objects is fed the results from many tests.
82
default_transport = LocalRelpathServer
85
MODULES_TO_DOCTEST = [
87
bzrlib.bundle.serializer,
100
def packages_to_test():
101
"""Return a list of packages to test.
103
The packages are not globally imported so that import failures are
104
triggered when running selftest, not when importing the command.
107
import bzrlib.tests.blackbox
108
import bzrlib.tests.branch_implementations
109
import bzrlib.tests.bzrdir_implementations
110
import bzrlib.tests.interrepository_implementations
111
import bzrlib.tests.interversionedfile_implementations
112
import bzrlib.tests.repository_implementations
113
import bzrlib.tests.revisionstore_implementations
114
import bzrlib.tests.workingtree_implementations
117
bzrlib.tests.blackbox,
118
bzrlib.tests.branch_implementations,
119
bzrlib.tests.bzrdir_implementations,
120
bzrlib.tests.interrepository_implementations,
121
bzrlib.tests.interversionedfile_implementations,
122
bzrlib.tests.repository_implementations,
123
bzrlib.tests.revisionstore_implementations,
124
bzrlib.tests.workingtree_implementations,
128
class _MyResult(unittest._TextTestResult):
129
"""Custom TestResult.
131
Shows output in a different format, including displaying runtime for tests.
139
133
stop_early = False
141
def __init__(self, stream, descriptions, verbosity,
145
"""Construct new TestResult.
147
:param bench_history: Optionally, a writable file object to accumulate
135
def __init__(self, stream, descriptions, verbosity, pb=None):
150
136
unittest._TextTestResult.__init__(self, stream, descriptions, verbosity)
151
if bench_history is not None:
152
from bzrlib.version import _get_bzr_source_tree
153
src_tree = _get_bzr_source_tree()
156
revision_id = src_tree.get_parent_ids()[0]
158
# XXX: if this is a brand new tree, do the same as if there
162
# XXX: If there's no branch, what should we do?
164
bench_history.write("--date %s %s\n" % (time.time(), revision_id))
165
self._bench_history = bench_history
166
self.ui = ui.ui_factory
169
self.failure_count = 0
170
self.known_failure_count = 0
172
self.not_applicable_count = 0
173
self.unsupported = {}
175
self._overall_start_time = time.time()
176
self._strict = strict
178
def stopTestRun(self):
181
stopTime = time.time()
182
timeTaken = stopTime - self.startTime
184
self.stream.writeln(self.separator2)
185
self.stream.writeln("%s %d test%s in %.3fs" % (actionTaken,
186
run, run != 1 and "s" or "", timeTaken))
187
self.stream.writeln()
188
if not self.wasSuccessful():
189
self.stream.write("FAILED (")
190
failed, errored = map(len, (self.failures, self.errors))
192
self.stream.write("failures=%d" % failed)
194
if failed: self.stream.write(", ")
195
self.stream.write("errors=%d" % errored)
196
if self.known_failure_count:
197
if failed or errored: self.stream.write(", ")
198
self.stream.write("known_failure_count=%d" %
199
self.known_failure_count)
200
self.stream.writeln(")")
202
if self.known_failure_count:
203
self.stream.writeln("OK (known_failures=%d)" %
204
self.known_failure_count)
206
self.stream.writeln("OK")
207
if self.skip_count > 0:
208
skipped = self.skip_count
209
self.stream.writeln('%d test%s skipped' %
210
(skipped, skipped != 1 and "s" or ""))
212
for feature, count in sorted(self.unsupported.items()):
213
self.stream.writeln("Missing feature '%s' skipped %d tests." %
216
ok = self.wasStrictlySuccessful()
218
ok = self.wasSuccessful()
219
if TestCase._first_thread_leaker_id:
221
'%s is leaking threads among %d leaking tests.\n' % (
222
TestCase._first_thread_leaker_id,
223
TestCase._leaking_threads_tests))
225
def _extractBenchmarkTime(self, testCase):
139
def extractBenchmarkTime(self, testCase):
226
140
"""Add a benchmark time for the current test case."""
227
return getattr(testCase, "_benchtime", None)
141
self._benchmarkTime = getattr(testCase, "_benchtime", None)
229
143
def _elapsedTestTimeString(self):
230
144
"""Return a time string for the overall time the current test has taken."""
231
145
return self._formatTime(time.time() - self._start_time)
233
def _testTimeString(self, testCase):
234
benchmark_time = self._extractBenchmarkTime(testCase)
235
if benchmark_time is not None:
236
return self._formatTime(benchmark_time) + "*"
147
def _testTimeString(self):
148
if self._benchmarkTime is not None:
150
self._formatTime(self._benchmarkTime),
151
self._elapsedTestTimeString())
238
return self._elapsedTestTimeString()
153
return " %s" % self._elapsedTestTimeString()
240
155
def _formatTime(self, seconds):
241
156
"""Format seconds as milliseconds with leading spaces."""
242
# some benchmarks can take thousands of seconds to run, so we need 8
244
return "%8dms" % (1000 * seconds)
157
return "%5dms" % (1000 * seconds)
246
def _shortened_test_description(self, test):
248
what = re.sub(r'^bzrlib\.(tests|benchmarks)\.', '', what)
159
def _ellipsise_unimportant_words(self, a_string, final_width,
161
"""Add ellipses (sp?) for overly long strings.
163
:param keep_start: If true preserve the start of a_string rather
167
if len(a_string) > final_width:
168
result = a_string[:final_width-3] + '...'
172
if len(a_string) > final_width:
173
result = '...' + a_string[3-final_width:]
176
return result.ljust(final_width)
251
178
def startTest(self, test):
252
179
unittest.TestResult.startTest(self, test)
255
self.report_test_start(test)
256
test.number = self.count
180
# In a short description, the important words are in
181
# the beginning, but in an id, the important words are
183
SHOW_DESCRIPTIONS = False
185
if not self.showAll and self.dots and self.pb is not None:
188
final_width = osutils.terminal_width()
189
final_width = final_width - 15 - 8
191
if SHOW_DESCRIPTIONS:
192
what = test.shortDescription()
194
what = self._ellipsise_unimportant_words(what, final_width, keep_start=True)
197
if what.startswith('bzrlib.tests.'):
199
what = self._ellipsise_unimportant_words(what, final_width)
201
self.stream.write(what)
202
elif self.dots and self.pb is not None:
203
self.pb.update(what, self.testsRun - 1, None)
257
205
self._recordTestStartTime()
259
def startTests(self):
261
if getattr(sys, 'frozen', None) is None:
262
bzr_path = osutils.realpath(sys.argv[0])
264
bzr_path = sys.executable
266
'testing: %s\n' % (bzr_path,))
269
bzrlib.__path__[0],))
271
' bzr-%s python-%s %s\n' % (
272
bzrlib.version_string,
273
bzrlib._format_version_tuple(sys.version_info),
274
platform.platform(aliased=1),
276
self.stream.write('\n')
278
207
def _recordTestStartTime(self):
279
208
"""Record that a test has started."""
280
209
self._start_time = time.time()
282
def _cleanupLogFile(self, test):
283
# We can only do this if we have one of our TestCases, not if
285
setKeepLogfile = getattr(test, 'setKeepLogfile', None)
286
if setKeepLogfile is not None:
289
211
def addError(self, test, err):
290
"""Tell result that test finished with an error.
292
Called from the TestCase run() method when the test
293
fails with an unexpected error.
295
self._testConcluded(test)
296
if isinstance(err[1], TestNotApplicable):
297
return self._addNotApplicable(test, err)
298
elif isinstance(err[1], UnavailableFeature):
299
return self.addNotSupported(test, err[1].args[0])
301
unittest.TestResult.addError(self, test, err)
302
self.error_count += 1
303
self.report_error(test, err)
306
self._cleanupLogFile(test)
212
if isinstance(err[1], TestSkipped):
213
return self.addSkipped(test, err)
214
unittest.TestResult.addError(self, test, err)
215
self.extractBenchmarkTime(test)
217
self.stream.writeln("ERROR %s" % self._testTimeString())
218
elif self.dots and self.pb is None:
219
self.stream.write('E')
221
self.pb.update(self._ellipsise_unimportant_words('ERROR', 13), self.testsRun, None)
308
226
def addFailure(self, test, err):
309
"""Tell result that test failed.
311
Called from the TestCase run() method when the test
312
fails because e.g. an assert() method failed.
314
self._testConcluded(test)
315
if isinstance(err[1], KnownFailure):
316
return self._addKnownFailure(test, err)
318
unittest.TestResult.addFailure(self, test, err)
319
self.failure_count += 1
320
self.report_failure(test, err)
323
self._cleanupLogFile(test)
227
unittest.TestResult.addFailure(self, test, err)
228
self.extractBenchmarkTime(test)
230
self.stream.writeln(" FAIL %s" % self._testTimeString())
231
elif self.dots and self.pb is None:
232
self.stream.write('F')
234
self.pb.update(self._ellipsise_unimportant_words('FAIL', 13), self.testsRun, None)
325
239
def addSuccess(self, test):
326
"""Tell result that test completed successfully.
328
Called from the TestCase run()
330
self._testConcluded(test)
331
if self._bench_history is not None:
332
benchmark_time = self._extractBenchmarkTime(test)
333
if benchmark_time is not None:
334
self._bench_history.write("%s %s\n" % (
335
self._formatTime(benchmark_time),
337
self.report_success(test)
338
self._cleanupLogFile(test)
339
unittest.TestResult.addSuccess(self, test)
340
test._log_contents = ''
342
def _testConcluded(self, test):
343
"""Common code when a test has finished.
345
Called regardless of whether it succeded, failed, etc.
349
def _addKnownFailure(self, test, err):
350
self.known_failure_count += 1
351
self.report_known_failure(test, err)
353
def addNotSupported(self, test, feature):
354
"""The test will not be run because of a missing feature.
356
# this can be called in two different ways: it may be that the
357
# test started running, and then raised (through addError)
358
# UnavailableFeature. Alternatively this method can be called
359
# while probing for features before running the tests; in that
360
# case we will see startTest and stopTest, but the test will never
362
self.unsupported.setdefault(str(feature), 0)
363
self.unsupported[str(feature)] += 1
364
self.report_unsupported(test, feature)
366
def addSkip(self, test, reason):
367
"""A test has not run for 'reason'."""
369
self.report_skip(test, reason)
371
def _addNotApplicable(self, test, skip_excinfo):
372
if isinstance(skip_excinfo[1], TestNotApplicable):
373
self.not_applicable_count += 1
374
self.report_not_applicable(test, skip_excinfo)
377
except KeyboardInterrupt:
380
self.addError(test, test.exc_info())
382
# seems best to treat this as success from point-of-view of unittest
383
# -- it actually does nothing so it barely matters :)
384
unittest.TestResult.addSuccess(self, test)
385
test._log_contents = ''
240
self.extractBenchmarkTime(test)
242
self.stream.writeln(' OK %s' % self._testTimeString())
243
for bench_called, stats in getattr(test, '_benchcalls', []):
244
self.stream.writeln('LSProf output for %s(%s, %s)' % bench_called)
245
stats.pprint(file=self.stream)
246
elif self.dots and self.pb is None:
247
self.stream.write('~')
249
self.pb.update(self._ellipsise_unimportant_words('OK', 13), self.testsRun, None)
251
unittest.TestResult.addSuccess(self, test)
253
def addSkipped(self, test, skip_excinfo):
254
self.extractBenchmarkTime(test)
256
print >>self.stream, ' SKIP %s' % self._testTimeString()
257
print >>self.stream, ' %s' % skip_excinfo[1]
258
elif self.dots and self.pb is None:
259
self.stream.write('S')
261
self.pb.update(self._ellipsise_unimportant_words('SKIP', 13), self.testsRun, None)
263
# seems best to treat this as success from point-of-view of unittest
264
# -- it actually does nothing so it barely matters :)
265
unittest.TestResult.addSuccess(self, test)
387
267
def printErrorList(self, flavour, errors):
388
268
for test, err in errors:
389
269
self.stream.writeln(self.separator1)
390
self.stream.write("%s: " % flavour)
391
self.stream.writeln(self.getDescription(test))
270
self.stream.writeln("%s: %s" % (flavour, self.getDescription(test)))
392
271
if getattr(test, '_get_log', None) is not None:
393
log_contents = test._get_log()
395
self.stream.write('\n')
397
('vvvv[log from %s]' % test.id()).ljust(78,'-'))
398
self.stream.write('\n')
399
self.stream.write(log_contents)
400
self.stream.write('\n')
402
('^^^^[log from %s]' % test.id()).ljust(78,'-'))
403
self.stream.write('\n')
273
print >>self.stream, \
274
('vvvv[log from %s]' % test.id()).ljust(78,'-')
275
print >>self.stream, test._get_log()
276
print >>self.stream, \
277
('^^^^[log from %s]' % test.id()).ljust(78,'-')
404
278
self.stream.writeln(self.separator2)
405
279
self.stream.writeln("%s" % err)
407
def progress(self, offset, whence):
408
"""The test is adjusting the count of tests to run."""
409
if whence == SUBUNIT_SEEK_SET:
410
self.num_tests = offset
411
elif whence == SUBUNIT_SEEK_CUR:
412
self.num_tests += offset
414
raise errors.BzrError("Unknown whence %r" % whence)
416
def report_cleaning_up(self):
419
def startTestRun(self):
420
self.startTime = time.time()
422
def report_success(self, test):
425
def wasStrictlySuccessful(self):
426
if self.unsupported or self.known_failure_count:
428
return self.wasSuccessful()
431
class TextTestResult(ExtendedTestResult):
432
"""Displays progress and results of tests in text form"""
434
def __init__(self, stream, descriptions, verbosity,
439
ExtendedTestResult.__init__(self, stream, descriptions, verbosity,
440
bench_history, strict)
441
# We no longer pass them around, but just rely on the UIFactory stack
444
warnings.warn("Passing pb to TextTestResult is deprecated")
445
self.pb = self.ui.nested_progress_bar()
446
self.pb.show_pct = False
447
self.pb.show_spinner = False
448
self.pb.show_eta = False,
449
self.pb.show_count = False
450
self.pb.show_bar = False
451
self.pb.update_latency = 0
452
self.pb.show_transport_activity = False
454
def stopTestRun(self):
455
# called when the tests that are going to run have run
458
super(TextTestResult, self).stopTestRun()
460
def startTestRun(self):
461
super(TextTestResult, self).startTestRun()
462
self.pb.update('[test 0/%d] Starting' % (self.num_tests))
464
def printErrors(self):
465
# clear the pb to make room for the error listing
467
super(TextTestResult, self).printErrors()
469
def _progress_prefix_text(self):
470
# the longer this text, the less space we have to show the test
472
a = '[%d' % self.count # total that have been run
473
# tests skipped as known not to be relevant are not important enough
475
## if self.skip_count:
476
## a += ', %d skip' % self.skip_count
477
## if self.known_failure_count:
478
## a += '+%dX' % self.known_failure_count
480
a +='/%d' % self.num_tests
482
runtime = time.time() - self._overall_start_time
484
a += '%dm%ds' % (runtime / 60, runtime % 60)
488
a += ', %d err' % self.error_count
489
if self.failure_count:
490
a += ', %d fail' % self.failure_count
492
a += ', %d missing' % len(self.unsupported)
496
def report_test_start(self, test):
499
self._progress_prefix_text()
501
+ self._shortened_test_description(test))
503
def _test_description(self, test):
504
return self._shortened_test_description(test)
506
def report_error(self, test, err):
507
self.pb.note('ERROR: %s\n %s\n',
508
self._test_description(test),
512
def report_failure(self, test, err):
513
self.pb.note('FAIL: %s\n %s\n',
514
self._test_description(test),
518
def report_known_failure(self, test, err):
519
self.pb.note('XFAIL: %s\n%s\n',
520
self._test_description(test), err[1])
522
def report_skip(self, test, reason):
525
def report_not_applicable(self, test, skip_excinfo):
528
def report_unsupported(self, test, feature):
529
"""test cannot be run because feature is missing."""
531
def report_cleaning_up(self):
532
self.pb.update('Cleaning up')
535
class VerboseTestResult(ExtendedTestResult):
536
"""Produce long output, with one line per test run plus times"""
538
def _ellipsize_to_right(self, a_string, final_width):
539
"""Truncate and pad a string, keeping the right hand side"""
540
if len(a_string) > final_width:
541
result = '...' + a_string[3-final_width:]
544
return result.ljust(final_width)
546
def startTestRun(self):
547
super(VerboseTestResult, self).startTestRun()
548
self.stream.write('running %d tests...\n' % self.num_tests)
550
def report_test_start(self, test):
552
name = self._shortened_test_description(test)
553
# width needs space for 6 char status, plus 1 for slash, plus an
554
# 11-char time string, plus a trailing blank
555
# when NUMBERED_DIRS: plus 5 chars on test number, plus 1 char on space
556
self.stream.write(self._ellipsize_to_right(name,
557
osutils.terminal_width()-18))
560
def _error_summary(self, err):
562
return '%s%s' % (indent, err[1])
564
def report_error(self, test, err):
565
self.stream.writeln('ERROR %s\n%s'
566
% (self._testTimeString(test),
567
self._error_summary(err)))
569
def report_failure(self, test, err):
570
self.stream.writeln(' FAIL %s\n%s'
571
% (self._testTimeString(test),
572
self._error_summary(err)))
574
def report_known_failure(self, test, err):
575
self.stream.writeln('XFAIL %s\n%s'
576
% (self._testTimeString(test),
577
self._error_summary(err)))
579
def report_success(self, test):
580
self.stream.writeln(' OK %s' % self._testTimeString(test))
581
for bench_called, stats in getattr(test, '_benchcalls', []):
582
self.stream.writeln('LSProf output for %s(%s, %s)' % bench_called)
583
stats.pprint(file=self.stream)
584
# flush the stream so that we get smooth output. This verbose mode is
585
# used to show the output in PQM.
588
def report_skip(self, test, reason):
589
self.stream.writeln(' SKIP %s\n%s'
590
% (self._testTimeString(test), reason))
592
def report_not_applicable(self, test, skip_excinfo):
593
self.stream.writeln(' N/A %s\n%s'
594
% (self._testTimeString(test),
595
self._error_summary(skip_excinfo)))
597
def report_unsupported(self, test, feature):
598
"""test cannot be run because feature is missing."""
599
self.stream.writeln("NODEP %s\n The feature '%s' is not available."
600
%(self._testTimeString(test), feature))
603
282
class TextTestRunner(object):
604
283
stop_on_failure = False
1950
815
sys.stderr = real_stderr
1951
816
sys.stdin = real_stdin
1953
def reduceLockdirTimeout(self):
1954
"""Reduce the default lock timeout for the duration of the test, so that
1955
if LockContention occurs during a test, it does so quickly.
1957
Tests that expect to provoke LockContention errors should call this.
1959
orig_timeout = bzrlib.lockdir._DEFAULT_TIMEOUT_SECONDS
1961
bzrlib.lockdir._DEFAULT_TIMEOUT_SECONDS = orig_timeout
1962
self.addCleanup(resetTimeout)
1963
bzrlib.lockdir._DEFAULT_TIMEOUT_SECONDS = 0
1965
def make_utf8_encoded_stringio(self, encoding_type=None):
1966
"""Return a StringIOWrapper instance, that will encode Unicode
1969
if encoding_type is None:
1970
encoding_type = 'strict'
1972
output_encoding = 'utf-8'
1973
sio = codecs.getwriter(output_encoding)(sio, errors=encoding_type)
1974
sio.encoding = output_encoding
1977
def disable_verb(self, verb):
1978
"""Disable a smart server verb for one test."""
1979
from bzrlib.smart import request
1980
request_handlers = request.request_handlers
1981
orig_method = request_handlers.get(verb)
1982
request_handlers.remove(verb)
1984
request_handlers.register(verb, orig_method)
1985
self.addCleanup(restoreVerb)
1988
class CapturedCall(object):
1989
"""A helper for capturing smart server calls for easy debug analysis."""
1991
def __init__(self, params, prefix_length):
1992
"""Capture the call with params and skip prefix_length stack frames."""
1995
# The last 5 frames are the __init__, the hook frame, and 3 smart
1996
# client frames. Beyond this we could get more clever, but this is good
1998
stack = traceback.extract_stack()[prefix_length:-5]
1999
self.stack = ''.join(traceback.format_list(stack))
2002
return self.call.method
2005
return self.call.method
2011
class TestCaseWithMemoryTransport(TestCase):
2012
"""Common test class for tests that do not need disk resources.
2014
Tests that need disk resources should derive from TestCaseWithTransport.
2016
TestCaseWithMemoryTransport sets the TEST_ROOT variable for all bzr tests.
2018
For TestCaseWithMemoryTransport the test_home_dir is set to the name of
2019
a directory which does not exist. This serves to help ensure test isolation
2020
is preserved. test_dir is set to the TEST_ROOT, as is cwd, because they
2021
must exist. However, TestCaseWithMemoryTransport does not offer local
2022
file defaults for the transport in tests, nor does it obey the command line
2023
override, so tests that accidentally write to the common directory should
2026
:cvar TEST_ROOT: Directory containing all temporary directories, plus
2027
a .bzr directory that stops us ascending higher into the filesystem.
2033
def __init__(self, methodName='runTest'):
2034
# allow test parameterization after test construction and before test
2035
# execution. Variables that the parameterizer sets need to be
2036
# ones that are not set by setUp, or setUp will trash them.
2037
super(TestCaseWithMemoryTransport, self).__init__(methodName)
2038
self.vfs_transport_factory = default_transport
2039
self.transport_server = None
2040
self.transport_readonly_server = None
2041
self.__vfs_server = None
2043
def get_transport(self, relpath=None):
2044
"""Return a writeable transport.
2046
This transport is for the test scratch space relative to
2049
:param relpath: a path relative to the base url.
2051
t = get_transport(self.get_url(relpath))
2052
self.assertFalse(t.is_readonly())
2055
def get_readonly_transport(self, relpath=None):
2056
"""Return a readonly transport for the test scratch space
2058
This can be used to test that operations which should only need
2059
readonly access in fact do not try to write.
2061
:param relpath: a path relative to the base url.
2063
t = get_transport(self.get_readonly_url(relpath))
2064
self.assertTrue(t.is_readonly())
2067
def create_transport_readonly_server(self):
2068
"""Create a transport server from class defined at init.
2070
This is mostly a hook for daughter classes.
2072
return self.transport_readonly_server()
2074
def get_readonly_server(self):
2075
"""Get the server instance for the readonly transport
2077
This is useful for some tests with specific servers to do diagnostics.
2079
if self.__readonly_server is None:
2080
if self.transport_readonly_server is None:
2081
# readonly decorator requested
2082
self.__readonly_server = ReadonlyServer()
2084
# explicit readonly transport.
2085
self.__readonly_server = self.create_transport_readonly_server()
2086
self.start_server(self.__readonly_server,
2087
self.get_vfs_only_server())
2088
return self.__readonly_server
2090
def get_readonly_url(self, relpath=None):
2091
"""Get a URL for the readonly transport.
2093
This will either be backed by '.' or a decorator to the transport
2094
used by self.get_url()
2095
relpath provides for clients to get a path relative to the base url.
2096
These should only be downwards relative, not upwards.
2098
base = self.get_readonly_server().get_url()
2099
return self._adjust_url(base, relpath)
2101
def get_vfs_only_server(self):
2102
"""Get the vfs only read/write server instance.
2104
This is useful for some tests with specific servers that need
2107
For TestCaseWithMemoryTransport this is always a MemoryServer, and there
2108
is no means to override it.
2110
if self.__vfs_server is None:
2111
self.__vfs_server = MemoryServer()
2112
self.start_server(self.__vfs_server)
2113
return self.__vfs_server
2115
def get_server(self):
2116
"""Get the read/write server instance.
2118
This is useful for some tests with specific servers that need
2121
This is built from the self.transport_server factory. If that is None,
2122
then the self.get_vfs_server is returned.
2124
if self.__server is None:
2125
if (self.transport_server is None or self.transport_server is
2126
self.vfs_transport_factory):
2127
self.__server = self.get_vfs_only_server()
2129
# bring up a decorated means of access to the vfs only server.
2130
self.__server = self.transport_server()
2131
self.start_server(self.__server, self.get_vfs_only_server())
2132
return self.__server
2134
def _adjust_url(self, base, relpath):
2135
"""Get a URL (or maybe a path) for the readwrite transport.
2137
This will either be backed by '.' or to an equivalent non-file based
2139
relpath provides for clients to get a path relative to the base url.
2140
These should only be downwards relative, not upwards.
2142
if relpath is not None and relpath != '.':
2143
if not base.endswith('/'):
2145
# XXX: Really base should be a url; we did after all call
2146
# get_url()! But sometimes it's just a path (from
2147
# LocalAbspathServer), and it'd be wrong to append urlescaped data
2148
# to a non-escaped local path.
2149
if base.startswith('./') or base.startswith('/'):
2152
base += urlutils.escape(relpath)
2155
def get_url(self, relpath=None):
2156
"""Get a URL (or maybe a path) for the readwrite transport.
2158
This will either be backed by '.' or to an equivalent non-file based
2160
relpath provides for clients to get a path relative to the base url.
2161
These should only be downwards relative, not upwards.
2163
base = self.get_server().get_url()
2164
return self._adjust_url(base, relpath)
2166
def get_vfs_only_url(self, relpath=None):
2167
"""Get a URL (or maybe a path for the plain old vfs transport.
2169
This will never be a smart protocol. It always has all the
2170
capabilities of the local filesystem, but it might actually be a
2171
MemoryTransport or some other similar virtual filesystem.
2173
This is the backing transport (if any) of the server returned by
2174
get_url and get_readonly_url.
2176
:param relpath: provides for clients to get a path relative to the base
2177
url. These should only be downwards relative, not upwards.
2180
base = self.get_vfs_only_server().get_url()
2181
return self._adjust_url(base, relpath)
2183
def _create_safety_net(self):
2184
"""Make a fake bzr directory.
2186
This prevents any tests propagating up onto the TEST_ROOT directory's
2189
root = TestCaseWithMemoryTransport.TEST_ROOT
2190
bzrdir.BzrDir.create_standalone_workingtree(root)
2192
def _check_safety_net(self):
2193
"""Check that the safety .bzr directory have not been touched.
2195
_make_test_root have created a .bzr directory to prevent tests from
2196
propagating. This method ensures than a test did not leaked.
2198
root = TestCaseWithMemoryTransport.TEST_ROOT
2199
wt = workingtree.WorkingTree.open(root)
2200
last_rev = wt.last_revision()
2201
if last_rev != 'null:':
2202
# The current test have modified the /bzr directory, we need to
2203
# recreate a new one or all the followng tests will fail.
2204
# If you need to inspect its content uncomment the following line
2205
# import pdb; pdb.set_trace()
2206
_rmtree_temp_dir(root + '/.bzr')
2207
self._create_safety_net()
2208
raise AssertionError('%s/.bzr should not be modified' % root)
2210
def _make_test_root(self):
2211
if TestCaseWithMemoryTransport.TEST_ROOT is None:
2212
root = osutils.mkdtemp(prefix='testbzr-', suffix='.tmp')
2213
TestCaseWithMemoryTransport.TEST_ROOT = root
2215
self._create_safety_net()
2217
# The same directory is used by all tests, and we're not
2218
# specifically told when all tests are finished. This will do.
2219
atexit.register(_rmtree_temp_dir, root)
2221
self.addCleanup(self._check_safety_net)
2223
def makeAndChdirToTestDir(self):
2224
"""Create a temporary directories for this one test.
2226
This must set self.test_home_dir and self.test_dir and chdir to
2229
For TestCaseWithMemoryTransport we chdir to the TEST_ROOT for this test.
2231
os.chdir(TestCaseWithMemoryTransport.TEST_ROOT)
2232
self.test_dir = TestCaseWithMemoryTransport.TEST_ROOT
2233
self.test_home_dir = self.test_dir + "/MemoryTransportMissingHomeDir"
2235
def make_branch(self, relpath, format=None):
2236
"""Create a branch on the transport at relpath."""
2237
repo = self.make_repository(relpath, format=format)
2238
return repo.bzrdir.create_branch()
2240
def make_bzrdir(self, relpath, format=None):
2242
# might be a relative or absolute path
2243
maybe_a_url = self.get_url(relpath)
2244
segments = maybe_a_url.rsplit('/', 1)
2245
t = get_transport(maybe_a_url)
2246
if len(segments) > 1 and segments[-1] not in ('', '.'):
2250
if isinstance(format, basestring):
2251
format = bzrdir.format_registry.make_bzrdir(format)
2252
return format.initialize_on_transport(t)
2253
except errors.UninitializableFormat:
2254
raise TestSkipped("Format %s is not initializable." % format)
2256
def make_repository(self, relpath, shared=False, format=None):
2257
"""Create a repository on our default transport at relpath.
2259
Note that relpath must be a relative path, not a full url.
2261
# FIXME: If you create a remoterepository this returns the underlying
2262
# real format, which is incorrect. Actually we should make sure that
2263
# RemoteBzrDir returns a RemoteRepository.
2264
# maybe mbp 20070410
2265
made_control = self.make_bzrdir(relpath, format=format)
2266
return made_control.create_repository(shared=shared)
2268
def make_smart_server(self, path):
2269
smart_server = server.SmartTCPServer_for_testing()
2270
self.start_server(smart_server, self.get_server())
2271
remote_transport = get_transport(smart_server.get_url()).clone(path)
2272
return remote_transport
2274
def make_branch_and_memory_tree(self, relpath, format=None):
2275
"""Create a branch on the default transport and a MemoryTree for it."""
2276
b = self.make_branch(relpath, format=format)
2277
return memorytree.MemoryTree.create_on_branch(b)
2279
def make_branch_builder(self, relpath, format=None):
2280
branch = self.make_branch(relpath, format=format)
2281
return branchbuilder.BranchBuilder(branch=branch)
2283
def overrideEnvironmentForTesting(self):
2284
os.environ['HOME'] = self.test_home_dir
2285
os.environ['BZR_HOME'] = self.test_home_dir
2288
super(TestCaseWithMemoryTransport, self).setUp()
2289
self._make_test_root()
2290
_currentdir = os.getcwdu()
2291
def _leaveDirectory():
2292
os.chdir(_currentdir)
2293
self.addCleanup(_leaveDirectory)
2294
self.makeAndChdirToTestDir()
2295
self.overrideEnvironmentForTesting()
2296
self.__readonly_server = None
2297
self.__server = None
2298
self.reduceLockdirTimeout()
2300
def setup_smart_server_with_call_log(self):
2301
"""Sets up a smart server as the transport server with a call log."""
2302
self.transport_server = server.SmartTCPServer_for_testing
2303
self.hpss_calls = []
2305
# Skip the current stack down to the caller of
2306
# setup_smart_server_with_call_log
2307
prefix_length = len(traceback.extract_stack()) - 2
2308
def capture_hpss_call(params):
2309
self.hpss_calls.append(
2310
CapturedCall(params, prefix_length))
2311
client._SmartClient.hooks.install_named_hook(
2312
'call', capture_hpss_call, None)
2314
def reset_smart_call_log(self):
2315
self.hpss_calls = []
2318
class TestCaseInTempDir(TestCaseWithMemoryTransport):
818
def merge(self, branch_from, wt_to):
819
"""A helper for tests to do a ui-less merge.
821
This should move to the main library when someone has time to integrate
824
# minimal ui-less merge.
825
wt_to.branch.fetch(branch_from)
826
base_rev = common_ancestor(branch_from.last_revision(),
827
wt_to.branch.last_revision(),
828
wt_to.branch.repository)
829
merge_inner(wt_to.branch, branch_from.basis_tree(),
830
wt_to.branch.repository.revision_tree(base_rev),
832
wt_to.add_pending_merge(branch_from.last_revision())
835
BzrTestBase = TestCase
838
class TestCaseInTempDir(TestCase):
2319
839
"""Derived class that runs a test within a temporary directory.
2321
841
This is useful for tests that need to create a branch, etc.
2563
1130
for readonly urls.
2565
1132
TODO RBC 20060127: make this an option to TestCaseWithTransport so it can
2566
be used without needed to redo it when a different
1133
be used without needed to redo it when a different
2567
1134
subclass is in use ?
2570
1137
def setUp(self):
2571
1138
super(ChrootedTestCase, self).setUp()
2572
if not self.vfs_transport_factory == MemoryServer:
2573
self.transport_readonly_server = HttpServer
2576
def condition_id_re(pattern):
2577
"""Create a condition filter which performs a re check on a test's id.
2579
:param pattern: A regular expression string.
2580
:return: A callable that returns True if the re matches.
2582
filter_re = osutils.re_compile_checked(pattern, 0,
2584
def condition(test):
2586
return filter_re.search(test_id)
2590
def condition_isinstance(klass_or_klass_list):
2591
"""Create a condition filter which returns isinstance(param, klass).
2593
:return: A callable which when called with one parameter obj return the
2594
result of isinstance(obj, klass_or_klass_list).
2597
return isinstance(obj, klass_or_klass_list)
2601
def condition_id_in_list(id_list):
2602
"""Create a condition filter which verify that test's id in a list.
2604
:param id_list: A TestIdList object.
2605
:return: A callable that returns True if the test's id appears in the list.
2607
def condition(test):
2608
return id_list.includes(test.id())
2612
def condition_id_startswith(starts):
2613
"""Create a condition filter verifying that test's id starts with a string.
2615
:param starts: A list of string.
2616
:return: A callable that returns True if the test's id starts with one of
2619
def condition(test):
2620
for start in starts:
2621
if test.id().startswith(start):
2627
def exclude_tests_by_condition(suite, condition):
2628
"""Create a test suite which excludes some tests from suite.
2630
:param suite: The suite to get tests from.
2631
:param condition: A callable whose result evaluates True when called with a
2632
test case which should be excluded from the result.
2633
:return: A suite which contains the tests found in suite that fail
2637
for test in iter_suite_tests(suite):
2638
if not condition(test):
2640
return TestUtil.TestSuite(result)
2643
def filter_suite_by_condition(suite, condition):
2644
"""Create a test suite by filtering another one.
2646
:param suite: The source suite.
2647
:param condition: A callable whose result evaluates True when called with a
2648
test case which should be included in the result.
2649
:return: A suite which contains the tests found in suite that pass
2653
for test in iter_suite_tests(suite):
2656
return TestUtil.TestSuite(result)
1139
if not self.transport_server == bzrlib.transport.memory.MemoryServer:
1140
self.transport_readonly_server = bzrlib.transport.http.HttpServer
2659
1143
def filter_suite_by_re(suite, pattern):
2660
"""Create a test suite by filtering another one.
2662
:param suite: the source suite
2663
:param pattern: pattern that names must match
2664
:returns: the newly created suite
2666
condition = condition_id_re(pattern)
2667
result_suite = filter_suite_by_condition(suite, condition)
2671
def filter_suite_by_id_list(suite, test_id_list):
2672
"""Create a test suite by filtering another one.
2674
:param suite: The source suite.
2675
:param test_id_list: A list of the test ids to keep as strings.
2676
:returns: the newly created suite
2678
condition = condition_id_in_list(test_id_list)
2679
result_suite = filter_suite_by_condition(suite, condition)
2683
def filter_suite_by_id_startswith(suite, start):
2684
"""Create a test suite by filtering another one.
2686
:param suite: The source suite.
2687
:param start: A list of string the test id must start with one of.
2688
:returns: the newly created suite
2690
condition = condition_id_startswith(start)
2691
result_suite = filter_suite_by_condition(suite, condition)
2695
def exclude_tests_by_re(suite, pattern):
2696
"""Create a test suite which excludes some tests from suite.
2698
:param suite: The suite to get tests from.
2699
:param pattern: A regular expression string. Test ids that match this
2700
pattern will be excluded from the result.
2701
:return: A TestSuite that contains all the tests from suite without the
2702
tests that matched pattern. The order of tests is the same as it was in
2705
return exclude_tests_by_condition(suite, condition_id_re(pattern))
2708
def preserve_input(something):
2709
"""A helper for performing test suite transformation chains.
2711
:param something: Anything you want to preserve.
2717
def randomize_suite(suite):
2718
"""Return a new TestSuite with suite's tests in random order.
2720
The tests in the input suite are flattened into a single suite in order to
2721
accomplish this. Any nested TestSuites are removed to provide global
2724
tests = list(iter_suite_tests(suite))
2725
random.shuffle(tests)
2726
return TestUtil.TestSuite(tests)
2729
def split_suite_by_condition(suite, condition):
2730
"""Split a test suite into two by a condition.
2732
:param suite: The suite to split.
2733
:param condition: The condition to match on. Tests that match this
2734
condition are returned in the first test suite, ones that do not match
2735
are in the second suite.
2736
:return: A tuple of two test suites, where the first contains tests from
2737
suite matching the condition, and the second contains the remainder
2738
from suite. The order within each output suite is the same as it was in
1144
result = TestUtil.TestSuite()
1145
filter_re = re.compile(pattern)
2743
1146
for test in iter_suite_tests(suite):
2745
matched.append(test)
2747
did_not_match.append(test)
2748
return TestUtil.TestSuite(matched), TestUtil.TestSuite(did_not_match)
2751
def split_suite_by_re(suite, pattern):
2752
"""Split a test suite into two by a regular expression.
2754
:param suite: The suite to split.
2755
:param pattern: A regular expression string. Test ids that match this
2756
pattern will be in the first test suite returned, and the others in the
2757
second test suite returned.
2758
:return: A tuple of two test suites, where the first contains tests from
2759
suite matching pattern, and the second contains the remainder from
2760
suite. The order within each output suite is the same as it was in
2763
return split_suite_by_condition(suite, condition_id_re(pattern))
1147
if filter_re.search(test.id()):
1148
result.addTest(test)
2766
1152
def run_suite(suite, name='test', verbose=False, pattern=".*",
2767
stop_on_failure=False,
2768
transport=None, lsprof_timed=None, bench_history=None,
2769
matching_tests_first=None,
2772
exclude_pattern=None,
2775
suite_decorators=None,
2777
result_decorators=None,
2779
"""Run a test suite for bzr selftest.
2781
:param runner_class: The class of runner to use. Must support the
2782
constructor arguments passed by run_suite which are more than standard
2784
:return: A boolean indicating success.
1153
stop_on_failure=False, keep_output=False,
1154
transport=None, lsprof_timed=None):
1155
TestCaseInTempDir._TEST_NAME = name
2786
1156
TestCase._gather_lsprof_in_benchmarks = lsprof_timed
2791
if runner_class is None:
2792
runner_class = TextTestRunner
2795
runner = runner_class(stream=stream,
1162
pb = progress.ProgressBar()
1163
runner = TextTestRunner(stream=sys.stdout,
2796
1164
descriptions=0,
2797
1165
verbosity=verbosity,
2798
bench_history=bench_history,
2800
result_decorators=result_decorators,
1166
keep_output=keep_output,
2802
1168
runner.stop_on_failure=stop_on_failure
2803
# built in decorator factories:
2805
random_order(random_seed, runner),
2806
exclude_tests(exclude_pattern),
2808
if matching_tests_first:
2809
decorators.append(tests_first(pattern))
2811
decorators.append(filter_tests(pattern))
2812
if suite_decorators:
2813
decorators.extend(suite_decorators)
2814
# tell the result object how many tests will be running: (except if
2815
# --parallel=fork is being used. Robert said he will provide a better
2816
# progress design later -- vila 20090817)
2817
if fork_decorator not in decorators:
2818
decorators.append(CountingDecorator)
2819
for decorator in decorators:
2820
suite = decorator(suite)
2822
# Done after test suite decoration to allow randomisation etc
2823
# to take effect, though that is of marginal benefit.
2825
stream.write("Listing tests only ...\n")
2826
for t in iter_suite_tests(suite):
2827
stream.write("%s\n" % (t.id()))
1170
suite = filter_suite_by_re(suite, pattern)
2829
1171
result = runner.run(suite)
2831
return result.wasStrictlySuccessful()
2833
return result.wasSuccessful()
2836
# A registry where get() returns a suite decorator.
2837
parallel_registry = registry.Registry()
2840
def fork_decorator(suite):
2841
concurrency = osutils.local_concurrency()
2842
if concurrency == 1:
2844
from testtools import ConcurrentTestSuite
2845
return ConcurrentTestSuite(suite, fork_for_tests)
2846
parallel_registry.register('fork', fork_decorator)
2849
def subprocess_decorator(suite):
2850
concurrency = osutils.local_concurrency()
2851
if concurrency == 1:
2853
from testtools import ConcurrentTestSuite
2854
return ConcurrentTestSuite(suite, reinvoke_for_tests)
2855
parallel_registry.register('subprocess', subprocess_decorator)
2858
def exclude_tests(exclude_pattern):
2859
"""Return a test suite decorator that excludes tests."""
2860
if exclude_pattern is None:
2861
return identity_decorator
2862
def decorator(suite):
2863
return ExcludeDecorator(suite, exclude_pattern)
2867
def filter_tests(pattern):
2869
return identity_decorator
2870
def decorator(suite):
2871
return FilterTestsDecorator(suite, pattern)
2875
def random_order(random_seed, runner):
2876
"""Return a test suite decorator factory for randomising tests order.
2878
:param random_seed: now, a string which casts to a long, or a long.
2879
:param runner: A test runner with a stream attribute to report on.
2881
if random_seed is None:
2882
return identity_decorator
2883
def decorator(suite):
2884
return RandomDecorator(suite, random_seed, runner.stream)
2888
def tests_first(pattern):
2890
return identity_decorator
2891
def decorator(suite):
2892
return TestFirstDecorator(suite, pattern)
2896
def identity_decorator(suite):
2901
class TestDecorator(TestSuite):
2902
"""A decorator for TestCase/TestSuite objects.
2904
Usually, subclasses should override __iter__(used when flattening test
2905
suites), which we do to filter, reorder, parallelise and so on, run() and
2909
def __init__(self, suite):
2910
TestSuite.__init__(self)
2913
def countTestCases(self):
2916
cases += test.countTestCases()
2923
def run(self, result):
2924
# Use iteration on self, not self._tests, to allow subclasses to hook
2927
if result.shouldStop:
2933
class CountingDecorator(TestDecorator):
2934
"""A decorator which calls result.progress(self.countTestCases)."""
2936
def run(self, result):
2937
progress_method = getattr(result, 'progress', None)
2938
if callable(progress_method):
2939
progress_method(self.countTestCases(), SUBUNIT_SEEK_SET)
2940
return super(CountingDecorator, self).run(result)
2943
class ExcludeDecorator(TestDecorator):
2944
"""A decorator which excludes test matching an exclude pattern."""
2946
def __init__(self, suite, exclude_pattern):
2947
TestDecorator.__init__(self, suite)
2948
self.exclude_pattern = exclude_pattern
2949
self.excluded = False
2953
return iter(self._tests)
2954
self.excluded = True
2955
suite = exclude_tests_by_re(self, self.exclude_pattern)
2957
self.addTests(suite)
2958
return iter(self._tests)
2961
class FilterTestsDecorator(TestDecorator):
2962
"""A decorator which filters tests to those matching a pattern."""
2964
def __init__(self, suite, pattern):
2965
TestDecorator.__init__(self, suite)
2966
self.pattern = pattern
2967
self.filtered = False
2971
return iter(self._tests)
2972
self.filtered = True
2973
suite = filter_suite_by_re(self, self.pattern)
2975
self.addTests(suite)
2976
return iter(self._tests)
2979
class RandomDecorator(TestDecorator):
2980
"""A decorator which randomises the order of its tests."""
2982
def __init__(self, suite, random_seed, stream):
2983
TestDecorator.__init__(self, suite)
2984
self.random_seed = random_seed
2985
self.randomised = False
2986
self.stream = stream
2990
return iter(self._tests)
2991
self.randomised = True
2992
self.stream.writeln("Randomizing test order using seed %s\n" %
2993
(self.actual_seed()))
2994
# Initialise the random number generator.
2995
random.seed(self.actual_seed())
2996
suite = randomize_suite(self)
2998
self.addTests(suite)
2999
return iter(self._tests)
3001
def actual_seed(self):
3002
if self.random_seed == "now":
3003
# We convert the seed to a long to make it reuseable across
3004
# invocations (because the user can reenter it).
3005
self.random_seed = long(time.time())
3007
# Convert the seed to a long if we can
3009
self.random_seed = long(self.random_seed)
3012
return self.random_seed
3015
class TestFirstDecorator(TestDecorator):
3016
"""A decorator which moves named tests to the front."""
3018
def __init__(self, suite, pattern):
3019
TestDecorator.__init__(self, suite)
3020
self.pattern = pattern
3021
self.filtered = False
3025
return iter(self._tests)
3026
self.filtered = True
3027
suites = split_suite_by_re(self, self.pattern)
3029
self.addTests(suites)
3030
return iter(self._tests)
3033
def partition_tests(suite, count):
3034
"""Partition suite into count lists of tests."""
3036
tests = list(iter_suite_tests(suite))
3037
tests_per_process = int(math.ceil(float(len(tests)) / count))
3038
for block in range(count):
3039
low_test = block * tests_per_process
3040
high_test = low_test + tests_per_process
3041
process_tests = tests[low_test:high_test]
3042
result.append(process_tests)
3046
def fork_for_tests(suite):
3047
"""Take suite and start up one runner per CPU by forking()
3049
:return: An iterable of TestCase-like objects which can each have
3050
run(result) called on them to feed tests to result.
3052
concurrency = osutils.local_concurrency()
3054
from subunit import TestProtocolClient, ProtocolTestCase
3056
from subunit.test_results import AutoTimingTestResultDecorator
3058
AutoTimingTestResultDecorator = lambda x:x
3059
class TestInOtherProcess(ProtocolTestCase):
3060
# Should be in subunit, I think. RBC.
3061
def __init__(self, stream, pid):
3062
ProtocolTestCase.__init__(self, stream)
3065
def run(self, result):
3067
ProtocolTestCase.run(self, result)
3069
os.waitpid(self.pid, os.WNOHANG)
3071
test_blocks = partition_tests(suite, concurrency)
3072
for process_tests in test_blocks:
3073
process_suite = TestSuite()
3074
process_suite.addTests(process_tests)
3075
c2pread, c2pwrite = os.pipe()
3080
# Leave stderr and stdout open so we can see test noise
3081
# Close stdin so that the child goes away if it decides to
3082
# read from stdin (otherwise its a roulette to see what
3083
# child actually gets keystrokes for pdb etc).
3086
stream = os.fdopen(c2pwrite, 'wb', 1)
3087
subunit_result = AutoTimingTestResultDecorator(
3088
TestProtocolClient(stream))
3089
process_suite.run(subunit_result)
3094
stream = os.fdopen(c2pread, 'rb', 1)
3095
test = TestInOtherProcess(stream, pid)
3100
def reinvoke_for_tests(suite):
3101
"""Take suite and start up one runner per CPU using subprocess().
3103
:return: An iterable of TestCase-like objects which can each have
3104
run(result) called on them to feed tests to result.
3106
concurrency = osutils.local_concurrency()
3108
from subunit import ProtocolTestCase
3109
class TestInSubprocess(ProtocolTestCase):
3110
def __init__(self, process, name):
3111
ProtocolTestCase.__init__(self, process.stdout)
3112
self.process = process
3113
self.process.stdin.close()
3116
def run(self, result):
3118
ProtocolTestCase.run(self, result)
3121
os.unlink(self.name)
3122
# print "pid %d finished" % finished_process
3123
test_blocks = partition_tests(suite, concurrency)
3124
for process_tests in test_blocks:
3125
# ugly; currently reimplement rather than reuses TestCase methods.
3126
bzr_path = os.path.dirname(os.path.dirname(bzrlib.__file__))+'/bzr'
3127
if not os.path.isfile(bzr_path):
3128
# We are probably installed. Assume sys.argv is the right file
3129
bzr_path = sys.argv[0]
3130
fd, test_list_file_name = tempfile.mkstemp()
3131
test_list_file = os.fdopen(fd, 'wb', 1)
3132
for test in process_tests:
3133
test_list_file.write(test.id() + '\n')
3134
test_list_file.close()
3136
argv = [bzr_path, 'selftest', '--load-list', test_list_file_name,
3138
if '--no-plugins' in sys.argv:
3139
argv.append('--no-plugins')
3140
# stderr=STDOUT would be ideal, but until we prevent noise on
3141
# stderr it can interrupt the subunit protocol.
3142
process = Popen(argv, stdin=PIPE, stdout=PIPE, stderr=PIPE,
3144
test = TestInSubprocess(process, test_list_file_name)
3147
os.unlink(test_list_file_name)
3152
class ForwardingResult(unittest.TestResult):
3154
def __init__(self, target):
3155
unittest.TestResult.__init__(self)
3156
self.result = target
3158
def startTest(self, test):
3159
self.result.startTest(test)
3161
def stopTest(self, test):
3162
self.result.stopTest(test)
3164
def startTestRun(self):
3165
self.result.startTestRun()
3167
def stopTestRun(self):
3168
self.result.stopTestRun()
3170
def addSkip(self, test, reason):
3171
self.result.addSkip(test, reason)
3173
def addSuccess(self, test):
3174
self.result.addSuccess(test)
3176
def addError(self, test, err):
3177
self.result.addError(test, err)
3179
def addFailure(self, test, err):
3180
self.result.addFailure(test, err)
3183
class BZRTransformingResult(ForwardingResult):
3185
def addError(self, test, err):
3186
feature = self._error_looks_like('UnavailableFeature: ', err)
3187
if feature is not None:
3188
self.result.addNotSupported(test, feature)
3190
self.result.addError(test, err)
3192
def addFailure(self, test, err):
3193
known = self._error_looks_like('KnownFailure: ', err)
3194
if known is not None:
3195
self.result._addKnownFailure(test, [KnownFailure,
3196
KnownFailure(known), None])
3198
self.result.addFailure(test, err)
3200
def _error_looks_like(self, prefix, err):
3201
"""Deserialize exception and returns the stringify value."""
3205
if isinstance(exc, subunit.RemoteException):
3206
# stringify the exception gives access to the remote traceback
3207
# We search the last line for 'prefix'
3208
lines = str(exc).split('\n')
3209
while lines and not lines[-1]:
3212
if lines[-1].startswith(prefix):
3213
value = lines[-1][len(prefix):]
3217
class ProfileResult(ForwardingResult):
3218
"""Generate profiling data for all activity between start and success.
3220
The profile data is appended to the test's _benchcalls attribute and can
3221
be accessed by the forwarded-to TestResult.
3223
While it might be cleaner do accumulate this in stopTest, addSuccess is
3224
where our existing output support for lsprof is, and this class aims to
3225
fit in with that: while it could be moved it's not necessary to accomplish
3226
test profiling, nor would it be dramatically cleaner.
3229
def startTest(self, test):
3230
self.profiler = bzrlib.lsprof.BzrProfiler()
3231
self.profiler.start()
3232
ForwardingResult.startTest(self, test)
3234
def addSuccess(self, test):
3235
stats = self.profiler.stop()
3237
calls = test._benchcalls
3238
except AttributeError:
3239
test._benchcalls = []
3240
calls = test._benchcalls
3241
calls.append(((test.id(), "", ""), stats))
3242
ForwardingResult.addSuccess(self, test)
3244
def stopTest(self, test):
3245
ForwardingResult.stopTest(self, test)
3246
self.profiler = None
3249
# Controlled by "bzr selftest -E=..." option
3250
# Currently supported:
3251
# -Eallow_debug Will no longer clear debug.debug_flags() so it
3252
# preserves any flags supplied at the command line.
3253
# -Edisable_lock_checks Turns errors in mismatched locks into simple prints
3254
# rather than failing tests. And no longer raise
3255
# LockContention when fctnl locks are not being used
3256
# with proper exclusion rules.
3257
selftest_debug_flags = set()
1172
return result.wasSuccessful()
3260
1175
def selftest(verbose=False, pattern=".*", stop_on_failure=True,
3261
1177
transport=None,
3262
1178
test_suite_factory=None,
3265
matching_tests_first=None,
3268
exclude_pattern=None,
3274
suite_decorators=None,
3278
1180
"""Run the whole test suite under the enhanced runner"""
3279
# XXX: Very ugly way to do this...
3280
# Disable warning about old formats because we don't want it to disturb
3281
# any blackbox tests.
3282
from bzrlib import repository
3283
repository._deprecation_warning_done = True
3285
1181
global default_transport
3286
1182
if transport is None:
3287
1183
transport = default_transport
3288
1184
old_transport = default_transport
3289
1185
default_transport = transport
3290
global selftest_debug_flags
3291
old_debug_flags = selftest_debug_flags
3292
if debug_flags is not None:
3293
selftest_debug_flags = set(debug_flags)
3295
if load_list is None:
3298
keep_only = load_test_id_list(load_list)
3300
starting_with = [test_prefix_alias_registry.resolve_alias(start)
3301
for start in starting_with]
3302
1187
if test_suite_factory is None:
3303
# Reduce loading time by loading modules based on the starting_with
3305
suite = test_suite(keep_only, starting_with)
1188
suite = test_suite()
3307
1190
suite = test_suite_factory()
3309
# But always filter as requested.
3310
suite = filter_suite_by_id_startswith(suite, starting_with)
3311
result_decorators = []
3313
result_decorators.append(ProfileResult)
3314
1191
return run_suite(suite, 'testbzr', verbose=verbose, pattern=pattern,
3315
stop_on_failure=stop_on_failure,
1192
stop_on_failure=stop_on_failure, keep_output=keep_output,
3316
1193
transport=transport,
3317
lsprof_timed=lsprof_timed,
3318
bench_history=bench_history,
3319
matching_tests_first=matching_tests_first,
3320
list_only=list_only,
3321
random_seed=random_seed,
3322
exclude_pattern=exclude_pattern,
3324
runner_class=runner_class,
3325
suite_decorators=suite_decorators,
3327
result_decorators=result_decorators,
1194
lsprof_timed=lsprof_timed)
3330
1196
default_transport = old_transport
3331
selftest_debug_flags = old_debug_flags
3334
def load_test_id_list(file_name):
3335
"""Load a test id list from a text file.
3337
The format is one test id by line. No special care is taken to impose
3338
strict rules, these test ids are used to filter the test suite so a test id
3339
that do not match an existing test will do no harm. This allows user to add
3340
comments, leave blank lines, etc.
3344
ftest = open(file_name, 'rt')
3346
if e.errno != errno.ENOENT:
3349
raise errors.NoSuchFile(file_name)
3351
for test_name in ftest.readlines():
3352
test_list.append(test_name.strip())
3357
def suite_matches_id_list(test_suite, id_list):
3358
"""Warns about tests not appearing or appearing more than once.
3360
:param test_suite: A TestSuite object.
3361
:param test_id_list: The list of test ids that should be found in
3364
:return: (absents, duplicates) absents is a list containing the test found
3365
in id_list but not in test_suite, duplicates is a list containing the
3366
test found multiple times in test_suite.
3368
When using a prefined test id list, it may occurs that some tests do not
3369
exist anymore or that some tests use the same id. This function warns the
3370
tester about potential problems in his workflow (test lists are volatile)
3371
or in the test suite itself (using the same id for several tests does not
3372
help to localize defects).
3374
# Build a dict counting id occurrences
3376
for test in iter_suite_tests(test_suite):
3378
tests[id] = tests.get(id, 0) + 1
3383
occurs = tests.get(id, 0)
3385
not_found.append(id)
3387
duplicates.append(id)
3389
return not_found, duplicates
3392
class TestIdList(object):
3393
"""Test id list to filter a test suite.
3395
Relying on the assumption that test ids are built as:
3396
<module>[.<class>.<method>][(<param>+)], <module> being in python dotted
3397
notation, this class offers methods to :
3398
- avoid building a test suite for modules not refered to in the test list,
3399
- keep only the tests listed from the module test suite.
3402
def __init__(self, test_id_list):
3403
# When a test suite needs to be filtered against us we compare test ids
3404
# for equality, so a simple dict offers a quick and simple solution.
3405
self.tests = dict().fromkeys(test_id_list, True)
3407
# While unittest.TestCase have ids like:
3408
# <module>.<class>.<method>[(<param+)],
3409
# doctest.DocTestCase can have ids like:
3412
# <module>.<function>
3413
# <module>.<class>.<method>
3415
# Since we can't predict a test class from its name only, we settle on
3416
# a simple constraint: a test id always begins with its module name.
3419
for test_id in test_id_list:
3420
parts = test_id.split('.')
3421
mod_name = parts.pop(0)
3422
modules[mod_name] = True
3424
mod_name += '.' + part
3425
modules[mod_name] = True
3426
self.modules = modules
3428
def refers_to(self, module_name):
3429
"""Is there tests for the module or one of its sub modules."""
3430
return self.modules.has_key(module_name)
3432
def includes(self, test_id):
3433
return self.tests.has_key(test_id)
3436
class TestPrefixAliasRegistry(registry.Registry):
3437
"""A registry for test prefix aliases.
3439
This helps implement shorcuts for the --starting-with selftest
3440
option. Overriding existing prefixes is not allowed but not fatal (a
3441
warning will be emitted).
3444
def register(self, key, obj, help=None, info=None,
3445
override_existing=False):
3446
"""See Registry.register.
3448
Trying to override an existing alias causes a warning to be emitted,
3449
not a fatal execption.
3452
super(TestPrefixAliasRegistry, self).register(
3453
key, obj, help=help, info=info, override_existing=False)
3455
actual = self.get(key)
3456
note('Test prefix alias %s is already used for %s, ignoring %s'
3457
% (key, actual, obj))
3459
def resolve_alias(self, id_start):
3460
"""Replace the alias by the prefix in the given string.
3462
Using an unknown prefix is an error to help catching typos.
3464
parts = id_start.split('.')
3466
parts[0] = self.get(parts[0])
3468
raise errors.BzrCommandError(
3469
'%s is not a known test prefix alias' % parts[0])
3470
return '.'.join(parts)
3473
test_prefix_alias_registry = TestPrefixAliasRegistry()
3474
"""Registry of test prefix aliases."""
3477
# This alias allows to detect typos ('bzrlin.') by making all valid test ids
3478
# appear prefixed ('bzrlib.' is "replaced" by 'bzrlib.').
3479
test_prefix_alias_registry.register('bzrlib', 'bzrlib')
3481
# Obvious higest levels prefixes, feel free to add your own via a plugin
3482
test_prefix_alias_registry.register('bd', 'bzrlib.doc')
3483
test_prefix_alias_registry.register('bu', 'bzrlib.utils')
3484
test_prefix_alias_registry.register('bt', 'bzrlib.tests')
3485
test_prefix_alias_registry.register('bb', 'bzrlib.tests.blackbox')
3486
test_prefix_alias_registry.register('bp', 'bzrlib.plugins')
3489
def _test_suite_testmod_names():
3490
"""Return the standard list of test module names to test."""
3493
'bzrlib.tests.blackbox',
3494
'bzrlib.tests.commands',
3495
'bzrlib.tests.per_branch',
3496
'bzrlib.tests.per_bzrdir',
3497
'bzrlib.tests.per_interrepository',
3498
'bzrlib.tests.per_intertree',
3499
'bzrlib.tests.per_inventory',
3500
'bzrlib.tests.per_interbranch',
3501
'bzrlib.tests.per_lock',
3502
'bzrlib.tests.per_transport',
3503
'bzrlib.tests.per_tree',
3504
'bzrlib.tests.per_pack_repository',
3505
'bzrlib.tests.per_repository',
3506
'bzrlib.tests.per_repository_chk',
3507
'bzrlib.tests.per_repository_reference',
3508
'bzrlib.tests.per_versionedfile',
3509
'bzrlib.tests.per_workingtree',
3510
'bzrlib.tests.test__annotator',
3511
'bzrlib.tests.test__chk_map',
3512
'bzrlib.tests.test__dirstate_helpers',
3513
'bzrlib.tests.test__groupcompress',
3514
'bzrlib.tests.test__known_graph',
3515
'bzrlib.tests.test__rio',
3516
'bzrlib.tests.test__static_tuple',
3517
'bzrlib.tests.test__static_tuple_interned',
3518
'bzrlib.tests.test__walkdirs_win32',
3519
'bzrlib.tests.test_ancestry',
3520
'bzrlib.tests.test_annotate',
3521
'bzrlib.tests.test_api',
3522
'bzrlib.tests.test_atomicfile',
3523
'bzrlib.tests.test_bad_files',
3524
'bzrlib.tests.test_bencode',
3525
'bzrlib.tests.test_bisect_multi',
3526
'bzrlib.tests.test_branch',
3527
'bzrlib.tests.test_branchbuilder',
3528
'bzrlib.tests.test_btree_index',
3529
'bzrlib.tests.test_bugtracker',
3530
'bzrlib.tests.test_bundle',
3531
'bzrlib.tests.test_bzrdir',
3532
'bzrlib.tests.test__chunks_to_lines',
3533
'bzrlib.tests.test_cache_utf8',
3534
'bzrlib.tests.test_chk_map',
3535
'bzrlib.tests.test_chk_serializer',
3536
'bzrlib.tests.test_chunk_writer',
3537
'bzrlib.tests.test_clean_tree',
3538
'bzrlib.tests.test_commands',
3539
'bzrlib.tests.test_commit',
3540
'bzrlib.tests.test_commit_merge',
3541
'bzrlib.tests.test_config',
3542
'bzrlib.tests.test_conflicts',
3543
'bzrlib.tests.test_counted_lock',
3544
'bzrlib.tests.test_crash',
3545
'bzrlib.tests.test_decorators',
3546
'bzrlib.tests.test_delta',
3547
'bzrlib.tests.test_debug',
3548
'bzrlib.tests.test_deprecated_graph',
3549
'bzrlib.tests.test_diff',
3550
'bzrlib.tests.test_directory_service',
3551
'bzrlib.tests.test_dirstate',
3552
'bzrlib.tests.test_email_message',
3553
'bzrlib.tests.test_eol_filters',
3554
'bzrlib.tests.test_errors',
3555
'bzrlib.tests.test_export',
3556
'bzrlib.tests.test_extract',
3557
'bzrlib.tests.test_fetch',
3558
'bzrlib.tests.test_fifo_cache',
3559
'bzrlib.tests.test_filters',
3560
'bzrlib.tests.test_ftp_transport',
3561
'bzrlib.tests.test_foreign',
3562
'bzrlib.tests.test_generate_docs',
3563
'bzrlib.tests.test_generate_ids',
3564
'bzrlib.tests.test_globbing',
3565
'bzrlib.tests.test_gpg',
3566
'bzrlib.tests.test_graph',
3567
'bzrlib.tests.test_groupcompress',
3568
'bzrlib.tests.test_hashcache',
3569
'bzrlib.tests.test_help',
3570
'bzrlib.tests.test_hooks',
3571
'bzrlib.tests.test_http',
3572
'bzrlib.tests.test_http_response',
3573
'bzrlib.tests.test_https_ca_bundle',
3574
'bzrlib.tests.test_identitymap',
3575
'bzrlib.tests.test_ignores',
3576
'bzrlib.tests.test_index',
3577
'bzrlib.tests.test_info',
3578
'bzrlib.tests.test_inv',
3579
'bzrlib.tests.test_inventory_delta',
3580
'bzrlib.tests.test_knit',
3581
'bzrlib.tests.test_lazy_import',
3582
'bzrlib.tests.test_lazy_regex',
3583
'bzrlib.tests.test_lock',
3584
'bzrlib.tests.test_lockable_files',
3585
'bzrlib.tests.test_lockdir',
3586
'bzrlib.tests.test_log',
3587
'bzrlib.tests.test_lru_cache',
3588
'bzrlib.tests.test_lsprof',
3589
'bzrlib.tests.test_mail_client',
3590
'bzrlib.tests.test_memorytree',
3591
'bzrlib.tests.test_merge',
3592
'bzrlib.tests.test_merge3',
3593
'bzrlib.tests.test_merge_core',
3594
'bzrlib.tests.test_merge_directive',
3595
'bzrlib.tests.test_missing',
3596
'bzrlib.tests.test_msgeditor',
3597
'bzrlib.tests.test_multiparent',
3598
'bzrlib.tests.test_mutabletree',
3599
'bzrlib.tests.test_nonascii',
3600
'bzrlib.tests.test_options',
3601
'bzrlib.tests.test_osutils',
3602
'bzrlib.tests.test_osutils_encodings',
3603
'bzrlib.tests.test_pack',
3604
'bzrlib.tests.test_patch',
3605
'bzrlib.tests.test_patches',
3606
'bzrlib.tests.test_permissions',
3607
'bzrlib.tests.test_plugins',
3608
'bzrlib.tests.test_progress',
3609
'bzrlib.tests.test_read_bundle',
3610
'bzrlib.tests.test_reconcile',
3611
'bzrlib.tests.test_reconfigure',
3612
'bzrlib.tests.test_registry',
3613
'bzrlib.tests.test_remote',
3614
'bzrlib.tests.test_rename_map',
3615
'bzrlib.tests.test_repository',
3616
'bzrlib.tests.test_revert',
3617
'bzrlib.tests.test_revision',
3618
'bzrlib.tests.test_revisionspec',
3619
'bzrlib.tests.test_revisiontree',
3620
'bzrlib.tests.test_rio',
3621
'bzrlib.tests.test_rules',
3622
'bzrlib.tests.test_sampler',
3623
'bzrlib.tests.test_selftest',
3624
'bzrlib.tests.test_serializer',
3625
'bzrlib.tests.test_setup',
3626
'bzrlib.tests.test_sftp_transport',
3627
'bzrlib.tests.test_shelf',
3628
'bzrlib.tests.test_shelf_ui',
3629
'bzrlib.tests.test_smart',
3630
'bzrlib.tests.test_smart_add',
3631
'bzrlib.tests.test_smart_request',
3632
'bzrlib.tests.test_smart_transport',
3633
'bzrlib.tests.test_smtp_connection',
3634
'bzrlib.tests.test_source',
3635
'bzrlib.tests.test_ssh_transport',
3636
'bzrlib.tests.test_status',
3637
'bzrlib.tests.test_store',
3638
'bzrlib.tests.test_strace',
3639
'bzrlib.tests.test_subsume',
3640
'bzrlib.tests.test_switch',
3641
'bzrlib.tests.test_symbol_versioning',
3642
'bzrlib.tests.test_tag',
3643
'bzrlib.tests.test_testament',
3644
'bzrlib.tests.test_textfile',
3645
'bzrlib.tests.test_textmerge',
3646
'bzrlib.tests.test_timestamp',
3647
'bzrlib.tests.test_trace',
3648
'bzrlib.tests.test_transactions',
3649
'bzrlib.tests.test_transform',
3650
'bzrlib.tests.test_transport',
3651
'bzrlib.tests.test_transport_log',
3652
'bzrlib.tests.test_tree',
3653
'bzrlib.tests.test_treebuilder',
3654
'bzrlib.tests.test_tsort',
3655
'bzrlib.tests.test_tuned_gzip',
3656
'bzrlib.tests.test_ui',
3657
'bzrlib.tests.test_uncommit',
3658
'bzrlib.tests.test_upgrade',
3659
'bzrlib.tests.test_upgrade_stacked',
3660
'bzrlib.tests.test_urlutils',
3661
'bzrlib.tests.test_version',
3662
'bzrlib.tests.test_version_info',
3663
'bzrlib.tests.test_weave',
3664
'bzrlib.tests.test_whitebox',
3665
'bzrlib.tests.test_win32utils',
3666
'bzrlib.tests.test_workingtree',
3667
'bzrlib.tests.test_workingtree_4',
3668
'bzrlib.tests.test_wsgi',
3669
'bzrlib.tests.test_xml',
3673
def _test_suite_modules_to_doctest():
3674
"""Return the list of modules to doctest."""
3677
'bzrlib.branchbuilder',
3680
'bzrlib.iterablefile',
3684
'bzrlib.symbol_versioning',
3687
'bzrlib.version_info_formats.format_custom',
3691
def test_suite(keep_only=None, starting_with=None):
3692
1200
"""Build and return TestSuite for the whole of bzrlib.
3694
:param keep_only: A list of test ids limiting the suite returned.
3696
:param starting_with: An id limiting the suite returned to the tests
3699
1202
This function can be replaced if you need to change the default test
3700
1203
suite on a global basis, but it is not encouraged.
1206
'bzrlib.tests.test_ancestry',
1207
'bzrlib.tests.test_api',
1208
'bzrlib.tests.test_bad_files',
1209
'bzrlib.tests.test_branch',
1210
'bzrlib.tests.test_bundle',
1211
'bzrlib.tests.test_bzrdir',
1212
'bzrlib.tests.test_command',
1213
'bzrlib.tests.test_commit',
1214
'bzrlib.tests.test_commit_merge',
1215
'bzrlib.tests.test_config',
1216
'bzrlib.tests.test_conflicts',
1217
'bzrlib.tests.test_decorators',
1218
'bzrlib.tests.test_diff',
1219
'bzrlib.tests.test_doc_generate',
1220
'bzrlib.tests.test_emptytree',
1221
'bzrlib.tests.test_errors',
1222
'bzrlib.tests.test_escaped_store',
1223
'bzrlib.tests.test_fetch',
1224
'bzrlib.tests.test_gpg',
1225
'bzrlib.tests.test_graph',
1226
'bzrlib.tests.test_hashcache',
1227
'bzrlib.tests.test_http',
1228
'bzrlib.tests.test_identitymap',
1229
'bzrlib.tests.test_inv',
1230
'bzrlib.tests.test_knit',
1231
'bzrlib.tests.test_lockdir',
1232
'bzrlib.tests.test_lockable_files',
1233
'bzrlib.tests.test_log',
1234
'bzrlib.tests.test_merge',
1235
'bzrlib.tests.test_merge3',
1236
'bzrlib.tests.test_merge_core',
1237
'bzrlib.tests.test_missing',
1238
'bzrlib.tests.test_msgeditor',
1239
'bzrlib.tests.test_nonascii',
1240
'bzrlib.tests.test_options',
1241
'bzrlib.tests.test_osutils',
1242
'bzrlib.tests.test_patch',
1243
'bzrlib.tests.test_patches',
1244
'bzrlib.tests.test_permissions',
1245
'bzrlib.tests.test_plugins',
1246
'bzrlib.tests.test_progress',
1247
'bzrlib.tests.test_reconcile',
1248
'bzrlib.tests.test_repository',
1249
'bzrlib.tests.test_revision',
1250
'bzrlib.tests.test_revisionnamespaces',
1251
'bzrlib.tests.test_revprops',
1252
'bzrlib.tests.test_revisiontree',
1253
'bzrlib.tests.test_rio',
1254
'bzrlib.tests.test_sampler',
1255
'bzrlib.tests.test_selftest',
1256
'bzrlib.tests.test_setup',
1257
'bzrlib.tests.test_sftp_transport',
1258
'bzrlib.tests.test_smart_add',
1259
'bzrlib.tests.test_source',
1260
'bzrlib.tests.test_status',
1261
'bzrlib.tests.test_store',
1262
'bzrlib.tests.test_symbol_versioning',
1263
'bzrlib.tests.test_testament',
1264
'bzrlib.tests.test_textfile',
1265
'bzrlib.tests.test_textmerge',
1266
'bzrlib.tests.test_trace',
1267
'bzrlib.tests.test_transactions',
1268
'bzrlib.tests.test_transform',
1269
'bzrlib.tests.test_transport',
1270
'bzrlib.tests.test_tsort',
1271
'bzrlib.tests.test_tuned_gzip',
1272
'bzrlib.tests.test_ui',
1273
'bzrlib.tests.test_upgrade',
1274
'bzrlib.tests.test_urlutils',
1275
'bzrlib.tests.test_versionedfile',
1276
'bzrlib.tests.test_weave',
1277
'bzrlib.tests.test_whitebox',
1278
'bzrlib.tests.test_workingtree',
1279
'bzrlib.tests.test_xml',
1281
test_transport_implementations = [
1282
'bzrlib.tests.test_transport_implementations',
1283
'bzrlib.tests.test_read_bundle',
1285
suite = TestUtil.TestSuite()
3703
1286
loader = TestUtil.TestLoader()
3705
if keep_only is not None:
3706
id_filter = TestIdList(keep_only)
3708
# We take precedence over keep_only because *at loading time* using
3709
# both options means we will load less tests for the same final result.
3710
def interesting_module(name):
3711
for start in starting_with:
3713
# Either the module name starts with the specified string
3714
name.startswith(start)
3715
# or it may contain tests starting with the specified string
3716
or start.startswith(name)
3720
loader = TestUtil.FilteredByModuleTestLoader(interesting_module)
3722
elif keep_only is not None:
3723
loader = TestUtil.FilteredByModuleTestLoader(id_filter.refers_to)
3724
def interesting_module(name):
3725
return id_filter.refers_to(name)
3728
loader = TestUtil.TestLoader()
3729
def interesting_module(name):
3730
# No filtering, all modules are interesting
3733
suite = loader.suiteClass()
3735
# modules building their suite with loadTestsFromModuleNames
3736
suite.addTest(loader.loadTestsFromModuleNames(_test_suite_testmod_names()))
3738
for mod in _test_suite_modules_to_doctest():
3739
if not interesting_module(mod):
3740
# No tests to keep here, move along
3743
# note that this really does mean "report only" -- doctest
3744
# still runs the rest of the examples
3745
doc_suite = doctest.DocTestSuite(mod,
3746
optionflags=doctest.REPORT_ONLY_FIRST_FAILURE)
3747
except ValueError, e:
3748
print '**failed to get doctest for: %s\n%s' % (mod, e)
3750
if len(doc_suite._tests) == 0:
3751
raise errors.BzrError("no doctests found in %s" % (mod,))
3752
suite.addTest(doc_suite)
3754
default_encoding = sys.getdefaultencoding()
3755
for name, plugin in bzrlib.plugin.plugins().items():
3756
if not interesting_module(plugin.module.__name__):
3758
plugin_suite = plugin.test_suite()
3759
# We used to catch ImportError here and turn it into just a warning,
3760
# but really if you don't have --no-plugins this should be a failure.
3761
# mbp 20080213 - see http://bugs.launchpad.net/bugs/189771
3762
if plugin_suite is None:
3763
plugin_suite = plugin.load_plugin_tests(loader)
3764
if plugin_suite is not None:
3765
suite.addTest(plugin_suite)
3766
if default_encoding != sys.getdefaultencoding():
3767
bzrlib.trace.warning(
3768
'Plugin "%s" tried to reset default encoding to: %s', name,
3769
sys.getdefaultencoding())
3771
sys.setdefaultencoding(default_encoding)
3773
if keep_only is not None:
3774
# Now that the referred modules have loaded their tests, keep only the
3776
suite = filter_suite_by_id_list(suite, id_filter)
3777
# Do some sanity checks on the id_list filtering
3778
not_found, duplicates = suite_matches_id_list(suite, keep_only)
3780
# The tester has used both keep_only and starting_with, so he is
3781
# already aware that some tests are excluded from the list, there
3782
# is no need to tell him which.
3785
# Some tests mentioned in the list are not in the test suite. The
3786
# list may be out of date, report to the tester.
3787
for id in not_found:
3788
bzrlib.trace.warning('"%s" not found in the test suite', id)
3789
for id in duplicates:
3790
bzrlib.trace.warning('"%s" is used as an id by several tests', id)
1287
from bzrlib.transport import TransportTestProviderAdapter
1288
adapter = TransportTestProviderAdapter()
1289
adapt_modules(test_transport_implementations, adapter, loader, suite)
1290
suite.addTest(loader.loadTestsFromModuleNames(testmod_names))
1291
for package in packages_to_test():
1292
suite.addTest(package.test_suite())
1293
for m in MODULES_TO_TEST:
1294
suite.addTest(loader.loadTestsFromModule(m))
1295
for m in MODULES_TO_DOCTEST:
1296
suite.addTest(doctest.DocTestSuite(m))
1297
for name, plugin in bzrlib.plugin.all_plugins().items():
1298
if getattr(plugin, 'test_suite', None) is not None:
1299
suite.addTest(plugin.test_suite())
3795
def multiply_scenarios(scenarios_left, scenarios_right):
3796
"""Multiply two sets of scenarios.
3798
:returns: the cartesian product of the two sets of scenarios, that is
3799
a scenario for every possible combination of a left scenario and a
3803
('%s,%s' % (left_name, right_name),
3804
dict(left_dict.items() + right_dict.items()))
3805
for left_name, left_dict in scenarios_left
3806
for right_name, right_dict in scenarios_right]
3809
def multiply_tests(tests, scenarios, result):
3810
"""Multiply tests_list by scenarios into result.
3812
This is the core workhorse for test parameterisation.
3814
Typically the load_tests() method for a per-implementation test suite will
3815
call multiply_tests and return the result.
3817
:param tests: The tests to parameterise.
3818
:param scenarios: The scenarios to apply: pairs of (scenario_name,
3819
scenario_param_dict).
3820
:param result: A TestSuite to add created tests to.
3822
This returns the passed in result TestSuite with the cross product of all
3823
the tests repeated once for each scenario. Each test is adapted by adding
3824
the scenario name at the end of its id(), and updating the test object's
3825
__dict__ with the scenario_param_dict.
3827
>>> import bzrlib.tests.test_sampler
3828
>>> r = multiply_tests(
3829
... bzrlib.tests.test_sampler.DemoTest('test_nothing'),
3830
... [('one', dict(param=1)),
3831
... ('two', dict(param=2))],
3833
>>> tests = list(iter_suite_tests(r))
3837
'bzrlib.tests.test_sampler.DemoTest.test_nothing(one)'
3843
for test in iter_suite_tests(tests):
3844
apply_scenarios(test, scenarios, result)
3848
def apply_scenarios(test, scenarios, result):
3849
"""Apply the scenarios in scenarios to test and add to result.
3851
:param test: The test to apply scenarios to.
3852
:param scenarios: An iterable of scenarios to apply to test.
3854
:seealso: apply_scenario
3856
for scenario in scenarios:
3857
result.addTest(apply_scenario(test, scenario))
3861
def apply_scenario(test, scenario):
3862
"""Copy test and apply scenario to it.
3864
:param test: A test to adapt.
3865
:param scenario: A tuple describing the scenarion.
3866
The first element of the tuple is the new test id.
3867
The second element is a dict containing attributes to set on the
3869
:return: The adapted test.
3871
new_id = "%s(%s)" % (test.id(), scenario[0])
3872
new_test = clone_test(test, new_id)
3873
for name, value in scenario[1].items():
3874
setattr(new_test, name, value)
3878
def clone_test(test, new_id):
3879
"""Clone a test giving it a new id.
3881
:param test: The test to clone.
3882
:param new_id: The id to assign to it.
3883
:return: The new test.
3885
new_test = copy(test)
3886
new_test.id = lambda: new_id
3890
def _rmtree_temp_dir(dirname):
3891
# If LANG=C we probably have created some bogus paths
3892
# which rmtree(unicode) will fail to delete
3893
# so make sure we are using rmtree(str) to delete everything
3894
# except on win32, where rmtree(str) will fail
3895
# since it doesn't have the property of byte-stream paths
3896
# (they are either ascii or mbcs)
3897
if sys.platform == 'win32':
3898
# make sure we are using the unicode win32 api
3899
dirname = unicode(dirname)
3901
dirname = dirname.encode(sys.getfilesystemencoding())
3903
osutils.rmtree(dirname)
3905
# We don't want to fail here because some useful display will be lost
3906
# otherwise. Polluting the tmp dir is bad, but not giving all the
3907
# possible info to the test runner is even worse.
3908
sys.stderr.write('Unable to remove testing dir %s\n%s'
3909
% (os.path.basename(dirname), e))
3912
class Feature(object):
3913
"""An operating system Feature."""
3916
self._available = None
3918
def available(self):
3919
"""Is the feature available?
3921
:return: True if the feature is available.
3923
if self._available is None:
3924
self._available = self._probe()
3925
return self._available
3928
"""Implement this method in concrete features.
3930
:return: True if the feature is available.
3932
raise NotImplementedError
3935
if getattr(self, 'feature_name', None):
3936
return self.feature_name()
3937
return self.__class__.__name__
3940
class _SymlinkFeature(Feature):
3943
return osutils.has_symlinks()
3945
def feature_name(self):
3948
SymlinkFeature = _SymlinkFeature()
3951
class _HardlinkFeature(Feature):
3954
return osutils.has_hardlinks()
3956
def feature_name(self):
3959
HardlinkFeature = _HardlinkFeature()
3962
class _OsFifoFeature(Feature):
3965
return getattr(os, 'mkfifo', None)
3967
def feature_name(self):
3968
return 'filesystem fifos'
3970
OsFifoFeature = _OsFifoFeature()
3973
class _UnicodeFilenameFeature(Feature):
3974
"""Does the filesystem support Unicode filenames?"""
3978
# Check for character combinations unlikely to be covered by any
3979
# single non-unicode encoding. We use the characters
3980
# - greek small letter alpha (U+03B1) and
3981
# - braille pattern dots-123456 (U+283F).
3982
os.stat(u'\u03b1\u283f')
3983
except UnicodeEncodeError:
3985
except (IOError, OSError):
3986
# The filesystem allows the Unicode filename but the file doesn't
3990
# The filesystem allows the Unicode filename and the file exists,
3994
UnicodeFilenameFeature = _UnicodeFilenameFeature()
3997
def probe_unicode_in_user_encoding():
3998
"""Try to encode several unicode strings to use in unicode-aware tests.
3999
Return first successfull match.
4001
:return: (unicode value, encoded plain string value) or (None, None)
4003
possible_vals = [u'm\xb5', u'\xe1', u'\u0410']
4004
for uni_val in possible_vals:
4006
str_val = uni_val.encode(osutils.get_user_encoding())
4007
except UnicodeEncodeError:
4008
# Try a different character
4011
return uni_val, str_val
4015
def probe_bad_non_ascii(encoding):
4016
"""Try to find [bad] character with code [128..255]
4017
that cannot be decoded to unicode in some encoding.
4018
Return None if all non-ascii characters is valid
4021
for i in xrange(128, 256):
4024
char.decode(encoding)
4025
except UnicodeDecodeError:
4030
class _HTTPSServerFeature(Feature):
4031
"""Some tests want an https Server, check if one is available.
4033
Right now, the only way this is available is under python2.6 which provides
4044
def feature_name(self):
4045
return 'HTTPSServer'
4048
HTTPSServerFeature = _HTTPSServerFeature()
4051
class _UnicodeFilename(Feature):
4052
"""Does the filesystem support Unicode filenames?"""
4057
except UnicodeEncodeError:
4059
except (IOError, OSError):
4060
# The filesystem allows the Unicode filename but the file doesn't
4064
# The filesystem allows the Unicode filename and the file exists,
4068
UnicodeFilename = _UnicodeFilename()
4071
class _UTF8Filesystem(Feature):
4072
"""Is the filesystem UTF-8?"""
4075
if osutils._fs_enc.upper() in ('UTF-8', 'UTF8'):
4079
UTF8Filesystem = _UTF8Filesystem()
4082
class _CaseInsCasePresFilenameFeature(Feature):
4083
"""Is the file-system case insensitive, but case-preserving?"""
4086
fileno, name = tempfile.mkstemp(prefix='MixedCase')
4088
# first check truly case-preserving for created files, then check
4089
# case insensitive when opening existing files.
4090
name = osutils.normpath(name)
4091
base, rel = osutils.split(name)
4092
found_rel = osutils.canonical_relpath(base, name)
4093
return (found_rel == rel
4094
and os.path.isfile(name.upper())
4095
and os.path.isfile(name.lower()))
4100
def feature_name(self):
4101
return "case-insensitive case-preserving filesystem"
4103
CaseInsCasePresFilenameFeature = _CaseInsCasePresFilenameFeature()
4106
class _CaseInsensitiveFilesystemFeature(Feature):
4107
"""Check if underlying filesystem is case-insensitive but *not* case
4110
# Note that on Windows, Cygwin, MacOS etc, the file-systems are far
4111
# more likely to be case preserving, so this case is rare.
4114
if CaseInsCasePresFilenameFeature.available():
4117
if TestCaseWithMemoryTransport.TEST_ROOT is None:
4118
root = osutils.mkdtemp(prefix='testbzr-', suffix='.tmp')
4119
TestCaseWithMemoryTransport.TEST_ROOT = root
4121
root = TestCaseWithMemoryTransport.TEST_ROOT
4122
tdir = osutils.mkdtemp(prefix='case-sensitive-probe-', suffix='',
4124
name_a = osutils.pathjoin(tdir, 'a')
4125
name_A = osutils.pathjoin(tdir, 'A')
4127
result = osutils.isdir(name_A)
4128
_rmtree_temp_dir(tdir)
4131
def feature_name(self):
4132
return 'case-insensitive filesystem'
4134
CaseInsensitiveFilesystemFeature = _CaseInsensitiveFilesystemFeature()
4137
class _SubUnitFeature(Feature):
4138
"""Check if subunit is available."""
4147
def feature_name(self):
4150
SubUnitFeature = _SubUnitFeature()
4151
# Only define SubUnitBzrRunner if subunit is available.
4153
from subunit import TestProtocolClient
4155
from subunit.test_results import AutoTimingTestResultDecorator
4157
AutoTimingTestResultDecorator = lambda x:x
4158
class SubUnitBzrRunner(TextTestRunner):
4159
def run(self, test):
4160
result = AutoTimingTestResultDecorator(
4161
TestProtocolClient(self.stream))
1303
def adapt_modules(mods_list, adapter, loader, suite):
1304
"""Adapt the modules in mods_list using adapter and add to suite."""
1305
for test in iter_suite_tests(loader.loadTestsFromModuleNames(mods_list)):
1306
suite.addTests(adapter.adapt(test))
added
removed
bzrlib/tests/__init__.py
