27
31
from cStringIO import StringIO
37
from pprint import pformat
42
from subprocess import Popen, PIPE
50
# nb: check this before importing anything else from within it
51
_testtools_version = getattr(testtools, '__version__', ())
52
if _testtools_version < (0, 9, 5):
53
raise ImportError("need at least testtools 0.9.5: %s is %r"
54
% (testtools.__file__, _testtools_version))
55
from testtools import content
58
51
from bzrlib import (
62
commands as _mod_commands,
71
plugin as _mod_plugin,
78
transport as _mod_transport,
65
import bzrlib.commands
66
import bzrlib.timestamp
68
import bzrlib.inventory
69
import bzrlib.iterablefile
82
72
import bzrlib.lsprof
83
73
except ImportError:
84
74
# lsprof not available
86
from bzrlib.smart import client, request
87
from bzrlib.transport import (
76
from bzrlib.merge import merge_inner
80
from bzrlib import symbol_versioning
91
81
from bzrlib.symbol_versioning import (
92
83
deprecated_function,
95
from bzrlib.tests import (
100
from bzrlib.ui import NullProgressView
101
from bzrlib.ui.text import TextUIFactory
88
from bzrlib.transport import get_transport
89
import bzrlib.transport
90
from bzrlib.transport.local import LocalURLServer
91
from bzrlib.transport.memory import MemoryServer
92
from bzrlib.transport.readonly import ReadonlyServer
93
from bzrlib.trace import mutter, note
94
from bzrlib.tests import TestUtil
95
from bzrlib.tests.http_server import HttpServer
96
from bzrlib.tests.TestUtil import (
100
from bzrlib.tests.treeshape import build_tree_contents
101
import bzrlib.version_info_formats.format_custom
102
from bzrlib.workingtree import WorkingTree, WorkingTreeFormat2
103
104
# Mark this python module as being part of the implementation
104
105
# of unittest: this gives us better tracebacks where the last
105
106
# shown frame is the test code, not our assertXYZ.
108
default_transport = test_server.LocalURLServer
111
_unitialized_attr = object()
112
"""A sentinel needed to act as a default value in a method signature."""
115
# Subunit result codes, defined here to prevent a hard dependency on subunit.
119
# These are intentionally brought into this namespace. That way plugins, etc
120
# can just "from bzrlib.tests import TestCase, TestLoader, etc"
121
TestSuite = TestUtil.TestSuite
122
TestLoader = TestUtil.TestLoader
124
# Tests should run in a clean and clearly defined environment. The goal is to
125
# keep them isolated from the running environment as mush as possible. The test
126
# framework ensures the variables defined below are set (or deleted if the
127
# value is None) before a test is run and reset to their original value after
128
# the test is run. Generally if some code depends on an environment variable,
129
# the tests should start without this variable in the environment. There are a
130
# few exceptions but you shouldn't violate this rule lightly.
134
# bzr now uses the Win32 API and doesn't rely on APPDATA, but the
135
# tests do check our impls match APPDATA
136
'BZR_EDITOR': None, # test_msgeditor manipulates this variable
140
'BZREMAIL': None, # may still be present in the environment
141
'EMAIL': 'jrandom@example.com', # set EMAIL as bzr does not guess
142
'BZR_PROGRESS_BAR': None,
143
# This should trap leaks to ~/.bzr.log. This occurs when tests use TestCase
144
# as a base class instead of TestCaseInTempDir. Tests inheriting from
145
# TestCase should not use disk resources, BZR_LOG is one.
146
'BZR_LOG': '/you-should-use-TestCaseInTempDir-if-you-need-a-log-file',
147
'BZR_PLUGIN_PATH': None,
148
'BZR_DISABLE_PLUGINS': None,
149
'BZR_PLUGINS_AT': None,
150
'BZR_CONCURRENCY': None,
151
# Make sure that any text ui tests are consistent regardless of
152
# the environment the test case is run in; you may want tests that
153
# test other combinations. 'dumb' is a reasonable guess for tests
154
# going to a pipe or a StringIO.
160
'SSH_AUTH_SOCK': None,
170
# Nobody cares about ftp_proxy, FTP_PROXY AFAIK. So far at
171
# least. If you do (care), please update this comment
175
'BZR_REMOTE_PATH': None,
176
# Generally speaking, we don't want apport reporting on crashes in
177
# the test envirnoment unless we're specifically testing apport,
178
# so that it doesn't leak into the real system environment. We
179
# use an env var so it propagates to subprocesses.
180
'APPORT_DISABLE': '1',
184
def override_os_environ(test, env=None):
185
"""Modify os.environ keeping a copy.
187
:param test: A test instance
189
:param env: A dict containing variable definitions to be installed
192
env = isolated_environ
193
test._original_os_environ = dict([(var, value)
194
for var, value in os.environ.iteritems()])
195
for var, value in env.iteritems():
196
osutils.set_or_unset_env(var, value)
197
if var not in test._original_os_environ:
198
# The var is new, add it with a value of None, so
199
# restore_os_environ will delete it
200
test._original_os_environ[var] = None
203
def restore_os_environ(test):
204
"""Restore os.environ to its original state.
206
:param test: A test instance previously passed to override_os_environ.
208
for var, value in test._original_os_environ.iteritems():
209
# Restore the original value (or delete it if the value has been set to
210
# None in override_os_environ).
211
osutils.set_or_unset_env(var, value)
214
class ExtendedTestResult(testtools.TextTestResult):
109
default_transport = LocalURLServer
112
class ExtendedTestResult(unittest._TextTestResult):
215
113
"""Accepts, reports and accumulates the results of running tests.
217
115
Compared to the unittest version this class adds support for
264
162
self.unsupported = {}
266
164
self._overall_start_time = time.time()
267
self._strict = strict
268
self._first_thread_leaker_id = None
269
self._tests_leaking_threads_count = 0
270
self._traceback_from_test = None
272
def stopTestRun(self):
275
stopTime = time.time()
276
timeTaken = stopTime - self.startTime
277
# GZ 2010-07-19: Seems testtools has no printErrors method, and though
278
# the parent class method is similar have to duplicate
279
self._show_list('ERROR', self.errors)
280
self._show_list('FAIL', self.failures)
281
self.stream.write(self.sep2)
282
self.stream.write("%s %d test%s in %.3fs\n\n" % (actionTaken,
283
run, run != 1 and "s" or "", timeTaken))
284
if not self.wasSuccessful():
285
self.stream.write("FAILED (")
286
failed, errored = map(len, (self.failures, self.errors))
288
self.stream.write("failures=%d" % failed)
290
if failed: self.stream.write(", ")
291
self.stream.write("errors=%d" % errored)
292
if self.known_failure_count:
293
if failed or errored: self.stream.write(", ")
294
self.stream.write("known_failure_count=%d" %
295
self.known_failure_count)
296
self.stream.write(")\n")
298
if self.known_failure_count:
299
self.stream.write("OK (known_failures=%d)\n" %
300
self.known_failure_count)
302
self.stream.write("OK\n")
303
if self.skip_count > 0:
304
skipped = self.skip_count
305
self.stream.write('%d test%s skipped\n' %
306
(skipped, skipped != 1 and "s" or ""))
308
for feature, count in sorted(self.unsupported.items()):
309
self.stream.write("Missing feature '%s' skipped %d tests.\n" %
312
ok = self.wasStrictlySuccessful()
314
ok = self.wasSuccessful()
315
if self._first_thread_leaker_id:
317
'%s is leaking threads among %d leaking tests.\n' % (
318
self._first_thread_leaker_id,
319
self._tests_leaking_threads_count))
320
# We don't report the main thread as an active one.
322
'%d non-main threads were left active in the end.\n'
323
% (len(self._active_threads) - 1))
325
def getDescription(self, test):
328
def _extractBenchmarkTime(self, testCase, details=None):
166
def _extractBenchmarkTime(self, testCase):
329
167
"""Add a benchmark time for the current test case."""
330
if details and 'benchtime' in details:
331
return float(''.join(details['benchtime'].iter_bytes()))
332
168
return getattr(testCase, "_benchtime", None)
334
170
def _elapsedTestTimeString(self):
335
171
"""Return a time string for the overall time the current test has taken."""
336
return self._formatTime(self._delta_to_float(
337
self._now() - self._start_datetime))
172
return self._formatTime(time.time() - self._start_time)
339
174
def _testTimeString(self, testCase):
340
175
benchmark_time = self._extractBenchmarkTime(testCase)
341
176
if benchmark_time is not None:
342
return self._formatTime(benchmark_time) + "*"
178
self._formatTime(benchmark_time),
179
self._elapsedTestTimeString())
344
return self._elapsedTestTimeString()
181
return " %s" % self._elapsedTestTimeString()
346
183
def _formatTime(self, seconds):
347
184
"""Format seconds as milliseconds with leading spaces."""
352
189
def _shortened_test_description(self, test):
354
what = re.sub(r'^bzrlib\.tests\.', '', what)
191
what = re.sub(r'^bzrlib\.(tests|benchmarks)\.', '', what)
357
# GZ 2010-10-04: Cloned tests may end up harmlessly calling this method
358
# multiple times in a row, because the handler is added for
359
# each test but the container list is shared between cases.
360
# See lp:498869 lp:625574 and lp:637725 for background.
361
def _record_traceback_from_test(self, exc_info):
362
"""Store the traceback from passed exc_info tuple till"""
363
self._traceback_from_test = exc_info[2]
365
194
def startTest(self, test):
366
super(ExtendedTestResult, self).startTest(test)
195
unittest.TestResult.startTest(self, test)
370
196
self.report_test_start(test)
371
197
test.number = self.count
372
198
self._recordTestStartTime()
373
# Make testtools cases give us the real traceback on failure
374
addOnException = getattr(test, "addOnException", None)
375
if addOnException is not None:
376
addOnException(self._record_traceback_from_test)
377
# Only check for thread leaks on bzrlib derived test cases
378
if isinstance(test, TestCase):
379
test.addCleanup(self._check_leaked_threads, test)
381
def stopTest(self, test):
382
super(ExtendedTestResult, self).stopTest(test)
383
# Manually break cycles, means touching various private things but hey
384
getDetails = getattr(test, "getDetails", None)
385
if getDetails is not None:
387
# Clear _type_equality_funcs to try to stop TestCase instances
388
# from wasting memory. 'clear' is not available in all Python
389
# versions (bug 809048)
390
type_equality_funcs = getattr(test, "_type_equality_funcs", None)
391
if type_equality_funcs is not None:
392
tef_clear = getattr(type_equality_funcs, "clear", None)
393
if tef_clear is None:
394
tef_instance_dict = getattr(type_equality_funcs, "__dict__", None)
395
if tef_instance_dict is not None:
396
tef_clear = tef_instance_dict.clear
397
if tef_clear is not None:
399
self._traceback_from_test = None
401
def startTests(self):
402
self.report_tests_starting()
403
self._active_threads = threading.enumerate()
405
def _check_leaked_threads(self, test):
406
"""See if any threads have leaked since last call
408
A sample of live threads is stored in the _active_threads attribute,
409
when this method runs it compares the current live threads and any not
410
in the previous sample are treated as having leaked.
412
now_active_threads = set(threading.enumerate())
413
threads_leaked = now_active_threads.difference(self._active_threads)
415
self._report_thread_leak(test, threads_leaked, now_active_threads)
416
self._tests_leaking_threads_count += 1
417
if self._first_thread_leaker_id is None:
418
self._first_thread_leaker_id = test.id()
419
self._active_threads = now_active_threads
421
200
def _recordTestStartTime(self):
422
201
"""Record that a test has started."""
423
self._start_datetime = self._now()
202
self._start_time = time.time()
204
def _cleanupLogFile(self, test):
205
# We can only do this if we have one of our TestCases, not if
207
setKeepLogfile = getattr(test, 'setKeepLogfile', None)
208
if setKeepLogfile is not None:
425
211
def addError(self, test, err):
426
212
"""Tell result that test finished with an error.
441
233
Called from the TestCase run() method when the test
442
234
fails because e.g. an assert() method failed.
444
self._post_mortem(self._traceback_from_test)
445
super(ExtendedTestResult, self).addFailure(test, err)
446
self.failure_count += 1
447
self.report_failure(test, err)
236
self._testConcluded(test)
237
if isinstance(err[1], KnownFailure):
238
return self._addKnownFailure(test, err)
240
unittest.TestResult.addFailure(self, test, err)
241
self.failure_count += 1
242
self.report_failure(test, err)
245
self._cleanupLogFile(test)
451
def addSuccess(self, test, details=None):
247
def addSuccess(self, test):
452
248
"""Tell result that test completed successfully.
454
250
Called from the TestCase run()
252
self._testConcluded(test)
456
253
if self._bench_history is not None:
457
benchmark_time = self._extractBenchmarkTime(test, details)
254
benchmark_time = self._extractBenchmarkTime(test)
458
255
if benchmark_time is not None:
459
256
self._bench_history.write("%s %s\n" % (
460
257
self._formatTime(benchmark_time),
462
259
self.report_success(test)
463
super(ExtendedTestResult, self).addSuccess(test)
260
self._cleanupLogFile(test)
261
unittest.TestResult.addSuccess(self, test)
464
262
test._log_contents = ''
466
def addExpectedFailure(self, test, err):
264
def _testConcluded(self, test):
265
"""Common code when a test has finished.
267
Called regardless of whether it succeded, failed, etc.
271
def _addKnownFailure(self, test, err):
467
272
self.known_failure_count += 1
468
273
self.report_known_failure(test, err)
470
def addUnexpectedSuccess(self, test, details=None):
471
"""Tell result the test unexpectedly passed, counting as a failure
473
When the minimum version of testtools required becomes 0.9.8 this
474
can be updated to use the new handling there.
476
super(ExtendedTestResult, self).addFailure(test, details=details)
477
self.failure_count += 1
478
self.report_unexpected_success(test,
479
"".join(details["reason"].iter_text()))
483
275
def addNotSupported(self, test, feature):
484
276
"""The test will not be run because of a missing feature.
486
278
# this can be called in two different ways: it may be that the
487
# test started running, and then raised (through requireFeature)
279
# test started running, and then raised (through addError)
488
280
# UnavailableFeature. Alternatively this method can be called
489
# while probing for features before running the test code proper; in
490
# that case we will see startTest and stopTest, but the test will
491
# never actually run.
281
# while probing for features before running the tests; in that
282
# case we will see startTest and stopTest, but the test will never
492
284
self.unsupported.setdefault(str(feature), 0)
493
285
self.unsupported[str(feature)] += 1
494
286
self.report_unsupported(test, feature)
496
def addSkip(self, test, reason):
497
"""A test has not run for 'reason'."""
499
self.report_skip(test, reason)
501
def addNotApplicable(self, test, reason):
502
self.not_applicable_count += 1
503
self.report_not_applicable(test, reason)
505
def _post_mortem(self, tb=None):
506
"""Start a PDB post mortem session."""
507
if os.environ.get('BZR_TEST_PDB', None):
511
def progress(self, offset, whence):
512
"""The test is adjusting the count of tests to run."""
513
if whence == SUBUNIT_SEEK_SET:
514
self.num_tests = offset
515
elif whence == SUBUNIT_SEEK_CUR:
516
self.num_tests += offset
518
raise errors.BzrError("Unknown whence %r" % whence)
520
def report_tests_starting(self):
521
"""Display information before the test run begins"""
522
if getattr(sys, 'frozen', None) is None:
523
bzr_path = osutils.realpath(sys.argv[0])
525
bzr_path = sys.executable
527
'bzr selftest: %s\n' % (bzr_path,))
530
bzrlib.__path__[0],))
532
' bzr-%s python-%s %s\n' % (
533
bzrlib.version_string,
534
bzrlib._format_version_tuple(sys.version_info),
535
platform.platform(aliased=1),
537
self.stream.write('\n')
539
def report_test_start(self, test):
540
"""Display information on the test just about to be run"""
542
def _report_thread_leak(self, test, leaked_threads, active_threads):
543
"""Display information on a test that leaked one or more threads"""
544
# GZ 2010-09-09: A leak summary reported separately from the general
545
# thread debugging would be nice. Tests under subunit
546
# need something not using stream, perhaps adding a
547
# testtools details object would be fitting.
548
if 'threads' in selftest_debug_flags:
549
self.stream.write('%s is leaking, active is now %d\n' %
550
(test.id(), len(active_threads)))
552
def startTestRun(self):
553
self.startTime = time.time()
288
def _addSkipped(self, test, skip_excinfo):
289
if isinstance(skip_excinfo[1], TestNotApplicable):
290
self.not_applicable_count += 1
291
self.report_not_applicable(test, skip_excinfo)
294
self.report_skip(test, skip_excinfo)
297
except KeyboardInterrupt:
300
self.addError(test, test._exc_info())
302
# seems best to treat this as success from point-of-view of unittest
303
# -- it actually does nothing so it barely matters :)
304
unittest.TestResult.addSuccess(self, test)
305
test._log_contents = ''
307
def printErrorList(self, flavour, errors):
308
for test, err in errors:
309
self.stream.writeln(self.separator1)
310
self.stream.write("%s: " % flavour)
311
self.stream.writeln(self.getDescription(test))
312
if getattr(test, '_get_log', None) is not None:
313
self.stream.write('\n')
315
('vvvv[log from %s]' % test.id()).ljust(78,'-'))
316
self.stream.write('\n')
317
self.stream.write(test._get_log())
318
self.stream.write('\n')
320
('^^^^[log from %s]' % test.id()).ljust(78,'-'))
321
self.stream.write('\n')
322
self.stream.writeln(self.separator2)
323
self.stream.writeln("%s" % err)
328
def report_cleaning_up(self):
555
331
def report_success(self, test):
694
461
return '%s%s' % (indent, err[1])
696
463
def report_error(self, test, err):
697
self.stream.write('ERROR %s\n%s\n'
464
self.stream.writeln('ERROR %s\n%s'
698
465
% (self._testTimeString(test),
699
466
self._error_summary(err)))
701
468
def report_failure(self, test, err):
702
self.stream.write(' FAIL %s\n%s\n'
469
self.stream.writeln(' FAIL %s\n%s'
703
470
% (self._testTimeString(test),
704
471
self._error_summary(err)))
706
473
def report_known_failure(self, test, err):
707
self.stream.write('XFAIL %s\n%s\n'
474
self.stream.writeln('XFAIL %s\n%s'
708
475
% (self._testTimeString(test),
709
476
self._error_summary(err)))
711
def report_unexpected_success(self, test, reason):
712
self.stream.write(' FAIL %s\n%s: %s\n'
713
% (self._testTimeString(test),
714
"Unexpected success. Should have failed",
717
478
def report_success(self, test):
718
self.stream.write(' OK %s\n' % self._testTimeString(test))
479
self.stream.writeln(' OK %s' % self._testTimeString(test))
719
480
for bench_called, stats in getattr(test, '_benchcalls', []):
720
self.stream.write('LSProf output for %s(%s, %s)\n' % bench_called)
481
self.stream.writeln('LSProf output for %s(%s, %s)' % bench_called)
721
482
stats.pprint(file=self.stream)
722
483
# flush the stream so that we get smooth output. This verbose mode is
723
484
# used to show the output in PQM.
724
485
self.stream.flush()
726
def report_skip(self, test, reason):
727
self.stream.write(' SKIP %s\n%s\n'
728
% (self._testTimeString(test), reason))
487
def report_skip(self, test, skip_excinfo):
488
self.stream.writeln(' SKIP %s\n%s'
489
% (self._testTimeString(test),
490
self._error_summary(skip_excinfo)))
730
def report_not_applicable(self, test, reason):
731
self.stream.write(' N/A %s\n %s\n'
732
% (self._testTimeString(test), reason))
492
def report_not_applicable(self, test, skip_excinfo):
493
self.stream.writeln(' N/A %s\n%s'
494
% (self._testTimeString(test),
495
self._error_summary(skip_excinfo)))
734
497
def report_unsupported(self, test, feature):
735
498
"""test cannot be run because feature is missing."""
736
self.stream.write("NODEP %s\n The feature '%s' is not available.\n"
499
self.stream.writeln("NODEP %s\n The feature '%s' is not available."
737
500
%(self._testTimeString(test), feature))
747
510
bench_history=None,
749
result_decorators=None,
751
"""Create a TextTestRunner.
753
:param result_decorators: An optional list of decorators to apply
754
to the result object being used by the runner. Decorators are
755
applied left to right - the first element in the list is the
758
# stream may know claim to know to write unicode strings, but in older
759
# pythons this goes sufficiently wrong that it is a bad idea. (
760
# specifically a built in file with encoding 'UTF-8' will still try
761
# to encode using ascii.
762
new_encoding = osutils.get_terminal_encoding()
763
codec = codecs.lookup(new_encoding)
764
if type(codec) is tuple:
768
encode = codec.encode
769
# GZ 2010-09-08: Really we don't want to be writing arbitrary bytes,
770
# so should swap to the plain codecs.StreamWriter
771
stream = osutils.UnicodeOrBytesToBytesWriter(encode, stream,
773
stream.encoding = new_encoding
513
self.stream = unittest._WritelnDecorator(stream)
775
514
self.descriptions = descriptions
776
515
self.verbosity = verbosity
777
516
self._bench_history = bench_history
778
self._strict = strict
779
self._result_decorators = result_decorators or []
517
self.list_only = list_only
781
519
def run(self, test):
782
520
"Run the given test case or test suite."
521
startTime = time.time()
783
522
if self.verbosity == 1:
784
523
result_class = TextTestResult
785
524
elif self.verbosity >= 2:
786
525
result_class = VerboseTestResult
787
original_result = result_class(self.stream,
526
result = result_class(self.stream,
788
527
self.descriptions,
790
529
bench_history=self._bench_history,
530
num_tests=test.countTestCases(),
793
# Signal to result objects that look at stop early policy to stop,
794
original_result.stop_early = self.stop_on_failure
795
result = original_result
796
for decorator in self._result_decorators:
797
result = decorator(result)
798
result.stop_early = self.stop_on_failure
799
result.startTestRun()
532
result.stop_early = self.stop_on_failure
533
result.report_starting()
535
if self.verbosity >= 2:
536
self.stream.writeln("Listing tests only ...\n")
538
for t in iter_suite_tests(test):
539
self.stream.writeln("%s" % (t.id()))
541
actionTaken = "Listed"
804
# higher level code uses our extended protocol to determine
805
# what exit code to give.
806
return original_result
544
run = result.testsRun
546
stopTime = time.time()
547
timeTaken = stopTime - startTime
549
self.stream.writeln(result.separator2)
550
self.stream.writeln("%s %d test%s in %.3fs" % (actionTaken,
551
run, run != 1 and "s" or "", timeTaken))
552
self.stream.writeln()
553
if not result.wasSuccessful():
554
self.stream.write("FAILED (")
555
failed, errored = map(len, (result.failures, result.errors))
557
self.stream.write("failures=%d" % failed)
559
if failed: self.stream.write(", ")
560
self.stream.write("errors=%d" % errored)
561
if result.known_failure_count:
562
if failed or errored: self.stream.write(", ")
563
self.stream.write("known_failure_count=%d" %
564
result.known_failure_count)
565
self.stream.writeln(")")
567
if result.known_failure_count:
568
self.stream.writeln("OK (known_failures=%d)" %
569
result.known_failure_count)
571
self.stream.writeln("OK")
572
if result.skip_count > 0:
573
skipped = result.skip_count
574
self.stream.writeln('%d test%s skipped' %
575
(skipped, skipped != 1 and "s" or ""))
576
if result.unsupported:
577
for feature, count in sorted(result.unsupported.items()):
578
self.stream.writeln("Missing feature '%s' skipped %d tests." %
809
584
def iter_suite_tests(suite):
810
585
"""Return all tests in a suite, recursing through nested suites"""
811
if isinstance(suite, unittest.TestCase):
813
elif isinstance(suite, unittest.TestSuite):
586
for item in suite._tests:
587
if isinstance(item, unittest.TestCase):
589
elif isinstance(item, unittest.TestSuite):
815
590
for r in iter_suite_tests(item):
818
raise Exception('unknown type %r for object %r'
819
% (type(suite), suite))
822
TestSkipped = testtools.testcase.TestSkipped
593
raise Exception('unknown object %r inside test suite %r'
597
class TestSkipped(Exception):
598
"""Indicates that a test was intentionally skipped, rather than failing."""
825
601
class TestNotApplicable(TestSkipped):
826
602
"""A test is not applicable to the situation where it was run.
828
This is only normally raised by parameterized tests, if they find that
829
the instance they're constructed upon does not support one aspect
604
This is only normally raised by parameterized tests, if they find that
605
the instance they're constructed upon does not support one aspect
830
606
of its interface.
834
# traceback._some_str fails to format exceptions that have the default
835
# __str__ which does an implicit ascii conversion. However, repr() on those
836
# objects works, for all that its not quite what the doctor may have ordered.
837
def _clever_some_str(value):
842
return repr(value).replace('\\n', '\n')
844
return '<unprintable %s object>' % type(value).__name__
846
traceback._some_str = _clever_some_str
849
# deprecated - use self.knownFailure(), or self.expectFailure.
850
KnownFailure = testtools.testcase._ExpectedFailure
610
class KnownFailure(AssertionError):
611
"""Indicates that a test failed in a precisely expected manner.
613
Such failures dont block the whole test suite from passing because they are
614
indicators of partially completed code or of future work. We have an
615
explicit error for them so that we can ensure that they are always visible:
616
KnownFailures are always shown in the output of bzr selftest.
853
620
class UnavailableFeature(Exception):
854
621
"""A feature required for this test was not available.
856
This can be considered a specialised form of SkippedTest.
858
623
The feature should be used to construct the exception.
627
class CommandFailed(Exception):
862
631
class StringIOWrapper(object):
863
632
"""A wrapper around cStringIO which just adds an encoding attribute.
865
634
Internally we can check sys.stdout to see what the output encoding
866
635
should be. However, cStringIO has no encoding attribute that we can
867
636
set. So we wrap it instead.
955
736
retrieved by _get_log(). We use a real OS file, not an in-memory object,
956
737
so that it can also capture file IO. When the test completes this file
957
738
is read into memory and removed from disk.
959
740
There are also convenience functions to invoke bzr's command-line
960
741
routine, and to build and check bzr trees.
962
743
In addition to the usual method of overriding tearDown(), this class also
963
allows subclasses to register cleanup functions via addCleanup, which are
744
allows subclasses to register functions into the _cleanups list, which is
964
745
run in order as the object is torn down. It's less likely this will be
965
746
accidentally overlooked.
749
_active_threads = None
750
_leaking_threads_tests = 0
751
_first_thread_leaker_id = None
752
_log_file_name = None
754
_keep_log_file = False
969
755
# record lsprof data when performing benchmark calls.
970
756
_gather_lsprof_in_benchmarks = False
757
attrs_to_keep = ('id', '_testMethodName', '_testMethodDoc',
758
'_log_contents', '_log_file_name', '_benchtime',
759
'_TestCase__testMethodName')
972
761
def __init__(self, methodName='testMethod'):
973
762
super(TestCase, self).__init__(methodName)
974
self._directory_isolation = True
975
self.exception_handlers.insert(0,
976
(UnavailableFeature, self._do_unsupported_or_skip))
977
self.exception_handlers.insert(0,
978
(TestNotApplicable, self._do_not_applicable))
981
super(TestCase, self).setUp()
982
for feature in getattr(self, '_test_needs_features', []):
983
self.requireFeature(feature)
766
unittest.TestCase.setUp(self)
984
767
self._cleanEnvironment()
985
768
self._silenceUI()
986
769
self._startLogFile()
987
770
self._benchcalls = []
988
771
self._benchtime = None
989
772
self._clear_hooks()
990
self._track_transports()
992
773
self._clear_debug_flags()
993
# Isolate global verbosity level, to make sure it's reproducible
994
# between tests. We should get rid of this altogether: bug 656694. --
996
self.overrideAttr(bzrlib.trace, '_verbosity_level', 0)
997
# Isolate config option expansion until its default value for bzrlib is
998
# settled on or a the FIXME associated with _get_expand_default_value
999
# is addressed -- vila 20110219
1000
self.overrideAttr(config, '_expand_default_value', None)
1001
self._log_files = set()
1002
# Each key in the ``_counters`` dict holds a value for a different
1003
# counter. When the test ends, addDetail() should be used to output the
1004
# counter values. This happens in install_counter_hook().
1006
if 'config_stats' in selftest_debug_flags:
1007
self._install_config_stats_hooks()
1012
pdb.Pdb().set_trace(sys._getframe().f_back)
1014
def discardDetail(self, name):
1015
"""Extend the addDetail, getDetails api so we can remove a detail.
1017
eg. bzr always adds the 'log' detail at startup, but we don't want to
1018
include it for skipped, xfail, etc tests.
1020
It is safe to call this for a detail that doesn't exist, in case this
1021
gets called multiple times.
1023
# We cheat. details is stored in __details which means we shouldn't
1024
# touch it. but getDetails() returns the dict directly, so we can
1026
details = self.getDetails()
1030
def install_counter_hook(self, hooks, name, counter_name=None):
1031
"""Install a counting hook.
1033
Any hook can be counted as long as it doesn't need to return a value.
1035
:param hooks: Where the hook should be installed.
1037
:param name: The hook name that will be counted.
1039
:param counter_name: The counter identifier in ``_counters``, defaults
1042
_counters = self._counters # Avoid closing over self
1043
if counter_name is None:
1045
if _counters.has_key(counter_name):
1046
raise AssertionError('%s is already used as a counter name'
1048
_counters[counter_name] = 0
1049
self.addDetail(counter_name, content.Content(content.UTF8_TEXT,
1050
lambda: ['%d' % (_counters[counter_name],)]))
1051
def increment_counter(*args, **kwargs):
1052
_counters[counter_name] += 1
1053
label = 'count %s calls' % (counter_name,)
1054
hooks.install_named_hook(name, increment_counter, label)
1055
self.addCleanup(hooks.uninstall_named_hook, name, label)
1057
def _install_config_stats_hooks(self):
1058
"""Install config hooks to count hook calls.
1061
for hook_name in ('get', 'set', 'remove', 'load', 'save'):
1062
self.install_counter_hook(config.ConfigHooks, hook_name,
1063
'config.%s' % (hook_name,))
1065
# The OldConfigHooks are private and need special handling to protect
1066
# against recursive tests (tests that run other tests), so we just do
1067
# manually what registering them into _builtin_known_hooks will provide
1069
self.overrideAttr(config, 'OldConfigHooks', config._OldConfigHooks())
1070
for hook_name in ('get', 'set', 'remove', 'load', 'save'):
1071
self.install_counter_hook(config.OldConfigHooks, hook_name,
1072
'old_config.%s' % (hook_name,))
774
TestCase._active_threads = threading.activeCount()
775
self.addCleanup(self._check_leaked_threads)
777
def _check_leaked_threads(self):
778
active = threading.activeCount()
779
leaked_threads = active - TestCase._active_threads
780
TestCase._active_threads = active
782
TestCase._leaking_threads_tests += 1
783
if TestCase._first_thread_leaker_id is None:
784
TestCase._first_thread_leaker_id = self.id()
785
# we're not specifically told when all tests are finished.
786
# This will do. We use a function to avoid keeping a reference
787
# to a TestCase object.
788
atexit.register(_report_leaked_threads)
1074
790
def _clear_debug_flags(self):
1075
791
"""Prevent externally set debug flags affecting tests.
1077
793
Tests that want to use debug flags can just set them in the
1078
794
debug_flags set during setup/teardown.
1080
# Start with a copy of the current debug flags we can safely modify.
1081
self.overrideAttr(debug, 'debug_flags', set(debug.debug_flags))
1082
796
if 'allow_debug' not in selftest_debug_flags:
797
self._preserved_debug_flags = set(debug.debug_flags)
1083
798
debug.debug_flags.clear()
1084
if 'disable_lock_checks' not in selftest_debug_flags:
1085
debug.debug_flags.add('strict_locks')
799
self.addCleanup(self._restore_debug_flags)
1087
801
def _clear_hooks(self):
1088
802
# prevent hooks affecting tests
1089
known_hooks = hooks.known_hooks
1090
self._preserved_hooks = {}
1091
for key, (parent, name) in known_hooks.iter_parent_objects():
1092
current_hooks = getattr(parent, name)
1093
self._preserved_hooks[parent] = (name, current_hooks)
1094
self._preserved_lazy_hooks = hooks._lazy_hooks
1095
hooks._lazy_hooks = {}
804
import bzrlib.smart.server
805
self._preserved_hooks = {
806
bzrlib.branch.Branch: bzrlib.branch.Branch.hooks,
807
bzrlib.mutabletree.MutableTree: bzrlib.mutabletree.MutableTree.hooks,
808
bzrlib.smart.server.SmartTCPServer: bzrlib.smart.server.SmartTCPServer.hooks,
1096
810
self.addCleanup(self._restoreHooks)
1097
for key, (parent, name) in known_hooks.iter_parent_objects():
1098
factory = known_hooks.get(key)
1099
setattr(parent, name, factory())
1100
# this hook should always be installed
1101
request._install_hook()
1103
def disable_directory_isolation(self):
1104
"""Turn off directory isolation checks."""
1105
self._directory_isolation = False
1107
def enable_directory_isolation(self):
1108
"""Enable directory isolation checks."""
1109
self._directory_isolation = True
811
# reset all hooks to an empty instance of the appropriate type
812
bzrlib.branch.Branch.hooks = bzrlib.branch.BranchHooks()
813
bzrlib.smart.server.SmartTCPServer.hooks = bzrlib.smart.server.SmartServerHooks()
1111
815
def _silenceUI(self):
1112
816
"""Turn off UI for duration of test"""
1113
817
# by default the UI is off; tests can turn it on if they want it.
1114
self.overrideAttr(ui, 'ui_factory', ui.SilentUIFactory())
1116
def _check_locks(self):
1117
"""Check that all lock take/release actions have been paired."""
1118
# We always check for mismatched locks. If a mismatch is found, we
1119
# fail unless -Edisable_lock_checks is supplied to selftest, in which
1120
# case we just print a warning.
1122
acquired_locks = [lock for action, lock in self._lock_actions
1123
if action == 'acquired']
1124
released_locks = [lock for action, lock in self._lock_actions
1125
if action == 'released']
1126
broken_locks = [lock for action, lock in self._lock_actions
1127
if action == 'broken']
1128
# trivially, given the tests for lock acquistion and release, if we
1129
# have as many in each list, it should be ok. Some lock tests also
1130
# break some locks on purpose and should be taken into account by
1131
# considering that breaking a lock is just a dirty way of releasing it.
1132
if len(acquired_locks) != (len(released_locks) + len(broken_locks)):
1134
'Different number of acquired and '
1135
'released or broken locks.\n'
1139
(acquired_locks, released_locks, broken_locks))
1140
if not self._lock_check_thorough:
1141
# Rather than fail, just warn
1142
print "Broken test %s: %s" % (self, message)
1146
def _track_locks(self):
1147
"""Track lock activity during tests."""
1148
self._lock_actions = []
1149
if 'disable_lock_checks' in selftest_debug_flags:
1150
self._lock_check_thorough = False
1152
self._lock_check_thorough = True
1154
self.addCleanup(self._check_locks)
1155
_mod_lock.Lock.hooks.install_named_hook('lock_acquired',
1156
self._lock_acquired, None)
1157
_mod_lock.Lock.hooks.install_named_hook('lock_released',
1158
self._lock_released, None)
1159
_mod_lock.Lock.hooks.install_named_hook('lock_broken',
1160
self._lock_broken, None)
1162
def _lock_acquired(self, result):
1163
self._lock_actions.append(('acquired', result))
1165
def _lock_released(self, result):
1166
self._lock_actions.append(('released', result))
1168
def _lock_broken(self, result):
1169
self._lock_actions.append(('broken', result))
1171
def permit_dir(self, name):
1172
"""Permit a directory to be used by this test. See permit_url."""
1173
name_transport = _mod_transport.get_transport_from_path(name)
1174
self.permit_url(name)
1175
self.permit_url(name_transport.base)
1177
def permit_url(self, url):
1178
"""Declare that url is an ok url to use in this test.
1180
Do this for memory transports, temporary test directory etc.
1182
Do not do this for the current working directory, /tmp, or any other
1183
preexisting non isolated url.
1185
if not url.endswith('/'):
1187
self._bzr_selftest_roots.append(url)
1189
def permit_source_tree_branch_repo(self):
1190
"""Permit the source tree bzr is running from to be opened.
1192
Some code such as bzrlib.version attempts to read from the bzr branch
1193
that bzr is executing from (if any). This method permits that directory
1194
to be used in the test suite.
1196
path = self.get_source_path()
1197
self.record_directory_isolation()
1200
workingtree.WorkingTree.open(path)
1201
except (errors.NotBranchError, errors.NoWorkingTree):
1202
raise TestSkipped('Needs a working tree of bzr sources')
1204
self.enable_directory_isolation()
1206
def _preopen_isolate_transport(self, transport):
1207
"""Check that all transport openings are done in the test work area."""
1208
while isinstance(transport, pathfilter.PathFilteringTransport):
1209
# Unwrap pathfiltered transports
1210
transport = transport.server.backing_transport.clone(
1211
transport._filter('.'))
1212
url = transport.base
1213
# ReadonlySmartTCPServer_for_testing decorates the backing transport
1214
# urls it is given by prepending readonly+. This is appropriate as the
1215
# client shouldn't know that the server is readonly (or not readonly).
1216
# We could register all servers twice, with readonly+ prepending, but
1217
# that makes for a long list; this is about the same but easier to
1219
if url.startswith('readonly+'):
1220
url = url[len('readonly+'):]
1221
self._preopen_isolate_url(url)
1223
def _preopen_isolate_url(self, url):
1224
if not self._directory_isolation:
1226
if self._directory_isolation == 'record':
1227
self._bzr_selftest_roots.append(url)
1229
# This prevents all transports, including e.g. sftp ones backed on disk
1230
# from working unless they are explicitly granted permission. We then
1231
# depend on the code that sets up test transports to check that they are
1232
# appropriately isolated and enable their use by calling
1233
# self.permit_transport()
1234
if not osutils.is_inside_any(self._bzr_selftest_roots, url):
1235
raise errors.BzrError("Attempt to escape test isolation: %r %r"
1236
% (url, self._bzr_selftest_roots))
1238
def record_directory_isolation(self):
1239
"""Gather accessed directories to permit later access.
1241
This is used for tests that access the branch bzr is running from.
1243
self._directory_isolation = "record"
1245
def start_server(self, transport_server, backing_server=None):
1246
"""Start transport_server for this test.
1248
This starts the server, registers a cleanup for it and permits the
1249
server's urls to be used.
1251
if backing_server is None:
1252
transport_server.start_server()
1254
transport_server.start_server(backing_server)
1255
self.addCleanup(transport_server.stop_server)
1256
# Obtain a real transport because if the server supplies a password, it
1257
# will be hidden from the base on the client side.
1258
t = _mod_transport.get_transport_from_url(transport_server.get_url())
1259
# Some transport servers effectively chroot the backing transport;
1260
# others like SFTPServer don't - users of the transport can walk up the
1261
# transport to read the entire backing transport. This wouldn't matter
1262
# except that the workdir tests are given - and that they expect the
1263
# server's url to point at - is one directory under the safety net. So
1264
# Branch operations into the transport will attempt to walk up one
1265
# directory. Chrooting all servers would avoid this but also mean that
1266
# we wouldn't be testing directly against non-root urls. Alternatively
1267
# getting the test framework to start the server with a backing server
1268
# at the actual safety net directory would work too, but this then
1269
# means that the self.get_url/self.get_transport methods would need
1270
# to transform all their results. On balance its cleaner to handle it
1271
# here, and permit a higher url when we have one of these transports.
1272
if t.base.endswith('/work/'):
1273
# we have safety net/test root/work
1274
t = t.clone('../..')
1275
elif isinstance(transport_server,
1276
test_server.SmartTCPServer_for_testing):
1277
# The smart server adds a path similar to work, which is traversed
1278
# up from by the client. But the server is chrooted - the actual
1279
# backing transport is not escaped from, and VFS requests to the
1280
# root will error (because they try to escape the chroot).
1282
while t2.base != t.base:
1285
self.permit_url(t.base)
1287
def _track_transports(self):
1288
"""Install checks for transport usage."""
1289
# TestCase has no safe place it can write to.
1290
self._bzr_selftest_roots = []
1291
# Currently the easiest way to be sure that nothing is going on is to
1292
# hook into bzr dir opening. This leaves a small window of error for
1293
# transport tests, but they are well known, and we can improve on this
1295
bzrdir.BzrDir.hooks.install_named_hook("pre_open",
1296
self._preopen_isolate_transport, "Check bzr directories are safe.")
818
saved = ui.ui_factory
820
ui.ui_factory = saved
821
ui.ui_factory = ui.SilentUIFactory()
822
self.addCleanup(_restore)
1298
824
def _ndiff_strings(self, a, b):
1299
825
"""Return ndiff between two strings containing lines.
1301
827
A trailing newline is added if missing to make the strings
1302
828
print properly."""
1303
829
if b and b[-1] != '\n':
1338
864
if message is None:
1339
865
message = "texts not equal:\n"
867
message = 'first string is missing a final newline.\n'
1340
868
if a + '\n' == b:
1341
message = 'first string is missing a final newline.\n'
1343
869
message = 'second string is missing a final newline.\n'
1344
870
raise AssertionError(message +
1345
871
self._ndiff_strings(a, b))
1347
873
def assertEqualMode(self, mode, mode_test):
1348
874
self.assertEqual(mode, mode_test,
1349
875
'mode mismatch %o != %o' % (mode, mode_test))
1351
def assertEqualStat(self, expected, actual):
1352
"""assert that expected and actual are the same stat result.
1354
:param expected: A stat result.
1355
:param actual: A stat result.
1356
:raises AssertionError: If the expected and actual stat values differ
1357
other than by atime.
1359
self.assertEqual(expected.st_size, actual.st_size,
1360
'st_size did not match')
1361
self.assertEqual(expected.st_mtime, actual.st_mtime,
1362
'st_mtime did not match')
1363
self.assertEqual(expected.st_ctime, actual.st_ctime,
1364
'st_ctime did not match')
1365
if sys.platform == 'win32':
1366
# On Win32 both 'dev' and 'ino' cannot be trusted. In python2.4 it
1367
# is 'dev' that varies, in python 2.5 (6?) it is st_ino that is
1368
# odd. We just force it to always be 0 to avoid any problems.
1369
self.assertEqual(0, expected.st_dev)
1370
self.assertEqual(0, actual.st_dev)
1371
self.assertEqual(0, expected.st_ino)
1372
self.assertEqual(0, actual.st_ino)
1374
self.assertEqual(expected.st_dev, actual.st_dev,
1375
'st_dev did not match')
1376
self.assertEqual(expected.st_ino, actual.st_ino,
1377
'st_ino did not match')
1378
self.assertEqual(expected.st_mode, actual.st_mode,
1379
'st_mode did not match')
1381
def assertLength(self, length, obj_with_len):
1382
"""Assert that obj_with_len is of length length."""
1383
if len(obj_with_len) != length:
1384
self.fail("Incorrect length: wanted %d, got %d for %r" % (
1385
length, len(obj_with_len), obj_with_len))
1387
def assertLogsError(self, exception_class, func, *args, **kwargs):
1388
"""Assert that `func(*args, **kwargs)` quietly logs a specific error.
1391
orig_log_exception_quietly = trace.log_exception_quietly
1394
orig_log_exception_quietly()
1395
captured.append(sys.exc_info()[1])
1396
trace.log_exception_quietly = capture
1397
func(*args, **kwargs)
1399
trace.log_exception_quietly = orig_log_exception_quietly
1400
self.assertLength(1, captured)
1402
self.assertIsInstance(err, exception_class)
1405
877
def assertPositive(self, val):
1406
878
"""Assert that val is greater than 0."""
1407
879
self.assertTrue(val > 0, 'expected a positive value, but got %s' % val)
1701
1174
The file is removed as the test is torn down.
1703
pseudo_log_file = StringIO()
1704
def _get_log_contents_for_weird_testtools_api():
1705
return [pseudo_log_file.getvalue().decode(
1706
"utf-8", "replace").encode("utf-8")]
1707
self.addDetail("log", content.Content(content.ContentType("text",
1708
"plain", {"charset": "utf8"}),
1709
_get_log_contents_for_weird_testtools_api))
1710
self._log_file = pseudo_log_file
1711
self._log_memento = trace.push_log_file(self._log_file)
1176
fileno, name = tempfile.mkstemp(suffix='.log', prefix='testbzr')
1177
self._log_file = os.fdopen(fileno, 'w+')
1178
self._log_memento = bzrlib.trace.push_log_file(self._log_file)
1179
self._log_file_name = name
1712
1180
self.addCleanup(self._finishLogFile)
1714
1182
def _finishLogFile(self):
1715
1183
"""Finished with the log file.
1717
Close the file and delete it.
1719
if trace._trace_file:
1720
# flush the log file, to get all content
1721
trace._trace_file.flush()
1722
trace.pop_log_file(self._log_memento)
1724
def thisFailsStrictLockCheck(self):
1725
"""It is known that this test would fail with -Dstrict_locks.
1727
By default, all tests are run with strict lock checking unless
1728
-Edisable_lock_checks is supplied. However there are some tests which
1729
we know fail strict locks at this point that have not been fixed.
1730
They should call this function to disable the strict checking.
1732
This should be used sparingly, it is much better to fix the locking
1733
issues rather than papering over the problem by calling this function.
1735
debug.debug_flags.discard('strict_locks')
1737
def overrideAttr(self, obj, attr_name, new=_unitialized_attr):
1738
"""Overrides an object attribute restoring it after the test.
1740
:note: This should be used with discretion; you should think about
1741
whether it's better to make the code testable without monkey-patching.
1743
:param obj: The object that will be mutated.
1745
:param attr_name: The attribute name we want to preserve/override in
1748
:param new: The optional value we want to set the attribute to.
1750
:returns: The actual attr value.
1752
value = getattr(obj, attr_name)
1753
# The actual value is captured by the call below
1754
self.addCleanup(setattr, obj, attr_name, value)
1755
if new is not _unitialized_attr:
1756
setattr(obj, attr_name, new)
1759
def overrideEnv(self, name, new):
1760
"""Set an environment variable, and reset it after the test.
1762
:param name: The environment variable name.
1764
:param new: The value to set the variable to. If None, the
1765
variable is deleted from the environment.
1767
:returns: The actual variable value.
1769
value = osutils.set_or_unset_env(name, new)
1770
self.addCleanup(osutils.set_or_unset_env, name, value)
1773
def recordCalls(self, obj, attr_name):
1774
"""Monkeypatch in a wrapper that will record calls.
1776
The monkeypatch is automatically removed when the test concludes.
1778
:param obj: The namespace holding the reference to be replaced;
1779
typically a module, class, or object.
1780
:param attr_name: A string for the name of the attribute to
1782
:returns: A list that will be extended with one item every time the
1783
function is called, with a tuple of (args, kwargs).
1787
def decorator(*args, **kwargs):
1788
calls.append((args, kwargs))
1789
return orig(*args, **kwargs)
1790
orig = self.overrideAttr(obj, attr_name, decorator)
1185
Close the file and delete it, unless setKeepLogfile was called.
1187
if self._log_file is None:
1189
bzrlib.trace.pop_log_file(self._log_memento)
1190
self._log_file.close()
1191
self._log_file = None
1192
if not self._keep_log_file:
1193
os.remove(self._log_file_name)
1194
self._log_file_name = None
1196
def setKeepLogfile(self):
1197
"""Make the logfile not be deleted when _finishLogFile is called."""
1198
self._keep_log_file = True
1200
def addCleanup(self, callable, *args, **kwargs):
1201
"""Arrange to run a callable when this case is torn down.
1203
Callables are run in the reverse of the order they are registered,
1204
ie last-in first-out.
1206
self._cleanups.append((callable, args, kwargs))
1793
1208
def _cleanEnvironment(self):
1794
for name, value in isolated_environ.iteritems():
1795
self.overrideEnv(name, value)
1210
'BZR_HOME': None, # Don't inherit BZR_HOME to all the tests.
1211
'HOME': os.getcwd(),
1212
'APPDATA': None, # bzr now use Win32 API and don't rely on APPDATA
1213
'BZR_EDITOR': None, # test_msgeditor manipulates this variable
1215
'BZREMAIL': None, # may still be present in the environment
1217
'BZR_PROGRESS_BAR': None,
1220
'SSH_AUTH_SOCK': None,
1224
'https_proxy': None,
1225
'HTTPS_PROXY': None,
1230
# Nobody cares about these ones AFAIK. So far at
1231
# least. If you do (care), please update this comment
1235
'BZR_REMOTE_PATH': None,
1238
self.addCleanup(self._restoreEnvironment)
1239
for name, value in new_env.iteritems():
1240
self._captureVar(name, value)
1242
def _captureVar(self, name, newvalue):
1243
"""Set an environment variable, and reset it when finished."""
1244
self.__old_env[name] = osutils.set_or_unset_env(name, newvalue)
1246
def _restore_debug_flags(self):
1247
debug.debug_flags.clear()
1248
debug.debug_flags.update(self._preserved_debug_flags)
1250
def _restoreEnvironment(self):
1251
for name, value in self.__old_env.iteritems():
1252
osutils.set_or_unset_env(name, value)
1797
1254
def _restoreHooks(self):
1798
for klass, (name, hooks) in self._preserved_hooks.items():
1799
setattr(klass, name, hooks)
1800
self._preserved_hooks.clear()
1801
bzrlib.hooks._lazy_hooks = self._preserved_lazy_hooks
1802
self._preserved_lazy_hooks.clear()
1255
for klass, hooks in self._preserved_hooks.items():
1256
setattr(klass, 'hooks', hooks)
1804
1258
def knownFailure(self, reason):
1805
"""Declare that this test fails for a known reason
1807
Tests that are known to fail should generally be using expectedFailure
1808
with an appropriate reverse assertion if a change could cause the test
1809
to start passing. Conversely if the test has no immediate prospect of
1810
succeeding then using skip is more suitable.
1812
When this method is called while an exception is being handled, that
1813
traceback will be used, otherwise a new exception will be thrown to
1814
provide one but won't be reported.
1816
self._add_reason(reason)
1259
"""This test has failed for some known reason."""
1260
raise KnownFailure(reason)
1262
def run(self, result=None):
1263
if result is None: result = self.defaultTestResult()
1264
for feature in getattr(self, '_test_needs_features', []):
1265
if not feature.available():
1266
result.startTest(self)
1267
if getattr(result, 'addNotSupported', None):
1268
result.addNotSupported(self, feature)
1270
result.addSuccess(self)
1271
result.stopTest(self)
1818
exc_info = sys.exc_info()
1819
if exc_info != (None, None, None):
1820
self._report_traceback(exc_info)
1823
raise self.failureException(reason)
1824
except self.failureException:
1825
exc_info = sys.exc_info()
1826
# GZ 02-08-2011: Maybe cleanup this err.exc_info attribute too?
1827
raise testtools.testcase._ExpectedFailure(exc_info)
1274
return unittest.TestCase.run(self, result)
1831
def _suppress_log(self):
1832
"""Remove the log info from details."""
1833
self.discardDetail('log')
1835
def _do_skip(self, result, reason):
1836
self._suppress_log()
1837
addSkip = getattr(result, 'addSkip', None)
1838
if not callable(addSkip):
1839
result.addSuccess(result)
1841
addSkip(self, reason)
1844
def _do_known_failure(self, result, e):
1845
self._suppress_log()
1846
err = sys.exc_info()
1847
addExpectedFailure = getattr(result, 'addExpectedFailure', None)
1848
if addExpectedFailure is not None:
1849
addExpectedFailure(self, err)
1851
result.addSuccess(self)
1854
def _do_not_applicable(self, result, e):
1856
reason = 'No reason given'
1859
self._suppress_log ()
1860
addNotApplicable = getattr(result, 'addNotApplicable', None)
1861
if addNotApplicable is not None:
1862
result.addNotApplicable(self, reason)
1864
self._do_skip(result, reason)
1867
def _report_skip(self, result, err):
1868
"""Override the default _report_skip.
1870
We want to strip the 'log' detail. If we waint until _do_skip, it has
1871
already been formatted into the 'reason' string, and we can't pull it
1874
self._suppress_log()
1875
super(TestCase, self)._report_skip(self, result, err)
1878
def _report_expected_failure(self, result, err):
1881
See _report_skip for motivation.
1883
self._suppress_log()
1884
super(TestCase, self)._report_expected_failure(self, result, err)
1887
def _do_unsupported_or_skip(self, result, e):
1889
self._suppress_log()
1890
addNotSupported = getattr(result, 'addNotSupported', None)
1891
if addNotSupported is not None:
1892
result.addNotSupported(self, reason)
1894
self._do_skip(result, reason)
1277
absent_attr = object()
1278
for attr_name in self.attrs_to_keep:
1279
attr = getattr(self, attr_name, absent_attr)
1280
if attr is not absent_attr:
1281
saved_attrs[attr_name] = attr
1282
self.__dict__ = saved_attrs
1286
unittest.TestCase.tearDown(self)
1896
1288
def time(self, callable, *args, **kwargs):
1897
1289
"""Run callable and accrue the time it takes to the benchmark time.
1899
1291
If lsprofiling is enabled (i.e. by --lsprof-time to bzr selftest) then
1900
1292
this will cause lsprofile statistics to be gathered and stored in
1901
1293
self._benchcalls.
1903
1295
if self._benchtime is None:
1904
self.addDetail('benchtime', content.Content(content.ContentType(
1905
"text", "plain"), lambda:[str(self._benchtime)]))
1906
1296
self._benchtime = 0
1907
1297
start = time.time()
2636
1982
def make_repository(self, relpath, shared=False, format=None):
2637
1983
"""Create a repository on our default transport at relpath.
2639
1985
Note that relpath must be a relative path, not a full url.
2641
1987
# FIXME: If you create a remoterepository this returns the underlying
2642
# real format, which is incorrect. Actually we should make sure that
1988
# real format, which is incorrect. Actually we should make sure that
2643
1989
# RemoteBzrDir returns a RemoteRepository.
2644
1990
# maybe mbp 20070410
2645
1991
made_control = self.make_bzrdir(relpath, format=format)
2646
1992
return made_control.create_repository(shared=shared)
2648
def make_smart_server(self, path, backing_server=None):
2649
if backing_server is None:
2650
backing_server = self.get_server()
2651
smart_server = test_server.SmartTCPServer_for_testing()
2652
self.start_server(smart_server, backing_server)
2653
remote_transport = _mod_transport.get_transport_from_url(smart_server.get_url()
2655
return remote_transport
2657
1994
def make_branch_and_memory_tree(self, relpath, format=None):
2658
1995
"""Create a branch on the default transport and a MemoryTree for it."""
2659
1996
b = self.make_branch(relpath, format=format)
2660
1997
return memorytree.MemoryTree.create_on_branch(b)
2662
1999
def make_branch_builder(self, relpath, format=None):
2663
branch = self.make_branch(relpath, format=format)
2664
return branchbuilder.BranchBuilder(branch=branch)
2000
url = self.get_url(relpath)
2001
tran = get_transport(url)
2002
return branchbuilder.BranchBuilder(get_transport(url), format=format)
2666
2004
def overrideEnvironmentForTesting(self):
2667
test_home_dir = self.test_home_dir
2668
if isinstance(test_home_dir, unicode):
2669
test_home_dir = test_home_dir.encode(sys.getfilesystemencoding())
2670
self.overrideEnv('HOME', test_home_dir)
2671
self.overrideEnv('BZR_HOME', test_home_dir)
2005
os.environ['HOME'] = self.test_home_dir
2006
os.environ['BZR_HOME'] = self.test_home_dir
2673
2008
def setUp(self):
2674
2009
super(TestCaseWithMemoryTransport, self).setUp()
2675
# Ensure that ConnectedTransport doesn't leak sockets
2676
def get_transport_from_url_with_cleanup(*args, **kwargs):
2677
t = orig_get_transport_from_url(*args, **kwargs)
2678
if isinstance(t, _mod_transport.ConnectedTransport):
2679
self.addCleanup(t.disconnect)
2682
orig_get_transport_from_url = self.overrideAttr(
2683
_mod_transport, 'get_transport_from_url',
2684
get_transport_from_url_with_cleanup)
2685
2010
self._make_test_root()
2686
self.addCleanup(os.chdir, os.getcwdu())
2011
_currentdir = os.getcwdu()
2012
def _leaveDirectory():
2013
os.chdir(_currentdir)
2014
self.addCleanup(_leaveDirectory)
2687
2015
self.makeAndChdirToTestDir()
2688
2016
self.overrideEnvironmentForTesting()
2689
2017
self.__readonly_server = None
2690
2018
self.__server = None
2691
2019
self.reduceLockdirTimeout()
2693
def setup_smart_server_with_call_log(self):
2694
"""Sets up a smart server as the transport server with a call log."""
2695
self.transport_server = test_server.SmartTCPServer_for_testing
2696
self.hpss_calls = []
2698
# Skip the current stack down to the caller of
2699
# setup_smart_server_with_call_log
2700
prefix_length = len(traceback.extract_stack()) - 2
2701
def capture_hpss_call(params):
2702
self.hpss_calls.append(
2703
CapturedCall(params, prefix_length))
2704
client._SmartClient.hooks.install_named_hook(
2705
'call', capture_hpss_call, None)
2707
def reset_smart_call_log(self):
2708
self.hpss_calls = []
2711
2022
class TestCaseInTempDir(TestCaseWithMemoryTransport):
2712
2023
"""Derived class that runs a test within a temporary directory.
3187
2464
list_only=False,
3188
2465
random_seed=None,
3189
2466
exclude_pattern=None,
3192
suite_decorators=None,
3194
result_decorators=None,
3196
"""Run a test suite for bzr selftest.
3198
:param runner_class: The class of runner to use. Must support the
3199
constructor arguments passed by run_suite which are more than standard
3201
:return: A boolean indicating success.
3203
2468
TestCase._gather_lsprof_in_benchmarks = lsprof_timed
3208
if runner_class is None:
3209
runner_class = TextTestRunner
3212
runner = runner_class(stream=stream,
2473
runner = TextTestRunner(stream=sys.stdout,
3213
2474
descriptions=0,
3214
2475
verbosity=verbosity,
3215
2476
bench_history=bench_history,
3217
result_decorators=result_decorators,
2477
list_only=list_only,
3219
2479
runner.stop_on_failure=stop_on_failure
3220
# built in decorator factories:
3222
random_order(random_seed, runner),
3223
exclude_tests(exclude_pattern),
3225
if matching_tests_first:
3226
decorators.append(tests_first(pattern))
2480
# Initialise the random number generator and display the seed used.
2481
# We convert the seed to a long to make it reuseable across invocations.
2482
random_order = False
2483
if random_seed is not None:
2485
if random_seed == "now":
2486
random_seed = long(time.time())
2488
# Convert the seed to a long if we can
2490
random_seed = long(random_seed)
2493
runner.stream.writeln("Randomizing test order using seed %s\n" %
2495
random.seed(random_seed)
2496
# Customise the list of tests if requested
2497
if exclude_pattern is not None:
2498
suite = exclude_tests_by_re(suite, exclude_pattern)
2500
order_changer = randomize_suite
3228
decorators.append(filter_tests(pattern))
3229
if suite_decorators:
3230
decorators.extend(suite_decorators)
3231
# tell the result object how many tests will be running: (except if
3232
# --parallel=fork is being used. Robert said he will provide a better
3233
# progress design later -- vila 20090817)
3234
if fork_decorator not in decorators:
3235
decorators.append(CountingDecorator)
3236
for decorator in decorators:
3237
suite = decorator(suite)
3239
# Done after test suite decoration to allow randomisation etc
3240
# to take effect, though that is of marginal benefit.
3242
stream.write("Listing tests only ...\n")
3243
for t in iter_suite_tests(suite):
3244
stream.write("%s\n" % (t.id()))
2502
order_changer = preserve_input
2503
if pattern != '.*' or random_order:
2504
if matching_tests_first:
2505
suites = map(order_changer, split_suite_by_re(suite, pattern))
2506
suite = TestUtil.TestSuite(suites)
2508
suite = order_changer(filter_suite_by_re(suite, pattern))
3246
2510
result = runner.run(suite)
3248
2513
return result.wasStrictlySuccessful()
3250
return result.wasSuccessful()
3253
# A registry where get() returns a suite decorator.
3254
parallel_registry = registry.Registry()
3257
def fork_decorator(suite):
3258
if getattr(os, "fork", None) is None:
3259
raise errors.BzrCommandError("platform does not support fork,"
3260
" try --parallel=subprocess instead.")
3261
concurrency = osutils.local_concurrency()
3262
if concurrency == 1:
3264
from testtools import ConcurrentTestSuite
3265
return ConcurrentTestSuite(suite, fork_for_tests)
3266
parallel_registry.register('fork', fork_decorator)
3269
def subprocess_decorator(suite):
3270
concurrency = osutils.local_concurrency()
3271
if concurrency == 1:
3273
from testtools import ConcurrentTestSuite
3274
return ConcurrentTestSuite(suite, reinvoke_for_tests)
3275
parallel_registry.register('subprocess', subprocess_decorator)
3278
def exclude_tests(exclude_pattern):
3279
"""Return a test suite decorator that excludes tests."""
3280
if exclude_pattern is None:
3281
return identity_decorator
3282
def decorator(suite):
3283
return ExcludeDecorator(suite, exclude_pattern)
3287
def filter_tests(pattern):
3289
return identity_decorator
3290
def decorator(suite):
3291
return FilterTestsDecorator(suite, pattern)
3295
def random_order(random_seed, runner):
3296
"""Return a test suite decorator factory for randomising tests order.
3298
:param random_seed: now, a string which casts to a long, or a long.
3299
:param runner: A test runner with a stream attribute to report on.
3301
if random_seed is None:
3302
return identity_decorator
3303
def decorator(suite):
3304
return RandomDecorator(suite, random_seed, runner.stream)
3308
def tests_first(pattern):
3310
return identity_decorator
3311
def decorator(suite):
3312
return TestFirstDecorator(suite, pattern)
3316
def identity_decorator(suite):
3321
class TestDecorator(TestUtil.TestSuite):
3322
"""A decorator for TestCase/TestSuite objects.
3324
Usually, subclasses should override __iter__(used when flattening test
3325
suites), which we do to filter, reorder, parallelise and so on, run() and
3329
def __init__(self, suite):
3330
TestUtil.TestSuite.__init__(self)
3333
def countTestCases(self):
3336
cases += test.countTestCases()
3343
def run(self, result):
3344
# Use iteration on self, not self._tests, to allow subclasses to hook
3347
if result.shouldStop:
3353
class CountingDecorator(TestDecorator):
3354
"""A decorator which calls result.progress(self.countTestCases)."""
3356
def run(self, result):
3357
progress_method = getattr(result, 'progress', None)
3358
if callable(progress_method):
3359
progress_method(self.countTestCases(), SUBUNIT_SEEK_SET)
3360
return super(CountingDecorator, self).run(result)
3363
class ExcludeDecorator(TestDecorator):
3364
"""A decorator which excludes test matching an exclude pattern."""
3366
def __init__(self, suite, exclude_pattern):
3367
TestDecorator.__init__(self, suite)
3368
self.exclude_pattern = exclude_pattern
3369
self.excluded = False
3373
return iter(self._tests)
3374
self.excluded = True
3375
suite = exclude_tests_by_re(self, self.exclude_pattern)
3377
self.addTests(suite)
3378
return iter(self._tests)
3381
class FilterTestsDecorator(TestDecorator):
3382
"""A decorator which filters tests to those matching a pattern."""
3384
def __init__(self, suite, pattern):
3385
TestDecorator.__init__(self, suite)
3386
self.pattern = pattern
3387
self.filtered = False
3391
return iter(self._tests)
3392
self.filtered = True
3393
suite = filter_suite_by_re(self, self.pattern)
3395
self.addTests(suite)
3396
return iter(self._tests)
3399
class RandomDecorator(TestDecorator):
3400
"""A decorator which randomises the order of its tests."""
3402
def __init__(self, suite, random_seed, stream):
3403
TestDecorator.__init__(self, suite)
3404
self.random_seed = random_seed
3405
self.randomised = False
3406
self.stream = stream
3410
return iter(self._tests)
3411
self.randomised = True
3412
self.stream.write("Randomizing test order using seed %s\n\n" %
3413
(self.actual_seed()))
3414
# Initialise the random number generator.
3415
random.seed(self.actual_seed())
3416
suite = randomize_suite(self)
3418
self.addTests(suite)
3419
return iter(self._tests)
3421
def actual_seed(self):
3422
if self.random_seed == "now":
3423
# We convert the seed to a long to make it reuseable across
3424
# invocations (because the user can reenter it).
3425
self.random_seed = long(time.time())
3427
# Convert the seed to a long if we can
3429
self.random_seed = long(self.random_seed)
3432
return self.random_seed
3435
class TestFirstDecorator(TestDecorator):
3436
"""A decorator which moves named tests to the front."""
3438
def __init__(self, suite, pattern):
3439
TestDecorator.__init__(self, suite)
3440
self.pattern = pattern
3441
self.filtered = False
3445
return iter(self._tests)
3446
self.filtered = True
3447
suites = split_suite_by_re(self, self.pattern)
3449
self.addTests(suites)
3450
return iter(self._tests)
3453
def partition_tests(suite, count):
3454
"""Partition suite into count lists of tests."""
3455
# This just assigns tests in a round-robin fashion. On one hand this
3456
# splits up blocks of related tests that might run faster if they shared
3457
# resources, but on the other it avoids assigning blocks of slow tests to
3458
# just one partition. So the slowest partition shouldn't be much slower
3460
partitions = [list() for i in range(count)]
3461
tests = iter_suite_tests(suite)
3462
for partition, test in itertools.izip(itertools.cycle(partitions), tests):
3463
partition.append(test)
3467
def workaround_zealous_crypto_random():
3468
"""Crypto.Random want to help us being secure, but we don't care here.
3470
This workaround some test failure related to the sftp server. Once paramiko
3471
stop using the controversial API in Crypto.Random, we may get rid of it.
3474
from Crypto.Random import atfork
3480
def fork_for_tests(suite):
3481
"""Take suite and start up one runner per CPU by forking()
3483
:return: An iterable of TestCase-like objects which can each have
3484
run(result) called on them to feed tests to result.
3486
concurrency = osutils.local_concurrency()
3488
from subunit import TestProtocolClient, ProtocolTestCase
3489
from subunit.test_results import AutoTimingTestResultDecorator
3490
class TestInOtherProcess(ProtocolTestCase):
3491
# Should be in subunit, I think. RBC.
3492
def __init__(self, stream, pid):
3493
ProtocolTestCase.__init__(self, stream)
3496
def run(self, result):
3498
ProtocolTestCase.run(self, result)
3500
os.waitpid(self.pid, 0)
3502
test_blocks = partition_tests(suite, concurrency)
3503
for process_tests in test_blocks:
3504
process_suite = TestUtil.TestSuite()
3505
process_suite.addTests(process_tests)
3506
c2pread, c2pwrite = os.pipe()
3509
workaround_zealous_crypto_random()
3512
# Leave stderr and stdout open so we can see test noise
3513
# Close stdin so that the child goes away if it decides to
3514
# read from stdin (otherwise its a roulette to see what
3515
# child actually gets keystrokes for pdb etc).
3518
stream = os.fdopen(c2pwrite, 'wb', 1)
3519
subunit_result = AutoTimingTestResultDecorator(
3520
TestProtocolClient(stream))
3521
process_suite.run(subunit_result)
3526
stream = os.fdopen(c2pread, 'rb', 1)
3527
test = TestInOtherProcess(stream, pid)
3532
def reinvoke_for_tests(suite):
3533
"""Take suite and start up one runner per CPU using subprocess().
3535
:return: An iterable of TestCase-like objects which can each have
3536
run(result) called on them to feed tests to result.
3538
concurrency = osutils.local_concurrency()
3540
from subunit import ProtocolTestCase
3541
class TestInSubprocess(ProtocolTestCase):
3542
def __init__(self, process, name):
3543
ProtocolTestCase.__init__(self, process.stdout)
3544
self.process = process
3545
self.process.stdin.close()
3548
def run(self, result):
3550
ProtocolTestCase.run(self, result)
3553
os.unlink(self.name)
3554
# print "pid %d finished" % finished_process
3555
test_blocks = partition_tests(suite, concurrency)
3556
for process_tests in test_blocks:
3557
# ugly; currently reimplement rather than reuses TestCase methods.
3558
bzr_path = os.path.dirname(os.path.dirname(bzrlib.__file__))+'/bzr'
3559
if not os.path.isfile(bzr_path):
3560
# We are probably installed. Assume sys.argv is the right file
3561
bzr_path = sys.argv[0]
3562
bzr_path = [bzr_path]
3563
if sys.platform == "win32":
3564
# if we're on windows, we can't execute the bzr script directly
3565
bzr_path = [sys.executable] + bzr_path
3566
fd, test_list_file_name = tempfile.mkstemp()
3567
test_list_file = os.fdopen(fd, 'wb', 1)
3568
for test in process_tests:
3569
test_list_file.write(test.id() + '\n')
3570
test_list_file.close()
3572
argv = bzr_path + ['selftest', '--load-list', test_list_file_name,
3574
if '--no-plugins' in sys.argv:
3575
argv.append('--no-plugins')
3576
# stderr=subprocess.STDOUT would be ideal, but until we prevent
3577
# noise on stderr it can interrupt the subunit protocol.
3578
process = subprocess.Popen(argv, stdin=subprocess.PIPE,
3579
stdout=subprocess.PIPE,
3580
stderr=subprocess.PIPE,
3582
test = TestInSubprocess(process, test_list_file_name)
3585
os.unlink(test_list_file_name)
3590
class ProfileResult(testtools.ExtendedToOriginalDecorator):
3591
"""Generate profiling data for all activity between start and success.
3593
The profile data is appended to the test's _benchcalls attribute and can
3594
be accessed by the forwarded-to TestResult.
3596
While it might be cleaner do accumulate this in stopTest, addSuccess is
3597
where our existing output support for lsprof is, and this class aims to
3598
fit in with that: while it could be moved it's not necessary to accomplish
3599
test profiling, nor would it be dramatically cleaner.
3602
def startTest(self, test):
3603
self.profiler = bzrlib.lsprof.BzrProfiler()
3604
# Prevent deadlocks in tests that use lsprof: those tests will
3606
bzrlib.lsprof.BzrProfiler.profiler_block = 0
3607
self.profiler.start()
3608
testtools.ExtendedToOriginalDecorator.startTest(self, test)
3610
def addSuccess(self, test):
3611
stats = self.profiler.stop()
3613
calls = test._benchcalls
3614
except AttributeError:
3615
test._benchcalls = []
3616
calls = test._benchcalls
3617
calls.append(((test.id(), "", ""), stats))
3618
testtools.ExtendedToOriginalDecorator.addSuccess(self, test)
3620
def stopTest(self, test):
3621
testtools.ExtendedToOriginalDecorator.stopTest(self, test)
3622
self.profiler = None
2515
return result.wasSuccessful()
3625
2518
# Controlled by "bzr selftest -E=..." option
3626
# Currently supported:
3627
# -Eallow_debug Will no longer clear debug.debug_flags() so it
3628
# preserves any flags supplied at the command line.
3629
# -Edisable_lock_checks Turns errors in mismatched locks into simple prints
3630
# rather than failing tests. And no longer raise
3631
# LockContention when fctnl locks are not being used
3632
# with proper exclusion rules.
3633
# -Ethreads Will display thread ident at creation/join time to
3634
# help track thread leaks
3636
# -Econfig_stats Will collect statistics using addDetail
3637
2519
selftest_debug_flags = set()
3867
2728
test_prefix_alias_registry.register('bp', 'bzrlib.plugins')
3870
def _test_suite_testmod_names():
3871
"""Return the standard list of test module names to test."""
3874
'bzrlib.tests.blackbox',
3875
'bzrlib.tests.commands',
3876
'bzrlib.tests.doc_generate',
3877
'bzrlib.tests.per_branch',
3878
'bzrlib.tests.per_bzrdir',
3879
'bzrlib.tests.per_controldir',
3880
'bzrlib.tests.per_controldir_colo',
3881
'bzrlib.tests.per_foreign_vcs',
3882
'bzrlib.tests.per_interrepository',
3883
'bzrlib.tests.per_intertree',
3884
'bzrlib.tests.per_inventory',
3885
'bzrlib.tests.per_interbranch',
3886
'bzrlib.tests.per_lock',
3887
'bzrlib.tests.per_merger',
3888
'bzrlib.tests.per_transport',
3889
'bzrlib.tests.per_tree',
3890
'bzrlib.tests.per_pack_repository',
3891
'bzrlib.tests.per_repository',
3892
'bzrlib.tests.per_repository_chk',
3893
'bzrlib.tests.per_repository_reference',
3894
'bzrlib.tests.per_repository_vf',
3895
'bzrlib.tests.per_uifactory',
3896
'bzrlib.tests.per_versionedfile',
3897
'bzrlib.tests.per_workingtree',
3898
'bzrlib.tests.test__annotator',
3899
'bzrlib.tests.test__bencode',
3900
'bzrlib.tests.test__btree_serializer',
3901
'bzrlib.tests.test__chk_map',
3902
'bzrlib.tests.test__dirstate_helpers',
3903
'bzrlib.tests.test__groupcompress',
3904
'bzrlib.tests.test__known_graph',
3905
'bzrlib.tests.test__rio',
3906
'bzrlib.tests.test__simple_set',
3907
'bzrlib.tests.test__static_tuple',
3908
'bzrlib.tests.test__walkdirs_win32',
3909
'bzrlib.tests.test_ancestry',
3910
'bzrlib.tests.test_annotate',
3911
'bzrlib.tests.test_api',
3912
'bzrlib.tests.test_atomicfile',
3913
'bzrlib.tests.test_bad_files',
3914
'bzrlib.tests.test_bisect_multi',
3915
'bzrlib.tests.test_branch',
3916
'bzrlib.tests.test_branchbuilder',
3917
'bzrlib.tests.test_btree_index',
3918
'bzrlib.tests.test_bugtracker',
3919
'bzrlib.tests.test_bundle',
3920
'bzrlib.tests.test_bzrdir',
3921
'bzrlib.tests.test__chunks_to_lines',
3922
'bzrlib.tests.test_cache_utf8',
3923
'bzrlib.tests.test_chk_map',
3924
'bzrlib.tests.test_chk_serializer',
3925
'bzrlib.tests.test_chunk_writer',
3926
'bzrlib.tests.test_clean_tree',
3927
'bzrlib.tests.test_cleanup',
3928
'bzrlib.tests.test_cmdline',
3929
'bzrlib.tests.test_commands',
3930
'bzrlib.tests.test_commit',
3931
'bzrlib.tests.test_commit_merge',
3932
'bzrlib.tests.test_config',
3933
'bzrlib.tests.test_conflicts',
3934
'bzrlib.tests.test_controldir',
3935
'bzrlib.tests.test_counted_lock',
3936
'bzrlib.tests.test_crash',
3937
'bzrlib.tests.test_decorators',
3938
'bzrlib.tests.test_delta',
3939
'bzrlib.tests.test_debug',
3940
'bzrlib.tests.test_diff',
3941
'bzrlib.tests.test_directory_service',
3942
'bzrlib.tests.test_dirstate',
3943
'bzrlib.tests.test_email_message',
3944
'bzrlib.tests.test_eol_filters',
3945
'bzrlib.tests.test_errors',
3946
'bzrlib.tests.test_export',
3947
'bzrlib.tests.test_export_pot',
3948
'bzrlib.tests.test_extract',
3949
'bzrlib.tests.test_features',
3950
'bzrlib.tests.test_fetch',
3951
'bzrlib.tests.test_fixtures',
3952
'bzrlib.tests.test_fifo_cache',
3953
'bzrlib.tests.test_filters',
3954
'bzrlib.tests.test_filter_tree',
3955
'bzrlib.tests.test_ftp_transport',
3956
'bzrlib.tests.test_foreign',
3957
'bzrlib.tests.test_generate_docs',
3958
'bzrlib.tests.test_generate_ids',
3959
'bzrlib.tests.test_globbing',
3960
'bzrlib.tests.test_gpg',
3961
'bzrlib.tests.test_graph',
3962
'bzrlib.tests.test_groupcompress',
3963
'bzrlib.tests.test_hashcache',
3964
'bzrlib.tests.test_help',
3965
'bzrlib.tests.test_hooks',
3966
'bzrlib.tests.test_http',
3967
'bzrlib.tests.test_http_response',
3968
'bzrlib.tests.test_https_ca_bundle',
3969
'bzrlib.tests.test_i18n',
3970
'bzrlib.tests.test_identitymap',
3971
'bzrlib.tests.test_ignores',
3972
'bzrlib.tests.test_index',
3973
'bzrlib.tests.test_import_tariff',
3974
'bzrlib.tests.test_info',
3975
'bzrlib.tests.test_inv',
3976
'bzrlib.tests.test_inventory_delta',
3977
'bzrlib.tests.test_knit',
3978
'bzrlib.tests.test_lazy_import',
3979
'bzrlib.tests.test_lazy_regex',
3980
'bzrlib.tests.test_library_state',
3981
'bzrlib.tests.test_lock',
3982
'bzrlib.tests.test_lockable_files',
3983
'bzrlib.tests.test_lockdir',
3984
'bzrlib.tests.test_log',
3985
'bzrlib.tests.test_lru_cache',
3986
'bzrlib.tests.test_lsprof',
3987
'bzrlib.tests.test_mail_client',
3988
'bzrlib.tests.test_matchers',
3989
'bzrlib.tests.test_memorytree',
3990
'bzrlib.tests.test_merge',
3991
'bzrlib.tests.test_merge3',
3992
'bzrlib.tests.test_merge_core',
3993
'bzrlib.tests.test_merge_directive',
3994
'bzrlib.tests.test_mergetools',
3995
'bzrlib.tests.test_missing',
3996
'bzrlib.tests.test_msgeditor',
3997
'bzrlib.tests.test_multiparent',
3998
'bzrlib.tests.test_mutabletree',
3999
'bzrlib.tests.test_nonascii',
4000
'bzrlib.tests.test_options',
4001
'bzrlib.tests.test_osutils',
4002
'bzrlib.tests.test_osutils_encodings',
4003
'bzrlib.tests.test_pack',
4004
'bzrlib.tests.test_patch',
4005
'bzrlib.tests.test_patches',
4006
'bzrlib.tests.test_permissions',
4007
'bzrlib.tests.test_plugins',
4008
'bzrlib.tests.test_progress',
4009
'bzrlib.tests.test_pyutils',
4010
'bzrlib.tests.test_read_bundle',
4011
'bzrlib.tests.test_reconcile',
4012
'bzrlib.tests.test_reconfigure',
4013
'bzrlib.tests.test_registry',
4014
'bzrlib.tests.test_remote',
4015
'bzrlib.tests.test_rename_map',
4016
'bzrlib.tests.test_repository',
4017
'bzrlib.tests.test_revert',
4018
'bzrlib.tests.test_revision',
4019
'bzrlib.tests.test_revisionspec',
4020
'bzrlib.tests.test_revisiontree',
4021
'bzrlib.tests.test_rio',
4022
'bzrlib.tests.test_rules',
4023
'bzrlib.tests.test_sampler',
4024
'bzrlib.tests.test_scenarios',
4025
'bzrlib.tests.test_script',
4026
'bzrlib.tests.test_selftest',
4027
'bzrlib.tests.test_serializer',
4028
'bzrlib.tests.test_setup',
4029
'bzrlib.tests.test_sftp_transport',
4030
'bzrlib.tests.test_shelf',
4031
'bzrlib.tests.test_shelf_ui',
4032
'bzrlib.tests.test_smart',
4033
'bzrlib.tests.test_smart_add',
4034
'bzrlib.tests.test_smart_request',
4035
'bzrlib.tests.test_smart_transport',
4036
'bzrlib.tests.test_smtp_connection',
4037
'bzrlib.tests.test_source',
4038
'bzrlib.tests.test_ssh_transport',
4039
'bzrlib.tests.test_status',
4040
'bzrlib.tests.test_store',
4041
'bzrlib.tests.test_strace',
4042
'bzrlib.tests.test_subsume',
4043
'bzrlib.tests.test_switch',
4044
'bzrlib.tests.test_symbol_versioning',
4045
'bzrlib.tests.test_tag',
4046
'bzrlib.tests.test_test_server',
4047
'bzrlib.tests.test_testament',
4048
'bzrlib.tests.test_textfile',
4049
'bzrlib.tests.test_textmerge',
4050
'bzrlib.tests.test_cethread',
4051
'bzrlib.tests.test_timestamp',
4052
'bzrlib.tests.test_trace',
4053
'bzrlib.tests.test_transactions',
4054
'bzrlib.tests.test_transform',
4055
'bzrlib.tests.test_transport',
4056
'bzrlib.tests.test_transport_log',
4057
'bzrlib.tests.test_tree',
4058
'bzrlib.tests.test_treebuilder',
4059
'bzrlib.tests.test_treeshape',
4060
'bzrlib.tests.test_tsort',
4061
'bzrlib.tests.test_tuned_gzip',
4062
'bzrlib.tests.test_ui',
4063
'bzrlib.tests.test_uncommit',
4064
'bzrlib.tests.test_upgrade',
4065
'bzrlib.tests.test_upgrade_stacked',
4066
'bzrlib.tests.test_urlutils',
4067
'bzrlib.tests.test_utextwrap',
4068
'bzrlib.tests.test_version',
4069
'bzrlib.tests.test_version_info',
4070
'bzrlib.tests.test_versionedfile',
4071
'bzrlib.tests.test_weave',
4072
'bzrlib.tests.test_whitebox',
4073
'bzrlib.tests.test_win32utils',
4074
'bzrlib.tests.test_workingtree',
4075
'bzrlib.tests.test_workingtree_4',
4076
'bzrlib.tests.test_wsgi',
4077
'bzrlib.tests.test_xml',
4081
def _test_suite_modules_to_doctest():
4082
"""Return the list of modules to doctest."""
4084
# GZ 2009-03-31: No docstrings with -OO so there's nothing to doctest
4088
'bzrlib.branchbuilder',
4089
'bzrlib.decorators',
4091
'bzrlib.iterablefile',
4096
'bzrlib.symbol_versioning',
4098
'bzrlib.tests.fixtures',
4100
'bzrlib.transport.http',
4101
'bzrlib.version_info_formats.format_custom',
4105
2731
def test_suite(keep_only=None, starting_with=None):
4106
2732
"""Build and return TestSuite for the whole of bzrlib.
4113
2739
This function can be replaced if you need to change the default test
4114
2740
suite on a global basis, but it is not encouraged.
2744
'bzrlib.util.tests.test_bencode',
2745
'bzrlib.tests.blackbox',
2746
'bzrlib.tests.branch_implementations',
2747
'bzrlib.tests.bzrdir_implementations',
2748
'bzrlib.tests.commands',
2749
'bzrlib.tests.inventory_implementations',
2750
'bzrlib.tests.interrepository_implementations',
2751
'bzrlib.tests.intertree_implementations',
2752
'bzrlib.tests.per_lock',
2753
'bzrlib.tests.per_repository',
2754
'bzrlib.tests.test__dirstate_helpers',
2755
'bzrlib.tests.test_ancestry',
2756
'bzrlib.tests.test_annotate',
2757
'bzrlib.tests.test_api',
2758
'bzrlib.tests.test_atomicfile',
2759
'bzrlib.tests.test_bad_files',
2760
'bzrlib.tests.test_bisect_multi',
2761
'bzrlib.tests.test_branch',
2762
'bzrlib.tests.test_branchbuilder',
2763
'bzrlib.tests.test_btree_index',
2764
'bzrlib.tests.test_bugtracker',
2765
'bzrlib.tests.test_bundle',
2766
'bzrlib.tests.test_bzrdir',
2767
'bzrlib.tests.test_cache_utf8',
2768
'bzrlib.tests.test_chunk_writer',
2769
'bzrlib.tests.test_commands',
2770
'bzrlib.tests.test_commit',
2771
'bzrlib.tests.test_commit_merge',
2772
'bzrlib.tests.test_config',
2773
'bzrlib.tests.test_conflicts',
2774
'bzrlib.tests.test_counted_lock',
2775
'bzrlib.tests.test_decorators',
2776
'bzrlib.tests.test_delta',
2777
'bzrlib.tests.test_deprecated_graph',
2778
'bzrlib.tests.test_diff',
2779
'bzrlib.tests.test_dirstate',
2780
'bzrlib.tests.test_directory_service',
2781
'bzrlib.tests.test_email_message',
2782
'bzrlib.tests.test_errors',
2783
'bzrlib.tests.test_extract',
2784
'bzrlib.tests.test_fetch',
2785
'bzrlib.tests.test_ftp_transport',
2786
'bzrlib.tests.test_generate_docs',
2787
'bzrlib.tests.test_generate_ids',
2788
'bzrlib.tests.test_globbing',
2789
'bzrlib.tests.test_gpg',
2790
'bzrlib.tests.test_graph',
2791
'bzrlib.tests.test_hashcache',
2792
'bzrlib.tests.test_help',
2793
'bzrlib.tests.test_hooks',
2794
'bzrlib.tests.test_http',
2795
'bzrlib.tests.test_http_implementations',
2796
'bzrlib.tests.test_http_response',
2797
'bzrlib.tests.test_https_ca_bundle',
2798
'bzrlib.tests.test_identitymap',
2799
'bzrlib.tests.test_ignores',
2800
'bzrlib.tests.test_index',
2801
'bzrlib.tests.test_info',
2802
'bzrlib.tests.test_inv',
2803
'bzrlib.tests.test_knit',
2804
'bzrlib.tests.test_lazy_import',
2805
'bzrlib.tests.test_lazy_regex',
2806
'bzrlib.tests.test_lockdir',
2807
'bzrlib.tests.test_lockable_files',
2808
'bzrlib.tests.test_log',
2809
'bzrlib.tests.test_lsprof',
2810
'bzrlib.tests.test_lru_cache',
2811
'bzrlib.tests.test_mail_client',
2812
'bzrlib.tests.test_memorytree',
2813
'bzrlib.tests.test_merge',
2814
'bzrlib.tests.test_merge3',
2815
'bzrlib.tests.test_merge_core',
2816
'bzrlib.tests.test_merge_directive',
2817
'bzrlib.tests.test_missing',
2818
'bzrlib.tests.test_msgeditor',
2819
'bzrlib.tests.test_multiparent',
2820
'bzrlib.tests.test_mutabletree',
2821
'bzrlib.tests.test_nonascii',
2822
'bzrlib.tests.test_options',
2823
'bzrlib.tests.test_osutils',
2824
'bzrlib.tests.test_osutils_encodings',
2825
'bzrlib.tests.test_pack',
2826
'bzrlib.tests.test_pack_repository',
2827
'bzrlib.tests.test_patch',
2828
'bzrlib.tests.test_patches',
2829
'bzrlib.tests.test_permissions',
2830
'bzrlib.tests.test_plugins',
2831
'bzrlib.tests.test_progress',
2832
'bzrlib.tests.test_read_bundle',
2833
'bzrlib.tests.test_reconfigure',
2834
'bzrlib.tests.test_reconcile',
2835
'bzrlib.tests.test_registry',
2836
'bzrlib.tests.test_remote',
2837
'bzrlib.tests.test_repository',
2838
'bzrlib.tests.per_repository_reference',
2839
'bzrlib.tests.test_revert',
2840
'bzrlib.tests.test_revision',
2841
'bzrlib.tests.test_revisionspec',
2842
'bzrlib.tests.test_revisiontree',
2843
'bzrlib.tests.test_rio',
2844
'bzrlib.tests.test_rules',
2845
'bzrlib.tests.test_sampler',
2846
'bzrlib.tests.test_selftest',
2847
'bzrlib.tests.test_setup',
2848
'bzrlib.tests.test_sftp_transport',
2849
'bzrlib.tests.test_smart',
2850
'bzrlib.tests.test_smart_add',
2851
'bzrlib.tests.test_smart_transport',
2852
'bzrlib.tests.test_smtp_connection',
2853
'bzrlib.tests.test_source',
2854
'bzrlib.tests.test_ssh_transport',
2855
'bzrlib.tests.test_status',
2856
'bzrlib.tests.test_store',
2857
'bzrlib.tests.test_strace',
2858
'bzrlib.tests.test_subsume',
2859
'bzrlib.tests.test_switch',
2860
'bzrlib.tests.test_symbol_versioning',
2861
'bzrlib.tests.test_tag',
2862
'bzrlib.tests.test_testament',
2863
'bzrlib.tests.test_textfile',
2864
'bzrlib.tests.test_textmerge',
2865
'bzrlib.tests.test_timestamp',
2866
'bzrlib.tests.test_trace',
2867
'bzrlib.tests.test_transactions',
2868
'bzrlib.tests.test_transform',
2869
'bzrlib.tests.test_transport',
2870
'bzrlib.tests.test_transport_implementations',
2871
'bzrlib.tests.test_transport_log',
2872
'bzrlib.tests.test_tree',
2873
'bzrlib.tests.test_treebuilder',
2874
'bzrlib.tests.test_tsort',
2875
'bzrlib.tests.test_tuned_gzip',
2876
'bzrlib.tests.test_ui',
2877
'bzrlib.tests.test_uncommit',
2878
'bzrlib.tests.test_upgrade',
2879
'bzrlib.tests.test_upgrade_stacked',
2880
'bzrlib.tests.test_urlutils',
2881
'bzrlib.tests.test_versionedfile',
2882
'bzrlib.tests.test_version',
2883
'bzrlib.tests.test_version_info',
2884
'bzrlib.tests.test__walkdirs_win32',
2885
'bzrlib.tests.test_weave',
2886
'bzrlib.tests.test_whitebox',
2887
'bzrlib.tests.test_win32utils',
2888
'bzrlib.tests.test_workingtree',
2889
'bzrlib.tests.test_workingtree_4',
2890
'bzrlib.tests.test_wsgi',
2891
'bzrlib.tests.test_xml',
2892
'bzrlib.tests.tree_implementations',
2893
'bzrlib.tests.workingtree_implementations',
4117
2896
loader = TestUtil.TestLoader()
4119
if keep_only is not None:
4120
id_filter = TestIdList(keep_only)
4121
2898
if starting_with:
2899
starting_with = [test_prefix_alias_registry.resolve_alias(start)
2900
for start in starting_with]
4122
2901
# We take precedence over keep_only because *at loading time* using
4123
2902
# both options means we will load less tests for the same final result.
4124
2903
def interesting_module(name):
4232
3062
for right_name, right_dict in scenarios_right]
4235
def multiply_tests(tests, scenarios, result):
4236
"""Multiply tests_list by scenarios into result.
4238
This is the core workhorse for test parameterisation.
4240
Typically the load_tests() method for a per-implementation test suite will
4241
call multiply_tests and return the result.
4243
:param tests: The tests to parameterise.
4244
:param scenarios: The scenarios to apply: pairs of (scenario_name,
4245
scenario_param_dict).
4246
:param result: A TestSuite to add created tests to.
4248
This returns the passed in result TestSuite with the cross product of all
4249
the tests repeated once for each scenario. Each test is adapted by adding
4250
the scenario name at the end of its id(), and updating the test object's
4251
__dict__ with the scenario_param_dict.
4253
>>> import bzrlib.tests.test_sampler
4254
>>> r = multiply_tests(
4255
... bzrlib.tests.test_sampler.DemoTest('test_nothing'),
4256
... [('one', dict(param=1)),
4257
... ('two', dict(param=2))],
4258
... TestUtil.TestSuite())
4259
>>> tests = list(iter_suite_tests(r))
4263
'bzrlib.tests.test_sampler.DemoTest.test_nothing(one)'
4269
for test in iter_suite_tests(tests):
4270
apply_scenarios(test, scenarios, result)
4274
def apply_scenarios(test, scenarios, result):
4275
"""Apply the scenarios in scenarios to test and add to result.
4277
:param test: The test to apply scenarios to.
4278
:param scenarios: An iterable of scenarios to apply to test.
4280
:seealso: apply_scenario
4282
for scenario in scenarios:
4283
result.addTest(apply_scenario(test, scenario))
4287
def apply_scenario(test, scenario):
4288
"""Copy test and apply scenario to it.
4290
:param test: A test to adapt.
4291
:param scenario: A tuple describing the scenarion.
4292
The first element of the tuple is the new test id.
4293
The second element is a dict containing attributes to set on the
4295
:return: The adapted test.
4297
new_id = "%s(%s)" % (test.id(), scenario[0])
4298
new_test = clone_test(test, new_id)
4299
for name, value in scenario[1].items():
4300
setattr(new_test, name, value)
4304
def clone_test(test, new_id):
4305
"""Clone a test giving it a new id.
4307
:param test: The test to clone.
4308
:param new_id: The id to assign to it.
4309
:return: The new test.
4311
new_test = copy.copy(test)
4312
new_test.id = lambda: new_id
4313
# XXX: Workaround <https://bugs.launchpad.net/testtools/+bug/637725>, which
4314
# causes cloned tests to share the 'details' dict. This makes it hard to
4315
# read the test output for parameterized tests, because tracebacks will be
4316
# associated with irrelevant tests.
4318
details = new_test._TestCase__details
4319
except AttributeError:
4320
# must be a different version of testtools than expected. Do nothing.
4323
# Reset the '__details' dict.
4324
new_test._TestCase__details = {}
4328
def permute_tests_for_extension(standard_tests, loader, py_module_name,
4330
"""Helper for permutating tests against an extension module.
4332
This is meant to be used inside a modules 'load_tests()' function. It will
4333
create 2 scenarios, and cause all tests in the 'standard_tests' to be run
4334
against both implementations. Setting 'test.module' to the appropriate
4335
module. See bzrlib.tests.test__chk_map.load_tests as an example.
4337
:param standard_tests: A test suite to permute
4338
:param loader: A TestLoader
4339
:param py_module_name: The python path to a python module that can always
4340
be loaded, and will be considered the 'python' implementation. (eg
4341
'bzrlib._chk_map_py')
4342
:param ext_module_name: The python path to an extension module. If the
4343
module cannot be loaded, a single test will be added, which notes that
4344
the module is not available. If it can be loaded, all standard_tests
4345
will be run against that module.
4346
:return: (suite, feature) suite is a test-suite that has all the permuted
4347
tests. feature is the Feature object that can be used to determine if
4348
the module is available.
4351
from bzrlib.tests.features import ModuleAvailableFeature
4352
py_module = pyutils.get_named_object(py_module_name)
4354
('python', {'module': py_module}),
4356
suite = loader.suiteClass()
4357
feature = ModuleAvailableFeature(ext_module_name)
4358
if feature.available():
4359
scenarios.append(('C', {'module': feature.module}))
4361
# the compiled module isn't available, so we add a failing test
4362
class FailWithoutFeature(TestCase):
4363
def test_fail(self):
4364
self.requireFeature(feature)
4365
suite.addTest(loader.loadTestsFromTestCase(FailWithoutFeature))
4366
result = multiply_tests(standard_tests, scenarios, suite)
4367
return result, feature
4370
def _rmtree_temp_dir(dirname, test_id=None):
3066
def adapt_modules(mods_list, adapter, loader, suite):
3067
"""Adapt the modules in mods_list using adapter and add to suite."""
3068
tests = loader.loadTestsFromModuleNames(mods_list)
3069
adapt_tests(tests, adapter, suite)
3072
def adapt_tests(tests_list, adapter, suite):
3073
"""Adapt the tests in tests_list using adapter and add to suite."""
3074
for test in iter_suite_tests(tests_list):
3075
suite.addTests(adapter.adapt(test))
3078
def _rmtree_temp_dir(dirname):
4371
3079
# If LANG=C we probably have created some bogus paths
4372
3080
# which rmtree(unicode) will fail to delete
4373
3081
# so make sure we are using rmtree(str) to delete everything
4383
3091
osutils.rmtree(dirname)
4384
3092
except OSError, e:
4385
# We don't want to fail here because some useful display will be lost
4386
# otherwise. Polluting the tmp dir is bad, but not giving all the
4387
# possible info to the test runner is even worse.
4389
ui.ui_factory.clear_term()
4390
sys.stderr.write('\nWhile running: %s\n' % (test_id,))
4391
# Ugly, but the last thing we want here is fail, so bear with it.
4392
printable_e = str(e).decode(osutils.get_user_encoding(), 'replace'
4393
).encode('ascii', 'replace')
4394
sys.stderr.write('Unable to remove testing dir %s\n%s'
4395
% (os.path.basename(dirname), printable_e))
3093
if sys.platform == 'win32' and e.errno == errno.EACCES:
3094
sys.stderr.write(('Permission denied: '
3095
'unable to remove testing dir '
3096
'%s\n' % os.path.basename(dirname)))
3101
class Feature(object):
3102
"""An operating system Feature."""
3105
self._available = None
3107
def available(self):
3108
"""Is the feature available?
3110
:return: True if the feature is available.
3112
if self._available is None:
3113
self._available = self._probe()
3114
return self._available
3117
"""Implement this method in concrete features.
3119
:return: True if the feature is available.
3121
raise NotImplementedError
3124
if getattr(self, 'feature_name', None):
3125
return self.feature_name()
3126
return self.__class__.__name__
3129
class _SymlinkFeature(Feature):
3132
return osutils.has_symlinks()
3134
def feature_name(self):
3137
SymlinkFeature = _SymlinkFeature()
3140
class _HardlinkFeature(Feature):
3143
return osutils.has_hardlinks()
3145
def feature_name(self):
3148
HardlinkFeature = _HardlinkFeature()
3151
class _OsFifoFeature(Feature):
3154
return getattr(os, 'mkfifo', None)
3156
def feature_name(self):
3157
return 'filesystem fifos'
3159
OsFifoFeature = _OsFifoFeature()
3162
class _UnicodeFilenameFeature(Feature):
3163
"""Does the filesystem support Unicode filenames?"""
3167
# Check for character combinations unlikely to be covered by any
3168
# single non-unicode encoding. We use the characters
3169
# - greek small letter alpha (U+03B1) and
3170
# - braille pattern dots-123456 (U+283F).
3171
os.stat(u'\u03b1\u283f')
3172
except UnicodeEncodeError:
3174
except (IOError, OSError):
3175
# The filesystem allows the Unicode filename but the file doesn't
3179
# The filesystem allows the Unicode filename and the file exists,
3183
UnicodeFilenameFeature = _UnicodeFilenameFeature()
3186
class TestScenarioApplier(object):
3187
"""A tool to apply scenarios to tests."""
3189
def adapt(self, test):
3190
"""Return a TestSuite containing a copy of test for each scenario."""
3191
result = unittest.TestSuite()
3192
for scenario in self.scenarios:
3193
result.addTest(self.adapt_test_to_scenario(test, scenario))
3196
def adapt_test_to_scenario(self, test, scenario):
3197
"""Copy test and apply scenario to it.
3199
:param test: A test to adapt.
3200
:param scenario: A tuple describing the scenarion.
3201
The first element of the tuple is the new test id.
3202
The second element is a dict containing attributes to set on the
3204
:return: The adapted test.
3206
from copy import deepcopy
3207
new_test = deepcopy(test)
3208
for name, value in scenario[1].items():
3209
setattr(new_test, name, value)
3210
new_id = "%s(%s)" % (new_test.id(), scenario[0])
3211
new_test.id = lambda: new_id
4398
3215
def probe_unicode_in_user_encoding():
added
removed
bzrlib/tests/__init__.py
