1
# Copyright (C) 2005 by Canonical Ltd
1
# Copyright (C) 2005-2013, 2015, 2016 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., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
17
"""Testing framework extensions"""
19
from __future__ import absolute_import
21
# NOTE: Some classes in here use camelCaseNaming() rather than
22
# underscore_naming(). That's for consistency with unittest; it's not the
23
# general style of bzrlib. Please continue that consistency when adding e.g.
24
# new assertFoo() methods.
18
29
from cStringIO import StringIO
30
import bzrlib.commands
33
import bzrlib.osutils as osutils
34
from bzrlib.selftest import TestUtil
35
from bzrlib.selftest.TestUtil import TestLoader, TestSuite
36
from bzrlib.selftest.treeshape import build_tree_contents
39
MODULES_TO_DOCTEST = []
41
from logging import debug, warning, error
45
class EarlyStoppingTestResultAdapter(object):
46
"""An adapter for TestResult to stop at the first first failure or error"""
48
def __init__(self, result):
51
def addError(self, test, err):
52
self._result.addError(test, err)
55
def addFailure(self, test, err):
56
self._result.addFailure(test, err)
59
def __getattr__(self, name):
60
return getattr(self._result, name)
62
def __setattr__(self, name, value):
64
object.__setattr__(self, name, value)
65
return setattr(self._result, name, value)
68
class _MyResult(unittest._TextTestResult):
72
No special behaviour for now.
75
def _elapsedTime(self):
76
return "(Took %.3fs)" % (time.time() - self._start_time)
54
# nb: check this before importing anything else from within it
55
_testtools_version = getattr(testtools, '__version__', ())
56
if _testtools_version < (0, 9, 5):
57
raise ImportError("need at least testtools 0.9.5: %s is %r"
58
% (testtools.__file__, _testtools_version))
59
from testtools import content
66
commands as _mod_commands,
76
plugin as _mod_plugin,
83
transport as _mod_transport,
89
# lsprof not available
91
from bzrlib.smart import client, request
92
from bzrlib.transport import (
96
from bzrlib.symbol_versioning import (
100
from bzrlib.tests import (
106
from bzrlib.ui import NullProgressView
107
from bzrlib.ui.text import TextUIFactory
108
from bzrlib.tests.features import _CompatabilityThunkFeature
110
# Mark this python module as being part of the implementation
111
# of unittest: this gives us better tracebacks where the last
112
# shown frame is the test code, not our assertXYZ.
115
default_transport = test_server.LocalURLServer
118
_unitialized_attr = object()
119
"""A sentinel needed to act as a default value in a method signature."""
122
# Subunit result codes, defined here to prevent a hard dependency on subunit.
126
# These are intentionally brought into this namespace. That way plugins, etc
127
# can just "from bzrlib.tests import TestCase, TestLoader, etc"
128
TestSuite = TestUtil.TestSuite
129
TestLoader = TestUtil.TestLoader
131
# Tests should run in a clean and clearly defined environment. The goal is to
132
# keep them isolated from the running environment as mush as possible. The test
133
# framework ensures the variables defined below are set (or deleted if the
134
# value is None) before a test is run and reset to their original value after
135
# the test is run. Generally if some code depends on an environment variable,
136
# the tests should start without this variable in the environment. There are a
137
# few exceptions but you shouldn't violate this rule lightly.
141
'XDG_CONFIG_HOME': None,
142
# bzr now uses the Win32 API and doesn't rely on APPDATA, but the
143
# tests do check our impls match APPDATA
144
'BZR_EDITOR': None, # test_msgeditor manipulates this variable
148
'BZREMAIL': None, # may still be present in the environment
149
'EMAIL': 'jrandom@example.com', # set EMAIL as bzr does not guess
150
'BZR_PROGRESS_BAR': None,
151
# This should trap leaks to ~/.bzr.log. This occurs when tests use TestCase
152
# as a base class instead of TestCaseInTempDir. Tests inheriting from
153
# TestCase should not use disk resources, BZR_LOG is one.
154
'BZR_LOG': '/you-should-use-TestCaseInTempDir-if-you-need-a-log-file',
155
'BZR_PLUGIN_PATH': None,
156
'BZR_DISABLE_PLUGINS': None,
157
'BZR_PLUGINS_AT': None,
158
'BZR_CONCURRENCY': None,
159
# Make sure that any text ui tests are consistent regardless of
160
# the environment the test case is run in; you may want tests that
161
# test other combinations. 'dumb' is a reasonable guess for tests
162
# going to a pipe or a StringIO.
168
'SSH_AUTH_SOCK': None,
178
# Nobody cares about ftp_proxy, FTP_PROXY AFAIK. So far at
179
# least. If you do (care), please update this comment
183
'BZR_REMOTE_PATH': None,
184
# Generally speaking, we don't want apport reporting on crashes in
185
# the test envirnoment unless we're specifically testing apport,
186
# so that it doesn't leak into the real system environment. We
187
# use an env var so it propagates to subprocesses.
188
'APPORT_DISABLE': '1',
192
def override_os_environ(test, env=None):
193
"""Modify os.environ keeping a copy.
195
:param test: A test instance
197
:param env: A dict containing variable definitions to be installed
200
env = isolated_environ
201
test._original_os_environ = dict([(var, value)
202
for var, value in os.environ.iteritems()])
203
for var, value in env.iteritems():
204
osutils.set_or_unset_env(var, value)
205
if var not in test._original_os_environ:
206
# The var is new, add it with a value of None, so
207
# restore_os_environ will delete it
208
test._original_os_environ[var] = None
211
def restore_os_environ(test):
212
"""Restore os.environ to its original state.
214
:param test: A test instance previously passed to override_os_environ.
216
for var, value in test._original_os_environ.iteritems():
217
# Restore the original value (or delete it if the value has been set to
218
# None in override_os_environ).
219
osutils.set_or_unset_env(var, value)
222
def _clear__type_equality_funcs(test):
223
"""Cleanup bound methods stored on TestCase instances
225
Clear the dict breaking a few (mostly) harmless cycles in the affected
226
unittests released with Python 2.6 and initial Python 2.7 versions.
228
For a few revisions between Python 2.7.1 and Python 2.7.2 that annoyingly
229
shipped in Oneiric, an object with no clear method was used, hence the
230
extra complications, see bug 809048 for details.
232
type_equality_funcs = getattr(test, "_type_equality_funcs", None)
233
if type_equality_funcs is not None:
234
tef_clear = getattr(type_equality_funcs, "clear", None)
235
if tef_clear is None:
236
tef_instance_dict = getattr(type_equality_funcs, "__dict__", None)
237
if tef_instance_dict is not None:
238
tef_clear = tef_instance_dict.clear
239
if tef_clear is not None:
243
class ExtendedTestResult(testtools.TextTestResult):
244
"""Accepts, reports and accumulates the results of running tests.
246
Compared to the unittest version this class adds support for
247
profiling, benchmarking, stopping as soon as a test fails, and
248
skipping tests. There are further-specialized subclasses for
249
different types of display.
251
When a test finishes, in whatever way, it calls one of the addSuccess,
252
addFailure or addError methods. These in turn may redirect to a more
253
specific case for the special test results supported by our extended
256
Note that just one of these objects is fed the results from many tests.
261
def __init__(self, stream, descriptions, verbosity,
265
"""Construct new TestResult.
267
:param bench_history: Optionally, a writable file object to accumulate
270
testtools.TextTestResult.__init__(self, stream)
271
if bench_history is not None:
272
from bzrlib.version import _get_bzr_source_tree
273
src_tree = _get_bzr_source_tree()
276
revision_id = src_tree.get_parent_ids()[0]
278
# XXX: if this is a brand new tree, do the same as if there
282
# XXX: If there's no branch, what should we do?
284
bench_history.write("--date %s %s\n" % (time.time(), revision_id))
285
self._bench_history = bench_history
286
self.ui = ui.ui_factory
289
self.failure_count = 0
290
self.known_failure_count = 0
292
self.not_applicable_count = 0
293
self.unsupported = {}
295
self._overall_start_time = time.time()
296
self._strict = strict
297
self._first_thread_leaker_id = None
298
self._tests_leaking_threads_count = 0
299
self._traceback_from_test = None
301
def stopTestRun(self):
304
stopTime = time.time()
305
timeTaken = stopTime - self.startTime
306
# GZ 2010-07-19: Seems testtools has no printErrors method, and though
307
# the parent class method is similar have to duplicate
308
self._show_list('ERROR', self.errors)
309
self._show_list('FAIL', self.failures)
310
self.stream.write(self.sep2)
311
self.stream.write("%s %d test%s in %.3fs\n\n" % (actionTaken,
312
run, run != 1 and "s" or "", timeTaken))
313
if not self.wasSuccessful():
314
self.stream.write("FAILED (")
315
failed, errored = map(len, (self.failures, self.errors))
317
self.stream.write("failures=%d" % failed)
319
if failed: self.stream.write(", ")
320
self.stream.write("errors=%d" % errored)
321
if self.known_failure_count:
322
if failed or errored: self.stream.write(", ")
323
self.stream.write("known_failure_count=%d" %
324
self.known_failure_count)
325
self.stream.write(")\n")
327
if self.known_failure_count:
328
self.stream.write("OK (known_failures=%d)\n" %
329
self.known_failure_count)
331
self.stream.write("OK\n")
332
if self.skip_count > 0:
333
skipped = self.skip_count
334
self.stream.write('%d test%s skipped\n' %
335
(skipped, skipped != 1 and "s" or ""))
337
for feature, count in sorted(self.unsupported.items()):
338
self.stream.write("Missing feature '%s' skipped %d tests.\n" %
341
ok = self.wasStrictlySuccessful()
343
ok = self.wasSuccessful()
344
if self._first_thread_leaker_id:
346
'%s is leaking threads among %d leaking tests.\n' % (
347
self._first_thread_leaker_id,
348
self._tests_leaking_threads_count))
349
# We don't report the main thread as an active one.
351
'%d non-main threads were left active in the end.\n'
352
% (len(self._active_threads) - 1))
354
def getDescription(self, test):
357
def _extractBenchmarkTime(self, testCase, details=None):
358
"""Add a benchmark time for the current test case."""
359
if details and 'benchtime' in details:
360
return float(''.join(details['benchtime'].iter_bytes()))
361
return getattr(testCase, "_benchtime", None)
363
def _delta_to_float(self, a_timedelta, precision):
364
# This calls ceiling to ensure that the most pessimistic view of time
365
# taken is shown (rather than leaving it to the Python %f operator
366
# to decide whether to round/floor/ceiling. This was added when we
367
# had pyp3 test failures that suggest a floor was happening.
368
shift = 10 ** precision
369
return math.ceil((a_timedelta.days * 86400.0 + a_timedelta.seconds +
370
a_timedelta.microseconds / 1000000.0) * shift) / shift
372
def _elapsedTestTimeString(self):
373
"""Return a time string for the overall time the current test has taken."""
374
return self._formatTime(self._delta_to_float(
375
self._now() - self._start_datetime, 3))
377
def _testTimeString(self, testCase):
378
benchmark_time = self._extractBenchmarkTime(testCase)
379
if benchmark_time is not None:
380
return self._formatTime(benchmark_time) + "*"
382
return self._elapsedTestTimeString()
384
def _formatTime(self, seconds):
385
"""Format seconds as milliseconds with leading spaces."""
386
# some benchmarks can take thousands of seconds to run, so we need 8
388
return "%8dms" % (1000 * seconds)
390
def _shortened_test_description(self, test):
392
what = re.sub(r'^bzrlib\.tests\.', '', what)
395
# GZ 2010-10-04: Cloned tests may end up harmlessly calling this method
396
# multiple times in a row, because the handler is added for
397
# each test but the container list is shared between cases.
398
# See lp:498869 lp:625574 and lp:637725 for background.
399
def _record_traceback_from_test(self, exc_info):
400
"""Store the traceback from passed exc_info tuple till"""
401
self._traceback_from_test = exc_info[2]
78
403
def startTest(self, test):
79
unittest.TestResult.startTest(self, test)
80
# TODO: Maybe show test.shortDescription somewhere?
81
what = test.shortDescription() or test.id()
83
self.stream.write('%-70.70s' % what)
85
self._start_time = time.time()
404
super(ExtendedTestResult, self).startTest(test)
408
self.report_test_start(test)
409
test.number = self.count
410
self._recordTestStartTime()
411
# Make testtools cases give us the real traceback on failure
412
addOnException = getattr(test, "addOnException", None)
413
if addOnException is not None:
414
addOnException(self._record_traceback_from_test)
415
# Only check for thread leaks on bzrlib derived test cases
416
if isinstance(test, TestCase):
417
test.addCleanup(self._check_leaked_threads, test)
419
def stopTest(self, test):
420
super(ExtendedTestResult, self).stopTest(test)
421
# Manually break cycles, means touching various private things but hey
422
getDetails = getattr(test, "getDetails", None)
423
if getDetails is not None:
425
_clear__type_equality_funcs(test)
426
self._traceback_from_test = None
428
def startTests(self):
429
self.report_tests_starting()
430
self._active_threads = threading.enumerate()
432
def _check_leaked_threads(self, test):
433
"""See if any threads have leaked since last call
435
A sample of live threads is stored in the _active_threads attribute,
436
when this method runs it compares the current live threads and any not
437
in the previous sample are treated as having leaked.
439
now_active_threads = set(threading.enumerate())
440
threads_leaked = now_active_threads.difference(self._active_threads)
442
self._report_thread_leak(test, threads_leaked, now_active_threads)
443
self._tests_leaking_threads_count += 1
444
if self._first_thread_leaker_id is None:
445
self._first_thread_leaker_id = test.id()
446
self._active_threads = now_active_threads
448
def _recordTestStartTime(self):
449
"""Record that a test has started."""
450
self._start_datetime = self._now()
87
452
def addError(self, test, err):
88
unittest.TestResult.addError(self, test, err)
90
self.stream.writeln("ERROR %s" % self._elapsedTime())
92
self.stream.write('E')
453
"""Tell result that test finished with an error.
455
Called from the TestCase run() method when the test
456
fails with an unexpected error.
458
self._post_mortem(self._traceback_from_test)
459
super(ExtendedTestResult, self).addError(test, err)
460
self.error_count += 1
461
self.report_error(test, err)
95
465
def addFailure(self, test, err):
96
unittest.TestResult.addFailure(self, test, err)
98
self.stream.writeln("FAIL %s" % self._elapsedTime())
100
self.stream.write('F')
103
def addSuccess(self, test):
105
self.stream.writeln('OK %s' % self._elapsedTime())
107
self.stream.write('~')
109
unittest.TestResult.addSuccess(self, test)
111
def printErrorList(self, flavour, errors):
112
for test, err in errors:
113
self.stream.writeln(self.separator1)
114
self.stream.writeln("%s: %s" % (flavour,self.getDescription(test)))
115
if hasattr(test, '_get_log'):
116
self.stream.writeln()
117
self.stream.writeln('log from this test:')
118
print >>self.stream, test._get_log()
119
self.stream.writeln(self.separator2)
120
self.stream.writeln("%s" % err)
123
class TextTestRunner(unittest.TextTestRunner):
466
"""Tell result that test failed.
468
Called from the TestCase run() method when the test
469
fails because e.g. an assert() method failed.
471
self._post_mortem(self._traceback_from_test)
472
super(ExtendedTestResult, self).addFailure(test, err)
473
self.failure_count += 1
474
self.report_failure(test, err)
478
def addSuccess(self, test, details=None):
479
"""Tell result that test completed successfully.
481
Called from the TestCase run()
483
if self._bench_history is not None:
484
benchmark_time = self._extractBenchmarkTime(test, details)
485
if benchmark_time is not None:
486
self._bench_history.write("%s %s\n" % (
487
self._formatTime(benchmark_time),
489
self.report_success(test)
490
super(ExtendedTestResult, self).addSuccess(test)
491
test._log_contents = ''
493
def addExpectedFailure(self, test, err):
494
self.known_failure_count += 1
495
self.report_known_failure(test, err)
497
def addUnexpectedSuccess(self, test, details=None):
498
"""Tell result the test unexpectedly passed, counting as a failure
500
When the minimum version of testtools required becomes 0.9.8 this
501
can be updated to use the new handling there.
503
super(ExtendedTestResult, self).addFailure(test, details=details)
504
self.failure_count += 1
505
self.report_unexpected_success(test,
506
"".join(details["reason"].iter_text()))
510
def addNotSupported(self, test, feature):
511
"""The test will not be run because of a missing feature.
513
# this can be called in two different ways: it may be that the
514
# test started running, and then raised (through requireFeature)
515
# UnavailableFeature. Alternatively this method can be called
516
# while probing for features before running the test code proper; in
517
# that case we will see startTest and stopTest, but the test will
518
# never actually run.
519
self.unsupported.setdefault(str(feature), 0)
520
self.unsupported[str(feature)] += 1
521
self.report_unsupported(test, feature)
523
def addSkip(self, test, reason):
524
"""A test has not run for 'reason'."""
526
self.report_skip(test, reason)
528
def addNotApplicable(self, test, reason):
529
self.not_applicable_count += 1
530
self.report_not_applicable(test, reason)
532
def _count_stored_tests(self):
533
"""Count of tests instances kept alive due to not succeeding"""
534
return self.error_count + self.failure_count + self.known_failure_count
536
def _post_mortem(self, tb=None):
537
"""Start a PDB post mortem session."""
538
if os.environ.get('BZR_TEST_PDB', None):
542
def progress(self, offset, whence):
543
"""The test is adjusting the count of tests to run."""
544
if whence == SUBUNIT_SEEK_SET:
545
self.num_tests = offset
546
elif whence == SUBUNIT_SEEK_CUR:
547
self.num_tests += offset
549
raise errors.BzrError("Unknown whence %r" % whence)
551
def report_tests_starting(self):
552
"""Display information before the test run begins"""
553
if getattr(sys, 'frozen', None) is None:
554
bzr_path = osutils.realpath(sys.argv[0])
556
bzr_path = sys.executable
558
'bzr selftest: %s\n' % (bzr_path,))
561
bzrlib.__path__[0],))
563
' bzr-%s python-%s %s\n' % (
564
bzrlib.version_string,
565
bzrlib._format_version_tuple(sys.version_info),
566
platform.platform(aliased=1),
568
self.stream.write('\n')
570
def report_test_start(self, test):
571
"""Display information on the test just about to be run"""
573
def _report_thread_leak(self, test, leaked_threads, active_threads):
574
"""Display information on a test that leaked one or more threads"""
575
# GZ 2010-09-09: A leak summary reported separately from the general
576
# thread debugging would be nice. Tests under subunit
577
# need something not using stream, perhaps adding a
578
# testtools details object would be fitting.
579
if 'threads' in selftest_debug_flags:
580
self.stream.write('%s is leaking, active is now %d\n' %
581
(test.id(), len(active_threads)))
583
def startTestRun(self):
584
self.startTime = time.time()
586
def report_success(self, test):
589
def wasStrictlySuccessful(self):
590
if self.unsupported or self.known_failure_count:
592
return self.wasSuccessful()
595
class TextTestResult(ExtendedTestResult):
596
"""Displays progress and results of tests in text form"""
598
def __init__(self, stream, descriptions, verbosity,
603
ExtendedTestResult.__init__(self, stream, descriptions, verbosity,
604
bench_history, strict)
605
# We no longer pass them around, but just rely on the UIFactory stack
608
warnings.warn("Passing pb to TextTestResult is deprecated")
609
self.pb = self.ui.nested_progress_bar()
610
self.pb.show_pct = False
611
self.pb.show_spinner = False
612
self.pb.show_eta = False,
613
self.pb.show_count = False
614
self.pb.show_bar = False
615
self.pb.update_latency = 0
616
self.pb.show_transport_activity = False
618
def stopTestRun(self):
619
# called when the tests that are going to run have run
622
super(TextTestResult, self).stopTestRun()
624
def report_tests_starting(self):
625
super(TextTestResult, self).report_tests_starting()
626
self.pb.update('[test 0/%d] Starting' % (self.num_tests))
628
def _progress_prefix_text(self):
629
# the longer this text, the less space we have to show the test
631
a = '[%d' % self.count # total that have been run
632
# tests skipped as known not to be relevant are not important enough
634
## if self.skip_count:
635
## a += ', %d skip' % self.skip_count
636
## if self.known_failure_count:
637
## a += '+%dX' % self.known_failure_count
639
a +='/%d' % self.num_tests
641
runtime = time.time() - self._overall_start_time
643
a += '%dm%ds' % (runtime / 60, runtime % 60)
646
total_fail_count = self.error_count + self.failure_count
648
a += ', %d failed' % total_fail_count
649
# if self.unsupported:
650
# a += ', %d missing' % len(self.unsupported)
654
def report_test_start(self, test):
656
self._progress_prefix_text()
658
+ self._shortened_test_description(test))
660
def _test_description(self, test):
661
return self._shortened_test_description(test)
663
def report_error(self, test, err):
664
self.stream.write('ERROR: %s\n %s\n' % (
665
self._test_description(test),
669
def report_failure(self, test, err):
670
self.stream.write('FAIL: %s\n %s\n' % (
671
self._test_description(test),
675
def report_known_failure(self, test, err):
678
def report_unexpected_success(self, test, reason):
679
self.stream.write('FAIL: %s\n %s: %s\n' % (
680
self._test_description(test),
681
"Unexpected success. Should have failed",
685
def report_skip(self, test, reason):
688
def report_not_applicable(self, test, reason):
691
def report_unsupported(self, test, feature):
692
"""test cannot be run because feature is missing."""
695
class VerboseTestResult(ExtendedTestResult):
696
"""Produce long output, with one line per test run plus times"""
698
def _ellipsize_to_right(self, a_string, final_width):
699
"""Truncate and pad a string, keeping the right hand side"""
700
if len(a_string) > final_width:
701
result = '...' + a_string[3-final_width:]
704
return result.ljust(final_width)
706
def report_tests_starting(self):
707
self.stream.write('running %d tests...\n' % self.num_tests)
708
super(VerboseTestResult, self).report_tests_starting()
710
def report_test_start(self, test):
711
name = self._shortened_test_description(test)
712
width = osutils.terminal_width()
713
if width is not None:
714
# width needs space for 6 char status, plus 1 for slash, plus an
715
# 11-char time string, plus a trailing blank
716
# when NUMBERED_DIRS: plus 5 chars on test number, plus 1 char on
718
self.stream.write(self._ellipsize_to_right(name, width-18))
720
self.stream.write(name)
723
def _error_summary(self, err):
725
return '%s%s' % (indent, err[1])
727
def report_error(self, test, err):
728
self.stream.write('ERROR %s\n%s\n'
729
% (self._testTimeString(test),
730
self._error_summary(err)))
732
def report_failure(self, test, err):
733
self.stream.write(' FAIL %s\n%s\n'
734
% (self._testTimeString(test),
735
self._error_summary(err)))
737
def report_known_failure(self, test, err):
738
self.stream.write('XFAIL %s\n%s\n'
739
% (self._testTimeString(test),
740
self._error_summary(err)))
742
def report_unexpected_success(self, test, reason):
743
self.stream.write(' FAIL %s\n%s: %s\n'
744
% (self._testTimeString(test),
745
"Unexpected success. Should have failed",
748
def report_success(self, test):
749
self.stream.write(' OK %s\n' % self._testTimeString(test))
750
for bench_called, stats in getattr(test, '_benchcalls', []):
751
self.stream.write('LSProf output for %s(%s, %s)\n' % bench_called)
752
stats.pprint(file=self.stream)
753
# flush the stream so that we get smooth output. This verbose mode is
754
# used to show the output in PQM.
757
def report_skip(self, test, reason):
758
self.stream.write(' SKIP %s\n%s\n'
759
% (self._testTimeString(test), reason))
761
def report_not_applicable(self, test, reason):
762
self.stream.write(' N/A %s\n %s\n'
763
% (self._testTimeString(test), reason))
765
def report_unsupported(self, test, feature):
766
"""test cannot be run because feature is missing."""
767
self.stream.write("NODEP %s\n The feature '%s' is not available.\n"
768
%(self._testTimeString(test), feature))
771
class TextTestRunner(object):
124
772
stop_on_failure = False
126
def _makeResult(self):
127
result = _MyResult(self.stream, self.descriptions, self.verbosity)
128
if self.stop_on_failure:
129
result = EarlyStoppingTestResultAdapter(result)
780
result_decorators=None,
782
"""Create a TextTestRunner.
784
:param result_decorators: An optional list of decorators to apply
785
to the result object being used by the runner. Decorators are
786
applied left to right - the first element in the list is the
789
# stream may know claim to know to write unicode strings, but in older
790
# pythons this goes sufficiently wrong that it is a bad idea. (
791
# specifically a built in file with encoding 'UTF-8' will still try
792
# to encode using ascii.
793
new_encoding = osutils.get_terminal_encoding()
794
codec = codecs.lookup(new_encoding)
795
if type(codec) is tuple:
799
encode = codec.encode
800
# GZ 2010-09-08: Really we don't want to be writing arbitrary bytes,
801
# so should swap to the plain codecs.StreamWriter
802
stream = osutils.UnicodeOrBytesToBytesWriter(encode, stream,
804
stream.encoding = new_encoding
806
self.descriptions = descriptions
807
self.verbosity = verbosity
808
self._bench_history = bench_history
809
self._strict = strict
810
self._result_decorators = result_decorators or []
813
"Run the given test case or test suite."
814
if self.verbosity == 1:
815
result_class = TextTestResult
816
elif self.verbosity >= 2:
817
result_class = VerboseTestResult
818
original_result = result_class(self.stream,
821
bench_history=self._bench_history,
824
# Signal to result objects that look at stop early policy to stop,
825
original_result.stop_early = self.stop_on_failure
826
result = original_result
827
for decorator in self._result_decorators:
828
result = decorator(result)
829
result.stop_early = self.stop_on_failure
830
result.startTestRun()
835
# higher level code uses our extended protocol to determine
836
# what exit code to give.
837
return original_result
133
840
def iter_suite_tests(suite):
134
841
"""Return all tests in a suite, recursing through nested suites"""
135
for item in suite._tests:
136
if isinstance(item, unittest.TestCase):
138
elif isinstance(item, unittest.TestSuite):
842
if isinstance(suite, unittest.TestCase):
844
elif isinstance(suite, unittest.TestSuite):
139
846
for r in iter_suite_tests(item):
142
raise Exception('unknown object %r inside test suite %r'
146
class TestSkipped(Exception):
147
"""Indicates that a test was intentionally skipped, rather than failing."""
151
class CommandFailed(Exception):
154
class TestCase(unittest.TestCase):
849
raise Exception('unknown type %r for object %r'
850
% (type(suite), suite))
853
TestSkipped = testtools.testcase.TestSkipped
856
class TestNotApplicable(TestSkipped):
857
"""A test is not applicable to the situation where it was run.
859
This is only normally raised by parameterized tests, if they find that
860
the instance they're constructed upon does not support one aspect
865
# traceback._some_str fails to format exceptions that have the default
866
# __str__ which does an implicit ascii conversion. However, repr() on those
867
# objects works, for all that its not quite what the doctor may have ordered.
868
def _clever_some_str(value):
873
return repr(value).replace('\\n', '\n')
875
return '<unprintable %s object>' % type(value).__name__
877
traceback._some_str = _clever_some_str
880
# deprecated - use self.knownFailure(), or self.expectFailure.
881
KnownFailure = testtools.testcase._ExpectedFailure
884
class UnavailableFeature(Exception):
885
"""A feature required for this test was not available.
887
This can be considered a specialised form of SkippedTest.
889
The feature should be used to construct the exception.
893
class StringIOWrapper(object):
894
"""A wrapper around cStringIO which just adds an encoding attribute.
896
Internally we can check sys.stdout to see what the output encoding
897
should be. However, cStringIO has no encoding attribute that we can
898
set. So we wrap it instead.
903
def __init__(self, s=None):
905
self.__dict__['_cstring'] = StringIO(s)
907
self.__dict__['_cstring'] = StringIO()
909
def __getattr__(self, name, getattr=getattr):
910
return getattr(self.__dict__['_cstring'], name)
912
def __setattr__(self, name, val):
913
if name == 'encoding':
914
self.__dict__['encoding'] = val
916
return setattr(self._cstring, name, val)
919
class TestUIFactory(TextUIFactory):
920
"""A UI Factory for testing.
922
Hide the progress bar but emit note()s.
924
Allows get_password to be tested without real tty attached.
926
See also CannedInputUIFactory which lets you provide programmatic input in
929
# TODO: Capture progress events at the model level and allow them to be
930
# observed by tests that care.
932
# XXX: Should probably unify more with CannedInputUIFactory or a
933
# particular configuration of TextUIFactory, or otherwise have a clearer
934
# idea of how they're supposed to be different.
935
# See https://bugs.launchpad.net/bzr/+bug/408213
937
def __init__(self, stdout=None, stderr=None, stdin=None):
938
if stdin is not None:
939
# We use a StringIOWrapper to be able to test various
940
# encodings, but the user is still responsible to
941
# encode the string and to set the encoding attribute
942
# of StringIOWrapper.
943
stdin = StringIOWrapper(stdin)
944
super(TestUIFactory, self).__init__(stdin, stdout, stderr)
946
def get_non_echoed_password(self):
947
"""Get password from stdin without trying to handle the echo mode"""
948
password = self.stdin.readline()
951
if password[-1] == '\n':
952
password = password[:-1]
955
def make_progress_view(self):
956
return NullProgressView()
959
def isolated_doctest_setUp(test):
960
override_os_environ(test)
963
def isolated_doctest_tearDown(test):
964
restore_os_environ(test)
967
def IsolatedDocTestSuite(*args, **kwargs):
968
"""Overrides doctest.DocTestSuite to handle isolation.
970
The method is really a factory and users are expected to use it as such.
973
kwargs['setUp'] = isolated_doctest_setUp
974
kwargs['tearDown'] = isolated_doctest_tearDown
975
return doctest.DocTestSuite(*args, **kwargs)
978
class TestCase(testtools.TestCase):
155
979
"""Base class for bzr unit tests.
157
Tests that need access to disk resources should subclass
981
Tests that need access to disk resources should subclass
158
982
TestCaseInTempDir not TestCase.
160
984
Error and debug log messages are redirected from their usual
161
985
location into a temporary file, the contents of which can be
162
retrieved by _get_log().
986
retrieved by _get_log(). We use a real OS file, not an in-memory object,
987
so that it can also capture file IO. When the test completes this file
988
is read into memory and removed from disk.
164
990
There are also convenience functions to invoke bzr's command-line
165
routine, and to build and check bzr trees."""
168
_log_file_name = None
991
routine, and to build and check bzr trees.
993
In addition to the usual method of overriding tearDown(), this class also
994
allows subclasses to register cleanup functions via addCleanup, which are
995
run in order as the object is torn down. It's less likely this will be
996
accidentally overlooked.
1000
# record lsprof data when performing benchmark calls.
1001
_gather_lsprof_in_benchmarks = False
1003
def __init__(self, methodName='testMethod'):
1004
super(TestCase, self).__init__(methodName)
1005
self._directory_isolation = True
1006
self.exception_handlers.insert(0,
1007
(UnavailableFeature, self._do_unsupported_or_skip))
1008
self.exception_handlers.insert(0,
1009
(TestNotApplicable, self._do_not_applicable))
170
1011
def setUp(self):
171
unittest.TestCase.setUp(self)
172
self.oldenv = os.environ.get('HOME', None)
173
os.environ['HOME'] = os.getcwd()
174
self.bzr_email = os.environ.get('BZREMAIL')
175
if self.bzr_email is not None:
176
del os.environ['BZREMAIL']
177
self.email = os.environ.get('EMAIL')
178
if self.email is not None:
179
del os.environ['EMAIL']
180
bzrlib.trace.disable_default_logging()
181
self._enable_file_logging()
1012
super(TestCase, self).setUp()
1014
# At this point we're still accessing the config files in $BZR_HOME (as
1015
# set by the user running selftest).
1016
timeout = config.GlobalStack().get('selftest.timeout')
1018
timeout_fixture = fixtures.TimeoutFixture(timeout)
1019
timeout_fixture.setUp()
1020
self.addCleanup(timeout_fixture.cleanUp)
1022
for feature in getattr(self, '_test_needs_features', []):
1023
self.requireFeature(feature)
1024
self._cleanEnvironment()
1026
if bzrlib.global_state is not None:
1027
self.overrideAttr(bzrlib.global_state, 'cmdline_overrides',
1028
config.CommandLineStore())
1031
self._startLogFile()
1032
self._benchcalls = []
1033
self._benchtime = None
1035
self._track_transports()
1037
self._clear_debug_flags()
1038
# Isolate global verbosity level, to make sure it's reproducible
1039
# between tests. We should get rid of this altogether: bug 656694. --
1041
self.overrideAttr(bzrlib.trace, '_verbosity_level', 0)
1042
self._log_files = set()
1043
# Each key in the ``_counters`` dict holds a value for a different
1044
# counter. When the test ends, addDetail() should be used to output the
1045
# counter values. This happens in install_counter_hook().
1047
if 'config_stats' in selftest_debug_flags:
1048
self._install_config_stats_hooks()
1049
# Do not use i18n for tests (unless the test reverses this)
1055
# The sys preserved stdin/stdout should allow blackbox tests debugging
1056
pdb.Pdb(stdin=sys.__stdin__, stdout=sys.__stdout__
1057
).set_trace(sys._getframe().f_back)
1059
def discardDetail(self, name):
1060
"""Extend the addDetail, getDetails api so we can remove a detail.
1062
eg. bzr always adds the 'log' detail at startup, but we don't want to
1063
include it for skipped, xfail, etc tests.
1065
It is safe to call this for a detail that doesn't exist, in case this
1066
gets called multiple times.
1068
# We cheat. details is stored in __details which means we shouldn't
1069
# touch it. but getDetails() returns the dict directly, so we can
1071
details = self.getDetails()
1075
def install_counter_hook(self, hooks, name, counter_name=None):
1076
"""Install a counting hook.
1078
Any hook can be counted as long as it doesn't need to return a value.
1080
:param hooks: Where the hook should be installed.
1082
:param name: The hook name that will be counted.
1084
:param counter_name: The counter identifier in ``_counters``, defaults
1087
_counters = self._counters # Avoid closing over self
1088
if counter_name is None:
1090
if _counters.has_key(counter_name):
1091
raise AssertionError('%s is already used as a counter name'
1093
_counters[counter_name] = 0
1094
self.addDetail(counter_name, content.Content(content.UTF8_TEXT,
1095
lambda: ['%d' % (_counters[counter_name],)]))
1096
def increment_counter(*args, **kwargs):
1097
_counters[counter_name] += 1
1098
label = 'count %s calls' % (counter_name,)
1099
hooks.install_named_hook(name, increment_counter, label)
1100
self.addCleanup(hooks.uninstall_named_hook, name, label)
1102
def _install_config_stats_hooks(self):
1103
"""Install config hooks to count hook calls.
1106
for hook_name in ('get', 'set', 'remove', 'load', 'save'):
1107
self.install_counter_hook(config.ConfigHooks, hook_name,
1108
'config.%s' % (hook_name,))
1110
# The OldConfigHooks are private and need special handling to protect
1111
# against recursive tests (tests that run other tests), so we just do
1112
# manually what registering them into _builtin_known_hooks will provide
1114
self.overrideAttr(config, 'OldConfigHooks', config._OldConfigHooks())
1115
for hook_name in ('get', 'set', 'remove', 'load', 'save'):
1116
self.install_counter_hook(config.OldConfigHooks, hook_name,
1117
'old_config.%s' % (hook_name,))
1119
def _clear_debug_flags(self):
1120
"""Prevent externally set debug flags affecting tests.
1122
Tests that want to use debug flags can just set them in the
1123
debug_flags set during setup/teardown.
1125
# Start with a copy of the current debug flags we can safely modify.
1126
self.overrideAttr(debug, 'debug_flags', set(debug.debug_flags))
1127
if 'allow_debug' not in selftest_debug_flags:
1128
debug.debug_flags.clear()
1129
if 'disable_lock_checks' not in selftest_debug_flags:
1130
debug.debug_flags.add('strict_locks')
1132
def _clear_hooks(self):
1133
# prevent hooks affecting tests
1134
known_hooks = hooks.known_hooks
1135
self._preserved_hooks = {}
1136
for key, (parent, name) in known_hooks.iter_parent_objects():
1137
current_hooks = getattr(parent, name)
1138
self._preserved_hooks[parent] = (name, current_hooks)
1139
self._preserved_lazy_hooks = hooks._lazy_hooks
1140
hooks._lazy_hooks = {}
1141
self.addCleanup(self._restoreHooks)
1142
for key, (parent, name) in known_hooks.iter_parent_objects():
1143
factory = known_hooks.get(key)
1144
setattr(parent, name, factory())
1145
# this hook should always be installed
1146
request._install_hook()
1148
def disable_directory_isolation(self):
1149
"""Turn off directory isolation checks."""
1150
self._directory_isolation = False
1152
def enable_directory_isolation(self):
1153
"""Enable directory isolation checks."""
1154
self._directory_isolation = True
1156
def _silenceUI(self):
1157
"""Turn off UI for duration of test"""
1158
# by default the UI is off; tests can turn it on if they want it.
1159
self.overrideAttr(ui, 'ui_factory', ui.SilentUIFactory())
1161
def _check_locks(self):
1162
"""Check that all lock take/release actions have been paired."""
1163
# We always check for mismatched locks. If a mismatch is found, we
1164
# fail unless -Edisable_lock_checks is supplied to selftest, in which
1165
# case we just print a warning.
1167
acquired_locks = [lock for action, lock in self._lock_actions
1168
if action == 'acquired']
1169
released_locks = [lock for action, lock in self._lock_actions
1170
if action == 'released']
1171
broken_locks = [lock for action, lock in self._lock_actions
1172
if action == 'broken']
1173
# trivially, given the tests for lock acquistion and release, if we
1174
# have as many in each list, it should be ok. Some lock tests also
1175
# break some locks on purpose and should be taken into account by
1176
# considering that breaking a lock is just a dirty way of releasing it.
1177
if len(acquired_locks) != (len(released_locks) + len(broken_locks)):
1179
'Different number of acquired and '
1180
'released or broken locks.\n'
1184
(acquired_locks, released_locks, broken_locks))
1185
if not self._lock_check_thorough:
1186
# Rather than fail, just warn
1187
print "Broken test %s: %s" % (self, message)
1191
def _track_locks(self):
1192
"""Track lock activity during tests."""
1193
self._lock_actions = []
1194
if 'disable_lock_checks' in selftest_debug_flags:
1195
self._lock_check_thorough = False
1197
self._lock_check_thorough = True
1199
self.addCleanup(self._check_locks)
1200
_mod_lock.Lock.hooks.install_named_hook('lock_acquired',
1201
self._lock_acquired, None)
1202
_mod_lock.Lock.hooks.install_named_hook('lock_released',
1203
self._lock_released, None)
1204
_mod_lock.Lock.hooks.install_named_hook('lock_broken',
1205
self._lock_broken, None)
1207
def _lock_acquired(self, result):
1208
self._lock_actions.append(('acquired', result))
1210
def _lock_released(self, result):
1211
self._lock_actions.append(('released', result))
1213
def _lock_broken(self, result):
1214
self._lock_actions.append(('broken', result))
1216
def permit_dir(self, name):
1217
"""Permit a directory to be used by this test. See permit_url."""
1218
name_transport = _mod_transport.get_transport_from_path(name)
1219
self.permit_url(name)
1220
self.permit_url(name_transport.base)
1222
def permit_url(self, url):
1223
"""Declare that url is an ok url to use in this test.
1225
Do this for memory transports, temporary test directory etc.
1227
Do not do this for the current working directory, /tmp, or any other
1228
preexisting non isolated url.
1230
if not url.endswith('/'):
1232
self._bzr_selftest_roots.append(url)
1234
def permit_source_tree_branch_repo(self):
1235
"""Permit the source tree bzr is running from to be opened.
1237
Some code such as bzrlib.version attempts to read from the bzr branch
1238
that bzr is executing from (if any). This method permits that directory
1239
to be used in the test suite.
1241
path = self.get_source_path()
1242
self.record_directory_isolation()
1245
workingtree.WorkingTree.open(path)
1246
except (errors.NotBranchError, errors.NoWorkingTree):
1247
raise TestSkipped('Needs a working tree of bzr sources')
1249
self.enable_directory_isolation()
1251
def _preopen_isolate_transport(self, transport):
1252
"""Check that all transport openings are done in the test work area."""
1253
while isinstance(transport, pathfilter.PathFilteringTransport):
1254
# Unwrap pathfiltered transports
1255
transport = transport.server.backing_transport.clone(
1256
transport._filter('.'))
1257
url = transport.base
1258
# ReadonlySmartTCPServer_for_testing decorates the backing transport
1259
# urls it is given by prepending readonly+. This is appropriate as the
1260
# client shouldn't know that the server is readonly (or not readonly).
1261
# We could register all servers twice, with readonly+ prepending, but
1262
# that makes for a long list; this is about the same but easier to
1264
if url.startswith('readonly+'):
1265
url = url[len('readonly+'):]
1266
self._preopen_isolate_url(url)
1268
def _preopen_isolate_url(self, url):
1269
if not self._directory_isolation:
1271
if self._directory_isolation == 'record':
1272
self._bzr_selftest_roots.append(url)
1274
# This prevents all transports, including e.g. sftp ones backed on disk
1275
# from working unless they are explicitly granted permission. We then
1276
# depend on the code that sets up test transports to check that they are
1277
# appropriately isolated and enable their use by calling
1278
# self.permit_transport()
1279
if not osutils.is_inside_any(self._bzr_selftest_roots, url):
1280
raise errors.BzrError("Attempt to escape test isolation: %r %r"
1281
% (url, self._bzr_selftest_roots))
1283
def record_directory_isolation(self):
1284
"""Gather accessed directories to permit later access.
1286
This is used for tests that access the branch bzr is running from.
1288
self._directory_isolation = "record"
1290
def start_server(self, transport_server, backing_server=None):
1291
"""Start transport_server for this test.
1293
This starts the server, registers a cleanup for it and permits the
1294
server's urls to be used.
1296
if backing_server is None:
1297
transport_server.start_server()
1299
transport_server.start_server(backing_server)
1300
self.addCleanup(transport_server.stop_server)
1301
# Obtain a real transport because if the server supplies a password, it
1302
# will be hidden from the base on the client side.
1303
t = _mod_transport.get_transport_from_url(transport_server.get_url())
1304
# Some transport servers effectively chroot the backing transport;
1305
# others like SFTPServer don't - users of the transport can walk up the
1306
# transport to read the entire backing transport. This wouldn't matter
1307
# except that the workdir tests are given - and that they expect the
1308
# server's url to point at - is one directory under the safety net. So
1309
# Branch operations into the transport will attempt to walk up one
1310
# directory. Chrooting all servers would avoid this but also mean that
1311
# we wouldn't be testing directly against non-root urls. Alternatively
1312
# getting the test framework to start the server with a backing server
1313
# at the actual safety net directory would work too, but this then
1314
# means that the self.get_url/self.get_transport methods would need
1315
# to transform all their results. On balance its cleaner to handle it
1316
# here, and permit a higher url when we have one of these transports.
1317
if t.base.endswith('/work/'):
1318
# we have safety net/test root/work
1319
t = t.clone('../..')
1320
elif isinstance(transport_server,
1321
test_server.SmartTCPServer_for_testing):
1322
# The smart server adds a path similar to work, which is traversed
1323
# up from by the client. But the server is chrooted - the actual
1324
# backing transport is not escaped from, and VFS requests to the
1325
# root will error (because they try to escape the chroot).
1327
while t2.base != t.base:
1330
self.permit_url(t.base)
1332
def _track_transports(self):
1333
"""Install checks for transport usage."""
1334
# TestCase has no safe place it can write to.
1335
self._bzr_selftest_roots = []
1336
# Currently the easiest way to be sure that nothing is going on is to
1337
# hook into bzr dir opening. This leaves a small window of error for
1338
# transport tests, but they are well known, and we can improve on this
1340
controldir.ControlDir.hooks.install_named_hook("pre_open",
1341
self._preopen_isolate_transport, "Check bzr directories are safe.")
183
1343
def _ndiff_strings(self, a, b):
184
1344
"""Return ndiff between two strings containing lines.
186
1346
A trailing newline is added if missing to make the strings
187
1347
print properly."""
188
1348
if b and b[-1] != '\n':
195
1355
charjunk=lambda x: False)
196
1356
return ''.join(difflines)
198
def assertEqualDiff(self, a, b):
1358
def assertEqual(self, a, b, message=''):
1362
except UnicodeError, e:
1363
# If we can't compare without getting a UnicodeError, then
1364
# obviously they are different
1365
trace.mutter('UnicodeError: %s', e)
1368
raise AssertionError("%snot equal:\na = %s\nb = %s\n"
1370
pprint.pformat(a), pprint.pformat(b)))
1372
# FIXME: This is deprecated in unittest2 but plugins may still use it so we
1373
# need a deprecation period for them -- vila 2016-02-01
1374
assertEquals = assertEqual
1376
def assertEqualDiff(self, a, b, message=None):
199
1377
"""Assert two texts are equal, if not raise an exception.
201
This is intended for use with multi-line strings where it can
1379
This is intended for use with multi-line strings where it can
202
1380
be hard to find the differences by eye.
204
# TODO: perhaps override assertEquals to call this for strings?
1382
# TODO: perhaps override assertEqual to call this for strings?
207
raise AssertionError("texts not equal:\n" +
208
self._ndiff_strings(a, b))
210
def assertContainsRe(self, haystack, needle_re):
1386
message = "texts not equal:\n"
1388
message = 'first string is missing a final newline.\n'
1390
message = 'second string is missing a final newline.\n'
1391
raise AssertionError(message +
1392
self._ndiff_strings(a, b))
1394
def assertEqualMode(self, mode, mode_test):
1395
self.assertEqual(mode, mode_test,
1396
'mode mismatch %o != %o' % (mode, mode_test))
1398
def assertEqualStat(self, expected, actual):
1399
"""assert that expected and actual are the same stat result.
1401
:param expected: A stat result.
1402
:param actual: A stat result.
1403
:raises AssertionError: If the expected and actual stat values differ
1404
other than by atime.
1406
self.assertEqual(expected.st_size, actual.st_size,
1407
'st_size did not match')
1408
self.assertEqual(expected.st_mtime, actual.st_mtime,
1409
'st_mtime did not match')
1410
self.assertEqual(expected.st_ctime, actual.st_ctime,
1411
'st_ctime did not match')
1412
if sys.platform == 'win32':
1413
# On Win32 both 'dev' and 'ino' cannot be trusted. In python2.4 it
1414
# is 'dev' that varies, in python 2.5 (6?) it is st_ino that is
1415
# odd. We just force it to always be 0 to avoid any problems.
1416
self.assertEqual(0, expected.st_dev)
1417
self.assertEqual(0, actual.st_dev)
1418
self.assertEqual(0, expected.st_ino)
1419
self.assertEqual(0, actual.st_ino)
1421
self.assertEqual(expected.st_dev, actual.st_dev,
1422
'st_dev did not match')
1423
self.assertEqual(expected.st_ino, actual.st_ino,
1424
'st_ino did not match')
1425
self.assertEqual(expected.st_mode, actual.st_mode,
1426
'st_mode did not match')
1428
def assertLength(self, length, obj_with_len):
1429
"""Assert that obj_with_len is of length length."""
1430
if len(obj_with_len) != length:
1431
self.fail("Incorrect length: wanted %d, got %d for %r" % (
1432
length, len(obj_with_len), obj_with_len))
1434
def assertLogsError(self, exception_class, func, *args, **kwargs):
1435
"""Assert that `func(*args, **kwargs)` quietly logs a specific error.
1438
orig_log_exception_quietly = trace.log_exception_quietly
1441
orig_log_exception_quietly()
1442
captured.append(sys.exc_info()[1])
1443
trace.log_exception_quietly = capture
1444
func(*args, **kwargs)
1446
trace.log_exception_quietly = orig_log_exception_quietly
1447
self.assertLength(1, captured)
1449
self.assertIsInstance(err, exception_class)
1452
def assertPositive(self, val):
1453
"""Assert that val is greater than 0."""
1454
self.assertTrue(val > 0, 'expected a positive value, but got %s' % val)
1456
def assertNegative(self, val):
1457
"""Assert that val is less than 0."""
1458
self.assertTrue(val < 0, 'expected a negative value, but got %s' % val)
1460
def assertStartsWith(self, s, prefix):
1461
if not s.startswith(prefix):
1462
raise AssertionError('string %r does not start with %r' % (s, prefix))
1464
def assertEndsWith(self, s, suffix):
1465
"""Asserts that s ends with suffix."""
1466
if not s.endswith(suffix):
1467
raise AssertionError('string %r does not end with %r' % (s, suffix))
1469
def assertContainsRe(self, haystack, needle_re, flags=0):
211
1470
"""Assert that a contains something matching a regular expression."""
212
if not re.search(needle_re, haystack):
213
raise AssertionError('pattern "%s" not found in "%s"'
1471
if not re.search(needle_re, haystack, flags):
1472
if '\n' in haystack or len(haystack) > 60:
1473
# a long string, format it in a more readable way
1474
raise AssertionError(
1475
'pattern "%s" not found in\n"""\\\n%s"""\n'
1476
% (needle_re, haystack))
1478
raise AssertionError('pattern "%s" not found in "%s"'
1479
% (needle_re, haystack))
1481
def assertNotContainsRe(self, haystack, needle_re, flags=0):
1482
"""Assert that a does not match a regular expression"""
1483
if re.search(needle_re, haystack, flags):
1484
raise AssertionError('pattern "%s" found in "%s"'
214
1485
% (needle_re, haystack))
216
def _enable_file_logging(self):
217
fileno, name = tempfile.mkstemp(suffix='.log', prefix='testbzr')
219
self._log_file = os.fdopen(fileno, 'w+')
221
hdlr = logging.StreamHandler(self._log_file)
222
hdlr.setLevel(logging.DEBUG)
223
hdlr.setFormatter(logging.Formatter('%(levelname)8s %(message)s'))
224
logging.getLogger('').addHandler(hdlr)
225
logging.getLogger('').setLevel(logging.DEBUG)
226
self._log_hdlr = hdlr
227
debug('opened log file %s', name)
1487
def assertContainsString(self, haystack, needle):
1488
if haystack.find(needle) == -1:
1489
self.fail("string %r not found in '''%s'''" % (needle, haystack))
1491
def assertNotContainsString(self, haystack, needle):
1492
if haystack.find(needle) != -1:
1493
self.fail("string %r found in '''%s'''" % (needle, haystack))
1495
def assertSubset(self, sublist, superlist):
1496
"""Assert that every entry in sublist is present in superlist."""
1497
missing = set(sublist) - set(superlist)
1498
if len(missing) > 0:
1499
raise AssertionError("value(s) %r not present in container %r" %
1500
(missing, superlist))
1502
def assertListRaises(self, excClass, func, *args, **kwargs):
1503
"""Fail unless excClass is raised when the iterator from func is used.
1505
Many functions can return generators this makes sure
1506
to wrap them in a list() call to make sure the whole generator
1507
is run, and that the proper exception is raised.
1510
list(func(*args, **kwargs))
1514
if getattr(excClass,'__name__', None) is not None:
1515
excName = excClass.__name__
1517
excName = str(excClass)
1518
raise self.failureException, "%s not raised" % excName
1520
def assertRaises(self, excClass, callableObj, *args, **kwargs):
1521
"""Assert that a callable raises a particular exception.
1523
:param excClass: As for the except statement, this may be either an
1524
exception class, or a tuple of classes.
1525
:param callableObj: A callable, will be passed ``*args`` and
1528
Returns the exception so that you can examine it.
1531
callableObj(*args, **kwargs)
1535
if getattr(excClass,'__name__', None) is not None:
1536
excName = excClass.__name__
1539
excName = str(excClass)
1540
raise self.failureException, "%s not raised" % excName
1542
def assertIs(self, left, right, message=None):
1543
if not (left is right):
1544
if message is not None:
1545
raise AssertionError(message)
1547
raise AssertionError("%r is not %r." % (left, right))
1549
def assertIsNot(self, left, right, message=None):
1551
if message is not None:
1552
raise AssertionError(message)
1554
raise AssertionError("%r is %r." % (left, right))
1556
def assertTransportMode(self, transport, path, mode):
1557
"""Fail if a path does not have mode "mode".
1559
If modes are not supported on this transport, the assertion is ignored.
1561
if not transport._can_roundtrip_unix_modebits():
1563
path_stat = transport.stat(path)
1564
actual_mode = stat.S_IMODE(path_stat.st_mode)
1565
self.assertEqual(mode, actual_mode,
1566
'mode of %r incorrect (%s != %s)'
1567
% (path, oct(mode), oct(actual_mode)))
1569
def assertIsSameRealPath(self, path1, path2):
1570
"""Fail if path1 and path2 points to different files"""
1571
self.assertEqual(osutils.realpath(path1),
1572
osutils.realpath(path2),
1573
"apparent paths:\na = %s\nb = %s\n," % (path1, path2))
1575
def assertIsInstance(self, obj, kls, msg=None):
1576
"""Fail if obj is not an instance of kls
229
self._log_file_name = name
232
os.environ['HOME'] = self.oldenv
233
if os.environ.get('BZREMAIL') is not None:
234
del os.environ['BZREMAIL']
235
if self.bzr_email is not None:
236
os.environ['BZREMAIL'] = self.bzr_email
237
if os.environ.get('EMAIL') is not None:
238
del os.environ['EMAIL']
239
if self.email is not None:
240
os.environ['EMAIL'] = self.email
241
logging.getLogger('').removeHandler(self._log_hdlr)
242
bzrlib.trace.enable_default_logging()
243
logging.debug('%s teardown', self.id())
244
self._log_file.close()
245
unittest.TestCase.tearDown(self)
1578
:param msg: Supplementary message to show if the assertion fails.
1580
if not isinstance(obj, kls):
1581
m = "%r is an instance of %s rather than %s" % (
1582
obj, obj.__class__, kls)
1587
def assertFileEqual(self, content, path):
1588
"""Fail if path does not contain 'content'."""
1589
self.assertPathExists(path)
1590
f = file(path, 'rb')
1595
self.assertEqualDiff(content, s)
1597
def assertDocstring(self, expected_docstring, obj):
1598
"""Fail if obj does not have expected_docstring"""
1600
# With -OO the docstring should be None instead
1601
self.assertIs(obj.__doc__, None)
1603
self.assertEqual(expected_docstring, obj.__doc__)
1605
@symbol_versioning.deprecated_method(symbol_versioning.deprecated_in((2, 4)))
1606
def failUnlessExists(self, path):
1607
return self.assertPathExists(path)
1609
def assertPathExists(self, path):
1610
"""Fail unless path or paths, which may be abs or relative, exist."""
1611
if not isinstance(path, basestring):
1613
self.assertPathExists(p)
1615
self.assertTrue(osutils.lexists(path),
1616
path + " does not exist")
1618
@symbol_versioning.deprecated_method(symbol_versioning.deprecated_in((2, 4)))
1619
def failIfExists(self, path):
1620
return self.assertPathDoesNotExist(path)
1622
def assertPathDoesNotExist(self, path):
1623
"""Fail if path or paths, which may be abs or relative, exist."""
1624
if not isinstance(path, basestring):
1626
self.assertPathDoesNotExist(p)
1628
self.assertFalse(osutils.lexists(path),
1631
def _capture_deprecation_warnings(self, a_callable, *args, **kwargs):
1632
"""A helper for callDeprecated and applyDeprecated.
1634
:param a_callable: A callable to call.
1635
:param args: The positional arguments for the callable
1636
:param kwargs: The keyword arguments for the callable
1637
:return: A tuple (warnings, result). result is the result of calling
1638
a_callable(``*args``, ``**kwargs``).
1641
def capture_warnings(msg, cls=None, stacklevel=None):
1642
# we've hooked into a deprecation specific callpath,
1643
# only deprecations should getting sent via it.
1644
self.assertEqual(cls, DeprecationWarning)
1645
local_warnings.append(msg)
1646
original_warning_method = symbol_versioning.warn
1647
symbol_versioning.set_warning_method(capture_warnings)
1649
result = a_callable(*args, **kwargs)
1651
symbol_versioning.set_warning_method(original_warning_method)
1652
return (local_warnings, result)
1654
def applyDeprecated(self, deprecation_format, a_callable, *args, **kwargs):
1655
"""Call a deprecated callable without warning the user.
1657
Note that this only captures warnings raised by symbol_versioning.warn,
1658
not other callers that go direct to the warning module.
1660
To test that a deprecated method raises an error, do something like
1661
this (remember that both assertRaises and applyDeprecated delays *args
1662
and **kwargs passing)::
1664
self.assertRaises(errors.ReservedId,
1665
self.applyDeprecated,
1666
deprecated_in((1, 5, 0)),
1670
:param deprecation_format: The deprecation format that the callable
1671
should have been deprecated with. This is the same type as the
1672
parameter to deprecated_method/deprecated_function. If the
1673
callable is not deprecated with this format, an assertion error
1675
:param a_callable: A callable to call. This may be a bound method or
1676
a regular function. It will be called with ``*args`` and
1678
:param args: The positional arguments for the callable
1679
:param kwargs: The keyword arguments for the callable
1680
:return: The result of a_callable(``*args``, ``**kwargs``)
1682
call_warnings, result = self._capture_deprecation_warnings(a_callable,
1684
expected_first_warning = symbol_versioning.deprecation_string(
1685
a_callable, deprecation_format)
1686
if len(call_warnings) == 0:
1687
self.fail("No deprecation warning generated by call to %s" %
1689
self.assertEqual(expected_first_warning, call_warnings[0])
1692
def callCatchWarnings(self, fn, *args, **kw):
1693
"""Call a callable that raises python warnings.
1695
The caller's responsible for examining the returned warnings.
1697
If the callable raises an exception, the exception is not
1698
caught and propagates up to the caller. In that case, the list
1699
of warnings is not available.
1701
:returns: ([warning_object, ...], fn_result)
1703
# XXX: This is not perfect, because it completely overrides the
1704
# warnings filters, and some code may depend on suppressing particular
1705
# warnings. It's the easiest way to insulate ourselves from -Werror,
1706
# though. -- Andrew, 20071062
1708
def _catcher(message, category, filename, lineno, file=None, line=None):
1709
# despite the name, 'message' is normally(?) a Warning subclass
1711
wlist.append(message)
1712
saved_showwarning = warnings.showwarning
1713
saved_filters = warnings.filters
1715
warnings.showwarning = _catcher
1716
warnings.filters = []
1717
result = fn(*args, **kw)
1719
warnings.showwarning = saved_showwarning
1720
warnings.filters = saved_filters
1721
return wlist, result
1723
def callDeprecated(self, expected, callable, *args, **kwargs):
1724
"""Assert that a callable is deprecated in a particular way.
1726
This is a very precise test for unusual requirements. The
1727
applyDeprecated helper function is probably more suited for most tests
1728
as it allows you to simply specify the deprecation format being used
1729
and will ensure that that is issued for the function being called.
1731
Note that this only captures warnings raised by symbol_versioning.warn,
1732
not other callers that go direct to the warning module. To catch
1733
general warnings, use callCatchWarnings.
1735
:param expected: a list of the deprecation warnings expected, in order
1736
:param callable: The callable to call
1737
:param args: The positional arguments for the callable
1738
:param kwargs: The keyword arguments for the callable
1740
call_warnings, result = self._capture_deprecation_warnings(callable,
1742
self.assertEqual(expected, call_warnings)
1745
def _startLogFile(self):
1746
"""Setup a in-memory target for bzr and testcase log messages"""
1747
pseudo_log_file = StringIO()
1748
def _get_log_contents_for_weird_testtools_api():
1749
return [pseudo_log_file.getvalue().decode(
1750
"utf-8", "replace").encode("utf-8")]
1751
self.addDetail("log", content.Content(content.ContentType("text",
1752
"plain", {"charset": "utf8"}),
1753
_get_log_contents_for_weird_testtools_api))
1754
self._log_file = pseudo_log_file
1755
self._log_memento = trace.push_log_file(self._log_file)
1756
self.addCleanup(self._finishLogFile)
1758
def _finishLogFile(self):
1759
"""Flush and dereference the in-memory log for this testcase"""
1760
if trace._trace_file:
1761
# flush the log file, to get all content
1762
trace._trace_file.flush()
1763
trace.pop_log_file(self._log_memento)
1764
# The logging module now tracks references for cleanup so discard ours
1765
del self._log_memento
1767
def thisFailsStrictLockCheck(self):
1768
"""It is known that this test would fail with -Dstrict_locks.
1770
By default, all tests are run with strict lock checking unless
1771
-Edisable_lock_checks is supplied. However there are some tests which
1772
we know fail strict locks at this point that have not been fixed.
1773
They should call this function to disable the strict checking.
1775
This should be used sparingly, it is much better to fix the locking
1776
issues rather than papering over the problem by calling this function.
1778
debug.debug_flags.discard('strict_locks')
1780
def overrideAttr(self, obj, attr_name, new=_unitialized_attr):
1781
"""Overrides an object attribute restoring it after the test.
1783
:note: This should be used with discretion; you should think about
1784
whether it's better to make the code testable without monkey-patching.
1786
:param obj: The object that will be mutated.
1788
:param attr_name: The attribute name we want to preserve/override in
1791
:param new: The optional value we want to set the attribute to.
1793
:returns: The actual attr value.
1795
# The actual value is captured by the call below
1796
value = getattr(obj, attr_name, _unitialized_attr)
1797
if value is _unitialized_attr:
1798
# When the test completes, the attribute should not exist, but if
1799
# we aren't setting a value, we don't need to do anything.
1800
if new is not _unitialized_attr:
1801
self.addCleanup(delattr, obj, attr_name)
1803
self.addCleanup(setattr, obj, attr_name, value)
1804
if new is not _unitialized_attr:
1805
setattr(obj, attr_name, new)
1808
def overrideEnv(self, name, new):
1809
"""Set an environment variable, and reset it after the test.
1811
:param name: The environment variable name.
1813
:param new: The value to set the variable to. If None, the
1814
variable is deleted from the environment.
1816
:returns: The actual variable value.
1818
value = osutils.set_or_unset_env(name, new)
1819
self.addCleanup(osutils.set_or_unset_env, name, value)
1822
def recordCalls(self, obj, attr_name):
1823
"""Monkeypatch in a wrapper that will record calls.
1825
The monkeypatch is automatically removed when the test concludes.
1827
:param obj: The namespace holding the reference to be replaced;
1828
typically a module, class, or object.
1829
:param attr_name: A string for the name of the attribute to
1831
:returns: A list that will be extended with one item every time the
1832
function is called, with a tuple of (args, kwargs).
1836
def decorator(*args, **kwargs):
1837
calls.append((args, kwargs))
1838
return orig(*args, **kwargs)
1839
orig = self.overrideAttr(obj, attr_name, decorator)
1842
def _cleanEnvironment(self):
1843
for name, value in isolated_environ.iteritems():
1844
self.overrideEnv(name, value)
1846
def _restoreHooks(self):
1847
for klass, (name, hooks) in self._preserved_hooks.items():
1848
setattr(klass, name, hooks)
1849
self._preserved_hooks.clear()
1850
bzrlib.hooks._lazy_hooks = self._preserved_lazy_hooks
1851
self._preserved_lazy_hooks.clear()
1853
def knownFailure(self, reason):
1854
"""Declare that this test fails for a known reason
1856
Tests that are known to fail should generally be using expectedFailure
1857
with an appropriate reverse assertion if a change could cause the test
1858
to start passing. Conversely if the test has no immediate prospect of
1859
succeeding then using skip is more suitable.
1861
When this method is called while an exception is being handled, that
1862
traceback will be used, otherwise a new exception will be thrown to
1863
provide one but won't be reported.
1865
self._add_reason(reason)
1867
exc_info = sys.exc_info()
1868
if exc_info != (None, None, None):
1869
self._report_traceback(exc_info)
1872
raise self.failureException(reason)
1873
except self.failureException:
1874
exc_info = sys.exc_info()
1875
# GZ 02-08-2011: Maybe cleanup this err.exc_info attribute too?
1876
raise testtools.testcase._ExpectedFailure(exc_info)
1880
def _suppress_log(self):
1881
"""Remove the log info from details."""
1882
self.discardDetail('log')
1884
def _do_skip(self, result, reason):
1885
self._suppress_log()
1886
addSkip = getattr(result, 'addSkip', None)
1887
if not callable(addSkip):
1888
result.addSuccess(result)
1890
addSkip(self, reason)
1893
def _do_known_failure(self, result, e):
1894
self._suppress_log()
1895
err = sys.exc_info()
1896
addExpectedFailure = getattr(result, 'addExpectedFailure', None)
1897
if addExpectedFailure is not None:
1898
addExpectedFailure(self, err)
1900
result.addSuccess(self)
1903
def _do_not_applicable(self, result, e):
1905
reason = 'No reason given'
1908
self._suppress_log ()
1909
addNotApplicable = getattr(result, 'addNotApplicable', None)
1910
if addNotApplicable is not None:
1911
result.addNotApplicable(self, reason)
1913
self._do_skip(result, reason)
1916
def _report_skip(self, result, err):
1917
"""Override the default _report_skip.
1919
We want to strip the 'log' detail. If we waint until _do_skip, it has
1920
already been formatted into the 'reason' string, and we can't pull it
1923
self._suppress_log()
1924
super(TestCase, self)._report_skip(self, result, err)
1927
def _report_expected_failure(self, result, err):
1930
See _report_skip for motivation.
1932
self._suppress_log()
1933
super(TestCase, self)._report_expected_failure(self, result, err)
1936
def _do_unsupported_or_skip(self, result, e):
1938
self._suppress_log()
1939
addNotSupported = getattr(result, 'addNotSupported', None)
1940
if addNotSupported is not None:
1941
result.addNotSupported(self, reason)
1943
self._do_skip(result, reason)
1945
def time(self, callable, *args, **kwargs):
1946
"""Run callable and accrue the time it takes to the benchmark time.
1948
If lsprofiling is enabled (i.e. by --lsprof-time to bzr selftest) then
1949
this will cause lsprofile statistics to be gathered and stored in
1952
if self._benchtime is None:
1953
self.addDetail('benchtime', content.Content(content.ContentType(
1954
"text", "plain"), lambda:[str(self._benchtime)]))
1958
if not self._gather_lsprof_in_benchmarks:
1959
return callable(*args, **kwargs)
1961
# record this benchmark
1962
ret, stats = bzrlib.lsprof.profile(callable, *args, **kwargs)
1964
self._benchcalls.append(((callable, args, kwargs), stats))
1967
self._benchtime += time.time() - start
247
1969
def log(self, *args):
251
"""Return as a string the log for this test"""
252
if self._log_file_name:
253
return open(self._log_file_name).read()
257
def capture(self, cmd):
258
"""Shortcut that splits cmd into words, runs, and returns stdout"""
259
return self.run_bzr_captured(cmd.split())[0]
261
def run_bzr_captured(self, argv, retcode=0):
262
"""Invoke bzr and return (result, stdout, stderr).
264
Useful for code that wants to check the contents of the
265
output, the way error messages are presented, etc.
267
This should be the main method for tests that want to exercise the
268
overall behavior of the bzr application (rather than a unit test
269
or a functional test of the library.)
271
Much of the old code runs bzr by forking a new copy of Python, but
272
that is slower, harder to debug, and generally not necessary.
274
This runs bzr through the interface that catches and reports
275
errors, and with logging set to something approximating the
276
default, so that error reporting can be checked.
278
argv -- arguments to invoke bzr
279
retcode -- expected return code, or None for don't-care.
283
self.log('run bzr: %s', ' '.join(argv))
284
handler = logging.StreamHandler(stderr)
285
handler.setFormatter(bzrlib.trace.QuietFormatter())
286
handler.setLevel(logging.INFO)
1973
"""Get a unicode string containing the log from bzrlib.trace.
1975
Undecodable characters are replaced.
1977
return u"".join(self.getDetails()['log'].iter_text())
1979
def requireFeature(self, feature):
1980
"""This test requires a specific feature is available.
1982
:raises UnavailableFeature: When feature is not available.
1984
if not feature.available():
1985
raise UnavailableFeature(feature)
1987
def _run_bzr_autosplit(self, args, retcode, encoding, stdin,
1989
"""Run bazaar command line, splitting up a string command line."""
1990
if isinstance(args, basestring):
1991
# shlex don't understand unicode strings,
1992
# so args should be plain string (bialix 20070906)
1993
args = list(shlex.split(str(args)))
1994
return self._run_bzr_core(args, retcode=retcode,
1995
encoding=encoding, stdin=stdin, working_dir=working_dir,
1998
def _run_bzr_core(self, args, retcode, encoding, stdin,
2000
# Clear chk_map page cache, because the contents are likely to mask
2002
chk_map.clear_cache()
2003
if encoding is None:
2004
encoding = osutils.get_user_encoding()
2005
stdout = StringIOWrapper()
2006
stderr = StringIOWrapper()
2007
stdout.encoding = encoding
2008
stderr.encoding = encoding
2010
self.log('run bzr: %r', args)
2011
# FIXME: don't call into logging here
2012
handler = trace.EncodedStreamHandler(stderr, errors="replace",
287
2014
logger = logging.getLogger('')
288
2015
logger.addHandler(handler)
2016
old_ui_factory = ui.ui_factory
2017
ui.ui_factory = TestUIFactory(stdin=stdin, stdout=stdout, stderr=stderr)
2020
if working_dir is not None:
2021
cwd = osutils.getcwd()
2022
os.chdir(working_dir)
290
result = self.apply_redirected(None, stdout, stderr,
291
bzrlib.commands.run_bzr_catch_errors,
2026
result = self.apply_redirected(
2027
ui.ui_factory.stdin,
2029
_mod_commands.run_bzr_catch_user_errors,
2031
except KeyboardInterrupt:
2032
# Reraise KeyboardInterrupt with contents of redirected stdout
2033
# and stderr as arguments, for tests which are interested in
2034
# stdout and stderr and are expecting the exception.
2035
out = stdout.getvalue()
2036
err = stderr.getvalue()
2038
self.log('output:\n%r', out)
2040
self.log('errors:\n%r', err)
2041
raise KeyboardInterrupt(out, err)
294
2043
logger.removeHandler(handler)
2044
ui.ui_factory = old_ui_factory
295
2048
out = stdout.getvalue()
296
2049
err = stderr.getvalue()
298
self.log('output:\n%s', out)
2051
self.log('output:\n%r', out)
300
self.log('errors:\n%s', err)
2053
self.log('errors:\n%r', err)
301
2054
if retcode is not None:
302
self.assertEquals(result, retcode)
2055
self.assertEqual(retcode, result,
2056
message='Unexpected return code')
2057
return result, out, err
305
def run_bzr(self, *args, **kwargs):
2059
def run_bzr(self, args, retcode=0, encoding=None, stdin=None,
2060
working_dir=None, error_regexes=[], output_encoding=None):
306
2061
"""Invoke bzr, as if it were run from the command line.
2063
The argument list should not include the bzr program name - the
2064
first argument is normally the bzr command. Arguments may be
2065
passed in three ways:
2067
1- A list of strings, eg ["commit", "a"]. This is recommended
2068
when the command contains whitespace or metacharacters, or
2069
is built up at run time.
2071
2- A single string, eg "add a". This is the most convenient
2072
for hardcoded commands.
2074
This runs bzr through the interface that catches and reports
2075
errors, and with logging set to something approximating the
2076
default, so that error reporting can be checked.
308
2078
This should be the main method for tests that want to exercise the
309
2079
overall behavior of the bzr application (rather than a unit test
310
2080
or a functional test of the library.)
312
2082
This sends the stdout/stderr results into the test's log,
313
2083
where it may be useful for debugging. See also run_captured.
315
retcode = kwargs.pop('retcode', 0)
316
return self.run_bzr_captured(args, retcode)
318
def check_inventory_shape(self, inv, shape):
319
"""Compare an inventory to a list of expected names.
2085
:keyword stdin: A string to be used as stdin for the command.
2086
:keyword retcode: The status code the command should return;
2088
:keyword working_dir: The directory to run the command in
2089
:keyword error_regexes: A list of expected error messages. If
2090
specified they must be seen in the error output of the command.
2092
retcode, out, err = self._run_bzr_autosplit(
2097
working_dir=working_dir,
2099
self.assertIsInstance(error_regexes, (list, tuple))
2100
for regex in error_regexes:
2101
self.assertContainsRe(err, regex)
2104
def run_bzr_error(self, error_regexes, *args, **kwargs):
2105
"""Run bzr, and check that stderr contains the supplied regexes
2107
:param error_regexes: Sequence of regular expressions which
2108
must each be found in the error output. The relative ordering
2110
:param args: command-line arguments for bzr
2111
:param kwargs: Keyword arguments which are interpreted by run_bzr
2112
This function changes the default value of retcode to be 3,
2113
since in most cases this is run when you expect bzr to fail.
2115
:return: (out, err) The actual output of running the command (in case
2116
you want to do more inspection)
2120
# Make sure that commit is failing because there is nothing to do
2121
self.run_bzr_error(['no changes to commit'],
2122
['commit', '-m', 'my commit comment'])
2123
# Make sure --strict is handling an unknown file, rather than
2124
# giving us the 'nothing to do' error
2125
self.build_tree(['unknown'])
2126
self.run_bzr_error(['Commit refused because there are unknown files'],
2127
['commit', --strict', '-m', 'my commit comment'])
2129
kwargs.setdefault('retcode', 3)
2130
kwargs['error_regexes'] = error_regexes
2131
out, err = self.run_bzr(*args, **kwargs)
2134
def run_bzr_subprocess(self, *args, **kwargs):
2135
"""Run bzr in a subprocess for testing.
2137
This starts a new Python interpreter and runs bzr in there.
2138
This should only be used for tests that have a justifiable need for
2139
this isolation: e.g. they are testing startup time, or signal
2140
handling, or early startup code, etc. Subprocess code can't be
2141
profiled or debugged so easily.
2143
:keyword retcode: The status code that is expected. Defaults to 0. If
2144
None is supplied, the status code is not checked.
2145
:keyword env_changes: A dictionary which lists changes to environment
2146
variables. A value of None will unset the env variable.
2147
The values must be strings. The change will only occur in the
2148
child, so you don't need to fix the environment after running.
2149
:keyword universal_newlines: Convert CRLF => LF
2150
:keyword allow_plugins: By default the subprocess is run with
2151
--no-plugins to ensure test reproducibility. Also, it is possible
2152
for system-wide plugins to create unexpected output on stderr,
2153
which can cause unnecessary test failures.
2155
env_changes = kwargs.get('env_changes', {})
2156
working_dir = kwargs.get('working_dir', None)
2157
allow_plugins = kwargs.get('allow_plugins', False)
2159
if isinstance(args[0], list):
2161
elif isinstance(args[0], basestring):
2162
args = list(shlex.split(args[0]))
2164
raise ValueError("passing varargs to run_bzr_subprocess")
2165
process = self.start_bzr_subprocess(args, env_changes=env_changes,
2166
working_dir=working_dir,
2167
allow_plugins=allow_plugins)
2168
# We distinguish between retcode=None and retcode not passed.
2169
supplied_retcode = kwargs.get('retcode', 0)
2170
return self.finish_bzr_subprocess(process, retcode=supplied_retcode,
2171
universal_newlines=kwargs.get('universal_newlines', False),
2174
def start_bzr_subprocess(self, process_args, env_changes=None,
2175
skip_if_plan_to_signal=False,
2177
allow_plugins=False, stderr=subprocess.PIPE):
2178
"""Start bzr in a subprocess for testing.
2180
This starts a new Python interpreter and runs bzr in there.
2181
This should only be used for tests that have a justifiable need for
2182
this isolation: e.g. they are testing startup time, or signal
2183
handling, or early startup code, etc. Subprocess code can't be
2184
profiled or debugged so easily.
2186
:param process_args: a list of arguments to pass to the bzr executable,
2187
for example ``['--version']``.
2188
:param env_changes: A dictionary which lists changes to environment
2189
variables. A value of None will unset the env variable.
2190
The values must be strings. The change will only occur in the
2191
child, so you don't need to fix the environment after running.
2192
:param skip_if_plan_to_signal: raise TestSkipped when true and system
2193
doesn't support signalling subprocesses.
2194
:param allow_plugins: If False (default) pass --no-plugins to bzr.
2195
:param stderr: file to use for the subprocess's stderr. Valid values
2196
are those valid for the stderr argument of `subprocess.Popen`.
2197
Default value is ``subprocess.PIPE``.
2199
:returns: Popen object for the started process.
2201
if skip_if_plan_to_signal:
2202
if os.name != "posix":
2203
raise TestSkipped("Sending signals not supported")
2205
if env_changes is None:
2207
# Because $HOME is set to a tempdir for the context of a test, modules
2208
# installed in the user dir will not be found unless $PYTHONUSERBASE
2209
# gets set to the computed directory of this parent process.
2210
if site.USER_BASE is not None:
2211
env_changes["PYTHONUSERBASE"] = site.USER_BASE
2214
def cleanup_environment():
2215
for env_var, value in env_changes.iteritems():
2216
old_env[env_var] = osutils.set_or_unset_env(env_var, value)
2218
def restore_environment():
2219
for env_var, value in old_env.iteritems():
2220
osutils.set_or_unset_env(env_var, value)
2222
bzr_path = self.get_bzr_path()
2225
if working_dir is not None:
2226
cwd = osutils.getcwd()
2227
os.chdir(working_dir)
2230
# win32 subprocess doesn't support preexec_fn
2231
# so we will avoid using it on all platforms, just to
2232
# make sure the code path is used, and we don't break on win32
2233
cleanup_environment()
2234
# Include the subprocess's log file in the test details, in case
2235
# the test fails due to an error in the subprocess.
2236
self._add_subprocess_log(trace._get_bzr_log_filename())
2237
command = [sys.executable]
2238
# frozen executables don't need the path to bzr
2239
if getattr(sys, "frozen", None) is None:
2240
command.append(bzr_path)
2241
if not allow_plugins:
2242
command.append('--no-plugins')
2243
command.extend(process_args)
2244
process = self._popen(command, stdin=subprocess.PIPE,
2245
stdout=subprocess.PIPE,
2248
restore_environment()
2254
def _add_subprocess_log(self, log_file_path):
2255
if len(self._log_files) == 0:
2256
# Register an addCleanup func. We do this on the first call to
2257
# _add_subprocess_log rather than in TestCase.setUp so that this
2258
# addCleanup is registered after any cleanups for tempdirs that
2259
# subclasses might create, which will probably remove the log file
2261
self.addCleanup(self._subprocess_log_cleanup)
2262
# self._log_files is a set, so if a log file is reused we won't grab it
2264
self._log_files.add(log_file_path)
2266
def _subprocess_log_cleanup(self):
2267
for count, log_file_path in enumerate(self._log_files):
2268
# We use buffer_now=True to avoid holding the file open beyond
2269
# the life of this function, which might interfere with e.g.
2270
# cleaning tempdirs on Windows.
2271
# XXX: Testtools 0.9.5 doesn't have the content_from_file helper
2272
#detail_content = content.content_from_file(
2273
# log_file_path, buffer_now=True)
2274
with open(log_file_path, 'rb') as log_file:
2275
log_file_bytes = log_file.read()
2276
detail_content = content.Content(content.ContentType("text",
2277
"plain", {"charset": "utf8"}), lambda: [log_file_bytes])
2278
self.addDetail("start_bzr_subprocess-log-%d" % (count,),
2281
def _popen(self, *args, **kwargs):
2282
"""Place a call to Popen.
2284
Allows tests to override this method to intercept the calls made to
2285
Popen for introspection.
2287
return subprocess.Popen(*args, **kwargs)
2289
def get_source_path(self):
2290
"""Return the path of the directory containing bzrlib."""
2291
return os.path.dirname(os.path.dirname(bzrlib.__file__))
2293
def get_bzr_path(self):
2294
"""Return the path of the 'bzr' executable for this test suite."""
2295
bzr_path = os.path.join(self.get_source_path(), "bzr")
2296
if not os.path.isfile(bzr_path):
2297
# We are probably installed. Assume sys.argv is the right file
2298
bzr_path = sys.argv[0]
2301
def finish_bzr_subprocess(self, process, retcode=0, send_signal=None,
2302
universal_newlines=False, process_args=None):
2303
"""Finish the execution of process.
2305
:param process: the Popen object returned from start_bzr_subprocess.
2306
:param retcode: The status code that is expected. Defaults to 0. If
2307
None is supplied, the status code is not checked.
2308
:param send_signal: an optional signal to send to the process.
2309
:param universal_newlines: Convert CRLF => LF
2310
:returns: (stdout, stderr)
2312
if send_signal is not None:
2313
os.kill(process.pid, send_signal)
2314
out, err = process.communicate()
2316
if universal_newlines:
2317
out = out.replace('\r\n', '\n')
2318
err = err.replace('\r\n', '\n')
2320
if retcode is not None and retcode != process.returncode:
2321
if process_args is None:
2322
process_args = "(unknown args)"
2323
trace.mutter('Output of bzr %s:\n%s', process_args, out)
2324
trace.mutter('Error for bzr %s:\n%s', process_args, err)
2325
self.fail('Command bzr %s failed with retcode %s != %s'
2326
% (process_args, retcode, process.returncode))
2329
def check_tree_shape(self, tree, shape):
2330
"""Compare a tree to a list of expected names.
321
2332
Fail if they are not precisely equal.
324
2335
shape = list(shape) # copy
325
for path, ie in inv.entries():
2336
for path, ie in tree.iter_entries_by_dir():
326
2337
name = path.replace('\\', '/')
2338
if ie.kind == 'directory':
328
2339
name = name + '/'
2341
pass # ignore root entry
330
2343
shape.remove(name)
332
2345
extras.append(name)
367
2380
sys.stderr = real_stderr
368
2381
sys.stdin = real_stdin
371
BzrTestBase = TestCase
374
class TestCaseInTempDir(TestCase):
2383
def reduceLockdirTimeout(self):
2384
"""Reduce the default lock timeout for the duration of the test, so that
2385
if LockContention occurs during a test, it does so quickly.
2387
Tests that expect to provoke LockContention errors should call this.
2389
self.overrideAttr(lockdir, '_DEFAULT_TIMEOUT_SECONDS', 0)
2391
def make_utf8_encoded_stringio(self, encoding_type=None):
2392
"""Return a StringIOWrapper instance, that will encode Unicode
2395
if encoding_type is None:
2396
encoding_type = 'strict'
2398
output_encoding = 'utf-8'
2399
sio = codecs.getwriter(output_encoding)(sio, errors=encoding_type)
2400
sio.encoding = output_encoding
2403
def disable_verb(self, verb):
2404
"""Disable a smart server verb for one test."""
2405
from bzrlib.smart import request
2406
request_handlers = request.request_handlers
2407
orig_method = request_handlers.get(verb)
2408
orig_info = request_handlers.get_info(verb)
2409
request_handlers.remove(verb)
2410
self.addCleanup(request_handlers.register, verb, orig_method,
2414
class CapturedCall(object):
2415
"""A helper for capturing smart server calls for easy debug analysis."""
2417
def __init__(self, params, prefix_length):
2418
"""Capture the call with params and skip prefix_length stack frames."""
2421
# The last 5 frames are the __init__, the hook frame, and 3 smart
2422
# client frames. Beyond this we could get more clever, but this is good
2424
stack = traceback.extract_stack()[prefix_length:-5]
2425
self.stack = ''.join(traceback.format_list(stack))
2428
return self.call.method
2431
return self.call.method
2437
class TestCaseWithMemoryTransport(TestCase):
2438
"""Common test class for tests that do not need disk resources.
2440
Tests that need disk resources should derive from TestCaseInTempDir
2441
orTestCaseWithTransport.
2443
TestCaseWithMemoryTransport sets the TEST_ROOT variable for all bzr tests.
2445
For TestCaseWithMemoryTransport the ``test_home_dir`` is set to the name of
2446
a directory which does not exist. This serves to help ensure test isolation
2447
is preserved. ``test_dir`` is set to the TEST_ROOT, as is cwd, because they
2448
must exist. However, TestCaseWithMemoryTransport does not offer local file
2449
defaults for the transport in tests, nor does it obey the command line
2450
override, so tests that accidentally write to the common directory should
2453
:cvar TEST_ROOT: Directory containing all temporary directories, plus a
2454
``.bzr`` directory that stops us ascending higher into the filesystem.
2460
def __init__(self, methodName='runTest'):
2461
# allow test parameterization after test construction and before test
2462
# execution. Variables that the parameterizer sets need to be
2463
# ones that are not set by setUp, or setUp will trash them.
2464
super(TestCaseWithMemoryTransport, self).__init__(methodName)
2465
self.vfs_transport_factory = default_transport
2466
self.transport_server = None
2467
self.transport_readonly_server = None
2468
self.__vfs_server = None
2471
super(TestCaseWithMemoryTransport, self).setUp()
2473
def _add_disconnect_cleanup(transport):
2474
"""Schedule disconnection of given transport at test cleanup
2476
This needs to happen for all connected transports or leaks occur.
2478
Note reconnections may mean we call disconnect multiple times per
2479
transport which is suboptimal but seems harmless.
2481
self.addCleanup(transport.disconnect)
2483
_mod_transport.Transport.hooks.install_named_hook('post_connect',
2484
_add_disconnect_cleanup, None)
2486
self._make_test_root()
2487
self.addCleanup(os.chdir, os.getcwdu())
2488
self.makeAndChdirToTestDir()
2489
self.overrideEnvironmentForTesting()
2490
self.__readonly_server = None
2491
self.__server = None
2492
self.reduceLockdirTimeout()
2493
# Each test may use its own config files even if the local config files
2494
# don't actually exist. They'll rightly fail if they try to create them
2496
self.overrideAttr(config, '_shared_stores', {})
2498
def get_transport(self, relpath=None):
2499
"""Return a writeable transport.
2501
This transport is for the test scratch space relative to
2504
:param relpath: a path relative to the base url.
2506
t = _mod_transport.get_transport_from_url(self.get_url(relpath))
2507
self.assertFalse(t.is_readonly())
2510
def get_readonly_transport(self, relpath=None):
2511
"""Return a readonly transport for the test scratch space
2513
This can be used to test that operations which should only need
2514
readonly access in fact do not try to write.
2516
:param relpath: a path relative to the base url.
2518
t = _mod_transport.get_transport_from_url(
2519
self.get_readonly_url(relpath))
2520
self.assertTrue(t.is_readonly())
2523
def create_transport_readonly_server(self):
2524
"""Create a transport server from class defined at init.
2526
This is mostly a hook for daughter classes.
2528
return self.transport_readonly_server()
2530
def get_readonly_server(self):
2531
"""Get the server instance for the readonly transport
2533
This is useful for some tests with specific servers to do diagnostics.
2535
if self.__readonly_server is None:
2536
if self.transport_readonly_server is None:
2537
# readonly decorator requested
2538
self.__readonly_server = test_server.ReadonlyServer()
2540
# explicit readonly transport.
2541
self.__readonly_server = self.create_transport_readonly_server()
2542
self.start_server(self.__readonly_server,
2543
self.get_vfs_only_server())
2544
return self.__readonly_server
2546
def get_readonly_url(self, relpath=None):
2547
"""Get a URL for the readonly transport.
2549
This will either be backed by '.' or a decorator to the transport
2550
used by self.get_url()
2551
relpath provides for clients to get a path relative to the base url.
2552
These should only be downwards relative, not upwards.
2554
base = self.get_readonly_server().get_url()
2555
return self._adjust_url(base, relpath)
2557
def get_vfs_only_server(self):
2558
"""Get the vfs only read/write server instance.
2560
This is useful for some tests with specific servers that need
2563
For TestCaseWithMemoryTransport this is always a MemoryServer, and there
2564
is no means to override it.
2566
if self.__vfs_server is None:
2567
self.__vfs_server = memory.MemoryServer()
2568
self.start_server(self.__vfs_server)
2569
return self.__vfs_server
2571
def get_server(self):
2572
"""Get the read/write server instance.
2574
This is useful for some tests with specific servers that need
2577
This is built from the self.transport_server factory. If that is None,
2578
then the self.get_vfs_server is returned.
2580
if self.__server is None:
2581
if (self.transport_server is None or self.transport_server is
2582
self.vfs_transport_factory):
2583
self.__server = self.get_vfs_only_server()
2585
# bring up a decorated means of access to the vfs only server.
2586
self.__server = self.transport_server()
2587
self.start_server(self.__server, self.get_vfs_only_server())
2588
return self.__server
2590
def _adjust_url(self, base, relpath):
2591
"""Get a URL (or maybe a path) for the readwrite transport.
2593
This will either be backed by '.' or to an equivalent non-file based
2595
relpath provides for clients to get a path relative to the base url.
2596
These should only be downwards relative, not upwards.
2598
if relpath is not None and relpath != '.':
2599
if not base.endswith('/'):
2601
# XXX: Really base should be a url; we did after all call
2602
# get_url()! But sometimes it's just a path (from
2603
# LocalAbspathServer), and it'd be wrong to append urlescaped data
2604
# to a non-escaped local path.
2605
if base.startswith('./') or base.startswith('/'):
2608
base += urlutils.escape(relpath)
2611
def get_url(self, relpath=None):
2612
"""Get a URL (or maybe a path) for the readwrite transport.
2614
This will either be backed by '.' or to an equivalent non-file based
2616
relpath provides for clients to get a path relative to the base url.
2617
These should only be downwards relative, not upwards.
2619
base = self.get_server().get_url()
2620
return self._adjust_url(base, relpath)
2622
def get_vfs_only_url(self, relpath=None):
2623
"""Get a URL (or maybe a path for the plain old vfs transport.
2625
This will never be a smart protocol. It always has all the
2626
capabilities of the local filesystem, but it might actually be a
2627
MemoryTransport or some other similar virtual filesystem.
2629
This is the backing transport (if any) of the server returned by
2630
get_url and get_readonly_url.
2632
:param relpath: provides for clients to get a path relative to the base
2633
url. These should only be downwards relative, not upwards.
2636
base = self.get_vfs_only_server().get_url()
2637
return self._adjust_url(base, relpath)
2639
def _create_safety_net(self):
2640
"""Make a fake bzr directory.
2642
This prevents any tests propagating up onto the TEST_ROOT directory's
2645
root = TestCaseWithMemoryTransport.TEST_ROOT
2647
# Make sure we get a readable and accessible home for .bzr.log
2648
# and/or config files, and not fallback to weird defaults (see
2649
# http://pad.lv/825027).
2650
self.assertIs(None, os.environ.get('BZR_HOME', None))
2651
os.environ['BZR_HOME'] = root
2652
wt = controldir.ControlDir.create_standalone_workingtree(root)
2653
del os.environ['BZR_HOME']
2654
except Exception, e:
2655
self.fail("Fail to initialize the safety net: %r\n" % (e,))
2656
# Hack for speed: remember the raw bytes of the dirstate file so that
2657
# we don't need to re-open the wt to check it hasn't changed.
2658
TestCaseWithMemoryTransport._SAFETY_NET_PRISTINE_DIRSTATE = (
2659
wt.control_transport.get_bytes('dirstate'))
2661
def _check_safety_net(self):
2662
"""Check that the safety .bzr directory have not been touched.
2664
_make_test_root have created a .bzr directory to prevent tests from
2665
propagating. This method ensures than a test did not leaked.
2667
root = TestCaseWithMemoryTransport.TEST_ROOT
2668
t = _mod_transport.get_transport_from_path(root)
2669
self.permit_url(t.base)
2670
if (t.get_bytes('.bzr/checkout/dirstate') !=
2671
TestCaseWithMemoryTransport._SAFETY_NET_PRISTINE_DIRSTATE):
2672
# The current test have modified the /bzr directory, we need to
2673
# recreate a new one or all the followng tests will fail.
2674
# If you need to inspect its content uncomment the following line
2675
# import pdb; pdb.set_trace()
2676
_rmtree_temp_dir(root + '/.bzr', test_id=self.id())
2677
self._create_safety_net()
2678
raise AssertionError('%s/.bzr should not be modified' % root)
2680
def _make_test_root(self):
2681
if TestCaseWithMemoryTransport.TEST_ROOT is None:
2682
# Watch out for tricky test dir (on OSX /tmp -> /private/tmp)
2683
root = osutils.realpath(osutils.mkdtemp(prefix='testbzr-',
2685
TestCaseWithMemoryTransport.TEST_ROOT = root
2687
self._create_safety_net()
2689
# The same directory is used by all tests, and we're not
2690
# specifically told when all tests are finished. This will do.
2691
atexit.register(_rmtree_temp_dir, root)
2693
self.permit_dir(TestCaseWithMemoryTransport.TEST_ROOT)
2694
self.addCleanup(self._check_safety_net)
2696
def makeAndChdirToTestDir(self):
2697
"""Create a temporary directories for this one test.
2699
This must set self.test_home_dir and self.test_dir and chdir to
2702
For TestCaseWithMemoryTransport we chdir to the TEST_ROOT for this test.
2704
os.chdir(TestCaseWithMemoryTransport.TEST_ROOT)
2705
self.test_dir = TestCaseWithMemoryTransport.TEST_ROOT
2706
self.test_home_dir = self.test_dir + "/MemoryTransportMissingHomeDir"
2707
self.permit_dir(self.test_dir)
2709
def make_branch(self, relpath, format=None, name=None):
2710
"""Create a branch on the transport at relpath."""
2711
repo = self.make_repository(relpath, format=format)
2712
return repo.bzrdir.create_branch(append_revisions_only=False,
2715
def get_default_format(self):
2718
def resolve_format(self, format):
2719
"""Resolve an object to a ControlDir format object.
2721
The initial format object can either already be
2722
a ControlDirFormat, None (for the default format),
2723
or a string with the name of the control dir format.
2725
:param format: Object to resolve
2726
:return A ControlDirFormat instance
2729
format = self.get_default_format()
2730
if isinstance(format, basestring):
2731
format = controldir.format_registry.make_bzrdir(format)
2734
def make_bzrdir(self, relpath, format=None):
2736
# might be a relative or absolute path
2737
maybe_a_url = self.get_url(relpath)
2738
segments = maybe_a_url.rsplit('/', 1)
2739
t = _mod_transport.get_transport(maybe_a_url)
2740
if len(segments) > 1 and segments[-1] not in ('', '.'):
2742
format = self.resolve_format(format)
2743
return format.initialize_on_transport(t)
2744
except errors.UninitializableFormat:
2745
raise TestSkipped("Format %s is not initializable." % format)
2747
def make_repository(self, relpath, shared=None, format=None):
2748
"""Create a repository on our default transport at relpath.
2750
Note that relpath must be a relative path, not a full url.
2752
# FIXME: If you create a remoterepository this returns the underlying
2753
# real format, which is incorrect. Actually we should make sure that
2754
# RemoteBzrDir returns a RemoteRepository.
2755
# maybe mbp 20070410
2756
made_control = self.make_bzrdir(relpath, format=format)
2757
return made_control.create_repository(shared=shared)
2759
def make_smart_server(self, path, backing_server=None):
2760
if backing_server is None:
2761
backing_server = self.get_server()
2762
smart_server = test_server.SmartTCPServer_for_testing()
2763
self.start_server(smart_server, backing_server)
2764
remote_transport = _mod_transport.get_transport_from_url(smart_server.get_url()
2766
return remote_transport
2768
def make_branch_and_memory_tree(self, relpath, format=None):
2769
"""Create a branch on the default transport and a MemoryTree for it."""
2770
b = self.make_branch(relpath, format=format)
2771
return memorytree.MemoryTree.create_on_branch(b)
2773
def make_branch_builder(self, relpath, format=None):
2774
branch = self.make_branch(relpath, format=format)
2775
return branchbuilder.BranchBuilder(branch=branch)
2777
def overrideEnvironmentForTesting(self):
2778
test_home_dir = self.test_home_dir
2779
if isinstance(test_home_dir, unicode):
2780
test_home_dir = test_home_dir.encode(sys.getfilesystemencoding())
2781
self.overrideEnv('HOME', test_home_dir)
2782
self.overrideEnv('BZR_HOME', test_home_dir)
2784
def setup_smart_server_with_call_log(self):
2785
"""Sets up a smart server as the transport server with a call log."""
2786
self.transport_server = test_server.SmartTCPServer_for_testing
2787
self.hpss_connections = []
2788
self.hpss_calls = []
2790
# Skip the current stack down to the caller of
2791
# setup_smart_server_with_call_log
2792
prefix_length = len(traceback.extract_stack()) - 2
2793
def capture_hpss_call(params):
2794
self.hpss_calls.append(
2795
CapturedCall(params, prefix_length))
2796
def capture_connect(transport):
2797
self.hpss_connections.append(transport)
2798
client._SmartClient.hooks.install_named_hook(
2799
'call', capture_hpss_call, None)
2800
_mod_transport.Transport.hooks.install_named_hook(
2801
'post_connect', capture_connect, None)
2803
def reset_smart_call_log(self):
2804
self.hpss_calls = []
2805
self.hpss_connections = []
2808
class TestCaseInTempDir(TestCaseWithMemoryTransport):
375
2809
"""Derived class that runs a test within a temporary directory.
377
2811
This is useful for tests that need to create a branch, etc.
381
2815
All test cases create their own directory within that. If the
382
2816
tests complete successfully, the directory is removed.
384
InTempDir is an old alias for FunctionalTestCase.
2818
:ivar test_base_dir: The path of the top-level directory for this
2819
test, which contains a home directory and a work directory.
2821
:ivar test_home_dir: An initially empty directory under test_base_dir
2822
which is used as $HOME for this test.
2824
:ivar test_dir: A directory under test_base_dir used as the current
2825
directory when the test proper is run.
389
2828
OVERRIDE_PYTHON = 'python'
2831
super(TestCaseInTempDir, self).setUp()
2832
# Remove the protection set in isolated_environ, we have a proper
2833
# access to disk resources now.
2834
self.overrideEnv('BZR_LOG', None)
391
2836
def check_file_contents(self, filename, expect):
392
2837
self.log("check contents of file %s" % filename)
393
contents = file(filename, 'r').read()
394
2843
if contents != expect:
395
2844
self.log("expected: %r" % expect)
396
2845
self.log("actually: %r" % contents)
397
2846
self.fail("contents of %s not as expected" % filename)
399
def _make_test_root(self):
400
if TestCaseInTempDir.TEST_ROOT is not None:
404
root = 'test%04d.tmp' % i
408
if e.errno == errno.EEXIST:
413
# successfully created
414
TestCaseInTempDir.TEST_ROOT = os.path.abspath(root)
416
# make a fake bzr directory there to prevent any tests propagating
417
# up onto the source directory's real branch
418
os.mkdir(os.path.join(TestCaseInTempDir.TEST_ROOT, '.bzr'))
421
self._make_test_root()
422
self._currentdir = os.getcwdu()
423
short_id = self.id().replace('bzrlib.selftest.', '') \
424
.replace('__main__.', '')
425
self.test_dir = os.path.join(self.TEST_ROOT, short_id)
2848
def _getTestDirPrefix(self):
2849
# create a directory within the top level test directory
2850
if sys.platform in ('win32', 'cygwin'):
2851
name_prefix = re.sub('[<>*=+",:;_/\\-]', '_', self.id())
2852
# windows is likely to have path-length limits so use a short name
2853
name_prefix = name_prefix[-30:]
2855
name_prefix = re.sub('[/]', '_', self.id())
2858
def makeAndChdirToTestDir(self):
2859
"""See TestCaseWithMemoryTransport.makeAndChdirToTestDir().
2861
For TestCaseInTempDir we create a temporary directory based on the test
2862
name and then create two subdirs - test and home under it.
2864
name_prefix = osutils.pathjoin(TestCaseWithMemoryTransport.TEST_ROOT,
2865
self._getTestDirPrefix())
2867
for i in range(100):
2868
if os.path.exists(name):
2869
name = name_prefix + '_' + str(i)
2871
# now create test and home directories within this dir
2872
self.test_base_dir = name
2873
self.addCleanup(self.deleteTestDir)
2874
os.mkdir(self.test_base_dir)
2876
self.permit_dir(self.test_base_dir)
2877
# 'sprouting' and 'init' of a branch both walk up the tree to find
2878
# stacking policy to honour; create a bzr dir with an unshared
2879
# repository (but not a branch - our code would be trying to escape
2880
# then!) to stop them, and permit it to be read.
2881
# control = controldir.ControlDir.create(self.test_base_dir)
2882
# control.create_repository()
2883
self.test_home_dir = self.test_base_dir + '/home'
2884
os.mkdir(self.test_home_dir)
2885
self.test_dir = self.test_base_dir + '/work'
426
2886
os.mkdir(self.test_dir)
427
2887
os.chdir(self.test_dir)
428
super(TestCaseInTempDir, self).setUp()
431
os.chdir(self._currentdir)
432
super(TestCaseInTempDir, self).tearDown()
434
def build_tree(self, shape):
2888
# put name of test inside
2889
f = file(self.test_base_dir + '/name', 'w')
2895
def deleteTestDir(self):
2896
os.chdir(TestCaseWithMemoryTransport.TEST_ROOT)
2897
_rmtree_temp_dir(self.test_base_dir, test_id=self.id())
2899
def build_tree(self, shape, line_endings='binary', transport=None):
435
2900
"""Build a test tree according to a pattern.
437
2902
shape is a sequence of file specifications. If the final
438
2903
character is '/', a directory is created.
2905
This assumes that all the elements in the tree being built are new.
440
2907
This doesn't add anything to a branch.
2909
:type shape: list or tuple.
2910
:param line_endings: Either 'binary' or 'native'
2911
in binary mode, exact contents are written in native mode, the
2912
line endings match the default platform endings.
2913
:param transport: A transport to write to, for building trees on VFS's.
2914
If the transport is readonly or None, "." is opened automatically.
442
# XXX: It's OK to just create them using forward slashes on windows?
2917
if type(shape) not in (list, tuple):
2918
raise AssertionError("Parameter 'shape' should be "
2919
"a list or a tuple. Got %r instead" % (shape,))
2920
# It's OK to just create them using forward slashes on windows.
2921
if transport is None or transport.is_readonly():
2922
transport = _mod_transport.get_transport_from_path(".")
443
2923
for name in shape:
444
assert isinstance(name, basestring)
2924
self.assertIsInstance(name, basestring)
445
2925
if name[-1] == '/':
449
print >>f, "contents of", name
452
def build_tree_contents(self, shape):
453
bzrlib.selftest.build_tree_contents(shape)
455
def failUnlessExists(self, path):
456
"""Fail unless path, which may be abs or relative, exists."""
457
self.failUnless(osutils.lexists(path))
459
def assertFileEqual(self, content, path):
460
"""Fail if path does not contain 'content'."""
461
self.failUnless(osutils.lexists(path))
462
self.assertEqualDiff(content, open(path, 'r').read())
465
class MetaTestLog(TestCase):
466
def test_logging(self):
467
"""Test logs are captured when a test fails."""
468
logging.info('an info message')
469
warning('something looks dodgy...')
470
logging.debug('hello, test is running')
2926
transport.mkdir(urlutils.escape(name[:-1]))
2928
if line_endings == 'binary':
2930
elif line_endings == 'native':
2933
raise errors.BzrError(
2934
'Invalid line ending request %r' % line_endings)
2935
content = "contents of %s%s" % (name.encode('utf-8'), end)
2936
transport.put_bytes_non_atomic(urlutils.escape(name), content)
2938
build_tree_contents = staticmethod(treeshape.build_tree_contents)
2940
def assertInWorkingTree(self, path, root_path='.', tree=None):
2941
"""Assert whether path or paths are in the WorkingTree"""
2943
tree = workingtree.WorkingTree.open(root_path)
2944
if not isinstance(path, basestring):
2946
self.assertInWorkingTree(p, tree=tree)
2948
self.assertIsNot(tree.path2id(path), None,
2949
path+' not in working tree.')
2951
def assertNotInWorkingTree(self, path, root_path='.', tree=None):
2952
"""Assert whether path or paths are not in the WorkingTree"""
2954
tree = workingtree.WorkingTree.open(root_path)
2955
if not isinstance(path, basestring):
2957
self.assertNotInWorkingTree(p,tree=tree)
2959
self.assertIs(tree.path2id(path), None, path+' in working tree.')
2962
class TestCaseWithTransport(TestCaseInTempDir):
2963
"""A test case that provides get_url and get_readonly_url facilities.
2965
These back onto two transport servers, one for readonly access and one for
2968
If no explicit class is provided for readonly access, a
2969
ReadonlyTransportDecorator is used instead which allows the use of non disk
2970
based read write transports.
2972
If an explicit class is provided for readonly access, that server and the
2973
readwrite one must both define get_url() as resolving to os.getcwd().
2977
super(TestCaseWithTransport, self).setUp()
2978
self.__vfs_server = None
2980
def get_vfs_only_server(self):
2981
"""See TestCaseWithMemoryTransport.
2983
This is useful for some tests with specific servers that need
2986
if self.__vfs_server is None:
2987
self.__vfs_server = self.vfs_transport_factory()
2988
self.start_server(self.__vfs_server)
2989
return self.__vfs_server
2991
def make_branch_and_tree(self, relpath, format=None):
2992
"""Create a branch on the transport and a tree locally.
2994
If the transport is not a LocalTransport, the Tree can't be created on
2995
the transport. In that case if the vfs_transport_factory is
2996
LocalURLServer the working tree is created in the local
2997
directory backing the transport, and the returned tree's branch and
2998
repository will also be accessed locally. Otherwise a lightweight
2999
checkout is created and returned.
3001
We do this because we can't physically create a tree in the local
3002
path, with a branch reference to the transport_factory url, and
3003
a branch + repository in the vfs_transport, unless the vfs_transport
3004
namespace is distinct from the local disk - the two branch objects
3005
would collide. While we could construct a tree with its branch object
3006
pointing at the transport_factory transport in memory, reopening it
3007
would behaving unexpectedly, and has in the past caused testing bugs
3008
when we tried to do it that way.
3010
:param format: The BzrDirFormat.
3011
:returns: the WorkingTree.
3013
# TODO: always use the local disk path for the working tree,
3014
# this obviously requires a format that supports branch references
3015
# so check for that by checking bzrdir.BzrDirFormat.get_default_format()
3017
format = self.resolve_format(format=format)
3018
if not format.supports_workingtrees:
3019
b = self.make_branch(relpath+'.branch', format=format)
3020
return b.create_checkout(relpath, lightweight=True)
3021
b = self.make_branch(relpath, format=format)
3023
return b.bzrdir.create_workingtree()
3024
except errors.NotLocalUrl:
3025
# We can only make working trees locally at the moment. If the
3026
# transport can't support them, then we keep the non-disk-backed
3027
# branch and create a local checkout.
3028
if self.vfs_transport_factory is test_server.LocalURLServer:
3029
# the branch is colocated on disk, we cannot create a checkout.
3030
# hopefully callers will expect this.
3031
local_controldir = controldir.ControlDir.open(
3032
self.get_vfs_only_url(relpath))
3033
wt = local_controldir.create_workingtree()
3034
if wt.branch._format != b._format:
3036
# Make sure that assigning to wt._branch fixes wt.branch,
3037
# in case the implementation details of workingtree objects
3039
self.assertIs(b, wt.branch)
3042
return b.create_checkout(relpath, lightweight=True)
3044
def assertIsDirectory(self, relpath, transport):
3045
"""Assert that relpath within transport is a directory.
3047
This may not be possible on all transports; in that case it propagates
3048
a TransportNotPossible.
3051
mode = transport.stat(relpath).st_mode
3052
except errors.NoSuchFile:
3053
self.fail("path %s is not a directory; no such file"
3055
if not stat.S_ISDIR(mode):
3056
self.fail("path %s is not a directory; has mode %#o"
3059
def assertTreesEqual(self, left, right):
3060
"""Check that left and right have the same content and properties."""
3061
# we use a tree delta to check for equality of the content, and we
3062
# manually check for equality of other things such as the parents list.
3063
self.assertEqual(left.get_parent_ids(), right.get_parent_ids())
3064
differences = left.changes_from(right)
3065
self.assertFalse(differences.has_changed(),
3066
"Trees %r and %r are different: %r" % (left, right, differences))
3068
def disable_missing_extensions_warning(self):
3069
"""Some tests expect a precise stderr content.
3071
There is no point in forcing them to duplicate the extension related
3074
config.GlobalStack().set('ignore_missing_extensions', True)
3077
class ChrootedTestCase(TestCaseWithTransport):
3078
"""A support class that provides readonly urls outside the local namespace.
3080
This is done by checking if self.transport_server is a MemoryServer. if it
3081
is then we are chrooted already, if it is not then an HttpServer is used
3084
TODO RBC 20060127: make this an option to TestCaseWithTransport so it can
3085
be used without needed to redo it when a different
3086
subclass is in use ?
3090
from bzrlib.tests import http_server
3091
super(ChrootedTestCase, self).setUp()
3092
if not self.vfs_transport_factory == memory.MemoryServer:
3093
self.transport_readonly_server = http_server.HttpServer
3096
def condition_id_re(pattern):
3097
"""Create a condition filter which performs a re check on a test's id.
3099
:param pattern: A regular expression string.
3100
:return: A callable that returns True if the re matches.
3102
filter_re = re.compile(pattern, 0)
3103
def condition(test):
3105
return filter_re.search(test_id)
3109
def condition_isinstance(klass_or_klass_list):
3110
"""Create a condition filter which returns isinstance(param, klass).
3112
:return: A callable which when called with one parameter obj return the
3113
result of isinstance(obj, klass_or_klass_list).
3116
return isinstance(obj, klass_or_klass_list)
3120
def condition_id_in_list(id_list):
3121
"""Create a condition filter which verify that test's id in a list.
3123
:param id_list: A TestIdList object.
3124
:return: A callable that returns True if the test's id appears in the list.
3126
def condition(test):
3127
return id_list.includes(test.id())
3131
def condition_id_startswith(starts):
3132
"""Create a condition filter verifying that test's id starts with a string.
3134
:param starts: A list of string.
3135
:return: A callable that returns True if the test's id starts with one of
3138
def condition(test):
3139
for start in starts:
3140
if test.id().startswith(start):
3146
def exclude_tests_by_condition(suite, condition):
3147
"""Create a test suite which excludes some tests from suite.
3149
:param suite: The suite to get tests from.
3150
:param condition: A callable whose result evaluates True when called with a
3151
test case which should be excluded from the result.
3152
:return: A suite which contains the tests found in suite that fail
3156
for test in iter_suite_tests(suite):
3157
if not condition(test):
3159
return TestUtil.TestSuite(result)
3162
def filter_suite_by_condition(suite, condition):
3163
"""Create a test suite by filtering another one.
3165
:param suite: The source suite.
3166
:param condition: A callable whose result evaluates True when called with a
3167
test case which should be included in the result.
3168
:return: A suite which contains the tests found in suite that pass
3172
for test in iter_suite_tests(suite):
3175
return TestUtil.TestSuite(result)
474
3178
def filter_suite_by_re(suite, pattern):
475
result = TestUtil.TestSuite()
476
filter_re = re.compile(pattern)
3179
"""Create a test suite by filtering another one.
3181
:param suite: the source suite
3182
:param pattern: pattern that names must match
3183
:returns: the newly created suite
3185
condition = condition_id_re(pattern)
3186
result_suite = filter_suite_by_condition(suite, condition)
3190
def filter_suite_by_id_list(suite, test_id_list):
3191
"""Create a test suite by filtering another one.
3193
:param suite: The source suite.
3194
:param test_id_list: A list of the test ids to keep as strings.
3195
:returns: the newly created suite
3197
condition = condition_id_in_list(test_id_list)
3198
result_suite = filter_suite_by_condition(suite, condition)
3202
def filter_suite_by_id_startswith(suite, start):
3203
"""Create a test suite by filtering another one.
3205
:param suite: The source suite.
3206
:param start: A list of string the test id must start with one of.
3207
:returns: the newly created suite
3209
condition = condition_id_startswith(start)
3210
result_suite = filter_suite_by_condition(suite, condition)
3214
def exclude_tests_by_re(suite, pattern):
3215
"""Create a test suite which excludes some tests from suite.
3217
:param suite: The suite to get tests from.
3218
:param pattern: A regular expression string. Test ids that match this
3219
pattern will be excluded from the result.
3220
:return: A TestSuite that contains all the tests from suite without the
3221
tests that matched pattern. The order of tests is the same as it was in
3224
return exclude_tests_by_condition(suite, condition_id_re(pattern))
3227
def preserve_input(something):
3228
"""A helper for performing test suite transformation chains.
3230
:param something: Anything you want to preserve.
3236
def randomize_suite(suite):
3237
"""Return a new TestSuite with suite's tests in random order.
3239
The tests in the input suite are flattened into a single suite in order to
3240
accomplish this. Any nested TestSuites are removed to provide global
3243
tests = list(iter_suite_tests(suite))
3244
random.shuffle(tests)
3245
return TestUtil.TestSuite(tests)
3248
def split_suite_by_condition(suite, condition):
3249
"""Split a test suite into two by a condition.
3251
:param suite: The suite to split.
3252
:param condition: The condition to match on. Tests that match this
3253
condition are returned in the first test suite, ones that do not match
3254
are in the second suite.
3255
:return: A tuple of two test suites, where the first contains tests from
3256
suite matching the condition, and the second contains the remainder
3257
from suite. The order within each output suite is the same as it was in
477
3262
for test in iter_suite_tests(suite):
478
if filter_re.search(test.id()):
3264
matched.append(test)
3266
did_not_match.append(test)
3267
return TestUtil.TestSuite(matched), TestUtil.TestSuite(did_not_match)
3270
def split_suite_by_re(suite, pattern):
3271
"""Split a test suite into two by a regular expression.
3273
:param suite: The suite to split.
3274
:param pattern: A regular expression string. Test ids that match this
3275
pattern will be in the first test suite returned, and the others in the
3276
second test suite returned.
3277
:return: A tuple of two test suites, where the first contains tests from
3278
suite matching pattern, and the second contains the remainder from
3279
suite. The order within each output suite is the same as it was in
3282
return split_suite_by_condition(suite, condition_id_re(pattern))
483
3285
def run_suite(suite, name='test', verbose=False, pattern=".*",
484
stop_on_failure=False):
485
TestCaseInTempDir._TEST_NAME = name
3286
stop_on_failure=False,
3287
transport=None, lsprof_timed=None, bench_history=None,
3288
matching_tests_first=None,
3291
exclude_pattern=None,
3294
suite_decorators=None,
3296
result_decorators=None,
3298
"""Run a test suite for bzr selftest.
3300
:param runner_class: The class of runner to use. Must support the
3301
constructor arguments passed by run_suite which are more than standard
3303
:return: A boolean indicating success.
3305
TestCase._gather_lsprof_in_benchmarks = lsprof_timed
490
runner = TextTestRunner(stream=sys.stdout,
3310
if runner_class is None:
3311
runner_class = TextTestRunner
3314
runner = runner_class(stream=stream,
3316
verbosity=verbosity,
3317
bench_history=bench_history,
3319
result_decorators=result_decorators,
493
3321
runner.stop_on_failure=stop_on_failure
495
suite = filter_suite_by_re(suite, pattern)
3322
if isinstance(suite, unittest.TestSuite):
3323
# Empty out _tests list of passed suite and populate new TestSuite
3324
suite._tests[:], suite = [], TestSuite(suite)
3325
# built in decorator factories:
3327
random_order(random_seed, runner),
3328
exclude_tests(exclude_pattern),
3330
if matching_tests_first:
3331
decorators.append(tests_first(pattern))
3333
decorators.append(filter_tests(pattern))
3334
if suite_decorators:
3335
decorators.extend(suite_decorators)
3336
# tell the result object how many tests will be running: (except if
3337
# --parallel=fork is being used. Robert said he will provide a better
3338
# progress design later -- vila 20090817)
3339
if fork_decorator not in decorators:
3340
decorators.append(CountingDecorator)
3341
for decorator in decorators:
3342
suite = decorator(suite)
3344
# Done after test suite decoration to allow randomisation etc
3345
# to take effect, though that is of marginal benefit.
3347
stream.write("Listing tests only ...\n")
3348
for t in iter_suite_tests(suite):
3349
stream.write("%s\n" % (t.id()))
496
3351
result = runner.run(suite)
497
# This is still a little bogus,
498
# but only a little. Folk not using our testrunner will
499
# have to delete their temp directories themselves.
500
if result.wasSuccessful():
501
if TestCaseInTempDir.TEST_ROOT is not None:
502
shutil.rmtree(TestCaseInTempDir.TEST_ROOT)
3353
return result.wasStrictlySuccessful()
504
print "Failed tests working directories are in '%s'\n" % TestCaseInTempDir.TEST_ROOT
505
return result.wasSuccessful()
508
def selftest(verbose=False, pattern=".*", stop_on_failure=True):
3355
return result.wasSuccessful()
3358
# A registry where get() returns a suite decorator.
3359
parallel_registry = registry.Registry()
3362
def fork_decorator(suite):
3363
if getattr(os, "fork", None) is None:
3364
raise errors.BzrCommandError("platform does not support fork,"
3365
" try --parallel=subprocess instead.")
3366
concurrency = osutils.local_concurrency()
3367
if concurrency == 1:
3369
from testtools import ConcurrentTestSuite
3370
return ConcurrentTestSuite(suite, fork_for_tests)
3371
parallel_registry.register('fork', fork_decorator)
3374
def subprocess_decorator(suite):
3375
concurrency = osutils.local_concurrency()
3376
if concurrency == 1:
3378
from testtools import ConcurrentTestSuite
3379
return ConcurrentTestSuite(suite, reinvoke_for_tests)
3380
parallel_registry.register('subprocess', subprocess_decorator)
3383
def exclude_tests(exclude_pattern):
3384
"""Return a test suite decorator that excludes tests."""
3385
if exclude_pattern is None:
3386
return identity_decorator
3387
def decorator(suite):
3388
return ExcludeDecorator(suite, exclude_pattern)
3392
def filter_tests(pattern):
3394
return identity_decorator
3395
def decorator(suite):
3396
return FilterTestsDecorator(suite, pattern)
3400
def random_order(random_seed, runner):
3401
"""Return a test suite decorator factory for randomising tests order.
3403
:param random_seed: now, a string which casts to a long, or a long.
3404
:param runner: A test runner with a stream attribute to report on.
3406
if random_seed is None:
3407
return identity_decorator
3408
def decorator(suite):
3409
return RandomDecorator(suite, random_seed, runner.stream)
3413
def tests_first(pattern):
3415
return identity_decorator
3416
def decorator(suite):
3417
return TestFirstDecorator(suite, pattern)
3421
def identity_decorator(suite):
3426
class TestDecorator(TestUtil.TestSuite):
3427
"""A decorator for TestCase/TestSuite objects.
3429
Contains rather than flattening suite passed on construction
3432
def __init__(self, suite=None):
3433
super(TestDecorator, self).__init__()
3434
if suite is not None:
3437
# Don't need subclass run method with suite emptying
3438
run = unittest.TestSuite.run
3441
class CountingDecorator(TestDecorator):
3442
"""A decorator which calls result.progress(self.countTestCases)."""
3444
def run(self, result):
3445
progress_method = getattr(result, 'progress', None)
3446
if callable(progress_method):
3447
progress_method(self.countTestCases(), SUBUNIT_SEEK_SET)
3448
return super(CountingDecorator, self).run(result)
3451
class ExcludeDecorator(TestDecorator):
3452
"""A decorator which excludes test matching an exclude pattern."""
3454
def __init__(self, suite, exclude_pattern):
3455
super(ExcludeDecorator, self).__init__(
3456
exclude_tests_by_re(suite, exclude_pattern))
3459
class FilterTestsDecorator(TestDecorator):
3460
"""A decorator which filters tests to those matching a pattern."""
3462
def __init__(self, suite, pattern):
3463
super(FilterTestsDecorator, self).__init__(
3464
filter_suite_by_re(suite, pattern))
3467
class RandomDecorator(TestDecorator):
3468
"""A decorator which randomises the order of its tests."""
3470
def __init__(self, suite, random_seed, stream):
3471
random_seed = self.actual_seed(random_seed)
3472
stream.write("Randomizing test order using seed %s\n\n" %
3474
# Initialise the random number generator.
3475
random.seed(random_seed)
3476
super(RandomDecorator, self).__init__(randomize_suite(suite))
3479
def actual_seed(seed):
3481
# We convert the seed to a long to make it reuseable across
3482
# invocations (because the user can reenter it).
3483
return long(time.time())
3485
# Convert the seed to a long if we can
3488
except (TypeError, ValueError):
3493
class TestFirstDecorator(TestDecorator):
3494
"""A decorator which moves named tests to the front."""
3496
def __init__(self, suite, pattern):
3497
super(TestFirstDecorator, self).__init__()
3498
self.addTests(split_suite_by_re(suite, pattern))
3501
def partition_tests(suite, count):
3502
"""Partition suite into count lists of tests."""
3503
# This just assigns tests in a round-robin fashion. On one hand this
3504
# splits up blocks of related tests that might run faster if they shared
3505
# resources, but on the other it avoids assigning blocks of slow tests to
3506
# just one partition. So the slowest partition shouldn't be much slower
3508
partitions = [list() for i in range(count)]
3509
tests = iter_suite_tests(suite)
3510
for partition, test in itertools.izip(itertools.cycle(partitions), tests):
3511
partition.append(test)
3515
def workaround_zealous_crypto_random():
3516
"""Crypto.Random want to help us being secure, but we don't care here.
3518
This workaround some test failure related to the sftp server. Once paramiko
3519
stop using the controversial API in Crypto.Random, we may get rid of it.
3522
from Crypto.Random import atfork
3528
def fork_for_tests(suite):
3529
"""Take suite and start up one runner per CPU by forking()
3531
:return: An iterable of TestCase-like objects which can each have
3532
run(result) called on them to feed tests to result.
3534
concurrency = osutils.local_concurrency()
3536
from subunit import ProtocolTestCase
3537
from subunit.test_results import AutoTimingTestResultDecorator
3538
class TestInOtherProcess(ProtocolTestCase):
3539
# Should be in subunit, I think. RBC.
3540
def __init__(self, stream, pid):
3541
ProtocolTestCase.__init__(self, stream)
3544
def run(self, result):
3546
ProtocolTestCase.run(self, result)
3548
pid, status = os.waitpid(self.pid, 0)
3549
# GZ 2011-10-18: If status is nonzero, should report to the result
3550
# that something went wrong.
3552
test_blocks = partition_tests(suite, concurrency)
3553
# Clear the tests from the original suite so it doesn't keep them alive
3554
suite._tests[:] = []
3555
for process_tests in test_blocks:
3556
process_suite = TestUtil.TestSuite(process_tests)
3557
# Also clear each split list so new suite has only reference
3558
process_tests[:] = []
3559
c2pread, c2pwrite = os.pipe()
3563
stream = os.fdopen(c2pwrite, 'wb', 1)
3564
workaround_zealous_crypto_random()
3566
# Leave stderr and stdout open so we can see test noise
3567
# Close stdin so that the child goes away if it decides to
3568
# read from stdin (otherwise its a roulette to see what
3569
# child actually gets keystrokes for pdb etc).
3571
subunit_result = AutoTimingTestResultDecorator(
3572
SubUnitBzrProtocolClient(stream))
3573
process_suite.run(subunit_result)
3575
# Try and report traceback on stream, but exit with error even
3576
# if stream couldn't be created or something else goes wrong.
3577
# The traceback is formatted to a string and written in one go
3578
# to avoid interleaving lines from multiple failing children.
3580
stream.write(traceback.format_exc())
3586
stream = os.fdopen(c2pread, 'rb', 1)
3587
test = TestInOtherProcess(stream, pid)
3592
def reinvoke_for_tests(suite):
3593
"""Take suite and start up one runner per CPU using subprocess().
3595
:return: An iterable of TestCase-like objects which can each have
3596
run(result) called on them to feed tests to result.
3598
concurrency = osutils.local_concurrency()
3600
from subunit import ProtocolTestCase
3601
class TestInSubprocess(ProtocolTestCase):
3602
def __init__(self, process, name):
3603
ProtocolTestCase.__init__(self, process.stdout)
3604
self.process = process
3605
self.process.stdin.close()
3608
def run(self, result):
3610
ProtocolTestCase.run(self, result)
3613
os.unlink(self.name)
3614
# print "pid %d finished" % finished_process
3615
test_blocks = partition_tests(suite, concurrency)
3616
for process_tests in test_blocks:
3617
# ugly; currently reimplement rather than reuses TestCase methods.
3618
bzr_path = os.path.dirname(os.path.dirname(bzrlib.__file__))+'/bzr'
3619
if not os.path.isfile(bzr_path):
3620
# We are probably installed. Assume sys.argv is the right file
3621
bzr_path = sys.argv[0]
3622
bzr_path = [bzr_path]
3623
if sys.platform == "win32":
3624
# if we're on windows, we can't execute the bzr script directly
3625
bzr_path = [sys.executable] + bzr_path
3626
fd, test_list_file_name = tempfile.mkstemp()
3627
test_list_file = os.fdopen(fd, 'wb', 1)
3628
for test in process_tests:
3629
test_list_file.write(test.id() + '\n')
3630
test_list_file.close()
3632
argv = bzr_path + ['selftest', '--load-list', test_list_file_name,
3634
if '--no-plugins' in sys.argv:
3635
argv.append('--no-plugins')
3636
# stderr=subprocess.STDOUT would be ideal, but until we prevent
3637
# noise on stderr it can interrupt the subunit protocol.
3638
process = subprocess.Popen(argv, stdin=subprocess.PIPE,
3639
stdout=subprocess.PIPE,
3640
stderr=subprocess.PIPE,
3642
test = TestInSubprocess(process, test_list_file_name)
3645
os.unlink(test_list_file_name)
3650
class ProfileResult(testtools.ExtendedToOriginalDecorator):
3651
"""Generate profiling data for all activity between start and success.
3653
The profile data is appended to the test's _benchcalls attribute and can
3654
be accessed by the forwarded-to TestResult.
3656
While it might be cleaner do accumulate this in stopTest, addSuccess is
3657
where our existing output support for lsprof is, and this class aims to
3658
fit in with that: while it could be moved it's not necessary to accomplish
3659
test profiling, nor would it be dramatically cleaner.
3662
def startTest(self, test):
3663
self.profiler = bzrlib.lsprof.BzrProfiler()
3664
# Prevent deadlocks in tests that use lsprof: those tests will
3666
bzrlib.lsprof.BzrProfiler.profiler_block = 0
3667
self.profiler.start()
3668
testtools.ExtendedToOriginalDecorator.startTest(self, test)
3670
def addSuccess(self, test):
3671
stats = self.profiler.stop()
3673
calls = test._benchcalls
3674
except AttributeError:
3675
test._benchcalls = []
3676
calls = test._benchcalls
3677
calls.append(((test.id(), "", ""), stats))
3678
testtools.ExtendedToOriginalDecorator.addSuccess(self, test)
3680
def stopTest(self, test):
3681
testtools.ExtendedToOriginalDecorator.stopTest(self, test)
3682
self.profiler = None
3685
# Controlled by "bzr selftest -E=..." option
3686
# Currently supported:
3687
# -Eallow_debug Will no longer clear debug.debug_flags() so it
3688
# preserves any flags supplied at the command line.
3689
# -Edisable_lock_checks Turns errors in mismatched locks into simple prints
3690
# rather than failing tests. And no longer raise
3691
# LockContention when fctnl locks are not being used
3692
# with proper exclusion rules.
3693
# -Ethreads Will display thread ident at creation/join time to
3694
# help track thread leaks
3695
# -Euncollected_cases Display the identity of any test cases that weren't
3696
# deallocated after being completed.
3697
# -Econfig_stats Will collect statistics using addDetail
3698
selftest_debug_flags = set()
3701
def selftest(verbose=False, pattern=".*", stop_on_failure=True,
3703
test_suite_factory=None,
3706
matching_tests_first=None,
3709
exclude_pattern=None,
3715
suite_decorators=None,
509
3719
"""Run the whole test suite under the enhanced runner"""
510
return run_suite(test_suite(), 'testbzr', verbose=verbose, pattern=pattern,
511
stop_on_failure=stop_on_failure)
515
"""Build and return TestSuite for the whole program."""
516
import bzrlib.store, bzrlib.inventory, bzrlib.branch
517
import bzrlib.osutils, bzrlib.merge3, bzrlib.plugin
518
from doctest import DocTestSuite
520
global MODULES_TO_TEST, MODULES_TO_DOCTEST
523
['bzrlib.selftest.MetaTestLog',
524
'bzrlib.selftest.testgpg',
525
'bzrlib.selftest.testidentitymap',
526
'bzrlib.selftest.testinv',
527
'bzrlib.selftest.test_ancestry',
528
'bzrlib.selftest.test_commit',
529
'bzrlib.selftest.test_commit_merge',
530
'bzrlib.selftest.testconfig',
531
'bzrlib.selftest.versioning',
532
'bzrlib.selftest.testmerge3',
533
'bzrlib.selftest.testmerge',
534
'bzrlib.selftest.testhashcache',
535
'bzrlib.selftest.teststatus',
536
'bzrlib.selftest.testlog',
537
'bzrlib.selftest.testrevisionnamespaces',
538
'bzrlib.selftest.testbranch',
539
'bzrlib.selftest.testrevision',
540
'bzrlib.selftest.test_revision_info',
541
'bzrlib.selftest.test_merge_core',
542
'bzrlib.selftest.test_smart_add',
543
'bzrlib.selftest.test_bad_files',
544
'bzrlib.selftest.testdiff',
545
'bzrlib.selftest.test_parent',
546
'bzrlib.selftest.test_xml',
547
'bzrlib.selftest.test_weave',
548
'bzrlib.selftest.testfetch',
549
'bzrlib.selftest.whitebox',
550
'bzrlib.selftest.teststore',
551
'bzrlib.selftest.blackbox',
552
'bzrlib.selftest.testsampler',
553
'bzrlib.selftest.testtransactions',
554
'bzrlib.selftest.testtransport',
555
'bzrlib.selftest.testgraph',
556
'bzrlib.selftest.testworkingtree',
557
'bzrlib.selftest.test_upgrade',
558
'bzrlib.selftest.test_conflicts',
559
'bzrlib.selftest.testtestament',
560
'bzrlib.selftest.testannotate',
561
'bzrlib.selftest.testrevprops',
562
'bzrlib.selftest.testoptions',
563
'bzrlib.selftest.testhttp',
564
'bzrlib.selftest.testnonascii',
567
for m in (bzrlib.store, bzrlib.inventory, bzrlib.branch,
568
bzrlib.osutils, bzrlib.commands, bzrlib.merge3,
571
if m not in MODULES_TO_DOCTEST:
572
MODULES_TO_DOCTEST.append(m)
574
TestCase.BZRPATH = os.path.join(os.path.realpath(os.path.dirname(bzrlib.__path__[0])), 'bzr')
575
print '%-30s %s' % ('bzr binary', TestCase.BZRPATH)
578
suite.addTest(TestLoader().loadTestsFromNames(testmod_names))
579
for m in MODULES_TO_TEST:
580
suite.addTest(TestLoader().loadTestsFromModule(m))
581
for m in (MODULES_TO_DOCTEST):
582
suite.addTest(DocTestSuite(m))
583
for p in bzrlib.plugin.all_plugins:
584
if hasattr(p, 'test_suite'):
585
suite.addTest(p.test_suite())
3720
# XXX: Very ugly way to do this...
3721
# Disable warning about old formats because we don't want it to disturb
3722
# any blackbox tests.
3723
from bzrlib import repository
3724
repository._deprecation_warning_done = True
3726
global default_transport
3727
if transport is None:
3728
transport = default_transport
3729
old_transport = default_transport
3730
default_transport = transport
3731
global selftest_debug_flags
3732
old_debug_flags = selftest_debug_flags
3733
if debug_flags is not None:
3734
selftest_debug_flags = set(debug_flags)
3736
if load_list is None:
3739
keep_only = load_test_id_list(load_list)
3741
starting_with = [test_prefix_alias_registry.resolve_alias(start)
3742
for start in starting_with]
3743
if test_suite_factory is None:
3744
# Reduce loading time by loading modules based on the starting_with
3746
suite = test_suite(keep_only, starting_with)
3748
suite = test_suite_factory()
3750
# But always filter as requested.
3751
suite = filter_suite_by_id_startswith(suite, starting_with)
3752
result_decorators = []
3754
result_decorators.append(ProfileResult)
3755
return run_suite(suite, 'testbzr', verbose=verbose, pattern=pattern,
3756
stop_on_failure=stop_on_failure,
3757
transport=transport,
3758
lsprof_timed=lsprof_timed,
3759
bench_history=bench_history,
3760
matching_tests_first=matching_tests_first,
3761
list_only=list_only,
3762
random_seed=random_seed,
3763
exclude_pattern=exclude_pattern,
3765
runner_class=runner_class,
3766
suite_decorators=suite_decorators,
3768
result_decorators=result_decorators,
3771
default_transport = old_transport
3772
selftest_debug_flags = old_debug_flags
3775
def load_test_id_list(file_name):
3776
"""Load a test id list from a text file.
3778
The format is one test id by line. No special care is taken to impose
3779
strict rules, these test ids are used to filter the test suite so a test id
3780
that do not match an existing test will do no harm. This allows user to add
3781
comments, leave blank lines, etc.
3785
ftest = open(file_name, 'rt')
3787
if e.errno != errno.ENOENT:
3790
raise errors.NoSuchFile(file_name)
3792
for test_name in ftest.readlines():
3793
test_list.append(test_name.strip())
3798
def suite_matches_id_list(test_suite, id_list):
3799
"""Warns about tests not appearing or appearing more than once.
3801
:param test_suite: A TestSuite object.
3802
:param test_id_list: The list of test ids that should be found in
3805
:return: (absents, duplicates) absents is a list containing the test found
3806
in id_list but not in test_suite, duplicates is a list containing the
3807
tests found multiple times in test_suite.
3809
When using a prefined test id list, it may occurs that some tests do not
3810
exist anymore or that some tests use the same id. This function warns the
3811
tester about potential problems in his workflow (test lists are volatile)
3812
or in the test suite itself (using the same id for several tests does not
3813
help to localize defects).
3815
# Build a dict counting id occurrences
3817
for test in iter_suite_tests(test_suite):
3819
tests[id] = tests.get(id, 0) + 1
3824
occurs = tests.get(id, 0)
3826
not_found.append(id)
3828
duplicates.append(id)
3830
return not_found, duplicates
3833
class TestIdList(object):
3834
"""Test id list to filter a test suite.
3836
Relying on the assumption that test ids are built as:
3837
<module>[.<class>.<method>][(<param>+)], <module> being in python dotted
3838
notation, this class offers methods to :
3839
- avoid building a test suite for modules not refered to in the test list,
3840
- keep only the tests listed from the module test suite.
3843
def __init__(self, test_id_list):
3844
# When a test suite needs to be filtered against us we compare test ids
3845
# for equality, so a simple dict offers a quick and simple solution.
3846
self.tests = dict().fromkeys(test_id_list, True)
3848
# While unittest.TestCase have ids like:
3849
# <module>.<class>.<method>[(<param+)],
3850
# doctest.DocTestCase can have ids like:
3853
# <module>.<function>
3854
# <module>.<class>.<method>
3856
# Since we can't predict a test class from its name only, we settle on
3857
# a simple constraint: a test id always begins with its module name.
3860
for test_id in test_id_list:
3861
parts = test_id.split('.')
3862
mod_name = parts.pop(0)
3863
modules[mod_name] = True
3865
mod_name += '.' + part
3866
modules[mod_name] = True
3867
self.modules = modules
3869
def refers_to(self, module_name):
3870
"""Is there tests for the module or one of its sub modules."""
3871
return self.modules.has_key(module_name)
3873
def includes(self, test_id):
3874
return self.tests.has_key(test_id)
3877
class TestPrefixAliasRegistry(registry.Registry):
3878
"""A registry for test prefix aliases.
3880
This helps implement shorcuts for the --starting-with selftest
3881
option. Overriding existing prefixes is not allowed but not fatal (a
3882
warning will be emitted).
3885
def register(self, key, obj, help=None, info=None,
3886
override_existing=False):
3887
"""See Registry.register.
3889
Trying to override an existing alias causes a warning to be emitted,
3890
not a fatal execption.
3893
super(TestPrefixAliasRegistry, self).register(
3894
key, obj, help=help, info=info, override_existing=False)
3896
actual = self.get(key)
3898
'Test prefix alias %s is already used for %s, ignoring %s'
3899
% (key, actual, obj))
3901
def resolve_alias(self, id_start):
3902
"""Replace the alias by the prefix in the given string.
3904
Using an unknown prefix is an error to help catching typos.
3906
parts = id_start.split('.')
3908
parts[0] = self.get(parts[0])
3910
raise errors.BzrCommandError(
3911
'%s is not a known test prefix alias' % parts[0])
3912
return '.'.join(parts)
3915
test_prefix_alias_registry = TestPrefixAliasRegistry()
3916
"""Registry of test prefix aliases."""
3919
# This alias allows to detect typos ('bzrlin.') by making all valid test ids
3920
# appear prefixed ('bzrlib.' is "replaced" by 'bzrlib.').
3921
test_prefix_alias_registry.register('bzrlib', 'bzrlib')
3923
# Obvious highest levels prefixes, feel free to add your own via a plugin
3924
test_prefix_alias_registry.register('bd', 'bzrlib.doc')
3925
test_prefix_alias_registry.register('bu', 'bzrlib.utils')
3926
test_prefix_alias_registry.register('bt', 'bzrlib.tests')
3927
test_prefix_alias_registry.register('bb', 'bzrlib.tests.blackbox')
3928
test_prefix_alias_registry.register('bp', 'bzrlib.plugins')
3931
def _test_suite_testmod_names():
3932
"""Return the standard list of test module names to test."""
3935
'bzrlib.tests.blackbox',
3936
'bzrlib.tests.commands',
3937
'bzrlib.tests.per_branch',
3938
'bzrlib.tests.per_bzrdir',
3939
'bzrlib.tests.per_controldir',
3940
'bzrlib.tests.per_controldir_colo',
3941
'bzrlib.tests.per_foreign_vcs',
3942
'bzrlib.tests.per_interrepository',
3943
'bzrlib.tests.per_intertree',
3944
'bzrlib.tests.per_inventory',
3945
'bzrlib.tests.per_interbranch',
3946
'bzrlib.tests.per_lock',
3947
'bzrlib.tests.per_merger',
3948
'bzrlib.tests.per_transport',
3949
'bzrlib.tests.per_tree',
3950
'bzrlib.tests.per_pack_repository',
3951
'bzrlib.tests.per_repository',
3952
'bzrlib.tests.per_repository_chk',
3953
'bzrlib.tests.per_repository_reference',
3954
'bzrlib.tests.per_repository_vf',
3955
'bzrlib.tests.per_uifactory',
3956
'bzrlib.tests.per_versionedfile',
3957
'bzrlib.tests.per_workingtree',
3958
'bzrlib.tests.test__annotator',
3959
'bzrlib.tests.test__bencode',
3960
'bzrlib.tests.test__btree_serializer',
3961
'bzrlib.tests.test__chk_map',
3962
'bzrlib.tests.test__dirstate_helpers',
3963
'bzrlib.tests.test__groupcompress',
3964
'bzrlib.tests.test__known_graph',
3965
'bzrlib.tests.test__rio',
3966
'bzrlib.tests.test__simple_set',
3967
'bzrlib.tests.test__static_tuple',
3968
'bzrlib.tests.test__walkdirs_win32',
3969
'bzrlib.tests.test_ancestry',
3970
'bzrlib.tests.test_annotate',
3971
'bzrlib.tests.test_api',
3972
'bzrlib.tests.test_atomicfile',
3973
'bzrlib.tests.test_bad_files',
3974
'bzrlib.tests.test_bisect_multi',
3975
'bzrlib.tests.test_branch',
3976
'bzrlib.tests.test_branchbuilder',
3977
'bzrlib.tests.test_btree_index',
3978
'bzrlib.tests.test_bugtracker',
3979
'bzrlib.tests.test_bundle',
3980
'bzrlib.tests.test_bzrdir',
3981
'bzrlib.tests.test__chunks_to_lines',
3982
'bzrlib.tests.test_cache_utf8',
3983
'bzrlib.tests.test_chk_map',
3984
'bzrlib.tests.test_chk_serializer',
3985
'bzrlib.tests.test_chunk_writer',
3986
'bzrlib.tests.test_clean_tree',
3987
'bzrlib.tests.test_cleanup',
3988
'bzrlib.tests.test_cmdline',
3989
'bzrlib.tests.test_commands',
3990
'bzrlib.tests.test_commit',
3991
'bzrlib.tests.test_commit_merge',
3992
'bzrlib.tests.test_config',
3993
'bzrlib.tests.test_conflicts',
3994
'bzrlib.tests.test_controldir',
3995
'bzrlib.tests.test_counted_lock',
3996
'bzrlib.tests.test_crash',
3997
'bzrlib.tests.test_decorators',
3998
'bzrlib.tests.test_delta',
3999
'bzrlib.tests.test_debug',
4000
'bzrlib.tests.test_diff',
4001
'bzrlib.tests.test_directory_service',
4002
'bzrlib.tests.test_dirstate',
4003
'bzrlib.tests.test_email_message',
4004
'bzrlib.tests.test_eol_filters',
4005
'bzrlib.tests.test_errors',
4006
'bzrlib.tests.test_estimate_compressed_size',
4007
'bzrlib.tests.test_export',
4008
'bzrlib.tests.test_export_pot',
4009
'bzrlib.tests.test_extract',
4010
'bzrlib.tests.test_features',
4011
'bzrlib.tests.test_fetch',
4012
'bzrlib.tests.test_fixtures',
4013
'bzrlib.tests.test_fifo_cache',
4014
'bzrlib.tests.test_filters',
4015
'bzrlib.tests.test_filter_tree',
4016
'bzrlib.tests.test_ftp_transport',
4017
'bzrlib.tests.test_foreign',
4018
'bzrlib.tests.test_generate_docs',
4019
'bzrlib.tests.test_generate_ids',
4020
'bzrlib.tests.test_globbing',
4021
'bzrlib.tests.test_gpg',
4022
'bzrlib.tests.test_graph',
4023
'bzrlib.tests.test_groupcompress',
4024
'bzrlib.tests.test_hashcache',
4025
'bzrlib.tests.test_help',
4026
'bzrlib.tests.test_hooks',
4027
'bzrlib.tests.test_http',
4028
'bzrlib.tests.test_http_response',
4029
'bzrlib.tests.test_https_ca_bundle',
4030
'bzrlib.tests.test_https_urllib',
4031
'bzrlib.tests.test_i18n',
4032
'bzrlib.tests.test_identitymap',
4033
'bzrlib.tests.test_ignores',
4034
'bzrlib.tests.test_index',
4035
'bzrlib.tests.test_import_tariff',
4036
'bzrlib.tests.test_info',
4037
'bzrlib.tests.test_inv',
4038
'bzrlib.tests.test_inventory_delta',
4039
'bzrlib.tests.test_knit',
4040
'bzrlib.tests.test_lazy_import',
4041
'bzrlib.tests.test_lazy_regex',
4042
'bzrlib.tests.test_library_state',
4043
'bzrlib.tests.test_lock',
4044
'bzrlib.tests.test_lockable_files',
4045
'bzrlib.tests.test_lockdir',
4046
'bzrlib.tests.test_log',
4047
'bzrlib.tests.test_lru_cache',
4048
'bzrlib.tests.test_lsprof',
4049
'bzrlib.tests.test_mail_client',
4050
'bzrlib.tests.test_matchers',
4051
'bzrlib.tests.test_memorytree',
4052
'bzrlib.tests.test_merge',
4053
'bzrlib.tests.test_merge3',
4054
'bzrlib.tests.test_merge_core',
4055
'bzrlib.tests.test_merge_directive',
4056
'bzrlib.tests.test_mergetools',
4057
'bzrlib.tests.test_missing',
4058
'bzrlib.tests.test_msgeditor',
4059
'bzrlib.tests.test_multiparent',
4060
'bzrlib.tests.test_mutabletree',
4061
'bzrlib.tests.test_nonascii',
4062
'bzrlib.tests.test_options',
4063
'bzrlib.tests.test_osutils',
4064
'bzrlib.tests.test_osutils_encodings',
4065
'bzrlib.tests.test_pack',
4066
'bzrlib.tests.test_patch',
4067
'bzrlib.tests.test_patches',
4068
'bzrlib.tests.test_permissions',
4069
'bzrlib.tests.test_plugins',
4070
'bzrlib.tests.test_progress',
4071
'bzrlib.tests.test_pyutils',
4072
'bzrlib.tests.test_read_bundle',
4073
'bzrlib.tests.test_reconcile',
4074
'bzrlib.tests.test_reconfigure',
4075
'bzrlib.tests.test_registry',
4076
'bzrlib.tests.test_remote',
4077
'bzrlib.tests.test_rename_map',
4078
'bzrlib.tests.test_repository',
4079
'bzrlib.tests.test_revert',
4080
'bzrlib.tests.test_revision',
4081
'bzrlib.tests.test_revisionspec',
4082
'bzrlib.tests.test_revisiontree',
4083
'bzrlib.tests.test_rio',
4084
'bzrlib.tests.test_rules',
4085
'bzrlib.tests.test_url_policy_open',
4086
'bzrlib.tests.test_sampler',
4087
'bzrlib.tests.test_scenarios',
4088
'bzrlib.tests.test_script',
4089
'bzrlib.tests.test_selftest',
4090
'bzrlib.tests.test_serializer',
4091
'bzrlib.tests.test_setup',
4092
'bzrlib.tests.test_sftp_transport',
4093
'bzrlib.tests.test_shelf',
4094
'bzrlib.tests.test_shelf_ui',
4095
'bzrlib.tests.test_smart',
4096
'bzrlib.tests.test_smart_add',
4097
'bzrlib.tests.test_smart_request',
4098
'bzrlib.tests.test_smart_signals',
4099
'bzrlib.tests.test_smart_transport',
4100
'bzrlib.tests.test_smtp_connection',
4101
'bzrlib.tests.test_source',
4102
'bzrlib.tests.test_ssh_transport',
4103
'bzrlib.tests.test_status',
4104
'bzrlib.tests.test_store',
4105
'bzrlib.tests.test_strace',
4106
'bzrlib.tests.test_subsume',
4107
'bzrlib.tests.test_switch',
4108
'bzrlib.tests.test_symbol_versioning',
4109
'bzrlib.tests.test_tag',
4110
'bzrlib.tests.test_test_server',
4111
'bzrlib.tests.test_testament',
4112
'bzrlib.tests.test_textfile',
4113
'bzrlib.tests.test_textmerge',
4114
'bzrlib.tests.test_cethread',
4115
'bzrlib.tests.test_timestamp',
4116
'bzrlib.tests.test_trace',
4117
'bzrlib.tests.test_transactions',
4118
'bzrlib.tests.test_transform',
4119
'bzrlib.tests.test_transport',
4120
'bzrlib.tests.test_transport_log',
4121
'bzrlib.tests.test_tree',
4122
'bzrlib.tests.test_treebuilder',
4123
'bzrlib.tests.test_treeshape',
4124
'bzrlib.tests.test_tsort',
4125
'bzrlib.tests.test_tuned_gzip',
4126
'bzrlib.tests.test_ui',
4127
'bzrlib.tests.test_uncommit',
4128
'bzrlib.tests.test_upgrade',
4129
'bzrlib.tests.test_upgrade_stacked',
4130
'bzrlib.tests.test_urlutils',
4131
'bzrlib.tests.test_utextwrap',
4132
'bzrlib.tests.test_version',
4133
'bzrlib.tests.test_version_info',
4134
'bzrlib.tests.test_versionedfile',
4135
'bzrlib.tests.test_vf_search',
4136
'bzrlib.tests.test_weave',
4137
'bzrlib.tests.test_whitebox',
4138
'bzrlib.tests.test_win32utils',
4139
'bzrlib.tests.test_workingtree',
4140
'bzrlib.tests.test_workingtree_4',
4141
'bzrlib.tests.test_wsgi',
4142
'bzrlib.tests.test_xml',
4146
def _test_suite_modules_to_doctest():
4147
"""Return the list of modules to doctest."""
4149
# GZ 2009-03-31: No docstrings with -OO so there's nothing to doctest
4153
'bzrlib.branchbuilder',
4154
'bzrlib.decorators',
4156
'bzrlib.iterablefile',
4161
'bzrlib.symbol_versioning',
4163
'bzrlib.tests.fixtures',
4165
'bzrlib.transport.http',
4166
'bzrlib.version_info_formats.format_custom',
4170
def test_suite(keep_only=None, starting_with=None):
4171
"""Build and return TestSuite for the whole of bzrlib.
4173
:param keep_only: A list of test ids limiting the suite returned.
4175
:param starting_with: An id limiting the suite returned to the tests
4178
This function can be replaced if you need to change the default test
4179
suite on a global basis, but it is not encouraged.
4182
loader = TestUtil.TestLoader()
4184
if keep_only is not None:
4185
id_filter = TestIdList(keep_only)
4187
# We take precedence over keep_only because *at loading time* using
4188
# both options means we will load less tests for the same final result.
4189
def interesting_module(name):
4190
for start in starting_with:
4192
# Either the module name starts with the specified string
4193
name.startswith(start)
4194
# or it may contain tests starting with the specified string
4195
or start.startswith(name)
4199
loader = TestUtil.FilteredByModuleTestLoader(interesting_module)
4201
elif keep_only is not None:
4202
loader = TestUtil.FilteredByModuleTestLoader(id_filter.refers_to)
4203
def interesting_module(name):
4204
return id_filter.refers_to(name)
4207
loader = TestUtil.TestLoader()
4208
def interesting_module(name):
4209
# No filtering, all modules are interesting
4212
suite = loader.suiteClass()
4214
# modules building their suite with loadTestsFromModuleNames
4215
suite.addTest(loader.loadTestsFromModuleNames(_test_suite_testmod_names()))
4217
for mod in _test_suite_modules_to_doctest():
4218
if not interesting_module(mod):
4219
# No tests to keep here, move along
4222
# note that this really does mean "report only" -- doctest
4223
# still runs the rest of the examples
4224
doc_suite = IsolatedDocTestSuite(
4225
mod, optionflags=doctest.REPORT_ONLY_FIRST_FAILURE)
4226
except ValueError, e:
4227
print '**failed to get doctest for: %s\n%s' % (mod, e)
4229
if len(doc_suite._tests) == 0:
4230
raise errors.BzrError("no doctests found in %s" % (mod,))
4231
suite.addTest(doc_suite)
4233
default_encoding = sys.getdefaultencoding()
4234
for name, plugin in _mod_plugin.plugins().items():
4235
if not interesting_module(plugin.module.__name__):
4237
plugin_suite = plugin.test_suite()
4238
# We used to catch ImportError here and turn it into just a warning,
4239
# but really if you don't have --no-plugins this should be a failure.
4240
# mbp 20080213 - see http://bugs.launchpad.net/bugs/189771
4241
if plugin_suite is None:
4242
plugin_suite = plugin.load_plugin_tests(loader)
4243
if plugin_suite is not None:
4244
suite.addTest(plugin_suite)
4245
if default_encoding != sys.getdefaultencoding():
4247
'Plugin "%s" tried to reset default encoding to: %s', name,
4248
sys.getdefaultencoding())
4250
sys.setdefaultencoding(default_encoding)
4252
if keep_only is not None:
4253
# Now that the referred modules have loaded their tests, keep only the
4255
suite = filter_suite_by_id_list(suite, id_filter)
4256
# Do some sanity checks on the id_list filtering
4257
not_found, duplicates = suite_matches_id_list(suite, keep_only)
4259
# The tester has used both keep_only and starting_with, so he is
4260
# already aware that some tests are excluded from the list, there
4261
# is no need to tell him which.
4264
# Some tests mentioned in the list are not in the test suite. The
4265
# list may be out of date, report to the tester.
4266
for id in not_found:
4267
trace.warning('"%s" not found in the test suite', id)
4268
for id in duplicates:
4269
trace.warning('"%s" is used as an id by several tests', id)
4274
def multiply_scenarios(*scenarios):
4275
"""Multiply two or more iterables of scenarios.
4277
It is safe to pass scenario generators or iterators.
4279
:returns: A list of compound scenarios: the cross-product of all
4280
scenarios, with the names concatenated and the parameters
4283
return reduce(_multiply_two_scenarios, map(list, scenarios))
4286
def _multiply_two_scenarios(scenarios_left, scenarios_right):
4287
"""Multiply two sets of scenarios.
4289
:returns: the cartesian product of the two sets of scenarios, that is
4290
a scenario for every possible combination of a left scenario and a
4294
('%s,%s' % (left_name, right_name),
4295
dict(left_dict.items() + right_dict.items()))
4296
for left_name, left_dict in scenarios_left
4297
for right_name, right_dict in scenarios_right]
4300
def multiply_tests(tests, scenarios, result):
4301
"""Multiply tests_list by scenarios into result.
4303
This is the core workhorse for test parameterisation.
4305
Typically the load_tests() method for a per-implementation test suite will
4306
call multiply_tests and return the result.
4308
:param tests: The tests to parameterise.
4309
:param scenarios: The scenarios to apply: pairs of (scenario_name,
4310
scenario_param_dict).
4311
:param result: A TestSuite to add created tests to.
4313
This returns the passed in result TestSuite with the cross product of all
4314
the tests repeated once for each scenario. Each test is adapted by adding
4315
the scenario name at the end of its id(), and updating the test object's
4316
__dict__ with the scenario_param_dict.
4318
>>> import bzrlib.tests.test_sampler
4319
>>> r = multiply_tests(
4320
... bzrlib.tests.test_sampler.DemoTest('test_nothing'),
4321
... [('one', dict(param=1)),
4322
... ('two', dict(param=2))],
4323
... TestUtil.TestSuite())
4324
>>> tests = list(iter_suite_tests(r))
4328
'bzrlib.tests.test_sampler.DemoTest.test_nothing(one)'
4334
for test in iter_suite_tests(tests):
4335
apply_scenarios(test, scenarios, result)
4339
def apply_scenarios(test, scenarios, result):
4340
"""Apply the scenarios in scenarios to test and add to result.
4342
:param test: The test to apply scenarios to.
4343
:param scenarios: An iterable of scenarios to apply to test.
4345
:seealso: apply_scenario
4347
for scenario in scenarios:
4348
result.addTest(apply_scenario(test, scenario))
4352
def apply_scenario(test, scenario):
4353
"""Copy test and apply scenario to it.
4355
:param test: A test to adapt.
4356
:param scenario: A tuple describing the scenario.
4357
The first element of the tuple is the new test id.
4358
The second element is a dict containing attributes to set on the
4360
:return: The adapted test.
4362
new_id = "%s(%s)" % (test.id(), scenario[0])
4363
new_test = clone_test(test, new_id)
4364
for name, value in scenario[1].items():
4365
setattr(new_test, name, value)
4369
def clone_test(test, new_id):
4370
"""Clone a test giving it a new id.
4372
:param test: The test to clone.
4373
:param new_id: The id to assign to it.
4374
:return: The new test.
4376
new_test = copy.copy(test)
4377
new_test.id = lambda: new_id
4378
# XXX: Workaround <https://bugs.launchpad.net/testtools/+bug/637725>, which
4379
# causes cloned tests to share the 'details' dict. This makes it hard to
4380
# read the test output for parameterized tests, because tracebacks will be
4381
# associated with irrelevant tests.
4383
details = new_test._TestCase__details
4384
except AttributeError:
4385
# must be a different version of testtools than expected. Do nothing.
4388
# Reset the '__details' dict.
4389
new_test._TestCase__details = {}
4393
def permute_tests_for_extension(standard_tests, loader, py_module_name,
4395
"""Helper for permutating tests against an extension module.
4397
This is meant to be used inside a modules 'load_tests()' function. It will
4398
create 2 scenarios, and cause all tests in the 'standard_tests' to be run
4399
against both implementations. Setting 'test.module' to the appropriate
4400
module. See bzrlib.tests.test__chk_map.load_tests as an example.
4402
:param standard_tests: A test suite to permute
4403
:param loader: A TestLoader
4404
:param py_module_name: The python path to a python module that can always
4405
be loaded, and will be considered the 'python' implementation. (eg
4406
'bzrlib._chk_map_py')
4407
:param ext_module_name: The python path to an extension module. If the
4408
module cannot be loaded, a single test will be added, which notes that
4409
the module is not available. If it can be loaded, all standard_tests
4410
will be run against that module.
4411
:return: (suite, feature) suite is a test-suite that has all the permuted
4412
tests. feature is the Feature object that can be used to determine if
4413
the module is available.
4416
from bzrlib.tests.features import ModuleAvailableFeature
4417
py_module = pyutils.get_named_object(py_module_name)
4419
('python', {'module': py_module}),
4421
suite = loader.suiteClass()
4422
feature = ModuleAvailableFeature(ext_module_name)
4423
if feature.available():
4424
scenarios.append(('C', {'module': feature.module}))
4426
# the compiled module isn't available, so we add a failing test
4427
class FailWithoutFeature(TestCase):
4428
def test_fail(self):
4429
self.requireFeature(feature)
4430
suite.addTest(loader.loadTestsFromTestCase(FailWithoutFeature))
4431
result = multiply_tests(standard_tests, scenarios, suite)
4432
return result, feature
4435
def _rmtree_temp_dir(dirname, test_id=None):
4436
# If LANG=C we probably have created some bogus paths
4437
# which rmtree(unicode) will fail to delete
4438
# so make sure we are using rmtree(str) to delete everything
4439
# except on win32, where rmtree(str) will fail
4440
# since it doesn't have the property of byte-stream paths
4441
# (they are either ascii or mbcs)
4442
if sys.platform == 'win32':
4443
# make sure we are using the unicode win32 api
4444
dirname = unicode(dirname)
4446
dirname = dirname.encode(sys.getfilesystemencoding())
4448
osutils.rmtree(dirname)
4450
# We don't want to fail here because some useful display will be lost
4451
# otherwise. Polluting the tmp dir is bad, but not giving all the
4452
# possible info to the test runner is even worse.
4454
ui.ui_factory.clear_term()
4455
sys.stderr.write('\nWhile running: %s\n' % (test_id,))
4456
# Ugly, but the last thing we want here is fail, so bear with it.
4457
printable_e = str(e).decode(osutils.get_user_encoding(), 'replace'
4458
).encode('ascii', 'replace')
4459
sys.stderr.write('Unable to remove testing dir %s\n%s'
4460
% (os.path.basename(dirname), printable_e))
4463
def probe_unicode_in_user_encoding():
4464
"""Try to encode several unicode strings to use in unicode-aware tests.
4465
Return first successfull match.
4467
:return: (unicode value, encoded plain string value) or (None, None)
4469
possible_vals = [u'm\xb5', u'\xe1', u'\u0410']
4470
for uni_val in possible_vals:
4472
str_val = uni_val.encode(osutils.get_user_encoding())
4473
except UnicodeEncodeError:
4474
# Try a different character
4477
return uni_val, str_val
4481
def probe_bad_non_ascii(encoding):
4482
"""Try to find [bad] character with code [128..255]
4483
that cannot be decoded to unicode in some encoding.
4484
Return None if all non-ascii characters is valid
4487
for i in xrange(128, 256):
4490
char.decode(encoding)
4491
except UnicodeDecodeError:
4496
# Only define SubUnitBzrRunner if subunit is available.
4498
from subunit import TestProtocolClient
4499
from subunit.test_results import AutoTimingTestResultDecorator
4500
class SubUnitBzrProtocolClient(TestProtocolClient):
4502
def stopTest(self, test):
4503
super(SubUnitBzrProtocolClient, self).stopTest(test)
4504
_clear__type_equality_funcs(test)
4506
def addSuccess(self, test, details=None):
4507
# The subunit client always includes the details in the subunit
4508
# stream, but we don't want to include it in ours.
4509
if details is not None and 'log' in details:
4511
return super(SubUnitBzrProtocolClient, self).addSuccess(
4514
class SubUnitBzrRunner(TextTestRunner):
4515
def run(self, test):
4516
result = AutoTimingTestResultDecorator(
4517
SubUnitBzrProtocolClient(self.stream))
4524
# API compatibility for old plugins; see bug 892622.
4527
'HTTPServerFeature',
4528
'ModuleAvailableFeature',
4529
'HTTPSServerFeature', 'SymlinkFeature', 'HardlinkFeature',
4530
'OsFifoFeature', 'UnicodeFilenameFeature',
4531
'ByteStringNamedFilesystem', 'UTF8Filesystem',
4532
'BreakinFeature', 'CaseInsCasePresFilenameFeature',
4533
'CaseInsensitiveFilesystemFeature', 'case_sensitive_filesystem_feature',
4534
'posix_permissions_feature',
4536
globals()[name] = _CompatabilityThunkFeature(
4537
symbol_versioning.deprecated_in((2, 5, 0)),
4538
'bzrlib.tests', name,
4539
name, 'bzrlib.tests.features')
4542
for (old_name, new_name) in [
4543
('UnicodeFilename', 'UnicodeFilenameFeature'),
4545
globals()[name] = _CompatabilityThunkFeature(
4546
symbol_versioning.deprecated_in((2, 5, 0)),
4547
'bzrlib.tests', old_name,
4548
new_name, 'bzrlib.tests.features')
added
removed
bzrlib/tests/__init__.py
