1
# Copyright (C) 2005-2011 Canonical Ltd
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
# GNU General Public License for more details.
13
# You should have received a copy of the GNU General Public License
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
17
"""Testing framework extensions"""
19
from __future__ import absolute_import
21
# NOTE: Some classes in here use camelCaseNaming() rather than
22
# underscore_naming(). That's for consistency with unittest; it's not the
23
# general style of bzrlib. Please continue that consistency when adding e.g.
24
# new assertFoo() methods.
29
from cStringIO import StringIO
53
# nb: check this before importing anything else from within it
54
_testtools_version = getattr(testtools, '__version__', ())
55
if _testtools_version < (0, 9, 5):
56
raise ImportError("need at least testtools 0.9.5: %s is %r"
57
% (testtools.__file__, _testtools_version))
58
from testtools import content
65
commands as _mod_commands,
75
plugin as _mod_plugin,
82
transport as _mod_transport,
88
# lsprof not available
90
from bzrlib.smart import client, request
91
from bzrlib.transport import (
95
from bzrlib.symbol_versioning import (
99
from bzrlib.tests import (
105
from bzrlib.ui import NullProgressView
106
from bzrlib.ui.text import TextUIFactory
107
from bzrlib.tests.features import _CompatabilityThunkFeature
109
# Mark this python module as being part of the implementation
110
# of unittest: this gives us better tracebacks where the last
111
# shown frame is the test code, not our assertXYZ.
114
default_transport = test_server.LocalURLServer
117
_unitialized_attr = object()
118
"""A sentinel needed to act as a default value in a method signature."""
121
# Subunit result codes, defined here to prevent a hard dependency on subunit.
125
# These are intentionally brought into this namespace. That way plugins, etc
126
# can just "from bzrlib.tests import TestCase, TestLoader, etc"
127
TestSuite = TestUtil.TestSuite
128
TestLoader = TestUtil.TestLoader
130
# Tests should run in a clean and clearly defined environment. The goal is to
131
# keep them isolated from the running environment as mush as possible. The test
132
# framework ensures the variables defined below are set (or deleted if the
133
# value is None) before a test is run and reset to their original value after
134
# the test is run. Generally if some code depends on an environment variable,
135
# the tests should start without this variable in the environment. There are a
136
# few exceptions but you shouldn't violate this rule lightly.
140
# bzr now uses the Win32 API and doesn't rely on APPDATA, but the
141
# tests do check our impls match APPDATA
142
'BZR_EDITOR': None, # test_msgeditor manipulates this variable
146
'BZREMAIL': None, # may still be present in the environment
147
'EMAIL': 'jrandom@example.com', # set EMAIL as bzr does not guess
148
'BZR_PROGRESS_BAR': None,
149
# This should trap leaks to ~/.bzr.log. This occurs when tests use TestCase
150
# as a base class instead of TestCaseInTempDir. Tests inheriting from
151
# TestCase should not use disk resources, BZR_LOG is one.
152
'BZR_LOG': '/you-should-use-TestCaseInTempDir-if-you-need-a-log-file',
153
'BZR_PLUGIN_PATH': None,
154
'BZR_DISABLE_PLUGINS': None,
155
'BZR_PLUGINS_AT': None,
156
'BZR_CONCURRENCY': None,
157
# Make sure that any text ui tests are consistent regardless of
158
# the environment the test case is run in; you may want tests that
159
# test other combinations. 'dumb' is a reasonable guess for tests
160
# going to a pipe or a StringIO.
166
'SSH_AUTH_SOCK': None,
176
# Nobody cares about ftp_proxy, FTP_PROXY AFAIK. So far at
177
# least. If you do (care), please update this comment
181
'BZR_REMOTE_PATH': None,
182
# Generally speaking, we don't want apport reporting on crashes in
183
# the test envirnoment unless we're specifically testing apport,
184
# so that it doesn't leak into the real system environment. We
185
# use an env var so it propagates to subprocesses.
186
'APPORT_DISABLE': '1',
190
def override_os_environ(test, env=None):
191
"""Modify os.environ keeping a copy.
193
:param test: A test instance
195
:param env: A dict containing variable definitions to be installed
198
env = isolated_environ
199
test._original_os_environ = dict([(var, value)
200
for var, value in os.environ.iteritems()])
201
for var, value in env.iteritems():
202
osutils.set_or_unset_env(var, value)
203
if var not in test._original_os_environ:
204
# The var is new, add it with a value of None, so
205
# restore_os_environ will delete it
206
test._original_os_environ[var] = None
209
def restore_os_environ(test):
210
"""Restore os.environ to its original state.
212
:param test: A test instance previously passed to override_os_environ.
214
for var, value in test._original_os_environ.iteritems():
215
# Restore the original value (or delete it if the value has been set to
216
# None in override_os_environ).
217
osutils.set_or_unset_env(var, value)
220
def _clear__type_equality_funcs(test):
221
"""Cleanup bound methods stored on TestCase instances
223
Clear the dict breaking a few (mostly) harmless cycles in the affected
224
unittests released with Python 2.6 and initial Python 2.7 versions.
226
For a few revisions between Python 2.7.1 and Python 2.7.2 that annoyingly
227
shipped in Oneiric, an object with no clear method was used, hence the
228
extra complications, see bug 809048 for details.
230
type_equality_funcs = getattr(test, "_type_equality_funcs", None)
231
if type_equality_funcs is not None:
232
tef_clear = getattr(type_equality_funcs, "clear", None)
233
if tef_clear is None:
234
tef_instance_dict = getattr(type_equality_funcs, "__dict__", None)
235
if tef_instance_dict is not None:
236
tef_clear = tef_instance_dict.clear
237
if tef_clear is not None:
241
class ExtendedTestResult(testtools.TextTestResult):
242
"""Accepts, reports and accumulates the results of running tests.
244
Compared to the unittest version this class adds support for
245
profiling, benchmarking, stopping as soon as a test fails, and
246
skipping tests. There are further-specialized subclasses for
247
different types of display.
249
When a test finishes, in whatever way, it calls one of the addSuccess,
250
addFailure or addError methods. These in turn may redirect to a more
251
specific case for the special test results supported by our extended
254
Note that just one of these objects is fed the results from many tests.
259
def __init__(self, stream, descriptions, verbosity,
263
"""Construct new TestResult.
265
:param bench_history: Optionally, a writable file object to accumulate
268
testtools.TextTestResult.__init__(self, stream)
269
if bench_history is not None:
270
from bzrlib.version import _get_bzr_source_tree
271
src_tree = _get_bzr_source_tree()
274
revision_id = src_tree.get_parent_ids()[0]
276
# XXX: if this is a brand new tree, do the same as if there
280
# XXX: If there's no branch, what should we do?
282
bench_history.write("--date %s %s\n" % (time.time(), revision_id))
283
self._bench_history = bench_history
284
self.ui = ui.ui_factory
287
self.failure_count = 0
288
self.known_failure_count = 0
290
self.not_applicable_count = 0
291
self.unsupported = {}
293
self._overall_start_time = time.time()
294
self._strict = strict
295
self._first_thread_leaker_id = None
296
self._tests_leaking_threads_count = 0
297
self._traceback_from_test = None
299
def stopTestRun(self):
302
stopTime = time.time()
303
timeTaken = stopTime - self.startTime
304
# GZ 2010-07-19: Seems testtools has no printErrors method, and though
305
# the parent class method is similar have to duplicate
306
self._show_list('ERROR', self.errors)
307
self._show_list('FAIL', self.failures)
308
self.stream.write(self.sep2)
309
self.stream.write("%s %d test%s in %.3fs\n\n" % (actionTaken,
310
run, run != 1 and "s" or "", timeTaken))
311
if not self.wasSuccessful():
312
self.stream.write("FAILED (")
313
failed, errored = map(len, (self.failures, self.errors))
315
self.stream.write("failures=%d" % failed)
317
if failed: self.stream.write(", ")
318
self.stream.write("errors=%d" % errored)
319
if self.known_failure_count:
320
if failed or errored: self.stream.write(", ")
321
self.stream.write("known_failure_count=%d" %
322
self.known_failure_count)
323
self.stream.write(")\n")
325
if self.known_failure_count:
326
self.stream.write("OK (known_failures=%d)\n" %
327
self.known_failure_count)
329
self.stream.write("OK\n")
330
if self.skip_count > 0:
331
skipped = self.skip_count
332
self.stream.write('%d test%s skipped\n' %
333
(skipped, skipped != 1 and "s" or ""))
335
for feature, count in sorted(self.unsupported.items()):
336
self.stream.write("Missing feature '%s' skipped %d tests.\n" %
339
ok = self.wasStrictlySuccessful()
341
ok = self.wasSuccessful()
342
if self._first_thread_leaker_id:
344
'%s is leaking threads among %d leaking tests.\n' % (
345
self._first_thread_leaker_id,
346
self._tests_leaking_threads_count))
347
# We don't report the main thread as an active one.
349
'%d non-main threads were left active in the end.\n'
350
% (len(self._active_threads) - 1))
352
def getDescription(self, test):
355
def _extractBenchmarkTime(self, testCase, details=None):
356
"""Add a benchmark time for the current test case."""
357
if details and 'benchtime' in details:
358
return float(''.join(details['benchtime'].iter_bytes()))
359
return getattr(testCase, "_benchtime", None)
361
def _elapsedTestTimeString(self):
362
"""Return a time string for the overall time the current test has taken."""
363
return self._formatTime(self._delta_to_float(
364
self._now() - self._start_datetime))
366
def _testTimeString(self, testCase):
367
benchmark_time = self._extractBenchmarkTime(testCase)
368
if benchmark_time is not None:
369
return self._formatTime(benchmark_time) + "*"
371
return self._elapsedTestTimeString()
373
def _formatTime(self, seconds):
374
"""Format seconds as milliseconds with leading spaces."""
375
# some benchmarks can take thousands of seconds to run, so we need 8
377
return "%8dms" % (1000 * seconds)
379
def _shortened_test_description(self, test):
381
what = re.sub(r'^bzrlib\.tests\.', '', what)
384
# GZ 2010-10-04: Cloned tests may end up harmlessly calling this method
385
# multiple times in a row, because the handler is added for
386
# each test but the container list is shared between cases.
387
# See lp:498869 lp:625574 and lp:637725 for background.
388
def _record_traceback_from_test(self, exc_info):
389
"""Store the traceback from passed exc_info tuple till"""
390
self._traceback_from_test = exc_info[2]
392
def startTest(self, test):
393
super(ExtendedTestResult, self).startTest(test)
397
self.report_test_start(test)
398
test.number = self.count
399
self._recordTestStartTime()
400
# Make testtools cases give us the real traceback on failure
401
addOnException = getattr(test, "addOnException", None)
402
if addOnException is not None:
403
addOnException(self._record_traceback_from_test)
404
# Only check for thread leaks on bzrlib derived test cases
405
if isinstance(test, TestCase):
406
test.addCleanup(self._check_leaked_threads, test)
408
def stopTest(self, test):
409
super(ExtendedTestResult, self).stopTest(test)
410
# Manually break cycles, means touching various private things but hey
411
getDetails = getattr(test, "getDetails", None)
412
if getDetails is not None:
414
_clear__type_equality_funcs(test)
415
self._traceback_from_test = None
417
def startTests(self):
418
self.report_tests_starting()
419
self._active_threads = threading.enumerate()
421
def _check_leaked_threads(self, test):
422
"""See if any threads have leaked since last call
424
A sample of live threads is stored in the _active_threads attribute,
425
when this method runs it compares the current live threads and any not
426
in the previous sample are treated as having leaked.
428
now_active_threads = set(threading.enumerate())
429
threads_leaked = now_active_threads.difference(self._active_threads)
431
self._report_thread_leak(test, threads_leaked, now_active_threads)
432
self._tests_leaking_threads_count += 1
433
if self._first_thread_leaker_id is None:
434
self._first_thread_leaker_id = test.id()
435
self._active_threads = now_active_threads
437
def _recordTestStartTime(self):
438
"""Record that a test has started."""
439
self._start_datetime = self._now()
441
def addError(self, test, err):
442
"""Tell result that test finished with an error.
444
Called from the TestCase run() method when the test
445
fails with an unexpected error.
447
self._post_mortem(self._traceback_from_test)
448
super(ExtendedTestResult, self).addError(test, err)
449
self.error_count += 1
450
self.report_error(test, err)
454
def addFailure(self, test, err):
455
"""Tell result that test failed.
457
Called from the TestCase run() method when the test
458
fails because e.g. an assert() method failed.
460
self._post_mortem(self._traceback_from_test)
461
super(ExtendedTestResult, self).addFailure(test, err)
462
self.failure_count += 1
463
self.report_failure(test, err)
467
def addSuccess(self, test, details=None):
468
"""Tell result that test completed successfully.
470
Called from the TestCase run()
472
if self._bench_history is not None:
473
benchmark_time = self._extractBenchmarkTime(test, details)
474
if benchmark_time is not None:
475
self._bench_history.write("%s %s\n" % (
476
self._formatTime(benchmark_time),
478
self.report_success(test)
479
super(ExtendedTestResult, self).addSuccess(test)
480
test._log_contents = ''
482
def addExpectedFailure(self, test, err):
483
self.known_failure_count += 1
484
self.report_known_failure(test, err)
486
def addUnexpectedSuccess(self, test, details=None):
487
"""Tell result the test unexpectedly passed, counting as a failure
489
When the minimum version of testtools required becomes 0.9.8 this
490
can be updated to use the new handling there.
492
super(ExtendedTestResult, self).addFailure(test, details=details)
493
self.failure_count += 1
494
self.report_unexpected_success(test,
495
"".join(details["reason"].iter_text()))
499
def addNotSupported(self, test, feature):
500
"""The test will not be run because of a missing feature.
502
# this can be called in two different ways: it may be that the
503
# test started running, and then raised (through requireFeature)
504
# UnavailableFeature. Alternatively this method can be called
505
# while probing for features before running the test code proper; in
506
# that case we will see startTest and stopTest, but the test will
507
# never actually run.
508
self.unsupported.setdefault(str(feature), 0)
509
self.unsupported[str(feature)] += 1
510
self.report_unsupported(test, feature)
512
def addSkip(self, test, reason):
513
"""A test has not run for 'reason'."""
515
self.report_skip(test, reason)
517
def addNotApplicable(self, test, reason):
518
self.not_applicable_count += 1
519
self.report_not_applicable(test, reason)
521
def _count_stored_tests(self):
522
"""Count of tests instances kept alive due to not succeeding"""
523
return self.error_count + self.failure_count + self.known_failure_count
525
def _post_mortem(self, tb=None):
526
"""Start a PDB post mortem session."""
527
if os.environ.get('BZR_TEST_PDB', None):
531
def progress(self, offset, whence):
532
"""The test is adjusting the count of tests to run."""
533
if whence == SUBUNIT_SEEK_SET:
534
self.num_tests = offset
535
elif whence == SUBUNIT_SEEK_CUR:
536
self.num_tests += offset
538
raise errors.BzrError("Unknown whence %r" % whence)
540
def report_tests_starting(self):
541
"""Display information before the test run begins"""
542
if getattr(sys, 'frozen', None) is None:
543
bzr_path = osutils.realpath(sys.argv[0])
545
bzr_path = sys.executable
547
'bzr selftest: %s\n' % (bzr_path,))
550
bzrlib.__path__[0],))
552
' bzr-%s python-%s %s\n' % (
553
bzrlib.version_string,
554
bzrlib._format_version_tuple(sys.version_info),
555
platform.platform(aliased=1),
557
self.stream.write('\n')
559
def report_test_start(self, test):
560
"""Display information on the test just about to be run"""
562
def _report_thread_leak(self, test, leaked_threads, active_threads):
563
"""Display information on a test that leaked one or more threads"""
564
# GZ 2010-09-09: A leak summary reported separately from the general
565
# thread debugging would be nice. Tests under subunit
566
# need something not using stream, perhaps adding a
567
# testtools details object would be fitting.
568
if 'threads' in selftest_debug_flags:
569
self.stream.write('%s is leaking, active is now %d\n' %
570
(test.id(), len(active_threads)))
572
def startTestRun(self):
573
self.startTime = time.time()
575
def report_success(self, test):
578
def wasStrictlySuccessful(self):
579
if self.unsupported or self.known_failure_count:
581
return self.wasSuccessful()
584
class TextTestResult(ExtendedTestResult):
585
"""Displays progress and results of tests in text form"""
587
def __init__(self, stream, descriptions, verbosity,
592
ExtendedTestResult.__init__(self, stream, descriptions, verbosity,
593
bench_history, strict)
594
# We no longer pass them around, but just rely on the UIFactory stack
597
warnings.warn("Passing pb to TextTestResult is deprecated")
598
self.pb = self.ui.nested_progress_bar()
599
self.pb.show_pct = False
600
self.pb.show_spinner = False
601
self.pb.show_eta = False,
602
self.pb.show_count = False
603
self.pb.show_bar = False
604
self.pb.update_latency = 0
605
self.pb.show_transport_activity = False
607
def stopTestRun(self):
608
# called when the tests that are going to run have run
611
super(TextTestResult, self).stopTestRun()
613
def report_tests_starting(self):
614
super(TextTestResult, self).report_tests_starting()
615
self.pb.update('[test 0/%d] Starting' % (self.num_tests))
617
def _progress_prefix_text(self):
618
# the longer this text, the less space we have to show the test
620
a = '[%d' % self.count # total that have been run
621
# tests skipped as known not to be relevant are not important enough
623
## if self.skip_count:
624
## a += ', %d skip' % self.skip_count
625
## if self.known_failure_count:
626
## a += '+%dX' % self.known_failure_count
628
a +='/%d' % self.num_tests
630
runtime = time.time() - self._overall_start_time
632
a += '%dm%ds' % (runtime / 60, runtime % 60)
635
total_fail_count = self.error_count + self.failure_count
637
a += ', %d failed' % total_fail_count
638
# if self.unsupported:
639
# a += ', %d missing' % len(self.unsupported)
643
def report_test_start(self, test):
645
self._progress_prefix_text()
647
+ self._shortened_test_description(test))
649
def _test_description(self, test):
650
return self._shortened_test_description(test)
652
def report_error(self, test, err):
653
self.stream.write('ERROR: %s\n %s\n' % (
654
self._test_description(test),
658
def report_failure(self, test, err):
659
self.stream.write('FAIL: %s\n %s\n' % (
660
self._test_description(test),
664
def report_known_failure(self, test, err):
667
def report_unexpected_success(self, test, reason):
668
self.stream.write('FAIL: %s\n %s: %s\n' % (
669
self._test_description(test),
670
"Unexpected success. Should have failed",
674
def report_skip(self, test, reason):
677
def report_not_applicable(self, test, reason):
680
def report_unsupported(self, test, feature):
681
"""test cannot be run because feature is missing."""
684
class VerboseTestResult(ExtendedTestResult):
685
"""Produce long output, with one line per test run plus times"""
687
def _ellipsize_to_right(self, a_string, final_width):
688
"""Truncate and pad a string, keeping the right hand side"""
689
if len(a_string) > final_width:
690
result = '...' + a_string[3-final_width:]
693
return result.ljust(final_width)
695
def report_tests_starting(self):
696
self.stream.write('running %d tests...\n' % self.num_tests)
697
super(VerboseTestResult, self).report_tests_starting()
699
def report_test_start(self, test):
700
name = self._shortened_test_description(test)
701
width = osutils.terminal_width()
702
if width is not None:
703
# width needs space for 6 char status, plus 1 for slash, plus an
704
# 11-char time string, plus a trailing blank
705
# when NUMBERED_DIRS: plus 5 chars on test number, plus 1 char on
707
self.stream.write(self._ellipsize_to_right(name, width-18))
709
self.stream.write(name)
712
def _error_summary(self, err):
714
return '%s%s' % (indent, err[1])
716
def report_error(self, test, err):
717
self.stream.write('ERROR %s\n%s\n'
718
% (self._testTimeString(test),
719
self._error_summary(err)))
721
def report_failure(self, test, err):
722
self.stream.write(' FAIL %s\n%s\n'
723
% (self._testTimeString(test),
724
self._error_summary(err)))
726
def report_known_failure(self, test, err):
727
self.stream.write('XFAIL %s\n%s\n'
728
% (self._testTimeString(test),
729
self._error_summary(err)))
731
def report_unexpected_success(self, test, reason):
732
self.stream.write(' FAIL %s\n%s: %s\n'
733
% (self._testTimeString(test),
734
"Unexpected success. Should have failed",
737
def report_success(self, test):
738
self.stream.write(' OK %s\n' % self._testTimeString(test))
739
for bench_called, stats in getattr(test, '_benchcalls', []):
740
self.stream.write('LSProf output for %s(%s, %s)\n' % bench_called)
741
stats.pprint(file=self.stream)
742
# flush the stream so that we get smooth output. This verbose mode is
743
# used to show the output in PQM.
746
def report_skip(self, test, reason):
747
self.stream.write(' SKIP %s\n%s\n'
748
% (self._testTimeString(test), reason))
750
def report_not_applicable(self, test, reason):
751
self.stream.write(' N/A %s\n %s\n'
752
% (self._testTimeString(test), reason))
754
def report_unsupported(self, test, feature):
755
"""test cannot be run because feature is missing."""
756
self.stream.write("NODEP %s\n The feature '%s' is not available.\n"
757
%(self._testTimeString(test), feature))
760
class TextTestRunner(object):
761
stop_on_failure = False
769
result_decorators=None,
771
"""Create a TextTestRunner.
773
:param result_decorators: An optional list of decorators to apply
774
to the result object being used by the runner. Decorators are
775
applied left to right - the first element in the list is the
778
# stream may know claim to know to write unicode strings, but in older
779
# pythons this goes sufficiently wrong that it is a bad idea. (
780
# specifically a built in file with encoding 'UTF-8' will still try
781
# to encode using ascii.
782
new_encoding = osutils.get_terminal_encoding()
783
codec = codecs.lookup(new_encoding)
784
if type(codec) is tuple:
788
encode = codec.encode
789
# GZ 2010-09-08: Really we don't want to be writing arbitrary bytes,
790
# so should swap to the plain codecs.StreamWriter
791
stream = osutils.UnicodeOrBytesToBytesWriter(encode, stream,
793
stream.encoding = new_encoding
795
self.descriptions = descriptions
796
self.verbosity = verbosity
797
self._bench_history = bench_history
798
self._strict = strict
799
self._result_decorators = result_decorators or []
802
"Run the given test case or test suite."
803
if self.verbosity == 1:
804
result_class = TextTestResult
805
elif self.verbosity >= 2:
806
result_class = VerboseTestResult
807
original_result = result_class(self.stream,
810
bench_history=self._bench_history,
813
# Signal to result objects that look at stop early policy to stop,
814
original_result.stop_early = self.stop_on_failure
815
result = original_result
816
for decorator in self._result_decorators:
817
result = decorator(result)
818
result.stop_early = self.stop_on_failure
819
result.startTestRun()
824
# higher level code uses our extended protocol to determine
825
# what exit code to give.
826
return original_result
829
def iter_suite_tests(suite):
830
"""Return all tests in a suite, recursing through nested suites"""
831
if isinstance(suite, unittest.TestCase):
833
elif isinstance(suite, unittest.TestSuite):
835
for r in iter_suite_tests(item):
838
raise Exception('unknown type %r for object %r'
839
% (type(suite), suite))
842
TestSkipped = testtools.testcase.TestSkipped
845
class TestNotApplicable(TestSkipped):
846
"""A test is not applicable to the situation where it was run.
848
This is only normally raised by parameterized tests, if they find that
849
the instance they're constructed upon does not support one aspect
854
# traceback._some_str fails to format exceptions that have the default
855
# __str__ which does an implicit ascii conversion. However, repr() on those
856
# objects works, for all that its not quite what the doctor may have ordered.
857
def _clever_some_str(value):
862
return repr(value).replace('\\n', '\n')
864
return '<unprintable %s object>' % type(value).__name__
866
traceback._some_str = _clever_some_str
869
# deprecated - use self.knownFailure(), or self.expectFailure.
870
KnownFailure = testtools.testcase._ExpectedFailure
873
class UnavailableFeature(Exception):
874
"""A feature required for this test was not available.
876
This can be considered a specialised form of SkippedTest.
878
The feature should be used to construct the exception.
882
class StringIOWrapper(object):
883
"""A wrapper around cStringIO which just adds an encoding attribute.
885
Internally we can check sys.stdout to see what the output encoding
886
should be. However, cStringIO has no encoding attribute that we can
887
set. So we wrap it instead.
892
def __init__(self, s=None):
894
self.__dict__['_cstring'] = StringIO(s)
896
self.__dict__['_cstring'] = StringIO()
898
def __getattr__(self, name, getattr=getattr):
899
return getattr(self.__dict__['_cstring'], name)
901
def __setattr__(self, name, val):
902
if name == 'encoding':
903
self.__dict__['encoding'] = val
905
return setattr(self._cstring, name, val)
908
class TestUIFactory(TextUIFactory):
909
"""A UI Factory for testing.
911
Hide the progress bar but emit note()s.
913
Allows get_password to be tested without real tty attached.
915
See also CannedInputUIFactory which lets you provide programmatic input in
918
# TODO: Capture progress events at the model level and allow them to be
919
# observed by tests that care.
921
# XXX: Should probably unify more with CannedInputUIFactory or a
922
# particular configuration of TextUIFactory, or otherwise have a clearer
923
# idea of how they're supposed to be different.
924
# See https://bugs.launchpad.net/bzr/+bug/408213
926
def __init__(self, stdout=None, stderr=None, stdin=None):
927
if stdin is not None:
928
# We use a StringIOWrapper to be able to test various
929
# encodings, but the user is still responsible to
930
# encode the string and to set the encoding attribute
931
# of StringIOWrapper.
932
stdin = StringIOWrapper(stdin)
933
super(TestUIFactory, self).__init__(stdin, stdout, stderr)
935
def get_non_echoed_password(self):
936
"""Get password from stdin without trying to handle the echo mode"""
937
password = self.stdin.readline()
940
if password[-1] == '\n':
941
password = password[:-1]
944
def make_progress_view(self):
945
return NullProgressView()
948
def isolated_doctest_setUp(test):
949
override_os_environ(test)
952
def isolated_doctest_tearDown(test):
953
restore_os_environ(test)
956
def IsolatedDocTestSuite(*args, **kwargs):
957
"""Overrides doctest.DocTestSuite to handle isolation.
959
The method is really a factory and users are expected to use it as such.
962
kwargs['setUp'] = isolated_doctest_setUp
963
kwargs['tearDown'] = isolated_doctest_tearDown
964
return doctest.DocTestSuite(*args, **kwargs)
967
class TestCase(testtools.TestCase):
968
"""Base class for bzr unit tests.
970
Tests that need access to disk resources should subclass
971
TestCaseInTempDir not TestCase.
973
Error and debug log messages are redirected from their usual
974
location into a temporary file, the contents of which can be
975
retrieved by _get_log(). We use a real OS file, not an in-memory object,
976
so that it can also capture file IO. When the test completes this file
977
is read into memory and removed from disk.
979
There are also convenience functions to invoke bzr's command-line
980
routine, and to build and check bzr trees.
982
In addition to the usual method of overriding tearDown(), this class also
983
allows subclasses to register cleanup functions via addCleanup, which are
984
run in order as the object is torn down. It's less likely this will be
985
accidentally overlooked.
989
# record lsprof data when performing benchmark calls.
990
_gather_lsprof_in_benchmarks = False
992
def __init__(self, methodName='testMethod'):
993
super(TestCase, self).__init__(methodName)
994
self._directory_isolation = True
995
self.exception_handlers.insert(0,
996
(UnavailableFeature, self._do_unsupported_or_skip))
997
self.exception_handlers.insert(0,
998
(TestNotApplicable, self._do_not_applicable))
1001
super(TestCase, self).setUp()
1003
timeout = config.GlobalStack().get('selftest.timeout')
1005
timeout_fixture = fixtures.TimeoutFixture(timeout)
1006
timeout_fixture.setUp()
1007
self.addCleanup(timeout_fixture.cleanUp)
1009
for feature in getattr(self, '_test_needs_features', []):
1010
self.requireFeature(feature)
1011
self._cleanEnvironment()
1013
if bzrlib.global_state is not None:
1014
self.overrideAttr(bzrlib.global_state, 'cmdline_overrides',
1015
config.CommandLineStore())
1018
self._startLogFile()
1019
self._benchcalls = []
1020
self._benchtime = None
1022
self._track_transports()
1024
self._clear_debug_flags()
1025
# Isolate global verbosity level, to make sure it's reproducible
1026
# between tests. We should get rid of this altogether: bug 656694. --
1028
self.overrideAttr(bzrlib.trace, '_verbosity_level', 0)
1029
self._log_files = set()
1030
# Each key in the ``_counters`` dict holds a value for a different
1031
# counter. When the test ends, addDetail() should be used to output the
1032
# counter values. This happens in install_counter_hook().
1034
if 'config_stats' in selftest_debug_flags:
1035
self._install_config_stats_hooks()
1036
# Do not use i18n for tests (unless the test reverses this)
1042
# The sys preserved stdin/stdout should allow blackbox tests debugging
1043
pdb.Pdb(stdin=sys.__stdin__, stdout=sys.__stdout__
1044
).set_trace(sys._getframe().f_back)
1046
def discardDetail(self, name):
1047
"""Extend the addDetail, getDetails api so we can remove a detail.
1049
eg. bzr always adds the 'log' detail at startup, but we don't want to
1050
include it for skipped, xfail, etc tests.
1052
It is safe to call this for a detail that doesn't exist, in case this
1053
gets called multiple times.
1055
# We cheat. details is stored in __details which means we shouldn't
1056
# touch it. but getDetails() returns the dict directly, so we can
1058
details = self.getDetails()
1062
def install_counter_hook(self, hooks, name, counter_name=None):
1063
"""Install a counting hook.
1065
Any hook can be counted as long as it doesn't need to return a value.
1067
:param hooks: Where the hook should be installed.
1069
:param name: The hook name that will be counted.
1071
:param counter_name: The counter identifier in ``_counters``, defaults
1074
_counters = self._counters # Avoid closing over self
1075
if counter_name is None:
1077
if _counters.has_key(counter_name):
1078
raise AssertionError('%s is already used as a counter name'
1080
_counters[counter_name] = 0
1081
self.addDetail(counter_name, content.Content(content.UTF8_TEXT,
1082
lambda: ['%d' % (_counters[counter_name],)]))
1083
def increment_counter(*args, **kwargs):
1084
_counters[counter_name] += 1
1085
label = 'count %s calls' % (counter_name,)
1086
hooks.install_named_hook(name, increment_counter, label)
1087
self.addCleanup(hooks.uninstall_named_hook, name, label)
1089
def _install_config_stats_hooks(self):
1090
"""Install config hooks to count hook calls.
1093
for hook_name in ('get', 'set', 'remove', 'load', 'save'):
1094
self.install_counter_hook(config.ConfigHooks, hook_name,
1095
'config.%s' % (hook_name,))
1097
# The OldConfigHooks are private and need special handling to protect
1098
# against recursive tests (tests that run other tests), so we just do
1099
# manually what registering them into _builtin_known_hooks will provide
1101
self.overrideAttr(config, 'OldConfigHooks', config._OldConfigHooks())
1102
for hook_name in ('get', 'set', 'remove', 'load', 'save'):
1103
self.install_counter_hook(config.OldConfigHooks, hook_name,
1104
'old_config.%s' % (hook_name,))
1106
def _clear_debug_flags(self):
1107
"""Prevent externally set debug flags affecting tests.
1109
Tests that want to use debug flags can just set them in the
1110
debug_flags set during setup/teardown.
1112
# Start with a copy of the current debug flags we can safely modify.
1113
self.overrideAttr(debug, 'debug_flags', set(debug.debug_flags))
1114
if 'allow_debug' not in selftest_debug_flags:
1115
debug.debug_flags.clear()
1116
if 'disable_lock_checks' not in selftest_debug_flags:
1117
debug.debug_flags.add('strict_locks')
1119
def _clear_hooks(self):
1120
# prevent hooks affecting tests
1121
known_hooks = hooks.known_hooks
1122
self._preserved_hooks = {}
1123
for key, (parent, name) in known_hooks.iter_parent_objects():
1124
current_hooks = getattr(parent, name)
1125
self._preserved_hooks[parent] = (name, current_hooks)
1126
self._preserved_lazy_hooks = hooks._lazy_hooks
1127
hooks._lazy_hooks = {}
1128
self.addCleanup(self._restoreHooks)
1129
for key, (parent, name) in known_hooks.iter_parent_objects():
1130
factory = known_hooks.get(key)
1131
setattr(parent, name, factory())
1132
# this hook should always be installed
1133
request._install_hook()
1135
def disable_directory_isolation(self):
1136
"""Turn off directory isolation checks."""
1137
self._directory_isolation = False
1139
def enable_directory_isolation(self):
1140
"""Enable directory isolation checks."""
1141
self._directory_isolation = True
1143
def _silenceUI(self):
1144
"""Turn off UI for duration of test"""
1145
# by default the UI is off; tests can turn it on if they want it.
1146
self.overrideAttr(ui, 'ui_factory', ui.SilentUIFactory())
1148
def _check_locks(self):
1149
"""Check that all lock take/release actions have been paired."""
1150
# We always check for mismatched locks. If a mismatch is found, we
1151
# fail unless -Edisable_lock_checks is supplied to selftest, in which
1152
# case we just print a warning.
1154
acquired_locks = [lock for action, lock in self._lock_actions
1155
if action == 'acquired']
1156
released_locks = [lock for action, lock in self._lock_actions
1157
if action == 'released']
1158
broken_locks = [lock for action, lock in self._lock_actions
1159
if action == 'broken']
1160
# trivially, given the tests for lock acquistion and release, if we
1161
# have as many in each list, it should be ok. Some lock tests also
1162
# break some locks on purpose and should be taken into account by
1163
# considering that breaking a lock is just a dirty way of releasing it.
1164
if len(acquired_locks) != (len(released_locks) + len(broken_locks)):
1166
'Different number of acquired and '
1167
'released or broken locks.\n'
1171
(acquired_locks, released_locks, broken_locks))
1172
if not self._lock_check_thorough:
1173
# Rather than fail, just warn
1174
print "Broken test %s: %s" % (self, message)
1178
def _track_locks(self):
1179
"""Track lock activity during tests."""
1180
self._lock_actions = []
1181
if 'disable_lock_checks' in selftest_debug_flags:
1182
self._lock_check_thorough = False
1184
self._lock_check_thorough = True
1186
self.addCleanup(self._check_locks)
1187
_mod_lock.Lock.hooks.install_named_hook('lock_acquired',
1188
self._lock_acquired, None)
1189
_mod_lock.Lock.hooks.install_named_hook('lock_released',
1190
self._lock_released, None)
1191
_mod_lock.Lock.hooks.install_named_hook('lock_broken',
1192
self._lock_broken, None)
1194
def _lock_acquired(self, result):
1195
self._lock_actions.append(('acquired', result))
1197
def _lock_released(self, result):
1198
self._lock_actions.append(('released', result))
1200
def _lock_broken(self, result):
1201
self._lock_actions.append(('broken', result))
1203
def permit_dir(self, name):
1204
"""Permit a directory to be used by this test. See permit_url."""
1205
name_transport = _mod_transport.get_transport_from_path(name)
1206
self.permit_url(name)
1207
self.permit_url(name_transport.base)
1209
def permit_url(self, url):
1210
"""Declare that url is an ok url to use in this test.
1212
Do this for memory transports, temporary test directory etc.
1214
Do not do this for the current working directory, /tmp, or any other
1215
preexisting non isolated url.
1217
if not url.endswith('/'):
1219
self._bzr_selftest_roots.append(url)
1221
def permit_source_tree_branch_repo(self):
1222
"""Permit the source tree bzr is running from to be opened.
1224
Some code such as bzrlib.version attempts to read from the bzr branch
1225
that bzr is executing from (if any). This method permits that directory
1226
to be used in the test suite.
1228
path = self.get_source_path()
1229
self.record_directory_isolation()
1232
workingtree.WorkingTree.open(path)
1233
except (errors.NotBranchError, errors.NoWorkingTree):
1234
raise TestSkipped('Needs a working tree of bzr sources')
1236
self.enable_directory_isolation()
1238
def _preopen_isolate_transport(self, transport):
1239
"""Check that all transport openings are done in the test work area."""
1240
while isinstance(transport, pathfilter.PathFilteringTransport):
1241
# Unwrap pathfiltered transports
1242
transport = transport.server.backing_transport.clone(
1243
transport._filter('.'))
1244
url = transport.base
1245
# ReadonlySmartTCPServer_for_testing decorates the backing transport
1246
# urls it is given by prepending readonly+. This is appropriate as the
1247
# client shouldn't know that the server is readonly (or not readonly).
1248
# We could register all servers twice, with readonly+ prepending, but
1249
# that makes for a long list; this is about the same but easier to
1251
if url.startswith('readonly+'):
1252
url = url[len('readonly+'):]
1253
self._preopen_isolate_url(url)
1255
def _preopen_isolate_url(self, url):
1256
if not self._directory_isolation:
1258
if self._directory_isolation == 'record':
1259
self._bzr_selftest_roots.append(url)
1261
# This prevents all transports, including e.g. sftp ones backed on disk
1262
# from working unless they are explicitly granted permission. We then
1263
# depend on the code that sets up test transports to check that they are
1264
# appropriately isolated and enable their use by calling
1265
# self.permit_transport()
1266
if not osutils.is_inside_any(self._bzr_selftest_roots, url):
1267
raise errors.BzrError("Attempt to escape test isolation: %r %r"
1268
% (url, self._bzr_selftest_roots))
1270
def record_directory_isolation(self):
1271
"""Gather accessed directories to permit later access.
1273
This is used for tests that access the branch bzr is running from.
1275
self._directory_isolation = "record"
1277
def start_server(self, transport_server, backing_server=None):
1278
"""Start transport_server for this test.
1280
This starts the server, registers a cleanup for it and permits the
1281
server's urls to be used.
1283
if backing_server is None:
1284
transport_server.start_server()
1286
transport_server.start_server(backing_server)
1287
self.addCleanup(transport_server.stop_server)
1288
# Obtain a real transport because if the server supplies a password, it
1289
# will be hidden from the base on the client side.
1290
t = _mod_transport.get_transport_from_url(transport_server.get_url())
1291
# Some transport servers effectively chroot the backing transport;
1292
# others like SFTPServer don't - users of the transport can walk up the
1293
# transport to read the entire backing transport. This wouldn't matter
1294
# except that the workdir tests are given - and that they expect the
1295
# server's url to point at - is one directory under the safety net. So
1296
# Branch operations into the transport will attempt to walk up one
1297
# directory. Chrooting all servers would avoid this but also mean that
1298
# we wouldn't be testing directly against non-root urls. Alternatively
1299
# getting the test framework to start the server with a backing server
1300
# at the actual safety net directory would work too, but this then
1301
# means that the self.get_url/self.get_transport methods would need
1302
# to transform all their results. On balance its cleaner to handle it
1303
# here, and permit a higher url when we have one of these transports.
1304
if t.base.endswith('/work/'):
1305
# we have safety net/test root/work
1306
t = t.clone('../..')
1307
elif isinstance(transport_server,
1308
test_server.SmartTCPServer_for_testing):
1309
# The smart server adds a path similar to work, which is traversed
1310
# up from by the client. But the server is chrooted - the actual
1311
# backing transport is not escaped from, and VFS requests to the
1312
# root will error (because they try to escape the chroot).
1314
while t2.base != t.base:
1317
self.permit_url(t.base)
1319
def _track_transports(self):
1320
"""Install checks for transport usage."""
1321
# TestCase has no safe place it can write to.
1322
self._bzr_selftest_roots = []
1323
# Currently the easiest way to be sure that nothing is going on is to
1324
# hook into bzr dir opening. This leaves a small window of error for
1325
# transport tests, but they are well known, and we can improve on this
1327
controldir.ControlDir.hooks.install_named_hook("pre_open",
1328
self._preopen_isolate_transport, "Check bzr directories are safe.")
1330
def _ndiff_strings(self, a, b):
1331
"""Return ndiff between two strings containing lines.
1333
A trailing newline is added if missing to make the strings
1335
if b and b[-1] != '\n':
1337
if a and a[-1] != '\n':
1339
difflines = difflib.ndiff(a.splitlines(True),
1341
linejunk=lambda x: False,
1342
charjunk=lambda x: False)
1343
return ''.join(difflines)
1345
def assertEqual(self, a, b, message=''):
1349
except UnicodeError, e:
1350
# If we can't compare without getting a UnicodeError, then
1351
# obviously they are different
1352
trace.mutter('UnicodeError: %s', e)
1355
raise AssertionError("%snot equal:\na = %s\nb = %s\n"
1357
pprint.pformat(a), pprint.pformat(b)))
1359
assertEquals = assertEqual
1361
def assertEqualDiff(self, a, b, message=None):
1362
"""Assert two texts are equal, if not raise an exception.
1364
This is intended for use with multi-line strings where it can
1365
be hard to find the differences by eye.
1367
# TODO: perhaps override assertEquals to call this for strings?
1371
message = "texts not equal:\n"
1373
message = 'first string is missing a final newline.\n'
1375
message = 'second string is missing a final newline.\n'
1376
raise AssertionError(message +
1377
self._ndiff_strings(a, b))
1379
def assertEqualMode(self, mode, mode_test):
1380
self.assertEqual(mode, mode_test,
1381
'mode mismatch %o != %o' % (mode, mode_test))
1383
def assertEqualStat(self, expected, actual):
1384
"""assert that expected and actual are the same stat result.
1386
:param expected: A stat result.
1387
:param actual: A stat result.
1388
:raises AssertionError: If the expected and actual stat values differ
1389
other than by atime.
1391
self.assertEqual(expected.st_size, actual.st_size,
1392
'st_size did not match')
1393
self.assertEqual(expected.st_mtime, actual.st_mtime,
1394
'st_mtime did not match')
1395
self.assertEqual(expected.st_ctime, actual.st_ctime,
1396
'st_ctime did not match')
1397
if sys.platform == 'win32':
1398
# On Win32 both 'dev' and 'ino' cannot be trusted. In python2.4 it
1399
# is 'dev' that varies, in python 2.5 (6?) it is st_ino that is
1400
# odd. We just force it to always be 0 to avoid any problems.
1401
self.assertEqual(0, expected.st_dev)
1402
self.assertEqual(0, actual.st_dev)
1403
self.assertEqual(0, expected.st_ino)
1404
self.assertEqual(0, actual.st_ino)
1406
self.assertEqual(expected.st_dev, actual.st_dev,
1407
'st_dev did not match')
1408
self.assertEqual(expected.st_ino, actual.st_ino,
1409
'st_ino did not match')
1410
self.assertEqual(expected.st_mode, actual.st_mode,
1411
'st_mode did not match')
1413
def assertLength(self, length, obj_with_len):
1414
"""Assert that obj_with_len is of length length."""
1415
if len(obj_with_len) != length:
1416
self.fail("Incorrect length: wanted %d, got %d for %r" % (
1417
length, len(obj_with_len), obj_with_len))
1419
def assertLogsError(self, exception_class, func, *args, **kwargs):
1420
"""Assert that `func(*args, **kwargs)` quietly logs a specific error.
1423
orig_log_exception_quietly = trace.log_exception_quietly
1426
orig_log_exception_quietly()
1427
captured.append(sys.exc_info()[1])
1428
trace.log_exception_quietly = capture
1429
func(*args, **kwargs)
1431
trace.log_exception_quietly = orig_log_exception_quietly
1432
self.assertLength(1, captured)
1434
self.assertIsInstance(err, exception_class)
1437
def assertPositive(self, val):
1438
"""Assert that val is greater than 0."""
1439
self.assertTrue(val > 0, 'expected a positive value, but got %s' % val)
1441
def assertNegative(self, val):
1442
"""Assert that val is less than 0."""
1443
self.assertTrue(val < 0, 'expected a negative value, but got %s' % val)
1445
def assertStartsWith(self, s, prefix):
1446
if not s.startswith(prefix):
1447
raise AssertionError('string %r does not start with %r' % (s, prefix))
1449
def assertEndsWith(self, s, suffix):
1450
"""Asserts that s ends with suffix."""
1451
if not s.endswith(suffix):
1452
raise AssertionError('string %r does not end with %r' % (s, suffix))
1454
def assertContainsRe(self, haystack, needle_re, flags=0):
1455
"""Assert that a contains something matching a regular expression."""
1456
if not re.search(needle_re, haystack, flags):
1457
if '\n' in haystack or len(haystack) > 60:
1458
# a long string, format it in a more readable way
1459
raise AssertionError(
1460
'pattern "%s" not found in\n"""\\\n%s"""\n'
1461
% (needle_re, haystack))
1463
raise AssertionError('pattern "%s" not found in "%s"'
1464
% (needle_re, haystack))
1466
def assertNotContainsRe(self, haystack, needle_re, flags=0):
1467
"""Assert that a does not match a regular expression"""
1468
if re.search(needle_re, haystack, flags):
1469
raise AssertionError('pattern "%s" found in "%s"'
1470
% (needle_re, haystack))
1472
def assertContainsString(self, haystack, needle):
1473
if haystack.find(needle) == -1:
1474
self.fail("string %r not found in '''%s'''" % (needle, haystack))
1476
def assertNotContainsString(self, haystack, needle):
1477
if haystack.find(needle) != -1:
1478
self.fail("string %r found in '''%s'''" % (needle, haystack))
1480
def assertSubset(self, sublist, superlist):
1481
"""Assert that every entry in sublist is present in superlist."""
1482
missing = set(sublist) - set(superlist)
1483
if len(missing) > 0:
1484
raise AssertionError("value(s) %r not present in container %r" %
1485
(missing, superlist))
1487
def assertListRaises(self, excClass, func, *args, **kwargs):
1488
"""Fail unless excClass is raised when the iterator from func is used.
1490
Many functions can return generators this makes sure
1491
to wrap them in a list() call to make sure the whole generator
1492
is run, and that the proper exception is raised.
1495
list(func(*args, **kwargs))
1499
if getattr(excClass,'__name__', None) is not None:
1500
excName = excClass.__name__
1502
excName = str(excClass)
1503
raise self.failureException, "%s not raised" % excName
1505
def assertRaises(self, excClass, callableObj, *args, **kwargs):
1506
"""Assert that a callable raises a particular exception.
1508
:param excClass: As for the except statement, this may be either an
1509
exception class, or a tuple of classes.
1510
:param callableObj: A callable, will be passed ``*args`` and
1513
Returns the exception so that you can examine it.
1516
callableObj(*args, **kwargs)
1520
if getattr(excClass,'__name__', None) is not None:
1521
excName = excClass.__name__
1524
excName = str(excClass)
1525
raise self.failureException, "%s not raised" % excName
1527
def assertIs(self, left, right, message=None):
1528
if not (left is right):
1529
if message is not None:
1530
raise AssertionError(message)
1532
raise AssertionError("%r is not %r." % (left, right))
1534
def assertIsNot(self, left, right, message=None):
1536
if message is not None:
1537
raise AssertionError(message)
1539
raise AssertionError("%r is %r." % (left, right))
1541
def assertTransportMode(self, transport, path, mode):
1542
"""Fail if a path does not have mode "mode".
1544
If modes are not supported on this transport, the assertion is ignored.
1546
if not transport._can_roundtrip_unix_modebits():
1548
path_stat = transport.stat(path)
1549
actual_mode = stat.S_IMODE(path_stat.st_mode)
1550
self.assertEqual(mode, actual_mode,
1551
'mode of %r incorrect (%s != %s)'
1552
% (path, oct(mode), oct(actual_mode)))
1554
def assertIsSameRealPath(self, path1, path2):
1555
"""Fail if path1 and path2 points to different files"""
1556
self.assertEqual(osutils.realpath(path1),
1557
osutils.realpath(path2),
1558
"apparent paths:\na = %s\nb = %s\n," % (path1, path2))
1560
def assertIsInstance(self, obj, kls, msg=None):
1561
"""Fail if obj is not an instance of kls
1563
:param msg: Supplementary message to show if the assertion fails.
1565
if not isinstance(obj, kls):
1566
m = "%r is an instance of %s rather than %s" % (
1567
obj, obj.__class__, kls)
1572
def assertFileEqual(self, content, path):
1573
"""Fail if path does not contain 'content'."""
1574
self.assertPathExists(path)
1575
f = file(path, 'rb')
1580
self.assertEqualDiff(content, s)
1582
def assertDocstring(self, expected_docstring, obj):
1583
"""Fail if obj does not have expected_docstring"""
1585
# With -OO the docstring should be None instead
1586
self.assertIs(obj.__doc__, None)
1588
self.assertEqual(expected_docstring, obj.__doc__)
1590
@symbol_versioning.deprecated_method(symbol_versioning.deprecated_in((2, 4)))
1591
def failUnlessExists(self, path):
1592
return self.assertPathExists(path)
1594
def assertPathExists(self, path):
1595
"""Fail unless path or paths, which may be abs or relative, exist."""
1596
if not isinstance(path, basestring):
1598
self.assertPathExists(p)
1600
self.assertTrue(osutils.lexists(path),
1601
path + " does not exist")
1603
@symbol_versioning.deprecated_method(symbol_versioning.deprecated_in((2, 4)))
1604
def failIfExists(self, path):
1605
return self.assertPathDoesNotExist(path)
1607
def assertPathDoesNotExist(self, path):
1608
"""Fail if path or paths, which may be abs or relative, exist."""
1609
if not isinstance(path, basestring):
1611
self.assertPathDoesNotExist(p)
1613
self.assertFalse(osutils.lexists(path),
1616
def _capture_deprecation_warnings(self, a_callable, *args, **kwargs):
1617
"""A helper for callDeprecated and applyDeprecated.
1619
:param a_callable: A callable to call.
1620
:param args: The positional arguments for the callable
1621
:param kwargs: The keyword arguments for the callable
1622
:return: A tuple (warnings, result). result is the result of calling
1623
a_callable(``*args``, ``**kwargs``).
1626
def capture_warnings(msg, cls=None, stacklevel=None):
1627
# we've hooked into a deprecation specific callpath,
1628
# only deprecations should getting sent via it.
1629
self.assertEqual(cls, DeprecationWarning)
1630
local_warnings.append(msg)
1631
original_warning_method = symbol_versioning.warn
1632
symbol_versioning.set_warning_method(capture_warnings)
1634
result = a_callable(*args, **kwargs)
1636
symbol_versioning.set_warning_method(original_warning_method)
1637
return (local_warnings, result)
1639
def applyDeprecated(self, deprecation_format, a_callable, *args, **kwargs):
1640
"""Call a deprecated callable without warning the user.
1642
Note that this only captures warnings raised by symbol_versioning.warn,
1643
not other callers that go direct to the warning module.
1645
To test that a deprecated method raises an error, do something like
1646
this (remember that both assertRaises and applyDeprecated delays *args
1647
and **kwargs passing)::
1649
self.assertRaises(errors.ReservedId,
1650
self.applyDeprecated,
1651
deprecated_in((1, 5, 0)),
1655
:param deprecation_format: The deprecation format that the callable
1656
should have been deprecated with. This is the same type as the
1657
parameter to deprecated_method/deprecated_function. If the
1658
callable is not deprecated with this format, an assertion error
1660
:param a_callable: A callable to call. This may be a bound method or
1661
a regular function. It will be called with ``*args`` and
1663
:param args: The positional arguments for the callable
1664
:param kwargs: The keyword arguments for the callable
1665
:return: The result of a_callable(``*args``, ``**kwargs``)
1667
call_warnings, result = self._capture_deprecation_warnings(a_callable,
1669
expected_first_warning = symbol_versioning.deprecation_string(
1670
a_callable, deprecation_format)
1671
if len(call_warnings) == 0:
1672
self.fail("No deprecation warning generated by call to %s" %
1674
self.assertEqual(expected_first_warning, call_warnings[0])
1677
def callCatchWarnings(self, fn, *args, **kw):
1678
"""Call a callable that raises python warnings.
1680
The caller's responsible for examining the returned warnings.
1682
If the callable raises an exception, the exception is not
1683
caught and propagates up to the caller. In that case, the list
1684
of warnings is not available.
1686
:returns: ([warning_object, ...], fn_result)
1688
# XXX: This is not perfect, because it completely overrides the
1689
# warnings filters, and some code may depend on suppressing particular
1690
# warnings. It's the easiest way to insulate ourselves from -Werror,
1691
# though. -- Andrew, 20071062
1693
def _catcher(message, category, filename, lineno, file=None, line=None):
1694
# despite the name, 'message' is normally(?) a Warning subclass
1696
wlist.append(message)
1697
saved_showwarning = warnings.showwarning
1698
saved_filters = warnings.filters
1700
warnings.showwarning = _catcher
1701
warnings.filters = []
1702
result = fn(*args, **kw)
1704
warnings.showwarning = saved_showwarning
1705
warnings.filters = saved_filters
1706
return wlist, result
1708
def callDeprecated(self, expected, callable, *args, **kwargs):
1709
"""Assert that a callable is deprecated in a particular way.
1711
This is a very precise test for unusual requirements. The
1712
applyDeprecated helper function is probably more suited for most tests
1713
as it allows you to simply specify the deprecation format being used
1714
and will ensure that that is issued for the function being called.
1716
Note that this only captures warnings raised by symbol_versioning.warn,
1717
not other callers that go direct to the warning module. To catch
1718
general warnings, use callCatchWarnings.
1720
:param expected: a list of the deprecation warnings expected, in order
1721
:param callable: The callable to call
1722
:param args: The positional arguments for the callable
1723
:param kwargs: The keyword arguments for the callable
1725
call_warnings, result = self._capture_deprecation_warnings(callable,
1727
self.assertEqual(expected, call_warnings)
1730
def _startLogFile(self):
1731
"""Setup a in-memory target for bzr and testcase log messages"""
1732
pseudo_log_file = StringIO()
1733
def _get_log_contents_for_weird_testtools_api():
1734
return [pseudo_log_file.getvalue().decode(
1735
"utf-8", "replace").encode("utf-8")]
1736
self.addDetail("log", content.Content(content.ContentType("text",
1737
"plain", {"charset": "utf8"}),
1738
_get_log_contents_for_weird_testtools_api))
1739
self._log_file = pseudo_log_file
1740
self._log_memento = trace.push_log_file(self._log_file)
1741
self.addCleanup(self._finishLogFile)
1743
def _finishLogFile(self):
1744
"""Flush and dereference the in-memory log for this testcase"""
1745
if trace._trace_file:
1746
# flush the log file, to get all content
1747
trace._trace_file.flush()
1748
trace.pop_log_file(self._log_memento)
1749
# The logging module now tracks references for cleanup so discard ours
1750
del self._log_memento
1752
def thisFailsStrictLockCheck(self):
1753
"""It is known that this test would fail with -Dstrict_locks.
1755
By default, all tests are run with strict lock checking unless
1756
-Edisable_lock_checks is supplied. However there are some tests which
1757
we know fail strict locks at this point that have not been fixed.
1758
They should call this function to disable the strict checking.
1760
This should be used sparingly, it is much better to fix the locking
1761
issues rather than papering over the problem by calling this function.
1763
debug.debug_flags.discard('strict_locks')
1765
def overrideAttr(self, obj, attr_name, new=_unitialized_attr):
1766
"""Overrides an object attribute restoring it after the test.
1768
:note: This should be used with discretion; you should think about
1769
whether it's better to make the code testable without monkey-patching.
1771
:param obj: The object that will be mutated.
1773
:param attr_name: The attribute name we want to preserve/override in
1776
:param new: The optional value we want to set the attribute to.
1778
:returns: The actual attr value.
1780
value = getattr(obj, attr_name)
1781
# The actual value is captured by the call below
1782
self.addCleanup(setattr, obj, attr_name, value)
1783
if new is not _unitialized_attr:
1784
setattr(obj, attr_name, new)
1787
def overrideEnv(self, name, new):
1788
"""Set an environment variable, and reset it after the test.
1790
:param name: The environment variable name.
1792
:param new: The value to set the variable to. If None, the
1793
variable is deleted from the environment.
1795
:returns: The actual variable value.
1797
value = osutils.set_or_unset_env(name, new)
1798
self.addCleanup(osutils.set_or_unset_env, name, value)
1801
def recordCalls(self, obj, attr_name):
1802
"""Monkeypatch in a wrapper that will record calls.
1804
The monkeypatch is automatically removed when the test concludes.
1806
:param obj: The namespace holding the reference to be replaced;
1807
typically a module, class, or object.
1808
:param attr_name: A string for the name of the attribute to
1810
:returns: A list that will be extended with one item every time the
1811
function is called, with a tuple of (args, kwargs).
1815
def decorator(*args, **kwargs):
1816
calls.append((args, kwargs))
1817
return orig(*args, **kwargs)
1818
orig = self.overrideAttr(obj, attr_name, decorator)
1821
def _cleanEnvironment(self):
1822
for name, value in isolated_environ.iteritems():
1823
self.overrideEnv(name, value)
1825
def _restoreHooks(self):
1826
for klass, (name, hooks) in self._preserved_hooks.items():
1827
setattr(klass, name, hooks)
1828
self._preserved_hooks.clear()
1829
bzrlib.hooks._lazy_hooks = self._preserved_lazy_hooks
1830
self._preserved_lazy_hooks.clear()
1832
def knownFailure(self, reason):
1833
"""Declare that this test fails for a known reason
1835
Tests that are known to fail should generally be using expectedFailure
1836
with an appropriate reverse assertion if a change could cause the test
1837
to start passing. Conversely if the test has no immediate prospect of
1838
succeeding then using skip is more suitable.
1840
When this method is called while an exception is being handled, that
1841
traceback will be used, otherwise a new exception will be thrown to
1842
provide one but won't be reported.
1844
self._add_reason(reason)
1846
exc_info = sys.exc_info()
1847
if exc_info != (None, None, None):
1848
self._report_traceback(exc_info)
1851
raise self.failureException(reason)
1852
except self.failureException:
1853
exc_info = sys.exc_info()
1854
# GZ 02-08-2011: Maybe cleanup this err.exc_info attribute too?
1855
raise testtools.testcase._ExpectedFailure(exc_info)
1859
def _suppress_log(self):
1860
"""Remove the log info from details."""
1861
self.discardDetail('log')
1863
def _do_skip(self, result, reason):
1864
self._suppress_log()
1865
addSkip = getattr(result, 'addSkip', None)
1866
if not callable(addSkip):
1867
result.addSuccess(result)
1869
addSkip(self, reason)
1872
def _do_known_failure(self, result, e):
1873
self._suppress_log()
1874
err = sys.exc_info()
1875
addExpectedFailure = getattr(result, 'addExpectedFailure', None)
1876
if addExpectedFailure is not None:
1877
addExpectedFailure(self, err)
1879
result.addSuccess(self)
1882
def _do_not_applicable(self, result, e):
1884
reason = 'No reason given'
1887
self._suppress_log ()
1888
addNotApplicable = getattr(result, 'addNotApplicable', None)
1889
if addNotApplicable is not None:
1890
result.addNotApplicable(self, reason)
1892
self._do_skip(result, reason)
1895
def _report_skip(self, result, err):
1896
"""Override the default _report_skip.
1898
We want to strip the 'log' detail. If we waint until _do_skip, it has
1899
already been formatted into the 'reason' string, and we can't pull it
1902
self._suppress_log()
1903
super(TestCase, self)._report_skip(self, result, err)
1906
def _report_expected_failure(self, result, err):
1909
See _report_skip for motivation.
1911
self._suppress_log()
1912
super(TestCase, self)._report_expected_failure(self, result, err)
1915
def _do_unsupported_or_skip(self, result, e):
1917
self._suppress_log()
1918
addNotSupported = getattr(result, 'addNotSupported', None)
1919
if addNotSupported is not None:
1920
result.addNotSupported(self, reason)
1922
self._do_skip(result, reason)
1924
def time(self, callable, *args, **kwargs):
1925
"""Run callable and accrue the time it takes to the benchmark time.
1927
If lsprofiling is enabled (i.e. by --lsprof-time to bzr selftest) then
1928
this will cause lsprofile statistics to be gathered and stored in
1931
if self._benchtime is None:
1932
self.addDetail('benchtime', content.Content(content.ContentType(
1933
"text", "plain"), lambda:[str(self._benchtime)]))
1937
if not self._gather_lsprof_in_benchmarks:
1938
return callable(*args, **kwargs)
1940
# record this benchmark
1941
ret, stats = bzrlib.lsprof.profile(callable, *args, **kwargs)
1943
self._benchcalls.append(((callable, args, kwargs), stats))
1946
self._benchtime += time.time() - start
1948
def log(self, *args):
1952
"""Get a unicode string containing the log from bzrlib.trace.
1954
Undecodable characters are replaced.
1956
return u"".join(self.getDetails()['log'].iter_text())
1958
def requireFeature(self, feature):
1959
"""This test requires a specific feature is available.
1961
:raises UnavailableFeature: When feature is not available.
1963
if not feature.available():
1964
raise UnavailableFeature(feature)
1966
def _run_bzr_autosplit(self, args, retcode, encoding, stdin,
1968
"""Run bazaar command line, splitting up a string command line."""
1969
if isinstance(args, basestring):
1970
# shlex don't understand unicode strings,
1971
# so args should be plain string (bialix 20070906)
1972
args = list(shlex.split(str(args)))
1973
return self._run_bzr_core(args, retcode=retcode,
1974
encoding=encoding, stdin=stdin, working_dir=working_dir,
1977
def _run_bzr_core(self, args, retcode, encoding, stdin,
1979
# Clear chk_map page cache, because the contents are likely to mask
1981
chk_map.clear_cache()
1982
if encoding is None:
1983
encoding = osutils.get_user_encoding()
1984
stdout = StringIOWrapper()
1985
stderr = StringIOWrapper()
1986
stdout.encoding = encoding
1987
stderr.encoding = encoding
1989
self.log('run bzr: %r', args)
1990
# FIXME: don't call into logging here
1991
handler = trace.EncodedStreamHandler(stderr, errors="replace",
1993
logger = logging.getLogger('')
1994
logger.addHandler(handler)
1995
old_ui_factory = ui.ui_factory
1996
ui.ui_factory = TestUIFactory(stdin=stdin, stdout=stdout, stderr=stderr)
1999
if working_dir is not None:
2000
cwd = osutils.getcwd()
2001
os.chdir(working_dir)
2005
result = self.apply_redirected(
2006
ui.ui_factory.stdin,
2008
_mod_commands.run_bzr_catch_user_errors,
2010
except KeyboardInterrupt:
2011
# Reraise KeyboardInterrupt with contents of redirected stdout
2012
# and stderr as arguments, for tests which are interested in
2013
# stdout and stderr and are expecting the exception.
2014
out = stdout.getvalue()
2015
err = stderr.getvalue()
2017
self.log('output:\n%r', out)
2019
self.log('errors:\n%r', err)
2020
raise KeyboardInterrupt(out, err)
2022
logger.removeHandler(handler)
2023
ui.ui_factory = old_ui_factory
2027
out = stdout.getvalue()
2028
err = stderr.getvalue()
2030
self.log('output:\n%r', out)
2032
self.log('errors:\n%r', err)
2033
if retcode is not None:
2034
self.assertEquals(retcode, result,
2035
message='Unexpected return code')
2036
return result, out, err
2038
def run_bzr(self, args, retcode=0, encoding=None, stdin=None,
2039
working_dir=None, error_regexes=[], output_encoding=None):
2040
"""Invoke bzr, as if it were run from the command line.
2042
The argument list should not include the bzr program name - the
2043
first argument is normally the bzr command. Arguments may be
2044
passed in three ways:
2046
1- A list of strings, eg ["commit", "a"]. This is recommended
2047
when the command contains whitespace or metacharacters, or
2048
is built up at run time.
2050
2- A single string, eg "add a". This is the most convenient
2051
for hardcoded commands.
2053
This runs bzr through the interface that catches and reports
2054
errors, and with logging set to something approximating the
2055
default, so that error reporting can be checked.
2057
This should be the main method for tests that want to exercise the
2058
overall behavior of the bzr application (rather than a unit test
2059
or a functional test of the library.)
2061
This sends the stdout/stderr results into the test's log,
2062
where it may be useful for debugging. See also run_captured.
2064
:keyword stdin: A string to be used as stdin for the command.
2065
:keyword retcode: The status code the command should return;
2067
:keyword working_dir: The directory to run the command in
2068
:keyword error_regexes: A list of expected error messages. If
2069
specified they must be seen in the error output of the command.
2071
retcode, out, err = self._run_bzr_autosplit(
2076
working_dir=working_dir,
2078
self.assertIsInstance(error_regexes, (list, tuple))
2079
for regex in error_regexes:
2080
self.assertContainsRe(err, regex)
2083
def run_bzr_error(self, error_regexes, *args, **kwargs):
2084
"""Run bzr, and check that stderr contains the supplied regexes
2086
:param error_regexes: Sequence of regular expressions which
2087
must each be found in the error output. The relative ordering
2089
:param args: command-line arguments for bzr
2090
:param kwargs: Keyword arguments which are interpreted by run_bzr
2091
This function changes the default value of retcode to be 3,
2092
since in most cases this is run when you expect bzr to fail.
2094
:return: (out, err) The actual output of running the command (in case
2095
you want to do more inspection)
2099
# Make sure that commit is failing because there is nothing to do
2100
self.run_bzr_error(['no changes to commit'],
2101
['commit', '-m', 'my commit comment'])
2102
# Make sure --strict is handling an unknown file, rather than
2103
# giving us the 'nothing to do' error
2104
self.build_tree(['unknown'])
2105
self.run_bzr_error(['Commit refused because there are unknown files'],
2106
['commit', --strict', '-m', 'my commit comment'])
2108
kwargs.setdefault('retcode', 3)
2109
kwargs['error_regexes'] = error_regexes
2110
out, err = self.run_bzr(*args, **kwargs)
2113
def run_bzr_subprocess(self, *args, **kwargs):
2114
"""Run bzr in a subprocess for testing.
2116
This starts a new Python interpreter and runs bzr in there.
2117
This should only be used for tests that have a justifiable need for
2118
this isolation: e.g. they are testing startup time, or signal
2119
handling, or early startup code, etc. Subprocess code can't be
2120
profiled or debugged so easily.
2122
:keyword retcode: The status code that is expected. Defaults to 0. If
2123
None is supplied, the status code is not checked.
2124
:keyword env_changes: A dictionary which lists changes to environment
2125
variables. A value of None will unset the env variable.
2126
The values must be strings. The change will only occur in the
2127
child, so you don't need to fix the environment after running.
2128
:keyword universal_newlines: Convert CRLF => LF
2129
:keyword allow_plugins: By default the subprocess is run with
2130
--no-plugins to ensure test reproducibility. Also, it is possible
2131
for system-wide plugins to create unexpected output on stderr,
2132
which can cause unnecessary test failures.
2134
env_changes = kwargs.get('env_changes', {})
2135
working_dir = kwargs.get('working_dir', None)
2136
allow_plugins = kwargs.get('allow_plugins', False)
2138
if isinstance(args[0], list):
2140
elif isinstance(args[0], basestring):
2141
args = list(shlex.split(args[0]))
2143
raise ValueError("passing varargs to run_bzr_subprocess")
2144
process = self.start_bzr_subprocess(args, env_changes=env_changes,
2145
working_dir=working_dir,
2146
allow_plugins=allow_plugins)
2147
# We distinguish between retcode=None and retcode not passed.
2148
supplied_retcode = kwargs.get('retcode', 0)
2149
return self.finish_bzr_subprocess(process, retcode=supplied_retcode,
2150
universal_newlines=kwargs.get('universal_newlines', False),
2153
def start_bzr_subprocess(self, process_args, env_changes=None,
2154
skip_if_plan_to_signal=False,
2156
allow_plugins=False, stderr=subprocess.PIPE):
2157
"""Start bzr in a subprocess for testing.
2159
This starts a new Python interpreter and runs bzr in there.
2160
This should only be used for tests that have a justifiable need for
2161
this isolation: e.g. they are testing startup time, or signal
2162
handling, or early startup code, etc. Subprocess code can't be
2163
profiled or debugged so easily.
2165
:param process_args: a list of arguments to pass to the bzr executable,
2166
for example ``['--version']``.
2167
:param env_changes: A dictionary which lists changes to environment
2168
variables. A value of None will unset the env variable.
2169
The values must be strings. The change will only occur in the
2170
child, so you don't need to fix the environment after running.
2171
:param skip_if_plan_to_signal: raise TestSkipped when true and system
2172
doesn't support signalling subprocesses.
2173
:param allow_plugins: If False (default) pass --no-plugins to bzr.
2174
:param stderr: file to use for the subprocess's stderr. Valid values
2175
are those valid for the stderr argument of `subprocess.Popen`.
2176
Default value is ``subprocess.PIPE``.
2178
:returns: Popen object for the started process.
2180
if skip_if_plan_to_signal:
2181
if os.name != "posix":
2182
raise TestSkipped("Sending signals not supported")
2184
if env_changes is None:
2186
# Because $HOME is set to a tempdir for the context of a test, modules
2187
# installed in the user dir will not be found unless $PYTHONUSERBASE
2188
# gets set to the computed directory of this parent process.
2189
if site.USER_BASE is not None:
2190
env_changes["PYTHONUSERBASE"] = site.USER_BASE
2193
def cleanup_environment():
2194
for env_var, value in env_changes.iteritems():
2195
old_env[env_var] = osutils.set_or_unset_env(env_var, value)
2197
def restore_environment():
2198
for env_var, value in old_env.iteritems():
2199
osutils.set_or_unset_env(env_var, value)
2201
bzr_path = self.get_bzr_path()
2204
if working_dir is not None:
2205
cwd = osutils.getcwd()
2206
os.chdir(working_dir)
2209
# win32 subprocess doesn't support preexec_fn
2210
# so we will avoid using it on all platforms, just to
2211
# make sure the code path is used, and we don't break on win32
2212
cleanup_environment()
2213
# Include the subprocess's log file in the test details, in case
2214
# the test fails due to an error in the subprocess.
2215
self._add_subprocess_log(trace._get_bzr_log_filename())
2216
command = [sys.executable]
2217
# frozen executables don't need the path to bzr
2218
if getattr(sys, "frozen", None) is None:
2219
command.append(bzr_path)
2220
if not allow_plugins:
2221
command.append('--no-plugins')
2222
command.extend(process_args)
2223
process = self._popen(command, stdin=subprocess.PIPE,
2224
stdout=subprocess.PIPE,
2227
restore_environment()
2233
def _add_subprocess_log(self, log_file_path):
2234
if len(self._log_files) == 0:
2235
# Register an addCleanup func. We do this on the first call to
2236
# _add_subprocess_log rather than in TestCase.setUp so that this
2237
# addCleanup is registered after any cleanups for tempdirs that
2238
# subclasses might create, which will probably remove the log file
2240
self.addCleanup(self._subprocess_log_cleanup)
2241
# self._log_files is a set, so if a log file is reused we won't grab it
2243
self._log_files.add(log_file_path)
2245
def _subprocess_log_cleanup(self):
2246
for count, log_file_path in enumerate(self._log_files):
2247
# We use buffer_now=True to avoid holding the file open beyond
2248
# the life of this function, which might interfere with e.g.
2249
# cleaning tempdirs on Windows.
2250
# XXX: Testtools 0.9.5 doesn't have the content_from_file helper
2251
#detail_content = content.content_from_file(
2252
# log_file_path, buffer_now=True)
2253
with open(log_file_path, 'rb') as log_file:
2254
log_file_bytes = log_file.read()
2255
detail_content = content.Content(content.ContentType("text",
2256
"plain", {"charset": "utf8"}), lambda: [log_file_bytes])
2257
self.addDetail("start_bzr_subprocess-log-%d" % (count,),
2260
def _popen(self, *args, **kwargs):
2261
"""Place a call to Popen.
2263
Allows tests to override this method to intercept the calls made to
2264
Popen for introspection.
2266
return subprocess.Popen(*args, **kwargs)
2268
def get_source_path(self):
2269
"""Return the path of the directory containing bzrlib."""
2270
return os.path.dirname(os.path.dirname(bzrlib.__file__))
2272
def get_bzr_path(self):
2273
"""Return the path of the 'bzr' executable for this test suite."""
2274
bzr_path = os.path.join(self.get_source_path(), "bzr")
2275
if not os.path.isfile(bzr_path):
2276
# We are probably installed. Assume sys.argv is the right file
2277
bzr_path = sys.argv[0]
2280
def finish_bzr_subprocess(self, process, retcode=0, send_signal=None,
2281
universal_newlines=False, process_args=None):
2282
"""Finish the execution of process.
2284
:param process: the Popen object returned from start_bzr_subprocess.
2285
:param retcode: The status code that is expected. Defaults to 0. If
2286
None is supplied, the status code is not checked.
2287
:param send_signal: an optional signal to send to the process.
2288
:param universal_newlines: Convert CRLF => LF
2289
:returns: (stdout, stderr)
2291
if send_signal is not None:
2292
os.kill(process.pid, send_signal)
2293
out, err = process.communicate()
2295
if universal_newlines:
2296
out = out.replace('\r\n', '\n')
2297
err = err.replace('\r\n', '\n')
2299
if retcode is not None and retcode != process.returncode:
2300
if process_args is None:
2301
process_args = "(unknown args)"
2302
trace.mutter('Output of bzr %s:\n%s', process_args, out)
2303
trace.mutter('Error for bzr %s:\n%s', process_args, err)
2304
self.fail('Command bzr %s failed with retcode %s != %s'
2305
% (process_args, retcode, process.returncode))
2308
def check_tree_shape(self, tree, shape):
2309
"""Compare a tree to a list of expected names.
2311
Fail if they are not precisely equal.
2314
shape = list(shape) # copy
2315
for path, ie in tree.iter_entries_by_dir():
2316
name = path.replace('\\', '/')
2317
if ie.kind == 'directory':
2320
pass # ignore root entry
2326
self.fail("expected paths not found in inventory: %r" % shape)
2328
self.fail("unexpected paths found in inventory: %r" % extras)
2330
def apply_redirected(self, stdin=None, stdout=None, stderr=None,
2331
a_callable=None, *args, **kwargs):
2332
"""Call callable with redirected std io pipes.
2334
Returns the return code."""
2335
if not callable(a_callable):
2336
raise ValueError("a_callable must be callable.")
2338
stdin = StringIO("")
2340
if getattr(self, "_log_file", None) is not None:
2341
stdout = self._log_file
2345
if getattr(self, "_log_file", None is not None):
2346
stderr = self._log_file
2349
real_stdin = sys.stdin
2350
real_stdout = sys.stdout
2351
real_stderr = sys.stderr
2356
return a_callable(*args, **kwargs)
2358
sys.stdout = real_stdout
2359
sys.stderr = real_stderr
2360
sys.stdin = real_stdin
2362
def reduceLockdirTimeout(self):
2363
"""Reduce the default lock timeout for the duration of the test, so that
2364
if LockContention occurs during a test, it does so quickly.
2366
Tests that expect to provoke LockContention errors should call this.
2368
self.overrideAttr(lockdir, '_DEFAULT_TIMEOUT_SECONDS', 0)
2370
def make_utf8_encoded_stringio(self, encoding_type=None):
2371
"""Return a StringIOWrapper instance, that will encode Unicode
2374
if encoding_type is None:
2375
encoding_type = 'strict'
2377
output_encoding = 'utf-8'
2378
sio = codecs.getwriter(output_encoding)(sio, errors=encoding_type)
2379
sio.encoding = output_encoding
2382
def disable_verb(self, verb):
2383
"""Disable a smart server verb for one test."""
2384
from bzrlib.smart import request
2385
request_handlers = request.request_handlers
2386
orig_method = request_handlers.get(verb)
2387
orig_info = request_handlers.get_info(verb)
2388
request_handlers.remove(verb)
2389
self.addCleanup(request_handlers.register, verb, orig_method,
2393
class CapturedCall(object):
2394
"""A helper for capturing smart server calls for easy debug analysis."""
2396
def __init__(self, params, prefix_length):
2397
"""Capture the call with params and skip prefix_length stack frames."""
2400
# The last 5 frames are the __init__, the hook frame, and 3 smart
2401
# client frames. Beyond this we could get more clever, but this is good
2403
stack = traceback.extract_stack()[prefix_length:-5]
2404
self.stack = ''.join(traceback.format_list(stack))
2407
return self.call.method
2410
return self.call.method
2416
class TestCaseWithMemoryTransport(TestCase):
2417
"""Common test class for tests that do not need disk resources.
2419
Tests that need disk resources should derive from TestCaseInTempDir
2420
orTestCaseWithTransport.
2422
TestCaseWithMemoryTransport sets the TEST_ROOT variable for all bzr tests.
2424
For TestCaseWithMemoryTransport the ``test_home_dir`` is set to the name of
2425
a directory which does not exist. This serves to help ensure test isolation
2426
is preserved. ``test_dir`` is set to the TEST_ROOT, as is cwd, because they
2427
must exist. However, TestCaseWithMemoryTransport does not offer local file
2428
defaults for the transport in tests, nor does it obey the command line
2429
override, so tests that accidentally write to the common directory should
2432
:cvar TEST_ROOT: Directory containing all temporary directories, plus a
2433
``.bzr`` directory that stops us ascending higher into the filesystem.
2439
def __init__(self, methodName='runTest'):
2440
# allow test parameterization after test construction and before test
2441
# execution. Variables that the parameterizer sets need to be
2442
# ones that are not set by setUp, or setUp will trash them.
2443
super(TestCaseWithMemoryTransport, self).__init__(methodName)
2444
self.vfs_transport_factory = default_transport
2445
self.transport_server = None
2446
self.transport_readonly_server = None
2447
self.__vfs_server = None
2449
def get_transport(self, relpath=None):
2450
"""Return a writeable transport.
2452
This transport is for the test scratch space relative to
2455
:param relpath: a path relative to the base url.
2457
t = _mod_transport.get_transport_from_url(self.get_url(relpath))
2458
self.assertFalse(t.is_readonly())
2461
def get_readonly_transport(self, relpath=None):
2462
"""Return a readonly transport for the test scratch space
2464
This can be used to test that operations which should only need
2465
readonly access in fact do not try to write.
2467
:param relpath: a path relative to the base url.
2469
t = _mod_transport.get_transport_from_url(
2470
self.get_readonly_url(relpath))
2471
self.assertTrue(t.is_readonly())
2474
def create_transport_readonly_server(self):
2475
"""Create a transport server from class defined at init.
2477
This is mostly a hook for daughter classes.
2479
return self.transport_readonly_server()
2481
def get_readonly_server(self):
2482
"""Get the server instance for the readonly transport
2484
This is useful for some tests with specific servers to do diagnostics.
2486
if self.__readonly_server is None:
2487
if self.transport_readonly_server is None:
2488
# readonly decorator requested
2489
self.__readonly_server = test_server.ReadonlyServer()
2491
# explicit readonly transport.
2492
self.__readonly_server = self.create_transport_readonly_server()
2493
self.start_server(self.__readonly_server,
2494
self.get_vfs_only_server())
2495
return self.__readonly_server
2497
def get_readonly_url(self, relpath=None):
2498
"""Get a URL for the readonly transport.
2500
This will either be backed by '.' or a decorator to the transport
2501
used by self.get_url()
2502
relpath provides for clients to get a path relative to the base url.
2503
These should only be downwards relative, not upwards.
2505
base = self.get_readonly_server().get_url()
2506
return self._adjust_url(base, relpath)
2508
def get_vfs_only_server(self):
2509
"""Get the vfs only read/write server instance.
2511
This is useful for some tests with specific servers that need
2514
For TestCaseWithMemoryTransport this is always a MemoryServer, and there
2515
is no means to override it.
2517
if self.__vfs_server is None:
2518
self.__vfs_server = memory.MemoryServer()
2519
self.start_server(self.__vfs_server)
2520
return self.__vfs_server
2522
def get_server(self):
2523
"""Get the read/write server instance.
2525
This is useful for some tests with specific servers that need
2528
This is built from the self.transport_server factory. If that is None,
2529
then the self.get_vfs_server is returned.
2531
if self.__server is None:
2532
if (self.transport_server is None or self.transport_server is
2533
self.vfs_transport_factory):
2534
self.__server = self.get_vfs_only_server()
2536
# bring up a decorated means of access to the vfs only server.
2537
self.__server = self.transport_server()
2538
self.start_server(self.__server, self.get_vfs_only_server())
2539
return self.__server
2541
def _adjust_url(self, base, relpath):
2542
"""Get a URL (or maybe a path) for the readwrite transport.
2544
This will either be backed by '.' or to an equivalent non-file based
2546
relpath provides for clients to get a path relative to the base url.
2547
These should only be downwards relative, not upwards.
2549
if relpath is not None and relpath != '.':
2550
if not base.endswith('/'):
2552
# XXX: Really base should be a url; we did after all call
2553
# get_url()! But sometimes it's just a path (from
2554
# LocalAbspathServer), and it'd be wrong to append urlescaped data
2555
# to a non-escaped local path.
2556
if base.startswith('./') or base.startswith('/'):
2559
base += urlutils.escape(relpath)
2562
def get_url(self, relpath=None):
2563
"""Get a URL (or maybe a path) for the readwrite transport.
2565
This will either be backed by '.' or to an equivalent non-file based
2567
relpath provides for clients to get a path relative to the base url.
2568
These should only be downwards relative, not upwards.
2570
base = self.get_server().get_url()
2571
return self._adjust_url(base, relpath)
2573
def get_vfs_only_url(self, relpath=None):
2574
"""Get a URL (or maybe a path for the plain old vfs transport.
2576
This will never be a smart protocol. It always has all the
2577
capabilities of the local filesystem, but it might actually be a
2578
MemoryTransport or some other similar virtual filesystem.
2580
This is the backing transport (if any) of the server returned by
2581
get_url and get_readonly_url.
2583
:param relpath: provides for clients to get a path relative to the base
2584
url. These should only be downwards relative, not upwards.
2587
base = self.get_vfs_only_server().get_url()
2588
return self._adjust_url(base, relpath)
2590
def _create_safety_net(self):
2591
"""Make a fake bzr directory.
2593
This prevents any tests propagating up onto the TEST_ROOT directory's
2596
root = TestCaseWithMemoryTransport.TEST_ROOT
2597
# Make sure we get a readable and accessible home for .bzr.log
2598
# and/or config files, and not fallback to weird defaults (see
2599
# http://pad.lv/825027).
2600
self.assertIs(None, os.environ.get('BZR_HOME', None))
2601
os.environ['BZR_HOME'] = root
2602
wt = controldir.ControlDir.create_standalone_workingtree(root)
2603
del os.environ['BZR_HOME']
2604
# Hack for speed: remember the raw bytes of the dirstate file so that
2605
# we don't need to re-open the wt to check it hasn't changed.
2606
TestCaseWithMemoryTransport._SAFETY_NET_PRISTINE_DIRSTATE = (
2607
wt.control_transport.get_bytes('dirstate'))
2609
def _check_safety_net(self):
2610
"""Check that the safety .bzr directory have not been touched.
2612
_make_test_root have created a .bzr directory to prevent tests from
2613
propagating. This method ensures than a test did not leaked.
2615
root = TestCaseWithMemoryTransport.TEST_ROOT
2616
t = _mod_transport.get_transport_from_path(root)
2617
self.permit_url(t.base)
2618
if (t.get_bytes('.bzr/checkout/dirstate') !=
2619
TestCaseWithMemoryTransport._SAFETY_NET_PRISTINE_DIRSTATE):
2620
# The current test have modified the /bzr directory, we need to
2621
# recreate a new one or all the followng tests will fail.
2622
# If you need to inspect its content uncomment the following line
2623
# import pdb; pdb.set_trace()
2624
_rmtree_temp_dir(root + '/.bzr', test_id=self.id())
2625
self._create_safety_net()
2626
raise AssertionError('%s/.bzr should not be modified' % root)
2628
def _make_test_root(self):
2629
if TestCaseWithMemoryTransport.TEST_ROOT is None:
2630
# Watch out for tricky test dir (on OSX /tmp -> /private/tmp)
2631
root = osutils.realpath(osutils.mkdtemp(prefix='testbzr-',
2633
TestCaseWithMemoryTransport.TEST_ROOT = root
2635
self._create_safety_net()
2637
# The same directory is used by all tests, and we're not
2638
# specifically told when all tests are finished. This will do.
2639
atexit.register(_rmtree_temp_dir, root)
2641
self.permit_dir(TestCaseWithMemoryTransport.TEST_ROOT)
2642
self.addCleanup(self._check_safety_net)
2644
def makeAndChdirToTestDir(self):
2645
"""Create a temporary directories for this one test.
2647
This must set self.test_home_dir and self.test_dir and chdir to
2650
For TestCaseWithMemoryTransport we chdir to the TEST_ROOT for this test.
2652
os.chdir(TestCaseWithMemoryTransport.TEST_ROOT)
2653
self.test_dir = TestCaseWithMemoryTransport.TEST_ROOT
2654
self.test_home_dir = self.test_dir + "/MemoryTransportMissingHomeDir"
2655
self.permit_dir(self.test_dir)
2657
def make_branch(self, relpath, format=None, name=None):
2658
"""Create a branch on the transport at relpath."""
2659
repo = self.make_repository(relpath, format=format)
2660
return repo.bzrdir.create_branch(append_revisions_only=False,
2663
def get_default_format(self):
2666
def resolve_format(self, format):
2667
"""Resolve an object to a ControlDir format object.
2669
The initial format object can either already be
2670
a ControlDirFormat, None (for the default format),
2671
or a string with the name of the control dir format.
2673
:param format: Object to resolve
2674
:return A ControlDirFormat instance
2677
format = self.get_default_format()
2678
if isinstance(format, basestring):
2679
format = controldir.format_registry.make_bzrdir(format)
2682
def make_bzrdir(self, relpath, format=None):
2684
# might be a relative or absolute path
2685
maybe_a_url = self.get_url(relpath)
2686
segments = maybe_a_url.rsplit('/', 1)
2687
t = _mod_transport.get_transport(maybe_a_url)
2688
if len(segments) > 1 and segments[-1] not in ('', '.'):
2690
format = self.resolve_format(format)
2691
return format.initialize_on_transport(t)
2692
except errors.UninitializableFormat:
2693
raise TestSkipped("Format %s is not initializable." % format)
2695
def make_repository(self, relpath, shared=None, format=None):
2696
"""Create a repository on our default transport at relpath.
2698
Note that relpath must be a relative path, not a full url.
2700
# FIXME: If you create a remoterepository this returns the underlying
2701
# real format, which is incorrect. Actually we should make sure that
2702
# RemoteBzrDir returns a RemoteRepository.
2703
# maybe mbp 20070410
2704
made_control = self.make_bzrdir(relpath, format=format)
2705
return made_control.create_repository(shared=shared)
2707
def make_smart_server(self, path, backing_server=None):
2708
if backing_server is None:
2709
backing_server = self.get_server()
2710
smart_server = test_server.SmartTCPServer_for_testing()
2711
self.start_server(smart_server, backing_server)
2712
remote_transport = _mod_transport.get_transport_from_url(smart_server.get_url()
2714
return remote_transport
2716
def make_branch_and_memory_tree(self, relpath, format=None):
2717
"""Create a branch on the default transport and a MemoryTree for it."""
2718
b = self.make_branch(relpath, format=format)
2719
return memorytree.MemoryTree.create_on_branch(b)
2721
def make_branch_builder(self, relpath, format=None):
2722
branch = self.make_branch(relpath, format=format)
2723
return branchbuilder.BranchBuilder(branch=branch)
2725
def overrideEnvironmentForTesting(self):
2726
test_home_dir = self.test_home_dir
2727
if isinstance(test_home_dir, unicode):
2728
test_home_dir = test_home_dir.encode(sys.getfilesystemencoding())
2729
self.overrideEnv('HOME', test_home_dir)
2730
self.overrideEnv('BZR_HOME', test_home_dir)
2733
super(TestCaseWithMemoryTransport, self).setUp()
2735
def _add_disconnect_cleanup(transport):
2736
"""Schedule disconnection of given transport at test cleanup
2738
This needs to happen for all connected transports or leaks occur.
2740
Note reconnections may mean we call disconnect multiple times per
2741
transport which is suboptimal but seems harmless.
2743
self.addCleanup(transport.disconnect)
2745
_mod_transport.Transport.hooks.install_named_hook('post_connect',
2746
_add_disconnect_cleanup, None)
2748
self._make_test_root()
2749
self.addCleanup(os.chdir, os.getcwdu())
2750
self.makeAndChdirToTestDir()
2751
self.overrideEnvironmentForTesting()
2752
self.__readonly_server = None
2753
self.__server = None
2754
self.reduceLockdirTimeout()
2756
def setup_smart_server_with_call_log(self):
2757
"""Sets up a smart server as the transport server with a call log."""
2758
self.transport_server = test_server.SmartTCPServer_for_testing
2759
self.hpss_connections = []
2760
self.hpss_calls = []
2762
# Skip the current stack down to the caller of
2763
# setup_smart_server_with_call_log
2764
prefix_length = len(traceback.extract_stack()) - 2
2765
def capture_hpss_call(params):
2766
self.hpss_calls.append(
2767
CapturedCall(params, prefix_length))
2768
def capture_connect(transport):
2769
self.hpss_connections.append(transport)
2770
client._SmartClient.hooks.install_named_hook(
2771
'call', capture_hpss_call, None)
2772
_mod_transport.Transport.hooks.install_named_hook(
2773
'post_connect', capture_connect, None)
2775
def reset_smart_call_log(self):
2776
self.hpss_calls = []
2777
self.hpss_connections = []
2780
class TestCaseInTempDir(TestCaseWithMemoryTransport):
2781
"""Derived class that runs a test within a temporary directory.
2783
This is useful for tests that need to create a branch, etc.
2785
The directory is created in a slightly complex way: for each
2786
Python invocation, a new temporary top-level directory is created.
2787
All test cases create their own directory within that. If the
2788
tests complete successfully, the directory is removed.
2790
:ivar test_base_dir: The path of the top-level directory for this
2791
test, which contains a home directory and a work directory.
2793
:ivar test_home_dir: An initially empty directory under test_base_dir
2794
which is used as $HOME for this test.
2796
:ivar test_dir: A directory under test_base_dir used as the current
2797
directory when the test proper is run.
2800
OVERRIDE_PYTHON = 'python'
2803
super(TestCaseInTempDir, self).setUp()
2804
# Remove the protection set in isolated_environ, we have a proper
2805
# access to disk resources now.
2806
self.overrideEnv('BZR_LOG', None)
2808
def check_file_contents(self, filename, expect):
2809
self.log("check contents of file %s" % filename)
2815
if contents != expect:
2816
self.log("expected: %r" % expect)
2817
self.log("actually: %r" % contents)
2818
self.fail("contents of %s not as expected" % filename)
2820
def _getTestDirPrefix(self):
2821
# create a directory within the top level test directory
2822
if sys.platform in ('win32', 'cygwin'):
2823
name_prefix = re.sub('[<>*=+",:;_/\\-]', '_', self.id())
2824
# windows is likely to have path-length limits so use a short name
2825
name_prefix = name_prefix[-30:]
2827
name_prefix = re.sub('[/]', '_', self.id())
2830
def makeAndChdirToTestDir(self):
2831
"""See TestCaseWithMemoryTransport.makeAndChdirToTestDir().
2833
For TestCaseInTempDir we create a temporary directory based on the test
2834
name and then create two subdirs - test and home under it.
2836
name_prefix = osutils.pathjoin(TestCaseWithMemoryTransport.TEST_ROOT,
2837
self._getTestDirPrefix())
2839
for i in range(100):
2840
if os.path.exists(name):
2841
name = name_prefix + '_' + str(i)
2843
# now create test and home directories within this dir
2844
self.test_base_dir = name
2845
self.addCleanup(self.deleteTestDir)
2846
os.mkdir(self.test_base_dir)
2848
self.permit_dir(self.test_base_dir)
2849
# 'sprouting' and 'init' of a branch both walk up the tree to find
2850
# stacking policy to honour; create a bzr dir with an unshared
2851
# repository (but not a branch - our code would be trying to escape
2852
# then!) to stop them, and permit it to be read.
2853
# control = controldir.ControlDir.create(self.test_base_dir)
2854
# control.create_repository()
2855
self.test_home_dir = self.test_base_dir + '/home'
2856
os.mkdir(self.test_home_dir)
2857
self.test_dir = self.test_base_dir + '/work'
2858
os.mkdir(self.test_dir)
2859
os.chdir(self.test_dir)
2860
# put name of test inside
2861
f = file(self.test_base_dir + '/name', 'w')
2867
def deleteTestDir(self):
2868
os.chdir(TestCaseWithMemoryTransport.TEST_ROOT)
2869
_rmtree_temp_dir(self.test_base_dir, test_id=self.id())
2871
def build_tree(self, shape, line_endings='binary', transport=None):
2872
"""Build a test tree according to a pattern.
2874
shape is a sequence of file specifications. If the final
2875
character is '/', a directory is created.
2877
This assumes that all the elements in the tree being built are new.
2879
This doesn't add anything to a branch.
2881
:type shape: list or tuple.
2882
:param line_endings: Either 'binary' or 'native'
2883
in binary mode, exact contents are written in native mode, the
2884
line endings match the default platform endings.
2885
:param transport: A transport to write to, for building trees on VFS's.
2886
If the transport is readonly or None, "." is opened automatically.
2889
if type(shape) not in (list, tuple):
2890
raise AssertionError("Parameter 'shape' should be "
2891
"a list or a tuple. Got %r instead" % (shape,))
2892
# It's OK to just create them using forward slashes on windows.
2893
if transport is None or transport.is_readonly():
2894
transport = _mod_transport.get_transport_from_path(".")
2896
self.assertIsInstance(name, basestring)
2898
transport.mkdir(urlutils.escape(name[:-1]))
2900
if line_endings == 'binary':
2902
elif line_endings == 'native':
2905
raise errors.BzrError(
2906
'Invalid line ending request %r' % line_endings)
2907
content = "contents of %s%s" % (name.encode('utf-8'), end)
2908
transport.put_bytes_non_atomic(urlutils.escape(name), content)
2910
build_tree_contents = staticmethod(treeshape.build_tree_contents)
2912
def assertInWorkingTree(self, path, root_path='.', tree=None):
2913
"""Assert whether path or paths are in the WorkingTree"""
2915
tree = workingtree.WorkingTree.open(root_path)
2916
if not isinstance(path, basestring):
2918
self.assertInWorkingTree(p, tree=tree)
2920
self.assertIsNot(tree.path2id(path), None,
2921
path+' not in working tree.')
2923
def assertNotInWorkingTree(self, path, root_path='.', tree=None):
2924
"""Assert whether path or paths are not in the WorkingTree"""
2926
tree = workingtree.WorkingTree.open(root_path)
2927
if not isinstance(path, basestring):
2929
self.assertNotInWorkingTree(p,tree=tree)
2931
self.assertIs(tree.path2id(path), None, path+' in working tree.')
2934
class TestCaseWithTransport(TestCaseInTempDir):
2935
"""A test case that provides get_url and get_readonly_url facilities.
2937
These back onto two transport servers, one for readonly access and one for
2940
If no explicit class is provided for readonly access, a
2941
ReadonlyTransportDecorator is used instead which allows the use of non disk
2942
based read write transports.
2944
If an explicit class is provided for readonly access, that server and the
2945
readwrite one must both define get_url() as resolving to os.getcwd().
2948
def get_vfs_only_server(self):
2949
"""See TestCaseWithMemoryTransport.
2951
This is useful for some tests with specific servers that need
2954
if self.__vfs_server is None:
2955
self.__vfs_server = self.vfs_transport_factory()
2956
self.start_server(self.__vfs_server)
2957
return self.__vfs_server
2959
def make_branch_and_tree(self, relpath, format=None):
2960
"""Create a branch on the transport and a tree locally.
2962
If the transport is not a LocalTransport, the Tree can't be created on
2963
the transport. In that case if the vfs_transport_factory is
2964
LocalURLServer the working tree is created in the local
2965
directory backing the transport, and the returned tree's branch and
2966
repository will also be accessed locally. Otherwise a lightweight
2967
checkout is created and returned.
2969
We do this because we can't physically create a tree in the local
2970
path, with a branch reference to the transport_factory url, and
2971
a branch + repository in the vfs_transport, unless the vfs_transport
2972
namespace is distinct from the local disk - the two branch objects
2973
would collide. While we could construct a tree with its branch object
2974
pointing at the transport_factory transport in memory, reopening it
2975
would behaving unexpectedly, and has in the past caused testing bugs
2976
when we tried to do it that way.
2978
:param format: The BzrDirFormat.
2979
:returns: the WorkingTree.
2981
# TODO: always use the local disk path for the working tree,
2982
# this obviously requires a format that supports branch references
2983
# so check for that by checking bzrdir.BzrDirFormat.get_default_format()
2985
format = self.resolve_format(format=format)
2986
if not format.supports_workingtrees:
2987
b = self.make_branch(relpath+'.branch', format=format)
2988
return b.create_checkout(relpath, lightweight=True)
2989
b = self.make_branch(relpath, format=format)
2991
return b.bzrdir.create_workingtree()
2992
except errors.NotLocalUrl:
2993
# We can only make working trees locally at the moment. If the
2994
# transport can't support them, then we keep the non-disk-backed
2995
# branch and create a local checkout.
2996
if self.vfs_transport_factory is test_server.LocalURLServer:
2997
# the branch is colocated on disk, we cannot create a checkout.
2998
# hopefully callers will expect this.
2999
local_controldir = controldir.ControlDir.open(
3000
self.get_vfs_only_url(relpath))
3001
wt = local_controldir.create_workingtree()
3002
if wt.branch._format != b._format:
3004
# Make sure that assigning to wt._branch fixes wt.branch,
3005
# in case the implementation details of workingtree objects
3007
self.assertIs(b, wt.branch)
3010
return b.create_checkout(relpath, lightweight=True)
3012
def assertIsDirectory(self, relpath, transport):
3013
"""Assert that relpath within transport is a directory.
3015
This may not be possible on all transports; in that case it propagates
3016
a TransportNotPossible.
3019
mode = transport.stat(relpath).st_mode
3020
except errors.NoSuchFile:
3021
self.fail("path %s is not a directory; no such file"
3023
if not stat.S_ISDIR(mode):
3024
self.fail("path %s is not a directory; has mode %#o"
3027
def assertTreesEqual(self, left, right):
3028
"""Check that left and right have the same content and properties."""
3029
# we use a tree delta to check for equality of the content, and we
3030
# manually check for equality of other things such as the parents list.
3031
self.assertEqual(left.get_parent_ids(), right.get_parent_ids())
3032
differences = left.changes_from(right)
3033
self.assertFalse(differences.has_changed(),
3034
"Trees %r and %r are different: %r" % (left, right, differences))
3037
super(TestCaseWithTransport, self).setUp()
3038
self.__vfs_server = None
3040
def disable_missing_extensions_warning(self):
3041
"""Some tests expect a precise stderr content.
3043
There is no point in forcing them to duplicate the extension related
3046
config.GlobalConfig().set_user_option('ignore_missing_extensions', True)
3049
class ChrootedTestCase(TestCaseWithTransport):
3050
"""A support class that provides readonly urls outside the local namespace.
3052
This is done by checking if self.transport_server is a MemoryServer. if it
3053
is then we are chrooted already, if it is not then an HttpServer is used
3056
TODO RBC 20060127: make this an option to TestCaseWithTransport so it can
3057
be used without needed to redo it when a different
3058
subclass is in use ?
3062
from bzrlib.tests import http_server
3063
super(ChrootedTestCase, self).setUp()
3064
if not self.vfs_transport_factory == memory.MemoryServer:
3065
self.transport_readonly_server = http_server.HttpServer
3068
def condition_id_re(pattern):
3069
"""Create a condition filter which performs a re check on a test's id.
3071
:param pattern: A regular expression string.
3072
:return: A callable that returns True if the re matches.
3074
filter_re = re.compile(pattern, 0)
3075
def condition(test):
3077
return filter_re.search(test_id)
3081
def condition_isinstance(klass_or_klass_list):
3082
"""Create a condition filter which returns isinstance(param, klass).
3084
:return: A callable which when called with one parameter obj return the
3085
result of isinstance(obj, klass_or_klass_list).
3088
return isinstance(obj, klass_or_klass_list)
3092
def condition_id_in_list(id_list):
3093
"""Create a condition filter which verify that test's id in a list.
3095
:param id_list: A TestIdList object.
3096
:return: A callable that returns True if the test's id appears in the list.
3098
def condition(test):
3099
return id_list.includes(test.id())
3103
def condition_id_startswith(starts):
3104
"""Create a condition filter verifying that test's id starts with a string.
3106
:param starts: A list of string.
3107
:return: A callable that returns True if the test's id starts with one of
3110
def condition(test):
3111
for start in starts:
3112
if test.id().startswith(start):
3118
def exclude_tests_by_condition(suite, condition):
3119
"""Create a test suite which excludes some tests from suite.
3121
:param suite: The suite to get tests from.
3122
:param condition: A callable whose result evaluates True when called with a
3123
test case which should be excluded from the result.
3124
:return: A suite which contains the tests found in suite that fail
3128
for test in iter_suite_tests(suite):
3129
if not condition(test):
3131
return TestUtil.TestSuite(result)
3134
def filter_suite_by_condition(suite, condition):
3135
"""Create a test suite by filtering another one.
3137
:param suite: The source suite.
3138
:param condition: A callable whose result evaluates True when called with a
3139
test case which should be included in the result.
3140
:return: A suite which contains the tests found in suite that pass
3144
for test in iter_suite_tests(suite):
3147
return TestUtil.TestSuite(result)
3150
def filter_suite_by_re(suite, pattern):
3151
"""Create a test suite by filtering another one.
3153
:param suite: the source suite
3154
:param pattern: pattern that names must match
3155
:returns: the newly created suite
3157
condition = condition_id_re(pattern)
3158
result_suite = filter_suite_by_condition(suite, condition)
3162
def filter_suite_by_id_list(suite, test_id_list):
3163
"""Create a test suite by filtering another one.
3165
:param suite: The source suite.
3166
:param test_id_list: A list of the test ids to keep as strings.
3167
:returns: the newly created suite
3169
condition = condition_id_in_list(test_id_list)
3170
result_suite = filter_suite_by_condition(suite, condition)
3174
def filter_suite_by_id_startswith(suite, start):
3175
"""Create a test suite by filtering another one.
3177
:param suite: The source suite.
3178
:param start: A list of string the test id must start with one of.
3179
:returns: the newly created suite
3181
condition = condition_id_startswith(start)
3182
result_suite = filter_suite_by_condition(suite, condition)
3186
def exclude_tests_by_re(suite, pattern):
3187
"""Create a test suite which excludes some tests from suite.
3189
:param suite: The suite to get tests from.
3190
:param pattern: A regular expression string. Test ids that match this
3191
pattern will be excluded from the result.
3192
:return: A TestSuite that contains all the tests from suite without the
3193
tests that matched pattern. The order of tests is the same as it was in
3196
return exclude_tests_by_condition(suite, condition_id_re(pattern))
3199
def preserve_input(something):
3200
"""A helper for performing test suite transformation chains.
3202
:param something: Anything you want to preserve.
3208
def randomize_suite(suite):
3209
"""Return a new TestSuite with suite's tests in random order.
3211
The tests in the input suite are flattened into a single suite in order to
3212
accomplish this. Any nested TestSuites are removed to provide global
3215
tests = list(iter_suite_tests(suite))
3216
random.shuffle(tests)
3217
return TestUtil.TestSuite(tests)
3220
def split_suite_by_condition(suite, condition):
3221
"""Split a test suite into two by a condition.
3223
:param suite: The suite to split.
3224
:param condition: The condition to match on. Tests that match this
3225
condition are returned in the first test suite, ones that do not match
3226
are in the second suite.
3227
:return: A tuple of two test suites, where the first contains tests from
3228
suite matching the condition, and the second contains the remainder
3229
from suite. The order within each output suite is the same as it was in
3234
for test in iter_suite_tests(suite):
3236
matched.append(test)
3238
did_not_match.append(test)
3239
return TestUtil.TestSuite(matched), TestUtil.TestSuite(did_not_match)
3242
def split_suite_by_re(suite, pattern):
3243
"""Split a test suite into two by a regular expression.
3245
:param suite: The suite to split.
3246
:param pattern: A regular expression string. Test ids that match this
3247
pattern will be in the first test suite returned, and the others in the
3248
second test suite returned.
3249
:return: A tuple of two test suites, where the first contains tests from
3250
suite matching pattern, and the second contains the remainder from
3251
suite. The order within each output suite is the same as it was in
3254
return split_suite_by_condition(suite, condition_id_re(pattern))
3257
def run_suite(suite, name='test', verbose=False, pattern=".*",
3258
stop_on_failure=False,
3259
transport=None, lsprof_timed=None, bench_history=None,
3260
matching_tests_first=None,
3263
exclude_pattern=None,
3266
suite_decorators=None,
3268
result_decorators=None,
3270
"""Run a test suite for bzr selftest.
3272
:param runner_class: The class of runner to use. Must support the
3273
constructor arguments passed by run_suite which are more than standard
3275
:return: A boolean indicating success.
3277
TestCase._gather_lsprof_in_benchmarks = lsprof_timed
3282
if runner_class is None:
3283
runner_class = TextTestRunner
3286
runner = runner_class(stream=stream,
3288
verbosity=verbosity,
3289
bench_history=bench_history,
3291
result_decorators=result_decorators,
3293
runner.stop_on_failure=stop_on_failure
3294
if isinstance(suite, unittest.TestSuite):
3295
# Empty out _tests list of passed suite and populate new TestSuite
3296
suite._tests[:], suite = [], TestSuite(suite)
3297
# built in decorator factories:
3299
random_order(random_seed, runner),
3300
exclude_tests(exclude_pattern),
3302
if matching_tests_first:
3303
decorators.append(tests_first(pattern))
3305
decorators.append(filter_tests(pattern))
3306
if suite_decorators:
3307
decorators.extend(suite_decorators)
3308
# tell the result object how many tests will be running: (except if
3309
# --parallel=fork is being used. Robert said he will provide a better
3310
# progress design later -- vila 20090817)
3311
if fork_decorator not in decorators:
3312
decorators.append(CountingDecorator)
3313
for decorator in decorators:
3314
suite = decorator(suite)
3316
# Done after test suite decoration to allow randomisation etc
3317
# to take effect, though that is of marginal benefit.
3319
stream.write("Listing tests only ...\n")
3320
for t in iter_suite_tests(suite):
3321
stream.write("%s\n" % (t.id()))
3323
result = runner.run(suite)
3325
return result.wasStrictlySuccessful()
3327
return result.wasSuccessful()
3330
# A registry where get() returns a suite decorator.
3331
parallel_registry = registry.Registry()
3334
def fork_decorator(suite):
3335
if getattr(os, "fork", None) is None:
3336
raise errors.BzrCommandError("platform does not support fork,"
3337
" try --parallel=subprocess instead.")
3338
concurrency = osutils.local_concurrency()
3339
if concurrency == 1:
3341
from testtools import ConcurrentTestSuite
3342
return ConcurrentTestSuite(suite, fork_for_tests)
3343
parallel_registry.register('fork', fork_decorator)
3346
def subprocess_decorator(suite):
3347
concurrency = osutils.local_concurrency()
3348
if concurrency == 1:
3350
from testtools import ConcurrentTestSuite
3351
return ConcurrentTestSuite(suite, reinvoke_for_tests)
3352
parallel_registry.register('subprocess', subprocess_decorator)
3355
def exclude_tests(exclude_pattern):
3356
"""Return a test suite decorator that excludes tests."""
3357
if exclude_pattern is None:
3358
return identity_decorator
3359
def decorator(suite):
3360
return ExcludeDecorator(suite, exclude_pattern)
3364
def filter_tests(pattern):
3366
return identity_decorator
3367
def decorator(suite):
3368
return FilterTestsDecorator(suite, pattern)
3372
def random_order(random_seed, runner):
3373
"""Return a test suite decorator factory for randomising tests order.
3375
:param random_seed: now, a string which casts to a long, or a long.
3376
:param runner: A test runner with a stream attribute to report on.
3378
if random_seed is None:
3379
return identity_decorator
3380
def decorator(suite):
3381
return RandomDecorator(suite, random_seed, runner.stream)
3385
def tests_first(pattern):
3387
return identity_decorator
3388
def decorator(suite):
3389
return TestFirstDecorator(suite, pattern)
3393
def identity_decorator(suite):
3398
class TestDecorator(TestUtil.TestSuite):
3399
"""A decorator for TestCase/TestSuite objects.
3401
Contains rather than flattening suite passed on construction
3404
def __init__(self, suite=None):
3405
super(TestDecorator, self).__init__()
3406
if suite is not None:
3409
# Don't need subclass run method with suite emptying
3410
run = unittest.TestSuite.run
3413
class CountingDecorator(TestDecorator):
3414
"""A decorator which calls result.progress(self.countTestCases)."""
3416
def run(self, result):
3417
progress_method = getattr(result, 'progress', None)
3418
if callable(progress_method):
3419
progress_method(self.countTestCases(), SUBUNIT_SEEK_SET)
3420
return super(CountingDecorator, self).run(result)
3423
class ExcludeDecorator(TestDecorator):
3424
"""A decorator which excludes test matching an exclude pattern."""
3426
def __init__(self, suite, exclude_pattern):
3427
super(ExcludeDecorator, self).__init__(
3428
exclude_tests_by_re(suite, exclude_pattern))
3431
class FilterTestsDecorator(TestDecorator):
3432
"""A decorator which filters tests to those matching a pattern."""
3434
def __init__(self, suite, pattern):
3435
super(FilterTestsDecorator, self).__init__(
3436
filter_suite_by_re(suite, pattern))
3439
class RandomDecorator(TestDecorator):
3440
"""A decorator which randomises the order of its tests."""
3442
def __init__(self, suite, random_seed, stream):
3443
random_seed = self.actual_seed(random_seed)
3444
stream.write("Randomizing test order using seed %s\n\n" %
3446
# Initialise the random number generator.
3447
random.seed(random_seed)
3448
super(RandomDecorator, self).__init__(randomize_suite(suite))
3451
def actual_seed(seed):
3453
# We convert the seed to a long to make it reuseable across
3454
# invocations (because the user can reenter it).
3455
return long(time.time())
3457
# Convert the seed to a long if we can
3460
except (TypeError, ValueError):
3465
class TestFirstDecorator(TestDecorator):
3466
"""A decorator which moves named tests to the front."""
3468
def __init__(self, suite, pattern):
3469
super(TestFirstDecorator, self).__init__()
3470
self.addTests(split_suite_by_re(suite, pattern))
3473
def partition_tests(suite, count):
3474
"""Partition suite into count lists of tests."""
3475
# This just assigns tests in a round-robin fashion. On one hand this
3476
# splits up blocks of related tests that might run faster if they shared
3477
# resources, but on the other it avoids assigning blocks of slow tests to
3478
# just one partition. So the slowest partition shouldn't be much slower
3480
partitions = [list() for i in range(count)]
3481
tests = iter_suite_tests(suite)
3482
for partition, test in itertools.izip(itertools.cycle(partitions), tests):
3483
partition.append(test)
3487
def workaround_zealous_crypto_random():
3488
"""Crypto.Random want to help us being secure, but we don't care here.
3490
This workaround some test failure related to the sftp server. Once paramiko
3491
stop using the controversial API in Crypto.Random, we may get rid of it.
3494
from Crypto.Random import atfork
3500
def fork_for_tests(suite):
3501
"""Take suite and start up one runner per CPU by forking()
3503
:return: An iterable of TestCase-like objects which can each have
3504
run(result) called on them to feed tests to result.
3506
concurrency = osutils.local_concurrency()
3508
from subunit import ProtocolTestCase
3509
from subunit.test_results import AutoTimingTestResultDecorator
3510
class TestInOtherProcess(ProtocolTestCase):
3511
# Should be in subunit, I think. RBC.
3512
def __init__(self, stream, pid):
3513
ProtocolTestCase.__init__(self, stream)
3516
def run(self, result):
3518
ProtocolTestCase.run(self, result)
3520
pid, status = os.waitpid(self.pid, 0)
3521
# GZ 2011-10-18: If status is nonzero, should report to the result
3522
# that something went wrong.
3524
test_blocks = partition_tests(suite, concurrency)
3525
# Clear the tests from the original suite so it doesn't keep them alive
3526
suite._tests[:] = []
3527
for process_tests in test_blocks:
3528
process_suite = TestUtil.TestSuite(process_tests)
3529
# Also clear each split list so new suite has only reference
3530
process_tests[:] = []
3531
c2pread, c2pwrite = os.pipe()
3535
stream = os.fdopen(c2pwrite, 'wb', 1)
3536
workaround_zealous_crypto_random()
3538
# Leave stderr and stdout open so we can see test noise
3539
# Close stdin so that the child goes away if it decides to
3540
# read from stdin (otherwise its a roulette to see what
3541
# child actually gets keystrokes for pdb etc).
3543
subunit_result = AutoTimingTestResultDecorator(
3544
SubUnitBzrProtocolClient(stream))
3545
process_suite.run(subunit_result)
3547
# Try and report traceback on stream, but exit with error even
3548
# if stream couldn't be created or something else goes wrong.
3549
# The traceback is formatted to a string and written in one go
3550
# to avoid interleaving lines from multiple failing children.
3552
stream.write(traceback.format_exc())
3558
stream = os.fdopen(c2pread, 'rb', 1)
3559
test = TestInOtherProcess(stream, pid)
3564
def reinvoke_for_tests(suite):
3565
"""Take suite and start up one runner per CPU using subprocess().
3567
:return: An iterable of TestCase-like objects which can each have
3568
run(result) called on them to feed tests to result.
3570
concurrency = osutils.local_concurrency()
3572
from subunit import ProtocolTestCase
3573
class TestInSubprocess(ProtocolTestCase):
3574
def __init__(self, process, name):
3575
ProtocolTestCase.__init__(self, process.stdout)
3576
self.process = process
3577
self.process.stdin.close()
3580
def run(self, result):
3582
ProtocolTestCase.run(self, result)
3585
os.unlink(self.name)
3586
# print "pid %d finished" % finished_process
3587
test_blocks = partition_tests(suite, concurrency)
3588
for process_tests in test_blocks:
3589
# ugly; currently reimplement rather than reuses TestCase methods.
3590
bzr_path = os.path.dirname(os.path.dirname(bzrlib.__file__))+'/bzr'
3591
if not os.path.isfile(bzr_path):
3592
# We are probably installed. Assume sys.argv is the right file
3593
bzr_path = sys.argv[0]
3594
bzr_path = [bzr_path]
3595
if sys.platform == "win32":
3596
# if we're on windows, we can't execute the bzr script directly
3597
bzr_path = [sys.executable] + bzr_path
3598
fd, test_list_file_name = tempfile.mkstemp()
3599
test_list_file = os.fdopen(fd, 'wb', 1)
3600
for test in process_tests:
3601
test_list_file.write(test.id() + '\n')
3602
test_list_file.close()
3604
argv = bzr_path + ['selftest', '--load-list', test_list_file_name,
3606
if '--no-plugins' in sys.argv:
3607
argv.append('--no-plugins')
3608
# stderr=subprocess.STDOUT would be ideal, but until we prevent
3609
# noise on stderr it can interrupt the subunit protocol.
3610
process = subprocess.Popen(argv, stdin=subprocess.PIPE,
3611
stdout=subprocess.PIPE,
3612
stderr=subprocess.PIPE,
3614
test = TestInSubprocess(process, test_list_file_name)
3617
os.unlink(test_list_file_name)
3622
class ProfileResult(testtools.ExtendedToOriginalDecorator):
3623
"""Generate profiling data for all activity between start and success.
3625
The profile data is appended to the test's _benchcalls attribute and can
3626
be accessed by the forwarded-to TestResult.
3628
While it might be cleaner do accumulate this in stopTest, addSuccess is
3629
where our existing output support for lsprof is, and this class aims to
3630
fit in with that: while it could be moved it's not necessary to accomplish
3631
test profiling, nor would it be dramatically cleaner.
3634
def startTest(self, test):
3635
self.profiler = bzrlib.lsprof.BzrProfiler()
3636
# Prevent deadlocks in tests that use lsprof: those tests will
3638
bzrlib.lsprof.BzrProfiler.profiler_block = 0
3639
self.profiler.start()
3640
testtools.ExtendedToOriginalDecorator.startTest(self, test)
3642
def addSuccess(self, test):
3643
stats = self.profiler.stop()
3645
calls = test._benchcalls
3646
except AttributeError:
3647
test._benchcalls = []
3648
calls = test._benchcalls
3649
calls.append(((test.id(), "", ""), stats))
3650
testtools.ExtendedToOriginalDecorator.addSuccess(self, test)
3652
def stopTest(self, test):
3653
testtools.ExtendedToOriginalDecorator.stopTest(self, test)
3654
self.profiler = None
3657
# Controlled by "bzr selftest -E=..." option
3658
# Currently supported:
3659
# -Eallow_debug Will no longer clear debug.debug_flags() so it
3660
# preserves any flags supplied at the command line.
3661
# -Edisable_lock_checks Turns errors in mismatched locks into simple prints
3662
# rather than failing tests. And no longer raise
3663
# LockContention when fctnl locks are not being used
3664
# with proper exclusion rules.
3665
# -Ethreads Will display thread ident at creation/join time to
3666
# help track thread leaks
3667
# -Euncollected_cases Display the identity of any test cases that weren't
3668
# deallocated after being completed.
3669
# -Econfig_stats Will collect statistics using addDetail
3670
selftest_debug_flags = set()
3673
def selftest(verbose=False, pattern=".*", stop_on_failure=True,
3675
test_suite_factory=None,
3678
matching_tests_first=None,
3681
exclude_pattern=None,
3687
suite_decorators=None,
3691
"""Run the whole test suite under the enhanced runner"""
3692
# XXX: Very ugly way to do this...
3693
# Disable warning about old formats because we don't want it to disturb
3694
# any blackbox tests.
3695
from bzrlib import repository
3696
repository._deprecation_warning_done = True
3698
global default_transport
3699
if transport is None:
3700
transport = default_transport
3701
old_transport = default_transport
3702
default_transport = transport
3703
global selftest_debug_flags
3704
old_debug_flags = selftest_debug_flags
3705
if debug_flags is not None:
3706
selftest_debug_flags = set(debug_flags)
3708
if load_list is None:
3711
keep_only = load_test_id_list(load_list)
3713
starting_with = [test_prefix_alias_registry.resolve_alias(start)
3714
for start in starting_with]
3715
if test_suite_factory is None:
3716
# Reduce loading time by loading modules based on the starting_with
3718
suite = test_suite(keep_only, starting_with)
3720
suite = test_suite_factory()
3722
# But always filter as requested.
3723
suite = filter_suite_by_id_startswith(suite, starting_with)
3724
result_decorators = []
3726
result_decorators.append(ProfileResult)
3727
return run_suite(suite, 'testbzr', verbose=verbose, pattern=pattern,
3728
stop_on_failure=stop_on_failure,
3729
transport=transport,
3730
lsprof_timed=lsprof_timed,
3731
bench_history=bench_history,
3732
matching_tests_first=matching_tests_first,
3733
list_only=list_only,
3734
random_seed=random_seed,
3735
exclude_pattern=exclude_pattern,
3737
runner_class=runner_class,
3738
suite_decorators=suite_decorators,
3740
result_decorators=result_decorators,
3743
default_transport = old_transport
3744
selftest_debug_flags = old_debug_flags
3747
def load_test_id_list(file_name):
3748
"""Load a test id list from a text file.
3750
The format is one test id by line. No special care is taken to impose
3751
strict rules, these test ids are used to filter the test suite so a test id
3752
that do not match an existing test will do no harm. This allows user to add
3753
comments, leave blank lines, etc.
3757
ftest = open(file_name, 'rt')
3759
if e.errno != errno.ENOENT:
3762
raise errors.NoSuchFile(file_name)
3764
for test_name in ftest.readlines():
3765
test_list.append(test_name.strip())
3770
def suite_matches_id_list(test_suite, id_list):
3771
"""Warns about tests not appearing or appearing more than once.
3773
:param test_suite: A TestSuite object.
3774
:param test_id_list: The list of test ids that should be found in
3777
:return: (absents, duplicates) absents is a list containing the test found
3778
in id_list but not in test_suite, duplicates is a list containing the
3779
test found multiple times in test_suite.
3781
When using a prefined test id list, it may occurs that some tests do not
3782
exist anymore or that some tests use the same id. This function warns the
3783
tester about potential problems in his workflow (test lists are volatile)
3784
or in the test suite itself (using the same id for several tests does not
3785
help to localize defects).
3787
# Build a dict counting id occurrences
3789
for test in iter_suite_tests(test_suite):
3791
tests[id] = tests.get(id, 0) + 1
3796
occurs = tests.get(id, 0)
3798
not_found.append(id)
3800
duplicates.append(id)
3802
return not_found, duplicates
3805
class TestIdList(object):
3806
"""Test id list to filter a test suite.
3808
Relying on the assumption that test ids are built as:
3809
<module>[.<class>.<method>][(<param>+)], <module> being in python dotted
3810
notation, this class offers methods to :
3811
- avoid building a test suite for modules not refered to in the test list,
3812
- keep only the tests listed from the module test suite.
3815
def __init__(self, test_id_list):
3816
# When a test suite needs to be filtered against us we compare test ids
3817
# for equality, so a simple dict offers a quick and simple solution.
3818
self.tests = dict().fromkeys(test_id_list, True)
3820
# While unittest.TestCase have ids like:
3821
# <module>.<class>.<method>[(<param+)],
3822
# doctest.DocTestCase can have ids like:
3825
# <module>.<function>
3826
# <module>.<class>.<method>
3828
# Since we can't predict a test class from its name only, we settle on
3829
# a simple constraint: a test id always begins with its module name.
3832
for test_id in test_id_list:
3833
parts = test_id.split('.')
3834
mod_name = parts.pop(0)
3835
modules[mod_name] = True
3837
mod_name += '.' + part
3838
modules[mod_name] = True
3839
self.modules = modules
3841
def refers_to(self, module_name):
3842
"""Is there tests for the module or one of its sub modules."""
3843
return self.modules.has_key(module_name)
3845
def includes(self, test_id):
3846
return self.tests.has_key(test_id)
3849
class TestPrefixAliasRegistry(registry.Registry):
3850
"""A registry for test prefix aliases.
3852
This helps implement shorcuts for the --starting-with selftest
3853
option. Overriding existing prefixes is not allowed but not fatal (a
3854
warning will be emitted).
3857
def register(self, key, obj, help=None, info=None,
3858
override_existing=False):
3859
"""See Registry.register.
3861
Trying to override an existing alias causes a warning to be emitted,
3862
not a fatal execption.
3865
super(TestPrefixAliasRegistry, self).register(
3866
key, obj, help=help, info=info, override_existing=False)
3868
actual = self.get(key)
3870
'Test prefix alias %s is already used for %s, ignoring %s'
3871
% (key, actual, obj))
3873
def resolve_alias(self, id_start):
3874
"""Replace the alias by the prefix in the given string.
3876
Using an unknown prefix is an error to help catching typos.
3878
parts = id_start.split('.')
3880
parts[0] = self.get(parts[0])
3882
raise errors.BzrCommandError(
3883
'%s is not a known test prefix alias' % parts[0])
3884
return '.'.join(parts)
3887
test_prefix_alias_registry = TestPrefixAliasRegistry()
3888
"""Registry of test prefix aliases."""
3891
# This alias allows to detect typos ('bzrlin.') by making all valid test ids
3892
# appear prefixed ('bzrlib.' is "replaced" by 'bzrlib.').
3893
test_prefix_alias_registry.register('bzrlib', 'bzrlib')
3895
# Obvious highest levels prefixes, feel free to add your own via a plugin
3896
test_prefix_alias_registry.register('bd', 'bzrlib.doc')
3897
test_prefix_alias_registry.register('bu', 'bzrlib.utils')
3898
test_prefix_alias_registry.register('bt', 'bzrlib.tests')
3899
test_prefix_alias_registry.register('bb', 'bzrlib.tests.blackbox')
3900
test_prefix_alias_registry.register('bp', 'bzrlib.plugins')
3903
def _test_suite_testmod_names():
3904
"""Return the standard list of test module names to test."""
3907
'bzrlib.tests.blackbox',
3908
'bzrlib.tests.commands',
3909
'bzrlib.tests.per_branch',
3910
'bzrlib.tests.per_bzrdir',
3911
'bzrlib.tests.per_controldir',
3912
'bzrlib.tests.per_controldir_colo',
3913
'bzrlib.tests.per_foreign_vcs',
3914
'bzrlib.tests.per_interrepository',
3915
'bzrlib.tests.per_intertree',
3916
'bzrlib.tests.per_inventory',
3917
'bzrlib.tests.per_interbranch',
3918
'bzrlib.tests.per_lock',
3919
'bzrlib.tests.per_merger',
3920
'bzrlib.tests.per_transport',
3921
'bzrlib.tests.per_tree',
3922
'bzrlib.tests.per_pack_repository',
3923
'bzrlib.tests.per_repository',
3924
'bzrlib.tests.per_repository_chk',
3925
'bzrlib.tests.per_repository_reference',
3926
'bzrlib.tests.per_repository_vf',
3927
'bzrlib.tests.per_uifactory',
3928
'bzrlib.tests.per_versionedfile',
3929
'bzrlib.tests.per_workingtree',
3930
'bzrlib.tests.test__annotator',
3931
'bzrlib.tests.test__bencode',
3932
'bzrlib.tests.test__btree_serializer',
3933
'bzrlib.tests.test__chk_map',
3934
'bzrlib.tests.test__dirstate_helpers',
3935
'bzrlib.tests.test__groupcompress',
3936
'bzrlib.tests.test__known_graph',
3937
'bzrlib.tests.test__rio',
3938
'bzrlib.tests.test__simple_set',
3939
'bzrlib.tests.test__static_tuple',
3940
'bzrlib.tests.test__walkdirs_win32',
3941
'bzrlib.tests.test_ancestry',
3942
'bzrlib.tests.test_annotate',
3943
'bzrlib.tests.test_api',
3944
'bzrlib.tests.test_atomicfile',
3945
'bzrlib.tests.test_bad_files',
3946
'bzrlib.tests.test_bisect_multi',
3947
'bzrlib.tests.test_branch',
3948
'bzrlib.tests.test_branchbuilder',
3949
'bzrlib.tests.test_btree_index',
3950
'bzrlib.tests.test_bugtracker',
3951
'bzrlib.tests.test_bundle',
3952
'bzrlib.tests.test_bzrdir',
3953
'bzrlib.tests.test__chunks_to_lines',
3954
'bzrlib.tests.test_cache_utf8',
3955
'bzrlib.tests.test_chk_map',
3956
'bzrlib.tests.test_chk_serializer',
3957
'bzrlib.tests.test_chunk_writer',
3958
'bzrlib.tests.test_clean_tree',
3959
'bzrlib.tests.test_cleanup',
3960
'bzrlib.tests.test_cmdline',
3961
'bzrlib.tests.test_commands',
3962
'bzrlib.tests.test_commit',
3963
'bzrlib.tests.test_commit_merge',
3964
'bzrlib.tests.test_config',
3965
'bzrlib.tests.test_conflicts',
3966
'bzrlib.tests.test_controldir',
3967
'bzrlib.tests.test_counted_lock',
3968
'bzrlib.tests.test_crash',
3969
'bzrlib.tests.test_decorators',
3970
'bzrlib.tests.test_delta',
3971
'bzrlib.tests.test_debug',
3972
'bzrlib.tests.test_diff',
3973
'bzrlib.tests.test_directory_service',
3974
'bzrlib.tests.test_dirstate',
3975
'bzrlib.tests.test_email_message',
3976
'bzrlib.tests.test_eol_filters',
3977
'bzrlib.tests.test_errors',
3978
'bzrlib.tests.test_estimate_compressed_size',
3979
'bzrlib.tests.test_export',
3980
'bzrlib.tests.test_export_pot',
3981
'bzrlib.tests.test_extract',
3982
'bzrlib.tests.test_features',
3983
'bzrlib.tests.test_fetch',
3984
'bzrlib.tests.test_fixtures',
3985
'bzrlib.tests.test_fifo_cache',
3986
'bzrlib.tests.test_filters',
3987
'bzrlib.tests.test_filter_tree',
3988
'bzrlib.tests.test_ftp_transport',
3989
'bzrlib.tests.test_foreign',
3990
'bzrlib.tests.test_generate_docs',
3991
'bzrlib.tests.test_generate_ids',
3992
'bzrlib.tests.test_globbing',
3993
'bzrlib.tests.test_gpg',
3994
'bzrlib.tests.test_graph',
3995
'bzrlib.tests.test_groupcompress',
3996
'bzrlib.tests.test_hashcache',
3997
'bzrlib.tests.test_help',
3998
'bzrlib.tests.test_hooks',
3999
'bzrlib.tests.test_http',
4000
'bzrlib.tests.test_http_response',
4001
'bzrlib.tests.test_https_ca_bundle',
4002
'bzrlib.tests.test_https_urllib',
4003
'bzrlib.tests.test_i18n',
4004
'bzrlib.tests.test_identitymap',
4005
'bzrlib.tests.test_ignores',
4006
'bzrlib.tests.test_index',
4007
'bzrlib.tests.test_import_tariff',
4008
'bzrlib.tests.test_info',
4009
'bzrlib.tests.test_inv',
4010
'bzrlib.tests.test_inventory_delta',
4011
'bzrlib.tests.test_knit',
4012
'bzrlib.tests.test_lazy_import',
4013
'bzrlib.tests.test_lazy_regex',
4014
'bzrlib.tests.test_library_state',
4015
'bzrlib.tests.test_lock',
4016
'bzrlib.tests.test_lockable_files',
4017
'bzrlib.tests.test_lockdir',
4018
'bzrlib.tests.test_log',
4019
'bzrlib.tests.test_lru_cache',
4020
'bzrlib.tests.test_lsprof',
4021
'bzrlib.tests.test_mail_client',
4022
'bzrlib.tests.test_matchers',
4023
'bzrlib.tests.test_memorytree',
4024
'bzrlib.tests.test_merge',
4025
'bzrlib.tests.test_merge3',
4026
'bzrlib.tests.test_merge_core',
4027
'bzrlib.tests.test_merge_directive',
4028
'bzrlib.tests.test_mergetools',
4029
'bzrlib.tests.test_missing',
4030
'bzrlib.tests.test_msgeditor',
4031
'bzrlib.tests.test_multiparent',
4032
'bzrlib.tests.test_mutabletree',
4033
'bzrlib.tests.test_nonascii',
4034
'bzrlib.tests.test_options',
4035
'bzrlib.tests.test_osutils',
4036
'bzrlib.tests.test_osutils_encodings',
4037
'bzrlib.tests.test_pack',
4038
'bzrlib.tests.test_patch',
4039
'bzrlib.tests.test_patches',
4040
'bzrlib.tests.test_permissions',
4041
'bzrlib.tests.test_plugins',
4042
'bzrlib.tests.test_progress',
4043
'bzrlib.tests.test_pyutils',
4044
'bzrlib.tests.test_read_bundle',
4045
'bzrlib.tests.test_reconcile',
4046
'bzrlib.tests.test_reconfigure',
4047
'bzrlib.tests.test_registry',
4048
'bzrlib.tests.test_remote',
4049
'bzrlib.tests.test_rename_map',
4050
'bzrlib.tests.test_repository',
4051
'bzrlib.tests.test_revert',
4052
'bzrlib.tests.test_revision',
4053
'bzrlib.tests.test_revisionspec',
4054
'bzrlib.tests.test_revisiontree',
4055
'bzrlib.tests.test_rio',
4056
'bzrlib.tests.test_rules',
4057
'bzrlib.tests.test_url_policy_open',
4058
'bzrlib.tests.test_sampler',
4059
'bzrlib.tests.test_scenarios',
4060
'bzrlib.tests.test_script',
4061
'bzrlib.tests.test_selftest',
4062
'bzrlib.tests.test_serializer',
4063
'bzrlib.tests.test_setup',
4064
'bzrlib.tests.test_sftp_transport',
4065
'bzrlib.tests.test_shelf',
4066
'bzrlib.tests.test_shelf_ui',
4067
'bzrlib.tests.test_smart',
4068
'bzrlib.tests.test_smart_add',
4069
'bzrlib.tests.test_smart_request',
4070
'bzrlib.tests.test_smart_signals',
4071
'bzrlib.tests.test_smart_transport',
4072
'bzrlib.tests.test_smtp_connection',
4073
'bzrlib.tests.test_source',
4074
'bzrlib.tests.test_ssh_transport',
4075
'bzrlib.tests.test_status',
4076
'bzrlib.tests.test_store',
4077
'bzrlib.tests.test_strace',
4078
'bzrlib.tests.test_subsume',
4079
'bzrlib.tests.test_switch',
4080
'bzrlib.tests.test_symbol_versioning',
4081
'bzrlib.tests.test_tag',
4082
'bzrlib.tests.test_test_server',
4083
'bzrlib.tests.test_testament',
4084
'bzrlib.tests.test_textfile',
4085
'bzrlib.tests.test_textmerge',
4086
'bzrlib.tests.test_cethread',
4087
'bzrlib.tests.test_timestamp',
4088
'bzrlib.tests.test_trace',
4089
'bzrlib.tests.test_transactions',
4090
'bzrlib.tests.test_transform',
4091
'bzrlib.tests.test_transport',
4092
'bzrlib.tests.test_transport_log',
4093
'bzrlib.tests.test_tree',
4094
'bzrlib.tests.test_treebuilder',
4095
'bzrlib.tests.test_treeshape',
4096
'bzrlib.tests.test_tsort',
4097
'bzrlib.tests.test_tuned_gzip',
4098
'bzrlib.tests.test_ui',
4099
'bzrlib.tests.test_uncommit',
4100
'bzrlib.tests.test_upgrade',
4101
'bzrlib.tests.test_upgrade_stacked',
4102
'bzrlib.tests.test_urlutils',
4103
'bzrlib.tests.test_utextwrap',
4104
'bzrlib.tests.test_version',
4105
'bzrlib.tests.test_version_info',
4106
'bzrlib.tests.test_versionedfile',
4107
'bzrlib.tests.test_vf_search',
4108
'bzrlib.tests.test_weave',
4109
'bzrlib.tests.test_whitebox',
4110
'bzrlib.tests.test_win32utils',
4111
'bzrlib.tests.test_workingtree',
4112
'bzrlib.tests.test_workingtree_4',
4113
'bzrlib.tests.test_wsgi',
4114
'bzrlib.tests.test_xml',
4118
def _test_suite_modules_to_doctest():
4119
"""Return the list of modules to doctest."""
4121
# GZ 2009-03-31: No docstrings with -OO so there's nothing to doctest
4125
'bzrlib.branchbuilder',
4126
'bzrlib.decorators',
4128
'bzrlib.iterablefile',
4133
'bzrlib.symbol_versioning',
4135
'bzrlib.tests.fixtures',
4137
'bzrlib.transport.http',
4138
'bzrlib.version_info_formats.format_custom',
4142
def test_suite(keep_only=None, starting_with=None):
4143
"""Build and return TestSuite for the whole of bzrlib.
4145
:param keep_only: A list of test ids limiting the suite returned.
4147
:param starting_with: An id limiting the suite returned to the tests
4150
This function can be replaced if you need to change the default test
4151
suite on a global basis, but it is not encouraged.
4154
loader = TestUtil.TestLoader()
4156
if keep_only is not None:
4157
id_filter = TestIdList(keep_only)
4159
# We take precedence over keep_only because *at loading time* using
4160
# both options means we will load less tests for the same final result.
4161
def interesting_module(name):
4162
for start in starting_with:
4164
# Either the module name starts with the specified string
4165
name.startswith(start)
4166
# or it may contain tests starting with the specified string
4167
or start.startswith(name)
4171
loader = TestUtil.FilteredByModuleTestLoader(interesting_module)
4173
elif keep_only is not None:
4174
loader = TestUtil.FilteredByModuleTestLoader(id_filter.refers_to)
4175
def interesting_module(name):
4176
return id_filter.refers_to(name)
4179
loader = TestUtil.TestLoader()
4180
def interesting_module(name):
4181
# No filtering, all modules are interesting
4184
suite = loader.suiteClass()
4186
# modules building their suite with loadTestsFromModuleNames
4187
suite.addTest(loader.loadTestsFromModuleNames(_test_suite_testmod_names()))
4189
for mod in _test_suite_modules_to_doctest():
4190
if not interesting_module(mod):
4191
# No tests to keep here, move along
4194
# note that this really does mean "report only" -- doctest
4195
# still runs the rest of the examples
4196
doc_suite = IsolatedDocTestSuite(
4197
mod, optionflags=doctest.REPORT_ONLY_FIRST_FAILURE)
4198
except ValueError, e:
4199
print '**failed to get doctest for: %s\n%s' % (mod, e)
4201
if len(doc_suite._tests) == 0:
4202
raise errors.BzrError("no doctests found in %s" % (mod,))
4203
suite.addTest(doc_suite)
4205
default_encoding = sys.getdefaultencoding()
4206
for name, plugin in _mod_plugin.plugins().items():
4207
if not interesting_module(plugin.module.__name__):
4209
plugin_suite = plugin.test_suite()
4210
# We used to catch ImportError here and turn it into just a warning,
4211
# but really if you don't have --no-plugins this should be a failure.
4212
# mbp 20080213 - see http://bugs.launchpad.net/bugs/189771
4213
if plugin_suite is None:
4214
plugin_suite = plugin.load_plugin_tests(loader)
4215
if plugin_suite is not None:
4216
suite.addTest(plugin_suite)
4217
if default_encoding != sys.getdefaultencoding():
4219
'Plugin "%s" tried to reset default encoding to: %s', name,
4220
sys.getdefaultencoding())
4222
sys.setdefaultencoding(default_encoding)
4224
if keep_only is not None:
4225
# Now that the referred modules have loaded their tests, keep only the
4227
suite = filter_suite_by_id_list(suite, id_filter)
4228
# Do some sanity checks on the id_list filtering
4229
not_found, duplicates = suite_matches_id_list(suite, keep_only)
4231
# The tester has used both keep_only and starting_with, so he is
4232
# already aware that some tests are excluded from the list, there
4233
# is no need to tell him which.
4236
# Some tests mentioned in the list are not in the test suite. The
4237
# list may be out of date, report to the tester.
4238
for id in not_found:
4239
trace.warning('"%s" not found in the test suite', id)
4240
for id in duplicates:
4241
trace.warning('"%s" is used as an id by several tests', id)
4246
def multiply_scenarios(*scenarios):
4247
"""Multiply two or more iterables of scenarios.
4249
It is safe to pass scenario generators or iterators.
4251
:returns: A list of compound scenarios: the cross-product of all
4252
scenarios, with the names concatenated and the parameters
4255
return reduce(_multiply_two_scenarios, map(list, scenarios))
4258
def _multiply_two_scenarios(scenarios_left, scenarios_right):
4259
"""Multiply two sets of scenarios.
4261
:returns: the cartesian product of the two sets of scenarios, that is
4262
a scenario for every possible combination of a left scenario and a
4266
('%s,%s' % (left_name, right_name),
4267
dict(left_dict.items() + right_dict.items()))
4268
for left_name, left_dict in scenarios_left
4269
for right_name, right_dict in scenarios_right]
4272
def multiply_tests(tests, scenarios, result):
4273
"""Multiply tests_list by scenarios into result.
4275
This is the core workhorse for test parameterisation.
4277
Typically the load_tests() method for a per-implementation test suite will
4278
call multiply_tests and return the result.
4280
:param tests: The tests to parameterise.
4281
:param scenarios: The scenarios to apply: pairs of (scenario_name,
4282
scenario_param_dict).
4283
:param result: A TestSuite to add created tests to.
4285
This returns the passed in result TestSuite with the cross product of all
4286
the tests repeated once for each scenario. Each test is adapted by adding
4287
the scenario name at the end of its id(), and updating the test object's
4288
__dict__ with the scenario_param_dict.
4290
>>> import bzrlib.tests.test_sampler
4291
>>> r = multiply_tests(
4292
... bzrlib.tests.test_sampler.DemoTest('test_nothing'),
4293
... [('one', dict(param=1)),
4294
... ('two', dict(param=2))],
4295
... TestUtil.TestSuite())
4296
>>> tests = list(iter_suite_tests(r))
4300
'bzrlib.tests.test_sampler.DemoTest.test_nothing(one)'
4306
for test in iter_suite_tests(tests):
4307
apply_scenarios(test, scenarios, result)
4311
def apply_scenarios(test, scenarios, result):
4312
"""Apply the scenarios in scenarios to test and add to result.
4314
:param test: The test to apply scenarios to.
4315
:param scenarios: An iterable of scenarios to apply to test.
4317
:seealso: apply_scenario
4319
for scenario in scenarios:
4320
result.addTest(apply_scenario(test, scenario))
4324
def apply_scenario(test, scenario):
4325
"""Copy test and apply scenario to it.
4327
:param test: A test to adapt.
4328
:param scenario: A tuple describing the scenarion.
4329
The first element of the tuple is the new test id.
4330
The second element is a dict containing attributes to set on the
4332
:return: The adapted test.
4334
new_id = "%s(%s)" % (test.id(), scenario[0])
4335
new_test = clone_test(test, new_id)
4336
for name, value in scenario[1].items():
4337
setattr(new_test, name, value)
4341
def clone_test(test, new_id):
4342
"""Clone a test giving it a new id.
4344
:param test: The test to clone.
4345
:param new_id: The id to assign to it.
4346
:return: The new test.
4348
new_test = copy.copy(test)
4349
new_test.id = lambda: new_id
4350
# XXX: Workaround <https://bugs.launchpad.net/testtools/+bug/637725>, which
4351
# causes cloned tests to share the 'details' dict. This makes it hard to
4352
# read the test output for parameterized tests, because tracebacks will be
4353
# associated with irrelevant tests.
4355
details = new_test._TestCase__details
4356
except AttributeError:
4357
# must be a different version of testtools than expected. Do nothing.
4360
# Reset the '__details' dict.
4361
new_test._TestCase__details = {}
4365
def permute_tests_for_extension(standard_tests, loader, py_module_name,
4367
"""Helper for permutating tests against an extension module.
4369
This is meant to be used inside a modules 'load_tests()' function. It will
4370
create 2 scenarios, and cause all tests in the 'standard_tests' to be run
4371
against both implementations. Setting 'test.module' to the appropriate
4372
module. See bzrlib.tests.test__chk_map.load_tests as an example.
4374
:param standard_tests: A test suite to permute
4375
:param loader: A TestLoader
4376
:param py_module_name: The python path to a python module that can always
4377
be loaded, and will be considered the 'python' implementation. (eg
4378
'bzrlib._chk_map_py')
4379
:param ext_module_name: The python path to an extension module. If the
4380
module cannot be loaded, a single test will be added, which notes that
4381
the module is not available. If it can be loaded, all standard_tests
4382
will be run against that module.
4383
:return: (suite, feature) suite is a test-suite that has all the permuted
4384
tests. feature is the Feature object that can be used to determine if
4385
the module is available.
4388
from bzrlib.tests.features import ModuleAvailableFeature
4389
py_module = pyutils.get_named_object(py_module_name)
4391
('python', {'module': py_module}),
4393
suite = loader.suiteClass()
4394
feature = ModuleAvailableFeature(ext_module_name)
4395
if feature.available():
4396
scenarios.append(('C', {'module': feature.module}))
4398
# the compiled module isn't available, so we add a failing test
4399
class FailWithoutFeature(TestCase):
4400
def test_fail(self):
4401
self.requireFeature(feature)
4402
suite.addTest(loader.loadTestsFromTestCase(FailWithoutFeature))
4403
result = multiply_tests(standard_tests, scenarios, suite)
4404
return result, feature
4407
def _rmtree_temp_dir(dirname, test_id=None):
4408
# If LANG=C we probably have created some bogus paths
4409
# which rmtree(unicode) will fail to delete
4410
# so make sure we are using rmtree(str) to delete everything
4411
# except on win32, where rmtree(str) will fail
4412
# since it doesn't have the property of byte-stream paths
4413
# (they are either ascii or mbcs)
4414
if sys.platform == 'win32':
4415
# make sure we are using the unicode win32 api
4416
dirname = unicode(dirname)
4418
dirname = dirname.encode(sys.getfilesystemencoding())
4420
osutils.rmtree(dirname)
4422
# We don't want to fail here because some useful display will be lost
4423
# otherwise. Polluting the tmp dir is bad, but not giving all the
4424
# possible info to the test runner is even worse.
4426
ui.ui_factory.clear_term()
4427
sys.stderr.write('\nWhile running: %s\n' % (test_id,))
4428
# Ugly, but the last thing we want here is fail, so bear with it.
4429
printable_e = str(e).decode(osutils.get_user_encoding(), 'replace'
4430
).encode('ascii', 'replace')
4431
sys.stderr.write('Unable to remove testing dir %s\n%s'
4432
% (os.path.basename(dirname), printable_e))
4435
def probe_unicode_in_user_encoding():
4436
"""Try to encode several unicode strings to use in unicode-aware tests.
4437
Return first successfull match.
4439
:return: (unicode value, encoded plain string value) or (None, None)
4441
possible_vals = [u'm\xb5', u'\xe1', u'\u0410']
4442
for uni_val in possible_vals:
4444
str_val = uni_val.encode(osutils.get_user_encoding())
4445
except UnicodeEncodeError:
4446
# Try a different character
4449
return uni_val, str_val
4453
def probe_bad_non_ascii(encoding):
4454
"""Try to find [bad] character with code [128..255]
4455
that cannot be decoded to unicode in some encoding.
4456
Return None if all non-ascii characters is valid
4459
for i in xrange(128, 256):
4462
char.decode(encoding)
4463
except UnicodeDecodeError:
4468
# Only define SubUnitBzrRunner if subunit is available.
4470
from subunit import TestProtocolClient
4471
from subunit.test_results import AutoTimingTestResultDecorator
4472
class SubUnitBzrProtocolClient(TestProtocolClient):
4474
def stopTest(self, test):
4475
super(SubUnitBzrProtocolClient, self).stopTest(test)
4476
_clear__type_equality_funcs(test)
4478
def addSuccess(self, test, details=None):
4479
# The subunit client always includes the details in the subunit
4480
# stream, but we don't want to include it in ours.
4481
if details is not None and 'log' in details:
4483
return super(SubUnitBzrProtocolClient, self).addSuccess(
4486
class SubUnitBzrRunner(TextTestRunner):
4487
def run(self, test):
4488
result = AutoTimingTestResultDecorator(
4489
SubUnitBzrProtocolClient(self.stream))
4496
# API compatibility for old plugins; see bug 892622.
4499
'HTTPServerFeature',
4500
'ModuleAvailableFeature',
4501
'HTTPSServerFeature', 'SymlinkFeature', 'HardlinkFeature',
4502
'OsFifoFeature', 'UnicodeFilenameFeature',
4503
'ByteStringNamedFilesystem', 'UTF8Filesystem',
4504
'BreakinFeature', 'CaseInsCasePresFilenameFeature',
4505
'CaseInsensitiveFilesystemFeature', 'case_sensitive_filesystem_feature',
4506
'posix_permissions_feature',
4508
globals()[name] = _CompatabilityThunkFeature(
4509
symbol_versioning.deprecated_in((2, 5, 0)),
4510
'bzrlib.tests', name,
4511
name, 'bzrlib.tests.features')
4514
for (old_name, new_name) in [
4515
('UnicodeFilename', 'UnicodeFilenameFeature'),
4517
globals()[name] = _CompatabilityThunkFeature(
4518
symbol_versioning.deprecated_in((2, 5, 0)),
4519
'bzrlib.tests', old_name,
4520
new_name, 'bzrlib.tests.features')
added
removed
bzrlib/tests/__init__.py
