1
# Copyright (C) 2005-2011 Canonical Ltd
1
# Copyright (C) 2005 by Canonical Ltd
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
5
5
# the Free Software Foundation; either version 2 of the License, or
6
6
# (at your option) any later version.
8
8
# This program is distributed in the hope that it will be useful,
9
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
11
# GNU General Public License for more details.
13
13
# You should have received a copy of the GNU General Public License
14
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
# TODO: Perhaps there should be an API to find out if bzr running under the
20
# test suite -- some plugins might want to avoid making intrusive changes if
21
# this is the case. However, we want behaviour under to test to diverge as
22
# little as possible, so this should be used rarely if it's added at all.
23
# (Suggestion from j-a-meinel, 2005-11-24)
25
# NOTE: Some classes in here use camelCaseNaming() rather than
26
# underscore_naming(). That's for consistency with unittest; it's not the
27
# general style of bzrlib. Please continue that consistency when adding e.g.
28
# new assertFoo() methods.
33
from cStringIO import StringIO
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
56
# nb: check this before importing anything else from within it
57
_testtools_version = getattr(testtools, '__version__', ())
58
if _testtools_version < (0, 9, 5):
59
raise ImportError("need at least testtools 0.9.5: %s is %r"
60
% (testtools.__file__, _testtools_version))
61
from testtools import content
68
commands as _mod_commands,
77
plugin as _mod_plugin,
84
transport as _mod_transport,
90
# lsprof not available
24
from warnings import warn
25
from cStringIO import StringIO
28
import bzrlib.commands
35
MODULES_TO_DOCTEST = []
37
from logging import debug, warning, error
39
class CommandFailed(Exception):
92
from bzrlib.smart import client, request
93
from bzrlib.transport import (
97
from bzrlib.tests import (
102
from bzrlib.ui import NullProgressView
103
from bzrlib.ui.text import TextUIFactory
105
# Mark this python module as being part of the implementation
106
# of unittest: this gives us better tracebacks where the last
107
# shown frame is the test code, not our assertXYZ.
110
default_transport = test_server.LocalURLServer
113
_unitialized_attr = object()
114
"""A sentinel needed to act as a default value in a method signature."""
117
# Subunit result codes, defined here to prevent a hard dependency on subunit.
121
# These are intentionally brought into this namespace. That way plugins, etc
122
# can just "from bzrlib.tests import TestCase, TestLoader, etc"
123
TestSuite = TestUtil.TestSuite
124
TestLoader = TestUtil.TestLoader
126
# Tests should run in a clean and clearly defined environment. The goal is to
127
# keep them isolated from the running environment as mush as possible. The test
128
# framework ensures the variables defined below are set (or deleted if the
129
# value is None) before a test is run and reset to their original value after
130
# the test is run. Generally if some code depends on an environment variable,
131
# the tests should start without this variable in the environment. There are a
132
# few exceptions but you shouldn't violate this rule lightly.
136
# bzr now uses the Win32 API and doesn't rely on APPDATA, but the
137
# tests do check our impls match APPDATA
138
'BZR_EDITOR': None, # test_msgeditor manipulates this variable
142
'BZREMAIL': None, # may still be present in the environment
143
'EMAIL': 'jrandom@example.com', # set EMAIL as bzr does not guess
144
'BZR_PROGRESS_BAR': None,
145
# This should trap leaks to ~/.bzr.log. This occurs when tests use TestCase
146
# as a base class instead of TestCaseInTempDir. Tests inheriting from
147
# TestCase should not use disk resources, BZR_LOG is one.
148
'BZR_LOG': '/you-should-use-TestCaseInTempDir-if-you-need-a-log-file',
149
'BZR_PLUGIN_PATH': None,
150
'BZR_DISABLE_PLUGINS': None,
151
'BZR_PLUGINS_AT': None,
152
'BZR_CONCURRENCY': None,
153
# Make sure that any text ui tests are consistent regardless of
154
# the environment the test case is run in; you may want tests that
155
# test other combinations. 'dumb' is a reasonable guess for tests
156
# going to a pipe or a StringIO.
162
'SSH_AUTH_SOCK': None,
172
# Nobody cares about ftp_proxy, FTP_PROXY AFAIK. So far at
173
# least. If you do (care), please update this comment
177
'BZR_REMOTE_PATH': None,
178
# Generally speaking, we don't want apport reporting on crashes in
179
# the test envirnoment unless we're specifically testing apport,
180
# so that it doesn't leak into the real system environment. We
181
# use an env var so it propagates to subprocesses.
182
'APPORT_DISABLE': '1',
186
def override_os_environ(test, env=None):
187
"""Modify os.environ keeping a copy.
189
:param test: A test instance
191
:param env: A dict containing variable definitions to be installed
194
env = isolated_environ
195
test._original_os_environ = dict([(var, value)
196
for var, value in os.environ.iteritems()])
197
for var, value in env.iteritems():
198
osutils.set_or_unset_env(var, value)
199
if var not in test._original_os_environ:
200
# The var is new, add it with a value of None, so
201
# restore_os_environ will delete it
202
test._original_os_environ[var] = None
205
def restore_os_environ(test):
206
"""Restore os.environ to its original state.
208
:param test: A test instance previously passed to override_os_environ.
210
for var, value in test._original_os_environ.iteritems():
211
# Restore the original value (or delete it if the value has been set to
212
# None in override_os_environ).
213
osutils.set_or_unset_env(var, value)
216
class ExtendedTestResult(testtools.TextTestResult):
217
"""Accepts, reports and accumulates the results of running tests.
219
Compared to the unittest version this class adds support for
220
profiling, benchmarking, stopping as soon as a test fails, and
221
skipping tests. There are further-specialized subclasses for
222
different types of display.
224
When a test finishes, in whatever way, it calls one of the addSuccess,
225
addFailure or addError classes. These in turn may redirect to a more
226
specific case for the special test results supported by our extended
229
Note that just one of these objects is fed the results from many tests.
234
def __init__(self, stream, descriptions, verbosity,
238
"""Construct new TestResult.
240
:param bench_history: Optionally, a writable file object to accumulate
243
testtools.TextTestResult.__init__(self, stream)
244
if bench_history is not None:
245
from bzrlib.version import _get_bzr_source_tree
246
src_tree = _get_bzr_source_tree()
249
revision_id = src_tree.get_parent_ids()[0]
251
# XXX: if this is a brand new tree, do the same as if there
255
# XXX: If there's no branch, what should we do?
257
bench_history.write("--date %s %s\n" % (time.time(), revision_id))
258
self._bench_history = bench_history
259
self.ui = ui.ui_factory
262
self.failure_count = 0
263
self.known_failure_count = 0
265
self.not_applicable_count = 0
266
self.unsupported = {}
268
self._overall_start_time = time.time()
269
self._strict = strict
270
self._first_thread_leaker_id = None
271
self._tests_leaking_threads_count = 0
272
self._traceback_from_test = None
274
def stopTestRun(self):
277
stopTime = time.time()
278
timeTaken = stopTime - self.startTime
279
# GZ 2010-07-19: Seems testtools has no printErrors method, and though
280
# the parent class method is similar have to duplicate
281
self._show_list('ERROR', self.errors)
282
self._show_list('FAIL', self.failures)
283
self.stream.write(self.sep2)
284
self.stream.write("%s %d test%s in %.3fs\n\n" % (actionTaken,
285
run, run != 1 and "s" or "", timeTaken))
286
if not self.wasSuccessful():
287
self.stream.write("FAILED (")
288
failed, errored = map(len, (self.failures, self.errors))
290
self.stream.write("failures=%d" % failed)
292
if failed: self.stream.write(", ")
293
self.stream.write("errors=%d" % errored)
294
if self.known_failure_count:
295
if failed or errored: self.stream.write(", ")
296
self.stream.write("known_failure_count=%d" %
297
self.known_failure_count)
298
self.stream.write(")\n")
300
if self.known_failure_count:
301
self.stream.write("OK (known_failures=%d)\n" %
302
self.known_failure_count)
304
self.stream.write("OK\n")
305
if self.skip_count > 0:
306
skipped = self.skip_count
307
self.stream.write('%d test%s skipped\n' %
308
(skipped, skipped != 1 and "s" or ""))
310
for feature, count in sorted(self.unsupported.items()):
311
self.stream.write("Missing feature '%s' skipped %d tests.\n" %
314
ok = self.wasStrictlySuccessful()
316
ok = self.wasSuccessful()
317
if self._first_thread_leaker_id:
319
'%s is leaking threads among %d leaking tests.\n' % (
320
self._first_thread_leaker_id,
321
self._tests_leaking_threads_count))
322
# We don't report the main thread as an active one.
324
'%d non-main threads were left active in the end.\n'
325
% (len(self._active_threads) - 1))
327
def getDescription(self, test):
330
def _extractBenchmarkTime(self, testCase, details=None):
331
"""Add a benchmark time for the current test case."""
332
if details and 'benchtime' in details:
333
return float(''.join(details['benchtime'].iter_bytes()))
334
return getattr(testCase, "_benchtime", None)
336
def _elapsedTestTimeString(self):
337
"""Return a time string for the overall time the current test has taken."""
338
return self._formatTime(self._delta_to_float(
339
self._now() - self._start_datetime))
341
def _testTimeString(self, testCase):
342
benchmark_time = self._extractBenchmarkTime(testCase)
343
if benchmark_time is not None:
344
return self._formatTime(benchmark_time) + "*"
346
return self._elapsedTestTimeString()
348
def _formatTime(self, seconds):
349
"""Format seconds as milliseconds with leading spaces."""
350
# some benchmarks can take thousands of seconds to run, so we need 8
352
return "%8dms" % (1000 * seconds)
354
def _shortened_test_description(self, test):
356
what = re.sub(r'^bzrlib\.tests\.', '', what)
359
# GZ 2010-10-04: Cloned tests may end up harmlessly calling this method
360
# multiple times in a row, because the handler is added for
361
# each test but the container list is shared between cases.
362
# See lp:498869 lp:625574 and lp:637725 for background.
363
def _record_traceback_from_test(self, exc_info):
364
"""Store the traceback from passed exc_info tuple till"""
365
self._traceback_from_test = exc_info[2]
367
def startTest(self, test):
368
super(ExtendedTestResult, self).startTest(test)
372
self.report_test_start(test)
373
test.number = self.count
374
self._recordTestStartTime()
375
# Make testtools cases give us the real traceback on failure
376
addOnException = getattr(test, "addOnException", None)
377
if addOnException is not None:
378
addOnException(self._record_traceback_from_test)
379
# Only check for thread leaks on bzrlib derived test cases
380
if isinstance(test, TestCase):
381
test.addCleanup(self._check_leaked_threads, test)
383
def stopTest(self, test):
384
super(ExtendedTestResult, self).stopTest(test)
385
# Manually break cycles, means touching various private things but hey
386
getDetails = getattr(test, "getDetails", None)
387
if getDetails is not None:
389
type_equality_funcs = getattr(test, "_type_equality_funcs", None)
390
if type_equality_funcs is not None:
391
type_equality_funcs.clear()
392
self._traceback_from_test = None
394
def startTests(self):
395
self.report_tests_starting()
396
self._active_threads = threading.enumerate()
398
def _check_leaked_threads(self, test):
399
"""See if any threads have leaked since last call
401
A sample of live threads is stored in the _active_threads attribute,
402
when this method runs it compares the current live threads and any not
403
in the previous sample are treated as having leaked.
405
now_active_threads = set(threading.enumerate())
406
threads_leaked = now_active_threads.difference(self._active_threads)
408
self._report_thread_leak(test, threads_leaked, now_active_threads)
409
self._tests_leaking_threads_count += 1
410
if self._first_thread_leaker_id is None:
411
self._first_thread_leaker_id = test.id()
412
self._active_threads = now_active_threads
414
def _recordTestStartTime(self):
415
"""Record that a test has started."""
416
self._start_datetime = self._now()
418
def addError(self, test, err):
419
"""Tell result that test finished with an error.
421
Called from the TestCase run() method when the test
422
fails with an unexpected error.
424
self._post_mortem(self._traceback_from_test)
425
super(ExtendedTestResult, self).addError(test, err)
426
self.error_count += 1
427
self.report_error(test, err)
431
def addFailure(self, test, err):
432
"""Tell result that test failed.
434
Called from the TestCase run() method when the test
435
fails because e.g. an assert() method failed.
437
self._post_mortem(self._traceback_from_test)
438
super(ExtendedTestResult, self).addFailure(test, err)
439
self.failure_count += 1
440
self.report_failure(test, err)
444
def addSuccess(self, test, details=None):
445
"""Tell result that test completed successfully.
447
Called from the TestCase run()
449
if self._bench_history is not None:
450
benchmark_time = self._extractBenchmarkTime(test, details)
451
if benchmark_time is not None:
452
self._bench_history.write("%s %s\n" % (
453
self._formatTime(benchmark_time),
455
self.report_success(test)
456
super(ExtendedTestResult, self).addSuccess(test)
457
test._log_contents = ''
459
def addExpectedFailure(self, test, err):
460
self.known_failure_count += 1
461
self.report_known_failure(test, err)
463
def addUnexpectedSuccess(self, test, details=None):
464
"""Tell result the test unexpectedly passed, counting as a failure
466
When the minimum version of testtools required becomes 0.9.8 this
467
can be updated to use the new handling there.
469
super(ExtendedTestResult, self).addFailure(test, details=details)
470
self.failure_count += 1
471
self.report_unexpected_success(test,
472
"".join(details["reason"].iter_text()))
476
def addNotSupported(self, test, feature):
477
"""The test will not be run because of a missing feature.
479
# this can be called in two different ways: it may be that the
480
# test started running, and then raised (through requireFeature)
481
# UnavailableFeature. Alternatively this method can be called
482
# while probing for features before running the test code proper; in
483
# that case we will see startTest and stopTest, but the test will
484
# never actually run.
485
self.unsupported.setdefault(str(feature), 0)
486
self.unsupported[str(feature)] += 1
487
self.report_unsupported(test, feature)
489
def addSkip(self, test, reason):
490
"""A test has not run for 'reason'."""
492
self.report_skip(test, reason)
494
def addNotApplicable(self, test, reason):
495
self.not_applicable_count += 1
496
self.report_not_applicable(test, reason)
498
def _post_mortem(self, tb=None):
499
"""Start a PDB post mortem session."""
500
if os.environ.get('BZR_TEST_PDB', None):
504
def progress(self, offset, whence):
505
"""The test is adjusting the count of tests to run."""
506
if whence == SUBUNIT_SEEK_SET:
507
self.num_tests = offset
508
elif whence == SUBUNIT_SEEK_CUR:
509
self.num_tests += offset
511
raise errors.BzrError("Unknown whence %r" % whence)
513
def report_tests_starting(self):
514
"""Display information before the test run begins"""
515
if getattr(sys, 'frozen', None) is None:
516
bzr_path = osutils.realpath(sys.argv[0])
518
bzr_path = sys.executable
520
'bzr selftest: %s\n' % (bzr_path,))
523
bzrlib.__path__[0],))
525
' bzr-%s python-%s %s\n' % (
526
bzrlib.version_string,
527
bzrlib._format_version_tuple(sys.version_info),
528
platform.platform(aliased=1),
530
self.stream.write('\n')
532
def report_test_start(self, test):
533
"""Display information on the test just about to be run"""
535
def _report_thread_leak(self, test, leaked_threads, active_threads):
536
"""Display information on a test that leaked one or more threads"""
537
# GZ 2010-09-09: A leak summary reported separately from the general
538
# thread debugging would be nice. Tests under subunit
539
# need something not using stream, perhaps adding a
540
# testtools details object would be fitting.
541
if 'threads' in selftest_debug_flags:
542
self.stream.write('%s is leaking, active is now %d\n' %
543
(test.id(), len(active_threads)))
545
def startTestRun(self):
546
self.startTime = time.time()
548
def report_success(self, test):
551
def wasStrictlySuccessful(self):
552
if self.unsupported or self.known_failure_count:
554
return self.wasSuccessful()
557
class TextTestResult(ExtendedTestResult):
558
"""Displays progress and results of tests in text form"""
560
def __init__(self, stream, descriptions, verbosity,
565
ExtendedTestResult.__init__(self, stream, descriptions, verbosity,
566
bench_history, strict)
567
# We no longer pass them around, but just rely on the UIFactory stack
570
warnings.warn("Passing pb to TextTestResult is deprecated")
571
self.pb = self.ui.nested_progress_bar()
572
self.pb.show_pct = False
573
self.pb.show_spinner = False
574
self.pb.show_eta = False,
575
self.pb.show_count = False
576
self.pb.show_bar = False
577
self.pb.update_latency = 0
578
self.pb.show_transport_activity = False
580
def stopTestRun(self):
581
# called when the tests that are going to run have run
584
super(TextTestResult, self).stopTestRun()
586
def report_tests_starting(self):
587
super(TextTestResult, self).report_tests_starting()
588
self.pb.update('[test 0/%d] Starting' % (self.num_tests))
590
def _progress_prefix_text(self):
591
# the longer this text, the less space we have to show the test
593
a = '[%d' % self.count # total that have been run
594
# tests skipped as known not to be relevant are not important enough
596
## if self.skip_count:
597
## a += ', %d skip' % self.skip_count
598
## if self.known_failure_count:
599
## a += '+%dX' % self.known_failure_count
601
a +='/%d' % self.num_tests
603
runtime = time.time() - self._overall_start_time
605
a += '%dm%ds' % (runtime / 60, runtime % 60)
608
total_fail_count = self.error_count + self.failure_count
610
a += ', %d failed' % total_fail_count
611
# if self.unsupported:
612
# a += ', %d missing' % len(self.unsupported)
616
def report_test_start(self, test):
618
self._progress_prefix_text()
620
+ self._shortened_test_description(test))
622
def _test_description(self, test):
623
return self._shortened_test_description(test)
625
def report_error(self, test, err):
626
self.stream.write('ERROR: %s\n %s\n' % (
627
self._test_description(test),
631
def report_failure(self, test, err):
632
self.stream.write('FAIL: %s\n %s\n' % (
633
self._test_description(test),
637
def report_known_failure(self, test, err):
640
def report_unexpected_success(self, test, reason):
641
self.stream.write('FAIL: %s\n %s: %s\n' % (
642
self._test_description(test),
643
"Unexpected success. Should have failed",
647
def report_skip(self, test, reason):
650
def report_not_applicable(self, test, reason):
653
def report_unsupported(self, test, feature):
654
"""test cannot be run because feature is missing."""
657
class VerboseTestResult(ExtendedTestResult):
658
"""Produce long output, with one line per test run plus times"""
660
def _ellipsize_to_right(self, a_string, final_width):
661
"""Truncate and pad a string, keeping the right hand side"""
662
if len(a_string) > final_width:
663
result = '...' + a_string[3-final_width:]
666
return result.ljust(final_width)
668
def report_tests_starting(self):
669
self.stream.write('running %d tests...\n' % self.num_tests)
670
super(VerboseTestResult, self).report_tests_starting()
672
def report_test_start(self, test):
673
name = self._shortened_test_description(test)
674
width = osutils.terminal_width()
675
if width is not None:
676
# width needs space for 6 char status, plus 1 for slash, plus an
677
# 11-char time string, plus a trailing blank
678
# when NUMBERED_DIRS: plus 5 chars on test number, plus 1 char on
680
self.stream.write(self._ellipsize_to_right(name, width-18))
682
self.stream.write(name)
685
def _error_summary(self, err):
687
return '%s%s' % (indent, err[1])
689
def report_error(self, test, err):
690
self.stream.write('ERROR %s\n%s\n'
691
% (self._testTimeString(test),
692
self._error_summary(err)))
694
def report_failure(self, test, err):
695
self.stream.write(' FAIL %s\n%s\n'
696
% (self._testTimeString(test),
697
self._error_summary(err)))
699
def report_known_failure(self, test, err):
700
self.stream.write('XFAIL %s\n%s\n'
701
% (self._testTimeString(test),
702
self._error_summary(err)))
704
def report_unexpected_success(self, test, reason):
705
self.stream.write(' FAIL %s\n%s: %s\n'
706
% (self._testTimeString(test),
707
"Unexpected success. Should have failed",
710
def report_success(self, test):
711
self.stream.write(' OK %s\n' % self._testTimeString(test))
712
for bench_called, stats in getattr(test, '_benchcalls', []):
713
self.stream.write('LSProf output for %s(%s, %s)\n' % bench_called)
714
stats.pprint(file=self.stream)
715
# flush the stream so that we get smooth output. This verbose mode is
716
# used to show the output in PQM.
719
def report_skip(self, test, reason):
720
self.stream.write(' SKIP %s\n%s\n'
721
% (self._testTimeString(test), reason))
723
def report_not_applicable(self, test, reason):
724
self.stream.write(' N/A %s\n %s\n'
725
% (self._testTimeString(test), reason))
727
def report_unsupported(self, test, feature):
728
"""test cannot be run because feature is missing."""
729
self.stream.write("NODEP %s\n The feature '%s' is not available.\n"
730
%(self._testTimeString(test), feature))
733
class TextTestRunner(object):
734
stop_on_failure = False
742
result_decorators=None,
744
"""Create a TextTestRunner.
746
:param result_decorators: An optional list of decorators to apply
747
to the result object being used by the runner. Decorators are
748
applied left to right - the first element in the list is the
751
# stream may know claim to know to write unicode strings, but in older
752
# pythons this goes sufficiently wrong that it is a bad idea. (
753
# specifically a built in file with encoding 'UTF-8' will still try
754
# to encode using ascii.
755
new_encoding = osutils.get_terminal_encoding()
756
codec = codecs.lookup(new_encoding)
757
if type(codec) is tuple:
761
encode = codec.encode
762
# GZ 2010-09-08: Really we don't want to be writing arbitrary bytes,
763
# so should swap to the plain codecs.StreamWriter
764
stream = osutils.UnicodeOrBytesToBytesWriter(encode, stream,
766
stream.encoding = new_encoding
768
self.descriptions = descriptions
769
self.verbosity = verbosity
770
self._bench_history = bench_history
771
self._strict = strict
772
self._result_decorators = result_decorators or []
775
"Run the given test case or test suite."
776
if self.verbosity == 1:
777
result_class = TextTestResult
778
elif self.verbosity >= 2:
779
result_class = VerboseTestResult
780
original_result = result_class(self.stream,
783
bench_history=self._bench_history,
786
# Signal to result objects that look at stop early policy to stop,
787
original_result.stop_early = self.stop_on_failure
788
result = original_result
789
for decorator in self._result_decorators:
790
result = decorator(result)
791
result.stop_early = self.stop_on_failure
792
result.startTestRun()
797
# higher level code uses our extended protocol to determine
798
# what exit code to give.
799
return original_result
802
def iter_suite_tests(suite):
803
"""Return all tests in a suite, recursing through nested suites"""
804
if isinstance(suite, unittest.TestCase):
806
elif isinstance(suite, unittest.TestSuite):
808
for r in iter_suite_tests(item):
811
raise Exception('unknown type %r for object %r'
812
% (type(suite), suite))
815
TestSkipped = testtools.testcase.TestSkipped
818
class TestNotApplicable(TestSkipped):
819
"""A test is not applicable to the situation where it was run.
821
This is only normally raised by parameterized tests, if they find that
822
the instance they're constructed upon does not support one aspect
827
# traceback._some_str fails to format exceptions that have the default
828
# __str__ which does an implicit ascii conversion. However, repr() on those
829
# objects works, for all that its not quite what the doctor may have ordered.
830
def _clever_some_str(value):
835
return repr(value).replace('\\n', '\n')
837
return '<unprintable %s object>' % type(value).__name__
839
traceback._some_str = _clever_some_str
842
# deprecated - use self.knownFailure(), or self.expectFailure.
843
KnownFailure = testtools.testcase._ExpectedFailure
846
class UnavailableFeature(Exception):
847
"""A feature required for this test was not available.
849
This can be considered a specialised form of SkippedTest.
851
The feature should be used to construct the exception.
855
class StringIOWrapper(object):
856
"""A wrapper around cStringIO which just adds an encoding attribute.
858
Internally we can check sys.stdout to see what the output encoding
859
should be. However, cStringIO has no encoding attribute that we can
860
set. So we wrap it instead.
865
def __init__(self, s=None):
867
self.__dict__['_cstring'] = StringIO(s)
869
self.__dict__['_cstring'] = StringIO()
871
def __getattr__(self, name, getattr=getattr):
872
return getattr(self.__dict__['_cstring'], name)
874
def __setattr__(self, name, val):
875
if name == 'encoding':
876
self.__dict__['encoding'] = val
878
return setattr(self._cstring, name, val)
881
class TestUIFactory(TextUIFactory):
882
"""A UI Factory for testing.
884
Hide the progress bar but emit note()s.
886
Allows get_password to be tested without real tty attached.
888
See also CannedInputUIFactory which lets you provide programmatic input in
891
# TODO: Capture progress events at the model level and allow them to be
892
# observed by tests that care.
894
# XXX: Should probably unify more with CannedInputUIFactory or a
895
# particular configuration of TextUIFactory, or otherwise have a clearer
896
# idea of how they're supposed to be different.
897
# See https://bugs.launchpad.net/bzr/+bug/408213
899
def __init__(self, stdout=None, stderr=None, stdin=None):
900
if stdin is not None:
901
# We use a StringIOWrapper to be able to test various
902
# encodings, but the user is still responsible to
903
# encode the string and to set the encoding attribute
904
# of StringIOWrapper.
905
stdin = StringIOWrapper(stdin)
906
super(TestUIFactory, self).__init__(stdin, stdout, stderr)
908
def get_non_echoed_password(self):
909
"""Get password from stdin without trying to handle the echo mode"""
910
password = self.stdin.readline()
913
if password[-1] == '\n':
914
password = password[:-1]
917
def make_progress_view(self):
918
return NullProgressView()
921
def isolated_doctest_setUp(test):
922
override_os_environ(test)
925
def isolated_doctest_tearDown(test):
926
restore_os_environ(test)
929
def IsolatedDocTestSuite(*args, **kwargs):
930
"""Overrides doctest.DocTestSuite to handle isolation.
932
The method is really a factory and users are expected to use it as such.
935
kwargs['setUp'] = isolated_doctest_setUp
936
kwargs['tearDown'] = isolated_doctest_tearDown
937
return doctest.DocTestSuite(*args, **kwargs)
940
class TestCase(testtools.TestCase):
42
class TestCase(unittest.TestCase):
941
43
"""Base class for bzr unit tests.
943
Tests that need access to disk resources should subclass
45
Tests that need access to disk resources should subclass
944
46
TestCaseInTempDir not TestCase.
946
48
Error and debug log messages are redirected from their usual
947
49
location into a temporary file, the contents of which can be
948
retrieved by _get_log(). We use a real OS file, not an in-memory object,
949
so that it can also capture file IO. When the test completes this file
950
is read into memory and removed from disk.
50
retrieved by _get_log().
952
52
There are also convenience functions to invoke bzr's command-line
953
routine, and to build and check bzr trees.
955
In addition to the usual method of overriding tearDown(), this class also
956
allows subclasses to register cleanup functions via addCleanup, which are
957
run in order as the object is torn down. It's less likely this will be
958
accidentally overlooked.
962
# record lsprof data when performing benchmark calls.
963
_gather_lsprof_in_benchmarks = False
965
def __init__(self, methodName='testMethod'):
966
super(TestCase, self).__init__(methodName)
967
self._directory_isolation = True
968
self.exception_handlers.insert(0,
969
(UnavailableFeature, self._do_unsupported_or_skip))
970
self.exception_handlers.insert(0,
971
(TestNotApplicable, self._do_not_applicable))
53
routine, and to build and check bzr trees."""
974
super(TestCase, self).setUp()
975
for feature in getattr(self, '_test_needs_features', []):
976
self.requireFeature(feature)
977
self._cleanEnvironment()
980
self._benchcalls = []
981
self._benchtime = None
983
self._track_transports()
985
self._clear_debug_flags()
986
# Isolate global verbosity level, to make sure it's reproducible
987
# between tests. We should get rid of this altogether: bug 656694. --
989
self.overrideAttr(bzrlib.trace, '_verbosity_level', 0)
990
# Isolate config option expansion until its default value for bzrlib is
991
# settled on or a the FIXME associated with _get_expand_default_value
992
# is addressed -- vila 20110219
993
self.overrideAttr(config, '_expand_default_value', None)
994
self._log_files = set()
995
# Each key in the ``_counters`` dict holds a value for a different
996
# counter. When the test ends, addDetail() should be used to output the
997
# counter values. This happens in install_counter_hook().
999
if 'config_stats' in selftest_debug_flags:
1000
self._install_config_stats_hooks()
1005
pdb.Pdb().set_trace(sys._getframe().f_back)
1007
def discardDetail(self, name):
1008
"""Extend the addDetail, getDetails api so we can remove a detail.
1010
eg. bzr always adds the 'log' detail at startup, but we don't want to
1011
include it for skipped, xfail, etc tests.
1013
It is safe to call this for a detail that doesn't exist, in case this
1014
gets called multiple times.
1016
# We cheat. details is stored in __details which means we shouldn't
1017
# touch it. but getDetails() returns the dict directly, so we can
1019
details = self.getDetails()
1023
def install_counter_hook(self, hooks, name, counter_name=None):
1024
"""Install a counting hook.
1026
Any hook can be counted as long as it doesn't need to return a value.
1028
:param hooks: Where the hook should be installed.
1030
:param name: The hook name that will be counted.
1032
:param counter_name: The counter identifier in ``_counters``, defaults
1035
_counters = self._counters # Avoid closing over self
1036
if counter_name is None:
1038
if _counters.has_key(counter_name):
1039
raise AssertionError('%s is already used as a counter name'
1041
_counters[counter_name] = 0
1042
self.addDetail(counter_name, content.Content(content.UTF8_TEXT,
1043
lambda: ['%d' % (_counters[counter_name],)]))
1044
def increment_counter(*args, **kwargs):
1045
_counters[counter_name] += 1
1046
label = 'count %s calls' % (counter_name,)
1047
hooks.install_named_hook(name, increment_counter, label)
1048
self.addCleanup(hooks.uninstall_named_hook, name, label)
1050
def _install_config_stats_hooks(self):
1051
"""Install config hooks to count hook calls.
1054
for hook_name in ('get', 'set', 'remove', 'load', 'save'):
1055
self.install_counter_hook(config.ConfigHooks, hook_name,
1056
'config.%s' % (hook_name,))
1058
# The OldConfigHooks are private and need special handling to protect
1059
# against recursive tests (tests that run other tests), so we just do
1060
# manually what registering them into _builtin_known_hooks will provide
1062
self.overrideAttr(config, 'OldConfigHooks', config._OldConfigHooks())
1063
for hook_name in ('get', 'set', 'remove', 'load', 'save'):
1064
self.install_counter_hook(config.OldConfigHooks, hook_name,
1065
'old_config.%s' % (hook_name,))
1067
def _clear_debug_flags(self):
1068
"""Prevent externally set debug flags affecting tests.
1070
Tests that want to use debug flags can just set them in the
1071
debug_flags set during setup/teardown.
1073
# Start with a copy of the current debug flags we can safely modify.
1074
self.overrideAttr(debug, 'debug_flags', set(debug.debug_flags))
1075
if 'allow_debug' not in selftest_debug_flags:
1076
debug.debug_flags.clear()
1077
if 'disable_lock_checks' not in selftest_debug_flags:
1078
debug.debug_flags.add('strict_locks')
1080
def _clear_hooks(self):
1081
# prevent hooks affecting tests
1082
known_hooks = hooks.known_hooks
1083
self._preserved_hooks = {}
1084
for key, (parent, name) in known_hooks.iter_parent_objects():
1085
current_hooks = getattr(parent, name)
1086
self._preserved_hooks[parent] = (name, current_hooks)
1087
self._preserved_lazy_hooks = hooks._lazy_hooks
1088
hooks._lazy_hooks = {}
1089
self.addCleanup(self._restoreHooks)
1090
for key, (parent, name) in known_hooks.iter_parent_objects():
1091
factory = known_hooks.get(key)
1092
setattr(parent, name, factory())
1093
# this hook should always be installed
1094
request._install_hook()
1096
def disable_directory_isolation(self):
1097
"""Turn off directory isolation checks."""
1098
self._directory_isolation = False
1100
def enable_directory_isolation(self):
1101
"""Enable directory isolation checks."""
1102
self._directory_isolation = True
1104
def _silenceUI(self):
1105
"""Turn off UI for duration of test"""
1106
# by default the UI is off; tests can turn it on if they want it.
1107
self.overrideAttr(ui, 'ui_factory', ui.SilentUIFactory())
1109
def _check_locks(self):
1110
"""Check that all lock take/release actions have been paired."""
1111
# We always check for mismatched locks. If a mismatch is found, we
1112
# fail unless -Edisable_lock_checks is supplied to selftest, in which
1113
# case we just print a warning.
1115
acquired_locks = [lock for action, lock in self._lock_actions
1116
if action == 'acquired']
1117
released_locks = [lock for action, lock in self._lock_actions
1118
if action == 'released']
1119
broken_locks = [lock for action, lock in self._lock_actions
1120
if action == 'broken']
1121
# trivially, given the tests for lock acquistion and release, if we
1122
# have as many in each list, it should be ok. Some lock tests also
1123
# break some locks on purpose and should be taken into account by
1124
# considering that breaking a lock is just a dirty way of releasing it.
1125
if len(acquired_locks) != (len(released_locks) + len(broken_locks)):
1127
'Different number of acquired and '
1128
'released or broken locks.\n'
1132
(acquired_locks, released_locks, broken_locks))
1133
if not self._lock_check_thorough:
1134
# Rather than fail, just warn
1135
print "Broken test %s: %s" % (self, message)
1139
def _track_locks(self):
1140
"""Track lock activity during tests."""
1141
self._lock_actions = []
1142
if 'disable_lock_checks' in selftest_debug_flags:
1143
self._lock_check_thorough = False
1145
self._lock_check_thorough = True
1147
self.addCleanup(self._check_locks)
1148
_mod_lock.Lock.hooks.install_named_hook('lock_acquired',
1149
self._lock_acquired, None)
1150
_mod_lock.Lock.hooks.install_named_hook('lock_released',
1151
self._lock_released, None)
1152
_mod_lock.Lock.hooks.install_named_hook('lock_broken',
1153
self._lock_broken, None)
1155
def _lock_acquired(self, result):
1156
self._lock_actions.append(('acquired', result))
1158
def _lock_released(self, result):
1159
self._lock_actions.append(('released', result))
1161
def _lock_broken(self, result):
1162
self._lock_actions.append(('broken', result))
1164
def permit_dir(self, name):
1165
"""Permit a directory to be used by this test. See permit_url."""
1166
name_transport = _mod_transport.get_transport(name)
1167
self.permit_url(name)
1168
self.permit_url(name_transport.base)
1170
def permit_url(self, url):
1171
"""Declare that url is an ok url to use in this test.
1173
Do this for memory transports, temporary test directory etc.
1175
Do not do this for the current working directory, /tmp, or any other
1176
preexisting non isolated url.
1178
if not url.endswith('/'):
1180
self._bzr_selftest_roots.append(url)
1182
def permit_source_tree_branch_repo(self):
1183
"""Permit the source tree bzr is running from to be opened.
1185
Some code such as bzrlib.version attempts to read from the bzr branch
1186
that bzr is executing from (if any). This method permits that directory
1187
to be used in the test suite.
1189
path = self.get_source_path()
1190
self.record_directory_isolation()
1193
workingtree.WorkingTree.open(path)
1194
except (errors.NotBranchError, errors.NoWorkingTree):
1195
raise TestSkipped('Needs a working tree of bzr sources')
1197
self.enable_directory_isolation()
1199
def _preopen_isolate_transport(self, transport):
1200
"""Check that all transport openings are done in the test work area."""
1201
while isinstance(transport, pathfilter.PathFilteringTransport):
1202
# Unwrap pathfiltered transports
1203
transport = transport.server.backing_transport.clone(
1204
transport._filter('.'))
1205
url = transport.base
1206
# ReadonlySmartTCPServer_for_testing decorates the backing transport
1207
# urls it is given by prepending readonly+. This is appropriate as the
1208
# client shouldn't know that the server is readonly (or not readonly).
1209
# We could register all servers twice, with readonly+ prepending, but
1210
# that makes for a long list; this is about the same but easier to
1212
if url.startswith('readonly+'):
1213
url = url[len('readonly+'):]
1214
self._preopen_isolate_url(url)
1216
def _preopen_isolate_url(self, url):
1217
if not self._directory_isolation:
1219
if self._directory_isolation == 'record':
1220
self._bzr_selftest_roots.append(url)
1222
# This prevents all transports, including e.g. sftp ones backed on disk
1223
# from working unless they are explicitly granted permission. We then
1224
# depend on the code that sets up test transports to check that they are
1225
# appropriately isolated and enable their use by calling
1226
# self.permit_transport()
1227
if not osutils.is_inside_any(self._bzr_selftest_roots, url):
1228
raise errors.BzrError("Attempt to escape test isolation: %r %r"
1229
% (url, self._bzr_selftest_roots))
1231
def record_directory_isolation(self):
1232
"""Gather accessed directories to permit later access.
1234
This is used for tests that access the branch bzr is running from.
1236
self._directory_isolation = "record"
1238
def start_server(self, transport_server, backing_server=None):
1239
"""Start transport_server for this test.
1241
This starts the server, registers a cleanup for it and permits the
1242
server's urls to be used.
1244
if backing_server is None:
1245
transport_server.start_server()
1247
transport_server.start_server(backing_server)
1248
self.addCleanup(transport_server.stop_server)
1249
# Obtain a real transport because if the server supplies a password, it
1250
# will be hidden from the base on the client side.
1251
t = _mod_transport.get_transport(transport_server.get_url())
1252
# Some transport servers effectively chroot the backing transport;
1253
# others like SFTPServer don't - users of the transport can walk up the
1254
# transport to read the entire backing transport. This wouldn't matter
1255
# except that the workdir tests are given - and that they expect the
1256
# server's url to point at - is one directory under the safety net. So
1257
# Branch operations into the transport will attempt to walk up one
1258
# directory. Chrooting all servers would avoid this but also mean that
1259
# we wouldn't be testing directly against non-root urls. Alternatively
1260
# getting the test framework to start the server with a backing server
1261
# at the actual safety net directory would work too, but this then
1262
# means that the self.get_url/self.get_transport methods would need
1263
# to transform all their results. On balance its cleaner to handle it
1264
# here, and permit a higher url when we have one of these transports.
1265
if t.base.endswith('/work/'):
1266
# we have safety net/test root/work
1267
t = t.clone('../..')
1268
elif isinstance(transport_server,
1269
test_server.SmartTCPServer_for_testing):
1270
# The smart server adds a path similar to work, which is traversed
1271
# up from by the client. But the server is chrooted - the actual
1272
# backing transport is not escaped from, and VFS requests to the
1273
# root will error (because they try to escape the chroot).
1275
while t2.base != t.base:
1278
self.permit_url(t.base)
1280
def _track_transports(self):
1281
"""Install checks for transport usage."""
1282
# TestCase has no safe place it can write to.
1283
self._bzr_selftest_roots = []
1284
# Currently the easiest way to be sure that nothing is going on is to
1285
# hook into bzr dir opening. This leaves a small window of error for
1286
# transport tests, but they are well known, and we can improve on this
1288
bzrdir.BzrDir.hooks.install_named_hook("pre_open",
1289
self._preopen_isolate_transport, "Check bzr directories are safe.")
1291
def _ndiff_strings(self, a, b):
1292
"""Return ndiff between two strings containing lines.
1294
A trailing newline is added if missing to make the strings
1296
if b and b[-1] != '\n':
1298
if a and a[-1] != '\n':
1300
difflines = difflib.ndiff(a.splitlines(True),
1302
linejunk=lambda x: False,
1303
charjunk=lambda x: False)
1304
return ''.join(difflines)
1306
def assertEqual(self, a, b, message=''):
1310
except UnicodeError, e:
1311
# If we can't compare without getting a UnicodeError, then
1312
# obviously they are different
1313
trace.mutter('UnicodeError: %s', e)
1316
raise AssertionError("%snot equal:\na = %s\nb = %s\n"
1318
pprint.pformat(a), pprint.pformat(b)))
1320
assertEquals = assertEqual
1322
def assertEqualDiff(self, a, b, message=None):
1323
"""Assert two texts are equal, if not raise an exception.
1325
This is intended for use with multi-line strings where it can
1326
be hard to find the differences by eye.
1328
# TODO: perhaps override assertEquals to call this for strings?
1332
message = "texts not equal:\n"
1334
message = 'first string is missing a final newline.\n'
1336
message = 'second string is missing a final newline.\n'
1337
raise AssertionError(message +
1338
self._ndiff_strings(a, b))
1340
def assertEqualMode(self, mode, mode_test):
1341
self.assertEqual(mode, mode_test,
1342
'mode mismatch %o != %o' % (mode, mode_test))
1344
def assertEqualStat(self, expected, actual):
1345
"""assert that expected and actual are the same stat result.
1347
:param expected: A stat result.
1348
:param actual: A stat result.
1349
:raises AssertionError: If the expected and actual stat values differ
1350
other than by atime.
1352
self.assertEqual(expected.st_size, actual.st_size,
1353
'st_size did not match')
1354
self.assertEqual(expected.st_mtime, actual.st_mtime,
1355
'st_mtime did not match')
1356
self.assertEqual(expected.st_ctime, actual.st_ctime,
1357
'st_ctime did not match')
1358
if sys.platform == 'win32':
1359
# On Win32 both 'dev' and 'ino' cannot be trusted. In python2.4 it
1360
# is 'dev' that varies, in python 2.5 (6?) it is st_ino that is
1361
# odd. We just force it to always be 0 to avoid any problems.
1362
self.assertEqual(0, expected.st_dev)
1363
self.assertEqual(0, actual.st_dev)
1364
self.assertEqual(0, expected.st_ino)
1365
self.assertEqual(0, actual.st_ino)
1367
self.assertEqual(expected.st_dev, actual.st_dev,
1368
'st_dev did not match')
1369
self.assertEqual(expected.st_ino, actual.st_ino,
1370
'st_ino did not match')
1371
self.assertEqual(expected.st_mode, actual.st_mode,
1372
'st_mode did not match')
1374
def assertLength(self, length, obj_with_len):
1375
"""Assert that obj_with_len is of length length."""
1376
if len(obj_with_len) != length:
1377
self.fail("Incorrect length: wanted %d, got %d for %r" % (
1378
length, len(obj_with_len), obj_with_len))
1380
def assertLogsError(self, exception_class, func, *args, **kwargs):
1381
"""Assert that `func(*args, **kwargs)` quietly logs a specific error.
1384
orig_log_exception_quietly = trace.log_exception_quietly
1387
orig_log_exception_quietly()
1388
captured.append(sys.exc_info()[1])
1389
trace.log_exception_quietly = capture
1390
func(*args, **kwargs)
1392
trace.log_exception_quietly = orig_log_exception_quietly
1393
self.assertLength(1, captured)
1395
self.assertIsInstance(err, exception_class)
1398
def assertPositive(self, val):
1399
"""Assert that val is greater than 0."""
1400
self.assertTrue(val > 0, 'expected a positive value, but got %s' % val)
1402
def assertNegative(self, val):
1403
"""Assert that val is less than 0."""
1404
self.assertTrue(val < 0, 'expected a negative value, but got %s' % val)
1406
def assertStartsWith(self, s, prefix):
1407
if not s.startswith(prefix):
1408
raise AssertionError('string %r does not start with %r' % (s, prefix))
1410
def assertEndsWith(self, s, suffix):
1411
"""Asserts that s ends with suffix."""
1412
if not s.endswith(suffix):
1413
raise AssertionError('string %r does not end with %r' % (s, suffix))
1415
def assertContainsRe(self, haystack, needle_re, flags=0):
1416
"""Assert that a contains something matching a regular expression."""
1417
if not re.search(needle_re, haystack, flags):
1418
if '\n' in haystack or len(haystack) > 60:
1419
# a long string, format it in a more readable way
1420
raise AssertionError(
1421
'pattern "%s" not found in\n"""\\\n%s"""\n'
1422
% (needle_re, haystack))
1424
raise AssertionError('pattern "%s" not found in "%s"'
1425
% (needle_re, haystack))
1427
def assertNotContainsRe(self, haystack, needle_re, flags=0):
1428
"""Assert that a does not match a regular expression"""
1429
if re.search(needle_re, haystack, flags):
1430
raise AssertionError('pattern "%s" found in "%s"'
1431
% (needle_re, haystack))
1433
def assertContainsString(self, haystack, needle):
1434
if haystack.find(needle) == -1:
1435
self.fail("string %r not found in '''%s'''" % (needle, haystack))
1437
def assertNotContainsString(self, haystack, needle):
1438
if haystack.find(needle) != -1:
1439
self.fail("string %r found in '''%s'''" % (needle, haystack))
1441
def assertSubset(self, sublist, superlist):
1442
"""Assert that every entry in sublist is present in superlist."""
1443
missing = set(sublist) - set(superlist)
1444
if len(missing) > 0:
1445
raise AssertionError("value(s) %r not present in container %r" %
1446
(missing, superlist))
1448
def assertListRaises(self, excClass, func, *args, **kwargs):
1449
"""Fail unless excClass is raised when the iterator from func is used.
1451
Many functions can return generators this makes sure
1452
to wrap them in a list() call to make sure the whole generator
1453
is run, and that the proper exception is raised.
1456
list(func(*args, **kwargs))
1460
if getattr(excClass,'__name__', None) is not None:
1461
excName = excClass.__name__
1463
excName = str(excClass)
1464
raise self.failureException, "%s not raised" % excName
1466
def assertRaises(self, excClass, callableObj, *args, **kwargs):
1467
"""Assert that a callable raises a particular exception.
1469
:param excClass: As for the except statement, this may be either an
1470
exception class, or a tuple of classes.
1471
:param callableObj: A callable, will be passed ``*args`` and
1474
Returns the exception so that you can examine it.
1477
callableObj(*args, **kwargs)
1481
if getattr(excClass,'__name__', None) is not None:
1482
excName = excClass.__name__
1485
excName = str(excClass)
1486
raise self.failureException, "%s not raised" % excName
1488
def assertIs(self, left, right, message=None):
1489
if not (left is right):
1490
if message is not None:
1491
raise AssertionError(message)
1493
raise AssertionError("%r is not %r." % (left, right))
1495
def assertIsNot(self, left, right, message=None):
1497
if message is not None:
1498
raise AssertionError(message)
1500
raise AssertionError("%r is %r." % (left, right))
1502
def assertTransportMode(self, transport, path, mode):
1503
"""Fail if a path does not have mode "mode".
1505
If modes are not supported on this transport, the assertion is ignored.
1507
if not transport._can_roundtrip_unix_modebits():
1509
path_stat = transport.stat(path)
1510
actual_mode = stat.S_IMODE(path_stat.st_mode)
1511
self.assertEqual(mode, actual_mode,
1512
'mode of %r incorrect (%s != %s)'
1513
% (path, oct(mode), oct(actual_mode)))
1515
def assertIsSameRealPath(self, path1, path2):
1516
"""Fail if path1 and path2 points to different files"""
1517
self.assertEqual(osutils.realpath(path1),
1518
osutils.realpath(path2),
1519
"apparent paths:\na = %s\nb = %s\n," % (path1, path2))
1521
def assertIsInstance(self, obj, kls, msg=None):
1522
"""Fail if obj is not an instance of kls
1524
:param msg: Supplementary message to show if the assertion fails.
1526
if not isinstance(obj, kls):
1527
m = "%r is an instance of %s rather than %s" % (
1528
obj, obj.__class__, kls)
1533
def assertFileEqual(self, content, path):
1534
"""Fail if path does not contain 'content'."""
1535
self.assertPathExists(path)
1536
f = file(path, 'rb')
1541
self.assertEqualDiff(content, s)
1543
def assertDocstring(self, expected_docstring, obj):
1544
"""Fail if obj does not have expected_docstring"""
1546
# With -OO the docstring should be None instead
1547
self.assertIs(obj.__doc__, None)
1549
self.assertEqual(expected_docstring, obj.__doc__)
1551
@symbol_versioning.deprecated_method(symbol_versioning.deprecated_in((2, 4)))
1552
def failUnlessExists(self, path):
1553
return self.assertPathExists(path)
1555
def assertPathExists(self, path):
1556
"""Fail unless path or paths, which may be abs or relative, exist."""
1557
if not isinstance(path, basestring):
1559
self.assertPathExists(p)
1561
self.assertTrue(osutils.lexists(path),
1562
path + " does not exist")
1564
@symbol_versioning.deprecated_method(symbol_versioning.deprecated_in((2, 4)))
1565
def failIfExists(self, path):
1566
return self.assertPathDoesNotExist(path)
1568
def assertPathDoesNotExist(self, path):
1569
"""Fail if path or paths, which may be abs or relative, exist."""
1570
if not isinstance(path, basestring):
1572
self.assertPathDoesNotExist(p)
1574
self.assertFalse(osutils.lexists(path),
1577
def _capture_deprecation_warnings(self, a_callable, *args, **kwargs):
1578
"""A helper for callDeprecated and applyDeprecated.
1580
:param a_callable: A callable to call.
1581
:param args: The positional arguments for the callable
1582
:param kwargs: The keyword arguments for the callable
1583
:return: A tuple (warnings, result). result is the result of calling
1584
a_callable(``*args``, ``**kwargs``).
1587
def capture_warnings(msg, cls=None, stacklevel=None):
1588
# we've hooked into a deprecation specific callpath,
1589
# only deprecations should getting sent via it.
1590
self.assertEqual(cls, DeprecationWarning)
1591
local_warnings.append(msg)
1592
original_warning_method = symbol_versioning.warn
1593
symbol_versioning.set_warning_method(capture_warnings)
1595
result = a_callable(*args, **kwargs)
1597
symbol_versioning.set_warning_method(original_warning_method)
1598
return (local_warnings, result)
1600
def applyDeprecated(self, deprecation_format, a_callable, *args, **kwargs):
1601
"""Call a deprecated callable without warning the user.
1603
Note that this only captures warnings raised by symbol_versioning.warn,
1604
not other callers that go direct to the warning module.
1606
To test that a deprecated method raises an error, do something like
1607
this (remember that both assertRaises and applyDeprecated delays *args
1608
and **kwargs passing)::
1610
self.assertRaises(errors.ReservedId,
1611
self.applyDeprecated,
1612
deprecated_in((1, 5, 0)),
1616
:param deprecation_format: The deprecation format that the callable
1617
should have been deprecated with. This is the same type as the
1618
parameter to deprecated_method/deprecated_function. If the
1619
callable is not deprecated with this format, an assertion error
1621
:param a_callable: A callable to call. This may be a bound method or
1622
a regular function. It will be called with ``*args`` and
1624
:param args: The positional arguments for the callable
1625
:param kwargs: The keyword arguments for the callable
1626
:return: The result of a_callable(``*args``, ``**kwargs``)
1628
call_warnings, result = self._capture_deprecation_warnings(a_callable,
1630
expected_first_warning = symbol_versioning.deprecation_string(
1631
a_callable, deprecation_format)
1632
if len(call_warnings) == 0:
1633
self.fail("No deprecation warning generated by call to %s" %
1635
self.assertEqual(expected_first_warning, call_warnings[0])
1638
def callCatchWarnings(self, fn, *args, **kw):
1639
"""Call a callable that raises python warnings.
1641
The caller's responsible for examining the returned warnings.
1643
If the callable raises an exception, the exception is not
1644
caught and propagates up to the caller. In that case, the list
1645
of warnings is not available.
1647
:returns: ([warning_object, ...], fn_result)
1649
# XXX: This is not perfect, because it completely overrides the
1650
# warnings filters, and some code may depend on suppressing particular
1651
# warnings. It's the easiest way to insulate ourselves from -Werror,
1652
# though. -- Andrew, 20071062
1654
def _catcher(message, category, filename, lineno, file=None, line=None):
1655
# despite the name, 'message' is normally(?) a Warning subclass
1657
wlist.append(message)
1658
saved_showwarning = warnings.showwarning
1659
saved_filters = warnings.filters
1661
warnings.showwarning = _catcher
1662
warnings.filters = []
1663
result = fn(*args, **kw)
1665
warnings.showwarning = saved_showwarning
1666
warnings.filters = saved_filters
1667
return wlist, result
1669
def callDeprecated(self, expected, callable, *args, **kwargs):
1670
"""Assert that a callable is deprecated in a particular way.
1672
This is a very precise test for unusual requirements. The
1673
applyDeprecated helper function is probably more suited for most tests
1674
as it allows you to simply specify the deprecation format being used
1675
and will ensure that that is issued for the function being called.
1677
Note that this only captures warnings raised by symbol_versioning.warn,
1678
not other callers that go direct to the warning module. To catch
1679
general warnings, use callCatchWarnings.
1681
:param expected: a list of the deprecation warnings expected, in order
1682
:param callable: The callable to call
1683
:param args: The positional arguments for the callable
1684
:param kwargs: The keyword arguments for the callable
1686
call_warnings, result = self._capture_deprecation_warnings(callable,
1688
self.assertEqual(expected, call_warnings)
1691
def _startLogFile(self):
1692
"""Send bzr and test log messages to a temporary file.
1694
The file is removed as the test is torn down.
1696
pseudo_log_file = StringIO()
1697
def _get_log_contents_for_weird_testtools_api():
1698
return [pseudo_log_file.getvalue().decode(
1699
"utf-8", "replace").encode("utf-8")]
1700
self.addDetail("log", content.Content(content.ContentType("text",
1701
"plain", {"charset": "utf8"}),
1702
_get_log_contents_for_weird_testtools_api))
1703
self._log_file = pseudo_log_file
1704
self._log_memento = trace.push_log_file(self._log_file)
1705
self.addCleanup(self._finishLogFile)
1707
def _finishLogFile(self):
1708
"""Finished with the log file.
1710
Close the file and delete it.
1712
if trace._trace_file:
1713
# flush the log file, to get all content
1714
trace._trace_file.flush()
1715
trace.pop_log_file(self._log_memento)
1717
def thisFailsStrictLockCheck(self):
1718
"""It is known that this test would fail with -Dstrict_locks.
1720
By default, all tests are run with strict lock checking unless
1721
-Edisable_lock_checks is supplied. However there are some tests which
1722
we know fail strict locks at this point that have not been fixed.
1723
They should call this function to disable the strict checking.
1725
This should be used sparingly, it is much better to fix the locking
1726
issues rather than papering over the problem by calling this function.
1728
debug.debug_flags.discard('strict_locks')
1730
def overrideAttr(self, obj, attr_name, new=_unitialized_attr):
1731
"""Overrides an object attribute restoring it after the test.
1733
:param obj: The object that will be mutated.
1735
:param attr_name: The attribute name we want to preserve/override in
1738
:param new: The optional value we want to set the attribute to.
1740
:returns: The actual attr value.
1742
value = getattr(obj, attr_name)
1743
# The actual value is captured by the call below
1744
self.addCleanup(setattr, obj, attr_name, value)
1745
if new is not _unitialized_attr:
1746
setattr(obj, attr_name, new)
1749
def overrideEnv(self, name, new):
1750
"""Set an environment variable, and reset it after the test.
1752
:param name: The environment variable name.
1754
:param new: The value to set the variable to. If None, the
1755
variable is deleted from the environment.
1757
:returns: The actual variable value.
1759
value = osutils.set_or_unset_env(name, new)
1760
self.addCleanup(osutils.set_or_unset_env, name, value)
1763
def _cleanEnvironment(self):
1764
for name, value in isolated_environ.iteritems():
1765
self.overrideEnv(name, value)
1767
def _restoreHooks(self):
1768
for klass, (name, hooks) in self._preserved_hooks.items():
1769
setattr(klass, name, hooks)
1770
self._preserved_hooks.clear()
1771
bzrlib.hooks._lazy_hooks = self._preserved_lazy_hooks
1772
self._preserved_lazy_hooks.clear()
1774
def knownFailure(self, reason):
1775
"""This test has failed for some known reason."""
1776
raise KnownFailure(reason)
1778
def _suppress_log(self):
1779
"""Remove the log info from details."""
1780
self.discardDetail('log')
1782
def _do_skip(self, result, reason):
1783
self._suppress_log()
1784
addSkip = getattr(result, 'addSkip', None)
1785
if not callable(addSkip):
1786
result.addSuccess(result)
1788
addSkip(self, reason)
1791
def _do_known_failure(self, result, e):
1792
self._suppress_log()
1793
err = sys.exc_info()
1794
addExpectedFailure = getattr(result, 'addExpectedFailure', None)
1795
if addExpectedFailure is not None:
1796
addExpectedFailure(self, err)
1798
result.addSuccess(self)
1801
def _do_not_applicable(self, result, e):
1803
reason = 'No reason given'
1806
self._suppress_log ()
1807
addNotApplicable = getattr(result, 'addNotApplicable', None)
1808
if addNotApplicable is not None:
1809
result.addNotApplicable(self, reason)
1811
self._do_skip(result, reason)
1814
def _report_skip(self, result, err):
1815
"""Override the default _report_skip.
1817
We want to strip the 'log' detail. If we waint until _do_skip, it has
1818
already been formatted into the 'reason' string, and we can't pull it
1821
self._suppress_log()
1822
super(TestCase, self)._report_skip(self, result, err)
1825
def _report_expected_failure(self, result, err):
1828
See _report_skip for motivation.
1830
self._suppress_log()
1831
super(TestCase, self)._report_expected_failure(self, result, err)
1834
def _do_unsupported_or_skip(self, result, e):
1836
self._suppress_log()
1837
addNotSupported = getattr(result, 'addNotSupported', None)
1838
if addNotSupported is not None:
1839
result.addNotSupported(self, reason)
1841
self._do_skip(result, reason)
1843
def time(self, callable, *args, **kwargs):
1844
"""Run callable and accrue the time it takes to the benchmark time.
1846
If lsprofiling is enabled (i.e. by --lsprof-time to bzr selftest) then
1847
this will cause lsprofile statistics to be gathered and stored in
1850
if self._benchtime is None:
1851
self.addDetail('benchtime', content.Content(content.ContentType(
1852
"text", "plain"), lambda:[str(self._benchtime)]))
1856
if not self._gather_lsprof_in_benchmarks:
1857
return callable(*args, **kwargs)
1859
# record this benchmark
1860
ret, stats = bzrlib.lsprof.profile(callable, *args, **kwargs)
1862
self._benchcalls.append(((callable, args, kwargs), stats))
1865
self._benchtime += time.time() - start
58
# this replaces the default testsweet.TestCase; we don't want logging changed
59
unittest.TestCase.setUp(self)
60
bzrlib.trace.disable_default_logging()
61
self._enable_file_logging()
64
def _enable_file_logging(self):
65
fileno, name = tempfile.mkstemp(suffix='.log', prefix='testbzr')
67
self._log_file = os.fdopen(fileno, 'w+')
69
hdlr = logging.StreamHandler(self._log_file)
70
hdlr.setLevel(logging.DEBUG)
71
hdlr.setFormatter(logging.Formatter('%(levelname)4.4s %(message)s'))
72
logging.getLogger('').addHandler(hdlr)
73
logging.getLogger('').setLevel(logging.DEBUG)
75
debug('opened log file %s', name)
77
self._log_file_name = name
80
logging.getLogger('').removeHandler(self._log_hdlr)
81
bzrlib.trace.enable_default_logging()
82
logging.debug('%s teardown', self.id())
83
self._log_file.close()
84
unittest.TestCase.tearDown(self)
1867
86
def log(self, *args):
1871
"""Get a unicode string containing the log from bzrlib.trace.
1873
Undecodable characters are replaced.
1875
return u"".join(self.getDetails()['log'].iter_text())
1877
def requireFeature(self, feature):
1878
"""This test requires a specific feature is available.
1880
:raises UnavailableFeature: When feature is not available.
1882
if not feature.available():
1883
raise UnavailableFeature(feature)
1885
def _run_bzr_autosplit(self, args, retcode, encoding, stdin,
1887
"""Run bazaar command line, splitting up a string command line."""
1888
if isinstance(args, basestring):
1889
# shlex don't understand unicode strings,
1890
# so args should be plain string (bialix 20070906)
1891
args = list(shlex.split(str(args)))
1892
return self._run_bzr_core(args, retcode=retcode,
1893
encoding=encoding, stdin=stdin, working_dir=working_dir,
1896
def _run_bzr_core(self, args, retcode, encoding, stdin,
1898
# Clear chk_map page cache, because the contents are likely to mask
1900
chk_map.clear_cache()
1901
if encoding is None:
1902
encoding = osutils.get_user_encoding()
1903
stdout = StringIOWrapper()
1904
stderr = StringIOWrapper()
1905
stdout.encoding = encoding
1906
stderr.encoding = encoding
1908
self.log('run bzr: %r', args)
1909
# FIXME: don't call into logging here
90
"""Return as a string the log for this test"""
91
return open(self._log_file_name).read()
94
def capture(self, cmd):
95
"""Shortcut that splits cmd into words, runs, and returns stdout"""
96
return self.run_bzr_captured(cmd.split())[0]
99
def run_bzr_captured(self, argv, retcode=0):
100
"""Invoke bzr and return (result, stdout, stderr).
102
Useful for code that wants to check the contents of the
103
output, the way error messages are presented, etc.
105
This should be the main method for tests that want to exercise the
106
overall behavior of the bzr application (rather than a unit test
107
or a functional test of the library.)
109
Much of the old code runs bzr by forking a new copy of Python, but
110
that is slower, harder to debug, and generally not necessary.
112
This runs bzr through the interface that catches and reports
113
errors, and with logging set to something approximating the
114
default, so that error reporting can be checked.
116
argv -- arguments to invoke bzr
117
retcode -- expected return code, or None for don't-care.
121
self.log('run bzr: %s', ' '.join(argv))
1910
122
handler = logging.StreamHandler(stderr)
123
handler.setFormatter(bzrlib.trace.QuietFormatter())
1911
124
handler.setLevel(logging.INFO)
1912
125
logger = logging.getLogger('')
1913
126
logger.addHandler(handler)
1914
old_ui_factory = ui.ui_factory
1915
ui.ui_factory = TestUIFactory(stdin=stdin, stdout=stdout, stderr=stderr)
1918
if working_dir is not None:
1919
cwd = osutils.getcwd()
1920
os.chdir(working_dir)
1924
result = self.apply_redirected(
1925
ui.ui_factory.stdin,
1927
_mod_commands.run_bzr_catch_user_errors,
1929
except KeyboardInterrupt:
1930
# Reraise KeyboardInterrupt with contents of redirected stdout
1931
# and stderr as arguments, for tests which are interested in
1932
# stdout and stderr and are expecting the exception.
1933
out = stdout.getvalue()
1934
err = stderr.getvalue()
1936
self.log('output:\n%r', out)
1938
self.log('errors:\n%r', err)
1939
raise KeyboardInterrupt(out, err)
128
result = self.apply_redirected(None, stdout, stderr,
129
bzrlib.commands.run_bzr_catch_errors,
1941
132
logger.removeHandler(handler)
1942
ui.ui_factory = old_ui_factory
1946
133
out = stdout.getvalue()
1947
134
err = stderr.getvalue()
1949
self.log('output:\n%r', out)
136
self.log('output:\n%s', out)
1951
self.log('errors:\n%r', err)
138
self.log('errors:\n%s', err)
1952
139
if retcode is not None:
1953
self.assertEquals(retcode, result,
1954
message='Unexpected return code')
1955
return result, out, err
1957
def run_bzr(self, args, retcode=0, encoding=None, stdin=None,
1958
working_dir=None, error_regexes=[], output_encoding=None):
140
self.assertEquals(result, retcode)
144
def run_bzr(self, *args, **kwargs):
1959
145
"""Invoke bzr, as if it were run from the command line.
1961
The argument list should not include the bzr program name - the
1962
first argument is normally the bzr command. Arguments may be
1963
passed in three ways:
1965
1- A list of strings, eg ["commit", "a"]. This is recommended
1966
when the command contains whitespace or metacharacters, or
1967
is built up at run time.
1969
2- A single string, eg "add a". This is the most convenient
1970
for hardcoded commands.
1972
This runs bzr through the interface that catches and reports
1973
errors, and with logging set to something approximating the
1974
default, so that error reporting can be checked.
1976
147
This should be the main method for tests that want to exercise the
1977
148
overall behavior of the bzr application (rather than a unit test
1978
149
or a functional test of the library.)
1980
151
This sends the stdout/stderr results into the test's log,
1981
152
where it may be useful for debugging. See also run_captured.
1983
:keyword stdin: A string to be used as stdin for the command.
1984
:keyword retcode: The status code the command should return;
1986
:keyword working_dir: The directory to run the command in
1987
:keyword error_regexes: A list of expected error messages. If
1988
specified they must be seen in the error output of the command.
1990
retcode, out, err = self._run_bzr_autosplit(
1995
working_dir=working_dir,
1997
self.assertIsInstance(error_regexes, (list, tuple))
1998
for regex in error_regexes:
1999
self.assertContainsRe(err, regex)
2002
def run_bzr_error(self, error_regexes, *args, **kwargs):
2003
"""Run bzr, and check that stderr contains the supplied regexes
2005
:param error_regexes: Sequence of regular expressions which
2006
must each be found in the error output. The relative ordering
2008
:param args: command-line arguments for bzr
2009
:param kwargs: Keyword arguments which are interpreted by run_bzr
2010
This function changes the default value of retcode to be 3,
2011
since in most cases this is run when you expect bzr to fail.
2013
:return: (out, err) The actual output of running the command (in case
2014
you want to do more inspection)
2018
# Make sure that commit is failing because there is nothing to do
2019
self.run_bzr_error(['no changes to commit'],
2020
['commit', '-m', 'my commit comment'])
2021
# Make sure --strict is handling an unknown file, rather than
2022
# giving us the 'nothing to do' error
2023
self.build_tree(['unknown'])
2024
self.run_bzr_error(['Commit refused because there are unknown files'],
2025
['commit', --strict', '-m', 'my commit comment'])
2027
kwargs.setdefault('retcode', 3)
2028
kwargs['error_regexes'] = error_regexes
2029
out, err = self.run_bzr(*args, **kwargs)
2032
def run_bzr_subprocess(self, *args, **kwargs):
2033
"""Run bzr in a subprocess for testing.
2035
This starts a new Python interpreter and runs bzr in there.
2036
This should only be used for tests that have a justifiable need for
2037
this isolation: e.g. they are testing startup time, or signal
2038
handling, or early startup code, etc. Subprocess code can't be
2039
profiled or debugged so easily.
2041
:keyword retcode: The status code that is expected. Defaults to 0. If
2042
None is supplied, the status code is not checked.
2043
:keyword env_changes: A dictionary which lists changes to environment
2044
variables. A value of None will unset the env variable.
2045
The values must be strings. The change will only occur in the
2046
child, so you don't need to fix the environment after running.
2047
:keyword universal_newlines: Convert CRLF => LF
2048
:keyword allow_plugins: By default the subprocess is run with
2049
--no-plugins to ensure test reproducibility. Also, it is possible
2050
for system-wide plugins to create unexpected output on stderr,
2051
which can cause unnecessary test failures.
2053
env_changes = kwargs.get('env_changes', {})
2054
working_dir = kwargs.get('working_dir', None)
2055
allow_plugins = kwargs.get('allow_plugins', False)
2057
if isinstance(args[0], list):
2059
elif isinstance(args[0], basestring):
2060
args = list(shlex.split(args[0]))
2062
raise ValueError("passing varargs to run_bzr_subprocess")
2063
process = self.start_bzr_subprocess(args, env_changes=env_changes,
2064
working_dir=working_dir,
2065
allow_plugins=allow_plugins)
2066
# We distinguish between retcode=None and retcode not passed.
2067
supplied_retcode = kwargs.get('retcode', 0)
2068
return self.finish_bzr_subprocess(process, retcode=supplied_retcode,
2069
universal_newlines=kwargs.get('universal_newlines', False),
2072
def start_bzr_subprocess(self, process_args, env_changes=None,
2073
skip_if_plan_to_signal=False,
2075
allow_plugins=False, stderr=subprocess.PIPE):
2076
"""Start bzr in a subprocess for testing.
2078
This starts a new Python interpreter and runs bzr in there.
2079
This should only be used for tests that have a justifiable need for
2080
this isolation: e.g. they are testing startup time, or signal
2081
handling, or early startup code, etc. Subprocess code can't be
2082
profiled or debugged so easily.
2084
:param process_args: a list of arguments to pass to the bzr executable,
2085
for example ``['--version']``.
2086
:param env_changes: A dictionary which lists changes to environment
2087
variables. A value of None will unset the env variable.
2088
The values must be strings. The change will only occur in the
2089
child, so you don't need to fix the environment after running.
2090
:param skip_if_plan_to_signal: raise TestSkipped when true and system
2091
doesn't support signalling subprocesses.
2092
:param allow_plugins: If False (default) pass --no-plugins to bzr.
2093
:param stderr: file to use for the subprocess's stderr. Valid values
2094
are those valid for the stderr argument of `subprocess.Popen`.
2095
Default value is ``subprocess.PIPE``.
2097
:returns: Popen object for the started process.
2099
if skip_if_plan_to_signal:
2100
if os.name != "posix":
2101
raise TestSkipped("Sending signals not supported")
2103
if env_changes is None:
2107
def cleanup_environment():
2108
for env_var, value in env_changes.iteritems():
2109
old_env[env_var] = osutils.set_or_unset_env(env_var, value)
2111
def restore_environment():
2112
for env_var, value in old_env.iteritems():
2113
osutils.set_or_unset_env(env_var, value)
2115
bzr_path = self.get_bzr_path()
2118
if working_dir is not None:
2119
cwd = osutils.getcwd()
2120
os.chdir(working_dir)
2123
# win32 subprocess doesn't support preexec_fn
2124
# so we will avoid using it on all platforms, just to
2125
# make sure the code path is used, and we don't break on win32
2126
cleanup_environment()
2127
# Include the subprocess's log file in the test details, in case
2128
# the test fails due to an error in the subprocess.
2129
self._add_subprocess_log(trace._get_bzr_log_filename())
2130
command = [sys.executable]
2131
# frozen executables don't need the path to bzr
2132
if getattr(sys, "frozen", None) is None:
2133
command.append(bzr_path)
2134
if not allow_plugins:
2135
command.append('--no-plugins')
2136
command.extend(process_args)
2137
process = self._popen(command, stdin=subprocess.PIPE,
2138
stdout=subprocess.PIPE,
2141
restore_environment()
2147
def _add_subprocess_log(self, log_file_path):
2148
if len(self._log_files) == 0:
2149
# Register an addCleanup func. We do this on the first call to
2150
# _add_subprocess_log rather than in TestCase.setUp so that this
2151
# addCleanup is registered after any cleanups for tempdirs that
2152
# subclasses might create, which will probably remove the log file
2154
self.addCleanup(self._subprocess_log_cleanup)
2155
# self._log_files is a set, so if a log file is reused we won't grab it
2157
self._log_files.add(log_file_path)
2159
def _subprocess_log_cleanup(self):
2160
for count, log_file_path in enumerate(self._log_files):
2161
# We use buffer_now=True to avoid holding the file open beyond
2162
# the life of this function, which might interfere with e.g.
2163
# cleaning tempdirs on Windows.
2164
# XXX: Testtools 0.9.5 doesn't have the content_from_file helper
2165
#detail_content = content.content_from_file(
2166
# log_file_path, buffer_now=True)
2167
with open(log_file_path, 'rb') as log_file:
2168
log_file_bytes = log_file.read()
2169
detail_content = content.Content(content.ContentType("text",
2170
"plain", {"charset": "utf8"}), lambda: [log_file_bytes])
2171
self.addDetail("start_bzr_subprocess-log-%d" % (count,),
2174
def _popen(self, *args, **kwargs):
2175
"""Place a call to Popen.
2177
Allows tests to override this method to intercept the calls made to
2178
Popen for introspection.
2180
return subprocess.Popen(*args, **kwargs)
2182
def get_source_path(self):
2183
"""Return the path of the directory containing bzrlib."""
2184
return os.path.dirname(os.path.dirname(bzrlib.__file__))
2186
def get_bzr_path(self):
2187
"""Return the path of the 'bzr' executable for this test suite."""
2188
bzr_path = os.path.join(self.get_source_path(), "bzr")
2189
if not os.path.isfile(bzr_path):
2190
# We are probably installed. Assume sys.argv is the right file
2191
bzr_path = sys.argv[0]
2194
def finish_bzr_subprocess(self, process, retcode=0, send_signal=None,
2195
universal_newlines=False, process_args=None):
2196
"""Finish the execution of process.
2198
:param process: the Popen object returned from start_bzr_subprocess.
2199
:param retcode: The status code that is expected. Defaults to 0. If
2200
None is supplied, the status code is not checked.
2201
:param send_signal: an optional signal to send to the process.
2202
:param universal_newlines: Convert CRLF => LF
2203
:returns: (stdout, stderr)
2205
if send_signal is not None:
2206
os.kill(process.pid, send_signal)
2207
out, err = process.communicate()
2209
if universal_newlines:
2210
out = out.replace('\r\n', '\n')
2211
err = err.replace('\r\n', '\n')
2213
if retcode is not None and retcode != process.returncode:
2214
if process_args is None:
2215
process_args = "(unknown args)"
2216
trace.mutter('Output of bzr %s:\n%s', process_args, out)
2217
trace.mutter('Error for bzr %s:\n%s', process_args, err)
2218
self.fail('Command bzr %s failed with retcode %s != %s'
2219
% (process_args, retcode, process.returncode))
2222
def check_tree_shape(self, tree, shape):
2223
"""Compare a tree to a list of expected names.
154
retcode = kwargs.pop('retcode', 0)
155
return self.run_bzr_captured(args, retcode)
158
def check_inventory_shape(self, inv, shape):
160
Compare an inventory to a list of expected names.
2225
162
Fail if they are not precisely equal.
2228
165
shape = list(shape) # copy
2229
for path, ie in tree.iter_entries_by_dir():
166
for path, ie in inv.entries():
2230
167
name = path.replace('\\', '/')
2231
if ie.kind == 'directory':
2232
169
name = name + '/'
2234
pass # ignore root entry
2236
171
shape.remove(name)
2238
173
extras.append(name)
2664
222
All test cases create their own directory within that. If the
2665
223
tests complete successfully, the directory is removed.
2667
:ivar test_base_dir: The path of the top-level directory for this
2668
test, which contains a home directory and a work directory.
2670
:ivar test_home_dir: An initially empty directory under test_base_dir
2671
which is used as $HOME for this test.
2673
:ivar test_dir: A directory under test_base_dir used as the current
2674
directory when the test proper is run.
225
InTempDir is an old alias for FunctionalTestCase.
2677
230
OVERRIDE_PYTHON = 'python'
2680
super(TestCaseInTempDir, self).setUp()
2681
# Remove the protection set in isolated_environ, we have a proper
2682
# access to disk resources now.
2683
self.overrideEnv('BZR_LOG', None)
2685
232
def check_file_contents(self, filename, expect):
2686
233
self.log("check contents of file %s" % filename)
234
contents = file(filename, 'r').read()
2692
235
if contents != expect:
2693
236
self.log("expected: %r" % expect)
2694
237
self.log("actually: %r" % contents)
2695
self.fail("contents of %s not as expected" % filename)
2697
def _getTestDirPrefix(self):
2698
# create a directory within the top level test directory
2699
if sys.platform in ('win32', 'cygwin'):
2700
name_prefix = re.sub('[<>*=+",:;_/\\-]', '_', self.id())
2701
# windows is likely to have path-length limits so use a short name
2702
name_prefix = name_prefix[-30:]
2704
name_prefix = re.sub('[/]', '_', self.id())
2707
def makeAndChdirToTestDir(self):
2708
"""See TestCaseWithMemoryTransport.makeAndChdirToTestDir().
2710
For TestCaseInTempDir we create a temporary directory based on the test
2711
name and then create two subdirs - test and home under it.
2713
name_prefix = osutils.pathjoin(TestCaseWithMemoryTransport.TEST_ROOT,
2714
self._getTestDirPrefix())
2716
for i in range(100):
2717
if os.path.exists(name):
2718
name = name_prefix + '_' + str(i)
2720
# now create test and home directories within this dir
2721
self.test_base_dir = name
2722
self.addCleanup(self.deleteTestDir)
2723
os.mkdir(self.test_base_dir)
2725
self.permit_dir(self.test_base_dir)
2726
# 'sprouting' and 'init' of a branch both walk up the tree to find
2727
# stacking policy to honour; create a bzr dir with an unshared
2728
# repository (but not a branch - our code would be trying to escape
2729
# then!) to stop them, and permit it to be read.
2730
# control = bzrdir.BzrDir.create(self.test_base_dir)
2731
# control.create_repository()
2732
self.test_home_dir = self.test_base_dir + '/home'
2733
os.mkdir(self.test_home_dir)
2734
self.test_dir = self.test_base_dir + '/work'
238
self.fail("contents of %s not as expected")
240
def _make_test_root(self):
241
if TestCaseInTempDir.TEST_ROOT is not None:
245
root = 'test%04d.tmp' % i
249
if e.errno == errno.EEXIST:
254
# successfully created
255
TestCaseInTempDir.TEST_ROOT = os.path.abspath(root)
257
# make a fake bzr directory there to prevent any tests propagating
258
# up onto the source directory's real branch
259
os.mkdir(os.path.join(TestCaseInTempDir.TEST_ROOT, '.bzr'))
262
super(TestCaseInTempDir, self).setUp()
263
self._make_test_root()
264
self._currentdir = os.getcwdu()
265
self.test_dir = os.path.join(self.TEST_ROOT, self.id())
2735
266
os.mkdir(self.test_dir)
2736
267
os.chdir(self.test_dir)
2737
# put name of test inside
2738
f = file(self.test_base_dir + '/name', 'w')
2744
def deleteTestDir(self):
2745
os.chdir(TestCaseWithMemoryTransport.TEST_ROOT)
2746
_rmtree_temp_dir(self.test_base_dir, test_id=self.id())
2748
def build_tree(self, shape, line_endings='binary', transport=None):
270
os.chdir(self._currentdir)
271
super(TestCaseInTempDir, self).tearDown()
274
def build_tree(self, shape):
2749
275
"""Build a test tree according to a pattern.
2751
277
shape is a sequence of file specifications. If the final
2752
278
character is '/', a directory is created.
2754
This assumes that all the elements in the tree being built are new.
2756
280
This doesn't add anything to a branch.
2758
:type shape: list or tuple.
2759
:param line_endings: Either 'binary' or 'native'
2760
in binary mode, exact contents are written in native mode, the
2761
line endings match the default platform endings.
2762
:param transport: A transport to write to, for building trees on VFS's.
2763
If the transport is readonly or None, "." is opened automatically.
2766
if type(shape) not in (list, tuple):
2767
raise AssertionError("Parameter 'shape' should be "
2768
"a list or a tuple. Got %r instead" % (shape,))
2769
# It's OK to just create them using forward slashes on windows.
2770
if transport is None or transport.is_readonly():
2771
transport = _mod_transport.get_transport(".")
282
# XXX: It's OK to just create them using forward slashes on windows?
2772
283
for name in shape:
2773
self.assertIsInstance(name, basestring)
284
assert isinstance(name, basestring)
2774
285
if name[-1] == '/':
2775
transport.mkdir(urlutils.escape(name[:-1]))
2777
if line_endings == 'binary':
2779
elif line_endings == 'native':
2782
raise errors.BzrError(
2783
'Invalid line ending request %r' % line_endings)
2784
content = "contents of %s%s" % (name.encode('utf-8'), end)
2785
transport.put_bytes_non_atomic(urlutils.escape(name), content)
2787
build_tree_contents = staticmethod(treeshape.build_tree_contents)
2789
def assertInWorkingTree(self, path, root_path='.', tree=None):
2790
"""Assert whether path or paths are in the WorkingTree"""
2792
tree = workingtree.WorkingTree.open(root_path)
2793
if not isinstance(path, basestring):
2795
self.assertInWorkingTree(p, tree=tree)
2797
self.assertIsNot(tree.path2id(path), None,
2798
path+' not in working tree.')
2800
def assertNotInWorkingTree(self, path, root_path='.', tree=None):
2801
"""Assert whether path or paths are not in the WorkingTree"""
2803
tree = workingtree.WorkingTree.open(root_path)
2804
if not isinstance(path, basestring):
2806
self.assertNotInWorkingTree(p,tree=tree)
2808
self.assertIs(tree.path2id(path), None, path+' in working tree.')
2811
class TestCaseWithTransport(TestCaseInTempDir):
2812
"""A test case that provides get_url and get_readonly_url facilities.
2814
These back onto two transport servers, one for readonly access and one for
2817
If no explicit class is provided for readonly access, a
2818
ReadonlyTransportDecorator is used instead which allows the use of non disk
2819
based read write transports.
2821
If an explicit class is provided for readonly access, that server and the
2822
readwrite one must both define get_url() as resolving to os.getcwd().
2825
def get_vfs_only_server(self):
2826
"""See TestCaseWithMemoryTransport.
2828
This is useful for some tests with specific servers that need
2831
if self.__vfs_server is None:
2832
self.__vfs_server = self.vfs_transport_factory()
2833
self.start_server(self.__vfs_server)
2834
return self.__vfs_server
2836
def make_branch_and_tree(self, relpath, format=None):
2837
"""Create a branch on the transport and a tree locally.
2839
If the transport is not a LocalTransport, the Tree can't be created on
2840
the transport. In that case if the vfs_transport_factory is
2841
LocalURLServer the working tree is created in the local
2842
directory backing the transport, and the returned tree's branch and
2843
repository will also be accessed locally. Otherwise a lightweight
2844
checkout is created and returned.
2846
We do this because we can't physically create a tree in the local
2847
path, with a branch reference to the transport_factory url, and
2848
a branch + repository in the vfs_transport, unless the vfs_transport
2849
namespace is distinct from the local disk - the two branch objects
2850
would collide. While we could construct a tree with its branch object
2851
pointing at the transport_factory transport in memory, reopening it
2852
would behaving unexpectedly, and has in the past caused testing bugs
2853
when we tried to do it that way.
2855
:param format: The BzrDirFormat.
2856
:returns: the WorkingTree.
2858
# TODO: always use the local disk path for the working tree,
2859
# this obviously requires a format that supports branch references
2860
# so check for that by checking bzrdir.BzrDirFormat.get_default_format()
2862
b = self.make_branch(relpath, format=format)
2864
return b.bzrdir.create_workingtree()
2865
except errors.NotLocalUrl:
2866
# We can only make working trees locally at the moment. If the
2867
# transport can't support them, then we keep the non-disk-backed
2868
# branch and create a local checkout.
2869
if self.vfs_transport_factory is test_server.LocalURLServer:
2870
# the branch is colocated on disk, we cannot create a checkout.
2871
# hopefully callers will expect this.
2872
local_controldir= bzrdir.BzrDir.open(self.get_vfs_only_url(relpath))
2873
wt = local_controldir.create_workingtree()
2874
if wt.branch._format != b._format:
2876
# Make sure that assigning to wt._branch fixes wt.branch,
2877
# in case the implementation details of workingtree objects
2879
self.assertIs(b, wt.branch)
2882
return b.create_checkout(relpath, lightweight=True)
2884
def assertIsDirectory(self, relpath, transport):
2885
"""Assert that relpath within transport is a directory.
2887
This may not be possible on all transports; in that case it propagates
2888
a TransportNotPossible.
2891
mode = transport.stat(relpath).st_mode
2892
except errors.NoSuchFile:
2893
self.fail("path %s is not a directory; no such file"
2895
if not stat.S_ISDIR(mode):
2896
self.fail("path %s is not a directory; has mode %#o"
2899
def assertTreesEqual(self, left, right):
2900
"""Check that left and right have the same content and properties."""
2901
# we use a tree delta to check for equality of the content, and we
2902
# manually check for equality of other things such as the parents list.
2903
self.assertEqual(left.get_parent_ids(), right.get_parent_ids())
2904
differences = left.changes_from(right)
2905
self.assertFalse(differences.has_changed(),
2906
"Trees %r and %r are different: %r" % (left, right, differences))
2909
super(TestCaseWithTransport, self).setUp()
2910
self.__vfs_server = None
2912
def disable_missing_extensions_warning(self):
2913
"""Some tests expect a precise stderr content.
2915
There is no point in forcing them to duplicate the extension related
2918
config.GlobalConfig().set_user_option('ignore_missing_extensions', True)
2921
class ChrootedTestCase(TestCaseWithTransport):
2922
"""A support class that provides readonly urls outside the local namespace.
2924
This is done by checking if self.transport_server is a MemoryServer. if it
2925
is then we are chrooted already, if it is not then an HttpServer is used
2928
TODO RBC 20060127: make this an option to TestCaseWithTransport so it can
2929
be used without needed to redo it when a different
2930
subclass is in use ?
2934
from bzrlib.tests import http_server
2935
super(ChrootedTestCase, self).setUp()
2936
if not self.vfs_transport_factory == memory.MemoryServer:
2937
self.transport_readonly_server = http_server.HttpServer
2940
def condition_id_re(pattern):
2941
"""Create a condition filter which performs a re check on a test's id.
2943
:param pattern: A regular expression string.
2944
:return: A callable that returns True if the re matches.
2946
filter_re = re.compile(pattern, 0)
2947
def condition(test):
2949
return filter_re.search(test_id)
2953
def condition_isinstance(klass_or_klass_list):
2954
"""Create a condition filter which returns isinstance(param, klass).
2956
:return: A callable which when called with one parameter obj return the
2957
result of isinstance(obj, klass_or_klass_list).
2960
return isinstance(obj, klass_or_klass_list)
2964
def condition_id_in_list(id_list):
2965
"""Create a condition filter which verify that test's id in a list.
2967
:param id_list: A TestIdList object.
2968
:return: A callable that returns True if the test's id appears in the list.
2970
def condition(test):
2971
return id_list.includes(test.id())
2975
def condition_id_startswith(starts):
2976
"""Create a condition filter verifying that test's id starts with a string.
2978
:param starts: A list of string.
2979
:return: A callable that returns True if the test's id starts with one of
2982
def condition(test):
2983
for start in starts:
2984
if test.id().startswith(start):
2990
def exclude_tests_by_condition(suite, condition):
2991
"""Create a test suite which excludes some tests from suite.
2993
:param suite: The suite to get tests from.
2994
:param condition: A callable whose result evaluates True when called with a
2995
test case which should be excluded from the result.
2996
:return: A suite which contains the tests found in suite that fail
3000
for test in iter_suite_tests(suite):
3001
if not condition(test):
3003
return TestUtil.TestSuite(result)
3006
def filter_suite_by_condition(suite, condition):
3007
"""Create a test suite by filtering another one.
3009
:param suite: The source suite.
3010
:param condition: A callable whose result evaluates True when called with a
3011
test case which should be included in the result.
3012
:return: A suite which contains the tests found in suite that pass
3016
for test in iter_suite_tests(suite):
3019
return TestUtil.TestSuite(result)
3022
def filter_suite_by_re(suite, pattern):
3023
"""Create a test suite by filtering another one.
3025
:param suite: the source suite
3026
:param pattern: pattern that names must match
3027
:returns: the newly created suite
3029
condition = condition_id_re(pattern)
3030
result_suite = filter_suite_by_condition(suite, condition)
3034
def filter_suite_by_id_list(suite, test_id_list):
3035
"""Create a test suite by filtering another one.
3037
:param suite: The source suite.
3038
:param test_id_list: A list of the test ids to keep as strings.
3039
:returns: the newly created suite
3041
condition = condition_id_in_list(test_id_list)
3042
result_suite = filter_suite_by_condition(suite, condition)
3046
def filter_suite_by_id_startswith(suite, start):
3047
"""Create a test suite by filtering another one.
3049
:param suite: The source suite.
3050
:param start: A list of string the test id must start with one of.
3051
:returns: the newly created suite
3053
condition = condition_id_startswith(start)
3054
result_suite = filter_suite_by_condition(suite, condition)
3058
def exclude_tests_by_re(suite, pattern):
3059
"""Create a test suite which excludes some tests from suite.
3061
:param suite: The suite to get tests from.
3062
:param pattern: A regular expression string. Test ids that match this
3063
pattern will be excluded from the result.
3064
:return: A TestSuite that contains all the tests from suite without the
3065
tests that matched pattern. The order of tests is the same as it was in
3068
return exclude_tests_by_condition(suite, condition_id_re(pattern))
3071
def preserve_input(something):
3072
"""A helper for performing test suite transformation chains.
3074
:param something: Anything you want to preserve.
3080
def randomize_suite(suite):
3081
"""Return a new TestSuite with suite's tests in random order.
3083
The tests in the input suite are flattened into a single suite in order to
3084
accomplish this. Any nested TestSuites are removed to provide global
3087
tests = list(iter_suite_tests(suite))
3088
random.shuffle(tests)
3089
return TestUtil.TestSuite(tests)
3092
def split_suite_by_condition(suite, condition):
3093
"""Split a test suite into two by a condition.
3095
:param suite: The suite to split.
3096
:param condition: The condition to match on. Tests that match this
3097
condition are returned in the first test suite, ones that do not match
3098
are in the second suite.
3099
:return: A tuple of two test suites, where the first contains tests from
3100
suite matching the condition, and the second contains the remainder
3101
from suite. The order within each output suite is the same as it was in
3106
for test in iter_suite_tests(suite):
3108
matched.append(test)
3110
did_not_match.append(test)
3111
return TestUtil.TestSuite(matched), TestUtil.TestSuite(did_not_match)
3114
def split_suite_by_re(suite, pattern):
3115
"""Split a test suite into two by a regular expression.
3117
:param suite: The suite to split.
3118
:param pattern: A regular expression string. Test ids that match this
3119
pattern will be in the first test suite returned, and the others in the
3120
second test suite returned.
3121
:return: A tuple of two test suites, where the first contains tests from
3122
suite matching pattern, and the second contains the remainder from
3123
suite. The order within each output suite is the same as it was in
3126
return split_suite_by_condition(suite, condition_id_re(pattern))
3129
def run_suite(suite, name='test', verbose=False, pattern=".*",
3130
stop_on_failure=False,
3131
transport=None, lsprof_timed=None, bench_history=None,
3132
matching_tests_first=None,
3135
exclude_pattern=None,
3138
suite_decorators=None,
3140
result_decorators=None,
3142
"""Run a test suite for bzr selftest.
3144
:param runner_class: The class of runner to use. Must support the
3145
constructor arguments passed by run_suite which are more than standard
3147
:return: A boolean indicating success.
3149
TestCase._gather_lsprof_in_benchmarks = lsprof_timed
3154
if runner_class is None:
3155
runner_class = TextTestRunner
3158
runner = runner_class(stream=stream,
3160
verbosity=verbosity,
3161
bench_history=bench_history,
3163
result_decorators=result_decorators,
3165
runner.stop_on_failure=stop_on_failure
3166
# built in decorator factories:
3168
random_order(random_seed, runner),
3169
exclude_tests(exclude_pattern),
3171
if matching_tests_first:
3172
decorators.append(tests_first(pattern))
3174
decorators.append(filter_tests(pattern))
3175
if suite_decorators:
3176
decorators.extend(suite_decorators)
3177
# tell the result object how many tests will be running: (except if
3178
# --parallel=fork is being used. Robert said he will provide a better
3179
# progress design later -- vila 20090817)
3180
if fork_decorator not in decorators:
3181
decorators.append(CountingDecorator)
3182
for decorator in decorators:
3183
suite = decorator(suite)
3185
# Done after test suite decoration to allow randomisation etc
3186
# to take effect, though that is of marginal benefit.
3188
stream.write("Listing tests only ...\n")
3189
for t in iter_suite_tests(suite):
3190
stream.write("%s\n" % (t.id()))
3192
result = runner.run(suite)
3194
return result.wasStrictlySuccessful()
3196
return result.wasSuccessful()
3199
# A registry where get() returns a suite decorator.
3200
parallel_registry = registry.Registry()
3203
def fork_decorator(suite):
3204
if getattr(os, "fork", None) is None:
3205
raise errors.BzrCommandError("platform does not support fork,"
3206
" try --parallel=subprocess instead.")
3207
concurrency = osutils.local_concurrency()
3208
if concurrency == 1:
3210
from testtools import ConcurrentTestSuite
3211
return ConcurrentTestSuite(suite, fork_for_tests)
3212
parallel_registry.register('fork', fork_decorator)
3215
def subprocess_decorator(suite):
3216
concurrency = osutils.local_concurrency()
3217
if concurrency == 1:
3219
from testtools import ConcurrentTestSuite
3220
return ConcurrentTestSuite(suite, reinvoke_for_tests)
3221
parallel_registry.register('subprocess', subprocess_decorator)
3224
def exclude_tests(exclude_pattern):
3225
"""Return a test suite decorator that excludes tests."""
3226
if exclude_pattern is None:
3227
return identity_decorator
3228
def decorator(suite):
3229
return ExcludeDecorator(suite, exclude_pattern)
3233
def filter_tests(pattern):
3235
return identity_decorator
3236
def decorator(suite):
3237
return FilterTestsDecorator(suite, pattern)
3241
def random_order(random_seed, runner):
3242
"""Return a test suite decorator factory for randomising tests order.
3244
:param random_seed: now, a string which casts to a long, or a long.
3245
:param runner: A test runner with a stream attribute to report on.
3247
if random_seed is None:
3248
return identity_decorator
3249
def decorator(suite):
3250
return RandomDecorator(suite, random_seed, runner.stream)
3254
def tests_first(pattern):
3256
return identity_decorator
3257
def decorator(suite):
3258
return TestFirstDecorator(suite, pattern)
3262
def identity_decorator(suite):
3267
class TestDecorator(TestUtil.TestSuite):
3268
"""A decorator for TestCase/TestSuite objects.
3270
Usually, subclasses should override __iter__(used when flattening test
3271
suites), which we do to filter, reorder, parallelise and so on, run() and
3275
def __init__(self, suite):
3276
TestUtil.TestSuite.__init__(self)
3279
def countTestCases(self):
3282
cases += test.countTestCases()
3289
def run(self, result):
3290
# Use iteration on self, not self._tests, to allow subclasses to hook
3293
if result.shouldStop:
3299
class CountingDecorator(TestDecorator):
3300
"""A decorator which calls result.progress(self.countTestCases)."""
3302
def run(self, result):
3303
progress_method = getattr(result, 'progress', None)
3304
if callable(progress_method):
3305
progress_method(self.countTestCases(), SUBUNIT_SEEK_SET)
3306
return super(CountingDecorator, self).run(result)
3309
class ExcludeDecorator(TestDecorator):
3310
"""A decorator which excludes test matching an exclude pattern."""
3312
def __init__(self, suite, exclude_pattern):
3313
TestDecorator.__init__(self, suite)
3314
self.exclude_pattern = exclude_pattern
3315
self.excluded = False
3319
return iter(self._tests)
3320
self.excluded = True
3321
suite = exclude_tests_by_re(self, self.exclude_pattern)
3323
self.addTests(suite)
3324
return iter(self._tests)
3327
class FilterTestsDecorator(TestDecorator):
3328
"""A decorator which filters tests to those matching a pattern."""
3330
def __init__(self, suite, pattern):
3331
TestDecorator.__init__(self, suite)
3332
self.pattern = pattern
3333
self.filtered = False
3337
return iter(self._tests)
3338
self.filtered = True
3339
suite = filter_suite_by_re(self, self.pattern)
3341
self.addTests(suite)
3342
return iter(self._tests)
3345
class RandomDecorator(TestDecorator):
3346
"""A decorator which randomises the order of its tests."""
3348
def __init__(self, suite, random_seed, stream):
3349
TestDecorator.__init__(self, suite)
3350
self.random_seed = random_seed
3351
self.randomised = False
3352
self.stream = stream
3356
return iter(self._tests)
3357
self.randomised = True
3358
self.stream.write("Randomizing test order using seed %s\n\n" %
3359
(self.actual_seed()))
3360
# Initialise the random number generator.
3361
random.seed(self.actual_seed())
3362
suite = randomize_suite(self)
3364
self.addTests(suite)
3365
return iter(self._tests)
3367
def actual_seed(self):
3368
if self.random_seed == "now":
3369
# We convert the seed to a long to make it reuseable across
3370
# invocations (because the user can reenter it).
3371
self.random_seed = long(time.time())
3373
# Convert the seed to a long if we can
3375
self.random_seed = long(self.random_seed)
3378
return self.random_seed
3381
class TestFirstDecorator(TestDecorator):
3382
"""A decorator which moves named tests to the front."""
3384
def __init__(self, suite, pattern):
3385
TestDecorator.__init__(self, suite)
3386
self.pattern = pattern
3387
self.filtered = False
3391
return iter(self._tests)
3392
self.filtered = True
3393
suites = split_suite_by_re(self, self.pattern)
3395
self.addTests(suites)
3396
return iter(self._tests)
3399
def partition_tests(suite, count):
3400
"""Partition suite into count lists of tests."""
3401
# This just assigns tests in a round-robin fashion. On one hand this
3402
# splits up blocks of related tests that might run faster if they shared
3403
# resources, but on the other it avoids assigning blocks of slow tests to
3404
# just one partition. So the slowest partition shouldn't be much slower
3406
partitions = [list() for i in range(count)]
3407
tests = iter_suite_tests(suite)
3408
for partition, test in itertools.izip(itertools.cycle(partitions), tests):
3409
partition.append(test)
3413
def workaround_zealous_crypto_random():
3414
"""Crypto.Random want to help us being secure, but we don't care here.
3416
This workaround some test failure related to the sftp server. Once paramiko
3417
stop using the controversial API in Crypto.Random, we may get rid of it.
3420
from Crypto.Random import atfork
3426
def fork_for_tests(suite):
3427
"""Take suite and start up one runner per CPU by forking()
3429
:return: An iterable of TestCase-like objects which can each have
3430
run(result) called on them to feed tests to result.
3432
concurrency = osutils.local_concurrency()
3434
from subunit import TestProtocolClient, ProtocolTestCase
3435
from subunit.test_results import AutoTimingTestResultDecorator
3436
class TestInOtherProcess(ProtocolTestCase):
3437
# Should be in subunit, I think. RBC.
3438
def __init__(self, stream, pid):
3439
ProtocolTestCase.__init__(self, stream)
3442
def run(self, result):
3444
ProtocolTestCase.run(self, result)
3446
os.waitpid(self.pid, 0)
3448
test_blocks = partition_tests(suite, concurrency)
3449
for process_tests in test_blocks:
3450
process_suite = TestUtil.TestSuite()
3451
process_suite.addTests(process_tests)
3452
c2pread, c2pwrite = os.pipe()
3455
workaround_zealous_crypto_random()
3458
# Leave stderr and stdout open so we can see test noise
3459
# Close stdin so that the child goes away if it decides to
3460
# read from stdin (otherwise its a roulette to see what
3461
# child actually gets keystrokes for pdb etc).
3464
stream = os.fdopen(c2pwrite, 'wb', 1)
3465
subunit_result = AutoTimingTestResultDecorator(
3466
TestProtocolClient(stream))
3467
process_suite.run(subunit_result)
3472
stream = os.fdopen(c2pread, 'rb', 1)
3473
test = TestInOtherProcess(stream, pid)
3478
def reinvoke_for_tests(suite):
3479
"""Take suite and start up one runner per CPU using subprocess().
3481
:return: An iterable of TestCase-like objects which can each have
3482
run(result) called on them to feed tests to result.
3484
concurrency = osutils.local_concurrency()
3486
from subunit import ProtocolTestCase
3487
class TestInSubprocess(ProtocolTestCase):
3488
def __init__(self, process, name):
3489
ProtocolTestCase.__init__(self, process.stdout)
3490
self.process = process
3491
self.process.stdin.close()
3494
def run(self, result):
3496
ProtocolTestCase.run(self, result)
3499
os.unlink(self.name)
3500
# print "pid %d finished" % finished_process
3501
test_blocks = partition_tests(suite, concurrency)
3502
for process_tests in test_blocks:
3503
# ugly; currently reimplement rather than reuses TestCase methods.
3504
bzr_path = os.path.dirname(os.path.dirname(bzrlib.__file__))+'/bzr'
3505
if not os.path.isfile(bzr_path):
3506
# We are probably installed. Assume sys.argv is the right file
3507
bzr_path = sys.argv[0]
3508
bzr_path = [bzr_path]
3509
if sys.platform == "win32":
3510
# if we're on windows, we can't execute the bzr script directly
3511
bzr_path = [sys.executable] + bzr_path
3512
fd, test_list_file_name = tempfile.mkstemp()
3513
test_list_file = os.fdopen(fd, 'wb', 1)
3514
for test in process_tests:
3515
test_list_file.write(test.id() + '\n')
3516
test_list_file.close()
3518
argv = bzr_path + ['selftest', '--load-list', test_list_file_name,
3520
if '--no-plugins' in sys.argv:
3521
argv.append('--no-plugins')
3522
# stderr=subprocess.STDOUT would be ideal, but until we prevent
3523
# noise on stderr it can interrupt the subunit protocol.
3524
process = subprocess.Popen(argv, stdin=subprocess.PIPE,
3525
stdout=subprocess.PIPE,
3526
stderr=subprocess.PIPE,
3528
test = TestInSubprocess(process, test_list_file_name)
3531
os.unlink(test_list_file_name)
3536
class ProfileResult(testtools.ExtendedToOriginalDecorator):
3537
"""Generate profiling data for all activity between start and success.
3539
The profile data is appended to the test's _benchcalls attribute and can
3540
be accessed by the forwarded-to TestResult.
3542
While it might be cleaner do accumulate this in stopTest, addSuccess is
3543
where our existing output support for lsprof is, and this class aims to
3544
fit in with that: while it could be moved it's not necessary to accomplish
3545
test profiling, nor would it be dramatically cleaner.
3548
def startTest(self, test):
3549
self.profiler = bzrlib.lsprof.BzrProfiler()
3550
# Prevent deadlocks in tests that use lsprof: those tests will
3552
bzrlib.lsprof.BzrProfiler.profiler_block = 0
3553
self.profiler.start()
3554
testtools.ExtendedToOriginalDecorator.startTest(self, test)
3556
def addSuccess(self, test):
3557
stats = self.profiler.stop()
3559
calls = test._benchcalls
3560
except AttributeError:
3561
test._benchcalls = []
3562
calls = test._benchcalls
3563
calls.append(((test.id(), "", ""), stats))
3564
testtools.ExtendedToOriginalDecorator.addSuccess(self, test)
3566
def stopTest(self, test):
3567
testtools.ExtendedToOriginalDecorator.stopTest(self, test)
3568
self.profiler = None
3571
# Controlled by "bzr selftest -E=..." option
3572
# Currently supported:
3573
# -Eallow_debug Will no longer clear debug.debug_flags() so it
3574
# preserves any flags supplied at the command line.
3575
# -Edisable_lock_checks Turns errors in mismatched locks into simple prints
3576
# rather than failing tests. And no longer raise
3577
# LockContention when fctnl locks are not being used
3578
# with proper exclusion rules.
3579
# -Ethreads Will display thread ident at creation/join time to
3580
# help track thread leaks
3582
# -Econfig_stats Will collect statistics using addDetail
3583
selftest_debug_flags = set()
3586
def selftest(verbose=False, pattern=".*", stop_on_failure=True,
3588
test_suite_factory=None,
3591
matching_tests_first=None,
3594
exclude_pattern=None,
3600
suite_decorators=None,
3604
"""Run the whole test suite under the enhanced runner"""
3605
# XXX: Very ugly way to do this...
3606
# Disable warning about old formats because we don't want it to disturb
3607
# any blackbox tests.
3608
from bzrlib import repository
3609
repository._deprecation_warning_done = True
3611
global default_transport
3612
if transport is None:
3613
transport = default_transport
3614
old_transport = default_transport
3615
default_transport = transport
3616
global selftest_debug_flags
3617
old_debug_flags = selftest_debug_flags
3618
if debug_flags is not None:
3619
selftest_debug_flags = set(debug_flags)
3621
if load_list is None:
3624
keep_only = load_test_id_list(load_list)
3626
starting_with = [test_prefix_alias_registry.resolve_alias(start)
3627
for start in starting_with]
3628
if test_suite_factory is None:
3629
# Reduce loading time by loading modules based on the starting_with
3631
suite = test_suite(keep_only, starting_with)
3633
suite = test_suite_factory()
3635
# But always filter as requested.
3636
suite = filter_suite_by_id_startswith(suite, starting_with)
3637
result_decorators = []
3639
result_decorators.append(ProfileResult)
3640
return run_suite(suite, 'testbzr', verbose=verbose, pattern=pattern,
3641
stop_on_failure=stop_on_failure,
3642
transport=transport,
3643
lsprof_timed=lsprof_timed,
3644
bench_history=bench_history,
3645
matching_tests_first=matching_tests_first,
3646
list_only=list_only,
3647
random_seed=random_seed,
3648
exclude_pattern=exclude_pattern,
3650
runner_class=runner_class,
3651
suite_decorators=suite_decorators,
3653
result_decorators=result_decorators,
3656
default_transport = old_transport
3657
selftest_debug_flags = old_debug_flags
3660
def load_test_id_list(file_name):
3661
"""Load a test id list from a text file.
3663
The format is one test id by line. No special care is taken to impose
3664
strict rules, these test ids are used to filter the test suite so a test id
3665
that do not match an existing test will do no harm. This allows user to add
3666
comments, leave blank lines, etc.
3670
ftest = open(file_name, 'rt')
3672
if e.errno != errno.ENOENT:
3675
raise errors.NoSuchFile(file_name)
3677
for test_name in ftest.readlines():
3678
test_list.append(test_name.strip())
3683
def suite_matches_id_list(test_suite, id_list):
3684
"""Warns about tests not appearing or appearing more than once.
3686
:param test_suite: A TestSuite object.
3687
:param test_id_list: The list of test ids that should be found in
3690
:return: (absents, duplicates) absents is a list containing the test found
3691
in id_list but not in test_suite, duplicates is a list containing the
3692
test found multiple times in test_suite.
3694
When using a prefined test id list, it may occurs that some tests do not
3695
exist anymore or that some tests use the same id. This function warns the
3696
tester about potential problems in his workflow (test lists are volatile)
3697
or in the test suite itself (using the same id for several tests does not
3698
help to localize defects).
3700
# Build a dict counting id occurrences
3702
for test in iter_suite_tests(test_suite):
3704
tests[id] = tests.get(id, 0) + 1
3709
occurs = tests.get(id, 0)
3711
not_found.append(id)
3713
duplicates.append(id)
3715
return not_found, duplicates
3718
class TestIdList(object):
3719
"""Test id list to filter a test suite.
3721
Relying on the assumption that test ids are built as:
3722
<module>[.<class>.<method>][(<param>+)], <module> being in python dotted
3723
notation, this class offers methods to :
3724
- avoid building a test suite for modules not refered to in the test list,
3725
- keep only the tests listed from the module test suite.
3728
def __init__(self, test_id_list):
3729
# When a test suite needs to be filtered against us we compare test ids
3730
# for equality, so a simple dict offers a quick and simple solution.
3731
self.tests = dict().fromkeys(test_id_list, True)
3733
# While unittest.TestCase have ids like:
3734
# <module>.<class>.<method>[(<param+)],
3735
# doctest.DocTestCase can have ids like:
3738
# <module>.<function>
3739
# <module>.<class>.<method>
3741
# Since we can't predict a test class from its name only, we settle on
3742
# a simple constraint: a test id always begins with its module name.
3745
for test_id in test_id_list:
3746
parts = test_id.split('.')
3747
mod_name = parts.pop(0)
3748
modules[mod_name] = True
3750
mod_name += '.' + part
3751
modules[mod_name] = True
3752
self.modules = modules
3754
def refers_to(self, module_name):
3755
"""Is there tests for the module or one of its sub modules."""
3756
return self.modules.has_key(module_name)
3758
def includes(self, test_id):
3759
return self.tests.has_key(test_id)
3762
class TestPrefixAliasRegistry(registry.Registry):
3763
"""A registry for test prefix aliases.
3765
This helps implement shorcuts for the --starting-with selftest
3766
option. Overriding existing prefixes is not allowed but not fatal (a
3767
warning will be emitted).
3770
def register(self, key, obj, help=None, info=None,
3771
override_existing=False):
3772
"""See Registry.register.
3774
Trying to override an existing alias causes a warning to be emitted,
3775
not a fatal execption.
3778
super(TestPrefixAliasRegistry, self).register(
3779
key, obj, help=help, info=info, override_existing=False)
3781
actual = self.get(key)
3783
'Test prefix alias %s is already used for %s, ignoring %s'
3784
% (key, actual, obj))
3786
def resolve_alias(self, id_start):
3787
"""Replace the alias by the prefix in the given string.
3789
Using an unknown prefix is an error to help catching typos.
3791
parts = id_start.split('.')
3793
parts[0] = self.get(parts[0])
3795
raise errors.BzrCommandError(
3796
'%s is not a known test prefix alias' % parts[0])
3797
return '.'.join(parts)
3800
test_prefix_alias_registry = TestPrefixAliasRegistry()
3801
"""Registry of test prefix aliases."""
3804
# This alias allows to detect typos ('bzrlin.') by making all valid test ids
3805
# appear prefixed ('bzrlib.' is "replaced" by 'bzrlib.').
3806
test_prefix_alias_registry.register('bzrlib', 'bzrlib')
3808
# Obvious highest levels prefixes, feel free to add your own via a plugin
3809
test_prefix_alias_registry.register('bd', 'bzrlib.doc')
3810
test_prefix_alias_registry.register('bu', 'bzrlib.utils')
3811
test_prefix_alias_registry.register('bt', 'bzrlib.tests')
3812
test_prefix_alias_registry.register('bb', 'bzrlib.tests.blackbox')
3813
test_prefix_alias_registry.register('bp', 'bzrlib.plugins')
3816
def _test_suite_testmod_names():
3817
"""Return the standard list of test module names to test."""
3820
'bzrlib.tests.blackbox',
3821
'bzrlib.tests.commands',
3822
'bzrlib.tests.doc_generate',
3823
'bzrlib.tests.per_branch',
3824
'bzrlib.tests.per_bzrdir',
3825
'bzrlib.tests.per_controldir',
3826
'bzrlib.tests.per_controldir_colo',
3827
'bzrlib.tests.per_foreign_vcs',
3828
'bzrlib.tests.per_interrepository',
3829
'bzrlib.tests.per_intertree',
3830
'bzrlib.tests.per_inventory',
3831
'bzrlib.tests.per_interbranch',
3832
'bzrlib.tests.per_lock',
3833
'bzrlib.tests.per_merger',
3834
'bzrlib.tests.per_transport',
3835
'bzrlib.tests.per_tree',
3836
'bzrlib.tests.per_pack_repository',
3837
'bzrlib.tests.per_repository',
3838
'bzrlib.tests.per_repository_chk',
3839
'bzrlib.tests.per_repository_reference',
3840
'bzrlib.tests.per_repository_vf',
3841
'bzrlib.tests.per_uifactory',
3842
'bzrlib.tests.per_versionedfile',
3843
'bzrlib.tests.per_workingtree',
3844
'bzrlib.tests.test__annotator',
3845
'bzrlib.tests.test__bencode',
3846
'bzrlib.tests.test__btree_serializer',
3847
'bzrlib.tests.test__chk_map',
3848
'bzrlib.tests.test__dirstate_helpers',
3849
'bzrlib.tests.test__groupcompress',
3850
'bzrlib.tests.test__known_graph',
3851
'bzrlib.tests.test__rio',
3852
'bzrlib.tests.test__simple_set',
3853
'bzrlib.tests.test__static_tuple',
3854
'bzrlib.tests.test__walkdirs_win32',
3855
'bzrlib.tests.test_ancestry',
3856
'bzrlib.tests.test_annotate',
3857
'bzrlib.tests.test_api',
3858
'bzrlib.tests.test_atomicfile',
3859
'bzrlib.tests.test_bad_files',
3860
'bzrlib.tests.test_bisect_multi',
3861
'bzrlib.tests.test_branch',
3862
'bzrlib.tests.test_branchbuilder',
3863
'bzrlib.tests.test_btree_index',
3864
'bzrlib.tests.test_bugtracker',
3865
'bzrlib.tests.test_bundle',
3866
'bzrlib.tests.test_bzrdir',
3867
'bzrlib.tests.test__chunks_to_lines',
3868
'bzrlib.tests.test_cache_utf8',
3869
'bzrlib.tests.test_chk_map',
3870
'bzrlib.tests.test_chk_serializer',
3871
'bzrlib.tests.test_chunk_writer',
3872
'bzrlib.tests.test_clean_tree',
3873
'bzrlib.tests.test_cleanup',
3874
'bzrlib.tests.test_cmdline',
3875
'bzrlib.tests.test_commands',
3876
'bzrlib.tests.test_commit',
3877
'bzrlib.tests.test_commit_merge',
3878
'bzrlib.tests.test_config',
3879
'bzrlib.tests.test_conflicts',
3880
'bzrlib.tests.test_controldir',
3881
'bzrlib.tests.test_counted_lock',
3882
'bzrlib.tests.test_crash',
3883
'bzrlib.tests.test_decorators',
3884
'bzrlib.tests.test_delta',
3885
'bzrlib.tests.test_debug',
3886
'bzrlib.tests.test_diff',
3887
'bzrlib.tests.test_directory_service',
3888
'bzrlib.tests.test_dirstate',
3889
'bzrlib.tests.test_email_message',
3890
'bzrlib.tests.test_eol_filters',
3891
'bzrlib.tests.test_errors',
3892
'bzrlib.tests.test_export',
3893
'bzrlib.tests.test_export_pot',
3894
'bzrlib.tests.test_extract',
3895
'bzrlib.tests.test_fetch',
3896
'bzrlib.tests.test_fixtures',
3897
'bzrlib.tests.test_fifo_cache',
3898
'bzrlib.tests.test_filters',
3899
'bzrlib.tests.test_ftp_transport',
3900
'bzrlib.tests.test_foreign',
3901
'bzrlib.tests.test_generate_docs',
3902
'bzrlib.tests.test_generate_ids',
3903
'bzrlib.tests.test_globbing',
3904
'bzrlib.tests.test_gpg',
3905
'bzrlib.tests.test_graph',
3906
'bzrlib.tests.test_groupcompress',
3907
'bzrlib.tests.test_hashcache',
3908
'bzrlib.tests.test_help',
3909
'bzrlib.tests.test_hooks',
3910
'bzrlib.tests.test_http',
3911
'bzrlib.tests.test_http_response',
3912
'bzrlib.tests.test_https_ca_bundle',
3913
'bzrlib.tests.test_i18n',
3914
'bzrlib.tests.test_identitymap',
3915
'bzrlib.tests.test_ignores',
3916
'bzrlib.tests.test_index',
3917
'bzrlib.tests.test_import_tariff',
3918
'bzrlib.tests.test_info',
3919
'bzrlib.tests.test_inv',
3920
'bzrlib.tests.test_inventory_delta',
3921
'bzrlib.tests.test_knit',
3922
'bzrlib.tests.test_lazy_import',
3923
'bzrlib.tests.test_lazy_regex',
3924
'bzrlib.tests.test_library_state',
3925
'bzrlib.tests.test_lock',
3926
'bzrlib.tests.test_lockable_files',
3927
'bzrlib.tests.test_lockdir',
3928
'bzrlib.tests.test_log',
3929
'bzrlib.tests.test_lru_cache',
3930
'bzrlib.tests.test_lsprof',
3931
'bzrlib.tests.test_mail_client',
3932
'bzrlib.tests.test_matchers',
3933
'bzrlib.tests.test_memorytree',
3934
'bzrlib.tests.test_merge',
3935
'bzrlib.tests.test_merge3',
3936
'bzrlib.tests.test_merge_core',
3937
'bzrlib.tests.test_merge_directive',
3938
'bzrlib.tests.test_mergetools',
3939
'bzrlib.tests.test_missing',
3940
'bzrlib.tests.test_msgeditor',
3941
'bzrlib.tests.test_multiparent',
3942
'bzrlib.tests.test_mutabletree',
3943
'bzrlib.tests.test_nonascii',
3944
'bzrlib.tests.test_options',
3945
'bzrlib.tests.test_osutils',
3946
'bzrlib.tests.test_osutils_encodings',
3947
'bzrlib.tests.test_pack',
3948
'bzrlib.tests.test_patch',
3949
'bzrlib.tests.test_patches',
3950
'bzrlib.tests.test_permissions',
3951
'bzrlib.tests.test_plugins',
3952
'bzrlib.tests.test_progress',
3953
'bzrlib.tests.test_pyutils',
3954
'bzrlib.tests.test_read_bundle',
3955
'bzrlib.tests.test_reconcile',
3956
'bzrlib.tests.test_reconfigure',
3957
'bzrlib.tests.test_registry',
3958
'bzrlib.tests.test_remote',
3959
'bzrlib.tests.test_rename_map',
3960
'bzrlib.tests.test_repository',
3961
'bzrlib.tests.test_revert',
3962
'bzrlib.tests.test_revision',
3963
'bzrlib.tests.test_revisionspec',
3964
'bzrlib.tests.test_revisiontree',
3965
'bzrlib.tests.test_rio',
3966
'bzrlib.tests.test_rules',
3967
'bzrlib.tests.test_sampler',
3968
'bzrlib.tests.test_scenarios',
3969
'bzrlib.tests.test_script',
3970
'bzrlib.tests.test_selftest',
3971
'bzrlib.tests.test_serializer',
3972
'bzrlib.tests.test_setup',
3973
'bzrlib.tests.test_sftp_transport',
3974
'bzrlib.tests.test_shelf',
3975
'bzrlib.tests.test_shelf_ui',
3976
'bzrlib.tests.test_smart',
3977
'bzrlib.tests.test_smart_add',
3978
'bzrlib.tests.test_smart_request',
3979
'bzrlib.tests.test_smart_transport',
3980
'bzrlib.tests.test_smtp_connection',
3981
'bzrlib.tests.test_source',
3982
'bzrlib.tests.test_ssh_transport',
3983
'bzrlib.tests.test_status',
3984
'bzrlib.tests.test_store',
3985
'bzrlib.tests.test_strace',
3986
'bzrlib.tests.test_subsume',
3987
'bzrlib.tests.test_switch',
3988
'bzrlib.tests.test_symbol_versioning',
3989
'bzrlib.tests.test_tag',
3990
'bzrlib.tests.test_test_server',
3991
'bzrlib.tests.test_testament',
3992
'bzrlib.tests.test_textfile',
3993
'bzrlib.tests.test_textmerge',
3994
'bzrlib.tests.test_cethread',
3995
'bzrlib.tests.test_timestamp',
3996
'bzrlib.tests.test_trace',
3997
'bzrlib.tests.test_transactions',
3998
'bzrlib.tests.test_transform',
3999
'bzrlib.tests.test_transport',
4000
'bzrlib.tests.test_transport_log',
4001
'bzrlib.tests.test_tree',
4002
'bzrlib.tests.test_treebuilder',
4003
'bzrlib.tests.test_treeshape',
4004
'bzrlib.tests.test_tsort',
4005
'bzrlib.tests.test_tuned_gzip',
4006
'bzrlib.tests.test_ui',
4007
'bzrlib.tests.test_uncommit',
4008
'bzrlib.tests.test_upgrade',
4009
'bzrlib.tests.test_upgrade_stacked',
4010
'bzrlib.tests.test_urlutils',
4011
'bzrlib.tests.test_utextwrap',
4012
'bzrlib.tests.test_version',
4013
'bzrlib.tests.test_version_info',
4014
'bzrlib.tests.test_versionedfile',
4015
'bzrlib.tests.test_weave',
4016
'bzrlib.tests.test_whitebox',
4017
'bzrlib.tests.test_win32utils',
4018
'bzrlib.tests.test_workingtree',
4019
'bzrlib.tests.test_workingtree_4',
4020
'bzrlib.tests.test_wsgi',
4021
'bzrlib.tests.test_xml',
4025
def _test_suite_modules_to_doctest():
4026
"""Return the list of modules to doctest."""
4028
# GZ 2009-03-31: No docstrings with -OO so there's nothing to doctest
4032
'bzrlib.branchbuilder',
4033
'bzrlib.decorators',
4035
'bzrlib.iterablefile',
4040
'bzrlib.symbol_versioning',
4042
'bzrlib.tests.fixtures',
4044
'bzrlib.transport.http',
4045
'bzrlib.version_info_formats.format_custom',
4049
def test_suite(keep_only=None, starting_with=None):
4050
"""Build and return TestSuite for the whole of bzrlib.
4052
:param keep_only: A list of test ids limiting the suite returned.
4054
:param starting_with: An id limiting the suite returned to the tests
4057
This function can be replaced if you need to change the default test
4058
suite on a global basis, but it is not encouraged.
4061
loader = TestUtil.TestLoader()
4063
if keep_only is not None:
4064
id_filter = TestIdList(keep_only)
4066
# We take precedence over keep_only because *at loading time* using
4067
# both options means we will load less tests for the same final result.
4068
def interesting_module(name):
4069
for start in starting_with:
4071
# Either the module name starts with the specified string
4072
name.startswith(start)
4073
# or it may contain tests starting with the specified string
4074
or start.startswith(name)
4078
loader = TestUtil.FilteredByModuleTestLoader(interesting_module)
4080
elif keep_only is not None:
4081
loader = TestUtil.FilteredByModuleTestLoader(id_filter.refers_to)
4082
def interesting_module(name):
4083
return id_filter.refers_to(name)
4086
loader = TestUtil.TestLoader()
4087
def interesting_module(name):
4088
# No filtering, all modules are interesting
4091
suite = loader.suiteClass()
4093
# modules building their suite with loadTestsFromModuleNames
4094
suite.addTest(loader.loadTestsFromModuleNames(_test_suite_testmod_names()))
4096
for mod in _test_suite_modules_to_doctest():
4097
if not interesting_module(mod):
4098
# No tests to keep here, move along
4101
# note that this really does mean "report only" -- doctest
4102
# still runs the rest of the examples
4103
doc_suite = IsolatedDocTestSuite(
4104
mod, optionflags=doctest.REPORT_ONLY_FIRST_FAILURE)
4105
except ValueError, e:
4106
print '**failed to get doctest for: %s\n%s' % (mod, e)
4108
if len(doc_suite._tests) == 0:
4109
raise errors.BzrError("no doctests found in %s" % (mod,))
4110
suite.addTest(doc_suite)
4112
default_encoding = sys.getdefaultencoding()
4113
for name, plugin in _mod_plugin.plugins().items():
4114
if not interesting_module(plugin.module.__name__):
4116
plugin_suite = plugin.test_suite()
4117
# We used to catch ImportError here and turn it into just a warning,
4118
# but really if you don't have --no-plugins this should be a failure.
4119
# mbp 20080213 - see http://bugs.launchpad.net/bugs/189771
4120
if plugin_suite is None:
4121
plugin_suite = plugin.load_plugin_tests(loader)
4122
if plugin_suite is not None:
4123
suite.addTest(plugin_suite)
4124
if default_encoding != sys.getdefaultencoding():
4126
'Plugin "%s" tried to reset default encoding to: %s', name,
4127
sys.getdefaultencoding())
4129
sys.setdefaultencoding(default_encoding)
4131
if keep_only is not None:
4132
# Now that the referred modules have loaded their tests, keep only the
4134
suite = filter_suite_by_id_list(suite, id_filter)
4135
# Do some sanity checks on the id_list filtering
4136
not_found, duplicates = suite_matches_id_list(suite, keep_only)
4138
# The tester has used both keep_only and starting_with, so he is
4139
# already aware that some tests are excluded from the list, there
4140
# is no need to tell him which.
4143
# Some tests mentioned in the list are not in the test suite. The
4144
# list may be out of date, report to the tester.
4145
for id in not_found:
4146
trace.warning('"%s" not found in the test suite', id)
4147
for id in duplicates:
4148
trace.warning('"%s" is used as an id by several tests', id)
4153
def multiply_scenarios(*scenarios):
4154
"""Multiply two or more iterables of scenarios.
4156
It is safe to pass scenario generators or iterators.
4158
:returns: A list of compound scenarios: the cross-product of all
4159
scenarios, with the names concatenated and the parameters
4162
return reduce(_multiply_two_scenarios, map(list, scenarios))
4165
def _multiply_two_scenarios(scenarios_left, scenarios_right):
4166
"""Multiply two sets of scenarios.
4168
:returns: the cartesian product of the two sets of scenarios, that is
4169
a scenario for every possible combination of a left scenario and a
4173
('%s,%s' % (left_name, right_name),
4174
dict(left_dict.items() + right_dict.items()))
4175
for left_name, left_dict in scenarios_left
4176
for right_name, right_dict in scenarios_right]
4179
def multiply_tests(tests, scenarios, result):
4180
"""Multiply tests_list by scenarios into result.
4182
This is the core workhorse for test parameterisation.
4184
Typically the load_tests() method for a per-implementation test suite will
4185
call multiply_tests and return the result.
4187
:param tests: The tests to parameterise.
4188
:param scenarios: The scenarios to apply: pairs of (scenario_name,
4189
scenario_param_dict).
4190
:param result: A TestSuite to add created tests to.
4192
This returns the passed in result TestSuite with the cross product of all
4193
the tests repeated once for each scenario. Each test is adapted by adding
4194
the scenario name at the end of its id(), and updating the test object's
4195
__dict__ with the scenario_param_dict.
4197
>>> import bzrlib.tests.test_sampler
4198
>>> r = multiply_tests(
4199
... bzrlib.tests.test_sampler.DemoTest('test_nothing'),
4200
... [('one', dict(param=1)),
4201
... ('two', dict(param=2))],
4202
... TestUtil.TestSuite())
4203
>>> tests = list(iter_suite_tests(r))
4207
'bzrlib.tests.test_sampler.DemoTest.test_nothing(one)'
4213
for test in iter_suite_tests(tests):
4214
apply_scenarios(test, scenarios, result)
4218
def apply_scenarios(test, scenarios, result):
4219
"""Apply the scenarios in scenarios to test and add to result.
4221
:param test: The test to apply scenarios to.
4222
:param scenarios: An iterable of scenarios to apply to test.
4224
:seealso: apply_scenario
4226
for scenario in scenarios:
4227
result.addTest(apply_scenario(test, scenario))
4231
def apply_scenario(test, scenario):
4232
"""Copy test and apply scenario to it.
4234
:param test: A test to adapt.
4235
:param scenario: A tuple describing the scenarion.
4236
The first element of the tuple is the new test id.
4237
The second element is a dict containing attributes to set on the
4239
:return: The adapted test.
4241
new_id = "%s(%s)" % (test.id(), scenario[0])
4242
new_test = clone_test(test, new_id)
4243
for name, value in scenario[1].items():
4244
setattr(new_test, name, value)
4248
def clone_test(test, new_id):
4249
"""Clone a test giving it a new id.
4251
:param test: The test to clone.
4252
:param new_id: The id to assign to it.
4253
:return: The new test.
4255
new_test = copy.copy(test)
4256
new_test.id = lambda: new_id
4257
# XXX: Workaround <https://bugs.launchpad.net/testtools/+bug/637725>, which
4258
# causes cloned tests to share the 'details' dict. This makes it hard to
4259
# read the test output for parameterized tests, because tracebacks will be
4260
# associated with irrelevant tests.
4262
details = new_test._TestCase__details
4263
except AttributeError:
4264
# must be a different version of testtools than expected. Do nothing.
4267
# Reset the '__details' dict.
4268
new_test._TestCase__details = {}
4272
def permute_tests_for_extension(standard_tests, loader, py_module_name,
4274
"""Helper for permutating tests against an extension module.
4276
This is meant to be used inside a modules 'load_tests()' function. It will
4277
create 2 scenarios, and cause all tests in the 'standard_tests' to be run
4278
against both implementations. Setting 'test.module' to the appropriate
4279
module. See bzrlib.tests.test__chk_map.load_tests as an example.
4281
:param standard_tests: A test suite to permute
4282
:param loader: A TestLoader
4283
:param py_module_name: The python path to a python module that can always
4284
be loaded, and will be considered the 'python' implementation. (eg
4285
'bzrlib._chk_map_py')
4286
:param ext_module_name: The python path to an extension module. If the
4287
module cannot be loaded, a single test will be added, which notes that
4288
the module is not available. If it can be loaded, all standard_tests
4289
will be run against that module.
4290
:return: (suite, feature) suite is a test-suite that has all the permuted
4291
tests. feature is the Feature object that can be used to determine if
4292
the module is available.
4295
py_module = pyutils.get_named_object(py_module_name)
4297
('python', {'module': py_module}),
4299
suite = loader.suiteClass()
4300
feature = ModuleAvailableFeature(ext_module_name)
4301
if feature.available():
4302
scenarios.append(('C', {'module': feature.module}))
4304
# the compiled module isn't available, so we add a failing test
4305
class FailWithoutFeature(TestCase):
4306
def test_fail(self):
4307
self.requireFeature(feature)
4308
suite.addTest(loader.loadTestsFromTestCase(FailWithoutFeature))
4309
result = multiply_tests(standard_tests, scenarios, suite)
4310
return result, feature
4313
def _rmtree_temp_dir(dirname, test_id=None):
4314
# If LANG=C we probably have created some bogus paths
4315
# which rmtree(unicode) will fail to delete
4316
# so make sure we are using rmtree(str) to delete everything
4317
# except on win32, where rmtree(str) will fail
4318
# since it doesn't have the property of byte-stream paths
4319
# (they are either ascii or mbcs)
4320
if sys.platform == 'win32':
4321
# make sure we are using the unicode win32 api
4322
dirname = unicode(dirname)
4324
dirname = dirname.encode(sys.getfilesystemencoding())
4326
osutils.rmtree(dirname)
4328
# We don't want to fail here because some useful display will be lost
4329
# otherwise. Polluting the tmp dir is bad, but not giving all the
4330
# possible info to the test runner is even worse.
4332
ui.ui_factory.clear_term()
4333
sys.stderr.write('\nWhile running: %s\n' % (test_id,))
4334
# Ugly, but the last thing we want here is fail, so bear with it.
4335
printable_e = str(e).decode(osutils.get_user_encoding(), 'replace'
4336
).encode('ascii', 'replace')
4337
sys.stderr.write('Unable to remove testing dir %s\n%s'
4338
% (os.path.basename(dirname), printable_e))
4341
class Feature(object):
4342
"""An operating system Feature."""
4345
self._available = None
4347
def available(self):
4348
"""Is the feature available?
4350
:return: True if the feature is available.
4352
if self._available is None:
4353
self._available = self._probe()
4354
return self._available
4357
"""Implement this method in concrete features.
4359
:return: True if the feature is available.
4361
raise NotImplementedError
4364
if getattr(self, 'feature_name', None):
4365
return self.feature_name()
4366
return self.__class__.__name__
4369
class _SymlinkFeature(Feature):
4372
return osutils.has_symlinks()
4374
def feature_name(self):
4377
SymlinkFeature = _SymlinkFeature()
4380
class _HardlinkFeature(Feature):
4383
return osutils.has_hardlinks()
4385
def feature_name(self):
4388
HardlinkFeature = _HardlinkFeature()
4391
class _OsFifoFeature(Feature):
4394
return getattr(os, 'mkfifo', None)
4396
def feature_name(self):
4397
return 'filesystem fifos'
4399
OsFifoFeature = _OsFifoFeature()
4402
class _UnicodeFilenameFeature(Feature):
4403
"""Does the filesystem support Unicode filenames?"""
4407
# Check for character combinations unlikely to be covered by any
4408
# single non-unicode encoding. We use the characters
4409
# - greek small letter alpha (U+03B1) and
4410
# - braille pattern dots-123456 (U+283F).
4411
os.stat(u'\u03b1\u283f')
4412
except UnicodeEncodeError:
4414
except (IOError, OSError):
4415
# The filesystem allows the Unicode filename but the file doesn't
4419
# The filesystem allows the Unicode filename and the file exists,
4423
UnicodeFilenameFeature = _UnicodeFilenameFeature()
4426
class _CompatabilityThunkFeature(Feature):
4427
"""This feature is just a thunk to another feature.
4429
It issues a deprecation warning if it is accessed, to let you know that you
4430
should really use a different feature.
4433
def __init__(self, dep_version, module, name,
4434
replacement_name, replacement_module=None):
4435
super(_CompatabilityThunkFeature, self).__init__()
4436
self._module = module
4437
if replacement_module is None:
4438
replacement_module = module
4439
self._replacement_module = replacement_module
4441
self._replacement_name = replacement_name
4442
self._dep_version = dep_version
4443
self._feature = None
4446
if self._feature is None:
4447
depr_msg = self._dep_version % ('%s.%s'
4448
% (self._module, self._name))
4449
use_msg = ' Use %s.%s instead.' % (self._replacement_module,
4450
self._replacement_name)
4451
symbol_versioning.warn(depr_msg + use_msg, DeprecationWarning)
4452
# Import the new feature and use it as a replacement for the
4454
self._feature = pyutils.get_named_object(
4455
self._replacement_module, self._replacement_name)
4459
return self._feature._probe()
4462
class ModuleAvailableFeature(Feature):
4463
"""This is a feature than describes a module we want to be available.
4465
Declare the name of the module in __init__(), and then after probing, the
4466
module will be available as 'self.module'.
4468
:ivar module: The module if it is available, else None.
4471
def __init__(self, module_name):
4472
super(ModuleAvailableFeature, self).__init__()
4473
self.module_name = module_name
4477
self._module = __import__(self.module_name, {}, {}, [''])
4484
if self.available(): # Make sure the probe has been done
4488
def feature_name(self):
4489
return self.module_name
4492
def probe_unicode_in_user_encoding():
4493
"""Try to encode several unicode strings to use in unicode-aware tests.
4494
Return first successfull match.
4496
:return: (unicode value, encoded plain string value) or (None, None)
4498
possible_vals = [u'm\xb5', u'\xe1', u'\u0410']
4499
for uni_val in possible_vals:
4501
str_val = uni_val.encode(osutils.get_user_encoding())
4502
except UnicodeEncodeError:
4503
# Try a different character
4506
return uni_val, str_val
4510
def probe_bad_non_ascii(encoding):
4511
"""Try to find [bad] character with code [128..255]
4512
that cannot be decoded to unicode in some encoding.
4513
Return None if all non-ascii characters is valid
4516
for i in xrange(128, 256):
4519
char.decode(encoding)
4520
except UnicodeDecodeError:
4525
class _HTTPSServerFeature(Feature):
4526
"""Some tests want an https Server, check if one is available.
4528
Right now, the only way this is available is under python2.6 which provides
4539
def feature_name(self):
4540
return 'HTTPSServer'
4543
HTTPSServerFeature = _HTTPSServerFeature()
4546
class _UnicodeFilename(Feature):
4547
"""Does the filesystem support Unicode filenames?"""
4552
except UnicodeEncodeError:
4554
except (IOError, OSError):
4555
# The filesystem allows the Unicode filename but the file doesn't
4559
# The filesystem allows the Unicode filename and the file exists,
4563
UnicodeFilename = _UnicodeFilename()
4566
class _ByteStringNamedFilesystem(Feature):
4567
"""Is the filesystem based on bytes?"""
4570
if os.name == "posix":
4574
ByteStringNamedFilesystem = _ByteStringNamedFilesystem()
4577
class _UTF8Filesystem(Feature):
4578
"""Is the filesystem UTF-8?"""
4581
if osutils._fs_enc.upper() in ('UTF-8', 'UTF8'):
4585
UTF8Filesystem = _UTF8Filesystem()
4588
class _BreakinFeature(Feature):
4589
"""Does this platform support the breakin feature?"""
4592
from bzrlib import breakin
4593
if breakin.determine_signal() is None:
4595
if sys.platform == 'win32':
4596
# Windows doesn't have os.kill, and we catch the SIGBREAK signal.
4597
# We trigger SIGBREAK via a Console api so we need ctypes to
4598
# access the function
4605
def feature_name(self):
4606
return "SIGQUIT or SIGBREAK w/ctypes on win32"
4609
BreakinFeature = _BreakinFeature()
4612
class _CaseInsCasePresFilenameFeature(Feature):
4613
"""Is the file-system case insensitive, but case-preserving?"""
4616
fileno, name = tempfile.mkstemp(prefix='MixedCase')
4618
# first check truly case-preserving for created files, then check
4619
# case insensitive when opening existing files.
4620
name = osutils.normpath(name)
4621
base, rel = osutils.split(name)
4622
found_rel = osutils.canonical_relpath(base, name)
4623
return (found_rel == rel
4624
and os.path.isfile(name.upper())
4625
and os.path.isfile(name.lower()))
4630
def feature_name(self):
4631
return "case-insensitive case-preserving filesystem"
4633
CaseInsCasePresFilenameFeature = _CaseInsCasePresFilenameFeature()
4636
class _CaseInsensitiveFilesystemFeature(Feature):
4637
"""Check if underlying filesystem is case-insensitive but *not* case
4640
# Note that on Windows, Cygwin, MacOS etc, the file-systems are far
4641
# more likely to be case preserving, so this case is rare.
4644
if CaseInsCasePresFilenameFeature.available():
4647
if TestCaseWithMemoryTransport.TEST_ROOT is None:
4648
root = osutils.mkdtemp(prefix='testbzr-', suffix='.tmp')
4649
TestCaseWithMemoryTransport.TEST_ROOT = root
4651
root = TestCaseWithMemoryTransport.TEST_ROOT
4652
tdir = osutils.mkdtemp(prefix='case-sensitive-probe-', suffix='',
4654
name_a = osutils.pathjoin(tdir, 'a')
4655
name_A = osutils.pathjoin(tdir, 'A')
4657
result = osutils.isdir(name_A)
4658
_rmtree_temp_dir(tdir)
4661
def feature_name(self):
4662
return 'case-insensitive filesystem'
4664
CaseInsensitiveFilesystemFeature = _CaseInsensitiveFilesystemFeature()
4667
class _CaseSensitiveFilesystemFeature(Feature):
4670
if CaseInsCasePresFilenameFeature.available():
4672
elif CaseInsensitiveFilesystemFeature.available():
4677
def feature_name(self):
4678
return 'case-sensitive filesystem'
4680
# new coding style is for feature instances to be lowercase
4681
case_sensitive_filesystem_feature = _CaseSensitiveFilesystemFeature()
4684
# Only define SubUnitBzrRunner if subunit is available.
4686
from subunit import TestProtocolClient
4687
from subunit.test_results import AutoTimingTestResultDecorator
4688
class SubUnitBzrProtocolClient(TestProtocolClient):
4690
def addSuccess(self, test, details=None):
4691
# The subunit client always includes the details in the subunit
4692
# stream, but we don't want to include it in ours.
4693
if details is not None and 'log' in details:
4695
return super(SubUnitBzrProtocolClient, self).addSuccess(
4698
class SubUnitBzrRunner(TextTestRunner):
4699
def run(self, test):
4700
result = AutoTimingTestResultDecorator(
4701
SubUnitBzrProtocolClient(self.stream))
4707
class _PosixPermissionsFeature(Feature):
4711
# create temporary file and check if specified perms are maintained.
4714
write_perms = stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR
4715
f = tempfile.mkstemp(prefix='bzr_perms_chk_')
4718
os.chmod(name, write_perms)
4720
read_perms = os.stat(name).st_mode & 0777
4722
return (write_perms == read_perms)
4724
return (os.name == 'posix') and has_perms()
4726
def feature_name(self):
4727
return 'POSIX permissions support'
4729
posix_permissions_feature = _PosixPermissionsFeature()
289
print >>f, "contents of", name
294
class MetaTestLog(TestCase):
295
def test_logging(self):
296
"""Test logs are captured when a test fails."""
297
logging.info('an info message')
298
warning('something looks dodgy...')
299
logging.debug('hello, test is running')
303
def selftest(verbose=False, pattern=".*"):
304
return testsweet.run_suite(test_suite(), 'testbzr', verbose=verbose, pattern=pattern)
308
from bzrlib.selftest.TestUtil import TestLoader, TestSuite
309
import bzrlib, bzrlib.store, bzrlib.inventory, bzrlib.branch
310
import bzrlib.osutils, bzrlib.commands, bzrlib.merge3, bzrlib.plugin
311
from doctest import DocTestSuite
313
global MODULES_TO_TEST, MODULES_TO_DOCTEST
316
['bzrlib.selftest.MetaTestLog',
317
'bzrlib.selftest.test_parent',
318
'bzrlib.selftest.testinv',
319
'bzrlib.selftest.testfetch',
320
'bzrlib.selftest.versioning',
321
'bzrlib.selftest.whitebox',
322
'bzrlib.selftest.testmerge3',
323
'bzrlib.selftest.testmerge',
324
'bzrlib.selftest.testhashcache',
325
'bzrlib.selftest.teststatus',
326
'bzrlib.selftest.testlog',
327
'bzrlib.selftest.blackbox',
328
'bzrlib.selftest.testrevisionnamespaces',
329
'bzrlib.selftest.testbranch',
330
'bzrlib.selftest.testremotebranch',
331
'bzrlib.selftest.testrevision',
332
'bzrlib.selftest.test_merge_core',
333
'bzrlib.selftest.test_smart_add',
334
'bzrlib.selftest.testdiff',
335
'bzrlib.selftest.test_xml',
337
'bzrlib.selftest.teststore',
338
'bzrlib.selftest.testtransport',
339
'bzrlib.selftest.testgraph',
342
for m in (bzrlib.store, bzrlib.inventory, bzrlib.branch,
343
bzrlib.osutils, bzrlib.commands, bzrlib.merge3):
344
if m not in MODULES_TO_DOCTEST:
345
MODULES_TO_DOCTEST.append(m)
347
TestCase.BZRPATH = os.path.join(os.path.realpath(os.path.dirname(bzrlib.__path__[0])), 'bzr')
348
print '%-30s %s' % ('bzr binary', TestCase.BZRPATH)
351
suite.addTest(TestLoader().loadTestsFromNames(testmod_names))
352
for m in MODULES_TO_TEST:
353
suite.addTest(TestLoader().loadTestsFromModule(m))
354
for m in (MODULES_TO_DOCTEST):
355
suite.addTest(DocTestSuite(m))
356
for p in bzrlib.plugin.all_plugins:
357
if hasattr(p, 'test_suite'):
358
suite.addTest(p.test_suite())
added
removed
bzrlib/selftest/__init__.py
